dohertj2/lmxopcua
1
0
Fork 0
Code Issues 14 Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source 
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.
This commit is contained in:
Joseph Doherty
2026-08-10 08:03:47 -04:00
parent 94b8633e68
commit eddf5f57a5
2 changed files with 233 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttBrowseSession.cs
+79 -13
View File
@@ -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"/> /
/// <see cref="MaxSegmentChars"/>; surfaced as its own marker node from
/// True once a topic or metric was rejected as unusable — for breaching
/// <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 label)
foreach (var c in text)
{
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
+154
View File
@@ -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()
{