Skip to content

fix: harden ldk node teardown on stop #1100

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
jvsena42 wants to merge 12 commits into
base: master
Choose a base branch
from
Draft

fix: harden ldk node teardown on stop #1100

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
dc4db07
fix: add stop debounce and prevent mid-stop cancellation from Viewmod…
jvsena42 Jul 22, 2026
a28fe1c
fix: prevent stop cancellation from context cancellation and release …
jvsena42 Jul 22, 2026
6abdfb3
doc: changelog
jvsena42 Jul 22, 2026
14f6e11
chore: rename changelog fragment
jvsena42 Jul 22, 2026
d0e1f13
fix: make stop scheduling and cancel atomic
jvsena42 Jul 22, 2026
3bb4624
fix: call release independently rather than from inline and add a tim…
jvsena42 Jul 23, 2026
a6cb8ff
fix: replace runCatching with runSuspendCatching
jvsena42 Jul 23, 2026
82fa056
fix: guard all stop against caller cancellation and cancel stop early…
jvsena42 Jul 24, 2026
2a786ed
fix: self-join deadlock when WakeNodeWorker calls stop() from inside …
jvsena42 Jul 27, 2026
c3b55db
fix: guard listener re-arm against self-join
jvsena42 Jul 27, 2026
04402b5
fix: prevent overlap node's instance after 1s bounded stop
jvsena42 Jul 27, 2026
140ba5e
fix: remove gating for electrum setting restart
jvsena42 Jul 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 54 additions & 9 deletions app/src/main/java/to/bitkit/repositories/LightningRepo.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -144,6 +145,8 @@ class LightningRepo @Inject constructor(
private val syncMutex = Mutex()
private val syncPending = AtomicBoolean(false)
private val syncRetryJob = AtomicReference<Job?>(null)
private val pendingStopJob = AtomicReference<Job?>(null)
private val pendingStopLock = Any()
private val lifecycleMutex = Mutex()
private val isChangingAddressType = AtomicBoolean(false)

Expand Down Expand Up @@ -270,6 +273,7 @@ class LightningRepo @Inject constructor(
customServerUrl: String? = null,
customRgsServerUrl: String? = null,
channelMigration: ChannelDataMigration? = null,
awaitRelease: Boolean = true,
) = withContext(bgDispatcher) {
runCatching {
val trustedPeers = fetchTrustedPeers()
Expand All @@ -279,6 +283,7 @@ class LightningRepo @Inject constructor(
customRgsServerUrl,
trustedPeers,
channelMigration,
awaitRelease,
)
}.onFailure {
Logger.error("Node setup error", it, context = TAG)
Expand All @@ -305,11 +310,14 @@ class LightningRepo @Inject constructor(
eventHandler: NodeEventHandler? = null,
channelMigration: ChannelDataMigration? = null,
shouldValidateGraph: Boolean = true,
awaitRelease: Boolean = true,
): Result<Unit> = withContext(bgDispatcher) {
if (_isRecoveryMode.value) {
return@withContext Result.failure(RecoveryModeError())
}

cancelPendingStop()

eventHandler?.let { _eventHandlers.add(it) }

// Track retry state outside mutex to avoid deadlock (Mutex is non-reentrant)
Expand All @@ -330,7 +338,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, awaitRelease)
if (setupResult.isFailure) {
_lightningState.update {
it.copy(
Expand Down Expand Up @@ -537,6 +546,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<Unit> = withContext(bgDispatcher) {
lifecycleMutex.withLock {
if (_lightningState.value.nodeLifecycleState.isStoppedOrStopping()) {
Expand All @@ -545,10 +572,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
Expand Down Expand Up @@ -756,9 +785,16 @@ class LightningRepo @Inject constructor(
start(
shouldRetry = false,
customServerUrl = newServerUrl,
// Skip the release gate for the server-change rebuild: @settings_10 switches servers
// back to back, and gating would serialize each change behind the previous (possibly
// wedged, ~40s) node's drain. A rebuild-over-drain here is no worse than the pre-PR GC
// behaviour; the destructive storage deletes (wipe / graph-reset / scorer) keep the gate.
awaitRelease = false,
).onFailure {
Logger.warn("Failed ldk-node config change, attempting recovery…", context = TAG)
restartWithPreviousConfig()
// 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() }

@ovitrif ovitrif Jul 28, 2026 •
edited
Loading

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see this detached recovery—and the identical RGS branch—racing the next Connect attempt. Because recovery and the new request acquire the lifecycle mutex separately for stop() and start(), recovery can start the old config first; the new start then takes the Running fast path, reports success, and persists its URL without applying it.

Please serialize Electrum changes, RGS changes, and their recovery behind one dedicated configuration mutex, and assign each request a monotonically increasing generation before it waits for that mutex. Run the detached recovery under the same mutex and skip it when its captured generation is no longer current:

private val configChangeMutex = Mutex()
private val configGeneration = AtomicLong(0)

private fun recoverPreviousConfig(generation: Long) = scope.launch {
    configChangeMutex.withLock {
        if (generation != configGeneration.get()) return@withLock
        restartWithPreviousConfig()
    }
}

Keep the successful settingsStore.update inside that serialized transaction. Add a two-request barrier test that fails request A, pauses its recovery, starts request B, then releases A; assert A’s recovery is skipped, B’s server is the running configuration, and only B’s URL is persisted.

}.onSuccess {
settingsStore.update { it.copy(electrumServer = newServerUrl) }

Expand Down Expand Up @@ -786,8 +822,10 @@ class LightningRepo @Inject constructor(
shouldRetry = false,
customRgsServerUrl = newRgsUrl,
).onFailure {
Logger.warn("Failed ldk-node config change, attempting recovery…", context = TAG)
restartWithPreviousConfig()
// 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) }

Expand Down Expand Up @@ -991,7 +1029,10 @@ class LightningRepo @Inject constructor(
Logger.debug("Starting node with previous config for recovery", context = TAG)

start(
// Recovery runs while holding the lifecycle mutex; gating the rebuild here would block
// the user's next server change behind the wedged node's drain. See restartWithElectrumServer.
shouldRetry = false,
awaitRelease = false,
).onSuccess {
Logger.debug("Successfully started node with previous config", context = TAG)
}.onFailure {
Expand Down Expand Up @@ -1767,6 +1808,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()
Expand Down Expand Up @@ -1809,6 +1853,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
Expand Down
115 changes: 103 additions & 12 deletions app/src/main/java/to/bitkit/services/LightningService.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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<EventListenerContext>
}

data class AddressDerivationInfo(
val address: String,
val index: Int,
Expand All @@ -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,
Expand All @@ -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,
Expand All @@ -121,13 +144,21 @@ 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

@Suppress("LongParameterList")
suspend fun setup(
walletIndex: Int,
customServerUrl: String? = null,
customRgsServerUrl: String? = null,
trustedPeers: List<PeerDetails>? = null,
channelMigration: ChannelDataMigration? = null,
awaitRelease: Boolean = true,
) {
if (awaitRelease) awaitNodeRelease()
Logger.debug("Building node…", context = TAG)

val config = config(walletIndex, trustedPeers)
Expand Down Expand Up @@ -295,7 +326,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) {
Expand All @@ -314,34 +345,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) {
Comment thread
jvsena42 marked this conversation as resolved.
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() }

@ovitrif ovitrif Jul 24, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I’m treating native destruction as part of the lifecycle gate here. After this timeout, stop() reports success and LightningRepo publishes Stopped, even though the old Rust node can still be draining for tens of seconds. Immediate restart paths can then build a replacement over the same LDK state, while wipeStorage() and graph-reset paths can mutate that storage before the old owner releases it. That recreates the overlapping native lifetime this PR is intended to remove. Please retain an explicit release gate that prevents rebuild and destructive storage work until destroy() completes, or cross a safe process-recovery boundary, and cover it with distinct dispatchers plus a controllably delayed destroy().

@jvsena42 jvsena42 Jul 27, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

fixed in fc385e5

@jvsena42 jvsena42 Jul 27, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

wipe-and-restore.webm

?: 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")
Expand Down Expand Up @@ -1039,13 +1120,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<Unit> = runCatching {
suspend fun startEventListener(onEvent: NodeEventHandler? = null): Result<Unit> = 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)
Expand All @@ -1058,7 +1146,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 {
Expand Down
Loading
Loading