From b27035b3a6518507038c2d25d762acd1e069fcc1 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 20 Aug 2026 09:31:29 -0400 Subject: [PATCH] fix(IPC-34): forward a WorkerCancel when the caller abandons a command, not only on gateway timeout IPC-08 closed with WorkerCancel sent from the gateway's own timeout arm only. Its rationale -- the gateway has stopped waiting but the worker has not stopped working, so a queued-but-not-started command should be dropped instead of occupying the single STA behind a caller that is already gone -- applies exactly as well to caller cancellation, which is the common abandonment path: gRPC clients routinely set per-call deadlines shorter than the gateway's CommandTimeout, and a client disconnect cancels the RPC's token. Both landed in arms that removed the pending entry and returned silently, leaving the correlation to execute in its turn. Under a retry storm against a slow provider the STA works through dead commands while live callers queue behind them -- precisely the backlog the cancel exists to shorten. - TrySendCancelForTimedOutCommand is now TrySendCancelForAbandonedCommand and takes the reason string, since it serves both arms. - Both caller-cancellation arms route through AbandonCommandOnCallerCancellation, which sends the cancel under the same `removed` ownership gate as the timeout arm (so a reply that won the race is never chased) and skips it once _stopCts has fired, where the outbound channel is being completed and every pending command is failed anyway. - New WorkerClientErrorCode.CallerCancelled replaces the GatewayShutdown label those arms used, so metrics and logs stop recording a shutdown that did not happen. It maps to StatusCode.Cancelled, the same status GatewayShutdown maps to, so nothing changes for clients. Docs: docs/Grpc.md mapping line, gateway.md cancellation rules, docs/GatewayProcessDesign.md pending-command section. Tests: InvokeAsync_WhenCallerCancels_SendsWorkerCancelForThatCorrelation (twin of the timeout test, driven by the caller token; verified failing without the send) and InvokeAsync_WhenClientStopsUnderInFlightCommand_DoesNotSendWorkerCancel. --- docs/GatewayProcessDesign.md | 23 +++- docs/Grpc.md | 2 +- gateway.md | 11 +- .../Grpc/MxAccessGatewayService.cs | 1 + .../Workers/WorkerClient.cs | 83 ++++++++++---- .../Workers/WorkerClientErrorCode.cs | 8 ++ .../Gateway/Workers/WorkerClientTests.cs | 106 ++++++++++++++++++ 7 files changed, 202 insertions(+), 32 deletions(-) diff --git a/docs/GatewayProcessDesign.md b/docs/GatewayProcessDesign.md index 0fe7cd9..035a9e1 100644 --- a/docs/GatewayProcessDesign.md +++ b/docs/GatewayProcessDesign.md @@ -593,9 +593,19 @@ Pending command handling: Timeouts should not assume the COM call stopped. A timed-out command may still finish inside the worker. -On timeout the client also forwards a `WorkerCancel` carrying the abandoned -correlation id, best-effort: the gateway has stopped waiting, but the worker has -not stopped working, and the worker owns a single STA. `WorkerPipeSession` routes +On timeout **and on caller cancellation** the client also forwards a +`WorkerCancel` carrying the abandoned correlation id, best-effort: the gateway +has stopped waiting, but the worker has not stopped working, and the worker owns +a single STA. Caller cancellation — a per-call deadline shorter than the +gateway's command timeout, or a client disconnect cancelling the RPC's token — is +the more common of the two paths, and under a retry storm it is exactly the +backlog the cancel exists to shorten, so both arms send it; the pending command +is failed as `CallerCancelled` rather than `GatewayShutdown` so metrics and logs +name the real cause (both still map to gRPC `Cancelled`). The send is gated on +the same ownership check as the timeout arm — only the arm that removed the +pending entry sends, so a reply that won the race is never chased — and is +skipped when the client's own stop token has fired, since the outbound channel is +being completed and every pending command is failed anyway. `WorkerPipeSession` routes the cancel to `CancelCommand`, which drops the correlation from the STA queue if it has not started and replies `Canceled` for it. A cancel that arrives after the command reached MXAccess is a no-op — there is no way to abort an in-flight COM @@ -606,9 +616,10 @@ the worker reads the command and then its cancel and drops it before execution. Cancels ride the same outbound channel as commands, whose capacity is `MaxPendingCommands + 4`: the reserve above the pending-command limit is what -absorbs them, so a burst of timeouts stays bounded and cannot deadlock the -enqueue path. Failing to send the cancel is logged at debug and never replaces -the `CommandTimeout` the caller is owed. +absorbs them, so a burst of timeouts or cancellations stays bounded and cannot +deadlock the enqueue path. Failing to send the cancel is logged at debug and +never replaces the `CommandTimeout` — or the `OperationCanceledException` — the +caller is owed. Cancellation outranks the deadline. When a caller's token is canceled around the same time the timeout fires, the command is reported as canceled diff --git a/docs/Grpc.md b/docs/Grpc.md index e548e91..a09d5cc 100644 --- a/docs/Grpc.md +++ b/docs/Grpc.md @@ -282,7 +282,7 @@ StatusCode statusCode = exception.ErrorCode switch }; ``` -`WorkerClientException` follows the same pattern: `CommandTimeout` becomes `DeadlineExceeded`, `GatewayShutdown` becomes `Cancelled`, `InvalidState` becomes `FailedPrecondition`, `ProtocolViolation` becomes `Internal`, `CommandTooLarge` becomes `ResourceExhausted`, and unmapped codes fall through to `Unavailable`. +`WorkerClientException` follows the same pattern: `CommandTimeout` becomes `DeadlineExceeded`, `GatewayShutdown` and `CallerCancelled` both become `Cancelled`, `InvalidState` becomes `FailedPrecondition`, `ProtocolViolation` becomes `Internal`, `CommandTooLarge` becomes `ResourceExhausted`, and unmapped codes fall through to `Unavailable`. `CallerCancelled` is the caller's own token firing — a per-call deadline shorter than the gateway command timeout, or a disconnect — and is separate from `GatewayShutdown` only so metrics and logs stop attributing those calls to a shutdown that did not happen; the status the client observes is the same. ## Event Streaming Model diff --git a/gateway.md b/gateway.md index 18fbb9b..08e03a0 100644 --- a/gateway.md +++ b/gateway.md @@ -917,10 +917,19 @@ Timeouts: Cancellation: -- canceling the gRPC call should stop waiting in the gateway, +- canceling the gRPC call stops the gateway waiting **and** forwards a + `WorkerCancel` for the abandoned correlation, the same best-effort frame the + command timeout sends — a client deadline or a disconnect is the common + abandonment path, and leaving the correlation on the worker's single STA queue + makes live callers queue behind work nobody is waiting for, +- the cancel is sent only by the caller that owned the pending-entry removal (so + a reply that won the race is never chased) and is skipped once the client is + stopping, where every pending command fails anyway, - it cannot safely abort an in-flight COM call on the STA, - the worker should finish the COM call and discard or log the late reply if the correlation was canceled, +- caller cancellation is reported as `CallerCancelled` (mapped to gRPC + `Cancelled`), distinct from a gateway shutdown, so metrics name the real cause, - hard cancellation means killing the worker process. ## Event Delivery And Backpressure diff --git a/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs b/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs index 4481c1d..a10699c 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs @@ -1051,6 +1051,7 @@ public sealed class MxAccessGatewayService( { WorkerClientErrorCode.CommandTimeout => StatusCode.DeadlineExceeded, WorkerClientErrorCode.GatewayShutdown => StatusCode.Cancelled, + WorkerClientErrorCode.CallerCancelled => StatusCode.Cancelled, WorkerClientErrorCode.InvalidState => StatusCode.FailedPrecondition, WorkerClientErrorCode.ProtocolViolation => StatusCode.Internal, WorkerClientErrorCode.CommandTooLarge => StatusCode.ResourceExhausted, diff --git a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs index ce83155..0044abc 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs @@ -291,7 +291,10 @@ public sealed class WorkerClient : IWorkerClient // throw, so it can never replace the timeout the caller is owed. if (removed) { - TrySendCancelForTimedOutCommand(correlationId, method, timeout); + TrySendCancelForAbandonedCommand( + correlationId, + method, + $"gateway command timeout after {timeout}"); } throw new WorkerClientException( @@ -300,11 +303,7 @@ public sealed class WorkerClient : IWorkerClient } catch (OperationCanceledException) { - RemovePendingCommandAsFailed( - correlationId, - pendingCommand, - WorkerClientErrorCode.GatewayShutdown, - "Command wait was canceled."); + AbandonCommandOnCallerCancellation(correlationId, pendingCommand, method); // WaitAsync surfaces TaskCanceledException; throwing through the token keeps the // exception the caller observes exactly what the hand-rolled wait produced. @@ -316,11 +315,7 @@ public sealed class WorkerClient : IWorkerClient // The deadline and the caller's cancellation raced and WaitAsync picked the timer. // The old wait classified this as cancellation, so this clause — reached only when // the filter above saw a canceled token — reproduces that treatment exactly. - RemovePendingCommandAsFailed( - correlationId, - pendingCommand, - WorkerClientErrorCode.GatewayShutdown, - "Command wait was canceled."); + AbandonCommandOnCallerCancellation(correlationId, pendingCommand, method); cancellationToken.ThrowIfCancellationRequested(); throw; } @@ -822,6 +817,45 @@ public sealed class WorkerClient : IWorkerClient return true; } + /// + /// Fails the pending command for a caller whose own token fired — a client deadline shorter + /// than the gateway's command timeout, or a disconnect that cancelled the RPC's token — and + /// forwards a WorkerCancel for the correlation the caller has abandoned (IPC-34). + /// + /// Caller cancellation is the common abandonment path, not the rare one, and it leaves the + /// worker in exactly the state the timeout arm's cancel exists for: the correlation is still + /// on the worker's single STA queue and would execute in its turn, behind live callers, for + /// a caller that is already gone. Under a burst of deadline-cancelled calls that backlog is + /// precisely what the cancel was added to shorten, so both arms send it. + /// + /// + /// Gated the same way as the timeout arm: only the caller that owned the removal sends, so a + /// reply that won the race is never chased. Additionally skipped once _stopCts has + /// fired — the client is closing, the outbound channel is being completed and every pending + /// command failed anyway, so a cancel would be at best pointless and at worst a write racing + /// teardown. + /// + /// + /// Correlation id of the abandoned command. + /// The pending command entry to fail. + /// Command method name, for the cancel reason and diagnostics. + private void AbandonCommandOnCallerCancellation( + string correlationId, + PendingCommand pendingCommand, + string method) + { + bool removed = RemovePendingCommandAsFailed( + correlationId, + pendingCommand, + WorkerClientErrorCode.CallerCancelled, + "Command wait was canceled."); + + if (removed && !_stopCts.IsCancellationRequested) + { + TrySendCancelForAbandonedCommand(correlationId, method, "caller cancelled the command"); + } + } + /// /// Forwards a WorkerCancel for a correlation the gateway has given up waiting for, so the /// worker can drop it from its STA queue (WorkerPipeSession routes the envelope to @@ -838,9 +872,10 @@ public sealed class WorkerClient : IWorkerClient /// /// The whole body sits under one catch-all that debug-logs, because the guarantee this method /// owes its caller is structural, not incidental: the caller is on the throw path for the - /// timeout, so anything escaping here — an envelope that failed to build, a TryWrite - /// against a disposed channel, a scheduler refusing the detached task — would replace the - /// the caller is owed with an unrelated + /// timeout or the cancellation, so anything escaping here — an envelope that failed to build, + /// a TryWrite against a disposed channel, a scheduler refusing the detached task — + /// would replace the (or the + /// ) the caller is owed with an unrelated /// exception. Losing the cancel costs the worker one wasted command; losing the timeout /// misreports why the call failed. The detached task carries its own handler for the same /// reason: its failures (including the from @@ -848,13 +883,13 @@ public sealed class WorkerClient : IWorkerClient /// and would otherwise be unobserved. It is deliberately not tracked or awaited — it holds no /// resource the shutdown path needs back, and the outbound channel is completed on close. /// - /// Correlation id of the command that timed out. - /// Command method name, for the cancel reason and diagnostics. - /// The elapsed command timeout, for the cancel reason. - private void TrySendCancelForTimedOutCommand( + /// Correlation id of the abandoned command. + /// Command method name, for diagnostics. + /// Why the gateway abandoned the command, carried on the WorkerCancel. + private void TrySendCancelForAbandonedCommand( string correlationId, string method, - TimeSpan timeout) + string reason) { try { @@ -862,7 +897,7 @@ public sealed class WorkerClient : IWorkerClient correlationId, envelope => envelope.WorkerCancel = new WorkerCancel { - Reason = $"gateway command timeout after {timeout}", + Reason = reason, }); if (_outboundEnvelopes.Writer.TryWrite(cancelEnvelope)) @@ -888,10 +923,10 @@ public sealed class WorkerClient : IWorkerClient } } - /// Records a cancel that could not be forwarded for a timed-out command. + /// Records a cancel that could not be forwarded for an abandoned command. /// The failure that stopped the cancel from being sent. - /// Command method name of the timed-out command. - /// Correlation id of the timed-out command. + /// Command method name of the abandoned command. + /// Correlation id of the abandoned command. private void LogCancelNotForwarded( Exception exception, string method, @@ -899,7 +934,7 @@ public sealed class WorkerClient : IWorkerClient { _logger.LogDebug( exception, - "Could not forward a cancel for timed-out worker command {Method} on session {SessionId} " + "Could not forward a cancel for abandoned worker command {Method} on session {SessionId} " + "and correlation {CorrelationId}.", method, SessionId, diff --git a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClientErrorCode.cs b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClientErrorCode.cs index b16c58d..bcc8d48 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClientErrorCode.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClientErrorCode.cs @@ -10,6 +10,14 @@ public enum WorkerClientErrorCode HeartbeatExpired, ShutdownTimeout, GatewayShutdown, + + // The caller's own cancellation token fired while it waited for the reply — a client deadline + // shorter than the gateway's command timeout, or a disconnect that cancelled the RPC's token. + // Distinct from GatewayShutdown (which these arms used to report) so metrics and logs stop + // recording a shutdown that did not happen; both map to Cancelled at the gRPC boundary, so the + // status the client observes is unchanged. + CallerCancelled, + WriteFailed, PendingCommandLimitExceeded, 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..eaab7e0 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs @@ -216,6 +216,112 @@ public sealed class WorkerClientTests Assert.Equal(WorkerClientState.Ready, client.State); } + /// + /// Caller cancellation abandons the correlation exactly as a gateway timeout does — a client + /// deadline shorter than the gateway's command timeout, or a disconnect, is the common + /// abandonment path — so it forwards the same WorkerCancel (IPC-34). Without it a + /// retry storm of deadline-cancelled calls leaves the worker's single STA working through + /// commands nobody is waiting for while live callers queue behind them. Twin of + /// , driven + /// by the caller's token instead of the timer. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task InvokeAsync_WhenCallerCancels_SendsWorkerCancelForThatCorrelation() + { + await using PipePair pipePair = await PipePair.CreateAsync(); + await using WorkerClient client = CreateClient(pipePair); + await CompleteHandshakeAsync(client, pipePair); + + using CancellationTokenSource callerCts = new(); + + // Advise rather than a control command: control commands bypass the worker's STA queue, so a + // data command is the case the cancel actually exists for. The timeout is the full test + // timeout so the cancellation, not the timer, is unambiguously what abandons the command. + Task invokeTask = client.InvokeAsync( + CreateCommand(MxCommandKind.Advise), + TestTimeout, + callerCts.Token); + + WorkerEnvelope commandEnvelope = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout); + Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerCommand, commandEnvelope.BodyCase); + + await callerCts.CancelAsync(); + + await Assert.ThrowsAnyAsync( + async () => await invokeTask.WaitAsync(TestTimeout)); + + WorkerEnvelope cancelEnvelope = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout); + + Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerCancel, cancelEnvelope.BodyCase); + Assert.Equal(commandEnvelope.CorrelationId, cancelEnvelope.CorrelationId); + Assert.False(string.IsNullOrWhiteSpace(cancelEnvelope.WorkerCancel.Reason)); + Assert.True( + cancelEnvelope.Sequence > commandEnvelope.Sequence, + $"The cancel arrived with sequence {cancelEnvelope.Sequence} after {commandEnvelope.Sequence}; " + + "envelope sequences must be strictly increasing in wire order."); + + // One caller gave up; the session did not fault. + Assert.Equal(WorkerClientState.Ready, client.State); + } + + /// + /// The cancel is forwarded for a caller that abandoned its command, never for the client's own + /// teardown (IPC-34): once the stop token has fired the outbound channel is being completed and + /// every pending command is failed anyway, so a cancel would at best be pointless and at worst + /// a write racing teardown. Asserted on the wire — the guard in the cancellation arm is the + /// code-level expression of it, and the completed outbound channel is the second line of + /// defense, so this pins the observable contract rather than either mechanism. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task InvokeAsync_WhenClientStopsUnderInFlightCommand_DoesNotSendWorkerCancel() + { + await using PipePair pipePair = await PipePair.CreateAsync(); + await using WorkerClient client = CreateClient(pipePair); + await CompleteHandshakeAsync(client, pipePair); + + using CancellationTokenSource callerCts = new(); + Task invokeTask = client.InvokeAsync( + CreateCommand(MxCommandKind.Advise), + TestTimeout, + callerCts.Token); + + WorkerEnvelope commandEnvelope = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout); + Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerCommand, commandEnvelope.BodyCase); + + // Read ahead before tearing down so anything the teardown path puts on the wire is caught. + // Exceptions are swallowed: the read is expected to be still pending at the assertion, and + // the stream is disposed underneath it when the test ends. + Task pendingRead = Task.Run(async () => + { + try + { + return await pipePair.WorkerReader.ReadAsync(); + } + catch (Exception) + { + return null; + } + }); + + // Kill fires the client's stop token and then fails the pending command; a caller token that + // is wired to gateway shutdown (as the session's is) cancels in the same window. + client.Kill("test teardown"); + await callerCts.CancelAsync(); + + await Assert.ThrowsAnyAsync(async () => await invokeTask.WaitAsync(TestTimeout)); + + await Task.Delay(TimeSpan.FromMilliseconds(250)); + if (pendingRead.IsCompleted) + { + WorkerEnvelope? unexpected = await pendingRead; + Assert.True( + unexpected is null || unexpected.BodyCase != WorkerEnvelope.BodyOneofCase.WorkerCancel, + "A stopping worker client must not put a WorkerCancel on the wire."); + } + } + /// /// The envelope sequence is a monotonic per-sender counter (gateway.md), so the values /// observed on the pipe must be strictly increasing in wire order. Stamping the sequence when