using Akka.Actor;
using Akka.Event;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management;
using ZB.MOM.WW.ScadaBridge.Commons.Observability;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.HealthMonitoring;
using ZB.MOM.WW.ScadaBridge.SiteEventLogging;
namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
///
/// Connection actor using Akka.NET Become for its lifecycle state machine.
///
/// States:
/// - Connecting: retain latest subscription requests, reject writes; attempts connection
/// - Connected: process pending subscriptions and live requests
/// - Reconnecting: push bad quality, retain latest subscriptions and reject writes,
/// fixed-interval reconnect
///
/// Auto-reconnect with bad quality on disconnect.
/// Transparent re-subscribe after reconnection.
/// Write-back support (synchronous failure to caller, no S&F).
/// Tag path resolution with retry.
/// Health reporting (connection status + tag resolution counts).
/// Subscription lifecycle (register on create, cleanup on stop).
///
public class DataConnectionActor : UntypedActor, IWithTimers
{
public enum ActiveEndpoint { Primary, Backup }
private readonly ILoggingAdapter _log = Context.GetLogger();
private readonly string _connectionName;
private IDataConnection _adapter;
private readonly DataConnectionOptions _options;
private readonly ISiteHealthCollector _healthCollector;
private readonly IDataConnectionFactory _factory;
///
/// Protocol of the configuration currently in force. NOT readonly: an in-place
/// reconfiguration () may change the protocol, and
/// the factory keys off it when the replacement adapter is built.
///
private string _protocolType;
private readonly ISiteEventLogger? _siteEventLogger;
/// Timer scheduler for reconnect and tag-resolution retry timers.
public ITimerScheduler Timers { get; set; } = null!;
///
/// Active subscriptions: instanceUniqueName → set of tag paths.
///
private readonly Dictionary> _subscriptionsByInstance = new();
///
/// Subscription IDs returned by the adapter: tagPath → subscriptionId.
///
private readonly Dictionary _subscriptionIds = new();
///
/// Per-tag COUNTED SET — the reverse index of which instances subscribe to each tag
/// path: tagPath → set of instanceUniqueName. Mirrors
/// inverted so the hot path fans a value out to exactly
/// the interested instances in O(subscribers) instead of scanning every instance's tag set
/// on every tag update. Maintained via /
/// at every tag-set mutation; a tag key is dropped
/// when its instance set empties. Derived purely from ,
/// so it is preserved across reconnect exactly as that map is (see ).
///
/// It is also THE authority for three other questions, so that no independently-mutated
/// counter can drift away from it:
///
/// - "does any instance still need this tag?" — 's
/// last-subscriber test is simply "did drop the key?" (O(1),
/// replacing a separate _tagSubscriberCount map that had to be kept in step);
/// - "was this tag unsubscribed while a batch subscribe was in flight?" — the discard
/// gates in ;
/// - the reported .
///
///
private readonly Dictionary> _instancesByTag = new();
///
/// Tags whose path resolution failed and are awaiting retry.
///
private readonly HashSet _unresolvedTags = new();
///
/// Tags whose retry SubscribeAsync is currently in flight.
/// They are excluded from the next retry tick so a slow attempt is not duplicated
/// (which would leak monitored items / subscription ids).
///
private readonly HashSet _resolutionInFlight = new();
///
/// Tags whose initial SubscribeAsync (issued from
/// ) is currently in flight. Two parallel
/// SubscribeTagsRequest messages for different instances sharing a tag
/// path would otherwise both observe "not subscribed" against
/// (the in-flight task has not yet posted its
/// ), both call _adapter.SubscribeAsync,
/// and the second subscription id gets silently dropped at the existing
/// _subscriptionIds.ContainsKey guard in
/// — orphaning the adapter's monitored item (duplicate notifications + leaked
/// memory until the connection drops). This set is read+written only on the
/// actor thread and cleared in for symmetry with
/// .
///
private readonly HashSet _subscribesInFlight = new();
///
/// Subscribers: instanceUniqueName → IActorRef (the Instance Actor).
///
private readonly Dictionary _subscribers = new();
// ── Native alarm subscriptions ──
// The connection opens one alarm feed per source reference; transitions are
// routed to subscribers (NativeAlarmActors) by source-object reference.
/// sourceReference → set of subscriber actor refs (NativeAlarmActors), for routing + ref-count.
private readonly Dictionary> _alarmSourceSubscribers = new();
///
/// sourceReference → raw condition filter string passed to the adapter (last subscriber wins).
/// The shared feed carries a single filter: overwrites it
/// unconditionally on every subscribe, so co-subscribers to one source reference must agree on
/// the condition filter (a second subscriber's filter re-gates the first subscriber's transitions).
///
private readonly Dictionary _alarmSourceFilter = new();
///
/// sourceReference → parsed condition-type predicate. The authoritative
/// client-side gate in ; applies uniformly
/// across OPC UA and the gateway-wide MxGateway feed.
///
private readonly Dictionary _alarmSourceFilterPredicate = new();
/// sourceReference → adapter alarm subscription id.
private readonly Dictionary _alarmSubscriptionIds = new();
/// sourceReferences whose adapter SubscribeAlarmsAsync is currently in flight.
private readonly HashSet _alarmSubscribesInFlight = new();
///
/// Total subscribed tags for health reporting: the number of DISTINCT tag paths that at
/// least one instance currently subscribes to.
///
/// DERIVED, never accumulated. This used to be an int field incremented and
/// decremented at five independent sites (initial subscribe success, initial subscribe
/// resolution failure, both unsubscribe branches) while the reconnect path cleared the
/// maps those decrements keyed off — so an unsubscribe that landed inside a reconnect
/// window matched NEITHER decrement branch and leaked +1 per churn cycle, permanently.
/// Reading the counted set at report time makes that class of drift unrepresentable:
/// the number can only ever be what the authoritative per-tag state says it is.
///
/// A tag counts from the moment it is registered against an instance, whatever its
/// resolution outcome — resolved, awaiting a resolution retry, or failed at connection
/// level (that last case used to be excluded, which let
/// climb ABOVE the total once the reconnect re-subscribe resolved it).
///
private int TotalSubscribedTagCount => _instancesByTag.Count;
///
/// Resolved tags for health reporting — DERIVED, for the same reason as
/// . A tag is resolved exactly when the adapter has
/// handed back a subscription handle for it, so IS the
/// count; the retired _resolvedTags field was incremented and decremented in
/// lock-step with every mutation of that dictionary anyway. Never exceeds
/// , because a handle is only ever stored for a tag
/// that is present in (enforced at every store site).
///
private int ResolvedTagCount => _subscriptionIds.Count;
private int _tagsGoodQuality;
private int _tagsBadQuality;
private int _tagsUncertainQuality;
private readonly Dictionary _lastTagQuality = new();
///
/// Set when a genuine quality TRANSITION has moved the counters since the last push to
/// the health collector. The push itself is coalesced onto a single-shot
/// quality-flush timer ()
/// instead of running on every received value: at 37,500 tags the per-message push was
/// pure overhead against a collector that is only read every 30s. Disconnect,
/// unsubscribe and the reconnect reset still flush SYNCHRONOUSLY — their correctness
/// depends on the collector being current at that instant.
///
private bool _qualityDirty;
///
/// Current tag-resolution retry interval. Starts at
/// , doubles after a
/// round in which no tag resolved, and is capped at
/// . Reset to the floor
/// whenever a tag resolves or the connection reconnects, so a dead device backs off
/// while a booting one is still picked up quickly.
///
private TimeSpan _tagResolutionInterval;
// ── Alarm subscriber prefix index ──
// HandleAlarmTransitionReceived used to scan EVERY subscribed source (two StartsWith
// per source) on every transition. Sources are bucketed by the FIRST path segment of
// their reference; a transition is matched only against the bucket for its own first
// segment, plus the residue list of sources that carry no separator at all (a prefix
// shorter than one segment, which can match transitions in other buckets). The
// per-source StartsWith + condition-type gate inside the bucket is unchanged, so
// routing decisions are identical to the linear scan.
private readonly Dictionary> _alarmSourcesByFirstSegment = new(StringComparer.Ordinal);
private readonly HashSet _alarmSourcesWithoutSeparator = new(StringComparer.Ordinal);
private IDictionary _connectionDetails;
// The three fields below describe the configuration currently in force. None are readonly:
// an in-place reconfiguration (ReconfigureConnectionCommand, deferred register row 97)
// replaces them wholesale and then swaps the adapter through the same machinery the
// primary→backup failover uses.
private IDictionary _primaryConfig;
private IDictionary? _backupConfig;
private int _failoverRetryCount;
private ActiveEndpoint _activeEndpoint = ActiveEndpoint.Primary;
private int _consecutiveFailures;
private int _consecutiveUnstableDisconnects;
private DateTimeOffset _lastConnectedAt;
///
/// Monotonically increasing tag that identifies the
/// current adapter instance. Subscription callbacks capture the generation in
/// effect when they were created; a whose
/// generation no longer matches comes from a disposed adapter and is dropped so
/// stale pre-failover device data is never forwarded to Instance Actors.
///
private int _adapterGeneration;
///
/// Captured Self reference for use from non-actor threads (event handlers, callbacks).
/// Akka.NET's Self property is only valid inside the actor's message loop.
///
private IActorRef _self = null!;
///
/// The delegate currently hooked to _adapter.Disconnected. It closes over the
/// in force when it was hooked, so the resulting
/// carries the generation of the adapter that raised it —
/// a method group could not. Kept in a field so and
/// unhook exactly the instance they hooked.
///
private Action? _disconnectedHandler;
///
/// Set when an for the
/// CURRENT adapter generation is received while the actor is still Connecting/Reconnecting —
/// i.e. the adapter connected and dropped again before the piped
/// for that attempt reached the mailbox.
///
///
/// Neither state used to handle the message at all, so it fell through to
/// Unhandled and the subsequent ConnectResult(true) declared the connection up
/// on an adapter that was already dead. Nothing recovers from there: an adapter fires
/// Disconnected at most once per successful connect (OpcUaDataConnection's
/// once-only _disconnectFired guard, reset only by the next successful
/// ConnectAsync), so a connection with zero subscriptions — nothing to fail at
/// connection level and drive the reconnect machine — sat Connected on a dead adapter until
/// the process restarted.
///
///
///
/// It is cleared at the top of , so it only ever refers to
/// the attempt currently in flight: a disconnect belonging to a session that ended BEFORE
/// this attempt began (e.g. the adapter's Disconnected trailing a connection-level
/// subscribe failure that already drove the actor into Reconnecting) is discarded rather than
/// replayed against the fresh session. That is also why the flag is used instead of
/// Stash: the stash has no selective drop, so a stashed disconnect would be replayed
/// after the next successful connect and cost a spurious reconnect cycle.
///
///
///
/// Residual (accepted, and in the safe direction): a disconnect raised by the PREVIOUS
/// session but delivered after the next attempt has already cleared the flag is attributed to
/// the new attempt, costing one extra reconnect cycle. Being wedged Connected on a dead
/// adapter is the failure this trades against.
///
///
private bool _disconnectDuringConnectAttempt;
///
/// Cancels the reconnect re-seed started by when the session it
/// belongs to ends (review 2026-10-21 02/L4). Without it a flapping link stacked one
/// full-tag-set seed read per flap, each alive for up to
/// against a device that was already
/// struggling — work whose values the stale-generation guard then dropped anyway.
///
private CancellationTokenSource? _sessionWorkCts;
///
/// Initializes the data connection actor with its adapter and configuration.
///
/// Human-readable name used in logs and health metrics.
/// The protocol adapter for the primary endpoint.
/// Data connection layer configuration options.
/// Collector for site health metrics.
/// Factory used to create replacement adapters on failover.
/// Protocol type identifier (e.g. "OpcUa").
/// Configuration dictionary for the primary endpoint.
/// Optional configuration dictionary for the backup endpoint.
/// Number of consecutive failures before switching to the backup endpoint.
/// Optional site event logger for operational events.
public DataConnectionActor(
string connectionName,
IDataConnection adapter,
DataConnectionOptions options,
ISiteHealthCollector healthCollector,
IDataConnectionFactory factory,
string protocolType,
IDictionary? primaryConfig = null,
IDictionary? backupConfig = null,
int failoverRetryCount = 3,
ISiteEventLogger? siteEventLogger = null)
{
_connectionName = connectionName;
_adapter = adapter;
_options = options;
_healthCollector = healthCollector;
_factory = factory;
_protocolType = protocolType;
_primaryConfig = primaryConfig ?? new Dictionary();
_backupConfig = backupConfig;
_failoverRetryCount = failoverRetryCount;
_siteEventLogger = siteEventLogger;
_connectionDetails = _primaryConfig;
_tagResolutionInterval = _options.TagResolutionRetryInterval;
}
///
protected override void PreStart()
{
_log.Info("DataConnectionActor [{0}] starting in Connecting state", _connectionName);
// Capture Self for use from non-actor threads (event handlers, callbacks).
// Akka.NET's Self property is only valid inside the actor's message loop.
_self = Self;
// Listen for unexpected adapter disconnections
HookAdapterDisconnected();
BecomeConnecting();
}
///
/// Subscribes to the CURRENT adapter's Disconnected event with a delegate that closes
/// over the generation in force right now, so the marshalled
/// identifies the adapter that raised it. Callers must
/// have bumped for the new adapter first
/// (see ).
///
private void HookAdapterDisconnected()
{
var generation = _adapterGeneration;
var self = _self;
// Marshal the event onto the actor's message loop using the captured _self reference.
// This runs on a background thread (keep-alive / gRPC stream reader), so Self would throw.
_disconnectedHandler = () => self.Tell(new AdapterDisconnected(generation));
_adapter.Disconnected += _disconnectedHandler;
}
/// Unhooks the delegate installed by , if any.
private void UnhookAdapterDisconnected()
{
if (_disconnectedHandler is null)
return;
_adapter.Disconnected -= _disconnectedHandler;
_disconnectedHandler = null;
}
///
protected override void PostStop()
{
_log.Info("DataConnectionActor [{0}] stopping — disposing adapter", _connectionName);
CancelSessionWork();
UnhookAdapterDisconnected();
// S9: fire-and-forget, but observe a faulted dispose so a failing adapter
// teardown is logged rather than swallowed into an unobserved task exception.
_adapter.DisposeAsync().AsTask().ContinueWith(
t => _log.Warning("[{0}] Adapter dispose faulted on stop: {1}",
_connectionName, t.Exception?.GetBaseException().Message),
TaskContinuationOptions.OnlyOnFaulted);
}
///
protected override void OnReceive(object message)
{
// Default handler — should not be reached due to Become
Unhandled(message);
}
// ── Connecting State ──
private void BecomeConnecting()
{
_log.Info("[{0}] Entering Connecting state", _connectionName);
_healthCollector.UpdateConnectionHealth(_connectionName, ConnectionHealth.Connecting);
_healthCollector.UpdateConnectionEndpoint(_connectionName, "Connecting");
// Register row 54 — mirror the state into the data-connection gauges. Kept alongside
// the health-collector write (rather than derived from it) so the metric never
// depends on the health-report path being alive.
ScadaBridgeTelemetry.RecordDataConnectionState(_connectionName, ConnectionHealth.Connecting);
Become(Connecting);
Self.Tell(new AttemptConnect());
}
private void Connecting(object message)
{
switch (message)
{
case AttemptConnect:
HandleAttemptConnect();
break;
case ConnectResult result:
HandleConnectResult(result);
break;
case AdapterDisconnected disconnected:
// The adapter connected and dropped again before this attempt's
// ConnectResult was applied. Remember it (see _disconnectDuringConnectAttempt)
// rather than acting now: acting would put the actor into Reconnecting and the
// in-flight ConnectResult(true) would then be applied THERE, re-subscribing and
// declaring the same dead adapter Connected anyway.
NotePendingDisconnect(disconnected);
break;
case SubscribeTagsRequest pending:
QueueSubscribe(pending);
break;
case SubscribeAlarmsRequest pendingAlarm:
QueueSubscribeAlarms(pendingAlarm);
break;
case WriteTagRequest write:
RejectDisconnectedWrite(write);
break;
case WriteTagBatchRequest batch:
RejectDisconnectedWrite(batch);
break;
case UnsubscribeTagsRequest unsubscribe:
HandleUnsubscribe(unsubscribe);
break;
case UnsubscribeAlarmsRequest unsubscribeAlarm:
HandleUnsubscribeAlarms(unsubscribeAlarm);
break;
case ReconfigureConnectionCommand rc:
// NEVER stashed. A reconfigure that arrived while the initial connect was still
// in flight would otherwise sit in the stash until the OLD configuration
// succeeded — and if it never succeeds (the very case an endpoint correction is
// sent to fix) it would sit there forever.
HandleReconfigure(rc);
break;
case SubscribeCompleted sc:
// A subscribe started while Connected can complete after a transition;
// apply it so its state survives into the next ReSubscribeAll.
HandleSubscribeCompleted(sc);
break;
case AlarmSubscribeCompleted asc:
HandleAlarmSubscribeCompleted(asc);
break;
case AlarmTransitionReceived:
// No live feed yet in Connecting; ignore (snapshot replays on subscribe).
break;
case BatchSubscribeCompleted:
// Re-subscribe / resolution-probe results from a previous connection —
// ReSubscribeAll re-issues everything once the link is back up.
break;
case RetryTagResolution:
// No session yet — a probe would fail for every tag. ReSubscribeAll's chunk
// completions re-arm the timer once the link is up.
break;
case QualityFlushTick:
FlushQualityCountersIfDirty();
break;
case BrowseNodeCommand browse:
// Browse is an interactive design-time query; never stash. The
// adapter has no session yet in this state, so reply with a
// typed ConnectionNotConnected failure so the dialog can render
// an inline banner.
HandleBrowse(browse);
break;
case SearchAddressSpaceCommand search:
// Search is the address-space analogue of browse — same rule:
// never stash; the adapter has no session yet here, so
// HandleSearch short-circuits to ConnectionNotConnected.
HandleSearch(search);
break;
case ReadTagValuesCommand read:
// Same rule as browse — never stash; adapter is not yet
// connected, so HandleReadTagValues short-circuits to
// ConnectionNotConnected.
HandleReadTagValues(read);
break;
case Terminated terminated:
HandleSubscriberTerminated(terminated.ActorRef);
break;
case GetHealthReport:
ReplyWithHealthReport();
break;
default:
Unhandled(message);
break;
}
}
// ── Connected State ──
private void BecomeConnected()
{
_log.Info("[{0}] Entering Connected state", _connectionName);
_lastConnectedAt = DateTimeOffset.UtcNow;
_healthCollector.UpdateConnectionHealth(_connectionName, ConnectionHealth.Connected);
_healthCollector.UpdateTagResolution(_connectionName, TotalSubscribedTagCount, ResolvedTagCount);
var endpointLabel = _backupConfig == null ? "Connected" : $"Connected to {_activeEndpoint.ToString().ToLower()}";
_healthCollector.UpdateConnectionEndpoint(_connectionName, endpointLabel);
// Register row 54 — the only transition that raises the connected gauge.
ScadaBridgeTelemetry.RecordDataConnectionState(_connectionName, ConnectionHealth.Connected);
Become(Connected);
foreach (var pending in _pendingTagSubscriptions.Values)
Self.Tell(pending.Request, pending.Owner);
_pendingTagSubscriptions.Clear();
foreach (var (key, request) in _pendingAlarmSubscriptions)
Self.Tell(request, key.Owner);
_pendingAlarmSubscriptions.Clear();
}
private void Connected(object message)
{
switch (message)
{
case SubscribeTagsRequest req:
HandleSubscribe(req);
break;
case SubscribeCompleted sc:
// In Connected state, a connection-level subscribe failure must drive
// the reconnection state machine.
if (HandleSubscribeCompleted(sc))
{
_log.Warning("[{0}] Connection-level subscribe failure — entering Reconnecting", _connectionName);
BecomeReconnecting();
}
break;
case UnsubscribeTagsRequest req:
HandleUnsubscribe(req);
break;
case SubscribeAlarmsRequest areq:
HandleSubscribeAlarms(areq);
break;
case UnsubscribeAlarmsRequest areq:
HandleUnsubscribeAlarms(areq);
break;
case AlarmSubscribeCompleted asc:
HandleAlarmSubscribeCompleted(asc);
break;
case AlarmTransitionReceived atr:
HandleAlarmTransitionReceived(atr);
break;
case WriteTagRequest req:
HandleWrite(req);
break;
case WriteTagBatchRequest req:
HandleWriteBatch(req);
break;
case TagValueReceived tvr:
HandleTagValueReceived(tvr);
break;
case BatchSubscribeCompleted bsc:
HandleBatchSubscribeCompleted(bsc);
break;
case QualityFlushTick:
FlushQualityCountersIfDirty();
break;
case AdapterDisconnected disconnected:
// Fence the disconnect on its adapter generation. The outgoing adapter's
// event can fire concurrently with SwapAdapter's unhook, so a Disconnected from
// the endpoint we have just failed over (or reconfigured) AWAY from could arrive
// here and knock the healthy replacement back into a reconnect cycle.
if (!IsStaleDisconnect(disconnected))
HandleDisconnect();
break;
case ReconfigureConnectionCommand rc:
HandleReconfigure(rc);
break;
case RetryTagResolution:
HandleRetryTagResolution();
break;
case BrowseNodeCommand browse:
HandleBrowse(browse);
break;
case SearchAddressSpaceCommand search:
HandleSearch(search);
break;
case ReadTagValuesCommand read:
HandleReadTagValues(read);
break;
case Terminated terminated:
HandleSubscriberTerminated(terminated.ActorRef);
break;
case GetHealthReport:
ReplyWithHealthReport();
break;
default:
Unhandled(message);
break;
}
}
// ── Reconnecting State ──
///
/// True when this transition was caused by an in-place reconfiguration
/// () rather than by losing the link. It suppresses the
/// unstable-disconnect accounting and the failover check — the adapter has just been
/// swapped ON PURPOSE against a brand-new configuration, so counting it as an unstable
/// cycle would let a reconfigure trip a failover onto the new backup before the new primary
/// has been tried even once — and it reconnects immediately instead of waiting out
/// , since there is nothing to wait for.
///
private void BecomeReconnecting(bool afterReconfigure = false)
{
_log.Warning("[{0}] Entering Reconnecting state{1}", _connectionName,
afterReconfigure ? " (reconfigured in place)" : string.Empty);
// The session is over: abandon its re-seed (02/L4).
CancelSessionWork();
_healthCollector.UpdateConnectionHealth(_connectionName, ConnectionHealth.Disconnected);
_healthCollector.UpdateConnectionEndpoint(_connectionName, "Disconnected");
// Register row 54 — drops this connection out of the connected gauge while the
// total holds, which is exactly the connected < total alert condition.
ScadaBridgeTelemetry.RecordDataConnectionState(_connectionName, ConnectionHealth.Disconnected);
if (!afterReconfigure)
{
// Track unstable connections toward failover.
// If we were connected for less than the stability threshold, this counts
// as an unstable cycle (e.g., connect succeeded but heartbeat went stale).
var connectionDuration = DateTimeOffset.UtcNow - _lastConnectedAt;
if (_lastConnectedAt != default && connectionDuration < _options.StableConnectionThreshold)
{
_consecutiveUnstableDisconnects++;
_log.Warning("[{0}] Unstable connection (lasted {1:F0}s) — consecutive unstable disconnects: {2}/{3}",
_connectionName, connectionDuration.TotalSeconds, _consecutiveUnstableDisconnects,
_backupConfig != null ? _failoverRetryCount : 0);
}
else
{
_consecutiveUnstableDisconnects = 0;
}
// Failover if we keep connecting and going stale repeatedly
if (_backupConfig != null && _consecutiveUnstableDisconnects >= _failoverRetryCount)
{
var previousEndpoint = _activeEndpoint;
_activeEndpoint = _activeEndpoint == ActiveEndpoint.Primary
? ActiveEndpoint.Backup
: ActiveEndpoint.Primary;
_consecutiveUnstableDisconnects = 0;
_consecutiveFailures = 0;
SwapAdapter(_activeEndpoint == ActiveEndpoint.Primary ? _primaryConfig : _backupConfig);
_log.Warning("[{0}] Failing over from {1} to {2} (unstable connection pattern)",
_connectionName, previousEndpoint, _activeEndpoint);
if (_siteEventLogger != null)
{
_ = _siteEventLogger.LogEventAsync(
"connection", "Warning", null, _connectionName,
$"Failover from {previousEndpoint} to {_activeEndpoint} (unstable connection)",
$"Connection lasted {connectionDuration.TotalSeconds:F0}s, threshold {_options.StableConnectionThreshold.TotalSeconds:F0}s");
}
}
}
// Log disconnect to site event log
if (_siteEventLogger != null)
{
_ = _siteEventLogger.LogEventAsync(
"connection", afterReconfigure ? "Info" : "Warning", null, _connectionName,
afterReconfigure
? "Connection reconfigured — reconnecting on the new configuration"
: "Connection lost — entering reconnect cycle",
null);
}
Become(Reconnecting);
// Push bad quality for all subscribed tags on disconnect
PushBadQualityForAllTags();
// Notify native alarm subscribers the source feed is unavailable
// (mark mirrored alarms uncertain; the reconnect snapshot reconciles them).
PushAlarmSourceUnavailable();
// Schedule reconnect attempt. StartSingleTimer REPLACES any reconnect already armed,
// so a reconfigure landing mid-backoff cannot leave two AttemptConnect messages racing.
Timers.StartSingleTimer("reconnect", new AttemptConnect(),
afterReconfigure ? TimeSpan.Zero : _options.ReconnectInterval);
}
///
/// Replaces the live adapter with one built for — the single
/// swap shared by both failover sites and by in-place reconfiguration
/// ().
///
///
/// The old adapter is unhooked first and disposed fire-and-forget (never awaited on the
/// actor thread), and is bumped LAST so every callback and
/// in-flight background task started against the outgoing adapter is recognised as stale and
/// dropped rather than delivered as if it came from the new endpoint. Subscriptions are
/// untouched: they live in and are re-issued by
/// once the replacement adapter connects.
///
///
private void SwapAdapter(IDictionary newConfig)
{
CancelSessionWork();
UnhookAdapterDisconnected();
_ = _adapter.DisposeAsync().AsTask();
_adapter = _factory.Create(_protocolType, newConfig);
_connectionDetails = newConfig;
_adapterGeneration++;
// Hooked AFTER the bump so the new adapter's handler stamps the NEW generation; the
// outgoing adapter's handler was captured with the old one and is now recognised as
// stale by IsStaleDisconnect even if it fires concurrently with the unhook above.
HookAdapterDisconnected();
}
///
/// Cancels background work bound to the current session (the reconnect re-seed). Cancel
/// only — never Dispose — because the background task may still be linking to the token.
///
private void CancelSessionWork()
{
_sessionWorkCts?.Cancel();
_sessionWorkCts = null;
}
///
/// Advances for a new SESSION on the SAME adapter — a plain
/// drop + reconnect, where never runs (review 2026-10-21 02/C1).
///
///
/// Without it, an initial subscribe issued against the dead session (in flight for up to
/// ) carried the current generation and
/// was applied against the new session: landing before the re-subscribe row for its tag it
/// parked a dead handle under the tag's key, so the LIVE row looked redundant and was
/// released (the tag reads resolved and is silent until the next reconnect); landing after,
/// its dead handle was released against the adapter whose new session may have reused that
/// handle number for another tag (MxGateway); and a connection-level failure row in it
/// knocked the freshly reconnected link back into Reconnecting.
///
///
///
/// The Disconnected handler stamps the generation it was hooked with, so it is
/// re-hooked for the new generation. The new handler is hooked BEFORE the old one is
/// removed: a disconnect raised in between is then delivered twice (old stamp dropped as
/// stale, new stamp acted on) rather than lost, which would wedge the actor Connected on a
/// dead session.
///
///
private void AdvanceSessionGeneration()
{
var previousHandler = _disconnectedHandler;
_adapterGeneration++;
HookAdapterDisconnected();
if (previousHandler is not null)
_adapter.Disconnected -= previousHandler;
}
///
/// Applies a new configuration to this LIVE connection in place (deferred register
/// row 97) — the non-destructive alternative to stopping and re-creating the connection
/// actor, which would strand every subscriber (the subscription registry lives here, and
/// InstanceActor cancels its subscribe-retry timer on the first success, so nothing
/// would ever re-subscribe).
///
///
/// It reuses the primary→backup failover machinery verbatim:
/// disposes the current adapter, builds one from the new configuration and bumps the
/// generation, then the reconnect cycle re-issues every subscription from
/// and every alarm feed from
/// . In-flight writes against the outgoing adapter may
/// fail — the same accepted consequence a failover has, and the failure is returned
/// synchronously to the calling script exactly as it would be there.
///
///
///
/// Endpoint selection is re-armed from the top. Whatever endpoint the PREVIOUS
/// configuration had failed over to, the new configuration's primary is the operator's
/// intent, so the active endpoint and both failure counters reset.
///
///
///
/// Idempotent. A reconfigure carrying the configuration already in force is a no-op.
/// The sender is hash-gated and does not send one, but a swap is destructive enough that a
/// duplicate must never cost a reconnect.
///
///
private void HandleReconfigure(ReconfigureConnectionCommand command)
{
if (string.Equals(_protocolType, command.ProtocolType, StringComparison.Ordinal)
&& _failoverRetryCount == command.FailoverRetryCount
&& DetailsEqual(_primaryConfig, command.PrimaryConnectionDetails)
&& DetailsEqual(_backupConfig, command.BackupConnectionDetails))
{
_log.Debug("[{0}] Reconfigure carries the configuration already in force — ignoring",
_connectionName);
return;
}
_log.Warning(
"[{0}] Reconfiguring in place (protocol {1} → {2}) — swapping the adapter and "
+ "re-subscribing {3} tag(s) for {4} instance(s) and {5} alarm source(s)",
_connectionName, _protocolType, command.ProtocolType,
_instancesByTag.Count, _subscriptionsByInstance.Count, _alarmSourceSubscribers.Count);
_protocolType = command.ProtocolType;
_primaryConfig = command.PrimaryConnectionDetails;
_backupConfig = command.BackupConnectionDetails;
_failoverRetryCount = command.FailoverRetryCount;
_activeEndpoint = ActiveEndpoint.Primary;
_consecutiveFailures = 0;
_consecutiveUnstableDisconnects = 0;
SwapAdapter(_primaryConfig);
BecomeReconnecting(afterReconfigure: true);
}
///
/// Order-insensitive value comparison of two flattened connection-detail maps, used only by
/// 's no-op guard.
///
private static bool DetailsEqual(
IDictionary? left, IDictionary? right)
{
if (ReferenceEquals(left, right)) return true;
if (left == null || right == null) return false;
if (left.Count != right.Count) return false;
foreach (var (key, value) in left)
{
if (!right.TryGetValue(key, out var other)
|| !string.Equals(value, other, StringComparison.Ordinal))
return false;
}
return true;
}
private void Reconnecting(object message)
{
switch (message)
{
case AttemptConnect:
HandleAttemptConnect();
break;
case ConnectResult result:
HandleReconnectResult(result);
break;
case AdapterDisconnected disconnected:
// Same gap as Connecting: a reconnect that succeeds and drops again before
// its ConnectResult is applied would otherwise be declared Connected on a dead
// adapter that will never raise Disconnected a second time.
NotePendingDisconnect(disconnected);
break;
case SubscribeTagsRequest pending:
QueueSubscribe(pending);
break;
case SubscribeAlarmsRequest pendingAlarm:
QueueSubscribeAlarms(pendingAlarm);
break;
case WriteTagRequest write:
RejectDisconnectedWrite(write);
break;
case WriteTagBatchRequest batch:
RejectDisconnectedWrite(batch);
break;
case UnsubscribeTagsRequest req:
// Allow unsubscribe even during reconnect (for cleanup on instance stop)
HandleUnsubscribe(req);
break;
case UnsubscribeAlarmsRequest areq:
// Allow alarm unsubscribe during reconnect (cleanup on instance stop).
HandleUnsubscribeAlarms(areq);
break;
case ReconfigureConnectionCommand rc:
// NEVER stashed — this is the state a connection to a wrong or dead endpoint
// spends its life in, i.e. exactly when a corrected configuration arrives.
HandleReconfigure(rc);
break;
case TagValueReceived:
// Ignore — stale callback from previous connection
break;
case AlarmTransitionReceived:
// Ignore — stale alarm callback from previous connection; ReSubscribeAll re-seeds.
break;
case BatchSubscribeCompleted:
// Ignore — stale results from previous connection; ReSubscribeAll runs after reconnect
break;
case RetryTagResolution:
// Ignore — the adapter has no live session; ReSubscribeAll's chunk
// completions re-arm the retry timer after reconnect.
break;
case QualityFlushTick:
FlushQualityCountersIfDirty();
break;
case SubscribeCompleted sc:
// A subscribe started while Connected can complete after a transition;
// apply it so its state survives into the next ReSubscribeAll.
HandleSubscribeCompleted(sc);
break;
case AlarmSubscribeCompleted asc:
HandleAlarmSubscribeCompleted(asc);
break;
case BrowseNodeCommand browse:
// Browse is design-time and never stashed. While reconnecting
// the adapter has no live session, so the adapter call will
// throw ConnectionNotConnectedException — mapped by HandleBrowse.
HandleBrowse(browse);
break;
case SearchAddressSpaceCommand search:
// Same rule as browse — never stashed; while reconnecting the
// adapter is not Connected so the adapter call throws
// ConnectionNotConnectedException, mapped by HandleSearch.
HandleSearch(search);
break;
case ReadTagValuesCommand read:
// Same rule as browse — never stashed; while reconnecting the
// adapter is not Connected so HandleReadTagValues short-circuits
// to a ConnectionNotConnected failure.
HandleReadTagValues(read);
break;
case Terminated terminated:
HandleSubscriberTerminated(terminated.ActorRef);
break;
case GetHealthReport:
ReplyWithHealthReport();
break;
default:
Unhandled(message);
break;
}
}
// ── Connection Management ──
private void HandleAttemptConnect()
{
_log.Debug("[{0}] Attempting connection...", _connectionName);
// Scope the pending-disconnect flag to THIS attempt. Anything noted before now
// belongs to a session that is already over and must not be replayed against the one
// about to start.
_disconnectDuringConnectAttempt = false;
var self = Self;
// Stamp the generation this attempt belongs to, exactly as every other piped
// completion does. A failover or an in-place reconfigure landing while ConnectAsync
// is still in flight swaps the adapter and bumps the counter, and the OUTGOING
// adapter's answer must never be applied to its replacement.
var generation = _adapterGeneration;
// Started on the thread pool, never inline: everything an adapter's ConnectAsync does
// before its first YIELDING await would otherwise run on this actor's dispatcher thread
// — OPC UA endpoint discovery was synchronous there and parked the thread for the SDK's
// operation timeout on every reconnect tick to a black-holed endpoint (review 2026-10-21
// 02/P2). Adapter and details are captured here, on the actor thread.
var adapter = _adapter;
var details = _connectionDetails;
Task.Run(() => adapter.ConnectAsync(details)).ContinueWith(t =>
{
if (t.IsCompletedSuccessfully)
return new ConnectResult(true, null, generation);
return new ConnectResult(false, t.Exception?.GetBaseException().Message, generation);
}).PipeTo(self);
}
///
/// True when this connect result was produced by an adapter that has since been swapped
/// out, in which case it is dropped WHOLE (arch-review 2026-08-19).
///
///
/// A success from the dead adapter used to drive and
/// against the NEW, never-connected adapter — and the new
/// adapter's own result then arrived in the Connected state, where nothing handles it. If the
/// new endpoint was bad the batch subscribes failed at connection level with a log line only,
/// no AdapterDisconnected ever fired, and the idempotent-reconfigure guard rejected
/// re-sending the same configuration: only a restart recovered the connection. A failure from
/// the dead adapter is just as wrong — it would count toward failover and re-arm the reconnect
/// timer underneath the attempt that is genuinely in flight.
///
///
///
/// Dropping cannot wedge the state machine: every site that bumps the generation
/// (, the failover branch of
/// , and through it) arms an
/// for the new generation, and the state stays Connecting /
/// Reconnecting until that attempt's own result arrives.
///
///
private bool IsStaleConnectResult(ConnectResult result)
{
if (result.AdapterGeneration == _adapterGeneration)
return false;
_log.Debug(
"[{0}] Dropping {1} connect result from adapter generation {2} (current {3}) — "
+ "the adapter it ran against has been replaced",
_connectionName, result.Success ? "successful" : "failed",
result.AdapterGeneration, _adapterGeneration);
return true;
}
///
/// True when this was raised by an adapter that has since
/// been swapped out (failover or in-place reconfigure), in which case it is dropped — the
/// disconnect twin of (arch-review 2026-08-19).
/// unhooks the outgoing adapter, but the event can already be
/// executing on a background thread when it does, so the message can still be delivered
/// after the swap.
///
private bool IsStaleDisconnect(AdapterDisconnected disconnected)
{
if (disconnected.AdapterGeneration == _adapterGeneration)
return false;
_log.Debug(
"[{0}] Dropping disconnect notification from adapter generation {1} (current {2}) — "
+ "the adapter that raised it has been replaced",
_connectionName, disconnected.AdapterGeneration, _adapterGeneration);
return true;
}
///
/// Records a current-generation disconnect that arrived while a connect attempt was still in
/// flight, so / can
/// refuse to declare the connection up. See .
///
private void NotePendingDisconnect(AdapterDisconnected disconnected)
{
if (IsStaleDisconnect(disconnected))
return;
_log.Warning(
"[{0}] Adapter reported a disconnect while a connect attempt was still in flight — "
+ "the attempt's result will be treated as a failed connection", _connectionName);
_disconnectDuringConnectAttempt = true;
}
///
/// Reads and clears , logging the demotion. True
/// means the successful connect result being applied belongs to a session the adapter has
/// already reported as dropped.
///
private bool ConsumeDisconnectDuringConnectAttempt()
{
if (!_disconnectDuringConnectAttempt)
return false;
_disconnectDuringConnectAttempt = false;
_log.Warning(
"[{0}] Connect succeeded but the adapter had already reported the session lost — "
+ "treating the attempt as a failed connect", _connectionName);
return true;
}
///
/// Applies a demoted connect success (see )
/// exactly as a connect FAILURE: the session the "success" describes is already gone, so it
/// must count toward the failover budget and re-arm the retry timer rather than declare the
/// connection up. Verbatim the tail of both result handlers' failure branches, which is the
/// point — a flapping endpoint that drops inside every connect window still reaches its
/// backup instead of retrying the same dead primary forever.
///
private void DemoteConnectAttemptToFailure()
{
_consecutiveFailures++;
CountFailureAndMaybeFailover(
"adapter reported the session lost before the connect result was applied");
Timers.StartSingleTimer("reconnect", new AttemptConnect(), _options.ReconnectInterval);
}
private void HandleConnectResult(ConnectResult result)
{
if (IsStaleConnectResult(result))
return;
if (result.Success)
{
// A disconnect that raced ahead of this result means the session this
// "success" describes is already gone; going Connected here would strand the
// connection on a dead adapter that never raises Disconnected again. Stay in
// Connecting and retry, exactly as a failed connect would.
if (ConsumeDisconnectDuringConnectAttempt())
{
DemoteConnectAttemptToFailure();
return;
}
_log.Info("[{0}] Connection established", _connectionName);
// Every connection transition reaches the site event log — the initial connect
// included, not only restore / loss / failover (review 2026-10-21 02/U4).
if (_siteEventLogger != null)
{
_ = _siteEventLogger.LogEventAsync(
"connection", "Info", null, _connectionName,
$"Connection established on {_activeEndpoint} endpoint", null);
}
BecomeConnected();
}
else
{
// The INITIAL connect must participate in the
// failover counter exactly like a reconnect. Without this a primary that is
// unreachable when the actor first starts (fresh deployment, site restart, or
// a primary simply down) is retried forever and the configured backup is
// never tried. Count the failure and switch endpoint once the retry count is
// exhausted, then re-arm the timer.
_consecutiveFailures++;
CountFailureAndMaybeFailover(result.Error);
Timers.StartSingleTimer("reconnect", new AttemptConnect(), _options.ReconnectInterval);
}
}
private void HandleReconnectResult(ConnectResult result)
{
if (IsStaleConnectResult(result))
return;
if (result.Success)
{
// Same demotion as the initial connect: the reconnect landed but the adapter
// has already reported the new session lost, so re-subscribing and going Connected
// would wedge on a dead adapter. Stay in Reconnecting and retry.
if (ConsumeDisconnectDuringConnectAttempt())
{
DemoteConnectAttemptToFailure();
return;
}
_log.Info("[{0}] Reconnected successfully on {1} endpoint", _connectionName, _activeEndpoint);
_consecutiveFailures = 0;
// Log restoration event to site event log
if (_siteEventLogger != null)
{
_ = _siteEventLogger.LogEventAsync(
"connection", "Info", null, _connectionName,
$"Connection restored on {_activeEndpoint} endpoint", null);
}
// The adapter object survived the drop but its SESSION did not. Advance the
// generation BEFORE re-subscribing so every completion, value and alarm callback
// started against the dead session takes the existing stale path (review
// 2026-10-21 02/C1) — exactly as it would after a swap.
AdvanceSessionGeneration();
// Transparent re-subscribe — re-establish all active subscriptions
ReSubscribeAll();
// Re-establish native alarm feeds (source replays a snapshot).
ReSubscribeAllAlarms();
BecomeConnected();
}
else
{
_consecutiveFailures++;
CountFailureAndMaybeFailover(result.Error);
Timers.StartSingleTimer("reconnect", new AttemptConnect(), _options.ReconnectInterval);
}
}
///
/// Shared connect-failure handling for both the initial connect (Connecting state)
/// and reconnect (Reconnecting state). Assumes has
/// already been incremented for the current failure. Switches to the other endpoint
/// once the retry count is exhausted and a backup is configured; the initial
/// connect follows this same path.
///
private void CountFailureAndMaybeFailover(string? error)
{
// Failover: switch endpoint after exhausting retry count (only if backup is configured)
if (_backupConfig != null && _consecutiveFailures >= _failoverRetryCount)
{
var previousEndpoint = _activeEndpoint;
_activeEndpoint = _activeEndpoint == ActiveEndpoint.Primary
? ActiveEndpoint.Backup
: ActiveEndpoint.Primary;
_consecutiveFailures = 0;
SwapAdapter(_activeEndpoint == ActiveEndpoint.Primary ? _primaryConfig : _backupConfig);
_log.Warning("[{0}] Failing over from {1} to {2}",
_connectionName, previousEndpoint, _activeEndpoint);
// Log failover event to site event log
if (_siteEventLogger != null)
{
_ = _siteEventLogger.LogEventAsync(
"connection", "Warning", null, _connectionName,
$"Failover from {previousEndpoint} to {_activeEndpoint}",
$"After {_failoverRetryCount} consecutive failures");
}
}
else
{
var retryLimit = _backupConfig != null ? _failoverRetryCount.ToString() : "∞";
_log.Warning("[{0}] Connect failed: {1}. Retrying in {2}s (attempt {3}/{4})",
_connectionName, error, _options.ReconnectInterval.TotalSeconds,
_consecutiveFailures, retryLimit);
}
}
private void HandleDisconnect()
{
_log.Warning("[{0}] AdapterDisconnected message received — transitioning to Reconnecting", _connectionName);
BecomeReconnecting();
}
// One latest request per owner/source while disconnected; retries cannot grow a stash.
private readonly Dictionary _pendingTagSubscriptions = new();
private readonly Dictionary<(IActorRef Owner, string Source), SubscribeAlarmsRequest> _pendingAlarmSubscriptions = new();
private void QueueSubscribe(SubscribeTagsRequest request)
{
Context.Watch(Sender);
_pendingTagSubscriptions.TryGetValue(request.InstanceUniqueName, out var previous);
_pendingTagSubscriptions[request.InstanceUniqueName] = (request, Sender);
if (previous.Owner is not null && !previous.Owner.Equals(Sender))
UnwatchIfUnused(previous.Owner);
}
private void QueueSubscribeAlarms(SubscribeAlarmsRequest request)
{
Context.Watch(Sender);
_pendingAlarmSubscriptions[(Sender, request.SourceReference)] = request;
}
private void RejectDisconnectedWrite(WriteTagRequest request) => Sender.Tell(new WriteTagResponse(
request.CorrelationId, false, "Connection is not connected; device write was not queued.", DateTimeOffset.UtcNow));
private void RejectDisconnectedWrite(WriteTagBatchRequest request) => Sender.Tell(new WriteTagBatchResponse(
request.CorrelationId, false, "Connection is not connected; device write was not queued.", DateTimeOffset.UtcNow));
private void UnwatchIfUnused(IActorRef owner)
{
if (!_subscribers.Values.Contains(owner)
&& !_alarmSourceSubscribers.Values.Any(subs => subs.Contains(owner))
&& !_pendingTagSubscriptions.Values.Any(p => p.Owner.Equals(owner))
&& !_pendingAlarmSubscriptions.Keys.Any(k => k.Owner.Equals(owner)))
Context.Unwatch(owner);
}
private void HandleSubscriberTerminated(IActorRef owner)
{
foreach (var instance in _pendingTagSubscriptions.Where(kv => kv.Value.Owner.Equals(owner)).Select(kv => kv.Key).ToList())
_pendingTagSubscriptions.Remove(instance);
foreach (var key in _pendingAlarmSubscriptions.Keys.Where(k => k.Owner.Equals(owner)).ToList())
_pendingAlarmSubscriptions.Remove(key);
foreach (var instance in _subscribers.Where(kv => kv.Value.Equals(owner)).Select(kv => kv.Key).ToList())
RemoveTagSubscriber(instance, owner);
foreach (var source in _alarmSourceSubscribers.Where(kv => kv.Value.Contains(owner)).Select(kv => kv.Key).ToList())
RemoveAlarmSubscriber(source, owner);
}
// ── Subscription Management ──
private void HandleSubscribe(SubscribeTagsRequest request)
{
_log.Debug("[{0}] Subscribing {1} tags for instance {2}",
_connectionName, request.TagPaths.Count, request.InstanceUniqueName);
Context.Watch(Sender);
if (_subscribers.TryGetValue(request.InstanceUniqueName, out var previousOwner) && !previousOwner.Equals(Sender))
RemoveTagSubscriber(request.InstanceUniqueName, previousOwner, request.TagPaths.ToHashSet());
_subscribers[request.InstanceUniqueName] = Sender;
if (previousOwner is not null && !previousOwner.Equals(Sender))
UnwatchIfUnused(previousOwner);
if (!_subscriptionsByInstance.TryGetValue(request.InstanceUniqueName, out var requestedTags))
_subscriptionsByInstance[request.InstanceUniqueName] = requestedTags = new HashSet();
// Record demand before I/O, so a replacement owner can adopt a shared in-flight result.
foreach (var tag in request.TagPaths)
{
requestedTags.Add(tag);
IndexTag(tag, request.InstanceUniqueName);
}
var self = Self;
var sender = Sender;
// Capture the current adapter generation so callbacks
// from this adapter can be distinguished from a later (post-failover) adapter.
var generation = _adapterGeneration;
// Capture the adapter reference on the actor thread too (S7). The background task
// below iterates multiple tags with an await between each SubscribeAsync — a failover
// that swaps the mutable _adapter field mid-loop would otherwise make later iterations
// (and the SeedTagsAsync read) run against the NEW adapter while carrying the OLD
// generation. The in-flight subscribe must finish against the adapter that started it;
// the generation guard drops any value the swapped-out adapter later produces.
var adapter = _adapter;
// Partition tags on the actor thread into "this
// request will issue _adapter.SubscribeAsync" vs. "already subscribed (by us
// or by another in-flight SubscribeTagsRequest)". A tag that is already in
// _subscriptionIds OR currently in _subscribesInFlight is treated as
// AlreadySubscribed — the eventual SubscribeCompleted of the in-flight
// request will populate _subscriptionIds, at which point a subsequent
// unsubscribe by either instance correctly references the adapter handle.
// The background task below must NOT read or mutate actor state — these
// partitioned lists are the only state it sees.
var tagsToSubscribe = new List(request.TagPaths.Count);
var preResolvedResults = new List();
foreach (var tagPath in request.TagPaths)
{
if (_subscriptionIds.ContainsKey(tagPath) || _subscribesInFlight.Contains(tagPath) || _resolutionInFlight.Contains(tagPath))
{
preResolvedResults.Add(new SubscribeTagResult(
tagPath, AlreadySubscribed: true, Success: true, null, null));
}
else
{
tagsToSubscribe.Add(tagPath);
_subscribesInFlight.Add(tagPath);
}
}
// ONE batch subscribe per chunk instead of one adapter round trip per tag. The
// Deployment Manager already staggers instance startup (SiteRuntimeOptions
// StartupBatchSize/Delay), so requests arrive pre-spaced on this path and the DCL
// deliberately adds NO extra delay between chunks here — double-staggering would
// only slow failover. A typical request is ~75 tags, i.e. a single chunk.
var batchSize = Math.Max(1, _options.SubscribeBatchSize);
Task.Run(async () =>
{
var results = new List(request.TagPaths.Count);
results.AddRange(preResolvedResults);
var tagsToSeed = new List(preResolvedResults.Count + tagsToSubscribe.Count);
foreach (var r in preResolvedResults)
{
tagsToSeed.Add(r.TagPath);
}
for (var offset = 0; offset < tagsToSubscribe.Count; offset += batchSize)
{
var chunk = tagsToSubscribe.GetRange(
offset, Math.Min(batchSize, tagsToSubscribe.Count - offset));
var chunkResults = await SubscribeTagsAsync(adapter, chunk, (path, value) =>
{
self.Tell(new TagValueReceived(path, value, generation));
});
results.AddRange(chunkResults);
foreach (var r in chunkResults)
{
if (r.Success)
tagsToSeed.Add(r.TagPath);
}
}
// Initial read — capture current values for resolved tags so the Instance
// Actor doesn't stay Uncertain until the next data-change notification.
// These are NOT delivered here. Emitting a
// TagValueReceived now (inside the background subscribe task) races ahead of
// the SubscribeCompleted that registers this instance's tags in
// _subscriptionsByInstance, so HandleTagValueReceived's fan-out finds no
// subscriber for the tag and drops the value. That's harmless for a tag that
// soon gets a real change notification, but for a STATIC tag (e.g. an idle
// MES field that never changes) the dropped seed is the only value it will
// ever produce — leaving the attribute Uncertain forever. So the seeds ride
// back on SubscribeCompleted and are delivered after registration.
// SeedTagsAsync retries the still-empty reads so a
// seed that races the just-created advise (returns VT_EMPTY) is not silently
// dropped, and logs any tag that never yields a value.
var seedValues = await SeedTagsAsync(adapter, tagsToSeed);
return new SubscribeCompleted(request, sender, results, seedValues, generation, adapter);
}).PipeTo(self);
}
///
/// Subscribes a CHUNK of tags against and returns one row
/// per requested tag. An adapter that advertises
/// does it in one round trip; any other
/// adapter falls back to the historical per-tag loop, so the capability is purely
/// additive.
///
///
/// Partial-failure contract: a per-tag fault is a result row with
/// Success:false. A THROWN exception means the whole chunk failed, and is
/// classified with exactly as the per-tag path
/// does — connection-level faults drive the reconnect state machine, everything else
/// is treated as a tag-resolution failure and retried on the backoff timer.
///
///
/// Runs on a background task: it touches no actor state.
///
private static async Task> SubscribeTagsAsync(
IDataConnection adapter, IReadOnlyList tags, SubscriptionCallback callback)
{
var results = new List(tags.Count);
if (tags.Count == 0)
return results;
if (adapter is IBatchSubscribableConnection batchAdapter)
{
IReadOnlyList rows;
try
{
rows = await batchAdapter.SubscribeBatchAsync(tags, callback);
}
catch (Exception ex)
{
var connectionLevel = IsConnectionLevelFailure(ex);
foreach (var tagPath in tags)
{
results.Add(new SubscribeTagResult(
tagPath, AlreadySubscribed: false, Success: false, null, ex.Message,
ConnectionLevelFailure: connectionLevel));
}
return results;
}
var byTag = new Dictionary(rows.Count, StringComparer.Ordinal);
foreach (var row in rows)
byTag[row.TagPath] = row;
foreach (var tagPath in tags)
{
results.Add(byTag.TryGetValue(tagPath, out var row)
? new SubscribeTagResult(
tagPath, AlreadySubscribed: false, row.Success, row.SubscriptionId, row.ErrorMessage)
: new SubscribeTagResult(
tagPath, AlreadySubscribed: false, Success: false, null,
"Adapter returned no subscribe result for this tag."));
}
return results;
}
foreach (var tagPath in tags)
{
try
{
var subId = await adapter.SubscribeAsync(tagPath, callback);
results.Add(new SubscribeTagResult(tagPath, AlreadySubscribed: false, Success: true, subId, null));
}
catch (Exception ex)
{
// Distinguish a connection-level fault
// (adapter not connected / transport down) from a genuine
// node-not-found. Connection-level faults must drive the
// reconnection state machine, not be retried as unresolved tags.
var connectionLevel = IsConnectionLevelFailure(ex);
results.Add(new SubscribeTagResult(
tagPath, AlreadySubscribed: false, Success: false, null, ex.Message,
ConnectionLevelFailure: connectionLevel));
}
}
return results;
}
///
/// Releases adapter subscription handles, in one round trip where the adapter supports
/// it. Fire-and-forget at every call site (the actor never awaits a release), so
/// faults are swallowed the same way the per-tag UnsubscribeAsync calls were.
///
private static Task UnsubscribeIdsAsync(IDataConnection adapter, IReadOnlyList subscriptionIds)
{
if (subscriptionIds.Count == 0)
return Task.CompletedTask;
if (adapter is IBatchSubscribableConnection batchAdapter)
return batchAdapter.UnsubscribeBatchAsync(subscriptionIds);
return Task.WhenAll(subscriptionIds.Select(id => adapter.UnsubscribeAsync(id)));
}
///
/// Best-effort release of handles created by a subscribe whose adapter has since been
/// swapped out. must target the adapter that CREATED the handles
/// — they are adapter-scoped (MxGateway's are plain ints that collide across sessions), so
/// releasing by handle against the replacement adapter can tear down a live subscription.
/// The outgoing adapter is disposed fire-and-forget by and may
/// already be gone, so a failure here is expected and logged at Debug rather than surfaced.
///
private void ReleaseStaleAgainstOwnAdapter(object owner, string description, Func release)
{
// Same-adapter reconnect (review 2026-10-21 02/C1): the generation advanced because the
// SESSION changed, not the adapter. The dead session's handles died with it, and the
// adapter now speaks for the NEW session — where a session-scoped handle number (an
// MxGateway int) may already belong to a live item. Releasing would tear that down.
if (ReferenceEquals(owner, _adapter))
{
_log.Debug("[{0}] Not releasing {1}: it belongs to a session of the current adapter that "
+ "has since ended, so it died with that session", _connectionName, description);
return;
}
try
{
release().ContinueWith(
t => _log.Debug("[{0}] Best-effort release of {1} failed: {2}",
_connectionName, description, t.Exception?.GetBaseException().Message),
TaskContinuationOptions.OnlyOnFaulted);
}
catch (Exception ex)
{
_log.Debug("[{0}] Best-effort release of {1} threw: {2}",
_connectionName, description, ex.Message);
}
}
///
/// Reads the current value of each tag so the Instance Actor
/// has an initial value, retrying the still-empty subset a bounded number of times. A
/// STATIC tag (one that emits no further OnDataChange after the advise) depends
/// entirely on this seed; on a cold/fresh advise the read can race the just-created
/// subscription and return an empty/failed result, which pre-fix was swallowed —
/// leaving the attribute Uncertain forever even though the source reads Good.
///
///
/// Reads are issued as CHUNKED bulk reads (
/// tags per chunk, chunks in
/// flight) against a batch-capable adapter, and per tag otherwise.
/// now bounds a CHUNK rather than a
/// single tag — chunking plus that per-chunk bound answers the "some gateways time out
/// on a large batch" caveat that originally motivated per-tag reads — and the whole
/// seed (all chunks, all retry rounds) is bounded by
/// , replacing the old
/// 30s-per-tag serial worst case.
///
///
/// The retry delay is applied once per round across the whole pending subset, so total
/// added latency is bounded to (attempts - 1) × SeedReadRetryDelay regardless of
/// tag count. Tags still empty after the budget (or when the deadline expires) are
/// logged (named) and left to heal from a future change.
/// Runs on a background task: it reads only the supplied
/// and returns the seeds — all actor-state mutation/delivery stays on the actor thread.
///
private async Task> SeedTagsAsync(
IDataConnection adapter, IReadOnlyCollection tags, CancellationToken sessionToken = default)
{
var seedValues = new List(tags.Count);
if (tags.Count == 0)
return seedValues;
var pending = new HashSet(tags);
var attempts = Math.Max(1, _options.SeedReadMaxAttempts);
var chunkSize = Math.Max(1, _options.SeedReadBatchSize);
var parallelism = Math.Max(1, _options.SeedReadMaxParallelism);
var useBatch = adapter is IBatchSubscribableConnection;
// ONE deadline across every chunk and every retry round, also cut short when the
// session this seed belongs to ends (02/L4).
using var overall = CancellationTokenSource.CreateLinkedTokenSource(sessionToken);
overall.CancelAfter(_options.SeedOverallTimeout);
var deadlineHit = false;
for (var attempt = 1; attempt <= attempts && pending.Count > 0 && !overall.IsCancellationRequested; attempt++)
{
var round = pending.ToList();
var chunks = new List>();
for (var offset = 0; offset < round.Count; offset += chunkSize)
chunks.Add(round.GetRange(offset, Math.Min(chunkSize, round.Count - offset)));
var gate = new SemaphoreSlim(parallelism);
var chunkTasks = chunks.Select(async chunk =>
{
await gate.WaitAsync(CancellationToken.None);
try
{
return await ReadSeedChunkAsync(adapter, chunk, useBatch, overall.Token);
}
finally
{
gate.Release();
}
}).ToList();
var chunkResults = await Task.WhenAll(chunkTasks);
foreach (var seeded in chunkResults.SelectMany(r => r))
{
if (pending.Remove(seeded.TagPath))
seedValues.Add(seeded);
}
if (overall.IsCancellationRequested)
{
deadlineHit = true;
break;
}
if (pending.Count > 0 && attempt < attempts)
{
try
{
await Task.Delay(_options.SeedReadRetryDelay, overall.Token);
}
catch (OperationCanceledException)
{
deadlineHit = true;
break;
}
}
}
if (sessionToken.IsCancellationRequested)
{
_log.Debug("[{0}] Seed read abandoned — the session it belonged to has ended", _connectionName);
return seedValues;
}
if (pending.Count > 0)
{
const int maxNamed = 20;
var named = string.Join(", ", pending.Take(maxNamed));
if (pending.Count > maxNamed)
named += $", … (+{pending.Count - maxNamed} more)";
_log.Warning(
"[{0}] Seed read returned no value for {1} tag(s) after {2} attempt(s){3}; they stay Uncertain until a change notification arrives: {4}",
_connectionName, pending.Count, attempts,
deadlineHit ? $" (overall {_options.SeedOverallTimeout.TotalSeconds:F0}s seed deadline expired)" : string.Empty,
named);
}
return seedValues;
}
///
/// Reads one seed chunk, bounded by
/// and by the caller's overall seed deadline. Every failure mode (bad read, chunk
/// timeout, deadline) yields "no seed for these tags", which the caller retries or
/// leaves Uncertain — identical to the historical per-tag behaviour.
///
private async Task> ReadSeedChunkAsync(
IDataConnection adapter, IReadOnlyList chunk, bool useBatch, CancellationToken overallToken)
{
var seeded = new List(chunk.Count);
using var cts = CancellationTokenSource.CreateLinkedTokenSource(overallToken);
cts.CancelAfter(_options.SeedReadTimeout);
if (useBatch)
{
try
{
var results = await adapter.ReadBatchAsync(chunk, cts.Token);
foreach (var tagPath in chunk)
{
if (results.TryGetValue(tagPath, out var readResult)
&& readResult.Success && readResult.Value is { Value: not null } value)
{
seeded.Add(new SeededValue(tagPath, value));
}
}
}
catch
{
// Best-effort read — retried by the caller, or logged once the budget is
// spent. Includes OperationCanceledException on the chunk/overall deadline.
}
return seeded;
}
foreach (var tagPath in chunk)
{
try
{
var readResult = await adapter.ReadAsync(tagPath, cts.Token);
if (readResult.Success && readResult.Value is { Value: not null } value)
seeded.Add(new SeededValue(tagPath, value));
}
catch
{
// Best-effort read — see above.
}
}
return seeded;
}
///
/// Applies the result of an asynchronous subscribe on the actor thread. ALL mutation
/// of subscription state and counters happens here — never on the background task —
/// so the actor model's single-threaded state guarantee holds.
/// Returns true if any tag failed at connection level,
/// signalling the caller (only the Connected state) to enter Reconnecting.
///
private bool HandleSubscribeCompleted(SubscribeCompleted msg)
{
var instanceName = msg.Request.InstanceUniqueName;
// Results produced against an adapter that has since been swapped out (failover or
// in-place reconfigure) are dropped WHOLE, mirroring the BatchSubscribeCompleted and
// TagValueReceived generation guards. An initial subscribe can take up to
// SeedOverallTimeout to complete, so this window is wide. Applying a stale completion
// stores a dead session's handle under the tag's key, which then makes the LIVE
// completion for the same tag look like a duplicate and get RELEASED (the tag reads
// resolved forever and delivers nothing); re-stamps the old endpoint's seed values with
// the CURRENT generation, so they bypass the stale-data guard and present as fresh; and
// lets a connection-level failure row from the dead adapter knock the healthy new
// connection back into Reconnecting.
if (msg.AdapterGeneration != _adapterGeneration)
{
var staleIds = msg.Results
.Where(r => r is { Success: true, AlreadySubscribed: false, SubscriptionId: not null })
.Select(r => r.SubscriptionId!)
.ToList();
_log.Debug(
"[{0}] Dropping stale subscribe completion for instance {1} from adapter " +
"generation {2} (current {3}); releasing {4} handle(s) against the adapter that created them",
_connectionName, instanceName, msg.AdapterGeneration, _adapterGeneration, staleIds.Count);
// Release against msg.Adapter, NEVER _adapter: handles are adapter-scoped and
// MxGateway's are plain ints that collide across sessions, so releasing by handle
// against the replacement adapter can tear down a live subscription.
ReleaseStaleAgainstOwnAdapter(
msg.Adapter,
$"{staleIds.Count} tag subscription handle(s) from adapter generation {msg.AdapterGeneration}",
() => UnsubscribeIdsAsync(msg.Adapter, staleIds));
// _subscribesInFlight is deliberately NOT cleared here: ReSubscribeAll empties it
// wholesale once the replacement adapter connects, and a marker the new generation
// has since placed for the same tag must not be removed by this dead message.
// Demand was registered before I/O, so ReSubscribeAll can recover it on the
// replacement adapter. Reply failure as well so the owner's correlated retry
// state never treats this stale completion as an established subscription.
msg.ReplyTo.Tell(new SubscribeTagsResponse(
msg.Request.CorrelationId, instanceName, false,
"adapter replaced while the subscribe was in flight — will retry",
DateTimeOffset.UtcNow));
return false;
}
if (!_subscribers.TryGetValue(instanceName, out var currentOwner) || !currentOwner.Equals(msg.ReplyTo)
|| !_subscriptionsByInstance.TryGetValue(instanceName, out var instanceTags))
{
// The initiating owner may be gone, but another instance/replacement can
// still need these tags. Reconcile against CURRENT per-tag demand instead
// of unconditionally releasing handles owned by shared in-flight work.
foreach (var row in msg.Results.Where(r => !r.AlreadySubscribed))
_subscribesInFlight.Remove(row.TagPath);
HandleBatchSubscribeCompleted(new BatchSubscribeCompleted(
msg.Results.Where(r => !r.AlreadySubscribed).ToList(), msg.AdapterGeneration, BatchSubscribeSource.Resubscribe));
return msg.Results.Any(r => r.ConnectionLevelFailure && _instancesByTag.ContainsKey(r.TagPath));
}
// If any tag failed because the adapter is not
// connected (a connection-level fault), the subscribe needs the reconnection
// state machine, not the tag-resolution retry. Drive a disconnect and let the
// request be re-stashed/retried after reconnect via ReSubscribeAll.
var connectionLevelFailure = msg.Results.Any(r => !r.Success && r.ConnectionLevelFailure);
// Handles this request created that turn out to be redundant; released in ONE
// round trip below, symmetric with HandleBatchSubscribeCompleted.
var redundantIds = new List();
foreach (var result in msg.Results)
{
// A result with AlreadySubscribed: false means
// this request was responsible for the SubscribeAsync call — the tag
// was added to _subscribesInFlight in HandleSubscribe. Clear it now so
// a later SubscribeTagsRequest for the same tag isn't forever treated
// as in-flight. AlreadySubscribed: true tags were not added to the
// set (another request owned the in-flight slot).
if (!result.AlreadySubscribed)
_subscribesInFlight.Remove(result.TagPath);
// Register the tag against this instance in both directions. The per-tag
// counted set _instancesByTag is what makes the tag "subscribed" for every
// consumer — the fan-out, the last-subscriber test in HandleUnsubscribe, the
// in-flight-unsubscribe discard gates, and TotalSubscribedTagCount. Both
// adds are idempotent, so this runs unconditionally: there is no separate
// counter left that a repeated add could inflate.
instanceTags.Add(result.TagPath);
IndexTag(result.TagPath, instanceName);
// Re-check against current state: another subscribe may have resolved the
// same tag while this request's I/O was in flight.
// AlreadySubscribed rows are partitioned out in HandleSubscribe and carry NO
// SubscriptionId of their own (another caller owns the adapter handle), so
// there is nothing to release for them.
if (result.AlreadySubscribed)
continue;
if (_subscriptionIds.ContainsKey(result.TagPath))
{
// Another path (a resolution probe, a reconnect re-subscribe or a
// concurrent request for a different instance) stored a handle for this
// tag while this request's I/O was in flight. This request issued its OWN
// SubscribeAsync, so dropping the row without releasing its handle leaks a
// monitored item forever — _subscriptionIds holds a single id per tag, so
// no later unsubscribe can ever reference this one. Release it, exactly as
// HandleBatchSubscribeCompleted does for its duplicate rows.
if (result is { Success: true, SubscriptionId: not null }
&& _subscriptionIds[result.TagPath] != result.SubscriptionId)
redundantIds.Add(result.SubscriptionId);
continue;
}
if (result.Success)
{
// Storing the handle IS the resolution: ResolvedTagCount is
// _subscriptionIds.Count, and the tag is already in the per-tag counted
// set, so it is already inside TotalSubscribedTagCount. The former
// fresh-subscribe vs. unresolved→resolved-promotion split existed only to
// decide which of the two scalar counters to bump (a promotion had to move
// _resolvedTags WITHOUT re-bumping _totalSubscribed, or a tag that a second
// instance resolved after a first instance failed it counted twice.
// Derived counts get both cases right for free;
// all that is left of the promotion is dropping the retry bookkeeping.
_subscriptionIds[result.TagPath] = result.SubscriptionId!;
if (_unresolvedTags.Remove(result.TagPath))
_resolutionInFlight.Remove(result.TagPath);
}
else if (result.ConnectionLevelFailure)
{
// Connection-level fault — do not count as an unresolved tag.
// ReSubscribeAll after reconnect derives the tag from
// _subscriptionsByInstance (already updated above).
_log.Warning("[{0}] Subscribe for {1} failed at connection level: {2}",
_connectionName, result.TagPath, result.Error);
}
else
{
// Genuine tag resolution failure — mark unresolved so the periodic retry
// timer picks it up. The tag still counts toward TotalSubscribedTagCount
// (an instance subscribes to it, so it is in the per-tag counted set); it
// simply does not count as resolved. Two instances failing the SAME tag is
// still one logical tag in both counts, because both counts are set sizes
// rather than accumulated increments.
_unresolvedTags.Add(result.TagPath);
_log.Debug("[{0}] Tag resolution failed for {1}: {2}",
_connectionName, result.TagPath, result.Error);
// Design doc Tag Path Resolution step 2:
// mark the attribute quality `bad` so the Instance Actor sees a
// signal rather than staying Uncertain indefinitely.
if (_subscribers.TryGetValue(instanceName, out var subscriber))
{
subscriber.Tell(new TagValueUpdate(
_connectionName, result.TagPath, null, QualityCode.Bad, DateTimeOffset.UtcNow));
}
}
}
// Fire-and-forget release of every handle this request created redundantly, against the
// adapter that created them (see the orphaned-release site above).
_ = UnsubscribeIdsAsync(msg.Adapter, redundantIds);
// Publish the resolution counts this request just changed. Without this push a
// connection whose tags ALL resolve on the first attempt reported 0/0 forever:
// the only other pushes are BecomeConnected's (which fires before any instance
// has subscribed, so its 0/0 is legitimate but is never superseded),
// HandleBatchSubscribeCompleted's (reconnect re-subscribe / resolution probe) and
// HandleUnsubscribe's (teardown) — and ScheduleTagResolutionRetry below cancels
// the retry timer outright when nothing is unresolved, so no later tick pushed
// either. The blind spot was therefore worst on a perfectly healthy connection,
// which is precisely when the "device still booting / tag path unresolved"
// diagnostic is supposed to read clean rather than empty. Mirrors the push site
// in HandleBatchSubscribeCompleted, and stays on the actor thread where both
// derived counts are stable.
_healthCollector.UpdateTagResolution(_connectionName, TotalSubscribedTagCount, ResolvedTagCount);
// Now that every tag is registered in
// _subscriptionsByInstance, deliver the values captured by the initial read.
// Re-entering via Self reuses HandleTagValueReceived's generation guard, fan-out
// and quality accounting — and crucially runs AFTER registration, so the value
// is no longer dropped. Only resolved tags (in _subscriptionIds) are seeded; an
// unresolved tag already got a Bad-quality update above and must not be masked.
if (!connectionLevelFailure)
{
foreach (var seed in msg.SeedValues)
{
if (_subscriptionIds.ContainsKey(seed.TagPath))
Self.Tell(new TagValueReceived(seed.TagPath, seed.Value, _adapterGeneration));
}
}
// Start the tag-resolution retry timer if any tags are unresolved.
ScheduleTagResolutionRetry();
// The response must match the actor's own assessment.
// When a connection-level failure is driving the actor into Reconnecting, the
// tags were never subscribed at the adapter — replying Success: true would tell
// the Instance Actor the subscribe succeeded when it did not. Genuine
// tag-resolution failures stay Success: true (they are a runtime quality concern
// tracked via _unresolvedTags, with a Bad-quality TagValueUpdate already pushed).
msg.ReplyTo.Tell(connectionLevelFailure
? new SubscribeTagsResponse(
msg.Request.CorrelationId, instanceName, false,
"connection unavailable — will re-subscribe on reconnect", DateTimeOffset.UtcNow)
: new SubscribeTagsResponse(
msg.Request.CorrelationId, instanceName, true, null, DateTimeOffset.UtcNow));
// The caller (Connected state only) decides whether to enter Reconnecting.
// In Connecting/Reconnecting the connection is not established anyway, so the
// existing reconnect cycle handles recovery without a re-trigger here.
return connectionLevelFailure;
}
///
/// Classifies a subscribe exception as a connection-level
/// fault (adapter not connected / transport down) versus a genuine tag-resolution
/// failure (the node does not exist on the device). Connection-level faults must
/// drive the reconnection state machine; resolution failures are retried on the
/// tag-resolution timer.
///
private static bool IsConnectionLevelFailure(Exception ex)
{
var baseEx = ex is AggregateException agg ? agg.GetBaseException() : ex;
return baseEx is InvalidOperationException
or System.Net.Sockets.SocketException
or TimeoutException
or System.IO.IOException
|| (baseEx is Opc.Ua.ServiceResultException sre && IsConnectionLevelStatus(sre.StatusCode));
}
///
/// OPC UA status codes that mean the SESSION or the SECURE CHANNEL is gone, not that one
/// node is unusable (arch-review 2026-08-19). The SDK reports a dead session by
/// throwing out of the batch subscribe, which
/// classified as a per-tag fault: every tag in the chunk was marked unresolved and each
/// backoff probe re-issued the whole batch against the same dead session, growing the
/// adapter's monitored-item bookkeeping on every round instead of reconnecting once.
///
///
/// Deliberately narrow. A per-node code — BadNodeIdUnknown, BadNodeIdInvalid,
/// BadUserAccessDenied — must stay a tag-resolution failure, because that is exactly
/// the "device still booting / tag path wrong" case the retry timer exists for; treating it
/// as connection-level would put the connection into a reconnect loop over one bad tag path.
///
///
private static bool IsConnectionLevelStatus(uint statusCode) => statusCode is
Opc.Ua.StatusCodes.BadSessionIdInvalid
or Opc.Ua.StatusCodes.BadSessionClosed
or Opc.Ua.StatusCodes.BadSessionNotActivated
or Opc.Ua.StatusCodes.BadSecureChannelClosed
or Opc.Ua.StatusCodes.BadSecureChannelIdInvalid
or Opc.Ua.StatusCodes.BadServerNotConnected
or Opc.Ua.StatusCodes.BadConnectionClosed
or Opc.Ua.StatusCodes.BadCommunicationError
or Opc.Ua.StatusCodes.BadNotConnected;
private void HandleUnsubscribe(UnsubscribeTagsRequest request)
{
if (_pendingTagSubscriptions.TryGetValue(request.InstanceUniqueName, out var pending) && pending.Owner.Equals(Sender))
_pendingTagSubscriptions.Remove(request.InstanceUniqueName);
RemoveTagSubscriber(request.InstanceUniqueName, Sender);
UnwatchIfUnused(Sender);
}
private void RemoveTagSubscriber(string instanceName, IActorRef owner, HashSet? preserveTags = null)
{
if (!_subscribers.TryGetValue(instanceName, out var currentOwner) || !currentOwner.Equals(owner)
|| !_subscriptionsByInstance.TryGetValue(instanceName, out var tags))
return;
// Released adapter handles are collected and released in ONE round trip below
// (batch-capable adapters) instead of one call per tag.
var idsToRelease = new List();
// Cleanup on Instance Actor stop
foreach (var tagPath in tags.Where(t => preserveTags == null || !preserveTags.Contains(t)).ToList())
{
// Drop this instance from the per-tag counted set. UnindexTag removes the tag
// key entirely when its last subscriber leaves, so "the key is gone" IS the
// last-subscriber test — O(1), and with no parallel reference count that could
// ever disagree with it. TotalSubscribedTagCount drops by exactly the number of
// keys this loop retires, whatever state the tag was in (resolved, unresolved,
// or mid-reconnect with both maps cleared — the case the retired scalar
// counters silently failed to decrement at all).
tags.Remove(tagPath);
UnindexTag(tagPath, instanceName);
if (_instancesByTag.ContainsKey(tagPath))
continue;
// Last subscriber gone. A tag with a subscription id is a resolved tag and its
// adapter handle must be released; an unresolved tag never has one.
if (_subscriptionIds.Remove(tagPath, out var subId))
{
idsToRelease.Add(subId);
// Drop the tag's tracked quality so it is no
// longer counted by PushBadQualityForAllTags (which sets _tagsBadQuality
// from _lastTagQuality.Count). Leaving it here drifts the quality
// counters above the reported subscribed total across disconnect cycles.
if (_lastTagQuality.Remove(tagPath, out var droppedQuality))
{
switch (droppedQuality)
{
case QualityCode.Good: _tagsGoodQuality--; break;
case QualityCode.Bad: _tagsBadQuality--; break;
case QualityCode.Uncertain: _tagsUncertainQuality--; break;
}
}
}
// Stop probing a tag nobody subscribes to any more. Unconditional and
// idempotent: a tag is in at most one of these sets (and in NEITHER during a
// reconnect window, where ReSubscribeAll has just cleared both), so there is
// no branch here to get wrong.
_unresolvedTags.Remove(tagPath);
_resolutionInFlight.Remove(tagPath);
}
if (preserveTags is null)
{
_subscriptionsByInstance.Remove(instanceName);
_subscribers.Remove(instanceName);
UnwatchIfUnused(owner);
}
// Fire-and-forget release of every handle this unsubscribe orphaned.
_ = UnsubscribeIdsAsync(_adapter, idsToRelease);
// Keep the reported quality counters in sync after the
// unsubscribed tags' buckets were decremented above. SYNCHRONOUS flush: the
// coalescing timer must never delay a count that just dropped.
FlushQualityCounters();
_healthCollector.UpdateTagResolution(_connectionName, TotalSubscribedTagCount, ResolvedTagCount);
}
// ── Write Support ──
private void HandleWrite(WriteTagRequest request)
{
_log.Debug("[{0}] Writing to tag {1}", _connectionName, request.TagPath);
var sender = Sender;
// Bound the write with WriteTimeout. A hung device
// write (TCP black-hole) would otherwise never complete, so PipeTo never
// fires and the calling script gets no DCL-level error. The CancellationToken
// is passed to the adapter; on timeout we translate cancellation into a
// failed WriteTagResponse so the failure is returned synchronously.
var cts = new CancellationTokenSource(_options.WriteTimeout);
// Capture the adapter on the actor thread (S7) so the write completes against the
// adapter that started it even if a failover swaps the _adapter field mid-flight.
var adapter = _adapter;
// Write through DCL to device, failure returned synchronously
adapter.WriteAsync(request.TagPath, request.Value, cts.Token).ContinueWith(t =>
{
cts.Dispose();
if (t.IsCompletedSuccessfully)
{
var result = t.Result;
return new WriteTagResponse(
request.CorrelationId, result.Success, result.ErrorMessage, DateTimeOffset.UtcNow);
}
if (t.IsCanceled || t.Exception?.GetBaseException() is OperationCanceledException)
{
return new WriteTagResponse(
request.CorrelationId, false,
$"Write timeout after {_options.WriteTimeout.TotalSeconds:F0}s", DateTimeOffset.UtcNow);
}
return new WriteTagResponse(
request.CorrelationId, false, t.Exception?.GetBaseException().Message, DateTimeOffset.UtcNow);
}).PipeTo(sender);
}
///
/// Batch write counterpart of . Writes every value in
/// to the device in ONE adapter
/// WriteBatchAsync round-trip, then (if present) writes the trigger flag with
/// a single WriteAsync. Both legs share one
/// budget. Any failed value in the batch, a failed trigger,
/// the timeout, or an adapter exception is translated into a failed
/// returned synchronously to the caller — never a
/// dropped reply. NOTE: this deliberately composes the batch + trigger primitives and
/// uses the EXISTING event-driven WaitForAttribute waiter for the wait half (the
/// adapters' former poll-based WriteBatchAndWaitAsync was removed — review
/// 2026-10-21 02/U3 — so no polling handshake can bypass that waiter).
///
private void HandleWriteBatch(WriteTagBatchRequest request)
{
_log.Debug("[{0}] Batch-writing {1} tag(s)", _connectionName, request.Values.Count);
var sender = Sender;
var cts = new CancellationTokenSource(_options.WriteTimeout);
// Capture the adapter on the actor thread (S7). The trigger write below runs AFTER
// the batch await, so re-reading the _adapter field there could hit a post-failover
// adapter; both legs must use the adapter that started this batch.
var adapter = _adapter;
async Task RunAsync()
{
try
{
var results = await adapter.WriteBatchAsync(
new Dictionary(request.Values), cts.Token);
var failed = results.Values.Where(r => !r.Success).Select(r => r.ErrorMessage).ToList();
if (failed.Count > 0)
return new WriteTagBatchResponse(
request.CorrelationId, false,
"Batch write failed: " + string.Join("; ", failed), DateTimeOffset.UtcNow);
if (!string.IsNullOrEmpty(request.TriggerTagPath))
{
var tr = await adapter.WriteAsync(request.TriggerTagPath, request.TriggerValue, cts.Token);
if (!tr.Success)
return new WriteTagBatchResponse(
request.CorrelationId, false,
"Trigger write failed: " + tr.ErrorMessage, DateTimeOffset.UtcNow);
}
return new WriteTagBatchResponse(request.CorrelationId, true, null, DateTimeOffset.UtcNow);
}
catch (OperationCanceledException)
{
return new WriteTagBatchResponse(
request.CorrelationId, false,
$"Write timeout after {_options.WriteTimeout.TotalSeconds:F0}s", DateTimeOffset.UtcNow);
}
catch (Exception ex)
{
return new WriteTagBatchResponse(
request.CorrelationId, false, ex.GetBaseException().Message, DateTimeOffset.UtcNow);
}
finally
{
cts.Dispose();
}
}
RunAsync().PipeTo(sender);
}
// ── OPC UA Tag Browser (interactive design-time query) ──
///
/// Handles a forwarded by the
/// . The capability check (does
/// this adapter support browsing?) and all browse-failure mapping live
/// here because the adapter is held by this actor, not the manager.
///
/// Failure mapping:
///
/// - — adapter is not , or it threw (browsable adapter, but the server/protocol cannot browse — e.g. a gateway build predating the browse RPC); message carried verbatim in the latter case.
/// - — adapter threw .
/// - — adapter threw .
/// - — any other exception, message carried verbatim.
///
///
/// The reply is sent via PipeTo(sender) — the same pattern used by
/// — so the captured is
/// safe to use from the continuation (which runs off the actor thread).
///
private void HandleBrowse(BrowseNodeCommand command)
{
var sender = Sender;
if (_adapter is not IBrowsableDataConnection browsable)
{
_log.Debug("[{0}] Browse requested but adapter does not implement IBrowsableDataConnection", _connectionName);
sender.Tell(new BrowseNodeResult(
Array.Empty(),
Truncated: false,
new BrowseFailure(
BrowseFailureKind.NotBrowsable,
$"Connection '{_connectionName}' does not support browsing.")));
return;
}
_log.Debug("[{0}] Browsing children of {1}", _connectionName, command.ParentNodeId ?? "(root)");
browsable.BrowseChildrenAsync(command.ParentNodeId, command.ContinuationToken).ContinueWith(t =>
{
if (t.IsCompletedSuccessfully)
{
// Bound the reply to stay under Akka's remote frame size before it
// crosses the site→central boundary (see CapBrowseChildren).
var (children, truncated) = CapBrowseChildren(t.Result.Children, t.Result.Truncated);
// Carry the adapter's continuation cursor through so the UI can ask
// for the next page (BrowseNext). Null when this is the final page.
return new BrowseNodeResult(children, truncated, Failure: null, t.Result.ContinuationToken);
}
var baseEx = t.Exception?.GetBaseException();
return baseEx switch
{
ConnectionNotConnectedException notConnected => new BrowseNodeResult(
Array.Empty(),
Truncated: false,
new BrowseFailure(BrowseFailureKind.ConnectionNotConnected, notConnected.Message)),
OperationCanceledException => new BrowseNodeResult(
Array.Empty(),
Truncated: false,
new BrowseFailure(BrowseFailureKind.Timeout, "Browse cancelled.")),
// Adapter reachable but the protocol/server cannot browse (e.g. an
// MxGateway build that predates the BrowseChildren RPC). Carry the
// adapter's explanatory message through as NotBrowsable.
NotSupportedException notSupported => new BrowseNodeResult(
Array.Empty(),
Truncated: false,
new BrowseFailure(BrowseFailureKind.NotBrowsable, notSupported.Message)),
_ => new BrowseNodeResult(
Array.Empty(),
Truncated: false,
new BrowseFailure(
BrowseFailureKind.ServerError,
baseEx?.Message ?? "Unknown browse error.")),
};
}).PipeTo(sender);
}
///
/// Handles a forwarded by the
/// — the address-space analogue of
/// . The capability check (does this adapter support
/// search?) and all failure mapping live here because the adapter is held by
/// this actor, not the manager. The search path reuses the browse
/// kinds rather than inventing a parallel set.
///
/// Failure mapping:
///
/// - — adapter is not , or it threw (searchable adapter, but the server/protocol cannot search); message carried verbatim in the latter case.
/// - — adapter threw .
/// - — adapter threw .
/// - — any other exception, message carried verbatim.
///
///
/// The adapter already caps the match list at MaxResults, so unlike
/// there is no frame-budget clip here — the bound is
/// supplied by the caller (B6) and chosen to stay well under Akka's remote
/// frame size. The reply is sent via PipeTo(sender), the same pattern
/// used by .
///
private void HandleSearch(SearchAddressSpaceCommand command)
{
var sender = Sender;
if (_adapter is not IAddressSpaceSearchable searchable)
{
_log.Debug("[{0}] Search requested but adapter does not implement IAddressSpaceSearchable", _connectionName);
sender.Tell(new SearchAddressSpaceResult(
Array.Empty(),
CapReached: false,
new BrowseFailure(
BrowseFailureKind.NotBrowsable,
$"Connection '{_connectionName}' does not support search.")));
return;
}
_log.Debug("[{0}] Searching address space for '{1}' (maxDepth={2}, maxResults={3})",
_connectionName, command.Query, command.MaxDepth, command.MaxResults);
searchable.SearchAddressSpaceAsync(command.Query, command.MaxDepth, command.MaxResults).ContinueWith(t =>
{
if (t.IsCompletedSuccessfully)
{
// The adapter already bounded the walk by MaxResults — pass the
// matches and the cap flag straight through.
return new SearchAddressSpaceResult(t.Result.Matches, t.Result.CapReached, Failure: null);
}
var baseEx = t.Exception?.GetBaseException();
return baseEx switch
{
ConnectionNotConnectedException notConnected => new SearchAddressSpaceResult(
Array.Empty(),
CapReached: false,
new BrowseFailure(BrowseFailureKind.ConnectionNotConnected, notConnected.Message)),
OperationCanceledException => new SearchAddressSpaceResult(
Array.Empty(),
CapReached: false,
new BrowseFailure(BrowseFailureKind.Timeout, "Search cancelled.")),
// Adapter reachable but the protocol/server cannot search. Carry
// the adapter's explanatory message through as NotBrowsable.
NotSupportedException notSupported => new SearchAddressSpaceResult(
Array.Empty(),
CapReached: false,
new BrowseFailure(BrowseFailureKind.NotBrowsable, notSupported.Message)),
_ => new SearchAddressSpaceResult(
Array.Empty(),
CapReached: false,
new BrowseFailure(
BrowseFailureKind.ServerError,
baseEx?.Message ?? "Unknown search error.")),
};
}).PipeTo(sender);
}
///
/// Estimated-byte ceiling for a single , kept
/// comfortably below Akka's default 128 KB remote frame size. A browse reply
/// crosses the site→central frame on a temp Ask actor; an oversized reply is
/// silently discarded by remoting (the picker then hangs on "loading…"). The
/// limit is a byte budget rather than a child count because the only thing
/// that actually consumes frame space is serialized size — OPC UA NodeIds and
/// MxGateway tag references vary widely in length, so a fixed count is not a
/// safe proxy.
///
private const int BrowseResultByteBudget = 100 * 1024;
///
/// Truncates a browse child list to using
/// a conservative per-node size estimate (JSON structural overhead plus the two
/// variable-length strings — ASCII NodeId/DisplayName ≈ 1 byte/char). Returns
/// the kept prefix and a Truncated flag OR-ed with the adapter's own
/// truncation signal, so the picker shows its "use manual entry" hint when the
/// level is clipped. Protocol-agnostic: every adapter's reply funnels through
/// here regardless of how it paginates upstream.
///
private static (IReadOnlyList Children, bool Truncated) CapBrowseChildren(
IReadOnlyList children, bool truncated)
{
var budget = 0;
var kept = new List(children.Count);
foreach (var node in children)
{
budget += 64 + (node.NodeId?.Length ?? 0) + (node.DisplayName?.Length ?? 0);
if (budget > BrowseResultByteBudget)
{
truncated = true;
break;
}
kept.Add(node);
}
return (kept, truncated);
}
// ── Test Bindings (one-shot live read of bound tags) ──
///
/// Handles a forwarded by the
/// . Short-circuits to a
/// failure
/// when the adapter is not currently Connected (Connecting / Reconnecting
/// states) so the dialog can render an inline banner without waiting for
/// the adapter to fail per-tag with a generic "client is not connected"
/// message. Otherwise calls _adapter.ReadBatchAsync and maps the
/// resulting per-tag map onto a list of
/// (preserving every requested tag — missing
/// adapter entries become failure outcomes).
///
/// Failure mapping mirrors :
///
/// - — adapter status is not .
/// - — batch cancelled ().
/// - — any other exception, message carried verbatim.
///
///
/// The reply is sent via PipeTo(sender) — same pattern as
/// and — so the
/// captured is safe to use from the continuation.
///
private void HandleReadTagValues(ReadTagValuesCommand command)
{
var sender = Sender;
if (_adapter.Status != ConnectionHealth.Connected)
{
_log.Debug("[{0}] Test-bindings read requested but adapter status is {1}", _connectionName, _adapter.Status);
sender.Tell(new ReadTagValuesResult(
Array.Empty(),
new ReadTagValuesFailure(
ReadTagValuesFailureKind.ConnectionNotConnected,
"Connection is not yet established.")));
return;
}
_log.Debug("[{0}] Test-bindings read of {1} tag(s)", _connectionName, command.TagPaths.Count);
var tagPaths = command.TagPaths.ToList();
_adapter.ReadBatchAsync(tagPaths).ContinueWith(t =>
{
if (t.IsCompletedSuccessfully)
{
var nowUtc = DateTimeOffset.UtcNow;
var outcomes = new List(tagPaths.Count);
foreach (var tagPath in tagPaths)
{
if (t.Result.TryGetValue(tagPath, out var result) && result.Success && result.Value is not null)
{
outcomes.Add(new TagReadOutcome(
tagPath,
Success: true,
Value: result.Value.Value,
Quality: result.Value.Quality.ToString(),
Timestamp: result.Value.Timestamp,
ErrorMessage: null));
}
else
{
var errMsg = result?.ErrorMessage
?? (t.Result.ContainsKey(tagPath)
? "Read returned no value."
: "Tag missing from adapter result.");
outcomes.Add(new TagReadOutcome(
tagPath,
Success: false,
Value: null,
Quality: "Bad",
Timestamp: nowUtc,
ErrorMessage: errMsg));
}
}
return new ReadTagValuesResult(outcomes, Failure: null);
}
var baseEx = t.Exception?.GetBaseException();
return baseEx switch
{
OperationCanceledException => new ReadTagValuesResult(
Array.Empty(),
new ReadTagValuesFailure(ReadTagValuesFailureKind.Timeout, "Read cancelled.")),
_ => new ReadTagValuesResult(
Array.Empty(),
new ReadTagValuesFailure(
ReadTagValuesFailureKind.ServerError,
baseEx?.Message ?? "Unknown read error.")),
};
}).PipeTo(sender);
}
// ── Tag Resolution Retry ──
private void HandleRetryTagResolution()
{
if (_unresolvedTags.Count == 0)
{
Timers.Cancel("tag-resolution-retry");
return;
}
var self = Self;
// Only dispatch retries for tags that do not already
// have an attempt in flight. A slow subscribe overlapping the next tick
// would otherwise produce duplicate concurrent subscribes for the same tag.
var toResolve = _unresolvedTags.Where(t => !_resolutionInFlight.Contains(t) && !_subscribesInFlight.Contains(t)).ToList();
if (toResolve.Count == 0)
{
_log.Debug("[{0}] Tag-resolution retry skipped — {1} attempt(s) still in flight",
_connectionName, _resolutionInFlight.Count);
return;
}
_log.Debug("[{0}] Retrying resolution for {1} unresolved tags (interval {2:F0}s)",
_connectionName, toResolve.Count, _tagResolutionInterval.TotalSeconds);
var generation = _adapterGeneration;
var adapter = _adapter;
var batchSize = Math.Max(1, _options.SubscribeBatchSize);
foreach (var tagPath in toResolve)
_resolutionInFlight.Add(tagPath);
// One probe ROUND = chunked batch subscribes over the whole not-in-flight
// unresolved set, reported back as a single completion. Rescheduling happens on
// that completion (never on a periodic tick), which is what makes the backoff
// possible AND preserves the anti-starvation property the old IsTimerActive gate
// defended: a fan-out of failures can never reset the clock.
Task.Run(async () =>
{
var rows = new List(toResolve.Count);
for (var offset = 0; offset < toResolve.Count; offset += batchSize)
{
var chunk = toResolve.GetRange(offset, Math.Min(batchSize, toResolve.Count - offset));
rows.AddRange(await SubscribeTagsAsync(adapter, chunk, (path, value) =>
{
self.Tell(new TagValueReceived(path, value, generation));
}));
}
// NOTE: `self` is the Self reference captured on the actor thread above — this runs on
// a background task, where Akka's Self property is invalid.
SeedNewlyResolvedTags(self, adapter, generation, rows);
return new BatchSubscribeCompleted(rows, generation, BatchSubscribeSource.ResolutionProbe);
}).PipeTo(self);
}
///
/// Reads the current value of every tag this probe round has just resolved and delivers it
/// through the normal generation-guarded path — the same seed
/// the initial subscribe () and the reconnect re-subscribe
/// () already perform, which the resolution probe was missing
/// (arch-review 2026-08-19).
///
///
/// A newly advised subscription only starts producing values on the source's next CHANGE.
/// OPC UA masks that with an initial data-change notification on subscription creation, but
/// an MxGateway advise can legitimately return VT_EMPTY, so a tag that resolved LATE
/// (device still booting, item created after deployment) and then never changes had no value
/// path at all — its attribute stayed Uncertain for the life of the connection, healed only
/// by an unrelated reconnect.
///
///
///
/// Unlike , the seeds are delivered directly rather than carried
/// back on the completion message: the probe only ever covers tags already present in
/// / , so the
/// registration-ordering hazard that forces the initial subscribe to defer does not exist
/// here. It runs as its own fire-and-forget task (mirroring the reconnect re-seed) so the
/// seed reads — bounded by SeedOverallTimeout, not by the subscribe — never delay the
/// that carries the round's resolution bookkeeping and
/// re-arms the retry timer. Values that arrive after a failover or a drop to Reconnecting are
/// discarded by the generation guard / state handler exactly like any other tag value.
///
///
///
/// The actor's own reference, captured on the actor thread by the caller — this method is
/// invoked from a background task, where Akka's Self property is invalid.
///
/// The adapter the probe round ran against.
/// The adapter generation the probe round was stamped with.
/// The probe round's per-tag subscribe results.
private void SeedNewlyResolvedTags(
IActorRef self, IDataConnection adapter, int generation, IReadOnlyList rows)
{
var resolved = new List(rows.Count);
foreach (var row in rows)
{
if (row is { Success: true, SubscriptionId: not null })
resolved.Add(row.TagPath);
}
if (resolved.Count == 0)
return;
_ = Task.Run(async () =>
{
try
{
var seeds = await SeedTagsAsync(adapter, resolved);
foreach (var seed in seeds)
self.Tell(new TagValueReceived(seed.TagPath, seed.Value, generation));
}
catch (Exception ex)
{
// Fire-and-forget: guarantee a trace rather than an unobserved task fault.
// SeedTagsAsync already swallows per-read errors, so reaching here is unexpected.
_log.Warning("[{0}] Tag-resolution re-seed task faulted: {1}", _connectionName, ex.Message);
}
});
}
///
/// Arms the single-shot tag-resolution retry timer at the current backoff interval
/// when unresolved tags remain, and cancels it when none do. Single-shot (rescheduled
/// on probe completion) rather than periodic — a periodic timer cannot back off, and
/// re-arming a periodic timer on every failure is what starved the retry before.
/// Gating on IsTimerActive keeps a burst of failed
/// subscribes from pushing the already-running deadline out.
///
private void ScheduleTagResolutionRetry()
{
if (_unresolvedTags.Count == 0)
{
Timers.Cancel("tag-resolution-retry");
return;
}
if (Timers.IsTimerActive("tag-resolution-retry"))
return;
Timers.StartSingleTimer("tag-resolution-retry", new RetryTagResolution(), _tagResolutionInterval);
}
/// Resets the tag-resolution backoff to its floor (a tag resolved, or the connection reconnected).
private void ResetTagResolutionBackoff() => _tagResolutionInterval = _options.TagResolutionRetryInterval;
///
/// Doubles the tag-resolution retry interval, capped at
/// . Called only after
/// a probe round in which NOTHING resolved, so a device that is merely slow to boot is
/// still picked up at the floor interval.
///
private void BackOffTagResolution() =>
_tagResolutionInterval = NextTagResolutionInterval(
_tagResolutionInterval, _options.TagResolutionRetryInterval, _options.TagResolutionRetryMaxInterval);
///
/// Pure backoff step: the interval after a fully-failed probe round — double the
/// current one, never above (itself never below
/// , so a misconfigured ceiling degrades to a fixed interval
/// rather than shrinking below the floor).
///
/// Interval that has just been used.
/// Configured floor interval.
/// Configured ceiling interval.
/// The next retry interval.
internal static TimeSpan NextTagResolutionInterval(TimeSpan current, TimeSpan floor, TimeSpan max)
{
var ceiling = max < floor ? floor : max;
if (current < floor)
current = floor;
var doubled = current + current;
return doubled > ceiling ? ceiling : doubled;
}
///
/// Applies one chunk of batch-subscribe results on the actor thread — the shared tail
/// of the reconnect re-subscribe and the tag-resolution probe. Both paths re-subscribe
/// tags that some instance already subscribes to (they come from
/// ), so they only ever move a tag between
/// unresolved and resolved — is unaffected by
/// construction, because neither path adds to or removes from the per-tag counted set.
///
private void HandleBatchSubscribeCompleted(BatchSubscribeCompleted msg)
{
// Results produced by a disposed adapter (post-failover) are dropped, mirroring
// the TagValueReceived generation guard.
if (msg.Generation != _adapterGeneration)
{
_log.Debug("[{0}] Dropping {1} stale batch-subscribe result(s) from adapter generation {2} (current {3})",
_connectionName, msg.Results.Count, msg.Generation, _adapterGeneration);
return;
}
var anyResolved = false;
var idsToRelease = new List();
foreach (var row in msg.Results)
{
_resolutionInFlight.Remove(row.TagPath);
if (row is { Success: true, SubscriptionId: not null })
{
var wasUnresolved = _unresolvedTags.Remove(row.TagPath);
// The tag lost its last subscriber while this batch was in flight — an
// instance unsubscribe (disable/undeploy/redeploy) raced the resolution
// probe or the reconnect re-subscribe. _instancesByTag, the per-tag counted
// set, is the authority: it is the inverse of _subscriptionsByInstance and,
// like it, is preserved across reconnect, so a missing key means nobody
// wants this tag any more. Applying the row anyway would store a handle no
// future unsubscribe can ever release (HandleUnsubscribe already ran for
// this tag) AND — because ResolvedTagCount is _subscriptionIds.Count while
// TotalSubscribedTagCount is _instancesByTag.Count — report a resolved tag
// that is not in the subscribed total at all, i.e. resolved > total. So
// discard every state mutation for this row and release the adapter handle
// the batch just created. This restores the gate the pre-merge per-tag
// handler had (it applied only when the tag was still in _unresolvedTags)
// and additionally releases the handle that path leaked.
//
// Deriving the counts is what makes this gate SUFFICIENT rather than merely
// necessary. It used to be one of several places that had to agree on scalar
// counters mutated elsewhere, and the counters could still drift when the
// unsubscribe landed inside a reconnect window (ReSubscribeAll clears the
// maps HandleUnsubscribe's decrement branches keyed off, so neither branch
// fired and the total leaked +1 per churn cycle). Now the gate has exactly
// one job — do not store an orphan handle — and both counts follow from the
// collections it protects.
if (!_instancesByTag.ContainsKey(row.TagPath))
{
_log.Debug(
"[{0}] Discarding batch-subscribe result for {1} — the tag was " +
"unsubscribed while the subscribe was in flight; releasing its handle.",
_connectionName, row.TagPath);
idsToRelease.Add(row.SubscriptionId);
continue;
}
if (_subscriptionIds.ContainsKey(row.TagPath))
{
// Another path already stored a handle for this tag while this one was
// in flight — release the redundant handle instead of leaking it
// (mirrors the duplicate-alarm-feed guard).
if (_subscriptionIds[row.TagPath] != row.SubscriptionId)
idsToRelease.Add(row.SubscriptionId);
}
else
{
_subscriptionIds[row.TagPath] = row.SubscriptionId;
anyResolved = true;
if (wasUnresolved)
_log.Info("[{0}] Tag resolved: {1}", _connectionName, row.TagPath);
}
}
else if (row.ConnectionLevelFailure)
{
// Connection-level fault: not a tag-resolution problem. The reconnect
// cycle re-issues everything from _subscriptionsByInstance.
_log.Warning("[{0}] Batch subscribe for {1} failed at connection level: {2}",
_connectionName, row.TagPath, row.Error);
}
else
{
_log.Debug("[{0}] Tag resolution still failing for {1}: {2}",
_connectionName, row.TagPath, row.Error);
// Same in-flight-unsubscribe race as the success branch above: the tag was
// already dropped from _unresolvedTags and from the per-tag counted set by
// HandleUnsubscribe, so re-adding it here would probe a tag nobody
// subscribes to, forever.
if (_instancesByTag.ContainsKey(row.TagPath))
_unresolvedTags.Add(row.TagPath);
}
}
_ = UnsubscribeIdsAsync(_adapter, idsToRelease);
_healthCollector.UpdateTagResolution(_connectionName, TotalSubscribedTagCount, ResolvedTagCount);
// Backoff bookkeeping belongs to the probe round only: a reconnect re-subscribe
// chunk is not a "retry round" and must not double the interval.
if (msg.Source == BatchSubscribeSource.ResolutionProbe)
{
if (anyResolved)
ResetTagResolutionBackoff();
else
BackOffTagResolution();
}
else if (anyResolved)
{
ResetTagResolutionBackoff();
}
ScheduleTagResolutionRetry();
}
// ── Bad Quality Push ──
private void PushBadQualityForAllTags()
{
var now = DateTimeOffset.UtcNow;
foreach (var (instanceName, tags) in _subscriptionsByInstance)
{
if (!_subscribers.TryGetValue(instanceName, out var subscriber))
continue;
subscriber.Tell(new ConnectionQualityChanged(_connectionName, QualityCode.Bad, now));
}
// All tags now bad quality
_tagsGoodQuality = 0;
_tagsUncertainQuality = 0;
_tagsBadQuality = _lastTagQuality.Count;
foreach (var key in _lastTagQuality.Keys.ToList())
_lastTagQuality[key] = QualityCode.Bad;
// SYNCHRONOUS — "immediate bad quality on disconnect" must never be deferred to
// the quality-flush coalescing timer.
FlushQualityCounters();
}
// ── Quality counter flush ──
///
/// Marks the quality counters dirty and arms the coalescing flush timer. Called only
/// on a genuine quality TRANSITION — an unchanged quality moves no counter, so there
/// is nothing to push.
///
private void MarkQualityDirty()
{
_qualityDirty = true;
if (!Timers.IsTimerActive("quality-flush"))
Timers.StartSingleTimer("quality-flush", new QualityFlushTick(), _options.QualityFlushInterval);
}
///
/// Coalescing-timer tick: pushes only when a transition is still pending. A tick that
/// races a synchronous flush (disconnect / unsubscribe) does nothing.
///
private void FlushQualityCountersIfDirty()
{
if (_qualityDirty)
FlushQualityCounters();
}
///
/// Pushes the current quality counters to the health collector and disarms the
/// coalescing timer. Used both by the timer tick and by the paths whose correctness
/// depends on an immediate push (disconnect, unsubscribe, reconnect reset).
///
private void FlushQualityCounters()
{
_qualityDirty = false;
Timers.Cancel("quality-flush");
_healthCollector.UpdateTagQuality(_connectionName, _tagsGoodQuality, _tagsBadQuality, _tagsUncertainQuality);
}
// ── Re-subscribe ──
private void ReSubscribeAll()
{
// Derive tag list from _subscriptionsByInstance (durable source of truth),
// not _subscriptionIds which gets cleared and is only repopulated on success.
var allTags = _subscriptionsByInstance.Values
.SelectMany(tags => tags)
.Distinct()
.ToList();
_log.Info("[{0}] Re-subscribing {1} tags after reconnect", _connectionName, allTags.Count);
var self = Self;
// NOTE: do NOT clear _instancesByTag here. It is the inverse of
// _subscriptionsByInstance, which this method deliberately preserves as the durable
// source of truth across reconnect — clearing the index would silently stop the
// post-reconnect value fan-out for every already-subscribed tag.
_subscriptionIds.Clear();
_unresolvedTags.Clear();
_resolutionInFlight.Clear();
// Symmetric with _resolutionInFlight — any pending
// initial-subscribe completions from the previous adapter generation will
// post SubscribeCompleted to the actor, but ReSubscribeAll has just emptied
// the in-flight tracking; the stale completion simply has nothing to
// remove (idempotent HashSet.Remove on a missing key).
_subscribesInFlight.Clear();
// NOTE: clearing _subscriptionIds above IS the resolved-count reset —
// ResolvedTagCount is derived from it. The retired _resolvedTags scalar had to be
// zeroed here by hand; its sibling _totalSubscribed deliberately was NOT, because
// the tags stay subscribed across a reconnect. That asymmetry is what left an
// unsubscribe arriving inside this window matching neither of _totalSubscribed's
// decrement branches (both keyed off maps this method just cleared), leaking a
// phantom tag into the reported total for the lifetime of the actor. Both counts
// now follow from the collections themselves: _instancesByTag is untouched here, so
// the total correctly rides through the reconnect and still drops on unsubscribe.
// Reset the quality tracking too. Otherwise tags
// resolved for the first time after reconnect (never in _lastTagQuality) only
// increment their bucket and the totals drift above the subscribed total. They are
// repopulated from fresh TagValueReceived messages once subscriptions activate.
_lastTagQuality.Clear();
_tagsGoodQuality = 0;
_tagsBadQuality = 0;
_tagsUncertainQuality = 0;
// SYNCHRONOUS reset push — never deferred to the coalescing timer.
FlushQualityCounters();
// A reconnect starts from the retry floor: the device just answered, so any
// backoff accumulated against the previous outage is stale.
ResetTagResolutionBackoff();
var generation = _adapterGeneration;
// Capture the adapter up front (S7), symmetric with reseedAdapter below, so every
// chunk runs against the adapter that started the re-subscribe even if a later
// failover swaps the _adapter field mid-flight.
var subscribeAdapter = _adapter;
var batchSize = Math.Max(1, _options.SubscribeBatchSize);
var chunkDelay = _options.SubscribeBatchDelay;
// Bounded, sequential chunks in ONE background task — replacing the previous
// "one fire-and-forget task per tag" storm (37,500 concurrent subscribes at the
// sizing target). Sequential because subscription creation on one session is
// serialized by the adapter's apply lock anyway, and sequencing bounds device load;
// the inter-chunk delay paces it further. Each chunk reports its per-tag results
// back to the actor as its own message, so progress is applied incrementally.
Task.Run(async () =>
{
for (var offset = 0; offset < allTags.Count; offset += batchSize)
{
if (offset > 0 && chunkDelay > TimeSpan.Zero)
await Task.Delay(chunkDelay);
var chunk = allTags.GetRange(offset, Math.Min(batchSize, allTags.Count - offset));
try
{
var rows = await SubscribeTagsAsync(subscribeAdapter, chunk, (path, value) =>
{
self.Tell(new TagValueReceived(path, value, generation));
});
self.Tell(new BatchSubscribeCompleted(rows, generation, BatchSubscribeSource.Resubscribe));
}
catch (Exception ex)
{
// SubscribeTagsAsync already converts faults into rows; reaching here is
// unexpected, so guarantee a trace rather than an unobserved task fault.
_log.Warning("[{0}] Reconnect re-subscribe chunk faulted: {1}", _connectionName, ex.Message);
}
}
});
// Re-advising alone does NOT restore a STATIC tag's value.
// PushBadQualityForAllTags flipped every tag Bad on disconnect, and a tag that fires
// no further OnDataChange would stay Bad/Uncertain across the reconnect even though
// the source reads Good — the same seed gap as the initial subscribe,
// which this path previously did not cover. Mirror HandleSubscribe and re-seed: read
// each re-advised tag's current value and deliver it through the normal
// generation-guarded path. Capture the adapter + generation up front so a later
// failover (which swaps _adapter and bumps _adapterGeneration) cannot misroute the
// reads or deliver a stale value — HandleTagValueReceived drops any value whose
// generation no longer matches. Delivery fans out via _subscriptionsByInstance
// (preserved across reconnect), so it does not depend on _subscriptionIds being
// repopulated for the new adapter generation. Seeds delivered after registration is
// already established, so the ordering hazard does not apply here.
var reseedAdapter = _adapter;
CancelSessionWork();
_sessionWorkCts = new CancellationTokenSource();
var reseedToken = _sessionWorkCts.Token;
Task.Run(async () =>
{
try
{
var seeds = await SeedTagsAsync(reseedAdapter, allTags, reseedToken);
foreach (var seed in seeds)
self.Tell(new TagValueReceived(seed.TagPath, seed.Value, generation));
}
catch (Exception ex)
{
// Fire-and-forget: guarantee a trace rather than an unobserved task fault.
// SeedTagsAsync already swallows per-read errors, so reaching here is unexpected.
_log.Warning("[{0}] Reconnect re-seed task faulted: {1}", _connectionName, ex.Message);
}
});
}
// ── Health Reporting ──
private void ReplyWithHealthReport()
{
var status = _adapter.Status;
var endpointLabel = _backupConfig == null
? "Primary (no backup)"
: _activeEndpoint.ToString();
Sender.Tell(new DataConnectionHealthReport(
_connectionName, status, TotalSubscribedTagCount, ResolvedTagCount,
endpointLabel, DateTimeOffset.UtcNow));
}
// ── Internal message handlers for piped async results ──
///
/// Adds to the reverse index for .
/// Call at every site that adds a tag to an instance's set.
///
private void IndexTag(string tag, string instance)
{
if (!_instancesByTag.TryGetValue(tag, out var insts))
{
insts = new HashSet();
_instancesByTag[tag] = insts;
}
insts.Add(instance);
}
///
/// Removes from the reverse index for ,
/// dropping the tag key entirely when its instance set empties. Call at every site that
/// removes a tag from an instance's set.
///
private void UnindexTag(string tag, string instance)
{
if (_instancesByTag.TryGetValue(tag, out var insts) && insts.Remove(instance) && insts.Count == 0)
_instancesByTag.Remove(tag);
}
private void HandleTagValueReceived(TagValueReceived msg)
{
// Drop values delivered by a disposed adapter. After a
// failover the old adapter's OPC UA SDK threads may still fire callbacks; those
// carry a stale generation and must not be forwarded to Instance Actors.
if (msg.AdapterGeneration != _adapterGeneration)
{
_log.Debug("[{0}] Dropping stale tag value for {1} from adapter generation {2} (current {3})",
_connectionName, msg.TagPath, msg.AdapterGeneration, _adapterGeneration);
return;
}
// A tag nobody subscribes to any more (a callback already queued, or raised before
// the device unadvise landed, after the last unsubscribe) is dropped WHOLE — including
// the quality bookkeeping below, which would otherwise re-insert the tag into
// _lastTagQuality and count a phantom bucket above the subscribed total until the
// next reconnect (review 2026-10-21 02/L2).
if (!_instancesByTag.TryGetValue(msg.TagPath, out var interestedInstances))
return;
// Fan out to exactly the instances subscribed to this tag via the reverse index —
// O(subscribers of this tag) instead of scanning every instance's tag set per update.
foreach (var instanceName in interestedInstances)
{
if (_subscribers.TryGetValue(instanceName, out var subscriber))
{
subscriber.Tell(new TagValueUpdate(
_connectionName, msg.TagPath, msg.Value.Value, msg.Value.Quality, msg.Value.Timestamp));
}
}
// Track quality transitions. A value whose quality is UNCHANGED moves no bucket —
// the old code still ran the decrement/increment pair and pushed the identical
// counters to the health collector on every single value. At 37,500 tags that push
// was the dominant cost of the value hot path, against a collector only read every
// 30s. Now: unchanged quality is a no-op; a genuine transition updates the buckets
// and arms the coalescing flush timer.
var hadPrevious = _lastTagQuality.TryGetValue(msg.TagPath, out var prevQuality);
if (hadPrevious && prevQuality == msg.Value.Quality)
return;
if (hadPrevious)
{
// Decrement old quality bucket
switch (prevQuality)
{
case QualityCode.Good: _tagsGoodQuality--; break;
case QualityCode.Bad: _tagsBadQuality--; break;
case QualityCode.Uncertain: _tagsUncertainQuality--; break;
}
}
// Increment new quality bucket
switch (msg.Value.Quality)
{
case QualityCode.Good: _tagsGoodQuality++; break;
case QualityCode.Bad: _tagsBadQuality++; break;
case QualityCode.Uncertain: _tagsUncertainQuality++; break;
}
_lastTagQuality[msg.TagPath] = msg.Value.Quality;
MarkQualityDirty();
}
// ── Native alarm subscriptions ──
private sealed class AlarmReplayState
{
/// The current alarm transitions keyed by alarm, as last published.
public Dictionary Current { get; set; } = new();
/// Transitions accumulated while a replay snapshot is being received.
public Dictionary Snapshot { get; } = new();
/// Live transitions that arrived while the snapshot was still in progress.
public Dictionary LiveDuringSnapshot { get; } = new();
/// True while a replay snapshot is being received.
public bool InSnapshot { get; set; }
/// True once the initial snapshot has completed and live updates may be published.
public bool Ready { get; set; }
/// The replay epoch the current state belongs to, if known.
public Guid? Epoch { get; set; }
}
private readonly Dictionary _alarmReplay = new();
private void CacheAlarmTransition(string source, NativeAlarmTransition transition)
{
if (!_alarmReplay.TryGetValue(source, out var state))
_alarmReplay[source] = state = new AlarmReplayState();
switch (transition.Kind)
{
case AlarmTransitionKind.SnapshotStart:
if (transition.SnapshotId is not null && transition.SnapshotId == state.Epoch) return;
state.Snapshot.Clear();
state.LiveDuringSnapshot.Clear();
state.InSnapshot = true;
state.Epoch = transition.SnapshotId;
break;
case AlarmTransitionKind.Snapshot:
if (transition.SnapshotId != state.Epoch || (transition.SnapshotId is not null && !state.InSnapshot)) return;
state.InSnapshot = true;
state.Snapshot[transition.SourceReference] = transition;
break;
case AlarmTransitionKind.SnapshotComplete:
if (transition.SnapshotId != state.Epoch || (transition.SnapshotId is not null && !state.InSnapshot)) return;
state.Current = SnapshotWithLiveChanges(state);
state.Snapshot.Clear();
state.LiveDuringSnapshot.Clear();
state.InSnapshot = false;
state.Ready = true;
break;
default:
if (state.LiveDuringSnapshot.TryGetValue(transition.SourceReference, out var recent)
&& transition.TransitionTime < recent.TransitionTime) return;
if (state.Current.TryGetValue(transition.SourceReference, out var prior)
&& transition.TransitionTime < prior.TransitionTime) return;
if (state.InSnapshot) state.LiveDuringSnapshot[transition.SourceReference] = transition;
if (!transition.Condition.Active && transition.Condition.Acknowledged)
state.Current.Remove(transition.SourceReference);
else
state.Current[transition.SourceReference] = transition;
break;
}
}
private static Dictionary SnapshotWithLiveChanges(AlarmReplayState state)
{
var result = new Dictionary(state.Snapshot);
foreach (var (key, live) in state.LiveDuringSnapshot)
{
if (result.TryGetValue(key, out var snapshot) && snapshot.TransitionTime > live.TransitionTime) continue;
if (!live.Condition.Active && live.Condition.Acknowledged) result.Remove(key);
else result[key] = live;
}
return result;
}
private void ReplayAlarmsTo(string source, IActorRef subscriber)
{
if (!_alarmReplay.TryGetValue(source, out var state) || (!state.Ready && !state.InSnapshot)) return;
var epoch = state.InSnapshot ? state.Epoch : Guid.NewGuid();
var boundary = new NativeAlarmTransition("", source, "", AlarmTransitionKind.SnapshotStart,
new AlarmConditionState(false, true, null, AlarmShelveState.Unshelved, false, 0),
"", "", "", "", "", null, DateTimeOffset.UtcNow, "", "", SnapshotSourceReference: source, SnapshotId: epoch);
// Legacy Mx snapshots have no start or epoch. Mid-frame subscribers get
// the prefix already received and then share the remaining live frame.
if (epoch is not null)
subscriber.Tell(new NativeAlarmTransitionUpdate(_connectionName, boundary));
foreach (var transition in (state.InSnapshot ? SnapshotWithLiveChanges(state) : state.Current).Values)
subscriber.Tell(new NativeAlarmTransitionUpdate(_connectionName, transition with
{ Kind = AlarmTransitionKind.Snapshot, SnapshotSourceReference = source, SnapshotId = epoch }));
if (!state.InSnapshot)
subscriber.Tell(new NativeAlarmTransitionUpdate(_connectionName, boundary with { Kind = AlarmTransitionKind.SnapshotComplete }));
}
private void HandleSubscribeAlarms(SubscribeAlarmsRequest request)
{
var subscriber = Sender;
var now = DateTimeOffset.UtcNow;
if (_adapter is not IAlarmSubscribableConnection alarmable)
{
subscriber.Tell(new SubscribeAlarmsResponse(
request.CorrelationId, request.InstanceUniqueName, false,
$"Connection '{_connectionName}' is not alarm-capable.", now));
return;
}
// Register the subscriber for routing (idempotent) before issuing the
// adapter subscribe so a transition that arrives mid-subscribe is routed.
if (!_alarmSourceSubscribers.TryGetValue(request.SourceReference, out var subs))
{
subs = new HashSet();
_alarmSourceSubscribers[request.SourceReference] = subs;
IndexAlarmSource(request.SourceReference);
}
var newlyAdded = subs.Add(subscriber);
Context.Watch(subscriber);
// S10: the adapter feed carries a single shared condition filter per source.
// A second subscriber with a different filter silently overwrites the first
// (last-writer-wins). Surface it so a co-subscriber mismatch is diagnosable —
// co-subscribers to one source are expected to agree on the filter.
if (_alarmSourceFilter.TryGetValue(request.SourceReference, out var existingFilter)
&& !string.Equals(existingFilter, request.ConditionFilter, StringComparison.Ordinal))
{
_log.Warning(
"[{0}] Alarm condition filter for source {1} overwritten: '{2}' -> '{3}' (last subscriber wins; co-subscribers must agree)",
_connectionName, request.SourceReference, existingFilter, request.ConditionFilter);
}
_alarmSourceFilter[request.SourceReference] = request.ConditionFilter;
// Parse the type-name filter once; this is the authoritative client-side
// gate consulted on every routed transition.
_alarmSourceFilterPredicate[request.SourceReference] = AlarmConditionFilter.Parse(request.ConditionFilter);
// If the adapter feed for this source is already (being) established, the
// existing subscription serves the new subscriber too.
if (_alarmSubscriptionIds.ContainsKey(request.SourceReference) ||
_alarmSubscribesInFlight.Contains(request.SourceReference))
{
if (newlyAdded) ReplayAlarmsTo(request.SourceReference, subscriber);
subscriber.Tell(new SubscribeAlarmsResponse(
request.CorrelationId, request.InstanceUniqueName, true, null, now));
return;
}
_alarmSubscribesInFlight.Add(request.SourceReference);
var self = Self;
var generation = _adapterGeneration;
var sourceRef = request.SourceReference;
var filter = request.ConditionFilter;
var corr = request.CorrelationId;
var inst = request.InstanceUniqueName;
alarmable.SubscribeAlarmsAsync(sourceRef, filter,
t => self.Tell(new AlarmTransitionReceived(t, generation)))
.ContinueWith(task => task.IsCompletedSuccessfully
? new AlarmSubscribeCompleted(
sourceRef, true, task.Result, null, subscriber, corr, inst, generation, alarmable) as object
: new AlarmSubscribeCompleted(sourceRef, false, null,
task.Exception?.GetBaseException().Message ?? "Unknown error",
subscriber, corr, inst, generation, alarmable))
.PipeTo(self);
}
private void HandleAlarmSubscribeCompleted(AlarmSubscribeCompleted msg)
{
// The alarm twin of HandleSubscribeCompleted's generation guard: a completion from an
// adapter that has since been swapped out is dropped whole. Applying it would store a
// dead feed's id — or, when the new adapter's feed is already stored, take the
// duplicate branch and release the LIVE feed, leaving the source mirrored but silent.
if (msg.AdapterGeneration != _adapterGeneration)
{
_log.Debug(
"[{0}] Dropping stale alarm subscribe completion for source {1} from adapter " +
"generation {2} (current {3})",
_connectionName, msg.SourceReference, msg.AdapterGeneration, _adapterGeneration);
// Release the orphaned feed against the adapter that created it, never the current
// one (see ReleaseStaleAgainstOwnAdapter).
if (msg is { Success: true, SubscriptionId: not null })
ReleaseStaleAgainstOwnAdapter(
msg.Adapter,
$"the alarm feed for {msg.SourceReference} from adapter generation {msg.AdapterGeneration}",
() => msg.Adapter.UnsubscribeAlarmsAsync(msg.SubscriptionId));
// _alarmSubscribesInFlight is left untouched — ReSubscribeAllAlarms clears it
// wholesale and may already have re-registered this source against the new adapter.
// The subscriber state (_alarmSourceSubscribers) survives the swap, so the feed is
// re-established by ReSubscribeAllAlarms; the failure reply just lets the
// NativeAlarmActor retry rather than wait out its response timeout.
msg.ReplyTo?.Tell(new SubscribeAlarmsResponse(
msg.CorrelationId ?? string.Empty, msg.InstanceUniqueName ?? string.Empty,
false, "adapter replaced while the alarm subscribe was in flight — will retry",
DateTimeOffset.UtcNow));
return;
}
_alarmSubscribesInFlight.Remove(msg.SourceReference);
// The last (or only) subscriber may have been
// unsubscribed while this alarm subscribe was in flight. HandleUnsubscribeAlarms
// emptied/removed _alarmSourceSubscribers for the source but could not tear down
// the adapter feed because the subscription id was not stored yet. Mirror the
// tag-path guard: if no subscriber remains, release the just-created
// adapter feed instead of storing an orphaned subscription id that would stream
// transitions to nobody for the lifetime of the adapter.
if (!_alarmSourceSubscribers.ContainsKey(msg.SourceReference))
{
if (msg.Success && msg.SubscriptionId != null)
{
_log.Warning(
"[{0}] AlarmSubscribeCompleted arrived for source {1} but the last " +
"subscriber unsubscribed while the subscribe was in flight; releasing " +
"the orphaned adapter alarm feed.",
_connectionName, msg.SourceReference);
// Against the adapter that created the feed (identical to _adapter here — the
// generation guard above already proved the completion is current).
_ = msg.Adapter.UnsubscribeAlarmsAsync(msg.SubscriptionId);
}
// No live requester remains to receive a response.
return;
}
if (msg.Success && msg.SubscriptionId != null)
{
// A concurrent unsubscribe clears the in-flight
// marker, so a fresh subscribe for the same source can issue a
// SECOND adapter feed before this completion fires — yielding two completions
// for one source. Mirror the tag-path re-check (see HandleTagSubscribeCompleted,
// the `_subscriptionIds.ContainsKey` guard): if a feed is already stored, THIS
// completion is the redundant one — release its feed rather than overwriting
// the stored id and leaking the already-tracked subscription.
if (_alarmSubscriptionIds.ContainsKey(msg.SourceReference))
{
_log.Warning(
"[{0}] Duplicate alarm feed for source {1}; releasing the redundant " +
"subscription instead of overwriting the stored one.",
_connectionName, msg.SourceReference);
_ = msg.Adapter.UnsubscribeAlarmsAsync(msg.SubscriptionId);
}
else
{
_alarmSubscriptionIds[msg.SourceReference] = msg.SubscriptionId;
_log.Info("[{0}] Alarm feed subscribed for source {1}", _connectionName, msg.SourceReference);
}
}
else if (!msg.Success)
{
_log.Warning("[{0}] Alarm subscribe failed for source {1}: {2}",
_connectionName, msg.SourceReference, msg.Error);
}
// ReplyTo is null for reconnect re-subscribes (no original requester to answer).
msg.ReplyTo?.Tell(new SubscribeAlarmsResponse(
msg.CorrelationId ?? string.Empty, msg.InstanceUniqueName ?? string.Empty,
msg.Success, msg.Error, DateTimeOffset.UtcNow));
}
private void HandleAlarmTransitionReceived(AlarmTransitionReceived msg)
{
// Drop transitions from a disposed adapter after failover.
if (msg.AdapterGeneration != _adapterGeneration)
return;
var transition = msg.Transition;
var notified = new HashSet();
// OPC UA refresh boundaries belong to one monitored source, whereas Mx
// uses one connection-wide alarm stream. Boundaries bypass condition filters
// but must retain their scope: broadcasting a second source's end marker
// would clear the first source's already-committed snapshot.
if (transition.Kind is AlarmTransitionKind.SnapshotStart or AlarmTransitionKind.SnapshotComplete)
{
// Explicit scope distinguishes the empty OPC UA Server binding from the
// connection-wide Mx stream. Legacy empty-scope sentinels stay connection-wide.
var scope = transition.SnapshotSourceReference
?? (string.IsNullOrEmpty(transition.SourceObjectReference) ? null : transition.SourceObjectReference);
var targets = scope is null
? _alarmSourceSubscribers
: _alarmSourceSubscribers.Where(kv => kv.Key == scope);
foreach (var (source, subscribers) in targets)
{
CacheAlarmTransition(source, transition);
foreach (var sub in subscribers)
if (notified.Add(sub))
sub.Tell(new NativeAlarmTransitionUpdate(_connectionName, transition));
}
return;
}
foreach (var sourceRef in CandidateAlarmSources(transition))
{
if (!_alarmSourceSubscribers.TryGetValue(sourceRef, out var subs))
continue;
// A subscriber bound to source S receives a transition whose source
// object (or full reference) falls under S.
var match = transition.SourceObjectReference.StartsWith(sourceRef, StringComparison.Ordinal)
|| transition.SourceReference.StartsWith(sourceRef, StringComparison.Ordinal);
if (!match)
continue;
// Authoritative client-side condition-type gate. Applied
// per matched source because two sources may share a prefix yet carry
// different filters. Empty filter = allow all (historical behaviour);
// framing sentinels (SnapshotComplete) are never dropped.
if (_alarmSourceFilterPredicate.TryGetValue(sourceRef, out var predicate) &&
!predicate.IsAllowed(transition))
continue;
CacheAlarmTransition(sourceRef, transition);
foreach (var sub in subs)
{
if (notified.Add(sub))
sub.Tell(new NativeAlarmTransitionUpdate(_connectionName, transition));
}
}
}
///
/// First path segment of a source reference (everything before the first .), or
/// the whole reference when it carries no separator.
///
private static string FirstSegment(string reference)
{
var separator = reference.IndexOf('.', StringComparison.Ordinal);
return separator < 0 ? reference : reference[..separator];
}
///
/// Adds a source reference to the routing index. A reference containing a separator
/// is bucketed by its first segment; one WITHOUT a separator is a prefix shorter than
/// a full segment (it can match transitions whose first segment merely starts with it)
/// and goes to the small residue list that every transition is checked against.
///
private void IndexAlarmSource(string sourceReference)
{
if (sourceReference.IndexOf('.', StringComparison.Ordinal) < 0)
{
_alarmSourcesWithoutSeparator.Add(sourceReference);
return;
}
var segment = FirstSegment(sourceReference);
if (!_alarmSourcesByFirstSegment.TryGetValue(segment, out var bucket))
{
bucket = new HashSet(StringComparer.Ordinal);
_alarmSourcesByFirstSegment[segment] = bucket;
}
bucket.Add(sourceReference);
}
/// Removes a source reference from the routing index, dropping an emptied bucket.
private void UnindexAlarmSource(string sourceReference)
{
if (sourceReference.IndexOf('.', StringComparison.Ordinal) < 0)
{
_alarmSourcesWithoutSeparator.Remove(sourceReference);
return;
}
var segment = FirstSegment(sourceReference);
if (_alarmSourcesByFirstSegment.TryGetValue(segment, out var bucket)
&& bucket.Remove(sourceReference) && bucket.Count == 0)
{
_alarmSourcesByFirstSegment.Remove(segment);
}
}
///
/// Source references that could possibly match this transition: the buckets for the
/// first segment of its source-object and full references, plus the separator-less
/// residue. Sound because a source reference containing a separator can only be a
/// prefix of a reference sharing its ENTIRE first segment — everything before the
/// first . must match character for character. The caller still applies the
/// full StartsWith test and the condition-type gate, so routing is identical to the
/// previous linear scan.
///
private IEnumerable CandidateAlarmSources(NativeAlarmTransition transition)
{
var objectSegment = FirstSegment(transition.SourceObjectReference);
if (_alarmSourcesByFirstSegment.TryGetValue(objectSegment, out var objectBucket))
{
foreach (var sourceRef in objectBucket)
yield return sourceRef;
}
var referenceSegment = FirstSegment(transition.SourceReference);
if (!string.Equals(referenceSegment, objectSegment, StringComparison.Ordinal)
&& _alarmSourcesByFirstSegment.TryGetValue(referenceSegment, out var referenceBucket))
{
foreach (var sourceRef in referenceBucket)
yield return sourceRef;
}
foreach (var sourceRef in _alarmSourcesWithoutSeparator)
yield return sourceRef;
}
private void HandleUnsubscribeAlarms(UnsubscribeAlarmsRequest request)
{
_pendingAlarmSubscriptions.Remove((Sender, request.SourceReference));
RemoveAlarmSubscriber(request.SourceReference, Sender);
UnwatchIfUnused(Sender);
}
private void RemoveAlarmSubscriber(string sourceReference, IActorRef owner)
{
if (!_alarmSourceSubscribers.TryGetValue(sourceReference, out var subs) || !subs.Remove(owner))
return;
UnwatchIfUnused(owner);
if (subs.Count > 0)
return;
// No subscribers remain for this source — tear down the adapter feed.
_alarmSourceSubscribers.Remove(sourceReference);
_alarmReplay.Remove(sourceReference);
UnindexAlarmSource(sourceReference);
_alarmSourceFilter.Remove(sourceReference);
_alarmSourceFilterPredicate.Remove(sourceReference);
// Clear the in-flight marker so that if an adapter
// subscribe is still in flight for this source, the late AlarmSubscribeCompleted
// is recognized as orphaned (its guard checks _alarmSourceSubscribers, now empty)
// and the just-created feed is released rather than stored and leaked.
_alarmSubscribesInFlight.Remove(sourceReference);
if (_alarmSubscriptionIds.Remove(sourceReference, out var subId) &&
_adapter is IAlarmSubscribableConnection alarmable)
{
_ = alarmable.UnsubscribeAlarmsAsync(subId);
}
}
/// Re-establishes all native alarm feeds after a reconnect; the source replays a snapshot.
private void ReSubscribeAllAlarms()
{
if (_adapter is not IAlarmSubscribableConnection alarmable || _alarmSourceSubscribers.Count == 0)
return;
_alarmSubscriptionIds.Clear();
_alarmReplay.Clear();
_alarmSubscribesInFlight.Clear();
var self = Self;
var generation = _adapterGeneration;
foreach (var sourceRef in _alarmSourceSubscribers.Keys.ToList())
{
var sr = sourceRef;
var filter = _alarmSourceFilter.GetValueOrDefault(sourceRef);
_alarmSubscribesInFlight.Add(sr);
alarmable.SubscribeAlarmsAsync(sr, filter,
t => self.Tell(new AlarmTransitionReceived(t, generation)))
.ContinueWith(task => task.IsCompletedSuccessfully
? new AlarmSubscribeCompleted(
sr, true, task.Result, null, null, null, null, generation, alarmable) as object
: new AlarmSubscribeCompleted(sr, false, null,
task.Exception?.GetBaseException().Message ?? "Unknown error",
null, null, null, generation, alarmable))
.PipeTo(self);
}
}
/// Notifies alarm subscribers that the source feed is unavailable (connection lost).
private void PushAlarmSourceUnavailable()
{
var now = DateTimeOffset.UtcNow;
foreach (var (sourceRef, subs) in _alarmSourceSubscribers)
{
foreach (var sub in subs)
sub.Tell(new NativeAlarmSourceUnavailable(_connectionName, sourceRef, now));
}
}
// ── Internal messages ──
internal record AttemptConnect;
///
/// Outcome of one . Carries the adapter generation it ran
/// against so a result produced by an adapter that has since been swapped out is dropped
/// rather than applied to its replacement (see ).
///
internal record ConnectResult(bool Success, string? Error, int AdapterGeneration);
///
/// The adapter's Disconnected event marshalled onto the actor thread, stamped with the
/// the raising adapter was hooked at so a notification from a
/// swapped-out adapter can be dropped ().
///
internal record AdapterDisconnected(int AdapterGeneration);
internal record TagValueReceived(string TagPath, TagValue Value, int AdapterGeneration);
internal record RetryTagResolution;
/// Coalescing-timer tick for the tag-quality counter push.
internal record QualityFlushTick;
/// Which path produced a .
internal enum BatchSubscribeSource
{
/// One chunk of the post-reconnect transparent re-subscribe.
Resubscribe,
/// A whole tag-resolution retry round (all chunks), which drives the backoff.
ResolutionProbe
}
///
/// Per-tag results of a batch subscribe issued off the actor thread. Carries the
/// adapter generation it ran against so post-failover results are dropped.
///
internal record BatchSubscribeCompleted(
IReadOnlyList Results, int Generation, BatchSubscribeSource Source);
internal record SubscribeTagResult(
string TagPath, bool AlreadySubscribed, bool Success, string? SubscriptionId, string? Error,
bool ConnectionLevelFailure = false);
///
/// Per-tag results of the initial subscribe issued off the actor thread by
/// . Carries the adapter generation it ran against — and the
/// adapter INSTANCE that produced the handles, because a stale completion's handles must be
/// released against their own adapter (see ).
///
internal record SubscribeCompleted(
SubscribeTagsRequest Request, IActorRef ReplyTo, IReadOnlyList Results,
IReadOnlyList SeedValues, int AdapterGeneration, IDataConnection Adapter);
/// An initial-read value captured during subscribe, delivered after the
/// instance's tags are registered for fan-out.
internal record SeededValue(string TagPath, TagValue Value);
internal record AlarmTransitionReceived(NativeAlarmTransition Transition, int AdapterGeneration);
///
/// Result of an alarm-feed subscribe issued off the actor thread. Carries the adapter
/// generation and the adapter instance for the same reason
/// does.
///
internal record AlarmSubscribeCompleted(
string SourceReference, bool Success, string? SubscriptionId, string? Error,
IActorRef? ReplyTo, string? CorrelationId, string? InstanceUniqueName,
int AdapterGeneration, IAlarmSubscribableConnection Adapter);
public record GetHealthReport;
}