downloads: preserve completed reset baselines

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

@@ -88,15 +88,19 @@ class DownloadRepository(
                 statesFlow.value = rows
                 if (sourceId != null) {
                     val completed = completedDownloads(rows)
+                    mutex.withLock {
                         completedBaselineMutex.withLock {
                             val previous = completedBySource[sourceId]
-                        if (previous != null && completed != previous) {
+                            if (previous == null) {
+                                completedBySource[sourceId] = completed
+                            } else if (completed != previous) {
                                 val point = pendingResetPoints.remove(sourceId) ?: resetPoint
                                 queueRepository.resetChangedOrigins(setOf(QueueOrigin(QueueKind.Downloaded, sourceId, "")), point)
                                 completedBySource[sourceId] = completed
                             }
                         }
                     }
+                }
                 resetPoint = queueRepository.captureResetPoint()
                 loaded.complete(Unit)
             }
@@ -712,14 +716,9 @@ class DownloadRepository(
                         row.downloadedAt
                     },
             )
-        if (updated != row &&
+        val completedBefore = row.takeIf { it.status == DownloadStatus.Completed }?.let { it.songId to it.downloadedAt }
-            (
+        val completedAfter = updated.takeIf { it.status == DownloadStatus.Completed }?.let { it.songId to it.downloadedAt }
-                updated.status == DownloadStatus.Completed || row.status == DownloadStatus.Completed ||
+        if (completedBefore != completedAfter) captureForReset(row.sourceId)
-                    updated.downloadedAt != row.downloadedAt
-            )
-        ) {
-            captureForReset(row.sourceId)
-        }
         if (updated != row) db.downloadDao().upsert(updated)
         if (updated.status != row.status &&
             (updated.status == DownloadStatus.Completed || row.status == DownloadStatus.Completed)

app/src/test/kotlin/com/subtracks/data/download/FakeDownloadEngine.kt

@@ -12,6 +12,7 @@ class FakeDownloadEngine : DownloadEngine {
     var enqueueDelayMs = 0L
     var failEnqueueFor: String? = null
     var failCancelOnEmpty = false
+    var beforeDownload: ((Long) -> Unit)? = null
 
     override fun enqueue(request: EngineRequest): Long {
         if (enqueueDelayMs > 0) Thread.sleep(enqueueDelayMs)
@@ -23,7 +24,10 @@ class FakeDownloadEngine : DownloadEngine {
         return id
     }
 
-    override fun download(id: Long): EngineDownload? = downloads[id]
+    override fun download(id: Long): EngineDownload? {
+        beforeDownload?.invoke(id)
+        return downloads[id]
+    }
 
     override fun cancel(ids: List<Long>) {
         if (ids.isEmpty() && failCancelOnEmpty) throw IllegalArgumentException("no ids")

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

@@ -54,6 +54,7 @@ import org.junit.runner.RunWith
 import java.io.File
 import java.io.IOException
 import java.util.concurrent.CopyOnWriteArrayList
+import java.util.concurrent.CountDownLatch
 import java.util.concurrent.TimeUnit
 
 @RunWith(AndroidJUnit4::class)
@@ -1558,27 +1559,124 @@ class DownloadRepositoryTest {
             )
         }
 
+    @Test
+    fun aFreshSourceResetsAnEditedDownloadedQueueOnTheSecondCompletion() =
+        runTest {
+            seedLibrary()
+            repository.start()
+            runBlocking {
+                repository.awaitLoaded()
+                repository.download(1, "s1")
+            }
+            var engineId = engine.requests.single().first
+            writeFile(1, "s1", "one")
+            engine.complete(engineId, total = 10)
+            await { repository.localUri("s1") != null }
+            runBlocking {
+                queues.replace(listOf(queues.downloadedEntry(1).copy(rangeStart = 0, rangeEnd = 0)))
+                repository.download(1, "s2")
+            }
+            val beforeReset = runBlocking { queues.snapshot().resetVersion }
+            engineId = engine.requests.last().first
+            writeFile(1, "s2", "two")
+            engine.complete(engineId, total = 10)
+
+            await { runBlocking { queues.snapshot().resetVersion > beforeReset } }
+
+            val snapshot = runBlocking { queues.snapshot() }
+            assertNull(
+                snapshot.entries
+                    .single()
+                    .entry.rangeStart,
+            )
+            assertEquals(2L, snapshot.size)
+        }
+
+    @Test
+    fun metadataOnlyUpdatesDoNotReuseAStaleResetPointForALaterCompletion() =
+        runTest {
+            seedLibrary()
+            seedSongs(0..0)
+            runBlocking {
+                db.downloadDao().upsert(
+                    SongDownload(1, "s2", DownloadStatus.Completed, engineId = 42, bytes = 10, total = 10, downloadedAt = 1),
+                )
+                db.downloadDao().upsert(SongDownload(1, "s1", DownloadStatus.Completed, bytes = 10, total = 10, downloadedAt = 2))
+            }
+            writeFile(1, "s1", "one")
+            writeFile(1, "s2", "two")
+            repository.start()
+            await { repository.localUri("s1") != null && repository.localUri("s2") != null }
+            runBlocking {
+                repository.awaitLoaded()
+                queues.replace(listOf(queues.downloadedEntry(1).copy(rangeStart = 0, rangeEnd = 1)))
+                queues.setCursor(0)
+            }
+            engine.complete(42, total = 20)
+            runBlocking { repository.reconcile() }
+            await { row(1, "s2")?.total == 20L }
+            runBlocking {
+                queues.setCursor(1)
+                repository.download(1, "s0")
+            }
+            val beforeReset = runBlocking { queues.snapshot().resetVersion }
+            val engineId = engine.requests.single().first
+            engine.complete(engineId, total = 10)
+            writeFile(1, "s0", "zero")
+            runBlocking { repository.reconcile() }
+
+            await { runBlocking { queues.snapshot().resetVersion > beforeReset } }
+
+            assertEquals(2L, runBlocking { queues.cursor() })
+        }
+
     @Test
     fun aMultiRowRecoveryKeepsTheResetPointFromBeforeTheWholeBatch() =
         runTest {
             seedLibrary()
-            seedSongs(3..3)
+            seedSongs(3..5)
             runBlocking {
-                db.downloadDao().upsert(SongDownload(1, "s3", DownloadStatus.Completed, downloadedAt = 1))
+                (1..5).forEach { track ->
-                queues.replace(listOf(queues.downloadedEntry(1)))
+                    val engineId =
-                queues.setCursor(0)
+                        if (track == 2) {
-                repository.captureForReset(1)
+                            102L
-                db.downloadDao().upsert(SongDownload(1, "s1", DownloadStatus.Completed, downloadedAt = 2))
+                        } else if (track == 3) {
-                repository.captureForReset(1)
+                            103L
-                db.downloadDao().upsert(SongDownload(1, "s2", DownloadStatus.Completed, downloadedAt = 3))
+                        } else {
-                repository.captureForReset(1)
+                            null
+                        }
+                    db.downloadDao().upsert(
+                        SongDownload(1, "s$track", DownloadStatus.Completed, engineId = engineId, downloadedAt = (6 - track).toLong()),
+                    )
+                }
+            }
+            (1..5).forEach { writeFile(1, "s$it", "audio$it") }
+            repository.start()
+            await { (1..5).all { repository.localUri("s$it") != null } }
+            runBlocking {
+                repository.awaitLoaded()
+                repository.reconcile()
+            }
+            val beforeReset =
+                runBlocking {
+                    queues.replace(listOf(queues.downloadedEntry(1).copy(rangeStart = 0, rangeEnd = 4)))
+                    queues.setCursor(2)
+                    queues.snapshot().resetVersion
+                }
+            val firstReset = CountDownLatch(1)
+            queues.setResetListener { firstReset.countDown() }
+            file(1, "s2").delete()
+            file(1, "s3").delete()
+            engine.beforeDownload = { engineId ->
+                if (engineId == 103L) firstReset.await(1, TimeUnit.SECONDS)
             }
-            val field = DownloadRepository::class.java.getDeclaredField("pendingResetPoints")
-            field.isAccessible = true
-            val pending = field.get(repository) as Map<*, *>
-            val point = pending[1L] as QueueResetPoint
 
-            assertEquals("s3", point.current?.songId)
+            runBlocking { repository.reconcile() }
+
+            await { runBlocking { queues.snapshot().resetVersion > beforeReset && queues.snapshot().size == 3L } }
+            val (snapshot, cursor) = runBlocking { queues.snapshot() to queues.cursor() }
+            assertEquals(2L, cursor)
+            assertEquals("s5", runBlocking { queues.itemAt(snapshot, cursor)?.song?.id })
         }
 
     @Test