From ae41b0e66aee569702e2f897c54477efdb38238c Mon Sep 17 00:00:00 2001 From: austinried <4966622+austinried@users.noreply.github.com> Date: Sun, 27 Sep 2026 06:35:54 +0000 Subject: [PATCH] publish the download map before signalling it is loaded Second review round on the downloads block: - `loaded` was completed upstream of the state flow that publishes the map, so `awaitLoaded()` could return while the map was still the initial empty one and the first window after a cold start would stream a track that is already on disk. The collector now assigns the map before completing, and the map is a plain state flow instead of a lazy eager `stateIn` with a bare touch to start it. - A failed download now deletes its partial file, in both the engine-failure and the recovery paths. A failed row used to protect its own partial bytes from the sweep, and the UI offers no delete for it. - A queued row is left queued when the download URL cannot be resolved yet, instead of being failed. At launch the repository's reconcile can run before the source repository has built its client, and failing there turned a resumable download into a dead one. - The poller only considers the active source's rows, so a queued row belonging to another source no longer keeps it polling the database every second. - `SourceRepository.downloadUri` takes the expected source id and returns null unless it is the active one, so a download cannot be enqueued against the previous server's URL during a source switch. - Tests: a stream with no content length still gets its declared length through the production data source factory (the http half of the chain, which the first round left uncovered), and an engine failure is asserted to leave no partial file. The engine's status mapping now runs as a plain unit test. --- .../subtracks/data/download/DownloadEngine.kt | 1 - .../data/download/SystemDownloadEngine.kt | 3 +- .../subtracks/data/repo/DownloadRepository.kt | 40 +++++++++---------- .../subtracks/data/repo/SourceRepository.kt | 7 +++- .../data/download/SystemDownloadEngineTest.kt | 3 -- .../data/repo/DownloadRepositoryTest.kt | 6 ++- .../subtracks/playback/MediaDataSourceTest.kt | 25 +++++++++++- 7 files changed, 55 insertions(+), 30 deletions(-) diff --git a/app/src/main/kotlin/com/subtracks/data/download/DownloadEngine.kt b/app/src/main/kotlin/com/subtracks/data/download/DownloadEngine.kt index 32d59e2b..cf66938e 100644 --- a/app/src/main/kotlin/com/subtracks/data/download/DownloadEngine.kt +++ b/app/src/main/kotlin/com/subtracks/data/download/DownloadEngine.kt @@ -4,7 +4,6 @@ data class EngineRequest( val uri: String, val path: String, val title: String, - val allowMetered: Boolean = false, ) enum class EngineStatus { Pending, Running, Completed, Failed } diff --git a/app/src/main/kotlin/com/subtracks/data/download/SystemDownloadEngine.kt b/app/src/main/kotlin/com/subtracks/data/download/SystemDownloadEngine.kt index 9915d002..26419e34 100644 --- a/app/src/main/kotlin/com/subtracks/data/download/SystemDownloadEngine.kt +++ b/app/src/main/kotlin/com/subtracks/data/download/SystemDownloadEngine.kt @@ -15,7 +15,8 @@ class SystemDownloadEngine( .setTitle(request.title) .setDestinationInExternalFilesDir(context, Environment.DIRECTORY_MUSIC, request.path) .setNotificationVisibility(DownloadManager.Request.VISIBILITY_VISIBLE) - .setAllowedOverMetered(request.allowMetered) + // Downloads wait for an unmetered network until the preference for them exists. + .setAllowedOverMetered(false) return manager().enqueue(download) } 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 6f6d50ba..6e794b2a 100644 --- a/app/src/main/kotlin/com/subtracks/data/repo/DownloadRepository.kt +++ b/app/src/main/kotlin/com/subtracks/data/repo/DownloadRepository.kt @@ -17,14 +17,12 @@ import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.cancel import kotlinx.coroutines.delay import kotlinx.coroutines.flow.Flow -import kotlinx.coroutines.flow.SharingStarted +import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.distinctUntilChanged import kotlinx.coroutines.flow.flatMapLatest import kotlinx.coroutines.flow.flowOf import kotlinx.coroutines.flow.map -import kotlinx.coroutines.flow.onEach -import kotlinx.coroutines.flow.stateIn import kotlinx.coroutines.launch import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock @@ -47,18 +45,19 @@ class DownloadRepository( @Volatile private var started = false - private val sharedStates: StateFlow> by lazy { - downloadStates() - .onEach { loaded.complete(Unit) } - .stateIn(scope, SharingStarted.Eagerly, emptyMap()) - } + private val statesFlow = MutableStateFlow>(emptyMap()) fun start() { if (started) return started = true downloadsDir.mkdirs() File(downloadsDir, NO_MEDIA).createNewFile() - sharedStates + scope.launch { + downloadStates().collect { rows -> + statesFlow.value = rows + loaded.complete(Unit) + } + } scope.launch { sourceRepository.activeSourceId().distinctUntilChanged().collect { reconcile() } } @@ -68,16 +67,14 @@ class DownloadRepository( scope.cancel() } - fun states(): StateFlow> = sharedStates + fun states(): StateFlow> = statesFlow suspend fun awaitLoaded() { - if (!started) return - states() - loaded.await() + if (started) loaded.await() } fun localUri(songId: String): String? = - sharedStates.value[songId] + statesFlow.value[songId] ?.takeIf { it.status == DownloadStatus.Completed } ?.let { file(it.sourceId, it.songId) } ?.takeIf { it.exists() } @@ -95,9 +92,9 @@ class DownloadRepository( val existing = db.downloadDao().find(sourceId, songId) if (existing != null && existing.status != DownloadStatus.Failed) return@withLock existing?.engineId?.let(engine::cancel) - val url = sourceRepository.downloadUri(songId) + val url = sourceRepository.downloadUri(sourceId, songId) if (url == null) { - showMessage("Can't download: the server address is unavailable") + showMessage(NO_ADDRESS) return@withLock } enqueue(sourceId, songId, url) @@ -157,7 +154,7 @@ class DownloadRepository( val title = db.libraryDao().songOnce(sourceId, songId)?.title ?: songId val engineId = engine.enqueue( - EngineRequest(uri = url, path = enginePath(sourceId, songId), title = title, allowMetered = false), + EngineRequest(uri = url, path = enginePath(sourceId, songId), title = title), ) db.downloadDao().upsert( SongDownload(sourceId = sourceId, songId = songId, status = DownloadStatus.Queued, engineId = engineId), @@ -173,8 +170,8 @@ class DownloadRepository( DownloadStatus.Queued -> { if (sourceRepository.activeSourceIdOnce() != row.sourceId) return - val url = sourceRepository.downloadUri(row.songId) - if (url == null) markFailed(row, NO_ADDRESS) else enqueue(row.sourceId, row.songId, url) + val url = sourceRepository.downloadUri(row.sourceId, row.songId) ?: return + enqueue(row.sourceId, row.songId, url) } DownloadStatus.Running -> { @@ -198,6 +195,7 @@ class DownloadRepository( EngineStatus.Completed -> if (file(row.sourceId, row.songId).exists()) DownloadStatus.Completed else DownloadStatus.Failed EngineStatus.Failed -> DownloadStatus.Failed } + if (status == DownloadStatus.Failed) file(row.sourceId, row.songId).delete() db.downloadDao().upsert( row.copy( status = status, @@ -213,6 +211,7 @@ class DownloadRepository( message: String, ) { row.engineId?.let(engine::cancel) + file(row.sourceId, row.songId).delete() db.downloadDao().upsert(row.copy(status = DownloadStatus.Failed, engineId = null, error = message)) } @@ -223,7 +222,8 @@ class DownloadRepository( while (true) { delay(POLL_MS) mutex.withLock { - val active = db.downloadDao().all().filter { it.status.isActive } + val sourceId = sourceRepository.activeSourceIdOnce() + val active = db.downloadDao().all().filter { it.status.isActive && it.sourceId == sourceId } if (active.isEmpty()) return@launch active.forEach { row -> val state = row.engineId?.let(engine::download) diff --git a/app/src/main/kotlin/com/subtracks/data/repo/SourceRepository.kt b/app/src/main/kotlin/com/subtracks/data/repo/SourceRepository.kt index 0483f9f0..81b5c432 100644 --- a/app/src/main/kotlin/com/subtracks/data/repo/SourceRepository.kt +++ b/app/src/main/kotlin/com/subtracks/data/repo/SourceRepository.kt @@ -38,6 +38,8 @@ class SourceRepository( @Volatile private var active: SubsonicMusicSource? = null + + @Volatile private var activeSourceId: Long? = null private val _quality = MutableStateFlow(StreamQuality()) @@ -94,7 +96,10 @@ class SourceRepository( return uri + streamLengthSuffix(length) } - fun downloadUri(songId: String): String? = active?.downloadUri(songId)?.toString() + fun downloadUri( + sourceId: Long, + songId: String, + ): String? = if (activeSourceId == sourceId) active?.downloadUri(songId)?.toString() else null suspend fun addSource( name: String, diff --git a/app/src/test/kotlin/com/subtracks/data/download/SystemDownloadEngineTest.kt b/app/src/test/kotlin/com/subtracks/data/download/SystemDownloadEngineTest.kt index bf478e81..ba4e90f4 100644 --- a/app/src/test/kotlin/com/subtracks/data/download/SystemDownloadEngineTest.kt +++ b/app/src/test/kotlin/com/subtracks/data/download/SystemDownloadEngineTest.kt @@ -1,12 +1,9 @@ package com.subtracks.data.download import android.app.DownloadManager -import androidx.test.ext.junit.runners.AndroidJUnit4 import org.junit.Assert.assertEquals import org.junit.Test -import org.junit.runner.RunWith -@RunWith(AndroidJUnit4::class) class SystemDownloadEngineTest { @Test fun successfulRunningAndFailedStatusesAreMapped() { 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 d859cd95..3e9dbb80 100644 --- a/app/src/test/kotlin/com/subtracks/data/repo/DownloadRepositoryTest.kt +++ b/app/src/test/kotlin/com/subtracks/data/repo/DownloadRepositoryTest.kt @@ -188,17 +188,19 @@ class DownloadRepositoryTest { } @Test - fun anEngineFailureIsMirroredWithItsMessage() = + fun anEngineFailureIsMirroredWithItsMessageAndLeavesNoPartialFile() = runTest { seedLibrary() runBlocking { repository.download(1, "s1") } val id = engine.requests.single().first + writeFile(1, "s1", "half") engine.fail(id) runBlocking { repository.reconcile() } assertEquals(DownloadStatus.Failed, row(1, "s1")?.status) assertEquals("Download failed", row(1, "s1")?.error) + assertFalse(file(1, "s1").exists()) } @Test @@ -291,7 +293,7 @@ class DownloadRepositoryTest { ), ) } - await { sources.downloadUri("s1") != null } + await { sources.downloadUri(1, "s1") != null } } private fun file( diff --git a/app/src/test/kotlin/com/subtracks/playback/MediaDataSourceTest.kt b/app/src/test/kotlin/com/subtracks/playback/MediaDataSourceTest.kt index 250b149c..b1f93367 100644 --- a/app/src/test/kotlin/com/subtracks/playback/MediaDataSourceTest.kt +++ b/app/src/test/kotlin/com/subtracks/playback/MediaDataSourceTest.kt @@ -6,7 +6,10 @@ import androidx.media3.datasource.DataSpec import androidx.media3.datasource.okhttp.OkHttpDataSource import androidx.test.core.app.ApplicationProvider import androidx.test.ext.junit.runners.AndroidJUnit4 +import com.subtracks.data.source.streamLengthSuffix import okhttp3.OkHttpClient +import okhttp3.mockwebserver.MockResponse +import okhttp3.mockwebserver.MockWebServer import org.junit.Assert.assertEquals import org.junit.Test import org.junit.runner.RunWith @@ -14,11 +17,14 @@ import java.io.File @RunWith(AndroidJUnit4::class) class MediaDataSourceTest { + private val context = ApplicationProvider.getApplicationContext() + + private fun dataSource() = mediaDataSourceFactory(context, OkHttpDataSource.Factory(OkHttpClient())).createDataSource() + @Test fun aDownloadedFileIsReadThroughTheStreamingChain() { - val context = ApplicationProvider.getApplicationContext() val file = File(context.cacheDir, "downloaded-${System.nanoTime()}").apply { writeBytes(byteArrayOf(1, 2, 3, 4)) } - val dataSource = mediaDataSourceFactory(context, OkHttpDataSource.Factory(OkHttpClient())).createDataSource() + val dataSource = dataSource() val length = dataSource.open(DataSpec(Uri.fromFile(file))) val buffer = ByteArray(4) @@ -29,4 +35,19 @@ class MediaDataSourceTest { assertEquals(4, read) assertEquals(listOf(1, 2, 3, 4), buffer.toList()) } + + @Test + fun aStreamWithoutAContentLengthGetsItsDeclaredLength() { + val server = MockWebServer() + server.enqueue(MockResponse().setChunkedBody("0123456789", 1)) + server.start() + val url = "http://${server.hostName}:${server.port}/rest/stream" + streamLengthSuffix(1_000_000) + val dataSource = dataSource() + + val length = dataSource.open(DataSpec(Uri.parse(url))) + dataSource.close() + server.shutdown() + + assertEquals(1_000_000L, length) + } }