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 1ed55064..ebb6a486 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs @@ -89,6 +89,43 @@ public enum MqttConnectionState Disposed = 4, } +/// +/// The broker completed the handshake and refused the CONNECT, returning a non-success +/// CONNACK reason code. +/// +/// +/// +/// This exists because MQTTnet 5 does not throw on an unsuccessful CONNACK — it returns +/// the outcome in and leaves the client +/// disconnected, and v5 removed v4's ThrowOnNonSuccessfulConnectResponse option, so +/// inspecting the result is the only way to see it. A caller that ignores the result gets a +/// connect that "succeeded" against a broker which rejected it: a driver configured with the +/// wrong broker password would report Healthy and never receive a single value. That is +/// exactly what the Task-13 live fixture caught. +/// +/// +/// distinguishes a rejection no amount of retrying can fix +/// (credentials, identity, protocol, an invalid Last-Will) from a genuinely transient refusal +/// (broker unavailable/busy, quota, rate limit). See +/// for the full mapping. +/// +/// +public sealed class MqttConnectRejectedException : Exception +{ + internal MqttConnectRejectedException(MqttClientConnectResultCode resultCode, bool isUnrecoverable, string message) + : base(message) + { + ResultCode = resultCode; + IsUnrecoverable = isUnrecoverable; + } + + /// Whether retrying against this broker with this configuration can never succeed. + public bool IsUnrecoverable { get; } + + /// The CONNACK reason code the broker returned. + public MqttClientConnectResultCode ResultCode { get; } +} + /// /// Owns the driver's single live MQTTnet-5 client: assembles the broker connection options /// from (endpoint, protocol version, session, credentials, @@ -450,6 +487,12 @@ public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport /// /// The connection was disposed, either before the attempt started or while it was in flight. /// + /// + /// The broker refused the CONNECT (non-success CONNACK — bad credentials, unsupported protocol + /// version, server unavailable …). MQTTnet reports this as a result, not an exception, + /// so this method inspects the CONNACK and raises it: a caller that treats "did not throw" as + /// "connected" would otherwise report a healthy session the broker never granted. + /// public async Task ConnectAsync(CancellationToken cancellationToken) { ObjectDisposedException.ThrowIf(Disposed, this); @@ -584,7 +627,15 @@ public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport return false; } - await client.ConnectAsync(clientOptions, linked.Token).ConfigureAwait(false); + var connack = await client.ConnectAsync(clientOptions, linked.Token).ConfigureAwait(false); + + // MQTTnet 5 does NOT throw for a broker that refused the CONNECT — it hands the reason + // back here and leaves the client disconnected. Skipping this check is how a wrong broker + // password produced a Healthy driver that never received a value (Task-13 live gate). + if (connack is not null && connack.ResultCode != MqttClientConnectResultCode.Success) + { + throw BuildRejection(connack, supervisorAttempt); + } if (AfterConnectHookForTests is { } hook) { @@ -618,6 +669,12 @@ public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport { throw; } + // A refused CONNECT is already fully classified — it must not be re-wrapped as a timeout or a + // cancellation just because a token happened to fire while the CONNACK was being read. + catch (MqttConnectRejectedException) + { + throw; + } catch (Exception ex) { throw Classify(ex, cancellationToken, deadline); @@ -628,6 +685,101 @@ public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport } } + /// + /// Turns a refused CONNACK into the thrown contract, setting + /// first when the refusal is unrecoverable — which is what stops the supervisor + /// ( returns on Faulted) from retrying every + /// backoff period against a broker that will never accept these credentials. + /// + private MqttConnectRejectedException BuildRejection(MqttClientConnectResult connack, bool supervisorAttempt) + { + var unrecoverable = IsUnrecoverableConnackRejection(connack.ResultCode); + var message = DescribeConnackRejection($"{_options.Host}:{_options.Port}", connack.ResultCode); + + if (unrecoverable) + { + SetState(MqttConnectionState.Faulted); + _logger?.LogError( + "MQTT driver '{DriverId}': {Message} Retrying cannot help; fix the configuration and redeploy.", + _driverId, + message); + } + else + { + // Retryable: leave the state alone — Disconnected before a first connect, Reconnecting + // under the supervisor — so the backoff loop keeps trying at its bounded rate. + _logger?.LogWarning( + "MQTT driver '{DriverId}': {Message} Treating as transient{Suffix}.", + _driverId, + message, + supervisorAttempt ? " and continuing to retry" : string.Empty); + } + + return new MqttConnectRejectedException(connack.ResultCode, unrecoverable, message); + } + + /// + /// Whether a refused CONNACK can never be fixed by retrying, and so must fault the connection + /// rather than leave it looping. + /// + /// + /// + /// Unrecoverable covers credentials/identity (BadUserNameOrPassword, + /// NotAuthorized, ClientIdentifierNotValid, Banned, + /// BadAuthenticationMethod), protocol mismatch (UnsupportedProtocolVersion, + /// MalformedPacket, ProtocolError, PacketTooLarge) and a CONNECT whose + /// Last-Will the broker will not accept (TopicNameInvalid, + /// PayloadFormatInvalid, RetainNotSupported, QoSNotSupported). Every + /// one is a property of the configuration we keep re-sending. + /// + /// + /// ServerMoved (0x9D) is unrecoverable and UseAnotherServer (0x9C) is not, + /// because MQTT 5 defines the first as "permanently use another server" and the second as + /// "temporarily use another server" — the spec, not a guess. + /// + /// + /// Everything else — including UnspecifiedError, ImplementationSpecificError + /// and any code a future broker invents — is treated as transient. That default is + /// deliberate: wrongly faulting a transient refusal takes a recoverable driver down until + /// someone redeploys, whereas wrongly retrying a permanent one costs a bounded, visibly + /// Reconnecting, backoff-paced attempt that names the reason code in the log. + /// + /// + internal static bool IsUnrecoverableConnackRejection(MqttClientConnectResultCode code) => code switch + { + MqttClientConnectResultCode.MalformedPacket + or MqttClientConnectResultCode.ProtocolError + or MqttClientConnectResultCode.UnsupportedProtocolVersion + or MqttClientConnectResultCode.ClientIdentifierNotValid + or MqttClientConnectResultCode.BadUserNameOrPassword + or MqttClientConnectResultCode.NotAuthorized + or MqttClientConnectResultCode.Banned + or MqttClientConnectResultCode.BadAuthenticationMethod + or MqttClientConnectResultCode.TopicNameInvalid + or MqttClientConnectResultCode.PacketTooLarge + or MqttClientConnectResultCode.PayloadFormatInvalid + or MqttClientConnectResultCode.RetainNotSupported + or MqttClientConnectResultCode.QoSNotSupported + or MqttClientConnectResultCode.ServerMoved => true, + _ => false, + }; + + /// + /// Renders a refused CONNACK as a targeted, credential-free operator message. Shared with + /// MqttDriverProbe so the AdminUI "Test connect" button and the running driver describe + /// the same rejection in the same words. + /// + internal static string DescribeConnackRejection(string target, MqttClientConnectResultCode code) => code switch + { + MqttClientConnectResultCode.NotAuthorized or MqttClientConnectResultCode.BadUserNameOrPassword + => $"Broker at {target} rejected the credentials ({code}).", + MqttClientConnectResultCode.UnsupportedProtocolVersion + => $"Broker at {target} does not support the configured protocol version ({code}).", + MqttClientConnectResultCode.Banned + => $"Broker at {target} has banned this client ({code}).", + _ => $"Broker at {target} refused CONNECT: {code}.", + }; + /// /// Starts the supervisor on the first successful caller connect. Always called while /// holding the lifecycle gate, so the ??= needs no further synchronisation. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriverProbe.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriverProbe.cs index 4a6e973a..d14886a4 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriverProbe.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriverProbe.cs @@ -152,16 +152,14 @@ public sealed class MqttDriverProbe : IDriverProbe /// Success — never an exception, see the type remarks) to a targeted, credential-free /// message. /// - private static string ClassifyRejectedConnack(string target, MqttClientConnectResultCode code) => code switch - { - MqttClientConnectResultCode.NotAuthorized or MqttClientConnectResultCode.BadUserNameOrPassword - => $"Broker at {target} rejected the credentials ({code}).", - MqttClientConnectResultCode.UnsupportedProtocolVersion - => $"Broker at {target} does not support the configured protocol version ({code}).", - MqttClientConnectResultCode.Banned - => $"Broker at {target} has banned this client ({code}).", - _ => $"Broker at {target} refused CONNECT: {code}.", - }; + /// + /// Delegates to so the "Test connect" + /// button and the running driver report an identical rejection in identical words — they now + /// both classify CONNACK, and disagreeing about it was the exact confusion the Task-13 live + /// gate surfaced. + /// + private static string ClassifyRejectedConnack(string target, MqttClientConnectResultCode code) + => MqttConnection.DescribeConnackRejection(target, code); /// /// Classifies a thrown connect failure. Precedence mirrors diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/PlainMqttLiveTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/PlainMqttLiveTests.cs index 6633f7c4..128ad67c 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/PlainMqttLiveTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/PlainMqttLiveTests.cs @@ -88,34 +88,30 @@ public sealed class PlainMqttLiveTests(MqttFixture fixture) /// allow_anonymous false, so if bad credentials produced a session the fixture would be /// misconfigured and "connected" would prove nothing about auth. /// - /// LIVE-GATE FINDING (2026-07-24) — the invariant asserted here is deliberately "no - /// session", not "ConnectAsync throws". Against real Mosquitto, - /// returns normally for a CONNECT the - /// broker rejects as not-authorized: MQTTnet 5 does not throw on an unsuccessful CONNACK, - /// and ConnectCoreAsync then unconditionally publishes - /// and starts the reconnect supervisor. - /// Measured at +300 ms with a wrong password: thrown=<none>, - /// IsConnected=False, State=Reconnecting (the broker's socket close had - /// fired DisconnectedAsync by then). Broker side: - /// Client … disconnected, not authorised. + /// LIVE-GATE FINDING (2026-07-24), now FIXED — this test was tightened accordingly. + /// Against real Mosquitto, originally + /// returned normally for a CONNECT the broker rejected as not-authorized: MQTTnet 5 + /// does not throw on an unsuccessful CONNACK, and ConnectCoreAsync then + /// unconditionally published and started the + /// reconnect supervisor. Measured at +300 ms with a wrong password: + /// thrown=<none>, IsConnected=False, State=Reconnecting. Broker + /// side: Client … disconnected, not authorised. The operator-visible consequence was + /// that InitializeAsync reported DriverState.Healthy / HostState.Running + /// off a connect that never authenticated, so a deployment carrying the wrong broker + /// password sealed green. /// /// - /// That is a driver-side defect, not a fixture one — InitializeAsync reports - /// DriverState.Healthy / HostState.Running off a connect that never - /// authenticated, so a deployment carrying the wrong broker password seals green and the - /// AdminUI "Test connect" probe is expected to agree. Fixing it belongs in the driver - /// (MqttConnection must inspect the connect result / opt into MQTTnet's - /// throw-on-unsuccessful-CONNACK), which is outside this task's file scope; it is reported - /// as a live-gate finding. - /// - /// - /// The test is written to hold either way: it records whether the call threw (as a - /// diagnostic, not an assertion) and asserts only the invariant that must be true under - /// both behaviours — no session is ever established. A driver fixed to throw still passes. + /// ConnectCoreAsync now inspects the CONNACK and raises + /// , so this asserts the strong form: the + /// rejection is surfaced, classified as unrecoverable, and lands the connection in + /// . The original weak invariant — polled over 5 s + /// so the optimistic Connected window cannot pass it for the wrong reason — is kept + /// underneath it, because "no session is ever established" is the property that actually + /// matters to a consumer and it must hold regardless of how the failure is reported. /// /// [Fact] - public async Task Tls_connect_with_wrong_password_never_establishes_a_session() + public async Task Tls_connect_with_wrong_password_is_rejected_and_surfaced() { SkipIfUnconfigured(); @@ -123,34 +119,32 @@ public sealed class PlainMqttLiveTests(MqttFixture fixture) var options = TlsOptions() with { Password = _fx.Password + "-definitely-wrong" }; await using var connection = new MqttConnection(options, "mqtt-live-badpass"); - Exception? thrown = null; - try - { - await connection.ConnectAsync(ct); - } - catch (Exception ex) - { - thrown = ex; - } + var rejection = await Should.ThrowAsync( + async () => await connection.ConnectAsync(ct)); - // Poll rather than sample once: the rejection arrives as a broker-initiated socket close a - // few milliseconds after ConnectAsync returns, so a single immediate read could catch the - // optimistic Connected state and pass for the wrong reason. Every sample must show no - // session — a correct password would make IsConnected true within the first sample. + // Mosquitto answers bad credentials with NotAuthorized (MQTT 5) / 0x05 (3.1.1); either maps to + // the same unrecoverable classification, which is what stops the reconnect supervisor. + rejection.IsUnrecoverable.ShouldBeTrue( + $"a broker that refused the credentials returned {rejection.ResultCode}, classified as retryable"); + rejection.Message.ShouldNotContain(options.Password ?? "", Case.Sensitive); + connection.State.ShouldBe(MqttConnectionState.Faulted); + + // Poll rather than sample once: a regression that went back to publishing an optimistic + // Connected would still read "not connected" in the first millisecond. Every sample must show + // no session — a correct password would make IsConnected true within the first sample. var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(5); while (DateTime.UtcNow < deadline) { connection.IsConnected.ShouldBeFalse( "a CONNECT the broker rejected as not-authorized must never yield a usable session"); + connection.State.ShouldNotBe( + MqttConnectionState.Connected, + "a rejected CONNECT must never advertise a healthy session"); await Task.Delay(100, ct); } - connection.State.ShouldNotBe( - MqttConnectionState.Connected, - "the settled state after a rejected CONNECT must not advertise a healthy session"); - TestContext.Current.SendDiagnosticMessage( - $"wrong-password connect: threw={thrown?.GetType().Name ?? ""}, " + $"wrong-password connect: {rejection.ResultCode} (unrecoverable={rejection.IsUnrecoverable}), " + $"IsConnected={connection.IsConnected}, State={connection.State}."); } 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 new file mode 100644 index 00000000..7950a240 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MiniBroker.cs @@ -0,0 +1,189 @@ +using System.Net; +using System.Net.Sockets; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests; + +/// +/// The smallest broker that can complete an MQTT 3.1.1 handshake: accept TCP, answer CONNECT with +/// a 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 () drive a real +/// DisconnectedAsync and a real reconnect, rather than a test seam standing in for one. +/// +/// +/// Shared across suites rather than nested in one, because both the connection-level and the +/// driver-level rejection tests need it: the driver-level one is the pin for the actual +/// operator-visible symptom (a rejected CONNECT must not report Healthy). +/// +internal sealed class MiniBroker : IDisposable +{ + /// CONNACK "Connection Accepted" (MQTT 3.1.1 return code 0). + internal const byte ReturnCodeAccepted = 0x00; + + /// CONNACK "Identifier rejected" — MQTTnet maps it to ClientIdentifierNotValid. + internal const byte ReturnCodeIdentifierRejected = 0x02; + + /// CONNACK "Server unavailable" — MQTTnet maps it to ServerUnavailable (transient). + internal const byte ReturnCodeServerUnavailable = 0x03; + + /// CONNACK "Bad user name or password" — MQTTnet maps it to BadUserNameOrPassword. + internal const byte ReturnCodeBadUserNameOrPassword = 0x04; + + /// CONNACK "Not authorized" — MQTTnet maps it to NotAuthorized (unrecoverable). + internal const byte ReturnCodeNotAuthorized = 0x05; + + private static readonly byte[] PingResp = [0xD0, 0x00]; + + private readonly CancellationTokenSource _cts = new(); + private readonly HashSet _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; } + + /// When set, TCP is still accepted but CONNECT is never answered (frozen peer). + public bool Blackhole { get; set; } + + /// + /// The MQTT 3.1.1 return code answered to every CONNECT. Anything but + /// is a broker that completes the handshake and then + /// refuses — the shape MQTTnet reports as a result rather than an exception, and the one + /// a real Mosquitto with the wrong password produces. + /// + public byte ConnAckReturnCode { get; set; } = ReturnCodeAccepted; + + /// Total TCP connections accepted since construction — counts reconnect attempts. + public int AcceptedCount => Volatile.Read(ref _acceptedCount); + + /// + /// 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. + /// + public int LiveConnections + { + get + { + lock (_live) + { + return _live.Count; + } + } + } + + /// Kills every established session — the "broker went away" event. + 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(new byte[] { 0x20, 0x02, 0x00, ConnAckReturnCode }, _cts.Token); + + // A real broker closes the socket straight after refusing, which is what drives + // MQTTnet's DisconnectedAsync. Mirroring it keeps the fake honest. + if (ConnAckReturnCode != ReturnCodeAccepted) + { + break; + } + } + 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. + } + } + } +} 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 a0986238..11287ce1 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 @@ -927,6 +927,168 @@ public sealed class MqttConnectionTests .ShouldBeTrue(); } + // --------------------------------------------------------------------------------- + // Refused CONNACK — MQTTnet 5 reports it as a RESULT, never an exception + // --------------------------------------------------------------------------------- + + /// + /// The Task-13 live-gate defect, reproduced offline: Mosquitto genuinely rejected a wrong + /// password (Client … disconnected, not authorised.) and ConnectAsync returned + /// normally, because MQTTnet 5 hands a refused CONNACK back in + /// MqttClientConnectResult.ResultCode and v5 removed v4's + /// ThrowOnNonSuccessfulConnectResponse option. + /// + [Fact] + public async Task ConnectAsync_BrokerRefusesTheCredentials_Throws_AndFaults() + { + using var broker = new MiniBroker { ConnAckReturnCode = MiniBroker.ReturnCodeNotAuthorized }; + await using var conn = new MqttConnection(BrokerOpts(broker.Port), driverId: "t", logger: null); + + var ex = await Should.ThrowAsync( + async () => await conn.ConnectAsync(CancellationToken.None)); + + ex.ResultCode.ShouldBe(MqttClientConnectResultCode.NotAuthorized); + ex.IsUnrecoverable.ShouldBeTrue(); + ex.Message.ShouldContain("rejected the credentials"); + conn.IsConnected.ShouldBeFalse(); + conn.State.ShouldBe(MqttConnectionState.Faulted); + } + + /// Credentials must never reach the message an operator (or a log sink) will see. + [Fact] + public async Task ConnectAsync_RefusedCredentials_DoesNotLeakThePasswordIntoTheRejection() + { + const string secret = "sup3r-s3cret-broker-pw"; + using var broker = new MiniBroker { ConnAckReturnCode = MiniBroker.ReturnCodeBadUserNameOrPassword }; + var opts = BrokerOpts(broker.Port) with { Username = "svc", Password = secret }; + await using var conn = new MqttConnection(opts, driverId: "t", logger: null); + + var ex = await Should.ThrowAsync( + async () => await conn.ConnectAsync(CancellationToken.None)); + + ex.ToString().ShouldNotContain(secret); + } + + /// + /// An unrecoverable rejection must not leave a supervisor grinding away every backoff period + /// against a broker that will never accept these credentials. + /// + [Fact] + public async Task ConnectAsync_RefusedCredentials_StartsNoReconnectSupervisor() + { + using var broker = new MiniBroker { ConnAckReturnCode = MiniBroker.ReturnCodeNotAuthorized }; + var opts = BrokerOpts(broker.Port) with { ReconnectMinBackoffSeconds = 1, ReconnectMaxBackoffSeconds = 1 }; + await using var conn = new MqttConnection(opts, driverId: "t", logger: null); + + await Should.ThrowAsync( + async () => await conn.ConnectAsync(CancellationToken.None)); + var acceptedAfterTheAttempt = broker.AcceptedCount; + + await Task.Delay(TimeSpan.FromSeconds(4), TestContext.Current.CancellationToken); + + broker.AcceptedCount.ShouldBe(acceptedAfterTheAttempt, "a hopeless rejection is being retried"); + conn.State.ShouldBe(MqttConnectionState.Faulted); + } + + /// + /// A genuinely transient refusal is NOT a fault. Faulting on ServerUnavailable would take + /// a recoverable driver down until someone redeployed it. + /// + [Fact] + public async Task ConnectAsync_TransientRefusal_Throws_ButStaysRetryable() + { + using var broker = new MiniBroker { ConnAckReturnCode = MiniBroker.ReturnCodeServerUnavailable }; + await using var conn = new MqttConnection(BrokerOpts(broker.Port), driverId: "t", logger: null); + + var ex = await Should.ThrowAsync( + async () => await conn.ConnectAsync(CancellationToken.None)); + + ex.ResultCode.ShouldBe(MqttClientConnectResultCode.ServerUnavailable); + ex.IsUnrecoverable.ShouldBeFalse(); + conn.State.ShouldNotBe(MqttConnectionState.Faulted); + } + + [Theory] + [InlineData(MqttClientConnectResultCode.NotAuthorized, true)] + [InlineData(MqttClientConnectResultCode.BadUserNameOrPassword, true)] + [InlineData(MqttClientConnectResultCode.ClientIdentifierNotValid, true)] + [InlineData(MqttClientConnectResultCode.UnsupportedProtocolVersion, true)] + [InlineData(MqttClientConnectResultCode.Banned, true)] + [InlineData(MqttClientConnectResultCode.BadAuthenticationMethod, true)] + [InlineData(MqttClientConnectResultCode.TopicNameInvalid, true)] + [InlineData(MqttClientConnectResultCode.QoSNotSupported, true)] + // MQTT 5: 0x9D "permanently use another server" vs 0x9C "temporarily use another server". + [InlineData(MqttClientConnectResultCode.ServerMoved, true)] + [InlineData(MqttClientConnectResultCode.UseAnotherServer, false)] + [InlineData(MqttClientConnectResultCode.ServerUnavailable, false)] + [InlineData(MqttClientConnectResultCode.ServerBusy, false)] + [InlineData(MqttClientConnectResultCode.QuotaExceeded, false)] + [InlineData(MqttClientConnectResultCode.ConnectionRateExceeded, false)] + // Carries no actionable information, so it must not be allowed to permanently fault a driver. + [InlineData(MqttClientConnectResultCode.UnspecifiedError, false)] + [InlineData(MqttClientConnectResultCode.ImplementationSpecificError, false)] + public void IsUnrecoverableConnackRejection_ClassifiesByWhetherRetryingCouldEverHelp( + MqttClientConnectResultCode code, + bool expected) + => MqttConnection.IsUnrecoverableConnackRejection(code).ShouldBe(expected); + + /// + /// A reason code this build has never seen must default to retryable — see the classification + /// remarks for why the ambiguous case fails toward availability, not toward Faulted. + /// + [Fact] + public void IsUnrecoverableConnackRejection_UnknownCode_DefaultsToRetryable() + => MqttConnection.IsUnrecoverableConnackRejection((MqttClientConnectResultCode)250).ShouldBeFalse(); + + /// + /// The supervisor's own attempts go through the same CONNACK check. A broker that starts + /// refusing mid-life (credentials revoked, client banned) must stop the loop, not feed it. + /// + [Fact] + public async Task Reconnect_BrokerStartsRefusing_FaultsAndStopsTheSupervisor() + { + using var broker = new MiniBroker(); + var opts = BrokerOpts(broker.Port) with { ReconnectMinBackoffSeconds = 1, ReconnectMaxBackoffSeconds = 1 }; + 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.ConnAckReturnCode = MiniBroker.ReturnCodeNotAuthorized; // credentials revoked + broker.DropAll(); + + (await WaitUntilAsync(() => conn.State == MqttConnectionState.Faulted, TimeSpan.FromSeconds(20))) + .ShouldBeTrue("a revoked-credentials reconnect must fault, not retry forever"); + var acceptedAtFault = broker.AcceptedCount; + await Task.Delay(TimeSpan.FromSeconds(4), TestContext.Current.CancellationToken); + + broker.AcceptedCount.ShouldBe(acceptedAtFault, "the supervisor kept retrying a hopeless rejection"); + Volatile.Read(ref resubscribes).ShouldBe(0, "nothing was ever re-subscribed on a refused session"); + } + + /// The mirror case: a transient refusal keeps the supervisor retrying under backoff. + [Fact] + public async Task Reconnect_TransientRefusal_KeepsRetryingUnderBackoff() + { + using var broker = new MiniBroker(); + var opts = BrokerOpts(broker.Port) with { 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.ConnAckReturnCode = MiniBroker.ReturnCodeServerUnavailable; + broker.DropAll(); + + (await WaitUntilAsync(() => broker.AcceptedCount >= acceptedWhileHealthy + 2, TimeSpan.FromSeconds(20))) + .ShouldBeTrue("the supervisor gave up on a transiently-unavailable broker"); + conn.State.ShouldBe(MqttConnectionState.Reconnecting, "a busy broker is retryable, not Faulted"); + + // And it recovers on its own once the broker starts accepting again. + broker.ConnAckReturnCode = MiniBroker.ReturnCodeAccepted; + (await WaitUntilAsync(() => conn.State == MqttConnectionState.Connected, TimeSpan.FromSeconds(20))) + .ShouldBeTrue(); + } + // --------------------------------------------------------------------------------- // helpers // --------------------------------------------------------------------------------- @@ -1150,156 +1312,4 @@ public sealed class MqttConnectionTests } } } - - /// - /// 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 - /// () drive a real DisconnectedAsync and a real reconnect, rather - /// than a test seam standing in for one. - /// - 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 _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; } - - /// When set, TCP is still accepted but CONNECT is never answered (frozen peer). - public bool Blackhole { get; set; } - - /// Total TCP connections accepted since construction — counts reconnect attempts. - public int AcceptedCount => Volatile.Read(ref _acceptedCount); - - /// - /// 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. - /// - public int LiveConnections - { - get - { - lock (_live) - { - return _live.Count; - } - } - } - - /// Kills every established session — the "broker went away" event. - 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. - } - } - } - } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverConnectRejectionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverConnectRejectionTests.cs new file mode 100644 index 00000000..8b7ddcca --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverConnectRejectionTests.cs @@ -0,0 +1,99 @@ +using System.Text.Json; +using MQTTnet; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests; + +/// +/// The operator-visible half of the refused-CONNACK defect. +/// pins that MqttConnection.ConnectAsync raises the rejection; this pins the thing an +/// operator actually sees — that a driver whose broker refused the CONNECT does not seal +/// green. +/// +/// +/// Worth its own suite because the two failures are independent. The connection could classify a +/// rejection perfectly and the driver could still report Healthy, simply by not letting the +/// failure reach InitializeCoreAsync's catch — which is exactly the shape the Task-13 live +/// fixture found (a wrong broker password produced DriverState.Healthy / +/// HostState.Running on a session that never authenticated). +/// +[Trait("Category", "Unit")] +public sealed class MqttDriverConnectRejectionTests +{ + /// + /// THE regression pin. A configured-wrong password must fail the deploy loudly, not produce a + /// healthy driver that will never receive a single value. + /// + [Fact] + public async Task InitializeAsync_BrokerRefusesTheCredentials_IsNotHealthy() + { + using var broker = new MiniBroker { ConnAckReturnCode = MiniBroker.ReturnCodeNotAuthorized }; + var options = BrokerOptions(broker.Port); + await using var driver = new MqttDriver(options, "d", null); + + var ex = await Should.ThrowAsync( + async () => await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken)); + + ex.ResultCode.ShouldBe(MqttClientConnectResultCode.NotAuthorized); + + var health = driver.GetHealth(); + health.State.ShouldNotBe(DriverState.Healthy, "a driver the broker refused must never seal green"); + health.State.ShouldBe(DriverState.Faulted); + health.LastError.ShouldNotBeNullOrWhiteSpace(); + health.LastError!.ShouldContain("rejected the credentials"); + } + + /// + /// The transient counterpart: still a failed initialize (the driver has no session), but it must + /// not be reported as the same unrecoverable shape — a broker restarting is not a misconfigured + /// deployment. + /// + [Fact] + public async Task InitializeAsync_BrokerTemporarilyUnavailable_FailsAsTransient() + { + using var broker = new MiniBroker { ConnAckReturnCode = MiniBroker.ReturnCodeServerUnavailable }; + var options = BrokerOptions(broker.Port); + await using var driver = new MqttDriver(options, "d", null); + + var ex = await Should.ThrowAsync( + async () => await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken)); + + ex.IsUnrecoverable.ShouldBeFalse(); + driver.GetHealth().State.ShouldNotBe(DriverState.Healthy); + } + + /// + /// The control. Without it, every assertion above would still hold if the driver simply never + /// went healthy against this fake broker — the suite would be green and prove nothing. + /// + [Fact] + public async Task InitializeAsync_BrokerAcceptsTheConnect_IsHealthy() + { + using var broker = new MiniBroker(); + var options = BrokerOptions(broker.Port); + await using var driver = new MqttDriver(options, "d", null); + + await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken); + + driver.GetHealth().State.ShouldBe(DriverState.Healthy); + } + + private static MqttDriverOptions BrokerOptions(int port) => new() + { + Mode = MqttMode.Plain, + Host = "127.0.0.1", + Port = port, + UseTls = false, + ProtocolVersion = MqttProtocolVersion.V311, + Username = "svc", + Password = "pw", + KeepAliveSeconds = 300, + ConnectTimeoutSeconds = 5, + ReconnectMinBackoffSeconds = 1, + ReconnectMaxBackoffSeconds = 2, + }; + + private static string Json(MqttDriverOptions options) => JsonSerializer.Serialize(options, MqttJson.Options); +}