stream sync with SAX and write the library per batch

The DOM-based sync materialised the whole library in memory before
writing: SubsonicSource accumulated every paged response into one list
and SubsonicClient built a full DOM per response, so peak memory grew
with the library size (hundreds of MB for large libraries).

Replace it with a streaming pipeline:

- SubsonicXml is now a SAX reader that maps entities straight from the
  stream. Album disc titles are accumulated from child elements and
  attached when the album closes.
- SubsonicClient exposes stream()/check() instead of returning a DOM.
- MusicSource returns batches as Flow<List<T>>; SubsonicSource emits one
  batch per response page with backpressure so only one batch is held.
- SyncService syncs each entity sequentially, commits one transaction
  per batch, and prunes with an existing-minus-seen stale set. Playlist
  song tails are pruned once the whole flow has been counted, and
  orphaned playlist_songs rows are removed with a single delete.

Sync is no longer atomic: on failure the committed batches remain and
the queue's library cache is now invalidated on failure too.
This commit is contained in:
austinried
2026-10-04 00:29:58 +00:00
parent bce074ecdc
commit b171b0b33c
13 changed files with 804 additions and 331 deletions
@@ -38,7 +38,7 @@ class SubsonicClientTest {
fun tokenAuthSendsSaltAndHash() {
server.enqueue(MockResponse().setBody("<subsonic-response status=\"ok\" version=\"1.16.1\"/>"))
client(tokenAuth = true).get("ping")
client(tokenAuth = true).check("ping")
val url = server.takeRequest().requestUrl!!
assertEquals("/rest/ping.view", url.encodedPath)
@@ -53,7 +53,7 @@ class SubsonicClientTest {
fun plaintextAuthSendsPassword() {
server.enqueue(MockResponse().setBody("<subsonic-response status=\"ok\" version=\"1.16.1\"/>"))
client(tokenAuth = false).get("ping")
client(tokenAuth = false).check("ping")
val url = server.takeRequest().requestUrl!!
assertEquals("secret", url.queryParameter("p"))
@@ -64,7 +64,7 @@ class SubsonicClientTest {
fun paramsAreForwarded() {
server.enqueue(MockResponse().setBody("<subsonic-response status=\"ok\" version=\"1.16.1\"/>"))
client(tokenAuth = true).get("getCoverArt", mapOf("id" to "cov", "size" to "256"))
client(tokenAuth = true).check("getCoverArt", mapOf("id" to "cov", "size" to "256"))
val url = server.takeRequest().requestUrl!!
assertEquals("cov", url.queryParameter("id"))
@@ -79,7 +79,7 @@ class SubsonicClientTest {
),
)
val error = assertThrows(SubsonicException::class.java) { client(tokenAuth = true).get("ping") }
val error = assertThrows(SubsonicException::class.java) { client(tokenAuth = true).check("ping") }
assertEquals(40, error.code)
assertEquals("Wrong username", error.message)
}
@@ -88,6 +88,6 @@ class SubsonicClientTest {
fun nonSubsonicResponseThrows() {
server.enqueue(MockResponse().setBody("<html><body>Sign in</body></html>"))
assertThrows(SubsonicException::class.java) { client(tokenAuth = true).get("ping") }
assertThrows(SubsonicException::class.java) { client(tokenAuth = true).check("ping") }
}
}
@@ -1,5 +1,6 @@
package com.subtracks.data.source.subsonic
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.runBlocking
import okhttp3.OkHttpClient
import okhttp3.mockwebserver.Dispatcher
@@ -53,7 +54,33 @@ class SubsonicSourceTest {
}
val source = SubsonicSource(1, client())
assertEquals(listOf("s1", "s2"), source.getSongs().map { it.id })
assertEquals(
listOf("s1", "s2"),
source
.songs()
.toList()
.flatten()
.map { it.id },
)
}
@Test
fun artistsAreEmittedInBoundedBatches() =
runBlocking {
val count = 1200
val body =
buildString {
append("<subsonic-response status=\"ok\"><artists>")
repeat(count) { index ->
append("<artist id=\"ar$index\" name=\"Artist $index\" albumCount=\"1\"/>")
}
append("</artists></subsonic-response>")
}
server.enqueue(MockResponse().setBody(body))
val sizes = SubsonicSource(1, client()).artists().toList().map { it.size }
assertEquals(listOf(500, 500, 200), sizes)
}
@Test
@@ -2,19 +2,37 @@ package com.subtracks.data.source.subsonic
import org.junit.Assert.assertEquals
import org.junit.Test
import org.w3c.dom.Document
import javax.xml.parsers.DocumentBuilderFactory
import org.xml.sax.Attributes
class SubsonicXmlTest {
private fun parse(xml: String): Document =
DocumentBuilderFactory
.newInstance()
.newDocumentBuilder()
.parse(xml.byteInputStream(Charsets.UTF_8))
private fun <T> parse(
xml: String,
tag: String,
create: (Attributes) -> T,
onChild: (T, String, Attributes) -> Unit = { _, _, _ -> },
): List<T> {
val result = mutableListOf<T>()
SubsonicXml.readEntities(
input = xml.byteInputStream(Charsets.UTF_8),
entityTag = tag,
create = create,
onChild = onChild,
onEnd = { result += it },
)
return result
}
private fun albums(xml: String) =
parse(
xml,
tag = "album",
create = { SubsonicXml.albumDraft(1, it) },
onChild = { draft, name, attrs -> if (name == "discTitles") draft.addDiscTitle(attrs) },
).map { it.toAlbum() }
@Test
fun mapsArtists() {
val document =
val artists =
parse(
"""
<subsonic-response status="ok">
@@ -24,10 +42,10 @@ class SubsonicXmlTest {
</artists>
</subsonic-response>
""".trimIndent(),
tag = "artist",
create = { SubsonicXml.artist(7, it) },
)
val artists = SubsonicXml.artists(7, document)
assertEquals(2, artists.size)
assertEquals("Radiohead", artists[0].name)
assertEquals(7L, artists[0].sourceId)
@@ -38,8 +56,8 @@ class SubsonicXmlTest {
@Test
fun mapsAlbumList() {
val document =
parse(
val albums =
albums(
"""
<subsonic-response status="ok">
<albumList2>
@@ -50,8 +68,6 @@ class SubsonicXmlTest {
""".trimIndent(),
)
val albums = SubsonicXml.albums(1, document)
assertEquals(2, albums.size)
assertEquals("OK Computer", albums[0].name)
assertEquals("Kid A", albums[1].name)
@@ -62,8 +78,8 @@ class SubsonicXmlTest {
@Test
fun mapsAlbumDiscTitles() {
val document =
parse(
val albums =
albums(
"""
<subsonic-response status="ok">
<albumList2>
@@ -77,8 +93,6 @@ class SubsonicXmlTest {
""".trimIndent(),
)
val albums = SubsonicXml.albums(1, document)
assertEquals(mapOf(1L to "The Calm", 2L to "The Storm"), albums.single().discTitles)
}
@@ -94,28 +108,52 @@ class SubsonicXmlTest {
@Test
fun mapsSongsAndPlaylistEntries() {
val document =
parse(
"""
<subsonic-response status="ok">
<searchResult3>
<song id="sg2" title="Idioteque" artist="Radiohead" album="Kid A" albumId="al2" artistId="ar1" duration="300" track="8" discNumber="1"/>
</searchResult3>
<playlist id="pl1" name="Favourites" songCount="1" created="2021-02-03T04:05:06.000Z">
<entry id="sg1" title="Everything In Its Right Place" artist="Radiohead" album="Kid A" albumId="al2" artistId="ar1" duration="251" track="1" discNumber="1"/>
</playlist>
</subsonic-response>
""".trimIndent(),
)
val xml =
"""
<subsonic-response status="ok">
<searchResult3>
<song id="sg2" title="Idioteque" artist="Radiohead" album="Kid A" albumId="al2" artistId="ar1" duration="300" track="8" discNumber="1"/>
</searchResult3>
<playlist id="pl1" name="Favourites" songCount="1" created="2021-02-03T04:05:06.000Z">
<entry id="sg1" title="Everything In Its Right Place" artist="Radiohead" album="Kid A" albumId="al2" artistId="ar1" duration="251" track="1" discNumber="1"/>
</playlist>
</subsonic-response>
""".trimIndent()
val songs = SubsonicXml.songs(1, document)
val entries = SubsonicXml.playlistSongs(1, "pl1", document)
val songs = parse(xml, tag = "song", create = { SubsonicXml.song(1, it) })
var position = 0L
val entries = parse(xml, tag = "entry", create = { SubsonicXml.playlistSong(1, "pl1", position++, it) })
assertEquals(1, songs.size)
assertEquals("Idioteque", songs[0].title)
assertEquals(300L, songs[0].duration)
assertEquals(0L, entries[0].position)
assertEquals("sg1", entries[0].songId)
assertEquals(251L, SubsonicXml.song(1, document.getElementsByTagName("entry").item(0) as org.w3c.dom.Element).duration)
}
@Test
fun statusOnlyReadRejectsAnHtmlLoginPage() {
val failure =
runCatching {
SubsonicXml.readStatus("<html><body>Sign in</body></html>".byteInputStream(Charsets.UTF_8))
}
assertEquals(true, failure.exceptionOrNull() is SubsonicException)
}
@Test
fun containsEntityStopsAtTheFirstMatch() {
val xml =
"""
<subsonic-response status="ok">
<searchResult3>
<song id="s1" title="One"/>
<song id="s2" title="Two"/>
</searchResult3>
</subsonic-response>
""".trimIndent()
assertEquals(true, SubsonicXml.containsEntity(xml.byteInputStream(Charsets.UTF_8), "song"))
assertEquals(false, SubsonicXml.containsEntity(xml.byteInputStream(Charsets.UTF_8), "artist"))
}
}
@@ -14,7 +14,9 @@ import com.subtracks.data.model.PlaylistSong
import com.subtracks.data.model.Song
import com.subtracks.data.model.Source
import com.subtracks.data.source.MusicSource
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.test.runTest
import org.junit.After
import org.junit.Assert.assertEquals
@@ -293,6 +295,103 @@ class SyncServiceTest {
assertEquals(mapOf(1L to "New"), storedDiscs("al1"))
}
@Test
fun syncCommitsMultipleBatchesAndPrunes() =
runTest {
insertSource()
val source =
FakeMusicSource(
artists = (1..5).map { artist("a$it") },
songs = (1..5).map { song("s$it") },
batchSize = 2,
)
SyncService(db, source).sync()
assertEquals((1..5).map { "a$it" }, db.libraryDao().artistIds(1).sorted())
assertEquals((1..5).map { "s$it" }, db.libraryDao().songIds(1).sorted())
source.artists = (1..3).map { artist("a$it") }
source.songs = (1..3).map { song("s$it") }
SyncService(db, source).sync()
assertEquals((1..3).map { "a$it" }, db.libraryDao().artistIds(1).sorted())
assertEquals((1..3).map { "s$it" }, db.libraryDao().songIds(1).sorted())
}
@Test
fun playlistSongsSplitAcrossBatchesKeepTheirTail() =
runTest {
insertSource()
val source =
FakeMusicSource(
songs = (1..5).map { song("s$it") },
playlists = listOf(playlist("p1")),
playlistSongs = (1..5).map { PlaylistSong(1, "p1", "s$it", (it - 1).toLong()) },
batchSize = 2,
)
SyncService(db, source).sync()
assertEquals(
5,
db
.libraryDao()
.playlistSongs(1, "p1")
.allRows()
.size,
)
source.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,
)
}
@Test
fun removingAPlaylistRemovesItsSongs() =
runTest {
insertSource()
val source =
FakeMusicSource(
songs = listOf(song("s1")),
playlists = listOf(playlist("p1")),
playlistSongs = listOf(PlaylistSong(1, "p1", "s1", 0)),
)
SyncService(db, source).sync()
assertEquals(
1,
db
.libraryDao()
.playlistSongs(1, "p1")
.allRows()
.size,
)
source.playlists = emptyList()
source.playlistSongs = emptyList()
SyncService(db, source).sync()
assertEquals(0, db.libraryDao().playlistIds(1).size)
assertEquals(
0,
db
.libraryDao()
.playlistSongs(1, "p1")
.allRows()
.size,
)
}
private suspend fun storedDiscs(albumId: String): Map<Long, String> =
db
.libraryDao()
@@ -359,18 +458,24 @@ private class FakeMusicSource(
var songs: List<Song> = emptyList(),
var playlists: List<Playlist> = emptyList(),
var playlistSongs: List<PlaylistSong> = emptyList(),
private val batchSize: Int = 1000,
) : MusicSource {
override val id: Long = 1
override suspend fun ping() = Unit
override suspend fun getArtists(): List<Artist> = artists
override fun artists(): Flow<List<Artist>> = batches(artists)
override suspend fun getAlbums(): List<Album> = albums
override fun albums(): Flow<List<Album>> = batches(albums)
override suspend fun getSongs(): List<Song> = songs
override fun songs(): Flow<List<Song>> = batches(songs)
override suspend fun getPlaylists(): List<Playlist> = playlists
override fun playlists(): Flow<List<Playlist>> = batches(playlists)
override suspend fun getPlaylistSongs(playlists: List<Playlist>): List<PlaylistSong> = playlistSongs
override fun playlistSongs(playlistIds: List<String>): Flow<List<PlaylistSong>> = batches(playlistSongs)
private fun <T> batches(items: List<T>): Flow<List<T>> =
flow {
items.chunked(batchSize).forEach { emit(it) }
}
}