diff --git a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs index bff47b1..6f32caa 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs @@ -418,12 +418,32 @@ public sealed class WorkerClient : IWorkerClient /// A task that represents the asynchronous operation. 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 // is about to be discarded, and the sum must not keep counting a client that is going away. _eventQueueDepthRegistration?.Dispose(); @@ -432,6 +452,16 @@ public sealed class WorkerClient : IWorkerClient _outboundEnvelopes.Writer.TryComplete(); _eventStaging.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( new WorkerClientException( 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. /// TryWrite is non-blocking, so the read loop keeps dispatching replies, heartbeats and /// faults regardless of how backed up the event path is. It returns false in two cases: - /// the staging channel has been completed during shutdown (the event is safely dropped because - /// the client is already terminal), or the channel is full at its + /// the staging channel has been completed by one of the three completing paths — + /// , or , 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 × bound. The latter means 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) — diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs index 807249d..ddd732c 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs @@ -14,6 +14,10 @@ public sealed class WorkerClientTests private const string SessionId = "session-worker-client"; private const string Nonce = "nonce-worker-client"; private const int WorkerProcessId = 4321; + + /// Upper bound on the events writes. + private const ulong EmittedEventCeiling = 2000; + private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(5); /// Verifies that StartAsync enters ready state after receiving worker hello and ready messages. @@ -560,6 +564,96 @@ public sealed class WorkerClientTests Assert.True(process.Disposed); } + /// + /// GWC-32: disposing a still-Ready client while the worker is emitting events must be + /// an ordinary teardown, not a fault. DisposeAsync completes the same three channels + /// MarkClosed and SetFaulted do, so it must enter a terminal state first; + /// otherwise the read loop — already inside DispatchEnvelope 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 + /// ProtocolViolation fault, a worker-event-staging 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 DisposeAsync awaits its background + /// tasks, so the metrics read afterwards is not racing a late fault. + /// + /// A task that represents the asynchronous operation. + [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); + } + + /// + /// Writes worker events from the fake worker side until cancelled or until + /// 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. + /// + /// The connected pipe pair whose worker side emits the events. + /// Stops the emission. + /// A task that represents the asynchronous operation. + 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. + } + } + /// /// Verifies that a heartbeat envelope updates the last-heartbeat timestamp and worker /// process id. Uses a so the timestamp advance is