fix(mqtt): stale observed refs BadNoCommunication on connection loss so the identical replay recovers (05/STAB-18 shape a)

archreview/plans/R3-02-stale-good-family-plan.md

@@ -608,3 +608,13 @@ test could only pass once the production change landed.
   nothing at all for an already-observed ref.
 - GREEN: 77 `MqttSubscriptionManager*` tests pass, incl. the suppression control (an identical message
   after the push is still swallowed). Full `Driver.Mqtt.Tests`: 790/790.
+
+### T3 — `MqttConnection.ConnectionLost` + plain-mode stale
+
+- RED: `--filter "…MqttSubscriptionManagerRecoveryTests|…ConnectionLost"` ⇒ **build failure**,
+  `CS1061: 'MqttSubscriptionManager' does not contain a definition for 'OnConnectionLost'` (×5) and
+  `CS1061: 'MqttConnection' does not contain a definition for 'ConnectionLost'` (×3).
+- GREEN: 11/11 (8 manager recovery + 3 connection edge). Full `Driver.Mqtt.Tests`: 797/797.
+- The event fires inside the existing `State == Connected` guard in `OnDisconnectedAsync`, which is
+  what makes "not on an initial-connect failure" and "not on dispose" true by construction rather
+  than by a second condition — `DisposeAsync` sets `Disposed` before tearing the client down.

archreview/plans/R3-02-stale-good-family-plan.md.tasks.json

