diff --git a/archreview/plans/R3-02-stale-good-family-plan.md b/archreview/plans/R3-02-stale-good-family-plan.md index d51c252d..6c3bac48 100644 --- a/archreview/plans/R3-02-stale-good-family-plan.md +++ b/archreview/plans/R3-02-stale-good-family-plan.md @@ -600,3 +600,11 @@ test could only pass once the production change landed. ⇒ **build failure**, 8 × `error CS0103: The name 'InputQuality' does not exist in the current context`. - GREEN: 13 passed. `Core.ScriptedAlarms.Tests` 94/94 green with `WorstInputStatus` re-pointed (the never-pushed/null skips stay in the alarm engine — they are alarm policy, not the shared rule). + +### T2 — MQTT plain-mode initial push on `SubscribeAsync` + +- RED: `--filter "FullyQualifiedName~MqttSubscriptionManagerRecoveryTests"` ⇒ **4 failed**, each on + `raised.Count` (0 observed, 1 expected) / `raised.Select(r => r.Reference)` — `SubscribeAsync` raised + 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. 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 24c53f7e..923ae830 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 @@ -11,7 +11,7 @@ { "id": "T2", "subject": "MQTT plain-mode initial push of cached snapshots on SubscribeAsync — push-model forceRaise (05/STAB-18 shape b)", - "status": "pending", + "status": "done", "blockedBy": [] }, { 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 4d28f0da..47919324 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs @@ -89,6 +89,13 @@ public sealed class MqttSubscriptionManager /// The broker refused, or never answered, the SUBSCRIBE for this reference's topic. private const uint StatusBadCommunicationError = 0x80050000u; + /// + /// No value has ever been observed for this reference — 's + /// answer for a RawPath nothing has published under yet. The initial push skips it: pushing + /// this would claim an observation the driver has not made. + /// + private const uint StatusBadWaitingForInitialData = 0x80320000u; + private readonly int _defaultQos; private readonly string _driverId; private readonly ILogger? _logger; @@ -356,6 +363,13 @@ public sealed class MqttSubscriptionManager _handleByRawPath[def.Name] = handle; } + // Initial push (05/STAB-18 shape b): every ref this driver has already observed is republished + // onto the new handle before any broker traffic. A poll driver gets this free — PollGroupEngine + // runs the first poll of a new subscription with forceRaise: true — and CalculationDriver does + // it explicitly; MQTT did not, so a node reseeded BadWaitingForInitialData by an address-space + // rebuild stayed there until the plant published again, which for a quiet topic is never. + PushObservedSnapshots(defs); + // One SUBSCRIBE per distinct topic, and only for topics not already live: re-SUBSCRIBEing an // established filter is legal but pointless, and it re-triggers the retained seed. var pending = defs.Where(d => !_subscribedTopics.ContainsKey(d.Topic)).ToList(); @@ -756,6 +770,12 @@ public sealed class MqttSubscriptionManager // timestamp — and only the notification is skipped. "Cached but not raised" is already a valid // state on this path (see Raise: an authored-but-unsubscribed tag does exactly that). // + // What makes this suppression SAFE is the pair of edge behaviours around it (05/STAB-18): + // PushObservedSnapshots re-delivers the cached snapshot on every (re)subscribe, and + // OnConnectionLost stales every observed ref to BadNoCommunication — so the reconnect's + // retained replay of an unchanged payload differs in STATUS and is not suppressed. Without + // those, this comparison is what pinned a comms-lost tag Bad until the plant value changed. + // // Sparkplug bypasses this entirely. SparkplugIngestor.cs updates Values and raises directly // (its own Values.Update()/Raise() calls, not through this Deliver method), so Sparkplug B // ingestion gets no unchanged-value suppression — deliberately out of scope here. @@ -788,6 +808,29 @@ public sealed class MqttSubscriptionManager return !Equals(previous, current); } + /// + /// Republishes the cached snapshot of every already-observed reference in + /// — the push-model equivalent of PollGroupEngine's + /// forceRaise: true on the first poll of a new subscription (05/STAB-18 shape b). + /// + /// + /// A never-observed reference is skipped: its cached status is + /// , and republishing that would assert an + /// observation this driver has not made. The cache is deliberately NOT re-Updated here — + /// nothing new was observed, only re-delivered, so the stored timestamps stay honest and the + /// steady-state suppression still compares against the real last observation. + /// + /// The definitions the new handle covers. + private void PushObservedSnapshots(IReadOnlyList defs) + { + for (var i = 0; i < defs.Count; i++) + { + var snapshot = Values.Read(defs[i].Name); + if (snapshot.StatusCode == StatusBadWaitingForInitialData) continue; + Raise(defs[i].Name, snapshot); + } + } + /// /// Raises for a reference, if a live subscription covers it. /// A throwing subscriber is contained so it cannot starve the rest of the fan-out — or, worse, 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 new file mode 100644 index 00000000..791df347 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerRecoveryTests.cs @@ -0,0 +1,139 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests; + +/// +/// The push-model edge behaviours lacked (archreview +/// 05/STAB-18): an initial push of the cached snapshot on every (re)subscribe, and a +/// connection-loss stale to BadNoCommunication with the value retained. +/// +/// Why both, and why the identical payload is load-bearing. The manager's steady-state +/// dedup suppresses a message whose value AND status equal the cached one, and that cache is +/// driver-owned — it survives a reconnect and a redeploy. So a broker that replays its retained +/// message unchanged after an outage raised nothing, and every node the host had latched Bad stayed +/// Bad until the plant value happened to change. A recovery test that replays a different +/// payload passes against the pre-fix code and proves nothing; every recovery assertion here +/// therefore replays the identical payload. +/// +/// The steady-state suppression itself must survive — it is what keeps a 10 Hz topic from +/// re-triggering the whole downstream fan-out per message — so each recovery case is paired with a +/// control that asserts an unchanged republish is still suppressed. +/// +public sealed class MqttSubscriptionManagerRecoveryTests +{ + // OPC UA status codes, verified against Opc.Ua.StatusCodes rather than recalled. + private const uint Good = 0x00000000u; + private const uint BadNoCommunication = 0x80310000u; + private const uint BadWaitingForInitialData = 0x80320000u; + + private const string TempPath = "Plant/Mqtt/broker1/OvenTemp"; + private const string PressPath = "Plant/Mqtt/broker1/OvenPressure"; + private const string TempTopic = "f/oven/temp"; + private const string PressTopic = "f/oven/press"; + + private static RawTagEntry Tag(string rawPath, string topic) => + new(rawPath, $$"""{"topic":"{{topic}}","payloadFormat":"Json","jsonPath":"$.value","dataType":"Float64"}""", + WriteIdempotent: false); + + private static MqttSubscriptionManager NewManager() + { + var mgr = new MqttSubscriptionManager(new MqttDriverOptions { Mode = MqttMode.Plain }, "mqtt-1"); + mgr.Register([Tag(TempPath, TempTopic), Tag(PressPath, PressTopic)]); + return mgr; + } + + private sealed record Raised(string Reference, object? Value, uint StatusCode, ISubscriptionHandle Handle); + + private static List Record(MqttSubscriptionManager mgr) + { + var raised = new List(); + mgr.OnDataChange += (_, e) => + raised.Add(new Raised(e.FullReference, e.Snapshot.Value, e.Snapshot.StatusCode, e.SubscriptionHandle)); + return raised; + } + + // --------------------------------------------------------------------------------- + // (A) Initial push on (re)subscribe — 05/STAB-18 shape (b) + // --------------------------------------------------------------------------------- + + /// + /// A ref observed before the subscription existed is pushed once, with its cached snapshot, + /// attributed to the new handle. This is the push-model equivalent of the poll engine's + /// forceRaise: true on the first poll of a new subscription, and it is what refills a + /// node the address-space rebuild reseeded BadWaitingForInitialData on a quiet topic. + /// + [Fact] + public async Task SubscribeAsync_pushes_the_cached_snapshot_for_an_observed_ref() + { + var mgr = NewManager(); + mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false); // cached; no handle yet + + var raised = Record(mgr); + var handle = await mgr.SubscribeAsync([TempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + raised.Count.ShouldBe(1); + raised[0].Reference.ShouldBe(TempPath); + raised[0].Value.ShouldBe(21.5); + raised[0].StatusCode.ShouldBe(Good); + raised[0].Handle.ShouldBe(handle); + } + + /// + /// A ref that has never been observed carries only the materialiser's + /// BadWaitingForInitialData placeholder — pushing that would claim an observation the + /// driver has not made, so the initial push skips it. + /// + [Fact] + public async Task SubscribeAsync_does_not_push_a_never_observed_ref() + { + var mgr = NewManager(); + mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false); + + var raised = Record(mgr); + await mgr.SubscribeAsync([TempPath, PressPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + raised.Select(r => r.Reference).ShouldBe([TempPath]); + mgr.Values.Read(PressPath).StatusCode.ShouldBe(BadWaitingForInitialData); + } + + /// + /// Control. The initial push must not weaken steady-state suppression: an identical + /// message arriving after the subscribe is still swallowed. Without this, "recovery works" + /// could simply mean "nothing is ever deduped any more". + /// + [Fact] + public async Task An_identical_message_after_the_initial_push_is_still_suppressed() + { + var mgr = NewManager(); + mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false); + + var raised = Record(mgr); + await mgr.SubscribeAsync([TempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false); + + raised.Count.ShouldBe(1); // the initial push only + } + + /// + /// The rebuild shape: unsubscribe and subscribe again — as every deploy does through + /// DriverInstanceActor's unsubscribe-then-subscribe — and the cached value is pushed + /// again onto the new handle rather than leaving the rebuilt node waiting for the plant. + /// + [Fact] + public async Task A_resubscribe_pushes_again() + { + var mgr = NewManager(); + mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false); + var first = await mgr.SubscribeAsync([TempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + await mgr.UnsubscribeAsync(first, TestContext.Current.CancellationToken); + + var raised = Record(mgr); + var second = await mgr.SubscribeAsync([TempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + raised.Count.ShouldBe(1); + raised[0].Value.ShouldBe(21.5); + raised[0].Handle.ShouldBe(second); + } +}