fix(historian): alarm adapter drains a bounded queue; drop the senderless receive (lmxopcua#746)

HistorianAdapterActor fired `_ = EnqueueAsync(...)` per alarm transition
with no in-flight cap, so an alarm storm (a native source re-raising
hundreds of conditions on reconnect) spawned hundreds of concurrent SQLite
inserts that serialised on the file lock and held pool threads. It also kept
a Receive<AlarmHistorianEvent> with no production sender: a transition told
directly AND published on `alerts` would have been historized twice.

Transitions now go to a bounded channel (10000) drained by one loop, so the
sink sees one enqueue at a time in arrival order; a full queue holds the
mailbox (backpressure, nothing dropped); PostStop flushes the queue within
the shared 5 s stop budget. The direct receive is deleted. The gate tests
drive transitions instead; a directly-told AlarmHistorianEvent is shown to
be unhandled against a positive control.

Rebased from the unpushed remed/lane-H2 branch (originally authored 2026-10-22
in session_01Ub9hGmYLQvn9XETzw7CMsT) onto current master.

Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV

src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/HistorianAdapterActor.cs

@@ -1,3 +1,4 @@
+using System.Threading.Channels;
 using Akka.Actor;
 using Akka.Cluster.Tools.PublishSubscribe;
 using Akka.Event;
@@ -12,9 +13,12 @@ using ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms;
 namespace ZB.MOM.WW.OtOpcUa.Runtime.Historian;
 
 /// <summary>
-/// Thin actor wrapper around <see cref="IAlarmHistorianSink"/>. Engine code (ScriptedAlarmActor,
-/// Galaxy native alarm bridge, AB CIP ALMD reader) tells <see cref="AlarmHistorianEvent"/>s to this
-/// actor; the actor enqueues them on the sink fire-and-forget. Production deployments register
+/// Thin actor wrapper around <see cref="IAlarmHistorianSink"/>. Live alarm transitions arrive off the cluster
+/// <c>alerts</c> topic; the Primary's adapter translates each to an <see cref="AlarmHistorianEvent"/> and queues it
+/// for the sink. The queue is bounded and drained by ONE loop, so the sink sees one enqueue at a time in arrival order
+/// (an alarm storm used to fan into one concurrent SQLite insert per transition, serialising on the file lock and
+/// holding pool threads); a full queue holds the mailbox until room frees (backpressure, nothing dropped), and stopping
+/// flushes the queue within the shared 5 s stop budget. Production deployments register
 /// <see cref="LocalDbStoreAndForwardSink"/> against <c>IAlarmHistorianSink</c>; the sink owns the
 /// durable queue + drain-to-HistorianGateway-SendEvent loop. The actor here owns nothing operational beyond
 /// the message contract — its job is to keep the engine actors on Akka's mailbox without blocking
@@ -34,6 +38,18 @@ public sealed class HistorianAdapterActor : ReceiveActor
     private readonly ILoggingAdapter _log = Context.GetLogger();
     private IActorRef? _mediator;
 
+    /// <summary>The most events queued for the sink before a transition waits for room (backpressure).</summary>
+    public const int QueueCapacity = 10_000;
+
+    /// <summary>How long stopping waits for the queue to flush (the runtime's shared actor-stop budget).</summary>
+    public static readonly TimeSpan FlushBudget = TimeSpan.FromSeconds(5);
+
+    private readonly Channel<AlarmHistorianEvent> _queue = Channel.CreateBounded<AlarmHistorianEvent>(
+        new BoundedChannelOptions(QueueCapacity) { SingleReader = true, SingleWriter = true, FullMode = BoundedChannelFullMode.Wait });
+
+    private readonly CancellationTokenSource _lifetime = new();
+    private Task _drain = Task.CompletedTask;
+
     /// <summary>Cached local <see cref="RedundancyRole"/> from the latest <see cref="RedundancyStateChanged"/>
     /// snapshot (null = unknown until the first snapshot arrives, or no <see cref="_localNode"/> wired). The
     /// durable sink enqueue in the <see cref="AlarmHistorianEvent"/> handler is gated on this: only the
@@ -60,10 +76,8 @@ public sealed class HistorianAdapterActor : ReceiveActor
         _sink = sink;
         _localNode = localNode;
 
-        // A direct AlarmHistorianEvent source is kept for a future engine→historian path; it goes through the
-        // SAME ShouldHistorize gate. WARNING: do NOT Tell an AlarmHistorianEvent for a transition that is ALSO
-        // published on the `alerts` topic — the sink would then double-write (both handlers enqueue it).
-        Receive<AlarmHistorianEvent>(evt => { if (ShouldHistorize()) _ = EnqueueAsync(evt); });
+        // (The direct AlarmHistorianEvent receive is gone: it had no production sender, and a transition told directly
+        // AND published on `alerts` would have been written twice.)
 
         // Live alarm transitions arrive off the cluster `alerts` DPS topic (subscribed in PreStart). The
         // Primary ScriptedAlarmHostActor publishes each transition ONCE, but DistributedPubSub fans that
@@ -76,7 +90,21 @@ public sealed class HistorianAdapterActor : ReceiveActor
         // an old-format message missing the field deserializes to null and is historized (default-on), so no
         // audit row is dropped at a handover — same posture as the AlarmTypeName null-coalesce in Translate.
         // The producer (ScriptedAlarmHostActor) always sets a concrete true/false.
-        Receive<AlarmTransitionEvent>(t => { if (ShouldHistorize() && t.HistorizeToAveva is not false) _ = EnqueueAsync(Translate(t)); });
+        ReceiveAsync<AlarmTransitionEvent>(async t =>
+        {
+            if (!ShouldHistorize() || t.HistorizeToAveva is false) return;
+            var evt = Translate(t);
+            // Off the mailbox unless the queue is full: then the mailbox waits for room (bounded memory, nothing dropped).
+            if (_queue.Writer.TryWrite(evt)) return;
+            try
+            {
+                await _queue.Writer.WriteAsync(evt, _lifetime.Token).ConfigureAwait(false);
+            }
+            catch (Exception ex) when (ex is OperationCanceledException or ChannelClosedException)
+            {
+                // Stopping raced the enqueue.
+            }
+        });
 
         Receive<GetStatus>(_ => Sender.Tell(_sink.GetStatus()));
 
@@ -134,9 +162,30 @@ public sealed class HistorianAdapterActor : ReceiveActor
         _mediator = DistributedPubSub.Get(Context.System).Mediator;
         _mediator.Tell(new Subscribe(OpcUaPublishActor.RedundancyStateTopic, Self));
         _mediator.Tell(new Subscribe(ScriptedAlarmHostActor.AlertsTopic, Self));
+        _drain = Task.Run(DrainAsync);
         base.PreStart();
     }
 
