diff --git a/docs/Metrics.md b/docs/Metrics.md index 1a8f7a3..10cd4a0 100644 --- a/docs/Metrics.md +++ b/docs/Metrics.md @@ -124,7 +124,7 @@ The recording call sites describe the code paths that write into each instrument ```csharp _metrics.SessionOpened(); ... -_metrics.Fault(SessionManagerErrorCode.OpenFailed.ToString()); +_metrics.Fault(SessionManagerErrorCode.OpenFailed.ToString()); // not for a cancelled open or GatewayStopping ... _metrics.SessionClosed(); ... @@ -185,10 +185,10 @@ metrics.StreamDisconnected(disconnectReason); // an EventStreamDisconnectReasons ... metrics.QueueOverflow("grpc-event-stream"); metrics.Fault(SessionManagerErrorCode.EventQueueOverflow.ToString()); -... -metrics.Fault(WorkerClientErrorCode.WorkerFaulted.ToString()); ``` +A worker fault seen by a subscriber records only the `WorkerFault` disconnect reason, not a fault. `WorkerClient` records the fault once when it faults, and every subscriber of the session observes that same fault, so counting it per subscriber made one worker fault read as 1+K faults for K subscribers. + Rather than pushing a running total on every streamed event (which took the subscriber channel's `Count` lock and the metric lock per event), each subscriber registers its channel as a backlog source for the duration of the stream. The gauge callback and `GetSnapshot` sum those sources' `Count` on demand — only when the metric is scraped or projected — so the streaming hot path does no per-event gauge bookkeeping (GWC-15). Disposing the registration in the `finally` removes the subscriber's contribution, so the gauge reflects only the remaining subscribers and returns to zero once the last stream detaches. Negative source returns are clamped to zero when summed. `Grpc/MxAccessGatewayService.cs` records gRPC event send latency around each response-stream write using the allocation-free timestamp API (no per-event `Stopwatch` object): diff --git a/src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamService.cs b/src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamService.cs index 7d6fafe..6b7b11b 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamService.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamService.cs @@ -174,12 +174,13 @@ public sealed class EventStreamService( { // The distributor pump completes every subscriber channel with the source // fault when the worker event stream terminates abnormally; that surfaces - // here. Mirror the original ProduceEventsAsync behavior: fault the - // session and record the metric, then propagate the terminal fault to the - // gRPC client. + // here. Fault the session and propagate the terminal fault to the gRPC + // client. The fault itself is NOT counted here: WorkerClient.SetFaulted + // already recorded it once, and every subscriber of the session sees the + // same fault, so counting it per subscriber inflated mxgateway.faults to + // 1+K. This subscriber's share is the WorkerFault disconnect reason. disconnectReason = EventStreamDisconnectReasons.WorkerFault; session.MarkFaulted(workerException.Message); - metrics.Fault(WorkerClientErrorCode.WorkerFaulted.ToString()); throw; } catch (SessionManagerException sessionException) 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 063dad2..85b305c 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/EventStreamServiceTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/EventStreamServiceTests.cs @@ -358,7 +358,63 @@ public sealed class EventStreamServiceTests Assert.Equal(WorkerClientErrorCode.WorkerFaulted, exception.ErrorCode); Assert.Equal(SessionState.Faulted, session.State); + + // The worker fault is counted once, by the worker client that faulted (this fake does + // not count it); a subscriber observing it records a disconnect reason, not a fault. + Assert.Equal(0, metrics.GetSnapshot().Faults); + } + + /// + /// One worker fault is one mxgateway.faults increment however many + /// StreamEvents subscribers observe it: records it when it + /// faults, and each subscriber records only its WorkerFault disconnect reason. + /// Counting it per subscriber as well inflated the metric to 1+K for K subscribers. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task StreamEventsAsync_WhenWorkerFaultsUnderTwoSubscribers_CountsTheFaultOnce() + { + await using Workers.Fakes.FakeWorkerHarness fakeWorker = await Workers.Fakes.FakeWorkerHarness.CreateConnectedPairAsync(); + using GatewayMetrics metrics = new(); + using DisconnectReasonRecorder reasons = new(metrics); + WorkerClient client = fakeWorker.CreateClient(metrics: metrics); + Task startTask = client.StartAsync(CancellationToken.None); + await fakeWorker.CompleteStartupAsync(); + await startTask.WaitAsync(TestTimeout); + GatewayOptions options = new() + { + Sessions = new SessionOptions { AllowMultipleEventSubscribers = true }, + }; + SessionManager manager = new( + new SessionRegistry(), + new SingleClientFactory(client), + Options.Create(options), + metrics); + GatewaySession session = await manager.OpenSessionAsync( + new SessionOpenRequest(RequestedBackend: null, ClientSessionName: "fault-count", ClientCorrelationId: null, CommandTimeout: null), + "client-1", + ownerKeyId: null, + CancellationToken.None); + EventStreamService service = new(manager, Options.Create(options), metrics); + await using IAsyncEnumerator first = service + .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None) + .GetAsyncEnumerator(); + await using IAsyncEnumerator second = service + .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None) + .GetAsyncEnumerator(); + Task firstMove = first.MoveNextAsync().AsTask(); + Task secondMove = second.MoveNextAsync().AsTask(); + await WaitUntilAsync(() => session.ActiveEventSubscriberCount == 2); + + await fakeWorker.DisposeWorkerSideAsync(); + + await Assert.ThrowsAsync(async () => await firstMove.WaitAsync(TestTimeout)); + await Assert.ThrowsAsync(async () => await secondMove.WaitAsync(TestTimeout)); + Assert.Equal(WorkerClientState.Faulted, client.State); Assert.Equal(1, metrics.GetSnapshot().Faults); + Assert.Equal( + [EventStreamDisconnectReasons.WorkerFault, EventStreamDisconnectReasons.WorkerFault], + reasons.Reasons); } /// @@ -1038,6 +1094,15 @@ public sealed class EventStreamServiceTests /// measurement recorded on one instance (filtered by meter /// reference, so parallel tests sharing the meter name cannot cross-talk). /// + private sealed class SingleClientFactory(IWorkerClient client) : ISessionWorkerClientFactory + { + /// + public Task CreateAsync(GatewaySession session, CancellationToken cancellationToken) + { + return Task.FromResult(client); + } + } + private sealed class DisconnectReasonRecorder : IDisposable { private readonly System.Diagnostics.Metrics.MeterListener _listener = new();