diff --git a/docs/drivers/OpcUaClient.md b/docs/drivers/OpcUaClient.md index 8cc07a27..59cf4fa2 100644 --- a/docs/drivers/OpcUaClient.md +++ b/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 diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDiagnostics.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDiagnostics.cs index b5afca01..966dd602 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDiagnostics.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDiagnostics.cs @@ -24,7 +24,7 @@ internal sealed class OpcUaClientDiagnostics public const string KeepAliveMiss = "keepalive_miss"; /// A keep-alive came back Good after a bad one on the same session (an in-place recovery, #670). public const string KeepAliveRecovered = "keepalive_recovered"; - /// A late replayed notification (older than the ref's last Good source time) was dropped inside the post-recovery replay window (#670 F2). + /// 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). public const string StaleNotificationDropped = "stale_notification_dropped"; private readonly string _driverInstanceId; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs index 76a7c64f..fe76821e 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs @@ -305,19 +305,27 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// /// #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 - /// . 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 + /// ; 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 = never armed). + /// 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(); + /// 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); @@ -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 } /// - /// #670 F2 — true when the replay guard is live for and 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 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; - 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; + } } /// @@ -1803,15 +1827,18 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// 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); + } } /// diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientDiagnosticsTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientDiagnosticsTests.cs index fa296091..de076b11 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientDiagnosticsTests.cs +++ b/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(); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientSessionLossTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientSessionLossTests.cs index 9d6b3b8d..992f5462 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientSessionLossTests.cs +++ b/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 Session, ManualClock Clock) NewConnectedDriver() + private static (OpcUaClientDriver Driver, Mock 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 /// /// #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). /// [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 }); + } + + /// + /// #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. + /// + [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(); + 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); + } + + /// + /// 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. + /// + [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(); + 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 }); + } + + /// A notification with no source timestamp passes through the open guard and does not move it. + [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(); + 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 }); } ///