From ac2f8dc80144339352d33f5d9c4e25da466ca1ac Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 20 Aug 2026 09:46:56 -0400 Subject: [PATCH] fix(mqtt): stale observed refs BadNoCommunication on connection loss so the identical replay recovers (05/STAB-18 shape a) --- .../plans/R3-02-stale-good-family-plan.md | 10 +++ ...R3-02-stale-good-family-plan.md.tasks.json | 2 +- .../MqttConnection.cs | 38 ++++++++ .../MqttSubscriptionManager.cs | 42 +++++++++ .../MqttConnectionTests.cs | 75 ++++++++++++++++ .../MqttSubscriptionManagerRecoveryTests.cs | 90 +++++++++++++++++++ 6 files changed, 256 insertions(+), 1 deletion(-) diff --git a/archreview/plans/R3-02-stale-good-family-plan.md b/archreview/plans/R3-02-stale-good-family-plan.md index 6c3bac48..8ffdb94e 100644 --- a/archreview/plans/R3-02-stale-good-family-plan.md +++ b/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. diff --git a/archreview/plans/R3-02-stale-good-family-plan.md.tasks.json b/archreview/plans/R3-02-stale-good-family-plan.md.tasks.json index 923ae830..159a053f 100644 --- a/archreview/plans/R3-02-stale-good-family-plan.md.tasks.json +++ b/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" ] diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs index 78dbdf1f..f2a01b32 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs @@ -349,6 +349,31 @@ public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport, /// public event Func? Reconnected; + /// + /// Raised on the Connected → Reconnecting edge — the session this connection had is + /// gone. Ingest paths stale their cached values to BadNoCommunication 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. + /// + /// + /// + /// Runs on MQTTnet's dispatcher thread (this is raised from the library's + /// DisconnectedAsync callback), so a handler must be non-blocking, must not do I/O + /// and must not await — the same constraint carries. A + /// throwing handler is caught and logged rather than escaping into the pump. + /// + /// + /// Exactly once per edge, and only for a real outage. It does not fire for a failed + /// initial connect (there was no session to lose — the values are still + /// BadWaitingForInitialData) nor during (an intentional + /// teardown, which every redeploy performs). It is deliberately NOT the mirror of + /// : recovery is a value the broker republishes, never an + /// assumption this connection makes on the caller's behalf. + /// + /// + public event Action? ConnectionLost; + /// /// 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 diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs index 47919324..57a1fe39 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs @@ -96,6 +96,9 @@ public sealed class MqttSubscriptionManager /// private const uint StatusBadWaitingForInitialData = 0x80320000u; + /// The broker session this driver was observing is gone — see . + 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; + } + + /// + /// Marks every already-observed reference BadNoCommunication, keeping its last + /// value, and publishes the change — the driver's answer to a lost broker session + /// (archreview 05/STAB-18 shape a). Wired to by + /// ; public so tests (and Sparkplug's ingestor) can drive the edge + /// directly. + /// + /// + /// + /// This is what makes the reconnect recover. The steady-state suppression in + /// 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. + /// + /// + /// A never-observed reference is left alone (it is already + /// , which says something more precise), and an + /// already-Bad one is not re-stamped. Runs on MQTTnet's dispatcher thread: dictionary work + /// and synchronous invokes only — exactly what + /// already does there — and never awaits. + /// + /// + 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); + } } /// diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttConnectionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttConnectionTests.cs index 315ca370..9311df52 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttConnectionTests.cs +++ b/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) + // --------------------------------------------------------------------------------- + + /// + /// fires exactly once per + /// Connected → Reconnecting 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. + /// + [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"); + } + + /// + /// An initial connect that never succeeds is not a loss — 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 BadWaitingForInitialData. + /// + [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(async () => await conn.ConnectAsync(CancellationToken.None)); + await Task.Delay(TimeSpan.FromMilliseconds(500), TestContext.Current.CancellationToken); + + Volatile.Read(ref losses).ShouldBe(0); + } + + /// + /// 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. + /// + [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 // --------------------------------------------------------------------------------- diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerRecoveryTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerRecoveryTests.cs index 791df347..66328430 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerRecoveryTests.cs +++ b/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) + // --------------------------------------------------------------------------------- + + /// + /// A lost broker session makes every observed reference BadNoCommunication — 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). + /// + [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); + } + + /// + /// The contract test for this seam. After the stale, the broker's retained replay of the + /// identical payload is raised Good — because the cached status now differs even though + /// the value does not. Replaying a different payload here would pass against the + /// pre-fix build and prove nothing; the identical payload is the whole point. + /// + [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); + } + + /// + /// 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. + /// + [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(); + } + + /// + /// 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. + /// + [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); + } }