sync: consume pending origins with queue resets

app/src/main/kotlin/com/subtracks/data/sync/SyncManager.kt

@@ -1,12 +1,13 @@
 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
 import com.subtracks.data.model.QueueOrigin
 import com.subtracks.data.prefs.UserPreferences
 import com.subtracks.data.repo.QueueRepository
-import com.subtracks.data.repo.QueueResetPoint
 import com.subtracks.data.repo.SourceRepository
 import com.subtracks.data.source.ServerActionSink
 import com.subtracks.data.source.subsonic.SubsonicException
@@ -83,26 +84,28 @@ class SyncManager(
         runCatching { preferences.setLastSyncAt(System.currentTimeMillis()) }
             .onFailure { Log.w("sync", "could not record the sync attempt", it) }
         val startedAt = System.nanoTime()
-        var summary: SyncSummary? = null
-        var resetPoint: QueueResetPoint? = null
-        var syncSourceId: Long? = null
         val result =
             try {
                 val source = sourceRepository.activeMusicSource() ?: throw NoServerException()
                 Log.i("sync", "started (source ${source.id})")
                 serverActions.flush()
-                resetPoint = queueRepository.captureResetPoint()
+                val resetPoint = queueRepository.captureResetPoint()
-                syncSourceId = source.id
                 val service = SyncService(db, source)
                 val completed = service.sync()
                 val pending = db.queueDao().pendingSyncOrigins(source.id).map { QueueOrigin(it.kind, it.sourceId, it.refId) }
-                summary = completed.copy(changedOrigins = pending.toSet() + completed.changedOrigins)
+                val changedOrigins = pending.toSet() + completed.changedOrigins
                 Log.i(
                     "sync",
                     "finished in ${elapsedMs(startedAt)}ms: " +
                         "${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)
+                    }
+                }
                 SyncStatus.Success
             } catch (cancellation: CancellationException) {
                 throw cancellation
@@ -122,14 +125,6 @@ class SyncManager(
                 if (!silent) showMessage(message)
                 SyncStatus.Failed(message)
             }
-        if (result is SyncStatus.Success) {
-            val completed = summary
-            val point = resetPoint
-            if (completed != null && point != null) {
-                queueRepository.resetChangedOrigins(completed.changedOrigins, point)
-                db.queueDao().clearPendingSyncOrigins(sourceId = checkNotNull(syncSourceId))
-            }
-        }
         _status.value = result
     }
 

app/src/test/kotlin/com/subtracks/data/sync/SyncManagerTest.kt

@@ -2,6 +2,7 @@ package com.subtracks.data.sync
 
 import android.content.Context
 import android.content.res.Resources
+import androidx.room3.useWriterConnection
 import androidx.test.core.app.ApplicationProvider
 import androidx.test.ext.junit.runners.AndroidJUnit4
 import com.subtracks.TEST_TIMEOUT_MS
@@ -278,6 +279,87 @@ class SyncManagerTest {
             assertTrue(db.queueDao().pendingSyncOrigins(1).isEmpty())
         }
 
+    @Test
+    fun aPendingClearFailureRollsBackTheQueueResetAndCanBeRetried() =
+        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 resetAttempted = CountDownLatch(1)
+            queueRepository.setResetListener { resetAttempted.countDown() }
+            failPendingOriginClear()
+
+            manager.requestSync()
+            assertTrue(resetAttempted.await(10, TimeUnit.SECONDS))
+            awaitUntil("waiting for the pending-origin clear failure") { manager.status.value !is SyncStatus.Running }
+
+            assertEquals(
+                0L,
+                queueRepository
+                    .snapshot()
+                    .entries
+                    .single()
+                    .entry.rangeStart,
+            )
+            assertTrue(manager.status.value is SyncStatus.Failed)
+            assertTrue(db.queueDao().pendingSyncOrigins(1).isNotEmpty())
+
+            allowPendingOriginClear()
+            manager.requestSync()
+            withTimeout(TEST_TIMEOUT_MS) { manager.status.drop(1).first { it is SyncStatus.Success } }
+
+            assertNull(
+                queueRepository
+                    .snapshot()
+                    .entries
+                    .single()
+                    .entry.rangeStart,
+            )
+            assertTrue(db.queueDao().pendingSyncOrigins(1).isEmpty())
+        }
+
+    @Test
+    fun aResetListenerFailureIsReportedAndCanBeRetried() =
+        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 resetAttempted = CountDownLatch(1)
+            val resetAttempts = AtomicInteger()
+            queueRepository.setResetListener {
+                resetAttempted.countDown()
+                if (resetAttempts.incrementAndGet() == 1) throw IllegalStateException("reconciliation failed")
+            }
+
+            manager.requestSync()
+            assertTrue(resetAttempted.await(10, TimeUnit.SECONDS))
+            awaitUntil("waiting for the queue reset listener failure") { manager.status.value !is SyncStatus.Running }
+
+            assertTrue(manager.status.value is SyncStatus.Failed)
+            assertEquals(
+                0L,
+                queueRepository
+                    .snapshot()
+                    .entries
+                    .single()
+                    .entry.rangeStart,
+            )
+            assertTrue(db.queueDao().pendingSyncOrigins(1).isNotEmpty())
+
+            manager.requestSync()
+            withTimeout(TEST_TIMEOUT_MS) { manager.status.drop(1).first { it is SyncStatus.Success } }
+
+            assertEquals(2, resetAttempts.get())
+            assertNull(
+                queueRepository
+                    .snapshot()
+                    .entries
+                    .single()
+                    .entry.rangeStart,
+            )
+            assertTrue(db.queueDao().pendingSyncOrigins(1).isEmpty())
+        }
+
     @Test
     fun aFailedSyncPreservesResolvedQueueShuffle() =
         runBlocking {
@@ -419,6 +501,21 @@ class SyncManagerTest {
             assertEquals(2, artistRequests.get())
         }
 
+    private suspend fun failPendingOriginClear() {
+        db.useWriterConnection { connection ->
+            connection.usePrepared(
+                "CREATE TRIGGER fail_pending_origin_clear BEFORE DELETE ON pending_sync_origins " +
+                    "BEGIN SELECT RAISE(FAIL, 'pending-origin clear failed'); END",
+            ) { it.step() }
+        }
+    }
+
+    private suspend fun allowPendingOriginClear() {
+        db.useWriterConnection { connection ->
+            connection.usePrepared("DROP TRIGGER fail_pending_origin_clear") { it.step() }
+        }
+    }
+
     private fun healthyDispatcher() =
         object : Dispatcher() {
             override fun dispatch(request: RecordedRequest): MockResponse =