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 826e1b0..30e7044 100644
--- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs
+++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs
@@ -1453,8 +1453,15 @@ public sealed class WorkerPipeSessionTests
/// 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
+ /// models the pump refresh with
+ /// , which
+ /// stamps activity at every heartbeat capture exactly as the pump's
+ /// per-iteration MarkActivity() does. The refresh has to be in
+ /// effect from construction, not from the moment the command blocks:
+ /// with a 50 ms grace, the idle window covering handshake and startup
+ /// carries no correlation id for the watchdog to suppress on, so a fake
+ /// whose activity timestamp is frozen at construction is reported
+ /// StaHung before the scenario under test even starts. Contrast
/// ,
/// where a frozen timestamp beyond the ceiling correctly faults; here
/// the refreshed timestamp must keep the fault suppressed and let the
@@ -1469,6 +1476,13 @@ public sealed class WorkerPipeSessionTests
FakeRuntimeSession runtime = new()
{
BlockDispatch = true,
+
+ // The pump refreshes STA activity on every wait iteration, so every
+ // heartbeat capture on a healthy worker sees fresh activity — while
+ // a command holds the STA and while it is idle alike. Armed before
+ // RunAsync so the very first beat, sent as soon as the session is
+ // Ready, is already covered.
+ RefreshStaActivityOnCapture = true,
};
WorkerPipeSession session = CreatePipeSession(
pipePair.WorkerStream,
@@ -1490,49 +1504,48 @@ public sealed class WorkerPipeSessionTests
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);
- }
- });
+ // Publish the in-flight shape the heartbeat then reports for the whole
+ // blocked window; only LastStaActivityUtc moves after this, refreshed by
+ // the modelled pump at each capture.
+ runtime.SetSnapshot(new WorkerRuntimeHeartbeatSnapshot(
+ DateTimeOffset.UtcNow,
+ pendingCommandCount: 1,
+ outboundEventQueueDepth: 0,
+ lastEventSequence: 0,
+ currentCommandCorrelationId: "long-bulk-read"));
// 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++)
+ int frameIndex = 0;
+ for (; frameIndex < framesToInspect; frameIndex++)
{
WorkerEnvelope envelope = await pipePair.GatewayReader
.ReadAsync(cancellation.Token);
- Assert.NotEqual(
- WorkerEnvelope.BodyOneofCase.WorkerFault,
- envelope.BodyCase);
+ AssertNotFault(envelope, frameIndex);
}
- // Stop refreshing and release the command; its reply must be delivered
- // because the session never faulted (state stayed Ready).
- pumpRefresh.Cancel();
- await refreshLoop;
+ // Release the command with the pump still running — as it is in
+ // production while the reply is marshalled off the STA. The reply must
+ // be delivered (the session never faulted, so its state stayed Ready),
+ // and no frame on the way to it may be a fault either.
runtime.ReleaseDispatch();
- WorkerEnvelope reply = await ReadUntilAsync(
- pipePair.GatewayReader,
- WorkerEnvelope.BodyOneofCase.WorkerCommandReply,
- envelope => envelope.CorrelationId == "long-bulk-read",
- cancellation.Token);
+ WorkerEnvelope reply;
+ while (true)
+ {
+ WorkerEnvelope envelope = await pipePair.GatewayReader
+ .ReadAsync(cancellation.Token);
+ AssertNotFault(envelope, frameIndex++);
+ if (envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerCommandReply
+ && envelope.CorrelationId == "long-bulk-read")
+ {
+ reply = envelope;
+ break;
+ }
+ }
+
Assert.Equal(
ProtocolStatusCode.Ok,
reply.WorkerCommandReply.Reply.ProtocolStatus.Code);
@@ -2217,6 +2230,26 @@ public sealed class WorkerPipeSessionTests
/// Predicate to match against envelope.
/// Token to cancel the asynchronous operation.
/// The matching envelope.
+ ///
+ /// Fails when the frame is a WorkerFault, naming the category and diagnostic message.
+ /// A bare body-case comparison reports only "expected not WorkerFault", which says nothing
+ /// about which watchdog or protocol path produced it — the one fact needed to tell a
+ /// regression from a harness that mis-models the runtime.
+ ///
+ /// Frame read from the gateway end.
+ /// Ordinal of the frame within the inspected run.
+ private static void AssertNotFault(WorkerEnvelope envelope, int frameIndex)
+ {
+ if (envelope.BodyCase != WorkerEnvelope.BodyOneofCase.WorkerFault)
+ {
+ return;
+ }
+
+ Assert.Fail(
+ $"Frame {frameIndex} is a WorkerFault ({envelope.WorkerFault.Category}): "
+ + envelope.WorkerFault.DiagnosticMessage);
+ }
+
private static async Task ReadUntilAsync(
WorkerFrameReader reader,
WorkerEnvelope.BodyOneofCase expectedBody,
diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/TestSupport/FakeRuntimeSession.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/TestSupport/FakeRuntimeSession.cs
index b5a4252..c7b9ebf 100644
--- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/TestSupport/FakeRuntimeSession.cs
+++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/TestSupport/FakeRuntimeSession.cs
@@ -127,11 +127,54 @@ internal sealed class FakeRuntimeSession : IWorkerRuntimeSession
});
}
+ private bool refreshStaActivityOnCapture;
+
+ ///
+ /// When set, stamps the snapshot's
+ /// LastStaActivityUtc with the capture time and leaves every other field as the last
+ /// left it. Models a live STA whose pump calls
+ /// MarkActivity() on each wait iteration (StaRuntime.ThreadMain), so a healthy
+ /// worker is never captured stale — which a watchdog test needs to hold for the whole
+ /// session, including the handshake window before any command exists for the watchdog to
+ /// suppress on. A test-owned refresh loop cannot hold it: it is a thread-pool continuation
+ /// racing a compressed grace, and the gap between this fake being constructed and that loop's
+ /// first tick is already enough to look hung.
+ ///
+ public bool RefreshStaActivityOnCapture
+ {
+ get
+ {
+ lock (gate)
+ {
+ return refreshStaActivityOnCapture;
+ }
+ }
+
+ set
+ {
+ lock (gate)
+ {
+ refreshStaActivityOnCapture = value;
+ }
+ }
+ }
+
///
public WorkerRuntimeHeartbeatSnapshot CaptureHeartbeat()
{
lock (gate)
{
+ if (refreshStaActivityOnCapture)
+ {
+ snapshot = new WorkerRuntimeHeartbeatSnapshot(
+ DateTimeOffset.UtcNow,
+ snapshot.PendingCommandCount,
+ snapshot.OutboundEventQueueDepth,
+ snapshot.LastEventSequence,
+ snapshot.CurrentCommandCorrelationId,
+ snapshot.StaCallInProgress);
+ }
+
return snapshot;
}
}