From 2298485bdadc19c7a622c0f886ccd55312e3ae94 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 10 Aug 2026 07:58:29 -0400 Subject: [PATCH] fix(mqtt): a reconnect that re-establishes nothing tears the session down MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Driver.Mqtt-007 (Low, error handling). OnReconnectedAsync throws when a re-subscribe is granted zero filters — MqttConnection's contract is that a throwing Reconnected subscriber tears the fresh session down and retries under backoff, because a connected-but-deaf driver is the one outcome this path must never serve. But when the live subscribed topics resolved to zero filters in the first place it logged a warning and returned, reaching the same "nothing was re-established" state by a different route and giving it the opposite answer. Throw there too, and say so in the remarks so the type states ONE rule. It costs no other tag — every live topic is in the set that resolved to nothing, so there is nothing left flowing to protect, which is the whole reason the partial- rejection case is exempt — and it is not permanent: the retry runs against whatever Register has published by then. Test reaches the state the only way production can: Register prunes _subscribedTopics of the topics its new table no longer carries, but a SUBSCRIBE already in flight records its own topic AFTER that prune ran, so a deploy landing inside the round-trip leaves a live topic no authored tag resolves to. Paired with the control that a reconnect with no live subscriptions at all stays a no-op — throwing there would tear down the session of every driver whose tags have not been subscribed yet. --- .../MqttSubscriptionManager.cs | 34 +++++++----- .../MqttSubscriptionManagerTests.cs | 53 +++++++++++++++++++ 2 files changed, 73 insertions(+), 14 deletions(-) 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 14dc9d1d..320a6a04 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs @@ -397,6 +397,15 @@ public sealed class MqttSubscriptionManager /// rest keep flowing. That is the plan's "SUBACK failure → per-ref Bad, not hang", scoped /// to the case where it does not cost the whole driver. /// + /// + /// Live topics resolving to zero filters is a total failure too — it re-establishes + /// nothing, so it throws by the same rule rather than warning and returning. It costs no + /// other tag (every live topic is in the set that resolved to nothing) and it is not + /// permanent, because the retry runs against whatever has published + /// by then. It is reachable only when a deploy lands inside a subscribe round-trip: + /// prunes _subscribedTopics of the topics its new table no + /// longer carries, but an in-flight SUBSCRIBE records its own topic after that prune ran. + /// /// /// The connection's lifetime token; cancelled by its dispose. /// A task that completes when the re-subscribe has been answered. @@ -421,24 +430,21 @@ public sealed class MqttSubscriptionManager 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. + // Silence here is how the wildcard defect hid: no exception, no log, no SUBSCRIBE. Live + // topics resolving to zero filters re-establishes nothing, which is the total-failure case + // by a different route, so it gets the total-failure answer — see the remarks. 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); + throw new InvalidOperationException( + $"MQTT driver '{_driverId}': reconnect resolved NO filters for {subscribed.Count} live " + + $"topic(s) ({string.Join(", ", subscribed)}); tearing the session down rather than " + + "serving a connection that receives nothing."); } + _logger?.LogDebug( + "MQTT driver '{DriverId}': reconnect had no live subscriptions to re-establish.", + _driverId); + return; } 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 354caf4c..743c70bf 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,51 @@ public sealed class MqttSubscriptionManagerTests mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(Good); } + /// + /// Driver.Mqtt-007. Live subscribed topics that resolve to ZERO filters re-establish + /// nothing, which is the same outcome as a total SUBACK failure and must get the same answer: + /// throw, so tears the fresh session down and retries rather than + /// serving a connected-but-deaf driver. A warning was not enough — the whole point of the type's + /// split contract is that "no subscription was re-established" is fatal to the session. + /// + /// + /// The state is reached the only way production can reach it: + /// prunes _subscribedTopics of the topics its new table no longer carries, but a SUBSCRIBE + /// already in flight records its topic after that prune has run — so a redeploy landing + /// inside the subscribe round-trip leaves a live topic no authored tag resolves to. + /// + [Fact] + public async Task OnReconnectedAsync_LiveTopicResolvingNoFilter_Throws_RatherThanServingADeafSession() + { + var transport = new FakeTransport(); + var mgr = NewManager(transport); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]); + + transport.DuringSubscribe = () => mgr.Register([]); // the redeploy lands mid-round-trip + await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + transport.DuringSubscribe = null; + + await Should.ThrowAsync( + () => mgr.OnReconnectedAsync(TestContext.Current.CancellationToken)); + } + + /// + /// The falsifying control: a reconnect with no live subscriptions at all re-establishes nothing + /// because there is nothing to establish, and must stay a no-op. Throwing here would tear down + /// the session of every driver whose tags have not been subscribed yet. + /// + [Fact] + public async Task OnReconnectedAsync_NoLiveSubscriptions_IsANoOp() + { + var transport = new FakeTransport(); + var mgr = NewManager(transport); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]); + + await Should.NotThrowAsync(() => mgr.OnReconnectedAsync(TestContext.Current.CancellationToken)); + + transport.Calls.Count.ShouldBe(0); + } + // --------------------------------------------------------------------------------- // 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 @@ -898,10 +943,18 @@ public sealed class MqttSubscriptionManagerTests /// When set, the whole subscribe call throws instead of answering. public Exception? Throw { get; set; } + /// + /// Runs inside the subscribe round-trip, so a test can land a concurrent + /// in the window between its prune and this + /// call recording its topic as live. + /// + public Action? DuringSubscribe { get; set; } + public Task> SubscribeAsync( IReadOnlyList filters, CancellationToken cancellationToken) { Calls.Add(filters); + DuringSubscribe?.Invoke(); if (Throw is not null) throw Throw; IReadOnlyList outcomes = [.. filters.Select(f => new MqttSubscribeOutcome(f.Topic, !Reject(f.Topic), Reject(f.Topic) ? "NotAuthorized" : "GrantedQoS1"))];