fix(android): sync pulls every page of the change log, and a 404 drops the cached row (#5654)
release / govulncheck (push) Successful in 32s
release / web (push) Successful in 1m21s
release / go (push) Successful in 1m37s
release / android (push) Failing after 3m58s
release / Attach APK to the Release (tag releases only) (push) Canceled after 0s
release / Build + push container image (push) Canceled after 0s
release / Verify release artifacts (tag releases only) (push) Canceled after 0s
release / integration (push) Canceled after 4m38s
release / Build signed APK (releases and dev) (push) Canceled after 4m39s
release / govulncheck (push) Successful in 32s
release / web (push) Successful in 1m21s
release / go (push) Successful in 1m37s
release / android (push) Failing after 3m58s
release / Attach APK to the Release (tag releases only) (push) Canceled after 0s
release / Build + push container image (push) Canceled after 0s
release / Verify release artifacts (tag releases only) (push) Canceled after 0s
release / integration (push) Canceled after 4m38s
release / Build signed APK (releases and dev) (push) Canceled after 4m39s
The server returns at most 1000 changes per /api/library/sync call and expects clients to loop with the returned cursor. SyncController took one page per trigger (cold start, reconnect), so the cache fell behind a log that a full re-read grows by ~26,000 rows and, past the 30-day compaction, reset through 410 to cursor 0 without ever reaching the merge deletes of 2026-10-09/10. The freshness sweep then re-asked for those tracks every 30 minutes: the same ~100 404s on every pass. - SyncController pages until 204, storing the cursor after each page, and stops on a cursor that does not advance. - MetadataProvider drops a cached album, artist or track the server answers 404 for, so rows orphaned before this fix stop being refetched. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
+33
-10
@@ -109,19 +109,36 @@ class SyncController @Inject constructor(
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Pulls pages until the server has nothing newer (#5654). The server
|
||||
* returns at most 1000 changes a call; taking one page per trigger left
|
||||
* the cache behind a change log that a full re-read grows by ~26,000
|
||||
* rows, so merge deletes never arrived and the freshness sweep kept
|
||||
* asking for tracks that were gone. The cursor is stored after each
|
||||
* page, so an interrupted pass resumes where it stopped.
|
||||
*/
|
||||
private suspend fun sync() {
|
||||
repeat(MAX_PAGES_PER_PASS) {
|
||||
if (!syncPage()) return
|
||||
}
|
||||
}
|
||||
|
||||
/** One page of the change log. True when the server may hold more. */
|
||||
private suspend fun syncPage(): Boolean {
|
||||
val meta = syncMetadataDao.get()
|
||||
val cursor = meta?.cursor ?: 0L
|
||||
val response = api.sync(cursor)
|
||||
when (response.code()) {
|
||||
return when (response.code()) {
|
||||
HTTP_NO_CONTENT -> {
|
||||
touchLastSyncAt(cursor)
|
||||
false
|
||||
}
|
||||
HTTP_GONE -> {
|
||||
wipeAndRetry()
|
||||
resetCursor()
|
||||
true
|
||||
}
|
||||
HTTP_OK -> {
|
||||
val body = response.body() ?: return
|
||||
val body = response.body() ?: return false
|
||||
applyDeltas(body.upserts.artist, body.upserts.album, body.upserts.track)
|
||||
applyDeletes(body.deletes.artist, body.deletes.album, body.deletes.track)
|
||||
syncMetadataDao.upsert(
|
||||
@@ -131,10 +148,14 @@ class SyncController @Inject constructor(
|
||||
lastSyncAt = Clock.System.now(),
|
||||
),
|
||||
)
|
||||
// A cursor that did not advance would ask for the same page
|
||||
// again; stop rather than spin.
|
||||
body.cursor > cursor
|
||||
}
|
||||
else -> {
|
||||
// Unhandled status — treat as a no-op for v1; the next
|
||||
// app start retries from the same cursor.
|
||||
false
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -165,25 +186,27 @@ class SyncController @Inject constructor(
|
||||
)
|
||||
}
|
||||
|
||||
private suspend fun wipeAndRetry() {
|
||||
private suspend fun resetCursor() {
|
||||
// The Library DAOs only expose deleteByIds; for the rare 410
|
||||
// path we re-issue the same request with cursor=0 after
|
||||
// resetting our cursor. The next response carries the full
|
||||
// path the next page is requested with cursor=0 after
|
||||
// resetting our cursor. The following pages carry the full
|
||||
// entity set, which upsertAll overwrites the existing rows
|
||||
// with — old rows for entities the server no longer knows
|
||||
// about will linger until a later 410 or until the user clears
|
||||
// app data. Acceptable for v1; a future "wipe all" DAO method
|
||||
// tightens this up.
|
||||
// about linger until MetadataProvider's refresh meets their
|
||||
// 404 and drops them.
|
||||
syncMetadataDao.upsert(
|
||||
SyncMetadataEntity(id = 0, cursor = 0, lastSyncAt = null),
|
||||
)
|
||||
sync()
|
||||
}
|
||||
|
||||
private companion object {
|
||||
const val HTTP_OK = 200
|
||||
const val HTTP_NO_CONTENT = 204
|
||||
const val HTTP_GONE = 410
|
||||
|
||||
// A guard, not a budget: 1000 pages is a million changes, far past
|
||||
// anything a library writes between two app starts.
|
||||
const val MAX_PAGES_PER_PASS = 1000
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -18,11 +18,13 @@ import kotlinx.coroutines.flow.onEach
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.sync.Semaphore
|
||||
import kotlinx.coroutines.sync.withPermit
|
||||
import retrofit2.HttpException
|
||||
import java.util.concurrent.ConcurrentHashMap
|
||||
import javax.inject.Inject
|
||||
import javax.inject.Singleton
|
||||
|
||||
private const val MAX_CONCURRENT_FETCHES = 4
|
||||
private const val HTTP_NOT_FOUND = 404
|
||||
|
||||
/**
|
||||
* Cache-first metadata reads with eager on-miss live-fetch.
|
||||
@@ -93,26 +95,47 @@ class MetadataProvider @Inject constructor(
|
||||
fun refreshArtist(id: String): Job = fetchArtist(id)
|
||||
fun refreshTrack(id: String): Job = fetchTrack(id)
|
||||
|
||||
private fun fetchAlbum(id: String): Job = launchDeduped(inFlightAlbums, id) {
|
||||
private fun fetchAlbum(id: String): Job = launchDeduped(
|
||||
inFlightAlbums,
|
||||
id,
|
||||
onGone = { albumDao.deleteByIds(listOf(id)) },
|
||||
) {
|
||||
libraryRepository.refreshAlbumDetail(id)
|
||||
}
|
||||
|
||||
private fun fetchArtist(id: String): Job = launchDeduped(inFlightArtists, id) {
|
||||
private fun fetchArtist(id: String): Job = launchDeduped(
|
||||
inFlightArtists,
|
||||
id,
|
||||
onGone = { artistDao.deleteByIds(listOf(id)) },
|
||||
) {
|
||||
libraryRepository.refreshArtistDetail(id)
|
||||
}
|
||||
|
||||
private fun fetchTrack(id: String): Job = launchDeduped(inFlightTracks, id) {
|
||||
private fun fetchTrack(id: String): Job = launchDeduped(
|
||||
inFlightTracks,
|
||||
id,
|
||||
onGone = { trackDao.deleteByIds(listOf(id)) },
|
||||
) {
|
||||
libraryRepository.refreshTrack(id)
|
||||
}
|
||||
|
||||
/**
|
||||
* [onGone] runs when the server answers 404: the row no longer exists
|
||||
* there (a duplicate merge, a deleted file), so the cached copy is
|
||||
* dropped. Kept, it stays stale forever and [FreshnessSweeper] asks for
|
||||
* it every pass (#5654). Any other failure leaves the row for a retry.
|
||||
*/
|
||||
private fun launchDeduped(
|
||||
inFlight: ConcurrentHashMap<String, Job>,
|
||||
id: String,
|
||||
onGone: suspend () -> Unit,
|
||||
block: suspend () -> Unit,
|
||||
): Job = inFlight.computeIfAbsent(id) {
|
||||
appScope.launch {
|
||||
try {
|
||||
fetchGate.withPermit { block() }
|
||||
} catch (e: HttpException) {
|
||||
if (e.code() == HTTP_NOT_FOUND) onGone()
|
||||
} catch (
|
||||
@Suppress("TooGenericExceptionCaught", "SwallowedException") e: Throwable,
|
||||
) {
|
||||
|
||||
+103
@@ -0,0 +1,103 @@
|
||||
package com.fabledsword.minstrel.cache.sync
|
||||
|
||||
import com.fabledsword.minstrel.api.endpoints.SyncApi
|
||||
import com.fabledsword.minstrel.auth.AuthStore
|
||||
import com.fabledsword.minstrel.cache.db.dao.CachedAlbumDao
|
||||
import com.fabledsword.minstrel.cache.db.dao.CachedArtistDao
|
||||
import com.fabledsword.minstrel.cache.db.dao.CachedTrackDao
|
||||
import com.fabledsword.minstrel.cache.db.dao.SyncMetadataDao
|
||||
import com.fabledsword.minstrel.cache.db.entities.SyncMetadataEntity
|
||||
import com.fabledsword.minstrel.connectivity.NetworkStatusController
|
||||
import com.fabledsword.minstrel.connectivity.ServerHealth
|
||||
import com.fabledsword.minstrel.models.wire.SyncDeletesWire
|
||||
import com.fabledsword.minstrel.models.wire.SyncResponseWire
|
||||
import io.mockk.coEvery
|
||||
import io.mockk.coVerify
|
||||
import io.mockk.every
|
||||
import io.mockk.mockk
|
||||
import kotlinx.coroutines.flow.MutableStateFlow
|
||||
import kotlinx.coroutines.test.runTest
|
||||
import okhttp3.ResponseBody.Companion.toResponseBody
|
||||
import org.junit.jupiter.api.Test
|
||||
import retrofit2.Response
|
||||
import retrofit2.Retrofit
|
||||
import kotlin.test.assertEquals
|
||||
|
||||
/**
|
||||
* The server returns at most 1000 changes a call (#5654). One page per
|
||||
* trigger left the cache behind the change log, so deletes from duplicate
|
||||
* merges never arrived; a pass pulls pages until the server answers 204.
|
||||
*/
|
||||
class SyncControllerTest {
|
||||
private val api: SyncApi = mockk()
|
||||
private val trackDao: CachedTrackDao = mockk(relaxed = true)
|
||||
private val retrofit: Retrofit = mockk {
|
||||
every { create(SyncApi::class.java) } returns api
|
||||
}
|
||||
private val authStore: AuthStore = mockk {
|
||||
every { sessionCookie } returns MutableStateFlow(null)
|
||||
}
|
||||
private val networkStatus: NetworkStatusController = mockk {
|
||||
every { state } returns MutableStateFlow(ServerHealth.Healthy)
|
||||
}
|
||||
|
||||
// The cursor round-trips through the DAO, so it is held for real.
|
||||
private var stored: SyncMetadataEntity? = null
|
||||
private val metadataDao: SyncMetadataDao = mockk {
|
||||
coEvery { get() } answers { stored }
|
||||
coEvery { upsert(any()) } answers { stored = firstArg() }
|
||||
}
|
||||
|
||||
private fun page(cursor: Long, deletedTracks: List<String> = emptyList()) =
|
||||
Response.success(SyncResponseWire(cursor = cursor, deletes = SyncDeletesWire(track = deletedTracks)))
|
||||
|
||||
private fun noContent(): Response<SyncResponseWire> = Response.success(204, null)
|
||||
|
||||
@Test
|
||||
fun `pages are pulled until the server has nothing newer`() = runTest {
|
||||
coEvery { api.sync(0) } returns page(1000)
|
||||
coEvery { api.sync(1000) } returns page(2000, deletedTracks = listOf("merged-away"))
|
||||
coEvery { api.sync(2000) } returns noContent()
|
||||
|
||||
SyncController(
|
||||
metadataDao, mockk<CachedArtistDao>(relaxed = true), mockk<CachedAlbumDao>(relaxed = true),
|
||||
trackDao, authStore, backgroundScope, networkStatus, retrofit,
|
||||
)
|
||||
testScheduler.advanceUntilIdle()
|
||||
|
||||
coVerify(exactly = 1) { api.sync(2000) }
|
||||
coVerify { trackDao.deleteByIds(listOf("merged-away")) }
|
||||
assertEquals(2000L, stored?.cursor)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a cursor that does not advance stops the pass`() = runTest {
|
||||
stored = SyncMetadataEntity(cursor = 500)
|
||||
coEvery { api.sync(500) } returns page(500)
|
||||
|
||||
SyncController(
|
||||
metadataDao, mockk<CachedArtistDao>(relaxed = true), mockk<CachedAlbumDao>(relaxed = true),
|
||||
trackDao, authStore, backgroundScope, networkStatus, retrofit,
|
||||
)
|
||||
testScheduler.advanceUntilIdle()
|
||||
|
||||
coVerify(exactly = 1) { api.sync(500) }
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a stale cursor resets to zero and keeps pulling`() = runTest {
|
||||
stored = SyncMetadataEntity(cursor = 7)
|
||||
coEvery { api.sync(7) } returns Response.error(410, "".toResponseBody())
|
||||
coEvery { api.sync(0) } returns page(1000)
|
||||
coEvery { api.sync(1000) } returns noContent()
|
||||
|
||||
SyncController(
|
||||
metadataDao, mockk<CachedArtistDao>(relaxed = true), mockk<CachedAlbumDao>(relaxed = true),
|
||||
trackDao, authStore, backgroundScope, networkStatus, retrofit,
|
||||
)
|
||||
testScheduler.advanceUntilIdle()
|
||||
|
||||
coVerify(exactly = 1) { api.sync(1000) }
|
||||
assertEquals(1000L, stored?.cursor)
|
||||
}
|
||||
}
|
||||
@@ -13,7 +13,10 @@ import kotlinx.coroutines.flow.flowOf
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.test.UnconfinedTestDispatcher
|
||||
import kotlinx.coroutines.test.runTest
|
||||
import okhttp3.ResponseBody.Companion.toResponseBody
|
||||
import org.junit.jupiter.api.Test
|
||||
import retrofit2.HttpException
|
||||
import retrofit2.Response
|
||||
import java.util.concurrent.atomic.AtomicInteger
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
@@ -36,6 +39,30 @@ class MetadataProviderTest {
|
||||
coVerify(exactly = 1) { library.refreshAlbumDetail("a1") }
|
||||
}
|
||||
|
||||
// A row the server no longer has (a duplicate merge) is dropped, or the
|
||||
// freshness sweep asks for it every 30 minutes forever (#5654).
|
||||
@Test
|
||||
fun `a 404 drops the cached row`() = runTest {
|
||||
coEvery { library.refreshTrack("t1") } throws httpError(404)
|
||||
val provider = MetadataProvider(albumDao, artistDao, trackDao, library, this)
|
||||
|
||||
provider.refreshTrack("t1").join()
|
||||
|
||||
coVerify(exactly = 1) { trackDao.deleteByIds(listOf("t1")) }
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `any other failure keeps the cached row`() = runTest {
|
||||
coEvery { library.refreshAlbumDetail("a1") } throws httpError(500)
|
||||
val provider = MetadataProvider(albumDao, artistDao, trackDao, library, this)
|
||||
|
||||
provider.refreshAlbum("a1").join()
|
||||
|
||||
coVerify(exactly = 0) { albumDao.deleteByIds(any()) }
|
||||
}
|
||||
|
||||
private fun httpError(status: Int) = HttpException(Response.error<Unit>(status, "".toResponseBody()))
|
||||
|
||||
@Test
|
||||
fun `at most four fetches run concurrently`() = runTest {
|
||||
val scope = this
|
||||
|
||||
Reference in New Issue
Block a user