diff --git a/src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamService.cs b/src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamService.cs index a41f67b..ea1cf48 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamService.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamService.cs @@ -1,4 +1,5 @@ using System.Runtime.CompilerServices; +using System.Threading.Channels; using Microsoft.Extensions.Options; using ZB.MOM.WW.MxGateway.Contracts.Proto; using ZB.MOM.WW.MxGateway.Server.Configuration; @@ -37,7 +38,7 @@ public sealed class EventStreamService( // non-blocking. When this subscriber's channel is full the pump applies the per-subscriber // backpressure policy and completes this subscriber's channel with a SessionManagerException // (SessionManagerErrorCode.EventQueueOverflow). That terminal fault surfaces here when the - // reader's MoveNextAsync throws, and it propagates to the gRPC client unchanged. The overflow + // reader's WaitToReadAsync throws, and it propagates to the gRPC client unchanged. The overflow // metric, and (in the legacy single-subscriber FailFast case) the session fault + fault metric, // are recorded by the distributor's overflow handler so the session, the pump, and other // subscribers are isolated from this subscriber's slowness. @@ -106,9 +107,14 @@ public sealed class EventStreamService( options.Value.Sessions.MaxEventSubscribersPerSession); } - IAsyncEnumerator reader = subscriber.Reader - .ReadAllAsync(cancellationToken) - .GetAsyncEnumerator(cancellationToken); + // Consume the subscriber channel directly (WaitToReadAsync + an inner TryRead drain) + // rather than through ReadAllAsync's IAsyncEnumerable wrapper. This is the hottest + // per-event path in the gateway and the wrapper added a second async state machine hop + // per event for no behavioral benefit: WaitToReadAsync observes cancellation and a + // faulted completion exactly as MoveNextAsync did, and TryRead drains what is already + // buffered without allocating a wait. StreamEventsAsync itself stays an async iterator — + // its `yield return` is what feeds the gRPC writer. + ChannelReader reader = subscriber.Reader; // GWC-15: register this subscriber's channel as a live backlog source instead of // reconciling the queue-depth gauge on every event. The gauge previously read the @@ -151,15 +157,14 @@ public sealed class EventStreamService( while (true) { - MxEvent mxEvent; + bool hasMore; try { - if (!await reader.MoveNextAsync().ConfigureAwait(false)) - { - break; - } - - mxEvent = reader.Current; + // A cleanly completed channel returns false here (end of stream); a channel + // completed WITH a fault rethrows that fault from the wait once the buffer + // is drained — the same surface MoveNextAsync presented, so the terminal + // SessionManagerException(EventQueueOverflow) still propagates unchanged. + hasMore = await reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false); } catch (WorkerClientException workerException) { @@ -173,24 +178,36 @@ public sealed class EventStreamService( throw; } - // Per-RPC filter stays at the subscriber boundary: each request may resume - // from a different AfterWorkerSequence, so the shared pump fans raw events and - // this loop drops the ones at or below the caller's watermark. - if (mxEvent.WorkerSequence <= afterWorkerSequence) + if (!hasMore) { - continue; + break; } - // The queue-depth gauge is maintained lazily via the backlog registration above - // (GWC-15): the metric reads this subscriber's channel Count only when scraped, - // so there is no per-event gauge bookkeeping on this hot path. - yield return mxEvent; + // Drain everything already buffered before waiting again. TryRead never throws; + // a fault left on the channel is observed by the next WaitToReadAsync above. + while (reader.TryRead(out MxEvent? mxEvent)) + { + // Per-RPC filter stays at the subscriber boundary: each request may resume + // from a different AfterWorkerSequence, so the shared pump fans raw events + // and this loop drops the ones at or below the caller's watermark. It + // applies to every live event, drained or awaited alike. + if (mxEvent.WorkerSequence <= afterWorkerSequence) + { + continue; + } + + // The queue-depth gauge is maintained lazily via the backlog registration + // above (GWC-15): the metric reads this subscriber's channel Count only when + // scraped, so there is no per-event gauge bookkeeping on this hot path. + yield return mxEvent; + } } } finally { - await reader.DisposeAsync().ConfigureAwait(false); - + // Nothing to dispose for the reader: consuming the ChannelReader directly means + // there is no enumerator wrapper holding the cancellation registration. + // // Remove this subscriber's live backlog contribution before disposing the lease so // the gauge stops counting a channel that is about to be completed; after this the // gauge reflects only the remaining subscribers (zero when none remain).