fix(opcuaclient): native-alarm ConditionRefresh + transition table, branch, dedup and thread-safety fixes (#619 re-review)

Re-review of 3de2cc3b/454775c6: - AlarmTransitions could not reach (inactive, unacked): (active,acked)->(inactive,unacked) emitted Clear (host stayed acked) and a first-seen cleared-unacked condition emitted nothing. The table is now total; an exhaustive test replays every (prior, target) pair through the projector's semantics. - The per-RawPath read-compute-emit was a lock-free TryGetValue/set across SDK publish threads; it now runs under one lock with the projection and emission. - Condition BRANCH events (non-null BranchId) were projected (flipping the trunk) and remembered as the ack target (acking the branch); they are dropped. BranchId is select clause 9. - EventId is copied, not aliased from the decoded notification. - An unchanged state re-emitted Unspecified; it now emits nothing unless message/severity changed (an Unspecified projects from the host projector's default after a redeploy reset it). - Exact upstream severity rides OpcUaSeverity (#626) instead of being re-bucketed. - ShutdownAsync no longer forgets the last-told state: a reinit re-subscribes the same set and the refresh would otherwise re-raise every acked alarm. A CHANGED set (host rebuilt) forgets it. ConditionRefresh: called after the alarm event item is created, after an in-place reconnect (subscriptions found on the new session by display name, since a recreate clones them) and on RefreshRequired. Refresh markers are never projected. BadNotSupported/BadMethodInvalid/... logs once. Wire test: a condition already active before subscribe is reported Raise; a release+resubscribe refresh re-raises nothing (falsified by disabling the refresh call). Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
2026-10-08 17:13:54 -04:00
parent 454775c67c
commit 8f5b265db8
4 changed files with 499 additions and 50 deletions
@@ -102,8 +102,19 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
    /// <summary>#619 review: the (Active, Acked) state last told to the host per reported condition id, from
    /// which <see cref="AlarmTransitions"/> picks the Kind sequence that moves the host's projector to the
    /// upstream event's state.</summary>
-    private readonly System.Collections.Concurrent.ConcurrentDictionary<string, (bool Active, bool Acked)> _alarmStateByRawPath =
-        new(StringComparer.Ordinal);
+    private readonly Dictionary<string, AlarmToldState> _alarmStateByRawPath = new(StringComparer.Ordinal);
+
+    /// <summary>#619 re-review: serialises the read-compute-emit over <see cref="_alarmStateByRawPath"/>. 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.</summary>
+    private readonly object _alarmStateLock = new();
+
+    /// <summary>#619 re-review: the ref set the last <see cref="SubscribeAlarmsAsync"/> carried. Survives
+    /// <see cref="ShutdownAsync"/> deliberately — see <see cref="BeginAlarmSubscription"/>.</summary>
+    private HashSet<string>? _lastAlarmRefSet;
+
+    /// <summary>Latched once the upstream server has refused ConditionRefresh as unsupported (logged once).</summary>
+    private int _conditionRefreshUnsupportedLogged;
    private long _nextAlarmSubscriptionId;

    /// <summary>Occurs when an alarm event is received from the remote OPC UA server.</summary>
@@ -891,8 +902,11 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
            catch { /* best-effort */ }
        }
        _alarmSubscriptions.Clear();
-        _alarmConditionByRawPath.Clear();
-        _alarmStateByRawPath.Clear();
+        // #619 re-review: 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.

        await ReleaseSessionAsync(cancellationToken).ConfigureAwait(false);
        _namespaceMap = null;
@@ -1637,6 +1651,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
    private const int AlarmFieldConditionId = 6;
    private const int AlarmFieldActiveState = 7;
    private const int AlarmFieldAckedState = 8;
+    private const int AlarmFieldBranchId = 9;

    /// <inheritdoc />
    public async Task<IAlarmSubscriptionHandle> SubscribeAlarmsAsync(
@@ -1664,15 +1679,11 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
                && !NamespaceMap.TryResolve(sessionForResolve, raw, out live)) return null;
            return AlarmIndexKey(mapForResolve, live);
        });
