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 1c563e83..432d539c 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 @@ -55,7 +55,27 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser; /// internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession { - /// Default cap on recorded nodes; a busy broker's # firehose is unbounded. + /// + /// Default cap on recorded nodes; a busy broker's # firehose is unbounded. + /// + /// Read this as a memory bound, not just a node count. Each node retains its full + /// Path (up to ), its Segment (up to + /// ) and, in plain mode, a snippet (up to + /// ), 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 TopicPrefix subscribes # — the whole broker — + /// which is exactly the shape that approaches them. + /// + /// + /// The aggregate across sessions is bounded by LIFETIME, not by a count. + /// BrowseSessionRegistry caps neither concurrent sessions nor sessions per user; + /// BrowseSessionReaper evicts and disposes anything untouched for its 2-minute idle + /// TTL on a 30-second tick, and DisposeAsync 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. + /// + /// internal const int DefaultNodeCap = 50_000; /// Maximum characters of a payload retained as the picker's confirm-it's-the-right-topic snippet. @@ -198,8 +218,9 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession internal bool Truncated => Volatile.Read(ref _truncated) != 0; /// - /// True once a topic was rejected for breaching / - /// ; surfaced as its own marker node from + /// True once a topic or metric was rejected as unusable — for breaching + /// / , or for carrying a control + /// character in a segment / identifier; surfaced as its own marker node from /// . Kept distinct from because the operator's /// remedy differs — narrow the prefix, versus this broker publishes absurd topic names. /// @@ -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 /// /// The candidate label. /// when the label may be recorded. - private static bool IsUsableLabel(string label) + private static bool IsUsableLabel(string label) => + label.Length != 0 && label.Length <= MaxSegmentChars && !HasControlCharacter(label); + + /// + /// Whether 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 for the one rule they deliberately differ on. + /// + /// The candidate label. + /// when a control character is present. + private static bool HasControlCharacter(string text) { - if (label.Length == 0 || label.Length > MaxSegmentChars) return false; - foreach (var c in label) + foreach (var c in text) { - if (char.IsControl(c)) return false; + if (char.IsControl(c)) return true; } - return true; + return false; } /// @@ -1095,10 +1148,23 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession /// Projects one level of the tree into browse nodes, ordered so the picker is stable. /// The child map to project. - /// The projected nodes, sorted by segment. + /// The projected nodes, sorted by segment and then by path. + /// + /// The sort must be TOTAL, which is why the path is the tiebreaker. Sorting by segment + /// alone is only a stable order when labels are unique, and the whole reason this map is keyed by + /// path (see ) is that in Sparkplug mode two siblings CAN share one — + /// a device folder Plant1/EdgeA/Temp and an edge-node-level metric + /// Plant1/EdgeA::Temp. is stable with respect to its input, but the input is + /// , 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 calls, and the operator cannot + /// see which one they are clicking. + /// private static List Project(ConcurrentDictionary 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 diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBrowseSessionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBrowseSessionTests.cs index b84fdb7d..bbc14bc1 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBrowseSessionTests.cs +++ b/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(); } + /// + /// The two ingest paths must agree on what may become a tree label. MQTT forbids only + /// U+0000 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. + /// + /// + /// 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 # firehose and unbounded memory, and a read-then- + /// increment pair lets N threads each see cap - 1 and all insert. + /// + [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"); + } + + /// + /// 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. + /// + [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é"); + } + + /// + /// An EMPTY topic level (a//b, or a trailing /) is legal MQTT and stays accepted — + /// deliberately NOT converged with IsUsableLabel's non-empty rule. Its terminal leaf is a + /// real, committable address (the commit reads AddressFields[topic], 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. + /// + [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"); } + /// + /// Siblings that share a label must still have a TOTAL order. The level is projected from a + /// ConcurrentDictionary 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. + /// + [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() {