address review: serialize syncs, classify parse errors, widen coverage

- SourceRepository now holds a Mutex around sync(), so overlapping requestSync() calls can no longer interleave per-batch writes and stale-set prunes. The per-call SyncService lock could never do this. - parse() wraps SAXParseException/IOException as SubsonicException, so a truncated or malformed response is a clean sync failure instead of a raw parser exception escaping the empty-search fallback. - A response with no status attribute is accepted again, matching the old DOM behaviour (only status="failed" fails). - Simplify the SAX handler (drop sawRoot and StatusHandler), make the entity mappers internal, and drop an unreachable guard in write(). Tests: pagination and empty-search caching for songs, album pagination, empty-id filtering, cancellation, SAX error paths and structural edge cases, cross-source prune isolation, emptied playlists, disc-set shrink, multi-chunk deletes, change detection via total_changes, playlist id filtering, and SyncManager cache invalidation on failure.
2026-11-10 15:11:58 +01:00 · 2026-10-04 00:29:58 +00:00
parent b171b0b33c
commit be6bcd11dc
7 changed files with 523 additions and 32 deletions
@@ -1,7 +1,9 @@
 package com.subtracks.data.source.subsonic

+import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
 import okhttp3.OkHttpClient
 import okhttp3.mockwebserver.Dispatcher
 import okhttp3.mockwebserver.MockResponse
@@ -13,6 +15,8 @@ import org.junit.Assert.assertFalse
 import org.junit.Assert.assertTrue
 import org.junit.Before
 import org.junit.Test
