diff --git a/docs/drivers/OpcUaClient.md b/docs/drivers/OpcUaClient.md index 4c2ffd8b..8cc07a27 100644 --- a/docs/drivers/OpcUaClient.md +++ b/docs/drivers/OpcUaClient.md @@ -66,12 +66,19 @@ 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. + 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 @@ -228,7 +235,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` (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/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..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,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 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 f054508a..76a7c64f 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,33 @@ 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 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); @@ -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)); } + /// #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[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); + } + /// /// #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 @@ -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])); 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..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 @@ -468,4 +468,173 @@ 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); + } + + 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. 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 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; + 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(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[] { 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] + 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(); + } }