Joseph Doherty
db057548e4
A wrong published value now reports its Success/Value/Reason/StatusCode, the
publish's wall time against the run budget, the run guard's in-flight /
refused / quarantined counters, thread-pool and GC state around the call,
the shadow stats, and the tails of the script and runner logs.
Reproduced under heavy load (32 `yes`, test at nice 20, load avg 300+):
3 of 8 runs failed, every one because the SIBLING's shadow run missed the
300 ms soft deadline in the starved worker ("shadow Failed [budget] script
timed out after 0.3s") and was rightly recorded as a sibling mismatch. No
published value was ever wrong. The out-of-process soft deadline is armed at
submit, so it covers the pipe round trip and worker scheduling too.
The run budget (which is also the worker's soft deadline) goes to the 2 s
production default; no assertion changes. The runaway still runs away and is
killed at T + HardGrace, so nothing the test proves depends on T being short.
410 lines
24 KiB
C#
410 lines
24 KiB
C#
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.Core.Scripting.OutOfProcess;
|
|
using ZB.MOM.WW.OtOpcUa.Driver.Calculation;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests;
|
|
|
|
/// <summary>
|
|
/// lmxopcua#780 P2 — <c>Scripting:Runner=Shadow</c> 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.
|
|
/// </summary>
|
|
[Collection(ScriptWorkerCollection.Name)]
|
|
[Trait("Category", "ScriptWorker")]
|
|
public sealed class ShadowModeTests
|
|
{
|
|
/// <summary>
|
|
/// True only inside the ScriptWorker: a test-only divergence that needs no sandbox-forbidden API. The worker runs
|
|
/// with <c>InvariantGlobalization</c> (ordinal string comparison) while the test host does linguistic (ICU/NLS)
|
|
/// comparison, so a culture-sensitive <c>string.Compare("a", "B")</c> is negative in the host and positive in the
|
|
/// worker. <see cref="Culture_sensitive_comparison_really_diverges_between_host_and_worker"/> pins the premise.
|
|
/// </summary>
|
|
private const string InWorker = "(string.Compare(\"a\", \"B\") > 0)";
|
|
|
|
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, CapturingSink? scriptLog = null)
|
|
=> new(NullLogger.Instance,
|
|
new ScriptRootLogger(scriptLog is null
|
|
? new LoggerConfiguration().CreateLogger()
|
|
: new LoggerConfiguration().MinimumLevel.Verbose().WriteTo.Sink(scriptLog).CreateLogger()),
|
|
budget ?? TimeSpan.FromSeconds(5), scriptRunner: runner);
|
|
|
|
private static Dictionary<string, DataValueSnapshot> Inputs(int i)
|
|
{
|
|
var now = DateTime.UtcNow;
|
|
DataValueSnapshot S(object v) => new(v, 0u, now, now);
|
|
return new Dictionary<string, DataValueSnapshot>(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 Culture_sensitive_comparison_really_diverges_between_host_and_worker()
|
|
{
|
|
// Premise of InWorker: the worker is a different process whose globalization mode differs from the host's.
|
|
string.Compare("a", "B").ShouldBeLessThan(0, "the test host compares linguistically");
|
|
await using var pool = Scripts.Runner(Scripts.Options());
|
|
var r = await pool.RunAsync(Scripts.Vt("probe", $"return {InWorker} ? 2 : 1;"), Ct);
|
|
r.Status.ShouldBe(ScriptRunStatus.Ok, r.Reason);
|
|
r.Value.ShouldBe(2, "the worker (InvariantGlobalization) compares ordinally");
|
|
}
|
|
|
|
[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<string, string>(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);
|
|
}
|
|
|
|
/// <summary>
|
|
/// The non-blocking contract of <see cref="IScriptShadow.Mirror"/>, proven causally rather than by a stopwatch
|
|
/// (lmxopcua#798: a 250 ms bound around the first evaluations measured cold Roslyn compile/JIT of the in-process
|
|
/// run, not shadow scheduling). <see cref="GatedShadow"/> sits between the REAL <see cref="ShadowScriptRunner"/>
|
|
/// and the REAL worker pool and holds the runaway's first shadow run at two test-owned barriers: before it reaches
|
|
/// the pool, and — after the worker has run it away, answered the soft timeout and been killed at its hard
|
|
/// deadline — before its outcome is handed back for comparison. Every published evaluation made while a barrier
|
|
/// is closed completed while its shadow was provably outstanding (only the test opens a barrier), so a published
|
|
/// path that waited for its shadow would hang here rather than merely run slow. Mutation check: make
|
|
/// <see cref="ShadowScriptRunner.Mirror"/> wait for the shadow run and the first held evaluation never returns.
|
|
/// A representative latency number is the opt-in <see cref="ScriptRunnerPerfTests"/> case (Category=Perf).
|
|
/// </summary>
|
|
[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();
|
|
var scriptLog = new CapturingSink();
|
|
await using var pool = Scripts.Runner(ShadowOptions(TimeSpan.FromMilliseconds(400)), log.CreateLogger());
|
|
var gate = new GatedShadow(pool);
|
|
using var shadow = new ShadowScriptRunner(gate, ShadowOptions(), log.CreateLogger());
|
|
using var shadowed = Calc(shadow, RunawayRunBudget, scriptLog);
|
|
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 i = 0;
|
|
var testClock = Stopwatch.StartNew();
|
|
|
|
// Off the test thread with a liveness bound (not a latency bound): a Mirror that waited for its held shadow would
|
|
// otherwise hang the test instead of failing it.
|
|
Task<VirtualTagEvalResult> Publish(string key, string source)
|
|
=> Task.Run(() => shadowed.EvaluateAsync(key, source, Inputs(i), Ct), Ct).WaitAsync(TimeSpan.FromSeconds(30), Ct);
|
|
|
|
// lmxopcua#811: a wrong published value must say what it was and what the host was doing — the one loaded
|
|
// failure seen so far was unattributable because only "should be Ok(…)" survived a truncated log.
|
|
async Task Expect(string phase, string key, string source, int expected)
|
|
{
|
|
var before = PublishProbe.Take(shadowed, shadow);
|
|
var sw = Stopwatch.StartNew();
|
|
var published = await Publish(key, source);
|
|
sw.Stop();
|
|
if (published == VirtualTagEvalResult.Ok(expected)) return;
|
|
var after = PublishProbe.Take(shadowed, shadow);
|
|
published.ShouldBe(VirtualTagEvalResult.Ok(expected),
|
|
$"{phase}: {key} (iteration {i}) published the wrong value — " +
|
|
$"Success={published.Success}, Value={published.Value ?? "null"} ({published.Value?.GetType().Name ?? "-"}), " +
|
|
$"Reason={published.Reason ?? "null"}, StatusCode={(published.StatusCode is { } sc ? $"0x{sc:X8}" : "null")}, " +
|
|
$"RetryWhen={(published.RetryWhen is null ? "null" : "set")}; publish took {sw.Elapsed.TotalMilliseconds:F0} ms " +
|
|
$"against a {RunawayRunBudget.TotalMilliseconds:F0} ms run budget, at t+{testClock.Elapsed.TotalSeconds:F1} s; " +
|
|
$"quarantined={(shadowed.RunGuard.IsQuarantined(key, out var why) ? why : "no")}" + Environment.NewLine +
|
|
$" before: {before}" + Environment.NewLine +
|
|
$" after: {after}" + Environment.NewLine +
|
|
" script log (last 20):" + Lines(scriptLog) + Environment.NewLine +
|
|
" runner log (last 20):" + Lines(log));
|
|
}
|
|
|
|
static string Lines(CapturingSink sink)
|
|
=> sink.Events.IsEmpty
|
|
? " (none)"
|
|
: Environment.NewLine + " " + string.Join(Environment.NewLine + " ",
|
|
sink.Events.TakeLast(20).Select(e => $"{e.Timestamp:HH:mm:ss.fff} {e.Level} {e.RenderMessage()}"));
|
|
|
|
async Task PublishBoth(string phase)
|
|
{
|
|
await Expect($"{phase}: published from the in-process run", "runaway", runaway, i * 2);
|
|
await Expect(phase, "sibling", sibling, i + 100);
|
|
i++;
|
|
}
|
|
|
|
// Barrier 1: the runaway's first shadow run is dispatched and held before the pool ever sees it.
|
|
var held = gate.Hold("runaway");
|
|
await PublishBoth("first evaluation");
|
|
await held.Entered.WaitAsync(TimeSpan.FromSeconds(30), Ct);
|
|
var droppedBefore = shadow.Stats().DroppedKeyBusy;
|
|
for (var k = 0; k < 3; k++)
|
|
{
|
|
await PublishBoth("shadow held before the worker");
|
|
held.ReachedPool.ShouldBeFalse("barrier 1 is still closed: this evaluation published while its shadow was blocked");
|
|
}
|
|
shadow.Stats().DroppedKeyBusy.ShouldBeGreaterThanOrEqualTo(droppedBefore + 3,
|
|
"each runaway trigger behind the held shadow run was dropped — not queued, not waited for");
|
|
|
|
// Open barrier 1: the shadow run really runs away in the worker, the pool answers the soft timeout T and the
|
|
// worker is killed at T + G. Barrier 2 still holds the outcome back from the comparison.
|
|
var worker = await Scripts.Eventually(() => pool.Shards[0].Current, TimeSpan.FromSeconds(30), "the shadow worker");
|
|
held.ReleaseToPool();
|
|
var answered = await held.PoolAnswered.WaitAsync(TimeSpan.FromSeconds(30), Ct);
|
|
answered.Status.ShouldBe(ScriptRunStatus.Failed, answered.Reason);
|
|
answered.Reason!.ShouldStartWith("script timed out after ", Case.Sensitive, "the worker ran it away and the pool answered at T");
|
|
await PublishBoth("shadow running away / being killed in the worker");
|
|
var exit = await worker.Exited.WaitAsync(TimeSpan.FromSeconds(30), Ct);
|
|
exit.Reason.ShouldBe(WorkerExitReason.KilledDeadline);
|
|
exit.CulpritKey.ShouldBe(InProcessScriptRunner.ScopedKey(ScriptContextKind.Calculation, "runaway"),
|
|
"only the runaway is blamed, never the sibling sharing its shard");
|
|
await PublishBoth("worker killed, shadow outcome still held");
|
|
shadow.Mismatches.ShouldBeEmpty("nothing about the held shadow run has been compared, and nothing published waited for it");
|
|
shadow.Stats().InFlight.ShouldBeGreaterThan(0, "the held shadow run is still outstanding");
|
|
|
|
// Open barrier 2: the comparison records the divergence; the published side never saw any of it.
|
|
held.Deliver();
|
|
(await shadow.WaitIdleAsync(TimeSpan.FromSeconds(30))).ShouldBeTrue();
|
|
shadow.Mismatches.ShouldHaveSingleItem().Key.ShouldBe("runaway");
|
|
|
|
// Un-gated from here: keep triggering until the shadow's second kill quarantines it in the pool.
|
|
var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(90);
|
|
while (!pool.QuarantinedKeys(ScriptContextKind.Calculation).Contains("runaway"))
|
|
{
|
|
if (DateTime.UtcNow > deadline)
|
|
throw new TimeoutException("the shadow never quarantined the runaway:" + Environment.NewLine + string.Join(Environment.NewLine, log.Rendered));
|
|
await PublishBoth("until quarantined");
|
|
await Task.Delay(100, Ct);
|
|
}
|
|
|
|
// Still publishing correctly after the shadow quarantined it, and never quarantined on the published side.
|
|
for (var j = 0; j < 5; j++) await PublishBoth("after the shadow quarantine");
|
|
shadowed.QuarantinedScripts.ShouldBeEmpty();
|
|
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");
|
|
}
|
|
|
|
/// <summary>
|
|
/// The in-process run budget of the runaway test's published path. It is also the soft deadline the shadow worker
|
|
/// answers the runaway at (the mirrored invocation carries the seam's budget), so it sets how long the shadow runs
|
|
/// away before it is killed.
|
|
/// <para>lmxopcua#811: it was 300 ms, and under heavy CPU load (32 <c>yes</c> processes, the test at nice 20, load
|
|
/// average 300+) 3 of 8 runs failed because the SIBLING's shadow — a one-line script — missed that soft deadline
|
|
/// in the worker ("shadow Failed [budget] script timed out after 0.3s"), which the comparison rightly records as a
|
|
/// sibling mismatch. The out-of-process soft deadline is armed at submit, so it also covers the pipe round trip
|
|
/// and the starved worker being scheduled at all. The published values were never wrong. 2 s is the production
|
|
/// default budget; the runaway still runs away (and is killed at T + HardGrace) whatever T is, so nothing the test
|
|
/// proves depends on T being short — only its runtime does (two kills, ~5 s).</para>
|
|
/// </summary>
|
|
private static readonly TimeSpan RunawayRunBudget = TimeSpan.FromSeconds(2);
|
|
|
|
/// <summary>Host-side state around one published evaluation, for a self-diagnosing failure message (lmxopcua#811).</summary>
|
|
private readonly record struct PublishProbe(
|
|
int GuardInFlight, long GuardStarted, long GuardRefused, long GuardQuarantined, int PoolThreads, long PoolPending,
|
|
int Gen0, int Gen1, int Gen2, double GcPauseMs, ScriptShadowStats Shadow)
|
|
{
|
|
public static PublishProbe Take(CalculationEvaluator evaluator, ShadowScriptRunner shadow) => new(
|
|
evaluator.RunGuard.InFlightCount, evaluator.RunGuard.StartedRunCount, evaluator.RunGuard.RefusedWhileRunningCount,
|
|
evaluator.RunGuard.QuarantinedCount, ThreadPool.ThreadCount, ThreadPool.PendingWorkItemCount,
|
|
GC.CollectionCount(0), GC.CollectionCount(1), GC.CollectionCount(2), GC.GetTotalPauseDuration().TotalMilliseconds,
|
|
shadow.Stats());
|
|
|
|
public override string ToString() =>
|
|
$"guard inFlight={GuardInFlight} started={GuardStarted} refused={GuardRefused} quarantined={GuardQuarantined}; " +
|
|
$"pool threads={PoolThreads} pending={PoolPending}; gc gen0/1/2={Gen0}/{Gen1}/{Gen2} totalPause={GcPauseMs:F1} ms; " +
|
|
$"shadow {Shadow}";
|
|
}
|
|
|
|
/// <summary>
|
|
/// A pass-through shadow runner with test-owned barriers on one key's next run: it holds that run before it reaches
|
|
/// the inner runner, and holds the inner runner's outcome before returning it. Everything else forwards untouched.
|
|
/// </summary>
|
|
private sealed class GatedShadow(IScriptRunner inner) : IScriptRunner
|
|
{
|
|
private readonly ConcurrentDictionary<string, Held> _holds = new(StringComparer.Ordinal);
|
|
|
|
public Held Hold(string key) => _holds[key] = new Held();
|
|
|
|
public async Task<ScriptRunOutcome> RunAsync(ScriptInvocation invocation, CancellationToken ct = default)
|
|
{
|
|
if (!_holds.TryRemove(invocation.Key, out var held)) return await inner.RunAsync(invocation, ct);
|
|
held.EnteredSource.TrySetResult();
|
|
await held.ToPool.Task.WaitAsync(ct);
|
|
held.ReachedPool = true;
|
|
var outcome = await inner.RunAsync(invocation, ct);
|
|
held.Answered.TrySetResult(outcome);
|
|
await held.Delivery.Task.WaitAsync(ct);
|
|
return outcome;
|
|
}
|
|
|
|
public void Reset(ScriptContextKind kind) => inner.Reset(kind);
|
|
|
|
public IReadOnlyCollection<string> QuarantinedKeys(ScriptContextKind kind) => inner.QuarantinedKeys(kind);
|
|
|
|
public sealed class Held
|
|
{
|
|
internal readonly TaskCompletionSource EnteredSource = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
internal readonly TaskCompletionSource ToPool = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
internal readonly TaskCompletionSource<ScriptRunOutcome> Answered = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
internal readonly TaskCompletionSource Delivery = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
private volatile bool _reachedPool;
|
|
|
|
/// <summary>The shadow runner dispatched the held run: it is outstanding from here on.</summary>
|
|
public Task Entered => EnteredSource.Task;
|
|
|
|
/// <summary>Whether the held run has been let through to the inner runner.</summary>
|
|
public bool ReachedPool { get => _reachedPool; internal set => _reachedPool = value; }
|
|
|
|
/// <summary>The inner runner's outcome for the held run, before it is returned to the shadow runner.</summary>
|
|
public Task<ScriptRunOutcome> PoolAnswered => Answered.Task;
|
|
|
|
public void ReleaseToPool() => ToPool.TrySetResult();
|
|
|
|
public void Deliver() => Delivery.TrySetResult();
|
|
}
|
|
}
|
|
|
|
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<ScriptedAlarmEvent>();
|
|
var refEvents = new ConcurrentQueue<ScriptedAlarmEvent>();
|
|
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);
|
|
}
|
|
}
|