+import java.util.concurrent.CopyOnWriteArrayList
+import java.util.concurrent.atomic.AtomicInteger

 class SubsonicSourceTest {
    private lateinit var server: MockWebServer
@@ -83,6 +87,98 @@ class SubsonicSourceTest {
            assertEquals(listOf(500, 500, 200), sizes)
        }

+    @Test
+    fun albumsPaginateUntilAShortPage() =
+        runBlocking {
+            val requestedOffsets = CopyOnWriteArrayList<Int>()
+            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())
+                        }
+                        val offset = url.queryParameter("offset")!!.toInt()
+                        requestedOffsets += offset
+                        val count = if (offset < 1000) 500 else 3
+                        return MockResponse().setBody(albumPage(offset, count))
+                    }
+                }
+
+            val albums = SubsonicSource(1, client()).albums().toList().flatten()
+
+            assertEquals(1003, albums.size)
+            assertEquals(listOf(0, 500, 1000), requestedOffsets.toList())
+            assertEquals("al-0", albums.first().id)
+            assertEquals("al-1002", albums.last().id)
+        }
+
+    @Test
+    fun songsUseEmptySearchAndPaginate() =
+        runBlocking {
+            val probeCount = AtomicInteger()
+            val requestedOffsets = CopyOnWriteArrayList<Int>()
+            server.dispatcher =
+                object : Dispatcher() {
+                    override fun dispatch(request: RecordedRequest): MockResponse {
+                        val url = request.requestUrl!!
+                        if (url.encodedPath != "/rest/search3.view") return MockResponse().setResponseCode(404)
+                        if (url.queryParameter("songCount") == "1") {
+                            probeCount.incrementAndGet()
+                            return MockResponse().setBody(songPage("probe", 0, 1))
+                        }
+                        val offset = url.queryParameter("songOffset")!!.toInt()
+                        requestedOffsets += offset
+                        val count = if (offset < 1000) 500 else 3
+                        return MockResponse().setBody(songPage("s", offset, count))
+                    }
+                }
+            val source = SubsonicSource(1, client())
+
+            val first = source.songs().toList().flatten()
+            val firstOffsets = requestedOffsets.toList()
+            val second = source.songs().toList().flatten()
+
+            assertEquals(1003, first.size)
+            assertEquals(1003, second.size)
+            assertEquals(1, probeCount.get())
+            assertEquals(listOf(0, 500, 1000), firstOffsets)
+        }
+
+    @Test
+    fun entitiesWithEmptyIdsAreDropped() =
+        runBlocking {
+            server.enqueue(
+                MockResponse().setBody(
+                    "<subsonic-response status=\"ok\"><artists>" +
+                        "<artist id=\"\" name=\"No id\"/>" +
+                        "<artist id=\"ar1\" name=\"Real\"/>" +
+                        "</artists></subsonic-response>",
+                ),
+            )
+
+            val artists = SubsonicSource(1, client()).artists().toList().flatten()
+
+            assertEquals(listOf("ar1"), artists.map { it.id })
+        }
+
+    @Test
+    fun cancellingAfterTheFirstBatchDoesNotHang() =
+        runBlocking {
+            val body =
+                buildString {
+                    append("<subsonic-response status=\"ok\"><artists>")
+                    repeat(100_000) { append("<artist id=\"ar$it\" name=\"A$it\" albumCount=\"1\"/>") }
+                    append("</artists></subsonic-response>")
+                }
+            server.enqueue(MockResponse().setBody(body))
+
+            val firstBatch = withTimeout(15_000) { SubsonicSource(1, client()).artists().first() }
+
+            assertEquals(500, firstBatch.size)
+        }
+
    @Test
    fun streamUriRequestsTranscodingWhenABitrateIsSet() {
        val uri = SubsonicSource(1, client(), maxBitrate = 128).streamUri("s1").toString()
@@ -122,6 +218,27 @@ class SubsonicSourceTest {
        message: String,
    ) = "<subsonic-response status=\"failed\"><error code=\"$code\" message=\"$message\"/></subsonic-response>"

+    private fun emptyAlbumList() = "<subsonic-response status=\"ok\"><albumList2></albumList2></subsonic-response>"
+
+    private fun albumPage(
+        offset: Int,
+        count: Int,
+    ) = "<subsonic-response status=\"ok\"><albumList2>" +
+        (offset until offset + count).joinToString("") {
+            "<album id=\"al-$it\" name=\"Album $it\" artist=\"Artist\" artistId=\"ar1\" songCount=\"1\"/>"
+        } +
+        "</albumList2></subsonic-response>"
+
+    private fun songPage(
+        prefix: String,
+        offset: Int,
+        count: Int,
+    ) = "<subsonic-response status=\"ok\"><searchResult3>" +
+        (offset until offset + count).joinToString("") {
+            "<song id=\"$prefix$it\" title=\"Song $it\"/>"
+        } +
+        "</searchResult3></subsonic-response>"
+
    private companion object {
        const val ALBUM_LIST =
            "<subsonic-response status=\"ok\" version=\"1.16.1\"><albumList2>" +
@@ -1,8 +1,11 @@
 package com.subtracks.data.source.subsonic

 import org.junit.Assert.assertEquals
+import org.junit.Assert.assertTrue
 import org.junit.Test
 import org.xml.sax.Attributes
+import java.io.FilterInputStream
+import java.io.InputStream

 class SubsonicXmlTest {
    private fun <T> parse(
@@ -156,4 +159,146 @@ class SubsonicXmlTest {
        assertEquals(true, SubsonicXml.containsEntity(xml.byteInputStream(Charsets.UTF_8), "song"))
        assertEquals(false, SubsonicXml.containsEntity(xml.byteInputStream(Charsets.UTF_8), "artist"))
    }
+
+    @Test
+    fun containsEntityStopsReadingAtTheFirstMatch() {
+        val xml =
+            buildString {
+                append("<subsonic-response status=\"ok\"><searchResult3>")
+                repeat(5_000) { append("<song id=\"s$it\" title=\"Song $it\"/>") }
+                append("</searchResult3></subsonic-response>")
+            }
+        val counting = CountingInputStream(xml.byteInputStream(Charsets.UTF_8))
+
+        assertEquals(true, SubsonicXml.containsEntity(counting, "song"))
+
+        assertTrue("read ${counting.bytesRead} of ${xml.length}", counting.bytesRead < 32_000)
+    }
+
+    @Test
+    fun rejectsATruncatedResponse() {
+        val failure =
+            runCatching {
+                SubsonicXml.readStatus("<subsonic-response status=\"ok\"><artists>".byteInputStream(Charsets.UTF_8))
+            }
+
+        assertEquals(true, failure.exceptionOrNull() is SubsonicException)
+    }
+
+    @Test
+    fun failedStatusWithoutAnErrorStillThrows() {
+        val failure =
+            runCatching {
+                SubsonicXml.readStatus(
+                    "<subsonic-response status=\"failed\"></subsonic-response>".byteInputStream(Charsets.UTF_8),
+                )
+            }
+
+        assertEquals(true, failure.exceptionOrNull() is SubsonicException)
+    }
+
+    @Test
+    fun errorElementWithoutStatusThrowsWithItsCode() {
+        val error =
+            runCatching {
+                SubsonicXml.readStatus(
+                    "<subsonic-response><error code=\"40\" message=\"Wrong username\"/></subsonic-response>"
+                        .byteInputStream(Charsets.UTF_8),
+                )
+            }.exceptionOrNull() as? SubsonicException
+
+        assertEquals(40, error?.code)
+        assertEquals("Wrong username", error?.message)
+    }
+
+    @Test
+    fun nestedSameNameElementsEmitOnce() {
+        val albums =
+            albums(
+                """
+                <subsonic-response status="ok">
+                  <albumList2>
+                    <album id="al1" name="Outer">
+                      <album id="al2" name="Inner"/>
+                    </album>
+                  </albumList2>
+                </subsonic-response>
+                """.trimIndent(),
+            )
+
+        assertEquals(listOf("al1"), albums.map { it.id })
+    }
+
+    @Test
+    fun ignoresDiscTitlesThatAreNotDirectChildren() {
+        val albums =
+            albums(
+                """
+                <subsonic-response status="ok">
+                  <albumList2>
+                    <album id="al1" name="The Wall">
+                      <wrapper><discTitles disc="1" title="The Calm"/></wrapper>
+                    </album>
+                  </albumList2>
+                </subsonic-response>
+                """.trimIndent(),
+            )
+
+        assertEquals(emptyMap<Long, String>(), albums.single().discTitles)
+    }
+
+    @Test
+    fun discTitleWithoutADiscNumberIsIgnored() {
+        val albums =
+            albums(
+                """
+                <subsonic-response status="ok">
+                  <albumList2>
+                    <album id="al1" name="The Wall">
+                      <discTitles title="No number"/>
+                      <discTitles disc="2" title="The Storm"/>
+                    </album>
+                  </albumList2>
+                </subsonic-response>
+                """.trimIndent(),
+            )
+
+        assertEquals(mapOf(2L to "The Storm"), albums.single().discTitles)
+    }
+
+    @Test
+    fun rejectsExternalEntities() {
+        val xml =
+            """
+            <!DOCTYPE subsonic-response [<!ENTITY xxe SYSTEM "file:///etc/passwd">]>
+            <subsonic-response status="ok"><artists><artist id="&xxe;" name="x"/></artists></subsonic-response>
+            """.trimIndent()
+        val failure =
+            runCatching {
+                SubsonicXml.readStatus(xml.byteInputStream(Charsets.UTF_8))
+            }
+
+        assertEquals(true, failure.exceptionOrNull() is SubsonicException)
+    }
+
+    private class CountingInputStream(
+        input: InputStream,
+    ) : FilterInputStream(input) {
+        var bytesRead = 0
+            private set
+
+        override fun read(): Int =
+            super.read().also {
+                if (it >= 0) bytesRead++
+            }
+
+        override fun read(
+            b: ByteArray,
+            off: Int,
+            len: Int,
+        ): Int =
+            super.read(b, off, len).also {
+                if (it > 0) bytesRead += it
+            }
+    }
 }
@@ -0,0 +1,60 @@
+package com.subtracks.data.sync
+
+import android.content.Context
+import androidx.room3.Room
+import androidx.sqlite.driver.bundled.BundledSQLiteDriver
+import androidx.test.core.app.ApplicationProvider
+import androidx.test.ext.junit.runners.AndroidJUnit4
+import com.subtracks.data.db.SubtracksDatabase
+import com.subtracks.data.prefs.fakeUserPreferences
+import com.subtracks.data.repo.QueueRepository
+import com.subtracks.data.repo.SourceRepository
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import okhttp3.OkHttpClient
+import org.junit.After
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertTrue
+import org.junit.Before
+import org.junit.Test
+import org.junit.runner.RunWith
+
+@RunWith(AndroidJUnit4::class)
+class SyncManagerTest {
+    private lateinit var db: SubtracksDatabase
+    private lateinit var sourceRepository: SourceRepository
+    private lateinit var queueRepository: QueueRepository
+    private lateinit var manager: SyncManager
+
+    @Before
+    fun setUp() {
+        val context = ApplicationProvider.getApplicationContext<Context>()
+        db =
+            Room
+                .inMemoryDatabaseBuilder(context, SubtracksDatabase::class.java)
+                .setDriver(BundledSQLiteDriver())
+                .build()
+        sourceRepository = SourceRepository(db, OkHttpClient(), fakeUserPreferences())
+        queueRepository = QueueRepository(db)
+        manager = SyncManager(sourceRepository, queueRepository)
+    }
+
+    @After
+    fun tearDown() {
+        sourceRepository.close()
+        db.close()
+    }
+
+    @Test
+    fun aFailedSyncStillInvalidatesTheLibraryCache() =
+        runBlocking {
+            val before = queueRepository.snapshot().version
+
+            manager.requestSync()
+            val status = withTimeout(10_000) { manager.status.first { it is SyncStatus.Failed } }
+
+            assertTrue(status is SyncStatus.Failed)
+            assertEquals(before + 1, queueRepository.snapshot().version)
+        }
+}
@@ -3,6 +3,7 @@ package com.subtracks.data.sync
 import android.content.Context
 import androidx.paging.PagingSource
 import androidx.room3.Room
+import androidx.room3.useWriterConnection
 import androidx.sqlite.driver.bundled.BundledSQLiteDriver
 import androidx.test.core.app.ApplicationProvider
 import androidx.test.ext.junit.runners.AndroidJUnit4
@@ -392,6 +393,167 @@ class SyncServiceTest {
            )
        }

+    @Test
+    fun playlistSongsAreFetchedOnlyForSurvivingPlaylists() =
+        runTest {
+            insertSource()
+            val source =
+                FakeMusicSource(
+                    songs = listOf(song("s1"), song("s2")),
+                    playlists = listOf(playlist("p1"), playlist("p2")),
+                    playlistSongs = listOf(PlaylistSong(1, "p1", "s1", 0), PlaylistSong(1, "p2", "s2", 0)),
+                )
+
+            SyncService(db, source).sync()
+
+            source.playlists = listOf(playlist("p1"))
+            source.playlistSongs = listOf(PlaylistSong(1, "p1", "s1", 0))
+
+            SyncService(db, source).sync()
+
+            assertEquals(listOf("p1"), source.requestedPlaylistIds.last())
+        }
+
+    @Test
+    fun playlistEmptiedToZeroRemovesAllSongs() =
+        runTest {
+            insertSource()
+            val source =
+                FakeMusicSource(
+                    songs = listOf(song("s1"), song("s2")),
+                    playlists = listOf(playlist("p1")),
+                    playlistSongs = listOf(PlaylistSong(1, "p1", "s1", 0), PlaylistSong(1, "p1", "s2", 1)),
+                )
+
+            SyncService(db, source).sync()
+            assertEquals(
+                2,
+                db
+                    .libraryDao()
+                    .playlistSongs(1, "p1")
+                    .allRows()
+                    .size,
+            )
+
+            source.playlistSongs = emptyList()
+
+            SyncService(db, source).sync()
+
+            assertEquals(1, db.libraryDao().playlistIds(1).size)
+            assertEquals(
+                0,
+                db
+                    .libraryDao()
+                    .playlistSongs(1, "p1")
+                    .allRows()
+                    .size,
+            )
+        }
+
+    @Test
+    fun discTitleSetShrinksForARetainedAlbum() =
+        runTest {
+            insertSource()
+            val source = FakeMusicSource(albums = listOf(album("al1").copy(discTitles = mapOf(1L to "A", 2L to "B"))))
+
+            SyncService(db, source).sync()
+            assertEquals(mapOf(1L to "A", 2L to "B"), storedDiscs("al1"))
+
+            source.albums = listOf(album("al1").copy(discTitles = mapOf(1L to "A")))
+
+            SyncService(db, source).sync()
+
+            assertEquals(mapOf(1L to "A"), storedDiscs("al1"))
+        }
+
+    @Test
+    fun syncingOneSourceDoesNotTouchAnother() =
+        runTest {
+            insertSource()
+            insertSource(id = 2, name = "other")
+            val sourceOne =
+                FakeMusicSource(
+                    artists = listOf(artist("a1"), artist("a2")),
+                    songs = listOf(song("s1")),
+                    playlists = listOf(playlist("p1")),
+                    playlistSongs = listOf(PlaylistSong(1, "p1", "s1", 0)),
+                )
+            val sourceTwo =
+                FakeMusicSource(
+                    id = 2,
+                    artists = listOf(artist("b1").copy(sourceId = 2)),
+                    songs = listOf(song("s2").copy(sourceId = 2)),
+                    playlists = listOf(playlist("p2").copy(sourceId = 2)),
+                    playlistSongs = listOf(PlaylistSong(2, "p2", "s2", 0)),
+                )
+            SyncService(db, sourceOne).sync()
+            SyncService(db, sourceTwo).sync()
+
+            sourceOne.artists = listOf(artist("a1"))
+            sourceOne.songs = emptyList()
+            sourceOne.playlists = emptyList()
+            sourceOne.playlistSongs = emptyList()
+
+            SyncService(db, sourceOne).sync()
+
+            assertEquals(listOf("b1"), db.libraryDao().artistIds(2))
+            assertEquals(listOf("s2"), db.libraryDao().songIds(2))
+            assertEquals(listOf("p2"), db.libraryDao().playlistIds(2))
+            assertEquals(
+                1,
+                db
+                    .libraryDao()
+                    .playlistSongs(2, "p2")
+                    .allRows()
+                    .size,
+            )
+        }
+
+    @Test
+    fun prunesMoreThanOneDeleteChunk() =
+        runTest {
+            insertSource()
+            val source = FakeMusicSource(artists = (1..1100).map { artist("a$it") })
+
+            SyncService(db, source).sync()
+            assertEquals(1100, db.libraryDao().artistIds(1).size)
+
+            source.artists = emptyList()
+
+            SyncService(db, source).sync()
+
+            assertEquals(0, db.libraryDao().artistIds(1).size)
+        }
+
+    @Test
+    fun anIdenticalSyncDoesNotRewriteRows() =
+        runTest {
+            insertSource()
+            val source =
+                FakeMusicSource(
+                    artists = listOf(artist("a1")),
+                    albums = listOf(album("al1")),
+                    songs = listOf(song("s1")),
+                    playlists = listOf(playlist("p1")),
+                    playlistSongs = listOf(PlaylistSong(1, "p1", "s1", 0)),
+                )
+
+            SyncService(db, source).sync()
+            val before = totalChanges()
+
+            SyncService(db, source).sync()
+
+            assertEquals(before, totalChanges())
+        }
+
+    private suspend fun totalChanges(): Long =
+        db.useWriterConnection { connection ->
+            connection.usePrepared("SELECT total_changes()") { statement ->
+                statement.step()
+                statement.getLong(0)
+            }
+        }
+
    private suspend fun storedDiscs(albumId: String): Map<Long, String> =
        db
            .libraryDao()
@@ -404,9 +566,12 @@ class SyncServiceTest {
        return (page as PagingSource.LoadResult.Page).data
    }

-    private suspend fun insertSource() {
+    private suspend fun insertSource(
+        id: Long = 1,
+        name: String = "test",
+    ) {
        db.sourcesDao().upsertSource(
-            Source(id = 1, name = "test", address = "http://localhost", isActive = true, createdAt = 0),
+            Source(id = id, name = name, address = "http://localhost/$id", isActive = id == 1L, createdAt = 0),
        )
    }

@@ -453,6 +618,7 @@ class SyncServiceTest {
 }

 private class FakeMusicSource(
+    override val id: Long = 1,
    var artists: List<Artist> = emptyList(),
    var albums: List<Album> = emptyList(),
    var songs: List<Song> = emptyList(),
@@ -460,7 +626,7 @@ private class FakeMusicSource(
    var playlistSongs: List<PlaylistSong> = emptyList(),
    private val batchSize: Int = 1000,
 ) : MusicSource {
-    override val id: Long = 1
+    val requestedPlaylistIds = mutableListOf<List<String>>()

    override suspend fun ping() = Unit

@@ -472,7 +638,10 @@ private class FakeMusicSource(

    override fun playlists(): Flow<List<Playlist>> = batches(playlists)

-    override fun playlistSongs(playlistIds: List<String>): Flow<List<PlaylistSong>> = batches(playlistSongs)
+    override fun playlistSongs(playlistIds: List<String>): Flow<List<PlaylistSong>> {
+        requestedPlaylistIds += playlistIds
+        return batches(playlistSongs)
+    }

    private fun <T> batches(items: List<T>): Flow<List<T>> =
        flow {