Joseph Doherty
4cd648b029
Reproduced 10/10 under 28 CPU burners (load 100-250): every run killed for 157-309 ms of silence with every compile on its dedicated thread - the OS scheduler, not the pool. A 1 s limit still failed 1/10 at load ~300 (1,012 ms). The limit is now production's (3 x the default 1 s heartbeat): 10/10 at load 360-430. The failure message now carries the burst length, the limit and the kill line. Also corrects the doc: its mutation check (compile on the pool again => killed for silence) no longer holds - re-checked on a 14-core Mac, the mutated worker passed at both 150 ms and 1 s, because the host's silence clock counts any frame and since #789 each compile writes its own LoadStarted/LoadResult frames inline. Co-Authored-By: Claude <noreply@anthropic.com>
742 lines
44 KiB
C#
742 lines
44 KiB
C#
using System.Collections.Concurrent;
|
||
using System.Diagnostics;
|
||
using System.Diagnostics.Metrics;
|
||
using System.Text;
|
||
using System.Text.Json;
|
||
using Serilog;
|
||
using Shouldly;
|
||
using Xunit;
|
||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
|
||
using ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess;
|
||
using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol;
|
||
using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.V1;
|
||
|
||
namespace ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests;
|
||
|
||
/// <summary>
|
||
/// lmxopcua#780 P1 supervision gaps: the quarantine race, victim re-evaluation on a shard's return, the idle
|
||
/// isolation worker, the startup orphan sweep, culprit-finding for a compile that crashes its worker, and the
|
||
/// runner state / crash-loop circuit. Real worker processes throughout.
|
||
/// </summary>
|
||
[Collection(ScriptWorkerCollection.Name)]
|
||
[Trait("Category", "ScriptWorker")]
|
||
public sealed class SupervisionTests
|
||
{
|
||
private const string Healthy = "return 7;";
|
||
private const string Runaway = "while (true) { }";
|
||
private const string Spin = "var ms = (int)ctx.GetTag(\"ms\").Value; var end = ctx.Now.AddMilliseconds(ms); while (ctx.Now < end) { } return ms;";
|
||
|
||
private static IReadOnlyDictionary<string, DataValueSnapshot> Ms(int ms) => Scripts.Input("ms", ms);
|
||
|
||
private static CancellationToken Ct => TestContext.Current.CancellationToken;
|
||
|
||
/// <summary>
|
||
/// A runner whose workers heartbeat every <paramref name="heartbeat"/> 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). <see cref="WarmTight"/> then tightens the limit to three intervals for the
|
||
/// phase under test.
|
||
/// </summary>
|
||
private static OutOfProcessScriptRunner TightHeartbeatRunner(TimeSpan heartbeat, TimeSpan loadTimeout, CapturingSink sink)
|
||
=> new(Scripts.Options(), sink.CreateLogger(), null, heartbeatInterval: heartbeat, loadTimeout: loadTimeout,
|
||
heartbeatSilenceLimit: TimeSpan.FromSeconds(30));
|
||
|
||
/// <summary>Warms the shard's worker with <paramref name="key"/>, then arms the three-interval silence kill.</summary>
|
||
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
|
||
|
||
/// <summary>
|
||
/// The ordering invariant behind the race fix, pinned deterministically at the connection: the exit attribution
|
||
/// callback runs BEFORE any outstanding evaluation or compile is completed and before <c>Exited</c> is signalled.
|
||
/// Mutation check: moving the attribution call after the completion loop in <c>WorkerConnection.Finish</c> turns
|
||
/// both assertions red.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task Exit_attribution_runs_before_any_outstanding_evaluation_completes()
|
||
{
|
||
var spec = new WorkerLaunchSpec("t", OutOfProcessScriptRunner.ResolveWorkerPath(new ScriptWorkerOptions()),
|
||
256L * 1024 * 1024, EvalThreads: 2, HardGrace: TimeSpan.FromSeconds(60), HeartbeatInterval: TimeSpan.FromSeconds(1),
|
||
HandshakeTimeout: TimeSpan.FromSeconds(15));
|
||
PendingEval? pending = null;
|
||
WorkerConnection? connection = null;
|
||
bool? pendingCompletedAtAttribution = null, exitedCompletedAtAttribution = null;
|
||
IReadOnlyList<string>? inFlightAtAttribution = null;
|
||
connection = await WorkerConnection.StartAsync(spec, new LoggerConfiguration().CreateLogger(), _ => { }, exit =>
|
||
{
|
||
pendingCompletedAtAttribution = pending?.Completion.IsCompleted;
|
||
exitedCompletedAtAttribution = connection?.Exited.IsCompleted;
|
||
inFlightAtAttribution = exit.InFlightKeys;
|
||
}, Ct);
|
||
await using var _ = connection;
|
||
|
||
var entry = await connection.EnsureLoadedAsync("k", ContextKind.VirtualTag, Runaway,
|
||
System.Security.Cryptography.SHA256.HashData(Encoding.UTF8.GetBytes(Runaway)), []).WaitAsync(TimeSpan.FromSeconds(60), Ct);
|
||
entry!.Outcome!.Ok.ShouldBeTrue();
|
||
pending = connection.Submit("k", new EvalRequest { Key = "k", SoftTimeoutMs = 60_000, HardGraceMs = 60_000, NowUtcTicks = DateTime.UtcNow.Ticks });
|
||
await Scripts.Eventually(() => connection.LastHeartbeat is { Inflight: 1 }, TimeSpan.FromSeconds(10), "the run in flight");
|
||
|
||
Process.GetProcessById(connection.Pid).Kill();
|
||
await connection.Exited.WaitAsync(TimeSpan.FromSeconds(10), Ct);
|
||
(await pending.Completion.WaitAsync(TimeSpan.FromSeconds(5), Ct)).ShouldBeNull("the worker went away");
|
||
|
||
inFlightAtAttribution.ShouldBe(["k"]);
|
||
pendingCompletedAtAttribution.ShouldBe(false, "attribution must precede the evaluation's completion");
|
||
exitedCompletedAtAttribution.ShouldBe(false, "attribution must precede Exited");
|
||
}
|
||
|
||
/// <summary>
|
||
/// The other half of the race fix, pinned without timing: the shard hands the connection an attribution callback
|
||
/// that reaches the runner's exit handler SYNCHRONOUSLY (the pre-fix shard called it from an Exited continuation,
|
||
/// after the evaluations had already been completed). Mutation check: launching with a no-op callback and
|
||
/// attributing from the Exited continuation instead leaves <c>attributed</c> false when the callback returns.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task The_shard_attributes_an_exit_synchronously_inside_the_connections_callback()
|
||
{
|
||
Action<WorkerExit>? captured = null;
|
||
var attributed = false;
|
||
await using var shard = new WorkerShard("t", isolation: false,
|
||
(onExiting, _) =>
|
||
{
|
||
captured = onExiting;
|
||
throw new InvalidOperationException("no process in this test");
|
||
},
|
||
(_, _) => attributed = true,
|
||
new LoggerConfiguration().CreateLogger());
|
||
(await shard.GetReadyAsync(TimeSpan.FromSeconds(5), Ct)).ShouldBeNull();
|
||
|
||
captured.ShouldNotBeNull();
|
||
captured(new WorkerExit(WorkerExitReason.Crash, null, 1, ["k"]));
|
||
attributed.ShouldBeTrue("the runner's verdict is recorded before the callback returns");
|
||
}
|
||
|
||
/// <summary>
|
||
/// A burst of cold compiles must not starve the worker's heartbeat. Found on windev by the §8 perf harness: 500
|
||
/// loads at once, each compiled synchronously on its own thread-pool work item, queued the heartbeat timer and
|
||
/// the pipe writer behind them; the host saw no frame for 3 s, killed the shard (KilledHeartbeat) and made all 500
|
||
/// keys suspects. Here a 50 ms heartbeat (kill after 3 s of silence, production's limit) and 1,500 distinct cold scripts at once:
|
||
/// every script answers and the shard's worker is the one that started.
|
||
/// <para>⚠️ The old "mutation check: compile on the thread pool again and the worker is killed for silence" does
|
||
/// NOT hold any more (re-checked for lmxopcua#843 on a 14-core Mac: the mutated worker passed at both the 150 ms
|
||
/// and the 1 s limit). The host's silence clock counts ANY frame, and since lmxopcua#789 a compile writes its own
|
||
/// <c>LoadStarted</c>/<c>LoadResult</c> frames inline — so pool-queued compiles keep the pipe busy themselves (the 3 s silence
|
||
/// windev saw presumably also needed the pre-#789 pool-scheduled writer stuck behind them). What this still pins is the
|
||
/// behaviour: a cold burst under a tight heartbeat is answered without a kill.</para>
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task A_burst_of_cold_compiles_does_not_starve_the_heartbeat()
|
||
{
|
||
var sink = new CapturingSink();
|
||
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);
|
||
// lmxopcua#843: three 50 ms intervals (150 ms) is a limit the OS scheduler alone breaks on an overloaded box — at
|
||
// load average 100–280 the worker or the host's reader waited 150–310 ms for a core (10/10 runs killed for silence
|
||
// under 28 CPU burners, every compile on its dedicated thread), and at ~300 even a 1 s limit was hit once in 10
|
||
// (1,012 ms). The limit is now PRODUCTION's (3 × the default 1 s heartbeat): the test kills only where a real node
|
||
// would, and the windev defect (no frame for 3 s, the whole burst) is still the size it catches. The 50 ms
|
||
// heartbeat keeps frames flowing for the silence clock.
|
||
worker.SilenceLimit = TimeSpan.FromSeconds(3);
|
||
|
||
const int count = 1500;
|
||
var burst = Stopwatch.StartNew();
|
||
var outcomes = await Task.WhenAll(Enumerable.Range(0, count).Select(i => runner.RunAsync(
|
||
Scripts.Vt("burst-" + i, $"return (int)ctx.GetTag(\"x\").Value * {i} + 1;", TimeSpan.FromSeconds(60),
|
||
Scripts.Input("x", 2)), Ct)));
|
||
burst.Stop();
|
||
|
||
// The kill line ("sent nothing for N ms") against the burst's length is the diagnosis: N close to the burst is
|
||
// the pool-starvation defect, N just over the limit is a starved box.
|
||
var log = $"burst took {burst.ElapsedMilliseconds} ms, silence limit {worker.SilenceLimit.TotalMilliseconds} ms\n"
|
||
+ string.Join(Environment.NewLine, sink.Rendered
|
||
.Where(l => l.Contains("ScriptWorker", StringComparison.Ordinal))
|
||
.Where(l => l.Contains("sent nothing", StringComparison.Ordinal) || l.Contains("kill", StringComparison.OrdinalIgnoreCase)
|
||
|| l.Contains("exit", StringComparison.OrdinalIgnoreCase))
|
||
.Take(10));
|
||
for (var i = 0; i < count; i++) outcomes[i].Value.ShouldBe(2 * i + 1, $"{outcomes[i].Status} {outcomes[i].Reason}\n{log}");
|
||
runner.Shards[0].Current.ShouldNotBeNull().Pid.ShouldBe(worker.Pid, log);
|
||
worker.Exited.IsCompleted.ShouldBeFalse(log);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Single flight must not refuse a caller's NEXT run because of the run it already awaited: the slot is freed
|
||
/// before the awaiting caller resumes. The §8 perf harness found the slot freed by a continuation unordered
|
||
/// against the caller's await, so 32 of 200 back-to-back runs of one key came back "previous evaluation still
|
||
/// running" — which the VirtualTag and Calculation seams publish as Bad. Since lmxopcua#789 the release runs in
|
||
/// the connection, on the reader, before the caller's resumption is handed to the result dispatcher; the second
|
||
/// phase makes every release slow (the test hook sleeps inside it) so the ordering, not luck, is what keeps the
|
||
/// caller from being refused. Mutation check: release after handing off the resumption and the second phase goes
|
||
/// red.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task A_caller_that_awaited_its_run_is_never_refused_by_it()
|
||
{
|
||
await using var runner = Scripts.Runner(Scripts.Options());
|
||
var refused = new List<string>();
|
||
for (var i = 0; i < 1_000; i++)
|
||
{
|
||
var outcome = await runner.RunAsync(
|
||
Scripts.Vt("seq", "return (int)ctx.GetTag(\"x\").Value + 7;", inputs: Scripts.Input("x", i)), Ct);
|
||
if (outcome.Status != ScriptRunStatus.Ok || !Equals(outcome.Value, i + 7)) refused.Add($"#{i}: {outcome.Reason}");
|
||
}
|
||
refused.ShouldBeEmpty();
|
||
|
||
runner.SlotReleasingForTest = _ => Thread.Sleep(5);
|
||
for (var i = 0; i < 50; i++)
|
||
{
|
||
var outcome = await runner.RunAsync(
|
||
Scripts.Vt("seq", "return (int)ctx.GetTag(\"x\").Value + 7;", inputs: Scripts.Input("x", i)), Ct);
|
||
if (outcome.Status != ScriptRunStatus.Ok || !Equals(outcome.Value, i + 7)) refused.Add($"slow release #{i}: {outcome.Reason}");
|
||
}
|
||
refused.ShouldBeEmpty();
|
||
}
|
||
|
||
/// <summary>
|
||
/// The race itself, end to end: with QuarantineAfterKills = 1, the moment the killed script's single-flight slot is
|
||
/// released the quarantine is already in place — so no trigger can re-run it in the restart gap, and the next
|
||
/// trigger is refused without any wait. Repeated, since the old ordering lost this race only some of the time.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task A_killed_script_is_quarantined_before_its_slot_is_released()
|
||
{
|
||
await using var runner = Scripts.Runner(Scripts.Options(hardGrace: TimeSpan.FromMilliseconds(250), quarantineAfterKills: 1));
|
||
var observed = new ConcurrentQueue<bool>();
|
||
runner.SlotReleasingForTest = key =>
|
||
{
|
||
if (key == VtKey("runaway")) observed.Enqueue(runner.QuarantinedKeys(ScriptContextKind.VirtualTag).Contains("runaway"));
|
||
};
|
||
(await runner.RunAsync(Scripts.Vt("sibling", Healthy), Ct)).Value.ShouldBe(7);
|
||
|
||
for (var round = 1; round <= 3; round++)
|
||
{
|
||
var worker = runner.Shards[0].Current.ShouldNotBeNull();
|
||
(await runner.RunAsync(Scripts.Vt("runaway", Runaway, TimeSpan.FromMilliseconds(250)), Ct)).Reason!
|
||
.ShouldStartWith("script timed out after ");
|
||
await worker.Exited.WaitAsync(TimeSpan.FromSeconds(10), Ct);
|
||
await Scripts.Eventually(() => observed.Count == round, TimeSpan.FromSeconds(5), "the slot release");
|
||
|
||
// No Eventually: the very next trigger after the kill is refused.
|
||
var next = await runner.RunAsync(Scripts.Vt("runaway", Runaway, TimeSpan.FromMilliseconds(250)), Ct);
|
||
next.Reason!.ShouldStartWith(ScriptRunGuard.QuarantinedReasonPrefix);
|
||
runner.Reset(ScriptContextKind.VirtualTag);
|
||
(await runner.RunAsync(Scripts.Vt("sibling", Healthy), Ct)).Value.ShouldBe(7);
|
||
}
|
||
observed.ShouldAllBe(quarantined => quarantined, "the slot was released before the kill strike was recorded");
|
||
}
|
||
|
||
// ---------------------------------------------------------------- victims re-evaluated on return
|
||
|
||
/// <summary>
|
||
/// A victim's Unavailable outcome carries a Recovery task that completes when its shard is back — and the shard
|
||
/// comes back on its own (proactive restart), with no further trigger. Mutation check: without the shard's
|
||
/// recovery loop, Recovery never completes and no worker is running.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task A_victims_recovery_completes_when_its_shard_restarts_by_itself()
|
||
{
|
||
await using var runner = Scripts.Runner(Scripts.Options());
|
||
(await runner.RunAsync(Scripts.Vt("victim", Spin, inputs: Ms(0)), Ct)).Value.ShouldBe(0);
|
||
(await runner.RunAsync(Scripts.Vt("sibling", Healthy), Ct)).Value.ShouldBe(7);
|
||
var worker = runner.Shards[0].Current.ShouldNotBeNull();
|
||
|
||
var inFlight = runner.RunAsync(Scripts.Vt("sibling", Spin, TimeSpan.FromSeconds(20), Ms(5000)), Ct);
|
||
await Scripts.Eventually(() => worker.InFlightCount == 1, TimeSpan.FromSeconds(10), "the run in flight");
|
||
Process.GetProcessById(worker.Pid).Kill();
|
||
|
||
var victim = await inFlight.WaitAsync(TimeSpan.FromSeconds(10), Ct);
|
||
victim.Status.ShouldBe(ScriptRunStatus.Unavailable);
|
||
_ = victim.Recovery.ShouldNotBeNull("a runner failure tells the seam when to re-evaluate");
|
||
victim.ToVirtualTagEvalResult().RetryWhen.ShouldBeSameAs(victim.Recovery);
|
||
|
||
// The crash made "sibling" a suspect, so its recovery is the isolation worker's; the crashed shard restarts by
|
||
// itself as well, so the scripts placed on it find it warm.
|
||
await victim.Recovery!.WaitAsync(TimeSpan.FromSeconds(15), Ct);
|
||
runner.Isolation.Current.ShouldNotBeNull("recovered without any new trigger");
|
||
await Scripts.Eventually(() => runner.Shards[0].Current, TimeSpan.FromSeconds(15), "the shard back");
|
||
runner.Shards[0].Current!.Pid.ShouldNotBe(worker.Pid);
|
||
(await runner.RunAsync(Scripts.Vt("sibling", Spin, inputs: Ms(0)), Ct)).Value.ShouldBe(0);
|
||
}
|
||
|
||
// ---------------------------------------------------------------- the idle isolation worker
|
||
|
||
/// <summary>
|
||
/// The isolation worker stops after IsolationIdleTimeout with nothing to run — an Idle exit, not a failure — and
|
||
/// starts again for the next suspect. Mutation check: with the maintenance pass not stopping it, it stays up.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task An_idle_isolation_worker_is_stopped_and_restarts_on_demand()
|
||
{
|
||
await using var runner = Scripts.Runner(Scripts.Options(isolationIdle: TimeSpan.FromSeconds(1)),
|
||
maintenanceInterval: TimeSpan.FromMilliseconds(200));
|
||
(await runner.RunAsync(Scripts.Vt("spin", Spin, inputs: Ms(0)), Ct)).Value.ShouldBe(0);
|
||
var worker = runner.Shards[0].Current.ShouldNotBeNull();
|
||
var run = runner.RunAsync(Scripts.Vt("spin", Spin, TimeSpan.FromSeconds(20), Ms(5000)), Ct);
|
||
await Scripts.Eventually(() => worker.InFlightCount == 1, TimeSpan.FromSeconds(10), "the run in flight");
|
||
Process.GetProcessById(worker.Pid).Kill();
|
||
(await run.WaitAsync(TimeSpan.FromSeconds(10), Ct)).Status.ShouldBe(ScriptRunStatus.Unavailable);
|
||
runner.SuspectKeys.ShouldBe([VtKey("spin")]);
|
||
|
||
(await runner.RunAsync(Scripts.Vt("spin", Spin, inputs: Ms(0)), Ct)).Value.ShouldBe(0);
|
||
var iso = runner.Isolation.Current.ShouldNotBeNull();
|
||
|
||
var exit = await iso.Exited.WaitAsync(TimeSpan.FromSeconds(10), Ct);
|
||
exit.Reason.ShouldBe(WorkerExitReason.Idle);
|
||
runner.Isolation.Current.ShouldBeNull();
|
||
runner.Isolation.ConsecutiveFailures.ShouldBe(0, "an idle stop is not a failure");
|
||
runner.Isolation.DownFor.ShouldBeNull("an idle stop does not make the isolation worker 'down'");
|
||
await Scripts.Eventually(() => runner.State == ScriptRunnerState.Ready, TimeSpan.FromSeconds(15), "the runner ready again");
|
||
|
||
(await runner.RunAsync(Scripts.Vt("spin", Spin, inputs: Ms(0)), Ct)).Value.ShouldBe(0);
|
||
runner.Isolation.Current.ShouldNotBeNull().Pid.ShouldNotBe(iso.Pid, "a fresh isolation worker for the next suspect run");
|
||
}
|
||
|
||
// ---------------------------------------------------------------- the orphan sweep
|
||
|
||
private static Process StartDecoyWorker()
|
||
{
|
||
// A real ScriptWorker that will sit waiting (up to 10 s) for a pipe that does not exist: the shape of a worker
|
||
// its host abandoned.
|
||
var psi = new ProcessStartInfo(WorkerConnection.ResolveMuxer())
|
||
{
|
||
UseShellExecute = false,
|
||
CreateNoWindow = true,
|
||
RedirectStandardError = true,
|
||
RedirectStandardOutput = true,
|
||
};
|
||
psi.ArgumentList.Add("exec");
|
||
psi.ArgumentList.Add(OutOfProcessScriptRunner.ResolveWorkerPath(new ScriptWorkerOptions()));
|
||
psi.Environment[ScriptWorkerEnvironment.PipeName] = "otopcua-sw-d" + Guid.NewGuid().ToString("N")[..8];
|
||
psi.Environment[ScriptWorkerEnvironment.Nonce] = "decoy";
|
||
var p = Process.Start(psi)!;
|
||
Thread.Sleep(300);
|
||
if (p.HasExited) throw new InvalidOperationException($"the decoy worker exited {p.ExitCode}: {p.StandardError.ReadToEnd()}");
|
||
return p;
|
||
}
|
||
|
||
private static (int Pid, long StartTicks) DeadProcess()
|
||
{
|
||
var psi = new ProcessStartInfo(WorkerConnection.ResolveMuxer(), "--version")
|
||
{
|
||
UseShellExecute = false, CreateNoWindow = true, RedirectStandardOutput = true,
|
||
};
|
||
using var p = Process.Start(psi)!;
|
||
var start = p.StartTime.ToUniversalTime().Ticks;
|
||
var pid = p.Id;
|
||
p.WaitForExit();
|
||
return (pid, start);
|
||
}
|
||
|
||
private static string WriteMarker(string dir, Process worker, int hostPid, long hostStartTicks, long? workerStartTicks = null)
|
||
{
|
||
Directory.CreateDirectory(dir);
|
||
var record = new WorkerLaunchMarker.MarkerRecord(worker.Id, workerStartTicks ?? worker.StartTime.ToUniversalTime().Ticks,
|
||
worker.ProcessName, hostPid, hostStartTicks, "decoy");
|
||
var path = Path.Combine(dir, worker.Id + ".worker");
|
||
File.WriteAllText(path, JsonSerializer.Serialize(record));
|
||
return path;
|
||
}
|
||
|
||
/// <summary>
|
||
/// Design §3.3: a worker left behind by a DEAD host of this instance is killed at the next start — through the
|
||
/// runner's constructor, so the wiring is pinned too. Mutation check: without the Sweep call the decoy survives.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task The_startup_sweep_kills_a_worker_whose_host_is_gone()
|
||
{
|
||
using var dir = new Scripts.TempDir();
|
||
using var decoy = StartDecoyWorker();
|
||
var (deadPid, deadStart) = DeadProcess();
|
||
var marker = WriteMarker(dir.Path, decoy, deadPid, deadStart);
|
||
|
||
await using var runner = Scripts.Runner(Scripts.Options(), markerDirectory: dir.Path);
|
||
(await Task.Run(() => decoy.WaitForExit(5000), Ct)).ShouldBeTrue("the orphan was killed");
|
||
File.Exists(marker).ShouldBeFalse();
|
||
}
|
||
|
||
/// <summary>
|
||
/// Never an unrelated process: a marker whose host is ALIVE (another instance, or a sibling runner in this
|
||
/// process) is left alone, and a marker whose pid now belongs to a different process (start time differs —
|
||
/// pid reuse) never kills it. Mutation checks: dropping either the host-alive check or the start-time check
|
||
/// kills the decoy.
|
||
/// </summary>
|
||
[Fact]
|
||
public void The_sweep_never_kills_a_live_hosts_worker_or_a_reused_pid()
|
||
{
|
||
using var dir = new Scripts.TempDir();
|
||
using var decoy = StartDecoyWorker();
|
||
using var self = Process.GetCurrentProcess();
|
||
try
|
||
{
|
||
var liveHostMarker = WriteMarker(dir.Path, decoy, self.Id, self.StartTime.ToUniversalTime().Ticks);
|
||
WorkerLaunchMarker.Sweep(dir.Path, new LoggerConfiguration().CreateLogger()).ShouldBeEmpty();
|
||
decoy.HasExited.ShouldBeFalse("a live host's worker is not an orphan");
|
||
File.Exists(liveHostMarker).ShouldBeTrue("the live host still owns its marker");
|
||
|
||
var (deadPid, deadStart) = DeadProcess();
|
||
var reused = WriteMarker(dir.Path, decoy, deadPid, deadStart,
|
||
workerStartTicks: decoy.StartTime.ToUniversalTime().AddHours(-1).Ticks);
|
||
WorkerLaunchMarker.Sweep(dir.Path, new LoggerConfiguration().CreateLogger()).ShouldBeEmpty();
|
||
decoy.HasExited.ShouldBeFalse("a pid whose start time does not match is some other process");
|
||
File.Exists(reused).ShouldBeFalse("a stale marker is cleaned up");
|
||
}
|
||
finally
|
||
{
|
||
try { decoy.Kill(); } catch { /* already gone */ }
|
||
}
|
||
}
|
||
|
||
/// <summary>A worker's marker exists while it runs and is removed when it stops cleanly.</summary>
|
||
[Fact]
|
||
public async Task A_running_worker_has_a_launch_marker_that_its_clean_stop_removes()
|
||
{
|
||
using var dir = new Scripts.TempDir();
|
||
var runner = Scripts.Runner(Scripts.Options(), markerDirectory: dir.Path);
|
||
(await runner.RunAsync(Scripts.Vt("vt", Healthy), Ct)).Value.ShouldBe(7);
|
||
var pid = runner.Shards[0].Current!.Pid;
|
||
var marker = Path.Combine(dir.Path, pid + ".worker");
|
||
File.Exists(marker).ShouldBeTrue();
|
||
JsonSerializer.Deserialize<WorkerLaunchMarker.MarkerRecord>(File.ReadAllText(marker))!.HostPid.ShouldBe(Environment.ProcessId);
|
||
await runner.DisposeAsync();
|
||
File.Exists(marker).ShouldBeFalse();
|
||
}
|
||
|
||
// ---------------------------------------------------------------- a compile that crashes its worker
|
||
|
||
/// <summary>
|
||
/// Design §5.3.5 culprit-finding: a worker that dies while a script is COMPILING (no evaluation in flight) makes
|
||
/// that script a suspect; it then compiles alone in the isolation worker, where a second death is unambiguous and
|
||
/// quarantines it — so a poison source cannot crash-loop its shard forever. Mutation check: leaving the compiling
|
||
/// keys out of the attribution leaves the key unsuspected and unquarantined.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task A_script_whose_compile_keeps_crashing_its_worker_is_found_in_isolation_and_quarantined()
|
||
{
|
||
await using var runner = Scripts.Runner(Scripts.Options(quarantineAfterKills: 1));
|
||
var body = new StringBuilder("long x = 0;\n");
|
||
for (var i = 0; i < 3000; i++) body.Append("x = x * 31 + ").Append(i).Append(";\n");
|
||
body.Append("return x;");
|
||
var poison = body.ToString();
|
||
|
||
async Task<ScriptRunOutcome> CrashDuringCompile(Func<WorkerConnection?> worker)
|
||
{
|
||
var run = runner.RunAsync(Scripts.Vt("poison", poison), Ct);
|
||
var live = await Scripts.Eventually(worker, TimeSpan.FromSeconds(15), "a live worker");
|
||
await Scripts.Eventually(() => live.LoadingCount == 1, TimeSpan.FromSeconds(15), "the compile in flight");
|
||
live.InFlightCount.ShouldBe(0, "no evaluation: only the compile is in flight");
|
||
Process.GetProcessById(live.Pid).Kill();
|
||
return await run.WaitAsync(TimeSpan.FromSeconds(15), Ct);
|
||
}
|
||
|
||
var first = await CrashDuringCompile(() => runner.Shards[0].Current);
|
||
first.Status.ShouldBe(ScriptRunStatus.Unavailable);
|
||
runner.SuspectKeys.ShouldBe([VtKey("poison")], "a death during its compile makes it a suspect");
|
||
runner.QuarantinedKeys(ScriptContextKind.VirtualTag).ShouldBeEmpty("ambiguous on a shard: no strike");
|
||
|
||
await CrashDuringCompile(() => runner.Isolation.Current);
|
||
runner.QuarantinedKeys(ScriptContextKind.VirtualTag).ShouldBe(["poison"]);
|
||
var refused = await runner.RunAsync(Scripts.Vt("poison", poison), Ct);
|
||
refused.Reason!.ShouldStartWith(ScriptRunGuard.QuarantinedReasonPrefix);
|
||
refused.Reason.ShouldContain("compile crashed the isolation worker");
|
||
(await runner.RunAsync(Scripts.Vt("sibling", Healthy), Ct)).Value.ShouldBe(7);
|
||
}
|
||
|
||
// ---------------------------------------------------------------- state, circuit, gauges
|
||
|
||
[Theory]
|
||
[InlineData(new double[] { }, null, ScriptRunnerState.Ready)]
|
||
[InlineData(new double[] { -1, -1 }, null, ScriptRunnerState.Ready)]
|
||
[InlineData(new double[] { -1, 5 }, null, ScriptRunnerState.Degraded)]
|
||
[InlineData(new double[] { -1, -1 }, 5.0, ScriptRunnerState.Degraded)]
|
||
[InlineData(new double[] { 200, 5 }, null, ScriptRunnerState.Degraded)]
|
||
[InlineData(new double[] { 200, 130 }, null, ScriptRunnerState.Unavailable)]
|
||
public void The_runner_state_rule(double[] shardsDownSeconds, double? isolationDownSeconds, ScriptRunnerState expected)
|
||
{
|
||
var shards = shardsDownSeconds.Select(s => s < 0 ? (TimeSpan?)null : TimeSpan.FromSeconds(s)).ToArray();
|
||
var iso = isolationDownSeconds is { } i ? TimeSpan.FromSeconds(i) : (TimeSpan?)null;
|
||
OutOfProcessScriptRunner.ComputeState(shards, iso, TimeSpan.FromMinutes(2)).ShouldBe(expected);
|
||
}
|
||
|
||
/// <summary>
|
||
/// The crash-loop circuit: every shard down past CircuitOpenAfter makes the runner Unavailable (gauge value 2),
|
||
/// logs one Error, and answers evaluations at once instead of after UnavailableGrace. Mutation check: without the
|
||
/// zero grace under an open circuit, the evaluation waits the full 3 s.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task Every_shard_down_opens_the_circuit_answers_at_once_and_reports_state_2()
|
||
{
|
||
var sink = new CapturingSink();
|
||
var missing = Path.Combine(Path.GetTempPath(), "no-such-dir-" + Guid.NewGuid().ToString("N"), OutOfProcessScriptRunner.WorkerAssemblyFileName);
|
||
await using var runner = Scripts.Runner(
|
||
Scripts.Options(workerPath: missing, unavailableGrace: TimeSpan.FromSeconds(3), circuitOpenAfter: TimeSpan.FromSeconds(1)),
|
||
sink.CreateLogger(), maintenanceInterval: TimeSpan.FromMilliseconds(100));
|
||
runner.State.ShouldBe(ScriptRunnerState.Ready, "nothing needed yet");
|
||
|
||
(await runner.RunAsync(Scripts.Vt("vt", "return 1;"), Ct)).Status.ShouldBe(ScriptRunStatus.Unavailable);
|
||
runner.State.ShouldNotBe(ScriptRunnerState.Ready);
|
||
await Scripts.Eventually(() => runner.State == ScriptRunnerState.Unavailable, TimeSpan.FromSeconds(10), "the circuit open");
|
||
|
||
var sw = Stopwatch.StartNew();
|
||
var fast = await runner.RunAsync(Scripts.Vt("vt", "return 1;"), Ct);
|
||
sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(1), "an open circuit answers without waiting out the 3 s grace");
|
||
fast.Status.ShouldBe(ScriptRunStatus.Unavailable);
|
||
fast.Reason.ShouldBe(ScriptRunOutcome.RunnerUnavailableReason);
|
||
|
||
await Scripts.Eventually(() => sink.Rendered.Any(m => m.Contains("circuit OPEN")), TimeSpan.FromSeconds(5), "the circuit Error");
|
||
await Task.Delay(500, Ct);
|
||
sink.Events.Count(e => e.Level == Serilog.Events.LogEventLevel.Error && e.RenderMessage().Contains("circuit OPEN"))
|
||
.ShouldBe(1, "logged once, not per maintenance pass");
|
||
|
||
var states = new List<int>();
|
||
using var listener = new MeterListener();
|
||
listener.InstrumentPublished = (instrument, l) =>
|
||
{
|
||
if (instrument.Name == "otopcua.script.runner.state") l.EnableMeasurementEvents(instrument);
|
||
};
|
||
listener.SetMeasurementEventCallback<int>((_, value, _, _) => states.Add(value));
|
||
listener.Start();
|
||
listener.RecordObservableInstruments();
|
||
states.ShouldContain(2);
|
||
runner.Status().State.ShouldBe(ScriptRunnerState.Unavailable);
|
||
}
|
||
|
||
/// <summary>The per-shard gauges observe the live worker (positive control: the shard is up and reports its pid).</summary>
|
||
[Fact]
|
||
public async Task The_shard_gauges_report_a_live_worker()
|
||
{
|
||
await using var runner = Scripts.Runner(Scripts.Options());
|
||
(await runner.RunAsync(Scripts.Vt("vt", Healthy), Ct)).Value.ShouldBe(7);
|
||
await Scripts.Eventually(() => runner.Shards[0].Current?.LastHeartbeat is not null, TimeSpan.FromSeconds(5), "a heartbeat");
|
||
|
||
var up = new List<(int Value, string? Shard)>();
|
||
var heap = new List<long>();
|
||
using var listener = new MeterListener();
|
||
listener.InstrumentPublished = (instrument, l) =>
|
||
{
|
||
if (instrument.Name is "otopcua.script.worker.up" or "otopcua.script.worker.heap_bytes") l.EnableMeasurementEvents(instrument);
|
||
};
|
||
listener.SetMeasurementEventCallback<int>((_, value, tags, _) =>
|
||
{
|
||
foreach (var tag in tags) if (tag.Key == "shard") up.Add((value, tag.Value as string));
|
||
});
|
||
listener.SetMeasurementEventCallback<long>((_, value, _, _) => heap.Add(value));
|
||
listener.Start();
|
||
listener.RecordObservableInstruments();
|
||
|
||
up.ShouldContain((1, "0"));
|
||
up.ShouldContain((0, "iso"), "the isolation worker is not running");
|
||
heap.ShouldContain(h => h > 0);
|
||
var status = runner.Status();
|
||
status.State.ShouldBe(ScriptRunnerState.Ready);
|
||
status.Shards.Single(s => s.Label == "0").Pid.ShouldBe(runner.Shards[0].Current!.Pid);
|
||
}
|
||
|
||
// ---------------------------------------------------------------- lmxopcua#789: who resumes the caller
|
||
|
||
/// <summary>
|
||
/// lmxopcua#789: callers resume on the connection's result dispatcher, never on the pipe reader — so a consumer
|
||
/// that blocks after its await (the Calculation pump publishes to OPC UA subscribers there) cannot stop the reader
|
||
/// from taking heartbeats and get a healthy worker killed for silence. Here a 100 ms heartbeat (kill after 300 ms
|
||
/// of silence) and a continuation that blocks for 1.5 s. Mutation check: resume callers inline on the reader (a
|
||
/// synchronous-continuation TaskCompletionSource completed by the reader, no dispatcher) and the worker is killed
|
||
/// (KilledHeartbeat).
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task A_caller_that_blocks_after_its_result_does_not_starve_the_reader()
|
||
{
|
||
var sink = new CapturingSink();
|
||
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 =>
|
||
{
|
||
// Runs wherever RunAsync's own continuation completed it: the thread that resumed the caller.
|
||
resumedOn = Thread.CurrentThread.Name;
|
||
t.Result.Value.ShouldBe(7);
|
||
Thread.Sleep(1_500);
|
||
}, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
|
||
|
||
// Behaviour first: the worker was not killed for silence while the caller blocked.
|
||
(await Task.WhenAny(worker.Exited, Task.Delay(1_000, Ct))).ShouldNotBe(worker.Exited,
|
||
string.Join(Environment.NewLine, sink.Rendered.Where(l => l.Contains("ScriptWorker", StringComparison.Ordinal))));
|
||
var log = string.Join(Environment.NewLine, sink.Rendered.Where(l => l.Contains("ScriptWorker", StringComparison.Ordinal)));
|
||
resumedOn.ShouldBe("ScriptWorker[0].Results", log);
|
||
(await runner.RunAsync(Scripts.Vt("warm", Healthy), Ct)).Value.ShouldBe(7);
|
||
runner.Shards[0].Current.ShouldNotBeNull().Pid.ShouldBe(worker.Pid, log);
|
||
}
|
||
|
||
// ---------------------------------------------------------------- lmxopcua#788: start-failure logging
|
||
|
||
/// <summary>
|
||
/// lmxopcua#788: a shard that keeps failing to start logs ONE Error with the exception for the down period; its
|
||
/// retries are a Warning without a stack, carrying the attempt count, rate-limited (the rest at Debug); and the
|
||
/// start that ends the outage is one Information line naming how many attempts failed. The docker-dev drill logged
|
||
/// 28 Error+stack entries in two minutes on one node before this. Mutation check: log every failure at Error with
|
||
/// the exception again and the single-Error assertion goes red (3 Errors).
|
||
/// </summary>
|
||
[Theory]
|
||
[InlineData(0, 2, 0)]
|
||
[InlineData(3_600_000, 0, 2)]
|
||
public async Task A_shard_that_cannot_start_logs_one_error_then_concise_rate_limited_retries_then_its_recovery(
|
||
int warnEveryMs, int expectedWarnings, int expectedDebugs)
|
||
{
|
||
var sink = new CapturingSink();
|
||
var log = sink.CreateLogger();
|
||
var spec = new WorkerLaunchSpec("t", OutOfProcessScriptRunner.ResolveWorkerPath(new ScriptWorkerOptions { ExecutablePath = Scripts.WorkerUnderTest }),
|
||
256L * 1024 * 1024, EvalThreads: 2, HardGrace: TimeSpan.FromSeconds(5), HeartbeatInterval: TimeSpan.FromSeconds(1),
|
||
HandshakeTimeout: TimeSpan.FromSeconds(15));
|
||
var attempts = 0;
|
||
await using var shard = new WorkerShard("t", isolation: false,
|
||
(onExiting, ct) => Interlocked.Increment(ref attempts) <= 3
|
||
? throw new InvalidOperationException("the worker exited (code 1) before connecting to its pipe")
|
||
: WorkerConnection.StartAsync(spec, log, _ => { }, onExiting, ct),
|
||
(_, _) => { }, log, startFailureWarningInterval: TimeSpan.FromMilliseconds(warnEveryMs));
|
||
|
||
// The background recovery retries at the backoff (250 ms, 500 ms, 1 s, ±20 %) until the 4th attempt starts.
|
||
await shard.WhenRecoveredAsync(null).WaitAsync(TimeSpan.FromSeconds(30), Ct);
|
||
shard.Current.ShouldNotBeNull();
|
||
attempts.ShouldBe(4);
|
||
|
||
var events = sink.Events.Where(e => e.MessageTemplate.Text.StartsWith("ScriptWorker[{Shard}]", StringComparison.Ordinal)).ToList();
|
||
string Dump() => string.Join(Environment.NewLine, events.Select(e => $"{e.Level} {e.RenderMessage()} {(e.Exception is null ? "" : "+exception")}"));
|
||
static int Attempts(Serilog.Events.LogEvent e) => int.Parse(e.Properties["Attempts"].ToString(), System.Globalization.CultureInfo.InvariantCulture);
|
||
|
||
var errors = events.Where(e => e.Level == Serilog.Events.LogEventLevel.Error).ToList();
|
||
errors.Count.ShouldBe(1, Dump());
|
||
errors[0].Exception.ShouldNotBeNull("the first failure of the outage carries the stack");
|
||
|
||
var warnings = events.Where(e => e.Level == Serilog.Events.LogEventLevel.Warning).ToList();
|
||
warnings.Count.ShouldBe(expectedWarnings, Dump());
|
||
warnings.ShouldAllBe(e => e.Exception == null);
|
||
warnings.Select(Attempts).ShouldBe(expectedWarnings == 0 ? [] : [2, 3]);
|
||
events.Count(e => e.Level == Serilog.Events.LogEventLevel.Debug && e.Properties.ContainsKey("Attempts")).ShouldBe(expectedDebugs, Dump());
|
||
|
||
var recovered = events.Single(e => e.Level == Serilog.Events.LogEventLevel.Information && e.Properties.ContainsKey("Attempts"));
|
||
Attempts(recovered).ShouldBe(3);
|
||
recovered.RenderMessage().ShouldContain("after 3 failed start attempt(s)");
|
||
}
|
||
|
||
// ---------------------------------------------------------------- lmxopcua#790: compile timeout from compile start
|
||
|
||
/// <summary>A script of <paramref name="lines"/> statements: Roslyn needs time roughly proportional to it.</summary>
|
||
private static string Lines(int lines, int salt)
|
||
{
|
||
var body = new StringBuilder("var a = ").Append(salt).Append(";\n");
|
||
for (var i = 0; i < lines; i++) body.Append("a += (int)ctx.GetTag(\"x\").Value * ").Append(i % 97).Append(";\n");
|
||
return body.Append("return a;").ToString();
|
||
}
|
||
|
||
private static int LinesValue(int lines, int salt) => salt + Enumerable.Range(0, lines).Sum(i => i % 97);
|
||
|
||
/// <summary>
|
||
/// lmxopcua#790: the compile timeout runs from when a compile thread STARTS a load, so a queue of compiles that
|
||
/// takes several load timeouts to drain kills nothing and strikes nobody. One compile thread; the compiler is
|
||
/// warmed under a generous timeout (a cold Roslyn start alone takes seconds on the Windows VM), then the live
|
||
/// connection's timeout drops to four times one measured compile (at least 3 s; a 500-statement script takes
|
||
/// ~0.1 s on an M-series Mac and ~1.5 s on the Windows VM) and a burst of such scripts, sized from that and grown
|
||
/// until it does, queues for over three timeouts — asserted, so the test cannot pass vacuously
|
||
/// — while each compile stays far inside the timeout. Mutation check: time the load from when the host SENT it
|
||
/// (the old <c>WaitAsync(LoadTimeout)</c> around the load) and the worker is killed and the queued keys are
|
||
/// struck.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task Compiles_queued_behind_others_longer_than_the_load_timeout_are_not_killed_or_struck()
|
||
{
|
||
const int lines = 500;
|
||
var sink = new CapturingSink();
|
||
await using var runner = new OutOfProcessScriptRunner(Scripts.Options(), sink.CreateLogger(), null,
|
||
heartbeatInterval: TimeSpan.FromSeconds(1), loadTimeout: TimeSpan.FromSeconds(120), compileThreads: 1);
|
||
(await runner.RunAsync(Scripts.Vt("first", Healthy), Ct)).Value.ShouldBe(7);
|
||
var worker = runner.Shards[0].Current.ShouldNotBeNull();
|
||
|
||
ScriptInvocation Script(string prefix, int i)
|
||
=> Scripts.Vt($"{prefix}-{i}", Lines(lines, i), TimeSpan.FromSeconds(300), Scripts.Input("x", 1));
|
||
|
||
// Warm the compiler, then calibrate: the compile time of one such script on this box.
|
||
for (var i = 0; i < 3; i++) (await runner.RunAsync(Script("warm", i), Ct)).Value.ShouldBe(LinesValue(lines, i));
|
||
var probe = Stopwatch.StartNew();
|
||
for (var i = 0; i < 3; i++) (await runner.RunAsync(Script("probe", i), Ct)).Value.ShouldBe(LinesValue(lines, i));
|
||
var perCompile = probe.Elapsed / 3;
|
||
var loadTimeout = TimeSpan.FromTicks(Math.Max(TimeSpan.FromSeconds(3).Ticks, perCompile.Ticks * 4));
|
||
worker.LoadTimeout = loadTimeout;
|
||
var count = (int)Math.Clamp(Math.Ceiling(7 * loadTimeout / perCompile), 10, 300);
|
||
|
||
// A burst compiles faster per script than the sequential probe (the compiler keeps warming up), so grow the burst
|
||
// until it really outlasts three load timeouts; each burst must pass the same no-kill checks.
|
||
ScriptRunOutcome[] outcomes = [];
|
||
var elapsed = TimeSpan.Zero;
|
||
for (var round = 0; round < 4; round++)
|
||
{
|
||
runner.Reset(ScriptContextKind.VirtualTag); // unload the previous round's scripts (worker heap)
|
||
var sw = Stopwatch.StartNew();
|
||
var prefix = "queued" + round;
|
||
outcomes = await Task.WhenAll(Enumerable.Range(0, count).Select(i => runner.RunAsync(Script(prefix, i), Ct)));
|
||
elapsed = sw.Elapsed;
|
||
for (var i = 0; i < count; i++) outcomes[i].Value.ShouldBe(LinesValue(lines, i), $"round {round}: {outcomes[i].Status} {outcomes[i].Reason}");
|
||
if (elapsed > loadTimeout * 3) break;
|
||
count = Math.Min(1_200, count * 2);
|
||
}
|
||
|
||
var log = string.Join(Environment.NewLine, sink.Rendered.Where(l => l.Contains("ScriptWorker", StringComparison.Ordinal)).Take(5));
|
||
elapsed.ShouldBeGreaterThan(loadTimeout * 3,
|
||
$"precondition: {count} compiles (probe {perCompile.TotalMilliseconds:F0} ms each) must queue for several load timeouts, or this proves nothing");
|
||
worker.Exited.IsCompleted.ShouldBeFalse(log);
|
||
runner.Shards[0].Current.ShouldNotBeNull().Pid.ShouldBe(worker.Pid, log);
|
||
runner.QuarantinedKeys(ScriptContextKind.VirtualTag).ShouldBeEmpty();
|
||
runner.SuspectKeys.ShouldBeEmpty();
|
||
sink.Rendered.ShouldNotContain(l => l.Contains("killing the worker", StringComparison.Ordinal));
|
||
}
|
||
|
||
/// <summary>
|
||
/// The positive control for the test above: a compile that is still running <c>LoadTimeout</c> after it STARTED is
|
||
/// the script's fault — the worker is killed (<c>KilledLoadTimeout</c>), the caller gets the compile-timeout
|
||
/// failure, the key takes a kill strike (the second quarantines it), and the healthy key on the same shard is
|
||
/// not blamed. The healthy key comes from a primed PE cache, so a fresh worker's cold Roslyn start never counts
|
||
/// against the short timeout. Mutation check: never arm the timer on <c>LoadStarted</c> and the huge compile
|
||
/// simply finishes (status Ok, red).
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task A_compile_still_running_a_load_timeout_after_it_started_is_killed_and_struck()
|
||
{
|
||
using var dir = new Scripts.TempDir();
|
||
var cache = Path.Combine(dir.Path, "pe-cache");
|
||
await using (var primer = Scripts.Runner(Scripts.Options(peCacheDir: cache)))
|
||
(await primer.RunAsync(Scripts.Vt("first", Healthy), Ct)).Value.ShouldBe(7);
|
||
|
||
var sink = new CapturingSink();
|
||
await using var runner = new OutOfProcessScriptRunner(Scripts.Options(quarantineAfterKills: 2, peCacheDir: cache),
|
||
sink.CreateLogger(), null, heartbeatInterval: TimeSpan.FromSeconds(1), loadTimeout: TimeSpan.FromSeconds(2), compileThreads: 1);
|
||
(await runner.RunAsync(Scripts.Vt("first", Healthy), Ct)).Value.ShouldBe(7);
|
||
|
||
// 20,000 statements: about 4 s of Roslyn on an M-series Mac, longer elsewhere.
|
||
var huge = Scripts.Vt("huge", Lines(20_000, 0), TimeSpan.FromSeconds(30), Scripts.Input("x", 1));
|
||
for (var strike = 1; strike <= 2; strike++)
|
||
{
|
||
var before = runner.Shards[0].Current ?? await runner.Shards[0].GetReadyAsync(TimeSpan.FromSeconds(30), Ct);
|
||
var outcome = await runner.RunAsync(huge, Ct).WaitAsync(TimeSpan.FromSeconds(60), Ct);
|
||
outcome.Status.ShouldBe(ScriptRunStatus.Failed, $"strike {strike}");
|
||
outcome.Reason.ShouldNotBeNull().ShouldContain("compile did not finish");
|
||
var exit = await before.ShouldNotBeNull().Exited.WaitAsync(TimeSpan.FromSeconds(10), Ct);
|
||
exit.Reason.ShouldBe(WorkerExitReason.KilledLoadTimeout);
|
||
exit.CulpritKey.ShouldBe(VtKey("huge"));
|
||
}
|
||
|
||
var third = await runner.RunAsync(huge, Ct);
|
||
third.Reason.ShouldNotBeNull().ShouldStartWith(ScriptRunGuard.QuarantinedReasonPrefix);
|
||
runner.QuarantinedKeys(ScriptContextKind.VirtualTag).ShouldBe(["huge"]);
|
||
(await runner.RunAsync(Scripts.Vt("first", Healthy), Ct)).Value.ShouldBe(7, "the healthy key on the same shard is not blamed");
|
||
}
|
||
}
|