perf(grpc): consume the subscriber channel directly in StreamEventsAsync — drops the ReadAllAsync iterator hop

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<MxEvent> 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<MxEvent> 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;
                 }
 
+                if (!hasMore)
+                {
+                    break;
+                }
+
+                // 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.
+                    // 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.
+                    // 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).