diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs index 8e8aa1e9..8438f456 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs @@ -59,6 +59,12 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt; /// public sealed class MqttSubscriptionManager { + /// + /// Default — 1 MiB. Generous for MQTT telemetry (which is + /// kilobytes) while still bounding the dispatcher-thread work any one publisher can impose. + /// + 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 IReadable. Defaults to a fresh one owned by this /// manager and exposed as . /// + /// + /// Ceiling on an inbound message body; see . Non-positive values + /// fall back to — "unbounded" is not offered, because the + /// bound exists to protect a shared dispatcher thread. + /// 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 /// public LastValueCache Values { get; } + /// + /// Ceiling on an inbound message body. A larger message is refused before any decode or parse + /// and degrades its own tags to BadDecodingError. + /// + /// + /// Decode and parse both run synchronously on MQTTnet's shared dispatcher thread, so their cost + /// is paid by every 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. + /// + /// Follow-up: this is a constructor knob, not yet an operator-facing one. Promoting + /// it to an MqttDriverOptions key (and pairing it with MQTTnet's + /// WithMaximumPacketSize, so an oversized body is refused at the protocol layer + /// rather than after the client has already buffered it) needs an edit to the + /// .Contracts project that this task was scoped out of. + /// + /// + public int MaxPayloadBytes { get; } + /// /// Wires this manager onto a live connection: inbound messages route to /// 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. /// /// - /// Runs on MQTTnet's dispatcher thread: no I/O, no locks, no allocation beyond the extracted - /// value, and no path throws. A tag whose payload is malformed degrades to a Bad status - /// code and the fan-out continues to the next tag. + /// + /// Runs on MQTTnet's dispatcher thread: no I/O, no locks, and no path throws. A tag + /// whose payload is malformed degrades to a Bad status code and the fan-out continues to + /// the next tag. + /// + /// + /// Bounded work. A body larger than is refused + /// before 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 + /// GetString / JsonDocument.Parse over a multi-megabyte body does exactly + /// that. The oversized message degrades its own tags to BadDecodingError. + /// + /// + /// Parsed once. 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. + /// /// /// The concrete topic the message arrived on. /// The message body; valid only for the duration of this call. @@ -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 } /// Marks every authored reference on as unreachable. + /// + /// is a subscribed filter string (it came from a requested + /// filter or a SUBACK outcome), so the lookup must go through ByFilter — the + /// concrete-only index misses every wildcard-authored tag and this method would then return + /// having silently degraded nothing, leaving the tag at BadWaitingForInitialData (or + /// worse, its last Good value) after the broker refused its subscription. + /// /// The topic filter whose subscription could not be established. /// The failure reason, for logs. 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 /// The message body. /// The message's retain flag. /// One clock read shared by the whole fan-out, so siblings agree. - private void Deliver(MqttTagDefinition def, ReadOnlySpan payload, bool retained, DateTime nowUtc) + /// Whether the body exceeded . + /// The fan-out's shared JSON parse, created on first use. + private void Deliver( + MqttTagDefinition def, + ReadOnlySpan 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 /// The authored tag definition. /// The message body. /// Timestamp for the snapshot — MQTT carries no source timestamp of its own. + /// The fan-out's shared JSON parse, created on first use by a Json-format tag. /// The extracted snapshot. - internal static DataValueSnapshot Extract(MqttTagDefinition def, ReadOnlySpan payload, DateTime nowUtc) + internal static DataValueSnapshot Extract( + MqttTagDefinition def, + ReadOnlySpan 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); - /// Parses the payload as JSON, selects the tag's JSONPath, and coerces to its data type. + /// + /// Selects the tag's JSONPath out of the fan-out's shared 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. + /// /// The authored tag definition. /// The message body. /// Timestamp for the snapshot. + /// The shared parse; parses on first call and caches the outcome. /// The extracted snapshot. - private static DataValueSnapshot ExtractJson(MqttTagDefinition def, ReadOnlySpan payload, DateTime nowUtc) + private static DataValueSnapshot ExtractJson( + MqttTagDefinition def, + ReadOnlySpan 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); + } + + /// + /// One message's JSON parse, shared across the whole fan-out. A mutable struct passed by + /// ref 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. + /// + internal struct PayloadJson : IDisposable + { + private JsonDocument? _doc; + private bool _attempted; + private bool _parsed; + + /// + /// 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. + /// + /// The message body; only read on the first call. + /// The document root when this returns . + /// when the payload is valid JSON. + public bool TryGetRoot(ReadOnlySpan 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; + } + + /// + 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 } } + /// Reads a JSON number that is written as an exact integer of the requested type. + /// The JSON number. + /// The target integer type. + /// The boxed value when this returns . + /// when the number is an exact integer in range. + 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; + } + + /// Parses text written as an exact integer of the requested type. + /// The trimmed text. + /// The target integer type. + /// The boxed value when this returns . + /// when the text is an exact integer in range. + 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; + } + + /// + /// Accepts a real that is exactly integral (5.0, 5E2) 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. + /// + /// + /// Parsed as rather than 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. + /// + /// The number's text (JSON raw text, or a Scalar payload). + /// The target integer type. + /// The boxed value when this returns . + /// when the text is an integral real in range. + 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; + } + /// /// 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; /// - /// 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. /// + /// + /// Two topic indexes, and the difference is load-bearing. Delivery matches an + /// incoming published topic, which never carries a wildcard, so it uses + /// (concrete only) plus a comparer scan of . + /// Every path that instead reconstructs state from a subscribed filter string — + /// reconnect re-subscribe and per-topic degradation — must use , which is + /// keyed by the tag's authored topic including wildcard patterns, because that is what + /// lands in _subscribedTopics 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. + /// /// RawPath → definition; the v3 resolver's backing table. - /// Concrete topic → every definition bound to it (the fan-out list). + /// + /// Authored topic filter string → every definition bound to it, wildcard patterns + /// included. Keyed the same way _subscribedTopics and SUBACK outcomes are. + /// + /// Concrete topic → its definitions; the delivery fast path. /// Definitions whose authored topic carries an MQTT wildcard. private sealed record AuthoredTable( FrozenDictionary ByRawPath, - FrozenDictionary ByTopic, + FrozenDictionary ByFilter, + FrozenDictionary ByExactTopic, MqttTagDefinition[] Wildcards) { public static readonly AuthoredTable Empty = Build(new Dictionary(StringComparer.Ordinal)); - /// Builds the snapshot, splitting concrete topics from wildcard-authored ones. + /// True when a topic string carries an MQTT wildcard and is therefore a filter, not a name. + /// The authored topic. + /// when it contains + or #. + public static bool IsWildcard(string topic) => + topic.Contains(MqttTopicFilterComparer.SingleLevelWildcard, StringComparison.Ordinal) + || topic.Contains(MqttTopicFilterComparer.MultiLevelWildcard, StringComparison.Ordinal); + + /// Builds the snapshot: every def into ByFilter, concrete/wildcard split for delivery. /// The mapped definitions, keyed by RawPath. /// The immutable snapshot. public static AuthoredTable Build(IReadOnlyDictionary byRawPath) { + var byFilter = new Dictionary>(StringComparer.Ordinal); var concrete = new Dictionary>(StringComparer.Ordinal); var wildcards = new List(); 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]); } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerTests.cs index c0f7b3f7..5fc1e966 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerTests.cs @@ -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"]); + } + + /// + /// 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. + /// + [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); + } + + /// + /// 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 filters.Count == 0 took a silent early + /// return and the caller was told everything was fine. + /// + [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(() => 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); + } + + /// A genuinely fractional value is still refused — never rounded into a wrong value. + [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 + } + /// Records every subscribe call so dedupe / re-subscribe can be asserted without a broker. private sealed class FakeTransport : IMqttSubscribeTransport {