dohertj2/natsdotnet
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add closed connection ring buffer for /connz?state=closed (Gap 10.1)

Browse Source 
Replace ConcurrentQueue<ClosedClient> with ClosedConnectionRingBuffer — a
fixed-size ring buffer that overwrites oldest entries when full, eliminating
the manual dequeue-to-cap loop. Adds TotalClosed lifetime counter, GetRecent(),
and Clear(). Wires the ring buffer into NatsServer including capacity resize
on config reload. Adds 10 unit tests covering capacity, ordering, wrapping,
TotalClosed tracking, and Clear behavior.
This commit is contained in:
Joseph Doherty
2026-02-25 13:07:16 -05:00
parent 9fb2ae205c
commit 9ece600ebc
3 changed files with 354 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
src/NATS.Server/NatsServer.cs
View File

@@ -32,7 +32,7 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
{
private readonly NatsOptions _options;
private readonly ConcurrentDictionary<ulong, NatsClient> _clients = new();
private readonly ConcurrentQueue<ClosedClient> _closedClients = new();
private ClosedConnectionRingBuffer _closedClients;
private readonly ServerInfo _serverInfo;
private readonly ILogger<NatsServer> _logger;
private readonly ILoggerFactory _loggerFactory;
@@ -118,7 +118,7 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
public Action? ReOpenLogFile { get; set; }
public IEnumerable<NatsClient> GetClients() => _clients.Values;
public IEnumerable<ClosedClient> GetClosedClients() => _closedClients;
public IReadOnlyList<ClosedClient> GetClosedClients() => _closedClients.GetAll();
public IEnumerable<Auth.Account> GetAccounts() => _accounts.Values;
public bool HasRemoteInterest(string subject) => _globalAccount.SubList.HasRemoteInterest(subject);
@@ -364,6 +364,7 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
_options = options;
_loggerFactory = loggerFactory;
_logger = loggerFactory.CreateLogger<NatsServer>();
_closedClients = new ClosedConnectionRingBuffer(options.MaxClosedClients);
_authService = AuthService.Build(options);
_globalAccount = new Account(Account.GlobalAccountName);
_accounts[Account.GlobalAccountName] = _globalAccount;
@@ -1509,8 +1510,8 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
_clients.TryRemove(client.Id, out _);
_logger.LogDebug("Removed client {ClientId}", client.Id);
// Snapshot for closed-connections tracking
_closedClients.Enqueue(new ClosedClient
// Snapshot for closed-connections tracking (ring buffer auto-overwrites oldest when full)
_closedClients.Add(new ClosedClient
{
Cid = client.Id,
Ip = client.RemoteIp ?? "",
@@ -1538,10 +1539,6 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
Proxy = client.ClientOpts?.Username?.StartsWith("proxy:", StringComparison.Ordinal) == true ? "true" : "",
});
// Cap closed clients list
while (_closedClients.Count > _options.MaxClosedClients)
_closedClients.TryDequeue(out _);
var subList = client.Account?.SubList ?? _globalAccount.SubList;
client.RemoveAllSubscriptions(subList);
client.Account?.RemoveClient(client.Id);
@@ -1830,7 +1827,11 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
_options.MaxSubs = newOpts.MaxSubs;
_options.MaxSubTokens = newOpts.MaxSubTokens;
_options.MaxTracedMsgLen = newOpts.MaxTracedMsgLen;
_options.MaxClosedClients = newOpts.MaxClosedClients;
if (newOpts.MaxClosedClients != _options.MaxClosedClients)
{
_options.MaxClosedClients = newOpts.MaxClosedClients;
_closedClients = new ClosedConnectionRingBuffer(newOpts.MaxClosedClients);
}
// TLS
_options.TlsCert = newOpts.TlsCert;