fix: stop swallowed errors from silently killing balance/updater sync

StatemineAssetBalance and NativeAssetBalance's startSyncingBalance()
wrapped setup and subscription failures in runCatching{}/catch{} that
returned emptyFlow()/NoCause instead of propagating. This meant
FullSyncPaymentUpdater's retryWhen (added for the Tron fix) never
actually engaged for these asset types - a transient failure silently
and permanently stopped sync for that asset (HEZ, PEZ, USDT, DOT, ...)
with only a logcat line as evidence. Let the exceptions propagate so
the existing retry boundary can do its job.

ChainRegistry.currentChains ran registerChain()/unregisterChain() per
chain, unguarded, inside a plain (non-Supervisor) CoroutineScope and an
Eagerly-shared flow that never restarts once it dies. One bad chain row
could permanently kill sync for every chain. Isolated each chain's
register/unregister in its own runCatching, and switched the scope to
a SupervisorJob as defense in depth.

ChainUpdaterGroupUpdateSystem.runUpdaters() called getRuntime() and
built per-chain updater flows with no error boundary; callers merge
several chains' results together, so one chain's failure (e.g. a
disabled chain throwing DisabledChainException) could kill
governance/staking/crowdloan sync for every other chain in the group.
Wrapped the body in flow{} + catch to isolate failures per chain.
This commit is contained in:
2026-07-08 13:54:44 -07:00
parent 4aebac305f
commit 7906a20f96
4 changed files with 81 additions and 62 deletions
@@ -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
}
@@ -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<Updater<*>>): Flow<Updater.SideEffect> {
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<Any?>).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<Any?>).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)
}
}
}