diff --git a/docs/drivers/OpcUaClient.md b/docs/drivers/OpcUaClient.md index 9c53b1a3..6c1f6da7 100644 --- a/docs/drivers/OpcUaClient.md +++ b/docs/drivers/OpcUaClient.md @@ -113,13 +113,15 @@ The session keep-alive (`KeepAliveInterval`, default 5 s) is the liveness signal source timestamp (a server that does not send them) is the server's current state and counts as newer. An **equal** source timestamp with different content IS published — it is the server's current state, not a replay, and it repairs the equal-timestamp status-only change the replay guard drops by design. An identical (value, status) is never - republished. An Information line names how many refs it had to repair. The trailing read and the replay-guard disarm + republished. A ref that received ANY delivery while the read was in flight is skipped (a per-ref delivery counter is + captured before the read is issued): without source timestamps the read's sample cannot be ordered against that + notification, and publishing it would regress the node. An Information line names how many refs it had to repair. The trailing read and the replay-guard disarm timer run on the driver clock and are cancelled when the session is released (re-initialise / shutdown / dispose). -- **No older value is ever delivered after a newer one.** The three paths that publish upstream data — live - notifications, the recovery read and the trailing read — make their "may I publish this?" decision and the publish - inside one driver-wide critical section. A single lock rather than one per ref: a session's notifications arrive on - its publish thread one message at a time, so the only contention is a recovery/trailing read overlapping a - notification, and an uncontended lock costs tens of nanoseconds per notification. Subscribers run inside it — the +- **No older value is ever delivered after a newer one.** The paths that publish upstream data — live notifications, + the recovery read, the trailing read and the comms-loss Bad fan — make their "may I publish this?" decision and the + publish inside one driver-wide critical section. A single lock rather than one per ref: it also serialises delivery + across subscriptions (the SDK dispatches each subscription's messages on its own worker), which is cheap because each + section is a dictionary update plus a subscriber `Tell` and nothing awaits under it. Subscribers run inside it — the host's handler only `Tell`s an actor — so a subscriber must never block on, or call back into, the driver. - **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 the #527 comms-loss watchdog's trip condition; `DriverInstanceActor` 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 b3cebfd9..a5d82caa 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs @@ -361,14 +361,25 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// Review of 3b88056f — serialises "decide whether this upstream value may be published" with "publish it" across /// the three paths that publish upstream data: live notifications, the recovery read and the trailing reconcile /// read. Without it a path could pass its newer-than check, lose the CPU to another path publishing a NEWER value, - /// and then deliver its older one last — regressing the node. One driver-wide lock rather than one per ref: SDK - /// notifications for a session arrive on its publish thread one message at a time, so the only contention is a - /// recovery/trailing read overlapping a notification (seconds after an outage), and an uncontended lock costs tens - /// of nanoseconds per notification. Subscribers are invoked under it — the host's handler only Tells an - /// actor — so a subscriber must never block on, or call back into, this driver. + /// and then deliver its older one last — regressing the node. It also covers the comms-loss Bad fan + /// (). One driver-wide lock rather than one per ref: it also serialises delivery + /// ACROSS subscriptions — with several subscriptions the SDK dispatches each subscription's messages on its own + /// worker, so their notifications no longer run in parallel. That is the cost, and it is small: each critical + /// section is a dictionary update plus a subscriber call that only Tells an actor (tens of nanoseconds to + /// a few microseconds), nothing awaits under it, and per-subscription message order is unaffected. Because + /// subscribers are invoked under it, a subscriber must never block on, or call back into, this driver. /// private readonly Lock _deliveryLock = new(); + /// + /// Re-review of 30a9d5f0 — per RawPath, a counter bumped (under ) every time a value is + /// delivered for it. The trailing re-read captures it before issuing its read and skips any ref delivered to since: + /// without source timestamps the read's value cannot be ordered against a notification that landed while the read + /// was in flight, and publishing it would regress the node to the value the read sampled. + /// + private readonly System.Collections.Concurrent.ConcurrentDictionary _deliverySeq = + new(StringComparer.Ordinal); + /// Review of 3b88056f — cancels the post-recovery background work (trailing read, replay-guard disarm timer) /// of the current session; cancelled and replaced whenever the session is released, and on dispose. private CancellationTokenSource _sessionWorkCts = new(); @@ -1077,9 +1088,11 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// Cancel the current session's background work and start a fresh token for the next one. private void CancelSessionWork() { + // Cancelled but deliberately NOT disposed (re-review of 30a9d5f0): the keep-alive thread may be reading the old + // source's Token concurrently, and a disposed source throws ObjectDisposedException there. Nothing here is + // timer-backed (no CancelAfter), so an undisposed source holds no unmanaged resource. var old = Interlocked.Exchange(ref _sessionWorkCts, new CancellationTokenSource()); - try { old.Cancel(); } catch (ObjectDisposedException) { } - old.Dispose(); + old.Cancel(); } /// Upper bound on the best-effort wire calls a teardown makes against a possibly-dead upstream. @@ -1884,6 +1897,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit } RecordUpstreamGood(fullRef, snapshot); _lastValues[fullRef] = snapshot; + _deliverySeq.AddOrUpdate(fullRef, 1, (_, n) => n + 1); _awaitingRecovery.TryRemove(fullRef, out _); OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, fullRef, snapshot)); } @@ -3030,18 +3044,23 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// private void PublishCommsLoss(DateTime now) { - foreach (var rs in _subscriptions.Values) + // Under the delivery lock (re-review of 30a9d5f0) so the Bad cannot interleave with a recovery publish. + lock (_deliveryLock) { - foreach (var pair in rs.ItemHandlers) + foreach (var rs in _subscriptions.Values) { - 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, now, now))); + foreach (var pair in rs.ItemHandlers) + { + var fullRef = RefOf(pair.Item); + _lastValues.TryGetValue(fullRef, out var last); + _awaitingRecovery[fullRef] = 0; + _deliverySeq.AddOrUpdate(fullRef, 1, (_, n) => n + 1); + // 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, now, now))); + } } } AnnotateAlarmQuality(StatusBadNoCommunication, now); @@ -3105,6 +3124,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit private void PublishUpstreamLocked(string fullRef, DataValueSnapshot value) { _lastValues[fullRef] = value; + _deliverySeq.AddOrUpdate(fullRef, 1, (_, n) => n + 1); RecordUpstreamGood(fullRef, value); if (value.SourceTimestampUtc is { } readAt && StatusCode.IsGood(value.StatusCode) && Interlocked.Read(ref _replayGuardUntilTicks) != 0) @@ -3146,6 +3166,9 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit .Distinct(StringComparer.Ordinal) .ToList(); if (refs.Count == 0) return; + // Captured BEFORE the read is issued: a ref delivered to while the read is in flight has a value at least as + // new as anything the read could have sampled, so the read must not overwrite it. + var seqAtIssue = refs.Select(r => _deliverySeq.GetValueOrDefault(r)).ToArray(); var (results, transportOk) = await ReadCoreAsync(refs, ct, session).ConfigureAwait(false); if (!transportOk) return; // a new outage — the keep-alive reports it var published = 0; @@ -3154,6 +3177,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit if (!Current()) return; lock (_deliveryLock) { + if (_deliverySeq.GetValueOrDefault(refs[i]) != seqAtIssue[i]) continue; if (!IsReconcileRepair(refs[i], results[i])) continue; BeforeReconcilePublishForTest?.Invoke(refs[i]); _awaitingRecovery.TryRemove(refs[i], out _); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientConnectionLossSignalTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientConnectionLossSignalTests.cs index e3f34208..d2db998f 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientConnectionLossSignalTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientConnectionLossSignalTests.cs @@ -614,4 +614,49 @@ public sealed class OpcUaClientConnectionLossSignalTests await drv.ReconcileTaskForTest.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); } + + /// + /// Re-review of 30a9d5f0: with no source timestamps to compare, "newer" cannot be judged from the values. The + /// trailing read samples V1; while it is in flight a live notification delivers V2; the read then returns V1 — + /// different from the last delivered value and with no timestamp to prove it older. It must NOT be published: a + /// ref delivered to after the read was issued is skipped (per-ref delivery sequence), or the node regresses to V1. + /// + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task A_notification_delivered_while_the_trailing_read_is_in_flight_is_not_overwritten(bool readHasSourceTime) + { + var (drv, session, clock, _) = NewConnectedDriver(); + using var _ = drv; + drv.PostRecoveryReconcileDelay = () => TimeSpan.Zero; + var handle = drv.AddSubscriptionForTest([SpeedRef]); + var t0 = clock.Now.UtcDateTime; + DataValue NoSource(object v, DateTime server) => new(new Variant(v), StatusCodes.Good, DateTime.MinValue, server); + drv.DeliverNotificationForTest(handle, SpeedRef, NoSource(10f, t0)); + var reads = 0; + session.Setup(s => s.ReadAsync( + It.IsAny(), It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny())) + .ReturnsAsync(() => + { + DataValue result; + if (Interlocked.Increment(ref reads) == 1) + result = NoSource(10f, t0.AddSeconds(1)); // recovery read: unchanged + else + { + // The trailing read has sampled V1 at the server; before its response is processed, the live + // subscription delivers the newer V2. + drv.DeliverNotificationForTest(handle, SpeedRef, NoSource(20f, t0.AddSeconds(9))); + result = readHasSourceTime + ? Good(15f, t0.AddSeconds(8)) // read carries a source time, V2 did not + : NoSource(15f, t0.AddSeconds(8)); + } + return new ReadResponse { ResponseHeader = new ResponseHeader(), Results = [result], DiagnosticInfos = [] }; + }); + + var seen = await RecoverAndReconcileAsync(drv); + + reads.ShouldBe(2, "the trailing read must have run"); + lock (seen) seen.Select(e => e.Value).ShouldBe([10f, 10f, 20f], "the read's stale 15 must not overwrite the newer 20"); + } }