feat: phase B distributed substrate test parity — 39 new tests across 5 subsystems
FileStore basics (4), MemStore/retention (10), RAFT election/append (16), config reload parity (3), monitoring endpoints varz/connz/healthz (6). 972 total tests passing, 0 failures.
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,163 @@
|
||||
// Ported from golang/nats-server/server/memstore_test.go:
|
||||
// TestMemStoreMsgLimit, TestMemStoreBytesLimit, TestMemStoreAgeLimit
|
||||
//
|
||||
// Retention limits are enforced by StreamManager (which calls MemStore.TrimToMaxMessages,
|
||||
// removes oldest messages by bytes, and prunes by age). These tests exercise the full
|
||||
// Limits-retention path via StreamManager.Capture, which is the code path the Go server
|
||||
// exercises through its StoreMsg integration.
|
||||
|
||||
using System.Text;
|
||||
using NATS.Server.JetStream;
|
||||
using NATS.Server.JetStream.Models;
|
||||
|
||||
namespace NATS.Server.Tests.JetStream.Storage;
|
||||
|
||||
public class StorageRetentionTests
|
||||
{
|
||||
// Go ref: TestMemStoreMsgLimit — store MaxMsgs+N messages; only MaxMsgs remain,
|
||||
// oldest are evicted, sequence window advances.
|
||||
[Fact]
|
||||
public async Task Max_msgs_limit_enforced()
|
||||
{
|
||||
const int maxMsgs = 10;
|
||||
const int overCount = 20;
|
||||
|
||||
var manager = new StreamManager();
|
||||
manager.CreateOrUpdate(new StreamConfig
|
||||
{
|
||||
Name = "MSGLIMIT",
|
||||
Subjects = ["msglimit.*"],
|
||||
MaxMsgs = maxMsgs,
|
||||
Storage = StorageType.Memory,
|
||||
}).Error.ShouldBeNull();
|
||||
|
||||
for (var i = 0; i < overCount; i++)
|
||||
manager.Capture("msglimit.foo", Encoding.UTF8.GetBytes($"msg{i}"));
|
||||
|
||||
manager.TryGet("MSGLIMIT", out var handle).ShouldBeTrue();
|
||||
var state = await handle.Store.GetStateAsync(default);
|
||||
|
||||
state.Messages.ShouldBe((ulong)maxMsgs);
|
||||
// The last stored sequence is overCount.
|
||||
state.LastSeq.ShouldBe((ulong)overCount);
|
||||
// The first kept sequence is overCount - maxMsgs + 1.
|
||||
state.FirstSeq.ShouldBe((ulong)(overCount - maxMsgs + 1));
|
||||
}
|
||||
|
||||
// Go ref: TestMemStoreBytesLimit — store messages until bytes exceed MaxBytes;
|
||||
// oldest messages are purged to keep total bytes at or below the limit.
|
||||
[Fact]
|
||||
public async Task Max_bytes_limit_enforced()
|
||||
{
|
||||
// Each payload is 100 bytes. Set MaxBytes to hold exactly 5 messages.
|
||||
var payload = new byte[100];
|
||||
const int payloadSize = 100;
|
||||
const int maxCapacity = 5;
|
||||
var maxBytes = (long)(payloadSize * maxCapacity);
|
||||
|
||||
var manager = new StreamManager();
|
||||
manager.CreateOrUpdate(new StreamConfig
|
||||
{
|
||||
Name = "BYTESLIMIT",
|
||||
Subjects = ["byteslimit.*"],
|
||||
MaxBytes = maxBytes,
|
||||
Storage = StorageType.Memory,
|
||||
}).Error.ShouldBeNull();
|
||||
|
||||
// Store exactly maxCapacity messages — should all fit.
|
||||
for (var i = 0; i < maxCapacity; i++)
|
||||
manager.Capture("byteslimit.foo", payload);
|
||||
|
||||
manager.TryGet("BYTESLIMIT", out var handle).ShouldBeTrue();
|
||||
var stateAtCapacity = await handle.Store.GetStateAsync(default);
|
||||
stateAtCapacity.Messages.ShouldBe((ulong)maxCapacity);
|
||||
stateAtCapacity.Bytes.ShouldBe((ulong)(payloadSize * maxCapacity));
|
||||
|
||||
// Store 5 more — each one should displace an old message.
|
||||
for (var i = 0; i < maxCapacity; i++)
|
||||
manager.Capture("byteslimit.foo", payload);
|
||||
|
||||
var stateFinal = await handle.Store.GetStateAsync(default);
|
||||
stateFinal.Messages.ShouldBe((ulong)maxCapacity);
|
||||
stateFinal.Bytes.ShouldBeLessThanOrEqualTo((ulong)maxBytes);
|
||||
stateFinal.LastSeq.ShouldBe((ulong)(maxCapacity * 2));
|
||||
}
|
||||
|
||||
// Go ref: TestMemStoreAgeLimit — messages older than MaxAge are pruned on the next Capture.
|
||||
// In the Go server, the memstore runs a background timer; in the .NET port, pruning happens
|
||||
// synchronously inside StreamManager.Capture via PruneExpiredMessages which compares
|
||||
// TimestampUtc against (now - MaxAge). We backdate stored messages to simulate expiry.
|
||||
[Fact]
|
||||
public async Task Max_age_limit_enforced()
|
||||
{
|
||||
// Use a 1-second MaxAge so we can reason clearly about cutoff.
|
||||
const int maxAgeMs = 1000;
|
||||
|
||||
var manager = new StreamManager();
|
||||
manager.CreateOrUpdate(new StreamConfig
|
||||
{
|
||||
Name = "AGELIMIT",
|
||||
Subjects = ["agelimit.*"],
|
||||
MaxAgeMs = maxAgeMs,
|
||||
Storage = StorageType.Memory,
|
||||
}).Error.ShouldBeNull();
|
||||
|
||||
// Store 5 messages that are logically "already expired" by storing them,
|
||||
// then relying on an additional capture after sleeping past MaxAge to trigger
|
||||
// the pruning pass.
|
||||
const int initialCount = 5;
|
||||
for (var i = 0; i < initialCount; i++)
|
||||
manager.Capture("agelimit.foo", Encoding.UTF8.GetBytes($"msg{i}"));
|
||||
|
||||
manager.TryGet("AGELIMIT", out var handle).ShouldBeTrue();
|
||||
var stateBefore = await handle.Store.GetStateAsync(default);
|
||||
stateBefore.Messages.ShouldBe((ulong)initialCount);
|
||||
|
||||
// Wait for MaxAge to elapse so the stored messages are now older than the cutoff.
|
||||
await Task.Delay(maxAgeMs + 50);
|
||||
|
||||
// A subsequent Capture triggers PruneExpiredMessages, which removes all messages
|
||||
// whose TimestampUtc < (now - MaxAge).
|
||||
manager.Capture("agelimit.foo", "trigger"u8.ToArray());
|
||||
|
||||
var stateAfter = await handle.Store.GetStateAsync(default);
|
||||
// Only the freshly-appended trigger message should remain.
|
||||
stateAfter.Messages.ShouldBe((ulong)1);
|
||||
stateAfter.Bytes.ShouldBeGreaterThan((ulong)0);
|
||||
}
|
||||
|
||||
// Go ref: TestMemStoreMsgLimit — verifies that sequence numbers keep incrementing even
|
||||
// after old messages are evicted; the store window moves forward rather than wrapping.
|
||||
[Fact]
|
||||
public async Task Sequence_numbers_monotonically_increase_through_eviction()
|
||||
{
|
||||
const int maxMsgs = 5;
|
||||
const int totalToStore = 15;
|
||||
|
||||
var manager = new StreamManager();
|
||||
manager.CreateOrUpdate(new StreamConfig
|
||||
{
|
||||
Name = "SEQMONOT",
|
||||
Subjects = ["seqmonot.*"],
|
||||
MaxMsgs = maxMsgs,
|
||||
Storage = StorageType.Memory,
|
||||
}).Error.ShouldBeNull();
|
||||
|
||||
for (var i = 1; i <= totalToStore; i++)
|
||||
manager.Capture("seqmonot.foo", Encoding.UTF8.GetBytes($"msg{i}"));
|
||||
|
||||
manager.TryGet("SEQMONOT", out var handle).ShouldBeTrue();
|
||||
var state = await handle.Store.GetStateAsync(default);
|
||||
|
||||
state.Messages.ShouldBe((ulong)maxMsgs);
|
||||
state.LastSeq.ShouldBe((ulong)totalToStore);
|
||||
state.FirstSeq.ShouldBe((ulong)(totalToStore - maxMsgs + 1));
|
||||
|
||||
// The first evicted sequence (1) is no longer loadable.
|
||||
(await handle.Store.LoadAsync(1, default)).ShouldBeNull();
|
||||
// The last evicted sequence is totalToStore - maxMsgs (= 10).
|
||||
(await handle.Store.LoadAsync((ulong)(totalToStore - maxMsgs), default)).ShouldBeNull();
|
||||
// The first surviving message is still present.
|
||||
(await handle.Store.LoadAsync((ulong)(totalToStore - maxMsgs + 1), default)).ShouldNotBeNull();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user