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.

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<List<SongDownload>>
 
+    @Query("SELECT * FROM song_downloads WHERE sourceId = :sourceId")
+    suspend fun downloadsOnce(sourceId: Long): List<SongDownload>
+
     @Query(
         "SELECT d.* FROM song_downloads d " +
             "JOIN songs ON songs.sourceId = d.sourceId AND songs.id = d.songId " +

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) ->
+            downloadStates().collect { (sourceId, emittedRows) ->
-                statesFlow.value = rows
+                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<String>,
     ) {
         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 {

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<String?>("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<SongDownload> = 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,