Merge fix/670-f2-regate: keep the post-recovery replay guard armed for the whole window (#670 live-gate R2)

docs/drivers/OpcUaClient.md

@@ -74,9 +74,13 @@ The session keep-alive (`KeepAliveInterval`, default 5 s) is the liveness signal
   unchanged value, and a `(value, status)` dedup downstream lets the identical value through), alarm quality goes
   back Good and a `ConditionRefresh` re-reports anything missed. A notification that lands first wins over the read.
   For a bounded **replay window** after such a recovery — max(30 s, 2 x the longest subscription keep-alive window) —
-  a notification whose source time is OLDER than the ref's last Good upstream source time (raised by the recovery
-  read) is dropped: the SDK was seen delivering a missed one 4–5 s late, which regressed the node (live-gate F2).
-  A ref disarms on its first in-order (equal or newer) notification. Outside that window every notification is
+  each ref is guarded by a high-water mark — its last Good upstream source time, raised by the recovery read and
+  advanced by every notification accepted in the window — and a notification whose source time is at or BELOW it
+  (equal included: OPC UA moves the source timestamp on every value or status change, so an equal one is a replay)
+  is dropped: the SDK was seen delivering a missed one 4–6 s late, which regressed the node (live-gate F2). The
+  guard holds for the WHOLE window — fresh notifications arrive ~1 s after recovery, before the late replay, so a
+  ref is never disarmed early (the F2 re-gate failed on exactly that). A notification with no source time passes
+  without moving the mark. Outside that window every notification is
   accepted whatever its timestamp, so an upstream clock that steps backwards (NTP, a PLC clock reset) never
   freezes a node. The outage Bad never moves the comparison; a re-initialise resets it.
 - **The session does not come back** (an upstream restart: the server forgot the session) ⇒ the driver stays

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDiagnostics.cs

@@ -24,7 +24,7 @@ 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>
+    /// <summary>A late replayed notification (source time at or below the ref's high-water mark) was dropped inside the post-recovery replay window (#670 F2).</summary>
     public const string StaleNotificationDropped = "stale_notification_dropped";
 
     private readonly string _driverInstanceId;

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs

@@ -305,19 +305,27 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
 
     /// <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
+    ///     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
+    ///     <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>
+    /// <summary>Clock ticks at which the replay guard expires (0 = not armed).</summary>
     private long _replayGuardUntilTicks;
 
+    /// <summary>Serialises arming against expiry, so an expiry racing a re-arm cannot wipe the new seeds.</summary>
+    private readonly object _replayGuardLock = new();
+
     /// <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);
 
@@ -1760,7 +1768,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
             _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,
+                "(source {Source:O} is not newer than {Newest:O}).", _driverInstanceId, fullRef, snapshot.SourceTimestampUtc,
                 new DateTime(newest, DateTimeKind.Utc));
             return;
         }
@@ -1778,23 +1786,39 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
     }
 
     /// <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.
+    ///     #670 F2 — true when the replay window is open and <paramref name="source"/> is at or below
+    ///     <paramref name="fullRef"/>'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.
     /// </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))
+        var until = Interlocked.Read(ref _replayGuardUntilTicks);
+        if (until == 0) return false;
+        if (Clock.GetUtcNow().UtcDateTime.Ticks >= until)
         {
-            _replayGuard.Clear();
+            lock (_replayGuardLock)
+            {
+                if (Interlocked.Read(ref _replayGuardUntilTicks) == until)
+                {
+                    Interlocked.Exchange(ref _replayGuardUntilTicks, 0);
+                    _replayGuard.Clear();
+                }
+            }
             return false;
         }
-        if (!_replayGuard.TryGetValue(fullRef, out newest)) return false;
-        if (src.Ticks < newest) return true;
-        _replayGuard.TryRemove(fullRef, out _);
-        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;
+        }
     }
 
     /// <summary>
@@ -1803,15 +1827,18 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
     /// </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);
+        lock (_replayGuardLock)
+        {
+            _replayGuard.Clear();
+            foreach (var kv in _lastGoodSourceTicks) _replayGuard[kv.Key] = kv.Value;
+            Interlocked.Exchange(ref _replayGuardUntilTicks, (now + window).Ticks);
+        }
     }
 
     /// <summary>

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientDiagnosticsTests.cs

