fix(database): route all one-shot DB writes through the merge drain barrier (#6233)

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

core/data/src/commonMain/kotlin/org/meshtastic/core/data/datasource/EventFirmwareEditionLocalDataSource.kt

@@ -27,20 +27,23 @@ class EventFirmwareEditionLocalDataSource(
     private val dbManager: DatabaseProvider,
     private val dispatchers: CoroutineDispatchers,
 ) {
+    // Reads may use the direct accessor; writes go through withDb so they register with the cross-transport merge
+    // drain barrier (see DatabaseProvider).
     private val dao
         get() = dbManager.currentDb.value.eventFirmwareEditionDao()
 
     suspend fun getByEdition(edition: String): EventFirmwareEditionEntity? =
         withContext(dispatchers.io) { dao.getByEdition(edition) }
 
-    suspend fun upsertAll(editions: List<EventFirmwareEditionEntity>) =
-        withContext(dispatchers.io) { dao.upsertAll(editions) }
+    suspend fun upsertAll(editions: List<EventFirmwareEditionEntity>) {
+        withContext(dispatchers.io) { dbManager.withDb { it.eventFirmwareEditionDao().upsertAll(editions) } }
+    }
 
     // No-op on empty: `NOT IN ()` is always true in SQLite, so forwarding an empty list would wipe the whole table.
     // Callers only prune against a non-empty upstream payload; an empty list means "keep everything".
     suspend fun deleteNotIn(keep: List<String>) {
         if (keep.isEmpty()) return
-        withContext(dispatchers.io) { dao.deleteNotIn(keep) }
+        withContext(dispatchers.io) { dbManager.withDb { it.eventFirmwareEditionDao().deleteNotIn(keep) } }
     }
 
     suspend fun count(): Int = withContext(dispatchers.io) { dao.count() }

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

@@ -153,51 +153,66 @@ open class MeshLogRepositoryImpl(
         .mapLatest { list -> list.map { it.asExternalModel() }.firstOrNull { it.myNodeInfo != null }?.myNodeInfo }
         .flowOn(dispatchers.io)
 
+    // Writes go through withDb so they register with the cross-transport merge drain barrier (see DatabaseProvider).
+
     /** Persists a new log entry to the database if logging is enabled in preferences. */
     override suspend fun insert(log: MeshLog) = withContext(dispatchers.io) {
         if (!meshLogPrefs.loggingEnabled.value) return@withContext
-        dbManager.currentDb.value.meshLogDao().insert(log.asEntity())
+        dbManager.withDb { it.meshLogDao().insert(log.asEntity()) }
+        Unit
     }
 
     /** Clears all logs from the database. */
-    override suspend fun deleteAll() =
-        withContext(dispatchers.io) { dbManager.currentDb.value.meshLogDao().deleteAll() }
+    override suspend fun deleteAll() {
+        withContext(dispatchers.io) { dbManager.withDb { it.meshLogDao().deleteAll() } }
+    }
 
     /** Deletes a specific log entry by its [uuid]. */
-    override suspend fun deleteLog(uuid: String) =
-        withContext(dispatchers.io) { dbManager.currentDb.value.meshLogDao().deleteLog(uuid) }
+    override suspend fun deleteLog(uuid: String) {
+        withContext(dispatchers.io) { dbManager.withDb { it.meshLogDao().deleteLog(uuid) } }
+    }
 
     /** Deletes all logs associated with a specific [nodeNum] and [portNum]. */
     override suspend fun deleteLogs(nodeNum: Int, portNum: Int) = withContext(dispatchers.io) {
         val myNodeNum = nodeInfoReadDataSource.myNodeInfoFlow().firstOrNull()?.myNodeNum
         val logId = if (nodeNum == myNodeNum) MeshLog.NODE_NUM_LOCAL else nodeNum
-        dbManager.currentDb.value.meshLogDao().deleteLogs(logId, portNum)
+        dbManager.withDb { it.meshLogDao().deleteLogs(logId, portNum) }
+        Unit
     }
 
     /** Deletes only local stats telemetry logs for [nodeNum], preserving other telemetry types. */
     override suspend fun deleteLocalStatsLogs(nodeNum: Int) = withContext(dispatchers.io) {
         val myNodeNum = nodeInfoReadDataSource.myNodeInfoFlow().firstOrNull()?.myNodeNum
         val logId = if (nodeNum == myNodeNum) MeshLog.NODE_NUM_LOCAL else nodeNum
-        val dao = dbManager.currentDb.value.meshLogDao()
-        val localStatsLogs =
-            dao.getLogsFrom(logId, PortNum.TELEMETRY_APP.value, Int.MAX_VALUE)
-                .firstOrNull()
-                .orEmpty()
-                .map { it.asExternalModel() }
-                .filter { parseTelemetryLog(it)?.local_stats != null }
+        // One withDb block: the query and the chunked deletes all land on the same DB instance (previously each
+        // chunk
+        // re-fetched the DAO to survive a mid-op switch; capturing one instance is the more correct behavior, and
+        // the
+        // drain barrier now covers the whole read-modify-delete against a concurrent merge).
+        dbManager.withDb { db ->
+            val dao = db.meshLogDao()
+            val localStatsLogIds =
+                dao.getLogsFrom(logId, PortNum.TELEMETRY_APP.value, Int.MAX_VALUE)
+                    .firstOrNull()
+                    .orEmpty()
+                    .map { it.asExternalModel() }
+                    .filter { parseTelemetryLog(it)?.local_stats != null }
+                    .map { it.uuid }
 
-        val localStatsLogIds = localStatsLogs.map { it.uuid }
-        // Chunk to stay under SQLite's bind-variable limit; re-fetch DAO per chunk if the active DB switches.
-        for (chunk in localStatsLogIds.chunked(DELETE_CHUNK_SIZE)) {
-            dbManager.currentDb.value.meshLogDao().deleteLogsByUuid(chunk)
+            // Chunk to stay under SQLite's bind-variable limit.
+            for (chunk in localStatsLogIds.chunked(DELETE_CHUNK_SIZE)) {
+                dao.deleteLogsByUuid(chunk)
+            }
         }
+        Unit
     }
 
     /** Prunes the log database based on the configured [retentionDays]. */
     @Suppress("MagicNumber")
     override suspend fun deleteLogsOlderThan(retentionDays: Int) = withContext(dispatchers.io) {
         val cutoffTime = nowMillis - (retentionDays.toLong() * 24 * 60 * 60 * 1000)
-        dbManager.currentDb.value.meshLogDao().deleteOlderThan(cutoffTime)
+        dbManager.withDb { it.meshLogDao().deleteOlderThan(cutoffTime) }
+        Unit
     }
 
     companion object {

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

@@ -31,6 +31,7 @@ import okio.ByteString.Companion.toByteString
 import org.koin.core.annotation.Single
 import org.meshtastic.core.database.DatabaseProvider
 import org.meshtastic.core.database.dao.NodeInfoDao
+import org.meshtastic.core.database.dao.PacketDao
 import org.meshtastic.core.database.entity.PacketEntity
 import org.meshtastic.core.database.entity.toReaction
 import org.meshtastic.core.di.CoroutineDispatchers
@@ -91,36 +92,46 @@ class PacketRepositoryImpl(private val dbManager: DatabaseProvider, private val
     override fun getUnreadCountTotal(): Flow<Int> =
         dbManager.currentDb.flatMapLatest { db -> db.packetDao().getUnreadCountTotal() }
 
-    override suspend fun clearUnreadCount(contact: String, timestamp: Long) =
+    // One-shot writes go through withDb so they register with the cross-transport merge drain barrier and pick up
+    // its closed-pool retry (see DatabaseProvider). Reads and Flow/Paging factories stay on currentDb by design.
+
+    override suspend fun clearUnreadCount(contact: String, timestamp: Long) {
         withContext(dispatchers.io + NonCancellable) {
-            dbManager.currentDb.value.packetDao().clearUnreadCount(contact, timestamp)
+            dbManager.withDb { it.packetDao().clearUnreadCount(contact, timestamp) }
         }
+    }
 
-    override suspend fun clearAllUnreadCounts() =
-        withContext(dispatchers.io + NonCancellable) { dbManager.currentDb.value.packetDao().clearAllUnreadCounts() }
+    override suspend fun clearAllUnreadCounts() {
+        withContext(dispatchers.io + NonCancellable) { dbManager.withDb { it.packetDao().clearAllUnreadCounts() } }
+    }
 
-    override suspend fun updateLastReadMessage(contact: String, messageUuid: Long, lastReadTimestamp: Long) =
+    override suspend fun updateLastReadMessage(contact: String, messageUuid: Long, lastReadTimestamp: Long) {
         withContext(dispatchers.io + NonCancellable) {
-            val dao = dbManager.currentDb.value.packetDao()
-            val current = dao.getContactSettings(contact)
-            val existingTimestamp = current?.lastReadMessageTimestamp ?: Long.MIN_VALUE
-            if (lastReadTimestamp <= existingTimestamp) {
-                return@withContext
+            // Read-modify-write in one withDb block so both statements land on the same DB instance.
+            dbManager.withDb { db ->
+                val dao = db.packetDao()
+                val current = dao.getContactSettings(contact)
+                val existingTimestamp = current?.lastReadMessageTimestamp ?: Long.MIN_VALUE
+                if (lastReadTimestamp <= existingTimestamp) {
+                    return@withDb
+                }
+                val updated =
+                    (current ?: ContactSettingsEntity(contact_key = contact)).copy(
+                        lastReadMessageUuid = messageUuid,
+                        lastReadMessageTimestamp = lastReadTimestamp,
+                    )
+                dao.upsertContactSettings(listOf(updated))
             }
-            val updated =
-                (current ?: ContactSettingsEntity(contact_key = contact)).copy(
-                    lastReadMessageUuid = messageUuid,
-                    lastReadMessageTimestamp = lastReadTimestamp,
-                )
-            dao.upsertContactSettings(listOf(updated))
         }
+    }
 
     override suspend fun getQueuedPackets(): List<DataPacket> = withContext(dispatchers.io) {
         dbManager.currentDb.value.packetDao().getAllDataPackets().filter { it.status == MessageStatus.QUEUED }
     }
 
-    suspend fun insertRoomPacket(packet: RoomPacket) =
-        withContext(dispatchers.io + NonCancellable) { dbManager.currentDb.value.packetDao().insert(packet) }
+    suspend fun insertRoomPacket(packet: RoomPacket) {
+        withContext(dispatchers.io + NonCancellable) { dbManager.withDb { it.packetDao().insert(packet) } }
+    }
 
     override suspend fun savePacket(
         myNodeNum: Int,
@@ -230,17 +241,21 @@ class PacketRepositoryImpl(private val dbManager: DatabaseProvider, private val
             }
         }
 
-    override suspend fun updateMessageStatus(d: DataPacket, m: MessageStatus) =
-        withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().updateMessageStatus(d, m) }
+    override suspend fun updateMessageStatus(d: DataPacket, m: MessageStatus) {
+        withContext(dispatchers.io) { dbManager.withDb { it.packetDao().updateMessageStatus(d, m) } }
+    }
 
-    override suspend fun updateMessageId(d: DataPacket, id: Int) =
-        withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().updateMessageId(d, id) }
+    override suspend fun updateMessageId(d: DataPacket, id: Int) {
+        withContext(dispatchers.io) { dbManager.withDb { it.packetDao().updateMessageId(d, id) } }
+    }
 
-    override suspend fun setMessageTranslation(uuid: Long, translatedText: String) =
-        withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().setTranslation(uuid, translatedText) }
+    override suspend fun setMessageTranslation(uuid: Long, translatedText: String) {
+        withContext(dispatchers.io) { dbManager.withDb { it.packetDao().setTranslation(uuid, translatedText) } }
+    }
 
-    override suspend fun setShowTranslated(uuid: Long, showTranslated: Boolean) =
-        withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().setShowTranslated(uuid, showTranslated) }
+    override suspend fun setShowTranslated(uuid: Long, showTranslated: Boolean) {
+        withContext(dispatchers.io) { dbManager.withDb { it.packetDao().setShowTranslated(uuid, showTranslated) } }
+    }
 
     override suspend fun getPacketById(id: Int): DataPacket? =
         withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().getPacketById(id)?.data }
@@ -296,29 +311,38 @@ class PacketRepositoryImpl(private val dbManager: DatabaseProvider, private val
     }
 
     override suspend fun update(packet: DataPacket, routingError: Int): Unit = withContext(dispatchers.io) {
-        val dao = dbManager.currentDb.value.packetDao()
-        // Match on key fields that identify the packet, rather than the entire data object
-        dao.findPacketsWithId(packet.id)
-            .find { it.data.id == packet.id && it.data.from == packet.from && it.data.to == packet.to }
-            ?.let { existing ->
-                val updated =
-                    if (routingError >= 0) {
-                        existing.copy(data = packet, routingError = routingError)
-                    } else {
-                        existing.copy(data = packet)
-                    }
-                dao.update(updated)
-            }
+        // Read-modify-write in one withDb block so the lookup and update land on the same DB instance.
+        dbManager.withDb { db ->
+            val dao = db.packetDao()
+            // Match on key fields that identify the packet, rather than the entire data object
+            dao.findPacketsWithId(packet.id)
+                .find { it.data.id == packet.id && it.data.from == packet.from && it.data.to == packet.to }
+                ?.let { existing ->
+                    val updated =
+                        if (routingError >= 0) {
+                            existing.copy(data = packet, routingError = routingError)
+                        } else {
+                            existing.copy(data = packet)
+                        }
+                    dao.update(updated)
+                }
+        }
+        Unit
     }
 
-    override suspend fun insertReaction(reaction: Reaction, myNodeNum: Int) =
-        withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().insert(reaction.toEntity(myNodeNum)) }
+    override suspend fun insertReaction(reaction: Reaction, myNodeNum: Int) {
+        withContext(dispatchers.io) { dbManager.withDb { it.packetDao().insert(reaction.toEntity(myNodeNum)) } }
+    }
 
-    override suspend fun updateReaction(reaction: Reaction) = withContext(dispatchers.io) {
-        val dao = dbManager.currentDb.value.packetDao()
-        dao.findReactionsWithId(reaction.packetId)
-            .find { it.userId == reaction.user.id && it.emoji == reaction.emoji }
-            ?.let { dao.update(reaction.toEntity(it.myNodeNum)) } ?: Unit
+    override suspend fun updateReaction(reaction: Reaction) {
+        withContext(dispatchers.io) {
+            dbManager.withDb { db ->
+                val dao = db.packetDao()
+                dao.findReactionsWithId(reaction.packetId)
+                    .find { it.userId == reaction.user.id && it.emoji == reaction.emoji }
+                    ?.let { dao.update(reaction.toEntity(it.myNodeNum)) }
+            }
+        }
     }
 
     override suspend fun getReactionByPacketId(packetId: Int): Reaction? = withContext(dispatchers.io) {
@@ -329,17 +353,10 @@ class PacketRepositoryImpl(private val dbManager: DatabaseProvider, private val
         dbManager.currentDb.value.packetDao().findPacketsWithId(packetId).map { it.data }
     }
 
-    private suspend fun findPacketsWithIdInternal(packetId: Int) =
-        withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().findPacketsWithId(packetId) }
-
     override suspend fun findReactionsWithId(packetId: Int): List<Reaction> = withContext(dispatchers.io) {
         dbManager.currentDb.value.packetDao().findReactionsWithId(packetId).toReaction { null }
     }
 
-    private suspend fun findReactionsWithIdInternal(packetId: Int) =
-        withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().findReactionsWithId(packetId) }
-
-    @Suppress("CyclomaticComplexMethod")
     override suspend fun updateSFPPStatus(
         packetId: Int,
         from: Int,
@@ -349,9 +366,28 @@ class PacketRepositoryImpl(private val dbManager: DatabaseProvider, private val
         rxTime: Long,
         myNodeNum: Int?,
     ) = withContext(dispatchers.io) {
-        val dao = dbManager.currentDb.value.packetDao()
-        val packets = findPacketsWithIdInternal(packetId)
-        val reactions = findReactionsWithIdInternal(packetId)
+        // One withDb block: the id lookups and the status updates all land on the same DB instance.
+        dbManager.withDb { db ->
+            val dao = db.packetDao()
+            val packets = dao.findPacketsWithId(packetId)
+            val reactions = dao.findReactionsWithId(packetId)
+            updateSFPPMatches(dao, packets, reactions, from, to, hash, status, rxTime, myNodeNum)
+        }
+        Unit
+    }
+
+    @Suppress("CyclomaticComplexMethod", "LongParameterList")
+    private suspend fun updateSFPPMatches(
+        dao: PacketDao,
+        packets: List<RoomPacket>,
+        reactions: List<RoomReaction>,
+        from: Int,
+        to: Int,
+        hash: ByteArray,
+        status: MessageStatus,
+        rxTime: Long,
+        myNodeNum: Int?,
+    ) {
         val fromId = NodeAddress.numToDefaultId(from)
         val isFromLocalNode = myNodeNum != null && from == myNodeNum
         val toId =
@@ -401,8 +437,7 @@ class PacketRepositoryImpl(private val dbManager: DatabaseProvider, private val
                     return@forEach
                 }
                 val newTime = if (rxTime > 0) rxTime * MILLISECONDS_IN_SECOND else reaction.timestamp
-                val updatedReaction =
-                    reaction.copy(status = status, sfpp_hash = hashByteString, timestamp = newTime)
+                val updatedReaction = reaction.copy(status = status, sfpp_hash = hashByteString, timestamp = newTime)
                 dao.update(updatedReaction)
             }
         }
@@ -410,46 +445,59 @@ class PacketRepositoryImpl(private val dbManager: DatabaseProvider, private val
 
     override suspend fun updateSFPPStatusByHash(hash: ByteArray, status: MessageStatus, rxTime: Long): Unit =
         withContext(dispatchers.io) {
-            val dao = dbManager.currentDb.value.packetDao()
-            val hashByteString = hash.toByteString()
-            dao.findPacketBySfppHash(hashByteString)?.let { packet ->
-                // If it's already confirmed, don't downgrade it
-                if (packet.data.status == MessageStatus.SFPP_CONFIRMED && status == MessageStatus.SFPP_ROUTING) {
-                    return@let
+            // Read-modify-write in one withDb block so the hash lookups and updates land on the same DB instance.
+            dbManager.withDb { db ->
+                val dao = db.packetDao()
+                val hashByteString = hash.toByteString()
+                dao.findPacketBySfppHash(hashByteString)?.let { packet ->
+                    // If it's already confirmed, don't downgrade it
+                    if (packet.data.status == MessageStatus.SFPP_CONFIRMED && status == MessageStatus.SFPP_ROUTING) {
+                        return@let
+                    }
+                    val newTime = if (rxTime > 0) rxTime * MILLISECONDS_IN_SECOND else packet.received_time
+                    val updatedData = packet.data.copy(status = status, sfppHash = hashByteString, time = newTime)
+                    dao.update(packet.copy(data = updatedData, sfpp_hash = hashByteString, received_time = newTime))
                 }
-                val newTime = if (rxTime > 0) rxTime * MILLISECONDS_IN_SECOND else packet.received_time
-                val updatedData = packet.data.copy(status = status, sfppHash = hashByteString, time = newTime)
-                dao.update(packet.copy(data = updatedData, sfpp_hash = hashByteString, received_time = newTime))
-            }
 
-            dao.findReactionBySfppHash(hashByteString)?.let { reaction ->
-                if (reaction.status == MessageStatus.SFPP_CONFIRMED && status == MessageStatus.SFPP_ROUTING) {
-                    return@let
+                dao.findReactionBySfppHash(hashByteString)?.let { reaction ->
+                    if (reaction.status == MessageStatus.SFPP_CONFIRMED && status == MessageStatus.SFPP_ROUTING) {
+                        return@let
+                    }
+                    val newTime = if (rxTime > 0) rxTime * MILLISECONDS_IN_SECOND else reaction.timestamp
+                    val updatedReaction =
+                        reaction.copy(status = status, sfpp_hash = hashByteString, timestamp = newTime)
+                    dao.update(updatedReaction)
                 }
-                val newTime = if (rxTime > 0) rxTime * MILLISECONDS_IN_SECOND else reaction.timestamp
-                val updatedReaction = reaction.copy(status = status, sfpp_hash = hashByteString, timestamp = newTime)
-                dao.update(updatedReaction)
             }
+            Unit
         }
 
     override suspend fun deleteMessages(uuidList: List<Long>) = withContext(dispatchers.io) {
-        for (chunk in uuidList.chunked(DELETE_CHUNK_SIZE)) {
-            // Fetch DAO per chunk to avoid holding a stale reference if the active DB switches
-            dbManager.currentDb.value.packetDao().deleteMessages(chunk)
+        // One withDb block: every chunk lands on the same DB instance (previously each chunk re-fetched the DAO to
+        // survive a mid-op switch; the drain barrier now covers the whole chunked delete instead).
+        dbManager.withDb { db ->
+            for (chunk in uuidList.chunked(DELETE_CHUNK_SIZE)) {
+                db.packetDao().deleteMessages(chunk)
+            }
         }
+        Unit
     }
 
-    override suspend fun deleteContacts(contactList: List<String>) =
-        withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().deleteContacts(contactList) }
+    override suspend fun deleteContacts(contactList: List<String>) {
+        withContext(dispatchers.io) { dbManager.withDb { it.packetDao().deleteContacts(contactList) } }
+    }
 
-    override suspend fun deleteWaypoint(id: Int) =
-        withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().deleteWaypoint(id) }
+    override suspend fun deleteWaypoint(id: Int) {
+        withContext(dispatchers.io) { dbManager.withDb { it.packetDao().deleteWaypoint(id) } }
+    }
 
-    suspend fun delete(packet: RoomPacket) =
-        withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().delete(packet) }
+    suspend fun delete(packet: RoomPacket) {
+        withContext(dispatchers.io) { dbManager.withDb { it.packetDao().delete(packet) } }
+    }
 
-    suspend fun update(packet: RoomPacket) =
-        withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().update(packet) }
+    suspend fun update(packet: RoomPacket) {
+        withContext(dispatchers.io) { dbManager.withDb { it.packetDao().update(packet) } }
+    }
 
     override fun getContactSettings(): Flow<Map<String, ContactSettings>> = dbManager.currentDb
         .flatMapLatest { db -> db.packetDao().getContactSettings() }
@@ -459,14 +507,17 @@ class PacketRepositoryImpl(private val dbManager: DatabaseProvider, private val
         dbManager.currentDb.value.packetDao().getContactSettings(contact)?.toShared() ?: ContactSettings(contact)
     }
 
-    override suspend fun setMuteUntil(contacts: List<String>, until: Long) =
-        withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().setMuteUntil(contacts, until) }
+    override suspend fun setMuteUntil(contacts: List<String>, until: Long) {
+        withContext(dispatchers.io) { dbManager.withDb { it.packetDao().setMuteUntil(contacts, until) } }
+    }
 
-    suspend fun insertReaction(reaction: RoomReaction) =
-        withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().insert(reaction) }
+    suspend fun insertReaction(reaction: RoomReaction) {
+        withContext(dispatchers.io) { dbManager.withDb { it.packetDao().insert(reaction) } }
+    }
 
