From 6874a562f31d8b0fc63f0f8c643d308a08a54bac Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 20 Aug 2026 09:55:11 -0400 Subject: [PATCH] test+docs(mqtt): end-to-end Bad-then-identical-replay recovery --- .../plans/R3-02-stale-good-family-plan.md | 16 ++++ ...R3-02-stale-good-family-plan.md.tasks.json | 2 +- docs/drivers/Mqtt.md | 35 +++++++++ .../MqttDriverSessionLifecycleTests.cs | 73 +++++++++++++++++++ 4 files changed, 125 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 15f9ef79..85e6f2f7 100644 --- a/archreview/plans/R3-02-stale-good-family-plan.md +++ b/archreview/plans/R3-02-stale-good-family-plan.md @@ -635,3 +635,19 @@ test could only pass once the production change landed. source announcing its own offline state, the stale is our transport losing the broker (and matches what the host's S15 fan-out writes). The existing const's "no driver emits BadNoCommunication at all" remark was corrected rather than left to rot. + +### T5 — MQTT driver-level end-to-end recovery + docs + +- The test is green on first run by construction (T2–T4 already landed), so the plan's TDD bar was met + the other way: **the wiring was deliberately broken and the test seen to fail.** Deleting + `connection.ConnectionLost += OnConnectionLost;` from `MqttSubscriptionManager.AttachTo` ⇒ + `ABrokerDrop_MarksCachedRefsBadNoCommunication_AndTheIdenticalReplayRecovers` **FAILED** with + `Timed out waiting for the broker drop to publish a BadNoCommunication snapshot` (15 s). Restored, + green. That is the point of this test: the manager-level suites drive `OnConnectionLost()` directly + and stay green against an unforwarded seam. +- The outage edge is a real socket drop through MiniBroker → MQTTnet `DisconnectedAsync`; ingest is fed + through `Subscriptions.HandleMessage` because MiniBroker answers CONNECT/PINGREQ only and cannot + PUBLISH. +- Docs: `docs/drivers/Mqtt.md` §"Quality on connection loss and re-subscribe" (links the contract in + `DriverLifecycle.md`, records the two distinct "not communicating" codes and the operator-visible + behaviour change). 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 ef1ec3b6..1ae9d324 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 @@ -33,7 +33,7 @@ { "id": "T5", "subject": "MQTT driver-level end-to-end recovery test (broker drop -> Bad -> identical retained replay -> Good) + docs/drivers/Mqtt.md section", - "status": "pending", + "status": "done", "blockedBy": [ "T4" ] diff --git a/docs/drivers/Mqtt.md b/docs/drivers/Mqtt.md index 0876c55a..d13bbf16 100644 --- a/docs/drivers/Mqtt.md +++ b/docs/drivers/Mqtt.md @@ -415,6 +415,41 @@ one session, and carrying it across made the host's re-subscribe resolve an empt SUBSCRIBE at all, and leave the driver Connected/Healthy and permanently deaf. Sparkplug's `SparkplugIngestor.AttachTo` resets its own session state for the same reason. +## Quality on connection loss and re-subscribe + +The quality contract this implements is stated once, in +[`docs/DriverLifecycle.md` §"The quality contract"](../DriverLifecycle.md#the-quality-contract). +Two edge behaviours make MQTT — a push-model driver — honour it. A poll driver gets both free from +`PollGroupEngine` (a fresh subscription's first poll runs with `forceRaise: true`); MQTT had neither, +which is archreview 05/STAB-18. + +- **Connection loss stales, keeping the value.** On the `Connected → Reconnecting` edge + `MqttConnection.ConnectionLost` fires, and both ingest paths mark every **already-observed** + reference `BadNoCommunication` with its **last value retained** and publish the change. A + never-observed reference (still `BadWaitingForInitialData`) and an already-Bad one are left alone. + The event does **not** fire for a failed *initial* connect or during dispose — neither is an outage. +- **Every (re)subscribe pushes the cached snapshots.** `SubscribeAsync` republishes each observed + reference's cached snapshot onto the new handle before any broker traffic. This is what refills a + node an address-space rebuild reseeded `BadWaitingForInitialData` on a quiet topic — a rebuild + triggers no retained replay and, in Sparkplug, no rebirth. + +**Why the stale is what makes recovery work.** The plain path suppresses a message whose value *and* +status equal the driver-owned cache, and that cache survives reconnects and redeploys. So the retained +message a broker replays on reconnect — routinely the identical payload — was suppressed, and every +node the host had latched Bad stayed Bad until the plant value happened to change. Staling the status +first makes the identical replay a change. (A regression test for this must replay the **identical** +payload; a different value defeats the suppression on its own and proves nothing.) + +**Two "not communicating" codes, deliberately.** `BadNoCommunication` (0x80310000) means *the link is +down* — the connection-loss stale, and what the host's comms-loss fan-out writes onto the value nodes. +`BadCommunicationError` (0x80050000) keeps its narrower Sparkplug meaning: the **source announced its +own death** (NDEATH/DDEATH), or a birth voided the scope feeding the tag. A refused SUBSCRIBE filter +also degrades to `BadCommunicationError`. + +**Operator-visible change:** an MQTT tag now goes Bad when the broker session drops, rather than +holding its last value at Good until the host's 90 s comms-loss watchdog notices — which, since +MQTTnet usually reconnects well inside that window, in practice meant never. + ## Write-through `IWritable` landed with [#508](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/508). Before it, diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverSessionLifecycleTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverSessionLifecycleTests.cs index a20c1fca..38bdeaa0 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverSessionLifecycleTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverSessionLifecycleTests.cs @@ -100,6 +100,79 @@ public sealed class MqttDriverSessionLifecycleTests "the authored 2s budget must still bound the connect, not the 15s default"); } + /// + /// End-to-end through (archreview 05/STAB-18). Deliver a value, + /// let the real broker drop the real session, and the driver publishes + /// BadNoCommunication with the value retained; feed the identical payload back — + /// the retained replay a broker sends on reconnect — and it publishes Good again. + /// + /// + /// + /// What this adds over MqttSubscriptionManagerRecoveryTests: the wiring. Those + /// drive OnConnectionLost() directly, so they stay green even if + /// MqttSubscriptionManager.AttachTo never subscribes to + /// — an unforwarded seam is this repo's + /// best-documented way to ship an inert fix. Here the edge comes from a genuine socket drop + /// through MQTTnet's own DisconnectedAsync. + /// + /// + /// Vacuity rule: the recovery payload is byte-identical to the pre-outage one. A test + /// that replays a different value passes against a build with no stale at all — the + /// value change alone would defeat the suppression — and is not the guard. + /// + /// + /// Ingest is fed through Subscriptions.HandleMessage, the method MQTTnet's dispatcher + /// calls, because answers CONNECT and PINGREQ and nothing else: it + /// can drop a session (which is what this test needs from it) but cannot PUBLISH. + /// + /// + [Fact] + public async Task ABrokerDrop_MarksCachedRefsBadNoCommunication_AndTheIdenticalReplayRecovers() + { + const string rawPath = "Plant/Mqtt/dev1/Temp"; + const string topic = "factory/oven/temp"; + const uint good = 0x00000000u; + const uint badNoCommunication = 0x80310000u; + + using var broker = new MiniBroker(); + var options = BrokerOptions(broker.Port) with + { + RawTags = + [ + new RawTagEntry( + rawPath, + $$"""{"topic":"{{topic}}","payloadFormat":"Json","jsonPath":"$.value","dataType":"Float64"}""", + WriteIdempotent: false), + ], + }; + await using var driver = new MqttDriver(options, "d", null); + await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken); + await driver.SubscribeAsync([rawPath], TimeSpan.Zero, TestContext.Current.CancellationToken); + + var raised = new List(); + driver.OnDataChange += (_, e) => { lock (raised) raised.Add(e.Snapshot); }; + + driver.Subscriptions.HandleMessage(topic, """{"value":21.5}"""u8.ToArray(), retained: false); + lock (raised) raised.Count.ShouldBe(1, "the first value must publish"); + + broker.DropAll(); + await WaitForAsync( + () => { lock (raised) return raised.Count == 2; }, + "the broker drop to publish a BadNoCommunication snapshot"); + DataValueSnapshot staled; + lock (raised) staled = raised[1]; + staled.StatusCode.ShouldBe(badNoCommunication); + staled.Value.ShouldBe(21.5, "the last known value is retained through the outage"); + + // The reconnect's retained replay — the IDENTICAL payload. + driver.Subscriptions.HandleMessage(topic, """{"value":21.5}"""u8.ToArray(), retained: true); + + DataValueSnapshot recovered; + lock (raised) { raised.Count.ShouldBe(3); recovered = raised[2]; } + recovered.StatusCode.ShouldBe(good); + recovered.Value.ShouldBe(21.5); + } + private static async Task WaitForAsync(Func condition, string what) { var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10);