diff --git a/feature-wallet-impl/src/main/java/io/novafoundation/nova/feature_wallet_impl/data/network/blockchain/assets/balances/statemine/StatemineAssetBalance.kt b/feature-wallet-impl/src/main/java/io/novafoundation/nova/feature_wallet_impl/data/network/blockchain/assets/balances/statemine/StatemineAssetBalance.kt index 8735685c..15860fbe 100644 --- a/feature-wallet-impl/src/main/java/io/novafoundation/nova/feature_wallet_impl/data/network/blockchain/assets/balances/statemine/StatemineAssetBalance.kt +++ b/feature-wallet-impl/src/main/java/io/novafoundation/nova/feature_wallet_impl/data/network/blockchain/assets/balances/statemine/StatemineAssetBalance.kt @@ -131,6 +131,10 @@ class StatemineAssetBalance( ) } + // Deliberately lets setup/subscription failures propagate as exceptions rather than swallowing them into + // emptyFlow()/BalanceSyncUpdate.NoCause: the caller, FullSyncPaymentUpdater.syncAsset(), wraps this whole + // call in a single retryWhen boundary that exists specifically to catch and retry failures like these. If + // we swallow here, that boundary never triggers - the asset silently stops syncing instead of retrying. override suspend fun startSyncingBalance( chain: Chain, chainAsset: Chain.Asset, @@ -138,40 +142,32 @@ class StatemineAssetBalance( accountId: AccountId, subscriptionBuilder: SharedRequestsBuilder ): Flow { - return runCatching { - val runtime = chainRegistry.getRuntime(chain.id) + val runtime = chainRegistry.getRuntime(chain.id) - val statemineType = chainAsset.requireStatemine() - val encodableAssetId = statemineType.prepareIdForEncoding(runtime) + val statemineType = chainAsset.requireStatemine() + val encodableAssetId = statemineType.prepareIdForEncoding(runtime) - val module = runtime.metadata.statemineModule(statemineType) + val module = runtime.metadata.statemineModule(statemineType) - val assetAccountStorage = module.storage("Account") - val assetAccountKey = assetAccountStorage.storageKey(runtime, encodableAssetId, accountId) + val assetAccountStorage = module.storage("Account") + val assetAccountKey = assetAccountStorage.storageKey(runtime, encodableAssetId, accountId) - val assetDetailsFlow = statemineAssetsRepository.subscribeAndSyncAssetDetails(chain.id, statemineType, subscriptionBuilder) + val assetDetailsFlow = statemineAssetsRepository.subscribeAndSyncAssetDetails(chain.id, statemineType, subscriptionBuilder) - combine( - subscriptionBuilder.subscribe(assetAccountKey), - assetDetailsFlow.map { it.status.transfersFrozen } - ) { balanceStorageChange, isAssetFrozen -> - val assetAccountDecoded = assetAccountStorage.decodeValue(balanceStorageChange.value, runtime) - val assetAccount = bindAssetAccountOrEmpty(assetAccountDecoded) + return combine( + subscriptionBuilder.subscribe(assetAccountKey), + assetDetailsFlow.map { it.status.transfersFrozen } + ) { balanceStorageChange, isAssetFrozen -> + val assetAccountDecoded = assetAccountStorage.decodeValue(balanceStorageChange.value, runtime) + val assetAccount = bindAssetAccountOrEmpty(assetAccountDecoded) - val assetChanged = updateAssetBalance(metaAccount.id, chainAsset, isAssetFrozen, assetAccount) + val assetChanged = updateAssetBalance(metaAccount.id, chainAsset, isAssetFrozen, assetAccount) - if (assetChanged) { - BalanceSyncUpdate.CauseFetchable(balanceStorageChange.block) - } else { - BalanceSyncUpdate.NoCause - } - }.catch { error -> - Log.e(LOG_TAG, "Balance sync failed for ${chainAsset.symbol} on ${chain.name}: ${error.message}") - emit(BalanceSyncUpdate.NoCause) + if (assetChanged) { + BalanceSyncUpdate.CauseFetchable(balanceStorageChange.block) + } else { + BalanceSyncUpdate.NoCause } - }.getOrElse { error -> - Log.e(LOG_TAG, "Failed to start balance sync for ${chainAsset.symbol} on ${chain.name}: ${error.message}") - emptyFlow() } } diff --git a/feature-wallet-impl/src/main/java/io/novafoundation/nova/feature_wallet_impl/data/network/blockchain/assets/balances/utility/NativeAssetBalance.kt b/feature-wallet-impl/src/main/java/io/novafoundation/nova/feature_wallet_impl/data/network/blockchain/assets/balances/utility/NativeAssetBalance.kt index f5f14bb4..8768fb15 100644 --- a/feature-wallet-impl/src/main/java/io/novafoundation/nova/feature_wallet_impl/data/network/blockchain/assets/balances/utility/NativeAssetBalance.kt +++ b/feature-wallet-impl/src/main/java/io/novafoundation/nova/feature_wallet_impl/data/network/blockchain/assets/balances/utility/NativeAssetBalance.kt @@ -151,6 +151,9 @@ class NativeAssetBalance( } } + // Setup/subscription failures are allowed to propagate rather than being swallowed into emptyFlow()/NoCause: + // the caller, FullSyncPaymentUpdater.syncAsset(), wraps this whole call in a single retryWhen boundary meant + // to catch and retry exactly these failures. Swallowing here would make that retry boundary never trigger. override suspend fun startSyncingBalance( chain: Chain, chainAsset: Chain.Asset, @@ -160,13 +163,7 @@ class NativeAssetBalance( ): Flow { val runtime = chainRegistry.getRuntime(chain.id) - val key = try { - runtime.metadata.system().storage("Account").storageKey(runtime, accountId) - } catch (e: Exception) { - Log.e(LOG_TAG, "Failed to construct account storage key: ${e.message} in ${chain.name}") - - return emptyFlow() - } + val key = runtime.metadata.system().storage("Account").storageKey(runtime, accountId) return subscriptionBuilder.subscribe(key) .map { change -> @@ -179,10 +176,6 @@ class NativeAssetBalance( BalanceSyncUpdate.NoCause } } - .catch { error -> - Log.e(LOG_TAG, "Balance sync failed for ${chainAsset.symbol} on ${chain.name}: ${error.message}") - emit(BalanceSyncUpdate.NoCause) - } } private fun bindBalanceHolds(dynamicInstance: Any?): List? { diff --git a/runtime/src/main/java/io/novafoundation/nova/runtime/multiNetwork/ChainRegistry.kt b/runtime/src/main/java/io/novafoundation/nova/runtime/multiNetwork/ChainRegistry.kt index 1ae735b7..518d6c49 100644 --- a/runtime/src/main/java/io/novafoundation/nova/runtime/multiNetwork/ChainRegistry.kt +++ b/runtime/src/main/java/io/novafoundation/nova/runtime/multiNetwork/ChainRegistry.kt @@ -41,6 +41,7 @@ import io.novafoundation.nova.runtime.multiNetwork.runtime.types.BaseTypeSynchro import io.novasama.substrate_sdk_android.wsrpc.SocketService import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.SharingStarted import kotlinx.coroutines.flow.distinctUntilChanged @@ -67,14 +68,28 @@ class ChainRegistry( private val runtimeSyncService: RuntimeSyncService, private val web3ApiPool: Web3ApiPool, private val gson: Gson -) : CoroutineScope by CoroutineScope(Dispatchers.Default) { + // SupervisorJob, not the plain Job a bare CoroutineScope(Dispatchers.Default) would give: without it, an + // uncaught exception in ANY coroutine sharing this scope (e.g. currentChains'/chainsById's shareIn, or any + // launch{} below) cancels every sibling, including the other one - a single malformed/leftover chain row + // would then permanently kill sync for every chain, not just the offending one. +) : CoroutineScope by CoroutineScope(SupervisorJob() + Dispatchers.Default) { val currentChains = chainDao.joinChainInfoFlow() .mapList { mapChainLocalToChain(it, gson) } .diffed() .map { diff -> - diff.removed.forEach { unregisterChain(it) } - diff.newOrUpdated.forEach { chain -> registerChain(chain) } + // Each chain's register/unregister is isolated: one malformed/leftover row (e.g. a chain persisted + // as disabled from an earlier session) must not throw out of this operator and kill this flow for + // every other chain - shareIn(..., Eagerly) never restarts once its upstream completes/throws, so + // any single unhandled exception here would silently and permanently break sync for the whole app. + diff.removed.forEach { chain -> + runCatching { unregisterChain(chain) } + .onFailure { Log.e(LOG_TAG, "Failed to unregister chain ${chain.name} (${chain.id})", it) } + } + diff.newOrUpdated.forEach { chain -> + runCatching { registerChain(chain) } + .onFailure { Log.e(LOG_TAG, "Failed to register chain ${chain.name} (${chain.id})", it) } + } diff.all } diff --git a/runtime/src/main/java/io/novafoundation/nova/runtime/network/updaters/ChainUpdaterGroupUpdateSystem.kt b/runtime/src/main/java/io/novafoundation/nova/runtime/network/updaters/ChainUpdaterGroupUpdateSystem.kt index 87e7df49..3d2f0b4b 100644 --- a/runtime/src/main/java/io/novafoundation/nova/runtime/network/updaters/ChainUpdaterGroupUpdateSystem.kt +++ b/runtime/src/main/java/io/novafoundation/nova/runtime/network/updaters/ChainUpdaterGroupUpdateSystem.kt @@ -14,7 +14,9 @@ import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.catch import kotlinx.coroutines.flow.emptyFlow +import kotlinx.coroutines.flow.emitAll import kotlinx.coroutines.flow.flatMapLatest +import kotlinx.coroutines.flow.flow import kotlinx.coroutines.flow.flowOn import kotlinx.coroutines.flow.merge import kotlin.coroutines.coroutineContext @@ -24,35 +26,48 @@ abstract class ChainUpdaterGroupUpdateSystem( private val storageSharedRequestsBuilderFactory: StorageSharedRequestsBuilderFactory, ) : UpdateSystem { + // Callers (MultiChainUpdateSystem, SingleChainUpdateSystem) merge several chains' runUpdaters() results + // into one flow. chainRegistry.getRuntime(chain.id) throws for a chain whose runtime isn't ready yet + // (including a disabled chain, via DisabledChainException) - if that throw escapes this function + // uncaught, it propagates through the merge and kills governance/staking/crowdloan sync for every OTHER + // chain in the group too, not just the failing one. Wrapping the whole body in flow{} + catch isolates + // that failure to this chain alone. protected suspend fun runUpdaters(chain: Chain, updaters: Collection>): Flow { - val runtimeMetadata = chainRegistry.getRuntime(chain.id).metadata + return flow { + val runtimeMetadata = chainRegistry.getRuntime(chain.id).metadata - val logTag = this@ChainUpdaterGroupUpdateSystem.LOG_TAG - val selfName = this@ChainUpdaterGroupUpdateSystem::class.java.simpleName + val logTag = this@ChainUpdaterGroupUpdateSystem.LOG_TAG + val selfName = this@ChainUpdaterGroupUpdateSystem::class.java.simpleName - val scopeFlows = updaters.groupBy(Updater<*>::scope).map { (scope, scopeUpdaters) -> - scope.invalidationFlow().flatMapLatest { scopeValue -> - val subscriptionBuilder = storageSharedRequestsBuilderFactory.create(chain.id) + val scopeFlows = updaters.groupBy(Updater<*>::scope).map { (scope, scopeUpdaters) -> + scope.invalidationFlow().flatMapLatest { scopeValue -> + val subscriptionBuilder = storageSharedRequestsBuilderFactory.create(chain.id) - val updatersFlow = scopeUpdaters - .filter { it.requiredModules.all(runtimeMetadata::hasModule) } - .map { updater -> - @Suppress("UNCHECKED_CAST") - (updater as Updater).listenForUpdates(subscriptionBuilder, scopeValue) - .catch { Log.e(logTag, "Failed to start ${updater.javaClass.simpleName} in $selfName for ${chain.name}", it) } - .flowOn(Dispatchers.Default) + val updatersFlow = scopeUpdaters + .filter { it.requiredModules.all(runtimeMetadata::hasModule) } + .map { updater -> + @Suppress("UNCHECKED_CAST") + (updater as Updater).listenForUpdates(subscriptionBuilder, scopeValue) + .catch { Log.e(logTag, "Failed to start ${updater.javaClass.simpleName} in $selfName for ${chain.name}", it) } + .flowOn(Dispatchers.Default) + } + + if (updatersFlow.isNotEmpty()) { + subscriptionBuilder.subscribe(coroutineContext) + + updatersFlow.merge() + } else { + emptyFlow() } - - if (updatersFlow.isNotEmpty()) { - subscriptionBuilder.subscribe(coroutineContext) - - updatersFlow.merge() - } else { - emptyFlow() } } - } - return scopeFlows.merge() + emitAll(scopeFlows.merge()) + }.catch { error -> + // Explicitly qualified: unqualified LOG_TAG here would resolve against the nearest implicit + // receiver, which is this catch lambda's FlowCollector, not this class - Any.LOG_TAG applies to + // any receiver, so it would silently compile but log the wrong (unhelpful) tag. + Log.e(this@ChainUpdaterGroupUpdateSystem.LOG_TAG, "Failed to start updaters in ${this@ChainUpdaterGroupUpdateSystem::class.java.simpleName} for ${chain.name}", error) + } } }