fix(grpc): count a worker fault once, not once per StreamEvents subscriber

WorkerClient.SetFaulted already records mxgateway.faults; EventStreamService added a
WorkerFaulted fault per subscriber, so one fault read as 1+K. The subscriber now records
only its WorkerFault disconnect reason. docs/Metrics.md updated.
This commit is contained in:
Joseph Doherty
2026-10-22 08:42:35 -04:00
parent a178d68b60
commit a0b5e91227
3 changed files with 73 additions and 7 deletions
+3 -3
View File
@@ -124,7 +124,7 @@ The recording call sites describe the code paths that write into each instrument
```csharp ```csharp
_metrics.SessionOpened(); _metrics.SessionOpened();
... ...
_metrics.Fault(SessionManagerErrorCode.OpenFailed.ToString()); _metrics.Fault(SessionManagerErrorCode.OpenFailed.ToString()); // not for a cancelled open or GatewayStopping
... ...
_metrics.SessionClosed(); _metrics.SessionClosed();
... ...
@@ -185,10 +185,10 @@ metrics.StreamDisconnected(disconnectReason); // an EventStreamDisconnectReasons
... ...
metrics.QueueOverflow("grpc-event-stream"); metrics.QueueOverflow("grpc-event-stream");
metrics.Fault(SessionManagerErrorCode.EventQueueOverflow.ToString()); 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. 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): `Grpc/MxAccessGatewayService.cs` records gRPC event send latency around each response-stream write using the allocation-free timestamp API (no per-event `Stopwatch` object):
@@ -174,12 +174,13 @@ public sealed class EventStreamService(
{ {
// The distributor pump completes every subscriber channel with the source // The distributor pump completes every subscriber channel with the source
// fault when the worker event stream terminates abnormally; that surfaces // fault when the worker event stream terminates abnormally; that surfaces
// here. Mirror the original ProduceEventsAsync behavior: fault the // here. Fault the session and propagate the terminal fault to the gRPC
// session and record the metric, then propagate the terminal fault to the // client. The fault itself is NOT counted here: WorkerClient.SetFaulted
// gRPC client. // 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; disconnectReason = EventStreamDisconnectReasons.WorkerFault;
session.MarkFaulted(workerException.Message); session.MarkFaulted(workerException.Message);
metrics.Fault(WorkerClientErrorCode.WorkerFaulted.ToString());
throw; throw;
} }
catch (SessionManagerException sessionException) catch (SessionManagerException sessionException)
@@ -358,7 +358,63 @@ public sealed class EventStreamServiceTests
Assert.Equal(WorkerClientErrorCode.WorkerFaulted, exception.ErrorCode); Assert.Equal(WorkerClientErrorCode.WorkerFaulted, exception.ErrorCode);
Assert.Equal(SessionState.Faulted, session.State); 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);
}
/// <summary>
/// One worker fault is one <c>mxgateway.faults</c> increment however many
/// <c>StreamEvents</c> subscribers observe it: <see cref="WorkerClient"/> records it when it
/// faults, and each subscriber records only its <c>WorkerFault</c> disconnect reason.
/// Counting it per subscriber as well inflated the metric to 1+K for K subscribers.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[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<MxEvent> first = service
.StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None)
.GetAsyncEnumerator();
await using IAsyncEnumerator<MxEvent> second = service
.StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None)
.GetAsyncEnumerator();
Task<bool> firstMove = first.MoveNextAsync().AsTask();
Task<bool> secondMove = second.MoveNextAsync().AsTask();
await WaitUntilAsync(() => session.ActiveEventSubscriberCount == 2);
await fakeWorker.DisposeWorkerSideAsync();
await Assert.ThrowsAsync<WorkerClientException>(async () => await firstMove.WaitAsync(TestTimeout));
await Assert.ThrowsAsync<WorkerClientException>(async () => await secondMove.WaitAsync(TestTimeout));
Assert.Equal(WorkerClientState.Faulted, client.State);
Assert.Equal(1, metrics.GetSnapshot().Faults); Assert.Equal(1, metrics.GetSnapshot().Faults);
Assert.Equal(
[EventStreamDisconnectReasons.WorkerFault, EventStreamDisconnectReasons.WorkerFault],
reasons.Reasons);
} }
/// <summary> /// <summary>
@@ -1038,6 +1094,15 @@ public sealed class EventStreamServiceTests
/// measurement recorded on one <see cref="GatewayMetrics"/> instance (filtered by meter /// measurement recorded on one <see cref="GatewayMetrics"/> instance (filtered by meter
/// reference, so parallel tests sharing the meter name cannot cross-talk). /// reference, so parallel tests sharing the meter name cannot cross-talk).
/// </summary> /// </summary>
private sealed class SingleClientFactory(IWorkerClient client) : ISessionWorkerClientFactory
{
/// <inheritdoc />
public Task<IWorkerClient> CreateAsync(GatewaySession session, CancellationToken cancellationToken)
{
return Task.FromResult(client);
}
}
private sealed class DisconnectReasonRecorder : IDisposable private sealed class DisconnectReasonRecorder : IDisposable
{ {
private readonly System.Diagnostics.Metrics.MeterListener _listener = new(); private readonly System.Diagnostics.Metrics.MeterListener _listener = new();