fix(mqtt): stop reporting transport failures as credential rejections (#6506)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

.skills/compose-ui/strings-index.txt

@@ -1057,6 +1057,10 @@ mpwrd_os
 mqtt
 mqtt_config
 mqtt_enabled
+mqtt_error_connection_lost
+mqtt_error_credentials_rejected
+mqtt_error_proxy_failed
+mqtt_error_rejected
 mqtt_probe_dns_failure
 mqtt_probe_other_failure
 mqtt_probe_rejected

core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MqttManagerImpl.kt

@@ -31,15 +31,25 @@ import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.flow.stateIn
 import org.koin.core.annotation.Named
 import org.koin.core.annotation.Single
+import org.meshtastic.core.common.util.safeCatchingAll
 import org.meshtastic.core.model.MqttConnectionState
 import org.meshtastic.core.model.MqttProbeStatus
 import org.meshtastic.core.network.repository.MQTTRepository
+import org.meshtastic.core.network.repository.MQTT_KEEPALIVE_SECONDS
+import org.meshtastic.core.network.repository.isCredentialRejection
 import org.meshtastic.core.network.repository.mqttTlsConfig
 import org.meshtastic.core.network.repository.resolveEndpoint
 import org.meshtastic.core.repository.MqttManager
 import org.meshtastic.core.repository.NodeRepository
 import org.meshtastic.core.repository.PacketHandler
 import org.meshtastic.core.repository.ServiceStateWriter
+import org.meshtastic.core.resources.Res
+import org.meshtastic.core.resources.getStringSuspend
+import org.meshtastic.core.resources.mqtt_error_connection_lost
+import org.meshtastic.core.resources.mqtt_error_credentials_rejected
+import org.meshtastic.core.resources.mqtt_error_proxy_failed
+import org.meshtastic.core.resources.mqtt_error_rejected
+import org.meshtastic.core.resources.unknown
 import org.meshtastic.mqtt.ConnectionState
 import org.meshtastic.mqtt.MqttClient
 import org.meshtastic.mqtt.MqttException
@@ -80,12 +90,26 @@ class MqttManagerImpl(
                     .onEach { message -> packetHandler.sendToRadio(ToRadio(mqttClientProxyMessage = message)) }
                     .catch { throwable ->
                         _proxyActive.value = false
+                        // safeCatchingAll swallows the Skiko ExceptionInInitializerError that
+                        // compose-resources raises on headless JVM tests; production resolves the
+                        // localized string and the error is still surfaced either way.
                         val message =
-                            when (throwable) {
-                                is MqttException.ConnectionRejected -> "MQTT: connection rejected (check credentials)"
-                                is MqttException.ConnectionLost -> "MQTT: connection lost"
-                                else -> "MQTT proxy failed: ${throwable.message}"
+                            safeCatchingAll {
+                                when {
+                                    throwable is MqttException.ConnectionRejected &&
+                                        throwable.isCredentialRejection() ->
+                                        getStringSuspend(Res.string.mqtt_error_credentials_rejected)
+
+                                    throwable is MqttException.ConnectionRejected ->
+                                        getStringSuspend(Res.string.mqtt_error_rejected, throwable.detail())
+
+                                    throwable is MqttException.ConnectionLost ->
+                                        getStringSuspend(Res.string.mqtt_error_connection_lost)
+
+                                    else -> getStringSuspend(Res.string.mqtt_error_proxy_failed, throwable.detail())
+                                }
                             }
+                                .getOrDefault("")
                         serviceStateWriter.setErrorMessage(text = message, severity = Severity.Warn)
                     }
                     .launchIn(scope)
@@ -144,6 +168,9 @@ class MqttManagerImpl(
                 // including its scoped private-CA trust hook — otherwise a probe would fail where a connect succeeds.
                 val tls = mqttTlsConfig()
                 transportFactory = TcpTransportFactory(tls) + WebSocketTransportFactory(tls)
+                // Mirror the live client's keepalive too: the library default is 0 (no keepalive),
+                // which some brokers reject — misleadingly, as CLIENT_IDENTIFIER_NOT_VALID.
+                keepAliveSeconds = MQTT_KEEPALIVE_SECONDS
                 // Per-connection random suffix: myId identifies the node (and is null →
                 // "unknown" before the node record loads), so two probes can collide on one
                 // client-id and evict each other (SESSION_TAKEN_OVER). See MQTTRepositoryImpl.
@@ -188,3 +215,6 @@ class MqttManagerImpl(
         is ProbeResult.Other -> MqttProbeStatus.Other(message = cause.message)
     }
 }
+
+/** Failure detail for a user-facing message; a throwable without a message still needs a placeholder to substitute. */
+private suspend fun Throwable.detail(): String = message ?: getStringSuspend(Res.string.unknown)

core/network/src/commonMain/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImpl.kt

@@ -52,6 +52,7 @@ import org.meshtastic.mqtt.MqttException
 import org.meshtastic.mqtt.MqttLogLevel
 import org.meshtastic.mqtt.MqttMessage
 import org.meshtastic.mqtt.QoS
+import org.meshtastic.mqtt.ReasonCode
 import org.meshtastic.mqtt.packet.Subscription
 import org.meshtastic.mqtt.plus
 import org.meshtastic.mqtt.transport.tcp.TcpTransportFactory
@@ -199,17 +200,17 @@ class MQTTRepositoryImpl(
                     }
                     Logger.i { "MQTT connected and subscribed" }
                 }
+                val failure = result.exceptionOrNull()
                 when {
                     result.isSuccess -> return@launch
 
-                    result.exceptionOrNull() is MqttException.ConnectionRejected -> {
-                        Logger.e(result.exceptionOrNull()) { "MQTT connection rejected (unrecoverable), stopping" }
-                        close(result.exceptionOrNull()!!)
+                    failure is MqttException.ConnectionRejected && failure.isCredentialRejection() -> {
+                        Logger.e(failure) { "MQTT connection rejected (unrecoverable), stopping" }
+                        close(failure)
                         return@launch
                     }
 
                     else -> {
-                        val failure = result.exceptionOrNull()
                         // Broker- and network-side failures are what this retry loop exists to absorb — an
                         // unreachable host, a TLS problem, a dropped connection, or a broker that violates the
                         // MQTT 5 spec (e.g. the topic-alias limit). None are defects in this app, and reporting
@@ -320,6 +321,24 @@ class MQTTRepositoryImpl(
     }
 }
 
+/**
+ * `true` only for CONNACK reason codes where retrying can never help — the broker examined our credentials or client
+ * identity and refused them. The library also wraps transport-level connect failures (timeout, TLS, socket EOF) as
+ * [MqttException.ConnectionRejected] with [ReasonCode.UNSPECIFIED_ERROR]; those are transient and must stay in the
+ * retry loop, not permanently stop the proxy with a "check credentials" dialog. Public (not internal) so
+ * `MqttManagerImpl` in `:core:data` can phrase its user-facing error from the same classification.
+ */
+fun MqttException.ConnectionRejected.isCredentialRejection(): Boolean = when (reasonCode) {
+    ReasonCode.BAD_USER_NAME_OR_PASSWORD,
+    ReasonCode.NOT_AUTHORIZED,
+    ReasonCode.BAD_AUTHENTICATION_METHOD,
+    ReasonCode.CLIENT_IDENTIFIER_NOT_VALID,
+    ReasonCode.BANNED,
+    -> true
+
+    else -> false
+}
+
 /**
  * `true` when an MQTT connect/subscribe failure is one the retry loop is designed to absorb, rather than a defect worth
  * reporting to Crashlytics/Datadog.
@@ -387,8 +406,9 @@ private fun defaultMqttClientFactory(setup: MqttClientSetup): MqttClientSession
         transportFactory = TcpTransportFactory(tls) + WebSocketTransportFactory(tls)
         keepAliveSeconds = MQTT_KEEPALIVE_SECONDS
         autoReconnect = true
-        username = setup.mqttConfig?.username
-        setup.mqttConfig?.password?.let { password(it) }
+        val (user, pass) = effectiveCredentials(setup.mqttConfig)
+        username = user
+        pass?.let { password(it) }
         logger = KermitMqttLogger()
         // WARN for production: the library emits endpoint addresses and topic strings at
         // INFO level. WARN messages (reconnect, timeout, retry) contain no PII and are
@@ -397,7 +417,9 @@ private fun defaultMqttClientFactory(setup: MqttClientSetup): MqttClientSession
     },
 )
 
-private const val MQTT_KEEPALIVE_SECONDS = 30
+// Public (not internal/private) so MqttManagerImpl's probe in :core:data can mirror the live client —
+// some brokers reject a keepalive-0 CONNECT (misleadingly, as CLIENT_IDENTIFIER_NOT_VALID).
+const val MQTT_KEEPALIVE_SECONDS = 30
 private const val MQTT_PORT_PLAIN = 1883
 private const val MQTT_PORT_TLS = 8883
 
@@ -426,6 +448,26 @@ fun resolveEndpoint(rawAddress: String, tlsEnabled: Boolean): MqttEndpoint = if
 
 private const val DEFAULT_PUBLIC_SERVER = "mqtt.meshtastic.org"
 
+// The public broker's well-known credentials, same values as the firmware's Default.h.
+private const val DEFAULT_MQTT_USERNAME = "meshdev"
+private const val DEFAULT_MQTT_PASSWORD = "large4cats"
+
+/**
+ * Mirrors the firmware's `PubSubConfig` rule: an empty `address` means "the public broker with its well-known
+ * credentials", substituting username and password together with the server — the stored username/password are ignored
+ * in that case, whatever they contain. Substituting only the address (as this repository does for the endpoint) while
+ * passing the stored empty credentials through makes the proxy connect anonymously, which the public broker rejects
+ * with BAD_USER_NAME_OR_PASSWORD — surfaced to the user as a bogus "check credentials" error on configs the firmware
+ * itself connects with happily. Empty-address configs occur in the wild: lockdown-enabled firmware hands
+ * unauthenticated clients a zeroed MQTTConfig.
+ */
+internal fun effectiveCredentials(config: ModuleConfig.MQTTConfig?): Pair<String?, String?> =
+    if (config?.address.isNullOrEmpty()) {
+        DEFAULT_MQTT_USERNAME to DEFAULT_MQTT_PASSWORD
+    } else {
+        config?.username to config?.password
+    }
+
 fun effectiveTlsEnabled(address: String, tlsEnabled: Boolean): Boolean =
     tlsEnabled || extractHost(address).equals(DEFAULT_PUBLIC_SERVER, ignoreCase = true)
 

core/network/src/commonTest/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImplTest.kt

@@ -37,6 +37,7 @@ import kotlinx.coroutines.test.runTest
 import kotlinx.serialization.json.Json
 import okio.ByteString.Companion.toByteString
 import org.meshtastic.core.common.BuildConfigProvider
+import org.meshtastic.core.common.util.safeCatching
 import org.meshtastic.core.di.CoroutineDispatchers
 import org.meshtastic.core.model.MqttJsonPayload
 import org.meshtastic.core.testing.FakeNodeRepository
@@ -184,6 +185,40 @@ class MQTTRepositoryImplTest {
 
     // endregion
 
+    // region effectiveCredentials — firmware-parity credential defaulting.
+
+    @Test
+    fun `empty address substitutes the public broker's well-known credentials`() {
+        // Mirrors firmware PubSubConfig: lockdown-redacted (zeroed) configs must not connect anonymously.
+        val creds = effectiveCredentials(ModuleConfig.MQTTConfig(address = "", username = "", password = ""))
+        assertEquals("meshdev" to "large4cats", creds)
+    }
+
+    @Test
+    fun `null config substitutes the public broker's well-known credentials`() {
+        assertEquals("meshdev" to "large4cats", effectiveCredentials(null))
+    }
+
+    @Test
+    fun `empty address ignores stored credentials entirely - firmware parity`() {
+        val creds = effectiveCredentials(ModuleConfig.MQTTConfig(address = "", username = "custom", password = "pw"))
+        assertEquals("meshdev" to "large4cats", creds)
+    }
+
+    @Test
+    fun `explicit address uses the stored credentials as-is`() {
+        val config = ModuleConfig.MQTTConfig(address = "broker.example.com", username = "user", password = "pass")
+        assertEquals("user" to "pass", effectiveCredentials(config))
+    }
+
+    @Test
+    fun `explicit default server address uses the stored credentials as-is - firmware parity`() {
+        val config = ModuleConfig.MQTTConfig(address = "mqtt.meshtastic.org", username = "user", password = "pass")
+        assertEquals("user" to "pass", effectiveCredentials(config))
+    }
+
+    // endregion
+
     // region extractHost — address canonicalization tests.
 
     @Test
@@ -337,6 +372,74 @@ class MQTTRepositoryImplTest {
         runCurrent()
     }
 
+    @Test
+    fun `transport failure wrapped as ConnectionRejected retries instead of stopping the proxy`() = runTest {
+        // The MQTT library wraps ANY connect failure — timeout, TLS, socket EOF — as
+        // ConnectionRejected with UNSPECIFIED_ERROR. Treating those as unrecoverable stopped
+        // the proxy permanently and showed users a bogus "check credentials" dialog.
+        val harness = createHarness()
+        harness.client.failConnectWith(
+            MqttException.ConnectionRejected(ReasonCode.UNSPECIFIED_ERROR, "Connection failed: Connection timed out"),
+        )
+
+        val collector = startProxyCollection(harness.repository)
+        runCurrent()
+        assertEquals(1, harness.client.connectCalls.size)
+
+        advanceTimeBy(1_000)
+        runCurrent()
+        assertEquals(2, harness.client.connectCalls.size)
+        assertEquals(1, harness.client.subscribeCalls.size)
+
+        collector.cancelAndJoin()
+        runCurrent()
+    }
+
+    @Test
+    fun `credential rejection stops the proxy permanently`() = runTest {
+        val harness = createHarness()
+        harness.client.failConnectWith(
+            MqttException.ConnectionRejected(ReasonCode.BAD_USER_NAME_OR_PASSWORD, "Connection refused"),
+        )
+
+        val outcome = backgroundScope.async { safeCatching { harness.repository.proxyMessageFlow.collect {} } }
+        runCurrent()
+        advanceTimeBy(60_000)
+        runCurrent()
+
+        assertEquals(1, harness.client.connectCalls.size)
+        assertIs<MqttException.ConnectionRejected>(outcome.await().exceptionOrNull())
+    }
+
+    @Test
+    fun `only credential and identity reason codes classify as credential rejections`() {
+        val fatal =
+            listOf(
+                ReasonCode.BAD_USER_NAME_OR_PASSWORD,
+                ReasonCode.NOT_AUTHORIZED,
+                ReasonCode.BAD_AUTHENTICATION_METHOD,
+                ReasonCode.CLIENT_IDENTIFIER_NOT_VALID,
+                ReasonCode.BANNED,
+            )
+        val transient =
+            listOf(
+                ReasonCode.UNSPECIFIED_ERROR,
+                ReasonCode.SERVER_UNAVAILABLE,
+                ReasonCode.SERVER_BUSY,
+                ReasonCode.CONNECTION_RATE_EXCEEDED,
+            )
+
+        fatal.forEach { code ->
+            assertTrue(MqttException.ConnectionRejected(code, "x").isCredentialRejection(), "expected fatal: $code")
+        }
+        transient.forEach { code ->
+            assertFalse(
+                MqttException.ConnectionRejected(code, "x").isCredentialRejection(),
+                "expected transient: $code",
+            )
+        }
+    }
+
     @Test
     fun `subscription failures trigger reconnect retry`() = runTest {
         val harness = createHarness()

core/resources/src/commonMain/composeResources/values/strings.xml

@@ -1090,6 +1090,10 @@
     <string name="mqtt">MQTT</string>
     <string name="mqtt_config">MQTT Config</string>
     <string name="mqtt_enabled">MQTT enabled</string>
+    <string name="mqtt_error_connection_lost">MQTT: connection lost</string>
+    <string name="mqtt_error_credentials_rejected">MQTT: connection rejected (check credentials)</string>
+    <string name="mqtt_error_proxy_failed">MQTT proxy failed: %1$s</string>
+    <string name="mqtt_error_rejected">MQTT: connection rejected: %1$s</string>
     <string name="mqtt_probe_dns_failure">Host not found</string>
     <string name="mqtt_probe_other_failure">Connection failed</string>
     <string name="mqtt_probe_rejected">Broker rejected: %1$s</string>