From 03686eb5d7f176f1077d3236d5d3b3aa4a55b8d0 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Tue, 10 Nov 2026 07:18:40 -0500 Subject: [PATCH] =?UTF-8?q?feat(scripting):=20Scripting:Runner=3DShadow=20?= =?UTF-8?q?=E2=80=94=20in-process=20results=20published,=20each=20compiled?= =?UTF-8?q?=20run=20mirrored=20to=20the=20ScriptWorker=20pool=20and=20comp?= =?UTF-8?q?ared=20(lmxopcua#780=20P2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ShadowScriptRunner (Core.Scripting): IScriptShadow.Mirror copies the inputs and queues the shadow run on the pool; compares value (structural), CLR type, status and failure class (compile / budget / threw+message / unsupported_type); runner-fault shadow outcomes (Unavailable, worker busy) are not compared. Bounded: Scripting:Shadow:MaxInFlight (256) and one per script; drops counted (behind | key_busy). - Metrics otopcua.script.shadow.{compared,mismatch{script},dropped,inflight}; one rate-limited Warning per script and reason with a diff summary (Scripting:Shadow:MismatchLogsPerMinute, re-armed at the seam's deploy). - The VirtualTag, Calculation and scripted-alarm seams recognise the shadow and keep their in-process path byte-for-byte (no coalescer, one-at-a-time pump, in-process engine), mirroring after each compiled run. - ScriptRunnerSelection builds the shadow over an OutOfProcessScriptRunner whose lines carry ScriptRunnerRole=Shadow; Status() is null under Shadow so readiness never sees the shadow pool; the script-runner health entry stays Healthy and reports the shadow counters as data. - docker-dev/docker-compose.script-runner-shadow.yml; docs/VirtualTags.md "Shadow mode". - Tests: comparison table, per-script metric + once-per-reason log, rate limit, hung/blocking/throwing shadow never reaches the caller, MaxInFlight drop accounting, input copy; real-worker divergent + unsupported-type scripts counted against exactly those scripts, a worker-only runaway killed/quarantined there with every published value intact, alarm transitions identical to in process across a kill -9 of the shadow worker. Mutation-proved (16 mutants, 15 red; the one survivor added an assertion). Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV --- .../docker-compose.script-runner-shadow.yml | 27 + docs/VirtualTags.md | 31 +- .../ScriptedAlarmEngine.cs | 31 +- .../OutOfProcessScriptRunner.cs | 2 +- .../IScriptRunner.cs | 9 +- .../ScriptRunnerOptions.cs | 57 +- .../ScriptingOptions.cs | 6 + .../ShadowScriptRunner.cs | 510 ++++++++++++++++++ .../CalculationEvaluator.cs | 21 +- .../Configuration/ScriptRunnerSelection.cs | 81 ++- .../Engines/RoslynVirtualTagEvaluator.cs | 24 +- .../Health/ScriptRunnerHealthCheck.cs | 60 ++- .../ShadowScriptRunnerTests.cs | 343 ++++++++++++ .../ScriptRunnerSelectionTests.cs | 48 +- ...lynVirtualTagEvaluatorScriptRunnerTests.cs | 34 ++ .../Health/ScriptRunnerHealthCheckTests.cs | 19 + .../ShadowModeTests.cs | 226 ++++++++ 17 files changed, 1477 insertions(+), 52 deletions(-) create mode 100644 docker-dev/docker-compose.script-runner-shadow.yml create mode 100644 src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ShadowScriptRunner.cs create mode 100644 tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ShadowScriptRunnerTests.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/ShadowModeTests.cs diff --git a/docker-dev/docker-compose.script-runner-shadow.yml b/docker-dev/docker-compose.script-runner-shadow.yml new file mode 100644 index 00000000..f9804e7d --- /dev/null +++ b/docker-dev/docker-compose.script-runner-shadow.yml @@ -0,0 +1,27 @@ +# docker-dev/docker-compose.script-runner-shadow.yml — optional overlay: run the central pair in SHADOW mode +# (lmxopcua#780 P2, Scripting:Runner=Shadow). Every compiled VirtualTag / Calculation / scripted-alarm evaluation runs +# IN PROCESS — that is what is published — and is mirrored to the ScriptWorker pool, whose result is only compared. +# Shadow doubles script CPU and keeps the in-process risk: a time-boxed diagnostic, never a production mode. +# +# Up (the image must be built from a branch that has the shadow runner): +# docker compose -f docker-dev/docker-compose.yml -f docker-dev/docker-compose.script-runner-shadow.yml up -d --no-deps central-1 central-2 +# Back to InProcess: re-run `up -d --no-deps central-1 central-2` with the base file alone. +# +# Watch it: `/metrics` on :9000 carries otopcua_script_shadow_compared_total{kind,result}, +# otopcua_script_shadow_mismatch_total{kind,reason,script}, otopcua_script_shadow_dropped_total{kind,reason} and +# otopcua_script_shadow_inflight; each script's first mismatch per reason logs one Warning "Script shadow mismatch: …" +# with a diff summary; `/health/ready`'s `script-runner` entry stays Healthy and carries the shadow counters as data. +x-script-runner-shadow-env: &script-runner-shadow-env + Scripting__Runner: "Shadow" + Scripting__Runner__Seams__0: "VirtualTag" + Scripting__Runner__Seams__1: "Calculation" + Scripting__Runner__Seams__2: "ScriptedAlarm" + Scripting__Worker__PeCacheDir: "/tmp/otopcua-pe-cache" + +services: + central-1: + environment: + <<: *script-runner-shadow-env + central-2: + environment: + <<: *script-runner-shadow-env diff --git a/docs/VirtualTags.md b/docs/VirtualTags.md index c54f0a9b..40ae5fd9 100644 --- a/docs/VirtualTags.md +++ b/docs/VirtualTags.md @@ -45,8 +45,7 @@ The sandbox cannot pre-empt a script. **Memory:** each run's own-thread allocati #### Out-of-process runner (`Scripting:Runner`, lmxopcua#780) `Scripting:Runner` selects where compiled scripts run: **`InProcess`** (the default and the rollback — everything -above, unchanged), **`OutOfProcess`**, or `Shadow` (design P2; not implemented yet, so it runs `InProcess` with a -startup Warning). `Scripting:Runner:Seams` selects the seams (`VirtualTag`, `Calculation`, `ScriptedAlarm`). Unset or +above, unchanged), **`OutOfProcess`**, or **`Shadow`** (design P2, see "Shadow mode" below). `Scripting:Runner:Seams` selects the seams (`VirtualTag`, `Calculation`, `ScriptedAlarm`). Unset or empty means `VirtualTag` + `Calculation`: the **`ScriptedAlarm` seam runs out of process only when it is listed explicitly** (alarms go last, design §14 #10). Under `OutOfProcess` the selected seams hand their compiled runs to `OutOfProcessScriptRunner` (`Core.Scripting.OutOfProcess`), which runs them in a pool of `ZB.MOM.WW.OtOpcUa.ScriptWorker` @@ -156,6 +155,34 @@ processes: Pinned by `tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests` (`Category=ScriptWorker`: real worker processes). +#### Shadow mode (`Scripting:Runner=Shadow`, lmxopcua#780 P2) + +A time-boxed diagnostic before the flip to `OutOfProcess`. The selected seams (same `Scripting:Runner:Seams` rule: +unset = VirtualTag + Calculation, `ScriptedAlarm` only when listed) keep their **in-process path, unchanged — that is +what is published** — and each compiled evaluation is then mirrored to the `ScriptWorker` pool by `ShadowScriptRunner` +(`Core.Scripting`), whose result is only compared. The passthrough fast path is never mirrored (it runs no user code). + +- **Compared:** value (structural for arrays), exact CLR type, status (Ok vs failed) and failure class (`compile`, + `budget` = timed out / killed / refused / quarantined, `threw` with its message, `unsupported_type`). A shadow outcome + that is the runner's own fault (worker restarted, none ready, worker busy) is **not compared** + (`result = shadow_unavailable`). An input or result the protocol cannot carry (design §14 #11) counts as a mismatch + with reason `unsupported_type` — out of process that evaluation would publish Bad. +- **Never reaches a published value or readiness.** The mirror copies the inputs and queues the shadow run on the pool; + a shadow-side timeout, kill, crash, quarantine, or a runner that hangs or throws stays on the shadow side. + `/health/ready`'s `script-runner` entry stays **Healthy** under Shadow and carries the shadow pool's state and the + counters as data. Kill / quarantine lines of the shadow pool go to the host log tagged `ScriptRunnerRole=Shadow`, + never to the script log; shadow runs' `ctx.Logger` lines are dropped (the in-process run wrote them). +- **Bounded.** At most `Scripting:Shadow:MaxInFlight` (default 256) shadow runs outstanding and at most one per script; + the rest are dropped and counted (`otopcua.script.shadow.dropped{kind, reason = behind | key_busy}`). +- **Metrics:** `otopcua.script.shadow.compared{kind, result = match | mismatch | shadow_unavailable}`, + `otopcua.script.shadow.mismatch{kind, reason, script}`, `otopcua.script.shadow.dropped`, `otopcua.script.shadow.inflight`. + The shadow pool's own `otopcua.script.worker.*` / `otopcua.script.quarantined` series are emitted as usual. +- **Log:** the first mismatch of each script for each reason logs one Warning `Script shadow mismatch: + () — published …; shadow …`, re-armed when the seam redeploys, capped at `Scripting:Shadow:MismatchLogsPerMinute` + (default 20) lines a minute; the next line reports how many were suppressed. +- **Cost.** Every selected script runs twice and the in-process risk stays. docker-dev: + `docker-dev/docker-compose.script-runner-shadow.yml` (all three seams). + #### Quarantine: what a client sees (status code) A quarantined script's output reads **`BadInternalError`** on OPC UA, the same code as a script that diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/ScriptedAlarmEngine.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/ScriptedAlarmEngine.cs index 5204b00f..b7e79c8b 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/ScriptedAlarmEngine.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/ScriptedAlarmEngine.cs @@ -87,6 +87,7 @@ public sealed class ScriptedAlarmEngine : IDisposable /// the pre-emptible worker pool, and a re-evaluation pass submits its whole dirty set as one batch. /// private readonly IScriptRunner? _runner; + private readonly IScriptShadow? _shadow; // Alarms whose predicate came back Unavailable (the runner, not the script, failed) and that are waiting for the // runner's Recovery to re-evaluate them once. Guarded by _dirtyLock. @@ -255,7 +256,9 @@ public sealed class ScriptedAlarmEngine : IDisposable /// silently ran with because the host never passed a value. /// Run-guard budget + quarantine threshold; null ⇒ defaults. /// lmxopcua#780 — the out-of-process runner when the ScriptedAlarm seam runs out of - /// process; null (the default) keeps the in-process guard path, unchanged. + /// process; null (the default) keeps the in-process guard path, unchanged. Under Scripting:Runner=Shadow it is an + /// : the in-process path stays the one that drives the state machine and each predicate run + /// is mirrored to it for comparison (P2). public ScriptedAlarmEngine( ITagUpstreamSource upstream, IAlarmStateStore store, @@ -266,7 +269,8 @@ public sealed class ScriptedAlarmEngine : IDisposable ScriptingOptions? scripting = null, IScriptRunner? scriptRunner = null) { - _runner = scriptRunner; + _shadow = scriptRunner as IScriptShadow; + _runner = _shadow is null ? scriptRunner : null; _upstream = upstream ?? throw new ArgumentNullException(nameof(upstream)); _store = store ?? throw new ArgumentNullException(nameof(store)); _loggerFactory = loggerFactory ?? throw new ArgumentNullException(nameof(loggerFactory)); @@ -286,6 +290,9 @@ public sealed class ScriptedAlarmEngine : IDisposable /// The out-of-process runner predicates run through, or null in process (diagnostics / tests). public IScriptRunner? ScriptRunner => _runner; + /// The shadow predicate runs are mirrored to (Scripting:Runner=Shadow), or null. + public IScriptShadow? ScriptShadow => _shadow; + /// /// The alarm ids whose predicate is quarantined — by the in-process guard, or by the out-of-process runner when /// one is in use (diagnostics / tests). @@ -366,6 +373,7 @@ public sealed class ScriptedAlarmEngine : IDisposable // its own through the guard's source identity; this covers "redeployed to retry"). _guard.ResetAll(); _runner?.Reset(ScriptContextKind.AlarmPredicate); + _shadow?.Reset(ScriptContextKind.AlarmPredicate); lock (_dirtyLock) _recoveryArmed.Clear(); var compileFailures = new List(); @@ -920,23 +928,29 @@ public sealed class ScriptedAlarmEngine : IDisposable } catch (ScriptTimeoutException tex) { + MirrorPredicate(state, scratch, ScriptRunOutcome.Failed( + $"{ZB.MOM.WW.OtOpcUa.Commons.Engines.ScriptEvalFailureReasons.TimedOutPrefix}{tex.Timeout.TotalSeconds:F1}s")); return PredicateDidNotComplete(state, seed, scratch, $"timed out after {tex.Timeout.TotalMilliseconds:F0} ms", pendingEmissions); } catch (ScriptRunRefusedException ex) { + MirrorPredicate(state, scratch, ScriptRunOutcome.Failed(ex.Message)); return PredicateDidNotComplete(state, seed, scratch, ex.Message, pendingEmissions); } catch (ScriptQuarantinedException ex) { + MirrorPredicate(state, scratch, ScriptRunOutcome.Failed(ex.Message)); return PredicateDidNotComplete(state, seed, scratch, ex.Message, pendingEmissions); } catch (Exception ex) { + MirrorPredicate(state, scratch, ScriptRunOutcome.Failed($"script threw: {ex.Message}")); scratch.LastNonCompletion = null; state.Logger.Error(ex, "Alarm predicate threw — state unchanged"); return seed; } + MirrorPredicate(state, scratch, ScriptRunOutcome.Ok(predicateTrue)); scratch.LastNonCompletion = null; // Track the bucket only once the predicate has COMPLETED (I3). The tracked bucket is the last @@ -963,6 +977,19 @@ public sealed class ScriptedAlarmEngine : IDisposable return result.State; } + /// + /// lmxopcua#780 P2 — Shadow: mirrors one in-process predicate run, after it has driven the state machine's input, + /// to the shadow runner for comparison. Never blocks; the inputs are copied before it returns, so the alarm's + /// reused read cache can be refilled at once. + /// + private void MirrorPredicate(AlarmState state, AlarmScratch scratch, ScriptRunOutcome published) + { + if (_shadow is null) return; + _shadow.Mirror(new ScriptInvocation( + state.Definition.AlarmId, ScriptContextKind.AlarmPredicate, state.Definition.PredicateScriptSource, + scratch.ReadCache, _scriptTimeout), published); + } + /// /// The out-of-process re-evaluation pass (lmxopcua#780, design §4.5): every dirty alarm's predicate goes to the /// runner in ONE batch, then the results are applied — Part 9 transition, persist-before-update, emissions — in 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 8cf22db5..1892febc 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 @@ -76,7 +76,7 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, public const string WorkerFolderName = "script-worker"; /// Failure reason when the worker could not take more work. - public const string WorkerBusyReason = "script worker busy"; + public const string WorkerBusyReason = ScriptRunOutcome.WorkerBusyReason; /// PE cache entries unused for this long are pruned at startup. internal static readonly TimeSpan PeCacheMaxAge = TimeSpan.FromDays(30); 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 118eaba7..1a1e7921 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/IScriptRunner.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/IScriptRunner.cs @@ -122,6 +122,9 @@ public sealed record ScriptRunOutcome(ScriptRunStatus Status, object? Value, str /// Reason when no worker became ready within . public const string RunnerUnavailableReason = "script runner unavailable"; + /// Reason when the worker could not take more work (the out-of-process runner's back-pressure answer). + public const string WorkerBusyReason = "script worker busy"; + /// /// 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 @@ -175,7 +178,11 @@ public static class ScriptRunOutcomeExtensions /// public interface IScriptRunnerProvider { - /// The runner must use, or null for its in-process path. + /// + /// The runner must use, or null for its in-process path. Under + /// Scripting:Runner=Shadow this is a : the seam recognises it as an + /// , keeps its in-process path and mirrors each evaluation to it. + /// /// The seam. /// The runner, or null. IScriptRunner? ForSeam(ScriptSeam seam); 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 98a33c4a..4d1c1539 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptRunnerOptions.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptRunnerOptions.cs @@ -10,8 +10,10 @@ public enum ScriptRunnerMode InProcess = 0, /// - /// Run in process AND submit to the worker, comparing the two (design P2). Not implemented yet: a node - /// configured for it runs and logs a startup Warning. + /// Run in process — that result is the one published — AND mirror each evaluation of the selected seams to the + /// worker pool, comparing the two (design P2, ). A time-boxed diagnostic: it + /// doubles script CPU and keeps the in-process risk. Nothing on the shadow side reaches a published value or + /// readiness. /// Shadow = 1, @@ -251,3 +253,54 @@ public sealed class ScriptWorkerOptions return null; } } + +/// +/// The shadow comparison's knobs (the Scripting:Shadow section, lmxopcua#780 P2). Only consulted when +/// is . +/// +public sealed class ScriptShadowOptions +{ + /// Default : 256. + public const int DefaultMaxInFlight = 256; + + /// Default : 20. + public const int DefaultMismatchLogsPerMinute = 20; + + /// + /// Shadow runs outstanding at once (all seams). An evaluation that would exceed it is not mirrored and is counted + /// as dropped (reason = behind): the shadow is best effort and never queues without bound. Default 256; + /// 1–100 000. + /// + public int MaxInFlight { get; set; } = DefaultMaxInFlight; + + /// + /// Mismatch Warning lines per minute (each script logs once per mismatch reason until its seam redeploys; this + /// caps a burst of many scripts). Default 20; 0 (counted, never logged) – 10 000. + /// + public int MismatchLogsPerMinute { get; set; } = DefaultMismatchLogsPerMinute; + + /// Returns a clamped copy; one line per out-of-range key in . + /// The clamps applied, naming each key. + /// A new, clamped instance. + public ScriptShadowOptions Sanitized(out IReadOnlyList adjustments) + { + const string prefix = ScriptingOptions.SectionName + ":Shadow:"; + var notes = new List(); + var inFlight = MaxInFlight; + if (inFlight < 1 || inFlight > 100_000) + { + var clamped = Math.Clamp(inFlight, 1, 100_000); + notes.Add($"{prefix}{nameof(MaxInFlight)} {inFlight} is outside 1–100000; clamped to {clamped}."); + inFlight = clamped; + } + var logs = MismatchLogsPerMinute; + if (logs < 0 || logs > 10_000) + { + var clamped = Math.Clamp(logs, 0, 10_000); + notes.Add($"{prefix}{nameof(MismatchLogsPerMinute)} {logs} is outside 0–10000; clamped to {clamped}."); + logs = clamped; + } + adjustments = notes; + return new ScriptShadowOptions { MaxInFlight = inFlight, MismatchLogsPerMinute = logs }; + } +} diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptingOptions.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptingOptions.cs index cb657b0b..3e141e10 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptingOptions.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptingOptions.cs @@ -89,6 +89,9 @@ public sealed class ScriptingOptions /// The out-of-process worker pool's knobs (Scripting:Worker). public ScriptWorkerOptions Worker { get; set; } = new(); + /// The shadow comparison's knobs (Scripting:Shadow), for . + public ScriptShadowOptions Shadow { get; set; } = new(); + /// Whether is selected for out-of-process execution by these options. /// The seam. /// true when is and the seam is in @@ -185,6 +188,8 @@ public sealed class ScriptingOptions var worker = (Worker ?? new ScriptWorkerOptions()).Sanitized(out var workerNotes); notes.AddRange(workerNotes); + var shadow = (Shadow ?? new ScriptShadowOptions()).Sanitized(out var shadowNotes); + notes.AddRange(shadowNotes); adjustments = notes; return new ScriptingOptions @@ -196,6 +201,7 @@ public sealed class ScriptingOptions Runner = Runner, RunnerSeams = RunnerSeams is null ? null : [.. RunnerSeams], Worker = worker, + Shadow = shadow, }; } } diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ShadowScriptRunner.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ShadowScriptRunner.cs new file mode 100644 index 00000000..403598d5 --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ShadowScriptRunner.cs @@ -0,0 +1,510 @@ +using System.Collections; +using System.Collections.Concurrent; +using System.Diagnostics.Metrics; +using System.Globalization; +using Serilog; +using Serilog.Core; +using ZB.MOM.WW.OtOpcUa.Commons.Engines; +using ZB.MOM.WW.OtOpcUa.Commons.Observability; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Core.Scripting; + +/// +/// The mirror half of Scripting:Runner=Shadow (lmxopcua#780 P2, design §12): a seam that has already produced +/// — and published — its in-process outcome hands the same evaluation here, and it is run again on the shadow +/// runner (the out-of-process worker pool) and compared. +/// +/// +/// A seam receives its as the it is constructed with (the +/// node's runner selection hands out , which is both). A seam that recognises it keeps +/// its own in-process path, unchanged, and calls after each compiled evaluation. +/// +public interface IScriptShadow +{ + /// + /// Mirrors one evaluation whose outcome the seam has already published. Never + /// blocks, never throws, and nothing it does reaches the seam: the shadow run is queued on the pool, or dropped + /// (and counted) when the shadow side is behind. + /// + /// The evaluation as the seam ran it. Its inputs are copied before this returns, so the + /// caller may reuse its read cache at once. + /// The in-process outcome — the source of truth. + void Mirror(ScriptInvocation invocation, ScriptRunOutcome published); + + /// The seam's deploy boundary: lifts the shadow runner's quarantines for and + /// re-arms its once-per-script mismatch log lines. + /// The seam's context kind. + void Reset(ScriptContextKind kind); +} + +/// How one shadow comparison came out. +public enum ScriptShadowVerdict +{ + /// The shadow run agreed with the published outcome. + Match = 0, + + /// The shadow run disagreed; names how. + Mismatch = 1, + + /// The shadow runner could not run the script (worker restarted, none ready, worker busy): not compared. + ShadowUnavailable = 2, +} + +/// The result of comparing a published outcome with its shadow. +/// Match, mismatch, or not compared. +/// For a mismatch, the metric reason: value, type, status, +/// failure_kind, exception or unsupported_type. +/// A one-line description of the published outcome. +/// A one-line description of the shadow outcome. +public sealed record ScriptShadowComparison(ScriptShadowVerdict Verdict, string? Reason, string Published, string Shadow) +{ + /// The diff summary a mismatch log line carries. + public string Summary => $"published {Published}; shadow {Shadow}"; +} + +/// Counters of one since it started. +/// Shadow runs that completed and were compared (match or mismatch). +/// Comparisons that agreed. +/// Comparisons that disagreed. +/// Shadow runs the runner could not run (not compared). +/// Evaluations not mirrored because shadow +/// runs were already outstanding. +/// Evaluations not mirrored because the same script's previous shadow run was still +/// outstanding. +/// Shadow runs outstanding now. +public sealed record ScriptShadowStats( + long Compared, long Matched, long Mismatched, long ShadowUnavailable, long DroppedBehind, long DroppedKeyBusy, int InFlight) +{ + /// Every evaluation that was not mirrored. + public long Dropped => DroppedBehind + DroppedKeyBusy; +} + +/// One script's mismatch record. +/// The seam's context kind. +/// The script key (un-namespaced). +/// Mismatches counted for it. +/// The reason of the latest one. +/// The diff summary of the latest one. +public sealed record ScriptShadowMismatch(ScriptContextKind Kind, string Key, long Count, string LastReason, string LastSummary); + +/// +/// Scripting:Runner=Shadow (lmxopcua#780 P2, design §12): every evaluation of a shadowed seam runs IN PROCESS — +/// that result is the one published — and is mirrored, best effort, to a shadow (the +/// out-of-process worker pool), whose result is only compared: value, CLR type, status and failure kind. +/// +/// +/// +/// The published value never depends on the shadow. copies the inputs and queues +/// the shadow run on the pool; a shadow-side timeout, kill, crash, quarantine, a runner that hangs or throws, all +/// stay here. +/// Bounded. At most shadow runs are outstanding and at +/// most one per script; anything beyond that is dropped and counted (otopcua.script.shadow.dropped, +/// reason = behind | key_busy). One per script also keeps the out-of-process runner's +/// single-flight rule from refusing a shadow run and reporting a false mismatch. +/// Counted per script. otopcua.script.shadow.compared{kind, result} and +/// otopcua.script.shadow.mismatch{kind, reason, script}; keeps the latest diff per +/// script. A mismatch logs one Warning with the diff summary the first time each script mismatches for each +/// reason (re-armed at the seam's deploy boundary), and at most +/// such lines a minute. +/// Not compared: a shadow outcome that is the runner's fault (, +/// or the worker busy) — that is node health, not a script divergence. +/// Unsupported types (design §14 #11): a value the protocol cannot carry makes the shadow answer +/// unsupported type, which counts as a mismatch with reason unsupported_type — the census the flip to +/// OutOfProcess needs, since out of process that evaluation would publish Bad. +/// +/// As an , runs an and mirrors +/// its outcome, for callers that only speak the runner interface. The three production seams recognise +/// instead and keep their own in-process code, so what they publish is byte-for-byte +/// what InProcess publishes. +/// +public sealed class ShadowScriptRunner : IScriptRunner, IScriptShadow, IDisposable +{ + private static readonly Counter ComparedCounter = OtOpcUaTelemetry.Meter.CreateCounter( + "otopcua.script.shadow.compared", unit: "{eval}", + description: "Shadow-mode evaluations whose shadow run completed (tags: kind; result = match | mismatch | shadow_unavailable)."); + + private static readonly Counter MismatchCounter = OtOpcUaTelemetry.Meter.CreateCounter( + "otopcua.script.shadow.mismatch", unit: "{eval}", + description: "Shadow-mode evaluations whose out-of-process result differed from the published in-process one " + + "(tags: kind; reason = value | type | status | failure_kind | exception | unsupported_type; script)."); + + private static readonly Counter DroppedCounter = OtOpcUaTelemetry.Meter.CreateCounter( + "otopcua.script.shadow.dropped", unit: "{eval}", + description: "Shadow-mode evaluations not mirrored because the shadow side was behind (tags: kind; reason = behind | key_busy)."); + + private static readonly ConcurrentDictionary LiveRunners = new(); + + static ShadowScriptRunner() + { + OtOpcUaTelemetry.Meter.CreateObservableGauge("otopcua.script.shadow.inflight", + static () => LiveRunners.Keys.Select(r => new Measurement(Volatile.Read(ref r._inFlight))), + unit: "{eval}", description: "Shadow-mode runs outstanding on the shadow runner."); + } + + private readonly IScriptRunner _shadow; + private readonly InProcessScriptRunner _primary; + private readonly ScriptShadowOptions _options; + private readonly ILogger _log; + private readonly Func _clock; + private readonly CancellationTokenSource _stopping = new(); + private readonly ConcurrentDictionary _busy = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary _logged = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary _mismatches = new(StringComparer.Ordinal); + private readonly object _logLock = new(); + private DateTime _logWindowStart; + private int _logsInWindow; + private long _logsSuppressed; + private int _inFlight; + private long _compared, _matched, _mismatched, _unavailable, _droppedBehind, _droppedBusy; + private int _disposed; + + /// Initializes the shadow runner. + /// The runner every mirrored evaluation is run on again (the out-of-process pool). Not owned. + /// The node's Scripting options (the in-process guard's knobs and Scripting:Shadow); + /// null ⇒ defaults. + /// Where mismatch lines go; null ⇒ Serilog's global logger. + /// The mismatch-log rate limiter's clock (tests); null ⇒ . + public ShadowScriptRunner(IScriptRunner shadow, ScriptingOptions? scripting = null, ILogger? log = null, Func? clock = null) + { + _shadow = shadow ?? throw new ArgumentNullException(nameof(shadow)); + var options = (scripting ?? new ScriptingOptions()).Sanitized(out _); + _options = options.Shadow; + _log = (log ?? Log.Logger).ForContext(); + _clock = clock ?? (() => DateTime.UtcNow); + _primary = new InProcessScriptRunner(options, log); + _logWindowStart = _clock(); + LiveRunners.TryAdd(this, 0); + } + + /// The runner shadow runs go to. + public IScriptRunner Shadow => _shadow; + + /// The counters since start. + /// A snapshot. + public ScriptShadowStats Stats() => new( + Interlocked.Read(ref _compared), Interlocked.Read(ref _matched), Interlocked.Read(ref _mismatched), + Interlocked.Read(ref _unavailable), Interlocked.Read(ref _droppedBehind), Interlocked.Read(ref _droppedBusy), + Volatile.Read(ref _inFlight)); + + /// Every script that has mismatched, with its count and latest diff. + public IReadOnlyCollection Mismatches => _mismatches.Values.ToArray(); + + /// + /// Waits until no shadow run is outstanding (tests and diagnostics: a shadow run is asynchronous to the + /// evaluation that triggered it). + /// + /// How long to wait. + /// true when idle within . + public async Task WaitIdleAsync(TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (Volatile.Read(ref _inFlight) > 0) + { + if (DateTime.UtcNow >= deadline) return false; + await Task.Delay(10).ConfigureAwait(false); + } + return true; + } + + /// + public async Task RunAsync(ScriptInvocation invocation, CancellationToken ct = default) + { + var published = await _primary.RunAsync(invocation, ct).ConfigureAwait(false); + if (!ct.IsCancellationRequested) Mirror(invocation, published); + return published; + } + + /// + public void Mirror(ScriptInvocation invocation, ScriptRunOutcome published) + { + if (invocation is null || published is null || Volatile.Read(ref _disposed) != 0) return; + var scoped = InProcessScriptRunner.ScopedKey(invocation.Kind, invocation.Key); + if (!_busy.TryAdd(scoped, 0)) + { + Drop(invocation.Kind, "key_busy", ref _droppedBusy); + return; + } + if (Interlocked.Increment(ref _inFlight) > _options.MaxInFlight) + { + Release(scoped); + Drop(invocation.Kind, "behind", ref _droppedBehind); + return; + } + + ScriptInvocation copy; + try + { + // The caller may refill its read cache as soon as this returns (the alarm engine reuses one per alarm). + copy = invocation with + { + Inputs = new Dictionary(invocation.Inputs, StringComparer.Ordinal), + // The in-process run already wrote this evaluation's ctx.Logger lines; the shadow's are not repeated. + ScriptLogger = Logger.None, + }; + } + catch (Exception ex) + { + Release(scoped); + _log.Debug(ex, "Script shadow: could not copy the inputs of {ScriptKey}; not mirrored", invocation.Key); + return; + } + + // Never on the caller's thread: a shadow runner may do synchronous work (encode, write) before its first await. + _ = Task.Run(() => RunShadowAsync(scoped, copy, published)); + } + + /// + public void Reset(ScriptContextKind kind) + { + _primary.Reset(kind); + try + { + _shadow.Reset(kind); + } + catch (Exception ex) + { + _log.Warning(ex, "Script shadow: the shadow runner's reset failed for {ScriptKind}", kind); + } + var prefix = InProcessScriptRunner.ScopedKey(kind, string.Empty); + foreach (var key in _logged.Keys) + if (key.StartsWith(prefix, StringComparison.Ordinal)) _logged.TryRemove(key, out _); + } + + /// + public IReadOnlyCollection QuarantinedKeys(ScriptContextKind kind) => _primary.QuarantinedKeys(kind); + + /// + public void Dispose() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) return; + LiveRunners.TryRemove(this, out _); + _stopping.Cancel(); + _primary.Dispose(); + _stopping.Dispose(); + } + + /// + /// Compares a published outcome with its shadow (pure; the table the mismatch reasons are tested from). + /// + /// The in-process outcome. + /// The shadow runner's outcome. + /// The comparison. + public static ScriptShadowComparison Compare(ScriptRunOutcome published, ScriptRunOutcome shadow) + { + ArgumentNullException.ThrowIfNull(published); + ArgumentNullException.ThrowIfNull(shadow); + var p = Describe(published); + var s = Describe(shadow); + + if (shadow.Status == ScriptRunStatus.Unavailable + || (shadow.Status == ScriptRunStatus.Failed && shadow.Reason == ScriptRunOutcome.WorkerBusyReason)) + return new ScriptShadowComparison(ScriptShadowVerdict.ShadowUnavailable, null, p, s); + + ScriptShadowComparison Mismatch(string reason) => new(ScriptShadowVerdict.Mismatch, reason, p, s); + var match = new ScriptShadowComparison(ScriptShadowVerdict.Match, null, p, s); + + if (published.Status == ScriptRunStatus.Ok && shadow.Status == ScriptRunStatus.Ok) + { + if (published.Value is null && shadow.Value is null) return match; + if (published.Value is null || shadow.Value is null) return Mismatch("value"); + if (published.Value.GetType() != shadow.Value.GetType()) return Mismatch("type"); + return ValuesEqual(published.Value, shadow.Value) ? match : Mismatch("value"); + } + + var publishedClass = published.Status == ScriptRunStatus.Ok ? null : FailureClass(published.Reason); + var shadowClass = shadow.Status == ScriptRunStatus.Ok ? null : FailureClass(shadow.Reason); + if (shadowClass == UnsupportedTypeClass && publishedClass != UnsupportedTypeClass) return Mismatch("unsupported_type"); + if (publishedClass is null || shadowClass is null) return Mismatch("status"); + if (publishedClass != shadowClass) return Mismatch("failure_kind"); + if (publishedClass == ThrewClass && !string.Equals(published.Reason, shadow.Reason, StringComparison.Ordinal)) + return Mismatch("exception"); + return match; + } + + /// + /// The failure class a reason belongs to, for comparison: compile, budget (timed out, killed, + /// refused behind a run still going, quarantined — the two runners reach these by different routes for the same + /// runaway), threw, unsupported_type, cancelled or other. + /// + /// A failure reason. + /// The class. + public static string FailureClass(string? reason) + { + var r = reason ?? string.Empty; + if (r.StartsWith("compile error:", StringComparison.Ordinal) + || r.StartsWith("sandbox violation:", StringComparison.Ordinal) + || r.StartsWith("compile failure:", StringComparison.Ordinal)) + return "compile"; + if (r.StartsWith(ScriptEvalFailureReasons.TimedOutPrefix, StringComparison.Ordinal) + || r.StartsWith("timed out after ", StringComparison.Ordinal) + || r == ScriptRunGuard.PreviousRunStillRunningReason + || r.StartsWith(ScriptRunGuard.QuarantinedReasonPrefix, StringComparison.Ordinal)) + return "budget"; + if (r.StartsWith("script threw:", StringComparison.Ordinal)) return ThrewClass; + if (r.StartsWith("unsupported type:", StringComparison.Ordinal)) return UnsupportedTypeClass; + if (r.StartsWith("evaluation cancelled", StringComparison.Ordinal)) return "cancelled"; + return "other"; + } + + private const string ThrewClass = "threw"; + private const string UnsupportedTypeClass = "unsupported_type"; + private const int MaxDescribedLength = 160; + + private async Task RunShadowAsync(string scoped, ScriptInvocation invocation, ScriptRunOutcome published) + { + try + { + ScriptRunOutcome shadow; + try + { + shadow = await _shadow.RunAsync(invocation, _stopping.Token).ConfigureAwait(false); + } + catch (Exception ex) + { + // A runner must not throw for a script fault; one that does is the runner's problem, not a divergence. + shadow = ScriptRunOutcome.Unavailable($"shadow runner threw: {ex.Message}"); + } + if (Volatile.Read(ref _disposed) != 0) return; + Record(invocation, published, shadow); + } + catch (Exception ex) + { + _log.Debug(ex, "Script shadow: comparing {ScriptKey} failed", invocation.Key); + } + finally + { + Release(scoped); + } + } + + private void Record(ScriptInvocation invocation, ScriptRunOutcome published, ScriptRunOutcome shadow) + { + var comparison = Compare(published, shadow); + var kindTag = new KeyValuePair("kind", KindTag(invocation.Kind)); + switch (comparison.Verdict) + { + case ScriptShadowVerdict.ShadowUnavailable: + Interlocked.Increment(ref _unavailable); + ComparedCounter.Add(1, kindTag, new KeyValuePair("result", "shadow_unavailable")); + return; + case ScriptShadowVerdict.Match: + Interlocked.Increment(ref _compared); + Interlocked.Increment(ref _matched); + ComparedCounter.Add(1, kindTag, new KeyValuePair("result", "match")); + return; + } + + Interlocked.Increment(ref _compared); + Interlocked.Increment(ref _mismatched); + var reason = comparison.Reason!; + ComparedCounter.Add(1, kindTag, new KeyValuePair("result", "mismatch")); + MismatchCounter.Add(1, kindTag, + new KeyValuePair("reason", reason), + new KeyValuePair("script", invocation.Key)); + + var scoped = InProcessScriptRunner.ScopedKey(invocation.Kind, invocation.Key); + _mismatches.AddOrUpdate(scoped, + _ => new ScriptShadowMismatch(invocation.Kind, invocation.Key, 1, reason, comparison.Summary), + (_, prior) => prior with { Count = prior.Count + 1, LastReason = reason, LastSummary = comparison.Summary }); + + if (!_logged.TryAdd($"{scoped}|{reason}", 0)) return; + long suppressed; + lock (_logLock) + { + var now = _clock(); + if (now - _logWindowStart >= TimeSpan.FromMinutes(1)) + { + _logWindowStart = now; + _logsInWindow = 0; + } + if (_logsInWindow >= _options.MismatchLogsPerMinute) + { + _logsSuppressed++; + return; + } + _logsInWindow++; + suppressed = _logsSuppressed; + _logsSuppressed = 0; + } + _log.Warning( + "Script shadow mismatch: {ScriptKind} {ScriptKey} ({MismatchReason}) — {MismatchSummary}{Suppressed}", + invocation.Kind, invocation.Key, reason, comparison.Summary, + suppressed > 0 ? $" ({suppressed} earlier mismatch line(s) suppressed by the rate limit)" : string.Empty); + } + + private void Release(string scoped) + { + _busy.TryRemove(scoped, out _); + Interlocked.Decrement(ref _inFlight); + } + + private static void Drop(ScriptContextKind kind, string reason, ref long counter) + { + Interlocked.Increment(ref counter); + DroppedCounter.Add(1, + new KeyValuePair("kind", KindTag(kind)), + new KeyValuePair("reason", reason)); + } + + private static bool ValuesEqual(object a, object b) + => a is Array || b is Array + ? StructuralComparisons.StructuralEqualityComparer.Equals(a, b) + : a.Equals(b); + + /// A one-line description of an outcome for a mismatch summary. + /// The outcome. + /// e.g. Ok Int32 42, Failed [threw] script threw: …. + public static string Describe(ScriptRunOutcome outcome) + { + ArgumentNullException.ThrowIfNull(outcome); + var text = outcome.Status switch + { + ScriptRunStatus.Ok => outcome.Value is null ? "Ok null" : $"Ok {outcome.Value.GetType().Name} {Format(outcome.Value)}", + ScriptRunStatus.Unavailable => $"Unavailable {outcome.Reason}", + _ => $"Failed [{FailureClass(outcome.Reason)}] {outcome.Reason}", + }; + return text.Length <= MaxDescribedLength ? text : string.Concat(text.AsSpan(0, MaxDescribedLength), "…"); + } + + private static string Format(object value) => value switch + { + string s => $"\"{s}\"", + DateTime d => $"{d.ToString("O", CultureInfo.InvariantCulture)} ({d.Kind})", + Array a => $"[{string.Join(", ", a.Cast().Take(8).Select(v => v is null ? "null" : Convert.ToString(v, CultureInfo.InvariantCulture)))}{(a.Length > 8 ? ", …" : string.Empty)}]", + IFormattable f => f.ToString(null, CultureInfo.InvariantCulture), + _ => value.ToString() ?? string.Empty, + }; + + private static string KindTag(ScriptContextKind kind) => kind switch + { + ScriptContextKind.VirtualTag => "virtual_tag", + ScriptContextKind.Calculation => "calculation", + ScriptContextKind.AlarmPredicate => "alarm", + _ => "other", + }; +} + +/// Maps a value seam's in-process result onto the shadow comparison's outcome type. +public static class ScriptShadowOutcomes +{ + /// + /// The equivalent of an in-process , or null when + /// the result is the host's own lifecycle (an evaluator disposed by a redeploy, a shutdown cancellation) rather + /// than the script's — those are not mirrored. + /// + /// The seam's in-process result. + /// The outcome to mirror, or null. + public static ScriptRunOutcome? ToMirroredOutcome(this VirtualTagEvalResult result) + { + ArgumentNullException.ThrowIfNull(result); + if (result.Success) return ScriptRunOutcome.Ok(result.Value); + var reason = result.Reason ?? "evaluator failure"; + if (reason is "evaluator disposed" or ScriptEvalFailureReasons.EvaluatorDisposedDuringEvaluation + || reason.StartsWith("evaluation cancelled", StringComparison.Ordinal)) + return null; + return result.StatusCode == Commons.OpcUa.OpcUaStatus.BadResourceUnavailable + ? ScriptRunOutcome.Unavailable(reason) + : ScriptRunOutcome.Failed(reason); + } +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationEvaluator.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationEvaluator.cs index 01a1fd05..2b2b7780 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationEvaluator.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationEvaluator.cs @@ -60,6 +60,7 @@ public sealed class CalculationEvaluator : IScriptCacheOwner, IScriptQuarantineO private readonly TimeSpan _runTimeout; private readonly ScriptRunGuard _guard; private readonly IScriptRunner? _runner; + private readonly IScriptShadow? _shadow; // Negative compile cache (R07:P4), keyed by source hash. A compile error is a property of the SOURCE, // which never changes between triggers (it is fixed in the deploy artifact) — yet every trigger of a @@ -78,7 +79,9 @@ public sealed class CalculationEvaluator : IScriptCacheOwner, IScriptQuarantineO /// Wall-clock budget per script run; defaults to 2 seconds (parity with the VT evaluator). /// Run-guard budget + quarantine threshold; null ⇒ defaults. /// lmxopcua#780 — the out-of-process runner when the Calculation seam runs out of - /// process; null (the default) keeps the in-process guard path, unchanged. + /// process; null (the default) keeps the in-process guard path, unchanged. Under Scripting:Runner=Shadow it is an + /// : the in-process path stays the published one and each compiled evaluation is mirrored to + /// it for comparison (P2). public CalculationEvaluator( ILogger logger, ScriptRootLogger scriptRoot, @@ -86,7 +89,8 @@ public sealed class CalculationEvaluator : IScriptCacheOwner, IScriptQuarantineO ScriptingOptions? scripting = null, IScriptRunner? scriptRunner = null) { - _runner = scriptRunner; + _shadow = scriptRunner as IScriptShadow; + _runner = _shadow is null ? scriptRunner : null; _logger = logger ?? throw new ArgumentNullException(nameof(logger)); _scriptRoot = (scriptRoot ?? throw new ArgumentNullException(nameof(scriptRoot))).Logger; _runTimeout = runTimeout ?? TimeSpan.FromSeconds(2); @@ -202,6 +206,17 @@ public sealed class CalculationEvaluator : IScriptCacheOwner, IScriptQuarantineO return outcome.ToVirtualTagEvalResult(); } + var result = await RunInProcessAsync(calcTagId, scriptSource, context, ct).ConfigureAwait(false); + // lmxopcua#780 P2 — Shadow: `result` is what gets published; the same evaluation is mirrored to the worker pool + // and only compared. Mirror never blocks and nothing it does reaches this result. + if (_shadow is not null && result.ToMirroredOutcome() is { } mirrored) + _shadow.Mirror(new ScriptInvocation(calcTagId, ScriptContextKind.Calculation, scriptSource, readCache, _runTimeout), mirrored); + return result; + } + + private async Task RunInProcessAsync( + string calcTagId, string scriptSource, VirtualTagContext context, CancellationToken ct) + { // A source that already failed to compile fails the same way without touching Roslyn (R07:P4). var sourceKey = HashSource(scriptSource); if (_compileFailures.TryGetValue(sourceKey, out var knownFailure)) @@ -366,6 +381,7 @@ public sealed class CalculationEvaluator : IScriptCacheOwner, IScriptQuarantineO // (an edited source already lifts its own through the guard's source identity). _guard.ResetAll(); _runner?.Reset(ScriptContextKind.Calculation); + _shadow?.Reset(ScriptContextKind.Calculation); } /// @@ -373,6 +389,7 @@ public sealed class CalculationEvaluator : IScriptCacheOwner, IScriptQuarantineO { _guard.ResetAll(); _runner?.Reset(ScriptContextKind.Calculation); + _shadow?.Reset(ScriptContextKind.Calculation); } /// 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 ebac1ffd..ca50071f 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/ScriptRunnerSelection.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/ScriptRunnerSelection.cs @@ -9,8 +9,10 @@ namespace ZB.MOM.WW.OtOpcUa.Host.Configuration; /// the default every seam gets null and keeps its in-process guard path, /// unchanged. Under the selected seams share one /// , which starts no worker until the first compiled evaluation, so a node with -/// no scripts (or an admin-only node) never launches one. Disposed with the host's service provider, which stops the -/// workers. +/// no scripts (or an admin-only node) never launches one. Under (P2) the selected +/// seams get a over such a runner: they keep their in-process path, which is what they +/// publish, and mirror each compiled evaluation to the worker pool for comparison. Disposed with the host's service +/// provider, which stops the workers. /// public sealed class ScriptRunnerSelection : IScriptRunnerProvider, IAsyncDisposable, IDisposable { @@ -26,25 +28,43 @@ public sealed class ScriptRunnerSelection : IScriptRunnerProvider, IAsyncDisposa public static readonly IReadOnlyList DefaultOutOfProcessSeams = [ScriptSeam.VirtualTag, ScriptSeam.Calculation]; private readonly OutOfProcessScriptRunner? _runner; + private readonly ShadowScriptRunner? _shadow; private readonly HashSet _seams; - private ScriptRunnerSelection(OutOfProcessScriptRunner? runner, IEnumerable seams) + private ScriptRunnerSelection(OutOfProcessScriptRunner? runner, ShadowScriptRunner? shadow, IEnumerable seams) { _runner = runner; + _shadow = shadow; _seams = [.. seams]; } - /// The out-of-process runner, when one is configured. + /// The out-of-process runner, when one is configured (under Shadow, the one shadow runs go to). public IScriptRunner? OutOfProcessRunner => _runner; - /// The out-of-process runner's status (the readiness check), or null when every seam runs in process. + /// The shadow comparison, under ; otherwise null. + public ShadowScriptRunner? Shadow => _shadow; + + /// + /// The out-of-process runner's status (the readiness check), or null when every seam publishes in-process + /// results — , and also , whose worker + /// pool must never move readiness (see ). + /// /// The status, or null. - public ScriptRunnerStatus? Status() => _runner?.Status(); + public ScriptRunnerStatus? Status() => _shadow is null ? _runner?.Status() : null; + + /// Under : the shadow worker pool's status and the comparison counters. + /// The shadow status, or null in any other mode. + public (ScriptRunnerStatus Runner, ScriptShadowStats Stats)? ShadowStatus() + => _shadow is not null && _runner is not null ? (_runner.Status(), _shadow.Stats()) : null; /// The runner must use, or null for its in-process path. /// The seam. - /// The runner, or null. - public IScriptRunner? ForSeam(ScriptSeam seam) => _runner is not null && _seams.Contains(seam) ? _runner : null; + /// The out-of-process runner; under Shadow the ; or null. + public IScriptRunner? ForSeam(ScriptSeam seam) + { + if (_runner is null || !_seams.Contains(seam)) return null; + return _shadow ?? (IScriptRunner)_runner; + } /// /// Startup checks for the runner options: returns the Warnings to log, and throws when the configuration must @@ -63,21 +83,27 @@ public sealed class ScriptRunnerSelection : IScriptRunnerProvider, IAsyncDisposa if (options.Worker.ValidatePaths(contentRootPath) is { } violation) throw new InvalidOperationException($"Scripting misconfiguration: {violation}"); - if (options.Runner == ScriptRunnerMode.Shadow) + var mode = $"Scripting:Runner={options.Runner}"; + var shadow = options.Runner == ScriptRunnerMode.Shadow; + if (shadow) { - warnings.Add("Scripting:Runner=Shadow is not implemented yet (lmxopcua#780 P2); every seam runs InProcess."); - return warnings; + warnings.Add($"{mode}: every selected script runs twice — in process (published) and in the ScriptWorker " + + $"(compared only: {string.Join(", ", RequestedSeams(options).Where(SupportedOutOfProcessSeams.Contains))}). " + + "It doubles script CPU and keeps the in-process risk: a time-boxed diagnostic, not a production mode."); } foreach (var seam in RequestedSeams(options)) { if (!SupportedOutOfProcessSeams.Contains(seam)) - warnings.Add($"Scripting:Runner=OutOfProcess: the {seam} seam cannot run out of process yet (lmxopcua#780); it stays InProcess."); + warnings.Add($"{mode}: the {seam} seam cannot run out of process yet (lmxopcua#780); it stays InProcess."); } var workerPath = OutOfProcessScriptRunner.ResolveWorkerPath(options.Worker); if (!File.Exists(workerPath)) - warnings.Add($"Scripting:Runner=OutOfProcess: the script worker '{workerPath}' does not exist; every out-of-process " + - "script reports BadResourceUnavailable (the runner fails closed, it never falls back to in process)."); + warnings.Add(shadow + ? $"{mode}: the script worker '{workerPath}' does not exist; every shadow run is skipped as shadow_unavailable " + + "(published values are unaffected)." + : $"{mode}: the script worker '{workerPath}' does not exist; every out-of-process " + + "script reports BadResourceUnavailable (the runner fails closed, it never falls back to in process)."); return warnings; } @@ -89,17 +115,34 @@ public sealed class ScriptRunnerSelection : IScriptRunnerProvider, IAsyncDisposa public static ScriptRunnerSelection Create(ScriptingOptions options, SerilogLogger log, SerilogLogger? scriptRoot) { ArgumentNullException.ThrowIfNull(options); - if (options.Runner != ScriptRunnerMode.OutOfProcess) return new ScriptRunnerSelection(null, []); + ArgumentNullException.ThrowIfNull(log); + if (options.Runner == ScriptRunnerMode.InProcess) return new ScriptRunnerSelection(null, null, []); var seams = RequestedSeams(options).Where(SupportedOutOfProcessSeams.Contains).ToArray(); - if (seams.Length == 0) return new ScriptRunnerSelection(null, []); - return new ScriptRunnerSelection(new OutOfProcessScriptRunner(options, log, scriptRoot), seams); + if (seams.Length == 0) return new ScriptRunnerSelection(null, null, []); + if (options.Runner == ScriptRunnerMode.OutOfProcess) + return new ScriptRunnerSelection(new OutOfProcessScriptRunner(options, log, scriptRoot), null, seams); + + // Shadow: the worker pool's own lines (kills, quarantines, restarts) are about shadow runs, so they carry + // ScriptRunnerRole=Shadow and go to the host log, never to the operator's script log — the published script was + // not killed or quarantined. ctx.Logger lines of shadow runs are dropped (the in-process run wrote them). + var shadowLog = log.ForContext("ScriptRunnerRole", "Shadow"); + var runner = new OutOfProcessScriptRunner(options, shadowLog, shadowLog); + return new ScriptRunnerSelection(runner, new ShadowScriptRunner(runner, options, log), seams); } /// - public ValueTask DisposeAsync() => _runner?.DisposeAsync() ?? ValueTask.CompletedTask; + public async ValueTask DisposeAsync() + { + _shadow?.Dispose(); + if (_runner is not null) await _runner.DisposeAsync().ConfigureAwait(false); + } /// - public void Dispose() => _runner?.Dispose(); + public void Dispose() + { + _shadow?.Dispose(); + _runner?.Dispose(); + } private static IEnumerable RequestedSeams(ScriptingOptions options) => options.RunnerSeams is { Count: > 0 } seams ? seams.Distinct() : DefaultOutOfProcessSeams; diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Engines/RoslynVirtualTagEvaluator.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Engines/RoslynVirtualTagEvaluator.cs index 0c130b2b..dc440ce2 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Engines/RoslynVirtualTagEvaluator.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Engines/RoslynVirtualTagEvaluator.cs @@ -54,6 +54,7 @@ public sealed class RoslynVirtualTagEvaluator : IVirtualTagEvaluator, IScriptCac private readonly TimeSpan _runTimeout; private readonly ScriptRunGuard _guard; private readonly IScriptRunner? _runner; + private readonly IScriptShadow? _shadow; private readonly ScriptBatchCoalescer? _batches; private bool _disposed; @@ -66,7 +67,8 @@ public sealed class RoslynVirtualTagEvaluator : IVirtualTagEvaluator, IScriptCac /// /// lmxopcua#780 — the out-of-process runner when Scripting:Runner=OutOfProcess selects the VirtualTag seam; /// null (the default, InProcess) keeps the in-process guard path below, unchanged. The passthrough fast path - /// never reaches it. + /// never reaches it. Under Scripting:Runner=Shadow it is an : the in-process path + /// below stays the published one and each compiled evaluation is mirrored to it for comparison (P2). /// public RoslynVirtualTagEvaluator( ILogger logger, @@ -79,13 +81,17 @@ public sealed class RoslynVirtualTagEvaluator : IVirtualTagEvaluator, IScriptCac _scriptRoot = (scriptRoot ?? throw new ArgumentNullException(nameof(scriptRoot))).Logger; _runTimeout = runTimeout ?? (scripting ?? new ScriptingOptions()).Sanitized(out _).RunTimeout; _guard = new ScriptRunGuard(scripting, _scriptRoot); - _runner = scriptRunner; - _batches = scriptRunner is null ? null : new ScriptBatchCoalescer(scriptRunner); + _shadow = scriptRunner as IScriptShadow; + _runner = _shadow is null ? scriptRunner : null; + _batches = _runner is null ? null : new ScriptBatchCoalescer(_runner); } /// The out-of-process runner this evaluator delegates compiled runs to, or null (in process). public IScriptRunner? ScriptRunner => _runner; + /// The shadow compiled runs are mirrored to (Scripting:Runner=Shadow), or null. + public IScriptShadow? ScriptShadow => _shadow; + /// The run guard every compiled evaluation goes through (diagnostics / tests). public ScriptRunGuard RunGuard => _guard; @@ -175,6 +181,16 @@ public sealed class RoslynVirtualTagEvaluator : IVirtualTagEvaluator, IScriptCac return outcome.ToVirtualTagEvalResult(); } + var result = await RunInProcessAsync(virtualTagId, expression, context).ConfigureAwait(false); + // lmxopcua#780 P2 — Shadow: `result` is what gets published; the same evaluation is mirrored to the worker pool + // and only compared. Mirror never blocks and nothing it does reaches this result. + if (_shadow is not null && result.ToMirroredOutcome() is { } mirrored) + _shadow.Mirror(new ScriptInvocation(virtualTagId, ScriptContextKind.VirtualTag, expression, readCache, _runTimeout), mirrored); + return result; + } + + private async Task RunInProcessAsync(string virtualTagId, string expression, VirtualTagContext context) + { // 02/S12: fetch + run inside a bounded loop so an ObjectDisposedException caused by a concurrent // apply-boundary ClearCompiledScripts (which disposed the evaluator between GetOrCompile and the // run's disposed guard) re-fetches and retries exactly once. The loop runs at most twice — the @@ -264,6 +280,7 @@ public sealed class RoslynVirtualTagEvaluator : IVirtualTagEvaluator, IScriptCac // (an edited expression already lifts its own through the guard's source identity). _guard.ResetAll(); _runner?.Reset(ScriptContextKind.VirtualTag); + _shadow?.Reset(ScriptContextKind.VirtualTag); } /// @@ -271,6 +288,7 @@ public sealed class RoslynVirtualTagEvaluator : IVirtualTagEvaluator, IScriptCac { _guard.ResetAll(); _runner?.Reset(ScriptContextKind.VirtualTag); + _shadow?.Reset(ScriptContextKind.VirtualTag); } /// diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/ScriptRunnerHealthCheck.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/ScriptRunnerHealthCheck.cs index e8afc64d..8f2b2ba7 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/ScriptRunnerHealthCheck.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/ScriptRunnerHealthCheck.cs @@ -1,4 +1,5 @@ using Microsoft.Extensions.Diagnostics.HealthChecks; +using ZB.MOM.WW.OtOpcUa.Core.Scripting; using ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess; using ZB.MOM.WW.OtOpcUa.Host.Configuration; @@ -16,26 +17,35 @@ public static class ScriptRunnerHealthDecision /// 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. + /// + /// Under Scripting:Runner=Shadow: the shadow worker pool's status and the comparison counters. Always + /// Healthy — every published value comes from the in-process path, so nothing on the shadow side (a + /// degraded pool, an open circuit, mismatches) may move readiness. It is reported as data only. + /// /// The status, a description and the per-shard data. - public static (HealthStatus Status, string Description, IReadOnlyDictionary Data) Evaluate(ScriptRunnerStatus? status) + public static (HealthStatus Status, string Description, IReadOnlyDictionary Data) Evaluate( + ScriptRunnerStatus? status, (ScriptRunnerStatus Runner, ScriptShadowStats Stats)? shadow = null) { + if (shadow is { } sh) + { + var shadowData = RunnerData(sh.Runner, "shadow."); + shadowData["mode"] = "Shadow"; + shadowData["shadow.compared"] = sh.Stats.Compared; + shadowData["shadow.mismatched"] = sh.Stats.Mismatched; + shadowData["shadow.unavailable"] = sh.Stats.ShadowUnavailable; + shadowData["shadow.dropped"] = sh.Stats.Dropped; + shadowData["shadow.inFlight"] = sh.Stats.InFlight; + return (HealthStatus.Healthy, + $"Scripts run in process (Scripting:Runner=Shadow); the out-of-process shadow runner is {sh.Runner.State}: " + + $"{sh.Stats.Compared} compared, {sh.Stats.Mismatched} mismatch(es), {sh.Stats.Dropped} dropped. " + + "Shadow results are never published and never affect readiness.", + shadowData); + } + 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 data = RunnerData(status, string.Empty); var down = status.Shards.Where(s => s.DownFor is not null).Select(s => s.Label).ToArray(); return status.State switch { @@ -48,6 +58,24 @@ public static class ScriptRunnerHealthDecision data), }; } + + private static Dictionary RunnerData(ScriptRunnerStatus status, string prefix) + { + var data = new Dictionary(StringComparer.Ordinal) + { + [$"{prefix}state"] = status.State.ToString(), + [$"{prefix}suspects"] = status.SuspectCount, + [$"{prefix}quarantined"] = status.QuarantinedCount, + [$"{prefix}peCache"] = status.PeCacheEnabled, + }; + foreach (var shard in status.Shards) + { + data[$"{prefix}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"; + } + return data; + } } /// The script-runner readiness component: Healthy, or Degraded-but-ready (see ). @@ -60,7 +88,7 @@ public sealed class ScriptRunnerHealthCheck(ScriptRunnerSelection? selection) : /// public Task CheckHealthAsync(HealthCheckContext context, CancellationToken cancellationToken = default) { - var (status, description, data) = ScriptRunnerHealthDecision.Evaluate(selection?.Status()); + var (status, description, data) = ScriptRunnerHealthDecision.Evaluate(selection?.Status(), selection?.ShadowStatus()); return Task.FromResult(status == HealthStatus.Healthy ? HealthCheckResult.Healthy(description, data) : HealthCheckResult.Degraded(description, data: data)); diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ShadowScriptRunnerTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ShadowScriptRunnerTests.cs new file mode 100644 index 00000000..a0f29ae5 --- /dev/null +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ShadowScriptRunnerTests.cs @@ -0,0 +1,343 @@ +using System.Collections.Concurrent; +using System.Diagnostics; +using System.Diagnostics.Metrics; +using Serilog; +using Serilog.Core; +using Serilog.Events; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Engines; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests; + +/// +/// lmxopcua#780 P2 — , the comparison half of Scripting:Runner=Shadow, against +/// fake shadow runners: the comparison table, per-script mismatch accounting and its rate-limited log, and the +/// guarantee that nothing on the shadow side — a runner that hangs, blocks its caller, or throws — reaches the +/// evaluation that published, plus the drop accounting that keeps the shadow bounded. +/// +public sealed class ShadowScriptRunnerTests +{ + private static readonly IReadOnlyDictionary NoInputs = + new Dictionary(StringComparer.Ordinal); + + private sealed class FakeShadow(Func> run) : IScriptRunner + { + public ConcurrentQueue Invocations { get; } = new(); + public ConcurrentQueue Resets { get; } = new(); + + public Task RunAsync(ScriptInvocation invocation, CancellationToken ct = default) + { + Invocations.Enqueue(invocation); + return run(invocation, ct); + } + + public void Reset(ScriptContextKind kind) => Resets.Enqueue(kind); + + public IReadOnlyCollection QuarantinedKeys(ScriptContextKind kind) => []; + } + + private sealed class Capture : ILogEventSink + { + public ConcurrentQueue Events { get; } = new(); + + public void Emit(LogEvent logEvent) => Events.Enqueue(logEvent); + + public ILogger Logger() => new LoggerConfiguration().MinimumLevel.Verbose().WriteTo.Sink(this).CreateLogger(); + + public string[] Mismatches => Events.Where(e => e.MessageTemplate.Text.StartsWith("Script shadow mismatch", StringComparison.Ordinal)) + .Select(e => e.RenderMessage()).ToArray(); + } + + private static ScriptInvocation Vt(string key, IReadOnlyDictionary? inputs = null, string source = "return 1;") + => new(key, ScriptContextKind.VirtualTag, source, inputs ?? NoInputs, TimeSpan.FromSeconds(2)); + + private static ScriptingOptions Options(int maxInFlight = 256, int logsPerMinute = 20) + => new() { Shadow = new ScriptShadowOptions { MaxInFlight = maxInFlight, MismatchLogsPerMinute = logsPerMinute } }; + + private static string NewKey() => "shadow-" + Guid.NewGuid().ToString("N")[..10]; + + public static TheoryData Table => new() + { + { ScriptRunOutcome.Ok(42), ScriptRunOutcome.Ok(42), ScriptShadowVerdict.Match, null }, + { ScriptRunOutcome.Ok(null), ScriptRunOutcome.Ok(null), ScriptShadowVerdict.Match, null }, + { ScriptRunOutcome.Ok(new[] { 1, 2 }), ScriptRunOutcome.Ok(new[] { 1, 2 }), ScriptShadowVerdict.Match, null }, + { ScriptRunOutcome.Ok(double.NaN), ScriptRunOutcome.Ok(double.NaN), ScriptShadowVerdict.Match, null }, + { ScriptRunOutcome.Ok(42), ScriptRunOutcome.Ok(43), ScriptShadowVerdict.Mismatch, "value" }, + { ScriptRunOutcome.Ok(new[] { 1, 2 }), ScriptRunOutcome.Ok(new[] { 1, 3 }), ScriptShadowVerdict.Mismatch, "value" }, + { ScriptRunOutcome.Ok(42), ScriptRunOutcome.Ok(null), ScriptShadowVerdict.Mismatch, "value" }, + { ScriptRunOutcome.Ok((short)42), ScriptRunOutcome.Ok(42), ScriptShadowVerdict.Mismatch, "type" }, + { ScriptRunOutcome.Ok(42), ScriptRunOutcome.Failed("script threw: boom"), ScriptShadowVerdict.Mismatch, "status" }, + { ScriptRunOutcome.Failed("script threw: boom"), ScriptRunOutcome.Ok(42), ScriptShadowVerdict.Mismatch, "status" }, + { ScriptRunOutcome.Failed("script threw: boom"), ScriptRunOutcome.Failed("script threw: boom"), ScriptShadowVerdict.Match, null }, + { ScriptRunOutcome.Failed("script threw: boom"), ScriptRunOutcome.Failed("script threw: bang"), ScriptShadowVerdict.Mismatch, "exception" }, + { ScriptRunOutcome.Failed("compile error: CS1002"), ScriptRunOutcome.Failed("compile error: CS1002 (worker)"), ScriptShadowVerdict.Match, null }, + { ScriptRunOutcome.Failed("compile error: CS1002"), ScriptRunOutcome.Failed("script threw: x"), ScriptShadowVerdict.Mismatch, "failure_kind" }, + // The same runaway reaches "budget" by different routes in the two runners: a timeout here, a kill-strike quarantine there. + { ScriptRunOutcome.Failed("script timed out after 2.0s"), ScriptRunOutcome.Failed("script quarantined: killed twice"), ScriptShadowVerdict.Match, null }, + { ScriptRunOutcome.Failed("previous evaluation still running"), ScriptRunOutcome.Failed("script timed out after 2.0s"), ScriptShadowVerdict.Match, null }, + // Design §14 #11: what the protocol cannot carry is surfaced in shadow first. + { ScriptRunOutcome.Ok(1.5m), ScriptRunOutcome.Failed("unsupported type: script returned a value of unsupported type System.Decimal"), ScriptShadowVerdict.Mismatch, "unsupported_type" }, + { ScriptRunOutcome.Failed("script threw: x"), ScriptRunOutcome.Failed("unsupported type: input 'a' has a value of unsupported type System.Guid"), ScriptShadowVerdict.Mismatch, "unsupported_type" }, + // The runner's fault, not the script's: not compared. + { ScriptRunOutcome.Ok(42), ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerRestartedReason), ScriptShadowVerdict.ShadowUnavailable, null }, + { ScriptRunOutcome.Ok(42), ScriptRunOutcome.Failed(ScriptRunOutcome.WorkerBusyReason), ScriptShadowVerdict.ShadowUnavailable, null }, + }; + + [Theory] + [MemberData(nameof(Table))] + public void Compare_classifies_each_pair(ScriptRunOutcome published, ScriptRunOutcome shadow, ScriptShadowVerdict verdict, string? reason) + { + var comparison = ShadowScriptRunner.Compare(published, shadow); + comparison.Verdict.ShouldBe(verdict); + comparison.Reason.ShouldBe(reason); + } + + [Fact] + public async Task A_divergent_script_is_counted_per_script_in_metrics_and_logged_once_with_a_diff_summary() + { + var divergent = NewKey(); + var healthy = NewKey(); + var shadow = new FakeShadow((inv, _) => Task.FromResult(inv.Key == divergent ? ScriptRunOutcome.Ok(2) : ScriptRunOutcome.Ok(1))); + var log = new Capture(); + using var runner = new ShadowScriptRunner(shadow, Options(), log.Logger()); + + var measured = new ConcurrentQueue<(long Value, string? Script, string? Reason, string? Kind)>(); + using var listener = new MeterListener(); + listener.InstrumentPublished = (instrument, l) => + { + if (instrument.Name == "otopcua.script.shadow.mismatch") l.EnableMeasurementEvents(instrument); + }; + listener.SetMeasurementEventCallback((_, value, tags, _) => + { + string? script = null, reason = null, kind = null; + foreach (var tag in tags) + { + if (tag.Key == "script") script = tag.Value as string; + if (tag.Key == "reason") reason = tag.Value as string; + if (tag.Key == "kind") kind = tag.Value as string; + } + if (script == divergent || script == healthy) measured.Enqueue((value, script, reason, kind)); + }); + listener.Start(); + + for (var i = 0; i < 5; i++) + { + runner.Mirror(Vt(divergent), ScriptRunOutcome.Ok(1)); + runner.Mirror(Vt(healthy), ScriptRunOutcome.Ok(1)); + (await runner.WaitIdleAsync(TimeSpan.FromSeconds(10))).ShouldBeTrue(); + } + + var stats = runner.Stats(); + stats.Compared.ShouldBe(10); + stats.Matched.ShouldBe(5); + stats.Mismatched.ShouldBe(5); + stats.Dropped.ShouldBe(0); + var record = runner.Mismatches.ShouldHaveSingleItem(); + record.Key.ShouldBe(divergent); + record.Count.ShouldBe(5); + record.LastReason.ShouldBe("value"); + record.LastSummary.ShouldBe("published Ok Int32 1; shadow Ok Int32 2"); + + measured.Count.ShouldBe(5); + measured.ShouldAllBe(m => m.Script == divergent && m.Reason == "value" && m.Kind == "virtual_tag" && m.Value == 1); + + var line = log.Mismatches.ShouldHaveSingleItem("once per script and reason, not once per evaluation"); + line.ShouldContain(divergent); + line.ShouldContain("published Ok Int32 1; shadow Ok Int32 2"); + } + + [Fact] + public async Task Mismatch_lines_are_rate_limited_per_minute_and_the_next_line_reports_what_was_suppressed() + { + var now = new DateTime(2026, 11, 10, 12, 0, 0, DateTimeKind.Utc); + var shadow = new FakeShadow((_, _) => Task.FromResult(ScriptRunOutcome.Ok(2))); + var log = new Capture(); + using var runner = new ShadowScriptRunner(shadow, Options(logsPerMinute: 2), log.Logger(), clock: () => now); + + for (var i = 0; i < 5; i++) runner.Mirror(Vt(NewKey()), ScriptRunOutcome.Ok(1)); + (await runner.WaitIdleAsync(TimeSpan.FromSeconds(10))).ShouldBeTrue(); + log.Mismatches.Length.ShouldBe(2); + runner.Stats().Mismatched.ShouldBe(5, "every mismatch is counted even when its line is not written"); + + now = now.AddMinutes(1); + runner.Mirror(Vt(NewKey()), ScriptRunOutcome.Ok(1)); + (await runner.WaitIdleAsync(TimeSpan.FromSeconds(10))).ShouldBeTrue(); + log.Mismatches.Length.ShouldBe(3); + log.Mismatches[^1].ShouldContain("3 earlier mismatch line(s) suppressed"); + } + + [Fact] + public async Task A_reset_re_arms_the_once_per_script_line_and_resets_the_shadow_runner() + { + var key = NewKey(); + var shadow = new FakeShadow((_, _) => Task.FromResult(ScriptRunOutcome.Ok(2))); + var log = new Capture(); + using var runner = new ShadowScriptRunner(shadow, Options(), log.Logger()); + + runner.Mirror(Vt(key), ScriptRunOutcome.Ok(1)); + (await runner.WaitIdleAsync(TimeSpan.FromSeconds(10))).ShouldBeTrue(); + runner.Mirror(Vt(key), ScriptRunOutcome.Ok(1)); + (await runner.WaitIdleAsync(TimeSpan.FromSeconds(10))).ShouldBeTrue(); + log.Mismatches.Length.ShouldBe(1); + + runner.Reset(ScriptContextKind.Calculation); + runner.Mirror(Vt(key), ScriptRunOutcome.Ok(1)); + (await runner.WaitIdleAsync(TimeSpan.FromSeconds(10))).ShouldBeTrue(); + log.Mismatches.Length.ShouldBe(1, "another seam's deploy does not re-arm a virtual tag's line"); + + runner.Reset(ScriptContextKind.VirtualTag); + runner.Mirror(Vt(key), ScriptRunOutcome.Ok(1)); + (await runner.WaitIdleAsync(TimeSpan.FromSeconds(10))).ShouldBeTrue(); + log.Mismatches.Length.ShouldBe(2); + shadow.Resets.ShouldBe([ScriptContextKind.Calculation, ScriptContextKind.VirtualTag]); + } + + [Fact] + public async Task A_shadow_runner_that_hangs_blocks_or_throws_never_reaches_the_caller_and_is_never_a_mismatch() + { + var release = new ManualResetEventSlim(false); + var hang = NewKey(); + var block = NewKey(); + var boom = NewKey(); + var shadow = new FakeShadow((inv, _) => + { + if (inv.Key == hang) return new TaskCompletionSource().Task; // never completes + if (inv.Key == block) release.Wait(TimeSpan.FromSeconds(2)); // blocks the thread that called RunAsync (bounded, so a regression fails instead of hanging) + if (inv.Key == boom) throw new InvalidOperationException("runner bug"); + return Task.FromResult(ScriptRunOutcome.Ok(1)); + }); + using var runner = new ShadowScriptRunner(shadow, Options(), new LoggerConfiguration().CreateLogger()); + + var sw = Stopwatch.StartNew(); + runner.Mirror(Vt(hang), ScriptRunOutcome.Ok(1)); + runner.Mirror(Vt(block), ScriptRunOutcome.Ok(1)); + runner.Mirror(Vt(boom), ScriptRunOutcome.Ok(1)); + sw.Elapsed.ShouldBeLessThan(TimeSpan.FromMilliseconds(500), "Mirror never runs the shadow on the caller's thread"); + + // Each key's run is outstanding (or done): a second mirror of the hung/blocked keys is dropped, never queued. + await Task.Delay(100, TestContext.Current.CancellationToken); + runner.Mirror(Vt(hang), ScriptRunOutcome.Ok(1)); + runner.Mirror(Vt(block), ScriptRunOutcome.Ok(1)); + runner.Stats().DroppedKeyBusy.ShouldBe(2); + + release.Set(); + await Task.Delay(200, TestContext.Current.CancellationToken); + var stats = runner.Stats(); + stats.Mismatched.ShouldBe(0); + stats.ShadowUnavailable.ShouldBe(1, "a runner that throws is the runner's fault, not a divergence"); + stats.Matched.ShouldBe(1, "the blocked one completed once released"); + stats.InFlight.ShouldBe(1, "the hung one is still outstanding — bounded at one per key"); + } + + [Fact] + public async Task Evaluations_beyond_MaxInFlight_are_dropped_and_counted_and_the_shadow_resumes_once_it_catches_up() + { + var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var shadow = new FakeShadow((_, _) => gate.Task); + using var runner = new ShadowScriptRunner(shadow, Options(maxInFlight: 3), new LoggerConfiguration().CreateLogger()); + + var measured = new ConcurrentQueue(); + using var listener = new MeterListener(); + listener.InstrumentPublished = (instrument, l) => + { + if (instrument.Name == "otopcua.script.shadow.dropped") l.EnableMeasurementEvents(instrument); + }; + var keys = Enumerable.Range(0, 10).Select(_ => NewKey()).ToArray(); + listener.SetMeasurementEventCallback((_, _, tags, _) => + { + foreach (var tag in tags) if (tag.Key == "reason") measured.Enqueue(tag.Value as string); + }); + listener.Start(); + + foreach (var key in keys) runner.Mirror(Vt(key), ScriptRunOutcome.Ok(1)); + await Scripts.Eventually(() => shadow.Invocations.Count == 3, "three shadow runs started"); + var stats = runner.Stats(); + stats.InFlight.ShouldBe(3); + stats.DroppedBehind.ShouldBe(7); + shadow.Invocations.Count.ShouldBe(3, "a dropped evaluation never reaches the shadow runner"); + + gate.SetResult(ScriptRunOutcome.Ok(1)); + (await runner.WaitIdleAsync(TimeSpan.FromSeconds(10))).ShouldBeTrue(); + runner.Mirror(Vt(keys[9]), ScriptRunOutcome.Ok(1)); + (await runner.WaitIdleAsync(TimeSpan.FromSeconds(10))).ShouldBeTrue(); + stats = runner.Stats(); + stats.Matched.ShouldBe(4); + stats.DroppedBehind.ShouldBe(7); + measured.Count(r => r == "behind").ShouldBeGreaterThanOrEqualTo(7); + } + + [Fact] + public async Task Mirror_copies_the_inputs_before_it_returns_and_never_forwards_the_script_logger() + { + var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var seen = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var shadow = new FakeShadow((inv, _) => + { + seen.TrySetResult(inv); + return gate.Task; + }); + using var runner = new ShadowScriptRunner(shadow, Options(), new LoggerConfiguration().CreateLogger()); + var cache = new Dictionary(StringComparer.Ordinal) + { + ["a"] = new(1, 0u, DateTime.UtcNow, DateTime.UtcNow), + }; + var scriptLog = new LoggerConfiguration().CreateLogger(); + + runner.Mirror(Vt(NewKey(), cache) with { ScriptLogger = scriptLog }, ScriptRunOutcome.Ok(1)); + cache["a"] = new DataValueSnapshot(99, 0u, DateTime.UtcNow, DateTime.UtcNow); // the alarm engine refills in place + var invocation = await seen.Task.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); + gate.SetResult(ScriptRunOutcome.Ok(1)); + + invocation.Inputs["a"].Value.ShouldBe(1); + invocation.Inputs.ShouldNotBeSameAs(cache); + invocation.ScriptLogger.ShouldBeSameAs(Logger.None, "the in-process run already wrote the script's log lines"); + } + + [Fact] + public async Task As_a_runner_it_publishes_the_in_process_result_and_mirrors_it() + { + var shadow = new FakeShadow((_, _) => Task.FromResult(ScriptRunOutcome.Ok(-1))); + using var runner = new ShadowScriptRunner(shadow, Options(), new LoggerConfiguration().CreateLogger()); + var inputs = new Dictionary(StringComparer.Ordinal) + { + ["a"] = new(20, 0u, DateTime.UtcNow, DateTime.UtcNow), + }; + + var outcome = await runner.RunAsync(Vt(NewKey(), inputs, "return (int)ctx.GetTag(\"a\").Value + 1;"), TestContext.Current.CancellationToken); + outcome.Status.ShouldBe(ScriptRunStatus.Ok); + outcome.Value.ShouldBe(21, "the in-process result, not the shadow's -1"); + (await runner.WaitIdleAsync(TimeSpan.FromSeconds(10))).ShouldBeTrue(); + runner.Stats().Mismatched.ShouldBe(1); + } + + [Theory] + [InlineData("evaluator disposed")] + [InlineData(ScriptEvalFailureReasons.EvaluatorDisposedDuringEvaluation)] + [InlineData("evaluation cancelled (driver shutting down)")] + public void Host_lifecycle_failures_are_not_mirrored(string reason) + => VirtualTagEvalResult.Failure(reason).ToMirroredOutcome().ShouldBeNull(); + + [Fact] + public void Script_results_map_onto_runner_outcomes() + { + VirtualTagEvalResult.Ok(5).ToMirroredOutcome()!.ShouldBe(ScriptRunOutcome.Ok(5)); + VirtualTagEvalResult.Failure("script threw: x").ToMirroredOutcome()!.ShouldBe(ScriptRunOutcome.Failed("script threw: x")); + VirtualTagEvalResult.Unavailable("script worker restarted").ToMirroredOutcome()!.Status.ShouldBe(ScriptRunStatus.Unavailable); + } + + private static class Scripts + { + public static async Task Eventually(Func probe, string what) + { + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10); + while (DateTime.UtcNow < deadline) + { + if (probe()) return; + await Task.Delay(10); + } + throw new TimeoutException($"timed out waiting for {what}"); + } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/ScriptRunnerSelectionTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/ScriptRunnerSelectionTests.cs index 1f458b1c..02cd7753 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/ScriptRunnerSelectionTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/ScriptRunnerSelectionTests.cs @@ -128,12 +128,52 @@ public sealed class ScriptRunnerSelectionTests } [Fact] - public void Shadow_is_not_implemented_yet_and_runs_in_process_with_a_warning() + public void Shadow_hands_the_selected_seams_a_shadow_runner_and_never_reports_its_pool_to_readiness() { + // lmxopcua#780 P2. Unset Seams: VirtualTag + Calculation; alarms only when listed (same rule as OutOfProcess). var options = new ScriptingOptions { Runner = ScriptRunnerMode.Shadow }; - ScriptRunnerSelection.Validate(options, "/srv").ShouldHaveSingleItem().ShouldContain("Shadow"); - using var selection = ScriptRunnerSelection.Create(options, Silent, null); - selection.OutOfProcessRunner.ShouldBeNull(); + ScriptRunnerSelection.Validate(options, "/srv").ShouldContain(w => w.Contains("runs twice") && w.Contains("time-boxed diagnostic")); + using (var selection = ScriptRunnerSelection.Create(options, Silent, null)) + { + var vt = selection.ForSeam(ScriptSeam.VirtualTag).ShouldBeOfType(); + selection.ForSeam(ScriptSeam.Calculation).ShouldBeSameAs(vt, "one node-wide shadow"); + selection.ForSeam(ScriptSeam.ScriptedAlarm).ShouldBeNull(); + vt.Shadow.ShouldBeSameAs(selection.OutOfProcessRunner); + selection.Shadow.ShouldBeSameAs(vt); + selection.Status().ShouldBeNull("readiness must not see the shadow pool"); + selection.ShadowStatus().ShouldNotBeNull(); + } + + var withAlarms = new ScriptingOptions { Runner = ScriptRunnerMode.Shadow, RunnerSeams = [ScriptSeam.ScriptedAlarm] }; + using (var selection = ScriptRunnerSelection.Create(withAlarms, Silent, null)) + { + selection.ForSeam(ScriptSeam.ScriptedAlarm).ShouldBeOfType(); + selection.ForSeam(ScriptSeam.VirtualTag).ShouldBeNull(); + } + + // Out of process: no shadow, the pool's status IS readiness. + using (var oop = ScriptRunnerSelection.Create(new ScriptingOptions { Runner = ScriptRunnerMode.OutOfProcess }, Silent, null)) + { + oop.Shadow.ShouldBeNull(); + oop.ShadowStatus().ShouldBeNull(); + oop.Status().ShouldNotBeNull(); + } + } + + [Fact] + public void Scripting_Shadow_binds_and_clamps() + { + var config = new ConfigurationBuilder().AddInMemoryCollection(new Dictionary + { + ["Scripting:Runner"] = "Shadow", + ["Scripting:Shadow:MaxInFlight"] = "0", + ["Scripting:Shadow:MismatchLogsPerMinute"] = "5", + }).Build(); + var bound = ScriptingOptionsBinding.Bind(config, out var adjustments); + bound.Runner.ShouldBe(ScriptRunnerMode.Shadow); + bound.Shadow.MaxInFlight.ShouldBe(1); + bound.Shadow.MismatchLogsPerMinute.ShouldBe(5); + adjustments.ShouldContain(a => a.Contains("Scripting:Shadow:MaxInFlight")); } [Fact] diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Engines/RoslynVirtualTagEvaluatorScriptRunnerTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Engines/RoslynVirtualTagEvaluatorScriptRunnerTests.cs index 2c5e87ed..32a37068 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Engines/RoslynVirtualTagEvaluatorScriptRunnerTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Engines/RoslynVirtualTagEvaluatorScriptRunnerTests.cs @@ -106,4 +106,38 @@ public sealed class RoslynVirtualTagEvaluatorScriptRunnerTests (await sut.EvaluateAsync("vt-1", "return (int)ctx.GetTag(\"a\").Value + 1;", Deps(41))).Value.ShouldBe(42); sut.RunGuard.StartedRunCount.ShouldBe(1); } + + [Fact] + public async Task Under_Shadow_the_in_process_result_is_published_and_each_compiled_run_is_mirrored() + { + // lmxopcua#780 P2: the shadow answers differently (and slowly) — the published value must still be the in-process one. + var shadowRunner = new FakeRunner(); + shadowRunner.Outcomes.Enqueue(ScriptRunOutcome.Ok(-1)); + using var shadow = new ShadowScriptRunner(shadowRunner, log: new LoggerConfiguration().CreateLogger()); + using var sut = Build(shadow); + sut.ScriptRunner.ShouldBeNull("Shadow keeps the in-process path"); + sut.ScriptShadow.ShouldBeSameAs(shadow); + + var result = await sut.EvaluateAsync("vt-s", "return (int)ctx.GetTag(\"a\").Value + 1;", Deps(41)); + result.Success.ShouldBeTrue(); + result.Value.ShouldBe(42); + sut.RunGuard.StartedRunCount.ShouldBe(1, "the in-process guard ran it"); + + (await shadow.WaitIdleAsync(TimeSpan.FromSeconds(10))).ShouldBeTrue(); + var mirrored = shadowRunner.Invocations.ShouldHaveSingleItem(); + mirrored.Key.ShouldBe("vt-s"); + mirrored.Kind.ShouldBe(ScriptContextKind.VirtualTag); + mirrored.Timeout.ShouldBe(TimeSpan.FromMilliseconds(1500)); + mirrored.Inputs["a"].Value.ShouldBe(41); + shadow.Mismatches.ShouldHaveSingleItem().LastSummary.ShouldBe("published Ok Int32 42; shadow Ok Int32 -1"); + + // The passthrough fast path runs no user code: nothing to mirror. + (await sut.EvaluateAsync("vt-p", "return ctx.GetTag(\"a\").Value;", Deps(7))).Value.ShouldBe(7); + (await shadow.WaitIdleAsync(TimeSpan.FromSeconds(10))).ShouldBeTrue(); + shadowRunner.Invocations.Count.ShouldBe(1); + + // A redeploy resets the shadow side too. + sut.ClearCompiledScripts(); + shadowRunner.Resets.ShouldBe([ScriptContextKind.VirtualTag]); + } } 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 index 42f4a426..230d9a3f 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Health/ScriptRunnerHealthCheckTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Health/ScriptRunnerHealthCheckTests.cs @@ -76,4 +76,23 @@ public sealed class ScriptRunnerHealthCheckTests (await check.CheckHealthAsync(new HealthCheckContext { Registration = registration }, CancellationToken.None)) .Status.ShouldBe(HealthStatus.Healthy); } + + [Theory] + [InlineData(ScriptRunnerState.Ready)] + [InlineData(ScriptRunnerState.Degraded)] + [InlineData(ScriptRunnerState.Unavailable)] + public void Shadow_is_healthy_whatever_its_worker_pool_does(ScriptRunnerState shadowState) + { + // lmxopcua#780 P2: under Shadow every published value is in process — the shadow pool must never move readiness. + var stats = new ZB.MOM.WW.OtOpcUa.Core.Scripting.ScriptShadowStats(10, 7, 3, 1, 2, 1, 0); + var (status, description, data) = ScriptRunnerHealthDecision.Evaluate( + null, (Status(shadowState, Shard("0", shadowState == ScriptRunnerState.Ready, 200)), stats)); + status.ShouldBe(HealthStatus.Healthy); + description.ShouldContain("Shadow"); + description.ShouldContain("3 mismatch(es)"); + data["mode"].ShouldBe("Shadow"); + data["shadow.state"].ShouldBe(shadowState.ToString()); + data["shadow.mismatched"].ShouldBe(3L); + data["shadow.dropped"].ShouldBe(3L); + } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/ShadowModeTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/ShadowModeTests.cs new file mode 100644 index 00000000..3655fa57 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/ShadowModeTests.cs @@ -0,0 +1,226 @@ +using System.Collections.Concurrent; +using System.Diagnostics; +using Microsoft.Extensions.Logging.Abstractions; +using Serilog; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Engines; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms; +using ZB.MOM.WW.OtOpcUa.Core.Scripting; +using ZB.MOM.WW.OtOpcUa.Driver.Calculation; + +namespace ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests; + +/// +/// lmxopcua#780 P2 — Scripting:Runner=Shadow over the REAL worker: the seams publish their in-process results +/// and mirror each compiled evaluation to the worker pool. Ordinary scripts compare equal; a deliberately divergent +/// one (it reads which process it runs in) and an unsupported return type (design §14 #11) are counted as mismatches +/// for exactly those scripts; and a runaway that only runs away in the worker is killed and quarantined THERE while +/// every published value stays the in-process one. Alarms: the transition sequence is the in-process engine's, a +/// killed shadow worker never annotates or clears a condition. +/// +[Collection(ScriptWorkerCollection.Name)] +[Trait("Category", "ScriptWorker")] +public sealed class ShadowModeTests +{ + /// True only inside the ScriptWorker (its base directory is script-worker/): a test-only divergence. + private const string InWorker = "System.AppContext.BaseDirectory.Contains(\"script-worker\")"; + + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private static ScriptingOptions ShadowOptions(TimeSpan? hardGrace = null) + { + var options = Scripts.Options(hardGrace: hardGrace ?? TimeSpan.FromMilliseconds(500), quarantineAfterKills: 2); + options.Runner = ScriptRunnerMode.Shadow; + return options; + } + + private static CalculationEvaluator Calc(IScriptRunner? runner, TimeSpan? budget = null) + => new(NullLogger.Instance, new ScriptRootLogger(new LoggerConfiguration().CreateLogger()), + budget ?? TimeSpan.FromSeconds(5), scriptRunner: runner); + + private static Dictionary Inputs(int i) + { + var now = DateTime.UtcNow; + DataValueSnapshot S(object v) => new(v, 0u, now, now); + return new Dictionary(StringComparer.Ordinal) + { + ["s"] = S((short)(5 + i)), + ["d"] = S(1.25 * i), + ["str"] = S("abc" + i), + ["i"] = S(i), + ["t"] = S(new DateTime(2026, 11, 10, 12, 0, 0, DateTimeKind.Utc).AddSeconds(i)), + ["arr"] = S(new[] { 1.5, 2.5 * i }), + }; + } + + private static bool Same(VirtualTagEvalResult a, VirtualTagEvalResult b) + => a.Success == b.Success && a.Reason == b.Reason && a.StatusCode == b.StatusCode + && System.Collections.StructuralComparisons.StructuralEqualityComparer.Equals(a.Value, b.Value) + && a.Value?.GetType() == b.Value?.GetType(); + + [Fact] + public async Task Ordinary_scripts_match_and_a_divergent_and_an_unsupported_one_are_counted_against_exactly_those_scripts() + { + var log = new CapturingSink(); + await using var pool = Scripts.Runner(ShadowOptions()); + using var shadow = new ShadowScriptRunner(pool, ShadowOptions(), log.CreateLogger()); + using var shadowed = Calc(shadow); + using var reference = Calc(null); + shadowed.SubmitsBatches.ShouldBeFalse("Shadow keeps the in-process one-at-a-time pump, not the runner path"); + + var scripts = new Dictionary(StringComparer.Ordinal) + { + ["short"] = "return (short)ctx.GetTag(\"s\").Value + 1;", + ["double"] = "return (double)ctx.GetTag(\"d\").Value * 2.5;", + ["string"] = "return \"v=\" + ctx.GetTag(\"str\").Value;", + ["intArray"] = "return new[] { (int)ctx.GetTag(\"i\").Value, 2 };", + ["doubleArray"] = "var a = (double[])ctx.GetTag(\"arr\").Value; return new[] { a[1], a[0] };", + ["datetime"] = "return ((DateTime)ctx.GetTag(\"t\").Value).AddSeconds(1);", + ["bool"] = "return (int)ctx.GetTag(\"i\").Value > 1;", + ["quality"] = "return ctx.GetTag(\"i\").StatusCode;", + ["throws"] = "if ((int)ctx.GetTag(\"i\").Value >= 0) throw new InvalidOperationException(\"boom \" + ctx.GetTag(\"i\").Value); return 0;", + ["divergent"] = $"return {InWorker} ? 2 : 1;", + ["decimal"] = "return 1.5m;", + }; + + for (var i = 0; i < 3; i++) + { + foreach (var (key, source) in scripts) + { + var published = await shadowed.EvaluateAsync(key, source, Inputs(i), Ct); + var expected = await reference.EvaluateAsync(key, source, Inputs(i), Ct); + Same(published, expected).ShouldBeTrue($"{key}: Shadow publishes exactly what InProcess publishes ({published} vs {expected})"); + } + (await shadow.WaitIdleAsync(TimeSpan.FromSeconds(60))).ShouldBeTrue(); + } + + var stats = shadow.Stats(); + stats.Dropped.ShouldBe(0); + stats.ShadowUnavailable.ShouldBe(0); + stats.Compared.ShouldBe(scripts.Count * 3); + stats.Mismatched.ShouldBe(6, string.Join(Environment.NewLine, shadow.Mismatches.Select(m => $"{m.Key}: {m.LastReason} — {m.LastSummary}"))); + var mismatches = shadow.Mismatches.ToDictionary(m => m.Key); + mismatches.Keys.ShouldBe(["decimal", "divergent"], ignoreOrder: true); + mismatches["divergent"].LastReason.ShouldBe("value"); + mismatches["divergent"].LastSummary.ShouldBe("published Ok Int32 1; shadow Ok Int32 2"); + mismatches["divergent"].Count.ShouldBe(3); + mismatches["decimal"].LastReason.ShouldBe("unsupported_type"); + mismatches["decimal"].LastSummary.ShouldContain("System.Decimal"); + + var lines = log.Events.Where(e => e.MessageTemplate.Text.StartsWith("Script shadow mismatch", StringComparison.Ordinal)).ToArray(); + lines.Length.ShouldBe(2, "one line per script and reason"); + lines.ShouldAllBe(e => e.Level == Serilog.Events.LogEventLevel.Warning); + } + + [Fact] + public async Task A_runaway_that_runs_away_only_in_the_worker_is_killed_and_quarantined_there_and_never_reaches_a_published_value() + { + var log = new CapturingSink(); + await using var pool = Scripts.Runner(ShadowOptions(TimeSpan.FromMilliseconds(400)), log.CreateLogger()); + using var shadow = new ShadowScriptRunner(pool, ShadowOptions(), log.CreateLogger()); + using var shadowed = Calc(shadow, TimeSpan.FromMilliseconds(300)); + const string runaway = $"if ({InWorker}) {{ while (true) {{ }} }} return (int)ctx.GetTag(\"i\").Value * 2;"; + const string sibling = "return (int)ctx.GetTag(\"i\").Value + 100;"; + + var slowest = TimeSpan.Zero; + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(90); + var i = 0; + for (; !pool.QuarantinedKeys(ScriptContextKind.Calculation).Contains("runaway"); i++) + { + if (DateTime.UtcNow > deadline) + throw new TimeoutException("the shadow never quarantined the runaway:" + Environment.NewLine + string.Join(Environment.NewLine, log.Rendered)); + var sw = Stopwatch.StartNew(); + var a = await shadowed.EvaluateAsync("runaway", runaway, Inputs(i), Ct); + var b = await shadowed.EvaluateAsync("sibling", sibling, Inputs(i), Ct); + if (sw.Elapsed > slowest) slowest = sw.Elapsed; + a.ShouldBe(VirtualTagEvalResult.Ok(i * 2), "published from the in-process run"); + b.ShouldBe(VirtualTagEvalResult.Ok(i + 100)); + await Task.Delay(100, Ct); + } + + // Still publishing correctly after the shadow quarantined it, and never quarantined on the published side. + for (var j = i; j < i + 5; j++) + (await shadowed.EvaluateAsync("runaway", runaway, Inputs(j), Ct)).ShouldBe(VirtualTagEvalResult.Ok(j * 2)); + shadowed.QuarantinedScripts.ShouldBeEmpty(); + slowest.ShouldBeLessThan(TimeSpan.FromMilliseconds(250), "a published evaluation never waits for its shadow"); + log.Rendered.ShouldContain(l => l.Contains("exceeded its hard deadline")); + + (await shadow.WaitIdleAsync(TimeSpan.FromSeconds(30))).ShouldBeTrue(); + var record = shadow.Mismatches.ShouldHaveSingleItem("the sibling on the killed shard is a victim (not compared), never a mismatch"); + record.Key.ShouldBe("runaway"); + record.LastReason.ShouldBe("status"); + record.LastSummary.ShouldContain("shadow Failed [budget]"); + shadow.Stats().DroppedKeyBusy.ShouldBeGreaterThan(0, "triggers while its shadow run was still going were dropped, not queued"); + } + + private static ScriptedAlarmDefinition Alarm(string id, string predicate) => + new(AlarmId: id, EquipmentPath: "Plant/Line1/Reactor", AlarmName: id, Kind: AlarmKind.AlarmCondition, + Severity: AlarmSeverity.High, MessageTemplate: "condition", PredicateScriptSource: predicate); + + private static ScriptedAlarmEngine Engine(TestUpstream up, IScriptRunner? runner) + { + var logger = new LoggerConfiguration().CreateLogger(); + return new ScriptedAlarmEngine(up, new InMemoryAlarmStateStore(), new ScriptLoggerFactory(logger), logger, + scriptTimeout: TimeSpan.FromSeconds(2), scriptRunner: runner); + } + + [Fact] + public async Task Alarms_in_shadow_transition_exactly_as_in_process_and_a_killed_shadow_worker_never_touches_a_condition() + { + await using var pool = Scripts.Runner(ShadowOptions()); + using var shadow = new ShadowScriptRunner(pool, ShadowOptions(), new LoggerConfiguration().CreateLogger()); + var up = new TestUpstream(); + var refUp = new TestUpstream(); + up.Set("x", 0); + refUp.Set("x", 0); + using var engine = Engine(up, shadow); + using var reference = Engine(refUp, null); + engine.ScriptRunner.ShouldBeNull("Shadow keeps the in-process engine path"); + var events = new ConcurrentQueue(); + var refEvents = new ConcurrentQueue(); + engine.OnEvent += (_, e) => events.Enqueue(e); + reference.OnEvent += (_, e) => refEvents.Enqueue(e); + ScriptedAlarmDefinition[] defs = + [ + Alarm("high", "return (int)ctx.GetTag(\"x\").Value > 5;"), + Alarm("divergent", $"return {InWorker};"), + ]; + await engine.LoadAsync(defs, Ct); + await reference.LoadAsync(defs, Ct); + + async Task Push(int x) + { + up.Push("x", x); + refUp.Push("x", x); + await Scripts.Eventually(() => events.Count == refEvents.Count, TimeSpan.FromSeconds(10), "both engines' emissions"); + await Task.Delay(100, Ct); + } + + await Push(10); + await Push(2); + (await shadow.WaitIdleAsync(TimeSpan.FromSeconds(30))).ShouldBeTrue(); + + // kill -9 the shadow's worker: the published engine neither annotates nor clears anything because of it. + var worker = pool.Shards[0].Current.ShouldNotBeNull(); + Process.GetProcessById(worker.Pid).Kill(); + await Push(12); + await Push(0); + (await shadow.WaitIdleAsync(TimeSpan.FromSeconds(30))).ShouldBeTrue(); + await Push(9); + (await shadow.WaitIdleAsync(TimeSpan.FromSeconds(30))).ShouldBeTrue(); + + events.Select(e => (e.AlarmId, e.Emission, e.WorstInputStatusCode)) + .ShouldBe(refEvents.Select(e => (e.AlarmId, e.Emission, e.WorstInputStatusCode)), "the in-process engine's sequence, exactly"); + events.Count(e => e is { AlarmId: "high", Emission: EmissionKind.Activated }).ShouldBe(3); + events.ShouldNotContain(e => e.WorstInputStatusCode != 0u, "no shadow-side failure annotates a published condition"); + engine.GetState("divergent")!.Active.ShouldBe(AlarmActiveState.Inactive, "the in-process predicate (false) drives it"); + + var record = shadow.Mismatches.ShouldHaveSingleItem(); + record.Key.ShouldBe("divergent"); + record.Kind.ShouldBe(ScriptContextKind.AlarmPredicate); + record.LastSummary.ShouldBe("published Ok Boolean False; shadow Ok Boolean True"); + shadow.Stats().Matched.ShouldBeGreaterThan(0); + } +}