-    suspend fun updateReaction(reaction: RoomReaction) =
-        withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().update(reaction) }
+    suspend fun updateReaction(reaction: RoomReaction) {
+        withContext(dispatchers.io) { dbManager.withDb { it.packetDao().update(reaction) } }
+    }
 
     override fun getFilteredCountFlow(contactKey: String): Flow<Int> =
         dbManager.currentDb.flatMapLatest { db -> db.packetDao().getFilteredCountFlow(contactKey) }
@@ -474,22 +525,25 @@ class PacketRepositoryImpl(private val dbManager: DatabaseProvider, private val
     override suspend fun getFilteredCount(contactKey: String): Int =
         withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().getFilteredCount(contactKey) }
 
-    override suspend fun setContactFilteringDisabled(contactKey: String, disabled: Boolean) =
+    override suspend fun setContactFilteringDisabled(contactKey: String, disabled: Boolean) {
         withContext(dispatchers.io) {
-            dbManager.currentDb.value.packetDao().setContactFilteringDisabled(contactKey, disabled)
+            dbManager.withDb { it.packetDao().setContactFilteringDisabled(contactKey, disabled) }
         }
+    }
 
-    override suspend fun clearPacketDB() =
-        withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().deleteAll() }
+    override suspend fun clearPacketDB() {
+        withContext(dispatchers.io) { dbManager.withDb { it.packetDao().deleteAll() } }
+    }
 
