address final review: coalesce sync requests, fix page-cap edge

- SyncManager now uses a conflated request channel instead of tryLock, so a
  request that arrives during a sync (e.g. add-source switching the active
  server) is queued and runs afterwards against the then-active source
  instead of being dropped.
- Paged loops count raw server rows, not accepted entities, so an empty-id
  row on an otherwise full page no longer ends the scan early and prunes
  the unseen tail.
- The cap now fails when the page just past it is non-empty, not only when
  it is full, matching the bound the error message reports.
- fetchRank rethrows the cap instead of downgrading an over-cap album scan
  to empty ranks.
- Docs: the mutex lives in SyncManager, and the prune still holds the
  existing-id set.

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

@@ -6,7 +6,7 @@ import okhttp3.Request
 import java.io.InputStream
 import java.security.MessageDigest
 
-class SubsonicException(
+open class SubsonicException(
     val code: Int,
     message: String,
 ) : Exception(message)

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

@@ -16,6 +16,7 @@ import kotlinx.coroutines.flow.channelFlow
 import kotlinx.coroutines.flow.emitAll
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.flowOn
+import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.withContext
 import org.xml.sax.Attributes
 
@@ -38,8 +39,7 @@ class SubsonicSource(
             method = "getArtists",
             tag = "artist",
             create = { SubsonicXml.artist(id, it) },
-            accept = { it.id.isNotEmpty() },
-        )
+        ).map { batch -> batch.filter { it.id.isNotEmpty() } }
 
     override fun albums(): Flow<List<Album>> =
         flow {
@@ -47,26 +47,31 @@ class SubsonicSource(
             var offset = 0
             var pages = 0
             while (true) {
-                var count = 0
+                var raw = 0
                 entityBatches(
                     method = "getAlbumList2",
                     params = page("newest", offset),
                     tag = "album",
                     create = { SubsonicXml.albumDraft(id, it) },
                     onChild = { draft, name, attrs -> if (name == "discTitles") draft.addDiscTitle(attrs) },
-                    accept = { it.id.isNotEmpty() },
                 ).collect { batch ->
-                    count += batch.size
-                    emit(
-                        batch.map {
-                            it.toAlbum().copy(
-                                frequentRank = frequent[it.id],
-                                recentRank = recent[it.id],
-                            )
-                        },
-                    )
+                    raw += batch.size
+                    val accepted = batch.filter { it.id.isNotEmpty() }
+                    if (accepted.isNotEmpty()) {
+                        emit(
+                            accepted.map {
+                                it.toAlbum().copy(
+                                    frequentRank = frequent[it.id],
+                                    recentRank = recent[it.id],
+                                )
+                            },
+                        )
+                    }
+                }
+                if (raw < PAGE_SIZE) {
+                    if (raw > 0 && pages >= maxPages) throw pageCapExceeded()
+                    break
                 }
-                if (count < PAGE_SIZE) break
                 pages++
                 if (pages > maxPages) throw pageCapExceeded()
                 offset += PAGE_SIZE
@@ -87,8 +92,7 @@ class SubsonicSource(
             method = "getPlaylists",
             tag = "playlist",
             create = { SubsonicXml.playlist(id, it) },
-            accept = { it.id.isNotEmpty() },
-        )
+        ).map { batch -> batch.filter { it.id.isNotEmpty() } }
 
     override fun playlistSongs(playlistIds: List<String>): Flow<List<PlaylistSong>> =
         flow {
@@ -102,8 +106,10 @@ class SubsonicSource(
                         val entryId = attrs.attr("id")
                         SubsonicXml.playlistSong(id, playlistId, if (entryId.isEmpty()) -1L else position++, attrs)
                     },
-                    accept = { it.songId.isNotEmpty() },
-                ).collect { emit(it) }
+                ).collect { batch ->
+                    val accepted = batch.filter { it.songId.isNotEmpty() }
+                    if (accepted.isNotEmpty()) emit(accepted)
+                }
             }
         }.flowOn(Dispatchers.IO).buffer(1)
 
