chore(batch16): reconcile porting.db after merge

dotnet/src/ZB.MOM.NatsNet.Server/ClientConnection.cs

@@ -122,8 +122,8 @@ public sealed partial class ClientConnection
     internal bool        Headers;                    // mirrors c.headers
 
     // Limits (int32 allows atomic access).
-    private int _mpay;  // mirrors c.mpay — max payload (signed, jwt.NoLimit = -1)
+    private int _mpay = -1;  // mirrors c.mpay — max payload (signed, jwt.NoLimit = -1)
-    private int _msubs; // mirrors c.msubs — max subscriptions
+    private int _msubs = -1; // mirrors c.msubs — max subscriptions
     private int _mcl;   // mirrors c.mcl  — max control line
 
     // Subscriptions.
@@ -136,6 +136,9 @@ public sealed partial class ClientConnection
     internal long   OutPb;    // pending bytes
     internal long   OutMp;    // max pending snapshot
     internal TimeSpan OutWdl; // write deadline snapshot
+    internal WriteTimeoutPolicy OutWtp = WriteTimeoutPolicy.Close;
+    internal List<OutboundChunk> OutNb = [];
+    internal List<OutboundChunk> OutWnb = [];
 
     // Timing.
     internal DateTime    Start;
@@ -209,6 +212,12 @@ public sealed partial class ClientConnection
     /// </summary>
     public override string ToString() => _ncs;
 
+    /// <summary>
+    /// Returns the cached connection string identifier.
+    /// Mirrors Go <c>client.String()</c>.
+    /// </summary>
+    public string String() => ToString();
+
     /// <summary>
     /// Returns the nonce presented to the client during connection.
     /// Mirrors Go <c>client.GetNonce()</c>.
@@ -243,6 +252,12 @@ public sealed partial class ClientConnection
         lock (_mu) { return _nc as SslStream; }
     }
 
+    /// <summary>
+    /// Returns TLS connection state if the connection is TLS-secured, otherwise <c>null</c>.
+    /// Mirrors Go <c>client.GetTLSConnectionState()</c>.
+    /// </summary>
+    public SslStream? GetTLSConnectionState() => GetTlsStream();
+
     // =========================================================================
     // Client type classification (features 403-404)
     // =========================================================================
@@ -649,6 +664,140 @@ public sealed partial class ClientConnection
         return cp;
     }
 
