fix(GWC-33): unregister terminal worker queue gauge

This commit is contained in:
Joseph Doherty
2026-08-25 10:56:23 -04:00
parent f2090fe1b5
commit b0156c685e
2 changed files with 49 additions and 3 deletions
@@ -444,11 +444,10 @@ public sealed class WorkerClient : IWorkerClient
{
_state = WorkerClientState.Closed;
}
UnregisterEventQueueDepthSource();
}
// Drop out of the worker queue-depth gauge before teardown: whatever this client still holds
// is about to be discarded, and the sum must not keep counting a client that is going away.
_eventQueueDepthRegistration?.Dispose();
KillOwnedProcess("Dispose");
_stopCts.Cancel();
_outboundEnvelopes.Writer.TryComplete();
@@ -1092,6 +1091,7 @@ public sealed class WorkerClient : IWorkerClient
}
_state = WorkerClientState.Closed;
UnregisterEventQueueDepthSource();
}
_stopCts.Cancel();
@@ -1126,6 +1126,7 @@ public sealed class WorkerClient : IWorkerClient
}
_state = WorkerClientState.Faulted;
UnregisterEventQueueDepthSource();
}
_stopCts.Cancel();
@@ -1139,6 +1140,16 @@ public sealed class WorkerClient : IWorkerClient
_logger.LogWarning(exception, "Worker client faulted for session {SessionId}: {Message}", SessionId, message);
}
/// <summary>
/// Removes this terminal client's abandoned backlog from the gateway-wide queue-depth
/// gauge. The registration handle is idempotent, so terminal transitions and later disposal
/// may all call this safely while the faulted client remains retained for diagnostics.
/// </summary>
private void UnregisterEventQueueDepthSource()
{
_eventQueueDepthRegistration?.Dispose();
}
private void KillOwnedProcess(string reason)
{
WorkerProcessHandle? processHandle = _connection.ProcessHandle;
@@ -1147,6 +1147,41 @@ public sealed class WorkerClientTests
Assert.Equal(0, metrics.GetSnapshot().WorkerEventQueueDepth);
}
/// <summary>
/// GWC-33: a faulted client can remain retained for diagnostics, but its abandoned event
/// backlog must stop contributing to the gateway-wide queue-depth gauge as soon as the
/// client enters a terminal state rather than waiting for disposal.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task WorkerEventQueueDepthGauge_WhenClientFaults_StopsCountingRetainedClient()
{
using GatewayMetrics metrics = new();
await using PipePair pipePair = await PipePair.CreateAsync();
await using WorkerClient client = CreateClient(
pipePair,
new WorkerClientOptions
{
EventChannelCapacity = 4,
EventChannelFullModeTimeout = TimeSpan.FromMinutes(5),
HeartbeatGrace = TimeSpan.FromSeconds(30),
HeartbeatCheckInterval = TimeSpan.FromSeconds(30),
},
metrics: metrics);
await CompleteHandshakeAsync(client, pipePair);
await pipePair.WriteAsync(CreateEventEnvelope(sequence: 1, MxEventFamily.OnDataChange));
await WaitUntilAsync(
() => metrics.GetSnapshot().WorkerEventQueueDepth == 1,
TestTimeout);
await pipePair.WriteAsync(CreateWorkerFaultEnvelope("scripted retained-client fault"));
await WaitUntilAsync(() => client.State == WorkerClientState.Faulted, TestTimeout);
Assert.Equal(WorkerClientState.Faulted, client.State);
Assert.Equal(0, metrics.GetSnapshot().WorkerEventQueueDepth);
}
private static WorkerClient CreateClient(
PipePair pipePair,
WorkerClientOptions? options = null,