fix(mqtt-browser): converge the plain-mode label guard, make the node cap atomic, total the sibling sort

Driver.Mqtt.Browser-002/-003/-004/-005.

-002: ObservePlainTopic's pre-pass judged segment LENGTH only, so a plain topic
segment carrying a newline or an ANSI escape became the picker's displayed label —
a rule ObserveSparkplugBirth already applied via IsUsableLabel. The control-character
half is now single-sourced (HasControlCharacter) and the plain marker text converges
with the Sparkplug one. An EMPTY topic level stays accepted on purpose: it is legal
MQTT and its terminal leaf is a real committable address.

-003: GetOrCreate read _nodeCount and incremented it later, so N observers could each
see cap-1 and all insert. The slot is now reserved with one Interlocked.Increment and
given back on both non-consuming paths.

-004: Project ordered a level by Segment alone, but the level is keyed by Path
precisely because two Sparkplug siblings CAN share a Segment — leaving the tie to
ConcurrentDictionary enumeration order, which .NET's randomised string hashing makes
differ per process. .ThenBy(Path) makes the sort total.

-005: no behaviour change — DefaultNodeCap's doc now states the real per-session
ceiling and records that the aggregate across sessions is bounded by the reaper's
idle TTL, not by any session cap.

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

@@ -55,7 +55,27 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser;
 /// </summary>
 internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
 {
-    /// <summary>Default cap on recorded nodes; a busy broker's <c>#</c> firehose is unbounded.</summary>
+    /// <summary>
+    ///     Default cap on recorded nodes; a busy broker's <c>#</c> firehose is unbounded.
+    ///     <para>
+    ///         <b>Read this as a memory bound, not just a node count.</b> Each node retains its full
+    ///         <c>Path</c> (up to <see cref="MaxTopicChars"/>), its <c>Segment</c> (up to
+    ///         <see cref="MaxSegmentChars"/>) and, in plain mode, a snippet (up to
+    ///         <see cref="PayloadSnippetMaxChars"/>), plus its own child dictionary — and every node is
+    ///         held twice, once in its parent's children and once in the flat path index. So a session
+    ///         near these ceilings costs on the order of 100 MB, not the few MB "50 000 nodes" suggests,
+    ///         and a plain-mode open with no <c>TopicPrefix</c> subscribes <c>#</c> — the whole broker —
+    ///         which is exactly the shape that approaches them.
+    ///     </para>
+    ///     <para>
+    ///         <b>The aggregate across sessions is bounded by LIFETIME, not by a count.</b>
+    ///         <c>BrowseSessionRegistry</c> caps neither concurrent sessions nor sessions per user;
+    ///         <c>BrowseSessionReaper</c> evicts and disposes anything untouched for its 2-minute idle
+    ///         TTL on a 30-second tick, and <c>DisposeAsync</c> clears both maps. If a per-session bound
+    ///         is ever needed, cap the number of concurrent sessions there rather than lowering this
+    ///         constant, which would make the picker less useful on a genuinely large broker.
+    ///     </para>
+    /// </summary>
     internal const int DefaultNodeCap = 50_000;
 
     /// <summary>Maximum characters of a payload retained as the picker's confirm-it's-the-right-topic snippet.</summary>
@@ -198,8 +218,9 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
     internal bool Truncated => Volatile.Read(ref _truncated) != 0;
 
     /// <summary>
-    ///     True once a topic was rejected for breaching <see cref="MaxTopicChars"/> /
+    ///     True once a topic or metric was rejected as unusable — for breaching
-    ///     <see cref="MaxSegmentChars"/>; surfaced as its own marker node from
+    ///     <see cref="MaxTopicChars"/> / <see cref="MaxSegmentChars"/>, or for carrying a control
+    ///     character in a segment / identifier; surfaced as its own marker node from
     ///     <see cref="RootAsync"/>. Kept distinct from <see cref="Truncated"/> because the operator's
     ///     remedy differs — narrow the prefix, versus this broker publishes absurd topic names.
     /// </summary>
@@ -245,9 +266,11 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
         {
             nodes.Add(new BrowseNode(
                 OversizedNodeId,
+                // Both modes word it the same way: since the plain path rejects a control-character
+                // segment as well as an oversized one, "too long" would name only half the cause.
                 Mode == MqttMode.SparkplugB
                     ? "⚠ Some metrics were unusable (oversized or malformed names) — use manual entry for those"
-                    : "⚠ Some topics were too long to show — use manual entry for those",
+                    : "⚠ Some topics were unusable (oversized or malformed names) — use manual entry for those",
                 BrowseNodeKind.Folder,
                 HasChildrenHint: false));
         }
@@ -477,9 +500,21 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
         var segments = topic.Split('/');
 
         // Checked as a pre-pass so a rejected topic leaves no half-built path behind.
+        //
+        // The control-character half is the same rule IsUsableLabel applies to a Sparkplug identifier,
+        // and for the same reason: a segment becomes the picker's displayed label, MQTT forbids only
+        // U+0000 in a topic NAME, and an embedded newline or ANSI escape breaks a single-line row. It is
+        // a REJECTION rather than a scrub because a scrubbed segment is a different topic, and the
+        // picker would commit it as the tag's address.
+        //
+        // An EMPTY segment is deliberately NOT rejected, so this is not simply IsUsableLabel: `a//b` and
+        // a trailing '/' are legal MQTT, and such a topic's terminal leaf is a real committable address
+        // (the commit reads AddressFields[topic], which carries the node's full path). Refusing the whole
+        // topic would cost the operator a legal address to tidy a blank row. The Sparkplug path has no
+        // such case — every id it labels came off SparkplugTopic.TryParse, which rejects an empty segment.
         foreach (var segment in segments)
         {
-            if (segment.Length > MaxSegmentChars)
+            if (segment.Length > MaxSegmentChars || HasControlCharacter(segment))
             {
                 Volatile.Write(ref _oversized, 1);
                 return;
@@ -682,8 +717,14 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
         // different nodes, and keying by label would silently drop whichever arrived second.
         if (level.TryGetValue(path, out var existing)) return existing;
 
-        if (Volatile.Read(ref _nodeCount) >= _nodeCap)
+        // The slot is RESERVED before the node is created, in one interlocked step, and given back on
+        // either path that does not consume it. Reading the count and then incrementing it would let N
+        // observers each see `cap - 1` and all insert, overshooting the cap by up to N-1 — the cap is
+        // the only bound between a `#` firehose and unbounded memory, so it must not depend on MQTTnet
+        // happening to deliver on a single dispatcher thread.
+        if (Interlocked.Increment(ref _nodeCount) > _nodeCap)
         {
+            Interlocked.Decrement(ref _nodeCount);
             Volatile.Write(ref _truncated, 1);
             return null;
         }
@@ -692,9 +733,12 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
         var child = level.GetOrAdd(path, created);
         if (ReferenceEquals(child, created))
         {
-            Interlocked.Increment(ref _nodeCount);
             _byPath[path] = child;
         }
+        else
+        {
+            Interlocked.Decrement(ref _nodeCount);   // lost the insert race; the reservation was not consumed
+        }
 
         return child;
     }
@@ -709,15 +753,24 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
     /// </summary>
     /// <param name="label">The candidate label.</param>
     /// <returns><see langword="true"/> when the label may be recorded.</returns>
-    private static bool IsUsableLabel(string label)
+    private static bool IsUsableLabel(string label) =>
+        label.Length != 0 && label.Length <= MaxSegmentChars && !HasControlCharacter(label);
+
+    /// <summary>
+    ///     Whether <paramref name="text"/> carries a character that would break the picker's single-line
+    ///     rows. Single-sourced so the plain and Sparkplug ingest paths cannot drift on what may become a
+    ///     tree label — see <see cref="ObservePlainTopic"/> for the one rule they deliberately differ on.
+    /// </summary>
+    /// <param name="text">The candidate label.</param>
+    /// <returns><see langword="true"/> when a control character is present.</returns>
+    private static bool HasControlCharacter(string text)
     {
-        if (label.Length == 0 || label.Length > MaxSegmentChars) return false;
+        foreach (var c in text)
-        foreach (var c in label)
         {
-            if (char.IsControl(c)) return false;
+            if (char.IsControl(c)) return true;
         }
 
-        return true;
+        return false;
     }
 
     /// <summary>
@@ -1095,10 +1148,23 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
 
     /// <summary>Projects one level of the tree into browse nodes, ordered so the picker is stable.</summary>
     /// <param name="level">The child map to project.</param>
-    /// <returns>The projected nodes, sorted by segment.</returns>
+    /// <returns>The projected nodes, sorted by segment and then by path.</returns>
+    /// <remarks>
+    ///     <b>The sort must be TOTAL, which is why the path is the tiebreaker.</b> Sorting by segment
+    ///     alone is only a stable order when labels are unique, and the whole reason this map is keyed by
+    ///     path (see <see cref="GetOrCreate"/>) is that in Sparkplug mode two siblings CAN share one —
+    ///     a device folder <c>Plant1/EdgeA/Temp</c> and an edge-node-level metric
+    ///     <c>Plant1/EdgeA::Temp</c>. <see cref="Enumerable.OrderBy{TSource, TKey}(IEnumerable{TSource},
+    ///     Func{TSource, TKey}, IComparer{TKey})"/> is stable with respect to its input, but the input is
+    ///     <see cref="ConcurrentDictionary{TKey, TValue}.Values"/>, whose enumeration order is not
+    ///     guaranteed and can change as the tree grows — so a tie left to it lets two visually identical
+    ///     rows swap places between successive <see cref="ExpandAsync"/> calls, and the operator cannot
+    ///     see which one they are clicking.
+    /// </remarks>
     private static List<BrowseNode> Project(ConcurrentDictionary<string, TopicNode> level) =>
         level.Values
             .OrderBy(n => n.Segment, StringComparer.Ordinal)
+            .ThenBy(n => n.Path, StringComparer.Ordinal)
             .Select(n =>
             {
                 // Plain mode infers: a childless node is always the terminal segment of a topic that

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

@@ -296,6 +296,123 @@ public sealed class MqttBrowseSessionTests
         (await s.RootAsync(TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
     }
 
+    /// <summary>
+    ///     The two ingest paths must agree on what may become a tree label. MQTT forbids only
+    ///     <c>U+0000</c> in a topic NAME, so a publisher can legally put a newline or an ANSI escape in
+    ///     one — and a plain segment goes straight into the picker's displayed label, exactly where the
+    ///     Sparkplug path already rejects one.
+    /// </summary>
+    /// <summary>
+    ///     The node cap must hold under concurrent observation. MQTTnet delivers on a single dispatcher
+    ///     per client today, so this is a guard on the shape rather than a live defect — but the cap is
+    ///     the only thing standing between a <c>#</c> firehose and unbounded memory, and a read-then-
+    ///     increment pair lets N threads each see <c>cap - 1</c> and all insert.
+    /// </summary>
+    [Fact]
+    public async Task NodeCap_HoldsUnderConcurrentObservation()
+    {
+        // One round would pass by luck; the overshoot is a race. Each round drives many threads at a
+        // cap of 1 through a barrier, so every thread reaches the cap check at once.
+        const int rounds = 40;
+        var workers = Math.Max(Environment.ProcessorCount * 2, 8);
+
+        for (var round = 0; round < rounds; round++)
+        {
+            await using var s = new MqttBrowseSession(MqttMode.Plain, nodeCap: 1);
+            using var barrier = new Barrier(workers);
+
+            // Dedicated threads, not Task.Run: every worker BLOCKS on the barrier, and blocking that
+            // many pool threads makes the pool inject new ones on its ~500 ms starvation timer — which
+            // both slows the test enormously and staggers the very arrivals it is trying to align.
+            var threads = Enumerable.Range(0, workers)
+                .Select(i => new Thread(() =>
+                {
+                    barrier.SignalAndWait();
+                    s.ObserveTopicForTest($"t{i}", "1");
+                }) { IsBackground = true })
+                .ToList();
+
+            foreach (var t in threads) t.Start();
+            foreach (var t in threads) t.Join();
+
+            var recorded = (await s.RootAsync(TestContext.Current.CancellationToken))
+                .Count(n => n.NodeId != MqttBrowseSession.TruncatedNodeId
+                            && n.NodeId != MqttBrowseSession.OversizedNodeId);
+            recorded.ShouldBeLessThanOrEqualTo(1, $"round {round} overshot the node cap");
+        }
+    }
+
+    [Fact]
+    public async Task PlainTopicSegmentWithAControlCharacter_IsRejectedWhole_AndSurfacesAMarker()
+    {
+        // Every segment here is well under MaxSegmentChars, so ONLY the control-character rule can
+        // reject it — the length guard must not be what makes this test pass.
+        await using var s = new MqttBrowseSession(MqttMode.Plain);
+        s.ObserveTopicForTest("good/topic", "1");
+        s.ObserveTopicForTest("bad/oven\nname/temp", "1");
+
+        s.OversizedRejected.ShouldBeTrue();
+        var root = await s.RootAsync(TestContext.Current.CancellationToken);
+        root.ShouldContain(n => n.NodeId == MqttBrowseSession.OversizedNodeId);
+        root.ShouldContain(n => n.DisplayName == "good");
+        root.ShouldNotContain(n => n.DisplayName == "bad");   // no half-built path left behind
+    }
+
+    [Theory]
+    [InlineData("\n")]
+    [InlineData("\r")]
+    [InlineData("\t")]
+    [InlineData("\u001b")]   // ANSI escape
+    [InlineData("\u0000")]
+    public async Task PlainTopic_EachControlCharacterForm_IsRejected(string control)
+    {
+        await using var s = new MqttBrowseSession(MqttMode.Plain);
+        s.ObserveTopicForTest($"a/b{control}c/d", "1");
+
+        s.OversizedRejected.ShouldBeTrue();
+        (await s.RootAsync(TestContext.Current.CancellationToken))
+            .ShouldNotContain(n => n.DisplayName == "a");
+    }
+
+    /// <summary>
+    ///     Falsification control for the rule above: the guard rejects CONTROL characters, not
+    ///     "anything unusual". A topic segment is arbitrary UTF-8 and legitimately carries accents,
+    ///     spaces and punctuation — rejecting those would drop real plant topics from the picker.
+    /// </summary>
+    [Fact]
+    public async Task PlainTopic_PrintableNonAsciiSegments_AreStillAccepted()
+    {
+        await using var s = new MqttBrowseSession(MqttMode.Plain);
+        s.ObserveTopicForTest("usiné/four n°3/temp %", "1");
+
+        s.OversizedRejected.ShouldBeFalse();
+        (await s.RootAsync(TestContext.Current.CancellationToken))
+            .ShouldHaveSingleItem().DisplayName.ShouldBe("usiné");
+    }
+
+    /// <summary>
+    ///     An EMPTY topic level (<c>a//b</c>, or a trailing <c>/</c>) is legal MQTT and stays accepted —
+    ///     deliberately NOT converged with <c>IsUsableLabel</c>'s non-empty rule. Its terminal leaf is a
+    ///     real, committable address (the commit reads <c>AddressFields[topic]</c>, which carries the
+    ///     node's full path), so rejecting the topic whole would cost the operator a legal address to
+    ///     tidy a blank picker row. The control-character rule is the one that had to converge: a label
+    ///     carrying a newline breaks the row it renders in.
+    /// </summary>
+    [Fact]
+    public async Task PlainTopic_WithAnEmptyLevel_IsStillRecorded()
+    {
+        await using var s = new MqttBrowseSession(MqttMode.Plain);
+        s.ObserveTopicForTest("a//b", "1");
+
+        s.OversizedRejected.ShouldBeFalse();
+        var leaf = (await s.ExpandAsync("a/", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
+        leaf.NodeId.ShouldBe("a//b");
+
+        var attr = (await s.AttributesAsync("a//b", TestContext.Current.CancellationToken))
+            .ShouldHaveSingleItem();
+        attr.AddressFields.ShouldNotBeNull()[MqttTagConfigKeys.Topic].ShouldBe("a//b");
+    }
+
     // ---------------------------------------------------------------- payload bounds
 
     [Fact]
@@ -554,6 +671,43 @@ public sealed class MqttBrowseSessionTests
             .ShouldHaveSingleItem().Name.ShouldBe("Temp");
     }
 
+    /// <summary>
+    ///     Siblings that share a label must still have a TOTAL order. The level is projected from a
+    ///     <c>ConcurrentDictionary</c> keyed by path — precisely because two Sparkplug siblings CAN share
+    ///     a segment — and its enumeration order is not guaranteed, so ordering by segment alone leaves
+    ///     the tie broken by whatever the dictionary happens to yield. The two rows are visually
+    ///     identical, so a reorder between successive expands is a picker that changes what a click means
+    ///     with nothing on screen to show it.
+    /// </summary>
+    [Fact]
+    public async Task SparkplugTree_SiblingsSharingALabel_AreOrderedDeterministically()
+    {
+        // MANY colliding pairs, not one. .NET string hashing is randomised per process, so the tie order
+        // a segment-only sort falls through to is whatever THIS run's dictionary yields — with a single
+        // pair it lands the right way round about half the time and the test would pass vacuously. With
+        // 24 pairs, every tie has to agree by accident for that to happen.
+        var names = Enumerable.Range(0, 24).Select(i => $"N{i:D2}").ToList();
+
+        await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
+        foreach (var name in names)
+        {
+            s.ObserveBirthForTest("Plant1", "EdgeA", null, name, SparkplugDataType.Float);    // metric
+            s.ObserveBirthForTest("Plant1", "EdgeA", name, "Value", SparkplugDataType.Float); // device
+        }
+
+        // Segment first (the names, in ordinal order), then path — and Ordinal puts '/' (0x2F) before
+        // ':' (0x3A), so each device folder precedes the same-named node-level metric.
+        var expected = names.SelectMany(n => new[] { $"Plant1/EdgeA/{n}", $"Plant1/EdgeA::{n}" }).ToList();
+
+        var children = await s.ExpandAsync("Plant1/EdgeA", TestContext.Current.CancellationToken);
+        children.Select(n => n.NodeId).ShouldBe(expected);
+
+        // …and it is stable: expanding the same level again yields the same order.
+        (await s.ExpandAsync("Plant1/EdgeA", TestContext.Current.CancellationToken))
+            .Select(n => n.NodeId)
+            .ShouldBe(expected);
+    }
+
     [Fact]
     public async Task SparkplugTree_EmptyBirth_StillRendersItsDeviceAsAFolder()
     {