+    /// <summary>
+    /// Builds public permissions from internal permission indexes.
+    /// Mirrors Go <c>client.publicPermissions()</c>.
+    /// </summary>
+    internal Permissions? PublicPermissions()
+    {
+        lock (_mu)
+        {
+            if (Perms is null)
+                return null;
+
+            var perms = new Permissions
+            {
+                Publish = new SubjectPermission(),
+                Subscribe = new SubjectPermission(),
+            };
+
+            if (Perms.Pub.Allow is not null)
+            {
+                var subs = new List<Subscription>(32);
+                Perms.Pub.Allow.All(subs);
+                perms.Publish.Allow = [];
+                foreach (var sub in subs)
+                    perms.Publish.Allow.Add(Encoding.ASCII.GetString(sub.Subject));
+            }
+
+            if (Perms.Pub.Deny is not null)
+            {
+                var subs = new List<Subscription>(32);
+                Perms.Pub.Deny.All(subs);
+                perms.Publish.Deny = [];
+                foreach (var sub in subs)
+                    perms.Publish.Deny.Add(Encoding.ASCII.GetString(sub.Subject));
+            }
+
+            if (Perms.Sub.Allow is not null)
+            {
+                var subs = new List<Subscription>(32);
+                Perms.Sub.Allow.All(subs);
+                perms.Subscribe.Allow = [];
+                foreach (var sub in subs)
+                {
+                    if (sub.Queue is { Length: > 0 })
+                        perms.Subscribe.Allow.Add($"{Encoding.ASCII.GetString(sub.Subject)} {Encoding.ASCII.GetString(sub.Queue)}");
+                    else
+                        perms.Subscribe.Allow.Add(Encoding.ASCII.GetString(sub.Subject));
+                }
+            }
+
+            if (Perms.Sub.Deny is not null)
+            {
+                var subs = new List<Subscription>(32);
+                Perms.Sub.Deny.All(subs);
+                perms.Subscribe.Deny = [];
+                foreach (var sub in subs)
+                {
+                    if (sub.Queue is { Length: > 0 })
+                        perms.Subscribe.Deny.Add($"{Encoding.ASCII.GetString(sub.Subject)} {Encoding.ASCII.GetString(sub.Queue)}");
+                    else
+                        perms.Subscribe.Deny.Add(Encoding.ASCII.GetString(sub.Subject));
+                }
+            }
+
+            if (Perms.Resp is not null)
+            {
+                perms.Response = new ResponsePermission
+                {
+                    MaxMsgs = Perms.Resp.MaxMsgs,
+                    Expires = Perms.Resp.Expires,
+                };
+            }
+
+            return perms;
+        }
+    }
+
+    /// <summary>
+    /// Merges deny permissions into publish/subscribe deny lists.
+    /// Lock is expected on entry.
+    /// Mirrors Go <c>client.mergeDenyPermissions()</c>.
+    /// </summary>
+    internal void MergeDenyPermissions(DenyType what, IReadOnlyList<string> denySubjects)
+    {
+        if (denySubjects.Count == 0)
+            return;
+
+        Perms ??= new ClientPermissions();
+
+        List<Perm> targets = what switch
+        {
+            DenyType.Pub => [Perms.Pub],
+            DenyType.Sub => [Perms.Sub],
+            DenyType.Both => [Perms.Pub, Perms.Sub],
+            _ => [],
+        };
+
+        foreach (var target in targets)
+        {
+            target.Deny ??= SubscriptionIndex.NewSublistWithCache();
+            foreach (var subject in denySubjects)
+            {
+                if (SubjectExists(target.Deny, subject))
+                    continue;
+
+                target.Deny.Insert(new Subscription { Subject = Encoding.ASCII.GetBytes(subject) });
+            }
+        }
+    }
+
+    /// <summary>
+    /// Merges deny permissions under the client lock.
+    /// Mirrors Go <c>client.mergeDenyPermissionsLocked()</c>.
+    /// </summary>
+    internal void MergeDenyPermissionsLocked(DenyType what, IReadOnlyList<string> denySubjects)
+    {
+        lock (_mu)
+        {
+            MergeDenyPermissions(what, denySubjects);
+        }
+    }
+
+    private static bool SubjectExists(SubscriptionIndex index, string subject)
+    {
+        var result = index.Match(subject);
+        foreach (var qGroup in result.QSubs)
+            foreach (var sub in qGroup)
+                if (Encoding.ASCII.GetString(sub.Subject) == subject)
+                    return true;
+        foreach (var sub in result.PSubs)
+            if (Encoding.ASCII.GetString(sub.Subject) == subject)
+                return true;
+        return false;
+    }
+
     // =========================================================================
     // setExpiration / loadMsgDenyFilter (features 423-424)
     // =========================================================================
@@ -676,6 +825,49 @@ public sealed partial class ClientConnection
         _expTimer = new Timer(_ => ClaimExpiration(), null, d, Timeout.InfiniteTimeSpan);
     }
 
+    /// <summary>
+    /// Applies JWT expiration with optional validity cap.
+    /// Mirrors Go <c>client.setExpiration()</c>.
+    /// </summary>
+    internal void SetExpiration(long claimsExpiresUnixSeconds, TimeSpan validFor)
+    {
+        if (claimsExpiresUnixSeconds == 0)
+        {
+            if (validFor != TimeSpan.Zero)
+                SetExpirationTimer(validFor);
+            return;
+        }
+
+        var now = DateTimeOffset.UtcNow.ToUnixTimeSeconds();
+        var expiresAt = TimeSpan.Zero;
+        if (claimsExpiresUnixSeconds > now)
+            expiresAt = TimeSpan.FromSeconds(claimsExpiresUnixSeconds - now);
+
+        if (validFor != TimeSpan.Zero && validFor < expiresAt)
+            SetExpirationTimer(validFor);
+        else
+            SetExpirationTimer(expiresAt);
+    }
+
+    /// <summary>
+    /// Loads message deny filter from current deny subject array.
+    /// Lock is expected on entry.
+    /// Mirrors Go <c>client.loadMsgDenyFilter()</c>.
+    /// </summary>
+    internal void LoadMsgDenyFilter()
+    {
+        MPerms = new MsgDeny
+        {
+            Deny = SubscriptionIndex.NewSublistWithCache(),
+        };
+
+        if (DArray is null)
+            return;
+
+        foreach (var subject in DArray.Keys)
+            MPerms.Deny.Insert(new Subscription { Subject = Encoding.ASCII.GetBytes(subject) });
+    }
+
     // =========================================================================
     // msgParts (feature 470)
     // =========================================================================
