Creepso/NuvioStreaming
1
0
Fork 0
mirror of https://github.com/tapframe/NuvioStreaming.git synced 2026-08-17 12:57:03 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix(sync): retry incomplete foreground pulls

Browse source
This commit is contained in:
tapframe 2026-07-22 21:31:08 +05:30
parent 7210e7cc68
commit 002a54e022
2 changed files with 88 additions and 47 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

102
composeApp/src/commonMain/kotlin/com/nuvio/app/core/sync/SyncManager.kt
View file

@ -62,6 +62,28 @@ internal data class ProfileSyncResult(
get() = failedSteps.isEmpty()
}
internal data class ProfilePullFreshness(
val profileId: Int? = null,
val completedAtEpochMs: Long = 0L,
) {
fun isRecent(profileId: Int, nowEpochMs: Long, minIntervalMs: Long): Boolean =
this.profileId == profileId && nowEpochMs - completedAtEpochMs < minIntervalMs
fun recordIfSuccessful(
profileId: Int,
completedAtEpochMs: Long,
result: ProfileSyncResult,
): ProfilePullFreshness =
if (result.succeeded) {
ProfilePullFreshness(
profileId = profileId,
completedAtEpochMs = completedAtEpochMs,
)
} else {
this
}
}
internal suspend fun runOrderedProfileSync(
profileId: Int,
pluginsEnabled: Boolean,
@ -211,8 +233,7 @@ object SyncManager {
private var foregroundPullProfileId: Int? = null
private var periodicNuvioSyncPullJob: Job? = null
private var periodicNuvioSyncProfileId: Int? = null
private var lastFullPullAtMs: Long = 0L
private var lastFullPullProfileId: Int? = null
private var pullFreshness = ProfilePullFreshness()
private val profileSyncOperations = ProfileSyncOperations(
pullAddons = { profileId -> AddonRepository.pullFromServer(profileId) },
@ -247,8 +268,7 @@ object SyncManager {
foregroundPullJob.also {
foregroundPullJob = null
foregroundPullProfileId = null
lastFullPullAtMs = 0L
lastFullPullProfileId = null
pullFreshness = ProfilePullFreshness()
}
}
foregroundJob?.cancel()
@ -304,8 +324,11 @@ object SyncManager {
private fun hasRecentFullPull(profileId: Int): Boolean =
synchronized(pullStateLock) {
lastFullPullProfileId == profileId &&
EpisodeReleaseDatePlatform.nowEpochMs() - lastFullPullAtMs < FOREGROUND_PULL_MIN_INTERVAL_MS
pullFreshness.isRecent(
profileId = profileId,
nowEpochMs = EpisodeReleaseDatePlatform.nowEpochMs(),
minIntervalMs = FOREGROUND_PULL_MIN_INTERVAL_MS,
)
}
private suspend fun pullForegroundForProfile(profileId: Int) {
@ -313,42 +336,25 @@ object SyncManager {
runCatching { ProfileRepository.pullProfiles() }
.onFailure { log.e(it) { "Foreground profiles pull failed" } }
runCatching { ProfileSettingsSync.pull(profileId) }
.onFailure { log.e(it) { "Foreground profile settings pull failed" } }
coroutineScope {
launch {
runCatching { AddonRepository.pullFromServer(profileId) }
.onFailure { log.e(it) { "Foreground addons pull failed" } }
}
if (AppFeaturePolicy.pluginsEnabled) {
launch {
runCatching { PluginRepository.pullFromServer(profileId) }
.onFailure { log.e(it) { "Foreground plugins pull failed" } }
}
}
launch {
runCatching { LibraryRepository.pullFromServer(profileId) }
.onFailure { log.e(it) { "Foreground library pull failed" } }
}
launch {
runCatching {
WatchProgressSourceCoordinator.refreshActiveSource(profileId = profileId, force = true)
}.onFailure { log.e(it) { "Foreground active watch source pull failed" } }
}
launch {
runCatching { CollectionSyncService.pullFromServer(profileId) }
.onFailure { log.e(it) { "Foreground collections pull failed" } }
}
launch {
runCatching { HomeCatalogSettingsSyncService.pullFromServer(profileId) }
.onFailure { log.e(it) { "Foreground home catalog settings pull failed" } }
}
}
val syncResult = runOrderedProfileSync(
profileId = profileId,
pluginsEnabled = AppFeaturePolicy.pluginsEnabled,
operations = profileSyncOperations,
onFailure = { step, error ->
log.e(error) { "Foreground profile sync step failed profile=$profileId step=$step" }
},
)
synchronized(pullStateLock) {
lastFullPullAtMs = EpisodeReleaseDatePlatform.nowEpochMs()
lastFullPullProfileId = profileId
pullFreshness = pullFreshness.recordIfSuccessful(
profileId = profileId,
completedAtEpochMs = EpisodeReleaseDatePlatform.nowEpochMs(),
result = syncResult,
)
}
if (!syncResult.succeeded) {
log.w {
"Foreground profile sync incomplete profile=$profileId failedSteps=${syncResult.failedSteps}"
}
}
log.i { "Foreground sync completed profile=$profileId" }
@ -386,12 +392,14 @@ object SyncManager {
} finally {
WatchProgressSourceCoordinator.resumeAutomaticTransitions()
}
if (syncResult.succeeded) {
synchronized(pullStateLock) {
lastFullPullAtMs = EpisodeReleaseDatePlatform.nowEpochMs()
lastFullPullProfileId = profileId
}
} else {
synchronized(pullStateLock) {
pullFreshness = pullFreshness.recordIfSuccessful(
profileId = profileId,
completedAtEpochMs = EpisodeReleaseDatePlatform.nowEpochMs(),
result = syncResult,
)
}
if (!syncResult.succeeded) {
log.w {
"Full profile sync incomplete profile=$profileId reason=$reason " +
"failedSteps=${syncResult.failedSteps}"

33
composeApp/src/commonTest/kotlin/com/nuvio/app/core/sync/SyncManagerTest.kt
View file

@ -131,6 +131,39 @@ class SyncManagerTest {
assertEquals(setOf(ProfileSyncStep.ActiveWatchSource), result.failedSteps)
}
@Test
fun `failed profile sync does not advance foreground freshness`() {
val previous = ProfilePullFreshness(
profileId = 3,
completedAtEpochMs = 1_000L,
)
val failed = previous.recordIfSuccessful(
profileId = 3,
completedAtEpochMs = 2_000L,
result = ProfileSyncResult(setOf(ProfileSyncStep.ActiveWatchSource)),
)
val succeeded = previous.recordIfSuccessful(
profileId = 3,
completedAtEpochMs = 2_000L,
result = ProfileSyncResult(emptySet()),
)
assertEquals(previous, failed)
assertEquals(2_000L, succeeded.completedAtEpochMs)
assertFalse(
ProfilePullFreshness()
.recordIfSuccessful(
profileId = 3,
completedAtEpochMs = 2_000L,
result = ProfileSyncResult(setOf(ProfileSyncStep.ActiveWatchSource)),
)
.isRecent(profileId = 3, nowEpochMs = 2_001L, minIntervalMs = 1_000L),
)
assertTrue(
succeeded.isRecent(profileId = 3, nowEpochMs = 2_001L, minIntervalMs = 1_000L),
)
}
@Test
fun `realtime invalidation queued during active sync runs once afterwards`() = runBlocking {
val gate = ProfileSyncRequestGate()