2 changed files with 132 additions and 25 deletions
@@ -98,14 +98,21 @@ public sealed class SessionEventDistributor : IAsyncDisposable
    private readonly object _lifecycleLock = new();

    // Replay ring buffer. Appended on the pump thread and queried from arbitrary
-    // threads via TryGetReplayFrom, so every access is under _replayLock. The deque
-    // keeps events in ascending WorkerSequence order (the pump fans in source order),
-    // so the oldest retained event is always at the front. Capacity == 0 disables
-    // retention; RetentionSeconds <= 0 disables age-based eviction.
+    // threads via TryGetReplayFrom, so every access is under _replayLock. Backed by a
+    // fixed-size circular array preallocated to the capacity so appending a retained
+    // event allocates no node (the LinkedList this replaced allocated one node per
+    // event on the fan-out hot path). Events are kept in ascending WorkerSequence order
+    // (the pump fans in source order): _replayHead is the logical front (oldest retained
+    // event) and _replayCount entries follow it, wrapping modulo the array length. The
+    // logical entry at position i is _replayBuffer[(_replayHead + i) % Length]. Capacity
+    // == 0 disables retention (the array is empty and never indexed); RetentionSeconds
+    // <= 0 disables age-based eviction.
    private readonly int _replayBufferCapacity;
    private readonly TimeSpan _replayRetention;
    private readonly bool _ageEvictionEnabled;
-    private readonly LinkedList<ReplayEntry> _replayBuffer = new();
+    private readonly ReplayEntry[] _replayBuffer;
+    private int _replayHead;
+    private int _replayCount;
    private readonly object _replayLock = new();
    private bool _anyEventSeen;
    private ulong _highestSequenceSeen;
@@ -245,6 +252,7 @@ public sealed class SessionEventDistributor : IAsyncDisposable
        _overflowHandler = overflowHandler;
        _shutdownTimeout = DefaultShutdownTimeout;
        _replayBufferCapacity = replayBufferCapacity;
+        _replayBuffer = new ReplayEntry[replayBufferCapacity];
        _ageEvictionEnabled = replayRetentionSeconds > 0;
        _replayRetention = _ageEvictionEnabled
            ? TimeSpan.FromSeconds(replayRetentionSeconds)
@@ -452,24 +460,25 @@ public sealed class SessionEventDistributor : IAsyncDisposable
            List<MxEvent> newer = [];
            ulong highestReplayed = afterSequence;

-            if (_replayBuffer.Count == 0)
+            if (_replayCount == 0)
            {
                gap = _anyEventSeen && afterSequence < _highestSequenceSeen;
                oldestAvailableSequence = 0; // meaningful only when gap == true; 0 here since nothing is retained
            }
            else
            {
-                ulong oldestRetained = _replayBuffer.First!.Value.Event.WorkerSequence;
+                ulong oldestRetained = ReplayEntryAt(0).Event.WorkerSequence;
                gap = oldestRetained > 0 && afterSequence < oldestRetained - 1;
                // Per the contract on OldestAvailableSequence: meaningful only when gap == true.
                oldestAvailableSequence = gap ? oldestRetained : 0;

-                foreach (ReplayEntry entry in _replayBuffer)
+                for (int i = 0; i < _replayCount; i++)
                {
-                    if (entry.Event.WorkerSequence > afterSequence)
+                    MxEvent retained = ReplayEntryAt(i).Event;
+                    if (retained.WorkerSequence > afterSequence)
                    {
-                        newer.Add(entry.Event);
-                        highestReplayed = entry.Event.WorkerSequence;
+                        newer.Add(retained);
+                        highestReplayed = retained.WorkerSequence;
                    }
                }
            }
@@ -729,7 +738,7 @@ public sealed class SessionEventDistributor : IAsyncDisposable
        {
            EvictAged();

-            if (_replayBuffer.Count == 0)
+            if (_replayCount == 0)
            {
                events = [];
                // Nothing retained. The caller missed events only if it is behind the
@@ -738,7 +747,7 @@ public sealed class SessionEventDistributor : IAsyncDisposable
                return true;
            }

-            ulong oldestRetained = _replayBuffer.First!.Value.Event.WorkerSequence;
+            ulong oldestRetained = ReplayEntryAt(0).Event.WorkerSequence;

            // A gap exists when at least one event newer than afterSequence was evicted,
            // i.e. afterSequence sits below the oldest-retained-minus-one boundary.
@@ -750,11 +759,12 @@ public sealed class SessionEventDistributor : IAsyncDisposable
            // O(n) scan over the retained buffer — acceptable because TryGetReplayFrom
            // is only called on subscriber reconnect, never on the hot fan-out path.
            List<MxEvent> newer = [];
-            foreach (ReplayEntry entry in _replayBuffer)
+            for (int i = 0; i < _replayCount; i++)
            {
-                if (entry.Event.WorkerSequence > afterSequence)
+                MxEvent retained = ReplayEntryAt(i).Event;
+                if (retained.WorkerSequence > afterSequence)
                {
-                    newer.Add(entry.Event);
+                    newer.Add(retained);
                }
            }

@@ -780,30 +790,43 @@ public sealed class SessionEventDistributor : IAsyncDisposable
                return;
            }

