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"))];