diff --git a/docs/MxAccessWorkerInstanceDesign.md b/docs/MxAccessWorkerInstanceDesign.md index d490f59..77b6fa7 100644 --- a/docs/MxAccessWorkerInstanceDesign.md +++ b/docs/MxAccessWorkerInstanceDesign.md @@ -656,11 +656,15 @@ the event queue implementation owns those counters. The STA watchdog currently emits a `WorkerFault` with `WorkerFaultCategory.StaHung` when `LastStaActivityUtc` is older than `WorkerPipeSessionOptions.HeartbeatGrace` **and no command is in flight**. -`StaRuntime.ProcessQueuedCommands` calls `MarkActivity()` only immediately -before and after each work item, so a synchronously long-running STA command -(for example a `ReadBulk` waiting `timeout_ms` for the first `OnDataChange`) -legitimately freezes `LastStaActivityUtc` for the duration of the wait while -the worker is healthy. The watchdog is therefore suppressed while the +`StaRuntime.ProcessQueuedCommands` calls `MarkActivity()` immediately before +and after each work item, so a synchronously long-running STA command that +neither completes work items nor pumps would freeze `LastStaActivityUtc` for +the duration of the wait while the worker is healthy. Commands that hold the +STA to wait for COM events (for example a `ReadBulk` waiting `timeout_ms` for +the first `OnDataChange`) avoid this: they pump via +`StaRuntime.PumpPendingMessages()`, which now refreshes `LastStaActivityUtc` +on every iteration (see the `HeartbeatStuckCeiling` discussion below). The +watchdog is additionally suppressed while the heartbeat snapshot's `CurrentCommandCorrelationId` is non-empty: the worker is busy executing a command, not hung, and the heartbeat already surfaces the in-flight correlation id so the gateway can apply its own per-command timeout @@ -684,10 +688,18 @@ session and only the gateway's per-command timeout would catch the hang — losing the worker-originated diagnostic (`StaHung` fault category, the stale-by interval) from the gateway audit trail. Once `LastStaActivityUtc` has been stale for longer than `HeartbeatStuckCeiling`, the watchdog fires -`StaHung` regardless of whether a command is in flight, on the assumption -that no legitimate STA command should run that long without periodically -refreshing activity. Deployments that legitimately run very long bulk -operations should raise the ceiling rather than disable it. +`StaHung` regardless of whether a command is in flight. This is now safe for +healthy long-running commands: `StaRuntime.PumpPendingMessages()` refreshes +`LastStaActivityUtc` (via `MarkActivity()`) every time it runs, and long-hold +STA commands invoke it on every wait iteration (`ReadBulk` routes its +per-tag wait through the `pumpStep` wired from `StaRuntime.PumpPendingMessages`). +A command that keeps pumping therefore keeps its activity timestamp fresh and +never reaches the ceiling, while a genuinely stuck STA — one that has stopped +pumping — accrues staleness and faults correctly. The ceiling is thus the +backstop for a command that both holds the thread and stops pumping, not a +guillotine for slow-but-healthy work. Deployments that legitimately run very +long bulk operations should still be able to raise the ceiling rather than +disable it. ## Shutdown 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 0c13b3a..117770a 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs @@ -704,6 +704,102 @@ public sealed class WorkerPipeSessionTests await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token); } + /// + /// WRK-01 regression: a long in-flight STA command that keeps pumping + /// must NOT self-fault as StaHung, and its reply must still be + /// delivered. The real fix makes StaRuntime.PumpPendingMessages + /// refresh LastActivityUtc on every wait iteration, so a healthy + /// ReadBulk holding the STA far longer than + /// HeartbeatStuckCeiling (75 s in production) keeps its activity + /// timestamp fresh. This test compresses the clock — a 100 ms ceiling + /// with a command in flight across a window many multiples longer — and + /// models the pump refresh by continuously advancing the snapshot's + /// LastStaActivityUtc while the command blocks. Contrast + /// , + /// where a frozen timestamp beyond the ceiling correctly faults; here + /// the refreshed timestamp must keep the fault suppressed and let the + /// reply through the Ready-state gate. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task RunAsync_LongInFlightCommandThatKeepsPumping_DoesNotFaultAndDeliversReply() + { + using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(20)); + using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token); + FakeRuntimeSession runtime = new() + { + BlockDispatch = true, + }; + WorkerPipeSession session = CreatePipeSession( + pipePair.WorkerStream, + runtime, + new WorkerPipeSessionOptions + { + HeartbeatInterval = TimeSpan.FromMilliseconds(20), + HeartbeatGrace = TimeSpan.FromMilliseconds(50), + HeartbeatStuckCeiling = TimeSpan.FromMilliseconds(100), + }); + Task runTask = session.RunAsync(cancellation.Token); + await CompleteGatewayHandshakeAsync(pipePair, cancellation.Token); + + // Kick off the long command; it blocks in DispatchAsync until released, + // so its correlation id stays in flight in the heartbeat snapshot. + await pipePair.GatewayWriter + .WriteAsync(CreateCommandEnvelope("long-bulk-read"), cancellation.Token); + Assert.True( + runtime.DispatchStarted.Wait(TimeSpan.FromSeconds(5)), + "The long command must reach the runtime and begin dispatch."); + + // Model the pump refreshing STA activity on each wait iteration: keep + // the snapshot's LastStaActivityUtc current while the command is in + // flight. + using CancellationTokenSource pumpRefresh = new(); + Task refreshLoop = Task.Run( + async () => + { + while (!pumpRefresh.IsCancellationRequested) + { + runtime.SetSnapshot(new WorkerRuntimeHeartbeatSnapshot( + DateTimeOffset.UtcNow, + pendingCommandCount: 1, + outboundEventQueueDepth: 0, + lastEventSequence: 0, + currentCommandCorrelationId: "long-bulk-read")); + await Task.Delay(TimeSpan.FromMilliseconds(20)).ConfigureAwait(false); + } + }); + + // Inspect a bounded number of frames over a window many multiples of the + // 100 ms ceiling (at least 30 heartbeats at 20 ms ~ 600 ms). None may be + // a WorkerFault while activity is continuously refreshed. + const int framesToInspect = 30; + for (int index = 0; index < framesToInspect; index++) + { + WorkerEnvelope envelope = await pipePair.GatewayReader + .ReadAsync(cancellation.Token); + Assert.NotEqual( + WorkerEnvelope.BodyOneofCase.WorkerFault, + envelope.BodyCase); + } + + // Stop refreshing and release the command; its reply must be delivered + // because the session never faulted (state stayed Ready). + pumpRefresh.Cancel(); + await refreshLoop; + runtime.ReleaseDispatch(); + + WorkerEnvelope reply = await ReadUntilAsync( + pipePair.GatewayReader, + WorkerEnvelope.BodyOneofCase.WorkerCommandReply, + envelope => envelope.CorrelationId == "long-bulk-read", + cancellation.Token); + Assert.Equal( + ProtocolStatusCode.Ok, + reply.WorkerCommandReply.Reply.ProtocolStatus.Code); + + await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token); + } + /// /// RunAsync must throw a diagnostic exception if the /// runtime-session factory returns null, rather than deferring the diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Sta/StaRuntimeTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Sta/StaRuntimeTests.cs index 3afaa85..f143324 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Sta/StaRuntimeTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Sta/StaRuntimeTests.cs @@ -85,6 +85,34 @@ public sealed class StaRuntimeTests Assert.True(updated); } + /// + /// Verifies that refreshes + /// . A long synchronous STA + /// command (for example ReadBulk waiting timeout_ms for + /// the first OnDataChange) invokes the pump step on every wait + /// iteration while it legitimately holds the STA thread; refreshing + /// activity here keeps the watchdog from mistaking a busy STA for a hung + /// one (WRK-01). The runtime is deliberately left unstarted so the only + /// source of activity is the pump call under test, not the idle loop. + /// + [Fact] + public void PumpPendingMessages_RefreshesLastActivity() + { + RecordingComApartmentInitializer initializer = new(); + using StaRuntime runtime = CreateRuntime(initializer); + DateTimeOffset before = runtime.LastActivityUtc; + + bool refreshed = SpinWait.SpinUntil( + () => + { + runtime.PumpPendingMessages(); + return runtime.LastActivityUtc > before; + }, + TimeSpan.FromSeconds(2)); + + Assert.True(refreshed, "PumpPendingMessages must advance LastActivityUtc."); + } + /// Verifies that InvokeAsync faults the returned task when a command raises an exception without stopping the runtime. /// A task that represents the asynchronous operation. [Fact] diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaRuntime.cs b/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaRuntime.cs index 2ba2d37..f8a83c1 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaRuntime.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaRuntime.cs @@ -80,14 +80,24 @@ public sealed class StaRuntime : IDisposable public bool IsRunning => startedEvent.IsSet && !stoppedEvent.IsSet; /// - /// Pumps any pending Windows messages on the calling thread. Intended - /// for commands that synchronously hold the STA (e.g. ReadBulk) and - /// must allow inbound MXAccess COM events to dispatch while they - /// wait. Callers must already be on the STA; the method is otherwise - /// safe (PeekMessage simply finds no messages). + /// Pumps any pending Windows messages on the calling thread and refreshes + /// the STA activity timestamp. Intended for commands that synchronously + /// hold the STA (e.g. ReadBulk) and must allow inbound MXAccess COM events + /// to dispatch while they wait. Because a long-running command invokes this + /// on every wait iteration, refreshing activity here keeps a busy STA from + /// being mistaken for a hung one: a healthy command that keeps pumping stays + /// fresh past HeartbeatStuckCeiling, while a genuinely stuck STA (no + /// pumping) still accrues staleness and faults correctly. Callers must + /// already be on the STA; the method is otherwise safe (PeekMessage simply + /// finds no messages). /// /// The number of messages pumped. - public int PumpPendingMessages() => messagePump.PumpPendingMessages(); + public int PumpPendingMessages() + { + int pumpedMessages = messagePump.PumpPendingMessages(); + MarkActivity(); + return pumpedMessages; + } /// /// Starts the STA thread.