From 7c1ea123311af2864f03f89ccb1ef338634d9796 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 12:28:09 -0400 Subject: [PATCH] perf(ipc): WaitAsync command timeouts + forward WorkerCancel so a timed-out COM call frees the STA --- docs/GatewayProcessDesign.md | 14 +++ .../Workers/WorkerClient.cs | 118 +++++++++++++++--- .../Gateway/Workers/WorkerClientTests.cs | 65 +++++++++- 3 files changed, 178 insertions(+), 19 deletions(-) diff --git a/docs/GatewayProcessDesign.md b/docs/GatewayProcessDesign.md index 6fd2ddd..98ce742 100644 --- a/docs/GatewayProcessDesign.md +++ b/docs/GatewayProcessDesign.md @@ -593,6 +593,20 @@ 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 +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 +call — so this shortens the STA backlog rather than freeing a call already +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. + ## Fault Model Fault categories: diff --git a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs index be74d2b..b0c6d84 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs @@ -232,6 +232,13 @@ public sealed class WorkerClient : IWorkerClient // session. Command envelopes are the only gateway-authored outbound payload whose // size the caller controls; checking here keeps a MessageTooLarge in the write loop a // genuine desync signal. + // + // PERF(GWC-31): this size cannot be handed to WorkerFrameWriter to spare its own + // CalculateSize. WriteLoopAsync stamps envelope.Sequence immediately before the write + // (GWC-28), and a non-zero varint field grows the encoding — so the number computed here + // is a lower bound on the frame the writer actually emits, never the frame length. Passing + // it as a knownSize would under-length the prefix and desync the worker's framing. The + // pre-check stays a pre-check: it is conservative in the right direction. int envelopeSize = commandEnvelope.CalculateSize(); if (envelopeSize > _connection.FrameOptions.MaxMessageBytes) { @@ -242,36 +249,49 @@ public sealed class WorkerClient : IWorkerClient } await EnqueueAsync(commandEnvelope, cancellationToken).ConfigureAwait(false); - using CancellationTokenSource timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); - Task timeoutTask = Task.Delay(timeout, timeoutCts.Token); - Task replyTask = pendingCommand.Task; - Task completedTask = await Task.WhenAny(replyTask, timeoutTask).ConfigureAwait(false); - if (completedTask == replyTask) + // GWC-31: one pooled timer instead of a linked CTS + Task.Delay + WhenAny per command. + // Task.WaitAsync arms a TimerQueueTimer on the shared timer queue and cancels it when the + // reply lands, so the steady-state cost of a command that replies in time is a single + // 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. + try { - await timeoutCts.CancelAsync().ConfigureAwait(false); - return await replyTask.ConfigureAwait(false); + return await pendingCommand.Task.WaitAsync(timeout, cancellationToken).ConfigureAwait(false); } - - if (cancellationToken.IsCancellationRequested) + catch (OperationCanceledException) { RemovePendingCommandAsFailed( correlationId, pendingCommand, WorkerClientErrorCode.GatewayShutdown, "Command wait was canceled."); - cancellationToken.ThrowIfCancellationRequested(); + throw; } + catch (TimeoutException) + { + string timeoutMessage = $"Worker command {method} timed out after {timeout}."; + RemovePendingCommandAsFailed( + correlationId, + pendingCommand, + WorkerClientErrorCode.CommandTimeout, + timeoutMessage); - RemovePendingCommandAsFailed( - correlationId, - pendingCommand, - WorkerClientErrorCode.CommandTimeout, - $"Worker command {method} timed out after {timeout}."); + // 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); - throw new WorkerClientException( - WorkerClientErrorCode.CommandTimeout, - $"Worker command {method} timed out after {timeout}."); + throw new WorkerClientException( + WorkerClientErrorCode.CommandTimeout, + timeoutMessage); + } } catch { @@ -752,6 +772,68 @@ public sealed class WorkerClient : IWorkerClient pendingCommand.SetException(new WorkerClientException(errorCode, message)); } + /// + /// 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 + /// CancelCommand). A cancel that arrives after the command already reached the COM call + /// is a no-op — MXAccess offers no way to abort an in-flight call — so this shortens the STA + /// backlog rather than freeing a call already running on it. + /// + /// A command whose envelope has not yet left _outboundEnvelopes is handled by the same + /// path rather than by pulling it back out: exposes no removal, and + /// the queue is FIFO, so the worker simply reads the command and then its cancel and drops the + /// correlation before it ever reaches the STA. Nothing is gained by dequeuing it here. + /// + /// + /// + /// 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. + /// + /// 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( + string correlationId, + string method, + TimeSpan timeout) + { + WorkerEnvelope cancelEnvelope = CreateEnvelope( + correlationId, + envelope => envelope.WorkerCancel = new WorkerCancel + { + Reason = $"gateway command timeout after {timeout}", + }); + + if (_outboundEnvelopes.Writer.TryWrite(cancelEnvelope)) + { + return; + } + + _ = Task.Run(async () => + { + 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); + } + }); + } + /// Reads and validates a handshake envelope. /// Expected envelope body type. /// Cancellation token. 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 5855572..b4f34dd 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs @@ -159,7 +159,11 @@ public sealed class WorkerClientTests CreateCommand(MxCommandKind.GetWorkerInfo), TestTimeout, CancellationToken.None); - WorkerEnvelope secondCommand = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout); + + // 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); await pipePair.WriteAsync( CreateCommandReplyEnvelope(secondCommand.CorrelationId, MxCommandKind.GetWorkerInfo)); @@ -169,6 +173,47 @@ public sealed class WorkerClientTests Assert.Equal(MxCommandKind.GetWorkerInfo, reply.Reply.Kind); } + /// + /// A command timeout abandons the gateway-side wait, but the worker keeps the correlation on + /// its single STA queue and would still run it — so the gateway forwards a WorkerCancel + /// for the abandoned correlation id (GWC-31). Without it, a client that retries after a + /// timeout stacks work the worker still intends to execute. Asserted on the wire because the + /// cancel is protocol behavior the worker depends on, not an internal detail. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task InvokeAsync_WhenCommandTimesOut_SendsWorkerCancelForThatCorrelation() + { + await using PipePair pipePair = await PipePair.CreateAsync(); + await using WorkerClient client = CreateClient(pipePair); + await CompleteHandshakeAsync(client, pipePair); + + Task invokeTask = client.InvokeAsync( + CreateCommand(MxCommandKind.Ping), + TimeSpan.FromMilliseconds(50), + CancellationToken.None); + + WorkerEnvelope commandEnvelope = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout); + Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerCommand, commandEnvelope.BodyCase); + + WorkerClientException exception = await Assert.ThrowsAsync( + async () => await invokeTask.WaitAsync(TestTimeout)); + Assert.Equal(WorkerClientErrorCode.CommandTimeout, exception.ErrorCode); + + 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."); + + // The timeout fails one command; it is not a session fault. + Assert.Equal(WorkerClientState.Ready, client.State); + } + /// /// 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 @@ -1030,6 +1075,24 @@ public sealed class WorkerClientTests return envelope; } + /// + /// 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. + /// + /// The connected pipe pair whose worker side is read. + /// The next command envelope written by the gateway. + private static async Task ReadNextCommandAsync(PipePair pipePair) + { + while (true) + { + WorkerEnvelope envelope = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout); + if (envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerCommand) + { + return envelope; + } + } + } + private static async Task WaitUntilAsync( Func predicate, TimeSpan timeout)