fix(mqtt/sparkplug): initial push on subscribe + BadNoCommunication on connection loss

2026-08-20 09:52:15 -04:00
parent ac2f8dc801
commit b7d20d14a8
4 changed files with 385 additions and 5 deletions
@@ -101,9 +101,28 @@ public sealed class SparkplugIngestor
    ///     OpcUaClient and this driver's own plain-mode path all emit it, while no driver emits
    ///     BadNoCommunication at all (it appears once, in a CLI formatter's label map). Following the
    ///     precedent keeps one code meaning "not communicating" across the whole fleet.
+    ///     <para><b>Superseded in part (archreview R3-02):</b> <see cref="StatusBadNoCommunication"/> now
+    ///     has a distinct, fleet-wide meaning — "the link to the source is down" — written by the host's
+    ///     comms-loss fan-out and by <see cref="OnConnectionLost"/>. This code keeps the narrower one it
+    ///     always had here: the <i>source itself</i> announced it is offline, or a birth voided the
+    ///     scope.</para>
    /// </summary>
    private const uint StatusBadCommunicationError = 0x80050000u;

+    /// <summary>
+    ///     The broker session this driver was observing is gone — see <see cref="OnConnectionLost"/>.
+    ///     Distinct from <see cref="StatusBadCommunicationError"/>, which means the source announced its
+    ///     own death.
+    /// </summary>
+    private const uint StatusBadNoCommunication = 0x80310000u;
+
+    /// <summary>
+    ///     No value has ever been observed for this reference — <c>LastValueCache.Read</c>'s answer for
+    ///     a RawPath nothing has published under yet. Neither the initial push nor the connection-loss
+    ///     stale touches such a reference: both would claim an observation that never happened.
+    /// </summary>
+    private const uint StatusBadWaitingForInitialData = 0x80320000u;
+
    /// <summary>The metric's value does not fit the tag's effective data type — including an explicit Sparkplug null.</summary>
    private const uint StatusBadTypeMismatch = 0x80740000u;

@@ -317,6 +336,47 @@ public sealed class SparkplugIngestor
        _publishTransport = connection;
        connection.MessageReceived += HandleMessage;
        connection.Reconnected += OnReconnectedAsync;
+        connection.ConnectionLost += OnConnectionLost;
+    }
+
+    /// <summary>
+    ///     Marks every already-observed metric <c>BadNoCommunication</c>, <b>keeping its last value</b>,
+    ///     and publishes the change — this 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 can drive the edge directly.
+    /// </summary>
+    /// <remarks>
+    ///     <para>
+    ///         <b>A death is a different event and keeps its own code.</b> NDEATH/DDEATH is the source
+    ///         announcing it is offline and stales with <see cref="StatusBadCommunicationError"/> as it
+    ///         always has; this is <i>our</i> transport losing the broker, and it reports
+    ///         <see cref="StatusBadNoCommunication"/> — the same code the host's comms-loss fan-out
+    ///         writes onto the value nodes, so an operator sees one story from both ends.
+    ///     </para>
+    ///     <para>
+    ///         <b>It is what makes the reconnect recover.</b> A reconnect replays NBIRTH/DBIRTH carrying
+    ///         values that are routinely unchanged; the birth path publishes them, but every node the
+    ///         host latched Bad needs a change to un-latch. Staling first guarantees the replay differs.
+    ///     </para>
+    ///     <para>
+    ///         Runs on MQTTnet's dispatcher thread: dictionary reads and synchronous
+    ///         <see cref="OnDataChange"/> invokes only, never an await. A never-observed metric and an
+    ///         already-Bad one are both left alone.
+    ///     </para>
+    /// </remarks>
+    public void OnConnectionLost()
+    {
+        var now = _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>
@@ -502,6 +562,13 @@ public sealed class SparkplugIngestor
            _handleByRawPath[def.Name] = handle;
        }

+        // Initial push (05/STAB-18 shape b): every metric this driver has already observed is
+        // republished onto the new handle. A poll driver gets this free from PollGroupEngine's
+        // forceRaise on the first poll of a new subscription; Sparkplug pushed nothing, so an
+        // address-space rebuild — which reseeds every node BadWaitingForInitialData and triggers no
+        // rebirth — left a quiet metric waiting for the plant indefinitely.
+        PushObservedSnapshots(defs);
+
        return Task.FromResult<ISubscriptionHandle>(handle);
    }

@@ -580,10 +647,16 @@ public sealed class SparkplugIngestor
    ///         already empty, so calling it three times during one connect costs three no-ops.
    ///     </para>
    ///     <para>
-    ///         Values are deliberately <b>not</b> staled here. A session change is a driver-side event,
-    ///         not a statement about the plant; last observed values remain the best available answer
-    ///         until a birth lands, and the connection's own health surface already reports the outage.
-    ///         Only a death — the source saying it is offline — stales a tag.
+    ///         <b>Values are not staled here — but they ARE staled on a connection loss</b> (archreview
+    ///         R3-02; this remark used to claim the opposite and that was the stale-Good bug). What is
+    ///         still true is the narrow part: a session reset for a <i>config rebuild</i> on a live
+    ///         connection says nothing about the plant, so last observed values remain the best answer
+    ///         until a birth lands. What was wrong was the reason given for the general case — "the
+    ///         connection's own health surface already reports the outage". Health is not quality: an
+    ///         OPC UA client subscribed to the metric's node sees neither the health page nor the
+    ///         outage, and reported Good over a dead broker. The transport loss is now handled by
+    ///         <see cref="OnConnectionLost"/> (<see cref="StatusBadNoCommunication"/>, value retained);
+    ///         a death still stales the way it always did (<see cref="StatusBadCommunicationError"/>).
    ///     </para>
    /// </remarks>
    /// <param name="reason">Why the reset is happening, for the diagnostic log.</param>
@@ -1376,6 +1449,28 @@ public sealed class SparkplugIngestor
        return new DataValueSnapshot(null, StatusBadTypeMismatch, stamp, serverUtc);
    }

+    /// <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"/>, which is a more precise statement than any
+    ///     value this could republish. The cache is not re-<c>Update</c>d — nothing new was observed,
+    ///     only re-delivered — so the stored timestamps stay honest.
+    /// </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>Marks every authored tag in <paramref name="scope"/> as no longer communicating.</summary>
    private void FanStale(SparkplugScope scope, DateTime? sourceUtc, string reason)
    {