diff --git a/docs/drivers/Mqtt.md b/docs/drivers/Mqtt.md index f8aef796..8400ab42 100644 --- a/docs/drivers/Mqtt.md +++ b/docs/drivers/Mqtt.md @@ -408,9 +408,18 @@ value. `MqttConnection` therefore raises `MqttConnectRejectedException`: code. (`ServerMoved` is permanent, `UseAnotherServer` is temporary — per the MQTT 5 spec.) A reconnect that completes without re-subscribing would be a healthy-looking connection that receives -nothing forever, so `OnReconnectedAsync` **throws** when zero filters are granted, tearing the session -down rather than swallowing it. A SUBACK failure degrades just the affected RawPaths to -`BadCommunicationError`. +nothing forever, so `OnReconnectedAsync` **throws** when none of the previously live filters is granted +(or the SUBSCRIBE gets no verdict), tearing the session down rather than swallowing it. A SUBACK failure +degrades just the affected RawPaths to `BadCommunicationError`. + +**A failed SUBSCRIBE is retried, never forgotten (R06:M-1).** Every topic whose SUBSCRIBE was refused or got +no verdict joins a *pending* set that the next `SubscribeAsync` and every reconnect re-request; a grant moves +it to the live set, and a redeploy that drops its tags drops it. Pending topics do not count toward the +reconnect's total-failure verdict, so one ACL-denied topic cannot reconnect-loop the driver. A plain-mode +`SubscribeAsync` whose SUBSCRIBE got **no verdict at all** (transport throw, SUBSCRIBE deadline) degrades its +refs and then **throws**, so `DriverInstanceActor` holds no handle and its health-tick reconcile re-subscribes; +a SUBACK *refusal* still returns a handle (the broker answered). A subscribe issued with no transport attached +records its topics as pending, never as established (M-11). There are **two** kinds of reconnect and only one of them fires that callback. A drop the connection's own supervisor recovers keeps the same `MqttConnection`, so `Reconnected` → `OnReconnectedAsync` 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 92c45051..9e4d8f64 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs @@ -113,6 +113,14 @@ public sealed class MqttSubscriptionManager /// Topics already SUBSCRIBEd on the live session — the dedupe set and the re-subscribe set. private readonly ConcurrentDictionary _subscribedTopics = new(StringComparer.Ordinal); + /// + /// Topics a subscribe asked for whose SUBSCRIBE failed (refused, or no verdict at all). Retried by + /// every later and by ; moved to + /// when granted. Without it a failed topic left every retry set and + /// stayed dark behind a Healthy connection until a redeploy. + /// + private readonly ConcurrentDictionary _pendingTopics = new(StringComparer.Ordinal); + /// RawPaths whose extraction failure has already been logged loudly once. private readonly ConcurrentDictionary _warnedRefs = new(StringComparer.Ordinal); @@ -212,8 +220,8 @@ public sealed class MqttSubscriptionManager /// The manager does not own the connection's lifetime and never detaches — the driver disposes /// both together. Call once per connection, and before the first /// : a subscribe issued with no transport attached records its - /// intent (and warns) but sends nothing to a broker, and would only be established if the - /// connection later dropped and recovered. + /// topics as pending (and warns) but sends nothing to a broker; the next subscribe or + /// reconnect on an attached session retries them. /// /// /// Binding to a session forgets the previous session's established filters — the same @@ -328,6 +336,14 @@ public sealed class MqttSubscriptionManager } } + foreach (var topic in _pendingTopics.Keys) + { + if (!table.ByFilter.ContainsKey(topic)) + { + _pendingTopics.TryRemove(topic, out _); + } + } + foreach (var rawPath in _handleByRawPath.Keys) { if (!table.ByRawPath.ContainsKey(rawPath)) @@ -356,12 +372,13 @@ public sealed class MqttSubscriptionManager /// opening SUBSCRIBE itself; only re-establishes it later. /// /// - /// Never throws, never hangs. A reference that is not an authored tag is recorded as - /// BadNodeIdUnknown; a SUBACK rejection, transport failure or subscribe deadline - /// degrades exactly the affected references to BadCommunicationError. Either way a - /// handle is returned — the OPC UA server's subscribe call must not fail because one topic - /// is ACL-denied. Recovery for a transient failure comes from the reconnect path; - /// a persistent broker-side refusal is visible as a Bad status on the affected nodes. + /// Never hangs; throws only when the broker gave no verdict. A reference that is not an + /// authored tag is recorded as BadNodeIdUnknown. A SUBACK rejection degrades exactly the + /// affected references to BadCommunicationError and still returns a handle — the OPC UA + /// server's subscribe must not fail because one topic is ACL-denied. A transport failure or + /// subscribe deadline degrades the same way and then throws, so the host holds no handle + /// and its reconcile re-subscribes. Every failed topic also joins a pending set that the next + /// subscribe and every reconnect retry. /// /// /// The RawPath references to subscribe. @@ -414,12 +431,43 @@ public sealed class MqttSubscriptionManager // One SUBSCRIBE per distinct topic, and only for topics not already live: re-SUBSCRIBEing an // established filter is legal but pointless, and it re-triggers the retained seed. - var pending = defs.Where(d => !_subscribedTopics.ContainsKey(d.Topic)).ToList(); - await EstablishAsync(BuildFilters(pending, _defaultQos), cancellationToken).ConfigureAwait(false); + // Topics an EARLIER subscribe failed to establish ride along: a redeploy's bulk re-subscribe is then a + // retry of a refusal the broker may since have lifted, instead of a no-op that leaves them dark. + var authored = _authored; + var pending = defs.Where(d => !_subscribedTopics.ContainsKey(d.Topic)) + .Concat(PendingDefinitions(authored)) + .ToList(); + var outcome = await EstablishAsync(BuildFilters(pending, _defaultQos), cancellationToken).ConfigureAwait(false); + if (outcome.TransportFailure is { } failure) + { + // No verdict at all (transport throw / SUBSCRIBE deadline): THROW, so the caller holds no handle + // and DriverInstanceActor's health-tick reconcile re-subscribes. Returning a handle here was the + // silent-death case — the actor thought the subscription live and never retried. The refs already + // carry BadCommunicationError (raised under this handle above), so the throw costs no visibility. + _refsByHandle.TryRemove(handle, out _); + foreach (var def in defs) + { + _handleByRawPath.TryRemove(new KeyValuePair(def.Name, handle)); + } + + throw new InvalidOperationException( + $"MQTT driver '{_driverId}': SUBSCRIBE for {pending.Count} tag(s) got no broker verdict " + + $"({failure.GetType().Name}); the subscription will be retried.", + failure); + } + + // A SUBACK refusal (an ACL denial, say) does NOT throw: the broker answered, the refused topics are + // Bad + pending, and failing the whole subscribe would take every other topic in it dark too. return handle; } + /// The authored definitions behind every still-authored pending topic. + private IEnumerable PendingDefinitions(AuthoredTable authored) + => _pendingTopics.Keys + .Select(topic => authored.ByFilter.TryGetValue(topic, out var defs) ? defs : []) + .SelectMany(defs => defs); + /// /// Stops attributing notifications to . References another live /// subscription also covers keep publishing. @@ -498,8 +546,13 @@ public sealed class MqttSubscriptionManager .SelectMany(defs => defs) .ToList(); - var filters = BuildFilters(live, _defaultQos); - if (filters.Count == 0) + // Pending topics (a failed earlier SUBSCRIBE) are retried here too — a reconnect re-subscribing + // only the granted set is how a failed topic stayed dark forever. They do NOT count toward the + // total-failure verdict below: re-refusing a topic that was never live loses nothing, and throwing + // for it would turn one ACL-denied topic into a reconnect loop for the whole driver. + var pendingDefs = PendingDefinitions(authored).ToList(); + var filters = BuildFilters(live.Concat(pendingDefs), _defaultQos); + if (live.Count == 0 && pendingDefs.Count == 0) { // 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 @@ -519,13 +572,17 @@ public sealed class MqttSubscriptionManager return; } - var granted = await EstablishAsync(filters, cancellationToken).ConfigureAwait(false); - if (granted == 0) + var outcome = await EstablishAsync(filters, cancellationToken).ConfigureAwait(false); + var liveTopics = live.Select(d => d.Topic).Distinct(StringComparer.Ordinal).ToList(); + var liveLost = outcome.TransportFailure is not null + || (subscribed.Count > 0 && (liveTopics.Count == 0 || !liveTopics.Any(outcome.Granted.Contains))); + if (liveLost) { throw new InvalidOperationException( $"MQTT driver '{_driverId}': no subscription could be re-established after reconnect " + $"({filters.Count} filter(s) requested); tearing the session down rather than serving a " - + "connection that receives nothing."); + + "connection that receives nothing.", + outcome.TransportFailure); } } @@ -633,35 +690,35 @@ public sealed class MqttSubscriptionManager /// /// The filters to establish; an empty list is a no-op. /// Cancellation for the round-trip. - /// How many filters the broker granted. - private async Task EstablishAsync( + /// The granted topics, plus the transport failure when the broker gave no verdict at all. + private async Task EstablishAsync( IReadOnlyList filters, CancellationToken cancellationToken) { if (filters.Count == 0) { - return 0; + return EstablishOutcome.Nothing; } var transport = _transport; if (transport is null) { - // No transport: the intent is recorded so the reconnect path establishes it, but NOTHING - // reaches a broker now. Loud, because outside tests this means the composition order is - // wrong — AttachTo must run before the OPC UA server's first subscribe, or these topics - // stay unestablished until the connection happens to drop and recover. + // No transport: NOTHING reaches a broker, so nothing is recorded as subscribed — recording it + // was a silent "subscribed, receives nothing" if the AttachTo-before-subscribe ordering ever + // changed. The topics wait in the pending set, which the next subscribe and every reconnect + // retry. Loud, because outside tests this means the composition order is wrong. foreach (var filter in filters) { - _subscribedTopics[filter.Topic] = 0; + _pendingTopics[filter.Topic] = 0; } _logger?.LogWarning( - "MQTT driver '{DriverId}': {FilterCount} subscription(s) recorded with no connection attached; " - + "nothing was sent to a broker. Call AttachTo before subscribing.", + "MQTT driver '{DriverId}': {FilterCount} subscription(s) requested with no connection attached; " + + "nothing was sent to a broker and they stay pending. Call AttachTo before subscribing.", _driverId, filters.Count); - return filters.Count; + return EstablishOutcome.Nothing; } IReadOnlyList outcomes; @@ -681,16 +738,17 @@ public sealed class MqttSubscriptionManager DegradeTopic(filter.Topic, ex.GetType().Name); } - return 0; + return new EstablishOutcome(new HashSet(StringComparer.Ordinal), ex); } - var granted = 0; + var granted = new HashSet(StringComparer.Ordinal); foreach (var outcome in outcomes) { if (outcome.Granted) { _subscribedTopics[outcome.Topic] = 0; - granted++; + _pendingTopics.TryRemove(outcome.Topic, out _); + granted.Add(outcome.Topic); } else { @@ -703,7 +761,16 @@ public sealed class MqttSubscriptionManager } } - return granted; + return new EstablishOutcome(granted, null); + } + + /// One SUBSCRIBE round-trip's verdict. + /// The filters the broker granted. + /// Set when the call produced no verdict at all (throw / deadline). + private sealed record EstablishOutcome(IReadOnlySet Granted, Exception? TransportFailure) + { + public static readonly EstablishOutcome Nothing = + new(new HashSet(StringComparer.Ordinal), null); } /// Marks every authored reference on as unreachable. @@ -722,6 +789,7 @@ public sealed class MqttSubscriptionManager if (!_authored.ByFilter.TryGetValue(topic, out var defs)) { + _pendingTopics.TryRemove(topic, out _); _logger?.LogWarning( "MQTT driver '{DriverId}': cannot degrade tags for unestablished filter '{Topic}' ({Reason}) — " + "no authored tag resolves to it.", @@ -731,6 +799,9 @@ public sealed class MqttSubscriptionManager return; } + // Still authored, so it must be retried — see _pendingTopics. + _pendingTopics[topic] = 0; + var now = DateTime.UtcNow; var snapshot = new DataValueSnapshot(null, StatusBadCommunicationError, null, now); foreach (var def in defs) diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MiniBroker.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MiniBroker.cs index 7950a240..77d0b762 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MiniBroker.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MiniBroker.cs @@ -59,6 +59,54 @@ internal sealed class MiniBroker : IDisposable /// public byte ConnAckReturnCode { get; set; } = ReturnCodeAccepted; + /// + /// When true (the default) a SUBSCRIBE is answered with a SUBACK granting each filter's requested + /// QoS. False is a broker that accepts SUBSCRIBE and never answers — the subscribe-deadline shape. + /// + public bool AnswerSubscribe { get; set; } = true; + + /// + /// Builds the SUBACK for one SUBSCRIBE packet read whole into , or null when + /// the packet is not parseable from one read (the fake handles one packet per read, like CONNECT). + /// + private static byte[]? TryBuildSubAck(byte[] buffer, int read) + { + // Fixed header: type byte, then the variable-length "remaining length". + var at = 1; + var remaining = 0; + var multiplier = 1; + byte digit; + do + { + if (at >= read) return null; + digit = buffer[at++]; + remaining += (digit & 0x7F) * multiplier; + multiplier *= 128; + } + while ((digit & 0x80) != 0); + + var end = at + remaining; + if (end > read || remaining < 2) return null; + + var packetIdHigh = buffer[at]; + var packetIdLow = buffer[at + 1]; + at += 2; + + var grants = new List(); + while (at + 2 <= end) + { + var length = (buffer[at] << 8) | buffer[at + 1]; + at += 2 + length; + if (at >= end) return null; + grants.Add((byte)(buffer[at] & 0x03)); + at++; + } + + var ack = new List { 0x90, (byte)(2 + grants.Count), packetIdHigh, packetIdLow }; + ack.AddRange(grants); + return [.. ack]; + } + /// Total TCP connections accepted since construction — counts reconnect attempts. public int AcceptedCount => Volatile.Read(ref _acceptedCount); @@ -163,6 +211,10 @@ internal sealed class MiniBroker : IDisposable { await stream.WriteAsync(PingResp, _cts.Token); } + else if (packetType == 0x80 && AnswerSubscribe && TryBuildSubAck(buffer, read) is { } subAck) + { + await stream.WriteAsync(subAck, _cts.Token); + } } } catch (Exception) diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverSessionLifecycleTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverSessionLifecycleTests.cs index ca4e667e..947fb081 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverSessionLifecycleTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverSessionLifecycleTests.cs @@ -13,9 +13,9 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests; /// ( / ). /// /// -/// Plain mode deliberately: it is the only mode whose initialize needs no SUBSCRIBE, and -/// answers CONNECT and PINGREQ and nothing else — so a Sparkplug driver -/// would park on its group SUBSCRIBE until the deadline and prove nothing about connections. +/// Plain mode deliberately: it is the only mode whose initialize needs no SUBSCRIBE. +/// answers CONNECT, SUBSCRIBE (unless told not to) and PINGREQ, and never +/// PUBLISHes, so ingest is driven through Subscriptions.HandleMessage. /// [Trait("Category", "Unit")] public sealed class MqttDriverSessionLifecycleTests @@ -122,7 +122,7 @@ public sealed class MqttDriverSessionLifecycleTests /// /// /// Ingest is fed through Subscriptions.HandleMessage, the method MQTTnet's dispatcher - /// calls, because answers CONNECT and PINGREQ and nothing else: it + /// calls, because answers CONNECT, SUBSCRIBE and PINGREQ only: it /// can drop a session (which is what this test needs from it) but cannot PUBLISH. /// /// @@ -174,6 +174,51 @@ public sealed class MqttDriverSessionLifecycleTests recovered.Value.ShouldBe(21.5); } + /// + /// R06:M-1, end to end. A SUBSCRIBE the broker never answers used to come back from + /// as a live handle while the topic left every retry set, so + /// DriverInstanceActor never re-subscribed and the tag stayed dark behind a Healthy connection. + /// It must throw (the actor's reconcile then retries), and the retry against a broker that answers + /// must establish and deliver — the positive control. + /// + [Trait("Category", "SlowUnit")] + [Fact] + public async Task ASubscribeTheBrokerNeverAnswers_Throws_AndTheRetryEstablishes() + { + const string rawPath = "Plant/Mqtt/dev1/Temp"; + const string topic = "factory/oven/temp"; + + using var broker = new MiniBroker { AnswerSubscribe = false }; + var options = BrokerOptions(broker.Port) with + { + RawTags = + [ + new RawTagEntry( + rawPath, + $$"""{"topic":"{{topic}}","payloadFormat":"Scalar","dataType":"Float64"}""", + WriteIdempotent: false), + ], + }; + await using var driver = new MqttDriver(options, "d", null); + await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken); + + await Should.ThrowAsync(() => + driver.SubscribeAsync([rawPath], TimeSpan.Zero, TestContext.Current.CancellationToken)); + driver.GetHealth().State.ShouldBe(DriverState.Healthy, "the connection itself is fine — that was the trap"); + + broker.AnswerSubscribe = true; + var handle = await driver.SubscribeAsync([rawPath], TimeSpan.Zero, TestContext.Current.CancellationToken); + + var raised = new List(); + driver.OnDataChange += (_, e) => { lock (raised) raised.Add(e); }; + driver.Subscriptions.HandleMessage(topic, "7"u8.ToArray(), retained: false); + lock (raised) + { + raised.ShouldHaveSingleItem().SubscriptionHandle.ShouldBe(handle); + raised[0].Snapshot.StatusCode.ShouldBe(0u); + } + } + private static async Task WaitForAsync(Func condition, string what) { var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10); 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 558d4477..632b0925 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 @@ -538,17 +538,171 @@ public sealed class MqttSubscriptionManagerTests mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(BadCommunicationError); } + /// + /// A SUBSCRIBE that never got a verdict (transport throw / deadline) degrades its refs AND throws, + /// so DriverInstanceActor holds no handle and its health-tick reconcile re-subscribes. It + /// used to return a live-looking handle with the topic dropped from every retry set — dark behind a + /// Healthy connection until a redeploy. + /// [Fact] - public async Task SubscribeAsync_TransportThrows_DegradesRefsToBad_DoesNotThrow() + public async Task SubscribeAsync_TransportThrows_DegradesRefsToBad_AndThrows() { var transport = new FakeTransport { Throw = new InvalidOperationException("broker gone") }; var mgr = NewManager(transport); mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]); + var raised = new List(); + mgr.OnDataChange += (_, e) => raised.Add(e); - await Should.NotThrowAsync(() => + await Should.ThrowAsync(() => mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken)); mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(BadCommunicationError); + // The Bad verdict still reached the subscriber before the throw. + raised.ShouldContain(e => e.FullReference == OvenTempPath && e.Snapshot.StatusCode == BadCommunicationError); + } + + /// + /// The reconcile path: once the actor re-subscribes after the throw, the topic IS re-SUBSCRIBEd and + /// delivery resumes on the new handle — the positive control for the throw above. + /// + [Fact] + public async Task SubscribeAsync_AfterATransportThrow_TheRetrySubscribesAndDelivers() + { + var transport = new FakeTransport { Throw = new InvalidOperationException("broker gone") }; + var mgr = NewManager(transport); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]); + await Should.ThrowAsync(() => + mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken)); + + transport.Throw = null; + var handle = await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + var raised = new List(); + mgr.OnDataChange += (_, e) => raised.Add(e); + mgr.HandleMessage(OvenTopic, "21.5"u8.ToArray(), retained: false); + + transport.Calls.Count.ShouldBe(2); + transport.Calls[1].Select(f => f.Topic).ShouldBe([OvenTopic]); + raised.ShouldHaveSingleItem().SubscriptionHandle.ShouldBe(handle); + raised[0].Snapshot.Value.ShouldBe(21.5); + } + + /// + /// A topic whose first SUBSCRIBE was refused stays in a pending set the reconnect re-subscribe + /// covers. The reconnect used to re-SUBSCRIBE only the granted set, so it was never retried. + /// + [Fact] + public async Task OnReconnectedAsync_RetriesATopicWhoseFirstSubscribeWasRefused() + { + var transport = new FakeTransport { Reject = _ => true }; + var mgr = NewManager(transport); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]); + var handle = await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + transport.Reject = _ => false; + await mgr.OnReconnectedAsync(TestContext.Current.CancellationToken); + + transport.Calls.Count.ShouldBe(2); + transport.Calls[1].Select(f => f.Topic).ShouldBe([OvenTopic]); + + var raised = new List(); + mgr.OnDataChange += (_, e) => raised.Add(e); + mgr.HandleMessage(OvenTopic, "3"u8.ToArray(), retained: false); + raised.ShouldHaveSingleItem().SubscriptionHandle.ShouldBe(handle); + raised[0].Snapshot.StatusCode.ShouldBe(Good); + } + + /// + /// A persistently ACL-denied topic must not turn every reconnect into a teardown loop: a pending + /// topic refused again is not "re-established nothing" — nothing that was live was lost. Both a + /// mixed set (the granted topic still re-subscribes) and a pending-only set are covered. + /// + [Fact] + public async Task OnReconnectedAsync_PendingTopicsRefusedAgain_DoesNotThrow() + { + var transport = new FakeTransport { Reject = t => t == "f/oven/press" }; + var mgr = NewManager(transport); + mgr.Register( + [ + Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}"""), + Tag(OvenPressPath, """{"topic":"f/oven/press","payloadFormat":"Scalar","dataType":"Float64"}"""), + ]); + await mgr.SubscribeAsync([OvenTempPath, OvenPressPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + await Should.NotThrowAsync(() => mgr.OnReconnectedAsync(TestContext.Current.CancellationToken)); + transport.Calls[1].Select(f => f.Topic).OrderBy(t => t, StringComparer.Ordinal) + .ShouldBe(["f/oven/press", OvenTopic]); + + var lone = new FakeTransport { Reject = _ => true }; + var mgr2 = NewManager(lone); + mgr2.Register([Tag(OvenPressPath, """{"topic":"f/oven/press","payloadFormat":"Scalar","dataType":"Float64"}""")]); + await mgr2.SubscribeAsync([OvenPressPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + await Should.NotThrowAsync(() => mgr2.OnReconnectedAsync(TestContext.Current.CancellationToken)); + lone.Calls.Count.ShouldBe(2); // it WAS retried + } + + /// + /// A later subscribe on the same session (a redeploy's bulk re-subscribe) also retries the pending + /// set, so a broker ACL fixed while connected recovers without waiting for a reconnect. + /// + [Fact] + public async Task SubscribeAsync_RetriesPendingTopicsFromAnEarlierRefusal() + { + var transport = new FakeTransport { Reject = t => t == "f/oven/press" }; + var mgr = NewManager(transport); + mgr.Register( + [ + Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}"""), + Tag(OvenPressPath, """{"topic":"f/oven/press","payloadFormat":"Scalar","dataType":"Float64"}"""), + ]); + await mgr.SubscribeAsync([OvenPressPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + transport.Reject = _ => false; + await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + transport.Calls[1].Select(f => f.Topic).OrderBy(t => t, StringComparer.Ordinal) + .ShouldBe(["f/oven/press", OvenTopic]); + } + + /// + /// R06:M-11. A subscribe issued before any transport is attached reaches no broker, so it must + /// not be recorded as established. It used to be — "subscribed, receives nothing" — and the first + /// reconnect then had nothing to re-establish. Now it stays pending, and the first reconnect on the + /// attached session attempts it: the Bad verdict from the unconnected session is the proof. + /// + [Fact] + public async Task SubscribeAsync_WithNoTransport_StaysPending_AndTheFirstReconnectAttemptsIt() + { + var mgr = NewManager(transport: null); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]); + await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(0x80320000u); // BadWaitingForInitialData + + await using var attached = new MqttConnection( + new MqttDriverOptions { Host = "127.0.0.1", Port = 1, UseTls = false }, "mqtt-1"); + mgr.AttachTo(attached); + + // The unconnected session gives no verdict, so the reconnect throws — and it ATTEMPTED the topic. + await Should.ThrowAsync(() => mgr.OnReconnectedAsync(TestContext.Current.CancellationToken)); + mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(BadCommunicationError); + } + + /// + /// Register drops a pending topic whose tags the redeploy removed — the same housekeeping the + /// granted set gets — so a deleted tag's denied topic is not retried forever. + /// + [Fact] + public async Task Register_DroppingATag_RemovesItsTopicFromThePendingSet() + { + var transport = new FakeTransport { Reject = _ => true }; + var mgr = NewManager(transport); + mgr.Register([Tag(OvenPressPath, """{"topic":"f/oven/press","payloadFormat":"Scalar","dataType":"Float64"}""")]); + await mgr.SubscribeAsync([OvenPressPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]); + await mgr.OnReconnectedAsync(TestContext.Current.CancellationToken); + + transport.Calls.Count.ShouldBe(1); // nothing left to retry } // --------------------------------------------------------------------------------- @@ -939,7 +1093,10 @@ public sealed class MqttSubscriptionManagerTests new MqttDriverOptions { Host = "127.0.0.1", Port = 1, UseTls = false }, "mqtt-1"); mgr.AttachTo(rebound); - await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + // The unconnected session gives no verdict, so the subscribe now THROWS (M-1) — the throw is a + // second proof the SUBSCRIBE was attempted against the new session. + await Should.ThrowAsync(() => + mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken)); mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe( BadCommunicationError,