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
parent 2a8f9af5e4
commit b028322568
6 changed files with 484 additions and 25 deletions
@@ -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)

@@ -161,6 +161,15 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
    /// <param name="session">The session instance to install, or <c>null</c> to clear it.</param>
    internal void SetSessionForTest(ISession? session) => Session = session;

+    /// <summary>Test-only seam (#670): the most recent in-place recovery task.</summary>
+    internal Task RecoveryTaskForTest => _recoveryTask;
+
+    /// <summary>Test-only seam (#670 review M1): raise the session's PublishError as the SDK publish loop does.</summary>
+    /// <param name="status">The publish error status.</param>
+    /// <param name="subscriptionId">The subscription id the error names (0 = none).</param>
+    internal void RaisePublishErrorForTest(StatusCode status, uint subscriptionId = 0) =>
+        HandlePublishError(Session, status, subscriptionId);
+
    /// <summary>Test-only seam (#670): install a health snapshot as a completed connect would.</summary>
    /// <param name="health">The health to install.</param>
    internal void SetHealthForTest(DriverHealth health) => _health = health;
@@ -256,6 +265,33 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
    /// </summary>
    private int _sessionLost;

+    /// <summary>
+    ///     #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 <see cref="DriverHealth.LastSuccessfulRead"/>
+    ///     (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.
+    /// </summary>
+    private int _subscriptionLost;
+
+    /// <summary>#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.</summary>
+    private long _publishingStoppedSinceTicks;
+
+    /// <summary>The last in-place recovery started (#670). Test seam: lets a test await it deterministically.</summary>
+    private Task _recoveryTask = Task.CompletedTask;
+
+    /// <summary>Test seam (#670 review M1): whether a subscription has stopped publishing. Production asks the SDK.</summary>
+    internal Func<Subscription, bool> IsPublishingStopped { get; set; } = static s => s.Created && s.PublishingStopped;
+
+    /// <summary>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 <c>PublishingStopped</c> already waits.</summary>
+    internal Func<Subscription, TimeSpan> PublishingStallGrace { get; set; } = static s =>
+        TimeSpan.FromMilliseconds(Math.Max(s.CurrentPublishingInterval, 1) * (s.CurrentKeepAliveCount + 1d))
+        + TimeSpan.FromSeconds(5);
+
    /// <summary>#670 — the last value delivered per subscribed RawPath, so an outage can publish
    /// <c>BadNoCommunication</c> with the value retained (the same shape the host's own comms-loss fan uses).</summary>
    private readonly System.Collections.Concurrent.ConcurrentDictionary<string, DataValueSnapshot> _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);
    }

    /// <summary>Upper bound on the best-effort wire calls a teardown makes against a possibly-dead upstream.</summary>
@@ -1002,9 +1042,11 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
    /// <summary>
    ///     The read itself, also used by the #670 in-place recovery. <c>TransportOk</c> is false when the wire read
    ///     did not happen or threw (every result is then a local <c>BadCommunicationError</c>, not upstream data).
+    ///     <paramref name="expectedSession"/>, 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.
    /// </summary>
    private async Task<(DataValueSnapshot[] Results, bool TransportOk)> ReadCoreAsync(
-        IReadOnlyList<string> fullReferences, CancellationToken cancellationToken)
+        IReadOnlyList<string> 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
    /// </summary>
    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);
+    }
+
+    /// <summary>
+    ///     Read-modify-write of <see cref="_health"/> 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.
+    /// </summary>
+    /// <param name="next">Maps the current health to the next one, or null to leave it unchanged.</param>
+    private void UpdateHealth(Func<DriverHealth, DriverHealth?> 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;
+        }
    }

    /// <summary>
@@ -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);
    }

+    /// <summary>
+    ///     #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.
+    /// </summary>
+    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;
+    }
+
+    /// <summary>
+    ///     #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.
+    /// </summary>
+    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);
+    }
+
+    /// <summary>
+    ///     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).
+    /// </summary>
+    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);
+    }
+
    /// <summary>
    ///     #670 — publish the start of an outage: <c>BadNoCommunication</c> 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.
    /// </summary>
-    private async Task RecoverAfterOutageAsync()
+    /// <param name="session">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.</param>
+    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);
    }

    /// <summary>Asynchronously disposes the driver and releases all associated resources.</summary>
@@ -2744,5 +2890,6 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
        }
        catch { /* timeout or already disposed */ }
        _gate.Dispose();
+        SdkTelemetry.Release(_sdkTelemetry);
    }
 }
@@ -17,19 +17,85 @@ internal static class SdkTelemetry
    /// <summary>Gets the shared, provider-less telemetry context.</summary>
    public static ITelemetryContext Context => Instance.Value;

-    /// <summary>Creates a telemetry context whose SDK log output at Warning and above goes to <paramref name="logger"/>.</summary>
+    /// <summary>
+    ///     Creates a telemetry context whose SDK log output at Warning and above goes to <paramref name="logger"/>,
+    ///     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 <see cref="ILoggerFactory"/>: release it with <see cref="Release"/>
+    ///     when the driver is disposed — <see cref="DefaultTelemetry"/> is not itself disposable.
+    /// </summary>
    /// <param name="logger">The driver instance's logger.</param>
    /// <returns>The telemetry context.</returns>
    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))));

-    /// <summary>Hands every SDK logger category the one driver logger.</summary>
+    /// <summary>Disposes the logger factory a <see cref="ForLogger"/> context owns. Never the shared one.</summary>
+    /// <param name="context">The context to release.</param>
+    public static void Release(ITelemetryContext context)
+    {
+        if (ReferenceEquals(context, Context)) return;
+        try { (context as TelemetryContextBase)?.LoggerFactory?.Dispose(); }
+        catch { /* best-effort */ }
+    }
+
+    /// <summary>SDK log entries forwarded per <see cref="Window"/> before the rest are dropped and counted.</summary>
+    internal const int MaxEntriesPerWindow = 20;
+
+    /// <summary>The rate-limit window.</summary>
+    internal static readonly TimeSpan Window = TimeSpan.FromMinutes(1);
+
+    /// <summary>Hands every SDK logger category the one (rate-limited) driver logger.</summary>
    private sealed class ForwardingLoggerProvider(ILogger target) : ILoggerProvider
    {
        public ILogger CreateLogger(string categoryName) => target;

        public void Dispose() { }
    }
+
+    /// <summary>
+    ///     Forwards at most <c>max</c> entries per window; the first entry of the next window carries a note of how
+    ///     many were suppressed.
+    /// </summary>
+    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>(TState state) where TState : notnull => target.BeginScope(state);
+
+        public bool IsEnabled(LogLevel logLevel) => target.IsEnabled(logLevel);
+
+        public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception,
+            Func<TState, Exception?, string> 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);
+        }
+    }
 }
@@ -31,6 +31,8 @@
        <InternalsVisibleTo Include="ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests"/>
        <!-- Gitea #656: the native-alarm re-statement end-to-end tests drive this driver into the real DriverHostActor. -->
        <InternalsVisibleTo Include="ZB.MOM.WW.OtOpcUa.Runtime.Tests"/>
+        <!-- Gitea #670 review M1: the wire test deletes the driver's subscription server-side through its own session. -->
+        <InternalsVisibleTo Include="ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests"/>
    </ItemGroup>

 </Project>
@@ -292,4 +292,180 @@ public sealed class OpcUaClientSessionLossTests

        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++;
+    }
 }
@@ -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);
    }
+
+    /// <summary>
+    ///     #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.
+    /// </summary>
+    [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);
+    }
 }