From b7d20d14a8d1e81d830edb04981bf9d3d1fc3b03 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 20 Aug 2026 09:52:15 -0400 Subject: [PATCH] fix(mqtt/sparkplug): initial push on subscribe + BadNoCommunication on connection loss --- .../plans/R3-02-stale-good-family-plan.md | 17 ++ ...R3-02-stale-good-family-plan.md.tasks.json | 2 +- .../Sparkplug/SparkplugIngestor.cs | 103 ++++++- .../SparkplugIngestorRecoveryTests.cs | 268 ++++++++++++++++++ 4 files changed, 385 insertions(+), 5 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugIngestorRecoveryTests.cs diff --git a/archreview/plans/R3-02-stale-good-family-plan.md b/archreview/plans/R3-02-stale-good-family-plan.md index 8ffdb94e..15f9ef79 100644 --- a/archreview/plans/R3-02-stale-good-family-plan.md +++ b/archreview/plans/R3-02-stale-good-family-plan.md @@ -618,3 +618,20 @@ test could only pass once the production change landed. - 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. + +### T4 — Sparkplug ingestor: initial push + connection-loss stale + +- RED: `--filter "…SparkplugIngestorRecoveryTests"` ⇒ **build failure**, `CS1061: 'SparkplugIngestor' + does not contain a definition for 'OnConnectionLost'` (×2). +- GREEN: 6/6. Full `Driver.Mqtt.Tests`: 803/803. +- ⚠️ **Deviation from the plan's test rationale (recorded, not silently accepted):** the plan asks for + the identical-DBIRTH-replay assertion as this seam's vacuity guard. It is present, but **it is not a + falsifier here** — Sparkplug ingestion has no unchanged-value suppression at all (births/data + publish directly, never through `Deliver`), so that replay was already raised Good before the + change. That is now asserted explicitly by an added control, + `An_identical_birth_replay_is_raised_even_without_a_preceding_stale`, and the suite header says so. + The load-bearing assertion on this seam is the **stale** itself. +- The stale reports `BadNoCommunication` while a DDEATH keeps `BadCommunicationError`: the death is the + 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. 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 159a053f..ef1ec3b6 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 @@ -25,7 +25,7 @@ { "id": "T4", "subject": "SparkplugIngestor: initial push on subscribe + connection-loss stale; rewrite the 'values are deliberately not staled' remark", - "status": "pending", + "status": "done", "blockedBy": [ "T3" ] diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugIngestor.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugIngestor.cs index 156d356e..4ccf883c 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugIngestor.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugIngestor.cs @@ -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. + /// Superseded in part (archreview R3-02): 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 . This code keeps the narrower one it + /// always had here: the source itself announced it is offline, or a birth voided the + /// scope. /// private const uint StatusBadCommunicationError = 0x80050000u; + /// + /// The broker session this driver was observing is gone — see . + /// Distinct from , which means the source announced its + /// own death. + /// + private const uint StatusBadNoCommunication = 0x80310000u; + + /// + /// No value has ever been observed for this reference — LastValueCache.Read'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. + /// + private const uint StatusBadWaitingForInitialData = 0x80320000u; + /// The metric's value does not fit the tag's effective data type — including an explicit Sparkplug null. private const uint StatusBadTypeMismatch = 0x80740000u; @@ -317,6 +336,47 @@ public sealed class SparkplugIngestor _publishTransport = connection; connection.MessageReceived += HandleMessage; connection.Reconnected += OnReconnectedAsync; + connection.ConnectionLost += OnConnectionLost; + } + + /// + /// Marks every already-observed metric BadNoCommunication, keeping its last value, + /// and publishes the change — this driver's answer to a lost broker session (archreview + /// 05/STAB-18 shape a). Wired to by + /// ; public so tests can drive the edge directly. + /// + /// + /// + /// A death is a different event and keeps its own code. NDEATH/DDEATH is the source + /// announcing it is offline and stales with as it + /// always has; this is our transport losing the broker, and it reports + /// — the same code the host's comms-loss fan-out + /// writes onto the value nodes, so an operator sees one story from both ends. + /// + /// + /// It is what makes the reconnect recover. 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. + /// + /// + /// Runs on MQTTnet's dispatcher thread: dictionary reads and synchronous + /// invokes only, never an await. A never-observed metric and an + /// already-Bad one are both left alone. + /// + /// + 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); + } } /// @@ -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(handle); } @@ -580,10 +647,16 @@ public sealed class SparkplugIngestor /// already empty, so calling it three times during one connect costs three no-ops. /// /// - /// Values are deliberately not 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. + /// Values are not staled here — but they ARE staled on a connection loss (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 config rebuild 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 + /// (, value retained); + /// a death still stales the way it always did (). /// /// /// Why the reset is happening, for the diagnostic log. @@ -1376,6 +1449,28 @@ public sealed class SparkplugIngestor return new DataValueSnapshot(null, StatusBadTypeMismatch, stamp, serverUtc); } + /// + /// 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 + /// , which is a more precise statement than any + /// value this could republish. The cache is not re-Updated — nothing new was observed, + /// only re-delivered — so the stored timestamps stay honest. + /// + /// 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); + } + } + /// Marks every authored tag in as no longer communicating. private void FanStale(SparkplugScope scope, DateTime? sourceUtc, string reason) { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugIngestorRecoveryTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugIngestorRecoveryTests.cs new file mode 100644 index 00000000..88817a47 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugIngestorRecoveryTests.cs @@ -0,0 +1,268 @@ +using Google.Protobuf; +using Org.Eclipse.Tahu.Protobuf; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug; +using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests; + +/// +/// The two push-model edge behaviours lacked (archreview +/// 05/STAB-18), matching on the plain path: an +/// initial push of the cached snapshot on every (re)subscribe, and a +/// connection-loss stale to BadNoCommunication with the value retained. +/// +/// Where the falsifier actually is on this seam — measured, not assumed. Recovery is +/// proved with an identical DBIRTH value, as on the plain path. But Sparkplug ingestion +/// carries no unchanged-value suppression (it publishes births and data directly rather +/// than through MqttSubscriptionManager.Deliver), so that replay was already raised Good +/// before this change — pinned below by +/// . The +/// load-bearing assertion here is therefore the stale: a Sparkplug tag now reports Bad on +/// the driver's own disconnect edge instead of Good over a dead broker. +/// +/// A death is not a connection loss. DDEATH/NDEATH is the source announcing it is +/// offline and keeps its own BadCommunicationError stale, unchanged; ConnectionLost +/// is this driver's transport losing the broker and reports BadNoCommunication — the same +/// code the host's comms-loss fan-out writes. The DDEATH regression case below pins that the two +/// paths stayed distinct. +/// +public sealed class SparkplugIngestorRecoveryTests +{ + // OPC UA status codes, verified against Opc.Ua.StatusCodes rather than recalled. + private const uint Good = 0x00000000u; + private const uint BadCommunicationError = 0x80050000u; + private const uint BadNoCommunication = 0x80310000u; + private const uint BadWaitingForInitialData = 0x80320000u; + + private const string Group = "Plant1"; + private const string Node = "EdgeA"; + private const string Device = "Filler1"; + + private const string TempPath = "Plant/Mqtt/spb/Filler1Temp"; + private const string PressPath = "Plant/Mqtt/spb/Filler1Press"; + + private const ulong TempAlias = 5UL; + private const ulong PressAlias = 6UL; + + private sealed record Raised(string Reference, object? Value, uint StatusCode); + + private static List Record(SparkplugIngestor ing) + { + var raised = new List(); + ing.OnDataChange += (_, e) => raised.Add(new Raised(e.FullReference, e.Snapshot.Value, e.Snapshot.StatusCode)); + return raised; + } + + // --------------------------------------------------------------------------------- + // (A) Initial push on (re)subscribe + // --------------------------------------------------------------------------------- + + /// + /// A metric observed before this subscription existed is pushed once from the cache. This is the + /// address-space-rebuild shape: a rebuild reseeds every node BadWaitingForInitialData and + /// triggers no rebirth, so without the push a quiet metric never refills. + /// + [Fact] + public async Task SubscribeAsync_pushes_the_cached_snapshot_for_an_observed_metric() + { + var ing = await NewSubscribedIngestorAsync(); + FeedBirths(ing); + ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 2, (TempAlias, TahuDataType.Float, 7f))); + + var raised = Record(ing); + await ing.SubscribeAsync([TempPath], TimeSpan.Zero, TestContext.Current.CancellationToken); + + raised.Count.ShouldBe(1); + raised[0].Reference.ShouldBe(TempPath); + raised[0].Value.ShouldBe(7f); + raised[0].StatusCode.ShouldBe(Good); + } + + /// + /// A metric that has never been observed carries only the cache's + /// BadWaitingForInitialData placeholder, and pushing that would claim an observation the + /// driver has not made. + /// + [Fact] + public async Task SubscribeAsync_does_not_push_a_never_observed_metric() + { + var ing = await NewSubscribedIngestorAsync(); + + var raised = Record(ing); + await ing.SubscribeAsync([TempPath, PressPath], TimeSpan.Zero, TestContext.Current.CancellationToken); + + raised.ShouldBeEmpty(); + ing.Values.Read(TempPath).StatusCode.ShouldBe(BadWaitingForInitialData); + } + + // --------------------------------------------------------------------------------- + // (B) Connection-loss stale + the identical-value recovery + // --------------------------------------------------------------------------------- + + /// + /// A lost broker session marks every observed metric BadNoCommunication with its last + /// value retained. ResetSessionState's previous remark argued the opposite — that the + /// connection's health surface already reports the outage — but health is not quality, and an + /// OPC UA client subscribed to the metric's node sees neither the health page nor the outage. + /// + [Fact] + public async Task OnConnectionLost_stales_every_observed_metric_keeping_its_value() + { + var ing = await NewSubscribedIngestorAsync(); + FeedBirths(ing); + ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 2, (TempAlias, TahuDataType.Float, 7f))); + + var raised = Record(ing); + ing.OnConnectionLost(); + + raised.ShouldContain(r => r.Reference == TempPath && r.StatusCode == BadNoCommunication && Equals(r.Value, 7f)); + ing.Values.Read(TempPath).StatusCode.ShouldBe(BadNoCommunication); + ing.Values.Read(TempPath).Value.ShouldBe(7f); + } + + /// + /// The contract test for this seam. The reconnect replays NBIRTH/DBIRTH carrying the + /// identical metric value, and it must be raised Good. Before the stale, the birth's + /// value equalled the cached one and the whole tag stayed Bad until the plant moved. + /// + [Fact] + public async Task After_the_stale_an_identical_birth_replay_is_raised_Good() + { + var ing = await NewSubscribedIngestorAsync(); + FeedBirths(ing, tempBirthValue: 7f); + ing.OnConnectionLost(); + ing.Values.Read(TempPath).StatusCode.ShouldBe(BadNoCommunication); + + var raised = Record(ing); + FeedBirths(ing, tempBirthValue: 7f, bdSeq: 2); // the reconnect's replayed births, value unchanged + + raised.ShouldContain(r => r.Reference == TempPath && r.StatusCode == Good && Equals(r.Value, 7f)); + } + + /// + /// Vacuity control, and it changes what the test above proves. Unlike the plain path, + /// Sparkplug ingestion has no unchanged-value suppression at all — a birth republishes its + /// metrics regardless — so an identical birth replay would have been raised Good even before + /// the stale existed. This asserts that directly. The consequence: on THIS seam the + /// load-bearing assertion is the stale itself (a Sparkplug tag now reports Bad on the driver's + /// own disconnect edge instead of Good over a dead broker), and the replay test guards only + /// that the stale is not sticky. + /// + [Fact] + public async Task An_identical_birth_replay_is_raised_even_without_a_preceding_stale() + { + var ing = await NewSubscribedIngestorAsync(); + FeedBirths(ing, tempBirthValue: 7f); + + var raised = Record(ing); + FeedBirths(ing, tempBirthValue: 7f, bdSeq: 2); + + raised.ShouldContain(r => r.Reference == TempPath && r.StatusCode == Good && Equals(r.Value, 7f)); + } + + /// + /// Regression: a DDEATH still stales with BadCommunicationError, the source-announced + /// code. The connection-loss stale is a second, distinct path — it did not replace this one. + /// + [Fact] + public async Task DDeath_still_stales_with_BadCommunicationError() + { + var ing = await NewSubscribedIngestorAsync(); + FeedBirths(ing, tempBirthValue: 7f); + + var raised = Record(ing); + ing.Dispatch(Topic(SparkplugMessageType.DDEATH, Device), Ddeath(seq: 2)); + + raised.ShouldContain(r => r.Reference == TempPath && r.StatusCode == BadCommunicationError); + ing.Values.Read(TempPath).StatusCode.ShouldBe(BadCommunicationError); + } + + // ================================================================================= + // Fixtures — deliberately a local copy of SparkplugIngestorTests' helpers (both are + // private to their own class; sharing them would make one suite's edits ripple into + // the other's fixtures). + // ================================================================================= + + private static RawTagEntry Tag(string rawPath, string tagConfig) => new(rawPath, tagConfig, WriteIdempotent: false); + + private static string Blob(string group, string node, string? device, string metric, string dataType) + { + var deviceKey = device is null ? "" : $"""{'"'}deviceId{'"'}:"{device}","""; + return $$"""{"groupId":"{{group}}","edgeNodeId":"{{node}}",{{deviceKey}}"metricName":"{{metric}}","dataType":"{{dataType}}"}"""; + } + + private static async Task NewSubscribedIngestorAsync() + { + var options = new MqttDriverOptions + { + Mode = MqttMode.SparkplugB, + Sparkplug = new MqttSparkplugOptions { GroupId = Group, HostId = "otopcua-host-1" }, + }; + var ingestor = new SparkplugIngestor(options, "mqtt-spb-1", rebirthDebounce: TimeSpan.Zero); + ingestor.Register( + [ + Tag(TempPath, Blob(Group, Node, Device, "Temperature", "Float32")), + Tag(PressPath, Blob(Group, Node, Device, "Pressure", "Float32")), + ]); + + await ingestor.SubscribeAsync([.. ingestor.AuthoredRawPaths], TimeSpan.Zero, TestContext.Current.CancellationToken); + return ingestor; + } + + private static void FeedBirths(SparkplugIngestor ing, float tempBirthValue = 0f, ulong bdSeq = 1) + { + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 0, bdSeq)); + ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Dbirth(seq: 1, tempBirthValue)); + } + + private static SparkplugTopic Topic(SparkplugMessageType type, string? device = null) => + new(type, Group, Node, device, null); + + private static SparkplugPayload Decode(Payload payload) => SparkplugCodec.Decode(payload.ToByteArray()); + + private static SparkplugPayload Nbirth(ulong seq, ulong bdSeq) + { + var payload = new Payload { Seq = seq, Timestamp = 1721822400000UL }; + payload.Metrics.Add(new Payload.Types.Metric + { + Name = "bdSeq", + Datatype = (uint)TahuDataType.Uint64, + LongValue = bdSeq, + }); + return Decode(payload); + } + + private static SparkplugPayload Dbirth(ulong seq, float tempValue) + { + var payload = new Payload { Seq = seq, Timestamp = 1721822400000UL }; + payload.Metrics.Add(BirthMetric("Temperature", TempAlias, tempValue)); + payload.Metrics.Add(BirthMetric("Pressure", PressAlias, 0f)); + return Decode(payload); + } + + private static SparkplugPayload Ddata(ulong seq, params (ulong Alias, TahuDataType Type, float Value)[] metrics) + { + var payload = new Payload { Seq = seq, Timestamp = 1721822405000UL }; + foreach (var (alias, _, value) in metrics) + { + // Deliberately NO Name and NO Datatype: the real post-birth wire shape. + payload.Metrics.Add(new Payload.Types.Metric { Alias = alias, FloatValue = value }); + } + + return Decode(payload); + } + + private static SparkplugPayload Ddeath(ulong seq) => + Decode(new Payload { Seq = seq, Timestamp = 1721822405000UL }); + + private static Payload.Types.Metric BirthMetric(string name, ulong alias, float value) => new() + { + Name = name, + Alias = alias, + Datatype = (uint)TahuDataType.Float, + FloatValue = value, + }; +}