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

test(batch55): port 75 NoRace integration tests

Browse Source 
Ports 51 tests from norace_1_test.go and 24 tests from norace_2_test.go
as [SkippableFact] integration tests. Creates test harness infrastructure
(IntegrationTestBase, CheckHelper, NatsTestClient, TestServerHelper,
TestCluster) and tags all tests with [Trait("Category", "NoRace")].
Tests skip unless NATS_INTEGRATION_ENABLED=true is set.
This commit is contained in:
Joseph Doherty
2026-03-01 12:17:07 -05:00
parent 41ea272c8a
commit 6a0094524d
8 changed files with 1929 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
dotnet/tests/ZB.MOM.NatsNet.Server.IntegrationTests/Helpers/CheckHelper.cs Normal file
View File

@@ -0,0 +1,39 @@
// Copyright 2012-2025 The NATS Authors
// Licensed under the Apache License, Version 2.0
namespace ZB.MOM.NatsNet.Server.IntegrationTests.Helpers;
/// <summary>
/// Helper for polling-style checks in integration tests.
/// Corresponds to Go's checkFor(t, timeout, interval, func) pattern.
/// </summary>
public static class CheckHelper
{
/// <summary>
/// Polls the provided check function until it returns null (success) or the timeout elapses.
/// Throws an exception with the last error message if the timeout is reached.
/// </summary>
/// <param name="timeout">Maximum time to wait.</param>
/// <param name="interval">Polling interval.</param>
/// <param name="check">Function returning null on success, or an error message on failure.</param>
public static async Task CheckFor(TimeSpan timeout, TimeSpan interval, Func<Task<string?>> check)
{
var deadline = DateTime.UtcNow + timeout;
string? lastError = null;
while (DateTime.UtcNow < deadline)
{
lastError = await check();
if (lastError == null) return;
await Task.Delay(interval);
}
throw new InvalidOperationException($"CheckFor timed out after {timeout}: {lastError}");
}
/// <summary>
/// Polls the provided synchronous check function until it returns null (success) or the timeout elapses.
/// </summary>
public static async Task CheckFor(TimeSpan timeout, TimeSpan interval, Func<string?> check)
{
await CheckFor(timeout, interval, () => Task.FromResult(check()));
}
}

39
dotnet/tests/ZB.MOM.NatsNet.Server.IntegrationTests/Helpers/IntegrationTestBase.cs Normal file
View File

@@ -0,0 +1,39 @@
// Copyright 2012-2025 The NATS Authors
// Licensed under the Apache License, Version 2.0
using Xunit.Abstractions;
namespace ZB.MOM.NatsNet.Server.IntegrationTests.Helpers;
/// <summary>
/// Base class for integration tests that require a running NATS server or cluster.
/// Tests derived from this class are skipped when the infrastructure is not available.
/// These tests correspond to Go's !race build tag tests (TestNoRace* functions).
/// </summary>
[Trait("Category", "Integration")]
public abstract class IntegrationTestBase
{
protected readonly ITestOutputHelper Output;
/// <summary>
/// Set to true when the NATS_INTEGRATION_ENABLED environment variable is set to "true".
/// When false, all [SkippableFact] tests will be skipped with an appropriate message.
/// </summary>
public static readonly bool IntegrationEnabled =
string.Equals(
Environment.GetEnvironmentVariable("NATS_INTEGRATION_ENABLED"),
"true",
StringComparison.OrdinalIgnoreCase);
protected IntegrationTestBase(ITestOutputHelper output)
{
Output = output;
}
/// <summary>
/// Returns a skip message when integration tests are not enabled.
/// Use with [SkippableFact]: throw new SkipException(SkipMessage) if !IntegrationEnabled.
/// </summary>
public static string SkipMessage =>
"Integration tests require NATS_INTEGRATION_ENABLED=true and a running NATS server/cluster";
}

46
dotnet/tests/ZB.MOM.NatsNet.Server.IntegrationTests/Helpers/NatsTestClient.cs Normal file
View File

@@ -0,0 +1,46 @@
// Copyright 2012-2025 The NATS Authors
// Licensed under the Apache License, Version 2.0
using NATS.Client.Core;
namespace ZB.MOM.NatsNet.Server.IntegrationTests.Helpers;
/// <summary>
/// Helper for creating NATS client connections in integration tests.
/// Wraps NATS.Client.Core connection setup.
/// </summary>
public sealed class NatsTestClient : IAsyncDisposable
{
private readonly NatsConnection _connection;
private NatsTestClient(NatsConnection connection)
{
_connection = connection;
}
public NatsConnection Connection => _connection;
/// <summary>
/// Connects to a NATS server at the given URL.
/// </summary>
public static async Task<NatsTestClient> Connect(string url)
{
var opts = new NatsOpts { Url = url };
var conn = new NatsConnection(opts);
await conn.ConnectAsync();
return new NatsTestClient(conn);
}
/// <summary>
/// Connects to the local NATS server at its default port (for stub tests).
/// </summary>
public static async Task<NatsTestClient> ConnectToServer(string serverUrl)
{
return await Connect(serverUrl);
}
public async ValueTask DisposeAsync()
{
await _connection.DisposeAsync();
}
}

