fix(mqtt): initial push of cached snapshots on (re)subscribe — the push-model forceRaise (05/STAB-18 shape b)

2026-08-20 09:42:25 -04:00
parent faea4ebfca
commit 7af9baab39
4 changed files with 191 additions and 1 deletions
@@ -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.
@@ -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": []
    },
    {
@@ -89,6 +89,13 @@ public sealed class MqttSubscriptionManager
    /// <summary>The broker refused, or never answered, the SUBSCRIBE for this reference's topic.</summary>
    private const uint StatusBadCommunicationError = 0x80050000u;

+    /// <summary>
+    ///     No value has ever been observed for this reference — <see cref="LastValueCache.Read"/>'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.
+    /// </summary>
+    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);
    }

+    /// <summary>
+    ///     Republishes the cached snapshot of every already-observed reference in
+    ///     <paramref name="defs"/> — the push-model equivalent of <c>PollGroupEngine</c>'s
+    ///     <c>forceRaise: true</c> on the first poll of a new subscription (05/STAB-18 shape b).
+    /// </summary>
+    /// <remarks>
+    ///     A never-observed reference is skipped: its cached status is
+    ///     <see cref="StatusBadWaitingForInitialData"/>, and republishing that would assert an
+    ///     observation this driver has not made. The cache is deliberately NOT re-<c>Update</c>d 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.
+    /// </remarks>
+    /// <param name="defs">The definitions the new handle covers.</param>
+    private void PushObservedSnapshots(IReadOnlyList<MqttTagDefinition> 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);
+        }
+    }
+
    /// <summary>
    ///     Raises <see cref="OnDataChange"/> 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,
@@ -0,0 +1,139 @@
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests;
+
+/// <summary>
+///     The push-model edge behaviours <see cref="MqttSubscriptionManager"/> lacked (archreview
+///     05/STAB-18): an <b>initial push</b> of the cached snapshot on every (re)subscribe, and a
+///     <b>connection-loss stale</b> to <c>BadNoCommunication</c> with the value retained.
+///
+///     <para><b>Why both, and why the identical payload is load-bearing.</b> 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 <em>different</em>
+///     payload passes against the pre-fix code and proves nothing; every recovery assertion here
+///     therefore replays the <b>identical</b> payload.</para>
+///
+///     <para>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.</para>
+/// </summary>
+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<Raised> Record(MqttSubscriptionManager mgr)
+    {
+        var raised = new List<Raised>();
+        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)
+    // ---------------------------------------------------------------------------------
+
+    /// <summary>
+    ///     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
+    ///     <c>forceRaise: true</c> on the first poll of a new subscription, and it is what refills a
+    ///     node the address-space rebuild reseeded <c>BadWaitingForInitialData</c> on a quiet topic.
+    /// </summary>
+    [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);
+    }
+
+    /// <summary>
+    ///     A ref that has never been observed carries only the materialiser's
+    ///     <c>BadWaitingForInitialData</c> placeholder — pushing that would claim an observation the
+    ///     driver has not made, so the initial push skips it.
+    /// </summary>
+    [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);
+    }
+
+    /// <summary>
+    ///     <b>Control.</b> 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".
+    /// </summary>
+    [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
+    }
+
+    /// <summary>
+    ///     The rebuild shape: unsubscribe and subscribe again — as every deploy does through
+    ///     <c>DriverInstanceActor</c>'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.
+    /// </summary>
+    [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);
+    }
+}