harden sync for large libraries: bounded prune, concurrent fetches, higher page cap

2026-11-11 07:29:14 +01:00 · 2026-10-04 00:29:59 +00:00
parent af38f34ecd
commit 34f15176d7
18 changed files with 324 additions and 125 deletions
@@ -65,18 +65,58 @@ interface LibraryDao {
    @Query("SELECT id FROM artists WHERE sourceId = :sourceId")
    suspend fun artistIds(sourceId: Long): List<String>

+    @Query("SELECT id FROM artists WHERE sourceId = :sourceId AND id > :afterId ORDER BY id LIMIT :limit")
+    suspend fun artistIdsAfter(
+        sourceId: Long,
+        afterId: String,
+        limit: Int,
+    ): List<String>
+
    @Query("SELECT id FROM albums WHERE sourceId = :sourceId")
    suspend fun albumIds(sourceId: Long): List<String>

+    @Query("SELECT id FROM albums WHERE sourceId = :sourceId AND id > :afterId ORDER BY id LIMIT :limit")
+    suspend fun albumIdsAfter(
+        sourceId: Long,
+        afterId: String,
+        limit: Int,
+    ): List<String>
+
    @Query("SELECT albumId, disc FROM discs WHERE sourceId = :sourceId")
    suspend fun discKeys(sourceId: Long): List<DiscKey>

+    @Query(
+        "SELECT albumId, disc FROM discs WHERE sourceId = :sourceId " +
+            "AND (albumId > :afterAlbumId OR (albumId = :afterAlbumId AND disc > :afterDisc)) " +
+            "ORDER BY albumId, disc LIMIT :limit",
+    )
+    suspend fun discKeysAfter(
+        sourceId: Long,
+        afterAlbumId: String,
+        afterDisc: Long,
+        limit: Int,
+    ): List<DiscKey>
+
    @Query("SELECT id FROM playlists WHERE sourceId = :sourceId")
    suspend fun playlistIds(sourceId: Long): List<String>

+    @Query("SELECT id FROM playlists WHERE sourceId = :sourceId AND id > :afterId ORDER BY id LIMIT :limit")
+    suspend fun playlistIdsAfter(
+        sourceId: Long,
+        afterId: String,
+        limit: Int,
+    ): List<String>
+
    @Query("SELECT id FROM songs WHERE sourceId = :sourceId")
    suspend fun songIds(sourceId: Long): List<String>

+    @Query("SELECT id FROM songs WHERE sourceId = :sourceId AND id > :afterId ORDER BY id LIMIT :limit")
+    suspend fun songIdsAfter(
+        sourceId: Long,
+        afterId: String,
+        limit: Int,
+    ): List<String>
+
    @Query("DELETE FROM artists WHERE sourceId = :sourceId AND id IN (:ids)")
    suspend fun deleteArtists(
        sourceId: Long,
@@ -127,6 +127,14 @@ val MIGRATION_12_13 =
        }
    }

+val MIGRATION_13_14 =
+    object : Migration(13, 14) {
+        override suspend fun migrate(connection: SQLiteConnection) {
+            connection.execSQL("ALTER TABLE `albums` DROP COLUMN `frequentRank`")
+            connection.execSQL("ALTER TABLE `albums` DROP COLUMN `recentRank`")
+        }
+    }
+
 val MIGRATIONS: Array<Migration> =
    arrayOf(
        MIGRATION_1_2,
@@ -141,4 +149,5 @@ val MIGRATIONS: Array<Migration> =
        MIGRATION_10_11,
        MIGRATION_11_12,
        MIGRATION_12_13,
+        MIGRATION_13_14,
    )
@@ -32,7 +32,7 @@ import com.subtracks.data.model.SubsonicSource
        ShuffleOrder::class,
        ArtworkSeed::class,
    ],
-    version = 13,
+    version = 14,
    exportSchema = true,
 )
@ColumnTypeConverters(QueueKindConverter::class)
@@ -84,8 +84,6 @@ data class Album(
    val year: Long?,
    val starred: Long?,
    val songCount: Long,
-    val frequentRank: Long?,
-    val recentRank: Long?,
    @Ignore val discTitles: Map<Long, String> = emptyMap(),
 )

@@ -9,7 +9,10 @@ import com.subtracks.data.source.MusicSource
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
 import kotlinx.coroutines.channels.trySendBlocking
+import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.buffer
 import kotlinx.coroutines.flow.channelFlow
@@ -17,6 +20,9 @@ import kotlinx.coroutines.flow.emitAll
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.flowOn
 import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.Semaphore
