dohertj2/mxaccessgw
1
0
Fork 0
Code Issues 14 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

fix(ipc): cancellation-priority timeout classification; structurally-enforced no-throw cancel send

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-15 12:41:38 -04:00
parent 1742e38c10
commit a1a38b5538
3 changed files with 130 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs
+105 -52
View File
@@ -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
/// <param name="pendingCommand">The pending command.</param>
/// <param name="errorCode">Error code.</param>
/// <param name="message">Error message.</param>
private void RemovePendingCommandAsFailed(
/// <returns>
/// <c>true</c> when this call removed the pending entry and owns the failure; <c>false</c> 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.
/// </returns>
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;
}
/// <summary>
@@ -786,14 +817,17 @@ public sealed class WorkerClient : IWorkerClient
/// </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.
/// 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 <c>TryWrite</c>
/// against a disposed channel, a scheduler refusing the detached task — would replace the
/// <see cref="WorkerClientErrorCode.CommandTimeout"/> 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 <see cref="ObjectDisposedException"/> from
/// <c>_stopCts</c> 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.
/// </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>
@@ -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);
}
}
/// <summary>Records a cancel that could not be forwarded for a timed-out command.</summary>
/// <param name="exception">The failure that stopped the cancel from being sent.</param>
/// <param name="method">Command method name of the timed-out command.</param>
/// <param name="correlationId">Correlation id of the timed-out command.</param>
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);
}
/// <summary>Reads and validates a handshake envelope.</summary>