Joseph Doherty
1143 lines
69 KiB
C#
1143 lines
69 KiB
C#
using Akka.Actor;
|
|
using Akka.Cluster;
|
|
using Akka.Cluster.Tools.PublishSubscribe;
|
|
using Akka.Event;
|
|
using Microsoft.EntityFrameworkCore;
|
|
using ZB.MOM.WW.OtOpcUa.Cluster.Redundancy;
|
|
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.Configuration;
|
|
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Authorization;
|
|
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.Health;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
|
|
|
|
/// <summary>
|
|
/// Single-threaded bridge between Akka messages and the OPC UA SDK address space. Hosted on
|
|
/// the pinned <c>opcua-synchronized-dispatcher</c> (HOCON) so the OPC UA SDK sees
|
|
/// only one thread per actor instance — its session/subscription locks expect strict
|
|
/// single-threaded access.
|
|
///
|
|
/// Address-space writes route through <see cref="IOpcUaAddressSpaceSink"/>; ServiceLevel
|
|
/// writes route through <see cref="IServiceLevelPublisher"/>. Production binds SDK-backed
|
|
/// implementations; dev/Mac/tests bind the Null* defaults so the actor stays decoupled from
|
|
/// <c>Opc.Ua.Server</c>. The remaining piece is wiring those bindings to a real
|
|
/// <c>StandardServer</c> address space — tracked as F10b.
|
|
/// </summary>
|
|
public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
|
|
{
|
|
public const string DispatcherId = "opcua-synchronized-dispatcher";
|
|
public const string RedundancyStateTopic = "redundancy-state";
|
|
|
|
/// <summary>Gets or sets the timer scheduler for the periodic DB-health refresh tick.</summary>
|
|
public ITimerScheduler Timers { get; set; } = null!;
|
|
|
|
/// <summary>Self-tick that drives the periodic Ask of the local <see cref="DbHealthProbeActor"/>
|
|
/// for its cached status. Private singleton — the actor pumps this for itself, no external sender.</summary>
|
|
private sealed class HealthTick { public static readonly HealthTick Instance = new(); private HealthTick() { } }
|
|
|
|
/// <summary>A driver/vtag value update for a single materialised Variable node. v3 Batch 4: carries the
|
|
/// node's <see cref="AddressSpaceRealm"/> so the sink resolves the right namespace — the driver fan-out
|
|
/// posts one of these per registered NodeId (the raw node in <see cref="AddressSpaceRealm.Raw"/> and each
|
|
/// referencing UNS node in <see cref="AddressSpaceRealm.Uns"/>) with identical value/status/timestamp.
|
|
/// <see cref="Realm"/> defaults to <see cref="AddressSpaceRealm.Uns"/> so pre-v3 callers (VirtualTag
|
|
/// equipment nodes) keep compiling; the driver fan-out passes it EXPLICITLY per node.
|
|
/// <para><see cref="StatusCode"/> is the full OPC UA status code, sub-code included (#519); the sink
|
|
/// hands it to the SDK verbatim.</para></summary>
|
|
public sealed record AttributeValueUpdate(string NodeId, object? Value, uint StatusCode, DateTime TimestampUtc, AddressSpaceRealm Realm);
|
|
|
|
/// <summary>Batched form of <see cref="AttributeValueUpdate"/>: every fan-out write of one driver
|
|
/// production burst in one message — the sink takes its lock once for the whole batch. Items carry
|
|
/// their realm each (<see cref="ValueWrite"/>); status codes ride verbatim (#519).</summary>
|
|
public sealed record AttributeValueBatchUpdate(IReadOnlyList<ValueWrite> Items);
|
|
|
|
/// <summary>Status-only batch: every named node keeps its CURRENT value and only its status +
|
|
/// timestamp change (archreview 03/S15). The comms-loss fan-out sends one of these covering every
|
|
/// NodeId a disconnected driver owns — the raw node and each referencing UNS node — so a client sees
|
|
/// <see cref="OpcUaStatus.BadNoCommunication"/> over the last known value instead of a stale Good.
|
|
/// <para>Deliberately a separate message from <see cref="AttributeValueBatchUpdate"/> rather than a
|
|
/// flag on it: the host holds no value cache (v3's single-source design), so it has nothing to put in
|
|
/// a <see cref="ValueWrite"/>, and only the node knows what value to retain.</para></summary>
|
|
public sealed record AttributeStatusBatchUpdate(IReadOnlyList<StatusWrite> Items);
|
|
|
|
/// <summary>Carries the full Part 9 condition state for a scripted alarm to the sink. The
|
|
/// <paramref name="State"/> snapshot is the Commons projection the Runtime host maps from the engine's
|
|
/// Core <c>AlarmConditionState</c> + severity/message — the actor stays decoupled from
|
|
/// <c>Core.ScriptedAlarms</c>.</summary>
|
|
/// <param name="AlarmNodeId">The alarm node id (== ScriptedAlarmId for scripted conditions; == RawPath for
|
|
/// v3 native raw conditions).</param>
|
|
/// <param name="State">The full condition state to project onto the node.</param>
|
|
/// <param name="TimestampUtc">The source timestamp of the transition in UTC.</param>
|
|
/// <param name="Realm">The namespace realm the condition lives in — <see cref="AddressSpaceRealm.Uns"/> for
|
|
/// scripted alarms (default), <see cref="AddressSpaceRealm.Raw"/> for v3 native raw conditions.</param>
|
|
public sealed record AlarmStateUpdate(string AlarmNodeId, AlarmConditionSnapshot State, DateTime TimestampUtc, AddressSpaceRealm Realm);
|
|
/// <summary>#477 — annotate a materialised condition's source-data Quality out of band from any alarm
|
|
/// transition (the driver-connectivity path: comms lost → <see cref="OpcUaStatus.BadNoCommunication"/>,
|
|
/// restored → <see cref="OpcUaStatus.Good"/>). Routed to <see cref="IOpcUaAddressSpaceSink.WriteAlarmQuality"/>,
|
|
/// which sets ONLY Quality and fires one Part 9 event on a status-code change.</summary>
|
|
/// <param name="AlarmNodeId">The condition node id (RawPath for a native alarm).</param>
|
|
/// <param name="StatusCode">The source-data status code to annotate, sub-code included.</param>
|
|
/// <param name="TimestampUtc">The connectivity transition timestamp in UTC.</param>
|
|
/// <param name="Realm">The namespace realm the condition lives in (<see cref="AddressSpaceRealm.Raw"/> for native).</param>
|
|
public sealed record AlarmQualityUpdate(string AlarmNodeId, uint StatusCode, DateTime TimestampUtc, AddressSpaceRealm Realm);
|
|
/// <summary>
|
|
/// Triggers an address-space rebuild. <paramref name="DeploymentId"/> is the deployment
|
|
/// just applied by the host; the rebuild loads THAT artifact so materialisation matches the
|
|
/// applied config + the SubscribeBulk pass. It is null only for legacy/dev callers, which
|
|
/// fall back to the latest sealed deployment (lags a not-yet-sealed apply by one revision).
|
|
/// </summary>
|
|
/// <param name="Correlation">Correlation id for tracing this rebuild.</param>
|
|
/// <param name="DeploymentId">The applied deployment whose artifact to materialise, or null for the latest-sealed fallback.</param>
|
|
/// <param name="Artifact">
|
|
/// The artifact bytes already in hand, used INSTEAD of loading them from the ConfigDb. This is
|
|
/// what makes boot-from-cache actually serve its address space: on a central-SQL outage the
|
|
/// host has the cached blob but <see cref="DeploymentId"/> alone would drive a ConfigDb read
|
|
/// that cannot succeed, leaving the rebuild a no-op and clients browsing an empty server. Null
|
|
/// on the normal path, where loading from the ConfigDb by id is correct.
|
|
/// </param>
|
|
/// <param name="ReportOutcome">
|
|
/// Gitea #601 — when <see langword="true"/>, the actor replies to the sender with a
|
|
/// <see cref="RebuildCompleted"/> once the apply outcome is known (clean or degraded), on EVERY path
|
|
/// through the handler. <c>DriverHostActor</c> sets it on the deploy path and sends the deployment's
|
|
/// ack only after that reply, so a degraded apply can no longer seal a deployment green. Off for the
|
|
/// boot/restore paths, which re-ack nothing.
|
|
/// </param>
|
|
public sealed record RebuildAddressSpace(
|
|
CorrelationId Correlation, DeploymentId? DeploymentId = null, byte[]? Artifact = null,
|
|
bool ReportOutcome = false);
|
|
|
|
/// <summary>
|
|
/// Gitea #601 — the reply to a <see cref="RebuildAddressSpace"/> carrying
|
|
/// <see cref="RebuildAddressSpace.ReportOutcome"/>: whether the address-space apply was clean.
|
|
/// </summary>
|
|
/// <param name="Correlation">The rebuild's correlation id, echoed.</param>
|
|
/// <param name="DeploymentId">The rebuild's deployment id, echoed (null for a legacy caller).</param>
|
|
/// <param name="Degraded">
|
|
/// <see langword="true"/> when the apply swallowed per-node sink faults, its rebuild failed, the
|
|
/// pipeline threw, or there was no usable artifact — i.e. the served address space may be partial or
|
|
/// stale. An empty plan (nothing changed) is clean.
|
|
/// </param>
|
|
/// <param name="DegradedReason">A short operator-readable reason; null when clean.</param>
|
|
public sealed record RebuildCompleted(
|
|
CorrelationId Correlation, DeploymentId? DeploymentId, bool Degraded, string? DegradedReason);
|
|
|
|
public sealed record ServiceLevelChanged(byte ServiceLevel);
|
|
|
|
private readonly IOpcUaAddressSpaceSink _sink;
|
|
private readonly IServiceLevelPublisher _serviceLevel;
|
|
private readonly bool _subscribeRedundancyTopic;
|
|
private readonly NodeId? _localNode;
|
|
private readonly IDbContextFactory<OtOpcUaConfigDbContext>? _dbFactory;
|
|
private readonly AddressSpaceApplier? _applier;
|
|
/// <summary>#520 stage 4 — receives the API keys + ACL grants carried by each applied artifact, so
|
|
/// OPC UA session activation can authenticate. Null when nothing consumes it (tests, and any node
|
|
/// that is not serving an OPC UA data plane).</summary>
|
|
private readonly IDataPlaneAuthoritySink? _dataPlaneAuthority;
|
|
private readonly IActorRef? _dbHealthProbe;
|
|
/// <summary>Per-cluster mesh Phase 4: this is a driver-only node with NO ConfigDb (LocalDb is its
|
|
/// config store). There is no DB to probe, so the ServiceLevel DB-reachability input is retired —
|
|
/// <c>DbReachable</c> is fed constant <c>true</c> and only snapshot staleness can demote. See
|
|
/// <see cref="RecomputeServiceLevel"/>.</summary>
|
|
private readonly bool _dbLess;
|
|
private readonly TimeSpan _staleWindow;
|
|
private readonly TimeSpan _probeFreshnessWindow;
|
|
private readonly TimeSpan _healthTickInterval;
|
|
private readonly Akka.Cluster.Cluster _cluster = Akka.Cluster.Cluster.Get(Context.System);
|
|
private readonly ILoggingAdapter _log = Context.GetLogger();
|
|
|
|
private int _writes;
|
|
private byte _lastServiceLevel;
|
|
private bool _publishedAtLeastOnce;
|
|
/// <summary>True once a real composition has been applied, i.e. there is a served address space worth
|
|
/// protecting. Distinguishes "the artifact went missing under a running server" (issue #485 — warn, and
|
|
/// hold what is materialised) from "nothing has been deployed here yet" (a quiet no-op at boot).</summary>
|
|
private bool _hasAppliedComposition;
|
|
/// <summary>
|
|
/// Gitea #601 review — <see cref="_lastApplied"/> was advanced by a DEGRADED rebuild (a rebuild advances
|
|
/// even with swallowed failures, see <see cref="HandleRebuildCore"/>). A re-deploy of the same revision
|
|
/// then diffs to an EMPTY plan; without this flag that empty plan reported clean, so the "re-deploy
|
|
/// re-applies" repair path sealed green over an address space still known to be partial. While set, an
|
|
/// empty plan re-materialises the whole composition (idempotent ensures) and reports clean only when
|
|
/// that pass is itself clean.
|
|
/// </summary>
|
|
private bool _lastAppliedDegraded;
|
|
private DbHealthProbeActor.DbHealthStatus? _lastDbHealth;
|
|
private RedundancyStateChanged? _lastSnapshot;
|
|
private (bool Ok, DateTime At)? _probeAboutMe;
|
|
private AddressSpaceComposition _lastApplied = new(
|
|
Array.Empty<UnsAreaProjection>(),
|
|
Array.Empty<UnsLineProjection>(),
|
|
Array.Empty<EquipmentNode>(),
|
|
Array.Empty<DriverInstancePlan>(),
|
|
Array.Empty<ScriptedAlarmPlan>());
|
|
|
|
/// <summary>Gets the number of writes performed.</summary>
|
|
public int WriteCount => _writes;
|
|
/// <summary>Gets the last published service level.</summary>
|
|
public byte LastServiceLevel => _lastServiceLevel;
|
|
|
|
/// <summary>Production Props — pins the OPC UA dispatcher + subscribes to the
|
|
/// <c>redundancy-state</c> DPS topic so cluster transitions drive the local ServiceLevel
|
|
/// publish path. When <paramref name="dbFactory"/> + <paramref name="applier"/> are supplied,
|
|
/// <see cref="RebuildAddressSpace"/> reads the latest deployment artifact + drives the
|
|
/// applier through the sink. When <paramref name="dbHealthProbe"/> is supplied the local
|
|
/// ServiceLevel is computed via <see cref="ServiceLevelCalculator"/> from real DB-health +
|
|
/// staleness + role-leader inputs; otherwise the legacy role-only switch is used.</summary>
|
|
/// <param name="sink">The OPC UA address space sink.</param>
|
|
/// <param name="serviceLevel">The service level publisher.</param>
|
|
/// <param name="localNode">The local cluster node ID.</param>
|
|
/// <param name="dbFactory">The optional database context factory.</param>
|
|
/// <param name="applier">The optional Phase 7 applier.</param>
|
|
/// <param name="dbHealthProbe">The optional <see cref="DbHealthProbeActor"/> ref; when null the
|
|
/// legacy role-only ServiceLevel seam is used until a <see cref="DbHealthProbeActor.DbHealthStatus"/> arrives.</param>
|
|
/// <param name="staleWindow">The window beyond which a DB-health sample or redundancy snapshot is
|
|
/// considered stale; defaults to 30 seconds.</param>
|
|
/// <param name="probeFreshnessWindow">The window beyond which a peer's OPC UA probe verdict about
|
|
/// this node is considered stale (and thus given the benefit of the doubt rather than demoting);
|
|
/// defaults to 30 seconds.</param>
|
|
/// <param name="healthTickInterval">The period between self-driven DB-health refresh ticks (each
|
|
/// Asks <paramref name="dbHealthProbe"/> for its cached status); defaults to 5 seconds. No timer is
|
|
/// started when <paramref name="dbHealthProbe"/> is null.</param>
|
|
/// <param name="dbLess">Per-cluster mesh Phase 4: this is a driver-only node with no ConfigDb, so the
|
|
/// ServiceLevel DB-reachability input is retired (<c>DbReachable</c> treated constant-true, only
|
|
/// staleness demotes). No <paramref name="dbHealthProbe"/> is wired on such a node.</param>
|
|
/// <returns>The configured <see cref="Props"/> for creating this actor, pinned to the
|
|
/// <see cref="DispatcherId"/> dispatcher.</returns>
|
|
public static Props Props(
|
|
IOpcUaAddressSpaceSink? sink = null,
|
|
IServiceLevelPublisher? serviceLevel = null,
|
|
NodeId? localNode = null,
|
|
IDbContextFactory<OtOpcUaConfigDbContext>? dbFactory = null,
|
|
AddressSpaceApplier? applier = null,
|
|
IActorRef? dbHealthProbe = null,
|
|
TimeSpan? staleWindow = null,
|
|
TimeSpan? probeFreshnessWindow = null,
|
|
TimeSpan? healthTickInterval = null,
|
|
bool dbLess = false,
|
|
IDataPlaneAuthoritySink? dataPlaneAuthority = null) =>
|
|
Akka.Actor.Props.Create(() => new OpcUaPublishActor(
|
|
sink ?? NullOpcUaAddressSpaceSink.Instance,
|
|
serviceLevel ?? NullServiceLevelPublisher.Instance,
|
|
subscribeRedundancyTopic: true,
|
|
localNode,
|
|
dbFactory,
|
|
applier,
|
|
dbHealthProbe,
|
|
staleWindow,
|
|
probeFreshnessWindow,
|
|
healthTickInterval,
|
|
dbLess,
|
|
dataPlaneAuthority)).WithDispatcher(DispatcherId);
|
|
|
|
/// <summary>Test-only Props that omits the pinned-dispatcher requirement and skips the
|
|
/// DPS subscribe so unit tests can spin up the actor on a vanilla TestKit cluster.</summary>
|
|
/// <param name="sink">The OPC UA address space sink.</param>
|
|
/// <param name="serviceLevel">The service level publisher.</param>
|
|
/// <param name="subscribeRedundancyTopic">Whether to subscribe to the redundancy topic.</param>
|
|
/// <param name="localNode">The local cluster node ID.</param>
|
|
/// <param name="dbFactory">The optional database context factory.</param>
|
|
/// <param name="applier">The optional Phase 7 applier.</param>
|
|
/// <param name="dbHealthProbe">The optional <see cref="DbHealthProbeActor"/> ref; when null the
|
|
/// legacy role-only ServiceLevel seam is used until a <see cref="DbHealthProbeActor.DbHealthStatus"/> arrives.</param>
|
|
/// <param name="staleWindow">The window beyond which a DB-health sample or redundancy snapshot is
|
|
/// considered stale; defaults to 30 seconds.</param>
|
|
/// <param name="probeFreshnessWindow">The window beyond which a peer's OPC UA probe verdict about
|
|
/// this node is considered stale (and thus given the benefit of the doubt rather than demoting);
|
|
/// defaults to 30 seconds.</param>
|
|
/// <param name="healthTickInterval">The period between self-driven DB-health refresh ticks (each
|
|
/// Asks <paramref name="dbHealthProbe"/> for its cached status); defaults to 5 seconds. No timer is
|
|
/// started when <paramref name="dbHealthProbe"/> is null.</param>
|
|
/// <param name="dbLess">Per-cluster mesh Phase 4: this is a driver-only node with no ConfigDb, so the
|
|
/// ServiceLevel DB-reachability input is retired (<c>DbReachable</c> treated constant-true, only
|
|
/// staleness demotes). No <paramref name="dbHealthProbe"/> is wired on such a node.</param>
|
|
/// <returns>The configured <see cref="Props"/> for creating this actor in tests, without dispatcher
|
|
/// pinning or the DPS subscribe.</returns>
|
|
public static Props PropsForTests(
|
|
IOpcUaAddressSpaceSink? sink = null,
|
|
IServiceLevelPublisher? serviceLevel = null,
|
|
bool subscribeRedundancyTopic = false,
|
|
NodeId? localNode = null,
|
|
IDbContextFactory<OtOpcUaConfigDbContext>? dbFactory = null,
|
|
AddressSpaceApplier? applier = null,
|
|
IActorRef? dbHealthProbe = null,
|
|
TimeSpan? staleWindow = null,
|
|
TimeSpan? probeFreshnessWindow = null,
|
|
TimeSpan? healthTickInterval = null,
|
|
bool dbLess = false,
|
|
IDataPlaneAuthoritySink? dataPlaneAuthority = null) =>
|
|
Akka.Actor.Props.Create(() => new OpcUaPublishActor(
|
|
sink ?? NullOpcUaAddressSpaceSink.Instance,
|
|
serviceLevel ?? NullServiceLevelPublisher.Instance,
|
|
subscribeRedundancyTopic,
|
|
localNode,
|
|
dbFactory,
|
|
applier,
|
|
dbHealthProbe,
|
|
staleWindow,
|
|
probeFreshnessWindow,
|
|
healthTickInterval,
|
|
dbLess,
|
|
dataPlaneAuthority));
|
|
|
|
/// <summary>Initializes a new instance of the <see cref="OpcUaPublishActor"/> class.</summary>
|
|
/// <param name="sink">The OPC UA address space sink.</param>
|
|
/// <param name="serviceLevel">The service level publisher.</param>
|
|
/// <param name="subscribeRedundancyTopic">Whether to subscribe to the redundancy topic.</param>
|
|
/// <param name="localNode">The local cluster node ID.</param>
|
|
/// <param name="dbFactory">The optional database context factory.</param>
|
|
/// <param name="applier">The optional Phase 7 applier.</param>
|
|
/// <param name="dbHealthProbe">The optional <see cref="DbHealthProbeActor"/> ref; when null the
|
|
/// legacy role-only ServiceLevel seam is used until a <see cref="DbHealthProbeActor.DbHealthStatus"/> arrives.</param>
|
|
/// <param name="staleWindow">The window beyond which a DB-health sample or redundancy snapshot is
|
|
/// considered stale; defaults to 30 seconds.</param>
|
|
/// <param name="probeFreshnessWindow">The window beyond which a peer's OPC UA probe verdict about
|
|
/// this node is considered stale (and thus given the benefit of the doubt rather than demoting);
|
|
/// defaults to 30 seconds.</param>
|
|
/// <param name="healthTickInterval">The period between self-driven DB-health refresh ticks (each
|
|
/// Asks <paramref name="dbHealthProbe"/> for its cached status); defaults to 5 seconds. No timer is
|
|
/// started when <paramref name="dbHealthProbe"/> is null.</param>
|
|
/// <param name="dbLess">Per-cluster mesh Phase 4: this is a driver-only node with no ConfigDb, so the
|
|
/// ServiceLevel DB-reachability input is retired (<c>DbReachable</c> treated constant-true, only
|
|
/// staleness demotes). No <paramref name="dbHealthProbe"/> is wired on such a node.</param>
|
|
public OpcUaPublishActor(
|
|
IOpcUaAddressSpaceSink sink,
|
|
IServiceLevelPublisher serviceLevel,
|
|
bool subscribeRedundancyTopic,
|
|
NodeId? localNode,
|
|
IDbContextFactory<OtOpcUaConfigDbContext>? dbFactory = null,
|
|
AddressSpaceApplier? applier = null,
|
|
IActorRef? dbHealthProbe = null,
|
|
TimeSpan? staleWindow = null,
|
|
TimeSpan? probeFreshnessWindow = null,
|
|
TimeSpan? healthTickInterval = null,
|
|
bool dbLess = false,
|
|
IDataPlaneAuthoritySink? dataPlaneAuthority = null)
|
|
{
|
|
_sink = sink;
|
|
_dataPlaneAuthority = dataPlaneAuthority;
|
|
_serviceLevel = serviceLevel;
|
|
_subscribeRedundancyTopic = subscribeRedundancyTopic;
|
|
_localNode = localNode;
|
|
_dbFactory = dbFactory;
|
|
_applier = applier;
|
|
_dbHealthProbe = dbHealthProbe;
|
|
_dbLess = dbLess;
|
|
_staleWindow = staleWindow ?? TimeSpan.FromSeconds(30);
|
|
_probeFreshnessWindow = probeFreshnessWindow ?? TimeSpan.FromSeconds(30);
|
|
_healthTickInterval = healthTickInterval ?? TimeSpan.FromSeconds(5);
|
|
|
|
Receive<AttributeValueUpdate>(HandleAttributeUpdate);
|
|
Receive<AttributeValueBatchUpdate>(HandleAttributeBatchUpdate);
|
|
Receive<AttributeStatusBatchUpdate>(HandleAttributeStatusBatchUpdate);
|
|
Receive<AlarmStateUpdate>(HandleAlarmUpdate);
|
|
Receive<AlarmQualityUpdate>(HandleAlarmQualityUpdate);
|
|
Receive<RebuildAddressSpace>(HandleRebuild);
|
|
Receive<ServiceLevelChanged>(HandleServiceLevelChanged);
|
|
Receive<RedundancyStateChanged>(HandleRedundancyStateChanged);
|
|
Receive<DbHealthProbeActor.DbHealthStatus>(HandleDbHealthStatus);
|
|
Receive<HealthTick>(_ => OnHealthTick());
|
|
Receive<PeerOpcUaProbeActor.OpcUaProbeResult>(HandlePeerProbe);
|
|
Receive<SubscribeAck>(_ => { /* PubSub ack */ });
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
protected override void PreStart()
|
|
{
|
|
if (_subscribeRedundancyTopic)
|
|
{
|
|
DistributedPubSub.Get(Context.System).Mediator.Tell(new Subscribe(RedundancyStateTopic, Self));
|
|
}
|
|
|
|
// Production delivery of DB-health: when a probe is wired, kick an immediate refresh + start the
|
|
// periodic tick so _lastDbHealth gets populated and kept fresh without any external pump. Gated
|
|
// ONLY on the probe being present (independent of the DPS subscribe) so the calculator path works
|
|
// in tests too. No probe → stay on the legacy role-only seam (no timer).
|
|
if (_dbHealthProbe is not null)
|
|
{
|
|
Self.Tell(HealthTick.Instance); // immediate first refresh
|
|
Timers.StartPeriodicTimer("db-health", HealthTick.Instance, _healthTickInterval);
|
|
}
|
|
}
|
|
|
|
private void HandleAttributeUpdate(AttributeValueUpdate msg)
|
|
{
|
|
try
|
|
{
|
|
_sink.WriteValue(msg.NodeId, msg.Value, msg.StatusCode, msg.TimestampUtc, msg.Realm);
|
|
Interlocked.Increment(ref _writes);
|
|
OtOpcUaTelemetry.OpcUaSinkWrite.Add(1, new KeyValuePair<string, object?>("kind", "value"));
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_log.Warning(ex, "OpcUaPublish: sink.WriteValue threw for {Node}", msg.NodeId);
|
|
}
|
|
}
|
|
|
|
/// <summary>Mirrors <see cref="HandleAttributeUpdate"/> for a whole burst: ONE
|
|
/// <see cref="IOpcUaAddressSpaceSink.WriteValues"/> call, but the write tally and the
|
|
/// <c>otopcua.opcua.sink.write</c> meter still advance PER ITEM — batching is a transport change, so
|
|
/// dashboards must not drop by the batch factor.
|
|
/// <para>A <see cref="ValueWriteBatchException"/> is a PARTIAL failure: the sink applied every item it
|
|
/// could and is telling us which one it couldn't. So it is caught ahead of the general catch to (a) name
|
|
/// the offending node in the log — a bare "the batch threw" cannot be acted on — and (b) still count the
|
|
/// items that DID land, keeping the per-item counting invariant honest instead of silently under-reporting
|
|
/// a whole batch because one item was bad.</para></summary>
|
|
private void HandleAttributeBatchUpdate(AttributeValueBatchUpdate msg)
|
|
{
|
|
try
|
|
{
|
|
_sink.WriteValues(msg.Items);
|
|
Interlocked.Add(ref _writes, msg.Items.Count);
|
|
OtOpcUaTelemetry.OpcUaSinkWrite.Add(msg.Items.Count, new KeyValuePair<string, object?>("kind", "value"));
|
|
}
|
|
catch (ValueWriteBatchException ex)
|
|
{
|
|
_log.Warning(ex,
|
|
"OpcUaPublish: sink.WriteValues partially failed for a {Count}-item batch — {Failed} failed " +
|
|
"(first: {Node}), {Succeeded} applied",
|
|
msg.Items.Count, ex.FailedCount, ex.FirstFailedNodeId, ex.SucceededCount);
|
|
Interlocked.Add(ref _writes, ex.SucceededCount);
|
|
OtOpcUaTelemetry.OpcUaSinkWrite.Add(ex.SucceededCount, new KeyValuePair<string, object?>("kind", "value"));
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_log.Warning(ex, "OpcUaPublish: sink.WriteValues threw for a {Count}-item batch", msg.Items.Count);
|
|
}
|
|
}
|
|
|
|
/// <summary>Routes a status-only batch to <see cref="IOpcUaAddressSpaceSink.WriteStatuses"/> — the
|
|
/// comms-loss annotation (03/S15). Same shape as <see cref="HandleAttributeBatchUpdate"/>: metered per
|
|
/// ITEM (batching is a transport change, so dashboards must not drop by the batch factor), a
|
|
/// <see cref="ValueWriteBatchException"/> is a partial failure that still counts what landed and names
|
|
/// the offender, and any other throw is swallowed so a sink fault cannot restart the publish actor and
|
|
/// drop live traffic. Metered under <c>kind:"status"</c> so a status annotation is separable from a
|
|
/// value write on the same counter.</summary>
|
|
private void HandleAttributeStatusBatchUpdate(AttributeStatusBatchUpdate msg)
|
|
{
|
|
try
|
|
{
|
|
_sink.WriteStatuses(msg.Items);
|
|
Interlocked.Add(ref _writes, msg.Items.Count);
|
|
OtOpcUaTelemetry.OpcUaSinkWrite.Add(msg.Items.Count, new KeyValuePair<string, object?>("kind", "status"));
|
|
}
|
|
catch (ValueWriteBatchException ex)
|
|
{
|
|
_log.Warning(ex,
|
|
"OpcUaPublish: sink.WriteStatuses partially failed for a {Count}-item batch — {Failed} failed " +
|
|
"(first: {Node}), {Succeeded} applied",
|
|
msg.Items.Count, ex.FailedCount, ex.FirstFailedNodeId, ex.SucceededCount);
|
|
Interlocked.Add(ref _writes, ex.SucceededCount);
|
|
OtOpcUaTelemetry.OpcUaSinkWrite.Add(ex.SucceededCount, new KeyValuePair<string, object?>("kind", "status"));
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_log.Warning(ex, "OpcUaPublish: sink.WriteStatuses threw for a {Count}-item batch", msg.Items.Count);
|
|
}
|
|
}
|
|
|
|
private void HandleAlarmUpdate(AlarmStateUpdate msg)
|
|
{
|
|
try
|
|
{
|
|
_sink.WriteAlarmCondition(msg.AlarmNodeId, msg.State, msg.TimestampUtc, msg.Realm);
|
|
Interlocked.Increment(ref _writes);
|
|
OtOpcUaTelemetry.OpcUaSinkWrite.Add(1, new KeyValuePair<string, object?>("kind", "alarm"));
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_log.Warning(ex, "OpcUaPublish: sink.WriteAlarmCondition threw for {Node}", msg.AlarmNodeId);
|
|
}
|
|
}
|
|
|
|
private void HandleAlarmQualityUpdate(AlarmQualityUpdate msg)
|
|
{
|
|
try
|
|
{
|
|
_sink.WriteAlarmQuality(msg.AlarmNodeId, msg.StatusCode, msg.TimestampUtc, msg.Realm);
|
|
Interlocked.Increment(ref _writes);
|
|
OtOpcUaTelemetry.OpcUaSinkWrite.Add(1, new KeyValuePair<string, object?>("kind", "alarm-quality"));
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_log.Warning(ex, "OpcUaPublish: sink.WriteAlarmQuality threw for {Node}", msg.AlarmNodeId);
|
|
}
|
|
}
|
|
|
|
private void HandleRebuild(RebuildAddressSpace msg)
|
|
{
|
|
// Captured up front: the pipeline below is synchronous on this actor's thread today, but the reply
|
|
// must reach whoever asked even if a future edit introduces an await into it.
|
|
var requester = Sender;
|
|
(bool Degraded, string? Reason) outcome;
|
|
try
|
|
{
|
|
outcome = HandleRebuildCore(msg);
|
|
}
|
|
finally
|
|
{
|
|
// Gitea #641 — close the apply on EVERY exit, early returns and a throwing pass included: the node
|
|
// manager reports here the native conditions a rebuild carried over (after the notifier wiring the
|
|
// passes did) and discards any captured state no materialise claimed. Skipping it on a throw would
|
|
// leave that state for a later surgical re-add of the same id to resurrect. No-op when nothing is
|
|
// pending; never throws out of the handler.
|
|
try { _sink.CompleteAddressSpaceApply(); }
|
|
catch (Exception ex)
|
|
{
|
|
_log.Error(ex, "OpcUaPublish: sink.CompleteAddressSpaceApply threw (correlation={Correlation})", msg.Correlation);
|
|
}
|
|
}
|
|
|
|
// Gitea #601 — the outcome goes out only after the apply (including #641's completion step) is done.
|
|
if (msg.ReportOutcome)
|
|
requester.Tell(new RebuildCompleted(msg.Correlation, msg.DeploymentId, outcome.Degraded, outcome.Reason));
|
|
}
|
|
|
|
/// <summary>The rebuild pipeline. Returns whether the apply was degraded, and why (Gitea #601).</summary>
|
|
private (bool Degraded, string? Reason) HandleRebuildCore(RebuildAddressSpace msg)
|
|
{
|
|
using var span = OtOpcUaTelemetry.StartAddressSpaceRebuildSpan();
|
|
span?.SetTag("otopcua.correlation_id", msg.Correlation.ToString());
|
|
|
|
// Two modes. With an applier wired, do a REAL diff-and-apply pass — driven either by the
|
|
// artifact bytes already in hand (a per-cluster-mesh driver-only / FetchAndCache node, whose
|
|
// ConfigDb is admin-only now so _dbFactory is null, hands its fetched/cached bytes in
|
|
// msg.Artifact) or, when no bytes are carried, by loading the artifact from the ConfigDb (the
|
|
// Direct path, which requires _dbFactory). Only a node with NO applier at all (the F10b/dev
|
|
// seam) falls back to the raw-sink rebuild — and that fallback WIPES the served address space
|
|
// without re-materialising, so it must NEVER be reached by a real driver node that has an
|
|
// applier but simply has no ConfigDb (that node diff-and-applies from its in-hand bytes below).
|
|
if (_applier is null)
|
|
{
|
|
try
|
|
{
|
|
_sink.RebuildAddressSpace();
|
|
OtOpcUaTelemetry.OpcUaSinkWrite.Add(1, new KeyValuePair<string, object?>("kind", "rebuild"));
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_log.Error(ex, "OpcUaPublish: sink.RebuildAddressSpace threw (correlation={Correlation})",
|
|
msg.Correlation);
|
|
return (true, $"address-space rebuild threw: {ex.Message}");
|
|
}
|
|
return (false, null);
|
|
}
|
|
|
|
try
|
|
{
|
|
// An in-hand artifact (boot-from-cache / FetchAndCache) wins: the host already has the
|
|
// cached/fetched bytes, and the ConfigDb read the DeploymentId path would do is exactly what
|
|
// is unreachable during the outage this exists to survive — and is simply absent on a
|
|
// driver-only node with no ConfigDb. Otherwise, when a ConfigDb is present (Direct mode),
|
|
// prefer the artifact of the deployment the host just applied — at apply time it is not yet
|
|
// Sealed, so LoadLatestArtifact would return the PREVIOUS revision and materialise a stale
|
|
// composition (variables that don't match the SubscribeBulk refs); fall back to latest-sealed
|
|
// only for legacy callers that carry neither. With NO bytes AND no ConfigDb there is nothing
|
|
// to load, so yield Array.Empty<byte>() — the #485 guard below reads that as "no answer" and
|
|
// abandons the rebuild, holding last-known-good rather than wiping the address space.
|
|
var artifact = msg.Artifact
|
|
?? (_dbFactory is not null
|
|
? (msg.DeploymentId is { } depId ? LoadArtifact(depId) : LoadLatestArtifact())
|
|
: Array.Empty<byte>());
|
|
|
|
// Issue #485 — an artifact we could not obtain is NOT an empty configuration. Parsing zero
|
|
// bytes yields an empty composition, which the planner diffs against the live one as a
|
|
// PureRemove and the applier then faithfully tears down: a transient ConfigDb blip empties the
|
|
// served address space. Nothing legitimate produces a zero-length blob (an operator who really
|
|
// deletes everything still deploys a JSON document with empty arrays), so the only honest
|
|
// reading of "no bytes" is "no answer" — hold the last-known-good address space and let the
|
|
// next deploy (or the next boot) supply a real one. Returning here also leaves _lastApplied
|
|
// intact, so the retry diffs against what is actually materialised.
|
|
if (artifact is null or { Length: 0 })
|
|
{
|
|
if (_hasAppliedComposition)
|
|
_log.Warning(
|
|
"OpcUaPublish: no usable artifact for deployment {Id} (correlation={Correlation}) — KEEPING the currently served address space rather than tearing it down",
|
|
msg.DeploymentId, msg.Correlation);
|
|
else
|
|
_log.Debug(
|
|
"OpcUaPublish: no artifact to materialise yet (deployment={Id}, correlation={Correlation})",
|
|
msg.DeploymentId, msg.Correlation);
|
|
return (true, "no usable artifact reached the address-space apply; the previously served address space was kept");
|
|
}
|
|
|
|
// Parse the artifact ONCE for this rebuild. Every consumer below (data-plane authority, then the
|
|
// composition) reads this one JsonDocument instead of rebuilding it per question — the rebuild
|
|
// used to pay a full parse-and-throw-away per static call. Disposed when the handler returns.
|
|
//
|
|
// Open() throws on bytes that are not valid JSON, where the static wrappers swallowed to an
|
|
// EMPTY result — and an empty composition is not a benign no-op here: the planner diffs it as a
|
|
// PureRemove and the applier tears the served address space down. So the throw is the honest
|
|
// outcome, and it lands in this method's catch below: logged at Error, _lastApplied untouched,
|
|
// last-known-good still served. Same "no answer ≠ empty configuration" rule as the #485 guard
|
|
// immediately above, applied to bytes that arrived unreadable rather than absent.
|
|
using var parsed = DeploymentArtifact.Open(artifact);
|
|
|
|
// #520 stage 4 — install this artifact's API keys + ACL grants BEFORE the address-space work
|
|
// below, and deliberately ahead of the `plan.IsEmpty` early-return: an artifact that revokes a
|
|
// key or edits a grant without touching a single node produces an EMPTY address-space plan, so
|
|
// feeding this after that return would leave a revoked key authenticating until some unrelated
|
|
// structural change happened to redeploy. Authorization and topology change on their own
|
|
// schedules.
|
|
FeedDataPlaneAuthority(parsed, artifact);
|
|
|
|
var composition = _localNode is { } ln
|
|
? parsed.ParseComposition(ln.Value,
|
|
inconsistency => _log.Warning("OpcUaPublish {Node}: cross-cluster binding — {Message}", ln, inconsistency))
|
|
: parsed.ParseComposition();
|
|
var plan = AddressSpacePlanner.Compute(_lastApplied, composition);
|
|
|
|
if (plan.IsEmpty)
|
|
{
|
|
if (_lastAppliedDegraded)
|
|
{
|
|
// #601 review — the composition is unchanged but the last apply of it was a degraded
|
|
// rebuild: re-ensure every node (each Ensure* no-ops on a node that stands) so the
|
|
// re-deploy actually repairs what that rebuild could not place.
|
|
var repairFailed = _applier.MaterialiseHierarchy(composition)
|
|
+ MaterialiseNodes(plan, composition, fullComposition: true);
|
|
if (repairFailed > 0)
|
|
{
|
|
OtOpcUaTelemetry.OpcUaApplyFailed.Add(1, new KeyValuePair<string, object?>("kind", "nodes"));
|
|
_log.Error(
|
|
"OpcUaPublish: re-apply of a degraded composition is STILL degraded (correlation={Correlation}, failedNodes={FailedNodes})",
|
|
msg.Correlation, repairFailed);
|
|
return (true, $"re-apply of a previously degraded address space is still degraded (failedNodes={repairFailed})");
|
|
}
|
|
|
|
_lastAppliedDegraded = false;
|
|
_log.Info("OpcUaPublish: re-apply repaired the previously degraded address space (correlation={Correlation})",
|
|
msg.Correlation);
|
|
return (false, null);
|
|
}
|
|
|
|
_log.Debug("OpcUaPublish: rebuild requested but plan is empty (correlation={Correlation})",
|
|
msg.Correlation);
|
|
return (false, null);
|
|
}
|
|
|
|
var outcome = _applier.Apply(plan);
|
|
|
|
// Sum swallowed per-node materialise failures across every pass together with Apply's own
|
|
// removal-pass tally (archreview 01/S-1). A degraded apply used to vanish into per-node
|
|
// Warnings + an optimistic Info line; now it surfaces at Error + a dedicated meter.
|
|
var failedNodes = outcome.FailedNodes;
|
|
// The UNS Area/Line/Equipment folder hierarchy is materialised in FULL on every apply, rebuild or
|
|
// not: it is O(areas + lines + equipment) — trivial next to the tag counts the delta passes below
|
|
// exist to bound — and every node the delta passes add needs its ancestor folders to already
|
|
// exist, so narrowing it would mean re-deriving each add's ancestor closure (strictly more work).
|
|
failedNodes += _applier.MaterialiseHierarchy(composition);
|
|
failedNodes += MaterialiseNodes(plan, composition, fullComposition: outcome.RebuildCalled);
|
|
|
|
// 2026-08-15 surgical-deploys review C1 — record the composition as applied ONLY when the apply was
|
|
// clean, or when a rebuild ran. Two reasons this sits HERE (after the materialise passes) rather
|
|
// than immediately after Apply:
|
|
//
|
|
// • A pass that THROWS outright leaves _lastApplied at the previous composition (the outer catch
|
|
// below takes over), so the next deploy re-diffs against what is actually materialised.
|
|
// • A DEGRADED non-rebuild apply (failedNodes > 0, no rebuild) must not advance either. Before the
|
|
// delta materialisation the full passes re-ensured every node on every apply, which silently
|
|
// repaired any prior swallowed sink fault — an accidental self-healing mechanism. The delta path
|
|
// removed it: a node whose EnsureVariable failed is inside _lastApplied, never appears in a diff
|
|
// set again, and EnsureVariable early-returns on the ones that DID land — so the failed node is
|
|
// lost until some unrelated change happens to force a rebuild. Holding the composition back is
|
|
// the replacement: the next deploy re-diffs the same composition and re-ensures what failed.
|
|
//
|
|
// What the re-diff costs, stated honestly: it also re-emits the previous apply's REMOVES. A
|
|
// re-remove of an already-removed node returns false from the surgical sink, which trips the
|
|
// applier's one-way ratchet into a full rebuild — the correct resync for an apply that has already
|
|
// been logged at Error as degraded, and a bounded one (it happens once, then the address space is
|
|
// known-good again).
|
|
//
|
|
// A rebuild DOES advance even with failures: it already re-materialised the whole composition, and
|
|
// the ratchet is the resync path for anything it could not place. Holding it back would make every
|
|
// subsequent deploy re-diff the entire address space as added.
|
|
if (outcome.RebuildCalled || failedNodes == 0)
|
|
{
|
|
_lastApplied = composition;
|
|
_hasAppliedComposition = true;
|
|
// #601 review — advanced despite failures only on a rebuild; remember it so an identical
|
|
// re-deploy repairs instead of short-circuiting clean on the empty plan.
|
|
_lastAppliedDegraded = outcome.RebuildFailed || failedNodes > 0;
|
|
}
|
|
else
|
|
{
|
|
_log.Error(
|
|
"OpcUaPublish: degraded non-rebuild apply (correlation={Correlation}, failedNodes={FailedNodes}) — NOT recording this composition as applied, so the next deploy re-diffs it and repairs the missing node(s)",
|
|
msg.Correlation, failedNodes);
|
|
}
|
|
|
|
// R2-07 T4b — on a NON-rebuild apply (PureAdd / AttributeOnly), the Materialise passes above
|
|
// just created exactly the added nodes WITHOUT tearing anything down, so announce them with a
|
|
// Part 3 GeneralModelChangeEvent(NodeAdded) per affected parent — model-aware clients then
|
|
// re-browse and pick up the new nodes while every existing MonitoredItem stays alive. Ordering
|
|
// is correct by construction: Apply returned before the passes ran, and this announce runs after
|
|
// them, so the nodes exist when clients re-browse. After a full rebuild the announcement is moot
|
|
// (subscriptions are dead anyway) — the guard skips it. AnnounceAddedNodes is Safe-wrapped
|
|
// internally, so a faulting announce can never break the deploy.
|
|
//
|
|
// The announce is realm-aware (Gitea #599): UNS parents in the Uns realm, and the parents of
|
|
// added RAW containers/tags in the Raw realm. It matters because of the delta work — before it, a
|
|
// plan carrying raw adds went through a full RebuildAddressSpace, so every client re-browsed
|
|
// anyway; now those adds are materialised surgically while existing subscriptions stay alive, and
|
|
// an unannounced add would stay invisible until a client happened to browse again.
|
|
if (!outcome.RebuildCalled) _applier.AnnounceAddedNodes(plan);
|
|
|
|
OtOpcUaTelemetry.OpcUaSinkWrite.Add(1, new KeyValuePair<string, object?>("kind", "rebuild"));
|
|
if (outcome.RebuildFailed || failedNodes > 0)
|
|
{
|
|
OtOpcUaTelemetry.OpcUaApplyFailed.Add(1,
|
|
new KeyValuePair<string, object?>("kind", outcome.RebuildFailed ? "rebuild" : "nodes"));
|
|
_log.Error(
|
|
"OpcUaPublish: address-space apply DEGRADED (correlation={Correlation}, rebuildFailed={RebuildFailed}, failedNodes={FailedNodes}, added={Added}, removed={Removed}, changed={Changed}) — the running address space may be stale or partial",
|
|
msg.Correlation, outcome.RebuildFailed, failedNodes,
|
|
outcome.AddedNodes, outcome.RemovedNodes, outcome.ChangedNodes);
|
|
return (true, outcome.RebuildFailed
|
|
? $"address-space rebuild failed (failedNodes={failedNodes})"
|
|
: $"address-space apply degraded (failedNodes={failedNodes}); not recorded as applied, the next deploy of this revision re-diffs and repairs it");
|
|
}
|
|
|
|
_log.Info("OpcUaPublish: applied rebuild (correlation={Correlation}, added={Added}, removed={Removed}, changed={Changed}, rebuild={Rebuild})",
|
|
msg.Correlation, outcome.AddedNodes, outcome.RemovedNodes, outcome.ChangedNodes, outcome.RebuildCalled);
|
|
return (false, null);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_log.Error(ex, "OpcUaPublish: rebuild pipeline threw (correlation={Correlation})", msg.Correlation);
|
|
return (true, $"address-space apply threw: {ex.Message}");
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Run the five node-materialising applier passes — Raw subtree, UNS references, scripted alarms,
|
|
/// equipment tags, equipment VirtualTags — either over the FULL composition (after a rebuild) or over
|
|
/// just this plan's delta (surgical-deploys Task 6). <c>MaterialiseHierarchy</c> is NOT one of them:
|
|
/// the caller always runs it in full, because parents must exist before any add.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// <para><b>Why the delta path exists.</b> The passes are idempotent, but idempotent is not free.
|
|
/// <c>EnsureFolder</c> / <c>EnsureVariable</c> early-return on a <c>ConcurrentDictionary</c> hit
|
|
/// BEFORE taking the node manager's lock, so an already-standing node costs a dictionary probe rather
|
|
/// than a lock acquisition — but <c>MaterialiseAlarmCondition</c> DOES take the lock every time, and
|
|
/// the probes are still O(address space) per apply plus the composition walk that produces them.
|
|
/// Running the passes over the whole composition made the cost of ANY apply proportional to the size
|
|
/// of the served address space rather than to the size of the change: adding one tag to a 50k-node
|
|
/// server was ~50k sink calls on the actor's pinned publish thread, with every client's publish
|
|
/// blocked behind it. The delta path makes it O(change).</para>
|
|
/// <para><b>What the full pass was silently doing, and its replacement.</b> Re-ensuring every node on
|
|
/// every apply also REPAIRED any node a previous apply had swallowed a sink fault on — accidental
|
|
/// self-healing nobody designed. The caller's degraded-apply no-advance rule (a non-rebuild apply
|
|
/// reporting <c>failedNodes > 0</c> does not record its composition in <c>_lastApplied</c>, so the
|
|
/// next deploy re-diffs and re-ensures it) is the deliberate replacement for it.</para>
|
|
/// <para><b>Why <see cref="AddressSpaceApplyOutcome.RebuildCalled"/> is the branch key</b> rather than
|
|
/// the classifier's kind: a surgical apply that hits a <c>false</c>/throw from the surgical sink falls
|
|
/// back to a full <c>RebuildAddressSpace()</c> INSIDE <c>Apply</c>, which wipes every node. That plan
|
|
/// is still classified non-Rebuild, so branching on the kind would re-materialise only the delta onto
|
|
/// a wiped address space. The outcome flag reports what actually happened.</para>
|
|
/// <para><b>Why changed UNS references are in the subset.</b> <c>Apply</c> repairs a
|
|
/// <c>UnsReferenceDelta</c> by REMOVING the previous node and leaving the recreate to this pass, so
|
|
/// the subset must carry the <i>current</i> side of every changed reference as well as the added ones
|
|
/// — otherwise a re-pointed, renamed, or backing-edited UNS node is removed and never comes back.</para>
|
|
/// <para><b>Accepted narrowing.</b> The plan diffs the THIN <c>ScriptedAlarmPlans</c> projection, so
|
|
/// the added-alarm subset is the rich <c>EquipmentScriptedAlarms</c> rows whose id the plan reports as
|
|
/// added. An alarm already present in both compositions whose predicate SCRIPT row only now arrived
|
|
/// (making it materialisable for the first time) is not in that set — but neither is it in any diff
|
|
/// set at all, so on its own it already produced an empty plan and materialised nothing. The delta
|
|
/// path extends that pre-existing blindness to the case where it rides along with another change.</para>
|
|
/// </remarks>
|
|
/// <param name="plan">The plan just applied — the source of the delta subsets.</param>
|
|
/// <param name="composition">The full composition, always passed as the passes' lookup context.</param>
|
|
/// <param name="fullComposition">When true, materialise the whole composition (post-rebuild).</param>
|
|
/// <returns>The summed count of swallowed per-node sink failures across the passes that ran.</returns>
|
|
private int MaterialiseNodes(AddressSpacePlan plan, AddressSpaceComposition composition, bool fullComposition)
|
|
{
|
|
var applier = _applier!;
|
|
var failedNodes = 0;
|
|
|
|
// v3 Batch 4 — the Raw device subtree (ns=Raw): containers (Folder→Driver→Device→TagGroup) then
|
|
// tag Variables keyed by RawPath. Materialised BEFORE the UNS references so each UNS Organizes→Raw
|
|
// edge finds its raw target. The driver binds live values to these raw NodeIds.
|
|
if (fullComposition)
|
|
{
|
|
failedNodes += applier.MaterialiseRawSubtree(composition);
|
|
}
|
|
else if (plan.AddedRawContainers.Count > 0 || plan.AddedRawTags.Count > 0)
|
|
{
|
|
failedNodes += applier.MaterialiseRawSubtree(composition, plan.AddedRawContainers, plan.AddedRawTags);
|
|
}
|
|
|
|
// v3 Batch 4 — the UNS reference Variables (ns=UNS): each projects a raw tag under its equipment
|
|
// folder (created by MaterialiseHierarchy) with an Organizes→Raw edge; values fan out from the raw
|
|
// node. Runs AFTER both the equipment folders AND the raw subtree exist.
|
|
if (fullComposition)
|
|
{
|
|
failedNodes += applier.MaterialiseUnsReferences(composition);
|
|
}
|
|
else
|
|
{
|
|
var unsReferences = DeltaUnsReferences(plan);
|
|
if (unsReferences.Count > 0) failedNodes += applier.MaterialiseUnsReferences(composition, unsReferences);
|
|
}
|
|
|
|
// T14 — scripted alarms get their own pass right after the hierarchy so the equipment
|
|
// folders they parent under already exist. Materialises real Part 9 AlarmConditionState
|
|
// nodes (keyed by ScriptedAlarmId so AlarmStateUpdate writes target them); disabled
|
|
// alarms are skipped.
|
|
if (fullComposition)
|
|
{
|
|
failedNodes += applier.MaterialiseScriptedAlarms(composition);
|
|
}
|
|
else
|
|
{
|
|
var addedAlarms = DeltaScriptedAlarms(plan, composition);
|
|
if (addedAlarms.Count > 0) failedNodes += applier.MaterialiseScriptedAlarms(composition, addedAlarms);
|
|
}
|
|
|
|
// Equipment-namespace tags get their own pass: ensures each signal's Variable (and any
|
|
// FolderPath sub-folder) exists under its already-materialised equipment folder so
|
|
// clients can browse them. Live values are pushed by DriverHostActor.ForwardBatchToMux after
|
|
// each subscription cycle; variables show BadWaitingForInitialData only until the first
|
|
// publish interval fires.
|
|
if (fullComposition)
|
|
{
|
|
failedNodes += applier.MaterialiseEquipmentTags(composition);
|
|
}
|
|
else if (plan.AddedEquipmentTags.Count > 0)
|
|
{
|
|
failedNodes += applier.MaterialiseEquipmentTags(composition, plan.AddedEquipmentTags);
|
|
}
|
|
|
|
// Equipment-namespace VirtualTags get their own pass right after the equipment tags:
|
|
// ensures each computed signal's Variable (and any FolderPath sub-folder) exists under its
|
|
// equipment folder with a folder-scoped NodeId. VirtualTagHostActor.OnResult pushes live
|
|
// values once the first dependency update arrives; until then variables show BadWaitingForInitialData.
|
|
if (fullComposition)
|
|
{
|
|
failedNodes += applier.MaterialiseEquipmentVirtualTags(composition);
|
|
}
|
|
else if (plan.AddedEquipmentVirtualTags.Count > 0)
|
|
{
|
|
failedNodes += applier.MaterialiseEquipmentVirtualTags(composition, plan.AddedEquipmentVirtualTags);
|
|
}
|
|
|
|
return failedNodes;
|
|
}
|
|
|
|
/// <summary>The UNS reference variables a non-rebuild apply must materialise: the plan's ADDED references
|
|
/// plus the CURRENT side of every CHANGED one (whose previous node the applier already removed in place).
|
|
/// </summary>
|
|
/// <param name="plan">The plan just applied.</param>
|
|
/// <returns>The references to materialise (possibly empty).</returns>
|
|
private static IReadOnlyList<UnsReferenceVariable> DeltaUnsReferences(AddressSpacePlan plan)
|
|
{
|
|
if (plan.ChangedUnsReferenceVariables.Count == 0) return plan.AddedUnsReferenceVariables;
|
|
var list = new List<UnsReferenceVariable>(
|
|
plan.AddedUnsReferenceVariables.Count + plan.ChangedUnsReferenceVariables.Count);
|
|
list.AddRange(plan.AddedUnsReferenceVariables);
|
|
foreach (var d in plan.ChangedUnsReferenceVariables) list.Add(d.Current);
|
|
return list;
|
|
}
|
|
|
|
/// <summary>The scripted-alarm host plans a non-rebuild apply must materialise: the rich
|
|
/// <c>EquipmentScriptedAlarms</c> rows whose <c>ScriptedAlarmId</c> the plan reports as ADDED (the plan
|
|
/// diffs only the thin projection — see the narrowing note on <see cref="MaterialiseNodes"/>).</summary>
|
|
/// <param name="plan">The plan just applied.</param>
|
|
/// <param name="composition">The full composition carrying the rich alarm plans.</param>
|
|
/// <returns>The alarms to materialise (possibly empty).</returns>
|
|
private static IReadOnlyList<EquipmentScriptedAlarmPlan> DeltaScriptedAlarms(
|
|
AddressSpacePlan plan, AddressSpaceComposition composition)
|
|
{
|
|
if (plan.AddedAlarms.Count == 0 || composition.EquipmentScriptedAlarms.Count == 0)
|
|
return Array.Empty<EquipmentScriptedAlarmPlan>();
|
|
|
|
var addedIds = new HashSet<string>(plan.AddedAlarms.Select(a => a.ScriptedAlarmId), StringComparer.Ordinal);
|
|
return composition.EquipmentScriptedAlarms.Where(a => addedIds.Contains(a.ScriptedAlarmId)).ToList();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Decode the artifact's data-plane authorization inputs (API keys + this cluster's ACL grants)
|
|
/// and install them for OPC UA session activation (#520 stage 4). No-op when nothing consumes
|
|
/// them.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// Wrapped so a decode failure can never break a deploy — the same discipline as the historised-ref
|
|
/// feed. The cost of that choice is explicit: the node keeps serving the PREVIOUS key set, which is
|
|
/// the safe direction for availability but means a revocation carried by an artifact this node
|
|
/// could not decode has not taken effect. It is logged at Error for exactly that reason.
|
|
/// </remarks>
|
|
/// <param name="parsed">The rebuild's single parse of the artifact, read for the keys + grants.</param>
|
|
/// <param name="artifactBytes">The RAW artifact bytes — hashed for the revision id, which must be
|
|
/// SHA-256 of the exact blob (see below); never re-parsed here.</param>
|
|
private void FeedDataPlaneAuthority(ParsedDeploymentArtifact parsed, byte[] artifactBytes)
|
|
{
|
|
if (_dataPlaneAuthority is null) return;
|
|
|
|
try
|
|
{
|
|
var nodeId = _localNode?.Value ?? string.Empty;
|
|
var projection = parsed.ParseDataPlaneAuthority(nodeId);
|
|
// The revision hash is SHA-256 of the artifact bytes — the same derivation ConfigComposer uses,
|
|
// which is what lets a node recognise a re-applied revision without the hash riding the wire.
|
|
// Hashed off the RAW bytes for exactly that reason: a re-serialisation of the parsed document
|
|
// would be a different byte sequence and would not match the deployment's RevisionHash.
|
|
var revision = Convert.ToHexStringLower(System.Security.Cryptography.SHA256.HashData(artifactBytes));
|
|
var snapshot = _dataPlaneAuthority.Apply(
|
|
projection.ClusterId, revision, projection.Keys, projection.Grants, projection.ScopePaths);
|
|
|
|
_log.Info(
|
|
"OpcUaPublish: data-plane authority generation {Generation} installed (cluster={Cluster}, revision={Revision}, keys={Keys}, grants={Grants}, resolvedScopes={ResolvedScopes})",
|
|
snapshot.GenerationId, projection.ClusterId ?? "(single-cluster)", revision[..12],
|
|
projection.Keys.Count, projection.Grants.Count, projection.ScopePaths.Count);
|
|
|
|
// Reported BY the trie build (#520 stage 5), not re-derived here. The previous version of this
|
|
// log counted unresolvable ScopeIds itself and so agreed with the builder on exactly one of its
|
|
// four drop reasons — a row with a blank ScopeId, a retired scope kind, or a scope path that
|
|
// does not start at a realm all vanished with the log still reading zero. A dropped row is an
|
|
// authored rule that can never match, and this is the operator's only warning of it.
|
|
if (snapshot.DroppedGrants.Count > 0)
|
|
{
|
|
_log.Warning(
|
|
"OpcUaPublish: {Dropped} ACL rule(s) in generation {Generation} were DROPPED and will never match — {Detail}",
|
|
snapshot.DroppedGrants.Count, snapshot.GenerationId,
|
|
string.Join("; ", snapshot.DroppedGrants
|
|
.Select(d => $"acl {d.NodeAclId} ({d.ScopeKind} '{d.ScopeId}') {d.Reason}")));
|
|
}
|
|
|
|
if (string.IsNullOrWhiteSpace(projection.ClusterId) && projection.Grants.Count > 0)
|
|
{
|
|
// No cluster identity means no trie was built at all, so nothing can be granted here. Say
|
|
// so explicitly: with enforcement on, this node denies every write, alarm command and
|
|
// HistoryRead, and the grant count logged above would otherwise suggest the opposite.
|
|
_log.Warning(
|
|
"OpcUaPublish: this node has no cluster identity, so NO permission trie was built for its {Grants} grant(s) — with Security:Acl:Enabled every scoped decision here will be denied",
|
|
projection.Grants.Count);
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_log.Error(ex,
|
|
"OpcUaPublish: failed to install data-plane authority from the applied artifact — the node keeps the PREVIOUS keys and grants, so any revocation in this deployment has NOT taken effect");
|
|
}
|
|
}
|
|
|
|
/// <summary>Read a specific deployment's artifact blob from ConfigDb (the one just applied,
|
|
/// which may not be Sealed yet). Empty array on any failure — <see cref="HandleRebuild"/> treats that as
|
|
/// "no answer" and abandons the rebuild, leaving the served address space untouched (issue #485).</summary>
|
|
private byte[] LoadArtifact(DeploymentId deploymentId)
|
|
{
|
|
try
|
|
{
|
|
using var db = _dbFactory!.CreateDbContext();
|
|
return db.Deployments.AsNoTracking()
|
|
.Where(d => d.DeploymentId == deploymentId.Value)
|
|
.Select(d => d.ArtifactBlob)
|
|
.FirstOrDefault() ?? Array.Empty<byte>();
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_log.Warning(ex, "OpcUaPublish: failed to load artifact for deployment {Id}; the rebuild is abandoned", deploymentId);
|
|
return Array.Empty<byte>();
|
|
}
|
|
}
|
|
|
|
/// <summary>Read the most recent <c>Sealed</c> deployment's artifact blob from ConfigDb.
|
|
/// Empty array on any failure — see <see cref="LoadArtifact"/> for how that is handled.</summary>
|
|
private byte[] LoadLatestArtifact()
|
|
{
|
|
try
|
|
{
|
|
using var db = _dbFactory!.CreateDbContext();
|
|
return db.Deployments.AsNoTracking()
|
|
.Where(d => d.Status == Configuration.Enums.DeploymentStatus.Sealed)
|
|
.OrderByDescending(d => d.SealedAtUtc)
|
|
.Select(d => d.ArtifactBlob)
|
|
.FirstOrDefault() ?? Array.Empty<byte>();
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_log.Warning(ex, "OpcUaPublish: failed to load latest deployment artifact; the rebuild is abandoned");
|
|
return Array.Empty<byte>();
|
|
}
|
|
}
|
|
|
|
private void HandleServiceLevelChanged(ServiceLevelChanged msg)
|
|
{
|
|
// Always publish the FIRST computed level, even if it equals the byte-default 0. Otherwise a
|
|
// node starting Detached/role-less (first level = 0) would be dedup'd away, leaving the SDK's
|
|
// built-in default (255 = full service) standing — a degraded node wrongly advertising 255.
|
|
if (_publishedAtLeastOnce && msg.ServiceLevel == _lastServiceLevel) return;
|
|
try
|
|
{
|
|
_serviceLevel.Publish(msg.ServiceLevel);
|
|
_lastServiceLevel = msg.ServiceLevel;
|
|
_publishedAtLeastOnce = true;
|
|
OtOpcUaTelemetry.ServiceLevelChange.Add(1,
|
|
new KeyValuePair<string, object?>("level", msg.ServiceLevel));
|
|
_log.Debug("OpcUaPublish: ServiceLevel={Level}", msg.ServiceLevel);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
// A failed write may be partially applied; even a previous successful value must retry.
|
|
_publishedAtLeastOnce = false;
|
|
_log.Warning(ex, "OpcUaPublish: ServiceLevel publisher threw at level {Level}", msg.ServiceLevel);
|
|
}
|
|
}
|
|
|
|
/// <summary>Caches the latest redundancy snapshot and recomputes the local ServiceLevel.
|
|
/// The actual byte is produced by <see cref="RecomputeServiceLevel"/> — either via the
|
|
/// health-aware <see cref="ServiceLevelCalculator"/> (once a DB-health probe + sample are
|
|
/// wired) or via the legacy role-only seam (back-compat / bootstrap).</summary>
|
|
private void HandleRedundancyStateChanged(RedundancyStateChanged msg)
|
|
{
|
|
_lastSnapshot = msg;
|
|
RecomputeServiceLevel();
|
|
}
|
|
|
|
/// <summary>Caches the latest DB-health sample and recomputes the local ServiceLevel. The
|
|
/// probe pushes these (or the actor Asks for them); either way the freshest sample feeds the
|
|
/// calculator's <c>DbReachable</c>/<c>Stale</c> inputs.</summary>
|
|
private void HandleDbHealthStatus(DbHealthProbeActor.DbHealthStatus msg)
|
|
{
|
|
_lastDbHealth = msg;
|
|
RecomputeServiceLevel();
|
|
}
|
|
|
|
/// <summary>Periodic self-tick: Asks the local <see cref="DbHealthProbeActor"/> for its cached
|
|
/// status and pipes the reply back to <see cref="Self"/>, where <see cref="HandleDbHealthStatus"/>
|
|
/// caches it + recomputes. A hung/late probe (Ask timeout) pipes a <c>Reachable=false</c> status so
|
|
/// the node FAIL-SAFE-DEMOTES rather than freezing the last-known-good. The continuation only
|
|
/// CONSTRUCTS a record off the actor thread — the actual state mutation happens on the actor thread
|
|
/// when the piped <see cref="DbHealthProbeActor.DbHealthStatus"/> is received, so no actor field is
|
|
/// touched here.</summary>
|
|
private void OnHealthTick()
|
|
{
|
|
_dbHealthProbe!.Ask<DbHealthProbeActor.DbHealthStatus>(
|
|
DbHealthProbeActor.GetStatus.Instance, TimeSpan.FromSeconds(1))
|
|
.ContinueWith(t => t.IsCompletedSuccessfully
|
|
? t.Result
|
|
: new DbHealthProbeActor.DbHealthStatus(false, DateTime.UtcNow, "db-health ask timeout"))
|
|
.PipeTo(Self);
|
|
}
|
|
|
|
/// <summary>Records a peer's OPC UA probe verdict about THIS node and recomputes the local
|
|
/// ServiceLevel. The probe's <see cref="PeerOpcUaProbeActor.OpcUaProbeResult.NodeId"/> is the
|
|
/// target that was probed, so a result whose <c>NodeId</c> is not this node is about a peer and
|
|
/// is ignored. A matching result is stamped with the receive time so <see cref="OpcUaProbeOk"/>
|
|
/// can debounce stale verdicts.</summary>
|
|
private void HandlePeerProbe(PeerOpcUaProbeActor.OpcUaProbeResult r)
|
|
{
|
|
// The result targets the probed node. If it isn't me, it's about a peer — ignore it.
|
|
if (_localNode is null || r.NodeId != _localNode.Value) return;
|
|
_probeAboutMe = (r.Ok, DateTime.UtcNow);
|
|
RecomputeServiceLevel();
|
|
}
|
|
|
|
/// <summary>The OPC UA self-probe input for the calculator: "did a peer recently observe MY OPC UA
|
|
/// endpoint as reachable?" Returns <c>true</c> (benefit of the doubt) when no peer verdict has
|
|
/// arrived yet (single-node / no peer) or when the latest verdict is older than
|
|
/// <see cref="_probeFreshnessWindow"/> (the peer went away — don't penalise this node for that).
|
|
/// Only an actively-observed, RECENT <c>Ok==false</c> demotes.</summary>
|
|
private bool OpcUaProbeOk()
|
|
{
|
|
if (_probeAboutMe is not { } verdict) return true;
|
|
if (DateTime.UtcNow - verdict.At > _probeFreshnessWindow) return true;
|
|
return verdict.Ok;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Computes the local OPC UA ServiceLevel and routes it through <see cref="ServiceLevelChanged"/>
|
|
/// (the dedup/publish/metric handler). The full <see cref="ServiceLevelCalculator"/> path is
|
|
/// used once a DB-health probe is wired AND a sample has arrived; until then (and when no probe
|
|
/// is supplied at all) a legacy role-only seam keeps the historical "primary-leader → 240,
|
|
/// secondary → 100, detached → 0" behaviour. The calculator does not model Detached, so a
|
|
/// detached local node is guarded to 0 before either path runs.
|
|
/// </summary>
|
|
private void RecomputeServiceLevel()
|
|
{
|
|
if (_localNode is null || _lastSnapshot is null) return;
|
|
|
|
var entry = _lastSnapshot.Nodes.FirstOrDefault(n => n.NodeId == _localNode.Value);
|
|
|
|
// The calculator does NOT model Detached — a healthy detached node would wrongly compute
|
|
// 240, so guard it (and the missing-entry case) to 0 here.
|
|
if (entry is null || entry.Role == RedundancyRole.Detached)
|
|
{
|
|
Self.Tell(new ServiceLevelChanged(0));
|
|
return;
|
|
}
|
|
|
|
// Per-cluster mesh Phase 4: on a driver-only (DB-less) node there is no ConfigDb to probe —
|
|
// LocalDb IS the config store and is in-process, so it can never be "unreachable". Retire the
|
|
// DB-reachability input by feeding DbReachable=true CONSTANT and deriving staleness from the
|
|
// snapshot age ALONE (no DB-health term). This is the survive-alone posture and it is
|
|
// client-visible: a healthy DB-less node must publish 240 (240+10=250 as Primary) even with
|
|
// central down, so clients keep steering to it — the old DbReachable=false would have computed
|
|
// the (false,true,false) → 0 arm and driven clients AWAY from a node serving perfectly.
|
|
// Staleness ("running behind on config") still demotes to 200. The Detached guard above still
|
|
// wins, so a DB-less Detached node stays 0.
|
|
if (_dbLess)
|
|
{
|
|
var utcNow = DateTime.UtcNow;
|
|
var dbLessInputs = new NodeHealthInputs(
|
|
MemberState: SafeSelfStatus(),
|
|
DbReachable: true,
|
|
OpcUaProbeOk: OpcUaProbeOk(),
|
|
Stale: (utcNow - entry.AsOfUtc) > _staleWindow,
|
|
IsDriverPrimary: entry.IsDriverPrimary);
|
|
Self.Tell(new ServiceLevelChanged(ServiceLevelCalculator.Compute(dbLessInputs)));
|
|
return;
|
|
}
|
|
|
|
// Legacy / back-compat seam: with no DB-health probe wired (or before the first sample
|
|
// arrives) fall back to the old role-only switch. This preserves historical behaviour and
|
|
// is the bootstrap value until the first DbHealthStatus lands.
|
|
if (_dbHealthProbe is null || _lastDbHealth is null)
|
|
{
|
|
Self.Tell(new ServiceLevelChanged(LegacyRoleOnly(entry)));
|
|
return;
|
|
}
|
|
|
|
var now = DateTime.UtcNow;
|
|
var inputs = new NodeHealthInputs(
|
|
MemberState: SafeSelfStatus(),
|
|
DbReachable: _lastDbHealth.Reachable,
|
|
OpcUaProbeOk: OpcUaProbeOk(),
|
|
Stale: !_lastDbHealth.Reachable
|
|
|| (now - _lastDbHealth.AsOfUtc) > _staleWindow
|
|
|| (now - entry.AsOfUtc) > _staleWindow,
|
|
IsDriverPrimary: entry.IsDriverPrimary);
|
|
|
|
Self.Tell(new ServiceLevelChanged(ServiceLevelCalculator.Compute(inputs)));
|
|
}
|
|
|
|
/// <summary>The legacy role-only ServiceLevel switch (primary-leader → 240, primary → 200,
|
|
/// secondary → 100, _ → 0). Preserved as the back-compat / bootstrap seam.</summary>
|
|
private static byte LegacyRoleOnly(NodeRedundancyState entry) => entry.Role switch
|
|
{
|
|
RedundancyRole.Primary when entry.IsDriverPrimary => 240,
|
|
RedundancyRole.Primary => 200,
|
|
RedundancyRole.Secondary => 100,
|
|
_ => 0,
|
|
};
|
|
|
|
/// <summary>Reads this node's cluster <see cref="MemberStatus"/>, returning
|
|
/// <see cref="MemberStatus.Removed"/> if the cluster is unavailable (so the calculator treats it
|
|
/// as untrusted → 0 rather than throwing).</summary>
|
|
private MemberStatus SafeSelfStatus()
|
|
{
|
|
try
|
|
{
|
|
return _cluster.SelfMember.Status;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_log.Debug(ex, "OpcUaPublish: SelfMember status unavailable; treating as Removed (ServiceLevel→0)");
|
|
return MemberStatus.Removed;
|
|
}
|
|
}
|
|
}
|