perf(ipc): WaitAsync command timeouts + forward WorkerCancel so a timed-out COM call frees the STA

2026-08-15 12:28:09 -04:00
parent 7171892984
commit 7c1ea12331
3 changed files with 178 additions and 19 deletions
@@ -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<WorkerCommandReply> 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));
    }

+    /// <summary>
+    ///     Forwards a <c>WorkerCancel</c> for a correlation the gateway has given up waiting for, so the
+    ///     worker can drop it from its STA queue (<c>WorkerPipeSession</c> routes the envelope to
+    ///     <c>CancelCommand</c>). 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.
+    ///     <para>
+    ///     A command whose envelope has not yet left <c>_outboundEnvelopes</c> is handled by the same
+    ///     path rather than by pulling it back out: <see cref="Channel{T}"/> 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.
+    ///     </para>
+    /// </summary>
+    /// <remarks>
+    ///     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 <see cref="WorkerClientErrorCode.CommandTimeout"/> the
+    ///     caller is owed. <c>TryWrite</c> 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
+    ///     <see cref="ObjectDisposedException"/> from <c>_stopCts</c> if the client is disposed
+    ///     underneath it — so nothing is left unobserved.
+    /// </remarks>
+    /// <param name="correlationId">Correlation id of the command that timed out.</param>
+    /// <param name="method">Command method name, for the cancel reason and diagnostics.</param>
+    /// <param name="timeout">The elapsed command timeout, for the cancel reason.</param>
+    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);
+            }
+        });
+    }
+
    /// <summary>Reads and validates a handshake envelope.</summary>
    /// <param name="expectedBody">Expected envelope body type.</param>
    /// <param name="cancellationToken">Cancellation token.</param>