Joseph Doherty
e77c8a3569
The second thing #516/§8.2 deliberately did not build. The original objection stands — Modbus, S7, MQTT, AbLegacy and Sql echo authored config from DiscoverAsync, so diffing discovered against authored is a tautology for them — but it is now ENCODED rather than used as a reason not to build. The discriminator already existed: ITagDiscovery.SupportsOnlineDiscovery is documented as "enumerates the tag set from the live backend rather than replaying pre-declared/authored tags", and it separates the fleet cleanly — FOCAS, TwinCAT, MTConnect and AbCip true; the five config-echo drivers explicitly false. The check runs only for a driver that is SupportsOnlineDiscovery AND reports the new AuthoredDiscoveryRefs. That is nullable rather than empty-by-default on purpose: an empty collection legitimately means "nothing authored, everything discovered is new", so conflating the two would report total drift for a driver that never opted in. Where the value is: AbCip and FOCAS browse a live backend but implement no IRediscoverable, so before this they had no change signal at all. A periodic compare is the only way to notice a PLC re-download. A finding that changed the design: FOCAS, TwinCAT and AbCip all re-emit their authored tags into the same DiscoverAsync stream as device-derived ones, so the browse picker can show an operator their existing tags. Discovered is therefore always a superset of authored, and a VANISHED tag can never appear missing — one whole direction is structurally undetectable for three of the four. Shipping a Vanished list that is permanently empty for them would have been the half-inert seam this whole exercise removes. So the driver declares DiscoveryStreamIncludesAuthoredTags, the detector does not compute the undetectable half, and the operator message says "missing-tag detection unavailable for this driver" rather than letting the absence of a missing-tag clause read as reassurance. MTConnect is the only driver where both directions work — its stream is purely probe-model-derived. Behaviour: a failed browse is not drift (an unreachable device would otherwise report every authored tag as vanished); a truncated capture is skipped for the same reason; an unchanged drift is reported once rather than re-stamping its timestamp; drift that resolves clears the prompt. Interval defaults to 5 minutes — it browses a real device, and a tag set changing is an engineering event, not a runtime one. It reuses the Stage-2 surfacing, raising the same /hosts re-browse prompt, and never rebuilds the served address space. Comparison logic is a pure, separately-tested type rather than actor-inline. 14 new tests. The gate was verified load-bearing by deleting the SupportsOnlineDiscovery half: the tautology-driver test goes red, and it asserts discovery was never invoked rather than merely "no prompt raised", which would have passed for the wrong reason if the timer never fired. Full suite green except the same three pre-existing fixture-gated integration suites, identical counts.
1152 lines
65 KiB
C#
1152 lines
65 KiB
C#
using Akka.Actor;
|
|
using Akka.Event;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Browsing;
|
|
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);
|
|
|
|
/// <summary>Default period of the tag-set drift check. Deliberately slow: it browses a real device, and
|
|
/// a tag set changing is an engineering event (a PLC re-download, a CNC option install), not a runtime
|
|
/// one. Five minutes bounds the wire cost while still surfacing drift the same shift it happens.</summary>
|
|
public static readonly TimeSpan DefaultDriftCheckInterval = TimeSpan.FromMinutes(5);
|
|
|
|
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 on a slow timer to re-browse a genuinely-browsable driver and compare what the
|
|
/// remote offers against what an operator authored. Only ever scheduled for a driver that opts in — see
|
|
/// <see cref="QualifiesForDriftCheck"/>.</summary>
|
|
private sealed record DriftCheckTick
|
|
{
|
|
public static readonly DriftCheckTick Instance = new();
|
|
private DriftCheckTick() { }
|
|
}
|
|
|
|
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;
|
|
|
|
/// <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>Period of the tag-set drift check; tests inject a tiny value.</summary>
|
|
private readonly TimeSpan _driftCheckInterval;
|
|
|
|
/// <summary>Signature of the last drift REPORTED, so an unchanged drift is not re-announced on every
|
|
/// check. Without this the operator's prompt would re-stamp its timestamp every interval forever,
|
|
/// which reads as "it just happened again" rather than "it is still true".</summary>
|
|
private string? _lastDriftSignature;
|
|
|
|
/// <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="driftCheckInterval">Optional period of the tag-set drift check; defaults to
|
|
/// <see cref="DefaultDriftCheckInterval"/>. Tests inject a tiny value so the check runs without a wait.</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,
|
|
TimeSpan? driftCheckInterval = null) =>
|
|
Akka.Actor.Props.Create(() => new DriverInstanceActor(
|
|
driver,
|
|
reconnectInterval ?? DefaultReconnectInterval,
|
|
startStubbed,
|
|
healthPublisher ?? NullDriverHealthPublisher.Instance,
|
|
clusterId ?? string.Empty,
|
|
invoker,
|
|
healthPollInterval,
|
|
driftCheckInterval));
|
|
|
|
/// <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="driftCheckInterval">Period of the tag-set drift check; defaults to
|
|
/// <see cref="DefaultDriftCheckInterval"/>.</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,
|
|
TimeSpan? driftCheckInterval = null)
|
|
{
|
|
_driver = driver;
|
|
_invoker = invoker ?? NullDriverCapabilityInvoker.Instance;
|
|
_driverInstanceId = driver.DriverInstanceId;
|
|
_clusterId = clusterId ?? string.Empty;
|
|
_healthPublisher = healthPublisher ?? NullDriverHealthPublisher.Instance;
|
|
_driftCheckInterval = driftCheckInterval ?? DefaultDriftCheckInterval;
|
|
_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();
|
|
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<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();
|
|
StartDriftCheck();
|
|
});
|
|
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<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();
|
|
StopDriftCheck();
|
|
Become(Reconnecting);
|
|
PublishHealthSnapshot();
|
|
});
|
|
Receive<ForceReconnect>(_ =>
|
|
{
|
|
_log.Info("DriverInstance {Id}: ForceReconnect requested by admin; re-entering Reconnecting", _driverInstanceId);
|
|
DetachSubscription();
|
|
StopDriftCheck();
|
|
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);
|
|
ReceiveAsync<DriftCheckTick>(HandleDriftCheckAsync);
|
|
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<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();
|
|
StartDriftCheck();
|
|
});
|
|
// 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<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>Stops the drift check. Called on every Connected exit — browsing a disconnected driver would
|
|
/// fail every pass, and a failed browse is deliberately NOT treated as drift.</summary>
|
|
private void StopDriftCheck() => Timers.Cancel("drift-check");
|
|
|
|
/// <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>
|
|
/// True when this driver is worth drift-checking: it must genuinely BROWSE its remote
|
|
/// (<see cref="ITagDiscovery.SupportsOnlineDiscovery"/>) and must report the refs its authored tags
|
|
/// bind (<see cref="ITagDiscovery.AuthoredDiscoveryRefs"/>).
|
|
/// <para>The first condition is what makes the check meaningful rather than reassuring. Most drivers
|
|
/// — Modbus, S7, AbLegacy, Sql, Mqtt — stream their AUTHORED tags back out of <c>DiscoverAsync</c>
|
|
/// rather than enumerating the device, so diffing the two sets for them is a tautology that would
|
|
/// report "no drift" forever. A check that cannot fail is worse than no check: it looks like
|
|
/// coverage.</para>
|
|
/// </summary>
|
|
private bool QualifiesForDriftCheck() =>
|
|
_driver is ITagDiscovery { SupportsOnlineDiscovery: true, AuthoredDiscoveryRefs: not null };
|
|
|
|
/// <summary>Starts the slow drift check on a Connected entry, for qualifying drivers only.</summary>
|
|
private void StartDriftCheck()
|
|
{
|
|
if (!QualifiesForDriftCheck()) return;
|
|
// Periodic, not fire-on-connect: a driver whose discovered shape fills in asynchronously after
|
|
// connect (the FOCAS FixedTree) would otherwise be compared against a half-populated browse and
|
|
// report phantom drift on every reconnect.
|
|
Timers.StartPeriodicTimer("drift-check", DriftCheckTick.Instance, _driftCheckInterval);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Re-browses the remote and compares it against the driver's authored refs. A CHANGED drift raises
|
|
/// the same operator prompt an <see cref="IRediscoverable"/> raise does — this is the signal for the
|
|
/// browsable drivers that have no native change notification (AbCip, FOCAS), where a periodic
|
|
/// compare is the only way to notice a PLC program re-download.
|
|
/// <para>Advisory, like every rediscovery signal: the served address space is not touched. Raw tags
|
|
/// are authored through the <c>/raw</c> browse-commit flow, so an operator re-browses and commits.</para>
|
|
/// </summary>
|
|
private async Task HandleDriftCheckAsync(DriftCheckTick _)
|
|
{
|
|
if (_driver is not ITagDiscovery discovery) return;
|
|
var authored = discovery.AuthoredDiscoveryRefs;
|
|
if (authored is null) return;
|
|
|
|
CapturedTree tree;
|
|
try
|
|
{
|
|
var builder = new CapturingAddressSpaceBuilder();
|
|
// Bounded: ReceiveAsync suspends the mailbox for the whole handler, so an unbounded browse would
|
|
// block writes, reconnects and health polls behind it.
|
|
using var cts = new CancellationTokenSource(DriftBrowseTimeout);
|
|
await _invoker.ExecuteAsync(
|
|
DriverCapability.Discover,
|
|
_driverInstanceId,
|
|
async ct => await discovery.DiscoverAsync(builder, ct),
|
|
cts.Token);
|
|
tree = builder.Build();
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
// A failed browse is not drift — the device may simply be unreachable, which the health surface
|
|
// already reports. Reporting drift here would turn every comms blip into "all your tags vanished".
|
|
_log.Debug(ex, "DriverInstance {Id}: drift check browse failed; skipping this pass", _driverInstanceId);
|
|
return;
|
|
}
|
|
|
|
if (tree.Truncated)
|
|
{
|
|
// A truncated capture is a PARTIAL view, so every un-captured authored ref would look vanished.
|
|
_log.Warning(
|
|
"DriverInstance {Id}: drift check skipped — the browse hit the capture cap and is partial",
|
|
_driverInstanceId);
|
|
return;
|
|
}
|
|
|
|
var discovered = new List<string>();
|
|
CollectLeafRefs(tree.Root, discovered);
|
|
var drift = TagSetDriftDetector.Compare(
|
|
discovered, authored, detectVanished: !discovery.DiscoveryStreamIncludesAuthoredTags);
|
|
|
|
if (!drift.HasDrift)
|
|
{
|
|
// Recovered: the device matches the authored set again, so drop the prompt and let the next
|
|
// genuine drift re-raise with a fresh timestamp.
|
|
if (_lastDriftSignature is not null)
|
|
{
|
|
_log.Info("DriverInstance {Id}: tag-set drift cleared — device matches the authored set",
|
|
_driverInstanceId);
|
|
_lastDriftSignature = null;
|
|
_rediscoveryNeededUtc = null;
|
|
_rediscoveryReason = null;
|
|
PublishHealthSnapshot();
|
|
}
|
|
|
|
return;
|
|
}
|
|
|
|
// Report a CHANGED drift once. An unchanged one re-stamping its timestamp every interval would read
|
|
// as "it just happened again" rather than "it is still true".
|
|
if (string.Equals(_lastDriftSignature, drift.Signature, StringComparison.Ordinal)) return;
|
|
_lastDriftSignature = drift.Signature;
|
|
_rediscoveryNeededUtc = DateTime.UtcNow;
|
|
_rediscoveryReason = "device tag set differs from authored: " + drift.Describe();
|
|
_log.Info(
|
|
"DriverInstance {Id}: tag-set drift detected ({Drift}) — surfaced for operator re-browse; the served address space is unchanged",
|
|
_driverInstanceId, drift.Describe());
|
|
PublishHealthSnapshot();
|
|
}
|
|
|
|
/// <summary>Per-pass timeout for the drift browse. Bounds the mailbox suspension.</summary>
|
|
private static readonly TimeSpan DriftBrowseTimeout = TimeSpan.FromSeconds(30);
|
|
|
|
/// <summary>Flattens a captured tree to its leaf ids — the driver-side <c>FullName</c>s, which is the
|
|
/// vocabulary <c>AuthoredDiscoveryRefs</c> is defined in.</summary>
|
|
private static void CollectLeafRefs(CapturedNode node, List<string> into)
|
|
{
|
|
if (node.IsLeaf) into.Add(node.Id);
|
|
foreach (var child in node.Children) CollectLeafRefs(child, into);
|
|
}
|
|
|
|
/// <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();
|
|
// _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.
|
|
var fingerprint = (health.State, health.LastSuccessfulRead, health.LastError, errorCount, _rediscoveryNeededUtc);
|
|
if (_lastPublishedFingerprint is { } prev && prev.Equals(fingerprint))
|
|
return;
|
|
_lastPublishedFingerprint = fingerprint;
|
|
_healthPublisher.Publish(
|
|
_clusterId, _driverInstanceId, health, errorCount, _rediscoveryNeededUtc, _rediscoveryReason);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_log.Warning(ex, "DriverInstance {Id}: GetHealth threw during health publish; skipping", _driverInstanceId);
|
|
}
|
|
}
|
|
|
|
/// <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)? _lastPublishedFingerprint;
|
|
|
|
/// <inheritdoc />
|
|
protected override void PostStop()
|
|
{
|
|
StopDriftCheck();
|
|
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();
|
|
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));
|
|
}
|
|
}
|