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.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/Stash pattern for lifecycle state machine. /// /// States: /// - Connecting: stash subscribe/write requests; attempts connection /// - Connected: unstash and process all requests /// - Reconnecting: push bad quality for all subscribed tags, stash new requests, /// 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, IWithStash, 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; private readonly string _protocolType; private readonly ISiteEventLogger? _siteEventLogger; /// Stash for holding messages while the connection is not in the Connected state. public IStash Stash { get; set; } = null!; /// 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; private readonly IDictionary _primaryConfig; private readonly IDictionary? _backupConfig; private readonly 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!; /// /// 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 _adapter.Disconnected += OnAdapterDisconnected; BecomeConnecting(); } private void OnAdapterDisconnected() { // Marshal the event onto the actor's message loop using captured _self reference. // This runs on a background thread (gRPC stream reader), so Self would throw. _self.Tell(new AdapterDisconnected()); } /// protected override void PostStop() { _log.Info("DataConnectionActor [{0}] stopping — disposing adapter", _connectionName); _adapter.Disconnected -= OnAdapterDisconnected; // 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"); Become(Connecting); Self.Tell(new AttemptConnect()); } private void Connecting(object message) { switch (message) { case AttemptConnect: HandleAttemptConnect(); break; case ConnectResult result: HandleConnectResult(result); break; case SubscribeTagsRequest: case WriteTagRequest: case WriteTagBatchRequest: case UnsubscribeTagsRequest: case SubscribeAlarmsRequest: case UnsubscribeAlarmsRequest: Stash.Stash(); 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 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); Become(Connected); Stash.UnstashAll(); } 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: HandleDisconnect(); break; case RetryTagResolution: HandleRetryTagResolution(); break; case BrowseNodeCommand browse: HandleBrowse(browse); break; case SearchAddressSpaceCommand search: HandleSearch(search); break; case ReadTagValuesCommand read: HandleReadTagValues(read); break; case GetHealthReport: ReplyWithHealthReport(); break; default: Unhandled(message); break; } } // ── Reconnecting State ── private void BecomeReconnecting() { _log.Warning("[{0}] Entering Reconnecting state", _connectionName); _healthCollector.UpdateConnectionHealth(_connectionName, ConnectionHealth.Disconnected); _healthCollector.UpdateConnectionEndpoint(_connectionName, "Disconnected"); // 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; var newConfig = _activeEndpoint == ActiveEndpoint.Primary ? _primaryConfig : _backupConfig; // Dispose old adapter _adapter.Disconnected -= OnAdapterDisconnected; _ = _adapter.DisposeAsync().AsTask(); // Create new adapter for the target endpoint _adapter = _factory.Create(_protocolType, newConfig); _connectionDetails = newConfig; _adapter.Disconnected += OnAdapterDisconnected; // New adapter — bump the generation so callbacks // from the disposed adapter are recognised as stale and dropped. _adapterGeneration++; _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", "Warning", null, _connectionName, $"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 Timers.StartSingleTimer("reconnect", new AttemptConnect(), _options.ReconnectInterval); } private void Reconnecting(object message) { switch (message) { case AttemptConnect: HandleAttemptConnect(); break; case ConnectResult result: HandleReconnectResult(result); break; case SubscribeTagsRequest: case WriteTagRequest: case WriteTagBatchRequest: case SubscribeAlarmsRequest: Stash.Stash(); 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 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 GetHealthReport: ReplyWithHealthReport(); break; default: Unhandled(message); break; } } // ── Connection Management ── private void HandleAttemptConnect() { _log.Debug("[{0}] Attempting connection...", _connectionName); var self = Self; _adapter.ConnectAsync(_connectionDetails).ContinueWith(t => { if (t.IsCompletedSuccessfully) return new ConnectResult(true, null); return new ConnectResult(false, t.Exception?.GetBaseException().Message); }).PipeTo(self); } private void HandleConnectResult(ConnectResult result) { if (result.Success) { _log.Info("[{0}] Connection established", _connectionName); 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 (result.Success) { _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); } // 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; var newConfig = _activeEndpoint == ActiveEndpoint.Primary ? _primaryConfig : _backupConfig; // Dispose old adapter (fire-and-forget — don't await in actor context) _adapter.Disconnected -= OnAdapterDisconnected; _ = _adapter.DisposeAsync().AsTask(); // Create new adapter for the target endpoint _adapter = _factory.Create(_protocolType, newConfig); _connectionDetails = newConfig; // Wire disconnect handler on new adapter _adapter.Disconnected += OnAdapterDisconnected; // New adapter — bump the generation so callbacks // from the disposed adapter are recognised as stale and dropped. _adapterGeneration++; _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(); } // ── Subscription Management ── private void HandleSubscribe(SubscribeTagsRequest request) { _log.Debug("[{0}] Subscribing {1} tags for instance {2}", _connectionName, request.TagPaths.Count, request.InstanceUniqueName); _subscribers[request.InstanceUniqueName] = Sender; if (!_subscriptionsByInstance.ContainsKey(request.InstanceUniqueName)) _subscriptionsByInstance[request.InstanceUniqueName] = new HashSet(); 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)) { 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); }).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))); } /// /// 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) { 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. using var overall = new CancellationTokenSource(_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 (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; if (!_subscriptionsByInstance.TryGetValue(instanceName, out var instanceTags)) { // The instance was unsubscribed while the // subscribe I/O was in flight. Re-creating the per-instance entry and // applying the handle mutations here would permanently leak state // — _subscriptionsByInstance[instanceName] resurrected with no // subscriber to receive callbacks, _instancesByTag re-indexed against // an instance no future HandleUnsubscribe will ever visit, and the // health counts derived from both reporting tags nobody subscribes to // for the rest of the adapter lifetime (also re-issued by ReSubscribeAll // on every reconnect). Instead: drop all state mutations for this stale // message and release the adapter-level monitored items we just // created so the device doesn't keep streaming change notifications // for a tag nobody is subscribed to. _log.Warning( "[{0}] SubscribeCompleted arrived for instance {1} but the instance " + "was unsubscribed while the subscribe was in flight; releasing " + "{2} adapter handle(s) and discarding state mutations.", _connectionName, instanceName, msg.Results.Count(r => r.Success && !r.AlreadySubscribed)); var orphanedIds = new List(); foreach (var result in msg.Results) { // Clear in-flight markers we placed in HandleSubscribe. if (!result.AlreadySubscribed) _subscribesInFlight.Remove(result.TagPath); // Fire-and-forget release of any subscription id this request // genuinely created. AlreadySubscribed=true means another caller // owns the adapter handle and unsubscribing it would break them. if (result is { Success: true, AlreadySubscribed: false, SubscriptionId: not null }) { orphanedIds.Add(result.SubscriptionId); } } _ = UnsubscribeIdsAsync(_adapter, orphanedIds); // The original sender is already gone (unsubscribed). Telling a dead // ref produces a dead letter, which is the harmless and observable // outcome — but skipping the reply altogether keeps dead-letter noise // out of the log when this race fires in the normal disable/redeploy // path. The unsubscribe message did NOT request a response of its own. return false; } // 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 }) 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 — // DataConnectionLayer-020). 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 (DataConnectionLayer-020). _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. _ = UnsubscribeIdsAsync(_adapter, redundantIds); // 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; } private void HandleUnsubscribe(UnsubscribeTagsRequest request) { _log.Debug("[{0}] Unsubscribing all tags for instance {1}", _connectionName, request.InstanceUniqueName); if (!_subscriptionsByInstance.TryGetValue(request.InstanceUniqueName, 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) { // 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). UnindexTag(tagPath, request.InstanceUniqueName); 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); } _subscriptionsByInstance.Remove(request.InstanceUniqueName); _subscribers.Remove(request.InstanceUniqueName); // 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; it does /// NOT call the adapter's poll-based WriteBatchAndWaitAsync. /// 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)).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)); })); } return new BatchSubscribeCompleted(rows, generation, BatchSubscribeSource.ResolutionProbe); }).PipeTo(self); } /// /// 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 /// (DataConnectionLayer-022). 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). 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; Task.Run(async () => { try { var seeds = await SeedTagsAsync(reseedAdapter, allTags); 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; } // 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. if (_instancesByTag.TryGetValue(msg.TagPath, out var interestedInstances)) { 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 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); } subs.Add(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)) { 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) as object : new AlarmSubscribeCompleted(sourceRef, false, null, task.Exception?.GetBaseException().Message ?? "Unknown error", subscriber, corr, inst)) .PipeTo(self); } private void HandleAlarmSubscribeCompleted(AlarmSubscribeCompleted msg) { _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 && _adapter is IAlarmSubscribableConnection alarmable) { _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); _ = alarmable.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)) { if (_adapter is IAlarmSubscribableConnection alarmable) { _log.Warning( "[{0}] Duplicate alarm feed for source {1}; releasing the redundant " + "subscription instead of overwriting the stored one.", _connectionName, msg.SourceReference); _ = alarmable.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(); // A SnapshotComplete is a connection-wide framing sentinel, not a real // condition: the mapper emits it with an empty SourceReference / // SourceObjectReference. It must reach EVERY alarm subscriber so each // NativeAlarmActor can atomically swap in the snapshot it just buffered. // The per-source prefix match below would drop it ("".StartsWith(".") // is false), which would strand statically-active conditions that are only // delivered in the snapshot (no later live transition) — they would buffer // forever and never surface. Broadcast the sentinel to all subscribers, // bypassing the source match and the condition-type filter (the sentinel // carries no condition; the buffered entries were already filtered). if (transition.Kind == AlarmTransitionKind.SnapshotComplete) { foreach (var subs in _alarmSourceSubscribers.Values) foreach (var sub in subs) 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; 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) { if (!_alarmSourceSubscribers.TryGetValue(request.SourceReference, out var subs)) return; subs.Remove(Sender); if (subs.Count > 0) return; // No subscribers remain for this source — tear down the adapter feed. _alarmSourceSubscribers.Remove(request.SourceReference); UnindexAlarmSource(request.SourceReference); _alarmSourceFilter.Remove(request.SourceReference); _alarmSourceFilterPredicate.Remove(request.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(request.SourceReference); if (_alarmSubscriptionIds.Remove(request.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(); _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) as object : new AlarmSubscribeCompleted(sr, false, null, task.Exception?.GetBaseException().Message ?? "Unknown error", null, null, null)) .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; internal record ConnectResult(bool Success, string? Error); internal record AdapterDisconnected; 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); internal record SubscribeCompleted( SubscribeTagsRequest Request, IActorRef ReplyTo, IReadOnlyList Results, IReadOnlyList SeedValues); /// 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); internal record AlarmSubscribeCompleted( string SourceReference, bool Success, string? SubscriptionId, string? Error, IActorRef? ReplyTo, string? CorrelationId, string? InstanceUniqueName); public record GetHealthReport; }