-    override suspend fun migrateChannelsByPSK(oldSettings: List<ChannelSettings>, newSettings: List<ChannelSettings>) =
+    override suspend fun migrateChannelsByPSK(oldSettings: List<ChannelSettings>, newSettings: List<ChannelSettings>) {
         withContext(dispatchers.io) {
-            dbManager.currentDb.value.packetDao().migrateChannelsByPSK(oldSettings, newSettings)
+            dbManager.withDb { it.packetDao().migrateChannelsByPSK(oldSettings, newSettings) }
         }
+    }
 
     override suspend fun updateFilteredBySender(senderId: String, filtered: Boolean) {
         val pattern = "%\"from\":\"${senderId}\"%"
-        withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().updateFilteredBySender(pattern, filtered) }
+        withContext(dispatchers.io) { dbManager.withDb { it.packetDao().updateFilteredBySender(pattern, filtered) } }
     }
 
     private fun org.meshtastic.core.database.dao.PacketDao.getAllWaypointsFlow(): Flow<List<RoomPacket>> =

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

@@ -34,21 +34,20 @@ class QuickChatActionRepositoryImpl(
     override fun getAllActions(): Flow<List<QuickChatAction>> =
         dbManager.currentDb.flatMapLatest { it.quickChatActionDao().getAll() }.flowOn(dispatchers.io)
 
+    // Writes go through withDb so they register with the cross-transport merge drain barrier (see DatabaseProvider).
     override suspend fun upsert(action: QuickChatAction) {
-        withContext(dispatchers.io) { dbManager.currentDb.value.quickChatActionDao().upsert(action) }
+        withContext(dispatchers.io) { dbManager.withDb { it.quickChatActionDao().upsert(action) } }
     }
 
     override suspend fun deleteAll() {
-        withContext(dispatchers.io) { dbManager.currentDb.value.quickChatActionDao().deleteAll() }
+        withContext(dispatchers.io) { dbManager.withDb { it.quickChatActionDao().deleteAll() } }
     }
 
     override suspend fun delete(action: QuickChatAction) {
-        withContext(dispatchers.io) { dbManager.currentDb.value.quickChatActionDao().delete(action) }
+        withContext(dispatchers.io) { dbManager.withDb { it.quickChatActionDao().delete(action) } }
     }
 
     override suspend fun setItemPosition(uuid: Long, newPos: Int) {
-        withContext(dispatchers.io) {
-            dbManager.currentDb.value.quickChatActionDao().updateActionPosition(uuid, newPos)
-        }
+        withContext(dispatchers.io) { dbManager.withDb { it.quickChatActionDao().updateActionPosition(uuid, newPos) } }
     }
 }

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

