fix(database): make DB updates atomic across device switches

androidApp/src/main/kotlin/org/meshtastic/app/MeshUtilApplication.kt

@@ -34,6 +34,7 @@ import kotlinx.coroutines.TimeoutCancellationException
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.withTimeout
 import org.koin.android.ext.android.get
 import org.koin.android.ext.koin.androidContext
@@ -135,11 +136,16 @@ open class MeshUtilApplication :
     }
 
     override fun onTerminate() {
-        // Shutdown managers (useful for Robolectric tests)
-        get<DatabaseManager>().close()
+        // Shutdown managers (useful for Robolectric tests).
+        // Non-blocking: cancelAndJoin inside runBlocking on the main thread can deadlock
+        // if any active coroutine is dispatching to Dispatchers.Main.
         applicationScope.cancel()
-        super.onTerminate()
-        org.koin.core.context.stopKoin()
+        try {
+            runBlocking { get<DatabaseManager>().close() }
+        } finally {
+            super.onTerminate()
+            org.koin.core.context.stopKoin()
+        }
     }
 
     private fun scheduleMeshLogCleanup() {

core/common/src/commonMain/kotlin/org/meshtastic/core/common/database/DatabaseManager.kt

@@ -33,17 +33,23 @@ interface DatabaseManager {
     suspend fun switchActiveDatabase(address: String?)
 
     /**
-     * Associates the currently-active connection with the physical device learned from the radio handshake, so the same
-     * device reached over a different transport (BLE / TCP / USB) shares one database. The first transport to report
-     * the device claims the active DB for it; a later transport folds its own data in and switches to the claimed DB.
-     * Idempotent once a device/transport pair has been unified.
+     * Associates the connection identified by [address] with the physical device learned from its radio handshake, so
+     * the same device reached over a different transport (BLE / TCP / USB) shares one database. Implementations must
+     * reject an address that is no longer active, preventing a stale identity emission from claiming or merging the
+     * newly selected transport's database. The first transport to report the device claims its active DB; a later
+     * transport folds its own data in and switches to the claimed DB. Idempotent once a device/transport pair has been
+     * unified.
      *
      * [deviceId] is the factory-burned hardware id from MyNodeInfo (null/blank when the platform or firmware doesn't
      * report one, or when lockdown zeroes it). It is the preferred claim key because it survives the node-number
      * changes firmware 2.8 introduced (`my_node_num = crc32(public_key)` renumbers on upgrade, erase, and re-key);
      * [nodeNum] remains the fallback claim for hardware without a device id and for claims made by older app versions.
+     *
+     * The caller must hold a transport-session lifecycle lease for the full call. [isSessionActive] must represent that
+     * lease, not merely whether new work may still be admitted, so association rollover cannot overlap the destination
+     * transaction commit.
      */
-    suspend fun associateDevice(nodeNum: Int, deviceId: String?)
+    suspend fun associateDevice(address: String, nodeNum: Int, deviceId: String?, isSessionActive: () -> Boolean)
 
     /** Returns true if a database exists for the given device address. */
     fun hasDatabaseFor(address: String?): Boolean

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

@@ -22,6 +22,7 @@ import org.meshtastic.core.repository.AdminPacketHandler
 import org.meshtastic.core.repository.MeshConfigFlowManager
 import org.meshtastic.core.repository.MeshConfigHandler
 import org.meshtastic.core.repository.NodeManager
+import org.meshtastic.core.repository.RadioSessionContext
 import org.meshtastic.core.repository.SessionManager
 import org.meshtastic.proto.AdminMessage
 import org.meshtastic.proto.MeshPacket
@@ -38,7 +39,7 @@ class AdminPacketHandlerImpl(
     private val sessionManager: SessionManager,
 ) : AdminPacketHandler {
 
-    override fun handleAdminMessage(packet: MeshPacket, myNodeNum: Int) {
+    override fun handleAdminMessage(packet: MeshPacket, myNodeNum: Int, session: RadioSessionContext) {
         val payload = packet.decoded?.payload ?: return
         val u = AdminMessage.ADAPTER.decode(payload)
         Logger.d { "Admin message from=${packet.from} fields=${u.summarize()}" }
@@ -52,22 +53,26 @@ class AdminPacketHandlerImpl(
         val fromNum = packet.from
         u.get_module_config_response?.let {
             if (fromNum == myNodeNum) {
-                configHandler.value.handleModuleConfig(it)
+                configHandler.value.handleModuleConfig(it, session)
             } else {
-                it.statusmessage?.node_status?.let { nodeManager.updateNodeStatus(fromNum, it) }
+                it.statusmessage?.node_status?.let { status ->
+                    nodeManager.updateNodeForSession(fromNum, session) { node ->
+                        node.copy(nodeStatus = status.takeIf(String::isNotEmpty))
+                    }
+                }
             }
         }
 
         if (fromNum == myNodeNum) {
-            u.get_config_response?.let { configHandler.value.handleDeviceConfig(it) }
-            u.get_channel_response?.let { configHandler.value.handleChannel(it) }
+            u.get_config_response?.let { configHandler.value.handleDeviceConfig(it, session) }
+            u.get_channel_response?.let { configHandler.value.handleChannel(it, session) }
         }
 
         u.get_device_metadata_response?.let {
             if (fromNum == myNodeNum) {
-                configFlowManager.value.handleLocalMetadata(it)
+                configFlowManager.value.handleLocalMetadata(it, session)
             } else {
-                nodeManager.insertMetadata(fromNum, it)
+                nodeManager.insertMetadata(fromNum, it, session)
             }
         }
     }

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

@@ -20,7 +20,6 @@ import co.touchlab.kermit.Logger
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.SupervisorJob
 import org.koin.core.annotation.Single
-import org.meshtastic.core.common.util.handledLaunch
 import org.meshtastic.core.common.util.ioDispatcher
 import org.meshtastic.core.model.util.isOtaStatusNotification
 import org.meshtastic.core.repository.FirmwareUpdateStatusRepository
@@ -32,6 +31,8 @@ import org.meshtastic.core.repository.MqttManager
 import org.meshtastic.core.repository.Notification
 import org.meshtastic.core.repository.NotificationManager
 import org.meshtastic.core.repository.PacketHandler
+import org.meshtastic.core.repository.RadioInterfaceService
+import org.meshtastic.core.repository.RadioSessionContext
 import org.meshtastic.core.repository.ServiceStateWriter
 import org.meshtastic.core.repository.XModemManager
 import org.meshtastic.core.resources.Res
@@ -46,6 +47,7 @@ import org.meshtastic.proto.ClientNotification
 import org.meshtastic.proto.FromRadio
 
 /** Implementation of [FromRadioPacketHandler] that dispatches [FromRadio] variants to specialized handlers. */
+@Suppress("LongParameterList")
 @Single
 class FromRadioPacketHandlerImpl(
     private val serviceStateWriter: ServiceStateWriter,
@@ -57,14 +59,15 @@ class FromRadioPacketHandlerImpl(
     private val notificationManager: NotificationManager,
     private val lockdownCoordinator: LockdownCoordinator,
     private val firmwareUpdateStatusRepository: FirmwareUpdateStatusRepository,
+    private val radioInterfaceService: RadioInterfaceService,
 ) : FromRadioPacketHandler {
 
     // Application-scoped coroutine context for suspend work (e.g. getStringSuspend).
     // This @Single lives for the entire app lifetime, so the SupervisorJob is never cancelled.
     private val scope = CoroutineScope(ioDispatcher + SupervisorJob())
 
     @Suppress("CyclomaticComplexMethod")
-    override fun handleFromRadio(proto: FromRadio) {
+    override fun handleFromRadio(proto: FromRadio, session: RadioSessionContext) {
         val myInfo = proto.my_info
         val metadata = proto.metadata
         val nodeInfo = proto.node_info
@@ -82,100 +85,130 @@ class FromRadioPacketHandlerImpl(
         val lockdownStatus = proto.lockdown_status
 
         when {
-            myInfo != null -> configFlowManager.value.handleMyInfo(myInfo)
+            myInfo != null -> configFlowManager.value.handleMyInfo(myInfo, session)
 
             // deviceuiConfig arrives immediately after my_info (STATE_SEND_UIDATA). It carries
             // the device's display, theme, node-filter, and other UI preferences.
-            deviceUIConfig != null -> configHandler.value.handleDeviceUIConfig(deviceUIConfig)
+            deviceUIConfig != null -> configHandler.value.handleDeviceUIConfig(deviceUIConfig, session)
 
-            metadata != null -> configFlowManager.value.handleLocalMetadata(metadata)
+            metadata != null -> configFlowManager.value.handleLocalMetadata(metadata, session)
 
             nodeInfo != null -> {
-                configFlowManager.value.handleNodeInfo(nodeInfo)
-                serviceStateWriter.setConnectionProgress("Nodes (${configFlowManager.value.newNodeCount})")
+                if (configFlowManager.value.handleNodeInfo(nodeInfo, session)) {
+                    runIfSessionActive(session, "node-list progress") {
+                        serviceStateWriter.setConnectionProgress("Nodes (${configFlowManager.value.newNodeCount})")
+                    }
+                }
             }
 
             configCompleteId != null -> {
-                configFlowManager.value.handleConfigComplete(configCompleteId)
-                lockdownCoordinator.onConfigComplete()
+                if (configFlowManager.value.handleConfigComplete(configCompleteId, session)) {
+                    runIfSessionActive(session, "lockdown config completion") { lockdownCoordinator.onConfigComplete() }
+                }
             }
 
-            mqttProxyMessage != null -> mqttManager.handleMqttProxyMessage(mqttProxyMessage)
+            mqttProxyMessage != null ->
+                runIfSessionActive(session, "MQTT proxy message") {
+                    mqttManager.handleMqttProxyMessage(mqttProxyMessage)
+                }
 
-            queueStatus != null -> packetHandler.handleQueueStatus(queueStatus)
+            queueStatus != null ->
+                runIfSessionActive(session, "queue status") { packetHandler.handleQueueStatus(queueStatus) }
 
-            config != null -> configHandler.value.handleDeviceConfig(config)
+            config != null -> configHandler.value.handleDeviceConfig(config, session)
 
-            moduleConfig != null -> configHandler.value.handleModuleConfig(moduleConfig)
+            moduleConfig != null -> configHandler.value.handleModuleConfig(moduleConfig, session)
 
-            channel != null -> configHandler.value.handleChannel(channel)
+            channel != null -> configHandler.value.handleChannel(channel, session)
 
-            fileInfo != null -> configFlowManager.value.handleFileInfo(fileInfo)
+            fileInfo != null -> configFlowManager.value.handleFileInfo(fileInfo, session)
 
             // region_presets arrives during the handshake (after metadata, before channels). It tells the client
             // which modem presets are legal per LoRa region. Absent on firmware < 2.8.
-            regionPresets != null -> configHandler.value.handleRegionPresets(regionPresets)
+            regionPresets != null -> configHandler.value.handleRegionPresets(regionPresets, session)
 
-            xmodemPacket != null -> xmodemManager.value.handleIncomingXModem(xmodemPacket)
+            xmodemPacket != null ->
+                runIfSessionActive(session, "XModem packet") { xmodemManager.value.handleIncomingXModem(xmodemPacket) }
 
-            lockdownStatus != null -> lockdownCoordinator.handleLockdownStatus(lockdownStatus)
+            lockdownStatus != null ->
+                runIfSessionActive(session, "lockdown status") {
+                    lockdownCoordinator.handleLockdownStatus(lockdownStatus)
+                }
 
-            clientNotification != null -> handleClientNotification(clientNotification)
+            clientNotification != null -> handleClientNotification(clientNotification, session)
 
             // Firmware rebooted without a transport-level disconnect (common on serial/TCP).
             // Re-handshake immediately rather than waiting for the 30s stall guard.
             proto.rebooted != null -> {
                 Logger.w { "Firmware rebooted (rebooted=${proto.rebooted}), re-initiating handshake" }
-                configFlowManager.value.triggerWantConfig()
+                configFlowManager.value.triggerWantConfig(session)
             }
         }
     }
 
-    private fun handleClientNotification(cn: ClientNotification) {
-        serviceStateWriter.setClientNotification(cn)
+    private fun runIfSessionActive(session: RadioSessionContext, operation: String, block: () -> Unit) {
+        if (!radioInterfaceService.runIfSessionActive(session, block)) {
+            Logger.d { "Discarding $operation from stale transport session" }
+        }
+    }
 
-        scope.handledLaunch {
-            if (cn.isOtaStatusNotification() && firmwareUpdateStatusRepository.status.value.isOtaUpdateActive) {
-                Logger.i { "OTA status ClientNotification received; skipping duplicate generic alert" }
-                return@handledLaunch
-            }
+    private fun handleClientNotification(cn: ClientNotification, session: RadioSessionContext) {
+        val admitted =
+            radioInterfaceService.runIfSessionActive(session) { serviceStateWriter.setClientNotification(cn) }
+        if (!admitted) {
+            Logger.d { "Discarding client notification from stale transport session" }
+            return
+        }
+        radioInterfaceService.launchSessionWork(
+            scope = scope,
+            session = session,
+            onRejected = { Logger.d { "Skipping client alert from stale transport session" } },
+        ) {
+            dispatchClientNotification(cn)
+        }
+    }
 
-            val inform = cn.key_verification_number_inform
-            val request = cn.key_verification_number_request
-            val verificationFinal = cn.key_verification_final
-            val (title, type) =
-                when {
-                    inform != null -> {
-                        Logger.i { "Key verification inform from ${inform.remote_longname}" }
-                        Pair(getStringSuspend(Res.string.key_verification_title), Notification.Type.Info)
-                    }
+    private suspend fun dispatchClientNotification(cn: ClientNotification) {
+        if (cn.isOtaStatusNotification() && firmwareUpdateStatusRepository.status.value.isOtaUpdateActive) {
+            Logger.i { "OTA status ClientNotification received; skipping duplicate generic alert" }
+            return
+        }
 
-                    request != null -> {
-                        Logger.i { "Key verification request from ${request.remote_longname}" }
-                        Pair(getStringSuspend(Res.string.key_verification_request_title), Notification.Type.Info)
-                    }
+        val inform = cn.key_verification_number_inform
+        val request = cn.key_verification_number_request
+        val verificationFinal = cn.key_verification_final
+        val (title, type) =
+            when {
+                inform != null -> {
+                    Logger.i { "Key verification inform received" }
+                    Pair(getStringSuspend(Res.string.key_verification_title), Notification.Type.Info)
+                }
 
-                    verificationFinal != null -> {
-                        Logger.i { "Key verification final from ${verificationFinal.remote_longname}" }
-                        Pair(getStringSuspend(Res.string.key_verification_final_title), Notification.Type.Info)
-                    }
+                request != null -> {
+                    Logger.i { "Key verification request received" }
+                    Pair(getStringSuspend(Res.string.key_verification_request_title), Notification.Type.Info)
+                }
 
-                    cn.duplicated_public_key != null -> {
-                        Logger.w { "Duplicated public key notification received" }
-                        Pair(getStringSuspend(Res.string.duplicated_public_key_title), Notification.Type.Warning)
-                    }
+                verificationFinal != null -> {
+                    Logger.i { "Key verification final received" }
+                    Pair(getStringSuspend(Res.string.key_verification_final_title), Notification.Type.Info)
+                }
 
-                    cn.low_entropy_key != null -> {
-                        Logger.w { "Low entropy key notification received" }
-                        Pair(getStringSuspend(Res.string.low_entropy_key_title), Notification.Type.Warning)
-                    }
+                cn.duplicated_public_key != null -> {
+                    Logger.w { "Duplicated public key notification received" }
+                    Pair(getStringSuspend(Res.string.duplicated_public_key_title), Notification.Type.Warning)
+                }
 
-                    else -> Pair(getStringSuspend(Res.string.client_notification), Notification.Type.Info)
+                cn.low_entropy_key != null -> {
+                    Logger.w { "Low entropy key notification received" }
+                    Pair(getStringSuspend(Res.string.low_entropy_key_title), Notification.Type.Warning)
                 }
 
-            notificationManager.dispatch(
-                Notification(title = title, type = type, message = cn.message, category = Notification.Category.Alert),
-            )
-        }
+                else -> Pair(getStringSuspend(Res.string.client_notification), Notification.Type.Info)
+            }
+
+        notificationManager.dispatch(
+            Notification(title = title, type = type, message = cn.message, category = Notification.Category.Alert),
+        )
     }
 }