From 4e72b1f38d1b0f6423705247a0b0007c4d896ab0 Mon Sep 17 00:00:00 2001 From: austinried <4966622+austinried@users.noreply.github.com> Date: Thu, 12 Nov 2026 10:12:53 +0000 Subject: [PATCH] downloads: preserve completed reset baselines --- .../subtracks/data/repo/DownloadRepository.kt | 27 ++-- .../data/download/FakeDownloadEngine.kt | 6 +- .../data/repo/DownloadRepositoryTest.kt | 126 ++++++++++++++++-- 3 files changed, 130 insertions(+), 29 deletions(-) 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 f4debf21..5c80cedd 100644 --- a/app/src/main/kotlin/com/subtracks/data/repo/DownloadRepository.kt +++ b/app/src/main/kotlin/com/subtracks/data/repo/DownloadRepository.kt @@ -88,12 +88,16 @@ class DownloadRepository( statesFlow.value = rows if (sourceId != null) { val completed = completedDownloads(rows) - completedBaselineMutex.withLock { - val previous = completedBySource[sourceId] - if (previous != null && completed != previous) { - val point = pendingResetPoints.remove(sourceId) ?: resetPoint - queueRepository.resetChangedOrigins(setOf(QueueOrigin(QueueKind.Downloaded, sourceId, "")), point) - completedBySource[sourceId] = completed + mutex.withLock { + completedBaselineMutex.withLock { + val previous = completedBySource[sourceId] + 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 + } } } } @@ -712,14 +716,9 @@ class DownloadRepository( row.downloadedAt }, ) - if (updated != row && - ( - updated.status == DownloadStatus.Completed || row.status == DownloadStatus.Completed || - updated.downloadedAt != row.downloadedAt - ) - ) { - captureForReset(row.sourceId) - } + 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 } + if (completedBefore != completedAfter) captureForReset(row.sourceId) if (updated != row) db.downloadDao().upsert(updated) if (updated.status != row.status && (updated.status == DownloadStatus.Completed || row.status == DownloadStatus.Completed) diff --git a/app/src/test/kotlin/com/subtracks/data/download/FakeDownloadEngine.kt b/app/src/test/kotlin/com/subtracks/data/download/FakeDownloadEngine.kt index 894e2248..97bac086 100644 --- a/app/src/test/kotlin/com/subtracks/data/download/FakeDownloadEngine.kt +++ b/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) { if (ids.isEmpty() && failCancelOnEmpty) throw IllegalArgumentException("no ids") 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 e54a73bb..2e15ce9a 100644 --- a/app/src/test/kotlin/com/subtracks/data/repo/DownloadRepositoryTest.kt +++ b/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)) - queues.replace(listOf(queues.downloadedEntry(1))) - queues.setCursor(0) - repository.captureForReset(1) - db.downloadDao().upsert(SongDownload(1, "s1", DownloadStatus.Completed, downloadedAt = 2)) - repository.captureForReset(1) - db.downloadDao().upsert(SongDownload(1, "s2", DownloadStatus.Completed, downloadedAt = 3)) - repository.captureForReset(1) + (1..5).forEach { track -> + val engineId = + if (track == 2) { + 102L + } else if (track == 3) { + 103L + } else { + 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