diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index f8d9d521e..d7d6b5eee 100644 --- a/src/wallet/mod.rs +++ b/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) -> 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 { 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 { - 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 { - 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| { diff --git a/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index e401c8218..03287a409 100644 --- a/tests/integration_tests_rust.rs +++ b/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, + wallet_write_gate: Arc>, + gate_engaged: Arc, + wallet_writes_completed: Arc, +} + +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, 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, + ) -> impl Future> + '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> + 'static + Send { + KVStore::remove(&*self.inner, primary_namespace, secondary_namespace, key, lazy) + } + + fn list( + &self, primary_namespace: &str, secondary_namespace: &str, + ) -> impl Future, 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, + ) -> impl Future> + '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();