@@ -139,7 +145,7 @@ class SubsonicSource(
             var offset = 0
             var pages = 0
             while (true) {
-                var count = 0
+                var raw = 0
                 entityBatches(
                     method = "search3",
                     params =
@@ -152,12 +158,15 @@ class SubsonicSource(
                         ),
                     tag = "song",
                     create = { SubsonicXml.song(id, it) },
-                    accept = { it.id.isNotEmpty() },
                 ).collect { batch ->
-                    count += batch.size
-                    emit(batch)
+                    raw += batch.size
+                    val accepted = batch.filter { it.id.isNotEmpty() }
+                    if (accepted.isNotEmpty()) emit(accepted)
+                }
+                if (raw < PAGE_SIZE) {
+                    if (raw > 0 && pages >= maxPages) throw pageCapExceeded()
+                    break
                 }
-                if (count < PAGE_SIZE) break
                 pages++
                 if (pages > maxPages) throw pageCapExceeded()
                 offset += PAGE_SIZE
@@ -169,24 +178,32 @@ class SubsonicSource(
             var offset = 0
             var pages = 0
             while (true) {
+                var raw = 0
                 val albums = ArrayList<String>(PAGE_SIZE)
                 entityBatches(
                     method = "getAlbumList2",
                     params = page("alphabeticalByName", offset),
                     tag = "album",
                     create = { it.attr("id") },
-                    accept = { it.isNotEmpty() },
-                ).collect { albums += it }
+                ).collect { batch ->
+                    raw += batch.size
+                    batch.forEach { if (it.isNotEmpty()) albums += it }
+                }
                 for (albumId in albums) {
                     entityBatches(
                         method = "getAlbum",
                         params = mapOf("id" to albumId),
                         tag = "song",
                         create = { SubsonicXml.song(id, it) },
-                        accept = { it.id.isNotEmpty() },
-                    ).collect { emit(it) }
+                    ).collect { batch ->
+                        val accepted = batch.filter { it.id.isNotEmpty() }
+                        if (accepted.isNotEmpty()) emit(accepted)
+                    }
+                }
+                if (raw < PAGE_SIZE) {
+                    if (raw > 0 && pages >= maxPages) throw pageCapExceeded()
+                    break
                 }
-                if (albums.size < PAGE_SIZE) break
                 pages++
                 if (pages > maxPages) throw pageCapExceeded()
                 offset += PAGE_SIZE
@@ -201,26 +218,33 @@ class SubsonicSource(
             var offset = 0
             var pages = 0
             while (true) {
+                var raw = 0
                 var index = 0
                 entityBatches(
                     method = "getAlbumList2",
                     params = page(type, offset),
                     tag = "album",
                     create = { it.attr("id") },
-                    accept = { it.isNotEmpty() },
                 ).collect { batch ->
+                    raw += batch.size
                     batch.forEach { albumId ->
-                        ranks[albumId] = (offset + index).toLong()
-                        index++
+                        if (albumId.isNotEmpty()) {
+                            ranks[albumId] = (offset + index).toLong()
+                            index++
+                        }
                     }
                 }
-                if (index < PAGE_SIZE) break
+                if (raw < PAGE_SIZE) {
+                    if (raw > 0 && pages >= maxPages) throw pageCapExceeded()
+                    break
+                }
                 pages++
                 if (pages > maxPages) throw pageCapExceeded()
                 offset += PAGE_SIZE
             }
             ranks
-        } catch (_: SubsonicException) {
+        } catch (failure: SubsonicException) {
+            if (failure is PageCapExceeded) throw failure
             emptyMap()
         }
 
@@ -239,7 +263,7 @@ class SubsonicSource(
         return supported
     }
 
-    private fun pageCapExceeded() = SubsonicException(-1, "Library exceeds the ${maxPages * PAGE_SIZE}-row page cap")
+    private fun pageCapExceeded() = PageCapExceeded("Library exceeds the ${maxPages * PAGE_SIZE}-row page cap")
 
     private fun <T> entityBatches(
         method: String,
@@ -247,7 +271,6 @@ class SubsonicSource(
         tag: String,
         create: (Attributes) -> T,
         onChild: (T, String, Attributes) -> Unit = { _, _, _ -> },
-        accept: (T) -> Boolean = { true },
     ): Flow<List<T>> =
         channelFlow {
             val buffer = ArrayList<T>(PAGE_SIZE)
@@ -267,10 +290,8 @@ class SubsonicSource(
                     create = create,
                     onChild = onChild,
                     onEnd = { entity ->
-                        if (accept(entity)) {
-                            buffer += entity
-                            if (buffer.size >= PAGE_SIZE) flush()
-                        }
+                        buffer += entity
+                        if (buffer.size >= PAGE_SIZE) flush()
                     },
                 )
             }
@@ -293,3 +314,7 @@ class SubsonicSource(
         const val THUMBNAIL_SIZE = 256
     }
 }
+
+private class PageCapExceeded(
+    message: String,
+) : SubsonicException(-1, message)

app/src/main/kotlin/com/subtracks/data/sync/SyncManager.kt

@@ -8,10 +8,10 @@ import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.launch
-import kotlinx.coroutines.sync.Mutex
 
 sealed interface SyncStatus {
     data object Idle : SyncStatus
@@ -31,42 +31,34 @@ class SyncManager(
     private val queueRepository: QueueRepository,
 ) {
     private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
-    private val syncLock = Mutex()
+    private val requests = Channel<Unit>(Channel.CONFLATED)
     private val _status = MutableStateFlow<SyncStatus>(SyncStatus.Idle)
     val status: StateFlow<SyncStatus> = _status
 
-    fun requestSync() {
-        if (!syncLock.tryLock()) return
+    init {
         scope.launch {
-            try {
-                runSync()
-            } finally {
-                syncLock.unlock()
-            }
+            for (ignored in requests) runSync()
         }
     }
 
+    fun requestSync() {
+        requests.trySend(Unit)
+    }
+
     private suspend fun runSync() {
         _status.value = SyncStatus.Running
-        val result =
+        _status.value =
             try {
                 val source = sourceRepository.activeMusicSource() ?: error("No server configured")
                 SyncService(db, source).sync()
-                Result.success(Unit)
+                SyncStatus.Success
             } catch (cancellation: CancellationException) {
                 throw cancellation
             } catch (failure: Exception) {
-                Result.failure(failure)
+                Log.w(TAG, "Sync failed", failure)
+                SyncStatus.Failed(failure.message ?: "Sync failed")
             }
         queueRepository.invalidateLibraryCache()
-        _status.value =
-            result.fold(
-                onSuccess = { SyncStatus.Success },
-                onFailure = { error ->
-                    Log.w(TAG, "Sync failed", error)
-                    SyncStatus.Failed(error.message ?: "Sync failed")
-                },
-            )
     }
 
     private companion object {

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

@@ -224,6 +224,52 @@ class SubsonicSourceTest {
             assertEquals(1000, albums.size)
         }
 
+    @Test
+    fun albumsAbortWhenThePageJustOverTheCapIsNotEmpty() =
+        runBlocking {
+            server.dispatcher =
+                object : Dispatcher() {
+                    override fun dispatch(request: RecordedRequest): MockResponse {
+                        val url = request.requestUrl!!
+                        if (url.encodedPath != "/rest/getAlbumList2.view") return MockResponse().setResponseCode(404)
+                        if (url.queryParameter("type") != "newest") return MockResponse().setBody(emptyAlbumList())
+                        return when (url.queryParameter("offset")) {
+                            "0", "500" -> MockResponse().setBody(albumPage(url.queryParameter("offset")!!.toInt(), 500))
+                            else -> MockResponse().setBody(albumPage(1000, 1))
+                        }
+                    }
+                }
+
+            val failure =
+                runCatching {
+                    SubsonicSource(1, client(), maxPages = 2).albums().toList()
+                }
+
+            assertEquals(true, failure.exceptionOrNull() is SubsonicException)
+        }
+
+    @Test
+    fun emptyIdRowsDoNotEndPaginationEarly() =
+        runBlocking {
+            server.dispatcher =
+                object : Dispatcher() {
+                    override fun dispatch(request: RecordedRequest): MockResponse {
+                        val url = request.requestUrl!!
+                        if (url.encodedPath != "/rest/getAlbumList2.view") return MockResponse().setResponseCode(404)
+                        if (url.queryParameter("type") != "newest") return MockResponse().setBody(emptyAlbumList())
+                        return when (url.queryParameter("offset")) {
+                            "0" -> MockResponse().setBody(albumPageWithEmptyId(0, 500))
+                            else -> MockResponse().setBody(albumPage(500, 3))
+                        }
+                    }
+                }
+
+            val albums = SubsonicSource(1, client()).albums().toList().flatten()
+
+            assertEquals(502, albums.size)
+            assertEquals("al-502", albums.last().id)
+        }
+
     @Test
     fun playlistEntriesWithoutIdsAreSkipped() =
         runBlocking {
@@ -293,6 +339,16 @@ class SubsonicSourceTest {
         } +
         "</albumList2></subsonic-response>"
 
+    private fun albumPageWithEmptyId(
+        offset: Int,
+        count: Int,
+    ) = "<subsonic-response status=\"ok\"><albumList2>" +
+        "<album id=\"\" name=\"No id\" artist=\"Artist\" artistId=\"ar1\" songCount=\"1\"/>" +
+        (offset until offset + count - 1).joinToString("") {
+            "<album id=\"al-$it\" name=\"Album $it\" artist=\"Artist\" artistId=\"ar1\" songCount=\"1\"/>"
+        } +
+        "</albumList2></subsonic-response>"
+
     private fun songPage(
         prefix: String,
         offset: Int,