lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/AlarmSeamTests.cs

using System.Collections.Concurrent;
using System.Diagnostics;
using Serilog;
using Shouldly;
using Xunit;
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;

namespace ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests;

/// <summary>A minimal <see cref="ITagUpstreamSource"/>: set or push a value; push notifies subscribers synchronously.</summary>
internal sealed class TestUpstream : ITagUpstreamSource
{
    private readonly ConcurrentDictionary<string, DataValueSnapshot> _values = new(StringComparer.Ordinal);
    private readonly ConcurrentDictionary<string, List<Action<string, DataValueSnapshot>>> _subs = new(StringComparer.Ordinal);

    public void Set(string path, object? value, uint status = 0u)
        => _values[path] = new DataValueSnapshot(value, status, DateTime.UtcNow, DateTime.UtcNow);

    public void Push(string path, object? value, uint status = 0u)
    {
        Set(path, value, status);
        if (!_subs.TryGetValue(path, out var list)) return;
        Action<string, DataValueSnapshot>[] observers;
        lock (list) observers = [.. list];
        foreach (var observer in observers) observer(path, _values[path]);
    }

    public DataValueSnapshot ReadTag(string path)
        => _values.TryGetValue(path, out var v) ? v : new DataValueSnapshot(null, 0x80320000u, null, DateTime.UtcNow);

    public IDisposable SubscribeTag(string path, Action<string, DataValueSnapshot> observer)
    {
        var list = _subs.GetOrAdd(path, _ => []);
        lock (list) list.Add(observer);
        return new Unsubscribe(() => { lock (list) list.Remove(observer); });
    }

    private sealed class Unsubscribe(Action dispose) : IDisposable
    {
        public void Dispose() => dispose();
    }
}

/// <summary>
///     lmxopcua#780 — the scripted-alarm seam over the REAL worker (design §9, alarm column): a runaway predicate is killed
///     at T + G and quarantined, its condition frozen with a <c>BadInternalError</c> annotation and never transitioned; a
///     sibling alarm on the same shard keeps evaluating; and a worker that dies under an in-flight predicate freezes its
///     condition <c>BadResourceUnavailable</c> — it never clears it — and the alarm is re-evaluated by itself on return.
/// </summary>
[Collection(ScriptWorkerCollection.Name)]
[Trait("Category", "ScriptWorker")]
public sealed class AlarmSeamTests
{
    private const uint StatusBadInternalError = 0x80020000u;
    private const uint StatusBadResourceUnavailable = 0x80040000u;

