From cc94dfc4584647bc319564b313358143dc63cb45 Mon Sep 17 00:00:00 2001 From: austinried <4966622+austinried@users.noreply.github.com> Date: Thu, 12 Nov 2026 13:38:00 +0000 Subject: [PATCH] downloads: reconcile visible queue reset baselines Consume captured reset points on the first collector emission and leave pending sources uninitialized during reconciliation. Build ordered baselines from queue-visible completed downloads and ignore pruned-row transitions while retaining all-row state and recovery. --- .../com/subtracks/data/db/DownloadDao.kt | 18 ++ .../subtracks/data/repo/DownloadRepository.kt | 67 ++++-- .../com/subtracks/data/db/DownloadDaoTest.kt | 76 +++++++ .../data/repo/DownloadRepositoryTest.kt | 202 ++++++++++++++++++ 4 files changed, 347 insertions(+), 16 deletions(-) create mode 100644 app/src/test/kotlin/com/subtracks/data/db/DownloadDaoTest.kt 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 a4d45784..70467ec4 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,24 @@ interface DownloadDao { @Query("SELECT * FROM song_downloads WHERE sourceId = :sourceId") fun downloads(sourceId: Long): Flow> + @Query( + "SELECT d.* FROM song_downloads d " + + "JOIN songs ON songs.sourceId = d.sourceId AND songs.id = d.songId " + + "WHERE d.sourceId = :sourceId AND d.status = 'Completed' " + + "ORDER BY d.downloadedAt DESC, songs.sortName COLLATE NOCASE, songs.id", + ) + suspend fun completedDownloads(sourceId: Long): List + + @Query( + "SELECT d.songId FROM song_downloads d " + + "JOIN songs ON songs.sourceId = d.sourceId AND songs.id = d.songId " + + "WHERE d.sourceId = :sourceId AND d.songId IN (:songIds)", + ) + suspend fun queueVisibleSongIds( + sourceId: Long, + songIds: Collection, + ): List + @Query("SELECT * FROM song_downloads WHERE status IN ('Queued', 'Running') AND (engineId IS NOT NULL OR sourceId = :sourceId)") fun activeDownloads(sourceId: Long): Flow> 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 5c80cedd..fd96c816 100644 --- a/app/src/main/kotlin/com/subtracks/data/repo/DownloadRepository.kt +++ b/app/src/main/kotlin/com/subtracks/data/repo/DownloadRepository.kt @@ -71,7 +71,8 @@ class DownloadRepository( private var started = false private val statesFlow = MutableStateFlow>(emptyMap()) - private val completedBySource = mutableMapOf>() + private val completedBySource = mutableMapOf>>() + private val completedRowsBySource = mutableMapOf>() private val completedBaselineMutex = Mutex() private val pendingResetPoints = mutableMapOf() private val attemptedBios = ConcurrentHashMap.newKeySet>() @@ -87,17 +88,33 @@ class DownloadRepository( downloadStates().collect { (sourceId, rows) -> statesFlow.value = rows if (sourceId != null) { - val completed = completedDownloads(rows) mutex.withLock { + val completed = completedDownloads(sourceId) + val completedRows = completedDownloadKeys(rows) 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 + 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, + ) + } + } + + previous != null && completed != previous && completedRows != previousRows -> { + queueRepository.resetChangedOrigins( + setOf(QueueOrigin(QueueKind.Downloaded, sourceId, "")), + resetPoint, + ) + } } + completedBySource[sourceId] = completed + completedRowsBySource[sourceId] = completedRows } } } @@ -125,16 +142,34 @@ class DownloadRepository( if (sourceId !in pendingResetPoints) pendingResetPoints[sourceId] = queueRepository.captureResetPoint() } + private suspend fun captureForReset( + sourceId: Long, + songIds: Collection, + ) { + if (songIds.isEmpty()) return + if (db.downloadDao().queueVisibleSongIds(sourceId, songIds).isNotEmpty()) { + captureForReset(sourceId) + } else { + completedBaselineMutex.withLock { + if (sourceId !in pendingResetPoints) completedBySource[sourceId] = completedDownloads(sourceId) + } + } + } + private suspend fun ensureCompletedBaselines(rows: List) = completedBaselineMutex.withLock { rows.groupBy { it.sourceId }.forEach { (sourceId, sourceRows) -> - if (sourceId !in completedBySource) { - completedBySource[sourceId] = completedDownloads(sourceRows.associateBy { it.songId }) + if (sourceId !in completedBySource && sourceId !in pendingResetPoints) { + completedBySource[sourceId] = completedDownloads(sourceId) + completedRowsBySource[sourceId] = completedDownloadKeys(sourceRows.associateBy { it.songId }) } } } - private fun completedDownloads(rows: Map): Map = + private suspend fun completedDownloads(sourceId: Long): List> = + db.downloadDao().completedDownloads(sourceId).map { it.songId to it.downloadedAt } + + private fun completedDownloadKeys(rows: Map): Map = rows.values .filter { it.status == DownloadStatus.Completed } .associate { it.songId to it.downloadedAt } @@ -394,7 +429,7 @@ class DownloadRepository( val engineIds = rows.mapNotNull { it.engineId } if (engineIds.isNotEmpty()) engine.cancel(engineIds) rows.groupBy { it.sourceId }.forEach { (sourceId, group) -> - if (group.any { it.status == DownloadStatus.Completed }) captureForReset(sourceId) + captureForReset(sourceId, group.filter { it.status == DownloadStatus.Completed }.map { it.songId }) group.forEach { file(sourceId, it.songId).delete() } group.map { it.songId }.chunked(DELETE_CHUNK).forEach { db.downloadDao().deleteSongs(sourceId, it) } if (group.any { it.status == DownloadStatus.Completed }) { @@ -416,7 +451,7 @@ class DownloadRepository( withContext(dispatcher) { mutex.withLock { val rows = db.downloadDao().all().filter { it.sourceId == sourceId } - if (rows.any { it.status == DownloadStatus.Completed }) captureForReset(sourceId) + captureForReset(sourceId, rows.filter { it.status == DownloadStatus.Completed }.map { it.songId }) rows.mapNotNull { it.engineId }.let(engine::cancel) db.downloadDao().deleteSource(sourceId) db.libraryDao().clearDownloadedArtistArt(sourceId) @@ -632,7 +667,7 @@ class DownloadRepository( songId: String, ): Boolean { val row = db.downloadDao().find(sourceId, songId) ?: return false - if (row.status == DownloadStatus.Completed) captureForReset(sourceId) + if (row.status == DownloadStatus.Completed) captureForReset(sourceId, listOf(songId)) row.engineId?.let { engine.cancel(listOf(it)) } file(sourceId, songId).delete() db.downloadDao().delete(sourceId, songId) @@ -718,7 +753,7 @@ class DownloadRepository( ) 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 (completedBefore != completedAfter) captureForReset(row.sourceId, listOf(row.songId)) if (updated != row) db.downloadDao().upsert(updated) if (updated.status != row.status && (updated.status == DownloadStatus.Completed || row.status == DownloadStatus.Completed) @@ -739,7 +774,7 @@ class DownloadRepository( row: SongDownload, error: DownloadError, ) { - if (row.status == DownloadStatus.Completed) captureForReset(row.sourceId) + if (row.status == DownloadStatus.Completed) captureForReset(row.sourceId, listOf(row.songId)) row.engineId?.let { engine.cancel(listOf(it)) } file(row.sourceId, row.songId).delete() db.downloadDao().upsert(row.copy(status = DownloadStatus.Failed, engineId = null, error = error)) diff --git a/app/src/test/kotlin/com/subtracks/data/db/DownloadDaoTest.kt b/app/src/test/kotlin/com/subtracks/data/db/DownloadDaoTest.kt new file mode 100644 index 00000000..e5001019 --- /dev/null +++ b/app/src/test/kotlin/com/subtracks/data/db/DownloadDaoTest.kt @@ -0,0 +1,76 @@ +package com.subtracks.data.db + +import androidx.test.ext.junit.runners.AndroidJUnit4 +import com.subtracks.data.model.DownloadStatus +import com.subtracks.data.model.Song +import com.subtracks.data.model.SongDownload +import com.subtracks.data.model.Source +import kotlinx.coroutines.test.runTest +import org.junit.After +import org.junit.Assert.assertEquals +import org.junit.Before +import org.junit.Test +import org.junit.runner.RunWith + +@RunWith(AndroidJUnit4::class) +class DownloadDaoTest { + private lateinit var db: SubtracksDatabase + + @Before + fun setUp() { + db = createTestDatabase() + } + + @After + fun tearDown() { + db.close() + } + + @Test + fun completedDownloadsUseTheDownloadedQueueVisibilityAndOrder() = + runTest { + val sourceId = + db.sourcesDao().insertSource( + Source(name = "source", address = "http://localhost", isActive = true, createdAt = 0), + ) + db.libraryDao().upsertSongs( + listOf( + song(sourceId, "s1", "Zulu"), + song(sourceId, "s2", "Alpha"), + song(sourceId, "s3", "Running"), + song(sourceId, "s4", "Pruned"), + ), + ) + db.downloadDao().upsert(SongDownload(sourceId, "s1", DownloadStatus.Completed, downloadedAt = 1)) + db.downloadDao().upsert(SongDownload(sourceId, "s2", DownloadStatus.Completed, downloadedAt = 1)) + db.downloadDao().upsert(SongDownload(sourceId, "s3", DownloadStatus.Running, downloadedAt = 2)) + db.downloadDao().upsert(SongDownload(sourceId, "s4", DownloadStatus.Completed, downloadedAt = 3)) + db.libraryDao().deleteSongs(sourceId, listOf("s4")) + + assertEquals(listOf("s2", "s1"), db.downloadDao().completedDownloads(sourceId).map { it.songId }) + assertEquals( + setOf("s1", "s2", "s3"), + db.downloadDao().queueVisibleSongIds(sourceId, listOf("s1", "s2", "s3", "s4")).toSet(), + ) + } + + private fun song( + sourceId: Long, + id: String, + sortName: String, + ) = Song( + sourceId = sourceId, + id = id, + albumId = "al1", + artistId = "ar1", + title = sortName, + sortName = sortName, + album = "Album", + artist = "Artist", + duration = 100, + track = 1, + disc = 1, + starred = null, + genre = null, + ) +} 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 2e15ce9a..34ad12a9 100644 --- a/app/src/test/kotlin/com/subtracks/data/repo/DownloadRepositoryTest.kt +++ b/app/src/test/kotlin/com/subtracks/data/repo/DownloadRepositoryTest.kt @@ -29,9 +29,11 @@ import com.subtracks.data.model.SubsonicSource import com.subtracks.data.model.artistImageKey import com.subtracks.data.model.coverArtKey import com.subtracks.data.prefs.fakeUserPreferences +import kotlinx.coroutines.CoroutineDispatcher import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.asCoroutineDispatcher import kotlinx.coroutines.flow.first import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking @@ -55,6 +57,7 @@ import java.io.File import java.io.IOException import java.util.concurrent.CopyOnWriteArrayList import java.util.concurrent.CountDownLatch +import java.util.concurrent.Executors import java.util.concurrent.TimeUnit @RunWith(AndroidJUnit4::class) @@ -1375,6 +1378,46 @@ class DownloadRepositoryTest { ) } + private fun useRepository( + scope: CoroutineScope, + dispatcher: CoroutineDispatcher, + ) { + repository = + DownloadRepository( + db, + sources, + engine, + dir, + artworkStore = artwork, + artworkFetcher = fetcher, + queueRepository = queues, + showMessage = { messages += it.resolve(resources) }, + dispatcher = dispatcher, + scope = scope, + ) + } + + private fun withBlockedCollector(block: (CountDownLatch) -> Unit) { + val releaseCollector = CountDownLatch(1) + val executor = Executors.newSingleThreadExecutor() + val dispatcher = executor.asCoroutineDispatcher() + val collectorScope = CoroutineScope(SupervisorJob() + dispatcher) + val collectorBlocked = CountDownLatch(1) + executor.execute { + collectorBlocked.countDown() + releaseCollector.await() + } + assertTrue(collectorBlocked.await(5, TimeUnit.SECONDS)) + useRepository(collectorScope, Dispatchers.Default) + try { + block(releaseCollector) + } finally { + releaseCollector.countDown() + cancelAndJoinBlocking(collectorScope) + dispatcher.close() + } + } + private fun requestedSongIds(): List = engine.requests.map { it.second.path.substringAfterLast('/') } private fun seedSongs(tracks: IntRange) { @@ -1704,6 +1747,165 @@ class DownloadRepositoryTest { } } + @Test + fun aCapturedDeletionBeforeTheFirstEmissionResetsAnEditedDownloadedQueue() = + runTest { + seedLibrary() + runBlocking { + db.downloadDao().upsert(SongDownload(1, "s1", DownloadStatus.Completed, downloadedAt = 1)) + db.downloadDao().upsert(SongDownload(1, "s2", DownloadStatus.Completed, downloadedAt = 2)) + queues.replace(listOf(queues.downloadedEntry(1).copy(rangeStart = 1))) + } + writeFile(1, "s1", "one") + writeFile(1, "s2", "two") + val beforeReset = runBlocking { queues.snapshot().resetVersion } + withBlockedCollector { releaseCollector -> + repository.start() + runBlocking { + repository.remove(1, "s2") + repository.reconcile() + } + releaseCollector.countDown() + runBlocking { repository.awaitLoaded() } + + val snapshot = runBlocking { queues.snapshot() } + assertEquals(beforeReset + 1, snapshot.resetVersion) + assertNull( + snapshot.entries + .single() + .entry.rangeStart, + ) + } + } + + @Test + fun aCapturedCompletionBeforeTheFirstEmissionResetsAnEditedDownloadedQueue() = + 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 -> + repository.start() + runBlocking { + repository.captureForReset(1) + db.downloadDao().upsert(SongDownload(1, "s2", DownloadStatus.Completed, downloadedAt = 2)) + repository.reconcile() + } + releaseCollector.countDown() + runBlocking { repository.awaitLoaded() } + + val snapshot = runBlocking { queues.snapshot() } + assertEquals(beforeReset + 1, snapshot.resetVersion) + assertNull( + snapshot.entries + .single() + .entry.rangeStart, + ) + assertEquals(2L, snapshot.size) + } + } + + @Test + fun aPrunedDownloadCompletingDoesNotResetAnEditedDownloadedQueue() = + runTest { + val beforeReset = preparePrunedDownload(DownloadStatus.Running) + writeFile(1, "s3", "three") + engine.complete(42, total = 10) + + runBlocking { repository.reconcile() } + await { repository.states().value["s3"]?.status == DownloadStatus.Completed } + Thread.sleep(200) + + val snapshot = runBlocking { queues.snapshot() } + assertEquals(beforeReset, snapshot.resetVersion) + assertEquals( + 1L, + snapshot.entries + .single() + .entry.rangeStart, + ) + } + + @Test + fun aPrunedCompletedDownloadFailingDoesNotResetAnEditedDownloadedQueue() = + runTest { + val beforeReset = preparePrunedDownload(DownloadStatus.Completed) + file(1, "s3").delete() + + runBlocking { repository.reconcile() } + await { repository.states().value["s3"]?.status == DownloadStatus.Failed } + Thread.sleep(200) + + val snapshot = runBlocking { queues.snapshot() } + assertEquals(beforeReset, snapshot.resetVersion) + assertEquals( + 1L, + snapshot.entries + .single() + .entry.rangeStart, + ) + } + + @Test + fun removingAPrunedCompletedDownloadDoesNotResetAnEditedDownloadedQueue() = + runTest { + val beforeReset = preparePrunedDownload(DownloadStatus.Completed) + + runBlocking { repository.remove(1, "s3") } + await { repository.states().value["s3"] == null } + Thread.sleep(200) + + val snapshot = runBlocking { queues.snapshot() } + assertEquals(beforeReset, snapshot.resetVersion) + assertEquals( + 1L, + snapshot.entries + .single() + .entry.rangeStart, + ) + } + + private fun preparePrunedDownload(status: DownloadStatus): Long { + seedLibrary() + seedSongs(3..3) + runBlocking { + db.downloadDao().upsert(SongDownload(1, "s1", DownloadStatus.Completed, downloadedAt = 1)) + db.downloadDao().upsert(SongDownload(1, "s2", DownloadStatus.Completed, downloadedAt = 2)) + db.downloadDao().upsert( + SongDownload( + 1, + "s3", + status, + engineId = 42L.takeIf { status == DownloadStatus.Running }, + downloadedAt = if (status == DownloadStatus.Completed) 3L else 0L, + ), + ) + } + writeFile(1, "s1", "one") + writeFile(1, "s2", "two") + writeFile(1, "s3", "three") + if (status == DownloadStatus.Running) engine.running(42, bytes = 1, total = 10) + repository.start() + await { + repository.localUri("s1") != null && + repository.localUri("s2") != null && + (status != DownloadStatus.Completed || repository.localUri("s3") != null) + } + return runBlocking { + repository.awaitLoaded() + repository.reconcile() + queues.replace(listOf(queues.downloadedEntry(1).copy(rangeStart = 1))) + db.libraryDao().deleteSongs(1, listOf("s3")) + queues.invalidateLibraryCache() + queues.snapshot().resetVersion + } + } + private fun seedPlaylist(songIds: List) { runBlocking { db.libraryDao().upsertPlaylists(listOf(Playlist(1, "pl1", "Playlist", null, null, songIds.size.toLong(), 0)))