dohertj2/mxaccessgw
1
0
Fork 0
Code Issues 14 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

docs(events): correct the capture-under-replayLock rationale

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-15 12:34:36 -04:00
parent f920b4cbf5
commit 07b83561d1
1 changed files with 70 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionEventDistributor.cs
+70 -37
View File
@@ -111,9 +111,13 @@ public sealed class SessionEventDistributor : IAsyncDisposable
// Copy-on-write fan-out snapshot of _subscribers.Values. Rebuilt (a whole new array)
// inside the _lifecycleLock section of every register/unregister; never mutated in
// place, so the pump can walk the array it captured with no lock and no allocation.
// Written with Volatile.Write / read with Volatile.Read so a reader on another core
// cannot observe a stale reference after the publishing store. See the type remarks
// for why fan-out must not touch ConcurrentDictionary.Values.
// Volatile.Write / Volatile.Read ORDER the access — they keep the publishing store from
// sinking past the lock release and keep the pump's read from being hoisted out of the
// fan-out loop. They do NOT promise freshness, and nothing here needs them to: a reader
// may legitimately observe the previous array, which IS the documented "late subscribers
// see events after they register" window. Where visibility must be guaranteed — the
// RegisterWithReplay handoff — it comes from the _replayLock edge, not from Volatile.
// See the type remarks for why fan-out must not touch ConcurrentDictionary.Values.
private Subscriber[] _subscriberSnapshot = [];
// Replay ring buffer. Appended on the pump thread and queried from arbitrary
@@ -451,9 +455,10 @@ public sealed class SessionEventDistributor : IAsyncDisposable
/// If the pump appended E before this critical section, E is in
/// <paramref name="replayedEvents"/> (when newer than
/// <paramref name="afterSequence"/>). The pump captured its subscriber array in
/// that same earlier section, so it cannot fan E into this not-yet-registered
/// channel. Even if it could, E's sequence is <c>&lt;= liveResumeSequence</c>, so
/// the caller's live filter DROPS it — no duplicate either way.
/// that same earlier section, so it cannot also fan E into this
/// not-yet-registered channel — no duplicate. Belt and braces: even if it did,
/// E's sequence is <c>&lt;= liveResumeSequence</c> and the caller's live filter
/// DROPS it.
/// </item>
/// <item>
/// If the pump appends E after this critical section, E is NOT in the snapshot,
@@ -464,6 +469,16 @@ public sealed class SessionEventDistributor : IAsyncDisposable
/// </item>
/// </list>
/// <para>
/// Capturing the fan-out array inside the append's <c>_replayLock</c> section is what
/// makes the first bullet's "cannot" hold. It is defense in depth rather than a
/// correctness fix: a capture taken after that lock released could not drop an event
/// either (the lock edge orders it), it could only produce the duplicate the live
/// filter already discards. Doing it under the lock costs nothing and stops
/// no-duplicate from depending on every caller remembering to apply the filter —
/// which callers MUST still do, since <paramref name="liveResumeSequence"/> remains
/// part of this method's contract.
/// </para>
/// <para>
/// Lock ordering: this is the only path that holds both <c>_replayLock</c> and
/// <c>_lifecycleLock</c>; it always takes <c>_replayLock</c> first then
/// <c>_lifecycleLock</c>. No other path acquires both, so there is no inversion.
@@ -627,10 +642,13 @@ public sealed class SessionEventDistributor : IAsyncDisposable
// Walk the captured copy-on-write array: no dictionary enumeration, no
// per-event allocation. A subscriber registered after this capture misses
// this event, which matches "late subscribers see events after they
// register". A subscriber unregistered after the capture is still written
// to — the identical window the previous ConcurrentDictionary.Values
// enumeration had (that property also materialized its list up front), so
// the race and its outcome are unchanged by the copy-on-write array.
// register". A subscriber UNREGISTERED after the capture is still written to,
// and TryWrite on its completed channel returns false — from here that is
// indistinguishable from a real overflow. The window predates the
// copy-on-write array (ConcurrentDictionary.Values materialized its list up
// front too) and its outcome is NOT benign, so telling a graceful unregister
// apart from a genuine overflow is OnSubscriberOverflow's job, not this
// loop's.
foreach (Subscriber subscriber in subscribers)
{
// Non-blocking write: TryWrite never blocks the pump on a slow reader.
@@ -854,14 +872,28 @@ public sealed class SessionEventDistributor : IAsyncDisposable
}
// Appends an event to the replay ring AND captures the fan-out array the pump will write it
// to, in ONE _replayLock section. The capture must share the append's critical section, not
// follow it: RegisterWithReplay snapshots the ring and registers under the same _replayLock,
// so mutual exclusion is what puts each event strictly before or strictly after a resume —
// "replayed, not fanned" or "fanned, not replayed", never neither. Capturing after the lock
// released would let a resume interleave between the append and the capture, replaying
// nothing for the event and fanning it to a stale array that omits the new subscriber: a
// silently dropped event. Returns the array; the pump fans OUTSIDE the lock so a slow
// reader can never stall replay.
// to, in ONE _replayLock section, making append+capture atomic with respect to
// RegisterWithReplay (which snapshots the ring and registers under that same lock). Each
// event therefore lands strictly on one side of a resume: replayed to that subscriber, or
// fanned to it live — never both.
//
// This is defense in depth, NOT a correctness fix; capturing after the lock released would
// also be correct. Monitor.Enter is an acquire (ECMA-335 I.12.6.5), so a later read cannot
// move above the append's lock acquisition, and a resume whose entire locked section
// (ring snapshot, registration, array republish) preceded the append is visible across that
// lock edge — no event can be silently dropped. What a late capture would allow is the
// benign case: an event both replayed AND written to the new subscriber's live channel, a
// duplicate the caller's liveResumeSequence filter discards. Capturing under the lock
// removes that duplicate at the source, so "no duplicate" no longer rests on the caller
// actually applying the filter — bought at zero cost, since the pump holds this lock anyway.
//
// Lock ordering: this helper only READS the already-published array, deliberately. The one
// permitted nesting in this type is RegisterWithReplay's _replayLock -> _lifecycleLock;
// every other path takes exactly one lock. Rebuilding here instead — an obvious-looking
// lock(_lifecycleLock) inside this _replayLock section — would drag the pump's hot path into
// that nesting and turn any future _lifecycleLock -> _replayLock path into a deadlock.
//
// Returns the array; the pump fans OUTSIDE the lock so a slow reader can never stall replay.
private Subscriber[] AppendToReplayBufferAndCaptureSubscribers(MxEvent mxEvent)
{
lock (_replayLock)
@@ -873,28 +905,29 @@ public sealed class SessionEventDistributor : IAsyncDisposable
}
// Capacity 0 disables retention: track the highest-seen sequence (so replay
// can still report a gap) but keep no events.
if (_replayBufferCapacity == 0)
// can still report a gap) but keep no events. The capture below still runs —
// retention being off says nothing about the fan-out set.
if (_replayBufferCapacity > 0)
{
return Volatile.Read(ref _subscriberSnapshot);
// 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[(_replayHead + _replayCount) % _replayBufferCapacity] = entry;
_replayCount++;
}
else
{
_replayBuffer[_replayHead] = entry;
_replayHead = (_replayHead + 1) % _replayBufferCapacity;
}
EvictAged();
}
// 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[(_replayHead + _replayCount) % _replayBufferCapacity] = entry;
_replayCount++;
}
else
{
_replayBuffer[_replayHead] = entry;
_replayHead = (_replayHead + 1) % _replayBufferCapacity;
}
EvictAged();
// Single capture point for both the retained and no-retention paths.
return Volatile.Read(ref _subscriberSnapshot);
}
}