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.
This commit is contained in:
@@ -216,6 +216,112 @@ public sealed class WorkerClientTests
|
||||
Assert.Equal(WorkerClientState.Ready, client.State);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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 <c>WorkerCancel</c> (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
|
||||
/// <see cref="InvokeAsync_WhenCommandTimesOut_SendsWorkerCancelForThatCorrelation"/>, driven
|
||||
/// by the caller's token instead of the timer.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[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<WorkerCommandReply> 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<OperationCanceledException>(
|
||||
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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[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<WorkerCommandReply> 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<WorkerEnvelope?> 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<Exception>(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.");
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The envelope <c>sequence</c> 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
|
||||
|
||||
Reference in New Issue
Block a user