fix(database): make completed cross-transport merges retry-idempotent (#6231)

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

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

@@ -25,6 +25,7 @@ import androidx.datastore.preferences.core.longPreferencesKey
 import androidx.datastore.preferences.core.stringPreferencesKey
 import co.touchlab.kermit.Logger
 import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.Job
@@ -45,6 +46,7 @@ import kotlinx.coroutines.launch
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
 import kotlinx.coroutines.withContext
+import kotlinx.coroutines.withTimeoutOrNull
 import org.koin.core.annotation.Named
 import org.koin.core.annotation.Single
 import org.meshtastic.core.common.util.normalizeAddress
@@ -66,6 +68,16 @@ open class DatabaseManager(
     private val managerScope = CoroutineScope(SupervisorJob() + dispatchers.default)
     private val mutex = Mutex()
 
+    // Per-source write barrier for merges. `withDb` deliberately does NOT take [mutex] (hot path), so a merge under
+    // [mutex] must still drain any in-flight writer that captured the source DB before folding it away — otherwise a
+    // late-committing write is lost when the source is retired. This dedicated lock (never held across a drain await,
+    // so it can't deadlock the merge) tracks live `withDb` blocks per captured DB instance. It also guards the merge's
+    // active-DB swap so a writer either registers against `source` before the swap and is drained, or captures `dest`
+    // after it and is safe — it can never slip through the gap between the two.
+    private val writerTrackerMutex = Mutex()
+    private val activeWriters = mutableMapOf<MeshtasticDatabase, Int>()
+    private val drainWaiters = mutableMapOf<MeshtasticDatabase, MutableList<CompletableDeferred<Unit>>>()
+
     private val cacheLimitKey = intPreferencesKey(DatabaseConstants.CACHE_LIMIT_KEY)
     private val legacyCleanedKey = booleanPreferencesKey(DatabaseConstants.LEGACY_DB_CLEANED_KEY)
 
@@ -235,22 +247,23 @@ open class DatabaseManager(
                     val source = _currentDb.value ?: return@withLock
                     val dest = withContext(dispatchers.io) { getOrOpenDatabase(claimed) }
 
-                    // Redirect live writers to the canonical DB BEFORE merging. Otherwise a concurrent withDb
-                    // write (connect triggers a full NodeDB re-dump) could land in `source` after the merge has
-                    // already snapshotted it, then be lost when `source` is retired — and withDb's closed-pool
-                    // retry can't recover it because the write itself succeeded. New writers now capture `dest`;
-                    // any still holding `source` are covered by that retry once retireDatabase closes it.
-                    _currentDb.value = dest
-                    currentDbName = claimed
+                    // Redirect live writers to the canonical DB BEFORE merging (a connect triggers a full NodeDB
+                    // re-dump). The swap is published under the writer lock so a concurrent withDb write registers
+                    // against `source` before it — and is drained below — or captures `dest` after it and is safe.
+                    // New writers now capture `dest`; drainWriters then waits out any still writing to `source` so
+                    // the merge can't snapshot `source` mid-write and lose it when `source` is later retired.
+                    publishActiveDb(dest, claimed)
                     try {
-                        withContext(dispatchers.io) { DatabaseMerger.merge(source, dest) }
+                        withContext(dispatchers.io) {
+                            drainWriters(source, sourceName)
+                            DatabaseMerger.merge(source, dest, sourceName)
+                        }
                     } catch (e: CancellationException) {
                         throw e
                     } catch (e: Exception) {
                         // merge() is atomic, so on failure `dest` is unchanged. Roll the active DB back to
                         // `source` so the address still resolves consistently and the merge retries next connect.
-                        _currentDb.value = source
-                        currentDbName = sourceName
+                        publishActiveDb(source, sourceName)
                         Logger.w(e) {
                             "Merge into ${anonymizeDbName(claimed)} failed; kept ${anonymizeDbName(sourceName)} active"
                         }
@@ -370,9 +383,57 @@ 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
+        activeWriters[db] = (activeWriters[db] ?: 0) + 1
+        db
+    }
+
+    /**
+     * Registers a writer against a specific [db] — used by the withDb retry paths, whose target is a recovered/new
+     * instance rather than the snapshotted active DB, so their writes stay visible to a concurrent drain too.
+     */
+    private suspend fun registerWriter(db: MeshtasticDatabase) =
+        writerTrackerMutex.withLock { activeWriters[db] = (activeWriters[db] ?: 0) + 1 }
+
+    /** Deregisters a writer and releases any merge waiting for [db] to quiesce. Cancellation-safe (see call site). */
+    private suspend fun endWrite(db: MeshtasticDatabase) = writerTrackerMutex.withLock {
+        val remaining = (activeWriters[db] ?: 1) - 1
+        if (remaining <= 0) {
+            activeWriters.remove(db)
+            drainWaiters.remove(db)?.forEach { it.complete(Unit) }
+        } else {
+            activeWriters[db] = remaining
+        }
+    }
+
+    /** Publishes [db]/[name] as active under the writer lock so concurrent [beginWrite]s order against the swap. */
+    private suspend fun publishActiveDb(db: MeshtasticDatabase, name: String) = writerTrackerMutex.withLock {
+        _currentDb.value = db
+        currentDbName = name
+    }
+
+    /**
+     * Suspends until every writer that captured [db] before this call has finished, so a merge never snapshots [db]
+     * while a write is still in flight (and then loses it when [db] is retired). Bounded by [WRITER_DRAIN_TIMEOUT_MS]
+     * so a wedged writer can't pin the merge — and [mutex] — forever; falling through on timeout is no worse than the
+     * old barrier-less behavior for that rare case.
+     */
+    private suspend fun drainWriters(db: MeshtasticDatabase, dbName: String) {
+        val waiter =
+            writerTrackerMutex.withLock {
+                if ((activeWriters[db] ?: 0) == 0) return
+                CompletableDeferred<Unit>().also { drainWaiters.getOrPut(db) { mutableListOf() }.add(it) }
+            }
+        if (withTimeoutOrNull(WRITER_DRAIN_TIMEOUT_MS) { waiter.await() } == null) {
+            Logger.w { "Timed out draining writers on ${anonymizeDbName(dbName)} before merge" }
+        }
+    }
+
     @Suppress("ReturnCount", "ThrowsCount", "TooGenericExceptionCaught", "CyclomaticComplexMethod")
     private suspend fun <T> withCurrentDb(block: suspend (MeshtasticDatabase) -> T): T? {
-        val db = _currentDb.value ?: return null
+        val db = beginWrite() ?: return null
         val active = currentDbName
         markLastUsed(active)
         try {
@@ -385,14 +446,7 @@ open class DatabaseManager(
             val retryDb = _currentDb.value
             if (retryDb != null && retryDb !== db && isDbClosedException(e)) {
                 Logger.w { "withDb: database closed during switch (${e.message}), retrying with current DB" }
-                return try {
-                    runCancellableDbBlock(retryDb, block)
-                } catch (retryCancel: CancellationException) {
-                    throw retryCancel
-                } catch (retryEx: Exception) {
-                    retryEx.addSuppressed(e)
-                    throw retryEx
-                }
+                return retryRegisteredDbBlock(retryDb, e, block)
             }
 
             // Same active DB but Room's connection pool is wedged — reopen onto a fresh active instance once.
@@ -406,17 +460,40 @@ open class DatabaseManager(
                         "withDb: active DB switched during timeout recovery; retrying with current DB"
                     }
                 }
-                return try {
-                    runCancellableDbBlock(recoveredDb, block)
-                } catch (retryCancel: CancellationException) {
-                    throw retryCancel
-                } catch (retryEx: Exception) {
-                    retryEx.addSuppressed(e)
-                    throw retryEx
-                }
+                return retryRegisteredDbBlock(recoveredDb, e, block)
             }
 
             throw e
+        } finally {
+            // NonCancellable so a cancelled withDb still deregisters — a leaked +1 would make every future
+            // drain on this DB instance time out.
+            withContext(NonCancellable) { endWrite(db) }
+        }
+    }
+
+    /**
+     * Retries [block] against [db] — the recovered/new instance a withDb retry targets instead of the DB it originally
+     * registered against. Registers a writer on [db] for the duration so the retry write stays visible to a concurrent
+     * merge draining [db], with the same NonCancellable deregistration guarantee as [withCurrentDb]'s outer
+     * registration (which remains held on the original DB until that finally runs — the overlap is harmless, counts
+     * balance per instance). Any retry failure carries the original failure [cause] as a suppressed exception.
+     */
+    @Suppress("TooGenericExceptionCaught")
+    private suspend fun <T> retryRegisteredDbBlock(
+        db: MeshtasticDatabase,
+        cause: Exception,
+        block: suspend (MeshtasticDatabase) -> T,
+    ): T {
+        registerWriter(db)
+        try {
+            return runCancellableDbBlock(db, block)
+        } catch (retryCancel: CancellationException) {
+            throw retryCancel
+        } catch (retryEx: Exception) {
+            retryEx.addSuppressed(cause)
+            throw retryEx
+        } finally {
+            withContext(NonCancellable) { endWrite(db) }
         }
     }
 
@@ -439,6 +516,11 @@ open class DatabaseManager(
     internal companion object {
         private const val BACKFILL_COLD_START_DELAY_MS = 2_000L
         private const val WITH_DB_SLOW_OPERATION_MS = 1_000L
+
+        /**
+         * Upper bound on how long a merge waits for in-flight writers on the source DB to drain (see [drainWriters]).
+         */
+        private const val WRITER_DRAIN_TIMEOUT_MS = 5_000L
         val DB_TERMS = listOf("pool", "database", "connection", "sqlite")
 
         private const val ROOM_POOL_ACQUIRE_TIMEOUT_PHRASE = "timed out attempting to acquire"

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

@@ -19,6 +19,8 @@ package org.meshtastic.core.database
 import androidx.room3.immediateTransaction
 import androidx.room3.useWriterConnection
 import co.touchlab.kermit.Logger
+import org.meshtastic.core.common.util.nowMillis
+import org.meshtastic.core.database.entity.MergeMarkerEntity
 
 /**
  * Folds the contents of one device database ([source]) into another ([dest]) when both turn out to belong to the same
@@ -36,15 +38,26 @@ import co.touchlab.kermit.Logger
  */
 object DatabaseMerger {
 
-    suspend fun merge(source: MeshtasticDatabase, dest: MeshtasticDatabase) {
+    /**
+     * Folds [source] into [dest], skipping the work if [sourceName] was already merged into [dest]. [sourceName] is the
+     * source database's file name — the stable key under which the merge is recorded (see [MergeMarkerEntity]).
+     */
+    suspend fun merge(source: MeshtasticDatabase, dest: MeshtasticDatabase, sourceName: String) {
         // All destination writes run in a single transaction so a crash or exception mid-merge rolls
-        // back cleanly instead of leaving `dest` half-merged. This also makes a retried merge safe: if
-        // associateDevice re-runs (e.g. a crash before the address alias is persisted), the rolled-back
-        // partial writes never happened, so the fresh-id packet/discovery re-inserts can't duplicate.
-        // Reads from `source` use its own connection pool and don't participate in this transaction.
+        // back cleanly instead of leaving `dest` half-merged. The merge marker is written inside that
+        // same transaction, so it commits atomically with the copied rows: on a retried merge (e.g. a
+        // crash after commit but before associateDevice persisted the address alias) the marker is already
+        // present and the whole merge is skipped, so the fresh-id packet/discovery re-inserts — which are
+        // NOT idempotent on their own — can never duplicate. Reads from `source` use its own connection
+        // pool and don't participate in this transaction.
         var packets = 0
+        var skipped = false
         dest.useWriterConnection { transactor ->
             transactor.immediateTransaction {
+                if (dest.mergeMarkerDao().isMerged(sourceName)) {
+                    skipped = true
+                    return@immediateTransaction
+                }
                 packets = mergePackets(source, dest)
                 mergeReactions(source, dest)
                 mergeContactSettings(source, dest)
@@ -55,9 +68,14 @@ object DatabaseMerger {
                 mergeLogs(source, dest)
                 mergeTraceroutePositions(source, dest)
                 mergeDiscovery(source, dest)
+                dest.mergeMarkerDao().insertMarker(MergeMarkerEntity(sourceDbName = sourceName, mergedAt = nowMillis))
             }
         }
-        Logger.i { "Merged $packets packets across transports into unified node DB" }
+        if (skipped) {
+            Logger.i { "Source ${anonymizeDbName(sourceName)} already merged; skipped duplicate merge" }
+        } else {
+            Logger.i { "Merged $packets packets across transports into unified node DB" }
+        }
     }
 
     /**

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

@@ -30,6 +30,7 @@ import org.meshtastic.core.database.dao.DeviceLinkDao
 import org.meshtastic.core.database.dao.DiscoveryDao
 import org.meshtastic.core.database.dao.EventFirmwareEditionDao
 import org.meshtastic.core.database.dao.FirmwareReleaseDao
+import org.meshtastic.core.database.dao.MergeMarkerDao
 import org.meshtastic.core.database.dao.MeshLogDao
 import org.meshtastic.core.database.dao.NodeInfoDao
 import org.meshtastic.core.database.dao.PacketDao
@@ -43,6 +44,7 @@ import org.meshtastic.core.database.entity.DiscoveryPresetResultEntity
 import org.meshtastic.core.database.entity.DiscoverySessionEntity
 import org.meshtastic.core.database.entity.EventFirmwareEditionEntity
 import org.meshtastic.core.database.entity.FirmwareReleaseEntity
+import org.meshtastic.core.database.entity.MergeMarkerEntity
 import org.meshtastic.core.database.entity.MeshLog
 import org.meshtastic.core.database.entity.MetadataEntity
 import org.meshtastic.core.database.entity.MyNodeEntity
@@ -73,6 +75,7 @@ import org.meshtastic.core.database.entity.TracerouteNodePositionEntity
         DiscoveryPresetResultEntity::class,
         DiscoveredNodeEntity::class,
         EventFirmwareEditionEntity::class,
+        MergeMarkerEntity::class,
     ],
     autoMigrations =
     [
@@ -121,8 +124,9 @@ import org.meshtastic.core.database.entity.TracerouteNodePositionEntity
         AutoMigration(from = 45, to = 46),
         AutoMigration(from = 46, to = 47),
         AutoMigration(from = 47, to = 48),
+        AutoMigration(from = 48, to = 49),
     ],
-    version = 48,
+    version = 49,
     exportSchema = true,
 )
 @androidx.room3.ConstructedBy(MeshtasticDatabaseConstructor::class)
@@ -149,6 +153,8 @@ abstract class MeshtasticDatabase : RoomDatabase() {
 
     abstract fun eventFirmwareEditionDao(): EventFirmwareEditionDao
 
+    abstract fun mergeMarkerDao(): MergeMarkerDao
+
     companion object {
         /**
          * Configures a [RoomDatabase.Builder] with standard settings for this project.

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

@@ -0,0 +1,35 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.meshtastic.core.database.dao
+
+import androidx.room3.Dao
+import androidx.room3.Insert
+import androidx.room3.OnConflictStrategy
+import androidx.room3.Query
+import org.meshtastic.core.database.entity.MergeMarkerEntity
+
+/** Tracks which source databases have already been merged into this (canonical) database. See [MergeMarkerEntity]. */
+@Dao
+interface MergeMarkerDao {
+
+    /** IGNORE keeps the earliest marker on a re-run; the row's mere existence is what matters, not its timestamp. */
+    @Insert(onConflict = OnConflictStrategy.IGNORE)
+    suspend fun insertMarker(marker: MergeMarkerEntity)
+
+    @Query("SELECT EXISTS(SELECT 1 FROM merge_marker WHERE source_db_name = :sourceDbName)")
+    suspend fun isMerged(sourceDbName: String): Boolean
+}

core/database/src/commonMain/kotlin/org/meshtastic/core/database/entity/MergeMarkerEntity.kt

@@ -0,0 +1,36 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.meshtastic.core.database.entity
+
+import androidx.room3.ColumnInfo
+import androidx.room3.Entity
+import androidx.room3.PrimaryKey
+
+/**
+ * A durable record, in the canonical (destination) database, that a given source database has already been folded in by
+ * [org.meshtastic.core.database.DatabaseMerger]. Written inside the same merge transaction as the copied rows, so it
+ * commits atomically with them. When [org.meshtastic.core.database.DatabaseManager.associateDevice] re-runs the merge
+ * on the next connection, `DatabaseMerger.merge` finds the marker at the top of its transaction and skips the whole
+ * merge — closing the crash window between the merge commit and the datastore alias write, where a retry would
+ * otherwise duplicate the fresh-id packet/discovery rows. Markers share the destination DB's lifecycle: if the
+ * canonical DB is LRU-evicted and later recreated empty, its markers vanish with the merged data they describe.
+ */
+@Entity(tableName = "merge_marker")
+data class MergeMarkerEntity(
+    @PrimaryKey @ColumnInfo(name = "source_db_name") val sourceDbName: String,
+    @ColumnInfo(name = "merged_at") val mergedAt: Long = 0L,
+)

core/database/src/commonTest/kotlin/org/meshtastic/core/database/DatabaseMergerTest.kt

@@ -52,6 +52,9 @@ class DatabaseMergerTest {
     // Same physical node reached over two transports → same myNodeNum in both DBs.
     private val nodeNum = 0x1234
 
+    // The source DB's file name — the key under which a completed merge is recorded in `dest`.
+    private val sourceName = "meshtastic_database_source"
+
     private val source: MeshtasticDatabase = getInMemoryDatabaseBuilder().build()
     private val dest: MeshtasticDatabase = getInMemoryDatabaseBuilder().build()
 
@@ -171,7 +174,7 @@ class DatabaseMergerTest {
             .discoveryDao()
             .insertDiscoveredNodes(listOf(DiscoveredNodeEntity(presetResultId = srcPreset, nodeNum = 30L)))
 
-        DatabaseMerger.merge(source, dest)
+        DatabaseMerger.merge(source, dest, sourceName)
 
         // Packets: 1 + 2, with fresh distinct non-zero uuids (no collision with dest's existing uuid).
         val packets = dest.packetDao().getAllPacketsSnapshot()
@@ -220,4 +223,51 @@ class DatabaseMergerTest {
             "node appended under rewired preset id",
         )
     }
+
+    /**
+     * Regression for #6230: a crash after the merge transaction commits but before the address alias is persisted
+     * re-runs [DatabaseMerger.merge] on the next connection. The non-idempotent tables (packets, discovery) must not be
+     * duplicated. The durable per-source merge marker, written inside the merge transaction, makes the second merge a
+     * no-op.
+     */
+    @Test
+    fun mergeIsIdempotentAcrossRetry() = runTest {
+        source.nodeInfoDao().setMyNodeInfo(myNode())
+        source.packetDao().insert(textPacket("0!broadcast", "srcmsgone", time = 200))
+        source.packetDao().insert(textPacket("0!broadcast", "srcmsgtwo", time = 300))
+        val srcSession =
+            source
+                .discoveryDao()
+                .insertSession(
+                    DiscoverySessionEntity(timestamp = 1, presetsScanned = "LongFast", homePreset = "LongFast"),
+                )
+        val srcPreset =
+            source
+                .discoveryDao()
+                .insertPresetResult(DiscoveryPresetResultEntity(sessionId = srcSession, presetName = "LongFast"))
+        source
+            .discoveryDao()
+            .insertDiscoveredNodes(listOf(DiscoveredNodeEntity(presetResultId = srcPreset, nodeNum = 30L)))
+
+        DatabaseMerger.merge(source, dest, sourceName)
+
+        val packetsAfterFirst = dest.packetDao().getAllPacketsSnapshot().size
+        val sessionsAfterFirst = dest.discoveryDao().getAllSessionsSnapshot().size
+        assertEquals(2, packetsAfterFirst, "both source packets merged once")
+        assertEquals(1, sessionsAfterFirst, "source discovery session merged once")
+
+        // Simulate the crash-window retry: same source folded into the same dest again.
+        DatabaseMerger.merge(source, dest, sourceName)
+
+        assertEquals(
+            packetsAfterFirst,
+            dest.packetDao().getAllPacketsSnapshot().size,
+            "packets not duplicated on retry",
+        )
+        assertEquals(
+            sessionsAfterFirst,
+            dest.discoveryDao().getAllSessionsSnapshot().size,
+            "discovery sessions not duplicated on retry",
+        )
+    }
 }