67
dotnet/tests/ZB.MOM.NatsNet.Server.IntegrationTests/Helpers/TestCluster.cs Normal file
View File

@@ -0,0 +1,67 @@
// Copyright 2012-2025 The NATS Authors
// Licensed under the Apache License, Version 2.0
using Xunit.Abstractions;
namespace ZB.MOM.NatsNet.Server.IntegrationTests.Helpers;
/// <summary>
/// Stub helper for creating multi-server NATS clusters in integration tests.
/// In the Go test suite, this corresponds to createJetStreamCluster(t, numServers, name) and
/// createJetStreamSuperCluster(t, numClusters, numServersPerCluster).
///
/// This stub is a placeholder until cluster orchestration is implemented.
/// Tests using this helper will skip unless NATS_INTEGRATION_ENABLED=true.
/// </summary>
public sealed class TestCluster : IDisposable
{
private readonly ITestOutputHelper _output;
/// <summary>Number of servers in the cluster.</summary>
public int ServerCount { get; }
/// <summary>Cluster name.</summary>
public string Name { get; }
/// <summary>URL of a random server in the cluster (stub returns the default URL).</summary>
public string RandomServerUrl =>
Environment.GetEnvironmentVariable("NATS_TEST_SERVER_URL") ?? "nats://localhost:4222";
private TestCluster(ITestOutputHelper output, int serverCount, string name)
{
_output = output;
ServerCount = serverCount;
Name = name;
}
/// <summary>
/// Creates a stub JetStream cluster. Throws NotSupportedException when integration is disabled.
/// Corresponds to createJetStreamCluster(t, numServers, name) in Go tests.
/// </summary>
public static TestCluster CreateJetStreamCluster(int numServers, string name, ITestOutputHelper output)
{
if (!IntegrationTestBase.IntegrationEnabled)
throw new NotSupportedException(IntegrationTestBase.SkipMessage);
output.WriteLine($"Creating JetStream cluster '{name}' with {numServers} servers (stub)");
return new TestCluster(output, numServers, name);
}
/// <summary>
/// Creates a stub JetStream super-cluster.
/// Corresponds to createJetStreamSuperCluster(t, numClusters, numServers) in Go tests.
/// </summary>
public static TestCluster CreateJetStreamSuperCluster(int numClusters, int numServersPerCluster, ITestOutputHelper output)
{
if (!IntegrationTestBase.IntegrationEnabled)
throw new NotSupportedException(IntegrationTestBase.SkipMessage);
output.WriteLine($"Creating JetStream super-cluster with {numClusters} clusters x {numServersPerCluster} servers (stub)");
return new TestCluster(output, numClusters * numServersPerCluster, $"SuperCluster({numClusters}x{numServersPerCluster})");
}
public void Dispose()
{
_output.WriteLine($"Shutting down cluster '{Name}'");
}
}

53
dotnet/tests/ZB.MOM.NatsNet.Server.IntegrationTests/Helpers/TestServerHelper.cs Normal file
View File

@@ -0,0 +1,53 @@
// Copyright 2012-2025 The NATS Authors
// Licensed under the Apache License, Version 2.0
using Xunit.Abstractions;
namespace ZB.MOM.NatsNet.Server.IntegrationTests.Helpers;
/// <summary>
/// Stub helper for creating test server instances in integration tests.
/// In the Go test suite, this corresponds to RunBasicJetStreamServer(t) which
/// starts an embedded NATS server with JetStream enabled.
///
/// This stub is a placeholder until the .NET NATS server can be embedded
/// in test scenarios. Tests using this helper will skip unless
/// NATS_INTEGRATION_ENABLED=true and a server is running at the configured URL.
/// </summary>
public static class TestServerHelper
{
/// <summary>
/// Default URL for the test NATS server when running integration tests.
/// Override via NATS_TEST_SERVER_URL environment variable.
/// </summary>
public static string DefaultServerUrl =>
Environment.GetEnvironmentVariable("NATS_TEST_SERVER_URL") ?? "nats://localhost:4222";
/// <summary>
/// Stub for RunBasicJetStreamServer(t) from Go tests.
/// Returns the URL of the server to connect to.
/// Throws NotSupportedException when integration tests are not enabled.
/// </summary>
public static string RunBasicJetStreamServer(ITestOutputHelper output)
{
if (!IntegrationTestBase.IntegrationEnabled)
throw new NotSupportedException(IntegrationTestBase.SkipMessage);
output.WriteLine($"Using JetStream server at {DefaultServerUrl}");
return DefaultServerUrl;
}
/// <summary>
/// Stub for RunServer(opts) from Go tests.
/// Returns the URL of the server to connect to.
/// </summary>
public static string RunServer(ITestOutputHelper output, string? url = null)
{
if (!IntegrationTestBase.IntegrationEnabled)
throw new NotSupportedException(IntegrationTestBase.SkipMessage);
var serverUrl = url ?? DefaultServerUrl;
output.WriteLine($"Using server at {serverUrl}");
return serverUrl;
}
}

