fix(scripting): soft notice resumes the caller on the dispatcher; a late notice still cancels and counts (#827 review F1, F2)

F1: PendingEval.SoftExpired runs on the pipe reader and completed the caller's outcome inline (production outcomes have no async continuations, by design, because results resume on the ResultDispatcher). The caller's code (the runner's Cancel write, the seam's publish) therefore ran on the reader, which could hold up heartbeats (a false silence kill) and, with blocking I/O, write to the pipe from the reader. The soft outcome is now posted to the dispatcher like a result. Test: a 1.5 s continuation on a soft timeout with a 300 ms silence limit resumes on ScriptWorker[t].Results and the worker lives (before: it resumed on the Reader). F2: after a transport timeout ("worker unresponsive"), a late soft notice armed G but nobody sent Cancel, so a polling script could not end itself and the whole worker was killed at notice + G with a kill strike. PendingEval now exposes LateSoftExpiry (asynchronous continuations, cancelled when the run answers); the runner sends the Cancel and counts the timeout (NoteTimeout) from it. A genuinely slow script whose notice is delayed past T + G is thus still a timeout. Test: stall past T + G, thaw, a polling script overruns T: the worker survives G and the key is quarantined for timeouts, not a hard deadline (before: the worker was killed).
2026-11-11 02:21:45 -05:00
parent 30c6c3a0bf
commit c5dffbd8b8
3 changed files with 122 additions and 7 deletions
@@ -591,10 +591,22 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
            {
                // lmxopcua#827: T + G passed with neither a result nor a soft-deadline notice. The worker runs the soft
                // clock from the run's start, so this is the pipe or a starved / stalled worker — the runner's fault,
-                // never the script's: no timeout strike, no Cancel (the run may not even have started), and the slot
-                // stays with the run until the worker answers. The seam re-evaluates when it does (or when the worker
-                // is replaced, if the heartbeat monitor kills it).
-                TransportTimeoutCounter.Add(1, new KeyValuePair<string, object?>("kind", KindTag(invocation.Kind)));
+                // not (yet) the script's: no timeout counted, no Cancel (the run may not even have started), and the
+                // slot stays with the run until the worker answers. The seam re-evaluates when it does (or when the
+                // worker is replaced, if the heartbeat monitor kills it).
+                TransportTimeoutCounter.Add(1, new KeyValuePair<string, object?>("kind", KindTag(invocation.Kind)),
+                    new KeyValuePair<string, object?>("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);
@@ -1128,6 +1128,7 @@ internal sealed class PendingEval
    private bool _started;
    private int _phase;
    private int _done;
+    private TaskCompletionSource? _lateSoftExpiry;

    public PendingEval(string key, ulong evalId, Action? onFinished = null, ResultDispatcher? dispatcher = null)
    {
@@ -1160,6 +1161,16 @@ internal sealed class PendingEval
    /// <summary>The worker's result, or null when the worker ended first.</summary>
    public Task<EvalResult?> Completion => _completion.Task;

+    /// <summary>
+    ///     After <see cref="TransportTimedOut"/> (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.
+    /// </summary>
+    internal Task? LateSoftExpiry
+    {
+        get { lock (_clock) return _lateSoftExpiry?.Task; }
+    }
+
    /// <summary>
    ///     What the caller waits on: the worker's result, null when the worker ended first, <see cref="SoftTimedOut"/> at
    ///     the soft deadline, or <see cref="TransportTimedOut"/> at the transport deadline (the run itself is still
@@ -1246,7 +1257,7 @@ internal sealed class PendingEval
    /// </summary>
    internal void SoftExpired()
    {
-        bool signal;
+        bool signal, late;
        lock (_clock)
        {
            if (_timer is null || !_workerClock || Volatile.Read(ref _done) != 0) return;
@@ -1255,12 +1266,23 @@ internal sealed class PendingEval
            _started = true;
            _deferred = false;
            signal = phase == Waiting;
+            late = phase == TransportReported;
            Interlocked.Exchange(ref _phase, SoftReported);
            _hardDue = Stopwatch.GetTimestamp() + Ticks(_grace);
            TryChange(_grace);
        }
-        // Outside the lock: a caller with a dispatcher resumes synchronously inside TrySetResult.
-        if (signal) _outcome.TrySetResult(SoftTimedOut);
+        // This runs on the pipe READER. The caller's resumption is the caller's code (the seam's publish, the runner's
+        // Cancel write): it goes through the dispatcher exactly like a result (ResultDispatcher remarks), never inline
+        // here, or a slow consumer would hold up heartbeats and results (#827 review F1).
+        if (signal)
+        {
+            if (_dispatcher is { } dispatcher) dispatcher.Post(this, SoftTimedOut);
+            else _outcome.TrySetResult(SoftTimedOut);
+        }
+        // The caller was already answered "worker unresponsive"; the script has now itself run for T. Its watcher
+        // (the runner) cancels the run's token and counts the timeout, off this thread: the completion source runs its
+        // continuations asynchronously (#827 review F2).
+        if (late) _lateSoftExpiry?.TrySetResult();
    }

    private void OnTimer()
@@ -1277,6 +1299,7 @@ internal sealed class PendingEval
                // A tick armed before a deferral/start can land early: re-arm for what is left.
                if (now < _transportDue) { TryChange(Stopwatch.GetElapsedTime(now, _transportDue)); return; }
                Interlocked.Exchange(ref _phase, TransportReported);
+                _lateSoftExpiry = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
                transport = true;
            }
            else if (phase == Waiting)
@@ -1316,6 +1339,7 @@ internal sealed class PendingEval
    {
        if (Interlocked.Exchange(ref _done, 1) != 0) return;
        _timer?.Dispose();
+        lock (_clock) _lateSoftExpiry?.TrySetCanceled();
        try { _onFinished?.Invoke(); }
        finally
        {
@@ -227,6 +227,85 @@ public sealed class WorkerSoftClockTests
        sw.Elapsed.ShouldBeGreaterThanOrEqualTo(softT + grace - TimeSpan.FromMilliseconds(50));
    }

+    // ---------------------------------------------------------------- review findings (#827)
+
+    /// <summary>
+    ///     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).
+    /// </summary>
+    [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();
+    }
+
+    /// <summary>
+    ///     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).
+    /// </summary>
+    [Fact]
+    public async Task A_slow_script_whose_notice_arrives_after_the_transport_deadline_is_cancelled_and_counted_not_killed()
+    {
+        var softT = TimeSpan.FromMilliseconds(200);
+        var grace = TimeSpan.FromMilliseconds(400);
+        var options = Scripts.Options(hardGrace: grace, quarantineAfterKills: 1);
+        options.QuarantineAfterConsecutiveTimeouts = 1;
+        await using var runner = new OutOfProcessScriptRunner(options, new LoggerConfiguration().CreateLogger(), null,
+            heartbeatInterval: TimeSpan.FromSeconds(2), loadTimeout: TimeSpan.FromSeconds(60));
+        const string polite = "if ((int)ctx.GetTag(\"spin\").Value == 1) { while (!ctx.CancellationToken.IsCancellationRequested) { } } return 1;";
+        (await runner.RunAsync(Scripts.Vt("polite", polite, TimeSpan.FromSeconds(30), Scripts.Input("spin", 0)), Ct)).Value.ShouldBe(1);
+        var worker = runner.Shards[0].Current.ShouldNotBeNull();
+
+        ScriptRunOutcome stalled;
+        using (ProcessFreezer.Freeze(worker.Pid))
+        {
+            stalled = await runner.RunAsync(Scripts.Vt("polite", polite, softT, Scripts.Input("spin", 1)), Ct).WaitAsync(TimeSpan.FromSeconds(10), Ct);
+        }
+        stalled.Reason.ShouldBe(ScriptRunOutcome.WorkerUnresponsiveReason);
+
+        // Thawed: the script starts, spins past T, the late notice cancels it and it returns. The worker survives G.
+        await stalled.Recovery.ShouldNotBeNull().WaitAsync(TimeSpan.FromSeconds(10), Ct);
+        await Task.Delay(softT + grace + TimeSpan.FromMilliseconds(500), Ct);
+        worker.IsAlive.ShouldBeTrue("a cancelled polling script must not cost its worker");
+        runner.Shards[0].Current.ShouldNotBeNull().Pid.ShouldBe(worker.Pid);
+
+        // …and it counted as the timeout it was (threshold 1), not as a kill.
+        await Scripts.Eventually(() => runner.QuarantinedKeys(ScriptContextKind.VirtualTag).Count == 1, TimeSpan.FromSeconds(5), "the timeout quarantine");
+        var refused = await runner.RunAsync(Scripts.Vt("polite", polite, softT, Scripts.Input("spin", 0)), Ct);
+        refused.Reason.ShouldNotBeNull().ShouldContain("timed out");
+        refused.Reason.ShouldNotContain("hard deadline");
+    }
+
    // ---------------------------------------------------------------- PendingEval, no process

    /// <summary>The host side of the worker clock, deterministic: no notice means the transport deadline at T + G.</summary>