fix(alarms): engine cache prune + bounded dispose, source Warning, cached dead-letter depth, script-log rate limit (E3)

- R05:1/12 ScriptedAlarmEngine: _valueCache is cleared in LoadAsync (it is re-seeded for the referenced paths) so retired tags no longer accrete across deploys; OnUpstreamChange no longer writes the cache after Dispose. - R05:3 ScriptedAlarmEngine.Dispose drains in-flight work for at most 5 s (DisposeDrainTimeout), then cancels it through a shutdown token every DrainDirtyAsync/ShelvingCheckAsync now runs on, logs a Warning and gives one second of grace — a stalled IAlarmStateStore.SaveAsync can no longer hang host shutdown. A save that completes inside the bound is still drained (positive control test). - R05:6 ScriptedAlarmSource logs a throwing OnAlarmEvent subscriber at Warning (naming the alarm) instead of swallowing it. - R05:7 LocalDbStoreAndForwardSink keeps a cached dead-letter counter in step with every dead-letter/revive/purge path and resyncs it with the queued counter on the same cadence; GetStatus no longer runs a COUNT(*) scan. - R05:9 ScriptLogRateLimiter (20 per source per minute, fixed window, one summary entry naming the dropped count when the window rolls) behind ScriptLogTopicSink (key = script|vtag|alarm) and ScriptLogCompanionSink (key = script name). Refs #734 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
2026-10-22 10:17:30 -04:00
parent 2e9160317f
commit 6cbf763719
10 changed files with 683 additions and 32 deletions
@@ -342,6 +342,40 @@ public sealed class LocalDbStoreAndForwardSinkTests : IDisposable
        sink.GetStatus().DeadLetterDepth.ShouldBe(0, "purged past retention");
    }

+    /// <summary>
+    ///     Review R05:7 — <c>GetStatus</c> no longer runs a <c>COUNT(*)</c> table scan per call for the
+    ///     dead-letter depth: it reads a cached counter kept in step by every dead-letter / revive / purge
+    ///     path (the other tests in this class pin those), and resynced from storage on the same cadence as
+    ///     the queued-row counter. A row flipped underneath the sink (the peer replicating a dead-letter in)
+    ///     is therefore invisible until the next resync — that is the documented trade, and the resync is
+    ///     what makes it bounded rather than permanent.
+    /// </summary>
+    [Fact]
+    public async Task DeadLetterDepth_is_served_from_a_cached_counter_that_the_periodic_resync_corrects()
+    {
+        var writer = new FakeWriter();
+        using var sink = new LocalDbStoreAndForwardSink(Db, writer, _log);
+        await sink.EnqueueAsync(Event("A1"), CancellationToken.None);
+        await sink.EnqueueAsync(Event("A2"), CancellationToken.None);
+        sink.GetStatus().DeadLetterDepth.ShouldBe(0);
+        var probesBefore = sink.DebugCapacityProbeCount;
+
+        // Drift the table underneath the sink, exactly as a replicated dead-letter from the pair peer would.
+        await Db.ExecuteAsync(
+            "UPDATE alarm_sf_events SET dead_lettered = 1 WHERE id IN (SELECT id FROM alarm_sf_events LIMIT 1)",
+            null, CancellationToken.None);
+
+        // Cached: GetStatus did not go to the table, so it still reports the pre-drift depth…
+        sink.GetStatus().DeadLetterDepth.ShouldBe(0, "GetStatus serves the cached counter, not a live COUNT(*)");
+        sink.DebugCapacityProbeCount.ShouldBe(probesBefore, "GetStatus must not probe storage");
+
+        // …and the resync (the one the enqueue cadence drives) brings both counters back in line.
+        await sink.DebugResyncCountersAsync(CancellationToken.None);
+        var status = sink.GetStatus();
+        status.DeadLetterDepth.ShouldBe(1);
+        status.QueueDepth.ShouldBe(1);
+    }
+
    /// <summary>Verifies that RetryDeadLettered requeues dead-lettered rows for retry.</summary>
    [Fact]
    public async Task RetryDeadLettered_requeues_for_retry()