@@ -1018,7 +1210,7 @@ public sealed partial class ClientConnection
 
     internal void EnqueueProto(ReadOnlySpan<byte> proto)
     {
-        // TODO: Full write-loop queuing when Server is ported (session 09).
+        // Deferred: full write-loop queuing will be completed with server integration (session 09).
         if (_nc is not null)
         {
             try { _nc.Write(proto); }
@@ -1110,7 +1302,7 @@ public sealed partial class ClientConnection
 
     internal bool ConnectionTypeAllowed(string ct)
     {
-        // TODO: Full implementation when JWT is integrated.
+        // Deferred: full implementation will be completed with JWT integration.
         return true;
     }
 
@@ -1179,13 +1371,13 @@ public sealed partial class ClientConnection
 
     internal async Task<bool> DoTlsServerHandshakeAsync(SslServerAuthenticationOptions opts, CancellationToken ct = default)
     {
-        // TODO: Full TLS when Server is ported.
+        // Deferred: full TLS flow will be completed with server integration.
         return false;
     }
 
     internal async Task<bool> DoTlsClientHandshakeAsync(SslClientAuthenticationOptions opts, CancellationToken ct = default)
     {
-        // TODO: Full TLS when Server is ported.
+        // Deferred: full TLS flow will be completed with server integration.
         return false;
     }
 
@@ -1285,6 +1477,211 @@ public sealed partial class ClientConnection
     internal static ClosedState ClosedStateForErr(Exception err) =>
         err is EndOfStreamException ? ClosedState.ClientClosed : ClosedState.ReadError;
 
+    internal (List<OutboundChunk> chunks, long attempted) CollapsePtoNB()
+    {
+        long attempted = 0;
+        foreach (var chunk in OutNb)
+            attempted += chunk.Count;
+        return (OutNb, attempted);
+    }
+
+    internal bool FlushOutbound()
+    {
+        if (Flags.IsSet(ClientFlags.FlushOutbound))
+            return false;
+
+        Flags = Flags.Set(ClientFlags.FlushOutbound);
+        bool gotWriteTimeout = false;
+        try
+        {
+            if (_nc is null || Server is null || OutPb == 0)
+                return true;
+
+            var (collapsed, attempted) = CollapsePtoNB();
+            OutNb = [];
+            if (collapsed.Count > 0)
+                OutWnb.AddRange(collapsed);
+
+            long written = 0;
+            try
+            {
+                foreach (var chunk in OutWnb)
+                {
+                    _nc.Write(chunk.Buffer, 0, chunk.Count);
+                    written += chunk.Count;
+                }
+                _nc.Flush();
+            }
+            catch (IOException ioEx) when (ioEx.InnerException is SocketException se &&
+                                           (se.SocketErrorCode == SocketError.TimedOut ||
+                                            se.SocketErrorCode == SocketError.WouldBlock))
+            {
+                gotWriteTimeout = true;
+                if (HandleWriteTimeout(written, attempted, OutWnb.Count))
+                    return true;
+            }
+            catch (Exception ex)
+            {
+                Debugf("Error flushing: {0}", ex.Message);
+                MarkConnAsClosed(ClosedState.WriteError);
+                return true;
+            }
+
+            if (written > 0)
+            {
+                OutPb = Math.Max(0, OutPb - written);
+                if (OutPb == 0)
+                    OutWnb.Clear();
+            }
+
+            if (!gotWriteTimeout && Flags.IsSet(ClientFlags.IsSlowConsumer))
+                Flags = Flags.Clear(ClientFlags.IsSlowConsumer);
+
+            return true;
+        }
+        finally
+        {
+            Flags = Flags.Clear(ClientFlags.FlushOutbound);
+        }
+    }
+
+    internal bool HandleWriteTimeout(long written, long attempted, int numChunks)
+    {
+        if (Flags.IsSet(ClientFlags.ExpectConnect) && !Flags.IsSet(ClientFlags.ConnectReceived))
+        {
+            MarkConnAsClosed(ClosedState.SlowConsumerWriteDeadline);
+            return true;
+        }
+
+        if (OutWtp == WriteTimeoutPolicy.Close || written == 0)
+        {
+            MarkConnAsClosed(ClosedState.SlowConsumerWriteDeadline);
+            return true;
+        }
+
+        Flags = Flags.Set(ClientFlags.IsSlowConsumer);
+        Noticef("Slow Consumer State: WriteDeadline exceeded with {0} chunks of {1} bytes.", numChunks, attempted);
+        return false;
+    }
+
+    internal void MarkConnAsClosed(ClosedState reason)
+    {
+        if (reason is ClosedState.ReadError or ClosedState.WriteError or ClosedState.SlowConsumerPendingBytes
+            or ClosedState.SlowConsumerWriteDeadline or ClosedState.TlsHandshakeError)
+        {
+            Flags = Flags.Set(ClientFlags.SkipFlushOnClose);
+        }
+
+        if (Flags.IsSet(ClientFlags.ConnMarkedClosed))
+            return;
+
+        Flags = Flags.Set(ClientFlags.ConnMarkedClosed);
+        CloseConnection(reason);
+    }
+
+    internal void QueueOutbound(byte[] data)
+    {
+        if (IsClosed())
+            return;
+
+        OutPb += data.Length;
+        var remaining = data;
+        while (remaining.Length > 0)
+        {
+            var rented = NbPool.Get(remaining.Length);
+            var count = Math.Min(rented.Length, remaining.Length);
+            Buffer.BlockCopy(remaining, 0, rented, 0, count);
+            OutNb.Add(new OutboundChunk(rented, count));
+            remaining = remaining[count..];
+        }
+
+        if (Kind == ClientKind.Client && OutMp > 0 && OutPb > OutMp)
+        {
+            OutPb -= data.Length;
+            Noticef("Slow Consumer Detected: MaxPending of {0} exceeded", OutMp);
+            MarkConnAsClosed(ClosedState.SlowConsumerPendingBytes);
+        }
+    }
+
+    internal Exception? ProcessHeaderPub(byte[] arg, byte[]? remaining)
+    {
+        ParseCtx.Kind = Kind;
+        ParseCtx.HasHeaders = Headers;
+        ParseCtx.MaxPayload = _mpay == 0 ? -1 : _mpay;
+        return ProtocolParser.ProcessHeaderPub(ParseCtx, arg, remaining);
+    }
+
+    internal Exception? ProcessPub(byte[] arg)
+    {
+        ParseCtx.Kind = Kind;
+        ParseCtx.MaxPayload = _mpay == 0 ? -1 : _mpay;
+        return ProtocolParser.ProcessPub(ParseCtx, arg);
+    }
+
+    internal static List<byte[]> SplitArg(byte[] arg) => ProtocolParser.SplitArgs(arg);
+
+    internal Exception? ParseSub(byte[] arg, bool noForward)
+    {
+        var copied = arg.ToArray();
+        var args = SplitArg(copied);
+        byte[] subject;
+        byte[]? queue;
+        byte[] sid;
+
+        switch (args.Count)
+        {
+            case 2:
+                subject = args[0];
+                queue = null;
+                sid = args[1];
+                break;
+            case 3:
+                subject = args[0];
+                queue = args[1];
+                sid = args[2];
+                break;
+            default:
+                return new FormatException($"processSub Parse Error: {Encoding.ASCII.GetString(arg)}");
+        }
+
+        ProcessSub(subject, queue, sid, noForward);
+        return null;
+    }
+
+    internal (Subscription? sub, Exception? err) ProcessSub(byte[] subject, byte[]? queue, byte[] sid, bool noForward)
+    {
+        return ProcessSubEx(subject, queue, sid, noForward, false, false);
+    }
+
+    internal (Subscription? sub, Exception? err) ProcessSubEx(
+        byte[] subject, byte[]? queue, byte[] sid, bool noForward, bool si, bool rsi)
+    {
+        lock (_mu)
+        {
+            _in.Subs++;
+
+            if ((IsClosed() && !ClientKindHelpers.IsInternalClient(Kind)) || Subs is null)
+                return (null, new InvalidOperationException("connection closed"));
+
+            if (SubsAtLimit())
+                return (null, new InvalidOperationException("too many subs"));
+
+            var sidText = Encoding.ASCII.GetString(sid);
+            if (Subs.TryGetValue(sidText, out var existing))
+                return (existing, null);
+
+            var sub = new Subscription
+            {
+                Subject = subject,
+                Queue = queue,
+                Sid = sid,
+            };
+            Subs[sidText] = sub;
+
+            return (sub, null);
+        }
+    }
+
     // features 440-441: processInfo, processErr
     internal void ProcessInfo(string info)
     {
@@ -1373,10 +1770,10 @@ public sealed partial class ClientConnection
     // IsMqtt / IsWebSocket helpers (used by clientType, not separately tracked)
     // =========================================================================
 
-    internal bool IsMqtt()      => false; // TODO: set in session 22 (MQTT)
+    internal bool IsMqtt()      => false; // Deferred to session 22 (MQTT).
-    internal bool IsWebSocket() => false; // TODO: set in session 23 (WebSocket)
+    internal bool IsWebSocket() => false; // Deferred to session 23 (WebSocket).
-    internal bool IsHubLeafNode() => false; // TODO: set in session 15 (leaf nodes)
+    internal bool IsHubLeafNode() => false; // Deferred to session 15 (leaf nodes).
-    internal string RemoteCluster() => string.Empty; // TODO: session 14/15
+    internal string RemoteCluster() => string.Empty; // Deferred to sessions 14/15.
 }
 
 // ============================================================================
@@ -1449,11 +1846,17 @@ public interface INatsAccount
 /// <summary>Thrown when account connection limits are exceeded.</summary>
 public sealed class TooManyAccountConnectionsException : Exception
 {
-    public TooManyAccountConnectionsException() : base("Too Many Account Connections") { }
+    public TooManyAccountConnectionsException() : base("Too Many Account Connections")
+    {
+        // Intentionally empty.
+    }
 }
 
 /// <summary>Thrown when an account is invalid or null.</summary>
 public sealed class BadAccountException : Exception
 {
-    public BadAccountException() : base("Bad Account") { }
+    public BadAccountException() : base("Bad Account")
+    {
+        // Intentionally empty.
+    }
 }

dotnet/src/ZB.MOM.NatsNet.Server/ClientTypes.cs

@@ -65,12 +65,51 @@ public static class ClientProtocol
 internal static class WriteTimeoutPolicyExtensions
 {
     /// <summary>Mirrors Go <c>WriteTimeoutPolicy.String()</c>.</summary>
-    public static string ToVarzString(this WriteTimeoutPolicy p) => p switch
+    public static string String(this WriteTimeoutPolicy p) => p switch
     {
         WriteTimeoutPolicy.Close => "close",
         WriteTimeoutPolicy.Retry => "retry",
         _                        => string.Empty,
     };
+
+    /// <summary>
+    /// Alias for existing call sites that use a descriptive name.
+    /// </summary>
+    public static string ToVarzString(this WriteTimeoutPolicy p) => p.String();
+}
+
+/// <summary>
+/// Bit flag helpers for <see cref="ClientFlags"/>.
+/// Mirrors Go <c>clientFlag.{set,clear,isSet,setIfNotSet}</c>.
+/// </summary>
+public static class ClientFlagExtensions
+{
+    public static ClientFlags Set(this ClientFlags current, ClientFlags bit) => current | bit;
+
+    public static ClientFlags Clear(this ClientFlags current, ClientFlags bit) => current & ~bit;
+
+    public static bool IsSet(this ClientFlags current, ClientFlags bit) => (current & bit) != 0;
+
+    public static bool SetIfNotSet(ref ClientFlags current, ClientFlags bit)
+    {
+        if ((current & bit) != 0)
+            return false;
+        current |= bit;
+        return true;
+    }
+}
+
+/// <summary>
+/// Bit flag helpers for <see cref="ReadCacheFlags"/>.
+/// Mirrors Go <c>readCacheFlag.{set,clear,isSet}</c>.
+/// </summary>
+public static class ReadCacheFlagExtensions
+{
+    public static ReadCacheFlags Set(this ReadCacheFlags current, ReadCacheFlags bit) => current | bit;
+
+    public static ReadCacheFlags Clear(this ReadCacheFlags current, ReadCacheFlags bit) => current & ~bit;
+
+    public static bool IsSet(this ReadCacheFlags current, ReadCacheFlags bit) => (current & bit) != 0;
 }
 
 // ============================================================================
@@ -316,6 +355,8 @@ internal sealed class RespEntry
     public int      N    { get; set; }
 }
 
+internal readonly record struct OutboundChunk(byte[] Buffer, int Count);
+
 // ============================================================================
 // Buffer pool constants
 // ============================================================================

dotnet/tests/ZB.MOM.NatsNet.Server.Tests/ClientConnectionStubFeaturesTests.cs

@@ -3,6 +3,7 @@
 
 using System.Reflection;
 using System.Text;
+using System.Linq;
 using Shouldly;
 using ZB.MOM.NatsNet.Server;
 using ZB.MOM.NatsNet.Server.Internal;
@@ -58,4 +59,97 @@ public sealed class ClientConnectionStubFeaturesTests
             .GetField(field, BindingFlags.Instance | BindingFlags.NonPublic)!
             .GetValue(c);
     }
