diff --git a/CLAUDE.md b/CLAUDE.md index f4636df1..6cef1de7 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -392,6 +392,8 @@ See `docs/ScriptEditor.md` for the full guide. Scripts may use the `{{equip}}` t **Out-of-process script runner (lmxopcua#780, opt-in).** `Scripting:Runner` = `InProcess` (default, unchanged) | `OutOfProcess` | `Shadow` (not yet). `OutOfProcess` runs the VirtualTag + Calculation seams' compiled scripts in `ZB.MOM.WW.OtOpcUa.ScriptWorker` processes (`src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess`, protocol in `Core.Scripting.Protocol`), shipped in `script-worker/` -beside the host; a runner-side failure publishes `BadResourceUnavailable` (ScadaBridge-visible, ScadaBridge#112). The worker tests +beside the host; a runner-side failure publishes `BadResourceUnavailable` (ScadaBridge-visible, ScadaBridge#112) and the seam re-evaluates +when the worker is back. Optional `Scripting:Worker:PeCacheDir` = HMAC'd compiled-PE cache (restart without Roslyn); the `script-runner` +readiness check is at worst Degraded (never fails ready). The worker tests are `tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests` (`Category=ScriptWorker`, real processes). See `docs/VirtualTags.md` §"Out-of-process runner". diff --git a/docs/ServiceHosting.md b/docs/ServiceHosting.md index 3ad4427d..899f8d55 100644 --- a/docs/ServiceHosting.md +++ b/docs/ServiceHosting.md @@ -94,7 +94,11 @@ Off by default. When enabled, a driver node launches up to `Scripting:Worker:Sha (`ZB.MOM.WW.OtOpcUa.ScriptWorker`, from `script-worker/` beside the host) on its first compiled script evaluation and kills them with it: a Job Object with kill-on-close on Windows, pipe end of stream plus a parent-pid poll elsewhere, and a Shutdown frame then a kill on host stop. Their stderr lands in the host log prefixed `ScriptWorker[]`. -The worker currently runs under the host's account through the dotnet muxer; the dedicated low-privilege account, +At startup the host kills workers a dead predecessor of the same install left behind, matched by launch marker +(pid + start time + image, and only when the marker's host is gone). `Scripting:Worker:PeCacheDir`, when set, holds the +compiled-PE cache, its HMAC key and the launch markers: put it outside the install folder (startup refuses a relative path +or one inside the content root) on a directory only the service account can read. The `script-runner` readiness component +is at worst Degraded. The worker currently runs under the host's account through the dotnet muxer; the dedicated low-privilege account, pipe ACL and firewall outbound rule (design §14 #5/#6) are installer work still open on lmxopcua#780. See `docs/VirtualTags.md` §"Out-of-process runner". diff --git a/docs/VirtualTags.md b/docs/VirtualTags.md index c8f91f40..f4d0e602 100644 --- a/docs/VirtualTags.md +++ b/docs/VirtualTags.md @@ -76,12 +76,49 @@ and Calculation evaluators hand each compiled run to `OutOfProcessScriptRunner` or when its parent pid disappears. Results cross as protobuf `Variant`s carrying the exact CLR type, so `(short)ctx.GetTag(x).Value` behaves as in process; an input or result type with no tag (a `List`, a `Guid`) is an explicit `unsupported type: …` failure rather than a conversion — a behaviour change from in process. +- **Attribution happens before release.** The worker's exit path runs attribution (strike, quarantine, suspect + marks) synchronously **before** it completes any evaluation the worker held, so a script killed for its deadline is + already quarantined when its single-flight slot is released — no trigger can re-run it in the restart gap. A worker + that dies while a script is **compiling** makes that script a suspect too; it then compiles alone in the isolation + worker, where a second death quarantines it (`"its compile crashed the isolation worker"`), so a poison source cannot + crash-loop its shard. +- **Recovery.** A worker that dies restarts by itself after its backoff (250 ms doubling to 30 s, ±20 %), not on the + next trigger. An `Unavailable` outcome carries a recovery task (`ScriptRunOutcome.Recovery` → + `VirtualTagEvalResult.RetryWhen`); the VirtualTag actor and the Calculation driver re-evaluate once, against their + **current** inputs, when it completes — so a victim with a static input does not stay `BadResourceUnavailable`. The + isolation worker stops after `Scripting:Worker:IsolationIdleTimeout` (10 min) with nothing to run and starts again for + the next suspect. +- **Crash-loop circuit.** When every shard has been down for `Scripting:Worker:CircuitOpenAfter` (2 min) the runner is + `Unavailable`: one Error is logged, evaluations answer `script runner unavailable` at once instead of each waiting out + `UnavailableGrace`, and the shards keep retrying at the backoff cap. +- **Compiled-PE cache** (`Scripting:Worker:PeCacheDir`, optional; absolute and outside the content root, or startup + fails). Workers write each compiled image to `/pe/.pe`, authenticated with HMAC-SHA256 under a per-node + random key (`/pe-cache.key`, created by the host; 0600 / directory 0700 on Linux and macOS, the directory ACL on + Windows), and a restarted worker loads the image instead of re-entering Roslyn. The id is SHA-256 over the source, the + context kind and a compile fingerprint (Roslyn version, the `Core.Scripting` build — analyzer, wrapper, stack guard —, + the sandbox reference set, the runtime), so a compiler or analyzer change misses every old entry. An entry that fails + verification (edited, truncated, another key, copied under another id) is deleted and the source recompiled; the + worker never loads bytes that did not verify. The key reaches the worker in `Hello`, never on disk or in its + environment. Entries unused for 30 days, and the least recently used beyond 50 000, are pruned at startup; the + directory can be deleted at any time. Measured on macOS (40 scripts, restart to every first result): 820 ms with no + cache, 840 ms with an empty one, **85 ms warm**; on windev (Windows) 9.5 s / 9.4 s / **1.0 s**. An unusable directory is a Warning and a cache-less runner. +- **Orphan sweep.** Every launched worker gets a marker (`/workers/.worker`, or a per-install + directory under temp) naming its pid, start time and image, and the host's pid and start time. At startup the runner + kills only a process whose marker's host is gone **and** whose pid, start time and image all still match — never + "every process with this name", never a live host's worker (mxaccessgw's `OrphanWorkerTerminator` lesson). +- **Observability.** On `OtOpcUaTelemetry.Meter`: `otopcua.script.runner.state` (0 ready / 1 degraded / 2 circuit + open), `otopcua.script.worker.up` / `.inflight` / `.queue_depth` / `.heap_bytes` (tag `shard`; queue and heap from + the worker heartbeat), `otopcua.script.worker.starts` / `.exits` (`reason` adds `idle`), `otopcua.script.ipc.roundtrip` + and `otopcua.script.eval.duration` (µs, tag `kind`; no per-key tag), `otopcua.script.compile` (`cache` = `hit` | + `miss` | `rejected` | `uncached` | `negative`) and `otopcua.script.isolation.suspects`. The `script-runner` health check + is on the **ready** tier and is at worst **Degraded** (still 200): a sick runner never fails readiness, so a pair + does not move ServiceLevel to its peer over a script problem (design §14 #12). - **Deployment.** The Host build copies the worker into `script-worker/` beside `OtOpcUa.Host.dll` (`ScriptWorker.Deploy.targets`, Build and Publish); `Scripting:Worker:ExecutablePath` overrides it. It is run through - the dotnet muxer, so it needs a framework-dependent host (the docker-dev image). Not yet done (lmxopcua#780 P1 - remainder): the PE cache (`Scripting:Worker:PeCacheDir` is validated — absolute, outside the content root — but not - used, so a worker restart recompiles), a self-contained worker for the Windows installs, the low-privilege worker - account and firewall rule, the runner-state gauge / readiness component, and the scripted-alarm seam. + the dotnet muxer, so it needs a framework-dependent host (the docker-dev image). Not yet done (lmxopcua#780): a + self-contained worker for the Windows installs, the low-privilege worker account, pipe ACL and firewall rule + (installer), the docker-dev live drill, the scripted-alarm seam, the per-seam batch submit (design §4.5) and the §8 + perf harness. Pinned by `tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests` (`Category=ScriptWorker`: real worker processes). diff --git a/docs/plans/2026-11-09-out-of-process-script-runner-design.md b/docs/plans/2026-11-09-out-of-process-script-runner-design.md index 6e851c80..0548265d 100644 --- a/docs/plans/2026-11-09-out-of-process-script-runner-design.md +++ b/docs/plans/2026-11-09-out-of-process-script-runner-design.md @@ -677,6 +677,15 @@ per-shard state, restarts in the last hour and the quarantine list. Quarantines publish + the low-privilege account / pipe ACL / firewall rule (installer), the runner-state gauge and readiness component, victim re-evaluation on shard return, isolation-worker idle stop, the orphan sweep by nonce file, the scripted-alarm seam, the docker-dev enablement and the §8 perf harness. + - **Second P1 slice (lmxopcua#780):** the HMAC'd PE cache (§7; keyed by source + kind + a compile fingerprint, tampered + or mismatched entries evicted, pruned by recency at startup instead of by generation — this runner has no + generation object), attribution before release (the quarantine race), loading keys counted in attribution + (culprit-finding for a compile that crashes its worker, by isolation rather than the §5.3.5 three-strikes + bisection — loads are per key here), proactive restart + `Recovery` / `RetryWhen` re-evaluation of victims in the + VT and Calc seams, isolation idle stop (`IsolationIdleTimeout`), the crash-loop circuit (`CircuitOpenAfter`), the + §10 gauges and histograms, the degraded-but-ready `script-runner` health check (§14 #12), and the orphan sweep by + launch marker (pid + start time + image, host gone). Still open: self-contained worker publish, installer (account, + pipe ACL, firewall), docker-dev live drill, scripted-alarm seam, the §4.5 batch submit, the §8 perf harness. 3. **P2, shadow mode on docker-dev and QA.** - `Shadow` runs the **in-process** evaluation as the published result, and *also* submits to the worker. - It compares value, status class and failure kind per evaluation, and counts diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Engines/IVirtualTagEvaluator.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Engines/IVirtualTagEvaluator.cs index 67f43a1b..228fb54e 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Engines/IVirtualTagEvaluator.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Engines/IVirtualTagEvaluator.cs @@ -95,6 +95,13 @@ public sealed record VirtualTagEvalResult(bool Success, object? Value, string? R /// public uint? StatusCode { get; init; } + /// + /// For a runner failure (): completes when the runner can run the script again — its + /// worker is back — so the seam re-evaluates it then against its latest inputs instead of waiting for the next + /// input change (lmxopcua#780 design §9). Null for every other result. + /// + public Task? RetryWhen { get; init; } + /// A failure that is the runner's, not the script's: published BadResourceUnavailable. /// Description of the failure cause. /// A failed evaluation result carrying . 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 e3e99528..3dcce7f1 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 @@ -1,17 +1,45 @@ using System.Collections.Concurrent; +using System.Diagnostics; +using System.Diagnostics.Metrics; using System.Security.Cryptography; using System.Text; using Serilog; -using Serilog.Core; using Serilog.Events; using ZB.MOM.WW.OtOpcUa.Commons.Engines; +using ZB.MOM.WW.OtOpcUa.Commons.Observability; using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol; using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.V1; namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess; +/// The out-of-process runner's health (the otopcua.script.runner.state gauge: 0 / 1 / 2). +public enum ScriptRunnerState +{ + /// Every shard that has been needed is up. + Ready = 0, + + /// At least one shard (or the isolation worker) is down after a failure; the rest serve. + Degraded = 1, + + /// The crash-loop circuit is open: every shard has been down for CircuitOpenAfter (design §5.3.6). + Unavailable = 2, +} + +/// A point-in-time view of the runner (the readiness check and diagnostics). +/// The overall state. +/// Every shard, the isolation worker last. +/// Keys currently routed to the isolation worker. +/// Keys currently quarantined (all seams). +/// Whether the compiled-PE cache is in use. +public sealed record ScriptRunnerStatus( + ScriptRunnerState State, + IReadOnlyList Shards, + int SuspectCount, + int QuarantinedCount, + bool PeCacheEnabled); + /// -/// The pre-emptible (lmxopcua#780, design §3–§9): compiled scripts run in a pool of +/// The pre-emptible (lmxopcua#780, design §3–§10): compiled scripts run in a pool of /// ScriptWorker processes — K shards placed by a stable hash of the script key, plus an isolation worker /// for suspects — so a script that overruns its hard deadline is stopped by killing its worker, and an OOM, a /// native crash or a fail-fast takes down a worker instead of the server. @@ -21,16 +49,22 @@ namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess; /// Two deadlines. At the soft deadline T the caller gets today's timeout failure (and the /// run token is cancelled inside the worker). At T + G () /// the worker is killed and the key takes a kill strike. -/// Attribution. A deadline kill blames exactly one key; the other evaluations in flight on that -/// worker are victims and report (no strike). A crash with -/// no deadline crossed makes every key in flight a suspect, and suspects run in the isolation worker one at a -/// time, where a crash names exactly one key. A suspect returns to its shard after +/// Attribution runs synchronously in the worker's exit path, before any evaluation it held is +/// completed — so a quarantine earned by a kill is in place before the key's slot is released and the next +/// trigger is admitted. A deadline kill blames exactly one key; the other evaluations in flight on that worker +/// are victims and report (no strike). A crash with no +/// deadline crossed makes every key in flight — evaluating or compiling — a suspect, and suspects run (and +/// compile) in the isolation worker one at a time, where a crash names exactly one key; that is also how a +/// script whose compile keeps crashing its worker is found. A suspect returns to its shard after /// clean runs. /// Quarantine lives here, in the host, so it survives worker restarts: kill strikes (within an /// hour), consecutive soft timeouts, or one allocation-budget breach. Reasons keep the /// , and a source change or lifts them. /// Fail closed. When no worker is ready within -/// the result is — never an in-process run. +/// the result is — never an in-process run — carrying a +/// task the seam re-evaluates on. +/// PE cache (design §7): with set, workers load +/// HMAC-verified compiled images instead of re-entering Roslyn after a restart. /// /// public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, IDisposable @@ -44,8 +78,52 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, /// Failure reason when the worker could not take more work. public const string WorkerBusyReason = "script worker busy"; + /// PE cache entries unused for this long are pruned at startup. + internal static readonly TimeSpan PeCacheMaxAge = TimeSpan.FromDays(30); + + /// At most this many PE cache entries are kept. + internal const int PeCacheMaxEntries = 50_000; + private static readonly TimeSpan KillStrikeWindow = TimeSpan.FromHours(1); + private static readonly ConcurrentDictionary LiveRunners = new(); + + private static readonly Counter CompileCounter = OtOpcUaTelemetry.Meter.CreateCounter( + "otopcua.script.compile", unit: "{compile}", + description: "Out-of-process script loads (tag: cache = hit | miss | rejected | uncached | negative)."); + + private static readonly Histogram RoundTripHistogram = OtOpcUaTelemetry.Meter.CreateHistogram( + "otopcua.script.ipc.roundtrip", unit: "us", + description: "Host-measured submit-to-result time of one out-of-process evaluation (tag: kind)."); + + private static readonly Histogram EvalDurationHistogram = OtOpcUaTelemetry.Meter.CreateHistogram( + "otopcua.script.eval.duration", unit: "us", + description: "Worker-measured run time of one out-of-process evaluation (tag: kind)."); + + static OutOfProcessScriptRunner() + { + // Gauges observe every live runner (one per node in production; tests make several). + OtOpcUaTelemetry.Meter.CreateObservableGauge("otopcua.script.runner.state", + static () => LiveRunners.Keys.Select(r => new Measurement((int)r.State)), + unit: "{state}", description: "Out-of-process script runner state: 0 ready, 1 degraded, 2 unavailable (circuit open)."); + OtOpcUaTelemetry.Meter.CreateObservableGauge("otopcua.script.worker.up", + static () => ShardMeasurements(s => s.Up ? 1 : 0), + unit: "{process}", description: "1 when the shard's ScriptWorker is live (tag: shard)."); + OtOpcUaTelemetry.Meter.CreateObservableGauge("otopcua.script.worker.inflight", + static () => ShardMeasurements(s => s.InFlight), + unit: "{eval}", description: "Evaluations outstanding on the shard's ScriptWorker (tag: shard)."); + OtOpcUaTelemetry.Meter.CreateObservableGauge("otopcua.script.worker.queue_depth", + static () => ShardMeasurements(s => s.Queued), + unit: "{eval}", description: "Evaluations queued inside the shard's ScriptWorker, from its heartbeat (tag: shard)."); + OtOpcUaTelemetry.Meter.CreateObservableGauge("otopcua.script.worker.heap_bytes", + static () => LiveRunners.Keys.SelectMany(r => r.AllShards.Select(s => s.Status())) + .Select(s => new Measurement(s.HeapBytes, new KeyValuePair("shard", s.Label))), + unit: "By", description: "The shard's ScriptWorker managed heap, from its heartbeat (tag: shard)."); + OtOpcUaTelemetry.Meter.CreateObservableGauge("otopcua.script.isolation.suspects", + static () => LiveRunners.Keys.Select(r => new Measurement(r.SuspectKeys.Count)), + unit: "{script}", description: "Scripts currently routed to the isolation worker."); + } + private readonly ScriptingOptions _options; private readonly ScriptWorkerOptions _worker; private readonly ILogger _log; @@ -54,6 +132,15 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, private readonly WorkerShard _isolation; private readonly ConcurrentDictionary _keys = new(StringComparer.Ordinal); private readonly ConcurrentDictionary _compileFailures = new(StringComparer.Ordinal); + private readonly Timer _maintenance; + private readonly string _markerDirectory; + private readonly string? _peCacheDirectory; + private readonly byte[]? _peCacheKey; + private long _peCacheHits; + private long _peCacheMisses; + private long _peCacheRejected; + private int _maintaining; + private int _circuitLogged; private int _disposed; /// Initializes the runner. No process starts until the first evaluation. @@ -66,7 +153,7 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, } internal OutOfProcessScriptRunner(ScriptingOptions options, ILogger? log, ILogger? scriptRoot, - TimeSpan heartbeatInterval, TimeSpan loadTimeout) + TimeSpan heartbeatInterval, TimeSpan loadTimeout, TimeSpan? maintenanceInterval = null, string? markerDirectory = null) { ArgumentNullException.ThrowIfNull(options); _options = options.Sanitized(out _); @@ -76,6 +163,32 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, LoadTimeout = loadTimeout; WorkerPath = ResolveWorkerPath(_worker); + // The PE cache is best effort: a directory or key that cannot be used is a Warning and a cache-less runner, + // never a startup failure (the path RULE is enforced at startup by ScriptRunnerSelection.Validate). + if (!string.IsNullOrWhiteSpace(_worker.PeCacheDir)) + { + try + { + _peCacheKey = ScriptPeCacheStore.LoadOrCreateKey(_worker.PeCacheDir); + _peCacheDirectory = _worker.PeCacheDir; + var pruned = new ScriptPeCacheStore(_peCacheDirectory, _peCacheKey).Prune(PeCacheMaxAge, PeCacheMaxEntries); + if (pruned > 0) _log.Information("Script PE cache {Directory}: pruned {Count} stale entr(ies)", _peCacheDirectory, pruned); + } + catch (Exception ex) when (ex is IOException or UnauthorizedAccessException) + { + _peCacheKey = null; + _peCacheDirectory = null; + _log.Warning(ex, "Script PE cache {Directory} is unusable; ScriptWorkers will recompile every script on each start", + _worker.PeCacheDir); + } + } + + // Design §3.3: workers a dead predecessor of THIS host instance left behind, identified by their launch + // markers — never by process name. + _markerDirectory = markerDirectory ?? WorkerLaunchMarker.DirectoryFor(_worker.PeCacheDir); + var swept = WorkerLaunchMarker.Sweep(_markerDirectory, _log); + if (swept.Count > 0) _log.Warning("Killed {Count} orphaned ScriptWorker process(es) at startup: {Pids}", swept.Count, swept); + var evalThreads = Math.Max(2, Environment.ProcessorCount / _worker.Shards); _shards = new WorkerShard[_worker.Shards]; for (var i = 0; i < _shards.Length; i++) @@ -84,11 +197,27 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, _shards[i] = NewShard(label, isolation: false, evalThreads, heartbeatInterval); } _isolation = NewShard("iso", isolation: true, evalThreads: 1, heartbeatInterval); + + var tick = maintenanceInterval ?? TimeSpan.FromSeconds(1); + _maintenance = new Timer(_ => _ = MaintainAsync(), null, tick, tick); + LiveRunners.TryAdd(this, 0); } /// The worker entry assembly this runner launches. public string WorkerPath { get; } + /// The overall state (the readiness check, the state gauge). + public ScriptRunnerState State => ComputeState(_shards.Select(s => s.DownFor).ToArray(), _isolation.DownFor, _worker.CircuitOpenAfter); + + /// A snapshot of the runner for diagnostics and health. + /// The status. + public ScriptRunnerStatus Status() => new( + State, + AllShards.Select(s => s.Status()).ToArray(), + SuspectKeys.Count, + _keys.Count(kv => kv.Value.QuarantineReason is not null), + _peCacheDirectory is not null); + /// How long a compile in the worker may take before the worker is killed. internal TimeSpan LoadTimeout { get; } @@ -98,10 +227,22 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, /// The isolation worker (tests). internal WorkerShard Isolation => _isolation; + /// The marker directory this runner writes launch markers to and sweeps at startup. + internal string MarkerDirectory => _markerDirectory; + + /// PE cache outcomes reported by this runner's workers (tests / diagnostics). + internal (long Hits, long Misses, long Rejected) PeCacheCounts => + (Interlocked.Read(ref _peCacheHits), Interlocked.Read(ref _peCacheMisses), Interlocked.Read(ref _peCacheRejected)); + + /// Test hook: runs inside the continuation that releases a key's single-flight slot, just before it does. + internal Action? SlotReleasingForTest { get; set; } + /// Scoped keys currently treated as suspects (tests / diagnostics). internal IReadOnlyCollection SuspectKeys => _keys.Where(kv => kv.Value.Suspect).Select(kv => kv.Key).ToArray(); + private IEnumerable AllShards => _shards.Append(_isolation); + /// The shard a key is placed on when it is not a suspect: a stable FNV-1a hash, so placement survives restarts. /// The scoped key. /// K. @@ -117,6 +258,23 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, return (int)(hash % (uint)shardCount); } + /// + /// The state rule: Unavailable when every shard has been down for at least , + /// Degraded when any shard or the isolation worker is down, Ready otherwise. + /// + /// Each regular shard's down time (null = up / never needed). + /// The isolation worker's. + /// The circuit threshold. + /// The state. + internal static ScriptRunnerState ComputeState(IReadOnlyList shardsDownFor, TimeSpan? isolationDownFor, TimeSpan circuitOpenAfter) + { + if (shardsDownFor.Count > 0 && shardsDownFor.All(d => d is { } down && down >= circuitOpenAfter)) + return ScriptRunnerState.Unavailable; + return shardsDownFor.Any(d => d is not null) || isolationDownFor is not null + ? ScriptRunnerState.Degraded + : ScriptRunnerState.Ready; + } + /// The default worker path, or the configured one. /// The worker options. /// The absolute worker path. @@ -137,7 +295,11 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, var sha = SHA256.HashData(Encoding.UTF8.GetBytes(invocation.Source)); var failureKey = InProcessScriptRunner.ScopedKey(invocation.Kind, Convert.ToHexString(sha)); - if (_compileFailures.TryGetValue(failureKey, out var knownFailure)) return ScriptRunOutcome.Failed(knownFailure); + if (_compileFailures.TryGetValue(failureKey, out var knownFailure)) + { + CompileCounter.Add(1, new KeyValuePair("cache", "negative")); + return ScriptRunOutcome.Failed(knownFailure); + } if (Interlocked.CompareExchange(ref state.InFlight, 1, 0) != 0) { @@ -147,19 +309,23 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, var handedOff = false; SemaphoreSlim? gate = null; + var shard = state.Suspect ? _isolation : _shards[ShardIndex(key, _shards.Length)]; + WorkerConnection? connection = null; try { state.ScriptLogger = invocation.ScriptLogger; - var shard = state.Suspect ? _isolation : _shards[ShardIndex(key, _shards.Length)]; + // Circuit open (design §5.3.6): answer at once instead of making every trigger wait out the grace. A zero + // grace still kicks a start that is due, so the shards keep retrying at the backoff cap. + var grace = State == ScriptRunnerState.Unavailable ? TimeSpan.Zero : _worker.UnavailableGrace; if (shard.IsolationGate is { } isolationGate) { - if (!await isolationGate.WaitAsync(_worker.UnavailableGrace, ct).ConfigureAwait(false)) - return ScriptRunOutcome.Unavailable(ScriptRunOutcome.RunnerUnavailableReason); + if (!await isolationGate.WaitAsync(grace, ct).ConfigureAwait(false)) + return Unavailable(ScriptRunOutcome.RunnerUnavailableReason, shard, null); gate = isolationGate; } - var connection = await shard.GetReadyAsync(_worker.UnavailableGrace, ct).ConfigureAwait(false); - if (connection is null) return ScriptRunOutcome.Unavailable(ScriptRunOutcome.RunnerUnavailableReason); + connection = await shard.GetReadyAsync(grace, ct).ConfigureAwait(false); + if (connection is null) return Unavailable(ScriptRunOutcome.RunnerUnavailableReason, shard, null); var paths = invocation.Inputs.Keys.Order(StringComparer.Ordinal).ToArray(); WorkerConnection.LoadEntry? entry; @@ -178,7 +344,8 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, connection.Kill(WorkerExitReason.KilledLoadTimeout, key); return ScriptRunOutcome.Failed($"compile failure: compile did not finish within {LoadTimeout.TotalSeconds:F0} s"); } - if (entry?.Outcome is not { } outcome) return ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerRestartedReason); + if (entry?.Outcome is not { } outcome) return Unavailable(ScriptRunOutcome.WorkerRestartedReason, RouteFor(key, state), connection); + NoteLoad(entry); if (!outcome.Ok) { var reason = outcome.Error?.Kind switch @@ -192,16 +359,19 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, return ScriptRunOutcome.Failed(reason); } + var submitted = Stopwatch.GetTimestamp(); var pending = connection.Submit(key, BuildRequest(key, entry, invocation)); ArmHardDeadline(connection, pending, invocation.Timeout + _worker.EffectiveHardGrace(invocation.Timeout)); // From here the in-flight slot (and the isolation gate) belong to the evaluation itself: they are released // when the worker answers or dies — not when this caller stops waiting at the soft deadline. That keeps a - // key single-flight between T and T + G, and the isolation worker strictly one-at-a-time. + // key single-flight between T and T + G, and the isolation worker strictly one-at-a-time. On a worker + // death, attribution (strike, quarantine, suspect) has already run when this continuation fires. handedOff = true; var heldGate = gate; _ = pending.Completion.ContinueWith(_ => { + SlotReleasingForTest?.Invoke(key); Interlocked.Exchange(ref state.InFlight, 0); heldGate?.Release(); }, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); @@ -221,8 +391,11 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, return ScriptRunOutcome.Failed($"{ScriptEvalFailureReasons.TimedOutPrefix}{invocation.Timeout.TotalSeconds:F1}s"); } - if (result is null) return ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerRestartedReason); - return Complete(key, state, result); + if (result is null) return Unavailable(ScriptRunOutcome.WorkerRestartedReason, RouteFor(key, state), connection); + var kindTag = new KeyValuePair("kind", KindTag(invocation.Kind)); + RoundTripHistogram.Record(Stopwatch.GetElapsedTime(submitted).TotalMicroseconds, kindTag); + EvalDurationHistogram.Record(result.RunMicros, kindTag); + return Complete(key, state, result, shard, connection); } catch (OperationCanceledException) when (ct.IsCancellationRequested) { @@ -251,7 +424,7 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, { if (key.StartsWith(prefix, StringComparison.Ordinal)) _compileFailures.TryRemove(key, out _); } - foreach (var shard in _shards.Append(_isolation)) + foreach (var shard in AllShards) shard.Current?.Unload(dropped); } @@ -270,12 +443,47 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, public async ValueTask DisposeAsync() { if (Interlocked.Exchange(ref _disposed, 1) != 0) return; - await Task.WhenAll(_shards.Append(_isolation).Select(s => s.DisposeAsync().AsTask())).ConfigureAwait(false); + LiveRunners.TryRemove(this, out _); + await _maintenance.DisposeAsync().ConfigureAwait(false); + await Task.WhenAll(AllShards.Select(s => s.DisposeAsync().AsTask())).ConfigureAwait(false); } /// Stops every worker (blocking). public void Dispose() => DisposeAsync().AsTask().GetAwaiter().GetResult(); + /// + /// The periodic pass: stop the isolation worker once it has been idle for + /// , and log the circuit opening / closing once each. + /// + /// A task for the pass. + internal async Task MaintainAsync() + { + if (Volatile.Read(ref _disposed) != 0 || Interlocked.Exchange(ref _maintaining, 1) != 0) return; + try + { + await _isolation.StopIfIdleAsync(_worker.IsolationIdleTimeout).ConfigureAwait(false); + + var open = State == ScriptRunnerState.Unavailable; + if (open && Interlocked.Exchange(ref _circuitLogged, 1) == 0) + { + _log.Error("Script runner circuit OPEN: every ScriptWorker shard has been down for at least {After}; every compiled script on this node reports {Reason} until a worker starts (retrying at the backoff cap)", + _worker.CircuitOpenAfter, ScriptRunOutcome.RunnerUnavailableReason); + } + else if (!open && Interlocked.Exchange(ref _circuitLogged, 0) == 1) + { + _log.Information("Script runner circuit closed: a ScriptWorker shard is serving again"); + } + } + catch (Exception ex) + { + _log.Warning(ex, "Script runner maintenance pass failed"); + } + finally + { + Interlocked.Exchange(ref _maintaining, 0); + } + } + private WorkerShard NewShard(string label, bool isolation, int evalThreads, TimeSpan heartbeatInterval) { var spec = new WorkerLaunchSpec( @@ -285,12 +493,45 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, EvalThreads: evalThreads, HardGrace: _worker.HardGrace ?? TimeSpan.FromSeconds(Math.Max(1, _options.RunTimeout.TotalSeconds)), HeartbeatInterval: heartbeatInterval, - HandshakeTimeout: TimeSpan.FromSeconds(15)); + HandshakeTimeout: TimeSpan.FromSeconds(15), + PeCacheDirectory: _peCacheDirectory, + PeCacheKey: _peCacheKey, + MarkerDirectory: _markerDirectory); return new WorkerShard(label, isolation, - ct => WorkerConnection.StartAsync(spec, _log, OnLogs, ct), + (onExiting, ct) => WorkerConnection.StartAsync(spec, _log, OnLogs, onExiting, ct), OnWorkerExit, _log); } + /// The shard a key's NEXT evaluation goes to (attribution may just have made it a suspect). + private WorkerShard RouteFor(string key, KeyState state) + => state.Suspect ? _isolation : _shards[ShardIndex(key, _shards.Length)]; + + private static ScriptRunOutcome Unavailable(string reason, WorkerShard shard, WorkerConnection? failed) + => ScriptRunOutcome.Unavailable(reason) with { Recovery = shard.WhenRecoveredAsync(failed) }; + + private void NoteLoad(WorkerConnection.LoadEntry entry) + { + // Counted once per compile, by the caller that waited for it (an already-loaded key is not a load). + if (Interlocked.Exchange(ref entry.Counted, 1) != 0 || entry.Outcome is not { Ok: true } outcome) return; + var tag = outcome.Cache switch + { + PeCache.Hit => "hit", + PeCache.Miss => "miss", + PeCache.Rejected => "rejected", + _ => "uncached", + }; + CompileCounter.Add(1, new KeyValuePair("cache", tag)); + switch (outcome.Cache) + { + case PeCache.Hit: Interlocked.Increment(ref _peCacheHits); break; + case PeCache.Miss: Interlocked.Increment(ref _peCacheMisses); break; + case PeCache.Rejected: + Interlocked.Increment(ref _peCacheRejected); + _log.Warning("Script {ScriptKey}: its PE cache entry failed verification and was deleted; recompiled", outcome.Key); + break; + } + } + private KeyState Admit(string key, string source) { while (true) @@ -355,7 +596,7 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, }, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); } - private ScriptRunOutcome Complete(string key, KeyState state, EvalResult result) + private ScriptRunOutcome Complete(string key, KeyState state, EvalResult result, WorkerShard shard, WorkerConnection connection) { // The script returned control (a value or a throw): the soft-timeout streak resets, as in process. Interlocked.Exchange(ref state.ConsecutiveTimeouts, 0); @@ -396,7 +637,7 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, ScriptError.Types.Kind.Threw => ScriptRunOutcome.Failed($"script threw: {result.Error.Message}"), ScriptError.Types.Kind.UnsupportedType => ScriptRunOutcome.Failed($"unsupported type: {result.Error.Message}"), ScriptError.Types.Kind.Busy => ScriptRunOutcome.Failed(WorkerBusyReason), - ScriptError.Types.Kind.NotLoaded => ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerRestartedReason), + ScriptError.Types.Kind.NotLoaded => Unavailable(ScriptRunOutcome.WorkerRestartedReason, shard, connection), _ => ScriptRunOutcome.Failed($"script failed: {result.Error.Message}"), }; default: @@ -415,7 +656,14 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, return true; } - private void OnWorkerExit(WorkerShard shard, WorkerExit exit) + /// + /// Attribution for one worker exit. Called synchronously from the worker's exit path BEFORE any evaluation or + /// compile it held is completed (see ), so its verdict is in place + /// before the affected keys' single-flight slots are released. + /// + /// The shard whose worker ended. + /// How it ended. + internal void OnWorkerExit(WorkerShard shard, WorkerExit exit) { switch (exit.Reason) { @@ -426,22 +674,25 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, KillStrike(culprit, "its compile did not finish and its worker was killed", "killed"); break; case WorkerExitReason.Crash or WorkerExitReason.KilledHeartbeat 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) { - // One evaluation at a time here: the crash is this key's (design §5.3.3). - foreach (var key in exit.InFlightKeys) - KillStrike(key, "it crashed the isolation worker", "isolated_crash"); + // One evaluation (or compile) at a time here: the crash is this key's (design §5.3.3). + foreach (var key in implicated) + KillStrike(key, exit.InFlightKeys.Contains(key) ? "it crashed the isolation worker" : "its compile crashed the isolation worker", + "isolated_crash"); } - else if (exit.InFlightKeys.Count > 0) + else if (implicated.Length > 0) { - foreach (var key in exit.InFlightKeys) + foreach (var key in implicated) { if (!_keys.TryGetValue(key, out var state)) continue; state.Suspect = true; Interlocked.Exchange(ref state.ProbationRuns, 0); } _log.Warning("ScriptWorker[{Shard}] ended ({Reason}) with no deadline crossed; {Count} script(s) in flight move to the isolation worker: {Suspects}", - shard.Label, exit.Reason, exit.InFlightKeys.Count, string.Join(", ", exit.InFlightKeys)); + shard.Label, exit.Reason, implicated.Length, string.Join(", ", implicated)); } break; } @@ -491,6 +742,18 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, _log.Warning("ScriptWorker dropped {Dropped} script log line(s): a script is logging faster than the pipe drains", batch.Dropped); } + private static string KindTag(ScriptContextKind kind) => kind switch + { + ScriptContextKind.VirtualTag => "virtual_tag", + ScriptContextKind.Calculation => "calculation", + ScriptContextKind.AlarmPredicate => "alarm", + _ => "other", + }; + + private static IEnumerable> ShardMeasurements(Func value) + => LiveRunners.Keys.SelectMany(r => r.AllShards.Select(s => s.Status())) + .Select(s => new Measurement(value(s), new KeyValuePair("shard", s.Label))); + private sealed class KeyState(string source) { public readonly string Source = source; 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 4b456f00..063fa6d0 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 @@ -33,6 +33,9 @@ internal enum WorkerExitReason /// The runner is shutting down. Disposed, + + /// The host stopped an idle worker (the isolation worker after IsolationIdleTimeout); not a failure. + Idle, } /// The end of one worker process. @@ -40,7 +43,16 @@ internal enum WorkerExitReason /// The scoped key the host blamed, for a deadline or load kill. /// The OS exit code, when known. /// The scoped keys that had an evaluation outstanding when it ended. -internal sealed record WorkerExit(WorkerExitReason Reason, string? CulpritKey, int? ExitCode, IReadOnlyList InFlightKeys); +/// The scoped keys whose compile (load) was outstanding when it ended. +internal sealed record WorkerExit(WorkerExitReason Reason, string? CulpritKey, int? ExitCode, IReadOnlyList InFlightKeys, + IReadOnlyList LoadingKeys) +{ + /// An exit with no load in flight. + public WorkerExit(WorkerExitReason reason, string? culpritKey, int? exitCode, IReadOnlyList inFlightKeys) + : this(reason, culpritKey, exitCode, inFlightKeys, []) + { + } +} /// What the host needs to launch one worker. /// The shard label (0, 1, … or iso). @@ -50,6 +62,9 @@ internal sealed record WorkerExit(WorkerExitReason Reason, string? CulpritKey, i /// The worker watchdog's grace past the soft deadline. /// How often the worker sends a heartbeat. /// How long a launch may take to reach Ready. +/// The compiled-PE cache directory handed to the worker; null = no cache. +/// The cache's HMAC key (with ). +/// Where the launch marker for the orphan sweep is written; null = none. internal sealed record WorkerLaunchSpec( string Shard, string WorkerPath, @@ -57,7 +72,10 @@ internal sealed record WorkerLaunchSpec( int EvalThreads, TimeSpan HardGrace, TimeSpan HeartbeatInterval, - TimeSpan HandshakeTimeout); + TimeSpan HandshakeTimeout, + string? PeCacheDirectory = null, + byte[]? PeCacheKey = null, + string? MarkerDirectory = null); /// /// One live ScriptWorker process and its pipe (lmxopcua#780, design §3–§6): launch with a cleared, allow-listed @@ -85,12 +103,17 @@ internal sealed class WorkerConnection : IAsyncDisposable private readonly Action _onLogs; private readonly Channel _outbound = Channel.CreateUnbounded(new UnboundedChannelOptions { SingleReader = true }); private readonly ConcurrentDictionary _pending = new(); - private readonly ConcurrentDictionary> _loads = new(); + private readonly ConcurrentDictionary Completion)> _loads = new(); private readonly ConcurrentDictionary _loaded = new(StringComparer.Ordinal); private readonly TaskCompletionSource _exited = new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly CancellationTokenSource _stopping = new(); private readonly TimeSpan _heartbeatInterval; + private readonly Action _onExiting; + private readonly string? _markerPath; private TaskCompletionSource? _diag; + private Heartbeat? _lastHeartbeat; + private long _lastUsedTimestamp; + private int _finished; private (WorkerExitReason Reason, string? Culprit)? _killedFor; private long _lastFrameTimestamp; private long _seq; @@ -100,8 +123,10 @@ internal sealed class WorkerConnection : IAsyncDisposable private int _disposed; private WorkerConnection(string shard, Process process, NamedPipeServerStream pipe, IDisposable? job, string workDir, - ILogger log, Action onLogs, TimeSpan heartbeatInterval) + ILogger log, Action onLogs, TimeSpan heartbeatInterval, Action onExiting, string? markerPath) { + _onExiting = onExiting; + _markerPath = markerPath; Shard = shard; _process = process; _pipe = pipe; @@ -131,13 +156,29 @@ internal sealed class WorkerConnection : IAsyncDisposable /// Evaluations currently outstanding on this worker. public int InFlightCount => _pending.Count; + /// Compiles currently outstanding on this worker. + public int LoadingCount => _loads.Count; + + /// The worker's last heartbeat (its in-flight / queued counts and heap), or null before the first. + public Heartbeat? LastHeartbeat => Volatile.Read(ref _lastHeartbeat); + + /// of the last load or evaluation submitted (or of Ready). + public long LastUsedTimestamp => Interlocked.Read(ref _lastUsedTimestamp); + /// Launches a worker and completes the handshake. /// What to launch. /// Host diagnostics (stderr lines, exits). /// Receives the worker's forwarded ctx.Logger batches. + /// + /// Attribution, called synchronously when the worker has ended and BEFORE any outstanding evaluation or load is + /// completed or is signalled. A kill strike or suspect mark recorded here is therefore in + /// place before any caller can see its evaluation end and trigger the script again (the quarantine race, + /// lmxopcua#780). + /// /// Cancellation. /// A ready connection. - public static async Task StartAsync(WorkerLaunchSpec spec, ILogger log, Action onLogs, CancellationToken ct) + public static async Task StartAsync(WorkerLaunchSpec spec, ILogger log, Action onLogs, + Action onExiting, CancellationToken ct) { var pipeName = "otopcua-sw-" + Convert.ToHexString(RandomNumberGenerator.GetBytes(12)).ToLowerInvariant(); var nonce = Convert.ToHexString(RandomNumberGenerator.GetBytes(32)); @@ -149,11 +190,15 @@ internal sealed class WorkerConnection : IAsyncDisposable PipeTransmissionMode.Byte, PipeOptions.Asynchronous | PipeOptions.CurrentUserOnly); Process? process = null; IDisposable? job = null; + string? markerPath = null; try { var psi = BuildStartInfo(spec, pipeName, nonce, workDir); process = new Process { StartInfo = psi, EnableRaisingEvents = true }; if (!process.Start()) throw new InvalidOperationException("the worker process did not start"); + // Before anything else can fail: a host that dies from here on leaves a marker the next start's orphan + // sweep can match to exactly this process (design §3.3). + if (spec.MarkerDirectory is { } markerDir) markerPath = WorkerLaunchMarker.Write(markerDir, process, pipeName, log); if (OperatingSystem.IsWindows()) { // Native and JIT memory backstop above the managed heap limit. @@ -190,9 +235,15 @@ internal sealed class WorkerConnection : IAsyncDisposable HeartbeatIntervalMs = (uint)spec.HeartbeatInterval.TotalMilliseconds, MaxLogLines = 1000, }; + if (spec.PeCacheDirectory is { } cacheDir && spec.PeCacheKey is { } cacheKey) + { + hello.PeCacheDir = cacheDir; + hello.PeCacheKey = Google.Protobuf.ByteString.CopyFrom(cacheKey); + } await HandshakeAsync(pipe, hello, nonce, process.Id, handshakeCts.Token).ConfigureAwait(false); - var connection = new WorkerConnection(spec.Shard, process, pipe, job, workDir, log, onLogs, spec.HeartbeatInterval); + var connection = new WorkerConnection(spec.Shard, process, pipe, job, workDir, log, onLogs, spec.HeartbeatInterval, + onExiting, markerPath); connection.Begin(); return connection; } @@ -203,6 +254,7 @@ internal sealed class WorkerConnection : IAsyncDisposable process?.Dispose(); await pipe.DisposeAsync().ConfigureAwait(false); TryDeleteDirectory(workDir); + WorkerLaunchMarker.TryDelete(markerPath); throw; } } @@ -308,6 +360,7 @@ internal sealed class WorkerConnection : IAsyncDisposable { ReadyTimestamp = Stopwatch.GetTimestamp(); Volatile.Write(ref _lastFrameTimestamp, ReadyTimestamp); + Interlocked.Exchange(ref _lastUsedTimestamp, ReadyTimestamp); _ = Task.Run(ReadLoopAsync); _ = Task.Run(WriteLoopAsync); _ = Task.Run(HeartbeatMonitorAsync); @@ -329,9 +382,10 @@ internal sealed class WorkerConnection : IAsyncDisposable var entry = new LoadEntry(hashHex, inputPaths); _loaded[key] = entry; + Interlocked.Exchange(ref _lastUsedTimestamp, Stopwatch.GetTimestamp()); var loadId = (ulong)Interlocked.Increment(ref _nextLoadId); var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - _loads[loadId] = tcs; + _loads[loadId] = (key, tcs); var load = new LoadScripts { LoadId = loadId }; var script = new ScriptSource { Key = key, Kind = kind, Source = source, SourceSha256 = Google.Protobuf.ByteString.CopyFrom(sourceHash) }; script.InputPaths.AddRange(inputPaths); @@ -372,6 +426,7 @@ internal sealed class WorkerConnection : IAsyncDisposable public PendingEval Submit(string key, EvalRequest request) { request.EvalId = (ulong)Interlocked.Increment(ref _nextEvalId); + Interlocked.Exchange(ref _lastUsedTimestamp, Stopwatch.GetTimestamp()); var pending = new PendingEval(key, request.EvalId); _pending[request.EvalId] = pending; var batch = new EvalBatch(); @@ -423,13 +478,18 @@ internal sealed class WorkerConnection : IAsyncDisposable /// Asks the worker to stop, waits briefly, then kills it. /// A task that completes when the worker has ended. - public async ValueTask DisposeAsync() + public ValueTask DisposeAsync() => StopAsync(WorkerExitReason.Disposed); + + /// As , recording as the exit reason. + /// Why the host is stopping it ( or ). + /// A task that completes when the worker has ended. + public async ValueTask StopAsync(WorkerExitReason reason) { // Both the shard (on exit) and the runner (on shutdown) dispose a connection; only the first does the work. if (Interlocked.Exchange(ref _disposed, 1) != 0) return; if (!_exited.Task.IsCompleted) { - lock (_pending) { _killedFor ??= (WorkerExitReason.Disposed, null); } + lock (_pending) { _killedFor ??= (reason, null); } Send(new Envelope { Shutdown = new Shutdown() }); try { @@ -447,6 +507,7 @@ internal sealed class WorkerConnection : IAsyncDisposable _job?.Dispose(); _process.Dispose(); TryDeleteDirectory(_workDir); + WorkerLaunchMarker.TryDelete(_markerPath); } /// Test hook: closes the host end of the pipe without touching the process, as a dying host would. @@ -521,7 +582,7 @@ internal sealed class WorkerConnection : IAsyncDisposable } break; case Envelope.BodyOneofCase.LoadResult: - if (_loads.TryRemove(frame.LoadResult.LoadId, out var load)) load.TrySetResult(frame.LoadResult); + if (_loads.TryRemove(frame.LoadResult.LoadId, out var load)) load.Completion.TrySetResult(frame.LoadResult); else throw new ScriptWorkerProtocolException($"load result for unknown load id {frame.LoadResult.LoadId}"); break; case Envelope.BodyOneofCase.Logs: @@ -529,6 +590,7 @@ internal sealed class WorkerConnection : IAsyncDisposable catch (Exception ex) { _log.Warning(ex, "ScriptWorker[{Shard}] log forwarding threw", Shard); } break; case Envelope.BodyOneofCase.Heartbeat: + Volatile.Write(ref _lastHeartbeat, frame.Heartbeat); break; case Envelope.BodyOneofCase.DiagReply: Volatile.Read(ref _diag)?.TrySetResult(frame.DiagReply); @@ -604,6 +666,11 @@ internal sealed class WorkerConnection : IAsyncDisposable private void Finish() { + // The reader (on EOF) and the process watcher both get here; only the first reports. + if (Interlocked.Exchange(ref _finished, 1) != 0) return; + // No new frame can be queued from here on, so every Submit / load either fails its Send (and completes itself) + // or was registered before this point and is completed below. + _outbound.Writer.TryComplete(); int? code = null; try { if (_process.HasExited) code = _process.ExitCode; } catch (InvalidOperationException) { } @@ -611,13 +678,21 @@ internal sealed class WorkerConnection : IAsyncDisposable lock (_pending) { killed = _killedFor; } var reason = killed?.Reason ?? (code == 0 ? WorkerExitReason.Clean : WorkerExitReason.Crash); var inFlight = _pending.Values.Select(p => p.Key).Distinct(StringComparer.Ordinal).ToArray(); - if (!_exited.TrySetResult(new WorkerExit(reason, killed?.Culprit, code, inFlight))) return; + var loading = _loads.Values.Select(l => l.Key).Distinct(StringComparer.Ordinal).ToArray(); + var exit = new WorkerExit(reason, killed?.Culprit, code, inFlight, loading); + + // Attribution FIRST, synchronously: a kill strike / quarantine / suspect mark must be in place before any + // evaluation below completes, because completing it releases the key's single-flight slot and the very next + // trigger is admitted against whatever the quarantine table says at that instant (lmxopcua#780 race fix). + try { _onExiting(exit); } + catch (Exception ex) { _log.Error(ex, "ScriptWorker[{Shard}] exit attribution threw", Shard); } + _exited.TrySetResult(exit); // Every outstanding evaluation and load ends now; null tells the waiter "the worker went away". foreach (var id in _pending.Keys.ToArray()) if (_pending.TryRemove(id, out var pending)) pending.Complete(null); foreach (var id in _loads.Keys.ToArray()) - if (_loads.TryRemove(id, out var load)) load.TrySetCanceled(); + if (_loads.TryRemove(id, out var load)) load.Completion.TrySetCanceled(); Volatile.Read(ref _diag)?.TrySetCanceled(); _outbound.Writer.TryComplete(); } @@ -638,6 +713,9 @@ internal sealed class WorkerConnection : IAsyncDisposable { private readonly TaskCompletionSource _ready = new(TaskCreationOptions.RunContinuationsAsynchronously); + /// Set once the runner has counted this load in its metrics. + public int Counted; + public LoadEntry(string sourceHash, IReadOnlyList inputPaths) { SourceHash = sourceHash; diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerLaunchMarker.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerLaunchMarker.cs new file mode 100644 index 00000000..d794138e --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerLaunchMarker.cs @@ -0,0 +1,166 @@ +using System.Diagnostics; +using System.Security.Cryptography; +using System.Text; +using System.Text.Json; +using Serilog; + +namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess; + +/// +/// The launch marker and the startup orphan sweep (lmxopcua#780 design §3.3). Every worker this host launches gets a +/// marker file in this host instance's marker directory naming the worker's pid, its start time and process name, +/// and the launching host's pid and start time. A host that dies uncleanly leaves its markers behind; the next start +/// of the same instance kills exactly the processes they name — and nothing else. +/// +/// +/// +/// Never "every process with this name". mxaccessgw's OrphanWorkerTerminator killed a live +/// gateway's workers when a second gateway started. Here a marker is acted on only when (1) the host that wrote +/// it is gone — no process with its pid and start time — and (2) a process with the marker's pid, the same +/// start time and the same process name is still running. A pid that was reused by an unrelated process +/// fails (2) and is left alone; a marker written by a live host (a second instance sharing the directory, or a +/// sibling runner in this very process) fails (1) and is left alone. +/// +/// +/// On Windows the Job Object (KILL_ON_JOB_CLOSE) already takes the workers down with the host, and on +/// Linux the worker exits on pipe EOF or when its parent pid disappears. The sweep is the backstop for a worker +/// wedged so hard it noticed neither. +/// +/// +internal static class WorkerLaunchMarker +{ + private const string Extension = ".worker"; + + /// Start times are compared with this tolerance (platforms report them at different precisions). + private static readonly TimeSpan StartTimeTolerance = TimeSpan.FromSeconds(2); + + /// The marker directory of this host instance: under the PE cache when there is one, else under temp. + /// The configured Scripting:Worker:PeCacheDir, or null. + /// The directory (not created). + public static string DirectoryFor(string? peCacheDir) + { + if (!string.IsNullOrWhiteSpace(peCacheDir)) return Path.Combine(peCacheDir, "workers"); + // One directory per installation (base directory), so two installs on a box do not see each other's markers. + var install = Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(AppContext.BaseDirectory)))[..16].ToLowerInvariant(); + return Path.Combine(Path.GetTempPath(), "otopcua-scriptworker", "hosts", install); + } + + /// Writes the marker for a just-started worker. Best effort: a failure is logged, never fatal. + /// The marker directory. + /// The started worker. + /// Its pipe name (diagnostics). + /// Diagnostics. + /// The marker path, or null when none was written. + public static string? Write(string directory, Process worker, string pipeName, ILogger log) + { + try + { + Directory.CreateDirectory(directory); + using var host = Process.GetCurrentProcess(); + var record = new MarkerRecord( + worker.Id, worker.StartTime.ToUniversalTime().Ticks, SafeName(worker), + host.Id, host.StartTime.ToUniversalTime().Ticks, pipeName); + var path = Path.Combine(directory, worker.Id.ToString(System.Globalization.CultureInfo.InvariantCulture) + Extension); + File.WriteAllText(path, JsonSerializer.Serialize(record)); + return path; + } + catch (Exception ex) when (ex is IOException or UnauthorizedAccessException or InvalidOperationException + or System.ComponentModel.Win32Exception or NotSupportedException) + { + log.Warning(ex, "Could not write the ScriptWorker launch marker under {Directory}; the startup orphan sweep cannot see pid {Pid}", + directory, SafeId(worker)); + return null; + } + } + + /// Deletes a marker (worker ended cleanly). + /// The marker path, or null. + public static void TryDelete(string? path) + { + if (path is null) return; + try { File.Delete(path); } catch (Exception ex) when (ex is IOException or UnauthorizedAccessException) { } + } + + /// Kills the workers a dead host of this instance left behind; returns the pids killed. + /// The marker directory. + /// Diagnostics. + /// The pids killed. + public static IReadOnlyList Sweep(string directory, ILogger log) + { + var killed = new List(); + if (!Directory.Exists(directory)) return killed; + IEnumerable files; + try { files = Directory.EnumerateFiles(directory, "*" + Extension).ToArray(); } + catch (Exception ex) when (ex is IOException or UnauthorizedAccessException) { return killed; } + + foreach (var path in files) + { + MarkerRecord? record; + try { record = JsonSerializer.Deserialize(File.ReadAllText(path)); } + catch (Exception ex) when (ex is IOException or UnauthorizedAccessException or JsonException) + { + record = null; + } + if (record is null) + { + TryDelete(path); + continue; + } + + // (1) A live host owns its markers — another instance sharing the directory, or a sibling runner here. + if (IsSameProcess(record.HostPid, record.HostStartTicks, expectedName: null)) continue; + + // (2) Only the exact process the dead host launched: same pid, same start time, same image name. + if (IsSameProcess(record.Pid, record.StartTicks, record.ProcessName)) + { + try + { + using var orphan = Process.GetProcessById(record.Pid); + orphan.Kill(entireProcessTree: true); + killed.Add(record.Pid); + log.Warning("Killed orphaned ScriptWorker pid {Pid}, left behind by host pid {HostPid} (pipe {Pipe})", + record.Pid, record.HostPid, record.PipeName); + } + catch (Exception ex) when (ex is ArgumentException or InvalidOperationException or System.ComponentModel.Win32Exception) + { + log.Warning(ex, "Could not kill orphaned ScriptWorker pid {Pid}", record.Pid); + continue; // keep the marker: the next start retries + } + } + TryDelete(path); + } + return killed; + } + + private static bool IsSameProcess(int pid, long startTicks, string? expectedName) + { + try + { + using var process = Process.GetProcessById(pid); + if (process.HasExited) return false; + var started = process.StartTime.ToUniversalTime().Ticks; + if (Math.Abs(started - startTicks) > StartTimeTolerance.Ticks) return false; + return expectedName is null || string.Equals(SafeName(process), expectedName, StringComparison.Ordinal); + } + catch (Exception ex) when (ex is ArgumentException or InvalidOperationException or System.ComponentModel.Win32Exception + or NotSupportedException) + { + return false; + } + } + + private static string SafeName(Process process) + { + try { return process.ProcessName; } + catch (InvalidOperationException) { return string.Empty; } + } + + private static int SafeId(Process process) + { + try { return process.Id; } + catch (InvalidOperationException) { return 0; } + } + + /// The marker's content. + internal sealed record MarkerRecord(int Pid, long StartTicks, string ProcessName, int HostPid, long HostStartTicks, string PipeName); +} 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 32febbab..7f432eeb 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 @@ -6,13 +6,39 @@ using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.V1; namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess; +/// A point-in-time view of one shard (diagnostics, the readiness check, the runner-state gauge). +/// The shard label (0…K-1, or iso). +/// Whether it is the isolation worker. +/// A worker is live and serving. +/// How long it has been down after a failure; null when up or never needed. +/// The live worker's pid. +/// Start failures and crashes since the last healthy period. +/// Evaluations outstanding on the live worker. +/// Evaluations queued in the worker (its last heartbeat). +/// The worker's managed heap (its last heartbeat). +public sealed record ScriptWorkerShardStatus( + string Label, + bool IsIsolation, + bool Up, + TimeSpan? DownFor, + int? Pid, + int ConsecutiveFailures, + int InFlight, + int Queued, + long HeapBytes); + /// /// Supervises one shard's worker process (design §3.3, §5.3.5): starts it lazily on first use, reports its end to -/// the runner (for attribution), and restarts it after an exponential backoff — 250 ms doubling to a 30 s cap with -/// ±20 % jitter, reset once a worker has stayed up for 60 s. It never runs a script in the host when the worker -/// cannot start (fail closed, §14 #4): returns null and the caller reports -/// BadResourceUnavailable. +/// the runner (for attribution, synchronously — see ), and restarts it after +/// an exponential backoff — 250 ms doubling to a 30 s cap with ±20 % jitter, reset once a worker has stayed up for +/// 60 s. It never runs a script in the host when the worker cannot start (fail closed, §14 #4): +/// returns null and the caller reports BadResourceUnavailable. /// +/// +/// A caller that got Unavailable waits on ; that both tells it when to +/// re-evaluate and makes the shard restart proactively (with backoff) instead of waiting for the next trigger, so +/// victims of a crash are re-evaluated on the shard's return (design §9). +/// internal sealed class WorkerShard : IAsyncDisposable { internal static readonly TimeSpan InitialBackoff = TimeSpan.FromMilliseconds(250); @@ -24,20 +50,24 @@ 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 | disposed)."); + description: "ScriptWorker processes ended (tags: shard, reason = clean | killed_deadline | killed_heartbeat | killed_load | crash | protocol | idle | disposed)."); - private readonly Func> _launch; + private readonly Func, CancellationToken, Task> _launch; private readonly Action _onExit; private readonly ILogger _log; private readonly object _gate = new(); private readonly Random _jitter = new(); + private readonly CancellationTokenSource _disposing = new(); private WorkerConnection? _current; private Task? _starting; + private TaskCompletionSource? _recovered; + private bool _recovering; private long _backoffUntil; + private long _downSince; private int _consecutiveFailures; private bool _disposed; - public WorkerShard(string label, bool isolation, Func> launch, + public WorkerShard(string label, bool isolation, Func, CancellationToken, Task> launch, Action onExit, ILogger log) { Label = label; @@ -70,9 +100,40 @@ internal sealed class WorkerShard : IAsyncDisposable get { lock (_gate) return _consecutiveFailures; } } + /// How long the shard has been down after a failure; null when it is up or was never needed. + public TimeSpan? DownFor + { + get + { + lock (_gate) + { + if (_current is { IsAlive: true } || _downSince == 0) return null; + return Stopwatch.GetElapsedTime(_downSince); + } + } + } + + /// A snapshot for diagnostics and health. + /// The status. + public ScriptWorkerShardStatus Status() + { + WorkerConnection? live; + int failures; + TimeSpan? downFor; + lock (_gate) + { + live = _current is { IsAlive: true } c ? c : null; + failures = _consecutiveFailures; + downFor = live is null && _downSince != 0 ? Stopwatch.GetElapsedTime(_downSince) : null; + } + var heartbeat = live?.LastHeartbeat; + return new ScriptWorkerShardStatus(Label, IsIsolation, live is not null, downFor, live?.Pid, failures, + live?.InFlightCount ?? 0, (int)(heartbeat?.Queued ?? 0), heartbeat?.HeapBytes ?? 0); + } + /// /// The ready worker, starting one if needed. Waits at most — through a start in progress - /// or a backoff — and returns null if no worker is ready by then. + /// or a backoff — and returns null if no worker is ready by then. A zero grace still kicks a start that is due. /// /// How long to wait. /// Cancellation. @@ -125,18 +186,123 @@ internal sealed class WorkerShard : IAsyncDisposable } } + /// + /// Completes when this shard has a live worker other than . Starts a background + /// recovery (start with backoff, retried at the cap) if none is live, so the caller does not depend on a later + /// trigger to bring the shard back. + /// + /// The worker the caller's evaluation died on, if any. + /// A task that completes on recovery (or when the shard is disposed). + public Task WhenRecoveredAsync(WorkerConnection? failed) + { + lock (_gate) + { + if (_disposed) return Task.CompletedTask; + if (_current is { IsAlive: true } live) + { + // Still the same worker (it answered NOT_LOADED, or the caller lost a race with a restart that has + // already happened): a short pause, so a seam cannot spin on a key the worker keeps refusing. + return ReferenceEquals(live, failed) ? Task.Delay(InitialBackoff) : Task.CompletedTask; + } + _recovered ??= new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var task = _recovered.Task; + EnsureRecoveringLocked(); + return task; + } + } + + private void EnsureRecoveringLocked() + { + if (_recovering || _disposed) return; + _recovering = true; + _ = Task.Run(RecoverAsync); + } + + /// Stops the live worker if it has had nothing to do for (the isolation worker). + /// The idle threshold. + /// True when a worker was stopped. + public async Task StopIfIdleAsync(TimeSpan idle) + { + WorkerConnection? victim; + // Holding the isolation gate means no evaluation can pick this worker up while it is being stopped. + var gate = IsolationGate; + try + { + if (gate is not null && !gate.Wait(0)) return false; + } + catch (ObjectDisposedException) + { + return false; + } + try + { + lock (_gate) + { + if (_disposed || _current is not { IsAlive: true } live) return false; + if (live.InFlightCount > 0 || live.LoadingCount > 0) return false; + if (Stopwatch.GetElapsedTime(live.LastUsedTimestamp) < idle) return false; + victim = live; + _current = null; + } + _log.Information("ScriptWorker[{Shard}] pid {Pid} idle for {Idle}; stopping it (it restarts on demand)", Label, victim.Pid, idle); + await victim.StopAsync(WorkerExitReason.Idle).ConfigureAwait(false); + return true; + } + finally + { + gate?.Release(); + } + } + + private async Task RecoverAsync() + { + try + { + while (true) + { + // A long grace: GetReadyAsync waits out the backoff itself and returns null only on a failed start. + var live = await GetReadyAsync(MaxBackoff * 2, _disposing.Token).ConfigureAwait(false); + lock (_gate) + { + if (_disposed || live is not null) + { + _recovering = false; + TakeRecoveredLocked()?.TrySetResult(); + return; + } + } + } + } + catch (OperationCanceledException) + { + lock (_gate) + { + _recovering = false; + TakeRecoveredLocked()?.TrySetResult(); + } + } + } + + private TaskCompletionSource? TakeRecoveredLocked() + { + var tcs = _recovered; + _recovered = null; + return tcs; + } + private async Task StartCoreAsync() { await Task.Yield(); WorkerConnection? connection = null; try { - connection = await _launch(CancellationToken.None).ConfigureAwait(false); + connection = await _launch(Attribute, CancellationToken.None).ConfigureAwait(false); StartsCounter.Add(1, new KeyValuePair("shard", Label)); _log.Information("ScriptWorker[{Shard}] started, pid {Pid}", Label, connection.Pid); var started = connection; _ = started.Exited.ContinueWith(t => OnExited(started, t.Result), CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously | TaskContinuationOptions.OnlyOnRanToCompletion, TaskScheduler.Default); + TaskCompletionSource? recovered; lock (_gate) { _starting = null; @@ -146,7 +312,10 @@ internal sealed class WorkerShard : IAsyncDisposable return null; } _current = started; + _downSince = 0; + recovered = TakeRecoveredLocked(); } + recovered?.TrySetResult(); return started; } catch (Exception ex) @@ -155,6 +324,7 @@ internal sealed class WorkerShard : IAsyncDisposable lock (_gate) { _starting = null; + MarkDownLocked(); ScheduleBackoffLocked(); } if (connection is not null) await connection.DisposeAsync().ConfigureAwait(false); @@ -162,29 +332,44 @@ internal sealed class WorkerShard : IAsyncDisposable } } + /// Synchronous attribution, from 's exit path, before any evaluation completes. + private void Attribute(WorkerExit exit) + { + try { _onExit(this, exit); } + catch (Exception ex) { _log.Error(ex, "ScriptWorker[{Shard}] exit handler threw", Label); } + } + private void OnExited(WorkerConnection connection, WorkerExit exit) { ExitsCounter.Add(1, new KeyValuePair("shard", Label), new KeyValuePair("reason", MetricReason(exit.Reason))); + var expected = exit.Reason is WorkerExitReason.Disposed or WorkerExitReason.Idle; lock (_gate) { if (ReferenceEquals(_current, connection)) _current = null; - if (!_disposed) + if (!_disposed && !expected) { if (Stopwatch.GetElapsedTime(connection.ReadyTimestamp) >= HealthyResetAfter) _consecutiveFailures = 0; + MarkDownLocked(); ScheduleBackoffLocked(); + // A worker that was serving and died comes back by itself (after the backoff), so the scripts placed + // on it find it warm and its victims are re-evaluated on its return (design §9). + EnsureRecoveringLocked(); } } - if (exit.Reason != WorkerExitReason.Disposed) + if (!expected) { - _log.Warning("ScriptWorker[{Shard}] pid {Pid} ended: {Reason} (exit code {ExitCode}, culprit {Culprit}, {InFlight} evaluation(s) in flight)", - Label, connection.Pid, exit.Reason, exit.ExitCode, exit.CulpritKey ?? "none", exit.InFlightKeys.Count); + _log.Warning("ScriptWorker[{Shard}] pid {Pid} ended: {Reason} (exit code {ExitCode}, culprit {Culprit}, {InFlight} evaluation(s) and {Loading} compile(s) in flight)", + Label, connection.Pid, exit.Reason, exit.ExitCode, exit.CulpritKey ?? "none", exit.InFlightKeys.Count, exit.LoadingKeys.Count); } - try { _onExit(this, exit); } - catch (Exception ex) { _log.Error(ex, "ScriptWorker[{Shard}] exit handler threw", Label); } _ = connection.DisposeAsync().AsTask(); } + private void MarkDownLocked() + { + if (_downSince == 0) _downSince = Stopwatch.GetTimestamp(); + } + private void ScheduleBackoffLocked() { _consecutiveFailures++; @@ -212,6 +397,7 @@ internal sealed class WorkerShard : IAsyncDisposable WorkerExitReason.KilledLoadTimeout => "killed_load", WorkerExitReason.ProtocolFault => "protocol", WorkerExitReason.Disposed => "disposed", + WorkerExitReason.Idle => "idle", _ => "crash", }; @@ -221,13 +407,18 @@ internal sealed class WorkerShard : IAsyncDisposable { WorkerConnection? current; Task? starting; + TaskCompletionSource? recovered; lock (_gate) { + if (_disposed) return; _disposed = true; current = _current; starting = _starting; _current = null; + recovered = TakeRecoveredLocked(); } + recovered?.TrySetResult(); + _disposing.Cancel(); if (starting is not null) { try { current ??= await starting.WaitAsync(TimeSpan.FromSeconds(15)).ConfigureAwait(false); } catch { /* shutting down */ } diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/Protos/script_worker.proto b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/Protos/script_worker.proto index f27c40b5..4c5573f0 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/Protos/script_worker.proto +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/Protos/script_worker.proto @@ -37,6 +37,8 @@ message Hello { uint32 hard_grace_ms = 5; // worker watchdog: fail fast past soft + grace (+ margin) uint32 heartbeat_interval_ms = 6; uint32 max_log_lines = 7; // outbound log buffer cap + string pe_cache_dir = 8; // the node's compiled-PE cache (design §7); empty = no cache + bytes pe_cache_key = 9; // its HMAC key; the worker loads only entries that verify under it } message Ready { @@ -69,6 +71,14 @@ message LoadOutcome { string key = 1; bool ok = 2; ScriptError error = 3; // COMPILE or SANDBOX when !ok + PeCache cache = 4; // how the PE cache served this load +} + +enum PeCache { + PE_CACHE_NONE = 0; // no cache configured (or the load failed before it was consulted) + PE_CACHE_HIT = 1; // loaded from a verified entry; Roslyn not entered + PE_CACHE_MISS = 2; // compiled; the entry was written + PE_CACHE_REJECTED = 3; // an entry failed verification, was deleted, and the source was compiled } message LoadResult { diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ScriptPeCacheStore.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ScriptPeCacheStore.cs new file mode 100644 index 00000000..edb0de1c --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ScriptPeCacheStore.cs @@ -0,0 +1,292 @@ +using System.Security.Cryptography; +using System.Text; + +namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol; + +/// What a found. +public enum PeCacheLookup +{ + /// No entry for the id. + Miss = 0, + + /// An entry whose HMAC verified; its PE bytes are returned. + Hit = 1, + + /// + /// An entry that exists but did not verify — truncated, edited, written under another key, or renamed from + /// another id. It has been deleted; the caller recompiles. + /// + Rejected = 2, +} + +/// +/// The node-local compiled-PE cache (lmxopcua#780 design §7, owner decision §14 #8): content-addressed files under +/// Scripting:Worker:PeCacheDir, each authenticated with HMAC-SHA256 under a per-node random key, so a worker +/// restart loads its scripts' PE images instead of re-entering Roslyn — and never loads bytes it did not write. +/// +/// +/// +/// Layout. <dir>/pe-cache.key (32 random bytes, created once by the host) and +/// <dir>/pe/<id>.pe, one file per entry: an 8-byte magic, the 32-byte tag, then the PE. +/// Identity. is SHA-256 over the compile fingerprint (Roslyn version, the +/// sandbox and analyzer build, the reference set, the runtime), the context kind and the source. A compiler or +/// analyzer change therefore misses every old entry; it can never be served an image a newer sandbox rule +/// would have refused. +/// Authentication. The tag covers the magic, the id and the PE, so an entry copied to another id +/// is rejected as well as an edited or truncated one. Loading unauthenticated PE bytes would be code execution +/// outside the analyzer; a rejected entry is deleted and the source recompiled. +/// Key. Readable only by the host account: on Linux/macOS the directory is 0700 and the key file +/// 0600. On Windows the file inherits the directory's ACL, which must be the installer's (lmxopcua#780 installer +/// work). The worker receives the key in Hello, never on disk or in its environment. +/// Writes go to a unique temp file and are renamed into place, so a concurrent reader (another +/// shard) sees the old entry, the new one, or none — never a torn file. +/// +/// +public sealed class ScriptPeCacheStore +{ + /// The key file's name under the cache directory. + public const string KeyFileName = "pe-cache.key"; + + /// The entries' subdirectory. + public const string EntriesDirectoryName = "pe"; + + /// The HMAC key length. + public const int KeyLength = 32; + + /// An entry above this size is not read (a 4 MiB source compiles to far less). + public const int MaxEntryBytes = 32 * 1024 * 1024; + + private const string EntryExtension = ".pe"; + private static readonly byte[] Magic = "OTPEC\u0001\0\0"u8.ToArray(); + private const int TagLength = 32; + + private readonly byte[] _key; + + /// Opens the store over with . + /// The cache directory (absolute). + /// The node's HMAC key ( bytes). + public ScriptPeCacheStore(string directory, byte[] key) + { + ArgumentException.ThrowIfNullOrWhiteSpace(directory); + ArgumentNullException.ThrowIfNull(key); + if (key.Length != KeyLength) throw new ArgumentException($"the PE cache key must be {KeyLength} bytes", nameof(key)); + Directory = directory; + EntriesDirectory = Path.Combine(directory, EntriesDirectoryName); + _key = (byte[])key.Clone(); + } + + /// The cache directory. + public string Directory { get; } + + /// The entries' directory. + public string EntriesDirectory { get; } + + /// + /// The host half: returns the node's key, creating the directory and the key on first use. An existing key of + /// the wrong length is not replaced (it may be a sign of tampering); the call throws and the caller runs without + /// a cache. + /// + /// The cache directory (absolute). + /// The key. + /// The directory or key cannot be created or read, or the key is malformed. + public static byte[] LoadOrCreateKey(string directory) + { + ArgumentException.ThrowIfNullOrWhiteSpace(directory); + CreatePrivateDirectory(directory); + CreatePrivateDirectory(Path.Combine(directory, EntriesDirectoryName)); + var keyPath = Path.Combine(directory, KeyFileName); + if (!File.Exists(keyPath)) + { + // Written to a temp name and moved without overwrite: two hosts racing on first start agree on one key. + var temp = keyPath + "." + Convert.ToHexString(RandomNumberGenerator.GetBytes(8)) + ".tmp"; + var options = new FileStreamOptions { Mode = FileMode.CreateNew, Access = FileAccess.Write, Share = FileShare.None }; + if (!OperatingSystem.IsWindows()) options.UnixCreateMode = UnixFileMode.UserRead | UnixFileMode.UserWrite; + using (var stream = new FileStream(temp, options)) + { + stream.Write(RandomNumberGenerator.GetBytes(KeyLength)); + } + try + { + File.Move(temp, keyPath, overwrite: false); + } + catch (IOException) when (File.Exists(keyPath)) + { + File.Delete(temp); + } + } + var key = File.ReadAllBytes(keyPath); + if (key.Length != KeyLength) + throw new IOException($"the PE cache key '{keyPath}' is {key.Length} bytes, not {KeyLength}; delete it to start a fresh cache"); + return key; + } + + /// The entry id for one compile: hex SHA-256 over the fingerprint, the kind and the source. + /// The compile fingerprint (ScriptEvaluator.CacheFingerprint). + /// The context kind (the wire enum value). + /// The script source. + /// A 64-character lowercase hex id. + public static string EntryId(ReadOnlySpan fingerprint, int kind, string source) + { + ArgumentNullException.ThrowIfNull(source); + using var hash = IncrementalHash.CreateHash(HashAlgorithmName.SHA256); + hash.AppendData(fingerprint); + hash.AppendData(BitConverter.GetBytes(kind)); + hash.AppendData(Encoding.UTF8.GetBytes(source)); + return Convert.ToHexString(hash.GetHashAndReset()).ToLowerInvariant(); + } + + /// The path of an entry. + /// The id. + /// The file path. + public string EntryPath(string entryId) + { + if (!IsValidId(entryId)) throw new ArgumentException("not a PE cache entry id", nameof(entryId)); + return Path.Combine(EntriesDirectory, entryId + EntryExtension); + } + + /// Reads and authenticates an entry. A rejected entry is deleted. + /// The id. + /// The PE bytes on a hit. + /// Hit, miss, or rejected. + public PeCacheLookup TryRead(string entryId, out byte[]? peImage) + { + peImage = null; + var path = EntryPath(entryId); + byte[] bytes; + try + { + var info = new FileInfo(path); + if (!info.Exists) return PeCacheLookup.Miss; + if (info.Length > MaxEntryBytes) + { + TryDelete(path); + return PeCacheLookup.Rejected; + } + bytes = File.ReadAllBytes(path); + } + catch (FileNotFoundException) + { + return PeCacheLookup.Miss; + } + catch (DirectoryNotFoundException) + { + return PeCacheLookup.Miss; + } + + if (bytes.Length <= Magic.Length + TagLength || !bytes.AsSpan(0, Magic.Length).SequenceEqual(Magic)) + { + TryDelete(path); + return PeCacheLookup.Rejected; + } + var tag = bytes.AsSpan(Magic.Length, TagLength); + var pe = bytes.AsSpan(Magic.Length + TagLength); + if (!CryptographicOperations.FixedTimeEquals(tag, Tag(entryId, pe))) + { + TryDelete(path); + return PeCacheLookup.Rejected; + } + + peImage = pe.ToArray(); + // Recency for Prune: a hit counts as a use. + try { File.SetLastWriteTimeUtc(path, DateTime.UtcNow); } catch (IOException) { } catch (UnauthorizedAccessException) { } + return PeCacheLookup.Hit; + } + + /// Writes an entry (atomically replacing any existing one). + /// The id. + /// The PE bytes. + /// False when the write failed (the cache is best effort: a failed write costs a recompile later). + public bool TryWrite(string entryId, ReadOnlySpan peImage) + { + var path = EntryPath(entryId); + var temp = Path.Combine(EntriesDirectory, $".{entryId}.{Convert.ToHexString(RandomNumberGenerator.GetBytes(6))}.tmp"); + try + { + System.IO.Directory.CreateDirectory(EntriesDirectory); + var options = new FileStreamOptions { Mode = FileMode.CreateNew, Access = FileAccess.Write, Share = FileShare.None }; + if (!OperatingSystem.IsWindows()) options.UnixCreateMode = UnixFileMode.UserRead | UnixFileMode.UserWrite; + using (var stream = new FileStream(temp, options)) + { + stream.Write(Magic); + stream.Write(Tag(entryId, peImage)); + stream.Write(peImage); + } + File.Move(temp, path, overwrite: true); + return true; + } + catch (Exception ex) when (ex is IOException or UnauthorizedAccessException) + { + TryDelete(temp); + return false; + } + } + + /// + /// Deletes entries not used for , then the least recently used beyond + /// , plus stray temp files. The design prunes "entries not in the current + /// generation"; this runner has no generation object, so recency stands in for it. + /// + /// Entries older than this go. + /// At most this many entries stay. + /// How many files were deleted. + public int Prune(TimeSpan maxAge, int maxEntries) + { + if (!System.IO.Directory.Exists(EntriesDirectory)) return 0; + var deleted = 0; + var now = DateTime.UtcNow; + var entries = new List(); + foreach (var file in new DirectoryInfo(EntriesDirectory).EnumerateFiles()) + { + var stale = now - file.LastWriteTimeUtc; + if (file.Name.EndsWith(".tmp", StringComparison.Ordinal)) + { + if (stale > TimeSpan.FromHours(1) && TryDelete(file.FullName)) deleted++; + continue; + } + if (!file.Name.EndsWith(EntryExtension, StringComparison.Ordinal)) continue; + if (stale > maxAge) + { + if (TryDelete(file.FullName)) deleted++; + continue; + } + entries.Add(file); + } + if (entries.Count > maxEntries) + { + foreach (var file in entries.OrderBy(f => f.LastWriteTimeUtc).Take(entries.Count - maxEntries)) + if (TryDelete(file.FullName)) deleted++; + } + return deleted; + } + + private byte[] Tag(string entryId, ReadOnlySpan pe) + { + using var hmac = IncrementalHash.CreateHMAC(HashAlgorithmName.SHA256, _key); + hmac.AppendData(Magic); + hmac.AppendData(Encoding.ASCII.GetBytes(entryId)); + hmac.AppendData(pe); + return hmac.GetHashAndReset(); + } + + private static bool IsValidId(string id) + => id is { Length: 64 } && id.All(static c => c is >= '0' and <= '9' or >= 'a' and <= 'f'); + + private static void CreatePrivateDirectory(string path) + { + if (OperatingSystem.IsWindows()) System.IO.Directory.CreateDirectory(path); + else System.IO.Directory.CreateDirectory(path, UnixFileMode.UserRead | UnixFileMode.UserWrite | UnixFileMode.UserExecute); + } + + private static bool TryDelete(string path) + { + try + { + File.Delete(path); + return true; + } + catch (Exception ex) when (ex is IOException or UnauthorizedAccessException) + { + return false; + } + } +} diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/IScriptRunner.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/IScriptRunner.cs index 0678896d..048a0229 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/IScriptRunner.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/IScriptRunner.cs @@ -97,6 +97,13 @@ public sealed record ScriptRunOutcome(ScriptRunStatus Status, object? Value, str /// Reason when no worker became ready within . public const string RunnerUnavailableReason = "script runner unavailable"; + /// + /// For : completes when the runner can run this script again (its worker + /// is back), so the seam re-evaluates it then with its latest inputs (design §9, "re-evaluated on the shard's + /// return"). Null when there is nothing to wait for. + /// + public Task? Recovery { get; init; } + /// A successful run. /// The script's result. /// The outcome. @@ -130,7 +137,7 @@ public static class ScriptRunOutcomeExtensions { ScriptRunStatus.Ok => Commons.Engines.VirtualTagEvalResult.Ok(outcome.Value), ScriptRunStatus.Unavailable => Commons.Engines.VirtualTagEvalResult.Unavailable( - outcome.Reason ?? ScriptRunOutcome.RunnerUnavailableReason), + outcome.Reason ?? ScriptRunOutcome.RunnerUnavailableReason) with { RetryWhen = outcome.Recovery }, _ => Commons.Engines.VirtualTagEvalResult.Failure(outcome.Reason ?? "evaluator failure"), }; } diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptEvaluator.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptEvaluator.cs index 5f5b23d2..8d717b12 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptEvaluator.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptEvaluator.cs @@ -70,7 +70,18 @@ public sealed class ScriptEvaluator : IDisposable /// Compiles user script source into an evaluator. /// The user script source code to compile. /// A ready-to-run wrapping the compiled script. - public static ScriptEvaluator Compile(string scriptSource) + public static ScriptEvaluator Compile(string scriptSource) => FromImage(CompileToImage(scriptSource)); + + /// + /// The compile half of : parse, sandbox compile, , the + /// stack-guard rewrite and the emit — returning the PE bytes without loading them. The out-of-process worker + /// caches these bytes on disk (lmxopcua#780 design §7) so a restart does not re-enter Roslyn. + /// + /// The user script source code to compile. + /// The emitted PE image. + /// The source does not compile. + /// The source uses a forbidden API. + public static byte[] CompileToImage(string scriptSource) { if (scriptSource is null) throw new ArgumentNullException(nameof(scriptSource)); @@ -142,8 +153,22 @@ public sealed class ScriptEvaluator : IDisposable throw new CompilationErrorException(emitErrors); } - peStream.Position = 0; - var alc = new ScriptAssemblyLoadContext(assemblyName); + return peStream.ToArray(); + } + + /// + /// The load half of : loads a PE image produced by into a fresh + /// collectible ALC and binds its Run method. The caller vouches for the image — only bytes this pipeline + /// emitted (the worker's HMAC-verified PE cache) may be passed, because loading arbitrary PE bytes would run + /// code the analyzer never saw. + /// + /// The image. + /// A ready-to-run evaluator. + public static ScriptEvaluator FromImage(byte[] peImage) + { + ArgumentNullException.ThrowIfNull(peImage); + using var peStream = new MemoryStream(peImage, writable: false); + var alc = new ScriptAssemblyLoadContext("ZB.MOM.WW.OtOpcUa.Core.Scripting.Compiled." + Guid.NewGuid().ToString("N")); Assembly assembly; try { @@ -182,6 +207,33 @@ public sealed class ScriptEvaluator : IDisposable return new ScriptEvaluator(alc, func); } + /// + /// A digest of everything that decides what emits for a given source, other than + /// the source itself (lmxopcua#780 design §7): the Roslyn version, this assembly's build identity (the + /// analyzer's deny-lists, the wrapper template, the stack-guard rewrite), the sandbox reference set, the + /// runtime version and / . A compiled-PE cache + /// keys on it, so an analyzer or compiler change invalidates every cached image and a cached PE can never skip + /// a newer sandbox rule. + /// + public static byte[] CacheFingerprint => FingerprintHolder.Value; + + private static readonly Lazy FingerprintHolder = new(ComputeFingerprint, LazyThreadSafetyMode.ExecutionAndPublication); + + private static byte[] ComputeFingerprint() + { + var sb = new StringBuilder("otopcua-script-pe/1\n"); + static void AppendAssembly(StringBuilder b, System.Reflection.Assembly a) + => b.Append(a.FullName).Append('|').Append(a.ManifestModule.ModuleVersionId.ToString("N")).Append('\n'); + AppendAssembly(sb, typeof(CSharpCompilation).Assembly); + AppendAssembly(sb, typeof(Compilation).Assembly); + AppendAssembly(sb, typeof(ScriptEvaluator<,>).Assembly); + sb.Append(Environment.Version).Append('\n'); + sb.Append(typeof(TContext).AssemblyQualifiedName).Append('\n'); + sb.Append(typeof(TResult).AssemblyQualifiedName).Append('\n'); + foreach (var identity in ScriptSandbox.ReferenceIdentities(typeof(TContext))) sb.Append(identity).Append('\n'); + return System.Security.Cryptography.SHA256.HashData(Encoding.UTF8.GetBytes(sb.ToString())); + } + /// Runs the script against an already-constructed context. /// The script context. /// Cancellation token for the operation. diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptRunnerOptions.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptRunnerOptions.cs index 54e03457..c8e7e431 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptRunnerOptions.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptRunnerOptions.cs @@ -74,6 +74,12 @@ public sealed class ScriptWorkerOptions /// Default : 20. public const int DefaultIsolationProbationRuns = 20; + /// Default : 10 min (design §3.3). + public static readonly TimeSpan DefaultIsolationIdleTimeout = TimeSpan.FromMinutes(10); + + /// Default : 2 min (design §5.3.6). + public static readonly TimeSpan DefaultCircuitOpenAfter = TimeSpan.FromMinutes(2); + /// Shard workers per node; a script is placed by a stable hash of its key. Clamped to 1–8. public int Shards { get; set; } = DefaultShards; @@ -102,9 +108,22 @@ public sealed class ScriptWorkerOptions public int IsolationProbationRuns { get; set; } = DefaultIsolationProbationRuns; /// - /// Directory of the node-local compiled-PE cache (design §7). Must be absolute and outside the content root - /// when set; startup fails otherwise (the family's fail-closed path rule). Reserved: the cache itself has - /// not landed yet (lmxopcua#780 P1 remainder), so a worker restart recompiles. + /// How long the isolation worker may sit with nothing to run before it is stopped (it is started again by the + /// next suspect). Default 10 min; 1 s – 24 h. + /// + public TimeSpan IsolationIdleTimeout { get; set; } = DefaultIsolationIdleTimeout; + + /// + /// The crash-loop circuit (design §5.3.6): when every shard worker has been down for this long, the runner + /// reports Unavailable, logs one Error and answers evaluations at once (no + /// wait) while it keeps retrying at the backoff cap. Default 2 min; 1 s – 1 h. + /// + public TimeSpan CircuitOpenAfter { get; set; } = DefaultCircuitOpenAfter; + + /// + /// Directory of the node-local compiled-PE cache (design §7): HMAC-authenticated PE images, so a worker restart + /// does not re-enter Roslyn. Must be absolute and outside the content root when set; startup fails otherwise + /// (the family's fail-closed path rule). Null (the default) runs without a cache: every worker start recompiles. /// public string? PeCacheDir { get; set; } @@ -175,9 +194,27 @@ public sealed class ScriptWorkerOptions probation = clamped; } + var idle = IsolationIdleTimeout; + if (idle < TimeSpan.FromSeconds(1) || idle > TimeSpan.FromHours(24)) + { + var clamped = idle < TimeSpan.FromSeconds(1) ? TimeSpan.FromSeconds(1) : TimeSpan.FromHours(24); + notes.Add($"{prefix}{nameof(IsolationIdleTimeout)} {idle} is outside 00:00:01–1.00:00:00; clamped to {clamped}."); + idle = clamped; + } + + var circuit = CircuitOpenAfter; + if (circuit < TimeSpan.FromSeconds(1) || circuit > TimeSpan.FromHours(1)) + { + var clamped = circuit < TimeSpan.FromSeconds(1) ? TimeSpan.FromSeconds(1) : TimeSpan.FromHours(1); + notes.Add($"{prefix}{nameof(CircuitOpenAfter)} {circuit} is outside 00:00:01–01:00:00; clamped to {clamped}."); + circuit = clamped; + } + adjustments = notes; return new ScriptWorkerOptions { + IsolationIdleTimeout = idle, + CircuitOpenAfter = circuit, Shards = shards, HardGrace = grace, HeapLimitBytes = heap, diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptSandbox.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptSandbox.cs index 2e495b6d..f0bbdc68 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptSandbox.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptSandbox.cs @@ -99,6 +99,23 @@ public static class ScriptSandbox return new SandboxConfig(references, imports); } + /// + /// A stable textual identity of the reference set compiles against for + /// — each pinned assembly's name and module version id, then each BCL path — without + /// opening any of them. Part of (lmxopcua#780 §7). + /// + /// The concrete script context type. + /// One line per reference, in a stable order. + public static IEnumerable ReferenceIdentities(Type contextType) + { + ArgumentNullException.ThrowIfNull(contextType); + var pinned = new[] { typeof(DataValueSnapshot).Assembly, typeof(ScriptContext).Assembly, typeof(Serilog.ILogger).Assembly, contextType.Assembly } + .Distinct() + .Select(static a => $"{a.FullName}|{a.ManifestModule.ModuleVersionId:N}") + .Order(StringComparer.Ordinal); + return pinned.Concat(EnumerateBclAssemblyPaths().Order(StringComparer.Ordinal)).ToArray(); + } + private static readonly System.Collections.Concurrent.ConcurrentDictionary PinnedReferences = new(); // The trusted-platform set is process-constant, so the BCL reference list is context-independent. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs index 1a12ae7c..76156fc8 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs @@ -647,6 +647,13 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba if (toPublish is not null || !string.Equals(st.LastFailureReason, reason, StringComparison.Ordinal)) failureToLog = reason; st.LastFailureReason = reason; + if (result.RetryWhen is { } retryWhen && !st.RecoveryArmed) + { + st.RecoveryArmed = true; + var rawPath = def.RawPath; + _ = retryWhen.ContinueWith(_ => OnRunnerRecovered(rawPath), CancellationToken.None, + TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); + } } else { @@ -659,6 +666,24 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba if (toPublish is { } p) Publish(def.RawPath, p.Value, p.StatusCode, pending.TimestampUtc); } + /// + /// The out-of-process runner that could not run 's script is serving again + /// (lmxopcua#780 design §9): queue one evaluation against the CURRENT inputs, so a calc tag with static inputs + /// does not stay BadResourceUnavailable until an input happens to change. + /// + /// The calc tag. + private void OnRunnerRecovered(string rawPath) + { + lock (_lock) + { + if (_stateByRawPath.TryGetValue(rawPath, out var st)) st.RecoveryArmed = false; + if (!_running || _disposed) return; + if (PrepareEvaluation(rawPath, DateTime.UtcNow) is not { } prepared) return; + _pending.Enqueue(prepared); + SchedulePump(); + } + } + /// Record a successful evaluation. MUST be called under ; returns what to /// publish, or null when the result is deduped. /// The dedup key is (value, status), not value alone — generalising the @@ -852,6 +877,9 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba /// The reason of the last failed evaluation (null after a success / inputs-Bad hold) — the /// gate that keeps the script-log entry to the transition and to a CHANGE of reason. public string? LastFailureReason; + + /// A runner-recovery re-evaluation is armed (lmxopcua#780): at most one per tag (latest-wins). + public bool RecoveryArmed; } private sealed record CalculationSubscriptionHandle(string DiagnosticId) : ISubscriptionHandle; diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/ScriptRunnerSelection.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/ScriptRunnerSelection.cs index 19c7b46c..2039c255 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/ScriptRunnerSelection.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/ScriptRunnerSelection.cs @@ -29,6 +29,10 @@ public sealed class ScriptRunnerSelection : IAsyncDisposable, IDisposable /// The out-of-process runner, when one is configured. public IScriptRunner? OutOfProcessRunner => _runner; + /// The out-of-process runner's status (the readiness check), or null when every seam runs in process. + /// The status, or null. + public ScriptRunnerStatus? Status() => _runner?.Status(); + /// The runner must use, or null for its in-process path. /// The seam. /// The runner, or null. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/HealthEndpoints.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/HealthEndpoints.cs index c3a48688..e7fe87e9 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/HealthEndpoints.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/HealthEndpoints.cs @@ -62,6 +62,16 @@ public static class HealthEndpoints sp => new OpcUaEndpointHealthCheck(sp.GetRequiredService()), failureStatus: null, tags: tags)); + + // lmxopcua#780 design §10 / owner decision §14 #12: the out-of-process script runner is a READY-tier + // component that is at worst Degraded (still 200): OPC UA keeps serving driver data while scripts are + // Bad, so a sick runner must never fail readiness (on a pair that would move ServiceLevel to the peer). + // Healthy when every seam runs in process. Not on the active tier. + checks.Add(new HealthCheckRegistration( + ScriptRunnerHealthCheck.Name, + sp => new ScriptRunnerHealthCheck(sp.GetService()), + failureStatus: null, + tags: new[] { ZbHealthTags.Ready })); } // configdb probe is admin-only (per-cluster mesh Phase 4). A driver-only node gets its config diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/ScriptRunnerHealthCheck.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/ScriptRunnerHealthCheck.cs new file mode 100644 index 00000000..e8afc64d --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/ScriptRunnerHealthCheck.cs @@ -0,0 +1,68 @@ +using Microsoft.Extensions.Diagnostics.HealthChecks; +using ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess; +using ZB.MOM.WW.OtOpcUa.Host.Configuration; + +namespace ZB.MOM.WW.OtOpcUa.Host.Health; + +/// +/// Pure decision function for the script-runner probe (lmxopcua#780 design §10, owner decision §14 #12), factored +/// out of so the matrix is table-testable without worker processes. +/// +public static class ScriptRunnerHealthDecision +{ + /// + /// Maps the runner's status to a health status. Never Unhealthy: a sick script runner does not stop the + /// node serving driver data over OPC UA, so it must not fail readiness — on a redundant pair that would shift + /// ServiceLevel to the peer over a script problem. Degraded is still ready (200) but visible. + /// + /// The out-of-process runner's status, or null when every seam runs in process. + /// The status, a description and the per-shard data. + public static (HealthStatus Status, string Description, IReadOnlyDictionary Data) Evaluate(ScriptRunnerStatus? status) + { + if (status is null) + return (HealthStatus.Healthy, "Scripts run in process (Scripting:Runner=InProcess).", new Dictionary()); + + var data = new Dictionary(StringComparer.Ordinal) + { + ["state"] = status.State.ToString(), + ["suspects"] = status.SuspectCount, + ["quarantined"] = status.QuarantinedCount, + ["peCache"] = status.PeCacheEnabled, + }; + foreach (var shard in status.Shards) + { + data[$"shard.{shard.Label}"] = shard.Up + ? $"up pid {shard.Pid}, {shard.InFlight} in flight" + : shard.DownFor is { } downFor ? $"down {downFor.TotalSeconds:F0} s ({shard.ConsecutiveFailures} failure(s))" : "not started"; + } + + var down = status.Shards.Where(s => s.DownFor is not null).Select(s => s.Label).ToArray(); + return status.State switch + { + ScriptRunnerState.Ready => (HealthStatus.Healthy, "Out-of-process script runner ready.", data), + ScriptRunnerState.Degraded => (HealthStatus.Degraded, + $"Out-of-process script runner degraded: ScriptWorker {string.Join(", ", down)} down; scripts placed there report BadResourceUnavailable until it restarts.", + data), + _ => (HealthStatus.Degraded, + "Out-of-process script runner unavailable (circuit open): every ScriptWorker shard is down; every compiled script reports BadResourceUnavailable. OPC UA keeps serving driver data.", + data), + }; + } +} + +/// The script-runner readiness component: Healthy, or Degraded-but-ready (see ). +/// The node's runner selection; null on nodes that never register one. +public sealed class ScriptRunnerHealthCheck(ScriptRunnerSelection? selection) : IHealthCheck +{ + /// The check's registration name. + public const string Name = "script-runner"; + + /// + public Task CheckHealthAsync(HealthCheckContext context, CancellationToken cancellationToken = default) + { + var (status, description, data) = ScriptRunnerHealthDecision.Evaluate(selection?.Status()); + return Task.FromResult(status == HealthStatus.Healthy + ? HealthCheckResult.Healthy(description, data) + : HealthCheckResult.Degraded(description, data: data)); + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagActor.cs index f55bc965..9940ea61 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagActor.cs @@ -51,6 +51,12 @@ public sealed class VirtualTagActor : ReceiveActor, IWithUnboundedStash /// when the tag has a TimerIntervalMs. private sealed class TimerTick { public static readonly TimerTick Instance = new(); private TimerTick() { } } + /// + /// The out-of-process script runner that could not run this tag's script is serving again (lmxopcua#780 design + /// §9): re-evaluate once against the latest inputs. Private to the actor. + /// + private sealed class RunnerRecovered { public static readonly RunnerRecovered Instance = new(); private RunnerRecovered() { } } + /// The asynchronous evaluation started by finished. Private to the actor. private sealed record EvaluationCompleted( Task Evaluation, uint WorstInputStatus, DateTime TimestampUtc); @@ -108,6 +114,8 @@ public sealed class VirtualTagActor : ReceiveActor, IWithUnboundedStash private readonly bool _changeTriggered; private readonly int? _timerIntervalMs; private ICancelable? _timer; + // One outstanding RunnerRecovered wait at most: a burst of Unavailable results re-evaluates once (latest-wins). + private bool _recoveryArmed; // R02:P6 — a script failing on every trigger of a 1 Hz input published one cluster-wide DPS // ScriptLogEntry per second per tag; 200 such tags shed everyone's entries out of the bounded // telemetry channels. The per-failure entry is now rate-limited per tag (20/min + one summary). @@ -212,6 +220,11 @@ public sealed class VirtualTagActor : ReceiveActor, IWithUnboundedStash Receive(OnDependencyBatch); Receive(_ => OnReassertValue()); Receive(_ => Evaluate(DateTime.UtcNow)); + Receive(_ => + { + _recoveryArmed = false; + Evaluate(DateTime.UtcNow); + }); // Only reachable for a completion that outlived its behaviour (a supervision restart mid-evaluation // resets the behaviour stack): the incarnation it belonged to is gone, so it is dropped. Receive(_ => { }); @@ -368,6 +381,21 @@ public sealed class VirtualTagActor : ReceiveActor, IWithUnboundedStash ReceiveAny(_ => Stash.Stash()); } + /// + /// A runner failure (BadResourceUnavailable) carries the task that completes when the runner is back: evaluate + /// again then, with whatever inputs are current, instead of holding the Bad until the next input change — which, + /// for a static input, would never come (lmxopcua#780 design §9, "re-evaluated on the shard's return"). + /// + /// The recovery task, or null. + private void ArmRecovery(Task? retryWhen) + { + if (retryWhen is null || _recoveryArmed) return; + _recoveryArmed = true; + var self = Self; + retryWhen.ContinueWith(_ => self.Tell(RunnerRecovered.Instance), + CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); + } + private void OnEvaluatorThrew(Exception ex, DateTime timestampUtc) { _log.Warning(ex, "VirtualTag {Id}: evaluator threw", _virtualTagId); @@ -393,6 +421,7 @@ public sealed class VirtualTagActor : ReceiveActor, IWithUnboundedStash { OnEvaluationFailed(result.Reason ?? "evaluator failure", "Warning", timestampUtc, result.StatusCode ?? OpcUaStatus.BadInternalError); + ArmRecovery(result.RetryWhen); return; } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker/WorkerHost.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker/WorkerHost.cs index 858f0328..8efc0496 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker/WorkerHost.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker/WorkerHost.cs @@ -44,6 +44,7 @@ internal sealed class WorkerHost private int _queueCapacity = 4096; private TimeSpan _hardGrace = TimeSpan.FromSeconds(1); private TimeSpan _heartbeatInterval = TimeSpan.FromSeconds(1); + private ScriptPeCacheStore? _peCache; private long _seq; public WorkerHost(NamedPipeClientStream pipe, string nonce, int parentPid) @@ -102,6 +103,11 @@ internal sealed class WorkerHost if (hello.QueueCapacity > 0) _queueCapacity = (int)hello.QueueCapacity; if (hello.HardGraceMs > 0) _hardGrace = TimeSpan.FromMilliseconds(hello.HardGraceMs); if (hello.HeartbeatIntervalMs > 0) _heartbeatInterval = TimeSpan.FromMilliseconds(hello.HeartbeatIntervalMs); + if (!string.IsNullOrEmpty(hello.PeCacheDir) && hello.PeCacheKey.Length == ScriptPeCacheStore.KeyLength) + { + // Design §7: the host owns the key; it reaches us only here, never through the environment or a file. + _peCache = new ScriptPeCacheStore(hello.PeCacheDir, hello.PeCacheKey.ToByteArray()); + } } private void StartBackground(int evalThreads) @@ -181,7 +187,8 @@ internal sealed class WorkerHost var outcome = new LoadOutcome { Key = script.Key, Ok = true }; try { - var loaded = LoadedScript.Compile(script); + var loaded = LoadedScript.Load(script, _peCache, out var cache); + outcome.Cache = cache; if (_scripts.TryRemove(script.Key, out var previous)) previous.Dispose(); _scripts[script.Key] = loaded; } @@ -499,15 +506,52 @@ internal sealed class WorkerHost public string[] InputPaths { get; } - public static LoadedScript Compile(ScriptSource source) => source.Kind switch + /// + /// Loads a script: from a verified PE cache entry when there is one (Roslyn is not entered), otherwise by + /// compiling it and writing the entry (design §7). A compile error is never cached — the host's negative + /// cache covers that — and an entry that fails verification is deleted by the store and recompiled. + /// + public static LoadedScript Load(ScriptSource source, ScriptPeCacheStore? cache, out PeCache how) => source.Kind switch { ContextKind.VirtualTag or ContextKind.Calculation => new LoadedScript( - [.. source.InputPaths], ScriptEvaluator.Compile(source.Source), null), + [.. source.InputPaths], Load(source, cache, out how), null), ContextKind.AlarmPredicate => new LoadedScript( - [.. source.InputPaths], null, ScriptEvaluator.Compile(source.Source)), + [.. source.InputPaths], null, Load(source, cache, out how)), _ => throw new ScriptWorkerProtocolException($"unknown script kind {source.Kind}"), }; + private static ScriptEvaluator Load(ScriptSource source, ScriptPeCacheStore? cache, + out PeCache how) where TContext : ScriptContext + { + if (cache is null) + { + how = PeCache.None; + return ScriptEvaluator.Compile(source.Source); + } + + var id = ScriptPeCacheStore.EntryId(ScriptEvaluator.CacheFingerprint, (int)source.Kind, source.Source); + var lookup = cache.TryRead(id, out var image); + if (lookup == PeCacheLookup.Hit) + { + try + { + how = PeCache.Hit; + return ScriptEvaluator.FromImage(image!); + } + catch (Exception ex) when (ex is BadImageFormatException or InvalidOperationException or TypeLoadException or MissingMethodException) + { + // Authenticated but unloadable: only possible across a runtime change the fingerprint missed. + try { File.Delete(cache.EntryPath(id)); } catch (IOException) { } + lookup = PeCacheLookup.Rejected; + } + } + + var compiled = ScriptEvaluator.CompileToImage(source.Source); + cache.TryWrite(id, compiled); + how = lookup == PeCacheLookup.Rejected ? PeCache.Rejected : PeCache.Miss; + return ScriptEvaluator.FromImage(compiled); + } + public object? Run(IReadOnlyDictionary inputs, ILogger logger, Func clock, CancellationToken ct, out bool isPredicate) { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationDriverScriptRunnerTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationDriverScriptRunnerTests.cs index d336f9ec..b61b5436 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationDriverScriptRunnerTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationDriverScriptRunnerTests.cs @@ -94,6 +94,38 @@ public sealed class CalculationDriverScriptRunnerTests scriptFault!.Snapshot.StatusCode.ShouldBe(StatusBadInternalError); } + /// + /// 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 + /// RetryWhen the tag stays BadResourceUnavailable. + /// + [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() { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Health/ScriptRunnerHealthCheckTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Health/ScriptRunnerHealthCheckTests.cs new file mode 100644 index 00000000..42f4a426 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Health/ScriptRunnerHealthCheckTests.cs @@ -0,0 +1,79 @@ +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Diagnostics.HealthChecks; +using Microsoft.Extensions.Options; +using Shouldly; +using Xunit; +using ZB.MOM.WW.Health; +using ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess; +using ZB.MOM.WW.OtOpcUa.Host.Health; + +namespace ZB.MOM.WW.OtOpcUa.Host.Tests.Health; + +/// +/// lmxopcua#780 design §10, owner decision §14 #12: the script runner is a READY-tier component that is degraded but +/// still ready when its circuit is open — a sick runner must never fail readiness (on a redundant pair that would move +/// ServiceLevel to the peer over a script problem while OPC UA still serves driver data). +/// +public sealed class ScriptRunnerHealthCheckTests +{ + private static ScriptWorkerShardStatus Shard(string label, bool up, double? downSeconds = null) + => new(label, label == "iso", up, downSeconds is { } s ? TimeSpan.FromSeconds(s) : null, up ? 1234 : null, up ? 0 : 3, 0, 0, 0); + + private static ScriptRunnerStatus Status(ScriptRunnerState state, params ScriptWorkerShardStatus[] shards) + => new(state, shards, SuspectCount: 0, QuarantinedCount: 0, PeCacheEnabled: true); + + [Fact] + public void In_process_is_healthy() + => ScriptRunnerHealthDecision.Evaluate(null).Status.ShouldBe(HealthStatus.Healthy); + + [Fact] + public void Ready_is_healthy() + => ScriptRunnerHealthDecision.Evaluate(Status(ScriptRunnerState.Ready, Shard("0", true), Shard("1", true))).Status + .ShouldBe(HealthStatus.Healthy); + + [Fact] + public void A_shard_down_is_degraded_and_names_it() + { + var (status, description, data) = ScriptRunnerHealthDecision.Evaluate( + Status(ScriptRunnerState.Degraded, Shard("0", true), Shard("1", false, 4))); + status.ShouldBe(HealthStatus.Degraded); + description.ShouldContain("ScriptWorker 1 down"); + data["shard.1"].ShouldBe("down 4 s (3 failure(s))"); + data["state"].ShouldBe("Degraded"); + } + + /// ★ decision 12: the circuit-open runner is Degraded — never Unhealthy, so /health/ready stays 200. + [Fact] + public void An_open_circuit_is_degraded_never_unhealthy() + { + var (status, description, _) = ScriptRunnerHealthDecision.Evaluate( + Status(ScriptRunnerState.Unavailable, Shard("0", false, 200), Shard("1", false, 200))); + status.ShouldBe(HealthStatus.Degraded); + description.ShouldContain("circuit open"); + } + + [Theory] + [InlineData(false, false)] + [InlineData(true, true)] + public async Task It_is_registered_on_the_ready_tier_only_on_driver_nodes(bool hasDriver, bool registered) + { + var registrations = new ServiceCollection() + .AddOtOpcUaHealth(hasAdmin: true, hasDriver: hasDriver) + .BuildServiceProvider() + .GetRequiredService>() + .Value.Registrations + .Where(r => r.Name == ScriptRunnerHealthCheck.Name) + .ToList(); + if (!registered) + { + registrations.ShouldBeEmpty(); + return; + } + var registration = registrations.ShouldHaveSingleItem(); + registration.Tags.ShouldBe([ZbHealthTags.Ready]); + // A node without a runner selection (in-process, or a harness) resolves it as Healthy. + var check = registration.Factory(new ServiceCollection().BuildServiceProvider()).ShouldBeOfType(); + (await check.CheckHealthAsync(new HealthCheckContext { Registration = registration }, CancellationToken.None)) + .Status.ShouldBe(HealthStatus.Healthy); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/VirtualTagActorRunnerUnavailableTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/VirtualTagActorRunnerUnavailableTests.cs index 252bfeb5..2d5bb0f8 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/VirtualTagActorRunnerUnavailableTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/VirtualTagActorRunnerUnavailableTests.cs @@ -54,6 +54,36 @@ public sealed class VirtualTagActorRunnerUnavailableTests : RuntimeActorTestBase scriptFault.Value.ShouldBe(5); } + /// + /// Design §9: a victim of a worker restart is re-evaluated when the runner is back, with the latest inputs — not + /// left Bad until an input happens to change (a static input never does). Mutation check: without the actor + /// arming , no second result arrives. + /// + [Fact] + public void A_runner_unavailable_result_is_re_evaluated_once_when_the_runner_recovers() + { + var parent = CreateTestProbe(); + var evaluator = new QueuedEvaluator(); + var recovered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + evaluator.Results.Enqueue(VirtualTagEvalResult.Ok(5)); + evaluator.Results.Enqueue(VirtualTagEvalResult.Unavailable("script worker restarted") with { RetryWhen = recovered.Task }); + evaluator.Results.Enqueue(VirtualTagEvalResult.Ok(7)); + var actor = parent.ChildActorOf(VirtualTagActor.Props("vt-1", "x", evaluator: evaluator, dependencyRefs: ["a"])); + + actor.Tell(new VirtualTagActor.DependencyValueChanged("a", 1, DateTime.UtcNow)); + parent.ExpectMsg().Value.ShouldBe(5); + actor.Tell(new VirtualTagActor.DependencyValueChanged("a", 2, DateTime.UtcNow)); + parent.ExpectMsg(PresenceBudget).StatusCode.ShouldBe(OpcUaStatus.BadResourceUnavailable); + parent.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); // nothing re-runs before the runner is back + + recovered.SetResult(); + var again = parent.ExpectMsg(PresenceBudget); + again.StatusCode.ShouldBe(OpcUaStatus.Good, "re-evaluated on recovery, with no new input"); + again.Value.ShouldBe(7); + evaluator.Results.ShouldBeEmpty(); + parent.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); // exactly one re-evaluation + } + [Fact] public void A_failure_without_a_status_keeps_BadInternalError() { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/PeCacheTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/PeCacheTests.cs new file mode 100644 index 00000000..637ceef3 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/PeCacheTests.cs @@ -0,0 +1,253 @@ +using System.Diagnostics; +using System.Security.Cryptography; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Scripting; +using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol; +using ZB.MOM.WW.OtOpcUa.Core.VirtualTags; + +namespace ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests; + +/// +/// lmxopcua#780 design §7 (owner decision §14 #8): the node-local HMAC'd compiled-PE cache. The store is tested +/// directly (authentication, eviction, keying); the worker-backed cases prove a restart loads from it instead of +/// recompiling, and that the worker never runs bytes that did not verify. +/// +public sealed class PeCacheStoreTests +{ + private static readonly byte[] Fingerprint = SHA256.HashData("fp"u8); + + private static (ScriptPeCacheStore Store, Scripts.TempDir Dir) NewStore() + { + var dir = new Scripts.TempDir(); + var key = ScriptPeCacheStore.LoadOrCreateKey(dir.Path); + return (new ScriptPeCacheStore(dir.Path, key), dir); + } + + [Fact] + public void A_written_entry_reads_back_and_an_absent_one_misses() + { + var (store, dir) = NewStore(); + using var cleanup = dir; + var id = ScriptPeCacheStore.EntryId(Fingerprint, 1, "return 1;"); + store.TryRead(id, out _).ShouldBe(PeCacheLookup.Miss); + store.TryWrite(id, [1, 2, 3, 4]).ShouldBeTrue(); + store.TryRead(id, out var pe).ShouldBe(PeCacheLookup.Hit); + pe.ShouldBe(new byte[] { 1, 2, 3, 4 }); + } + + /// Every way an entry can be made to lie is rejected AND deleted. Positive control: the untouched entry hits. + [Theory] + [InlineData("flip-pe")] + [InlineData("flip-tag")] + [InlineData("truncate")] + [InlineData("magic")] + [InlineData("other-key")] + [InlineData("renamed")] + public void A_tampered_or_mismatched_entry_is_rejected_and_evicted(string tamper) + { + var (store, dir) = NewStore(); + using var cleanup = dir; + var id = ScriptPeCacheStore.EntryId(Fingerprint, 1, "return 1;"); + var otherId = ScriptPeCacheStore.EntryId(Fingerprint, 1, "return 2;"); + store.TryWrite(id, [10, 20, 30, 40, 50]).ShouldBeTrue(); + store.TryRead(id, out _).ShouldBe(PeCacheLookup.Hit, "positive control"); + var path = store.EntryPath(id); + var bytes = File.ReadAllBytes(path); + var readId = id; + switch (tamper) + { + case "flip-pe": bytes[^1] ^= 0xFF; File.WriteAllBytes(path, bytes); break; + case "flip-tag": bytes[10] ^= 0x01; File.WriteAllBytes(path, bytes); break; + case "truncate": File.WriteAllBytes(path, bytes[..(bytes.Length - 2)]); break; + case "magic": bytes[0] = (byte)'X'; File.WriteAllBytes(path, bytes); break; + case "other-key": + new ScriptPeCacheStore(dir.Path, RandomNumberGenerator.GetBytes(32)).TryWrite(id, [10, 20, 30, 40, 50]).ShouldBeTrue(); + break; + case "renamed": + // A valid entry for one source copied under another source's id: the tag binds the id. + File.Move(path, store.EntryPath(otherId)); + path = store.EntryPath(otherId); + readId = otherId; + break; + } + + store.TryRead(readId, out var pe).ShouldBe(PeCacheLookup.Rejected); + pe.ShouldBeNull(); + File.Exists(path).ShouldBeFalse("a rejected entry is evicted"); + } + + [Fact] + public void The_entry_id_changes_with_the_compiler_fingerprint_the_kind_and_the_source() + { + var baseline = ScriptPeCacheStore.EntryId(Fingerprint, 1, "return 1;"); + ScriptPeCacheStore.EntryId(SHA256.HashData("fp2"u8), 1, "return 1;").ShouldNotBe(baseline, "a compiler/analyzer change misses"); + ScriptPeCacheStore.EntryId(Fingerprint, 2, "return 1;").ShouldNotBe(baseline); + ScriptPeCacheStore.EntryId(Fingerprint, 1, "return 1; ").ShouldNotBe(baseline); + ScriptPeCacheStore.EntryId(Fingerprint, 1, "return 1;").ShouldBe(baseline); + } + + [Fact] + public void The_compile_fingerprint_covers_the_context_and_result_types() + { + ScriptEvaluator.CacheFingerprint.Length.ShouldBe(32); + ScriptEvaluator.CacheFingerprint + .ShouldNotBe(ScriptEvaluator.CacheFingerprint); + } + + [Fact] + public void The_key_is_created_once_private_and_reused() + { + using var dir = new Scripts.TempDir(); + var key = ScriptPeCacheStore.LoadOrCreateKey(dir.Path); + key.Length.ShouldBe(ScriptPeCacheStore.KeyLength); + ScriptPeCacheStore.LoadOrCreateKey(dir.Path).ShouldBe(key, "the second start reuses it"); + if (!OperatingSystem.IsWindows()) + { + File.GetUnixFileMode(Path.Combine(dir.Path, ScriptPeCacheStore.KeyFileName)) + .ShouldBe(UnixFileMode.UserRead | UnixFileMode.UserWrite, "readable only by the host account"); + } + } + + [Fact] + public void A_malformed_key_is_refused_not_silently_replaced() + { + using var dir = new Scripts.TempDir(); + File.WriteAllBytes(Path.Combine(dir.Path, ScriptPeCacheStore.KeyFileName), [1, 2, 3]); + Should.Throw(() => ScriptPeCacheStore.LoadOrCreateKey(dir.Path)); + File.ReadAllBytes(Path.Combine(dir.Path, ScriptPeCacheStore.KeyFileName)).Length.ShouldBe(3); + } + + [Fact] + public void Prune_drops_stale_entries_then_the_least_recently_used() + { + var (store, dir) = NewStore(); + using var cleanup = dir; + var ids = Enumerable.Range(0, 5).Select(i => ScriptPeCacheStore.EntryId(Fingerprint, 1, "s" + i)).ToArray(); + for (var i = 0; i < ids.Length; i++) + { + store.TryWrite(ids[i], [1]).ShouldBeTrue(); + File.SetLastWriteTimeUtc(store.EntryPath(ids[i]), DateTime.UtcNow.AddDays(-i * 10)); + } + store.Prune(maxAge: TimeSpan.FromDays(25), maxEntries: 2).ShouldBe(3); + File.Exists(store.EntryPath(ids[0])).ShouldBeTrue(); + File.Exists(store.EntryPath(ids[1])).ShouldBeTrue(); + File.Exists(store.EntryPath(ids[2])).ShouldBeFalse("LRU beyond the cap"); + File.Exists(store.EntryPath(ids[3])).ShouldBeFalse("older than the max age"); + } +} + +/// The PE cache through real workers. +[Collection(ScriptWorkerCollection.Name)] +[Trait("Category", "ScriptWorker")] +public sealed class PeCacheWorkerTests +{ + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private static string Source(int i) => $"var x = (int)ctx.GetTag(\"x\").Value; var acc = 0L; for (var j = 0; j < 10; j++) acc += x * {i} + j; return acc;"; + + private static long Expected(int i) => Enumerable.Range(0, 10).Sum(j => 3L * i + j); + + /// Time from a fresh runner (fresh worker process) to the first result of every one of scripts. + private static async Task<(TimeSpan Elapsed, (long Hits, long Misses, long Rejected) Counts)> ColdStart(string? cacheDir, int count) + { + await using var runner = Scripts.Runner(Scripts.Options(peCacheDir: cacheDir)); + var sw = Stopwatch.StartNew(); + for (var i = 0; i < count; i++) + { + var outcome = await runner.RunAsync(Scripts.Vt("vt-" + i, Source(i), inputs: Scripts.Input("x", 3)), Ct); + outcome.Status.ShouldBe(ScriptRunStatus.Ok, outcome.Reason); + outcome.Value.ShouldBe(Expected(i)); + } + return (sw.Elapsed, runner.PeCacheCounts); + } + + /// + /// The measurement the slice asks for: worker restart to first result for N scripts — with no cache (every + /// start recompiles), with an empty cache (compiles and writes) and with a warm cache (loads only). The warm + /// start must serve every script from the cache and beat the cold one. Mutation check: a worker that ignores the + /// cache reports misses on the warm start. + /// + [Fact] + public async Task A_warm_restart_loads_every_script_from_the_cache_and_is_faster_than_a_cold_one() + { + const int count = 40; + using var dir = new Scripts.TempDir(); + var cacheDir = Path.Combine(dir.Path, "pe-cache"); + + var uncached = await ColdStart(null, count); + var cold = await ColdStart(cacheDir, count); + var warm = await ColdStart(cacheDir, count); + + TestContext.Current.TestOutputHelper?.WriteLine( + $"{count} scripts, restart to all first results: no cache {uncached.Elapsed.TotalMilliseconds:F0} ms; " + + $"empty cache {cold.Elapsed.TotalMilliseconds:F0} ms; warm cache {warm.Elapsed.TotalMilliseconds:F0} ms"); + + uncached.Counts.ShouldBe((0L, 0L, 0L), "no cache configured"); + cold.Counts.ShouldBe((0L, (long)count, 0L), "the first start compiles and writes every entry"); + warm.Counts.ShouldBe(((long)count, 0L, 0L), "the restart loads every script from the cache"); + warm.Elapsed.ShouldBeLessThan(cold.Elapsed, "loading a PE is cheaper than entering Roslyn"); + Directory.EnumerateFiles(Path.Combine(cacheDir, ScriptPeCacheStore.EntriesDirectoryName), "*.pe").Count().ShouldBe(count); + } + + /// + /// The security property: a cache entry replaced with a DIFFERENT, validly framed PE (one that returns 666) under + /// a forged tag is never run — the worker rejects it, deletes it, recompiles the real source and returns 1. + /// Positive control: the forged image itself runs and returns 666 when loaded directly. + /// + [Fact] + public async Task A_forged_cache_entry_is_rejected_and_the_real_source_recompiled() + { + using var dir = new Scripts.TempDir(); + var cacheDir = Path.Combine(dir.Path, "pe-cache"); + await using (var first = Scripts.Runner(Scripts.Options(peCacheDir: cacheDir))) + (await first.RunAsync(Scripts.Vt("vt", "return 1;"), Ct)).Value.ShouldBe(1); + var entry = Directory.EnumerateFiles(Path.Combine(cacheDir, ScriptPeCacheStore.EntriesDirectoryName), "*.pe").ShouldHaveSingleItem(); + + var evil = ScriptEvaluator.CompileToImage("return 666;"); + using (var direct = ScriptEvaluator.FromImage(evil)) + { + (await direct.RunAsync(new VirtualTagContext(Scripts.NoInputs, static (_, _) => { }, Serilog.Core.Logger.None, () => DateTime.UtcNow), Ct)) + .ShouldBe(666, "positive control: the forged image is a working script"); + } + var original = File.ReadAllBytes(entry); + var forged = new byte[8 + 32 + evil.Length]; + original.AsSpan(0, 8).CopyTo(forged); // the real magic + RandomNumberGenerator.GetBytes(32).CopyTo(forged, 8); // a tag the attacker cannot compute + evil.CopyTo(forged, 40); + File.WriteAllBytes(entry, forged); + + await using var second = Scripts.Runner(Scripts.Options(peCacheDir: cacheDir)); + var outcome = await second.RunAsync(Scripts.Vt("vt", "return 1;"), Ct); + outcome.Value.ShouldBe(1, "the forged bytes never ran"); + second.PeCacheCounts.Rejected.ShouldBe(1); + var key = File.ReadAllBytes(Path.Combine(cacheDir, ScriptPeCacheStore.KeyFileName)); + new ScriptPeCacheStore(cacheDir, key).TryRead(Path.GetFileNameWithoutExtension(entry), out _) + .ShouldBe(PeCacheLookup.Hit, "the entry was rewritten from the real compile"); + } + + /// A compile error is never cached (the host's negative cache covers it) and does not poison the directory. + [Fact] + public async Task A_compile_error_writes_no_entry() + { + using var dir = new Scripts.TempDir(); + var cacheDir = Path.Combine(dir.Path, "pe-cache"); + await using var runner = Scripts.Runner(Scripts.Options(peCacheDir: cacheDir)); + (await runner.RunAsync(Scripts.Vt("bad", "return nope;"), Ct)).Reason!.ShouldStartWith("compile error:"); + Directory.EnumerateFiles(Path.Combine(cacheDir, ScriptPeCacheStore.EntriesDirectoryName), "*.pe").ShouldBeEmpty(); + } + + /// An unusable cache directory is a Warning and a cache-less runner, never a failure. + [Fact] + public async Task An_unusable_cache_directory_degrades_to_no_cache() + { + using var dir = new Scripts.TempDir(); + var blocker = Path.Combine(dir.Path, "a-file"); + File.WriteAllText(blocker, "not a directory"); + var sink = new CapturingSink(); + await using var runner = Scripts.Runner(Scripts.Options(peCacheDir: Path.Combine(blocker, "pe-cache")), sink.CreateLogger()); + (await runner.RunAsync(Scripts.Vt("vt", "return 5;"), Ct)).Value.ShouldBe(5); + runner.Status().PeCacheEnabled.ShouldBeFalse(); + sink.Rendered.ShouldContain(m => m.Contains("is unusable")); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/SupervisionTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/SupervisionTests.cs new file mode 100644 index 00000000..f2c5f047 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/SupervisionTests.cs @@ -0,0 +1,441 @@ +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; + +/// +/// 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. +/// +[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 Ms(int ms) => Scripts.Input("ms", ms); + + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private static string VtKey(string key) => InProcessScriptRunner.ScopedKey(ScriptContextKind.VirtualTag, key); + + // ---------------------------------------------------------------- the quarantine race + + /// + /// 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 Exited is signalled. + /// Mutation check: moving the attribution call after the completion loop in WorkerConnection.Finish turns + /// both assertions red. + /// + [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? 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"); + } + + /// + /// 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 attributed false when the callback returns. + /// + [Fact] + public async Task The_shard_attributes_an_exit_synchronously_inside_the_connections_callback() + { + Action? 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"); + } + + /// + /// 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. + /// + [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(); + 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 + + /// + /// 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. + /// + [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 + + /// + /// 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. + /// + [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; + } + + /// + /// 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. + /// + [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(); + } + + /// + /// 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. + /// + [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 */ } + } + } + + /// A worker's marker exists while it runs and is removed when it stops cleanly. + [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(File.ReadAllText(marker))!.HostPid.ShouldBe(Environment.ProcessId); + await runner.DisposeAsync(); + File.Exists(marker).ShouldBeFalse(); + } + + // ---------------------------------------------------------------- a compile that crashes its worker + + /// + /// 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. + /// + [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 CrashDuringCompile(Func 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); + } + + /// + /// 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. + /// + [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(); + using var listener = new MeterListener(); + listener.InstrumentPublished = (instrument, l) => + { + if (instrument.Name == "otopcua.script.runner.state") l.EnableMeasurementEvents(instrument); + }; + listener.SetMeasurementEventCallback((_, value, _, _) => states.Add(value)); + listener.Start(); + listener.RecordObservableInstruments(); + states.ShouldContain(2); + runner.Status().State.ShouldBe(ScriptRunnerState.Unavailable); + } + + /// The per-shard gauges observe the live worker (positive control: the shard is up and reports its pid). + [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(); + 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((_, value, tags, _) => + { + foreach (var tag in tags) if (tag.Key == "shard") up.Add((value, tag.Value as string)); + }); + listener.SetMeasurementEventCallback((_, 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); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/TestSupport.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/TestSupport.cs index d0612986..6bf8086e 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/TestSupport.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/TestSupport.cs @@ -37,7 +37,8 @@ internal static class Scripts /// Options for a single-shard runner with short budgets, so the containment tests run in seconds. public static ScriptingOptions Options(int shards = 1, TimeSpan? hardGrace = null, int quarantineAfterKills = 2, - int probation = 20, TimeSpan? unavailableGrace = null, string? workerPath = null, long heapLimit = 256L * 1024 * 1024) + int probation = 20, TimeSpan? unavailableGrace = null, string? workerPath = null, long heapLimit = 256L * 1024 * 1024, + string? peCacheDir = null, TimeSpan? isolationIdle = null, TimeSpan? circuitOpenAfter = null) => new() { Runner = ScriptRunnerMode.OutOfProcess, @@ -51,12 +52,34 @@ internal static class Scripts UnavailableGrace = unavailableGrace ?? TimeSpan.FromSeconds(30), ExecutablePath = workerPath, HeapLimitBytes = heapLimit, + PeCacheDir = peCacheDir, + IsolationIdleTimeout = isolationIdle ?? ScriptWorkerOptions.DefaultIsolationIdleTimeout, + CircuitOpenAfter = circuitOpenAfter ?? ScriptWorkerOptions.DefaultCircuitOpenAfter, }, }; - public static OutOfProcessScriptRunner Runner(ScriptingOptions options, ILogger? log = null) + /// A fresh, empty temp directory, deleted by . + public sealed class TempDir : IDisposable + { + public TempDir() + { + Path = System.IO.Path.Combine(System.IO.Path.GetTempPath(), "otopcua-sw-test-" + Guid.NewGuid().ToString("N")[..12]); + Directory.CreateDirectory(Path); + } + + public string Path { get; } + + public void Dispose() + { + try { Directory.Delete(Path, recursive: true); } catch { /* best effort */ } + } + } + + public static OutOfProcessScriptRunner Runner(ScriptingOptions options, ILogger? log = null, + TimeSpan? maintenanceInterval = null, string? markerDirectory = null) => new(options, log ?? new LoggerConfiguration().CreateLogger(), null, - heartbeatInterval: TimeSpan.FromSeconds(1), loadTimeout: TimeSpan.FromSeconds(60)); + heartbeatInterval: TimeSpan.FromSeconds(1), loadTimeout: TimeSpan.FromSeconds(60), + maintenanceInterval: maintenanceInterval, markerDirectory: markerDirectory); public static async Task Eventually(Func probe, TimeSpan timeout, string what) where T : class {