Derive shutdown scripts without blocking on wallet persistence

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();