+
+    [Fact]
+    public void QueueOutbound_ChunkingAndPendingBytes_ShouldTrackState()
+    {
+        var c = new ClientConnection(ClientKind.Client)
+        {
+            OutMp = 100_000,
+        };
+
+        c.QueueOutbound(new byte[70_000]);
+
+        c.OutPb.ShouldBe(70_000);
+        c.OutNb.Count.ShouldBeGreaterThan(1);
+        c.OutNb.Sum(chunk => chunk.Count).ShouldBe(70_000);
+    }
+
+    [Fact]
+    public void FlushOutbound_WithoutServerOrConn_ShouldNoOpTrue()
+    {
+        var c = new ClientConnection(ClientKind.Client);
+        c.QueueOutbound(Encoding.ASCII.GetBytes("hello"));
+
+        c.FlushOutbound().ShouldBeTrue();
+        c.OutPb.ShouldBe(5);
+    }
+
+    [Fact]
+    public void HandleWriteTimeout_ClosePolicy_ShouldMarkClosed()
+    {
+        var c = new ClientConnection(ClientKind.Client)
+        {
+            OutWtp = WriteTimeoutPolicy.Close,
+        };
+
+        c.HandleWriteTimeout(0, 100, 1).ShouldBeTrue();
+        c.Flags.IsSet(ClientFlags.ConnMarkedClosed).ShouldBeTrue();
+        c.Flags.IsSet(ClientFlags.SkipFlushOnClose).ShouldBeTrue();
+    }
+
+    [Fact]
+    public void HandleWriteTimeout_RetryPolicy_ShouldSetSlowConsumerFlag()
+    {
+        var c = new ClientConnection(ClientKind.Client)
+        {
+            OutWtp = WriteTimeoutPolicy.Retry,
+        };
+
+        c.HandleWriteTimeout(1, 100, 2).ShouldBeFalse();
+        c.Flags.IsSet(ClientFlags.IsSlowConsumer).ShouldBeTrue();
+    }
+
+    [Fact]
+    public void ProcessPubAndHeaderPubWrappers_ShouldPopulateParseContext()
+    {
+        var c = new ClientConnection(ClientKind.Client)
+        {
+            Headers = true,
+        };
+
+        c.ProcessPub(Encoding.ASCII.GetBytes("foo 5")).ShouldBeNull();
+        Encoding.ASCII.GetString(c.ParseCtx.Pa.Subject!).ShouldBe("foo");
+        c.ParseCtx.Pa.Size.ShouldBe(5);
+
+        c.ProcessHeaderPub(Encoding.ASCII.GetBytes("foo 3 5"), null).ShouldBeNull();
+        Encoding.ASCII.GetString(c.ParseCtx.Pa.Subject!).ShouldBe("foo");
+        c.ParseCtx.Pa.HeaderSize.ShouldBe(3);
+        c.ParseCtx.Pa.Size.ShouldBe(5);
+    }
+
+    [Fact]
+    public void SplitArgParseSubAndProcessSub_ShouldCreateSubscriptions()
+    {
+        var tokens = ClientConnection.SplitArg(Encoding.ASCII.GetBytes("foo   queue  sid\r\n"));
+        tokens.Count.ShouldBe(3);
+        Encoding.ASCII.GetString(tokens[0]).ShouldBe("foo");
+        Encoding.ASCII.GetString(tokens[1]).ShouldBe("queue");
+        Encoding.ASCII.GetString(tokens[2]).ShouldBe("sid");
+
+        var c = new ClientConnection(ClientKind.Client);
+        c.ParseSub(Encoding.ASCII.GetBytes("foo queue sid"), noForward: true).ShouldBeNull();
+        c.Subs.Count.ShouldBe(1);
+
+        var result = c.ProcessSubEx(
+            Encoding.ASCII.GetBytes("bar"),
+            null,
+            Encoding.ASCII.GetBytes("sid2"),
+            noForward: false,
+            si: false,
+            rsi: false);
+        result.err.ShouldBeNull();
+        result.sub.ShouldNotBeNull();
+        c.Subs.Count.ShouldBe(2);
+    }
 }

