diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/HistorianAdapterActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/HistorianAdapterActor.cs index af60b86b..a5629f94 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/HistorianAdapterActor.cs +++ b/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; /// -/// Thin actor wrapper around . Engine code (ScriptedAlarmActor, -/// Galaxy native alarm bridge, AB CIP ALMD reader) tells s to this -/// actor; the actor enqueues them on the sink fire-and-forget. Production deployments register +/// Thin actor wrapper around . Live alarm transitions arrive off the cluster +/// alerts topic; the Primary's adapter translates each to an 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 /// against IAlarmHistorianSink; 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; + /// The most events queued for the sink before a transition waits for room (backpressure). + public const int QueueCapacity = 10_000; + + /// How long stopping waits for the queue to flush (the runtime's shared actor-stop budget). + public static readonly TimeSpan FlushBudget = TimeSpan.FromSeconds(5); + + private readonly Channel _queue = Channel.CreateBounded( + new BoundedChannelOptions(QueueCapacity) { SingleReader = true, SingleWriter = true, FullMode = BoundedChannelFullMode.Wait }); + + private readonly CancellationTokenSource _lifetime = new(); + private Task _drain = Task.CompletedTask; + /// Cached local from the latest /// snapshot (null = unknown until the first snapshot arrives, or no wired). The /// durable sink enqueue in the 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(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(t => { if (ShouldHistorize() && t.HistorizeToAveva is not false) _ = EnqueueAsync(Translate(t)); }); + ReceiveAsync(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(_ => 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(); } + /// + 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(); + } + + /// The single loop feeding the sink, in arrival order. Never throws: a rejected event is logged. + private async Task DrainAsync() + { + await foreach (var evt in _queue.Reader.ReadAllAsync(CancellationToken.None).ConfigureAwait(false)) + await EnqueueAsync(evt).ConfigureAwait(false); + } + /// Caches this node's from a cluster redundancy snapshot so the /// handler can gate the durable sink enqueue to the Primary. A snapshot /// that doesn't mention (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) { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Health/HealthProbeActorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Health/HealthProbeActorTests.cs index 0dacfebf..4a113136 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Health/HealthProbeActorTests.cs +++ b/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( diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/HistorianAdapterActorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/HistorianAdapterActorTests.cs index d58ff74d..826137fc 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/HistorianAdapterActorTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/HistorianAdapterActorTests.cs @@ -80,19 +80,6 @@ public sealed class HistorianAdapterActorTests : RuntimeActorTestBase DrainState: HistorianDrainState.Idle); } - /// Builds a minimal for the gate tests. - 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); - /// Tell a snapshot marking /// with so the gate observes the local role. 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); } + + /// + /// The direct receive is gone (it had no production sender and was a latent + /// double-historize trap: a transition told directly AND published on alerts would be written twice). A + /// directly-told event is unhandled; the positive control — a transition after it — is historized exactly once. + /// + [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"); + } + + /// A sink whose enqueue takes time and reports how many enqueues overlap. + private sealed class SlowSink(TimeSpan delay) : IAlarmHistorianSink + { + private int _inFlight; + private int _maxInFlight; + private readonly object _lock = new(); + private readonly List _order = new(); + + public int MaxInFlight => Volatile.Read(ref _maxInFlight); + public IReadOnlyList 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); + } + + /// + /// 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. + /// + [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}")); + } + + /// Stopping the adapter flushes the events still queued within the stop budget. + [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); + } }