@@ -17,7 +17,7 @@
     {
       "id": "T3",
       "subject": "MqttConnection.ConnectionLost event + plain-mode stale observed refs to BadNoCommunication (value retained) so the identical replay recovers (05/STAB-18 shape a)",
-      "status": "pending",
+      "status": "done",
       "blockedBy": [
         "T2"
       ]

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs

@@ -349,6 +349,31 @@ public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport,
     /// </remarks>
     public event Func<CancellationToken, Task>? Reconnected;
 
+    /// <summary>
+    ///     Raised on the <c>Connected → Reconnecting</c> edge — the session this connection had is
+    ///     gone. Ingest paths stale their cached values to <c>BadNoCommunication</c> here (archreview
+    ///     05/STAB-18): without it, MQTT reports Good over a dead broker until the host's 90 s
+    ///     comms-loss watchdog notices, and MQTTnet's own supervisor usually reconnects well inside
+    ///     that window so the outage is never surfaced at all.
+    /// </summary>
+    /// <remarks>
+    ///     <para>
+    ///         <b>Runs on MQTTnet's dispatcher thread</b> (this is raised from the library's
+    ///         <c>DisconnectedAsync</c> callback), so a handler must be non-blocking, must not do I/O
+    ///         and must not await — the same constraint <see cref="MessageReceived"/> carries. A
+    ///         throwing handler is caught and logged rather than escaping into the pump.
+    ///     </para>
+    ///     <para>
+    ///         <b>Exactly once per edge, and only for a real outage.</b> It does not fire for a failed
+    ///         <i>initial</i> connect (there was no session to lose — the values are still
+    ///         <c>BadWaitingForInitialData</c>) nor during <see cref="DisposeAsync"/> (an intentional
+    ///         teardown, which every redeploy performs). It is deliberately NOT the mirror of
+    ///         <see cref="Reconnected"/>: recovery is a value the broker republishes, never an
+    ///         assumption this connection makes on the caller's behalf.
+    ///     </para>
+    /// </remarks>
+    public event Action? ConnectionLost;
+
     /// <summary>
     ///     Raised for every inbound application message, on MQTTnet's own dispatcher thread. This
     ///     connection does no routing, parsing or typing of its own — that is the subscription
@@ -939,6 +964,19 @@ public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport,
         if (State == MqttConnectionState.Connected)
         {
             SetState(MqttConnectionState.Reconnecting);
+
+            // The quality edge (05/STAB-18): exactly here, once per outage, and never for an initial
+            // connect failure or a dispose — both of which are excluded by the state guard above.
+            // Contained, because this runs on the library's dispatcher: a throwing ingest handler must
+            // not stop the reconnect wake below.
+            try
+            {
+                ConnectionLost?.Invoke();
+            }
+            catch (Exception ex)
+            {
+                _logger?.LogError(ex, "MQTT driver '{DriverId}': a connection-lost handler threw.", _driverId);
+            }
         }
 
         // Released unconditionally rather than only when the client had been connected: a missed

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs

@@ -96,6 +96,9 @@ public sealed class MqttSubscriptionManager
     /// </summary>
     private const uint StatusBadWaitingForInitialData = 0x80320000u;
 
+    /// <summary>The broker session this driver was observing is gone — see <see cref="OnConnectionLost"/>.</summary>
+    private const uint StatusBadNoCommunication = 0x80310000u;
+
     private readonly int _defaultQos;
     private readonly string _driverId;
     private readonly ILogger? _logger;
@@ -237,6 +240,45 @@ public sealed class MqttSubscriptionManager
         _transport = connection;
         connection.MessageReceived += HandleMessage;
         connection.Reconnected += OnReconnectedAsync;
+        connection.ConnectionLost += OnConnectionLost;
+    }
+
+    /// <summary>
+    ///     Marks every already-observed reference <c>BadNoCommunication</c>, <b>keeping its last
+    ///     value</b>, and publishes the change — the driver's answer to a lost broker session
+    ///     (archreview 05/STAB-18 shape a). Wired to <see cref="MqttConnection.ConnectionLost"/> by
+    ///     <see cref="AttachTo"/>; public so tests (and Sparkplug's ingestor) can drive the edge
+    ///     directly.
+    /// </summary>
+    /// <remarks>
+    ///     <para>
+    ///         <b>This is what makes the reconnect recover.</b> The steady-state suppression in
+    ///         <see cref="Deliver"/> compares value AND status against the driver-owned cache, which
+    ///         survives reconnects and redeploys — so the retained replay of an unchanged payload was
+    ///         suppressed and every node latched Bad by the host stayed Bad until the plant value
+    ///         happened to change. Staling the status first makes that identical replay a change.
+    ///     </para>
+    ///     <para>
+    ///         A never-observed reference is left alone (it is already
+    ///         <see cref="StatusBadWaitingForInitialData"/>, which says something more precise), and an
+    ///         already-Bad one is not re-stamped. Runs on MQTTnet's dispatcher thread: dictionary work
+    ///         and synchronous <see cref="OnDataChange"/> invokes only — exactly what
+    ///         <see cref="Deliver"/> already does there — and never awaits.
+    ///     </para>
+    /// </remarks>
+    public void OnConnectionLost()
+    {
+        var now = DateTime.UtcNow;
+        foreach (var rawPath in _authored.ByRawPath.Keys)
+        {
+            var prior = Values.Read(rawPath);
+            if (prior.StatusCode == StatusBadWaitingForInitialData) continue;   // never observed
+            if ((prior.StatusCode >> 30) >= 2) continue;                        // already Bad
+
+            var staled = prior with { StatusCode = StatusBadNoCommunication, ServerTimestampUtc = now };
+            Values.Update(rawPath, staled);
+            Raise(rawPath, staled);
+        }
     }
 
     /// <summary>

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttConnectionTests.cs

@@ -615,6 +615,81 @@ public sealed class MqttConnectionTests
         blackhole.AcceptedCount.ShouldBe(acceptedAfterTheAttempt);
     }
 
