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

refactor: extract NATS.Server.JetStream.Tests project

Browse Source 
Move 225 JetStream-related test files from NATS.Server.Tests into a
dedicated NATS.Server.JetStream.Tests project. This includes root-level
JetStream*.cs files, storage test files (FileStore, MemStore,
StreamStoreContract), and the full JetStream/ subfolder tree (Api,
Cluster, Consumers, MirrorSource, Snapshots, Storage, Streams).

Updated all namespaces, added InternalsVisibleTo, registered in the
solution file, and added the JETSTREAM_INTEGRATION_MATRIX define.
This commit is contained in:
Joseph Doherty
2026-03-12 15:58:10 -04:00
parent 36b9dfa654
commit 78b4bc2486
228 changed files with 253 additions and 227 deletions
Download Patch File Download Diff File Expand all files Collapse all files

206
tests/NATS.Server.JetStream.Tests/JetStream/Api/AdvisoryEventTests.cs Normal file
View File

@@ -0,0 +1,206 @@
// Go reference: jetstream_api.go — advisory event publication for stream/consumer lifecycle.
// Advisory subjects use the pattern $JS.EVENT.ADVISORY.{type}.{stream}[.{consumer}].
using NATS.Server.Events;
using NATS.Server.JetStream.Api;
namespace NATS.Server.JetStream.Tests.JetStream.Api;
public class AdvisoryEventTests
{
private static (AdvisoryPublisher Publisher, List<(string Subject, object Body)> Published) CreatePublisher()
{
var published = new List<(string Subject, object Body)>();
var publisher = new AdvisoryPublisher((s, b) => published.Add((s, b)));
return (publisher, published);
}
// Go reference: jetstream_api.go — stream created advisory on $JS.EVENT.ADVISORY.STREAM.CREATED.{stream}.
[Fact]
public void StreamCreated_publishes_advisory_to_correct_subject()
{
var (publisher, published) = CreatePublisher();
publisher.StreamCreated("ORDERS");
published.Count.ShouldBe(1);
published[0].Subject.ShouldBe("$JS.EVENT.ADVISORY.STREAM.CREATED.ORDERS");
}
// Go reference: jetstream_api.go — stream deleted advisory includes stream name in subject.
[Fact]
public void StreamDeleted_publishes_advisory_with_stream_name()
{
var (publisher, published) = CreatePublisher();
publisher.StreamDeleted("PAYMENTS");
published.Count.ShouldBe(1);
published[0].Subject.ShouldBe("$JS.EVENT.ADVISORY.STREAM.DELETED.PAYMENTS");
var evt = published[0].Body.ShouldBeOfType<AdvisoryEvent>();
evt.Stream.ShouldBe("PAYMENTS");
}
// Go reference: jetstream_api.go — stream updated advisory carries optional detail payload.
[Fact]
public void StreamUpdated_publishes_advisory_with_detail()
{
var (publisher, published) = CreatePublisher();
var detail = new { Reason = "config_change" };
publisher.StreamUpdated("EVENTS", detail);
published.Count.ShouldBe(1);
published[0].Subject.ShouldBe("$JS.EVENT.ADVISORY.STREAM.UPDATED.EVENTS");
var evt = published[0].Body.ShouldBeOfType<AdvisoryEvent>();
evt.Detail.ShouldNotBeNull();
}
// Go reference: jetstream_api.go — consumer created advisory on $JS.EVENT.ADVISORY.CONSUMER.CREATED.{stream}.{consumer}.
[Fact]
public void ConsumerCreated_publishes_advisory_with_stream_and_consumer()
{
var (publisher, published) = CreatePublisher();
publisher.ConsumerCreated("ORDERS", "push-consumer");
published.Count.ShouldBe(1);
published[0].Subject.ShouldBe("$JS.EVENT.ADVISORY.CONSUMER.CREATED.ORDERS.push-consumer");
var evt = published[0].Body.ShouldBeOfType<AdvisoryEvent>();
evt.Stream.ShouldBe("ORDERS");
evt.Consumer.ShouldBe("push-consumer");
}
// Go reference: jetstream_api.go — consumer deleted advisory type field identifies event kind.
[Fact]
public void ConsumerDeleted_publishes_advisory_with_correct_type()
{
var (publisher, published) = CreatePublisher();
publisher.ConsumerDeleted("ORDERS", "my-consumer");
published.Count.ShouldBe(1);
var evt = published[0].Body.ShouldBeOfType<AdvisoryEvent>();
evt.Type.ShouldBe("io.nats.jetstream.advisory.consumer_deleted");
}
// Go reference: jetstream_api.go — publish count tracks all emitted advisories atomically.
[Fact]
public void PublishCount_increments_for_each_advisory()
{
var (publisher, _) = CreatePublisher();
publisher.PublishCount.ShouldBe(0);
publisher.StreamCreated("S1");
publisher.PublishCount.ShouldBe(1);
publisher.StreamDeleted("S1");
publisher.PublishCount.ShouldBe(2);
publisher.ConsumerCreated("S1", "C1");
publisher.PublishCount.ShouldBe(3);
}
// Go reference: jetstream_api.go — each advisory type has its own descriptive type string.
[Fact]
public void Advisory_event_has_correct_type_field()
{
var (publisher, published) = CreatePublisher();
publisher.StreamCreated("S");
published[0].Body.ShouldBeOfType<AdvisoryEvent>().Type
.ShouldBe("io.nats.jetstream.advisory.stream_created");
publisher.StreamDeleted("S");
published[1].Body.ShouldBeOfType<AdvisoryEvent>().Type
.ShouldBe("io.nats.jetstream.advisory.stream_deleted");
publisher.StreamUpdated("S");
published[2].Body.ShouldBeOfType<AdvisoryEvent>().Type
.ShouldBe("io.nats.jetstream.advisory.stream_updated");
publisher.ConsumerCreated("S", "C");
published[3].Body.ShouldBeOfType<AdvisoryEvent>().Type
.ShouldBe("io.nats.jetstream.advisory.consumer_created");
publisher.ConsumerDeleted("S", "C");
published[4].Body.ShouldBeOfType<AdvisoryEvent>().Type
.ShouldBe("io.nats.jetstream.advisory.consumer_deleted");
}
// Go reference: jetstream_api.go — advisory timestamps use UTC to ensure cross-cluster consistency.
[Fact]
public void Advisory_event_has_utc_timestamp()
{
var (publisher, published) = CreatePublisher();
var before = DateTime.UtcNow;
publisher.StreamCreated("TEST");
var after = DateTime.UtcNow;
var evt = published[0].Body.ShouldBeOfType<AdvisoryEvent>();
evt.TimeStamp.Kind.ShouldBe(DateTimeKind.Utc);
evt.TimeStamp.ShouldBeGreaterThanOrEqualTo(before);
evt.TimeStamp.ShouldBeLessThanOrEqualTo(after);
}
// Go reference: jetstream_api.go — advisory subjects are derived from EventSubjects constants.
[Fact]
public void Advisory_subjects_format_correctly()
{
string.Format(EventSubjects.JsAdvisoryStreamCreated, "MY_STREAM")
.ShouldBe("$JS.EVENT.ADVISORY.STREAM.CREATED.MY_STREAM");
string.Format(EventSubjects.JsAdvisoryStreamDeleted, "MY_STREAM")
.ShouldBe("$JS.EVENT.ADVISORY.STREAM.DELETED.MY_STREAM");
string.Format(EventSubjects.JsAdvisoryStreamUpdated, "MY_STREAM")
.ShouldBe("$JS.EVENT.ADVISORY.STREAM.UPDATED.MY_STREAM");
string.Format(EventSubjects.JsAdvisoryConsumerCreated, "MY_STREAM", "MY_CONSUMER")
.ShouldBe("$JS.EVENT.ADVISORY.CONSUMER.CREATED.MY_STREAM.MY_CONSUMER");
string.Format(EventSubjects.JsAdvisoryConsumerDeleted, "MY_STREAM", "MY_CONSUMER")
.ShouldBe("$JS.EVENT.ADVISORY.CONSUMER.DELETED.MY_STREAM.MY_CONSUMER");
string.Format(EventSubjects.JsAdvisoryStreamSnapshotCreated, "MY_STREAM")
.ShouldBe("$JS.EVENT.ADVISORY.STREAM.SNAPSHOT_CREATE.MY_STREAM");
string.Format(EventSubjects.JsAdvisoryStreamSnapshotCompleted, "MY_STREAM")
.ShouldBe("$JS.EVENT.ADVISORY.STREAM.SNAPSHOT_COMPLETE.MY_STREAM");
string.Format(EventSubjects.JsAdvisoryStreamRestoreCreated, "MY_STREAM")
.ShouldBe("$JS.EVENT.ADVISORY.STREAM.RESTORE_CREATE.MY_STREAM");
string.Format(EventSubjects.JsAdvisoryStreamRestoreCompleted, "MY_STREAM")
.ShouldBe("$JS.EVENT.ADVISORY.STREAM.RESTORE_COMPLETE.MY_STREAM");
string.Format(EventSubjects.JsAdvisoryStreamLeaderElected, "MY_STREAM")
.ShouldBe("$JS.EVENT.ADVISORY.STREAM.LEADER_ELECTED.MY_STREAM");
string.Format(EventSubjects.JsAdvisoryStreamQuorumLost, "MY_STREAM")
.ShouldBe("$JS.EVENT.ADVISORY.STREAM.QUORUM_LOST.MY_STREAM");
}
// Go reference: jetstream_api.go — full lifecycle sequence (create, update, delete) emits all advisories.
[Fact]
public void Multiple_advisories_all_published()
{
var (publisher, published) = CreatePublisher();
publisher.StreamCreated("LIFECYCLE");
publisher.StreamUpdated("LIFECYCLE", new { Reason = "retention_change" });
publisher.ConsumerCreated("LIFECYCLE", "worker");
publisher.ConsumerDeleted("LIFECYCLE", "worker");
publisher.StreamDeleted("LIFECYCLE");
published.Count.ShouldBe(5);
published[0].Subject.ShouldBe("$JS.EVENT.ADVISORY.STREAM.CREATED.LIFECYCLE");
published[1].Subject.ShouldBe("$JS.EVENT.ADVISORY.STREAM.UPDATED.LIFECYCLE");
published[2].Subject.ShouldBe("$JS.EVENT.ADVISORY.CONSUMER.CREATED.LIFECYCLE.worker");
published[3].Subject.ShouldBe("$JS.EVENT.ADVISORY.CONSUMER.DELETED.LIFECYCLE.worker");
published[4].Subject.ShouldBe("$JS.EVENT.ADVISORY.STREAM.DELETED.LIFECYCLE");
publisher.PublishCount.ShouldBe(5);
}
}

124
tests/NATS.Server.JetStream.Tests/JetStream/Api/ApiEndpointParityTests.cs Normal file
View File