@@ -14,7 +14,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests;
 [Trait("Category", "Unit")]
 public sealed class OpcUaClientDiagnosticsTests
 {
-    private sealed class Capture : IDisposable
+    internal sealed class Capture : IDisposable
     {
         private readonly MeterListener _listener = new();
         private readonly object _gate = new();

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientSessionLossTests.cs

@@ -46,9 +46,10 @@ public sealed class OpcUaClientSessionLossTests
         return mock;
     }
 
-    private static (OpcUaClientDriver Driver, Mock<ISession> Session, ManualClock Clock) NewConnectedDriver()
+    private static (OpcUaClientDriver Driver, Mock<ISession> Session, ManualClock Clock) NewConnectedDriver(
+        string driverInstanceId = "opcua-670")
     {
-        var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions { RawTags = Tags }, "opcua-670");
+        var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions { RawTags = Tags }, driverInstanceId);
         var session = NewSession();
         var clock = new ManualClock(new DateTimeOffset(2026, 10, 11, 12, 0, 0, TimeSpan.Zero));
         drv.Clock = clock;
@@ -518,11 +519,11 @@ public sealed class OpcUaClientSessionLossTests
     /// <summary>
     ///     #670 follow-up F2 (live gate): after a same-session recovery the SDK delivered one notification late with
     ///     an OLDER source time, and the node regressed to it. Inside the post-recovery replay window an older
-    ///     notification is dropped; an equal one (a status-only change) passes and disarms the ref, after which the
-    ///     ref is back to accepting anything.
+    ///     notification is dropped, and so is one carrying the exact source time already accepted (OPC UA moves the
+    ///     source timestamp on every value OR status change, so an equal one is a replay, not a status-only change).
     /// </summary>
     [Fact]
-    public async Task After_a_recovery_a_late_older_notification_is_dropped_until_the_ref_is_back_in_order()
+    public async Task After_a_recovery_a_late_older_or_equal_notification_is_dropped()
     {
         var (drv, session, clock) = NewConnectedDriver();
         using var _ = drv;
@@ -536,12 +537,93 @@ public sealed class OpcUaClientSessionLossTests
 
         drv.DeliverNotificationForTest(handle, SpeedRef, Good(667, t0.AddSeconds(1)));          // late replay
         drv.DeliverNotificationForTest(handle, SpeedRef,
-            new DataValue(new Variant(668), StatusCodes.UncertainLastUsableValue, t0.AddSeconds(2), t0.AddSeconds(2))); // same ts, status-only
+            new DataValue(new Variant(668), StatusCodes.UncertainLastUsableValue, t0.AddSeconds(2), t0.AddSeconds(2))); // same ts
         drv.DeliverNotificationForTest(handle, SpeedRef, Good(669, t0.AddSeconds(3)));
-        drv.DeliverNotificationForTest(handle, SpeedRef, Good(665, t0));                        // disarmed: accepted
+        drv.DeliverNotificationForTest(handle, SpeedRef, Good(665, t0));                        // still guarded
 
