sync: preserve queue lock order during resets

Move pending-origin clearing into the queue mutation lock, acquire the cursor lock before Room, and clear pending rows before cache, revision, or listener side effects. Keep sync listener reconciliation transactional for retries while preserving direct reset commit visibility.

app/src/main/kotlin/com/subtracks/data/repo/QueueRepository.kt

@@ -11,11 +11,15 @@ import com.subtracks.data.model.QueueKind
 import com.subtracks.data.model.QueueOrigin
 import com.subtracks.data.model.SongItem
 import com.subtracks.data.model.UpNextEntry
+import kotlinx.coroutines.currentCoroutineContext
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
+import kotlinx.coroutines.withContext
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.atomic.AtomicLong
+import kotlin.coroutines.AbstractCoroutineContextElement
+import kotlin.coroutines.CoroutineContext
 
 const val QUEUE_CHUNK = 60
 
@@ -183,6 +187,10 @@ data class QueueReset(
     val snapshot: QueueSnapshot,
 )
 
+private class CursorLockContext : AbstractCoroutineContextElement(CursorLockContext) {
+    companion object Key : CoroutineContext.Key<CursorLockContext>
+}
+
 private interface QueueKindResolver {
     fun supportsKeyset(key: OrderKey): Boolean = false
 
@@ -414,6 +422,13 @@ class QueueRepository(
 
     internal val cursorMutex = Mutex()
 
+    private suspend fun <T> withCursorLock(block: suspend () -> T): T {
+        if (currentCoroutineContext()[CursorLockContext] != null) return block()
+        return cursorMutex.withLock {
+            withContext(CursorLockContext()) { block() }
+        }
+    }
+
     fun playlistEntry(
         sourceId: Long,
         playlistId: String,
@@ -508,7 +523,7 @@ class QueueRepository(
         writeLocked(withOriginEntryIds(entries, assignFragmentIds = false))
         writeUpNextLocked(withOriginEntryIds(upNext, assignFragmentIds = false))
         setShuffleLocked(shuffleSeed)
-        cursorMutex.withLock {
+        withCursorLock {
             dao.setCursor(cursorRow().copy(queuePosition = 0, positionMs = 0, upNextAnchor = upNextAnchor))
         }
     }
@@ -549,7 +564,7 @@ class QueueRepository(
 
     private suspend fun setShuffleLocked(seed: Long?) {
         val size = if (seed != null) contextSize() else 0L
-        cursorMutex.withLock {
+        withCursorLock {
             dao.setCursor(
                 cursorRow().copy(
                     shuffleEnabled = seed != null,
@@ -563,6 +578,8 @@ class QueueRepository(
 
     fun contentVersion(): Long = queueVersion.get()
 
+    internal fun libraryCacheVersion(): Long = libraryVersion.get()
+
     fun invalidateLibraryCache() {
         lengthCache.clear()
         keyCache.clear()
@@ -584,85 +601,132 @@ class QueueRepository(
         point: QueueResetPoint,
     ) {
         if (changed.isEmpty()) return
-        withMutationLock {
+        withMutationLock { resetChangedOriginsLocked(changed, point) }
+    }
+
+    suspend fun resetChangedOriginsAndClearPendingOrigins(
+        changed: Set<QueueOrigin>,
+        point: QueueResetPoint,
+        sourceId: Long,
+    ) = withMutationLock {
+        resetChangedOriginsLocked(changed, point, sourceId)
+    }
+
+    private suspend fun QueueMutationScope.resetChangedOriginsLocked(
+        changed: Set<QueueOrigin>,
+        point: QueueResetPoint,
+        pendingOriginSourceId: Long? = null,
+    ) {
+        if (changed.isNotEmpty()) {
             val before = snapshot()
             val entries = canonicalEntries(before.entries.map { it.entry }, changed)
             val upNext = canonicalEntries(before.upNext.map { it.entry }, changed)
-            val matchingOrigin =
-                before.entries.any { it.entry.origin in changed } || before.upNext.any { it.entry.origin in changed }
-            if (!matchingOrigin) return@withMutationLock
-            val updated = writeResetLocked(before, entries, upNext, point)
-            resetListener?.invoke(this, QueueReset(point, updated))
+            if (entries != before.entries.map { it.entry } || upNext != before.upNext.map { it.entry }) {
+                writeResetLocked(before, entries, upNext, point, pendingOriginSourceId) { updated ->
+                    resetListener?.invoke(this, QueueReset(point, updated))
+                }
+                return
+            }
         }
+        if (pendingOriginSourceId != null) clearPendingSyncOriginsLocked(pendingOriginSourceId)
     }
 
+    private suspend fun clearPendingSyncOriginsLocked(sourceId: Long) =
+        withCursorLock {
+            db.useWriterConnection { connection ->
+                connection.immediateTransaction {
+                    dao.clearPendingSyncOrigins(sourceId)
+                }
+            }
+        }
+
     private suspend fun writeResetLocked(
         before: QueueSnapshot,
         entries: List<QueueEntry>,
         upNext: List<QueueEntry>,
         point: QueueResetPoint,
-    ): QueueSnapshot {
-        invalidateLibraryCache()
-        val resolvedEntries = entries.map { ResolvedQueueEntry(it, it.resolvedLength()) }
-        val resolvedUpNext = upNext.map { ResolvedQueueEntry(it, it.resolvedLength()) }
-        val shuffle = before.shuffleSeed?.let { ShuffleState(it, resolvedEntries.sumOf { it.length }) }
-        var updated =
-            QueueSnapshot(
-                entries = resolvedEntries,
-                upNext = resolvedUpNext,
-                shuffle = shuffle,
-                upNextAnchor = before.upNextAnchor,
-                version = queueVersion.get(),
-                resetVersion = resetVersion.get(),
-            )
-        val anchorChanged = before.upNextSize > 0L && before.upNextAnchor != point.upNextAnchor
-        val latestAnchor = if (anchorChanged) resetItemAt(before, before.anchorPlay) else null
-        val anchor =
-            if (anchorChanged) {
-                remapAnchor(updated, latestAnchor) ?: fallbackAnchor(updated, before.anchorPlay)
-            } else {
-                remapAnchor(updated, point.anchor)
-                    ?: fallbackAnchor(updated, point.anchor?.position ?: before.upNextAnchor)
-            }
-        updated = updated.copy(upNextAnchor = anchor)
-        cursorMutex.withLock {
-            val row = cursorRow()
-            val current = point.current ?: resetItemAt(before, row.queuePosition)
-            val remappedCurrent = current?.let { remapPosition(updated, it) }
-            val cursor =
-                when {
-                    updated.size == 0L -> 0L
-                    remappedCurrent != null -> remappedCurrent.coerceIn(0, updated.size - 1)
-                    current != null -> current.position.coerceIn(0, updated.size - 1)
-                    else -> row.queuePosition.coerceIn(0, updated.size - 1)
-                }
-            val previousSongId = current?.songId
-            val updatedSongId = itemAt(updated, cursor)?.song?.id
-            val positionMs = if (remappedCurrent != null && previousSongId == updatedSongId) row.positionMs else 0L
-            db.useWriterConnection { connection ->
-                connection.immediateTransaction {
-                    dao.clear()
-                    dao.insert(entries.mapIndexed { index, entry -> entry.copy(id = 0, position = index.toLong()) })
-                    dao.clearUpNext()
-                    dao.insertUpNext(
-                        upNext.mapIndexed { index, entry -> UpNextEntry.from(entry).copy(id = 0, position = index.toLong()) },
-                    )
-                    dao.setCursor(
-                        row.copy(
-                            queuePosition = cursor,
-                            positionMs = positionMs,
-                            upNextAnchor = anchor,
-                            shuffleEnabled = shuffle != null,
-                            shuffleSeed = shuffle?.seed ?: 0L,
-                            shuffleSize = shuffle?.domain ?: 0L,
-                        ),
-                    )
+        pendingOriginSourceId: Long? = null,
+        onReset: suspend (QueueSnapshot) -> Unit,
+    ) {
+        val resetSnapshot =
+            withCursorLock {
+                db.useWriterConnection { connection ->
+                    connection.immediateTransaction {
+                        if (pendingOriginSourceId != null) dao.clearPendingSyncOrigins(pendingOriginSourceId)
+                        invalidateLibraryCache()
+                        val resolvedEntries = entries.map { ResolvedQueueEntry(it, it.resolvedLength()) }
+                        val resolvedUpNext = upNext.map { ResolvedQueueEntry(it, it.resolvedLength()) }
+                        val shuffle = before.shuffleSeed?.let { ShuffleState(it, resolvedEntries.sumOf { it.length }) }
+                        var updated =
+                            QueueSnapshot(
+                                entries = resolvedEntries,
+                                upNext = resolvedUpNext,
+                                shuffle = shuffle,
+                                upNextAnchor = before.upNextAnchor,
+                                version = queueVersion.get(),
+                                resetVersion = resetVersion.get(),
+                            )
+                        val anchorChanged = before.upNextSize > 0L && before.upNextAnchor != point.upNextAnchor
+                        val latestAnchor = if (anchorChanged) resetItemAt(before, before.anchorPlay) else null
+                        val anchor =
+                            if (anchorChanged) {
+                                remapAnchor(updated, latestAnchor) ?: fallbackAnchor(updated, before.anchorPlay)
+                            } else {
+                                remapAnchor(updated, point.anchor)
+                                    ?: fallbackAnchor(updated, point.anchor?.position ?: before.upNextAnchor)
+                            }
+                        updated = updated.copy(upNextAnchor = anchor)
+                        val row = cursorRow()
+                        val current = point.current ?: resetItemAt(before, row.queuePosition)
+                        val remappedCurrent = current?.let { remapPosition(updated, it) }
+                        val cursor =
+                            when {
+                                updated.size == 0L -> 0L
+                                remappedCurrent != null -> remappedCurrent.coerceIn(0, updated.size - 1)
+                                current != null -> current.position.coerceIn(0, updated.size - 1)
+                                else -> row.queuePosition.coerceIn(0, updated.size - 1)
+                            }
+                        val previousSongId = current?.songId
+                        val updatedSongId = itemAt(updated, cursor)?.song?.id
+                        val positionMs = if (remappedCurrent != null && previousSongId == updatedSongId) row.positionMs else 0L
+                        dao.clear()
+                        dao.insert(
+                            withOriginEntryIds(entries).mapIndexed { index, entry ->
+                                entry.copy(id = 0, position = index.toLong())
+                            },
+                        )
+                        dao.clearUpNext()
+                        dao.insertUpNext(
+                            withOriginEntryIds(upNext).mapIndexed { index, entry ->
+                                UpNextEntry.from(entry).copy(id = 0, position = index.toLong())
+                            },
+                        )
+                        dao.setCursor(
+                            row.copy(
+                                queuePosition = cursor,
+                                positionMs = positionMs,
+                                upNextAnchor = anchor,
+                                shuffleEnabled = shuffle != null,
+                                shuffleSeed = shuffle?.seed ?: 0L,
+                                shuffleSize = shuffle?.domain ?: 0L,
+                            ),
+                        )
+                        if (pendingOriginSourceId == null) {
+                            updated
+                        } else {
+                            queueVersion.incrementAndGet()
+                            resetVersion.incrementAndGet()
+                            onReset(updated.copy(version = queueVersion.get(), resetVersion = resetVersion.get()))
+                            null
+                        }
+                    }
                 }
             }
+        if (resetSnapshot != null) {
+            queueVersion.incrementAndGet()
+            resetVersion.incrementAndGet()
+            onReset(resetSnapshot.copy(version = queueVersion.get(), resetVersion = resetVersion.get()))
         }
-        queueVersion.incrementAndGet()
-        resetVersion.incrementAndGet()
-        return updated.copy(version = queueVersion.get(), resetVersion = resetVersion.get())
     }
 
     private suspend fun remapAnchor(
@@ -874,7 +938,7 @@ class QueueRepository(
         return null
     }
 
-    suspend fun setRepeat(mode: Int) = cursorMutex.withLock { dao.setCursor(cursorRow().copy(repeatMode = mode)) }
+    suspend fun setRepeat(mode: Int) = withCursorLock { dao.setCursor(cursorRow().copy(repeatMode = mode)) }
 
     suspend fun itemAt(
         snapshot: QueueSnapshot,
@@ -963,7 +1027,7 @@ class QueueRepository(
                 current.upNextAnchor
             }
         writeUpNextLocked(list)
-        cursorMutex.withLock { dao.setCursor(cursorRow().copy(upNextAnchor = anchor)) }
+        withCursorLock { dao.setCursor(cursorRow().copy(upNextAnchor = anchor)) }
         if (current.upNextSize == 0L) return currentPosition
         val before = current.upNext.take(insertIndex).sumOf { it.length }
         return if (currentPosition >= current.anchorPlay + 1 + before) currentPosition + 1 else currentPosition
@@ -978,7 +1042,7 @@ class QueueRepository(
         val play = snapshot.contextPlay(target)
         val flat = snapshot.flatContext(play) ?: return Reanchored(target, false)
         if (flat == snapshot.upNextAnchor) return Reanchored(target, false)
-        cursorMutex.withLock { dao.setCursor(cursorRow().copy(upNextAnchor = flat)) }
+        withCursorLock { dao.setCursor(cursorRow().copy(upNextAnchor = flat)) }
         queueVersion.incrementAndGet()
         return Reanchored(play, true)
     }
@@ -1007,7 +1071,7 @@ class QueueRepository(
     }
 
     private suspend fun adjustAnchorOnRemove(removedFlat: Long) =
-        cursorMutex.withLock {
+        withCursorLock {
             val row = cursorRow()
             val anchor = row.upNextAnchor
             // Removing the track the block sits behind leaves the block next in line, so move the
@@ -1020,7 +1084,7 @@ class QueueRepository(
     private suspend fun adjustAnchorOnMove(
         flatFrom: Long,
         flatTo: Long,
-    ) = cursorMutex.withLock {
+    ) = withCursorLock {
         val row = cursorRow()
         val anchor = row.upNextAnchor
         val moved =
@@ -1073,12 +1137,12 @@ class QueueRepository(
     suspend fun cursorPositionMs(): Long = dao.cursor()?.positionMs ?: 0
 
     suspend fun setPosition(positionMs: Long) =
-        cursorMutex.withLock {
+        withCursorLock {
             dao.setCursor(cursorRow().copy(positionMs = positionMs.coerceAtLeast(0)))
         }
 
     suspend fun setCursor(position: Long) =
-        cursorMutex.withLock {
+        withCursorLock {
             val row = cursorRow()
             dao.setCursor(row.copy(queuePosition = position, positionMs = if (row.queuePosition == position) row.positionMs else 0))
         }

app/src/main/kotlin/com/subtracks/data/sync/SyncManager.kt

@@ -1,7 +1,5 @@
 package com.subtracks.data.sync
 
-import androidx.room3.deferredTransaction
-import androidx.room3.useWriterConnection
 import com.subtracks.R
 import com.subtracks.UiMessage
 import com.subtracks.data.db.SubtracksDatabase
@@ -100,12 +98,7 @@ class SyncManager(
                         "${completed.artists} artists, ${completed.albums} albums, ${completed.songs} songs, " +
                         "${completed.playlists} playlists, ${completed.playlistSongs} playlist tracks, ${completed.pruned} pruned",
                 )
-                db.useWriterConnection { connection ->
-                    connection.deferredTransaction {
-                        queueRepository.resetChangedOrigins(changedOrigins, resetPoint)
-                        db.queueDao().clearPendingSyncOrigins(source.id)
-                    }
-                }
+                queueRepository.resetChangedOriginsAndClearPendingOrigins(changedOrigins, resetPoint, source.id)
                 SyncStatus.Success
             } catch (cancellation: CancellationException) {
                 throw cancellation

app/src/test/kotlin/com/subtracks/data/repo/QueueRepositoryTest.kt

@@ -15,9 +15,13 @@ import com.subtracks.data.model.QueueOrigin
 import com.subtracks.data.model.Song
 import com.subtracks.data.model.SongDownload
 import com.subtracks.data.model.Source
+import kotlinx.coroutines.CoroutineStart
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.sync.withLock
 import kotlinx.coroutines.test.runTest
+import kotlinx.coroutines.withTimeout
 import kotlinx.coroutines.yield
 import org.junit.After
 import org.junit.Assert.assertEquals
@@ -27,6 +31,8 @@ import org.junit.Assert.assertTrue
 import org.junit.Before
 import org.junit.Test
 import org.junit.runner.RunWith
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
 
 @RunWith(AndroidJUnit4::class)
 class QueueRepositoryTest {
@@ -922,6 +928,53 @@ class QueueRepositoryTest {
             assertEquals(2, repository.modes().repeat)
         }
 
+    @Test
+    fun aConcurrentPositionWriteDoesNotDeadlockAPendingOriginReset() =
+        runBlocking {
+            seedLibrary()
+            repository.snapshotAfter(listOf(repository.albumEntry(1, "al1").copy(rangeStart = 0, rangeEnd = 0)))
+            val point = repository.captureResetPoint()
+            val positionStarted = CountDownLatch(1)
+            var cursorLocked = true
+            repository.cursorMutex.lock()
+            try {
+                withTimeout(5_000) {
+                    val position =
+                        launch(Dispatchers.Default) {
+                            repository.withMutationLock {
+                                positionStarted.countDown()
+                                repository.setPosition(42_000L)
+                            }
+                        }
+                    assertTrue(positionStarted.await(5, TimeUnit.SECONDS))
+                    val reset =
+                        launch(Dispatchers.Default, start = CoroutineStart.UNDISPATCHED) {
+                            repository.resetChangedOriginsAndClearPendingOrigins(
+                                setOf(QueueOrigin(QueueKind.Album, 1, "al1")),
+                                point,
+                                sourceId = 1,
+                            )
+                        }
+                    yield()
+                    repository.cursorMutex.unlock()
+                    cursorLocked = false
+                    position.join()
+                    reset.join()
+                }
+            } finally {
+                if (cursorLocked) repository.cursorMutex.unlock()
+            }
+
+            assertNull(
+                repository
+                    .snapshot()
+                    .entries
+                    .single()
+                    .entry.rangeStart,
+            )
+            assertEquals(42_000L, repository.cursorPositionMs())
+        }
+
     @Test
     fun upNextResolvesAfterTheCurrentTrack() =
         runTest {

app/src/test/kotlin/com/subtracks/data/sync/SyncManagerTest.kt

@@ -285,13 +285,14 @@ class SyncManagerTest {
             server.dispatcher = healthyDispatcher()
             sourceRepository.addSource("Local", server.url("/").toString(), "u", "p", useTokenAuth = false)
             queueRepository.replace(listOf(queueRepository.albumEntry(1, "al1").copy(rangeStart = 0, rangeEnd = 0)))
-            val resetAttempted = CountDownLatch(1)
-            queueRepository.setResetListener { resetAttempted.countDown() }
+            val before = queueRepository.snapshot()
+            val cacheVersion = queueRepository.libraryCacheVersion()
+            val resetAttempts = AtomicInteger()
+            queueRepository.setResetListener { resetAttempts.incrementAndGet() }
             failPendingOriginClear()
 
             manager.requestSync()
-            assertTrue(resetAttempted.await(10, TimeUnit.SECONDS))
-            awaitUntil("waiting for the pending-origin clear failure") { manager.status.value !is SyncStatus.Running }
+            withTimeout(TEST_TIMEOUT_MS) { manager.status.first { it is SyncStatus.Failed } }
 
             assertEquals(
                 0L,
@@ -301,8 +302,16 @@ class SyncManagerTest {
                     .single()
                     .entry.rangeStart,
             )
-            assertTrue(manager.status.value is SyncStatus.Failed)
             assertTrue(db.queueDao().pendingSyncOrigins(1).isNotEmpty())
+            assertEquals(
+                listOf(0L, before.version, before.resetVersion, cacheVersion),
+                listOf(
+                    resetAttempts.get().toLong(),
+                    queueRepository.contentVersion(),
+                    queueRepository.snapshot().resetVersion,
+                    queueRepository.libraryCacheVersion(),
+                ),
+            )
 
             allowPendingOriginClear()
             manager.requestSync()
@@ -316,6 +325,7 @@ class SyncManagerTest {
                     .entry.rangeStart,
             )
             assertTrue(db.queueDao().pendingSyncOrigins(1).isEmpty())
+            assertEquals(1, resetAttempts.get())
         }
 
     @Test
@@ -327,6 +337,7 @@ class SyncManagerTest {
             val resetAttempted = CountDownLatch(1)
             val resetAttempts = AtomicInteger()
             queueRepository.setResetListener {
+                queueRepository.setPosition(42_000L)
                 resetAttempted.countDown()
                 if (resetAttempts.incrementAndGet() == 1) throw IllegalStateException("reconciliation failed")
             }