@@ -0,0 +1,124 @@
using NATS.Server.TestUtilities;
// Go reference: golang/nats-server/server/jetstream.go — $JS.API.* subject dispatch
// Covers create/info/update/delete for streams, create/info/list/delete for consumers,
// direct-get access, account info, and 404 routing for unknown subjects.
namespace NATS.Server.JetStream.Tests;
public class ApiEndpointParityTests
{
// Go ref: jsStreamCreateT handler — stream create persists config and info round-trips correctly.
[Fact]
public async Task Stream_create_info_update_delete_lifecycle()
{
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("EVENTS", "events.*");
var info = await fx.RequestLocalAsync("$JS.API.STREAM.INFO.EVENTS", "{}");
info.Error.ShouldBeNull();
info.StreamInfo.ShouldNotBeNull();
info.StreamInfo!.Config.Name.ShouldBe("EVENTS");
info.StreamInfo.Config.Subjects.ShouldContain("events.*");
var update = await fx.RequestLocalAsync(
"$JS.API.STREAM.UPDATE.EVENTS",
"{\"name\":\"EVENTS\",\"subjects\":[\"events.*\"],\"max_msgs\":100}");
update.Error.ShouldBeNull();
update.StreamInfo.ShouldNotBeNull();
update.StreamInfo!.Config.MaxMsgs.ShouldBe(100);
var delete = await fx.RequestLocalAsync("$JS.API.STREAM.DELETE.EVENTS", "{}");
delete.Error.ShouldBeNull();
delete.Success.ShouldBeTrue();
var infoAfterDelete = await fx.RequestLocalAsync("$JS.API.STREAM.INFO.EVENTS", "{}");
infoAfterDelete.Error.ShouldNotBeNull();
infoAfterDelete.Error!.Code.ShouldBe(404);
}
// Go ref: jsConsumerCreateT / jsConsumerInfoT handlers — consumer create then info returns config.
[Fact]
public async Task Consumer_create_info_list_delete_lifecycle()
{
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("ORDERS", "orders.*");
var create = await fx.CreateConsumerAsync("ORDERS", "MON", "orders.created");
create.Error.ShouldBeNull();
create.ConsumerInfo.ShouldNotBeNull();
create.ConsumerInfo!.Config.DurableName.ShouldBe("MON");
var info = await fx.RequestLocalAsync("$JS.API.CONSUMER.INFO.ORDERS.MON", "{}");
info.Error.ShouldBeNull();
info.ConsumerInfo.ShouldNotBeNull();
info.ConsumerInfo!.Config.FilterSubject.ShouldBe("orders.created");
var names = await fx.RequestLocalAsync("$JS.API.CONSUMER.NAMES.ORDERS", "{}");
names.Error.ShouldBeNull();
names.ConsumerNames.ShouldNotBeNull();
names.ConsumerNames.ShouldContain("MON");
var list = await fx.RequestLocalAsync("$JS.API.CONSUMER.LIST.ORDERS", "{}");
list.Error.ShouldBeNull();
list.ConsumerNames.ShouldNotBeNull();
list.ConsumerNames.ShouldContain("MON");
var del = await fx.RequestLocalAsync("$JS.API.CONSUMER.DELETE.ORDERS.MON", "{}");
del.Error.ShouldBeNull();
del.Success.ShouldBeTrue();
var infoAfterDelete = await fx.RequestLocalAsync("$JS.API.CONSUMER.INFO.ORDERS.MON", "{}");
infoAfterDelete.Error.ShouldNotBeNull();
infoAfterDelete.Error!.Code.ShouldBe(404);
}
// Go ref: jsDirectMsgGetT handler — direct get returns message payload at correct sequence.
[Fact]
public async Task Direct_get_returns_message_at_sequence()
{
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("LOGS", "logs.*");
var ack = await fx.PublishAndGetAckAsync("logs.app", "hello-direct");
var direct = await fx.RequestLocalAsync("$JS.API.DIRECT.GET.LOGS", $"{{\"seq\":{ack.Seq}}}");
direct.Error.ShouldBeNull();
direct.DirectMessage.ShouldNotBeNull();
direct.DirectMessage!.Sequence.ShouldBe(ack.Seq);
direct.DirectMessage.Payload.ShouldBe("hello-direct");
}
// Go ref: jsStreamNamesT / $JS.API.INFO handler — names list reflects created streams,
// account info reflects total stream and consumer counts.
[Fact]
public async Task Stream_names_and_account_info_reflect_state()
{
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("ALPHA", "alpha.*");
_ = await fx.CreateStreamAsync("BETA", ["beta.*"]);
_ = await fx.CreateConsumerAsync("ALPHA", "C1", "alpha.>");
_ = await fx.CreateConsumerAsync("BETA", "C2", "beta.>");
var names = await fx.RequestLocalAsync("$JS.API.STREAM.NAMES", "{}");
names.Error.ShouldBeNull();
names.StreamNames.ShouldNotBeNull();
names.StreamNames.ShouldContain("ALPHA");
names.StreamNames.ShouldContain("BETA");
var accountInfo = await fx.RequestLocalAsync("$JS.API.INFO", "{}");
accountInfo.Error.ShouldBeNull();
accountInfo.AccountInfo.ShouldNotBeNull();
accountInfo.AccountInfo!.Streams.ShouldBe(2);
accountInfo.AccountInfo.Consumers.ShouldBe(2);
}
// Go ref: JetStreamApiRouter dispatch — subjects not matching any handler return 404 error shape.
[Fact]
public async Task Unknown_api_subject_returns_404_error_response()
{
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("ORDERS", "orders.*");
var response = await fx.RequestLocalAsync("$JS.API.STREAM.FROBNICATE.ORDERS", "{}");
response.Error.ShouldNotBeNull();
response.Error!.Code.ShouldBe(404);
response.StreamInfo.ShouldBeNull();
response.ConsumerInfo.ShouldBeNull();
response.Success.ShouldBeFalse();
}
}

186
tests/NATS.Server.JetStream.Tests/JetStream/Api/ApiRateLimiterTests.cs Normal file
View File

@@ -0,0 +1,186 @@
// Go reference: jetstream_api.go — rate limiting via maxConcurrentRequests semaphore and
// request deduplication via the dedup cache keyed by Nats-Msg-Id header.
// The Go server uses a configurable semaphore (default 256) to throttle concurrent API
// requests, and caches responses for duplicate request IDs within a TTL window.
namespace NATS.Server.JetStream.Tests.JetStream.Api;
using NATS.Server.JetStream.Api;
public class ApiRateLimiterTests : IDisposable
{
private readonly ApiRateLimiter _limiter = new(maxConcurrent: 4);
public void Dispose() => _limiter.Dispose();
// Go reference: jetstream_api.go — semaphore.TryAcquire(0) used for non-blocking attempt.
[Fact]
public async Task TryAcquire_succeeds_when_slots_available()
{
var acquired = await _limiter.TryAcquireAsync();
acquired.ShouldBeTrue();
}
// Go reference: jetstream_api.go — when all slots are taken, new requests are rejected.
[Fact]
public async Task TryAcquire_fails_when_all_slots_taken()
{
// Fill all 4 slots.
for (var i = 0; i < 4; i++)
(await _limiter.TryAcquireAsync()).ShouldBeTrue();
// 5th attempt should fail.
var rejected = await _limiter.TryAcquireAsync();
rejected.ShouldBeFalse();
}
// Go reference: jetstream_api.go — releasing a slot allows a subsequent request to proceed.
[Fact]
public async Task Release_frees_slot_for_next_request()
{
// Fill all slots.
for (var i = 0; i < 4; i++)
(await _limiter.TryAcquireAsync()).ShouldBeTrue();
// Currently full.
(await _limiter.TryAcquireAsync()).ShouldBeFalse();
// Release one slot.
_limiter.Release();
// Now one slot is free.
(await _limiter.TryAcquireAsync()).ShouldBeTrue();
}
// Go reference: jetstream_api.go — active count reflects in-flight requests.
[Fact]
public async Task ActiveCount_tracks_concurrent_requests()
{
_limiter.ActiveCount.ShouldBe(0);
await _limiter.TryAcquireAsync();
await _limiter.TryAcquireAsync();
await _limiter.TryAcquireAsync();
_limiter.ActiveCount.ShouldBe(3);
}
// Go reference: jetstream_api.go — unknown request ID returns null (cache miss).
[Fact]
public void GetCachedResponse_returns_null_for_unknown_id()
{
var result = _limiter.GetCachedResponse("nonexistent-id");
result.ShouldBeNull();
}
// Go reference: jetstream_api.go — dedup cache stores response keyed by Nats-Msg-Id.
[Fact]
public void CacheResponse_and_get_returns_cached()
{
var response = JetStreamApiResponse.SuccessResponse();
_limiter.CacheResponse("req-001", response);
var cached = _limiter.GetCachedResponse("req-001");
cached.ShouldNotBeNull();
cached!.Success.ShouldBeTrue();
}
// Go reference: jetstream_api.go — dedup window expires after TTL (dedupWindow config).
[SlopwatchSuppress("SW004", "TTL expiry test requires real wall-clock time to elapse; no synchronisation primitive can replace observing a time-based cache eviction")]
[Fact]
public async Task GetCachedResponse_returns_null_after_ttl_expiry()
{
using var shortLimiter = new ApiRateLimiter(maxConcurrent: 4, dedupTtl: TimeSpan.FromMilliseconds(50));
var response = JetStreamApiResponse.SuccessResponse();
shortLimiter.CacheResponse("req-ttl", response);
// Verify it's cached before expiry.
shortLimiter.GetCachedResponse("req-ttl").ShouldNotBeNull();
// Wait for TTL to expire.
await Task.Delay(120);
// Should be null after expiry.
shortLimiter.GetCachedResponse("req-ttl").ShouldBeNull();
}
// Go reference: jetstream_api.go — null/empty Nats-Msg-Id is ignored for dedup.
[Fact]
public void CacheResponse_ignores_null_request_id()
{
var response = JetStreamApiResponse.SuccessResponse();
// These should not throw and should not increment the cache count.
_limiter.CacheResponse(null, response);
_limiter.CacheResponse("", response);
_limiter.CacheResponse(string.Empty, response);
_limiter.DedupCacheCount.ShouldBe(0);
_limiter.GetCachedResponse(null).ShouldBeNull();
_limiter.GetCachedResponse("").ShouldBeNull();
}
// Go reference: jetstream_api.go — periodic sweep removes expired dedup entries.
[SlopwatchSuppress("SW004", "TTL expiry test requires real wall-clock time to elapse; no synchronisation primitive can replace observing a time-based cache eviction")]
[Fact]
public async Task PurgeExpired_removes_old_entries()
{
using var shortLimiter = new ApiRateLimiter(maxConcurrent: 4, dedupTtl: TimeSpan.FromMilliseconds(50));
shortLimiter.CacheResponse("req-a", JetStreamApiResponse.SuccessResponse());
shortLimiter.CacheResponse("req-b", JetStreamApiResponse.SuccessResponse());
shortLimiter.CacheResponse("req-c", JetStreamApiResponse.SuccessResponse());
shortLimiter.DedupCacheCount.ShouldBe(3);
// Wait for all entries to expire.
await Task.Delay(120);
var removed = shortLimiter.PurgeExpired();
removed.ShouldBe(3);
shortLimiter.DedupCacheCount.ShouldBe(0);
}
// Go reference: jetstream_api.go — dedup cache count is observable.
[Fact]
public void DedupCacheCount_tracks_cached_entries()
{
_limiter.DedupCacheCount.ShouldBe(0);
_limiter.CacheResponse("req-1", JetStreamApiResponse.Ok());
_limiter.CacheResponse("req-2", JetStreamApiResponse.Ok());
_limiter.CacheResponse("req-3", JetStreamApiResponse.Ok());
_limiter.DedupCacheCount.ShouldBe(3);
}
// Go reference: jetstream_api.go — semaphore enforces max-concurrent across goroutines.
[Fact]
public async Task Concurrent_acquire_respects_max()
{
using var limiter = new ApiRateLimiter(maxConcurrent: 5);
// Spin up 10 tasks, only 5 should succeed.
var results = await Task.WhenAll(
Enumerable.Range(0, 10).Select(_ => limiter.TryAcquireAsync()));
var acquired = results.Count(r => r);
acquired.ShouldBe(5);
}
// Go reference: jetstream_api.go — default maxConcurrentRequests = 256.
[Fact]
public async Task Default_max_concurrent_is_256()
{
using var defaultLimiter = new ApiRateLimiter();
// Acquire 256 slots — all should succeed.
var tasks = Enumerable.Range(0, 256).Select(_ => defaultLimiter.TryAcquireAsync());
var results = await Task.WhenAll(tasks);
results.ShouldAllBe(r => r);
// 257th should fail.
var rejected = await defaultLimiter.TryAcquireAsync();
rejected.ShouldBeFalse();
}
}

