diff --git a/docs/MxAccessWorkerInstanceDesign.md b/docs/MxAccessWorkerInstanceDesign.md index d53543b..1e35078 100644 --- a/docs/MxAccessWorkerInstanceDesign.md +++ b/docs/MxAccessWorkerInstanceDesign.md @@ -1011,7 +1011,11 @@ Graceful shutdown sequence: 1. Pipe reader receives `WorkerShutdown`. 2. Worker host marks shutdown requested. -3. Reject new commands. +3. Reject new commands. A command that reaches the dispatcher after intake has + stopped is answered with a `WorkerUnavailable` reply and logged as + `WorkerCommandRejectedAfterShutdown`, so the gateway's correlation wait ends + at once instead of running out its own timeout. The read loop is serial and + stops after the shutdown frame, so this guard is defensive today. 4. Let current STA command finish if within timeout. 5. Optionally run MXAccess cleanup: - `UnAdvise`, diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs index 033b5c8..1b08781 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs @@ -1943,6 +1943,60 @@ public sealed class WorkerPipeSessionTests && (string?)correlationId == "command-dropped-after-shutdown"); } + /// + /// A command that reaches the dispatcher after shutdown has stopped command intake must be + /// answered with (and logged) rather than + /// dropped silently, which would leave the gateway's correlation wait to run out its own + /// timeout with no trace (#182). + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task CommandAfterShutdown_RepliesWorkerUnavailable() + { + using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(10)); + using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token); + FakeRuntimeSession runtime = new(); + RecordingWorkerLogger logger = new(); + WorkerFrameProtocolOptions options = CreateOptions(); + WorkerPipeSession session = new( + new WorkerFrameReader(pipePair.WorkerStream, options), + new WorkerFrameWriter(pipePair.WorkerStream, options), + options, + () => 1234, + new WorkerPipeSessionOptions + { + HeartbeatInterval = TimeSpan.FromSeconds(1), + HeartbeatGrace = TimeSpan.FromSeconds(5), + }, + () => runtime, + logger); + Task runTask = session.RunAsync(cancellation.Token); + await CompleteGatewayHandshakeAsync(pipePair, cancellation.Token); + + session.StopAcceptingCommands(); + await pipePair.GatewayWriter.WriteAsync( + CreateCommandEnvelope("command-after-shutdown"), + cancellation.Token); + + WorkerEnvelope reply = await ReadUntilAsync( + pipePair.GatewayReader, + WorkerEnvelope.BodyOneofCase.WorkerCommandReply, + envelope => envelope.CorrelationId == "command-after-shutdown", + cancellation.Token); + + Assert.Equal(ProtocolStatusCode.WorkerUnavailable, reply.WorkerCommandReply.Reply.ProtocolStatus.Code); + Assert.Equal(MxCommandKind.Register, reply.WorkerCommandReply.Reply.Kind); + Assert.Equal(SessionId, reply.WorkerCommandReply.Reply.SessionId); + Assert.False(runtime.DispatchStarted.IsSet); + Assert.Contains( + logger.Events, + entry => entry.EventName == "WorkerCommandRejectedAfterShutdown" + && entry.Fields.TryGetValue("correlation_id", out object? correlationId) + && (string?)correlationId == "command-after-shutdown"); + + await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token); + } + private static WorkerPipeSession CreateSession( Stream inbound, Stream outbound, diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs index f4f2de0..6176f95 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs @@ -748,7 +748,11 @@ public sealed class WorkerPipeSession return await HandleControlCommandAsync(envelope, cancellationToken).ConfigureAwait(false); } - TryStartCommandTask(envelope, cancellationToken); + if (!TryStartCommandTask(envelope, cancellationToken)) + { + await RejectCommandAfterShutdownAsync(envelope, cancellationToken).ConfigureAwait(false); + } + return true; case WorkerEnvelope.BodyOneofCase.WorkerShutdown: await ShutdownAsync(envelope.WorkerShutdown, cancellationToken).ConfigureAwait(false); @@ -1230,7 +1234,12 @@ public sealed class WorkerPipeSession } } - private void TryStartCommandTask( + /// + /// Starts the command on the runtime session unless shutdown has already stopped command + /// intake. Returns false when the command was refused, so the caller can answer the + /// correlation instead of leaving the gateway's wait to run out its own timeout. + /// + private bool TryStartCommandTask( WorkerEnvelope envelope, CancellationToken cancellationToken) { @@ -1239,7 +1248,7 @@ public sealed class WorkerPipeSession { if (!_acceptingCommands) { - return; + return false; } commandTask = ProcessCommandAsync(envelope, cancellationToken); @@ -1254,6 +1263,60 @@ public sealed class WorkerPipeSession // would fault the discarded task with no observer. Close the gap the same way the rest of // the file does. ObserveFaultWhenever(ObserveCommandTaskAsync(commandTask)); + return true; + } + + /// + /// Answers a command that arrived after shutdown stopped command intake with a + /// reply, and logs it. The worker is + /// going away, so the reply is best effort: if the pipe is already unusable the failure is + /// logged and the gateway falls back to its own correlation timeout, as it did before. + /// + private async Task RejectCommandAfterShutdownAsync( + WorkerEnvelope envelope, + CancellationToken cancellationToken) + { + MxCommandKind kind = envelope.WorkerCommand?.Command?.Kind ?? MxCommandKind.Unspecified; + const string message = "The worker is shutting down and no longer accepts commands."; + _logger?.Information( + "WorkerCommandRejectedAfterShutdown", + new Dictionary + { + ["correlation_id"] = envelope.CorrelationId, + ["command_kind"] = kind.ToString(), + ["worker_state"] = _state.ToString(), + }); + + MxCommandReply reply = new() + { + SessionId = _options.SessionId, + CorrelationId = envelope.CorrelationId, + Kind = kind, + DiagnosticMessage = message, + ProtocolStatus = new ProtocolStatus + { + Code = ProtocolStatusCode.WorkerUnavailable, + Message = message, + }, + }; + + try + { + await WriteControlReplyAsync(reply, cancellationToken).ConfigureAwait(false); + } + catch (Exception exception) when (exception is IOException + || exception is ObjectDisposedException + || exception is WorkerFrameProtocolException) + { + _logger?.Error( + "WorkerCommandRejectionReplyFailed", + new Dictionary + { + ["correlation_id"] = envelope.CorrelationId, + ["command_kind"] = kind.ToString(), + ["reason"] = exception.Message, + }); + } } private async Task ObserveCommandTaskAsync(Task commandTask) @@ -1274,7 +1337,12 @@ public sealed class WorkerPipeSession } } - private void StopAcceptingCommands() + /// + /// Stops command intake. Internal so tests can reach the refusal path directly: the read + /// loop is serial and ends after a shutdown, so a command cannot race it over the pipe today, + /// but the guard must still answer if a future change lets one through. + /// + internal void StopAcceptingCommands() { lock (_commandTaskGate) {