From dda971b5ce81ed5dd9549dbaba1d8548bbfa4126 Mon Sep 17 00:00:00 2001 From: James Rich <2199651+jamesarich@users.noreply.github.com> Date: Tue, 11 Aug 2026 19:10:30 -0700 Subject: [PATCH] fix(messaging): time out orphaned "Sending..." messages into a retryable failure (#6630) Co-authored-by: Claude Fable 5 --- .../data/manager/MeshConnectionManagerImpl.kt | 1 + .../core/data/manager/PacketHandlerImpl.kt | 58 ++++++++++++- .../data/repository/PacketRepositoryImpl.kt | 9 ++ .../data/manager/PacketHandlerImplTest.kt | 83 +++++++++++++++++++ .../meshtastic/core/database/dao/PacketDao.kt | 16 ++++ .../core/database/dao/CommonPacketDaoTest.kt | 52 ++++++++++++ .../core/repository/PacketHandler.kt | 7 ++ .../core/repository/PacketRepository.kt | 11 +++ 8 files changed, 235 insertions(+), 2 deletions(-) diff --git a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshConnectionManagerImpl.kt b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshConnectionManagerImpl.kt index 709199b5a1..2cdcb9d721 100644 --- a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshConnectionManagerImpl.kt +++ b/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 -> diff --git a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/PacketHandlerImpl.kt b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/PacketHandlerImpl.kt index 3a32a5bd6f..017f6a4a9c 100644 --- a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/PacketHandlerImpl.kt +++ b/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>() private val routingResponse = mutableMapOf>() + private val timeoutMutex = Mutex() + private val sendAckTimeoutJobs = mutableMapOf() + 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) { diff --git a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/PacketRepositoryImpl.kt b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/PacketRepositoryImpl.kt index f08de3ea3c..9bb9c4affe 100644 --- a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/PacketRepositoryImpl.kt +++ b/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 = 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) } } } diff --git a/core/data/src/commonTest/kotlin/org/meshtastic/core/data/manager/PacketHandlerImplTest.kt b/core/data/src/commonTest/kotlin/org/meshtastic/core/data/manager/PacketHandlerImplTest.kt index bd97138817..41afc9443e 100644 --- a/core/data/src/commonTest/kotlin/org/meshtastic/core/data/manager/PacketHandlerImplTest.kt +++ b/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) } + } } diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/PacketDao.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/PacketDao.kt index b53c21a808..8570027e37 100644 --- a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/PacketDao.kt +++ b/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. diff --git a/core/database/src/commonTest/kotlin/org/meshtastic/core/database/dao/CommonPacketDaoTest.kt b/core/database/src/commonTest/kotlin/org/meshtastic/core/database/dao/CommonPacketDaoTest.kt index 083c36097e..f7c040e5fc 100644 --- a/core/database/src/commonTest/kotlin/org/meshtastic/core/database/dao/CommonPacketDaoTest.kt +++ b/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 } } diff --git a/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/PacketHandler.kt b/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/PacketHandler.kt index 1ede06c8ee..4a3f364044 100644 --- a/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/PacketHandler.kt +++ b/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() } diff --git a/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/PacketRepository.kt b/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/PacketRepository.kt index c9c6de0564..458c0725a6 100644 --- a/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/PacketRepository.kt +++ b/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 + /** Returns all sent packets still awaiting a routing ACK/NAK (status [MessageStatus.ENROUTE]). */ + suspend fun getEnroutePackets(): List + + /** + * 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. *