Files
lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientSessionLossTests.cs
T
Joseph Doherty b028322568 fix(opcuaclient): #670 review — subscription loss with a live session, stale recovery latch
M1: a subscription that dies while the session lives (outage longer than
the subscription lifetime but shorter than the session timeout) used to
end Healthy with fresh keep-alive stamps and frozen values. PublishError
BadNoSubscription / BadSubscriptionIdInvalid naming a tracked
subscription, or a tracked subscription whose SDK PublishingStopped
persists one more keep-alive window + 5 s across good keep-alives, now
declares it lost: Degraded, outage published once, and nothing advances
LastSuccessfulRead until the session is released, so the #527 watchdog
re-initialises and re-subscribes. Both detectors wire-tested.

M2: in-place recovery is bound to the session whose keep-alive recovered;
one overtaken by a re-initialise no longer re-arms the new session's
outage latch or degrades its health (ReadCoreAsync expectedSession).

L1 CAS health updates; L2 re-check before each recovery publish; L3 skip
CloseSession on a lost session; L4 SDK log forwarding rate-limited
(20/min); L5 per-driver SDK LoggerFactory disposed with the driver.
Wire test asserts the read stamp stays frozen while the server is down.
2026-10-11 08:41:27 -04:00

472 lines
21 KiB
C#

