perf(mqtt): drop payload-less client-proxy downlinks before forwarding (#6171)

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

@@ -40,6 +40,7 @@ import org.koin.core.annotation.Single
 import org.meshtastic.core.common.util.safeCatching
 import org.meshtastic.core.di.CoroutineDispatchers
 import org.meshtastic.core.model.MqttJsonPayload
+import org.meshtastic.core.model.util.decodeOrNull
 import org.meshtastic.core.model.util.subscribeList
 import org.meshtastic.core.repository.NodeRepository
 import org.meshtastic.core.repository.RadioConfigRepository
@@ -56,6 +57,7 @@ import org.meshtastic.mqtt.transport.tcp.TcpTransportFactory
 import org.meshtastic.mqtt.transport.ws.WebSocketTransportFactory
 import org.meshtastic.proto.ModuleConfig
 import org.meshtastic.proto.MqttClientProxyMessage
+import org.meshtastic.proto.ServiceEnvelope
 import kotlin.concurrent.Volatile
 import kotlin.uuid.Uuid
 
@@ -162,7 +164,13 @@ class MQTTRepositoryImpl(
                     )
                 }
             }
-            add(Subscription("$rootTopic${DEFAULT_TOPIC_LEVEL}PKI/+", maxQos = QoS.AT_LEAST_ONCE, noLocal = true))
+            add(
+                Subscription(
+                    "$rootTopic$DEFAULT_TOPIC_LEVEL$PKI_CHANNEL_ID/+",
+                    maxQos = QoS.AT_LEAST_ONCE,
+                    noLocal = true,
+                ),
+            )
         }
 
         // Collect from the SharedFlow before connecting to avoid missing retained messages
@@ -248,6 +256,20 @@ class MQTTRepositoryImpl(
                 Logger.e(e) { "Failed to parse MQTT JSON: ${e.message}" }
             }
         } else {
+            // Drop provably-undeliverable downlink packets before spending BLE bandwidth on
+            // them. In client-proxy mode the public broker floods payload-less packet-header
+            // stubs the node can only decrypt-fail and discard; stopping them here saves BLE
+            // airtime, a decrypt attempt, and a node-side warning per packet.
+            if (isUndeliverableDownlink(payload, nodeRepository.myId.value)) {
+                Logger.d {
+                    if (buildConfigProvider.isDebug) {
+                        "MQTT downlink dropped (no payload): $topic"
+                    } else {
+                        "MQTT downlink dropped (no payload)"
+                    }
+                }
+                return
+            }
             trySend(MqttClientProxyMessage(topic = topic, data_ = payload.toByteString(), retained = msg.retain))
         }
     }
@@ -391,3 +413,35 @@ internal fun extractHost(address: String): String {
     // Remove path (if any), then remove port
     return afterScheme.substringBefore("/").substringBefore(":")
 }
+
+private const val PKI_CHANNEL_ID = "PKI"
+
+/**
+ * Returns true when a downlink [ServiceEnvelope] is provably un-usable by the node and should be dropped before
+ * forwarding over BLE (MQTT client-proxy mode).
+ *
+ * Fails open — returns false (forward) for anything it cannot positively prove undeliverable: unparseable bytes,
+ * traffic on the `PKI` channel (public-key-encrypted direct messages the node accepts without first decrypting), our
+ * own echoed-back packets (used as implicit ACKs), a packet-less envelope, or any packet that actually carries a
+ * payload. The only drop case is Tier 1: a [MeshPacket] with neither `decoded` nor `encrypted` set — no legitimate
+ * Meshtastic packet is payload-less.
+ *
+ * Extracted as an internal top-level function so [MQTTRepositoryImplTest] can exercise every branch without spinning up
+ * the full repository.
+ *
+ * @param payload the raw MQTT payload bytes (a serialized ServiceEnvelope on the binary topic).
+ * @param myId this device's node id in `!xxxxxxxx` hex form, or null before the local node loads.
+ */
+internal fun isUndeliverableDownlink(payload: ByteArray, myId: String?): Boolean {
+    // Decode without a logger: this is fail-open and, on a public broker, may see arbitrary bytes on the binary
+    // topic. Passing Logger would emit an error log per unparseable downlink — expensive noise for an expected case.
+    val envelope = ServiceEnvelope.ADAPTER.decodeOrNull(payload) ?: return false // fail open on garbage
+    // Guards — never drop: PKI-channel traffic (accepted without decryption) or our own echoed-back
+    // packets (used as implicit ACKs).
+    val isGuarded = envelope.channel_id == PKI_CHANNEL_ID || (myId != null && envelope.gateway_id == myId)
+    // Tier 1: a packet with neither `decoded` nor `encrypted` carries nothing the node can use.
+    // A packet-less envelope (packet == null) has nothing to prove, so it is treated as deliverable.
+    val packet = envelope.packet
+    val hasNoPayload = packet != null && packet.decoded == null && packet.encrypted == null
+    return hasNoPayload && !isGuarded
+}

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

