fix(opcuaclient): #676 re-review — trailing read skips refs delivered to while in flight
1. A per-ref delivery counter (bumped under the delivery lock by every publish path) is captured before the trailing read is issued; a ref delivered to since is skipped. Without source timestamps the read's sample could not be ordered against a notification that landed mid-read, and publishing it regressed the node. Test injects the notification inside the mocked read, with and without a read source time (red before). 2. Delivery-lock comment corrected (it serialises across subscriptions, which the SDK dispatches on separate workers); PublishCommsLoss now publishes under the lock too. 3. CancelSessionWork cancels without disposing, so a keep-alive thread reading the old token cannot hit ObjectDisposedException.
This commit is contained in:
Joseph Doherty
@@ -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`
|
||||
|
||||
@@ -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 <c>Tell</c>s 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
|
||||
/// (<see cref="PublishCommsLoss"/>). 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 <c>Tell</c>s 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.
|
||||
/// </summary>
|
||||
private readonly Lock _deliveryLock = new();
|
||||
|
||||
/// <summary>
|
||||
/// Re-review of 30a9d5f0 — per RawPath, a counter bumped (under <see cref="_deliveryLock"/>) 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.
|
||||
/// </summary>
|
||||
private readonly System.Collections.Concurrent.ConcurrentDictionary<string, long> _deliverySeq =
|
||||
new(StringComparer.Ordinal);
|
||||
|
||||
/// <summary>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.</summary>
|
||||
private CancellationTokenSource _sessionWorkCts = new();
|
||||
@@ -1077,9 +1088,11 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
|
||||
/// <summary>Cancel the current session's background work and start a fresh token for the next one.</summary>
|
||||
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();
|
||||
}
|
||||
|
||||
/// <summary>Upper bound on the best-effort wire calls a teardown makes against a possibly-dead upstream.</summary>
|
||||
@@ -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
|
||||
/// </summary>
|
||||
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 _);
|
||||
|
||||
+45
@@ -614,4 +614,49 @@ public sealed class OpcUaClientConnectionLossSignalTests
|
||||
|
||||
await drv.ReconcileTaskForTest.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[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<RequestHeader>(), It.IsAny<double>(), It.IsAny<TimestampsToReturn>(),
|
||||
It.IsAny<ReadValueIdCollection>(), It.IsAny<CancellationToken>()))
|
||||
.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");
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user