address arch review: guard the page cap, coalesce syncs, tighten buffers

- Paged loops now throw when they hit maxPages with a full last page, so a
  truncated enumeration aborts the entity's collect and skips its prune
  instead of deleting rows the sync never saw. The cap is injectable for
  tests.
- Move sync serialization into SyncManager (tryLock coalesces duplicate
  requests) and construct SyncService there; SourceRepository only exposes
  the active MusicSource. Removes the repo <-> sync import cycle.
- Re-throw CancellationException instead of reporting a cancelled sync as
  failed.
- Add buffer(1) to the outer albums/songs/playlistSongs flows so their
  flowOn channel cannot hold 64 batches.
- Rank prefetch is best-effort (missing frequent/recent no longer aborts
  the album sync).
- Skip playlist entries without an id, keeping positions contiguous.
- Chunk stale-disc deletes like the other prunes.
- Update the implementation plan line that still described the old
  single-transaction sync.

app/src/main/kotlin/com/subtracks/data/repo/SourceRepository.kt

@@ -6,8 +6,8 @@ import com.subtracks.data.model.Source
 import com.subtracks.data.model.SubsonicConfig
 import com.subtracks.data.model.SubsonicSource
 import com.subtracks.data.prefs.UserPreferences
+import com.subtracks.data.source.MusicSource
 import com.subtracks.data.source.subsonic.SubsonicClient
-import com.subtracks.data.sync.SyncService
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
@@ -16,8 +16,6 @@ import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.combine
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.launch
-import kotlinx.coroutines.sync.Mutex
-import kotlinx.coroutines.sync.withLock
 import kotlinx.coroutines.withContext
 import okhttp3.HttpUrl.Companion.toHttpUrl
 import okhttp3.OkHttpClient
