lmxopcua/tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/ScriptedAlarmEngineRunGuardTests.cs

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;

/// <summary>
///     <see cref="ScriptedAlarmEngine"/> runs every predicate through a <see cref="ScriptRunGuard"/> 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.
/// </summary>
[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<ScriptedAlarmEvent>();
        eng.OnEvent += (_, e) => { lock (events) events.Add(e); };

        // The load itself evaluates once (startup recovery) — that is the first, abandoned run.
        // The healthy predicate deliberately spins ~100 ms (well inside the 500 ms budget) so every run of it
        // is observably in flight for a while. That keeps the #911 window open on EVERY run of this test
        // instead of only under CPU load: the guard counts a run on StartedRunCount AND InFlightCount the
        // moment it is admitted, but releases InFlightCount only when the run finishes.
        await eng.LoadAsync(
            [Alarm("runaway", $"var t = ctx.GetTag(\"Temp\"); {SpinPredicate}"),
             Alarm("healthy", $$"""{{HealthySpin}} 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);
        }

        // #911: wait for the sibling's last run to FINISH, not merely to start. StartedRunCount reaching 6 only
        // proves the healthy alarm's 4th-push run was admitted — it is still executing (InFlightCount 2: the
        // orphan plus that run) until its finally releases the slot. Waiting on the started count and then
        // asserting the in-flight count was the race.
        await WaitForAsync(() => eng.RunGuard.StartedRunCount >= 1 + 1 + pushes && eng.RunGuard.InFlightCount == 1);

        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, "exactly one orphan, never one per push");
        eng.GetState("runaway")!.Active.ShouldBe(AlarmActiveState.Inactive, "state is held, never invented");

        // Positive control: the healthy sibling still transitions on the same engine. The same push re-evaluates
        // the runaway alarm too, and it must be refused once more — proving the single in-flight run counted
        // above is the runaway's orphan (still holding its slot), not a leftover healthy run.
        up.Push("Temp", 150);
        await WaitForAsync(() => SnapshotOf(events).Any(e => e.AlarmId == "healthy" && e.Emission == EmissionKind.Activated));
        eng.RunGuard.RefusedWhileRunningCount.ShouldBe(pushes + 1);
        eng.RunGuard.StartedRunCount.ShouldBe(1 + 1 + pushes + 1, "only the healthy alarm was dispatched");

        // 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. Checked only AFTER the healthy Activated event
        // arrived: the single drainer fires each pass's events in push order, so every earlier refusal's
        // emissions have already been delivered and a per-push re-annotation could not still be in transit.
        var quality = SnapshotOf(events).Where(e => e.AlarmId == "runaway" && e.Emission == EmissionKind.QualityChanged).ToArray();
        quality.Length.ShouldBe(1);
        quality[0].WorstInputStatusCode.ShouldBe(StatusBadInternalError);
    }

    private const string HealthySpin =
        "var spinUntil = System.DateTime.UtcNow.AddMilliseconds(100); while (System.DateTime.UtcNow < spinUntil) { }";

    private static ScriptedAlarmEvent[] SnapshotOf(List<ScriptedAlarmEvent> events)
    {
        lock (events) return events.ToArray();
    }

    /// <summary>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.</summary>
    [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<ScriptedAlarmEvent>();
        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, TestContext.Current.CancellationToken); // 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<bool> 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");
    }
}