@@ -43,20 +43,25 @@ class TracerouteSnapshotRepositoryImpl(
         .flowOn(dispatchers.io)
         .conflate()
 
-    override suspend fun upsertSnapshotPositions(logUuid: String, requestId: Int, positions: Map<Int, Position>) =
+    // The delete+insert pair runs in one withDb block so both land on the same DB instance and stay visible to the
+    // cross-transport merge drain barrier (see DatabaseProvider).
+    override suspend fun upsertSnapshotPositions(logUuid: String, requestId: Int, positions: Map<Int, Position>) {
         withContext(dispatchers.io) {
-            val dao = dbManager.currentDb.value.tracerouteNodePositionDao()
-            dao.deleteByLogUuid(logUuid)
-            if (positions.isEmpty()) return@withContext
-            val entities =
-                positions.map { (nodeNum, position) ->
-                    TracerouteNodePositionEntity(
-                        logUuid = logUuid,
-                        requestId = requestId,
-                        nodeNum = nodeNum,
-                        position = position,
-                    )
-                }
-            dao.insertAll(entities)
+            dbManager.withDb {
+                val dao = it.tracerouteNodePositionDao()
+                dao.deleteByLogUuid(logUuid)
+                if (positions.isEmpty()) return@withDb
+                val entities =
+                    positions.map { (nodeNum, position) ->
+                        TracerouteNodePositionEntity(
+                            logUuid = logUuid,
+                            requestId = requestId,
+                            nodeNum = nodeNum,
+                            position = position,
+                        )
+                    }
+                dao.insertAll(entities)
+            }
         }
+    }
 }

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

