From e4d97f7aba165d2f3a03b62011d9e43c0d760d59 Mon Sep 17 00:00:00 2001 From: austinried <4966622+austinried@users.noreply.github.com> Date: Sat, 14 Nov 2026 01:56:04 +0000 Subject: [PATCH] downloads: harden queue reset reconciliation Refresh per-source rows under the reconciliation mutex before reset decisions and chunk queue-visibility checks to respect SQLite bind limits. Cover a blocked pre-completion emission and source deletion at the bundled SQLite variable limit. --- .../com/subtracks/data/db/DownloadDao.kt | 3 + .../subtracks/data/repo/DownloadRepository.kt | 12 ++- .../data/repo/DownloadRepositoryTest.kt | 101 ++++++++++++++++++ 3 files changed, 113 insertions(+), 3 deletions(-) diff --git a/app/src/main/kotlin/com/subtracks/data/db/DownloadDao.kt b/app/src/main/kotlin/com/subtracks/data/db/DownloadDao.kt index 70467ec4..b0605839 100644 --- a/app/src/main/kotlin/com/subtracks/data/db/DownloadDao.kt +++ b/app/src/main/kotlin/com/subtracks/data/db/DownloadDao.kt @@ -56,6 +56,9 @@ interface DownloadDao { @Query("SELECT * FROM song_downloads WHERE sourceId = :sourceId") fun downloads(sourceId: Long): Flow> + @Query("SELECT * FROM song_downloads WHERE sourceId = :sourceId") + suspend fun downloadsOnce(sourceId: Long): List + @Query( "SELECT d.* FROM song_downloads d " + "JOIN songs ON songs.sourceId = d.sourceId AND songs.id = d.songId " + 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 fd96c816..a2e7cc90 100644 --- a/app/src/main/kotlin/com/subtracks/data/repo/DownloadRepository.kt +++ b/app/src/main/kotlin/com/subtracks/data/repo/DownloadRepository.kt @@ -85,10 +85,12 @@ class DownloadRepository( File(downloadsDir, NO_MEDIA).createNewFile() scope.launch { var resetPoint = queueRepository.captureResetPoint() - downloadStates().collect { (sourceId, rows) -> - statesFlow.value = rows + 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 { @@ -147,7 +149,11 @@ class DownloadRepository( songIds: Collection, ) { if (songIds.isEmpty()) return - if (db.downloadDao().queueVisibleSongIds(sourceId, songIds).isNotEmpty()) { + val queueVisible = + songIds.asSequence().chunked(DELETE_CHUNK).any { ids -> + db.downloadDao().queueVisibleSongIds(sourceId, ids).isNotEmpty() + } + if (queueVisible) { captureForReset(sourceId) } else { completedBaselineMutex.withLock { 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 34ad12a9..589c2c84 100644 --- a/app/src/test/kotlin/com/subtracks/data/repo/DownloadRepositoryTest.kt +++ b/app/src/test/kotlin/com/subtracks/data/repo/DownloadRepositoryTest.kt @@ -3,6 +3,9 @@ package com.subtracks.data.repo import android.content.Context import android.content.res.Resources import android.net.Uri +import androidx.room3.immediateTransaction +import androidx.room3.useReaderConnection +import androidx.room3.useWriterConnection import androidx.test.core.app.ApplicationProvider import androidx.test.ext.junit.runners.AndroidJUnit4 import com.subtracks.awaitUntil @@ -59,6 +62,7 @@ import java.util.concurrent.CopyOnWriteArrayList import java.util.concurrent.CountDownLatch import java.util.concurrent.Executors import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicReference @RunWith(AndroidJUnit4::class) class DownloadRepositoryTest { @@ -369,6 +373,39 @@ class DownloadRepositoryTest { ) } + @Test + fun removingASourceBeyondTheSqliteBindLimitDeletesAllDownloads() = + runTest { + val count = runBlocking { sqliteVariableLimit() } + runBlocking { + db.sourcesDao().upsertSource(Source(1, "source", "http://localhost", isActive = true, createdAt = 0)) + db.useWriterConnection { connection -> + connection.immediateTransaction { + connection.usePrepared( + "WITH RECURSIVE ids(id) AS (SELECT 0 UNION ALL SELECT id + 1 FROM ids WHERE id + 1 < ?) " + + "INSERT INTO songs (sourceId, id, title) SELECT 1, 's' || id, 'Song' FROM ids", + ) { statement -> + statement.bindLong(1, count.toLong()) + statement.step() + } + connection.usePrepared( + "WITH RECURSIVE ids(id) AS (SELECT 0 UNION ALL SELECT id + 1 FROM ids WHERE id + 1 < ?) " + + "INSERT INTO song_downloads (sourceId, songId, status, downloadedAt) " + + "SELECT 1, 's' || id, 'Completed', id FROM ids", + ) { statement -> + statement.bindLong(1, count.toLong()) + statement.step() + } + } + } + } + + assertEquals(count, allRows().size) + runBlocking { repository.removeSource(1) } + + assertTrue(allRows().isEmpty()) + } + @Test fun deleteAllCapturesTheCurrentQueuePositionBeforeRemovingDownloads() = runTest { @@ -1810,6 +1847,58 @@ class DownloadRepositoryTest { } } + @Test + fun aResetAfterAStaleEmissionObservesTheCompletedLocalFile() = + runTest { + seedLibrary() + runBlocking { + db.downloadDao().upsert(SongDownload(1, "s1", DownloadStatus.Running, engineId = 42, total = 10)) + db.downloadDao().upsert(SongDownload(1, "s2", DownloadStatus.Completed, downloadedAt = 2)) + } + writeFile(1, "s1", "one") + writeFile(1, "s2", "two") + engine.running(42, bytes = 1, total = 10) + repository.start() + await { + repository.states().value["s1"]?.status == DownloadStatus.Running && repository.localUri("s2") != null + } + runBlocking { + repository.awaitLoaded() + repository.reconcile() + queues.replace(listOf(queues.downloadedEntry(1).copy(rangeStart = 0, rangeEnd = 0))) + } + val reconciliationStarted = CountDownLatch(1) + val finishReconciliation = CountDownLatch(1) + val reconciliationTimedOut = AtomicReference(false) + engine.beforeDownload = { engineId -> + if (engineId == 42L) { + reconciliationStarted.countDown() + reconciliationTimedOut.set(!finishReconciliation.await(5, TimeUnit.SECONDS)) + } + } + val reset = CountDownLatch(1) + val localUriAtReset = AtomicReference("not reset") + queues.setResetListener { + localUriAtReset.set(repository.localUri("s1")) + reset.countDown() + } + val reconciling = launch(Dispatchers.IO) { repository.reconcile() } + assertTrue(reconciliationStarted.await(5, TimeUnit.SECONDS)) + runBlocking { + val running = db.downloadDao().find(1, "s1")!! + db.downloadDao().upsert(running.copy(bytes = 2)) + } + await { repository.states().value["s1"]?.bytes == 2L } + assertFalse(reconciliationTimedOut.get()) + + engine.complete(42, total = 10) + finishReconciliation.countDown() + reconciling.join() + + assertTrue(reset.await(5, TimeUnit.SECONDS)) + assertNotNull(localUriAtReset.get()) + } + @Test fun aPrunedDownloadCompletingDoesNotResetAnEditedDownloadedQueue() = runTest { @@ -2003,6 +2092,18 @@ class DownloadRepositoryTest { private fun allRows(): List = runBlocking { db.downloadDao().all() } + private suspend fun sqliteVariableLimit(): Int = + db.useReaderConnection { connection -> + connection.usePrepared("PRAGMA compile_options") { statement -> + var limit = 32_766 + while (statement.step()) { + val option = statement.getText(0) + if (option.startsWith("MAX_VARIABLE_NUMBER=")) limit = option.substringAfter('=').toInt() + } + limit + } + } + private fun writeFile( sourceId: Long, songId: String,