241
tests/NATS.Server.JetStream.Tests/JetStream/Api/ClusteredApiTests.cs Normal file
View File

@@ -0,0 +1,241 @@
// Go reference: jetstream_cluster.go:7620-8265 — clustered stream/consumer API handlers
// propose to the meta RAFT group rather than applying locally to StreamManager/ConsumerManager.
using System.Text;
using NATS.Server.JetStream.Api.Handlers;
using NATS.Server.JetStream.Cluster;
namespace NATS.Server.JetStream.Tests.JetStream.Api;
public class ClusteredApiTests
{
// ---------------------------------------------------------------
// Stream clustered handlers
// ---------------------------------------------------------------
/// <summary>
/// A successful clustered create proposes to the meta group, resulting in a new stream
/// assignment tracked under the provided name.
/// Go reference: jetstream_cluster.go:7620 jsClusteredStreamRequest.
/// </summary>
[Fact]
public async Task HandleClusteredCreate_proposes_to_meta_group()
{
var metaGroup = new JetStreamMetaGroup(nodes: 1);
var payload = Encoding.UTF8.GetBytes("""{"name":"ORDERS","subjects":["orders.>"]}""");
var response = await StreamApiHandlers.HandleClusteredCreateAsync(
"$JS.API.STREAM.CREATE.ORDERS", payload, metaGroup, CancellationToken.None);
response.Error.ShouldBeNull();
response.Success.ShouldBeTrue();
metaGroup.GetStreamAssignment("ORDERS").ShouldNotBeNull();
}
/// <summary>
/// A duplicate clustered create for the same stream name returns an error response.
/// Go reference: jetstream_cluster.go — duplicate stream proposal returns error.
/// </summary>
[Fact]
public async Task HandleClusteredCreate_returns_error_for_duplicate()
{
var metaGroup = new JetStreamMetaGroup(nodes: 1);
var payload = Encoding.UTF8.GetBytes("""{"name":"ORDERS","subjects":["orders.>"]}""");
// First create succeeds.
var first = await StreamApiHandlers.HandleClusteredCreateAsync(
"$JS.API.STREAM.CREATE.ORDERS", payload, metaGroup, CancellationToken.None);
first.Error.ShouldBeNull();
// Second create for same name returns error.
var second = await StreamApiHandlers.HandleClusteredCreateAsync(
"$JS.API.STREAM.CREATE.ORDERS", payload, metaGroup, CancellationToken.None);
second.Error.ShouldNotBeNull();
second.Error!.Description.ShouldContain("ORDERS");
}
/// <summary>
/// When this node is not the meta-group leader, clustered create returns a not-leader error.
/// Go reference: jetstream_cluster.go:7620 — leader check before proposing.
/// </summary>
[Fact]
public async Task HandleClusteredCreate_returns_error_when_not_leader()
{
// selfIndex=2, leaderIndex defaults to 1 — this node is NOT the leader.
var metaGroup = new JetStreamMetaGroup(nodes: 3, selfIndex: 2);
var payload = Encoding.UTF8.GetBytes("""{"name":"ORDERS","subjects":["orders.>"]}""");
var response = await StreamApiHandlers.HandleClusteredCreateAsync(
"$JS.API.STREAM.CREATE.ORDERS", payload, metaGroup, CancellationToken.None);
response.Error.ShouldNotBeNull();
response.Error!.Code.ShouldBe(10003);
response.Error.Description.ShouldBe("not leader");
}
/// <summary>
/// Clustered update proposes a config change to an existing stream assignment.
/// Go reference: jetstream_cluster.go jsClusteredStreamUpdateRequest.
/// </summary>
[Fact]
public async Task HandleClusteredUpdate_updates_existing_stream()
{
var metaGroup = new JetStreamMetaGroup(nodes: 1);
// Create the stream first.
var createPayload = Encoding.UTF8.GetBytes("""{"name":"EVENTS","subjects":["events.>"]}""");
await StreamApiHandlers.HandleClusteredCreateAsync(
"$JS.API.STREAM.CREATE.EVENTS", createPayload, metaGroup, CancellationToken.None);
// Now update it with a max_msgs constraint.
var updatePayload = Encoding.UTF8.GetBytes("""{"name":"EVENTS","subjects":["events.>"],"max_msgs":500}""");
var response = await StreamApiHandlers.HandleClusteredUpdateAsync(
"$JS.API.STREAM.UPDATE.EVENTS", updatePayload, metaGroup, CancellationToken.None);
response.Error.ShouldBeNull();
response.Success.ShouldBeTrue();
// The assignment should still exist.
metaGroup.GetStreamAssignment("EVENTS").ShouldNotBeNull();
}
/// <summary>
/// Clustered delete proposes removal of a stream from the meta group.
/// Go reference: jetstream_cluster.go processStreamRemoval via meta leader.
/// </summary>
[Fact]
public async Task HandleClusteredDelete_proposes_deletion()
{
var metaGroup = new JetStreamMetaGroup(nodes: 1);
var createPayload = Encoding.UTF8.GetBytes("""{"name":"ORDERS","subjects":["orders.>"]}""");
await StreamApiHandlers.HandleClusteredCreateAsync(
"$JS.API.STREAM.CREATE.ORDERS", createPayload, metaGroup, CancellationToken.None);
metaGroup.GetStreamAssignment("ORDERS").ShouldNotBeNull();
var response = await StreamApiHandlers.HandleClusteredDeleteAsync(
"$JS.API.STREAM.DELETE.ORDERS", metaGroup, CancellationToken.None);
response.Error.ShouldBeNull();
response.Success.ShouldBeTrue();
metaGroup.GetStreamAssignment("ORDERS").ShouldBeNull();
}
/// <summary>
/// Clustered delete of a non-existent stream returns a 404 not-found error.
/// Go reference: jetstream_cluster.go — delete missing stream returns error.
/// </summary>
[Fact]
public async Task HandleClusteredDelete_returns_error_for_missing_stream()
{
var metaGroup = new JetStreamMetaGroup(nodes: 1);
var response = await StreamApiHandlers.HandleClusteredDeleteAsync(
"$JS.API.STREAM.DELETE.GHOST", metaGroup, CancellationToken.None);
response.Error.ShouldNotBeNull();
response.Error!.Code.ShouldBe(404);
}
// ---------------------------------------------------------------
// Consumer clustered handlers
// ---------------------------------------------------------------
/// <summary>
/// Clustered consumer create proposes to the meta group, adding the consumer to the
/// stream's assignment map.
/// Go reference: jetstream_cluster.go:8100 jsClusteredConsumerRequest.
/// </summary>
[Fact]
public async Task Consumer_clustered_create_proposes_to_meta()
{
var metaGroup = new JetStreamMetaGroup(nodes: 1);
// Create parent stream first.
await StreamApiHandlers.HandleClusteredCreateAsync(
"$JS.API.STREAM.CREATE.ORDERS",
Encoding.UTF8.GetBytes("""{"name":"ORDERS","subjects":["orders.>"]}"""),
metaGroup,
CancellationToken.None);
var consumerPayload = Encoding.UTF8.GetBytes("""{"durable_name":"MON","filter_subject":"orders.created"}""");
var response = await ConsumerApiHandlers.HandleClusteredCreateAsync(
"$JS.API.CONSUMER.CREATE.ORDERS.MON", consumerPayload, metaGroup, CancellationToken.None);
response.Error.ShouldBeNull();
response.Success.ShouldBeTrue();
metaGroup.GetConsumerAssignment("ORDERS", "MON").ShouldNotBeNull();
}
/// <summary>
/// Creating a consumer on a stream that does not exist in the meta group returns an error.
/// Go reference: jetstream_cluster.go — consumer proposal validates stream existence.
/// </summary>
[Fact]
public async Task Consumer_clustered_create_returns_error_for_missing_stream()
{
var metaGroup = new JetStreamMetaGroup(nodes: 1);
var payload = Encoding.UTF8.GetBytes("""{"durable_name":"MON","filter_subject":"orders.created"}""");
var response = await ConsumerApiHandlers.HandleClusteredCreateAsync(
"$JS.API.CONSUMER.CREATE.GHOST.MON", payload, metaGroup, CancellationToken.None);
response.Error.ShouldNotBeNull();
response.Error!.Description.ShouldContain("GHOST");
}
/// <summary>
/// Clustered consumer delete removes the consumer from the stream assignment.
/// Go reference: jetstream_cluster.go processConsumerRemoval via meta leader.
/// </summary>
[Fact]
public async Task Consumer_clustered_delete_removes_consumer()
{
var metaGroup = new JetStreamMetaGroup(nodes: 1);
// Set up stream and consumer.
await StreamApiHandlers.HandleClusteredCreateAsync(
"$JS.API.STREAM.CREATE.ORDERS",
Encoding.UTF8.GetBytes("""{"name":"ORDERS","subjects":["orders.>"]}"""),
metaGroup,
CancellationToken.None);
await ConsumerApiHandlers.HandleClusteredCreateAsync(
"$JS.API.CONSUMER.CREATE.ORDERS.MON",
Encoding.UTF8.GetBytes("""{"durable_name":"MON"}"""),
metaGroup,
CancellationToken.None);
metaGroup.GetConsumerAssignment("ORDERS", "MON").ShouldNotBeNull();
var response = await ConsumerApiHandlers.HandleClusteredDeleteAsync(
"$JS.API.CONSUMER.DELETE.ORDERS.MON", metaGroup, CancellationToken.None);
response.Error.ShouldBeNull();
response.Success.ShouldBeTrue();
metaGroup.GetConsumerAssignment("ORDERS", "MON").ShouldBeNull();
}
/// <summary>
/// Deleting a non-existent consumer returns a 404 not-found error.
/// Go reference: jetstream_cluster.go — consumer delete validates existence.
/// </summary>
[Fact]
public async Task Consumer_clustered_delete_returns_not_found_for_missing()
{
var metaGroup = new JetStreamMetaGroup(nodes: 1);
// Create the stream but not the consumer.
await StreamApiHandlers.HandleClusteredCreateAsync(
"$JS.API.STREAM.CREATE.ORDERS",
Encoding.UTF8.GetBytes("""{"name":"ORDERS","subjects":["orders.>"]}"""),
metaGroup,
CancellationToken.None);
var response = await ConsumerApiHandlers.HandleClusteredDeleteAsync(
"$JS.API.CONSUMER.DELETE.ORDERS.GHOST", metaGroup, CancellationToken.None);
response.Error.ShouldNotBeNull();
response.Error!.Code.ShouldBe(404);
}
}

303
tests/NATS.Server.JetStream.Tests/JetStream/Api/ClusteredRequestTests.cs Normal file
View File

