fix(opcuaclient): #670 live-gate follow-ups — historize the outage Bad, drop late replays

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

@@ -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: <status>"`, 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.

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 <see cref="PollOverrunDriverInstanceTag"/> (<c>driver_instance</c>) +
     /// <c>event</c> = <c>publish_response|notification|data_change|bad_status_value|publish_error|
-    /// keepalive_miss|keepalive_recovered</c>.
+    /// keepalive_miss|keepalive_recovered|stale_notification_dropped</c>.
     /// </summary>
     public static readonly Counter<long> OpcUaClientEvents =
         Meter.CreateCounter<long>("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 ----------------
 

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDiagnostics.cs

@@ -24,6 +24,8 @@ internal sealed class OpcUaClientDiagnostics
     public const string KeepAliveMiss = "keepalive_miss";
     /// <summary>A keep-alive came back Good after a bad one on the same session (an in-place recovery, #670).</summary>
     public const string KeepAliveRecovered = "keepalive_recovered";
+    /// <summary>A notification older than the newest Good source time already published for its ref was dropped (#670 F2).</summary>
+    public const string StaleNotificationDropped = "stale_notification_dropped";
 
     private readonly string _driverInstanceId;
 

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs

@@ -294,6 +294,15 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
 
     /// <summary>#670 — the last value delivered per subscribed RawPath, so an outage can publish
     /// <c>BadNoCommunication</c> with the value retained (the same shape the host's own comms-loss fan uses).</summary>
+    /// <summary>
+    ///     #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).
+    /// </summary>
+    private readonly System.Collections.Concurrent.ConcurrentDictionary<string, long> _lastGoodSourceTicks =
+        new(StringComparer.Ordinal);
+
     private readonly System.Collections.Concurrent.ConcurrentDictionary<string, DataValueSnapshot> _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));
     }
 
+    /// <summary>#670 F2 — remember the newest Good upstream source time for <paramref name="fullRef"/>.</summary>
+    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));
+    }
+
     /// <summary>
     ///     #670 — advance <see cref="DriverHealth.LastSuccessfulRead"/> 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]));

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientSessionLossTests.cs

@@ -468,4 +468,105 @@ public sealed class OpcUaClientSessionLossTests
         public void Log<TState>(Microsoft.Extensions.Logging.LogLevel logLevel, Microsoft.Extensions.Logging.EventId eventId,
             TState state, Exception? exception, Func<TState, Exception?, string> formatter) => Count++;
     }
+
+    /// <summary>
+    ///     #670 follow-up F1 (live gate): the outage Bad carried the last Good sample's SOURCE timestamp, so the
+    ///     recorder (which historizes <c>SourceTimestampUtc ?? ServerTimestampUtc</c>) 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.
+    /// </summary>
+    [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<DataChangeEventArgs>();
+        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);
+    }
+
+    /// <summary>
+    ///     #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.
+    /// </summary>
+    [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<DataChangeEventArgs>();
+        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<DataChangeEventArgs>();
+        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<RequestHeader>(), It.IsAny<double>(), It.IsAny<TimestampsToReturn>(),
+                It.IsAny<ReadValueIdCollection>(), It.IsAny<CancellationToken>()))
+            .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<DataChangeEventArgs>();
+        drv.OnDataChange += (_, e) => seen.Add(e);
+        drv.DeliverNotificationForTest(handle, SpeedRef, Good(667, t0.AddSeconds(5))); // replayed, older than the read
+        seen.ShouldBeEmpty();
+    }
 }