fix(GWC-32): make WorkerClient disposal a terminal transition
DisposeAsync is the third path that completes the outbound, staging and event
channels, but unlike MarkClosed and SetFaulted it left _state untouched. A
client disposed while still Ready therefore kept reporting
IsTerminalState() == false, so a read loop already inside DispatchEnvelope when
the staging channel was completed took StageWorkerEvent's overflow arm: a
spurious ProtocolViolation fault, a worker-event-staging overflow metric, a
redundant kill and a WorkerStopped("ProtocolViolation") metric during an
intentional teardown — on exactly the path an operator inspects after a failed
SessionManager.OpenSessionAsync.
DisposeAsync now takes _syncRoot once, claims _disposed there, and transitions
to Closed only when the state is not already Closed or Faulted, so a
dispose-after-fault keeps Faulted and a double dispose is a no-op. Because the
transition is now real, the dispose path records the worker stop through the
existing RecordWorkerStoppedOnce latch, which also stops the running-worker
gauge leaking when a Ready client is disposed without a prior close or fault.
MarkClosed is deliberately not reused: it would overwrite Faulted with Closed
and would re-run a teardown sequence DisposeAsync performs itself.
The StageWorkerEvent XML doc now names all three completion paths.
Test: DisposeAsync_WhileWorkerIsEmittingEvents_ClosesWithoutFaultingOrOverflowing
disposes a Ready client mid-event-stream and asserts Closed state, zero faults,
zero queue overflows, zero kills and exactly one recorded worker exit. It fails
on the pre-fix code.
This commit is contained in:
@@ -418,12 +418,32 @@ public sealed class WorkerClient : IWorkerClient
|
|||||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||||
public async ValueTask DisposeAsync()
|
public async ValueTask DisposeAsync()
|
||||||
{
|
{
|
||||||
if (_disposed)
|
// GWC-32: disposal is the third channel-completing path, and it must enter a terminal state
|
||||||
|
// like the other two (MarkClosed, SetFaulted) BEFORE the channels are completed. Otherwise a
|
||||||
|
// client disposed while still Ready keeps reporting IsTerminalState() == false, and a read
|
||||||
|
// loop already inside DispatchEnvelope (the _stopCts.Cancel below only interrupts the NEXT
|
||||||
|
// ReadAsync) takes StageWorkerEvent's overflow arm on the completed staging channel: a
|
||||||
|
// spurious ProtocolViolation fault, a worker-event-staging overflow metric and a redundant
|
||||||
|
// kill, all during an intentional teardown. MarkClosed is deliberately not reused here: it
|
||||||
|
// would overwrite Faulted with Closed and it re-runs the whole teardown sequence this method
|
||||||
|
// performs itself. The transition is skipped when the state is already terminal, so a
|
||||||
|
// dispose-after-fault keeps Faulted and a double dispose changes nothing.
|
||||||
|
bool becameTerminalHere;
|
||||||
|
lock (_syncRoot)
|
||||||
{
|
{
|
||||||
return;
|
if (_disposed)
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
_disposed = true;
|
||||||
|
becameTerminalHere = _state is not (WorkerClientState.Closed or WorkerClientState.Faulted);
|
||||||
|
if (becameTerminalHere)
|
||||||
|
{
|
||||||
|
_state = WorkerClientState.Closed;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
_disposed = true;
|
|
||||||
// Drop out of the worker queue-depth gauge before teardown: whatever this client still holds
|
// 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.
|
// is about to be discarded, and the sum must not keep counting a client that is going away.
|
||||||
_eventQueueDepthRegistration?.Dispose();
|
_eventQueueDepthRegistration?.Dispose();
|
||||||
@@ -432,6 +452,16 @@ public sealed class WorkerClient : IWorkerClient
|
|||||||
_outboundEnvelopes.Writer.TryComplete();
|
_outboundEnvelopes.Writer.TryComplete();
|
||||||
_eventStaging.Writer.TryComplete();
|
_eventStaging.Writer.TryComplete();
|
||||||
_events.Writer.TryComplete();
|
_events.Writer.TryComplete();
|
||||||
|
if (becameTerminalHere)
|
||||||
|
{
|
||||||
|
// A worker that reached Ready and is now being disposed without a prior Close or fault
|
||||||
|
// (SessionManager.OpenSessionAsync's failure arm does exactly this) stopped, so the
|
||||||
|
// running-worker gauge must come down. RecordWorkerStoppedOnce is latched on
|
||||||
|
// _workerStartRecorded, so a client that already recorded its stop through MarkClosed or
|
||||||
|
// SetFaulted cannot record a second one here.
|
||||||
|
RecordWorkerStoppedOnce("dispose");
|
||||||
|
}
|
||||||
|
|
||||||
CompletePendingCommands(
|
CompletePendingCommands(
|
||||||
new WorkerClientException(
|
new WorkerClientException(
|
||||||
WorkerClientErrorCode.GatewayShutdown,
|
WorkerClientErrorCode.GatewayShutdown,
|
||||||
@@ -633,8 +663,10 @@ public sealed class WorkerClient : IWorkerClient
|
|||||||
/// Hands a received worker event to the dedicated event writer without blocking the read loop.
|
/// Hands a received worker event to the dedicated event writer without blocking the read loop.
|
||||||
/// <c>TryWrite</c> is non-blocking, so the read loop keeps dispatching replies, heartbeats and
|
/// <c>TryWrite</c> is non-blocking, so the read loop keeps dispatching replies, heartbeats and
|
||||||
/// faults regardless of how backed up the event path is. It returns <c>false</c> in two cases:
|
/// faults regardless of how backed up the event path is. It returns <c>false</c> in two cases:
|
||||||
/// the staging channel has been completed during shutdown (the event is safely dropped because
|
/// the staging channel has been completed by one of the three completing paths —
|
||||||
/// the client is already terminal), or the channel is full at its
|
/// <see cref="MarkClosed"/>, <see cref="SetFaulted"/> or <see cref="DisposeAsync"/>, each of
|
||||||
|
/// which enters a terminal state before completing the channels (GWC-32), so the event is
|
||||||
|
/// safely dropped because the client is already terminal — or the channel is full at its
|
||||||
/// 2 × <see cref="WorkerClientOptions.EventChannelCapacity"/> bound. The latter means the
|
/// 2 × <see cref="WorkerClientOptions.EventChannelCapacity"/> bound. The latter means the
|
||||||
/// consumer has been draining slower than the worker produces for the whole time it took the
|
/// consumer has been draining slower than the worker produces for the whole time it took the
|
||||||
/// worker to emit that many further events, so the client is faulted immediately (GWC-24) —
|
/// worker to emit that many further events, so the client is faulted immediately (GWC-24) —
|
||||||
|
|||||||
@@ -14,6 +14,10 @@ public sealed class WorkerClientTests
|
|||||||
private const string SessionId = "session-worker-client";
|
private const string SessionId = "session-worker-client";
|
||||||
private const string Nonce = "nonce-worker-client";
|
private const string Nonce = "nonce-worker-client";
|
||||||
private const int WorkerProcessId = 4321;
|
private const int WorkerProcessId = 4321;
|
||||||
|
|
||||||
|
/// <summary>Upper bound on the events <see cref="EmitEventsUntilStoppedAsync"/> writes.</summary>
|
||||||
|
private const ulong EmittedEventCeiling = 2000;
|
||||||
|
|
||||||
private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(5);
|
private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(5);
|
||||||
|
|
||||||
/// <summary>Verifies that StartAsync enters ready state after receiving worker hello and ready messages.</summary>
|
/// <summary>Verifies that StartAsync enters ready state after receiving worker hello and ready messages.</summary>
|
||||||
@@ -560,6 +564,96 @@ public sealed class WorkerClientTests
|
|||||||
Assert.True(process.Disposed);
|
Assert.True(process.Disposed);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// GWC-32: disposing a still-<c>Ready</c> client while the worker is emitting events must be
|
||||||
|
/// an ordinary teardown, not a fault. <c>DisposeAsync</c> completes the same three channels
|
||||||
|
/// <c>MarkClosed</c> and <c>SetFaulted</c> do, so it must enter a terminal state first;
|
||||||
|
/// otherwise the read loop — already inside <c>DispatchEnvelope</c> when the staging channel
|
||||||
|
/// is completed, or reading a stream that is torn down underneath it — misdiagnoses the
|
||||||
|
/// completed channel as the GWC-24 sustained-backpressure overflow and raises a spurious
|
||||||
|
/// <c>ProtocolViolation</c> fault, a <c>worker-event-staging</c> overflow metric and a
|
||||||
|
/// redundant kill during an intentional dispose. The event channel is sized far above the
|
||||||
|
/// burst so a genuine overflow is impossible, and <c>DisposeAsync</c> awaits its background
|
||||||
|
/// tasks, so the metrics read afterwards is not racing a late fault.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task DisposeAsync_WhileWorkerIsEmittingEvents_ClosesWithoutFaultingOrOverflowing()
|
||||||
|
{
|
||||||
|
// The burst is capped at EmittedEventCeiling and the staging channel holds 2 x capacity, so
|
||||||
|
// even if every emitted event were staged and none consumed the GWC-24 bound is unreachable:
|
||||||
|
// an overflow metric in this test can only be the dispose-race misdiagnosis.
|
||||||
|
const int capacity = 2048;
|
||||||
|
using GatewayMetrics metrics = new();
|
||||||
|
await using PipePair pipePair = await PipePair.CreateAsync();
|
||||||
|
WorkerClient client = CreateClient(
|
||||||
|
pipePair,
|
||||||
|
new WorkerClientOptions
|
||||||
|
{
|
||||||
|
EventChannelCapacity = capacity,
|
||||||
|
EventChannelFullModeTimeout = TimeSpan.FromMinutes(5),
|
||||||
|
HeartbeatGrace = TimeSpan.FromSeconds(30),
|
||||||
|
HeartbeatCheckInterval = TimeSpan.FromSeconds(30),
|
||||||
|
},
|
||||||
|
metrics: metrics);
|
||||||
|
await CompleteHandshakeAsync(client, pipePair);
|
||||||
|
|
||||||
|
using CancellationTokenSource emitCts = new(TestTimeout);
|
||||||
|
Task emitTask = EmitEventsUntilStoppedAsync(pipePair, emitCts.Token);
|
||||||
|
|
||||||
|
// Proves the read loop is mid-stream — not parked before its first event — when dispose runs.
|
||||||
|
await WaitUntilAsync(() => metrics.GetSnapshot().WorkerEventQueueDepth > 0, TestTimeout);
|
||||||
|
|
||||||
|
await client.DisposeAsync().AsTask().WaitAsync(TestTimeout);
|
||||||
|
await emitCts.CancelAsync();
|
||||||
|
await emitTask.WaitAsync(TestTimeout);
|
||||||
|
|
||||||
|
Assert.Equal(WorkerClientState.Closed, client.State);
|
||||||
|
|
||||||
|
GatewayMetricsSnapshot snapshot = metrics.GetSnapshot();
|
||||||
|
Assert.Equal(0L, snapshot.Faults);
|
||||||
|
Assert.Equal(0L, snapshot.QueueOverflows);
|
||||||
|
Assert.Equal(0L, snapshot.WorkerKills);
|
||||||
|
|
||||||
|
// Disposal is now a terminal transition, so the worker that reached Ready is recorded as
|
||||||
|
// stopped exactly once — the running-worker gauge no longer leaks on this path.
|
||||||
|
Assert.Equal(0, snapshot.WorkersRunning);
|
||||||
|
Assert.Equal(1L, snapshot.WorkerExits);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Writes worker events from the fake worker side until cancelled or until
|
||||||
|
/// <see cref="EmittedEventCeiling"/> events have been written, so the gateway read loop has a
|
||||||
|
/// continuous stream to dispatch while a test tears the client down. The ceiling keeps the
|
||||||
|
/// burst structurally below the staging bound the caller configures. Every failure is
|
||||||
|
/// swallowed: disposal closes the gateway end of the pipe underneath this writer, which is
|
||||||
|
/// the situation under test rather than a test failure.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="pipePair">The connected pipe pair whose worker side emits the events.</param>
|
||||||
|
/// <param name="cancellationToken">Stops the emission.</param>
|
||||||
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||||
|
private static async Task EmitEventsUntilStoppedAsync(
|
||||||
|
PipePair pipePair,
|
||||||
|
CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
await Task.Yield();
|
||||||
|
try
|
||||||
|
{
|
||||||
|
for (ulong sequence = 1;
|
||||||
|
sequence <= EmittedEventCeiling && !cancellationToken.IsCancellationRequested;
|
||||||
|
sequence++)
|
||||||
|
{
|
||||||
|
await pipePair.WorkerWriter.WriteAsync(
|
||||||
|
CreateEventEnvelope(sequence, MxEventFamily.OnDataChange),
|
||||||
|
cancellationToken);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
catch (Exception)
|
||||||
|
{
|
||||||
|
// Broken pipe, disposed stream or cancellation — all expected once dispose starts.
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Verifies that a heartbeat envelope updates the last-heartbeat timestamp and worker
|
/// Verifies that a heartbeat envelope updates the last-heartbeat timestamp and worker
|
||||||
/// process id. Uses a <see cref="ManualTimeProvider"/> so the timestamp advance is
|
/// process id. Uses a <see cref="ManualTimeProvider"/> so the timestamp advance is
|
||||||
|
|||||||
Reference in New Issue
Block a user