From 7da52b65b7002fd19bcade77df7cc64389981fca Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Tue, 18 Aug 2026 06:46:08 -0400 Subject: [PATCH] =?UTF-8?q?fix(worker-tests):=20model=20the=20STA=20pump?= =?UTF-8?q?=20at=20heartbeat=20capture=20=E2=80=94=20WorkerPipeSessionTest?= =?UTF-8?q?s=20long-in-flight=20repro?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RunAsync_LongInFlightCommandThatKeepsPumping_DoesNotFaultAndDeliversReply failed deterministically on the Windows box with a StaHung fault whose command_method was empty and whose staleness was 233 ms — i.e. a fault raised with no command in flight, before the scenario under test began. FakeRuntimeSession stamps LastStaActivityUtc once, at construction, and the test only started refreshing it after the blocked dispatch signalled. Everything between those two points — handshake, STA init, the first heartbeat — captured a snapshot already stale past the compressed 50 ms grace, with no correlation id for the watchdog to suppress on, so the watchdog correctly reported the fake as hung. The harness, not the product, was wrong: StaRuntime.ThreadMain calls MarkActivity() on every WaitForWorkOrMessages iteration, so a live worker is never captured stale, idle or busy. Model that where it belongs — FakeRuntimeSession.RefreshStaActivityOnCapture (opt in, default off) stamps activity at each CaptureHeartbeat and leaves the rest of the snapshot alone — and arm it before RunAsync so the first beat is covered. The test-owned refresh loop goes away with it; a thread-pool loop racing a compressed grace could not have held the invariant anyway. Scenario intent is unchanged and slightly stronger: the command still blocks in dispatch across 30 heartbeats (~600 ms, many multiples of the 100 ms stuck ceiling), no frame may be a fault, and the reply must still arrive. The reply leg is now fault-checked too (previously it skipped frames blindly), and the pump keeps running across the release, as it does in production while the reply is marshalled off the STA. Fault assertions now report the category and diagnostic message instead of a bare body-case mismatch. Test-only change; no product code, frame protocol, or STA rule touched. --- .../Ipc/WorkerPipeSessionTests.cs | 99 ++++++++++++------- .../TestSupport/FakeRuntimeSession.cs | 43 ++++++++ 2 files changed, 109 insertions(+), 33 deletions(-) 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; } }