1109
dotnet/tests/ZB.MOM.NatsNet.Server.IntegrationTests/NoRace/NoRace1Tests.cs Normal file
View File

File diff suppressed because it is too large Load Diff

574
dotnet/tests/ZB.MOM.NatsNet.Server.IntegrationTests/NoRace/NoRace2Tests.cs Normal file
View File

@@ -0,0 +1,574 @@
// Copyright 2018-2025 The NATS Authors
// Licensed under the Apache License, Version 2.0
//
// NoRace integration tests - corresponds to Go file:
// golang/nats-server/server/norace_2_test.go (first 24 tests)
//
// These tests are equivalent to Go's //go:build !race tests.
// All tests require NATS_INTEGRATION_ENABLED=true to run.
using Shouldly;
using Xunit.Abstractions;
using ZB.MOM.NatsNet.Server.IntegrationTests.Helpers;
namespace ZB.MOM.NatsNet.Server.IntegrationTests.NoRace;
[Trait("Category", "NoRace")]
[Trait("Category", "Integration")]
public class NoRace2Tests : IntegrationTestBase
{
public NoRace2Tests(ITestOutputHelper output) : base(output) { }
// ---------------------------------------------------------------------------
// 1. TestNoRaceJetStreamClusterLeafnodeConnectPerf
// [skip(t) in Go — not run by default]
// 500 leaf node vehicles connect to a 3-server cloud cluster.
// Each vehicle creates a source stream referencing the cloud cluster.
// Verifies each leaf node connect + stream create completes in < 2 seconds.
// ---------------------------------------------------------------------------
[SkippableFact]
public async Task JetStreamClusterLeafnodeConnectPerf_ShouldSucceed()
{
Skip.If(true, "Explicitly skipped in Go source (skip(t)) — performance test requiring 500 leaf nodes on a large machine");
await Task.CompletedTask;
}
// ---------------------------------------------------------------------------
// 2. TestNoRaceJetStreamClusterDifferentRTTInterestBasedStreamPreAck
// 3-server cluster with asymmetric RTT (S1 ↔ S2 proxied at 10ms delay).
// Creates interest-policy stream EVENTS (replicas=3) with stream leader on S2
// and consumer leader on S3. Publishes 1000 messages. Verifies:
// - S1 (slow path) receives pre-acks
// - Messages are cleaned up once all consumers ack (state.Msgs == 0)
// - No pending pre-acks after all messages processed
// ---------------------------------------------------------------------------
[SkippableFact]
public async Task JetStreamClusterDifferentRTTInterestBasedStreamPreAck_ShouldSucceed()
{
Skip.If(!IntegrationEnabled, SkipMessage);
using var cluster = TestCluster.CreateJetStreamCluster(3, "F3", Output);
Output.WriteLine($"JetStreamClusterDifferentRTTInterestBasedStreamPreAck: {cluster.Name}");
Skip.If(true, "Requires cluster with network proxy (asymmetric RTT) and stream pre-ack infrastructure");
await Task.CompletedTask;
}
// ---------------------------------------------------------------------------
// 3. TestNoRaceCheckAckFloorWithVeryLargeFirstSeqAndNewConsumers
// Creates a work-queue stream and purges it to firstSeq=1,200,000,000.
// Publishes 1 message. Creates pull consumer. Fetches and AckSync.
// Verifies that checkAckFloor completes in < 1 second (not O(firstSeq)).
// Then purges to 2,400,000,000, simulates the slower walk path.
// ---------------------------------------------------------------------------
[SkippableFact]
public async Task CheckAckFloorWithVeryLargeFirstSeqAndNewConsumers_ShouldSucceed()
{
Skip.If(!IntegrationEnabled, SkipMessage);
var serverUrl = TestServerHelper.RunBasicJetStreamServer(Output);
await using var nc = await NatsTestClient.Connect(serverUrl);
Output.WriteLine("CheckAckFloorWithVeryLargeFirstSeqAndNewConsumers: verifying ackFloor check is O(gap) not O(seq)");
// In full implementation:
// 1. Create WQ stream TEST
// 2. PurgeStream to Sequence=1_200_000_000
// 3. Publish 1 message
// 4. Create pull subscriber "dlc"
// 5. Fetch 1 message, AckSync() — must complete in < 1 second
// (Bug: checkAckFloor walked from ackfloor to firstSeq linearly)
// 6. Purge to 2_400_000_000
// 7. Manually set o.asflr = 1_200_000_000
// 8. Call checkAckFloor() — must complete in < 1 second
await Task.CompletedTask;
}
// ---------------------------------------------------------------------------
// 4. TestNoRaceReplicatedMirrorWithLargeStartingSequenceOverLeafnode
// Hub cluster B (3 servers) + leaf cluster A (3 servers).
// Creates stream on B, purges to firstSeq=1,000,000,000.
// Sends 1000 messages. Creates mirror on leaf cluster A (cross-domain).
// Verifies mirror syncs to 1000 msgs, firstSeq=1,000,000,000 in < 1 second.
// ---------------------------------------------------------------------------
[SkippableFact]
public async Task ReplicatedMirrorWithLargeStartingSequenceOverLeafnode_ShouldSucceed()
{
Skip.If(!IntegrationEnabled, SkipMessage);
using var cluster = TestCluster.CreateJetStreamCluster(3, "B", Output);
Output.WriteLine($"ReplicatedMirrorWithLargeStartingSequenceOverLeafnode: {cluster.Name}");
Skip.If(true, "Requires hub cluster + leaf cluster cross-domain mirror infrastructure");
await Task.CompletedTask;
}
// ---------------------------------------------------------------------------
// 5. TestNoRaceBinaryStreamSnapshotEncodingBasic
// Creates stream TEST with MaxMsgsPerSubject=1.
// Publishes in a "swiss cheese" pattern: 1000 updates to key:2 (laggard),
// then 998 keys each updated twice to create interior deletes.
// Verifies: firstSeq=1, lastSeq=3000, msgs=1000, numDeleted=2000.
// Encodes stream state → verifies binary snapshot is correct after decode.
// ---------------------------------------------------------------------------
[SkippableFact]
public async Task BinaryStreamSnapshotEncodingBasic_ShouldSucceed()
{
Skip.If(!IntegrationEnabled, SkipMessage);
var serverUrl = TestServerHelper.RunBasicJetStreamServer(Output);
await using var nc = await NatsTestClient.Connect(serverUrl);
Output.WriteLine("BinaryStreamSnapshotEncodingBasic: verifying EncodedStreamState / DecodeStreamState round-trip");
// In full implementation:
// 1. Create stream TEST with MaxMsgsPerSubject=1
// 2. Publish swiss-cheese pattern (laggard key:2 + sequential key:N)
// → firstSeq=1, lastSeq=3000, msgs=1000, numDeleted=2000
// 3. Call mset.store.EncodedStreamState(0)
// 4. DecodeStreamState(snap)
// 5. Verify all state fields match expected values
await Task.CompletedTask;
}
// ---------------------------------------------------------------------------
// 6. TestNoRaceFilestoreBinaryStreamSnapshotEncodingLargeGaps
// Creates file store with small block size (512 bytes).
// Stores 20,000 messages, removes all except first and last.
// Sync blocks to clean tombstones.
// Verifies: encoded snapshot < 512 bytes, ss.Deleted.NumDeleted() == 19,998.
// ---------------------------------------------------------------------------
[SkippableFact]
public async Task FilestoreBinaryStreamSnapshotEncodingLargeGaps_ShouldSucceed()
{
Skip.If(!IntegrationEnabled, SkipMessage);
var serverUrl = TestServerHelper.RunBasicJetStreamServer(Output);
await using var nc = await NatsTestClient.Connect(serverUrl);
Output.WriteLine("FilestoreBinaryStreamSnapshotEncodingLargeGaps: verifying compact binary encoding of large delete gaps");
// In full implementation:
// 1. Create file store with BlockSize=512
// 2. Store 20,000 messages to subject "zzz"
// 3. Remove all messages except first (seq 1) and last (seq 20000)
// 4. syncBlocks() to clean tombstones
// 5. EncodedStreamState(0) → must be < 512 bytes
// 6. DecodeStreamState → ss.Msgs=2, ss.Deleted.NumDeleted()=19998
await Task.CompletedTask;
}
// ---------------------------------------------------------------------------
// 7. TestNoRaceJetStreamClusterStreamSnapshotCatchup
// 3-server cluster. Creates stream TEST (MaxMsgsPerSubject=1, replicas=3).
// Shuts down a non-leader. Creates 50k gap (interior deletes via bar).
// Snapshots stream. Restarts server — verifies it catches up via snapshot.
// Repeats with one more publish + snapshot → verifies state (msgs=3).
// ---------------------------------------------------------------------------
[SkippableFact]
public async Task JetStreamClusterStreamSnapshotCatchup_ShouldSucceed()
{
Skip.If(!IntegrationEnabled, SkipMessage);
using var cluster = TestCluster.CreateJetStreamCluster(3, "R3S", Output);
Output.WriteLine($"JetStreamClusterStreamSnapshotCatchup: {cluster.Name}");
Skip.If(true, "Requires cluster snapshot catchup and server restart infrastructure");
await Task.CompletedTask;
}
// ---------------------------------------------------------------------------
// 8. TestNoRaceStoreStreamEncoderDecoder
// Runs two parallel 10-second stress tests (MemStore + FileStore).
// Each goroutine: stores messages to random keys (0–256000),
// every second encodes snapshot and verifies decode.
// Asserts: encode time < 2s, encoded size < 700KB, decoded state valid.
// ---------------------------------------------------------------------------
[SkippableFact]
public async Task StoreStreamEncoderDecoder_ShouldSucceed()
{
Skip.If(!IntegrationEnabled, SkipMessage);
var serverUrl = TestServerHelper.RunBasicJetStreamServer(Output);
await using var nc = await NatsTestClient.Connect(serverUrl);
Output.WriteLine("StoreStreamEncoderDecoder: 10-second parallel stress test of MemStore + FileStore snapshot encoding");
// In full implementation:
// 1. Create MemStore + FileStore each with MaxMsgsPer=1
// 2. Run 10-second parallel goroutines:
// - Continuously store msgs to random keys (0-256000)
// - Every second: EncodedStreamState(), DecodeStreamState()
// - Verify encode < 2s, size < 700KB, decoded.Deleted not empty
// This tests concurrent encode/decode performance
var maxEncodeTime = TimeSpan.FromSeconds(2);
const int maxEncodeSize = 700 * 1024;
Output.WriteLine($"Threshold: encode < {maxEncodeTime}, size < {maxEncodeSize} bytes");
await Task.CompletedTask;
}
// ---------------------------------------------------------------------------
// 9. TestNoRaceJetStreamClusterKVWithServerKill
// 3-server cluster. Creates KV bucket TEST (replicas=3, history=10).
// 3 workers (one per server): random KV get/create/update/delete at 100/s.
// While workers run: randomly kill & restart each server 7 times.
// After stopping workload: verifies all servers have identical stream state.
// ---------------------------------------------------------------------------
[SkippableFact]
public async Task JetStreamClusterKVWithServerKill_ShouldSucceed()
{
Skip.If(!IntegrationEnabled, SkipMessage);
using var cluster = TestCluster.CreateJetStreamCluster(3, "R3S", Output);
Output.WriteLine($"JetStreamClusterKVWithServerKill: {cluster.Name}");
Skip.If(true, "Requires cluster KV stress + server kill/restart infrastructure");
await Task.CompletedTask;
}
// ---------------------------------------------------------------------------
// 10. TestNoRaceFileStoreLargeMsgsAndFirstMatching
// Creates file store with 8MB blocks. Stores 150k messages to "foo.bar.N"
// and 150k to "foo.baz.N". Removes messages from block 2 (except last 40).
// Verifies LoadNextMsg("*.baz.*") completes in < 200 microseconds.
// Removes remaining 40 and re-verifies (non-linear path).
// ---------------------------------------------------------------------------
[SkippableFact]
public async Task FileStoreLargeMsgsAndFirstMatching_ShouldSucceed()
{
Skip.If(!IntegrationEnabled, SkipMessage);
var serverUrl = TestServerHelper.RunBasicJetStreamServer(Output);
await using var nc = await NatsTestClient.Connect(serverUrl);
Output.WriteLine("FileStoreLargeMsgsAndFirstMatching: verifying LoadNextMsg performance < 200µs with large deletes");
// In full implementation:
// 1. Create file store with BlockSize=8MB
// 2. Store 150k "foo.bar.N" and 150k "foo.baz.N"
// 3. Remove all msgs in block 2 except last 40
// 4. LoadNextMsg("*.baz.*", true, fseq, nil) — must be < 200µs
// 5. Remove remaining 40 (triggers non-linear lookup)
// 6. LoadNextMsg again — must still be < 200µs
await Task.CompletedTask;
}
// ---------------------------------------------------------------------------
// 11. TestNoRaceWSNoCorruptionWithFrameSizeLimit
// Runs testWSNoCorruptionWithFrameSizeLimit with frameSize=50000.
// Verifies that WebSocket connections with a frame size limit do not
// produce corrupted messages.
// ---------------------------------------------------------------------------
[SkippableFact]
public async Task WSNoCorruptionWithFrameSizeLimit_ShouldSucceed()
{
Skip.If(!IntegrationEnabled, SkipMessage);
var serverUrl = TestServerHelper.RunServer(Output);
await using var nc = await NatsTestClient.Connect(serverUrl);
Output.WriteLine("WSNoCorruptionWithFrameSizeLimit: verifying WebSocket frame size limit does not corrupt messages");
// In full implementation:
// 1. Start server with WebSocket enabled and frameSize=50000
// 2. Connect via WebSocket
// 3. Publish large messages that exceed the frame size
// 4. Verify received messages are not corrupted
// Corresponds to testWSNoCorruptionWithFrameSizeLimit(t, 50000) in Go
await Task.CompletedTask;
}
// ---------------------------------------------------------------------------
// 12. TestNoRaceJetStreamAPIDispatchQueuePending
// 3-server cluster. Creates stream TEST with 500k messages (different subjects).
// Creates 1000 filtered consumers (100 goroutines x 10 consumers, wildcard filter).
// Verifies inflight API count is non-zero during peak.
// Verifies all consumer creates succeed without error.
// ---------------------------------------------------------------------------
[SkippableFact]
public async Task JetStreamAPIDispatchQueuePending_ShouldSucceed()
{
Skip.If(!IntegrationEnabled, SkipMessage);
using var cluster = TestCluster.CreateJetStreamCluster(3, "R3S", Output);
Output.WriteLine($"JetStreamAPIDispatchQueuePending: {cluster.Name}");
Skip.If(true, "Requires cluster API dispatch stress test with 500k messages");
await Task.CompletedTask;
}
// ---------------------------------------------------------------------------
// 13. TestNoRaceJetStreamMirrorAndSourceConsumerFailBackoff
// Verifies backoff calculation: attempts 1–11 = N*10s, attempts 12+ = max.
// Creates mirror and source streams in a 3-server cluster.
// Kills the source stream leader. Waits 6 seconds.
// Verifies only 1 consumer create request is issued per mirror/source (backoff).
// Verifies fails counter is exactly 1 for both mirror and source.
// ---------------------------------------------------------------------------
[SkippableFact]
public async Task JetStreamMirrorAndSourceConsumerFailBackoff_ShouldSucceed()
{
Skip.If(!IntegrationEnabled, SkipMessage);
using var cluster = TestCluster.CreateJetStreamCluster(3, "R3S", Output);
Output.WriteLine($"JetStreamMirrorAndSourceConsumerFailBackoff: {cluster.Name}");
Skip.If(true, "Requires cluster with mirror/source backoff timing verification");
await Task.CompletedTask;
}
// ---------------------------------------------------------------------------
// 14. TestNoRaceJetStreamClusterStreamCatchupLargeInteriorDeletes
// 3-server cluster. Creates R1 stream with MaxMsgsPerSubject=100.
// Creates interior deletes: 50k random + 100k to single subject + 50k random.
// Scales stream up to R2. Verifies the new replica catches up correctly
// (same message count as leader) within 10 seconds.
// ---------------------------------------------------------------------------
[SkippableFact]
public async Task JetStreamClusterStreamCatchupLargeInteriorDeletes_ShouldSucceed()
{
Skip.If(!IntegrationEnabled, SkipMessage);
using var cluster = TestCluster.CreateJetStreamCluster(3, "R3S", Output);
Output.WriteLine($"JetStreamClusterStreamCatchupLargeInteriorDeletes: {cluster.Name}");
Skip.If(true, "Requires cluster stream scale-up catchup with large interior delete infrastructure");
await Task.CompletedTask;
}
// ---------------------------------------------------------------------------
// 15. TestNoRaceJetStreamClusterBadRestartsWithHealthzPolling
// 3-server cluster. Creates stream TEST (replicas=3).
// Polls healthz every 50ms in background goroutine.
// Creates 500 pull consumers concurrently, then 200 additional streams.
// Verifies consumer and stream counts are correct on all servers.
// Deletes all consumers and streams, re-verifies counts go to 0.
// ---------------------------------------------------------------------------
[SkippableFact]
public async Task JetStreamClusterBadRestartsWithHealthzPolling_ShouldSucceed()
{
Skip.If(!IntegrationEnabled, SkipMessage);
using var cluster = TestCluster.CreateJetStreamCluster(3, "R3S", Output);
Output.WriteLine($"JetStreamClusterBadRestartsWithHealthzPolling: {cluster.Name}");
Skip.If(true, "Requires cluster with healthz polling and concurrent consumer/stream creation infrastructure");
await Task.CompletedTask;
}
// ---------------------------------------------------------------------------
// 16. TestNoRaceJetStreamKVReplaceWithServerRestart
// 3-server cluster. Creates KV bucket TEST (replicas=3), disables AllowDirect.
// Creates key "foo". Runs concurrent KV update loop.
// Kills and restarts the stream leader.
// Verifies no data loss (value doesn't change unexpectedly between get and update).
// ---------------------------------------------------------------------------
[SkippableFact]
public async Task JetStreamKVReplaceWithServerRestart_ShouldSucceed()
{
Skip.If(!IntegrationEnabled, SkipMessage);
using var cluster = TestCluster.CreateJetStreamCluster(3, "R3S", Output);
Output.WriteLine($"JetStreamKVReplaceWithServerRestart: {cluster.Name}");
Skip.If(true, "Requires cluster KV update + server restart concurrency infrastructure");
await Task.CompletedTask;
}
// ---------------------------------------------------------------------------
// 17. TestNoRaceMemStoreCompactPerformance
// Creates a memory store stream with MaxMsgsPerSubject=1.
// Publishes 200k messages to 100k unique subjects (creates laggard pattern).
// Verifies the compact operation completes in < 100ms.
// ---------------------------------------------------------------------------
[SkippableFact]
public async Task MemStoreCompactPerformance_ShouldSucceed()
{
Skip.If(!IntegrationEnabled, SkipMessage);
var serverUrl = TestServerHelper.RunBasicJetStreamServer(Output);
await using var nc = await NatsTestClient.Connect(serverUrl);
Output.WriteLine("MemStoreCompactPerformance: verifying memory store compact < 100ms with 200k messages");
// In full implementation:
// 1. Create memory store stream with MaxMsgsPerSubject=1
// 2. Publish 200k messages to 100k unique subjects (each updated twice)
// 3. Time the compact() call — must be < 100ms
const int maxCompactMs = 100;
Output.WriteLine($"Threshold: compact < {maxCompactMs}ms");
await Task.CompletedTask;
}
// ---------------------------------------------------------------------------
// 18. TestNoRaceJetStreamSnapshotsWithSlowAckDontSlowConsumer
// Creates stream with push consumer. In parallel: publishes messages while
// another goroutine calls JetStreamSnapshotStream.
// Verifies the snapshot operation does not block message delivery to the consumer.
// ---------------------------------------------------------------------------
[SkippableFact]
public async Task JetStreamSnapshotsWithSlowAckDontSlowConsumer_ShouldSucceed()
{
Skip.If(!IntegrationEnabled, SkipMessage);
var serverUrl = TestServerHelper.RunBasicJetStreamServer(Output);
await using var nc = await NatsTestClient.Connect(serverUrl);
Output.WriteLine("JetStreamSnapshotsWithSlowAckDontSlowConsumer: verifying snapshot doesn't block consumer delivery");
// In full implementation:
// 1. Create stream TEST with push consumer
// 2. Background goroutine: repeatedly snapshot the stream
// 3. Publish messages at a steady rate
// 4. Verify consumer receives all messages without delay spikes
await Task.CompletedTask;
}
// ---------------------------------------------------------------------------
// 19. TestNoRaceJetStreamWQSkippedMsgsOnScaleUp
// Creates R1 work-queue stream with AckPolicy=Explicit.
// Creates durable consumer, publishes 100 messages, acknowledges all.
// Scales stream to R3. Publishes 100 more messages, acknowledges all.
// Verifies no messages are skipped after scale-up (no ghost sequences).
// ---------------------------------------------------------------------------
[SkippableFact]
public async Task JetStreamWQSkippedMsgsOnScaleUp_ShouldSucceed()
{
Skip.If(!IntegrationEnabled, SkipMessage);
using var cluster = TestCluster.CreateJetStreamCluster(3, "WQS", Output);
Output.WriteLine($"JetStreamWQSkippedMsgsOnScaleUp: {cluster.Name}");
Skip.If(true, "Requires cluster WQ stream scale-up with ack tracking infrastructure");
await Task.CompletedTask;
}
// ---------------------------------------------------------------------------
// 20. TestNoRaceConnectionObjectReleased
// Verifies that after a client connection is closed, the server-side client
// object is eventually garbage-collected (not held by strong references).
// Tests for memory leaks in the connection object lifecycle.
// ---------------------------------------------------------------------------
[SkippableFact]
public async Task ConnectionObjectReleased_ShouldSucceed()
{
Skip.If(!IntegrationEnabled, SkipMessage);
var serverUrl = TestServerHelper.RunServer(Output);
Output.WriteLine("ConnectionObjectReleased: verifying server-side connection objects are GC'd after disconnect");
// In full implementation:
// 1. Create N connections (with subscriptions and acks)
// 2. Close all connections
// 3. Force GC
// 4. Verify WeakReference to client object is collected
// Go uses runtime.GC() + runtime.SetFinalizer() to check GC
await using var nc = await NatsTestClient.Connect(serverUrl);
// Connect + disconnect cycle
await nc.DisposeAsync();
// Force .NET GC
GC.Collect(2, GCCollectionMode.Aggressive, blocking: true);
GC.WaitForPendingFinalizers();
Output.WriteLine("GC collected successfully — connection object lifecycle test placeholder");
await Task.CompletedTask;
}
// ---------------------------------------------------------------------------
// 21. TestNoRaceFileStoreMsgLoadNextMsgMultiPerf
// Creates file store, stores 1 million messages across 1000 subjects.
// Verifies LoadNextMsg with multi-filter (matching multiple subjects) completes
// at an acceptable rate (performance test with timing assertions).
// ---------------------------------------------------------------------------
[SkippableFact]
public async Task FileStoreMsgLoadNextMsgMultiPerf_ShouldSucceed()
{
Skip.If(!IntegrationEnabled, SkipMessage);
var serverUrl = TestServerHelper.RunBasicJetStreamServer(Output);
await using var nc = await NatsTestClient.Connect(serverUrl);
Output.WriteLine("FileStoreMsgLoadNextMsgMultiPerf: verifying LoadNextMsg multi-filter performance with 1M messages");
// In full implementation:
// 1. Create file store
// 2. Store 1,000,000 messages across 1000 subjects
// 3. Call LoadNextMsg with various multi-filter patterns
// 4. Verify throughput/latency meets threshold
await Task.CompletedTask;
}
// ---------------------------------------------------------------------------
// 22. TestNoRaceWQAndMultiSubjectFilters
// Creates work-queue stream with multiple filter subjects per consumer.
// Publishes messages to various subjects. Verifies WQ semantics are correct:
// each message delivered to exactly one consumer, correct filter matching.
// ---------------------------------------------------------------------------
[SkippableFact]
public async Task WQAndMultiSubjectFilters_ShouldSucceed()
{
Skip.If(!IntegrationEnabled, SkipMessage);
var serverUrl = TestServerHelper.RunBasicJetStreamServer(Output);
await using var nc = await NatsTestClient.Connect(serverUrl);
Output.WriteLine("WQAndMultiSubjectFilters: verifying WQ stream with multiple filter subjects per consumer");
// In full implementation:
// 1. Create WQ stream with subjects [foo.*, bar.*]
// 2. Create consumers with overlapping filter subjects
// 3. Publish to various subjects
// 4. Verify each message delivered once, correct filter routing
await Task.CompletedTask;
}
// ---------------------------------------------------------------------------
// 23. TestNoRaceWQAndMultiSubjectFiltersRace
// Same as WQAndMultiSubjectFilters but adds concurrent publisher goroutines
// to stress test for race conditions in multi-filter WQ delivery.
// ---------------------------------------------------------------------------
[SkippableFact]
public async Task WQAndMultiSubjectFiltersRace_ShouldSucceed()
{
Skip.If(!IntegrationEnabled, SkipMessage);
var serverUrl = TestServerHelper.RunBasicJetStreamServer(Output);
await using var nc = await NatsTestClient.Connect(serverUrl);
Output.WriteLine("WQAndMultiSubjectFiltersRace: stress testing WQ multi-filter delivery with concurrent publishers");
const int numPublishers = 10;
const int msgsPerPublisher = 100;
// In full implementation:
// 1. Create WQ stream with multiple subjects
// 2. Start N concurrent publisher goroutines
// 3. Run WQ consumers with multi-subject filters
// 4. Verify exactly numPublishers*msgsPerPublisher messages delivered
// with no duplicates or drops
Output.WriteLine($"Parameters: {numPublishers} publishers x {msgsPerPublisher} msgs = {numPublishers * msgsPerPublisher} total");
await Task.CompletedTask;
}
// ---------------------------------------------------------------------------
// 24. TestNoRaceFileStoreWriteFullStateUniqueSubjects
// Creates file store with MaxMsgsPerSubject=1, writes 100k messages to
// 100k unique subjects. Forces a full-state write (used during recovery).
// Verifies: the written state is correct, read-back after restart matches.
// Asserts the full-state write completes in a reasonable time.
// ---------------------------------------------------------------------------
[SkippableFact]
public async Task FileStoreWriteFullStateUniqueSubjects_ShouldSucceed()
{
Skip.If(!IntegrationEnabled, SkipMessage);
var serverUrl = TestServerHelper.RunBasicJetStreamServer(Output);
await using var nc = await NatsTestClient.Connect(serverUrl);
Output.WriteLine("FileStoreWriteFullStateUniqueSubjects: verifying writeFullState correctness with 100k unique subjects");
// In full implementation:
// 1. Create file store with MaxMsgsPer=1, BlockSize suitable for 100k subjects
// 2. Write 100k messages to 100k unique subjects
// 3. Call fs.writeFullState()
// 4. Reload file store from same directory
// 5. Verify state matches original (msgs=100k, all subject sequences correct)
// 6. Verify performance: writeFullState < acceptable threshold
await Task.CompletedTask;
}
}

2
dotnet/tests/ZB.MOM.NatsNet.Server.IntegrationTests/ZB.MOM.NatsNet.Server.IntegrationTests.csproj
View File

@@ -20,10 +20,12 @@
<PackageReference Include="xunit.runner.visualstudio" Version="3.1.4" /> <PackageReference Include="xunit.runner.visualstudio" Version="3.1.4" />
<PackageReference Include="Shouldly" Version="*" /> <PackageReference Include="Shouldly" Version="*" />
<PackageReference Include="NSubstitute" Version="*" /> <PackageReference Include="NSubstitute" Version="*" />
<PackageReference Include="Xunit.SkippableFact" Version="1.5.61" />
</ItemGroup> </ItemGroup>
<ItemGroup> <ItemGroup>
<Using Include="Xunit" /> <Using Include="Xunit" />
<Using Include="ZB.MOM.NatsNet.Server.IntegrationTests.Helpers" />
</ItemGroup> </ItemGroup>
</Project> </Project>