dohertj2/lmxopcua
1
0
Fork 0
Code Issues 17 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(mqtt): hand-rolled reconnect loop with bounded backoff + resubscribe

Browse Source 
Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
2026-07-24 14:58:37 -04:00
parent cacc30a60d
commit 768fd87774
2 changed files with 1101 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttConnectionTests.cs
+513
View File
@@ -376,10 +376,366 @@ public sealed class MqttConnectionTests
conn.IsConnected.ShouldBeFalse();
}
// ---------------------------------------------------------------------------------
// Task 4 — NextBackoff (pure): bounded, monotone, cannot hot-loop, cannot overflow
// ---------------------------------------------------------------------------------
[Theory]
[InlineData(0, 1)]
[InlineData(1, 2)]
[InlineData(2, 4)]
[InlineData(3, 8)]
[InlineData(4, 16)]
[InlineData(5, 30)] // 32 → clamped
[InlineData(10, 30)]
public void NextBackoff_IsExponentialClampedToMax(int attempt, int expectedSeconds)
=> MqttConnection.NextBackoff(attempt, minSeconds: 1, maxSeconds: 30)
.ShouldBe(TimeSpan.FromSeconds(expectedSeconds));
/// <summary>
/// A shift-based implementation (<c>min &lt;&lt; attempt</c>) wraps: <c>1 &lt;&lt; 32</c> is
/// <c>1</c>, not 4294967296, so a broker that has been down for half an hour would suddenly
/// be hammered once a second. The calculator must saturate, not wrap.
/// </summary>
[Fact]
public void NextBackoff_HugeAttemptCount_SaturatesAtMax_NeverWrapsBackToMin()
{
foreach (var attempt in new[] { 31, 32, 33, 40, 62, 63, 64, 1_000, int.MaxValue })
{
MqttConnection.NextBackoff(attempt, minSeconds: 1, maxSeconds: 30)
.ShouldBe(TimeSpan.FromSeconds(30), $"attempt {attempt}");
}
}
[Fact]
public void NextBackoff_IsMonotoneNonDecreasing_AndAlwaysWithinMinMax()
{
var previous = TimeSpan.Zero;
for (var attempt = 0; attempt <= 64; attempt++)
{
var backoff = MqttConnection.NextBackoff(attempt, minSeconds: 2, maxSeconds: 30);
backoff.ShouldBeGreaterThanOrEqualTo(previous, $"attempt {attempt} went backwards");
backoff.ShouldBeGreaterThanOrEqualTo(TimeSpan.FromSeconds(2), $"attempt {attempt} below min");
backoff.ShouldBeLessThanOrEqualTo(TimeSpan.FromSeconds(30), $"attempt {attempt} above max");
previous = backoff;
}
}
/// <summary>
/// A misconfigured <c>max &lt; min</c> must not collapse the floor — the floor is the only
/// thing standing between a dead broker and a hot reconnect loop.
/// </summary>
[Fact]
public void NextBackoff_MaxBelowMin_HonoursMin_SoTheLoopCannotGoHot()
=> MqttConnection.NextBackoff(0, minSeconds: 10, maxSeconds: 5).ShouldBe(TimeSpan.FromSeconds(10));
[Fact]
public void NextBackoff_NonPositiveAttempt_IsTheFirstAttemptDelay()
{
MqttConnection.NextBackoff(0, minSeconds: 3, maxSeconds: 30).ShouldBe(TimeSpan.FromSeconds(3));
MqttConnection.NextBackoff(-5, minSeconds: 3, maxSeconds: 30).ShouldBe(TimeSpan.FromSeconds(3));
}
// ---------------------------------------------------------------------------------
// Task 4 — the Reconnected callback (the load-bearing re-subscribe hook)
// ---------------------------------------------------------------------------------
[Fact]
public async Task FireReconnected_InvokesEverySubscribedCallback()
{
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; };
await conn.FireReconnectedAsync();
first.ShouldBe(1);
second.ShouldBe(1);
}
/// <summary>
/// Awaiting a multicast <c>Func&lt;Task&gt;</c> directly returns only the LAST handler's task
/// and abandons the earlier ones — so one throwing subscriber would silently skip every
/// subscriber behind it. The fan-out must walk the invocation list and still surface the
/// failure, because a re-subscribe that failed is a connection that has gone dark.
/// </summary>
[Fact]
public async Task FireReconnected_OneHandlerThrows_StillInvokesTheRest_AndSurfacesTheFailure()
{
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; };
await Should.ThrowAsync<Exception>(async () => await conn.FireReconnectedAsync());
behindTheThrower.ShouldBe(1);
}
[Fact]
public async Task FireReconnected_NoSubscribers_IsANoOp()
{
await using var conn = new MqttConnection(LoopbackOpts(), driverId: "t", logger: null);
await Should.NotThrowAsync(async () => await conn.FireReconnectedAsync());
}
// ---------------------------------------------------------------------------------
// Task 4 — State + LastMessage
// ---------------------------------------------------------------------------------
[Fact]
public void State_AfterCtor_IsDisconnected()
=> new MqttConnection(LoopbackOpts(), driverId: "t", logger: null)
.State.ShouldBe(MqttConnectionState.Disconnected);
[Fact]
public async Task State_AfterDispose_IsDisposed()
{
var conn = new MqttConnection(LoopbackOpts(), driverId: "t", logger: null);
await conn.DisposeAsync();
conn.State.ShouldBe(MqttConnectionState.Disposed);
}
/// <summary>
/// Faulted is reserved for config that cannot succeed no matter how long we retry. A broker
/// that is merely down is <see cref="MqttConnectionState.Reconnecting"/>, indefinitely.
/// </summary>
[Fact]
public async Task State_UnrecoverableProtocolConfig_IsFaulted()
{
var opts = new MqttDriverOptions
{
Host = "127.0.0.1",
Port = 1,
UseTls = false,
ConnectTimeoutSeconds = 1,
ProtocolVersion = (MqttProtocolVersion)99,
};
await using var conn = new MqttConnection(opts, driverId: "t", logger: null);
await Should.ThrowAsync<ArgumentOutOfRangeException>(async () => await conn.ConnectAsync(CancellationToken.None));
conn.State.ShouldBe(MqttConnectionState.Faulted);
}
/// <summary>A broker that is simply unreachable is retryable, never <c>Faulted</c>.</summary>
[Fact]
public async Task State_UnreachableBroker_IsNotFaulted()
{
using var blackhole = new BlackholeBroker();
var opts = new MqttDriverOptions
{
Host = "127.0.0.1", Port = blackhole.Port, UseTls = false, ConnectTimeoutSeconds = 1,
};
await using var conn = new MqttConnection(opts, driverId: "t", logger: null);
await Should.ThrowAsync<TimeoutException>(async () => await conn.ConnectAsync(CancellationToken.None));
conn.State.ShouldNotBe(MqttConnectionState.Faulted);
}
[Fact]
public void LastMessage_BeforeAnyTraffic_IsNull()
{
var conn = new MqttConnection(LoopbackOpts(), driverId: "t", logger: null);
conn.LastMessageUtc.ShouldBeNull();
conn.LastMessageAge.ShouldBeNull();
}
/// <summary>
/// A failed FIRST connect is the caller's problem (driver-host resilience retries
/// <c>InitializeAsync</c>) — it must not silently leave a background reconnect worker
/// hammering an endpoint the caller has already given up on.
/// </summary>
[Fact]
public async Task ConnectAsync_InitialAttemptFails_LeavesNoBackgroundReconnectStorm()
{
using var blackhole = new BlackholeBroker();
var opts = new MqttDriverOptions
{
Host = "127.0.0.1",
Port = blackhole.Port,
UseTls = false,
ConnectTimeoutSeconds = 1,
ReconnectMinBackoffSeconds = 1,
ReconnectMaxBackoffSeconds = 1,
};
await using var conn = new MqttConnection(opts, driverId: "t", logger: null);
await Should.ThrowAsync<TimeoutException>(async () => await conn.ConnectAsync(CancellationToken.None));
var acceptedAfterTheAttempt = blackhole.AcceptedCount;
await Task.Delay(TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken);
blackhole.AcceptedCount.ShouldBe(acceptedAfterTheAttempt);
}
// ---------------------------------------------------------------------------------
// Task 4 — the reconnect supervisor, against a real (minimal) broker
// ---------------------------------------------------------------------------------
/// <summary>
/// THE load-bearing invariant. MQTT subscriptions do not survive a clean session, so a
/// reconnect that completes without firing <c>Reconnected</c> leaves a healthy-looking
/// connection that receives nothing, forever, with no error to show for it. Driven through
/// the real socket → real MQTTnet <c>DisconnectedAsync</c> → real re-CONNECT path, not a
/// test seam.
/// </summary>
[Fact]
public async Task BrokerDropsTheConnection_ReconnectsAndFiresReconnected()
{
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);
conn.State.ShouldBe(MqttConnectionState.Connected);
conn.IsConnected.ShouldBeTrue();
Volatile.Read(ref resubscribes).ShouldBe(0, "the initial connect is not a reconnect");
broker.DropAll();
(await WaitUntilAsync(() => Volatile.Read(ref resubscribes) >= 1, TimeSpan.FromSeconds(30)))
.ShouldBeTrue("the reconnect never fired Reconnected — subscriptions would be silently gone");
(await WaitUntilAsync(() => conn.IsConnected, TimeSpan.FromSeconds(10))).ShouldBeTrue();
conn.State.ShouldBe(MqttConnectionState.Connected);
broker.AcceptedCount.ShouldBeGreaterThanOrEqualTo(2);
}
/// <summary>
/// A broker that stays down keeps the supervisor in <c>Reconnecting</c> indefinitely (never
/// <c>Faulted</c>) — and the backoff keeps the retry rate low enough that a handful of
/// attempts, not hundreds, land in the observation window.
/// </summary>
[Fact]
public async Task BrokerStaysDown_RetriesUnderBackoff_WithoutHotLooping()
{
using var broker = new MiniBroker();
var opts = BrokerOpts(broker.Port) with
{
ConnectTimeoutSeconds = 1, ReconnectMinBackoffSeconds = 1, ReconnectMaxBackoffSeconds = 2,
};
await using var conn = new MqttConnection(opts, driverId: "t", logger: null);
await conn.ConnectAsync(CancellationToken.None);
var acceptedWhileHealthy = broker.AcceptedCount;
broker.Blackhole = true; // accepts TCP, never answers CONNACK again
broker.DropAll();
(await WaitUntilAsync(() => conn.State == MqttConnectionState.Reconnecting, TimeSpan.FromSeconds(10)))
.ShouldBeTrue();
await Task.Delay(TimeSpan.FromSeconds(8), TestContext.Current.CancellationToken);
conn.State.ShouldBe(MqttConnectionState.Reconnecting, "a down broker is retryable, not Faulted");
var retries = broker.AcceptedCount - acceptedWhileHealthy;
retries.ShouldBeGreaterThanOrEqualTo(1, "the supervisor gave up on a down broker");
retries.ShouldBeLessThan(15, $"backoff is not bounding the retry rate ({retries} attempts in ~8s)");
}
/// <summary>
/// Dispose during a reconnect backoff must stop the supervisor dead — no leaked task, no
/// later attempt resurrecting a connection the caller has torn down.
/// </summary>
[Fact]
public async Task Dispose_DuringReconnectBackoff_StopsTheSupervisor_NoFurtherAttempts()
{
using var broker = new MiniBroker();
var opts = BrokerOpts(broker.Port) with
{
ConnectTimeoutSeconds = 1, ReconnectMinBackoffSeconds = 1, ReconnectMaxBackoffSeconds = 2,
};
var conn = new MqttConnection(opts, driverId: "t", logger: null);
await conn.ConnectAsync(CancellationToken.None);
broker.Blackhole = true;
broker.DropAll();
(await WaitUntilAsync(() => conn.State == MqttConnectionState.Reconnecting, TimeSpan.FromSeconds(10)))
.ShouldBeTrue();
await conn.DisposeAsync();
var acceptedAtDispose = broker.AcceptedCount;
conn.State.ShouldBe(MqttConnectionState.Disposed);
conn.IsConnected.ShouldBeFalse();
await Task.Delay(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken);
broker.AcceptedCount.ShouldBe(acceptedAtDispose, "the supervisor outlived DisposeAsync");
}
/// <summary>
/// The T3 connect/dispose leak, reproduced at the exact interleaving the reviewer traced:
/// the client has already CONNECTED when dispose flags the instance and blocks on the
/// lifecycle gate this connect is holding — so dispose disposes nothing. Without the
/// post-connect re-check the client stays live with an open socket that no later
/// <c>DisposeAsync</c> can ever reach; the broker sees the leak as a connection that never
/// closes.
/// </summary>
[Fact]
public async Task DisposeRacingAnInFlightConnect_LeavesNoLiveBrokerConnection()
{
using var broker = new MiniBroker();
var conn = new MqttConnection(BrokerOpts(broker.Port), driverId: "t", logger: null);
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");
};
await Should.ThrowAsync<ObjectDisposedException>(async () => await conn.ConnectAsync(CancellationToken.None));
(await WaitUntilAsync(() => broker.LiveConnections == 0, TimeSpan.FromSeconds(15)))
.ShouldBeTrue($"a live broker connection leaked ({broker.LiveConnections} still open)");
conn.IsConnected.ShouldBeFalse();
}
// ---------------------------------------------------------------------------------
// helpers
// ---------------------------------------------------------------------------------
private static MqttDriverOptions LoopbackOpts()
=> new() { Host = "127.0.0.1", Port = 1, UseTls = false, ConnectTimeoutSeconds = 1 };
/// <summary>Options aimed at <see cref="MiniBroker"/>: plaintext v3.1.1, keep-alive parked out of the way.</summary>
private static MqttDriverOptions BrokerOpts(int port)
=> new()
{
Host = "127.0.0.1",
Port = port,
UseTls = false,
ProtocolVersion = MqttProtocolVersion.V311,
CleanSession = true,
KeepAliveSeconds = 300,
ConnectTimeoutSeconds = 5,
ReconnectMinBackoffSeconds = 1,
ReconnectMaxBackoffSeconds = 2,
};
private static async Task<bool> WaitUntilAsync(Func<bool> condition, TimeSpan timeout)
{
var deadline = Stopwatch.StartNew();
while (deadline.Elapsed < timeout)
{
if (condition())
{
return true;
}
await Task.Delay(25);
}
return condition();
}
private static MqttClientTlsOptions BuildTls(MqttDriverOptions opts)
=> MqttConnection.BuildClientOptions(opts, clientIdSuffix: null)
.ChannelOptions.ShouldBeOfType<MqttClientTcpOptions>().TlsOptions;
@@ -475,6 +831,7 @@ public sealed class MqttConnectionTests
private readonly List<TcpClient> _accepted = [];
private readonly CancellationTokenSource _cts = new();
private readonly TcpListener _listener;
private int _acceptedCount;
public BlackholeBroker()
{
@@ -486,6 +843,9 @@ public sealed class MqttConnectionTests
public int Port { get; }
/// <summary>Total TCP connections accepted — how a caller detects a background retry storm.</summary>
public int AcceptedCount => Volatile.Read(ref _acceptedCount);
public void Dispose()
{
_cts.Cancel();
@@ -510,6 +870,7 @@ public sealed class MqttConnectionTests
while (!_cts.IsCancellationRequested)
{
var client = await _listener.AcceptTcpClientAsync(_cts.Token);
Interlocked.Increment(ref _acceptedCount);
lock (_accepted)
{
_accepted.Add(client);
@@ -522,4 +883,156 @@ public sealed class MqttConnectionTests
}
}
}
/// <summary>
/// The smallest broker that can complete an MQTT 3.1.1 handshake: accept TCP, answer CONNECT
/// with a success CONNACK, answer PINGREQ with PINGRESP, and nothing else. That is enough for
/// MQTTnet to report a real connection — which is what makes a real drop
/// (<see cref="DropAll"/>) drive a real <c>DisconnectedAsync</c> and a real reconnect, rather
/// than a test seam standing in for one.
/// </summary>
private sealed class MiniBroker : IDisposable
{
private static readonly byte[] ConnAckAccepted = [0x20, 0x02, 0x00, 0x00];
private static readonly byte[] PingResp = [0xD0, 0x00];
private readonly CancellationTokenSource _cts = new();
private readonly HashSet<TcpClient> _live = [];
private readonly TcpListener _listener;
private int _acceptedCount;
public MiniBroker()
{
_listener = new TcpListener(IPAddress.Loopback, 0);
_listener.Start();
Port = ((IPEndPoint)_listener.LocalEndpoint).Port;
_ = Task.Run(AcceptLoopAsync);
}
public int Port { get; }
/// <summary>When set, TCP is still accepted but CONNECT is never answered (frozen peer).</summary>
public bool Blackhole { get; set; }
/// <summary>Total TCP connections accepted since construction — counts reconnect attempts.</summary>
public int AcceptedCount => Volatile.Read(ref _acceptedCount);
/// <summary>
/// Sockets the broker still has open. Drops to zero only once the peer actually closes,
/// so a leaked-but-live client keeps this above zero.
/// </summary>
public int LiveConnections
{
get
{
lock (_live)
{
return _live.Count;
}
}
}
/// <summary>Kills every established session — the "broker went away" event.</summary>
public void DropAll()
{
lock (_live)
{
foreach (var client in _live)
{
try
{
client.Close();
}
catch (Exception)
{
// Already gone.
}
}
_live.Clear();
}
}
public void Dispose()
{
_cts.Cancel();
_listener.Stop();
DropAll();
_cts.Dispose();
}
private async Task AcceptLoopAsync()
{
try
{
while (!_cts.IsCancellationRequested)
{
var client = await _listener.AcceptTcpClientAsync(_cts.Token);
Interlocked.Increment(ref _acceptedCount);
lock (_live)
{
_live.Add(client);
}
_ = Task.Run(() => ServeAsync(client));
}
}
catch (Exception)
{
// Listener stopped / token cancelled — the fixture is going away.
}
}
private async Task ServeAsync(TcpClient client)
{
var buffer = new byte[1024];
try
{
var stream = client.GetStream();
while (!_cts.IsCancellationRequested)
{
var read = await stream.ReadAsync(buffer, _cts.Token);
if (read == 0)
{
break; // peer closed — this is how a disposed client stops counting as live
}
if (Blackhole)
{
continue;
}
var packetType = buffer[0] & 0xF0;
if (packetType == 0x10)
{
await stream.WriteAsync(ConnAckAccepted, _cts.Token);
}
else if (packetType == 0xC0)
{
await stream.WriteAsync(PingResp, _cts.Token);
}
}
}
catch (Exception)
{
// Socket torn down by either side.
}
finally
{
lock (_live)
{
_live.Remove(client);
}
try
{
client.Dispose();
}
catch (Exception)
{
// Already gone.
}
}
}
}
}