diff --git a/docs/drivers/OpcUaClient.md b/docs/drivers/OpcUaClient.md index 90363537..8cc07a27 100644 --- a/docs/drivers/OpcUaClient.md +++ b/docs/drivers/OpcUaClient.md @@ -73,10 +73,12 @@ The session keep-alive (`KeepAliveInterval`, default 5 s) is the liveness signal the refs still Bad are re-read once and published Good (a server whose subscription survived does not resend an 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. - A notification whose source time is OLDER than the newest Good upstream source time already published for its ref - (notification or recovery read) is dropped — the SDK can deliver a missed one seconds late after a same-session - recovery, which used to regress the node (live-gate F2). Equal timestamps pass (status-only changes); the outage - Bad never moves the comparison; a re-initialise resets it. + 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 + 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 `Degraded` with its read stamp frozen, which is exactly the #527 comms-loss watchdog's trip condition. After 90 s `DriverInstanceActor` runs `ShutdownAsync` + `InitializeAsync` (retried every 10 s while the server is down) and @@ -234,7 +236,7 @@ Session/publish events are published on the shared OTel meter as the counter `otopcua.driver.opcuaclient.events`, tagged `driver_instance` (the driver instance id) and `event` = `publish_response`, `notification` (items), `data_change`, `bad_status_value`, `publish_error`, `keepalive_miss`, `keepalive_recovered` (a bad keep-alive followed by a good one on the same session), -`stale_notification_dropped` (an out-of-order late notification, see above). A climbing +`stale_notification_dropped` (a late replay dropped inside the post-recovery window, see above). A climbing `keepalive_miss` / `keepalive_recovered` rate is the signature of a flapping upstream (`reconnect_*` were removed with the reconnect handler, #670). They are not part of `DriverHealth` or the `/hosts` panel. 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 6b4b1418..b5afca01 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 notification older than the newest Good source time already published for its ref was dropped (#670 F2). + /// A late replayed notification (older than the ref's last Good source time) 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 81e3b7eb..76a7c64f 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs @@ -295,14 +295,32 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// #670 — 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). /// - /// #670 follow-up F2 — per RawPath, the newest SOURCE timestamp (ticks) of a Good value that came from the - /// upstream (a notification or the recovery read). A notification older than it is a late replay — the SDK can - /// deliver a missed notification seconds after a same-session recovery — and is dropped rather than regressing - /// the node. The driver's own outage Bad never writes it (its stamp is the detection time, not upstream data). + /// #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. /// private readonly System.Collections.Concurrent.ConcurrentDictionary _lastGoodSourceTicks = new(StringComparer.Ordinal); + /// + /// #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 + /// 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). + private long _replayGuardUntilTicks; + + /// 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); @@ -995,6 +1013,8 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit // 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 @@ -1734,14 +1754,13 @@ 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: never let a late, older notification regress the node. Equal timestamps pass (a status-only change). - if (snapshot.SourceTimestampUtc is { } src - && _lastGoodSourceTicks.TryGetValue(fullRef, out var newest) && src.Ticks < newest) + // 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 notification for {Ref} (source {Source:O} is " + - "older than the newest Good source {Newest:O}).", _driverInstanceId, fullRef, src, + "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; } @@ -1751,11 +1770,48 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, fullRef, snapshot)); } - /// #670 F2 — remember the newest Good upstream source time for . + /// #670 F2 — 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.AddOrUpdate(fullRef, src.Ticks, (_, prior) => Math.Max(prior, src.Ticks)); + _lastGoodSourceTicks[fullRef] = src.Ticks; + } + + /// + /// #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. + /// + 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; + } + + /// + /// #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). + /// + 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); } /// @@ -2623,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; " + @@ -2775,6 +2832,9 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit 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])); 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 386a4852..9d6b3b8d 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 @@ -496,13 +496,60 @@ public sealed class OpcUaClientSessionLossTests bad.ServerTimestampUtc.ShouldBe(clock.Now.UtcDateTime); } + private static void SetupReadAt(Mock session, object value, DateTime sourceTime) => + session.Setup(s => s.ReadAsync( + It.IsAny(), It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny())) + .ReturnsAsync(new ReadResponse + { + ResponseHeader = new ResponseHeader(), + Results = [new DataValue(new Variant(value), StatusCodes.Good, sourceTime, sourceTime)], + DiagnosticInfos = [], + }); + + /// Drives a same-session outage + recovery (bad → good keep-alive) and waits for the recovery read. + private static async Task RecoverInPlaceAsync(OpcUaClientDriver drv) + { + drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); + drv.RaiseKeepAliveForTest(StatusCodes.Good); + await drv.RecoveryTaskForTest; + } + /// /// #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. A notification older than the last Good source time - /// published for its ref is dropped; an equal one (a status-only change) and a newer one pass. + /// 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. /// [Fact] - public void A_notification_older_than_the_last_good_source_time_is_dropped() + public async Task After_a_recovery_a_late_older_notification_is_dropped_until_the_ref_is_back_in_order() + { + var (drv, session, clock) = NewConnectedDriver(); + using var _ = drv; + var handle = drv.AddSubscriptionForTest([SpeedRef]); + var t0 = clock.Now.UtcDateTime; + drv.DeliverNotificationForTest(handle, SpeedRef, Good(668, t0.AddSeconds(2))); + SetupReadAt(session, 668, t0.AddSeconds(2)); + await RecoverInPlaceAsync(drv); + var seen = new List(); + drv.OnDataChange += (_, e) => seen.Add(e); + + 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 + drv.DeliverNotificationForTest(handle, SpeedRef, Good(669, t0.AddSeconds(3))); + drv.DeliverNotificationForTest(handle, SpeedRef, Good(665, t0)); // disarmed: accepted + + seen.Select(e => e.Snapshot.Value).ShouldBe(new object[] { 668, 669, 665 }); + seen[0].Snapshot.StatusCode.ShouldBe((uint)StatusCodes.UncertainLastUsableValue); + } + + /// + /// Coordinator review of F2: with no recovery in play, an upstream clock that steps BACKWARDS (NTP correction, + /// a PLC with a reset clock) must not freeze the node — every notification is published, whatever its time. + /// + [Fact] + public void In_steady_state_a_clock_step_backwards_still_publishes() { var (drv, _, clock) = NewConnectedDriver(); using var _ = drv; @@ -511,14 +558,35 @@ public sealed class OpcUaClientSessionLossTests var seen = new List(); drv.OnDataChange += (_, e) => seen.Add(e); - drv.DeliverNotificationForTest(handle, SpeedRef, Good(668, t0.AddSeconds(2))); - 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 - drv.DeliverNotificationForTest(handle, SpeedRef, Good(669, t0.AddSeconds(3))); + drv.DeliverNotificationForTest(handle, SpeedRef, Good(1, t0.AddHours(1))); + drv.DeliverNotificationForTest(handle, SpeedRef, Good(2, t0)); // clock stepped back an hour + drv.DeliverNotificationForTest(handle, SpeedRef, Good(3, t0.AddSeconds(1))); - seen.Select(e => e.Snapshot.Value).ShouldBe(new object[] { 668, 668, 669 }); - seen[1].Snapshot.StatusCode.ShouldBe((uint)StatusCodes.UncertainLastUsableValue); + seen.Select(e => e.Snapshot.Value).ShouldBe(new object[] { 1, 2, 3 }); + } + + /// The replay guard is bounded in time: once its window has passed, an older notification is accepted + /// even for a ref that never saw an in-order one. + [Fact] + public async Task The_replay_guard_expires() + { + var (drv, session, clock) = NewConnectedDriver(); + using var _ = drv; + var handle = drv.AddSubscriptionForTest([SpeedRef]); + var t0 = clock.Now.UtcDateTime; + drv.DeliverNotificationForTest(handle, SpeedRef, Good(10, t0.AddSeconds(5))); + SetupReadAt(session, 10, t0.AddSeconds(5)); + await RecoverInPlaceAsync(drv); + var seen = new List(); + drv.OnDataChange += (_, e) => seen.Add(e); + + clock.Advance(TimeSpan.FromSeconds(29)); + drv.DeliverNotificationForTest(handle, SpeedRef, Good(9, t0)); + seen.ShouldBeEmpty("still inside the 30 s window"); + + clock.Advance(TimeSpan.FromSeconds(2)); + drv.DeliverNotificationForTest(handle, SpeedRef, Good(8, t0)); + seen.ShouldHaveSingleItem().Snapshot.Value.ShouldBe(8); } [Fact]