-        // Forget ack targets + last-told states for alarm tags no longer subscribed (removed/renamed RawPaths).
-        foreach (var known in _alarmConditionByRawPath.Keys)
-            if (!sourceNodeIds.Contains(known, StringComparer.Ordinal)) _alarmConditionByRawPath.TryRemove(known, out _);
-        foreach (var known in _alarmStateByRawPath.Keys)
-            if (!sourceNodeIds.Contains(known, StringComparer.Ordinal)) _alarmStateByRawPath.TryRemove(known, out _);
+        BeginAlarmSubscription(sourceNodeIds);

        var subscription = new Subscription(telemetry: null!, new SubscriptionOptions
        {
-            DisplayName = $"opcua-alarm-sub-{id}",
+            DisplayName = $"{AlarmSubscriptionNamePrefix}{id}",
            PublishingInterval = 500,   // 500ms — alarms don't need fast polling; the server pushes
            KeepAliveCount = _options.SubscriptionKeepAliveCount,
            LifetimeCount = _options.SubscriptionLifetimeCount,
@@ -1721,6 +1732,14 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
            BrowsePath = [new QualifiedName(BrowseNames.AckedState), new QualifiedName(BrowseNames.Id)],
            AttributeId = Attributes.Value,
        });
+        // #619 re-review: 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,
+        });

        await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
        try
@@ -1757,6 +1776,11 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
            await subscription.CreateItemsAsync(cancellationToken).ConfigureAwait(false);

            _alarmSubscriptions[id] = new RemoteAlarmSubscription(subscription, handle, eventItem, notifHandler);
+
+            // #619: 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 { _gate.Release(); }

@@ -1898,30 +1922,194 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
        MonitoredItemNotificationEventArgs args)
    {
        if (args.NotificationValue is not EventFieldList efl) return;
-        var e = ProjectEvent(efl.EventFields, sourceFilter, handle,
-            n => AlarmIndexKey(_namespaceMap, n), (raw, target) => _alarmConditionByRawPath[raw] = target,
-            out var state);
-        if (e is null) return;
-        if (state is not { } target)
+        RouteAlarmEvent(efl.EventFields, sourceFilter, handle);
+    }
+
+    /// <summary>
+    ///     #619 re-review: open (or re-open) the host's alarm subscription over <paramref name="sourceNodeIds"/>.
+    ///     A CHANGED set means the host rebuilt its address space (an alarm-bearing delta always does), recreating
+    ///     every condition node inactive+acked — so every last-told state is forgotten and the post-subscribe
+    ///     ConditionRefresh re-reports from that 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.
+    /// </summary>
+    /// <param name="sourceNodeIds">The refs the host is subscribing.</param>
+    internal void BeginAlarmSubscription(IReadOnlyList<string> sourceNodeIds)
+    {
+        var set = new HashSet<string>(sourceNodeIds, StringComparer.Ordinal);
+        lock (_alarmStateLock)
        {
-            // Not a condition (no ActiveState/AckedState) — a message/severity-only update, as before.
-            OnAlarmEvent?.Invoke(this, e);
+            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 _);
+        }
+    }
+
+    /// <summary>
+    ///     #619 — 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.
+    /// </summary>
+    /// <param name="fields">The event's fields, in select-clause order.</param>
+    /// <param name="sourceFilter">The subscription's upstream-key → RawPath index.</param>
+    /// <param name="handle">The subscription handle to stamp.</param>
+    internal void RouteAlarmEvent(
+        IReadOnlyList<Variant> fields, IReadOnlyDictionary<string, string> sourceFilter, IAlarmSubscriptionHandle handle)
+    {
+        if (IsRefreshRequired(fields))
+        {
+            // Part 9 §5.11.7: the server lost events (queue overflow / restart) and asks for a refresh.
+            _ = Task.Run(() => RefreshAllAlarmSubscriptionsAsync(CancellationToken.None));
            return;
        }

-        // Fresh prior = the host projector's own default (inactive, acked).
-        var prior = _alarmStateByRawPath.TryGetValue(e.ConditionId, out var p) ? p : (Active: false, Acked: true);
-        _alarmStateByRawPath[e.ConditionId] = target;
-        foreach (var kind in AlarmTransitions(prior, target))
-            OnAlarmEvent?.Invoke(this, e with { Kind = kind });
+        // 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 prior = _alarmStateByRawPath.TryGetValue(e.ConditionId, out var p)
+                ? p
+                : new AlarmToldState(false, true, null, null);
+            _alarmStateByRawPath[e.ConditionId] = told;
+
+            var kinds = AlarmTransitions((prior.Active, prior.Acked), target);
+            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];
+            }
+            foreach (var kind in kinds)
+                OnAlarmEvent?.Invoke(this, e with { Kind = kind });
+        }
    }

