address review: test the source wiring, one default, clamp the stored value

- Add a `SourceRepositoryTest` that sets the preference to 1 and proves the
  active source fetches sequentially; removing the `maxConcurrentFetches`
  wiring now fails it.
- `playlistSongsRespectsASmallerFetchBound` now holds the first `bound`
  fetches and asserts no `(bound + 1)`th starts, instead of timing an
  overlap; it fails when the semaphore ignores the bound.
- One `DEFAULT_FETCH_CONCURRENCY` (in `data.source`) backs both the source
  default and the preference default, so they cannot drift.
- Clamp the stored preference in `syncConcurrency()` too, so a bad stored
  value can't make `SubsonicSource` throw on construction.

app/src/main/kotlin/com/subtracks/data/prefs/UserPreferences.kt

@@ -8,6 +8,7 @@ import androidx.datastore.preferences.core.intPreferencesKey
 import androidx.datastore.preferences.core.stringPreferencesKey
 import androidx.datastore.preferences.preferencesDataStore
 import com.subtracks.data.net.NetworkMode
+import com.subtracks.data.source.DEFAULT_FETCH_CONCURRENCY
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.map
@@ -43,8 +44,6 @@ data class StreamQuality(
     val transcodes: Boolean get() = maxBitrate > 0 || !format.isNullOrEmpty()
 }
 