-            _replayBuffer.AddLast(new ReplayEntry(mxEvent, _timeProvider.GetUtcNow()));
-
-            // Capacity eviction: drop oldest until within bound.
-            while (_replayBuffer.Count > _replayBufferCapacity)
+            // Append at the logical tail. When the ring is full the oldest entry is
+            // overwritten in place (its slot becomes the new tail) and the head advances,
+            // so the newest _replayBufferCapacity events are retained with no allocation.
+            ReplayEntry entry = new(mxEvent, _timeProvider.GetUtcNow());
+            if (_replayCount < _replayBufferCapacity)
            {
-                _replayBuffer.RemoveFirst();
+                _replayBuffer[(_replayHead + _replayCount) % _replayBufferCapacity] = entry;
+                _replayCount++;
+            }
+            else
+            {
+                _replayBuffer[_replayHead] = entry;
+                _replayHead = (_replayHead + 1) % _replayBufferCapacity;
            }

            EvictAged();
        }
    }

-    // Must be called under _replayLock. Drops entries older than the retention window.
+    // Returns the logical entry at position i (0 == oldest retained). Must be called
+    // under _replayLock with 0 <= i < _replayCount (so the array length is nonzero).
+    private ReplayEntry ReplayEntryAt(int i) => _replayBuffer[(_replayHead + i) % _replayBuffer.Length];
+
+    // Must be called under _replayLock. Drops entries older than the retention window
+    // by advancing the head past them (no dealloc; slots are reused on the next append).
    private void EvictAged()
    {
-        if (!_ageEvictionEnabled || _replayBuffer.Count == 0)
+        if (!_ageEvictionEnabled || _replayCount == 0)
        {
            return;
        }

        DateTimeOffset cutoff = _timeProvider.GetUtcNow() - _replayRetention;
-        while (_replayBuffer.First is { } first && first.Value.RetainedAt < cutoff)
+        while (_replayCount > 0 && _replayBuffer[_replayHead].RetainedAt < cutoff)
        {
-            _replayBuffer.RemoveFirst();
+            _replayHead = (_replayHead + 1) % _replayBuffer.Length;
+            _replayCount--;
        }
    }

@@ -185,6 +185,90 @@ public sealed class SessionEventDistributorTests
        Assert.Equal(new ulong[] { 3, 4, 5 }, replay.Select(e => e.WorkerSequence));
    }

+    /// <summary>
+    ///     Appending far more events than the capacity wraps the circular buffer's head
+    ///     index past the array boundary multiple times; the retained window stays the
+    ///     newest <c>capacity</c> events, in ascending order, and a replay from before the
+    ///     window reports a gap and returns the whole retained tail.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task ReplayBuffer_WrapsRingMultipleTimes_RetainsNewestInAscendingOrder()
+    {
+        Channel<MxEvent> source = Channel.CreateUnbounded<MxEvent>();
+        await using SessionEventDistributor distributor = CreateDistributor(
+            source.Reader,
+            replayBufferCapacity: 4,
+            replayRetentionSeconds: 0);
+        await distributor.StartAsync(CancellationToken.None);
+
+        // 13 events through a capacity-4 ring advances the head index 13 - 4 = 9 slots,
+        // wrapping the 4-slot array's boundary more than twice.
+        using IEventSubscriberLease lease = distributor.Register();
+        for (ulong sequence = 1; sequence <= 13; sequence++)
+        {
+            source.Writer.TryWrite(Event(sequence));
+        }
+
+        for (ulong sequence = 1; sequence <= 13; sequence++)
+        {
+            MxEvent e = await ReadOneAsync(lease.Reader);
+            Assert.Equal(sequence, e.WorkerSequence);
+        }
+
+        // Newest four (10, 11, 12, 13) retained in ascending order despite the wraps.
+        bool found = distributor.TryGetReplayFrom(0, out IReadOnlyList<MxEvent> replay, out bool gap);
+
+        Assert.True(found);
+        Assert.True(gap);
+        Assert.Equal(new ulong[] { 10, 11, 12, 13 }, replay.Select(e => e.WorkerSequence));
+
+        // A replay from inside the wrapped window returns only the newer entries, no gap,
+        // proving the modular scan reads the logical order and not the physical slots.
+        bool foundInner = distributor.TryGetReplayFrom(11, out IReadOnlyList<MxEvent> inner, out bool innerGap);
+
+        Assert.True(foundInner);
+        Assert.False(innerGap);
+        Assert.Equal(new ulong[] { 12, 13 }, inner.Select(e => e.WorkerSequence));
+    }
+
+    /// <summary>
+    ///     A capacity-1 ring retains only the single newest event; each append overwrites
+    ///     the sole slot, and a replay from before it reports a gap.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task ReplayBuffer_Capacity1_RetainsOnlyNewest()
+    {
+        Channel<MxEvent> source = Channel.CreateUnbounded<MxEvent>();
+        await using SessionEventDistributor distributor = CreateDistributor(
+            source.Reader,
+            replayBufferCapacity: 1,
+            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);
+        }
+
+        // Only sequence 3 is retained; a request from 0 missed 1 and 2 => gap.
+        bool found = distributor.TryGetReplayFrom(0, out IReadOnlyList<MxEvent> replay, out bool gap);
+
+        Assert.True(found);
+        Assert.True(gap);
+        Assert.Equal(new ulong[] { 3 }, replay.Select(e => e.WorkerSequence));
+
+        // A request from exactly the newest retained sequence is caught up: empty, no gap.
+        bool foundCaughtUp = distributor.TryGetReplayFrom(3, out IReadOnlyList<MxEvent> caughtUp, out bool caughtUpGap);
+
+        Assert.True(foundCaughtUp);
+        Assert.False(caughtUpGap);
+        Assert.Empty(caughtUp);
+    }
+
    /// <summary>Requesting replay from a sequence still inside the retained window returns only the newer events, with no gap.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]