diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs index 1d7556180..48aa3f588 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs @@ -591,10 +591,22 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, { // lmxopcua#827: T + G passed with neither a result nor a soft-deadline notice. The worker runs the soft // clock from the run's start, so this is the pipe or a starved / stalled worker — the runner's fault, - // never the script's: no timeout strike, no Cancel (the run may not even have started), and the slot - // stays with the run until the worker answers. The seam re-evaluates when it does (or when the worker - // is replaced, if the heartbeat monitor kills it). - TransportTimeoutCounter.Add(1, new KeyValuePair("kind", KindTag(invocation.Kind))); + // not (yet) the script's: no timeout counted, no Cancel (the run may not even have started), and the + // slot stays with the run until the worker answers. The seam re-evaluates when it does (or when the + // worker is replaced, if the heartbeat monitor kills it). + TransportTimeoutCounter.Add(1, new KeyValuePair("kind", KindTag(invocation.Kind)), + new KeyValuePair("answer", "stalled")); + // …unless the worker then reports the script itself over T (slow AND late — a notice delayed past + // T + G by the watchdog poll and the pipe lands here too): it still gets its Cancel, so a polling + // script ends itself instead of costing the whole worker at notice + G, and it counts as a timeout + // (#827 review F2). Off the reader: the completion source runs this asynchronously. + var timeout = invocation.Timeout; + pending.LateSoftExpiry?.ContinueWith(_ => + { + connection.Cancel(pending); + NoteTimeout(key, state, timeout); + }, CancellationToken.None, TaskContinuationOptions.OnlyOnRanToCompletion | TaskContinuationOptions.ExecuteSynchronously, + TaskScheduler.Default); // Read by a refusal only while it is incomplete, so a write that loses a race with the slot release // is harmless. Volatile.Write(ref state.StalledRun, pending.Completion); diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs index 228068a9d..911f20ee8 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs @@ -1128,6 +1128,7 @@ internal sealed class PendingEval private bool _started; private int _phase; private int _done; + private TaskCompletionSource? _lateSoftExpiry; public PendingEval(string key, ulong evalId, Action? onFinished = null, ResultDispatcher? dispatcher = null) { @@ -1160,6 +1161,16 @@ internal sealed class PendingEval /// The worker's result, or null when the worker ended first. public Task Completion => _completion.Task; + /// + /// After (null before): completes when the worker then reports the script itself + /// over T (slow AND late: it still gets its Cancel and counts as a timeout, #827 review F2), and is cancelled + /// when the worker answers or ends first. Continuations run asynchronously, never on the reader. + /// + internal Task? LateSoftExpiry + { + get { lock (_clock) return _lateSoftExpiry?.Task; } + } + /// /// What the caller waits on: the worker's result, null when the worker ended first, at /// the soft deadline, or at the transport deadline (the run itself is still @@ -1246,7 +1257,7 @@ internal sealed class PendingEval /// internal void SoftExpired() { - bool signal; + bool signal, late; lock (_clock) { if (_timer is null || !_workerClock || Volatile.Read(ref _done) != 0) return; @@ -1255,12 +1266,23 @@ internal sealed class PendingEval _started = true; _deferred = false; signal = phase == Waiting; + late = phase == TransportReported; Interlocked.Exchange(ref _phase, SoftReported); _hardDue = Stopwatch.GetTimestamp() + Ticks(_grace); TryChange(_grace); } - // Outside the lock: a caller with a dispatcher resumes synchronously inside TrySetResult. - if (signal) _outcome.TrySetResult(SoftTimedOut); + // This runs on the pipe READER. The caller's resumption is the caller's code (the seam's publish, the runner's + // Cancel write): it goes through the dispatcher exactly like a result (ResultDispatcher remarks), never inline + // here, or a slow consumer would hold up heartbeats and results (#827 review F1). + if (signal) + { + if (_dispatcher is { } dispatcher) dispatcher.Post(this, SoftTimedOut); + else _outcome.TrySetResult(SoftTimedOut); + } + // The caller was already answered "worker unresponsive"; the script has now itself run for T. Its watcher + // (the runner) cancels the run's token and counts the timeout, off this thread: the completion source runs its + // continuations asynchronously (#827 review F2). + if (late) _lateSoftExpiry?.TrySetResult(); } private void OnTimer() @@ -1277,6 +1299,7 @@ internal sealed class PendingEval // A tick armed before a deferral/start can land early: re-arm for what is left. if (now < _transportDue) { TryChange(Stopwatch.GetElapsedTime(now, _transportDue)); return; } Interlocked.Exchange(ref _phase, TransportReported); + _lateSoftExpiry = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); transport = true; } else if (phase == Waiting) @@ -1316,6 +1339,7 @@ internal sealed class PendingEval { if (Interlocked.Exchange(ref _done, 1) != 0) return; _timer?.Dispose(); + lock (_clock) _lateSoftExpiry?.TrySetCanceled(); try { _onFinished?.Invoke(); } finally { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs index 359e3f86f..7a35adfd4 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs @@ -227,6 +227,85 @@ public sealed class WorkerSoftClockTests sw.Elapsed.ShouldBeGreaterThanOrEqualTo(softT + grace - TimeSpan.FromMilliseconds(50)); } + // ---------------------------------------------------------------- review findings (#827) + + /// + /// Review F1: the soft notice is handled on the pipe READER, and the caller's resumption is the caller's code. It + /// must go through the result dispatcher like a result does: a continuation that blocks for 1.5 s must not hold up + /// the reader, or the next heartbeats sit unread and a healthy worker is killed for silence (300 ms here). Before + /// the fix the continuation ran inline on the reader and the worker was killed (KilledHeartbeat). + /// + [Fact] + public async Task A_slow_continuation_of_a_soft_timeout_does_not_stall_the_reader() + { + var heartbeat = TimeSpan.FromMilliseconds(100); + var spec = new WorkerLaunchSpec("t", OutOfProcessScriptRunner.ResolveWorkerPath(new ScriptWorkerOptions { ExecutablePath = Scripts.WorkerUnderTest }), + 256L * 1024 * 1024, EvalThreads: 2, HardGrace: TimeSpan.FromSeconds(60), HeartbeatInterval: heartbeat, + HandshakeTimeout: TimeSpan.FromSeconds(15), HeartbeatSilenceLimit: TimeSpan.FromSeconds(30)); + var connection = await WorkerConnection.StartAsync(spec, new LoggerConfiguration().CreateLogger(), _ => { }, _ => { }, Ct); + await using var _ = connection; + var entry = await connection.EnsureLoadedAsync("k", ContextKind.VirtualTag, Runaway, + SHA256.HashData(Encoding.UTF8.GetBytes(Runaway)), []).WaitAsync(TimeSpan.FromSeconds(60), Ct); + entry.Outcome.ShouldNotBeNull().Ok.ShouldBeTrue(); + connection.SilenceLimit = heartbeat * 3; // armed only once the cold start is over (see SupervisionTests) + + var softT = TimeSpan.FromMilliseconds(200); + var pending = connection.Submit("k", + new EvalRequest { Key = "k", SoftTimeoutMs = (uint)softT.TotalMilliseconds, HardGraceMs = 60_000, NowUtcTicks = DateTime.UtcNow.Ticks }, + softT, TimeSpan.FromSeconds(60), null); + string? resumedOn = null; + var slowConsumer = pending.Outcome.ContinueWith(t => + { + resumedOn = Thread.CurrentThread.Name; + t.Result.ShouldBeSameAs(PendingEval.SoftTimedOut); + Thread.Sleep(1_500); + }, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); + + await slowConsumer.WaitAsync(TimeSpan.FromSeconds(10), Ct); + resumedOn.ShouldBe("ScriptWorker[t].Results", "the caller resumes on the dispatcher, never on the reader"); + connection.Exited.IsCompleted.ShouldBeFalse("the reader kept reading heartbeats while the consumer was slow"); + connection.IsAlive.ShouldBeTrue(); + } + + /// + /// Review F2: a stall past T + G answers "worker unresponsive"; when the worker then runs the script and it really + /// overruns T, the late soft notice must still cancel its token — a polling script ends itself instead of the whole + /// worker being killed G later — and the run counts as a timeout (QuarantineAfterConsecutiveTimeouts = 1 here). + /// Before the fix nobody sent the Cancel: the worker was killed at notice + G (KilledDeadline, a kill strike). + /// + [Fact] + public async Task A_slow_script_whose_notice_arrives_after_the_transport_deadline_is_cancelled_and_counted_not_killed() + { + var softT = TimeSpan.FromMilliseconds(200); + var grace = TimeSpan.FromMilliseconds(400); + var options = Scripts.Options(hardGrace: grace, quarantineAfterKills: 1); + options.QuarantineAfterConsecutiveTimeouts = 1; + await using var runner = new OutOfProcessScriptRunner(options, new LoggerConfiguration().CreateLogger(), null, + heartbeatInterval: TimeSpan.FromSeconds(2), loadTimeout: TimeSpan.FromSeconds(60)); + const string polite = "if ((int)ctx.GetTag(\"spin\").Value == 1) { while (!ctx.CancellationToken.IsCancellationRequested) { } } return 1;"; + (await runner.RunAsync(Scripts.Vt("polite", polite, TimeSpan.FromSeconds(30), Scripts.Input("spin", 0)), Ct)).Value.ShouldBe(1); + var worker = runner.Shards[0].Current.ShouldNotBeNull(); + + ScriptRunOutcome stalled; + using (ProcessFreezer.Freeze(worker.Pid)) + { + stalled = await runner.RunAsync(Scripts.Vt("polite", polite, softT, Scripts.Input("spin", 1)), Ct).WaitAsync(TimeSpan.FromSeconds(10), Ct); + } + stalled.Reason.ShouldBe(ScriptRunOutcome.WorkerUnresponsiveReason); + + // Thawed: the script starts, spins past T, the late notice cancels it and it returns. The worker survives G. + await stalled.Recovery.ShouldNotBeNull().WaitAsync(TimeSpan.FromSeconds(10), Ct); + await Task.Delay(softT + grace + TimeSpan.FromMilliseconds(500), Ct); + worker.IsAlive.ShouldBeTrue("a cancelled polling script must not cost its worker"); + runner.Shards[0].Current.ShouldNotBeNull().Pid.ShouldBe(worker.Pid); + + // …and it counted as the timeout it was (threshold 1), not as a kill. + await Scripts.Eventually(() => runner.QuarantinedKeys(ScriptContextKind.VirtualTag).Count == 1, TimeSpan.FromSeconds(5), "the timeout quarantine"); + var refused = await runner.RunAsync(Scripts.Vt("polite", polite, softT, Scripts.Input("spin", 0)), Ct); + refused.Reason.ShouldNotBeNull().ShouldContain("timed out"); + refused.Reason.ShouldNotContain("hard deadline"); + } + // ---------------------------------------------------------------- PendingEval, no process /// The host side of the worker clock, deterministic: no notice means the transport deadline at T + G.