From 4e51f906fe345bc0a39a58383cb3c1227af31c59 Mon Sep 17 00:00:00 2001 From: austinried <4966622+austinried@users.noreply.github.com> Date: Thu, 12 Nov 2026 10:15:53 +0000 Subject: [PATCH] sync: consume pending origins with queue resets --- .../com/subtracks/data/sync/SyncManager.kt | 25 ++--- .../subtracks/data/sync/SyncManagerTest.kt | 97 +++++++++++++++++++ 2 files changed, 107 insertions(+), 15 deletions(-) diff --git a/app/src/main/kotlin/com/subtracks/data/sync/SyncManager.kt b/app/src/main/kotlin/com/subtracks/data/sync/SyncManager.kt index 46597fe2..75e0362a 100644 --- a/app/src/main/kotlin/com/subtracks/data/sync/SyncManager.kt +++ b/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() - syncSourceId = source.id + val resetPoint = queueRepository.captureResetPoint() 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 } 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 077a8dfb..22bb20c5 100644 --- a/app/src/test/kotlin/com/subtracks/data/sync/SyncManagerTest.kt +++ b/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 =