Merge fix/670-followups: outage Bad stamped at detection; post-recovery replay guard (#670 live-gate F1/F2)

2026-10-11 09:16:03 -04:00
parent 984e069f5d 87e8aff786
commit 699871040a
5 changed files with 277 additions and 5 deletions
@@ -24,6 +24,8 @@ internal sealed class OpcUaClientDiagnostics
    public const string KeepAliveMiss = "keepalive_miss";
    /// <summary>A keep-alive came back Good after a bad one on the same session (an in-place recovery, #670).</summary>
    public const string KeepAliveRecovered = "keepalive_recovered";
+    /// <summary>A late replayed notification (older than the ref's last Good source time) was dropped inside the post-recovery replay window (#670 F2).</summary>
+    public const string StaleNotificationDropped = "stale_notification_dropped";

    private readonly string _driverInstanceId;

@@ -294,6 +294,33 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit

    /// <summary>#670 — the last value delivered per subscribed RawPath, so an outage can publish
    /// <c>BadNoCommunication</c> with the value retained (the same shape the host's own comms-loss fan uses).</summary>
+    /// <summary>
+    ///     #670 follow-up F2 — 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.
+    /// </summary>
+    private readonly System.Collections.Concurrent.ConcurrentDictionary<string, long> _lastGoodSourceTicks =
+        new(StringComparer.Ordinal);
+
+    /// <summary>
+    ///     #670 follow-up F2 — 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
+    ///     the newest Good upstream source time at arming (raised by the recovery read); a notification older than it
+    ///     is dropped. A ref disarms on its first in-order notification, and the whole guard expires at
+    ///     <see cref="_replayGuardUntilTicks"/>. 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.
+    /// </summary>
+    private readonly System.Collections.Concurrent.ConcurrentDictionary<string, long> _replayGuard =
+        new(StringComparer.Ordinal);
+
+    /// <summary>Clock ticks at which the replay guard expires (0 = never armed).</summary>
+    private long _replayGuardUntilTicks;
+
+    /// <summary>Floor of the replay-guard window; the live gate saw the late replay 4–5 s after recovery.</summary>
+    private static readonly TimeSpan ReplayGuardMinimum = TimeSpan.FromSeconds(30);
+
    private readonly System.Collections.Concurrent.ConcurrentDictionary<string, DataValueSnapshot> _lastValues =
        new(StringComparer.Ordinal);

@@ -985,6 +1012,9 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
        // 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();
+        _replayGuard.Clear();
+        Interlocked.Exchange(ref _replayGuardUntilTicks, 0);
        _awaitingRecovery.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
@@ -1724,11 +1754,66 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
            StatusCode: dv.StatusCode.Code,
            SourceTimestampUtc: dv.SourceTimestamp == DateTime.MinValue ? null : dv.SourceTimestamp,
            ServerTimestampUtc: dv.ServerTimestamp == DateTime.MinValue ? DateTime.UtcNow : dv.ServerTimestamp);
+        // 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);
+            _logger.LogDebug(
+                "OpcUaClientDriver '{DriverInstanceId}': dropped a late replayed notification for {Ref} after a recovery " +
+                "(source {Source:O} is older than {Newest:O}).", _driverInstanceId, fullRef, snapshot.SourceTimestampUtc,
+                new DateTime(newest, DateTimeKind.Utc));
+            return;
+        }
+        RecordUpstreamGood(fullRef, snapshot);
        _lastValues[fullRef] = snapshot;
        _awaitingRecovery.TryRemove(fullRef, out _);
        OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, fullRef, snapshot));
    }

+    /// <summary>#670 F2 — remember the last accepted Good upstream source time for <paramref name="fullRef"/>.</summary>
+    private void RecordUpstreamGood(string fullRef, DataValueSnapshot snapshot)
+    {
+        if (snapshot.SourceTimestampUtc is not { } src || !StatusCode.IsGood(snapshot.StatusCode)) return;
+        _lastGoodSourceTicks[fullRef] = src.Ticks;
+    }
+
+    /// <summary>
+    ///     #670 F2 — true when the replay guard is live for <paramref name="fullRef"/> and <paramref name="source"/> is
+    ///     older than its armed time. An in-order (equal or newer) notification disarms the ref; an expired window
+    ///     disarms everything.
+    /// </summary>
+    private bool IsLateReplay(string fullRef, DateTime? source, out long newest)
+    {
+        newest = 0;
+        if (source is not { } src || _replayGuard.IsEmpty) return false;
+        if (Clock.GetUtcNow().UtcDateTime.Ticks >= Interlocked.Read(ref _replayGuardUntilTicks))
+        {
+            _replayGuard.Clear();
+            return false;
+        }
+        if (!_replayGuard.TryGetValue(fullRef, out newest)) return false;
+        if (src.Ticks < newest) return true;
+        _replayGuard.TryRemove(fullRef, out _);
+        return false;
+    }
+
+    /// <summary>
+    ///     #670 F2 — 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).
+    /// </summary>
+    private void ArmReplayGuard(DateTime now)
+    {
+        _replayGuard.Clear();
+        foreach (var kv in _lastGoodSourceTicks) _replayGuard[kv.Key] = kv.Value;
+        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;
+        }
+        Interlocked.Exchange(ref _replayGuardUntilTicks, (now + window).Ticks);
+    }
+
    /// <summary>
    ///     #670 — advance <see cref="DriverHealth.LastSuccessfulRead"/> 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
@@ -2594,6 +2679,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
                UpdateHealth(_ => 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);
+                ArmReplayGuard(now);
                _diag.Record(OpcUaClientDiagnostics.KeepAliveRecovered);
                _logger.LogInformation(
                    "OpcUaClientDriver '{DriverInstanceId}': upstream keep-alive recovered on the same session; " +
@@ -2705,8 +2791,11 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
                var fullRef = RefOf(pair.Item);
                _lastValues.TryGetValue(fullRef, out var last);
                _awaitingRecovery[fullRef] = 0;
+                // 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, last?.SourceTimestampUtc, now)));
+                    last?.Value, StatusBadNoCommunication, now, now)));
            }
        }
        AnnotateAlarmQuality(StatusBadNoCommunication, now);
@@ -2742,6 +2831,10 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
                    if (!Current()) return;
                    if (!_awaitingRecovery.TryRemove(refs[i], out _)) continue; // a notification got there first
                    _lastValues[refs[i]] = results[i];
+                    RecordUpstreamGood(refs[i], results[i]);
+                    // The read is upstream data published inside the replay window: a replay older than it is late too.
+                    if (results[i].SourceTimestampUtc is { } readAt && StatusCode.IsGood(results[i].StatusCode))
+                        _replayGuard.AddOrUpdate(refs[i], readAt.Ticks, (_, prior) => Math.Max(prior, readAt.Ticks));
                    foreach (var rs in _subscriptions.Values)
                        if (rs.ItemHandlers.Any(p => RefOf(p.Item) == refs[i]))
                            RaiseDataChange(new DataChangeEventArgs(rs.Handle, refs[i], results[i]));