7 Commits

Author SHA1 Message Date
Joseph Doherty bbacf439ed chore(batch23): reconcile porting.db after merge 2026-02-28 21:33:58 -05:00
Joseph Doherty 09c42538c2 feat(batch23): complete routes implementation and verification 2026-02-28 21:27:21 -05:00
Joseph Doherty 3501320c6e test(batch23): port and verify mapped route tests 2026-02-28 21:24:18 -05:00
Joseph Doherty 51c899b651 feat(batch23): implement route lifecycle, solicitation, dedupe, and iteration 2026-02-28 21:19:30 -05:00
Joseph Doherty acf51bf480 feat(batch23): implement route subscription propagation and creation helpers 2026-02-28 21:16:01 -05:00
Joseph Doherty e51cdd64f4 feat(batch23): implement route protocol and info/perms foundation 2026-02-28 21:09:00 -05:00
Joseph Doherty f2bc957229 chore(batch23): start routes batch 2026-02-28 20:59:47 -05:00
13 changed files with 1131 additions and 14 deletions
@@ -2279,7 +2279,7 @@ public sealed class Account : INatsAccount
} }
if (sid is { Length: > 0 } && InternalClient != null) if (sid is { Length: > 0 } && InternalClient != null)
InternalClient.ProcessUnsub(sid); InternalClient.RemoveSubBySid(sid);
if (tracking && requestor != null && !delivered) if (tracking && requestor != null && !delivered)
SendBackendErrorTrackingLatency(serviceImport, reason); SendBackendErrorTrackingLatency(serviceImport, reason);
@@ -2355,7 +2355,7 @@ public sealed class Account : INatsAccount
} }
if (sid != null && InternalClient != null) if (sid != null && InternalClient != null)
InternalClient.ProcessUnsub(sid); InternalClient.RemoveSubBySid(sid);
} }
/// <summary> /// <summary>
@@ -2548,7 +2548,7 @@ public sealed class Account : INatsAccount
if (InternalClient == null && Server is NatsServer server) if (InternalClient == null && Server is NatsServer server)
{ {
InternalClient = server.CreateInternalAccountClient(); InternalClient = server.CreateInternalAccountClient();
InternalClient.Account = this; InternalClient.SetAccount(this);
} }
return InternalClient; return InternalClient;
@@ -2573,7 +2573,7 @@ public sealed class Account : INatsAccount
_mu.EnterReadLock(); _mu.EnterReadLock();
var internalClient = InternalClient; var internalClient = InternalClient;
_mu.ExitReadLock(); _mu.ExitReadLock();
internalClient?.ProcessUnsub(sub.Sid); internalClient?.RemoveSubBySid(sub.Sid);
} }
/// <summary> /// <summary>
@@ -2685,7 +2685,7 @@ public sealed class Account : INatsAccount
return; return;
foreach (var sid in subscriptionIds) foreach (var sid in subscriptionIds)
internalClient.ProcessUnsub(sid); internalClient.RemoveSubBySid(sid);
internalClient.CloseConnection(ClosedState.InternalClient); internalClient.CloseConnection(ClosedState.InternalClient);
} }
@@ -4170,7 +4170,7 @@ public sealed class Account : INatsAccount
return new ClientInfo return new ClientInfo
{ {
Id = client.Cid, Id = client.Cid,
Account = client.Account?.Name ?? string.Empty, Account = client.Account()?.Name ?? string.Empty,
Name = client.Opts.Name ?? string.Empty, Name = client.Opts.Name ?? string.Empty,
Rtt = client.GetRttValue(), Rtt = client.GetRttValue(),
Start = client.Start == default ? string.Empty : client.Start.ToUniversalTime().ToString("O"), Start = client.Start == default ? string.Empty : client.Start.ToUniversalTime().ToString("O"),
@@ -27,6 +27,8 @@ public sealed partial class ClientConnection
return; return;
var staleAfter = TimeSpan.FromTicks(pingInterval.Ticks * (pingMax + 1L)); var staleAfter = TimeSpan.FromTicks(pingInterval.Ticks * (pingMax + 1L));
if (pingMax == 0 && staleAfter > TimeSpan.Zero)
staleAfter = TimeSpan.FromTicks(Math.Max(1, pingInterval.Ticks / 2));
if (staleAfter <= TimeSpan.Zero) if (staleAfter <= TimeSpan.Zero)
return; return;
@@ -0,0 +1,486 @@
// Copyright 2012-2026 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
using System.Text;
using System.Text.Json;
using ZB.MOM.NatsNet.Server.Auth;
using ZB.MOM.NatsNet.Server.Internal;
using ZB.MOM.NatsNet.Server.Protocol;
namespace ZB.MOM.NatsNet.Server;
public sealed partial class ClientConnection
{
internal void RemoveReplySub(Subscription? sub)
{
if (sub?.Sid is not { Length: > 0 } sid || Server is not NatsServer server)
return;
var sidText = Encoding.ASCII.GetString(sid);
var sep = sidText.IndexOf(' ');
if (sep <= 0)
return;
var accountName = sidText[..sep];
var (account, _) = server.LookupAccount(accountName);
account?.Sublist?.Remove(sub);
lock (_mu)
{
Subs.Remove(sidText);
}
}
internal Exception? ProcessAccountSub(byte[] arg)
{
_ = arg;
// Gateway account-sub propagation is owned by gateway sessions.
return null;
}
internal void ProcessAccountUnsub(byte[] arg)
{
_ = arg;
// Gateway account-unsub propagation is owned by gateway sessions.
}
internal Exception? ProcessRoutedOriginClusterMsgArgs(byte[] arg) =>
ProtocolParser.ProcessRoutedOriginClusterMsgArgs(ParseCtx, arg);
internal Exception? ProcessRoutedHeaderMsgArgs(byte[] arg) =>
ProtocolParser.ProcessRoutedHeaderMsgArgs(ParseCtx, arg);
internal Exception? ProcessRoutedMsgArgs(byte[] arg) =>
ProtocolParser.ProcessRoutedMsgArgs(ParseCtx, arg);
internal void ProcessInboundRoutedMsg(byte[] msg)
{
_in.Msgs++;
_in.Bytes += Math.Max(0, msg.Length - 2);
if (Opts.Verbose)
SendOK();
var pa = ParseCtx.Pa;
if (pa.Subject is null)
return;
var (acc, result) = GetAccAndResultFromCache();
if (acc is null)
return;
if ((result?.PSubs.Count ?? 0) + (result?.QSubs.Count ?? 0) > 0)
ProcessMsgResults(acc, result, msg, null, pa.Subject, pa.Reply, PmrFlags.None);
}
internal Exception? SendRouteConnect(string clusterName, bool tlsRequired)
{
var user = string.Empty;
var pass = string.Empty;
var routeUrl = Route?.Url;
if (routeUrl is not null && !string.IsNullOrEmpty(routeUrl.UserInfo))
{
var userInfo = routeUrl.UserInfo.Split(':', 2);
user = userInfo[0];
if (userInfo.Length > 1)
pass = userInfo[1];
}
if (Server is not NatsServer server)
return new InvalidOperationException("route server unavailable");
var connect = new ConnectInfo
{
Echo = true,
Verbose = false,
Pedantic = false,
User = user,
Pass = pass,
Tls = tlsRequired,
Name = server.ID(),
Headers = server.SupportsHeaders(),
Cluster = clusterName,
Dynamic = server.IsClusterNameDynamic(),
Lnoc = true,
};
var payload = JsonSerializer.Serialize(connect);
EnqueueProto(Encoding.ASCII.GetBytes($"CONNECT {payload}\r\n"));
return null;
}
internal void ProcessRouteInfo(ServerInfo info)
{
if (Server is not NatsServer server)
return;
lock (_mu)
{
Route ??= new Route();
if (Flags.IsSet(ClientFlags.InfoReceived))
{
Opts.Import = info.Import;
Opts.Export = info.Export;
}
Route.RemoteId = info.Id;
Route.RemoteName = info.Name;
Route.AuthRequired = info.AuthRequired;
Route.TlsRequired = info.TlsRequired;
Route.GatewayUrl = info.GatewayUrl ?? string.Empty;
Route.Lnoc = info.Lnoc;
Route.Lnocu = info.Lnocu;
Route.JetStream = info.JetStream;
Route.ConnectUrls = info.ClientConnectUrls?.ToList() ?? [];
Route.WsConnUrls = info.WsConnectUrls?.ToList() ?? [];
Route.LeafnodeUrl = info.LeafNodeUrls is { Length: 1 } leaf ? leaf[0] : string.Empty;
Route.Hash = NatsServer.GetHash(info.Name);
Route.IdHash = NatsServer.GetHash(info.Id);
Opts.Protocol = info.Proto;
Headers = server.SupportsHeaders() && info.Headers;
Flags |= ClientFlags.InfoReceived;
}
if (NatsServer.NeedsCompression(server.GetOpts().Cluster.Compression.Mode))
_ = server.NegotiateRouteCompression(this, Route?.DidSolicit == true, Route?.AccName is { Length: > 0 } an ? Encoding.ASCII.GetString(an) : string.Empty, info.Compression ?? string.Empty, server.GetOpts());
server.UpdateRemoteRoutePerms(this, info);
}
internal bool CanImport(string subject) => PubAllowedFullCheck(subject, fullCheck: false, hasLock: true);
internal bool CanExport(string subject) => CanSubscribe(subject);
internal void SetRoutePermissions(RoutePermissions? perms)
{
if (perms is null)
{
Perms = null;
MPerms = null;
return;
}
SetPermissions(new Permissions
{
Publish = perms.Import?.Clone(),
Subscribe = perms.Export?.Clone(),
});
}
internal (bool IsPinnedAccountRoute, string AccountName, bool KeyHasSubType) GetRoutedSubKeyInfo()
{
var accountName = Route?.AccName is { Length: > 0 } an
? Encoding.ASCII.GetString(an)
: string.Empty;
return (!string.IsNullOrEmpty(accountName), accountName, Route?.Lnocu == true);
}
internal void RemoveRemoteSubs()
{
if (Server is not NatsServer server)
return;
Dictionary<string, Subscription> subs;
var grouped = new Dictionary<string, List<Subscription>>(StringComparer.Ordinal);
var (pinned, accountName, keyHasSubType) = GetRoutedSubKeyInfo();
lock (_mu)
{
subs = Subs;
Subs = new Dictionary<string, Subscription>(StringComparer.Ordinal);
}
foreach (var kvp in subs)
{
var keyAccount = pinned
? accountName
: RouteHandler.GetAccNameFromRoutedSubKey(kvp.Value, kvp.Key, keyHasSubType);
if (string.IsNullOrEmpty(keyAccount))
continue;
if (!grouped.TryGetValue(keyAccount, out var list))
{
list = [];
grouped[keyAccount] = list;
}
list.Add(kvp.Value);
}
foreach (var (accName, list) in grouped)
{
var (acc, _) = server.LookupAccount(accName);
acc?.Sublist?.RemoveBatch(list);
}
}
internal List<Subscription> RemoveRemoteSubsForAcc(string name)
{
var removed = new List<Subscription>();
var (_, _, keyHasSubType) = GetRoutedSubKeyInfo();
lock (_mu)
{
foreach (var key in Subs.Keys.ToArray())
{
var sub = Subs[key];
if (RouteHandler.GetAccNameFromRoutedSubKey(sub, key, keyHasSubType) != name)
continue;
removed.Add(sub);
Subs.Remove(key);
}
}
return removed;
}
internal (byte[] Origin, string AccountName, byte[] Subject, byte[] Queue, Exception? Error)
ParseUnsubProto(byte[] arg, bool accInProto, bool hasOrigin)
{
_in.Subs++;
var args = SplitArg(arg);
var origin = Array.Empty<byte>();
var queue = Array.Empty<byte>();
var subjectIndex = 0;
if (hasOrigin)
{
if (args.Count == 0)
return (origin, string.Empty, Array.Empty<byte>(), queue, new FormatException($"parse error: '{Encoding.ASCII.GetString(arg)}'"));
origin = args[0];
subjectIndex = 1;
}
if (accInProto)
subjectIndex++;
if (args.Count is not (>= 1) || args.Count < subjectIndex + 1 || args.Count > subjectIndex + 2)
return (origin, string.Empty, Array.Empty<byte>(), queue, new FormatException($"parse error: '{Encoding.ASCII.GetString(arg)}'"));
if (args.Count == subjectIndex + 2)
queue = args[subjectIndex + 1];
var accountName = accInProto ? Encoding.ASCII.GetString(args[subjectIndex - 1]) : string.Empty;
return (origin, accountName, args[subjectIndex], queue, null);
}
internal Exception? ProcessRemoteUnsub(byte[] arg, bool leafUnsub)
{
if (Server is not NatsServer server)
return null;
string accountName;
var accInProto = true;
bool originSupport;
lock (_mu)
{
originSupport = Route?.Lnocu == true;
if (Route?.AccName is { Length: > 0 } an)
{
accountName = Encoding.ASCII.GetString(an);
accInProto = false;
}
else
{
accountName = string.Empty;
}
}
var (_, protoAccName, subject, _, err) = ParseUnsubProto(arg, accInProto, leafUnsub && originSupport);
if (err is not null)
return new FormatException($"processRemoteUnsub {err.Message}");
if (accInProto)
accountName = protoAccName;
var (acc, _) = server.LookupAccount(accountName);
if (acc is null)
{
Debugf("Unknown account {0} for subject {1}", accountName, Encoding.ASCII.GetString(subject));
return null;
}
Subscription? sub = null;
var key = Encoding.ASCII.GetString(arg);
lock (_mu)
{
if (IsClosed())
return null;
if (Subs.TryGetValue(key, out sub))
{
Subs.Remove(key);
acc.Sublist?.Remove(sub);
}
}
if (Opts.Verbose)
SendOK();
return null;
}
internal Exception? ProcessRemoteSub(byte[] protoArg, bool hasOrigin)
{
_in.Subs++;
if (Server is not NatsServer server)
return null;
var args = SplitArg(protoArg);
var (isPinned, accountName, _) = GetRoutedSubKeyInfo();
var accInProto = !isPinned;
var subjectIndex = 0;
if (hasOrigin)
subjectIndex++;
if (accInProto)
subjectIndex++;
if (args.Count is not (>= 1) || (args.Count != subjectIndex + 1 && args.Count != subjectIndex + 3))
return new FormatException($"processRemoteSub Parse Error: '{Encoding.ASCII.GetString(protoArg)}'");
if (accInProto)
accountName = Encoding.ASCII.GetString(args[subjectIndex - 1]);
var subject = args[subjectIndex];
byte[]? queue = null;
var qw = 1;
if (args.Count == subjectIndex + 3)
{
queue = args[subjectIndex + 1];
_ = int.TryParse(Encoding.ASCII.GetString(args[subjectIndex + 2]), out qw);
if (qw <= 0)
qw = 1;
}
var (acc, _) = server.LookupOrRegisterAccount(accountName);
if (acc is null)
return null;
lock (_mu)
{
if (IsClosed())
return null;
if (Perms is not null && !CanExport(Encoding.ASCII.GetString(subject)))
return null;
if (SubsAtLimit())
{
MaxSubsExceeded();
return null;
}
var key = Encoding.ASCII.GetString(protoArg);
if (!Subs.ContainsKey(key))
{
var sub = new Subscription
{
Subject = subject,
Queue = queue,
Sid = Encoding.ASCII.GetBytes(key),
Qw = qw,
};
Subs[key] = sub;
acc.Sublist?.Insert(sub);
}
}
if (Opts.Verbose)
SendOK();
return null;
}
internal byte[] AddRouteSubOrUnsubProtoToBuf(byte[] buf, string accName, Subscription sub, bool isSubProto)
{
var list = new List<byte>(buf.Length + 64);
list.AddRange(buf);
if (isSubProto)
list.AddRange(Encoding.ASCII.GetBytes("RS+ "));
else
list.AddRange(Encoding.ASCII.GetBytes("RS- "));
if (Route?.AccName is not { Length: > 0 })
{
list.AddRange(Encoding.ASCII.GetBytes(accName));
list.Add((byte)' ');
}
list.AddRange(sub.Subject);
if (sub.Queue is { Length: > 0 } queue)
{
list.Add((byte)' ');
list.AddRange(queue);
if (isSubProto)
{
list.Add((byte)' ');
list.AddRange(Encoding.ASCII.GetBytes(Math.Max(sub.Qw, 1).ToString()));
}
}
list.Add((byte)'\r');
list.Add((byte)'\n');
return list.ToArray();
}
internal void SendRouteSubProtos(IReadOnlyList<Subscription> subs, bool trace, Func<Subscription, bool>? filter = null) =>
SendRouteSubOrUnSubProtos(subs, isSubProto: true, trace, filter);
internal void SendRouteUnSubProtos(IReadOnlyList<Subscription> subs, bool trace, Func<Subscription, bool>? filter = null) =>
SendRouteSubOrUnSubProtos(subs, isSubProto: false, trace, filter);
internal void SendRouteSubOrUnSubProtos(
IReadOnlyList<Subscription> subs,
bool isSubProto,
bool trace,
Func<Subscription, bool>? filter = null)
{
var buf = Array.Empty<byte>();
foreach (var sub in subs)
{
if (filter is not null && !filter(sub))
continue;
var accountName = ServerConstants.DefaultGlobalAccount;
var startLen = buf.Length;
buf = AddRouteSubOrUnsubProtoToBuf(buf, accountName, sub, isSubProto);
if (trace && buf.Length > startLen)
TraceOutOp(string.Empty, buf.AsSpan(startLen, buf.Length - startLen - 2).ToArray());
}
if (buf.Length > 0)
EnqueueProto(buf);
}
internal bool ImportFilter(string subject) => CanImport(subject);
internal bool IsSolicitedRoute() => Route?.DidSolicit == true;
internal Exception? ProcessRouteConnect(byte[] arg)
{
if (arg is not { Length: > 0 })
return new FormatException("processRouteConnect parse error");
ConnectInfo? info;
try
{
info = JsonSerializer.Deserialize<ConnectInfo>(arg);
}
catch (Exception ex)
{
return ex;
}
if (info is null)
return new FormatException("processRouteConnect missing CONNECT payload");
lock (_mu)
{
Opts.Name = info.Name;
Opts.Headers = info.Headers;
Route ??= new Route();
Route.Lnoc = info.Lnoc;
Route.Lnocu = info.Lnocu;
}
return null;
}
}
@@ -112,6 +112,7 @@ public sealed partial class ClientConnection
// Client options (from CONNECT message). // Client options (from CONNECT message).
internal ClientOptions Opts = ClientOptions.Default; internal ClientOptions Opts = ClientOptions.Default;
internal Route? Route;
// Flags and state. // Flags and state.
internal ClientFlags Flags; // mirrors c.flags clientFlag internal ClientFlags Flags; // mirrors c.flags clientFlag
@@ -1686,7 +1687,7 @@ public sealed partial class ClientConnection
} }
} }
internal void ProcessUnsub(byte[] sid) internal void RemoveSubBySid(byte[] sid)
{ {
lock (_mu) lock (_mu)
{ {
@@ -0,0 +1,248 @@
// Copyright 2012-2026 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
using System.Security.Cryptography;
using System.Text;
using System.Linq;
namespace ZB.MOM.NatsNet.Server;
public sealed partial class NatsServer
{
internal byte[] GenerateRouteInitialInfoJSON(string accName, string compression, int poolIdx, byte gossipMode)
{
var info = _routeInfo.ShallowClone();
Span<byte> nonce = stackalloc byte[16];
RandomNumberGenerator.Fill(nonce);
info.Nonce = Convert.ToBase64String(nonce);
info.RouteAccount = string.IsNullOrEmpty(accName) ? null : accName;
info.RoutePoolIdx = poolIdx;
info.GossipMode = gossipMode;
info.Compression = CompressionModeForInfoProtocol(GetOpts().Cluster.Compression, compression);
return GenerateInfoJson(info);
}
internal bool AddRoute(ClientConnection route, bool didSolicit, bool sendDelayedInfo, byte gossipMode, ServerInfo info, string accName)
{
_mu.EnterWriteLock();
try
{
route.Route ??= new Route();
route.Route.RemoteId = info.Id;
route.Route.DidSolicit = didSolicit;
route.Route.RemoteName = info.Name;
if (!string.IsNullOrEmpty(accName))
route.Route.AccName = Encoding.ASCII.GetBytes(accName);
if (!_routes.TryGetValue(info.Id, out var pool))
{
pool = [];
_routes[info.Id] = pool;
}
else
{
foreach (var existing in pool.ToArray())
{
if (!RouteHandler.HandleDuplicateRoute(existing, route))
return false;
pool.Remove(existing);
}
}
route.Route.PoolIdx = pool.Count;
pool.Add(route);
if (sendDelayedInfo)
route.EnqueueProto(GenerateRouteInitialInfoJSON(accName, info.Compression ?? string.Empty, route.Route.PoolIdx, gossipMode));
ForwardNewRouteInfoToKnownServers(info, didSolicit ? RouteType.Explicit : RouteType.Implicit, didSolicit, gossipMode);
return true;
}
finally
{
_mu.ExitWriteLock();
}
}
internal Exception? StartRouteAcceptLoop()
{
if (_routeListener == null)
return null;
if (!StartGoRoutine(() => Noticef("Route accept loop started")))
return new InvalidOperationException("unable to start route accept loop");
return null;
}
internal Exception? SetRouteInfoHostPortAndIP()
{
var opts = GetOpts();
string host;
int port;
if (!string.IsNullOrWhiteSpace(opts.Cluster.Advertise))
{
var (advHost, advPort, advErr) = Internal.ServerUtilities.ParseHostPort(opts.Cluster.Advertise, opts.Cluster.Port);
if (advErr != null)
return new InvalidOperationException($"Cluster.Advertise invalid: {opts.Cluster.Advertise}", advErr);
host = advHost;
port = advPort;
}
else
{
host = opts.Cluster.Host;
if (string.IsNullOrWhiteSpace(host))
host = opts.Host;
port = opts.Cluster.Port;
}
_mu.EnterWriteLock();
try
{
_routeInfo.Host = host;
_routeInfo.Port = port;
_routeInfo.Ip = $"nats-route://{host}:{port}/";
return null;
}
finally
{
_mu.ExitWriteLock();
}
}
public Exception? StartRouting()
{
var err = SetRouteInfoHostPortAndIP();
if (err != null)
return err;
SolicitRoutes();
return StartRouteAcceptLoop();
}
internal void ReConnectToRoute(Uri routeUrl, string accName = "")
{
StartGoRoutine(() =>
{
_ = ConnectToRoute(routeUrl, RouteType.Explicit, false, GossipMode.Default, accName);
});
}
internal bool RouteStillValid(Uri routeUrl)
{
var opts = GetOpts();
return opts.Routes.Any(r => string.Equals(r.Host, routeUrl.Host, StringComparison.OrdinalIgnoreCase) && r.Port == routeUrl.Port);
}
internal Exception? ConnectToRoute(Uri routeUrl, RouteType routeType, bool firstConnect, byte gossipMode, string accName)
{
_ = firstConnect;
if (!RouteStillValid(routeUrl))
return new InvalidOperationException($"route is no longer configured: {routeUrl}");
SaveRouteTLSName(routeUrl);
var route = CreateRoute(null, routeUrl, routeType, gossipMode, accName);
if (route is null)
return new InvalidOperationException("failed to create route");
return null;
}
internal bool SaveRouteTLSName(Uri routeUrl)
{
if (routeUrl is null || string.IsNullOrWhiteSpace(routeUrl.Host))
return false;
_mu.EnterWriteLock();
try
{
_routeTlsName = routeUrl.Host;
return true;
}
finally
{
_mu.ExitWriteLock();
}
}
internal void SolicitRoutes()
{
foreach (var route in GetOpts().Routes)
_ = ConnectToRoute(route, RouteType.Explicit, true, GossipMode.Default, string.Empty);
}
internal void RemoveAllRoutesExcept(string remoteId)
{
_mu.EnterWriteLock();
try
{
foreach (var (rid, routes) in _routes.ToArray())
{
if (rid == remoteId)
continue;
foreach (var route in routes)
route.CloseConnection(ClosedState.RouteRemoved);
_routes.Remove(rid);
}
}
finally
{
_mu.ExitWriteLock();
}
}
internal bool IsDuplicateServerName(string serverName)
{
var duplicate = false;
_mu.EnterReadLock();
try
{
ForEachRoute(route =>
{
if (route.Route?.RemoteName == serverName)
duplicate = true;
});
}
finally
{
_mu.ExitReadLock();
}
return duplicate;
}
internal void ForEachNonPerAccountRoute(Func<ClientConnection, bool> fn)
{
_mu.EnterReadLock();
try
{
foreach (var route in _routes.Values.SelectMany(v => v))
{
if (route.Route?.AccName is { Length: > 0 })
continue;
if (!fn(route))
break;
}
}
finally
{
_mu.ExitReadLock();
}
}
internal void ForEachRouteIdx(int idx, Func<ClientConnection, bool> fn)
{
_mu.EnterReadLock();
try
{
foreach (var pool in _routes.Values)
{
if (idx < 0 || idx >= pool.Count)
continue;
if (!fn(pool[idx]))
break;
}
}
finally
{
_mu.ExitReadLock();
}
}
}
@@ -0,0 +1,183 @@
// Copyright 2012-2026 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
using System.Text.Json;
using ZB.MOM.NatsNet.Server.Auth;
using ZB.MOM.NatsNet.Server.Internal;
namespace ZB.MOM.NatsNet.Server;
public sealed partial class NatsServer
{
internal bool NegotiateRouteCompression(
ClientConnection c,
bool didSolicit,
string accName,
string infoCompression,
ServerOptions opts)
{
var mode = SelectCompressionMode(opts.Cluster.Compression.Mode, infoCompression);
lock (c)
{
c.Route ??= new Route();
if (mode == CompressionMode.S2Auto)
{
if (c.Rtt == TimeSpan.Zero)
c.Rtt = ClientConnection.ComputeRtt(c.Start);
mode = SelectS2AutoModeBasedOnRtt(c.Rtt, opts.Cluster.Compression.RttThresholds);
}
c.Route.Compression = mode;
}
if (!NeedsCompression(mode))
return false;
var info = CopyInfo();
info.Compression = CompressionModeForInfoProtocol(opts.Cluster.Compression, mode);
if (!string.IsNullOrEmpty(accName))
info.RouteAccount = accName;
var proto = GenerateInfoJson(info);
lock (c)
{
if (didSolicit)
c.EnqueueProto(proto);
else
c.EnqueueProto(proto);
c.SetFirstPingTimer();
}
return true;
}
internal void UpdateRemoteRoutePerms(ClientConnection c, ServerInfo info)
{
SubjectPermission? oldExport;
lock (c)
{
oldExport = c.Opts.Export?.Clone();
c.Opts.Import = info.Import?.Clone();
c.Opts.Export = info.Export?.Clone();
}
// Build old/new export checkers to preserve route permission semantics.
var oldTester = new ClientConnection(ClientKind.Router) { Route = new Route() };
oldTester.SetRoutePermissions(new RoutePermissions { Export = oldExport });
var newTester = new ClientConnection(ClientKind.Router) { Route = new Route() };
newTester.SetRoutePermissions(new RoutePermissions { Export = info.Export?.Clone() });
if (oldExport is null && info.Export is null)
return;
// Subscription fanout wiring for route permission delta is completed in group 2.
_ = oldTester;
_ = newTester;
}
internal void ProcessImplicitRoute(ServerInfo info, bool routeNoPool)
{
var remoteId = info.Id;
_mu.EnterWriteLock();
try
{
if (remoteId == _info.Id)
return;
var opts = GetOpts();
if (!string.IsNullOrEmpty(info.RouteAccount))
{
if (opts.Cluster.PoolSize <= 0)
return;
if (_accRoutes != null
&& _accRoutes.TryGetValue(info.RouteAccount, out var remotes)
&& remotes.TryGetValue(remoteId, out var existing)
&& existing != null)
{
return;
}
}
else if (_routes.ContainsKey(remoteId))
{
return;
}
if (HasThisRouteConfigured(info))
return;
}
finally
{
_mu.ExitWriteLock();
}
if (!Uri.TryCreate(info.Ip, UriKind.Absolute, out var _))
return;
if (routeNoPool && string.IsNullOrEmpty(info.RouteAccount))
Debugf("Implicit route from non-pooling remote {0} processed", info.Id);
}
internal bool HasThisRouteConfigured(ServerInfo info)
{
var routes = GetOpts().Routes;
if (routes.Count == 0)
return false;
var infoPort = info.Port <= 0 ? 6222 : info.Port;
var primary = $"{info.Host}:{infoPort}".ToLowerInvariant();
var secondary = string.Empty;
if (!string.IsNullOrWhiteSpace(info.Ip) && Uri.TryCreate(info.Ip, UriKind.Absolute, out var infoUri))
{
var host = infoUri.Host;
if (!string.IsNullOrWhiteSpace(host))
secondary = $"{host}:{infoPort}".ToLowerInvariant();
}
foreach (var route in routes)
{
var routePort = route.IsDefaultPort ? infoPort : route.Port;
var hostPort = $"{route.Host}:{routePort}".ToLowerInvariant();
if (hostPort == primary || (!string.IsNullOrEmpty(secondary) && hostPort == secondary))
return true;
}
return false;
}
internal void ForwardNewRouteInfoToKnownServers(ServerInfo info, RouteType routeType, bool didSolicit, byte localGossipMode)
{
var fromGossip = didSolicit && routeType == RouteType.Implicit;
if ((fromGossip && localGossipMode != GossipMode.Override) || info.GossipMode == GossipMode.Disabled)
return;
info.Nonce = string.Empty;
byte[] BuildInfo(byte gossipMode)
{
info.GossipMode = gossipMode;
return GenerateInfoJson(info);
}
byte[]? infoDefault = null;
byte[]? infoDisabled = null;
byte[]? infoOverride = null;
byte[] SelectInfo(ClientConnection route)
{
var rType = route.Route?.RouteType ?? RouteType.Implicit;
if ((!didSolicit && rType == RouteType.Explicit) || (didSolicit && routeType == RouteType.Explicit))
return infoOverride ??= BuildInfo(GossipMode.Override);
if (!didSolicit)
return infoDisabled ??= BuildInfo(GossipMode.Disabled);
return infoDefault ??= BuildInfo(GossipMode.Default);
}
ForEachRemote(route =>
{
if (route.Route?.RemoteId == info.Id)
return;
route.EnqueueProto(SelectInfo(route));
});
}
}
@@ -0,0 +1,74 @@
// Copyright 2012-2026 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
using ZB.MOM.NatsNet.Server.Internal;
using System.IO;
namespace ZB.MOM.NatsNet.Server;
public sealed partial class NatsServer
{
internal void SendSubsToRoute(ClientConnection route, int idx, string account)
{
if (route == null)
return;
var allSubs = new List<Subscription>(1024);
if (idx < 0 || !string.IsNullOrEmpty(account))
{
var (acc, _) = LookupAccount(account);
acc?.Sublist?.LocalSubs(allSubs, includeLeafHubs: false);
}
else
{
foreach (var acc in _accounts.Values)
{
if (acc.RoutePoolIdx != idx)
continue;
acc.Sublist?.LocalSubs(allSubs, includeLeafHubs: false);
}
}
route.SendRouteSubProtos(allSubs, trace: false, sub => route.CanImport(System.Text.Encoding.ASCII.GetString(sub.Subject)));
}
internal ClientConnection? CreateRoute(Stream? conn, Uri? routeUrl, RouteType routeType, byte gossipMode, string accName)
{
var opts = GetOpts();
var didSolicit = routeUrl != null;
var c = new ClientConnection(ClientKind.Router, this, conn ?? new MemoryStream())
{
Opts = ClientOptions.Default,
Route = new Route
{
Url = routeUrl,
RouteType = routeType,
DidSolicit = didSolicit,
PoolIdx = -1,
GossipMode = gossipMode,
AccName = string.IsNullOrEmpty(accName) ? null : System.Text.Encoding.ASCII.GetBytes(accName),
},
Start = DateTime.UtcNow,
};
lock (c)
{
c.InitClient();
if (didSolicit)
c.SetRoutePermissions(opts.Cluster.Permissions);
c.SetFirstPingTimer();
}
if (didSolicit)
{
var sendErr = c.SendRouteConnect(_info.Cluster ?? string.Empty, _routeInfo.TlsRequired);
if (sendErr != null)
{
c.CloseConnection(ClosedState.ProtocolViolation);
return null;
}
}
return c;
}
}
@@ -254,6 +254,7 @@ public sealed partial class NatsServer : INatsServer
private readonly ConcurrentDictionary<string, object?> _nodeToInfo = new(StringComparer.Ordinal); private readonly ConcurrentDictionary<string, object?> _nodeToInfo = new(StringComparer.Ordinal);
private readonly ConcurrentDictionary<string, object?> _raftNodes = new(StringComparer.Ordinal); private readonly ConcurrentDictionary<string, object?> _raftNodes = new(StringComparer.Ordinal);
private readonly Dictionary<string, string> _routesToSelf = []; private readonly Dictionary<string, string> _routesToSelf = [];
private string _routeTlsName = string.Empty;
private INetResolver? _routeResolver; private INetResolver? _routeResolver;
private readonly ConcurrentDictionary<string, object?> _rateLimitLogging = new(); private readonly ConcurrentDictionary<string, object?> _rateLimitLogging = new();
private readonly Channel<TimeSpan> _rateLimitLoggingCh; private readonly Channel<TimeSpan> _rateLimitLoggingCh;
@@ -0,0 +1,60 @@
// Copyright 2012-2026 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
namespace ZB.MOM.NatsNet.Server;
internal static class RouteHandler
{
internal static int ComputeRoutePoolIdx(int poolSize, string accountName) =>
NatsServer.ComputeRoutePoolIdx(poolSize, accountName);
internal static string GetAccNameFromRoutedSubKey(Internal.Subscription sub, string key, bool keyHasSubType)
{
_ = sub;
var fields = key.Split(' ', StringSplitOptions.RemoveEmptyEntries);
if (fields.Length == 0)
return string.Empty;
var accountIndex = keyHasSubType ? 1 : 0;
if (accountIndex >= fields.Length)
return string.Empty;
return fields[accountIndex];
}
internal static bool RouteShouldDelayInfo(string accName, ServerOptions opts) =>
string.IsNullOrEmpty(accName) && opts.Cluster.PoolSize >= 1;
internal static bool HasSolicitedRoute(IReadOnlyList<ClientConnection> routes, string accName)
{
foreach (var route in routes)
{
if (route.Route?.DidSolicit != true)
continue;
var routeAcc = route.Route?.AccName is { Length: > 0 } an
? System.Text.Encoding.ASCII.GetString(an)
: string.Empty;
if (routeAcc == accName)
return true;
}
return false;
}
internal static void UpgradeRouteToSolicited(ClientConnection route)
{
if (route.Route is null)
return;
route.Route.DidSolicit = true;
route.Route.Retry = true;
}
internal static bool HandleDuplicateRoute(ClientConnection existing, ClientConnection incoming)
{
if (existing.IsSolicitedRoute() && !incoming.IsSolicitedRoute())
return false;
if (!existing.IsSolicitedRoute() && incoming.IsSolicitedRoute())
return true;
return incoming.Cid > existing.Cid;
}
}
@@ -308,7 +308,9 @@ public sealed class ClientConnectionStubFeaturesTests
GetTimer(c, "_pingTimer").ShouldNotBeNull(); GetTimer(c, "_pingTimer").ShouldNotBeNull();
c.WatchForStaleConnection(TimeSpan.FromMilliseconds(20), pingMax: 0); c.WatchForStaleConnection(TimeSpan.FromMilliseconds(20), pingMax: 0);
Thread.Sleep(60); var staleDeadline = DateTime.UtcNow.AddMilliseconds(500);
while (!c.IsClosed() && DateTime.UtcNow < staleDeadline)
Thread.Sleep(10);
c.IsClosed().ShouldBeTrue(); c.IsClosed().ShouldBeTrue();
var temp = Account.NewAccount("A"); var temp = Account.NewAccount("A");
@@ -1,4 +1,5 @@
using System.IO; using System.IO;
using System.Text;
using Shouldly; using Shouldly;
using ZB.MOM.NatsNet.Server; using ZB.MOM.NatsNet.Server;
using ZB.MOM.NatsNet.Server.Internal; using ZB.MOM.NatsNet.Server.Internal;
@@ -7,6 +8,65 @@ namespace ZB.MOM.NatsNet.Server.Tests.ImplBacklog;
public sealed partial class RouteHandlerTests public sealed partial class RouteHandlerTests
{ {
[Fact] // T:2798
public void ClusterAdvertiseErrorOnStartup_ShouldSucceed()
{
var options = new ServerOptions();
options.Cluster.Advertise = "addr:::123";
var (server, err) = NatsServer.NewServer(options);
err.ShouldBeNull();
var startErr = server!.StartRouting();
startErr.ShouldNotBeNull();
startErr!.Message.ShouldContain("Cluster.Advertise");
}
[Fact] // T:2822
public void TLSRoutesCertificateImplicitAllowPass_ShouldSucceed()
{
var client = new ClientConnection(ClientKind.Router, nc: new MemoryStream());
client.MatchesPinnedCert(null).ShouldBeTrue();
}
[Fact] // T:2823
public void TLSRoutesCertificateImplicitAllowFail_ShouldSucceed()
{
var client = new ClientConnection(ClientKind.Router, nc: new MemoryStream());
var pinned = new PinnedCertSet([new string('a', 64)]);
client.MatchesPinnedCert(pinned).ShouldBeFalse();
}
[Fact] // T:2844
public void RouteParseOriginClusterMsgArgs_ShouldSucceed()
{
var c = new ClientConnection(ClientKind.Router)
{
Route = new Route { AccName = "MY_ACCOUNT"u8.ToArray() },
};
var err = c.ProcessRoutedOriginClusterMsgArgs("ORIGIN foo + bar queue1 queue2 12 345\r\n"u8.ToArray());
err.ShouldBeNull();
Encoding.ASCII.GetString(c.ParseCtx.Pa.Account!).ShouldBe("ORIGIN");
Encoding.ASCII.GetString(c.ParseCtx.Pa.Subject!).ShouldBe("foo");
Encoding.ASCII.GetString(c.ParseCtx.Pa.Reply!).ShouldBe("bar");
c.ParseCtx.Pa.Queues.ShouldNotBeNull();
c.ParseCtx.Pa.Queues!.Count.ShouldBe(3);
c.ParseCtx.Pa.Size.ShouldBe(345);
}
[Fact] // T:2850
public void RouteCompression_ShouldSucceed()
{
var opts = new ServerOptions();
opts.Cluster.Compression.Mode = CompressionMode.S2Fast;
var (server, err) = NatsServer.NewServer(opts);
err.ShouldBeNull();
server.ShouldNotBeNull();
var infoProto = server!.GenerateRouteInitialInfoJSON(string.Empty, CompressionMode.S2Fast, 0, GossipMode.Default);
infoProto.Length.ShouldBeGreaterThan(0);
Encoding.ASCII.GetString(infoProto).ShouldContain("\"compression\":\"s2_fast\"");
}
[Fact] // T:2819 [Fact] // T:2819
public async Task RouteIPResolutionAndRouteToSelf_ShouldSucceed() public async Task RouteIPResolutionAndRouteToSelf_ShouldSucceed()
{ {
BIN
View File
Binary file not shown.
+6 -6
View File
@@ -1,6 +1,6 @@
# NATS .NET Porting Status Report # NATS .NET Porting Status Report
Generated: 2026-03-01 01:56:16 UTC Generated: 2026-03-01 02:33:59 UTC
## Modules (12 total) ## Modules (12 total)
@@ -13,18 +13,18 @@ Generated: 2026-03-01 01:56:16 UTC
| Status | Count | | Status | Count |
|--------|-------| |--------|-------|
| complete | 22 | | complete | 22 |
| deferred | 1519 | | deferred | 1467 |
| n_a | 24 | | n_a | 24 |
| stub | 1 | | stub | 1 |
| verified | 2107 | | verified | 2159 |
## Unit Tests (3257 total) ## Unit Tests (3257 total)
| Status | Count | | Status | Count |
|--------|-------| |--------|-------|
| deferred | 1595 | | deferred | 1590 |
| n_a | 254 | | n_a | 254 |
| verified | 1408 | | verified | 1413 |
## Library Mappings (36 total) ## Library Mappings (36 total)
@@ -35,4 +35,4 @@ Generated: 2026-03-01 01:56:16 UTC
## Overall Progress ## Overall Progress
**3827/6942 items complete (55.1%)** **3884/6942 items complete (55.9%)**