@@ -0,0 +1,303 @@
// Go reference: jetstream_cluster.go:7620-7701 — jsClusteredStreamRequest lifecycle:
// propose to meta RAFT → wait for result → deliver or time out.
// ClusteredRequestProcessor tracks pending requests and delivers results when RAFT entries
// are applied, matching the Go server's callback-based completion mechanism.
using NATS.Server.JetStream.Api;
namespace NATS.Server.JetStream.Tests.JetStream.Api;
public class ClusteredRequestTests
{
// ---------------------------------------------------------------
// RegisterPending
// ---------------------------------------------------------------
/// <summary>
/// Each call to RegisterPending returns a distinct, non-empty string identifier.
/// Go reference: jetstream_cluster.go:7620 — each clustered request gets a unique ID
/// used to correlate the RAFT apply callback with the waiting caller.
/// </summary>
[Fact]
public void RegisterPending_returns_unique_id()
{
var processor = new ClusteredRequestProcessor();
var id1 = processor.RegisterPending();
var id2 = processor.RegisterPending();
id1.ShouldNotBeNullOrWhiteSpace();
id2.ShouldNotBeNullOrWhiteSpace();
id1.ShouldNotBe(id2);
}
// ---------------------------------------------------------------
// WaitForResult
// ---------------------------------------------------------------
/// <summary>
/// When a result is delivered for a pending request, WaitForResultAsync returns that response.
/// Go reference: jetstream_cluster.go:7620 — the waiting goroutine receives the result
/// via channel once the RAFT leader applies the entry.
/// </summary>
[Fact]
public async Task WaitForResult_returns_delivered_response()
{
var processor = new ClusteredRequestProcessor(timeout: TimeSpan.FromSeconds(5));
var requestId = processor.RegisterPending();
var expected = JetStreamApiResponse.SuccessResponse();
// Use a semaphore so the wait starts before delivery occurs — no timing dependency.
var waitStarted = new SemaphoreSlim(0, 1);
var deliverTask = Task.Run(async () =>
{
// Wait until WaitForResultAsync has been entered before delivering.
await waitStarted.WaitAsync();
processor.DeliverResult(requestId, expected);
});
// Signal the deliver task once we begin waiting.
waitStarted.Release();
var result = await processor.WaitForResultAsync(requestId);
await deliverTask;
result.ShouldBeSameAs(expected);
}
/// <summary>
/// When no result is delivered within the timeout, WaitForResultAsync returns a 408 error.
/// Go reference: jetstream_cluster.go:7620 — if the RAFT group does not respond in time,
/// the request is considered timed out and an error is returned to the client.
/// </summary>
[Fact]
public async Task WaitForResult_times_out_after_timeout()
{
var processor = new ClusteredRequestProcessor(timeout: TimeSpan.FromMilliseconds(50));
var requestId = processor.RegisterPending();
var result = await processor.WaitForResultAsync(requestId);
result.Error.ShouldNotBeNull();
result.Error!.Code.ShouldBe(408);
result.Error.Description.ShouldContain("timeout");
}
/// <summary>
/// WaitForResultAsync returns a 500 error for an ID that was never registered.
/// Go reference: jetstream_cluster.go — requesting a result for an unknown request ID
/// is a programming error; return an internal server error.
/// </summary>
[Fact]
public async Task WaitForResult_returns_error_for_unknown_id()
{
var processor = new ClusteredRequestProcessor();
var result = await processor.WaitForResultAsync("nonexistent-id");
result.Error.ShouldNotBeNull();
result.Error!.Code.ShouldBe(500);
result.Error.Description.ShouldContain("not found");
}
/// <summary>
/// When the caller's CancellationToken is triggered, WaitForResultAsync returns a timeout error.
/// Go reference: jetstream_cluster.go:7620 — callers can cancel waiting for a RAFT result
/// if their own request context is cancelled.
/// </summary>
[Fact]
public async Task WaitForResult_respects_cancellation_token()
{
var processor = new ClusteredRequestProcessor(timeout: TimeSpan.FromSeconds(30));
var requestId = processor.RegisterPending();
using var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(50));
var result = await processor.WaitForResultAsync(requestId, cts.Token);
result.Error.ShouldNotBeNull();
result.Error!.Code.ShouldBe(408);
}
// ---------------------------------------------------------------
// DeliverResult
// ---------------------------------------------------------------
/// <summary>
/// DeliverResult returns true when the request ID is known and pending.
/// Go reference: jetstream_cluster.go:7620 — the RAFT apply callback signals success
/// by resolving the pending request.
/// </summary>
[Fact]
public void DeliverResult_returns_true_for_pending_request()
{
var processor = new ClusteredRequestProcessor();
var requestId = processor.RegisterPending();
var delivered = processor.DeliverResult(requestId, JetStreamApiResponse.SuccessResponse());
delivered.ShouldBeTrue();
}
/// <summary>
/// DeliverResult returns false when the request ID is not found.
/// Go reference: jetstream_cluster.go — delivering a result for an unknown or already-completed
/// request is a no-op; return false so the caller knows the result was not consumed.
/// </summary>
[Fact]
public void DeliverResult_returns_false_for_unknown_request()
{
var processor = new ClusteredRequestProcessor();
var delivered = processor.DeliverResult("unknown-id", JetStreamApiResponse.SuccessResponse());
delivered.ShouldBeFalse();
}
// ---------------------------------------------------------------
// PendingCount
// ---------------------------------------------------------------
/// <summary>
/// PendingCount increases with each RegisterPending call and decreases when a result is
/// delivered or the request times out.
/// Go reference: jetstream_cluster.go — the server tracks pending RAFT proposals for
/// observability and to detect stuck requests.
/// </summary>
[Fact]
public async Task PendingCount_tracks_active_requests()
{
var processor = new ClusteredRequestProcessor(timeout: TimeSpan.FromMilliseconds(50));
processor.PendingCount.ShouldBe(0);
var id1 = processor.RegisterPending();
processor.PendingCount.ShouldBe(1);
var id2 = processor.RegisterPending();
processor.PendingCount.ShouldBe(2);
// Deliver one request.
processor.DeliverResult(id1, JetStreamApiResponse.SuccessResponse());
processor.PendingCount.ShouldBe(1);
// Let id2 time out.
await processor.WaitForResultAsync(id2);
processor.PendingCount.ShouldBe(0);
}
// ---------------------------------------------------------------
// CancelAll
// ---------------------------------------------------------------
/// <summary>
/// CancelAll completes all pending requests with a 503 error response.
/// Go reference: jetstream_cluster.go — when this node loses RAFT leadership, all
/// in-flight proposals must be failed so callers do not hang indefinitely.
/// </summary>
[Fact]
public async Task CancelAll_completes_all_pending_with_error()
{
var processor = new ClusteredRequestProcessor(timeout: TimeSpan.FromSeconds(30));
var id1 = processor.RegisterPending();
var id2 = processor.RegisterPending();
var task1 = processor.WaitForResultAsync(id1);
var task2 = processor.WaitForResultAsync(id2);
processor.CancelAll("leadership changed");
var result1 = await task1;
var result2 = await task2;
result1.Error.ShouldNotBeNull();
result1.Error!.Code.ShouldBe(503);
result1.Error.Description.ShouldContain("leadership changed");
result2.Error.ShouldNotBeNull();
result2.Error!.Code.ShouldBe(503);
result2.Error.Description.ShouldContain("leadership changed");
}
/// <summary>
/// After CancelAll, PendingCount drops to zero.
/// Go reference: jetstream_cluster.go — a leadership change clears all pending state.
/// </summary>
[Fact]
public void CancelAll_clears_pending_count()
{
var processor = new ClusteredRequestProcessor(timeout: TimeSpan.FromSeconds(30));
processor.RegisterPending();
processor.RegisterPending();
processor.RegisterPending();
processor.PendingCount.ShouldBe(3);
processor.CancelAll();
processor.PendingCount.ShouldBe(0);
}
/// <summary>
/// CancelAll uses a default reason of "leadership changed" when no reason is provided.
/// Go reference: jetstream_cluster.go — default cancellation reason matches NATS cluster semantics.
/// </summary>
[Fact]
public async Task CancelAll_uses_default_reason()
{
var processor = new ClusteredRequestProcessor(timeout: TimeSpan.FromSeconds(30));
var id = processor.RegisterPending();
var task = processor.WaitForResultAsync(id);
processor.CancelAll(); // no reason argument
var result = await task;
result.Error.ShouldNotBeNull();
result.Error!.Description.ShouldContain("leadership changed");
}
// ---------------------------------------------------------------
// Concurrency
// ---------------------------------------------------------------
/// <summary>
/// Concurrent registrations and deliveries all receive the correct response.
/// Go reference: jetstream_cluster.go — in a cluster, many API requests may be in-flight
/// simultaneously, each waiting for its own RAFT entry to be applied.
/// </summary>
[Fact]
public async Task Concurrent_register_and_deliver()
{
const int count = 50;
var processor = new ClusteredRequestProcessor(timeout: TimeSpan.FromSeconds(10));
var requestIds = new string[count];
for (var i = 0; i < count; i++)
requestIds[i] = processor.RegisterPending();
// Start all waits concurrently.
var waitTasks = requestIds.Select(id => processor.WaitForResultAsync(id)).ToArray();
// Deliver all results concurrently — no delay needed; the ThreadPool provides
// sufficient interleaving to exercise concurrent access patterns.
var deliverTasks = requestIds.Select((id, i) => Task.Run(() =>
{
processor.DeliverResult(id, JetStreamApiResponse.ErrorResponse(200 + i, $"response-{i}"));
})).ToArray();
await Task.WhenAll(deliverTasks);
var results = await Task.WhenAll(waitTasks);
// Every result should be a valid response (no null errors from "not found").
results.Length.ShouldBe(count);
foreach (var result in results)
{
// Each result was an explicitly delivered response with a known code.
result.Error.ShouldNotBeNull();
result.Error!.Code.ShouldBeGreaterThanOrEqualTo(200);
result.Error.Code.ShouldBeLessThan(200 + count);
}
}
}

182
tests/NATS.Server.JetStream.Tests/JetStream/Api/ConsumerPauseApiTests.cs Normal file
View File

