diff --git a/app/src/main/java/to/bitkit/repositories/BackupRepo.kt b/app/src/main/java/to/bitkit/repositories/BackupRepo.kt index a9d095a1dd..cc37493627 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 @@ -82,10 +89,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 @@ -147,19 +156,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() @@ -168,10 +191,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 } @@ -213,16 +261,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) @@ -237,15 +298,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 { @@ -256,39 +318,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 -> @@ -298,144 +343,127 @@ 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) + + 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.USDT_RECOVERY, - ) + observeBackupChanges( + owner, + usdtRepo.get().state.map { it.transfers }.distinctUntilChanged(), + BackupCategory.WALLET, ) - dataListenerJobs.add( - observeBackupChanges( - usdtRepo.get().state.map { it.transfers }.distinctUntilChanged(), - BackupCategory.WALLET, - ) - ) - dataListenerJobs.add(observeBackupChanges(paykitPresentationStore.backupStateVersion, BackupCategory.WALLET)) - dataListenerJobs.add(observeBackupChanges(paykitAllowanceStore.backupStateVersion, BackupCategory.WALLET)) - dataListenerJobs.add( - observeBackupChanges( - privatePaykitAddressReservationRepo.get().backupStateVersion, - BackupCategory.WALLET, - ) + observeBackupChanges(owner, paykitPresentationStore.backupStateVersion, BackupCategory.WALLET) + observeBackupChanges(owner, paykitAllowanceStore.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 { @@ -443,87 +471,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() + // Coalescing pending work must not delay an explicit recovery acknowledgement. + delay(BACKUP_DEBOUNCE) + categoryMutexes.getValue(category).withLock { + runningBackups += category + try { + updateBackupStatus(category, expectedGeneration) { it.copy(running = true) } + if (category == BackupCategory.WALLET) { + paykitSdkService.isPaymentSubmissionActive.first { !it } + } + val status = cacheStore.backupStatuses.first()[category] ?: BackupItemStatus() + if (status.isRequired && !shouldSkipBackup(category) && + failedBackupRequired[category] != status.required + ) { + 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 (!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() @@ -535,18 +619,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), @@ -557,77 +641,145 @@ class BackupRepo @Inject constructor( } } - suspend fun triggerBackup(category: BackupCategory): Result = withContext(ioDispatcher) { + suspend fun triggerBackup(category: BackupCategory): Result = performExplicitBackup(category) + + 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) + } + } + } + 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) { + schedulePendingBackup(category, expectedGeneration) + } + } + } + } + + private suspend fun performBackup( + category: BackupCategory, + expectedGeneration: Long, + usdtSnapshot: String? = null, + explicitAttempt: Boolean = false, + reservedRequired: Long? = null, + ): Result = if (category == BackupCategory.WALLET || category == BackupCategory.METADATA) { - walletBackupMutex.withLock { performBackup(category) } + walletBackupMutex.withLock { + performSerializedBackup(category, expectedGeneration, usdtSnapshot, explicitAttempt, reservedRequired) + } } else { - performBackup(category) + performSerializedBackup(category, expectedGeneration, usdtSnapshot, explicitAttempt, reservedRequired) } - } - private suspend fun performBackup(category: BackupCategory): Result { + private suspend fun performSerializedBackup( + category: BackupCategory, + expectedGeneration: Long, + usdtSnapshot: String?, + explicitAttempt: Boolean, + reservedRequired: Long?, + ): Result { 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) + 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) } - - val upload = when (category) { - BackupCategory.WALLET -> uploadWalletBackup() - BackupCategory.USDT_RECOVERY -> uploadUsdtRecoveryBackup() - else -> runSuspendCatching { - vssBackupClient.putObject(key = category.name, data = getBackupDataBytes(category)).getOrThrow() - } + 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) { - cacheStore.updateBackupStatus(category) { - it.copy(running = false, synced = backupRequired) + } + 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 } - if (category == BackupCategory.WALLET) { + it.copy(running = false, synced = maxOf(it.synced, backupRequired)) + } + if (category == BackupCategory.WALLET) { + updateBackupStatus(BackupCategory.METADATA, expectedGeneration) { failedBackupRequired -= BackupCategory.METADATA - cacheStore.updateBackupStatus(BackupCategory.METADATA) { - if (it.isRequired) { - it.copy(required = maxOf(currentTimeMillis(), it.required + 1, it.synced + 1)) - } else { - it - } + if (it.isRequired) { + it.copy(required = maxOf(currentTimeMillis(), it.required + 1, it.synced + 1)) + } else { + it } } - 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)) { - withContext(NonCancellable) { - cacheStore.updateBackupStatus(category) { it.copy(running = false) } } + Logger.info("Backup succeeded for: '$category'", context = TAG) } - } + .onFailure { markBackupFailed(category, backupRequired, expectedGeneration, it, explicitAttempt) } + .map { } } - suspend fun persistWalletBackup(): Result = - withContext(ioDispatcher) { - walletBackupMutex.withLock { uploadWalletBackup() } - }.onFailure { markBackupRequired(BackupCategory.WALLET) } + suspend fun persistWalletBackup(): Result = withBackupOperation { expectedGeneration -> + walletBackupMutex.withLock { uploadWalletBackup() } + .onFailure { markBackupRequired(BackupCategory.WALLET, expectedGeneration) } + } - suspend fun persistUsdtRecoveryBackup(snapshot: String): Result = runSuspendCatching { - cacheStore.updateBackupStatus(BackupCategory.USDT_RECOVERY) { - it.copy(required = maxOf(currentTimeMillis(), it.required + 1, it.synced + 1)) + suspend fun persistUsdtRecoveryBackup(snapshot: String): Result = + performExplicitBackup(BackupCategory.USDT_RECOVERY, snapshot) + + 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) } + } + try { + task.await() + } catch (e: CancellationException) { + currentCoroutineContext().ensureActive() + Result.failure(UsdtException.BackupUnavailable().apply { initCause(e) }) + } finally { + withContext(NonCancellable) { task.cancelAndJoin() } } - uploadUsdtRecoveryBackup(snapshot).getOrThrow() - }.onFailure { markBackupRequired(BackupCategory.USDT_RECOVERY) } + } private suspend fun uploadUsdtRecoveryBackup(snapshot: String? = null): Result = withContext(ioDispatcher) { runSuspendCatching { @@ -663,38 +815,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) { @@ -1217,10 +1399,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) } } @@ -1380,5 +1563,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/main/java/to/bitkit/repositories/PubkyRepo.kt b/app/src/main/java/to/bitkit/repositories/PubkyRepo.kt index e722c3dd25..0402b95e29 100644 --- a/app/src/main/java/to/bitkit/repositories/PubkyRepo.kt +++ b/app/src/main/java/to/bitkit/repositories/PubkyRepo.kt @@ -1104,16 +1104,6 @@ class PubkyRepo @Inject constructor( } } - suspend fun deleteProfileWithSessionRetry(): Result = withContext(ioDispatcher) { - val initialResult = deleteProfile() - if (initialResult.isSuccess) return@withContext initialResult - - val refreshedSession = refreshSessionIfPossible().getOrDefault(false) - if (!refreshedSession) return@withContext initialResult - - deleteProfile() - } - suspend fun deleteProfile(): Result = runSuspendCatching { withContext(ioDispatcher) { requireNotNull(keychain.loadString(Keychain.Key.PAYKIT_SESSION.name)) { @@ -1830,20 +1820,6 @@ class PubkyRepo @Inject constructor( } } - suspend fun refreshSessionIfPossible(): Result = runSuspendCatching { - withContext(ioDispatcher) { - val storedSecretKeyHex = activeSecretKeyHex() ?: return@withContext false - - pubkyService.signIn(storedSecretKeyHex) - val publicKey = pubkyService.publicKeyFromSecret(storedSecretKeyHex).ensurePubkyPrefix() - - notifyBackupStateChanged() - continueSignIn(publicKey) - - true - } - } - // endregion // region Sign out diff --git a/app/src/main/java/to/bitkit/ui/screens/profile/EditProfileViewModel.kt b/app/src/main/java/to/bitkit/ui/screens/profile/EditProfileViewModel.kt index e2628806cf..06bb4bd4c5 100644 --- a/app/src/main/java/to/bitkit/ui/screens/profile/EditProfileViewModel.kt +++ b/app/src/main/java/to/bitkit/ui/screens/profile/EditProfileViewModel.kt @@ -273,7 +273,7 @@ class EditProfileViewModel @Inject constructor( try { privatePaykitRepo.beginProfileDeletion() privatePaykitRepo.removePublishedEndpointsForCleanup(TAG, deferAppSync = true) - val result = pubkyRepo.deleteProfileWithSessionRetry() + val result = pubkyRepo.deleteProfile() if (result.isSuccess) { privatePaykitRepo.closeAndClear() ToastEventBus.send( diff --git a/app/src/test/java/to/bitkit/repositories/BackupRepoTest.kt b/app/src/test/java/to/bitkit/repositories/BackupRepoTest.kt index 1d82f0b6e7..1938c2c811 100644 --- a/app/src/test/java/to/bitkit/repositories/BackupRepoTest.kt +++ b/app/src/test/java/to/bitkit/repositories/BackupRepoTest.kt @@ -1,16 +1,23 @@ 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.Dispatchers 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.coroutineScope @@ -20,11 +27,13 @@ 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 kotlinx.serialization.json.jsonPrimitive @@ -68,6 +77,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.data.withPublicSharingDisabledForRestore import to.bitkit.di.json import to.bitkit.models.ActivityBackupV1 @@ -86,13 +96,17 @@ import to.bitkit.models.WatchOnlyAccountRecord import to.bitkit.models.WatchOnlyAccountSetupState import to.bitkit.services.AddressDerivationInfo 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.CountDownLatch import java.util.concurrent.Executors import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger import javax.inject.Provider import kotlin.coroutines.cancellation.CancellationException import kotlin.test.assertEquals @@ -172,8 +186,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) @@ -281,7 +299,7 @@ class BackupRepoTest : BaseUnitTest() { false } Executors.newSingleThreadExecutor().asCoroutineDispatcher().use { ioDispatcher -> - sut = createSut(ioDispatcher = ioDispatcher) + sut = createSut(dispatcher = ioDispatcher) try { val read = async(Dispatchers.Main) { sut.hasPendingWalletRestore() } val storageThread = withContext(Dispatchers.Default) { @@ -352,14 +370,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)) @@ -1040,16 +1058,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 @@ -1316,7 +1338,7 @@ class BackupRepoTest : BaseUnitTest() { verify(settingsStore, never()).update(any()) verify(keychain, never()).delete(Keychain.Key.PAYKIT_PENDING_BACKUP_RESTORE.name) - sut = createSut(createReservationRepo(reservationStore)) + sut = createSut(reservationRepo = createReservationRepo(reservationStore)) assertTrue(sut.hasPendingWalletRestore()) assertTrue(keychain.exists(Keychain.Key.USDT_PENDING_BACKUP_RESTORE.name)) assertTrue(sut.triggerBackup(BackupCategory.WALLET).isFailure) @@ -1339,7 +1361,7 @@ class BackupRepoTest : BaseUnitTest() { val restartedReservations = createReservationRepo(reservationStore) assertTrue(restartedReservations.isUnavailableForReusableReceive(publicAddress)) - sut = createSut(restartedReservations) + sut = createSut(reservationRepo = restartedReservations) sut.triggerBackup(BackupCategory.WALLET).getOrThrow() val uploaded = argumentCaptor() @@ -1359,7 +1381,7 @@ class BackupRepoTest : BaseUnitTest() { assertTrue(sut.hasPendingWalletRestore()) assertFalse(settingsData.value.backupVerified) - sut = createSut(createReservationRepo(reservationStore)) + sut = createSut(reservationRepo = createReservationRepo(reservationStore)) whenever(vssBackupClient.getObject(BackupCategory.METADATA.name)) .thenReturn(Result.failure(BackupRepoTestError("metadata download unavailable"))) @@ -1372,7 +1394,7 @@ class BackupRepoTest : BaseUnitTest() { assertFalse(sut.hasPendingWalletRestore()) val restartedReservations = createReservationRepo(reservationStore) assertTrue(restartedReservations.isUnavailableForReusableReceive(publicAddress)) - sut = createSut(restartedReservations) + sut = createSut(reservationRepo = restartedReservations) sut.triggerBackup(BackupCategory.WALLET).getOrThrow() val uploaded = argumentCaptor() @@ -1444,7 +1466,7 @@ class BackupRepoTest : BaseUnitTest() { assertFalse(reservations.preparePublicAddressRestore()) reservations.capturePublicAddressRestoreSource(metadata.toByteArray()) val source = pendingPublicAddressRestoreRecord("source") - sut = createSut(createReservationRepo(reservationStore)) + sut = createSut(reservationRepo = createReservationRepo(reservationStore)) for (download in listOf(null, "$metadata ")) { whenever(vssBackupClient.getObject(BackupCategory.METADATA.name)).thenReturn( @@ -1494,7 +1516,7 @@ class BackupRepoTest : BaseUnitTest() { assertTrue(sut.triggerBackup(BackupCategory.METADATA).isFailure) verify(vssBackupClient, never()).putObject(any(), any()) - sut = createSut(createReservationRepo(reservationStore)) + sut = createSut(reservationRepo = createReservationRepo(reservationStore)) stubMetadataRestore(envelope = metadataEnvelope()) sut.performFullRestoreFromLatestBackup().getOrThrow() @@ -1516,7 +1538,7 @@ class BackupRepoTest : BaseUnitTest() { assertTrue(reservationStore.data.first().quarantinedPublicAddresses.isEmpty()) verify(vssBackupClient, never()).getObject(BackupCategory.WALLET.name) - sut = createSut(createReservationRepo(reservationStore)) + sut = createSut(reservationRepo = createReservationRepo(reservationStore)) val missingMetadata = listOf( Result.failure(BackupRepoTestError("metadata download unavailable")), Result.success(null), @@ -1573,7 +1595,7 @@ class BackupRepoTest : BaseUnitTest() { assertFalse(settingsData.value.backupVerified) pendingPublicAddressRestoreRecord("source") assertTrue(reservationStore.data.first().quarantinedPublicAddresses.isEmpty()) - sut = createSut(createReservationRepo(reservationStore)) + sut = createSut(reservationRepo = createReservationRepo(reservationStore)) whenever(vssBackupClient.getObject(BackupCategory.METADATA.name)) .thenReturn(Result.failure(BackupRepoTestError("metadata download unavailable"))) @@ -1659,7 +1681,7 @@ class BackupRepoTest : BaseUnitTest() { assertEquals(setOf(publicAddress), reservationStore.data.first().quarantinedPublicAddresses) assertEquals(publicAddress, pendingPublicAddressRestoreRecord("address")["address"]?.jsonPrimitive?.content) - sut = createSut(createReservationRepo(reservationStore)) + sut = createSut(reservationRepo = createReservationRepo(reservationStore)) whenever(vssBackupClient.getObject(BackupCategory.METADATA.name)) .thenReturn(Result.failure(BackupRepoTestError("metadata download unavailable"))) sut.performFullRestoreFromLatestBackup().getOrThrow() @@ -1741,6 +1763,85 @@ class BackupRepoTest : BaseUnitTest() { assertFalse(cacheData.value.backupStatuses.getValue(BackupCategory.METADATA).isRequired) } + @Test + fun `wallet acknowledgement automatically resumes blocked metadata observation`() = test { + val publicAddress = "bcrt1qpublic" + val reservationStore = stubPublicAddressRestore(publicAddress) + val legacyMetadata = envelopeWithRawField( + base = metadataEnvelope(cache = AppCacheData(onchainAddress = publicAddress)), + field = "tagMetadata", + rawJson = LEGACY_METADATA_JSON, + ) + val remote = mutableMapOf( + BackupCategory.METADATA.name to legacyMetadata.toByteArray(), + BackupCategory.WALLET.name to """{"version":1,"createdAt":123,"transfers":[]}""".toByteArray(), + ) + whenever(vssBackupClient.getObject(any())).doSuspendableAnswer { + val key = it.getArgument(0) + Result.success(remote[key]?.let { value -> VssItem(key = key, value = value, version = 1) }) + } + var walletUploadAvailable = false + val uploadedCategories = mutableListOf() + whenever(vssBackupClient.putObject(any(), any())).doSuspendableAnswer { + val key = it.getArgument(0) + val value = it.getArgument(1) + if (key == BackupCategory.WALLET.name && !walletUploadAvailable) { + Result.failure(BackupRepoTestError("wallet upload unavailable")) + } else { + remote[key] = value + uploadedCategories += key + Result.success(VssItem(key = key, value = value, version = 1)) + } + } + + sut.performFullRestoreFromLatestBackup().getOrThrow() + + assertTrue(settingsData.value.backupVerified) + assertFalse(sut.hasPendingWalletRestore()) + assertEquals(setOf(publicAddress), reservationStore.data.first().quarantinedPublicAddresses) + assertEquals(legacyMetadata, remote.getValue(BackupCategory.METADATA.name).decodeToString()) + assertTrue(cacheData.value.backupStatuses.getValue(BackupCategory.WALLET).isRequired) + assertTrue(cacheData.value.backupStatuses.getValue(BackupCategory.METADATA).isRequired) + verify(vssBackupClient, never()).putObject(eq(BackupCategory.WALLET.name), any()) + + sut.triggerBackup(BackupCategory.METADATA) + assertTrue(sut.triggerBackup(BackupCategory.WALLET).isFailure) + sut.triggerBackup(BackupCategory.METADATA) + assertEquals(legacyMetadata, remote.getValue(BackupCategory.METADATA.name).decodeToString()) + assertTrue(cacheData.value.backupStatuses.getValue(BackupCategory.WALLET).isRequired) + assertTrue(cacheData.value.backupStatuses.getValue(BackupCategory.METADATA).isRequired) + verify(vssBackupClient, never()).putObject(eq(BackupCategory.METADATA.name), any()) + + stubBackupObservers() + sut.startObservingBackups() + runCurrent() + advanceTimeBy(5_000) + runCurrent() + assertEquals(legacyMetadata, remote.getValue(BackupCategory.METADATA.name).decodeToString()) + + try { + walletUploadAvailable = true + sut.triggerBackup(BackupCategory.WALLET).getOrThrow() + val wallet = json.decodeFromString( + remote.getValue(BackupCategory.WALLET.name).decodeToString(), + ) + assertEquals(listOf(publicAddress), wallet.publicPaykitLegacyAddresses) + advanceTimeBy(5_000) + runCurrent() + assertEquals(1, uploadedCategories.count { it == BackupCategory.METADATA.name }) + val metadata = json.decodeFromString( + remote.getValue(BackupCategory.METADATA.name).decodeToString(), + ) + assertEquals("", metadata.cache.onchainAddress) + assertEquals(BackupCategory.WALLET.name, uploadedCategories.first()) + assertTrue(uploadedCategories.indexOf(BackupCategory.METADATA.name) > 0) + assertFalse(cacheData.value.backupStatuses.getValue(BackupCategory.WALLET).isRequired) + assertFalse(cacheData.value.backupStatuses.getValue(BackupCategory.METADATA).isRequired) + } finally { + sut.reset() + } + } + @Test fun `metadata restore waits for admitted metadata capture and upload to drain`() = test { stubMetadataBackupReads() @@ -1828,6 +1929,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)) @@ -1866,156 +2020,1332 @@ 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) + } + + sut.startObservingBackups() + setupStarted.await() + sut.setWiping(true) + val reset = async { sut.reset() } + runCurrent() + allowSetup.complete(Unit) + reset.await() + runCurrent() + + verify(vssBackupClient).reset() + verify(vssBackupClientLdk).reset() + verify(vssBackupClientLdk, never()).setup() + } + + @Test + fun `consumed settings change survives observation stop`() = test { + assertConsumedSettingsChangeSurvivesStop(reset = false) + } + + @Test + 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(5_000) runCurrent() - sut.stopObservingBackups() - advanceTimeBy(20_000) + release.complete(Unit) + resetJob?.await() runCurrent() - verify(vssBackupClient, never()).putObject(eq(BackupCategory.WALLET.name), any()) - + assertTrue(statuses.value.getValue(category).isRequired) sut.startObservingBackups() - runCurrent() advanceTimeBy(5_000) runCurrent() - - verify(vssBackupClient).putObject(eq(BackupCategory.WALLET.name), any()) + 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 `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 `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() - advanceTimeBy(31.minutes.inWholeMilliseconds) + assertFalse(reset.isCompleted) + verify(vssBackupClient, never()).reset() + verify(vssBackupClientLdk, never()).reset() + verify(vssBackupClient, times(1)).setupWithRetry(any(), any(), any()) + reset.cancel() runCurrent() - - verify(context, atLeastOnce()).getString(R.string.settings__backup__failed_title) + assertFalse(reset.isCompleted) } finally { - sut.stopObservingBackups() + release.complete(Unit) + reset.join() } + verify(vssBackupClient).reset() + verify(vssBackupClientLdk).reset() + verify(vssBackupClientLdk, never()).setup() } @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 `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() - verify(vssBackupClient, never()).putObject(eq(BackupCategory.WALLET.name), any()) - - allowWalletClear.complete(Unit) + 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() - - assertTrue(delayedWalletClear) - verify(vssBackupClient).putObject(eq(BackupCategory.WALLET.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 `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( + 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), ) ) - val allowWalletClear = CompletableDeferred().apply { complete(Unit) } - stubBackupStatuses(backupStatuses, allowWalletClear) {} + 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() } + } + } + 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).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 `successful upload leaves a newer required snapshot queued`() = 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.success(VssItem(key = BackupCategory.SETTINGS.name, value = byteArrayOf(), version = 1)) - } else { - Result.success( + 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() } + } + } + 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 { + advanceTimeBy(20_000) + runCurrent() + verify(privatePaykitRepo, times(1)).backupSnapshot() + verify(vssBackupClient, never()).putObject(eq(BackupCategory.WALLET.name), any()) + release.complete(Unit) + runCurrent() + advanceTimeBy(5_000) + runCurrent() + verify(vssBackupClient).putObject(eq(BackupCategory.WALLET.name), any()) + assertFalse(statuses.value.getValue(BackupCategory.WALLET).isRequired) + } finally { + 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 `committed recovery revision survives cancellation after an older permanent failure`() = test { + val category = BackupCategory.USDT_RECOVERY + val statuses = MutableStateFlow>(emptyMap()) + stubBackupStatuses(statuses, CompletableDeferred(Unit)) {} + stubBackupObservers() + whenever(usdtRepo.backupSnapshot()).thenReturn("signed") + var attempts = 0 + whenever { vssBackupClient.putObject(argThat { startsWith(UsdtRecoveryBackupStore.PREFIX) }, any()) } + .doSuspendableAnswer { + attempts++ + if (attempts == 1) { + Result.failure(VssException.UnknownException("invalid data")) + } else { + Result.success(VssItem(key = "recovery", value = byteArrayOf(), version = 1)) + } + } + assertTrue(sut.persistUsdtRecoveryBackup("signed").isFailure) + val failedRevision = statuses.value.getValue(category).required + sut.startObservingBackups() + runCurrent() + advanceTimeBy(10_000) + runCurrent() + assertEquals(1, attempts) + val committed = CompletableDeferred() + val release = CompletableDeferred() + whenever { cacheStore.update(any()) }.doSuspendableAnswer { + val transform = it.getArgument<(AppCacheData) -> AppCacheData>(0) + statuses.update { current -> transform(AppCacheData(backupStatuses = current)).backupStatuses } + val status = statuses.value.getValue(category) + if (!committed.isCompleted && status.required > failedRevision && !status.running) { + committed.complete(Unit) + release.await() + } + } + val service = UsdtService(keychain, Provider { sut }) + val callback = launch { service.recoveryBackup.persist("signed") } + try { + committed.await() + callback.cancel() + runCurrent() + release.complete(Unit) + callback.join() + assertEquals(0L, service.recoveryGeneration) + assertTrue(statuses.value.getValue(category).required > failedRevision) + advanceTimeBy(10_000) + runCurrent() + assertEquals(2, attempts) + assertFalse(statuses.value.getValue(category).isRequired) + assertEquals(0L, service.recoveryGeneration) + } finally { + release.complete(Unit) + callback.cancelAndJoin() + sut.reset() + } + } + + @Test + fun `explicit recovery callback bypasses scheduled debounce and owns its acknowledgement`() = test { + val category = BackupCategory.USDT_RECOVERY + val statuses = MutableStateFlow(mapOf(category to BackupItemStatus(required = 1_000))) + stubBackupStatuses(statuses, CompletableDeferred(Unit)) {} + stubBackupObservers() + whenever(usdtRepo.backupSnapshot()).thenReturn("background") + val uploads = mutableListOf() + whenever { vssBackupClient.putObject(argThat { startsWith(UsdtRecoveryBackupStore.PREFIX) }, any()) } + .doSuspendableAnswer { + val payload = json.decodeFromString(it.getArgument(1).decodeToString()) + uploads += payload.usdtWallet + Result.success(VssItem(key = "recovery", value = byteArrayOf(), version = 1)) + } + sut.startObservingBackups() + runCurrent() + advanceTimeBy(150) + val service = UsdtService(keychain, Provider { sut }) + val callback = async { service.recoveryBackup.persist("signed") } + try { + runCurrent() + assertTrue(callback.isCompleted, "Explicit recovery must not wait for scheduled debounce") + callback.await() + assertEquals(listOf("signed"), uploads) + assertEquals(1L, service.recoveryGeneration) + advanceTimeBy(5_000) + runCurrent() + assertEquals(listOf("signed"), uploads) + } finally { + callback.cancelAndJoin() + 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(), @@ -2029,6 +3359,7 @@ class BackupRepoTest : BaseUnitTest() { stubBackupObservers() try { + sut.scheduleFullBackup() sut.startObservingBackups() runCurrent() advanceTimeBy(5_000) @@ -2653,7 +3984,7 @@ class BackupRepoTest : BaseUnitTest() { holdDownload = false val restartedReservations = createReservationRepo(reservationStore) - sut = createSut(restartedReservations) + sut = createSut(reservationRepo = restartedReservations) sut.setRestorePending(false) assertTrue(sut.performFullRestoreFromLatestBackup().isSuccess) @@ -2755,7 +4086,7 @@ class BackupRepoTest : BaseUnitTest() { val transform = it.getArgument<(PrivatePaykitReservationData) -> PrivatePaykitReservationData>(0) reservationData.update(transform) } - sut = createSut(createReservationRepo(reservationStore)) + sut = createSut(reservationRepo = createReservationRepo(reservationStore)) return reservationStore } @@ -2931,17 +4262,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 } } } @@ -2978,12 +4310,13 @@ class BackupRepoTest : BaseUnitTest() { } private fun createSut( + backupCache: CacheStore = cacheStore, + dispatcher: CoroutineDispatcher = testDispatcher, reservationRepo: PrivatePaykitAddressReservationRepo = privatePaykitAddressReservationRepo, - ioDispatcher: CoroutineDispatcher = testDispatcher, ) = BackupRepo( context = context, - ioDispatcher = ioDispatcher, - cacheStore = cacheStore, + ioDispatcher = dispatcher, + cacheStore = backupCache, vssBackupClient = vssBackupClient, vssBackupClientLdk = vssBackupClientLdk, settingsStore = settingsStore, diff --git a/app/src/test/java/to/bitkit/repositories/PubkyRepoTest.kt b/app/src/test/java/to/bitkit/repositories/PubkyRepoTest.kt index 81a7f15999..bb8b3a12bd 100644 --- a/app/src/test/java/to/bitkit/repositories/PubkyRepoTest.kt +++ b/app/src/test/java/to/bitkit/repositories/PubkyRepoTest.kt @@ -1375,36 +1375,46 @@ class PubkyRepoTest : BaseUnitTest() { } @Test - fun `deleteProfileWithSessionRetry should refresh session and retry delete`() = test { - val expiredSession = "expired_session" - val newSession = "new_session" + fun `profile deletion does not replay after an uncertain response`() = test { val secretKey = "local_secret" - authenticateForTesting(publicKey = VALID_SELF_KEY, secret = expiredSession) - whenever(keychain.loadString(Keychain.Key.PAYKIT_SESSION.name)).thenReturn(expiredSession, newSession) + authenticateForTesting(publicKey = VALID_SELF_KEY, secret = "test_session") whenever(keychain.loadString(Keychain.Key.PUBKY_SECRET_KEY.name)).thenReturn(secretKey) - whenever(pubkyService.deletePaykitProfile()).thenAnswer { throw TestAppError("Expired") }.thenReturn(Unit) - whenever(pubkyService.signIn(secretKey)).thenReturn(Unit) whenever(pubkyService.publicKeyFromSecret(secretKey)).thenReturn(VALID_SELF_KEY.removePrefix("pubky")) + val uncertainResponse = TestAppError("Connection closed after profile deletion") + whenever(pubkyService.deletePaykitProfile()).thenAnswer { throw uncertainResponse }.thenReturn(Unit) - val result = sut.deleteProfileWithSessionRetry() + val result = sut.deleteProfile() - assertTrue(result.isSuccess) - verifyBlocking(pubkyService, times(2)) { deletePaykitProfile() } + assertSame(uncertainResponse, result.exceptionOrNull()) + verify(pubkyService).deletePaykitProfile() + verify(pubkyService, never()).signIn(any()) + verify(pubkyService, never()).signOut() } @Test - fun `deleteProfileWithSessionRetry should return failure when session cannot refresh`() = test { - val expiredSession = "expired_session" - authenticateForTesting(publicKey = VALID_SELF_KEY, secret = expiredSession) - whenever(keychain.loadString(Keychain.Key.PAYKIT_SESSION.name)).thenReturn(expiredSession) - whenever(keychain.loadString(Keychain.Key.PUBKY_SECRET_KEY.name)).thenReturn(null) - whenever(pubkyService.deletePaykitProfile()).thenAnswer { throw TestAppError("Expired") } - - val result = sut.deleteProfileWithSessionRetry() + fun `failed profile deletion does not restore an identity after sign out`() = test { + val secretKey = "local_secret" + authenticateForTesting(publicKey = VALID_SELF_KEY, secret = "test_session") + whenever(keychain.loadString(Keychain.Key.PUBKY_SECRET_KEY.name)).thenReturn(secretKey) + whenever(pubkyService.publicKeyFromSecret(secretKey)).thenReturn(VALID_SELF_KEY.removePrefix("pubky")) + val deletionStarted = CompletableDeferred() + val finishDeletion = CompletableDeferred() + val uncertainResponse = TestAppError("Connection closed after profile deletion") + whenever(pubkyService.deletePaykitProfile()).doSuspendableAnswer { + deletionStarted.complete(Unit) + withContext(NonCancellable) { finishDeletion.await() } + throw uncertainResponse + } + val deletion = async { sut.deleteProfile() } + deletionStarted.await() + sut.signOut().getOrThrow() + finishDeletion.complete(Unit) - assertTrue(result.isFailure) - verifyBlocking(pubkyService) { deletePaykitProfile() } - verifyBlocking(pubkyService, never()) { signIn(any()) } + assertSame(uncertainResponse, deletion.await().exceptionOrNull()) + assertNull(sut.publicKey.value) + assertFalse(sut.isAuthenticated.value) + verify(pubkyService, never()).signIn(any()) + verify(pubkyService).deletePaykitProfile() } @Test @@ -3562,31 +3572,6 @@ class PubkyRepoTest : BaseUnitTest() { verify(pubkyStore).reset() } - @Test - fun `refreshSessionIfPossible should refresh session when local secret key exists`() = test { - val secretKey = "local_secret" - whenever(keychain.loadString(Keychain.Key.PUBKY_SECRET_KEY.name)).thenReturn(secretKey) - whenever(pubkyService.signIn(secretKey)).thenReturn(Unit) - whenever(pubkyService.publicKeyFromSecret(secretKey)).thenReturn(VALID_SELF_KEY.removePrefix("pubky")) - - val result = sut.refreshSessionIfPossible() - - assertEquals(true, result.getOrNull()) - assertEquals(VALID_SELF_KEY, sut.publicKey.value) - assertTrue(sut.isAuthenticated.value) - } - - @Test - fun `refreshSessionIfPossible should return false when local secret key is missing`() = test { - whenever(keychain.loadString(Keychain.Key.PUBKY_SECRET_KEY.name)).thenReturn(null) - - val result = sut.refreshSessionIfPossible() - - assertEquals(false, result.getOrNull()) - assertNull(sut.publicKey.value) - assertFalse(sut.isAuthenticated.value) - } - @Test fun `restoreSessionBackupState should derive local secret key for local seed backups`() = test { whenever(keychain.loadString(Keychain.Key.BIP39_MNEMONIC.name)).thenReturn("test mnemonic") @@ -4490,7 +4475,7 @@ class PubkyRepoTest : BaseUnitTest() { } @Test - fun `a contact edit started before the session of its identity is refreshed still saves`() = test { + fun `a contact edit started before the session of its identity is restored still saves`() = test { val store = stubGatedPubkyStore() authenticateForTesting(publicKey = VALID_SELF_KEY) whenever(keychain.loadString(Keychain.Key.PUBKY_SECRET_KEY.name)).thenReturn("local_secret") @@ -4498,7 +4483,7 @@ class PubkyRepoTest : BaseUnitTest() { whenever(pubkyService.publicKeyFromSecret("local_secret")).thenReturn(VALID_SELF_KEY.removePrefix("pubky")) val signIn = checkNotNull(sut.currentSignIn()) - assertEquals(true, sut.refreshSessionIfPossible().getOrNull()) + sut.initialize().getOrThrow() val result = sut.updateContact(signIn, VALID_CONTACT_KEY_A, "Alice", "", null, emptyList(), listOf("Friend")) assertTrue(result.isSuccess) diff --git a/app/src/test/java/to/bitkit/ui/screens/profile/EditProfileViewModelTest.kt b/app/src/test/java/to/bitkit/ui/screens/profile/EditProfileViewModelTest.kt index f9c005d101..381c03e5e4 100644 --- a/app/src/test/java/to/bitkit/ui/screens/profile/EditProfileViewModelTest.kt +++ b/app/src/test/java/to/bitkit/ui/screens/profile/EditProfileViewModelTest.kt @@ -60,7 +60,7 @@ class EditProfileViewModelTest : BaseUnitTest() { sut.showDeleteConfirmation() advanceUntilIdle() verify(pubkyRepo, times(1)).forgetUnrestoredIdentity() - verify(pubkyRepo, never()).deleteProfileWithSessionRetry() + verify(pubkyRepo, never()).deleteProfile() verify(pubkyRepo, never()).saveProfile(any(), any(), any(), any(), any()) assertFalse(sut.uiState.value.showDeleteDialog) @@ -113,7 +113,7 @@ class EditProfileViewModelTest : BaseUnitTest() { val sut = createSut() advanceUntilIdle() val result = CompletableDeferred>() - whenever(pubkyRepo.deleteProfileWithSessionRetry()).doSuspendableAnswer { result.await() } + whenever(pubkyRepo.deleteProfile()).doSuspendableAnswer { result.await() } sut.showDeleteConfirmation() sut.deleteProfile() @@ -129,7 +129,7 @@ class EditProfileViewModelTest : BaseUnitTest() { sut.disconnectProfile() sut.showDeleteConfirmation() advanceUntilIdle() - verify(pubkyRepo, times(1)).deleteProfileWithSessionRetry() + verify(pubkyRepo, times(1)).deleteProfile() verify(pubkyRepo, never()).saveProfile(any(), any(), any(), any(), any()) verify(pubkyRepo, never()).signOut() assertFalse(sut.uiState.value.showDeleteDialog) @@ -161,7 +161,7 @@ class EditProfileViewModelTest : BaseUnitTest() { sut.showDeleteConfirmation() advanceUntilIdle() verify(pubkyRepo, times(1)).saveProfile(any(), any(), any(), any(), any()) - verify(pubkyRepo, never()).deleteProfileWithSessionRetry() + verify(pubkyRepo, never()).deleteProfile() verify(pubkyRepo, never()).forgetUnrestoredIdentity() assertFalse(sut.uiState.value.showDeleteDialog) @@ -183,7 +183,7 @@ class EditProfileViewModelTest : BaseUnitTest() { @Test fun `deleteProfile should emit success when repository delete succeeds`() = test { - whenever(pubkyRepo.deleteProfileWithSessionRetry()).thenReturn(Result.success(Unit)) + whenever(pubkyRepo.deleteProfile()).thenReturn(Result.success(Unit)) val sut = createSut() advanceUntilIdle() @@ -197,14 +197,14 @@ class EditProfileViewModelTest : BaseUnitTest() { assertFalse(sut.uiState.value.showDeleteFailureDialog) inOrder(privatePaykitRepo, pubkyRepo).apply { verify(privatePaykitRepo).removePublishedEndpointsForCleanup(any(), deferAppSync = eq(true)) - verify(pubkyRepo).deleteProfileWithSessionRetry() + verify(pubkyRepo).deleteProfile() verify(privatePaykitRepo).closeAndClear() } } @Test fun `profile created in Bitkit keeps delete copy`() = test { - whenever(pubkyRepo.deleteProfileWithSessionRetry()).thenReturn(Result.success(Unit)) + whenever(pubkyRepo.deleteProfile()).thenReturn(Result.success(Unit)) val sut = createSut() advanceUntilIdle() @@ -218,7 +218,7 @@ class EditProfileViewModelTest : BaseUnitTest() { @Test fun `ring identity is unlinked with the same cleanup`() = test { - whenever(pubkyRepo.deleteProfileWithSessionRetry()).thenReturn(Result.success(Unit)) + whenever(pubkyRepo.deleteProfile()).thenReturn(Result.success(Unit)) val sut = createSut(isRingIdentity = true) advanceUntilIdle() @@ -234,7 +234,7 @@ class EditProfileViewModelTest : BaseUnitTest() { verify(context, never()).getString(R.string.profile__delete_success) inOrder(privatePaykitRepo, pubkyRepo).apply { verify(privatePaykitRepo).removePublishedEndpointsForCleanup(any(), deferAppSync = eq(true)) - verify(pubkyRepo).deleteProfileWithSessionRetry() + verify(pubkyRepo).deleteProfile() verify(privatePaykitRepo).closeAndClear() } } @@ -256,7 +256,7 @@ class EditProfileViewModelTest : BaseUnitTest() { @Test fun `retryDeleteProfile should emit success when repository delete succeeds`() = test { - whenever(pubkyRepo.deleteProfileWithSessionRetry()).thenReturn(Result.success(Unit)) + whenever(pubkyRepo.deleteProfile()).thenReturn(Result.success(Unit)) val sut = createSut() advanceUntilIdle() @@ -270,14 +270,14 @@ class EditProfileViewModelTest : BaseUnitTest() { assertFalse(sut.uiState.value.showDeleteFailureDialog) inOrder(privatePaykitRepo, pubkyRepo).apply { verify(privatePaykitRepo).removePublishedEndpointsForCleanup(any(), deferAppSync = eq(true)) - verify(pubkyRepo).deleteProfileWithSessionRetry() + verify(pubkyRepo).deleteProfile() verify(privatePaykitRepo).closeAndClear() } } @Test fun `deleteProfile should show retry dialog when delete still fails`() = test { - whenever(pubkyRepo.deleteProfileWithSessionRetry()).thenReturn( + whenever(pubkyRepo.deleteProfile()).thenReturn( Result.failure(TestAppError("expired session")), ) @@ -295,7 +295,7 @@ class EditProfileViewModelTest : BaseUnitTest() { @Test fun `disconnectProfile should emit disconnect success`() = test { - whenever(pubkyRepo.deleteProfileWithSessionRetry()).thenReturn( + whenever(pubkyRepo.deleteProfile()).thenReturn( Result.failure(TestAppError("expired session")), ) whenever(pubkyRepo.signOut()).thenReturn(Result.success(Unit)) @@ -354,7 +354,7 @@ class EditProfileViewModelTest : BaseUnitTest() { val sut = createSut() whenever { privatePaykitRepo.removePublishedEndpointsForCleanup(any(), deferAppSync = eq(true)) } .thenReturn(Result.failure(TestAppError("cleanup failed"))) - whenever(pubkyRepo.deleteProfileWithSessionRetry()).thenReturn(Result.success(Unit)) + whenever(pubkyRepo.deleteProfile()).thenReturn(Result.success(Unit)) advanceUntilIdle() sut.effects.test { @@ -367,14 +367,14 @@ class EditProfileViewModelTest : BaseUnitTest() { assertFalse(sut.uiState.value.isDeleting) inOrder(privatePaykitRepo, pubkyRepo).apply { verify(privatePaykitRepo).removePublishedEndpointsForCleanup(any(), deferAppSync = eq(true)) - verify(pubkyRepo).deleteProfileWithSessionRetry() + verify(pubkyRepo).deleteProfile() verify(privatePaykitRepo).closeAndClear() } } @Test fun `dismissDeleteFailureDialog should hide retry dialog`() = test { - whenever(pubkyRepo.deleteProfileWithSessionRetry()).thenReturn( + whenever(pubkyRepo.deleteProfile()).thenReturn( Result.failure(TestAppError("expired session")), ) diff --git a/changelog.d/next/1688.fixed.md b/changelog.d/next/1688.fixed.md new file mode 100644 index 0000000000..fb97b9ecd4 --- /dev/null +++ b/changelog.d/next/1688.fixed.md @@ -0,0 +1 @@ +Prevent uncertain profile deletion failures from automatically repeating contact cleanup and deletion. diff --git a/changelog.d/next/1689.fixed.md b/changelog.d/next/1689.fixed.md new file mode 100644 index 0000000000..8e1c25855d --- /dev/null +++ b/changelog.d/next/1689.fixed.md @@ -0,0 +1 @@ +Keep interrupted backups within their wallet lifecycle, retry newer pending data after failures, and let explicit recovery acknowledgements proceed without the background delay. 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. diff --git a/journeys/profile/delete-profile-uncertain-response.xml b/journeys/profile/delete-profile-uncertain-response.xml new file mode 100644 index 0000000000..fe825c36ff --- /dev/null +++ b/journeys/profile/delete-profile-uncertain-response.xml @@ -0,0 +1,17 @@ + + + Requires a disposable test identity and a controlled homeserver proxy; never use a personal profile. + Configure the proxy to accept one profile deletion but lose its acknowledgement, retaining a redacted + request counter. This is a fault-injection fixture requirement, not evidence that this journey has run. + Normal SDK recovery may retry an individual request only after a confirmed authentication rejection; + the app must not restart contact cleanup and profile deletion because a request has an uncertain outcome. + + + Open Edit Profile and confirm deletion + Drop the acknowledgement after the server accepts the profile deletion + Verify failure options are shown and no success notification or onboarding navigation occurs + Wait through the network timeout and verify the app starts no second contact cleanup, profile deletion or sign-in + Cancel the failure dialog and verify no mutation is retried + Restart the app without choosing Retry and verify it does not automatically repeat the deletion + +