-        seen.Select(e => e.Snapshot.Value).ShouldBe(new object[] { 668, 669, 665 });
-        seen[0].Snapshot.StatusCode.ShouldBe((uint)StatusCodes.UncertainLastUsableValue);
+        seen.Select(e => e.Snapshot.Value).ShouldBe(new object[] { 669 });
+    }
+
+    /// <summary>
+    ///     #670 F2 re-gate (live gate FAILED on the per-ref disarm): fresh notifications arrived ~1 s after recovery
+    ///     and disarmed each ref, then the SDK's late replay arrived 4–6 s later and regressed the node (value 670 with
+    ///     a pre-outage source time delivered after newer values). The guard must hold for the WHOLE window against the
+    ///     newest ACCEPTED source time per ref — older, equal-to-the-recovery-read and equal-to-the-last-accepted
+    ///     replays are all dropped, and counted.
+    /// </summary>
+    [Fact]
+    public async Task A_late_replay_after_a_fresh_in_order_notification_is_still_dropped_inside_the_window()
+    {
+        var instance = $"opcua-670-regate-{Guid.NewGuid():N}";
+        using var drops = new OpcUaClientDiagnosticsTests.Capture(instance);
+        var (drv, session, clock) = NewConnectedDriver(instance);
+        using var _ = drv;
+        var handle = drv.AddSubscriptionForTest([SpeedRef]);
+        var t0 = clock.Now.UtcDateTime;
+        drv.DeliverNotificationForTest(handle, SpeedRef, Good(100, t0.AddSeconds(-5)));
+        SetupReadAt(session, 105, t0);                                // recovery read seeds T0
+        await RecoverInPlaceAsync(drv);
+        var seen = new List<DataChangeEventArgs>();
+        drv.OnDataChange += (_, e) => seen.Add(e);
+
+        clock.Advance(TimeSpan.FromSeconds(1));
+        drv.DeliverNotificationForTest(handle, SpeedRef, Good(116, t0.AddSeconds(6)));   // fresh, in order: T1
+        clock.Advance(TimeSpan.FromSeconds(5));
+        drv.DeliverNotificationForTest(handle, SpeedRef, Good(109, t0.AddSeconds(3)));   // late replay, older than T1
+        drv.DeliverNotificationForTest(handle, SpeedRef, Good(105, t0));                 // replay == T0 (recovery read)
+        drv.DeliverNotificationForTest(handle, SpeedRef, Good(116, t0.AddSeconds(6)));   // replay == T1
+        drv.DeliverNotificationForTest(handle, SpeedRef, Good(670, t0.AddSeconds(-10))); // pre-outage replay
+        drv.DeliverNotificationForTest(handle, SpeedRef, Good(117, t0.AddSeconds(7)));   // fresh again
+
+        seen.Select(e => e.Snapshot.Value).ShouldBe(new object[] { 116, 117 });
+        drops[OpcUaClientDiagnostics.StaleNotificationDropped].ShouldBe(4);
+    }
+
+    /// <summary>
+    ///     The re-gated guard still expires: after the window an older source time (an upstream clock step-back) is
+    ///     accepted, even though the ref accepted newer notifications inside the window.
+    /// </summary>
+    [Fact]
+    public async Task After_the_window_a_clock_step_back_is_accepted_even_after_in_window_notifications()
+    {
+        var (drv, session, clock) = NewConnectedDriver();
+        using var _ = drv;
+        var handle = drv.AddSubscriptionForTest([SpeedRef]);
+        var t0 = clock.Now.UtcDateTime;
+        drv.DeliverNotificationForTest(handle, SpeedRef, Good(1, t0.AddSeconds(-5)));
+        SetupReadAt(session, 2, t0);
+        await RecoverInPlaceAsync(drv);
+        var seen = new List<DataChangeEventArgs>();
+        drv.OnDataChange += (_, e) => seen.Add(e);
+
+        drv.DeliverNotificationForTest(handle, SpeedRef, Good(3, t0.AddSeconds(20)));
+        clock.Advance(TimeSpan.FromSeconds(31));
+        drv.DeliverNotificationForTest(handle, SpeedRef, Good(4, t0.AddSeconds(1)));     // clock stepped back
+        drv.DeliverNotificationForTest(handle, SpeedRef, Good(5, t0));                   // and again
+
+        seen.Select(e => e.Snapshot.Value).ShouldBe(new object[] { 3, 4, 5 });
+    }
+
+    /// <summary>A notification with no source timestamp passes through the open guard and does not move it.</summary>
+    [Fact]
+    public async Task A_notification_without_a_source_time_passes_the_guard_and_does_not_advance_it()
+    {
+        var (drv, session, clock) = NewConnectedDriver();
+        using var _ = drv;
+        var handle = drv.AddSubscriptionForTest([SpeedRef]);
+        var t0 = clock.Now.UtcDateTime;
+        drv.DeliverNotificationForTest(handle, SpeedRef, Good(1, t0.AddSeconds(-5)));
+        SetupReadAt(session, 2, t0);
+        await RecoverInPlaceAsync(drv);
+        var seen = new List<DataChangeEventArgs>();
+        drv.OnDataChange += (_, e) => seen.Add(e);
+
+        drv.DeliverNotificationForTest(handle, SpeedRef,
+            new DataValue(new Variant(3), StatusCodes.Good, DateTime.MinValue, t0));      // no source time
+        drv.DeliverNotificationForTest(handle, SpeedRef, Good(4, t0.AddSeconds(-1)));     // still guarded at T0
+        drv.DeliverNotificationForTest(handle, SpeedRef, Good(5, t0.AddSeconds(1)));
+
+        seen.Select(e => e.Snapshot.Value).ShouldBe(new object[] { 3, 5 });
     }
 
     /// <summary>