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 48aa3f588..723eb0a42 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 @@ -910,7 +910,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) 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 911f20ee8..6a2b922ac 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, @@ -155,6 +161,12 @@ 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; @@ -235,6 +247,32 @@ internal sealed class WorkerConnection : IAsyncDisposable 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; @@ -592,6 +630,7 @@ 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, WorkerSoftClock); _pending[request.EvalId] = pending; @@ -617,6 +656,7 @@ 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]; @@ -800,6 +840,8 @@ 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. @@ -915,6 +957,17 @@ internal sealed class WorkerConnection : IAsyncDisposable 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) { } 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/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs index 7a35adfd4..f0da183bc 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/WorkerSoftClockTests.cs @@ -306,6 +306,65 @@ public sealed class WorkerSoftClockTests 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.