diff --git a/docs/Grpc.md b/docs/Grpc.md index 5c0bf4b..dd1c7d9 100644 --- a/docs/Grpc.md +++ b/docs/Grpc.md @@ -375,7 +375,7 @@ Under `FailFast` the session is faulted so subsequent commands return `FailedPre ### Cancellation and cleanup -The handler creates a linked cancellation token (`streamCts`) so that completing the consumer (client disconnect, error, or graceful end-of-stream) also cancels the producer. The `finally` block cancels the source, disposes the subscriber slot, awaits the producer (swallowing the expected cancellation), and emits `StreamDisconnected("Detached")` so dashboards see the disconnection regardless of cause. +The handler creates a linked cancellation token (`streamCts`) so that completing the consumer (client disconnect, error, or graceful end-of-stream) also cancels the producer. The `finally` block cancels the source, disposes the subscriber slot, awaits the producer (swallowing the expected cancellation), and emits `StreamDisconnected(reason)` so dashboards see every disconnection. The reason is the bounded constant (`EventStreamDisconnectReasons`) for how the stream ended — `SessionClosed`, `Overflow`, `WorkerFault`, `ClientCancelled`, `Error`, or `Detached` — rather than one constant for every cause, so an operator can tell slow consumers from worker faults from routine client disconnects. `WorkerClientException` thrown by the producer marks the session as faulted before completing the channel — the worker is presumed gone, and any subsequent command on that session must observe the fault rather than silently retry. diff --git a/docs/Metrics.md b/docs/Metrics.md index e3b9bf6..1a8f7a3 100644 --- a/docs/Metrics.md +++ b/docs/Metrics.md @@ -45,7 +45,7 @@ All counters are `Counter`. Tag values come from the call sites listed und | `mxgateway.workers.killed` | `reason` | Forced terminations of worker processes. | | `mxgateway.workers.exited` | `reason` | Clean or fault-driven worker exits. | | `mxgateway.heartbeats.failed` | *(none)* | Aggregate worker heartbeat misses. The exported counter carries **no** `session_id` tag — per-session attribution is unbounded cardinality (every session mints a new time series), so it is kept out of the exporter; use the dashboard snapshot or the log scope for per-session detail. | -| `mxgateway.grpc.streams.disconnected` | `reason` | Detachments of the dashboard or client gRPC event stream. | +| `mxgateway.grpc.streams.disconnected` | `reason` | Ends of a client `StreamEvents` stream. `reason` is one of a fixed set (`EventStreamDisconnectReasons`): `SessionClosed` (the session event stream completed), `Overflow` (subscriber fell behind its bounded queue), `WorkerFault` (the worker event stream faulted), `ClientCancelled` (the client cancelled or disconnected), `Error` (unclassified), `Detached` (the consumer stopped reading without cancelling). | | `mxgateway.retries.attempted` | `area` | Resilience retries executed by gateway components. | ### Histograms @@ -181,7 +181,7 @@ IDisposable backlogRegistration = metrics.RegisterEventStreamBacklogSource( () => subscriber.Reader.CanCount ? subscriber.Reader.Count : 0); ... backlogRegistration.Dispose(); -metrics.StreamDisconnected("Detached"); +metrics.StreamDisconnected(disconnectReason); // an EventStreamDisconnectReasons constant ... metrics.QueueOverflow("grpc-event-stream"); metrics.Fault(SessionManagerErrorCode.EventQueueOverflow.ToString()); diff --git a/src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamDisconnectReasons.cs b/src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamDisconnectReasons.cs new file mode 100644 index 0000000..edbba0f --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamDisconnectReasons.cs @@ -0,0 +1,30 @@ +namespace ZB.MOM.WW.MxGateway.Server.Grpc; + +/// +/// The bounded set of reason tag values recorded on +/// mxgateway.grpc.streams.disconnected when a StreamEvents stream ends. +/// +/// +/// Constants rather than exception messages or free text so the metric's tag cardinality stays +/// fixed: an exporter mints one time series per distinct tag value. +/// +public static class EventStreamDisconnectReasons +{ + /// The session's event stream completed cleanly (session closed or the worker stream ended). + public const string SessionClosed = "SessionClosed"; + + /// The subscriber fell behind its bounded queue and was disconnected. + public const string Overflow = "Overflow"; + + /// The worker event stream terminated with a fault. + public const string WorkerFault = "WorkerFault"; + + /// The client cancelled the call or disconnected. + public const string ClientCancelled = "ClientCancelled"; + + /// The stream ended on an unclassified error. + public const string Error = "Error"; + + /// The consumer stopped reading without cancelling the call. + public const string Detached = "Detached"; +} diff --git a/src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamService.cs b/src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamService.cs index 671e933..7d6fafe 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamService.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamService.cs @@ -129,6 +129,10 @@ public sealed class EventStreamService( IDisposable backlogRegistration = metrics.RegisterEventStreamBacklogSource( () => subscriber.Reader.CanCount ? subscriber.Reader.Count : 0); + // Why this stream ended, recorded as a bounded constant on the disconnect metric. Set at + // the point the end is observed; a consumer that simply stops iterating keeps Detached. + string disconnectReason = EventStreamDisconnectReasons.Detached; + try { // Emit order for a resume: the ReplayGap sentinel FIRST (only when events were @@ -173,13 +177,31 @@ public sealed class EventStreamService( // here. Mirror the original ProduceEventsAsync behavior: fault the // session and record the metric, then propagate the terminal fault to the // gRPC client. + disconnectReason = EventStreamDisconnectReasons.WorkerFault; session.MarkFaulted(workerException.Message); metrics.Fault(WorkerClientErrorCode.WorkerFaulted.ToString()); throw; } + catch (SessionManagerException sessionException) + when (sessionException.ErrorCode == SessionManagerErrorCode.EventQueueOverflow) + { + disconnectReason = EventStreamDisconnectReasons.Overflow; + throw; + } + catch (OperationCanceledException) + { + disconnectReason = EventStreamDisconnectReasons.ClientCancelled; + throw; + } + catch (Exception) + { + disconnectReason = EventStreamDisconnectReasons.Error; + throw; + } if (!hasMore) { + disconnectReason = EventStreamDisconnectReasons.SessionClosed; break; } @@ -214,7 +236,15 @@ public sealed class EventStreamService( backlogRegistration.Dispose(); subscriber.Dispose(); - metrics.StreamDisconnected("Detached"); + // A consumer that stopped iterating mid-stream lands here with the Detached default; + // when the call token has fired, that early stop was the client going away. + if (disconnectReason == EventStreamDisconnectReasons.Detached + && cancellationToken.IsCancellationRequested) + { + disconnectReason = EventStreamDisconnectReasons.ClientCancelled; + } + + metrics.StreamDisconnected(disconnectReason); } } diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/EventStreamServiceTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/EventStreamServiceTests.cs index 177e6e9..063dad2 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/EventStreamServiceTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/EventStreamServiceTests.cs @@ -263,7 +263,7 @@ public sealed class EventStreamServiceTests GatewayMetricsSnapshot snapshot = metrics.GetSnapshot(); Assert.Equal(1, snapshot.QueueOverflows); Assert.Equal(1, snapshot.Faults); - // The finally block in StreamEventsAsync calls StreamDisconnected("Detached") on the + // The finally block in StreamEventsAsync records StreamDisconnected (reason Overflow) on the // overflow+fault path too; pin it here so a regression removing that call is caught. Assert.Equal(1, snapshot.StreamDisconnects); } @@ -572,6 +572,142 @@ public sealed class EventStreamServiceTests } } + /// + /// A stream that ends because the session's event stream completed records the + /// SessionClosed disconnect reason rather than a blanket constant. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task StreamEventsAsync_WhenEventStreamCompletes_RecordsSessionClosedReason() + { + FakeWorkerClient workerClient = new(); + using GatewayMetrics metrics = new(); + using DisconnectReasonRecorder reasons = new(metrics); + GatewaySession session = CreateReadySession(workerClient); + EventStreamService service = CreateService(new FakeSessionManager(session), metrics); + workerClient.Events.Add(CreateWorkerEvent(sequence: 1, MxEventFamily.OnDataChange)); + workerClient.CompleteAfterConfiguredEvents = true; + + await CollectEventsAsync(service, session.SessionId); + + Assert.Equal([EventStreamDisconnectReasons.SessionClosed], reasons.Reasons); + } + + /// A per-subscriber overflow records the Overflow disconnect reason. + /// A task that represents the asynchronous operation. + [Fact] + public async Task StreamEventsAsync_WhenStreamQueueOverflows_RecordsOverflowReason() + { + FakeWorkerClient workerClient = new(); + using GatewayMetrics metrics = new(); + using DisconnectReasonRecorder reasons = new(metrics); + GatewaySession session = CreateReadySession( + workerClient, + queueCapacity: 1, + metrics: metrics, + backpressurePolicy: EventBackpressurePolicy.DisconnectSubscriber); + EventStreamService service = CreateService( + new FakeSessionManager(session), + metrics, + queueCapacity: 1, + backpressurePolicy: EventBackpressurePolicy.DisconnectSubscriber); + for (ulong sequence = 1; sequence <= 50; sequence++) + { + workerClient.Events.Add(CreateWorkerEvent(sequence, MxEventFamily.OnDataChange)); + } + + workerClient.CompleteAfterConfiguredEvents = true; + await using (IAsyncEnumerator subscriber = service + .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None) + .GetAsyncEnumerator()) + { + await Assert.ThrowsAsync( + async () => + { + while (await subscriber.MoveNextAsync().AsTask().WaitAsync(TestTimeout)) + { + } + }); + } + + Assert.Equal([EventStreamDisconnectReasons.Overflow], reasons.Reasons); + } + + /// A terminal worker fault records the WorkerFault disconnect reason. + /// A task that represents the asynchronous operation. + [Fact] + public async Task StreamEventsAsync_WhenWorkerEventStreamFaults_RecordsWorkerFaultReason() + { + FakeWorkerClient workerClient = new() + { + TerminalException = new WorkerClientException( + WorkerClientErrorCode.WorkerFaulted, + "worker terminal fault"), + }; + GatewaySession session = CreateReadySession(workerClient); + using GatewayMetrics metrics = new(); + using DisconnectReasonRecorder reasons = new(metrics); + EventStreamService service = CreateService(new FakeSessionManager(session), metrics); + await using (IAsyncEnumerator subscriber = service + .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None) + .GetAsyncEnumerator()) + { + await Assert.ThrowsAsync( + async () => await subscriber.MoveNextAsync().AsTask().WaitAsync(TestTimeout)); + } + + Assert.Equal([EventStreamDisconnectReasons.WorkerFault], reasons.Reasons); + } + + /// A client cancellation records the ClientCancelled disconnect reason. + /// A task that represents the asynchronous operation. + [Fact] + public async Task StreamEventsAsync_WhenCanceled_RecordsClientCancelledReason() + { + FakeWorkerClient workerClient = new(); + GatewaySession session = CreateReadySession(workerClient); + using GatewayMetrics metrics = new(); + using DisconnectReasonRecorder reasons = new(metrics); + EventStreamService service = CreateService(new FakeSessionManager(session), metrics); + using CancellationTokenSource cancellationTokenSource = new(); + await using (IAsyncEnumerator subscriber = service + .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, cancellationTokenSource.Token) + .GetAsyncEnumerator(cancellationTokenSource.Token)) + { + Task moveTask = subscriber.MoveNextAsync().AsTask(); + await WaitUntilAsync(() => session.ActiveEventSubscriberCount == 1); + await cancellationTokenSource.CancelAsync(); + await Assert.ThrowsAnyAsync( + async () => await moveTask.WaitAsync(TestTimeout)); + } + + Assert.Equal([EventStreamDisconnectReasons.ClientCancelled], reasons.Reasons); + } + + /// + /// A consumer that stops reading without cancelling (the gRPC writer gave up) keeps the + /// Detached reason. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task StreamEventsAsync_WhenConsumerStopsEarly_RecordsDetachedReason() + { + FakeWorkerClient workerClient = new(); + GatewaySession session = CreateReadySession(workerClient); + using GatewayMetrics metrics = new(); + using DisconnectReasonRecorder reasons = new(metrics); + EventStreamService service = CreateService(new FakeSessionManager(session), metrics); + workerClient.Events.Add(CreateWorkerEvent(sequence: 1, MxEventFamily.OnDataChange)); + await using (IAsyncEnumerator subscriber = service + .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None) + .GetAsyncEnumerator()) + { + await ReadNextAsync(subscriber); + } + + Assert.Equal([EventStreamDisconnectReasons.Detached], reasons.Reasons); + } + private static async Task ReadNextAsync(IAsyncEnumerator enumerator) { Assert.True(await enumerator.MoveNextAsync().AsTask().WaitAsync(TestTimeout)); @@ -896,4 +1032,63 @@ public sealed class EventStreamServiceTests return ValueTask.CompletedTask; } } + + /// + /// Captures the reason tag of every mxgateway.grpc.streams.disconnected + /// measurement recorded on one instance (filtered by meter + /// reference, so parallel tests sharing the meter name cannot cross-talk). + /// + private sealed class DisconnectReasonRecorder : IDisposable + { + private readonly System.Diagnostics.Metrics.MeterListener _listener = new(); + private readonly List _reasons = []; + + /// Initializes a new instance of the class. + /// The metrics instance whose meter to observe. + public DisconnectReasonRecorder(GatewayMetrics metrics) + { + _listener.InstrumentPublished = (instrument, listener) => + { + if (ReferenceEquals(instrument.Meter, metrics.Meter) + && instrument.Name == "mxgateway.grpc.streams.disconnected") + { + listener.EnableMeasurementEvents(instrument); + } + }; + _listener.SetMeasurementEventCallback((instrument, _, tags, _) => + { + if (!ReferenceEquals(instrument.Meter, metrics.Meter)) + { + return; + } + + foreach (KeyValuePair tag in tags) + { + if (tag.Key == "reason" && tag.Value is string reason) + { + lock (_reasons) + { + _reasons.Add(reason); + } + } + } + }); + _listener.Start(); + } + + /// Gets the recorded reasons, in order. + public string[] Reasons + { + get + { + lock (_reasons) + { + return [.. _reasons]; + } + } + } + + /// + public void Dispose() => _listener.Dispose(); + } }