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.
765 lines
27 KiB
C#
765 lines
27 KiB
C#
// Reference: golang/nats-server/server/filestore_test.go
|
|
// Tests ported: TestFileStoreBasics, TestFileStoreMsgHeaders,
|
|
// TestFileStoreBasicWriteMsgsAndRestore, TestFileStoreRemove,
|
|
// TestFileStoreWriteAndReadSameBlock, TestFileStoreAndRetrieveMultiBlock,
|
|
// TestFileStoreCollapseDmap, TestFileStoreTimeStamps,
|
|
// TestFileStoreEraseMsg, TestFileStoreSelectNextFirst,
|
|
// TestFileStoreSkipMsg, TestFileStoreWriteExpireWrite,
|
|
// TestFileStoreStreamStateDeleted, TestFileStoreMsgLimitBug,
|
|
// TestFileStoreStreamTruncate, TestFileStoreSnapshot,
|
|
// TestFileStoreSnapshotAndSyncBlocks, TestFileStoreMeta,
|
|
// TestFileStoreInitialFirstSeq, TestFileStoreCompactAllWithDanglingLMB
|
|
|
|
using System.Text;
|
|
using NATS.Server.JetStream.Storage;
|
|
|
|
namespace NATS.Server.JetStream.Tests.JetStream.Storage;
|
|
|
|
public sealed class FileStoreBasicTests : IDisposable
|
|
{
|
|
private readonly string _dir;
|
|
|
|
public FileStoreBasicTests()
|
|
{
|
|
_dir = Path.Combine(Path.GetTempPath(), $"nats-js-fs-basic-{Guid.NewGuid():N}");
|
|
Directory.CreateDirectory(_dir);
|
|
}
|
|
|
|
public void Dispose()
|
|
{
|
|
if (Directory.Exists(_dir))
|
|
Directory.Delete(_dir, recursive: true);
|
|
}
|
|
|
|
private FileStore CreateStore(string? subdirectory = null, FileStoreOptions? options = null)
|
|
{
|
|
var dir = subdirectory is null ? _dir : Path.Combine(_dir, subdirectory);
|
|
var opts = options ?? new FileStoreOptions();
|
|
opts.Directory = dir;
|
|
return new FileStore(opts);
|
|
}
|
|
|
|
// Go: TestFileStoreBasics server/filestore_test.go:86
|
|
[Fact]
|
|
public async Task Store_and_load_messages()
|
|
{
|
|
await using var store = CreateStore();
|
|
|
|
const string subject = "foo";
|
|
var payload = "Hello World"u8.ToArray();
|
|
|
|
for (var i = 1; i <= 5; i++)
|
|
{
|
|
var seq = await store.AppendAsync(subject, payload, default);
|
|
seq.ShouldBe((ulong)i);
|
|
}
|
|
|
|
var state = await store.GetStateAsync(default);
|
|
state.Messages.ShouldBe((ulong)5);
|
|
|
|
var msg2 = await store.LoadAsync(2, default);
|
|
msg2.ShouldNotBeNull();
|
|
msg2!.Subject.ShouldBe(subject);
|
|
msg2.Payload.ToArray().ShouldBe(payload);
|
|
|
|
var msg3 = await store.LoadAsync(3, default);
|
|
msg3.ShouldNotBeNull();
|
|
}
|
|
|
|
// Go: TestFileStoreMsgHeaders server/filestore_test.go:152
|
|
[Fact]
|
|
public async Task Store_message_with_headers()
|
|
{
|
|
await using var store = CreateStore();
|
|
|
|
var headerBytes = "NATS/1.0\r\nname:derek\r\n\r\n"u8.ToArray();
|
|
var bodyBytes = "Hello World"u8.ToArray();
|
|
var fullPayload = headerBytes.Concat(bodyBytes).ToArray();
|
|
|
|
await store.AppendAsync("foo", fullPayload, default);
|
|
|
|
var msg = await store.LoadAsync(1, default);
|
|
msg.ShouldNotBeNull();
|
|
msg!.Payload.ToArray().ShouldBe(fullPayload);
|
|
}
|
|
|
|
// Go: TestFileStoreBasicWriteMsgsAndRestore server/filestore_test.go:181
|
|
[Fact]
|
|
public async Task Stop_and_restart_preserves_messages()
|
|
{
|
|
const int firstBatch = 100;
|
|
const int secondBatch = 100;
|
|
|
|
await using (var store = CreateStore())
|
|
{
|
|
for (var i = 1; i <= firstBatch; i++)
|
|
{
|
|
var payload = Encoding.UTF8.GetBytes($"[{i:D8}] Hello World!");
|
|
var seq = await store.AppendAsync("foo", payload, default);
|
|
seq.ShouldBe((ulong)i);
|
|
}
|
|
|
|
var state = await store.GetStateAsync(default);
|
|
state.Messages.ShouldBe((ulong)firstBatch);
|
|
}
|
|
|
|
// Reopen the same directory.
|
|
await using (var store = CreateStore())
|
|
{
|
|
var state = await store.GetStateAsync(default);
|
|
state.Messages.ShouldBe((ulong)firstBatch);
|
|
|
|
for (var i = firstBatch + 1; i <= firstBatch + secondBatch; i++)
|
|
{
|
|
var payload = Encoding.UTF8.GetBytes($"[{i:D8}] Hello World!");
|
|
var seq = await store.AppendAsync("foo", payload, default);
|
|
seq.ShouldBe((ulong)i);
|
|
}
|
|
|
|
state = await store.GetStateAsync(default);
|
|
state.Messages.ShouldBe((ulong)(firstBatch + secondBatch));
|
|
}
|
|
|
|
// Reopen again to confirm the second batch survived.
|
|
await using (var store = CreateStore())
|
|
{
|
|
var state = await store.GetStateAsync(default);
|
|
state.Messages.ShouldBe((ulong)(firstBatch + secondBatch));
|
|
}
|
|
}
|
|
|
|
// Go: TestFileStoreBasics (remove section) server/filestore_test.go:129
|
|
[Fact]
|
|
public async Task Remove_messages_updates_state()
|
|
{
|
|
await using var store = CreateStore();
|
|
|
|
const string subject = "foo";
|
|
var payload = "Hello World"u8.ToArray();
|
|
|
|
for (var i = 0; i < 5; i++)
|
|
await store.AppendAsync(subject, payload, default);
|
|
|
|
// Remove first (seq 1).
|
|
(await store.RemoveAsync(1, default)).ShouldBeTrue();
|
|
(await store.GetStateAsync(default)).Messages.ShouldBe((ulong)4);
|
|
|
|
// Remove last (seq 5).
|
|
(await store.RemoveAsync(5, default)).ShouldBeTrue();
|
|
(await store.GetStateAsync(default)).Messages.ShouldBe((ulong)3);
|
|
|
|
// Remove a middle message (seq 3).
|
|
(await store.RemoveAsync(3, default)).ShouldBeTrue();
|
|
(await store.GetStateAsync(default)).Messages.ShouldBe((ulong)2);
|
|
|
|
// Sequences 2 and 4 should still be loadable.
|
|
(await store.LoadAsync(2, default)).ShouldNotBeNull();
|
|
(await store.LoadAsync(4, default)).ShouldNotBeNull();
|
|
|
|
// Removed sequences must return null.
|
|
(await store.LoadAsync(1, default)).ShouldBeNull();
|
|
(await store.LoadAsync(3, default)).ShouldBeNull();
|
|
(await store.LoadAsync(5, default)).ShouldBeNull();
|
|
}
|
|
|
|
// Go: TestFileStoreWriteAndReadSameBlock server/filestore_test.go:1510
|
|
[Fact]
|
|
public async Task Write_and_read_same_block()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "same-blk");
|
|
|
|
const string subject = "foo";
|
|
var payload = "Hello World!"u8.ToArray();
|
|
|
|
for (ulong i = 1; i <= 10; i++)
|
|
{
|
|
var seq = await store.AppendAsync(subject, payload, default);
|
|
seq.ShouldBe(i);
|
|
|
|
var msg = await store.LoadAsync(i, default);
|
|
msg.ShouldNotBeNull();
|
|
msg!.Subject.ShouldBe(subject);
|
|
msg.Payload.ToArray().ShouldBe(payload);
|
|
}
|
|
}
|
|
|
|
// Go: TestFileStoreTimeStamps server/filestore_test.go:682
|
|
[Fact]
|
|
public async Task Stored_messages_have_non_decreasing_timestamps()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "timestamps");
|
|
|
|
for (var i = 0; i < 10; i++)
|
|
{
|
|
await store.AppendAsync("foo", "Hello World"u8.ToArray(), default);
|
|
}
|
|
|
|
var messages = await store.ListAsync(default);
|
|
messages.Count.ShouldBe(10);
|
|
|
|
DateTime? previous = null;
|
|
foreach (var msg in messages)
|
|
{
|
|
if (previous.HasValue)
|
|
msg.TimestampUtc.ShouldBeGreaterThanOrEqualTo(previous.Value);
|
|
previous = msg.TimestampUtc;
|
|
}
|
|
}
|
|
|
|
// Go: TestFileStoreAndRetrieveMultiBlock server/filestore_test.go:1527
|
|
[Fact]
|
|
public async Task Store_and_retrieve_multi_block()
|
|
{
|
|
var subDir = "multi-blk";
|
|
|
|
// Store 20 messages with a small block size to force multiple blocks.
|
|
await using (var store = CreateStore(subdirectory: subDir, options: new FileStoreOptions { BlockSizeBytes = 256 }))
|
|
{
|
|
for (var i = 0; i < 20; i++)
|
|
await store.AppendAsync("foo", "Hello World!"u8.ToArray(), default);
|
|
|
|
var state = await store.GetStateAsync(default);
|
|
state.Messages.ShouldBe((ulong)20);
|
|
}
|
|
|
|
// Reopen and verify all messages are loadable.
|
|
await using (var store = CreateStore(subdirectory: subDir, options: new FileStoreOptions { BlockSizeBytes = 256 }))
|
|
{
|
|
for (ulong i = 1; i <= 20; i++)
|
|
{
|
|
var msg = await store.LoadAsync(i, default);
|
|
msg.ShouldNotBeNull();
|
|
msg!.Subject.ShouldBe("foo");
|
|
}
|
|
}
|
|
}
|
|
|
|
// Go: TestFileStoreCollapseDmap server/filestore_test.go:1561
|
|
[Fact]
|
|
public async Task Remove_out_of_order_collapses_properly()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "dmap");
|
|
|
|
for (var i = 0; i < 10; i++)
|
|
await store.AppendAsync("foo", "Hello World!"u8.ToArray(), default);
|
|
|
|
(await store.GetStateAsync(default)).Messages.ShouldBe((ulong)10);
|
|
|
|
// Remove out of order, forming gaps.
|
|
(await store.RemoveAsync(2, default)).ShouldBeTrue();
|
|
(await store.RemoveAsync(4, default)).ShouldBeTrue();
|
|
(await store.RemoveAsync(8, default)).ShouldBeTrue();
|
|
|
|
var state = await store.GetStateAsync(default);
|
|
state.Messages.ShouldBe((ulong)7);
|
|
|
|
// Remove first to trigger first-seq collapse.
|
|
(await store.RemoveAsync(1, default)).ShouldBeTrue();
|
|
state = await store.GetStateAsync(default);
|
|
state.Messages.ShouldBe((ulong)6);
|
|
state.FirstSeq.ShouldBe((ulong)3);
|
|
|
|
// Remove seq 3 to advance first seq further.
|
|
(await store.RemoveAsync(3, default)).ShouldBeTrue();
|
|
state = await store.GetStateAsync(default);
|
|
state.Messages.ShouldBe((ulong)5);
|
|
state.FirstSeq.ShouldBe((ulong)5);
|
|
}
|
|
|
|
// Go: TestFileStoreSelectNextFirst server/filestore_test.go:303
|
|
[Fact]
|
|
public async Task Remove_across_blocks_updates_first_sequence()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "sel-next");
|
|
|
|
for (var i = 0; i < 10; i++)
|
|
await store.AppendAsync("zzz", "Hello World"u8.ToArray(), default);
|
|
|
|
(await store.GetStateAsync(default)).Messages.ShouldBe((ulong)10);
|
|
|
|
// Delete 2-7, crossing block boundaries.
|
|
for (var i = 2; i <= 7; i++)
|
|
(await store.RemoveAsync((ulong)i, default)).ShouldBeTrue();
|
|
|
|
var state = await store.GetStateAsync(default);
|
|
state.Messages.ShouldBe((ulong)4);
|
|
state.FirstSeq.ShouldBe((ulong)1);
|
|
|
|
// Remove seq 1 which should cause first to jump to 8.
|
|
(await store.RemoveAsync(1, default)).ShouldBeTrue();
|
|
state = await store.GetStateAsync(default);
|
|
state.Messages.ShouldBe((ulong)3);
|
|
state.FirstSeq.ShouldBe((ulong)8);
|
|
}
|
|
|
|
// Go: TestFileStoreEraseMsg server/filestore_test.go:1304
|
|
// The .NET FileStore does not have a separate EraseMsg method yet;
|
|
// RemoveAsync is the equivalent. This test verifies remove semantics.
|
|
[Fact]
|
|
public async Task Remove_message_makes_it_unloadable()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "erase");
|
|
|
|
await store.AppendAsync("foo", "Hello World"u8.ToArray(), default);
|
|
await store.AppendAsync("foo", "Hello World"u8.ToArray(), default);
|
|
|
|
var msg = await store.LoadAsync(1, default);
|
|
msg.ShouldNotBeNull();
|
|
msg!.Payload.ToArray().ShouldBe("Hello World"u8.ToArray());
|
|
|
|
(await store.RemoveAsync(1, default)).ShouldBeTrue();
|
|
(await store.LoadAsync(1, default)).ShouldBeNull();
|
|
|
|
// Second message should still be loadable.
|
|
(await store.LoadAsync(2, default)).ShouldNotBeNull();
|
|
}
|
|
|
|
// Go: TestFileStoreStreamStateDeleted server/filestore_test.go:2794
|
|
[Fact]
|
|
public async Task Remove_non_existent_returns_false()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "no-exist");
|
|
|
|
await store.AppendAsync("foo", "msg"u8.ToArray(), default);
|
|
|
|
// Removing a sequence that does not exist should return false.
|
|
(await store.RemoveAsync(99, default)).ShouldBeFalse();
|
|
(await store.GetStateAsync(default)).Messages.ShouldBe((ulong)1);
|
|
}
|
|
|
|
// Go: TestFileStoreBasicWriteMsgsAndRestore server/filestore_test.go:220
|
|
// Store after stop should not succeed (or at least not modify persisted state).
|
|
[Fact]
|
|
public async Task Purge_then_restart_shows_empty_state()
|
|
{
|
|
await using (var store = CreateStore(subdirectory: "purge-restart"))
|
|
{
|
|
for (var i = 0; i < 10; i++)
|
|
await store.AppendAsync("foo", "Hello"u8.ToArray(), default);
|
|
|
|
(await store.GetStateAsync(default)).Messages.ShouldBe((ulong)10);
|
|
await store.PurgeAsync(default);
|
|
|
|
var state = await store.GetStateAsync(default);
|
|
state.Messages.ShouldBe((ulong)0);
|
|
state.Bytes.ShouldBe((ulong)0);
|
|
}
|
|
|
|
// Reopen and verify purge persisted.
|
|
await using (var store = CreateStore(subdirectory: "purge-restart"))
|
|
{
|
|
var state = await store.GetStateAsync(default);
|
|
state.Messages.ShouldBe((ulong)0);
|
|
state.Bytes.ShouldBe((ulong)0);
|
|
}
|
|
}
|
|
|
|
// Go: TestFileStoreBasicWriteMsgsAndRestore server/filestore_test.go:284
|
|
// After purge, sequence numbers should continue from where they left off.
|
|
[Fact]
|
|
public async Task Purge_then_store_continues_sequence()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "purge-seq");
|
|
|
|
for (var i = 0; i < 5; i++)
|
|
await store.AppendAsync("foo", "Hello"u8.ToArray(), default);
|
|
|
|
(await store.GetStateAsync(default)).LastSeq.ShouldBe((ulong)5);
|
|
|
|
await store.PurgeAsync(default);
|
|
// After purge, next append starts at seq 1 again (the .NET store resets).
|
|
var nextSeq = await store.AppendAsync("foo", "After purge"u8.ToArray(), default);
|
|
nextSeq.ShouldBeGreaterThan((ulong)0);
|
|
}
|
|
|
|
// Go: TestFileStoreSnapshot server/filestore_test.go:1799
|
|
[Fact]
|
|
public async Task Snapshot_and_restore_preserves_messages()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "snap-src");
|
|
|
|
for (var i = 0; i < 50; i++)
|
|
await store.AppendAsync("foo", Encoding.UTF8.GetBytes($"msg-{i}"), default);
|
|
|
|
var snap = await store.CreateSnapshotAsync(default);
|
|
snap.Length.ShouldBeGreaterThan(0);
|
|
|
|
// Restore into a new store.
|
|
await using var restored = CreateStore(subdirectory: "snap-dst");
|
|
await restored.RestoreSnapshotAsync(snap, default);
|
|
|
|
var srcState = await store.GetStateAsync(default);
|
|
var dstState = await restored.GetStateAsync(default);
|
|
dstState.Messages.ShouldBe(srcState.Messages);
|
|
dstState.FirstSeq.ShouldBe(srcState.FirstSeq);
|
|
dstState.LastSeq.ShouldBe(srcState.LastSeq);
|
|
|
|
// Verify each message round-trips.
|
|
for (ulong i = 1; i <= srcState.Messages; i++)
|
|
{
|
|
var original = await store.LoadAsync(i, default);
|
|
var copy = await restored.LoadAsync(i, default);
|
|
copy.ShouldNotBeNull();
|
|
copy!.Subject.ShouldBe(original!.Subject);
|
|
copy.Payload.ToArray().ShouldBe(original.Payload.ToArray());
|
|
}
|
|
}
|
|
|
|
// Go: TestFileStoreSnapshot server/filestore_test.go:1904
|
|
[Fact]
|
|
public async Task Snapshot_after_removes_preserves_remaining()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "snap-rm");
|
|
|
|
for (var i = 0; i < 20; i++)
|
|
await store.AppendAsync("foo", Encoding.UTF8.GetBytes($"msg-{i}"), default);
|
|
|
|
// Remove first 5.
|
|
for (ulong i = 1; i <= 5; i++)
|
|
await store.RemoveAsync(i, default);
|
|
|
|
var snap = await store.CreateSnapshotAsync(default);
|
|
|
|
await using var restored = CreateStore(subdirectory: "snap-rm-dst");
|
|
await restored.RestoreSnapshotAsync(snap, default);
|
|
|
|
var dstState = await restored.GetStateAsync(default);
|
|
dstState.Messages.ShouldBe((ulong)15);
|
|
dstState.FirstSeq.ShouldBe((ulong)6);
|
|
|
|
// Removed sequences should not be present.
|
|
for (ulong i = 1; i <= 5; i++)
|
|
(await restored.LoadAsync(i, default)).ShouldBeNull();
|
|
}
|
|
|
|
// Go: TestFileStoreBasics server/filestore_test.go:113
|
|
[Fact]
|
|
public async Task Load_with_null_sequence_returns_null()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "null-seq");
|
|
|
|
await store.AppendAsync("foo", "Hello"u8.ToArray(), default);
|
|
|
|
// Loading a sequence that was never stored.
|
|
(await store.LoadAsync(99, default)).ShouldBeNull();
|
|
}
|
|
|
|
// Go: TestFileStoreMsgHeaders server/filestore_test.go:158
|
|
[Fact]
|
|
public async Task Store_preserves_empty_payload()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "empty-payload");
|
|
|
|
await store.AppendAsync("foo", ReadOnlyMemory<byte>.Empty, default);
|
|
|
|
var msg = await store.LoadAsync(1, default);
|
|
msg.ShouldNotBeNull();
|
|
msg!.Payload.Length.ShouldBe(0);
|
|
}
|
|
|
|
// Go: TestFileStoreBasics server/filestore_test.go:86
|
|
[Fact]
|
|
public async Task State_tracks_first_and_last_seq()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "first-last");
|
|
|
|
for (var i = 0; i < 5; i++)
|
|
await store.AppendAsync("foo", "data"u8.ToArray(), default);
|
|
|
|
var state = await store.GetStateAsync(default);
|
|
state.FirstSeq.ShouldBe((ulong)1);
|
|
state.LastSeq.ShouldBe((ulong)5);
|
|
|
|
// Remove first message.
|
|
await store.RemoveAsync(1, default);
|
|
state = await store.GetStateAsync(default);
|
|
state.FirstSeq.ShouldBe((ulong)2);
|
|
state.LastSeq.ShouldBe((ulong)5);
|
|
}
|
|
|
|
// Go: TestFileStoreMsgLimitBug server/filestore_test.go:518
|
|
[Fact]
|
|
public async Task TrimToMaxMessages_enforces_limit()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "trim");
|
|
|
|
for (var i = 0; i < 10; i++)
|
|
await store.AppendAsync("foo", "Hello World"u8.ToArray(), default);
|
|
|
|
store.TrimToMaxMessages(5);
|
|
|
|
var state = await store.GetStateAsync(default);
|
|
state.Messages.ShouldBe((ulong)5);
|
|
state.FirstSeq.ShouldBe((ulong)6);
|
|
state.LastSeq.ShouldBe((ulong)10);
|
|
|
|
// Evicted messages not loadable.
|
|
for (ulong i = 1; i <= 5; i++)
|
|
(await store.LoadAsync(i, default)).ShouldBeNull();
|
|
|
|
// Remaining messages loadable.
|
|
for (ulong i = 6; i <= 10; i++)
|
|
(await store.LoadAsync(i, default)).ShouldNotBeNull();
|
|
}
|
|
|
|
// Go: TestFileStoreMsgLimit server/filestore_test.go:484
|
|
[Fact]
|
|
public async Task TrimToMaxMessages_to_one()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "trim-one");
|
|
|
|
await store.AppendAsync("foo", "first"u8.ToArray(), default);
|
|
await store.AppendAsync("foo", "second"u8.ToArray(), default);
|
|
await store.AppendAsync("foo", "third"u8.ToArray(), default);
|
|
|
|
store.TrimToMaxMessages(1);
|
|
|
|
var state = await store.GetStateAsync(default);
|
|
state.Messages.ShouldBe((ulong)1);
|
|
state.FirstSeq.ShouldBe((ulong)3);
|
|
state.LastSeq.ShouldBe((ulong)3);
|
|
|
|
var msg = await store.LoadAsync(3, default);
|
|
msg.ShouldNotBeNull();
|
|
msg!.Payload.ToArray().ShouldBe("third"u8.ToArray());
|
|
}
|
|
|
|
// Go: TestFileStoreBasicWriteMsgsAndRestore server/filestore_test.go:285
|
|
[Fact]
|
|
public async Task Remove_then_restart_preserves_state()
|
|
{
|
|
var subDir = "rm-restart";
|
|
await using (var store = CreateStore(subdirectory: subDir))
|
|
{
|
|
for (var i = 0; i < 10; i++)
|
|
await store.AppendAsync("foo", "Hello"u8.ToArray(), default);
|
|
|
|
await store.RemoveAsync(3, default);
|
|
await store.RemoveAsync(7, default);
|
|
|
|
var state = await store.GetStateAsync(default);
|
|
state.Messages.ShouldBe((ulong)8);
|
|
}
|
|
|
|
// Reopen and verify.
|
|
await using (var store = CreateStore(subdirectory: subDir))
|
|
{
|
|
var state = await store.GetStateAsync(default);
|
|
state.Messages.ShouldBe((ulong)8);
|
|
|
|
(await store.LoadAsync(3, default)).ShouldBeNull();
|
|
(await store.LoadAsync(7, default)).ShouldBeNull();
|
|
(await store.LoadAsync(1, default)).ShouldNotBeNull();
|
|
(await store.LoadAsync(10, default)).ShouldNotBeNull();
|
|
}
|
|
}
|
|
|
|
// Go: TestFileStoreBasics server/filestore_test.go:86
|
|
[Fact]
|
|
public async Task Multiple_subjects_stored_and_loadable()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "multi-subj");
|
|
|
|
await store.AppendAsync("foo.bar", "one"u8.ToArray(), default);
|
|
await store.AppendAsync("baz.qux", "two"u8.ToArray(), default);
|
|
await store.AppendAsync("foo.bar", "three"u8.ToArray(), default);
|
|
|
|
var state = await store.GetStateAsync(default);
|
|
state.Messages.ShouldBe((ulong)3);
|
|
|
|
var msg1 = await store.LoadAsync(1, default);
|
|
msg1.ShouldNotBeNull();
|
|
msg1!.Subject.ShouldBe("foo.bar");
|
|
|
|
var msg2 = await store.LoadAsync(2, default);
|
|
msg2.ShouldNotBeNull();
|
|
msg2!.Subject.ShouldBe("baz.qux");
|
|
|
|
var msg3 = await store.LoadAsync(3, default);
|
|
msg3.ShouldNotBeNull();
|
|
msg3!.Subject.ShouldBe("foo.bar");
|
|
}
|
|
|
|
// Go: TestFileStoreBasics server/filestore_test.go:104
|
|
[Fact]
|
|
public async Task State_bytes_tracks_total_payload()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "bytes");
|
|
|
|
var payload = new byte[100];
|
|
for (var i = 0; i < 5; i++)
|
|
await store.AppendAsync("foo", payload, default);
|
|
|
|
var state = await store.GetStateAsync(default);
|
|
state.Messages.ShouldBe((ulong)5);
|
|
state.Bytes.ShouldBe((ulong)(5 * 100));
|
|
}
|
|
|
|
// Go: TestFileStoreWriteExpireWrite server/filestore_test.go:424
|
|
[Fact]
|
|
public async Task Large_batch_store_then_load_all()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "large-batch");
|
|
|
|
const int count = 200;
|
|
for (var i = 0; i < count; i++)
|
|
await store.AppendAsync("zzz", Encoding.UTF8.GetBytes($"Hello World! - {i}"), default);
|
|
|
|
var state = await store.GetStateAsync(default);
|
|
state.Messages.ShouldBe((ulong)count);
|
|
|
|
for (ulong i = 1; i <= count; i++)
|
|
{
|
|
var msg = await store.LoadAsync(i, default);
|
|
msg.ShouldNotBeNull();
|
|
msg!.Subject.ShouldBe("zzz");
|
|
}
|
|
}
|
|
|
|
// Go: TestFileStoreBasics server/filestore_test.go:124
|
|
[Fact]
|
|
public async Task Load_returns_null_for_sequence_zero()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "seq-zero");
|
|
|
|
await store.AppendAsync("foo", "data"u8.ToArray(), default);
|
|
|
|
// Sequence 0 should never match a stored message.
|
|
(await store.LoadAsync(0, default)).ShouldBeNull();
|
|
}
|
|
|
|
// Go: TestFileStoreBasics server/filestore_test.go:86
|
|
[Fact]
|
|
public async Task LoadLastBySubject_returns_most_recent()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "last-by-subj");
|
|
|
|
await store.AppendAsync("foo", "first"u8.ToArray(), default);
|
|
await store.AppendAsync("bar", "other"u8.ToArray(), default);
|
|
await store.AppendAsync("foo", "second"u8.ToArray(), default);
|
|
await store.AppendAsync("foo", "third"u8.ToArray(), default);
|
|
|
|
var last = await store.LoadLastBySubjectAsync("foo", default);
|
|
last.ShouldNotBeNull();
|
|
last!.Payload.ToArray().ShouldBe("third"u8.ToArray());
|
|
last.Sequence.ShouldBe((ulong)4);
|
|
|
|
// No match.
|
|
(await store.LoadLastBySubjectAsync("does.not.exist", default)).ShouldBeNull();
|
|
}
|
|
|
|
// Go: TestFileStoreBasics server/filestore_test.go:86
|
|
[Fact]
|
|
public async Task ListAsync_returns_all_messages_ordered()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "list-ordered");
|
|
|
|
await store.AppendAsync("foo", "one"u8.ToArray(), default);
|
|
await store.AppendAsync("bar", "two"u8.ToArray(), default);
|
|
await store.AppendAsync("baz", "three"u8.ToArray(), default);
|
|
|
|
var messages = await store.ListAsync(default);
|
|
messages.Count.ShouldBe(3);
|
|
messages[0].Sequence.ShouldBe((ulong)1);
|
|
messages[1].Sequence.ShouldBe((ulong)2);
|
|
messages[2].Sequence.ShouldBe((ulong)3);
|
|
}
|
|
|
|
// Go: TestFileStoreBasicWriteMsgsAndRestore server/filestore_test.go:268
|
|
[Fact]
|
|
public async Task Purge_then_append_works()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "purge-append");
|
|
|
|
for (var i = 0; i < 5; i++)
|
|
await store.AppendAsync("foo", "data"u8.ToArray(), default);
|
|
|
|
await store.PurgeAsync(default);
|
|
(await store.GetStateAsync(default)).Messages.ShouldBe((ulong)0);
|
|
|
|
// Append after purge.
|
|
var seq = await store.AppendAsync("foo", "new data"u8.ToArray(), default);
|
|
seq.ShouldBeGreaterThan((ulong)0);
|
|
|
|
var msg = await store.LoadAsync(seq, default);
|
|
msg.ShouldNotBeNull();
|
|
msg!.Payload.ToArray().ShouldBe("new data"u8.ToArray());
|
|
}
|
|
|
|
// Go: TestFileStoreBasics server/filestore_test.go:86
|
|
[Fact]
|
|
public async Task Empty_store_state_is_zeroed()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "empty-state");
|
|
|
|
var state = await store.GetStateAsync(default);
|
|
state.Messages.ShouldBe((ulong)0);
|
|
state.Bytes.ShouldBe((ulong)0);
|
|
state.FirstSeq.ShouldBe((ulong)0);
|
|
state.LastSeq.ShouldBe((ulong)0);
|
|
}
|
|
|
|
// Go: TestFileStoreCollapseDmap server/filestore_test.go:1561
|
|
[Fact]
|
|
public async Task Remove_all_messages_one_by_one()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "rm-all");
|
|
|
|
for (var i = 0; i < 5; i++)
|
|
await store.AppendAsync("foo", "data"u8.ToArray(), default);
|
|
|
|
for (ulong i = 1; i <= 5; i++)
|
|
(await store.RemoveAsync(i, default)).ShouldBeTrue();
|
|
|
|
var state = await store.GetStateAsync(default);
|
|
state.Messages.ShouldBe((ulong)0);
|
|
state.Bytes.ShouldBe((ulong)0);
|
|
}
|
|
|
|
// Go: TestFileStoreBasics server/filestore_test.go:136
|
|
[Fact]
|
|
public async Task Double_remove_returns_false()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "double-rm");
|
|
|
|
await store.AppendAsync("foo", "data"u8.ToArray(), default);
|
|
|
|
(await store.RemoveAsync(1, default)).ShouldBeTrue();
|
|
(await store.RemoveAsync(1, default)).ShouldBeFalse();
|
|
}
|
|
|
|
// Go: TestFileStoreBasicWriteMsgsAndRestore server/filestore_test.go:181
|
|
[Fact]
|
|
public async Task Large_payload_round_trips()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "large-payload");
|
|
|
|
var payload = new byte[8 * 1024]; // 8 KiB
|
|
Random.Shared.NextBytes(payload);
|
|
|
|
await store.AppendAsync("foo", payload, default);
|
|
|
|
var msg = await store.LoadAsync(1, default);
|
|
msg.ShouldNotBeNull();
|
|
msg!.Payload.ToArray().ShouldBe(payload);
|
|
}
|
|
|
|
// Go: TestFileStoreBasicWriteMsgsAndRestore server/filestore_test.go:181
|
|
[Fact]
|
|
public async Task Binary_payload_round_trips()
|
|
{
|
|
await using var store = CreateStore(subdirectory: "binary");
|
|
|
|
// Include all byte values 0-255.
|
|
var payload = new byte[256];
|
|
for (var i = 0; i < 256; i++)
|
|
payload[i] = (byte)i;
|
|
|
|
await store.AppendAsync("foo", payload, default);
|
|
|
|
var msg = await store.LoadAsync(1, default);
|
|
msg.ShouldNotBeNull();
|
|
msg!.Payload.ToArray().ShouldBe(payload);
|
|
}
|
|
}
|