+    /// <summary>#619 re-review — 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.</summary>
+    /// <param name="Active">ActiveState/Id last told.</param>
+    /// <param name="Acked">AckedState/Id last told.</param>
+    /// <param name="Message">Message last told (null before any event).</param>
+    /// <param name="Severity">Exact severity last told (null before any event).</param>
+    private readonly record struct AlarmToldState(bool Active, bool Acked, string? Message, int? Severity);
+
+    /// <summary>True for a <c>RefreshRequiredEventType</c> event (the server asks the client to ConditionRefresh).</summary>
+    /// <param name="fields">The event's fields.</param>
+    /// <returns>Whether the event is a RefreshRequired marker.</returns>
+    internal static bool IsRefreshRequired(IReadOnlyList<Variant> fields) =>
+        fields.Count > AlarmFieldEventType && fields[AlarmFieldEventType].Value is NodeId t
+        && t == ObjectTypeIds.RefreshRequiredEventType;
+
+    /// <summary>True for the three ConditionRefresh marker event types — never alarms.</summary>
+    /// <param name="eventType">The event's EventType.</param>
+    /// <returns>Whether it is a refresh marker.</returns>
+    private static bool IsRefreshMarker(object? eventType) =>
+        eventType is NodeId t
+        && (t == ObjectTypeIds.RefreshStartEventType || t == ObjectTypeIds.RefreshEndEventType
+            || t == ObjectTypeIds.RefreshRequiredEventType);
+
+    /// <summary>A ConditionRefresh refusal meaning "this server does not do refresh" (tolerated, logged once).</summary>
+    /// <param name="code">The refusal status.</param>
+    /// <returns>Whether the server simply lacks ConditionRefresh.</returns>
+    internal static bool IsConditionRefreshUnsupported(StatusCode code) =>
+        code == StatusCodes.BadNotSupported || code == StatusCodes.BadMethodInvalid
+        || code == StatusCodes.BadNotImplemented || code == StatusCodes.BadServiceUnsupported
+        || code == StatusCodes.BadNodeIdUnknown;
+
+    /// <summary>
+    ///     #619 — call ConditionRefresh for <paramref name="subscription"/> so conditions already active upstream
+    ///     are reported. Never throws: an unsupported server is logged once, any other failure warns.
+    /// </summary>
+    /// <param name="subscription">The (created) alarm subscription.</param>
+    /// <param name="cancellationToken">Cancellation.</param>
+    /// <returns>A task completing when the refresh call returns.</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.
+        }
+    }
+
+    /// <summary>
+    ///     #619 — ConditionRefresh every alarm subscription on the current session: after an in-place reconnect
+    ///     (transferred or SDK-recreated subscriptions — matched by display name, since a recreate clones the
+    ///     <see cref="Subscription"/> objects) and on a RefreshRequired event. Events missed during the outage
+    ///     reach the host through the normal deduplicated path.
+    /// </summary>
+    /// <param name="cancellationToken">Cancellation.</param>
+    /// <returns>A task completing when every refresh call has returned.</returns>
+    private async Task RefreshAllAlarmSubscriptionsAsync(CancellationToken cancellationToken)
+    {
+        if (_alarmSubscriptions.IsEmpty) return;
+        try
+        {
+            await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
+        }
+        catch (ObjectDisposedException) { return; }
+        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 { _gate.Release(); } catch (ObjectDisposedException) { }
+        }
+    }
+
+    /// <summary>Display-name prefix of every alarm subscription (how a refresh sweep finds them on a session).</summary>
+    private const string AlarmSubscriptionNamePrefix = "opcua-alarm-sub-";
+
    /// <summary>
    ///     #619 review — the <see cref="AlarmTransitionKind"/> sequence that moves the host's
    ///     <c>NativeAlarmProjector</c> (which knows only kinds: Raise ⇒ active+unacked, Acknowledge ⇒ acked,
    ///     Clear ⇒ inactive keeping acked) from <paramref name="prior"/> to <paramref name="target"/>. A single
    ///     upstream event can need two (an alarm that went active and was acknowledged between two notifications).
