From e7e84e05771dd457c4ae84fb6eefb107d96622b5 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 16:39:24 -0400 Subject: [PATCH] fix(mqtt): review minors - QoS 0 is not answer evidence, caller cancels stay cancels, rebirth wait cap, bound/trie pins (R06 G2 MQTT review minors 2 3 4 6 7 8) 2: only a PUBACKed (QoS >= 1) publish advances LastSuccessfulRead; a QoS 0 Sparkplug command completing means only that the bytes left the local socket. 3: a no-verdict SUBSCRIBE caused by the caller's own cancellation now throws OperationCanceledException (refs still Bad + pending), so the breaker no longer counts a pipeline timeout as a failure. 4: the rebirth wait is the window capped at 15 s, below the AdminUI 20 s per-call timeout. 6: a 10-topic test pins the 8-in-flight publish bound. 7: partial-level wildcards and $share filters are pinned never to confirm. 8: the intended init-vs-shutdown supersede is commented at the re-check. Refs #744 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- docs/drivers/Mqtt.md | 4 +- .../MqttBrowseSession.cs | 12 ++- .../MqttDriver.cs | 13 +++- .../MqttSubscriptionManager.cs | 13 ++++ .../MqttBirthObservationWindowTests.cs | 15 ++++ .../MqttDriverSessionLifecycleTests.cs | 73 +++++++++++++++++++ .../MqttSubscriptionManagerTests.cs | 28 ++++++- .../TopicFilterTrieTests.cs | 21 ++++++ 8 files changed, 171 insertions(+), 8 deletions(-) diff --git a/docs/drivers/Mqtt.md b/docs/drivers/Mqtt.md index d5c33835..5952b4fc 100644 --- a/docs/drivers/Mqtt.md +++ b/docs/drivers/Mqtt.md @@ -128,7 +128,7 @@ driver are pure organisational grouping in the RawPath. | `HostId` | `""` | Sparkplug Host Application identity — the `spBv1.0/STATE/{HostId}` topic. Optional for observation; **required** to claim. Set without `ActAsPrimaryHost`, the driver only observes | | `ActAsPrimaryHost` | `false` | Claim the Sparkplug Primary Host Application identity. **Primary-gated** — only the node holding the redundancy Primary role claims it. See [Primary host](#primary-host-actasprimaryhost) | | `RequestRebirthOnGap` | `true` | On a detected sequence-number gap, publish a rebirth-request NCMD instead of running on stale metric state | -| `BirthObservationWindowSeconds` | `15` | The Sparkplug browse picker's birth-collection window, clamped to 1–120 s (R06:M-4). The open budget grows by it; the open lingers while NBIRTH/DBIRTH traffic is still arriving (returning after a 1 s quiet spell — an idle plant costs 1 s); and an operator **Request rebirth** waits, bounded by it, for the births it asked for to arrive and go quiet for 500 ms, so the tree re-rendered afterwards carries them. A caller deadline during that wait still reports the publish. The runtime driver does not read it (its `DiscoverAsync` has no production caller — `SupportsOnlineDiscovery` is false) | +| `BirthObservationWindowSeconds` | `15` | The Sparkplug browse picker's birth-collection window, clamped to 1–120 s (R06:M-4). The open budget grows by it; the open lingers while NBIRTH/DBIRTH traffic is still arriving (returning after a 1 s quiet spell — an idle plant costs 1 s); and an operator **Request rebirth** waits, bounded by it and capped at 15 s (below the AdminUI's 20 s per-call timeout), for the births it asked for to arrive and go quiet for 500 ms, so the tree re-rendered afterwards carries them. A caller deadline during that wait still reports the publish. The runtime driver does not read it (its `DiscoverAsync` has no production caller — `SupportsOnlineDiscovery` is false) | `MqttDriverForm` authors all five in Sparkplug mode. The sub-object is **merged, never replaced** (a key a newer driver adds inside it survives an older AdminUI), and in `Plain` mode it is not touched at @@ -494,7 +494,7 @@ subscribe — including the device's own reconnect — so one stale write would **A batch publishes concurrently, one topic at a time (R06:M-8).** Writes to *different* topics in one `WriteAsync` go out together, at most 8 in flight, so a stalled broker costs a batch about one publish deadline rather than one per request; writes to the *same* topic stay sequential in request order, so the last one wins. -A successful publish (PUBACK, or a sent QoS 0 command) advances `LastSuccessfulRead` like an inbound message. +A PUBACKed (QoS 1) publish advances `LastSuccessfulRead` like an inbound message; a QoS 0 Sparkplug command does not, because its completion only means the bytes left the local socket. **Plain QoS 1 is at-least-once.** Know the consequences before pointing a Plain `writeTopic` at a non-idempotent device topic: diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttBrowseSession.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttBrowseSession.cs index b6bf8d9c..1b6b691c 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttBrowseSession.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttBrowseSession.cs @@ -149,6 +149,13 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession /// private static readonly UTF8Encoding Utf8Strict = new(encoderShouldEmitUTF8Identifier: false, throwOnInvalidBytes: true); + /// + /// Ceiling on the rebirth request's wait, below the AdminUI's 20 s BrowserSessionService.PerCallTimeout + /// (review minor 4): a configured window above it could never take effect there, and a rebirth aimed at an + /// offline edge node must not hold the operator's button for the whole window. + /// + internal static readonly TimeSpan MaxRebirthWait = TimeSpan.FromSeconds(15); + /// The quiet spell that ends a rebirth request's wait once its births have started arriving. private static readonly TimeSpan RebirthQuietInterval = TimeSpan.FromMilliseconds(500); @@ -225,6 +232,9 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession /// The Sparkplug birth-observation window this session was opened with (R06:M-4). internal TimeSpan BirthWindow { get; } + /// How long a rebirth request waits for its births: the window, capped at . + internal TimeSpan RebirthWait => BirthWindow < MaxRebirthWait ? BirthWindow : MaxRebirthWait; + /// How many NBIRTH/DBIRTH messages this session has observed — the settle loop's signal. internal long BirthCount => Interlocked.Read(ref _birthCount); @@ -987,7 +997,7 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession try { await SettleBirthsAsync( - () => BirthCount, baseline, awaitFirstBirth: true, RebirthQuietInterval, BirthWindow, cancellationToken) + () => BirthCount, baseline, awaitFirstBirth: true, RebirthQuietInterval, RebirthWait, cancellationToken) .ConfigureAwait(false); } catch (OperationCanceledException) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs index dec7569c..c265dce1 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs @@ -806,9 +806,11 @@ public sealed class MqttDriver .ConfigureAwait(false); results[index] = new WriteResult(0u); - // A PUBACK (or a sent QoS 0 command) is the broker answering — evidence the session works, - // so it feeds LastSuccessfulRead alongside inbound messages. - Interlocked.Exchange(ref _lastPublishUtcTicks, DateTime.UtcNow.Ticks); + // A PUBACK is the broker answering — evidence the session works, so it feeds + // LastSuccessfulRead alongside inbound messages. A completed QoS 0 publish is NOT: it only + // means the bytes reached the local socket buffer, and frequent Sparkplug commands would + // otherwise keep the watchdog's clock fresh over a half-open session (review minor 2). + if (encoded.Qos >= 1) Interlocked.Exchange(ref _lastPublishUtcTicks, DateTime.UtcNow.Ticks); } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { @@ -1118,7 +1120,10 @@ public sealed class MqttDriver await StartPrimaryHostAsync(cancellationToken).ConfigureAwait(false); // Re-checked after publishing: a shutdown whose gate wait expired between the check above and - // here captured whatever it found and left these fields for us. We still hold the gate, so no + // here captured whatever it found and left these fields for us. Intended consequence (review + // minor 8): a perfectly healthy init that is just completing when a ShutdownAsync arrives also + // releases itself and throws OperationCanceledException — the queued shutdown would tear it down + // anyway, and the actor serialises init/shutdown, so only an abandoned init can meet this. We still hold the gate, so no // newer initialize can have published anything — everything in them is ours to release. if (!IsCurrentLifecycle(generation)) { 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 0c8b1a9a..2a80209d 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs @@ -454,6 +454,10 @@ public sealed class MqttSubscriptionManager if (outcome.TransportFailure is { } failure) { + // The refs are already Bad and pending either way. But a cancel of the CALLER's token (a pipeline + // timeout, host shutdown) is a cancellation, not a broker fault: surfacing it as + // InvalidOperationException made the resilience breaker count it as a failure (review minor 3). + var callerCancelled = cancellationToken.IsCancellationRequested; // No verdict at all (transport throw / SUBSCRIBE deadline): THROW, so the caller holds no handle // and DriverInstanceActor's health-tick reconcile re-subscribes. Returning a handle here was the // silent-death case — the actor thought the subscription live and never retried. The refs already @@ -464,6 +468,15 @@ public sealed class MqttSubscriptionManager _handleByRawPath.TryRemove(new KeyValuePair(def.Name, handle)); } + if (callerCancelled) + { + throw new OperationCanceledException( + $"MQTT driver '{_driverId}': SUBSCRIBE for {pending.Count} tag(s) was cancelled by the caller; " + + "the topics stay pending and will be retried.", + failure, + cancellationToken); + } + throw new InvalidOperationException( $"MQTT driver '{_driverId}': SUBSCRIBE for {pending.Count} tag(s) got no broker verdict " + $"({failure.GetType().Name}); the subscription will be retried.", diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBirthObservationWindowTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBirthObservationWindowTests.cs index b7f920af..3132d3b2 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBirthObservationWindowTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBirthObservationWindowTests.cs @@ -43,6 +43,21 @@ public sealed class MqttBirthObservationWindowTests public void BirthWindow_IsClamped(int configured, int expectedSeconds) => MqttDriverBrowser.BirthWindow(Sparkplug(configured)).ShouldBe(TimeSpan.FromSeconds(expectedSeconds)); + /// + /// Review minor 4: the rebirth wait is capped below the AdminUI's 20 s per-call timeout, so a long + /// configured window cannot hold the button past it (and a short one still governs). + /// + [Theory] + [InlineData(120, 15)] + [InlineData(15, 15)] + [InlineData(5, 5)] + public async Task RebirthWait_IsTheWindowCappedAt15Seconds(int windowSeconds, int expectedSeconds) + { + await using var s = new MqttBrowseSession( + MqttMode.SparkplugB, client: null, logger: null, birthWindow: TimeSpan.FromSeconds(windowSeconds)); + s.RebirthWait.ShouldBe(TimeSpan.FromSeconds(expectedSeconds)); + } + /// An idle plant costs one quiet interval, not the whole window. [Fact] public async Task Settle_WithNoBirths_ReturnsAfterOneQuietInterval() diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverSessionLifecycleTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverSessionLifecycleTests.cs index 14dbda13..8abf72aa 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverSessionLifecycleTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverSessionLifecycleTests.cs @@ -361,6 +361,79 @@ public sealed class MqttDriverSessionLifecycleTests driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull(); } + /// + /// Review minor 6: the bound itself. Ten topics against a broker that never PUBACKs: exactly + /// PUBLISHes go out and the count then holds — a + /// regression to unbounded concurrency sends all ten. The stability check re-reads the count across + /// several condition polls rather than sleeping once. + /// + [Fact] + public async Task AWriteBatch_NeverHasMoreThanTheBoundInFlight() + { + using var broker = new MiniBroker(); // never PUBACKs + var options = BrokerOptions(broker.Port) with + { + ConnectTimeoutSeconds = 30, + RawTags = [.. Enumerable.Range(0, 10).Select(i => WritableTag($"Plant/Mqtt/dev1/Sp{i}", $"cmd/sp{i}"))], + }; + await using var driver = new MqttDriver(options, "d", null); + await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken); + + using var cts = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken); + var write = driver.WriteAsync( + [.. Enumerable.Range(0, 10).Select(i => new WriteRequest($"Plant/Mqtt/dev1/Sp{i}", (double)i))], + cts.Token); + + await WaitForAsync(() => broker.Published.Count >= MqttDriver.MaxConcurrentPublishes, "the bound to fill"); + for (var i = 0; i < 10; i++) + { + await Task.Delay(50, TestContext.Current.CancellationToken); + broker.Published.Count.ShouldBe(MqttDriver.MaxConcurrentPublishes, "a slot freed with no PUBACK received"); + } + + await cts.CancelAsync(); + var results = await write.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); + results.Count.ShouldBe(10); + results.ShouldAllBe(r => r.StatusCode == 0x80050000u); + } + + /// + /// Review minor 2: a QoS 0 publish (a Sparkplug command) completing only means the bytes reached the + /// local socket buffer; the broker has not answered, so it must not advance LastSuccessfulRead. + /// The positive control is , where a QoS 1 + /// PUBACK does advance it. + /// + [Fact] + public async Task ASparkplugQos0Command_DoesNotAdvanceLastSuccessfulRead() + { + using var broker = new MiniBroker(); + var options = BrokerOptions(broker.Port) with + { + Mode = MqttMode.SparkplugB, + Sparkplug = new MqttSparkplugOptions { GroupId = "G", RequestRebirthOnGap = false }, + RawTags = + [ + new RawTagEntry( + "Plant/Mqtt/spb/Setpoint", + """{"groupId":"G","edgeNodeId":"E","metricName":"Setpoint","dataType":"Float64"}""", + WriteIdempotent: false), + ], + }; + await using var driver = new MqttDriver(options, "d", null); + await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken); + var before = driver.GetHealth().LastSuccessfulRead; + var publishedBefore = broker.Published.Count; + + var results = await driver.WriteAsync( + [new WriteRequest("Plant/Mqtt/spb/Setpoint", 4.5)], TestContext.Current.CancellationToken); + + results.ShouldHaveSingleItem().StatusCode.ShouldBe(0u); + await WaitForAsync( + () => broker.Published.Skip(publishedBefore).Any(p => p.Topic == "spBv1.0/G/NCMD/E"), + "the NCMD to reach the broker — the control that the write really went out"); + driver.GetHealth().LastSuccessfulRead.ShouldBe(before); + } + private static async Task WaitForAsync(Func condition, string what) { var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10); 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 906c6da4..3757c169 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 @@ -553,7 +553,7 @@ public sealed class MqttSubscriptionManagerTests var raised = new List(); mgr.OnDataChange += (_, e) => raised.Add(e); - await Should.ThrowAsync(() => + await Should.ThrowAsync(() => mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken)); mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(BadCommunicationError); @@ -561,6 +561,32 @@ public sealed class MqttSubscriptionManagerTests raised.ShouldContain(e => e.FullReference == OvenTempPath && e.Snapshot.StatusCode == BadCommunicationError); } + /// + /// Review minor 3: a cancel of the CALLER's token (a pipeline timeout, host shutdown) is a cancellation, + /// not a broker fault — it must surface as OperationCanceledException, or the resilience breaker counts + /// it as a failure. The refs are still Bad and pending, so the next reconnect retries them (the positive + /// control). The uncancelled no-verdict case throws InvalidOperationException + /// (). + /// + [Fact] + public async Task SubscribeAsync_CallerCancelled_ThrowsOperationCanceled_AndTheTopicStaysPending() + { + var transport = new FakeTransport { Throw = new OperationCanceledException() }; + var mgr = NewManager(transport); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]); + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + + var ex = await Should.ThrowAsync(() => + mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), cts.Token)); + ex.ShouldNotBeOfType(); + mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(BadCommunicationError); + + transport.Throw = null; + await mgr.OnReconnectedAsync(TestContext.Current.CancellationToken); + transport.Calls.Last().Select(f => f.Topic).ShouldBe([OvenTopic]); + } + /// /// The reconcile path: once the actor re-subscribes after the throw, the topic IS re-SUBSCRIBEd and /// delivery resumes on the new handle — the positive control for the throw above. diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/TopicFilterTrieTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/TopicFilterTrieTests.cs index f21677e1..191b4596 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/TopicFilterTrieTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/TopicFilterTrieTests.cs @@ -69,6 +69,27 @@ public sealed class TopicFilterTrieTests collect.Seen.Contains(1).ShouldBe(expected); } + /// + /// Review minor 7: partial-level wildcards and $share/… filters are outside the trie's level rules. + /// Whatever it yields, confirmation by the broker's comparer must produce no match — pinned so a later + /// "smarter" trie cannot start delivering them. + /// + [Theory] + [InlineData("a+", "ab")] + [InlineData("a/b#", "a/bc")] + [InlineData("a/b#", "a/b")] + [InlineData("$share/g/a/b", "a/b")] + [InlineData("$share/g/#", "a/b")] + public void Match_partialLevelWildcardsAndSharedFilters_neverConfirm(string filter, string topic) + { + var trie = TopicFilterTrie.Build([(filter, 1)]); + var collect = new Collect(); + trie.Match(topic, ref collect); + + collect.Seen.Where(_ => MqttTopicFilterComparer.Compare(topic, filter) == MqttTopicFilterCompareResult.IsMatch) + .ShouldBeEmpty(); + } + [Fact] public void Empty_MatchesNothing() {