From 705c93053f2f65804e316a8ffaa3e6d6439d8161 Mon Sep 17 00:00:00 2001 From: austinried <4966622+austinried@users.noreply.github.com> Date: Sat, 14 Nov 2026 02:02:39 +0000 Subject: [PATCH] sync: publish queue resets after commit Prepare queue and playback reconciliation inside the reset transaction without publishing listener, cache, or revision state. Publish through a non-suspending, non-cancellable post-commit step so committed resets are never retried over later edits. Keep the pending-origin clear under the cursor lock without a redundant single-statement transaction, and cover listener and post-clear write failures. --- .../subtracks/data/repo/QueueRepository.kt | 193 ++++++++++-------- .../subtracks/playback/PlaybackController.kt | 176 +++++++++++----- .../subtracks/data/sync/SyncManagerTest.kt | 87 +++++++- .../playback/PlaybackControllerTest.kt | 36 +--- 4 files changed, 322 insertions(+), 170 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 58dfe871..8daf1811 100644 --- a/app/src/main/kotlin/com/subtracks/data/repo/QueueRepository.kt +++ b/app/src/main/kotlin/com/subtracks/data/repo/QueueRepository.kt @@ -11,6 +11,7 @@ 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.NonCancellable import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.flow.first import kotlinx.coroutines.sync.Mutex @@ -20,6 +21,7 @@ import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.atomic.AtomicLong import kotlin.coroutines.AbstractCoroutineContextElement import kotlin.coroutines.CoroutineContext +import kotlin.coroutines.EmptyCoroutineContext const val QUEUE_CHUNK = 60 @@ -187,10 +189,20 @@ data class QueueReset( val snapshot: QueueSnapshot, ) +data class QueueResetPreparation( + val snapshot: QueueSnapshot, + val publicationContext: CoroutineContext = EmptyCoroutineContext, + val publish: QueueRepository.QueueMutationScope.(QueueSnapshot) -> Unit = {}, +) + private class CursorLockContext : AbstractCoroutineContextElement(CursorLockContext) { companion object Key : CoroutineContext.Key } +private class ResetPreparationContext : AbstractCoroutineContextElement(ResetPreparationContext) { + companion object Key : CoroutineContext.Key +} + private interface QueueKindResolver { fun supportsKeyset(key: OrderKey): Boolean = false @@ -396,9 +408,9 @@ class QueueRepository( private val resetVersion = AtomicLong() private val libraryVersion = AtomicLong() private val queueMutex = Mutex() - private var resetListener: (suspend QueueMutationScope.(QueueReset) -> Unit)? = null + private var resetListener: (suspend QueueMutationScope.(QueueReset) -> QueueResetPreparation)? = null - fun setResetListener(listener: suspend QueueMutationScope.(QueueReset) -> Unit) { + fun setResetListener(listener: suspend QueueMutationScope.(QueueReset) -> QueueResetPreparation) { resetListener = listener } @@ -622,9 +634,7 @@ class QueueRepository( val entries = canonicalEntries(before.entries.map { it.entry }, changed) val upNext = canonicalEntries(before.upNext.map { it.entry }, changed) 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)) - } + writeResetLocked(before, entries, upNext, point, pendingOriginSourceId) return } } @@ -633,99 +643,91 @@ class QueueRepository( private suspend fun clearPendingSyncOriginsLocked(sourceId: Long) = withCursorLock { - db.useWriterConnection { connection -> - connection.immediateTransaction { - dao.clearPendingSyncOrigins(sourceId) - } - } + dao.clearPendingSyncOrigins(sourceId) } - private suspend fun writeResetLocked( + private suspend fun QueueMutationScope.writeResetLocked( before: QueueSnapshot, entries: List, upNext: List, point: QueueResetPoint, pendingOriginSourceId: Long? = null, - onReset: suspend (QueueSnapshot) -> Unit, ) { - val resetSnapshot = + val preparation = 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(), + withContext(ResetPreparationContext()) { + db.useWriterConnection { connection -> + connection.immediateTransaction { + if (pendingOriginSourceId != null) dao.clearPendingSyncOrigins(pendingOriginSourceId) + 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() + 1, + resetVersion = resetVersion.get() + 1, + ) + 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()) + }, ) - 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 + 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, + ), + ) + resetListener?.invoke(this@writeResetLocked, QueueReset(point, updated)) + ?: QueueResetPreparation(updated) } } } } - if (resetSnapshot != null) { - queueVersion.incrementAndGet() - resetVersion.incrementAndGet() - onReset(resetSnapshot.copy(version = queueVersion.get(), resetVersion = resetVersion.get())) + invalidateLibraryCache() + queueVersion.incrementAndGet() + resetVersion.incrementAndGet() + val published = preparation.snapshot.copy(version = queueVersion.get(), resetVersion = resetVersion.get()) + withContext(preparation.publicationContext + NonCancellable) { + preparation.publish(this@writeResetLocked, published) } } @@ -1033,6 +1035,18 @@ class QueueRepository( return if (currentPosition >= current.anchorPlay + 1 + before) currentPosition + 1 else currentPosition } + private suspend fun reanchorForResetLocked( + snapshot: QueueSnapshot, + target: Long, + ): Reanchored { + if (snapshot.upNextSize == 0L || snapshot.isUpNext(target)) return Reanchored(target, false) + val play = snapshot.contextPlay(target) + val flat = snapshot.flatContext(play) ?: return Reanchored(target, false) + if (flat == snapshot.upNextAnchor) return Reanchored(target, false) + withCursorLock { dao.setCursor(cursorRow().copy(upNextAnchor = flat)) } + return Reanchored(play, true) + } + private suspend fun reanchorForLocked( snapshot: QueueSnapshot, target: Long, @@ -1198,6 +1212,11 @@ class QueueRepository( target: Long, ): Reanchored = reanchorForLocked(snapshot, target) + suspend fun reanchorForReset( + snapshot: QueueSnapshot, + target: Long, + ): Reanchored = reanchorForResetLocked(snapshot, target) + suspend fun clearUpNext() = writeUpNextLocked(emptyList()) suspend fun removeUpNextAt( @@ -1342,7 +1361,8 @@ class QueueRepository( return resolverFor(entry.kind).rows(dao, entry, offset, limit).also { resolvedRowCount += it.size } } val ref = QueueOrigin(entry.kind, entry.sourceId, entry.refId) - val anchored = keyCache[ref]?.let { keyedRows(entry, offset, limit, it) } + val preparingReset = currentCoroutineContext()[ResetPreparationContext] != null + val anchored = if (preparingReset) null else keyCache[ref]?.let { keyedRows(entry, offset, limit, it) } if (anchored != null) { keysetSeeks++ if (anchored.isNotEmpty()) remember(ref, offset, anchored) @@ -1351,7 +1371,7 @@ class QueueRepository( } offsetSeeks++ val fresh = resolverFor(entry.kind).rows(dao, entry, offset, limit) - if (fresh.isNotEmpty()) remember(ref, offset, fresh) + if (fresh.isNotEmpty() && !preparingReset) remember(ref, offset, fresh) resolvedRowCount += fresh.size return fresh } @@ -1424,6 +1444,7 @@ class QueueRepository( entry: QueueEntry, version: Long, ): Long { + if (currentCoroutineContext()[ResetPreparationContext] != null) return entry.resolvedLength() val key = entry.lengthKey(version) lengthCache[key]?.let { return it } val length = entry.resolvedLength() diff --git a/app/src/main/kotlin/com/subtracks/playback/PlaybackController.kt b/app/src/main/kotlin/com/subtracks/playback/PlaybackController.kt index a8022faa..a2708823 100644 --- a/app/src/main/kotlin/com/subtracks/playback/PlaybackController.kt +++ b/app/src/main/kotlin/com/subtracks/playback/PlaybackController.kt @@ -14,6 +14,7 @@ import com.subtracks.data.repo.QueueRepository import com.subtracks.data.repo.QueueRepository.QueueMutationScope import com.subtracks.data.repo.QueueReset import com.subtracks.data.repo.QueueResetItem +import com.subtracks.data.repo.QueueResetPreparation import com.subtracks.data.repo.QueueSnapshot import com.subtracks.data.repo.QueueWindowItem import com.subtracks.data.repo.Shuffle @@ -115,6 +116,14 @@ private data class PendingPlay( val behavior: SongTapBehavior?, ) +private data class PreparedWindow( + val items: List, + val actual: Long, + val startIndex: Int, + val autoplay: Boolean, + val startPositionMs: Long, +) + class PlaybackController( private val sourceRepository: SourceRepository, private val queueRepository: QueueRepository, @@ -166,7 +175,7 @@ class PlaybackController( init { queueRepository.setResetListener { reset -> withContext(playbackDispatcher) { - reconcileLibraryResetLocked(reset) + prepareLibraryResetLocked(reset) } } scope.launch { @@ -1209,42 +1218,58 @@ class PlaybackController( ) { val player = player ?: return val snapshot = snapshot ?: return + val prepared = prepareWindow(snapshot, position, autoplay, startPositionMs) + if (prepared == null) { + stopLocked() + return + } + publishWindow(player, snapshot, prepared) + } + + private suspend fun prepareWindow( + snapshot: QueueSnapshot, + position: Long, + autoplay: Boolean, + startPositionMs: Long, + ): PreparedWindow? { val requested = if (offline) { - downloadedAtOrAfter(snapshot, position) ?: downloadedAtOrBefore(snapshot, position - 1) ?: run { - stopLocked() - return - } + downloadedAtOrAfter(snapshot, position) ?: downloadedAtOrBefore(snapshot, position - 1) ?: return null } else { position } val raw = queueRepository.window(snapshot, requested, QUEUE_WINDOW_RADIUS) val window = if (offline) raw.filter { downloads.localUri(it.item.song.id) != null } else raw - if (window.isEmpty()) { - stopLocked() - return - } + if (window.isEmpty()) return null var startIndex = window.indexOfFirst { it.position == requested } if (startIndex < 0) { startIndex = window.indexOfFirst { it.position > requested } if (startIndex < 0) startIndex = window.lastIndex } val actual = window[startIndex].position - clearDetached() - windowPositions = LongArray(window.size) { window[it].position } - windowFiltered = offline - windowStart = window.first().position - windowEnd = window.last().position - lastPosition = actual if (actual != requested) queueRepository.setCursor(actual) + return PreparedWindow(window, actual, startIndex, autoplay, startPositionMs) + } + + private fun publishWindow( + player: PlayerHandle, + snapshot: QueueSnapshot, + prepared: PreparedWindow, + ) { + clearDetached() + windowPositions = LongArray(prepared.items.size) { prepared.items[it].position } + windowFiltered = offline + windowStart = prepared.items.first().position + windowEnd = prepared.items.last().position + lastPosition = prepared.actual updating = true - player.setWindow(window.map { it.playerItem(snapshot) }, startIndex, startPositionMs) + player.setWindow(prepared.items.map { it.playerItem(snapshot) }, prepared.startIndex, prepared.startPositionMs) updating = false - if (autoplay) { + if (prepared.autoplay) { player.prepare() player.play() } - refresh(actual) + refresh(prepared.actual) } private suspend fun reloadQuality() = @@ -1421,10 +1446,10 @@ class PlaybackController( lastPosition = target } - private suspend fun QueueMutationScope.reconcileLibraryResetLocked(reset: QueueReset) { - val player = player ?: return - val previous = snapshot ?: return - if (previous.version > reset.snapshot.version) return + private suspend fun QueueMutationScope.prepareLibraryResetLocked(reset: QueueReset): QueueResetPreparation { + val player = player ?: return QueueResetPreparation(reset.snapshot) + val previous = snapshot ?: return QueueResetPreparation(reset.snapshot) + if (previous.version > reset.snapshot.version) return QueueResetPreparation(reset.snapshot) var current = reset.point.current val actualItem = player.currentItem val actualId = actualItem?.id @@ -1447,45 +1472,72 @@ class PlaybackController( current?.copy(songId = actualId, position = actualPosition, upNext = upNext) } } - lastEdit = null - windowJob?.cancel() - endedHandled = false var updated = reset.snapshot - shuffleEnabled = updated.shuffled - layoutVersion++ - this@PlaybackController.snapshot = updated val target = current?.let { queueRepository.remapPosition(updated, it) } if (target != null) { - val anchor = reanchorFor(updated, target) - if (anchor.stale) return + val anchor = reanchorForReset(updated, target) if (anchor.changed) { - updated = snapshot() - this@PlaybackController.snapshot = updated + val flat = updated.flatContext(updated.contextPlay(target)) ?: updated.upNextAnchor + updated = updated.copy(upNextAnchor = flat) } val position = anchor.position.coerceIn(0, updated.size - 1) queueRepository.setCursor(position) if (detachedWindow && currentPosition() == null) { - rebuildDetachedSurvivors(updated, position) - } else { - clearDetached() - rebuildWindow(queueRepository.window(updated, position, QUEUE_WINDOW_RADIUS), position) + val window = prepareDetachedWindow(updated, position) + return QueueResetPreparation(updated, playbackDispatcher) { committed -> + beginLibraryResetPublication(committed) + publishDetachedSurvivors(player, committed, position, window) + refresh(position) + } + } + val raw = queueRepository.window(updated, position, QUEUE_WINDOW_RADIUS) + val ordered = if (offline) raw.filter { downloads.localUri(it.item.song.id) != null } else raw + val index = ordered.indexOfFirst { it.position == position } + if (index < 0) { + val prepared = prepareWindow(updated, position, player.playWhenReady, player.currentPositionMs) + return QueueResetPreparation(updated, playbackDispatcher) { committed -> + beginLibraryResetPublication(committed) + if (prepared == null) stopLocked() else publishWindow(player, committed, prepared) + refresh(position) + } + } + return QueueResetPreparation(updated, playbackDispatcher) { committed -> + beginLibraryResetPublication(committed) + publishRebuiltWindow(player, committed, ordered, index, position) + refresh(position) } - refresh(position) - return } if (player.itemCount == 0) { if (updated.size == 0L) { - stopLocked() - } else { - loadWindow(0, autoplay = false) + return QueueResetPreparation(updated, playbackDispatcher) { committed -> + beginLibraryResetPublication(committed) + stopLocked() + } + } + val prepared = prepareWindow(updated, 0, autoplay = false, startPositionMs = 0) + return QueueResetPreparation(updated, playbackDispatcher) { committed -> + beginLibraryResetPublication(committed) + if (prepared == null) stopLocked() else publishWindow(player, committed, prepared) } - return } val nearest = (current?.position ?: lastPosition ?: 0L).coerceIn(0, (updated.size - 1).coerceAtLeast(0)) - rebuildDetachedSurvivors(updated, nearest) - queueRepository.setCursor(lastPosition ?: 0L) - refresh(lastPosition) + val window = prepareDetachedWindow(updated, nearest) + queueRepository.setCursor(window.firstOrNull()?.position ?: 0L) + return QueueResetPreparation(updated, playbackDispatcher) { committed -> + beginLibraryResetPublication(committed) + publishDetachedSurvivors(player, committed, nearest, window) + refresh(lastPosition) + } + } + + private fun beginLibraryResetPublication(updated: QueueSnapshot) { + lastEdit = null + windowJob?.cancel() + endedHandled = false + shuffleEnabled = updated.shuffled + layoutVersion++ + snapshot = updated } private fun trimToCurrent(player: PlayerHandle) { @@ -1554,6 +1606,16 @@ class PlaybackController( loadWindow(position, autoplay = player.playWhenReady, startPositionMs = player.currentPositionMs) return } + publishRebuiltWindow(player, snapshot, ordered, index, position) + } + + private fun publishRebuiltWindow( + player: PlayerHandle, + snapshot: QueueSnapshot, + ordered: List, + index: Int, + position: Long, + ) { // A track downloaded while it was streaming still holds a stream URL; fix the current item's // source in place (without resetting the window) so an offline switch can play the local file. val current = ordered[index] @@ -1594,11 +1656,27 @@ class PlaybackController( position: Long, ) { val player = player ?: return + val window = prepareDetachedWindow(snapshot, position) + publishDetachedSurvivors(player, snapshot, position, window) + } + + private suspend fun prepareDetachedWindow( + snapshot: QueueSnapshot, + position: Long, + ): List { + val nearest = position.coerceIn(0, (snapshot.size - 1).coerceAtLeast(0)) + return queueRepository + .range(snapshot, nearest, (nearest + QUEUE_WINDOW_RADIUS).coerceAtMost(snapshot.size - 1)) + .filter { !offline || downloads.localUri(it.item.song.id) != null } + } + + private fun publishDetachedSurvivors( + player: PlayerHandle, + snapshot: QueueSnapshot, + position: Long, + window: List, + ) { val nearest = position.coerceIn(0, (snapshot.size - 1).coerceAtLeast(0)) - val window = - queueRepository - .range(snapshot, nearest, (nearest + QUEUE_WINDOW_RADIUS).coerceAtMost(snapshot.size - 1)) - .filter { !offline || downloads.localUri(it.item.song.id) != null } trimToCurrent(player) updating = true try { 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 63f4e274..c29d364e 100644 --- a/app/src/test/kotlin/com/subtracks/data/sync/SyncManagerTest.kt +++ b/app/src/test/kotlin/com/subtracks/data/sync/SyncManagerTest.kt @@ -20,6 +20,7 @@ import com.subtracks.data.prefs.fakeUserPreferences import com.subtracks.data.repo.DownloadRepository import com.subtracks.data.repo.NetworkServerActionSink import com.subtracks.data.repo.QueueRepository +import com.subtracks.data.repo.QueueResetPreparation import com.subtracks.data.repo.SourceRepository import com.subtracks.data.repo.replace import com.subtracks.data.repo.setShuffle @@ -288,7 +289,10 @@ class SyncManagerTest { val before = queueRepository.snapshot() val cacheVersion = queueRepository.libraryCacheVersion() val resetAttempts = AtomicInteger() - queueRepository.setResetListener { resetAttempts.incrementAndGet() } + queueRepository.setResetListener { reset -> + resetAttempts.incrementAndGet() + QueueResetPreparation(reset.snapshot) + } failPendingOriginClear() manager.requestSync() @@ -334,12 +338,15 @@ 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 before = queueRepository.snapshot() + val cacheVersion = queueRepository.libraryCacheVersion() val resetAttempted = CountDownLatch(1) val resetAttempts = AtomicInteger() - queueRepository.setResetListener { + queueRepository.setResetListener { reset -> queueRepository.setPosition(42_000L) resetAttempted.countDown() if (resetAttempts.incrementAndGet() == 1) throw IllegalStateException("reconciliation failed") + QueueResetPreparation(reset.snapshot) } manager.requestSync() @@ -356,6 +363,15 @@ class SyncManagerTest { .entry.rangeStart, ) assertTrue(db.queueDao().pendingSyncOrigins(1).isNotEmpty()) + assertEquals( + listOf(before.version, before.resetVersion, cacheVersion, 0L), + listOf( + queueRepository.contentVersion(), + queueRepository.snapshot().resetVersion, + queueRepository.libraryCacheVersion(), + queueRepository.cursorPositionMs(), + ), + ) manager.requestSync() withTimeout(TEST_TIMEOUT_MS) { manager.status.drop(1).first { it is SyncStatus.Success } } @@ -371,6 +387,58 @@ class SyncManagerTest { assertTrue(db.queueDao().pendingSyncOrigins(1).isEmpty()) } + @Test + fun aQueueWriteFailureAfterPendingClearRollsBackWithoutPublishing() = + runBlocking { + 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 before = queueRepository.snapshot() + val cacheVersion = queueRepository.libraryCacheVersion() + val resetAttempts = AtomicInteger() + queueRepository.setResetListener { reset -> + resetAttempts.incrementAndGet() + QueueResetPreparation(reset.snapshot) + } + failQueueResetInsert() + + manager.requestSync() + withTimeout(TEST_TIMEOUT_MS) { manager.status.first { it is SyncStatus.Failed } } + + assertEquals( + 0L, + queueRepository + .snapshot() + .entries + .single() + .entry.rangeStart, + ) + 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(), + ), + ) + + allowQueueResetInsert() + manager.requestSync() + withTimeout(TEST_TIMEOUT_MS) { manager.status.drop(1).first { it is SyncStatus.Success } } + + assertEquals(1, resetAttempts.get()) + assertNull( + queueRepository + .snapshot() + .entries + .single() + .entry.rangeStart, + ) + assertTrue(db.queueDao().pendingSyncOrigins(1).isEmpty()) + } + @Test fun aFailedSyncPreservesResolvedQueueShuffle() = runBlocking { @@ -527,6 +595,21 @@ class SyncManagerTest { } } + private suspend fun failQueueResetInsert() { + db.useWriterConnection { connection -> + connection.usePrepared( + "CREATE TRIGGER fail_queue_reset_insert BEFORE INSERT ON queue_entries " + + "BEGIN SELECT RAISE(FAIL, 'queue reset insert failed'); END", + ) { it.step() } + } + } + + private suspend fun allowQueueResetInsert() { + db.useWriterConnection { connection -> + connection.usePrepared("DROP TRIGGER fail_queue_reset_insert") { it.step() } + } + } + private fun healthyDispatcher() = object : Dispatcher() { override fun dispatch(request: RecordedRequest): MockResponse = diff --git a/app/src/test/kotlin/com/subtracks/playback/PlaybackControllerTest.kt b/app/src/test/kotlin/com/subtracks/playback/PlaybackControllerTest.kt index d47be08c..e5e386b7 100644 --- a/app/src/test/kotlin/com/subtracks/playback/PlaybackControllerTest.kt +++ b/app/src/test/kotlin/com/subtracks/playback/PlaybackControllerTest.kt @@ -3726,14 +3726,6 @@ class PlaybackControllerTest { controller.playPlaylist(1, "p1", 2) await { handle.currentItem?.id == "s3" && controller.state.value.isPlaying } val point = runBlocking { queues.captureResetPoint() } - val initialId = - runBlocking { - db - .queueDao() - .entries() - .single() - .id - } val dispatcherBlocked = CountDownLatch(1) val releaseDispatcher = CountDownLatch(1) controllerScope.launch { @@ -3745,22 +3737,10 @@ class PlaybackControllerTest { controller.seekTo(42_000L) setPlaylist("s1", "s2", "s4", "s5") val reset = - repoScope.launch { + repoScope.launch(start = CoroutineStart.UNDISPATCHED) { queues.resetChangedOrigins(setOf(QueueOrigin(QueueKind.Playlist, 1, "p1")), point) } - try { - await { - runBlocking { - db - .queueDao() - .entries() - .single() - .id != initialId - } - } - } finally { - releaseDispatcher.countDown() - } + releaseDispatcher.countDown() runBlocking { withTimeout(TEST_TIMEOUT_MS) { reset.join() } } Thread.sleep(200) @@ -3998,19 +3978,9 @@ class PlaybackControllerTest { assertTrue(dispatcherBlocked.await(TEST_TIMEOUT_MS, TimeUnit.MILLISECONDS)) handle.advanceTo(handle.currentIndex + 1) val reset = - repoScope.launch { + repoScope.launch(start = CoroutineStart.UNDISPATCHED) { queues.resetChangedOrigins(setOf(QueueOrigin(QueueKind.Playlist, 1, "p1")), point) } - await { - runBlocking { - db - .queueDao() - .upNextEntries() - .single() - .rangeStart == null - } - } - releaseDispatcher.countDown() runBlocking { withTimeout(TEST_TIMEOUT_MS) { reset.join() } } Thread.sleep(200)