using System.Text.Json;
using Moq;
using Opc.Ua;
using Opc.Ua.Client;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests;
/// <summary>
/// Gitea #670 — upstream session loss must reach the driver's health and its subscribers. Before the fix a bad
/// keep-alive only flipped the host status: the subscribed values stayed Good forever, health stayed Healthy (so the
/// #527 comms-loss watchdog could never trip — its first conjunct is "non-Healthy"), and <c>LastSuccessfulRead</c>
/// only moved on init / <c>ReadAsync</c>, which a subscription-only driver never calls.
/// </summary>
[Trait("Category", "Unit")]
public sealed class OpcUaClientSessionLossTests
{
private const uint BadNoCommunication = 0x80310000u;
private const string SpeedRef = "Plant/dev/Speed";
private const string CounterRef = "Plant/dev/Counter";
private sealed class ManualClock(DateTimeOffset start) : TimeProvider
{
public DateTimeOffset Now { get; set; } = start;
public override DateTimeOffset GetUtcNow() => Now;
public void Advance(TimeSpan by) => Now += by;
}
private static RawTagEntry Tag(string rawPath, string nodeId) =>
new(rawPath, JsonSerializer.Serialize(new { nodeId }), WriteIdempotent: true);
private static readonly IReadOnlyList<RawTagEntry> Tags =
[
Tag(SpeedRef, "ns=0;s=Speed"),
Tag(CounterRef, "ns=0;s=Counter"),
];
private static Mock<ISession> NewSession()
{
var mock = new Mock<ISession>(MockBehavior.Loose);
mock.SetupGet(s => s.MessageContext)
.Returns(new ServiceMessageContext(SdkTelemetry.Context));
mock.SetupGet(s => s.NamespaceUris).Returns(new NamespaceTable());
return mock;
}
private static (OpcUaClientDriver Driver, Mock<ISession> Session, ManualClock Clock) NewConnectedDriver()
{
var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions { RawTags = Tags }, "opcua-670");
var session = NewSession();
var clock = new ManualClock(new DateTimeOffset(2026, 10, 11, 12, 0, 0, TimeSpan.Zero));
drv.Clock = clock;
drv.SetSessionForTest(session.Object);
drv.SetNamespaceMapForTest(NamespaceMap.FromTable(new NamespaceTable(), Tags));
drv.SetHealthForTest(new DriverHealth(DriverState.Healthy, clock.Now.UtcDateTime, null));
return (drv, session, clock);
}
private static DataValue Good(object value, DateTime ts) =>
new(new Variant(value), StatusCodes.Good, ts, ts);
private static void SetupRead(Mock<ISession> session, params object[] values) =>
session.Setup(s => s.ReadAsync(
It.IsAny<RequestHeader>(), It.IsAny<double>(), It.IsAny<TimestampsToReturn>(),
It.IsAny<ReadValueIdCollection>(), It.IsAny<CancellationToken>()))
.ReturnsAsync((RequestHeader _, double _, TimestampsToReturn _, ReadValueIdCollection nodes, CancellationToken _) =>
new ReadResponse
{
ResponseHeader = new ResponseHeader(),
Results = [.. nodes.Select((_, i) => new DataValue(new Variant(values[i]), StatusCodes.Good, DateTime.UtcNow, DateTime.UtcNow))],
DiagnosticInfos = [],
});
private static async Task WaitForAsync(Func<bool> condition, string because)
{
var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10);
while (DateTime.UtcNow < deadline)
{
if (condition()) return;
await Task.Delay(10);
}
condition().ShouldBeTrue(because);
}
[Fact]
public void A_bad_keep_alive_degrades_health_and_preserves_the_last_successful_read()
{
var (drv, _, clock) = NewConnectedDriver();
using var _ = drv;
var before = drv.GetHealth().LastSuccessfulRead;
clock.Advance(TimeSpan.FromSeconds(5));
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
var h = drv.GetHealth();
h.State.ShouldNotBe(DriverState.Healthy, "a lost session must be visible to the #527 watchdog's first conjunct");
h.State.ShouldBe(DriverState.Degraded);
h.LastSuccessfulRead.ShouldBe(before, "the watchdog measures silence from this stamp, so loss must not move it");
h.LastError.ShouldNotBeNull().ShouldContain("keep-alive");
}
[Fact]
public void A_bad_keep_alive_publishes_BadNoCommunication_once_per_ref_retaining_the_value()
{
var (drv, _, clock) = NewConnectedDriver();
using var _ = drv;
var handle = drv.AddSubscriptionForTest([SpeedRef, CounterRef]);
var t0 = clock.Now.UtcDateTime;
drv.DeliverNotificationForTest(handle, SpeedRef, Good(42.5f, t0));
drv.DeliverNotificationForTest(handle, CounterRef, Good(7, t0));
var seen = new List<DataChangeEventArgs>();
drv.OnDataChange += (_, e) => seen.Add(e);
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
seen.Count.ShouldBe(2, "exactly one Bad per subscribed ref");
var speed = seen.Single(e => e.FullReference == SpeedRef);
speed.Snapshot.StatusCode.ShouldBe(BadNoCommunication);
speed.Snapshot.Value.ShouldBe(42.5f, "the value is retained — only the quality changes (as the host does)");
speed.SubscriptionHandle.ShouldBe(handle);
seen.Single(e => e.FullReference == CounterRef).Snapshot.Value.ShouldBe(7);
// The SDK re-raises a bad keep-alive every interval while the channel stays down: no further publishes.
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
drv.RaiseKeepAliveForTest(StatusCodes.BadConnectionClosed);
seen.Count.ShouldBe(2, "a second bad keep-alive in the same outage must publish nothing");
}
[Fact]
public void Good_keep_alives_and_notifications_advance_the_last_successful_read()
{
var (drv, _, clock) = NewConnectedDriver();
using var _ = drv;
var handle = drv.AddSubscriptionForTest([SpeedRef]);
clock.Advance(TimeSpan.FromSeconds(30));
drv.RaiseKeepAliveForTest(StatusCodes.Good);
drv.GetHealth().LastSuccessfulRead.ShouldBe(clock.Now.UtcDateTime);
clock.Advance(TimeSpan.FromSeconds(30));
drv.DeliverNotificationForTest(handle, SpeedRef, Good(1.0f, clock.Now.UtcDateTime));
drv.GetHealth().LastSuccessfulRead.ShouldBe(clock.Now.UtcDateTime);
drv.GetHealth().State.ShouldBe(DriverState.Healthy);
}
/// <summary>
/// Falsification control for the R2 half of the fix: a long-running, idle, subscription-only driver whose
/// server is perfectly alive must never look dead to the #527 watchdog (non-Healthy AND silent ≥ 90 s).
/// Without good keep-alives advancing <c>LastSuccessfulRead</c>, fixing R1 alone would trip it on the first
/// bad tick of any connection older than 90 s.
/// </summary>
[Fact]
public void An_idle_healthy_subscription_driver_never_looks_dead_to_the_watchdog()
{
var (drv, _, clock) = NewConnectedDriver();
using var _ = drv;
drv.AddSubscriptionForTest([SpeedRef]); // static value — no notifications ever arrive
for (var i = 0; i < 60; i++) // 5 minutes of 5 s keep-alives
{
clock.Advance(TimeSpan.FromSeconds(5));
drv.RaiseKeepAliveForTest(StatusCodes.Good);
var h = drv.GetHealth();
h.State.ShouldBe(DriverState.Healthy);
(clock.Now.UtcDateTime - h.LastSuccessfulRead!.Value).ShouldBeLessThan(TimeSpan.FromSeconds(90));
}
}
[Fact]
public async Task Recovery_on_the_same_session_republishes_the_identical_value_Good()
{
var (drv, session, clock) = NewConnectedDriver();
using var _ = drv;
var handle = drv.AddSubscriptionForTest([SpeedRef]);
drv.DeliverNotificationForTest(handle, SpeedRef, Good(42.5f, clock.Now.UtcDateTime));
SetupRead(session, 42.5f); // the server still holds the SAME value — it will not resend it
var seen = new List<DataChangeEventArgs>();
drv.OnDataChange += (_, e) => { lock (seen) seen.Add(e); };
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
clock.Advance(TimeSpan.FromSeconds(20));
drv.RaiseKeepAliveForTest(StatusCodes.Good);
drv.GetHealth().State.ShouldBe(DriverState.Healthy);
drv.GetHealth().LastSuccessfulRead.ShouldBe(clock.Now.UtcDateTime);
await WaitForAsync(() => { lock (seen) return seen.Count >= 2; }, "the recovery read must publish");
lock (seen)
{
seen.Count.ShouldBe(2);
seen[0].Snapshot.StatusCode.ShouldBe(BadNoCommunication);
seen[1].FullReference.ShouldBe(SpeedRef);
seen[1].SubscriptionHandle.ShouldBe(handle);
seen[1].Snapshot.StatusCode.ShouldBe(0u);
seen[1].Snapshot.Value.ShouldBe(42.5f, "the IDENTICAL value must come back Good — a (value,status) dedup lets it through");
}
}
[Fact]
public async Task A_notification_during_recovery_wins_over_the_recovery_read()
{
var (drv, session, clock) = NewConnectedDriver();
using var _ = drv;
var handle = drv.AddSubscriptionForTest([SpeedRef]);
drv.DeliverNotificationForTest(handle, SpeedRef, Good(1.0f, clock.Now.UtcDateTime));
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
// A fresh notification arrives before the keep-alive recovers: the read must not overwrite it.
drv.DeliverNotificationForTest(handle, SpeedRef, Good(2.0f, clock.Now.UtcDateTime));
SetupRead(session, 1.0f);
var seen = new List<DataChangeEventArgs>();
drv.OnDataChange += (_, e) => { lock (seen) seen.Add(e); };
drv.RaiseKeepAliveForTest(StatusCodes.Good);
await Task.Delay(200, TestContext.Current.CancellationToken);
lock (seen) seen.ShouldBeEmpty();
}
[Fact]
public void A_bad_keep_alive_annotates_subscribed_alarm_conditions_quality_only_once()
{
var (drv, _, _) = NewConnectedDriver();
using var _ = drv;
drv.AddAlarmSubscriptionForTest([SpeedRef]);
var seen = new List<AlarmEventArgs>();
drv.OnAlarmEvent += (_, e) => seen.Add(e);
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
var e = seen.ShouldHaveSingleItem();
e.ConditionId.ShouldBe(SpeedRef);
e.IsQualityOnly.ShouldBeTrue("an outage says nothing about the alarm state — quality only, no transition");
e.QualityStatusCode.ShouldBe(BadNoCommunication);
}
[Fact]
public async Task Recovery_restores_alarm_condition_quality_Good()
{
var (drv, session, _) = NewConnectedDriver();
using var _ = drv;
SetupRead(session);
drv.AddAlarmSubscriptionForTest([SpeedRef]);
var seen = new List<AlarmEventArgs>();
drv.OnAlarmEvent += (_, e) => { lock (seen) seen.Add(e); };
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
drv.RaiseKeepAliveForTest(StatusCodes.Good);
await WaitForAsync(() => { lock (seen) return seen.Count >= 2; }, "recovery must re-annotate Good");
lock (seen)
{
seen[1].IsQualityOnly.ShouldBeTrue();
seen[1].QualityStatusCode.ShouldBe(0u);
}
}
[Fact]
public void A_keep_alive_from_a_superseded_session_is_ignored()
{
var (drv, _, _) = NewConnectedDriver();
using var _ = drv;
var stale = NewSession();
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication, stale.Object);
drv.GetHealth().State.ShouldBe(DriverState.Healthy);
}
[Fact]
public async Task Shutdown_clears_the_outage_so_a_fresh_session_reports_its_own_loss()
{
var (drv, _, _) = NewConnectedDriver();
using var _ = drv;
drv.AddSubscriptionForTest([SpeedRef]);
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
await drv.ShutdownAsync(TestContext.Current.CancellationToken);
// As after a watchdog reinit: a new session + subscription, and a new outage must be reported again.
var fresh = NewSession();
drv.SetSessionForTest(fresh.Object);
drv.SetNamespaceMapForTest(NamespaceMap.FromTable(new NamespaceTable(), Tags));
drv.AddSubscriptionForTest([SpeedRef]);
var seen = new List<DataChangeEventArgs>();
drv.OnDataChange += (_, e) => seen.Add(e);
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
seen.ShouldHaveSingleItem().Snapshot.StatusCode.ShouldBe(BadNoCommunication);
}
/// <summary>
/// Review M2: a recovery overtaken by a re-initialise (watchdog Shutdown + Initialize) must not leak its
/// outcome onto the NEW session — previously its failed read re-armed the outage latch, so the new session's
/// first real outage was swallowed.
/// </summary>
[Fact]
public async Task A_recovery_overtaken_by_a_reinit_does_not_latch_the_new_session()
{
var ct = TestContext.Current.CancellationToken;
var (drv, _, clock) = NewConnectedDriver();
using var _ = drv;
var handle = drv.AddSubscriptionForTest([SpeedRef]);
drv.DeliverNotificationForTest(handle, SpeedRef, Good(1.0f, clock.Now.UtcDateTime));
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
await drv.Gate.WaitAsync(ct); // park the recovery read on the gate
drv.RaiseKeepAliveForTest(StatusCodes.Good);
await drv.ShutdownAsync(ct); // the watchdog's re-initialise overtakes it …
var fresh = NewSession(); // … and a new session is up (its reads would fail)
drv.SetSessionForTest(fresh.Object);
drv.SetNamespaceMapForTest(NamespaceMap.FromTable(new NamespaceTable(), Tags));
drv.SetHealthForTest(new DriverHealth(DriverState.Healthy, clock.Now.UtcDateTime, null));
drv.AddSubscriptionForTest([SpeedRef]);
drv.Gate.Release();
await drv.RecoveryTaskForTest;
drv.GetHealth().State.ShouldBe(DriverState.Healthy, "the stale recovery must not degrade the new session");
var seen = new List<DataChangeEventArgs>();
drv.OnDataChange += (_, e) => seen.Add(e);
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
seen.ShouldHaveSingleItem("the new session's first outage must be published, not swallowed by a stale latch")
.Snapshot.StatusCode.ShouldBe(BadNoCommunication);
}
/// <summary>
/// Review M1: the subscription dies while the session lives (an outage longer than the subscription lifetime
/// but shorter than the session timeout). Keep-alives stay Good, so without detection the driver ends Healthy
/// with fresh stamps and frozen values forever. It must go Degraded, publish the loss once, and STOP advancing
/// the read stamp — so the watchdog's reinit recreates the subscription.
/// </summary>
[Fact]
public void A_subscription_that_stops_publishing_while_the_session_lives_freezes_the_stamp_and_goes_Bad()
{
var (drv, _, clock) = NewConnectedDriver();
using var _ = drv;
var handle = drv.AddSubscriptionForTest([SpeedRef]);
var otherHandle = drv.AddSubscriptionForTest([CounterRef]);
drv.DeliverNotificationForTest(handle, SpeedRef, Good(3.0f, clock.Now.UtcDateTime));
var stalled = false;
drv.IsPublishingStopped = sub => stalled && sub.DisplayName == "opcua-sub-1";
drv.PublishingStallGrace = _ => TimeSpan.FromSeconds(10);
var seen = new List<DataChangeEventArgs>();
drv.OnDataChange += (_, e) => seen.Add(e);
stalled = true;
clock.Advance(TimeSpan.FromSeconds(5));
drv.RaiseKeepAliveForTest(StatusCodes.Good); // first observation — within grace
drv.GetHealth().State.ShouldBe(DriverState.Healthy);
clock.Advance(TimeSpan.FromSeconds(10));
drv.RaiseKeepAliveForTest(StatusCodes.Good); // grace elapsed
var h = drv.GetHealth();
h.State.ShouldBe(DriverState.Degraded);
h.LastError.ShouldNotBeNull().ShouldContain("subscription lost");
var frozen = h.LastSuccessfulRead;
seen.Count(e => e.Snapshot.StatusCode == BadNoCommunication).ShouldBe(2, "every subscribed ref published Bad once");
seen.Single(e => e.FullReference == SpeedRef).Snapshot.Value.ShouldBe(3.0f);
// Nothing may revive the stamp: not good keep-alives, not a bad→good keep-alive cycle, not notifications
// from a still-healthy sibling subscription.
for (var i = 0; i < 30; i++)
{
clock.Advance(TimeSpan.FromSeconds(5));
drv.RaiseKeepAliveForTest(StatusCodes.Good);
drv.DeliverNotificationForTest(otherHandle, CounterRef, Good(i, clock.Now.UtcDateTime));
}
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
drv.RaiseKeepAliveForTest(StatusCodes.Good);
h = drv.GetHealth();
h.State.ShouldBe(DriverState.Degraded);
h.LastSuccessfulRead.ShouldBe(frozen, "the #527 watchdog needs a frozen stamp to re-initialise the driver");
seen.Count(e => e.Snapshot.StatusCode == BadNoCommunication).ShouldBe(2, "the loss is published once per connection");
}
/// <summary>Control for M1: a subscription that is briefly silent (e.g. right after an outage) and then resumes
/// within its grace is NOT declared dead.</summary>
[Fact]
public void A_subscription_silent_for_less_than_its_grace_is_not_declared_lost()
{
var (drv, _, clock) = NewConnectedDriver();
using var _ = drv;
drv.AddSubscriptionForTest([SpeedRef]);
var stalled = true;
drv.IsPublishingStopped = _ => stalled;
drv.PublishingStallGrace = _ => TimeSpan.FromSeconds(10);
drv.RaiseKeepAliveForTest(StatusCodes.Good);
clock.Advance(TimeSpan.FromSeconds(9));
drv.RaiseKeepAliveForTest(StatusCodes.Good);
stalled = false;
for (var i = 0; i < 10; i++)
{
clock.Advance(TimeSpan.FromSeconds(5));
drv.RaiseKeepAliveForTest(StatusCodes.Good);
}
stalled = true; // a NEW stall restarts the grace rather than inheriting the old one
clock.Advance(TimeSpan.FromSeconds(5));
drv.RaiseKeepAliveForTest(StatusCodes.Good);
drv.GetHealth().State.ShouldBe(DriverState.Healthy);
drv.GetHealth().LastSuccessfulRead.ShouldBe(clock.Now.UtcDateTime);
}
[Fact]
public void A_BadNoSubscription_publish_error_with_tracked_subscriptions_declares_the_subscription_lost()
{
var (drv, _, _) = NewConnectedDriver();
using var _ = drv;
drv.AddSubscriptionForTest([SpeedRef]);
var seen = new List<DataChangeEventArgs>();
drv.OnDataChange += (_, e) => seen.Add(e);
drv.RaisePublishErrorForTest(StatusCodes.BadSubscriptionIdInvalid, subscriptionId: 4242); // not ours
drv.GetHealth().State.ShouldBe(DriverState.Healthy);
drv.RaisePublishErrorForTest(StatusCodes.BadNoSubscription);
drv.GetHealth().State.ShouldBe(DriverState.Degraded);
seen.ShouldHaveSingleItem().Snapshot.StatusCode.ShouldBe(BadNoCommunication);
}
[Fact]
public void A_BadNoSubscription_with_nothing_tracked_is_not_a_loss()
{
var (drv, _, _) = NewConnectedDriver();
using var _ = drv;
drv.RaisePublishErrorForTest(StatusCodes.BadNoSubscription);
drv.GetHealth().State.ShouldBe(DriverState.Healthy);
}
[Fact]
public async Task Shutdown_clears_a_subscription_loss()
{
var (drv, _, clock) = NewConnectedDriver();
using var _ = drv;
drv.AddSubscriptionForTest([SpeedRef]);
drv.RaisePublishErrorForTest(StatusCodes.BadNoSubscription);
await drv.ShutdownAsync(TestContext.Current.CancellationToken);
drv.SetSessionForTest(NewSession().Object);
drv.SetHealthForTest(new DriverHealth(DriverState.Healthy, clock.Now.UtcDateTime, null));
clock.Advance(TimeSpan.FromSeconds(5));
drv.RaiseKeepAliveForTest(StatusCodes.Good);
drv.GetHealth().LastSuccessfulRead.ShouldBe(clock.Now.UtcDateTime, "the re-subscribed connection starts clean");
}
[Fact]
public void Sdk_log_forwarding_is_rate_limited()
{
var target = new CountingLogger();
var limited = new SdkTelemetry.RateLimitedLogger(target, max: 3, window: TimeSpan.FromHours(1));
for (var i = 0; i < 50; i++)
Microsoft.Extensions.Logging.LoggerExtensions.LogWarning(limited, "sdk warning {I}", i);
target.Count.ShouldBe(3);
}
private sealed class CountingLogger : Microsoft.Extensions.Logging.ILogger
{
public int Count;
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
public bool IsEnabled(Microsoft.Extensions.Logging.LogLevel logLevel) => true;
public void Log<TState>(Microsoft.Extensions.Logging.LogLevel logLevel, Microsoft.Extensions.Logging.EventId eventId,
TState state, Exception? exception, Func<TState, Exception?, string> formatter) => Count++;
}
}