+    /// <inheritdoc />
+    protected override void PostStop()
+    {
+        // Flush what the Primary already accepted from the alerts topic, bounded by the shared stop budget.
+        _queue.Writer.TryComplete();
+        if (!_drain.Wait(FlushBudget))
+            _log.Warning("Historian adapter stopped with alarm events still queued for the sink; abandoned after {Seconds} s.",
+                FlushBudget.TotalSeconds);
+        _lifetime.Cancel();
+        _lifetime.Dispose();
+        base.PostStop();
+    }
+
+    /// <summary>The single loop feeding the sink, in arrival order. Never throws: a rejected event is logged.</summary>
+    private async Task DrainAsync()
+    {
+        await foreach (var evt in _queue.Reader.ReadAllAsync(CancellationToken.None).ConfigureAwait(false))
+            await EnqueueAsync(evt).ConfigureAwait(false);
+    }
+
     /// <summary>Caches this node's <see cref="RedundancyRole"/> from a cluster redundancy snapshot so the
     /// <see cref="AlarmHistorianEvent"/> handler can gate the durable sink enqueue to the Primary. A snapshot
     /// that doesn't mention <see cref="_localNode"/> (or no local node wired) leaves the cached role unchanged
@@ -160,7 +209,11 @@ public sealed class HistorianAdapterActor : ReceiveActor
     {
         try
         {
-            await _sink.EnqueueAsync(evt, CancellationToken.None);
+            await _sink.EnqueueAsync(evt, _lifetime.Token).ConfigureAwait(false);
+        }
+        catch (OperationCanceledException) when (_lifetime.IsCancellationRequested)
+        {
+            // Stopped past the flush budget.
         }
         catch (Exception ex)
         {

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Health/HealthProbeActorTests.cs

@@ -77,18 +77,19 @@ public sealed class HealthProbeActorTests : RuntimeActorTestBase
         var sink = new RecordingSink();
         var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink));
 
+        // Transitions arrive as AlarmTransitionEvent (the alerts-topic shape); the direct AlarmHistorianEvent receive
+        // had no production sender and was removed.
         for (var i = 0; i < 5; i++)