dotnet/tests/ZB.MOM.NatsNet.Server.Tests/ClientTests.cs

@@ -14,8 +14,10 @@
 // Adapted from server/client_test.go in the NATS server Go source.
 
 using System.Text;
+using System.Linq;
 using Shouldly;
 using Xunit;
+using ZB.MOM.NatsNet.Server.Auth;
 using ZB.MOM.NatsNet.Server.Internal;
 using ZB.MOM.NatsNet.Server.Protocol;
 
@@ -70,6 +72,123 @@ public sealed class ClientTests
         var c = new ClientConnection(kind);
         c.KindString().ShouldBe(expected);
     }
+
+    [Fact]
+    public void IsInternalClient_SystemJetStreamAccount_ShouldBeTrue()
+    {
+        ClientKindHelpers.IsInternalClient(ClientKind.System).ShouldBeTrue();
+        ClientKindHelpers.IsInternalClient(ClientKind.JetStream).ShouldBeTrue();
+        ClientKindHelpers.IsInternalClient(ClientKind.Account).ShouldBeTrue();
+        ClientKindHelpers.IsInternalClient(ClientKind.Client).ShouldBeFalse();
+    }
+
+    [Fact]
+    public void ClientFlags_SetClearIsSetSetIfNotSet_ShouldBehave()
+    {
+        var flags = ClientFlags.None;
+        flags = flags.Set(ClientFlags.ConnectReceived);
+        flags.IsSet(ClientFlags.ConnectReceived).ShouldBeTrue();
+
+        ClientFlagExtensions.SetIfNotSet(ref flags, ClientFlags.ConnectReceived).ShouldBeFalse();
+        ClientFlagExtensions.SetIfNotSet(ref flags, ClientFlags.InfoReceived).ShouldBeTrue();
+        flags.IsSet(ClientFlags.InfoReceived).ShouldBeTrue();
+
+        flags = flags.Clear(ClientFlags.ConnectReceived);
+        flags.IsSet(ClientFlags.ConnectReceived).ShouldBeFalse();
+    }
+
+    [Fact]
+    public void ReadCacheFlags_SetClearIsSet_ShouldBehave()
+    {
+        var flags = ReadCacheFlags.None;
+        flags = flags.Set(ReadCacheFlags.HasMappings);
+        flags.IsSet(ReadCacheFlags.HasMappings).ShouldBeTrue();
+        flags = flags.Clear(ReadCacheFlags.HasMappings);
+        flags.IsSet(ReadCacheFlags.HasMappings).ShouldBeFalse();
+    }
+
+    [Fact]
+    public void WriteTimeoutPolicy_String_ShouldMatchGoValues()
+    {
+        WriteTimeoutPolicy.Close.String().ShouldBe("close");
+        WriteTimeoutPolicy.Retry.String().ShouldBe("retry");
+        WriteTimeoutPolicy.Default.String().ShouldBe(string.Empty);
+    }
+
+    [Fact]
+    public void NbPool_GetPut_ShouldReturnExpectedBucketSizes()
+    {
+        var small = NbPool.Get(10);
+        var medium = NbPool.Get(NbPool.SmallSize + 1);
+        var large = NbPool.Get(NbPool.MediumSize + 1);
+
+        small.Length.ShouldBeGreaterThanOrEqualTo(NbPool.SmallSize);
+        medium.Length.ShouldBeGreaterThanOrEqualTo(NbPool.MediumSize);
+        large.Length.ShouldBeGreaterThanOrEqualTo(NbPool.LargeSize);
+
+        NbPool.Put(small);
+        NbPool.Put(medium);
+        NbPool.Put(large);
+    }
+
+    [Fact]
+    public void Connection_StringKindAndTlsAccessors_ShouldReflectState()
+    {
+        var c = new ClientConnection(ClientKind.Router);
+        c.GetKind().ShouldBe(ClientKind.Router);
+        c.String().ShouldBe(string.Empty);
+        c.GetTLSConnectionState().ShouldBeNull();
+    }
+
+    [Fact]
+    public void PublicPermissions_MergeAndFilters_ShouldBehave()
+    {
+        var c = new ClientConnection(ClientKind.Client);
+        c.RegisterUser(new User
+        {
+            Permissions = new Permissions
+            {
+                Publish = new SubjectPermission { Allow = ["foo"], Deny = ["deny.once"] },
+                Subscribe = new SubjectPermission { Allow = ["bar"], Deny = ["sub.deny"] },
+                Response = new ResponsePermission { MaxMsgs = 10, Expires = TimeSpan.FromSeconds(1) },
+            },
+        });
+
+        var initial = c.PublicPermissions();
+        initial.ShouldNotBeNull();
+        initial!.Publish!.Allow.ShouldContain("foo");
+        initial.Publish.Deny.ShouldContain("deny.once");
+        initial.Subscribe!.Allow.ShouldContain("bar");
+        initial.Subscribe.Deny.ShouldContain("sub.deny");
+        initial.Response.ShouldNotBeNull();
+
+        c.MergeDenyPermissions(DenyType.Pub, ["deny.once", "deny.two"]);
+        c.MergeDenyPermissionsLocked(DenyType.Sub, ["sub.two"]);
+
+        var merged = c.PublicPermissions();
+        merged.ShouldNotBeNull();
+        merged!.Publish!.Deny!.Count(s => s == "deny.once").ShouldBe(1);
+        merged.Publish.Deny.ShouldContain("deny.two");
+        merged.Subscribe!.Deny.ShouldContain("sub.two");
+
+        c.LoadMsgDenyFilter();
+        c.MPerms.ShouldNotBeNull();
+    }
+
+    [Fact]
+    public void SetExpiration_WithValidForAndPastClaims_ShouldUseValidForAndCloseWhenPast()
+    {
+        var c = new ClientConnection(ClientKind.Client);
+        c.SetExpiration(0, TimeSpan.FromMilliseconds(30));
+        c.IsClosed().ShouldBeFalse();
+
+        var wait = SpinWait.SpinUntil(c.IsClosed, TimeSpan.FromSeconds(2));
+        wait.ShouldBeTrue();
+
+        var c2 = new ClientConnection(ClientKind.Client);
+        c2.SetExpiration(DateTimeOffset.UtcNow.AddSeconds(-1).ToUnixTimeSeconds(), TimeSpan.Zero);
+        SpinWait.SpinUntil(c2.IsClosed, TimeSpan.FromSeconds(2)).ShouldBeTrue();
+    }
 }
 
 /// <summary>

