diff --git a/app/src/main/java/to/bitkit/repositories/LightningRepo.kt b/app/src/main/java/to/bitkit/repositories/LightningRepo.kt index 74dd9b49a..b3544cac2 100644 --- a/app/src/main/java/to/bitkit/repositories/LightningRepo.kt +++ b/app/src/main/java/to/bitkit/repositories/LightningRepo.kt @@ -18,6 +18,7 @@ import kotlinx.collections.immutable.persistentListOf import kotlinx.collections.immutable.toImmutableList import kotlinx.coroutines.CoroutineDispatcher import kotlinx.coroutines.Job +import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableStateFlow @@ -71,6 +72,7 @@ import to.bitkit.ext.toPeerDetailsList import to.bitkit.ext.totalNextOutboundHtlcLimitSats import to.bitkit.models.ALL_ADDRESS_TYPE_STRINGS import to.bitkit.models.CoinSelectionPreference +import to.bitkit.models.ElectrumServer import to.bitkit.models.NATIVE_WITNESS_TYPES import to.bitkit.models.NodeLifecycleState import to.bitkit.models.OpenChannelResult @@ -84,6 +86,7 @@ import to.bitkit.models.toCoreNetwork import to.bitkit.models.toSettingsString import to.bitkit.services.AddressDerivationInfo import to.bitkit.services.CoreService +import to.bitkit.services.ElectrumProbeService import to.bitkit.services.LightningService import to.bitkit.services.LnurlChannelResponse import to.bitkit.services.LnurlService @@ -123,6 +126,7 @@ class LightningRepo @Inject constructor( private val connectivityRepo: ConnectivityRepo, private val vssBackupClientLdk: VssBackupClientLdk, private val urlValidator: UrlValidator, + private val electrumProbeService: ElectrumProbeService, ) { private val _lightningState = MutableStateFlow(LightningState()) val lightningState = _lightningState.asStateFlow() @@ -144,7 +148,18 @@ class LightningRepo @Inject constructor( private val syncMutex = Mutex() private val syncPending = AtomicBoolean(false) private val syncRetryJob = AtomicReference(null) + private val pendingStopJob = AtomicReference(null) + private val pendingStopLock = Any() private val lifecycleMutex = Mutex() + + /** + * Serializes a whole server-change transaction (stop, start, persist) against the background + * recovery a failed change launches. [lifecycleMutex] is taken separately by `stop()` and + * `start()`, so without this a detached recovery can restart the previous config between a + * request's stop and start, leaving the request to report and persist a config that never + * started. Recovery is cheap to wait on: it skips the release gate, so it is a plain rebuild. + */ + private val configChangeMutex = Mutex() private val isChangingAddressType = AtomicBoolean(false) init { @@ -310,6 +325,8 @@ class LightningRepo @Inject constructor( return@withContext Result.failure(RecoveryModeError()) } + cancelPendingStop() + eventHandler?.let { _eventHandlers.add(it) } // Track retry state outside mutex to avoid deadlock (Mutex is non-reentrant) @@ -320,9 +337,11 @@ class LightningRepo @Inject constructor( val result = lifecycleMutex.withLock { initialLifecycleState = _lightningState.value.nodeLifecycleState if (initialLifecycleState.isRunningOrStarting()) { - Logger.info("LDK node start skipped, lifecycle state: $initialLifecycleState", context = TAG) - lightningService.startEventListener(::onEvent) - return@withLock Result.success(Unit) + return@withLock skipStartForRunningNode( + lifecycleState = initialLifecycleState, + customServerUrl = customServerUrl, + customRgsServerUrl = customRgsServerUrl, + ) } runCatching { @@ -330,7 +349,8 @@ class LightningRepo @Inject constructor( // Setup if needed if (lightningService.node == null) { - val setupResult = setup(walletIndex, customServerUrl, customRgsServerUrl, channelMigration) + val setupResult = + setup(walletIndex, customServerUrl, customRgsServerUrl, channelMigration) if (setupResult.isFailure) { _lightningState.update { it.copy( @@ -435,6 +455,23 @@ class LightningRepo @Inject constructor( result } + private suspend fun skipStartForRunningNode( + lifecycleState: NodeLifecycleState, + customServerUrl: String?, + customRgsServerUrl: String?, + ): Result { + if (customServerUrl != null || customRgsServerUrl != null) { + // A node that is already up was not built with this config, so reporting success would + // let the caller persist a server that never started. + Logger.warn("Skipped LDK node start with custom config, state: $lifecycleState", context = TAG) + return Result.failure(NodeConfigNotAppliedError()) + } + + Logger.info("LDK node start skipped, lifecycle state: $lifecycleState", context = TAG) + lightningService.startEventListener(::onEvent) + return Result.success(Unit) + } + fun removeEventHandler(handler: NodeEventHandler) { _eventHandlers.remove(handler) } @@ -537,6 +574,24 @@ class LightningRepo @Inject constructor( _lightningState.update { it.copy(nodeLifecycleState = NodeLifecycleState.Initializing) } } + /** + * Defers [stop] so a brief background/foreground cycle does not tear the node down and rebuild it. + * Runs on the repo scope so a cancelled ViewModel cannot drop the pending stop. + * + * Scheduling and cancelling are atomic: [cancelPendingStop] runs on the repo dispatcher while this + * runs on the caller thread, so an interleaved cancel could otherwise miss the job being installed + * and stop the node after the app is back in the foreground. + */ + fun stopDebounced() = synchronized(pendingStopLock) { + val job = scope.launch { + delay(BACKGROUND_STOP_DELAY) + stop() + } + pendingStopJob.getAndSet(job)?.cancel() + } + + fun cancelPendingStop() = synchronized(pendingStopLock) { pendingStopJob.getAndSet(null)?.cancel() } + suspend fun stop(): Result = withContext(bgDispatcher) { lifecycleMutex.withLock { if (_lightningState.value.nodeLifecycleState.isStoppedOrStopping()) { @@ -545,10 +600,12 @@ class LightningRepo @Inject constructor( } runCatching { - _lightningState.update { it.copy(nodeLifecycleState = NodeLifecycleState.Stopping) } - lightningService.stop() - clearProbeOutcomes() - _lightningState.update { LightningState(nodeLifecycleState = NodeLifecycleState.Stopped) } + withContext(NonCancellable) { + _lightningState.update { it.copy(nodeLifecycleState = NodeLifecycleState.Stopping) } + lightningService.stop() + clearProbeOutcomes() + _lightningState.update { LightningState(nodeLifecycleState = NodeLifecycleState.Stopped) } + } }.onFailure { Logger.error("Node stop error", it, context = TAG) // On failure, check actual node state and update accordingly @@ -745,24 +802,33 @@ class LightningRepo @Inject constructor( suspend fun restartWithElectrumServer(newServerUrl: String): Result = withContext(bgDispatcher) { Logger.info("Changing ldk-node electrum server to: '$newServerUrl'", context = TAG) - waitForNodeToStop().onFailure { return@withContext Result.failure(it) } - stop().onFailure { - Logger.error("Failed to stop node during electrum server change", it, context = TAG) + validateElectrumServer(newServerUrl).onFailure { + Logger.warn("Rejected electrum server '$newServerUrl'", it, context = TAG) return@withContext Result.failure(it) } - Logger.debug("Starting node with new electrum server: '$newServerUrl'", context = TAG) + configChangeMutex.withLock { + waitForNodeToStop().onFailure { return@withContext Result.failure(it) } + stop().onFailure { + Logger.error("Failed to stop node during electrum server change", it, context = TAG) + return@withContext Result.failure(it) + } - start( - shouldRetry = false, - customServerUrl = newServerUrl, - ).onFailure { - Logger.warn("Failed ldk-node config change, attempting recovery…", context = TAG) - restartWithPreviousConfig() - }.onSuccess { - settingsStore.update { it.copy(electrumServer = newServerUrl) } + Logger.debug("Starting node with new electrum server: '$newServerUrl'", context = TAG) - Logger.info("Successfully changed electrum server", context = TAG) + start( + shouldRetry = false, + customServerUrl = newServerUrl, + ).onFailure { + // Recover in the background: a wedged node's release can gate the rebuild for tens of + // seconds, and the caller must surface this failure now rather than block on recovery. + Logger.warn("Failed ldk-node config change, recovering in background…", context = TAG) + scope.launch { restartWithPreviousConfig() } + }.onSuccess { + settingsStore.update { it.copy(electrumServer = newServerUrl) } + + Logger.info("Successfully changed electrum server", context = TAG) + } } } @@ -774,24 +840,28 @@ class LightningRepo @Inject constructor( return@withContext Result.failure(it) } - waitForNodeToStop().onFailure { return@withContext Result.failure(it) } - stop().onFailure { - Logger.error("Failed to stop node during RGS server change", it, context = TAG) - return@withContext Result.failure(it) - } - - Logger.debug("Starting node with new RGS server: '$newRgsUrl'", context = TAG) + configChangeMutex.withLock { + waitForNodeToStop().onFailure { return@withContext Result.failure(it) } + stop().onFailure { + Logger.error("Failed to stop node during RGS server change", it, context = TAG) + return@withContext Result.failure(it) + } - start( - shouldRetry = false, - customRgsServerUrl = newRgsUrl, - ).onFailure { - Logger.warn("Failed ldk-node config change, attempting recovery…", context = TAG) - restartWithPreviousConfig() - }.onSuccess { - settingsStore.update { it.copy(rgsServerUrl = newRgsUrl) } + Logger.debug("Starting node with new RGS server: '$newRgsUrl'", context = TAG) - Logger.info("Successfully changed RGS server", context = TAG) + start( + shouldRetry = false, + customRgsServerUrl = newRgsUrl, + ).onFailure { + // Recover in the background: a wedged node's release can gate the rebuild for tens of + // seconds, and the caller must surface this failure now rather than block on recovery. + Logger.warn("Failed ldk-node config change, recovering in background…", context = TAG) + scope.launch { restartWithPreviousConfig() } + }.onSuccess { + settingsStore.update { it.copy(rgsServerUrl = newRgsUrl) } + + Logger.info("Successfully changed RGS server", context = TAG) + } } } @@ -801,6 +871,14 @@ class LightningRepo @Inject constructor( urlValidator.validate(testUrl) } + private suspend fun validateElectrumServer(url: String): Result = withContext(bgDispatcher) { + runSuspendCatching { ElectrumServer.parse(url) } + .fold( + onSuccess = { electrumProbeService.probe(it) }, + onFailure = { Result.failure(it) }, + ) + } + suspend fun getBalanceForAddressType(addressType: AddressType): Result = withContext(bgDispatcher) { executeWhenNodeRunning("getBalanceForAddressType") { runCatching { @@ -981,21 +1059,23 @@ class LightningRepo @Inject constructor( } private suspend fun restartWithPreviousConfig(): Result = withContext(bgDispatcher) { - Logger.debug("Stopping node for recovery attempt", context = TAG) - - stop().onFailure { e -> - Logger.error("Failed to stop node during recovery", e, context = TAG) - return@withContext Result.failure(e) - } + // Runs detached from the failed change, so it takes the same transaction lock: without it + // this recovery can restart the previous config between a later change's stop and start. + configChangeMutex.withLock { + Logger.debug("Stopping node for recovery attempt", context = TAG) + + stop().onFailure { e -> + Logger.error("Failed to stop node during recovery", e, context = TAG) + return@withContext Result.failure(e) + } - Logger.debug("Starting node with previous config for recovery", context = TAG) + Logger.debug("Starting node with previous config for recovery", context = TAG) - start( - shouldRetry = false, - ).onSuccess { - Logger.debug("Successfully started node with previous config", context = TAG) - }.onFailure { - Logger.error("Failed starting node with previous config", it, context = TAG) + start(shouldRetry = false).onSuccess { + Logger.debug("Successfully started node with previous config", context = TAG) + }.onFailure { + Logger.error("Failed starting node with previous config", it, context = TAG) + } } } @@ -1767,6 +1847,9 @@ class LightningRepo @Inject constructor( check(lifecycleState == NodeLifecycleState.Stopped) { "Node lifecycle changed to '$lifecycleState' during pathfinding scores reset" } + // Gate the destructive VSS deletes on the previous node's release, so the old node cannot + // re-persist scores over the delete while it is still draining. + lightningService.awaitNodeRelease() vssBackupClientLdk.setup(walletIndex).getOrThrow() vssBackupClientLdk.deleteObject(VSS_KEY_SCORER).getOrThrow() vssBackupClientLdk.deleteObject(VSS_KEY_EXTERNAL_SCORES_CACHE).getOrThrow() @@ -1809,6 +1892,7 @@ class LightningRepo @Inject constructor( private const val VSS_KEY_EXTERNAL_SCORES_CACHE = "external_pathfinding_scores_cache" private const val MS_SYNC_LOOP_DEBOUNCE = 500L private const val SYNC_RETRY_DELAY_MS = 15_000L + private val BACKGROUND_STOP_DELAY = 3.seconds private val CHANNELS_USABLE_TIMEOUT = 15.seconds private val NO_USABLE_CHANNELS_FEEDBACK_DELAY = 2_500.milliseconds val SEND_LN_TIMEOUT = 10.seconds @@ -1819,6 +1903,7 @@ class LightningRepo @Inject constructor( class RecoveryModeError : AppError("App in recovery mode, skipping node start") class NodeSetupError : AppError("Unknown node setup error") class NodeStopTimeoutError : AppError("Timeout waiting for node to stop") +class NodeConfigNotAppliedError : AppError("Node already running, requested config was not applied") class NodeRunTimeoutError(opName: String) : AppError("Timeout waiting for node to run and execute: '$opName'") class GetPaymentsError : AppError("It wasn't possible get the payments") class SyncUnhealthyError : AppError("Wallet sync failed before send") diff --git a/app/src/main/java/to/bitkit/services/ElectrumProbeService.kt b/app/src/main/java/to/bitkit/services/ElectrumProbeService.kt new file mode 100644 index 000000000..85cb1e6a4 --- /dev/null +++ b/app/src/main/java/to/bitkit/services/ElectrumProbeService.kt @@ -0,0 +1,203 @@ +package to.bitkit.services + +import kotlinx.coroutines.CoroutineDispatcher +import kotlinx.coroutines.withContext +import kotlinx.serialization.Serializable +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.JsonElement +import kotlinx.serialization.json.JsonNull +import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.contentOrNull +import kotlinx.serialization.json.jsonPrimitive +import org.lightningdevkit.ldknode.Network +import to.bitkit.di.IoDispatcher +import to.bitkit.env.Env +import to.bitkit.ext.runSuspendCatching +import to.bitkit.models.ElectrumProtocol +import to.bitkit.models.ElectrumServer +import to.bitkit.utils.AppError +import to.bitkit.utils.Logger +import java.io.BufferedReader +import java.io.Writer +import java.net.InetSocketAddress +import java.net.Socket +import javax.inject.Inject +import javax.inject.Singleton +import javax.net.ssl.SSLSocket +import javax.net.ssl.SSLSocketFactory +import kotlin.time.Duration.Companion.seconds + +/** + * Probes an electrum server over its own socket before the node is asked to use it. + * + * A failed `node.start()` leaves the node's electrum background tasks wedged, so `free_node` then + * blocks for tens of seconds instead of milliseconds. Rejecting a misconfigured server here means + * the node is never torn down and rebuilt for one, so that path stops producing wedged releases. + */ +@Singleton +class ElectrumProbeService @Inject constructor( + @IoDispatcher private val ioDispatcher: CoroutineDispatcher, +) { + companion object { + private const val TAG = "ElectrumProbeService" + + /** Budget for the TCP connect and, on SSL, the TLS handshake. */ + private val CONNECT_TIMEOUT = 5.seconds + + /** Budget for each JSON-RPC response line. */ + private val RESPONSE_TIMEOUT = 5.seconds + + private const val CLIENT_NAME = "bitkit" + private const val PROTOCOL_VERSION = "1.4" + + /** JSON-RPC id of the `server.version` request, echoed back by a well-behaved server. */ + private const val VERSION_REQUEST_ID = 0 + + /** JSON-RPC id of the `server.features` request. */ + private const val FEATURES_REQUEST_ID = 1 + } + + // Deliberately not the injected Json: that one sets prettyPrint, and electrum is line-delimited, + // so a multi-line request would be read as a truncated line. encodeDefaults keeps `jsonrpc` and + // an empty `params` on the wire, which servers expect. + private val json = Json { + ignoreUnknownKeys = true + encodeDefaults = true + } + + suspend fun probe( + server: ElectrumServer, + network: Network = Env.network, + ): Result = withContext(ioDispatcher) { + runSuspendCatching { + openSocket(server).use { socket -> + socket.soTimeout = RESPONSE_TIMEOUT.inWholeMilliseconds.toInt() + val reader = socket.getInputStream().bufferedReader() + val writer = socket.getOutputStream().bufferedWriter() + + // A host that accepts the connection but does not speak electrum drops or resets it + // mid-exchange, so report that as a probe verdict rather than a raw socket error. + runCatching { + // Version negotiation has to succeed before server.features may be treated as + // optional, otherwise a server erroring on both would probe clean. + request(reader, writer, VERSION_REQUEST_ID, "server.version", versionParams()).getOrThrow() + + val features = request(reader, writer, FEATURES_REQUEST_ID, "server.features") + verifyNetwork(features.getOrNull(), server, network) + }.getOrElse { + throw it as? ElectrumProbeError ?: ElectrumProbeError.NotElectrum(server, it) + } + } + Logger.info("Probed electrum server '$server' successfully", context = TAG) + } + } + + private fun openSocket(server: ElectrumServer): Socket { + val plain = Socket() + runCatching { + plain.connect(InetSocketAddress(server.host, server.getPort()), CONNECT_TIMEOUT.inWholeMilliseconds.toInt()) + }.onFailure { + plain.runCatching { close() } + throw ElectrumProbeError.Unreachable(server, it) + } + + if (server.protocol == ElectrumProtocol.TCP) return plain + + // A TLS handshake against a plain-TCP server hangs without a read timeout, which is the + // misconfiguration that wedges the node's release when it is left to node.start(). + return runCatching { + val factory = SSLSocketFactory.getDefault() as SSLSocketFactory + val ssl = factory.createSocket(plain, server.host, server.getPort(), true) as SSLSocket + ssl.soTimeout = CONNECT_TIMEOUT.inWholeMilliseconds.toInt() + ssl.startHandshake() + ssl + }.getOrElse { + plain.runCatching { close() } + throw ElectrumProbeError.ProtocolMismatch(server, it) + } + } + + private fun request( + reader: BufferedReader, + writer: Writer, + id: Int, + method: String, + params: List = emptyList(), + ): Result = runCatching { + writer.appendLine(json.encodeToString(RpcRequest(id = id, method = method, params = params))) + writer.flush() + + val line = reader.readLine() ?: throw AppError("Closed connection before answering '$method'") + val response = json.decodeFromString(line) + + when { + response.id != id -> + throw AppError("Answered '$method' with id '${response.id}', expected '$id'") + + !response.error.isNullOrJsonNull() -> + throw AppError("Answered '$method' with error '${response.error}'") + + response.result.isNullOrJsonNull() -> + throw AppError("Answered '$method' without a result") + + else -> response + } + } + + private fun verifyNetwork(features: RpcResponse?, server: ElectrumServer, network: Network) { + val genesis = (features?.result as? JsonObject)?.get("genesis_hash")?.jsonPrimitive?.contentOrNull + if (genesis == null) { + // server.features is optional, so a server that answered version negotiation but cannot + // report a genesis hash stays usable; only the network check is skipped. + Logger.warn("Skipped network check, server '$server' reported no genesis hash", context = TAG) + return + } + + val expected = genesisHashOf(network) + if (!genesis.equals(expected, ignoreCase = true)) { + throw ElectrumProbeError.NetworkMismatch(server, expected = expected, actual = genesis) + } + } + + private fun versionParams() = listOf(CLIENT_NAME, PROTOCOL_VERSION) + + private fun genesisHashOf(network: Network): String = when (network) { + Network.BITCOIN -> "000000000019d6689c085ae165831e934ff763ae46a2a6c172b3f1b60a8ce26f" + Network.TESTNET -> "000000000933ea01ad0ee984209779baaec3ced90fa3f408719526f8d77f4943" + Network.SIGNET -> "00000008819873e925422c1ff0f99f7cc9bbb232af63a077a480a3633bee1ef6" + Network.REGTEST -> "0f9188f13cb7b2c71f2a335e3a4fc328bf5beb436012afca590b1a11466e2206" + } +} + +/** A JSON-RPC call, matching the envelope the probe expects back. */ +@Serializable +private data class RpcRequest( + val id: Int, + val jsonrpc: String = "2.0", + val method: String, + val params: List = emptyList(), +) + +/** A JSON-RPC reply, kept only as far as the probe needs to trust it. */ +@Serializable +private data class RpcResponse( + val id: Int? = null, + val result: JsonElement? = null, + val error: JsonElement? = null, +) + +private fun JsonElement?.isNullOrJsonNull() = this == null || this is JsonNull + +sealed class ElectrumProbeError(message: String, cause: Throwable? = null) : AppError(message, cause) { + class Unreachable(server: ElectrumServer, cause: Throwable) : + ElectrumProbeError("Could not reach electrum server '$server'", cause) + + class ProtocolMismatch(server: ElectrumServer, cause: Throwable) : + ElectrumProbeError("Failed TLS handshake with electrum server '$server', check the protocol", cause) + + class NotElectrum(server: ElectrumServer, cause: Throwable? = null) : + ElectrumProbeError("Received no electrum response from '$server'", cause) + + class NetworkMismatch(server: ElectrumServer, expected: String, actual: String) : + ElectrumProbeError("Rejected electrum server '$server' on wrong network, expected '$expected' got '$actual'") +} diff --git a/app/src/main/java/to/bitkit/services/LightningService.kt b/app/src/main/java/to/bitkit/services/LightningService.kt index 75419bda3..f0b92f5b0 100644 --- a/app/src/main/java/to/bitkit/services/LightningService.kt +++ b/app/src/main/java/to/bitkit/services/LightningService.kt @@ -3,7 +3,9 @@ package to.bitkit.services import com.synonym.bitkitcore.AddressType import kotlinx.coroutines.CoroutineDispatcher import kotlinx.coroutines.Job +import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.cancelAndJoin +import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.ensureActive import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.SharedFlow @@ -12,6 +14,7 @@ import kotlinx.coroutines.flow.first import kotlinx.coroutines.launch import kotlinx.coroutines.withContext import kotlinx.coroutines.withTimeout +import kotlinx.coroutines.withTimeoutOrNull import kotlinx.serialization.Serializable import org.lightningdevkit.ldknode.Address import org.lightningdevkit.ldknode.AddressTypeBalance @@ -48,8 +51,10 @@ import to.bitkit.data.SettingsStore import to.bitkit.data.backup.VssStoreIdProvider import to.bitkit.data.keychain.Keychain import to.bitkit.di.BgDispatcher +import to.bitkit.di.IoDispatcher import to.bitkit.env.Defaults import to.bitkit.env.Env +import to.bitkit.ext.runSuspendCatching import to.bitkit.ext.uByteList import to.bitkit.ext.uri import to.bitkit.models.OpenChannelResult @@ -65,13 +70,21 @@ import to.bitkit.utils.jsonLogOf import java.io.File import javax.inject.Inject import javax.inject.Singleton +import kotlin.coroutines.AbstractCoroutineContextElement +import kotlin.coroutines.CoroutineContext import kotlin.coroutines.cancellation.CancellationException import kotlin.io.path.Path import kotlin.time.Duration +import kotlin.time.Duration.Companion.seconds import org.lightningdevkit.ldknode.AddressType as LdkAddressType typealias NodeEventHandler = suspend (Event) -> Unit +/** Tags the event-listener coroutine so a handler-triggered stop can skip joining its own job. */ +private class EventListenerContext : AbstractCoroutineContextElement(Key) { + companion object Key : CoroutineContext.Key +} + data class AddressDerivationInfo( val address: String, val index: Int, @@ -81,6 +94,7 @@ data class AddressDerivationInfo( @Singleton class LightningService @Inject constructor( @BgDispatcher private val bgDispatcher: CoroutineDispatcher, + @IoDispatcher private val ioDispatcher: CoroutineDispatcher, private val keychain: Keychain, private val vssStoreIdProvider: VssStoreIdProvider, private val settingsStore: SettingsStore, @@ -97,6 +111,15 @@ class LightningService @Inject constructor( private const val SCORING_HISTORICAL_LIQUIDITY_PENALTY_AMOUNT_MULTIPLIER_MSAT = 20_000uL private const val SCORING_CONSIDERED_IMPOSSIBLE_PENALTY_MSAT = 1_000_000_000_000uL + /** How long a node stop waits for the rust handle to be released before moving on. */ + private val NODE_RELEASE_TIMEOUT = 1.seconds + + /** + * Upper bound for the release gate. Longer than an observed wedged free_node (~40s) so a real + * drain still gates, but bounded so a stuck free_node cannot permanently block rebuilds. + */ + private val NODE_RELEASE_GATE_TIMEOUT = 90.seconds + private val DEFAULT_SCORING_FEE_PARAMETERS = ScoringFeeParameters( basePenaltyMsat = 1_024uL, basePenaltyAmountMultiplierMsat = 131_072uL, @@ -121,6 +144,11 @@ class LightningService @Inject constructor( private var listenerJob: Job? = null + // Release of the previous node's rust handle (free_node). Rebuild and destructive storage work + // wait on this so a new node never touches storage the old one is still draining. + @Volatile + private var releaseJob: Job? = null + suspend fun setup( walletIndex: Int, customServerUrl: String? = null, @@ -128,6 +156,7 @@ class LightningService @Inject constructor( trustedPeers: List? = null, channelMigration: ChannelDataMigration? = null, ) { + awaitNodeRelease() Logger.debug("Building node…", context = TAG) val config = config(walletIndex, trustedPeers) @@ -295,7 +324,7 @@ class LightningService @Inject constructor( // start event listener after node started onEvent?.let { eventHandler -> shouldListenForEvents = true - listenerJob = launch { + listenerJob = launch(EventListenerContext()) { runCatching { Logger.debug("LDK event listener started", context = TAG) if (timeout != null) { @@ -314,34 +343,84 @@ class LightningService @Inject constructor( Logger.info("Node started", context = TAG) } - suspend fun stop() { + // Teardown must not be abandoned midway: a cancelled caller would leave the rust node alive and + // let the GC free it later on the finalizer thread, racing the next node. This covers the whole + // body, including the listener join, so cancellation during listener cleanup cannot skip it. + suspend fun stop() = withContext(NonCancellable) { shouldListenForEvents = false - listenerJob?.cancelAndJoin() + // A stop requested from inside an event handler runs on the listener job itself; joining it + // here would deadlock, so let the loop exit on shouldListenForEvents instead. + if (currentCoroutineContext()[EventListenerContext.Key] == null) { + listenerJob?.cancelAndJoin() + } listenerJob = null - val node = this.node ?: run { + val node = this@LightningService.node ?: run { Logger.debug("Node already stopped", context = TAG) - return + return@withContext } Logger.debug("Stopping node…", context = TAG) ServiceQueue.LDK.background { - runCatching { node.stop() } - .onFailure { if (it !is NodeException.NotRunning) throw it } + runSuspendCatching { node.stop() } + .onFailure { + if (it !is NodeException.NotRunning) Logger.warn("Node stop error", it, context = TAG) + } this@LightningService.node = null } + releaseHandle(node) Logger.info("Node stopped", context = TAG) } - fun wipeStorage(walletIndex: Int) { + /** + * Releases the rust handle instead of leaving it to the GC finalizer, keeping it off the + * single-threaded LDK queue and off the lifecycle critical path. + * + * `free_node` only returns once the node's background tasks drain, which takes tens of seconds + * when a failed start left them wedged, so the wait is bounded and the release is joined rather + * than run inline: cancelling a blocking FFI call does nothing, but abandoning the join lets the + * caller continue while the release finishes on its own. [awaitNodeRelease] gates the next + * rebuild or destructive storage work on the same job so native lifetimes never overlap. + */ + private suspend fun releaseHandle(node: Node) { + val release = launch(ioDispatcher) { + runSuspendCatching { node.destroy() } + .onFailure { Logger.warn("Node handle release error", it, context = TAG) } + } + releaseJob = release + // Clear once done so a later gate check no-ops instead of joining a stale completed job. + release.invokeOnCompletion { if (releaseJob === release) releaseJob = null } + withTimeoutOrNull(NODE_RELEASE_TIMEOUT) { release.join() } + ?: Logger.warn("Node handle release still pending after $NODE_RELEASE_TIMEOUT", context = TAG) + } + + /** + * Blocks until the previous node's release ([releaseHandle] / `free_node`) has finished, so a + * rebuild or destructive storage operation never overlaps the old node's native lifetime. The + * wait is bounded: a stuck free_node throws rather than proceeding (which would overlap) or + * blocking forever (which would brick every future rebuild). + */ + suspend fun awaitNodeRelease(timeout: Duration = NODE_RELEASE_GATE_TIMEOUT) { + val release = releaseJob ?: return + if (release.isActive) Logger.debug("Waiting for previous node release to finish…", context = TAG) + withTimeoutOrNull(timeout) { release.join() } + ?: run { + Logger.error("Previous node release did not finish within $timeout", context = TAG) + throw ServiceError.NodeReleaseTimeout() + } + } + + suspend fun wipeStorage(walletIndex: Int) { if (node != null) throw ServiceError.NodeStillRunning() + awaitNodeRelease() Logger.warn("Wiping LDK storage…", context = TAG) Path(Env.ldkStoragePath(walletIndex)).toFile().deleteRecursively() Logger.info("LDK storage wiped", context = TAG) } - fun resetNetworkGraph(walletIndex: Int) { + suspend fun resetNetworkGraph(walletIndex: Int) { if (node != null) throw ServiceError.NodeStillRunning() + awaitNodeRelease() Logger.warn("Resetting network graph cache…", context = TAG) val ldkPath = Path(Env.ldkStoragePath(walletIndex)).toFile() val graphFile = ldkPath.resolve("network_graph_cache") @@ -1039,13 +1118,20 @@ class LightningService @Inject constructor( // endregion // region events + // Volatile: written by stop()/startEventListener() and read by the listener loop on another thread. + @Volatile private var shouldListenForEvents = true - suspend fun startEventListener(onEvent: NodeEventHandler? = null): Result = runCatching { + suspend fun startEventListener(onEvent: NodeEventHandler? = null): Result = runSuspendCatching { val node = this.node ?: throw ServiceError.NodeNotSetup() + // A re-arm requested from inside an event handler runs on the listener job itself: joining it + // would deadlock and relaunching would double-poll the node. Skip re-arming and keep the + // running listener. This drops the passed onEvent, which is safe only because the callers + // (LightningRepo.start) always re-arm with the same dispatch handler the listener already runs. + if (currentCoroutineContext()[EventListenerContext.Key] != null) return@runSuspendCatching listenerJob?.cancelAndJoin() shouldListenForEvents = true - listenerJob = launch { + listenerJob = launch(EventListenerContext()) { runCatching { Logger.debug("LDK event listener started", context = TAG) listenForEvents(node, onEvent) @@ -1058,7 +1144,10 @@ class LightningService @Inject constructor( } private suspend fun listenForEvents(node: Node, onEvent: NodeEventHandler? = null) = withContext(bgDispatcher) { - while (shouldListenForEvents) { + // Key the loop on node identity, not just the shared flag: a handler-triggered stop nulls + // and frees this node, and a racing start() could flip shouldListenForEvents back on. Without + // the identity check the loop would call nextEventAsync() on the freed node. + while (shouldListenForEvents && node === this@LightningService.node) { ensureActive() val event = runCatching { node.nextEventAsync() }.getOrElse { diff --git a/app/src/main/java/to/bitkit/utils/Errors.kt b/app/src/main/java/to/bitkit/utils/Errors.kt index 5ec25ced0..b42c09099 100644 --- a/app/src/main/java/to/bitkit/utils/Errors.kt +++ b/app/src/main/java/to/bitkit/utils/Errors.kt @@ -17,6 +17,7 @@ sealed class ServiceError(message: String) : AppError(message) { class NodeNotStarted : ServiceError("Node is not started") class MnemonicNotFound : ServiceError("Mnemonic not found") class NodeStillRunning : ServiceError("Node is still running") + class NodeReleaseTimeout : ServiceError("Previous node release did not finish in time") class InvalidNodeSigningMessage : ServiceError("Invalid node signing message") class CurrencyRateUnavailable : ServiceError("Currency rate unavailable") class BlocktankInfoUnavailable : ServiceError("Blocktank info not available") diff --git a/app/src/main/java/to/bitkit/viewmodels/WalletViewModel.kt b/app/src/main/java/to/bitkit/viewmodels/WalletViewModel.kt index a22494de1..9c99ef1d7 100644 --- a/app/src/main/java/to/bitkit/viewmodels/WalletViewModel.kt +++ b/app/src/main/java/to/bitkit/viewmodels/WalletViewModel.kt @@ -270,6 +270,10 @@ class WalletViewModel @Inject constructor( fun setInitNodeLifecycleState() = lightningRepo.setInitNodeLifecycleState() fun start(walletIndex: Int = 0) { + // Cancel before the guards: a start that short-circuits never reaches LightningRepo.start, + // so a stop deferred by an earlier background would fire while the app is foregrounded. + lightningRepo.cancelPendingStop() + if (!walletExists || isStarting) return viewModelScope.launch(bgDispatcher) { @@ -424,12 +428,8 @@ class WalletViewModel @Inject constructor( swapUpdatesRunning = false viewModelScope.launch(bgDispatcher) { stopSwapUpdates() - lightningRepo.stop() - .onFailure { - Logger.error("Node stop error", it) - ToastEventBus.send(it) - } } + lightningRepo.stopDebounced() } private suspend fun stopSwapUpdates() { diff --git a/app/src/test/java/to/bitkit/repositories/LightningRepoTest.kt b/app/src/test/java/to/bitkit/repositories/LightningRepoTest.kt index 920974cf8..1dfbd9e3f 100644 --- a/app/src/test/java/to/bitkit/repositories/LightningRepoTest.kt +++ b/app/src/test/java/to/bitkit/repositories/LightningRepoTest.kt @@ -15,10 +15,12 @@ import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.async +import kotlinx.coroutines.cancelAndJoin import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.flowOf +import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking import org.junit.Before import org.junit.Test @@ -56,12 +58,15 @@ import to.bitkit.data.keychain.Keychain import to.bitkit.ext.createChannelDetails import to.bitkit.ext.of import to.bitkit.models.CoinSelectionPreference +import to.bitkit.models.ElectrumServer import to.bitkit.models.NodeLifecycleState import to.bitkit.models.OpenChannelResult import to.bitkit.models.TransactionSpeed import to.bitkit.services.ActivityService import to.bitkit.services.BlocktankService import to.bitkit.services.CoreService +import to.bitkit.services.ElectrumProbeError +import to.bitkit.services.ElectrumProbeService import to.bitkit.services.LightningService import to.bitkit.services.LnurlService import to.bitkit.services.LspNotificationsService @@ -82,6 +87,7 @@ import kotlin.time.Duration.Companion.seconds class LightningRepoTest : BaseUnitTest() { companion object { private const val NO_USABLE_CHANNELS_FEEDBACK_DELAY_MS = 2_500L + private const val BACKGROUND_STOP_DELAY_MS = 3_000L } private lateinit var sut: LightningRepo @@ -97,6 +103,7 @@ class LightningRepoTest : BaseUnitTest() { private val lnurlService = mock() private val connectivityRepo = mock() private val vssBackupClientLdk = mock() + private val electrumProbeService = mock() private val urlValidator = UrlValidator { Result.success(Unit) } private val probePaymentA = "probe-payment-a" private val probePaymentB = "probe-payment-b" @@ -106,8 +113,10 @@ class LightningRepoTest : BaseUnitTest() { @Before fun setUp() = runBlocking { - whenever(lightningService.setup(any(), anyOrNull(), anyOrNull(), anyOrNull(), anyOrNull())).thenReturn(Unit) + whenever(lightningService.setup(any(), anyOrNull(), anyOrNull(), anyOrNull(), anyOrNull())) + .thenReturn(Unit) whenever(lightningService.start(anyOrNull(), any())).thenReturn(Unit) + whenever { electrumProbeService.probe(any(), any()) }.thenReturn(Result.success(Unit)) whenever(coreService.isGeoBlocked()).thenReturn(false) whenever(cacheStore.data).thenReturn(flowOf(AppCacheData())) whenever(connectivityRepo.isOnline).thenReturn(MutableStateFlow(ConnectivityState.CONNECTED)) @@ -127,6 +136,7 @@ class LightningRepoTest : BaseUnitTest() { connectivityRepo = connectivityRepo, vssBackupClientLdk = vssBackupClientLdk, urlValidator = urlValidator, + electrumProbeService = electrumProbeService, ) } @@ -353,6 +363,79 @@ class LightningRepoTest : BaseUnitTest() { } } + @Test + fun `stopDebounced does not stop the node before the delay elapses`() = test { + startNodeForTesting() + + sut.stopDebounced() + testScheduler.advanceTimeBy(BACKGROUND_STOP_DELAY_MS - 1) + + verify(lightningService, never()).stop() + } + + @Test + fun `stopDebounced stops the node after the delay elapses`() = test { + startNodeForTesting() + + sut.stopDebounced() + testScheduler.advanceTimeBy(BACKGROUND_STOP_DELAY_MS) + testScheduler.advanceUntilIdle() + + verify(lightningService).stop() + assertEquals(NodeLifecycleState.Stopped, sut.lightningState.value.nodeLifecycleState) + } + + @Test + fun `stopDebounced called twice only stops once`() = test { + startNodeForTesting() + + sut.stopDebounced() + sut.stopDebounced() + testScheduler.advanceUntilIdle() + + verify(lightningService, times(1)).stop() + } + + // Regression: a brief background and foreground cycle must not tear the node down and rebuild it + @Test + fun `a background and foreground cycle within the debounce window never stops the node`() = test { + startNodeForTesting() + + sut.stopDebounced() + testScheduler.advanceTimeBy(BACKGROUND_STOP_DELAY_MS - 1) + sut.start() + testScheduler.advanceUntilIdle() + + verify(lightningService, never()).stop() + assertEquals(NodeLifecycleState.Running, sut.lightningState.value.nodeLifecycleState) + } + + // Regression: node teardown must complete before the next start rebuilds, never overlap it + @Test + fun `stop tears down the node before a subsequent start rebuilds it`() = test { + startNodeForTesting() + whenever(lightningService.node).thenReturn(null) + + sut.stop() + sut.start() + + val inOrder = inOrder(lightningService) + inOrder.verify(lightningService).stop() + inOrder.verify(lightningService).setup(any(), anyOrNull(), anyOrNull(), anyOrNull(), anyOrNull()) + inOrder.verify(lightningService).start(anyOrNull(), any()) + } + + // Regression: a cancelled caller must not strand lifecycle state at Stopping + @Test + fun `stop leaves lifecycle state Stopped when the caller is cancelled`() = test { + startNodeForTesting() + + val job = launch { sut.stop() } + job.cancelAndJoin() + + assertEquals(NodeLifecycleState.Stopped, sut.lightningState.value.nodeLifecycleState) + } + @Test fun `resetNetworkGraph clears local cache and VSS copy`() = test { whenever(vssBackupClientLdk.setup(any())).thenReturn(Result.success(Unit)) @@ -835,7 +918,8 @@ class LightningRepoTest : BaseUnitTest() { assertTrue(result.isSuccess) val inOrder = inOrder(lightningService) inOrder.verify(lightningService).stop() - inOrder.verify(lightningService).setup(any(), eq(customServerUrl), anyOrNull(), anyOrNull(), anyOrNull()) + inOrder.verify(lightningService) + .setup(any(), eq(customServerUrl), anyOrNull(), anyOrNull(), anyOrNull()) inOrder.verify(lightningService).start(anyOrNull(), any()) assertEquals(NodeLifecycleState.Running, sut.lightningState.value.nodeLifecycleState) } @@ -851,6 +935,109 @@ class LightningRepoTest : BaseUnitTest() { assertTrue(result.isFailure) } + // Regression: recovery must not block the caller. A wedged node's release can gate the rebuild + // for tens of seconds; the failure has to surface immediately while recovery runs in background. + @Test + fun `restartWithElectrumServer surfaces failure before background recovery completes`() = test { + startNodeForTesting() + val badUrl = "ssl://10.0.2.2:60001" + whenever(lightningService.node).thenReturn(null) + whenever(lightningService.stop()).thenReturn(Unit) + // The switch to the new server fails. + whenever(lightningService.setup(any(), eq(badUrl), anyOrNull(), anyOrNull(), anyOrNull())) + .thenThrow(RuntimeException("start failed")) + // The background recovery (previous config) blocks until released. + val recoveryStarted = CompletableDeferred() + val releaseRecovery = CompletableDeferred() + whenever { lightningService.setup(any(), isNull(), isNull(), anyOrNull(), anyOrNull()) } + .doSuspendableAnswer { + recoveryStarted.complete(Unit) + releaseRecovery.await() + } + + val result = sut.restartWithElectrumServer(badUrl) + + assertTrue(result.isFailure) // surfaced without awaiting recovery + assertTrue(recoveryStarted.isCompleted) // recovery was launched in the background + assertFalse(releaseRecovery.isCompleted) // ... and is still draining + + releaseRecovery.complete(Unit) + testScheduler.advanceUntilIdle() + } + + // Regression: a server that cannot work is rejected by the probe before the node is touched. + // Tearing the node down for one is what leaves its electrum tasks wedged, so that a later + // free_node blocks for tens of seconds instead of milliseconds. + @Test + fun `restartWithElectrumServer rejects a bad server without stopping the node`() = test { + startNodeForTesting() + val badUrl = "ssl://10.0.2.2:60001" + val probeError = ElectrumProbeError.NotElectrum(ElectrumServer.parse(badUrl)) + whenever { electrumProbeService.probe(any(), any()) }.thenReturn(Result.failure(probeError)) + + val result = sut.restartWithElectrumServer(badUrl) + + assertEquals(probeError, result.exceptionOrNull()) + verify(lightningService, never()).stop() + verify(lightningService, never()) + .setup(any(), eq(badUrl), anyOrNull(), anyOrNull(), anyOrNull()) + assertEquals(NodeLifecycleState.Running, sut.lightningState.value.nodeLifecycleState) + } + + // Regression: a start carrying an explicit config must never be satisfied by an already-running + // node, which was not built with it. Returning success here let the caller persist a server that + // never started, e.g. when a detached recovery restarted the previous config mid-change. + @Test + fun `start with a custom server url fails when the node is already running`() = test { + startNodeForTesting() + val newServerUrl = "ssl://next.example.com:50002" + + val result = sut.start(customServerUrl = newServerUrl) + + assertTrue(result.isFailure) + assertTrue(result.exceptionOrNull() is NodeConfigNotAppliedError) + verify(lightningService, never()) + .setup(any(), eq(newServerUrl), anyOrNull(), anyOrNull(), anyOrNull()) + } + + // Two-request barrier: a change issued while the background recovery of a previous failed + // change is in flight waits for it, then genuinely rebuilds with the requested server rather + // than riding the node recovery just brought up. Pins the outcome, not the interleaving: under + // a single-threaded test dispatcher recovery holds the lifecycle mutex for its whole rebuild, + // so the window the transaction lock closes only opens on a truly concurrent dispatcher. + @Test + fun `restartWithElectrumServer serializes a second change behind an in-flight recovery`() = test { + startNodeForTesting() + val badUrl = "ssl://10.0.2.2:60001" + val nextUrl = "ssl://next.example.com:50002" + whenever(lightningService.node).thenReturn(null) + whenever(lightningService.stop()).thenReturn(Unit) + whenever(lightningService.setup(any(), eq(badUrl), anyOrNull(), anyOrNull(), anyOrNull())) + .thenThrow(RuntimeException("start failed")) + // Hold the background recovery mid-flight so the second change is issued while it runs. + val recoveryStarted = CompletableDeferred() + val releaseRecovery = CompletableDeferred() + whenever { lightningService.setup(any(), isNull(), isNull(), anyOrNull(), anyOrNull()) } + .doSuspendableAnswer { + recoveryStarted.complete(Unit) + releaseRecovery.await() + } + + assertTrue(sut.restartWithElectrumServer(badUrl).isFailure) + recoveryStarted.await() + + val secondChange = async { sut.restartWithElectrumServer(nextUrl) } + testScheduler.advanceUntilIdle() + assertFalse(secondChange.isCompleted) // serialized behind the in-flight recovery + + releaseRecovery.complete(Unit) + val result = secondChange.await() + + // Success must mean the node was actually rebuilt with the requested server. + assertTrue(result.isSuccess) + verify(lightningService).setup(any(), eq(nextUrl), anyOrNull(), anyOrNull(), anyOrNull()) + } + @Test fun `restartWithRgsServer should setup with new rgs server`() = test { startNodeForTesting() @@ -883,8 +1070,9 @@ class LightningRepoTest : BaseUnitTest() { startNodeForTesting() whenever(lightningService.node).thenReturn(null) whenever(lightningService.stop()).thenReturn(Unit) - whenever(lightningService.setup(any(), isNull(), eq("https://bad.rgs/snapshot"), anyOrNull(), anyOrNull())) - .thenThrow(RuntimeException("Failed to start node")) + whenever( + lightningService.setup(any(), isNull(), eq("https://bad.rgs/snapshot"), anyOrNull(), anyOrNull()), + ).thenThrow(RuntimeException("Failed to start node")) val result = sut.restartWithRgsServer("https://bad.rgs/snapshot") @@ -908,6 +1096,7 @@ class LightningRepoTest : BaseUnitTest() { connectivityRepo = connectivityRepo, vssBackupClientLdk = vssBackupClientLdk, urlValidator = failingValidator, + electrumProbeService = electrumProbeService, ) sutWithFailingValidator.setInitNodeLifecycleState() whenever(lightningService.node).thenReturn(mock()) diff --git a/app/src/test/java/to/bitkit/services/ElectrumProbeServiceTest.kt b/app/src/test/java/to/bitkit/services/ElectrumProbeServiceTest.kt new file mode 100644 index 000000000..5afd2b359 --- /dev/null +++ b/app/src/test/java/to/bitkit/services/ElectrumProbeServiceTest.kt @@ -0,0 +1,242 @@ +package to.bitkit.services + +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.int +import kotlinx.serialization.json.jsonArray +import kotlinx.serialization.json.jsonObject +import kotlinx.serialization.json.jsonPrimitive +import org.junit.After +import org.junit.Test +import org.lightningdevkit.ldknode.Network +import to.bitkit.models.ElectrumProtocol +import to.bitkit.models.ElectrumServer +import to.bitkit.test.BaseUnitTest +import java.io.BufferedReader +import java.net.InetAddress +import java.net.ServerSocket +import java.net.Socket +import kotlin.concurrent.thread +import kotlin.test.assertEquals +import kotlin.test.assertIs +import kotlin.test.assertTrue + +private const val REGTEST_GENESIS = "0f9188f13cb7b2c71f2a335e3a4fc328bf5beb436012afca590b1a11466e2206" +private const val MAINNET_GENESIS = "000000000019d6689c085ae165831e934ff763ae46a2a6c172b3f1b60a8ce26f" + +@OptIn(ExperimentalCoroutinesApi::class) +class ElectrumProbeServiceTest : BaseUnitTest() { + private val sut = ElectrumProbeService(ioDispatcher = Dispatchers.IO) + + private var server: ServerSocket? = null + + @After + fun tearDown() { + server?.runCatching { close() } + server = null + } + + @Test + fun `probe succeeds against an electrum server on the expected network`() = test { + val port = startFakeElectrum(genesisHash = REGTEST_GENESIS) + + val result = sut.probe(serverAt(port), network = Network.REGTEST) + + assertTrue(result.isSuccess) + } + + @Test + fun `probe rejects a server on a different network`() = test { + val port = startFakeElectrum(genesisHash = MAINNET_GENESIS) + + val result = sut.probe(serverAt(port), network = Network.REGTEST) + + assertIs(result.exceptionOrNull()) + } + + @Test + fun `probe accepts a server that does not report a genesis hash`() = test { + val port = startFakeElectrum(genesisHash = null) + + val result = sut.probe(serverAt(port), network = Network.REGTEST) + + assertTrue(result.isSuccess) // server.features is optional, so this must not reject + } + + // Regression: the version reply must be validated as a JSON-RPC envelope, not merely parsed. + // A server that errors on version negotiation is one the real LDK client rejects at startup, + // and letting it through here recreates the failed-start wedge the probe exists to prevent. + @Test + fun `probe rejects a server that errors on version negotiation`() = test { + val port = startFakeElectrum(versionReply = """{"id":0,"error":{"code":1,"message":"unsupported"}}""") + + val result = sut.probe(serverAt(port), network = Network.REGTEST) + + assertIs(result.exceptionOrNull()) + } + + @Test + fun `probe rejects a version reply with no result`() = test { + val port = startFakeElectrum(versionReply = """{"id":0,"jsonrpc":"2.0"}""") + + val result = sut.probe(serverAt(port), network = Network.REGTEST) + + assertIs(result.exceptionOrNull()) + } + + @Test + fun `probe rejects a version reply answering a different id`() = test { + val port = startFakeElectrum(versionReply = """{"id":99,"result":["fake-electrs","1.4"]}""") + + val result = sut.probe(serverAt(port), network = Network.REGTEST) + + assertIs(result.exceptionOrNull()) + } + + // The rejection reason has to survive as the cause, otherwise the log says only "no electrum + // response" and a wrong protocol version looks the same as a dropped connection. + @Test + fun `probe keeps why the version reply was rejected as the cause`() = test { + val port = startFakeElectrum(versionReply = """{"id":0,"error":{"code":1,"message":"unsupported"}}""") + + val error = sut.probe(serverAt(port), network = Network.REGTEST).exceptionOrNull() + + assertIs(error) + val cause = error.cause?.message.orEmpty() + assertTrue("server.version" in cause, "cause should name the request, was '$cause'") + assertTrue("error" in cause, "cause should report the rejection reason, was '$cause'") + } + + // Regression: a features reply that fails validation must not be read as "no genesis hash" on a + // server whose version negotiation never succeeded — that combination used to probe clean. + @Test + fun `probe rejects a server that errors on both version and features`() = test { + val port = startFakeElectrum( + versionReply = """{"id":0,"error":{"code":1,"message":"unsupported"}}""", + featuresReply = """{"id":1,"error":{"code":-32601,"message":"unknown method"}}""", + ) + + val result = sut.probe(serverAt(port), network = Network.REGTEST) + + assertIs(result.exceptionOrNull()) + } + + @Test + fun `probe rejects a host that never answers the electrum handshake`() = test { + val port = startSilentServer() + + val result = sut.probe(serverAt(port), network = Network.REGTEST) + + assertIs(result.exceptionOrNull()) + } + + @Test + fun `probe rejects an unreachable host`() = test { + val port = ServerSocket(0).use { it.localPort } // closed immediately, nothing listens + + val result = sut.probe(serverAt(port), network = Network.REGTEST) + + assertIs(result.exceptionOrNull()) + } + + // The @settings_10 wedge condition: TLS pointed at a plain-TCP electrum server. Left to + // node.start() this hangs and wedges the node's release; the probe must refuse it instead. + @Test + fun `probe rejects TLS against a plain tcp server`() = test { + val port = startFakeElectrum(genesisHash = REGTEST_GENESIS) + + val result = sut.probe(serverAt(port, ElectrumProtocol.SSL), network = Network.REGTEST) + + assertIs(result.exceptionOrNull()) + } + + @Test + fun `probe reports the requested server in its error`() = test { + val port = startSilentServer() + + val error = sut.probe(serverAt(port), network = Network.REGTEST).exceptionOrNull() + + assertEquals(true, error?.message?.contains("$port")) + } + + // The requests are serialized rather than hand-built, so pin the envelope actually put on the + // wire: a real electrum server has to accept it, and no fake-server assertion covers that. + @Test + fun `probe sends well formed json rpc requests`() = test { + val port = startFakeElectrum(genesisHash = REGTEST_GENESIS) + + sut.probe(serverAt(port), network = Network.REGTEST) + + val sent = synchronized(received) { received.toList() } + assertEquals(2, sent.size) + + val version = Json.parseToJsonElement(sent[0]).jsonObject + assertEquals(0, version.getValue("id").jsonPrimitive.int) + assertEquals("2.0", version.getValue("jsonrpc").jsonPrimitive.content) + assertEquals("server.version", version.getValue("method").jsonPrimitive.content) + assertEquals( + listOf("bitkit", "1.4"), + version.getValue("params").jsonArray.map { it.jsonPrimitive.content }, + ) + + val features = Json.parseToJsonElement(sent[1]).jsonObject + assertEquals(1, features.getValue("id").jsonPrimitive.int) + assertEquals("server.features", features.getValue("method").jsonPrimitive.content) + assertTrue(features.getValue("params").jsonArray.isEmpty()) + } + + private fun serverAt(port: Int, protocol: ElectrumProtocol = ElectrumProtocol.TCP) = ElectrumServer( + host = "127.0.0.1", + tcp = port, + ssl = port, + protocol = protocol, + ) + + /** + * Answers server.version then server.features. Defaults are a well-formed pair; either reply can + * be overridden to exercise a malformed envelope. + */ + private fun startFakeElectrum( + genesisHash: String? = null, + versionReply: String = """{"id":0,"jsonrpc":"2.0","result":["fake-electrs","1.4"]}""", + featuresReply: String = genesisHash + ?.let { """{"id":1,"jsonrpc":"2.0","result":{"genesis_hash":"$it"}}""" } + ?: """{"id":1,"error":{"code":-32601,"message":"unknown method"}}""", + ): Int { + val socket = ServerSocket(0, 1, InetAddress.getByName("127.0.0.1")).also { server = it } + thread(isDaemon = true) { + runCatching { + socket.accept().use { client -> serveElectrum(client, versionReply, featuresReply) } + } + } + return socket.localPort + } + + private fun serveElectrum(client: Socket, versionReply: String, featuresReply: String) { + val reader = client.getInputStream().bufferedReader() + val writer = client.getOutputStream().bufferedWriter() + + readAndRespond(reader, writer) { versionReply } + readAndRespond(reader, writer) { featuresReply } + } + + /** Requests the fake server received, in order, so the encoded envelope can be asserted. */ + private val received = mutableListOf() + + private fun readAndRespond(reader: BufferedReader, writer: java.io.Writer, response: () -> String) { + val request = reader.readLine() ?: return + synchronized(received) { received += request } + writer.write(response() + "\n") + writer.flush() + } + + /** Accepts the connection but never speaks electrum, like a non-electrum service on the port. */ + private fun startSilentServer(): Int { + val socket = ServerSocket(0, 1, InetAddress.getByName("127.0.0.1")).also { server = it } + thread(isDaemon = true) { + runCatching { socket.accept().use { it.getInputStream().read() } } + } + return socket.localPort + } +} diff --git a/app/src/test/java/to/bitkit/services/LightningServiceTest.kt b/app/src/test/java/to/bitkit/services/LightningServiceTest.kt index e77bf24ae..41d95d025 100644 --- a/app/src/test/java/to/bitkit/services/LightningServiceTest.kt +++ b/app/src/test/java/to/bitkit/services/LightningServiceTest.kt @@ -1,18 +1,50 @@ package to.bitkit.services +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.NonCancellable +import kotlinx.coroutines.awaitCancellation +import kotlinx.coroutines.cancelAndJoin +import kotlinx.coroutines.delay +import kotlinx.coroutines.launch +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withContext +import kotlinx.coroutines.withTimeout import org.junit.Before +import org.junit.Rule import org.junit.Test +import org.junit.rules.TemporaryFolder +import org.lightningdevkit.ldknode.Event import org.lightningdevkit.ldknode.Node +import org.lightningdevkit.ldknode.NodeException +import org.mockito.kotlin.doSuspendableAnswer import org.mockito.kotlin.mock +import org.mockito.kotlin.never +import org.mockito.kotlin.timeout +import org.mockito.kotlin.times +import org.mockito.kotlin.verify import org.mockito.kotlin.whenever +import to.bitkit.async.newSingleThreadDispatcher import to.bitkit.data.SettingsStore import to.bitkit.data.backup.VssStoreIdProvider import to.bitkit.data.keychain.Keychain +import to.bitkit.env.Env import to.bitkit.ext.createChannelDetails import to.bitkit.test.BaseUnitTest import to.bitkit.utils.LoggerLdk +import to.bitkit.utils.ServiceError +import java.util.concurrent.CountDownLatch +import java.util.concurrent.atomic.AtomicBoolean import kotlin.test.assertFalse +import kotlin.test.assertNull import kotlin.test.assertTrue +import kotlin.time.Duration.Companion.milliseconds + +private const val VERIFY_TIMEOUT_MS = 2_000L +private const val RELEASE_GATE_PROBE_MS = 200L +private const val STOP_BOUND_MS = 5_000L +private val SHORT_GATE_TIMEOUT = 200.milliseconds class LightningServiceTest : BaseUnitTest() { private val keychain = mock() @@ -21,12 +53,16 @@ class LightningServiceTest : BaseUnitTest() { private val loggerLdk = mock() private val node = mock() + @get:Rule + val tempFolder = TemporaryFolder() + private lateinit var sut: LightningService @Before fun setUp() { sut = LightningService( bgDispatcher = testDispatcher, + ioDispatcher = testDispatcher, keychain = keychain, vssStoreIdProvider = vssStoreIdProvider, settingsStore = settingsStore, @@ -56,4 +92,307 @@ class LightningServiceTest : BaseUnitTest() { assertTrue(sut.canReceive()) } + + @Test + fun `stop destroys the node handle and clears it`() = test { + sut.stop() + + verify(node).stop() + verify(node).destroy() + assertNull(sut.node) + } + + @Test + fun `stop destroys the node handle when it is already not running`() = test { + whenever(node.stop()).thenThrow(NodeException.NotRunning("not running")) + + sut.stop() + + verify(node).destroy() + assertNull(sut.node) + } + + @Test + fun `stop is a no-op when no node is set`() = test { + sut.node = null + + sut.stop() + + verify(node, never()).destroy() + } + + // Regression: a cancelled caller must not abandon teardown, leaving the rust node to the GC finalizer + @Test + fun `stop completes teardown when the caller is cancelled`() = test { + val job = launch { sut.stop() } + + job.cancelAndJoin() + + verify(node).destroy() + assertNull(sut.node) + } + + // Regression: cancelling while the listener is still winding down must not skip native teardown + @Test + fun `stop completes teardown when cancelled during listener cleanup`() = test { + val listenerEntered = CompletableDeferred() + val listenerCleanup = CompletableDeferred() + whenever(node.nextEventAsync()).doSuspendableAnswer { + listenerEntered.complete(Unit) + try { + awaitCancellation() + } finally { + withContext(NonCancellable) { listenerCleanup.await() } + } + } + sut.startEventListener() + listenerEntered.await() + + val stopJob = launch { sut.stop() } + stopJob.cancel() + listenerCleanup.complete(Unit) + stopJob.join() + testScheduler.advanceUntilIdle() + + verify(node).stop() + verify(node).destroy() + assertNull(sut.node) + } + + // Regression: an external stop (not from inside a handler) must cancel AND join the listener, so + // the loop has fully exited before teardown — the counterpart to the handler-triggered skip. + @Test + fun `external stop cancels and joins the running listener`() = test { + val listening = CompletableDeferred() + val listenerExited = CompletableDeferred() + whenever(node.nextEventAsync()).doSuspendableAnswer { + listening.complete(Unit) + try { + awaitCancellation() + } finally { + listenerExited.complete(Unit) + } + } + sut.startEventListener() + listening.await() // listener is parked inside nextEventAsync + + sut.stop() + + // stop() returned, so cancelAndJoin completed: the listener has exited, not been left running. + assertTrue(listenerExited.isCompleted) + verify(node).destroy() + } + + // Regression: a stop requested from inside an event handler runs on the listener job, so joining + // that job would deadlock; teardown must still complete + @Test + fun `stop from within an event handler completes teardown without deadlock`() = test { + val event = Event.PaymentSuccessful(null, "hash", null, null) + // Gate the first event so startEventListener returns and assigns listenerJob before the + // handler runs; otherwise the eager test dispatcher would hide the self-join. + val releaseEvent = CompletableDeferred() + var delivered = false + whenever(node.nextEventAsync()).doSuspendableAnswer { + // A second poll would be on the node destroy() already freed; the loop must exit instead. + check(!delivered) { "polled the node after teardown freed it" } + delivered = true + releaseEvent.await() + } + // The handler stops the node from inside the listener job; the loop then exits on the flag. + val handler: NodeEventHandler = { sut.stop() } + + sut.startEventListener(handler) + releaseEvent.complete(event) + testScheduler.advanceUntilIdle() + + // timeout() polls in real time: teardown finishes on the LDK/IO threads, not virtual time. + // Without the self-join guard, stop() would deadlock and node.stop() would never run. + verify(node, timeout(VERIFY_TIMEOUT_MS)).stop() + verify(node, timeout(VERIFY_TIMEOUT_MS)).destroy() + // The loop re-checks its guard after the handler returns instead of polling the freed node. + verify(node, times(1)).nextEventAsync() + } + + // Regression: startEventListener() also joins listenerJob, so a re-arm requested from inside a + // handler (e.g. LightningRepo.start on an already-running node) would self-cancel the listener + // it runs on — its CancellationException swallowed by runCatching — silently killing it. The + // listener must survive the re-arm and keep delivering events. + @Test + fun `startEventListener from within a handler keeps the listener running`() = test { + // Gate the first event so the outer startEventListener returns and assigns listenerJob before + // the handler re-arms; otherwise the eager test dispatcher leaves listenerJob null and hides it. + val firstEvent = CompletableDeferred() + val secondEvent = CompletableDeferred() + var polls = 0 + whenever(node.nextEventAsync()).doSuspendableAnswer { + if (++polls == 1) firstEvent.await() else secondEvent.await() + } + var handlerCalls = 0 + val handler: NodeEventHandler = { + when (++handlerCalls) { + 1 -> sut.startEventListener { } // re-arm from inside the listener; must not kill it + 2 -> sut.stop() // reached only if the listener survived; also terminates the loop + } + } + + sut.startEventListener(handler) + firstEvent.complete(Event.PaymentSuccessful(null, "h1", null, null)) + secondEvent.complete(Event.PaymentSuccessful(null, "h2", null, null)) + testScheduler.advanceUntilIdle() + + // Reached only if the second event was delivered, i.e. the re-arm did not kill the listener. + verify(node, timeout(VERIFY_TIMEOUT_MS)).stop() + verify(node, timeout(VERIFY_TIMEOUT_MS)).destroy() + } + + // Regression: stop() nulls listenerJob while the old loop is still unwinding, so a racing start() + // can install a new node and re-arm shouldListenForEvents before the old loop returns. The loop + // must key on node identity and exit, not poll the node the previous teardown freed. + @Test + fun `listener stops polling the old node once a new node is swapped in`() = test { + val newNode = mock() + val releaseEvent = CompletableDeferred() + var delivered = false + whenever(node.nextEventAsync()).doSuspendableAnswer { + check(!delivered) { "polled the stale node after it was swapped out" } + delivered = true + releaseEvent.await() + } + // Simulate a concurrent start(): swap in a new node with the listener flag still on. + val handler: NodeEventHandler = { sut.node = newNode } + + sut.startEventListener(handler) + releaseEvent.complete(Event.PaymentSuccessful(null, "hash", null, null)) + testScheduler.advanceUntilIdle() + + verify(node, times(1)).nextEventAsync() + verify(newNode, never()).nextEventAsync() + } + + private class GateFixture( + scope: CoroutineScope, + val gated: LightningService, + val destroyGate: CountDownLatch, + val destroyed: AtomicBoolean, + ) : CoroutineScope by scope + + // Runs [block] against a LightningService on real dispatchers whose node.destroy() blocks on + // destroyGate, so the release gate can be observed. Real dispatchers + a controllably delayed + // destroy() are exactly the setup the gate needs to be tested independently of virtual time. + private fun runGateTest(block: suspend GateFixture.() -> Unit) = runBlocking { + Env.initAppStoragePath(tempFolder.root.absolutePath) + val io = newSingleThreadDispatcher("test-gate-io") + val bg = newSingleThreadDispatcher("test-gate-bg") + val gated = LightningService(bg, io, keychain, vssStoreIdProvider, settingsStore, loggerLdk) + gated.node = node + val destroyGate = CountDownLatch(1) + val destroyed = AtomicBoolean(false) + whenever(node.destroy()).thenAnswer { + destroyGate.await() + destroyed.set(true) + } + try { + GateFixture(this, gated, destroyGate, destroyed).block() + } finally { + destroyGate.countDown() // release any wedged destroy so the io thread can exit + io.close() + bg.close() + } + } + + // Regression: destructive storage work must wait for the previous node's release (free_node) so + // native lifetimes never overlap. Also pins (d): stop() returns within its bound despite the wedge. + @Test + fun `resetNetworkGraph waits for the previous node release to finish`() = runGateTest { + withTimeout(STOP_BOUND_MS) { gated.stop() } // stop returns within its bound though destroy() is wedged + + val resetDone = CompletableDeferred() + launch(Dispatchers.Default) { + gated.resetNetworkGraph(walletIndex = 0) + resetDone.complete(Unit) + } + + delay(RELEASE_GATE_PROBE_MS) + assertFalse(resetDone.isCompleted) // gate holds while the release is still draining + assertFalse(destroyed.get()) + + destroyGate.countDown() + resetDone.await() + assertTrue(destroyed.get()) // release finished before resetNetworkGraph proceeded + } + + // Regression (b): wipeStorage deletes storage and must wait for the release just like the rebuild. + @Test + fun `wipeStorage waits for the previous node release to finish`() = runGateTest { + withTimeout(STOP_BOUND_MS) { gated.stop() } + + val wipeDone = CompletableDeferred() + launch(Dispatchers.Default) { + gated.wipeStorage(walletIndex = 0) + wipeDone.complete(Unit) + } + + delay(RELEASE_GATE_PROBE_MS) + assertFalse(wipeDone.isCompleted) + assertFalse(destroyed.get()) + + destroyGate.countDown() + wipeDone.await() + assertTrue(destroyed.get()) + } + + // Regression (a): the rebuild path (setup) must wait for the release before building a new node. + @Test + fun `setup waits for the previous node release before rebuilding`() = runGateTest { + withTimeout(STOP_BOUND_MS) { gated.stop() } + + val setupDone = CompletableDeferred>() + launch(Dispatchers.Default) { + setupDone.complete(runCatching { gated.setup(walletIndex = 0) }) + } + + delay(RELEASE_GATE_PROBE_MS) + assertFalse(setupDone.isCompleted) // gate blocks the rebuild while the release drains + + destroyGate.countDown() + setupDone.await() // proceeds once released (build then fails on the mocked deps, which is fine) + } + + // Regression (c): with no pending release the gate must add no latency. + @Test + fun `wipeStorage proceeds immediately when no release is pending`() = runGateTest { + gated.node = null // no stop() has run, so releaseJob is null + + withTimeout(RELEASE_GATE_PROBE_MS) { gated.wipeStorage(walletIndex = 0) } // completes without waiting + } + + // Regression: a stuck free_node must not block rebuilds forever; the gate throws once bounded out. + @Test + fun `awaitNodeRelease throws when the release never finishes`() = runGateTest { + withTimeout(STOP_BOUND_MS) { gated.stop() } // release launched; destroy() stays wedged (never counted down) + + val error = runCatching { gated.awaitNodeRelease(timeout = SHORT_GATE_TIMEOUT) }.exceptionOrNull() + + assertTrue(error is ServiceError.NodeReleaseTimeout) + } + + // Regression: a failing node stop must still release the handle instead of rethrowing and leaking it + @Test + fun `stop destroys the node handle when node stop throws`() = test { + whenever(node.stop()).thenThrow(NodeException.ConnectionFailed("boom")) + + sut.stop() + + verify(node).destroy() + assertNull(sut.node) + } + + // Regression: destroying twice would surface later as IllegalStateException from callWithPointer + @Test + fun `consecutive stop calls destroy the handle only once`() = test { + sut.stop() + sut.stop() + + verify(node, times(1)).destroy() + } } diff --git a/app/src/test/java/to/bitkit/ui/WalletViewModelTest.kt b/app/src/test/java/to/bitkit/ui/WalletViewModelTest.kt index 771f4736a..46f82e2e2 100644 --- a/app/src/test/java/to/bitkit/ui/WalletViewModelTest.kt +++ b/app/src/test/java/to/bitkit/ui/WalletViewModelTest.kt @@ -1,6 +1,7 @@ package to.bitkit.ui import android.content.Context +import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableStateFlow @@ -12,8 +13,10 @@ import org.junit.Test import org.lightningdevkit.ldknode.PeerDetails import org.mockito.kotlin.any import org.mockito.kotlin.anyOrNull +import org.mockito.kotlin.doSuspendableAnswer import org.mockito.kotlin.mock import org.mockito.kotlin.never +import org.mockito.kotlin.times import org.mockito.kotlin.verify import org.mockito.kotlin.verifyBlocking import org.mockito.kotlin.whenever @@ -350,6 +353,65 @@ class WalletViewModelTest : BaseUnitTest() { verify(testWalletRepo).refreshBip21() } + // Regression: a start that short-circuits on the isStarting guard never reaches + // LightningRepo.start, so it must cancel a deferred stop itself or the node stops while foregrounded + @Test + fun `foreground start cancels deferred stop while startup is active`() = test { + val testWalletRepo: WalletRepo = mock() + val testLightningRepo: LightningRepo = mock() + val testWalletState = MutableStateFlow(WalletState(walletExists = true)) + + whenever(testWalletRepo.walletState).thenReturn(testWalletState) + whenever(testWalletRepo.balanceState).thenReturn(balanceState) + whenever(testWalletRepo.walletExists()).thenReturn(true) + whenever(testLightningRepo.lightningState).thenReturn(lightningState) + whenever(testLightningRepo.isRecoveryMode).thenReturn(isRecoveryMode) + + val startEntered = CompletableDeferred() + val finishStart = CompletableDeferred() + whenever( + testLightningRepo.start( + any(), + anyOrNull(), + any(), + anyOrNull(), + anyOrNull(), + anyOrNull(), + anyOrNull(), + any(), + ), + ).doSuspendableAnswer { + startEntered.complete(Unit) + finishStart.await() + Result.success(Unit) + } + + 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.start() + startEntered.await() + testSut.stop() + testSut.start() + + verify(testLightningRepo).stopDebounced() + verify(testLightningRepo, times(2)).cancelPendingStop() + + finishStart.complete(Unit) + advanceUntilIdle() + } + @Test fun `start should skip refreshBip21 when restore is in progress`() = test { // Create fresh mocks for this test diff --git a/changelog.d/next/1100.fixed.md b/changelog.d/next/1100.fixed.md new file mode 100644 index 000000000..e4cb04a71 --- /dev/null +++ b/changelog.d/next/1100.fixed.md @@ -0,0 +1 @@ +Lightning node teardown now releases native resources deterministically and no longer restarts the node when the app is only briefly backgrounded.