diff --git a/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionEventDistributor.cs b/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionEventDistributor.cs index 441ce1c..9ea3ec4 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionEventDistributor.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionEventDistributor.cs @@ -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 /// (when newer than /// ). 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 <= liveResumeSequence, 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 <= liveResumeSequence and the caller's live filter + /// DROPS it. /// /// /// If the pump appends E after this critical section, E is NOT in the snapshot, @@ -464,6 +469,16 @@ public sealed class SessionEventDistributor : IAsyncDisposable /// /// /// + /// Capturing the fan-out array inside the append's _replayLock 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 remains + /// part of this method's contract. + /// + /// /// Lock ordering: this is the only path that holds both _replayLock and /// _lifecycleLock; it always takes _replayLock first then /// _lifecycleLock. 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); } }