From dea5b84af0d8b21cb2c4e11fcfd6213c062848a6 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 01:38:46 -0500 Subject: [PATCH 1/7] test(runtime): late-burst SourceRemoved test no longer races the 5 s PostStop budget (#828) The test parks the removed child in PostStop (ShutdownGate) to hold the retired-but-not-terminated window open, but PostStop abandons a shutdown after DefaultShutdownTimeout (5 s). Under a loaded suite run more than 5 s can pass between the stop and the host processing the stale burst; the child then terminates, the host forgets it on Terminated, and the burst sent AS it is forwarded and re-retained. Reproduced deterministically with a 6 s sleep before the burst. Not a production path: a terminated actor cannot send. DriverHostActor gains a childShutdownTimeout test seam (last parameter, passed to DriverInstanceActor.Props). The test raises it to 5 min, asserts the child was really stopped (ShutdownCount) before the burst, checks the premise as a positive control (no Terminated by the time the fence has passed), and expects Terminated once the gate is released. With an 8 s sleep before the burst the fixed test passes. --- .../Drivers/DriverHostActor.cs | 24 +++++++-- .../DriverHostActorSourceRemovedTests.cs | 53 ++++++++++++++----- 2 files changed, 59 insertions(+), 18 deletions(-) diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs index 6879a972d..3d58971c9 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -491,6 +491,14 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// () means the same thing on both sides. private readonly TimeProvider _writeClock; + /// + /// Test seam: the PostStop shutdown budget every real (non-stub) child is spawned with; null = + /// . A test that parks a stopped child in its + /// PostStop to observe the retired-but-not-terminated window raises it, so the window cannot close under + /// load before the test has looked (lmxopcua#828). + /// + private readonly TimeSpan? _childShutdownTimeout; + /// Whether any redundancy snapshot has reached this host — the HistoryUpdate gate stays /// closed until one has (see ). private bool _redundancySnapshotReceived; @@ -738,6 +746,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// Ask is derived (). Null ⇒ . /// The clock the write bounds (and the host's write Ask) run on, shared with every /// child. Null ⇒ . + /// Test seam: each real child's PostStop shutdown budget; null ⇒ + /// (lmxopcua#828). /// The Akka.NET used to spawn this actor. public static Props Props( IDbContextFactory? dbFactory, @@ -776,7 +786,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS NativeAlarmAckPreflightView? ackPreflightView = null, DriverWriteBounds? writeBounds = null, TimeProvider? writeClock = null, - IScriptRunner? scriptedAlarmRunner = null) => + IScriptRunner? scriptedAlarmRunner = null, + TimeSpan? childShutdownTimeout = null) => // WARNING: this forwarding list is POSITIONAL, and Props.Create compiles it into an // expression tree. Six IActorRef? parameters and several interface-typed ones mean a // mis-ordered argument is usually type-compatible and therefore compiles clean, then binds @@ -790,7 +801,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS alarmStateStore, telemetryHub, primaryRoleSource, historizationRoleView, sharesHistorizationScopeWithLocal, applyWedgeTimeout, reprovisionBackoff, commsRecoveryAuditDelay, sharesAlarmQueueWithLocal, applyOutcomeTimeout, captureWhilePrimaryUnreachable, historyUpdateRoleView, - scripting, ackPreflightView, writeBounds, writeClock, scriptedAlarmRunner)); + scripting, ackPreflightView, writeBounds, writeClock, scriptedAlarmRunner, childShutdownTimeout)); /// Initializes a new DriverHostActor with the specified dependencies. /// Database context factory for configuration database access. @@ -885,6 +896,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// Ask is derived (). Null ⇒ . /// The clock the write bounds (and the host's write Ask) run on, shared with every /// child. Null ⇒ . + /// Test seam: each real child's PostStop shutdown budget; null ⇒ + /// (lmxopcua#828). public DriverHostActor( IDbContextFactory? dbFactory, CommonsNodeId localNode, @@ -922,8 +935,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS NativeAlarmAckPreflightView? ackPreflightView = null, DriverWriteBounds? writeBounds = null, TimeProvider? writeClock = null, - IScriptRunner? scriptedAlarmRunner = null) + IScriptRunner? scriptedAlarmRunner = null, + TimeSpan? childShutdownTimeout = null) { + _childShutdownTimeout = childShutdownTimeout; _scriptedAlarmRunner = scriptedAlarmRunner; _writeBounds = writeBounds ?? DriverWriteBounds.Default; _writeClock = writeClock ?? TimeProvider.System; @@ -3957,7 +3972,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS clusterId: clusterId, invoker: invoker, writeTimeout: _writeBounds.Timeout, - writeClock: _writeClock), + writeClock: _writeClock, + shutdownTimeout: _childShutdownTimeout), actorName); child.Tell(new DriverInstanceActor.InitializeRequested(spec.DriverConfig)); } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorSourceRemovedTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorSourceRemovedTests.cs index 4cf7ca254..0eab7f6f0 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorSourceRemovedTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorSourceRemovedTests.cs @@ -47,7 +47,8 @@ public sealed class DriverHostActorSourceRemovedTests : RuntimeActorTestBase /// Spawns a host over a real mux with two enabled drivers, waits for both children to subscribe, /// registers a replay-opting holder (so the mux retains both refs) and publishes one value per driver. - private void ArrangeTwoDriversWithRetainedValues() + /// Each child's PostStop shutdown budget (null = the production 5 s). + private void ArrangeTwoDriversWithRetainedValues(TimeSpan? childShutdownTimeout = null) { _db = NewInMemoryDbFactory(); var revA = V3ArtifactSeeder.SeedV3Deployment(_db, RevA, BothDrivers, NoRefs, enabled: true); @@ -61,7 +62,8 @@ public sealed class DriverHostActorSourceRemovedTests : RuntimeActorTestBase localRoles: new HashSet { "driver" }, dependencyMux: _mux, virtualTagEvaluator: NullVirtualTagEvaluator.Instance, - virtualTagHostOverride: CreateTestProbe().Ref)); + virtualTagHostOverride: CreateTestProbe().Ref, + childShutdownTimeout: childShutdownTimeout)); V3ArtifactSeeder.Dispatch(_host, _coordinator, revA, RevA); AwaitAssert(() => _factory.Created.Count.ShouldBe(2), PresenceBudget); // The spawn hands a child the PREVIOUS apply's set (empty on the first deploy) and the bulk push then @@ -137,28 +139,51 @@ public sealed class DriverHostActorSourceRemovedTests : RuntimeActorTestBase /// Review I-1: a burst the removed child sent after the eviction (it stops asynchronously, so it keeps /// publishing through the rest of the apply) must be dropped by the host, not forwarded to re-retain the value. + /// + /// lmxopcua#828: the window this test needs — the child RETIRED by the host but not yet TERMINATED — is held + /// open by parking the child in its PostStop on . PostStop + /// abandons a shutdown after its budget (5 s in production), the child then terminates, the host forgets it on + /// Terminated, and a burst sent AS it afterwards is forwarded — so under a loaded suite run (more than + /// 5 s from the stop to the host processing the burst) the test failed for a reason that is not a production + /// path (a terminated actor cannot send). Reproduced deterministically by sleeping 6 s before the burst. The + /// budget is therefore raised for this test, and the premise is a checked positive control: the child must + /// still be alive when the fence has passed, or the absence assertion below would mean nothing. + /// [Fact] public void A_late_burst_from_a_removed_child_does_not_re_retain_its_value() { - ArrangeTwoDriversWithRetainedValues(); + ArrangeTwoDriversWithRetainedValues(childShutdownTimeout: TimeSpan.FromMinutes(5)); var oldChild = ChildOf("DI-1"); + var watcher = CreateTestProbe(); + watcher.Watch(oldChild); // Hold the removed child in its PostStop so it has NOT terminated when the stale burst lands; a terminated // actor cannot send, and the host forgets a retired child on Terminated. var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); Driver("DI-1").ShutdownGate = gate; + try + { + var revB = V3ArtifactSeeder.SeedV3Deployment(_db, RevB, [("DI-2", "Modbus2", "dev2", "flow", "Double", false)], NoRefs, enabled: true); + V3ArtifactSeeder.Dispatch(_host, _coordinator, revB, RevB); + // The child really was stopped (it is parked in PostStop now), so the burst below is a RETIRED child's. + AwaitAssert(() => Driver("DI-1").ShutdownCount.ShouldBe(1), PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); - var revB = V3ArtifactSeeder.SeedV3Deployment(_db, RevB, [("DI-2", "Modbus2", "dev2", "flow", "Double", false)], NoRefs, enabled: true); - V3ArtifactSeeder.Dispatch(_host, _coordinator, revB, RevB); + // The stale burst, sent AS the stopped child. + _host.Tell(ValuePublishes.Single("DI-1", Path1, 11.0, OpcUaStatus.Good, DateTime.UtcNow), oldChild); + // Positive control / ordering fence: a live driver's value sent after it is delivered, so the stale burst + // has been processed by the host by the time we look. + Driver("DI-2").FireDataChange(Path2, 23.0); + AwaitHolder(Path2, 23.0); + // Premise: the burst came from a retired child that had NOT terminated (it is still parked on the gate). + watcher.HasMessages.ShouldBeFalse("the removed child terminated before the fence; the test premise is void"); - // The stale burst, sent AS the stopped child. - _host.Tell(ValuePublishes.Single("DI-1", Path1, 11.0, OpcUaStatus.Good, DateTime.UtcNow), oldChild); - // Positive control / ordering fence: a live driver's value sent after it is delivered, so the stale burst - // has been processed by the host by the time we look. - Driver("DI-2").FireDataChange(Path2, 23.0); - AwaitHolder(Path2, 23.0); - - Replayed().Keys.ToArray().ShouldBe(new[] { Path2 }, "the removed child's late burst re-retained its last Good"); - gate.TrySetResult(); + Replayed().Keys.ToArray().ShouldBe(new[] { Path2 }, "the removed child's late burst re-retained its last Good"); + } + finally + { + gate.TrySetResult(); + } + // And the gate really was what held it: released, the child terminates. + watcher.ExpectTerminated(oldChild, PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); } /// Review m-3: a config edit respawns the driver (stop + spawn); the OLD object's value is evicted, and From f7ca2c582679fbcf7441cd0c9a806193b70db5c9 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 01:38:46 -0500 Subject: [PATCH 2/7] fix(scripting): out-of-process soft deadline measures the script's own run, not the pipe or a starved worker (#827) The soft deadline T was armed on the host at submit, so the pipe round trip and the time a starved worker waits to be scheduled counted against the script: under load a trivial script reported 'script timed out', in Shadow mode that was a script mismatch, and at T + G the host killed the worker and struck the key. - Protocol (additive, negotiated): Hello/Ready.worker_soft_clock and EvalResults.soft_expired_eval_ids. protocol_version stays 1; a worker that does not echo the flag keeps the pre-#827 host clock. - Worker: the watchdog runs each run's soft clock from the moment an evaluation thread starts it and reports the crossing (woken early for budgets shorter than its 50 ms poll). It still never cancels by itself; its fail-fast is anchored at the soft notice + G + margin. - Host (PendingEval): the soft notice answers the caller and arms the kill G later. The host's own timer becomes a transport deadline, T + G from submit (restarted at a deferred run's start); passing it with no result and no notice yields TransportTimedOut -> Unavailable 'script worker unresponsive' (BadResourceUnavailable, recovery = the run's completion), no strike, no kill, no Cancel. A re-trigger refused by that still-unanswered run is Unavailable too, not 'previous run still running'. Counted as otopcua.script.ipc.timeouts{kind}. - Frame writer: coalescing now carries soft_expired_eval_ids (it silently dropped them); the host processes soft notices before results in a frame. - Shadow: unresponsive is shadow_unavailable (Unavailable), never a mismatch. Tests: WorkerSoftClockTests suspends the real worker (SIGSTOP via libc kill / NtSuspendProcess) - a stall past T still returns 7; a stall past T + G is a runner fault with nothing struck; a late-starting runaway is still timed out at T from its start and killed G after; Shadow stays match / shadow_unavailable; both clocks enforce T and G; PendingEval state machine. --- CLAUDE.md | 2 +- docs/VirtualTags.md | 17 +- .../OutOfProcessScriptRunner.cs | 49 ++- .../WorkerConnection.cs | 151 ++++++-- .../Protos/script_worker.proto | 15 + .../ScriptWorkerFramePipe.cs | 1 + .../IScriptRunner.cs | 6 + .../ShadowScriptRunner.cs | 2 +- .../WorkerHost.cs | 107 +++++- .../ShadowScriptRunnerTests.cs | 2 + .../WorkerSoftClockTests.cs | 339 ++++++++++++++++++ 11 files changed, 649 insertions(+), 42 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs diff --git a/CLAUDE.md b/CLAUDE.md index c70af7e8e..39413236e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -395,7 +395,7 @@ lists `ScriptedAlarm` — never by default) in `ZB.MOM.WW.OtOpcUa.ScriptWorker` (`src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess`, protocol in `Core.Scripting.Protocol`), shipped in `script-worker/` beside the host (self-contained + started via its apphost on a `-r win-x64 --self-contained` publish, through the muxer otherwise); a runner-side failure publishes `BadResourceUnavailable` (ScadaBridge-visible, ScadaBridge#112) and the seam re-evaluates when the worker is back. `IScriptRunner.RunBatch` sends ready evaluations as one frame per worker (Calc pump, VT coalescer, -alarm dirty set); a run queued behind busy worker threads has its deadlines restarted when it starts, so it is never blamed for a sibling. Optional `Scripting:Worker:PeCacheDir` = HMAC'd compiled-PE cache (restart without Roslyn); optional (Windows) `Scripting:Worker:Identity` = `NT SERVICE\OtOpcUaScriptWorker` runs workers under a restricted token, set up by `scripts/install/Install-ScriptWorkerIsolation.ps1` (ACLs + outbound firewall block; `docs/ServiceHosting.md`); the `script-runner` +alarm dirty set); a run queued behind busy worker threads has its deadlines restarted when it starts, so it is never blamed for a sibling. The soft deadline is the **worker's** clock from the run's start (#827, negotiated via `Hello`/`Ready.worker_soft_clock`); the host only keeps a T + G transport deadline, and passing it is a runner fault (`script worker unresponsive` → `BadResourceUnavailable`, Shadow `shadow_unavailable`), never a script timeout. Optional `Scripting:Worker:PeCacheDir` = HMAC'd compiled-PE cache (restart without Roslyn); optional (Windows) `Scripting:Worker:Identity` = `NT SERVICE\OtOpcUaScriptWorker` runs workers under a restricted token, set up by `scripts/install/Install-ScriptWorkerIsolation.ps1` (ACLs + outbound firewall block; `docs/ServiceHosting.md`); the `script-runner` readiness check is at worst Degraded (never fails ready). The worker tests are `tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests` (`Category=ScriptWorker`, real processes; `Category=Perf` harness opt-in via `OTOPCUA_SCRIPT_PERF=1`). docker-dev overlay: `docker-dev/docker-compose.script-runner.yml`. See `docs/VirtualTags.md` diff --git a/docs/VirtualTags.md b/docs/VirtualTags.md index 5c66d01ad..a456933c9 100644 --- a/docs/VirtualTags.md +++ b/docs/VirtualTags.md @@ -58,6 +58,18 @@ processes: `script timed out after …` failure and the run token is cancelled inside the worker; at `T + G` (`Scripting:Worker:HardGrace`, default `max(1 s, T)`) the worker is **killed** (`Process.Kill(entireProcessTree)`, plus a worker-side fail-fast watchdog) — the thread, its memory and its ALC go with it. +- **`T` is the script's own run time (lmxopcua#827).** As in process (where the guard's clock starts when the + thread pool picks the run up), `T` starts when a worker evaluation thread starts the run — not at submit — so the + pipe round trip and the time a starved worker waits to be scheduled never count against the script. The worker runs + that clock and reports the crossing (`EvalResults.soft_expired_eval_ids`); the host answers its caller, sends the + cancel, and arms the kill `G` after the notice. What the host still times on its own is a **transport deadline**, + `T + G` from submit (restarted at a queued run's start): no result and no soft notice by then is the runner's fault — + the tag publishes **`BadResourceUnavailable`** (`"script worker unresponsive"`), nothing is struck or quarantined, + the worker is not killed (a really hung one goes silent and the heartbeat monitor kills it), and the seam + re-evaluates when the worker answers. A re-trigger while that run is still unanswered is the runner's fault too, not + `previous run still running`. Counted as `otopcua.script.ipc.timeouts{kind}`. The clock is negotiated in the handshake + (`Hello`/`Ready.worker_soft_clock`): a worker that does not echo it keeps the host clock from submit (the pre-#827 + contract), so a mismatched host/worker pair degrades to the old behaviour instead of losing its deadlines. - **Attribution and quarantine** live in the host and survive restarts: a deadline kill strikes exactly its key (other evaluations in flight are victims, not strikes); a crash with no deadline crossed makes every key in flight a suspect that must complete `Scripting:Worker:IsolationProbationRuns` (20) clean runs in the isolation worker, @@ -108,7 +120,7 @@ processes: "every process with this name", never a live host's worker (mxaccessgw's `OrphanWorkerTerminator` lesson). - **Observability.** On `OtOpcUaTelemetry.Meter`: `otopcua.script.runner.state` (0 ready / 1 degraded / 2 circuit open), `otopcua.script.worker.up` / `.inflight` / `.queue_depth` / `.heap_bytes` (tag `shard`; queue and heap from - the worker heartbeat), `otopcua.script.worker.starts` / `.exits` (`reason` adds `idle`), `otopcua.script.ipc.roundtrip` + the worker heartbeat), `otopcua.script.worker.starts` / `.exits` (`reason` adds `idle`), `otopcua.script.ipc.roundtrip`, `otopcua.script.ipc.timeouts` (transport deadline, lmxopcua#827) and `otopcua.script.eval.duration` (µs, tag `kind`; no per-key tag), `otopcua.script.compile` (`cache` = `hit` | `miss` | `rejected` | `uncached` | `negative`) and `otopcua.script.isolation.suspects`. The `script-runner` health check is on the **ready** tier and is at worst **Degraded** (still 200): a sick runner never fails readiness, so a pair @@ -168,7 +180,8 @@ what is published** — and each compiled evaluation is then mirrored to the `Sc - **Compared:** value (structural for arrays), exact CLR type, status (Ok vs failed) and failure class (`compile`, `budget` = timed out / killed / refused / quarantined, `threw` with its message, `unsupported_type`). A shadow outcome - that is the runner's own fault (worker restarted, none ready, worker busy) is **not compared** + that is the runner's own fault (worker restarted, none ready, worker busy, worker unresponsive — a stalled or starved + worker past the transport deadline, lmxopcua#827) is **not compared** (`result = shadow_unavailable`). An input or result the protocol cannot carry (design §14 #11) counts as a mismatch with reason `unsupported_type` — out of process that evaluation would publish Bad. - **Never reaches a published value or readiness.** The mirror copies the inputs and queues the shadow run on the pool; diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs index 4497440c2..06b748b4d 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs @@ -100,6 +100,10 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, "otopcua.script.eval.duration", unit: "us", description: "Worker-measured run time of one out-of-process evaluation (tag: kind)."); + private static readonly Counter TransportTimeoutCounter = OtOpcUaTelemetry.Meter.CreateCounter( + "otopcua.script.ipc.timeouts", unit: "{eval}", + description: "Out-of-process evaluations answered 'script worker unresponsive': no result and no soft-deadline notice from the worker within T + G of submit — a starved or stalled worker, not a slow script (tag: kind; lmxopcua#827)."); + static OutOfProcessScriptRunner() { // Gauges observe every live runner (one per node in production; tests make several). @@ -372,11 +376,7 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, return ScriptRunOutcome.Failed(knownFailure); } - if (Interlocked.CompareExchange(ref state.InFlight, 1, 0) != 0) - { - ScriptRunGuard.RefusedCounter.Add(1, new KeyValuePair("script", key)); - return ScriptRunOutcome.Failed(ScriptRunGuard.PreviousRunStillRunningReason); - } + if (Interlocked.CompareExchange(ref state.InFlight, 1, 0) != 0) return Refused(key, state); var handedOff = false; SemaphoreSlim? gate = null; @@ -533,8 +533,7 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, if (Interlocked.CompareExchange(ref state.InFlight, 1, 0) != 0) { - ScriptRunGuard.RefusedCounter.Add(1, new KeyValuePair("script", key)); - answered = ScriptRunOutcome.Failed(ScriptRunGuard.PreviousRunStillRunningReason); + answered = Refused(key, state); return BatchPreparation.Answered; } state.ScriptLogger = invocation.ScriptLogger; @@ -559,6 +558,7 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, { if (Interlocked.Exchange(ref released, 1) != 0) return; SlotReleasingForTest?.Invoke(key); + Volatile.Write(ref state.StalledRun, null); Interlocked.Exchange(ref state.InFlight, 0); heldGate?.Release(); }; @@ -583,6 +583,21 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, return ScriptRunOutcome.Failed($"{ScriptEvalFailureReasons.TimedOutPrefix}{invocation.Timeout.TotalSeconds:F1}s"); } + if (ReferenceEquals(result, PendingEval.TransportTimedOut)) + { + // lmxopcua#827: T + G passed with neither a result nor a soft-deadline notice. The worker runs the soft + // clock from the run's start, so this is the pipe or a starved / stalled worker — the runner's fault, + // never the script's: no timeout strike, no Cancel (the run may not even have started), and the slot + // stays with the run until the worker answers. The seam re-evaluates when it does (or when the worker + // is replaced, if the heartbeat monitor kills it). + TransportTimeoutCounter.Add(1, new KeyValuePair("kind", KindTag(invocation.Kind))); + // Read by a refusal only while it is incomplete, so a write that loses a race with the slot release + // is harmless. + Volatile.Write(ref state.StalledRun, pending.Completion); + if (ct.IsCancellationRequested) return ScriptRunOutcome.Failed("evaluation cancelled"); + return ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerUnresponsiveReason) with { Recovery = pending.Completion }; + } + if (result is null) return Unavailable(ScriptRunOutcome.WorkerRestartedReason, RouteFor(key, state), connection); var kindTag = new KeyValuePair("kind", KindTag(invocation.Kind)); RoundTripHistogram.Record(Stopwatch.GetElapsedTime(submitted).TotalMicroseconds, kindTag); @@ -717,6 +732,20 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, private static ScriptRunOutcome Unavailable(string reason, WorkerShard shard, WorkerConnection? failed) => ScriptRunOutcome.Unavailable(reason) with { Recovery = shard.WhenRecoveredAsync(failed) }; + /// + /// The answer to an evaluation whose key's previous run still holds its single-flight slot: the usual + /// script-attributable refusal — unless that run was answered "worker unresponsive" (lmxopcua#827) and the + /// worker still has not answered it, in which case nothing says the script is slow, and the refusal is the + /// runner's fault too (re-evaluated when the stalled run completes). + /// + private static ScriptRunOutcome Refused(string key, KeyState state) + { + if (Volatile.Read(ref state.StalledRun) is { IsCompleted: false } stalled) + return ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerUnresponsiveReason) with { Recovery = stalled }; + ScriptRunGuard.RefusedCounter.Add(1, new KeyValuePair("script", key)); + return ScriptRunOutcome.Failed(ScriptRunGuard.PreviousRunStillRunningReason); + } + private void NoteLoad(WorkerConnection.LoadEntry entry) { // Counted once per compile, by the caller that waited for it (an already-loaded key is not a load). @@ -986,5 +1015,11 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, public volatile bool Suspect; public string? QuarantineReason; public ILogger? ScriptLogger; + + /// + /// The completion of this key's in-flight run while it is answered "worker unresponsive" and not yet answered + /// by the worker (lmxopcua#827); null otherwise. Cleared by the slot release. + /// + public Task? StalledRun; } } diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs index 85b6f6d94..a9fce4fa4 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs @@ -81,6 +81,10 @@ internal sealed record BatchSubmission(string Key, EvalRequest Request, TimeSpan /// Windows only: the worker identity's SID (Scripting:Worker:Identity). Set, the worker starts under a restricted /// token () and the pipe ACL admits it beside the host; null keeps the host's token. /// +/// +/// Ask the worker to run the soft clock from each run's start (lmxopcua#827; see ). +/// False (tests only) keeps the pre-#827 host clock, which is also what a worker that does not echo it gets. +/// internal sealed record WorkerLaunchSpec( string Shard, string WorkerPath, @@ -94,7 +98,8 @@ internal sealed record WorkerLaunchSpec( string? MarkerDirectory = null, TimeSpan? LoadTimeout = null, int CompileThreads = 0, - string? WorkerSid = null) + string? WorkerSid = null, + bool WorkerSoftClock = true) { /// The default compile timeout. public static readonly TimeSpan DefaultLoadTimeout = TimeSpan.FromSeconds(60); @@ -151,8 +156,9 @@ internal sealed class WorkerConnection : IAsyncDisposable private WorkerConnection(string shard, Process process, NamedPipeServerStream pipe, IDisposable? job, IDisposable? stdio, string workDir, ILogger log, Action onLogs, TimeSpan heartbeatInterval, TimeSpan loadTimeout, - Action onExiting, string? markerPath) + Action onExiting, string? markerPath, bool workerSoftClock) { + WorkerSoftClock = workerSoftClock; _stdio = stdio; _loadTimeout = loadTimeout; _onExiting = onExiting; @@ -178,6 +184,15 @@ internal sealed class WorkerConnection : IAsyncDisposable /// The worker's process id. public int Pid { get; } + /// + /// lmxopcua#827: the worker runs each evaluation's soft clock from the moment an evaluation thread starts it and + /// reports the crossing, so the soft deadline measures the script's own execution — not the pipe, not the time + /// a starved worker takes to be scheduled. The host keeps a transport deadline of its own instead (see + /// ). Negotiated: asked for in Hello, true only when the worker echoed it in + /// Ready; otherwise the host's clock, started at submit, is the soft deadline (the pre-#827 contract). + /// + public bool WorkerSoftClock { get; } + /// Completes when the worker has ended, with why. public Task Exited => _exited.Task; @@ -300,19 +315,19 @@ internal sealed class WorkerConnection : IAsyncDisposable HeartbeatIntervalMs = (uint)spec.HeartbeatInterval.TotalMilliseconds, MaxLogLines = 1000, CompileThreads = (uint)Math.Max(0, spec.CompileThreads), + WorkerSoftClock = spec.WorkerSoftClock, }; if (spec.PeCacheDirectory is { } cacheDir && spec.PeCacheKey is { } cacheKey) { hello.PeCacheDir = cacheDir; hello.PeCacheKey = Google.Protobuf.ByteString.CopyFrom(cacheKey); } - if (ScriptWorkerFrameCodec.UseBlockingIo) - await HandshakeBlockingAsync(pipe, hello, nonce, process.Id, handshakeCts.Token).ConfigureAwait(false); - else - await HandshakeAsync(pipe, hello, nonce, process.Id, handshakeCts.Token).ConfigureAwait(false); + var ready = ScriptWorkerFrameCodec.UseBlockingIo + ? await HandshakeBlockingAsync(pipe, hello, nonce, process.Id, handshakeCts.Token).ConfigureAwait(false) + : await HandshakeAsync(pipe, hello, nonce, process.Id, handshakeCts.Token).ConfigureAwait(false); var connection = new WorkerConnection(spec.Shard, process, pipe, job, stdio, workDir, log, onLogs, spec.HeartbeatInterval, - spec.EffectiveLoadTimeout, onExiting, markerPath); + spec.EffectiveLoadTimeout, onExiting, markerPath, workerSoftClock: spec.WorkerSoftClock && ready.WorkerSoftClock); connection.Begin(); return connection; } @@ -557,7 +572,7 @@ internal sealed class WorkerConnection : IAsyncDisposable request.EvalId = (ulong)Interlocked.Increment(ref _nextEvalId); Interlocked.Exchange(ref _lastUsedTimestamp, Stopwatch.GetTimestamp()); var pending = new PendingEval(key, request.EvalId, onFinished, _dispatcher); - if (softTimeout is { } soft) pending.Arm(soft, hardGrace, OnHardDeadline); + if (softTimeout is { } soft) pending.Arm(soft, hardGrace, OnHardDeadline, WorkerSoftClock); _pending[request.EvalId] = pending; var batch = new EvalBatch(); batch.Items.Add(request); @@ -586,7 +601,7 @@ internal sealed class WorkerConnection : IAsyncDisposable var item = items[i]; item.Request.EvalId = (ulong)Interlocked.Increment(ref _nextEvalId); var pending = new PendingEval(item.Key, item.Request.EvalId, item.OnFinished, _dispatcher); - pending.Arm(item.SoftTimeout, item.HardGrace, OnHardDeadline); + pending.Arm(item.SoftTimeout, item.HardGrace, OnHardDeadline, WorkerSoftClock); _pending[item.Request.EvalId] = pending; batch.Items.Add(item.Request); pendings[i] = pending; @@ -764,6 +779,16 @@ internal sealed class WorkerConnection : IAsyncDisposable switch (frame.BodyCase) { case Envelope.BodyOneofCase.Results: + // Worker soft clock (lmxopcua#827): the script itself has run for T. Only a worker that agreed to run + // the clock may send these. Before the results: the writer coalesces frames, and a notice the worker + // sent ahead of the same run's result must still answer the caller with its timeout, as in process. + if (frame.Results.SoftExpiredEvalIds.Count > 0 && !WorkerSoftClock) + throw new ScriptWorkerProtocolException("soft-deadline notice from a worker that does not run the soft clock"); + foreach (var id in frame.Results.SoftExpiredEvalIds) + { + if (_pending.TryGetValue(id, out var expired)) expired.SoftExpired(); + else CheckIssued(id, "soft-deadline notice"); + } foreach (var result in frame.Results.Items) { // An eval id with no outstanding request is a protocol violation (design §6 "what the host @@ -1026,18 +1051,45 @@ internal sealed class WorkerConnection : IAsyncDisposable /// /// One evaluation outstanding on a worker. Its deadlines share ONE timer (lmxopcua#789: the per-eval /// WaitAsync(timeout) timer, the hard-deadline CancellationTokenSource and their continuations were a -/// measurable share of the host's CPU): it fires at the soft deadline T and is re-armed for the grace G. +/// measurable share of the host's CPU). /// /// -/// The deadlines are absolute timestamps the timer callback re-checks, so a deferral (the worker queued the run behind -/// busy evaluation threads, lmxopcua#780 batch submit) can stop the clock and a start notice can restart it — the full -/// T, then G, from the start — without a stale tick firing early. +/// +/// Worker soft clock (lmxopcua#827, ): the soft deadline T is +/// the worker's — it reports, through , that the script has itself run for T, and the +/// hard deadline G is armed from that notice. The host's timer is then only a TRANSPORT deadline: T + G from +/// submit (restarted at a deferred run's start). Passing it with neither a result nor a soft notice means the +/// pipe or the worker process is late, not the script — yields +/// (a runner fault: nothing is struck and the worker is not killed; a worker that +/// is really hung goes silent and the heartbeat monitor kills it). So a starved worker can no longer make a +/// trivial script report "timed out", which in Shadow mode read as a script mismatch. +/// +/// +/// Host clock (a worker that did not agree to the above): the timer fires at the soft deadline T, counted +/// from submit, and is re-armed for the grace G. +/// +/// +/// The deadlines are absolute timestamps the timer callback re-checks, so a deferral (the worker queued the run +/// behind busy evaluation threads, lmxopcua#780 batch submit) can stop the clock and a start notice can restart +/// it — the full T (or transport deadline), then G, from the start — without a stale tick firing early. +/// /// internal sealed class PendingEval { /// What yields when the soft deadline passed first. Never on the wire. internal static readonly EvalResult SoftTimedOut = new(); + /// + /// What yields when the transport deadline passed with no word from the worker (worker soft + /// clock only, lmxopcua#827): a runner fault, not the script's. Never on the wire. + /// + internal static readonly EvalResult TransportTimedOut = new(); + + private const int Waiting = 0; + private const int SoftReported = 1; + private const int HardFired = 2; + private const int TransportReported = 3; + private readonly TaskCompletionSource _completion = new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly TaskCompletionSource _outcome; private readonly ResultDispatcher? _dispatcher; @@ -1047,8 +1099,10 @@ internal sealed class PendingEval private Action? _onHardDeadline; private TimeSpan _soft; private TimeSpan _grace; + private bool _workerClock; private long _softDue; private long _hardDue; + private long _transportDue; private bool _deferred; private bool _started; private int _phase; @@ -1086,24 +1140,31 @@ internal sealed class PendingEval public Task Completion => _completion.Task; /// - /// What the caller waits on: the worker's result, null when the worker ended first, or - /// at the soft deadline (the run itself is still outstanding then). + /// What the caller waits on: the worker's result, null when the worker ended first, at + /// the soft deadline, or at the transport deadline (the run itself is still + /// outstanding in both of the last two cases). /// public Task Outcome => _outcome.Task; /// Starts the deadline timer. Called once, before the request is sent. - internal void Arm(TimeSpan soft, TimeSpan grace, Action onHardDeadline) + /// T. + /// G. + /// The kill. + /// The worker runs the soft clock (lmxopcua#827): this timer is the transport deadline. + internal void Arm(TimeSpan soft, TimeSpan grace, Action onHardDeadline, bool workerClock = false) { _soft = soft; _grace = grace; _onHardDeadline = onHardDeadline; + _workerClock = workerClock; HardDeadline = soft + grace; var now = Stopwatch.GetTimestamp(); _softDue = now + Ticks(soft); _hardDue = _softDue + Ticks(grace); + _transportDue = _hardDue; // Created disarmed and assigned before it can fire, so the soft tick can always re-arm it for the grace. _timer = new Timer(static s => ((PendingEval)s!).OnTimer(), this, Timeout.Infinite, Timeout.Infinite); - _timer.Change(soft, Timeout.InfiniteTimeSpan); + _timer.Change(workerClock ? soft + grace : soft, Timeout.InfiniteTimeSpan); } /// @@ -1122,7 +1183,8 @@ internal sealed class PendingEval /// /// The worker started this run: a deferred run's deadlines restart from now (T, or T + G when its soft deadline - /// had already been reported). A run that was never deferred keeps the clock it has had since submit. + /// had already been reported; with the worker soft clock, the transport deadline T + G). A run that was never + /// deferred keeps the clock it has had since submit. /// internal void Start() { @@ -1132,7 +1194,16 @@ internal sealed class PendingEval if (_timer is null || !_deferred || Volatile.Read(ref _done) != 0) return; _deferred = false; var now = Stopwatch.GetTimestamp(); - if (Volatile.Read(ref _phase) == 0) + if (_workerClock) + { + // Its soft deadline is the worker's from here; the transport deadline restarts. Already answered (a + // deferral notice that arrived after the transport deadline): nothing to re-arm — a soft notice or the + // result is what comes next. + if (Volatile.Read(ref _phase) != Waiting) return; + _transportDue = now + Ticks(_soft + _grace); + TryChange(_soft + _grace); + } + else if (Volatile.Read(ref _phase) == Waiting) { _softDue = now + Ticks(_soft); _hardDue = _softDue + Ticks(_grace); @@ -1148,31 +1219,63 @@ internal sealed class PendingEval } } + /// + /// The worker reports that the script has itself run for T (worker soft clock, lmxopcua#827): the caller gets its + /// soft timeout now — unless the transport deadline already answered it — and the hard deadline is G from here. + /// + internal void SoftExpired() + { + bool signal; + lock (_clock) + { + if (_timer is null || !_workerClock || Volatile.Read(ref _done) != 0) return; + var phase = Volatile.Read(ref _phase); + if (phase is SoftReported or HardFired) return; + _started = true; + _deferred = false; + signal = phase == Waiting; + Interlocked.Exchange(ref _phase, SoftReported); + _hardDue = Stopwatch.GetTimestamp() + Ticks(_grace); + TryChange(_grace); + } + // Outside the lock: a caller with a dispatcher resumes synchronously inside TrySetResult. + if (signal) _outcome.TrySetResult(SoftTimedOut); + } + private void OnTimer() { if (Volatile.Read(ref _done) != 0) return; - bool soft = false, hard = false; + bool soft = false, hard = false, transport = false; lock (_clock) { if (_deferred || Volatile.Read(ref _done) != 0) return; var now = Stopwatch.GetTimestamp(); - if (Volatile.Read(ref _phase) == 0) + var phase = Volatile.Read(ref _phase); + if (phase == Waiting && _workerClock) + { + // A tick armed before a deferral/start can land early: re-arm for what is left. + if (now < _transportDue) { TryChange(Stopwatch.GetElapsedTime(now, _transportDue)); return; } + Interlocked.Exchange(ref _phase, TransportReported); + transport = true; + } + else if (phase == Waiting) { // A tick armed before a deferral/start can land early: re-arm for what is left. if (now < _softDue) { TryChange(Stopwatch.GetElapsedTime(now, _softDue)); return; } - Interlocked.Exchange(ref _phase, 1); + Interlocked.Exchange(ref _phase, SoftReported); soft = true; TryChange(_hardDue > now ? Stopwatch.GetElapsedTime(now, _hardDue) : TimeSpan.Zero); } - else if (Volatile.Read(ref _phase) == 1) + else if (phase == SoftReported) { if (now < _hardDue) { TryChange(Stopwatch.GetElapsedTime(now, _hardDue)); return; } - Interlocked.Exchange(ref _phase, 2); + Interlocked.Exchange(ref _phase, HardFired); hard = true; } } // Outside the lock: a caller with a dispatcher resumes synchronously inside TrySetResult. if (soft) _outcome.TrySetResult(SoftTimedOut); + if (transport) _outcome.TrySetResult(TransportTimedOut); if (hard && Volatile.Read(ref _done) == 0) _onHardDeadline?.Invoke(this); } diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/Protos/script_worker.proto b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/Protos/script_worker.proto index 7e918208e..5d73da467 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/Protos/script_worker.proto +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/Protos/script_worker.proto @@ -41,12 +41,17 @@ message Hello { string pe_cache_dir = 8; // the node's compiled-PE cache (design §7); empty = no cache bytes pe_cache_key = 9; // its HMAC key; the worker loads only entries that verify under it uint32 compile_threads = 10; // dedicated compile threads; 0 = the worker's default (min(8, max(1, CPUs/2))) + // lmxopcua#827: the host asks the worker to run each evaluation's SOFT clock itself, from when an evaluation thread + // starts the run, and to report the crossing in EvalResults.soft_expired_eval_ids. The host then keeps only a + // transport deadline of its own (see EvalResults). Honoured only when the worker echoes Ready.worker_soft_clock. + bool worker_soft_clock = 11; } message Ready { string nonce = 1; int32 pid = 2; string runtime_version = 3; + bool worker_soft_clock = 4; // lmxopcua#827: this worker runs the soft clock (Hello.worker_soft_clock was set) } enum ContextKind { @@ -141,10 +146,20 @@ message EvalResult { // when the evaluation appears in `started_eval_ids`, which an evaluation thread sends alongside the result of the run it // just finished. So a run waiting behind a sibling is never timed out, struck or killed for the sibling's time. // Either list may be the only content of a frame. A host that predates these fields ignores them. +// +// Worker soft clock (lmxopcua#827, negotiated by Hello/Ready.worker_soft_clock): the soft deadline T measures the +// script's OWN execution — the worker starts it when an evaluation thread starts the run and reports the crossing in +// `soft_expired_eval_ids` (it does not cancel the run itself: the host answers its caller first, then sends Cancel, +// keeping the in-process ordering). The host arms the hard deadline G from that notice. What the host keeps on its own +// clock is a TRANSPORT deadline, T + G from submit (restarted at a deferred run's start): no result and no soft notice +// by then means the pipe or the worker process is late — the caller is answered as a runner fault ("script worker +// unresponsive", BadResourceUnavailable), never as a script timeout, and nothing is struck. A starved or stalled worker +// therefore cannot make a trivial script look over budget; a hung one is still caught by the heartbeat silence kill. message EvalResults { repeated EvalResult items = 1; repeated uint64 started_eval_ids = 2; repeated uint64 deferred_eval_ids = 3; + repeated uint64 soft_expired_eval_ids = 4; } message ScriptError { diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ScriptWorkerFramePipe.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ScriptWorkerFramePipe.cs index 98a90ba7e..260be8bbf 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ScriptWorkerFramePipe.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ScriptWorkerFramePipe.cs @@ -237,6 +237,7 @@ public sealed class ScriptWorkerFrameWriter results.Items.AddRange(envelope.Results.Items); results.StartedEvalIds.AddRange(envelope.Results.StartedEvalIds); results.DeferredEvalIds.AddRange(envelope.Results.DeferredEvalIds); + results.SoftExpiredEvalIds.AddRange(envelope.Results.SoftExpiredEvalIds); break; default: if (evals is not null) { Append(new Envelope { Eval = evals }); evals = null; } diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/IScriptRunner.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/IScriptRunner.cs index 1a1e79215..77953ac11 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/IScriptRunner.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/IScriptRunner.cs @@ -125,6 +125,12 @@ public sealed record ScriptRunOutcome(ScriptRunStatus Status, object? Value, str /// Reason when the worker could not take more work (the out-of-process runner's back-pressure answer). public const string WorkerBusyReason = "script worker busy"; + /// + /// Reason when the worker neither answered nor reported the script over its budget within T + G of submit — the + /// pipe or a starved / stalled worker process, not the script (lmxopcua#827). + /// + public const string WorkerUnresponsiveReason = "script worker unresponsive"; + /// /// For : completes when the runner can run this script again (its worker /// is back), so the seam re-evaluates it then with its latest inputs (design §9, "re-evaluated on the shard's diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ShadowScriptRunner.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ShadowScriptRunner.cs index 403598d5c..647e76432 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ShadowScriptRunner.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ShadowScriptRunner.cs @@ -47,7 +47,7 @@ public enum ScriptShadowVerdict /// The shadow run disagreed; names how. Mismatch = 1, - /// The shadow runner could not run the script (worker restarted, none ready, worker busy): not compared. + /// The shadow runner could not run the script (worker restarted, none ready, worker busy, worker unresponsive): not compared. ShadowUnavailable = 2, } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker/WorkerHost.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker/WorkerHost.cs index 5b0d99854..4d9250859 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker/WorkerHost.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker/WorkerHost.cs @@ -29,6 +29,9 @@ internal sealed class WorkerHost /// The watchdog fails fast this long after the host's own hard deadline, so the host normally wins. private static readonly TimeSpan WatchdogMargin = TimeSpan.FromMilliseconds(500); + /// The watchdog's longest sleep between passes. + private static readonly TimeSpan WatchdogPoll = TimeSpan.FromMilliseconds(50); + private readonly NamedPipeClientStream _pipe; private readonly string _nonce; private readonly int _parentPid; @@ -75,6 +78,19 @@ internal sealed class WorkerHost private TimeSpan _hardGrace = TimeSpan.FromSeconds(1); private TimeSpan _heartbeatInterval = TimeSpan.FromSeconds(1); private ScriptPeCacheStore? _peCache; + + /// + /// lmxopcua#827: this worker runs each evaluation's soft clock from the run's start and reports the crossing + /// (EvalResults.soft_expired_eval_ids); the host asked for it in Hello and we echoed it in + /// Ready. Off (an older host): the host's own clock, started at submit, is the soft deadline. + /// + private bool _workerSoftClock; + + /// Wakes the watchdog early when a run's soft deadline is sooner than its next planned pass. + private readonly AutoResetEvent _watchdogWake = new(false); + + /// of the watchdog's next planned pass. + private long _watchdogNextPass; private long _seq; public WorkerHost(NamedPipeClientStream pipe, string nonce, int parentPid) @@ -109,6 +125,7 @@ internal sealed class WorkerHost Nonce = _nonce, Pid = Environment.ProcessId, RuntimeVersion = Environment.Version.ToString(), + WorkerSoftClock = _workerSoftClock, }, }); if (ScriptWorkerFrameCodec.UseBlockingIo) ScriptWorkerFrameCodec.Write(_pipe, ready, _maxFrameBytes); @@ -140,6 +157,7 @@ internal sealed class WorkerHost if (hello.HardGraceMs > 0) _hardGrace = TimeSpan.FromMilliseconds(hello.HardGraceMs); if (hello.HeartbeatIntervalMs > 0) _heartbeatInterval = TimeSpan.FromMilliseconds(hello.HeartbeatIntervalMs); if (hello.CompileThreads > 0) _compileThreads = (int)Math.Min(hello.CompileThreads, 64); + _workerSoftClock = hello.WorkerSoftClock; if (!string.IsNullOrEmpty(hello.PeCacheDir) && hello.PeCacheKey.Length == ScriptPeCacheStore.KeyLength) { // Design §7: the host owns the key; it reaches us only here, never through the environment or a file. @@ -425,11 +443,17 @@ internal sealed class WorkerHost var logger = _scriptLogger.ForContext(ForwardingLogSink.KeyProperty, request.Key); var soft = TimeSpan.FromMilliseconds(Math.Max(1, request.SoftTimeoutMs)); - // Cancelled by the host's Cancel frame at the soft deadline, never by our own clock (see the proto). + // Cancelled by the host's Cancel frame at the soft deadline, never by our own clock (see the proto): with the + // worker soft clock (lmxopcua#827) the watchdog only REPORTS the crossing; the host answers its caller, then + // cancels. using var cts = new CancellationTokenSource(); var grace = request.HardGraceMs > 0 ? TimeSpan.FromMilliseconds(request.HardGraceMs) : _hardGrace; - var running = new RunningEval(request.Key, Stopwatch.GetTimestamp(), soft + grace + WatchdogMargin, cts); + var started = Stopwatch.GetTimestamp(); + var running = new RunningEval(request.EvalId, request.Key, started, soft, grace, cts); _running[(long)request.EvalId] = running; + // The watchdog plans its next pass at most WatchdogPoll ahead; a budget shorter than that wakes it now, so the + // soft notice is not late by up to a poll interval. + if (_workerSoftClock && running.SoftDue < Interlocked.Read(ref _watchdogNextPass)) _watchdogWake.Set(); var before = GC.GetAllocatedBytesForCurrentThread(); var sw = Stopwatch.StartNew(); try @@ -465,22 +489,68 @@ internal sealed class WorkerHost return result; } + /// + /// The watchdog: reports soft-deadline crossings (worker soft clock, lmxopcua#827) and fails the process fast for + /// a run past its hard deadline that the host has not killed. + /// + /// + /// The hard deadline it enforces is the host's plus , from the same anchor the host + /// uses: the run's start (T + G) on the host clock, or — with the worker soft clock — the soft notice it sent + /// (G), which is where the host arms its kill. So the host normally wins and the exit names the culprit. + /// private void WatchdogLoop() { while (true) { - Thread.Sleep(50); + var now = Stopwatch.GetTimestamp(); + var nextPass = now + Ticks(WatchdogPoll); + List? expired = null; foreach (var (_, run) in _running) { - if (Stopwatch.GetElapsedTime(run.StartedTimestamp) > run.FailFastAfter) + if (_workerSoftClock) { - // The host should have killed us already; this names the culprit in the exit/dump. - Environment.FailFast($"ScriptWorker: script '{run.Key}' exceeded its hard deadline ({run.FailFastAfter.TotalMilliseconds:F0} ms)"); + if (run.SoftReportedAt == 0) + { + if (now >= run.SoftDue) + { + run.SoftReportedAt = now; + (expired ??= []).Add(run.EvalId); + } + else if (run.SoftDue < nextPass) + { + nextPass = run.SoftDue; + } + } + else if (now - run.SoftReportedAt > Ticks(run.Grace + WatchdogMargin)) + { + FailFast(run, run.Grace + WatchdogMargin, "its soft deadline"); + } + } + else if (now - run.StartedTimestamp > Ticks(run.Soft + run.Grace + WatchdogMargin)) + { + FailFast(run, run.Soft + run.Grace + WatchdogMargin, "its start"); } } + if (expired is not null) + { + var notice = new EvalResults(); + notice.SoftExpiredEvalIds.AddRange(expired); + Send(new Envelope { Results = notice }); + } + Interlocked.Exchange(ref _watchdogNextPass, nextPass); + var wait = Stopwatch.GetElapsedTime(Stopwatch.GetTimestamp(), nextPass); + if (wait > TimeSpan.Zero) _watchdogWake.WaitOne(wait); + } + + static void FailFast(RunningEval run, TimeSpan after, string anchor) + { + // The host should have killed us already; this names the culprit in the exit/dump. + Environment.FailFast($"ScriptWorker: script '{run.Key}' exceeded its hard deadline ({after.TotalMilliseconds:F0} ms after {anchor})"); } } + private static long Ticks(TimeSpan span) => (long)(span.TotalSeconds * Stopwatch.Frequency); + private async Task HeartbeatLoopAsync() { using var timer = new PeriodicTimer(_heartbeatInterval); @@ -609,7 +679,30 @@ internal sealed class WorkerHost Environment.Exit(ScriptWorkerEnvironment.ExitForbiddenAssembly); } - private sealed record RunningEval(string Key, long StartedTimestamp, TimeSpan FailFastAfter, CancellationTokenSource Cancellation); + /// One evaluation on an evaluation thread, as the watchdog sees it. + private sealed class RunningEval(ulong evalId, string key, long startedTimestamp, TimeSpan soft, TimeSpan grace, + CancellationTokenSource cancellation) + { + public ulong EvalId { get; } = evalId; + + public string Key { get; } = key; + + /// when the evaluation thread started the run. + public long StartedTimestamp { get; } = startedTimestamp; + + /// T. + public TimeSpan Soft { get; } = soft; + + /// The run's start + T. + public long SoftDue { get; } = startedTimestamp + Ticks(soft); + + public TimeSpan Grace { get; } = grace; + + public CancellationTokenSource Cancellation { get; } = cancellation; + + /// When the watchdog reported the soft crossing to the host; 0 = not yet. Watchdog thread only. + public long SoftReportedAt { get; set; } + } /// One loaded script: its compiled evaluator and its input slot order. private sealed class LoadedScript : IDisposable diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ShadowScriptRunnerTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ShadowScriptRunnerTests.cs index a0f29ae58..f320c3fd5 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ShadowScriptRunnerTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ShadowScriptRunnerTests.cs @@ -83,6 +83,8 @@ public sealed class ShadowScriptRunnerTests // The runner's fault, not the script's: not compared. { ScriptRunOutcome.Ok(42), ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerRestartedReason), ScriptShadowVerdict.ShadowUnavailable, null }, { ScriptRunOutcome.Ok(42), ScriptRunOutcome.Failed(ScriptRunOutcome.WorkerBusyReason), ScriptShadowVerdict.ShadowUnavailable, null }, + // lmxopcua#827: a stalled / starved worker past the transport deadline is the runner's fault, not a mismatch. + { ScriptRunOutcome.Ok(42), ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerUnresponsiveReason), ScriptShadowVerdict.ShadowUnavailable, null }, }; [Theory] diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs new file mode 100644 index 000000000..359e3f86f --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs @@ -0,0 +1,339 @@ +using System.Diagnostics; +using System.Runtime.InteropServices; +using System.Security.Cryptography; +using System.Text; +using Serilog; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Scripting; +using ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess; +using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.V1; + +namespace ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests; + +/// +/// lmxopcua#827: the soft deadline T measures the script's OWN execution — the worker runs it from the moment an +/// evaluation thread starts the run — so the pipe round trip and the time a starved worker waits to be scheduled +/// no longer count against the script. The host keeps a transport deadline (T + G from submit) and reports passing +/// it as a runner fault (, BadResourceUnavailable), never as a +/// script timeout. The starved worker is made real here: its process is suspended (SIGSTOP / NtSuspendProcess) +/// while a trivial script is submitted. +/// +[Collection(ScriptWorkerCollection.Name)] +[Trait("Category", "ScriptWorker")] +public sealed class WorkerSoftClockTests +{ + private const string Trivial = "return 7;"; + private const string Runaway = "while (true) { }"; + + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + /// + /// A runner whose worker heartbeats every 2 s, so the host's silence kill (3 intervals) stays well clear of the + /// suspensions below: what is measured is the deadline, not the heartbeat monitor. + /// + private static OutOfProcessScriptRunner Runner(TimeSpan hardGrace, ILogger? log = null) + => new(Scripts.Options(hardGrace: hardGrace, quarantineAfterKills: 1), log ?? new LoggerConfiguration().CreateLogger(), null, + heartbeatInterval: TimeSpan.FromSeconds(2), loadTimeout: TimeSpan.FromSeconds(60)); + + private static async Task WarmAsync(OutOfProcessScriptRunner runner, string key) + { + var warm = await runner.RunAsync(Scripts.Vt(key, Trivial, TimeSpan.FromSeconds(30)), Ct); + warm.Status.ShouldBe(ScriptRunStatus.Ok, warm.Reason); + var worker = runner.Shards[0].Current.ShouldNotBeNull(); + worker.WorkerSoftClock.ShouldBeTrue("the worker negotiated the soft clock"); + return worker; + } + + /// + /// The issue's headline: a worker that cannot be scheduled for longer than T, but shorter than the transport + /// deadline, answers a trivial script with its value. Before #827 the host's clock, started at submit, answered + /// script timed out after 0.3s for return 7;. + /// + [Fact] + public async Task A_worker_stalled_for_longer_than_T_still_answers_a_trivial_script_with_its_value() + { + var softT = TimeSpan.FromMilliseconds(300); + await using var runner = Runner(hardGrace: TimeSpan.FromSeconds(3)); + var worker = await WarmAsync(runner, "trivial"); + + Task run; + using (ProcessFreezer.Freeze(worker.Pid)) + { + run = runner.RunAsync(Scripts.Vt("trivial", Trivial, softT), Ct); + await Task.Delay(softT * 4, Ct); // 1.2 s: well past T on the host's clock + run.IsCompleted.ShouldBeFalse("nothing can answer while the worker is suspended"); + } + + var outcome = await run.WaitAsync(TimeSpan.FromSeconds(30), Ct); + outcome.Status.ShouldBe(ScriptRunStatus.Ok, outcome.Reason); + outcome.Value.ShouldBe(7); + runner.Shards[0].Current.ShouldNotBeNull().Pid.ShouldBe(worker.Pid, "a stall is not a deadline kill"); + runner.QuarantinedKeys(ScriptContextKind.VirtualTag).ShouldBeEmpty(); + } + + /// + /// A stall past the transport deadline (T + G) is the runner's fault: the caller gets + /// as (published + /// BadResourceUnavailable) with a recovery that completes when the worker answers; a re-trigger meanwhile is the + /// runner's fault too (not "previous run still running"); the worker is not killed and the key is not struck. + /// Before #827 this was a script timeout at T and a deadline kill at T + G that struck the key — with + /// QuarantineAfterKills = 1, a quarantine for return 7;. + /// + [Fact] + public async Task A_worker_stalled_past_the_transport_deadline_is_a_runner_fault_and_nothing_is_struck() + { + var softT = TimeSpan.FromMilliseconds(200); + var grace = TimeSpan.FromMilliseconds(400); + await using var runner = Runner(grace); + var worker = await WarmAsync(runner, "trivial"); + + ScriptRunOutcome stalled, retriggered; + using (ProcessFreezer.Freeze(worker.Pid)) + { + stalled = await runner.RunAsync(Scripts.Vt("trivial", Trivial, softT), Ct).WaitAsync(TimeSpan.FromSeconds(10), Ct); + retriggered = await runner.RunAsync(Scripts.Vt("trivial", Trivial, softT), Ct).WaitAsync(TimeSpan.FromSeconds(10), Ct); + stalled.Recovery.ShouldNotBeNull().IsCompleted.ShouldBeFalse("the worker has not answered yet"); + } + + stalled.Status.ShouldBe(ScriptRunStatus.Unavailable, stalled.Reason); + stalled.Reason.ShouldBe(ScriptRunOutcome.WorkerUnresponsiveReason); + retriggered.Status.ShouldBe(ScriptRunStatus.Unavailable, retriggered.Reason); + retriggered.Reason.ShouldBe(ScriptRunOutcome.WorkerUnresponsiveReason); + stalled.ToVirtualTagEvalResult().StatusCode.ShouldBe(Commons.OpcUa.OpcUaStatus.BadResourceUnavailable); + + await stalled.Recovery!.WaitAsync(TimeSpan.FromSeconds(30), Ct); + var next = await runner.RunAsync(Scripts.Vt("trivial", Trivial, softT), Ct); + next.Status.ShouldBe(ScriptRunStatus.Ok, next.Reason); + next.Value.ShouldBe(7); + runner.Shards[0].Current.ShouldNotBeNull().Pid.ShouldBe(worker.Pid, "a stall is not a deadline kill"); + runner.QuarantinedKeys(ScriptContextKind.VirtualTag).ShouldBeEmpty(); + } + + /// + /// The other side of the contract is unchanged: a script that really runs past T is answered at T from its own + /// start and killed G after that, even with the worker stalled before it started. Positive control for the two + /// tests above — they would also pass if the runner simply never timed anything out. + /// + [Fact] + public async Task A_runaway_that_starts_late_is_still_timed_out_at_T_from_its_start_and_killed_at_G() + { + var softT = TimeSpan.FromMilliseconds(300); + var grace = TimeSpan.FromMilliseconds(600); + var stall = TimeSpan.FromMilliseconds(500); + await using var runner = Runner(grace); + var worker = await WarmAsync(runner, "warm"); + var load = await runner.RunAsync(Scripts.Vt("runaway", "return 0;", TimeSpan.FromSeconds(30)), Ct); + load.Status.ShouldBe(ScriptRunStatus.Ok, load.Reason); + + var sw = Stopwatch.StartNew(); + Task run; + using (ProcessFreezer.Freeze(worker.Pid)) + { + run = runner.RunAsync(Scripts.Vt("runaway", Runaway, softT), Ct); + await Task.Delay(stall, Ct); + } + var thawed = sw.Elapsed; + var outcome = await run.WaitAsync(TimeSpan.FromSeconds(30), Ct); + var answered = sw.Elapsed; + + outcome.Status.ShouldBe(ScriptRunStatus.Failed); + outcome.Reason.ShouldNotBeNull().ShouldStartWith("script timed out after "); + (answered - thawed).ShouldBeGreaterThanOrEqualTo(softT - TimeSpan.FromMilliseconds(50), "T counts from the run's start, after the stall"); + + var exit = await worker.Exited.WaitAsync(TimeSpan.FromSeconds(30), Ct); + exit.Reason.ShouldBe(WorkerExitReason.KilledDeadline); + exit.CulpritKey.ShouldBe(InProcessScriptRunner.ScopedKey(ScriptContextKind.VirtualTag, "runaway")); + (sw.Elapsed - thawed).ShouldBeGreaterThanOrEqualTo(softT + grace - TimeSpan.FromMilliseconds(50), "G counts from the soft deadline"); + } + + /// + /// Shadow (lmxopcua#780 P2) compares the shadow pool with the published in-process result; a starved worker used to + /// show up there as a script mismatch (published Ok, shadow "timed out") and skew the census for the OutOfProcess + /// flip. Now a stall shorter than the transport deadline is a match, and a longer one is shadow_unavailable + /// — never a mismatch — including the re-trigger that finds the stalled run still holding its slot. + /// + [Fact] + public async Task In_Shadow_a_stalled_worker_is_a_match_or_shadow_unavailable_never_a_mismatch() + { + var softT = TimeSpan.FromMilliseconds(300); + var grace = TimeSpan.FromSeconds(1); + var options = Scripts.Options(hardGrace: grace, quarantineAfterKills: 1); + options.Runner = ScriptRunnerMode.Shadow; + await using var pool = Runner(grace); + using var shadow = new ShadowScriptRunner(pool, options, new LoggerConfiguration().CreateLogger()); + var worker = await WarmAsync(pool, "trivial"); + var published = ScriptRunOutcome.Ok(7); + + // Stall past T, short of T + G: compared, and a match. + using (ProcessFreezer.Freeze(worker.Pid)) + { + shadow.Mirror(Scripts.Vt("trivial", Trivial, softT), published); + await Task.Delay(TimeSpan.FromMilliseconds(900), Ct); + } + (await shadow.WaitIdleAsync(TimeSpan.FromSeconds(30))).ShouldBeTrue(); + shadow.Stats().Matched.ShouldBe(1, Summary(shadow)); + + // Stall past T + G: not compared. The second mirror arrives while the first still holds the key's slot. + using (ProcessFreezer.Freeze(worker.Pid)) + { + shadow.Mirror(Scripts.Vt("trivial", Trivial, softT), published); + await Scripts.Eventually(() => shadow.Stats().ShadowUnavailable == 1, TimeSpan.FromSeconds(10), "the transport deadline"); + shadow.Mirror(Scripts.Vt("trivial", Trivial, softT), published); + await Scripts.Eventually(() => shadow.Stats().ShadowUnavailable == 2, TimeSpan.FromSeconds(10), "the stalled re-trigger"); + } + (await shadow.WaitIdleAsync(TimeSpan.FromSeconds(30))).ShouldBeTrue(); + + var stats = shadow.Stats(); + stats.Mismatched.ShouldBe(0, Summary(shadow)); + stats.ShadowUnavailable.ShouldBe(2, Summary(shadow)); + stats.Matched.ShouldBe(1); + pool.Shards[0].Current.ShouldNotBeNull().Pid.ShouldBe(worker.Pid); + } + + /// + /// Protocol compatibility: the soft clock is negotiated (Hello.worker_soft_clock → Ready.worker_soft_clock). + /// A host that does not ask for it keeps the pre-#827 contract — the soft deadline is the host's, from submit, + /// and the worker sends no soft notices — and a runaway is still answered at T and killed at T + G. + /// + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task The_soft_clock_is_negotiated_and_both_clocks_enforce_T_and_G(bool workerSoftClock) + { + var spec = new WorkerLaunchSpec("t", OutOfProcessScriptRunner.ResolveWorkerPath(new ScriptWorkerOptions { ExecutablePath = Scripts.WorkerUnderTest }), + 256L * 1024 * 1024, EvalThreads: 2, HardGrace: TimeSpan.FromSeconds(1), HeartbeatInterval: TimeSpan.FromSeconds(1), + HandshakeTimeout: TimeSpan.FromSeconds(15), WorkerSoftClock: workerSoftClock); + var connection = await WorkerConnection.StartAsync(spec, new LoggerConfiguration().CreateLogger(), _ => { }, _ => { }, Ct); + await using var _ = connection; + connection.WorkerSoftClock.ShouldBe(workerSoftClock); + + var entry = await connection.EnsureLoadedAsync("k", ContextKind.VirtualTag, Runaway, + SHA256.HashData(Encoding.UTF8.GetBytes(Runaway)), []).WaitAsync(TimeSpan.FromSeconds(60), Ct); + entry.Outcome.ShouldNotBeNull().Ok.ShouldBeTrue(); + + var softT = TimeSpan.FromMilliseconds(300); + var grace = TimeSpan.FromMilliseconds(500); + var sw = Stopwatch.StartNew(); + var pending = connection.Submit("k", + new EvalRequest { Key = "k", SoftTimeoutMs = (uint)softT.TotalMilliseconds, HardGraceMs = (uint)grace.TotalMilliseconds, NowUtcTicks = DateTime.UtcNow.Ticks }, + softT, grace, null); + (await pending.Outcome.WaitAsync(TimeSpan.FromSeconds(10), Ct)).ShouldBeSameAs(PendingEval.SoftTimedOut); + sw.Elapsed.ShouldBeGreaterThanOrEqualTo(softT - TimeSpan.FromMilliseconds(20)); + connection.Cancel(pending); + + var exit = await connection.Exited.WaitAsync(TimeSpan.FromSeconds(10), Ct); + exit.Reason.ShouldBe(WorkerExitReason.KilledDeadline); + sw.Elapsed.ShouldBeGreaterThanOrEqualTo(softT + grace - TimeSpan.FromMilliseconds(50)); + } + + // ---------------------------------------------------------------- PendingEval, no process + + /// The host side of the worker clock, deterministic: no notice means the transport deadline at T + G. + [Fact] + public async Task Worker_clock_with_no_word_from_the_worker_answers_transport_timed_out_at_T_plus_G_and_never_kills() + { + var killed = 0; + var pending = new PendingEval("k", 1); + var sw = Stopwatch.StartNew(); + pending.Arm(TimeSpan.FromMilliseconds(100), TimeSpan.FromMilliseconds(200), _ => Interlocked.Increment(ref killed), workerClock: true); + + (await pending.Outcome.WaitAsync(TimeSpan.FromSeconds(10), Ct)).ShouldBeSameAs(PendingEval.TransportTimedOut); + sw.Elapsed.ShouldBeGreaterThanOrEqualTo(TimeSpan.FromMilliseconds(280)); + await Task.Delay(500, Ct); + killed.ShouldBe(0, "a late worker is not the script's fault"); + + // A soft notice after the transport answer still arms the kill: the script itself ran for T. + pending.SoftExpired(); + await Scripts.Eventually(() => Volatile.Read(ref killed) == 1, TimeSpan.FromSeconds(10), "the kill G after the soft notice"); + } + + /// The soft notice answers the caller and arms the kill G later; a result before then cancels it. + [Fact] + public async Task Worker_clock_soft_notice_answers_soft_timed_out_and_arms_the_kill_G_later() + { + var killed = 0; + var pending = new PendingEval("k", 1); + pending.Arm(TimeSpan.FromSeconds(30), TimeSpan.FromMilliseconds(300), _ => Interlocked.Increment(ref killed), workerClock: true); + pending.Outcome.IsCompleted.ShouldBeFalse(); + + var sw = Stopwatch.StartNew(); + pending.SoftExpired(); + (await pending.Outcome.WaitAsync(TimeSpan.FromSeconds(1), Ct)).ShouldBeSameAs(PendingEval.SoftTimedOut); + await Scripts.Eventually(() => Volatile.Read(ref killed) == 1, TimeSpan.FromSeconds(10), "the kill"); + sw.Elapsed.ShouldBeGreaterThanOrEqualTo(TimeSpan.FromMilliseconds(280)); + + var answered = new PendingEval("k2", 2); + answered.Arm(TimeSpan.FromSeconds(30), TimeSpan.FromMilliseconds(100), _ => Interlocked.Increment(ref killed), workerClock: true); + answered.SoftExpired(); + answered.Complete(new EvalResult { EvalId = 2 }); + await Task.Delay(400, Ct); + killed.ShouldBe(1, "a run that answered inside G is not killed"); + } + + /// Under the host clock a soft notice means nothing (a worker that did not negotiate it cannot send one). + [Fact] + public void Host_clock_ignores_a_soft_notice() + { + var pending = new PendingEval("k", 1); + pending.Arm(TimeSpan.FromSeconds(30), TimeSpan.FromSeconds(30), _ => { }, workerClock: false); + pending.SoftExpired(); + pending.Outcome.IsCompleted.ShouldBeFalse(); + pending.Complete(null); + } + + private static string Summary(ShadowScriptRunner shadow) + => $"{shadow.Stats()}; " + string.Join("; ", shadow.Mismatches.Select(m => $"{m.Key}: {m.LastReason} — {m.LastSummary}")); +} + +/// Suspends a process (SIGSTOP, or NtSuspendProcess on Windows) until disposed. +internal sealed class ProcessFreezer : IDisposable +{ + private readonly int _pid; + private int _thawed; + + private ProcessFreezer(int pid) + { + _pid = pid; + Signal(pid, stop: true); + } + + public static ProcessFreezer Freeze(int pid) => new(pid); + + public void Dispose() + { + if (Interlocked.Exchange(ref _thawed, 1) == 0) Signal(_pid, stop: false); + } + + private static void Signal(int pid, bool stop) + { + if (OperatingSystem.IsWindows()) + { + using var process = Process.GetProcessById(pid); + var status = stop ? NtSuspendProcess(process.Handle) : NtResumeProcess(process.Handle); + status.ShouldBe(0, $"Nt{(stop ? "Suspend" : "Resume")}Process"); + return; + } + // libc kill(2), not the kill(1) command: a child process's exit is observed through the same SIGCHLD + // handling that the suspended worker (another child of this process) just disturbed, and WaitForExit was + // seen to take seconds. + var signal = stop ? SigStop : SigCont; + Kill(pid, signal).ShouldBe(0, $"kill({pid}, {signal}): errno {Marshal.GetLastPInvokeError()}"); + } + + /// SIGSTOP / SIGCONT: 17 / 19 on macOS (BSD), 19 / 18 on Linux. + private static int SigStop => OperatingSystem.IsMacOS() ? 17 : 19; + + private static int SigCont => OperatingSystem.IsMacOS() ? 19 : 18; + +#pragma warning disable SYSLIB1054 // DllImport kept: LibraryImport would need AllowUnsafeBlocks for three test-only calls. + [DllImport("libc", EntryPoint = "kill", SetLastError = true)] + private static extern int Kill(int pid, int signal); + + [DllImport("ntdll.dll")] + private static extern int NtSuspendProcess(IntPtr processHandle); + + [DllImport("ntdll.dll")] + private static extern int NtResumeProcess(IntPtr processHandle); +#pragma warning restore SYSLIB1054 +} From 30c6c3a0bfa8e9daba67167401aae2e3b04c4b54 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 01:59:11 -0500 Subject: [PATCH 3/7] test(scriptworker): tight-heartbeat tests warm the worker before arming the 3-interval silence kill (#827 audit) A_burst_of_cold_compiles_does_not_starve_the_heartbeat (50 ms heartbeat) and A_caller_that_blocks_after_its_result_does_not_starve_the_reader (100 ms) launched their worker under the tight silence kill, so its cold start (launch, first compile, Roslyn JIT) had to stay within 150/300 ms of silence. Under 28 busy processes on 14 cores that failed 2 in 10 runs in the WARM-UP, killed for ~165 ms of silence, and replacement workers were killed the same way. New test seam: WorkerLaunchSpec.HeartbeatSilenceLimit (runner internal ctor heartbeatSilenceLimit) and WorkerConnection.SilenceLimit. The tests launch with a 30 s limit, warm, then tighten to three intervals for the phase they are about; the assertions on that phase are unchanged. --- .../OutOfProcessScriptRunner.cs | 9 +++-- .../WorkerConnection.cs | 29 ++++++++++++--- .../SupervisionTests.cs | 35 ++++++++++++++----- 3 files changed, 59 insertions(+), 14 deletions(-) diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs index 06b748b4d..1d7556180 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs @@ -141,6 +141,9 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, private readonly string? _peCacheDirectory; private readonly byte[]? _peCacheKey; private readonly int _compileThreads; + + /// Test seam: the launched workers' initial silence limit (null = three heartbeat intervals). + private readonly TimeSpan? _heartbeatSilenceLimit; private long _peCacheHits; private long _peCacheMisses; private long _peCacheRejected; @@ -159,9 +162,10 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, internal OutOfProcessScriptRunner(ScriptingOptions options, ILogger? log, ILogger? scriptRoot, TimeSpan heartbeatInterval, TimeSpan loadTimeout, TimeSpan? maintenanceInterval = null, string? markerDirectory = null, - int compileThreads = 0, int evalThreads = 0) + int compileThreads = 0, int evalThreads = 0, TimeSpan? heartbeatSilenceLimit = null) { _compileThreads = compileThreads; + _heartbeatSilenceLimit = heartbeatSilenceLimit; ArgumentNullException.ThrowIfNull(options); _options = options.Sanitized(out _); _worker = _options.Worker; @@ -719,7 +723,8 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, MarkerDirectory: _markerDirectory, LoadTimeout: LoadTimeout, CompileThreads: _compileThreads, - WorkerSid: WorkerSid); + WorkerSid: WorkerSid, + HeartbeatSilenceLimit: _heartbeatSilenceLimit); return new WorkerShard(label, isolation, (onExiting, ct) => WorkerConnection.StartAsync(spec, _log, OnLogs, onExiting, ct), OnWorkerExit, _log); diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs index a9fce4fa4..228068a9d 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs @@ -81,6 +81,10 @@ internal sealed record BatchSubmission(string Key, EvalRequest Request, TimeSpan /// Windows only: the worker identity's SID (Scripting:Worker:Identity). Set, the worker starts under a restricted /// token () and the pipe ACL admits it beside the host; null keeps the host's token. /// +/// +/// How long the worker may send nothing before it is killed (); null = +/// three s. Test seam (with ). +/// /// /// Ask the worker to run the soft clock from each run's start (lmxopcua#827; see ). /// False (tests only) keeps the pre-#827 host clock, which is also what a worker that does not echo it gets. @@ -99,7 +103,8 @@ internal sealed record WorkerLaunchSpec( TimeSpan? LoadTimeout = null, int CompileThreads = 0, string? WorkerSid = null, - bool WorkerSoftClock = true) + bool WorkerSoftClock = true, + TimeSpan? HeartbeatSilenceLimit = null) { /// The default compile timeout. public static readonly TimeSpan DefaultLoadTimeout = TimeSpan.FromSeconds(60); @@ -156,7 +161,7 @@ internal sealed class WorkerConnection : IAsyncDisposable private WorkerConnection(string shard, Process process, NamedPipeServerStream pipe, IDisposable? job, IDisposable? stdio, string workDir, ILogger log, Action onLogs, TimeSpan heartbeatInterval, TimeSpan loadTimeout, - Action onExiting, string? markerPath, bool workerSoftClock) + Action onExiting, string? markerPath, bool workerSoftClock, TimeSpan? heartbeatSilenceLimit) { WorkerSoftClock = workerSoftClock; _stdio = stdio; @@ -171,6 +176,7 @@ internal sealed class WorkerConnection : IAsyncDisposable _log = log; _onLogs = onLogs; _heartbeatInterval = heartbeatInterval; + _silenceLimitTicks = (heartbeatSilenceLimit ?? heartbeatInterval * 3).Ticks; Pid = process.Id; // The handshake used seq 1. A failed write means the pipe is gone: the worker is dead or dying. _writer = new ScriptWorkerFrameWriter(pipe, OnWriteFault, ScriptWorkerFrameCodec.DefaultMaxFrameBytes, firstSeq: 1, @@ -215,6 +221,20 @@ internal sealed class WorkerConnection : IAsyncDisposable set => _loadTimeout = value; } + /// + /// How long the worker may send nothing before it is killed for silence: three heartbeat intervals unless the + /// launch spec said otherwise. Test hook: a test with a tight heartbeat launches and warms the worker (its cold + /// start JITs Roslyn, which on a loaded box can pause it for longer than three short intervals) under a generous + /// limit, then tightens it for the phase it is about. + /// + internal TimeSpan SilenceLimit + { + get => TimeSpan.FromTicks(Volatile.Read(ref _silenceLimitTicks)); + set => Volatile.Write(ref _silenceLimitTicks, value.Ticks); + } + + private long _silenceLimitTicks; + /// Compiles currently outstanding on this worker. public int LoadingCount => _loads.Count; @@ -327,7 +347,8 @@ internal sealed class WorkerConnection : IAsyncDisposable : await HandshakeAsync(pipe, hello, nonce, process.Id, handshakeCts.Token).ConfigureAwait(false); var connection = new WorkerConnection(spec.Shard, process, pipe, job, stdio, workDir, log, onLogs, spec.HeartbeatInterval, - spec.EffectiveLoadTimeout, onExiting, markerPath, workerSoftClock: spec.WorkerSoftClock && ready.WorkerSoftClock); + spec.EffectiveLoadTimeout, onExiting, markerPath, workerSoftClock: spec.WorkerSoftClock && ready.WorkerSoftClock, + heartbeatSilenceLimit: spec.HeartbeatSilenceLimit); connection.Begin(); return connection; } @@ -888,7 +909,7 @@ internal sealed class WorkerConnection : IAsyncDisposable { if (_exited.Task.IsCompleted) return; var silent = Stopwatch.GetElapsedTime(Volatile.Read(ref _lastFrameTimestamp)); - if (silent > _heartbeatInterval * 3) + if (silent > SilenceLimit) { _log.Warning("ScriptWorker[{Shard}] pid {Pid} sent nothing for {Silent} ms; killing it", Shard, Pid, (long)silent.TotalMilliseconds); Kill(WorkerExitReason.KilledHeartbeat, null); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/SupervisionTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/SupervisionTests.cs index 5f075fcfa..df214c4b6 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/SupervisionTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/SupervisionTests.cs @@ -31,6 +31,27 @@ public sealed class SupervisionTests private static CancellationToken Ct => TestContext.Current.CancellationToken; + /// + /// A runner whose workers heartbeat every but start with a generous silence limit: + /// the cold start (launch, first compile, Roslyn's JIT) is not what the tight-heartbeat tests are about, and on a + /// loaded box it can pause the worker for longer than three short intervals (lmxopcua#827 audit: 2 in 10 runs + /// under 28 busy processes on 14 cores were killed for 160 ms of silence during the warm-up compile, and the + /// replacement workers were too). then tightens the limit to three intervals for the + /// phase under test. + /// + private static OutOfProcessScriptRunner TightHeartbeatRunner(TimeSpan heartbeat, TimeSpan loadTimeout, CapturingSink sink) + => new(Scripts.Options(), sink.CreateLogger(), null, heartbeatInterval: heartbeat, loadTimeout: loadTimeout, + heartbeatSilenceLimit: TimeSpan.FromSeconds(30)); + + /// Warms the shard's worker with , then arms the three-interval silence kill. + private static WorkerConnection WarmTight(OutOfProcessScriptRunner runner, ScriptRunOutcome warm, TimeSpan heartbeat, CapturingSink sink) + { + warm.Value.ShouldBe(7, $"{warm.Status} {warm.Reason}\n{string.Join(Environment.NewLine, sink.Rendered)}"); + var worker = runner.Shards[0].Current.ShouldNotBeNull(); + worker.SilenceLimit = heartbeat * 3; + return worker; + } + private static string VtKey(string key) => InProcessScriptRunner.ScopedKey(ScriptContextKind.VirtualTag, key); // ---------------------------------------------------------------- the quarantine race @@ -112,10 +133,9 @@ public sealed class SupervisionTests public async Task A_burst_of_cold_compiles_does_not_starve_the_heartbeat() { var sink = new CapturingSink(); - await using var runner = new OutOfProcessScriptRunner(Scripts.Options(), sink.CreateLogger(), null, - heartbeatInterval: TimeSpan.FromMilliseconds(50), loadTimeout: TimeSpan.FromSeconds(120)); - (await runner.RunAsync(Scripts.Vt("first", Healthy), Ct)).Value.ShouldBe(7); - var worker = runner.Shards[0].Current.ShouldNotBeNull(); + var heartbeat = TimeSpan.FromMilliseconds(50); + await using var runner = TightHeartbeatRunner(heartbeat, TimeSpan.FromSeconds(120), sink); + var worker = WarmTight(runner, await runner.RunAsync(Scripts.Vt("first", Healthy), Ct), heartbeat, sink); const int count = 1500; var outcomes = await Task.WhenAll(Enumerable.Range(0, count).Select(i => runner.RunAsync( @@ -514,10 +534,9 @@ public sealed class SupervisionTests public async Task A_caller_that_blocks_after_its_result_does_not_starve_the_reader() { var sink = new CapturingSink(); - await using var runner = new OutOfProcessScriptRunner(Scripts.Options(), sink.CreateLogger(), null, - heartbeatInterval: TimeSpan.FromMilliseconds(100), loadTimeout: TimeSpan.FromSeconds(60)); - (await runner.RunAsync(Scripts.Vt("warm", Healthy), Ct)).Value.ShouldBe(7); - var worker = runner.Shards[0].Current.ShouldNotBeNull(); + var heartbeat = TimeSpan.FromMilliseconds(100); + await using var runner = TightHeartbeatRunner(heartbeat, TimeSpan.FromSeconds(60), sink); + var worker = WarmTight(runner, await runner.RunAsync(Scripts.Vt("warm", Healthy), Ct), heartbeat, sink); string? resumedOn = null; await runner.RunAsync(Scripts.Vt("warm", Healthy), Ct).ContinueWith(t => From c5dffbd8b8c126436f02b8e1561aa192f62cf8c2 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 02:21:45 -0500 Subject: [PATCH 4/7] fix(scripting): soft notice resumes the caller on the dispatcher; a late notice still cancels and counts (#827 review F1, F2) F1: PendingEval.SoftExpired runs on the pipe reader and completed the caller's outcome inline (production outcomes have no async continuations, by design, because results resume on the ResultDispatcher). The caller's code (the runner's Cancel write, the seam's publish) therefore ran on the reader, which could hold up heartbeats (a false silence kill) and, with blocking I/O, write to the pipe from the reader. The soft outcome is now posted to the dispatcher like a result. Test: a 1.5 s continuation on a soft timeout with a 300 ms silence limit resumes on ScriptWorker[t].Results and the worker lives (before: it resumed on the Reader). F2: after a transport timeout ("worker unresponsive"), a late soft notice armed G but nobody sent Cancel, so a polling script could not end itself and the whole worker was killed at notice + G with a kill strike. PendingEval now exposes LateSoftExpiry (asynchronous continuations, cancelled when the run answers); the runner sends the Cancel and counts the timeout (NoteTimeout) from it. A genuinely slow script whose notice is delayed past T + G is thus still a timeout. Test: stall past T + G, thaw, a polling script overruns T: the worker survives G and the key is quarantined for timeouts, not a hard deadline (before: the worker was killed). --- .../OutOfProcessScriptRunner.cs | 20 ++++- .../WorkerConnection.cs | 30 ++++++- .../WorkerSoftClockTests.cs | 79 +++++++++++++++++++ 3 files changed, 122 insertions(+), 7 deletions(-) diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs index 1d7556180..48aa3f588 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs @@ -591,10 +591,22 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, { // lmxopcua#827: T + G passed with neither a result nor a soft-deadline notice. The worker runs the soft // clock from the run's start, so this is the pipe or a starved / stalled worker — the runner's fault, - // never the script's: no timeout strike, no Cancel (the run may not even have started), and the slot - // stays with the run until the worker answers. The seam re-evaluates when it does (or when the worker - // is replaced, if the heartbeat monitor kills it). - TransportTimeoutCounter.Add(1, new KeyValuePair("kind", KindTag(invocation.Kind))); + // not (yet) the script's: no timeout counted, no Cancel (the run may not even have started), and the + // slot stays with the run until the worker answers. The seam re-evaluates when it does (or when the + // worker is replaced, if the heartbeat monitor kills it). + TransportTimeoutCounter.Add(1, new KeyValuePair("kind", KindTag(invocation.Kind)), + new KeyValuePair("answer", "stalled")); + // …unless the worker then reports the script itself over T (slow AND late — a notice delayed past + // T + G by the watchdog poll and the pipe lands here too): it still gets its Cancel, so a polling + // script ends itself instead of costing the whole worker at notice + G, and it counts as a timeout + // (#827 review F2). Off the reader: the completion source runs this asynchronously. + var timeout = invocation.Timeout; + pending.LateSoftExpiry?.ContinueWith(_ => + { + connection.Cancel(pending); + NoteTimeout(key, state, timeout); + }, CancellationToken.None, TaskContinuationOptions.OnlyOnRanToCompletion | TaskContinuationOptions.ExecuteSynchronously, + TaskScheduler.Default); // Read by a refusal only while it is incomplete, so a write that loses a race with the slot release // is harmless. Volatile.Write(ref state.StalledRun, pending.Completion); diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs index 228068a9d..911f20ee8 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs @@ -1128,6 +1128,7 @@ internal sealed class PendingEval private bool _started; private int _phase; private int _done; + private TaskCompletionSource? _lateSoftExpiry; public PendingEval(string key, ulong evalId, Action? onFinished = null, ResultDispatcher? dispatcher = null) { @@ -1160,6 +1161,16 @@ internal sealed class PendingEval /// The worker's result, or null when the worker ended first. public Task Completion => _completion.Task; + /// + /// After (null before): completes when the worker then reports the script itself + /// over T (slow AND late: it still gets its Cancel and counts as a timeout, #827 review F2), and is cancelled + /// when the worker answers or ends first. Continuations run asynchronously, never on the reader. + /// + internal Task? LateSoftExpiry + { + get { lock (_clock) return _lateSoftExpiry?.Task; } + } + /// /// What the caller waits on: the worker's result, null when the worker ended first, at /// the soft deadline, or at the transport deadline (the run itself is still @@ -1246,7 +1257,7 @@ internal sealed class PendingEval /// internal void SoftExpired() { - bool signal; + bool signal, late; lock (_clock) { if (_timer is null || !_workerClock || Volatile.Read(ref _done) != 0) return; @@ -1255,12 +1266,23 @@ internal sealed class PendingEval _started = true; _deferred = false; signal = phase == Waiting; + late = phase == TransportReported; Interlocked.Exchange(ref _phase, SoftReported); _hardDue = Stopwatch.GetTimestamp() + Ticks(_grace); TryChange(_grace); } - // Outside the lock: a caller with a dispatcher resumes synchronously inside TrySetResult. - if (signal) _outcome.TrySetResult(SoftTimedOut); + // This runs on the pipe READER. The caller's resumption is the caller's code (the seam's publish, the runner's + // Cancel write): it goes through the dispatcher exactly like a result (ResultDispatcher remarks), never inline + // here, or a slow consumer would hold up heartbeats and results (#827 review F1). + if (signal) + { + if (_dispatcher is { } dispatcher) dispatcher.Post(this, SoftTimedOut); + else _outcome.TrySetResult(SoftTimedOut); + } + // The caller was already answered "worker unresponsive"; the script has now itself run for T. Its watcher + // (the runner) cancels the run's token and counts the timeout, off this thread: the completion source runs its + // continuations asynchronously (#827 review F2). + if (late) _lateSoftExpiry?.TrySetResult(); } private void OnTimer() @@ -1277,6 +1299,7 @@ internal sealed class PendingEval // A tick armed before a deferral/start can land early: re-arm for what is left. if (now < _transportDue) { TryChange(Stopwatch.GetElapsedTime(now, _transportDue)); return; } Interlocked.Exchange(ref _phase, TransportReported); + _lateSoftExpiry = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); transport = true; } else if (phase == Waiting) @@ -1316,6 +1339,7 @@ internal sealed class PendingEval { if (Interlocked.Exchange(ref _done, 1) != 0) return; _timer?.Dispose(); + lock (_clock) _lateSoftExpiry?.TrySetCanceled(); try { _onFinished?.Invoke(); } finally { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs index 359e3f86f..7a35adfd4 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs @@ -227,6 +227,85 @@ public sealed class WorkerSoftClockTests sw.Elapsed.ShouldBeGreaterThanOrEqualTo(softT + grace - TimeSpan.FromMilliseconds(50)); } + // ---------------------------------------------------------------- review findings (#827) + + /// + /// Review F1: the soft notice is handled on the pipe READER, and the caller's resumption is the caller's code. It + /// must go through the result dispatcher like a result does: a continuation that blocks for 1.5 s must not hold up + /// the reader, or the next heartbeats sit unread and a healthy worker is killed for silence (300 ms here). Before + /// the fix the continuation ran inline on the reader and the worker was killed (KilledHeartbeat). + /// + [Fact] + public async Task A_slow_continuation_of_a_soft_timeout_does_not_stall_the_reader() + { + var heartbeat = TimeSpan.FromMilliseconds(100); + var spec = new WorkerLaunchSpec("t", OutOfProcessScriptRunner.ResolveWorkerPath(new ScriptWorkerOptions { ExecutablePath = Scripts.WorkerUnderTest }), + 256L * 1024 * 1024, EvalThreads: 2, HardGrace: TimeSpan.FromSeconds(60), HeartbeatInterval: heartbeat, + HandshakeTimeout: TimeSpan.FromSeconds(15), HeartbeatSilenceLimit: TimeSpan.FromSeconds(30)); + var connection = await WorkerConnection.StartAsync(spec, new LoggerConfiguration().CreateLogger(), _ => { }, _ => { }, Ct); + await using var _ = connection; + var entry = await connection.EnsureLoadedAsync("k", ContextKind.VirtualTag, Runaway, + SHA256.HashData(Encoding.UTF8.GetBytes(Runaway)), []).WaitAsync(TimeSpan.FromSeconds(60), Ct); + entry.Outcome.ShouldNotBeNull().Ok.ShouldBeTrue(); + connection.SilenceLimit = heartbeat * 3; // armed only once the cold start is over (see SupervisionTests) + + var softT = TimeSpan.FromMilliseconds(200); + var pending = connection.Submit("k", + new EvalRequest { Key = "k", SoftTimeoutMs = (uint)softT.TotalMilliseconds, HardGraceMs = 60_000, NowUtcTicks = DateTime.UtcNow.Ticks }, + softT, TimeSpan.FromSeconds(60), null); + string? resumedOn = null; + var slowConsumer = pending.Outcome.ContinueWith(t => + { + resumedOn = Thread.CurrentThread.Name; + t.Result.ShouldBeSameAs(PendingEval.SoftTimedOut); + Thread.Sleep(1_500); + }, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); + + await slowConsumer.WaitAsync(TimeSpan.FromSeconds(10), Ct); + resumedOn.ShouldBe("ScriptWorker[t].Results", "the caller resumes on the dispatcher, never on the reader"); + connection.Exited.IsCompleted.ShouldBeFalse("the reader kept reading heartbeats while the consumer was slow"); + connection.IsAlive.ShouldBeTrue(); + } + + /// + /// Review F2: a stall past T + G answers "worker unresponsive"; when the worker then runs the script and it really + /// overruns T, the late soft notice must still cancel its token — a polling script ends itself instead of the whole + /// worker being killed G later — and the run counts as a timeout (QuarantineAfterConsecutiveTimeouts = 1 here). + /// Before the fix nobody sent the Cancel: the worker was killed at notice + G (KilledDeadline, a kill strike). + /// + [Fact] + public async Task A_slow_script_whose_notice_arrives_after_the_transport_deadline_is_cancelled_and_counted_not_killed() + { + var softT = TimeSpan.FromMilliseconds(200); + var grace = TimeSpan.FromMilliseconds(400); + var options = Scripts.Options(hardGrace: grace, quarantineAfterKills: 1); + options.QuarantineAfterConsecutiveTimeouts = 1; + await using var runner = new OutOfProcessScriptRunner(options, new LoggerConfiguration().CreateLogger(), null, + heartbeatInterval: TimeSpan.FromSeconds(2), loadTimeout: TimeSpan.FromSeconds(60)); + const string polite = "if ((int)ctx.GetTag(\"spin\").Value == 1) { while (!ctx.CancellationToken.IsCancellationRequested) { } } return 1;"; + (await runner.RunAsync(Scripts.Vt("polite", polite, TimeSpan.FromSeconds(30), Scripts.Input("spin", 0)), Ct)).Value.ShouldBe(1); + var worker = runner.Shards[0].Current.ShouldNotBeNull(); + + ScriptRunOutcome stalled; + using (ProcessFreezer.Freeze(worker.Pid)) + { + stalled = await runner.RunAsync(Scripts.Vt("polite", polite, softT, Scripts.Input("spin", 1)), Ct).WaitAsync(TimeSpan.FromSeconds(10), Ct); + } + stalled.Reason.ShouldBe(ScriptRunOutcome.WorkerUnresponsiveReason); + + // Thawed: the script starts, spins past T, the late notice cancels it and it returns. The worker survives G. + await stalled.Recovery.ShouldNotBeNull().WaitAsync(TimeSpan.FromSeconds(10), Ct); + await Task.Delay(softT + grace + TimeSpan.FromMilliseconds(500), Ct); + worker.IsAlive.ShouldBeTrue("a cancelled polling script must not cost its worker"); + runner.Shards[0].Current.ShouldNotBeNull().Pid.ShouldBe(worker.Pid); + + // …and it counted as the timeout it was (threshold 1), not as a kill. + await Scripts.Eventually(() => runner.QuarantinedKeys(ScriptContextKind.VirtualTag).Count == 1, TimeSpan.FromSeconds(5), "the timeout quarantine"); + var refused = await runner.RunAsync(Scripts.Vt("polite", polite, softT, Scripts.Input("spin", 0)), Ct); + refused.Reason.ShouldNotBeNull().ShouldContain("timed out"); + refused.Reason.ShouldNotContain("hard deadline"); + } + // ---------------------------------------------------------------- PendingEval, no process /// The host side of the worker clock, deterministic: no notice means the transport deadline at T + G. From 3f5fdd1d901b3fc9b34270234f5bec332879010c Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 02:24:01 -0500 Subject: [PATCH 5/7] fix(scripting): host progress ceiling kills a worker that heartbeats but never answers (#827 review F3) With the worker soft clock the host's per-run timer is only the transport deadline and the kill G is armed by the worker's soft notice, so a worker whose watchdog never sends one (the design treats the worker as untrusted) held the run, the key's slot and the tag's BadResourceUnavailable forever; a deferred run had no host clock at all. WorkerConnection now tracks evaluation progress (any Results frame, or the submit that made the worker busy again). The heartbeat monitor kills the worker when evaluations are outstanding and nothing progressed for 3 x the largest outstanding T + G plus the silence limit: WorkerExitReason.KilledUnresponsive (metric reason killed_unresponsive), no culprit, attributed like a heartbeat kill (suspects, no strike on a shard). A legitimate run has answered or been killed by T + G. Test seams: ProgressCeilingOverride, DropSoftNoticesForTest. Tests: a worker whose notices are dropped (G = 60 s, so its own fail-fast cannot end it) is killed KilledUnresponsive after the 1.5 s ceiling and the run is released (with the check disabled the test times out); positive control: 20 queued 150 ms runs behind one thread under a 1 s ceiling are all answered. --- .../OutOfProcessScriptRunner.cs | 2 +- .../WorkerConnection.cs | 53 +++++++++++++++++ .../WorkerShard.cs | 3 +- .../WorkerSoftClockTests.cs | 59 +++++++++++++++++++ 4 files changed, 115 insertions(+), 2 deletions(-) diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs index 48aa3f588..723eb0a42 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs @@ -910,7 +910,7 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, case WorkerExitReason.KilledLoadTimeout when exit.CulpritKey is { } culprit: KillStrike(culprit, "its compile did not finish and its worker was killed", "killed"); break; - case WorkerExitReason.Crash or WorkerExitReason.KilledHeartbeat or WorkerExitReason.ProtocolFault: + case WorkerExitReason.Crash or WorkerExitReason.KilledHeartbeat or WorkerExitReason.KilledUnresponsive or WorkerExitReason.ProtocolFault: // Compiles count as well as evaluations: a source that crashes Roslyn (or the load) must be found too. var implicated = exit.InFlightKeys.Concat(exit.LoadingKeys).Distinct(StringComparer.Ordinal).ToArray(); if (shard.IsIsolation) diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs index 911f20ee8..6a2b922ac 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs @@ -21,6 +21,12 @@ internal enum WorkerExitReason /// The host killed it: no frame for three heartbeat intervals. KilledHeartbeat, + /// + /// The host killed it: it kept heartbeating but answered nothing — no result, no start, no soft-deadline notice — + /// for the progress ceiling while evaluations were outstanding (#827 review F3). Attributed like a heartbeat kill. + /// + KilledUnresponsive, + /// The host killed it: a compile did not finish in time (). KilledLoadTimeout, @@ -155,6 +161,12 @@ internal sealed class WorkerConnection : IAsyncDisposable private int _finished; private (WorkerExitReason Reason, string? Culprit)? _killedFor; private long _lastFrameTimestamp; + + /// + /// of the last evaluation progress — a Results frame of any kind — or of the + /// submit that made the worker busy again. See . + /// + private long _lastProgressTimestamp; private long _nextEvalId; private long _nextLoadId; private int _disposed; @@ -235,6 +247,32 @@ internal sealed class WorkerConnection : IAsyncDisposable private long _silenceLimitTicks; + /// + /// The host's own bound on a worker that heartbeats but makes no evaluation progress (#827 review F3). With the + /// worker soft clock the host's per-run timer is only the transport deadline, and the kill G is armed by the + /// worker's soft notice — so a worker whose watchdog never sends one (a bug; the design treats the worker as + /// untrusted) would hold a key's slot, and its tag at BadResourceUnavailable, forever; a deferred run has no host + /// clock at all. So: while evaluations are outstanding and none of them has produced a result, start or soft + /// notice for 3 × the largest outstanding T + G plus the silence limit, the worker is killed + /// (, no culprit: attributed like a heartbeat kill, so no key is + /// struck on a shard). A legitimate run never gets near it: by T + G it has answered or been killed. Null = that + /// rule; set = an override (tests). + /// + internal TimeSpan? ProgressCeilingOverride { get; set; } + + /// Test seam: ignore the worker's soft-deadline notices, as a worker whose watchdog never sends them would. + internal bool DropSoftNoticesForTest { get; set; } + + /// The progress ceiling now (see ); null = no outstanding deadline. + private TimeSpan? ProgressCeiling() + { + if (ProgressCeilingOverride is { } fixedCeiling) return fixedCeiling; + var largest = TimeSpan.Zero; + foreach (var (_, pending) in _pending) + if (pending.HardDeadline > largest) largest = pending.HardDeadline; + return largest == TimeSpan.Zero ? null : (largest * 3) + SilenceLimit; + } + /// Compiles currently outstanding on this worker. public int LoadingCount => _loads.Count; @@ -592,6 +630,7 @@ internal sealed class WorkerConnection : IAsyncDisposable { request.EvalId = (ulong)Interlocked.Increment(ref _nextEvalId); Interlocked.Exchange(ref _lastUsedTimestamp, Stopwatch.GetTimestamp()); + if (_pending.IsEmpty) Interlocked.Exchange(ref _lastProgressTimestamp, Stopwatch.GetTimestamp()); var pending = new PendingEval(key, request.EvalId, onFinished, _dispatcher); if (softTimeout is { } soft) pending.Arm(soft, hardGrace, OnHardDeadline, WorkerSoftClock); _pending[request.EvalId] = pending; @@ -617,6 +656,7 @@ internal sealed class WorkerConnection : IAsyncDisposable var pendings = new PendingEval[items.Count]; var batch = new EvalBatch(); Interlocked.Exchange(ref _lastUsedTimestamp, Stopwatch.GetTimestamp()); + if (_pending.IsEmpty) Interlocked.Exchange(ref _lastProgressTimestamp, Stopwatch.GetTimestamp()); for (var i = 0; i < items.Count; i++) { var item = items[i]; @@ -800,6 +840,8 @@ internal sealed class WorkerConnection : IAsyncDisposable switch (frame.BodyCase) { case Envelope.BodyOneofCase.Results: + Volatile.Write(ref _lastProgressTimestamp, Stopwatch.GetTimestamp()); + if (DropSoftNoticesForTest) frame.Results.SoftExpiredEvalIds.Clear(); // Worker soft clock (lmxopcua#827): the script itself has run for T. Only a worker that agreed to run // the clock may send these. Before the results: the writer coalesces frames, and a notice the worker // sent ahead of the same run's result must still answer the caller with its timeout, as in process. @@ -915,6 +957,17 @@ internal sealed class WorkerConnection : IAsyncDisposable Kill(WorkerExitReason.KilledHeartbeat, null); return; } + if (!_pending.IsEmpty && ProgressCeiling() is { } ceiling) + { + var stuck = Stopwatch.GetElapsedTime(Volatile.Read(ref _lastProgressTimestamp)); + if (stuck > ceiling) + { + _log.Warning("ScriptWorker[{Shard}] pid {Pid} has {Count} evaluation(s) outstanding and made no progress for {Stuck} ms (ceiling {Ceiling} ms) while still heartbeating; killing it", + Shard, Pid, _pending.Count, (long)stuck.TotalMilliseconds, (long)ceiling.TotalMilliseconds); + Kill(WorkerExitReason.KilledUnresponsive, null); + return; + } + } } } catch (OperationCanceledException) { } diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerShard.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerShard.cs index e8359f949..7f1208150 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerShard.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerShard.cs @@ -56,7 +56,7 @@ internal sealed class WorkerShard : IAsyncDisposable private static readonly Counter ExitsCounter = OtOpcUaTelemetry.Meter.CreateCounter( "otopcua.script.worker.exits", unit: "{process}", - description: "ScriptWorker processes ended (tags: shard, reason = clean | killed_deadline | killed_heartbeat | killed_load | crash | protocol | idle | disposed)."); + description: "ScriptWorker processes ended (tags: shard, reason = clean | killed_deadline | killed_heartbeat | killed_unresponsive | killed_load | crash | protocol | idle | disposed)."); private readonly Func, CancellationToken, Task> _launch; private readonly Action _onExit; @@ -439,6 +439,7 @@ internal sealed class WorkerShard : IAsyncDisposable WorkerExitReason.Clean => "clean", WorkerExitReason.KilledDeadline => "killed_deadline", WorkerExitReason.KilledHeartbeat => "killed_heartbeat", + WorkerExitReason.KilledUnresponsive => "killed_unresponsive", WorkerExitReason.KilledLoadTimeout => "killed_load", WorkerExitReason.ProtocolFault => "protocol", WorkerExitReason.Disposed => "disposed", diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs index 7a35adfd4..f0da183bc 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs @@ -306,6 +306,65 @@ public sealed class WorkerSoftClockTests refused.Reason.ShouldNotContain("hard deadline"); } + /// + /// Review F3: the host keeps its own bound. A worker that heartbeats but never sends a soft notice (a bug, or a + /// compromised worker — simulated by the host dropping its notices) used to hold the run, the key's slot and the + /// tag's BadResourceUnavailable forever: the transport deadline answers the caller once and the kill G is armed + /// only by a notice. Now no progress for the ceiling, with evaluations outstanding, kills the worker + /// (, no culprit) and the run completes. G is 60 s here, so the + /// worker's own fail-fast watchdog (notice + G + 0.5 s) cannot be what ends it. + /// + [Fact] + public async Task A_worker_that_heartbeats_but_never_reports_progress_is_killed_at_the_host_ceiling() + { + var spec = new WorkerLaunchSpec("t", OutOfProcessScriptRunner.ResolveWorkerPath(new ScriptWorkerOptions { ExecutablePath = Scripts.WorkerUnderTest }), + 256L * 1024 * 1024, EvalThreads: 2, HardGrace: TimeSpan.FromSeconds(60), HeartbeatInterval: TimeSpan.FromMilliseconds(200), + HandshakeTimeout: TimeSpan.FromSeconds(15), HeartbeatSilenceLimit: TimeSpan.FromSeconds(30)); + var connection = await WorkerConnection.StartAsync(spec, new LoggerConfiguration().CreateLogger(), _ => { }, _ => { }, Ct); + await using var _ = connection; + var entry = await connection.EnsureLoadedAsync("k", ContextKind.VirtualTag, Runaway, + SHA256.HashData(Encoding.UTF8.GetBytes(Runaway)), []).WaitAsync(TimeSpan.FromSeconds(60), Ct); + entry.Outcome.ShouldNotBeNull().Ok.ShouldBeTrue(); + connection.DropSoftNoticesForTest = true; + connection.ProgressCeilingOverride = TimeSpan.FromSeconds(1.5); + + var softT = TimeSpan.FromMilliseconds(200); + var sw = Stopwatch.StartNew(); + var pending = connection.Submit("k", + new EvalRequest { Key = "k", SoftTimeoutMs = (uint)softT.TotalMilliseconds, HardGraceMs = 60_000, NowUtcTicks = DateTime.UtcNow.Ticks }, + softT, TimeSpan.FromSeconds(60), null); + + var exit = await connection.Exited.WaitAsync(TimeSpan.FromSeconds(15), Ct); + exit.Reason.ShouldBe(WorkerExitReason.KilledUnresponsive); + exit.CulpritKey.ShouldBeNull("nothing says which script, so nothing is struck"); + exit.InFlightKeys.ShouldBe(["k"]); + sw.Elapsed.ShouldBeGreaterThanOrEqualTo(TimeSpan.FromSeconds(1.4), "not before the ceiling"); + (await pending.Completion.WaitAsync(TimeSpan.FromSeconds(5), Ct)).ShouldBeNull("the run is released"); + } + + /// Positive control for the ceiling: the default rule leaves a busy, answering worker alone. + [Fact] + public async Task The_progress_ceiling_leaves_a_worker_that_keeps_answering_alone() + { + var spec = new WorkerLaunchSpec("t", OutOfProcessScriptRunner.ResolveWorkerPath(new ScriptWorkerOptions { ExecutablePath = Scripts.WorkerUnderTest }), + 256L * 1024 * 1024, EvalThreads: 1, HardGrace: TimeSpan.FromSeconds(60), HeartbeatInterval: TimeSpan.FromMilliseconds(100), + HandshakeTimeout: TimeSpan.FromSeconds(15), HeartbeatSilenceLimit: TimeSpan.FromSeconds(30)); + var connection = await WorkerConnection.StartAsync(spec, new LoggerConfiguration().CreateLogger(), _ => { }, _ => { }, Ct); + await using var _ = connection; + const string spin = "var end = ctx.Now.AddMilliseconds(150); while (ctx.Now < end) { } return 1;"; + var entry = await connection.EnsureLoadedAsync("k", ContextKind.VirtualTag, spin, + SHA256.HashData(Encoding.UTF8.GetBytes(spin)), []).WaitAsync(TimeSpan.FromSeconds(60), Ct); + entry.Outcome.ShouldNotBeNull().Ok.ShouldBeTrue(); + // Ceiling 1 s; 20 runs of 150 ms queued behind one evaluation thread take ~3 s, but each answers. + connection.ProgressCeilingOverride = TimeSpan.FromSeconds(1); + var soft = TimeSpan.FromSeconds(30); + var pendings = connection.SubmitBatch(Enumerable.Range(0, 20).Select(_ => new BatchSubmission("k", + new EvalRequest { Key = "k", SoftTimeoutMs = 30_000, HardGraceMs = 60_000, NowUtcTicks = DateTime.UtcNow.Ticks }, + soft, TimeSpan.FromSeconds(60), null)).ToArray()); + foreach (var p in pendings) (await p.Completion.WaitAsync(TimeSpan.FromSeconds(30), Ct)).ShouldNotBeNull(); + connection.IsAlive.ShouldBeTrue(); + } + // ---------------------------------------------------------------- PendingEval, no process /// The host side of the worker clock, deterministic: no notice means the transport deadline at T + G. From f50209082e6ec7bf04baf216faf3241a3e9aea65 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 02:25:17 -0500 Subject: [PATCH 6/7] fix(scripting): count re-triggers answered 'worker unresponsive' (#827 review F5) otopcua.script.ipc.timeouts gains the tag answer = stalled | retrigger: a later trigger refused while the key's stalled run is still unanswered was counted nowhere (not a script refusal, so not ScriptRunGuard's counter). KeyState carries its Kind for the tag. Test: the stalled-worker test listens for the instrument and sees one stalled and one retrigger (before: only stalled). --- .../OutOfProcessScriptRunner.cs | 12 ++++++++++-- .../WorkerSoftClockTests.cs | 14 ++++++++++++++ 2 files changed, 24 insertions(+), 2 deletions(-) diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs index 723eb0a42..740bd84bf 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs @@ -102,7 +102,7 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, private static readonly Counter TransportTimeoutCounter = OtOpcUaTelemetry.Meter.CreateCounter( "otopcua.script.ipc.timeouts", unit: "{eval}", - description: "Out-of-process evaluations answered 'script worker unresponsive': no result and no soft-deadline notice from the worker within T + G of submit — a starved or stalled worker, not a slow script (tag: kind; lmxopcua#827)."); + description: "Out-of-process evaluations answered 'script worker unresponsive' (lmxopcua#827; tags: kind, answer = stalled | retrigger). stalled: no result and no soft-deadline notice from the worker within T + G of submit — a starved or stalled worker, not a slow script; retrigger: a later trigger of the same script refused while that stalled run is still unanswered."); static OutOfProcessScriptRunner() { @@ -758,7 +758,12 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, private static ScriptRunOutcome Refused(string key, KeyState state) { if (Volatile.Read(ref state.StalledRun) is { IsCompleted: false } stalled) + { + // Counted, like the run it waits on (#827 review F5): it is not a script refusal, so not ScriptRunGuard's. + TransportTimeoutCounter.Add(1, new KeyValuePair("kind", KindTag(state.Kind)), + new KeyValuePair("answer", "retrigger")); return ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerUnresponsiveReason) with { Recovery = stalled }; + } ScriptRunGuard.RefusedCounter.Add(1, new KeyValuePair("script", key)); return ScriptRunOutcome.Failed(ScriptRunGuard.PreviousRunStillRunningReason); } @@ -1014,6 +1019,9 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, public readonly string Source = source; + /// The seam (metric tags). + public readonly ScriptContextKind Kind = kind; + /// SHA-256 of (computed on first use; a benign race computes the same bytes). public byte[] Sha => _sha ??= SHA256.HashData(Encoding.UTF8.GetBytes(Source)); @@ -1021,7 +1029,7 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, public string ShaHex => _shaHex ??= Convert.ToHexString(Sha); /// The negative compile cache's key for this source (kind-scoped). - public string FailureKey => _failureKey ??= InProcessScriptRunner.ScopedKey(kind, ShaHex); + public string FailureKey => _failureKey ??= InProcessScriptRunner.ScopedKey(Kind, ShaHex); /// The slot order last used for this key (see ). public string[]? Paths; diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs index f0da183bc..eb7631fac 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs @@ -87,6 +87,19 @@ public sealed class WorkerSoftClockTests var grace = TimeSpan.FromMilliseconds(400); await using var runner = Runner(grace); var worker = await WarmAsync(runner, "trivial"); + // Review F5: both answers are counted (otopcua.script.ipc.timeouts, answer = stalled | retrigger). + var answers = new System.Collections.Concurrent.ConcurrentQueue(); + using var listener = new System.Diagnostics.Metrics.MeterListener(); + listener.InstrumentPublished = (instrument, l) => + { + if (instrument.Name == "otopcua.script.ipc.timeouts") l.EnableMeasurementEvents(instrument); + }; + listener.SetMeasurementEventCallback((_, _, tags, _) => + { + foreach (var tag in tags) + if (tag.Key == "answer") answers.Enqueue((string)tag.Value!); + }); + listener.Start(); ScriptRunOutcome stalled, retriggered; using (ProcessFreezer.Freeze(worker.Pid)) @@ -101,6 +114,7 @@ public sealed class WorkerSoftClockTests retriggered.Status.ShouldBe(ScriptRunStatus.Unavailable, retriggered.Reason); retriggered.Reason.ShouldBe(ScriptRunOutcome.WorkerUnresponsiveReason); stalled.ToVirtualTagEvalResult().StatusCode.ShouldBe(Commons.OpcUa.OpcUaStatus.BadResourceUnavailable); + answers.ToArray().ShouldBe(["stalled", "retrigger"], ignoreOrder: true); await stalled.Recovery!.WaitAsync(TimeSpan.FromSeconds(30), Ct); var next = await runner.RunAsync(Scripts.Vt("trivial", Trivial, softT), Ct); From 5962cc3d4121f5f19f2c9946c21204c017070c9a Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 02:26:28 -0500 Subject: [PATCH 7/7] docs(scripting): T starts at the run's start but is wall clock from there; document the late-notice, ceiling and counter rules (#827 review F4) The proto comment, VirtualTags.md and CLAUDE.md said the soft deadline measures the script's own execution and that starvation never counts. Only the time before the run STARTS is excluded; from the start it is wall-clock time, as in process, so a GC pause or a starved / suspended worker during the run still counts. A mid-run freeze test is not added: it races the script's own return against the watchdog's notice; documented as not covered. Also documents F2 (late notice cancels and counts), F3 (progress ceiling, killed_unresponsive) and F5 (answer = stalled | retrigger). --- CLAUDE.md | 2 +- docs/VirtualTags.md | 33 ++++++++++++------- .../WorkerConnection.cs | 5 +-- .../Protos/script_worker.proto | 19 ++++++----- .../WorkerSoftClockTests.cs | 7 ++-- 5 files changed, 40 insertions(+), 26 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 39413236e..e4b0a48f9 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -395,7 +395,7 @@ lists `ScriptedAlarm` — never by default) in `ZB.MOM.WW.OtOpcUa.ScriptWorker` (`src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess`, protocol in `Core.Scripting.Protocol`), shipped in `script-worker/` beside the host (self-contained + started via its apphost on a `-r win-x64 --self-contained` publish, through the muxer otherwise); a runner-side failure publishes `BadResourceUnavailable` (ScadaBridge-visible, ScadaBridge#112) and the seam re-evaluates when the worker is back. `IScriptRunner.RunBatch` sends ready evaluations as one frame per worker (Calc pump, VT coalescer, -alarm dirty set); a run queued behind busy worker threads has its deadlines restarted when it starts, so it is never blamed for a sibling. The soft deadline is the **worker's** clock from the run's start (#827, negotiated via `Hello`/`Ready.worker_soft_clock`); the host only keeps a T + G transport deadline, and passing it is a runner fault (`script worker unresponsive` → `BadResourceUnavailable`, Shadow `shadow_unavailable`), never a script timeout. Optional `Scripting:Worker:PeCacheDir` = HMAC'd compiled-PE cache (restart without Roslyn); optional (Windows) `Scripting:Worker:Identity` = `NT SERVICE\OtOpcUaScriptWorker` runs workers under a restricted token, set up by `scripts/install/Install-ScriptWorkerIsolation.ps1` (ACLs + outbound firewall block; `docs/ServiceHosting.md`); the `script-runner` +alarm dirty set); a run queued behind busy worker threads has its deadlines restarted when it starts, so it is never blamed for a sibling. The soft deadline is the **worker's** clock from the run's start (#827, negotiated via `Hello`/`Ready.worker_soft_clock`); wall clock from there, so a pause during the run still counts. The host keeps a T + G transport deadline (passing it is a runner fault: `script worker unresponsive` → `BadResourceUnavailable`, Shadow `shadow_unavailable`; a later soft notice still cancels and counts the timeout) and a progress ceiling (`killed_unresponsive`) so it never depends on the worker's notices. Optional `Scripting:Worker:PeCacheDir` = HMAC'd compiled-PE cache (restart without Roslyn); optional (Windows) `Scripting:Worker:Identity` = `NT SERVICE\OtOpcUaScriptWorker` runs workers under a restricted token, set up by `scripts/install/Install-ScriptWorkerIsolation.ps1` (ACLs + outbound firewall block; `docs/ServiceHosting.md`); the `script-runner` readiness check is at worst Degraded (never fails ready). The worker tests are `tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests` (`Category=ScriptWorker`, real processes; `Category=Perf` harness opt-in via `OTOPCUA_SCRIPT_PERF=1`). docker-dev overlay: `docker-dev/docker-compose.script-runner.yml`. See `docs/VirtualTags.md` diff --git a/docs/VirtualTags.md b/docs/VirtualTags.md index a456933c9..05845b0ad 100644 --- a/docs/VirtualTags.md +++ b/docs/VirtualTags.md @@ -58,18 +58,27 @@ processes: `script timed out after …` failure and the run token is cancelled inside the worker; at `T + G` (`Scripting:Worker:HardGrace`, default `max(1 s, T)`) the worker is **killed** (`Process.Kill(entireProcessTree)`, plus a worker-side fail-fast watchdog) — the thread, its memory and its ALC go with it. -- **`T` is the script's own run time (lmxopcua#827).** As in process (where the guard's clock starts when the - thread pool picks the run up), `T` starts when a worker evaluation thread starts the run — not at submit — so the - pipe round trip and the time a starved worker waits to be scheduled never count against the script. The worker runs - that clock and reports the crossing (`EvalResults.soft_expired_eval_ids`); the host answers its caller, sends the - cancel, and arms the kill `G` after the notice. What the host still times on its own is a **transport deadline**, - `T + G` from submit (restarted at a queued run's start): no result and no soft notice by then is the runner's fault — - the tag publishes **`BadResourceUnavailable`** (`"script worker unresponsive"`), nothing is struck or quarantined, - the worker is not killed (a really hung one goes silent and the heartbeat monitor kills it), and the seam - re-evaluates when the worker answers. A re-trigger while that run is still unanswered is the runner's fault too, not - `previous run still running`. Counted as `otopcua.script.ipc.timeouts{kind}`. The clock is negotiated in the handshake - (`Hello`/`Ready.worker_soft_clock`): a worker that does not echo it keeps the host clock from submit (the pre-#827 - contract), so a mismatched host/worker pair degrades to the old behaviour instead of losing its deadlines. +- **`T` starts when the run starts (lmxopcua#827).** As in process (where the guard's clock starts when the thread + pool picks the run up), `T` starts when a worker evaluation thread starts the run — not at submit — so the pipe + round trip and the time a starved worker takes to **start** the run never count against the script. From the start + on it is wall-clock time, exactly as in process: a GC pause, a CPU-starved or suspended worker **during** the run + still counts against `T` (not separated, and not tested beyond that equivalence). The worker runs that clock and + reports the crossing (`EvalResults.soft_expired_eval_ids`); the host resumes its caller on the result dispatcher + (never on the pipe reader), sends the cancel, and arms the kill `G` after the notice. + - **Transport deadline.** What the host still times on its own is `T + G` from submit (restarted at a queued run's + start): no result and no soft notice by then is the runner's fault — the tag publishes **`BadResourceUnavailable`** + (`"script worker unresponsive"`), nothing is struck, the worker is not killed, and the seam re-evaluates when the + worker answers. A re-trigger while that run is still unanswered gets the same answer, not + `previous run still running`. If the worker then reports the script over `T` after all (slow **and** late — a + notice delayed past `T + G` lands here too), the run still gets its cancel and counts as a timeout. Counted as + `otopcua.script.ipc.timeouts{kind, answer = stalled | retrigger}`. + - **Progress ceiling.** The host does not depend on the worker's notices to end a run: while evaluations are + outstanding, a worker that keeps heartbeating but produces no result, start or soft notice for + `3 × the largest outstanding T + G + 3 heartbeat intervals` is killed (`killed_unresponsive`, no culprit, + attributed like a heartbeat kill). A worker that stops heartbeating is killed by the silence rule as before. + - **Negotiated.** The clock is agreed in the handshake (`Hello`/`Ready.worker_soft_clock`): a worker that does not + echo it keeps the host clock from submit (the pre-#827 contract), so a mismatched host/worker pair degrades to the + old behaviour instead of losing its deadlines. - **Attribution and quarantine** live in the host and survive restarts: a deadline kill strikes exactly its key (other evaluations in flight are victims, not strikes); a crash with no deadline crossed makes every key in flight a suspect that must complete `Scripting:Worker:IsolationProbationRuns` (20) clean runs in the isolation worker, diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs index 6a2b922ac..31a128db9 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs @@ -204,8 +204,9 @@ internal sealed class WorkerConnection : IAsyncDisposable /// /// lmxopcua#827: the worker runs each evaluation's soft clock from the moment an evaluation thread starts it and - /// reports the crossing, so the soft deadline measures the script's own execution — not the pipe, not the time - /// a starved worker takes to be scheduled. The host keeps a transport deadline of its own instead (see + /// reports the crossing, so the soft deadline no longer counts the pipe or the time a starved worker takes to + /// START the run. From the start it is wall-clock time, as in process: a pause during the run still counts. The + /// host keeps a transport deadline of its own instead (see /// ). Negotiated: asked for in Hello, true only when the worker echoed it in /// Ready; otherwise the host's clock, started at submit, is the soft deadline (the pre-#827 contract). /// diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/Protos/script_worker.proto b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/Protos/script_worker.proto index 5d73da467..ed67c6a63 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/Protos/script_worker.proto +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/Protos/script_worker.proto @@ -147,14 +147,17 @@ message EvalResult { // just finished. So a run waiting behind a sibling is never timed out, struck or killed for the sibling's time. // Either list may be the only content of a frame. A host that predates these fields ignores them. // -// Worker soft clock (lmxopcua#827, negotiated by Hello/Ready.worker_soft_clock): the soft deadline T measures the -// script's OWN execution — the worker starts it when an evaluation thread starts the run and reports the crossing in -// `soft_expired_eval_ids` (it does not cancel the run itself: the host answers its caller first, then sends Cancel, -// keeping the in-process ordering). The host arms the hard deadline G from that notice. What the host keeps on its own -// clock is a TRANSPORT deadline, T + G from submit (restarted at a deferred run's start): no result and no soft notice -// by then means the pipe or the worker process is late — the caller is answered as a runner fault ("script worker -// unresponsive", BadResourceUnavailable), never as a script timeout, and nothing is struck. A starved or stalled worker -// therefore cannot make a trivial script look over budget; a hung one is still caught by the heartbeat silence kill. +// Worker soft clock (lmxopcua#827, negotiated by Hello/Ready.worker_soft_clock): the soft deadline T starts when an +// evaluation thread starts the run (not at submit) and is wall-clock time from there, as in process — a pause or +// starvation DURING the run still counts. The worker reports the crossing in `soft_expired_eval_ids` (it does not +// cancel the run itself: the host answers its caller first, then sends Cancel, keeping the in-process ordering). The +// host arms the hard deadline G from that notice. What the host keeps on its own clock is a TRANSPORT deadline, T + G +// from submit (restarted at a deferred run's start): no result and no soft notice by then means the pipe or the worker +// process is late — the caller is answered as a runner fault ("script worker unresponsive", BadResourceUnavailable), +// not as a script timeout, and nothing is struck (a soft notice that arrives later still cancels the run and counts as +// a timeout). A worker that starts runs late therefore cannot make a trivial script look over budget; one that stops +// heartbeating is killed for silence, and one that heartbeats without progress is killed at the host's progress +// ceiling. message EvalResults { repeated EvalResult items = 1; repeated uint64 started_eval_ids = 2; diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs index eb7631fac..bc9acb9af 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs @@ -12,9 +12,10 @@ using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.V1; namespace ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests; /// -/// lmxopcua#827: the soft deadline T measures the script's OWN execution — the worker runs it from the moment an -/// evaluation thread starts the run — so the pipe round trip and the time a starved worker waits to be scheduled -/// no longer count against the script. The host keeps a transport deadline (T + G from submit) and reports passing +/// lmxopcua#827: the soft deadline T starts when an evaluation thread starts the run — the worker runs that clock — +/// so the pipe round trip and the time a starved worker takes to START the run no longer count against the script. +/// From the start it is wall-clock time, as in process: a pause DURING the run still counts, and is not tested +/// here (a mid-run freeze races the script's own return against the watchdog's notice). The host keeps a transport deadline (T + G from submit) and reports passing /// it as a runner fault (, BadResourceUnavailable), never as a /// script timeout. The starved worker is made real here: its process is suspended (SIGSTOP / NtSuspendProcess) /// while a trivial script is submitted.