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

fix(mqtt): wildcard tags went dark on reconnect — index by filter, not topic

Browse Source 
C1 (CRITICAL). AuthoredTable routed any wildcard-authored tag into `Wildcards`
and never into the concrete topic index. But the two paths that reconstruct
state from a SUBSCRIBED FILTER STRING — OnReconnectedAsync's re-subscribe set
and DegradeTopic — both looked up that concrete-only index, and
`_subscribedTopics` keys on the wildcard PATTERN ("f/+/temp"). So:

- reconnect resolved zero filters for a wildcard tag, hit a silent early
  return, and issued NO subscribe. The cache keeps its last Good value, so the
  tag never turns Bad — it just stops updating forever behind a connection
  reporting healthy. Exactly the silent-death mode MqttConnection's own remarks
  warn about, left unguarded for wildcards.
- DegradeTopic missed too, so a SUBACK rejection of a wildcard filter degraded
  nothing and left the tag at BadWaitingForInitialData.

Fix: AuthoredTable now carries `ByFilter` (EVERY def, keyed by its authored
topic filter, wildcards included) alongside `ByExactTopic` (concrete only) and
`Wildcards`. Delivery matches an incoming published topic — which never carries
a wildcard — so it keeps using ByExactTopic + the comparer scan. Every path
keyed on a filter string uses ByFilter. The distinction is documented on the
record. The silent early return is now a Warning naming the orphaned topics.

Also in this pass:

- I2: bounded decode on the dispatcher thread. A body over `MaxPayloadBytes`
  (default 1 MiB, ctor-settable) is refused BEFORE any GetString/JsonDocument
  parse and degrades its own tags. Unbounded decode on the shared dispatcher is
  paid by every subscription, not just the offending topic. NOTE: promoting this
  to an operator-facing MqttDriverOptions key (+ WithMaximumPacketSize) needs a
  .Contracts edit this task is scoped out of; flagged in the XML docs.
- I3: integral-valued reals now coerce to integer tags. JS/Python edge gateways
  serialize an integer as 5.0, and TryGetInt32/int.TryParse are syntactic — an
  Int32 tag fed by such a gateway silently never received data. Parsed via
  decimal so integrality and range are exact across Int64/UInt64. A genuinely
  fractional 5.5 is still refused, never rounded.
- I4: the JSON document is parsed ONCE per message and shared across the whole
  fan-out (the documented "one document, one JSONPath per signal" shape was
  re-parsing per tag). Zero-alloc via a ref struct when no Json tag matches.
