fix(grpc): record the real StreamEvents disconnect reason

StreamDisconnected always recorded "Detached". The reason tag is now a bounded EventStreamDisconnectReasons constant for how the stream ended: SessionClosed, Overflow, WorkerFault, ClientCancelled, Error, or Detached.
2026-10-22 06:29:31 -04:00
parent 0b4a1682c8
commit 6ae5ed905f
5 changed files with 260 additions and 5 deletions
@@ -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
        }
    }

+    /// <summary>
+    ///     A stream that ends because the session's event stream completed records the
+    ///     <c>SessionClosed</c> disconnect reason rather than a blanket constant.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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);
+    }
+
+    /// <summary>A per-subscriber overflow records the <c>Overflow</c> disconnect reason.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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<MxEvent> subscriber = service
+            .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None)
+            .GetAsyncEnumerator())
+        {
+            await Assert.ThrowsAsync<SessionManagerException>(
+                async () =>
+                {
+                    while (await subscriber.MoveNextAsync().AsTask().WaitAsync(TestTimeout))
+                    {
+                    }
+                });
+        }
+
+        Assert.Equal([EventStreamDisconnectReasons.Overflow], reasons.Reasons);
+    }
+
+    /// <summary>A terminal worker fault records the <c>WorkerFault</c> disconnect reason.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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<MxEvent> subscriber = service
+            .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None)
+            .GetAsyncEnumerator())
+        {
+            await Assert.ThrowsAsync<WorkerClientException>(
+                async () => await subscriber.MoveNextAsync().AsTask().WaitAsync(TestTimeout));
+        }
+
+        Assert.Equal([EventStreamDisconnectReasons.WorkerFault], reasons.Reasons);
+    }
+
+    /// <summary>A client cancellation records the <c>ClientCancelled</c> disconnect reason.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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<MxEvent> subscriber = service
+            .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, cancellationTokenSource.Token)
+            .GetAsyncEnumerator(cancellationTokenSource.Token))
+        {
+            Task<bool> moveTask = subscriber.MoveNextAsync().AsTask();
+            await WaitUntilAsync(() => session.ActiveEventSubscriberCount == 1);
+            await cancellationTokenSource.CancelAsync();
+            await Assert.ThrowsAnyAsync<OperationCanceledException>(
+                async () => await moveTask.WaitAsync(TestTimeout));
+        }
+
+        Assert.Equal([EventStreamDisconnectReasons.ClientCancelled], reasons.Reasons);
+    }
+
+    /// <summary>
+    ///     A consumer that stops reading without cancelling (the gRPC writer gave up) keeps the
+    ///     <c>Detached</c> reason.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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<MxEvent> subscriber = service
+            .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None)
+            .GetAsyncEnumerator())
+        {
+            await ReadNextAsync(subscriber);
+        }
+
+        Assert.Equal([EventStreamDisconnectReasons.Detached], reasons.Reasons);
+    }
+
    private static async Task<MxEvent> ReadNextAsync(IAsyncEnumerator<MxEvent> enumerator)
    {
        Assert.True(await enumerator.MoveNextAsync().AsTask().WaitAsync(TestTimeout));
@@ -896,4 +1032,63 @@ public sealed class EventStreamServiceTests
            return ValueTask.CompletedTask;
        }
    }
+
+    /// <summary>
+    ///     Captures the <c>reason</c> tag of every <c>mxgateway.grpc.streams.disconnected</c>
+    ///     measurement recorded on one <see cref="GatewayMetrics"/> instance (filtered by meter
+    ///     reference, so parallel tests sharing the meter name cannot cross-talk).
+    /// </summary>
+    private sealed class DisconnectReasonRecorder : IDisposable
+    {
+        private readonly System.Diagnostics.Metrics.MeterListener _listener = new();
+        private readonly List<string> _reasons = [];
+
+        /// <summary>Initializes a new instance of the <see cref="DisconnectReasonRecorder"/> class.</summary>
+        /// <param name="metrics">The metrics instance whose meter to observe.</param>
+        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<long>((instrument, _, tags, _) =>
+            {
+                if (!ReferenceEquals(instrument.Meter, metrics.Meter))
+                {
+                    return;
+                }
+
+                foreach (KeyValuePair<string, object?> tag in tags)
+                {
+                    if (tag.Key == "reason" && tag.Value is string reason)
+                    {
+                        lock (_reasons)
+                        {
+                            _reasons.Add(reason);
+                        }
+                    }
+                }
+            });
+            _listener.Start();
+        }
+
+        /// <summary>Gets the recorded reasons, in order.</summary>
+        public string[] Reasons
+        {
+            get
+            {
+                lock (_reasons)
+                {
+                    return [.. _reasons];
+                }
+            }
+        }
+
+        /// <inheritdoc />
+        public void Dispose() => _listener.Dispose();
+    }
 }