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.

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 =

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)