Files
lmxopcua/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs
T
Joseph Doherty 2d20aeb779 fix(opcua,drivers): carry driver status sub-codes to clients (#519) + health-write ordering (#517)
#519 — a driver's OPC UA status code now reaches a client with its sub-code intact.
DriverInstanceActor projected the driver's uint onto a 3-state Good/Uncertain/Bad enum
and OtOpcUaNodeManager re-expanded Bad to a bare 0x80000000, so BadNoCommunication,
BadTypeMismatch and BadNotSupported were indistinguishable on the wire and every
driver's status constants (incl. the 16 #497 had just corrected) were diagnostic in
logs only. Resolved by option (b): the raw uint rides AttributeValuePublished ->
AttributeValueUpdate -> IOpcUaAddressSpaceSink.WriteValue, OpcUaQuality is deleted, and
severity classification is a helper (OpcUaStatus.IsGood/IsUncertain/IsBad).

The surface was wider than the issue costed (41 files, not "the sink and 3
implementations"): OpcUaQuality had non-driver producers with no uint to carry.
Each got its natural code — BadInternalError for a failed virtual-tag script,
BadNoCommunication for driver comms loss — and ScriptedAlarmHostActor's pair of
mutually-inverse uint<->enum mappers is deleted, having round-tripped a code the
engine already held. AddressSpaceApplier's removal writes deliberately keep the
generic Bad: not losing a code the driver had is this issue; inventing one for the
deploy path is a separate call.

Behaviour change: WriteAlarmQuality fires its Part 9 event on a status-code change
rather than a severity-bucket change. While the annotation was 3-state those were the
same predicate; now a move between two Bad reasons changes what a subscriber reads.

New SDK-free named constants live on Commons.OpcUa.OpcUaStatus, and
StatusCodeParityTests is widened to cover them — that guard is reflection-driven, so
it can only see a named constant, never an inline literal.

#517 — the symptom is real but the issue named the wrong path and its fix was unsafe.
ShutdownAsync cannot race: DriverInstanceActor calls it from PostStop, after the
mailbox drains. (Core/Hosting/DriverHost, the one class with a concurrent GetHealth,
has no production call site.) InitializeAsync can: it runs on a fire-and-forget
Task.Run, and MTConnectDriver writes Healthy then starts loops that can throw into a
catch which disposes the client before writing Faulted.

Moving the write earlier, as suggested, inverts the terminal state — teardown drains
in-flight loops that write health as they unwind, so a late loop could leave Healthy at
rest with every connection closed. The fix is a bracket (conservative before, terminal
after), applied to MTConnect's init-failure path only; shutdown keeps its single
trailing write, and the rule plus the reachability table are documented on
IDriver.ShutdownAsync.

CalculationDriver.GetHealth returned a hard-coded Healthy that consulted nothing —
Healthy before initialize, after shutdown and after dispose. Now derived from _running.
Its test, named Health_is_always_Connected, asserted the defect and is replaced.

Every new guard was falsified by breaking the production code and watching it go red.
The two sub-code tests each assert two different Bad codes, since a single-code
assertion could pass against a hard-coded constant under the old projection.

Pre-existing failures, verified unchanged on a stashed tree and not chased: 4 in
Driver.AbLegacy.IntegrationTests (slc500/micrologix/plc5 fixtures not up) and 3 in
Host.IntegrationTests.

Closes #517
Closes #519

Claude-Session: https://claude.ai/code/session_015p7wGqy3YpZNCpDzTpGMKo
2026-07-31 05:25:27 -04:00

759 lines
45 KiB
C#
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
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.Observability;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry;
using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
namespace ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms;
/// <summary>
/// Akka host that owns one <see cref="ScriptedAlarmEngine"/> for an Equipment-namespace
/// driver node, feeds it live tag values, and bridges its emissions to OPC UA publish + the
/// cluster <c>alerts</c> DistributedPubSub topic.
///
/// <para>
/// The host is the engine's lifecycle owner: the caller (the driver-role startup) builds
/// the <see cref="DependencyMuxTagUpstreamSource"/>, constructs a <see cref="ScriptedAlarmEngine"/>
/// <b>around that same upstream</b>, and passes both here. The host disposes the engine in
/// <see cref="PostStop"/>.
/// </para>
///
/// <para>
/// <b>Data flow.</b> <see cref="DependencyMuxActor"/> delivers a
/// <see cref="VirtualTagActor.DependencyValueChanged"/> for every tag the loaded alarms
/// depend on; the host pushes each into the upstream. The engine self-evaluates: its own
/// <see cref="ITagUpstreamSource.SubscribeTag"/> observer re-runs the predicates and raises
/// <see cref="ScriptedAlarmEngine.OnEvent"/>. The host does NOT call evaluate — it only
/// feeds values in and fans emissions out.
/// </para>
///
/// <para>
/// <b>Thread-safety.</b> <see cref="ScriptedAlarmEngine.OnEvent"/> fires on the engine's
/// background worker thread (its fire-and-forget re-evaluation), NOT on the actor thread.
/// The ctor subscribes a handler that does nothing but <c>Self.Tell(new EngineEmission(e))</c>
/// — <see cref="ICanTell.Tell"/> is thread-safe. ALL sink work (touching <see cref="Context"/>,
/// the publish actor ref, the DPS mediator) then happens on the actor thread in
/// <see cref="OnEngineEmission"/>. No <see cref="Context"/> or actor state is ever read or
/// written from the OnEvent callback.
/// </para>
///
/// <para>
/// <b>Historization.</b> The host publishes each transition as an
/// <see cref="AlarmTransitionEvent"/> to the <c>alerts</c> topic ONLY. That topic is the
/// historization path: <c>HistorianAdapterActor</c>'s upstream and the Admin UI Alerts page
/// both consume <c>alerts</c>. The host deliberately does NOT also <c>Tell</c> the historian
/// adapter directly — doing so would double-historize every transition. (T9 plan called for a
/// "direct historian tell"; that was dropped because the alerts topic already feeds the
/// historian path, so a direct tell would duplicate every row.)
/// </para>
/// </summary>
public sealed class ScriptedAlarmHostActor : ReceiveActor
{
/// <summary>The cluster DistributedPubSub topic every alarm transition is published to. Matches
/// the constant the (retired) <c>ScriptedAlarmActor</c> used so subscribers stay wired.</summary>
public const string AlertsTopic = "alerts";
/// <summary>The cluster DistributedPubSub topic inbound OPC UA Part 9 alarm method calls
/// (Acknowledge / Confirm / Shelve / AddComment) are routed onto as <see cref="AlarmCommand"/>s.
/// The OPC UA node manager's condition handlers build the command (after the <c>AlarmAck</c> role
/// gate, T18) or the AdminUI path republishes via <c>AdminOperationsActor</c> (T21); this host's
/// boot wiring subscribes; T19's engine-side handler consumes it. Re-exports the single Commons
/// const so every publisher/subscriber shares one literal.</summary>
public const string AlarmCommandsTopic = ZB.MOM.WW.OtOpcUa.Commons.OpcUa.AlarmCommandsTopic.Name;
/// <summary>Reconcile the loaded alarm set to exactly the enabled subset of <paramref name="Plans"/>:
/// builds <see cref="ScriptedAlarmDefinition"/>s (skipping disabled plans), reloads the engine, and
/// re-registers mux interest for the union of dependency refs.</summary>
/// <param name="Plans">The desired Equipment-namespace scripted-alarm plans.</param>
public sealed record ApplyScriptedAlarms(IReadOnlyList<EquipmentScriptedAlarmPlan> Plans);
/// <summary>Marshals an engine emission off the engine's worker thread onto the actor thread.
/// Carries the <see cref="ScriptedAlarmEvent"/> the engine raised on <c>OnEvent</c>.</summary>
private sealed record EngineEmission(ScriptedAlarmEvent Event);
/// <summary>Pipe-back completion of an in-flight <see cref="ScriptedAlarmEngine.LoadAsync"/>:
/// carries the union of dependency refs to register mux interest for AFTER the load completed
/// (so the engine's upstream subscriptions exist before any value can arrive). <see cref="Generation"/>
/// is the load generation captured when <see cref="OnApply"/> kicked the load off; a stale
/// continuation (an earlier generation completing after a newer apply) is discarded in
/// <see cref="OnAlarmsLoaded"/> so out-of-order completions can't register stale dep refs.</summary>
private sealed record AlarmsLoaded(IReadOnlyList<string> DepRefs, int Generation);
/// <summary>Pipe-back marker for a <see cref="ScriptedAlarmEngine.LoadAsync"/> that was cancelled
/// because the actor is stopping (its <see cref="_cts"/> fired in <see cref="PostStop"/>). It is a
/// no-op — there is nothing to register and no fault to log — and exists only so a cancelled load
/// pipes back a quiet message instead of a <see cref="Status.Failure"/> that would log a spurious
/// Warning on every clean shutdown.</summary>
private sealed record AlarmsLoadCanceled;
private readonly IActorRef _publishActor;
private readonly IActorRef? _mux;
private readonly DependencyMuxTagUpstreamSource _upstream;
private readonly ScriptedAlarmEngine _engine;
private readonly NodeId? _localNode;
private readonly ILoggingAdapter _log = Context.GetLogger();
private readonly CancellationTokenSource _cts = new();
private readonly EventHandler<ScriptedAlarmEvent> _onEngineEvent;
/// <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
/// cluster-wide <c>alerts</c> publish in <see cref="OnEngineEmission"/> resolves this through
/// <see cref="Drivers.PrimaryGatePolicy"/>: only the Primary publishes; an unknown role default-emits on a
/// single-driver cluster but default-DENIES on a multi-driver one, so the dual-primary boot window can't
/// historize duplicate AVEVA alarm rows (archreview 03/S4).</summary>
private RedundancyRole? _localRole;
/// <summary>Test seam (archreview 03/S4): overrides the count of Up <c>driver</c>-role cluster members the
/// alerts-emit gate reads while the role is unknown. Null ⇒ read the live cluster state.</summary>
private readonly Func<int>? _driverMemberCountProvider;
/// <summary>Optional node-local live-telemetry hub (Phase 5): each cluster-wide <c>alerts</c> transition
/// is also emitted here for the gRPC streaming service. Null (tests / no hub wired) skips the emit.</summary>
private readonly ITelemetryLocalHub? _telemetryHub;
/// <summary>Monotonic load generation, bumped on every <see cref="OnApply"/>. The continuation that
/// pipes back an <see cref="AlarmsLoaded"/> captures the generation it was started under; a stale
/// completion (an earlier generation arriving after a newer apply) is discarded in
/// <see cref="OnAlarmsLoaded"/> so out-of-order loads never register superseded dep refs.</summary>
private int _loadGeneration;
/// <summary>Cached cluster DistributedPubSub mediator, resolved once in <see cref="PreStart"/> (on the
/// actor thread) and reused for every emission instead of re-resolving it per-publish.</summary>
private IActorRef _mediator = null!;
/// <summary>Factory method to create Props for a <see cref="ScriptedAlarmHostActor"/>.</summary>
/// <param name="publishActor">The OPC UA publish actor that consumes
/// <see cref="OpcUaPublishActor.AlarmStateUpdate"/> bridged from engine emissions.</param>
/// <param name="mux">Optional dependency multiplexer the host registers interest with so it
/// receives a <see cref="VirtualTagActor.DependencyValueChanged"/> per dependency tag. Null on the
/// dev/Mac path (no live values).</param>
/// <param name="upstream">The mux-fed upstream the engine reads + subscribes from. MUST be the
/// same instance the <paramref name="engine"/> was constructed around.</param>
/// <param name="engine">The scripted-alarm engine this host owns + disposes.</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 publishes the cluster-wide <c>alerts</c>
/// transition. Null (the default) leaves the role unknown ⇒ default-emit (single-node deploys + tests).</param>
/// <param name="driverMemberCountProvider">Test seam (archreview 03/S4): overrides the count of Up
/// <c>driver</c>-role cluster members the alerts-emit gate reads while the role is unknown.</param>
/// <param name="telemetryHub">Optional node-local live-telemetry hub (Phase 5) each cluster-wide
/// <c>alerts</c> transition is also emitted into; null (tests) skips the emit.</param>
/// <returns>The <see cref="Akka.Actor.Props"/> used to instantiate the actor.</returns>
public static Props Props(
IActorRef publishActor,
IActorRef? mux,
DependencyMuxTagUpstreamSource upstream,
ScriptedAlarmEngine engine,
NodeId? localNode = null,
Func<int>? driverMemberCountProvider = null,
ITelemetryLocalHub? telemetryHub = null) =>
Akka.Actor.Props.Create(() => new ScriptedAlarmHostActor(publishActor, mux, upstream, engine, localNode, driverMemberCountProvider, telemetryHub));
/// <summary>Initializes a new instance of the <see cref="ScriptedAlarmHostActor"/> class.</summary>
/// <param name="publishActor">The OPC UA publish actor emissions are bridged to.</param>
/// <param name="mux">Optional dependency multiplexer the host registers dependency interest with.</param>
/// <param name="upstream">The mux-fed upstream the engine reads + subscribes from (same instance the engine wraps).</param>
/// <param name="engine">The scripted-alarm engine this host owns + disposes.</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 publishes the cluster-wide <c>alerts</c>
/// transition. Null leaves the role unknown ⇒ default-emit (single-node deploys + tests).</param>
/// <param name="driverMemberCountProvider">Test seam (archreview 03/S4): overrides the count of Up
/// <c>driver</c>-role cluster members the alerts-emit gate reads while the role is unknown. Null reads
/// the live cluster state (0 on a non-cluster ActorRefProvider ⇒ single-node default-emit).</param>
/// <param name="telemetryHub">Optional node-local live-telemetry hub (Phase 5) each cluster-wide
/// <c>alerts</c> transition is also emitted into; null (tests) skips the emit.</param>
public ScriptedAlarmHostActor(
IActorRef publishActor,
IActorRef? mux,
DependencyMuxTagUpstreamSource upstream,
ScriptedAlarmEngine engine,
NodeId? localNode = null,
Func<int>? driverMemberCountProvider = null,
ITelemetryLocalHub? telemetryHub = null)
{
ArgumentNullException.ThrowIfNull(publishActor);
ArgumentNullException.ThrowIfNull(upstream);
ArgumentNullException.ThrowIfNull(engine);
_publishActor = publishActor;
_mux = mux;
_upstream = upstream;
_engine = engine;
_localNode = localNode;
_driverMemberCountProvider = driverMemberCountProvider;
_telemetryHub = telemetryHub;
// OnEvent fires on the engine's worker thread. NEVER touch Context / actor state here —
// marshal onto the actor thread via the thread-safe Self.Tell. Keep the handler in a field
// so PostStop can unsubscribe it. (Self is captured once; it is stable for the actor's life.)
var self = Self;
_onEngineEvent = (_, e) => self.Tell(new EngineEmission(e));
_engine.OnEvent += _onEngineEvent;
Receive<ApplyScriptedAlarms>(OnApply);
Receive<AlarmsLoaded>(OnAlarmsLoaded);
Receive<VirtualTagActor.DependencyValueChanged>(OnDependencyChanged);
Receive<EngineEmission>(OnEngineEmission);
// Cluster redundancy snapshots (published on the `redundancy-state` topic, subscribed in PreStart)
// cache this node's role so OnEngineEmission can gate the cluster-wide alerts publish to the Primary.
Receive<RedundancyStateChanged>(OnRedundancyStateChanged);
// Inbound OPC UA Part 9 alarm method calls arrive as AlarmCommands on the cluster
// `alarm-commands` DPS topic (T18 publishes them after the AlarmAck role gate). The topic is a
// cluster-wide broadcast — every host node receives every command — so OnAlarmCommand filters to
// the alarms THIS host's engine owns before driving the matching engine op. The engine ops are
// async, and this project's Akka analyzer (AK2003) forbids an async-void Receive delegate, so
// the handler is a Task-returning ReceiveAsync: Akka suspends the mailbox until the op completes
// (ordered, awaited on the actor thread) and routes any escaped fault through supervision.
ReceiveAsync<AlarmCommand>(OnAlarmCommand);
// A faulted LoadAsync pipes back a Status.Failure (see OnApply) — log it and stay inert so the
// failure doesn't hit the dead-letter log.
Receive<Status.Failure>(OnLoadFailed);
// A LoadAsync cancelled by PostStop's _cts pipes back this marker. The actor is stopping, so
// there's nothing to do — swallow it quietly (no Warning, no dead letter).
Receive<AlarmsLoadCanceled>(_ => { });
// 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>(_ => { });
// DPS Subscribe (PreStart) acks back here once the mediator has registered Self on the topic.
// No-op — the subscription is live the moment the ack arrives; we only need to keep it off the
// dead-letter log. Matches OpcUaPublishActor / DriverHostActor's SubscribeAck convention.
Receive<SubscribeAck>(_ => { /* PubSub ack */ });
}
private void OnApply(ApplyScriptedAlarms msg)
{
// Bump the load generation and capture it into this apply's continuation. Rapid back-to-back
// applies kick off concurrent LoadAsyncs whose continuations may complete out of order; tagging
// each with its generation lets OnAlarmsLoaded discard any but the latest.
var gen = ++_loadGeneration;
// Skip disabled plans entirely — the engine has no Enabled flag, so a disabled alarm is simply
// not loaded (no predicate, no upstream subscription, no events).
var enabled = msg.Plans.Where(p => p.Enabled).ToList();
var defs = enabled.Select(ToDefinition).ToList();
// Union of dependency refs across the loaded (enabled) alarms — interest is registered with
// the mux only AFTER LoadAsync completes (see OnAlarmsLoaded), because the engine establishes
// its upstream SubscribeTag subscriptions inside LoadAsync; values must not arrive before then.
var depRefs = enabled
.SelectMany(p => p.DependencyRefs)
.Distinct(StringComparer.Ordinal)
.ToList();
// PipeTo marshals the completion back onto the actor thread. The continuation runs with
// CancellationToken.None so it ALWAYS executes (even after _cts fired on stop) and branches on
// the load's outcome: a clean cancel (actor stopping) pipes back a quiet AlarmsLoadCanceled; a
// genuine fault pipes back a Status.Failure we log as Warning (the prior generation's mux
// subscription stays — a failed reload should not silently drop live values); success pipes back
// AlarmsLoaded carrying the refs + the generation it was loaded under.
_engine.LoadAsync(defs, _cts.Token)
.ContinueWith(
t => t.IsCanceled ? (object)new AlarmsLoadCanceled()
: t.IsFaulted ? new Status.Failure(t.Exception!.GetBaseException())
: new AlarmsLoaded(depRefs, gen),
CancellationToken.None,
TaskContinuationOptions.ExecuteSynchronously,
TaskScheduler.Default)
.PipeTo(Self);
_log.Debug("ScriptedAlarmHost: applying (enabled={Enabled}/{Total}, depRefs={Refs})",
enabled.Count, msg.Plans.Count, depRefs.Count);
}
private void OnLoadFailed(Status.Failure msg)
=> _log.Warning(msg.Cause, "ScriptedAlarmHost: engine LoadAsync failed — alarms not (re)loaded");
private void OnAlarmsLoaded(AlarmsLoaded msg)
{
// Discard a stale completion: if a newer apply has since bumped the generation, this load's
// dep refs are superseded and registering them would re-introduce the old set. The latest
// generation's continuation will (or already did) register the current interest set.
if (msg.Generation != _loadGeneration)
{
return;
}
// Register mux interest only now that LoadAsync has returned — the engine's upstream
// subscriptions exist, so any DependencyValueChanged the mux delivers will be observed by the
// engine. RegisterInterest replaces a subscriber's prior interest set (see DependencyMuxActor),
// so a re-Apply with a fresh union simply supersedes the old one — no explicit unregister needed.
_mux?.Tell(new DependencyMuxActor.RegisterInterest(msg.DepRefs, Self));
_log.Debug("ScriptedAlarmHost: loaded; registered mux interest for {Count} dep refs", msg.DepRefs.Count);
ReassertConditionNodes();
}
/// <summary>
/// #487 — re-project every loaded alarm that is NOT in the Part 9 "no-event" position back onto
/// its condition node, right after a (re)load.
///
/// <para>
/// <b>The bug this closes.</b> A deploy that triggers a FULL address-space rebuild clears
/// <c>_alarmConditions</c> and re-materialises every condition node fresh — inactive, acked,
/// confirmed, unshelved. The engine, meanwhile, restores each alarm's persisted state and
/// re-derives Active from the predicate, so a still-active alarm ends the reload
/// <i>correctly Active in the engine</i> but produces <see cref="EmissionKind.None"/> (there
/// is no transition to report) — and <see cref="OnEngineEmission"/> drops None. Nothing then
/// writes the node, so an active alarm with static dependencies reads NORMAL to every OPC UA
/// client until its next real transition, which may never come. Same shape as the VirtualTag
/// redeploy-reset the <c>ReassertValue</c> fix closed.
/// </para>
///
/// <para>
/// <b>Node-only, by construction.</b> This sends <see cref="OpcUaPublishActor.AlarmStateUpdate"/>
/// and nothing else — it never publishes to the <c>alerts</c> topic and never touches the
/// telemetry hub. That is the whole reason it reads <see cref="ScriptedAlarmEngine.GetProjections"/>
/// (a plain read returning <see cref="ScriptedAlarmProjection"/>) rather than re-emitting: an
/// <c>alerts</c> row per alarm per deploy would append a duplicate historian / AVEVA record
/// every time anyone deploys — the alarm analogue of the VirtualTag M1 historian issue.
/// </para>
///
/// <para>
/// <b>Ordering.</b> <c>DriverHostActor</c> Tells <c>RebuildAddressSpace</c> to the publish
/// actor BEFORE it Tells <see cref="ApplyScriptedAlarms"/> here, and this runs later still —
/// after an <c>await</c>ed <see cref="ScriptedAlarmEngine.LoadAsync"/> pipes back. Both Tells
/// enqueue synchronously into the publish actor's local mailbox, so the re-materialise is
/// already queued ahead of these writes: the node exists by the time they are handled.
/// </para>
///
/// <para>
/// <b>Why only non-normal alarms.</b> An alarm sitting in the no-event position already
/// matches what the materialise just built, so writing it would be a pure no-op on state —
/// but it would still overwrite the node's Message (the materialise seeds it with the alarm's
/// display name; a projection carries the resolved template). Skipping them keeps a
/// never-fired alarm byte-identical to a deploy without this pass, and keeps the write count
/// at zero on the overwhelmingly common all-normal deploy.
/// </para>
///
/// <para>
/// <b>Duplicate events.</b> None. <c>WriteAlarmCondition</c>'s delta-gate compares the
/// incoming snapshot against the node's CURRENT state, so a re-assert onto a freshly
/// materialised (normal) node is a genuine delta and fires one Part 9 event — correct, since
/// the rebuild reset what clients see — while a re-assert onto a surgically-preserved node
/// that already holds the same state is suppressed. This write is ungated by redundancy role,
/// like every other node write here: the Secondary keeps its address space warm for failover.
/// </para>
/// </summary>
private void ReassertConditionNodes()
{
var reasserted = 0;
foreach (var projection in _engine.GetProjections())
{
if (IsInNoEventPosition(projection.Condition))
{
continue;
}
_publishActor.Tell(new OpcUaPublishActor.AlarmStateUpdate(
AlarmNodeId: projection.AlarmId,
State: ToSnapshot(projection),
// The instant the state we are re-asserting actually came about — NOT the deploy time.
// A client reading Time/ReceiveTime after a rebuild should still see when the alarm went
// active, not when the address space happened to be rebuilt.
TimestampUtc: projection.Condition.LastTransitionUtc,
Realm: AddressSpaceRealm.Uns));
reasserted++;
}
if (reasserted > 0)
{
_log.Info("ScriptedAlarmHost: re-asserted {Count} non-normal condition node(s) after (re)load", reasserted);
}
}
/// <summary>Whether <paramref name="condition"/> sits in the Part 9 "no-event" position — exactly the
/// state <see cref="AlarmConditionState.Fresh"/> produces and a freshly materialised condition node
/// already carries. Anything else (active, unacknowledged, unconfirmed, disabled, or shelved) is state
/// a rebuild would silently lose, so it is what <see cref="ReassertConditionNodes"/> re-projects.</summary>
/// <param name="condition">The engine's current condition state.</param>
/// <returns><c>true</c> when the condition carries nothing a rebuild could lose.</returns>
private static bool IsInNoEventPosition(AlarmConditionState condition)
=> condition.Enabled == AlarmEnabledState.Enabled
&& condition.Active == AlarmActiveState.Inactive
&& condition.Acked == AlarmAckedState.Acknowledged
&& condition.Confirmed == AlarmConfirmedState.Confirmed
&& condition.Shelving.Kind == ShelvingKind.Unshelved;
private void OnDependencyChanged(VirtualTagActor.DependencyValueChanged msg)
{
// Feed the live value into the upstream the engine subscribes from. #478 — carry the source
// driver's quality (mapped to an OPC UA StatusCode) so the engine can derive a scripted
// condition's worst-of-input quality; a Bad/Uncertain input is no longer silently treated as Good.
// #519 — the status code rides through verbatim. This used to convert the incoming value onto a
// 3-state enum and straight back again through a pair of mutually-inverse mappers, which is where
// the input's sub-code was lost before the engine ever cached it.
_upstream.Push(msg.TagId,
new DataValueSnapshot(msg.Value, msg.StatusCode, msg.TimestampUtc, msg.TimestampUtc));
}
private void OnEngineEmission(EngineEmission msg)
{
var e = msg.Event;
// None = no meaningful change; Suppressed = shelving ate the emission. Neither should reach a
// sink. (The engine already filters these out of BuildEmission, but guard defensively.)
if (e.Emission is EmissionKind.None or EmissionKind.Suppressed)
{
return;
}
// #478 — QualityChanged is a source-quality annotation (worst-of-input bucket moved with no Part 9
// state transition — e.g. an input went Bad, freezing the condition). Route it to the dedicated
// WriteAlarmQuality node path (sets ONLY Quality, one Part 9 event on a bucket change): NO full-state
// projection (would clobber severity/message) and NO /alerts row (it is not a state transition, so it
// must not historize). Same out-of-band quality path native comms-loss uses (#477 Layer 2).
if (e.Emission == EmissionKind.QualityChanged)
{
_publishActor.Tell(new OpcUaPublishActor.AlarmQualityUpdate(
AlarmNodeId: e.AlarmId,
StatusCode: e.WorstInputStatusCode,
TimestampUtc: e.TimestampUtc,
Realm: AddressSpaceRealm.Uns));
return;
}
// Bridge to OPC UA: project the FULL Part 9 condition state (enabled/active/acked/confirmed/
// shelving/severity/message) onto the materialised condition node via the Commons snapshot.
// e.AlarmId is the materialised condition's NodeId (T14 aligned it to the ScriptedAlarmId).
// Scripted alarms are per-equipment condition nodes in the UNS realm.
_publishActor.Tell(new OpcUaPublishActor.AlarmStateUpdate(
AlarmNodeId: e.AlarmId,
State: ToSnapshot(e),
TimestampUtc: e.TimestampUtc,
Realm: AddressSpaceRealm.Uns));
// Publish the transition to the cluster `alerts` topic — the single historization + live
// fan-out path. The mediator was cached on the ACTOR thread in PreStart; we only Tell it here.
var evt = new AlarmTransitionEvent(
AlarmId: e.AlarmId,
EquipmentPath: e.EquipmentPath,
AlarmName: e.AlarmName,
TransitionKind: e.Emission.ToString(),
Severity: SeverityToInt(e.Severity),
Message: e.Message,
User: TransitionUser(e),
TimestampUtc: e.TimestampUtc,
// Historian feed prep: carry the Part-9 subtype name (e.Kind.ToString() yields
// LimitAlarm/DiscreteAlarm/OffNormalAlarm/AlarmCondition) + any operator comment.
AlarmTypeName: e.Kind.ToString(),
Comment: e.Comment,
// Per-alarm DURABLE-historization opt-out — honored downstream by HistorianAdapterActor to
// suppress only the sink write. This publish (and the live `/alerts` fan-out) is NOT gated on it.
HistorizeToAveva: e.HistorizeToAveva);
// Warm-standby dedup: only the Primary publishes the cluster-wide transition (archreview 03/S4 —
// same PrimaryGatePolicy as DriverHostActor). A KNOWN Secondary/Detached is denied; an UNKNOWN role
// default-emits on a single-driver cluster but default-DENIES on a multi-driver one (so the
// dual-primary boot window can't historize duplicate AVEVA alarm rows). The OPC UA node write above +
// inbound command processing stay ungated (the secondary keeps its state warm for failover).
if (!PrimaryGatePolicy.ShouldServiceAsPrimary(_localRole, DriverMemberCount()))
{
OtOpcUaTelemetry.PrimaryGateDenied.Add(1,
new KeyValuePair<string, object?>("site", "alarm-emit"),
new KeyValuePair<string, object?>("reason", _localRole switch
{
RedundancyRole.Secondary => "secondary",
RedundancyRole.Detached => "detached",
_ => "role-unknown",
}));
return;
}
_mediator.Tell(new Publish(AlertsTopic, evt));
// Phase 5: fan the same transition into the node-local live-telemetry hub (no-op until a gRPC
// client subscribes). The DPS publish above is unchanged — a strictly additive tap, and it rides
// the same Primary gate as the publish (only the Primary reaches this line).
_telemetryHub?.Emit(new TelemetryItem.Alarm(evt));
}
/// <summary>Count of Up cluster members carrying the <c>driver</c> role, for the alerts-emit gate. Uses
/// the injected test seam when present; otherwise reads the live cluster state guarded by try/catch so a
/// non-cluster ActorRefProvider yields 0 ⇒ the single-node default-emit posture (archreview 03/S4).</summary>
/// <returns>The number of Up driver-role members, or 0 when the cluster extension is unavailable.</returns>
private int DriverMemberCount()
{
if (_driverMemberCountProvider is not null) return _driverMemberCountProvider();
try
{
return Akka.Cluster.Cluster.Get(Context.System).State.Members
.Count(m => m.Status == Akka.Cluster.MemberStatus.Up && m.Roles.Contains(ServiceCollectionExtensions.DriverRole));
}
catch (Exception)
{
return 0;
}
}
/// <summary>Caches this node's <see cref="RedundancyRole"/> from a cluster redundancy snapshot so
/// <see cref="OnEngineEmission"/> can gate the cluster-wide <c>alerts</c> publish to the Primary. A
/// snapshot that doesn't mention <see cref="_localNode"/> (or no local node wired) leaves the cached
/// role unchanged ⇒ default-emit. 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;
}
}
/// <summary>
/// Drives an inbound OPC UA Part 9 alarm method call (delivered as an <see cref="AlarmCommand"/>
/// on the cluster <c>alarm-commands</c> topic) onto the matching <see cref="ScriptedAlarmEngine"/>
/// operation.
///
/// <para>
/// <b>Ownership filter.</b> The topic is a cluster-wide broadcast; every host node receives
/// every command, but each owns a disjoint subset of alarms (its engine's loaded set). A
/// command for an alarm this engine does NOT own is a no-op — the owning node will act on it.
/// </para>
///
/// <para>
/// <b>No re-projection.</b> The engine op raises <see cref="ScriptedAlarmEngine.OnEvent"/> on
/// success, which already marshals back to <see cref="OnEngineEmission"/> and re-projects the
/// condition to the OPC UA node + the alerts topic. So this handler just calls the op and
/// awaits; it never touches the publish actor directly.
/// </para>
///
/// <para>
/// <b>Async on the actor thread.</b> The handler is a <c>Task</c>-returning
/// <c>ReceiveAsync</c> (this project's AK2003 analyzer forbids an async-void Receive
/// delegate). Akka suspends the actor's mailbox until the returned task completes, so the op
/// runs ordered + awaited on the actor thread — never overlapping the next message. The engine
/// also serialises every operation behind its own <c>_evalGate</c> and marshals every emission
/// back via <c>Self.Tell</c> (never touching <see cref="Context"/> off-thread). The whole body
/// is wrapped in a try/catch so a faulting op can never escape the handler and fault the actor
/// — failures are logged like <see cref="OnLoadFailed"/> and swallowed.
/// </para>
/// </summary>
/// <param name="cmd">The inbound alarm command.</param>
private async Task OnAlarmCommand(AlarmCommand cmd)
{
// Ownership filter FIRST: ignore commands for alarms this engine doesn't own. The topic is a
// cluster-wide broadcast, so the same command lands on every host — only the owner acts.
// Note: a concurrent LoadAsync can momentarily clear the loaded set, so a racing command may
// surface a spurious ArgumentException from the engine below — the outer catch absorbs it (benign).
if (!_engine.LoadedAlarmIds.Contains(cmd.AlarmId))
{
_log.Debug("ScriptedAlarmHost: ignoring AlarmCommand {Op} for unowned alarm {AlarmId}",
cmd.Operation, cmd.AlarmId);
return;
}
try
{
switch (cmd.Operation)
{
case "Acknowledge":
await _engine.AcknowledgeAsync(cmd.AlarmId, cmd.User, cmd.Comment, CancellationToken.None);
break;
case "Confirm":
await _engine.ConfirmAsync(cmd.AlarmId, cmd.User, cmd.Comment, CancellationToken.None);
break;
case "OneShotShelve":
await _engine.OneShotShelveAsync(cmd.AlarmId, cmd.User, CancellationToken.None);
break;
case "TimedShelve":
// A timed shelve needs the absolute unshelve instant. T18 derives it from the OPC UA
// Duration (UtcNow + shelvingTime); a command missing it is malformed — log + reject
// rather than throw (a fault escaping the async Task would trigger a supervision restart).
if (cmd.UnshelveAtUtc is not { } unshelveAt)
{
_log.Warning("ScriptedAlarmHost: rejecting TimedShelve for {AlarmId} — missing UnshelveAtUtc",
cmd.AlarmId);
return;
}
await _engine.TimedShelveAsync(cmd.AlarmId, cmd.User, unshelveAt, CancellationToken.None);
break;
case "Unshelve":
await _engine.UnshelveAsync(cmd.AlarmId, cmd.User, CancellationToken.None);
break;
case "Enable":
await _engine.EnableAsync(cmd.AlarmId, cmd.User, CancellationToken.None);
break;
case "Disable":
await _engine.DisableAsync(cmd.AlarmId, cmd.User, CancellationToken.None);
break;
case "AddComment":
if (string.IsNullOrWhiteSpace(cmd.Comment))
{
_log.Warning("ScriptedAlarmHost: rejecting AddComment for {AlarmId} — empty comment text", cmd.AlarmId);
return;
}
await _engine.AddCommentAsync(cmd.AlarmId, cmd.User, cmd.Comment, CancellationToken.None);
break;
default:
_log.Warning("ScriptedAlarmHost: ignoring AlarmCommand with unknown operation {Op} for {AlarmId}",
cmd.Operation, cmd.AlarmId);
break;
}
}
catch (Exception ex)
{
// A failing engine op must not crash the actor — mirror OnLoadFailed's log-and-stay-inert style.
_log.Warning(ex, "ScriptedAlarmHost: engine op {Op} failed for alarm {AlarmId}",
cmd.Operation, cmd.AlarmId);
}
}
/// <inheritdoc />
protected override void PreStart()
{
// Resolve the cluster DPS mediator once, on the actor thread, so emissions only Tell it.
_mediator = DistributedPubSub.Get(Context.System).Mediator;
// Subscribe to the `alarm-commands` topic so inbound OPC UA Part 9 method calls (published by
// the node manager's condition handlers, T18) land here as AlarmCommands. The Subscribe is sent
// from Self so the SubscribeAck returns to this actor (handled as a no-op in the ctor wiring).
_mediator.Tell(new Subscribe(AlarmCommandsTopic, Self));
// Also subscribe to the `redundancy-state` topic so cluster role changes land as
// RedundancyStateChanged and cache this node's role — OnEngineEmission gates the cluster-wide
// alerts publish to the Primary so a 2-node single-cluster deploy doesn't double-emit transitions.
_mediator.Tell(new Subscribe(OpcUaPublishActor.RedundancyStateTopic, Self));
// Per-cluster mesh Phase 2 dark switch. An AdminUI-originated ack/shelve arrives from
// NodeCommunicationActor on the node-local EventStream under
// MeshTransport:Mode = ClusterClient; the DPS subscribe above still carries the node-LOCAL
// publisher (OtOpcUaServerHostedService, the OPC UA Part 9 method calls), which never
// crosses the boundary and stays on DPS in both modes. Subscribing to both is safe: exactly
// one publisher feeds each path.
Context.System.EventStream.Subscribe(Self, typeof(AlarmCommand));
base.PreStart();
}
/// <inheritdoc />
protected override void PostStop()
{
// Unregister mux interest first so no further DependencyValueChanged arrives while we tear down.
_mux?.Tell(new DependencyMuxActor.UnregisterInterest(Self));
// Cancel any in-flight LoadAsync, detach the OnEvent handler (so a late engine emission can't
// Tell a stopping actor), then dispose the engine (which drains its background work + clears).
_cts.Cancel();
_engine.OnEvent -= _onEngineEvent;
_engine.Dispose();
_cts.Dispose();
base.PostStop();
}
/// <summary>Maps an <see cref="EquipmentScriptedAlarmPlan"/> to the engine's
/// <see cref="ScriptedAlarmDefinition"/>. <see cref="EquipmentScriptedAlarmPlan.AlarmType"/> is a
/// string that must parse to a defined <see cref="AlarmKind"/> NAME; an unrecognised type — including
/// a numeric string like "5" or "-1", which <see cref="Enum.TryParse{TEnum}(string, out TEnum)"/>
/// would otherwise accept as an undefined value — falls back to <see cref="AlarmKind.AlarmCondition"/>
/// rather than dropping the alarm.</summary>
private static ScriptedAlarmDefinition ToDefinition(EquipmentScriptedAlarmPlan p) => new(
AlarmId: p.ScriptedAlarmId,
EquipmentPath: p.EquipmentId,
AlarmName: p.Name,
Kind: Enum.TryParse<AlarmKind>(p.AlarmType, out var k) && Enum.IsDefined(k) ? k : AlarmKind.AlarmCondition,
Severity: SeverityFromInt(p.Severity),
MessageTemplate: p.MessageTemplate,
PredicateScriptSource: p.PredicateSource,
HistorizeToAveva: p.HistorizeToAveva,
Retain: p.Retain);
/// <summary>Maps a <see cref="ScriptedAlarmEvent"/>'s Core <see cref="AlarmConditionState"/> +
/// severity/message down to the Commons <see cref="AlarmConditionSnapshot"/> the SDK sink projects.
/// Severity is the OPC UA 1..1000 value <see cref="SeverityToInt"/> derives from the coarse engine
/// bucket, cast to the <c>ushort</c> the SDK <c>SetSeverity</c> expects. Shelving's 3-way Core kind
/// maps 1:1 onto the Commons <see cref="AlarmShelvingKind"/>.</summary>
private static AlarmConditionSnapshot ToSnapshot(ScriptedAlarmEvent e)
=> ToSnapshot(e.Condition, e.Severity, e.Message, e.WorstInputStatusCode);
/// <summary>#487 — the same projection for a <see cref="ScriptedAlarmProjection"/>: a state READ used
/// to restore a condition node after a rebuild, never an emission. Shares
/// <see cref="ToSnapshot(AlarmConditionState, AlarmSeverity, string, uint)"/> with the transition path
/// so a re-asserted node is byte-identical to what the alarm's own transition would have written.</summary>
/// <param name="p">The engine projection to map.</param>
/// <returns>The Commons snapshot the SDK sink projects onto the condition node.</returns>
private static AlarmConditionSnapshot ToSnapshot(ScriptedAlarmProjection p)
=> ToSnapshot(p.Condition, p.Severity, p.Message, p.WorstInputStatusCode);
/// <summary>The single Core → Commons condition projection, shared by the emission and re-assert
/// paths.</summary>
/// <param name="condition">The Part 9 condition state.</param>
/// <param name="severity">The engine's coarse severity bucket.</param>
/// <param name="message">The resolved condition message.</param>
/// <param name="worstInputStatusCode">The worst OPC UA StatusCode across the script's input tags.</param>
/// <returns>The Commons snapshot the SDK sink projects onto the condition node.</returns>
private static AlarmConditionSnapshot ToSnapshot(
AlarmConditionState condition, AlarmSeverity severity, string message, uint worstInputStatusCode) => new(
Active: condition.Active == AlarmActiveState.Active,
Acknowledged: condition.Acked == AlarmAckedState.Acknowledged,
Confirmed: condition.Confirmed == AlarmConfirmedState.Confirmed,
Enabled: condition.Enabled == AlarmEnabledState.Enabled,
Shelving: MapShelving(condition.Shelving.Kind),
Severity: (ushort)SeverityToInt(severity),
Message: message,
// #478 — the condition's status is the worst status across the script's input tags at evaluation
// time (carried on the event by the engine). A transition fired while an input is Uncertain projects
// Uncertain here so the full-snapshot write doesn't clobber quality back to Good.
StatusCode: worstInputStatusCode);
/// <summary>Maps the Core <see cref="ShelvingKind"/> onto the Commons <see cref="AlarmShelvingKind"/>
/// mirror (the Commons assembly can't see the Core enum).</summary>
private static AlarmShelvingKind MapShelving(ShelvingKind kind) => kind switch
{
ShelvingKind.OneShot => AlarmShelvingKind.OneShot,
ShelvingKind.Timed => AlarmShelvingKind.Timed,
_ => AlarmShelvingKind.Unshelved,
};
/// <summary>The acting user for an <see cref="AlarmTransitionEvent"/>. Engine-driven
/// Activated / Cleared transitions are <c>"system"</c>; operator Acknowledged / Confirmed carry the
/// recorded user from the condition state, falling back to <c>"system"</c> when none was recorded.
/// CommentAdded carries the commenter's identity from the last entry in
/// <see cref="AlarmConditionState.Comments"/>.</summary>
private static string TransitionUser(ScriptedAlarmEvent e) => e.Emission switch
{
EmissionKind.Acknowledged => e.Condition.LastAckUser ?? "system",
EmissionKind.Confirmed => e.Condition.LastConfirmUser ?? "system",
// Shelve / unshelve / enable / disable / comment ops each append the acting user as the LAST audit
// entry on the emitted condition (engine auto-unshelve appends "system"); read it from there.
EmissionKind.CommentAdded
or EmissionKind.Shelved
or EmissionKind.Unshelved
or EmissionKind.Enabled
or EmissionKind.Disabled => e.Condition.Comments.Count > 0 ? e.Condition.Comments[^1].User : "system",
_ => "system",
};
// Severity conversion convention: the engine + plan disagree on type (engine = AlarmSeverity enum,
// plan + AlarmTransitionEvent = OPC UA 1–1000 int). We bucket the int into quartiles on the way IN
// and emit the quartile ceiling on the way OUT, so a round-trip is stable within a bucket.
/// <summary>Buckets an OPC UA 1–1000 severity into the engine's coarse <see cref="AlarmSeverity"/>
/// enum: ≤250 Low, ≤500 Medium, ≤750 High, else Critical.</summary>
private static AlarmSeverity SeverityFromInt(int s) =>
s <= 250 ? AlarmSeverity.Low
: s <= 500 ? AlarmSeverity.Medium
: s <= 750 ? AlarmSeverity.High
: AlarmSeverity.Critical;
/// <summary>Maps the engine's coarse <see cref="AlarmSeverity"/> back to an OPC UA 1–1000 severity
/// at each bucket's ceiling: Low=250, Medium=500, High=750, Critical=1000.</summary>
private static int SeverityToInt(AlarmSeverity s) => s switch
{
AlarmSeverity.Low => 250,
AlarmSeverity.Medium => 500,
AlarmSeverity.High => 750,
AlarmSeverity.Critical => 1000,
_ => 500,
};
}