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.

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 }

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)
     }
 

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<Map<String, SongDownload>> by lazy {
-        downloadStates()
-            .onEach { loaded.complete(Unit) }
-            .stateIn(scope, SharingStarted.Eagerly, emptyMap())
-    }
+    private val statesFlow = MutableStateFlow<Map<String, SongDownload>>(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<Map<String, SongDownload>> = sharedStates
+    fun states(): StateFlow<Map<String, SongDownload>> = 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)

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,

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() {

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(

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<Context>()
+
+    private fun dataSource() = mediaDataSourceFactory(context, OkHttpDataSource.Factory(OkHttpClient())).createDataSource()
+
     @Test
     fun aDownloadedFileIsReadThroughTheStreamingChain() {
-        val context = ApplicationProvider.getApplicationContext<Context>()
         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<Byte>(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)
+    }
 }