Merge branch 'feature/core-lifecycle' into main
Reconcile close reason tracking: feature branch's MarkClosed() and ShouldSkipFlush/FlushAndCloseAsync now use main's ClientClosedReason enum. ClosedState enum retained for forward compatibility.
This commit is contained in:
Joseph Doherty
@@ -213,6 +213,39 @@ public class ServerTests : IAsyncLifetime
|
||||
}
|
||||
}
|
||||
|
||||
public class EphemeralPortTests
|
||||
{
|
||||
[Fact]
|
||||
public async Task Server_resolves_ephemeral_port()
|
||||
{
|
||||
using var cts = new CancellationTokenSource();
|
||||
var server = new NatsServer(new NatsOptions { Port = 0 }, NullLoggerFactory.Instance);
|
||||
|
||||
_ = server.StartAsync(cts.Token);
|
||||
await server.WaitForReadyAsync();
|
||||
|
||||
try
|
||||
{
|
||||
// Port should have been resolved to a real port
|
||||
server.Port.ShouldBeGreaterThan(0);
|
||||
|
||||
// Connect a raw socket to prove the port actually works
|
||||
using var client = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
await client.ConnectAsync(IPAddress.Loopback, server.Port);
|
||||
|
||||
var buf = new byte[4096];
|
||||
var n = await client.ReceiveAsync(buf, SocketFlags.None);
|
||||
var response = Encoding.ASCII.GetString(buf, 0, n);
|
||||
response.ShouldStartWith("INFO ");
|
||||
}
|
||||
finally
|
||||
{
|
||||
await cts.CancelAsync();
|
||||
server.Dispose();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public class MaxConnectionsTests : IAsyncLifetime
|
||||
{
|
||||
private readonly NatsServer _server;
|
||||
@@ -423,3 +456,448 @@ public class PingKeepaliveTests : IAsyncLifetime
|
||||
client.Dispose();
|
||||
}
|
||||
}
|
||||
|
||||
public class CloseReasonTests : IAsyncLifetime
|
||||
{
|
||||
private readonly NatsServer _server;
|
||||
private readonly int _port;
|
||||
private readonly CancellationTokenSource _cts = new();
|
||||
|
||||
public CloseReasonTests()
|
||||
{
|
||||
_port = GetFreePort();
|
||||
_server = new NatsServer(new NatsOptions { Port = _port }, NullLoggerFactory.Instance);
|
||||
}
|
||||
|
||||
public async Task InitializeAsync()
|
||||
{
|
||||
_ = _server.StartAsync(_cts.Token);
|
||||
await _server.WaitForReadyAsync();
|
||||
}
|
||||
|
||||
public async Task DisposeAsync()
|
||||
{
|
||||
await _cts.CancelAsync();
|
||||
_server.Dispose();
|
||||
}
|
||||
|
||||
private static int GetFreePort()
|
||||
{
|
||||
using var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
sock.Bind(new IPEndPoint(IPAddress.Loopback, 0));
|
||||
return ((IPEndPoint)sock.LocalEndPoint!).Port;
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Client_close_reason_set_on_normal_disconnect()
|
||||
{
|
||||
// Connect a raw TCP client
|
||||
using var client = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
await client.ConnectAsync(IPAddress.Loopback, _port);
|
||||
|
||||
// Read INFO
|
||||
var buf = new byte[4096];
|
||||
await client.ReceiveAsync(buf, SocketFlags.None);
|
||||
|
||||
// Send CONNECT + PING, wait for PONG
|
||||
await client.SendAsync(Encoding.ASCII.GetBytes("CONNECT {}\r\nPING\r\n"));
|
||||
var sb = new StringBuilder();
|
||||
using var readCts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
while (!sb.ToString().Contains("PONG"))
|
||||
{
|
||||
var n = await client.ReceiveAsync(buf, SocketFlags.None, readCts.Token);
|
||||
if (n == 0) break;
|
||||
sb.Append(Encoding.ASCII.GetString(buf, 0, n));
|
||||
}
|
||||
sb.ToString().ShouldContain("PONG");
|
||||
|
||||
// Get the NatsClient from the server
|
||||
var natsClient = _server.GetClients().First();
|
||||
|
||||
// Close the TCP socket (normal client disconnect)
|
||||
client.Shutdown(SocketShutdown.Both);
|
||||
client.Close();
|
||||
|
||||
// Wait for the server to detect the disconnect
|
||||
await Task.Delay(500);
|
||||
|
||||
// The close reason should be ClientClosed (normal disconnect falls through to finally)
|
||||
natsClient.CloseReason.ShouldBe(ClientClosedReason.ClientClosed);
|
||||
}
|
||||
}
|
||||
|
||||
public class ServerIdentityTests
|
||||
{
|
||||
[Fact]
|
||||
public void Server_creates_system_account()
|
||||
{
|
||||
var server = new NatsServer(new NatsOptions { Port = 0 }, NullLoggerFactory.Instance);
|
||||
server.SystemAccount.ShouldNotBeNull();
|
||||
server.SystemAccount.Name.ShouldBe("$SYS");
|
||||
server.Dispose();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Server_generates_nkey_identity()
|
||||
{
|
||||
var server = new NatsServer(new NatsOptions { Port = 0 }, NullLoggerFactory.Instance);
|
||||
server.ServerNKey.ShouldNotBeNullOrEmpty();
|
||||
// Server NKey public keys start with 'N'
|
||||
server.ServerNKey[0].ShouldBe('N');
|
||||
server.Dispose();
|
||||
}
|
||||
}
|
||||
|
||||
public class FlushBeforeCloseTests
|
||||
{
|
||||
private static int GetFreePort()
|
||||
{
|
||||
using var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
sock.Bind(new IPEndPoint(IPAddress.Loopback, 0));
|
||||
return ((IPEndPoint)sock.LocalEndPoint!).Port;
|
||||
}
|
||||
|
||||
private static async Task<string> ReadUntilAsync(Socket sock, string expected, int timeoutMs = 5000)
|
||||
{
|
||||
using var cts = new CancellationTokenSource(timeoutMs);
|
||||
var sb = new StringBuilder();
|
||||
var buf = new byte[4096];
|
||||
while (!sb.ToString().Contains(expected))
|
||||
{
|
||||
var n = await sock.ReceiveAsync(buf, SocketFlags.None, cts.Token);
|
||||
if (n == 0) break;
|
||||
sb.Append(Encoding.ASCII.GetString(buf, 0, n));
|
||||
}
|
||||
return sb.ToString();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Shutdown_flushes_pending_data_to_clients()
|
||||
{
|
||||
var port = GetFreePort();
|
||||
var server = new NatsServer(new NatsOptions { Port = port }, NullLoggerFactory.Instance);
|
||||
_ = server.StartAsync(CancellationToken.None);
|
||||
await server.WaitForReadyAsync();
|
||||
|
||||
try
|
||||
{
|
||||
// Connect a subscriber via raw socket
|
||||
using var sub = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
await sub.ConnectAsync(IPAddress.Loopback, port);
|
||||
|
||||
// Read INFO
|
||||
var buf = new byte[4096];
|
||||
await sub.ReceiveAsync(buf, SocketFlags.None);
|
||||
|
||||
// Subscribe to "foo"
|
||||
await sub.SendAsync(Encoding.ASCII.GetBytes("CONNECT {}\r\nSUB foo 1\r\nPING\r\n"));
|
||||
var pong = await ReadUntilAsync(sub, "PONG");
|
||||
pong.ShouldContain("PONG");
|
||||
|
||||
// Connect a publisher
|
||||
using var pub = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
await pub.ConnectAsync(IPAddress.Loopback, port);
|
||||
await pub.ReceiveAsync(buf, SocketFlags.None); // INFO
|
||||
|
||||
// Publish "Hello" to "foo"
|
||||
await pub.SendAsync(Encoding.ASCII.GetBytes("CONNECT {}\r\nPUB foo 5\r\nHello\r\n"));
|
||||
|
||||
// Wait briefly for delivery
|
||||
await Task.Delay(200);
|
||||
|
||||
// Read from subscriber to verify MSG was received
|
||||
var msg = await ReadUntilAsync(sub, "Hello\r\n");
|
||||
msg.ShouldContain("MSG foo 1 5\r\nHello\r\n");
|
||||
}
|
||||
finally
|
||||
{
|
||||
await server.ShutdownAsync();
|
||||
server.Dispose();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public class GracefulShutdownTests
|
||||
{
|
||||
private static int GetFreePort()
|
||||
{
|
||||
using var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
sock.Bind(new IPEndPoint(IPAddress.Loopback, 0));
|
||||
return ((IPEndPoint)sock.LocalEndPoint!).Port;
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ShutdownAsync_disconnects_all_clients()
|
||||
{
|
||||
var port = GetFreePort();
|
||||
var server = new NatsServer(new NatsOptions { Port = port }, NullLoggerFactory.Instance);
|
||||
_ = server.StartAsync(CancellationToken.None);
|
||||
await server.WaitForReadyAsync();
|
||||
|
||||
// Connect 2 raw TCP clients
|
||||
using var client1 = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
await client1.ConnectAsync(IPAddress.Loopback, port);
|
||||
var buf = new byte[4096];
|
||||
await client1.ReceiveAsync(buf, SocketFlags.None); // INFO
|
||||
|
||||
using var client2 = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
await client2.ConnectAsync(IPAddress.Loopback, port);
|
||||
await client2.ReceiveAsync(buf, SocketFlags.None); // INFO
|
||||
|
||||
// Send CONNECT so both are registered
|
||||
await client1.SendAsync(Encoding.ASCII.GetBytes("CONNECT {}\r\nPING\r\n"));
|
||||
await client2.SendAsync(Encoding.ASCII.GetBytes("CONNECT {}\r\nPING\r\n"));
|
||||
|
||||
// Wait for PONG from both (confirming they are registered)
|
||||
using var readCts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
await client1.ReceiveAsync(buf, SocketFlags.None, readCts.Token);
|
||||
await client2.ReceiveAsync(buf, SocketFlags.None, readCts.Token);
|
||||
|
||||
server.ClientCount.ShouldBe(2);
|
||||
|
||||
await server.ShutdownAsync();
|
||||
|
||||
server.ClientCount.ShouldBe(0);
|
||||
server.Dispose();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task WaitForShutdown_blocks_until_shutdown()
|
||||
{
|
||||
var port = GetFreePort();
|
||||
var server = new NatsServer(new NatsOptions { Port = port }, NullLoggerFactory.Instance);
|
||||
_ = server.StartAsync(CancellationToken.None);
|
||||
await server.WaitForReadyAsync();
|
||||
|
||||
// Start WaitForShutdown in background
|
||||
var waitTask = Task.Run(() => server.WaitForShutdown());
|
||||
|
||||
// Give it a moment -- it should NOT complete yet
|
||||
await Task.Delay(200);
|
||||
waitTask.IsCompleted.ShouldBeFalse();
|
||||
|
||||
// Trigger shutdown
|
||||
await server.ShutdownAsync();
|
||||
|
||||
// WaitForShutdown should complete within 5 seconds
|
||||
var completed = await Task.WhenAny(waitTask, Task.Delay(TimeSpan.FromSeconds(5)));
|
||||
completed.ShouldBe(waitTask);
|
||||
|
||||
server.Dispose();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ShutdownAsync_is_idempotent()
|
||||
{
|
||||
var port = GetFreePort();
|
||||
var server = new NatsServer(new NatsOptions { Port = port }, NullLoggerFactory.Instance);
|
||||
_ = server.StartAsync(CancellationToken.None);
|
||||
await server.WaitForReadyAsync();
|
||||
|
||||
// Call ShutdownAsync 3 times -- should not throw
|
||||
await server.ShutdownAsync();
|
||||
await server.ShutdownAsync();
|
||||
await server.ShutdownAsync();
|
||||
|
||||
server.IsShuttingDown.ShouldBeTrue();
|
||||
server.Dispose();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Accept_loop_waits_for_active_clients()
|
||||
{
|
||||
var port = GetFreePort();
|
||||
var server = new NatsServer(new NatsOptions { Port = port }, NullLoggerFactory.Instance);
|
||||
_ = server.StartAsync(CancellationToken.None);
|
||||
await server.WaitForReadyAsync();
|
||||
|
||||
// Connect a client
|
||||
using var client = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
await client.ConnectAsync(IPAddress.Loopback, port);
|
||||
var buf = new byte[4096];
|
||||
await client.ReceiveAsync(buf, SocketFlags.None); // INFO
|
||||
await client.SendAsync(Encoding.ASCII.GetBytes("CONNECT {}\r\nPING\r\n"));
|
||||
using var readCts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
await client.ReceiveAsync(buf, SocketFlags.None, readCts.Token); // PONG
|
||||
|
||||
// ShutdownAsync should complete within 10 seconds (doesn't hang)
|
||||
var shutdownTask = server.ShutdownAsync();
|
||||
var completed = await Task.WhenAny(shutdownTask, Task.Delay(TimeSpan.FromSeconds(10)));
|
||||
completed.ShouldBe(shutdownTask);
|
||||
|
||||
server.Dispose();
|
||||
}
|
||||
}
|
||||
|
||||
public class LameDuckTests
|
||||
{
|
||||
private static int GetFreePort()
|
||||
{
|
||||
using var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
sock.Bind(new IPEndPoint(IPAddress.Loopback, 0));
|
||||
return ((IPEndPoint)sock.LocalEndPoint!).Port;
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task LameDuckShutdown_stops_accepting_new_connections()
|
||||
{
|
||||
var port = GetFreePort();
|
||||
var server = new NatsServer(
|
||||
new NatsOptions
|
||||
{
|
||||
Port = port,
|
||||
LameDuckDuration = TimeSpan.FromSeconds(3),
|
||||
LameDuckGracePeriod = TimeSpan.FromMilliseconds(500),
|
||||
},
|
||||
NullLoggerFactory.Instance);
|
||||
|
||||
_ = server.StartAsync(CancellationToken.None);
|
||||
await server.WaitForReadyAsync();
|
||||
|
||||
try
|
||||
{
|
||||
// Connect 1 client
|
||||
using var client1 = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
await client1.ConnectAsync(IPAddress.Loopback, port);
|
||||
var buf = new byte[4096];
|
||||
await client1.ReceiveAsync(buf, SocketFlags.None); // INFO
|
||||
await client1.SendAsync(Encoding.ASCII.GetBytes("CONNECT {}\r\nPING\r\n"));
|
||||
using var readCts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
await client1.ReceiveAsync(buf, SocketFlags.None, readCts.Token); // PONG
|
||||
|
||||
// Start lame duck (don't await yet)
|
||||
var lameDuckTask = server.LameDuckShutdownAsync();
|
||||
|
||||
// Wait briefly for listener to close
|
||||
await Task.Delay(300);
|
||||
|
||||
// Verify lame duck mode is active
|
||||
server.IsLameDuckMode.ShouldBeTrue();
|
||||
|
||||
// Try connecting a new client -- should fail (connection refused)
|
||||
using var client2 = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
var connectAction = async () =>
|
||||
{
|
||||
await client2.ConnectAsync(IPAddress.Loopback, port);
|
||||
};
|
||||
await connectAction.ShouldThrowAsync<SocketException>();
|
||||
|
||||
// Await the lame duck task with timeout
|
||||
var completed = await Task.WhenAny(lameDuckTask, Task.Delay(TimeSpan.FromSeconds(15)));
|
||||
completed.ShouldBe(lameDuckTask);
|
||||
}
|
||||
finally
|
||||
{
|
||||
server.Dispose();
|
||||
}
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task LameDuckShutdown_eventually_closes_all_clients()
|
||||
{
|
||||
var port = GetFreePort();
|
||||
var server = new NatsServer(
|
||||
new NatsOptions
|
||||
{
|
||||
Port = port,
|
||||
LameDuckDuration = TimeSpan.FromSeconds(2),
|
||||
LameDuckGracePeriod = TimeSpan.FromMilliseconds(200),
|
||||
},
|
||||
NullLoggerFactory.Instance);
|
||||
|
||||
_ = server.StartAsync(CancellationToken.None);
|
||||
await server.WaitForReadyAsync();
|
||||
|
||||
try
|
||||
{
|
||||
// Connect 3 clients via raw sockets
|
||||
var clients = new List<Socket>();
|
||||
var buf = new byte[4096];
|
||||
for (int i = 0; i < 3; i++)
|
||||
{
|
||||
var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
await sock.ConnectAsync(IPAddress.Loopback, port);
|
||||
await sock.ReceiveAsync(buf, SocketFlags.None); // INFO
|
||||
await sock.SendAsync(Encoding.ASCII.GetBytes("CONNECT {}\r\nPING\r\n"));
|
||||
using var readCts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
await sock.ReceiveAsync(buf, SocketFlags.None, readCts.Token); // PONG
|
||||
clients.Add(sock);
|
||||
}
|
||||
|
||||
server.ClientCount.ShouldBe(3);
|
||||
|
||||
// Await LameDuckShutdownAsync
|
||||
var lameDuckTask = server.LameDuckShutdownAsync();
|
||||
var completed = await Task.WhenAny(lameDuckTask, Task.Delay(TimeSpan.FromSeconds(15)));
|
||||
completed.ShouldBe(lameDuckTask);
|
||||
|
||||
server.ClientCount.ShouldBe(0);
|
||||
|
||||
foreach (var sock in clients)
|
||||
sock.Dispose();
|
||||
}
|
||||
finally
|
||||
{
|
||||
server.Dispose();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public class PidFileTests : IDisposable
|
||||
{
|
||||
private readonly string _tempDir = Path.Combine(Path.GetTempPath(), $"nats-test-{Guid.NewGuid():N}");
|
||||
|
||||
public PidFileTests() => Directory.CreateDirectory(_tempDir);
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
if (Directory.Exists(_tempDir))
|
||||
Directory.Delete(_tempDir, recursive: true);
|
||||
}
|
||||
|
||||
private static int GetFreePort()
|
||||
{
|
||||
using var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
sock.Bind(new IPEndPoint(IPAddress.Loopback, 0));
|
||||
return ((IPEndPoint)sock.LocalEndPoint!).Port;
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Server_writes_pid_file_on_startup()
|
||||
{
|
||||
var pidFile = Path.Combine(_tempDir, "nats.pid");
|
||||
var port = GetFreePort();
|
||||
var server = new NatsServer(new NatsOptions { Port = port, PidFile = pidFile }, NullLoggerFactory.Instance);
|
||||
_ = server.StartAsync(CancellationToken.None);
|
||||
await server.WaitForReadyAsync();
|
||||
|
||||
File.Exists(pidFile).ShouldBeTrue();
|
||||
var content = await File.ReadAllTextAsync(pidFile);
|
||||
int.Parse(content).ShouldBe(Environment.ProcessId);
|
||||
|
||||
await server.ShutdownAsync();
|
||||
File.Exists(pidFile).ShouldBeFalse();
|
||||
|
||||
server.Dispose();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Server_writes_ports_file_on_startup()
|
||||
{
|
||||
var port = GetFreePort();
|
||||
var server = new NatsServer(new NatsOptions { Port = port, PortsFileDir = _tempDir }, NullLoggerFactory.Instance);
|
||||
_ = server.StartAsync(CancellationToken.None);
|
||||
await server.WaitForReadyAsync();
|
||||
|
||||
var portsFiles = Directory.GetFiles(_tempDir, "*.ports");
|
||||
portsFiles.Length.ShouldBe(1);
|
||||
|
||||
var content = await File.ReadAllTextAsync(portsFiles[0]);
|
||||
content.ShouldContain($"\"client\":{port}");
|
||||
|
||||
await server.ShutdownAsync();
|
||||
Directory.GetFiles(_tempDir, "*.ports").Length.ShouldBe(0);
|
||||
|
||||
server.Dispose();
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user