- I5: pinned the documented JSON-string-holding-a-number coercion end-to-end.
- Minor: Register now prunes `_subscribedTopics` / `_handleByRawPath` entries for
  tags a redeploy dropped, so a deleted topic stops being re-subscribed forever.

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
2026-07-24 16:40:31 -04:00
parent 4ec01bdfc1
commit abf3116bd4
2 changed files with 622 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs
+378 -89
View File
@@ -59,6 +59,12 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
/// </remarks>
public sealed class MqttSubscriptionManager
{
/// <summary>
/// Default <see cref="MaxPayloadBytes"/> — 1 MiB. Generous for MQTT telemetry (which is
/// kilobytes) while still bounding the dispatcher-thread work any one publisher can impose.
/// </summary>
public const int DefaultMaxPayloadBytes = 1024 * 1024;
// OPC UA status codes (values verified against Opc.Ua.StatusCodes, not recalled).
private const uint StatusGood = 0x00000000u;
@@ -113,16 +119,23 @@ public sealed class MqttSubscriptionManager
/// The last-value store backing <c>IReadable</c>. Defaults to a fresh one owned by this
/// manager and exposed as <see cref="Values"/>.
/// </param>
/// <param name="maxPayloadBytes">
/// Ceiling on an inbound message body; see <see cref="MaxPayloadBytes"/>. Non-positive values
/// fall back to <see cref="DefaultMaxPayloadBytes"/> — "unbounded" is not offered, because the
/// bound exists to protect a shared dispatcher thread.
/// </param>
public MqttSubscriptionManager(
MqttDriverOptions options,
string driverId,
IMqttSubscribeTransport? transport = null,
ILogger? logger = null,
LastValueCache? cache = null)
LastValueCache? cache = null,
int maxPayloadBytes = DefaultMaxPayloadBytes)
{
ArgumentNullException.ThrowIfNull(options);
ArgumentException.ThrowIfNullOrWhiteSpace(driverId);
MaxPayloadBytes = maxPayloadBytes > 0 ? maxPayloadBytes : DefaultMaxPayloadBytes;
_mode = options.Mode;
_defaultQos = options.Plain?.DefaultQos ?? 1;
_driverId = driverId;
@@ -142,6 +155,27 @@ public sealed class MqttSubscriptionManager
/// </summary>
public LastValueCache Values { get; }
/// <summary>
/// Ceiling on an inbound message body. A larger message is refused before any decode or parse
/// and degrades its own tags to <c>BadDecodingError</c>.
/// </summary>
/// <remarks>
/// Decode and parse both run synchronously on MQTTnet's shared dispatcher thread, so their cost
/// is paid by <i>every</i> subscription, not just the offending topic. Without a bound, one
/// publisher shipping a multi-megabyte body — or a hostile one shipping a huge nested document
/// — stalls the whole driver's delivery for as long as the parse takes. Sized for telemetry
/// payloads, which are kilobytes; a deployment that legitimately needs more should raise this
/// deliberately rather than run unbounded.
/// <para>
/// <b>Follow-up:</b> this is a constructor knob, not yet an operator-facing one. Promoting
/// it to an <c>MqttDriverOptions</c> key (and pairing it with MQTTnet's
/// <c>WithMaximumPacketSize</c>, so an oversized body is refused at the protocol layer
/// rather than after the client has already buffered it) needs an edit to the
/// <c>.Contracts</c> project that this task was scoped out of.
/// </para>
/// </remarks>
public int MaxPayloadBytes { get; }
/// <summary>
/// Wires this manager onto a live connection: inbound messages route to
/// <see cref="HandleMessage"/> and every reconnect re-establishes the subscriptions through
@@ -197,8 +231,29 @@ public sealed class MqttSubscriptionManager
}
}
_authored = AuthoredTable.Build(byRawPath);
var table = AuthoredTable.Build(byRawPath);
_authored = table;
_warnedRefs.Clear();
// Drop state belonging to tags this deploy removed. Without this, a topic whose last tag was
// deleted stays in _subscribedTopics and is dutifully re-SUBSCRIBEd on every future reconnect,
// forever — and its RawPaths keep a handle mapping that can never fire again.
foreach (var topic in _subscribedTopics.Keys)
{
if (!table.ByFilter.ContainsKey(topic))
{
_subscribedTopics.TryRemove(topic, out _);
}
}
foreach (var rawPath in _handleByRawPath.Keys)
{
if (!table.ByRawPath.ContainsKey(rawPath))
{
_handleByRawPath.TryRemove(rawPath, out _);
}
}
return byRawPath.Count;
}
@@ -334,14 +389,38 @@ public sealed class MqttSubscriptionManager
public async Task OnReconnectedAsync(CancellationToken cancellationToken)
{
var authored = _authored;
var live = _subscribedTopics.Keys
.Select(topic => authored.ByTopic.TryGetValue(topic, out var defs) ? defs : [])
var subscribed = _subscribedTopics.Keys.ToList();
// ByFilter, NOT ByExactTopic: these keys are subscribed FILTER strings, so a wildcard-authored
// tag's key is its pattern ("f/+/temp"). Reading the concrete-only index here misses every
// wildcard tag, drops it out of the re-subscribe set, and leaves it dark forever behind a
// connection that reports healthy — see the AuthoredTable remarks.
var live = subscribed
.Select(topic => authored.ByFilter.TryGetValue(topic, out var defs) ? defs : [])
.SelectMany(defs => defs)
.ToList();
var filters = BuildFilters(live, _defaultQos);
if (filters.Count == 0)
{
// Silence here is how the wildcard defect hid: no exception, no log, no SUBSCRIBE. If any
// topic was live, resolving zero filters for it is an anomaly and must be loud.
if (subscribed.Count > 0)
{
_logger?.LogWarning(
"MQTT driver '{DriverId}': reconnect resolved NO filters for {TopicCount} live topic(s) "
+ "({Topics}); those subscriptions are not being re-established.",
_driverId,
subscribed.Count,
string.Join(", ", subscribed));
}
else
{
_logger?.LogDebug(
"MQTT driver '{DriverId}': reconnect had no live subscriptions to re-establish.",
_driverId);
}
return;
}
@@ -361,9 +440,23 @@ public sealed class MqttSubscriptionManager
/// currently subscribed.
/// </summary>
/// <remarks>
/// Runs on MQTTnet's dispatcher thread: no I/O, no locks, no allocation beyond the extracted
/// value, and <b>no path throws</b>. A tag whose payload is malformed degrades to a Bad status
/// code and the fan-out continues to the next tag.
/// <para>
/// Runs on MQTTnet's dispatcher thread: no I/O, no locks, and <b>no path throws</b>. A tag
/// whose payload is malformed degrades to a Bad status code and the fan-out continues to
/// the next tag.
/// </para>
/// <para>
/// <b>Bounded work.</b> A body larger than <see cref="MaxPayloadBytes"/> is refused
/// <i>before</i> any decode or parse: the whole point of the dispatcher discipline is that
/// one topic cannot stall delivery for every other subscription, and an unbounded
/// <c>GetString</c> / <c>JsonDocument.Parse</c> over a multi-megabyte body does exactly
/// that. The oversized message degrades its own tags to <c>BadDecodingError</c>.
/// </para>
/// <para>
/// <b>Parsed once.</b> The documented shape is several tags sharing one JSON document with
/// a JSONPath each, so the document is parsed at most once per message and shared across
/// the whole fan-out rather than re-parsed per tag.
/// </para>
/// </remarks>
/// <param name="topic">The concrete topic the message arrived on.</param>
/// <param name="payload">The message body; valid only for the duration of this call.</param>
@@ -377,25 +470,35 @@ public sealed class MqttSubscriptionManager
var authored = _authored;
var now = DateTime.UtcNow;
var oversized = payload.Length > MaxPayloadBytes;
// Concrete topics — the overwhelmingly common case — cost one lookup, so a chatty broker
// publishing thousands of topics this driver does not care about stays cheap.
if (authored.ByTopic.TryGetValue(topic, out var exact))
// Zero-alloc until a Json-format tag actually matches; disposed in the finally below.
var json = default(PayloadJson);
try
{
foreach (var def in exact)
// Concrete topics — the overwhelmingly common case — cost one lookup, so a chatty broker
// publishing thousands of topics this driver does not care about stays cheap.
if (authored.ByExactTopic.TryGetValue(topic, out var exact))
{
Deliver(def, payload, retained, now);
foreach (var def in exact)
{
Deliver(def, payload, retained, now, oversized, ref json);
}
}
// A published topic name may not itself contain a wildcard, so a wildcard-authored
// definition can never also be an exact-index hit: no tag is delivered twice.
foreach (var def in authored.Wildcards)
{
if (MqttTopicFilterComparer.Compare(topic, def.Topic) == MqttTopicFilterCompareResult.IsMatch)
{
Deliver(def, payload, retained, now, oversized, ref json);
}
}
}
// A published topic name may not itself contain a wildcard, so a wildcard-authored definition
// can never also be an exact-index hit: no tag is delivered twice.
foreach (var def in authored.Wildcards)
finally
{
if (MqttTopicFilterComparer.Compare(topic, def.Topic) == MqttTopicFilterCompareResult.IsMatch)
{
Deliver(def, payload, retained, now);
}
json.Dispose();
}
}
@@ -509,14 +612,27 @@ public sealed class MqttSubscriptionManager
}
/// <summary>Marks every authored reference on <paramref name="topic"/> as unreachable.</summary>
/// <remarks>
/// <paramref name="topic"/> is a subscribed <i>filter</i> string (it came from a requested
/// filter or a SUBACK outcome), so the lookup must go through <c>ByFilter</c> — the
/// concrete-only index misses every wildcard-authored tag and this method would then return
/// having silently degraded nothing, leaving the tag at <c>BadWaitingForInitialData</c> (or
/// worse, its last Good value) after the broker refused its subscription.
/// </remarks>
/// <param name="topic">The topic filter whose subscription could not be established.</param>
/// <param name="reason">The failure reason, for logs.</param>
private void DegradeTopic(string topic, string reason)
{
_subscribedTopics.TryRemove(topic, out _);
if (!_authored.ByTopic.TryGetValue(topic, out var defs))
if (!_authored.ByFilter.TryGetValue(topic, out var defs))
{
_logger?.LogWarning(
"MQTT driver '{DriverId}': cannot degrade tags for unestablished filter '{Topic}' ({Reason}) — "
+ "no authored tag resolves to it.",
_driverId,
topic,
reason);
return;
}
@@ -544,7 +660,15 @@ public sealed class MqttSubscriptionManager
/// <param name="payload">The message body.</param>
/// <param name="retained">The message's retain flag.</param>
/// <param name="nowUtc">One clock read shared by the whole fan-out, so siblings agree.</param>
private void Deliver(MqttTagDefinition def, ReadOnlySpan<byte> payload, bool retained, DateTime nowUtc)
/// <param name="oversized">Whether the body exceeded <see cref="MaxPayloadBytes"/>.</param>
/// <param name="json">The fan-out's shared JSON parse, created on first use.</param>
private void Deliver(
MqttTagDefinition def,
ReadOnlySpan<byte> payload,
bool retained,
DateTime nowUtc,
bool oversized,
ref PayloadJson json)
{
// MQTT 3.1.1 has no retain-handling option, so the broker always replays its retained message
// on subscribe; a tag that opted out of seeding is filtered here rather than at the broker.
@@ -556,7 +680,11 @@ public sealed class MqttSubscriptionManager
DataValueSnapshot snapshot;
try
{
snapshot = Extract(def, payload, nowUtc);
// The cap is enforced before any decode/parse, so an oversized body costs one comparison
// per matched tag rather than a full pass over the payload.
snapshot = oversized
? new DataValueSnapshot(null, StatusBadDecodingError, null, nowUtc)
: Extract(def, payload, nowUtc, ref json);
}
catch (Exception ex)
{
@@ -623,13 +751,18 @@ public sealed class MqttSubscriptionManager
/// <param name="def">The authored tag definition.</param>
/// <param name="payload">The message body.</param>
/// <param name="nowUtc">Timestamp for the snapshot — MQTT carries no source timestamp of its own.</param>
/// <param name="json">The fan-out's shared JSON parse, created on first use by a Json-format tag.</param>
/// <returns>The extracted snapshot.</returns>
internal static DataValueSnapshot Extract(MqttTagDefinition def, ReadOnlySpan<byte> payload, DateTime nowUtc)
internal static DataValueSnapshot Extract(
MqttTagDefinition def,
ReadOnlySpan<byte> payload,
DateTime nowUtc,
ref PayloadJson json)
{
switch (def.PayloadFormat)
{
case MqttPayloadFormat.Json:
return ExtractJson(def, payload, nowUtc);
return ExtractJson(def, payload, nowUtc, ref json);
case MqttPayloadFormat.Raw:
// "Bytes as-is": the payload's UTF-8 text, published verbatim. dataType is deliberately
@@ -657,38 +790,81 @@ public sealed class MqttSubscriptionManager
private static DataValueSnapshot Bad(uint status, DateTime nowUtc) => new(null, status, null, nowUtc);
/// <summary>Parses the payload as JSON, selects the tag's JSONPath, and coerces to its data type.</summary>
/// <summary>
/// Selects the tag's JSONPath out of the fan-out's <b>shared</b> parse and coerces it to the
/// tag's data type. The document is parsed at most once per message however many tags share the
/// topic — the documented "one JSON document, one JSONPath per signal" shape would otherwise
/// re-parse it once per tag, on the dispatcher thread.
/// </summary>
/// <param name="def">The authored tag definition.</param>
/// <param name="payload">The message body.</param>
/// <param name="nowUtc">Timestamp for the snapshot.</param>
/// <param name="json">The shared parse; parses on first call and caches the outcome.</param>
/// <returns>The extracted snapshot.</returns>
private static DataValueSnapshot ExtractJson(MqttTagDefinition def, ReadOnlySpan<byte> payload, DateTime nowUtc)
private static DataValueSnapshot ExtractJson(
MqttTagDefinition def,
ReadOnlySpan<byte> payload,
DateTime nowUtc,
ref PayloadJson json)
{
JsonDocument? doc = null;
try
{
var reader = new Utf8JsonReader(payload);
if (!JsonDocument.TryParseValue(ref reader, out doc))
{
return Bad(StatusBadDecodingError, nowUtc);
}
if (!TryEvaluateJsonPath(doc.RootElement, def.JsonPath, out var selected))
{
return Bad(StatusBadDecodingError, nowUtc);
}
return TryCoerce(selected, def.DataType, out var value)
? new DataValueSnapshot(value, StatusGood, nowUtc, nowUtc)
: Bad(StatusBadTypeMismatch, nowUtc);
}
catch (JsonException)
if (!json.TryGetRoot(payload, out var root))
{
return Bad(StatusBadDecodingError, nowUtc);
}
finally
if (!TryEvaluateJsonPath(root, def.JsonPath, out var selected))
{
doc?.Dispose();
return Bad(StatusBadDecodingError, nowUtc);
}
return TryCoerce(selected, def.DataType, out var value)
? new DataValueSnapshot(value, StatusGood, nowUtc, nowUtc)
: Bad(StatusBadTypeMismatch, nowUtc);
}
/// <summary>
/// One message's JSON parse, shared across the whole fan-out. A mutable struct passed by
/// <c>ref</c> so the common case — a message no Json-format tag matched — allocates nothing at
/// all, while a message several Json tags share is parsed exactly once.
/// </summary>
internal struct PayloadJson : IDisposable
{
private JsonDocument? _doc;
private bool _attempted;
private bool _parsed;
/// <summary>
/// The parsed document root, parsing on first call. A malformed payload is remembered as a
/// failure so the fan-out's remaining tags do not each retry the same doomed parse.
/// </summary>
/// <param name="payload">The message body; only read on the first call.</param>
/// <param name="root">The document root when this returns <see langword="true"/>.</param>
/// <returns><see langword="true"/> when the payload is valid JSON.</returns>
public bool TryGetRoot(ReadOnlySpan<byte> payload, out JsonElement root)
{
if (!_attempted)
{
_attempted = true;
try
{
var reader = new Utf8JsonReader(payload);
_parsed = JsonDocument.TryParseValue(ref reader, out _doc);
}
catch (JsonException)
{
_parsed = false;
}
}
root = _parsed ? _doc!.RootElement : default;
return _parsed;
}
/// <inheritdoc/>
public void Dispose()
{
_doc?.Dispose();
_doc = null;
}
}
@@ -830,28 +1006,28 @@ public sealed class MqttSubscriptionManager
return false;
case DriverDataType.Int16:
if (element.ValueKind == JsonValueKind.Number && element.TryGetInt16(out var i16)) { value = i16; return true; }
return false;
case DriverDataType.Int32:
if (element.ValueKind == JsonValueKind.Number && element.TryGetInt32(out var i32)) { value = i32; return true; }
return false;
case DriverDataType.Int64:
if (element.ValueKind == JsonValueKind.Number && element.TryGetInt64(out var i64)) { value = i64; return true; }
return false;
case DriverDataType.UInt16:
if (element.ValueKind == JsonValueKind.Number && element.TryGetUInt16(out var u16)) { value = u16; return true; }
return false;
case DriverDataType.UInt32:
if (element.ValueKind == JsonValueKind.Number && element.TryGetUInt32(out var u32)) { value = u32; return true; }
return false;
case DriverDataType.UInt64:
if (element.ValueKind == JsonValueKind.Number && element.TryGetUInt64(out var u64)) { value = u64; return true; }
return false;
if (element.ValueKind != JsonValueKind.Number)
{
return false;
}
// Fast path: the JSON number is already written as an integer.
if (TryExactInteger(element, type, out value))
{
return true;
}
// Fallback: an integral-valued REAL. JS/Python edge gateways routinely serialize an
// integer as 5.0 (or 5E2), and System.Text.Json's TryGetInt32/64 are syntactic — they
// refuse it. Refusing here would mean an integer tag fed by such a gateway silently
// never receives data, so an exactly-integral real is accepted; a genuinely fractional
// value (5.5) is still refused rather than rounded.
return TryIntegralReal(element.GetRawText(), type, out value);
case DriverDataType.Float32:
if (element.ValueKind == JsonValueKind.Number && element.TryGetSingle(out var f32)) { value = f32; return true; }
@@ -896,28 +1072,19 @@ public sealed class MqttSubscriptionManager
return false;
case DriverDataType.Int16:
if (short.TryParse(trimmed, NumberStyles.Integer, inv, out var i16)) { value = i16; return true; }
return false;
case DriverDataType.Int32:
if (int.TryParse(trimmed, NumberStyles.Integer, inv, out var i32)) { value = i32; return true; }
return false;
case DriverDataType.Int64:
if (long.TryParse(trimmed, NumberStyles.Integer, inv, out var i64)) { value = i64; return true; }
return false;
case DriverDataType.UInt16:
if (ushort.TryParse(trimmed, NumberStyles.Integer, inv, out var u16)) { value = u16; return true; }
return false;
case DriverDataType.UInt32:
if (uint.TryParse(trimmed, NumberStyles.Integer, inv, out var u32)) { value = u32; return true; }
return false;
case DriverDataType.UInt64:
if (ulong.TryParse(trimmed, NumberStyles.Integer, inv, out var u64)) { value = u64; return true; }
return false;
// Same two-step as the JSON path: exact integer text first, then an integral-valued
// real ("5.0"), which a Scalar-format edge gateway emits just as readily.
if (TryExactIntegerText(trimmed, type, out value))
{
return true;
}
return TryIntegralReal(trimmed, type, out value);
case DriverDataType.Float32:
if (float.TryParse(trimmed, NumberStyles.Float, inv, out var f32)) { value = f32; return true; }
@@ -945,6 +1112,95 @@ public sealed class MqttSubscriptionManager
}
}
/// <summary>Reads a JSON number that is written as an exact integer of the requested type.</summary>
/// <param name="element">The JSON number.</param>
/// <param name="type">The target integer type.</param>
/// <param name="value">The boxed value when this returns <see langword="true"/>.</param>
/// <returns><see langword="true"/> when the number is an exact integer in range.</returns>
private static bool TryExactInteger(JsonElement element, DriverDataType type, out object? value)
{
value = type switch
{
DriverDataType.Int16 => element.TryGetInt16(out var v) ? v : null,
DriverDataType.Int32 => element.TryGetInt32(out var v) ? v : null,
DriverDataType.Int64 => element.TryGetInt64(out var v) ? v : null,
DriverDataType.UInt16 => element.TryGetUInt16(out var v) ? v : null,
DriverDataType.UInt32 => element.TryGetUInt32(out var v) ? v : null,
DriverDataType.UInt64 => element.TryGetUInt64(out var v) ? v : (object?)null,
_ => null,
};
return value is not null;
}
/// <summary>Parses text written as an exact integer of the requested type.</summary>
/// <param name="text">The trimmed text.</param>
/// <param name="type">The target integer type.</param>
/// <param name="value">The boxed value when this returns <see langword="true"/>.</param>
/// <returns><see langword="true"/> when the text is an exact integer in range.</returns>
private static bool TryExactIntegerText(string text, DriverDataType type, out object? value)
{
var inv = CultureInfo.InvariantCulture;
value = type switch
{
DriverDataType.Int16 => short.TryParse(text, NumberStyles.Integer, inv, out var v) ? v : null,
DriverDataType.Int32 => int.TryParse(text, NumberStyles.Integer, inv, out var v) ? v : null,
DriverDataType.Int64 => long.TryParse(text, NumberStyles.Integer, inv, out var v) ? v : null,
DriverDataType.UInt16 => ushort.TryParse(text, NumberStyles.Integer, inv, out var v) ? v : null,
DriverDataType.UInt32 => uint.TryParse(text, NumberStyles.Integer, inv, out var v) ? v : null,
DriverDataType.UInt64 => ulong.TryParse(text, NumberStyles.Integer, inv, out var v) ? v : (object?)null,
_ => null,
};
return value is not null;
}
/// <summary>
/// Accepts a real that is <b>exactly</b> integral (<c>5.0</c>, <c>5E2</c>) as an integer of the
/// requested type. A fractional value is refused, never rounded — silently turning 5.5 into 5
/// or 6 would be a wrong value, which is worse than no value.
/// </summary>
/// <remarks>
/// Parsed as <see cref="decimal"/> rather than <see cref="double"/> so the integrality test and
/// the conversion are exact across the whole Int64/UInt64 range, where double loses precision
/// above 2^53. Out-of-range values overflow the conversion and are refused.
/// </remarks>
/// <param name="text">The number's text (JSON raw text, or a Scalar payload).</param>
/// <param name="type">The target integer type.</param>
/// <param name="value">The boxed value when this returns <see langword="true"/>.</param>
/// <returns><see langword="true"/> when the text is an integral real in range.</returns>
private static bool TryIntegralReal(string text, DriverDataType type, out object? value)
{
value = null;
if (!decimal.TryParse(text, NumberStyles.Float, CultureInfo.InvariantCulture, out var d)
|| d != decimal.Truncate(d))
{
return false;
}
try
{
value = type switch
{
DriverDataType.Int16 => (short)d,
DriverDataType.Int32 => (int)d,
DriverDataType.Int64 => (long)d,
DriverDataType.UInt16 => (ushort)d,
DriverDataType.UInt32 => (uint)d,
DriverDataType.UInt64 => (ulong)d,
_ => null,
};
}
catch (OverflowException)
{
// decimal → integer conversions throw rather than wrap; out of range is a refusal.
value = null;
}
return value is not null;
}
/// <summary>
/// Decodes a payload as UTF-8, refusing invalid byte sequences rather than rendering a genuinely
/// binary body as a field of replacement characters (the same choice the browse session makes).
@@ -990,32 +1246,64 @@ public sealed class MqttSubscriptionManager
private sealed record MqttSubscriptionHandle(string DiagnosticId) : ISubscriptionHandle;
/// <summary>
/// The authored tag table and its derived topic index, published as one immutable snapshot so
/// the dispatcher thread reads a consistent view while a redeploy rebuilds it.
/// The authored tag table and its derived indexes, published as one immutable snapshot so the
/// dispatcher thread reads a consistent view while a redeploy rebuilds it.
/// </summary>
/// <remarks>
/// <b>Two topic indexes, and the difference is load-bearing.</b> Delivery matches an
/// <i>incoming published topic</i>, which never carries a wildcard, so it uses
/// <see cref="ByExactTopic"/> (concrete only) plus a comparer scan of <see cref="Wildcards"/>.
/// Every path that instead reconstructs state from a <i>subscribed filter string</i> —
/// reconnect re-subscribe and per-topic degradation — must use <see cref="ByFilter"/>, which is
/// keyed by the tag's authored topic <b>including wildcard patterns</b>, because that is what
/// lands in <c>_subscribedTopics</c> and in a SUBACK outcome. Looking a wildcard filter up in
/// the concrete index always misses, and a miss on those paths is silent: the tag simply stops
/// updating behind a healthy-looking connection, keeping its last Good value forever.
/// </remarks>
/// <param name="ByRawPath">RawPath → definition; the v3 resolver's backing table.</param>
/// <param name="ByTopic">Concrete topic → every definition bound to it (the fan-out list).</param>
/// <param name="ByFilter">
/// Authored topic <i>filter</i> string → every definition bound to it, wildcard patterns
/// included. Keyed the same way <c>_subscribedTopics</c> and SUBACK outcomes are.
/// </param>
/// <param name="ByExactTopic">Concrete topic → its definitions; the delivery fast path.</param>
/// <param name="Wildcards">Definitions whose authored topic carries an MQTT wildcard.</param>
private sealed record AuthoredTable(
FrozenDictionary<string, MqttTagDefinition> ByRawPath,
FrozenDictionary<string, MqttTagDefinition[]> ByTopic,
FrozenDictionary<string, MqttTagDefinition[]> ByFilter,
FrozenDictionary<string, MqttTagDefinition[]> ByExactTopic,
MqttTagDefinition[] Wildcards)
{
public static readonly AuthoredTable Empty =
Build(new Dictionary<string, MqttTagDefinition>(StringComparer.Ordinal));
/// <summary>Builds the snapshot, splitting concrete topics from wildcard-authored ones.</summary>
/// <summary>True when a topic string carries an MQTT wildcard and is therefore a filter, not a name.</summary>
/// <param name="topic">The authored topic.</param>
/// <returns><see langword="true"/> when it contains <c>+</c> or <c>#</c>.</returns>
public static bool IsWildcard(string topic) =>
topic.Contains(MqttTopicFilterComparer.SingleLevelWildcard, StringComparison.Ordinal)
|| topic.Contains(MqttTopicFilterComparer.MultiLevelWildcard, StringComparison.Ordinal);
/// <summary>Builds the snapshot: every def into <c>ByFilter</c>, concrete/wildcard split for delivery.</summary>
/// <param name="byRawPath">The mapped definitions, keyed by RawPath.</param>
/// <returns>The immutable snapshot.</returns>
public static AuthoredTable Build(IReadOnlyDictionary<string, MqttTagDefinition> byRawPath)
{
var byFilter = new Dictionary<string, List<MqttTagDefinition>>(StringComparer.Ordinal);
var concrete = new Dictionary<string, List<MqttTagDefinition>>(StringComparer.Ordinal);
var wildcards = new List<MqttTagDefinition>();
foreach (var def in byRawPath.Values)
{
if (def.Topic.Contains(MqttTopicFilterComparer.SingleLevelWildcard, StringComparison.Ordinal)
|| def.Topic.Contains(MqttTopicFilterComparer.MultiLevelWildcard, StringComparison.Ordinal))
// EVERY def lands here, wildcard or not — this is the index the reconnect and degrade
// paths read, and omitting wildcards from it is exactly the C1 silent-death defect.
if (!byFilter.TryGetValue(def.Topic, out var all))
{
byFilter[def.Topic] = all = [];
}
all.Add(def);
if (IsWildcard(def.Topic))
{
wildcards.Add(def);
continue;
@@ -1031,6 +1319,7 @@ public sealed class MqttSubscriptionManager
return new AuthoredTable(
byRawPath.ToFrozenDictionary(StringComparer.Ordinal),
byFilter.ToFrozenDictionary(kv => kv.Key, kv => kv.Value.ToArray(), StringComparer.Ordinal),
concrete.ToFrozenDictionary(kv => kv.Key, kv => kv.Value.ToArray(), StringComparer.Ordinal),
[.. wildcards]);
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerTests.cs
+244
View File
@@ -583,6 +583,250 @@ public sealed class MqttSubscriptionManagerTests
mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(Good);
}
// ---------------------------------------------------------------------------------
// C1 — wildcard-authored tags on the paths that reconstruct state from a FILTER string.
// Delivery matches an incoming topic (concrete); reconnect + degrade key off the subscribed
// filter, which for a wildcard tag is its PATTERN. Reading the concrete-only index there
// silently drops the tag: no exception, no log, last Good value frozen forever.
// ---------------------------------------------------------------------------------
[Fact]
public async Task OnReconnectedAsync_WildcardAuthoredTopic_IsReSubscribed()
{
var transport = new FakeTransport();
var mgr = NewManager(transport);
mgr.Register([Tag(OvenTempPath, """{"topic":"f/+/temp","payloadFormat":"Scalar","dataType":"Float64"}""")]);
await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
transport.Calls.Count.ShouldBe(1);
await mgr.OnReconnectedAsync(TestContext.Current.CancellationToken);
transport.Calls.Count.ShouldBe(2);
transport.Calls[1].Select(f => f.Topic).ShouldBe(["f/+/temp"]);
}
/// <summary>
/// The end-to-end shape of the defect: after a reconnect the wildcard tag must still receive
/// data. A dropped re-subscribe is invisible in every other assertion — the cache keeps its last
/// Good value, so nothing turns Bad.
/// </summary>
[Fact]
public async Task OnReconnectedAsync_WildcardAuthoredTopic_KeepsDeliveringAfterwards()
{
var transport = new FakeTransport();
var mgr = NewManager(transport);
mgr.Register([Tag(OvenTempPath, """{"topic":"f/+/temp","payloadFormat":"Scalar","dataType":"Float64"}""")]);
await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
await mgr.OnReconnectedAsync(TestContext.Current.CancellationToken);
// The broker only forwards what it was re-subscribed to; model that by asserting the filter is
// live, then that a matching publish still lands.
transport.Calls[^1].Select(f => f.Topic).ShouldContain("f/+/temp");
object? got = null;
mgr.OnDataChange += (_, e) => got = e.Snapshot.Value;
mgr.HandleMessage("f/oven/temp", "9.5"u8.ToArray(), retained: false);
got.ShouldBe(9.5);
}
/// <summary>
/// A total re-subscribe failure must throw for a wildcard-only driver too. Before the fix the
/// wildcard tag never reached the filter set, so <c>filters.Count == 0</c> took a silent early
/// return and the caller was told everything was fine.
/// </summary>
[Fact]
public async Task OnReconnectedAsync_WildcardOnly_TotalFailure_StillThrows()
{
var transport = new FakeTransport();
var mgr = NewManager(transport);
mgr.Register([Tag(OvenTempPath, """{"topic":"f/+/temp","payloadFormat":"Scalar","dataType":"Float64"}""")]);
await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
transport.Throw = new InvalidOperationException("broker gone");
await Should.ThrowAsync<Exception>(() => mgr.OnReconnectedAsync(TestContext.Current.CancellationToken));
}
[Fact]
public async Task SubscribeAsync_WildcardAuthoredTopic_SubackRejection_DegradesItsRefs()
{
var transport = new FakeTransport { Reject = _ => true };
var mgr = NewManager(transport);
mgr.Register([Tag(OvenTempPath, """{"topic":"f/+/temp","payloadFormat":"Scalar","dataType":"Float64"}""")]);
await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
// Before the fix this stayed at BadWaitingForInitialData — the broker refused the subscription
// and the tag never learned it was unreachable.
mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(BadCommunicationError);
}
[Fact]
public async Task OnReconnectedAsync_WildcardAuthoredTopic_PartialRejection_DegradesTheWildcardRefs()
{
var transport = new FakeTransport();
var mgr = NewManager(transport);
mgr.Register(
[
Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}"""),
Tag(OvenPressPath, """{"topic":"f/+/press","payloadFormat":"Scalar","dataType":"Float64"}"""),
]);
await mgr.SubscribeAsync([OvenTempPath, OvenPressPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
mgr.HandleMessage(OvenTopic, "1"u8.ToArray(), retained: false);
transport.Reject = t => t == "f/+/press";
await Should.NotThrowAsync(() => mgr.OnReconnectedAsync(TestContext.Current.CancellationToken));
mgr.Values.Read(OvenPressPath).StatusCode.ShouldBe(BadCommunicationError);
mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(Good);
}
// ---------------------------------------------------------------------------------
// I2 — bounded decode on the dispatcher thread
// ---------------------------------------------------------------------------------
[Fact]
public async Task HandleMessage_PayloadOverTheCeiling_IsRefusedBeforeDecode()
{
var mgr = new MqttSubscriptionManager(
new MqttDriverOptions { Mode = MqttMode.Plain }, "mqtt-1", maxPayloadBytes: 16);
mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"String"}""")]);
await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
DataValueSnapshot? snap = null;
mgr.OnDataChange += (_, e) => snap = e.Snapshot;
mgr.HandleMessage(OvenTopic, new byte[17], retained: false);
snap!.StatusCode.ShouldBe(BadDecodingError);
// …and a body at the ceiling still flows.
mgr.HandleMessage(OvenTopic, "0123456789ABCDEF"u8.ToArray(), retained: false);
snap.StatusCode.ShouldBe(Good);
}
[Fact]
public void MaxPayloadBytes_NonPositive_FallsBackToTheDefault() =>
new MqttSubscriptionManager(new MqttDriverOptions(), "d", maxPayloadBytes: 0)
.MaxPayloadBytes.ShouldBe(MqttSubscriptionManager.DefaultMaxPayloadBytes);
// ---------------------------------------------------------------------------------
// I3 — integral-valued reals against integer tags
// ---------------------------------------------------------------------------------
[Theory]
[InlineData("Int32", """{"value":5.0}""", 5)]
[InlineData("Int32", """{"value":5}""", 5)]
[InlineData("Int64", """{"value":5.0}""", 5L)]
[InlineData("Int16", """{"value":5.0}""", (short)5)]
[InlineData("UInt32", """{"value":5.0}""", 5u)]
public async Task HandleMessage_Json_IntegralValuedReal_CoercesToIntegerTag(
string dataType, string payload, object expected)
{
var mgr = NewManager();
mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Json","jsonPath":"$.value","dataType":"{{dataType}}"}""")]);
await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
DataValueSnapshot? snap = null;
mgr.OnDataChange += (_, e) => snap = e.Snapshot;
mgr.HandleMessage(OvenTopic, System.Text.Encoding.UTF8.GetBytes(payload), retained: false);
snap!.StatusCode.ShouldBe(Good);
snap.Value.ShouldBe(expected);
}
/// <summary>A genuinely fractional value is still refused — never rounded into a wrong value.</summary>
[Fact]
public async Task HandleMessage_Json_FractionalReal_StillRefusedByIntegerTag()
{
var mgr = NewManager();
mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Json","jsonPath":"$.value","dataType":"Int32"}""")]);
await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
DataValueSnapshot? snap = null;
mgr.OnDataChange += (_, e) => snap = e.Snapshot;
mgr.HandleMessage(OvenTopic, """{"value":5.5}"""u8.ToArray(), retained: false);
snap!.StatusCode.ShouldBe(BadTypeMismatch);
snap.Value.ShouldBeNull();
}
[Fact]
public async Task HandleMessage_Json_OutOfRangeIntegralReal_IsRefused()
{
var mgr = NewManager();
mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Json","jsonPath":"$.value","dataType":"Int16"}""")]);
await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
DataValueSnapshot? snap = null;
mgr.OnDataChange += (_, e) => snap = e.Snapshot;
mgr.HandleMessage(OvenTopic, """{"value":99999.0}"""u8.ToArray(), retained: false);
snap!.StatusCode.ShouldBe(BadTypeMismatch);
}
[Fact]
public async Task HandleMessage_Scalar_IntegralValuedReal_CoercesToIntegerTag()
{
var mgr = NewManager();
mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Int32"}""")]);
await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
DataValueSnapshot? snap = null;
mgr.OnDataChange += (_, e) => snap = e.Snapshot;
mgr.HandleMessage(OvenTopic, "5.0"u8.ToArray(), retained: false);
snap!.StatusCode.ShouldBe(Good);
snap.Value.ShouldBe(5);
}
// ---------------------------------------------------------------------------------
// I5 — a JSON string holding a number, against a numeric tag (documented as supported)
// ---------------------------------------------------------------------------------
[Theory]
[InlineData("Float64", """{"value":"21.5"}""", 21.5)]
[InlineData("Int32", """{"value":"5"}""", 5)]
[InlineData("Boolean", """{"value":"true"}""", true)]
public async Task HandleMessage_Json_StringHoldingANumber_CoercesToTheNumericTag(
string dataType, string payload, object expected)
{
var mgr = NewManager();
mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Json","jsonPath":"$.value","dataType":"{{dataType}}"}""")]);
await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
DataValueSnapshot? snap = null;
mgr.OnDataChange += (_, e) => snap = e.Snapshot;
mgr.HandleMessage(OvenTopic, System.Text.Encoding.UTF8.GetBytes(payload), retained: false);
snap!.StatusCode.ShouldBe(Good);
snap.Value.ShouldBe(expected);
}
// ---------------------------------------------------------------------------------
// Stale-state pruning on redeploy
// ---------------------------------------------------------------------------------
[Fact]
public async Task Register_DroppingATag_StopsItsTopicBeingReSubscribedForever()
{
var transport = new FakeTransport();
var mgr = NewManager(transport);
mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]);
await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
mgr.Register([]); // redeploy drops the tag
await mgr.OnReconnectedAsync(TestContext.Current.CancellationToken);
transport.Calls.Count.ShouldBe(1); // the initial subscribe only — no zombie re-subscribe
}
/// <summary>Records every subscribe call so dedupe / re-subscribe can be asserted without a broker.</summary>
private sealed class FakeTransport : IMqttSubscribeTransport
{