@@ -383,9 +383,15 @@ open class DatabaseManager(
         }
     }
 
-    /** Atomically snapshots the active DB and registers a writer against it. Returns null if none is open. */
-    private suspend fun beginWrite(): MeshtasticDatabase? = writerTrackerMutex.withLock {
-        val db = _currentDb.value ?: return@withLock null
+    /**
+     * Atomically snapshots the active DB and registers a writer against it. Before the first [switchActiveDatabase]
+     * `_currentDb` is still null, so fall back to the public [currentDb] view — the eagerly-opened default DB — giving
+     * withDb the exact DB-resolution semantics of a direct `currentDb.value` caller. Desktop never calls [init] until a
+     * device is selected, so without the fallback pre-connection writes (quick-chat actions, firmware/hardware cache
+     * refreshes) would silently no-op there instead of landing in the default DB the pre-connection flows read from.
+     */
+    private suspend fun beginWrite(): MeshtasticDatabase = writerTrackerMutex.withLock {
+        val db = _currentDb.value ?: currentDb.value
         activeWriters[db] = (activeWriters[db] ?: 0) + 1
         db
     }
@@ -433,7 +439,7 @@ open class DatabaseManager(
 
     @Suppress("ReturnCount", "ThrowsCount", "TooGenericExceptionCaught", "CyclomaticComplexMethod")
     private suspend fun <T> withCurrentDb(block: suspend (MeshtasticDatabase) -> T): T? {
-        val db = beginWrite() ?: return null
+        val db = beginWrite()
         val active = currentDbName
         markLastUsed(active)
         try {

core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseProvider.kt

@@ -21,6 +21,14 @@ import kotlinx.coroutines.flow.StateFlow
 /**
  * Provides multiplatform access to the current [MeshtasticDatabase] and a safe transactional helper. Platform
  * implementations manage the concrete lifecycle (Room on Android, etc.).
+ *
+ * **Write policy:** every one-shot DAO *write* (insert/upsert/update/delete/clear) must go through [withDb], never
+ * `currentDb.value` directly. [withDb] registers the write with the cross-transport merge drain barrier (see
+ * [DatabaseManager.associateDevice]) so a merge can't snapshot a database while the write is still in flight and lose
+ * it when that database is retired — and it retries once if the pool closes under a concurrent DB switch. Direct
+ * `currentDb.value` is fine for one-shot *reads* (a torn read against a just-retired DB is recoverable and reads don't
+ * need drain visibility), and `currentDb` itself is the right latch for Flow/Paging factories, which must re-latch on
+ * DB switch and must stay out of [withDb]'s containment lane.
  */
 interface DatabaseProvider {
     /** Reactive stream of the currently active database instance. */