From b028322568a692f108a55f76307e00e835587caa Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sun, 11 Oct 2026 08:41:27 -0400 Subject: [PATCH] =?UTF-8?q?fix(opcuaclient):=20#670=20review=20=E2=80=94?= =?UTF-8?q?=20subscription=20loss=20with=20a=20live=20session,=20stale=20r?= =?UTF-8?q?ecovery=20latch?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- docs/drivers/OpcUaClient.md | 16 +- .../OpcUaClientDriver.cs | 187 ++++++++++++++++-- .../SdkTelemetry.cs | 72 ++++++- ...B.MOM.WW.OtOpcUa.Driver.OpcUaClient.csproj | 2 + .../OpcUaClientSessionLossTests.cs | 176 +++++++++++++++++ .../OpcUaClientDriverReconnectWireTests.cs | 56 ++++++ 6 files changed, 484 insertions(+), 25 deletions(-) diff --git a/docs/drivers/OpcUaClient.md b/docs/drivers/OpcUaClient.md index b9abdd31..4c2ffd8b 100644 --- a/docs/drivers/OpcUaClient.md +++ b/docs/drivers/OpcUaClient.md @@ -78,12 +78,24 @@ The session keep-alive (`KeepAliveInterval`, default 5 s) is the liveness signal re-subscribes data + alarms; the new subscription's initial values come back Good. Alarm last-told state survives the re-initialise (#619), so the post-subscribe `ConditionRefresh` raises no un-ack flicker; the host's reconnect annotation restores condition quality. +- **The subscription dies but the session lives** (an outage longer than the subscription lifetime — + `LifetimeCount` x publishing interval, often well under the session timeout — or a server that drops it) ⇒ keep-alives + stay Good, so they cannot be the signal. A `PublishError` of `BadNoSubscription` (while subscriptions are tracked) or + `BadSubscriptionIdInvalid` naming a tracked subscription, or a tracked subscription whose SDK `PublishingStopped` + persists for one more subscription keep-alive window + 5 s across good keep-alives, declares it lost: `Degraded` + (`"subscription lost: …"`), the outage published once, and — sticky until the session is released — **nothing** + advances `LastSuccessfulRead` any more (not good keep-alives, not a sibling subscription's notifications), so the + watchdog's re-initialise recreates it. Both detectors are wire-tested independently. +- An in-place recovery is bound to the session whose keep-alive recovered: one overtaken by a re-initialise neither + publishes, nor degrades, nor re-arms the outage latch of the new session. There is **no in-driver reconnect** any more. The SDK's `SessionReconnectHandler` (removed) retried forever in SDK 1.5.378 (its "exhausted" path was unreachable), recreated sessions by cloning subscriptions the driver never re-tracked, and kept health `Healthy` throughout — so on docker-dev a stopped opc-plc left nodes Good on their last -value indefinitely and they never recovered. A teardown against a lost session skips the wire deletes and bounds -`CloseSession` to 5 s. SDK warnings (keep-alive / publish / channel errors) now reach the driver's own logger. +value indefinitely and they never recovered. A teardown against a lost session skips the wire deletes and `CloseSession` +(bounded to 5 s otherwise). SDK warnings (keep-alive / publish / channel errors) now reach the driver's own logger, +rate-limited to 20 per minute with a suppressed-count note; the per-driver SDK logger factory is disposed with the +driver. ### Namespace assignment (retired, Gitea #648) 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 671ed97a..f054508a 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs @@ -161,6 +161,15 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// The session instance to install, or null to clear it. internal void SetSessionForTest(ISession? session) => Session = session; + /// Test-only seam (#670): the most recent in-place recovery task. + internal Task RecoveryTaskForTest => _recoveryTask; + + /// Test-only seam (#670 review M1): raise the session's PublishError as the SDK publish loop does. + /// The publish error status. + /// The subscription id the error names (0 = none). + internal void RaisePublishErrorForTest(StatusCode status, uint subscriptionId = 0) => + HandlePublishError(Session, status, subscriptionId); + /// Test-only seam (#670): install a health snapshot as a completed connect would. /// The health to install. internal void SetHealthForTest(DriverHealth health) => _health = health; @@ -256,6 +265,33 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// private int _sessionLost; + /// + /// #670 review M1 — 1 once a tracked subscription is known dead while the SESSION lives (it stopped + /// publishing past its keep-alive window, or the server said BadNoSubscription / BadSubscriptionIdInvalid — + /// e.g. an outage longer than the subscription lifetime but shorter than the session timeout). Sticky until + /// the session is released: health stays Degraded and NOTHING advances + /// (good keep-alives would otherwise keep a driver whose data plane is dead looking alive forever), so the + /// host's #527 watchdog re-initialises the driver and the re-subscribe recreates the subscription. + /// + private int _subscriptionLost; + + /// #670 review M1 — ticks of the first good keep-alive that saw a tracked subscription not publishing + /// (0 = none stalled). Only the keep-alive thread and the session release touch it. + private long _publishingStoppedSinceTicks; + + /// The last in-place recovery started (#670). Test seam: lets a test await it deterministically. + private Task _recoveryTask = Task.CompletedTask; + + /// Test seam (#670 review M1): whether a subscription has stopped publishing. Production asks the SDK. + internal Func IsPublishingStopped { get; set; } = static s => s.Created && s.PublishingStopped; + + /// Test seam (#670 review M1): how long a subscription may keep reporting stopped (measured on the + /// driver's keep-alive cadence) before it is declared dead. Production: one more subscription keep-alive window + /// plus 5 s, on top of the window the SDK's own PublishingStopped already waits. + internal Func PublishingStallGrace { get; set; } = static s => + TimeSpan.FromMilliseconds(Math.Max(s.CurrentPublishingInterval, 1) * (s.CurrentKeepAliveCount + 1d)) + + TimeSpan.FromSeconds(5); + /// #670 — the last value delivered per subscribed RawPath, so an outage can publish /// BadNoCommunication with the value retained (the same shape the host's own comms-loss fan uses). private readonly System.Collections.Concurrent.ConcurrentDictionary _lastValues = @@ -898,10 +934,12 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit } // #670: bound the close — against a dead upstream (the comms-loss watchdog's whole use case) an unbounded - // CloseSession would sit out the full operation timeout on every reconnect attempt. + // CloseSession would sit out the full operation timeout on every reconnect attempt — and skip it outright + // when the keep-alive already says the session is unreachable (review L3). + var sessionKnownLost = Volatile.Read(ref _sessionLost) == 1; try { - if (sessionToClose is Session s) + if (sessionToClose is Session s && !sessionKnownLost) { using var closeCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); closeCts.CancelAfter(SessionCloseTimeout); @@ -911,6 +949,8 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit catch { /* best-effort */ } try { sessionToClose?.Dispose(); } catch { } Interlocked.Exchange(ref _sessionLost, 0); + Interlocked.Exchange(ref _subscriptionLost, 0); + Interlocked.Exchange(ref _publishingStoppedSinceTicks, 0); } /// Upper bound on the best-effort wire calls a teardown makes against a possibly-dead upstream. @@ -1002,9 +1042,11 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// /// The read itself, also used by the #670 in-place recovery. TransportOk is false when the wire read /// did not happen or threw (every result is then a local BadCommunicationError, not upstream data). + /// , when given, must still be the current session inside the gate — a + /// recovery overtaken by a re-initialise neither reads on, nor writes the health of, the new session. /// private async Task<(DataValueSnapshot[] Results, bool TransportOk)> ReadCoreAsync( - IReadOnlyList fullReferences, CancellationToken cancellationToken) + IReadOnlyList fullReferences, CancellationToken cancellationToken, ISession? expectedSession = null) { var results = new DataValueSnapshot[fullReferences.Count]; var now = Clock.GetUtcNow().UtcDateTime; @@ -1017,7 +1059,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit // session. NodeId parsing is namespace-relative, so it must also use the current // session's namespace table. var session = Session; - if (session is null) + if (session is null || (expectedSession is not null && !ReferenceEquals(session, expectedSession))) { for (var i = 0; i < fullReferences.Count; i++) results[i] = new DataValueSnapshot(null, StatusBadCommunicationError, null, now); @@ -1694,13 +1736,29 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// private void RecordLivenessEvidence() { + // A dead subscription (M1) must keep the stamp frozen even while OTHER subscriptions still deliver. + if (Volatile.Read(ref _subscriptionLost) == 1) return; var now = Clock.GetUtcNow().UtcDateTime; var last = Interlocked.Read(ref _lastEvidenceStampTicks); if (now.Ticks - last < TimeSpan.TicksPerSecond && last != 0) return; Interlocked.Exchange(ref _lastEvidenceStampTicks, now.Ticks); - var h = _health; - if (h.State is DriverState.Healthy or DriverState.Degraded) - _health = h with { LastSuccessfulRead = now }; + UpdateHealth(h => h.State is DriverState.Healthy or DriverState.Degraded ? h with { LastSuccessfulRead = now } : null); + } + + /// + /// Read-modify-write of as a CAS loop (#670 review L1): the keep-alive timer, publish + /// threads and the recovery task all derive the next health from the current one. + /// + /// Maps the current health to the next one, or null to leave it unchanged. + private void UpdateHealth(Func next) + { + while (true) + { + var current = Volatile.Read(ref _health); + var proposed = next(current); + if (proposed is null) return; + if (ReferenceEquals(Interlocked.CompareExchange(ref _health, proposed, current), current)) return; + } } /// @@ -2475,7 +2533,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit foreach (var nd in data) _diag.Record(OpcUaClientDiagnostics.Notification, CountNotifications(nd)); }; - _publishErrorHandler ??= (_, _) => _diag.Record(OpcUaClientDiagnostics.PublishError); + _publishErrorHandler ??= OnPublishError; session.Notification -= _notificationHandler; session.Notification += _notificationHandler; session.PublishError -= _publishErrorHandler; @@ -2523,29 +2581,43 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit if (!ServiceResult.IsBad(e.Status)) { + TransitionTo(HostState.Running); + // M1: the session answers but a subscription is dead — no stamp, no recovery; the watchdog reinit is the fix. + if (Volatile.Read(ref _subscriptionLost) == 1) + { + Interlocked.Exchange(ref _sessionLost, 0); + return; + } + if (Interlocked.Exchange(ref _sessionLost, 0) == 1) { - _health = new DriverHealth(DriverState.Healthy, now, null); - TransitionTo(HostState.Running); + UpdateHealth(_ => new DriverHealth(DriverState.Healthy, now, null)); + // A fresh grace: every subscription reports stopped right after an outage until its next message. + Interlocked.Exchange(ref _publishingStoppedSinceTicks, 0); _diag.Record(OpcUaClientDiagnostics.KeepAliveRecovered); _logger.LogInformation( "OpcUaClientDriver '{DriverInstanceId}': upstream keep-alive recovered on the same session; " + "re-reading the refs published Bad during the outage.", _driverInstanceId); - _ = Task.Run(RecoverAfterOutageAsync); + _recoveryTask = Task.Run(() => RecoverAfterOutageAsync(sender)); return; } - var h = _health; - if (h.State is DriverState.Healthy or DriverState.Degraded) - _health = h with { LastSuccessfulRead = now }; - TransitionTo(HostState.Running); + if (FindStalledSubscription(now) is { } stalled) + { + DeclareSubscriptionLost($"subscription '{stalled}' stopped publishing while the session stayed up", now); + return; + } + + UpdateHealth(h => h.State is DriverState.Healthy or DriverState.Degraded ? h with { LastSuccessfulRead = now } : null); return; } + Interlocked.Exchange(ref _publishingStoppedSinceTicks, 0); TransitionTo(HostState.Stopped); _diag.Record(OpcUaClientDiagnostics.KeepAliveMiss); - _health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, $"keep-alive: {e.Status}"); + UpdateHealth(h => new DriverHealth(DriverState.Degraded, h.LastSuccessfulRead, $"keep-alive: {e.Status}")); if (Interlocked.Exchange(ref _sessionLost, 1) == 1) return; + if (Volatile.Read(ref _subscriptionLost) == 1) return; // already published Bad for this connection _logger.LogWarning( "OpcUaClientDriver '{DriverInstanceId}': upstream session keep-alive failed ({Status}); subscribed values " + @@ -2554,6 +2626,71 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit PublishCommsLoss(now); } + /// + /// #670 review M1 — the display name of a tracked subscription that has reported "publishing stopped" for + /// longer than its grace, measured across consecutive good keep-alives; null when none has. + /// + private string? FindStalledSubscription(DateTime now) + { + Subscription? worst = null; + var grace = TimeSpan.Zero; + foreach (var sub in _subscriptions.Values.Select(r => r.Subscription) + .Concat(_alarmSubscriptions.Values.Select(r => r.Subscription))) + { + if (!IsPublishingStopped(sub)) continue; + var g = PublishingStallGrace(sub); + if (worst is null || g > grace) { worst = sub; grace = g; } + } + + if (worst is null) + { + Interlocked.Exchange(ref _publishingStoppedSinceTicks, 0); + return null; + } + + var since = Interlocked.CompareExchange(ref _publishingStoppedSinceTicks, now.Ticks, 0); + if (since == 0) since = now.Ticks; + return now.Ticks - since >= grace.Ticks ? worst.DisplayName ?? $"#{worst.Id}" : null; + } + + /// + /// #670 review M1 — the server dropped a subscription the driver still tracks. Degraded with the read stamp + /// frozen (sticky until the session is released), and the outage published once, so the host's watchdog + /// re-initialises and re-subscribes. + /// + private void DeclareSubscriptionLost(string reason, DateTime now) + { + if (Interlocked.Exchange(ref _subscriptionLost, 1) == 1) return; + UpdateHealth(h => new DriverHealth(DriverState.Degraded, h.LastSuccessfulRead, $"subscription lost: {reason}")); + _logger.LogWarning( + "OpcUaClientDriver '{DriverInstanceId}': {Reason}; subscribed values are BadNoCommunication until the " + + "host's comms-loss watchdog re-initialises the driver and re-subscribes.", _driverInstanceId, reason); + if (Volatile.Read(ref _sessionLost) == 0) PublishCommsLoss(now); + } + + /// + /// Session PublishError: always counted (#276); a BadNoSubscription / BadSubscriptionIdInvalid naming a + /// subscription the driver tracks means the server dropped it while the session lived (#670 review M1). + /// + private void OnPublishError(object? sender, PublishErrorEventArgs e) => + HandlePublishError(sender, e.Status?.StatusCode ?? StatusCodes.Good, e.SubscriptionId); + + private void HandlePublishError(object? sender, StatusCode code, uint subscriptionId) + { + _diag.Record(OpcUaClientDiagnostics.PublishError); + if (sender is not null && !ReferenceEquals(sender, Session)) return; + bool tracked; + if (code == StatusCodes.BadNoSubscription) + tracked = !_subscriptions.IsEmpty || !_alarmSubscriptions.IsEmpty; + else if (code == StatusCodes.BadSubscriptionIdInvalid) + tracked = _subscriptions.Values.Any(r => r.Subscription.Id == subscriptionId) + || _alarmSubscriptions.Values.Any(r => r.Subscription.Id == subscriptionId); + else + return; + if (tracked) + DeclareSubscriptionLost($"server reported {code} for a tracked subscription", Clock.GetUtcNow().UtcDateTime); + } + /// /// #670 — publish the start of an outage: BadNoCommunication once per (subscription, ref) with the last /// delivered value retained, and a quality-only annotation on every subscribed alarm condition (no state @@ -2581,21 +2718,28 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// and ConditionRefresh so transitions missed during the outage are re-reported (deduplicated). A failed /// read re-arms the outage so the next good keep-alive tries again. /// - private async Task RecoverAfterOutageAsync() + /// The session whose keep-alive recovered. Every effect — publishing, re-arming the + /// latch — is conditional on it still being the current session and still out of outage (#670 review M2/L2): a + /// recovery overtaken by a re-initialise must not leak its outcome onto the NEW session. + private async Task RecoverAfterOutageAsync(ISession session) { + bool Current() => ReferenceEquals(session, Session) && Volatile.Read(ref _sessionLost) == 0; try { + if (!Current()) return; var refs = _awaitingRecovery.Keys.ToList(); - if (refs.Count > 0 && Session is not null) + if (refs.Count > 0) { - var (results, transportOk) = await ReadCoreAsync(refs, CancellationToken.None).ConfigureAwait(false); + var (results, transportOk) = await ReadCoreAsync(refs, CancellationToken.None, session).ConfigureAwait(false); if (!transportOk) { - Interlocked.Exchange(ref _sessionLost, 1); + // Re-arm only for the session this recovery belongs to. + if (ReferenceEquals(session, Session)) Interlocked.Exchange(ref _sessionLost, 1); return; } for (var i = 0; i < refs.Count; i++) { + if (!Current()) return; if (!_awaitingRecovery.TryRemove(refs[i], out _)) continue; // a notification got there first _lastValues[refs[i]] = results[i]; foreach (var rs in _subscriptions.Values) @@ -2604,6 +2748,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit } } + if (!Current()) return; AnnotateAlarmQuality(StatusGood, Clock.GetUtcNow().UtcDateTime); if (!_alarmSubscriptions.IsEmpty) await RefreshAllAlarmSubscriptionsAsync(CancellationToken.None).ConfigureAwait(false); @@ -2716,6 +2861,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit } catch { /* timeout or already disposed — proceed */ } _gate.Dispose(); + SdkTelemetry.Release(_sdkTelemetry); } /// Asynchronously disposes the driver and releases all associated resources. @@ -2744,5 +2890,6 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit } catch { /* timeout or already disposed */ } _gate.Dispose(); + SdkTelemetry.Release(_sdkTelemetry); } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/SdkTelemetry.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/SdkTelemetry.cs index d8c7a200..dbe3f79c 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/SdkTelemetry.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/SdkTelemetry.cs @@ -17,19 +17,85 @@ internal static class SdkTelemetry /// Gets the shared, provider-less telemetry context. public static ITelemetryContext Context => Instance.Value; - /// Creates a telemetry context whose SDK log output at Warning and above goes to . + /// + /// Creates a telemetry context whose SDK log output at Warning and above goes to , + /// rate-limited (a dead upstream makes the SDK warn on every keep-alive and publish for as long as the + /// outage lasts). The context owns a : release it with + /// when the driver is disposed — is not itself disposable. + /// /// The driver instance's logger. /// The telemetry context. public static ITelemetryContext ForLogger(ILogger logger) => DefaultTelemetry.Create(builder => builder .SetMinimumLevel(LogLevel.Warning) - .AddProvider(new ForwardingLoggerProvider(logger))); + .AddProvider(new ForwardingLoggerProvider(new RateLimitedLogger(logger, MaxEntriesPerWindow, Window)))); - /// Hands every SDK logger category the one driver logger. + /// Disposes the logger factory a context owns. Never the shared one. + /// The context to release. + public static void Release(ITelemetryContext context) + { + if (ReferenceEquals(context, Context)) return; + try { (context as TelemetryContextBase)?.LoggerFactory?.Dispose(); } + catch { /* best-effort */ } + } + + /// SDK log entries forwarded per before the rest are dropped and counted. + internal const int MaxEntriesPerWindow = 20; + + /// The rate-limit window. + internal static readonly TimeSpan Window = TimeSpan.FromMinutes(1); + + /// Hands every SDK logger category the one (rate-limited) driver logger. private sealed class ForwardingLoggerProvider(ILogger target) : ILoggerProvider { public ILogger CreateLogger(string categoryName) => target; public void Dispose() { } } + + /// + /// Forwards at most max entries per window; the first entry of the next window carries a note of how + /// many were suppressed. + /// + internal sealed class RateLimitedLogger(ILogger target, int max, TimeSpan window) : ILogger + { + private readonly object _lock = new(); + private DateTime _windowStart = DateTime.MinValue; + private int _count; + private int _suppressed; + + public IDisposable? BeginScope(TState state) where TState : notnull => target.BeginScope(state); + + public bool IsEnabled(LogLevel logLevel) => target.IsEnabled(logLevel); + + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, + Func formatter) + { + int suppressedNote; + lock (_lock) + { + var now = DateTime.UtcNow; + if (now - _windowStart >= window) + { + suppressedNote = _suppressed; + _windowStart = now; + _count = 0; + _suppressed = 0; + } + else suppressedNote = 0; + + if (_count >= max) + { + _suppressed++; + return; + } + _count++; + } + + if (suppressedNote > 0) + target.LogWarning("OPC UA SDK: {Suppressed} further log entries were suppressed in the last {Window}.", + suppressedNote, window); + target.Log(logLevel, eventId, state, exception, formatter); + } + } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.csproj b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.csproj index cfba9159..d8e6892c 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.csproj +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.csproj @@ -31,6 +31,8 @@ + + diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientSessionLossTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientSessionLossTests.cs index 3cde5d9c..93bdeff7 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientSessionLossTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientSessionLossTests.cs @@ -292,4 +292,180 @@ public sealed class OpcUaClientSessionLossTests seen.ShouldHaveSingleItem().Snapshot.StatusCode.ShouldBe(BadNoCommunication); } + + /// + /// 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. + /// + [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(); + 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); + } + + /// + /// 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. + /// + [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(); + 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"); + } + + /// 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. + [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(); + 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 state) where TState : notnull => null; + public bool IsEnabled(Microsoft.Extensions.Logging.LogLevel logLevel) => true; + public void Log(Microsoft.Extensions.Logging.LogLevel logLevel, Microsoft.Extensions.Logging.EventId eventId, + TState state, Exception? exception, Func formatter) => Count++; + } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverReconnectWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverReconnectWireTests.cs index 97c22141..17f6f54a 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverReconnectWireTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverReconnectWireTests.cs @@ -61,6 +61,12 @@ public sealed class OpcUaClientDriverReconnectWireTests lock (seen) seen.Single(s => s.Status == BadNoCommunication).Value.ShouldBe(10.0f); drv.GetHealth().State.ShouldNotBe(DriverState.Healthy); drv.GetHostStatuses().Single().State.ShouldNotBe(HostState.Running); + + // The read stamp must stay frozen for the whole outage — it is the watchdog's silence measure. + var frozen = drv.GetHealth().LastSuccessfulRead; + frozen.ShouldNotBeNull(); + await Task.Delay(TimeSpan.FromSeconds(3), ct); // three keep-alive intervals + drv.GetHealth().LastSuccessfulRead.ShouldBe(frozen); }); // The restarted server knows nothing of the old session, so the driver stays down on its own … @@ -95,4 +101,54 @@ public sealed class OpcUaClientDriverReconnectWireTests back[0].StatusCode.ShouldBe(0u); Convert.ToSingle(back[0].Value).ShouldBe(88.0f); } + + /// + /// #670 review M1 — the subscription dies while the SESSION lives (in production: an outage longer than the + /// subscription lifetime but shorter than the session timeout; here: the server deletes it). Keep-alives stay + /// Good throughout, so a driver that only watched them would end Healthy with fresh read stamps and a frozen + /// value forever. It must go Bad, stay non-Healthy with the stamp frozen (so the #527 watchdog fires), and the + /// watchdog's re-initialise must bring the value back. + /// + [Fact] + public async Task A_subscription_lost_while_the_session_lives_is_reported_and_recovered_by_reinit() + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await OpcUaClientDriverWireTests.StartRigAsync("drv-sublost", ct); + OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 5.0f); + + await using var drv = OpcUaClientDriverWireTests.NewDriver(rig, "wire-sublost"); + await drv.InitializeAsync("{}", ct); + var seen = new List<(uint Status, object? Value)>(); + drv.OnDataChange += (_, e) => + { + if (e.FullReference == OpcUaClientDriverWireTests.SpeedRef) + lock (seen) seen.Add((e.Snapshot.StatusCode, e.Snapshot.Value)); + }; + await drv.SubscribeAsync([OpcUaClientDriverWireTests.SpeedRef], TimeSpan.FromMilliseconds(100), ct); + await WaitAsync(() => { lock (seen) return seen.Any(s => s is (0u, 5.0f)); }, "initial value"); + + // Delete the subscription on the server behind the SDK's back; the session itself stays healthy. + var session = drv.Session.ShouldNotBeNull(); + var subId = session.Subscriptions.Single(s => s.Created).Id; + await session.DeleteSubscriptionsAsync(null, [subId], ct); + + await WaitAsync(() => { lock (seen) return seen.Any(s => s.Status == BadNoCommunication); }, + "a subscription the server dropped must be reported as comms loss", TimeSpan.FromSeconds(30)); + drv.GetHealth().State.ShouldNotBe(DriverState.Healthy); + var frozen = drv.GetHealth().LastSuccessfulRead; + await Task.Delay(TimeSpan.FromSeconds(3), ct); // good keep-alives keep arriving … + drv.GetHealth().State.ShouldNotBe(DriverState.Healthy, "… and must not revive a driver whose data plane is dead"); + drv.GetHealth().LastSuccessfulRead.ShouldBe(frozen); + drv.GetHostStatuses().Single().State.ShouldBe(HostState.Running, "the session itself is fine"); + + // The watchdog's re-initialise recreates the subscription and the value comes back Good. + await drv.ShutdownAsync(ct); + await drv.InitializeAsync("{}", ct); + int before; + lock (seen) before = seen.Count; + await drv.SubscribeAsync([OpcUaClientDriverWireTests.SpeedRef], TimeSpan.FromMilliseconds(100), ct); + await WaitAsync(() => { lock (seen) return seen.Skip(before).Any(s => s is (0u, 5.0f)); }, + "the re-created subscription must deliver the value Good"); + drv.GetHealth().State.ShouldBe(DriverState.Healthy); + } }