From e0cbf447dc70216ead6eaee1ceda43fce1642e3d Mon Sep 17 00:00:00 2001 From: tapframe <85391825+tapframe@users.noreply.github.com> Date: Sun, 2 Aug 2026 16:23:48 +0530 Subject: [PATCH] fix: contain Trakt watched rate limits (NUVIO-MOBILE-Q1) --- .../app/features/watched/WatchedRepository.kt | 31 ++++++++--- .../watching/sync/TraktWatchedSyncAdapter.kt | 54 ++++++++++++++----- .../features/watched/WatchedRepositoryTest.kt | 29 ++++++++++ .../sync/TraktWatchedSyncAdapterTest.kt | 53 ++++++++++++++++++ 4 files changed, 148 insertions(+), 19 deletions(-) diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watched/WatchedRepository.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watched/WatchedRepository.kt index d65129a5b..e3f305c6b 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watched/WatchedRepository.kt +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watched/WatchedRepository.kt @@ -101,6 +101,18 @@ internal fun replaceWatchedItemsForSource( target.putAll(replacement) } +internal suspend fun watchedProviderRefreshOrNull( + refresh: suspend () -> T, + onFailure: (Throwable) -> Unit, +): T? = try { + refresh() +} catch (error: CancellationException) { + throw error +} catch (error: Throwable) { + onFailure(error) + null +} + object WatchedRepository { private data class WatchedRefreshOperation( val profileId: Int, @@ -1054,16 +1066,21 @@ object WatchedRepository { .collectLatest { extraKeys -> val keysChanged = providerExtraWatchedKeys[providerId] != extraKeys if (keysChanged) { - providerExtraWatchedKeys[providerId] = extraKeys - // Re-pull items from provider to reflect snapshot changes - // (e.g. after remote episode removal) - val freshItems = adapter.pull( - profileId = currentProfileId, - pageSize = watchedItemsPageSize, - ) + val freshItems = watchedProviderRefreshOrNull( + refresh = { + adapter.pull( + profileId = currentProfileId, + pageSize = watchedItemsPageSize, + ) + }, + onFailure = { error -> + log.w(error) { "Failed to refresh watched items from ${providerId.storageId}" } + }, + ) ?: return@collectLatest val itemsByKey = freshItems.associateBy { item -> watchedItemKey(item.type, item.id, item.season, item.episode) }.toMutableMap() + providerExtraWatchedKeys[providerId] = extraKeys providerItemsByKey[providerId] = itemsByKey publish() } diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watching/sync/TraktWatchedSyncAdapter.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watching/sync/TraktWatchedSyncAdapter.kt index 76f5bd0fa..9534f31ea 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watching/sync/TraktWatchedSyncAdapter.kt +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watching/sync/TraktWatchedSyncAdapter.kt @@ -14,6 +14,7 @@ import com.nuvio.app.features.watched.normalizeWatchedMarkedAtEpochMs import kotlinx.coroutines.CancellationException import kotlinx.coroutines.async import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.delay import kotlinx.serialization.SerialName import kotlinx.serialization.Serializable import kotlinx.serialization.encodeToString @@ -24,6 +25,8 @@ private const val WATCHED_PAGE_LIMIT = 250 private const val WATCHED_MAX_PAGES = 1_000 private const val WATCHED_SHOWS_EXTENDED = "progress" internal const val TRAKT_WATCHED_MAX_RESPONSE_BODY_BYTES = 8 * 1024 * 1024 +private const val TRAKT_WATCHED_MAX_ATTEMPTS = 2 +private const val TRAKT_WATCHED_MAX_RETRY_DELAY_MS = 60_000L internal fun interface TraktWatchedHttpEngine { suspend fun get( @@ -35,15 +38,48 @@ internal fun interface TraktWatchedHttpEngine { internal class TraktWatchedPageClient( private val engine: TraktWatchedHttpEngine, + private val sleep: suspend (Long) -> Unit = { delayMs -> delay(delayMs) }, ) { - suspend fun get(url: String, headers: Map): RawHttpResponse = - engine.get( - url = url, - headers = headers, - maxResponseBodyBytes = TRAKT_WATCHED_MAX_RESPONSE_BODY_BYTES, - ) + suspend fun get(url: String, headers: Map): RawHttpResponse { + repeat(TRAKT_WATCHED_MAX_ATTEMPTS) { attempt -> + val response = engine.get( + url = url, + headers = headers, + maxResponseBodyBytes = TRAKT_WATCHED_MAX_RESPONSE_BODY_BYTES, + ) + if (response.status in 200..299) return response + if (!isTransientTraktWatchedStatus(response.status) || attempt == TRAKT_WATCHED_MAX_ATTEMPTS - 1) { + throw TraktWatchedHttpException(response.status) + } + sleep( + traktWatchedRetryDelayMs( + attempt = attempt, + retryAfterSeconds = response.headers.entries + .firstOrNull { (name, _) -> name.equals("retry-after", ignoreCase = true) } + ?.value, + ), + ) + } + error("Trakt watched request exhausted without a response") + } } +internal class TraktWatchedHttpException( + val status: Int, +) : Exception("Trakt watched request failed: $status") + +internal fun isTransientTraktWatchedStatus(status: Int): Boolean = + status == 429 || status in 500..599 + +internal fun traktWatchedRetryDelayMs(attempt: Int, retryAfterSeconds: String?): Long = + retryAfterSeconds + ?.trim() + ?.toLongOrNull() + ?.coerceAtLeast(0L) + ?.times(1_000L) + ?.coerceAtMost(TRAKT_WATCHED_MAX_RETRY_DELAY_MS) + ?: (1_000L shl attempt.coerceIn(0, 5)).coerceAtMost(TRAKT_WATCHED_MAX_RETRY_DELAY_MS) + private val platformTraktWatchedHttpEngine = TraktWatchedHttpEngine { url, headers, maxResponseBodyBytes -> httpRequestRaw( method = "GET", @@ -153,9 +189,6 @@ object TraktWatchedSyncAdapter : TrackingWatchedProvider { url = "$BASE_URL/sync/watched/movies?page=$page&limit=$WATCHED_PAGE_LIMIT", headers = headers, ) - if (response.status !in 200..299) { - error("Trakt watched movies request failed: ${response.status}") - } val pageItems = json.decodeFromString>(response.body) if (pageItems.isEmpty()) break items.addAll(pageItems) @@ -177,9 +210,6 @@ object TraktWatchedSyncAdapter : TrackingWatchedProvider { url = "$BASE_URL/sync/watched/shows?page=$page&limit=$WATCHED_PAGE_LIMIT&extended=$WATCHED_SHOWS_EXTENDED", headers = headers, ) - if (response.status !in 200..299) { - error("Trakt watched shows request failed: ${response.status}") - } val pageItems = json.decodeFromString>(response.body) if (pageItems.isEmpty()) break items.addAll(pageItems) diff --git a/composeApp/src/commonTest/kotlin/com/nuvio/app/features/watched/WatchedRepositoryTest.kt b/composeApp/src/commonTest/kotlin/com/nuvio/app/features/watched/WatchedRepositoryTest.kt index 38f65f330..86f320b00 100644 --- a/composeApp/src/commonTest/kotlin/com/nuvio/app/features/watched/WatchedRepositoryTest.kt +++ b/composeApp/src/commonTest/kotlin/com/nuvio/app/features/watched/WatchedRepositoryTest.kt @@ -4,12 +4,41 @@ import com.nuvio.app.features.details.MetaDetails import com.nuvio.app.features.details.MetaVideo import com.nuvio.app.features.tracking.TrackingProviderId import com.nuvio.app.features.tracking.WatchProgressSource +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.runBlocking import kotlin.test.Test import kotlin.test.assertEquals +import kotlin.test.assertFailsWith import kotlin.test.assertFalse +import kotlin.test.assertNull import kotlin.test.assertTrue class WatchedRepositoryTest { + @Test + fun providerRefreshFailure_isContainedWithoutReplacingState() = runBlocking { + val failure = IllegalStateException("rate limited") + var observedFailure: Throwable? = null + + val result = watchedProviderRefreshOrNull( + refresh = { throw failure }, + onFailure = { observedFailure = it }, + ) + + assertNull(result) + assertEquals(failure, observedFailure) + } + + @Test + fun providerRefreshCancellation_isNotContained() = runBlocking { + assertFailsWith { + watchedProviderRefreshOrNull( + refresh = { throw CancellationException("cancelled") }, + onFailure = {}, + ) + } + Unit + } + @Test fun watchedItemKey_isTypeAware() { assertEquals("movie:tt1:-1:-1", watchedItemKey(type = "movie", id = "tt1")) diff --git a/composeApp/src/commonTest/kotlin/com/nuvio/app/features/watching/sync/TraktWatchedSyncAdapterTest.kt b/composeApp/src/commonTest/kotlin/com/nuvio/app/features/watching/sync/TraktWatchedSyncAdapterTest.kt index 5f6021298..d63af85aa 100644 --- a/composeApp/src/commonTest/kotlin/com/nuvio/app/features/watching/sync/TraktWatchedSyncAdapterTest.kt +++ b/composeApp/src/commonTest/kotlin/com/nuvio/app/features/watching/sync/TraktWatchedSyncAdapterTest.kt @@ -5,6 +5,7 @@ import com.nuvio.app.features.addons.RawHttpResponse import kotlinx.coroutines.runBlocking import kotlin.test.Test import kotlin.test.assertEquals +import kotlin.test.assertFailsWith import kotlin.test.assertFalse class TraktWatchedSyncAdapterTest { @@ -35,4 +36,56 @@ class TraktWatchedSyncAdapterTest { assertFalse(response.body.length < body.length) assertEquals(body, response.body) } + + @Test + fun `rate limited watched request retries once after retry after`() = runBlocking { + var attempts = 0 + val delays = mutableListOf() + val client = TraktWatchedPageClient( + engine = TraktWatchedHttpEngine { _, _, _ -> + attempts += 1 + response( + status = if (attempts == 1) 429 else 200, + headers = if (attempts == 1) mapOf("Retry-After" to "2") else emptyMap(), + ) + }, + sleep = delays::add, + ) + + val result = client.get("https://api.trakt.tv/sync/watched/shows", emptyMap()) + + assertEquals(200, result.status) + assertEquals(2, attempts) + assertEquals(listOf(2_000L), delays) + } + + @Test + fun `repeated rate limit becomes a bounded typed failure`() = runBlocking { + var attempts = 0 + val client = TraktWatchedPageClient( + engine = TraktWatchedHttpEngine { _, _, _ -> + attempts += 1 + response(status = 429) + }, + sleep = {}, + ) + + val error = assertFailsWith { + client.get("https://api.trakt.tv/sync/watched/shows", emptyMap()) + } + + assertEquals(429, error.status) + assertEquals(2, attempts) + } + + private fun response( + status: Int, + headers: Map = emptyMap(), + ): RawHttpResponse = RawHttpResponse( + status = status, + statusText = "", + url = "https://api.trakt.tv/sync/watched/shows", + body = "[]", + headers = headers, + ) }