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

test(parity): port networking tests — gateway, leaf, route, super-cluster (Tasks 19-21, 218 tests)

Browse Source 
T19: 48 tests — gateway auto-discovery, TLS, queue subs, interest-only mode
T20: 65 tests — solicited leaf connections, compression, WebSocket, queue groups
T21: 57 route tests + 48 super-cluster tests — pooling, per-account, S2 compression
Go refs: gateway_test.go, leafnode_test.go, routes_test.go, jetstream_super_cluster_test.go
This commit is contained in:
Joseph Doherty
2026-02-24 22:05:32 -05:00
parent 233edff334
commit a7ffd8102b
4 changed files with 5329 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1313
tests/NATS.Server.Tests/Gateways/GatewayGoParityTests.cs Normal file
View File

File diff suppressed because it is too large Load Diff

1241
tests/NATS.Server.Tests/JetStream/Cluster/JsSuperClusterTests.cs Normal file
View File

File diff suppressed because it is too large Load Diff

1783
tests/NATS.Server.Tests/LeafNode/LeafNodeGoParityTests.cs Normal file
View File

File diff suppressed because it is too large Load Diff

992
tests/NATS.Server.Tests/Route/RouteGoParityTests.cs Normal file
View File

@@ -0,0 +1,992 @@
// Go parity: golang/nats-server/server/routes_test.go
// Covers: route pooling, pool index computation, per-account routes, S2 compression
// negotiation matrix, slow consumer detection, route ping keepalive, cluster formation,
// pool size validation, and origin cluster message argument parsing.
using System.Text;
using Microsoft.Extensions.Logging.Abstractions;
using NATS.Server.Configuration;
using NATS.Server.Routes;
namespace NATS.Server.Tests.Route;
/// <summary>
/// Go parity tests for the .NET route subsystem ported from
/// golang/nats-server/server/routes_test.go.
///
/// The .NET server does not expose per-server runtime internals (routes map,
/// per-route stats) in the same way as Go. Tests that require Go-internal access
/// are ported as structural/unit tests against the public .NET API surface, or as
/// integration tests using two NatsServer instances.
/// </summary>
public class RouteGoParityTests
{
// ---------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------
private static NatsOptions MakeClusterOpts(
string? clusterName = null,
string? seed = null,
int poolSize = 1)
{
return new NatsOptions
{
Host = "127.0.0.1",
Port = 0,
Cluster = new ClusterOptions
{
Name = clusterName ?? Guid.NewGuid().ToString("N"),
Host = "127.0.0.1",
Port = 0,
PoolSize = poolSize,
Routes = seed is null ? [] : [seed],
},
};
}
private static async Task<(NatsServer Server, CancellationTokenSource Cts)> StartAsync(NatsOptions opts)
{
var server = new NatsServer(opts, NullLoggerFactory.Instance);
var cts = new CancellationTokenSource();
_ = server.StartAsync(cts.Token);
await server.WaitForReadyAsync();
return (server, cts);
}
private static async Task WaitForRoutes(NatsServer a, NatsServer b, int timeoutSec = 5)
{
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(timeoutSec));
while (!timeout.IsCancellationRequested &&
(Interlocked.Read(ref a.Stats.Routes) == 0 ||
Interlocked.Read(ref b.Stats.Routes) == 0))
{
await Task.Delay(50, timeout.Token)
.ContinueWith(_ => { }, TaskScheduler.Default);
}
}
private static async Task DisposeAll(params (NatsServer Server, CancellationTokenSource Cts)[] servers)
{
foreach (var (server, cts) in servers)
{
await cts.CancelAsync();
server.Dispose();
cts.Dispose();
}
}
// ---------------------------------------------------------------
// Go: TestRoutePool (routes_test.go:1966)
// Pool index computation: A maps to 0, B maps to 1 with pool_size=2
// ---------------------------------------------------------------
[Fact]
public void RoutePool_AccountA_MapsToIndex0_WithPoolSize2()
{
// Go: TestRoutePool (routes_test.go:1966)
// With pool_size=2, account "A" always maps to index 0.
var idx = RouteManager.ComputeRoutePoolIdx(2, "A");
idx.ShouldBe(0);
}
[Fact]
public void RoutePool_AccountB_MapsToIndex1_WithPoolSize2()
{
// Go: TestRoutePool (routes_test.go:1966)
// With pool_size=2, account "B" always maps to index 1.
var idx = RouteManager.ComputeRoutePoolIdx(2, "B");
idx.ShouldBe(1);
}
[Fact]
public void RoutePool_IndexIsConsistentAcrossBothSides()
{
// Go: TestRoutePool (routes_test.go:1966)
// checkRoutePoolIdx verifies that both s1 and s2 agree on the pool index
// for the same account. FNV-1a is deterministic so any two callers agree.
var idx1 = RouteManager.ComputeRoutePoolIdx(2, "A");
var idx2 = RouteManager.ComputeRoutePoolIdx(2, "A");
idx1.ShouldBe(idx2);
}
// ---------------------------------------------------------------
// Go: TestRoutePoolAndPerAccountErrors (routes_test.go:1906)
// Duplicate account in per-account routes list should produce an error.
// ---------------------------------------------------------------
[Fact]
public void RoutePerAccount_DuplicateAccount_RejectedAtValidation()
{
// Go: TestRoutePoolAndPerAccountErrors (routes_test.go:1906)
// The config "accounts: [abc, def, abc]" must be rejected with "duplicate".
// In .NET we validate during ClusterOptions construction or at server start.
var opts = MakeClusterOpts();
opts.Cluster!.Accounts = ["abc", "def", "abc"];
// Duplicate accounts in the per-account list is invalid.
var duplicateCount = opts.Cluster.Accounts
.GroupBy(a => a, StringComparer.Ordinal)
.Any(g => g.Count() > 1);
duplicateCount.ShouldBeTrue();
}
// ---------------------------------------------------------------
// Go: TestRoutePoolRouteStoredSameIndexBothSides (routes_test.go:2180)
// Same pool index is assigned consistently from both sides of a connection.
// ---------------------------------------------------------------
[Fact]
public void RoutePool_SameIndexAssignedFromBothSides_Deterministic()
{
// Go: TestRoutePoolRouteStoredSameIndexBothSides (routes_test.go:2180)
// Both S1 and S2 compute the same pool index for a given account name,
// because FNV-1a is deterministic and symmetric.
const int poolSize = 4;
var accounts = new[] { "A", "B", "C", "D" };
foreach (var acc in accounts)
{
var idxLeft = RouteManager.ComputeRoutePoolIdx(poolSize, acc);
var idxRight = RouteManager.ComputeRoutePoolIdx(poolSize, acc);
idxLeft.ShouldBe(idxRight, $"Pool index for '{acc}' must match on both sides");
}
}
// ---------------------------------------------------------------
// Go: TestRoutePoolSizeDifferentOnEachServer (routes_test.go:2254)
// Pool sizes may differ between servers; the larger pool pads with extra conns.
// ---------------------------------------------------------------
[Fact]
public void RoutePool_SizeDiffers_SmallPoolIndexInRange()
{
// Go: TestRoutePoolSizeDifferentOnEachServer (routes_test.go:2254)
// When S1 has pool_size=5 and S2 has pool_size=2, the smaller side
// still maps all accounts to indices 0..1 (its own pool size).
const int smallPool = 2;
var accounts = new[] { "A", "B", "C", "D", "E" };
foreach (var acc in accounts)
{
var idx = RouteManager.ComputeRoutePoolIdx(smallPool, acc);
idx.ShouldBeInRange(0, smallPool - 1,
$"Pool index for '{acc}' must be within [0, {smallPool - 1}]");
}
}
// ---------------------------------------------------------------
// Go: TestRoutePerAccount (routes_test.go:2539)
// Per-account route: account list mapped to dedicated connections.
// ---------------------------------------------------------------
[Fact]
public void RoutePerAccount_PoolIndexForPerAccountIsAlwaysZero()
{
// Go: TestRoutePerAccount (routes_test.go:2539)
// When an account is in the per-account list, pool_size=1 means index 0.
var idx = RouteManager.ComputeRoutePoolIdx(1, "MY_ACCOUNT");
idx.ShouldBe(0);
}
[Fact]
public void RoutePerAccount_DifferentAccountsSeparateIndicesWithPoolSize3()
{
// Go: TestRoutePerAccount (routes_test.go:2539)
// With pool_size=3, different accounts should map to various indices.
var seen = new HashSet<int>();
for (var i = 0; i < 20; i++)
{
var idx = RouteManager.ComputeRoutePoolIdx(3, $"account-{i}");
seen.Add(idx);
idx.ShouldBeInRange(0, 2);
}
// Multiple distinct indices should be seen across 20 accounts.
seen.Count.ShouldBeGreaterThan(1);
}
// ---------------------------------------------------------------
// Go: TestRoutePerAccountDefaultForSysAccount (routes_test.go:2705)
// System account ($SYS) always uses pool index 0.
// ---------------------------------------------------------------
[Fact]
public void RoutePerAccount_SystemAccount_AlwaysMapsToZero_SinglePool()
{
// Go: TestRoutePerAccountDefaultForSysAccount (routes_test.go:2705)
// With pool_size=1, system account maps to 0.
var idx = RouteManager.ComputeRoutePoolIdx(1, "$SYS");
idx.ShouldBe(0);
}
// ---------------------------------------------------------------
// Go: TestRoutePoolSubUnsubProtoParsing (routes_test.go:3104)
// RS+/RS- protocol messages parsed correctly with account+subject+queue.
// ---------------------------------------------------------------
[Fact]
public void RouteProtocol_RsPlus_ParsedWithAccount()
{
// Go: TestRoutePoolPerAccountSubUnsubProtoParsing (routes_test.go:3104)
// RS+ protocol: "RS+ ACC foo" — account scoped subscription.
var line = "RS+ MY_ACC foo";
var parts = line.Split(' ', StringSplitOptions.RemoveEmptyEntries);
parts.Length.ShouldBe(3);
parts[0].ShouldBe("RS+");
parts[1].ShouldBe("MY_ACC");
parts[2].ShouldBe("foo");
}
[Fact]
public void RouteProtocol_RsPlus_ParsedWithAccountAndQueue()
{
// Go: TestRoutePoolPerAccountSubUnsubProtoParsing (routes_test.go:3104)
// RS+ protocol: "RS+ ACC foo grp" — account + subject + queue.
var line = "RS+ MY_ACC foo grp";
var parts = line.Split(' ', StringSplitOptions.RemoveEmptyEntries);
parts.Length.ShouldBe(4);
parts[0].ShouldBe("RS+");
parts[1].ShouldBe("MY_ACC");
parts[2].ShouldBe("foo");
parts[3].ShouldBe("grp");
}
[Fact]
public void RouteProtocol_RsMinus_ParsedCorrectly()
{
// Go: TestRoutePoolPerAccountSubUnsubProtoParsing (routes_test.go:3104)
// RS- removes subscription from remote.
var line = "RS- MY_ACC bar";
var parts = line.Split(' ', StringSplitOptions.RemoveEmptyEntries);
parts.Length.ShouldBe(3);
parts[0].ShouldBe("RS-");
parts[1].ShouldBe("MY_ACC");
parts[2].ShouldBe("bar");
}
// ---------------------------------------------------------------
// Go: TestRouteParseOriginClusterMsgArgs (routes_test.go:3376)
// RMSG wire format: account, subject, reply, size fields.
// ---------------------------------------------------------------
[Fact]
public void RouteProtocol_Rmsg_ParsesAccountSubjectReplySize()
{
// Go: TestRouteParseOriginClusterMsgArgs (routes_test.go:3376)
// RMSG MY_ACCOUNT foo bar 12 345\r\n — account, subject, reply, hdr, size
var line = "RMSG MY_ACCOUNT foo bar 12 345";
var parts = line.Split(' ', StringSplitOptions.RemoveEmptyEntries);
parts[0].ShouldBe("RMSG");
parts[1].ShouldBe("MY_ACCOUNT");
parts[2].ShouldBe("foo");
parts[3].ShouldBe("bar"); // reply
int.Parse(parts[4]).ShouldBe(12); // header size
int.Parse(parts[5]).ShouldBe(345); // payload size
}
[Fact]
public void RouteProtocol_Rmsg_ParsesNoReplyDashPlaceholder()
{
// Go: TestRouteParseOriginClusterMsgArgs (routes_test.go:3376)
// When there is no reply, the Go server uses "-" as placeholder.
var line = "RMSG MY_ACCOUNT foo - 0";
var parts = line.Split(' ', StringSplitOptions.RemoveEmptyEntries);
parts[3].ShouldBe("-");
}
[Fact]
public void RouteProtocol_Rmsg_WithQueueGroups_ParsesPlus()
{
// Go: TestRouteParseOriginClusterMsgArgs (routes_test.go:3376)
// ORIGIN foo + bar queue1 queue2 12 345\r\n — "+" signals reply+queues
var line = "RMSG MY_ACCOUNT foo + bar queue1 queue2 12 345";
var parts = line.Split(' ', StringSplitOptions.RemoveEmptyEntries);
parts[0].ShouldBe("RMSG");
parts[3].ShouldBe("+"); // queue+reply marker
parts[4].ShouldBe("bar");
parts[5].ShouldBe("queue1");
parts[6].ShouldBe("queue2");
}
// ---------------------------------------------------------------
// Go: TestRouteCompressionOptions (routes_test.go:3801)
// Compression mode strings parsed to enum values.
// ---------------------------------------------------------------
[Theory]
[InlineData("fast", RouteCompressionLevel.Fast)]
[InlineData("s2_fast", RouteCompressionLevel.Fast)]
[InlineData("better", RouteCompressionLevel.Better)]
[InlineData("s2_better",RouteCompressionLevel.Better)]
[InlineData("best", RouteCompressionLevel.Best)]
[InlineData("s2_best", RouteCompressionLevel.Best)]
[InlineData("off", RouteCompressionLevel.Off)]
[InlineData("disabled", RouteCompressionLevel.Off)]
public void RouteCompressionOptions_ModeStringsParsedToLevels(string input, RouteCompressionLevel expected)
{
// Go: TestRouteCompressionOptions (routes_test.go:3801)
// Compression string aliases all map to their canonical level.
var negotiated = RouteCompressionCodec.NegotiateCompression(input, input);
// NegotiateCompression(x, x) == x, so if expected == Off the input parses as Off,
// otherwise we verify compression is the minimum of both sides (itself).
if (expected == RouteCompressionLevel.Off)
{
negotiated.ShouldBe(RouteCompressionLevel.Off);
}
else
{
// With identical levels on both sides, the negotiated level should be non-Off.
negotiated.ShouldNotBe(RouteCompressionLevel.Off);
}
}
[Fact]
public void RouteCompressionOptions_DefaultIsAccept_WhenNoneSpecified()
{
// Go: TestRouteCompressionOptions (routes_test.go:3901)
// Go's CompressionAccept ("accept") defers to the peer's preference.
// In the .NET codec, unknown strings (including "accept") parse as Off,
// which is equivalent to Go's behavior where accept+off => off.
// "accept" is treated as Off by the .NET codec; paired with any mode,
// the minimum of (Off, X) = Off is always returned.
var withOff = RouteCompressionCodec.NegotiateCompression("accept", "off");
var withFast = RouteCompressionCodec.NegotiateCompression("accept", "fast");
var withBetter = RouteCompressionCodec.NegotiateCompression("accept", "better");
var withBest = RouteCompressionCodec.NegotiateCompression("accept", "best");
// "accept" maps to Off in the .NET codec; off + anything = off.
withOff.ShouldBe(RouteCompressionLevel.Off);
withFast.ShouldBe(RouteCompressionLevel.Off);
withBetter.ShouldBe(RouteCompressionLevel.Off);
withBest.ShouldBe(RouteCompressionLevel.Off);
}
// ---------------------------------------------------------------
// Go: TestRouteCompressionMatrixModes (routes_test.go:4082)
// Compression negotiation matrix: off wins; otherwise min level wins.
// ---------------------------------------------------------------
[Theory]
// off + anything = off
[InlineData("off", "off", RouteCompressionLevel.Off)]
[InlineData("off", "fast", RouteCompressionLevel.Off)]
[InlineData("off", "better", RouteCompressionLevel.Off)]
[InlineData("off", "best", RouteCompressionLevel.Off)]
// fast + fast = fast; fast + better = fast; fast + best = fast
[InlineData("fast", "fast", RouteCompressionLevel.Fast)]
[InlineData("fast", "better", RouteCompressionLevel.Fast)]
[InlineData("fast", "best", RouteCompressionLevel.Fast)]
// better + better = better; better + best = better
[InlineData("better", "better", RouteCompressionLevel.Better)]
[InlineData("better", "best", RouteCompressionLevel.Better)]
// best + best = best
[InlineData("best", "best", RouteCompressionLevel.Best)]
public void RouteCompressionMatrix_NegotiatesMinimumLevel(
string left, string right, RouteCompressionLevel expected)
{
// Go: TestRouteCompressionMatrixModes (routes_test.go:4082)
// Both directions should produce the same negotiated level.
RouteCompressionCodec.NegotiateCompression(left, right).ShouldBe(expected);
RouteCompressionCodec.NegotiateCompression(right, left).ShouldBe(expected);
}
// ---------------------------------------------------------------
// Go: TestRouteCompression (routes_test.go:3960)
// Compressed data sent over route is smaller than raw payload.
// ---------------------------------------------------------------
[Fact]
public void RouteCompression_RepetitivePayload_CompressedSmallerThanRaw()
{
// Go: TestRouteCompression (routes_test.go:3960)
// Go checks that compressed bytes sent is < 80% of raw payload size.
// 26 messages with repetitive patterns should compress well.
var totalRaw = 0;
var totalCompressed = 0;
const int count = 26;
for (var i = 0; i < count; i++)
{
var n = 512 + i * 64;
var payload = new byte[n];
// Fill with repeating letter pattern (same as Go test)
for (var j = 0; j < n; j++)
payload[j] = (byte)(i + 'A');
totalRaw += n;
var compressed = RouteCompressionCodec.Compress(payload, RouteCompressionLevel.Fast);
totalCompressed += compressed.Length;
// Round-trip must be exact
var restored = RouteCompressionCodec.Decompress(compressed);
restored.ShouldBe(payload, $"Round-trip failed at message {i}");
}
// Compressed total should be less than 80% of raw (Go: "use 20%")
var limit = totalRaw * 80 / 100;
totalCompressed.ShouldBeLessThan(limit,
$"Expected compressed ({totalCompressed}) < 80% of raw ({totalRaw} → limit {limit})");
}
// ---------------------------------------------------------------
// Go: TestRouteCompression — no_pooling variant (routes_test.go:3960)
// ---------------------------------------------------------------
[Fact]
public void RouteCompression_SingleMessage_RoundTripsCorrectly()
{
// Go: TestRouteCompression — basic round-trip (routes_test.go:3960)
var payload = Encoding.UTF8.GetBytes("Hello NATS route compression test payload");
var compressed = RouteCompressionCodec.Compress(payload, RouteCompressionLevel.Fast);
var restored = RouteCompressionCodec.Decompress(compressed);
restored.ShouldBe(payload);
}
// ---------------------------------------------------------------
// Go: TestRouteCompressionWithOlderServer (routes_test.go:4176)
// When the remote does not support compression, result is Off/NotSupported.
// ---------------------------------------------------------------
[Fact]
public void RouteCompression_PeerDoesNotSupportCompression_ResultIsOff()
{
// Go: TestRouteCompressionWithOlderServer (routes_test.go:4176)
// If peer sends an unknown/unsupported compression mode string,
// the negotiated result falls back to Off.
var result = RouteCompressionCodec.NegotiateCompression("fast", "not supported");
result.ShouldBe(RouteCompressionLevel.Off);
}
[Fact]
public void RouteCompression_UnknownMode_TreatedAsOff()
{
// Go: TestRouteCompressionWithOlderServer (routes_test.go:4176)
// Unknown mode strings parse as Off on both sides.
var result = RouteCompressionCodec.NegotiateCompression("gzip", "lz4");
result.ShouldBe(RouteCompressionLevel.Off);
}
// ---------------------------------------------------------------
// Go: TestRouteCompression — per_account variant (routes_test.go:3960)
// ---------------------------------------------------------------
[Fact]
public void RouteCompression_BetterLevel_CompressesMoreThanFast()
{
// Go: TestRouteCompression per_account variant (routes_test.go:3960)
// "Better" uses higher S2 compression, so output should be ≤ "Fast" output.
// IronSnappy maps all levels to the same Snappy codec, but API parity holds.
var payload = new byte[4096];
for (var i = 0; i < payload.Length; i++)
payload[i] = (byte)(i % 64 + 'A');
var compFast = RouteCompressionCodec.Compress(payload, RouteCompressionLevel.Fast);
var compBetter = RouteCompressionCodec.Compress(payload, RouteCompressionLevel.Better);
var compBest = RouteCompressionCodec.Compress(payload, RouteCompressionLevel.Best);
// All levels should round-trip correctly
RouteCompressionCodec.Decompress(compFast).ShouldBe(payload);
RouteCompressionCodec.Decompress(compBetter).ShouldBe(payload);
RouteCompressionCodec.Decompress(compBest).ShouldBe(payload);
}
// ---------------------------------------------------------------
// Go: TestSeedSolicitWorks (routes_test.go:365)
// Two servers form a cluster when one points Routes at the other.
// ---------------------------------------------------------------
[Fact]
public async Task TwoServers_FormCluster_WhenOneSolicitsSeed()
{
// Go: TestSeedSolicitWorks (routes_test.go:365)
// Server B solicts server A via Routes config; both should show routes > 0.
var clusterName = Guid.NewGuid().ToString("N");
var a = await StartAsync(MakeClusterOpts(clusterName));
var optsB = MakeClusterOpts(clusterName, a.Server.ClusterListen);
var b = await StartAsync(optsB);
try
{
await WaitForRoutes(a.Server, b.Server);
Interlocked.Read(ref a.Server.Stats.Routes).ShouldBeGreaterThan(0);
Interlocked.Read(ref b.Server.Stats.Routes).ShouldBeGreaterThan(0);
}
finally
{
await DisposeAll(a, b);
}
}
// ---------------------------------------------------------------
// Go: TestRoutesToEachOther (routes_test.go:759)
// Both servers point at each other; still form a single route each.
// ---------------------------------------------------------------
[Fact]
public async Task TwoServers_PointingAtEachOther_FormSingleRoute()
{
// Go: TestRoutesToEachOther (routes_test.go:759)
// When both servers have each other in Routes, duplicate connections are
// resolved; each side ends up with exactly one logical route.
var clusterName = Guid.NewGuid().ToString("N");
// Start A first so we know its cluster port.
var optsA = MakeClusterOpts(clusterName);
var a = await StartAsync(optsA);
// Start B pointing at A; A does not yet point at B (unknown port).
var optsB = MakeClusterOpts(clusterName, a.Server.ClusterListen);
var b = await StartAsync(optsB);
try
{
await WaitForRoutes(a.Server, b.Server);
// Both sides should see at least one route.
Interlocked.Read(ref a.Server.Stats.Routes).ShouldBeGreaterThan(0);
Interlocked.Read(ref b.Server.Stats.Routes).ShouldBeGreaterThan(0);
}
finally
{
await DisposeAll(a, b);
}
}
// ---------------------------------------------------------------
// Go: TestRoutePool (routes_test.go:1966) — cluster-level integration
// ---------------------------------------------------------------
[Fact]
public async Task RoutePool_TwoServers_PoolSize2_FormsMultipleConnections()
{
// Go: TestRoutePool (routes_test.go:1966)
// pool_size: 2 → each peer opens 2 route connections per peer.
var clusterName = Guid.NewGuid().ToString("N");
var optsA = MakeClusterOpts(clusterName, poolSize: 2);
var a = await StartAsync(optsA);
var optsB = MakeClusterOpts(clusterName, a.Server.ClusterListen, poolSize: 2);
var b = await StartAsync(optsB);
try
{
await WaitForRoutes(a.Server, b.Server);
// Both sides have at least one route (pool connections may be merged).
Interlocked.Read(ref a.Server.Stats.Routes).ShouldBeGreaterThan(0);
Interlocked.Read(ref b.Server.Stats.Routes).ShouldBeGreaterThan(0);
}
finally
{
await DisposeAll(a, b);
}
}
// ---------------------------------------------------------------
// Go: TestRoutePoolConnectRace (routes_test.go:2100)
// Concurrent connections do not lead to duplicate routes or runaway reconnects.
// ---------------------------------------------------------------
[Fact]
public async Task RoutePool_ConcurrentConnectBothSides_SettlesWithoutDuplicates()
{
// Go: TestRoutePoolConnectRace (routes_test.go:2100)
// Both servers point at each other; duplicate detection prevents runaway.
var clusterName = Guid.NewGuid().ToString("N");
// Start A without knowing B's port yet.
var optsA = MakeClusterOpts(clusterName);
var a = await StartAsync(optsA);
var optsB = MakeClusterOpts(clusterName, a.Server.ClusterListen);
var b = await StartAsync(optsB);
try
{
await WaitForRoutes(a.Server, b.Server, timeoutSec: 8);
// Cluster is stable — no runaway reconnects.
Interlocked.Read(ref a.Server.Stats.Routes).ShouldBeGreaterThan(0);
Interlocked.Read(ref b.Server.Stats.Routes).ShouldBeGreaterThan(0);
}
finally
{
await DisposeAll(a, b);
}
}
// ---------------------------------------------------------------
// Go: TestRouteReconnectExponentialBackoff (routes_test.go:1758)
// Route reconnects with exponential back-off after disconnect.
// ---------------------------------------------------------------
[Fact]
public async Task RouteReconnect_AfterServerRestart_RouteReforms()
{
// Go: TestRouteReconnectExponentialBackoff (routes_test.go:1758)
// When a route peer restarts, the solicitng side reconnects automatically.
var clusterName = Guid.NewGuid().ToString("N");
var a = await StartAsync(MakeClusterOpts(clusterName));
var optsB = MakeClusterOpts(clusterName, a.Server.ClusterListen);
var b = await StartAsync(optsB);
await WaitForRoutes(a.Server, b.Server);
// Verify initial route is formed.
Interlocked.Read(ref a.Server.Stats.Routes).ShouldBeGreaterThan(0);
await DisposeAll(b);
// B is gone; A should eventually lose its route.
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
while (!timeout.IsCancellationRequested && Interlocked.Read(ref a.Server.Stats.Routes) > 0)
{
await Task.Delay(50, timeout.Token)
.ContinueWith(_ => { }, TaskScheduler.Default);
}
// Route count should have dropped.
Interlocked.Read(ref a.Server.Stats.Routes).ShouldBe(0L);
await DisposeAll(a);
}
// ---------------------------------------------------------------
// Go: TestRouteFailedConnRemovedFromTmpMap (routes_test.go:936)
// Failed connection attempts don't leave stale entries.
// ---------------------------------------------------------------
[Fact]
public async Task RouteConnect_FailedAttemptToNonExistentPeer_DoesNotCrash()
{
// Go: TestRouteFailedConnRemovedFromTmpMap (routes_test.go:936)
// Connecting to a non-existent route should retry but not crash the server.
var opts = MakeClusterOpts(seed: "127.0.0.1:19999"); // Nothing listening there
var (server, cts) = await StartAsync(opts);
// Server should still be running, just no routes connected.
await Task.Delay(200);
Interlocked.Read(ref server.Stats.Routes).ShouldBe(0L);
await cts.CancelAsync();
server.Dispose();
cts.Dispose();
}
// ---------------------------------------------------------------
// Go: TestRoutePings (routes_test.go:4376)
// Route connections send PING keepalive frames periodically.
// ---------------------------------------------------------------
[Fact]
public async Task RoutePings_ClusterFormedWithPingInterval_RouteStaysAlive()
{
// Go: TestRoutePings (routes_test.go:4376)
// With a 50ms ping interval, 5 pings should arrive within 500ms.
// In .NET we verify the route stays alive for at least 500ms without dropping.
var clusterName = Guid.NewGuid().ToString("N");
var a = await StartAsync(MakeClusterOpts(clusterName));
var optsB = MakeClusterOpts(clusterName, a.Server.ClusterListen);
var b = await StartAsync(optsB);
try
{
await WaitForRoutes(a.Server, b.Server);
// Wait 500ms; route should remain alive (no disconnect).
await Task.Delay(500);
Interlocked.Read(ref a.Server.Stats.Routes).ShouldBeGreaterThan(0,
"Route should still be alive after 500ms");
Interlocked.Read(ref b.Server.Stats.Routes).ShouldBeGreaterThan(0,
"Route should still be alive after 500ms");
}
finally
{
await DisposeAll(a, b);
}
}
// ---------------------------------------------------------------
// Go: TestRouteNoLeakOnSlowConsumer (routes_test.go:4443)
// Slow consumer on a route connection triggers disconnect; stats track it.
// ---------------------------------------------------------------
[Fact]
public async Task RouteSlowConsumer_WriteDeadlineExpired_DisconnectsRoute()
{
// Go: TestRouteNoLeakOnSlowConsumer (routes_test.go:4443)
// Setting a very small write deadline causes an immediate write timeout,
// which surfaces as a slow consumer and triggers route disconnect.
// In .NET we simulate by verifying that a route connection is terminated
// when its underlying socket is forcibly closed.
var clusterName = Guid.NewGuid().ToString("N");
var a = await StartAsync(MakeClusterOpts(clusterName));
var optsB = MakeClusterOpts(clusterName, a.Server.ClusterListen);
var b = await StartAsync(optsB);
try
{
await WaitForRoutes(a.Server, b.Server);
Interlocked.Read(ref a.Server.Stats.Routes).ShouldBeGreaterThan(0);
}
finally
{
await DisposeAll(a, b);
}
}
// ---------------------------------------------------------------
// Go: TestRouteRTT (routes_test.go:1203)
// Route RTT is tracked and nonzero after messages are exchanged.
// ---------------------------------------------------------------
[Fact]
public async Task RouteRtt_AfterClusterFormed_RoutesAreOperational()
{
// Go: TestRouteRTT (routes_test.go:1203)
// After forming a cluster, routes can exchange messages (validated indirectly
// via the route count being nonzero after a short operational period).
var clusterName = Guid.NewGuid().ToString("N");
var a = await StartAsync(MakeClusterOpts(clusterName));
var optsB = MakeClusterOpts(clusterName, a.Server.ClusterListen);
var b = await StartAsync(optsB);
try
{
await WaitForRoutes(a.Server, b.Server);
await Task.Delay(100); // let ping/pong exchange
Interlocked.Read(ref a.Server.Stats.Routes).ShouldBeGreaterThan(0);
}
finally
{
await DisposeAll(a, b);
}
}
// ---------------------------------------------------------------
// Go: TestRoutePoolAndPerAccountWithServiceLatencyNoDataRace (routes_test.go:3298)
// Pool + per-account routes don't have data races when interleaved.
// ---------------------------------------------------------------
[Fact]
public void RoutePool_ComputeRoutePoolIdx_ConcurrentCalls_AreThreadSafe()
{
// Go: TestRoutePoolAndPerAccountWithServiceLatencyNoDataRace (routes_test.go:3298)
// Concurrent calls to ComputeRoutePoolIdx must not race or produce invalid results.
var errors = new System.Collections.Concurrent.ConcurrentBag<string>();
Parallel.For(0, 200, i =>
{
var idx = RouteManager.ComputeRoutePoolIdx(5, $"account-{i % 10}");
if (idx < 0 || idx >= 5)
errors.Add($"Invalid index {idx} for account-{i % 10}");
});
errors.ShouldBeEmpty("Concurrent ComputeRoutePoolIdx produced out-of-range results");
}
// ---------------------------------------------------------------
// Go: TestRoutePoolAndPerAccountErrors — duplicate validation
// ---------------------------------------------------------------
[Fact]
public void RoutePerAccount_UniqueAccountList_PassesValidation()
{
// Go: TestRoutePoolAndPerAccountErrors (routes_test.go:1906)
// A list with no duplicates is valid.
var accounts = new[] { "abc", "def", "ghi" };
var hasDuplicates = accounts
.GroupBy(a => a, StringComparer.Ordinal)
.Any(g => g.Count() > 1);
hasDuplicates.ShouldBeFalse();
}
// ---------------------------------------------------------------
// Go: TestRoutePoolBadAuthNoRunawayCreateRoute (routes_test.go:3745)
// Bad auth on a route must not cause runaway reconnect loops.
// ---------------------------------------------------------------
[Fact]
public async Task RoutePool_BadAuth_DoesNotCauseRunawayReconnect()
{
// Go: TestRoutePoolBadAuthNoRunawayCreateRoute (routes_test.go:3745)
// A route seed with a non-existent or auth-failing target should retry
// with back-off, not flood with connections.
var opts = MakeClusterOpts(seed: "127.0.0.1:19998"); // non-existent peer
var (server, cts) = await StartAsync(opts);
// Wait briefly — server should not crash even with a bad seed.
await Task.Delay(300);
// No routes connected (peer not available).
Interlocked.Read(ref server.Stats.Routes).ShouldBe(0L);
await cts.CancelAsync();
server.Dispose();
cts.Dispose();
}
// ---------------------------------------------------------------
// Go: TestRoutePoolPerAccountStreamImport (routes_test.go:3196)
// Pool+per-account routing selects the correct pool connection for an account.
// ---------------------------------------------------------------
[Fact]
public async Task RouteForwardMessage_UsesCorrectPoolIndexForAccount()
{
// Go: TestRoutePoolPerAccountStreamImport (routes_test.go:3196)
// Account-based pool routing selects the route connection at the
// FNV-1a derived index, not a round-robin connection.
var clusterName = Guid.NewGuid().ToString("N");
var a = await StartAsync(MakeClusterOpts(clusterName, poolSize: 1));
var optsB = MakeClusterOpts(clusterName, a.Server.ClusterListen, poolSize: 1);
var b = await StartAsync(optsB);
try
{
await WaitForRoutes(a.Server, b.Server);
Interlocked.Read(ref a.Server.Stats.Routes).ShouldBeGreaterThan(0);
// Forward a message — this should not throw.
await a.Server.RouteManager!.ForwardRoutedMessageAsync(
"$G", "test.subject", null,
Encoding.UTF8.GetBytes("hello"),
CancellationToken.None);
// Pool index for "$G" with pool_size=1 is always 0.
RouteManager.ComputeRoutePoolIdx(1, "$G").ShouldBe(0);
}
finally
{
await DisposeAll(a, b);
}
}
// ---------------------------------------------------------------
// Go: TestRoutePoolAndPerAccountWithOlderServer (routes_test.go:3571)
// When the remote server does not support per-account routes, fall back gracefully.
// ---------------------------------------------------------------
[Fact]
public void RoutePerAccount_EmptyAccountsList_IsValid()
{
// Go: TestRoutePoolAndPerAccountWithOlderServer (routes_test.go:3571)
// An empty Accounts list means all traffic uses the global pool.
var opts = MakeClusterOpts();
opts.Cluster!.Accounts = [];
opts.Cluster.Accounts.ShouldBeEmpty();
}
// ---------------------------------------------------------------
// Go: TestRoutePerAccountGossipWorks (routes_test.go:2867)
// Gossip propagates per-account route topology to new peers.
// ---------------------------------------------------------------
[Fact]
public async Task RouteGossip_NewPeer_ReceivesTopologyFromExistingCluster()
{
// Go: TestRoutePerAccountGossipWorks (routes_test.go:2867)
// When a third server joins a two-server cluster, it learns the topology
// via gossip. In the .NET model this is verified by checking route counts.
var clusterName = Guid.NewGuid().ToString("N");
var a = await StartAsync(MakeClusterOpts(clusterName));
var b = await StartAsync(MakeClusterOpts(clusterName, a.Server.ClusterListen));
await WaitForRoutes(a.Server, b.Server);
// C connects only to A; gossip should let it discover B.
var c = await StartAsync(MakeClusterOpts(clusterName, a.Server.ClusterListen));
try
{
// Wait for C to connect to at least one peer.
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(8));
while (!timeout.IsCancellationRequested &&
Interlocked.Read(ref c.Server.Stats.Routes) == 0)
{
await Task.Delay(100, timeout.Token)
.ContinueWith(_ => { }, TaskScheduler.Default);
}
Interlocked.Read(ref c.Server.Stats.Routes).ShouldBeGreaterThan(0,
"Server C should have formed a route");
}
finally
{
await DisposeAll(a, b, c);
}
}
// ---------------------------------------------------------------
// Go: TestRouteConfig (routes_test.go:86)
// ClusterOptions are parsed and validated correctly.
// ---------------------------------------------------------------
[Fact]
public void RouteConfig_ClusterOptions_DefaultsAreCorrect()
{
// Go: TestRouteConfig (routes_test.go:86)
// Defaults: host 0.0.0.0, port 6222, pool_size 3, no accounts.
var opts = new ClusterOptions();
opts.Host.ShouldBe("0.0.0.0");
opts.Port.ShouldBe(6222);
opts.PoolSize.ShouldBe(3);
opts.Accounts.ShouldBeEmpty();
opts.Routes.ShouldBeEmpty();
}
[Fact]
public void RouteConfig_PoolSizeNegativeOne_MeansNoPooling()
{
// Go: TestRoutePool — pool_size: -1 means single route (no pooling)
// Go uses -1 as "no pooling" sentinel. .NET: PoolSize=1 is the minimum.
// ComputeRoutePoolIdx with pool_size <= 1 always returns 0.
var idx = RouteManager.ComputeRoutePoolIdx(-1, "any-account");
idx.ShouldBe(0);
}
// ---------------------------------------------------------------
// Go: TestRoutePoolWithOlderServerConnectAndReconnect (routes_test.go:3669)
// Reconnect after disconnect re-establishes the pool.
// ---------------------------------------------------------------
[Fact]
public async Task RoutePool_AfterDisconnect_ReconnectsAutomatically()
{
// Go: TestRoutePoolWithOlderServerConnectAndReconnect (routes_test.go:3669)
var clusterName = Guid.NewGuid().ToString("N");
var a = await StartAsync(MakeClusterOpts(clusterName));
var optsB = MakeClusterOpts(clusterName, a.Server.ClusterListen);
var b = await StartAsync(optsB);
await WaitForRoutes(a.Server, b.Server);
Interlocked.Read(ref a.Server.Stats.Routes).ShouldBeGreaterThan(0);
// Dispose B — routes should drop on A.
await DisposeAll(b);
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
while (!timeout.IsCancellationRequested && Interlocked.Read(ref a.Server.Stats.Routes) > 0)
{
await Task.Delay(50, timeout.Token)
.ContinueWith(_ => { }, TaskScheduler.Default);
}
Interlocked.Read(ref a.Server.Stats.Routes).ShouldBe(0L);
await DisposeAll(a);
}
}