@@ -51,12 +51,16 @@ import org.meshtastic.mqtt.ReasonCode
 import org.meshtastic.mqtt.packet.Subscription
 import org.meshtastic.proto.ChannelSet
 import org.meshtastic.proto.ChannelSettings
+import org.meshtastic.proto.Data
 import org.meshtastic.proto.LocalModuleConfig
+import org.meshtastic.proto.MeshPacket
 import org.meshtastic.proto.ModuleConfig
+import org.meshtastic.proto.ServiceEnvelope
 import kotlin.test.BeforeTest
 import kotlin.test.Test
 import kotlin.test.assertContentEquals
 import kotlin.test.assertEquals
+import kotlin.test.assertFalse
 import kotlin.test.assertIs
 import kotlin.test.assertNull
 import kotlin.test.assertTrue
@@ -427,6 +431,89 @@ class MQTTRepositoryImplTest {
 
     // endregion
 
+    // region isUndeliverableDownlink — Tier 1 drop filter for MQTT client-proxy downlink packets.
+
+    private fun envelopeBytes(
+        channelId: String = "LongFast",
+        gatewayId: String = "!aabbccdd",
+        packet: MeshPacket? = MeshPacket(),
+    ): ByteArray =
+        ServiceEnvelope.ADAPTER.encode(ServiceEnvelope(packet = packet, channel_id = channelId, gateway_id = gatewayId))
+
+    @Test
+    fun `payload-less packet is undeliverable`() {
+        // The observed LongFast flood: a packet with neither decoded nor encrypted set.
+        val bytes = envelopeBytes(packet = MeshPacket(from = 1, to = 2))
+        assertTrue(isUndeliverableDownlink(bytes, myId = "!12345678"))
+    }
+
+    @Test
+    fun `decoded packet is deliverable`() {
+        val bytes = envelopeBytes(packet = MeshPacket(decoded = Data()))
+        assertFalse(isUndeliverableDownlink(bytes, myId = "!12345678"))
+    }
+
+    @Test
+    fun `encrypted packet is deliverable`() {
+        val bytes = envelopeBytes(packet = MeshPacket(encrypted = byteArrayOf(1, 2, 3).toByteString()))
+        assertFalse(isUndeliverableDownlink(bytes, myId = "!12345678"))
+    }
+
+    @Test
+    fun `PKI payload-less packet is deliverable - guard`() {
+        val bytes = envelopeBytes(channelId = "PKI", packet = MeshPacket(from = 1))
+        assertFalse(isUndeliverableDownlink(bytes, myId = "!12345678"))
+    }
+
+    @Test
+    fun `own echo payload-less packet is deliverable - guard`() {
+        val bytes = envelopeBytes(gatewayId = "!12345678", packet = MeshPacket(from = 1))
+        assertFalse(isUndeliverableDownlink(bytes, myId = "!12345678"))
+    }
+
+    @Test
+    fun `a different gateway's payload-less packet is undeliverable`() {
+        val bytes = envelopeBytes(gatewayId = "!deadbeef", packet = MeshPacket(from = 1))
+        assertTrue(isUndeliverableDownlink(bytes, myId = "!12345678"))
+    }
+
+    @Test
+    fun `null myId does not suppress dropping`() {
+        val bytes = envelopeBytes(gatewayId = "!deadbeef", packet = MeshPacket(from = 1))
+        assertTrue(isUndeliverableDownlink(bytes, myId = null))
+    }
+
+    @Test
+    fun `packet-less envelope is deliverable - fail open`() {
+        val bytes = envelopeBytes(packet = null)
+        assertFalse(isUndeliverableDownlink(bytes, myId = "!12345678"))
+    }
+
+    @Test
+    fun `garbage bytes are deliverable - fail open`() {
+        // Non-protobuf bytes must never be dropped.
+        val bytes = byteArrayOf(0xFF.toByte(), 0xFF.toByte(), 0xFF.toByte(), 0x42)
+        assertFalse(isUndeliverableDownlink(bytes, myId = "!12345678"))
+    }
+
+    @Test
+    fun `payload-less downlink stubs are dropped before forwarding`() = runTest {
+        val harness = createHarness()
+        val stub = envelopeBytes(packet = MeshPacket(from = 1, to = 2)) // no payload → dropped
+        val real = envelopeBytes(packet = MeshPacket(decoded = Data())) // has payload → forwarded
+
+        val nextMessage = backgroundScope.async { harness.repository.proxyMessageFlow.first() }
+        runCurrent()
+        harness.client.emitMessage(MqttMessage(topic = "msh/2/e/alpha/node", payload = stub, retain = false))
+        harness.client.emitMessage(MqttMessage(topic = "msh/2/e/alpha/node", payload = real, retain = false))
+
+        // first() returns the first *forwarded* message; the stub was dropped, so it must be `real`.
+        val proxyMessage = nextMessage.await()
+        assertContentEquals(real, proxyMessage.data_?.toByteArray())
+    }
+
+    // endregion
+
     private fun TestScope.createHarness(
         radioConfigRepository: FakeRadioConfigRepository = defaultRadioConfigRepository(),
         nodeRepository: FakeNodeRepository = FakeNodeRepository().apply { setMyId("!12345678") },