78b4bc2486
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.
810 lines
31 KiB
C#
810 lines
31 KiB
C#
// Go reference: golang/nats-server/server/jetstream_test.go
|
|
// Ports a representative subset (~35 tests) covering stream CRUD, consumer
|
|
// create/delete, publish/subscribe flow, purge, retention policies,
|
|
// mirror/source, and validation. All mapped to existing .NET infrastructure.
|
|
|
|
using NATS.Server.JetStream;
|
|
using NATS.Server.JetStream.Api;
|
|
using NATS.Server.JetStream.Models;
|
|
using NATS.Server.TestUtilities;
|
|
|
|
namespace NATS.Server.JetStream.Tests.JetStream;
|
|
|
|
/// <summary>
|
|
/// Go parity tests ported from jetstream_test.go for core JetStream behaviors
|
|
/// including stream lifecycle, publish/subscribe, purge, retention, mirroring,
|
|
/// and configuration validation.
|
|
/// </summary>
|
|
public class JetStreamGoParityTests
|
|
{
|
|
// =========================================================================
|
|
// TestJetStreamAddStream — jetstream_test.go:178
|
|
// Adding a stream and publishing messages should update state correctly.
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task AddStream_and_publish_updates_state()
|
|
{
|
|
// Go: TestJetStreamAddStream jetstream_test.go:178
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("foo", "foo");
|
|
|
|
var ack1 = await fx.PublishAndGetAckAsync("foo", "Hello World!");
|
|
ack1.ErrorCode.ShouldBeNull();
|
|
ack1.Seq.ShouldBe(1UL);
|
|
|
|
var state = await fx.GetStreamStateAsync("foo");
|
|
state.Messages.ShouldBe(1UL);
|
|
|
|
var ack2 = await fx.PublishAndGetAckAsync("foo", "Hello World Again!");
|
|
ack2.Seq.ShouldBe(2UL);
|
|
|
|
state = await fx.GetStreamStateAsync("foo");
|
|
state.Messages.ShouldBe(2UL);
|
|
}
|
|
|
|
// =========================================================================
|
|
// TestJetStreamAddStreamDiscardNew — jetstream_test.go:236
|
|
// Discard new policy rejects messages when stream is full.
|
|
// =========================================================================
|
|
|
|
[Fact(Skip = "DiscardPolicy.New enforcement for MaxMsgs not yet implemented in .NET server — only MaxBytes is checked")]
|
|
public async Task AddStream_discard_new_rejects_when_full()
|
|
{
|
|
// Go: TestJetStreamAddStreamDiscardNew jetstream_test.go:236
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamConfigAsync(new StreamConfig
|
|
{
|
|
Name = "foo",
|
|
Subjects = ["foo"],
|
|
MaxMsgs = 3,
|
|
Discard = DiscardPolicy.New,
|
|
});
|
|
|
|
for (int i = 0; i < 3; i++)
|
|
{
|
|
var ack = await fx.PublishAndGetAckAsync("foo", $"msg{i}");
|
|
ack.ErrorCode.ShouldBeNull();
|
|
}
|
|
|
|
// 4th message should be rejected
|
|
var rejected = await fx.PublishAndGetAckAsync("foo", "overflow", expectError: true);
|
|
rejected.ErrorCode.ShouldNotBeNull();
|
|
}
|
|
|
|
// =========================================================================
|
|
// TestJetStreamAddStreamMaxMsgSize — jetstream_test.go:450
|
|
// MaxMsgSize enforcement on stream.
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task AddStream_max_msg_size_rejects_oversized()
|
|
{
|
|
// Go: TestJetStreamAddStreamMaxMsgSize jetstream_test.go:450
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamConfigAsync(new StreamConfig
|
|
{
|
|
Name = "SIZED",
|
|
Subjects = ["sized.>"],
|
|
MaxMsgSize = 10,
|
|
});
|
|
|
|
var small = await fx.PublishAndGetAckAsync("sized.ok", "tiny");
|
|
small.ErrorCode.ShouldBeNull();
|
|
|
|
var big = await fx.PublishAndGetAckAsync("sized.big", "this-is-way-too-large-for-the-limit");
|
|
big.ErrorCode.ShouldNotBeNull();
|
|
}
|
|
|
|
// =========================================================================
|
|
// TestJetStreamAddStreamCanonicalNames — jetstream_test.go:502
|
|
// Stream name is preserved exactly as created.
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task AddStream_canonical_name_preserved()
|
|
{
|
|
// Go: TestJetStreamAddStreamCanonicalNames jetstream_test.go:502
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("MyStream", "my.>");
|
|
|
|
var info = await fx.RequestLocalAsync("$JS.API.STREAM.INFO.MyStream", "{}");
|
|
info.Error.ShouldBeNull();
|
|
info.StreamInfo!.Config.Name.ShouldBe("MyStream");
|
|
}
|
|
|
|
// =========================================================================
|
|
// TestJetStreamAddStreamSameConfigOK — jetstream_test.go:701
|
|
// Re-creating a stream with the same config is idempotent.
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task AddStream_same_config_is_idempotent()
|
|
{
|
|
// Go: TestJetStreamAddStreamSameConfigOK jetstream_test.go:701
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("ORDERS", "orders.*");
|
|
|
|
var second = await fx.RequestLocalAsync(
|
|
"$JS.API.STREAM.CREATE.ORDERS",
|
|
"""{"name":"ORDERS","subjects":["orders.*"]}""");
|
|
second.Error.ShouldBeNull();
|
|
second.StreamInfo!.Config.Name.ShouldBe("ORDERS");
|
|
}
|
|
|
|
// =========================================================================
|
|
// TestJetStreamPubAck — jetstream_test.go:354
|
|
// Publish acknowledges with correct stream name and sequence.
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task PubAck_returns_correct_stream_and_sequence()
|
|
{
|
|
// Go: TestJetStreamPubAck jetstream_test.go:354
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("PUBACK", "foo");
|
|
|
|
for (ulong i = 1; i <= 10; i++)
|
|
{
|
|
var ack = await fx.PublishAndGetAckAsync("foo", $"HELLO-{i}");
|
|
ack.ErrorCode.ShouldBeNull();
|
|
ack.Stream.ShouldBe("PUBACK");
|
|
ack.Seq.ShouldBe(i);
|
|
}
|
|
}
|
|
|
|
// =========================================================================
|
|
// TestJetStreamBasicAckPublish — jetstream_test.go:737
|
|
// Basic ack publish with sequence tracking.
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task BasicAckPublish_sequences_increment()
|
|
{
|
|
// Go: TestJetStreamBasicAckPublish jetstream_test.go:737
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("TEST", "test.>");
|
|
|
|
var ack1 = await fx.PublishAndGetAckAsync("test.a", "msg1");
|
|
ack1.Seq.ShouldBe(1UL);
|
|
|
|
var ack2 = await fx.PublishAndGetAckAsync("test.b", "msg2");
|
|
ack2.Seq.ShouldBe(2UL);
|
|
|
|
var ack3 = await fx.PublishAndGetAckAsync("test.c", "msg3");
|
|
ack3.Seq.ShouldBe(3UL);
|
|
}
|
|
|
|
// =========================================================================
|
|
// Stream state after publish — jetstream_test.go:770
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task Stream_state_tracks_messages_and_bytes()
|
|
{
|
|
// Go: TestJetStreamStateTimestamps jetstream_test.go:770
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("STATE", "state.>");
|
|
|
|
var state0 = await fx.GetStreamStateAsync("STATE");
|
|
state0.Messages.ShouldBe(0UL);
|
|
|
|
await fx.PublishAndGetAckAsync("state.a", "hello");
|
|
var state1 = await fx.GetStreamStateAsync("STATE");
|
|
state1.Messages.ShouldBe(1UL);
|
|
state1.Bytes.ShouldBeGreaterThan(0UL);
|
|
|
|
await fx.PublishAndGetAckAsync("state.b", "world");
|
|
var state2 = await fx.GetStreamStateAsync("STATE");
|
|
state2.Messages.ShouldBe(2UL);
|
|
state2.Bytes.ShouldBeGreaterThan(state1.Bytes);
|
|
}
|
|
|
|
// =========================================================================
|
|
// TestJetStreamStreamPurge — jetstream_test.go:4182
|
|
// Purging a stream resets message count and timestamps.
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task Stream_purge_resets_state()
|
|
{
|
|
// Go: TestJetStreamStreamPurge jetstream_test.go:4182
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("DC", "DC");
|
|
|
|
// Publish 100 messages
|
|
for (int i = 0; i < 100; i++)
|
|
await fx.PublishAndGetAckAsync("DC", $"msg{i}");
|
|
|
|
var state = await fx.GetStreamStateAsync("DC");
|
|
state.Messages.ShouldBe(100UL);
|
|
|
|
// Purge
|
|
var purgeResponse = await fx.RequestLocalAsync("$JS.API.STREAM.PURGE.DC", "{}");
|
|
purgeResponse.Error.ShouldBeNull();
|
|
|
|
state = await fx.GetStreamStateAsync("DC");
|
|
state.Messages.ShouldBe(0UL);
|
|
|
|
// Publish after purge
|
|
await fx.PublishAndGetAckAsync("DC", "after-purge");
|
|
state = await fx.GetStreamStateAsync("DC");
|
|
state.Messages.ShouldBe(1UL);
|
|
}
|
|
|
|
// =========================================================================
|
|
// TestJetStreamStreamPurgeWithConsumer — jetstream_test.go:4238
|
|
// Purging a stream that has consumers attached.
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task Stream_purge_with_consumer_attached()
|
|
{
|
|
// Go: TestJetStreamStreamPurgeWithConsumer jetstream_test.go:4238
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("DC", "DC");
|
|
await fx.CreateConsumerAsync("DC", "C1", "DC");
|
|
|
|
for (int i = 0; i < 50; i++)
|
|
await fx.PublishAndGetAckAsync("DC", $"msg{i}");
|
|
|
|
var state = await fx.GetStreamStateAsync("DC");
|
|
state.Messages.ShouldBe(50UL);
|
|
|
|
await fx.RequestLocalAsync("$JS.API.STREAM.PURGE.DC", "{}");
|
|
|
|
state = await fx.GetStreamStateAsync("DC");
|
|
state.Messages.ShouldBe(0UL);
|
|
}
|
|
|
|
// =========================================================================
|
|
// Consumer create and delete
|
|
// =========================================================================
|
|
|
|
// TestJetStreamMaxConsumers — jetstream_test.go:553
|
|
[Fact]
|
|
public async Task Consumer_create_succeeds()
|
|
{
|
|
// Go: TestJetStreamMaxConsumers jetstream_test.go:553
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("TEST", "test.>");
|
|
|
|
var r1 = await fx.CreateConsumerAsync("TEST", "C1", "test.a");
|
|
r1.Error.ShouldBeNull();
|
|
|
|
var r2 = await fx.CreateConsumerAsync("TEST", "C2", "test.b");
|
|
r2.Error.ShouldBeNull();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Consumer_delete_succeeds()
|
|
{
|
|
// Go: TestJetStreamConsumerDelete consumer tests
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("TEST", "test.>");
|
|
await fx.CreateConsumerAsync("TEST", "C1", "test.a");
|
|
|
|
var delete = await fx.RequestLocalAsync("$JS.API.CONSUMER.DELETE.TEST.C1", "{}");
|
|
delete.Error.ShouldBeNull();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Consumer_info_returns_config()
|
|
{
|
|
// Go: consumer info endpoint
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("TEST", "test.>");
|
|
await fx.CreateConsumerAsync("TEST", "C1", "test.a",
|
|
ackPolicy: AckPolicy.Explicit, ackWaitMs: 5000);
|
|
|
|
var info = await fx.GetConsumerInfoAsync("TEST", "C1");
|
|
info.Config.DurableName.ShouldBe("C1");
|
|
info.Config.AckPolicy.ShouldBe(AckPolicy.Explicit);
|
|
}
|
|
|
|
// =========================================================================
|
|
// TestJetStreamSubjectFiltering — jetstream_test.go:1385
|
|
// Subject filtering on consumers.
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task Subject_filtering_on_consumer()
|
|
{
|
|
// Go: TestJetStreamSubjectFiltering jetstream_test.go:1385
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("FILTER", ">");
|
|
|
|
await fx.CreateConsumerAsync("FILTER", "CF", "orders.*");
|
|
|
|
await fx.PublishAndGetAckAsync("orders.created", "o1");
|
|
await fx.PublishAndGetAckAsync("payments.settled", "p1");
|
|
await fx.PublishAndGetAckAsync("orders.updated", "o2");
|
|
|
|
var batch = await fx.FetchAsync("FILTER", "CF", 10);
|
|
batch.Messages.Count.ShouldBe(2);
|
|
batch.Messages.All(m => m.Subject.StartsWith("orders.", StringComparison.Ordinal)).ShouldBeTrue();
|
|
}
|
|
|
|
// =========================================================================
|
|
// TestJetStreamWildcardSubjectFiltering — jetstream_test.go:1522
|
|
// Wildcard subject filtering.
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task Wildcard_subject_filtering_on_consumer()
|
|
{
|
|
// Go: TestJetStreamWildcardSubjectFiltering jetstream_test.go:1522
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("WF", ">");
|
|
|
|
await fx.CreateConsumerAsync("WF", "CF", "data.*.info");
|
|
|
|
await fx.PublishAndGetAckAsync("data.us.info", "us-info");
|
|
await fx.PublishAndGetAckAsync("data.eu.info", "eu-info");
|
|
await fx.PublishAndGetAckAsync("data.us.debug", "us-debug");
|
|
|
|
var batch = await fx.FetchAsync("WF", "CF", 10);
|
|
batch.Messages.Count.ShouldBe(2);
|
|
batch.Messages.All(m => m.Subject.EndsWith(".info", StringComparison.Ordinal)).ShouldBeTrue();
|
|
}
|
|
|
|
// =========================================================================
|
|
// TestJetStreamBasicWorkQueue — jetstream_test.go:1000
|
|
// Work queue retention policy.
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task WorkQueue_retention_deletes_on_ack()
|
|
{
|
|
// Go: TestJetStreamBasicWorkQueue jetstream_test.go:1000
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamConfigAsync(new StreamConfig
|
|
{
|
|
Name = "WQ",
|
|
Subjects = ["wq.>"],
|
|
Retention = RetentionPolicy.WorkQueue,
|
|
});
|
|
|
|
await fx.CreateConsumerAsync("WQ", "WORKER", "wq.>",
|
|
ackPolicy: AckPolicy.Explicit);
|
|
|
|
await fx.PublishAndGetAckAsync("wq.task1", "job1");
|
|
await fx.PublishAndGetAckAsync("wq.task2", "job2");
|
|
|
|
var state = await fx.GetStreamStateAsync("WQ");
|
|
state.Messages.ShouldBe(2UL);
|
|
}
|
|
|
|
// =========================================================================
|
|
// TestJetStreamInterestRetentionStream — jetstream_test.go:4411
|
|
// Interest retention policy.
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task Interest_retention_stream_creation()
|
|
{
|
|
// Go: TestJetStreamInterestRetentionStream jetstream_test.go:4411
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamConfigAsync(new StreamConfig
|
|
{
|
|
Name = "IR",
|
|
Subjects = ["ir.>"],
|
|
Retention = RetentionPolicy.Interest,
|
|
});
|
|
|
|
var info = await fx.RequestLocalAsync("$JS.API.STREAM.INFO.IR", "{}");
|
|
info.Error.ShouldBeNull();
|
|
info.StreamInfo!.Config.Retention.ShouldBe(RetentionPolicy.Interest);
|
|
}
|
|
|
|
// =========================================================================
|
|
// Mirror configuration
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task Mirror_stream_configuration()
|
|
{
|
|
// Go: mirror-related tests in jetstream_test.go
|
|
await using var fx = await JetStreamApiFixture.StartWithMirrorSetupAsync();
|
|
|
|
var mirrorInfo = await fx.RequestLocalAsync("$JS.API.STREAM.INFO.ORDERS_MIRROR", "{}");
|
|
mirrorInfo.Error.ShouldBeNull();
|
|
mirrorInfo.StreamInfo!.Config.Mirror.ShouldBe("ORDERS");
|
|
}
|
|
|
|
// =========================================================================
|
|
// Source configuration
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task Source_stream_configuration()
|
|
{
|
|
// Go: source-related tests in jetstream_test.go
|
|
await using var fx = await JetStreamApiFixture.StartWithMultipleSourcesAsync();
|
|
|
|
var aggInfo = await fx.RequestLocalAsync("$JS.API.STREAM.INFO.AGG", "{}");
|
|
aggInfo.Error.ShouldBeNull();
|
|
aggInfo.StreamInfo!.Config.Sources.Count.ShouldBe(2);
|
|
}
|
|
|
|
// =========================================================================
|
|
// Stream list
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task Stream_list_returns_all_streams()
|
|
{
|
|
// Go: stream list API
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("S1", "s1.>");
|
|
|
|
var r2 = await fx.CreateStreamAsync("S2", ["s2.>"]);
|
|
r2.Error.ShouldBeNull();
|
|
|
|
var list = await fx.RequestLocalAsync("$JS.API.STREAM.LIST", "{}");
|
|
list.Error.ShouldBeNull();
|
|
}
|
|
|
|
// =========================================================================
|
|
// Consumer list
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task Consumer_list_returns_all_consumers()
|
|
{
|
|
// Go: consumer list API
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("TEST", ">");
|
|
await fx.CreateConsumerAsync("TEST", "C1", "one");
|
|
await fx.CreateConsumerAsync("TEST", "C2", "two");
|
|
|
|
var list = await fx.RequestLocalAsync("$JS.API.CONSUMER.LIST.TEST", "{}");
|
|
list.Error.ShouldBeNull();
|
|
}
|
|
|
|
// =========================================================================
|
|
// TestJetStreamPublishDeDupe — jetstream_test.go:2657
|
|
// Deduplication via Nats-Msg-Id header.
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task Publish_dedup_with_msg_id()
|
|
{
|
|
// Go: TestJetStreamPublishDeDupe jetstream_test.go:2657
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamConfigAsync(new StreamConfig
|
|
{
|
|
Name = "DEDUP",
|
|
Subjects = ["dedup.>"],
|
|
DuplicateWindowMs = 60_000,
|
|
});
|
|
|
|
var ack1 = await fx.PublishAndGetAckAsync("dedup.test", "msg1", msgId: "unique-1");
|
|
ack1.ErrorCode.ShouldBeNull();
|
|
ack1.Seq.ShouldBe(1UL);
|
|
|
|
// Same msg ID should be deduplicated — publisher sets ErrorCode (not Duplicate flag)
|
|
var ack2 = await fx.PublishAndGetAckAsync("dedup.test", "msg1-again", msgId: "unique-1");
|
|
ack2.ErrorCode.ShouldNotBeNull();
|
|
|
|
// Different msg ID should succeed
|
|
var ack3 = await fx.PublishAndGetAckAsync("dedup.test", "msg2", msgId: "unique-2");
|
|
ack3.ErrorCode.ShouldBeNull();
|
|
ack3.Seq.ShouldBe(2UL);
|
|
}
|
|
|
|
// =========================================================================
|
|
// TestJetStreamPublishExpect — jetstream_test.go:2817
|
|
// Publish with expected last sequence precondition.
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task Publish_with_expected_last_seq()
|
|
{
|
|
// Go: TestJetStreamPublishExpect jetstream_test.go:2817
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("EXPECT", "expect.>");
|
|
|
|
var ack1 = await fx.PublishAndGetAckAsync("expect.a", "msg1");
|
|
ack1.Seq.ShouldBe(1UL);
|
|
|
|
// Correct expected last seq should succeed
|
|
var ack2 = await fx.PublishWithExpectedLastSeqAsync("expect.b", "msg2", 1UL);
|
|
ack2.ErrorCode.ShouldBeNull();
|
|
|
|
// Wrong expected last seq should fail
|
|
var ack3 = await fx.PublishWithExpectedLastSeqAsync("expect.c", "msg3", 99UL);
|
|
ack3.ErrorCode.ShouldNotBeNull();
|
|
}
|
|
|
|
// =========================================================================
|
|
// Stream delete
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task Stream_delete_removes_stream()
|
|
{
|
|
// Go: mset.delete() in various tests
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("DEL", "del.>");
|
|
|
|
await fx.PublishAndGetAckAsync("del.a", "msg1");
|
|
|
|
var deleteResponse = await fx.RequestLocalAsync("$JS.API.STREAM.DELETE.DEL", "{}");
|
|
deleteResponse.Error.ShouldBeNull();
|
|
|
|
var info = await fx.RequestLocalAsync("$JS.API.STREAM.INFO.DEL", "{}");
|
|
info.Error.ShouldNotBeNull();
|
|
}
|
|
|
|
// =========================================================================
|
|
// Fetch with no messages returns empty batch
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task Fetch_with_no_messages_returns_empty()
|
|
{
|
|
// Go: basic fetch behavior
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("EMPTY", "empty.>");
|
|
await fx.CreateConsumerAsync("EMPTY", "C1", "empty.>");
|
|
|
|
var batch = await fx.FetchWithNoWaitAsync("EMPTY", "C1", 10);
|
|
batch.Messages.ShouldBeEmpty();
|
|
}
|
|
|
|
// =========================================================================
|
|
// Fetch returns published messages in order
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task Fetch_returns_messages_in_order()
|
|
{
|
|
// Go: basic fetch behavior
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("ORDERED", "ordered.>");
|
|
await fx.CreateConsumerAsync("ORDERED", "C1", "ordered.>");
|
|
|
|
for (int i = 0; i < 5; i++)
|
|
await fx.PublishAndGetAckAsync("ordered.test", $"msg{i}");
|
|
|
|
var batch = await fx.FetchAsync("ORDERED", "C1", 10);
|
|
batch.Messages.Count.ShouldBe(5);
|
|
|
|
for (int i = 1; i < batch.Messages.Count; i++)
|
|
{
|
|
batch.Messages[i].Sequence.ShouldBeGreaterThan(batch.Messages[i - 1].Sequence);
|
|
}
|
|
}
|
|
|
|
// =========================================================================
|
|
// MaxMsgs enforcement — old messages evicted
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task MaxMsgs_evicts_old_messages()
|
|
{
|
|
// Go: limits retention with MaxMsgs
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamConfigAsync(new StreamConfig
|
|
{
|
|
Name = "LIM",
|
|
Subjects = ["lim.>"],
|
|
MaxMsgs = 5,
|
|
});
|
|
|
|
for (int i = 0; i < 10; i++)
|
|
await fx.PublishAndGetAckAsync("lim.test", $"msg{i}");
|
|
|
|
var state = await fx.GetStreamStateAsync("LIM");
|
|
state.Messages.ShouldBe(5UL);
|
|
}
|
|
|
|
// =========================================================================
|
|
// MaxBytes enforcement
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task MaxBytes_limits_stream_size()
|
|
{
|
|
// Go: max_bytes enforcement in various tests
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamConfigAsync(new StreamConfig
|
|
{
|
|
Name = "MB",
|
|
Subjects = ["mb.>"],
|
|
MaxBytes = 100,
|
|
});
|
|
|
|
// Keep publishing until we exceed max_bytes
|
|
for (int i = 0; i < 20; i++)
|
|
await fx.PublishAndGetAckAsync("mb.test", $"data-{i}");
|
|
|
|
var state = await fx.GetStreamStateAsync("MB");
|
|
state.Bytes.ShouldBeLessThanOrEqualTo(100UL + 100); // Allow some overhead
|
|
}
|
|
|
|
// =========================================================================
|
|
// MaxMsgsPer enforcement per subject
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task MaxMsgsPer_limits_per_subject()
|
|
{
|
|
// Go: MaxMsgsPer subject tests
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamConfigAsync(new StreamConfig
|
|
{
|
|
Name = "MPS",
|
|
Subjects = ["mps.>"],
|
|
MaxMsgsPer = 2,
|
|
});
|
|
|
|
await fx.PublishAndGetAckAsync("mps.a", "a1");
|
|
await fx.PublishAndGetAckAsync("mps.a", "a2");
|
|
await fx.PublishAndGetAckAsync("mps.a", "a3"); // should evict a1
|
|
await fx.PublishAndGetAckAsync("mps.b", "b1");
|
|
|
|
var state = await fx.GetStreamStateAsync("MPS");
|
|
// Should have at most 2 for "mps.a" + 1 for "mps.b" = 3
|
|
state.Messages.ShouldBe(3UL);
|
|
}
|
|
|
|
// =========================================================================
|
|
// Ack All semantics
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task AckAll_acknowledges_up_to_sequence()
|
|
{
|
|
// Go: TestJetStreamAckAllRedelivery jetstream_test.go:1921
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("AA", "aa.>");
|
|
await fx.CreateConsumerAsync("AA", "ACKALL", "aa.>",
|
|
ackPolicy: AckPolicy.All);
|
|
|
|
await fx.PublishAndGetAckAsync("aa.1", "msg1");
|
|
await fx.PublishAndGetAckAsync("aa.2", "msg2");
|
|
await fx.PublishAndGetAckAsync("aa.3", "msg3");
|
|
|
|
var batch = await fx.FetchAsync("AA", "ACKALL", 5);
|
|
batch.Messages.Count.ShouldBe(3);
|
|
|
|
// AckAll up to sequence 2
|
|
await fx.AckAllAsync("AA", "ACKALL", 2);
|
|
var pending = await fx.GetPendingCountAsync("AA", "ACKALL");
|
|
pending.ShouldBeLessThanOrEqualTo(1);
|
|
}
|
|
|
|
// =========================================================================
|
|
// Consumer with DeliverPolicy.Last
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task Consumer_deliver_last()
|
|
{
|
|
// Go: deliver last policy tests
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("DL", "dl.>");
|
|
|
|
await fx.PublishAndGetAckAsync("dl.test", "first");
|
|
await fx.PublishAndGetAckAsync("dl.test", "second");
|
|
await fx.PublishAndGetAckAsync("dl.test", "third");
|
|
|
|
await fx.CreateConsumerAsync("DL", "LAST", "dl.>",
|
|
deliverPolicy: DeliverPolicy.Last);
|
|
|
|
var batch = await fx.FetchAsync("DL", "LAST", 10);
|
|
batch.Messages.ShouldNotBeEmpty();
|
|
// With deliver last, we should get the latest message(s)
|
|
batch.Messages[0].Sequence.ShouldBeGreaterThanOrEqualTo(3UL);
|
|
}
|
|
|
|
// =========================================================================
|
|
// Consumer with DeliverPolicy.New
|
|
// =========================================================================
|
|
|
|
[Fact(Skip = "DeliverPolicy.New initial sequence resolved lazily at fetch time, not at consumer creation — sees post-fetch state")]
|
|
public async Task Consumer_deliver_new_only_gets_new_messages()
|
|
{
|
|
// Go: deliver new policy tests
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("DN", "dn.>");
|
|
|
|
// Pre-existing messages
|
|
await fx.PublishAndGetAckAsync("dn.test", "old1");
|
|
await fx.PublishAndGetAckAsync("dn.test", "old2");
|
|
|
|
// Create consumer with deliver new
|
|
await fx.CreateConsumerAsync("DN", "NEW", "dn.>",
|
|
deliverPolicy: DeliverPolicy.New);
|
|
|
|
// Publish new message after consumer creation
|
|
await fx.PublishAndGetAckAsync("dn.test", "new1");
|
|
|
|
var batch = await fx.FetchAsync("DN", "NEW", 10);
|
|
batch.Messages.ShouldNotBeEmpty();
|
|
// Should only get messages published after consumer creation
|
|
batch.Messages.All(m => m.Sequence >= 3UL).ShouldBeTrue();
|
|
}
|
|
|
|
// =========================================================================
|
|
// Stream update changes subjects
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task Stream_update_changes_subjects()
|
|
{
|
|
// Go: TestJetStreamUpdateStream jetstream_test.go:6409
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("UPD", "upd.old.*");
|
|
|
|
// Update subjects
|
|
var update = await fx.RequestLocalAsync(
|
|
"$JS.API.STREAM.UPDATE.UPD",
|
|
"""{"name":"UPD","subjects":["upd.new.*"]}""");
|
|
update.Error.ShouldBeNull();
|
|
|
|
// Old subject should no longer match
|
|
var ack = await fx.PublishAndGetAckAsync("upd.new.test", "msg1");
|
|
ack.ErrorCode.ShouldBeNull();
|
|
}
|
|
|
|
// =========================================================================
|
|
// Stream overlapping subjects rejected
|
|
// =========================================================================
|
|
|
|
[Fact(Skip = "Overlapping subject validation across streams not yet implemented in .NET server")]
|
|
public async Task Stream_overlapping_subjects_rejected()
|
|
{
|
|
// Go: TestJetStreamAddStreamOverlappingSubjects jetstream_test.go:615
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("S1", "foo.>");
|
|
|
|
// Creating another stream with overlapping subjects should fail
|
|
var response = await fx.CreateStreamAsync("S2", ["foo.bar"]);
|
|
response.Error.ShouldNotBeNull();
|
|
}
|
|
|
|
// =========================================================================
|
|
// Multiple streams with disjoint subjects
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task Multiple_streams_disjoint_subjects()
|
|
{
|
|
// Go: multiple streams with non-overlapping subjects
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamAsync("S1", "orders.>");
|
|
|
|
var r2 = await fx.CreateStreamAsync("S2", ["payments.>"]);
|
|
r2.Error.ShouldBeNull();
|
|
|
|
var ack1 = await fx.PublishAndGetAckAsync("orders.new", "o1");
|
|
ack1.Stream.ShouldBe("S1");
|
|
|
|
var ack2 = await fx.PublishAndGetAckAsync("payments.new", "p1");
|
|
ack2.Stream.ShouldBe("S2");
|
|
}
|
|
|
|
// =========================================================================
|
|
// Stream sealed prevents new messages
|
|
// =========================================================================
|
|
|
|
[Fact(Skip = "Sealed stream publish rejection not yet implemented in .NET server Capture path")]
|
|
public async Task Stream_sealed_prevents_publishing()
|
|
{
|
|
// Go: sealed stream tests
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamConfigAsync(new StreamConfig
|
|
{
|
|
Name = "SEALED",
|
|
Subjects = ["sealed.>"],
|
|
Sealed = true,
|
|
});
|
|
|
|
var ack = await fx.PublishAndGetAckAsync("sealed.test", "msg", expectError: true);
|
|
ack.ErrorCode.ShouldNotBeNull();
|
|
}
|
|
|
|
// =========================================================================
|
|
// Storage type selection
|
|
// =========================================================================
|
|
|
|
[Fact]
|
|
public async Task Stream_memory_storage_type()
|
|
{
|
|
// Go: Storage type tests
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamConfigAsync(new StreamConfig
|
|
{
|
|
Name = "MEM",
|
|
Subjects = ["mem.>"],
|
|
Storage = StorageType.Memory,
|
|
});
|
|
|
|
var backendType = await fx.GetStreamBackendTypeAsync("MEM");
|
|
backendType.ShouldBe("memory");
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Stream_file_storage_type()
|
|
{
|
|
// Go: Storage type tests
|
|
await using var fx = await JetStreamApiFixture.StartWithStreamConfigAsync(new StreamConfig
|
|
{
|
|
Name = "FILE",
|
|
Subjects = ["file.>"],
|
|
Storage = StorageType.File,
|
|
});
|
|
|
|
var backendType = await fx.GetStreamBackendTypeAsync("FILE");
|
|
backendType.ShouldBe("file");
|
|
}
|
|
}
|