fix(WRK-01): STA pump step refreshes activity to prevent false StaHung
A long legitimate ReadBulk pumped Windows messages without refreshing LastStaActivityUtc, so the watchdog false-positived StaHung past HeartbeatStuckCeiling and then silently dropped every reply. PumpPendingMessages() now calls MarkActivity() after pumping; a genuinely stuck STA (no pumping) still accrues staleness and faults correctly. No MXAccess parity change. archreview: WRK-01 (P0). Verified on the Windows host (x86): worker builds clean, StaRuntimeTests + WorkerPipeSessionTests 33/33 pass.
This commit is contained in:
Joseph Doherty
@@ -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
|
||||
|
||||
|
||||
@@ -704,6 +704,102 @@ public sealed class WorkerPipeSessionTests
|
||||
await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// WRK-01 regression: a long in-flight STA command that keeps pumping
|
||||
/// must NOT self-fault as <c>StaHung</c>, and its reply must still be
|
||||
/// delivered. The real fix makes <c>StaRuntime.PumpPendingMessages</c>
|
||||
/// refresh <c>LastActivityUtc</c> on every wait iteration, so a healthy
|
||||
/// <c>ReadBulk</c> holding the STA far longer than
|
||||
/// <c>HeartbeatStuckCeiling</c> (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
|
||||
/// <c>LastStaActivityUtc</c> while the command blocks. Contrast
|
||||
/// <see cref="RunAsync_WhenStaActivityIsStaleBeyondCeilingWithCommandInFlight_WritesWatchdogFault"/>,
|
||||
/// where a frozen timestamp beyond the ceiling correctly faults; here
|
||||
/// the refreshed timestamp must keep the fault suppressed and let the
|
||||
/// reply through the <c>Ready</c>-state gate.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// <c>RunAsync</c> must throw a diagnostic exception if the
|
||||
/// runtime-session factory returns null, rather than deferring the
|
||||
|
||||
@@ -85,6 +85,34 @@ public sealed class StaRuntimeTests
|
||||
Assert.True(updated);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Verifies that <see cref="StaRuntime.PumpPendingMessages"/> refreshes
|
||||
/// <see cref="StaRuntime.LastActivityUtc"/>. A long synchronous STA
|
||||
/// command (for example <c>ReadBulk</c> waiting <c>timeout_ms</c> for
|
||||
/// the first <c>OnDataChange</c>) 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.
|
||||
/// </summary>
|
||||
[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.");
|
||||
}
|
||||
|
||||
/// <summary>Verifies that InvokeAsync faults the returned task when a command raises an exception without stopping the runtime.</summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
|
||||
@@ -80,14 +80,24 @@ public sealed class StaRuntime : IDisposable
|
||||
public bool IsRunning => startedEvent.IsSet && !stoppedEvent.IsSet;
|
||||
|
||||
/// <summary>
|
||||
/// 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 <c>HeartbeatStuckCeiling</c>, 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).
|
||||
/// </summary>
|
||||
/// <returns>The number of messages pumped.</returns>
|
||||
public int PumpPendingMessages() => messagePump.PumpPendingMessages();
|
||||
public int PumpPendingMessages()
|
||||
{
|
||||
int pumpedMessages = messagePump.PumpPendingMessages();
|
||||
MarkActivity();
|
||||
return pumpedMessages;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Starts the STA thread.
|
||||
|
||||
Reference in New Issue
Block a user