-            actor.Tell(new AlarmHistorianEvent(
+            actor.Tell(new ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts.AlarmTransitionEvent(
                 AlarmId: $"alm-{i}",
                 EquipmentPath: "Plant/LineA",
                 AlarmName: $"Alarm{i}",
-                AlarmTypeName: "LimitAlarm",
-                Severity: AlarmSeverity.High,
-                EventKind: "Activated",
+                TransitionKind: "Activated",
+                Severity: 700,
                 Message: $"Test alarm {i}",
                 User: "system",
-                Comment: null,
-                TimestampUtc: DateTime.UtcNow));
+                TimestampUtc: DateTime.UtcNow,
+                AlarmTypeName: "LimitAlarm"));
 
         AwaitCondition(() => sink.Enqueued.Count == 5, TimeSpan.FromSeconds(2));
         sink.Enqueued.Select(e => e.AlarmId).OrderBy(s => s).ShouldBe(
@@ -102,9 +103,9 @@ public sealed class HealthProbeActorTests : RuntimeActorTestBase
         var sink = new RecordingSink();
         var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink));
 
-        actor.Tell(new AlarmHistorianEvent(
-            "alm-x", "Plant/LineB", "OffNormal", "OffNormalAlarm",
-            AlarmSeverity.Low, "Activated", "msg", "system", null, DateTime.UtcNow));
+        actor.Tell(new ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts.AlarmTransitionEvent(
+            AlarmId: "alm-x", EquipmentPath: "Plant/LineB", AlarmName: "OffNormal", TransitionKind: "Activated",
+            Severity: 200, Message: "msg", User: "system", TimestampUtc: DateTime.UtcNow, AlarmTypeName: "OffNormalAlarm"));
         AwaitCondition(() => sink.Enqueued.Count == 1, TimeSpan.FromSeconds(2));
 
         var status = await actor.Ask<HistorianSinkStatus>(

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/HistorianAdapterActorTests.cs

@@ -80,19 +80,6 @@ public sealed class HistorianAdapterActorTests : RuntimeActorTestBase
             DrainState: HistorianDrainState.Idle);
     }
 
-    /// <summary>Builds a minimal <see cref="AlarmHistorianEvent"/> for the gate tests.</summary>
-    private static AlarmHistorianEvent SampleEvent() => new(
-        AlarmId: "alm-1",
-        EquipmentPath: "Area/Line/Equip",
-        AlarmName: "HiHi",
-        AlarmTypeName: "LimitAlarm",
-        Severity: AlarmSeverity.High,
-        EventKind: "Activated",
-        Message: "level high",
-        User: "system",
-        Comment: null,
-        TimestampUtc: DateTime.UtcNow);
-
     /// <summary>Tell <paramref name="actor"/> a <see cref="RedundancyStateChanged"/> snapshot marking
     /// <see cref="LocalNode"/> with <paramref name="role"/> so the gate observes the local role.</summary>
     private static void TellRedundancyRole(IActorRef actor, RedundancyRole role) =>
@@ -117,7 +104,7 @@ public sealed class HistorianAdapterActorTests : RuntimeActorTestBase
         var sink = new RecordingSink();
         var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink, LocalNode));
 
-        actor.Tell(SampleEvent());
+        actor.Tell(SampleTransition());
 
         AwaitAssert(() => sink.EnqueueCount.ShouldBe(1), AssertTimeout);
     }
@@ -131,7 +118,7 @@ public sealed class HistorianAdapterActorTests : RuntimeActorTestBase
         var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink, LocalNode));
 
         TellRedundancyRole(actor, RedundancyRole.Secondary);
-        actor.Tell(SampleEvent());
+        actor.Tell(SampleTransition());
 
         // Give the (suppressed) fire-and-forget a stable window, then assert nothing landed.
         ExpectNoMsg(Settle);
@@ -146,7 +133,7 @@ public sealed class HistorianAdapterActorTests : RuntimeActorTestBase
         var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink, LocalNode));
 
         TellRedundancyRole(actor, RedundancyRole.Detached);
-        actor.Tell(SampleEvent());
+        actor.Tell(SampleTransition());
 
         ExpectNoMsg(Settle);
         sink.EnqueueCount.ShouldBe(0);
@@ -161,7 +148,7 @@ public sealed class HistorianAdapterActorTests : RuntimeActorTestBase
         var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink, LocalNode));
 
         TellRedundancyRole(actor, RedundancyRole.Primary);
-        actor.Tell(SampleEvent());
+        actor.Tell(SampleTransition());
 
         AwaitAssert(() => sink.EnqueueCount.ShouldBe(1), AssertTimeout);
     }
@@ -189,7 +176,7 @@ public sealed class HistorianAdapterActorTests : RuntimeActorTestBase
             },
             CorrelationId.NewId()));
 
