fix(opcuaclient): #676 review fix-ups (port-less rewrite, full reconcile, delivery lock, timers)
1. RewriteEndpointUrl keeps the advertised port when the configured URL names none (Uri.Port -1 would have sent the session to 4840); port-less + IPv6 tests. 2. The trailing re-read repairs any (value, status) difference that is not older than the last delivered value — Good->Bad/Uncertain included; a read without a source timestamp counts as newer; an equal source time with different content is published (repairs the replay guard's by-design equal-timestamp drop). 3. One driver-wide delivery lock makes decide+publish atomic across live notifications, the recovery read and the trailing read, so an older value can never land after a newer one (deterministic race test via a seam). 4. Known gap documented: a ConnectionLost raised while the host is not Connected is swallowed and that session falls back to the watchdog. 5. Only a SERVER-reported subscription loss raises ConnectionLost; the publishing-stall heuristic stays with the #527 watchdog. 6. Post-recovery timers run on Task.Delay(…, Clock, ct) with a session-work CTS cancelled on session release and dispose. 7. Replay summary at Information only when something was dropped (Debug otherwise); a re-arm reports the superseded window's drops and counts afresh. 8. DriverInstanceActor: AttachHostStatusSource doc comment restored; Reconnecting swallows a stray DisconnectObserved (test). 9. Stall wire test: 120 s session timeout, 3000-cycle subscription lifetime, one further bad keep-alive tick, 60 s headroom on the post-recovery waits (5/5 green).
This commit is contained in:
Joseph Doherty
+219
-10
@@ -38,10 +38,12 @@ public sealed class OpcUaClientConnectionLossSignalTests
|
||||
lock (Entries) Entries.Add((logLevel, formatter(state, exception)));
|
||||
}
|
||||
|
||||
public string[] Info(string contains)
|
||||
public string[] Info(string contains) => At(LogLevel.Information, contains);
|
||||
|
||||
public string[] At(LogLevel level, string contains)
|
||||
{
|
||||
lock (Entries)
|
||||
return [.. Entries.Where(e => e.Level == LogLevel.Information && e.Message.Contains(contains)).Select(e => e.Message)];
|
||||
return [.. Entries.Where(e => e.Level == level && e.Message.Contains(contains)).Select(e => e.Message)];
|
||||
}
|
||||
}
|
||||
|
||||
@@ -231,6 +233,31 @@ public sealed class OpcUaClientConnectionLossSignalTests
|
||||
drv.GetHealth().State.ShouldBe(DriverState.Degraded);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Review of 3b88056f: only a SERVER-reported subscription loss is definitive. The publishing-stall heuristic
|
||||
/// (SDK <c>PublishingStopped</c> persisting past a grace) is an inference from silence — it still degrades health
|
||||
/// and freezes the stamp, but recovery stays with the #527 watchdog.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void The_publishing_stall_heuristic_does_not_raise_ConnectionLost()
|
||||
{
|
||||
var (drv, _, clock, _) = NewConnectedDriver();
|
||||
using var _ = drv;
|
||||
drv.AddSubscriptionForTest([SpeedRef]);
|
||||
drv.IsPublishingStopped = _ => true;
|
||||
drv.PublishingStallGrace = _ => TimeSpan.FromSeconds(10);
|
||||
var raised = 0;
|
||||
drv.ConnectionLost += (_, _) => raised++;
|
||||
|
||||
drv.RaiseKeepAliveForTest(StatusCodes.Good);
|
||||
clock.Advance(TimeSpan.FromSeconds(11));
|
||||
drv.RaiseKeepAliveForTest(StatusCodes.Good);
|
||||
|
||||
// Positive control: the heuristic DID fire.
|
||||
drv.GetHealth().LastError.ShouldNotBeNull().ShouldContain("subscription lost");
|
||||
raised.ShouldBe(0);
|
||||
}
|
||||
|
||||
// ── Replay-drop summary ──────────────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
[Fact]
|
||||
@@ -264,7 +291,28 @@ public sealed class OpcUaClientConnectionLossSignalTests
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_window_with_no_drops_still_reports_it_closed_and_counts_restart_per_window()
|
||||
public async Task A_window_with_no_drops_closes_at_Debug_not_Information()
|
||||
{
|
||||
var (drv, session, clock, log) = NewConnectedDriver();
|
||||
using var _ = drv;
|
||||
drv.PostRecoveryReconcileDelay = () => TimeSpan.FromDays(1);
|
||||
var handle = drv.AddSubscriptionForTest([SpeedRef]);
|
||||
var t0 = clock.Now.UtcDateTime;
|
||||
drv.DeliverNotificationForTest(handle, SpeedRef, Good(1, t0));
|
||||
SetupRead(session, () => t0, 1);
|
||||
|
||||
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
|
||||
drv.RaiseKeepAliveForTest(StatusCodes.Good);
|
||||
await drv.RecoveryTaskForTest;
|
||||
clock.Advance(TimeSpan.FromSeconds(31));
|
||||
drv.DeliverNotificationForTest(handle, SpeedRef, Good(2, t0.AddSeconds(40)));
|
||||
|
||||
log.Info("replay window closed").ShouldBeEmpty("nothing was dropped — not worth an Information line per recovery");
|
||||
log.At(LogLevel.Debug, "replay window closed").ShouldHaveSingleItem().ShouldContain("dropped 0");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_re_arm_inside_an_open_window_reports_the_earlier_drops_once_and_starts_counting_afresh()
|
||||
{
|
||||
var (drv, session, clock, log) = NewConnectedDriver();
|
||||
using var _ = drv;
|
||||
@@ -278,19 +326,18 @@ public sealed class OpcUaClientConnectionLossSignalTests
|
||||
drv.RaiseKeepAliveForTest(StatusCodes.Good);
|
||||
await drv.RecoveryTaskForTest;
|
||||
drv.DeliverNotificationForTest(handle, SpeedRef, Good(0, t0.AddSeconds(-1))); // one drop in window 1
|
||||
clock.Advance(TimeSpan.FromSeconds(31));
|
||||
drv.DeliverNotificationForTest(handle, SpeedRef, Good(2, t0.AddSeconds(40)));
|
||||
|
||||
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
|
||||
clock.Advance(TimeSpan.FromSeconds(5)); // window 1 still open …
|
||||
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); // … a second outage re-arms it
|
||||
drv.RaiseKeepAliveForTest(StatusCodes.Good);
|
||||
await drv.RecoveryTaskForTest;
|
||||
clock.Advance(TimeSpan.FromSeconds(31));
|
||||
drv.DeliverNotificationForTest(handle, SpeedRef, Good(3, t0.AddSeconds(80)));
|
||||
|
||||
log.Info("replay window closed").ShouldBe([
|
||||
"OpcUaClientDriver 'opcua-676': post-recovery replay window closed — dropped 1 late notifications across 1 refs after recovery.",
|
||||
"OpcUaClientDriver 'opcua-676': post-recovery replay window closed — dropped 0 late notifications across 0 refs after recovery.",
|
||||
]);
|
||||
log.Info("replay window closed").ShouldBe(
|
||||
["OpcUaClientDriver 'opcua-676': post-recovery replay window closed — dropped 1 late notifications across 1 refs after recovery."],
|
||||
"window 1's drop is reported once, when the re-arm supersedes it — not carried into window 2's summary");
|
||||
log.At(LogLevel.Debug, "replay window closed").ShouldHaveSingleItem().ShouldContain("dropped 0");
|
||||
}
|
||||
|
||||
// ── Trailing reconcile read ──────────────────────────────────────────────────────────────────────────────────
|
||||
@@ -405,4 +452,166 @@ public sealed class OpcUaClientConnectionLossSignalTests
|
||||
It.IsAny<ReadValueIdCollection>(), It.IsAny<CancellationToken>()), Times.Once,
|
||||
"only the recovery read — the trailing read must not run against a session that is no longer current");
|
||||
}
|
||||
|
||||
/// <summary>Wires the mock session so the first read (the recovery read) returns <paramref name="recovery"/> and
|
||||
/// every later one (the trailing read) returns <paramref name="trailing"/>.</summary>
|
||||
private static Func<int> SetupRecoveryThenTrailing(Mock<ISession> session, DataValue recovery, DataValue trailing)
|
||||
{
|
||||
var reads = 0;
|
||||
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 = [Interlocked.Increment(ref reads) > 1 ? trailing : recovery],
|
||||
DiagnosticInfos = [],
|
||||
});
|
||||
return () => Volatile.Read(ref reads);
|
||||
}
|
||||
|
||||
private static async Task<List<(uint Status, object? Value)>> RecoverAndReconcileAsync(
|
||||
OpcUaClientDriver drv, string fullRef = SpeedRef)
|
||||
{
|
||||
var seen = new List<(uint Status, object? Value)>();
|
||||
drv.OnDataChange += (_, e) => { if (e.FullReference == fullRef) lock (seen) seen.Add((e.Snapshot.StatusCode, e.Snapshot.Value)); };
|
||||
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
|
||||
drv.RaiseKeepAliveForTest(StatusCodes.Good);
|
||||
await drv.RecoveryTaskForTest;
|
||||
await drv.ReconcileTaskForTest;
|
||||
return seen;
|
||||
}
|
||||
|
||||
/// <summary>Review of 3b88056f: a Good→Bad change lost in the stalled pipeline left the node stale-GOOD; the
|
||||
/// trailing read repairs any (value, status) difference, not just Good ones.</summary>
|
||||
[Fact]
|
||||
public async Task The_trailing_re_read_repairs_a_lost_change_to_Bad()
|
||||
{
|
||||
var (drv, session, clock, _) = NewConnectedDriver();
|
||||
using var _ = drv;
|
||||
drv.PostRecoveryReconcileDelay = () => TimeSpan.Zero;
|
||||
var handle = drv.AddSubscriptionForTest([SpeedRef]);
|
||||
var t0 = clock.Now.UtcDateTime;
|
||||
drv.DeliverNotificationForTest(handle, SpeedRef, Good(10f, t0));
|
||||
var reads = SetupRecoveryThenTrailing(session, Good(10f, t0),
|
||||
new DataValue(new Variant(10f), StatusCodes.BadSensorFailure, t0.AddSeconds(9), t0.AddSeconds(9)));
|
||||
|
||||
var seen = await RecoverAndReconcileAsync(drv);
|
||||
|
||||
reads().ShouldBe(2);
|
||||
seen.ShouldBe([(BadNoCommunication, (object?)10f), (0u, 10f), ((uint)StatusCodes.BadSensorFailure, 10f)]);
|
||||
}
|
||||
|
||||
/// <summary>A server that sends no source timestamp: the re-read is the server's CURRENT state, read after the
|
||||
/// stalled pipeline drained, so a differing value is newer than anything delivered before — it is published.</summary>
|
||||
[Fact]
|
||||
public async Task The_trailing_re_read_repairs_a_server_that_sends_no_source_timestamp()
|
||||
{
|
||||
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 = SetupRecoveryThenTrailing(session, NoSource(10f, t0.AddSeconds(1)), NoSource(40f, t0.AddSeconds(9)));
|
||||
|
||||
var seen = await RecoverAndReconcileAsync(drv);
|
||||
|
||||
reads().ShouldBe(2);
|
||||
seen.ShouldBe([(BadNoCommunication, (object?)10f), (0u, 10f), (0u, 40f)]);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Equal source timestamp, different status: the re-read is the server's current state, not a replay, so it is
|
||||
/// published — which also repairs the one case the replay guard drops by design (a status-only change that kept
|
||||
/// its source timestamp inside the window). An IDENTICAL (value, status) is not republished.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task The_trailing_re_read_publishes_a_status_only_change_at_an_equal_source_time_but_not_an_identical_value()
|
||||
{
|
||||
var (drv, session, clock, _) = NewConnectedDriver();
|
||||
using var _ = drv;
|
||||
drv.PostRecoveryReconcileDelay = () => TimeSpan.Zero;
|
||||
var handle = drv.AddSubscriptionForTest([SpeedRef, CounterRef]);
|
||||
var t0 = clock.Now.UtcDateTime;
|
||||
drv.DeliverNotificationForTest(handle, SpeedRef, Good(10f, t0));
|
||||
drv.DeliverNotificationForTest(handle, CounterRef, Good(7, 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(() => new ReadResponse
|
||||
{
|
||||
ResponseHeader = new ResponseHeader(),
|
||||
Results = Interlocked.Increment(ref reads) > 1
|
||||
? [new DataValue(new Variant(10f), StatusCodes.UncertainLastUsableValue, t0, t0), Good(7, t0)]
|
||||
: [Good(10f, t0), Good(7, t0)],
|
||||
DiagnosticInfos = [],
|
||||
});
|
||||
var seen = new List<(string Ref, uint Status)>();
|
||||
drv.OnDataChange += (_, e) => { lock (seen) seen.Add((e.FullReference, e.Snapshot.StatusCode)); };
|
||||
|
||||
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
|
||||
drv.RaiseKeepAliveForTest(StatusCodes.Good);
|
||||
await drv.RecoveryTaskForTest;
|
||||
await drv.ReconcileTaskForTest;
|
||||
|
||||
reads.ShouldBe(2);
|
||||
seen.Where(e => e.Status != BadNoCommunication).ShouldBe([
|
||||
(SpeedRef, 0u), (CounterRef, 0u), // recovery read
|
||||
(SpeedRef, (uint)StatusCodes.UncertainLastUsableValue), // trailing read: status-only change at t0
|
||||
]);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Review of 3b88056f: the trailing read's compare and its publish are one critical section with the live
|
||||
/// notification path. Here a NEWER notification arrives exactly between the trailing read's decision and its
|
||||
/// publish; it must wait, so the older re-read can never land after it and regress the node.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_notification_racing_the_trailing_publish_is_serialised_after_it()
|
||||
{
|
||||
var (drv, session, clock, _) = NewConnectedDriver();
|
||||
using var _ = drv;
|
||||
drv.PostRecoveryReconcileDelay = () => TimeSpan.Zero;
|
||||
var handle = drv.AddSubscriptionForTest([SpeedRef]);
|
||||
var t0 = clock.Now.UtcDateTime;
|
||||
drv.DeliverNotificationForTest(handle, SpeedRef, Good(10f, t0));
|
||||
SetupRecoveryThenTrailing(session, Good(10f, t0), Good(40f, t0.AddSeconds(5)));
|
||||
Task? racer = null;
|
||||
drv.BeforeReconcilePublishForTest = _ =>
|
||||
{
|
||||
racer = Task.Run(() => drv.DeliverNotificationForTest(handle, SpeedRef, Good(50f, t0.AddSeconds(8))));
|
||||
// Bounded: with the critical section the racer cannot finish until this publish is done.
|
||||
racer.Wait(TimeSpan.FromMilliseconds(300));
|
||||
};
|
||||
|
||||
var seen = await RecoverAndReconcileAsync(drv);
|
||||
await racer!.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken);
|
||||
|
||||
lock (seen) seen.Select(e => e.Value).ShouldBe([10f, 10f, 40f, 50f], "the newer 50 must be the last value delivered");
|
||||
}
|
||||
|
||||
/// <summary>Review of 3b88056f: the background timers are bound to the session — a shutdown cancels a pending
|
||||
/// trailing read instead of leaving it to fire (and read) against whatever session exists later.</summary>
|
||||
[Fact]
|
||||
public async Task Shutdown_cancels_a_pending_trailing_read()
|
||||
{
|
||||
var (drv, session, clock, _) = NewConnectedDriver();
|
||||
using var _ = drv;
|
||||
drv.PostRecoveryReconcileDelay = () => TimeSpan.FromHours(1);
|
||||
var handle = drv.AddSubscriptionForTest([SpeedRef]);
|
||||
drv.DeliverNotificationForTest(handle, SpeedRef, Good(10f, clock.Now.UtcDateTime));
|
||||
SetupRead(session, () => clock.Now.UtcDateTime, 10f);
|
||||
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
|
||||
drv.RaiseKeepAliveForTest(StatusCodes.Good);
|
||||
await drv.RecoveryTaskForTest;
|
||||
drv.ReconcileTaskForTest.IsCompleted.ShouldBeFalse("pending for an hour");
|
||||
|
||||
await drv.ShutdownAsync(TestContext.Current.CancellationToken);
|
||||
|
||||
await drv.ReconcileTaskForTest.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
+16
@@ -34,6 +34,22 @@ public sealed class OpcUaClientEndpointRewriteTests
|
||||
.ShouldBe("opc.tcp://127.0.0.1:61000/OtOpcUa");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void A_configured_url_without_a_port_keeps_the_advertised_port()
|
||||
{
|
||||
// Review of 3b88056f: Uri.Port is -1 for a port-less opc.tcp URL; copying it would drop the advertised port and
|
||||
// the SDK would fall back to 4840.
|
||||
OpcUaClientDriver.RewriteEndpointUrl("opc.tcp://internal-host:62541/Server", "opc.tcp://gateway/Server")
|
||||
.ShouldBe("opc.tcp://gateway:62541/Server");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void An_IPv6_configured_host_is_bracketed()
|
||||
{
|
||||
OpcUaClientDriver.RewriteEndpointUrl("opc.tcp://internal-host:4840/OtOpcUa", "opc.tcp://[fd00::1]:54840/OtOpcUa")
|
||||
.ShouldBe("opc.tcp://[fd00::1]:54840/OtOpcUa");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void A_different_scheme_is_left_alone()
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user