Joseph Doherty
4e25f4f821
Both off-mailbox loops read _lifetime.Token per operation. PostStop disposes that source after the flush budget, and a disposed source's Token throws, so an operation started past the budget would have logged a spurious failure. Each loop now captures the token when it starts. Also rewrites the recorder's stale "capture Self before the await" note for the enqueue path. 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
227 lines
12 KiB
C#
227 lines
12 KiB
C#
using System.Threading.Channels;
|
||
using Akka.Actor;
|
||
using Akka.Cluster.Tools.PublishSubscribe;
|
||
using Akka.Event;
|
||
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts;
|
||
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
|
||
using ZB.MOM.WW.OtOpcUa.Commons.Types;
|
||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
|
||
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
|
||
using ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms;
|
||
|
||
namespace ZB.MOM.WW.OtOpcUa.Runtime.Historian;
|
||
|
||
/// <summary>
|
||
/// 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
|
||
/// them on disk I/O or gateway round-trips.
|
||
///
|
||
/// Query queue depth + drain health via <see cref="GetStatus"/>.
|
||
/// </summary>
|
||
public sealed class HistorianAdapterActor : ReceiveActor
|
||
{
|
||
public sealed record GetStatus
|
||
{
|
||
public static readonly GetStatus Instance = new();
|
||
}
|
||
|
||
private readonly IAlarmHistorianSink _sink;
|
||
private readonly NodeId? _localNode;
|
||
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
|
||
/// Primary historizes (default-write while unknown so single-node deploys + the boot window never drop
|
||
/// historization).</summary>
|
||
private RedundancyRole? _localRole;
|
||
|
||
/// <summary>Creates the props for a HistorianAdapterActor instance.</summary>
|
||
/// <param name="sink">The alarm historian sink implementation, or null to use a null sink.</param>
|
||
/// <param name="localNode">The local cluster node id, used to read this node's <see cref="RedundancyRole"/>
|
||
/// from the <c>redundancy-state</c> topic so only the Primary historizes to the durable sink. Null (the
|
||
/// default) leaves the role unknown ⇒ default-write (single-node deploys + tests).</param>
|
||
/// <returns>Props configured for creating a HistorianAdapterActor.</returns>
|
||
public static Props Props(IAlarmHistorianSink? sink = null, NodeId? localNode = null) =>
|
||
Akka.Actor.Props.Create(() => new HistorianAdapterActor(sink ?? NullAlarmHistorianSink.Instance, localNode));
|
||
|
||
/// <summary>Initializes a new instance of the HistorianAdapterActor class.</summary>
|
||
/// <param name="sink">The alarm historian sink to forward enqueued events to.</param>
|
||
/// <param name="localNode">The local cluster node id, used to read this node's <see cref="RedundancyRole"/>
|
||
/// from the <c>redundancy-state</c> topic so only the Primary historizes to the durable sink. Null leaves
|
||
/// the role unknown ⇒ default-write (single-node deploys + tests).</param>
|
||
public HistorianAdapterActor(IAlarmHistorianSink sink, NodeId? localNode = null)
|
||
{
|
||
_sink = sink;
|
||
_localNode = localNode;
|
||
|
||
// (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
|
||
// single message to EVERY node's subscriber — including BOTH central nodes' historian adapters. The
|
||
// ShouldHistorize gate keeps only the Primary writing ⇒ exactly-once across the warm pair.
|
||
// NOTE: Translate is intentionally inside the gate so Secondary/Detached nodes never allocate a
|
||
// discarded AlarmHistorianEvent.
|
||
// t.HistorizeToAveva is not false: only explicit false suppresses the durable sink write. null
|
||
// (CLR default for bool?) and true both historize. null is the rolling-restart / cross-version case:
|
||
// 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.
|
||
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()));
|
||
|
||
// Cluster redundancy snapshots (published on the `redundancy-state` topic, subscribed in PreStart)
|
||
// cache this node's role so the historize gate can scope the durable sink enqueue to the Primary.
|
||
// The PubSub Subscribe acks (redundancy-state + alerts) are acked back to Self (no-op below).
|
||
Receive<RedundancyStateChanged>(OnRedundancyStateChanged);
|
||
// The redundancy-state topic also carries OpcUaProbeResult (OpcUaPublishActor peer-probes).
|
||
// We don't consume it here — drop it so it doesn't dead-letter (matches PeerProbeSupervisor).
|
||
Receive<ZB.MOM.WW.OtOpcUa.Runtime.Health.PeerOpcUaProbeActor.OpcUaProbeResult>(_ => { });
|
||
Receive<SubscribeAck>(_ => { });
|
||
}
|
||
|
||
/// <summary>True iff this node should write to the durable sink: only the Primary historizes
|
||
/// (default-write while the role is unknown so single-node deploys + the boot window never drop
|
||
/// historization). DPS fans the Primary's single alerts publish to BOTH nodes' historians, so this
|
||
/// gate is what keeps the durable write exactly-once across the warm-redundant pair.</summary>
|
||
private bool ShouldHistorize() => _localRole is not (RedundancyRole.Secondary or RedundancyRole.Detached);
|
||
|
||
/// <summary>Translates a live <see cref="AlarmTransitionEvent"/> (the alerts-topic shape) into the
|
||
/// historian's <see cref="AlarmHistorianEvent"/>. <c>AlarmTypeName</c> is null-coalesced to
|
||
/// "AlarmCondition": during a rolling restart Akka's JSON serializer applies the CLR default (null) to
|
||
/// an old-format message's <c>AlarmTypeName</c> rather than the record's call-site default, and the
|
||
/// historian must never store a null alarm type.</summary>
|
||
/// <param name="t">The live transition published on the <c>alerts</c> topic.</param>
|
||
/// <returns>The translated historian event.</returns>
|
||
private static AlarmHistorianEvent Translate(AlarmTransitionEvent t) => new(
|
||
AlarmId: t.AlarmId,
|
||
EquipmentPath: t.EquipmentPath,
|
||
AlarmName: t.AlarmName,
|
||
AlarmTypeName: string.IsNullOrEmpty(t.AlarmTypeName) ? "AlarmCondition" : t.AlarmTypeName,
|
||
Severity: ToSeverity(t.Severity),
|
||
EventKind: t.TransitionKind,
|
||
Message: t.Message,
|
||
User: t.User,
|
||
Comment: t.Comment,
|
||
TimestampUtc: t.TimestampUtc);
|
||
|
||
/// <summary>Maps the OPC UA 1–1000 numeric severity carried on a transition back to the coarse
|
||
/// <see cref="AlarmSeverity"/>, inverting <c>ScriptedAlarmHostActor.SeverityToInt</c>'s bucket ceilings
|
||
/// (Low=250, Medium=500, High=750, Critical=1000).</summary>
|
||
/// <param name="severity">The OPC UA 1–1000 numeric severity.</param>
|
||
/// <returns>The coarse alarm severity bucket.</returns>
|
||
private static AlarmSeverity ToSeverity(int severity) => severity switch
|
||
{
|
||
<= 250 => AlarmSeverity.Low,
|
||
<= 500 => AlarmSeverity.Medium,
|
||
<= 750 => AlarmSeverity.High,
|
||
_ => AlarmSeverity.Critical,
|
||
};
|
||
|
||
/// <inheritdoc />
|
||
protected override void PreStart()
|
||
{
|
||
_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()
|
||
{
|
||
// Captured now: PostStop disposes the source after the flush budget, and a disposed source's Token throws.
|
||
var lifetime = _lifetime.Token;
|
||
await foreach (var evt in _queue.Reader.ReadAllAsync(CancellationToken.None).ConfigureAwait(false))
|
||
await EnqueueAsync(evt, lifetime).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
|
||
/// ⇒ default-write. Mirrors <see cref="OpcUaPublishActor"/>'s handler.</summary>
|
||
/// <param name="msg">The cluster redundancy snapshot.</param>
|
||
private void OnRedundancyStateChanged(RedundancyStateChanged msg)
|
||
{
|
||
if (_localNode is null)
|
||
{
|
||
return;
|
||
}
|
||
|
||
var local = msg.Nodes.FirstOrDefault(n => n.NodeId == _localNode.Value);
|
||
if (local is not null)
|
||
{
|
||
_localRole = local.Role;
|
||
}
|
||
}
|
||
|
||
private async Task EnqueueAsync(AlarmHistorianEvent evt, CancellationToken lifetime)
|
||
{
|
||
try
|
||
{
|
||
await _sink.EnqueueAsync(evt, lifetime).ConfigureAwait(false);
|
||
}
|
||
catch (OperationCanceledException) when (lifetime.IsCancellationRequested)
|
||
{
|
||
// Stopped past the flush budget.
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_log.Error(ex, "Historian sink rejected event for {AlarmId} at {Ts}",
|
||
evt.AlarmId, evt.TimestampUtc);
|
||
}
|
||
}
|
||
}
|