sync: make queue reset publication recoverable

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)
         }
     }
 

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()

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

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 {

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)
         }

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)