diff --git a/app/src/main/java/to/bitkit/repositories/BackupRepo.kt b/app/src/main/java/to/bitkit/repositories/BackupRepo.kt index ad7339fc70..b06ed2e9a4 100644 --- a/app/src/main/java/to/bitkit/repositories/BackupRepo.kt +++ b/app/src/main/java/to/bitkit/repositories/BackupRepo.kt @@ -4,14 +4,20 @@ import android.content.Context import com.synonym.bitkitcore.UsdtException import com.synonym.vssclient.VssException import dagger.hilt.android.qualifiers.ApplicationContext +import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.CoroutineDispatcher +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.CoroutineStart import kotlinx.coroutines.FlowPreview import kotlinx.coroutines.Job import kotlinx.coroutines.NonCancellable +import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.async +import kotlinx.coroutines.cancelAndJoin import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.delay +import kotlinx.coroutines.ensureActive import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow @@ -23,6 +29,7 @@ import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.map import kotlinx.coroutines.flow.update import kotlinx.coroutines.isActive +import kotlinx.coroutines.joinAll import kotlinx.coroutines.launch import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock @@ -79,10 +86,12 @@ import to.bitkit.utils.AppError import to.bitkit.utils.Logger import to.bitkit.utils.NetworkValidationHelper import to.bitkit.utils.jsonLogOf +import java.net.URI import java.util.concurrent.ConcurrentHashMap import javax.inject.Inject import javax.inject.Provider import javax.inject.Singleton +import kotlin.coroutines.cancellation.CancellationException import kotlin.time.Clock import kotlin.time.Duration.Companion.minutes import kotlin.time.Duration.Companion.seconds @@ -144,19 +153,33 @@ class BackupRepo @Inject constructor( private val usdtRecoveryBackup = UsdtRecoveryBackupStore(vssBackupClient) private val scope = appScope(ioDispatcher, TAG) + private val lifecycleLock = Any() + private val resetMutex = Mutex() + private var generation = 0L + private var resetting = false + + // Reset owns all pending work; stopping observation only cancels its child scope. + // Retiring observation parents remain attached until their cancellation cleanup finishes. + private var pendingScope = newBackupScope(scope) + private var observation: Observation? = null + private val retiringObservations = mutableSetOf() private val backupJobs = mutableMapOf() - private val statusObserverJobs = mutableListOf() - private val dataListenerJobs = mutableListOf() - private var periodicCheckJob: Job? = null + private val explicitBackups = mutableSetOf() + private val categoryMutexes = BackupCategory.entries.associateWith { Mutex() } private val runningBackups = ConcurrentHashMap.newKeySet() // Tracks active jobs since app start private val failedBackupRequired = ConcurrentHashMap() private val temporaryFailureRetries = ConcurrentHashMap() private val temporaryFailureSince = ConcurrentHashMap() - - private var isObserving = false private var lastNotificationTime = 0L + private class Observation(val scope: CoroutineScope, val generation: Long) { + val ready = CompletableDeferred() + } + + private fun newBackupScope(parent: CoroutineScope) = + CoroutineScope(parent.coroutineContext + SupervisorJob(parent.coroutineContext[Job])) + private val _isRestoring = MutableStateFlow(false) val isRestoring: StateFlow = _isRestoring.asStateFlow() @@ -165,10 +188,35 @@ class BackupRepo @Inject constructor( private val restorePendingUntil = MutableStateFlow(0L) - fun reset() { - stopObservingBackups() - vssBackupClient.reset() - vssBackupClientLdk.reset() + suspend fun reset() = withContext(ioDispatcher) { + resetMutex.withLock { + val pending = synchronized(lifecycleLock) { + resetting = true + observation = null + pendingScope.coroutineContext[Job] + } + withContext(NonCancellable) { + try { + pending?.cancel() + pending?.join() + synchronized(lifecycleLock) { generation += 1 } + vssBackupClient.reset() + vssBackupClientLdk.reset() + synchronized(lifecycleLock) { + backupJobs.clear() + explicitBackups.clear() + failedBackupRequired.clear() + temporaryFailureRetries.clear() + temporaryFailureSince.clear() + } + } finally { + synchronized(lifecycleLock) { + pendingScope = newBackupScope(scope) + resetting = false + } + } + } + } } fun setWiping(isWiping: Boolean) = _isWiping.update { isWiping } @@ -210,16 +258,29 @@ class BackupRepo @Inject constructor( failedBackupRequired[category] != this.required fun startObservingBackups() { - if (isObserving) return - if (_isWiping.value) { - Logger.debug("Skipped observing backups while wiping", context = TAG) - return + val startup = synchronized(lifecycleLock) { + if (observation != null || resetting || _isWiping.value) return + val previous = retiringObservations.toList() + val owner = Observation(newBackupScope(pendingScope), generation) + observation = owner + val parent = checkNotNull(owner.scope.coroutineContext[Job]) + retiringObservations.add(parent) + parent.invokeOnCompletion { synchronized(lifecycleLock) { retiringObservations.remove(parent) } } + owner.scope.launch(start = CoroutineStart.LAZY) { + // Record new changes during the drain; only uploads and client setup must wait. + startDataStoreListeners(owner) + previous.joinAll() + currentCoroutineContext().ensureActive() + owner.ready.complete(Unit) + startObservation(owner) + } } + startup.start() + } - isObserving = true + private fun startObservation(owner: Observation) { Logger.debug("Start observing backup statuses and data store changes", context = TAG) - - scope.launch { + owner.scope.launch { vssBackupClient.setupWithRetry { onSuccess = { attempt -> Logger.debug("VSS client setup succeeded on attempt $attempt", context = TAG) @@ -234,15 +295,16 @@ class BackupRepo @Inject constructor( Logger.warn("VSS client setup failed after $maxAttempts attempts", context = TAG) } }.onSuccess { - scope.launch { vssBackupClientLdk.setup() } + currentCoroutineContext().ensureActive() + vssBackupClientLdk.setup() } } - scope.launch { + owner.scope.launch { BackupCategory.entries.forEach { category -> if (category !in runningBackups) { - cacheStore.updateBackupStatus(category) { status -> - if (status.running) { + updateBackupStatus(category, owner.generation) { status -> + if (status.running && category !in runningBackups) { Logger.debug("Clearing stale running flag for: '$category'", context = TAG) status.copy(running = false) } else { @@ -253,39 +315,22 @@ class BackupRepo @Inject constructor( } } - startBackupStatusObservers() - startDataStoreListeners() - startPeriodicBackupFailureCheck() + startBackupStatusObservers(owner) + startPeriodicBackupFailureCheck(owner) } fun stopObservingBackups() { - if (!isObserving) return - - isObserving = false - - // Cancel all backup jobs - backupJobs.values.forEach { it.cancel() } - backupJobs.clear() - - // Cancel backup status observer jobs - statusObserverJobs.forEach { it.cancel() } - statusObserverJobs.clear() - - // Cancel data store listener jobs - dataListenerJobs.forEach { it.cancel() } - dataListenerJobs.clear() - - // Cancel periodic check job - periodicCheckJob?.cancel() - periodicCheckJob = null - + val parent = synchronized(lifecycleLock) { + observation?.scope?.coroutineContext?.get(Job).also { observation = null } + } + parent?.cancel() Logger.debug("Stopped observing backup statuses and data store changes", context = TAG) } - private fun startBackupStatusObservers() { + private fun startBackupStatusObservers(owner: Observation) { // Observe backup status changes for each category BackupCategory.entries.forEach { category -> - val job = scope.launch { + owner.scope.launch { cacheStore.backupStatuses .map { statuses -> statuses[category] ?: BackupItemStatus() } .distinctUntilChanged { old, new -> @@ -295,143 +340,126 @@ class BackupRepo @Inject constructor( } .collect { status -> if (status.shouldBackup(category)) { - scheduleBackup(category) + scheduleBackup(category, owner) } } } - statusObserverJobs.add(job) } - - Logger.debug("Started ${statusObserverJobs.size} backup status observers", context = TAG) } - private fun startDataStoreListeners() { - val settingsJob = scope.launch { + private fun startDataStoreListeners(owner: Observation) { + owner.scope.launch { settingsStore.data .distinctUntilChanged() .drop(1) .collect { if (shouldSkipBackup()) return@collect - markBackupRequired(BackupCategory.SETTINGS) + markBackupRequired(owner, BackupCategory.SETTINGS) } } - dataListenerJobs.add(settingsJob) - val widgetsJob = scope.launch { + owner.scope.launch { widgetsStore.data .distinctUntilChanged() .drop(1) .collect { if (shouldSkipBackup()) return@collect - markBackupRequired(BackupCategory.WIDGETS) + markBackupRequired(owner, BackupCategory.WIDGETS) } } - dataListenerJobs.add(widgetsJob) // WALLET - Observe transfers - val transfersJob = scope.launch { + owner.scope.launch { db.transferDao().observeAll() .distinctUntilChanged() .drop(1) .collect { if (shouldSkipBackup()) return@collect - markBackupRequired(BackupCategory.WALLET) + markBackupRequired(owner, BackupCategory.WALLET) } } - dataListenerJobs.add(transfersJob) - val watchOnlyAccountsJob = scope.launch { + owner.scope.launch { watchOnlyAccountStore.data .distinctUntilChanged() .drop(1) .collect { if (shouldSkipBackup()) return@collect - markBackupRequired(BackupCategory.WALLET) + markBackupRequired(owner, BackupCategory.WALLET) } } - dataListenerJobs.add(watchOnlyAccountsJob) // METADATA - Observe entire CacheStore excluding backup statuses - val cacheMetadataJob = scope.launch { + owner.scope.launch { cacheStore.data .map { it.copy(backupStatuses = mapOf()) } .distinctUntilChanged() .drop(1) .collect { if (shouldSkipBackup()) return@collect - markBackupRequired(BackupCategory.METADATA) + markBackupRequired(owner, BackupCategory.METADATA) } } - dataListenerJobs.add(cacheMetadataJob) // METADATA - Observe pre-activity metadata changes - val preActivityMetadataJob = scope.launch { + owner.scope.launch { preActivityMetadataRepo.preActivityMetadataChanged .drop(1) .collect { if (shouldSkipBackup()) return@collect - markBackupRequired(BackupCategory.METADATA) + markBackupRequired(owner, BackupCategory.METADATA) } } - dataListenerJobs.add(preActivityMetadataJob) // METADATA - Observe hardware wallet names only: the store is also rewritten by every connect, // and reconnect traffic must not re-upload the whole metadata envelope. - val hwWalletNamesJob = scope.launch { + owner.scope.launch { hwWalletStore.data .map { it.hwWalletNames() } .distinctUntilChanged() .drop(1) .collect { if (shouldSkipBackup()) return@collect - markBackupRequired(BackupCategory.METADATA) + markBackupRequired(owner, BackupCategory.METADATA) } } - dataListenerJobs.add(hwWalletNamesJob) - - dataListenerJobs.add(observeBackupChanges(pubkyRepo.backupStateVersion, BackupCategory.METADATA)) - dataListenerJobs.add(observeBackupChanges(privatePaykitRepo.get().backupStateVersion, BackupCategory.WALLET)) - dataListenerJobs.add(observeBackupChanges(paykitSdkService.backupStateVersion, BackupCategory.WALLET)) - dataListenerJobs.add(observeBackupChanges(paykitPaymentProofStore.backupStateVersion, BackupCategory.WALLET)) - dataListenerJobs.add(observeBackupChanges(onchainSendAttemptStore.backupStateVersion, BackupCategory.WALLET)) - dataListenerJobs.add( - observeBackupChanges(paykitUsdtPaymentRepo.get().backupStateVersion, BackupCategory.WALLET) - ) - dataListenerJobs.add( - observeBackupChanges(paykitUsdtPaymentRepo.get().backupStateVersion, BackupCategory.USDT_RECOVERY) - ) - dataListenerJobs.add( - observeBackupChanges( - usdtRepo.get().state.map { it.transfers }.distinctUntilChanged(), - BackupCategory.USDT_RECOVERY, - ) + + observeBackupChanges(owner, pubkyRepo.backupStateVersion, BackupCategory.METADATA) + observeBackupChanges(owner, privatePaykitRepo.get().backupStateVersion, BackupCategory.WALLET) + observeBackupChanges(owner, paykitSdkService.backupStateVersion, BackupCategory.WALLET) + observeBackupChanges(owner, paykitPaymentProofStore.backupStateVersion, BackupCategory.WALLET) + observeBackupChanges(owner, onchainSendAttemptStore.backupStateVersion, BackupCategory.WALLET) + observeBackupChanges(owner, paykitUsdtPaymentRepo.get().backupStateVersion, BackupCategory.WALLET) + observeBackupChanges(owner, paykitUsdtPaymentRepo.get().backupStateVersion, BackupCategory.USDT_RECOVERY) + observeBackupChanges( + owner, + usdtRepo.get().state.map { it.transfers }.distinctUntilChanged(), + BackupCategory.USDT_RECOVERY, ) - dataListenerJobs.add( - observeBackupChanges( - usdtRepo.get().state.map { it.transfers }.distinctUntilChanged(), - BackupCategory.WALLET, - ) + observeBackupChanges( + owner, + usdtRepo.get().state.map { it.transfers }.distinctUntilChanged(), + BackupCategory.WALLET, ) - dataListenerJobs.add(observeBackupChanges(paykitPresentationStore.backupStateVersion, BackupCategory.WALLET)) - dataListenerJobs.add( - observeBackupChanges( - privatePaykitAddressReservationRepo.get().backupStateVersion, - BackupCategory.WALLET, - ) + observeBackupChanges(owner, paykitPresentationStore.backupStateVersion, BackupCategory.WALLET) + observeBackupChanges( + owner, + privatePaykitAddressReservationRepo.get().backupStateVersion, + BackupCategory.WALLET, ) // BLOCKTANK - Observe blocktank state changes (orders, cjitEntries, info) - dataListenerJobs.add(observeBackupChanges(blocktankRepo.blocktankState, BackupCategory.BLOCKTANK)) + observeBackupChanges(owner, blocktankRepo.blocktankState, BackupCategory.BLOCKTANK) // ACTIVITY - Observe activity changes - dataListenerJobs.add(observeBackupChanges(activityRepo.activitiesChanged, BackupCategory.ACTIVITY)) + observeBackupChanges(owner, activityRepo.activitiesChanged, BackupCategory.ACTIVITY) // Hardware tags are carried by the metadata backup. Observe the narrower tag signal so ordinary // payment and sync traffic does not re-upload the whole metadata envelope. - dataListenerJobs.add(observeBackupChanges(activityRepo.activityTagsChanged, BackupCategory.METADATA)) + observeBackupChanges(owner, activityRepo.activityTagsChanged, BackupCategory.METADATA) // LIGHTNING_CONNECTIONS - Only display sync timestamp, ldk-node manages its own backups @OptIn(FlowPreview::class) - val lightningConnectionsJob = scope.launch { + owner.scope.launch { lightningService.syncStatusChanged .debounce(SYNC_STATUS_DEBOUNCE) .collect { @@ -439,87 +467,143 @@ class BackupRepo @Inject constructor( ?.let { it * 1000 } // Convert seconds to millis ?: return@collect if (shouldSkipBackup()) return@collect - cacheStore.updateBackupStatus(BackupCategory.LIGHTNING_CONNECTIONS) { + updateBackupStatus(BackupCategory.LIGHTNING_CONNECTIONS, owner.generation) { it.copy(required = lastSync, synced = lastSync, running = false) } } } - dataListenerJobs.add(lightningConnectionsJob) - - Logger.debug("Started ${dataListenerJobs.size} data store listeners", context = TAG) } - private fun observeBackupChanges(flow: Flow<*>, category: BackupCategory): Job = - scope.launch { + private fun observeBackupChanges(owner: Observation, flow: Flow<*>, category: BackupCategory): Job = + owner.scope.launch { flow.drop(1).collect { if (shouldSkipBackup(category)) return@collect - markBackupRequired(category) + markBackupRequired(owner, category) } } - private fun startPeriodicBackupFailureCheck() { - periodicCheckJob = scope.launch { + private fun startPeriodicBackupFailureCheck(owner: Observation) { + owner.scope.launch { while (currentCoroutineContext().isActive) { delay(BACKUP_CHECK_INTERVAL) - checkForFailedBackups() + checkForFailedBackups(owner) } } } - private fun markBackupRequired(category: BackupCategory) { - scope.launch { - failedBackupRequired -= category - cacheStore.updateBackupStatus(category) { + private suspend fun markBackupRequired(owner: Observation, category: BackupCategory) = + markBackupRequired(category, owner.generation) + + private suspend fun markBackupRequired(category: BackupCategory, expectedGeneration: Long) = + withContext(NonCancellable) { + // A consumed change must remain pending even if observation stops before the cache write. + updateBackupStatus(category, expectedGeneration) { + failedBackupRequired -= category it.copy(required = maxOf(currentTimeMillis(), it.required + 1, it.synced + 1)) } - Logger.verbose("Marked backup required for: '$category'", context = TAG) } - } - - private fun scheduleBackup(category: BackupCategory) { - backupJobs[category]?.cancel() - - Logger.verbose("Scheduling backup for: '$category'", context = TAG) - backupJobs[category] = scope.launch { - runningBackups += category - cacheStore.updateBackupStatus(category) { - it.copy(running = true) + private suspend fun updateBackupStatus( + category: BackupCategory, + expectedGeneration: Long, + transform: (BackupItemStatus) -> BackupItemStatus, + ) { + cacheStore.update { data -> + synchronized(lifecycleLock) { + // Reset drains admitted writers before advancing the generation or clearing clients. + if (generation != expectedGeneration) return@update data + val status = data.backupStatuses[category] ?: BackupItemStatus() + data.copy(backupStatuses = data.backupStatuses + (category to transform(status))) } + } + } - delay(BACKUP_DEBOUNCE) - if (category == BackupCategory.WALLET) paykitSdkService.isPaymentSubmissionActive.first { !it } - - val status = cacheStore.backupStatuses.first()[category] ?: BackupItemStatus() - if (status.isRequired && !shouldSkipBackup(category)) { - triggerBackup(category) - } else { - Logger.debug("Backup no longer needed for: '$category'", context = TAG) - runningBackups -= category - cacheStore.updateBackupStatus(category) { - it.copy(running = false) - } + private fun scheduleBackup( + category: BackupCategory, + owner: Observation? = null, + requestedGeneration: Long? = null, + ) { + val job = synchronized(lifecycleLock) { + if (resetting || _isWiping.value || (owner != null && observation !== owner) || + (requestedGeneration != null && requestedGeneration != generation) + ) { + return } - }.also { job -> - job.invokeOnCompletion { exception -> - if (exception != null) { - Logger.debug("Backup job cancelled for: '$category'", context = TAG) - scope.launch { - runningBackups -= category - cacheStore.updateBackupStatus(category) { - it.copy(running = false) + if (owner == null) explicitBackups += category + val existing = backupJobs[category] + if (existing != null && !existing.isCompleted && !existing.isCancelled) return + val expectedGeneration = generation + val jobScope = owner?.scope ?: pendingScope + jobScope.launch(start = CoroutineStart.LAZY) { + try { + owner?.ready?.await() + categoryMutexes.getValue(category).withLock { + runningBackups += category + try { + updateBackupStatus(category, expectedGeneration) { it.copy(running = true) } + delay(BACKUP_DEBOUNCE) + if (category == BackupCategory.WALLET) { + paykitSdkService.isPaymentSubmissionActive.first { !it } + } + val status = cacheStore.backupStatuses.first()[category] ?: BackupItemStatus() + if (status.isRequired && !shouldSkipBackup(category) && + !failedBackupRequired.containsKey(category) + ) { + performBackup(category, expectedGeneration) + } + } finally { + finishBackup(category, expectedGeneration) } } + } finally { + val currentJob = currentCoroutineContext()[Job] + synchronized(lifecycleLock) { backupJobs.remove(category, currentJob) } + withContext(NonCancellable) { + schedulePendingBackup(category, expectedGeneration, owner) + } } + }.also { backupJobs[category] = it } + } + job.start() + } + + private suspend fun schedulePendingBackup( + category: BackupCategory, + expectedGeneration: Long, + previousOwner: Observation? = null, + ) { + var status = BackupItemStatus() + updateBackupStatus(category, expectedGeneration) { + status = it + if (!it.isRequired) explicitBackups -= category + it + } + synchronized(lifecycleLock) { + if (resetting || generation != expectedGeneration) return + if (failedBackupRequired.containsKey(category)) return + if (!status.shouldBackup(category)) return + if (category in explicitBackups) { + scheduleBackup(category, requestedGeneration = expectedGeneration) + } else { + val owner = observation?.takeIf { previousOwner == null || previousOwner === it } ?: return + if (owner.scope.isActive) scheduleBackup(category, owner) } } } - private fun checkForFailedBackups() { + private suspend fun finishBackup(category: BackupCategory, expectedGeneration: Long) = withContext(NonCancellable) { + try { + updateBackupStatus(category, expectedGeneration) { it.copy(running = false) } + } finally { + runningBackups -= category + } + } + + private fun checkForFailedBackups(owner: Observation) { val currentTime = currentTimeMillis() // find if there are any backup categories that have been failing for more than 30 minutes - scope.launch { + owner.scope.launch { val backupStatuses = cacheStore.backupStatuses.first() val hasFailedBackups = BackupCategory.entries.any { category -> val status = backupStatuses[category] ?: BackupItemStatus() @@ -531,18 +615,18 @@ class BackupRepo @Inject constructor( } if (hasFailedBackups) { - showBackupFailureNotification(currentTime) + showBackupFailureNotification(owner, currentTime) } } } - private fun showBackupFailureNotification(currentTime: Long) { + private fun showBackupFailureNotification(owner: Observation, currentTime: Long) { // Throttle notifications to avoid spam if (currentTime - lastNotificationTime < FAILED_BACKUP_NOTIFICATION_INTERVAL) return lastNotificationTime = currentTime - scope.launch { + owner.scope.launch { ToastEventBus.send( type = Toast.ToastType.ERROR, title = context.getString(R.string.settings__backup__failed_title), @@ -553,57 +637,119 @@ class BackupRepo @Inject constructor( } } - suspend fun triggerBackup(category: BackupCategory): Result = withContext(ioDispatcher) { - Logger.debug("Backup starting for: '$category'", context = TAG) - var backupRequired = currentTimeMillis() - runningBackups += category - failedBackupRequired -= category - try { - cacheStore.updateBackupStatus(category) { - backupRequired = maxOf(backupRequired, it.required, it.synced + 1) - it.copy(running = true, required = backupRequired) - } + suspend fun triggerBackup(category: BackupCategory): Result = performExplicitBackup(category) - val upload = when (category) { - BackupCategory.WALLET -> uploadWalletBackup() - BackupCategory.USDT_RECOVERY -> uploadUsdtRecoveryBackup() - else -> runSuspendCatching { - vssBackupClient.putObject(key = category.name, data = getBackupDataBytes(category)).getOrThrow() + private suspend fun performExplicitBackup( + category: BackupCategory, + usdtSnapshot: String? = null, + ): Result = withBackupOperation { expectedGeneration -> + var reservedRequired: Long? = null + if (usdtSnapshot != null) { + withContext(NonCancellable) { + updateBackupStatus(category, expectedGeneration) { + val required = maxOf(currentTimeMillis(), it.required + 1, it.synced + 1) + reservedRequired = required + it.copy(required = required) } } - upload - .onSuccess { - withContext(NonCancellable) { - cacheStore.updateBackupStatus(category) { - it.copy(running = false, synced = backupRequired) - } - runningBackups -= category - failedBackupRequired -= category - temporaryFailureRetries -= category - temporaryFailureSince -= category - } - Logger.info("Backup succeeded for: '$category'", context = TAG) - } - .onFailure { markBackupFailed(category, backupRequired, it) } - .map {} - } finally { - if (runningBackups.remove(category)) { + } + categoryMutexes.getValue(category).withLock { + if (synchronized(lifecycleLock) { resetting || generation != expectedGeneration }) { + return@withLock Result.failure(UsdtException.BackupUnavailable()) + } + runningBackups += category + try { + performBackup( + category, + expectedGeneration, + usdtSnapshot, + explicitAttempt = true, + reservedRequired = reservedRequired, + ) + } finally { + finishBackup(category, expectedGeneration) withContext(NonCancellable) { - cacheStore.updateBackupStatus(category) { it.copy(running = false) } + schedulePendingBackup(category, expectedGeneration) } } } } - suspend fun persistWalletBackup(): Result = - uploadWalletBackup().onFailure { markBackupRequired(BackupCategory.WALLET) } + private suspend fun performBackup( + category: BackupCategory, + expectedGeneration: Long, + usdtSnapshot: String? = null, + explicitAttempt: Boolean = false, + reservedRequired: Long? = null, + ): Result { + Logger.debug("Backup starting for: '$category'", context = TAG) + var backupRequired = reservedRequired ?: currentTimeMillis() + updateBackupStatus(category, expectedGeneration) { + if (reservedRequired == null) { + val required = if (explicitAttempt) it.required + 1 else it.required + backupRequired = maxOf(backupRequired, required, it.synced + 1) + } + if ((failedBackupRequired[category] ?: 0) <= backupRequired) failedBackupRequired -= category + it.copy(running = true, required = maxOf(it.required, backupRequired)) + } + if (synchronized(lifecycleLock) { resetting || generation != expectedGeneration }) { + return Result.failure(UsdtException.BackupUnavailable()) + } + val upload = when (category) { + BackupCategory.WALLET -> uploadWalletBackup() + BackupCategory.USDT_RECOVERY -> uploadUsdtRecoveryBackup(usdtSnapshot) + else -> runSuspendCatching { + vssBackupClient.putObject(key = category.name, data = getBackupDataBytes(category)).getOrThrow() + } + } + return upload + .onSuccess { + withContext(NonCancellable) { + updateBackupStatus(category, expectedGeneration) { + if ((failedBackupRequired[category] ?: 0) <= backupRequired) { + failedBackupRequired -= category + temporaryFailureRetries -= category + temporaryFailureSince -= category + } + if (it.required <= backupRequired) { + explicitBackups -= category + } else if (explicitAttempt) { + explicitBackups += category + } + it.copy(running = false, synced = maxOf(it.synced, backupRequired)) + } + } + Logger.info("Backup succeeded for: '$category'", context = TAG) + } + .onFailure { markBackupFailed(category, backupRequired, expectedGeneration, it, explicitAttempt) } + .map { } + } + + suspend fun persistWalletBackup(): Result = withBackupOperation { expectedGeneration -> + uploadWalletBackup().onFailure { markBackupRequired(BackupCategory.WALLET, expectedGeneration) } + } + + suspend fun persistUsdtRecoveryBackup(snapshot: String): Result = + performExplicitBackup(BackupCategory.USDT_RECOVERY, snapshot) - suspend fun persistUsdtRecoveryBackup(snapshot: String): Result = runSuspendCatching { - cacheStore.updateBackupStatus(BackupCategory.USDT_RECOVERY) { - it.copy(required = maxOf(currentTimeMillis(), it.required + 1, it.synced + 1)) + private suspend fun withBackupOperation( + operation: suspend (Long) -> Result, + ): Result = withContext(ioDispatcher) { + val callerContext = currentCoroutineContext().minusKey(Job) + val task = synchronized(lifecycleLock) { + if (resetting || _isWiping.value) return@withContext Result.failure(UsdtException.BackupUnavailable()) + val expectedGeneration = generation + pendingScope.async(context = callerContext, start = CoroutineStart.LAZY) { operation(expectedGeneration) } } - uploadUsdtRecoveryBackup(snapshot).getOrThrow() - }.onFailure { markBackupRequired(BackupCategory.USDT_RECOVERY) } + try { + task.await() + } catch (e: CancellationException) { + currentCoroutineContext().ensureActive() + Result.failure(UsdtException.BackupUnavailable().apply { initCause(e) }) + } finally { + withContext(NonCancellable) { task.cancelAndJoin() } + } + } private suspend fun uploadUsdtRecoveryBackup(snapshot: String? = null): Result = withContext(ioDispatcher) { runSuspendCatching { @@ -641,38 +787,68 @@ class BackupRepo @Inject constructor( }.map { } } - private suspend fun markBackupFailed(category: BackupCategory, backupRequired: Long, e: Throwable) { + private suspend fun markBackupFailed( + category: BackupCategory, + backupRequired: Long, + expectedGeneration: Long, + e: Throwable, + explicitAttempt: Boolean, + ) { + val temporary = e.isTemporaryBackupFailure() withContext(NonCancellable) { - cacheStore.updateBackupStatus(category) { + updateBackupStatus(category, expectedGeneration) { + if (temporary && explicitAttempt) explicitBackups += category + if (!temporary && e !is WalletBackupSnapshotChanged && it.required == backupRequired) { + explicitBackups -= category + } if (it.required == backupRequired && e !is WalletBackupSnapshotChanged) { failedBackupRequired[category] = backupRequired - } else { + } else if ((failedBackupRequired[category] ?: 0) <= backupRequired) { failedBackupRequired -= category } it.copy(running = false) } - runningBackups -= category } when { e is WalletBackupSnapshotChanged -> Logger.debug("Deferred changing backup snapshot for: '$category'", context = TAG) - e.isTemporaryBackupFailure() -> retryAfterTemporaryFailure(category, backupRequired, e) + temporary -> retryAfterTemporaryFailure(category, backupRequired, expectedGeneration, e) else -> Logger.error("Backup failed for: '$category'", e, context = TAG) } } - private fun retryAfterTemporaryFailure(category: BackupCategory, backupRequired: Long, e: Throwable) { - val attempt = temporaryFailureRetries.merge(category, 1, Int::plus) ?: 1 - temporaryFailureSince.putIfAbsent(category, backupRequired) - val retryDelay = (TEMPORARY_FAILURE_RETRY_DELAY * (1 shl (attempt - 1).coerceAtMost(MAX_RETRY_SHIFT))) - .coerceAtMost(TEMPORARY_FAILURE_RETRY_MAX_DELAY) - Logger.warn("Retrying backup for: '$category' in '$retryDelay'", e, context = TAG) - scope.launch { - delay(retryDelay) - if (!failedBackupRequired.remove(category, backupRequired) || !isObserving) return@launch - val status = cacheStore.backupStatuses.first()[category] ?: return@launch - if (status.shouldBackup(category)) scheduleBackup(category) + private fun retryAfterTemporaryFailure( + category: BackupCategory, + backupRequired: Long, + expectedGeneration: Long, + e: Throwable, + ) { + val job = synchronized(lifecycleLock) { + if (resetting || generation != expectedGeneration || + (failedBackupRequired[category] ?: 0) > backupRequired + ) { + return + } + val attempt = temporaryFailureRetries.merge(category, 1, Int::plus) ?: 1 + temporaryFailureSince.putIfAbsent(category, backupRequired) + val retryDelay = (TEMPORARY_FAILURE_RETRY_DELAY * (1 shl (attempt - 1).coerceAtMost(MAX_RETRY_SHIFT))) + .coerceAtMost(TEMPORARY_FAILURE_RETRY_MAX_DELAY) + Logger.warn("Retrying backup for: '$category' in '$retryDelay'", e, context = TAG) + pendingScope.launch(start = CoroutineStart.LAZY) { + delay(retryDelay) + synchronized(lifecycleLock) { + if (resetting || generation != expectedGeneration) return@launch + val failed = failedBackupRequired[category] + if (failed == backupRequired) { + failedBackupRequired.remove(category, backupRequired) + } else if (failed != null || category !in explicitBackups) { + return@launch + } + } + schedulePendingBackup(category, expectedGeneration) + } } + job.start() } private suspend fun getBackupDataBytes(category: BackupCategory): ByteArray = when (category) { @@ -1157,10 +1333,11 @@ class BackupRepo @Inject constructor( fun scheduleFullBackup() { Logger.debug("Scheduling backups for all categories", context = TAG) + val requestedGeneration = synchronized(lifecycleLock) { generation } BackupCategory.entries .filter { it != BackupCategory.LIGHTNING_CONNECTIONS } .forEach { - scheduleBackup(it) + scheduleBackup(it, requestedGeneration = requestedGeneration) } } @@ -1307,5 +1484,13 @@ internal fun Throwable.isTemporaryBackupFailure(): Boolean = isPaykitTemporarily is VssException.StoreException, is VssException.PutException, -> true + is VssException.UnknownException -> isVssRequestSendFailure(errorDetails) else -> false } + +private fun isVssRequestSendFailure(details: String): Boolean { + val prefix = "error sending request for url (" + if (!details.startsWith(prefix) || !details.endsWith(")")) return false + val url = runCatching { URI(details.removePrefix(prefix).dropLast(1)) }.getOrNull() ?: return false + return url.scheme in setOf("http", "https") && !url.host.isNullOrEmpty() && url.path.endsWith("/putObjects") +} diff --git a/app/src/test/java/to/bitkit/repositories/BackupRepoTest.kt b/app/src/test/java/to/bitkit/repositories/BackupRepoTest.kt index 79c44c5a73..d839c91989 100644 --- a/app/src/test/java/to/bitkit/repositories/BackupRepoTest.kt +++ b/app/src/test/java/to/bitkit/repositories/BackupRepoTest.kt @@ -1,22 +1,37 @@ package to.bitkit.repositories import android.content.Context +import androidx.datastore.core.DataStore +import androidx.datastore.core.DataStoreFactory import com.synonym.bitkitcore.ActivityTags import com.synonym.bitkitcore.PreActivityMetadata import com.synonym.paykit.PaykitException +import com.synonym.vssclient.VssException import com.synonym.vssclient.VssItem import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.CoroutineDispatcher +import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.NonCancellable +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.asCoroutineDispatcher import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.awaitCancellation import kotlinx.coroutines.cancelAndJoin import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.map import kotlinx.coroutines.flow.update import kotlinx.coroutines.launch +import kotlinx.coroutines.test.TestScope import kotlinx.coroutines.test.advanceTimeBy import kotlinx.coroutines.test.currentTime import kotlinx.coroutines.test.runCurrent +import kotlinx.coroutines.withContext +import kotlinx.coroutines.withTimeout +import kotlinx.coroutines.yield import kotlinx.serialization.json.JsonObject import kotlinx.serialization.json.jsonObject import org.junit.Before @@ -57,6 +72,7 @@ import to.bitkit.data.backup.VssBackupClientLdk import to.bitkit.data.dao.TransferDao import to.bitkit.data.entities.TransferEntity import to.bitkit.data.keychain.Keychain +import to.bitkit.data.serializers.AppCacheSerializer import to.bitkit.di.json import to.bitkit.models.ActivityBackupV1 import to.bitkit.models.BackupCategory @@ -71,10 +87,15 @@ import to.bitkit.models.WalletScope import to.bitkit.models.WatchOnlyAccountRecord import to.bitkit.models.WatchOnlyAccountSetupState import to.bitkit.services.LightningService +import to.bitkit.services.PaykitSdkOperationLock import to.bitkit.services.PaykitSdkService +import to.bitkit.services.UsdtService import to.bitkit.test.BaseUnitTest import to.bitkit.test.connectionRecoveryFixture import to.bitkit.utils.AppError +import java.nio.file.Files +import java.util.concurrent.Executors +import java.util.concurrent.atomic.AtomicInteger import javax.inject.Provider import kotlin.coroutines.cancellation.CancellationException import kotlin.test.assertEquals @@ -84,6 +105,7 @@ import kotlin.test.assertNull import kotlin.test.assertTrue import kotlin.time.Clock import kotlin.time.Duration.Companion.minutes +import kotlin.time.Duration.Companion.seconds import kotlin.time.ExperimentalTime import kotlin.time.Instant @@ -152,8 +174,12 @@ class BackupRepoTest : BaseUnitTest() { } whenever { transferDao.upsert(any>()) }.thenReturn(Unit) whenever { cacheStore.updateBackupStatus(any(), any()) }.thenReturn(Unit) - whenever { cacheStore.update(any()) }.thenReturn(Unit) + whenever { cacheStore.update(any()) }.doSuspendableAnswer { + val transform = it.getArgument<(AppCacheData) -> AppCacheData>(0) + cacheData.update(transform) + } whenever(cacheStore.data).thenReturn(cacheData) + whenever(cacheStore.backupStatuses).thenReturn(cacheData.map { it.backupStatuses }) whenever(settingsStore.data).thenReturn(settingsData) whenever { settingsStore.update(any()) }.thenReturn(Unit) whenever(widgetsStore.data).thenReturn(widgetsData) @@ -298,14 +324,14 @@ class BackupRepoTest : BaseUnitTest() { val statusWriteStarted = CompletableDeferred() val finishStatusWrite = CompletableDeferred() whenever(cacheStore.backupStatuses).thenReturn(statuses) - whenever(cacheStore.updateBackupStatus(eq(category), any())).doSuspendableAnswer { - val transform = it.getArgument<(BackupItemStatus) -> BackupItemStatus>(1) - val updated = transform(statuses.value.getValue(category)) - if (!updated.running) { + whenever(cacheStore.update(any())).doSuspendableAnswer { + val transform = it.getArgument<(AppCacheData) -> AppCacheData>(0) + val updated = transform(AppCacheData(backupStatuses = statuses.value)).backupStatuses + if (!updated.getValue(category).running) { statusWriteStarted.complete(Unit) finishStatusWrite.await() } - statuses.update { current -> current + (category to updated) } + statuses.value = updated } val uploadResult = if (uploadSucceeds) { Result.success(VssItem(key = category.name, value = byteArrayOf(), version = 1)) @@ -964,16 +990,20 @@ class BackupRepoTest : BaseUnitTest() { whenever(privatePaykitRepo.backupSnapshot()).thenReturn( Result.failure(PaykitException.SharedStateBusy("shared_state_busy", "lock unavailable")), ) - assertTrue(sut.triggerBackup(BackupCategory.WALLET).isFailure) - sut.persistUsdtRecoveryBackup("same-signed-operation").getOrThrow() - val payload = argumentCaptor() - verify(vssBackupClient).putObject(argThat { startsWith(UsdtRecoveryBackupStore.PREFIX) }, payload.capture()) - assertEquals( - "same-signed-operation", - json.decodeFromString(payload.firstValue.decodeToString()).usdtWallet, - ) - verify(privatePaykitRepo, times(1)).backupSnapshot() - verify(vssBackupClient, never()).putObject(eq(BackupCategory.WALLET.name), any()) + try { + assertTrue(sut.triggerBackup(BackupCategory.WALLET).isFailure) + sut.persistUsdtRecoveryBackup("same-signed-operation").getOrThrow() + val payload = argumentCaptor() + verify(vssBackupClient).putObject(argThat { startsWith(UsdtRecoveryBackupStore.PREFIX) }, payload.capture()) + assertEquals( + "same-signed-operation", + json.decodeFromString(payload.firstValue.decodeToString()).usdtWallet, + ) + verify(privatePaykitRepo, times(1)).backupSnapshot() + verify(vssBackupClient, never()).putObject(eq(BackupCategory.WALLET.name), any()) + } finally { + sut.reset() + } } @Test @@ -1230,6 +1260,59 @@ class BackupRepoTest : BaseUnitTest() { } } + @Test + fun `released VSS connection failure is temporary without broad unknown retry`() { + assertTrue( + VssException.UnknownException("error sending request for url (http://127.0.0.1:1/putObjects)") + .isTemporaryBackupFailure(), + ) + listOf( + "invalid ciphertext", + "permission denied", + "HTTP status client error (401 Unauthorized)", + "error sending request for url (not a URL)", + "prefix error sending request for url (https://example.com)", + "error sending request for url (https://example.com/getObject)", + "error sending request for url (file://example.com/putObjects)", + "error sending request for url (https:///putObjects)", + "error sending request for url (https://example.com/putObjects) suffix", + ).forEach { assertFalse(VssException.UnknownException(it).isTemporaryBackupFailure()) } + } + + @Test + fun `settings backup retries the released VSS connection error and acknowledges recovery`() = test { + whenever(clock.now()).thenReturn(Instant.fromEpochMilliseconds(3_000)) + val category = BackupCategory.SETTINGS + val backupStatuses = MutableStateFlow(mapOf(category to BackupItemStatus(synced = 1_000, required = 2_000))) + stubBackupStatuses(backupStatuses, CompletableDeferred(Unit)) {} + stubBackupObservers() + var attempts = 0 + whenever(vssBackupClient.putObject(eq(category.name), any())).doSuspendableAnswer { + attempts += 1 + if (attempts == 1) { + Result.failure( + VssException.UnknownException("error sending request for url (http://127.0.0.1:1/putObjects)") + ) + } else { + Result.success(VssItem(category.name, it.getArgument(1), 1)) + } + } + try { + sut.startObservingBackups() + runCurrent() + advanceTimeBy(5_000) + runCurrent() + assertEquals(1, attempts) + assertTrue(backupStatuses.value.getValue(category).isRequired) + advanceTimeBy(20_000) + runCurrent() + assertEquals(2, attempts) + assertFalse(backupStatuses.value.getValue(category).isRequired) + } finally { + sut.stopObservingBackups() + } + } + @Test fun `wallet backup retries after a busy Paykit shared state`() = test { whenever(clock.now()).thenReturn(Instant.fromEpochMilliseconds(3_000)) @@ -1268,229 +1351,1315 @@ class BackupRepoTest : BaseUnitTest() { } @Test - fun `busy wallet backup retries after observation pauses and resumes`() = test { - whenever(clock.now()).thenReturn(Instant.fromEpochMilliseconds(3_000)) - whenever { privatePaykitRepo.backupSnapshot() } - .thenReturn(Result.failure(PaykitException.SharedStateBusy("shared_state_busy", "locked"))) - .thenReturn(Result.success(PrivatePaykitBackupSnapshot(null, 0))) - val backupStatuses = MutableStateFlow( - mapOf(BackupCategory.WALLET to BackupItemStatus(synced = 1_000, required = 2_000)) - ) + fun `reset prevents an interrupted observer setup from initializing the old LDK client`() = test { + val setupStarted = CompletableDeferred() + val allowSetup = CompletableDeferred() + val backupStatuses = MutableStateFlow(emptyMap()) val allowWalletClear = CompletableDeferred().apply { complete(Unit) } stubBackupStatuses(backupStatuses, allowWalletClear) {} stubBackupObservers() + whenever { vssBackupClient.setupWithRetry(any(), any(), any()) }.doSuspendableAnswer { + setupStarted.complete(Unit) + allowSetup.await() + Result.success(Unit) + } - try { - sut.startObservingBackups() - runCurrent() - advanceTimeBy(5_000) - runCurrent() - sut.stopObservingBackups() - advanceTimeBy(20_000) - runCurrent() - verify(vssBackupClient, never()).putObject(eq(BackupCategory.WALLET.name), any()) + sut.startObservingBackups() + setupStarted.await() + sut.setWiping(true) + val reset = async { sut.reset() } + runCurrent() + allowSetup.complete(Unit) + reset.await() + runCurrent() - sut.startObservingBackups() - runCurrent() - advanceTimeBy(5_000) - runCurrent() + verify(vssBackupClient).reset() + verify(vssBackupClientLdk).reset() + verify(vssBackupClientLdk, never()).setup() + } - verify(vssBackupClient).putObject(eq(BackupCategory.WALLET.name), any()) - } finally { - sut.stopObservingBackups() - } + @Test + fun `consumed settings change survives observation stop`() = test { + assertConsumedSettingsChangeSurvivesStop(reset = false) } @Test - fun `busy wallet backup still warns once it stays pending past the failure window`() = test { - val startedAt = currentTime - whenever(clock.now()).thenAnswer { Instant.fromEpochMilliseconds(3_000 + currentTime - startedAt) } - whenever { privatePaykitRepo.backupSnapshot() } - .thenReturn(Result.failure(PaykitException.SharedStateBusy("shared_state_busy", "locked"))) - whenever(context.getString(R.string.settings__backup__failed_title)).thenReturn("Backup failed") - whenever(context.getString(R.string.settings__backup__failed_message)).thenReturn("Retrying") - val backupStatuses = MutableStateFlow( - mapOf(BackupCategory.WALLET to BackupItemStatus(synced = 1_000, required = 2_000)) - ) - val allowWalletClear = CompletableDeferred().apply { complete(Unit) } - stubBackupStatuses(backupStatuses, allowWalletClear) {} + fun `consumed settings change survives reset when wipe does not clear data`() = test { + assertConsumedSettingsChangeSurvivesStop(reset = true) + } + + private suspend fun TestScope.assertConsumedSettingsChangeSurvivesStop(reset: Boolean) { + val category = BackupCategory.SETTINGS + val statuses = MutableStateFlow(mapOf(category to BackupItemStatus(required = 1_000, synced = 1_000))) + stubBackupStatuses(statuses, CompletableDeferred(Unit)) {} stubBackupObservers() + sut.startObservingBackups() + runCurrent() + val entered = CompletableDeferred() + val release = CompletableDeferred() + var holdNextWrite = true + whenever { cacheStore.update(any()) }.doSuspendableAnswer { + val transform = it.getArgument<(AppCacheData) -> AppCacheData>(0) + if (holdNextWrite) { + holdNextWrite = false + entered.complete(Unit) + release.await() + } + statuses.update { current -> transform(AppCacheData(backupStatuses = current)).backupStatuses } + } + settingsData.update { it.copy(selectedCurrency = "EUR") } + entered.await() + val resetJob = if (reset) async { sut.reset() } else null + if (!reset) sut.stopObservingBackups() try { - sut.startObservingBackups() runCurrent() - advanceTimeBy(31.minutes.inWholeMilliseconds) + release.complete(Unit) + resetJob?.await() runCurrent() - - verify(context, atLeastOnce()).getString(R.string.settings__backup__failed_title) + assertTrue(statuses.value.getValue(category).isRequired) + sut.startObservingBackups() + advanceTimeBy(5_000) + runCurrent() + val payload = argumentCaptor() + verify(vssBackupClient).putObject(eq(category.name), payload.capture()) + assertTrue(payload.firstValue.decodeToString().contains("EUR")) + assertFalse(statuses.value.getValue(category).isRequired) } finally { - sut.stopObservingBackups() + release.complete(Unit) + resetJob?.join() + sut.reset() } } @Test - fun `start observing backs up stale required status after clearing running flag`() = test { - val backupStatuses = MutableStateFlow( - mapOf( - BackupCategory.WALLET to BackupItemStatus( - running = true, - synced = 1_000, - required = 2_000, - ), - ) - ) - val allowWalletClear = CompletableDeferred() - var delayedWalletClear = false - stubBackupStatuses(backupStatuses, allowWalletClear) { - delayedWalletClear = true - } + fun `reset drains cancelled setup even after stop and restart`() = test { + val entered = CompletableDeferred() + val release = CompletableDeferred() + val statuses = MutableStateFlow(emptyMap()) + stubBackupStatuses(statuses, CompletableDeferred(Unit)) {} stubBackupObservers() + whenever { vssBackupClient.setupWithRetry(any(), any(), any()) }.doSuspendableAnswer { + entered.complete(Unit) + withContext(NonCancellable) { release.await() } + Result.success(Unit) + } + sut.startObservingBackups() + entered.await() + sut.stopObservingBackups() + sut.startObservingBackups() + sut.setWiping(true) + val reset = async { sut.reset() } try { - sut.startObservingBackups() - runCurrent() - verify(vssBackupClient, never()).putObject(eq(BackupCategory.WALLET.name), any()) - - allowWalletClear.complete(Unit) runCurrent() - advanceTimeBy(5_000) + assertFalse(reset.isCompleted) + verify(vssBackupClient, never()).reset() + verify(vssBackupClientLdk, never()).reset() + verify(vssBackupClient, times(1)).setupWithRetry(any(), any(), any()) + reset.cancel() runCurrent() - - assertTrue(delayedWalletClear) - verify(vssBackupClient).putObject(eq(BackupCategory.WALLET.name), any()) + assertFalse(reset.isCompleted) } finally { - sut.stopObservingBackups() + release.complete(Unit) + reset.join() } + verify(vssBackupClient).reset() + verify(vssBackupClientLdk).reset() + verify(vssBackupClientLdk, never()).setup() } @Test - fun `failed automatic backup waits for new required timestamp before retrying`() = test { - whenever(clock.now()).thenReturn(Instant.fromEpochMilliseconds(3_000)) - whenever { vssBackupClient.putObject(eq(BackupCategory.SETTINGS.name), any()) } - .thenReturn(Result.failure(BackupRepoTestError("backup failed"))) - val backupStatuses = MutableStateFlow( - mapOf( - BackupCategory.SETTINGS to BackupItemStatus( - synced = 1_000, - required = 2_000, - ), - ) - ) - val allowWalletClear = CompletableDeferred().apply { complete(Unit) } - stubBackupStatuses(backupStatuses, allowWalletClear) {} + fun `restart waits for prior upload cleanup before scheduling replacement`() = test { + val entered = CompletableDeferred() + val release = CompletableDeferred() + val category = BackupCategory.SETTINGS + val statuses = MutableStateFlow(mapOf(category to BackupItemStatus(required = 1_000))) + stubBackupStatuses(statuses, CompletableDeferred(Unit)) {} stubBackupObservers() + var attempts = 0 + whenever { vssBackupClient.putObject(eq(category.name), any()) }.doSuspendableAnswer { + attempts += 1 + if (attempts == 1) { + entered.complete(Unit) + try { + awaitCancellation() + } finally { + withContext(NonCancellable) { release.await() } + } + } + Result.success(VssItem(key = category.name, value = byteArrayOf(), version = 1)) + } + sut.startObservingBackups() + advanceTimeBy(5_000) + runCurrent() + entered.await() + sut.stopObservingBackups() + sut.startObservingBackups() try { - sut.startObservingBackups() runCurrent() - advanceTimeBy(5_000) + assertTrue(statuses.value.getValue(category).running) + verify(vssBackupClient, times(1)).setupWithRetry(any(), any(), any()) + assertEquals(1, attempts) + release.complete(Unit) runCurrent() + verify(vssBackupClient, times(2)).setupWithRetry(any(), any(), any()) advanceTimeBy(5_000) runCurrent() - - verify(vssBackupClient).putObject(eq(BackupCategory.SETTINGS.name), any()) + assertEquals(2, attempts) + assertFalse(statuses.value.getValue(category).running) + assertFalse(statuses.value.getValue(category).isRequired) } finally { - sut.stopObservingBackups() + release.complete(Unit) + sut.reset() } } @Test - fun `successful upload leaves a newer required snapshot queued`() = test { - whenever(clock.now()).thenReturn(Instant.fromEpochMilliseconds(3_000)) - val backupStatuses = MutableStateFlow( + fun `settings changed during restart cleanup remain pending without another edit`() = test { + val entered = CompletableDeferred() + val release = CompletableDeferred() + val statuses = MutableStateFlow( mapOf( - BackupCategory.SETTINGS to BackupItemStatus( - synced = 1_000, - required = 2_000, - ), + BackupCategory.WIDGETS to BackupItemStatus(required = 1_000), + BackupCategory.SETTINGS to BackupItemStatus(required = 1_000, synced = 1_000), ) ) - var uploadAttempts = 0 - whenever { vssBackupClient.putObject(eq(BackupCategory.SETTINGS.name), any()) } - .doSuspendableAnswer { - uploadAttempts += 1 - if (uploadAttempts == 1) { - backupStatuses.update { statuses -> - val status = statuses.getValue(BackupCategory.SETTINGS) - statuses + (BackupCategory.SETTINGS to status.copy(required = 4_000)) - } - whenever(clock.now()).thenReturn(Instant.fromEpochMilliseconds(5_000)) - Result.success(VssItem(key = BackupCategory.SETTINGS.name, value = byteArrayOf(), version = 1)) - } else { - Result.success( - VssItem( - key = BackupCategory.SETTINGS.name, - value = byteArrayOf(), - version = 1, - ) - ) + stubBackupStatuses(statuses, CompletableDeferred(Unit)) {} + stubBackupObservers() + var widgetAttempts = 0 + whenever { vssBackupClient.putObject(eq(BackupCategory.WIDGETS.name), any()) }.doSuspendableAnswer { + widgetAttempts += 1 + if (widgetAttempts == 1) { + entered.complete(Unit) + try { + awaitCancellation() + } finally { + withContext(NonCancellable) { release.await() } } } - val allowWalletClear = CompletableDeferred().apply { complete(Unit) } - stubBackupStatuses(backupStatuses, allowWalletClear) {} - stubBackupObservers() - + Result.success(VssItem(key = BackupCategory.WIDGETS.name, value = byteArrayOf(), version = 1)) + } + sut.startObservingBackups() + advanceTimeBy(5_000) + runCurrent() + entered.await() + sut.stopObservingBackups() + sut.startObservingBackups() try { - sut.startObservingBackups() runCurrent() - advanceTimeBy(5_000) + settingsData.update { it.copy(selectedCurrency = "EUR") } + runCurrent() + verify(vssBackupClient, never()).putObject(eq(BackupCategory.SETTINGS.name), any()) + release.complete(Unit) runCurrent() advanceTimeBy(5_000) runCurrent() - - verify(vssBackupClient, times(2)).putObject(eq(BackupCategory.SETTINGS.name), any()) + val payload = argumentCaptor() + verify(vssBackupClient).putObject(eq(BackupCategory.SETTINGS.name), payload.capture()) + assertTrue(payload.firstValue.decodeToString().contains("EUR")) + assertFalse(statuses.value.getValue(BackupCategory.SETTINGS).isRequired) } finally { - sut.stopObservingBackups() + release.complete(Unit) + sut.reset() } } @Test - fun `failed automatic backup retries newer required timestamp`() = test { - whenever(clock.now()).thenReturn(Instant.fromEpochMilliseconds(3_000)) - val backupStatuses = MutableStateFlow( + fun `temporary retry waits for other category cleanup during restart`() = test { + val entered = CompletableDeferred() + val release = CompletableDeferred() + val statuses = MutableStateFlow( mapOf( - BackupCategory.SETTINGS to BackupItemStatus( - synced = 1_000, - required = 2_000, - ), + BackupCategory.WIDGETS to BackupItemStatus(required = 1_000), + BackupCategory.WALLET to BackupItemStatus(required = 1_000), ) ) - var uploadAttempts = 0 - whenever { vssBackupClient.putObject(eq(BackupCategory.SETTINGS.name), any()) } - .doSuspendableAnswer { - uploadAttempts += 1 - if (uploadAttempts == 1) { - backupStatuses.update { statuses -> - val status = statuses.getValue(BackupCategory.SETTINGS) - statuses + (BackupCategory.SETTINGS to status.copy(required = 4_000)) - } - whenever(clock.now()).thenReturn(Instant.fromEpochMilliseconds(5_000)) - Result.failure(BackupRepoTestError("backup failed")) - } else { - Result.success( - VssItem( - key = BackupCategory.SETTINGS.name, - value = byteArrayOf(), - version = 1, - ) - ) + stubBackupStatuses(statuses, CompletableDeferred(Unit)) {} + stubBackupObservers() + whenever { privatePaykitRepo.backupSnapshot() } + .thenReturn(Result.failure(PaykitException.SharedStateBusy("shared_state_busy", "locked"))) + .thenReturn(Result.success(PrivatePaykitBackupSnapshot(null, 0))) + var widgetAttempts = 0 + whenever { vssBackupClient.putObject(eq(BackupCategory.WIDGETS.name), any()) }.doSuspendableAnswer { + widgetAttempts += 1 + if (widgetAttempts == 1) { + entered.complete(Unit) + try { + awaitCancellation() + } finally { + withContext(NonCancellable) { release.await() } } } - val allowWalletClear = CompletableDeferred().apply { complete(Unit) } - stubBackupStatuses(backupStatuses, allowWalletClear) {} - stubBackupObservers() - + Result.success(VssItem(key = BackupCategory.WIDGETS.name, value = byteArrayOf(), version = 1)) + } + sut.startObservingBackups() + advanceTimeBy(5_000) + runCurrent() + entered.await() + verify(privatePaykitRepo).backupSnapshot() + sut.stopObservingBackups() + sut.startObservingBackups() try { - sut.startObservingBackups() + advanceTimeBy(20_000) runCurrent() - advanceTimeBy(5_000) + verify(privatePaykitRepo, times(1)).backupSnapshot() + verify(vssBackupClient, never()).putObject(eq(BackupCategory.WALLET.name), any()) + release.complete(Unit) runCurrent() advanceTimeBy(5_000) runCurrent() - - verify(vssBackupClient, times(2)).putObject(eq(BackupCategory.SETTINGS.name), any()) + verify(vssBackupClient).putObject(eq(BackupCategory.WALLET.name), any()) + assertFalse(statuses.value.getValue(BackupCategory.WALLET).isRequired) } finally { - sut.stopObservingBackups() + release.complete(Unit) + sut.reset() + } + } + + @Test + fun `full backup requested during observer cleanup resumes without restart`() = test { + val entered = CompletableDeferred() + val release = CompletableDeferred() + val category = BackupCategory.SETTINGS + val statuses = MutableStateFlow(mapOf(category to BackupItemStatus(required = 1_000))) + stubBackupStatuses(statuses, CompletableDeferred(Unit)) {} + stubBackupObservers() + var attempts = 0 + whenever { vssBackupClient.putObject(eq(category.name), any()) }.doSuspendableAnswer { + attempts += 1 + if (attempts == 1) { + entered.complete(Unit) + try { + awaitCancellation() + } finally { + withContext(NonCancellable) { release.await() } + } + } + Result.success(VssItem(key = category.name, value = byteArrayOf(), version = 1)) + } + + sut.startObservingBackups() + advanceTimeBy(5_000) + runCurrent() + entered.await() + sut.stopObservingBackups() + sut.scheduleFullBackup() + try { + runCurrent() + assertTrue(statuses.value.getValue(category).running) + verify(vssBackupClient, times(1)).setupWithRetry(any(), any(), any()) + assertEquals(1, attempts) + release.complete(Unit) + runCurrent() + verify(vssBackupClient, times(1)).setupWithRetry(any(), any(), any()) + advanceTimeBy(5_000) + runCurrent() + assertEquals(2, attempts) + assertFalse(statuses.value.getValue(category).running) + assertFalse(statuses.value.getValue(category).isRequired) + } finally { + release.complete(Unit) + sut.reset() + } + } + + @Test + fun `manual finalization cannot recreate status entries after reset`() = test { + val data = MutableStateFlow(AppCacheData()) + val store = mock>() + whenever(store.data).thenReturn(data) + val entered = CompletableDeferred() + val release = CompletableDeferred() + var holdNextWrite = false + whenever { store.updateData(any()) }.doSuspendableAnswer { + val transform = it.getArgument AppCacheData>(0) + if (holdNextWrite) { + holdNextWrite = false + entered.complete(Unit) + withContext(NonCancellable) { release.await() } + } + transform(data.value).also { value -> data.value = value } + } + whenever { vssBackupClient.putObject(eq(BackupCategory.SETTINGS.name), any()) }.doSuspendableAnswer { + holdNextWrite = true + Result.success(VssItem(key = BackupCategory.SETTINGS.name, value = byteArrayOf(), version = 1)) + } + val repo = createSut(backupCache = CacheStore(store)) + val manual = async { repo.triggerBackup(BackupCategory.SETTINGS) } + entered.await() + val reset = async { repo.reset() } + try { + runCurrent() + assertFalse(reset.isCompleted) + } finally { + release.complete(Unit) + reset.await() + } + data.value = AppCacheData() + assertTrue(manual.await().isFailure) + assertTrue(data.value.backupStatuses.isEmpty()) + } + + @Test + fun `reset drains manual upload so failed wipe can resume pending backups`() = test { + val entered = CompletableDeferred() + val release = CompletableDeferred() + val category = BackupCategory.SETTINGS + val statuses = MutableStateFlow(mapOf(category to BackupItemStatus(required = 1_000))) + stubBackupStatuses(statuses, CompletableDeferred(Unit)) {} + stubBackupObservers() + var attempts = 0 + whenever { vssBackupClient.putObject(eq(category.name), any()) }.doSuspendableAnswer { + attempts += 1 + if (attempts == 1) { + entered.complete(Unit) + withContext(NonCancellable) { release.await() } + return@doSuspendableAnswer Result.failure(AppError("upload interrupted")) + } + Result.success(VssItem(key = category.name, value = byteArrayOf(), version = 1)) + } + val manual = async { sut.triggerBackup(category) } + entered.await() + val reset = async { sut.reset() } + try { + runCurrent() + assertFalse(reset.isCompleted) + verify(vssBackupClient, never()).reset() + } finally { + release.complete(Unit) + reset.await() + manual.await() + } + try { + sut.startObservingBackups() + advanceTimeBy(5_000) + runCurrent() + assertEquals(2, attempts) + assertFalse(statuses.value.getValue(category).running) + assertFalse(statuses.value.getValue(category).isRequired) + } finally { + sut.reset() + } + } + + @Test + fun `failed explicit USDT callback retries backup after observation stops without acknowledging Core`() = test { + val statuses = MutableStateFlow>(emptyMap()) + stubBackupStatuses(statuses, CompletableDeferred(Unit)) {} + stubBackupObservers() + sut.startObservingBackups() + sut.stopObservingBackups() + runCurrent() + val service = UsdtService(keychain, Provider { sut }) + whenever(usdtRepo.backupSnapshot()).thenReturn("same-signed-operation") + whenever { vssBackupClient.putObject(argThat { startsWith(UsdtRecoveryBackupStore.PREFIX) }, any()) } + .thenReturn( + Result.failure( + VssException.UnknownException("error sending request for url (http://127.0.0.1:1/putObjects)") + ) + ) + .thenReturn(Result.success(VssItem(key = "recovery", value = byteArrayOf(), version = 1))) + try { + assertTrue(runCatching { service.recoveryBackup.persist("same-signed-operation") }.isFailure) + assertEquals(0L, service.recoveryGeneration) + advanceTimeBy(15_001) + runCurrent() + assertFalse(statuses.value.getValue(BackupCategory.USDT_RECOVERY).isRequired) + assertEquals(0L, service.recoveryGeneration) + service.recoveryBackup.persist("same-signed-operation") + assertEquals(1L, service.recoveryGeneration) + val keys = argumentCaptor() + val payloads = argumentCaptor() + verify(vssBackupClient, times(3)).putObject(keys.capture(), payloads.capture()) + assertEquals(1, keys.allValues.distinct().size) + assertTrue(payloads.allValues.all { it.contentEquals(payloads.firstValue) }) + verify(privatePaykitRepo, never()).backupSnapshot() + } finally { + sut.reset() + } + } + + @Test + fun `manual backup failure retries without observation`() = test { + val category = BackupCategory.SETTINGS + val statuses = MutableStateFlow>(emptyMap()) + stubBackupStatuses(statuses, CompletableDeferred(Unit)) {} + whenever { vssBackupClient.putObject(eq(category.name), any()) } + .thenReturn(Result.failure(VssException.NetworkException("offline"))) + .thenReturn(Result.success(VssItem(key = category.name, value = byteArrayOf(), version = 1))) + try { + assertTrue(sut.triggerBackup(category).isFailure) + advanceTimeBy(15_001) + runCurrent() + verify(vssBackupClient, times(2)).putObject(eq(category.name), any()) + assertFalse(statuses.value.getValue(category).isRequired) + } finally { + sut.reset() + } + } + + @Test + fun `reset and wipe revoke explicit retry admission before client replacement`() = test { + val category = BackupCategory.SETTINGS + val statuses = MutableStateFlow>(emptyMap()) + stubBackupStatuses(statuses, CompletableDeferred(Unit)) {} + whenever { vssBackupClient.putObject(eq(category.name), any()) } + .thenReturn(Result.failure(VssException.NetworkException("offline"))) + .thenReturn(Result.success(VssItem(key = category.name, value = byteArrayOf(), version = 1))) + assertTrue(sut.triggerBackup(category).isFailure) + sut.setWiping(true) + sut.reset() + advanceTimeBy(20_000) + runCurrent() + verify(vssBackupClient, times(1)).putObject(eq(category.name), any()) + assertTrue(sut.triggerBackup(category).isFailure) + sut.setWiping(false) + sut.triggerBackup(category).getOrThrow() + advanceTimeBy(20_000) + runCurrent() + verify(vssBackupClient, times(2)).putObject(eq(category.name), any()) + assertFalse(statuses.value.getValue(category).isRequired) + sut.reset() + } + + @Test + fun `explicit retry retains newer required data until the successor succeeds`() = test { + val category = BackupCategory.SETTINGS + val statuses = MutableStateFlow>(emptyMap()) + stubBackupStatuses(statuses, CompletableDeferred(Unit)) {} + var attempts = 0 + whenever { vssBackupClient.putObject(eq(category.name), any()) }.doSuspendableAnswer { + attempts += 1 + if (attempts == 1) return@doSuspendableAnswer Result.failure(VssException.NetworkException("offline")) + if (attempts == 2) { + statuses.update { data -> + val current = data.getValue(category) + data + (category to current.copy(required = current.required + 1)) + } + } + Result.success(VssItem(key = category.name, value = byteArrayOf(), version = 1)) + } + try { + assertTrue(sut.triggerBackup(category).isFailure) + advanceTimeBy(20_001) + runCurrent() + assertEquals(3, attempts) + assertFalse(statuses.value.getValue(category).isRequired) + } finally { + sut.reset() + } + } + + @Test + fun `full request survives coalescing with a cancelled observer upload`() = test { + val category = BackupCategory.SETTINGS + val statuses = MutableStateFlow(mapOf(category to BackupItemStatus(required = 1_000))) + stubBackupStatuses(statuses, CompletableDeferred(Unit)) {} + stubBackupObservers() + val entered = CompletableDeferred() + val release = CompletableDeferred() + var attempts = 0 + whenever { vssBackupClient.putObject(eq(category.name), any()) }.doSuspendableAnswer { + attempts += 1 + if (attempts == 1) { + entered.complete(Unit) + try { + awaitCancellation() + } finally { + withContext(NonCancellable) { release.await() } + } + } + Result.success(VssItem(key = category.name, value = byteArrayOf(), version = 1)) + } + sut.startObservingBackups() + advanceTimeBy(5_000) + runCurrent() + entered.await() + sut.scheduleFullBackup() + sut.stopObservingBackups() + try { + advanceTimeBy(20_000) + runCurrent() + assertEquals(1, attempts) + release.complete(Unit) + runCurrent() + advanceTimeBy(5_000) + runCurrent() + assertEquals(2, attempts) + assertFalse(statuses.value.getValue(category).isRequired) + } finally { + release.complete(Unit) + sut.reset() + } + } + + @Test + fun `permanent failure and caller cancellation do not admit explicit retry`() = test { + val statuses = MutableStateFlow>(emptyMap()) + stubBackupStatuses(statuses, CompletableDeferred(Unit)) {} + val entered = CompletableDeferred() + var attempts = 0 + whenever { vssBackupClient.putObject(argThat { startsWith(UsdtRecoveryBackupStore.PREFIX) }, any()) } + .doSuspendableAnswer { + attempts += 1 + if (attempts == 1) { + Result.failure(VssException.UnknownException("invalid credentials")) + } else { + entered.complete(Unit) + awaitCancellation() + } + } + assertTrue(sut.persistUsdtRecoveryBackup("signed").isFailure) + advanceTimeBy(600_000) + runCurrent() + assertEquals(1, attempts) + val caller = launch { sut.persistUsdtRecoveryBackup("signed") } + entered.await() + caller.cancelAndJoin() + advanceTimeBy(600_000) + runCurrent() + assertEquals(2, attempts) + assertTrue(statuses.value.getValue(BackupCategory.USDT_RECOVERY).isRequired) + sut.reset() + } + + @Test + fun `manual success follows a consumed change after observers stop`() = test { + val category = BackupCategory.SETTINGS + val statuses = MutableStateFlow(mapOf(category to BackupItemStatus(required = 1_000, synced = 1_000))) + stubBackupStatuses(statuses, CompletableDeferred(Unit)) {} + stubBackupObservers() + sut.startObservingBackups() + runCurrent() + val entered = CompletableDeferred() + val release = CompletableDeferred() + var attempts = 0 + whenever { vssBackupClient.putObject(eq(category.name), any()) }.doSuspendableAnswer { + attempts += 1 + if (attempts == 1) { + entered.complete(Unit) + release.await() + } + Result.success(VssItem(key = category.name, value = byteArrayOf(), version = 1)) + } + val manual = async { sut.triggerBackup(category) } + entered.await() + settingsData.update { it.copy(selectedCurrency = "EUR") } + runCurrent() + sut.stopObservingBackups() + try { + release.complete(Unit) + manual.await().getOrThrow() + advanceTimeBy(5_001) + runCurrent() + assertEquals(2, attempts) + assertFalse(statuses.value.getValue(category).isRequired) + val payloads = argumentCaptor() + verify(vssBackupClient, times(2)).putObject(eq(category.name), payloads.capture()) + assertTrue(payloads.secondValue.decodeToString().contains("EUR")) + } finally { + release.complete(Unit) + sut.reset() + } + } + + @Test + fun `consumed change cannot orphan a retained explicit retry timer`() = test { + val category = BackupCategory.SETTINGS + val statuses = MutableStateFlow(mapOf(category to BackupItemStatus(required = 1_000, synced = 1_000))) + stubBackupStatuses(statuses, CompletableDeferred(Unit)) {} + stubBackupObservers() + sut.startObservingBackups() + runCurrent() + whenever { vssBackupClient.putObject(eq(category.name), any()) } + .thenReturn(Result.failure(VssException.NetworkException("offline"))) + .thenReturn(Result.success(VssItem(key = category.name, value = byteArrayOf(), version = 1))) + assertTrue(sut.triggerBackup(category).isFailure) + val entered = CompletableDeferred() + val release = CompletableDeferred() + var holdNextWrite = true + whenever { cacheStore.update(any()) }.doSuspendableAnswer { + val transform = it.getArgument<(AppCacheData) -> AppCacheData>(0) + if (holdNextWrite) { + holdNextWrite = false + entered.complete(Unit) + release.await() + } + statuses.update { current -> transform(AppCacheData(backupStatuses = current)).backupStatuses } + } + settingsData.update { it.copy(selectedCurrency = "EUR") } + entered.await() + sut.stopObservingBackups() + try { + release.complete(Unit) + runCurrent() + advanceTimeBy(15_001) + runCurrent() + verify(vssBackupClient, times(2)).putObject(eq(category.name), any()) + assertFalse(statuses.value.getValue(category).isRequired) + } finally { + release.complete(Unit) + sut.reset() + } + } + + @Test + fun `older retry timer cannot shorten a newer explicit failure backoff`() = test { + val category = BackupCategory.SETTINGS + val statuses = MutableStateFlow>(emptyMap()) + stubBackupStatuses(statuses, CompletableDeferred(Unit)) {} + var attempts = 0 + whenever { vssBackupClient.putObject(eq(category.name), any()) }.doSuspendableAnswer { + attempts += 1 + if (attempts < 3) { + Result.failure(VssException.NetworkException("offline")) + } else { + Result.success(VssItem(key = category.name, value = byteArrayOf(), version = 1)) + } + } + try { + assertTrue(sut.triggerBackup(category).isFailure) + advanceTimeBy(5_000) + assertTrue(sut.triggerBackup(category).isFailure) + advanceTimeBy(15_000) + runCurrent() + assertEquals(2, attempts) + advanceTimeBy(10_001) + runCurrent() + assertEquals(3, attempts) + assertFalse(statuses.value.getValue(category).isRequired) + } finally { + sut.reset() + } + } + + @Test + fun `stale timer continuation respects a concurrent explicit failure backoff`() = test { + val category = BackupCategory.SETTINGS + val statuses = MutableStateFlow>(emptyMap()) + stubBackupStatuses(statuses, CompletableDeferred(Unit)) {} + var attempts = 0 + whenever { vssBackupClient.putObject(eq(category.name), any()) }.doSuspendableAnswer { + attempts += 1 + if (attempts < 3) { + Result.failure(VssException.NetworkException("offline")) + } else { + Result.success(VssItem(key = category.name, value = byteArrayOf(), version = 1)) + } + } + assertTrue(sut.triggerBackup(category).isFailure) + val entered = CompletableDeferred() + val release = CompletableDeferred() + var holdNextReturn = true + whenever { cacheStore.update(any()) }.doSuspendableAnswer { + val transform = it.getArgument<(AppCacheData) -> AppCacheData>(0) + statuses.update { current -> transform(AppCacheData(backupStatuses = current)).backupStatuses } + if (holdNextReturn) { + holdNextReturn = false + entered.complete(Unit) + release.await() + } + } + advanceTimeBy(10_000) + runCurrent() + entered.await() + try { + assertTrue(sut.triggerBackup(category).isFailure) + release.complete(Unit) + runCurrent() + advanceTimeBy(5_001) + runCurrent() + assertEquals(2, attempts) + advanceTimeBy(20_000) + runCurrent() + assertEquals(3, attempts) + assertFalse(statuses.value.getValue(category).isRequired) + } finally { + release.complete(Unit) + sut.reset() + } + } + + @Test + fun `cancelled queued USDT callback retains its required revision`() = test { + val category = BackupCategory.USDT_RECOVERY + val statuses = MutableStateFlow>(emptyMap()) + stubBackupStatuses(statuses, CompletableDeferred(Unit)) {} + val release = CompletableDeferred() + val uploads = mutableListOf() + val service = UsdtService(keychain, Provider { sut }) + whenever(usdtRepo.backupSnapshot()).thenReturn("older") + whenever { vssBackupClient.putObject(argThat { startsWith(UsdtRecoveryBackupStore.PREFIX) }, any()) } + .doSuspendableAnswer { + val payload = json.decodeFromString(it.getArgument(1).decodeToString()) + uploads += payload.usdtWallet + release.await() + Result.success(VssItem(key = "recovery", value = byteArrayOf(), version = 1)) + } + val older = async { sut.triggerBackup(category) } + runCurrent() + val olderRequired = statuses.value.getValue(category).required + val queued = launch { service.recoveryBackup.persist("newer") } + try { + runCurrent() + queued.cancelAndJoin() + release.complete(Unit) + older.await().getOrThrow() + runCurrent() + val status = statuses.value.getValue(category) + assertTrue(status.required > olderRequired) + assertTrue(status.isRequired) + assertEquals(0L, service.recoveryGeneration) + assertEquals(listOf("older"), uploads) + } finally { + queued.cancelAndJoin() + release.complete(Unit) + sut.reset() + } + } + + @Test + fun `queued USDT callback cannot satisfy a later cancelled revision`() = test { + val category = BackupCategory.USDT_RECOVERY + val statuses = MutableStateFlow>(emptyMap()) + stubBackupStatuses(statuses, CompletableDeferred(Unit)) {} + val release = CompletableDeferred() + val uploads = mutableListOf() + whenever(usdtRepo.backupSnapshot()).thenReturn("older") + whenever { vssBackupClient.putObject(argThat { startsWith(UsdtRecoveryBackupStore.PREFIX) }, any()) } + .doSuspendableAnswer { + val payload = json.decodeFromString(it.getArgument(1).decodeToString()) + uploads += payload.usdtWallet + if (payload.usdtWallet == "older") release.await() + Result.success(VssItem(key = "recovery", value = byteArrayOf(), version = 1)) + } + val older = async { sut.triggerBackup(category) } + runCurrent() + val first = async { sut.persistUsdtRecoveryBackup("first") } + runCurrent() + val second = launch { sut.persistUsdtRecoveryBackup("second") } + try { + runCurrent() + second.cancelAndJoin() + release.complete(Unit) + older.await().getOrThrow() + first.await().getOrThrow() + runCurrent() + assertTrue(statuses.value.getValue(category).isRequired) + assertEquals(listOf("older", "first"), uploads) + } finally { + second.cancelAndJoin() + release.complete(Unit) + sut.reset() + } + } + + @Test + fun `older reserved callback preserves a newer failure backoff`() = test { + val category = BackupCategory.USDT_RECOVERY + val statuses = MutableStateFlow>(emptyMap()) + stubBackupStatuses(statuses, CompletableDeferred(Unit)) {} + whenever(usdtRepo.backupSnapshot()).thenReturn("newer") + val uploads = mutableListOf() + whenever { vssBackupClient.putObject(argThat { startsWith(UsdtRecoveryBackupStore.PREFIX) }, any()) } + .doSuspendableAnswer { + val payload = json.decodeFromString(it.getArgument(1).decodeToString()) + uploads += payload.usdtWallet + if (uploads.size == 1) { + Result.failure(VssException.NetworkException("offline")) + } else { + Result.success(VssItem(key = "recovery", value = byteArrayOf(), version = 1)) + } + } + val entered = CompletableDeferred() + val release = CompletableDeferred() + var holdNextReturn = true + whenever { cacheStore.update(any()) }.doSuspendableAnswer { + val transform = it.getArgument<(AppCacheData) -> AppCacheData>(0) + statuses.update { current -> transform(AppCacheData(backupStatuses = current)).backupStatuses } + if (holdNextReturn) { + holdNextReturn = false + entered.complete(Unit) + release.await() + } + } + val older = async { sut.persistUsdtRecoveryBackup("older") } + entered.await() + try { + assertTrue(sut.persistUsdtRecoveryBackup("newer").isFailure) + advanceTimeBy(5_000) + release.complete(Unit) + older.await().getOrThrow() + runCurrent() + advanceTimeBy(5_001) + runCurrent() + assertEquals(listOf("newer", "older"), uploads) + assertTrue(statuses.value.getValue(category).isRequired) + advanceTimeBy(5_000) + runCurrent() + assertEquals(listOf("newer", "older", "newer"), uploads) + assertFalse(statuses.value.getValue(category).isRequired) + } finally { + release.complete(Unit) + sut.reset() + } + } + + @Test + fun `USDT callback does not acknowledge reset cancellation and retries the same recovery record`() = test { + val entered = CompletableDeferred() + val release = CompletableDeferred() + val service = UsdtService(keychain, Provider { sut }) + var attempts = 0 + whenever { vssBackupClient.putObject(argThat { startsWith(UsdtRecoveryBackupStore.PREFIX) }, any()) } + .doSuspendableAnswer { + attempts += 1 + if (attempts == 1) { + entered.complete(Unit) + withContext(NonCancellable) { release.await() } + } + Result.success(VssItem(key = "recovery", value = byteArrayOf(), version = 1)) + } + val pending = async { runCatching { service.recoveryBackup.persist("same-signed-operation") } } + entered.await() + val reset = async { sut.reset() } + try { + runCurrent() + assertFalse(reset.isCompleted) + } finally { + release.complete(Unit) + reset.await() + } + assertTrue(pending.await().isFailure) + assertEquals(0L, service.recoveryGeneration) + service.recoveryBackup.persist("same-signed-operation") + assertEquals(1L, service.recoveryGeneration) + val keys = argumentCaptor() + val payloads = argumentCaptor() + verify(vssBackupClient, times(2)).putObject(keys.capture(), payloads.capture()) + assertEquals(keys.firstValue, keys.secondValue) + assertTrue(payloads.firstValue.contentEquals(payloads.secondValue)) + sut.reset() + } + + @Test + fun `explicit backups are rejected throughout wallet wipe`() = test { + sut.setWiping(true) + sut.reset() + assertTrue(sut.triggerBackup(BackupCategory.SETTINGS).isFailure) + assertTrue(sut.persistWalletBackup().isFailure) + assertTrue(sut.persistUsdtRecoveryBackup("signed").isFailure) + sut.scheduleFullBackup() + verify(vssBackupClient, never()).setup(any()) + verify(vssBackupClient, never()).putObject(any(), any()) + verify(cacheStore, never()).update(any()) + sut.setWiping(false) + sut.triggerBackup(BackupCategory.SETTINGS).getOrThrow() + verify(vssBackupClient).putObject(eq(BackupCategory.SETTINGS.name), any()) + } + + @Test + fun `explicit USDT recovery backup works after observers stop`() = test { + stubBackupObservers() + stubBackupStatuses(MutableStateFlow(emptyMap()), CompletableDeferred(Unit)) {} + sut.startObservingBackups() + sut.stopObservingBackups() + whenever(privatePaykitRepo.backupSnapshot()).thenReturn( + Result.failure(PaykitException.SharedStateBusy("shared_state_busy", "lock unavailable")), + ) + sut.persistUsdtRecoveryBackup("same-signed-operation").getOrThrow() + verify(vssBackupClient).putObject(argThat { startsWith(UsdtRecoveryBackupStore.PREFIX) }, any()) + verify(privatePaykitRepo, never()).backupSnapshot() + sut.reset() + } + + @Test + fun `wipe admission drains a backup queued at the real SDK operation barrier`() = test { + val sdkLock = PaykitSdkOperationLock() + val held = CompletableDeferred() + val release = CompletableDeferred() + val exportQueued = CompletableDeferred() + val blocker = async { + sdkLock.withLock { + held.complete(Unit) + release.await() + } + } + held.await() + val statuses = MutableStateFlow(mapOf(BackupCategory.WALLET to BackupItemStatus(required = 1_000))) + stubBackupStatuses(statuses, CompletableDeferred(Unit)) {} + stubBackupObservers() + var exported = false + whenever { privatePaykitRepo.backupSnapshot() }.doSuspendableAnswer { + exportQueued.complete(Unit) + sdkLock.withLock(PaykitSdkOperationLock.Priority.Background) { + exported = true + Result.success(PrivatePaykitBackupSnapshot(null, 0)) + } + } + sut.startObservingBackups() + advanceTimeBy(5_000) + runCurrent() + exportQueued.await() + val wipe = async { + sdkLock.withWalletWipe { + sut.setWiping(true) + sut.reset() + } + } + runCurrent() + try { + release.complete(Unit) + withTimeout(5.seconds) { wipe.await() } + blocker.await() + assertFalse(exported) + verify(vssBackupClient, never()).putObject(eq(BackupCategory.WALLET.name), any()) + verify(vssBackupClient).reset() + } finally { + release.complete(Unit) + sut.reset() + } + } + + @Test + fun `concurrent lifecycle and scheduling drains with a real cache store`() = test { + stubBackupObservers() + val directory = Files.createTempDirectory("backup-lifecycle-test").toFile() + val dispatcher = Executors.newFixedThreadPool(2).asCoroutineDispatcher() + val dataJob = SupervisorJob() + val store = DataStoreFactory.create( + serializer = AppCacheSerializer, + scope = CoroutineScope(dispatcher + dataJob), + produceFile = { directory.resolve("cache.json") }, + ) + val repo = createSut(backupCache = CacheStore(store), dispatcher = dispatcher) + val setupCalls = AtomicInteger() + whenever { vssBackupClient.setupWithRetry(any(), any(), any()) }.doSuspendableAnswer { + setupCalls.incrementAndGet() + Result.success(Unit) + } + try { + withContext(dispatcher) { + withTimeout(10.seconds) { + val start = CompletableDeferred() + val tasks = listOf( + async { + start.await() + repeat(25) { + repo.startObservingBackups() + repo.scheduleFullBackup() + yield() + } + }, + async { + start.await() + repeat(25) { + repo.stopObservingBackups() + yield() + } + }, + ) + start.complete(Unit) + tasks.awaitAll() + repo.reset() + val beforeRestart = setupCalls.get() + val restarted = CompletableDeferred() + whenever { vssBackupClient.setupWithRetry(any(), any(), any()) }.doSuspendableAnswer { + setupCalls.incrementAndGet() + restarted.complete(Unit) + Result.success(Unit) + } + repeat(25) { repo.startObservingBackups() } + restarted.await() + repo.reset() + assertEquals(beforeRestart + 1, setupCalls.get()) + } + } + } finally { + repo.reset() + dataJob.cancelAndJoin() + dispatcher.close() + directory.deleteRecursively() + } + } + + @Test + fun `busy wallet backup retries after observation pauses and resumes`() = test { + whenever(clock.now()).thenReturn(Instant.fromEpochMilliseconds(3_000)) + whenever { privatePaykitRepo.backupSnapshot() } + .thenReturn(Result.failure(PaykitException.SharedStateBusy("shared_state_busy", "locked"))) + .thenReturn(Result.success(PrivatePaykitBackupSnapshot(null, 0))) + val backupStatuses = MutableStateFlow( + mapOf(BackupCategory.WALLET to BackupItemStatus(synced = 1_000, required = 2_000)) + ) + val allowWalletClear = CompletableDeferred().apply { complete(Unit) } + stubBackupStatuses(backupStatuses, allowWalletClear) {} + stubBackupObservers() + + try { + sut.startObservingBackups() + runCurrent() + advanceTimeBy(5_000) + runCurrent() + sut.stopObservingBackups() + advanceTimeBy(20_000) + runCurrent() + verify(vssBackupClient, never()).putObject(eq(BackupCategory.WALLET.name), any()) + + sut.startObservingBackups() + runCurrent() + advanceTimeBy(5_000) + runCurrent() + + verify(vssBackupClient).putObject(eq(BackupCategory.WALLET.name), any()) + } finally { + sut.stopObservingBackups() + } + } + + @Test + fun `busy wallet backup still warns once it stays pending past the failure window`() = test { + val startedAt = currentTime + whenever(clock.now()).thenAnswer { Instant.fromEpochMilliseconds(3_000 + currentTime - startedAt) } + whenever { privatePaykitRepo.backupSnapshot() } + .thenReturn(Result.failure(PaykitException.SharedStateBusy("shared_state_busy", "locked"))) + whenever(context.getString(R.string.settings__backup__failed_title)).thenReturn("Backup failed") + whenever(context.getString(R.string.settings__backup__failed_message)).thenReturn("Retrying") + val backupStatuses = MutableStateFlow( + mapOf(BackupCategory.WALLET to BackupItemStatus(synced = 1_000, required = 2_000)) + ) + val allowWalletClear = CompletableDeferred().apply { complete(Unit) } + stubBackupStatuses(backupStatuses, allowWalletClear) {} + stubBackupObservers() + + try { + sut.startObservingBackups() + runCurrent() + advanceTimeBy(31.minutes.inWholeMilliseconds) + runCurrent() + + verify(context, atLeastOnce()).getString(R.string.settings__backup__failed_title) + } finally { + sut.stopObservingBackups() + } + } + + @Test + fun `start observing backs up stale required status after clearing running flag`() = test { + val backupStatuses = MutableStateFlow( + mapOf( + BackupCategory.WALLET to BackupItemStatus( + running = true, + synced = 1_000, + required = 2_000, + ), + ) + ) + val allowWalletClear = CompletableDeferred() + var delayedWalletClear = false + stubBackupStatuses(backupStatuses, allowWalletClear) { + delayedWalletClear = true + } + stubBackupObservers() + + try { + sut.startObservingBackups() + runCurrent() + verify(vssBackupClient, never()).putObject(eq(BackupCategory.WALLET.name), any()) + + allowWalletClear.complete(Unit) + runCurrent() + advanceTimeBy(5_000) + runCurrent() + + assertTrue(delayedWalletClear) + verify(vssBackupClient).putObject(eq(BackupCategory.WALLET.name), any()) + } finally { + sut.stopObservingBackups() + } + } + + @Test + fun `failed automatic backup waits for new required timestamp before retrying`() = test { + whenever(clock.now()).thenReturn(Instant.fromEpochMilliseconds(3_000)) + whenever { vssBackupClient.putObject(eq(BackupCategory.SETTINGS.name), any()) } + .thenReturn(Result.failure(BackupRepoTestError("backup failed"))) + val backupStatuses = MutableStateFlow( + mapOf( + BackupCategory.SETTINGS to BackupItemStatus( + synced = 1_000, + required = 2_000, + ), + ) + ) + val allowWalletClear = CompletableDeferred().apply { complete(Unit) } + stubBackupStatuses(backupStatuses, allowWalletClear) {} + stubBackupObservers() + + try { + sut.startObservingBackups() + runCurrent() + advanceTimeBy(5_000) + runCurrent() + advanceTimeBy(5_000) + runCurrent() + + verify(vssBackupClient).putObject(eq(BackupCategory.SETTINGS.name), any()) + } finally { + sut.stopObservingBackups() + } + } + + @Test + fun `successful upload leaves a newer required snapshot queued`() = test { + whenever(clock.now()).thenReturn(Instant.fromEpochMilliseconds(3_000)) + val backupStatuses = MutableStateFlow( + mapOf( + BackupCategory.SETTINGS to BackupItemStatus( + synced = 1_000, + required = 2_000, + ), + ) + ) + var uploadAttempts = 0 + whenever { vssBackupClient.putObject(eq(BackupCategory.SETTINGS.name), any()) } + .doSuspendableAnswer { + uploadAttempts += 1 + if (uploadAttempts == 1) { + backupStatuses.update { statuses -> + val status = statuses.getValue(BackupCategory.SETTINGS) + statuses + (BackupCategory.SETTINGS to status.copy(required = 4_000)) + } + whenever(clock.now()).thenReturn(Instant.fromEpochMilliseconds(5_000)) + Result.success(VssItem(key = BackupCategory.SETTINGS.name, value = byteArrayOf(), version = 1)) + } else { + Result.success( + VssItem( + key = BackupCategory.SETTINGS.name, + value = byteArrayOf(), + version = 1, + ) + ) + } + } + val allowWalletClear = CompletableDeferred().apply { complete(Unit) } + stubBackupStatuses(backupStatuses, allowWalletClear) {} + stubBackupObservers() + + try { + sut.startObservingBackups() + runCurrent() + advanceTimeBy(5_000) + runCurrent() + advanceTimeBy(5_000) + runCurrent() + + verify(vssBackupClient, times(2)).putObject(eq(BackupCategory.SETTINGS.name), any()) + } finally { + sut.stopObservingBackups() + } + } + + @Test + fun `full backup follows newer required data after observers coalesce`() = test { + whenever(clock.now()).thenReturn(Instant.fromEpochMilliseconds(3_000)) + val backupStatuses = MutableStateFlow( + mapOf( + BackupCategory.SETTINGS to BackupItemStatus( + synced = 1_000, + required = 2_000, + ), + ) + ) + var uploadAttempts = 0 + whenever { vssBackupClient.putObject(eq(BackupCategory.SETTINGS.name), any()) } + .doSuspendableAnswer { + uploadAttempts += 1 + if (uploadAttempts == 1) { + backupStatuses.update { statuses -> + val status = statuses.getValue(BackupCategory.SETTINGS) + statuses + (BackupCategory.SETTINGS to status.copy(required = 4_000)) + } + whenever(clock.now()).thenReturn(Instant.fromEpochMilliseconds(5_000)) + Result.success(VssItem(key = BackupCategory.SETTINGS.name, value = byteArrayOf(), version = 1)) + } else { + Result.success( + VssItem( + key = BackupCategory.SETTINGS.name, + value = byteArrayOf(), + version = 1, + ) + ) + } + } + val allowWalletClear = CompletableDeferred().apply { complete(Unit) } + stubBackupStatuses(backupStatuses, allowWalletClear) {} + stubBackupObservers() + + try { + sut.scheduleFullBackup() + sut.startObservingBackups() + runCurrent() + advanceTimeBy(5_000) + runCurrent() + advanceTimeBy(5_000) + runCurrent() + + verify(vssBackupClient, times(2)).putObject(eq(BackupCategory.SETTINGS.name), any()) + } finally { + sut.stopObservingBackups() + } + } + + @Test + fun `failed automatic backup retries newer required timestamp`() = test { + whenever(clock.now()).thenReturn(Instant.fromEpochMilliseconds(3_000)) + val backupStatuses = MutableStateFlow( + mapOf( + BackupCategory.SETTINGS to BackupItemStatus( + synced = 1_000, + required = 2_000, + ), + ) + ) + var uploadAttempts = 0 + whenever { vssBackupClient.putObject(eq(BackupCategory.SETTINGS.name), any()) } + .doSuspendableAnswer { + uploadAttempts += 1 + if (uploadAttempts == 1) { + backupStatuses.update { statuses -> + val status = statuses.getValue(BackupCategory.SETTINGS) + statuses + (BackupCategory.SETTINGS to status.copy(required = 4_000)) + } + whenever(clock.now()).thenReturn(Instant.fromEpochMilliseconds(5_000)) + Result.failure(BackupRepoTestError("backup failed")) + } else { + Result.success( + VssItem( + key = BackupCategory.SETTINGS.name, + value = byteArrayOf(), + version = 1, + ) + ) + } + } + val allowWalletClear = CompletableDeferred().apply { complete(Unit) } + stubBackupStatuses(backupStatuses, allowWalletClear) {} + stubBackupObservers() + + try { + sut.startObservingBackups() + runCurrent() + advanceTimeBy(5_000) + runCurrent() + advanceTimeBy(5_000) + runCurrent() + + verify(vssBackupClient, times(2)).putObject(eq(BackupCategory.SETTINGS.name), any()) + } finally { + sut.stopObservingBackups() } } @@ -1670,12 +2839,9 @@ class BackupRepoTest : BaseUnitTest() { fun `metadata restore preserves the Blocktank refund address pointer`() = test { val refundAddress = BlocktankRefundAddress(address = "bcrt1qrefund", index = 7) stubMetadataRestore(envelope = metadataEnvelope(cache = AppCacheData(blocktankRefundAddress = refundAddress))) - val cacheTransform = argumentCaptor<(AppCacheData) -> AppCacheData>() - sut.performFullRestoreFromLatestBackup() - verifyBlocking(cacheStore) { update(cacheTransform.capture()) } - assertEquals(refundAddress, cacheTransform.firstValue(AppCacheData()).blocktankRefundAddress) + assertEquals(refundAddress, cacheData.value.blocktankRefundAddress) } @Test @@ -2182,17 +3348,18 @@ class BackupRepoTest : BaseUnitTest() { onWalletClearDelayed: () -> Unit, ) { whenever(cacheStore.backupStatuses).thenReturn(backupStatuses) - whenever { cacheStore.updateBackupStatus(any(), any()) }.doSuspendableAnswer { - val category = it.getArgument(0) - val transform = it.getArgument<(BackupItemStatus) -> BackupItemStatus>(1) - if (category == BackupCategory.WALLET && !allowWalletClear.isCompleted) { - onWalletClearDelayed() - allowWalletClear.await() - } - backupStatuses.update { statuses -> - val currentStatus = statuses[category] ?: BackupItemStatus() - statuses + (category to transform(currentStatus)) + whenever { cacheStore.update(any()) }.doSuspendableAnswer { + val transform = it.getArgument<(AppCacheData) -> AppCacheData>(0) + if (!allowWalletClear.isCompleted) { + val current = AppCacheData(backupStatuses = backupStatuses.value) + if (transform(current).backupStatuses[BackupCategory.WALLET] != + current.backupStatuses[BackupCategory.WALLET] + ) { + onWalletClearDelayed() + allowWalletClear.await() + } } + backupStatuses.update { statuses -> transform(AppCacheData(backupStatuses = statuses)).backupStatuses } } } @@ -2227,10 +3394,13 @@ class BackupRepoTest : BaseUnitTest() { verify(vssBackupClient, never()).getObject(any()) } - private fun createSut() = BackupRepo( + private fun createSut( + backupCache: CacheStore = cacheStore, + dispatcher: CoroutineDispatcher = testDispatcher, + ) = BackupRepo( context = context, - ioDispatcher = testDispatcher, - cacheStore = cacheStore, + ioDispatcher = dispatcher, + cacheStore = backupCache, vssBackupClient = vssBackupClient, vssBackupClientLdk = vssBackupClientLdk, settingsStore = settingsStore, diff --git a/changelog.d/next/1689.fixed.md b/changelog.d/next/1689.fixed.md new file mode 100644 index 0000000000..b2ac2cf3ec --- /dev/null +++ b/changelog.d/next/1689.fixed.md @@ -0,0 +1 @@ +Prevent interrupted backups from crossing wallet resets and keep explicit backup retries running through temporary connection failures even when observation stops. diff --git a/journeys/backup-restore/wallet-snapshot-consistency.md b/journeys/backup-restore/wallet-snapshot-consistency.md index 6b58ef5e31..4dccf9d071 100644 --- a/journeys/backup-restore/wallet-snapshot-consistency.md +++ b/journeys/backup-restore/wallet-snapshot-consistency.md @@ -25,3 +25,43 @@ funded personal wallet or shorten payment safety waits. Record the tested commit, gate locations, attempts and upload counts, and the restored state. Unit coverage does not establish a device pass or a maximum backup delay during sustained activity. + +## Backup observation shutdown + +The following controls require the same dedicated lifecycle/capture fixture; +these are not ordinary UI journey capabilities. Use isolated fixture accounts; +do not reset a funded wallet. + +1. Hold a scheduled backup in its cancellable debounce, stop observation, and + request a manual metadata backup while cancellation cleanup remains held. + Verify a same-category manual upload and observation restart wait for cleanup. + Change synchronized settings after requesting restart while cleanup is still + held. Release cleanup; verify those settings and pending wallet work upload + without another edit, and no upload or client setup ran before the drain. + Repeat with a retained temporary-failure retry for another category becoming + due during the wait; that automatic retry must wait too. +2. Request a new USDT recovery backup during cleanup and again after observation + is stopped. Verify both can finish without restarting Lightning, and transient + backup failures retain bounded retry without repeating the payment. +3. Repeatedly start and immediately stop observation. Change a wallet record only + after the stopped listeners finish; verify no new automatic upload starts. + Restart observation and verify a single set of listeners resumes pending work. +4. Hold setup, upload or final status persistence, then request fixture wallet + reset. Verify both VSS clients remain unchanged until old work finishes, even + if the reset caller is canceled. Once reset returns, simulate failed node stop + and resume observation without clearing the cache; pending work must recover. +5. Keep wallet wipe admission closed after reset returns. Request manual retry, + full backup and USDT persistence; all must avoid new uploads/status changes. + After wipe finishes, new explicit backups work with observation stopped. +6. Change required data during an explicit full backup while observation runs. + Verify the new revision uploads after the first job completes without another + mutation. + +Android stop invalidates observation immediately; reset awaits all retained work, +and restart awaits prior observer cleanup. iOS awaits its asynchronous shutdown +boundary. These platform differences do not make observation a prerequisite for +explicit USDT persistence. Component tests cover controlled cancellation, the real +SDK operation barrier, and concurrent calls with a real DataStore. They do not +establish a complete device fault journey or a deployed Paykit/VSS outage pass. + +For the consumption-to-status interval, begin with synchronized settings, change the currency, and hold the required-status cache write. Stop observation or request a fixture reset that leaves the data intact, then release the write and restart without another edit. The changed settings must upload once and become synchronized. Reset must drain the write before a later cache clear; an acknowledged upload need not repeat.