Derive shutdown scripts without blocking on wallet persistence

LDK invokes the sync `SignerProvider::get_shutdown_scriptpubkey` and
`get_destination_script` callbacks on runtime worker threads while
holding channel locks, e.g. when accepting an inbound channel. Blocking
there on wallet persistence could deadlock the runtime: the parked
callback still held the channel locks, other tasks blocking
synchronously on those locks captured the remaining worker cores, and
the persistence future the callback waited on could then never be
polled. Observed as a permanent hang of integration test runs.

Instead, reveal the address without waiting and persist the staged
change set in the background. A persistence failure therefore no longer
rejects the channel; if the node crashes before the flush lands, the
revealed index may be handed out again after restart, which BDK's
keychain lookahead tolerates.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: jkczyz

src/wallet/mod.rs

@@ -518,6 +518,36 @@ impl Wallet {
 		Ok(address_info.address)
 	}
 
+	/// Returns a new address, persisting the revealed derivation index in the background rather
+	/// than waiting on it.
+	///
+	/// This exists for sync callbacks (e.g., [`SignerProvider`]) that LDK invokes on runtime
+	/// worker threads while holding channel locks. Blocking such a callback on persistence can
+	/// deadlock the runtime: other tasks blocking synchronously on the same channel locks capture
+	/// the remaining workers, leaving none to drive the persistence future the callback waits on.
+	///
+	/// If the node crashes before the background flush lands, the revealed index is lost and the
+	/// address may be handed out again after restart. BDK's keychain lookahead still detects any
+	/// funds it receives.
+	pub(crate) fn get_new_address_deferring_persist(self: &Arc<Self>) -> bitcoin::Address {
+		let address_info =
+			self.inner.lock().expect("lock").reveal_next_address(KeychainKind::External);
+
+		// Leave the change set staged: whichever flow next takes the persister lock and calls
+		// `take_staged` (possibly the task spawned here) persists the reveal, preserving the
+		// ordering that serializing those two steps under the persister lock establishes.
+		let wallet = Arc::clone(self);
+		self.runtime.spawn_background_task(async move {
+			let mut locked_persister = wallet.persister.lock().await;
+			let change_set = wallet.inner.lock().expect("lock").take_staged().unwrap_or_default();
+			if let Err(e) = locked_persister.persist_changeset(change_set).await {
+				log_error!(wallet.logger, "Failed to persist wallet: {}", e);
+			}
+		});
+
+		address_info.address
+	}
+
 	pub(crate) async fn get_new_internal_address(&self) -> Result<bitcoin::Address, Error> {
 		let mut locked_persister = self.persister.lock().await;
 		let (address_info, change_set) = {
@@ -2090,16 +2120,16 @@ impl SignerProvider for WalletKeysManager {
 	}
 
 	fn get_destination_script(&self, _channel_keys_id: [u8; 32]) -> Result<ScriptBuf, ()> {
-		let address = self.wallet.runtime.block_on(self.wallet.get_new_address()).map_err(|e| {
-			log_error!(self.logger, "Failed to retrieve new address from wallet: {}", e);
-		})?;
+		// LDK may invoke this callback on a runtime worker thread while holding channel locks.
+		// It must not block on the runtime, or the runtime can deadlock.
+		let address = self.wallet.get_new_address_deferring_persist();
 		Ok(address.script_pubkey())
 	}
 
 	fn get_shutdown_scriptpubkey(&self) -> Result<ShutdownScript, ()> {
-		let address = self.wallet.runtime.block_on(self.wallet.get_new_address()).map_err(|e| {
-			log_error!(self.logger, "Failed to retrieve new address from wallet: {}", e);
-		})?;
+		// LDK may invoke this callback on a runtime worker thread while holding channel locks.
+		// It must not block on the runtime, or the runtime can deadlock.
+		let address = self.wallet.get_new_address_deferring_persist();
 
 		match address.witness_program() {
 			Some(program) => ShutdownScript::new_witness_program(&program).map_err(|e| {

tests/integration_tests_rust.rs

@@ -10,7 +10,7 @@ mod common;
 use std::collections::HashSet;
 use std::future::Future;
 use std::str::FromStr;
-use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::{mpsc, Arc};
 use std::time::Duration;
 
@@ -24,10 +24,10 @@ use common::{
 	expect_channel_pending_event, expect_channel_ready_event, expect_channel_ready_events,
 	expect_event, expect_payment_claimable_event, expect_payment_received_event,
 	expect_payment_successful_event, expect_splice_negotiated_event, generate_blocks_and_wait,
-	generate_listening_addresses, invalidate_blocks, open_channel, open_channel_push_amt,
-	open_channel_with_all, premine_and_distribute_funds, premine_blocks, prepare_rbf,
-	random_chain_source, random_config, setup_bitcoind_and_electrsd, setup_builder, setup_node,
-	setup_two_nodes, splice_in_with_all, wait_for_block, wait_for_tx, InMemoryStore,
+	generate_listening_addresses, invalidate_blocks, open_channel, open_channel_no_wait,
+	open_channel_push_amt, open_channel_with_all, premine_and_distribute_funds, premine_blocks,
+	prepare_rbf, random_chain_source, random_config, setup_bitcoind_and_electrsd, setup_builder,
+	setup_node, setup_two_nodes, splice_in_with_all, wait_for_block, wait_for_tx, InMemoryStore,
 	TestChainSource, TestConfig, TestStoreType, TestSyncStore,
 };
 use electrsd::corepc_node::{self, Node as BitcoinD};
@@ -216,6 +216,164 @@ fn wallet_store_contention_does_not_stall_runtime() {
 	result.unwrap_or_else(|e| panic!("wallet contention test failed: {e}"));
 }
 
+#[derive(Clone)]
+struct WalletPersistGatedStore {
+	inner: Arc<InMemoryStore>,
+	wallet_write_gate: Arc<tokio::sync::RwLock<()>>,
+	gate_engaged: Arc<AtomicBool>,
+	wallet_writes_completed: Arc<AtomicUsize>,
+}
+
+impl WalletPersistGatedStore {
+	fn new() -> Self {
+		Self {
+			inner: Arc::new(InMemoryStore::new()),
+			wallet_write_gate: Arc::new(tokio::sync::RwLock::new(())),
+			gate_engaged: Arc::new(AtomicBool::new(false)),
+			wallet_writes_completed: Arc::new(AtomicUsize::new(0)),
+		}
+	}
+}
+
+impl KVStore for WalletPersistGatedStore {
+	fn read(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
+	) -> impl Future<Output = Result<Vec<u8>, lightning::io::Error>> + 'static + Send {
+		KVStore::read(&*self.inner, primary_namespace, secondary_namespace, key)
+	}
+
+	fn write(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
+	) -> impl Future<Output = Result<(), lightning::io::Error>> + 'static + Send {
+		let inner = Arc::clone(&self.inner);
+		let wallet_write_gate = Arc::clone(&self.wallet_write_gate);
+		let gate_engaged = Arc::clone(&self.gate_engaged);
+		let wallet_writes_completed = Arc::clone(&self.wallet_writes_completed);
+		let primary_namespace = primary_namespace.to_string();
+		let secondary_namespace = secondary_namespace.to_string();
+		let key = key.to_string();
+		async move {
+			let is_wallet_write = primary_namespace == "bdk_wallet";
+			if is_wallet_write && gate_engaged.load(Ordering::Acquire) {
+				let _guard = wallet_write_gate.read().await;
+			}
+			let res =
+				KVStore::write(&*inner, &primary_namespace, &secondary_namespace, &key, buf).await;
+			if is_wallet_write && res.is_ok() {
+				wallet_writes_completed.fetch_add(1, Ordering::AcqRel);
+			}
+			res
+		}
+	}
+
+	fn remove(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
+	) -> impl Future<Output = Result<(), lightning::io::Error>> + 'static + Send {
+		KVStore::remove(&*self.inner, primary_namespace, secondary_namespace, key, lazy)
+	}
+
+	fn list(
+		&self, primary_namespace: &str, secondary_namespace: &str,
+	) -> impl Future<Output = Result<Vec<String>, lightning::io::Error>> + 'static + Send {
+		KVStore::list(&*self.inner, primary_namespace, secondary_namespace)
+	}
+}
+
+impl PaginatedKVStore for WalletPersistGatedStore {
+	fn list_paginated(
+		&self, primary_namespace: &str, secondary_namespace: &str, page_token: Option<PageToken>,
+	) -> impl Future<Output = Result<PaginatedListResponse, lightning::io::Error>> + 'static + Send
+	{
+		PaginatedKVStore::list_paginated(
+			&*self.inner,
+			primary_namespace,
+			secondary_namespace,
+			page_token,
+		)
+	}
+}
+
+// LDK invokes the sync `SignerProvider::get_shutdown_scriptpubkey` callback on a runtime worker
+// thread while holding channel locks when a node accepts (or opens) a channel. If deriving the
+// shutdown script waits on wallet persistence, a contended wallet store wedges the event handler
+// while it holds those locks, and other runtime tasks blocking on the same locks can capture the
+// remaining workers, deadlocking the runtime. Gate node B's BDK wallet writes and assert the
+// channel open still completes, with the revealed address persisted once the store recovers.
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+async fn channel_open_completes_while_wallet_persistence_is_stalled() {
+	let (bitcoind, electrsd) = setup_bitcoind_and_electrsd();
+	let esplora_url = format!("http://{}", electrsd.esplora_url.as_ref().unwrap());
+	let chain_source = TestChainSource::Esplora(&electrsd);
+
+	let config_a = random_config();
+	let node_a = setup_node(&chain_source, config_a);
+
+	let config_b = random_config();
+	setup_builder!(builder_b, config_b.node_config);
+	let mut sync_config = EsploraSyncConfig::default();
+	sync_config.background_sync_config = None;
+	builder_b.set_chain_source_esplora(esplora_url, Some(sync_config));
+	let store = WalletPersistGatedStore::new();
+	let node_b = builder_b.build_with_store(config_b.node_entropy.into(), store.clone()).unwrap();
+	node_b.start().unwrap();
+
+	// Fund both nodes so node B passes the anchor reserve check on the accept path.
+	let address_a = node_a.onchain_payment().new_address().unwrap();
+	let address_b = node_b.onchain_payment().new_address().unwrap();
+	premine_and_distribute_funds(
+		&bitcoind.client,
+		&electrsd.client,
+		vec![address_a, address_b],
+		Amount::from_sat(5_000_000),
+	)
+	.await;
+	node_a.sync_wallets().unwrap();
+	node_b.sync_wallets().unwrap();
+
+	// Stall writes of node B's BDK wallet data before the channel open reaches the accept path.
+	// The gate guard lives on a plain thread with a deadline: if the gated write wedges the
+	// runtime (the bug under test captures all workers, so even timers stop firing), the gate
+	// force-reopens after the event timeouts below have expired, letting them fail the test
+	// cleanly instead of hanging it.
+	let (release_gate_sender, release_gate_receiver) = mpsc::sync_channel::<()>(1);
+	let (gate_held_sender, gate_held_receiver) = mpsc::sync_channel::<()>(1);
+	let gate = Arc::clone(&store.wallet_write_gate);
+	std::thread::spawn(move || {
+		let _guard = gate.blocking_write();
+		let _ = gate_held_sender.send(());
+		let _ = release_gate_receiver.recv_timeout(Duration::from_secs(90));
+	});
+	gate_held_receiver.recv().unwrap();
+	store.gate_engaged.store(true, Ordering::Release);
+	let wallet_writes_before = store.wallet_writes_completed.load(Ordering::Acquire);
+
+	// Accepting the channel must not wait on wallet persistence: `open_channel_no_wait` times out
+	// waiting for the `ChannelPending` events otherwise.
+	let funding_txo = open_channel_no_wait(&node_a, &node_b, 500_000, None, false).await;
+
+	// Reopen the gate and verify the deferred persist of the revealed shutdown-script address
+	// eventually lands.
+	store.gate_engaged.store(false, Ordering::Release);
+	// The watchdog thread may have force-released the gate already on a slow run.
+	let _ = release_gate_sender.send(());
+	let persisted = async {
+		while store.wallet_writes_completed.load(Ordering::Acquire) <= wallet_writes_before {
+			tokio::time::sleep(Duration::from_millis(50)).await;
+		}
+	};
+	tokio::time::timeout(Duration::from_secs(common::INTEROP_TIMEOUT_SECS), persisted)
+		.await
+		.expect("timed out waiting for the deferred wallet persist");
+
+	// The channel and both nodes remain fully functional.
+	wait_for_tx(&electrsd.client, funding_txo.txid).await;
+	generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await;
+	node_a.sync_wallets().unwrap();
+	node_b.sync_wallets().unwrap();
+	expect_channel_ready_event!(node_a, node_b.node_id());
+	expect_channel_ready_event!(node_b, node_a.node_id());
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 async fn channel_full_cycle() {
 	let (bitcoind, electrsd) = setup_bitcoind_and_electrsd();