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 8daf1811..8b30c504 100644 --- a/app/src/main/kotlin/com/subtracks/data/repo/QueueRepository.kt +++ b/app/src/main/kotlin/com/subtracks/data/repo/QueueRepository.kt @@ -11,8 +11,11 @@ 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 com.subtracks.log.Log +import kotlinx.coroutines.Job import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.currentCoroutineContext +import kotlinx.coroutines.ensureActive import kotlinx.coroutines.flow.first import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock @@ -193,6 +196,7 @@ data class QueueResetPreparation( val snapshot: QueueSnapshot, val publicationContext: CoroutineContext = EmptyCoroutineContext, val publish: QueueRepository.QueueMutationScope.(QueueSnapshot) -> Unit = {}, + val recover: QueueRepository.QueueMutationScope.(QueueSnapshot) -> Unit = {}, ) private class CursorLockContext : AbstractCoroutineContextElement(CursorLockContext) { @@ -633,7 +637,9 @@ class QueueRepository( val before = snapshot() 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 }) { + val matchingOrigin = + before.entries.any { it.entry.origin in changed } || before.upNext.any { it.entry.origin in changed } + if (matchingOrigin) { writeResetLocked(before, entries, upNext, point, pendingOriginSourceId) return } @@ -653,81 +659,89 @@ class QueueRepository( point: QueueResetPoint, pendingOriginSourceId: Long? = null, ) { - val preparation = - withCursorLock { - 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 callerJob = currentCoroutineContext()[Job] + withContext(NonCancellable) { + val preparation = + withCursorLock { + 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, - ), - ) - resetListener?.invoke(this@writeResetLocked, QueueReset(point, updated)) - ?: QueueResetPreparation(updated) + 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)) + .also { callerJob?.ensureActive() } + } } } } + invalidateLibraryCache() + queueVersion.incrementAndGet() + resetVersion.incrementAndGet() + val published = preparation.snapshot.copy(version = queueVersion.get(), resetVersion = resetVersion.get()) + runCatching { + withContext(preparation.publicationContext) { + preparation.publish(this@writeResetLocked, published) + } + }.onFailure { failure -> + Log.w("queue", "reset publication failed; recovering from committed state", failure) + runCatching { preparation.recover(this@writeResetLocked, published) } } - 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) } } diff --git a/app/src/main/kotlin/com/subtracks/playback/PlaybackController.kt b/app/src/main/kotlin/com/subtracks/playback/PlaybackController.kt index a2708823..df52bb25 100644 --- a/app/src/main/kotlin/com/subtracks/playback/PlaybackController.kt +++ b/app/src/main/kotlin/com/subtracks/playback/PlaybackController.kt @@ -49,6 +49,24 @@ data class QueueItem( val resetItem: QueueResetItem? = null, ) +private data class PlayerGeneration( + val itemId: String?, + val index: Int, + val resetItem: QueueResetItem?, +) { + companion object { + fun from(player: PlayerHandle): PlayerGeneration { + val item = player.currentItem + return PlayerGeneration(item?.id, player.currentIndex, item?.resetItem) + } + } + + fun matches(player: PlayerHandle): Boolean { + val item = player.currentItem + return item?.id == itemId && player.currentIndex == index && item?.resetItem == resetItem + } +} + enum class RepeatMode { Off, All, One } data class PlaybackState( @@ -1450,6 +1468,7 @@ class PlaybackController( val player = player ?: return QueueResetPreparation(reset.snapshot) val previous = snapshot ?: return QueueResetPreparation(reset.snapshot) if (previous.version > reset.snapshot.version) return QueueResetPreparation(reset.snapshot) + val generation = PlayerGeneration.from(player) var current = reset.point.current val actualItem = player.currentItem val actualId = actualItem?.id @@ -1484,7 +1503,7 @@ class PlaybackController( queueRepository.setCursor(position) if (detachedWindow && currentPosition() == null) { val window = prepareDetachedWindow(updated, position) - return QueueResetPreparation(updated, playbackDispatcher) { committed -> + return resetPreparation(updated, generation) { committed -> beginLibraryResetPublication(committed) publishDetachedSurvivors(player, committed, position, window) refresh(position) @@ -1495,28 +1514,32 @@ class PlaybackController( val index = ordered.indexOfFirst { it.position == position } if (index < 0) { val prepared = prepareWindow(updated, position, player.playWhenReady, player.currentPositionMs) - return QueueResetPreparation(updated, playbackDispatcher) { committed -> + return resetPreparation(updated, generation) { committed -> beginLibraryResetPublication(committed) if (prepared == null) stopLocked() else publishWindow(player, committed, prepared) refresh(position) } } - return QueueResetPreparation(updated, playbackDispatcher) { committed -> + return resetPreparation(updated, generation) { committed -> beginLibraryResetPublication(committed) - publishRebuiltWindow(player, committed, ordered, index, position) + publishWindow( + player, + committed, + PreparedWindow(ordered, position, index, player.playWhenReady, player.currentPositionMs), + ) refresh(position) } } if (player.itemCount == 0) { if (updated.size == 0L) { - return QueueResetPreparation(updated, playbackDispatcher) { committed -> + return resetPreparation(updated, generation) { committed -> beginLibraryResetPublication(committed) stopLocked() } } val prepared = prepareWindow(updated, 0, autoplay = false, startPositionMs = 0) - return QueueResetPreparation(updated, playbackDispatcher) { committed -> + return resetPreparation(updated, generation) { committed -> beginLibraryResetPublication(committed) if (prepared == null) stopLocked() else publishWindow(player, committed, prepared) } @@ -1524,13 +1547,46 @@ class PlaybackController( val nearest = (current?.position ?: lastPosition ?: 0L).coerceIn(0, (updated.size - 1).coerceAtLeast(0)) val window = prepareDetachedWindow(updated, nearest) queueRepository.setCursor(window.firstOrNull()?.position ?: 0L) - return QueueResetPreparation(updated, playbackDispatcher) { committed -> + return resetPreparation(updated, generation) { committed -> beginLibraryResetPublication(committed) publishDetachedSurvivors(player, committed, nearest, window) refresh(lastPosition) } } + private fun resetPreparation( + updated: QueueSnapshot, + generation: PlayerGeneration, + publish: QueueMutationScope.(QueueSnapshot) -> Unit, + ) = QueueResetPreparation( + updated, + playbackDispatcher, + publish = { committed -> + if (generation.matches(requireNotNull(player))) { + publish(committed) + } else { + recoverLibraryReset() + } + }, + recover = { recoverLibraryReset() }, + ) + + private fun recoverLibraryReset() { + scope.launch { + queueRepository.withMutationLock { + val player = player ?: return@withMutationLock + val autoplay = player.playWhenReady + val latest = queueRepository.snapshot() + stopLocked() + this@PlaybackController.snapshot = latest + if (latest.size > 0L) { + val position = queueRepository.cursor().coerceIn(0, latest.size - 1) + loadWindow(position, autoplay = autoplay, startPositionMs = 0) + } + } + } + } + private fun beginLibraryResetPublication(updated: QueueSnapshot) { lastEdit = null windowJob?.cancel() 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 94023449..9dd313d4 100644 --- a/app/src/test/kotlin/com/subtracks/data/repo/QueueRepositoryTest.kt +++ b/app/src/test/kotlin/com/subtracks/data/repo/QueueRepositoryTest.kt @@ -474,6 +474,25 @@ class QueueRepositoryTest { repository.removeAt(snapshot, 0) assertEquals(listOf("s2", "s3"), resolveAll(repository.snapshot())) + fun aMatchingOriginStillPublishesWhenItsEntriesDoNotChange() = + runTest { + seedLibrary() + db.libraryDao().upsertSongs(listOf(song("s0", "al1", track = 0, album = "First Album"))) + setPlaylist("p1", "s1", "s2", "s3") + repository.snapshotAfter(listOf(repository.playlistEntry(1, "p1"))) + repository.setCursor(1) + val point = repository.captureResetPoint() + var resets = 0 + repository.setResetListener { + resets++ + QueueResetPreparation(it.snapshot) + } + + setPlaylist("p1", "s0", "s1", "s2", "s3") + repository.resetChangedOrigins(setOf(QueueOrigin(QueueKind.Playlist, 1, "p1")), point) + + assertEquals(1, resets) + assertEquals(2L, repository.cursor()) } @Test 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 c29d364e..9178c49f 100644 --- a/app/src/test/kotlin/com/subtracks/data/sync/SyncManagerTest.kt +++ b/app/src/test/kotlin/com/subtracks/data/sync/SyncManagerTest.kt @@ -387,6 +387,42 @@ class SyncManagerTest { assertTrue(db.queueDao().pendingSyncOrigins(1).isEmpty()) } + @Test + fun aPublicationFailureRecoversFromCommittedReset() = + 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 recoveries = AtomicInteger() + queueRepository.setResetListener { reset -> + QueueResetPreparation( + reset.snapshot, + publish = { throw IllegalStateException("publish failed") }, + recover = { published -> + assertEquals(before.version + 1, published.version) + assertEquals(before.resetVersion + 1, published.resetVersion) + recoveries.incrementAndGet() + }, + ) + } + + manager.requestSync() + withTimeout(TEST_TIMEOUT_MS) { manager.status.first { it is SyncStatus.Success } } + + assertEquals(1, recoveries.get()) + assertNull( + queueRepository + .snapshot() + .entries + .single() + .entry.rangeStart, + ) + assertTrue(db.queueDao().pendingSyncOrigins(1).isEmpty()) + assertEquals(before.version + 1, queueRepository.contentVersion()) + assertEquals(before.resetVersion + 1, queueRepository.snapshot().resetVersion) + } + @Test fun aQueueWriteFailureAfterPendingClearRollsBackWithoutPublishing() = runBlocking { diff --git a/app/src/test/kotlin/com/subtracks/playback/FakePlayer.kt b/app/src/test/kotlin/com/subtracks/playback/FakePlayer.kt index cb7e9f80..276128ca 100644 --- a/app/src/test/kotlin/com/subtracks/playback/FakePlayer.kt +++ b/app/src/test/kotlin/com/subtracks/playback/FakePlayer.kt @@ -48,6 +48,9 @@ class FakePlayerHandle : PlayerHandle { @Volatile var beforeRemoveFirst: (() -> Unit)? = null + @Volatile + var beforeCurrentItem: (() -> Unit)? = null + @Volatile var roundTripItemsThroughMedia3 = false @@ -86,6 +89,7 @@ class FakePlayerHandle : PlayerHandle { override val currentItem: QueueItem? get() { + beforeCurrentItem?.invoke() checkThread() return items.getOrNull(index) } diff --git a/app/src/test/kotlin/com/subtracks/playback/PlaybackControllerTest.kt b/app/src/test/kotlin/com/subtracks/playback/PlaybackControllerTest.kt index e5e386b7..b4c62731 100644 --- a/app/src/test/kotlin/com/subtracks/playback/PlaybackControllerTest.kt +++ b/app/src/test/kotlin/com/subtracks/playback/PlaybackControllerTest.kt @@ -3598,6 +3598,36 @@ class PlaybackControllerTest { assertEquals(listOf("s0", "s1", "s2", "s3", "s4", "s5"), handle.items.map { it.id }) } + @Test + fun aTransitionBetweenResetPreparationAndPublicationRebuildsFromCommittedState() { + seedPlaylist(5) + seedSong("s0", "al1") + controller.playPlaylist(1, "p1", 2) + await { handle.currentItem?.id == "s3" && controller.state.value.isPlaying } + val point = runBlocking { queues.captureResetPoint() } + var currentItemReads = 0 + handle.beforeCurrentItem = { + currentItemReads++ + if (currentItemReads == 2) { + handle.beforeCurrentItem = null + handle.advanceTo(3) + } + } + handle.beforeRemoveFirst = { + handle.beforeRemoveFirst = null + handle.advanceTo(3) + } + + setPlaylist("s0", "s1", "s2", "s3", "s4", "s5") + runBlocking { queues.resetChangedOrigins(setOf(QueueOrigin(QueueKind.Playlist, 1, "p1")), point) } + + await { + handle.currentItem?.id == "s4" && controller.state.value.position == 3L && + handle.items.size == 5 && handle.items.distinctBy { it.id }.size == 5 + } + assertEquals(listOf("s1", "s2", "s3", "s4", "s5"), handle.items.map { it.id }) + } + @Test fun aSourceResetLetsTheRemovedCurrentFinishBeforeTheClampedTrack() { seedPlaylist(5)