From 30c6c3a0bfa8e9daba67167401aae2e3b04c4b54 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 01:59:11 -0500 Subject: [PATCH] test(scriptworker): tight-heartbeat tests warm the worker before arming the 3-interval silence kill (#827 audit) A_burst_of_cold_compiles_does_not_starve_the_heartbeat (50 ms heartbeat) and A_caller_that_blocks_after_its_result_does_not_starve_the_reader (100 ms) launched their worker under the tight silence kill, so its cold start (launch, first compile, Roslyn JIT) had to stay within 150/300 ms of silence. Under 28 busy processes on 14 cores that failed 2 in 10 runs in the WARM-UP, killed for ~165 ms of silence, and replacement workers were killed the same way. New test seam: WorkerLaunchSpec.HeartbeatSilenceLimit (runner internal ctor heartbeatSilenceLimit) and WorkerConnection.SilenceLimit. The tests launch with a 30 s limit, warm, then tighten to three intervals for the phase they are about; the assertions on that phase are unchanged. --- .../OutOfProcessScriptRunner.cs | 9 +++-- .../WorkerConnection.cs | 29 ++++++++++++--- .../SupervisionTests.cs | 35 ++++++++++++++----- 3 files changed, 59 insertions(+), 14 deletions(-) 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 06b748b4d..1d7556180 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 @@ -141,6 +141,9 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, private readonly string? _peCacheDirectory; private readonly byte[]? _peCacheKey; private readonly int _compileThreads; + + /// Test seam: the launched workers' initial silence limit (null = three heartbeat intervals). + private readonly TimeSpan? _heartbeatSilenceLimit; private long _peCacheHits; private long _peCacheMisses; private long _peCacheRejected; @@ -159,9 +162,10 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, internal OutOfProcessScriptRunner(ScriptingOptions options, ILogger? log, ILogger? scriptRoot, TimeSpan heartbeatInterval, TimeSpan loadTimeout, TimeSpan? maintenanceInterval = null, string? markerDirectory = null, - int compileThreads = 0, int evalThreads = 0) + int compileThreads = 0, int evalThreads = 0, TimeSpan? heartbeatSilenceLimit = null) { _compileThreads = compileThreads; + _heartbeatSilenceLimit = heartbeatSilenceLimit; ArgumentNullException.ThrowIfNull(options); _options = options.Sanitized(out _); _worker = _options.Worker; @@ -719,7 +723,8 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, MarkerDirectory: _markerDirectory, LoadTimeout: LoadTimeout, CompileThreads: _compileThreads, - WorkerSid: WorkerSid); + WorkerSid: WorkerSid, + HeartbeatSilenceLimit: _heartbeatSilenceLimit); return new WorkerShard(label, isolation, (onExiting, ct) => WorkerConnection.StartAsync(spec, _log, OnLogs, onExiting, ct), OnWorkerExit, _log); 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 a9fce4fa4..228068a9d 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. /// +/// +/// How long the worker may send nothing before it is killed (); null = +/// three s. Test seam (with ). +/// /// /// 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. @@ -99,7 +103,8 @@ internal sealed record WorkerLaunchSpec( TimeSpan? LoadTimeout = null, int CompileThreads = 0, string? WorkerSid = null, - bool WorkerSoftClock = true) + bool WorkerSoftClock = true, + TimeSpan? HeartbeatSilenceLimit = null) { /// The default compile timeout. public static readonly TimeSpan DefaultLoadTimeout = TimeSpan.FromSeconds(60); @@ -156,7 +161,7 @@ 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, bool workerSoftClock) + Action onExiting, string? markerPath, bool workerSoftClock, TimeSpan? heartbeatSilenceLimit) { WorkerSoftClock = workerSoftClock; _stdio = stdio; @@ -171,6 +176,7 @@ internal sealed class WorkerConnection : IAsyncDisposable _log = log; _onLogs = onLogs; _heartbeatInterval = heartbeatInterval; + _silenceLimitTicks = (heartbeatSilenceLimit ?? heartbeatInterval * 3).Ticks; Pid = process.Id; // The handshake used seq 1. A failed write means the pipe is gone: the worker is dead or dying. _writer = new ScriptWorkerFrameWriter(pipe, OnWriteFault, ScriptWorkerFrameCodec.DefaultMaxFrameBytes, firstSeq: 1, @@ -215,6 +221,20 @@ internal sealed class WorkerConnection : IAsyncDisposable set => _loadTimeout = value; } + /// + /// How long the worker may send nothing before it is killed for silence: three heartbeat intervals unless the + /// launch spec said otherwise. Test hook: a test with a tight heartbeat launches and warms the worker (its cold + /// start JITs Roslyn, which on a loaded box can pause it for longer than three short intervals) under a generous + /// limit, then tightens it for the phase it is about. + /// + internal TimeSpan SilenceLimit + { + get => TimeSpan.FromTicks(Volatile.Read(ref _silenceLimitTicks)); + set => Volatile.Write(ref _silenceLimitTicks, value.Ticks); + } + + private long _silenceLimitTicks; + /// Compiles currently outstanding on this worker. public int LoadingCount => _loads.Count; @@ -327,7 +347,8 @@ internal sealed class WorkerConnection : IAsyncDisposable : 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, workerSoftClock: spec.WorkerSoftClock && ready.WorkerSoftClock); + spec.EffectiveLoadTimeout, onExiting, markerPath, workerSoftClock: spec.WorkerSoftClock && ready.WorkerSoftClock, + heartbeatSilenceLimit: spec.HeartbeatSilenceLimit); connection.Begin(); return connection; } @@ -888,7 +909,7 @@ internal sealed class WorkerConnection : IAsyncDisposable { if (_exited.Task.IsCompleted) return; var silent = Stopwatch.GetElapsedTime(Volatile.Read(ref _lastFrameTimestamp)); - if (silent > _heartbeatInterval * 3) + if (silent > SilenceLimit) { _log.Warning("ScriptWorker[{Shard}] pid {Pid} sent nothing for {Silent} ms; killing it", Shard, Pid, (long)silent.TotalMilliseconds); Kill(WorkerExitReason.KilledHeartbeat, null); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/SupervisionTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/SupervisionTests.cs index 5f075fcfa..df214c4b6 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/SupervisionTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/SupervisionTests.cs @@ -31,6 +31,27 @@ public sealed class SupervisionTests private static CancellationToken Ct => TestContext.Current.CancellationToken; + /// + /// A runner whose workers heartbeat every but start with a generous silence limit: + /// the cold start (launch, first compile, Roslyn's JIT) is not what the tight-heartbeat tests are about, and on a + /// loaded box it can pause the worker for longer than three short intervals (lmxopcua#827 audit: 2 in 10 runs + /// under 28 busy processes on 14 cores were killed for 160 ms of silence during the warm-up compile, and the + /// replacement workers were too). then tightens the limit to three intervals for the + /// phase under test. + /// + private static OutOfProcessScriptRunner TightHeartbeatRunner(TimeSpan heartbeat, TimeSpan loadTimeout, CapturingSink sink) + => new(Scripts.Options(), sink.CreateLogger(), null, heartbeatInterval: heartbeat, loadTimeout: loadTimeout, + heartbeatSilenceLimit: TimeSpan.FromSeconds(30)); + + /// Warms the shard's worker with , then arms the three-interval silence kill. + private static WorkerConnection WarmTight(OutOfProcessScriptRunner runner, ScriptRunOutcome warm, TimeSpan heartbeat, CapturingSink sink) + { + warm.Value.ShouldBe(7, $"{warm.Status} {warm.Reason}\n{string.Join(Environment.NewLine, sink.Rendered)}"); + var worker = runner.Shards[0].Current.ShouldNotBeNull(); + worker.SilenceLimit = heartbeat * 3; + return worker; + } + private static string VtKey(string key) => InProcessScriptRunner.ScopedKey(ScriptContextKind.VirtualTag, key); // ---------------------------------------------------------------- the quarantine race @@ -112,10 +133,9 @@ public sealed class SupervisionTests public async Task A_burst_of_cold_compiles_does_not_starve_the_heartbeat() { var sink = new CapturingSink(); - await using var runner = new OutOfProcessScriptRunner(Scripts.Options(), sink.CreateLogger(), null, - heartbeatInterval: TimeSpan.FromMilliseconds(50), loadTimeout: TimeSpan.FromSeconds(120)); - (await runner.RunAsync(Scripts.Vt("first", Healthy), Ct)).Value.ShouldBe(7); - var worker = runner.Shards[0].Current.ShouldNotBeNull(); + var heartbeat = TimeSpan.FromMilliseconds(50); + await using var runner = TightHeartbeatRunner(heartbeat, TimeSpan.FromSeconds(120), sink); + var worker = WarmTight(runner, await runner.RunAsync(Scripts.Vt("first", Healthy), Ct), heartbeat, sink); const int count = 1500; var outcomes = await Task.WhenAll(Enumerable.Range(0, count).Select(i => runner.RunAsync( @@ -514,10 +534,9 @@ public sealed class SupervisionTests public async Task A_caller_that_blocks_after_its_result_does_not_starve_the_reader() { var sink = new CapturingSink(); - await using var runner = new OutOfProcessScriptRunner(Scripts.Options(), sink.CreateLogger(), null, - heartbeatInterval: TimeSpan.FromMilliseconds(100), loadTimeout: TimeSpan.FromSeconds(60)); - (await runner.RunAsync(Scripts.Vt("warm", Healthy), Ct)).Value.ShouldBe(7); - var worker = runner.Shards[0].Current.ShouldNotBeNull(); + var heartbeat = TimeSpan.FromMilliseconds(100); + await using var runner = TightHeartbeatRunner(heartbeat, TimeSpan.FromSeconds(60), sink); + var worker = WarmTight(runner, await runner.RunAsync(Scripts.Vt("warm", Healthy), Ct), heartbeat, sink); string? resumedOn = null; await runner.RunAsync(Scripts.Vt("warm", Healthy), Ct).ContinueWith(t =>