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();