@@ -0,0 +1,159 @@
+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>
+///     Review R05:1 / R05:12 / R05:3 — the engine's value cache is pruned on every reload (it used to keep
+///     every path ever referenced, forever), is no longer written after Dispose, and Dispose's in-flight
+///     drain is bounded (a hung store no longer hangs host shutdown).
+/// </summary>
+[Trait("Category", "Unit")]
+public sealed class ScriptedAlarmEngineLifecycleTests
+{
+    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, IAlarmStateStore? store = null)
+    {
+        var logger = new LoggerConfiguration().CreateLogger();
+        return new ScriptedAlarmEngine(up, store ?? new InMemoryAlarmStateStore(), new ScriptLoggerFactory(logger), logger);
+    }
+
+    [Fact]
+    public async Task Reload_prunes_value_cache_entries_for_paths_the_new_alarm_set_no_longer_references()
+    {
+        var up = new FakeUpstream();
+        up.Set("Temp", 50);
+        up.Set("Pressure", 5);
+        using var eng = Build(up);
+        var ct = TestContext.Current.CancellationToken;
+
+        await eng.LoadAsync(
+            [Alarm("a", """return (int)ctx.GetTag("Temp").Value > 100 || (int)ctx.GetTag("Pressure").Value > 10;""")], ct);
+        eng.ValueCacheCountForTest.ShouldBe(2);
+
+        // The redeploy drops Pressure from every predicate: its entry must go with it.
+        await eng.LoadAsync([Alarm("a", """return (int)ctx.GetTag("Temp").Value > 100;""")], ct);
+        eng.ValueCacheCountForTest.ShouldBe(1);
+        eng.ValueCacheContainsForTest("Pressure").ShouldBeFalse();
+        eng.ValueCacheContainsForTest("Temp").ShouldBeTrue();
+    }
+
+    [Fact]
+    public async Task A_push_after_Dispose_does_not_write_the_value_cache()
+    {
+        var up = new FakeUpstream();
+        up.Set("Temp", 50);
+        var eng = Build(up);
+        await eng.LoadAsync([Alarm("a", """return (int)ctx.GetTag("Temp").Value > 100;""")],
+            TestContext.Current.CancellationToken);
+        eng.Dispose();
+        eng.ValueCacheCountForTest.ShouldBe(1);
+
+        eng.OnUpstreamChange("Late", new DataValueSnapshot(1, 0u, DateTime.UtcNow, DateTime.UtcNow));
+
+        eng.ValueCacheContainsForTest("Late").ShouldBeFalse("a disposed engine must not grow its cache");
+    }
+
+    [Fact]
+    public async Task Dispose_gives_up_on_a_store_that_never_returns_after_the_drain_bound_and_cancels_it()
+    {
+        var up = new FakeUpstream();
+        up.Set("Temp", 50);
+        var store = new NeverReturningSaveStore();
+        var eng = Build(up, store);
+        eng.DisposeDrainTimeout = TimeSpan.FromMilliseconds(300);
+        await eng.LoadAsync([Alarm("A", """return (int)ctx.GetTag("Temp").Value > 100;""")],
+            TestContext.Current.CancellationToken);
+
+        store.BlockNext = true;
+        up.Push("Temp", 150); // Activated → SaveAsync → blocks forever (until cancelled)
+        await WaitForAsync(() => store.SaveInProgress);
+
+        var sw = System.Diagnostics.Stopwatch.StartNew();
+        eng.Dispose();
+        sw.Stop();
+
+        sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(5), "the drain is bounded; a hung store cannot hang shutdown");
+        await WaitForAsync(() => store.ObservedCancellation, timeoutMs: 2000);
+        store.ObservedCancellation.ShouldBeTrue("the in-flight save was cancelled through the engine's shutdown token");
+    }
+
+    [Fact]
+    public async Task Dispose_still_waits_for_a_save_that_completes_inside_the_bound()
+    {
+        // Positive control for the bound: a save that finishes promptly is drained, not cut off.
+        var up = new FakeUpstream();
+        up.Set("Temp", 50);
+        var store = new NeverReturningSaveStore { ReleaseAfter = TimeSpan.FromMilliseconds(100) };
+        var eng = Build(up, store);
+        eng.DisposeDrainTimeout = TimeSpan.FromSeconds(5);
+        await eng.LoadAsync([Alarm("A", """return (int)ctx.GetTag("Temp").Value > 100;""")],
+            TestContext.Current.CancellationToken);
+
+        store.BlockNext = true;
+        up.Push("Temp", 150);
+        await WaitForAsync(() => store.SaveInProgress);
+
+        eng.Dispose();
+
+        store.SaveInProgress.ShouldBeFalse();
+        store.ObservedCancellation.ShouldBeFalse("a save that completes within the bound is never cancelled");
+        store.Saved.ShouldContain("A");
+    }
+
+    private static async Task WaitForAsync(Func<bool> cond, int timeoutMs = 5000)
+    {
+        var deadline = DateTime.UtcNow.AddMilliseconds(timeoutMs);
+        while (DateTime.UtcNow < deadline)
+        {
+            if (cond()) return;
+            await Task.Delay(20);
+        }
+        throw new TimeoutException("Condition did not become true in time");
+    }
+
+    /// <summary>A store whose next SaveAsync blocks until cancelled (or until <see cref="ReleaseAfter"/>).</summary>
+    private sealed class NeverReturningSaveStore : IAlarmStateStore
+    {
+        private readonly InMemoryAlarmStateStore _inner = new();
+        public volatile bool BlockNext;
+        public volatile bool SaveInProgress;
+        public volatile bool ObservedCancellation;
+        public TimeSpan? ReleaseAfter { get; set; }
+        public List<string> Saved { get; } = [];
+
+        public Task<AlarmConditionState?> LoadAsync(string alarmId, CancellationToken ct) => _inner.LoadAsync(alarmId, ct);
+        public Task<IReadOnlyList<AlarmConditionState>> LoadAllAsync(CancellationToken ct) => _inner.LoadAllAsync(ct);
+        public Task RemoveAsync(string alarmId, CancellationToken ct) => _inner.RemoveAsync(alarmId, ct);
+
+        public async Task SaveAsync(AlarmConditionState state, CancellationToken ct)
+        {
+            if (BlockNext)
+            {
+                BlockNext = false;
+                SaveInProgress = true;
+                try
+                {
+                    await Task.Delay(ReleaseAfter ?? Timeout.InfiniteTimeSpan, ct).ConfigureAwait(false);
+                }
+                catch (OperationCanceledException)
+                {
+                    ObservedCancellation = true;
+                    throw;
+                }
+                finally
+                {
+                    SaveInProgress = false;
+                }
+            }
+            lock (Saved) Saved.Add(state.AlarmId);
+            await _inner.SaveAsync(state, ct).ConfigureAwait(false);
+        }
+    }
+}
@@ -0,0 +1,59 @@
+using Serilog;
+using Serilog.Core;
+using Serilog.Events;
+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>Review R05:6 — a throwing <see cref="ScriptedAlarmSource.OnAlarmEvent"/> subscriber used to be
+/// swallowed with no diagnostic: a lost alarm event with no trace. It is now logged at Warning.</summary>
+[Trait("Category", "Unit")]
+public sealed class ScriptedAlarmSourceLoggingTests
+{
+    private sealed class CapturingSink : ILogEventSink
+    {
+        public List<LogEvent> Events { get; } = [];
+        public void Emit(LogEvent logEvent) { lock (Events) Events.Add(logEvent); }
+    }
+
+    [Fact]
+    public async Task A_throwing_subscriber_is_logged_at_Warning_and_does_not_crash_the_adapter()
+    {
+        var sinkCapture = new CapturingSink();
+        var logger = new LoggerConfiguration().MinimumLevel.Verbose().WriteTo.Sink(sinkCapture).CreateLogger();
+        var up = new FakeUpstream();
+        up.Set("Temp", 50);
+        var engine = new ScriptedAlarmEngine(up, new InMemoryAlarmStateStore(), new ScriptLoggerFactory(logger), logger);
+        await engine.LoadAsync(
+            [new ScriptedAlarmDefinition("Plant/Line1::HighTemp", "Plant/Line1", "HighTemp", AlarmKind.LimitAlarm,
+                AlarmSeverity.High, "Temp {Temp}C", """return (int)ctx.GetTag("Temp").Value > 100;""")],
+            TestContext.Current.CancellationToken);
+        using var _e = engine;
+        using var source = new ScriptedAlarmSource(engine, logger);
+        await source.SubscribeAlarmsAsync(Array.Empty<string>(), TestContext.Current.CancellationToken);
+        source.OnAlarmEvent += (_, _) => throw new InvalidOperationException("subscriber broke");
+
+        up.Push("Temp", 150);
+
+        var deadline = DateTime.UtcNow.AddSeconds(10);
+        while (DateTime.UtcNow < deadline)
+        {
+            lock (sinkCapture.Events)
+            {
+                if (sinkCapture.Events.Any(e => e.Level == LogEventLevel.Warning && e.Exception?.Message == "subscriber broke"))
+                    break;
+            }
+            await Task.Delay(25);
+        }
+
+        LogEvent warning;
+        lock (sinkCapture.Events)
+            warning = sinkCapture.Events.Single(e => e.Level == LogEventLevel.Warning && e.Exception?.Message == "subscriber broke");
+        warning.RenderMessage().ShouldContain("Plant/Line1::HighTemp");
+        // The adapter survived: the engine's state still moved.
+        engine.GetState("Plant/Line1::HighTemp")!.Active.ShouldBe(AlarmActiveState.Active);
+    }
+}
@@ -0,0 +1,137 @@
+using Serilog;
+using Serilog.Core;
+using Serilog.Events;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging;
+
+namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests;
+
+/// <summary>
+///     Per-source rate limit on the script-log fan-out (review R05:9 / R02:P6): a script that logs — or
+///     fails — on every evaluation of a 1 Hz input used to put one entry per second per script onto the
+///     cluster bus and the main log. The limiter admits the first 20 entries per source per minute, drops
+///     the rest, and surfaces one summary line (with the dropped count) when the next entry arrives after
+///     the window rolled — so nothing is silently lost, and a chatty script cannot shed everyone else's
+///     entries out of the bounded telemetry channels.
+/// </summary>
+[Trait("Category", "Unit")]
+public sealed class ScriptLogRateLimiterTests
+{
+    private sealed class Clock
+    {
+        public DateTime Now = new(2026, 1, 1, 12, 0, 0, DateTimeKind.Utc);
+        public void Advance(TimeSpan by) => Now += by;
+    }
+
+    [Fact]
+    public void Admits_the_first_N_per_window_then_suppresses_and_summarises_once_the_window_rolls()
+    {
+        var clock = new Clock();
+        var limiter = new ScriptLogRateLimiter(limitPerWindow: 3, window: TimeSpan.FromMinutes(1), clock: () => clock.Now);
+
+        limiter.Admit("s1").Admitted.ShouldBeTrue();
+        limiter.Admit("s1").Admitted.ShouldBeTrue();
+        limiter.Admit("s1").Admitted.ShouldBeTrue();
+        limiter.Admit("s1").Admitted.ShouldBeFalse();
+        limiter.Admit("s1").Admitted.ShouldBeFalse();
+        limiter.Admit("s1").SuppressedInPreviousWindow.ShouldBe(0, "no summary until the window rolls");
+
+        clock.Advance(TimeSpan.FromMinutes(1));
+        var first = limiter.Admit("s1");
+        first.Admitted.ShouldBeTrue("a fresh window admits again");
+        first.SuppressedInPreviousWindow.ShouldBe(3, "the caller emits one summary for the three dropped entries");
+        limiter.Admit("s1").SuppressedInPreviousWindow.ShouldBe(0, "the summary is reported exactly once");
+    }
+
+    [Fact]
+    public void Sources_are_independent()
+    {
+        var limiter = new ScriptLogRateLimiter(limitPerWindow: 1);
+        limiter.Admit("a").Admitted.ShouldBeTrue();
+        limiter.Admit("a").Admitted.ShouldBeFalse();
+        limiter.Admit("b").Admitted.ShouldBeTrue("a chatty source must not shed another source's entries");
+    }
+
+    [Fact]
+    public void Defaults_are_20_per_minute()
+    {
+        ScriptLogRateLimiter.DefaultLimitPerWindow.ShouldBe(20);
+        ScriptLogRateLimiter.DefaultWindow.ShouldBe(TimeSpan.FromMinutes(1));
+        var limiter = new ScriptLogRateLimiter();
+        for (var i = 0; i < 20; i++) limiter.Admit("s").Admitted.ShouldBeTrue();
+        limiter.Admit("s").Admitted.ShouldBeFalse();
+    }
+
+    // ── the two sinks ────────────────────────────────────────────────────────────────────────
+
+    private sealed class FakePublisher : IScriptLogPublisher
+    {
+        public List<ScriptLogEntry> Published { get; } = [];
+        public void Publish(ScriptLogEntry entry) => Published.Add(entry);
+    }
+
+    private static LogEvent Evt(LogEventLevel level, string template, params (string Key, string Val)[] props)
+    {
+        var tmpl = new Serilog.Parsing.MessageTemplateParser().Parse(template);
+        var logProps = props.Select(p => new LogEventProperty(p.Key, new ScalarValue(p.Val)));
+        return new LogEvent(DateTimeOffset.UtcNow, level, exception: null, tmpl, logProps);
+    }
+
+    [Fact]
+    public void Topic_sink_rate_limits_per_source_and_publishes_a_summary_entry()
+    {
+        var clock = new Clock();
+        var publisher = new FakePublisher();
+        var sink = new ScriptLogTopicSink(publisher, LogEventLevel.Information,
+            new ScriptLogRateLimiter(limitPerWindow: 2, clock: () => clock.Now));
+
+        for (var i = 0; i < 5; i++)
+            sink.Emit(Evt(LogEventLevel.Warning, "chatty", (ScriptLoggerFactory.ScriptIdProperty, "S1"), (ScriptLoggerFactory.VirtualTagIdProperty, "V1")));
+        sink.Emit(Evt(LogEventLevel.Warning, "other", (ScriptLoggerFactory.ScriptIdProperty, "S2")));
+
+        publisher.Published.Count(e => e.ScriptId == "S1").ShouldBe(2, "the limit is per source");
+        publisher.Published.Count(e => e.ScriptId == "S2").ShouldBe(1, "another source is unaffected");
+
+        clock.Advance(TimeSpan.FromMinutes(1));
+        sink.Emit(Evt(LogEventLevel.Warning, "chatty again", (ScriptLoggerFactory.ScriptIdProperty, "S1"), (ScriptLoggerFactory.VirtualTagIdProperty, "V1")));
+
+        var s1 = publisher.Published.Where(e => e.ScriptId == "S1").ToList();
+        s1.Count.ShouldBe(4, "2 admitted + 1 summary + 1 admitted in the new window");
+        var summary = s1[2];
+        summary.Message.ShouldContain("3");
+        summary.Message.ShouldContain("suppressed");
+        summary.VirtualTagId.ShouldBe("V1", "the summary carries the source's identity so the page attributes it");
+        summary.Level.ShouldBe("Warning");
+        s1[3].Message.ShouldBe("chatty again");
+    }
+
+    private sealed class CapturingSink : ILogEventSink
+    {
+        public List<LogEvent> Events { get; } = [];
+        public void Emit(LogEvent logEvent) => Events.Add(logEvent);
+    }
+
+    [Fact]
+    public void Companion_sink_rate_limits_per_script_and_mirrors_a_summary_line()
+    {
+        var clock = new Clock();
+        var main = new CapturingSink();
+        var mainLogger = new LoggerConfiguration().MinimumLevel.Verbose().WriteTo.Sink(main).CreateLogger();
+        var sink = new ScriptLogCompanionSink(mainLogger, LogEventLevel.Error,
+            new ScriptLogRateLimiter(limitPerWindow: 2, clock: () => clock.Now));
+
+        for (var i = 0; i < 5; i++)
+            sink.Emit(Evt(LogEventLevel.Error, "boom", (ScriptLoggerFactory.ScriptNameProperty, "HighTemp")));
+        main.Events.Count.ShouldBe(2);
+
+        clock.Advance(TimeSpan.FromMinutes(1));
+        sink.Emit(Evt(LogEventLevel.Error, "boom", (ScriptLoggerFactory.ScriptNameProperty, "HighTemp")));
+
+        main.Events.Count.ShouldBe(4);
+        var summary = main.Events[2].RenderMessage();
+        summary.ShouldContain("HighTemp");
+        summary.ShouldContain("3");
+        summary.ShouldContain("suppressed");
+    }
+}