@@ -0,0 +1,182 @@
// Go reference: server/consumer.go — pauseConsumer / resumeConsumer / isPaused
// Tests for the consumer pause/resume API endpoint, including pause_until (RFC3339)
// time-bounded pauses and response body containing pause state.
using NATS.Server.JetStream.Api;
using NATS.Server.TestUtilities;
namespace NATS.Server.JetStream.Tests.JetStream.Api;
public class ConsumerPauseApiTests : IAsyncLifetime
{
private JetStreamApiFixture _fx = null!;
public async Task InitializeAsync()
{
_fx = await JetStreamApiFixture.StartWithStreamAsync("ORDERS", "orders.*");
_ = await _fx.CreateConsumerAsync("ORDERS", "MON", "orders.created");
}
public async Task DisposeAsync() => await _fx.DisposeAsync();
// Go ref: consumer.go pauseConsumer — pause=true pauses consumer.
[Fact]
public async Task HandlePause_with_pause_true_pauses_consumer()
{
var resp = await _fx.RequestLocalAsync(
"$JS.API.CONSUMER.PAUSE.ORDERS.MON",
"{\"pause\":true}");
resp.Error.ShouldBeNull();
resp.Success.ShouldBeTrue();
resp.Paused.ShouldBe(true);
}
// Go ref: consumer.go resumeConsumer — pause=false resumes consumer.
[Fact]
public async Task HandlePause_with_pause_false_resumes_consumer()
{
// First pause
await _fx.RequestLocalAsync("$JS.API.CONSUMER.PAUSE.ORDERS.MON", "{\"pause\":true}");
// Then resume
var resp = await _fx.RequestLocalAsync(
"$JS.API.CONSUMER.PAUSE.ORDERS.MON",
"{\"pause\":false}");
resp.Error.ShouldBeNull();
resp.Success.ShouldBeTrue();
resp.Paused.ShouldBe(false);
}
// Go ref: consumer.go pauseConsumer — pause_until sets deadline UTC datetime.
[Fact]
public async Task HandlePause_with_pause_until_sets_deadline()
{
var future = DateTime.UtcNow.AddHours(1);
var iso = future.ToString("O"); // RFC3339 round-trip format
var resp = await _fx.RequestLocalAsync(
"$JS.API.CONSUMER.PAUSE.ORDERS.MON",
$"{{\"pause_until\":\"{iso}\"}}");
resp.Error.ShouldBeNull();
resp.PauseUntil.ShouldNotBeNull();
resp.PauseUntil!.Value.Should_Be_Close_To_Utc(future, tolerance: TimeSpan.FromSeconds(2));
}
// Go ref: consumer.go pauseConsumer — pause_until implies pause=true.
[Fact]
public async Task HandlePause_with_pause_until_implies_pause_true()
{
var future = DateTime.UtcNow.AddHours(1);
var iso = future.ToString("O");
var resp = await _fx.RequestLocalAsync(
"$JS.API.CONSUMER.PAUSE.ORDERS.MON",
$"{{\"pause_until\":\"{iso}\"}}");
resp.Error.ShouldBeNull();
resp.Paused.ShouldBe(true);
}
// Go ref: consumer.go isPaused — response includes current pause state.
[Fact]
public async Task HandlePause_returns_pause_state_in_response()
{
var resp = await _fx.RequestLocalAsync(
"$JS.API.CONSUMER.PAUSE.ORDERS.MON",
"{\"pause\":true}");
resp.Paused.ShouldBe(true);
var resumeResp = await _fx.RequestLocalAsync(
"$JS.API.CONSUMER.PAUSE.ORDERS.MON",
"{\"pause\":false}");
resumeResp.Paused.ShouldBe(false);
}
// Go ref: consumer.go pauseUntil — response includes pause_until when set.
[Fact]
public async Task HandlePause_returns_pause_until_in_response()
{
var future = DateTime.UtcNow.AddMinutes(30);
var iso = future.ToString("O");
var resp = await _fx.RequestLocalAsync(
"$JS.API.CONSUMER.PAUSE.ORDERS.MON",
$"{{\"pause_until\":\"{iso}\"}}");
resp.PauseUntil.ShouldNotBeNull();
resp.PauseUntil!.Value.Kind.ShouldBe(DateTimeKind.Utc);
}
// Go ref: consumer.go pauseConsumer — 404 when consumer not found.
[Fact]
public async Task HandlePause_returns_not_found_for_missing_consumer()
{
var resp = await _fx.RequestLocalAsync(
"$JS.API.CONSUMER.PAUSE.ORDERS.NONEXISTENT",
"{\"pause\":true}");
resp.Error.ShouldNotBeNull();
resp.Error!.Code.ShouldBe(404);
}
// Go ref: consumer.go resumeConsumer — empty payload resumes consumer.
[Fact]
public async Task HandlePause_with_empty_payload_resumes()
{
// Pause first
await _fx.RequestLocalAsync("$JS.API.CONSUMER.PAUSE.ORDERS.MON", "{\"pause\":true}");
// Empty body = resume
var resp = await _fx.RequestLocalAsync("$JS.API.CONSUMER.PAUSE.ORDERS.MON", "");
resp.Error.ShouldBeNull();
resp.Success.ShouldBeTrue();
resp.Paused.ShouldBe(false);
}
// Go ref: consumer.go pauseConsumer — past pause_until auto-resumes immediately.
[Fact]
public async Task HandlePause_with_past_pause_until_auto_resumes()
{
var past = DateTime.UtcNow.AddHours(-1);
var iso = past.ToString("O");
var resp = await _fx.RequestLocalAsync(
"$JS.API.CONSUMER.PAUSE.ORDERS.MON",
$"{{\"pause_until\":\"{iso}\"}}");
// Deadline already passed — consumer should auto-resume, so paused=false.
resp.Error.ShouldBeNull();
resp.Success.ShouldBeTrue();
resp.Paused.ShouldBe(false);
}
// Go ref: jsConsumerPauseT — bad subject (not matching stream.consumer pattern) returns 404.
[Fact]
public async Task HandlePause_returns_not_found_for_bad_subject()
{
var resp = await _fx.RequestLocalAsync(
"$JS.API.CONSUMER.PAUSE.ONLY_ONE_TOKEN",
"{\"pause\":true}");
resp.Error.ShouldNotBeNull();
resp.Error!.Code.ShouldBe(404);
}
}
/// <summary>
/// Shouldly-compatible extension for DateTime proximity assertions.
/// </summary>
internal static class DateTimeAssertExtensions
{
public static void Should_Be_Close_To_Utc(this DateTime actual, DateTime expected, TimeSpan tolerance)
{
var diff = (actual.ToUniversalTime() - expected.ToUniversalTime()).Duration();
diff.ShouldBeLessThanOrEqualTo(tolerance);
}
}

108
tests/NATS.Server.JetStream.Tests/JetStream/Api/JetStreamApiLimitsParityBatch1Tests.cs Normal file
View File

@@ -0,0 +1,108 @@
using NATS.Server.JetStream.Api;
using NATS.Server.JetStream.Models;
using NATS.Server.JetStream.Validation;
using NATS.Server.JetStream;
namespace NATS.Server.JetStream.Tests.JetStream.Api;
public class JetStreamApiLimitsParityBatch1Tests
{
[Fact]
public void Constants_match_go_reference_values()
{
JetStreamApiLimits.JSMaxDescriptionLen.ShouldBe(4_096);
JetStreamApiLimits.JSMaxMetadataLen.ShouldBe(128 * 1024);
JetStreamApiLimits.JSMaxNameLen.ShouldBe(255);
JetStreamApiLimits.JSDefaultRequestQueueLimit.ShouldBe(10_000);
}
[Theory]
[InlineData(null, false)]
[InlineData("", false)]
[InlineData(" ", false)]
[InlineData("ORDERS", true)]
[InlineData("ORD ERS", false)]
[InlineData("ORDERS.*", false)]
[InlineData("ORDERS.>", false)]
public void IsValidName_enforces_expected_rules(string? name, bool expected)
{
JetStreamConfigValidator.IsValidName(name).ShouldBe(expected);
}
[Fact]
public void Stream_create_rejects_name_over_max_length()
{
var manager = new StreamManager();
var response = manager.CreateOrUpdate(new StreamConfig
{
Name = new string('S', JetStreamApiLimits.JSMaxNameLen + 1),
Subjects = ["a"],
});
response.Error.ShouldNotBeNull();
response.Error!.Description.ShouldBe("invalid stream name");
}
[Fact]
public void Stream_create_rejects_description_over_max_bytes()
{
var manager = new StreamManager();
var response = manager.CreateOrUpdate(new StreamConfig
{
Name = "LIMITDESC",
Subjects = ["a"],
Description = new string('d', JetStreamApiLimits.JSMaxDescriptionLen + 1),
});
response.Error.ShouldNotBeNull();
response.Error!.Description.ShouldBe("stream description is too long");
}
[Fact]
public void Stream_create_rejects_metadata_over_max_bytes()
{
var manager = new StreamManager();
var response = manager.CreateOrUpdate(new StreamConfig
{
Name = "LIMITMETA",
Subjects = ["a"],
Metadata = new Dictionary<string, string>
{
["k"] = new string('m', JetStreamApiLimits.JSMaxMetadataLen),
},
});
response.Error.ShouldNotBeNull();
response.Error!.Description.ShouldBe("stream metadata exceeds maximum size");
}
[Fact]
public void Consumer_create_rejects_durable_name_over_max_length()
{
var manager = new ConsumerManager();
var response = manager.CreateOrUpdate("S", new ConsumerConfig
{
DurableName = new string('C', JetStreamApiLimits.JSMaxNameLen + 1),
});
response.Error.ShouldNotBeNull();
response.Error!.Description.ShouldBe("invalid durable name");
}
[Fact]
public void Consumer_create_rejects_metadata_over_max_bytes()
{
var manager = new ConsumerManager();
var response = manager.CreateOrUpdate("S", new ConsumerConfig
{
DurableName = "C1",
Metadata = new Dictionary<string, string>
{
["k"] = new string('m', JetStreamApiLimits.JSMaxMetadataLen),
},
});
response.Error.ShouldNotBeNull();
response.Error!.Description.ShouldBe("consumer metadata exceeds maximum size");
}
}

412
tests/NATS.Server.JetStream.Tests/JetStream/Api/LeaderForwardingTests.cs Normal file
View File

