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

fix(opcuaclient): #670 F2 re-gate — hold the replay guard for the whole window

Browse Source 
The F2 live gate failed (10 value regressions over 4 blips): IsLateReplay
disarmed a ref on its first in-order notification, which arrives ~1 s after
recovery, while the SDK's late replay lands 4-6 s later. Equal source times
also passed, so a replay equal to the recovery read got through.

While the window is open each ref now carries a high-water mark (seeded at
arming, raised by the recovery read, advanced by every accepted notification);
a notification at or below it is dropped (equal included — OPC UA moves the
source timestamp on every value/status change). Null source time passes
without moving the mark. Window expiry still disarms entirely, so an upstream
clock step-back is never frozen outside the window. Arm/expiry serialised by a
lock so an expiry racing a re-arm cannot wipe fresh seeds.
This commit is contained in:
Joseph Doherty
2026-10-11 09:42:39 -04:00
parent 57deaeba16
commit 099bc1b6e0
5 changed files with 146 additions and 33 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 late replayed notification (older than the ref's last Good source time) was dropped inside the post-recovery replay window (#670 F2).</summary>
/// <summary>A late replayed notification (source time at or below the ref's high-water mark) 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
+46 -19
View File
@@ -305,19 +305,27 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
/// <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
/// SDK was observed delivering a missed notification seconds late with an older source time. Per ref it holds a
/// HIGH-WATER MARK: seeded with the last Good upstream source time at arming, raised by the recovery read, and
/// advanced by EVERY notification accepted while the window is open. A notification whose source time is at or
/// below the mark is dropped — equal included, because OPC UA moves the source timestamp on every value or
/// status change, so an equal one is a replay of what was already accepted. The guard holds for the WHOLE
/// window: the F2 live re-gate saw fresh notifications ~1 s after recovery and the SDK's late replay 4–6 s
/// later, so the original "disarm a ref on its first in-order notification" let the replay through. A
/// notification with no source time passes and leaves the mark alone. 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>
/// <summary>Clock ticks at which the replay guard expires (0 = not armed).</summary>
private long _replayGuardUntilTicks;
/// <summary>Serialises arming against expiry, so an expiry racing a re-arm cannot wipe the new seeds.</summary>
private readonly object _replayGuardLock = new();
/// <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);
@@ -1760,7 +1768,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
_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,
"(source {Source:O} is not newer than {Newest:O}).", _driverInstanceId, fullRef, snapshot.SourceTimestampUtc,
new DateTime(newest, DateTimeKind.Utc));
return;
}
@@ -1778,23 +1786,39 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
}
/// <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.
/// #670 F2 — true when the replay window is open and <paramref name="source"/> is at or below
/// <paramref name="fullRef"/>'s high-water mark. Otherwise (inside the window) the notification is accepted and
/// its source time becomes the ref's new mark, so the ref stays guarded for the whole window. A notification
/// with no source time passes without moving the mark. 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))
var until = Interlocked.Read(ref _replayGuardUntilTicks);
if (until == 0) return false;
if (Clock.GetUtcNow().UtcDateTime.Ticks >= until)
{
_replayGuard.Clear();
lock (_replayGuardLock)
{
if (Interlocked.Read(ref _replayGuardUntilTicks) == until)
{
Interlocked.Exchange(ref _replayGuardUntilTicks, 0);
_replayGuard.Clear();
}
}
return false;
}
if (!_replayGuard.TryGetValue(fullRef, out newest)) return false;
if (src.Ticks < newest) return true;
_replayGuard.TryRemove(fullRef, out _);
return false;
if (source is not { } src) return false;
var ticks = src.Ticks;
while (true)
{
if (_replayGuard.TryGetValue(fullRef, out newest))
{
if (ticks <= newest) return true;
if (_replayGuard.TryUpdate(fullRef, ticks, newest)) return false;
}
else if (_replayGuard.TryAdd(fullRef, ticks)) return false;
}
}
/// <summary>
@@ -1803,15 +1827,18 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
/// </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);
lock (_replayGuardLock)
{
_replayGuard.Clear();
foreach (var kv in _lastGoodSourceTicks) _replayGuard[kv.Key] = kv.Value;
Interlocked.Exchange(ref _replayGuardUntilTicks, (now + window).Ticks);
}
}
/// <summary>