diff --git a/docs/GatewayProcessDesign.md b/docs/GatewayProcessDesign.md index 98ce742..194800b 100644 --- a/docs/GatewayProcessDesign.md +++ b/docs/GatewayProcessDesign.md @@ -604,8 +604,16 @@ running on it, and the rule above still holds. A command whose envelope is still in the gateway's outbound queue needs no special handling: the queue is FIFO, so the worker reads the command and then its cancel and drops it before execution. -Failing to send the cancel is logged at debug and never replaces the -`CommandTimeout` the caller is owed. +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. + +Cancellation outranks the deadline. When a caller's token is canceled around the +same time the timeout fires, the command is reported as canceled +(`GatewayShutdown`, `OperationCanceledException`), not as `CommandTimeout`, and +no cancel is forwarded. ## Fault Model diff --git a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs index b0c6d84..bff47b1 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs @@ -256,26 +256,20 @@ public sealed class WorkerClient : IWorkerClient // continuation — the old shape allocated a linked CancellationTokenSource, its // registration, a delay Task, and the WhenAny Task on every invoke, and left the delay // Task rooted until the cancel completed. WaitAsync raises TimeoutException for the - // deadline and OperationCanceledException for the caller's token, which is exactly the - // two-way split the old completedTask/IsCancellationRequested test made by hand; the - // error codes and messages below are unchanged. + // deadline and OperationCanceledException for the caller's token — but unlike the old + // wait it races the two and reports whichever fired first, whereas the old code inspected + // cancellationToken.IsCancellationRequested BEFORE classifying a won delay as a timeout. + // The filter on the CommandTimeout clause restores that priority: a token canceled around + // the deadline is still classified as cancellation, never as CommandTimeout. Error codes + // and messages are unchanged. try { return await pendingCommand.Task.WaitAsync(timeout, cancellationToken).ConfigureAwait(false); } - catch (OperationCanceledException) - { - RemovePendingCommandAsFailed( - correlationId, - pendingCommand, - WorkerClientErrorCode.GatewayShutdown, - "Command wait was canceled."); - throw; - } - catch (TimeoutException) + catch (TimeoutException) when (!cancellationToken.IsCancellationRequested) { string timeoutMessage = $"Worker command {method} timed out after {timeout}."; - RemovePendingCommandAsFailed( + bool removed = RemovePendingCommandAsFailed( correlationId, pendingCommand, WorkerClientErrorCode.CommandTimeout, @@ -284,14 +278,45 @@ public sealed class WorkerClient : IWorkerClient // The gateway has stopped waiting, but the worker has not stopped working: the // correlation is still on its single STA queue and would execute (or keep executing) // regardless. Tell it, so a queued-but-not-started command is dropped instead of - // occupying the STA behind a caller that is already gone. Best-effort by design — a - // failure here must never replace the timeout the caller is owed. - TrySendCancelForTimedOutCommand(correlationId, method, timeout); + // occupying the STA behind a caller that is already gone. Gated on the removal so a + // reply that won the race — the pending entry is already gone and the caller is about + // to see it — never has a cancel chase it. Best-effort by design; the send cannot + // throw, so it can never replace the timeout the caller is owed. + if (removed) + { + TrySendCancelForTimedOutCommand(correlationId, method, timeout); + } throw new WorkerClientException( WorkerClientErrorCode.CommandTimeout, timeoutMessage); } + catch (OperationCanceledException) + { + RemovePendingCommandAsFailed( + correlationId, + pendingCommand, + WorkerClientErrorCode.GatewayShutdown, + "Command wait was canceled."); + + // WaitAsync surfaces TaskCanceledException; throwing through the token keeps the + // exception the caller observes exactly what the hand-rolled wait produced. + cancellationToken.ThrowIfCancellationRequested(); + throw; + } + catch (TimeoutException) + { + // 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."); + cancellationToken.ThrowIfCancellationRequested(); + throw; + } } catch { @@ -755,7 +780,12 @@ public sealed class WorkerClient : IWorkerClient /// The pending command. /// Error code. /// Error message. - private void RemovePendingCommandAsFailed( + /// + /// true when this call removed the pending entry and owns the failure; false when + /// the entry was already gone — a reply, fault, or shutdown got there first, so the caller must + /// not take any further action on behalf of that correlation. + /// + private bool RemovePendingCommandAsFailed( string correlationId, PendingCommand pendingCommand, WorkerClientErrorCode errorCode, @@ -763,13 +793,14 @@ public sealed class WorkerClient : IWorkerClient { if (!_pendingCommands.TryRemove(correlationId, out _)) { - return; + return false; } ReleasePendingCommandSlot(); TimeSpan duration = _timeProvider.GetElapsedTime(pendingCommand.StartTimestamp); _metrics?.CommandFailed(pendingCommand.Method, errorCode.ToString(), duration); pendingCommand.SetException(new WorkerClientException(errorCode, message)); + return true; } /// @@ -786,14 +817,17 @@ public sealed class WorkerClient : IWorkerClient /// /// /// - /// Every failure is swallowed and logged at debug: the caller is on the throw path for the - /// timeout, and losing the cancel costs the worker one wasted command, whereas letting an - /// exception escape would replace the the - /// caller is owed. TryWrite is the fast path and cannot throw for a closed or completed - /// channel; only when the outbound channel is momentarily at its bound is a detached task - /// started, and that task catches everything it can observe — including the - /// from _stopCts if the client is disposed - /// underneath it — so nothing is left unobserved. + /// 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 + /// 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 + /// _stopCts if the client is disposed underneath it) happen after this method returns + /// 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. @@ -803,35 +837,54 @@ public sealed class WorkerClient : IWorkerClient string method, TimeSpan timeout) { - WorkerEnvelope cancelEnvelope = CreateEnvelope( - correlationId, - envelope => envelope.WorkerCancel = new WorkerCancel + try + { + WorkerEnvelope cancelEnvelope = CreateEnvelope( + correlationId, + envelope => envelope.WorkerCancel = new WorkerCancel + { + Reason = $"gateway command timeout after {timeout}", + }); + + if (_outboundEnvelopes.Writer.TryWrite(cancelEnvelope)) { - Reason = $"gateway command timeout after {timeout}", + return; + } + + _ = Task.Run(async () => + { + try + { + await EnqueueAsync(cancelEnvelope, _stopCts.Token).ConfigureAwait(false); + } + catch (Exception exception) + { + LogCancelNotForwarded(exception, method, correlationId); + } }); - - if (_outboundEnvelopes.Writer.TryWrite(cancelEnvelope)) - { - return; } - - _ = Task.Run(async () => + catch (Exception exception) { - try - { - await EnqueueAsync(cancelEnvelope, _stopCts.Token).ConfigureAwait(false); - } - catch (Exception exception) - { - _logger.LogDebug( - exception, - "Could not forward a cancel for timed-out worker command {Method} on session {SessionId} " - + "and correlation {CorrelationId}.", - method, - SessionId, - correlationId); - } - }); + LogCancelNotForwarded(exception, method, correlationId); + } + } + + /// Records a cancel that could not be forwarded for a timed-out 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. + private void LogCancelNotForwarded( + Exception exception, + string method, + string correlationId) + { + _logger.LogDebug( + exception, + "Could not forward a cancel for timed-out worker command {Method} on session {SessionId} " + + "and correlation {CorrelationId}.", + method, + SessionId, + correlationId); } /// Reads and validates a handshake envelope. 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 b4f34dd..807249d 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs @@ -163,7 +163,7 @@ public sealed class WorkerClientTests // The timeout also emits a WorkerCancel for the abandoned correlation (GWC-31), which sits // ahead of the second command on the FIFO pipe; skip it rather than mistaking it for the // command this assertion is about. - WorkerEnvelope secondCommand = await ReadNextCommandAsync(pipePair); + WorkerEnvelope secondCommand = await ReadNextCommandAsync(pipePair, timedOutCommand.CorrelationId); await pipePair.WriteAsync( CreateCommandReplyEnvelope(secondCommand.CorrelationId, MxCommandKind.GetWorkerInfo)); @@ -188,8 +188,10 @@ public sealed class WorkerClientTests await using WorkerClient client = CreateClient(pipePair); await CompleteHandshakeAsync(client, pipePair); + // 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. Task invokeTask = client.InvokeAsync( - CreateCommand(MxCommandKind.Ping), + CreateCommand(MxCommandKind.Advise), TimeSpan.FromMilliseconds(50), CancellationToken.None); @@ -1076,12 +1078,17 @@ public sealed class WorkerClientTests } /// - /// Reads gateway envelopes until a WorkerCommand arrives, skipping the control envelopes - /// (such as the WorkerCancel a command timeout emits) that may precede it on the FIFO pipe. + /// Reads gateway envelopes until a WorkerCommand arrives, skipping the WorkerCancel + /// a command timeout emits for . Anything else on the pipe + /// fails the test rather than being skipped: the point of the skip is to tolerate exactly the one + /// known interleaving, not to make the assertion blind to unexpected gateway traffic. /// /// The connected pipe pair whose worker side is read. + /// Correlation id of the timed-out command whose cancel is expected. /// The next command envelope written by the gateway. - private static async Task ReadNextCommandAsync(PipePair pipePair) + private static async Task ReadNextCommandAsync( + PipePair pipePair, + string canceledCorrelationId) { while (true) { @@ -1090,6 +1097,9 @@ public sealed class WorkerClientTests { return envelope; } + + Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerCancel, envelope.BodyCase); + Assert.Equal(canceledCorrelationId, envelope.CorrelationId); } }