using Serilog;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
namespace ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests;
///
/// runs every predicate through a keyed by
/// alarm id. A runaway predicate used to leak one pool thread per upstream push (each re-evaluation
/// spawned another abandoned run); now the orphan is one, later pushes are refused (and a
/// predicate that times out repeatedly is quarantined) — frozen at its prior state and annotated Bad quality so the operator can see it — while a
/// sibling alarm on the same engine keeps evaluating. A reload lifts the quarantine.
///
[Trait("Category", "Unit")]
public sealed class ScriptedAlarmEngineRunGuardTests
{
private const uint StatusBadInternalError = 0x80020000u;
private const string SpinPredicate =
"var end = System.DateTime.UtcNow.AddSeconds(6); while (System.DateTime.UtcNow < end) { } return false;";
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 Build(FakeUpstream up, ScriptingOptions? scripting = null)
{
var logger = new LoggerConfiguration().CreateLogger();
// 500 ms, not the 60 ms the guard's own tests use: the HEALTHY alarm's first run is JIT-cold and
// must finish inside the budget on a loaded CI box, or the positive control itself times out. The
// spin predicate outlives this by 12x, so the runaway alarm still cannot finish first.
return new ScriptedAlarmEngine(up, new InMemoryAlarmStateStore(), new ScriptLoggerFactory(logger), logger,
scriptTimeout: TimeSpan.FromMilliseconds(500), scripting: scripting);
}
[Fact]
public async Task Runaway_predicate_leaks_one_orphan_is_refused_annotated_Bad_once_and_a_sibling_keeps_evaluating()
{
var up = new FakeUpstream();
up.Set("Temp", 50);
using var eng = Build(up);
var events = new List();
eng.OnEvent += (_, e) => { lock (events) events.Add(e); };
// The load itself evaluates once (startup recovery) — that is the first, abandoned run.
await eng.LoadAsync(
[Alarm("runaway", $"var t = ctx.GetTag(\"Temp\"); {SpinPredicate}"),
Alarm("healthy", """return (int)ctx.GetTag("Temp").Value > 100;""")],
TestContext.Current.CancellationToken);
// N more pushes while the orphan runs: every one is refused (a refusal is not a strike — I1), and
// the Bad annotation is emitted ONCE for the whole streak, not once per push (I3).
const int pushes = 4;
for (var i = 0; i < pushes; i++)
{
up.Push("Temp", 60 + i);
var expected = i + 1;
await WaitForAsync(() => eng.RunGuard.RefusedWhileRunningCount >= expected);
}
await WaitForAsync(() => eng.RunGuard.StartedRunCount >= 1 + 1 + pushes);
eng.RunGuard.IsQuarantined("runaway", out _).ShouldBeFalse("refusals are not strikes");
eng.RunGuard.StartedRunCount.ShouldBe(1 + 1 + pushes, "one dispatch for the runaway alarm, one per evaluation for the healthy one");
eng.RunGuard.InFlightCount.ShouldBe(1);
// The frozen condition is annotated Bad out of band — the operator sees WHY it is not moving —
// exactly once across the whole run of non-completions.
ScriptedAlarmEvent[] snapshot;
lock (events) snapshot = events.ToArray();
var quality = snapshot.Where(e => e.AlarmId == "runaway" && e.Emission == EmissionKind.QualityChanged).ToArray();
quality.Length.ShouldBe(1);
quality[0].WorstInputStatusCode.ShouldBe(StatusBadInternalError);
eng.GetState("runaway")!.Active.ShouldBe(AlarmActiveState.Inactive, "state is held, never invented");
// Positive control: the healthy sibling still transitions on the same engine.
up.Push("Temp", 150);
await WaitForAsync(() => eng.GetState("healthy")!.Active == AlarmActiveState.Active);
}
/// I3 — a QUARANTINED predicate is re-attempted on every push and refused before dispatch; the
/// Bad annotation must still be emitted once, not once per push.
[Fact]
public async Task A_quarantined_predicate_emits_QualityChanged_once_across_many_pushes()
{
var up = new FakeUpstream();
up.Set("Temp", 50);
using var eng = Build(up, new ScriptingOptions { QuarantineAfterConsecutiveTimeouts = 1 });
var events = new List();
eng.OnEvent += (_, e) => { lock (events) events.Add(e); };
await eng.LoadAsync(
[Alarm("q", $"var t = ctx.GetTag(\"Temp\"); {SpinPredicate}"),
Alarm("healthy", """return (int)ctx.GetTag("Temp").Value > 1000;""")],
TestContext.Current.CancellationToken);
eng.RunGuard.IsQuarantined("q", out _).ShouldBeTrue();
// Each push re-evaluates BOTH alarms (same input). The healthy sibling's dispatch count is the
// positive control proving every push was actually processed — so the quarantined alarm was
// re-attempted N times, not coalesced into one.
const int pushes = 5;
for (var i = 0; i < pushes; i++)
{
var before = eng.RunGuard.StartedRunCount;
up.Push("Temp", 60 + i);
await WaitForAsync(() => eng.RunGuard.StartedRunCount > before);
}
eng.RunGuard.StartedRunCount.ShouldBe(1 + 1 + pushes, "one runaway dispatch, then healthy-only dispatches");
await Task.Delay(300); // absence window for trailing emissions
ScriptedAlarmEvent[] snapshot;
lock (events) snapshot = events.ToArray();
var quality = snapshot.Where(e => e.AlarmId == "q" && e.Emission == EmissionKind.QualityChanged).ToArray();
quality.Length.ShouldBe(1, "one Bad annotation for the quarantine, not one per push");
quality[0].WorstInputStatusCode.ShouldBe(StatusBadInternalError);
snapshot.ShouldNotContain(e => e.AlarmId == "healthy" && e.Emission == EmissionKind.QualityChanged);
}
[Fact]
public async Task A_reload_lifts_the_quarantine()
{
var up = new FakeUpstream();
up.Set("Temp", 50);
using var eng = Build(up, new ScriptingOptions { QuarantineAfterConsecutiveTimeouts = 1 });
await eng.LoadAsync([Alarm("a", $"var t = ctx.GetTag(\"Temp\"); {SpinPredicate}")],
TestContext.Current.CancellationToken);
eng.RunGuard.IsQuarantined("a", out _).ShouldBeTrue();
// The operator fixed the predicate and redeployed.
await eng.LoadAsync([Alarm("a", """return (int)ctx.GetTag("Temp").Value > 100;""")],
TestContext.Current.CancellationToken);
eng.RunGuard.IsQuarantined("a", out _).ShouldBeFalse();
up.Push("Temp", 150);
await WaitForAsync(() => eng.GetState("a")!.Active == AlarmActiveState.Active);
}
private static async Task WaitForAsync(Func cond, int timeoutMs = 10_000)
{
var deadline = DateTime.UtcNow.AddMilliseconds(timeoutMs);
while (DateTime.UtcNow < deadline)
{
if (cond()) return;
await Task.Delay(25);
}
throw new TimeoutException("Condition did not become true in time");
}
}