downloads: recover reset observation and skip progress work

app/src/main/kotlin/com/subtracks/data/repo/DownloadRepository.kt

@@ -51,6 +51,19 @@ import kotlinx.coroutines.withTimeoutOrNull
 import java.io.File
 import java.util.concurrent.ConcurrentHashMap
 
+private fun SongDownload.sameResetRelevantState(other: SongDownload): Boolean =
+    sourceId == other.sourceId && songId == other.songId && status == other.status && engineId == other.engineId &&
+        total == other.total && error == other.error && downloadedAt == other.downloadedAt
+
+private fun sameResetRelevantEmission(
+    previous: Pair<Long?, Map<String, SongDownload>>?,
+    sourceId: Long?,
+    rows: Map<String, SongDownload>,
+): Boolean {
+    if (previous == null || previous.first != sourceId || previous.second.size != rows.size) return false
+    return previous.second.all { (songId, row) -> rows[songId]?.let { row.sameResetRelevantState(it) } == true }
+}
+
 class DownloadRepository(
     private val db: SubtracksDatabase,
     private val sourceRepository: SourceRepository,
@@ -84,44 +97,66 @@ class DownloadRepository(
         downloadsDir.mkdirs()
         File(downloadsDir, NO_MEDIA).createNewFile()
         scope.launch {
-            var resetPoint = queueRepository.captureResetPoint()
+            var resetPoint: QueueResetPoint? = null
-            downloadStates().collect { (sourceId, emittedRows) ->
+            var reconciled: Pair<Long?, Map<String, SongDownload>>? = null
-                statesFlow.value = emittedRows
+            while (true) {
-                if (sourceId != null) {
+                try {
-                    mutex.withLock {
+                    if (resetPoint == null) resetPoint = queueRepository.captureResetPoint()
-                        val rows = db.downloadDao().downloadsOnce(sourceId).associateBy { it.songId }
+                    downloadStates().collect { (sourceId, emittedRows) ->
-                        statesFlow.value = rows
+                        statesFlow.value = emittedRows
-                        val completed = completedDownloads(sourceId)
+                        val unchanged = sameResetRelevantEmission(reconciled, sourceId, emittedRows)
-                        val completedRows = completedDownloadKeys(rows)
+                        val pending =
-                        completedBaselineMutex.withLock {
+                            unchanged && sourceId != null &&
-                            val previous = completedBySource[sourceId]
+                                completedBaselineMutex.withLock { sourceId in pendingResetPoints }
-                            val previousRows = completedRowsBySource[sourceId]
+                        if (unchanged && !pending) {
-                            val pending = pendingResetPoints[sourceId]
+                            loaded.complete(Unit)
-                            when {
+                            return@collect
-                                pending != null -> {
+                        }
-                                    pendingResetPoints.remove(sourceId)
+                        var rows = emittedRows
-                                    if (previous == null || completed != previous) {
+                        if (sourceId != null) {
-                                        queueRepository.resetChangedOrigins(
+                            mutex.withLock {
-                                            setOf(QueueOrigin(QueueKind.Downloaded, sourceId, "")),
+                                rows = db.downloadDao().downloadsOnce(sourceId).associateBy { it.songId }
-                                            pending,
+                                statesFlow.value = rows
-                                        )
+                                val completed = completedDownloads(sourceId)
-                                    }
+                                val completedRows = completedDownloadKeys(rows)
-                                }
+                                completedBaselineMutex.withLock {
+                                    val previous = completedBySource[sourceId]
+                                    val previousRows = completedRowsBySource[sourceId]
+                                    val pending = pendingResetPoints[sourceId]
+                                    when {
+                                        pending != null -> {
+                                            if (previous == null || completed != previous) {
+                                                queueRepository.resetChangedOrigins(
+                                                    setOf(QueueOrigin(QueueKind.Downloaded, sourceId, "")),
+                                                    pending,
+                                                )
+                                            }
+                                            pendingResetPoints.remove(sourceId)
+                                        }
 
-                                previous != null && completed != previous && completedRows != previousRows -> {
+                                        previous != null && completed != previous && completedRows != previousRows -> {
-                                    queueRepository.resetChangedOrigins(
+                                            queueRepository.resetChangedOrigins(
-                                        setOf(QueueOrigin(QueueKind.Downloaded, sourceId, "")),
+                                                setOf(QueueOrigin(QueueKind.Downloaded, sourceId, "")),
-                                        resetPoint,
+                                                resetPoint!!,
-                                    )
+                                            )
+                                        }
+                                    }
+                                    completedBySource[sourceId] = completed
+                                    completedRowsBySource[sourceId] = completedRows
                                 }
                             }
-                            completedBySource[sourceId] = completed
-                            completedRowsBySource[sourceId] = completedRows
                         }
+                        resetPoint = queueRepository.captureResetPoint()
+                        reconciled = sourceId to rows
+                        loaded.complete(Unit)
                     }
+                } catch (e: CancellationException) {
+                    throw e
+                } catch (failure: Exception) {
+                    Log.w("download", "state reconciliation failed; retrying", failure)
+                    loaded.complete(Unit)
+                    delay(POLL_MS)
                 }
-                resetPoint = queueRepository.captureResetPoint()
-                loaded.complete(Unit)
             }
         }
         scope.launch {
@@ -298,7 +333,6 @@ class DownloadRepository(
         val ids = songIds.toHashSet()
         mutex.withLock {
             removeRows(db.downloadDao().all().filter { it.sourceId == sourceId && it.songId in ids })
-            sweep(db.downloadDao().all())
         }
     }
 

app/src/test/kotlin/com/subtracks/data/repo/DownloadRepositoryTest.kt

@@ -6,10 +6,13 @@ import android.net.Uri
 import androidx.room3.immediateTransaction
 import androidx.room3.useReaderConnection
 import androidx.room3.useWriterConnection
+import androidx.sqlite.driver.bundled.BundledSQLiteDriver
 import androidx.test.core.app.ApplicationProvider
 import androidx.test.ext.junit.runners.AndroidJUnit4
 import com.subtracks.awaitUntil
 import com.subtracks.cancelAndJoinBlocking
+import com.subtracks.data.db.LoggingSQLiteDriver
+import com.subtracks.data.db.SqlExecution
 import com.subtracks.data.db.SubtracksDatabase
 import com.subtracks.data.db.createTestDatabase
 import com.subtracks.data.download.ArtworkFetcher
@@ -41,6 +44,7 @@ import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.test.runTest
+import kotlinx.coroutines.withTimeout
 import okhttp3.OkHttpClient
 import okhttp3.mockwebserver.Dispatcher
 import okhttp3.mockwebserver.MockResponse
@@ -58,6 +62,7 @@ import org.junit.Test
 import org.junit.runner.RunWith
 import java.io.File
 import java.io.IOException
+import java.util.Collections
 import java.util.concurrent.CopyOnWriteArrayList
 import java.util.concurrent.CountDownLatch
 import java.util.concurrent.Executors
@@ -78,6 +83,7 @@ class DownloadRepositoryTest {
     private val messages = CopyOnWriteArrayList<String>()
     private lateinit var resources: Resources
     private val requestedArt = CopyOnWriteArrayList<String>()
+    private val auxiliaryDbs = mutableListOf<SubtracksDatabase>()
     private val artFetchedAfterTheEngineRequest = CopyOnWriteArrayList<Boolean>()
     private var failArtworkFetch = false
     private val failingArtUrls = CopyOnWriteArrayList<String>()
@@ -117,6 +123,7 @@ class DownloadRepositoryTest {
     @After
     fun tearDown() {
         cancelAndJoinBlocking(scope)
+        auxiliaryDbs.forEach { it.close() }
         db.close()
         dir.deleteRecursively()
     }
@@ -1415,6 +1422,14 @@ class DownloadRepositoryTest {
             )
         }
 
+    private fun useLoggingDatabase(executions: MutableList<SqlExecution>) {
+        auxiliaryDbs += db
+        db = createTestDatabase(LoggingSQLiteDriver(BundledSQLiteDriver()) { executions += it })
+        sources = SourceRepository(db, OkHttpClient(), prefs, artwork, scope = scope)
+        queues = QueueRepository(db)
+        useRepository(scope, Dispatchers.Default)
+    }
+
     private fun useRepository(
         scope: CoroutineScope,
         dispatcher: CoroutineDispatcher,
@@ -1963,6 +1978,145 @@ class DownloadRepositoryTest {
             )
         }
 
+    @Test
+    fun aTransientResetFailureIsRetriedWithTheOriginalResetPoint() =
+        runTest {
+            seedLibrary()
+            runBlocking {
+                db.downloadDao().upsert(SongDownload(1, "s1", DownloadStatus.Completed, downloadedAt = 1))
+                db.downloadDao().upsert(SongDownload(1, "s2", DownloadStatus.Completed, downloadedAt = 2))
+            }
+            writeFile(1, "s1", "one")
+            writeFile(1, "s2", "two")
+            repository.start()
+            runBlocking {
+                repository.awaitLoaded()
+                repository.reconcile()
+                queues.replace(listOf(queues.downloadedEntry(1).copy(rangeStart = 0, rangeEnd = 1)))
+                queues.setCursor(1)
+            }
+            val beforeReset = runBlocking { queues.snapshot().resetVersion }
+            val attempts = CopyOnWriteArrayList<QueueResetPoint>()
+            val secondAttempt = CountDownLatch(1)
+            queues.setResetListener { reset ->
+                attempts += reset.point
+                if (attempts.size == 1) throw IllegalStateException("reset failed")
+                secondAttempt.countDown()
+                QueueResetPreparation(reset.snapshot)
+            }
+
+            runBlocking { repository.remove(1, "s2") }
+
+            assertTrue(secondAttempt.await(5, TimeUnit.SECONDS))
+            await { runBlocking { queues.snapshot().resetVersion > beforeReset } }
+            assertEquals(2, attempts.size)
+            assertEquals(attempts[0], attempts[1])
+            val snapshot = runBlocking { queues.snapshot() }
+            assertNull(
+                snapshot.entries
+                    .single()
+                    .entry.rangeStart,
+            )
+            assertEquals(1L, snapshot.size)
+            assertEquals(0L, runBlocking { queues.cursor() })
+        }
+
+    @Test
+    fun aFirstEmissionResetFailureCompletesLoadingAndRetries() =
+        runTest {
+            seedLibrary()
+            runBlocking {
+                db.downloadDao().upsert(SongDownload(1, "s1", DownloadStatus.Completed, downloadedAt = 1))
+                queues.replace(listOf(queues.downloadedEntry(1).copy(rangeStart = 0, rangeEnd = 0)))
+            }
+            writeFile(1, "s1", "one")
+            writeFile(1, "s2", "two")
+            val beforeReset = runBlocking { queues.snapshot().resetVersion }
+            withBlockedCollector { releaseCollector ->
+                val attempts = CopyOnWriteArrayList<QueueResetPoint>()
+                val secondAttempt = CountDownLatch(1)
+                queues.setResetListener { reset ->
+                    attempts += reset.point
+                    if (attempts.size == 1) throw IllegalStateException("reset failed")
+                    secondAttempt.countDown()
+                    QueueResetPreparation(reset.snapshot)
+                }
+                repository.start()
+                runBlocking {
+                    repository.captureForReset(1)
+                    db.downloadDao().upsert(SongDownload(1, "s2", DownloadStatus.Completed, downloadedAt = 2))
+                    repository.reconcile()
+                }
+                releaseCollector.countDown()
+
+                runBlocking { withTimeout(5_000) { repository.awaitLoaded() } }
+
+                assertTrue(secondAttempt.await(5, TimeUnit.SECONDS))
+                await { runBlocking { queues.snapshot().resetVersion > beforeReset } }
+                assertEquals(2, attempts.size)
+                assertEquals(attempts[0], attempts[1])
+                assertNull(
+                    runBlocking {
+                        queues
+                            .snapshot()
+                            .entries
+                            .single()
+                            .entry.rangeStart
+                    },
+                )
+            }
+        }
+
+    @Test
+    fun byteOnlyProgressSkipsDownloadResetQueries() =
+        runTest {
+            val executions = Collections.synchronizedList(mutableListOf<SqlExecution>())
+            useLoggingDatabase(executions)
+            seedLibrary()
+            runBlocking {
+                db.downloadDao().upsert(
+                    SongDownload(1, "s1", DownloadStatus.Completed, bytes = 1, total = 10, downloadedAt = 1),
+                )
+            }
+            writeFile(1, "s1", "one")
+            repository.start()
+            runBlocking { repository.awaitLoaded() }
+            await { repository.localUri("s1") != null }
+            synchronized(executions) { executions.clear() }
+
+            runBlocking {
+                val row = db.downloadDao().find(1, "s1")!!
+                db.downloadDao().upsert(row.copy(bytes = 2))
+            }
+            await { repository.states().value["s1"]?.bytes == 2L }
+            Thread.sleep(300)
+
+            val queries = synchronized(executions) { executions.toList() }
+            assertEquals(1, queries.count { it.sql == "SELECT * FROM song_downloads WHERE sourceId = ?" })
+            assertTrue(queries.none { it.sql.contains("ORDER BY d.downloadedAt DESC") })
+        }
+
+    @Test
+    fun bulkRemoveSweepsArtworkOnce() =
+        runTest {
+            val executions = Collections.synchronizedList(mutableListOf<SqlExecution>())
+            useLoggingDatabase(executions)
+            seedLibrary()
+            runBlocking {
+                db.downloadDao().upsert(SongDownload(1, "s1", DownloadStatus.Completed, downloadedAt = 1))
+                db.downloadDao().upsert(SongDownload(1, "s2", DownloadStatus.Completed, downloadedAt = 2))
+            }
+            writeFile(1, "s1", "one")
+            writeFile(1, "s2", "two")
+            synchronized(executions) { executions.clear() }
+
+            runBlocking { repository.remove(1, listOf("s1", "s2")) }
+
+            val queries = synchronized(executions) { executions.toList() }
+            assertEquals(1, queries.count { it.sql.contains("AS albumRow") })
+            assertEquals(1, queries.count { it.sql.startsWith("SELECT DISTINCT ps.sourceId AS sourceId") })
+        }
+
     private fun preparePrunedDownload(status: DownloadStatus): Long {
         seedLibrary()
         seedSongs(3..3)