diff --git a/android/app/src/main/java/com/fabledsword/minstrel/cache/sync/SyncController.kt b/android/app/src/main/java/com/fabledsword/minstrel/cache/sync/SyncController.kt index ba290b9d..5ae67c21 100644 --- a/android/app/src/main/java/com/fabledsword/minstrel/cache/sync/SyncController.kt +++ b/android/app/src/main/java/com/fabledsword/minstrel/cache/sync/SyncController.kt @@ -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 } } diff --git a/android/app/src/main/java/com/fabledsword/minstrel/metadata/MetadataProvider.kt b/android/app/src/main/java/com/fabledsword/minstrel/metadata/MetadataProvider.kt index 24290d88..4e6c06e4 100644 --- a/android/app/src/main/java/com/fabledsword/minstrel/metadata/MetadataProvider.kt +++ b/android/app/src/main/java/com/fabledsword/minstrel/metadata/MetadataProvider.kt @@ -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, 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, ) { diff --git a/android/app/src/test/java/com/fabledsword/minstrel/cache/sync/SyncControllerTest.kt b/android/app/src/test/java/com/fabledsword/minstrel/cache/sync/SyncControllerTest.kt new file mode 100644 index 00000000..fd9125a0 --- /dev/null +++ b/android/app/src/test/java/com/fabledsword/minstrel/cache/sync/SyncControllerTest.kt @@ -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 = emptyList()) = + Response.success(SyncResponseWire(cursor = cursor, deletes = SyncDeletesWire(track = deletedTracks))) + + private fun noContent(): Response = 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(relaxed = true), mockk(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(relaxed = true), mockk(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(relaxed = true), mockk(relaxed = true), + trackDao, authStore, backgroundScope, networkStatus, retrofit, + ) + testScheduler.advanceUntilIdle() + + coVerify(exactly = 1) { api.sync(1000) } + assertEquals(1000L, stored?.cursor) + } +} diff --git a/android/app/src/test/java/com/fabledsword/minstrel/metadata/MetadataProviderTest.kt b/android/app/src/test/java/com/fabledsword/minstrel/metadata/MetadataProviderTest.kt index e9eddc06..4d0e8cb9 100644 --- a/android/app/src/test/java/com/fabledsword/minstrel/metadata/MetadataProviderTest.kt +++ b/android/app/src/test/java/com/fabledsword/minstrel/metadata/MetadataProviderTest.kt @@ -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(status, "".toResponseBody())) + @Test fun `at most four fetches run concurrently`() = runTest { val scope = this