From 99c13e44593278f54c0e9de347f9855aef934ffd Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sun, 11 Oct 2026 09:11:34 -0400 Subject: [PATCH 1/2] =?UTF-8?q?fix(opcuaclient):=20#670=20live-gate=20foll?= =?UTF-8?q?ow-ups=20=E2=80=94=20historize=20the=20outage=20Bad,=20drop=20l?= =?UTF-8?q?ate=20replays?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit F1: the driver's outage BadNoCommunication carried the last Good sample's source timestamp. The recorder historizes SourceTimestampUtc ?? ServerTimestampUtc, so the Bad landed on the Good row's (TagId, SourceTimeUtc) key and the insert-if-absent historian dropped it - no outage marker until the host's watchdog marker ~95 s in, none at all for a short blip. Now stamped with the detection time (source + server), as the host's comms-loss fan does. F2: after a same-session recovery the SDK delivered a missed notification seconds late with an older source time and the node regressed. A notification older than the newest Good upstream source time for its ref (notification or recovery read) is dropped and counted (stale_notification_dropped); equal timestamps pass; the outage Bad never moves the comparison; a re-initialise resets it. --- docs/drivers/OpcUaClient.md | 10 +- .../Observability/OtOpcUaTelemetry.cs | 4 +- .../OpcUaClientDiagnostics.cs | 2 + .../OpcUaClientDriver.cs | 35 +++++- .../OpcUaClientSessionLossTests.cs | 101 ++++++++++++++++++ 5 files changed, 147 insertions(+), 5 deletions(-) diff --git a/docs/drivers/OpcUaClient.md b/docs/drivers/OpcUaClient.md index 4c2ffd8b..90363537 100644 --- a/docs/drivers/OpcUaClient.md +++ b/docs/drivers/OpcUaClient.md @@ -66,12 +66,17 @@ The session keep-alive (`KeepAliveInterval`, default 5 s) is the liveness signal subscription-only driver never calls `ReadAsync`, so without this its read stamp froze at connect time. - **Bad keep-alive** ⇒ health `Degraded` (`LastError = "keep-alive: "`, read stamp preserved), host status `Stopped`, and — **once per outage**, not per keep-alive — every subscribed ref is published `BadNoCommunication` - with its last value retained, plus a quality-only `BadNoCommunication` annotation (#626 pattern, no alarm-state + with its last value retained and **source + server timestamp = the detection time** (live-gate F1: stamped with the + last Good's source time it collided with that row's `(TagId, SourceTimeUtc)` history key and was never stored), plus a quality-only `BadNoCommunication` annotation (#626 pattern, no alarm-state change, no `/alerts` row) on every subscribed alarm condition. - **The session survives** (a network blip; the keep-alive turns Good again) ⇒ recovered in place: health `Healthy`, 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. - **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 @@ -228,7 +233,8 @@ every `InitializeAsync`, including comms-loss reconnects). The discovery channel 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). A climbing +`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 `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/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs index bb3817dc..2d8f4dc8 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs @@ -80,11 +80,11 @@ public static class OtOpcUaTelemetry /// Session/publish-level events of an OPC UA Client driver instance (#276): the operator's view of a /// flapping upstream. Tagged (driver_instance) + /// event = publish_response|notification|data_change|bad_status_value|publish_error| - /// keepalive_miss|keepalive_recovered. + /// keepalive_miss|keepalive_recovered|stale_notification_dropped. /// public static readonly Counter OpcUaClientEvents = Meter.CreateCounter("otopcua.driver.opcuaclient.events", unit: "{event}", - description: "OPC UA Client driver session/publish events (event=publish_response|notification|data_change|bad_status_value|publish_error|keepalive_miss|keepalive_recovered)."); + description: "OPC UA Client driver session/publish events (event=publish_response|notification|data_change|bad_status_value|publish_error|keepalive_miss|keepalive_recovered|stale_notification_dropped)."); // ---------------- VirtualTag / ScriptedAlarm engines ---------------- 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 16c993e8..6b4b1418 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDiagnostics.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDiagnostics.cs @@ -24,6 +24,8 @@ 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). + 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 f054508a..81e3b7eb 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs @@ -294,6 +294,15 @@ 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). + /// + private readonly System.Collections.Concurrent.ConcurrentDictionary _lastGoodSourceTicks = + new(StringComparer.Ordinal); + private readonly System.Collections.Concurrent.ConcurrentDictionary _lastValues = new(StringComparer.Ordinal); @@ -985,6 +994,7 @@ 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(); _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 +1734,30 @@ 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) + { + _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, + new DateTime(newest, DateTimeKind.Utc)); + return; + } + RecordUpstreamGood(fullRef, snapshot); _lastValues[fullRef] = snapshot; _awaitingRecovery.TryRemove(fullRef, out _); OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, fullRef, snapshot)); } + /// #670 F2 — remember the newest 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)); + } + /// /// #670 — advance 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 @@ -2705,8 +2734,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 +2774,7 @@ 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]); 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 93bdeff7..386a4852 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 @@ -468,4 +468,105 @@ public sealed class OpcUaClientSessionLossTests public void Log(Microsoft.Extensions.Logging.LogLevel logLevel, Microsoft.Extensions.Logging.EventId eventId, TState state, Exception? exception, Func formatter) => Count++; } + + /// + /// #670 follow-up F1 (live gate): the outage Bad carried the last Good sample's SOURCE timestamp, so the + /// recorder (which historizes SourceTimestampUtc ?? ServerTimestampUtc) wrote it at the Good row's key — + /// the historian's insert-if-absent (TagId, SourceTimeUtc) key dropped it and no outage marker was ever stored. + /// The Bad is a fresh observation: it is stamped with the detection time, like the host's own comms-loss fan. + /// + [Fact] + public void The_outage_Bad_is_stamped_with_the_detection_time_not_the_last_good_source_time() + { + var (drv, _, clock) = NewConnectedDriver(); + using var _ = drv; + var handle = drv.AddSubscriptionForTest([SpeedRef]); + var goodAt = clock.Now.UtcDateTime; + drv.DeliverNotificationForTest(handle, SpeedRef, Good(42.5f, goodAt)); + var seen = new List(); + drv.OnDataChange += (_, e) => seen.Add(e); + + clock.Advance(TimeSpan.FromSeconds(37)); + drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); + + var bad = seen.ShouldHaveSingleItem().Snapshot; + bad.StatusCode.ShouldBe(BadNoCommunication); + bad.Value.ShouldBe(42.5f, "the value is still retained"); + bad.SourceTimestampUtc.ShouldBe(clock.Now.UtcDateTime, "a Bad at the Good row's source time collides with it in history"); + bad.ServerTimestampUtc.ShouldBe(clock.Now.UtcDateTime); + } + + /// + /// #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. + /// + [Fact] + public void A_notification_older_than_the_last_good_source_time_is_dropped() + { + var (drv, _, clock) = NewConnectedDriver(); + using var _ = drv; + var handle = drv.AddSubscriptionForTest([SpeedRef]); + var t0 = clock.Now.UtcDateTime; + 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))); + + seen.Select(e => e.Snapshot.Value).ShouldBe(new object[] { 668, 668, 669 }); + seen[1].Snapshot.StatusCode.ShouldBe((uint)StatusCodes.UncertainLastUsableValue); + } + + [Fact] + public void The_outage_Bad_does_not_poison_the_ordering_check() + { + var (drv, _, clock) = NewConnectedDriver(); + using var _ = drv; + var handle = drv.AddSubscriptionForTest([SpeedRef]); + var t0 = clock.Now.UtcDateTime; + drv.DeliverNotificationForTest(handle, SpeedRef, Good(1, t0)); + clock.Advance(TimeSpan.FromSeconds(60)); + drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); // stamped t0+60 s + + var seen = new List(); + drv.OnDataChange += (_, e) => seen.Add(e); + // An upstream sample produced during the outage (server-side) and delivered on recovery: older than the Bad's + // detection stamp, newer than the last Good — it is real data and must pass. + drv.DeliverNotificationForTest(handle, SpeedRef, Good(2, t0.AddSeconds(30))); + + seen.ShouldHaveSingleItem().Snapshot.Value.ShouldBe(2); + } + + [Fact] + public async Task A_late_notification_older_than_the_recovery_read_is_dropped() + { + var (drv, session, clock) = NewConnectedDriver(); + using var _ = drv; + var handle = drv.AddSubscriptionForTest([SpeedRef]); + var t0 = clock.Now.UtcDateTime; + drv.DeliverNotificationForTest(handle, SpeedRef, Good(666, t0)); + var readAt = t0.AddSeconds(10); + 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(670), StatusCodes.Good, readAt, readAt)], + DiagnosticInfos = [], + }); + + drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); + drv.RaiseKeepAliveForTest(StatusCodes.Good); + await drv.RecoveryTaskForTest; + + var seen = new List(); + drv.OnDataChange += (_, e) => seen.Add(e); + drv.DeliverNotificationForTest(handle, SpeedRef, Good(667, t0.AddSeconds(5))); // replayed, older than the read + seen.ShouldBeEmpty(); + } } From 87e8aff786273b974f65f9723a946d39f6fee3ac Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sun, 11 Oct 2026 09:15:51 -0400 Subject: [PATCH 2/2] fix(opcuaclient): scope the #670 F2 late-replay drop to the post-recovery window The global 'never older than the last Good source time' rule froze a node whenever an upstream clock stepped backwards (NTP correction, PLC clock reset) until the clock caught up. The check is now a replay guard armed only by a same-session recovery (keep-alive bad->good): per ref it holds the last Good upstream source time (raised by the recovery read), a ref disarms on its first in-order notification, and the whole guard expires after max(30 s, 2x the longest subscription keep-alive window). Outside it every notification is accepted. The steady-state tracker records the LAST accepted time, not a high-water mark. --- docs/drivers/OpcUaClient.md | 12 +-- .../OpcUaClientDiagnostics.cs | 2 +- .../OpcUaClientDriver.cs | 82 ++++++++++++++--- .../OpcUaClientSessionLossTests.cs | 88 ++++++++++++++++--- 4 files changed, 157 insertions(+), 27 deletions(-) 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]