diff --git a/CLAUDE.md b/CLAUDE.md index c70af7e8e..39413236e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -395,7 +395,7 @@ lists `ScriptedAlarm` — never by default) in `ZB.MOM.WW.OtOpcUa.ScriptWorker` (`src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess`, protocol in `Core.Scripting.Protocol`), shipped in `script-worker/` beside the host (self-contained + started via its apphost on a `-r win-x64 --self-contained` publish, through the muxer otherwise); a runner-side failure publishes `BadResourceUnavailable` (ScadaBridge-visible, ScadaBridge#112) and the seam re-evaluates when the worker is back. `IScriptRunner.RunBatch` sends ready evaluations as one frame per worker (Calc pump, VT coalescer, -alarm dirty set); a run queued behind busy worker threads has its deadlines restarted when it starts, so it is never blamed for a sibling. Optional `Scripting:Worker:PeCacheDir` = HMAC'd compiled-PE cache (restart without Roslyn); optional (Windows) `Scripting:Worker:Identity` = `NT SERVICE\OtOpcUaScriptWorker` runs workers under a restricted token, set up by `scripts/install/Install-ScriptWorkerIsolation.ps1` (ACLs + outbound firewall block; `docs/ServiceHosting.md`); the `script-runner` +alarm dirty set); a run queued behind busy worker threads has its deadlines restarted when it starts, so it is never blamed for a sibling. The soft deadline is the **worker's** clock from the run's start (#827, negotiated via `Hello`/`Ready.worker_soft_clock`); the host only keeps a T + G transport deadline, and passing it is a runner fault (`script worker unresponsive` → `BadResourceUnavailable`, Shadow `shadow_unavailable`), never a script timeout. Optional `Scripting:Worker:PeCacheDir` = HMAC'd compiled-PE cache (restart without Roslyn); optional (Windows) `Scripting:Worker:Identity` = `NT SERVICE\OtOpcUaScriptWorker` runs workers under a restricted token, set up by `scripts/install/Install-ScriptWorkerIsolation.ps1` (ACLs + outbound firewall block; `docs/ServiceHosting.md`); the `script-runner` readiness check is at worst Degraded (never fails ready). The worker tests are `tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests` (`Category=ScriptWorker`, real processes; `Category=Perf` harness opt-in via `OTOPCUA_SCRIPT_PERF=1`). docker-dev overlay: `docker-dev/docker-compose.script-runner.yml`. See `docs/VirtualTags.md` diff --git a/docs/VirtualTags.md b/docs/VirtualTags.md index 5c66d01ad..a456933c9 100644 --- a/docs/VirtualTags.md +++ b/docs/VirtualTags.md @@ -58,6 +58,18 @@ processes: `script timed out after …` failure and the run token is cancelled inside the worker; at `T + G` (`Scripting:Worker:HardGrace`, default `max(1 s, T)`) the worker is **killed** (`Process.Kill(entireProcessTree)`, plus a worker-side fail-fast watchdog) — the thread, its memory and its ALC go with it. +- **`T` is the script's own run time (lmxopcua#827).** As in process (where the guard's clock starts when the + thread pool picks the run up), `T` starts when a worker evaluation thread starts the run — not at submit — so the + pipe round trip and the time a starved worker waits to be scheduled never count against the script. The worker runs + that clock and reports the crossing (`EvalResults.soft_expired_eval_ids`); the host answers its caller, sends the + cancel, and arms the kill `G` after the notice. What the host still times on its own is a **transport deadline**, + `T + G` from submit (restarted at a queued run's start): no result and no soft notice by then is the runner's fault — + the tag publishes **`BadResourceUnavailable`** (`"script worker unresponsive"`), nothing is struck or quarantined, + the worker is not killed (a really hung one goes silent and the heartbeat monitor kills it), and the seam + re-evaluates when the worker answers. A re-trigger while that run is still unanswered is the runner's fault too, not + `previous run still running`. Counted as `otopcua.script.ipc.timeouts{kind}`. The clock is negotiated in the handshake + (`Hello`/`Ready.worker_soft_clock`): a worker that does not echo it keeps the host clock from submit (the pre-#827 + contract), so a mismatched host/worker pair degrades to the old behaviour instead of losing its deadlines. - **Attribution and quarantine** live in the host and survive restarts: a deadline kill strikes exactly its key (other evaluations in flight are victims, not strikes); a crash with no deadline crossed makes every key in flight a suspect that must complete `Scripting:Worker:IsolationProbationRuns` (20) clean runs in the isolation worker, @@ -108,7 +120,7 @@ processes: "every process with this name", never a live host's worker (mxaccessgw's `OrphanWorkerTerminator` lesson). - **Observability.** On `OtOpcUaTelemetry.Meter`: `otopcua.script.runner.state` (0 ready / 1 degraded / 2 circuit open), `otopcua.script.worker.up` / `.inflight` / `.queue_depth` / `.heap_bytes` (tag `shard`; queue and heap from - the worker heartbeat), `otopcua.script.worker.starts` / `.exits` (`reason` adds `idle`), `otopcua.script.ipc.roundtrip` + the worker heartbeat), `otopcua.script.worker.starts` / `.exits` (`reason` adds `idle`), `otopcua.script.ipc.roundtrip`, `otopcua.script.ipc.timeouts` (transport deadline, lmxopcua#827) and `otopcua.script.eval.duration` (µs, tag `kind`; no per-key tag), `otopcua.script.compile` (`cache` = `hit` | `miss` | `rejected` | `uncached` | `negative`) and `otopcua.script.isolation.suspects`. The `script-runner` health check is on the **ready** tier and is at worst **Degraded** (still 200): a sick runner never fails readiness, so a pair @@ -168,7 +180,8 @@ what is published** — and each compiled evaluation is then mirrored to the `Sc - **Compared:** value (structural for arrays), exact CLR type, status (Ok vs failed) and failure class (`compile`, `budget` = timed out / killed / refused / quarantined, `threw` with its message, `unsupported_type`). A shadow outcome - that is the runner's own fault (worker restarted, none ready, worker busy) is **not compared** + that is the runner's own fault (worker restarted, none ready, worker busy, worker unresponsive — a stalled or starved + worker past the transport deadline, lmxopcua#827) is **not compared** (`result = shadow_unavailable`). An input or result the protocol cannot carry (design §14 #11) counts as a mismatch with reason `unsupported_type` — out of process that evaluation would publish Bad. - **Never reaches a published value or readiness.** The mirror copies the inputs and queues the shadow run on the pool; diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs index 4497440c2..06b748b4d 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs @@ -100,6 +100,10 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, "otopcua.script.eval.duration", unit: "us", description: "Worker-measured run time of one out-of-process evaluation (tag: kind)."); + private static readonly Counter TransportTimeoutCounter = OtOpcUaTelemetry.Meter.CreateCounter( + "otopcua.script.ipc.timeouts", unit: "{eval}", + description: "Out-of-process evaluations answered 'script worker unresponsive': no result and no soft-deadline notice from the worker within T + G of submit — a starved or stalled worker, not a slow script (tag: kind; lmxopcua#827)."); + static OutOfProcessScriptRunner() { // Gauges observe every live runner (one per node in production; tests make several). @@ -372,11 +376,7 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, return ScriptRunOutcome.Failed(knownFailure); } - if (Interlocked.CompareExchange(ref state.InFlight, 1, 0) != 0) - { - ScriptRunGuard.RefusedCounter.Add(1, new KeyValuePair("script", key)); - return ScriptRunOutcome.Failed(ScriptRunGuard.PreviousRunStillRunningReason); - } + if (Interlocked.CompareExchange(ref state.InFlight, 1, 0) != 0) return Refused(key, state); var handedOff = false; SemaphoreSlim? gate = null; @@ -533,8 +533,7 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, if (Interlocked.CompareExchange(ref state.InFlight, 1, 0) != 0) { - ScriptRunGuard.RefusedCounter.Add(1, new KeyValuePair("script", key)); - answered = ScriptRunOutcome.Failed(ScriptRunGuard.PreviousRunStillRunningReason); + answered = Refused(key, state); return BatchPreparation.Answered; } state.ScriptLogger = invocation.ScriptLogger; @@ -559,6 +558,7 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, { if (Interlocked.Exchange(ref released, 1) != 0) return; SlotReleasingForTest?.Invoke(key); + Volatile.Write(ref state.StalledRun, null); Interlocked.Exchange(ref state.InFlight, 0); heldGate?.Release(); }; @@ -583,6 +583,21 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, return ScriptRunOutcome.Failed($"{ScriptEvalFailureReasons.TimedOutPrefix}{invocation.Timeout.TotalSeconds:F1}s"); } + if (ReferenceEquals(result, PendingEval.TransportTimedOut)) + { + // lmxopcua#827: T + G passed with neither a result nor a soft-deadline notice. The worker runs the soft + // clock from the run's start, so this is the pipe or a starved / stalled worker — the runner's fault, + // never the script's: no timeout strike, no Cancel (the run may not even have started), and the slot + // stays with the run until the worker answers. The seam re-evaluates when it does (or when the worker + // is replaced, if the heartbeat monitor kills it). + TransportTimeoutCounter.Add(1, new KeyValuePair("kind", KindTag(invocation.Kind))); + // Read by a refusal only while it is incomplete, so a write that loses a race with the slot release + // is harmless. + Volatile.Write(ref state.StalledRun, pending.Completion); + if (ct.IsCancellationRequested) return ScriptRunOutcome.Failed("evaluation cancelled"); + return ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerUnresponsiveReason) with { Recovery = pending.Completion }; + } + if (result is null) return Unavailable(ScriptRunOutcome.WorkerRestartedReason, RouteFor(key, state), connection); var kindTag = new KeyValuePair("kind", KindTag(invocation.Kind)); RoundTripHistogram.Record(Stopwatch.GetElapsedTime(submitted).TotalMicroseconds, kindTag); @@ -717,6 +732,20 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, private static ScriptRunOutcome Unavailable(string reason, WorkerShard shard, WorkerConnection? failed) => ScriptRunOutcome.Unavailable(reason) with { Recovery = shard.WhenRecoveredAsync(failed) }; + /// + /// The answer to an evaluation whose key's previous run still holds its single-flight slot: the usual + /// script-attributable refusal — unless that run was answered "worker unresponsive" (lmxopcua#827) and the + /// worker still has not answered it, in which case nothing says the script is slow, and the refusal is the + /// runner's fault too (re-evaluated when the stalled run completes). + /// + private static ScriptRunOutcome Refused(string key, KeyState state) + { + if (Volatile.Read(ref state.StalledRun) is { IsCompleted: false } stalled) + return ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerUnresponsiveReason) with { Recovery = stalled }; + ScriptRunGuard.RefusedCounter.Add(1, new KeyValuePair("script", key)); + return ScriptRunOutcome.Failed(ScriptRunGuard.PreviousRunStillRunningReason); + } + private void NoteLoad(WorkerConnection.LoadEntry entry) { // Counted once per compile, by the caller that waited for it (an already-loaded key is not a load). @@ -986,5 +1015,11 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, public volatile bool Suspect; public string? QuarantineReason; public ILogger? ScriptLogger; + + /// + /// The completion of this key's in-flight run while it is answered "worker unresponsive" and not yet answered + /// by the worker (lmxopcua#827); null otherwise. Cleared by the slot release. + /// + public Task? StalledRun; } } diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs index 85b6f6d94..a9fce4fa4 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs @@ -81,6 +81,10 @@ internal sealed record BatchSubmission(string Key, EvalRequest Request, TimeSpan /// Windows only: the worker identity's SID (Scripting:Worker:Identity). Set, the worker starts under a restricted /// token () and the pipe ACL admits it beside the host; null keeps the host's token. /// +/// +/// Ask the worker to run the soft clock from each run's start (lmxopcua#827; see ). +/// False (tests only) keeps the pre-#827 host clock, which is also what a worker that does not echo it gets. +/// internal sealed record WorkerLaunchSpec( string Shard, string WorkerPath, @@ -94,7 +98,8 @@ internal sealed record WorkerLaunchSpec( string? MarkerDirectory = null, TimeSpan? LoadTimeout = null, int CompileThreads = 0, - string? WorkerSid = null) + string? WorkerSid = null, + bool WorkerSoftClock = true) { /// The default compile timeout. public static readonly TimeSpan DefaultLoadTimeout = TimeSpan.FromSeconds(60); @@ -151,8 +156,9 @@ internal sealed class WorkerConnection : IAsyncDisposable private WorkerConnection(string shard, Process process, NamedPipeServerStream pipe, IDisposable? job, IDisposable? stdio, string workDir, ILogger log, Action onLogs, TimeSpan heartbeatInterval, TimeSpan loadTimeout, - Action onExiting, string? markerPath) + Action onExiting, string? markerPath, bool workerSoftClock) { + WorkerSoftClock = workerSoftClock; _stdio = stdio; _loadTimeout = loadTimeout; _onExiting = onExiting; @@ -178,6 +184,15 @@ internal sealed class WorkerConnection : IAsyncDisposable /// The worker's process id. public int Pid { get; } + /// + /// lmxopcua#827: the worker runs each evaluation's soft clock from the moment an evaluation thread starts it and + /// reports the crossing, so the soft deadline measures the script's own execution — not the pipe, not the time + /// a starved worker takes to be scheduled. The host keeps a transport deadline of its own instead (see + /// ). Negotiated: asked for in Hello, true only when the worker echoed it in + /// Ready; otherwise the host's clock, started at submit, is the soft deadline (the pre-#827 contract). + /// + public bool WorkerSoftClock { get; } + /// Completes when the worker has ended, with why. public Task Exited => _exited.Task; @@ -300,19 +315,19 @@ internal sealed class WorkerConnection : IAsyncDisposable HeartbeatIntervalMs = (uint)spec.HeartbeatInterval.TotalMilliseconds, MaxLogLines = 1000, CompileThreads = (uint)Math.Max(0, spec.CompileThreads), + WorkerSoftClock = spec.WorkerSoftClock, }; if (spec.PeCacheDirectory is { } cacheDir && spec.PeCacheKey is { } cacheKey) { hello.PeCacheDir = cacheDir; hello.PeCacheKey = Google.Protobuf.ByteString.CopyFrom(cacheKey); } - if (ScriptWorkerFrameCodec.UseBlockingIo) - await HandshakeBlockingAsync(pipe, hello, nonce, process.Id, handshakeCts.Token).ConfigureAwait(false); - else - await HandshakeAsync(pipe, hello, nonce, process.Id, handshakeCts.Token).ConfigureAwait(false); + var ready = ScriptWorkerFrameCodec.UseBlockingIo + ? await HandshakeBlockingAsync(pipe, hello, nonce, process.Id, handshakeCts.Token).ConfigureAwait(false) + : await HandshakeAsync(pipe, hello, nonce, process.Id, handshakeCts.Token).ConfigureAwait(false); var connection = new WorkerConnection(spec.Shard, process, pipe, job, stdio, workDir, log, onLogs, spec.HeartbeatInterval, - spec.EffectiveLoadTimeout, onExiting, markerPath); + spec.EffectiveLoadTimeout, onExiting, markerPath, workerSoftClock: spec.WorkerSoftClock && ready.WorkerSoftClock); connection.Begin(); return connection; } @@ -557,7 +572,7 @@ internal sealed class WorkerConnection : IAsyncDisposable request.EvalId = (ulong)Interlocked.Increment(ref _nextEvalId); Interlocked.Exchange(ref _lastUsedTimestamp, Stopwatch.GetTimestamp()); var pending = new PendingEval(key, request.EvalId, onFinished, _dispatcher); - if (softTimeout is { } soft) pending.Arm(soft, hardGrace, OnHardDeadline); + if (softTimeout is { } soft) pending.Arm(soft, hardGrace, OnHardDeadline, WorkerSoftClock); _pending[request.EvalId] = pending; var batch = new EvalBatch(); batch.Items.Add(request); @@ -586,7 +601,7 @@ internal sealed class WorkerConnection : IAsyncDisposable var item = items[i]; item.Request.EvalId = (ulong)Interlocked.Increment(ref _nextEvalId); var pending = new PendingEval(item.Key, item.Request.EvalId, item.OnFinished, _dispatcher); - pending.Arm(item.SoftTimeout, item.HardGrace, OnHardDeadline); + pending.Arm(item.SoftTimeout, item.HardGrace, OnHardDeadline, WorkerSoftClock); _pending[item.Request.EvalId] = pending; batch.Items.Add(item.Request); pendings[i] = pending; @@ -764,6 +779,16 @@ internal sealed class WorkerConnection : IAsyncDisposable switch (frame.BodyCase) { case Envelope.BodyOneofCase.Results: + // Worker soft clock (lmxopcua#827): the script itself has run for T. Only a worker that agreed to run + // the clock may send these. Before the results: the writer coalesces frames, and a notice the worker + // sent ahead of the same run's result must still answer the caller with its timeout, as in process. + if (frame.Results.SoftExpiredEvalIds.Count > 0 && !WorkerSoftClock) + throw new ScriptWorkerProtocolException("soft-deadline notice from a worker that does not run the soft clock"); + foreach (var id in frame.Results.SoftExpiredEvalIds) + { + if (_pending.TryGetValue(id, out var expired)) expired.SoftExpired(); + else CheckIssued(id, "soft-deadline notice"); + } foreach (var result in frame.Results.Items) { // An eval id with no outstanding request is a protocol violation (design §6 "what the host @@ -1026,18 +1051,45 @@ internal sealed class WorkerConnection : IAsyncDisposable /// /// One evaluation outstanding on a worker. Its deadlines share ONE timer (lmxopcua#789: the per-eval /// WaitAsync(timeout) timer, the hard-deadline CancellationTokenSource and their continuations were a -/// measurable share of the host's CPU): it fires at the soft deadline T and is re-armed for the grace G. +/// measurable share of the host's CPU). /// /// -/// The deadlines are absolute timestamps the timer callback re-checks, so a deferral (the worker queued the run behind -/// busy evaluation threads, lmxopcua#780 batch submit) can stop the clock and a start notice can restart it — the full -/// T, then G, from the start — without a stale tick firing early. +/// +/// Worker soft clock (lmxopcua#827, ): the soft deadline T is +/// the worker's — it reports, through , that the script has itself run for T, and the +/// hard deadline G is armed from that notice. The host's timer is then only a TRANSPORT deadline: T + G from +/// submit (restarted at a deferred run's start). Passing it with neither a result nor a soft notice means the +/// pipe or the worker process is late, not the script — yields +/// (a runner fault: nothing is struck and the worker is not killed; a worker that +/// is really hung goes silent and the heartbeat monitor kills it). So a starved worker can no longer make a +/// trivial script report "timed out", which in Shadow mode read as a script mismatch. +/// +/// +/// Host clock (a worker that did not agree to the above): the timer fires at the soft deadline T, counted +/// from submit, and is re-armed for the grace G. +/// +/// +/// The deadlines are absolute timestamps the timer callback re-checks, so a deferral (the worker queued the run +/// behind busy evaluation threads, lmxopcua#780 batch submit) can stop the clock and a start notice can restart +/// it — the full T (or transport deadline), then G, from the start — without a stale tick firing early. +/// /// internal sealed class PendingEval { /// What yields when the soft deadline passed first. Never on the wire. internal static readonly EvalResult SoftTimedOut = new(); + /// + /// What yields when the transport deadline passed with no word from the worker (worker soft + /// clock only, lmxopcua#827): a runner fault, not the script's. Never on the wire. + /// + internal static readonly EvalResult TransportTimedOut = new(); + + private const int Waiting = 0; + private const int SoftReported = 1; + private const int HardFired = 2; + private const int TransportReported = 3; + private readonly TaskCompletionSource _completion = new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly TaskCompletionSource _outcome; private readonly ResultDispatcher? _dispatcher; @@ -1047,8 +1099,10 @@ internal sealed class PendingEval private Action? _onHardDeadline; private TimeSpan _soft; private TimeSpan _grace; + private bool _workerClock; private long _softDue; private long _hardDue; + private long _transportDue; private bool _deferred; private bool _started; private int _phase; @@ -1086,24 +1140,31 @@ internal sealed class PendingEval public Task Completion => _completion.Task; /// - /// What the caller waits on: the worker's result, null when the worker ended first, or - /// at the soft deadline (the run itself is still outstanding then). + /// What the caller waits on: the worker's result, null when the worker ended first, at + /// the soft deadline, or at the transport deadline (the run itself is still + /// outstanding in both of the last two cases). /// public Task Outcome => _outcome.Task; /// Starts the deadline timer. Called once, before the request is sent. - internal void Arm(TimeSpan soft, TimeSpan grace, Action onHardDeadline) + /// T. + /// G. + /// The kill. + /// The worker runs the soft clock (lmxopcua#827): this timer is the transport deadline. + internal void Arm(TimeSpan soft, TimeSpan grace, Action onHardDeadline, bool workerClock = false) { _soft = soft; _grace = grace; _onHardDeadline = onHardDeadline; + _workerClock = workerClock; HardDeadline = soft + grace; var now = Stopwatch.GetTimestamp(); _softDue = now + Ticks(soft); _hardDue = _softDue + Ticks(grace); + _transportDue = _hardDue; // Created disarmed and assigned before it can fire, so the soft tick can always re-arm it for the grace. _timer = new Timer(static s => ((PendingEval)s!).OnTimer(), this, Timeout.Infinite, Timeout.Infinite); - _timer.Change(soft, Timeout.InfiniteTimeSpan); + _timer.Change(workerClock ? soft + grace : soft, Timeout.InfiniteTimeSpan); } /// @@ -1122,7 +1183,8 @@ internal sealed class PendingEval /// /// The worker started this run: a deferred run's deadlines restart from now (T, or T + G when its soft deadline - /// had already been reported). A run that was never deferred keeps the clock it has had since submit. + /// had already been reported; with the worker soft clock, the transport deadline T + G). A run that was never + /// deferred keeps the clock it has had since submit. /// internal void Start() { @@ -1132,7 +1194,16 @@ internal sealed class PendingEval if (_timer is null || !_deferred || Volatile.Read(ref _done) != 0) return; _deferred = false; var now = Stopwatch.GetTimestamp(); - if (Volatile.Read(ref _phase) == 0) + if (_workerClock) + { + // Its soft deadline is the worker's from here; the transport deadline restarts. Already answered (a + // deferral notice that arrived after the transport deadline): nothing to re-arm — a soft notice or the + // result is what comes next. + if (Volatile.Read(ref _phase) != Waiting) return; + _transportDue = now + Ticks(_soft + _grace); + TryChange(_soft + _grace); + } + else if (Volatile.Read(ref _phase) == Waiting) { _softDue = now + Ticks(_soft); _hardDue = _softDue + Ticks(_grace); @@ -1148,31 +1219,63 @@ internal sealed class PendingEval } } + /// + /// The worker reports that the script has itself run for T (worker soft clock, lmxopcua#827): the caller gets its + /// soft timeout now — unless the transport deadline already answered it — and the hard deadline is G from here. + /// + internal void SoftExpired() + { + bool signal; + lock (_clock) + { + if (_timer is null || !_workerClock || Volatile.Read(ref _done) != 0) return; + var phase = Volatile.Read(ref _phase); + if (phase is SoftReported or HardFired) return; + _started = true; + _deferred = false; + signal = phase == Waiting; + Interlocked.Exchange(ref _phase, SoftReported); + _hardDue = Stopwatch.GetTimestamp() + Ticks(_grace); + TryChange(_grace); + } + // Outside the lock: a caller with a dispatcher resumes synchronously inside TrySetResult. + if (signal) _outcome.TrySetResult(SoftTimedOut); + } + private void OnTimer() { if (Volatile.Read(ref _done) != 0) return; - bool soft = false, hard = false; + bool soft = false, hard = false, transport = false; lock (_clock) { if (_deferred || Volatile.Read(ref _done) != 0) return; var now = Stopwatch.GetTimestamp(); - if (Volatile.Read(ref _phase) == 0) + var phase = Volatile.Read(ref _phase); + if (phase == Waiting && _workerClock) + { + // A tick armed before a deferral/start can land early: re-arm for what is left. + if (now < _transportDue) { TryChange(Stopwatch.GetElapsedTime(now, _transportDue)); return; } + Interlocked.Exchange(ref _phase, TransportReported); + transport = true; + } + else if (phase == Waiting) { // A tick armed before a deferral/start can land early: re-arm for what is left. if (now < _softDue) { TryChange(Stopwatch.GetElapsedTime(now, _softDue)); return; } - Interlocked.Exchange(ref _phase, 1); + Interlocked.Exchange(ref _phase, SoftReported); soft = true; TryChange(_hardDue > now ? Stopwatch.GetElapsedTime(now, _hardDue) : TimeSpan.Zero); } - else if (Volatile.Read(ref _phase) == 1) + else if (phase == SoftReported) { if (now < _hardDue) { TryChange(Stopwatch.GetElapsedTime(now, _hardDue)); return; } - Interlocked.Exchange(ref _phase, 2); + Interlocked.Exchange(ref _phase, HardFired); hard = true; } } // Outside the lock: a caller with a dispatcher resumes synchronously inside TrySetResult. if (soft) _outcome.TrySetResult(SoftTimedOut); + if (transport) _outcome.TrySetResult(TransportTimedOut); if (hard && Volatile.Read(ref _done) == 0) _onHardDeadline?.Invoke(this); } diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/Protos/script_worker.proto b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/Protos/script_worker.proto index 7e918208e..5d73da467 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/Protos/script_worker.proto +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/Protos/script_worker.proto @@ -41,12 +41,17 @@ message Hello { string pe_cache_dir = 8; // the node's compiled-PE cache (design §7); empty = no cache bytes pe_cache_key = 9; // its HMAC key; the worker loads only entries that verify under it uint32 compile_threads = 10; // dedicated compile threads; 0 = the worker's default (min(8, max(1, CPUs/2))) + // lmxopcua#827: the host asks the worker to run each evaluation's SOFT clock itself, from when an evaluation thread + // starts the run, and to report the crossing in EvalResults.soft_expired_eval_ids. The host then keeps only a + // transport deadline of its own (see EvalResults). Honoured only when the worker echoes Ready.worker_soft_clock. + bool worker_soft_clock = 11; } message Ready { string nonce = 1; int32 pid = 2; string runtime_version = 3; + bool worker_soft_clock = 4; // lmxopcua#827: this worker runs the soft clock (Hello.worker_soft_clock was set) } enum ContextKind { @@ -141,10 +146,20 @@ message EvalResult { // when the evaluation appears in `started_eval_ids`, which an evaluation thread sends alongside the result of the run it // just finished. So a run waiting behind a sibling is never timed out, struck or killed for the sibling's time. // Either list may be the only content of a frame. A host that predates these fields ignores them. +// +// Worker soft clock (lmxopcua#827, negotiated by Hello/Ready.worker_soft_clock): the soft deadline T measures the +// script's OWN execution — the worker starts it when an evaluation thread starts the run and reports the crossing in +// `soft_expired_eval_ids` (it does not cancel the run itself: the host answers its caller first, then sends Cancel, +// keeping the in-process ordering). The host arms the hard deadline G from that notice. What the host keeps on its own +// clock is a TRANSPORT deadline, T + G from submit (restarted at a deferred run's start): no result and no soft notice +// by then means the pipe or the worker process is late — the caller is answered as a runner fault ("script worker +// unresponsive", BadResourceUnavailable), never as a script timeout, and nothing is struck. A starved or stalled worker +// therefore cannot make a trivial script look over budget; a hung one is still caught by the heartbeat silence kill. message EvalResults { repeated EvalResult items = 1; repeated uint64 started_eval_ids = 2; repeated uint64 deferred_eval_ids = 3; + repeated uint64 soft_expired_eval_ids = 4; } message ScriptError { diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ScriptWorkerFramePipe.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ScriptWorkerFramePipe.cs index 98a90ba7e..260be8bbf 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ScriptWorkerFramePipe.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ScriptWorkerFramePipe.cs @@ -237,6 +237,7 @@ public sealed class ScriptWorkerFrameWriter results.Items.AddRange(envelope.Results.Items); results.StartedEvalIds.AddRange(envelope.Results.StartedEvalIds); results.DeferredEvalIds.AddRange(envelope.Results.DeferredEvalIds); + results.SoftExpiredEvalIds.AddRange(envelope.Results.SoftExpiredEvalIds); break; default: if (evals is not null) { Append(new Envelope { Eval = evals }); evals = null; } diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/IScriptRunner.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/IScriptRunner.cs index 1a1e79215..77953ac11 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/IScriptRunner.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/IScriptRunner.cs @@ -125,6 +125,12 @@ public sealed record ScriptRunOutcome(ScriptRunStatus Status, object? Value, str /// Reason when the worker could not take more work (the out-of-process runner's back-pressure answer). public const string WorkerBusyReason = "script worker busy"; + /// + /// Reason when the worker neither answered nor reported the script over its budget within T + G of submit — the + /// pipe or a starved / stalled worker process, not the script (lmxopcua#827). + /// + public const string WorkerUnresponsiveReason = "script worker unresponsive"; + /// /// For : completes when the runner can run this script again (its worker /// is back), so the seam re-evaluates it then with its latest inputs (design §9, "re-evaluated on the shard's diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ShadowScriptRunner.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ShadowScriptRunner.cs index 403598d5c..647e76432 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ShadowScriptRunner.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ShadowScriptRunner.cs @@ -47,7 +47,7 @@ public enum ScriptShadowVerdict /// The shadow run disagreed; names how. Mismatch = 1, - /// The shadow runner could not run the script (worker restarted, none ready, worker busy): not compared. + /// The shadow runner could not run the script (worker restarted, none ready, worker busy, worker unresponsive): not compared. ShadowUnavailable = 2, } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker/WorkerHost.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker/WorkerHost.cs index 5b0d99854..4d9250859 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker/WorkerHost.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker/WorkerHost.cs @@ -29,6 +29,9 @@ internal sealed class WorkerHost /// The watchdog fails fast this long after the host's own hard deadline, so the host normally wins. private static readonly TimeSpan WatchdogMargin = TimeSpan.FromMilliseconds(500); + /// The watchdog's longest sleep between passes. + private static readonly TimeSpan WatchdogPoll = TimeSpan.FromMilliseconds(50); + private readonly NamedPipeClientStream _pipe; private readonly string _nonce; private readonly int _parentPid; @@ -75,6 +78,19 @@ internal sealed class WorkerHost private TimeSpan _hardGrace = TimeSpan.FromSeconds(1); private TimeSpan _heartbeatInterval = TimeSpan.FromSeconds(1); private ScriptPeCacheStore? _peCache; + + /// + /// lmxopcua#827: this worker runs each evaluation's soft clock from the run's start and reports the crossing + /// (EvalResults.soft_expired_eval_ids); the host asked for it in Hello and we echoed it in + /// Ready. Off (an older host): the host's own clock, started at submit, is the soft deadline. + /// + private bool _workerSoftClock; + + /// Wakes the watchdog early when a run's soft deadline is sooner than its next planned pass. + private readonly AutoResetEvent _watchdogWake = new(false); + + /// of the watchdog's next planned pass. + private long _watchdogNextPass; private long _seq; public WorkerHost(NamedPipeClientStream pipe, string nonce, int parentPid) @@ -109,6 +125,7 @@ internal sealed class WorkerHost Nonce = _nonce, Pid = Environment.ProcessId, RuntimeVersion = Environment.Version.ToString(), + WorkerSoftClock = _workerSoftClock, }, }); if (ScriptWorkerFrameCodec.UseBlockingIo) ScriptWorkerFrameCodec.Write(_pipe, ready, _maxFrameBytes); @@ -140,6 +157,7 @@ internal sealed class WorkerHost if (hello.HardGraceMs > 0) _hardGrace = TimeSpan.FromMilliseconds(hello.HardGraceMs); if (hello.HeartbeatIntervalMs > 0) _heartbeatInterval = TimeSpan.FromMilliseconds(hello.HeartbeatIntervalMs); if (hello.CompileThreads > 0) _compileThreads = (int)Math.Min(hello.CompileThreads, 64); + _workerSoftClock = hello.WorkerSoftClock; if (!string.IsNullOrEmpty(hello.PeCacheDir) && hello.PeCacheKey.Length == ScriptPeCacheStore.KeyLength) { // Design §7: the host owns the key; it reaches us only here, never through the environment or a file. @@ -425,11 +443,17 @@ internal sealed class WorkerHost var logger = _scriptLogger.ForContext(ForwardingLogSink.KeyProperty, request.Key); var soft = TimeSpan.FromMilliseconds(Math.Max(1, request.SoftTimeoutMs)); - // Cancelled by the host's Cancel frame at the soft deadline, never by our own clock (see the proto). + // Cancelled by the host's Cancel frame at the soft deadline, never by our own clock (see the proto): with the + // worker soft clock (lmxopcua#827) the watchdog only REPORTS the crossing; the host answers its caller, then + // cancels. using var cts = new CancellationTokenSource(); var grace = request.HardGraceMs > 0 ? TimeSpan.FromMilliseconds(request.HardGraceMs) : _hardGrace; - var running = new RunningEval(request.Key, Stopwatch.GetTimestamp(), soft + grace + WatchdogMargin, cts); + var started = Stopwatch.GetTimestamp(); + var running = new RunningEval(request.EvalId, request.Key, started, soft, grace, cts); _running[(long)request.EvalId] = running; + // The watchdog plans its next pass at most WatchdogPoll ahead; a budget shorter than that wakes it now, so the + // soft notice is not late by up to a poll interval. + if (_workerSoftClock && running.SoftDue < Interlocked.Read(ref _watchdogNextPass)) _watchdogWake.Set(); var before = GC.GetAllocatedBytesForCurrentThread(); var sw = Stopwatch.StartNew(); try @@ -465,22 +489,68 @@ internal sealed class WorkerHost return result; } + /// + /// The watchdog: reports soft-deadline crossings (worker soft clock, lmxopcua#827) and fails the process fast for + /// a run past its hard deadline that the host has not killed. + /// + /// + /// The hard deadline it enforces is the host's plus , from the same anchor the host + /// uses: the run's start (T + G) on the host clock, or — with the worker soft clock — the soft notice it sent + /// (G), which is where the host arms its kill. So the host normally wins and the exit names the culprit. + /// private void WatchdogLoop() { while (true) { - Thread.Sleep(50); + var now = Stopwatch.GetTimestamp(); + var nextPass = now + Ticks(WatchdogPoll); + List? expired = null; foreach (var (_, run) in _running) { - if (Stopwatch.GetElapsedTime(run.StartedTimestamp) > run.FailFastAfter) + if (_workerSoftClock) { - // The host should have killed us already; this names the culprit in the exit/dump. - Environment.FailFast($"ScriptWorker: script '{run.Key}' exceeded its hard deadline ({run.FailFastAfter.TotalMilliseconds:F0} ms)"); + if (run.SoftReportedAt == 0) + { + if (now >= run.SoftDue) + { + run.SoftReportedAt = now; + (expired ??= []).Add(run.EvalId); + } + else if (run.SoftDue < nextPass) + { + nextPass = run.SoftDue; + } + } + else if (now - run.SoftReportedAt > Ticks(run.Grace + WatchdogMargin)) + { + FailFast(run, run.Grace + WatchdogMargin, "its soft deadline"); + } + } + else if (now - run.StartedTimestamp > Ticks(run.Soft + run.Grace + WatchdogMargin)) + { + FailFast(run, run.Soft + run.Grace + WatchdogMargin, "its start"); } } + if (expired is not null) + { + var notice = new EvalResults(); + notice.SoftExpiredEvalIds.AddRange(expired); + Send(new Envelope { Results = notice }); + } + Interlocked.Exchange(ref _watchdogNextPass, nextPass); + var wait = Stopwatch.GetElapsedTime(Stopwatch.GetTimestamp(), nextPass); + if (wait > TimeSpan.Zero) _watchdogWake.WaitOne(wait); + } + + static void FailFast(RunningEval run, TimeSpan after, string anchor) + { + // The host should have killed us already; this names the culprit in the exit/dump. + Environment.FailFast($"ScriptWorker: script '{run.Key}' exceeded its hard deadline ({after.TotalMilliseconds:F0} ms after {anchor})"); } } + private static long Ticks(TimeSpan span) => (long)(span.TotalSeconds * Stopwatch.Frequency); + private async Task HeartbeatLoopAsync() { using var timer = new PeriodicTimer(_heartbeatInterval); @@ -609,7 +679,30 @@ internal sealed class WorkerHost Environment.Exit(ScriptWorkerEnvironment.ExitForbiddenAssembly); } - private sealed record RunningEval(string Key, long StartedTimestamp, TimeSpan FailFastAfter, CancellationTokenSource Cancellation); + /// One evaluation on an evaluation thread, as the watchdog sees it. + private sealed class RunningEval(ulong evalId, string key, long startedTimestamp, TimeSpan soft, TimeSpan grace, + CancellationTokenSource cancellation) + { + public ulong EvalId { get; } = evalId; + + public string Key { get; } = key; + + /// when the evaluation thread started the run. + public long StartedTimestamp { get; } = startedTimestamp; + + /// T. + public TimeSpan Soft { get; } = soft; + + /// The run's start + T. + public long SoftDue { get; } = startedTimestamp + Ticks(soft); + + public TimeSpan Grace { get; } = grace; + + public CancellationTokenSource Cancellation { get; } = cancellation; + + /// When the watchdog reported the soft crossing to the host; 0 = not yet. Watchdog thread only. + public long SoftReportedAt { get; set; } + } /// One loaded script: its compiled evaluator and its input slot order. private sealed class LoadedScript : IDisposable diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ShadowScriptRunnerTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ShadowScriptRunnerTests.cs index a0f29ae58..f320c3fd5 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ShadowScriptRunnerTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ShadowScriptRunnerTests.cs @@ -83,6 +83,8 @@ public sealed class ShadowScriptRunnerTests // The runner's fault, not the script's: not compared. { ScriptRunOutcome.Ok(42), ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerRestartedReason), ScriptShadowVerdict.ShadowUnavailable, null }, { ScriptRunOutcome.Ok(42), ScriptRunOutcome.Failed(ScriptRunOutcome.WorkerBusyReason), ScriptShadowVerdict.ShadowUnavailable, null }, + // lmxopcua#827: a stalled / starved worker past the transport deadline is the runner's fault, not a mismatch. + { ScriptRunOutcome.Ok(42), ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerUnresponsiveReason), ScriptShadowVerdict.ShadowUnavailable, null }, }; [Theory] diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs new file mode 100644 index 000000000..359e3f86f --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs @@ -0,0 +1,339 @@ +using System.Diagnostics; +using System.Runtime.InteropServices; +using System.Security.Cryptography; +using System.Text; +using Serilog; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Scripting; +using ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess; +using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.V1; + +namespace ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests; + +/// +/// lmxopcua#827: the soft deadline T measures the script's OWN execution — the worker runs it from the moment an +/// evaluation thread starts the run — so the pipe round trip and the time a starved worker waits to be scheduled +/// no longer count against the script. The host keeps a transport deadline (T + G from submit) and reports passing +/// it as a runner fault (, BadResourceUnavailable), never as a +/// script timeout. The starved worker is made real here: its process is suspended (SIGSTOP / NtSuspendProcess) +/// while a trivial script is submitted. +/// +[Collection(ScriptWorkerCollection.Name)] +[Trait("Category", "ScriptWorker")] +public sealed class WorkerSoftClockTests +{ + private const string Trivial = "return 7;"; + private const string Runaway = "while (true) { }"; + + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + /// + /// A runner whose worker heartbeats every 2 s, so the host's silence kill (3 intervals) stays well clear of the + /// suspensions below: what is measured is the deadline, not the heartbeat monitor. + /// + private static OutOfProcessScriptRunner Runner(TimeSpan hardGrace, ILogger? log = null) + => new(Scripts.Options(hardGrace: hardGrace, quarantineAfterKills: 1), log ?? new LoggerConfiguration().CreateLogger(), null, + heartbeatInterval: TimeSpan.FromSeconds(2), loadTimeout: TimeSpan.FromSeconds(60)); + + private static async Task WarmAsync(OutOfProcessScriptRunner runner, string key) + { + var warm = await runner.RunAsync(Scripts.Vt(key, Trivial, TimeSpan.FromSeconds(30)), Ct); + warm.Status.ShouldBe(ScriptRunStatus.Ok, warm.Reason); + var worker = runner.Shards[0].Current.ShouldNotBeNull(); + worker.WorkerSoftClock.ShouldBeTrue("the worker negotiated the soft clock"); + return worker; + } + + /// + /// The issue's headline: a worker that cannot be scheduled for longer than T, but shorter than the transport + /// deadline, answers a trivial script with its value. Before #827 the host's clock, started at submit, answered + /// script timed out after 0.3s for return 7;. + /// + [Fact] + public async Task A_worker_stalled_for_longer_than_T_still_answers_a_trivial_script_with_its_value() + { + var softT = TimeSpan.FromMilliseconds(300); + await using var runner = Runner(hardGrace: TimeSpan.FromSeconds(3)); + var worker = await WarmAsync(runner, "trivial"); + + Task run; + using (ProcessFreezer.Freeze(worker.Pid)) + { + run = runner.RunAsync(Scripts.Vt("trivial", Trivial, softT), Ct); + await Task.Delay(softT * 4, Ct); // 1.2 s: well past T on the host's clock + run.IsCompleted.ShouldBeFalse("nothing can answer while the worker is suspended"); + } + + var outcome = await run.WaitAsync(TimeSpan.FromSeconds(30), Ct); + outcome.Status.ShouldBe(ScriptRunStatus.Ok, outcome.Reason); + outcome.Value.ShouldBe(7); + runner.Shards[0].Current.ShouldNotBeNull().Pid.ShouldBe(worker.Pid, "a stall is not a deadline kill"); + runner.QuarantinedKeys(ScriptContextKind.VirtualTag).ShouldBeEmpty(); + } + + /// + /// A stall past the transport deadline (T + G) is the runner's fault: the caller gets + /// as (published + /// BadResourceUnavailable) with a recovery that completes when the worker answers; a re-trigger meanwhile is the + /// runner's fault too (not "previous run still running"); the worker is not killed and the key is not struck. + /// Before #827 this was a script timeout at T and a deadline kill at T + G that struck the key — with + /// QuarantineAfterKills = 1, a quarantine for return 7;. + /// + [Fact] + public async Task A_worker_stalled_past_the_transport_deadline_is_a_runner_fault_and_nothing_is_struck() + { + var softT = TimeSpan.FromMilliseconds(200); + var grace = TimeSpan.FromMilliseconds(400); + await using var runner = Runner(grace); + var worker = await WarmAsync(runner, "trivial"); + + ScriptRunOutcome stalled, retriggered; + using (ProcessFreezer.Freeze(worker.Pid)) + { + stalled = await runner.RunAsync(Scripts.Vt("trivial", Trivial, softT), Ct).WaitAsync(TimeSpan.FromSeconds(10), Ct); + retriggered = await runner.RunAsync(Scripts.Vt("trivial", Trivial, softT), Ct).WaitAsync(TimeSpan.FromSeconds(10), Ct); + stalled.Recovery.ShouldNotBeNull().IsCompleted.ShouldBeFalse("the worker has not answered yet"); + } + + stalled.Status.ShouldBe(ScriptRunStatus.Unavailable, stalled.Reason); + stalled.Reason.ShouldBe(ScriptRunOutcome.WorkerUnresponsiveReason); + retriggered.Status.ShouldBe(ScriptRunStatus.Unavailable, retriggered.Reason); + retriggered.Reason.ShouldBe(ScriptRunOutcome.WorkerUnresponsiveReason); + stalled.ToVirtualTagEvalResult().StatusCode.ShouldBe(Commons.OpcUa.OpcUaStatus.BadResourceUnavailable); + + await stalled.Recovery!.WaitAsync(TimeSpan.FromSeconds(30), Ct); + var next = await runner.RunAsync(Scripts.Vt("trivial", Trivial, softT), Ct); + next.Status.ShouldBe(ScriptRunStatus.Ok, next.Reason); + next.Value.ShouldBe(7); + runner.Shards[0].Current.ShouldNotBeNull().Pid.ShouldBe(worker.Pid, "a stall is not a deadline kill"); + runner.QuarantinedKeys(ScriptContextKind.VirtualTag).ShouldBeEmpty(); + } + + /// + /// The other side of the contract is unchanged: a script that really runs past T is answered at T from its own + /// start and killed G after that, even with the worker stalled before it started. Positive control for the two + /// tests above — they would also pass if the runner simply never timed anything out. + /// + [Fact] + public async Task A_runaway_that_starts_late_is_still_timed_out_at_T_from_its_start_and_killed_at_G() + { + var softT = TimeSpan.FromMilliseconds(300); + var grace = TimeSpan.FromMilliseconds(600); + var stall = TimeSpan.FromMilliseconds(500); + await using var runner = Runner(grace); + var worker = await WarmAsync(runner, "warm"); + var load = await runner.RunAsync(Scripts.Vt("runaway", "return 0;", TimeSpan.FromSeconds(30)), Ct); + load.Status.ShouldBe(ScriptRunStatus.Ok, load.Reason); + + var sw = Stopwatch.StartNew(); + Task run; + using (ProcessFreezer.Freeze(worker.Pid)) + { + run = runner.RunAsync(Scripts.Vt("runaway", Runaway, softT), Ct); + await Task.Delay(stall, Ct); + } + var thawed = sw.Elapsed; + var outcome = await run.WaitAsync(TimeSpan.FromSeconds(30), Ct); + var answered = sw.Elapsed; + + outcome.Status.ShouldBe(ScriptRunStatus.Failed); + outcome.Reason.ShouldNotBeNull().ShouldStartWith("script timed out after "); + (answered - thawed).ShouldBeGreaterThanOrEqualTo(softT - TimeSpan.FromMilliseconds(50), "T counts from the run's start, after the stall"); + + var exit = await worker.Exited.WaitAsync(TimeSpan.FromSeconds(30), Ct); + exit.Reason.ShouldBe(WorkerExitReason.KilledDeadline); + exit.CulpritKey.ShouldBe(InProcessScriptRunner.ScopedKey(ScriptContextKind.VirtualTag, "runaway")); + (sw.Elapsed - thawed).ShouldBeGreaterThanOrEqualTo(softT + grace - TimeSpan.FromMilliseconds(50), "G counts from the soft deadline"); + } + + /// + /// Shadow (lmxopcua#780 P2) compares the shadow pool with the published in-process result; a starved worker used to + /// show up there as a script mismatch (published Ok, shadow "timed out") and skew the census for the OutOfProcess + /// flip. Now a stall shorter than the transport deadline is a match, and a longer one is shadow_unavailable + /// — never a mismatch — including the re-trigger that finds the stalled run still holding its slot. + /// + [Fact] + public async Task In_Shadow_a_stalled_worker_is_a_match_or_shadow_unavailable_never_a_mismatch() + { + var softT = TimeSpan.FromMilliseconds(300); + var grace = TimeSpan.FromSeconds(1); + var options = Scripts.Options(hardGrace: grace, quarantineAfterKills: 1); + options.Runner = ScriptRunnerMode.Shadow; + await using var pool = Runner(grace); + using var shadow = new ShadowScriptRunner(pool, options, new LoggerConfiguration().CreateLogger()); + var worker = await WarmAsync(pool, "trivial"); + var published = ScriptRunOutcome.Ok(7); + + // Stall past T, short of T + G: compared, and a match. + using (ProcessFreezer.Freeze(worker.Pid)) + { + shadow.Mirror(Scripts.Vt("trivial", Trivial, softT), published); + await Task.Delay(TimeSpan.FromMilliseconds(900), Ct); + } + (await shadow.WaitIdleAsync(TimeSpan.FromSeconds(30))).ShouldBeTrue(); + shadow.Stats().Matched.ShouldBe(1, Summary(shadow)); + + // Stall past T + G: not compared. The second mirror arrives while the first still holds the key's slot. + using (ProcessFreezer.Freeze(worker.Pid)) + { + shadow.Mirror(Scripts.Vt("trivial", Trivial, softT), published); + await Scripts.Eventually(() => shadow.Stats().ShadowUnavailable == 1, TimeSpan.FromSeconds(10), "the transport deadline"); + shadow.Mirror(Scripts.Vt("trivial", Trivial, softT), published); + await Scripts.Eventually(() => shadow.Stats().ShadowUnavailable == 2, TimeSpan.FromSeconds(10), "the stalled re-trigger"); + } + (await shadow.WaitIdleAsync(TimeSpan.FromSeconds(30))).ShouldBeTrue(); + + var stats = shadow.Stats(); + stats.Mismatched.ShouldBe(0, Summary(shadow)); + stats.ShadowUnavailable.ShouldBe(2, Summary(shadow)); + stats.Matched.ShouldBe(1); + pool.Shards[0].Current.ShouldNotBeNull().Pid.ShouldBe(worker.Pid); + } + + /// + /// Protocol compatibility: the soft clock is negotiated (Hello.worker_soft_clock → Ready.worker_soft_clock). + /// A host that does not ask for it keeps the pre-#827 contract — the soft deadline is the host's, from submit, + /// and the worker sends no soft notices — and a runaway is still answered at T and killed at T + G. + /// + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task The_soft_clock_is_negotiated_and_both_clocks_enforce_T_and_G(bool workerSoftClock) + { + var spec = new WorkerLaunchSpec("t", OutOfProcessScriptRunner.ResolveWorkerPath(new ScriptWorkerOptions { ExecutablePath = Scripts.WorkerUnderTest }), + 256L * 1024 * 1024, EvalThreads: 2, HardGrace: TimeSpan.FromSeconds(1), HeartbeatInterval: TimeSpan.FromSeconds(1), + HandshakeTimeout: TimeSpan.FromSeconds(15), WorkerSoftClock: workerSoftClock); + var connection = await WorkerConnection.StartAsync(spec, new LoggerConfiguration().CreateLogger(), _ => { }, _ => { }, Ct); + await using var _ = connection; + connection.WorkerSoftClock.ShouldBe(workerSoftClock); + + var entry = await connection.EnsureLoadedAsync("k", ContextKind.VirtualTag, Runaway, + SHA256.HashData(Encoding.UTF8.GetBytes(Runaway)), []).WaitAsync(TimeSpan.FromSeconds(60), Ct); + entry.Outcome.ShouldNotBeNull().Ok.ShouldBeTrue(); + + var softT = TimeSpan.FromMilliseconds(300); + var grace = TimeSpan.FromMilliseconds(500); + var sw = Stopwatch.StartNew(); + var pending = connection.Submit("k", + new EvalRequest { Key = "k", SoftTimeoutMs = (uint)softT.TotalMilliseconds, HardGraceMs = (uint)grace.TotalMilliseconds, NowUtcTicks = DateTime.UtcNow.Ticks }, + softT, grace, null); + (await pending.Outcome.WaitAsync(TimeSpan.FromSeconds(10), Ct)).ShouldBeSameAs(PendingEval.SoftTimedOut); + sw.Elapsed.ShouldBeGreaterThanOrEqualTo(softT - TimeSpan.FromMilliseconds(20)); + connection.Cancel(pending); + + var exit = await connection.Exited.WaitAsync(TimeSpan.FromSeconds(10), Ct); + exit.Reason.ShouldBe(WorkerExitReason.KilledDeadline); + sw.Elapsed.ShouldBeGreaterThanOrEqualTo(softT + grace - TimeSpan.FromMilliseconds(50)); + } + + // ---------------------------------------------------------------- PendingEval, no process + + /// The host side of the worker clock, deterministic: no notice means the transport deadline at T + G. + [Fact] + public async Task Worker_clock_with_no_word_from_the_worker_answers_transport_timed_out_at_T_plus_G_and_never_kills() + { + var killed = 0; + var pending = new PendingEval("k", 1); + var sw = Stopwatch.StartNew(); + pending.Arm(TimeSpan.FromMilliseconds(100), TimeSpan.FromMilliseconds(200), _ => Interlocked.Increment(ref killed), workerClock: true); + + (await pending.Outcome.WaitAsync(TimeSpan.FromSeconds(10), Ct)).ShouldBeSameAs(PendingEval.TransportTimedOut); + sw.Elapsed.ShouldBeGreaterThanOrEqualTo(TimeSpan.FromMilliseconds(280)); + await Task.Delay(500, Ct); + killed.ShouldBe(0, "a late worker is not the script's fault"); + + // A soft notice after the transport answer still arms the kill: the script itself ran for T. + pending.SoftExpired(); + await Scripts.Eventually(() => Volatile.Read(ref killed) == 1, TimeSpan.FromSeconds(10), "the kill G after the soft notice"); + } + + /// The soft notice answers the caller and arms the kill G later; a result before then cancels it. + [Fact] + public async Task Worker_clock_soft_notice_answers_soft_timed_out_and_arms_the_kill_G_later() + { + var killed = 0; + var pending = new PendingEval("k", 1); + pending.Arm(TimeSpan.FromSeconds(30), TimeSpan.FromMilliseconds(300), _ => Interlocked.Increment(ref killed), workerClock: true); + pending.Outcome.IsCompleted.ShouldBeFalse(); + + var sw = Stopwatch.StartNew(); + pending.SoftExpired(); + (await pending.Outcome.WaitAsync(TimeSpan.FromSeconds(1), Ct)).ShouldBeSameAs(PendingEval.SoftTimedOut); + await Scripts.Eventually(() => Volatile.Read(ref killed) == 1, TimeSpan.FromSeconds(10), "the kill"); + sw.Elapsed.ShouldBeGreaterThanOrEqualTo(TimeSpan.FromMilliseconds(280)); + + var answered = new PendingEval("k2", 2); + answered.Arm(TimeSpan.FromSeconds(30), TimeSpan.FromMilliseconds(100), _ => Interlocked.Increment(ref killed), workerClock: true); + answered.SoftExpired(); + answered.Complete(new EvalResult { EvalId = 2 }); + await Task.Delay(400, Ct); + killed.ShouldBe(1, "a run that answered inside G is not killed"); + } + + /// Under the host clock a soft notice means nothing (a worker that did not negotiate it cannot send one). + [Fact] + public void Host_clock_ignores_a_soft_notice() + { + var pending = new PendingEval("k", 1); + pending.Arm(TimeSpan.FromSeconds(30), TimeSpan.FromSeconds(30), _ => { }, workerClock: false); + pending.SoftExpired(); + pending.Outcome.IsCompleted.ShouldBeFalse(); + pending.Complete(null); + } + + private static string Summary(ShadowScriptRunner shadow) + => $"{shadow.Stats()}; " + string.Join("; ", shadow.Mismatches.Select(m => $"{m.Key}: {m.LastReason} — {m.LastSummary}")); +} + +/// Suspends a process (SIGSTOP, or NtSuspendProcess on Windows) until disposed. +internal sealed class ProcessFreezer : IDisposable +{ + private readonly int _pid; + private int _thawed; + + private ProcessFreezer(int pid) + { + _pid = pid; + Signal(pid, stop: true); + } + + public static ProcessFreezer Freeze(int pid) => new(pid); + + public void Dispose() + { + if (Interlocked.Exchange(ref _thawed, 1) == 0) Signal(_pid, stop: false); + } + + private static void Signal(int pid, bool stop) + { + if (OperatingSystem.IsWindows()) + { + using var process = Process.GetProcessById(pid); + var status = stop ? NtSuspendProcess(process.Handle) : NtResumeProcess(process.Handle); + status.ShouldBe(0, $"Nt{(stop ? "Suspend" : "Resume")}Process"); + return; + } + // libc kill(2), not the kill(1) command: a child process's exit is observed through the same SIGCHLD + // handling that the suspended worker (another child of this process) just disturbed, and WaitForExit was + // seen to take seconds. + var signal = stop ? SigStop : SigCont; + Kill(pid, signal).ShouldBe(0, $"kill({pid}, {signal}): errno {Marshal.GetLastPInvokeError()}"); + } + + /// SIGSTOP / SIGCONT: 17 / 19 on macOS (BSD), 19 / 18 on Linux. + private static int SigStop => OperatingSystem.IsMacOS() ? 17 : 19; + + private static int SigCont => OperatingSystem.IsMacOS() ? 19 : 18; + +#pragma warning disable SYSLIB1054 // DllImport kept: LibraryImport would need AllowUnsafeBlocks for three test-only calls. + [DllImport("libc", EntryPoint = "kill", SetLastError = true)] + private static extern int Kill(int pid, int signal); + + [DllImport("ntdll.dll")] + private static extern int NtSuspendProcess(IntPtr processHandle); + + [DllImport("ntdll.dll")] + private static extern int NtResumeProcess(IntPtr processHandle); +#pragma warning restore SYSLIB1054 +}