f50b29abae
v2-ci / unit-tests (push) Failing after 1h12m10s
Second P1 slice of docs/plans/2026-11-09-out-of-process-script-runner-design.md; still opt-in behind Scripting:Runner=OutOfProcess (InProcess unchanged). - Compiled-PE cache (design §7, §14 #8): Scripting:Worker:PeCacheDir holds <dir>/pe/<id>.pe entries authenticated with HMAC-SHA256 under a per-node random key (<dir>/pe-cache.key, created by the host, 0600/0700 on Unix), handed to the worker in Hello. Entry id = SHA-256(compile fingerprint + kind + source); the fingerprint covers the Roslyn version, the Core.Scripting build (analyzer, wrapper, stack guard), the sandbox reference set and the runtime. Edited, truncated, other-key or renamed entries are rejected, deleted and recompiled; the worker never loads unverified bytes. Pruned by recency at startup (30 d / 50 000). ScriptEvaluator splits into CompileToImage + FromImage. Measured on 40 scripts, restart to every first result: macOS 820 ms no cache / 840 ms empty / 85 ms warm; windev (Windows) 9.5 s / 9.4 s / 1.0 s. - Quarantine race: WorkerConnection runs exit attribution synchronously BEFORE completing the evaluations/loads it held (and before Exited), and the shard passes it straight to the runner — a kill strike is in place before the killed key's slot is released. - Supervision: a dead worker restarts by itself after its backoff; Unavailable outcomes carry ScriptRunOutcome.Recovery (-> VirtualTagEvalResult.RetryWhen) and the VT actor / Calculation driver re-evaluate once against current inputs when it completes. Isolation worker stops after Scripting:Worker:IsolationIdleTimeout (10 min). Keys compiling when a worker dies are suspects too (isolation finds a poison compile and quarantines it). Startup orphan sweep by launch marker: kills only a process whose marker's host is gone and whose pid + start time + image match. - Crash-loop circuit (Scripting:Worker:CircuitOpenAfter, 2 min): runner Unavailable, one Error, evaluations answered at once while shards keep retrying at the cap. - Observability (§10): otopcua.script.runner.state, worker.up/inflight/queue_depth/heap_bytes (shard), ipc.roundtrip + eval.duration (kind), compile{cache}, isolation.suspects, exits reason=idle. `script-runner` health check on the ready tier, at worst Degraded (§14 #12: degraded but still ready). Left on #780: self-contained worker publish, installer (account, pipe ACL, firewall), docker-dev live drill, scripted-alarm seam, the §4.5 per-seam batch submit, the §8 perf harness. Tests: SupervisionTests, PeCacheStoreTests/PeCacheWorkerTests (real workers), ScriptRunnerHealthCheckTests, VT + Calc recovery seam tests; 14 mutations each turned their test red. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
155 lines
7.5 KiB
C#
155 lines
7.5 KiB
C#
using System.Collections.Concurrent;
|
|
using Serilog;
|
|
using Shouldly;
|
|
using Xunit;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests;
|
|
|
|
/// <summary>
|
|
/// lmxopcua#780 — with <c>Scripting:Runner=OutOfProcess</c> selecting the Calculation seam, the driver hands each
|
|
/// compiled run to the node's <see cref="IScriptRunner"/>: keyed by the calc tag's RawPath, with the driver's own
|
|
/// budget. A runner-side failure publishes <c>BadResourceUnavailable</c> (0x80040000); a script failure keeps
|
|
/// <c>BadInternalError</c>. The passthrough fast path never reaches the runner.
|
|
/// </summary>
|
|
public sealed class CalculationDriverScriptRunnerTests
|
|
{
|
|
private const uint StatusGood = 0u;
|
|
private const uint StatusBadInternalError = 0x80020000u;
|
|
private const uint StatusBadResourceUnavailable = 0x80040000u;
|
|
private const string Tag = "calc/e/x";
|
|
|
|
private sealed class FakeRunner : IScriptRunner
|
|
{
|
|
public ConcurrentQueue<ScriptRunOutcome> Outcomes { get; } = new();
|
|
public ConcurrentQueue<ScriptInvocation> Invocations { get; } = new();
|
|
public List<ScriptContextKind> Resets { get; } = [];
|
|
|
|
public Task<ScriptRunOutcome> RunAsync(ScriptInvocation invocation, CancellationToken ct = default)
|
|
{
|
|
Invocations.Enqueue(invocation);
|
|
return Task.FromResult(Outcomes.TryDequeue(out var next) ? next : ScriptRunOutcome.Ok(null));
|
|
}
|
|
|
|
public void Reset(ScriptContextKind kind) => Resets.Add(kind);
|
|
|
|
public IReadOnlyCollection<string> QuarantinedKeys(ScriptContextKind kind) => ["q-" + kind];
|
|
}
|
|
|
|
private static RawTagEntry CalcTag(string source)
|
|
{
|
|
var json = $"{{\"scriptId\":\"s\",\"changeTriggered\":true,\"scriptSource\":{System.Text.Json.JsonSerializer.Serialize(source)}}}";
|
|
return new RawTagEntry(Tag, json, WriteIdempotent: false);
|
|
}
|
|
|
|
private static async Task Settle(CalculationDriver driver)
|
|
{
|
|
var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(30);
|
|
while (DateTime.UtcNow < deadline && !driver.IsEvaluationIdle) await Task.Delay(10);
|
|
driver.IsEvaluationIdle.ShouldBeTrue();
|
|
}
|
|
|
|
private static (CalculationDriver Driver, ConcurrentQueue<DataChangeEventArgs> Published) Build(FakeRunner runner, string source)
|
|
{
|
|
var driver = new CalculationDriver(
|
|
new CalculationDriverOptions { RawTags = [CalcTag(source)], RunTimeout = TimeSpan.FromMilliseconds(1234), ScriptRunner = runner },
|
|
"calc-drv", new ScriptRootLogger(new LoggerConfiguration().CreateLogger()));
|
|
var published = new ConcurrentQueue<DataChangeEventArgs>();
|
|
driver.OnDataChange += (_, e) => published.Enqueue(e);
|
|
return (driver, published);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task A_runner_failure_publishes_BadResourceUnavailable_and_a_script_failure_BadInternalError()
|
|
{
|
|
var runner = new FakeRunner();
|
|
runner.Outcomes.Enqueue(ScriptRunOutcome.Ok(30.0));
|
|
runner.Outcomes.Enqueue(ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerRestartedReason));
|
|
runner.Outcomes.Enqueue(ScriptRunOutcome.Failed("script threw: boom"));
|
|
var (driver, published) = Build(runner, "return (double)ctx.GetTag(\"src/a\").Value * 3;");
|
|
await driver.InitializeAsync("{}", TestContext.Current.CancellationToken);
|
|
|
|
driver.OnDependencyValue("src/a", 10.0, StatusGood, DateTime.UtcNow);
|
|
await Settle(driver);
|
|
published.TryDequeue(out var good).ShouldBeTrue();
|
|
good!.Snapshot.StatusCode.ShouldBe(StatusGood);
|
|
good.Snapshot.Value.ShouldBe(30.0);
|
|
|
|
var invocation = runner.Invocations.Single();
|
|
invocation.Key.ShouldBe(Tag);
|
|
invocation.Kind.ShouldBe(ScriptContextKind.Calculation);
|
|
invocation.Timeout.ShouldBe(TimeSpan.FromMilliseconds(1234));
|
|
invocation.Inputs["src/a"].Value.ShouldBe(10.0);
|
|
|
|
driver.OnDependencyValue("src/a", 11.0, StatusGood, DateTime.UtcNow);
|
|
await Settle(driver);
|
|
published.TryDequeue(out var unavailable).ShouldBeTrue();
|
|
unavailable!.Snapshot.StatusCode.ShouldBe(StatusBadResourceUnavailable);
|
|
unavailable.Snapshot.Value.ShouldBe(30.0, "the last known value rides out with the Bad status");
|
|
|
|
driver.OnDependencyValue("src/a", 12.0, StatusGood, DateTime.UtcNow);
|
|
await Settle(driver);
|
|
published.TryDequeue(out var scriptFault).ShouldBeTrue();
|
|
scriptFault!.Snapshot.StatusCode.ShouldBe(StatusBadInternalError);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Design §9: a calc tag whose run was a runner failure is re-evaluated when the runner recovers, against the
|
|
/// CURRENT inputs, with no input change. Mutation check: without the driver arming
|
|
/// <c>RetryWhen</c> the tag stays BadResourceUnavailable.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task A_runner_unavailable_result_is_re_evaluated_with_current_inputs_when_the_runner_recovers()
|
|
{
|
|
var runner = new FakeRunner();
|
|
var recovered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
runner.Outcomes.Enqueue(ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerRestartedReason) with { Recovery = recovered.Task });
|
|
runner.Outcomes.Enqueue(ScriptRunOutcome.Ok(33.0));
|
|
var (driver, published) = Build(runner, "return (double)ctx.GetTag(\"src/a\").Value * 3;");
|
|
await driver.InitializeAsync("{}", TestContext.Current.CancellationToken);
|
|
|
|
driver.OnDependencyValue("src/a", 11.0, StatusGood, DateTime.UtcNow);
|
|
await Settle(driver);
|
|
published.TryDequeue(out var unavailable).ShouldBeTrue();
|
|
unavailable!.Snapshot.StatusCode.ShouldBe(StatusBadResourceUnavailable);
|
|
runner.Invocations.Count.ShouldBe(1);
|
|
|
|
recovered.SetResult();
|
|
var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10);
|
|
while (DateTime.UtcNow < deadline && published.IsEmpty) await Task.Delay(10, TestContext.Current.CancellationToken);
|
|
await Settle(driver);
|
|
published.TryDequeue(out var good).ShouldBeTrue("re-evaluated on recovery with no input change");
|
|
good!.Snapshot.StatusCode.ShouldBe(StatusGood);
|
|
good.Snapshot.Value.ShouldBe(33.0);
|
|
runner.Invocations.Count.ShouldBe(2);
|
|
runner.Invocations.Last().Inputs["src/a"].Value.ShouldBe(11.0, "the current input");
|
|
}
|
|
|
|
[Fact]
|
|
public async Task The_passthrough_fast_path_never_reaches_the_runner()
|
|
{
|
|
var runner = new FakeRunner();
|
|
var (driver, published) = Build(runner, "return ctx.GetTag(\"src/a\").Value;");
|
|
await driver.InitializeAsync("{}", TestContext.Current.CancellationToken);
|
|
|
|
driver.OnDependencyValue("src/a", 4.0, StatusGood, DateTime.UtcNow);
|
|
await Settle(driver);
|
|
published.TryDequeue(out var mirrored).ShouldBeTrue();
|
|
mirrored!.Snapshot.Value.ShouldBe(4.0);
|
|
runner.Invocations.ShouldBeEmpty();
|
|
}
|
|
|
|
[Fact]
|
|
public void The_deploy_boundary_resets_the_runners_calculation_seam_and_quarantines_are_reported()
|
|
{
|
|
var runner = new FakeRunner();
|
|
using var evaluator = new CalculationEvaluator(Microsoft.Extensions.Logging.Abstractions.NullLogger.Instance,
|
|
new ScriptRootLogger(new LoggerConfiguration().CreateLogger()), scriptRunner: runner);
|
|
evaluator.ClearCompiledScripts();
|
|
evaluator.ResetQuarantines();
|
|
runner.Resets.ShouldBe([ScriptContextKind.Calculation, ScriptContextKind.Calculation]);
|
|
evaluator.QuarantinedScripts.ShouldBe(["q-Calculation"]);
|
|
}
|
|
}
|