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.
This commit is contained in:
Joseph Doherty
@@ -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()
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user