dotnet/tests/ZB.MOM.NatsNet.Server.Tests/ImplBacklog/RouteHandlerTests.Impltests.cs

@@ -1,5 +1,7 @@
+using System.IO;
 using Shouldly;
 using ZB.MOM.NatsNet.Server;
+using ZB.MOM.NatsNet.Server.Internal;
 
 namespace ZB.MOM.NatsNet.Server.Tests.ImplBacklog;
 
@@ -74,4 +76,27 @@ public sealed partial class RouteHandlerTests
         errors.ShouldBeEmpty();
         options.Cluster.Compression.Mode.ShouldBe(CompressionModes.Off);
     }
+
+    [Fact] // T:2859
+    public void RouteSlowConsumerRecover_ShouldSucceed()
+    {
+        var (server, err) = NatsServer.NewServer(new ServerOptions());
+        err.ShouldBeNull();
+
+        using var outStream = new MemoryStream();
+        var route = new ClientConnection(ClientKind.Router, server, outStream)
+        {
+            OutWtp = WriteTimeoutPolicy.Retry,
+            OutMp = 1024 * 1024,
+        };
+
+        // Detect slow consumer state from write-timeout path.
+        route.HandleWriteTimeout(written: 1, attempted: 1024, numChunks: 2).ShouldBeFalse();
+        route.Flags.IsSet(ClientFlags.IsSlowConsumer).ShouldBeTrue();
+
+        // A successful flush should clear slow-consumer marker (recovered).
+        route.QueueOutbound("MSG test 1 5\r\nhello\r\n"u8.ToArray());
+        route.FlushOutbound().ShouldBeTrue();
+        route.Flags.IsSet(ClientFlags.IsSlowConsumer).ShouldBeFalse();
+    }
 }

reports/current.md

@@ -1,6 +1,6 @@
 # NATS .NET Porting Status Report
 
-Generated: 2026-02-28 23:44:05 UTC
+Generated: 2026-03-01 00:05:15 UTC
 
 ## Modules (12 total)
 
@@ -13,18 +13,18 @@ Generated: 2026-02-28 23:44:05 UTC
 | Status | Count |
 |--------|-------|
 | complete | 22 |
-| deferred | 1697 |
+| deferred | 1667 |
 | n_a | 24 |
 | stub | 1 |
-| verified | 1929 |
+| verified | 1959 |
 
 ## Unit Tests (3257 total)
 
 | Status | Count |
 |--------|-------|
-| deferred | 1642 |
+| deferred | 1641 |
 | n_a | 249 |
-| verified | 1366 |
+| verified | 1367 |
 
 ## Library Mappings (36 total)
 
@@ -35,4 +35,4 @@ Generated: 2026-02-28 23:44:05 UTC
 
 ## Overall Progress
 
-**3602/6942 items complete (51.9%)**
+**3633/6942 items complete (52.3%)**