From e48640e4734003cebc9292d7a45400a674f1612a Mon Sep 17 00:00:00 2001 From: James Rich <2199651+jamesarich@users.noreply.github.com> Date: Sat, 11 Jul 2026 19:12:57 -0500 Subject: [PATCH] fix(database): route all one-shot DB writes through the merge drain barrier (#6233) Co-authored-by: Claude Opus 4.8 (1M context) --- .../EventFirmwareEditionLocalDataSource.kt | 9 +- .../data/repository/MeshLogRepositoryImpl.kt | 51 ++-- .../data/repository/PacketRepositoryImpl.kt | 250 +++++++++++------- .../QuickChatActionRepositoryImpl.kt | 11 +- .../TracerouteSnapshotRepositoryImpl.kt | 33 ++- .../core/database/DatabaseManager.kt | 14 +- .../core/database/DatabaseProvider.kt | 8 + 7 files changed, 233 insertions(+), 143 deletions(-) diff --git a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/datasource/EventFirmwareEditionLocalDataSource.kt b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/datasource/EventFirmwareEditionLocalDataSource.kt index dfc214ead..891ed6efc 100644 --- a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/datasource/EventFirmwareEditionLocalDataSource.kt +++ b/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) = - withContext(dispatchers.io) { dao.upsertAll(editions) } + suspend fun upsertAll(editions: List) { + 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) { 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() } diff --git a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/MeshLogRepositoryImpl.kt b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/MeshLogRepositoryImpl.kt index 57e6312b0..52698b526 100644 --- a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/MeshLogRepositoryImpl.kt +++ b/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 { 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 8ea396944..0f42fef0a 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 @@ -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 = 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 = 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 = 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, + reactions: List, + 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) = 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) = - withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().deleteContacts(contactList) } + override suspend fun deleteContacts(contactList: List) { + 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> = 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, until: Long) = - withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().setMuteUntil(contacts, until) } + override suspend fun setMuteUntil(contacts: List, 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 = 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, newSettings: List) = + override suspend fun migrateChannelsByPSK(oldSettings: List, newSettings: List) { 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> = diff --git a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/QuickChatActionRepositoryImpl.kt b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/QuickChatActionRepositoryImpl.kt index 6546b6504..2e0f994aa 100644 --- a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/QuickChatActionRepositoryImpl.kt +++ b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/QuickChatActionRepositoryImpl.kt @@ -34,21 +34,20 @@ class QuickChatActionRepositoryImpl( override fun getAllActions(): Flow> = 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) } } } } diff --git a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/TracerouteSnapshotRepositoryImpl.kt b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/TracerouteSnapshotRepositoryImpl.kt index 4ad7afc24..769d18479 100644 --- a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/TracerouteSnapshotRepositoryImpl.kt +++ b/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) = + // 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) { 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) + } } + } } diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseManager.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseManager.kt index e98b40705..885458a8c 100644 --- a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseManager.kt +++ b/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 withCurrentDb(block: suspend (MeshtasticDatabase) -> T): T? { - val db = beginWrite() ?: return null + val db = beginWrite() val active = currentDbName markLastUsed(active) try { diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseProvider.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseProvider.kt index e406ed7fa..fc9daabee 100644 --- a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseProvider.kt +++ b/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. */