@@ -0,0 +1,412 @@
// Go reference: jetstream_api.go:200-300 — API requests at non-leader nodes must be
// forwarded to the current leader. Mutating operations return a not-leader error with
// a leader_hint field; read-only operations are handled locally on any node.
using System.Text;
using NATS.Server.JetStream;
using NATS.Server.JetStream.Api;
using NATS.Server.JetStream.Cluster;
namespace NATS.Server.JetStream.Tests.JetStream.Api;
/// <summary>
/// Simple test double for ILeaderForwarder.
/// Returns a predetermined response or null depending on the constructor.
/// </summary>
file sealed class StubForwarder(JetStreamApiResponse? response) : ILeaderForwarder
{
public int CallCount { get; private set; }
public string? LastSubject { get; private set; }
public ReadOnlyMemory<byte> LastPayload { get; private set; }
public string? LastLeaderName { get; private set; }
public Task<JetStreamApiResponse?> ForwardAsync(
string subject, ReadOnlyMemory<byte> payload, string leaderName, CancellationToken ct)
{
CallCount++;
LastSubject = subject;
LastPayload = payload;
LastLeaderName = leaderName;
return Task.FromResult(response);
}
}
/// <summary>
/// Test double that throws OperationCanceledException to simulate a timeout.
/// </summary>
file sealed class TimeoutForwarder : ILeaderForwarder
{
public Task<JetStreamApiResponse?> ForwardAsync(
string subject, ReadOnlyMemory<byte> payload, string leaderName, CancellationToken ct)
=> Task.FromException<JetStreamApiResponse?>(new OperationCanceledException("simulated timeout"));
}
public class LeaderForwardingTests
{
/// <summary>
/// When this node IS the leader, mutating requests are handled locally.
/// Go reference: jetstream_api.go — leader handles requests directly.
/// </summary>
[Fact]
public void Route_WhenLeader_HandlesLocally()
{
// selfIndex=1 matches default leaderIndex=1, so this node is the leader.
var metaGroup = new JetStreamMetaGroup(nodes: 3, selfIndex: 1);
var streamManager = new StreamManager(metaGroup);
var consumerManager = new ConsumerManager();
var router = new JetStreamApiRouter(streamManager, consumerManager, metaGroup);
// Create a stream first so the purge has something to operate on.
var createPayload = Encoding.UTF8.GetBytes("""{"name":"TEST","subjects":["test.>"]}""");
var createResult = router.Route("$JS.API.STREAM.CREATE.TEST", createPayload);
createResult.Error.ShouldBeNull();
createResult.StreamInfo.ShouldNotBeNull();
// A mutating operation (delete) should succeed locally.
var deleteResult = router.Route("$JS.API.STREAM.DELETE.TEST", ReadOnlySpan<byte>.Empty);
deleteResult.Error.ShouldBeNull();
deleteResult.Success.ShouldBeTrue();
}
/// <summary>
/// When this node is NOT the leader, mutating operations return a not-leader error
/// with the current leader's identifier in the leader_hint field.
/// Go reference: jetstream_api.go:200-300 — not-leader response.
/// </summary>
[Fact]
public void Route_WhenNotLeader_MutatingOp_ReturnsNotLeaderError()
{
// selfIndex=2, leaderIndex defaults to 1 — this node is NOT the leader.
var metaGroup = new JetStreamMetaGroup(nodes: 3, selfIndex: 2);
var streamManager = new StreamManager(metaGroup);
var consumerManager = new ConsumerManager();
var router = new JetStreamApiRouter(streamManager, consumerManager, metaGroup);
var payload = Encoding.UTF8.GetBytes("""{"name":"TEST","subjects":["test.>"]}""");
var result = router.Route("$JS.API.STREAM.CREATE.TEST", payload);
result.Error.ShouldNotBeNull();
result.Error!.Code.ShouldBe(10003);
result.Error.Description.ShouldBe("not leader");
result.Error.LeaderHint.ShouldNotBeNull();
result.Error.LeaderHint.ShouldBe("meta-1");
}
/// <summary>
/// Read-only operations (INFO, NAMES, LIST) are handled locally even when
/// this node is not the leader.
/// Go reference: jetstream_api.go — read operations do not require leadership.
/// </summary>
[Fact]
public void Route_WhenNotLeader_ReadOp_HandlesLocally()
{
// selfIndex=2, leaderIndex defaults to 1 — this node is NOT the leader.
var metaGroup = new JetStreamMetaGroup(nodes: 3, selfIndex: 2);
var streamManager = new StreamManager(metaGroup);
var consumerManager = new ConsumerManager();
var router = new JetStreamApiRouter(streamManager, consumerManager, metaGroup);
// $JS.API.INFO is a read-only operation.
var infoResult = router.Route("$JS.API.INFO", ReadOnlySpan<byte>.Empty);
infoResult.Error.ShouldBeNull();
// $JS.API.STREAM.NAMES is a read-only operation.
var namesResult = router.Route("$JS.API.STREAM.NAMES", ReadOnlySpan<byte>.Empty);
namesResult.Error.ShouldBeNull();
namesResult.StreamNames.ShouldNotBeNull();
// $JS.API.STREAM.LIST is a read-only operation.
var listResult = router.Route("$JS.API.STREAM.LIST", ReadOnlySpan<byte>.Empty);
listResult.Error.ShouldBeNull();
listResult.StreamNames.ShouldNotBeNull();
}
/// <summary>
/// When there is no meta-group (single-server mode), all operations are handled
/// locally regardless of the subject type.
/// Go reference: jetstream_api.go — standalone servers have no meta-group.
/// </summary>
[Fact]
public void Route_NoMetaGroup_HandlesLocally()
{
// No meta-group — single server mode.
var streamManager = new StreamManager();
var consumerManager = new ConsumerManager();
var router = new JetStreamApiRouter(streamManager, consumerManager, metaGroup: null);
var payload = Encoding.UTF8.GetBytes("""{"name":"TEST","subjects":["test.>"]}""");
var result = router.Route("$JS.API.STREAM.CREATE.TEST", payload);
// Should succeed — no leader check in single-server mode.
result.Error.ShouldBeNull();
result.StreamInfo.ShouldNotBeNull();
result.StreamInfo!.Config.Name.ShouldBe("TEST");
}
/// <summary>
/// IsLeaderRequired returns true for Create, Update, Delete, and Purge operations.
/// Go reference: jetstream_api.go:200-300 — mutating operations require leader.
/// </summary>
[Fact]
public void IsLeaderRequired_CreateUpdate_ReturnsTrue()
{
JetStreamApiRouter.IsLeaderRequired("$JS.API.STREAM.CREATE.TEST").ShouldBeTrue();
JetStreamApiRouter.IsLeaderRequired("$JS.API.STREAM.UPDATE.TEST").ShouldBeTrue();
JetStreamApiRouter.IsLeaderRequired("$JS.API.STREAM.DELETE.TEST").ShouldBeTrue();
JetStreamApiRouter.IsLeaderRequired("$JS.API.STREAM.PURGE.TEST").ShouldBeTrue();
JetStreamApiRouter.IsLeaderRequired("$JS.API.STREAM.RESTORE.TEST").ShouldBeTrue();
JetStreamApiRouter.IsLeaderRequired("$JS.API.STREAM.MSG.DELETE.TEST").ShouldBeTrue();
JetStreamApiRouter.IsLeaderRequired("$JS.API.CONSUMER.CREATE.STREAM.CON").ShouldBeTrue();
JetStreamApiRouter.IsLeaderRequired("$JS.API.CONSUMER.DELETE.STREAM.CON").ShouldBeTrue();
}
/// <summary>
/// IsLeaderRequired returns false for Info, Names, List, and other read operations.
/// Go reference: jetstream_api.go — read-only operations do not need leadership.
/// </summary>
[Fact]
public void IsLeaderRequired_InfoList_ReturnsFalse()
{
JetStreamApiRouter.IsLeaderRequired("$JS.API.INFO").ShouldBeFalse();
JetStreamApiRouter.IsLeaderRequired("$JS.API.STREAM.INFO.TEST").ShouldBeFalse();
JetStreamApiRouter.IsLeaderRequired("$JS.API.STREAM.NAMES").ShouldBeFalse();
JetStreamApiRouter.IsLeaderRequired("$JS.API.STREAM.LIST").ShouldBeFalse();
JetStreamApiRouter.IsLeaderRequired("$JS.API.STREAM.MSG.GET.TEST").ShouldBeFalse();
JetStreamApiRouter.IsLeaderRequired("$JS.API.STREAM.SNAPSHOT.TEST").ShouldBeFalse();
JetStreamApiRouter.IsLeaderRequired("$JS.API.CONSUMER.INFO.STREAM.CON").ShouldBeFalse();
JetStreamApiRouter.IsLeaderRequired("$JS.API.CONSUMER.NAMES.STREAM").ShouldBeFalse();
JetStreamApiRouter.IsLeaderRequired("$JS.API.CONSUMER.LIST.STREAM").ShouldBeFalse();
JetStreamApiRouter.IsLeaderRequired("$JS.API.CONSUMER.MSG.NEXT.STREAM.CON").ShouldBeFalse();
JetStreamApiRouter.IsLeaderRequired("$JS.API.DIRECT.GET.TEST").ShouldBeFalse();
}
// ---------------------------------------------------------------
// New tests for Task 19: Leader Forwarding (Gap 7.1)
// Go reference: jetstream_api.go:200-300 — jsClusteredStreamXxxRequest helpers.
// ---------------------------------------------------------------
/// <summary>
/// When not leader and a forwarder is provided, RouteAsync calls forward for mutating ops.
/// Go reference: jetstream_api.go — non-leader nodes forward mutating ops to the leader.
/// </summary>
[Fact]
public async Task Route_forwards_mutating_request_when_not_leader()
{
var metaGroup = new JetStreamMetaGroup(nodes: 3, selfIndex: 2);
var streamManager = new StreamManager(metaGroup);
var consumerManager = new ConsumerManager();
var forwarded = JetStreamApiResponse.SuccessResponse();
var forwarder = new StubForwarder(forwarded);
var router = new JetStreamApiRouter(streamManager, consumerManager, metaGroup, forwarder);
var payload = Encoding.UTF8.GetBytes("""{"name":"FWD","subjects":["fwd.>"]}""");
var result = await router.RouteAsync("$JS.API.STREAM.CREATE.FWD", payload.AsMemory());
forwarder.CallCount.ShouldBe(1);
forwarder.LastSubject.ShouldBe("$JS.API.STREAM.CREATE.FWD");
forwarder.LastLeaderName.ShouldBe("meta-1");
result.ShouldBeSameAs(forwarded);
}
/// <summary>
/// When not leader and no forwarder is provided, RouteAsync returns a NotLeader error.
/// Go reference: jetstream_api.go — fallback to not-leader error when no forwarder.
/// </summary>
[Fact]
public async Task Route_returns_not_leader_when_no_forwarder()
{
var metaGroup = new JetStreamMetaGroup(nodes: 3, selfIndex: 2);
var streamManager = new StreamManager(metaGroup);
var consumerManager = new ConsumerManager();
var router = new JetStreamApiRouter(streamManager, consumerManager, metaGroup, forwarder: null);
var payload = Encoding.UTF8.GetBytes("""{"name":"TEST","subjects":["test.>"]}""");
var result = await router.RouteAsync("$JS.API.STREAM.CREATE.TEST", payload.AsMemory());
result.Error.ShouldNotBeNull();
result.Error!.Code.ShouldBe(10003);
result.Error.Description.ShouldBe("not leader");
result.Error.LeaderHint.ShouldBe("meta-1");
}
/// <summary>
/// Read-only operations are handled locally even when not leader and a forwarder is set.
/// Go reference: jetstream_api.go — read ops do not require leadership, never forwarded.
/// </summary>
[Fact]
public async Task Route_does_not_forward_read_only_requests()
{
var metaGroup = new JetStreamMetaGroup(nodes: 3, selfIndex: 2);
var streamManager = new StreamManager(metaGroup);
var consumerManager = new ConsumerManager();
var forwarder = new StubForwarder(JetStreamApiResponse.SuccessResponse());
var router = new JetStreamApiRouter(streamManager, consumerManager, metaGroup, forwarder);
// $JS.API.INFO — read only
var infoResult = await router.RouteAsync("$JS.API.INFO", ReadOnlyMemory<byte>.Empty);
infoResult.Error.ShouldBeNull();
// $JS.API.STREAM.NAMES — read only
var namesResult = await router.RouteAsync("$JS.API.STREAM.NAMES", ReadOnlyMemory<byte>.Empty);
namesResult.Error.ShouldBeNull();
// Forwarder should never have been called for read-only subjects.
forwarder.CallCount.ShouldBe(0);
}
/// <summary>
/// When the forwarder returns null, RouteAsync falls back to a NotLeader response.
/// Go reference: jetstream_api.go — null forward result means forwarding unavailable.
/// </summary>
[Fact]
public async Task Route_handles_forward_returning_null_gracefully()
{
var metaGroup = new JetStreamMetaGroup(nodes: 3, selfIndex: 2);
var streamManager = new StreamManager(metaGroup);
var consumerManager = new ConsumerManager();
var forwarder = new StubForwarder(null); // returns null
var router = new JetStreamApiRouter(streamManager, consumerManager, metaGroup, forwarder);
var payload = Encoding.UTF8.GetBytes("""{"name":"TEST","subjects":["test.>"]}""");
var result = await router.RouteAsync("$JS.API.STREAM.CREATE.TEST", payload.AsMemory());
forwarder.CallCount.ShouldBe(1);
result.Error.ShouldNotBeNull();
result.Error!.Code.ShouldBe(10003);
result.Error.Description.ShouldBe("not leader");
}
/// <summary>
/// When the forwarder throws OperationCanceledException (timeout), RouteAsync falls back to NotLeader.
/// Go reference: jetstream_api.go — timeout/cancellation during forwarding falls back gracefully.
/// </summary>
[Fact]
public async Task Route_handles_forward_timeout()
{
var metaGroup = new JetStreamMetaGroup(nodes: 3, selfIndex: 2);
var streamManager = new StreamManager(metaGroup);
var consumerManager = new ConsumerManager();
var forwarder = new TimeoutForwarder();
var router = new JetStreamApiRouter(streamManager, consumerManager, metaGroup, forwarder);
var payload = Encoding.UTF8.GetBytes("""{"name":"TEST","subjects":["test.>"]}""");
var result = await router.RouteAsync("$JS.API.STREAM.CREATE.TEST", payload.AsMemory());
result.Error.ShouldNotBeNull();
result.Error!.Code.ShouldBe(10003);
result.Error.Description.ShouldBe("not leader");
}
/// <summary>
/// ForwardedCount increments on each successful (non-null) forward result.
/// Go reference: jetstream_api.go — monitoring/observability for forwarded requests.
/// </summary>
[Fact]
public async Task ForwardedCount_increments_on_successful_forward()
{
var metaGroup = new JetStreamMetaGroup(nodes: 3, selfIndex: 2);
var streamManager = new StreamManager(metaGroup);
var consumerManager = new ConsumerManager();
var forwarder = new StubForwarder(JetStreamApiResponse.SuccessResponse());
var router = new JetStreamApiRouter(streamManager, consumerManager, metaGroup, forwarder);
router.ForwardedCount.ShouldBe(0);
var payload = Encoding.UTF8.GetBytes("""{"name":"A","subjects":["a.>"]}""");
await router.RouteAsync("$JS.API.STREAM.CREATE.A", payload.AsMemory());
router.ForwardedCount.ShouldBe(1);
await router.RouteAsync("$JS.API.STREAM.DELETE.A", ReadOnlyMemory<byte>.Empty);
router.ForwardedCount.ShouldBe(2);
}
/// <summary>
/// When this node is the leader, RouteAsync handles requests locally and does not call the forwarder.
/// Go reference: jetstream_api.go — leader handles requests directly.
/// </summary>
[Fact]
public async Task Route_processes_locally_when_leader()
{
var metaGroup = new JetStreamMetaGroup(nodes: 3, selfIndex: 1); // IS leader
var streamManager = new StreamManager(metaGroup);
var consumerManager = new ConsumerManager();
var forwarder = new StubForwarder(JetStreamApiResponse.SuccessResponse());
var router = new JetStreamApiRouter(streamManager, consumerManager, metaGroup, forwarder);
var payload = Encoding.UTF8.GetBytes("""{"name":"LOCAL","subjects":["local.>"]}""");
var result = await router.RouteAsync("$JS.API.STREAM.CREATE.LOCAL", payload.AsMemory());
forwarder.CallCount.ShouldBe(0);
result.Error.ShouldBeNull();
result.StreamInfo.ShouldNotBeNull();
result.StreamInfo!.Config.Name.ShouldBe("LOCAL");
}
/// <summary>
/// When no meta-group is configured (single-server), RouteAsync handles all requests locally.
/// Go reference: jetstream_api.go — standalone servers have no meta-group.
/// </summary>
[Fact]
public async Task Route_processes_locally_when_no_meta_group()
{
var streamManager = new StreamManager();
var consumerManager = new ConsumerManager();
var forwarder = new StubForwarder(JetStreamApiResponse.SuccessResponse());
var router = new JetStreamApiRouter(streamManager, consumerManager, metaGroup: null, forwarder);
var payload = Encoding.UTF8.GetBytes("""{"name":"SOLO","subjects":["solo.>"]}""");
var result = await router.RouteAsync("$JS.API.STREAM.CREATE.SOLO", payload.AsMemory());
forwarder.CallCount.ShouldBe(0);
result.Error.ShouldBeNull();
result.StreamInfo.ShouldNotBeNull();
result.StreamInfo!.Config.Name.ShouldBe("SOLO");
}
/// <summary>
/// RouteAsync passes the payload bytes verbatim to the forwarder.
/// Go reference: jetstream_api.go — forwarded request includes the original payload.
/// </summary>
[Fact]
public async Task RouteAsync_forwards_to_leader_with_payload()
{
var metaGroup = new JetStreamMetaGroup(nodes: 3, selfIndex: 2);
var streamManager = new StreamManager(metaGroup);
var consumerManager = new ConsumerManager();
var forwarded = JetStreamApiResponse.SuccessResponse();
var forwarder = new StubForwarder(forwarded);
var router = new JetStreamApiRouter(streamManager, consumerManager, metaGroup, forwarder);
var payloadBytes = Encoding.UTF8.GetBytes("""{"name":"PAYLOAD","subjects":["p.>"]}""");
await router.RouteAsync("$JS.API.STREAM.CREATE.PAYLOAD", payloadBytes.AsMemory());
forwarder.LastPayload.Length.ShouldBe(payloadBytes.Length);
var receivedBytes = forwarder.LastPayload.ToArray();
receivedBytes.ShouldBe(payloadBytes);
}
/// <summary>
/// DefaultLeaderForwarder accepts a custom timeout value.
/// Go reference: jetstream_api.go — configurable forward timeout for slow leader responses.
/// </summary>
[Fact]
public void Forward_timeout_configurable()
{
var customTimeout = TimeSpan.FromSeconds(10);
var forwarder = new DefaultLeaderForwarder(customTimeout);
forwarder.Timeout.ShouldBe(customTimeout);
}
/// <summary>
/// DefaultLeaderForwarder uses a 5-second default timeout when none is provided.
/// Go reference: jetstream_api.go — default forward timeout.
/// </summary>
[Fact]
public void Forward_timeout_defaults_to_five_seconds()
{
var forwarder = new DefaultLeaderForwarder();
forwarder.Timeout.ShouldBe(TimeSpan.FromSeconds(5));
}
}

