diff --git a/CLAUDE.md b/CLAUDE.md index 89375e172..201d68593 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -398,7 +398,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`); 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 5c66d01ad..05845b0ad 100644 --- a/docs/VirtualTags.md +++ b/docs/VirtualTags.md @@ -58,6 +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` 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, @@ -108,7 +129,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 +189,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..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 @@ -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' (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() { // Gauges observe every live runner (one per node in production; tests make several). @@ -137,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; @@ -155,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; @@ -372,11 +380,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 +537,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 +562,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 +587,33 @@ 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, + // 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); + 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); @@ -704,7 +735,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); @@ -717,6 +749,25 @@ 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) + { + // 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); + } + 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). @@ -864,7 +915,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) @@ -968,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)); @@ -975,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; @@ -986,5 +1040,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..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 @@ -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, @@ -81,6 +87,14 @@ 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. +/// internal sealed record WorkerLaunchSpec( string Shard, string WorkerPath, @@ -94,7 +108,9 @@ internal sealed record WorkerLaunchSpec( string? MarkerDirectory = null, TimeSpan? LoadTimeout = null, int CompileThreads = 0, - string? WorkerSid = null) + string? WorkerSid = null, + bool WorkerSoftClock = true, + TimeSpan? HeartbeatSilenceLimit = null) { /// The default compile timeout. public static readonly TimeSpan DefaultLoadTimeout = TimeSpan.FromSeconds(60); @@ -145,14 +161,21 @@ 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; 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, TimeSpan? heartbeatSilenceLimit) { + WorkerSoftClock = workerSoftClock; _stdio = stdio; _loadTimeout = loadTimeout; _onExiting = onExiting; @@ -165,6 +188,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, @@ -178,6 +202,16 @@ 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 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). + /// + public bool WorkerSoftClock { get; } + /// Completes when the worker has ended, with why. public Task Exited => _exited.Task; @@ -200,6 +234,46 @@ 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; + + /// + /// 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; @@ -300,19 +374,20 @@ 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, + heartbeatSilenceLimit: spec.HeartbeatSilenceLimit); connection.Begin(); return connection; } @@ -556,8 +631,9 @@ 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); + if (softTimeout is { } soft) pending.Arm(soft, hardGrace, OnHardDeadline, WorkerSoftClock); _pending[request.EvalId] = pending; var batch = new EvalBatch(); batch.Items.Add(request); @@ -581,12 +657,13 @@ 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]; 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 +841,18 @@ 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. + 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 @@ -863,12 +952,23 @@ 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); 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) { } @@ -1026,18 +1126,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,12 +1174,15 @@ 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; private int _done; + private TaskCompletionSource? _lateSoftExpiry; public PendingEval(string key, ulong evalId, Action? onFinished = null, ResultDispatcher? dispatcher = null) { @@ -1086,24 +1216,41 @@ 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). + /// 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 + /// 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 +1269,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 +1280,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 +1305,75 @@ 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, late; + 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; + late = phase == TransportReported; + Interlocked.Exchange(ref _phase, SoftReported); + _hardDue = Stopwatch.GetTimestamp() + Ticks(_grace); + TryChange(_grace); + } + // 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() { 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); + _lateSoftExpiry = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + 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); } @@ -1192,6 +1393,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/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/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..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 @@ -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,23 @@ 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 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; 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.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/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.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 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 => 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..bc9acb9af --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs @@ -0,0 +1,492 @@ +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 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. +/// +[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"); + // 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)) + { + 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); + 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); + 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)); + } + + // ---------------------------------------------------------------- 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"); + } + + /// + /// 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. + [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 +}