Joseph Doherty
30d0697c28
`IHostConnectivityProbe` was a dead surface: eleven drivers implement it, `GetHostStatuses()` had ZERO production call sites, and `OnHostStatusChanged` had no subscriber outside the Galaxy driver's own aggregator. Per-host connectivity was computed by every driver and read by nobody. The issue offered "build the publisher or delete it". The publisher as its entity doc described it — driver nodes upserting `DriverHostStatus` rows — is not buildable: per-cluster mesh Phase 4 gates `AddOtOpcUaConfigDb` on the `admin` role, so a driver-only node has no ConfigDb connection to write rows with. So the capability is kept and the transport changed. `DriverHealthChanged.HostStatuses` now carries the probe result to `/hosts` as a Hosts column. That channel already reached the page, already survives the mesh split via the Phase 5 gRPC telemetry stream, and already replays a last-value snapshot on re-subscribe — so per-host state re-primes after a reconnect without a durable store. Both halves of the interface finally do what they are for: the event triggers a prompt publish, the pull is the source of truth. The point of the column is the case the driver-level state chip structurally cannot express: a multi-device driver stays aggregate-Healthy while ONE of its devices is unreachable. Two traps, both pinned by tests that were falsified against the prod code: - The host digest MUST be in the publish fingerprint. On a single-host-down transition every other fingerprint component is unchanged, so the dedup would swallow exactly the publish carrying the news — the trap that already bit the rediscovery signal. Removing it turns the guard test red, verified. - null (no probe) must stay distinct from empty (probe with no hosts). proto3 cannot tell an absent repeated field from an empty one, hence the explicit `has_host_statuses` flag; collapsing them would render every probe-less driver as one whose devices are all fine. Dropped: the DriverHostStatus entity, enum, DbSet, model config and table (migration DropDriverHostStatusTable — empty on every deployment, so the scaffolder's data-loss warning is moot, and Down() recreates it exactly). Found en route, NOT fixed here: `DriverInstanceResilienceStatus` is the identical defect — no writer, no reader, only a DbSet declaration, while the live data rides the `driver-resilience-status` telemetry channel. Its doc-comment now states that rather than describing the sampler and AdminUI join that were never built. Filed as #524 rather than widening this schema change beyond what was asked. Claude-Session: https://claude.ai/code/session_015p7wGqy3YpZNCpDzTpGMKo
1079 lines
62 KiB
C#
1079 lines
62 KiB
C#
using Akka.Actor;
|
|
using Akka.Event;
|
|
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;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
|
|
|
|
// private timer key type — file-scoped so the name stays unique per-file
|
|
file sealed class HealthPollTick
|
|
{
|
|
public static readonly HealthPollTick Instance = new();
|
|
private HealthPollTick() { }
|
|
}
|
|
|
|
/// <summary>
|
|
/// Akka wrapper for a single <see cref="IDriver"/> instance. States:
|
|
///
|
|
/// <list type="bullet">
|
|
/// <item><c>Connecting</c> — calling <see cref="IDriver.InitializeAsync"/>.</item>
|
|
/// <item><c>Connected</c> — initialised; serving Read/Write/Subscribe requests.</item>
|
|
/// <item><c>Reconnecting</c> — disconnect observed; periodic retry of Initialize.</item>
|
|
/// <item><c>Failed</c> — terminal until parent restarts the actor.</item>
|
|
/// </list>
|
|
///
|
|
/// Engine wiring (subscriptions → AttributeValueUpdate publishes, ApplyDelta-driven Reinitialize,
|
|
/// per-tag write Asks) is staged for follow-up F7. This skeleton compiles + has a working
|
|
/// state machine so the Phase 6 control-plane integration tests can target it.
|
|
/// </summary>
|
|
public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
|
{
|
|
public static readonly TimeSpan DefaultReconnectInterval = TimeSpan.FromSeconds(10);
|
|
|
|
public sealed record InitializeRequested(string DriverConfigJson);
|
|
public sealed record InitializeSucceeded(int Generation);
|
|
public sealed record InitializeFailed(string Reason, int Generation);
|
|
public sealed record DisconnectObserved(string Reason);
|
|
/// <summary>#477 — sent to the parent (<see cref="DriverHostActor"/>) on every connectivity transition:
|
|
/// <c>Connected=true</c> on entering the Connected state, <c>false</c> on entering Reconnecting. The host
|
|
/// annotates this driver's native alarm conditions' source-data Quality from it (comms lost → Bad,
|
|
/// restored → Good) — independently of alarm transitions, since a comms-lost driver emits no alarm
|
|
/// events. Fire-and-forget, mirroring <see cref="DeltaApplied"/>.</summary>
|
|
public sealed record ConnectivityChanged(string DriverInstanceId, bool Connected);
|
|
public sealed record ApplyDelta(string DriverConfigJson, CorrelationId Correlation);
|
|
public sealed record ApplyResult(bool Success, string? Reason, CorrelationId Correlation);
|
|
/// <summary>
|
|
/// Sent to the parent (<see cref="DriverHostActor"/>) AFTER an <see cref="ApplyDelta"/> has fully
|
|
/// re-initialised the driver (i.e. <see cref="IDriver.ReinitializeAsync"/> completed). The host uses
|
|
/// it to re-register this driver's dependency-consumer mux adapter: a driver may rebuild its declared
|
|
/// <see cref="IDependencyConsumer.DependencyRefs"/> during reinit (the Calculation driver re-binds its
|
|
/// calc tags), so the re-register MUST happen after reinit — NOT synchronously alongside the ApplyDelta
|
|
/// dispatch, which would re-read the STALE ref set. Emitted only on a successful apply.
|
|
/// </summary>
|
|
/// <param name="DriverInstanceId">The instance whose adapter should be re-registered.</param>
|
|
public sealed record DeltaApplied(string DriverInstanceId);
|
|
public sealed record WriteAttribute(string TagId, object Value);
|
|
public sealed record WriteAttributeResult(bool Success, string? Reason);
|
|
/// <summary>
|
|
/// Sent by <see cref="DriverHostActor"/> when an OPC UA client Acknowledges a NATIVE Part 9
|
|
/// condition (resolved from the condition NodeId to this driver via the host's alarm inverse map).
|
|
/// The actor forwards it to the driver's <see cref="IAlarmSource.AcknowledgeAsync"/>, carrying the
|
|
/// authored alarm full-reference (= the driver's <c>ConditionId</c>/AlarmFullReference) and the
|
|
/// authenticated principal. Mirrors <see cref="WriteAttribute"/>, but the ack is fire-and-forget:
|
|
/// the driver's <see cref="IAlarmSource.AcknowledgeAsync"/> returns no per-condition status and the
|
|
/// OPC UA Part 9 ack already committed the local condition state, so there is no reply to surface.
|
|
/// </summary>
|
|
/// <param name="ConditionId">The authored alarm full-reference the driver correlates the ack on
|
|
/// (= the equipment tag's <c>FullName</c>/AlarmFullReference).</param>
|
|
/// <param name="Comment">Operator-supplied comment forwarded to the upstream alarm system; null when none.</param>
|
|
/// <param name="OperatorUser">The authenticated principal performing the acknowledge.</param>
|
|
public sealed record RouteAlarmAck(string ConditionId, string? Comment, string OperatorUser);
|
|
public sealed record Subscribe(IReadOnlyList<string> FullReferences, TimeSpan PublishingInterval);
|
|
/// <summary>
|
|
/// Sets the set of references this driver should keep subscribed for the lifetime of the
|
|
/// current deployment. Unlike the one-shot <see cref="Subscribe"/>, the desired set is
|
|
/// retained and (re)established automatically every time the actor (re)enters
|
|
/// <c>Connected</c> — closing the F8b "re-subscribe across reconnects" gap and giving
|
|
/// <see cref="DriverHostActor"/> a single message to drive the SubscribeBulk pass after an
|
|
/// apply. Sending an empty set clears the desired subscription.
|
|
/// </summary>
|
|
/// <param name="AlarmReferences">
|
|
/// The native-alarm references (alarm-bearing equipment-tag FullNames = the driver's
|
|
/// <c>ConditionId</c>/AlarmFullReference) this driver should keep an alarm subscription open for.
|
|
/// An <see cref="IAlarmSource"/> driver suppresses <see cref="IAlarmSource.OnAlarmEvent"/> until at
|
|
/// least one alarm subscription exists, so the actor calls
|
|
/// <see cref="IAlarmSource.SubscribeAlarmsAsync"/> with this set to un-gate the native feed. Empty
|
|
/// (or null) means the driver has no alarm tags. Defaults to null so non-alarm callers are unchanged.
|
|
/// </param>
|
|
public sealed record SetDesiredSubscriptions(
|
|
IReadOnlyList<string> FullReferences,
|
|
TimeSpan PublishingInterval,
|
|
IReadOnlyList<string>? AlarmReferences = null);
|
|
public sealed record SubscriptionEstablished(string DiagnosticId, int ReferenceCount);
|
|
public sealed record SubscriptionFailed(string Reason);
|
|
public sealed record Unsubscribe;
|
|
/// <summary>
|
|
/// Sent by <see cref="DriverHostActor"/> when the AdminUI issues a Reconnect operation.
|
|
/// Pushes the actor out of <c>Connected</c> into <c>Reconnecting</c> so the transport is
|
|
/// re-established without fully stopping and respawning the actor. Safe to send in any
|
|
/// state — a no-op when already Reconnecting or Connecting.
|
|
/// </summary>
|
|
public sealed record ForceReconnect;
|
|
/// <summary>Published to the actor's parent whenever the subscribed IDriver fires
|
|
/// <see cref="ISubscribable.OnDataChange"/>. The parent forwards to OpcUaPublishActor.</summary>
|
|
public sealed record AttributeValuePublished(string DriverInstanceId, string FullReference, object? Value, OpcUaQuality Quality, DateTime TimestampUtc);
|
|
private sealed record DataChangeForward(string FullReference, DataValueSnapshot Snapshot);
|
|
/// <summary>Published to the parent whenever the subscribed driver (an <see cref="IAlarmSource"/>) fires
|
|
/// <see cref="IAlarmSource.OnAlarmEvent"/>. The parent (<see cref="DriverHostActor"/>) projects + routes it
|
|
/// to the materialised Part 9 condition. Parallels <see cref="AttributeValuePublished"/>.</summary>
|
|
public sealed record AttributeAlarmPublished(string DriverInstanceId, AlarmEventArgs Args);
|
|
private sealed record NativeAlarmRaised(AlarmEventArgs Args);
|
|
/// <summary>Self-sent on Connected entry / when alarm refs are (re)pushed, to establish the native-alarm
|
|
/// subscription that un-gates an <see cref="IAlarmSource"/> driver's feed. Handled async so the
|
|
/// <see cref="IAlarmSource.SubscribeAlarmsAsync"/> call is bounded + off the synchronous handlers.</summary>
|
|
private sealed record SubscribeAlarms;
|
|
/// <summary>Self-sent when the wrapped driver raises <see cref="IRediscoverable.OnRediscoveryNeeded"/> —
|
|
/// it observed that the remote's tag set may have changed. Marshals the event off the driver's thread
|
|
/// onto the actor thread. Handled in EVERY behaviour, including Stubbed and Reconnecting: the raise has no
|
|
/// connection affinity (a Galaxy redeploy or a TwinCAT symbol-version bump can land while the driver is
|
|
/// between connects), and dropping it in one state would lose the signal silently.</summary>
|
|
private sealed record RediscoveryRaised(RediscoveryEventArgs Args);
|
|
/// <summary>Self-sent when the wrapped driver raises <see cref="IHostConnectivityProbe.OnHostStatusChanged"/>
|
|
/// — one of its hosts went Running ↔ Stopped ↔ Faulted. Marshals the event off the driver's probe thread
|
|
/// onto the actor thread. Carries NO payload: the handler re-pulls
|
|
/// <see cref="IHostConnectivityProbe.GetHostStatuses"/>, so the driver stays the single source of truth
|
|
/// for the host set and a host that appeared since the last publish is picked up too. Handled in every
|
|
/// behaviour for the same reason as <see cref="RediscoveryRaised"/> — a probe tick can land while the
|
|
/// driver is between connects.</summary>
|
|
private sealed record HostStatusRaised;
|
|
|
|
public sealed class RetryConnect
|
|
{
|
|
public static readonly RetryConnect Instance = new();
|
|
private RetryConnect() { }
|
|
}
|
|
|
|
/// <summary>Interval between periodic health-poll heartbeats sent to the snapshot store.</summary>
|
|
public static readonly TimeSpan HealthPollInterval = TimeSpan.FromSeconds(30);
|
|
|
|
private readonly IDriver _driver;
|
|
|
|
/// <summary>Phase 6.1 resilience seam: every guarded driver-capability call this actor makes is
|
|
/// routed through this invoker (retry / breaker / in-flight accounting / tracker telemetry). Defaults to
|
|
/// <see cref="NullDriverCapabilityInvoker"/> (a genuine pass-through) when none is supplied — so
|
|
/// unit tests + the F7 skeleton path behave exactly as a raw driver call. The fused Host builds a
|
|
/// real per-instance invoker via <see cref="IDriverCapabilityInvokerFactory"/> and injects it at
|
|
/// spawn. Consumed through the <see cref="IDriverCapabilityInvoker"/> abstraction because Runtime
|
|
/// is deliberately Polly-free (it does not reference <c>ZB.MOM.WW.OtOpcUa.Core</c>).</summary>
|
|
private readonly IDriverCapabilityInvoker _invoker;
|
|
private readonly string _driverInstanceId;
|
|
private readonly string _clusterId;
|
|
private readonly IDriverHealthPublisher _healthPublisher;
|
|
private readonly TimeSpan _reconnectInterval;
|
|
|
|
private readonly TimeSpan _healthPollInterval;
|
|
|
|
private readonly ILoggingAdapter _log = Context.GetLogger();
|
|
private string? _currentConfigJson;
|
|
|
|
/// <summary>Monotonic token tagging each <see cref="InitializeAsync"/> attempt. An init result is
|
|
/// honoured only when its generation matches the latest; an older result is from a superseded attempt
|
|
/// (e.g. an <see cref="ApplyDelta"/> adopted a new config mid-(re)connect) and is dropped. Touched only
|
|
/// on the actor thread, so no lock is needed.</summary>
|
|
private int _initGeneration;
|
|
|
|
/// <summary>
|
|
/// Timestamps of recent Faulted-state transitions; used to compute the 5-minute error count.
|
|
/// No lock needed — every read/write site runs inside an Akka message handler, which is
|
|
/// single-threaded per actor instance.
|
|
/// </summary>
|
|
private readonly Queue<DateTime> _faultTimestamps = new();
|
|
|
|
/// <summary>Active subscription handle (null when not subscribed). Tracks the current live
|
|
/// subscription; the actor auto-(re)subscribes on (re)connect and on each <see cref="Subscribe"/>
|
|
/// message via <see cref="ResubscribeDesired"/> / <see cref="HandleSubscribeAsync"/>, so callers
|
|
/// do not need to re-send subscription requests after a reconnect.</summary>
|
|
private ISubscriptionHandle? _subscriptionHandle;
|
|
private EventHandler<DataChangeEventArgs>? _dataChangeHandler;
|
|
private EventHandler<AlarmEventArgs>? _alarmEventHandler;
|
|
private EventHandler<RediscoveryEventArgs>? _rediscoveryHandler;
|
|
private EventHandler<HostStatusChangedEventArgs>? _hostStatusHandler;
|
|
|
|
/// <summary>When the driver last raised <see cref="IRediscoverable.OnRediscoveryNeeded"/>, and the reason
|
|
/// it gave. Null until the first raise. Carried on every subsequent health snapshot so the AdminUI can
|
|
/// prompt an operator to re-browse the device.
|
|
/// <para>Deliberately <b>sticky</b> — it is not cleared on reconnect or on a later clean pass. The remote's
|
|
/// tag set stayed changed; only an operator re-browsing and committing resolves it, and this actor cannot
|
|
/// observe that happening. A redeploy respawns the child, which clears it naturally.</para></summary>
|
|
private DateTime? _rediscoveryNeededUtc;
|
|
private string? _rediscoveryReason;
|
|
|
|
/// <summary>The references the host wants kept subscribed (set by <see cref="SetDesiredSubscriptions"/>).
|
|
/// Re-applied on every entry into <c>Connected</c> so values resume after a reconnect or redeploy.</summary>
|
|
private IReadOnlyList<string> _desiredRefs = Array.Empty<string>();
|
|
private TimeSpan _desiredInterval = TimeSpan.FromSeconds(1);
|
|
|
|
/// <summary>The native-alarm references the host wants kept subscribed (set by
|
|
/// <see cref="SetDesiredSubscriptions"/>). Re-applied on every <c>Connected</c> entry so the alarm
|
|
/// feed is re-un-gated after a reconnect/redeploy.</summary>
|
|
private IReadOnlyList<string> _desiredAlarmRefs = Array.Empty<string>();
|
|
|
|
/// <summary>The active native-alarm subscription handle for an <see cref="IAlarmSource"/> driver, or
|
|
/// null when none is established. Reset on <see cref="DetachAlarmSource"/> so the next Connected entry
|
|
/// re-subscribes against the freshly re-initialised driver; the null check makes the subscribe
|
|
/// idempotent across repeated <see cref="SetDesiredSubscriptions"/> pushes.</summary>
|
|
private IAlarmSubscriptionHandle? _alarmSubscriptionHandle;
|
|
|
|
/// <summary>
|
|
/// Gets or sets the timer scheduler for scheduling reconnection attempts.
|
|
/// </summary>
|
|
public ITimerScheduler Timers { get; set; } = null!;
|
|
|
|
/// <summary>
|
|
/// Creates a Props object for instantiating a <see cref="DriverInstanceActor"/>.
|
|
/// </summary>
|
|
/// <param name="driver">The driver instance to wrap.</param>
|
|
/// <param name="reconnectInterval">Optional interval for reconnection attempts; defaults to 10 seconds.</param>
|
|
/// <param name="startStubbed">If true, the actor starts in stub mode for testing or unavailable platforms.</param>
|
|
/// <param name="healthPublisher">Optional health publisher; defaults to <see cref="NullDriverHealthPublisher"/> so tests and
|
|
/// stub paths don't need to provide one.</param>
|
|
/// <param name="clusterId">Optional cluster identifier forwarded in <see cref="DriverHealthChanged"/> messages;
|
|
/// defaults to an empty string when not provided (e.g. in unit tests).</param>
|
|
/// <param name="invoker">Optional Phase 6.1 resilience invoker wrapping this driver's capability
|
|
/// calls; defaults to <see cref="NullDriverCapabilityInvoker"/> (pass-through) when not supplied.</param>
|
|
/// <param name="healthPollInterval">Optional period of the health-poll heartbeat, which also drives the
|
|
/// subscription reconcile (<see cref="ReconcileSubscription"/>); defaults to
|
|
/// <see cref="HealthPollInterval"/>. Exists so a test can drive the reconcile without waiting 30 s.</param>
|
|
/// <returns>A <see cref="Akka.Actor.Props"/> instance configured to create the wrapped <see cref="DriverInstanceActor"/>.</returns>
|
|
public static Props Props(
|
|
IDriver driver,
|
|
TimeSpan? reconnectInterval = null,
|
|
bool startStubbed = false,
|
|
IDriverHealthPublisher? healthPublisher = null,
|
|
string? clusterId = null,
|
|
IDriverCapabilityInvoker? invoker = null,
|
|
TimeSpan? healthPollInterval = null) =>
|
|
Akka.Actor.Props.Create(() => new DriverInstanceActor(
|
|
driver,
|
|
reconnectInterval ?? DefaultReconnectInterval,
|
|
startStubbed,
|
|
healthPublisher ?? NullDriverHealthPublisher.Instance,
|
|
clusterId ?? string.Empty,
|
|
invoker,
|
|
healthPollInterval));
|
|
|
|
/// <summary>
|
|
/// Returns true when the driver should boot in DEV-STUB mode based on host platform and
|
|
/// configured roles. Only the legacy v1 in-process <c>"Galaxy"</c> type stays Windows-only:
|
|
/// the legacy MXAccess COM proxy (retired in PR 7.2; gated for any leftover DriverInstance
|
|
/// rows that still reference the old type name).
|
|
/// The v2 <c>"GalaxyMxGateway"</c> driver talks gRPC to an external mxaccessgw process,
|
|
/// so it runs on any platform .NET 10 supports — Linux containers included. Not stubbed.
|
|
/// </summary>
|
|
/// <param name="driverType">The type identifier of the driver.</param>
|
|
/// <param name="roles">Operational roles configured for this instance.</param>
|
|
/// <returns>True if the driver should start in DEV-STUB mode; otherwise false.</returns>
|
|
public static bool ShouldStub(string driverType, IEnumerable<string> roles)
|
|
{
|
|
var isWindowsOnly = driverType is "Galaxy";
|
|
if (!OperatingSystem.IsWindows() && isWindowsOnly) return true;
|
|
if (roles.Contains("dev") && isWindowsOnly) return true;
|
|
return false;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Initializes a new instance of the <see cref="DriverInstanceActor"/> class.
|
|
/// </summary>
|
|
/// <param name="driver">The driver instance to wrap and manage.</param>
|
|
/// <param name="reconnectInterval">Interval between reconnection attempts.</param>
|
|
/// <param name="startStubbed">If true, start in stub mode for testing or unavailable platforms.</param>
|
|
/// <param name="healthPublisher">Sink for health-change notifications; must not be null.</param>
|
|
/// <param name="clusterId">Cluster identifier forwarded in health snapshots.</param>
|
|
/// <param name="invoker">Phase 6.1 resilience invoker wrapping this driver's capability calls;
|
|
/// defaults to <see cref="NullDriverCapabilityInvoker"/> (pass-through) when null.</param>
|
|
/// <param name="healthPollInterval">Period of the health-poll heartbeat, which also drives the
|
|
/// subscription reconcile; defaults to <see cref="HealthPollInterval"/> when null.</param>
|
|
public DriverInstanceActor(
|
|
IDriver driver,
|
|
TimeSpan reconnectInterval,
|
|
bool startStubbed = false,
|
|
IDriverHealthPublisher? healthPublisher = null,
|
|
string? clusterId = null,
|
|
IDriverCapabilityInvoker? invoker = null,
|
|
TimeSpan? healthPollInterval = null)
|
|
{
|
|
_driver = driver;
|
|
_invoker = invoker ?? NullDriverCapabilityInvoker.Instance;
|
|
_driverInstanceId = driver.DriverInstanceId;
|
|
_clusterId = clusterId ?? string.Empty;
|
|
_healthPublisher = healthPublisher ?? NullDriverHealthPublisher.Instance;
|
|
_reconnectInterval = reconnectInterval;
|
|
_healthPollInterval = healthPollInterval ?? HealthPollInterval;
|
|
OtOpcUaTelemetry.DriverInstanceLifecycle.Add(1,
|
|
new KeyValuePair<string, object?>("event", startStubbed ? "spawn_stub" : "spawn"),
|
|
new KeyValuePair<string, object?>("driver_type", driver.DriverType));
|
|
if (startStubbed)
|
|
{
|
|
Context.GetLogger().Info("[DEV-STUB] driver={Name} type={Type}",
|
|
_driverInstanceId, driver.DriverType);
|
|
Become(Stubbed);
|
|
}
|
|
else
|
|
{
|
|
Become(Connecting);
|
|
}
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
protected override void PreStart()
|
|
{
|
|
// Warm up the snapshot store immediately so AdminUI sees current state as soon as the
|
|
// actor starts, before any state transition fires. Also start the periodic heartbeat so
|
|
// long-lived Healthy drivers keep their snapshot fresh for newly-joined SignalR clients.
|
|
// Attach the rediscovery signal before the first publish. Not per-connect: an IRediscoverable raise
|
|
// has no connection affinity, and a driver can observe a remote change while disconnected.
|
|
AttachRediscoverySource();
|
|
AttachHostStatusSource();
|
|
PublishHealthSnapshot();
|
|
Timers.StartPeriodicTimer("health-poll", HealthPollTick.Instance, _healthPollInterval);
|
|
}
|
|
|
|
private void Stubbed()
|
|
{
|
|
// Stubbed drivers accept the standard message contracts but return deterministic
|
|
// success without touching real hardware. Read returns null; Write succeeds.
|
|
Receive<InitializeRequested>(_ => { /* no-op */ });
|
|
Receive<ApplyDelta>(msg => Sender.Tell(new ApplyResult(true, "stubbed", msg.Correlation)));
|
|
Receive<WriteAttribute>(_ => Sender.Tell(new WriteAttributeResult(true, "stubbed")));
|
|
// Stubbed drivers have no upstream alarm system — swallow the ack (it's fire-and-forget, no reply).
|
|
Receive<RouteAlarmAck>(_ => { /* stubbed drivers have no alarm backend */ });
|
|
Receive<DisconnectObserved>(_ => { /* stubbed drivers don't disconnect */ });
|
|
Receive<ForceReconnect>(_ => { /* stubbed drivers don't reconnect */ });
|
|
Receive<SetDesiredSubscriptions>(StoreDesiredSubscriptions);
|
|
// Stubbed drivers never enter Connected, so they never kick discovery; swallow defensively in case a
|
|
// re-discovery self-tick is ever routed here so it doesn't surface as an Akka Unhandled message.
|
|
Receive<RediscoveryRaised>(HandleRediscoveryRaised);
|
|
Receive<HostStatusRaised>(_ => PublishHealthSnapshot());
|
|
Receive<HealthPollTick>(_ => PublishHealthSnapshot());
|
|
}
|
|
|
|
private void Connecting()
|
|
{
|
|
Receive<InitializeRequested>(msg => InitializeAsync(msg.DriverConfigJson));
|
|
// Fast-fail writes while still connecting — without this the inbound WriteAttribute dead-letters
|
|
// and DriverHostActor.HandleRouteNodeWrite waits its full 8s Ask before reporting a generic
|
|
// "write timeout". Synchronous Receive: Sender.Tell on the actor thread is safe.
|
|
Receive<WriteAttribute>(_ =>
|
|
Sender.Tell(new WriteAttributeResult(false, "driver not connected")));
|
|
// An ack arriving while still connecting can't reach the upstream alarm system; drop it (the ack is
|
|
// fire-and-forget — no reply to surface — and the OPC UA condition state already committed locally).
|
|
Receive<RouteAlarmAck>(_ =>
|
|
_log.Debug("DriverInstance {Id}: alarm ack arrived during connect — dropped (driver not connected)", _driverInstanceId));
|
|
Receive<ApplyDelta>(AdoptConfigDuringInit);
|
|
Receive<InitializeSucceeded>(msg =>
|
|
{
|
|
if (msg.Generation != _initGeneration) return;
|
|
_log.Info("DriverInstance {Id}: connected", _driverInstanceId);
|
|
Become(Connected);
|
|
PublishHealthSnapshot();
|
|
ResubscribeDesired();
|
|
AttachAlarmSource();
|
|
SubscribeDesiredAlarms();
|
|
});
|
|
Receive<InitializeFailed>(msg =>
|
|
{
|
|
if (msg.Generation != _initGeneration) return;
|
|
_log.Warning("DriverInstance {Id}: initialize failed: {Reason}", _driverInstanceId, msg.Reason);
|
|
RecordFault();
|
|
Become(Reconnecting);
|
|
PublishHealthSnapshot();
|
|
});
|
|
Receive<SetDesiredSubscriptions>(StoreDesiredSubscriptions);
|
|
Receive<ForceReconnect>(_ => { /* already connecting — no-op */ });
|
|
// ResubscribeDesired self-Tells Subscribe; HandleSubscribeAsync replies SubscriptionEstablished to the
|
|
// sender, which on the self-resubscribe path is Self. Swallow it (trace only) so it doesn't dead-letter.
|
|
Receive<SubscriptionEstablished>(msg =>
|
|
_log.Debug("DriverInstance {Id}: subscription established ({Count} refs, {Diag})",
|
|
_driverInstanceId, msg.ReferenceCount, msg.DiagnosticId));
|
|
// Symmetric to the SubscriptionEstablished swallow: a failed self-resubscribe replies SubscriptionFailed
|
|
// to Self (HandleSubscribeAsync already logged the underlying cause). Swallow it so it doesn't dead-letter.
|
|
Receive<SubscriptionFailed>(msg =>
|
|
_log.Debug("DriverInstance {Id}: resubscribe reported failure: {Reason}", _driverInstanceId, msg.Reason));
|
|
// A native alarm transition can race in while still (re)connecting (the driver's feed runs on its
|
|
// own thread); drop it — the feed re-delivers active alarms once Connected. Trace only.
|
|
Receive<NativeAlarmRaised>(_ =>
|
|
_log.Debug("DriverInstance {Id}: native alarm arrived during connect — dropped (feed re-delivers)", _driverInstanceId));
|
|
// A SubscribeAlarms self-tell (from Connected) can be overtaken by an already-queued disconnect into
|
|
// this state; swallow it so it doesn't dead-letter — the next Connected entry re-subscribes.
|
|
Receive<SubscribeAlarms>(_ => { });
|
|
Receive<RediscoveryRaised>(HandleRediscoveryRaised);
|
|
Receive<HostStatusRaised>(_ => PublishHealthSnapshot());
|
|
Receive<HealthPollTick>(_ => PublishHealthSnapshot());
|
|
}
|
|
|
|
private void Connected()
|
|
{
|
|
// #477 — announce connectivity to the host so it can clear any comms-lost Quality annotation on this
|
|
// driver's native alarm conditions (Bad → Good). Fire-and-forget; the host defaults conditions to a
|
|
// non-Good "waiting for initial data" quality at materialise, and this is what confirms them Good.
|
|
Context.Parent.Tell(new ConnectivityChanged(_driverInstanceId, Connected: true));
|
|
|
|
ReceiveAsync<ApplyDelta>(HandleApplyDeltaAsync);
|
|
Receive<DisconnectObserved>(msg =>
|
|
{
|
|
_log.Warning("DriverInstance {Id}: disconnect observed ({Reason}); reconnecting",
|
|
_driverInstanceId, msg.Reason);
|
|
DetachSubscription();
|
|
RecordFault();
|
|
Become(Reconnecting);
|
|
PublishHealthSnapshot();
|
|
});
|
|
Receive<ForceReconnect>(_ =>
|
|
{
|
|
_log.Info("DriverInstance {Id}: ForceReconnect requested by admin; re-entering Reconnecting", _driverInstanceId);
|
|
DetachSubscription();
|
|
Become(Reconnecting);
|
|
PublishHealthSnapshot();
|
|
});
|
|
ReceiveAsync<WriteAttribute>(HandleWriteAsync);
|
|
ReceiveAsync<RouteAlarmAck>(HandleAcknowledgeAsync);
|
|
ReceiveAsync<Subscribe>(HandleSubscribeAsync);
|
|
ReceiveAsync<Unsubscribe>(_ => UnsubscribeAsync());
|
|
Receive<SetDesiredSubscriptions>(msg =>
|
|
{
|
|
StoreDesiredSubscriptions(msg);
|
|
if (_desiredRefs.Count > 0) Self.Tell(new Subscribe(_desiredRefs, _desiredInterval));
|
|
else if (_subscriptionHandle is not null) Self.Tell(new Unsubscribe());
|
|
// Native-alarm analogue: un-gate the IAlarmSource feed when alarm tags are (now) present. The
|
|
// common live path — a deploy delivers SetDesiredSubscriptions while the driver is already
|
|
// Connected — flows through HERE, so the alarm subscribe must happen on this message, not only
|
|
// on Connected entry. Unlike the value path above, there is deliberately no unsubscribe-on-empty:
|
|
// a removed alarm tag's condition node is torn down on the address-space rebuild and
|
|
// DriverHostActor.ForwardNativeAlarm drops any transition whose ConditionId no longer maps, so a
|
|
// lingering session-less alarm subscription can never surface a removed alarm.
|
|
SubscribeDesiredAlarms();
|
|
});
|
|
ReceiveAsync<SubscribeAlarms>(HandleSubscribeAlarmsAsync);
|
|
Receive<DataChangeForward>(OnDataChangeForward);
|
|
// Native alarm transition marshaled onto the actor thread from the driver's OnAlarmEvent;
|
|
// project it to the parent the same way DataChangeForward projects AttributeValuePublished.
|
|
Receive<NativeAlarmRaised>(m => Context.Parent.Tell(new AttributeAlarmPublished(_driverInstanceId, m.Args)));
|
|
// ResubscribeDesired self-Tells Subscribe; HandleSubscribeAsync replies SubscriptionEstablished to the
|
|
// sender, which on the self-resubscribe path is Self. Swallow it (trace only) so it doesn't dead-letter.
|
|
Receive<SubscriptionEstablished>(msg =>
|
|
_log.Debug("DriverInstance {Id}: subscription established ({Count} refs, {Diag})",
|
|
_driverInstanceId, msg.ReferenceCount, msg.DiagnosticId));
|
|
// Symmetric to the SubscriptionEstablished swallow: a failed self-resubscribe replies SubscriptionFailed
|
|
// to Self (HandleSubscribeAsync already logged the underlying cause). Swallow it so it doesn't dead-letter.
|
|
Receive<SubscriptionFailed>(msg =>
|
|
_log.Debug("DriverInstance {Id}: resubscribe reported failure: {Reason}", _driverInstanceId, msg.Reason));
|
|
Receive<RediscoveryRaised>(HandleRediscoveryRaised);
|
|
Receive<HostStatusRaised>(_ => PublishHealthSnapshot());
|
|
Receive<HealthPollTick>(_ =>
|
|
{
|
|
PublishHealthSnapshot();
|
|
ReconcileSubscription();
|
|
});
|
|
}
|
|
|
|
/// <summary>
|
|
/// Periodic desired-vs-actual subscription reconcile (#488), riding the existing <c>health-poll</c>
|
|
/// tick. While Connected, a driver that has a desired subscription set but no live handle is
|
|
/// re-subscribed.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// <para>
|
|
/// The desired set is otherwise only (re)applied on a <b>Connected transition</b>
|
|
/// (<see cref="ResubscribeDesired"/>) or on a deploy's <see cref="SetDesiredSubscriptions"/>.
|
|
/// Nothing re-asserted it for a subscription that was lost while the driver <i>stayed</i>
|
|
/// Connected — a failed subscribe whose retry never came, or a handle dropped down an error
|
|
/// path. That is the hole this closes.
|
|
/// </para>
|
|
/// <para>
|
|
/// <b>This is a state-machine consistency check, not a probe.</b> It can only see that this
|
|
/// actor holds no handle; it cannot detect a handle that is present but dead server-side,
|
|
/// because <c>ISubscriptionHandle</c> is opaque (a <c>DiagnosticId</c> string and nothing
|
|
/// else). Detecting that needs a liveness member on <c>ISubscribable</c> implemented across
|
|
/// all eight drivers, each with different subscription mechanics — deliberately deferred
|
|
/// until a real occurrence shows the handle outliving the subscription.
|
|
/// </para>
|
|
/// <para>
|
|
/// Gated on <see cref="ISubscribable"/>: a driver that cannot subscribe at all may still have
|
|
/// been handed a desired set, and re-telling <c>Subscribe</c> to it every tick would fail
|
|
/// every tick, forever.
|
|
/// </para>
|
|
/// <para>
|
|
/// A redundant <c>Subscribe</c> is possible but harmless: a tick sitting in the mailbox ahead
|
|
/// of an already-queued <c>Subscribe</c> observes the same "no handle" state and tells a
|
|
/// second one. <see cref="HandleSubscribeAsync"/> is a <c>ReceiveAsync</c> (so no tick can be
|
|
/// processed mid-subscribe) and drops any prior subscription before establishing the new one,
|
|
/// so the duplicate costs one extra round-trip and converges. Suppressing it would need a
|
|
/// pending-subscribe flag threaded through every self-tell site — more state than the race
|
|
/// is worth.
|
|
/// </para>
|
|
/// </remarks>
|
|
private void ReconcileSubscription()
|
|
{
|
|
if (_driver is not ISubscribable) return;
|
|
if (_desiredRefs.Count == 0 || _subscriptionHandle is not null) return;
|
|
|
|
_log.Info(
|
|
"DriverInstance {Id}: connected with {Count} desired subscription ref(s) but no live subscription handle — re-subscribing (periodic reconcile)",
|
|
_driverInstanceId, _desiredRefs.Count);
|
|
Self.Tell(new Subscribe(_desiredRefs, _desiredInterval));
|
|
}
|
|
|
|
private void Reconnecting()
|
|
{
|
|
// #477 — announce comms loss to the host so it annotates this driver's native alarm conditions Bad
|
|
// (a comms-lost driver emits no alarm events, so this is the ONLY signal that the source is unreachable).
|
|
Context.Parent.Tell(new ConnectivityChanged(_driverInstanceId, Connected: false));
|
|
|
|
Receive<RetryConnect>(_ => InitializeAsync(_currentConfigJson ?? "{}"));
|
|
// Fast-fail writes while reconnecting (same reason as Connecting — avoids the 8s host Ask
|
|
// timeout on an inbound write to a transiently-down driver). Synchronous Receive.
|
|
Receive<WriteAttribute>(_ =>
|
|
Sender.Tell(new WriteAttributeResult(false, "driver not connected")));
|
|
// An ack arriving while reconnecting can't reach the upstream alarm system; drop it (fire-and-forget,
|
|
// no reply — the OPC UA condition state already committed locally on the Part 9 ack).
|
|
Receive<RouteAlarmAck>(_ =>
|
|
_log.Debug("DriverInstance {Id}: alarm ack arrived during reconnect — dropped (driver not connected)", _driverInstanceId));
|
|
Receive<ApplyDelta>(AdoptConfigDuringInit);
|
|
Receive<InitializeSucceeded>(msg =>
|
|
{
|
|
if (msg.Generation != _initGeneration) return;
|
|
Timers.Cancel("retry-connect");
|
|
_log.Info("DriverInstance {Id}: reconnected", _driverInstanceId);
|
|
Become(Connected);
|
|
PublishHealthSnapshot();
|
|
ResubscribeDesired();
|
|
AttachAlarmSource();
|
|
SubscribeDesiredAlarms();
|
|
});
|
|
// A failure here is a no-op regardless of generation — the retry timer keeps trying the
|
|
// current config; only a (generation-matched) InitializeSucceeded transitions state.
|
|
Receive<InitializeFailed>(_ => { /* keep retrying via timer */ });
|
|
Receive<SetDesiredSubscriptions>(StoreDesiredSubscriptions);
|
|
Receive<ForceReconnect>(_ => { /* already reconnecting — no-op */ });
|
|
// ResubscribeDesired self-Tells Subscribe; HandleSubscribeAsync replies SubscriptionEstablished to the
|
|
// sender, which on the self-resubscribe path is Self. Swallow it (trace only) so it doesn't dead-letter.
|
|
Receive<SubscriptionEstablished>(msg =>
|
|
_log.Debug("DriverInstance {Id}: subscription established ({Count} refs, {Diag})",
|
|
_driverInstanceId, msg.ReferenceCount, msg.DiagnosticId));
|
|
// Symmetric to the SubscriptionEstablished swallow: a failed self-resubscribe replies SubscriptionFailed
|
|
// to Self (HandleSubscribeAsync already logged the underlying cause). Swallow it so it doesn't dead-letter.
|
|
Receive<SubscriptionFailed>(msg =>
|
|
_log.Debug("DriverInstance {Id}: resubscribe reported failure: {Reason}", _driverInstanceId, msg.Reason));
|
|
// A native alarm transition can race in while still reconnecting (the driver's feed runs on its
|
|
// own thread); drop it — the feed re-delivers active alarms once Connected. Trace only.
|
|
Receive<NativeAlarmRaised>(_ =>
|
|
_log.Debug("DriverInstance {Id}: native alarm arrived during reconnect — dropped (feed re-delivers)", _driverInstanceId));
|
|
// A SubscribeAlarms self-tell (from Connected) can be overtaken by an already-queued disconnect into
|
|
// this state; swallow it so it doesn't dead-letter — the next Connected entry re-subscribes.
|
|
Receive<SubscribeAlarms>(_ => { });
|
|
Receive<RediscoveryRaised>(HandleRediscoveryRaised);
|
|
Receive<HostStatusRaised>(_ => PublishHealthSnapshot());
|
|
Receive<HealthPollTick>(_ => PublishHealthSnapshot());
|
|
Timers.StartPeriodicTimer("retry-connect", RetryConnect.Instance, _reconnectInterval);
|
|
}
|
|
|
|
private void InitializeAsync(string driverConfigJson)
|
|
{
|
|
_currentConfigJson = driverConfigJson;
|
|
var generation = ++_initGeneration;
|
|
var self = Self;
|
|
_ = Task.Run(async () =>
|
|
{
|
|
try
|
|
{
|
|
await _driver.InitializeAsync(driverConfigJson, CancellationToken.None);
|
|
self.Tell(new InitializeSucceeded(generation));
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
self.Tell(new InitializeFailed(ex.Message, generation));
|
|
}
|
|
});
|
|
}
|
|
|
|
/// <summary>Adopt a new config while not connected: ApplyDelta in Connecting/Reconnecting re-inits
|
|
/// immediately with the new config. <see cref="InitializeAsync"/> swaps <c>_currentConfigJson</c> and
|
|
/// bumps the generation, so the in-flight (old-config) init is superseded and its result is dropped.
|
|
/// The actor stays in its current state; the new init's result drives the next transition. In
|
|
/// Reconnecting the retry timer is left running — if this immediate attempt fails it keeps retrying
|
|
/// the new config (a redundant concurrent attempt is deduped by the generation guard).</summary>
|
|
private void AdoptConfigDuringInit(ApplyDelta msg)
|
|
{
|
|
_log.Info("DriverInstance {Id}: ApplyDelta during (re)connect — adopting new config, re-initialising now",
|
|
_driverInstanceId);
|
|
InitializeAsync(msg.DriverConfigJson);
|
|
Sender.Tell(new ApplyResult(true, "config adopted; reinitializing", msg.Correlation));
|
|
}
|
|
|
|
private async Task HandleApplyDeltaAsync(ApplyDelta msg)
|
|
{
|
|
var replyTo = Sender;
|
|
try
|
|
{
|
|
await _driver.ReinitializeAsync(msg.DriverConfigJson, CancellationToken.None);
|
|
_currentConfigJson = msg.DriverConfigJson;
|
|
// The driver may have rebuilt its dependency-consumer interest during reinit (the Calculation
|
|
// driver re-binds its calc tags + dependency refs). Notify the parent so it re-registers this
|
|
// driver's mux adapter AFTER reinit completed — a synchronous re-register at ApplyDelta-dispatch
|
|
// time would re-read the driver's STALE DependencyRefs. Parent Tell (fire-and-forget); the
|
|
// continuation resumes on the actor context (no ConfigureAwait) so Context.Parent is live.
|
|
Context.Parent.Tell(new DeltaApplied(_driverInstanceId));
|
|
replyTo.Tell(new ApplyResult(true, null, msg.Correlation));
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
replyTo.Tell(new ApplyResult(false, ex.Message, msg.Correlation));
|
|
}
|
|
}
|
|
|
|
private async Task HandleWriteAsync(WriteAttribute msg)
|
|
{
|
|
var replyTo = Sender;
|
|
if (_driver is not IWritable writable)
|
|
{
|
|
replyTo.Tell(new WriteAttributeResult(false, "Driver does not implement IWritable"));
|
|
return;
|
|
}
|
|
|
|
var request = new[] { new WriteRequest(msg.TagId, msg.Value) };
|
|
// Bound the write so a hung backend can't pin this actor forever — retry stays
|
|
// off by default, but a stalled call still needs an answer.
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
|
try
|
|
{
|
|
// Route through the resilience invoker (Write pipeline: timeout, per-host breaker). Writes
|
|
// default to NON-idempotent (03/S12): a timed-out write may already have committed at the
|
|
// device, and replaying a command-shaped write (pulse coil, counter increment, Galaxy
|
|
// supervisory write) can duplicate an irreversible field action. The invoker's no-retry arm is
|
|
// therefore authoritative regardless of any configured Write retry. Per-tag opt-in to retries via
|
|
// WriteIdempotentAttribute is the unshipped forward path (see WriteIdempotentAttribute). The outer
|
|
// cts is retained as a hard backstop above the (typically tighter) tier timeout.
|
|
var results = await _invoker.ExecuteWriteAsync(
|
|
ResolveHost(msg.TagId),
|
|
isIdempotent: false,
|
|
async ct => await writable.WriteAsync(request, ct),
|
|
cts.Token);
|
|
if (results is { Count: 1 } && IsGoodStatus(results[0].StatusCode))
|
|
{
|
|
replyTo.Tell(new WriteAttributeResult(true, null));
|
|
return;
|
|
}
|
|
var status = results is { Count: > 0 } ? results[0].StatusCode : 0xFFFFFFFF;
|
|
replyTo.Tell(new WriteAttributeResult(false, $"StatusCode=0x{status:X8}"));
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
replyTo.Tell(new WriteAttributeResult(false, "write timeout"));
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
replyTo.Tell(new WriteAttributeResult(false, ex.Message));
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Forwards an inbound native-condition acknowledge (routed by <see cref="DriverHostActor"/> from a
|
|
/// resolved condition NodeId) to the driver's <see cref="IAlarmSource.AcknowledgeAsync"/>. The driver
|
|
/// correlates on <see cref="AlarmAcknowledgeRequest.ConditionId"/> (= the authored alarm
|
|
/// full-reference); <see cref="AlarmAcknowledgeRequest.SourceNodeId"/> carries the same reference (the
|
|
/// driver's ack path keys on ConditionId). Bounded to 5s so a hung backend can't pin this actor.
|
|
/// Fire-and-forget — the OPC UA Part 9 ack already committed the local condition state and
|
|
/// <see cref="IAlarmSource.AcknowledgeAsync"/> returns no per-condition status — so there is no reply;
|
|
/// a failure is logged and dropped (the local condition stays Acknowledged regardless).
|
|
/// </summary>
|
|
private async Task HandleAcknowledgeAsync(RouteAlarmAck msg)
|
|
{
|
|
if (_driver is not IAlarmSource src)
|
|
{
|
|
_log.Warning("DriverInstance {Id}: alarm ack dropped — driver does not implement IAlarmSource", _driverInstanceId);
|
|
return;
|
|
}
|
|
|
|
var request = new[]
|
|
{
|
|
new AlarmAcknowledgeRequest(
|
|
SourceNodeId: msg.ConditionId,
|
|
ConditionId: msg.ConditionId,
|
|
Comment: msg.Comment,
|
|
OperatorUser: msg.OperatorUser),
|
|
};
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
|
try
|
|
{
|
|
// Ack is a write-shaped operation — the AlarmAcknowledge tier policy never retries.
|
|
await _invoker.ExecuteAsync(
|
|
DriverCapability.AlarmAcknowledge,
|
|
ResolveHost(msg.ConditionId),
|
|
async ct => await src.AcknowledgeAsync(request, ct),
|
|
cts.Token);
|
|
_log.Info("DriverInstance {Id}: acknowledged native condition {Cond} by {User}",
|
|
_driverInstanceId, msg.ConditionId, msg.OperatorUser);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_log.Warning(ex, "DriverInstance {Id}: native-alarm acknowledge of {Cond} failed",
|
|
_driverInstanceId, msg.ConditionId);
|
|
}
|
|
}
|
|
|
|
private async Task HandleSubscribeAsync(Subscribe msg)
|
|
{
|
|
// Capture Sender/Self BEFORE any await. The re-subscribe path below awaits
|
|
// UnsubscribeAsync, and a real async backend can resume the continuation off Akka's
|
|
// ActorContext — reading raw Sender/Self/Context past that point throws
|
|
// NotSupportedException ("no active ActorContext"). Keep ConfigureAwait off the awaits
|
|
// in this handler so continuations resume on the actor context.
|
|
var replyTo = Sender;
|
|
var self = Self;
|
|
if (_driver is not ISubscribable subscribable)
|
|
{
|
|
replyTo.Tell(new SubscriptionFailed("Driver does not implement ISubscribable"));
|
|
return;
|
|
}
|
|
if (_subscriptionHandle is not null)
|
|
{
|
|
// Subscribe-twice — drop the prior subscription before establishing the new one.
|
|
await UnsubscribeAsync();
|
|
}
|
|
|
|
try
|
|
{
|
|
_dataChangeHandler = (_, args) => self.Tell(new DataChangeForward(args.FullReference, args.Snapshot));
|
|
subscribable.OnDataChange += _dataChangeHandler;
|
|
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
|
|
// A bulk subscribe spans potentially many hosts, so key the pipeline on the driver
|
|
// instance (single-host fallback) rather than any one reference's host.
|
|
_subscriptionHandle = await _invoker.ExecuteAsync(
|
|
DriverCapability.Subscribe,
|
|
_driverInstanceId,
|
|
async ct => await subscribable.SubscribeAsync(msg.FullReferences, msg.PublishingInterval, ct),
|
|
cts.Token);
|
|
|
|
replyTo.Tell(new SubscriptionEstablished(_subscriptionHandle.DiagnosticId, msg.FullReferences.Count));
|
|
_log.Info("DriverInstance {Id}: subscribed to {Count} refs ({Diag})",
|
|
_driverInstanceId, msg.FullReferences.Count, _subscriptionHandle.DiagnosticId);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
DetachSubscription();
|
|
_log.Warning(ex, "DriverInstance {Id}: subscribe failed", _driverInstanceId);
|
|
replyTo.Tell(new SubscriptionFailed(ex.Message));
|
|
}
|
|
}
|
|
|
|
private async Task UnsubscribeAsync()
|
|
{
|
|
if (_driver is not ISubscribable subscribable || _subscriptionHandle is null)
|
|
{
|
|
DetachSubscription();
|
|
return;
|
|
}
|
|
try
|
|
{
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
|
var handle = _subscriptionHandle; // non-null per the guard above; capture before the async lambda
|
|
await _invoker.ExecuteAsync(
|
|
DriverCapability.Subscribe,
|
|
_driverInstanceId,
|
|
async ct => await subscribable.UnsubscribeAsync(handle, ct),
|
|
cts.Token);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_log.Warning(ex, "DriverInstance {Id}: unsubscribe threw (continuing)", _driverInstanceId);
|
|
}
|
|
finally
|
|
{
|
|
DetachSubscription();
|
|
}
|
|
}
|
|
|
|
/// <summary>Tear down the data-change + native-alarm event handlers + null the handle. Called from the
|
|
/// Unsubscribe path, on PostStop, and on Connected → Reconnecting transitions so a stale handler doesn't
|
|
/// push data-change / alarm events to an actor that has lost its driver connection.</summary>
|
|
private void DetachSubscription()
|
|
{
|
|
if (_driver is ISubscribable subscribable && _dataChangeHandler is not null)
|
|
{
|
|
subscribable.OnDataChange -= _dataChangeHandler;
|
|
}
|
|
_dataChangeHandler = null;
|
|
_subscriptionHandle = null;
|
|
DetachAlarmSource();
|
|
}
|
|
|
|
/// <summary>Subscribe the driver's native alarm event (if it is an <see cref="IAlarmSource"/>),
|
|
/// marshaling each transition to the actor thread. Idempotent; mirrors the OnDataChange attach.</summary>
|
|
private void AttachAlarmSource()
|
|
{
|
|
if (_driver is not IAlarmSource src || _alarmEventHandler is not null) return;
|
|
var self = Self;
|
|
_alarmEventHandler = (_, e) => self.Tell(new NativeAlarmRaised(e));
|
|
src.OnAlarmEvent += _alarmEventHandler;
|
|
}
|
|
|
|
/// <summary>Subscribe the driver's <see cref="IRediscoverable.OnRediscoveryNeeded"/> (if it is one),
|
|
/// marshaling each raise to the actor thread. Idempotent; mirrors <see cref="AttachAlarmSource"/>.
|
|
/// <para>Attached once in <c>PreStart</c> rather than per-connect, because the interesting raises
|
|
/// (a Galaxy redeploy, a TwinCAT symbol-version bump) can happen while the driver is between
|
|
/// connects, and the event carries no connection affinity.</para></summary>
|
|
private void AttachRediscoverySource()
|
|
{
|
|
if (_driver is not IRediscoverable src || _rediscoveryHandler is not null) return;
|
|
var self = Self;
|
|
_rediscoveryHandler = (_, e) => self.Tell(new RediscoveryRaised(e));
|
|
src.OnRediscoveryNeeded += _rediscoveryHandler;
|
|
}
|
|
|
|
/// <summary>Symmetric teardown, called from PostStop. Load-bearing: the <see cref="IDriver"/> instance
|
|
/// can OUTLIVE this actor (the host respawns a child around the same driver object), so a missing
|
|
/// unsubscribe would accumulate one handler per respawn, each holding a dead <c>Self</c>.</summary>
|
|
private void DetachRediscoverySource()
|
|
{
|
|
if (_driver is IRediscoverable src && _rediscoveryHandler is not null)
|
|
src.OnRediscoveryNeeded -= _rediscoveryHandler;
|
|
_rediscoveryHandler = null;
|
|
}
|
|
|
|
/// <summary>Subscribe the driver's <see cref="IHostConnectivityProbe.OnHostStatusChanged"/> (if it is
|
|
/// one), marshaling each transition to the actor thread. Idempotent; mirrors
|
|
/// <see cref="AttachRediscoverySource"/>, including the PreStart-not-per-connect placement — a probe
|
|
/// loop can report a host down while the driver itself is between connects.</summary>
|
|
private void AttachHostStatusSource()
|
|
{
|
|
if (_driver is not IHostConnectivityProbe src || _hostStatusHandler is not null) return;
|
|
var self = Self;
|
|
_hostStatusHandler = (_, _) => self.Tell(new HostStatusRaised());
|
|
src.OnHostStatusChanged += _hostStatusHandler;
|
|
}
|
|
|
|
/// <summary>Symmetric teardown, called from PostStop — same leak argument as
|
|
/// <see cref="DetachRediscoverySource"/>.</summary>
|
|
private void DetachHostStatusSource()
|
|
{
|
|
if (_driver is IHostConnectivityProbe src && _hostStatusHandler is not null)
|
|
src.OnHostStatusChanged -= _hostStatusHandler;
|
|
_hostStatusHandler = null;
|
|
}
|
|
|
|
/// <summary>Current per-host connectivity, or null when the driver is not an
|
|
/// <see cref="IHostConnectivityProbe"/>. Pulled fresh rather than cached: the driver owns the host set,
|
|
/// and a stale local copy would be a second source of truth to keep in sync.
|
|
/// <para><c>GetHostStatuses()</c> is called UNGUARDED (not through <see cref="_invoker"/>) on purpose —
|
|
/// it is a pure in-memory snapshot with no I/O, which is exactly why
|
|
/// <c>UnwrappedCapabilityCallAnalyzer</c> exempts it. A driver that makes it do I/O breaks that
|
|
/// contract; the try/catch below keeps a misbehaving one from killing the health publish.</para></summary>
|
|
private IReadOnlyList<HostConnectivityStatus>? CurrentHostStatuses()
|
|
{
|
|
if (_driver is not IHostConnectivityProbe probe) return null;
|
|
try
|
|
{
|
|
return probe.GetHostStatuses();
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_log.Warning(ex, "DriverInstance {Id}: GetHostStatuses threw during health publish; omitting host detail", _driverInstanceId);
|
|
return null;
|
|
}
|
|
}
|
|
|
|
/// <summary>Records the driver's rediscovery raise and re-publishes health so the signal reaches the
|
|
/// AdminUI promptly rather than waiting for the next 30 s heartbeat.
|
|
/// <para><b>Advisory only.</b> The served address space is deliberately NOT rebuilt: v3 authors raw tags
|
|
/// explicitly through the <c>/raw</c> browse-commit flow, so a runtime graft would create nodes no
|
|
/// operator approved and that no deployment artifact records. This prompts a human to re-browse.</para></summary>
|
|
private void HandleRediscoveryRaised(RediscoveryRaised msg)
|
|
{
|
|
_rediscoveryNeededUtc = DateTime.UtcNow;
|
|
_rediscoveryReason = msg.Args.Reason;
|
|
_log.Info(
|
|
"DriverInstance {Id}: driver reports its tag set may have changed ({Reason}) — surfaced for operator re-browse; the served address space is unchanged",
|
|
_driverInstanceId, msg.Args.Reason);
|
|
PublishHealthSnapshot();
|
|
}
|
|
|
|
/// <summary>Symmetric teardown — called from <see cref="DetachSubscription"/> and PostStop so a stale
|
|
/// handler never pushes to a disconnected actor.</summary>
|
|
private void DetachAlarmSource()
|
|
{
|
|
if (_driver is IAlarmSource src && _alarmEventHandler is not null)
|
|
src.OnAlarmEvent -= _alarmEventHandler;
|
|
_alarmEventHandler = null;
|
|
// Drop our handle so the next Connected entry re-subscribes; the desired alarm refs persist across
|
|
// reconnects. NOTE: this does NOT tear down the driver-side subscription. For a session-bound
|
|
// IAlarmSource the old subscription dies with the session (no accumulation). For a session-less feed
|
|
// (GalaxyDriver's always-on central monitor) it survives an in-place reconnect, so the re-subscribe
|
|
// is additive — but the driver now collapses to a single live handle on each SubscribeAlarmsAsync
|
|
// (GalaxyDriver.SubscribeAlarmsAsync clears the set before adding), so handles no longer accumulate
|
|
// across reconnects. The gate (Count > 0) and the one-shot fan-out are unchanged.
|
|
_alarmSubscriptionHandle = null;
|
|
}
|
|
|
|
/// <summary>Establish the native-alarm subscription that un-gates an <see cref="IAlarmSource"/> driver's
|
|
/// feed — the driver suppresses <see cref="IAlarmSource.OnAlarmEvent"/> until at least one alarm
|
|
/// subscription exists. Self-sends the async <see cref="SubscribeAlarms"/> the Connected behaviour
|
|
/// handles. Idempotent: a no-op unless the driver is an <see cref="IAlarmSource"/>, alarm refs are
|
|
/// desired, and no subscription is yet established.</summary>
|
|
private void SubscribeDesiredAlarms()
|
|
{
|
|
if (_driver is IAlarmSource && _desiredAlarmRefs.Count > 0 && _alarmSubscriptionHandle is null)
|
|
Self.Tell(new SubscribeAlarms());
|
|
}
|
|
|
|
/// <summary>Calls the driver's <see cref="IAlarmSource.SubscribeAlarmsAsync"/> (bounded) to register the
|
|
/// alarm subscription that un-gates its native feed, caching the returned handle. Re-checks the guard
|
|
/// (the desired set may have cleared, or another SubscribeAlarms may have already established a handle,
|
|
/// while this was queued). Failures are logged and retried on the next Connected entry — the feed simply
|
|
/// stays gated until then.</summary>
|
|
private async Task HandleSubscribeAlarmsAsync(SubscribeAlarms _)
|
|
{
|
|
if (_driver is not IAlarmSource src || _desiredAlarmRefs.Count == 0 || _alarmSubscriptionHandle is not null)
|
|
return;
|
|
var refs = _desiredAlarmRefs;
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
|
|
try
|
|
{
|
|
_alarmSubscriptionHandle = await _invoker.ExecuteAsync(
|
|
DriverCapability.AlarmSubscribe,
|
|
_driverInstanceId,
|
|
async ct => await src.SubscribeAlarmsAsync(refs, ct),
|
|
cts.Token);
|
|
_log.Info("DriverInstance {Id}: native-alarm subscription established for {Count} alarm ref(s) ({Diag})",
|
|
_driverInstanceId, refs.Count, _alarmSubscriptionHandle.DiagnosticId);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_log.Warning(ex, "DriverInstance {Id}: native-alarm subscription failed — feed stays gated until reconnect",
|
|
_driverInstanceId);
|
|
}
|
|
}
|
|
|
|
|
|
/// <summary>Records the host's desired subscription set without touching the live subscription.
|
|
/// The set is (re)applied by <see cref="ResubscribeDesired"/> on the next <c>Connected</c> entry.</summary>
|
|
private void StoreDesiredSubscriptions(SetDesiredSubscriptions msg)
|
|
{
|
|
_desiredRefs = msg.FullReferences;
|
|
_desiredInterval = msg.PublishingInterval;
|
|
_desiredAlarmRefs = msg.AlarmReferences ?? Array.Empty<string>();
|
|
}
|
|
|
|
/// <summary>Re-establish the desired subscription after (re)connecting. Self-sends the one-shot
|
|
/// <see cref="Subscribe"/> the Connected behaviour already handles (which drops any prior handle
|
|
/// first), so values resume streaming after a reconnect or redeploy without host involvement.</summary>
|
|
private void ResubscribeDesired()
|
|
{
|
|
if (_desiredRefs.Count > 0)
|
|
{
|
|
Self.Tell(new Subscribe(_desiredRefs, _desiredInterval));
|
|
}
|
|
}
|
|
|
|
private void OnDataChangeForward(DataChangeForward msg)
|
|
{
|
|
var quality = QualityFromStatus(msg.Snapshot.StatusCode);
|
|
var ts = msg.Snapshot.SourceTimestampUtc ?? msg.Snapshot.ServerTimestampUtc;
|
|
Context.Parent.Tell(new AttributeValuePublished(_driverInstanceId, msg.FullReference, msg.Snapshot.Value, quality, ts));
|
|
}
|
|
|
|
/// <summary>Translate an OPC UA status code to the 3-state <see cref="OpcUaQuality"/> projection
|
|
/// the publish actor consumes. Severity bits (top 2): 00 = Good, 01 = Uncertain, 10/11 = Bad.</summary>
|
|
private static OpcUaQuality QualityFromStatus(uint statusCode)
|
|
{
|
|
var severity = statusCode >> 30;
|
|
return severity switch
|
|
{
|
|
0 => OpcUaQuality.Good,
|
|
1 => OpcUaQuality.Uncertain,
|
|
_ => OpcUaQuality.Bad,
|
|
};
|
|
}
|
|
|
|
private static bool IsGoodStatus(uint statusCode) => (statusCode >> 30) == 0;
|
|
|
|
/// <summary>Resolve the resilience-pipeline host key for a single driver-side reference. Multi-host
|
|
/// drivers (e.g. Modbus across N PLCs) implement <see cref="IPerCallHostResolver"/> so one dead host
|
|
/// trips only its own breaker; single-host drivers fall back to the driver-instance id. Used for the
|
|
/// single-reference capability calls (write, alarm ack); bulk/whole-driver calls key on the instance
|
|
/// id directly.</summary>
|
|
private string ResolveHost(string reference) =>
|
|
_driver is IPerCallHostResolver resolver ? resolver.ResolveHost(reference) : _driverInstanceId;
|
|
|
|
/// <summary>Records a transition into a Faulted / error state for the 5-minute sliding counter.</summary>
|
|
private void RecordFault()
|
|
{
|
|
_faultTimestamps.Enqueue(DateTime.UtcNow);
|
|
}
|
|
|
|
/// <summary>Returns how many fault transitions occurred in the last 5 minutes.</summary>
|
|
private int ErrorCount5Min()
|
|
{
|
|
var cutoff = DateTime.UtcNow.AddMinutes(-5);
|
|
while (_faultTimestamps.Count > 0 && _faultTimestamps.Peek() < cutoff)
|
|
_faultTimestamps.Dequeue();
|
|
return _faultTimestamps.Count;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Polls <see cref="IDriver.GetHealth"/> and forwards the snapshot to the health publisher.
|
|
/// Called on every observable state change and by the periodic <see cref="HealthPollTick"/>
|
|
/// so the AdminUI snapshot store is warmed up for newly-joined SignalR clients.
|
|
/// Deduplicates: if the resulting (state, lastSuccess, lastError, errorCount) tuple matches
|
|
/// the last publish, this call is a no-op. Stops flood-publishing identical Healthy snapshots
|
|
/// every 30s when nothing has changed. Newly-joined SignalR clients still get the current
|
|
/// snapshot via <c>DriverStatusHub.JoinDriver</c> which reads the store directly.
|
|
/// </summary>
|
|
private void PublishHealthSnapshot()
|
|
{
|
|
try
|
|
{
|
|
var health = _driver.GetHealth();
|
|
var errorCount = ErrorCount5Min();
|
|
var hostStatuses = CurrentHostStatuses();
|
|
// _rediscoveryNeededUtc is PART OF THE FINGERPRINT on purpose. A rediscovery raise on an
|
|
// otherwise-unchanged Healthy driver leaves (state, lastSuccess, lastError, errorCount)
|
|
// identical, so without it the dedup below would swallow the very publish that carries the
|
|
// signal and the operator would never see the prompt.
|
|
//
|
|
// The host-status digest is in the fingerprint for EXACTLY the same reason, and the failure
|
|
// mode is the more likely one of the two: a multi-device driver stays aggregate-Healthy when a
|
|
// single device drops, so every other fingerprint component is unchanged and the transition —
|
|
// the whole point of this channel — would be deduped away. It is a flattened STRING because a
|
|
// tuple holding IReadOnlyList compares by reference, which would never match and so would
|
|
// defeat the dedup in the opposite direction (re-publishing every 30 s heartbeat).
|
|
var fingerprint = (health.State, health.LastSuccessfulRead, health.LastError, errorCount,
|
|
_rediscoveryNeededUtc, HostStatusDigest(hostStatuses));
|
|
if (_lastPublishedFingerprint is { } prev && prev.Equals(fingerprint))
|
|
return;
|
|
_lastPublishedFingerprint = fingerprint;
|
|
_healthPublisher.Publish(
|
|
_clusterId, _driverInstanceId, health, errorCount, _rediscoveryNeededUtc, _rediscoveryReason,
|
|
hostStatuses);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_log.Warning(ex, "DriverInstance {Id}: GetHealth threw during health publish; skipping", _driverInstanceId);
|
|
}
|
|
}
|
|
|
|
/// <summary>Order-insensitive value digest of a host-status list for the publish fingerprint. Null in,
|
|
/// null out — so "driver is not a probe" and "probe reports no hosts" stay distinguishable rather than
|
|
/// both collapsing to the empty string.
|
|
/// <para>Ordered by host name because a driver is free to return its hosts in any order (several build
|
|
/// the list from a <c>Dictionary</c>), and an order flip would otherwise read as a real transition and
|
|
/// re-publish forever.</para></summary>
|
|
private static string? HostStatusDigest(IReadOnlyList<HostConnectivityStatus>? statuses) =>
|
|
statuses is null
|
|
? null
|
|
: string.Join('|', statuses
|
|
.OrderBy(s => s.HostName, StringComparer.Ordinal)
|
|
.Select(s => $"{s.HostName}={s.State}@{s.LastChangedUtc:O}"));
|
|
|
|
/// <summary>Fingerprint of the last <see cref="PublishHealthSnapshot"/> call; null until first publish.</summary>
|
|
private (DriverState State, DateTime? LastSuccess, string? LastError, int ErrorCount,
|
|
DateTime? RediscoveryNeededUtc, string? HostStatusDigest)? _lastPublishedFingerprint;
|
|
|
|
/// <inheritdoc />
|
|
protected override void PostStop()
|
|
{
|
|
DetachSubscription();
|
|
// MUST happen: the IDriver instance can outlive this actor (the host respawns a child around the
|
|
// same driver object), so a missing unsubscribe accumulates a handler per respawn holding a dead Self.
|
|
DetachRediscoverySource();
|
|
DetachHostStatusSource();
|
|
try { _driver.ShutdownAsync(CancellationToken.None).GetAwaiter().GetResult(); }
|
|
catch (Exception ex) { _log.Warning(ex, "DriverInstance {Id}: ShutdownAsync threw on PostStop", _driverInstanceId); }
|
|
OtOpcUaTelemetry.DriverInstanceLifecycle.Add(1,
|
|
new KeyValuePair<string, object?>("event", "stop"),
|
|
new KeyValuePair<string, object?>("driver_type", _driver.DriverType));
|
|
}
|
|
}
|