317
tests/NATS.Server.JetStream.Tests/JetStream/Api/SnapshotApiTests.cs Normal file
View File

@@ -0,0 +1,317 @@
// Go reference: server/jetstream_api.go — jsStreamSnapshotT and jsStreamRestoreT handlers.
// Snapshot creates a serialized byte representation of stream state; restore re-applies it.
// The async variants (HandleSnapshotAsync / HandleRestoreAsync) add stream name and chunk
// metadata to the response and provide richer error codes compared to the sync stubs.
using System.Text;
using NATS.Server.JetStream;
using NATS.Server.JetStream.Api;
using NATS.Server.JetStream.Api.Handlers;
using NATS.Server.JetStream.Models;
namespace NATS.Server.JetStream.Tests.JetStream.Api;
public class SnapshotApiTests
{
// ---------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------
private static StreamManager CreateManagerWithStream(string streamName, string subjectPattern)
{
var sm = new StreamManager();
sm.CreateOrUpdate(new StreamConfig
{
Name = streamName,
Subjects = [subjectPattern],
});
return sm;
}
private static async Task AppendAsync(StreamManager sm, string subject, string payload)
{
var handle = sm.FindBySubject(subject);
handle.ShouldNotBeNull();
await handle!.Store.AppendAsync(subject, Encoding.UTF8.GetBytes(payload), default);
}
// ---------------------------------------------------------------
// HandleSnapshot (sync, existing)
// ---------------------------------------------------------------
/// <summary>
/// Go ref: jsStreamSnapshotT — snapshot of an existing stream returns a non-empty base64 payload.
/// </summary>
[Fact]
public async Task HandleSnapshot_returns_base64_payload_for_existing_stream()
{
var sm = CreateManagerWithStream("ORDERS", "orders.>");
await AppendAsync(sm, "orders.1", "hello");
var response = StreamApiHandlers.HandleSnapshot("$JS.API.STREAM.SNAPSHOT.ORDERS", sm);
response.Error.ShouldBeNull();
response.Snapshot.ShouldNotBeNull();
response.Snapshot!.Payload.ShouldNotBeNullOrEmpty();
// Verify it is valid base64.
var bytes = Convert.FromBase64String(response.Snapshot.Payload);
bytes.ShouldNotBeEmpty();
}
/// <summary>
/// Go ref: jsStreamSnapshotT — snapshot of a non-existent stream returns 404.
/// </summary>
[Fact]
public void HandleSnapshot_returns_not_found_for_missing_stream()
{
var sm = new StreamManager();
var response = StreamApiHandlers.HandleSnapshot("$JS.API.STREAM.SNAPSHOT.MISSING", sm);
response.Error.ShouldNotBeNull();
response.Error!.Code.ShouldBe(404);
}
// ---------------------------------------------------------------
// HandleRestore (sync, existing)
// ---------------------------------------------------------------
/// <summary>
/// Go ref: jsStreamRestoreT — restore with a valid base64 snapshot payload succeeds.
/// </summary>
[Fact]
public async Task HandleRestore_succeeds_with_valid_payload()
{
var sm = CreateManagerWithStream("ORDERS", "orders.>");
await AppendAsync(sm, "orders.1", "msg1");
// Obtain a snapshot first.
var snapshotResponse = StreamApiHandlers.HandleSnapshot("$JS.API.STREAM.SNAPSHOT.ORDERS", sm);
snapshotResponse.Snapshot.ShouldNotBeNull();
var base64 = snapshotResponse.Snapshot!.Payload;
// Restore back using the base64 bytes directly as the payload.
var payloadBytes = Encoding.UTF8.GetBytes(base64);
var response = StreamApiHandlers.HandleRestore(
"$JS.API.STREAM.RESTORE.ORDERS",
payloadBytes,
sm);
response.Error.ShouldBeNull();
response.Success.ShouldBeTrue();
}
/// <summary>
/// Go ref: jsStreamRestoreT — empty payload returns a 400 error.
/// </summary>
[Fact]
public void HandleRestore_returns_error_for_empty_payload()
{
var sm = CreateManagerWithStream("ORDERS", "orders.>");
var response = StreamApiHandlers.HandleRestore(
"$JS.API.STREAM.RESTORE.ORDERS",
ReadOnlySpan<byte>.Empty,
sm);
response.Error.ShouldNotBeNull();
response.Error!.Code.ShouldBe(400);
}
/// <summary>
/// Go ref: jsStreamRestoreT — bad subject token (no trailing stream name) returns 404.
/// </summary>
[Fact]
public void HandleRestore_returns_not_found_for_bad_subject()
{
var sm = new StreamManager();
// Subject without trailing token — ExtractTrailingToken returns null.
var payload = Encoding.UTF8.GetBytes(Convert.ToBase64String([1, 2, 3]));
var response = StreamApiHandlers.HandleRestore(
"$JS.API.STREAM.RESTORE.",
payload,
sm);
response.Error.ShouldNotBeNull();
response.Error!.Code.ShouldBe(404);
}
// ---------------------------------------------------------------
// HandleSnapshotAsync (new)
// ---------------------------------------------------------------
/// <summary>
/// Go ref: jsStreamSnapshotT — async handler populates StreamName in the response.
/// </summary>
[Fact]
public async Task HandleSnapshotAsync_includes_stream_name_in_response()
{
var sm = CreateManagerWithStream("EVENTS", "events.>");
await AppendAsync(sm, "events.1", "data");
var response = await StreamApiHandlers.HandleSnapshotAsync(
"$JS.API.STREAM.SNAPSHOT.EVENTS",
sm,
CancellationToken.None);
response.Error.ShouldBeNull();
response.Snapshot.ShouldNotBeNull();
response.Snapshot!.StreamName.ShouldBe("EVENTS");
}
/// <summary>
/// Go ref: jsStreamSnapshotT — async handler sets NumChunks=1 and BlkSize equal to the
/// length of the raw (pre-base64) snapshot bytes.
/// </summary>
[Fact]
public async Task HandleSnapshotAsync_includes_chunk_metadata()
{
var sm = CreateManagerWithStream("EVENTS", "events.>");
await AppendAsync(sm, "events.1", "payload-data");
var response = await StreamApiHandlers.HandleSnapshotAsync(
"$JS.API.STREAM.SNAPSHOT.EVENTS",
sm,
CancellationToken.None);
response.Error.ShouldBeNull();
var snap = response.Snapshot!;
snap.NumChunks.ShouldBe(1);
snap.BlkSize.ShouldBeGreaterThan(0);
// BlkSize should match the raw snapshot byte count.
var rawBytes = Convert.FromBase64String(snap.Payload);
snap.BlkSize.ShouldBe(rawBytes.Length);
}
/// <summary>
/// HandleSnapshotAsync returns 404 when the stream does not exist.
/// </summary>
[Fact]
public async Task HandleSnapshotAsync_returns_not_found_for_missing_stream()
{
var sm = new StreamManager();
var response = await StreamApiHandlers.HandleSnapshotAsync(
"$JS.API.STREAM.SNAPSHOT.NOPE",
sm,
CancellationToken.None);
response.Error.ShouldNotBeNull();
response.Error!.Code.ShouldBe(404);
}
// ---------------------------------------------------------------
// HandleRestoreAsync (new)
// ---------------------------------------------------------------
/// <summary>
/// Go ref: jsStreamRestoreT — async restore validates the base64 payload and succeeds.
/// </summary>
[Fact]
public async Task HandleRestoreAsync_validates_base64_payload()
{
var sm = CreateManagerWithStream("ORDERS", "orders.>");
await AppendAsync(sm, "orders.1", "hello");
// Take a snapshot, then restore it using the async path.
var snapResp = await StreamApiHandlers.HandleSnapshotAsync(
"$JS.API.STREAM.SNAPSHOT.ORDERS",
sm,
CancellationToken.None);
snapResp.Snapshot.ShouldNotBeNull();
var base64Payload = Encoding.UTF8.GetBytes(snapResp.Snapshot!.Payload);
var response = await StreamApiHandlers.HandleRestoreAsync(
"$JS.API.STREAM.RESTORE.ORDERS",
base64Payload,
sm,
CancellationToken.None);
response.Error.ShouldBeNull();
response.Success.ShouldBeTrue();
}
/// <summary>
/// HandleRestoreAsync returns 400 when given an empty payload array.
/// </summary>
[Fact]
public async Task HandleRestoreAsync_returns_error_for_empty_payload()
{
var sm = CreateManagerWithStream("ORDERS", "orders.>");
var response = await StreamApiHandlers.HandleRestoreAsync(
"$JS.API.STREAM.RESTORE.ORDERS",
[],
sm,
CancellationToken.None);
response.Error.ShouldNotBeNull();
response.Error!.Code.ShouldBe(400);
}
// ---------------------------------------------------------------
// Round-trip
// ---------------------------------------------------------------
/// <summary>
/// Go ref: jsStreamSnapshotT / jsStreamRestoreT — full snapshot-then-restore round-trip:
/// messages written before snapshot are recoverable after restore.
/// </summary>
[Fact]
public async Task Snapshot_round_trip_create_and_restore()
{
var sm = CreateManagerWithStream("LOGS", "logs.>");
await AppendAsync(sm, "logs.a", "alpha");
await AppendAsync(sm, "logs.b", "beta");
await AppendAsync(sm, "logs.c", "gamma");
var stateBefore = await sm.GetStateAsync("LOGS", default);
stateBefore.Messages.ShouldBe(3UL);
// Snapshot via async handler.
var snapResp = await StreamApiHandlers.HandleSnapshotAsync(
"$JS.API.STREAM.SNAPSHOT.LOGS",
sm,
CancellationToken.None);
snapResp.Error.ShouldBeNull();
var base64Payload = Encoding.UTF8.GetBytes(snapResp.Snapshot!.Payload);
// Restore via async handler.
var restoreResp = await StreamApiHandlers.HandleRestoreAsync(
"$JS.API.STREAM.RESTORE.LOGS",
base64Payload,
sm,
CancellationToken.None);
restoreResp.Error.ShouldBeNull();
restoreResp.Success.ShouldBeTrue();
// State should still be consistent (restore does not clear — it re-applies).
var stateAfter = await sm.GetStateAsync("LOGS", default);
stateAfter.Messages.ShouldBeGreaterThanOrEqualTo(3UL);
}
// ---------------------------------------------------------------
// Subject extraction
// ---------------------------------------------------------------
/// <summary>
/// Go ref: jsStreamSnapshotT — the stream name is correctly extracted from the API subject.
/// </summary>
[Fact]
public async Task HandleSnapshot_extracts_stream_name_from_subject()
{
var sm = CreateManagerWithStream("MY_STREAM", "mystream.>");
var response = await StreamApiHandlers.HandleSnapshotAsync(
"$JS.API.STREAM.SNAPSHOT.MY_STREAM",
sm,
CancellationToken.None);
response.Error.ShouldBeNull();
response.Snapshot.ShouldNotBeNull();
response.Snapshot!.StreamName.ShouldBe("MY_STREAM");
}
}

