feat(sessions): add bounded replay ring buffer to SessionEventDistributor

2026-06-15 12:42:15 -04:00
parent 7773bdebbd
commit e962737d2c
6 changed files with 375 additions and 3 deletions
@@ -1,5 +1,6 @@
 using System.Threading.Channels;
 using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Time.Testing;
 using ZB.MOM.WW.MxGateway.Contracts.Proto;
 using ZB.MOM.WW.MxGateway.Server.Sessions;

@@ -112,12 +113,142 @@ public sealed class SessionEventDistributorTests
        Assert.Throws<ObjectDisposedException>(() => distributor.Register());
    }

+    [Fact]
+    public async Task ReplayBuffer_OverCapacity_EvictsOldestFirst_AndReportsGap()
+    {
+        Channel<MxEvent> source = Channel.CreateUnbounded<MxEvent>();
+        await using SessionEventDistributor distributor = CreateDistributor(
+            source.Reader,
+            replayBufferCapacity: 3,
+            replayRetentionSeconds: 0);
+        await distributor.StartAsync(CancellationToken.None);
+
+        // A live subscriber forces the pump to fan (and thereby retain) each event,
+        // and gives us a deterministic point to know the pump has processed event 5.
+        using IEventSubscriberLease lease = distributor.Register();
+        for (ulong sequence = 1; sequence <= 5; sequence++)
+        {
+            source.Writer.TryWrite(Event(sequence));
+        }
+
+        for (ulong sequence = 1; sequence <= 5; sequence++)
+        {
+            MxEvent e = await ReadOneAsync(lease.Reader);
+            Assert.Equal(sequence, e.WorkerSequence);
+        }
+
+        // Capacity 3 retains only the newest three: sequences 3, 4, 5. Events 1 and 2
+        // were evicted, so a caller asking from 0 missed events => gap=true, and it
+        // gets only the retained tail.
+        bool found = distributor.TryGetReplayFrom(0, out IReadOnlyList<MxEvent> replay, out bool gap);
+
+        Assert.True(found);
+        Assert.True(gap);
+        Assert.Equal(new ulong[] { 3, 4, 5 }, replay.Select(e => e.WorkerSequence));
+    }
+
+    [Fact]
+    public async Task ReplayBuffer_WithinRetainedWindow_ReturnsNewerEvents_NoGap()
+    {
+        Channel<MxEvent> source = Channel.CreateUnbounded<MxEvent>();
+        await using SessionEventDistributor distributor = CreateDistributor(
+            source.Reader,
+            replayBufferCapacity: 10,
+            replayRetentionSeconds: 0);
+        await distributor.StartAsync(CancellationToken.None);
+
+        using IEventSubscriberLease lease = distributor.Register();
+        for (ulong sequence = 1; sequence <= 5; sequence++)
+        {
+            source.Writer.TryWrite(Event(sequence));
+            _ = await ReadOneAsync(lease.Reader);
+        }
+
+        // afterSequence 2 is still inside the retained window [1..5], so no gap and
+        // exactly the newer events 3, 4, 5 come back.
+        bool found = distributor.TryGetReplayFrom(2, out IReadOnlyList<MxEvent> replay, out bool gap);
+
+        Assert.True(found);
+        Assert.False(gap);
+        Assert.Equal(new ulong[] { 3, 4, 5 }, replay.Select(e => e.WorkerSequence));
+    }
+
+    [Fact]
+    public async Task ReplayBuffer_AgedEntries_AreEvictedAfterRetentionElapses()
+    {
+        FakeTimeProvider time = new(new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero));
+        Channel<MxEvent> source = Channel.CreateUnbounded<MxEvent>();
+        await using SessionEventDistributor distributor = CreateDistributor(
+            source.Reader,
+            replayBufferCapacity: 100,
+            replayRetentionSeconds: 30,
+            timeProvider: time);
+        await distributor.StartAsync(CancellationToken.None);
+
+        using IEventSubscriberLease lease = distributor.Register();
+
+        // Two old events, then advance the clock well past the retention window.
+        source.Writer.TryWrite(Event(1));
+        source.Writer.TryWrite(Event(2));
+        _ = await ReadOneAsync(lease.Reader);
+        _ = await ReadOneAsync(lease.Reader);
+
+        time.Advance(TimeSpan.FromSeconds(60));
+
+        // A fresh event triggers age-eviction of the now-stale entries 1 and 2.
+        source.Writer.TryWrite(Event(3));
+        _ = await ReadOneAsync(lease.Reader);
+
+        bool found = distributor.TryGetReplayFrom(0, out IReadOnlyList<MxEvent> replay, out bool gap);
+
+        Assert.True(found);
+        // Events 1 and 2 aged out; only 3 remains, and 0 predates the oldest retained.
+        Assert.Equal(new ulong[] { 3 }, replay.Select(e => e.WorkerSequence));
+        Assert.True(gap);
+    }
+
+    [Fact]
+    public async Task ReplayBuffer_AfterSequenceNewerThanAllRetained_ReturnsEmpty_NoGap()
+    {
+        Channel<MxEvent> source = Channel.CreateUnbounded<MxEvent>();
+        await using SessionEventDistributor distributor = CreateDistributor(
+            source.Reader,
+            replayBufferCapacity: 10,
+            replayRetentionSeconds: 0);
+        await distributor.StartAsync(CancellationToken.None);
+
+        using IEventSubscriberLease lease = distributor.Register();
+        for (ulong sequence = 1; sequence <= 3; sequence++)
+        {
+            source.Writer.TryWrite(Event(sequence));
+            _ = await ReadOneAsync(lease.Reader);
+        }
+
+        // afterSequence 3 is at/after the newest retained; nothing newer, and the
+        // caller is fully caught up => empty list, gap=false.
+        bool found = distributor.TryGetReplayFrom(3, out IReadOnlyList<MxEvent> replay, out bool gap);
+
+        Assert.True(found);
+        Assert.False(gap);
+        Assert.Empty(replay);
+    }
+
    private static SessionEventDistributor CreateDistributor(ChannelReader<MxEvent> source)
+        => CreateDistributor(source, replayBufferCapacity: 1024, replayRetentionSeconds: 300);
+
+    private static SessionEventDistributor CreateDistributor(
+        ChannelReader<MxEvent> source,
+        int replayBufferCapacity,
+        double replayRetentionSeconds,
+        TimeProvider? timeProvider = null)
        => new(
            "session-test",
            ct => source.ReadAllAsync(ct),
            subscriberQueueCapacity: 64,
-            NullLogger<SessionEventDistributor>.Instance);
+            replayBufferCapacity: replayBufferCapacity,
+            replayRetentionSeconds: replayRetentionSeconds,
+            NullLogger<SessionEventDistributor>.Instance,
+            timeProvider ?? TimeProvider.System);

    private static MxEvent Event(ulong sequence)
        => new() { SessionId = "session-test", WorkerSequence = sequence };