-    ///     No state change ⇒ one <see cref="AlarmTransitionKind.Unspecified"/> (message/severity refresh).
+    ///     No state change ⇒ empty (the caller decides whether a message/severity change warrants an Unspecified).
    /// </summary>
    /// <param name="prior">The state last told to the host.</param>
    /// <param name="target">The upstream event's ActiveState/Id + AckedState/Id.</param>
@@ -1929,28 +2117,25 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
    internal static IReadOnlyList<AlarmTransitionKind> AlarmTransitions(
        (bool Active, bool Acked) prior, (bool Active, bool Acked) target)
    {
-        var kinds = new List<AlarmTransitionKind>(2);
-        if (target.Active && !prior.Active)
+        if (prior == target) return [];
+        return target switch
        {
-            kinds.Add(AlarmTransitionKind.Raise);
-            if (target.Acked) kinds.Add(AlarmTransitionKind.Acknowledge);
-        }
-        else if (!target.Active && prior.Active)
-        {
-            kinds.Add(AlarmTransitionKind.Clear);
-            if (target.Acked && !prior.Acked) kinds.Add(AlarmTransitionKind.Acknowledge);
-        }
-        else if (target.Acked && !prior.Acked)
-        {
-            kinds.Add(AlarmTransitionKind.Acknowledge);
-        }
-        else if (target.Active && !target.Acked && prior.Acked)
-        {
-            kinds.Add(AlarmTransitionKind.Retrigger);
-        }
-
-        if (kinds.Count == 0) kinds.Add(AlarmTransitionKind.Unspecified);
-        return kinds;
+            // Active+unacked: a fresh raise, or a re-arm of an active-acked condition.
+            (true, false) => [prior.Active ? AlarmTransitionKind.Retrigger : AlarmTransitionKind.Raise],
+            // Active+acked: raise first when it was inactive (Raise always lands unacked).
+            (true, true) => prior.Active
+                ? [AlarmTransitionKind.Acknowledge]
+                : [AlarmTransitionKind.Raise, AlarmTransitionKind.Acknowledge],
+            // 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):
+            // the only kinds that un-ack are Raise/Retrigger, so pass through active, then Clear.
+            (false, false) => prior.Active
+                ? prior.Acked ? [AlarmTransitionKind.Retrigger, AlarmTransitionKind.Clear] : [AlarmTransitionKind.Clear]
+                : [AlarmTransitionKind.Raise, AlarmTransitionKind.Clear],
+        };
    }

    /// <summary>
@@ -1997,6 +2182,12 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
    {
        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)
@@ -2029,7 +2220,9 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit

        if (sourceIndex.Count != 0 && upstreamCondition.Length > 0)
            rememberCondition(reported, new AlarmAckTarget(
-                upstreamCondition, fields[AlarmFieldEventId].Value as byte[] ?? Array.Empty<byte>()));
+                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<byte>()));

        return new AlarmEventArgs(
            SubscriptionHandle: handle,
@@ -2038,7 +2231,9 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
            AlarmType: eventType,
            Message: message,
            Severity: MapSeverity(severity),
-            SourceTimestampUtc: time);
+            SourceTimestampUtc: time,
+            // #626: the upstream severity is already the exact Part 9 value — carry it, don't re-bucket.
+            OpcUaSeverity: severity == 0 ? null : severity);
    }

    /// <summary>#619 — what Acknowledge needs for one RawPath: the upstream condition node (server-stable
@@ -2562,6 +2757,10 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
            TransitionTo(HostState.Running);
            _health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null);
            _diag.Record(OpcUaClientDiagnostics.ReconnectSuccess);
+            // #619: transitions during the outage were missed (a recreated subscription starts empty; a transferred
+            // one may have overflowed). ConditionRefresh re-reports current states; unchanged ones dedup to nothing.
+            if (!_alarmSubscriptions.IsEmpty)
+                _ = Task.Run(() => RefreshAllAlarmSubscriptionsAsync(CancellationToken.None));
            return;
        }