193
tests/NATS.Server.JetStream.Tests/JetStream/Api/StreamPurgeOptionsTests.cs Normal file
View File

@@ -0,0 +1,193 @@
// Go reference: jetstream_api.go:1200-1350 — stream purge supports options: subject filter,
// sequence cutoff, and keep-last-N. Combinations like filter+keep allow keeping the last N
// messages per matching subject.
using System.Text;
using NATS.Server.JetStream;
using NATS.Server.JetStream.Api;
namespace NATS.Server.JetStream.Tests.JetStream.Api;
public class StreamPurgeOptionsTests
{
private static JetStreamApiRouter CreateRouterWithStream(string streamName, string subjectPattern, out StreamManager streamManager)
{
streamManager = new StreamManager();
var consumerManager = new ConsumerManager();
var router = new JetStreamApiRouter(streamManager, consumerManager);
var payload = Encoding.UTF8.GetBytes($$$"""{"name":"{{{streamName}}}","subjects":["{{{subjectPattern}}}"]}""");
var result = router.Route($"$JS.API.STREAM.CREATE.{streamName}", payload);
result.Error.ShouldBeNull();
return router;
}
private static async Task PublishAsync(StreamManager streamManager, string subject, string payload)
{
var stream = streamManager.FindBySubject(subject);
stream.ShouldNotBeNull();
await stream.Store.AppendAsync(subject, Encoding.UTF8.GetBytes(payload), default);
}
/// <summary>
/// Purge with no options removes all messages and returns the count.
/// Go reference: jetstream_api.go — basic purge with empty request body.
/// </summary>
[Fact]
public async Task Purge_NoOptions_RemovesAll()
{
var router = CreateRouterWithStream("TEST", "test.>", out var sm);
await PublishAsync(sm, "test.a", "1");
await PublishAsync(sm, "test.b", "2");
await PublishAsync(sm, "test.c", "3");
var result = router.Route("$JS.API.STREAM.PURGE.TEST", Encoding.UTF8.GetBytes("{}"));
result.Error.ShouldBeNull();
result.Success.ShouldBeTrue();
result.Purged.ShouldBe(3UL);
var state = await sm.GetStateAsync("TEST", default);
state.Messages.ShouldBe(0UL);
}
/// <summary>
/// Purge with a subject filter removes only messages matching the pattern.
/// Go reference: jetstream_api.go:1200-1350 — filter option.
/// </summary>
[Fact]
public async Task Purge_WithSubjectFilter_RemovesOnlyMatching()
{
var router = CreateRouterWithStream("TEST", ">", out var sm);
await PublishAsync(sm, "orders.a", "1");
await PublishAsync(sm, "orders.b", "2");
await PublishAsync(sm, "logs.x", "3");
await PublishAsync(sm, "orders.c", "4");
var payload = Encoding.UTF8.GetBytes("""{"filter":"orders.*"}""");
var result = router.Route("$JS.API.STREAM.PURGE.TEST", payload);
result.Error.ShouldBeNull();
result.Success.ShouldBeTrue();
result.Purged.ShouldBe(3UL);
var state = await sm.GetStateAsync("TEST", default);
state.Messages.ShouldBe(1UL);
}
/// <summary>
/// Purge with seq option removes all messages with sequence strictly less than the given value.
/// Go reference: jetstream_api.go:1200-1350 — seq option.
/// </summary>
[Fact]
public async Task Purge_WithSeq_RemovesBelowSequence()
{
var router = CreateRouterWithStream("TEST", "test.>", out var sm);
await PublishAsync(sm, "test.a", "1"); // seq 1
await PublishAsync(sm, "test.b", "2"); // seq 2
await PublishAsync(sm, "test.c", "3"); // seq 3
await PublishAsync(sm, "test.d", "4"); // seq 4
await PublishAsync(sm, "test.e", "5"); // seq 5
// Remove all messages with seq < 4 (i.e., sequences 1, 2, 3).
var payload = Encoding.UTF8.GetBytes("""{"seq":4}""");
var result = router.Route("$JS.API.STREAM.PURGE.TEST", payload);
result.Error.ShouldBeNull();
result.Success.ShouldBeTrue();
result.Purged.ShouldBe(3UL);
var state = await sm.GetStateAsync("TEST", default);
state.Messages.ShouldBe(2UL);
}
/// <summary>
/// Purge with keep option retains the last N messages globally.
/// Go reference: jetstream_api.go:1200-1350 — keep option.
/// </summary>
[Fact]
public async Task Purge_WithKeep_KeepsLastN()
{
var router = CreateRouterWithStream("TEST", "test.>", out var sm);
await PublishAsync(sm, "test.a", "1"); // seq 1
await PublishAsync(sm, "test.b", "2"); // seq 2
await PublishAsync(sm, "test.c", "3"); // seq 3
await PublishAsync(sm, "test.d", "4"); // seq 4
await PublishAsync(sm, "test.e", "5"); // seq 5
// Keep the last 2 messages (seq 4, 5); purge 1, 2, 3.
var payload = Encoding.UTF8.GetBytes("""{"keep":2}""");
var result = router.Route("$JS.API.STREAM.PURGE.TEST", payload);
result.Error.ShouldBeNull();
result.Success.ShouldBeTrue();
result.Purged.ShouldBe(3UL);
var state = await sm.GetStateAsync("TEST", default);
state.Messages.ShouldBe(2UL);
}
/// <summary>
/// Purge with both filter and keep retains the last N messages per matching subject.
/// Go reference: jetstream_api.go:1200-1350 — filter+keep combination.
/// </summary>
[Fact]
public async Task Purge_FilterAndKeep_KeepsLastNPerFilter()
{
var router = CreateRouterWithStream("TEST", ">", out var sm);
// Publish multiple messages on two subjects.
await PublishAsync(sm, "orders.a", "o1"); // seq 1
await PublishAsync(sm, "orders.a", "o2"); // seq 2
await PublishAsync(sm, "orders.a", "o3"); // seq 3
await PublishAsync(sm, "logs.x", "l1"); // seq 4 — not matching filter
await PublishAsync(sm, "orders.b", "ob1"); // seq 5
await PublishAsync(sm, "orders.b", "ob2"); // seq 6
// Keep last 1 per matching subject "orders.*".
// orders.a has 3 msgs -> keep seq 3, purge seq 1, 2
// orders.b has 2 msgs -> keep seq 6, purge seq 5
// logs.x is unaffected (does not match filter)
var payload = Encoding.UTF8.GetBytes("""{"filter":"orders.*","keep":1}""");
var result = router.Route("$JS.API.STREAM.PURGE.TEST", payload);
result.Error.ShouldBeNull();
result.Success.ShouldBeTrue();
result.Purged.ShouldBe(3UL);
var state = await sm.GetStateAsync("TEST", default);
// Remaining: orders.a seq 3, logs.x seq 4, orders.b seq 6 = 3 messages
state.Messages.ShouldBe(3UL);
}
/// <summary>
/// Purge on a non-existent stream returns a 404 not-found error.
/// Go reference: jetstream_api.go — stream not found.
/// </summary>
[Fact]
public void Purge_InvalidStream_ReturnsNotFound()
{
var streamManager = new StreamManager();
var consumerManager = new ConsumerManager();
var router = new JetStreamApiRouter(streamManager, consumerManager);
var result = router.Route("$JS.API.STREAM.PURGE.NONEXISTENT", Encoding.UTF8.GetBytes("{}"));
result.Error.ShouldNotBeNull();
result.Error!.Code.ShouldBe(404);
}
/// <summary>
/// Purge on an empty stream returns success with zero purged count.
/// Go reference: jetstream_api.go — purge on empty stream.
/// </summary>
[Fact]
public void Purge_EmptyStream_ReturnsZeroPurged()
{
var router = CreateRouterWithStream("TEST", "test.>", out _);
var result = router.Route("$JS.API.STREAM.PURGE.TEST", Encoding.UTF8.GetBytes("{}"));
result.Error.ShouldBeNull();
result.Success.ShouldBeTrue();
result.Purged.ShouldBe(0UL);
}
}