fix: keep provider state during storage outages

This commit is contained in:
2026-08-24 02:17:56 +08:00
parent a0f55b041a
commit cf2c59a493
5 changed files with 28 additions and 85 deletions
@@ -4,7 +4,6 @@ import androidx.datastore.core.DataStore
import androidx.datastore.preferences.core.Preferences
import androidx.datastore.preferences.core.booleanPreferencesKey
import androidx.datastore.preferences.core.edit
import androidx.datastore.preferences.core.emptyPreferences
import androidx.datastore.preferences.core.stringPreferencesKey
import androidx.datastore.preferences.core.stringSetPreferencesKey
import com.flagship.abox.manager.domain.DomainResult
@@ -18,35 +17,26 @@ import java.io.IOException
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.distinctUntilChanged
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.flatMapLatest
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.onStart
import kotlinx.coroutines.flow.retryWhen
@OptIn(kotlinx.coroutines.ExperimentalCoroutinesApi::class)
class ProviderProfileStoreImpl(
private val dataStore: DataStore<Preferences>,
) : ProviderProfileStore {
private val refresh = MutableSharedFlow<Unit>(extraBufferCapacity = 1)
private val safeData: Flow<Preferences>
get() = refresh
.onStart { emit(Unit) }
.flatMapLatest { readDataOnce() }
get() = dataStore.data
.retryWhen { cause, attempt ->
if (cause !is IOException) return@retryWhen false
val delayMillis = (READ_RETRY_DELAY_MILLIS shl
attempt.coerceAtMost(MAX_BACKOFF_SHIFT).toInt())
.coerceAtMost(MAX_READ_RETRY_DELAY_MILLIS)
delay(delayMillis)
true
}
.distinctUntilChanged()
private fun readDataOnce(): Flow<Preferences> = dataStore.data
.retryWhen { cause, attempt ->
val retry = cause is IOException && attempt < MAX_READ_RETRIES
if (retry) delay(READ_RETRY_DELAY_MILLIS)
retry
}
.catchIOExceptionAsEmpty()
override fun observeProfiles(): Flow<List<ProviderProfile>> = safeData.map(::profilesFrom)
override fun observeActiveProfile(): Flow<ProviderProfile?> = safeData.map(::activeProfileFrom)
@@ -67,7 +57,6 @@ class ProviderProfileStoreImpl(
preferences[profileTypeKey(profile.id)] = profile.type.name
preferences[profileNameKey(profile.id)] = profile.displayName
}
refresh.tryEmit(Unit)
}
override suspend fun setActiveProfile(profileId: ProviderProfileId?): DomainResult<Unit> = operation {
@@ -75,7 +64,6 @@ class ProviderProfileStoreImpl(
if (profileId == null) preferences.remove(ACTIVE_PROFILE_ID)
else preferences[ACTIVE_PROFILE_ID] = profileId.value
}
refresh.tryEmit(Unit)
}
override suspend fun saveLoginPreferences(
@@ -89,7 +77,6 @@ class ProviderProfileStoreImpl(
if (username == null) values.remove(usernameKey(profileId))
else values[usernameKey(profileId)] = username
}
refresh.tryEmit(Unit)
}
private fun profilesFrom(preferences: Preferences): List<ProviderProfile> =
@@ -131,8 +118,9 @@ class ProviderProfileStoreImpl(
}
companion object {
private const val MAX_READ_RETRIES = 2L
private const val READ_RETRY_DELAY_MILLIS = 50L
private const val READ_RETRY_DELAY_MILLIS = 100L
private const val MAX_READ_RETRY_DELAY_MILLIS = 5_000L
private const val MAX_BACKOFF_SHIFT = 6L
private val PROFILE_IDS = stringSetPreferencesKey("profile_ids")
private val ACTIVE_PROFILE_ID = stringPreferencesKey("active_profile_id")
@@ -152,7 +140,3 @@ class ProviderProfileStoreImpl(
booleanPreferencesKey("profile_${id.value}_auto_login")
}
}
private fun Flow<Preferences>.catchIOExceptionAsEmpty(): Flow<Preferences> = catch { error ->
if (error is IOException) emit(emptyPreferences()) else throw error
}
@@ -3,58 +3,32 @@ package com.flagship.abox.manager.session
import androidx.datastore.core.DataStore
import androidx.datastore.preferences.core.Preferences
import androidx.datastore.preferences.core.emptyPreferences
import com.flagship.abox.manager.domain.ProviderProfile
import com.flagship.abox.manager.domain.ProviderProfileId
import com.flagship.abox.manager.domain.ProviderType
import com.flagship.abox.manager.session.profile.ProviderProfileStoreImpl
import java.io.IOException
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.take
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.launch
import kotlinx.coroutines.test.advanceTimeBy
import kotlinx.coroutines.test.advanceUntilIdle
import kotlinx.coroutines.test.runTest
import org.junit.Assert.assertEquals
import kotlinx.coroutines.withTimeout
import org.junit.Assert.assertFalse
import org.junit.Test
class ProviderProfileStoreFailureTest {
@Test
fun ioFailureFallsBackToEmptyPreferences() = runTest {
val store = ProviderProfileStoreImpl(FailingReadDataStore())
assertEquals(emptyList<Any>(), store.observeProfiles().first())
assertEquals(null, store.observeActiveProfile().first())
assertEquals(emptyList<Any>(), store.profiles())
assertEquals(null, store.activeProfile())
assertEquals(false, store.loginPreferences(ProviderProfileId("profile")).autoLoginEnabled)
}
@OptIn(kotlinx.coroutines.ExperimentalCoroutinesApi::class)
@Test
fun observerResubscribesAfterSuccessfulWrite() = runTest {
val store = ProviderProfileStoreImpl(RecoveringReadDataStore())
val emissions = mutableListOf<ProviderProfile?>()
val collector = backgroundScope.launch {
store.observeActiveProfile().take(2).toList(emissions)
fun persistentIoFailureDoesNotEmitFalseEmptyState() = runTest {
val store = ProviderProfileStoreImpl(FailingReadDataStore())
var completed = false
backgroundScope.launch {
runCatching { withTimeout(250) { store.activeProfile() } }
completed = true
}
advanceTimeBy(200)
advanceUntilIdle()
assertEquals(listOf<ProviderProfile?>(null), emissions)
val profile = ProviderProfile(
ProviderProfileId("recovered"),
ProviderType.LOCAL_EXPERIENCE,
"Recovered",
)
store.saveProfile(profile)
store.setActiveProfile(profile.id)
advanceUntilIdle()
collector.join()
assertEquals(listOf(null, profile), emissions)
assertFalse(completed)
}
private class FailingReadDataStore : DataStore<Preferences> {
@@ -66,22 +40,4 @@ class ProviderProfileStoreFailureTest {
transform: suspend (t: Preferences) -> Preferences,
): Preferences = transform(emptyPreferences())
}
private class RecoveringReadDataStore : DataStore<Preferences> {
private var attempts = 0
private var current: Preferences = emptyPreferences()
override val data: Flow<Preferences>
get() = flow {
if (attempts++ < 3) throw IOException("temporarily unavailable")
emit(current)
}
override suspend fun updateData(
transform: suspend (t: Preferences) -> Preferences,
): Preferences {
current = transform(current)
return current
}
}
}