using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; using Opc.Ua; using Opc.Ua.Client; using Opc.Ua.Configuration; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.Secrets.Abstractions; namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient; /// /// OPC UA Client (gateway) driver. Opens a against a remote OPC UA /// server and re-exposes its address space through the local OtOpcUa server. PR 66 ships /// the scaffold: only (connect / close / health). Browse, read, /// write, subscribe, and probe land in PRs 67-69. /// /// /// /// Builds its own rather than reusing /// Client.Shared — Client.Shared is oriented at the interactive CLI; this /// driver is an always-on service component with different session-lifetime needs /// (keep-alive monitor, session transfer on reconnect, multi-year uptime). /// /// /// Session lifetime: a single per driver instance. /// Subscriptions multiplex onto that session. A lost session is reported (health Degraded, /// subscribed values BadNoCommunication) and recovered by the host's comms-loss /// watchdog re-initialising the driver and re-subscribing; a session whose keep-alive comes /// back on its own recovers in place. /// /// public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWritable, ISubscribable, IHostConnectivityProbe, IAlarmSource, ISkippedTagReporter, IDriverConnectionLossSource, IDisposable, IAsyncDisposable { /// Authored tags whose TagConfig carried no usable nodeId, so there is no upstream node /// to dial. This driver's skip path was the quietest of the twelve: NamespaceMap /// drops such an entry with a bare continue and not even a log line. private readonly SkippedTagLedger _skippedTags = new(); /// public SkippedTagReport? GetSkippedTags() => _skippedTags.Snapshot(); /// /// Recomputes the skip report from the freshly-built map. Derived by re-asking the map to resolve /// each authored RawPath rather than by instrumenting NamespaceMap, so what is reported /// skipped is exactly what read/write/subscribe will fail to resolve — and the shared Contracts /// type stays free of driver bookkeeping. /// /// The map just built from the live session. private void RecordSkippedTags(NamespaceMap map) { var authored = _options.RawTags; _skippedTags.BeginRebuild(authored.Count); foreach (var entry in authored) { if (string.IsNullOrWhiteSpace(entry.RawPath)) _skippedTags.Skip("", "entry has no RawPath"); else if (!map.TryResolve(entry.RawPath, out _)) _skippedTags.Skip(entry.RawPath, "TagConfig has no usable 'nodeId' — no upstream node to dial"); } _skippedTags.Commit(); } private readonly ILogger _logger; /// Initializes a new instance of the class. /// Driver configuration. /// Stable logical ID from the config DB. /// Optional logger; defaults to NullLogger when not supplied. /// /// Optional shared secret resolver used to resolve secret:-prefixed /// / /// at session-open. /// Defaults to the null-object (which reports every /// secret absent) for the test/parse-only path; the production factory threads the /// DI-registered resolver. A secret: ref against the null-object fails closed. /// public OpcUaClientDriver(OpcUaClientDriverOptions options, string driverInstanceId, ILogger? logger = null, ISecretResolver? secretResolver = null) { _options = options; _driverInstanceId = driverInstanceId; _diag = new OpcUaClientDiagnostics(driverInstanceId); _logger = logger ?? NullLogger.Instance; _secretResolver = secretResolver ?? NullSecretResolver.Instance; _sdkTelemetry = logger is null ? SdkTelemetry.Context : SdkTelemetry.ForLogger(logger); PostRecoveryReconcileDelay = () => _options.Timeout + _options.KeepAliveInterval; } /// Mutable so can adopt a re-parsed config on reinitialize. /// Only ever written on the init path, before any reader/session uses it. private OpcUaClientDriverOptions _options; private readonly ISecretResolver _secretResolver; private readonly string _driverInstanceId; // ---- IAlarmSource state ---- private readonly System.Collections.Concurrent.ConcurrentDictionary _alarmSubscriptions = new(); /// RawPath → the upstream condition NodeId last seen for it. The host acknowledges by /// RawPath (its only identity for the condition) but Part 9 Acknowledge must be called on the CONDITION /// node, which differs from the authored source node. private readonly System.Collections.Concurrent.ConcurrentDictionary _alarmConditionByRawPath = new(StringComparer.Ordinal); /// The (Active, Acked) state last told to the host per reported condition id, from /// which picks the Kind sequence that moves the host's projector to the /// upstream event's state. private readonly Dictionary _alarmStateByRawPath = new(StringComparer.Ordinal); /// Serialises the read-compute-emit over . Event /// notifications arrive on SDK publish threads (and, across a resubscribe, from two subscriptions at once), so /// a lock-free prior/target exchange could emit two kind sequences computed from the same prior. private readonly object _alarmStateLock = new(); /// The ref set the last carried. Survives /// deliberately — see . private HashSet? _lastAlarmRefSet; /// Latched once the upstream server has refused ConditionRefresh as unsupported (logged once). private int _conditionRefreshUnsupportedLogged; private long _nextAlarmSubscriptionId; /// Occurs when an alarm event is received from the remote OPC UA server. public event EventHandler? OnAlarmEvent; // ---- ISubscribable + IHostConnectivityProbe state ---- private readonly System.Collections.Concurrent.ConcurrentDictionary _subscriptions = new(); private long _nextSubscriptionId; private readonly object _probeLock = new(); private HostState _hostState = HostState.Unknown; private DateTime _hostStateChangedUtc = DateTime.UtcNow; private KeepAliveEventHandler? _keepAliveHandler; /// Occurs when a monitored item's data value changes on the remote OPC UA server. public event EventHandler? OnDataChange; /// Occurs when the host connectivity status of the remote OPC UA server changes. public event EventHandler? OnHostStatusChanged; /// // Raised at most once per session, and only on a DEFINITIVE loss // (IsDefinitiveSessionLoss, or a tracked subscription the SERVER reported dropped — never the // publishing-stall heuristic). Re-armed when the session is released (re-initialise / shutdown). Known gap: the host // acts on it only while Connected, so a raise it receives before it has processed the successful initialise is // swallowed and that session falls back to the comms-loss watchdog. public event EventHandler? ConnectionLost; // OPC UA StatusCode constants the driver surfaces for local-side faults. Upstream-server // StatusCodes are passed through verbatim per driver-specs.md §8 "cascading quality" — // downstream clients need to distinguish 'remote source down' from 'local driver failure'. private const uint StatusBadNodeIdInvalid = 0x80330000u; private const uint StatusBadInternalError = 0x80020000u; private const uint StatusBadCommunicationError = 0x80050000u; private const uint StatusBadNoCommunication = 0x80310000u; private const uint StatusGood = 0x00000000u; private readonly SemaphoreSlim _gate = new(1, 1); /// Active OPC UA session. Null until returns cleanly. internal ISession? Session { get; private set; } /// /// R06:O-6 — subscription management (data + alarm subscribe/unsubscribe, ConditionRefresh sweeps) has its own /// gate, so a slow subscription create no longer queues reads, and the reconcile reads never wait on it. /// private readonly SemaphoreSlim _subscriptionGate = new(1, 1); /// Read gate: serialises reads (incl. the recovery / reconcile reads) and acknowledges. Writes and the /// discovery browse take no gate (the SDK session is safe for concurrent service calls); subscription management /// takes . internal SemaphoreSlim Gate => _gate; /// Test seam: the subscription-management gate. internal SemaphoreSlim SubscriptionGate => _subscriptionGate; /// /// Test-only seam to swap the active the way a re-initialise does at /// runtime, so the stale-session race the read/history paths guard against (re-read Session /// inside _gate, not before) can be exercised deterministically without a live connect. /// Production code never calls this — owns the real swap. /// /// The session instance to install, or null to clear it. internal void SetSessionForTest(ISession? session) => Session = session; /// Test-only seam: the most recent in-place recovery task. internal Task RecoveryTaskForTest => _recoveryTask; /// Test seam: the trailing reconcile read scheduled by the last in-place recovery. internal Task ReconcileTaskForTest => _reconcileTask; /// Test-only seam: raise the session's PublishError as the SDK publish loop does. /// The publish error status. /// The subscription id the error names (0 = none). internal void RaisePublishErrorForTest(StatusCode status, uint subscriptionId = 0) => HandlePublishError(Session, status, subscriptionId); /// Test-only seam: install a health snapshot as a completed connect would. /// The health to install. internal void SetHealthForTest(DriverHealth health) => _health = health; /// Test-only seam: raise the session keep-alive callback as the SDK timer does. /// The keep-alive status. /// The raising session; defaults to the current one. internal void RaiseKeepAliveForTest(StatusCode status, ISession? sender = null) => OnKeepAlive(sender ?? Session!, new KeepAliveEventArgs(new ServiceResult(status), ServerState.Running, DateTime.UtcNow)); /// /// Test-only seam: register a data subscription over as /// would, without the wire create. /// /// The RawPaths subscribed. /// The subscription handle. internal ISubscriptionHandle AddSubscriptionForTest(IReadOnlyList refs) { var id = Interlocked.Increment(ref _nextSubscriptionId); var handle = new OpcUaSubscriptionHandle(id); var subscription = new Subscription(telemetry: null!, new SubscriptionOptions { DisplayName = $"opcua-sub-{id}" }); var items = new List(); foreach (var r in refs) { var item = new MonitoredItem(telemetry: null!, new MonitoredItemOptions { DisplayName = r }) { Handle = r }; MonitoredItemNotificationEventHandler h = (mi, args) => OnMonitoredItemNotification(handle, mi, args); item.Notification += h; items.Add(new MonitoredItemNotificationHandle(item, h)); } _subscriptions[id] = new RemoteSubscription(subscription, handle, items); return handle; } /// Test-only seam: deliver one data notification as the SDK publish thread would. /// The subscription handle. /// The RawPath. /// The delivered value. internal void DeliverNotificationForTest(ISubscriptionHandle handle, string fullRef, DataValue value) => PublishNotification((OpcUaSubscriptionHandle)handle, fullRef, value); /// Test-only seam: register an alarm subscription over as /// would, without the wire create. /// The alarm RawPaths subscribed. /// The alarm subscription handle. internal IAlarmSubscriptionHandle AddAlarmSubscriptionForTest(IReadOnlyList refs) { var id = Interlocked.Increment(ref _nextAlarmSubscriptionId); var handle = new OpcUaAlarmSubscriptionHandle(id); BeginAlarmSubscription(refs); var item = new MonitoredItem(telemetry: null!, new MonitoredItemOptions { DisplayName = "Server/Events" }); _alarmSubscriptions[id] = new RemoteAlarmSubscription( new Subscription(telemetry: null!, new SubscriptionOptions { DisplayName = $"{AlarmSubscriptionNamePrefix}{id}" }), handle, item, (_, _) => { }); return handle; } /// /// Test-only seam to install the that connect builds at /// runtime, so the v3 RawPath → upstream node id resolution can be exercised without a live /// connect (which is what populates ). Production code never calls /// this — owns the real build. /// /// The namespace map to install, or null to clear it. internal void SetNamespaceMapForTest(NamespaceMap? map) => _namespaceMap = map; /// Test-only view of the diagnostics counters. internal OpcUaClientDiagnostics DiagnosticsForTest => _diag; private DriverHealth _health = new(DriverState.Unknown, null, null); /// Per-instance event counters published as OTel metrics. private readonly OpcUaClientDiagnostics _diag; private NotificationEventHandler? _notificationHandler; private PublishErrorEventHandler? _publishErrorHandler; private bool _disposed; /// URL of the endpoint the driver actually connected to. Exposed via . private string? _connectedEndpointUrl; /// /// Cert-validation delegate wired when /// is true. Stored so / can /// detach it from the (potentially process-shared) /// and avoid leaking the closure. /// private CertificateValidationEventHandler? _certValidationHandler; /// The that owns . private CertificateValidator? _certValidatorRef; /// /// 1 while the CURRENT session's keep-alive is bad (an outage is in progress), else 0. The /// 0→1 edge publishes the outage once (Bad per subscribed ref + a quality-only alarm annotation); the 1→0 /// edge on a good keep-alive recovers in place. Reset whenever the session is released, so a fresh session /// reports its own loss. There is deliberately no in-driver reconnect any more: a session that never comes /// back keeps health Degraded with frozen, which is exactly what /// the host's comms-loss watchdog trips on (→ ShutdownAsync + InitializeAsync + re-subscribe). /// private int _sessionLost; /// /// 1 once a tracked subscription is known dead while the SESSION lives (it stopped /// publishing past its keep-alive window, or the server said BadNoSubscription / BadSubscriptionIdInvalid — /// e.g. an outage longer than the subscription lifetime but shorter than the session timeout). Sticky until /// the session is released: health stays Degraded and NOTHING advances /// (good keep-alives would otherwise keep a driver whose data plane is dead looking alive forever), so the /// host's comms-loss watchdog re-initialises the driver and the re-subscribe recreates the subscription. /// private int _subscriptionLost; /// Ticks of the first good keep-alive that saw a tracked subscription not publishing /// (0 = none stalled). Only the keep-alive thread and the session release touch it. private long _publishingStoppedSinceTicks; /// /// How long after the driver deleted its LAST server-side subscription an anonymous BadNoSubscription /// (subscription id 0) is attributed to that delete rather than to a loss of whatever was subscribed since /// (Gitea #763). A publish request the SDK had queued while the session held no subscription is answered /// BadNoSubscription, and its PublishError can be handled AFTER the next subscription is registered /// — an unsubscribe→resubscribe of the same set (DriverInstanceActor's reconcile) then looked like the /// server dropping the new subscription. The SDK names no subscription on that status, so the delete is the /// only thing it can be attributed to. Bounded on purpose: a genuine loss inside the window still degrades the /// driver through the publishing-stall heuristic (); only the immediate /// is forgone for it, and only within this window. /// internal static readonly TimeSpan StaleNoSubscriptionGrace = TimeSpan.FromSeconds(5); /// /// UTC ticks of the last time the driver deleted a subscription that left it tracking none (0 = never on this /// session) — the moment the server could legitimately answer BadNoSubscription. See /// . Reset when the session is released. /// private long _lastSubscriptionsEmptiedTicks; /// The last in-place recovery started. Test seam: lets a test await it deterministically. private Task _recoveryTask = Task.CompletedTask; /// 1 once has been raised for the CURRENT session (once per session); /// cleared when the session is released, so a re-initialised driver reports its next loss. private int _connectionLostRaised; /// Closed-channel keep-alive failures () in the current /// outage; reset by a good keep-alive and when the session is released. Two make a closed channel "sustained". private int _closedChannelFailures; /// Closed-channel keep-alive failures in one outage that make the loss definitive (a single one may be a /// transient that the next good keep-alive contradicts). private const int SustainedClosedChannelFailures = 2; /// The trailing reconcile read scheduled by the last in-place recovery. Test seam. private Task _reconcileTask = Task.CompletedTask; /// Late notifications the replay guard dropped since it was armed, and the refs they were for; /// summarised in ONE Information line when the window disarms. Guarded by . private int _replayDrops; private readonly HashSet _replayDropRefs = new(StringComparer.Ordinal); /// /// Test seam: how long after an in-place recovery the trailing reconcile read runs. Production: the /// operation timeout plus one keep-alive interval — by then every request that was in flight across the stall /// (including the Publish responses whose notifications the SDK can lose) has completed or timed out. /// internal Func PostRecoveryReconcileDelay { get; set; } /// Test seam: whether a subscription has stopped publishing. Production asks the SDK. internal Func IsPublishingStopped { get; set; } = static s => s.Created && s.PublishingStopped; /// Test seam: how long a subscription may keep reporting stopped (measured on the /// driver's keep-alive cadence) before it is declared dead. Production: one more subscription keep-alive window /// plus 5 s, on top of the window the SDK's own PublishingStopped already waits. internal Func PublishingStallGrace { get; set; } = static s => TimeSpan.FromMilliseconds(Math.Max(s.CurrentPublishingInterval, 1) * (s.CurrentKeepAliveCount + 1d)) + TimeSpan.FromSeconds(5); /// The last value delivered per subscribed RawPath, so an outage can publish /// BadNoCommunication with the value retained (the same shape the host's own comms-loss fan uses). /// /// per RawPath, the source time (ticks) of the last Good value ACCEPTED from upstream (a /// notification, or the recovery read). Last, not max: in steady state it simply follows the upstream clock, /// so a clock that steps backwards is tracked rather than remembered as a high-water mark. It only seeds the /// replay guard below; nothing is ever dropped against it outside that guard's window. /// private readonly System.Collections.Concurrent.ConcurrentDictionary _lastGoodSourceTicks = new(StringComparer.Ordinal); /// /// the REPLAY guard. Armed only by a same-session recovery (keep-alive bad→good), when the /// SDK was observed delivering a missed notification seconds late with an older source time. Per ref it holds a /// HIGH-WATER MARK: seeded with the last Good upstream source time at arming, raised by the recovery read, and /// advanced by EVERY notification accepted while the window is open. A notification whose source time is at or /// below the mark is dropped — equal included, because OPC UA moves the source timestamp on every value or /// status change, so an equal one is a replay of what was already accepted. The guard holds for the WHOLE /// window: the F2 live re-gate saw fresh notifications ~1 s after recovery and the SDK's late replay 4–6 s /// later, so the original "disarm a ref on its first in-order notification" let the replay through. A /// notification with no source time passes and leaves the mark alone. The whole guard expires at /// ; outside it every notification is accepted whatever its timestamp — a /// global "never go backwards" rule would freeze a node for as long as an upstream clock that stepped back /// (NTP correction, a PLC with a reset clock) takes to catch up. /// private readonly System.Collections.Concurrent.ConcurrentDictionary _replayGuard = new(StringComparer.Ordinal); /// Clock ticks at which the replay guard expires (0 = not armed). private long _replayGuardUntilTicks; /// Serialises arming against expiry, so an expiry racing a re-arm cannot wipe the new seeds. private readonly object _replayGuardLock = new(); /// /// Review of 3b88056f — serialises "decide whether this upstream value may be published" with "publish it" across /// the three paths that publish upstream data: live notifications, the recovery read and the trailing reconcile /// read. Without it a path could pass its newer-than check, lose the CPU to another path publishing a NEWER value, /// and then deliver its older one last — regressing the node. It also covers the comms-loss Bad fan /// (). One driver-wide lock rather than one per ref: it also serialises delivery /// ACROSS subscriptions — with several subscriptions the SDK dispatches each subscription's messages on its own /// worker, so their notifications no longer run in parallel. That is the cost, and it is small: each critical /// section is a dictionary update plus a subscriber call that only Tells an actor (tens of nanoseconds to /// a few microseconds), nothing awaits under it, and per-subscription message order is unaffected. Because /// subscribers are invoked under it, a subscriber must never block on, or call back into, this driver. /// private readonly Lock _deliveryLock = new(); /// /// Re-review of 30a9d5f0 — per RawPath, a counter bumped (under ) every time a value is /// delivered for it. The trailing re-read captures it before issuing its read and skips any ref delivered to since: /// without source timestamps the read's value cannot be ordered against a notification that landed while the read /// was in flight, and publishing it would regress the node to the value the read sampled. /// private readonly System.Collections.Concurrent.ConcurrentDictionary _deliverySeq = new(StringComparer.Ordinal); /// Review of 3b88056f — cancels the post-recovery background work (trailing read, replay-guard disarm timer) /// of the current session; cancelled and replaced whenever the session is released, and on dispose. private CancellationTokenSource _sessionWorkCts = new(); /// Test seam (review of 3b88056f): invoked inside the trailing read's critical section, after it decided /// to publish a ref and before it does — where a racing notification would otherwise overtake it. internal Action? BeforeReconcilePublishForTest { get; set; } /// Floor of the replay-guard window; the live gate saw the late replay 4–5 s after recovery. private static readonly TimeSpan ReplayGuardMinimum = TimeSpan.FromSeconds(30); private readonly System.Collections.Concurrent.ConcurrentDictionary _lastValues = new(StringComparer.Ordinal); /// Refs published Bad by the current outage and not yet re-published Good. A notification /// clears its ref (fresh data beats the recovery read); the recovery read publishes only what is left. private readonly System.Collections.Concurrent.ConcurrentDictionary _awaitingRecovery = new(StringComparer.Ordinal); /// Ticks of the last advance from a notification — /// throttles the per-notification health write to at most one per second. private long _lastEvidenceStampTicks; /// Clock for liveness stamps. Test seam; production uses . internal TimeProvider Clock { get; set; } = TimeProvider.System; /// /// SDK telemetry for this instance's sessions: SDK warnings (keep-alive / publish / channel errors) go to the /// driver's own logger (the shared has no provider, which made every /// SDK-side session error invisible). /// private readonly ITelemetryContext _sdkTelemetry; /// /// Bidirectional namespace map built at connect time from session.NamespaceUris. /// Stored NodeIds embed the server-stable namespace URI rather than the /// session-relative ns=N index, so a remote-server namespace-table reorder /// across a restart does not silently re-point stored references at the wrong /// namespace (driver-specs.md §8 "Namespace Remapping"). /// Null until returns cleanly. /// private NamespaceMap? _namespaceMap; /// public string DriverInstanceId => _driverInstanceId; /// public string DriverType => "OpcUaClient"; /// True when the supplied DriverConfig JSON carries a real body. The bootstrapper always /// passes a populated document; some unit tests pass "{}" or an empty string to exercise /// lifecycle shape without a config — those keep the constructor-supplied options. private static bool HasConfigBody(string? driverConfigJson) { if (string.IsNullOrWhiteSpace(driverConfigJson)) return false; var trimmed = driverConfigJson.Trim(); return trimmed is not "{}" and not "[]"; } /// public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) { // Every lifecycle transition (initialise / shutdown) takes a new generation. Everything this attempt // publishes — the session, the namespace map, health — is published only while its generation is still // the current one, so an init the host abandoned (30 s bound) that finishes after the next ShutdownAsync // or InitializeAsync cannot resurrect a session, or overwrite the newer health, behind its back. var generation = Interlocked.Increment(ref _lifecycleGeneration); // InitializeAsync is reachable on a live driver with no teardown first (the IDriver contract // allows it, and a reconnect retry can overlap a running attempt); `Session = session` below would otherwise orphan the previous session, // its keep-alive handler and any reconnect handler. await ReleaseSessionAsync(generation, cancellationToken).ConfigureAwait(false); WriteLifecycleHealth(generation, _ => new DriverHealth(DriverState.Initializing, null, null)); ISession? session = null; KeepAliveEventHandler? keepAliveHandler = null; var certValidation = new CertValidationHook(); var installed = false; try { // re-parse the supplied DriverConfig so a config change delivered through the IDriver // contract is honoured. Without this the driver keeps serving the options it was CONSTRUCTED // with — the endpoint, security policy and tag set were all frozen at construction, and only // secret rotation was picked up. if (HasConfigBody(driverConfigJson)) { var parsed = OpcUaClientDriverFactoryExtensions.ParseOptions(_driverInstanceId, driverConfigJson); // Review m2: a superseded init must not overwrite the options a newer one adopted. lock (_probeLock) { if (Volatile.Read(ref _lifecycleGeneration) == generation) _options = parsed; } } ValidateTuning(_options); var appConfig = await BuildApplicationConfigurationAsync(certValidation, cancellationToken).ConfigureAwait(false); // when a DiscoveryUrl is configured, FindServers/GetEndpoints on it first and try the // discovered URLs ahead of the static list. Non-fatal — an LDS outage must not block init. IReadOnlyList discovered = []; if (!string.IsNullOrWhiteSpace(_options.DiscoveryUrl)) { IReadOnlyList found = []; try { found = await DiscoverEndpointsAsync( appConfig, _options.DiscoveryUrl!, _options.SecurityPolicy, _options.SecurityMode, cancellationToken).ConfigureAwait(false); } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { throw; } catch (Exception ex) { _logger.LogWarning(ex, "OpcUaClientDriver '{DriverInstanceId}': discovery against '{DiscoveryUrl}' failed; " + "using the configured endpoint list only.", _driverInstanceId, _options.DiscoveryUrl); } // Selection is OUTSIDE the try on purpose: an ambiguous LDS must fail the connect with a // clear message, not be swallowed into "discovery unavailable". discovered = SelectDiscoveredUrls(found, _options.DiscoveryServerUri); } var candidates = ResolveEndpointCandidates(_options, discovered); // G-2b: resolve any secret:-prefixed Password / UserCertificatePassword through the // shared secret store ONCE, here in the async connect flow, just before the credentials // are consumed. _options stays the raw config (with secret: refs) — only this // connect-scoped local carries plaintext, and only for the duration of the connect. // Resolving on every InitializeAsync (a full reconnect calls it) picks up rotations. // Both credential consumers — the Username password below and the Certificate password // in BuildCertificateIdentity — flow through BuildUserIdentity(resolvedOptions), so a // single resolve covers both paths. var resolvedOptions = await OpcUaClientSecretResolution .ResolveSecretRefsAsync(_options, _secretResolver, cancellationToken).ConfigureAwait(false); var identity = BuildUserIdentity(resolvedOptions); // Failover sweep: try each endpoint in order, return the session from the first // one that successfully connects. Per-endpoint failures are captured so the final // aggregate exception names every URL that was tried and why — critical diag for // operators debugging 'why did the failover pick #3?'. var attemptErrors = new List(candidates.Count); string? connectedUrl = null; foreach (var url in candidates) { try { session = OpenSessionForTest is { } open ? await open(url, cancellationToken).ConfigureAwait(false) : await OpenSessionOnEndpointAsync( appConfig, url, _options.SecurityPolicy, _options.SecurityMode, identity, cancellationToken).ConfigureAwait(false); connectedUrl = url; break; } catch (Exception ex) { attemptErrors.Add($"{url} -> {ex.GetType().Name}: {ex.Message}"); } } if (session is null) throw new AggregateException( "OPC UA Client failed to connect to any of the configured endpoints. " + "Tried:\n " + string.Join("\n ", attemptErrors), attemptErrors.Select(e => new InvalidOperationException(e))); // Wire the session's keep-alive channel into HostState + health. OPC UA keep-alives // are authoritative for session liveness: the SDK pings on KeepAliveInterval. A good one is // liveness evidence (advances LastSuccessfulRead); a bad one degrades health and publishes the // outage. Recovery from a session that does not come back is the host's comms-loss watchdog // (ShutdownAsync + InitializeAsync + re-subscribe) — there is no in-driver reconnect handler. // The handler is attached to THIS session only; OnKeepAlive ignores any sender that is not the // installed session, so a session discarded below never reaches health. keepAliveHandler = OnKeepAlive; session.KeepAlive += keepAliveHandler; AttachDiagnostics(session); // Build the bidirectional namespace map from the freshly negotiated session's // NamespaceUris (driver-specs.md §8 "Namespace Remapping"). Stored NodeIds carry // the namespace URI, not the session-relative ns=N index, so a remote namespace // reorder across a restart can't silently misaddress nodes. v3: the same map also // ingests the authored RawTags so read/write/subscribe/history references (RawPaths) // resolve to their upstream node ids. var map = NamespaceMap.FromSession(session, _options.RawTags); // Publish the session only while this attempt is still the current lifecycle generation. The // generation check and the install share _probeLock with ReleaseSessionAsync's take, so a // ShutdownAsync either sees (and releases) this session or makes this attempt discard it. lock (_probeLock) { if (Volatile.Read(ref _lifecycleGeneration) == generation) { _namespaceMap = map; Session = session; _keepAliveHandler = keepAliveHandler; _connectedEndpointUrl = connectedUrl; // Review m2: the cert-validation handler is published with the session — a previous one (a // re-initialise without shutdown) is detached first so it is not leaked on its validator. DetachCertValidation(_certValidatorRef, _certValidationHandler); _certValidationHandler = certValidation.Handler; _certValidatorRef = certValidation.Validator; installed = true; } } if (!installed) throw new OperationCanceledException( $"OpcUaClientDriver '{_driverInstanceId}': initialise was superseded by a later lifecycle " + "transition (shutdown or re-initialise); its session was discarded."); RecordSkippedTags(map); WriteLifecycleHealth(generation, _ => new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null)); if (Volatile.Read(ref _lifecycleGeneration) == generation) TransitionTo(HostState.Running); } catch (Exception ex) { // R06:O-2 — close the LOCAL session this attempt opened (the property is still null, or belongs to a // newer attempt), so a throw after the session activated (AttachDiagnostics, the namespace map, the // superseded check) does not orphan it and its keep-alive timer until GC. if (session is not null && !installed) await DiscardSessionAsync(session, keepAliveHandler).ConfigureAwait(false); if (!installed) DetachCertValidation(certValidation.Validator, certValidation.Handler); if (Volatile.Read(ref _lifecycleGeneration) == generation) { if (installed) await ReleaseSessionAsync(generation, CancellationToken.None).ConfigureAwait(false); WriteLifecycleHealth(generation, h => new DriverHealth(DriverState.Faulted, h.LastSuccessfulRead, ex.Message)); } throw; } } /// The auto-accept certificate-validation handler one initialise attached, handed back to it (review m2). private sealed class CertValidationHook { public CertificateValidationEventHandler? Handler { get; set; } public CertificateValidator? Validator { get; set; } } private static void DetachCertValidation(CertificateValidator? validator, CertificateValidationEventHandler? handler) { if (validator is null || handler is null) return; try { validator.CertificateValidation -= handler; } catch { /* best-effort */ } } /// /// Test seam: opens the session for one endpoint URL in place of the real discovery + session create, so /// the post-connect half of (and its superseded / failure paths) can be driven /// against a mock session. Production leaves it null. /// internal Func>? OpenSessionForTest { get; set; } /// /// Close and dispose a session this driver never installed (a failed or superseded initialise): detach the /// handlers attached to it, close it within , dispose it. Best-effort. /// /// The session to discard. /// The keep-alive handler attached to it, if any. private async Task DiscardSessionAsync(ISession session, KeepAliveEventHandler? keepAliveHandler) { if (keepAliveHandler is not null) { try { session.KeepAlive -= keepAliveHandler; } catch { /* best-effort */ } } DetachDiagnostics(session); try { using var closeCts = new CancellationTokenSource(SessionCloseTimeout); await CloseBoundedAsync(session, closeCts.Token).ConfigureAwait(false); } catch { /* best-effort */ } try { session.Dispose(); } catch { /* best-effort */ } } /// /// Close a session, abandoning the wait when fires even if the SDK call itself ignores /// the token (a half-dead upstream must not hold a teardown past its budget). /// private static async Task CloseBoundedAsync(ISession session, CancellationToken ct) { var close = session.CloseAsync((int)SessionCloseTimeout.TotalMilliseconds, true, ct) ?? Task.FromResult(StatusCodes.Good); await close.WaitAsync(ct).ConfigureAwait(false); } /// /// Write a lifecycle health snapshot only while is still the current lifecycle /// generation (CAS against concurrent keep-alive / data-path writers, re-checking the generation each try). /// /// The generation of the init / shutdown making the write. /// Maps the current health to the one to write. private void WriteLifecycleHealth(long generation, Func next) => UpdateHealth(generation, next); /// /// Build a minimal in-memory . Certificates live /// under the OS user profile — on Windows that's %LocalAppData%\OtOpcUa\pki /// — so multiple driver instances in the same OtOpcUa server process share one /// certificate store without extra config. /// /// Receives the auto-accept handler this build attached (if any) — published by the /// caller only when its initialise is still current (review m2), detached otherwise. /// Cancellation token. private async Task BuildApplicationConfigurationAsync( CertValidationHook certValidation, CancellationToken ct) { // The ITelemetryContext overload is the non-obsolete ctor; SDK warnings reach the driver's // logger when one was supplied. var app = new ApplicationInstance(_sdkTelemetry) { ApplicationName = _options.SessionName, ApplicationType = ApplicationType.Client, }; var pkiRoot = Path.Combine( Environment.GetFolderPath(Environment.SpecialFolder.LocalApplicationData), "OtOpcUa", "pki"); var config = new ApplicationConfiguration { ApplicationName = _options.SessionName, ApplicationType = ApplicationType.Client, ApplicationUri = _options.ApplicationUri, SecurityConfiguration = new SecurityConfiguration { ApplicationCertificate = new CertificateIdentifier { StoreType = CertificateStoreType.Directory, StorePath = Path.Combine(pkiRoot, "own"), SubjectName = $"CN={_options.SessionName}", }, TrustedPeerCertificates = new CertificateTrustList { StoreType = CertificateStoreType.Directory, StorePath = Path.Combine(pkiRoot, "trusted"), }, TrustedIssuerCertificates = new CertificateTrustList { StoreType = CertificateStoreType.Directory, // the issuer store's crl/ sub-folder is where the SDK reads issuer CRLs, so an // operator-supplied path is the CRL knob. Unset = the historical default. StorePath = string.IsNullOrWhiteSpace(_options.CertificateValidation.IssuerStorePath) ? Path.Combine(pkiRoot, "issuers") : _options.CertificateValidation.IssuerStorePath, }, RejectedCertificateStore = new CertificateTrustList { StoreType = CertificateStoreType.Directory, StorePath = Path.Combine(pkiRoot, "rejected"), }, AutoAcceptUntrustedCertificates = _options.AutoAcceptCertificates, RejectUnknownRevocationStatus = _options.CertificateValidation.RejectUnknownRevocationStatus, }, TransportQuotas = new TransportQuotas { OperationTimeout = (int)_options.Timeout.TotalMilliseconds }, ClientConfiguration = new ClientConfiguration { DefaultSessionTimeout = (int)_options.SessionTimeout.TotalMilliseconds, }, DisableHiResClock = true, }; await config.ValidateAsync(ApplicationType.Client, ct).ConfigureAwait(false); // AutoAccept=true is a dev-only escape hatch. Emit a prominent warning so a // production misconfiguration is immediately visible in logs. if (_options.AutoAcceptCertificates) { _logger.LogWarning( "OpcUaClientDriver '{DriverInstanceId}': AutoAcceptCertificates=true — all " + "remote server certificate errors are accepted, including expired / wrong-host " + "/ chain-incomplete. This MUST be false in production to prevent MITM attacks " + "against the opc.tcp channel.", _driverInstanceId); // Accept the full set of certificate-validation error codes: a real dev cert can // fail with BadCertificateChainIncomplete, BadCertificateTimeInvalid, or // BadCertificateHostNameInvalid, not only BadCertificateUntrusted. Only accepting // the latter would silently fail for those certs. A REVOKED certificate is the one // exception: auto-accept is for dev self-signed certs, never for a cert its issuer // has explicitly withdrawn. CertificateValidationEventHandler handler = (_, e) => { if (IsRevocationFailure(e.Error.StatusCode)) { _logger.LogWarning( "OpcUaClientDriver '{DriverInstanceId}': REVOKED server certificate rejected " + "(subject={Subject}, status=0x{Status:X8}) — AutoAcceptCertificates does not apply to revoked certs.", _driverInstanceId, e.Certificate?.Subject, e.Error.StatusCode.Code); e.Accept = false; return; } e.Accept = true; }; config.CertificateValidator.CertificateValidation += handler; // Handed back to the caller, which stores the refs (so ShutdownAsync + Dispose can detach the delegate) // only while its initialise is the current lifecycle generation. certValidation.Handler = handler; certValidation.Validator = config.CertificateValidator; } // Ensure an application certificate exists. The SDK auto-generates one if missing. app.ApplicationConfiguration = config; await app.CheckApplicationInstanceCertificatesAsync(silent: true, lifeTimeInMonths: null, ct) .ConfigureAwait(false); return config; } /// True for the two hard "this certificate (or its issuer) is revoked" results. /// The SDK's certificate-validation status. /// true when the status is Revoked / IssuerRevoked. internal static bool IsRevocationFailure(StatusCode status) => status.Code == StatusCodes.BadCertificateRevoked || status.Code == StatusCodes.BadCertificateIssuerRevoked; /// /// Resolve the ordered failover candidate list. EndpointUrls wins when /// non-empty; otherwise fall back to EndpointUrl as a single-URL shortcut so /// existing single-endpoint configs keep working without migration. /// /// Driver options containing endpoint configuration. /// The ordered list of endpoint URLs to try. internal static IReadOnlyList ResolveEndpointCandidates(OpcUaClientDriverOptions opts) => ResolveEndpointCandidates(opts, []); /// /// Resolve the ordered failover list with optional discovery results. The configured /// endpoints always come FIRST — a discovered endpoint can only ever be a fallback, never outrank or /// replace the server the operator named. Discovered URLs follow, de-duplicated case-insensitively. /// With no discovered URLs the result is exactly the pre-discovery behaviour. /// /// Driver options containing endpoint configuration. /// URLs resolved via (may be empty). /// The ordered list of endpoint URLs to try. internal static IReadOnlyList ResolveEndpointCandidates( OpcUaClientDriverOptions opts, IReadOnlyList discovered) { IReadOnlyList staticList = opts.EndpointUrls is { Count: > 0 } ? opts.EndpointUrls : [opts.EndpointUrl]; if (discovered.Count == 0) return staticList; var seen = new HashSet(StringComparer.OrdinalIgnoreCase); var merged = new List(discovered.Count + staticList.Count); foreach (var u in staticList) if (!string.IsNullOrWhiteSpace(u) && seen.Add(u)) merged.Add(u); foreach (var u in discovered) if (!string.IsNullOrWhiteSpace(u) && seen.Add(u)) merged.Add(u); return merged; } /// /// FindServers on (an LDS, or the server itself), then /// GetEndpoints on each returned server's discovery URL, keeping the endpoints whose security /// policy + mode match. One unreachable server in the list is skipped, not fatal. The discovery /// channel is unsecured by design (OPC UA Part 4 §5.4) regardless of the session's security mode. /// /// Application configuration (client identity for the discovery channel). /// The LDS / server discovery URL. /// Required security policy. /// Required security mode. /// Cancellation token. /// Matching endpoint URLs in discovery order, de-duplicated. internal static async Task> DiscoverEndpointsAsync( ApplicationConfiguration appConfig, string discoveryUrl, OpcUaSecurityPolicy policy, OpcUaSecurityMode mode, CancellationToken ct) { var wantedPolicyUri = MapSecurityPolicy(policy); var wantedMode = ToMessageSecurityMode(mode); ApplicationDescriptionCollection servers; using (var lds = await DiscoveryClient.CreateAsync( appConfig, new Uri(discoveryUrl), Opc.Ua.DiagnosticsMasks.None, ct).ConfigureAwait(false)) { servers = await lds.FindServersAsync(null, ct).ConfigureAwait(false); } var found = new List(); foreach (var server in servers) { if (server.DiscoveryUrls is null) continue; foreach (var serverDiscoveryUrl in server.DiscoveryUrls) { if (string.IsNullOrWhiteSpace(serverDiscoveryUrl)) continue; EndpointDescriptionCollection endpoints; try { using var ep = await DiscoveryClient.CreateAsync( appConfig, new Uri(serverDiscoveryUrl), Opc.Ua.DiagnosticsMasks.None, ct).ConfigureAwait(false); endpoints = await ep.GetEndpointsAsync(null, ct).ConfigureAwait(false); } catch (OperationCanceledException) when (ct.IsCancellationRequested) { throw; } catch { continue; } found.AddRange(FilterEndpointUrls(endpoints, wantedPolicyUri, wantedMode) .Select(u => new DiscoveredEndpoint(server.ApplicationUri ?? string.Empty, u))); } } return found; } /// One endpoint URL found by discovery, with the ApplicationUri of the server advertising it. /// The advertising server's ApplicationUri (empty when it published none). /// The endpoint URL. internal readonly record struct DiscoveredEndpoint(string ApplicationUri, string Url); /// /// Decide which discovered endpoints may be used. An LDS can register many servers (identical /// line PLCs, a gateway plus a historian), and the first one that accepts a connection would be bound /// silently — mis-attributing every value and write. So: with set, only /// that ApplicationUri's endpoints; unset and exactly one distinct server, that server; unset and /// several, throw naming them so the operator pins one. /// /// Matching endpoints from every server the LDS listed. /// Optional DiscoveryServerUri filter. /// The usable URLs, de-duplicated, in discovery order (possibly empty). /// Several servers discovered and no filter configured. internal static IReadOnlyList SelectDiscoveredUrls( IReadOnlyList found, string? serverUri) { IEnumerable chosen = found; if (!string.IsNullOrWhiteSpace(serverUri)) { chosen = found.Where(e => string.Equals(e.ApplicationUri, serverUri, StringComparison.Ordinal)); } else { var servers = found.Select(e => e.ApplicationUri).Distinct(StringComparer.Ordinal).ToList(); if (servers.Count > 1) throw new InvalidOperationException( "OpcUaClient discovery found endpoints for more than one server " + $"({string.Join(", ", servers.Select(u => string.IsNullOrEmpty(u) ? "" : u))}). " + "Set DiscoveryServerUri to the ApplicationUri of the server this driver should use."); } return chosen.Select(e => e.Url).Distinct(StringComparer.OrdinalIgnoreCase).ToList(); } /// The endpoint URLs in matching the wanted policy URI + mode. /// Endpoints advertised by a server. /// Required SecurityPolicyUri. /// Required message security mode. /// Matching EndpointUrls, in advertised order. internal static IEnumerable FilterEndpointUrls( IEnumerable endpoints, string wantedPolicyUri, MessageSecurityMode wantedMode) => endpoints .Where(e => e.SecurityPolicyUri == wantedPolicyUri && e.SecurityMode == wantedMode && !string.IsNullOrWhiteSpace(e.EndpointUrl)) .Select(e => e.EndpointUrl); private static MessageSecurityMode ToMessageSecurityMode(OpcUaSecurityMode mode) => mode switch { OpcUaSecurityMode.None => MessageSecurityMode.None, OpcUaSecurityMode.Sign => MessageSecurityMode.Sign, OpcUaSecurityMode.SignAndEncrypt => MessageSecurityMode.SignAndEncrypt, _ => throw new ArgumentOutOfRangeException(nameof(mode)), }; /// /// Range checks for the subscription-tuning and discovery options. Throws at /// so a bad config fails like the namespace-kind check does rather /// than surfacing later as a server-side BadInvalidArgument on every subscribe. /// /// The options to validate. internal static void ValidateTuning(OpcUaClientDriverOptions opts) { if (opts.SubscriptionKeepAliveCount < 1) throw new InvalidOperationException("OpcUaClient SubscriptionKeepAliveCount must be >= 1."); if (opts.SubscriptionLifetimeCount < 3L * opts.SubscriptionKeepAliveCount) throw new InvalidOperationException( "OpcUaClient SubscriptionLifetimeCount must be at least 3x SubscriptionKeepAliveCount " + $"(got lifetime={opts.SubscriptionLifetimeCount}, keep-alive={opts.SubscriptionKeepAliveCount})."); if (opts.SubscriptionPublishingInterval is { } pi && pi <= TimeSpan.Zero) throw new InvalidOperationException("OpcUaClient SubscriptionPublishingInterval must be positive when set."); if (!string.IsNullOrWhiteSpace(opts.DiscoveryUrl) && !Uri.TryCreate(opts.DiscoveryUrl, UriKind.Absolute, out _)) throw new InvalidOperationException( $"OpcUaClient DiscoveryUrl '{opts.DiscoveryUrl}' is not an absolute URI."); } /// /// Publishing interval (ms) a new subscription gets: the configured override if set, else the /// interval the runtime requested, floored at 50 ms either way (servers negotiate up anyway, and /// sub-50 ms just wastes negotiation). /// /// Driver options. /// The per-call interval from SubscribeAsync. /// The publishing interval in milliseconds. internal static int ResolvePublishingIntervalMs(OpcUaClientDriverOptions opts, TimeSpan requested) { var effective = opts.SubscriptionPublishingInterval ?? requested; return effective < TimeSpan.FromMilliseconds(50) ? 50 : (int)effective.TotalMilliseconds; } /// /// Build the user-identity token from the driver options. Split out of /// so the failover sweep reuses one identity across /// every endpoint attempt — generating it N times would re-unlock the user cert's /// private key N times, wasteful + keeps the password in memory longer. /// /// Driver options containing authentication configuration. /// The constructed for session activation. internal static UserIdentity BuildUserIdentity(OpcUaClientDriverOptions options) => OpcUaUserIdentityFactory.Build(options); /// /// Open a session against a single endpoint URL. Bounded by /// so the failover /// sweep doesn't spend its full budget on one dead server. Moved out of /// so the failover loop body stays readable. /// private async Task OpenSessionOnEndpointAsync( ApplicationConfiguration appConfig, string endpointUrl, OpcUaSecurityPolicy policy, OpcUaSecurityMode mode, UserIdentity identity, CancellationToken ct) { using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct); cts.CancelAfter(_options.PerEndpointConnectTimeout); var selected = await SelectMatchingEndpointAsync( appConfig, endpointUrl, policy, mode, cts.Token).ConfigureAwait(false); var endpointConfig = EndpointConfiguration.Create(appConfig); endpointConfig.OperationTimeout = (int)_options.Timeout.TotalMilliseconds; var endpoint = new ConfiguredEndpoint(null, selected, endpointConfig); var session = await new DefaultSessionFactory(_sdkTelemetry).CreateAsync( appConfig, endpoint, false, // updateBeforeConnect _options.SessionName, (uint)_options.SessionTimeout.TotalMilliseconds, identity, null, // preferredLocales cts.Token).ConfigureAwait(false); session.KeepAliveInterval = (int)_options.KeepAliveInterval.TotalMilliseconds; return session; } /// /// Select the remote endpoint matching both the requested /// and . The SDK's CoreClientUtils.SelectEndpointAsync /// only honours a boolean "use security" flag; we need policy-aware matching so an /// operator asking for Basic256Sha256 against a server that also offers /// Basic128Rsa15 doesn't silently end up on the weaker cipher. /// private static async Task SelectMatchingEndpointAsync( ApplicationConfiguration appConfig, string endpointUrl, OpcUaSecurityPolicy policy, OpcUaSecurityMode mode, CancellationToken ct) { // GetEndpoints returns everything the server advertises; policy + mode filter is // applied client-side so the selection is explicit and fails loudly if the operator // asks for a combination the server doesn't publish. DiscoveryClient.CreateAsync // is the non-obsolete path in SDK 1.5.378; the synchronous Create(..) variants are // all deprecated. using var client = await DiscoveryClient.CreateAsync( appConfig, new Uri(endpointUrl), Opc.Ua.DiagnosticsMasks.None, ct).ConfigureAwait(false); var all = await client.GetEndpointsAsync(null, ct).ConfigureAwait(false); var wantedPolicyUri = MapSecurityPolicy(policy); var wantedMode = ToMessageSecurityMode(mode); var match = all.FirstOrDefault(e => e.SecurityPolicyUri == wantedPolicyUri && e.SecurityMode == wantedMode); if (match is null) { var advertised = string.Join(", ", all .Select(e => $"{ShortPolicyName(e.SecurityPolicyUri)}/{e.SecurityMode}")); throw new InvalidOperationException( $"No endpoint at '{endpointUrl}' matches SecurityPolicy={policy} + SecurityMode={mode}. " + $"Server advertises: {advertised}"); } // connect where the operator pointed us, not where the server says it lives. match.EndpointUrl = RewriteEndpointUrl(match.EndpointUrl, endpointUrl); return match; } /// /// the address to open the session on. A server reports its endpoints with ITS view of its own /// address; behind NAT, a port map, a container port publish or a proxy that address is unreachable from here /// (or bypasses the proxy). When the advertised URL's scheme matches the configured one, the configured host /// AND port (when it names one) replace the advertised ones and the advertised path is kept — the SDK's own /// CoreClientUtils.SelectEndpoint rule, which this driver's policy-aware selection bypassed. A different /// scheme is a different transport and is left alone; an unparseable advertised URL falls back to the /// configured one. /// /// The EndpointUrl the server returned from GetEndpoints. /// The URL the driver was configured with (the one GetEndpoints was sent to). /// The URL to connect the session to. internal static string RewriteEndpointUrl(string? advertisedUrl, string configuredUrl) { if (!Uri.TryCreate(configuredUrl, UriKind.Absolute, out var configured)) return advertisedUrl ?? configuredUrl; if (string.IsNullOrWhiteSpace(advertisedUrl) || !Uri.TryCreate(advertisedUrl, UriKind.Absolute, out var advertised)) return configuredUrl; if (!string.Equals(advertised.Scheme, configured.Scheme, StringComparison.OrdinalIgnoreCase)) return advertisedUrl; // A port-less configured URL (Uri.Port == -1 for opc.tcp) keeps the advertised port — copying -1 would drop it // and the SDK would fall back to 4840 (review of 3b88056f; same fix as the client's EndpointSelector). var builder = new UriBuilder(advertised) { Host = configured.DnsSafeHost, Port = configured.Port > 0 ? configured.Port : advertised.Port, }; return builder.Uri.ToString(); } /// /// Build a carrying a client user-authentication /// certificate loaded from . /// Used when the remote server's endpoint advertises Certificate-type user tokens. /// Fails fast if the path is missing, the file doesn't exist, or the certificate /// lacks a private key (the private key is required to sign the user-token /// challenge during session activation). /// /// Driver options containing certificate configuration. /// The certificate-backed . internal static UserIdentity BuildCertificateIdentity(OpcUaClientDriverOptions options) => OpcUaUserIdentityFactory.FromCertificate(options); /// Convert a driver to the OPC UA policy URI. /// The driver security policy to map. /// The corresponding OPC UA security policy URI. internal static string MapSecurityPolicy(OpcUaSecurityPolicy policy) => policy switch { OpcUaSecurityPolicy.None => SecurityPolicies.None, OpcUaSecurityPolicy.Basic128Rsa15 => SecurityPolicies.Basic128Rsa15, OpcUaSecurityPolicy.Basic256 => SecurityPolicies.Basic256, OpcUaSecurityPolicy.Basic256Sha256 => SecurityPolicies.Basic256Sha256, OpcUaSecurityPolicy.Aes128_Sha256_RsaOaep => SecurityPolicies.Aes128_Sha256_RsaOaep, OpcUaSecurityPolicy.Aes256_Sha256_RsaPss => SecurityPolicies.Aes256_Sha256_RsaPss, _ => throw new ArgumentOutOfRangeException(nameof(policy), policy, null), }; private static string ShortPolicyName(string policyUri) => policyUri?.Substring(policyUri.LastIndexOf('#') + 1) ?? "(null)"; /// public async Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) { await ShutdownAsync(cancellationToken).ConfigureAwait(false); await InitializeAsync(driverConfigJson, cancellationToken).ConfigureAwait(false); } /// /// Detaches the keep-alive handler and closes + disposes the current session. The session part of /// ; also run at the top of so a re-initialise on a /// live driver does not orphan the previous session. Subscriptions are NOT touched here. Clears the /// outage latch, so the next session reports its own loss. Does nothing when is /// no longer the current lifecycle generation — the session installed then belongs to a NEWER init, which a /// late (abandoned) shutdown must not close. /// /// The lifecycle generation of the caller. /// Bounds the close (on top of ). private async Task ReleaseSessionAsync(long generation, CancellationToken cancellationToken) { // Take the session reference under _probeLock and clear Session to null so any concurrent caller that // checks inside _gate sees null immediately after release begins. ISession? sessionToClose; lock (_probeLock) { if (Volatile.Read(ref _lifecycleGeneration) != generation) return; sessionToClose = Session; if (_keepAliveHandler is not null && sessionToClose is not null) { try { sessionToClose.KeepAlive -= _keepAliveHandler; } catch { } } _keepAliveHandler = null; if (sessionToClose is not null) DetachDiagnostics(sessionToClose); Session = null; } // bound the close — against a dead upstream (the comms-loss watchdog's whole use case) an unbounded // CloseSession would sit out the full operation timeout on every reconnect attempt — and skip it outright // when the keep-alive already says the session is unreachable (review L3). var sessionKnownLost = Volatile.Read(ref _sessionLost) == 1; try { if (sessionToClose is not null && !sessionKnownLost) { using var closeCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); closeCts.CancelAfter(SessionCloseTimeout); await CloseBoundedAsync(sessionToClose, closeCts.Token).ConfigureAwait(false); } } catch { /* best-effort */ } try { sessionToClose?.Dispose(); } catch { } Interlocked.Exchange(ref _sessionLost, 0); Interlocked.Exchange(ref _subscriptionLost, 0); Interlocked.Exchange(ref _publishingStoppedSinceTicks, 0); Interlocked.Exchange(ref _lastSubscriptionsEmptiedTicks, 0); // re-arm the definitive-loss signal for the next session. Interlocked.Exchange(ref _connectionLostRaised, 0); Interlocked.Exchange(ref _closedChannelFailures, 0); // Review of 3b88056f: the released session's post-recovery timers must not outlive it. CancelSessionWork(); } /// Cancel the current session's background work and start a fresh token for the next one. private void CancelSessionWork() { // Cancelled but deliberately NOT disposed (re-review of 30a9d5f0): the keep-alive thread may be reading the old // source's Token concurrently, and a disposed source throws ObjectDisposedException there. Nothing here is // timer-backed (no CancelAfter), so an undisposed source holds no unmanaged resource. var old = Interlocked.Exchange(ref _sessionWorkCts, new CancellationTokenSource()); old.Cancel(); } /// Upper bound on the best-effort wire calls a teardown makes against a possibly-dead upstream. private static readonly TimeSpan SessionCloseTimeout = TimeSpan.FromSeconds(5); /// /// R06:O-9 — the ONE budget the whole of spends on wire calls (every subscription /// delete, concurrently, then the session close). It used to be 5 s per subscription, serially, so a /// half-dead upstream held the actor's PostStop for N×5 s. Under the host's 5 s PostStop budget for /// shutdown + dispose together, so the local teardown still runs inside it. Test seam. /// internal TimeSpan TeardownBudget { get; set; } = TimeSpan.FromSeconds(4); /// Test seam: the wire delete of one subscription. Production calls the SDK's silent delete. internal Func DeleteSubscriptionAsync { get; set; } = static (subscription, ct) => subscription.DeleteAsync(silent: true, ct); /// public async Task ShutdownAsync(CancellationToken cancellationToken) { // A new lifecycle generation: an init still running (abandoned by the host) can no longer install its // session or write health, and keep-alive / data-path health writes of the old session stop landing. var generation = Interlocked.Increment(ref _lifecycleGeneration); // R06:O-9 — one budget for every wire call below, linked to the caller's token. using var budget = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); budget.CancelAfter(TeardownBudget); // r2 R2-1: cancel the session's background work FIRST — a ConditionRefresh sweep holding the subscription gate // against a stalling upstream would otherwise make the gate wait below spend the whole wire budget, leaving the // deletes and the close none. (ReleaseSessionAsync cancels it again; that is idempotent.) CancelSessionWork(); // r2 R2-2: a later DisposeAsync for this same lifecycle skips its nested shutdown. Interlocked.Exchange(ref _shutdownStartedGeneration, generation); // r1 N-2: subscription management is held off for the whole teardown. A subscribe already in flight finishes // first (bounded by the budget) and is enumerated below; one that arrives during the delete phase waits and // then finds the session released, instead of registering a subscription on the closing session after the // registry was enumerated. A wait the budget cut short proceeds without the gate (best effort). var holdsSubscriptionGate = false; try { await _subscriptionGate.WaitAsync(budget.Token).ConfigureAwait(false); holdsSubscriptionGate = true; } catch (OperationCanceledException) { /* budget spent on an in-flight subscribe — tear down regardless */ } catch (ObjectDisposedException) { /* disposed concurrently — tear down regardless */ } try { await ShutdownCoreAsync(generation, budget).ConfigureAwait(false); } finally { if (holdsSubscriptionGate) { try { _subscriptionGate.Release(); } catch (ObjectDisposedException) { } } } } private async Task ShutdownCoreAsync(long generation, CancellationTokenSource budget) { // Tear down remote subscriptions first — otherwise Session.Close will try and may fail // with BadSubscriptionIdInvalid noise in the upstream log. _subscriptions is cleared // whether or not the wire-side delete succeeds since the local handles are useless // after close anyway. Before deleting each subscription we detach the Notification // handlers we attached at subscribe time so the SDK's invocation list no longer // holds the driver instance through the closure. // against a session already known lost (bad keep-alive) the wire deletes cannot succeed and would each // sit out the operation timeout — skip them; disposing the session releases the local objects. var sessionAlive = Volatile.Read(ref _sessionLost) == 0; var deletes = new List(); foreach (var key in _subscriptions.Keys) { if (!_subscriptions.TryRemove(key, out var rs)) continue; DetachNotificationHandlers(rs.ItemHandlers); if (sessionAlive) deletes.Add(DeleteSubscriptionBestEffortAsync(rs.Subscription, budget.Token)); } foreach (var key in _alarmSubscriptions.Keys) { if (!_alarmSubscriptions.TryRemove(key, out var ras)) continue; try { ras.EventItem.Notification -= ras.Handler; } catch { /* best-effort */ } if (sessionAlive) deletes.Add(DeleteSubscriptionBestEffortAsync(ras.Subscription, budget.Token)); } // Concurrently, under the one budget (each delete already swallows its own failure). await Task.WhenAll(deletes).ConfigureAwait(false); // The host re-subscribes after a reinit and the new subscriptions deliver initial values; nothing of the // previous connection's value bookkeeping applies to them. _lastValues.Clear(); _lastGoodSourceTicks.Clear(); lock (_replayGuardLock) { _replayGuard.Clear(); Interlocked.Exchange(ref _replayGuardUntilTicks, 0); _replayDrops = 0; _replayDropRefs.Clear(); } _awaitingRecovery.Clear(); // the last-told alarm states (and ack targets) deliberately SURVIVE a shutdown. A // ReinitializeAsync (comms-loss watchdog / Force Reconnect) is followed by the host re-subscribing the // same set over condition nodes it did NOT rebuild; forgetting here would make the post-subscribe // ConditionRefresh re-raise every active+acked alarm (an un-ack flicker). BeginAlarmSubscription forgets // them when the set changes, which is when the host did rebuild. // The session close spends what is left of the same budget (none left ⇒ dispose without closing). await ReleaseSessionAsync(generation, budget.Token).ConfigureAwait(false); lock (_probeLock) { // A newer initialise may have installed its own map / endpoint meanwhile — leave those alone. if (Volatile.Read(ref _lifecycleGeneration) == generation) { _namespaceMap = null; _connectedEndpointUrl = null; // Detach the cert-validation handler so the (potentially process-shared) CertificateValidator doesn't // hold a delegate to a shutting-down driver — generation-checked like the map (r1 N-3): a newer init // that installed its own handler meanwhile keeps it. DetachCertValidation(_certValidatorRef, _certValidationHandler); _certValidationHandler = null; _certValidatorRef = null; } } if (Volatile.Read(ref _lifecycleGeneration) == generation) TransitionTo(HostState.Unknown); WriteLifecycleHealth(generation, h => new DriverHealth(DriverState.Unknown, h.LastSuccessfulRead, null)); } /// /// Best-effort delete of one subscription (shutdown path), bounded by the shutdown's budget: the wait is /// abandoned when the budget fires even if the SDK call ignores the token. /// private async Task DeleteSubscriptionBestEffortAsync(Subscription subscription, CancellationToken budget) { try { await DeleteSubscriptionAsync(subscription, budget).WaitAsync(budget).ConfigureAwait(false); } catch { /* best-effort */ } } /// public DriverHealth GetHealth() => _health; // ---- IReadable ---- /// public async Task> ReadAsync( IReadOnlyList fullReferences, CancellationToken cancellationToken) { // Make sure a session exists before queuing on the gate, but do NOT bind the wire // call to this reference — a reconnect can swap Session while we wait on _gate. The // session actually used is re-read inside the gate. _ = RequireSession(); var (results, _) = await ReadCoreAsync(fullReferences, cancellationToken).ConfigureAwait(false); return results; } /// /// The read itself, also used by the in-place recovery. TransportOk is false when the wire read /// did not happen or threw (every result is then a local BadCommunicationError, not upstream data). /// , when given, must still be the current session inside the gate — a /// recovery overtaken by a re-initialise neither reads on, nor writes the health of, the new session. /// private async Task<(DataValueSnapshot[] Results, bool TransportOk)> ReadCoreAsync( IReadOnlyList fullReferences, CancellationToken cancellationToken, ISession? expectedSession = null) { var results = new DataValueSnapshot[fullReferences.Count]; var now = Clock.GetUtcNow().UtcDateTime; await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); try { // Re-read Session inside the critical section: if a reconnect completed while we // were blocked on _gate, a re-initialise has already swapped in the new // session. NodeId parsing is namespace-relative, so it must also use the current // session's namespace table. The generation is read first: a shutdown / re-initialise that releases this // session while the read is in flight bumps it, so the read's health outcome never lands on the next lifecycle. var generation = Volatile.Read(ref _lifecycleGeneration); var session = Session; if (session is null || (expectedSession is not null && !ReferenceEquals(session, expectedSession))) { for (var i = 0; i < fullReferences.Count; i++) results[i] = new DataValueSnapshot(null, StatusBadCommunicationError, null, now); return (results, false); } // Resolve each RawPath reference (v3) to a live NodeId against the current session. // A RawPath the authored RawTags table doesn't know — or a node id that won't parse — // gets BadNodeIdInvalid and is omitted from the wire request, saving a round-trip for a // fault the driver can detect locally. var toSend = new ReadValueIdCollection(); var indexMap = new List(fullReferences.Count); // maps wire-index -> results-index for (var i = 0; i < fullReferences.Count; i++) { if (!TryResolveTagNode(session, fullReferences[i], out var nodeId)) { results[i] = new DataValueSnapshot(null, StatusBadNodeIdInvalid, null, now); continue; } toSend.Add(new ReadValueId { NodeId = nodeId, AttributeId = Attributes.Value }); indexMap.Add(i); } if (toSend.Count == 0) return (results, true); try { var resp = await session.ReadAsync( requestHeader: null, maxAge: 0, timestampsToReturn: TimestampsToReturn.Both, nodesToRead: toSend, ct: cancellationToken).ConfigureAwait(false); var values = resp.Results; for (var w = 0; w < values.Count; w++) { var r = indexMap[w]; var dv = values[w]; // Preserve the upstream StatusCode verbatim — including Bad codes per // §8's cascading-quality rule. Also preserve SourceTimestamp so downstream // clients can detect stale upstream data. results[r] = new DataValueSnapshot( Value: dv.Value, StatusCode: dv.StatusCode.Code, SourceTimestampUtc: dv.SourceTimestamp == DateTime.MinValue ? null : dv.SourceTimestamp, ServerTimestampUtc: dv.ServerTimestamp == DateTime.MinValue ? now : dv.ServerTimestamp); } RecordDeviceAnswered(generation, now); } catch (Exception ex) { // Transport / timeout / session-dropped — fan out the same fault across every // tag in this batch. Per-tag StatusCode stays BadCommunicationError (not // BadInternalError) so operators distinguish "upstream unreachable" from // "driver bug". for (var w = 0; w < indexMap.Count; w++) { var r = indexMap[w]; results[r] = new DataValueSnapshot(null, StatusBadCommunicationError, null, now); } RecordCommsFault(generation, ex.Message); return (results, false); } } finally { _gate.Release(); } return (results, true); } // ---- IWritable ---- /// public async Task> WriteAsync( IReadOnlyList writes, CancellationToken cancellationToken) { // R06:O-6 — writes take NO gate: a write stalled for the operation timeout must not hold up every read // and subscription create behind it (the host already bounds write concurrency and serialises per tag; // the SDK session is safe for concurrent service calls). The session is read once, here. _ = RequireSession(); var results = new WriteResult[writes.Count]; var generation = Volatile.Read(ref _lifecycleGeneration); var session = Session; if (session is null) { // Writes are non-idempotent — but here the request never reached the wire, so // BadCommunicationError ("definitely did not happen") is the honest code. for (var i = 0; i < writes.Count; i++) results[i] = new WriteResult(StatusBadCommunicationError); return results; } var toSend = new WriteValueCollection(); var indexMap = new List(writes.Count); for (var i = 0; i < writes.Count; i++) { if (!TryResolveTagNode(session, writes[i].FullReference, out var nodeId)) { results[i] = new WriteResult(StatusBadNodeIdInvalid); continue; } toSend.Add(new WriteValue { NodeId = nodeId, AttributeId = Attributes.Value, Value = new DataValue(new Variant(writes[i].Value)), }); indexMap.Add(i); } if (toSend.Count == 0) return results; try { var resp = await session.WriteAsync( requestHeader: null, nodesToWrite: toSend, ct: cancellationToken).ConfigureAwait(false); var codes = resp.Results; for (var w = 0; w < codes.Count; w++) { var r = indexMap[w]; // Pass upstream WriteResult StatusCode through verbatim. Success codes // include Good (0) and any warning-level Good* variants; anything with // the severity bits set is a Bad. results[r] = new WriteResult(codes[w].Code); } // A response — refusals included (BadNotWritable, BadTypeMismatch, ...) — is the server answering. RecordDeviceAnswered(generation, Clock.GetUtcNow().UtcDateTime); } catch (OperationCanceledException ex) { // Timeout / cancellation after the wire request may have been dispatched. // Writes are non-idempotent — BadTimeout ("outcome unknown, do not blindly // retry") is more honest than BadCommunicationError ("definitely did not // happen"). Downstream callers that need retry semantics check for BadTimeout // and can decide whether to re-issue. const uint StatusBadTimeout = 0x800A0000u; for (var w = 0; w < indexMap.Count; w++) results[indexMap[w]] = new WriteResult(StatusBadTimeout); // No reply arrived. Only a timeout is a comms fault — a caller that cancelled says nothing about upstream. if (!cancellationToken.IsCancellationRequested) RecordCommsFault(generation, $"write: {ex.Message}"); } catch (Exception ex) { // Pre-wire transport failure — the write definitely did not reach the server. for (var w = 0; w < indexMap.Count; w++) results[indexMap[w]] = new WriteResult(StatusBadCommunicationError); RecordCommsFault(generation, $"write: {ex.Message}"); } return results; } /// /// (v3) Resolve a tag's RawPath reference to a live against the /// 's current namespace table. Two stages: /// /// Map the RawPath (the wire reference handed to the driver under v3) to its /// upstream node id string via the authored RawTags table on /// (). /// Re-bind that node id string — a server-stable nsu=<uri>;… form or a plain /// ns=2;s=… serialized NodeId — against the current session so a remote /// namespace-table reorder across a restart is transparently corrected (driver-specs.md §8). /// /// An unknown/un-authored RawPath, a missing map, or a node id that won't parse returns false; /// the driver surfaces that as without a wire round-trip. /// /// The OPC UA session to resolve the node ID against. /// The tag's RawPath reference to resolve. /// The resolved live node ID when successful. /// true if the RawPath resolved to a live node ID; otherwise false. internal bool TryResolveTagNode(ISession session, string rawPath, out NodeId nodeId) { nodeId = NodeId.Null; var map = _namespaceMap; if (map is null) return false; // Stage 1: RawPath -> upstream node id string (authored as TagConfig.nodeId). if (!map.TryResolve(rawPath, out var upstreamNodeId)) return false; // Stage 2: node id string -> live NodeId re-bound against the current session table. return NamespaceMap.TryResolve(session, upstreamNodeId, out nodeId); } /// /// Render a discovered NodeId in the server-stable form persisted into the local /// address space. Falls back to the raw serialized NodeId if the namespace map is not /// yet built (it always is by the time runs). /// private string StableReference(NodeId nodeId) => _namespaceMap?.ToStableReference(nodeId) ?? nodeId.ToString() ?? string.Empty; private ISession RequireSession() => Session ?? throw new InvalidOperationException("OpcUaClientDriver not initialized"); // ---- ITagDiscovery ---- /// // Run-once: DiscoverAsync recursively browses the remote server's address space and // registers every variable within the single call (browse + enrich passes are fully // awaited) — nothing fills in asynchronously after connect, so a single discovery pass // is sufficient. public DiscoveryRediscoverPolicy RediscoverPolicy => DiscoveryRediscoverPolicy.Once; /// public async Task DiscoverAsync(IAddressSpaceBuilder builder, CancellationToken cancellationToken) { ArgumentNullException.ThrowIfNull(builder); // R06:O-6 — no gate across the recursive browse: it can run for as long as the remote tree takes to walk, // and holding the read gate for that long would stall the data plane behind it. The browse reads the session // once and uses it throughout (the SDK session is safe for concurrent service calls). var session = Session ?? throw new InvalidOperationException("OpcUaClientDriver not initialized"); var rootFolder = builder.Folder("Remote", "Remote"); var visited = new HashSet(); var discovered = 0; var pendingVariables = new List(); var root = !string.IsNullOrEmpty(_options.BrowseRoot) ? NodeId.Parse(session.MessageContext, _options.BrowseRoot) : ObjectIds.ObjectsFolder; // Pass 1: browse hierarchy + create folders inline, collect variables into a // pending list. Defers variable registration until attributes are resolved — the // address-space builder's Variable call is the one-way commit, so doing it only // once per variable (with correct DataType/SecurityClass/IsArray) avoids the // alternative (register with placeholders + mutate later) which the // IAddressSpaceBuilder contract doesn't expose. await BrowseRecursiveAsync(session, root, rootFolder, visited, depth: 0, discovered: () => discovered, increment: () => discovered++, pendingVariables: pendingVariables, ct: cancellationToken).ConfigureAwait(false); // Pass 2: batch-read DataType + AccessLevel + ValueRank + Historizing per // variable. One wire request for up to ~N variables; for 10k-node servers this is // still a couple of hundred ms total since the SDK chunks ReadAsync automatically. await EnrichAndRegisterVariablesAsync(session, pendingVariables, cancellationToken) .ConfigureAwait(false); } /// /// A variable collected during the browse pass, waiting for attribute enrichment /// before being registered on the address-space builder. /// private readonly record struct PendingVariable( IAddressSpaceBuilder ParentFolder, string BrowseName, string DisplayName, NodeId NodeId); private async Task BrowseRecursiveAsync( ISession session, NodeId node, IAddressSpaceBuilder folder, HashSet visited, int depth, Func discovered, Action increment, List pendingVariables, CancellationToken ct) { if (depth >= _options.MaxBrowseDepth) return; if (discovered() >= _options.MaxDiscoveredNodes) return; if (!visited.Add(node)) return; var browseDescriptions = new BrowseDescriptionCollection { new() { NodeId = node, BrowseDirection = BrowseDirection.Forward, ReferenceTypeId = ReferenceTypeIds.HierarchicalReferences, IncludeSubtypes = true, NodeClassMask = (uint)(NodeClass.Object | NodeClass.Variable), ResultMask = (uint)(BrowseResultMask.BrowseName | BrowseResultMask.DisplayName | BrowseResultMask.NodeClass | BrowseResultMask.TypeDefinition), } }; ReferenceDescriptionCollection refs; try { var resp = await session.BrowseAsync( requestHeader: null, view: null, requestedMaxReferencesPerNode: 0, nodesToBrowse: browseDescriptions, ct: ct).ConfigureAwait(false); if (resp.Results.Count == 0) return; var result = resp.Results[0]; refs = result.References; // Follow browse continuation points. OPC UA servers cap the references returned // per node in a single response; when a folder has more children than the cap, // BrowseResult.ContinuationPoint is non-empty and the remainder must be pulled // with BrowseNext. Without this loop a large remote folder is silently truncated // and discovered tags go missing from the local address space. var continuationPoint = result.ContinuationPoint; while (continuationPoint is { Length: > 0 }) { BrowseNextResponse next; try { next = await session.BrowseNextAsync( requestHeader: null, releaseContinuationPoints: false, continuationPoints: [continuationPoint], ct: ct).ConfigureAwait(false); } catch (OperationCanceledException) { // Caller cancelled mid-pagination. Release the server-side cursor so it // doesn't hold resources until session close. Use CancellationToken.None // for the release — the original token is already cancelled so we must not // gate the release on it. Fire-and-forget: if the server is also unreachable, // the release fails silently and the cursor times out server-side anyway. try { await session.BrowseNextAsync( requestHeader: null, releaseContinuationPoints: true, continuationPoints: [continuationPoint], ct: CancellationToken.None).ConfigureAwait(false); } catch { /* best-effort — server may have already cleaned up */ } throw; } if (next.Results.Count == 0) break; var nextResult = next.Results[0]; if (nextResult.References is { Count: > 0 }) refs.AddRange(nextResult.References); continuationPoint = nextResult.ContinuationPoint; } } catch (OperationCanceledException) { // Propagate cancellation — don't silently swallow it like a transient browse // failure. The caller's CancellationToken was honoured; let it observe that. throw; } catch { // Transient browse failure on a sub-tree — don't kill the whole discovery, just // skip this branch. The driver's health surface will reflect the cascade via the // probe loop (PR 69). return; } foreach (var rf in refs) { if (discovered() >= _options.MaxDiscoveredNodes) break; var childId = ExpandedNodeId.ToNodeId(rf.NodeId, session.NamespaceUris); if (NodeId.IsNull(childId)) continue; var browseName = rf.BrowseName?.Name ?? childId.ToString(); var displayName = rf.DisplayName?.Text ?? browseName; if (rf.NodeClass == NodeClass.Object) { var subFolder = folder.Folder(browseName, displayName); increment(); await BrowseRecursiveAsync(session, childId, subFolder, visited, depth + 1, discovered, increment, pendingVariables, ct).ConfigureAwait(false); } else if (rf.NodeClass == NodeClass.Variable) { pendingVariables.Add(new PendingVariable(folder, browseName, displayName, childId)); increment(); } } } /// /// Pass 2 of discovery: batch-read DataType + ValueRank + AccessLevel + Historizing /// for every collected variable in one Session.ReadAsync (the SDK chunks internally /// to respect the server's per-request limits). Then register each variable on its /// parent folder with the real . /// /// /// /// Attributes read: DataType (NodeId of the value type), /// ValueRank (-1 = scalar, 1 = array), UserAccessLevel (the /// effective access mask for our session — more accurate than AccessLevel which /// is the server-side configured mask before user filtering), and /// Historizing (server flags whether historian data is available). /// /// /// When the upstream server returns Bad on any attribute, the variable falls back /// to safe defaults (Int32 / ViewOnly / not-array / not-historized) and is still /// registered — a partial enrichment failure shouldn't drop entire variables from /// the address space. Operators reading the Admin dashboard see the variable /// with conservative metadata which is obviously wrong and easy to triage. /// /// private async Task EnrichAndRegisterVariablesAsync( ISession session, IReadOnlyList pending, CancellationToken ct) { if (pending.Count == 0) return; // 4 attributes per variable: DataType, ValueRank, UserAccessLevel, Historizing. var nodesToRead = new ReadValueIdCollection(pending.Count * 4); foreach (var pv in pending) { nodesToRead.Add(new ReadValueId { NodeId = pv.NodeId, AttributeId = Attributes.DataType }); nodesToRead.Add(new ReadValueId { NodeId = pv.NodeId, AttributeId = Attributes.ValueRank }); nodesToRead.Add(new ReadValueId { NodeId = pv.NodeId, AttributeId = Attributes.UserAccessLevel }); nodesToRead.Add(new ReadValueId { NodeId = pv.NodeId, AttributeId = Attributes.Historizing }); } DataValueCollection values; try { var resp = await session.ReadAsync( requestHeader: null, maxAge: 0, timestampsToReturn: TimestampsToReturn.Neither, nodesToRead: nodesToRead, ct: ct).ConfigureAwait(false); values = resp.Results; } catch { // Enrichment-read failed wholesale (server unreachable mid-browse). Register the // pending variables with conservative defaults rather than dropping them — the // downstream catalog is still useful for reading via IReadable. foreach (var pv in pending) RegisterFallback(pv); return; } for (var i = 0; i < pending.Count; i++) { var pv = pending[i]; var baseIdx = i * 4; var dataTypeDv = values[baseIdx]; var valueRankDv = values[baseIdx + 1]; var accessDv = values[baseIdx + 2]; var histDv = values[baseIdx + 3]; var dataType = StatusCode.IsGood(dataTypeDv.StatusCode) && dataTypeDv.Value is NodeId dtId ? MapUpstreamDataType(dtId) : DriverDataType.Int32; var valueRank = StatusCode.IsGood(valueRankDv.StatusCode) && valueRankDv.Value is int vr ? vr : -1; // OPC UA Part 3 ValueRank constants: -3 = ScalarOrOneDimension, -2 = Any, // -1 = Scalar, 0 = OneOrMoreDimensions, 1 = OneDimension, >1 = N specific dimensions. // Deliberate choice: treat anything >= 0 as an array (the spec guarantees -3/-2/-1 // are the only negative values, and any non-negative rank denotes at least one // array dimension). -3 ScalarOrOneDimension and -2 Any are conservatively treated // as scalar — array-of-one is exposed as scalar to the local address space until // the upstream variable carries a concrete dimensioned rank. var isArray = UpstreamAttributeMapping.IsArray(valueRank); var access = StatusCode.IsGood(accessDv.StatusCode) && accessDv.Value is byte ab ? ab : (byte)0; var securityClass = MapAccessLevelToSecurityClass(access); var historizing = StatusCode.IsGood(histDv.StatusCode) && histDv.Value is bool b && b; pv.ParentFolder.Variable(pv.BrowseName, pv.DisplayName, new DriverAttributeInfo( FullName: StableReference(pv.NodeId), DriverDataType: dataType, IsArray: isArray, ArrayDim: null, SecurityClass: securityClass, IsHistorized: historizing, IsAlarm: false)); } void RegisterFallback(PendingVariable pv) { pv.ParentFolder.Variable(pv.BrowseName, pv.DisplayName, new DriverAttributeInfo( FullName: StableReference(pv.NodeId), DriverDataType: DriverDataType.Int32, IsArray: false, ArrayDim: null, SecurityClass: SecurityClassification.ViewOnly, IsHistorized: false, IsAlarm: false)); } } /// /// Map an upstream OPC UA built-in DataType NodeId (via DataTypeIds.*) to a /// . Unknown / custom types fall through to /// which is the safest passthrough for /// Variant-wrapped structs + enums + extension objects; downstream clients see a /// string rendering but the cascading-quality path still preserves upstream /// StatusCode + timestamps. /// /// The OPC UA data type NodeId to map. /// The mapped . internal static DriverDataType MapUpstreamDataType(NodeId dataType) => UpstreamAttributeMapping.MapDataType(dataType); /// /// Map an OPC UA AccessLevel/UserAccessLevel attribute value (AccessLevels /// bitmask) to a the local node-manager's ACL /// layer can gate writes off. CurrentWrite-capable variables surface as /// ; read-only as . /// /// The OPC UA access level bitmask. /// The mapped security classification. internal static SecurityClassification MapAccessLevelToSecurityClass(byte accessLevel) => UpstreamAttributeMapping.MapAccessLevel(accessLevel); // ---- ISubscribable ---- /// public async Task SubscribeAsync( IReadOnlyList fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken) { // Fast-fail before queuing if the driver isn't connected, but do NOT bind the // subscription to this reference — a reconnect can swap Session while we wait on _gate. // The session actually used is re-read inside the gate. _ = RequireSession(); var id = Interlocked.Increment(ref _nextSubscriptionId); var handle = new OpcUaSubscriptionHandle(id); // Floor the publishing interval at 50ms — OPC UA servers routinely negotiate // minimum-supported intervals up anyway, but sending sub-50ms wastes negotiation // bandwidth on every subscription create. the options may pin the interval. var intervalMs = ResolvePublishingIntervalMs(_options, publishingInterval); var subscription = new Subscription(telemetry: null!, new SubscriptionOptions { DisplayName = $"opcua-sub-{id}", PublishingInterval = intervalMs, KeepAliveCount = _options.SubscriptionKeepAliveCount, LifetimeCount = _options.SubscriptionLifetimeCount, MaxNotificationsPerPublish = _options.SubscriptionMaxNotificationsPerPublish, PublishingEnabled = true, Priority = _options.SubscriptionPriority, TimestampsToReturn = TimestampsToReturn.Both, }); long subscribeGeneration = 0; await _subscriptionGate.WaitAsync(cancellationToken).ConfigureAwait(false); try { // Re-read Session inside the critical section: a reconnect completing while we were // blocked on _gate has already swapped in the new session via a re-initialise. subscribeGeneration = Volatile.Read(ref _lifecycleGeneration); var session = Session ?? throw new InvalidOperationException( "OpcUaClient session was lost before the subscription could be created."); session.AddSubscription(subscription); // Track each (MonitoredItem, handler) pair so UnsubscribeAsync / ShutdownAsync // can detach the Notification delegate before disposing the session. The lambda // captures `handle`, so we must hold the exact delegate instance returned by // `+=` to be able to remove it. var itemHandlers = new List(); try { await CreateDataItemsAsync(session, itemHandlers).ConfigureAwait(false); } catch { // R06:O-1 — the host's reconcile retries a failed subscribe every health tick, so a half-built // subscription left on the session (created server-side when CreateAsync succeeded, its item // handlers still attached, tracked by nothing) would accumulate one live, publishing subscription // per retry. Mirror the alarm path: detach, bounded remove, rethrow. DetachNotificationHandlers(itemHandlers); try { using var removeCts = new CancellationTokenSource(SessionCloseTimeout); await session.RemoveSubscriptionAsync(subscription, removeCts.Token).ConfigureAwait(false); } catch { /* best-effort — the session may be the thing that failed */ } // a half-built subscription can have been the server's only one, so its delete opens the same window. NoteSubscriptionDeleted(); throw; } _subscriptions[id] = new RemoteSubscription(subscription, handle, itemHandlers); } finally { _subscriptionGate.Release(); } return handle; async Task CreateDataItemsAsync(ISession session, List itemHandlers) { await subscription.CreateAsync(cancellationToken).ConfigureAwait(false); foreach (var fullRef in fullReferences) { if (!TryResolveTagNode(session, fullRef, out var nodeId)) continue; // The RawPath (v3 tag identity) is routed through MonitoredItem.Handle so the // Notification handler reports the change under the same reference the runtime // subscribed with — StartNodeId is the resolved upstream node, Handle stays the RawPath. var item = new MonitoredItem(telemetry: null!, new MonitoredItemOptions { DisplayName = fullRef, StartNodeId = nodeId, AttributeId = Attributes.Value, MonitoringMode = MonitoringMode.Reporting, SamplingInterval = intervalMs, QueueSize = 1, DiscardOldest = true, }) { Handle = fullRef, }; MonitoredItemNotificationEventHandler notifHandler = (mi, args) => OnMonitoredItemNotification(handle, mi, args); item.Notification += notifHandler; itemHandlers.Add(new MonitoredItemNotificationHandle(item, notifHandler)); subscription.AddItem(item); } await subscription.CreateItemsAsync(cancellationToken).ConfigureAwait(false); ThrowIfSubscribeSuperseded(subscribeGeneration); } } /// /// Review m10: a subscribe that a shutdown / re-initialise overtook must not register its subscription — the /// teardown has already enumerated the registry, so the entry would outlive it and the next session's stall /// check would declare a dead subscription lost. Thrown inside the create's try, so the half-built /// subscription is removed like any other failed create. /// private void ThrowIfSubscribeSuperseded(long subscribeGeneration) { if (Volatile.Read(ref _lifecycleGeneration) != subscribeGeneration) throw new InvalidOperationException( "OpcUaClient session was released (shutdown or re-initialise) while the subscription was being created."); } /// public async Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) { if (handle is not OpcUaSubscriptionHandle h) return; if (!_subscriptions.TryRemove(h.Id, out var rs)) return; var emptied = NoteSubscriptionDeleted(); await _subscriptionGate.WaitAsync(cancellationToken).ConfigureAwait(false); try { // Detach Notification handlers BEFORE deleting the subscription so the SDK's // MonitoredItem.Notification multicast invocation list no longer holds a // closure that captures the driver instance. The delegate stored on // RemoteSubscription is the exact instance that was added, so `-=` removes it // cleanly. DetachNotificationHandlers(rs.ItemHandlers); try { await rs.Subscription.DeleteAsync(silent: true, cancellationToken).ConfigureAwait(false); } catch { /* best-effort — the subscription may already be gone on reconnect */ } } finally { // the server answers the stale publish only once the delete is done, so the window runs from here. if (emptied) NoteSubscriptionDeleted(force: true); _subscriptionGate.Release(); } } private static void DetachNotificationHandlers(IReadOnlyList items) { for (var i = 0; i < items.Count; i++) { var pair = items[i]; try { pair.Item.Notification -= pair.Handler; } catch { /* best-effort — SDK may have already cleared its invocation list on session loss */ } } } private void OnMonitoredItemNotification(OpcUaSubscriptionHandle handle, MonitoredItem item, MonitoredItemNotificationEventArgs args) { // args.NotificationValue arrives as a MonitoredItemNotification for value-change // subscriptions; extract its DataValue. The Handle property carries our tag string. if (args.NotificationValue is not MonitoredItemNotification mn) return; var dv = mn.Value; if (dv is null) return; PublishNotification(handle, RefOf(item), dv); } private static string RefOf(MonitoredItem item) => (item.Handle as string) ?? item.DisplayName ?? string.Empty; private void PublishNotification(OpcUaSubscriptionHandle handle, string fullRef, DataValue dv) { _diag.Record(OpcUaClientDiagnostics.DataChange); if (StatusCode.IsBad(dv.StatusCode)) _diag.Record(OpcUaClientDiagnostics.BadStatusValue); // a delivered notification is liveness evidence — the subscription path never calls ReadAsync, so // without this a busy subscription-only driver's LastSuccessfulRead stayed at its connect time. RecordLivenessEvidence(); var snapshot = new DataValueSnapshot( Value: dv.Value, StatusCode: dv.StatusCode.Code, SourceTimestampUtc: dv.SourceTimestamp == DateTime.MinValue ? null : dv.SourceTimestamp, ServerTimestampUtc: dv.ServerTimestamp == DateTime.MinValue ? DateTime.UtcNow : dv.ServerTimestamp); // Review of 3b88056f: the replay check and the publish are one critical section shared with the recovery and // trailing reads, so an older value can never be delivered after a newer one (see _deliveryLock). lock (_deliveryLock) { // F2: inside the post-recovery replay window only, a late OLDER notification must not regress the node. if (IsLateReplay(fullRef, snapshot.SourceTimestampUtc, out var newest)) { _diag.Record(OpcUaClientDiagnostics.StaleNotificationDropped); // the OTel counter and the Debug line below are invisible on a default-configured node; the // per-recovery Information summary logged at disarm is what makes the guard observable. lock (_replayGuardLock) { _replayDrops++; _replayDropRefs.Add(fullRef); } _logger.LogDebug( "OpcUaClientDriver '{DriverInstanceId}': dropped a late replayed notification for {Ref} after a recovery " + "(source {Source:O} is not newer than {Newest:O}).", _driverInstanceId, fullRef, snapshot.SourceTimestampUtc, new DateTime(newest, DateTimeKind.Utc)); return; } RecordUpstreamGood(fullRef, snapshot); _lastValues[fullRef] = snapshot; _deliverySeq.AddOrUpdate(fullRef, 1, (_, n) => n + 1); _awaitingRecovery.TryRemove(fullRef, out _); OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, fullRef, snapshot)); } } /// Remember the last accepted Good upstream source time for . private void RecordUpstreamGood(string fullRef, DataValueSnapshot snapshot) { if (snapshot.SourceTimestampUtc is not { } src || !StatusCode.IsGood(snapshot.StatusCode)) return; _lastGoodSourceTicks[fullRef] = src.Ticks; } /// /// true when the replay window is open and is at or below /// 's high-water mark. Otherwise (inside the window) the notification is accepted and /// its source time becomes the ref's new mark, so the ref stays guarded for the whole window. A notification /// with no source time passes without moving the mark. An expired window disarms everything. /// private bool IsLateReplay(string fullRef, DateTime? source, out long newest) { newest = 0; var until = Interlocked.Read(ref _replayGuardUntilTicks); if (until == 0) return false; if (Clock.GetUtcNow().UtcDateTime.Ticks >= until) { DisarmReplayGuard(until); return false; } if (source is not { } src) return false; var ticks = src.Ticks; while (true) { if (_replayGuard.TryGetValue(fullRef, out newest)) { if (ticks <= newest) return true; if (_replayGuard.TryUpdate(fullRef, ticks, newest)) return false; } else if (_replayGuard.TryAdd(fullRef, ticks)) return false; } } /// /// Expire the replay guard armed until — a no-op when it was re-armed or already disarmed /// since — and log ONE Information summary of what it dropped. The guard's OTel counter and per-drop Debug /// line are invisible on a default-configured node, so before this a recovery's drops left no trace at all. /// /// The expiry the caller observed; only that arming is disarmed. private void DisarmReplayGuard(long until) { int drops, refs; lock (_replayGuardLock) { if (Interlocked.Read(ref _replayGuardUntilTicks) != until) return; Interlocked.Exchange(ref _replayGuardUntilTicks, 0); _replayGuard.Clear(); drops = _replayDrops; refs = _replayDropRefs.Count; _replayDrops = 0; _replayDropRefs.Clear(); } LogReplaySummary(drops, refs); } /// One line per replay window: Information when the guard dropped something, Debug otherwise. private void LogReplaySummary(int drops, int refs) => _logger.Log(drops > 0 ? LogLevel.Information : LogLevel.Debug, "OpcUaClientDriver '{DriverInstanceId}': post-recovery replay window closed — dropped {Drops} late " + "notifications across {Refs} refs after recovery.", _driverInstanceId, drops, refs); /// /// arm the replay guard on a same-session recovery: every ref with a known Good upstream source time, /// for max(30 s, 2 × the longest tracked subscription keep-alive window). /// /// The window's expiry in clock ticks. private long ArmReplayGuard(DateTime now) { var window = ReplayGuardMinimum; foreach (var sub in _subscriptions.Values.Select(r => r.Subscription)) { var keepAliveWindow = TimeSpan.FromMilliseconds(sub.CurrentPublishingInterval * (double)sub.CurrentKeepAliveCount); if (keepAliveWindow * 2 > window) window = keepAliveWindow * 2; } int supersededDrops, supersededRefs; lock (_replayGuardLock) { // A re-arm inside an open window ends that window: report what it dropped and count afresh. supersededDrops = _replayDrops; supersededRefs = _replayDropRefs.Count; _replayDrops = 0; _replayDropRefs.Clear(); _replayGuard.Clear(); foreach (var kv in _lastGoodSourceTicks) _replayGuard[kv.Key] = kv.Value; Interlocked.Exchange(ref _replayGuardUntilTicks, (now + window).Ticks); } if (supersededDrops > 0) LogReplaySummary(supersededDrops, supersededRefs); return (now + window).Ticks; } /// /// disarm the replay guard (and log its summary) when its window ends, even if no notification arrives to /// disarm it lazily in — a quiet subscription would otherwise never log the summary. /// private async Task DisarmReplayGuardWhenDueAsync(long until, CancellationToken ct) { try { var wait = new DateTime(until, DateTimeKind.Utc) - Clock.GetUtcNow().UtcDateTime; if (wait > TimeSpan.Zero) await Task.Delay(wait + TimeSpan.FromMilliseconds(100), Clock, ct).ConfigureAwait(false); if (Clock.GetUtcNow().UtcDateTime.Ticks >= until) DisarmReplayGuard(until); } catch (OperationCanceledException) when (ct.IsCancellationRequested) { /* session released */ } catch (Exception ex) { _logger.LogDebug(ex, "OpcUaClientDriver '{DriverInstanceId}': replay-guard disarm timer faulted.", _driverInstanceId); } } /// /// advance on upstream traffic (a notification), at most /// once per second. The state is left as it is: a notification is evidence of comms, but the keep-alive /// is what declares an outage over. /// private void RecordLivenessEvidence() { // A dead subscription must keep the stamp frozen even while OTHER subscriptions still deliver. if (Volatile.Read(ref _subscriptionLost) == 1) return; var now = Clock.GetUtcNow().UtcDateTime; var last = Interlocked.Read(ref _lastEvidenceStampTicks); if (now.Ticks - last < TimeSpan.TicksPerSecond && last != 0) return; Interlocked.Exchange(ref _lastEvidenceStampTicks, now.Ticks); UpdateHealth(h => h.State is DriverState.Healthy or DriverState.Degraded ? h with { LastSuccessfulRead = now } : null); } /// /// A service call got a RESPONSE from the upstream server (whatever per-node statuses it carried — a refusal /// is still the server answering): health Healthy, advanced. Not /// while a tracked subscription is known dead — that latch is what lets the comms-loss watchdog re-initialise /// a driver whose session answers but whose data plane is gone. /// /// The lifecycle generation the call ran under. /// When the response arrived. private void RecordDeviceAnswered(long generation, DateTime now) { if (Volatile.Read(ref _subscriptionLost) == 1) return; // Review m7: during a published outage (bad keep-alive) a reply that was in flight is evidence — the stamp // advances — but it does not declare the outage over; the keep-alive recovery path does that (it also // re-publishes the refs it published Bad), so Healthy is only written once the outage latch is clear. if (Volatile.Read(ref _sessionLost) == 1) { UpdateHealth(generation, h => h.State is DriverState.Healthy or DriverState.Degraded ? h with { LastSuccessfulRead = now } : null); return; } UpdateHealth(generation, _ => new DriverHealth(DriverState.Healthy, now, null)); } /// /// A service call got NO response (transport failure / timeout): health Degraded with /// preserved, so the comms-loss watchdog measures the silence. /// /// The lifecycle generation the call ran under. /// The failure, for . private void RecordCommsFault(long generation, string error) => UpdateHealth(generation, h => new DriverHealth(DriverState.Degraded, h.LastSuccessfulRead, error)); /// /// Read-modify-write of as a CAS loop: the keep-alive timer, publish /// threads and the recovery task all derive the next health from the current one. /// /// Maps the current health to the next one, or null to leave it unchanged. private void UpdateHealth(Func next) { while (true) { var current = Volatile.Read(ref _health); var proposed = next(current); if (proposed is null) return; if (ReferenceEquals(Interlocked.CompareExchange(ref _health, proposed, current), current)) return; } } /// /// stamped with a lifecycle generation: the /// write lands only while is still current, re-checked on every CAS attempt. Every /// lifecycle transition bumps the generation BEFORE it writes its own health, so a superseded writer (a late /// keep-alive of a released session, a read that outlived a shutdown, an abandoned init) either loses the /// generation check or is overwritten by the transition's own later write. /// /// The lifecycle generation the writer belongs to. /// Maps the current health to the next one, or null to leave it unchanged. private void UpdateHealth(long generation, Func next) => UpdateHealth(h => Volatile.Read(ref _lifecycleGeneration) == generation ? next(h) : null); /// /// Bumped by every and . Session install, the /// namespace-map / endpoint publish and every health write carry the generation they were started under. /// private long _lifecycleGeneration; /// The lifecycle generation the last started under (r2 R2-2). private long _shutdownStartedGeneration = -1; /// Test seam: the current lifecycle generation. internal long LifecycleGenerationForTest => Volatile.Read(ref _lifecycleGeneration); /// /// Live data-change subscription bookkeeping. Holds the SDK , /// the local handle, and the per-MonitoredItem (item, handler) pairs so /// / can detach the /// Notification delegates before the SDK disposes the subscription. /// private sealed record RemoteSubscription( Subscription Subscription, OpcUaSubscriptionHandle Handle, IReadOnlyList ItemHandlers); /// /// One (MonitoredItem, handler-delegate-instance) pair captured at subscribe time so /// the same delegate instance can be `-=` removed at unsubscribe time. The lambda /// captures the local OpcUaSubscriptionHandle, which is what makes detach /// necessary — without it the SDK's multicast invocation list holds the driver /// through the closure until the session itself is disposed. /// private sealed record MonitoredItemNotificationHandle( MonitoredItem Item, MonitoredItemNotificationEventHandler Handler); private sealed record OpcUaSubscriptionHandle(long Id) : ISubscriptionHandle { /// public string DiagnosticId => $"opcua-sub-{Id}"; } // ---- IAlarmSource ---- /// /// Field positions in the EventFilter SelectClauses below. Used to index into the /// EventFieldList.EventFields Variant collection when an event arrives. /// private const int AlarmFieldEventId = 0; private const int AlarmFieldEventType = 1; private const int AlarmFieldSourceNode = 2; private const int AlarmFieldMessage = 3; private const int AlarmFieldSeverity = 4; private const int AlarmFieldTime = 5; private const int AlarmFieldConditionId = 6; private const int AlarmFieldActiveState = 7; private const int AlarmFieldAckedState = 8; private const int AlarmFieldBranchId = 9; /// public async Task SubscribeAlarmsAsync( IReadOnlyList sourceNodeIds, CancellationToken cancellationToken) { // Fast-fail before queuing if the driver isn't connected, but do NOT bind the alarm // subscription to this reference — a reconnect can swap Session while we wait on _gate. // The session actually used is re-read inside the gate. _ = RequireSession(); var id = Interlocked.Increment(ref _nextAlarmSubscriptionId); var handle = new OpcUaAlarmSubscriptionHandle(id); // Pre-resolve the source-node filter set so the per-event notification handler can // match in O(1) without re-parsing on every event. // under v3 the host passes RawPaths, never upstream node ids. Translate each to the upstream // source id its tag authored (TagConfig.nodeId) so the per-event match compares like with like. // Keys are server-stable (AlarmIndexKey) so a remote namespace reorder cannot orphan them. var sessionForResolve = Session; var mapForResolve = _namespaceMap; var sourceFilter = BuildAlarmSourceIndex(sourceNodeIds, raw => { if (sessionForResolve is null) return null; if (!TryResolveTagNode(sessionForResolve, raw, out var live) && !NamespaceMap.TryResolve(sessionForResolve, raw, out live)) return null; return AlarmIndexKey(mapForResolve, live); }); BeginAlarmSubscription(sourceNodeIds); var subscription = new Subscription(telemetry: null!, new SubscriptionOptions { DisplayName = $"{AlarmSubscriptionNamePrefix}{id}", PublishingInterval = 500, // 500ms — alarms don't need fast polling; the server pushes KeepAliveCount = _options.SubscriptionKeepAliveCount, LifetimeCount = _options.SubscriptionLifetimeCount, MaxNotificationsPerPublish = _options.SubscriptionMaxNotificationsPerPublish, PublishingEnabled = true, Priority = 0, TimestampsToReturn = TimestampsToReturn.Both, }); // EventFilter SelectClauses — pick the standard BaseEventType fields we need to // materialize an AlarmEventArgs. Field positions are indexed by the AlarmField* // constants so the notification handler indexes in O(1) without re-examining the // QualifiedName BrowsePaths. var filter = new EventFilter(); void AddField(string browseName) => filter.SelectClauses.Add(new SimpleAttributeOperand { TypeDefinitionId = ObjectTypeIds.BaseEventType, BrowsePath = [new QualifiedName(browseName)], AttributeId = Attributes.Value, }); AddField("EventId"); AddField("EventType"); AddField("SourceNode"); AddField("Message"); AddField("Severity"); AddField("Time"); // ConditionId on ConditionType nodes is the branch identifier for // acknowledgeable conditions. Not a BaseEventType field — reach it via the typed path. filter.SelectClauses.Add(new SimpleAttributeOperand { TypeDefinitionId = ObjectTypeIds.ConditionType, BrowsePath = [], // empty path = the condition node itself AttributeId = Attributes.NodeId, }); // the host derives Active/Acked solely from AlarmEventArgs.Kind, so the condition state // must be selected. TWO-element browse paths — a single "ActiveState/Id" QualifiedName names nothing. filter.SelectClauses.Add(new SimpleAttributeOperand { TypeDefinitionId = ObjectTypeIds.AlarmConditionType, BrowsePath = [new QualifiedName(BrowseNames.ActiveState), new QualifiedName(BrowseNames.Id)], AttributeId = Attributes.Value, }); filter.SelectClauses.Add(new SimpleAttributeOperand { TypeDefinitionId = ObjectTypeIds.AcknowledgeableConditionType, BrowsePath = [new QualifiedName(BrowseNames.AckedState), new QualifiedName(BrowseNames.Id)], AttributeId = Attributes.Value, }); // BranchId — a non-null one marks a condition BRANCH event (a previous state, with its own // EventId). Only the trunk (null BranchId) is the condition's current state. filter.SelectClauses.Add(new SimpleAttributeOperand { TypeDefinitionId = ObjectTypeIds.ConditionType, BrowsePath = [new QualifiedName(BrowseNames.BranchId)], AttributeId = Attributes.Value, }); // Set by CreateAlarmEventItemAsync so a failure after the handler is attached can detach it. MonitoredItem? eventItem = null; MonitoredItemNotificationEventHandler? notifHandler = null; long subscribeGeneration = 0; await _subscriptionGate.WaitAsync(cancellationToken).ConfigureAwait(false); try { // Re-read Session inside the critical section: a reconnect completing while we were // blocked on _gate has already swapped in the new session via a re-initialise. subscribeGeneration = Volatile.Read(ref _lifecycleGeneration); var session = Session ?? throw new InvalidOperationException( "OpcUaClient session was lost before the alarm subscription could be created."); session.AddSubscription(subscription); try { await CreateAlarmEventItemAsync().ConfigureAwait(false); } catch { // the host retries a failed alarm subscribe from its reconcile, so a half-built subscription // left on the session would accumulate one per retry for the length of an outage. Detach the // handler first — the remove is best-effort and can itself fail against a broken server — and // bound it, since _gate is held. OnEventNotification also drops anything from an unregistered id. if (eventItem is not null && notifHandler is not null) { try { eventItem.Notification -= notifHandler; } catch { /* best-effort */ } } try { using var removeCts = new CancellationTokenSource(TimeSpan.FromSeconds(5)); await session.RemoveSubscriptionAsync(subscription, removeCts.Token).ConfigureAwait(false); } catch { /* best-effort — the session may be the thing that failed */ } // a half-built subscription can have been the server's only one, so its delete opens the same window. NoteSubscriptionDeleted(); throw; } // a new event item only reports FUTURE transitions — a condition already active upstream would // stay invisible until it next changed. ConditionRefresh re-delivers every retained condition's // current state through the normal path (deduplicated against the last-told state). await TryConditionRefreshAsync(subscription, cancellationToken).ConfigureAwait(false); } finally { _subscriptionGate.Release(); } return handle; async Task CreateAlarmEventItemAsync() { await subscription.CreateAsync(cancellationToken).ConfigureAwait(false); eventItem = new MonitoredItem(telemetry: null!, new MonitoredItemOptions { DisplayName = "Server/Events", StartNodeId = ObjectIds.Server, AttributeId = Attributes.EventNotifier, MonitoringMode = MonitoringMode.Reporting, QueueSize = 1000, // deep queue — a server can fire many alarms in bursts DiscardOldest = false, Filter = filter, }) { Handle = handle, }; // Capture the exact delegate instance so UnsubscribeAlarmsAsync / ShutdownAsync // can `-=` it later. The lambda captures `handle` and `sourceFilter`, so without // the explicit detach the SDK's invocation list keeps the driver instance alive // until the session itself is disposed. notifHandler = (mi, args) => OnEventNotification(handle, sourceFilter, mi, args); eventItem.Notification += notifHandler; subscription.AddItem(eventItem); await subscription.CreateItemsAsync(cancellationToken).ConfigureAwait(false); ThrowIfSubscribeSuperseded(subscribeGeneration); _alarmSubscriptions[id] = new RemoteAlarmSubscription(subscription, handle, eventItem, notifHandler); } } /// public async Task UnsubscribeAlarmsAsync(IAlarmSubscriptionHandle handle, CancellationToken cancellationToken) { if (handle is not OpcUaAlarmSubscriptionHandle h) return; if (!_alarmSubscriptions.TryRemove(h.Id, out var rs)) return; var emptied = NoteSubscriptionDeleted(); await _subscriptionGate.WaitAsync(cancellationToken).ConfigureAwait(false); try { // Detach the Notification handler before deleting the subscription so the SDK's // multicast invocation list no longer holds the driver instance through the // closure. try { rs.EventItem.Notification -= rs.Handler; } catch { /* best-effort */ } try { await rs.Subscription.DeleteAsync(silent: true, cancellationToken).ConfigureAwait(false); } catch { /* best-effort — session may already be gone across a reconnect */ } } finally { // the server answers the stale publish only once the delete is done, so the window runs from here. if (emptied) NoteSubscriptionDeleted(force: true); _subscriptionGate.Release(); } } /// public async Task AcknowledgeAsync( IReadOnlyList acknowledgements, CancellationToken cancellationToken) { // Short-circuit empty batch BEFORE touching the session so callers can pass an empty // list without guarding the size themselves — e.g. a bulk-ack UI that built an empty // list because the filter matched nothing. if (acknowledgements.Count == 0) return; // Fast-fail before queuing if the driver isn't connected, but do NOT bind the ack calls // (or the namespace-relative ConditionId parse) to this reference — a reconnect can swap // Session while we wait on _gate. The session actually used is re-read inside the gate. // Review m5: the same BadNotConnected as the in-gate check, so "no session" is one status to the caller. if (Session is null) throw new ServiceResultException(StatusCodes.BadNotConnected, "OpcUaClientDriver is not connected; the acknowledgement was not sent."); await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); try { // Re-read Session inside the critical section: a reconnect completing while we were // blocked on _gate has already swapped in the new session via a re-initialise. // ConditionId parsing is namespace-relative, so it must use the current session's // namespace table too. A session lost while this ack queued is a failure, not a silent success: the // host would otherwise log the ack as applied (R06:O-8). var generation = Volatile.Read(ref _lifecycleGeneration); var session = Session ?? throw new ServiceResultException(StatusCodes.BadNotConnected, "OpcUaClient session was lost before the acknowledgement could be sent."); // OPC UA A&C: call the AcknowledgeableConditionType.Acknowledge method on each // condition node with EventId + Comment arguments. CallAsync accepts a batch — // one CallMethodRequest per ack. var callRequests = new CallMethodRequestCollection(); foreach (var ack in acknowledgements) { // the host acks by RawPath; Acknowledge must hit the upstream CONDITION node. // The remembered target pairs the condition node with the EventId of the event that named it, // so the ack can never mix one condition's node with another's EventId. Nothing seen yet ⇒ the // tag's authored node (correct when the tag was authored on the condition node), else the id // verbatim (pre-v3 callers that pass an upstream condition id directly). NodeId conditionId; byte[] eventId; if (_alarmConditionByRawPath.TryGetValue(ack.ConditionId, out var seen)) { if (!NamespaceMap.TryResolve(session, seen.ConditionRef, out conditionId)) continue; eventId = seen.EventId; } else { if (!TryResolveTagNode(session, ack.ConditionId, out conditionId) && !NamespaceMap.TryResolve(session, ack.ConditionId, out conditionId)) continue; eventId = Array.Empty(); } callRequests.Add(new CallMethodRequest { ObjectId = conditionId, MethodId = MethodIds.AcknowledgeableConditionType_Acknowledge, InputArguments = [ new Variant(eventId), // Part 9 §5.7.3: the EventId being acknowledged (empty only when none seen) new Variant(new LocalizedText(ack.Comment ?? string.Empty)), ], }); } if (callRequests.Count == 0) { // Review m3 / r1 N-1: nothing resolved to a condition node, so nothing was sent — say so at Warning. NOT // thrown: this is a deterministic per-ack refusal, and a throw would count toward the host's // AlarmAcknowledge circuit breaker (5 failures), so an operator re-acking an unresolvable condition // would block every genuine ack on this driver. Only a no-reply transport failure throws (F2 owns a // per-ack result surface). _logger.LogWarning( "OpcUaClientDriver '{DriverInstanceId}': AcknowledgeAsync resolved none of {Count} condition id(s) " + "against the live session; nothing was sent upstream and the acknowledgement was NOT applied.", _driverInstanceId, acknowledgements.Count); return; } try { var resp = await session.CallAsync( requestHeader: null, methodsToCall: callRequests, ct: cancellationToken).ConfigureAwait(false); // A response — whatever the per-ack statuses say — is the server answering. RecordDeviceAnswered(generation, Clock.GetUtcNow().UtcDateTime); // Inspect per-ack results — the upstream server can reject individual acks // (BadConditionAlreadyAcked, BadNodeIdUnknown, BadUserAccessDenied) even when // the batch transport succeeds. Operators acking a critical alarm deserve to // know if the ack didn't take. if (resp?.Results is not null) { for (var i = 0; i < resp.Results.Count; i++) { var result = resp.Results[i]; if (StatusCode.IsBad(result.StatusCode)) { _logger.LogWarning( "OpcUaClientDriver '{DriverInstanceId}': AcknowledgeAsync ack[{Index}] " + "rejected by upstream server with StatusCode {StatusCode:X8}. " + "The acknowledgement may not have been applied.", _driverInstanceId, i, result.StatusCode.Code); } } } } catch (OperationCanceledException ex) { // Transport-level timeout / cancellation — propagate so the caller's // retry / re-ack mechanism can decide what to do. _logger.LogWarning(ex, "OpcUaClientDriver '{DriverInstanceId}': AcknowledgeAsync transport error.", _driverInstanceId); // Review m4: a timeout (not the caller cancelling) is a call with no reply, like a read or write. if (!cancellationToken.IsCancellationRequested) RecordCommsFault(generation, $"ack: {ex.Message}"); throw; } catch (Exception ex) { // R06:O-8 — a transport failure means NO reply: the acknowledgement was not applied upstream, so the // caller must see it (with the transport status, e.g. BadConnectionClosed) instead of a success. _logger.LogWarning(ex, "OpcUaClientDriver '{DriverInstanceId}': AcknowledgeAsync failed; " + "acknowledgements were not applied.", _driverInstanceId); RecordCommsFault(generation, $"ack: {ex.Message}"); throw; } } finally { _gate.Release(); } } private void OnEventNotification( OpcUaAlarmSubscriptionHandle handle, IReadOnlyDictionary sourceFilter, MonitoredItem item, MonitoredItemNotificationEventArgs args) { if (args.NotificationValue is not EventFieldList efl) return; // only a subscription SubscribeAlarmsAsync completed (and nothing has since released) may // deliver. A half-built one whose cleanup failed against a broken server can survive on the session; // without this it would feed RouteAlarmEvent alongside the real one. if (!_alarmSubscriptions.ContainsKey(handle.Id)) return; RecordLivenessEvidence(); RouteAlarmEvent(efl.EventFields, sourceFilter, handle); } /// /// open (or re-open) the host's alarm subscription over . /// A CHANGED set means the host applied an alarm-bearing delta, so every last-told state is forgotten and the /// post-subscribe ConditionRefresh re-reports as first sight. A rebuild now KEEPS native condition state, /// so the host may hold ANY state here — which is why first sight is flagged rather than assuming /// an inactive+acked default. The SAME set (a reinit / reconnect re-subscribe) keeps /// them, so the refresh copy of an already-reported state emits nothing. Ack targets of refs no longer /// subscribed are pruned either way. /// /// The refs the host is subscribing. internal void BeginAlarmSubscription(IReadOnlyList sourceNodeIds) { var set = new HashSet(sourceNodeIds, StringComparer.Ordinal); lock (_alarmStateLock) { if (_lastAlarmRefSet is null || !_lastAlarmRefSet.SetEquals(set)) _alarmStateByRawPath.Clear(); _lastAlarmRefSet = set; foreach (var known in _alarmConditionByRawPath.Keys) if (!set.Contains(known)) _alarmConditionByRawPath.TryRemove(known, out _); } } /// /// route one upstream event: drop refresh markers (a RefreshRequired asks for a new refresh), project /// it, and emit the kinds that move the host from the state last told to the event's. An event that changes /// nothing the host shows (a ConditionRefresh copy) emits nothing. /// /// The event's fields, in select-clause order. /// The subscription's upstream-key → RawPath index. /// The subscription handle to stamp. internal void RouteAlarmEvent( IReadOnlyList fields, IReadOnlyDictionary sourceFilter, IAlarmSubscriptionHandle handle) { if (IsRefreshRequired(fields)) { // Part 9 §5.11.7: the server lost events (queue overflow / restart) and asks for a refresh. // r2 R2-1: bound to the session's work token, which a shutdown cancels first thing. var sessionWork = _sessionWorkCts.Token; _ = Task.Run(() => RefreshAllAlarmSubscriptionsAsync(sessionWork)); return; } // Projection, the per-RawPath bookkeeping AND the emission run under one lock, so two notifications for // one condition (two publish threads; two subscriptions across a resubscribe) can never compute their // kinds from the same prior, nor reach the host out of order. The host's handler only marshals. lock (_alarmStateLock) { var e = ProjectEvent(fields, sourceFilter, handle, n => AlarmIndexKey(_namespaceMap, n), (raw, target) => _alarmConditionByRawPath[raw] = target, out var state); if (e is null) return; if (state is not { } target) { // Not a condition (no ActiveState/AckedState) — a message/severity-only event, as before. OnAlarmEvent?.Invoke(this, e); return; } // Fresh prior = the host projector's own default (inactive, acked). var told = new AlarmToldState(target.Active, target.Acked, e.Message, e.OpcUaSeverity); var priorKnown = _alarmStateByRawPath.TryGetValue(e.ConditionId, out var p); var prior = priorKnown ? p! : new AlarmToldState(false, true, null, null); _alarmStateByRawPath[e.ConditionId] = told; var kinds = AlarmTransitions((prior.Active, prior.Acked), target, priorKnown); if (kinds.Count == 0) { // Same state: a message/severity change still reaches the host; an identical copy does not. if (prior == told) return; kinds = [AlarmTransitionKind.Unspecified]; } // on first sight (a start or deploy respawn) this instance is RE-STATING upstream's current state, // not reporting a transition it watched — flagged so the host writes no duplicate history row when its // condition already shows exactly this. History only; the host still projects the state. foreach (var kind in kinds) OnAlarmEvent?.Invoke(this, e with { Kind = kind, IsRestatement = !priorKnown }); } } /// What the host was last told about one condition: its state plus the message and /// exact severity, so a refresh copy is recognised as a duplicate. /// ActiveState/Id last told. /// AckedState/Id last told. /// Message last told (null before any event). /// Exact severity last told (null before any event). private readonly record struct AlarmToldState(bool Active, bool Acked, string? Message, int? Severity); /// True for a RefreshRequiredEventType event (the server asks the client to ConditionRefresh). /// The event's fields. /// Whether the event is a RefreshRequired marker. internal static bool IsRefreshRequired(IReadOnlyList fields) => fields.Count > AlarmFieldEventType && fields[AlarmFieldEventType].Value is NodeId t && t == ObjectTypeIds.RefreshRequiredEventType; /// True for the three ConditionRefresh marker event types — never alarms. /// The event's EventType. /// Whether it is a refresh marker. private static bool IsRefreshMarker(object? eventType) => eventType is NodeId t && (t == ObjectTypeIds.RefreshStartEventType || t == ObjectTypeIds.RefreshEndEventType || t == ObjectTypeIds.RefreshRequiredEventType); /// A ConditionRefresh refusal meaning "this server does not do refresh" (tolerated, logged once). /// The refusal status. /// Whether the server simply lacks ConditionRefresh. internal static bool IsConditionRefreshUnsupported(StatusCode code) => code == StatusCodes.BadNotSupported || code == StatusCodes.BadMethodInvalid || code == StatusCodes.BadNotImplemented || code == StatusCodes.BadServiceUnsupported || code == StatusCodes.BadNodeIdUnknown; /// /// call ConditionRefresh for so conditions already active upstream /// are reported. Never throws: an unsupported server is logged once, any other failure warns. /// /// The (created) alarm subscription. /// Cancellation. /// A task completing when the refresh call returns. private async Task TryConditionRefreshAsync(Subscription subscription, CancellationToken cancellationToken) { try { await subscription.ConditionRefreshAsync(cancellationToken).ConfigureAwait(false); } catch (ServiceResultException ex) when (IsConditionRefreshUnsupported(ex.StatusCode)) { if (Interlocked.Exchange(ref _conditionRefreshUnsupportedLogged, 1) == 0) _logger.LogInformation( "OpcUaClientDriver '{DriverInstanceId}': upstream server does not support ConditionRefresh " + "({StatusCode}); conditions already active are reported only on their next transition.", _driverInstanceId, ex.StatusCode); } catch (Exception ex) when (ex is not OperationCanceledException || !cancellationToken.IsCancellationRequested) { _logger.LogWarning(ex, "OpcUaClientDriver '{DriverInstanceId}': ConditionRefresh on alarm subscription {SubscriptionId} failed; " + "conditions already active upstream may not be reported until their next transition.", _driverInstanceId, subscription.Id); } catch (OperationCanceledException) { // The caller cancelled; the subscription itself is in place. } } /// /// ConditionRefresh every alarm subscription on the current session (matched by display name): after /// a keep-alive outage the session survived and on a RefreshRequired event. Events missed during the /// outage reach the host through the normal deduplicated path. /// /// Cancellation. /// A task completing when every refresh call has returned. private async Task RefreshAllAlarmSubscriptionsAsync(CancellationToken cancellationToken) { if (_alarmSubscriptions.IsEmpty) return; try { await _subscriptionGate.WaitAsync(cancellationToken).ConfigureAwait(false); } catch (ObjectDisposedException) { return; } catch (OperationCanceledException) { return; /* the session's work was cancelled (shutdown / re-init) */ } try { var session = Session; if (session is null) return; foreach (var sub in session.Subscriptions.ToList()) { if (sub.DisplayName?.StartsWith(AlarmSubscriptionNamePrefix, StringComparison.Ordinal) == true && sub.Created) await TryConditionRefreshAsync(sub, cancellationToken).ConfigureAwait(false); } } catch (Exception ex) { _logger.LogWarning(ex, "OpcUaClientDriver '{DriverInstanceId}': alarm ConditionRefresh sweep failed.", _driverInstanceId); } finally { try { _subscriptionGate.Release(); } catch (ObjectDisposedException) { } } } /// Display-name prefix of every alarm subscription (how a refresh sweep finds them on a session). private const string AlarmSubscriptionNamePrefix = "opcua-alarm-sub-"; /// /// the sequence that moves the host's /// NativeAlarmProjector (which knows only kinds: Raise ⇒ active+unacked — or, on an already-active condition, a re-statement keeping the ack — Acknowledge ⇒ acked, /// ActiveAcknowledged ⇒ active+acked in one step, InactiveUnacknowledged ⇒ inactive+unacked in one step, Clear ⇒ inactive keeping acked) from to . A single /// upstream event can need two (an alarm that went active and was acknowledged between two notifications). /// No state change ⇒ empty (the caller decides whether a message/severity change warrants an Unspecified). /// /// The state last told to the host. /// The upstream event's ActiveState/Id + AckedState/Id. /// /// False when this driver instance has told the host nothing about the condition yet (first sight after a /// start or a deploy respawn), so is only the projector's default. The host now treats a Raise on an already-active condition as a re-statement that KEEPS its ack, so an /// upstream that says unacknowledged must then be told with /// — the upstream's ack state is authoritative, and the host may be holding an ack from before the respawn. /// /// The kinds to emit, in order. internal static IReadOnlyList AlarmTransitions( (bool Active, bool Acked) prior, (bool Active, bool Acked) target, bool priorKnown = true) { // on first sight the host may hold ANY state (it retains a still-deployed condition across a // respawn), so an inactive target is stated in kinds that land on it from every host state: inactive+unacked // as ONE InactiveUnacknowledged; inactive+acked as Clear then Acknowledge (flagged by the caller, so a host // already showing it writes no row, and a host retaining it Active gets a real "Cleared"). Pre-fix an // inactive+acked first sight equalled the default prior and went out as Unspecified — an "Activated" row for // an inactive alarm, and a retained-Active host condition was never cleared. if (!priorKnown && !target.Active) return target.Acked ? [AlarmTransitionKind.Clear, AlarmTransitionKind.Acknowledge] : [AlarmTransitionKind.InactiveUnacknowledged]; if (prior == target) return []; var kinds = Transitions(prior, target); return priorKnown || target.Acked ? kinds : [.. kinds.Select(k => k == AlarmTransitionKind.Raise ? AlarmTransitionKind.Retrigger : k)]; } private static IReadOnlyList Transitions( (bool Active, bool Acked) prior, (bool Active, bool Acked) target) { return target switch { // Active+unacked: a fresh raise, or a re-arm of an active-acked condition. (true, false) => [prior.Active ? AlarmTransitionKind.Retrigger : AlarmTransitionKind.Raise], // Active+acked from inactive: ONE atomic ActiveAcknowledged — Raise-then-Acknowledge showed // clients an Active+UNacked intermediate event before the ack. (true, true) => prior.Active ? [AlarmTransitionKind.Acknowledge] : [AlarmTransitionKind.ActiveAcknowledged], // Inactive+acked: Clear keeps the prior ack, so ack after it when that was unacked. (false, true) => prior.Active ? prior.Acked ? [AlarmTransitionKind.Clear] : [AlarmTransitionKind.Clear, AlarmTransitionKind.Acknowledge] : [AlarmTransitionKind.Acknowledge], // Inactive+UNacked (went active and cleared unacknowledged — e.g. first seen by a ConditionRefresh): ONE // atomic InactiveUnacknowledged — Raise/Retrigger-then-Clear showed clients the condition // Active again for an instant. From Active+unacked a plain Clear already lands there. (false, false) => prior is (true, false) ? [AlarmTransitionKind.Clear] : [AlarmTransitionKind.InactiveUnacknowledged], }; } /// /// index an alarm subscription's source refs by UPSTREAM source id. The host (v3) subscribes with /// the alarm tags' RawPaths; an upstream event carries its own source NodeId, so the match key is /// the upstream id the tag authored and the value is the RawPath to report back. A ref /// cannot resolve is kept verbatim (direct upstream-id use). /// /// The refs handed to SubscribeAlarmsAsync. /// RawPath → serialized live upstream NodeId, or null when unknown. /// Upstream source id → ref to report. internal static Dictionary BuildAlarmSourceIndex( IReadOnlyList refs, Func resolveUpstream) { var index = new Dictionary(StringComparer.Ordinal); foreach (var r in refs) index[resolveUpstream(r) ?? r] = r; return index; } /// /// Project one upstream event into an . ConditionId and /// SourceNodeId are the authored RawPath, because DriverHostActor routes native alarms on /// (driver, RawPath); the upstream condition node (needed to acknowledge) is handed to /// . Null when the event's source is not an authored one. /// /// The event's select-clause field values. /// Upstream source id → authored RawPath. /// The alarm subscription handle the event belongs to. /// Maps an upstream NodeId to its stable string key. /// Callback that records the upstream condition node for later acknowledgement. /// The projected event, or null when the event is not for an authored source. internal static AlarmEventArgs? ProjectEvent( IReadOnlyList fields, IReadOnlyDictionary sourceIndex, IAlarmSubscriptionHandle handle, Func keyOf, Action rememberCondition) => ProjectEvent(fields, sourceIndex, handle, keyOf, rememberCondition, out _); /// As the five-argument overload, also returning the event's (ActiveState/Id, AckedState/Id) through /// state when it is an alarm condition (null otherwise). /// The event's select-clause field values. /// Upstream source id → authored RawPath. /// The alarm subscription handle the event belongs to. /// Maps an upstream NodeId to its stable string key. /// Callback that records the upstream condition node for later acknowledgement. /// Receives the event's (Active, Acked) state when it is an alarm condition; otherwise null. /// The projected event, or null when the event is not for an authored source. internal static AlarmEventArgs? ProjectEvent( IReadOnlyList fields, IReadOnlyDictionary sourceIndex, IAlarmSubscriptionHandle handle, Func keyOf, Action rememberCondition, out (bool Active, bool Acked)? state) { state = null; if (fields.Count <= AlarmFieldConditionId) return null; // ConditionRefresh brackets its re-delivery with RefreshStart/RefreshEnd — markers, never alarms. if (IsRefreshMarker(fields[AlarmFieldEventType].Value)) return null; // A condition BRANCH event reports a previous state with its own EventId: projecting it would flip the // trunk's state, remembering it would make Acknowledge ack the branch. if (fields.Count > AlarmFieldBranchId && fields[AlarmFieldBranchId].Value is NodeId branch && !NodeId.IsNull(branch)) return null; if (fields.Count > AlarmFieldAckedState && fields[AlarmFieldActiveState].Value is bool active && fields[AlarmFieldAckedState].Value is bool acked) state = (active, acked); // Pass-through (no authored refs) keeps the raw serialized ids; an authored index compares // server-stable keys (see AlarmIndexKey), matching the CONDITION node first — the only way to tell // several conditions on one source apart — and then the source node. var sourceField = fields[AlarmFieldSourceNode].Value; var conditionField = fields[AlarmFieldConditionId].Value; string reported, upstreamCondition; if (sourceIndex.Count == 0) { reported = sourceField?.ToString() ?? string.Empty; upstreamCondition = conditionField?.ToString() ?? string.Empty; } else { upstreamCondition = conditionField is NodeId cn && !NodeId.IsNull(cn) ? keyOf(cn) : string.Empty; var sourceKey = sourceField is NodeId sn && !NodeId.IsNull(sn) ? keyOf(sn) : string.Empty; if (!(upstreamCondition.Length > 0 && sourceIndex.TryGetValue(upstreamCondition, out reported!)) && !(sourceKey.Length > 0 && sourceIndex.TryGetValue(sourceKey, out reported!))) return null; } var eventType = fields[AlarmFieldEventType].Value?.ToString() ?? "BaseEventType"; var message = (fields[AlarmFieldMessage].Value as LocalizedText)?.Text ?? string.Empty; var severity = fields[AlarmFieldSeverity].Value is ushort sev ? sev : (ushort)0; var time = fields[AlarmFieldTime].Value is DateTime t ? t : DateTime.UtcNow; if (sourceIndex.Count != 0 && upstreamCondition.Length > 0) rememberCondition(reported, new AlarmAckTarget( upstreamCondition, // A copy: the decoded notification's array is not ours to alias for the ack's lifetime. (fields[AlarmFieldEventId].Value as byte[])?.ToArray() ?? Array.Empty())); return new AlarmEventArgs( SubscriptionHandle: handle, SourceNodeId: reported, ConditionId: sourceIndex.Count == 0 ? upstreamCondition : reported, AlarmType: eventType, Message: message, Severity: MapSeverity(severity), SourceTimestampUtc: time, // the upstream severity is already the exact Part 9 value — carry it, don't re-bucket. OpcUaSeverity: severity == 0 ? null : severity); } /// What Acknowledge needs for one RawPath: the upstream condition node (server-stable /// reference) and the EventId of the event that named it. /// Upstream condition NodeId, rendered by . /// EventId of the most recent event seen for that condition. internal readonly record struct AlarmAckTarget(string ConditionRef, byte[] EventId); /// The key both sides of the alarm index use: server-stable (nsu=) when a map is /// available, so a remote namespace-table reorder cannot orphan it. /// The current session's namespace map, or null. /// The NodeId to key. /// The index key. internal static string AlarmIndexKey(NamespaceMap? map, NodeId nodeId) => map?.ToStableReference(nodeId) ?? nodeId.ToString() ?? string.Empty; /// /// Map an OPC UA BaseEventType.Severity (1..1000) to our coarse-grained /// bucket. Thresholds match the OPC UA A&C Part 9 /// guidance: 1-200 Low, 201-500 Medium, 501-800 High, 801-1000 Critical. /// /// The OPC UA severity value (1-1000). /// The mapped coarse-grained alarm severity bucket. internal static AlarmSeverity MapSeverity(ushort opcSeverity) => opcSeverity switch { <= 200 => AlarmSeverity.Low, <= 500 => AlarmSeverity.Medium, <= 800 => AlarmSeverity.High, _ => AlarmSeverity.Critical, }; /// /// Live alarm-event subscription bookkeeping. Holds the SDK , /// the local handle, the single event-MonitoredItem (`Server/Events`), and the exact /// handler delegate instance so unsubscribe / shutdown can detach the Notification /// event before the SDK disposes the subscription. /// private sealed record RemoteAlarmSubscription( Subscription Subscription, OpcUaAlarmSubscriptionHandle Handle, MonitoredItem EventItem, MonitoredItemNotificationEventHandler Handler); private sealed record OpcUaAlarmSubscriptionHandle(long Id) : IAlarmSubscriptionHandle { /// public string DiagnosticId => $"opcua-alarm-sub-{Id}"; } // ---- IHostConnectivityProbe ---- /// /// Endpoint-URL-keyed host identity for the Admin /hosts dashboard. Reflects the /// endpoint the driver actually connected to after the failover sweep — not the /// first URL in the candidate list — so operators see which of the configured /// endpoints is currently serving traffic. Falls back to the first configured URL /// pre-init so the dashboard has something to render before the first connect. /// public string HostName => _connectedEndpointUrl ?? ResolveEndpointCandidates(_options).FirstOrDefault() ?? _options.EndpointUrl; /// public IReadOnlyList GetHostStatuses() { lock (_probeLock) return [new HostConnectivityStatus(HostName, _hostState, _hostStateChangedUtc)]; } /// Hook the SDK's publish-side events into the diagnostics counters. /// The session to observe. private void AttachDiagnostics(ISession session) { _notificationHandler ??= (_, e) => { _diag.Record(OpcUaClientDiagnostics.PublishResponse); var msg = e.NotificationMessage; if (msg?.NotificationData is { } data) foreach (var nd in data) _diag.Record(OpcUaClientDiagnostics.Notification, CountNotifications(nd)); }; _publishErrorHandler ??= OnPublishError; session.Notification -= _notificationHandler; session.Notification += _notificationHandler; session.PublishError -= _publishErrorHandler; session.PublishError += _publishErrorHandler; } private void DetachDiagnostics(ISession session) { try { if (_notificationHandler is not null) session.Notification -= _notificationHandler; if (_publishErrorHandler is not null) session.PublishError -= _publishErrorHandler; } catch { /* best-effort — SDK may have already cleared its invocation lists on session loss */ } } private static int CountNotifications(ExtensionObject nd) => nd.Body switch { DataChangeNotification dc => dc.MonitoredItems?.Count ?? 0, EventNotificationList ev => ev.Events?.Count ?? 0, StatusChangeNotification => 1, _ => 1, }; /// /// Session keep-alive handler. A good keep-alive is liveness evidence: it advances /// , and after a bad one it ends the outage in place (the session /// survived — e.g. a network blip): health back to Healthy and the outage's Bad refs re-read and re-published /// Good, because a server whose subscription survived will not resend values that did not change. A bad /// keep-alive degrades health (stamp preserved) and, once per outage, publishes BadNoCommunication for /// every subscribed ref (value retained) plus a quality-only annotation on every subscribed alarm condition. /// /// /// There is deliberately no SessionReconnectHandler any more. In SDK 1.5.378 it retried forever, its /// session-recreate path cloned subscriptions the driver never re-tracked, and it kept health Healthy while /// it did so — the symptom. A session that does not come back keeps health Degraded with the read stamp /// frozen, which is exactly what the host's comms-loss watchdog trips on: ShutdownAsync + InitializeAsync /// + re-subscribe, whose initial values come back Good. When the keep-alive status proves the session is gone /// for good () the driver also raises , /// so the host runs that sequence at once instead of after the watchdog's silence window. /// private void OnKeepAlive(ISession sender, KeepAliveEventArgs e) { // Captured BEFORE the session check: a shutdown / re-initialise that releases this session after the check // has bumped the generation, so every health write below lands only for the session it was raised for. var generation = Volatile.Read(ref _lifecycleGeneration); // A late callback from a session already released (re-initialise) says nothing about the current one. if (!ReferenceEquals(sender, Session)) return; var now = Clock.GetUtcNow().UtcDateTime; if (!ServiceResult.IsBad(e.Status)) { Interlocked.Exchange(ref _closedChannelFailures, 0); TransitionTo(HostState.Running); // The session answers but a subscription is dead — no stamp, no recovery; the watchdog reinit is the fix. if (Volatile.Read(ref _subscriptionLost) == 1) { Interlocked.Exchange(ref _sessionLost, 0); return; } if (Interlocked.Exchange(ref _sessionLost, 0) == 1) { UpdateHealth(generation, _ => new DriverHealth(DriverState.Healthy, now, null)); // A fresh grace: every subscription reports stopped right after an outage until its next message. Interlocked.Exchange(ref _publishingStoppedSinceTicks, 0); var guardUntil = ArmReplayGuard(now); _diag.Record(OpcUaClientDiagnostics.KeepAliveRecovered); _logger.LogInformation( "OpcUaClientDriver '{DriverInstanceId}': upstream keep-alive recovered on the same session; " + "re-reading the refs published Bad during the outage.", _driverInstanceId); var recovery = Task.Run(() => RecoverAfterOutageAsync(sender)); _recoveryTask = recovery; var sessionWork = _sessionWorkCts.Token; _reconcileTask = ReconcileAfterRecoveryAsync(sender, recovery, PostRecoveryReconcileDelay(), sessionWork); _ = DisarmReplayGuardWhenDueAsync(guardUntil, sessionWork); return; } if (FindStalledSubscription(now) is { } stalled) { DeclareSubscriptionLost(generation, $"subscription '{stalled}' stopped publishing while the session stayed up", now, serverReported: false); return; } // A good keep-alive is the server answering. With no outage and no dead subscription latched, a Degraded // here can only be a failed read/write the server has since contradicted, so it clears too — otherwise a // subscription-only driver (which never reads again) stayed Degraded until its next re-initialise. UpdateHealth(generation, h => h.State switch { DriverState.Healthy => h with { LastSuccessfulRead = now }, DriverState.Degraded => new DriverHealth(DriverState.Healthy, now, null), _ => null, }); return; } Interlocked.Exchange(ref _publishingStoppedSinceTicks, 0); TransitionTo(HostState.Stopped); _diag.Record(OpcUaClientDiagnostics.KeepAliveMiss); UpdateHealth(generation, h => new DriverHealth(DriverState.Degraded, h.LastSuccessfulRead, $"keep-alive: {e.Status}")); // the first bad keep-alive is often ambiguous and a definitive one follows, so this is checked on EVERY // bad keep-alive — but only after the outage below has been published, so the Bad reaches subscribers first. var keepAliveCode = e.Status?.StatusCode ?? StatusCodes.Good; var definitive = IsDefinitiveSessionLoss(keepAliveCode) || (IsClosedChannel(keepAliveCode) && Interlocked.Increment(ref _closedChannelFailures) >= SustainedClosedChannelFailures); // A subscription loss already published Bad for this connection. if (Interlocked.Exchange(ref _sessionLost, 1) == 0 && Volatile.Read(ref _subscriptionLost) == 0) { _logger.LogWarning( "OpcUaClientDriver '{DriverInstanceId}': upstream session keep-alive failed ({Status}); subscribed values " + "are BadNoCommunication until it recovers or the host's comms-loss watchdog reconnects the driver.", _driverInstanceId, e.Status); PublishCommsLoss(now); } if (definitive) RaiseConnectionLost($"upstream session keep-alive reported {StatusCodes.GetBrowseName(keepAliveCode.Code)}"); } /// /// the connection under the session is CLOSED (not merely silent). A real upstream process stop /// (docker restart, SIGINT, docker kill) never produced a session status on the docker-dev gate: the keep-alives /// reported BadSecureChannelClosed once, then BadConnectionClosed interleaved with /// BadNoCommunication for the whole outage. Reproduced in-process (a proxy that resets every connection /// while the server and its session stay alive): the SDK 1.5.378 session never opens a new connection on its /// own — over 48 s with the server reachable again, not one connection attempt — so a closed channel can never /// recover in place. It is treated as definitive once SUSTAINED ( in /// one outage), so a single transient contradicted by the next good keep-alive does not count. A stall (a paused /// upstream, a congested link) produces BadRequestTimeout/BadNoCommunication instead and keeps /// its in-place recovery. /// /// A keep-alive status. /// True for a closed secure channel or connection. internal static bool IsClosedChannel(StatusCode status) => status == StatusCodes.BadSecureChannelClosed || status == StatusCodes.BadConnectionClosed; /// /// keep-alive statuses that prove the session can NEVER recover in place, so waiting out the /// comms-loss watchdog would only delay the inevitable re-initialise. The server forgot the session /// (BadSessionIdInvalid — what a re-opened channel to a restarted server answers), closed it /// (BadSessionClosed) or never activated it (BadSessionNotActivated); or the server HALTED /// (BadServerHalted) — measured on the restart wire test, the SDK never re-opens a channel the /// server halted: 15 s after the server was back the same session still failed every request with /// BadServerHalted, so a halted session is dead for good. A closed connection is definitive too, but only /// once sustained — see . Everything else — BadNoCommunication, /// BadNotConnected, request timeouts — is AMBIGUOUS (a stalled link that resumes keeps its session, see the /// stall wire test) and stays with the watchdog. /// /// A keep-alive status. /// True when the session is definitively gone. internal static bool IsDefinitiveSessionLoss(StatusCode status) => status == StatusCodes.BadSessionIdInvalid || status == StatusCodes.BadSessionClosed || status == StatusCodes.BadSessionNotActivated || status == StatusCodes.BadServerHalted; /// Raise once for the current session, isolating a faulting subscriber. /// Why the connection is definitively gone, for the host's log. private void RaiseConnectionLost(string reason) { if (Interlocked.Exchange(ref _connectionLostRaised, 1) == 1) return; _logger.LogWarning( "OpcUaClientDriver '{DriverInstanceId}': connection definitively lost ({Reason}); asking the host to " + "re-initialise now instead of waiting for the comms-loss watchdog.", _driverInstanceId, reason); try { ConnectionLost?.Invoke(this, new DriverConnectionLossEventArgs(reason)); } catch (Exception ex) { _logger.LogWarning(ex, "OpcUaClientDriver '{DriverInstanceId}': a ConnectionLost subscriber faulted.", _driverInstanceId); } } /// /// the display name of a tracked subscription that has reported "publishing stopped" for /// longer than its grace, measured across consecutive good keep-alives; null when none has. /// private string? FindStalledSubscription(DateTime now) { Subscription? worst = null; var grace = TimeSpan.Zero; foreach (var sub in _subscriptions.Values.Select(r => r.Subscription) .Concat(_alarmSubscriptions.Values.Select(r => r.Subscription))) { if (!IsPublishingStopped(sub)) continue; var g = PublishingStallGrace(sub); if (worst is null || g > grace) { worst = sub; grace = g; } } if (worst is null) { Interlocked.Exchange(ref _publishingStoppedSinceTicks, 0); return null; } var since = Interlocked.CompareExchange(ref _publishingStoppedSinceTicks, now.Ticks, 0); if (since == 0) since = now.Ticks; return now.Ticks - since >= grace.Ticks ? worst.DisplayName ?? $"#{worst.Id}" : null; } /// /// the server dropped a subscription the driver still tracks. Degraded with the read stamp /// frozen (sticky until the session is released), and the outage published once, so the host's watchdog /// re-initialises and re-subscribes. /// /// The lifecycle generation the detection belongs to. /// Why, for the log and health. /// Detection time. /// True when the SERVER said the subscription is gone (BadNoSubscription / /// BadSubscriptionIdInvalid) — definitive, so is raised. False for the publishing-stall /// heuristic, an inference from silence that stays with the comms-loss watchdog. private void DeclareSubscriptionLost(long generation, string reason, DateTime now, bool serverReported) { if (Interlocked.Exchange(ref _subscriptionLost, 1) == 1) return; UpdateHealth(generation, h => new DriverHealth(DriverState.Degraded, h.LastSuccessfulRead, $"subscription lost: {reason}")); _logger.LogWarning( "OpcUaClientDriver '{DriverInstanceId}': {Reason}; subscribed values are BadNoCommunication until the " + "host re-initialises the driver and re-subscribes.", _driverInstanceId, reason); if (Volatile.Read(ref _sessionLost) == 0) PublishCommsLoss(now); // a subscription the SERVER says is gone never comes back on this session — only the re-subscribe after a // re-initialise recreates it — so ask the host now rather than after the watchdog's 90 s. The stall heuristic is // an inference, not a server statement, and keeps going through the watchdog. if (serverReported) RaiseConnectionLost($"subscription lost: {reason}"); } /// /// Session PublishError: counted, except an anonymous BadNoSubscription while the driver tracks no /// subscription at all (#764) — the SDK keeps publishing on a session with none, and the server's answer is /// idle noise, not an error. A BadNoSubscription / BadSubscriptionIdInvalid naming a /// subscription the driver tracks means the server dropped it while the session lived. An anonymous /// BadNoSubscription means the server holds none at all — a loss of anything tracked, except within /// of the driver deleting its last subscription (#763). /// private void OnPublishError(object? sender, PublishErrorEventArgs e) => HandlePublishError(sender, e.Status?.StatusCode ?? StatusCodes.Good, e.SubscriptionId); private void HandlePublishError(object? sender, StatusCode code, uint subscriptionId) { // #764: with nothing tracked there is nothing to lose, and the SDK's publish loop keeps asking a session // that holds no subscription — every answer is an anonymous BadNoSubscription. Counting those made // publish_error climb on every idle driver; they are neither counted nor acted on. if (code == StatusCodes.BadNoSubscription && subscriptionId == 0 && _subscriptions.IsEmpty && _alarmSubscriptions.IsEmpty) { return; } _diag.Record(OpcUaClientDiagnostics.PublishError); var generation = Volatile.Read(ref _lifecycleGeneration); if (sender is not null && !ReferenceEquals(sender, Session)) return; var now = Clock.GetUtcNow().UtcDateTime; bool tracked; if (code == StatusCodes.BadSubscriptionIdInvalid || (code == StatusCodes.BadNoSubscription && subscriptionId != 0)) { // The error names a subscription: it is a loss only if that subscription is one the driver still tracks // (one it deleted itself, or never registered, is not). tracked = IsTrackedSubscription(subscriptionId); } else if (code == StatusCodes.BadNoSubscription) { // Anonymous: the server holds NO subscription for the session. True of anything tracked — unless the // driver just deleted its last subscription, when it is the answer to a publish queued in that gap (#763). tracked = !_subscriptions.IsEmpty || !_alarmSubscriptions.IsEmpty; if (tracked && IsWithinStaleNoSubscriptionGrace(now)) { _logger.LogDebug( "OpcUaClientDriver '{DriverInstanceId}': ignoring BadNoSubscription within {Grace} of deleting the " + "last subscription — it answers a publish queued while the session held none, not a loss of the " + "subscription registered since.", _driverInstanceId, StaleNoSubscriptionGrace); return; } } else return; if (tracked) DeclareSubscriptionLost(generation, $"server reported {code} for a tracked subscription", now, serverReported: true); } private bool IsTrackedSubscription(uint subscriptionId) => _subscriptions.Values.Any(r => r.Subscription.Id == subscriptionId) || _alarmSubscriptions.Values.Any(r => r.Subscription.Id == subscriptionId); /// /// Stamp when a delete leaves the driver tracking no subscription /// (or unconditionally with , to restart the window once that delete has completed). /// /// Stamp even though a subscription has been registered since (the delete it follows did /// empty the set). /// True when the stamp was taken. private bool NoteSubscriptionDeleted(bool force = false) { if (!force && (!_subscriptions.IsEmpty || !_alarmSubscriptions.IsEmpty)) return false; Interlocked.Exchange(ref _lastSubscriptionsEmptiedTicks, Clock.GetUtcNow().UtcDateTime.Ticks); return true; } private bool IsWithinStaleNoSubscriptionGrace(DateTime now) { var emptied = Interlocked.Read(ref _lastSubscriptionsEmptiedTicks); return emptied != 0 && now.Ticks - emptied < StaleNoSubscriptionGrace.Ticks; } /// /// publish the start of an outage: BadNoCommunication once per (subscription, ref) with the last /// delivered value retained, and a quality-only annotation on every subscribed alarm condition (no state /// change — an outage says nothing about whether the alarm is active). /// private void PublishCommsLoss(DateTime now) { // Under the delivery lock (re-review of 30a9d5f0) so the Bad cannot interleave with a recovery publish. lock (_deliveryLock) { foreach (var rs in _subscriptions.Values) { foreach (var pair in rs.ItemHandlers) { var fullRef = RefOf(pair.Item); _lastValues.TryGetValue(fullRef, out var last); _awaitingRecovery[fullRef] = 0; _deliverySeq.AddOrUpdate(fullRef, 1, (_, n) => n + 1); // F1: stamped with the DETECTION time, as the host's own comms-loss fan does. The recorder historizes // SourceTimestampUtc ?? ServerTimestampUtc, so the last Good's source time here put the Bad at the // Good row's (TagId, SourceTimeUtc) key, where the insert-if-absent historian dropped it. RaiseDataChange(new DataChangeEventArgs(rs.Handle, fullRef, new DataValueSnapshot( last?.Value, StatusBadNoCommunication, now, now))); } } } AnnotateAlarmQuality(StatusBadNoCommunication, now); } /// /// the session survived the outage: re-read the refs still awaiting recovery and publish them (a ref /// a notification already refreshed is skipped — fresh data beats the read), restore alarm-condition quality /// and ConditionRefresh so transitions missed during the outage are re-reported (deduplicated). A failed /// read re-arms the outage so the next good keep-alive tries again. /// /// The session whose keep-alive recovered. Every effect — publishing, re-arming the /// latch — is conditional on it still being the current session and still out of outage: a /// recovery overtaken by a re-initialise must not leak its outcome onto the NEW session. private async Task RecoverAfterOutageAsync(ISession session) { bool Current() => ReferenceEquals(session, Session) && Volatile.Read(ref _sessionLost) == 0; try { if (!Current()) return; var refs = _awaitingRecovery.Keys.ToList(); if (refs.Count > 0) { var (results, transportOk) = await ReadCoreAsync(refs, CancellationToken.None, session).ConfigureAwait(false); if (!transportOk) { // Re-arm only for the session this recovery belongs to. if (ReferenceEquals(session, Session)) Interlocked.Exchange(ref _sessionLost, 1); return; } for (var i = 0; i < refs.Count; i++) { if (!Current()) return; // One critical section with the live notification path (_deliveryLock): a notification that // removed the ref first wins, and none can overtake this publish once it has decided. lock (_deliveryLock) { if (!_awaitingRecovery.TryRemove(refs[i], out _)) continue; // a notification got there first PublishUpstreamLocked(refs[i], results[i]); } } } if (!Current()) return; AnnotateAlarmQuality(StatusGood, Clock.GetUtcNow().UtcDateTime); if (!_alarmSubscriptions.IsEmpty) await RefreshAllAlarmSubscriptionsAsync(_sessionWorkCts.Token).ConfigureAwait(false); } catch (Exception ex) { _logger.LogWarning(ex, "OpcUaClientDriver '{DriverInstanceId}': in-place recovery after an outage failed.", _driverInstanceId); } } /// /// Publish an upstream value read (not notified) for to every subscription carrying it, /// recording it as the last delivered value and — when Good with a source time — as the replay guard's mark, so a /// replay older than the read is late too. Caller holds . /// private void PublishUpstreamLocked(string fullRef, DataValueSnapshot value) { _lastValues[fullRef] = value; _deliverySeq.AddOrUpdate(fullRef, 1, (_, n) => n + 1); RecordUpstreamGood(fullRef, value); if (value.SourceTimestampUtc is { } readAt && StatusCode.IsGood(value.StatusCode) && Interlocked.Read(ref _replayGuardUntilTicks) != 0) _replayGuard.AddOrUpdate(fullRef, readAt.Ticks, (_, prior) => Math.Max(prior, readAt.Ticks)); foreach (var rs in _subscriptions.Values) if (rs.ItemHandlers.Any(p => RefOf(p.Item) == fullRef)) RaiseDataChange(new DataChangeEventArgs(rs.Handle, fullRef, value)); } /// /// the trailing reconcile read after an in-place recovery. Found by the TCP-stall wire test: a value change /// the server published just after a stall ended rode a Publish response that had been in flight across the /// stall; the SDK timed that request out (BadTimeout), discarded the response, and never republished the /// message (the following keep-alives advanced the sequence number past it). The node stayed on the value from /// the recovery read — stale, indefinitely, for a value that does not change again. Once every request that was /// in flight across the stall has completed or timed out (), every /// subscribed ref is read once more and published when says the read differs from /// what was last delivered and is not older than it. The decision and the publish are one critical section with /// the live notification path (), so a newer notification can never be overtaken. /// Bound to the session's work token: a re-initialise or shutdown cancels it. /// /// The session that recovered; nothing happens if it is no longer current or is in outage. /// The recovery read, which runs first so this read is compared against its result. /// How long to wait before reading (on ). /// The session's background-work token. private async Task ReconcileAfterRecoveryAsync(ISession session, Task recovery, TimeSpan delay, CancellationToken ct) { bool Current() => !ct.IsCancellationRequested && ReferenceEquals(session, Session) && Volatile.Read(ref _sessionLost) == 0 && Volatile.Read(ref _subscriptionLost) == 0; try { if (delay > TimeSpan.Zero) await Task.Delay(delay, Clock, ct).ConfigureAwait(false); await recovery.WaitAsync(ct).ConfigureAwait(false); if (!Current()) return; var refs = _subscriptions.Values .SelectMany(rs => rs.ItemHandlers.Select(p => RefOf(p.Item))) .Distinct(StringComparer.Ordinal) .ToList(); if (refs.Count == 0) return; // Captured BEFORE the read is issued: a ref delivered to while the read is in flight has a value at least as // new as anything the read could have sampled, so the read must not overwrite it. var seqAtIssue = refs.Select(r => _deliverySeq.GetValueOrDefault(r)).ToArray(); var (results, transportOk) = await ReadCoreAsync(refs, ct, session).ConfigureAwait(false); if (!transportOk) return; // a new outage — the keep-alive reports it var published = 0; for (var i = 0; i < refs.Count; i++) { if (!Current()) return; lock (_deliveryLock) { if (_deliverySeq.GetValueOrDefault(refs[i]) != seqAtIssue[i]) continue; if (!IsReconcileRepair(refs[i], results[i])) continue; BeforeReconcilePublishForTest?.Invoke(refs[i]); _awaitingRecovery.TryRemove(refs[i], out _); PublishUpstreamLocked(refs[i], results[i]); } published++; } if (published > 0) _logger.LogInformation( "OpcUaClientDriver '{DriverInstanceId}': trailing re-read after recovery found {Count} of {Total} refs " + "whose latest change never arrived through the subscription; published them.", _driverInstanceId, published, refs.Count); } catch (OperationCanceledException) when (ct.IsCancellationRequested) { /* session released */ } catch (Exception ex) { _logger.LogWarning(ex, "OpcUaClientDriver '{DriverInstanceId}': trailing re-read after a recovery failed.", _driverInstanceId); } } /// /// Whether the trailing read's repairs : its (value, status) /// differs from the last value delivered for the ref, and it is not OLDER than it. Any status counts — a Good→Bad /// or Good→Uncertain change lost in the stalled pipeline is as stale as a lost value. "Older" is judged on source /// timestamps only when both carry one; a read with no source timestamp (a server that does not send them) is the /// server's CURRENT state, read after the stalled pipeline drained, so it counts as newer. An EQUAL source /// timestamp with different content is published too: it is not a replay but the server's current state, and it /// repairs the one change the replay guard drops by design (a status-only change that kept its source time inside /// the window). Caller holds . /// private bool IsReconcileRepair(string fullRef, DataValueSnapshot read) { if (!_lastValues.TryGetValue(fullRef, out var last)) return true; if (last.StatusCode == read.StatusCode && Utils.IsEqual(last.Value, read.Value)) return false; return read.SourceTimestampUtc is not { } src || last.SourceTimestampUtc is not { } lastSrc || src >= lastSrc; } /// Quality-only annotation of every subscribed alarm condition. private void AnnotateAlarmQuality(uint quality, DateTime now) { if (_alarmSubscriptions.IsEmpty || OnAlarmEvent is null) return; var handle = _alarmSubscriptions.Values.First().Handle; lock (_alarmStateLock) { if (_lastAlarmRefSet is null) return; foreach (var conditionId in _lastAlarmRefSet) { _alarmStateByRawPath.TryGetValue(conditionId, out var told); var e = new AlarmEventArgs(handle, SourceNodeId: conditionId, ConditionId: conditionId, AlarmType: "ConditionType", Message: told.Message ?? string.Empty, Severity: told.Severity is { } sev ? MapSeverity((ushort)sev) : AlarmSeverity.Low, SourceTimestampUtc: now, Kind: AlarmTransitionKind.Unspecified, OpcUaSeverity: told.Severity, QualityStatusCode: quality); try { OnAlarmEvent?.Invoke(this, e); } catch (Exception ex) { _logger.LogWarning(ex, "OpcUaClientDriver '{DriverInstanceId}': alarm subscriber faulted on a quality " + "annotation for {ConditionId}.", _driverInstanceId, conditionId); } } } } /// Raise , isolating a faulting subscriber from the SDK callback thread. private void RaiseDataChange(DataChangeEventArgs args) { try { OnDataChange?.Invoke(this, args); } catch (Exception ex) { _logger.LogWarning(ex, "OpcUaClientDriver '{DriverInstanceId}': data subscriber faulted for {Ref}.", _driverInstanceId, args.FullReference); } } private void TransitionTo(HostState newState) { HostState old; lock (_probeLock) { old = _hostState; if (old == newState) return; _hostState = newState; _hostStateChangedUtc = DateTime.UtcNow; } OnHostStatusChanged?.Invoke(this, new HostStatusChangedEventArgs(HostName, old, newState)); } /// /// Synchronous disposal. Detaches the keep-alive /// hook synchronously (no async work on this hot path), then fires the cert-validation /// handler detach. The async session-close is intentionally skipped — it requires a /// live session + network round-trip and is unsafe to block-on from a potentially /// single-threaded context (OPC UA stack thread). The session will be cleaned up by /// the SDK's own finalizer on GC (no sync-over-async). /// public void Dispose() { if (_disposed) return; _disposed = true; // Supersede any init still running, so it discards rather than installs its session. Interlocked.Increment(ref _lifecycleGeneration); lock (_probeLock) { // Detach keep-alive and null Session so in-flight gated callers see null // after their next _gate.WaitAsync — they return BadCommunicationError cleanly. if (_keepAliveHandler is not null && Session is not null) { try { Session.KeepAlive -= _keepAliveHandler; } catch { } } _keepAliveHandler = null; Session = null; } // Detach the cert-validation handler registered during InitializeAsync so the // CertificateValidator (which may be process-shared) doesn't hold a reference to // a disposed driver. if (_certValidationHandler is not null && _certValidatorRef is not null) { try { _certValidatorRef.CertificateValidation -= _certValidationHandler; } catch { } _certValidationHandler = null; _certValidatorRef = null; } // Acquire the gate once so any in-flight gated operation (ReadAsync / WriteAsync / // DiscoverAsync) has definitely released before we dispose the gate. Without this // drain, a background read that calls _gate.Release() after Dispose throws // ObjectDisposedException. foreach (var gate in new[] { _gate, _subscriptionGate }) { try { if (gate.Wait(TimeSpan.FromMilliseconds(500))) gate.Release(); } catch { /* timeout or already disposed — proceed */ } gate.Dispose(); } try { _sessionWorkCts.Cancel(); } catch (ObjectDisposedException) { } SdkTelemetry.Release(_sdkTelemetry); } /// Asynchronously disposes the driver and releases all associated resources. /// A task that represents the asynchronous operation. public async ValueTask DisposeAsync() { if (_disposed) return; _disposed = true; // r2 R2-2: the host's PostStop runs ShutdownAsync then DisposeAsync under one budget. When nothing has happened // since that shutdown started (no init since), a nested shutdown would only re-wait the gate the first one may // still hold and overrun the dispose bound — the running / finished shutdown already covers the teardown. if (Volatile.Read(ref _shutdownStartedGeneration) != Volatile.Read(ref _lifecycleGeneration)) { try { await ShutdownAsync(CancellationToken.None).ConfigureAwait(false); } catch { /* disposal is best-effort */ } } // Detach the cert-validation handler. if (_certValidationHandler is not null && _certValidatorRef is not null) { try { _certValidatorRef.CertificateValidation -= _certValidationHandler; } catch { } _certValidationHandler = null; _certValidatorRef = null; } // Drain the gate before disposal so no in-flight _gate.Release() fires after // Dispose. foreach (var gate in new[] { _gate, _subscriptionGate }) { try { // Short: the host's PostStop gives shutdown + dispose 5 s together, and ShutdownAsync above // already spent its own budget; an op still holding a gate finishes on its own token. if (await gate.WaitAsync(TimeSpan.FromMilliseconds(500)).ConfigureAwait(false)) gate.Release(); } catch { /* timeout or already disposed */ } gate.Dispose(); } try { _sessionWorkCts.Cancel(); } catch (ObjectDisposedException) { } SdkTelemetry.Release(_sdkTelemetry); } }