@@ -29,7 +27,6 @@ class SourceRepository(
     private val prefs: UserPreferences,
 ) {
     private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
-    private val syncLock = Mutex()
 
     @Volatile
     private var active: SubsonicMusicSource? = null
@@ -118,13 +115,10 @@ class SourceRepository(
             runCatching { client(address, username, password, useTokenAuth).check("ping") }
         }
 
-    suspend fun sync(): Result<Unit> =
+    suspend fun activeMusicSource(): MusicSource? {
-        syncLock.withLock {
+        val config = db.sourcesDao().activeSubsonicConfigOnce() ?: return null
-            runCatching {
+        return config.toMusicSource(prefs.maxBitrate.first(), prefs.streamFormat.first())
-                val config = db.sourcesDao().activeSubsonicConfigOnce() ?: error("No server configured")
+    }
-                SyncService(db, config.toMusicSource(prefs.maxBitrate.first(), prefs.streamFormat.first())).sync()
-            }
-        }
 
     private fun SubsonicConfig.toMusicSource(
         maxBitrate: Int,

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

@@ -25,6 +25,7 @@ class SubsonicSource(
     private val client: SubsonicClient,
     private val maxBitrate: Int = 0,
     private val streamFormat: String? = null,
+    private val maxPages: Int = MAX_PAGES,
 ) : MusicSource {
     private var emptyQuerySearchSupported: Boolean? = null
 
@@ -45,7 +46,7 @@ class SubsonicSource(
             val (frequent, recent) = fetchRanks()
             var offset = 0
             var pages = 0
-            while (pages < MAX_PAGES) {
+            while (pages < maxPages) {
                 var count = 0
                 entityBatches(
                     method = "getAlbumList2",
@@ -69,7 +70,8 @@ class SubsonicSource(
                 offset += PAGE_SIZE
                 pages++
             }
-        }.flowOn(Dispatchers.IO)
+            checkPageCap(pages)
+        }.flowOn(Dispatchers.IO).buffer(1)
 
     override fun songs(): Flow<List<Song>> =
         flow {
@@ -78,7 +80,7 @@ class SubsonicSource(
             } else {
                 emitAll(albumSongs())
             }
-        }.flowOn(Dispatchers.IO)
+        }.flowOn(Dispatchers.IO).buffer(1)
 
     override fun playlists(): Flow<List<Playlist>> =
         entityBatches(
@@ -96,10 +98,14 @@ class SubsonicSource(
                     method = "getPlaylist",
                     params = mapOf("id" to playlistId),
                     tag = "entry",
-                    create = { attrs -> SubsonicXml.playlistSong(id, playlistId, position++, attrs) },
+                    create = { attrs ->
+                        val entryId = attrs.attr("id")
+                        SubsonicXml.playlistSong(id, playlistId, if (entryId.isEmpty()) -1L else position++, attrs)
+                    },
+                    accept = { it.songId.isNotEmpty() },
                 ).collect { emit(it) }
             }
-        }.flowOn(Dispatchers.IO)
+        }.flowOn(Dispatchers.IO).buffer(1)
 
     fun streamUri(songId: String) =
         client.uri(
@@ -132,7 +138,7 @@ class SubsonicSource(
         flow {
             var offset = 0
             var pages = 0
-            while (pages < MAX_PAGES) {
+            while (pages < maxPages) {
                 var count = 0
                 entityBatches(
                     method = "search3",
@@ -155,13 +161,14 @@ class SubsonicSource(
                 offset += PAGE_SIZE
                 pages++
             }
+            checkPageCap(pages)
         }
 
     private fun albumSongs(): Flow<List<Song>> =
         flow {
             var offset = 0
             var pages = 0
-            while (pages < MAX_PAGES) {
+            while (pages < maxPages) {
                 val albums = ArrayList<String>(PAGE_SIZE)
                 entityBatches(
                     method = "getAlbumList2",
@@ -183,34 +190,39 @@ class SubsonicSource(
                 offset += PAGE_SIZE
                 pages++
             }
+            checkPageCap(pages)
         }
 
     private suspend fun fetchRanks(): Pair<Map<String, Long>, Map<String, Long>> = fetchRank("frequent") to fetchRank("recent")
 
-    private suspend fun fetchRank(type: String): Map<String, Long> {
+    private suspend fun fetchRank(type: String): Map<String, Long> =
-        val ranks = HashMap<String, Long>()
+        try {
-        var offset = 0
+            val ranks = HashMap<String, Long>()
-        var pages = 0
+            var offset = 0
-        while (pages < MAX_PAGES) {
+            var pages = 0
-            var index = 0
+            while (pages < maxPages) {
-            entityBatches(
+                var index = 0
-                method = "getAlbumList2",
+                entityBatches(
-                params = page(type, offset),
+                    method = "getAlbumList2",
-                tag = "album",
+                    params = page(type, offset),
-                create = { it.attr("id") },
+                    tag = "album",
-                accept = { it.isNotEmpty() },
+                    create = { it.attr("id") },
-            ).collect { batch ->
+                    accept = { it.isNotEmpty() },
-                batch.forEach { albumId ->
+                ).collect { batch ->
-                    ranks[albumId] = (offset + index).toLong()
+                    batch.forEach { albumId ->
-                    index++
+                        ranks[albumId] = (offset + index).toLong()
+                        index++
+                    }
                 }
+                if (index < PAGE_SIZE) break
+                offset += PAGE_SIZE
+                pages++
             }
-            if (index < PAGE_SIZE) break
+            checkPageCap(pages)
-            offset += PAGE_SIZE
+            ranks
-            pages++
+        } catch (_: SubsonicException) {
+            emptyMap()
         }
-        return ranks
-    }
 
     private fun supportsEmptyQuerySearch(): Boolean {
         emptyQuerySearchSupported?.let { return it }
@@ -227,6 +239,10 @@ class SubsonicSource(
         return supported
     }
 
+    private fun checkPageCap(pages: Int) {
+        if (pages >= maxPages) throw SubsonicException(-1, "Library exceeds the $maxPages page cap")
+    }
+
     private fun <T> entityBatches(
         method: String,
         params: Map<String, String> = emptyMap(),

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

@@ -1,14 +1,17 @@
 package com.subtracks.data.sync
 
 import android.util.Log
+import com.subtracks.data.db.SubtracksDatabase
 import com.subtracks.data.repo.QueueRepository
 import com.subtracks.data.repo.SourceRepository
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
 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
@@ -23,20 +26,38 @@ sealed interface SyncStatus {
 }
 
 class SyncManager(
+    private val db: SubtracksDatabase,
     private val sourceRepository: SourceRepository,
     private val queueRepository: QueueRepository,
 ) {
     private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
+    private val syncLock = Mutex()
     private val _status = MutableStateFlow<SyncStatus>(SyncStatus.Idle)
     val status: StateFlow<SyncStatus> = _status
 
     fun requestSync() {
-        scope.launch { runSync() }
+        if (!syncLock.tryLock()) return
+        scope.launch {
+            try {
+                runSync()
+            } finally {
+                syncLock.unlock()
+            }
+        }
     }
 
     private suspend fun runSync() {
         _status.value = SyncStatus.Running
-        val result = sourceRepository.sync()
+        val result =
+            try {
+                val source = sourceRepository.activeMusicSource() ?: error("No server configured")
+                SyncService(db, source).sync()
+                Result.success(Unit)
+            } catch (cancellation: CancellationException) {
+                throw cancellation
+            } catch (failure: Exception) {
+                Result.failure(failure)
+            }
         queueRepository.invalidateLibraryCache()
         _status.value =
             result.fold(

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

@@ -117,17 +117,18 @@ class SyncService(
     }
 
     private suspend fun deleteStaleDiscs(stale: Set<DiscKey>) {
-        if (stale.isEmpty()) return
+        stale.chunked(DELETE_CHUNK).forEach { chunk ->
-        db.useWriterConnection { connection ->
+            db.useWriterConnection { connection ->
-            connection.immediateTransaction {
+                connection.immediateTransaction {
-                connection.usePrepared("DELETE FROM discs WHERE sourceId = ? AND albumId = ? AND disc = ?") { statement ->
+                    connection.usePrepared("DELETE FROM discs WHERE sourceId = ? AND albumId = ? AND disc = ?") { statement ->
-                    stale.forEach { key ->
+                        chunk.forEach { key ->
-                        statement.bind(1, source.id)
+                            statement.bind(1, source.id)
-                        statement.bind(2, key.albumId)
+                            statement.bind(2, key.albumId)
-                        statement.bind(3, key.disc)
+                            statement.bind(3, key.disc)
-                        statement.step()
+                            statement.step()
-                        statement.reset()
+                            statement.reset()
-                        statement.clearBindings()
+                            statement.clearBindings()
+                        }
                     }
                 }
             }

app/src/main/kotlin/com/subtracks/di/AppModule.kt

@@ -34,7 +34,7 @@ fun appModule(
     single { SourceRepository(get(), get(), get()) }
     single { LibraryRepository(get(), get()) }
     single<ArtworkSeedStore> { ArtworkSeedRepository(get()) }
-    single { SyncManager(get(), get()) }
+    single { SyncManager(get(), get(), get()) }
     single { QueueRepository(get()) }
     single<PlayerConnection> { MediaSessionConnection(context.applicationContext, get()) }
     single { PlaybackController(get(), get(), get()) }

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

@@ -179,6 +179,49 @@ class SubsonicSourceTest {
             assertEquals(500, firstBatch.size)
         }
 
+    @Test
+    fun albumsAbortWhenThePageCapIsHit() =
+        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)
+                        return if (url.queryParameter("type") == "newest") {
+                            MockResponse().setBody(albumPage(0, 500))
+                        } else {
+                            MockResponse().setBody(emptyAlbumList())
+                        }
+                    }
+                }
+
+            val failure =
+                runCatching {
+                    SubsonicSource(1, client(), maxPages = 2).albums().toList()
+                }
+
+            assertEquals(true, failure.exceptionOrNull() is SubsonicException)
+        }
+
+    @Test
+    fun playlistEntriesWithoutIdsAreSkipped() =
+        runBlocking {
+            server.enqueue(
+                MockResponse().setBody(
+                    "<subsonic-response status=\"ok\"><playlist id=\"p1\">" +
+                        "<entry id=\"s1\" title=\"One\"/>" +
+                        "<entry title=\"No id\"/>" +
+                        "<entry id=\"s2\" title=\"Two\"/>" +
+                        "</playlist></subsonic-response>",
+                ),
+            )
+
+            val entries = SubsonicSource(1, client()).playlistSongs(listOf("p1")).toList().flatten()
+
+            assertEquals(listOf("s1", "s2"), entries.map { it.songId })
+            assertEquals(listOf(0L, 1L), entries.map { it.position })
+        }
+
     @Test
     fun streamUriRequestsTranscodingWhenABitrateIsSet() {
         val uri = SubsonicSource(1, client(), maxBitrate = 128).streamUri("s1").toString()

app/src/test/kotlin/com/subtracks/data/sync/SyncManagerTest.kt

@@ -37,7 +37,7 @@ class SyncManagerTest {
                 .build()
         sourceRepository = SourceRepository(db, OkHttpClient(), fakeUserPreferences())
         queueRepository = QueueRepository(db)
-        manager = SyncManager(sourceRepository, queueRepository)
+        manager = SyncManager(db, sourceRepository, queueRepository)
     }
 
     @After

docs/implementation-plan.md

@@ -12,7 +12,7 @@ Everything needed to mirror a server library locally:
 
 - `MusicSource` abstraction with `SubsonicClient`, `SubsonicXml` and `SubsonicSource`.
 - Room 3 schema on bundled SQLite, so the SQLite version is identical on every device.
-- `SyncService`: fetch first, then a single transaction of upserts with diff-based pruning.
+- `SyncService`: stream each entity in batches, upserting per batch and pruning the IDs the server no longer lists; the sync is not atomic.
 - Unit tests (client, XML, sync), native integration tests against navidrome and gonic, and CI (`unit`, `lint`, `integration`) with warm caches.
 
 ## Phase 2 - UI vertical slice (done)