+    // ---------------------------------------------------------------------------------
+    // ConnectionLost — the quality edge (archreview 05/STAB-18 shape a)
+    // ---------------------------------------------------------------------------------
+
+    /// <summary>
+    ///     <see cref="MqttConnection.ConnectionLost"/> fires exactly once per
+    ///     <c>Connected → Reconnecting</c> edge. It is the signal the ingest paths stale their cached
+    ///     values on, so an extra fire is downstream churn and a missing one leaves every tag
+    ///     reporting Good over a dead session.
+    /// </summary>
+    [Fact]
+    public async Task BrokerDropsTheConnection_FiresConnectionLostOncePerEdge()
+    {
+        using var broker = new MiniBroker();
+        await using var conn = new MqttConnection(BrokerOpts(broker.Port), driverId: "t", logger: null);
+        var losses = 0;
+        conn.ConnectionLost += () => Interlocked.Increment(ref losses);
+
+        await conn.ConnectAsync(CancellationToken.None);
+        Volatile.Read(ref losses).ShouldBe(0, "a successful connect is not a loss");
+
+        broker.DropAll();
+        (await WaitUntilAsync(() => Volatile.Read(ref losses) >= 1, TimeSpan.FromSeconds(30))).ShouldBeTrue();
+        (await WaitUntilAsync(() => conn.State == MqttConnectionState.Connected, TimeSpan.FromSeconds(30)))
+            .ShouldBeTrue();
+        Volatile.Read(ref losses).ShouldBe(1, "the reconnect itself is not a second loss");
+
+        broker.DropAll();
+        (await WaitUntilAsync(() => Volatile.Read(ref losses) >= 2, TimeSpan.FromSeconds(30)))
+            .ShouldBeTrue("a second outage is a second edge");
+    }
+
+    /// <summary>
+    ///     An initial connect that never succeeds is not a <i>loss</i> — there was no session to lose,
+    ///     and staling on it would flip a never-observed tag Bad on a shape the cache already reports
+    ///     as <c>BadWaitingForInitialData</c>.
+    /// </summary>
+    [Fact]
+    public async Task ConnectionLost_DoesNotFireOnAnInitialConnectFailure()
+    {
+        using var blackhole = new BlackholeBroker();
+        var opts = new MqttDriverOptions
+        {
+            Host = "127.0.0.1", Port = blackhole.Port, UseTls = false, ConnectTimeoutSeconds = 1,
+        };
+        await using var conn = new MqttConnection(opts, driverId: "t", logger: null);
+        var losses = 0;
+        conn.ConnectionLost += () => Interlocked.Increment(ref losses);
+
+        await Should.ThrowAsync<TimeoutException>(async () => await conn.ConnectAsync(CancellationToken.None));
+        await Task.Delay(TimeSpan.FromMilliseconds(500), TestContext.Current.CancellationToken);
+
+        Volatile.Read(ref losses).ShouldBe(0);
+    }
+
+    /// <summary>
+    ///     Dispose tears the client down, which the library reports as a disconnect. That is an
+    ///     intentional teardown, not an outage: firing here would stale every tag on an ordinary
+    ///     redeploy or shutdown.
+    /// </summary>
+    [Fact]
+    public async Task ConnectionLost_DoesNotFireOnDispose()
+    {
+        using var broker = new MiniBroker();
+        var conn = new MqttConnection(BrokerOpts(broker.Port), driverId: "t", logger: null);
+        var losses = 0;
+        conn.ConnectionLost += () => Interlocked.Increment(ref losses);
+
+        await conn.ConnectAsync(CancellationToken.None);
+        await conn.DisposeAsync();
+        await Task.Delay(TimeSpan.FromMilliseconds(500), TestContext.Current.CancellationToken);
+
+        Volatile.Read(ref losses).ShouldBe(0);
+    }
+
     // ---------------------------------------------------------------------------------
     // Task 4 — the reconnect supervisor, against a real (minimal) broker
     // ---------------------------------------------------------------------------------

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerRecoveryTests.cs

@@ -136,4 +136,94 @@ public sealed class MqttSubscriptionManagerRecoveryTests
         raised[0].Value.ShouldBe(21.5);
         raised[0].Handle.ShouldBe(second);
     }
