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

@@ -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:

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttBrowseSession.cs

@@ -149,6 +149,13 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
     /// </summary>
     private static readonly UTF8Encoding Utf8Strict = new(encoderShouldEmitUTF8Identifier: false, throwOnInvalidBytes: true);
 
+    /// <summary>
+    ///     Ceiling on the rebirth request's wait, below the AdminUI's 20 s <c>BrowserSessionService.PerCallTimeout</c>
+    ///     (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.
+    /// </summary>
+    internal static readonly TimeSpan MaxRebirthWait = TimeSpan.FromSeconds(15);
+
     /// <summary>The quiet spell that ends a rebirth request's wait once its births have started arriving.</summary>
     private static readonly TimeSpan RebirthQuietInterval = TimeSpan.FromMilliseconds(500);
 
@@ -225,6 +232,9 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
     /// <summary>The Sparkplug birth-observation window this session was opened with (R06:M-4).</summary>
     internal TimeSpan BirthWindow { get; }
 
+    /// <summary>How long a rebirth request waits for its births: the window, capped at <see cref="MaxRebirthWait"/>.</summary>
+    internal TimeSpan RebirthWait => BirthWindow < MaxRebirthWait ? BirthWindow : MaxRebirthWait;
+
     /// <summary>How many NBIRTH/DBIRTH messages this session has observed — the settle loop's signal.</summary>
     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)

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))
             {

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<string, ISubscriptionHandle>(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.",

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));
 
+    /// <summary>
+    ///     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).
+    /// </summary>
+    [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));
+    }
+
     /// <summary>An idle plant costs one quiet interval, not the whole window.</summary>
     [Fact]
     public async Task Settle_WithNoBirths_ReturnsAfterOneQuietInterval()

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverSessionLifecycleTests.cs

@@ -361,6 +361,79 @@ public sealed class MqttDriverSessionLifecycleTests
         driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull();
     }
 
+    /// <summary>
+    ///     Review minor 6: the bound itself. Ten topics against a broker that never PUBACKs: exactly
+    ///     <see cref="MqttDriver.MaxConcurrentPublishes"/> 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.
+    /// </summary>
+    [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);
+    }
+
+    /// <summary>
+    ///     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 <c>LastSuccessfulRead</c>.
+    ///     The positive control is <see cref="AWriteBatch_KeepsSameTopicWritesInRequestOrder"/>, where a QoS 1
+    ///     PUBACK does advance it.
+    /// </summary>
+    [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<bool> condition, string what)
     {
         var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10);

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerTests.cs

@@ -553,7 +553,7 @@ public sealed class MqttSubscriptionManagerTests
         var raised = new List<DataChangeEventArgs>();
         mgr.OnDataChange += (_, e) => raised.Add(e);
 
-        await Should.ThrowAsync<Exception>(() =>
+        await Should.ThrowAsync<InvalidOperationException>(() =>
             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);
     }
 
+    /// <summary>
+    ///     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
+    ///     (<see cref="SubscribeAsync_TransportThrows_DegradesRefsToBad_AndThrows"/>).
+    /// </summary>
+    [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<OperationCanceledException>(() =>
+            mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), cts.Token));
+        ex.ShouldNotBeOfType<InvalidOperationException>();
+        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]);
+    }
+
     /// <summary>
     ///     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.

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);
     }
 
+    /// <summary>
+    ///     Review minor 7: partial-level wildcards and <c>$share/…</c> 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.
+    /// </summary>
+    [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<int>.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()
     {