diff --git a/app/src/main/java/to/bitkit/repositories/LightningRepo.kt b/app/src/main/java/to/bitkit/repositories/LightningRepo.kt index 2ac1ab04d6..6c5297b164 100644 --- a/app/src/main/java/to/bitkit/repositories/LightningRepo.kt +++ b/app/src/main/java/to/bitkit/repositories/LightningRepo.kt @@ -2204,13 +2204,13 @@ class LightningRepo @Inject constructor( */ suspend fun restartNodeDetached(): Result = scope.async { restartNode() }.await() - suspend fun restartNode(): Result = withContext(bgDispatcher) { + suspend fun restartNode(channelMigration: ChannelDataMigration? = null): Result = withContext(bgDispatcher) { Logger.info("Restarting node", context = TAG) stop().onFailure { Logger.error("Failed to stop node during restart", it, context = TAG) return@withContext Result.failure(it) } - startNode(shouldRetryYieldToStop = true).onFailure { + startNode(channelMigration = channelMigration, shouldRetryYieldToStop = true).onFailure { if (it is NodeStartYieldedToStopError) { Logger.info("Deferred node restart to a requested stop", context = TAG) } else { diff --git a/app/src/main/java/to/bitkit/services/MigrationService.kt b/app/src/main/java/to/bitkit/services/MigrationService.kt index 24696ceb1a..ebbc5a81c3 100644 --- a/app/src/main/java/to/bitkit/services/MigrationService.kt +++ b/app/src/main/java/to/bitkit/services/MigrationService.kt @@ -25,6 +25,7 @@ import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.update +import kotlinx.coroutines.sync.Mutex import kotlinx.serialization.SerialName import kotlinx.serialization.Serializable import kotlinx.serialization.json.JsonObject @@ -47,6 +48,7 @@ import to.bitkit.data.keychain.Keychain import to.bitkit.data.resetPin import to.bitkit.di.json import to.bitkit.env.Env +import to.bitkit.ext.runSuspendCatching import to.bitkit.models.BitcoinDisplayUnit import to.bitkit.models.CoinSelectionPreference import to.bitkit.models.DEFAULT_ADDRESS_TYPE_STRING @@ -112,6 +114,12 @@ class MigrationService @Inject constructor( internal const val BLOCKTANK_ORDER_IDS_CHUNK = 20 } + private val channelMigrationMutex = Mutex() + + suspend fun lockChannelMigration() = channelMigrationMutex.lock() + + fun unlockChannelMigration() = channelMigrationMutex.unlock() + private val rnMigrationStore = context.rnMigrationDataStore private inline fun decodeBackupData(data: ByteArray): T { @@ -318,6 +326,11 @@ class MigrationService @Inject constructor( Logger.debug("Cleared all persisted migration data", context = TAG) } + suspend fun hasPendingMigrationRetries(): Boolean { + loadPersistedMigrationData() + return !canCleanupAfterMigration + } + val canCleanupAfterMigration: Boolean get() { if (pendingBlocktankOrderIds != null || pendingRemotePaidOrders != null) { @@ -328,6 +341,10 @@ class MigrationService @Inject constructor( Logger.debug("Cannot cleanup: pending metadata/transfers/boosts exists", context = TAG) return false } + if (peekPendingChannelMigration() != null) { + Logger.debug("Cannot cleanup: pending channel migration exists", context = TAG) + return false + } return true } @@ -941,13 +958,14 @@ class MigrationService @Inject constructor( } } - private suspend fun applyRNMetadata(metadata: RNMetadata) { + private suspend fun applyRNMetadata(metadata: RNMetadata): Set { val tags = metadata.tags if (tags.isNullOrEmpty()) { Logger.debug("No tags to apply in metadata", context = TAG) - return + return emptySet() } + val unapplied = mutableSetOf() var applied = 0 val allTags = tags.mapNotNull { (activityId, tagList) -> val onchain = activityRepo.getOnchainActivityByTxId(activityId) @@ -968,7 +986,8 @@ class MigrationService @Inject constructor( tags = tagList, ) } else { - Logger.warn("Activity not found for tags: id=$activityId", context = TAG) + Logger.warn("Activity not found for tags: id='$activityId'", context = TAG) + unapplied += activityId null } } @@ -979,9 +998,11 @@ class MigrationService @Inject constructor( coreService.activity.upsertTags(allTags) Logger.info("Applied $applied/${tags.size} pending tags", context = TAG) }.onFailure { - Logger.error("Failed to upsert tags: $it", it, context = TAG) + Logger.error("Failed to upsert tags", it, context = TAG) + unapplied += tags.keys } } + return unapplied } private suspend fun applyRNTodos(todos: RNTodos) { @@ -1672,7 +1693,7 @@ class MigrationService @Inject constructor( loadPersistedMigrationData() // Handle MMKV (local) migration data - apply activities FIRST, then metadata - if (hasRNMmkvData()) { + if (needsPostMigrationSync() && hasRNMmkvData()) { loadRNMmkvData()?.let { mmkvData -> extractRNActivities(mmkvData)?.let { activities -> Logger.info("Applying ${activities.size} MMKV activities", context = TAG) @@ -1682,19 +1703,13 @@ class MigrationService @Inject constructor( extractRNWalletBackup(mmkvData)?.let { (transfers, boosts) -> if (transfers.isNotEmpty()) { Logger.info("Applying ${transfers.size} local transfer markers", context = TAG) - applyRemoteTransfers(transfers) + persistTransfers(pendingRemoteTransfers.orEmpty() + transfers) } if (boosts.isNotEmpty()) { Logger.info("Applying ${boosts.size} local boost markers", context = TAG) - applyBoostTransactions(boosts) + persistBoosts(pendingRemoteBoosts.orEmpty() + boosts) } } - - // Apply MMKV metadata (tags) AFTER activities are created - extractRNMetadata(mmkvData)?.let { metadata -> - Logger.info("Applying MMKV metadata (tags: ${metadata.tags?.size})", context = TAG) - applyRNMetadata(metadata) - } } } @@ -1707,21 +1722,25 @@ class MigrationService @Inject constructor( pendingRemoteTransfers?.let { transfers -> Logger.info("Applying ${transfers.size} remote transfer markers", context = TAG) - applyRemoteTransfers(transfers) - clearPersistedTransfers() + val remaining = applyRemoteTransfers(transfers) + if (remaining.isEmpty()) clearPersistedTransfers() else persistTransfers(remaining) } pendingRemoteBoosts?.let { boosts -> Logger.info("Applying ${boosts.size} remote boost markers", context = TAG) - applyBoostTransactions(boosts) - clearPersistedBoosts() + val remaining = applyBoostTransactions(boosts) + if (remaining.isEmpty()) clearPersistedBoosts() else persistBoosts(remaining) } // Apply remote metadata (tags) AFTER activities are created pendingRemoteMetadata?.let { metadata -> Logger.info("Applying remote metadata (tags: ${metadata.tags?.size})", context = TAG) - applyRNMetadata(metadata) - clearPersistedMetadata() + val remaining = unappliedRnMetadata(metadata, applyRNMetadata(metadata)) + if (remaining == null) { + clearPersistedMetadata() + } else { + persistMetadata(remaining) + } } var blocktankFetchFailed = false @@ -1775,15 +1794,20 @@ class MigrationService @Inject constructor( } } - private suspend fun applyRemoteTransfers(transfers: Map) { + internal suspend fun applyRemoteTransfers(transfers: Map): Map { + val remaining = transfers.toMutableMap() transfers.forEach { (txId, channelId) -> val onchain = activityRepo.getOnchainActivityByTxId(txId) ?: return@forEach val updated = onchain.copy(isTransfer = true, channelId = channelId) - activityRepo.updateActivity(onchain.id, Activity.Onchain(updated)) + activityRepo.updateActivity(onchain.id, Activity.Onchain(updated)).onSuccess { + remaining.remove(txId) + } } + return remaining } - private suspend fun applyBoostTransactions(boosts: Map) { + internal suspend fun applyBoostTransactions(boosts: Map): Map { + val remaining = boosts.toMutableMap() var applied = 0 boosts.forEach { (oldTxId, newTxId) -> @@ -1807,9 +1831,10 @@ class MigrationService @Inject constructor( boostTxIds = newOnchain.boostTxIds.filter { it != oldTxId }, ) - runCatching { - activityRepo.updateActivity(parentOnchain.id, Activity.Onchain(parentOnchain)) - activityRepo.updateActivity(updatedNewOnchain.id, Activity.Onchain(updatedNewOnchain)) + runSuspendCatching { + activityRepo.updateActivity(parentOnchain.id, Activity.Onchain(parentOnchain)).getOrThrow() + activityRepo.updateActivity(updatedNewOnchain.id, Activity.Onchain(updatedNewOnchain)).getOrThrow() + remaining.remove(oldTxId) applied++ }.onFailure { e -> Logger.error( @@ -1829,8 +1854,9 @@ class MigrationService @Inject constructor( boostTxIds = updatedBoostTxIds, ) - runCatching { - activityRepo.updateActivity(updated.id, Activity.Onchain(updated)) + runSuspendCatching { + activityRepo.updateActivity(updated.id, Activity.Onchain(updated)).getOrThrow() + remaining.remove(oldTxId) applied++ }.onFailure { e -> Logger.error("Failed to apply RBF boost for tx $newTxId: $e", e, context = TAG) @@ -1839,6 +1865,7 @@ class MigrationService @Inject constructor( } Logger.info("Applied $applied/${boosts.size} boost markers", context = TAG) + return remaining } private suspend fun applyBoostedParents(boostedParents: List, txId: String) { @@ -2201,6 +2228,13 @@ data class RNMetadata( val lastUsedTags: List? = null, ) +internal fun unappliedRnMetadata(metadata: RNMetadata, unappliedActivityIds: Set): RNMetadata? { + if (unappliedActivityIds.isEmpty()) return null + val remaining = metadata.tags.orEmpty().filterKeys { it in unappliedActivityIds } + if (remaining.isEmpty()) return null + return metadata.copy(tags = remaining, lastUsedTags = null) +} + @Serializable data class RNTodos( val hide: Map? = null, diff --git a/app/src/main/java/to/bitkit/viewmodels/AppViewModel.kt b/app/src/main/java/to/bitkit/viewmodels/AppViewModel.kt index 9bb6e638ee..31dabb0103 100644 --- a/app/src/main/java/to/bitkit/viewmodels/AppViewModel.kt +++ b/app/src/main/java/to/bitkit/viewmodels/AppViewModel.kt @@ -1841,7 +1841,7 @@ class AppViewModel @Inject constructor( when { (isShowingLoading || needsPostMigrationSync) && !isCompletingMigration -> completeMigration() - isRestoringRemote -> completeRNRemoteBackupRestore() + isRestoringRemote && !isCompletingMigration -> completeRNRemoteBackupRestore() pendingPrune -> { settingsStore.update { it.copy(pendingRestoreAddressTypePrune = false) } delay(POST_RESTORE_PRUNE_DELAY_MS) @@ -1849,7 +1849,10 @@ class AppViewModel @Inject constructor( walletRepo.debounceSyncByEvent() } - !isShowingLoading && !needsPostMigrationSync && !isCompletingMigration -> walletRepo.debounceSyncByEvent() + !isShowingLoading && !needsPostMigrationSync && !isCompletingMigration -> { + retryPendingMigrationData() + walletRepo.debounceSyncByEvent() + } else -> Unit } @@ -1863,6 +1866,17 @@ class AppViewModel @Inject constructor( } } + private suspend fun retryPendingMigrationData() { + isCompletingMigration = true + try { + if (!migrationService.hasPendingMigrationRetries()) return + migrationService.reapplyMetadataAfterSync() + if (migrationService.canCleanupAfterMigration) migrationService.cleanupAfterMigration() + } finally { + isCompletingMigration = false + } + } + private suspend fun completePendingRestoreActivitySeen(syncedBlockHeight: UInt) { // Claimed before the first suspension, so a later sync completing while this sweep runs cannot record its // higher tip and silence a new receive confirmed in between. @@ -1888,47 +1902,69 @@ class AppViewModel @Inject constructor( } private suspend fun completeRNRemoteBackupRestore() { - val channelMigration = buildChannelMigrationIfAvailable() + isCompletingMigration = true + try { + if (!applyPendingChannelMigration()) { + finishMigrationWithError() + return + } + + lightningRepo.getPayments().onSuccess { activityRepo.syncLdkNodePayments(it) } + migrationService.reapplyMetadataAfterSync() + activityRepo.syncActivities() + walletRepo.syncBalances() + + migrationService.setNeedsPostMigrationSync(false) + migrationService.setRestoringFromRNRemoteBackup(false) + if (migrationService.canCleanupAfterMigration) { + migrationService.cleanupAfterMigration() + migrationService.setShowingMigrationLoading(false) + } else { + Logger.info("Post-migration sync incomplete (remote restore), will retry on next sync", context = TAG) + migrationService.setShowingMigrationLoading(false) + } + } finally { + isCompletingMigration = false + } + } - if (channelMigration != null) { + private fun buildChannelMigrationIfAvailable(): ChannelDataMigration? { + val migration = migrationService.peekPendingChannelMigration() ?: return null + return ChannelDataMigration( + channelManager = migration.channelManager.map { it.toUByte() }, + channelMonitors = migration.channelMonitors.map { monitor -> monitor.map { it.toUByte() } }, + ) + } + + private suspend fun applyPendingChannelMigration(): Boolean { + migrationService.lockChannelMigration() + try { + val channelMigration = buildChannelMigrationIfAvailable() ?: return true lightningRepo.stop().onFailure { Logger.error("Failed to stop node during remote restore restart", it, context = TAG) } + if (lightningRepo.lightningState.value.nodeLifecycleState.isRunningOrStarting()) { + Logger.error("Node still running, pending channel migration was not applied", context = TAG) + return false + } delay(REMOTE_RESTORE_NODE_RESTART_DELAY_MS) + var applied = false lightningRepo.start(channelMigration = channelMigration, shouldRetry = false) .onSuccess { migrationService.consumePendingChannelMigration() walletRepo.syncNodeAndWallet() walletRepo.syncBalances() + applied = true } .onFailure { e -> - Logger.error("Failed to restart node after remote restore: $e", e, context = TAG) + Logger.error("Failed to restart node after remote restore", e, context = TAG) } - } - - lightningRepo.getPayments().onSuccess { activityRepo.syncLdkNodePayments(it) } - migrationService.reapplyMetadataAfterSync() - activityRepo.syncActivities() - walletRepo.syncBalances() - - if (migrationService.canCleanupAfterMigration) { - migrationService.cleanupAfterMigration() - migrationService.setRestoringFromRNRemoteBackup(false) - migrationService.setShowingMigrationLoading(false) - } else { - Logger.info("Post-migration sync incomplete (remote restore), will retry on next sync", context = TAG) - migrationService.setShowingMigrationLoading(false) + return applied + } finally { + migrationService.unlockChannelMigration() } } - private fun buildChannelMigrationIfAvailable(): ChannelDataMigration? { - val migration = migrationService.peekPendingChannelMigration() ?: return null - return ChannelDataMigration( - channelManager = migration.channelManager.map { it.toUByte() }, - channelMonitors = migration.channelMonitors.map { monitor -> monitor.map { it.toUByte() } }, - ) - } - private suspend fun completeMigration() { if (isCompletingMigration) return isCompletingMigration = true @@ -1941,7 +1977,10 @@ class AppViewModel @Inject constructor( } activityRepo.markAllUnseenActivitiesAsSeen() - migrationService.consumePendingChannelMigration() + if (!applyPendingChannelMigration()) { + finishMigrationWithError() + return@runCatching + } walletRepo.syncNodeAndWallet() .onSuccess { finishMigrationSuccessfully() } @@ -1964,6 +2003,8 @@ class AppViewModel @Inject constructor( transferRepo.syncTransferStates() migrationService.reapplyMetadataAfterSync() + migrationService.setNeedsPostMigrationSync(false) + migrationService.setRestoringFromRNRemoteBackup(false) if (migrationService.canCleanupAfterMigration) { migrationService.cleanupAfterMigration() migrationService.setShowingMigrationLoading(false) @@ -1983,6 +2024,8 @@ class AppViewModel @Inject constructor( transferRepo.syncTransferStates() migrationService.reapplyMetadataAfterSync() + migrationService.setNeedsPostMigrationSync(false) + migrationService.setRestoringFromRNRemoteBackup(false) if (migrationService.canCleanupAfterMigration) { migrationService.cleanupAfterMigration() migrationService.setShowingMigrationLoading(false) diff --git a/app/src/main/java/to/bitkit/viewmodels/WalletViewModel.kt b/app/src/main/java/to/bitkit/viewmodels/WalletViewModel.kt index b21f0f3515..a64e01ea46 100644 --- a/app/src/main/java/to/bitkit/viewmodels/WalletViewModel.kt +++ b/app/src/main/java/to/bitkit/viewmodels/WalletViewModel.kt @@ -39,6 +39,7 @@ import to.bitkit.repositories.BlocktankRepo import to.bitkit.repositories.ConnectivityRepo import to.bitkit.repositories.ConnectivityState import to.bitkit.repositories.LightningRepo +import to.bitkit.repositories.NodeStartYieldedToStopError import to.bitkit.repositories.PubkyRepo import to.bitkit.repositories.RecoveryModeError import to.bitkit.repositories.SyncSource @@ -181,8 +182,13 @@ class WalletViewModel @Inject constructor( pendingWalletStart = true return@launch } - val channelMigration = buildChannelMigrationIfAvailable() - startNode(0, channelMigration) + migrationService.lockChannelMigration() + try { + val channelMigration = buildChannelMigrationIfAvailable() + startNode(0, channelMigration) + } finally { + migrationService.unlockChannelMigration() + } } else { migrationService.setShowingMigrationLoading(false) } @@ -289,7 +295,17 @@ class WalletViewModel @Inject constructor( fun onRestoreRetry() = viewModelScope.launch(bgDispatcher) { _restoreState.update { it.countRetry() } setInitNodeLifecycleState() - lightningRepo.restartNode() + migrationService.lockChannelMigration() + try { + val channelMigration = buildChannelMigrationIfAvailable() + if (channelMigration == null) { + lightningRepo.restartNode() + } else { + startNode(0, channelMigration, isRestoreRetry = true) + } + } finally { + migrationService.unlockChannelMigration() + } } fun onBackupRestoreRetry() = viewModelScope.launch(bgDispatcher) { @@ -332,8 +348,13 @@ class WalletViewModel @Inject constructor( waitForRestoreIfNeeded() - val channelMigration = buildChannelMigrationIfAvailable() - startNode(walletIndex, channelMigration) + migrationService.lockChannelMigration() + try { + val channelMigration = buildChannelMigrationIfAvailable() + startNode(walletIndex, channelMigration) + } finally { + migrationService.unlockChannelMigration() + } } finally { isStarting = false } @@ -342,9 +363,16 @@ class WalletViewModel @Inject constructor( private suspend fun waitForRestoreIfNeeded() { if (!_restoreState.value.isOngoing()) return - withTimeoutOrNull(TIMEOUT_RESTORE_WAIT) { + val finished = withTimeoutOrNull(TIMEOUT_RESTORE_WAIT) { + _restoreState.first { !it.isOngoing() } + } + if (finished != null) return + if (migrationService.isRestoringFromRNRemoteBackup.value) { + Logger.warn("RN remote restore still running, waiting for channel state", context = TAG) _restoreState.first { !it.isOngoing() } - } ?: Logger.warn("waitForRestoreIfNeeded timeout, proceeding anyway", context = TAG) + return + } + Logger.warn("waitForRestoreIfNeeded timeout, proceeding anyway", context = TAG) } private fun buildChannelMigrationIfAvailable(): ChannelDataMigration? = @@ -358,8 +386,14 @@ class WalletViewModel @Inject constructor( private suspend fun startNode( walletIndex: Int = 0, channelMigration: ChannelDataMigration?, + isRestoreRetry: Boolean = false, ) { - lightningRepo.start(walletIndex, channelMigration = channelMigration) + val result = if (isRestoreRetry) { + lightningRepo.restartNode(channelMigration) + } else { + lightningRepo.start(walletIndex, channelMigration = channelMigration) + } + result .onSuccess { if (channelMigration != null) { migrationService.consumePendingChannelMigration() @@ -375,7 +409,7 @@ class WalletViewModel @Inject constructor( // checkForOrphanedChannelMonitorRecovery() } .onFailure { - if (it is RecoveryModeError || it is WipeInProgressError) { + if (it is RecoveryModeError || it is WipeInProgressError || it is NodeStartYieldedToStopError) { Logger.debug("Skipped node start: '${it.message}'", context = TAG) return@onFailure } diff --git a/app/src/test/java/to/bitkit/repositories/LightningRepoTest.kt b/app/src/test/java/to/bitkit/repositories/LightningRepoTest.kt index e76eb0c097..a83b903709 100644 --- a/app/src/test/java/to/bitkit/repositories/LightningRepoTest.kt +++ b/app/src/test/java/to/bitkit/repositories/LightningRepoTest.kt @@ -29,6 +29,7 @@ import org.junit.Before import org.junit.Test import org.lightningdevkit.ldknode.AddressTypeBalance import org.lightningdevkit.ldknode.BalanceDetails +import org.lightningdevkit.ldknode.ChannelDataMigration import org.lightningdevkit.ldknode.ChannelDetails import org.lightningdevkit.ldknode.Event import org.lightningdevkit.ldknode.Node @@ -741,6 +742,41 @@ class LightningRepoTest : BaseUnitTest() { verifyBlocking(lightningService, times(2)) { start(anyOrNull(), any()) } } + @Test + fun `channel migration restart skips the start retry when a stop is requested during the failed attempt`() = test { + stubNodeForRestart() + val error = AppError("Feerate estimation update timeout") + var attempts = 0 + whenever(lightningService.start(anyOrNull(), any())).thenAnswer { + attempts++ + if (attempts == 1) { + sut.stopDebounced() + throw error + } + Unit + } + + val result = sut.restartNode(ChannelDataMigration(listOf(1.toUByte()), listOf(listOf(2.toUByte())))) + + val yieldError = assertIs(result.exceptionOrNull()) + assertEquals(error, yieldError.cause) + assertEquals(NodeLifecycleState.Stopped, sut.lightningState.value.nodeLifecycleState) + verifyBlocking(lightningService, times(1)) { start(anyOrNull(), any()) } + verify(lightningService, times(1)).stop() + + testScheduler.advanceUntilIdle() + + verify(lightningService, times(2)).stop() + verifyBlocking(lightningService, times(1)) { start(anyOrNull(), any()) } + assertEquals(NodeLifecycleState.Stopped, sut.lightningState.value.nodeLifecycleState) + + val foregroundResult = sut.start() + + assertTrue(foregroundResult.isSuccess) + assertEquals(NodeLifecycleState.Running, sut.lightningState.value.nodeLifecycleState) + verifyBlocking(lightningService, times(2)) { start(anyOrNull(), any()) } + } + @Test fun `restartNode retries when a foreground cancels the stop requested during the failed attempt`() = test { stubNodeForRestart() @@ -794,6 +830,49 @@ class LightningRepoTest : BaseUnitTest() { assertEquals(NodeLifecycleState.Stopped, sut.lightningState.value.nodeLifecycleState) } + @Test + fun `channel migration restart keeps the backup across bounded start attempts`() = test { + stubNodeForRestart() + whenever(lightningService.node).thenReturn(null) + val migration = ChannelDataMigration(listOf(1.toUByte()), listOf(listOf(2.toUByte()))) + var attempts = 0 + whenever(lightningService.start(anyOrNull(), any())).thenAnswer { + attempts++ + if (attempts == 1) throw AppError("transient start failure") + Unit + } + + val result = sut.restartNode(migration) + + assertTrue(result.isSuccess) + verifyBlocking(lightningService, times(2)) { + setup(any(), anyOrNull(), anyOrNull(), anyOrNull(), eq(migration)) + } + assertEquals(NodeLifecycleState.Running, sut.lightningState.value.nodeLifecycleState) + } + + @Test + fun `channel migration restart retry does not cancel a stop requested during the retry`() = test { + stubNodeForRestart() + var attempts = 0 + whenever(lightningService.start(anyOrNull(), any())).thenAnswer { + attempts++ + if (attempts == 1) throw AppError("Feerate estimation update timeout") + sut.stopDebounced() + Unit + } + + val result = sut.restartNode(ChannelDataMigration(listOf(1.toUByte()), listOf(listOf(2.toUByte())))) + + assertTrue(result.isSuccess) + verifyBlocking(lightningService, times(2)) { start(anyOrNull(), any()) } + + testScheduler.advanceUntilIdle() + + verify(lightningService, times(2)).stop() + assertEquals(NodeLifecycleState.Stopped, sut.lightningState.value.nodeLifecycleState) + } + // Regression #1125: a caller cancelled inside the retry delay must not drop the bounded retry @Test fun `restartNodeDetached completes the start retry after the caller is cancelled`() = test { diff --git a/app/src/test/java/to/bitkit/services/MigrationServiceTest.kt b/app/src/test/java/to/bitkit/services/MigrationServiceTest.kt index 40cdd4572f..ec2eb68d8c 100644 --- a/app/src/test/java/to/bitkit/services/MigrationServiceTest.kt +++ b/app/src/test/java/to/bitkit/services/MigrationServiceTest.kt @@ -1,16 +1,92 @@ package to.bitkit.services +import android.content.Context +import com.synonym.bitkitcore.Activity import com.synonym.bitkitcore.IBtOrder +import com.synonym.bitkitcore.OnchainActivity +import com.synonym.bitkitcore.PaymentType import kotlinx.coroutines.test.runTest import org.junit.Test +import org.mockito.kotlin.any import org.mockito.kotlin.doReturn import org.mockito.kotlin.mock +import org.mockito.kotlin.never +import org.mockito.kotlin.verify +import org.mockito.kotlin.whenever +import to.bitkit.ext.create +import to.bitkit.repositories.ActivityRepo import to.bitkit.test.BaseUnitTest +import to.bitkit.utils.AppError import kotlin.test.assertEquals import kotlin.test.assertTrue class MigrationServiceTest : BaseUnitTest() { + @Test + fun `missing transfer markers remain pending until their activity exists`() = test { + val activityRepo = mock() + val service = createService(activityRepo) + val markers = mapOf("transfer" to "channel") + + val pending = service.applyRemoteTransfers(markers) + assertEquals(markers, pending) + verify(activityRepo, never()).updateActivity(any(), any(), any()) + + val activity = onchain("transfer") + whenever(activityRepo.getOnchainActivityByTxId("transfer")).thenReturn(activity) + whenever(activityRepo.updateActivity(any(), any(), any())).thenReturn(Result.success(Unit)) + assertTrue(service.applyRemoteTransfers(pending).isEmpty()) + verify(activityRepo).updateActivity( + "transfer", + Activity.Onchain(activity.copy(isTransfer = true, channelId = "channel")), + ) + } + + @Test + fun `missing boost markers remain pending until their activities exist`() = test { + val activityRepo = mock() + val service = createService(activityRepo) + val markers = mapOf("parent" to "child") + val pending = service.applyBoostTransactions(markers) + assertEquals(markers, pending) + + whenever(activityRepo.getOnchainActivityByTxId("parent")).thenReturn(onchain("parent")) + whenever(activityRepo.getOnchainActivityByTxId("child")).thenReturn(onchain("child")) + whenever(activityRepo.updateActivity(any(), any(), any())).thenReturn(Result.success(Unit)) + assertTrue(service.applyBoostTransactions(pending).isEmpty()) + } + + @Test + fun `failed transfer and boost writes remain pending`() = test { + val activityRepo = mock() + val service = createService(activityRepo) + whenever(activityRepo.getOnchainActivityByTxId(any(), any())).thenReturn(onchain("activity")) + whenever(activityRepo.updateActivity(any(), any(), any())).thenReturn(Result.failure(AppError("write failed"))) + + val transfer = mapOf("transfer" to "channel") + val boost = mapOf("parent" to "child") + assertEquals(transfer, service.applyRemoteTransfers(transfer)) + assertEquals(boost, service.applyBoostTransactions(boost)) + } + + private fun createService(activityRepo: ActivityRepo): MigrationService { + val context = mock() + whenever(context.applicationContext).thenReturn(context) + return MigrationService(context, mock(), mock(), mock(), mock(), activityRepo, mock(), mock(), mock(), mock()) + } + + private fun onchain(id: String) = OnchainActivity.create( + walletId = "wallet0", + id = id, + txType = PaymentType.SENT, + txId = id, + value = 1000uL, + fee = 100uL, + feeRate = 1uL, + address = "address", + timestamp = 1uL, + ) + @Test fun `fetchOrdersInChunks should split ids into chunks and concatenate results`() = runTest { val orderIds = (1..25).map { "order$it" } @@ -51,4 +127,31 @@ class MigrationServiceTest : BaseUnitTest() { assertTrue(requests.isEmpty()) assertTrue(result.isEmpty()) } + + @Test + fun `unapplied metadata keeps only tags whose activity is missing`() { + val metadata = RNMetadata( + tags = mapOf("known" to listOf("sent"), "missing" to listOf("received")), + lastUsedTags = listOf("sent"), + ) + + val remaining = unappliedRnMetadata(metadata, setOf("missing")) + + assertEquals(mapOf("missing" to listOf("received")), remaining?.tags) + assertEquals(null, remaining?.lastUsedTags) + } + + @Test + fun `unapplied metadata is dropped when every tag was applied`() { + val metadata = RNMetadata(tags = mapOf("known" to listOf("sent")), lastUsedTags = listOf("sent")) + + assertEquals(null, unappliedRnMetadata(metadata, emptySet())) + } + + @Test + fun `unapplied metadata is dropped when there are no tags`() { + val metadata = RNMetadata(tags = null, lastUsedTags = listOf("sent")) + + assertEquals(null, unappliedRnMetadata(metadata, setOf("missing"))) + } } diff --git a/app/src/test/java/to/bitkit/ui/WalletViewModelTest.kt b/app/src/test/java/to/bitkit/ui/WalletViewModelTest.kt index 88f4ee7d27..942b155b75 100644 --- a/app/src/test/java/to/bitkit/ui/WalletViewModelTest.kt +++ b/app/src/test/java/to/bitkit/ui/WalletViewModelTest.kt @@ -7,12 +7,15 @@ import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.flowOf import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.test.advanceTimeBy import kotlinx.coroutines.test.advanceUntilIdle import org.junit.Before import org.junit.Test +import org.lightningdevkit.ldknode.ChannelDataMigration import org.lightningdevkit.ldknode.PeerDetails import org.mockito.kotlin.any import org.mockito.kotlin.anyOrNull +import org.mockito.kotlin.argumentCaptor import org.mockito.kotlin.doSuspendableAnswer import org.mockito.kotlin.inOrder import org.mockito.kotlin.mock @@ -32,12 +35,14 @@ import to.bitkit.repositories.ConnectivityRepo import to.bitkit.repositories.ConnectivityState import to.bitkit.repositories.LightningRepo import to.bitkit.repositories.LightningState +import to.bitkit.repositories.NodeStartYieldedToStopError import to.bitkit.repositories.PubkyRepo import to.bitkit.repositories.SyncSource import to.bitkit.repositories.WalletRepo import to.bitkit.repositories.WalletState import to.bitkit.services.BoltzService import to.bitkit.services.MigrationService +import to.bitkit.services.PendingChannelMigration import to.bitkit.test.BaseUnitTest import to.bitkit.utils.AppError import to.bitkit.viewmodels.RestoreState @@ -46,6 +51,7 @@ import kotlin.coroutines.cancellation.CancellationException import kotlin.test.assertEquals import kotlin.test.assertFalse import kotlin.test.assertTrue +import kotlin.time.Duration.Companion.seconds @OptIn(ExperimentalCoroutinesApi::class) class WalletViewModelTest : BaseUnitTest() { @@ -75,6 +81,8 @@ class WalletViewModelTest : BaseUnitTest() { whenever(lightningRepo.lightningState).thenReturn(lightningState) whenever(migrationService.isMigrationChecked()).thenReturn(true) whenever(migrationService.isChannelRecoveryChecked()).thenReturn(true) + whenever(migrationService.isRestoringFromRNRemoteBackup).thenReturn(MutableStateFlow(false)) + whenever(migrationService.isShowingMigrationLoading).thenReturn(MutableStateFlow(false)) whenever(migrationService.tryFetchMigrationPeersFromBackup()).thenReturn(emptyList()) whenever { migrationService.getRNRemoteBackupTimestamp() }.thenReturn(null) whenever { backupRepo.hasPendingWalletRestore() }.thenReturn(false) @@ -661,6 +669,179 @@ class WalletViewModelTest : BaseUnitTest() { verify(testWalletRepo, never()).refreshBip21() } + @Test + fun `start waits out an rn remote restore instead of opening the node`() = test { + val testWalletRepo: WalletRepo = mock() + val testLightningRepo: LightningRepo = mock() + val testWalletState = MutableStateFlow(WalletState(walletExists = true)) + + whenever(migrationService.isRestoringFromRNRemoteBackup).thenReturn(MutableStateFlow(true)) + whenever(testWalletRepo.walletState).thenReturn(testWalletState) + whenever(testWalletRepo.balanceState).thenReturn(balanceState) + whenever(testWalletRepo.walletExists()).thenReturn(true) + whenever(testWalletRepo.restoreWallet(any(), anyOrNull())).thenReturn(Result.success(Unit)) + whenever(testLightningRepo.lightningState).thenReturn(lightningState) + whenever(testLightningRepo.isRecoveryMode).thenReturn(isRecoveryMode) + stubSuccessfulNodeStart(testLightningRepo) + + val testSut = WalletViewModel( + context = context, + bgDispatcher = testDispatcher, + walletRepo = testWalletRepo, + lightningRepo = testLightningRepo, + settingsStore = settingsStore, + backupRepo = backupRepo, + blocktankRepo = blocktankRepo, + pubkyRepo = pubkyRepo, + migrationService = migrationService, + connectivityRepo = connectivityRepo, + boltzService = boltzService, + ) + + testSut.restoreWallet("mnemonic", null) + testSut.start() + advanceTimeBy(31.seconds) + + verifyNodeNotStarted(testLightningRepo) + + testSut.onRestoreContinue() + advanceUntilIdle() + + verifyNodeStarted(testLightningRepo) + } + + @Test + fun `restore retry rebuilds the node with the pending channel file`() = test { + val pending = PendingChannelMigration(channelManager = byteArrayOf(1), channelMonitors = listOf(byteArrayOf(2))) + whenever(migrationService.peekPendingChannelMigration()).thenReturn(pending) + whenever(lightningRepo.restartNode(anyOrNull())).thenReturn(Result.success(Unit)) + + sut.onRestoreRetry() + advanceUntilIdle() + + verifyNodeNotStarted(lightningRepo) + val migration = argumentCaptor() + verify(lightningRepo).restartNode(migration.capture()) + assertEquals(listOf(1.toUByte()), migration.firstValue.channelManager) + assertEquals(listOf(listOf(2.toUByte())), migration.firstValue.channelMonitors) + inOrder(migrationService) { + verify(migrationService).lockChannelMigration() + verify(migrationService).peekPendingChannelMigration() + verify(migrationService).consumePendingChannelMigration() + verify(migrationService).unlockChannelMigration() + } + } + + @Test + fun `restore retry keeps pending channels when restart fails`() = test { + val pending = PendingChannelMigration(channelManager = byteArrayOf(1), channelMonitors = listOf(byteArrayOf(2))) + whenever(migrationService.peekPendingChannelMigration()).thenReturn(pending) + whenever(lightningRepo.restartNode(anyOrNull())).thenReturn(Result.failure(AppError("restart failed"))) + + sut.onRestoreRetry() + advanceUntilIdle() + + verifyNodeNotStarted(lightningRepo) + verify(lightningRepo).restartNode(anyOrNull()) + verify(migrationService, never()).consumePendingChannelMigration() + verify(migrationService).unlockChannelMigration() + } + + @Test + fun `restore retry keeps pending channels when restart yields to background stop`() = test { + val pending = PendingChannelMigration(channelManager = byteArrayOf(1), channelMonitors = listOf(byteArrayOf(2))) + whenever(migrationService.peekPendingChannelMigration()).thenReturn(pending) + whenever(lightningRepo.restartNode(anyOrNull())).thenReturn( + Result.failure(NodeStartYieldedToStopError(AppError("start failed"))), + ) + + sut.onRestoreRetry() + advanceUntilIdle() + + verifyNodeNotStarted(lightningRepo) + verify(lightningRepo).restartNode(anyOrNull()) + verify(migrationService, never()).consumePendingChannelMigration() + verify(migrationService).unlockChannelMigration() + } + + @Test + fun `rn local migration start holds the channel migration lock until consumption`() = test { + advanceUntilIdle() + val pending = PendingChannelMigration(channelManager = byteArrayOf(1), channelMonitors = listOf(byteArrayOf(2))) + whenever(migrationService.isMigrationChecked()).thenReturn(false) + whenever(migrationService.hasRNWalletData()).thenReturn(true) + whenever(migrationService.peekPendingChannelMigration()).thenReturn(pending) + whenever(walletRepo.walletExists()).thenReturn(true) + stubSuccessfulNodeStart(lightningRepo) + + WalletViewModel( + context = context, + bgDispatcher = testDispatcher, + walletRepo = walletRepo, + lightningRepo = lightningRepo, + settingsStore = settingsStore, + backupRepo = backupRepo, + blocktankRepo = blocktankRepo, + pubkyRepo = pubkyRepo, + migrationService = migrationService, + connectivityRepo = connectivityRepo, + boltzService = boltzService, + ) + advanceUntilIdle() + + verifyNodeStarted(lightningRepo) + inOrder(migrationService) { + verify(migrationService).lockChannelMigration() + verify(migrationService).peekPendingChannelMigration() + verify(migrationService).consumePendingChannelMigration() + verify(migrationService).unlockChannelMigration() + } + } + + private suspend fun stubSuccessfulNodeStart(repo: LightningRepo) { + whenever( + repo.start( + any(), + anyOrNull(), + any(), + anyOrNull(), + anyOrNull(), + anyOrNull(), + anyOrNull(), + any(), + ), + ).thenReturn(Result.success(Unit)) + } + + private suspend fun verifyNodeNotStarted(repo: LightningRepo) { + verify( + repo, + never(), + ).start( + any(), + anyOrNull(), + any(), + anyOrNull(), + anyOrNull(), + anyOrNull(), + anyOrNull(), + any(), + ) + } + + private suspend fun verifyNodeStarted(repo: LightningRepo) { + verify(repo).start( + any(), + anyOrNull(), + any(), + anyOrNull(), + anyOrNull(), + anyOrNull(), + anyOrNull(), + any(), + ) + } + private fun stubSettingsUpdate(): MutableStateFlow { val settingsData = MutableStateFlow(SettingsData()) whenever { settingsStore.update(any()) }.thenAnswer { diff --git a/app/src/test/java/to/bitkit/viewmodels/AppViewModelSendFlowTest.kt b/app/src/test/java/to/bitkit/viewmodels/AppViewModelSendFlowTest.kt index 83192292e1..7a03726a21 100644 --- a/app/src/test/java/to/bitkit/viewmodels/AppViewModelSendFlowTest.kt +++ b/app/src/test/java/to/bitkit/viewmodels/AppViewModelSendFlowTest.kt @@ -176,6 +176,7 @@ import to.bitkit.services.CoreService import to.bitkit.services.MigrationService import to.bitkit.services.NodeServiceFgState import to.bitkit.services.PaykitSdkOperationLock.Priority +import to.bitkit.services.PendingChannelMigration import to.bitkit.services.PubkyService import to.bitkit.test.BaseUnitTest import to.bitkit.ui.Routes @@ -6188,6 +6189,75 @@ class AppViewModelSendFlowTest : BaseUnitTest() { assertTrue(settingsData.value.pendingRestoreActivitySeen) } + @Test + fun `background tag retry does not sweep new activities or restart the node`() = test { + whenever(migrationService.isRestoringFromRNRemoteBackup).thenReturn(MutableStateFlow(false)) + whenever(migrationService.needsPostMigrationSync()).thenReturn(false) + whenever(migrationService.hasPendingMigrationRetries()).thenReturn(true) + whenever(migrationService.canCleanupAfterMigration).thenReturn(false) + + emitNodeEvent(Event.SyncCompleted(syncType = SyncType.ONCHAIN_WALLET, syncedBlockHeight = 100u)) + advanceUntilIdle() + + verify(migrationService).reapplyMetadataAfterSync() + verify(activityRepo, never()).markAllUnseenActivitiesAsSeen(anyOrNull()) + verify(lightningRepo, never()).stop() + verify(migrationService, never()).cleanupAfterMigration() + } + + @Test + fun `migration completion rechecks pending channels after startup releases the lock`() = test { + whenever(migrationService.isRestoringFromRNRemoteBackup).thenReturn(MutableStateFlow(false)) + val startup = CompletableDeferred() + whenever(migrationService.lockChannelMigration()).doSuspendableAnswer { startup.await() } + whenever(migrationService.needsPostMigrationSync()).thenReturn(true) + whenever(lightningRepo.getPayments()).thenReturn(Result.success(emptyList())) + whenever(activityRepo.markAllUnseenActivitiesAsSeen()).thenReturn(Result.success(Unit)) + whenever(migrationService.peekPendingChannelMigration()).thenReturn( + PendingChannelMigration(byteArrayOf(1), listOf(byteArrayOf(2))), + ) + + emitNodeEvent(Event.SyncCompleted(syncType = SyncType.ONCHAIN_WALLET, syncedBlockHeight = 100u)) + runCurrent() + verify(lightningRepo, never()).stop() + whenever(migrationService.peekPendingChannelMigration()).thenReturn(null) + whenever(walletRepo.syncNodeAndWallet()).thenReturn(Result.success(Unit)) + startup.complete(Unit) + advanceUntilIdle() + + verify(lightningRepo, never()).stop() + verify(migrationService).unlockChannelMigration() + verify(migrationService).setNeedsPostMigrationSync(false) + } + + @Test + fun `pending channel migration is kept when the node stays running after stop`() = test { + val nodeState = MutableStateFlow(LightningState(nodeLifecycleState = NodeLifecycleState.Running)) + whenever(lightningRepo.lightningState).thenReturn(nodeState) + whenever(lightningRepo.getPayments()).thenReturn(Result.success(emptyList())) + whenever(lightningRepo.stop()).thenReturn(Result.success(Unit)) + whenever(activityRepo.markAllUnseenActivitiesAsSeen()).thenReturn(Result.success(Unit)) + whenever(migrationService.needsPostMigrationSync()).thenReturn(true) + whenever(migrationService.peekPendingChannelMigration()).thenReturn( + PendingChannelMigration(byteArrayOf(1), listOf(byteArrayOf(2))), + ) + + emitNodeEvent(Event.SyncCompleted(syncType = SyncType.ONCHAIN_WALLET, syncedBlockHeight = 100u)) + advanceUntilIdle() + + verify(lightningRepo, never()).start( + any(), + anyOrNull(), + any(), + anyOrNull(), + anyOrNull(), + anyOrNull(), + anyOrNull(), + any(), + ) + verify(migrationService, never()).consumePendingChannelMigration() + } + @Test fun `onchain sync without a pending restore leaves unseen activities untouched`() = test { emitNodeEvent(Event.SyncCompleted(syncType = SyncType.ONCHAIN_WALLET, syncedBlockHeight = 100u)) diff --git a/changelog.d/next/1396.fixed.md b/changelog.d/next/1396.fixed.md new file mode 100644 index 0000000000..8c9aeb3380 --- /dev/null +++ b/changelog.d/next/1396.fixed.md @@ -0,0 +1 @@ +Restoring a React Native wallet waits for its Lightning channel backup, and activity tags that are not ready yet are kept for a later sync.