diff --git a/appnav/src/main/kotlin/io/element/android/appnav/session/SyncOrchestrator.kt b/appnav/src/main/kotlin/io/element/android/appnav/session/SyncOrchestrator.kt index 3a72c10ab7..786365b7be 100644 --- a/appnav/src/main/kotlin/io/element/android/appnav/session/SyncOrchestrator.kt +++ b/appnav/src/main/kotlin/io/element/android/appnav/session/SyncOrchestrator.kt @@ -28,10 +28,14 @@ import kotlinx.coroutines.flow.combine import kotlinx.coroutines.flow.debounce import kotlinx.coroutines.flow.distinctUntilChanged import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.launchIn import kotlinx.coroutines.flow.onCompletion +import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.launch import timber.log.Timber import java.util.concurrent.atomic.AtomicBoolean +import java.util.concurrent.atomic.AtomicInteger +import kotlin.time.Duration import kotlin.time.Duration.Companion.milliseconds import kotlin.time.Duration.Companion.seconds @@ -60,6 +64,12 @@ class SyncOrchestrator( private val started = AtomicBoolean(false) + /** + * Number of consecutive attempts at restarting a sync service which stopped on its own. + * Reset once the sync service has been running for [RESTART_BACKOFF_RESET_DELAY]. + */ + private val restartAttempt = AtomicInteger(0) + /** * Starting observing the app state and network state to start/stop the sync service. * @@ -91,6 +101,17 @@ class SyncOrchestrator( @VisibleForTesting(otherwise = VisibleForTesting.PRIVATE) internal fun observeStates() = coroutineScope.launch { Timber.tag(tag).d("start observing the app and network state") + // Reset the restart backoff once the sync service has been steadily running for a while. Note that + // SyncService.startSync() moves the state to Running right away, so resetting on any Running value + // would defeat the backoff when the sync service dies again immediately after each restart. + syncService.syncState + .debounce(RESTART_BACKOFF_RESET_DELAY) + .onEach { syncState -> + if (syncState == SyncState.Running) { + restartAttempt.set(0) + } + } + .launchIn(this) val isAppActiveFlows = listOf( appForegroundStateService.isInForeground, appForegroundStateService.isInCall, @@ -117,14 +138,26 @@ class SyncOrchestrator( SyncStateAction.StopSync } else if (syncState == SyncState.Idle && isAppActive && isNetworkAvailable) { SyncStateAction.StartSync + } else if (syncState.isDead() && isAppActive && isNetworkAvailable) { + // The sync service has given up, for instance because the homeserver has expired our sliding sync + // session and answered 400 M_UNKNOWN_POS. Nothing else will ever restart it, so do it from here, + // else the application is stuck until its data is cleared. + SyncStateAction.RestartSync } else { SyncStateAction.NoOp } } .distinctUntilChanged() .debounce { action -> - // Don't stop the sync immediately, wait a bit to avoid starting/stopping the sync too often - if (action == SyncStateAction.StopSync) 3.seconds else 0.seconds + when (action) { + // Don't stop the sync immediately, wait a bit to avoid starting/stopping the sync too often + SyncStateAction.StopSync -> 3.seconds + // Back off before restarting a dead sync service, to avoid hammering a failing homeserver. + // A state change in the meantime cancels the pending restart. + SyncStateAction.RestartSync -> restartDelay(restartAttempt.get()) + SyncStateAction.StartSync, + SyncStateAction.NoOp -> 0.seconds + } } .onCompletion { Timber.tag(tag).d("has been stopped") @@ -134,6 +167,11 @@ class SyncOrchestrator( SyncStateAction.StartSync -> { syncService.startSync() } + SyncStateAction.RestartSync -> { + val attempt = restartAttempt.incrementAndGet() + Timber.tag(tag).w("sync service is dead, restarting it (attempt $attempt)") + syncService.startSync() + } SyncStateAction.StopSync -> { syncService.stopSync() } @@ -143,8 +181,34 @@ class SyncOrchestrator( } } +@VisibleForTesting +internal val FIRST_RESTART_DELAY = 1.seconds + +@VisibleForTesting +internal val MAX_RESTART_DELAY = 30.seconds + +/** + * Duration during which the sync service has to stay running before the restart backoff is reset. + */ +@VisibleForTesting +internal val RESTART_BACKOFF_RESET_DELAY = 30.seconds + +/** + * Return true if the sync service has stopped and will not restart by itself. + */ +private fun SyncState.isDead() = this == SyncState.Error || this == SyncState.Terminated + +/** + * Exponential backoff, from [FIRST_RESTART_DELAY], doubling on every attempt, up to [MAX_RESTART_DELAY]. + */ +private fun restartDelay(attempt: Int): Duration { + val exponent = attempt.coerceIn(0, 5) + return (FIRST_RESTART_DELAY * (1 shl exponent)).coerceAtMost(MAX_RESTART_DELAY) +} + private enum class SyncStateAction { StartSync, + RestartSync, StopSync, NoOp, } diff --git a/appnav/src/test/kotlin/io/element/android/appnav/SyncOrchestratorTest.kt b/appnav/src/test/kotlin/io/element/android/appnav/SyncOrchestratorTest.kt index 32d0ec8f56..9475399a85 100644 --- a/appnav/src/test/kotlin/io/element/android/appnav/SyncOrchestratorTest.kt +++ b/appnav/src/test/kotlin/io/element/android/appnav/SyncOrchestratorTest.kt @@ -8,6 +8,9 @@ package io.element.android.appnav +import io.element.android.appnav.session.FIRST_RESTART_DELAY +import io.element.android.appnav.session.MAX_RESTART_DELAY +import io.element.android.appnav.session.RESTART_BACKOFF_RESET_DELAY import io.element.android.appnav.session.SyncOrchestrator import io.element.android.features.networkmonitor.api.NetworkStatus import io.element.android.features.networkmonitor.test.FakeNetworkMonitor @@ -414,6 +417,150 @@ class SyncOrchestratorTest { startSyncRecorder.assertions().isNeverCalled() } + @Test + fun `when the sync service is in error, it will be restarted after a delay`() = runTest { + val startSyncRecorder = lambdaRecorder> { Result.success(Unit) } + val syncService = FakeSyncService(initialSyncState = SyncState.Error).apply { + startSyncLambda = startSyncRecorder + } + val syncOrchestrator = createSyncOrchestrator( + matrixClient = FakeMatrixClient(syncService = syncService), + networkMonitor = FakeNetworkMonitor(initialStatus = NetworkStatus.Connected), + appForegroundStateService = FakeAppForegroundStateService(initialForegroundValue = true), + ) + + // We start observing + syncOrchestrator.observeStates() + + // The restart does not happen right away, we back off first + advanceTimeBy(FIRST_RESTART_DELAY / 2) + startSyncRecorder.assertions().isNeverCalled() + + // But it does happen once the backoff has elapsed + advanceTimeBy(FIRST_RESTART_DELAY) + startSyncRecorder.assertions().isCalledOnce() + } + + @Test + fun `when the sync service is terminated, it will be restarted after a delay`() = runTest { + val startSyncRecorder = lambdaRecorder> { Result.success(Unit) } + val syncService = FakeSyncService(initialSyncState = SyncState.Terminated).apply { + startSyncLambda = startSyncRecorder + } + val syncOrchestrator = createSyncOrchestrator( + matrixClient = FakeMatrixClient(syncService = syncService), + networkMonitor = FakeNetworkMonitor(initialStatus = NetworkStatus.Connected), + appForegroundStateService = FakeAppForegroundStateService(initialForegroundValue = true), + ) + + // We start observing + syncOrchestrator.observeStates() + + advanceTimeBy(FIRST_RESTART_DELAY * 2) + startSyncRecorder.assertions().isCalledOnce() + } + + @Test + fun `when the sync service keeps dying, the delay between the restarts increases`() = runTest { + val startSyncRecorder = lambdaRecorder> { Result.success(Unit) } + val syncService = FakeSyncService(initialSyncState = SyncState.Error).apply { + startSyncLambda = startSyncRecorder + } + val syncOrchestrator = createSyncOrchestrator( + matrixClient = FakeMatrixClient(syncService = syncService), + networkMonitor = FakeNetworkMonitor(initialStatus = NetworkStatus.Connected), + appForegroundStateService = FakeAppForegroundStateService(initialForegroundValue = true), + ) + + // We start observing + syncOrchestrator.observeStates() + + advanceTimeBy(FIRST_RESTART_DELAY * 2) + startSyncRecorder.assertions().isCalledOnce() + + // The sync service starts, but dies again straight away + syncService.emitSyncState(SyncState.Running) + advanceTimeBy(200.milliseconds) + syncService.emitSyncState(SyncState.Error) + + // The second restart is not attempted after the first delay anymore + advanceTimeBy(FIRST_RESTART_DELAY + 200.milliseconds) + startSyncRecorder.assertions().isCalledOnce() + + // But after twice that delay + advanceTimeBy(FIRST_RESTART_DELAY) + startSyncRecorder.assertions().isCalledExactly(2) + } + + @Test + fun `when the sync service recovers, the delay between the restarts is reset`() = runTest { + val startSyncRecorder = lambdaRecorder> { Result.success(Unit) } + val syncService = FakeSyncService(initialSyncState = SyncState.Error).apply { + startSyncLambda = startSyncRecorder + } + val syncOrchestrator = createSyncOrchestrator( + matrixClient = FakeMatrixClient(syncService = syncService), + networkMonitor = FakeNetworkMonitor(initialStatus = NetworkStatus.Connected), + appForegroundStateService = FakeAppForegroundStateService(initialForegroundValue = true), + ) + + // We start observing + syncOrchestrator.observeStates() + + advanceTimeBy(FIRST_RESTART_DELAY * 2) + startSyncRecorder.assertions().isCalledOnce() + + // The sync service is now running, and keeps running for a while + syncService.emitSyncState(SyncState.Running) + advanceTimeBy(RESTART_BACKOFF_RESET_DELAY + 1.seconds) + startSyncRecorder.assertions().isCalledOnce() + + // It dies again, and is restarted after the first delay, not a longer one + syncService.emitSyncState(SyncState.Error) + advanceTimeBy(FIRST_RESTART_DELAY / 2) + startSyncRecorder.assertions().isCalledOnce() + advanceTimeBy(FIRST_RESTART_DELAY) + startSyncRecorder.assertions().isCalledExactly(2) + } + + @Test + fun `when the sync service is in error but the app is in background, it is not restarted`() = runTest { + val startSyncRecorder = lambdaRecorder> { Result.success(Unit) } + val syncService = FakeSyncService(initialSyncState = SyncState.Error).apply { + startSyncLambda = startSyncRecorder + } + val syncOrchestrator = createSyncOrchestrator( + matrixClient = FakeMatrixClient(syncService = syncService), + networkMonitor = FakeNetworkMonitor(initialStatus = NetworkStatus.Connected), + appForegroundStateService = FakeAppForegroundStateService(initialForegroundValue = false), + ) + + // We start observing + syncOrchestrator.observeStates() + + advanceTimeBy(MAX_RESTART_DELAY * 2) + startSyncRecorder.assertions().isNeverCalled() + } + + @Test + fun `when the sync service is in error but there is no network, it is not restarted`() = runTest { + val startSyncRecorder = lambdaRecorder> { Result.success(Unit) } + val syncService = FakeSyncService(initialSyncState = SyncState.Error).apply { + startSyncLambda = startSyncRecorder + } + val syncOrchestrator = createSyncOrchestrator( + matrixClient = FakeMatrixClient(syncService = syncService), + networkMonitor = FakeNetworkMonitor(initialStatus = NetworkStatus.Disconnected), + appForegroundStateService = FakeAppForegroundStateService(initialForegroundValue = true), + ) + + // We start observing + syncOrchestrator.observeStates() + + advanceTimeBy(MAX_RESTART_DELAY * 2) + startSyncRecorder.assertions().isNeverCalled() + } + private fun TestScope.createSyncOrchestrator( matrixClient: FakeMatrixClient = FakeMatrixClient(), networkMonitor: FakeNetworkMonitor = FakeNetworkMonitor(),