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(); + } }