dohertj2/lmxopcua
1
0
Fork 0
Code Issues 32 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(opcuaclient): scope the #670 F2 late-replay drop to the post-recovery window

Browse Source 
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.
This commit is contained in:
Joseph Doherty
2026-10-11 09:15:51 -04:00
parent 99c13e4459
commit 87e8aff786
4 changed files with 157 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDiagnostics.cs
+1 -1
View File
@@ -24,7 +24,7 @@ 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>
/// <summary>A late replayed notification (older than the ref's last Good source time) was dropped inside the post-recovery replay window (#670 F2).</summary>
public const string StaleNotificationDropped = "stale_notification_dropped";
private readonly string _driverInstanceId;
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs
+71 -11
View File
@@ -295,14 +295,32 @@ 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).
/// #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.
/// </summary>
private readonly System.Collections.Concurrent.ConcurrentDictionary<string, long> _lastGoodSourceTicks =
new(StringComparer.Ordinal);
/// <summary>
/// #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
/// <see cref="_replayGuardUntilTicks"/>. 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.
/// </summary>
private readonly System.Collections.Concurrent.ConcurrentDictionary<string, long> _replayGuard =
new(StringComparer.Ordinal);
/// <summary>Clock ticks at which the replay guard expires (0 = never armed).</summary>
private long _replayGuardUntilTicks;
/// <summary>Floor of the replay-guard window; the live gate saw the late replay 4–5 s after recovery.</summary>
private static readonly TimeSpan ReplayGuardMinimum = TimeSpan.FromSeconds(30);
private readonly System.Collections.Concurrent.ConcurrentDictionary<string, DataValueSnapshot> _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));
}
/// <summary>#670 F2 — remember the newest Good upstream source time for <paramref name="fullRef"/>.</summary>
/// <summary>#670 F2 — remember the last accepted 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));
_lastGoodSourceTicks[fullRef] = src.Ticks;
}
/// <summary>
/// #670 F2 — true when the replay guard is live for <paramref name="fullRef"/> and <paramref name="source"/> is
/// older than its armed time. An in-order (equal or newer) notification disarms the ref; an expired window
/// disarms everything.
/// </summary>
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;
}
/// <summary>
/// #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).
/// </summary>
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);
}
/// <summary>
@@ -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]));