-        actor.Tell(SampleEvent());
+        actor.Tell(SampleTransition());
 
         // Local role is still unknown ⇒ default-historize path: sink must record exactly one enqueue.
         AwaitAssert(() => sink.EnqueueCount.ShouldBe(1), AssertTimeout);
@@ -346,4 +333,81 @@ public sealed class HistorianAdapterActorTests : RuntimeActorTestBase
             () => sink.Events.ShouldHaveSingleItem().AlarmTypeName.ShouldBe("AlarmCondition"),
             AssertTimeout);
     }
+
+    /// <summary>
+    ///     The direct <see cref="AlarmHistorianEvent"/> receive is gone (it had no production sender and was a latent
+    ///     double-historize trap: a transition told directly AND published on <c>alerts</c> would be written twice). A
+    ///     directly-told event is unhandled; the positive control — a transition after it — is historized exactly once.
+    /// </summary>
+    [Fact]
+    public void A_directly_told_AlarmHistorianEvent_is_not_historized()
+    {
+        var sink = new RecordingSink();
+        var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink));
+
+        actor.Tell(new AlarmHistorianEvent("alm-1", "Area/Line/Equip", "HiHi", "LimitAlarm", AlarmSeverity.High,
+            "Activated", "level high", "system", null, DateTime.UtcNow));
+        actor.Tell(SampleTransition());
+
+        AwaitAssert(() => sink.EnqueueCount.ShouldBe(1), AssertTimeout);
+        ExpectNoMsg(Settle);
+        sink.Events.ShouldHaveSingleItem().AlarmId.ShouldBe("alm-9");
+    }
+
+    /// <summary>A sink whose enqueue takes time and reports how many enqueues overlap.</summary>
+    private sealed class SlowSink(TimeSpan delay) : IAlarmHistorianSink
+    {
+        private int _inFlight;
+        private int _maxInFlight;
+        private readonly object _lock = new();
+        private readonly List<string> _order = new();
+
+        public int MaxInFlight => Volatile.Read(ref _maxInFlight);
+        public IReadOnlyList<string> Order { get { lock (_lock) return _order.ToArray(); } }
+
+        public async Task EnqueueAsync(AlarmHistorianEvent evt, CancellationToken cancellationToken)
+        {
+            var now = Interlocked.Increment(ref _inFlight);
+            int seen;
+            while ((seen = Volatile.Read(ref _maxInFlight)) < now && Interlocked.CompareExchange(ref _maxInFlight, now, seen) != seen) { }
+            await Task.Delay(delay, cancellationToken);
+            lock (_lock) _order.Add(evt.Message ?? "");
+            Interlocked.Decrement(ref _inFlight);
+        }
+
+        public HistorianSinkStatus GetStatus() => new(0, 0, null, null, null, HistorianDrainState.Idle);
+    }
+
+    /// <summary>
+    ///     An alarm storm no longer fans into one concurrent sink insert per transition: the adapter drains a bounded
+    ///     queue with one loop, so the sink sees one enqueue at a time, in arrival order, and every event lands.
+    /// </summary>
+    [Fact]
+    public void An_alarm_storm_is_enqueued_one_at_a_time_in_order()
+    {
+        var sink = new SlowSink(TimeSpan.FromMilliseconds(5));
+        var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink));
+
+        for (var i = 0; i < 40; i++) actor.Tell(SampleTransition() with { Message = $"m{i:00}" });
+
+        AwaitAssert(() => sink.Order.Count.ShouldBe(40), TimeSpan.FromSeconds(10));
+        sink.MaxInFlight.ShouldBe(1, "one loop drains the queue — never one concurrent insert per transition");
+        sink.Order.ShouldBe(Enumerable.Range(0, 40).Select(i => $"m{i:00}"));
+    }
+
+    /// <summary>Stopping the adapter flushes the events still queued within the stop budget.</summary>
+    [Fact]
+    public void Stopping_flushes_the_queued_events()
+    {
+        var sink = new SlowSink(TimeSpan.FromMilliseconds(20));
+        var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink));
+        for (var i = 0; i < 10; i++) actor.Tell(SampleTransition() with { Message = $"m{i:00}" });
+
+        // PoisonPill queues behind the transitions (Sys.Stop would overtake them in the mailbox): all ten are accepted
+        // into the adapter's queue, most still waiting on the slow sink when the actor stops — and they are flushed.
+        Watch(actor);
+        actor.Tell(PoisonPill.Instance);
+        ExpectTerminated(actor, TimeSpan.FromSeconds(10));
+        sink.Order.Count.ShouldBe(10);
+    }
 }