-const val DEFAULT_SYNC_CONCURRENCY = 4
-
 private val Context.preferences: DataStore<Preferences> by preferencesDataStore("user_prefs")
 
 data class ArtworkSeedValue(
@@ -79,7 +78,7 @@ class UserPreferences(
 
     private fun streamQualityKey(mode: NetworkMode) = stringPreferencesKey("stream_quality_${mode.key}")
 
-    fun syncConcurrency(): Flow<Int> = store.data.map { prefs -> prefs[syncConcurrencyKey] ?: DEFAULT_SYNC_CONCURRENCY }
+    fun syncConcurrency(): Flow<Int> = store.data.map { prefs -> (prefs[syncConcurrencyKey] ?: DEFAULT_FETCH_CONCURRENCY).coerceAtLeast(1) }
 
     suspend fun setSyncConcurrency(value: Int) {
         store.edit { prefs -> prefs[syncConcurrencyKey] = value.coerceAtLeast(1) }

app/src/main/kotlin/com/subtracks/data/source/MusicSource.kt

@@ -7,6 +7,8 @@ import com.subtracks.data.model.PlaylistSong
 import com.subtracks.data.model.Song
 import kotlinx.coroutines.flow.Flow
 
+const val DEFAULT_FETCH_CONCURRENCY = 4
+
 interface MusicSource {
     val id: Long
 

app/src/main/kotlin/com/subtracks/data/source/subsonic/SubsonicSource.kt

@@ -5,6 +5,7 @@ import com.subtracks.data.model.Artist
 import com.subtracks.data.model.Playlist
 import com.subtracks.data.model.PlaylistSong
 import com.subtracks.data.model.Song
+import com.subtracks.data.source.DEFAULT_FETCH_CONCURRENCY
 import com.subtracks.data.source.MusicSource
 import com.subtracks.data.source.StarType
 import kotlinx.coroutines.CancellationException
@@ -34,7 +35,7 @@ class SubsonicSource(
     private val maxBitrate: Int = 0,
     private val streamFormat: String? = null,
     private val maxPages: Int = MAX_PAGES,
-    private val maxConcurrentFetches: Int = MAX_CONCURRENT_FETCHES,
+    private val maxConcurrentFetches: Int = DEFAULT_FETCH_CONCURRENCY,
 ) : MusicSource {
     private var emptyQuerySearchSupported: Boolean? = null
 
@@ -306,8 +307,6 @@ class SubsonicSource(
     }
 }
 
-internal const val MAX_CONCURRENT_FETCHES = 4
-
 private class PageCapExceeded(
     message: String,
 ) : SubsonicException(-1, message)

app/src/main/kotlin/com/subtracks/ui/settings/SettingsViewModel.kt

@@ -4,11 +4,11 @@ import androidx.lifecycle.ViewModel
 import androidx.lifecycle.viewModelScope
 import com.subtracks.data.model.Source
 import com.subtracks.data.net.NetworkMode
-import com.subtracks.data.prefs.DEFAULT_SYNC_CONCURRENCY
 import com.subtracks.data.prefs.StreamQuality
 import com.subtracks.data.prefs.UserPreferences
 import com.subtracks.data.repo.DownloadRepository
 import com.subtracks.data.repo.SourceRepository
+import com.subtracks.data.source.DEFAULT_FETCH_CONCURRENCY
 import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.stateIn
@@ -38,7 +38,7 @@ class SettingsViewModel(
     val syncConcurrency: StateFlow<Int> =
         userPreferences
             .syncConcurrency()
-            .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), DEFAULT_SYNC_CONCURRENCY)
+            .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), DEFAULT_FETCH_CONCURRENCY)
 
     fun selectSource(id: Long) {
         viewModelScope.launch { sourceRepository.selectSource(id) }

app/src/test/kotlin/com/subtracks/data/prefs/UserPreferencesTest.kt

@@ -3,6 +3,7 @@ package com.subtracks.data.prefs
 import androidx.datastore.preferences.core.PreferenceDataStoreFactory
 import androidx.test.ext.junit.runners.AndroidJUnit4
 import com.subtracks.data.net.NetworkMode
+import com.subtracks.data.source.DEFAULT_FETCH_CONCURRENCY
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.test.runTest
@@ -65,7 +66,7 @@ class UserPreferencesTest {
             val store = PreferenceDataStoreFactory.create(scope = backgroundScope) { file }
             val prefs = UserPreferences(store)
 
-            assertEquals(DEFAULT_SYNC_CONCURRENCY, prefs.syncConcurrency().first())
+            assertEquals(DEFAULT_FETCH_CONCURRENCY, prefs.syncConcurrency().first())
 
             prefs.setSyncConcurrency(8)
             assertEquals(8, prefs.syncConcurrency().first())

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

@@ -9,15 +9,21 @@ import com.subtracks.data.db.SubtracksDatabase
 import com.subtracks.data.download.ArtworkStore
 import com.subtracks.data.model.CoverArtRef
 import com.subtracks.data.model.coverArtKey
+import com.subtracks.data.prefs.UserPreferences
 import com.subtracks.data.prefs.fakeUserPreferences
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.async
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.test.runTest
 import kotlinx.coroutines.withTimeout
 import okhttp3.OkHttpClient
+import okhttp3.mockwebserver.Dispatcher
 import okhttp3.mockwebserver.MockResponse
 import okhttp3.mockwebserver.MockWebServer
+import okhttp3.mockwebserver.RecordedRequest
 import org.junit.After
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertFalse
@@ -26,10 +32,14 @@ import org.junit.Before
 import org.junit.Test
 import org.junit.runner.RunWith
 import java.io.File
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicInteger
 
 @RunWith(AndroidJUnit4::class)
 class SourceRepositoryTest {
     private lateinit var db: SubtracksDatabase
+    private lateinit var prefs: UserPreferences
     private lateinit var repository: SourceRepository
     private lateinit var artwork: ArtworkStore
     private val messages = ArrayList<String>()
@@ -43,7 +53,8 @@ class SourceRepositoryTest {
                 .setDriver(BundledSQLiteDriver())
                 .build()
         artwork = ArtworkStore(File(context.cacheDir, "art-${System.nanoTime()}"))
-        repository = SourceRepository(db, OkHttpClient(), fakeUserPreferences(), artwork, showMessage = { messages += it })
+        prefs = fakeUserPreferences()
+        repository = SourceRepository(db, OkHttpClient(), prefs, artwork, showMessage = { messages += it })
     }
 
     @After
@@ -88,6 +99,46 @@ class SourceRepositoryTest {
             }
         }
 
+    @Test
+    fun theActiveSourceFetchConcurrencyComesFromThePreference() =
+        runBlocking {
+            withServer { server ->
+                val dispatches = AtomicInteger()
+                val secondStarted = CountDownLatch(1)
+                val release = CountDownLatch(1)
+                server.dispatcher =
+                    object : Dispatcher() {
+                        override fun dispatch(request: RecordedRequest): MockResponse {
+                            val url = request.requestUrl!!
+                            if (url.encodedPath != "/rest/getPlaylist.view") return MockResponse().setResponseCode(404)
+                            if (dispatches.incrementAndGet() == 2) secondStarted.countDown()
+                            release.await(5, TimeUnit.SECONDS)
+                            val id = url.queryParameter("id")!!
+                            return MockResponse().setBody(
+                                "<subsonic-response status=\"ok\"><playlist id=\"$id\">" +
+                                    (1..3).joinToString("") { "<entry id=\"$id-s$it\" title=\"t$it\"/>" } +
+                                    "</playlist></subsonic-response>",
+                            )
+                        }
+                    }
+
+                prefs.setSyncConcurrency(1)
+                repository.addSource("nav", server.url("/").toString(), "u", "p", false)
+
+                val source = repository.activeMusicSource()!!
+                val fetching =
+                    async(Dispatchers.IO) {
+                        source.playlistSongs(listOf("p1", "p2", "p3")).toList().flatten()
+                    }
+
+                // With concurrency 1 the second fetch must not start while the first is in flight;
+                // if the preference is not wired the source defaults to a higher bound and it does.
+                assertFalse(secondStarted.await(500, TimeUnit.MILLISECONDS))
+                release.countDown()
+                assertEquals(9, fetching.await().size)
+            }
+        }
+
     @Test
     fun pingReportsWhetherItFellBackToThePassword() =
         runBlocking {

app/src/test/kotlin/com/subtracks/data/source/subsonic/SubsonicSourceTest.kt

@@ -1,5 +1,6 @@
 package com.subtracks.data.source.subsonic
 
+import com.subtracks.data.source.DEFAULT_FETCH_CONCURRENCY
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.async
 import kotlinx.coroutines.flow.first
@@ -314,7 +315,7 @@ class SubsonicSourceTest {
                 assertEquals(listOf(0L, 1L, 2L), batch.sortedBy { it.position }.map { it.position })
                 assertEquals((1..3).map { "$playlistId-s$it" }, batch.sortedBy { it.position }.map { it.songId })
             }
-            assertTrue(maxInFlight.get() in 2..MAX_CONCURRENT_FETCHES)
+            assertTrue(maxInFlight.get() in 2..DEFAULT_FETCH_CONCURRENCY)
         }
 
     @Test
@@ -349,20 +350,17 @@ class SubsonicSourceTest {
     fun playlistSongsRespectsASmallerFetchBound() =
         runBlocking {
             val bound = 2
-            val inFlight = AtomicInteger()
-            val maxInFlight = AtomicInteger()
             val started = CountDownLatch(bound)
+            val beyondBound = CountDownLatch(bound + 1)
             val release = CountDownLatch(1)
             server.dispatcher =
                 object : Dispatcher() {
                     override fun dispatch(request: RecordedRequest): MockResponse {
                         val url = request.requestUrl!!
                         if (url.encodedPath != "/rest/getPlaylist.view") return MockResponse().setResponseCode(404)
-                        val current = inFlight.incrementAndGet()
+                        beyondBound.countDown()
-                        maxInFlight.updateAndGet { maxOf(it, current) }
                         started.countDown()
                         release.await(5, TimeUnit.SECONDS)
-                        inFlight.decrementAndGet()
                         return MockResponse().setBody(playlistResponse(url.queryParameter("id")!!))
                     }
                 }
@@ -376,11 +374,15 @@ class SubsonicSourceTest {
                 }
 
             assertTrue("$bound fetches should have started together", started.await(5, TimeUnit.SECONDS))
+            // While those are held, the semaphore must block any further fetch.
+            assertFalse(
+                "a ${bound + 1}th fetch started past the bound",
+                beyondBound.await(500, TimeUnit.MILLISECONDS),
+            )
             release.countDown()
             val entries = fetching.await()
 
             assertEquals(12, entries.size)
-            assertEquals(bound, maxInFlight.get())
         }
 
     @Test
@@ -496,7 +498,7 @@ class SubsonicSourceTest {
             val songs = SubsonicSource(1, client()).songs().toList().flatten()
 
             assertEquals(4, songs.size)
-            assertTrue(maxInFlight.get() in 2..MAX_CONCURRENT_FETCHES)
+            assertTrue(maxInFlight.get() in 2..DEFAULT_FETCH_CONCURRENCY)
         }
 
     @Test