    private static CancellationToken Ct => TestContext.Current.CancellationToken;

    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, TimeSpan budget)
    {
        var logger = new LoggerConfiguration().CreateLogger();
        return new ScriptedAlarmEngine(up, new InMemoryAlarmStateStore(), new ScriptLoggerFactory(logger), logger,
            scriptTimeout: budget, scriptRunner: runner);
    }

    [Fact]
    public async Task A_runaway_predicate_is_killed_and_quarantined_its_alarm_frozen_Bad_and_a_sibling_keeps_evaluating()
    {
        var budget = TimeSpan.FromMilliseconds(300);
        var log = new CapturingSink();
        await using var runner = Scripts.Runner(Scripts.Options(hardGrace: TimeSpan.FromMilliseconds(500), quarantineAfterKills: 2), log.CreateLogger());
        var up = new TestUpstream();
        up.Set("x", 0);
        using var engine = Engine(up, runner, budget);
        var events = new ConcurrentQueue<ScriptedAlarmEvent>();
        engine.OnEvent += (_, e) => events.Enqueue(e);
        await engine.LoadAsync(
        [
            Alarm("runaway", "if ((int)ctx.GetTag(\"x\").Value > 5) { while (true) { } } return false;"),
            Alarm("sibling", "return (int)ctx.GetTag(\"x\").Value > 1;"),
        ], Ct);
        var first = runner.Shards[0].Current.ShouldNotBeNull();

        up.Push("x", 10);
        await Scripts.Eventually(() => events.Any(e => e is { AlarmId: "sibling", Emission: EmissionKind.Activated }), TimeSpan.FromSeconds(10), "the sibling's activation");
        await Scripts.Eventually(() => events.Any(e => e is { AlarmId: "runaway", Emission: EmissionKind.QualityChanged }), TimeSpan.FromSeconds(10), "the runaway's annotation");
        events.Single(e => e.AlarmId == "runaway").WorstInputStatusCode.ShouldBe(StatusBadInternalError);
        (await first.Exited.WaitAsync(TimeSpan.FromSeconds(10), Ct)).CulpritKey
            .ShouldBe(InProcessScriptRunner.ScopedKey(ScriptContextKind.AlarmPredicate, "runaway"));

        // The second kill quarantines it; the sibling (same shard, worker restarted twice) keeps evaluating. Re-pushed:
        // Exited completes before the dead worker's evaluations are released, so the first push after it can find the
        // runaway's slot still held and be refused (not a strike, exactly as in process) — the next push runs it.
        var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(60);
        for (var x = 11; !engine.QuarantinedAlarmIds.Contains("runaway"); x++)
        {
            if (DateTime.UtcNow > deadline)
                throw new TimeoutException("no kill quarantine; events: " +
                    string.Join(", ", events.Select(e => $"{e.AlarmId}:{e.Emission}:{e.WorstInputStatusCode:X8}")) +
                    Environment.NewLine + string.Join(Environment.NewLine, log.Rendered));
            up.Push("x", x);
            await Task.Delay(TimeSpan.FromSeconds(2), Ct);
        }
        engine.QuarantinedAlarmIds.ShouldBe(["runaway"]);
        up.Push("x", 0);
        await Scripts.Eventually(() => events.Any(e => e is { AlarmId: "sibling", Emission: EmissionKind.Cleared }), TimeSpan.FromSeconds(60), "the sibling's clear");

        engine.GetState("runaway")!.Active.ShouldBe(AlarmActiveState.Inactive, "a predicate that never returned never transitions");
        events.Where(e => e.AlarmId == "runaway").ShouldAllBe(e => e.Emission == EmissionKind.QualityChanged);

        // A redeploy (reload) lifts the quarantine, as in process.
        await engine.LoadAsync([Alarm("runaway", "return (int)ctx.GetTag(\"x\").Value > 5;"), Alarm("sibling", "return (int)ctx.GetTag(\"x\").Value > 1;")], Ct);
        engine.QuarantinedAlarmIds.ShouldBeEmpty();
    }

    [Fact]
    public async Task A_worker_that_dies_under_a_predicate_freezes_it_BadResourceUnavailable_never_clears_it_and_it_is_re_evaluated_on_return()
    {
        await using var runner = Scripts.Runner(Scripts.Options(probation: 1));
        var up = new TestUpstream();
        up.Set("ms", 1);
        using var engine = Engine(up, runner, TimeSpan.FromSeconds(5));
        var events = new ConcurrentQueue<ScriptedAlarmEvent>();
        engine.OnEvent += (_, e) => events.Enqueue(e);
        await engine.LoadAsync([Alarm("slow",
            "var ms = (int)ctx.GetTag(\"ms\").Value; var end = ctx.Now.AddMilliseconds(ms); while (ctx.Now < end) { } return ms > 0;")], Ct);
        engine.GetState("slow")!.Active.ShouldBe(AlarmActiveState.Active);
        var worker = runner.Shards[0].Current.ShouldNotBeNull();
        events.Clear();

        up.Push("ms", 1500);
        await Task.Delay(300, Ct); // the predicate is spinning in the worker
        Process.GetProcessById(worker.Pid).Kill();

        await Scripts.Eventually(() => events.Any(e => e.Emission == EmissionKind.QualityChanged), TimeSpan.FromSeconds(10), "the runner-failure annotation");
        events.First().WorstInputStatusCode.ShouldBe(StatusBadResourceUnavailable);
        engine.GetState("slow")!.Active.ShouldBe(AlarmActiveState.Active, "a dead worker never clears an alarm");

        // No further input change: the runner's recovery re-evaluates it (in the isolation worker) and the quality returns.
        await Scripts.Eventually(() => events.Any(e => e is { Emission: EmissionKind.QualityChanged, WorstInputStatusCode: 0u }),
            TimeSpan.FromSeconds(60), "the re-evaluation on return");
        engine.GetState("slow")!.Active.ShouldBe(AlarmActiveState.Active);
        events.ShouldNotContain(e => e.Emission == EmissionKind.Cleared);
    }
}