fix(messaging): time out orphaned "Sending..." messages into a retryable failure (#6630)

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

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

@@ -495,6 +495,7 @@ class MeshConnectionManagerImpl(
     }
 
     override fun onRadioConfigLoaded() {
+        packetHandler.rearmSendAckTimeouts()
         scope.handledLaunch {
             val queuedPackets = packetRepository.getQueuedPackets()
             queuedPackets.forEach { packet ->

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

@@ -48,8 +48,11 @@ import org.meshtastic.core.repository.RadioInterfaceService
 import org.meshtastic.proto.FromRadio
 import org.meshtastic.proto.MeshPacket
 import org.meshtastic.proto.QueueStatus
+import org.meshtastic.proto.Routing
 import org.meshtastic.proto.ToRadio
+import kotlin.time.Duration
 import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.Duration.Companion.minutes
 import kotlin.time.Duration.Companion.seconds
 import kotlin.uuid.Uuid
 
@@ -66,6 +69,19 @@ class PacketHandlerImpl(
     companion object {
         private val TIMEOUT = 5.seconds
 
+        /**
+         * Grace period after which a sent packet still [MessageStatus.ENROUTE] is stamped [Routing.Error.TIMEOUT]
+         * (retryable) instead of showing as sending forever. Generous — well past the radio's retransmit window — and
+         * matches iOS's sendAckTimeout so both apps time out alike.
+         */
+        internal val SEND_ACK_TIMEOUT = 5.minutes
+
+        /**
+         * Minimum re-arm delay on reconnect: the firmware's phone-queue backlog may still deliver the missing ACK/NAK
+         * just after the config handshake, so give it a moment before stamping a timeout.
+         */
+        internal val REARM_GRACE = 30.seconds
+
         /**
          * Firmware-internal `ErrorCode` (MeshTypes.h `ERRNO_SHOULD_RELEASE`) leaked into `QueueStatus.res`: "no error,
          * but the packet should still be released". Firmware 2.8+ returns it for self-addressed packets, which are
@@ -89,6 +105,9 @@ class PacketHandlerImpl(
     private val queueResponse = mutableMapOf<Int, CompletableDeferred<Boolean>>()
     private val routingResponse = mutableMapOf<Int, CompletableDeferred<Boolean>>()
 
+    private val timeoutMutex = Mutex()
+    private val sendAckTimeoutJobs = mutableMapOf<Int, Job>()
+
     override fun sendToRadio(p: ToRadio) {
         Logger.d { "Sending to radio ${p.toPIIString()}" }
         val b = p.encode()
@@ -258,12 +277,47 @@ class PacketHandlerImpl(
     private fun changeStatus(packetId: Int, m: MessageStatus) = scope.handledLaunch {
         if (packetId != 0) {
             getDataPacketById(packetId)?.let { p ->
-                if (p.status == m) return@handledLaunch
-                packetRepository.value.updateMessageStatus(p, m)
+                if (p.status != m) {
+                    packetRepository.value.updateMessageStatus(p, m)
+                }
+                if (m == MessageStatus.ENROUTE) {
+                    scheduleSendAckTimeout(packetId)
+                }
             }
         }
     }
 
+    override fun rearmSendAckTimeouts() {
+        scope.handledLaunch {
+            packetRepository.value.getEnroutePackets().forEach { p ->
+                val remaining = p.time + SEND_ACK_TIMEOUT.inWholeMilliseconds - nowMillis
+                scheduleSendAckTimeout(p.id, remaining.milliseconds.coerceAtLeast(REARM_GRACE))
+            }
+        }
+    }
+
+    /**
+     * A send whose routing ACK/NAK never reaches the app (typically because it was disconnected when the radio's
+     * response arrived) would stay ENROUTE — "Sending…" — forever. Stamp it as a retryable timeout instead; a late ACK
+     * still upgrades it via handleAckNak.
+     *
+     * One timer per packet: re-arming supersedes the pending one, so repeated reconnects cannot pile up timers for the
+     * same send. Timers deliberately survive a disconnect — the ack genuinely never arrived, and the resulting state is
+     * retryable — so a user who never reconnects still sees the send resolve.
+     */
+    private suspend fun scheduleSendAckTimeout(packetId: Int, delayFor: Duration = SEND_ACK_TIMEOUT) {
+        timeoutMutex.withLock {
+            sendAckTimeoutJobs.remove(packetId)?.cancel()
+            sendAckTimeoutJobs.values.removeAll { it.isCompleted }
+            sendAckTimeoutJobs[packetId] =
+                scope.handledLaunch {
+                    delay(delayFor)
+                    // Conditional in the DAO transaction: an ACK/NAK landing while this timer waited must win.
+                    packetRepository.value.timeOutEnroutePacket(packetId, Routing.Error.TIMEOUT.value)
+                }
+        }
+    }
+
     private suspend fun getDataPacketById(packetId: Int): DataPacket? = withTimeoutOrNull(1.seconds) {
         var dataPacket: DataPacket? = null
         while (dataPacket == null) {

core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/PacketRepositoryImpl.kt

@@ -113,6 +113,15 @@ class PacketRepositoryImpl(private val dbManager: DatabaseProvider, private val
         dbManager.currentDb.value.packetDao().getAllDataPackets().filter { it.status == MessageStatus.QUEUED }
     }
 
+    override suspend fun getEnroutePackets(): List<DataPacket> = withContext(dispatchers.io) {
+        dbManager.currentDb.value.packetDao().getAllDataPackets().filter { it.status == MessageStatus.ENROUTE }
+    }
+
+    // A null from withDb means no database was available, so nothing was timed out.
+    override suspend fun timeOutEnroutePacket(packetId: Int, routingError: Int): Boolean = withContext(dispatchers.io) {
+        dbManager.withDb { it.packetDao().timeOutEnroutePacket(packetId, routingError) } ?: false
+    }
+
     suspend fun insertRoomPacket(packet: RoomPacket) {
         withContext(dispatchers.io + NonCancellable) { dbManager.withDb { it.packetDao().insert(packet) } }
     }

core/data/src/commonTest/kotlin/org/meshtastic/core/data/manager/PacketHandlerImplTest.kt

@@ -20,9 +20,11 @@ import dev.mokkery.MockMode
 import dev.mokkery.answering.returns
 import dev.mokkery.answering.throws
 import dev.mokkery.every
+import dev.mokkery.everySuspend
 import dev.mokkery.matcher.any
 import dev.mokkery.mock
 import dev.mokkery.verify
+import dev.mokkery.verify.VerifyMode.Companion.exactly
 import dev.mokkery.verifySuspend
 import io.kotest.property.Arb
 import io.kotest.property.arbitrary.int
@@ -33,7 +35,10 @@ import kotlinx.coroutines.test.StandardTestDispatcher
 import kotlinx.coroutines.test.TestScope
 import kotlinx.coroutines.test.runTest
 import org.meshtastic.core.common.di.asServiceScope
+import org.meshtastic.core.common.util.nowMillis
 import org.meshtastic.core.model.ConnectionState
+import org.meshtastic.core.model.DataPacket
+import org.meshtastic.core.model.MessageStatus
 import org.meshtastic.core.repository.MeshLogRepository
 import org.meshtastic.core.repository.PacketRepository
 import org.meshtastic.core.repository.RadioInterfaceService
@@ -42,6 +47,7 @@ import org.meshtastic.proto.Data
 import org.meshtastic.proto.MeshPacket
 import org.meshtastic.proto.PortNum
 import org.meshtastic.proto.QueueStatus
+import org.meshtastic.proto.Routing
 import org.meshtastic.proto.ToRadio
 import kotlin.test.BeforeTest
 import kotlin.test.Test
@@ -49,6 +55,7 @@ import kotlin.test.assertEquals
 import kotlin.test.assertFalse
 import kotlin.test.assertNotNull
 import kotlin.test.assertTrue
+import kotlin.time.Duration.Companion.seconds
 
 class PacketHandlerImplTest {
 
@@ -275,4 +282,80 @@ class PacketHandlerImplTest {
 
         verifySuspend { meshLogRepository.insert(any()) }
     }
+
+    private fun enrouteDataPacket(id: Int, time: Long = 0L) =
+        DataPacket(to = "!12345678", bytes = null, dataType = 1, id = id, time = time, status = MessageStatus.ENROUTE)
+
+    @Test
+    fun `unacked ENROUTE send times out to a retryable ERROR TIMEOUT`() = runTest(testDispatcher) {
+        connectionStateFlow.value = ConnectionState.Connected
+        everySuspend { packetRepository.getPacketById(123) } returns enrouteDataPacket(123)
+
+        handler.sendToRadio(ToRadio(packet = MeshPacket(id = 123)))
+        testScheduler.advanceTimeBy(PacketHandlerImpl.SEND_ACK_TIMEOUT + 1.seconds)
+        testScheduler.runCurrent()
+
+        verifySuspend { packetRepository.timeOutEnroutePacket(123, Routing.Error.TIMEOUT.value) }
+    }
+
+    @Test
+    fun `the timeout never fires before its deadline`() = runTest(testDispatcher) {
+        connectionStateFlow.value = ConnectionState.Connected
+        everySuspend { packetRepository.getPacketById(124) } returns enrouteDataPacket(124)
+
+        handler.sendToRadio(ToRadio(packet = MeshPacket(id = 124)))
+        testScheduler.advanceTimeBy(PacketHandlerImpl.SEND_ACK_TIMEOUT - 1.seconds)
+        testScheduler.runCurrent()
+
+        verifySuspend(exactly(0)) { packetRepository.timeOutEnroutePacket(any(), any()) }
+    }
+
+    @Test
+    fun `rearm times out a stale persisted ENROUTE packet after the reconnect grace`() = runTest(testDispatcher) {
+        val stale = enrouteDataPacket(321, time = 0L)
+        everySuspend { packetRepository.getEnroutePackets() } returns listOf(stale)
+
+        handler.rearmSendAckTimeouts()
+        testScheduler.advanceTimeBy(PacketHandlerImpl.REARM_GRACE + 1.seconds)
+        testScheduler.runCurrent()
+
+        verifySuspend { packetRepository.timeOutEnroutePacket(321, Routing.Error.TIMEOUT.value) }
+    }
+
+    @Test
+    fun `rearm gives a fresh ENROUTE packet its full ack window`() = runTest(testDispatcher) {
+        val fresh = enrouteDataPacket(322, time = nowMillis)
+        everySuspend { packetRepository.getEnroutePackets() } returns listOf(fresh)
+
+        handler.rearmSendAckTimeouts()
+        testScheduler.advanceTimeBy(PacketHandlerImpl.REARM_GRACE + 1.seconds)
+        testScheduler.runCurrent()
+        verifySuspend(exactly(0)) { packetRepository.timeOutEnroutePacket(any(), any()) }
+
+        testScheduler.advanceTimeBy(PacketHandlerImpl.SEND_ACK_TIMEOUT + 1.seconds)
+        testScheduler.runCurrent()
+        verifySuspend { packetRepository.timeOutEnroutePacket(322, Routing.Error.TIMEOUT.value) }
+    }
+
+    @Test
+    fun `rearming supersedes the pending timer instead of stacking a second one`() = runTest(testDispatcher) {
+        // Repeated reconnects must not accumulate timers for the same send, and the superseded timer must not
+        // fire on its own original deadline.
+        connectionStateFlow.value = ConnectionState.Connected
+        val packet = enrouteDataPacket(325, time = nowMillis)
+        everySuspend { packetRepository.getPacketById(325) } returns packet
+        everySuspend { packetRepository.getEnroutePackets() } returns listOf(packet)
+
+        handler.sendToRadio(ToRadio(packet = MeshPacket(id = 325)))
+        testScheduler.runCurrent()
+        repeat(3) {
+            handler.rearmSendAckTimeouts()
+            testScheduler.runCurrent()
+        }
+
+        testScheduler.advanceTimeBy(PacketHandlerImpl.SEND_ACK_TIMEOUT * 2 + 1.seconds)
+        testScheduler.runCurrent()
+
+        verifySuspend(exactly(1)) { packetRepository.timeOutEnroutePacket(325, Routing.Error.TIMEOUT.value) }
+    }
 }

core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/PacketDao.kt

@@ -631,6 +631,22 @@ interface PacketDao {
             }
     }
 
+    /**
+     * Stamps [routingError] on a sent packet only while it is still [MessageStatus.ENROUTE]. The read and the write
+     * share one transaction so an ACK/NAK that resolves the packet concurrently is never overwritten by a send-ack
+     * timeout that sampled the row before it landed.
+     *
+     * @return true if a row was timed out.
+     */
+    @Transaction
+    suspend fun timeOutEnroutePacket(packetId: Int, routingError: Int): Boolean {
+        val enroute = findPacketsWithId(packetId).filter { it.data.status == MessageStatus.ENROUTE }
+        enroute.forEach { existing ->
+            update(existing.copy(data = existing.data.copy(status = MessageStatus.ERROR), routingError = routingError))
+        }
+        return enroute.isNotEmpty()
+    }
+
     /**
      * Atomically finds reactions by [replacement]'s packetId + userId + emoji and updates every ownership-scoped copy,
      * borrowing [myNodeNum][ReactionEntity.myNodeNum] from each existing row. No-op if no match is found.

core/database/src/commonTest/kotlin/org/meshtastic/core/database/dao/CommonPacketDaoTest.kt

@@ -29,6 +29,7 @@ import org.meshtastic.core.model.DataPacket
 import org.meshtastic.core.model.MessageStatus
 import org.meshtastic.core.model.NodeAddress
 import org.meshtastic.proto.PortNum
+import org.meshtastic.proto.Routing
 import kotlin.test.AfterTest
 import kotlin.test.Test
 import kotlin.test.assertEquals
@@ -152,6 +153,55 @@ abstract class CommonPacketDaoTest {
         assertEquals(MessageStatus.DELIVERED, updatedMessages.first { it.packet.data.id == 999 }.packet.data.status)
     }
 
+    private suspend fun insertSentPacket(packetId: Int, status: MessageStatus) {
+        packetDao.insert(
+            Packet(
+                uuid = 0L,
+                myNodeNum = myNodeNum,
+                port_num = PortNum.TEXT_MESSAGE_APP.value,
+                contact_key = "sent",
+                received_time = nowMillis,
+                read = true,
+                packetId = packetId,
+                data =
+                DataPacket(
+                    to = NodeAddress.ID_BROADCAST,
+                    bytes = "Sent".encodeToByteArray().toByteString(),
+                    dataType = PortNum.TEXT_MESSAGE_APP.value,
+                    id = packetId,
+                    status = status,
+                ),
+            ),
+        )
+    }
+
+    @Test
+    fun timeOutEnroutePacketFailsOnlyStillEnroutePackets() = runTest {
+        createDb()
+        insertSentPacket(packetId = 8001, status = MessageStatus.ENROUTE)
+
+        assertTrue(packetDao.timeOutEnroutePacket(8001, TIMEOUT_ROUTING_ERROR))
+
+        val timedOut = packetDao.getPacketByPacketId(8001)
+        assertNotNull(timedOut)
+        assertEquals(MessageStatus.ERROR, timedOut.packet.data.status)
+        assertEquals(TIMEOUT_ROUTING_ERROR, timedOut.packet.routingError)
+    }
+
+    @Test
+    fun timeOutEnroutePacketLeavesAnAlreadyResolvedPacketAlone() = runTest {
+        createDb()
+        // The ACK that resolved this packet landed while a send-ack timeout was pending; the timeout must not
+        // overwrite the delivered status it sampled before the ACK arrived.
+        insertSentPacket(packetId = 8002, status = MessageStatus.DELIVERED)
+
+        assertFalse(packetDao.timeOutEnroutePacket(8002, TIMEOUT_ROUTING_ERROR))
+
+        val untouched = packetDao.getPacketByPacketId(8002)
+        assertNotNull(untouched)
+        assertEquals(MessageStatus.DELIVERED, untouched.packet.data.status)
+    }
+
     @Test
     fun testGetQueuedPackets() = runTest {
         createDb()
@@ -323,5 +373,7 @@ abstract class CommonPacketDaoTest {
 
     companion object {
         private const val SAMPLE_SIZE = 10
+
+        private val TIMEOUT_ROUTING_ERROR = Routing.Error.TIMEOUT.value
     }
 }

core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/PacketHandler.kt

@@ -48,4 +48,11 @@ interface PacketHandler {
 
     /** Stops the packet queue. */
     fun stopPacketQueue()
+
+    /**
+     * Re-arms the send-ACK timeout for every persisted packet still awaiting its routing ACK/NAK, so sends orphaned by
+     * a disconnect or app restart become retryable instead of showing as sending forever. Call once per connection,
+     * after the radio config is loaded.
+     */
+    fun rearmSendAckTimeouts()
 }

core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/PacketRepository.kt

@@ -73,6 +73,17 @@ interface PacketRepository {
     /** Returns all packets currently queued for transmission. */
     suspend fun getQueuedPackets(): List<DataPacket>
 
+    /** Returns all sent packets still awaiting a routing ACK/NAK (status [MessageStatus.ENROUTE]). */
+    suspend fun getEnroutePackets(): List<DataPacket>
+
+    /**
+     * Atomically marks a still-[MessageStatus.ENROUTE] packet as failed with [routingError], leaving it untouched if an
+     * ACK/NAK already resolved it.
+     *
+     * @return true if the packet was timed out.
+     */
+    suspend fun timeOutEnroutePacket(packetId: Int, routingError: Int): Boolean
+
     /**
      * Persists a packet in the database.
      *