From a6370a26f8c182c8aeda0d23818c3e51307912ae Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 24 Jul 2026 15:32:41 -0400 Subject: [PATCH] fix(mqtt): make connect idempotent, bound the reconnect callback, stop State lying Closes the connect-vs-connect defect: MQTTnet throws (and raises no DisconnectedAsync) on a connect-while-connected, so a caller's ConnectAsync and the reconnect supervisor corrupted each other in both directions. Both paths now check first; when the supervisor finds the session already restored it stands down WITHOUT firing Reconnected. Reconnected becomes Func, fed from the lifetime token and capped at ConnectTimeoutSeconds, so a hung re-subscribe can no longer park the supervisor. Connected is published only after the re-subscribe succeeds; a supervisor that dies now reports Faulted instead of Reconnecting forever. Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW --- .../MqttConnection.cs | 314 +++++++++++++++--- .../MqttConnectionTests.cs | 287 +++++++++++++++- 2 files changed, 545 insertions(+), 56 deletions(-) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs index d01d2444..02f81e7d 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs @@ -22,8 +22,10 @@ public enum MqttConnectionState Reconnecting = 2, /// - /// Unrecoverable configuration: retrying cannot help, so the supervisor stops. Only - /// failures raised while assembling the client options (before any I/O) land here. + /// Unrecoverable without a driver restart: retrying cannot help, so the supervisor has + /// stopped. Two causes — unusable configuration (a failure raised while assembling the client + /// options, before any I/O), or the supervisor itself dying on an unexpected exception. A + /// broker that is merely unreachable never lands here. /// Faulted = 3, @@ -69,7 +71,64 @@ public enum MqttConnectionState /// persistent-session reconnects, where re-subscribing is merely redundant. Re-subscribing /// an already-subscribed topic is harmless; missing one is silent death. If the callback /// fails, the freshly-established session is torn down and retried rather than left -/// connected-but-deaf. +/// connected-but-deaf. It is invoked under a cancellation token and a +/// ceiling: a subscriber that hangs on +/// a broker which accepts SUBSCRIBE and never SUBACKs — the frozen-peer shape again — must +/// not be able to park the supervisor or outlive . +/// +/// +/// Connect is idempotent. MQTTnet throws (and raises no DisconnectedAsync) if +/// asked to connect a client that is already connected, so both connect paths check first and +/// return a no-op. The two paths genuinely race: the supervisor is told to expect the driver +/// host to re-run InitializeAsync, so a caller can restore the session while the +/// supervisor sits in backoff, and the supervisor can restore it a moment before a caller +/// asks. Without the check, the first case leaves the supervisor failing forever against a +/// perfectly healthy connection (inflating attempt, so the next genuine outage +/// waits instead of min) and the +/// second throws an undocumented out of +/// on a working session. When the supervisor finds the session +/// already restored it stops retrying without firing — the +/// caller that reconnected owns its own subscribe, per the +/// contract. +/// +/// +/// State transitions. is terminal; nothing +/// moves off it. +/// +/// +/// +/// +/// set by on success (or on finding the session +/// already up), and by the supervisor only after +/// has completed — so on the reconnect path +/// Connected means "connected and re-subscribed". +/// +/// +/// +/// +/// +/// set by the DisconnectedAsync callback when a live session drops, and +/// held by the supervisor for the whole retry + re-subscribe sequence. +/// +/// +/// +/// +/// +/// set when the client options cannot be assembled, or when the supervisor exits +/// on an unexpected exception. Both mean no further recovery without a restart. +/// +/// +/// +/// +/// set first thing in . +/// +/// +/// and can legitimately disagree in both +/// directions and are not interchangeable: Reconnecting with +/// IsConnected == true is the window where the socket is up but the re-subscribe has +/// not finished, and Connected with IsConnected == false is the instant between +/// the socket dropping and MQTTnet raising its callback. is the health +/// surface; is the transport fact. /// /// /// Credentials never leak. Nothing here logs or formats @@ -99,7 +158,15 @@ public sealed class MqttConnection : IAsyncDisposable /// private readonly SemaphoreSlim _lifecycleGate = new(1, 1); - /// Cancelled by ; aborts the supervisor and any in-flight connect. + /// + /// Cancelled by ; aborts the supervisor, any in-flight connect and + /// any in-flight callback. Never disposed — deliberately, like the + /// two semaphores: a connect racing dispose reads + /// after its own disposed check, and disposing the source would turn that benign loser into an + /// naming the wrong type. It holds no timer and no + /// surviving registrations (every linked source here is using-scoped), so leaving it + /// undisposed costs nothing. + /// private readonly CancellationTokenSource _lifetimeCts = new(); private readonly ILogger? _logger; @@ -135,12 +202,20 @@ public sealed class MqttConnection : IAsyncDisposable /// /// Raised after every successful reconnect (never after the initial - /// , which the caller already sequences its own subscribe behind). + /// , which the caller already sequences its own subscribe behind, + /// and never when the supervisor merely finds the session already restored by such a caller). /// Subscribers must re-establish their MQTT subscriptions here — see the type remarks for /// why skipping it is silent death. Every subscriber is invoked even if an earlier one /// throws; a throwing subscriber causes the session to be torn down and retried. /// - public event Func? Reconnected; + /// + /// The supplied token is cancelled by , and the whole fan-out is + /// additionally capped at — a subscriber + /// that ignores its token cannot park the supervisor or make it outlive teardown. Subscribers + /// should flow the token into their own network calls rather than relying on that ceiling, + /// which abandons rather than stops the offending work. + /// + public event Func? Reconnected; /// Whether the underlying client currently holds an established MQTT session. public bool IsConnected => _client?.IsConnected ?? false; @@ -285,11 +360,20 @@ public sealed class MqttConnection : IAsyncDisposable /// supervisor takes over keeping the session alive. /// /// - /// May be retried on the same instance after a failed attempt — the underlying - /// is created once and reused across attempts. Safe to call - /// concurrently with : the two are serialised, and a connect that - /// loses the race disposes whatever it built and throws - /// . + /// + /// May be retried on the same instance after a failed attempt — the underlying + /// is created once and reused across attempts. Safe to call + /// concurrently with : the two are serialised, and a connect + /// that loses the race disposes whatever it built and throws + /// . + /// + /// + /// Idempotent. Calling it on a session that is already established — including one + /// the reconnect supervisor restored a moment earlier — is a no-op that returns normally, + /// not the MQTTnet would raise for a + /// connect-while-connected. The caller still owns re-establishing its own subscriptions + /// after this returns; is not fired for this path. + /// /// /// Caller cancellation; linked with the connect deadline. /// The connect deadline elapsed. @@ -297,21 +381,22 @@ public sealed class MqttConnection : IAsyncDisposable /// /// The connection was disposed, either before the attempt started or while it was in flight. /// - public Task ConnectAsync(CancellationToken cancellationToken) + public async Task ConnectAsync(CancellationToken cancellationToken) { ObjectDisposedException.ThrowIf(Disposed, this); - return ConnectCoreAsync(startSupervisor: true, cancellationToken); + await ConnectCoreAsync(supervisorAttempt: false, cancellationToken).ConfigureAwait(false); } /// /// Invokes every subscriber. Walks the invocation list explicitly: - /// awaiting a multicast directly yields only the last - /// subscriber's task and abandons the earlier ones, so a single throwing subscriber would - /// silently skip every subscriber behind it — and a skipped re-subscribe is a topic that goes - /// dark. Failures are collected and rethrown after every subscriber has had its turn. + /// awaiting a multicast delegate directly yields only the last subscriber's task and + /// abandons the earlier ones, so a single throwing subscriber would silently skip every + /// subscriber behind it — and a skipped re-subscribe is a topic that goes dark. Failures are + /// collected and rethrown after every subscriber has had its turn. /// - internal async Task FireReconnectedAsync() + /// Passed to every subscriber; cancelled by . + internal async Task FireReconnectedAsync(CancellationToken cancellationToken) { var subscribers = Reconnected; if (subscribers is null) @@ -320,11 +405,11 @@ public sealed class MqttConnection : IAsyncDisposable } List? failures = null; - foreach (var subscriber in subscribers.GetInvocationList().Cast>()) + foreach (var subscriber in subscribers.GetInvocationList().Cast>()) { try { - await subscriber().ConfigureAwait(false); + await subscriber(cancellationToken).ConfigureAwait(false); } catch (Exception ex) { @@ -340,7 +425,23 @@ public sealed class MqttConnection : IAsyncDisposable } } - private async Task ConnectCoreAsync(bool startSupervisor, CancellationToken cancellationToken) + /// + /// The single gated connect both entry points funnel through. + /// + /// + /// true when the reconnect supervisor is calling. It suppresses publishing + /// on success, because on the reconnect path + /// "connected" is only true once has re-subscribed — publishing it + /// here would advertise a healthy connection during the window in which it has no + /// subscriptions at all. It also suppresses starting the supervisor (it is the supervisor). + /// + /// Caller cancellation; linked with the connect deadline. + /// + /// true if this call established the session; false if it found one already + /// established and did nothing. The supervisor uses false to stop retrying without + /// firing — whoever established that session owns its subscribe. + /// + private async Task ConnectCoreAsync(bool supervisorAttempt, CancellationToken cancellationToken) { // The options are rebuilt per attempt so a rotated CA file is picked up on the next connect // rather than being pinned for the life of the process. This is also the only step that can @@ -397,6 +498,23 @@ public sealed class MqttConnection : IAsyncDisposable _client = client; } + // Idempotence, and the whole point of the gate: MQTTnet throws + // "not allowed to connect with a server after the connection is established" for a + // connect-while-connected AND raises no DisconnectedAsync for it, so neither the retry + // loop nor Classify would ever recover. Both entry points race for real — see the type + // remarks — so whichever arrives second must find the session and leave it alone. + if (client.IsConnected) + { + SetState(MqttConnectionState.Connected); + StartSupervisorIfCallerConnect(supervisorAttempt); + _logger?.LogDebug( + "MQTT driver '{DriverId}': connect to {Host}:{Port} skipped — the session is already established.", + _driverId, + _options.Host, + _options.Port); + return false; + } + await client.ConnectAsync(clientOptions, linked.Token).ConfigureAwait(false); if (AfterConnectHookForTests is { } hook) @@ -417,12 +535,15 @@ public sealed class MqttConnection : IAsyncDisposable + $"{_options.Host}:{_options.Port} was in flight; the established session was torn down."); } - SetState(MqttConnectionState.Connected); - - if (startSupervisor) + // On the supervisor path the session is up but has no subscriptions yet, so State stays + // Reconnecting until FireReconnectedAsync succeeds — see the state table in the remarks. + if (!supervisorAttempt) { - _reconnectWorker ??= Task.Run(() => ReconnectSupervisorAsync(_lifetimeCts.Token)); + SetState(MqttConnectionState.Connected); } + + StartSupervisorIfCallerConnect(supervisorAttempt); + return true; } catch (ObjectDisposedException) { @@ -438,6 +559,18 @@ public sealed class MqttConnection : IAsyncDisposable } } + /// + /// Starts the supervisor on the first successful caller connect. Always called while + /// holding the lifecycle gate, so the ??= needs no further synchronisation. + /// + private void StartSupervisorIfCallerConnect(bool supervisorAttempt) + { + if (!supervisorAttempt) + { + _reconnectWorker ??= Task.Run(() => ReconnectSupervisorAsync(_lifetimeCts.Token)); + } + } + /// /// Maps a connect failure onto this type's documented contract. MQTTnet wraps a cancelled /// connect in MqttConnectingFailedException rather than letting the @@ -537,6 +670,14 @@ public sealed class MqttConnection : IAsyncDisposable for (var attempt = 0; !cancellationToken.IsCancellationRequested && !Disposed; attempt++) { + // A caller's ConnectAsync may have restored the session while we slept. Bail + // before burning the backoff, not just before burning an attempt. + if (IsConnected) + { + AdoptSessionRestoredByCaller(); + break; + } + await Task .Delay( NextBackoff( @@ -546,11 +687,13 @@ public sealed class MqttConnection : IAsyncDisposable cancellationToken) .ConfigureAwait(false); + bool established; try { - await ConnectCoreAsync(startSupervisor: false, cancellationToken).ConfigureAwait(false); + established = await ConnectCoreAsync(supervisorAttempt: true, cancellationToken) + .ConfigureAwait(false); } - catch (ObjectDisposedException) + catch (ObjectDisposedException) when (Disposed) { return; } @@ -575,25 +718,24 @@ public sealed class MqttConnection : IAsyncDisposable continue; } - try + if (!established) { - // ALWAYS, on every reconnect — including persistent sessions, where this is - // merely redundant. Subscriptions do not survive a clean session, and a - // reconnect that skips this leaves a healthy-looking, permanently deaf client. - await FireReconnectedAsync().ConfigureAwait(false); + // Someone else restored the session between the backoff and the connect. + // Firing Reconnected here would re-subscribe on top of a caller that is about + // to do exactly that itself, and — worse — leaving the loop running would + // keep failing against a healthy connection forever. + AdoptSessionRestoredByCaller(); + break; } - catch (Exception ex) + + if (!await TryReSubscribeAsync(cancellationToken).ConfigureAwait(false)) { - _logger?.LogError( - ex, - "MQTT driver '{DriverId}': re-subscribe after reconnect failed; tearing the session down " - + "and retrying rather than serving a connection that receives nothing.", - _driverId); - SetState(MqttConnectionState.Reconnecting); - await ForceDisconnectAsync().ConfigureAwait(false); continue; } + // Only now is "Connected" true in the sense this type sells it: connected AND + // subscribed. + SetState(MqttConnectionState.Connected); _logger?.LogInformation( "MQTT driver '{DriverId}': reconnected to {Host}:{Port} after {Attempts} attempt(s); " + "subscriptions re-established.", @@ -612,7 +754,9 @@ public sealed class MqttConnection : IAsyncDisposable catch (Exception ex) { // Reaching here means the connection is permanently dark, which is exactly the failure - // this supervisor exists to prevent — it must never be swallowed quietly. + // this supervisor exists to prevent — it must never be swallowed quietly, and State must + // not keep reporting Reconnecting, which every consumer reads as "recovering". + SetState(MqttConnectionState.Faulted); _logger?.LogError( ex, "MQTT driver '{DriverId}': reconnect supervisor stopped unexpectedly; the broker session will not " @@ -621,6 +765,71 @@ public sealed class MqttConnection : IAsyncDisposable } } + /// + /// The session came back without this supervisor's help — a caller's + /// won the race while we were in backoff. Publish the truth and stand down; deliberately does + /// not fire , because that caller owns its own re-subscribe. + /// + private void AdoptSessionRestoredByCaller() + { + SetState(MqttConnectionState.Connected); + _logger?.LogDebug( + "MQTT driver '{DriverId}': reconnect stood down — the session to {Host}:{Port} was restored elsewhere.", + _driverId, + _options.Host, + _options.Port); + } + + /// + /// Fires — ALWAYS, on every reconnect, including persistent + /// sessions where it is merely redundant, because subscriptions do not survive a clean + /// session and a reconnect that skips this leaves a healthy-looking, permanently deaf client. + /// + /// + /// false if the re-subscribe failed or overran its deadline, having torn the session + /// down so the next attempt starts from a clean CONNECT — serving a connection that receives + /// nothing is the one outcome this type must never produce. + /// + private async Task TryReSubscribeAsync(CancellationToken cancellationToken) + { + // Bounded like every other wait here. A subscriber that flows the token cancels promptly; one + // that ignores it is abandoned at the ceiling rather than being allowed to park the + // supervisor past DisposeAsync. + var fanOut = FireReconnectedAsync(cancellationToken); + + // The ceiling abandons the task rather than stopping it, so its eventual failure must still + // be observed or it resurfaces as an unobserved TaskException on the finalizer thread. + _ = fanOut.ContinueWith( + static abandoned => _ = abandoned.Exception, + CancellationToken.None, + TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, + TaskScheduler.Default); + + try + { + await fanOut + .WaitAsync(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds), cancellationToken) + .ConfigureAwait(false); + return true; + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; // Teardown — let the supervisor's own cancellation handling end the loop. + } + catch (Exception ex) + { + _logger?.LogError( + ex, + "MQTT driver '{DriverId}': re-subscribe after reconnect failed or overran {TimeoutSeconds}s; tearing " + + "the session down and retrying rather than serving a connection that receives nothing.", + _driverId, + _options.ConnectTimeoutSeconds); + SetState(MqttConnectionState.Reconnecting); + await ForceDisconnectAsync().ConfigureAwait(false); + return false; + } + } + /// /// Drops a session that is connected but unusable, so the next supervisor pass starts from a /// clean CONNECT. Best-effort and bounded — failure here just means the next attempt sees a @@ -666,15 +875,28 @@ public sealed class MqttConnection : IAsyncDisposable } } - /// is terminal — nothing may move off it. + /// + /// is terminal — nothing may move off it. A + /// check-then-write would let a thread preempted between the two steps resurrect a disposed + /// connection's state, so the transition is a compare-and-swap: the terminal check and the + /// write are the same atomic operation. + /// private void SetState(MqttConnectionState state) { - if (Volatile.Read(ref _state) == (int)MqttConnectionState.Disposed) + var desired = (int)state; + while (true) { - return; - } + var current = Volatile.Read(ref _state); + if (current == (int)MqttConnectionState.Disposed || current == desired) + { + return; + } - Volatile.Write(ref _state, (int)state); + if (Interlocked.CompareExchange(ref _state, desired, current) == current) + { + return; + } + } } /// diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttConnectionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttConnectionTests.cs index b4b5a3b9..a0986238 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttConnectionTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttConnectionTests.cs @@ -4,6 +4,7 @@ using System.Net.Security; using System.Net.Sockets; using System.Security.Cryptography; using System.Security.Cryptography.X509Certificates; +using Microsoft.Extensions.Logging; using MQTTnet; using Shouldly; using Xunit; @@ -447,10 +448,10 @@ public sealed class MqttConnectionTests await using var conn = new MqttConnection(LoopbackOpts(), driverId: "t", logger: null); var first = 0; var second = 0; - conn.Reconnected += () => { Interlocked.Increment(ref first); return Task.CompletedTask; }; - conn.Reconnected += () => { Interlocked.Increment(ref second); return Task.CompletedTask; }; + conn.Reconnected += _ => { Interlocked.Increment(ref first); return Task.CompletedTask; }; + conn.Reconnected += _ => { Interlocked.Increment(ref second); return Task.CompletedTask; }; - await conn.FireReconnectedAsync(); + await conn.FireReconnectedAsync(TestContext.Current.CancellationToken); first.ShouldBe(1); second.ShouldBe(1); @@ -467,10 +468,10 @@ public sealed class MqttConnectionTests { await using var conn = new MqttConnection(LoopbackOpts(), driverId: "t", logger: null); var behindTheThrower = 0; - conn.Reconnected += () => throw new InvalidOperationException("SUBACK refused"); - conn.Reconnected += () => { Interlocked.Increment(ref behindTheThrower); return Task.CompletedTask; }; + conn.Reconnected += _ => throw new InvalidOperationException("SUBACK refused"); + conn.Reconnected += _ => { Interlocked.Increment(ref behindTheThrower); return Task.CompletedTask; }; - await Should.ThrowAsync(async () => await conn.FireReconnectedAsync()); + await Should.ThrowAsync(async () => await conn.FireReconnectedAsync(TestContext.Current.CancellationToken)); behindTheThrower.ShouldBe(1); } @@ -480,7 +481,7 @@ public sealed class MqttConnectionTests { await using var conn = new MqttConnection(LoopbackOpts(), driverId: "t", logger: null); - await Should.NotThrowAsync(async () => await conn.FireReconnectedAsync()); + await Should.NotThrowAsync(async () => await conn.FireReconnectedAsync(TestContext.Current.CancellationToken)); } // --------------------------------------------------------------------------------- @@ -594,7 +595,7 @@ public sealed class MqttConnectionTests using var broker = new MiniBroker(); await using var conn = new MqttConnection(BrokerOpts(broker.Port), driverId: "t", logger: null); var resubscribes = 0; - conn.Reconnected += () => { Interlocked.Increment(ref resubscribes); return Task.CompletedTask; }; + conn.Reconnected += _ => { Interlocked.Increment(ref resubscribes); return Task.CompletedTask; }; await conn.ConnectAsync(CancellationToken.None); @@ -684,20 +685,248 @@ public sealed class MqttConnectionTests { using var broker = new MiniBroker(); var conn = new MqttConnection(BrokerOpts(broker.Port), driverId: "t", logger: null); + + // No assertions inside the hook: it runs inside the production try, so a Shouldly throw would + // be laundered through Classify and resurface as a confusing TimeoutException. Record, assert + // outside. + var raceReproduced = false; conn.AfterConnectHookForTests = async () => { _ = Task.Run(async () => await conn.DisposeAsync()); - (await WaitUntilAsync(() => conn.State == MqttConnectionState.Disposed, TimeSpan.FromSeconds(5))) - .ShouldBeTrue("dispose never flagged the instance — the race was not reproduced"); + raceReproduced = await WaitUntilAsync( + () => conn.State == MqttConnectionState.Disposed, + TimeSpan.FromSeconds(5)); }; await Should.ThrowAsync(async () => await conn.ConnectAsync(CancellationToken.None)); + raceReproduced.ShouldBeTrue("dispose never flagged the instance — the race was not reproduced"); (await WaitUntilAsync(() => broker.LiveConnections == 0, TimeSpan.FromSeconds(15))) .ShouldBeTrue($"a live broker connection leaked ({broker.LiveConnections} still open)"); conn.IsConnected.ShouldBeFalse(); } + // --------------------------------------------------------------------------------- + // C1 — connect-vs-connect: the supervisor and ConnectAsync must not corrupt each other + // --------------------------------------------------------------------------------- + + /// + /// Direction (b), the simplest form: MQTTnet throws + /// InvalidOperationException: It is not allowed to connect with a server after the + /// connection is established for a connect-while-connected. That is outside this type's + /// documented contract, so a caller re-running InitializeAsync against a healthy + /// session would look like a hard driver failure. + /// + [Fact] + public async Task ConnectAsync_OnAnAlreadyEstablishedSession_IsANoOp_NotAnInvalidOperation() + { + using var broker = new MiniBroker(); + await using var conn = new MqttConnection(BrokerOpts(broker.Port), driverId: "t", logger: null); + + await conn.ConnectAsync(CancellationToken.None); + await Should.NotThrowAsync(async () => await conn.ConnectAsync(CancellationToken.None)); + + conn.IsConnected.ShouldBeTrue(); + conn.State.ShouldBe(MqttConnectionState.Connected); + broker.AcceptedCount.ShouldBe(1, "the second connect opened a second socket instead of no-opping"); + } + + /// + /// Direction (b) through the supervisor: the supervisor restores the session, then the driver + /// host re-runs InitializeAsync — exactly the sequence this design relies on. + /// + [Fact] + public async Task ConnectAsync_AfterTheSupervisorAlreadyReconnected_IsANoOp() + { + using var broker = new MiniBroker(); + await using var conn = new MqttConnection(BrokerOpts(broker.Port), driverId: "t", logger: null); + var resubscribes = 0; + conn.Reconnected += _ => { Interlocked.Increment(ref resubscribes); return Task.CompletedTask; }; + + await conn.ConnectAsync(CancellationToken.None); + broker.DropAll(); + (await WaitUntilAsync( + () => Volatile.Read(ref resubscribes) >= 1 && conn.State == MqttConnectionState.Connected, + TimeSpan.FromSeconds(30))) + .ShouldBeTrue(); + + await Should.NotThrowAsync(async () => await conn.ConnectAsync(CancellationToken.None)); + + conn.IsConnected.ShouldBeTrue(); + conn.State.ShouldBe(MqttConnectionState.Connected); + } + + /// + /// Direction (a): a caller reconnects while the supervisor is asleep in backoff. The + /// supervisor must find the restored session and stand down — silently, without firing + /// Reconnected (the caller owns its own re-subscribe). Left unfixed it fails forever + /// against a perfectly healthy connection, at Debug level, inflating attempt so the + /// NEXT genuine outage waits max backoff instead of min. + /// + [Fact] + public async Task CallerReconnectsWhileTheSupervisorIsInBackoff_SupervisorStandsDownSilently() + { + using var broker = new MiniBroker(); + var log = new CapturingLogger(); + var opts = BrokerOpts(broker.Port) with { ReconnectMinBackoffSeconds = 3, ReconnectMaxBackoffSeconds = 3 }; + await using var conn = new MqttConnection(opts, driverId: "t", logger: log); + var resubscribes = 0; + conn.Reconnected += _ => { Interlocked.Increment(ref resubscribes); return Task.CompletedTask; }; + + await conn.ConnectAsync(CancellationToken.None); + broker.DropAll(); + (await WaitUntilAsync(() => conn.State == MqttConnectionState.Reconnecting, TimeSpan.FromSeconds(10))) + .ShouldBeTrue(); + + // The caller wins the race while the supervisor sleeps off its 3 s backoff. + await conn.ConnectAsync(CancellationToken.None); + conn.IsConnected.ShouldBeTrue(); + var failuresBefore = log.CountContaining("reconnect attempt"); + + // Three more supervisor wake points would land in this window if it were still looping. + await Task.Delay(TimeSpan.FromSeconds(9), TestContext.Current.CancellationToken); + + log.CountContaining("reconnect attempt") + .ShouldBe(failuresBefore, "the supervisor kept failing against a healthy connection"); + conn.State.ShouldBe(MqttConnectionState.Connected); + conn.IsConnected.ShouldBeTrue(); + Volatile.Read(ref resubscribes).ShouldBe(0, "the caller owns its own re-subscribe; Reconnected must not fire"); + } + + /// + /// The operational consequence of direction (a), measured rather than argued: after a caller + /// has won a reconnect race, the supervisor must be parked at its outer wait with its attempt + /// counter reset, so the NEXT genuine outage recovers at MIN backoff. + /// + /// + /// The timings are load-bearing, not arbitrary. A supervisor still flailing against the healthy + /// connection would have burned attempts 0 and 1 (at t≈2 s and t≈6 s) and be asleep in an 8 s + /// backoff running to t≈14 s, so the outage induced at t≈7 s cannot be answered before then. + /// A stood-down supervisor answers it one min-backoff (2 s) later. The 5 s window separates the + /// two by a clear margin; widening the burn only widens the separation. + /// + [Fact] + public async Task AfterACallerWinsTheRace_TheNextGenuineOutageStillRecoversAtMinBackoff() + { + using var broker = new MiniBroker(); + var opts = BrokerOpts(broker.Port) with { ReconnectMinBackoffSeconds = 2, ReconnectMaxBackoffSeconds = 30 }; + await using var conn = new MqttConnection(opts, driverId: "t", logger: null); + var resubscribes = 0; + conn.Reconnected += _ => { Interlocked.Increment(ref resubscribes); return Task.CompletedTask; }; + + await conn.ConnectAsync(CancellationToken.None); + broker.DropAll(); + (await WaitUntilAsync(() => conn.State == MqttConnectionState.Reconnecting, TimeSpan.FromSeconds(10))) + .ShouldBeTrue(); + await conn.ConnectAsync(CancellationToken.None); // caller wins the race + await Task.Delay(TimeSpan.FromSeconds(7), TestContext.Current.CancellationToken); // a broken loop climbs here + + broker.DropAll(); // the NEXT genuine outage + var recovered = await WaitUntilAsync(() => Volatile.Read(ref resubscribes) >= 1, TimeSpan.FromSeconds(5)); + + recovered.ShouldBeTrue("recovery took far longer than min backoff — the attempt counter never reset"); + conn.State.ShouldBe(MqttConnectionState.Connected); + } + + // --------------------------------------------------------------------------------- + // I2 / M3 — the Reconnected callback is bounded, and Connected means "connected AND subscribed" + // --------------------------------------------------------------------------------- + + [Fact] + public async Task FireReconnected_PassesTheSuppliedTokenToEverySubscriber() + { + await using var conn = new MqttConnection(LoopbackOpts(), driverId: "t", logger: null); + using var cts = new CancellationTokenSource(); + var seen = new List(); + conn.Reconnected += ct => { seen.Add(ct); return Task.CompletedTask; }; + conn.Reconnected += ct => { seen.Add(ct); return Task.CompletedTask; }; + + await conn.FireReconnectedAsync(cts.Token); + + seen.Count.ShouldBe(2); + seen.ShouldAllBe(t => t == cts.Token); + } + + /// + /// A re-subscribe that hangs — broker accepts SUBSCRIBE and never SUBACKs, the frozen-peer + /// shape again — must not park the supervisor. This subscriber deliberately ignores its + /// token, so only the fan-out's own ceiling can end the wait. + /// + /// + /// Asserted at the transport, not at dispose timing: with the ceiling, each hung re-subscribe + /// is abandoned, the session torn down, and another CONNECT reaches the broker; without it, + /// the supervisor is parked forever on the first one and the broker never sees another. A + /// dispose-elapsed assertion cannot tell those apart — DisposeAsync's own bounded join + /// returns promptly either way, leaving the supervisor alive behind it. + /// + [Fact] + public async Task ReSubscribeThatHangsIgnoringItsToken_DoesNotParkTheSupervisor() + { + using var broker = new MiniBroker(); + var opts = BrokerOpts(broker.Port) with + { + ConnectTimeoutSeconds = 2, ReconnectMinBackoffSeconds = 1, ReconnectMaxBackoffSeconds = 2, + }; + var conn = new MqttConnection(opts, driverId: "t", logger: null); + var entered = new TaskCompletionSource(); + conn.Reconnected += async _ => + { + entered.TrySetResult(); + await Task.Delay(Timeout.InfiniteTimeSpan); // no token — only the ceiling bounds this + }; + + await conn.ConnectAsync(CancellationToken.None); + broker.DropAll(); + await entered.Task.WaitAsync(TimeSpan.FromSeconds(30), TestContext.Current.CancellationToken); + + // The first reconnect is accept #2; a supervisor that recovered from the hang makes at least + // one more. A parked one never does. + (await WaitUntilAsync(() => broker.AcceptedCount >= 3, TimeSpan.FromSeconds(25))) + .ShouldBeTrue($"the supervisor is parked behind a hung re-subscribe ({broker.AcceptedCount} connects seen)"); + + var sw = Stopwatch.StartNew(); + await conn.DisposeAsync(); + sw.Stop(); + + sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(20), "dispose was parked behind a hung re-subscribe"); + conn.State.ShouldBe(MqttConnectionState.Disposed); + var acceptedAtDispose = broker.AcceptedCount; + await Task.Delay(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + broker.AcceptedCount.ShouldBe(acceptedAtDispose, "the supervisor outlived DisposeAsync"); + } + + /// + /// M3: on the reconnect path the socket comes up before the subscriptions do. Publishing + /// Connected in that window advertises exactly the healthy-looking-but-deaf condition + /// this type exists to prevent. It is also the documented case where State and + /// IsConnected legitimately disagree. + /// + [Fact] + public async Task Reconnect_DoesNotPublishConnectedUntilTheReSubscribeHasCompleted() + { + using var broker = new MiniBroker(); + await using var conn = new MqttConnection(BrokerOpts(broker.Port), driverId: "t", logger: null); + var inSubscriber = new TaskCompletionSource(); + var release = new TaskCompletionSource(); + conn.Reconnected += async _ => + { + inSubscriber.TrySetResult(); + await release.Task; + }; + + await conn.ConnectAsync(CancellationToken.None); + broker.DropAll(); + await inSubscriber.Task.WaitAsync(TimeSpan.FromSeconds(30), TestContext.Current.CancellationToken); + + conn.IsConnected.ShouldBeTrue("the socket is up"); + conn.State.ShouldBe(MqttConnectionState.Reconnecting, "…but it has no subscriptions yet"); + + release.SetResult(); + + (await WaitUntilAsync(() => conn.State == MqttConnectionState.Connected, TimeSpan.FromSeconds(10))) + .ShouldBeTrue(); + } + // --------------------------------------------------------------------------------- // helpers // --------------------------------------------------------------------------------- @@ -720,6 +949,44 @@ public sealed class MqttConnectionTests ReconnectMaxBackoffSeconds = 2, }; + /// + /// Captures formatted log messages. The supervisor's failed-attempt path is Debug-only and has + /// no other externally visible effect — MQTTnet refuses a connect-while-connected before + /// opening a socket, so the broker never sees it — which is precisely why the C1 defect was + /// silent. The log is the observable. + /// + private sealed class CapturingLogger : ILogger + { + private readonly List _messages = []; + + public int CountContaining(string fragment) + { + lock (_messages) + { + return _messages.Count(m => m.Contains(fragment, StringComparison.OrdinalIgnoreCase)); + } + } + + public IDisposable? BeginScope(TState state) + where TState : notnull + => null; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log( + LogLevel logLevel, + EventId eventId, + TState state, + Exception? exception, + Func formatter) + { + lock (_messages) + { + _messages.Add(formatter(state, exception)); + } + } + } + private static async Task WaitUntilAsync(Func condition, TimeSpan timeout) { var deadline = Stopwatch.StartNew();