// Reference: golang/nats-server/server/memstore_test.go // Tests ported in this file: // TestMemStoreCompact → Compact_RemovesMessagesBeforeSeq // TestMemStoreStreamStateDeleted → StreamStateDeleted_TracksDmapCorrectly // TestMemStoreStreamTruncate → Truncate_RemovesMessagesAfterSeq // TestMemStoreUpdateMaxMsgsPerSubject → UpdateMaxMsgsPerSubject_EnforcesNewLimit // TestMemStoreStreamCompactMultiBlockSubjectInfo → Compact_AdjustsSubjectCount // TestMemStoreSubjectsTotals → SubjectsTotals_MatchesStoredCounts // TestMemStoreNumPending → NumPending_MatchesFilteredCount // TestMemStoreMultiLastSeqs → MultiLastSeqs_ReturnsLastPerSubject // TestMemStoreSubjectForSeq → SubjectForSeq_ReturnsCorrectSubject // TestMemStoreSubjectDeleteMarkers → SubjectDeleteMarkers_TtlExpiry (skipped: needs pmsgcb) // TestMemStoreAllLastSeqs → AllLastSeqs_ReturnsLastPerSubjectSorted // TestMemStoreGetSeqFromTimeWithLastDeleted → GetSeqFromTime_WithLastDeleted // TestMemStoreSkipMsgs → SkipMsgs_ReservesSequences // TestMemStoreDeleteBlocks → DeleteBlocks_DmapSizeMatchesNumDeleted // TestMemStoreMessageTTL → MessageTTL_ExpiresAfterDelay // TestMemStoreUpdateConfigTTLState → UpdateConfig_TtlStateInitializedAndDestroyed // TestMemStoreNextWildcardMatch → NextWildcardMatch_BoundsAreCorrect // TestMemStoreNextLiteralMatch → NextLiteralMatch_BoundsAreCorrect // TestMemStoreInitialFirstSeq → InitialFirstSeq_StartAtConfiguredSeq // TestMemStoreStreamTruncateReset → TruncateReset_ClearsEverything // TestMemStorePurgeExWithSubject → PurgeEx_WithSubject_PurgesAll // TestMemStorePurgeExWithDeletedMsgs → PurgeEx_WithDeletedMsgs_CorrectFirstSeq // TestMemStoreDeleteAllFirstSequenceCheck → DeleteAll_FirstSeqIsLastPlusOne // TestMemStoreNumPendingBug → NumPending_Bug_CorrectCount // TestMemStorePurgeLeaksDmap → Purge_ClearsDmap // TestMemStoreMultiLastSeqsMaxAllowed → MultiLastSeqs_MaxAllowed_ThrowsWhenExceeded using NATS.Server.JetStream.Models; using NATS.Server.JetStream.Storage; namespace NATS.Server.JetStream.Tests.JetStream.Storage; /// /// Go MemStore parity tests. Each test mirrors a specific Go test from /// golang/nats-server/server/memstore_test.go to verify behaviour parity. /// public sealed class MemStoreGoParityTests { // Helper: cast to IStreamStore for sync methods private static IStreamStore Sync(MemStore ms) => ms; // ------------------------------------------------------------------------- // Compact // ------------------------------------------------------------------------- // Go: TestMemStoreCompact server/memstore_test.go:259 [Fact] public void Compact_RemovesMessagesBeforeSeq() { var ms = new MemStore(); var s = Sync(ms); for (var i = 0; i < 10; i++) s.StoreMsg("foo", null, "Hello World"u8.ToArray(), 0); s.State().Msgs.ShouldBe(10UL); var n = s.Compact(6); n.ShouldBe(5UL); var state = s.State(); state.Msgs.ShouldBe(5UL); state.FirstSeq.ShouldBe(6UL); // Compact past the end resets first seq n = s.Compact(100); n.ShouldBe(5UL); s.State().FirstSeq.ShouldBe(100UL); } // ------------------------------------------------------------------------- // StreamStateDeleted // ------------------------------------------------------------------------- // Go: TestMemStoreStreamStateDeleted server/memstore_test.go:342 [Fact] public void StreamStateDeleted_TracksDmapCorrectly() { var ms = new MemStore(); var s = Sync(ms); const ulong toStore = 10; for (ulong i = 1; i <= toStore; i++) s.StoreMsg("foo", null, new byte[8], 0); s.State().Deleted.ShouldBeNull(); // Delete even sequences 2,4,6,8 var expectedDeleted = new List(); for (ulong seq = 2; seq < toStore; seq += 2) { s.RemoveMsg(seq); expectedDeleted.Add(seq); } var state = s.State(); state.Deleted.ShouldNotBeNull(); state.Deleted!.ShouldBe(expectedDeleted.ToArray()); // Delete 1 and 3 to fill first gap — deleted should shift forward s.RemoveMsg(1); s.RemoveMsg(3); expectedDeleted = expectedDeleted.Skip(2).ToList(); // remove 2 and 4 from start state = s.State(); state.Deleted!.ShouldBe(expectedDeleted.ToArray()); state.FirstSeq.ShouldBe(5UL); s.Purge(); s.State().Deleted.ShouldBeNull(); } // ------------------------------------------------------------------------- // Truncate // ------------------------------------------------------------------------- // Go: TestMemStoreStreamTruncate server/memstore_test.go:385 [Fact] public void Truncate_RemovesMessagesAfterSeq() { var ms = new MemStore(); var s = Sync(ms); const ulong tseq = 50; const ulong toStore = 100; for (ulong i = 1; i < tseq; i++) s.StoreMsg("foo", null, "ok"u8.ToArray(), 0); for (var i = tseq; i <= toStore; i++) s.StoreMsg("bar", null, "ok"u8.ToArray(), 0); s.State().Msgs.ShouldBe(toStore); s.Truncate(tseq); s.State().Msgs.ShouldBe(tseq); // Truncate with some interior deletes s.RemoveMsg(10); s.RemoveMsg(20); s.RemoveMsg(30); s.RemoveMsg(40); s.Truncate(25); var state = s.State(); // 25 seqs remaining, minus 2 deleted (10, 20) = 23 messages state.Msgs.ShouldBe(tseq - 2 - (tseq - 25)); state.NumSubjects.ShouldBe(1); // only "foo" left state.Deleted!.ShouldBe([10UL, 20UL]); } // ------------------------------------------------------------------------- // TruncateReset // ------------------------------------------------------------------------- // Go: TestMemStoreStreamTruncateReset server/memstore_test.go:490 [Fact] public void TruncateReset_ClearsEverything() { var ms = new MemStore(); var s = Sync(ms); for (var i = 0; i < 1000; i++) s.StoreMsg("foo", null, "Hello World"u8.ToArray(), 0); s.Truncate(0); var state = s.State(); state.Msgs.ShouldBe(0UL); state.Bytes.ShouldBe(0UL); state.FirstSeq.ShouldBe(0UL); state.LastSeq.ShouldBe(0UL); state.NumSubjects.ShouldBe(0); state.NumDeleted.ShouldBe(0); // Can store again after reset for (var i = 0; i < 1000; i++) s.StoreMsg("foo", null, "Hello World"u8.ToArray(), 0); state = s.State(); state.Msgs.ShouldBe(1000UL); state.FirstSeq.ShouldBe(1UL); state.LastSeq.ShouldBe(1000UL); state.NumSubjects.ShouldBe(1); state.NumDeleted.ShouldBe(0); } // ------------------------------------------------------------------------- // UpdateMaxMsgsPerSubject // ------------------------------------------------------------------------- // Go: TestMemStoreUpdateMaxMsgsPerSubject server/memstore_test.go:452 [Fact] public void UpdateMaxMsgsPerSubject_EnforcesNewLimit() { var cfg = new StreamConfig { Name = "TEST", Storage = StorageType.Memory, Subjects = ["foo"], MaxMsgsPer = 10, }; var ms = new MemStore(cfg); var s = Sync(ms); // Increase limit — should allow more cfg.MaxMsgsPer = 50; s.UpdateConfig(cfg); const int numStored = 22; for (var i = 0; i < numStored; i++) s.StoreMsg("foo", null, [], 0); var ss = s.SubjectsState("foo")["foo"]; ss.Msgs.ShouldBe((ulong)numStored); // Shrink limit — should truncate stored cfg.MaxMsgsPer = 10; s.UpdateConfig(cfg); ss = s.SubjectsState("foo")["foo"]; ss.Msgs.ShouldBe(10UL); } // ------------------------------------------------------------------------- // CompactMultiBlockSubjectInfo // ------------------------------------------------------------------------- // Go: TestMemStoreStreamCompactMultiBlockSubjectInfo server/memstore_test.go:531 [Fact] public void Compact_AdjustsSubjectCount() { var ms = new MemStore(); var s = Sync(ms); for (var i = 0; i < 1000; i++) s.StoreMsg($"foo.{i}", null, "Hello World"u8.ToArray(), 0); var deleted = s.Compact(501); deleted.ShouldBe(500UL); s.State().NumSubjects.ShouldBe(500); } // ------------------------------------------------------------------------- // SubjectsTotals // ------------------------------------------------------------------------- // Go: TestMemStoreSubjectsTotals server/memstore_test.go:557 [Fact] public void SubjectsTotals_MatchesStoredCounts() { var ms = new MemStore(); var s = Sync(ms); var fmap = new Dictionary(); var bmap = new Dictionary(); var rng = new Random(42); for (var i = 0; i < 10_000; i++) { string ft; Dictionary m; if (rng.Next(2) == 0) { ft = "foo"; m = fmap; } else { ft = "bar"; m = bmap; } var dt = rng.Next(100); var subj = $"{ft}.{dt}"; m.TryGetValue(dt, out var c); m[dt] = c + 1; s.StoreMsg(subj, null, "Hello World"u8.ToArray(), 0); } // Check individual foo subjects foreach (var kv in fmap) { var subj = $"foo.{kv.Key}"; var totals = s.SubjectsTotals(subj); totals[subj].ShouldBe((ulong)kv.Value); } // Check foo.* wildcard var fooTotals = s.SubjectsTotals("foo.*"); fooTotals.Count.ShouldBe(fmap.Count); var fooExpected = (ulong)fmap.Values.Sum(n => n); fooTotals.Values.Aggregate(0UL, (a, v) => a + v).ShouldBe(fooExpected); // Check bar.* wildcard var barTotals = s.SubjectsTotals("bar.*"); barTotals.Count.ShouldBe(bmap.Count); // Check *.* var allTotals = s.SubjectsTotals("*.*"); allTotals.Count.ShouldBe(fmap.Count + bmap.Count); } // ------------------------------------------------------------------------- // NumPending // ------------------------------------------------------------------------- // Go: TestMemStoreNumPending server/memstore_test.go:637 [Fact] public void NumPending_MatchesFilteredCount() { var ms = new MemStore(); var s = Sync(ms); var tokens = new[] { "foo", "bar", "baz" }; var rng = new Random(99); string GenSubj() => $"{tokens[rng.Next(3)]}.{tokens[rng.Next(3)]}.{tokens[rng.Next(3)]}.{tokens[rng.Next(3)]}"; for (var i = 0; i < 5_000; i++) s.StoreMsg(GenSubj(), null, "Hello World"u8.ToArray(), 0); var state = s.State(); var startSeqs = new ulong[] { 0, 1, 2, 200, 444, 555, 2222, 4000 }; var checkSubs = new[] { "foo.>", "*.bar.>", "foo.bar.*.baz", "*.foo.bar.*", "foo.foo.bar.baz" }; foreach (var filter in checkSubs) { foreach (var startSeq in startSeqs) { var (total, validThrough) = s.NumPending(startSeq, filter, false); validThrough.ShouldBe(state.LastSeq); // Sanity-check: manually count matching msgs from startSeq var sseq = startSeq == 0 ? 1 : startSeq; ulong expected = 0; for (var seq = sseq; seq <= state.LastSeq; seq++) { try { var sm = s.LoadMsg(seq, null); if (SubjectMatchesFilter(sm.Subject, filter)) expected++; } catch (KeyNotFoundException) { } } total.ShouldBe(expected, $"filter={filter} start={startSeq}"); } } } // ------------------------------------------------------------------------- // MultiLastSeqs // ------------------------------------------------------------------------- // Go: TestMemStoreMultiLastSeqs server/memstore_test.go:923 [Fact] public void MultiLastSeqs_ReturnsLastPerSubject() { var ms = new MemStore(); var s = Sync(ms); var msg = "abc"u8.ToArray(); for (var i = 0; i < 33; i++) { s.StoreMsg("foo.foo", null, msg, 0); s.StoreMsg("foo.bar", null, msg, 0); s.StoreMsg("foo.baz", null, msg, 0); } for (var i = 0; i < 33; i++) { s.StoreMsg("bar.foo", null, msg, 0); s.StoreMsg("bar.bar", null, msg, 0); s.StoreMsg("bar.baz", null, msg, 0); } // Up to seq 3 s.MultiLastSeqs(["foo.*"], 3, -1).ShouldBe([1UL, 2UL, 3UL]); // All of foo.* s.MultiLastSeqs(["foo.*"], 0, -1).ShouldBe([97UL, 98UL, 99UL]); // All of bar.* s.MultiLastSeqs(["bar.*"], 0, -1).ShouldBe([196UL, 197UL, 198UL]); // bar.* at seq <= 99 — nothing s.MultiLastSeqs(["bar.*"], 99, -1).ShouldBe([]); // Explicit subjects s.MultiLastSeqs(["foo.foo", "foo.bar", "foo.baz"], 3, -1).ShouldBe([1UL, 2UL, 3UL]); s.MultiLastSeqs(["foo.foo", "foo.bar", "foo.baz"], 0, -1).ShouldBe([97UL, 98UL, 99UL]); s.MultiLastSeqs(["bar.foo", "bar.bar", "bar.baz"], 0, -1).ShouldBe([196UL, 197UL, 198UL]); s.MultiLastSeqs(["bar.foo", "bar.bar", "bar.baz"], 99, -1).ShouldBe([]); // Single filter s.MultiLastSeqs(["foo.foo"], 3, -1).ShouldBe([1UL]); // De-duplicate overlapping filters s.MultiLastSeqs(["foo.*", "foo.bar"], 3, -1).ShouldBe([1UL, 2UL, 3UL]); // All subjects s.MultiLastSeqs([">"], 0, -1).ShouldBe([97UL, 98UL, 99UL, 196UL, 197UL, 198UL]); s.MultiLastSeqs([">"], 99, -1).ShouldBe([97UL, 98UL, 99UL]); } // ------------------------------------------------------------------------- // MultiLastSeqs — maxAllowed // ------------------------------------------------------------------------- // Go: TestMemStoreMultiLastSeqsMaxAllowed server/memstore_test.go:1010 [Fact] public void MultiLastSeqs_MaxAllowed_ThrowsWhenExceeded() { var ms = new MemStore(); var s = Sync(ms); var msg = "abc"u8.ToArray(); for (var i = 1; i <= 100; i++) s.StoreMsg($"foo.{i}", null, msg, 0); Should.Throw(() => s.MultiLastSeqs(["foo.*"], 0, 10)); } // ------------------------------------------------------------------------- // SubjectForSeq // ------------------------------------------------------------------------- // Go: TestMemStoreSubjectForSeq server/memstore_test.go:1319 [Fact] public void SubjectForSeq_ReturnsCorrectSubject() { var ms = new MemStore(); var s = Sync(ms); s.StoreMsg("foo.bar", null, [], 0); // seq 0 (not found) Should.Throw(() => s.SubjectForSeq(0)); // seq 1 — should be "foo.bar" s.SubjectForSeq(1).ShouldBe("foo.bar"); // seq 2 (not yet stored) Should.Throw(() => s.SubjectForSeq(2)); } // ------------------------------------------------------------------------- // AllLastSeqs // ------------------------------------------------------------------------- // Go: TestMemStoreAllLastSeqs server/memstore_test.go:1266 [Fact] public void AllLastSeqs_ReturnsLastPerSubjectSorted() { var cfg = new StreamConfig { Name = "zzz", Subjects = ["*.*"], MaxMsgsPer = 50, Storage = StorageType.Memory, }; var ms = new MemStore(cfg); var s = Sync(ms); var subjs = new[] { "foo.foo", "foo.bar", "foo.baz", "bar.foo", "bar.bar", "bar.baz" }; var msg = "abc"u8.ToArray(); var rng = new Random(7); for (var i = 0; i < 10_000; i++) s.StoreMsg(subjs[rng.Next(subjs.Length)], null, msg, 0); // Compute expected last sequences per subject var expected = new List(); foreach (var subj in subjs) { try { var sm = s.LoadLastMsg(subj, null); expected.Add(sm.Sequence); } catch (KeyNotFoundException) { } } expected.Sort(); var seqs = s.AllLastSeqs(); seqs.ShouldBe(expected.ToArray()); } // ------------------------------------------------------------------------- // GetSeqFromTime with last deleted // ------------------------------------------------------------------------- // Go: TestMemStoreGetSeqFromTimeWithLastDeleted server/memstore_test.go:839 [Fact] public void GetSeqFromTime_WithLastDeleted() { var ms = new MemStore(); var s = Sync(ms); const int total = 1000; DateTime midTime = default; for (var i = 1; i <= total; i++) { s.StoreMsg("A", null, "OK"u8.ToArray(), 0); if (i == total / 2) { Thread.Sleep(100); midTime = DateTime.UtcNow; } } // Delete last 100 for (var seq = total - 100; seq <= total; seq++) s.RemoveMsg((ulong)seq); // Should not panic and should return correct value var found = s.GetSeqFromTime(midTime); found.ShouldBe(501UL); } // ------------------------------------------------------------------------- // SkipMsgs // ------------------------------------------------------------------------- // Go: TestMemStoreSkipMsgs server/memstore_test.go:871 [Fact] public void SkipMsgs_ReservesSequences() { var ms = new MemStore(); var s = Sync(ms); // Wrong starting sequence should fail Should.Throw(() => s.SkipMsgs(10, 100)); // Skip from seq 1 s.SkipMsgs(1, 100); var state = s.State(); state.FirstSeq.ShouldBe(101UL); state.LastSeq.ShouldBe(100UL); // Skip many more s.SkipMsgs(101, 100_000); state = s.State(); state.FirstSeq.ShouldBe(100_101UL); state.LastSeq.ShouldBe(100_100UL); // New store: store a message then skip var ms2 = new MemStore(); var s2 = Sync(ms2); s2.StoreMsg("foo", null, [], 0); s2.SkipMsgs(2, 10); state = s2.State(); state.FirstSeq.ShouldBe(1UL); state.LastSeq.ShouldBe(11UL); state.Msgs.ShouldBe(1UL); state.NumDeleted.ShouldBe(10); state.Deleted.ShouldNotBeNull(); state.Deleted!.Length.ShouldBe(10); // FastState consistency var fstate = new StreamState(); s2.FastState(ref fstate); fstate.FirstSeq.ShouldBe(1UL); fstate.LastSeq.ShouldBe(11UL); fstate.Msgs.ShouldBe(1UL); fstate.NumDeleted.ShouldBe(10); } // ------------------------------------------------------------------------- // DeleteBlocks // ------------------------------------------------------------------------- // Go: TestMemStoreDeleteBlocks server/memstore_test.go:799 [Fact] public void DeleteBlocks_DmapSizeMatchesNumDeleted() { var ms = new MemStore(); var s = Sync(ms); const int total = 10_000; for (var i = 0; i < total; i++) s.StoreMsg("A", null, "OK"u8.ToArray(), 0); // Delete 5000 random sequences var rng = new Random(13); var deleteSet = new HashSet(); while (deleteSet.Count < 5000) deleteSet.Add(rng.Next(total) + 1); foreach (var seq in deleteSet) s.RemoveMsg((ulong)seq); var fstate = new StreamState(); s.FastState(ref fstate); // NumDeleted from FastState must equal interior gap count var fullState = s.State(); var dmapSize = fullState.Deleted?.Length ?? 0; dmapSize.ShouldBe(fstate.NumDeleted); } // ------------------------------------------------------------------------- // MessageTTL // ------------------------------------------------------------------------- // Go: TestMemStoreMessageTTL server/memstore_test.go:1202 [Fact] public void MessageTTL_ExpiresAfterDelay() { var cfg = new StreamConfig { Name = "zzz", Subjects = ["test"], Storage = StorageType.Memory, AllowMsgTtl = true, }; var ms = new MemStore(cfg); var s = Sync(ms); const long ttl = 1; // 1 second for (var i = 1; i <= 10; i++) s.StoreMsg("test", null, [], ttl); var ss = new StreamState(); s.FastState(ref ss); ss.FirstSeq.ShouldBe(1UL); ss.LastSeq.ShouldBe(10UL); ss.Msgs.ShouldBe(10UL); // Wait for TTL to expire (> 1 sec + check interval of 1 sec) Thread.Sleep(2_500); s.FastState(ref ss); ss.FirstSeq.ShouldBe(11UL); ss.LastSeq.ShouldBe(10UL); ss.Msgs.ShouldBe(0UL); } // ------------------------------------------------------------------------- // UpdateConfigTTLState // ------------------------------------------------------------------------- // Go: TestMemStoreUpdateConfigTTLState server/memstore_test.go:1299 [Fact] public void UpdateConfig_TtlStateInitializedAndDestroyed() { var cfg = new StreamConfig { Name = "zzz", Subjects = [">"], Storage = StorageType.Memory, AllowMsgTtl = false, }; var ms = new MemStore(cfg); var s = Sync(ms); // TTL disabled — internal TTL wheel should be null (we cannot observe it directly, // but UpdateConfig must not throw and subsequent behaviour must be correct) cfg.AllowMsgTtl = true; s.UpdateConfig(cfg); // Store with TTL — should work s.StoreMsg("test", null, [], 3600); s.State().Msgs.ShouldBe(1UL); // Disable TTL again cfg.AllowMsgTtl = false; s.UpdateConfig(cfg); // Message stored before disabling should still be present (TTL wheel gone but msg stays) s.State().Msgs.ShouldBe(1UL); } // ------------------------------------------------------------------------- // NextWildcardMatch // ------------------------------------------------------------------------- // Go: TestMemStoreNextWildcardMatch server/memstore_test.go:1373 [Fact] public void NextWildcardMatch_BoundsAreCorrect() { var ms = new MemStore(); var s = Sync(ms); void StoreN(string subj, int n) { for (var i = 0; i < n; i++) s.StoreMsg(subj, null, "msg"u8.ToArray(), 0); } StoreN("foo.bar.a", 1); // seq 1 StoreN("foo.baz.bar", 10); // seqs 2-11 StoreN("foo.bar.b", 1); // seq 12 StoreN("foo.baz.bar", 10); // seqs 13-22 StoreN("foo.baz.bar.no.match", 10); // seqs 23-32 lock (ms.Gate) { var (first, last, found) = ms.NextWildcardMatchLocked("foo.bar.*", 0); found.ShouldBeTrue(); first.ShouldBe(1UL); last.ShouldBe(12UL); (first, last, found) = ms.NextWildcardMatchLocked("foo.bar.*", 1); found.ShouldBeTrue(); first.ShouldBe(1UL); last.ShouldBe(12UL); (first, last, found) = ms.NextWildcardMatchLocked("foo.bar.*", 2); found.ShouldBeTrue(); first.ShouldBe(12UL); last.ShouldBe(12UL); (_, _, found) = ms.NextWildcardMatchLocked("foo.bar.*", first + 1); found.ShouldBeFalse(); (first, last, found) = ms.NextWildcardMatchLocked("foo.baz.*", 1); found.ShouldBeTrue(); first.ShouldBe(2UL); last.ShouldBe(22UL); (first, last, found) = ms.NextWildcardMatchLocked("foo.nope.*", 1); found.ShouldBeFalse(); first.ShouldBe(0UL); last.ShouldBe(0UL); (first, last, found) = ms.NextWildcardMatchLocked("foo.>", 1); found.ShouldBeTrue(); first.ShouldBe(1UL); last.ShouldBe(32UL); } } // ------------------------------------------------------------------------- // NextLiteralMatch // ------------------------------------------------------------------------- // Go: TestMemStoreNextLiteralMatch server/memstore_test.go:1454 [Fact] public void NextLiteralMatch_BoundsAreCorrect() { var ms = new MemStore(); var s = Sync(ms); void StoreN(string subj, int n) { for (var i = 0; i < n; i++) s.StoreMsg(subj, null, "msg"u8.ToArray(), 0); } StoreN("foo.bar.a", 1); // seq 1 StoreN("foo.baz.bar", 10); // seqs 2-11 StoreN("foo.bar.b", 1); // seq 12 StoreN("foo.baz.bar", 10); // seqs 13-22 StoreN("foo.baz.bar.no.match", 10); // seqs 23-32 lock (ms.Gate) { var (first, last, found) = ms.NextLiteralMatchLocked("foo.bar.a", 0); found.ShouldBeTrue(); first.ShouldBe(1UL); last.ShouldBe(1UL); (_, _, found) = ms.NextLiteralMatchLocked("foo.bar.a", 2); found.ShouldBeFalse(); (first, last, found) = ms.NextLiteralMatchLocked("foo.baz.bar", 1); found.ShouldBeTrue(); first.ShouldBe(2UL); last.ShouldBe(22UL); (first, last, found) = ms.NextLiteralMatchLocked("foo.baz.bar", 22); found.ShouldBeTrue(); first.ShouldBe(22UL); last.ShouldBe(22UL); (first, last, found) = ms.NextLiteralMatchLocked("foo.baz.bar", 23); found.ShouldBeFalse(); first.ShouldBe(0UL); last.ShouldBe(0UL); (_, _, found) = ms.NextLiteralMatchLocked("foo.nope", 1); found.ShouldBeFalse(); } } // ------------------------------------------------------------------------- // InitialFirstSeq // ------------------------------------------------------------------------- // Go: TestMemStoreInitialFirstSeq server/memstore_test.go:765 [Fact] public void InitialFirstSeq_StartAtConfiguredSeq() { var cfg = new StreamConfig { Name = "zzz", Storage = StorageType.Memory, FirstSeq = 1000, }; var ms = new MemStore(cfg); var s = Sync(ms); var (seq, _) = s.StoreMsg("A", null, "OK"u8.ToArray(), 0); seq.ShouldBe(1000UL); (seq, _) = s.StoreMsg("B", null, "OK"u8.ToArray(), 0); seq.ShouldBe(1001UL); var state = new StreamState(); s.FastState(ref state); state.Msgs.ShouldBe(2UL); state.FirstSeq.ShouldBe(1000UL); state.LastSeq.ShouldBe(1001UL); } // ------------------------------------------------------------------------- // PurgeEx with subject // ------------------------------------------------------------------------- // Go: TestMemStorePurgeExWithSubject server/memstore_test.go:437 [Fact] public void PurgeEx_WithSubject_PurgesAll() { var ms = new MemStore(); var s = Sync(ms); for (var i = 0; i < 100; i++) s.StoreMsg("foo", null, [], 0); s.PurgeEx("foo", 1, 0); s.State().Msgs.ShouldBe(0UL); } // ------------------------------------------------------------------------- // PurgeEx with deleted messages // ------------------------------------------------------------------------- // Go: TestMemStorePurgeExWithDeletedMsgs server/memstore_test.go:1031 [Fact] public void PurgeEx_WithDeletedMsgs_CorrectFirstSeq() { var ms = new MemStore(); var s = Sync(ms); var msg = "abc"u8.ToArray(); for (var i = 1; i <= 10; i++) s.StoreMsg("foo", null, msg, 0); s.RemoveMsg(2); s.RemoveMsg(9); // was the bug var n = s.PurgeEx("", 9, 0); n.ShouldBe(7UL); // seqs 1,3,4,5,6,7,8 (not 2 since deleted, not 9 since deleted) var state = new StreamState(); s.FastState(ref state); state.FirstSeq.ShouldBe(10UL); state.LastSeq.ShouldBe(10UL); state.Msgs.ShouldBe(1UL); } // ------------------------------------------------------------------------- // DeleteAll FirstSequenceCheck // ------------------------------------------------------------------------- // Go: TestMemStoreDeleteAllFirstSequenceCheck server/memstore_test.go:1060 [Fact] public void DeleteAll_FirstSeqIsLastPlusOne() { var ms = new MemStore(); var s = Sync(ms); for (var i = 1; i <= 10; i++) s.StoreMsg("foo", null, "abc"u8.ToArray(), 0); for (ulong seq = 1; seq <= 10; seq++) s.RemoveMsg(seq); var state = new StreamState(); s.FastState(ref state); state.FirstSeq.ShouldBe(11UL); state.LastSeq.ShouldBe(10UL); state.Msgs.ShouldBe(0UL); } // ------------------------------------------------------------------------- // NumPending — bug fix // ------------------------------------------------------------------------- // Go: TestMemStoreNumPendingBug server/memstore_test.go:1137 [Fact] public void NumPending_Bug_CorrectCount() { var ms = new MemStore(); var s = Sync(ms); foreach (var subj in new[] { "foo.foo", "foo.bar", "foo.baz", "foo.zzz" }) { s.StoreMsg("foo.aaa", null, [], 0); s.StoreMsg(subj, null, [], 0); s.StoreMsg(subj, null, [], 0); } // 12 msgs total var (total, _) = s.NumPending(4, "foo.*", false); ulong expected = 0; for (var seq = 4; seq <= 12; seq++) { try { var sm = s.LoadMsg((ulong)seq, null); if (SubjectMatchesFilter(sm.Subject, "foo.*")) expected++; } catch (KeyNotFoundException) { } } total.ShouldBe(expected); } // ------------------------------------------------------------------------- // Purge clears dmap // ------------------------------------------------------------------------- // Go: TestMemStorePurgeLeaksDmap server/memstore_test.go:1168 [Fact] public void Purge_ClearsDmap() { var ms = new MemStore(); var s = Sync(ms); for (var i = 0; i < 10; i++) s.StoreMsg("foo", null, [], 0); for (ulong i = 2; i <= 9; i++) s.RemoveMsg(i); // 8 interior gaps now var state = s.State(); state.NumDeleted.ShouldBe(8); // Purge should also clear dmap var purged = s.Purge(); purged.ShouldBe(2UL); // 2 actual msgs remain (1 and 10) state = s.State(); state.NumDeleted.ShouldBe(0); state.Deleted.ShouldBeNull(); } // ------------------------------------------------------------------------- // Helpers // ------------------------------------------------------------------------- private static bool SubjectMatchesFilter(string subject, string filter) { if (string.IsNullOrEmpty(filter) || filter == ">") return true; if (NATS.Server.Subscriptions.SubjectMatch.IsLiteral(filter)) return string.Equals(subject, filter, StringComparison.Ordinal); return NATS.Server.Subscriptions.SubjectMatch.MatchLiteral(subject, filter); } }