From 95c8a2a9158821b483286fd588de6843493afa50 Mon Sep 17 00:00:00 2001 From: austinried <4966622+austinried@users.noreply.github.com> Date: Thu, 12 Nov 2026 14:35:32 +0000 Subject: [PATCH] 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. --- .../subtracks/data/repo/QueueRepository.kt | 212 ++++++++++++------ .../com/subtracks/data/sync/SyncManager.kt | 9 +- .../data/repo/QueueRepositoryTest.kt | 53 +++++ .../subtracks/data/sync/SyncManagerTest.kt | 21 +- 4 files changed, 208 insertions(+), 87 deletions(-) diff --git a/app/src/main/kotlin/com/subtracks/data/repo/QueueRepository.kt b/app/src/main/kotlin/com/subtracks/data/repo/QueueRepository.kt index f7ca47c8..58dfe871 100644 --- a/app/src/main/kotlin/com/subtracks/data/repo/QueueRepository.kt +++ b/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 +} + private interface QueueKindResolver { fun supportsKeyset(key: OrderKey): Boolean = false @@ -414,6 +422,13 @@ class QueueRepository( internal val cursorMutex = Mutex() + private suspend fun 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, + point: QueueResetPoint, + sourceId: Long, + ) = withMutationLock { + resetChangedOriginsLocked(changed, point, sourceId) + } + + private suspend fun QueueMutationScope.resetChangedOriginsLocked( + changed: Set, + 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, upNext: List, 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)) } diff --git a/app/src/main/kotlin/com/subtracks/data/sync/SyncManager.kt b/app/src/main/kotlin/com/subtracks/data/sync/SyncManager.kt index 75e0362a..ef232f56 100644 --- a/app/src/main/kotlin/com/subtracks/data/sync/SyncManager.kt +++ b/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 diff --git a/app/src/test/kotlin/com/subtracks/data/repo/QueueRepositoryTest.kt b/app/src/test/kotlin/com/subtracks/data/repo/QueueRepositoryTest.kt index d16a16ad..94023449 100644 --- a/app/src/test/kotlin/com/subtracks/data/repo/QueueRepositoryTest.kt +++ b/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 { diff --git a/app/src/test/kotlin/com/subtracks/data/sync/SyncManagerTest.kt b/app/src/test/kotlin/com/subtracks/data/sync/SyncManagerTest.kt index 22bb20c5..63f4e274 100644 --- a/app/src/test/kotlin/com/subtracks/data/sync/SyncManagerTest.kt +++ b/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") }