+import kotlinx.coroutines.sync.withPermit
 import kotlinx.coroutines.withContext
 import org.xml.sax.Attributes

@@ -43,7 +49,6 @@ class SubsonicSource(

    override fun albums(): Flow<List<Album>> =
        flow {
-            val (frequent, recent) = fetchRanks()
            var offset = 0
            var pages = 0
            while (true) {
@@ -57,16 +62,7 @@ class SubsonicSource(
                ).collect { batch ->
                    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 (accepted.isNotEmpty()) emit(accepted.map { it.toAlbum() })
                }
                if (raw < PAGE_SIZE) {
                    if (raw > 0 && pages >= maxPages) throw pageCapExceeded()
@@ -95,20 +91,25 @@ class SubsonicSource(
        ).map { batch -> batch.filter { it.id.isNotEmpty() } }

    override fun playlistSongs(playlistIds: List<String>): Flow<List<PlaylistSong>> =
-        flow {
-            for (playlistId in playlistIds) {
-                var position = 0L
-                entityBatches(
-                    method = "getPlaylist",
-                    params = mapOf("id" to playlistId),
-                    tag = "entry",
-                    create = { attrs ->
-                        val entryId = attrs.attr("id")
-                        SubsonicXml.playlistSong(id, playlistId, if (entryId.isEmpty()) -1L else position++, attrs)
-                    },
-                ).collect { batch ->
-                    val accepted = batch.filter { it.songId.isNotEmpty() }
-                    if (accepted.isNotEmpty()) emit(accepted)
+        channelFlow {
+            val permits = Semaphore(MAX_CONCURRENT_FETCHES)
+            playlistIds.forEach { playlistId ->
+                launch {
+                    permits.withPermit {
+                        var position = 0L
+                        entityBatches(
+                            method = "getPlaylist",
+                            params = mapOf("id" to playlistId),
+                            tag = "entry",
+                            create = { attrs ->
+                                val entryId = attrs.attr("id")
+                                SubsonicXml.playlistSong(id, playlistId, if (entryId.isEmpty()) -1L else position++, attrs)
+                            },
+                        ).collect { batch ->
+                            val accepted = batch.filter { it.songId.isNotEmpty() }
+                            if (accepted.isNotEmpty()) send(accepted)
+                        }
+                    }
                }
            }
        }.flowOn(Dispatchers.IO).buffer(1)
@@ -174,7 +175,8 @@ class SubsonicSource(
        }

    private fun albumSongs(): Flow<List<Song>> =
-        flow {
+        channelFlow {
+            val permits = Semaphore(MAX_CONCURRENT_FETCHES)
            var offset = 0
            var pages = 0
            while (true) {
@@ -189,16 +191,23 @@ class SubsonicSource(
                    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) },
-                    ).collect { batch ->
-                        val accepted = batch.filter { it.id.isNotEmpty() }
-                        if (accepted.isNotEmpty()) emit(accepted)
-                    }
+                coroutineScope {
+                    albums
+                        .map { albumId ->
+                            async {
+                                permits.withPermit {
+                                    entityBatches(
+                                        method = "getAlbum",
+                                        params = mapOf("id" to albumId),
+                                        tag = "song",
+                                        create = { SubsonicXml.song(id, it) },
+                                    ).collect { batch ->
+                                        val accepted = batch.filter { it.id.isNotEmpty() }
+                                        if (accepted.isNotEmpty()) send(accepted)
+                                    }
+                                }
+                            }
+                        }.awaitAll()
                }
                if (raw < PAGE_SIZE) {
                    if (raw > 0 && pages >= maxPages) throw pageCapExceeded()
@@ -210,44 +219,6 @@ class SubsonicSource(
            }
        }

-    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> =
-        try {
-            val ranks = HashMap<String, Long>()
-            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") },
-                ).collect { batch ->
-                    raw += batch.size
-                    batch.forEach { albumId ->
-                        if (albumId.isNotEmpty()) {
-                            ranks[albumId] = (offset + index).toLong()
-                            index++
-                        }
-                    }
-                }
-                if (raw < PAGE_SIZE) {
-                    if (raw > 0 && pages >= maxPages) throw pageCapExceeded()
-                    break
-                }
-                pages++
-                if (pages > maxPages) throw pageCapExceeded()
-                offset += PAGE_SIZE
-            }
-            ranks
-        } catch (failure: SubsonicException) {
-            if (failure is PageCapExceeded) throw failure
-            emptyMap()
-        }
-
    private fun supportsEmptyQuerySearch(): Boolean {
        emptyQuerySearchSupported?.let { return it }
        val supported =
@@ -263,7 +234,7 @@ class SubsonicSource(
        return supported
    }

-    private fun pageCapExceeded() = PageCapExceeded("Library exceeds the ${maxPages * PAGE_SIZE}-row page cap")
+    private fun pageCapExceeded() = PageCapExceeded("Library is too large to sync: it exceeds ${maxPages.toLong() * PAGE_SIZE} rows")

    private fun <T> entityBatches(
        method: String,
@@ -310,7 +281,8 @@ class SubsonicSource(

    private companion object {
        const val PAGE_SIZE = 500
-        const val MAX_PAGES = 1000
+        const val MAX_PAGES = 20_000
+        const val MAX_CONCURRENT_FETCHES = 4
        const val THUMBNAIL_SIZE = 256
    }
 }
@@ -155,8 +155,6 @@ internal class AlbumDraft(
            year = year,
            starred = starred,
            songCount = songCount,
-            frequentRank = null,
-            recentRank = null,
            discTitles = discTitles.toMap(),
        )
 }
@@ -13,6 +13,7 @@ import com.subtracks.data.model.Playlist
 import com.subtracks.data.model.PlaylistSong
 import com.subtracks.data.model.Song
 import com.subtracks.data.source.MusicSource
+import java.util.Arrays

 class SyncService(
    private val db: SubtracksDatabase,
@@ -28,19 +29,19 @@ class SyncService(

    private suspend fun syncArtists() {
        val library = db.libraryDao()
-        val stale = library.artistIds(source.id).toHashSet()
+        val seen = HashedIds()
        source.artists().collect { batch ->
            if (batch.isEmpty()) return@collect
            write("artists", artistColumns, primaryKey, batch) { it.values() }
-            stale -= batch.mapTo(HashSet()) { it.id }
+            batch.forEach { seen.add(idHash(it.id)) }
        }
-        deleteStale(stale) { library.deleteArtists(source.id, it) }
+        pruneStaleIds(seen, { library.artistIdsAfter(source.id, it, PRUNE_PAGE) }, { library.deleteArtists(source.id, it) })
    }

    private suspend fun syncAlbums() {
        val library = db.libraryDao()
-        val staleAlbums = library.albumIds(source.id).toHashSet()
-        val staleDiscs = library.discKeys(source.id).toHashSet()
+        val seenAlbums = HashedIds()
+        val seenDiscs = HashedIds()
        source.albums().collect { batch ->
            if (batch.isEmpty()) return@collect
            val discs =
@@ -48,33 +49,35 @@ class SyncService(
                    album.discTitles.map { (disc, title) -> Disc(album.sourceId, album.id, disc, title) }
                }
            writeAlbumsAndDiscs(batch, discs)
-            staleAlbums -= batch.mapTo(HashSet()) { it.id }
-            staleDiscs -= discs.mapTo(HashSet()) { DiscKey(it.albumId, it.disc) }
+            batch.forEach { seenAlbums.add(idHash(it.id)) }
+            discs.forEach { seenDiscs.add(discHash(it.albumId, it.disc)) }
+        }
+        pruneStaleIds(seenAlbums, { library.albumIdsAfter(source.id, it, PRUNE_PAGE) }, { library.deleteAlbums(source.id, it) })
+        pruneStaleDiscs(seenDiscs, { albumId, disc -> library.discKeysAfter(source.id, albumId, disc, PRUNE_PAGE) }) {
+            deleteStaleDiscs(it)
        }
-        deleteStale(staleAlbums) { library.deleteAlbums(source.id, it) }
-        deleteStaleDiscs(staleDiscs)
    }

    private suspend fun syncSongs() {
        val library = db.libraryDao()
-        val stale = library.songIds(source.id).toHashSet()
+        val seen = HashedIds()
        source.songs().collect { batch ->
            if (batch.isEmpty()) return@collect
            write("songs", songColumns, primaryKey, batch) { it.values() }
-            stale -= batch.mapTo(HashSet()) { it.id }
+            batch.forEach { seen.add(idHash(it.id)) }
        }
-        deleteStale(stale) { library.deleteSongs(source.id, it) }
+        pruneStaleIds(seen, { library.songIdsAfter(source.id, it, PRUNE_PAGE) }, { library.deleteSongs(source.id, it) })
    }

    private suspend fun syncPlaylists() {
        val library = db.libraryDao()
-        val stale = library.playlistIds(source.id).toHashSet()
+        val seen = HashedIds()
        source.playlists().collect { batch ->
            if (batch.isEmpty()) return@collect
            write("playlists", playlistColumns, primaryKey, batch) { it.values() }
-            stale -= batch.mapTo(HashSet()) { it.id }
+            batch.forEach { seen.add(idHash(it.id)) }
        }
-        deleteStale(stale) { library.deletePlaylists(source.id, it) }
+        pruneStaleIds(seen, { library.playlistIdsAfter(source.id, it, PRUNE_PAGE) }, { library.deletePlaylists(source.id, it) })
    }

    private suspend fun syncPlaylistSongs() {
@@ -118,7 +121,7 @@ class SyncService(
        }
    }

-    private suspend fun deleteStaleDiscs(stale: Set<DiscKey>) {
+    private suspend fun deleteStaleDiscs(stale: Collection<DiscKey>) {
        stale.chunked(DELETE_CHUNK).forEach { chunk ->
            db.useWriterConnection { connection ->
                connection.immediateTransaction {
@@ -137,15 +140,77 @@ class SyncService(
        }
    }

-    private suspend fun deleteStale(
-        stale: Set<String>,
+    private suspend fun pruneStaleIds(
+        seen: HashedIds,
+        pageOf: suspend (afterId: String) -> List<String>,
        delete: suspend (Collection<String>) -> Unit,
    ) {
-        stale.chunked(DELETE_CHUNK).forEach { delete(it) }
+        seen.freeze()
+        var afterId = ""
+        while (true) {
+            val ids = pageOf(afterId)
+            if (ids.isEmpty()) return
+            val stale = ids.filterNot { seen.contains(idHash(it)) }
+            stale.chunked(DELETE_CHUNK).forEach { delete(it) }
+            if (ids.size < PRUNE_PAGE) return
+            afterId = ids.last()
+        }
+    }
+
+    private suspend fun pruneStaleDiscs(
+        seen: HashedIds,
+        pageOf: suspend (afterAlbumId: String, afterDisc: Long) -> List<DiscKey>,
+        delete: suspend (Collection<DiscKey>) -> Unit,
+    ) {
+        seen.freeze()
+        var afterAlbumId = ""
+        var afterDisc = 0L
+        while (true) {
+            val keys = pageOf(afterAlbumId, afterDisc)
+            if (keys.isEmpty()) return
+            val stale = keys.filterNot { seen.contains(discHash(it.albumId, it.disc)) }
+            stale.chunked(DELETE_CHUNK).forEach { delete(it) }
+            if (keys.size < PRUNE_PAGE) return
+            afterAlbumId = keys.last().albumId
+            afterDisc = keys.last().disc
+        }
    }
 }

 private const val DELETE_CHUNK = 500
+private const val PRUNE_PAGE = 500
+private const val INITIAL_HASH_CAPACITY = 256
+private const val FNV_OFFSET = -3750763034362895579L
+private const val FNV_PRIME = 1099511628211L
+
+private class HashedIds {
+    private var values = LongArray(0)
+    private var count = 0
+
+    fun add(value: Long) {
+        if (count == values.size) {
+            values = values.copyOf(if (values.isEmpty()) INITIAL_HASH_CAPACITY else values.size * 2)
+        }
+        values[count++] = value
+    }
+
+    fun freeze() {
+        Arrays.sort(values, 0, count)
+    }
+
+    operator fun contains(value: Long): Boolean = Arrays.binarySearch(values, 0, count, value) >= 0
+}
+
+private fun idHash(id: String): Long {
+    var hash = FNV_OFFSET
+    id.forEach { hash = (hash xor it.code.toLong()) * FNV_PRIME }
+    return hash
+}
+
+private fun discHash(
+    albumId: String,
+    disc: Long,
+): Long = idHash(albumId) * 31 + disc

 private val primaryKey = listOf("sourceId", "id")

@@ -164,8 +229,6 @@ private val albumColumns =
        "year",
        "starred",
        "songCount",
-        "frequentRank",
-        "recentRank",
    )

 private val discKey = listOf("sourceId", "albumId", "disc")
@@ -197,8 +260,7 @@ private val playlistSongKey = listOf("sourceId", "playlistId", "position")

 private fun Artist.values() = listOf<Any?>(sourceId, id, name, albumCount, starred, coverArt)

-private fun Album.values() =
-    listOf<Any?>(sourceId, id, artistId, name, albumArtist, created, coverArt, genre, year, starred, songCount, frequentRank, recentRank)
+private fun Album.values() = listOf<Any?>(sourceId, id, artistId, name, albumArtist, created, coverArt, genre, year, starred, songCount)

 private fun Disc.values() = listOf<Any?>(sourceId, albumId, disc, title)