fix(scripting): #789 review — reader faults end the worker, write stalls kill it, Cancel off the worker reader

- Host reader: the frame read and its handling are caught separately. Only the read's I/O exceptions mean the pipe ended; anything HandleFrame's inline completion work throws now kills that worker as ProtocolFault (outstanding runs -> worker restarted / BadResourceUnavailable) instead of escaping a dedicated thread (process crash) or being swallowed as a clean end (an InvalidOperationException used to be, leaving the worker unread until a Crash 2 s later). - Worker reader: the same catch-all around HandleFrame (exit ProtocolFault). - Worker: Cancel() runs on the pool, so a script's blocking ctx.CancellationToken callback cannot stop the worker reading its pipe. - Host write-stall check: ScriptWorkerFrameWriter.WriteInProgressFor; the heartbeat monitor kills a worker whose pipe write has been blocked longer than max(10 s, silence limit) as KilledUnresponsive, which fails the write and releases the producer (seam submit, deploy load, Cancel). Fixes the wrong 'heartbeat monitor ends it' comment on the pipe buffer quota. - StopAsync sends Shutdown off the caller so its 1 s budget holds on a full pipe. - WindowsBlockingPipeStream refuses the async surface (Stream's base versions serialise a write behind a waiting read). - Tests: PipeFaultTests (throwing completion callback -> ProtocolFault and a BadResourceUnavailable sibling, shard serves again; a worker that stops reading but heartbeats is killed within the bound and the blocked producer returns, via the Hello.test_stall_reader_on_diag seam); BlockingPipeStreamTests dispose with a write blocked on a full pipe, and with a read and a write both pending; Category=ScriptWorker on the protocol, handshake and pipe test classes. Mutation checks: no stall kill -> stall test red; catch-all without the fault -> callback test red (Crash). Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
2026-11-11 20:26:02 -05:00
parent d4a2aa6142
commit 03ac57d5d6
10 changed files with 345 additions and 19 deletions
@@ -91,6 +91,10 @@ internal sealed record BatchSubmission(string Key, EvalRequest Request, TimeSpan
 ///     How long the worker may send nothing before it is killed (<see cref="WorkerExitReason.KilledHeartbeat"/>); null =
 ///     three <paramref name="HeartbeatInterval"/>s. Test seam (with <see cref="WorkerConnection.SilenceLimit"/>).
 /// </param>
+/// <param name="TestStallReaderOnDiag">
+///     Test seam: the worker's reader stops reading for good at the first diagnostics request while its heartbeats go on —
+///     a worker that keeps talking but stopped listening (the write-stall check, lmxopcua#789).
+/// </param>
 /// <param name="WorkerSoftClock">
 ///     Ask the worker to run the soft clock from each run's start (lmxopcua#827; see <see cref="WorkerConnection.WorkerSoftClock"/>).
 ///     False (tests only) keeps the pre-#827 host clock, which is also what a worker that does not echo it gets.
@@ -110,7 +114,8 @@ internal sealed record WorkerLaunchSpec(
    int CompileThreads = 0,
    string? WorkerSid = null,
    bool WorkerSoftClock = true,
-    TimeSpan? HeartbeatSilenceLimit = null)
+    TimeSpan? HeartbeatSilenceLimit = null,
+    bool TestStallReaderOnDiag = false)
 {
    /// <summary>The default compile timeout.</summary>
    public static readonly TimeSpan DefaultLoadTimeout = TimeSpan.FromSeconds(60);
@@ -265,6 +270,20 @@ internal sealed class WorkerConnection : IAsyncDisposable
    /// </summary>
    internal TimeSpan? ProgressCeilingOverride { get; set; }

+    /// <summary>
+    ///     How long one pipe write may stay blocked before the worker is killed (<see cref="WorkerExitReason.KilledUnresponsive"/>,
+    ///     lmxopcua#789). Writes are blocking, so a full pipe — a worker whose reader stopped while its heartbeats go on —
+    ///     would otherwise hold whichever producer is draining the writer (a seam submitting, a deploy loading scripts, a
+    ///     Cancel) forever: the silence check never fires, and with only loads queued nothing else does. A worker that is
+    ///     reading drains the 1 MiB buffer far faster than this. Null = the larger of 10 s and the silence limit; set = an
+    ///     override (tests).
+    /// </summary>
+    internal TimeSpan? WriteStallLimitOverride { get; set; }
+
+    /// <summary>The write-stall bound now (see <see cref="WriteStallLimitOverride"/>).</summary>
+    internal TimeSpan WriteStallLimit
+        => WriteStallLimitOverride ?? (SilenceLimit > TimeSpan.FromSeconds(10) ? SilenceLimit : TimeSpan.FromSeconds(10));
+
    /// <summary>Test seam: ignore the worker's soft-deadline notices, as a worker whose watchdog never sends them would.</summary>
    internal bool DropSoftNoticesForTest { get; set; }

@@ -380,6 +399,7 @@ internal sealed class WorkerConnection : IAsyncDisposable
                MaxLogLines = 1000,
                CompileThreads = (uint)Math.Max(0, spec.CompileThreads),
                WorkerSoftClock = spec.WorkerSoftClock,
+                TestStallReaderOnDiag = spec.TestStallReaderOnDiag,
            };
            if (spec.PeCacheDirectory is { } cacheDir && spec.PeCacheKey is { } cacheKey)
            {
@@ -755,7 +775,9 @@ internal sealed class WorkerConnection : IAsyncDisposable
        if (!_exited.Task.IsCompleted)
        {
            lock (_pending) { _killedFor ??= (reason, null); }
-            Send(new Envelope { Shutdown = new Shutdown() });
+            // Off this thread: the send may drain the writer, and a write blocks on a full pipe — the 1 s budget below
+            // must hold regardless (the kill then fails that write).
+            _ = Task.Run(() => Send(new Envelope { Shutdown = new Shutdown() }));
            try
            {
                await _process.WaitForExitAsync().WaitAsync(TimeSpan.FromSeconds(1)).ConfigureAwait(false);
@@ -797,17 +819,44 @@ internal sealed class WorkerConnection : IAsyncDisposable
    {
        WorkerExitReason? fault = null;
        var reader = new ScriptWorkerFrameReader(_pipe);
-        try
-        {
-            while (reader.Read() is { } frame) HandleFrame(frame);
-        }
-        catch (ScriptWorkerProtocolException ex)
-        {
-            fault = ProtocolFault(ex);
-        }
-        catch (Exception ex) when (ex is IOException or ObjectDisposedException or InvalidOperationException
-                                       or System.Net.Sockets.SocketException)
+        while (true)
        {
+            Envelope? frame;
+            try
+            {
+                frame = reader.Read();
+            }
+            catch (ScriptWorkerProtocolException ex)
+            {
+                fault = ProtocolFault(ex);
+                break;
+            }
+            catch (Exception ex) when (ex is IOException or ObjectDisposedException or InvalidOperationException
+                                           or System.Net.Sockets.SocketException)
+            {
+                break; // the pipe ended (these come from the read only, never from the frame's handling below)
+            }
+            if (frame is null) break;
+
+            try
+            {
+                HandleFrame(frame);
+            }
+            catch (ScriptWorkerProtocolException ex)
+            {
+                fault = ProtocolFault(ex);
+                break;
+            }
+            catch (Exception ex)
+            {
+                // HandleFrame runs completion work inline (slot releases, load bookkeeping). On this dedicated thread an
+                // exception that escaped would take the whole server process down, and treating it as the pipe ending
+                // would leave the worker running unread; so it ends this worker as a fault: every outstanding evaluation
+                // completes as "worker restarted" (BadResourceUnavailable) and the shard restarts.
+                _log.Error(ex, "ScriptWorker[{Shard}] failed handling a {Frame} frame; killing pid {Pid}", Shard, frame.BodyCase, Pid);
+                fault = WorkerExitReason.ProtocolFault;
+                break;
+            }
        }
        _ = PipeEndedAsync(fault);
    }
@@ -941,6 +990,15 @@ internal sealed class WorkerConnection : IAsyncDisposable
                    Kill(WorkerExitReason.KilledHeartbeat, null);
                    return;
                }
+                if (_writer.WriteInProgressFor is { } blocked && blocked > WriteStallLimit)
+                {
+                    // It talks but does not listen: the pipe is full and a producer is blocked in the write. The kill
+                    // fails that write and releases it (lmxopcua#789).
+                    _log.Warning("ScriptWorker[{Shard}] pid {Pid} has not read from its pipe for {Blocked} ms (a write is blocked on a full pipe) while still heartbeating; killing it",
+                        Shard, Pid, (long)blocked.TotalMilliseconds);
+                    Kill(WorkerExitReason.KilledUnresponsive, null);
+                    return;
+                }
                if (!_pending.IsEmpty && ProgressCeiling() is { } ceiling)
                {
                    var stuck = Stopwatch.GetElapsedTime(Volatile.Read(ref _lastProgressTimestamp));