From 2c708f55600996b3f229f0479847d1078ee96a1b Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 21:42:00 -0400 Subject: [PATCH] perf(mqtt): suppress unchanged-value republish fan-out --- .../MqttSubscriptionManager.cs | 34 ++++++- .../MqttSubscriptionManagerTests.cs | 93 +++++++++++++++++++ 2 files changed, 126 insertions(+), 1 deletion(-) 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 abf760f6..8a4ed603 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs @@ -1,3 +1,4 @@ +using System.Collections; using System.Collections.Concurrent; using System.Collections.Frozen; using System.Globalization; @@ -748,8 +749,39 @@ public sealed class MqttSubscriptionManager snapshot.StatusCode); } + // Unchanged-value suppression. Every poll driver gets this dedupe for free at the edge from + // PollGroupEngine; MQTT is push-based, so a broker that timer-republishes an identical payload + // would otherwise re-trigger the ENTIRE downstream fan-out (actor hop → dependency mux → OPC UA + // sink write) per message. The cache is still freshened — a read must serve the current + // 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). + var prior = Values.Read(def.Name); + var unchanged = prior.StatusCode == snapshot.StatusCode && !ValuesAreDifferent(prior.Value, snapshot.Value); + Values.Update(def.Name, snapshot); - Raise(def.Name, snapshot); + if (!unchanged) Raise(def.Name, snapshot); + } + + /// + /// Returns true when and represent + /// different values. Mirrors PollGroupEngine.ValuesAreDifferent — deliberately duplicated + /// rather than shared, because that one is private to the poll-loop engine this push-based driver + /// does not run through. + /// + /// + /// Arrays are compared structurally: a JSON array payload decodes to a fresh array instance on + /// every message, so reference equality would report every republish as a change and suppress + /// nothing at all for exactly the tags whose payloads are largest. + /// + /// The last cached value for the reference. + /// The value just extracted from the inbound message. + /// true when the two differ. + private static bool ValuesAreDifferent(object? previous, object? current) + { + if (previous is Array prevArr && current is Array currArr) + return !StructuralComparisons.StructuralEqualityComparer.Equals(prevArr, currArr); + + return !Equals(previous, current); } /// diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerTests.cs index 9c80c78d..6ee9255a 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerTests.cs @@ -964,6 +964,99 @@ public sealed class MqttSubscriptionManagerTests transport.Calls.Count.ShouldBe(1); } + // --------------------------------------------------------------------------------- + // Unchanged-value suppression — MQTT is push-based, so the edge dedupe every poll driver + // gets from PollGroupEngine has to live here instead. + // --------------------------------------------------------------------------------- + + /// + /// A broker that timer-republishes an identical payload must not re-trigger the whole + /// downstream fan-out (actor hop → mux → OPC UA sink write). The value is still cached + /// — the republish freshens the snapshot so IReadable.ReadAsync stays current — but no + /// OnDataChange is raised. "Cached but not raised" is an established state here; see + /// . + /// + [Fact] + public async Task An_identical_republish_raises_nothing() + { + var mgr = NewManager(); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Json","jsonPath":"$.value","dataType":"Float64"}""")]); + await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + var raised = new List(); + mgr.OnDataChange += (_, e) => raised.Add(e.Snapshot); + + mgr.HandleMessage(OvenTopic, """{"value":21.5}"""u8.ToArray(), retained: false); + mgr.HandleMessage(OvenTopic, """{"value":21.5}"""u8.ToArray(), retained: false); + + raised.Count.ShouldBe(1); + raised[0].Value.ShouldBe(21.5); + + // The suppressed republish still WROTE the cache: the cached snapshot is the second message's + // instance, not the one that was raised. Asserting on the timestamp alone would be flaky — + // two dispatcher-thread calls can share a clock tick — while the instance identity is exact. + var cached = mgr.Values.Read(OvenTempPath); + cached.Value.ShouldBe(21.5); + cached.StatusCode.ShouldBe(Good); + cached.ShouldNotBeSameAs(raised[0]); + cached.ServerTimestampUtc.ShouldBeGreaterThanOrEqualTo(raised[0].ServerTimestampUtc); + } + + /// + /// The falsification control for : + /// suppression must be value-based, not once-only. An implementation that simply + /// raised the first message per reference and went quiet passes that test and takes the + /// driver permanently dark. + /// + [Fact] + public async Task A_changed_value_after_a_suppressed_republish_raises() + { + var mgr = NewManager(); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Json","jsonPath":"$.value","dataType":"Float64"}""")]); + await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + var raised = new List(); + mgr.OnDataChange += (_, e) => raised.Add(e.Snapshot.Value); + + mgr.HandleMessage(OvenTopic, """{"value":21.5}"""u8.ToArray(), retained: false); + mgr.HandleMessage(OvenTopic, """{"value":21.5}"""u8.ToArray(), retained: false); // suppressed + mgr.HandleMessage(OvenTopic, """{"value":22.5}"""u8.ToArray(), retained: false); + mgr.HandleMessage(OvenTopic, """{"value":21.5}"""u8.ToArray(), retained: false); // back again — a real change + + raised.ShouldBe([21.5, 22.5, 21.5]); + } + + /// + /// Suppression compares status as well as value, so a quality transition on an otherwise + /// unchanged reference still publishes. The case that matters is + /// DegradeTopic: a Good value followed by a refused subscription must reach the + /// subscriber as Bad, or the client keeps a stale Good forever. + /// + [Fact] + public async Task A_status_transition_still_raises() + { + var transport = new FakeTransport(); + var mgr = NewManager(transport); + mgr.Register( + [ + Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}"""), + Tag(OvenPressPath, """{"topic":"f/oven/press","payloadFormat":"Scalar","dataType":"Float64"}"""), + ]); + await mgr.SubscribeAsync([OvenTempPath, OvenPressPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + mgr.HandleMessage(OvenTopic, "21.5"u8.ToArray(), retained: false); + + var raised = new List<(string Reference, uint Status)>(); + mgr.OnDataChange += (_, e) => raised.Add((e.FullReference, e.Snapshot.StatusCode)); + + // Only the oven-temp topic is refused, so the re-subscribe is a PARTIAL rejection and does + // not throw — one topic is still granted. + transport.Reject = t => t == OvenTopic; + await mgr.OnReconnectedAsync(TestContext.Current.CancellationToken); + + raised.ShouldContain((OvenTempPath, BadCommunicationError)); + mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(BadCommunicationError); + } + /// Records every subscribe call so dedupe / re-subscribe can be asserted without a broker. private sealed class FakeTransport : IMqttSubscribeTransport {