+
+    // ---------------------------------------------------------------------------------
+    // (B) Connection-loss stale — 05/STAB-18 shape (a)
+    // ---------------------------------------------------------------------------------
+
+    /// <summary>
+    ///     A lost broker session makes every observed reference <c>BadNoCommunication</c> — with its
+    ///     last value retained, the Part 8 last-known-value convention a SCADA client expects. MQTT
+    ///     previously marked nothing at all on its own disconnect edge, so the tags stayed Good until
+    ///     the host's 90 s comms-loss watchdog noticed (if it noticed at all — MQTTnet usually
+    ///     reconnects well inside that window).
+    /// </summary>
+    [Fact]
+    public async Task OnConnectionLost_stales_every_observed_ref_keeping_its_value()
+    {
+        var mgr = NewManager();
+        mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false);
+        await mgr.SubscribeAsync([TempPath, PressPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
+
+        var raised = Record(mgr);   // recorded AFTER the subscribe, so the initial push is not counted
+        mgr.OnConnectionLost();
+
+        raised.Count.ShouldBe(1, "the never-observed ref has nothing to stale");
+        raised[0].Reference.ShouldBe(TempPath);
+        raised[0].StatusCode.ShouldBe(BadNoCommunication);
+        raised[0].Value.ShouldBe(21.5, "the last known value is retained, not nulled");
+        mgr.Values.Read(TempPath).StatusCode.ShouldBe(BadNoCommunication);
+        mgr.Values.Read(TempPath).Value.ShouldBe(21.5);
+    }
+
+    /// <summary>
+    ///     <b>The contract test for this seam.</b> After the stale, the broker's retained replay of the
+    ///     <b>identical</b> payload is raised Good — because the cached status now differs even though
+    ///     the value does not. Replaying a <em>different</em> payload here would pass against the
+    ///     pre-fix build and prove nothing; the identical payload is the whole point.
+    /// </summary>
+    [Fact]
+    public async Task After_the_stale_the_identical_retained_replay_is_raised_Good()
+    {
+        var mgr = NewManager();
+        mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false);
+        await mgr.SubscribeAsync([TempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
+        mgr.OnConnectionLost();
+
+        var raised = Record(mgr);
+        mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: true);   // the retained seed
+
+        raised.Count.ShouldBe(1);
+        raised[0].Reference.ShouldBe(TempPath);
+        raised[0].Value.ShouldBe(21.5);
+        raised[0].StatusCode.ShouldBe(Good);
+    }
+
+    /// <summary>
+    ///     A second loss with no intervening message raises nothing: the ref is already Bad, and a
+    ///     re-stamp would be pure churn on the downstream fan-out.
+    /// </summary>
+    [Fact]
+    public async Task OnConnectionLost_does_not_re_raise_an_already_bad_ref()
+    {
+        var mgr = NewManager();
+        mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false);
+        await mgr.SubscribeAsync([TempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
+        mgr.OnConnectionLost();
+
+        var raised = Record(mgr);
+        mgr.OnConnectionLost();
+
+        raised.ShouldBeEmpty();
+    }
+
+    /// <summary>
+    ///     The stale runs on MQTTnet's dispatcher thread, so it must complete synchronously and never
+    ///     await: parking that thread stalls delivery for every other subscription on the connection.
+    /// </summary>
+    [Fact]
+    public async Task OnConnectionLost_completes_on_the_calling_thread()
+    {
+        var mgr = NewManager();
+        mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false);
+        await mgr.SubscribeAsync([TempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
+
+        var callingThread = Environment.CurrentManagedThreadId;
+        var raisedOn = 0;
+        mgr.OnDataChange += (_, _) => raisedOn = Environment.CurrentManagedThreadId;
+
+        mgr.OnConnectionLost();
+
+        raisedOn.ShouldBe(callingThread);
+    }
 }