diff --git a/app/src/main/kotlin/com/subtracks/data/repo/DownloadRepository.kt b/app/src/main/kotlin/com/subtracks/data/repo/DownloadRepository.kt index 37136298..f02e73e0 100644 --- a/app/src/main/kotlin/com/subtracks/data/repo/DownloadRepository.kt +++ b/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>?, + sourceId: Long?, + rows: Map, +): 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() - downloadStates().collect { (sourceId, emittedRows) -> - statesFlow.value = emittedRows - if (sourceId != null) { - mutex.withLock { - val rows = db.downloadDao().downloadsOnce(sourceId).associateBy { it.songId } - 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 -> { - pendingResetPoints.remove(sourceId) - if (previous == null || completed != previous) { - queueRepository.resetChangedOrigins( - setOf(QueueOrigin(QueueKind.Downloaded, sourceId, "")), - pending, - ) - } - } + var resetPoint: QueueResetPoint? = null + var reconciled: Pair>? = null + while (true) { + try { + if (resetPoint == null) resetPoint = queueRepository.captureResetPoint() + downloadStates().collect { (sourceId, emittedRows) -> + statesFlow.value = emittedRows + val unchanged = sameResetRelevantEmission(reconciled, sourceId, emittedRows) + val pending = + unchanged && sourceId != null && + completedBaselineMutex.withLock { sourceId in pendingResetPoints } + if (unchanged && !pending) { + loaded.complete(Unit) + return@collect + } + var rows = emittedRows + if (sourceId != null) { + mutex.withLock { + rows = db.downloadDao().downloadsOnce(sourceId).associateBy { it.songId } + 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 -> { - queueRepository.resetChangedOrigins( - setOf(QueueOrigin(QueueKind.Downloaded, sourceId, "")), - resetPoint, - ) + previous != null && completed != previous && completedRows != previousRows -> { + queueRepository.resetChangedOrigins( + setOf(QueueOrigin(QueueKind.Downloaded, sourceId, "")), + 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()) } } diff --git a/app/src/test/kotlin/com/subtracks/data/repo/DownloadRepositoryTest.kt b/app/src/test/kotlin/com/subtracks/data/repo/DownloadRepositoryTest.kt index 130fe74e..8b916090 100644 --- a/app/src/test/kotlin/com/subtracks/data/repo/DownloadRepositoryTest.kt +++ b/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() private lateinit var resources: Resources private val requestedArt = CopyOnWriteArrayList() + private val auxiliaryDbs = mutableListOf() private val artFetchedAfterTheEngineRequest = CopyOnWriteArrayList() private var failArtworkFetch = false private val failingArtUrls = CopyOnWriteArrayList() @@ -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) { + 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() + 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() + 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()) + 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()) + 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)