From 57ffbf79e5609738337a77b4906483445c1796c1 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 15:28:33 -0400 Subject: [PATCH] fix(mqtt): prune LastValueCache to the authored set on Register (R06:M-2) Both ingest paths now drop cache entries whose RawPath is no longer authored when a deploy re-registers, so a removed tag's value and every non-authored subscribe reference's BadNodeIdUnknown marker no longer accumulate for the driver's lifetime. Still-authored tags keep their observed values. Refs #744 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- .../LastValueCache.cs | 24 +++++++++++++++++ .../MqttSubscriptionManager.cs | 5 ++++ .../Sparkplug/SparkplugIngestor.cs | 5 ++++ .../MqttSubscriptionManagerTests.cs | 27 +++++++++++++++++++ .../SparkplugIngestorTests.cs | 22 +++++++++++++++ 5 files changed, 83 insertions(+) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/LastValueCache.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/LastValueCache.cs index 7df9dd0e..5ba310bf 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/LastValueCache.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/LastValueCache.cs @@ -45,6 +45,30 @@ public sealed class LastValueCache _values[rawPath] = snapshot; } + /// How many RawPaths currently hold an entry. Diagnostics and the bounded-growth tests. + internal int Count => _values.Count; + + /// + /// Drops every entry whose RawPath rejects — called by each ingest path's + /// Register with "is this still an authored tag", so the cache is bounded by the CURRENT + /// authored set rather than by the union of every RawPath (and every non-authored reference ever + /// subscribed) across the driver's lifetime (R06:M-2). + /// + /// Returns for a RawPath whose entry must survive. + /// How many entries were dropped. + public int RetainOnly(Func keep) + { + ArgumentNullException.ThrowIfNull(keep); + var dropped = 0; + foreach (var rawPath in _values.Keys) + { + if (!keep(rawPath) && _values.TryRemove(rawPath, out _)) + dropped++; + } + + return dropped; + } + /// /// Read the last observed value for . Never throws: a /// null/empty or a RawPath never observed both return a 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 9e4d8f64..1504328a 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs @@ -325,6 +325,11 @@ public sealed class MqttSubscriptionManager _authored = table; _warnedRefs.Clear(); + // Bound the shared last-value cache by the CURRENT authored set (R06:M-2): a dropped tag's value + // and every non-authored reference's BadNodeIdUnknown marker would otherwise live forever. A + // still-authored tag keeps its value, so a redeploy never blanks what has been observed. + Values.RetainOnly(table.ByRawPath.ContainsKey); + // Drop state belonging to tags this deploy removed. Without this, a topic whose last tag was // deleted stays in _subscribedTopics and is dutifully re-SUBSCRIBEd on every future reconnect, // forever — and its RawPaths keep a handle mapping that can never fire again. 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 cd368eae..5d63e1b7 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 @@ -416,6 +416,11 @@ public sealed class SparkplugIngestor _authored = table; _warned.Clear(); + // Bound the shared last-value cache by the CURRENT authored set (R06:M-2): a dropped tag's value + // and every non-authored reference's BadNodeIdUnknown marker would otherwise live forever. A + // still-authored tag keeps its value, so a redeploy never blanks what has been observed. + Values.RetainOnly(table.ByRawPath.ContainsKey); + // Drop state belonging to edge nodes this deploy removed. Without this a tracker (and its // rebirth debounce slot) for a node nobody reads survives every future redeploy. foreach (var key in _trackers.Keys) 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 632b0925..906c6da4 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 @@ -664,6 +664,33 @@ public sealed class MqttSubscriptionManagerTests .ShouldBe(["f/oven/press", OvenTopic]); } + /// + /// R06:M-2. Register prunes the cache to the authored set — an unauthored reference's + /// BadNodeIdUnknown marker and a dropped tag's value go. Control: a still-authored tag keeps its + /// value across the re-register (a redeploy must not blank observed values). + /// + [Fact] + public async Task Register_PrunesCacheEntriesForReferencesNoLongerAuthored() + { + var mgr = NewManager(new FakeTransport()); + 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, "Plant/Mqtt/broker1/Nope"], TimeSpan.Zero, TestContext.Current.CancellationToken); + mgr.HandleMessage(OvenTopic, "1"u8.ToArray(), retained: false); + mgr.HandleMessage("f/oven/press", "2"u8.ToArray(), retained: false); + mgr.Values.Read("Plant/Mqtt/broker1/Nope").StatusCode.ShouldBe(BadNodeIdUnknown); + + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]); + + mgr.Values.Read(OvenTempPath).Value.ShouldBe(1.0); + mgr.Values.Read(OvenPressPath).StatusCode.ShouldBe(0x80320000u); + mgr.Values.Read("Plant/Mqtt/broker1/Nope").StatusCode.ShouldBe(0x80320000u); + mgr.Values.Count.ShouldBe(1); + } + /// /// R06:M-11. A subscribe issued before any transport is attached reaches no broker, so it must /// not be recorded as established. It used to be — "subscribed, receives nothing" — and the first diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugIngestorTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugIngestorTests.cs index 1bd3b830..43c44103 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugIngestorTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugIngestorTests.cs @@ -852,6 +852,28 @@ public sealed class SparkplugIngestorTests ing.TryResolve(TempPath, out _).ShouldBeFalse(); } + /// + /// R06:M-2. Register prunes the shared cache to the authored set: an unauthored reference's + /// BadNodeIdUnknown marker and a dropped tag's value go, so the cache cannot grow with every RawPath + /// and stray reference ever seen. Control: a still-authored tag's value survives the re-register. + /// + [Fact] + public async Task Register_PrunesCacheEntriesForReferencesNoLongerAuthored() + { + var ing = NewIngestor(); + await ing.SubscribeAsync(["Plant/Mqtt/spb/NotAThing"], TimeSpan.Zero, TestContext.Current.CancellationToken); + var kept = new DataValueSnapshot(1.0f, 0u, null, DateTime.UtcNow); + ing.Values.Update(TempPath, kept); + ing.Values.Update(PressPath, kept); + + ing.Register([Tag(TempPath, Blob(Group, Node, Device, "Temperature", "Float32"))]); + + ing.Values.Read(TempPath).ShouldBe(kept); + ing.Values.Read(PressPath).StatusCode.ShouldBe(0x80320000u); // dropped tag: no stale value + ing.Values.Read("Plant/Mqtt/spb/NotAThing").StatusCode.ShouldBe(0x80320000u); + ing.Values.Count.ShouldBe(1); + } + [Fact] public async Task SubscribeAsync_UnauthoredReference_ReportsBadNodeIdUnknown() {