From 273e4d06a7ae15caf5ee1bae418072bbf7cdc0d7 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Thu, 30 Jul 2026 11:20:50 -0500 Subject: [PATCH 1/5] 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 --- src/wallet/mod.rs | 42 ++++++-- tests/integration_tests_rust.rs | 168 +++++++++++++++++++++++++++++++- 2 files changed, 199 insertions(+), 11 deletions(-) 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(); From 1b27621d55b692ac77054f322d22a58a36b052c4 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Thu, 6 Aug 2026 11:31:37 -0500 Subject: [PATCH 2/5] f - Pool pre-persisted addresses instead of deferring the reveal's persistence Co-Authored-By: Claude Fable 5 --- src/builder.rs | 7 + src/io/mod.rs | 5 + src/wallet/mod.rs | 611 ++++++++++++++++++++++++++++++-- src/wallet/persist.rs | 72 ++++ tests/integration_tests_rust.rs | 69 +++- 5 files changed, 736 insertions(+), 28 deletions(-) diff --git a/src/builder.rs b/src/builder.rs index f11780099..3222f4b93 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -1754,6 +1754,13 @@ fn build_with_store_internal( Arc::clone(&pending_payment_store), )); + // Fill the address pool up front so LDK's sync `SignerProvider` callbacks can hand out + // pre-persisted addresses without waiting on wallet persistence. + runtime.block_on(wallet.initialize_address_pool()).map_err(|e| { + log_error!(logger, "Failed to initialize the wallet's address pool: {}", e); + BuildError::WalletSetupFailed + })?; + tx_broadcaster.set_wallet(Arc::downgrade(&wallet)); // Initialize the KeysManager diff --git a/src/io/mod.rs b/src/io/mod.rs index a01aa59a8..c70c68d96 100644 --- a/src/io/mod.rs +++ b/src/io/mod.rs @@ -80,6 +80,11 @@ pub(crate) const BDK_WALLET_INDEXER_PRIMARY_NAMESPACE: &str = "bdk_wallet"; pub(crate) const BDK_WALLET_INDEXER_SECONDARY_NAMESPACE: &str = ""; pub(crate) const BDK_WALLET_INDEXER_KEY: &str = "indexer"; +/// The derivation indices of the wallet's address pool will be persisted under this key. +pub(crate) const BDK_WALLET_ADDRESS_POOL_PRIMARY_NAMESPACE: &str = "bdk_wallet"; +pub(crate) const BDK_WALLET_ADDRESS_POOL_SECONDARY_NAMESPACE: &str = ""; +pub(crate) const BDK_WALLET_ADDRESS_POOL_KEY: &str = "address_pool"; + /// [`StaticInvoice`]s will be persisted under this key. /// /// [`StaticInvoice`]: lightning::offers::static_invoice::StaticInvoice diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index d7d6b5eee..2e3d966a8 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -5,7 +5,7 @@ // http://opensource.org/licenses/MIT>, at your option. You may not use this file except in // accordance with one or both of these licenses. -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; use std::future::Future; use std::ops::Deref; use std::str::FromStr; @@ -81,10 +81,39 @@ pub(crate) mod ser; const DUST_LIMIT_SATS: u64 = 546; +/// The number of external addresses kept revealed, persisted, and ready for handout via +/// [`Wallet::pop_pooled_address`]. +/// +/// Each channel open consumes two pooled addresses (one for the destination script and one for +/// the upfront shutdown script), and the pool is refilled after every handout, so this bounds +/// how many channels can be opened while wallet persistence is unavailable rather than steady +/// state throughput. Pooled addresses are revealed-but-unused, so this value also widens +/// incremental chain syncs accordingly and must stay below the default full-scan stop gap +/// ([`DEFAULT_FULL_SCAN_STOP_GAP`]) lest a from-seed restore's full scan stop inside the pool's +/// unused tail. +/// +/// [`DEFAULT_FULL_SCAN_STOP_GAP`]: crate::config::DEFAULT_FULL_SCAN_STOP_GAP +pub(crate) const ADDRESS_POOL_TARGET_SIZE: usize = 16; + +/// A pool of pre-revealed external addresses whose derivation indices are already persisted, +/// allowing LDK's synchronous [`SignerProvider`] callbacks to obtain fresh addresses without +/// waiting on wallet persistence. +struct AddressPool { + /// Addresses ready for handout: their reveal is durably persisted, so every chain sync path + /// watches their scripts. + available: VecDeque<(u32, bitcoin::Address)>, + /// Addresses revealed in-memory whose persistence has not succeeded yet. They are published + /// to `available` by the next successful [`Wallet::refill_address_pool`] run. + unpublished: Vec<(u32, bitcoin::Address)>, +} + pub(crate) struct Wallet { // A BDK on-chain wallet. inner: Mutex>, persister: tokio::sync::Mutex, + address_pool: Mutex, + // Serializes refill runs so concurrent pops never over-reveal. + address_pool_refill_lock: tokio::sync::Mutex<()>, broadcaster: Arc, fee_estimator: Arc, chain_source: Arc, @@ -105,9 +134,14 @@ impl Wallet { ) -> Self { let inner = Mutex::new(wallet); let persister = tokio::sync::Mutex::new(wallet_persister); + let address_pool = + Mutex::new(AddressPool { available: VecDeque::new(), unpublished: Vec::new() }); + let address_pool_refill_lock = tokio::sync::Mutex::new(()); Self { inner, persister, + address_pool, + address_pool_refill_lock, broadcaster, fee_estimator, chain_source, @@ -518,34 +552,132 @@ impl Wallet { Ok(address_info.address) } - /// Returns a new address, persisting the revealed derivation index in the background rather - /// than waiting on it. + /// Loads the persisted address pool and tops it up to [`ADDRESS_POOL_TARGET_SIZE`]. /// - /// 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. + /// Must be called once before the node starts handing out pooled addresses; reloading the + /// persisted indices is what keeps restarts from burning fresh derivation indices on every + /// run. + pub(crate) async fn initialize_address_pool(&self) -> Result<(), Error> { + let persisted_indices = { + let locked_persister = self.persister.lock().await; + locked_persister.read_address_pool().await.map_err(|e| { + log_error!(self.logger, "Failed to read address pool: {}", e); + Error::PersistenceFailed + })? + }; + { + let locked_wallet = self.inner.lock().expect("lock"); + let last_revealed = locked_wallet.derivation_index(KeychainKind::External); + let mut locked_pool = self.address_pool.lock().expect("lock"); + for index in persisted_indices { + // Only trust indices the persisted wallet actually revealed: anything beyond + // `last_revealed` would hand out a script no chain sync path watches. + if last_revealed.map_or(false, |last| index <= last) { + let address = locked_wallet.peek_address(KeychainKind::External, index).address; + locked_pool.available.push_back((index, address)); + } else { + log_error!( + self.logger, + "Dropping persisted address pool index {} beyond the wallet's last revealed index", + index + ); + } + } + } + self.refill_address_pool().await + } + + /// Returns an address whose reveal is already durably persisted, or `None` if the pool is + /// exhausted. /// - /// 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. + /// This is safe to call from sync callbacks (e.g., [`SignerProvider`]) that LDK invokes on + /// runtime worker threads while holding channel locks: it never waits on persistence, only + /// popping from the pre-persisted pool and scheduling a background refill. Blocking such a + /// callback on persistence can deadlock the runtime, as other tasks blocking synchronously on + /// the same channel locks may capture the remaining workers, leaving none to drive the + /// persistence future the callback would wait on. + /// + /// Failing closed on an empty pool (rather than revealing an unpersisted address) ensures we + /// never hand out a script that would go unwatched if the node crashed before its reveal + /// landed: incremental chain syncs only query scripts the persisted wallet has revealed. + /// + /// The handout itself is not persisted: if the node restarts before the refill scheduled here + /// rewrites the pool record, the popped address may be handed out again after the restart. + /// Its reveal is durable either way, so the script always stays watched — the cost is bounded + /// address reuse, not fund visibility. + pub(crate) fn pop_pooled_address(self: &Arc) -> Option { + let popped = self.address_pool.lock().expect("lock").available.pop_front(); + 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); + if let Err(e) = wallet.refill_address_pool().await { + log_error!(wallet.logger, "Failed to refill the address pool: {}", e); } }); - address_info.address + popped.map(|(_, address)| address) + } + + /// Tops the address pool up to [`ADDRESS_POOL_TARGET_SIZE`], publishing newly revealed + /// addresses only after their reveal has been durably persisted. + pub(crate) async fn refill_address_pool(&self) -> Result<(), Error> { + let _refill_guard = self.address_pool_refill_lock.lock().await; + + { + let locked_pool = self.address_pool.lock().expect("lock"); + if locked_pool.unpublished.is_empty() + && locked_pool.available.len() >= ADDRESS_POOL_TARGET_SIZE + { + return Ok(()); + } + } + + let mut locked_persister = self.persister.lock().await; + let (change_set, indices) = { + let mut locked_wallet = self.inner.lock().expect("lock"); + let mut locked_pool = self.address_pool.lock().expect("lock"); + let needed = ADDRESS_POOL_TARGET_SIZE + .saturating_sub(locked_pool.available.len() + locked_pool.unpublished.len()); + for _ in 0..needed { + let address_info = locked_wallet.reveal_next_address(KeychainKind::External); + locked_pool.unpublished.push((address_info.index, address_info.address)); + } + let indices: Vec = locked_pool + .available + .iter() + .chain(locked_pool.unpublished.iter()) + .map(|(index, _)| *index) + .collect(); + (locked_wallet.take_staged().unwrap_or_default(), indices) + }; + + // Persist the pool record before the reveals. A crash between the two writes then leaves + // record entries the persisted wallet doesn't cover, which reloading drops and the next + // refill re-derives to the same indices — rather than durably revealed indices missing + // from the record, which no path would ever pool or hand out again (burning them). + // Writing the record first also drops popped indices from it as early as possible, + // narrowing the restart window in which a handed-out address is handed out again. + let record_res = locked_persister.persist_address_pool(indices).await; + // Attempt the change-set persist even if the record write failed: the reveals were + // already taken from the wallet, so they must reach the persister's pending change set + // (either persisted now or retained for retry) to not be lost. + let change_set_res = locked_persister.persist_changeset(change_set).await; + record_res.map_err(|e| { + log_error!(self.logger, "Failed to persist address pool: {}", e); + Error::PersistenceFailed + })?; + // On failure the reveals stay in `unpublished` (never handed out) and the persister + // retains the change set, so the next refill run retries both. + change_set_res.map_err(|e| { + log_error!(self.logger, "Failed to persist wallet: {}", e); + Error::PersistenceFailed + })?; + + // Both writes are durable, so the addresses may be handed out. + let mut locked_pool = self.address_pool.lock().expect("lock"); + let unpublished = core::mem::take(&mut locked_pool.unpublished); + locked_pool.available.extend(unpublished); + Ok(()) } pub(crate) async fn get_new_internal_address(&self) -> Result { @@ -2122,14 +2254,18 @@ impl SignerProvider for WalletKeysManager { fn get_destination_script(&self, _channel_keys_id: [u8; 32]) -> Result { // 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(); + let address = self.wallet.pop_pooled_address().ok_or_else(|| { + log_error!(self.logger, "Failed to retrieve a destination script: address pool empty"); + })?; Ok(address.script_pubkey()) } fn get_shutdown_scriptpubkey(&self) -> Result { // 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(); + let address = self.wallet.pop_pooled_address().ok_or_else(|| { + log_error!(self.logger, "Failed to retrieve a shutdown script: address pool empty"); + })?; match address.witness_program() { Some(program) => ShutdownScript::new_witness_program(&program).map_err(|e| { @@ -2184,3 +2320,430 @@ fn ldk_to_bdk_satisfaction_weight(ldk_satisfaction_weight: u64) -> Weight { .saturating_sub(EMPTY_SCRIPT_SIG_WEIGHT + EMPTY_WITNESS_COUNT_WEIGHT), ) } + +#[cfg(test)] +mod tests { + use std::sync::atomic::{AtomicBool, Ordering}; + + use bdk_wallet::Wallet as BdkWallet; + use bitcoin::Network; + use lightning::io; + use lightning::util::persist::{KVStore, PageToken, PaginatedKVStore, PaginatedListResponse}; + + use super::*; + use crate::config::EsploraSyncConfig; + use crate::io::test_utils::InMemoryStore; + use crate::io::{ + BDK_WALLET_ADDRESS_POOL_KEY, BDK_WALLET_ADDRESS_POOL_PRIMARY_NAMESPACE, + BDK_WALLET_ADDRESS_POOL_SECONDARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, + PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, + PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + }; + use crate::types::{DynStore, DynStoreWrapper}; + use crate::{NodeMetrics, PersistedNodeMetrics}; + + const EXTERNAL_DESCRIPTOR: &str = "wpkh(tprv8ZgxMBicQKsPdy6LMhUtFHAgpocR8GC6QmwMSFpZs7h6Eziw3SpThFfczTDh5rW2krkqffa11UpX3XkeTTB2FvzZKWXqPY54Y6Rq4AQ5R8L/84'/1'/0'/0/*)"; + const INTERNAL_DESCRIPTOR: &str = "wpkh(tprv8ZgxMBicQKsPdy6LMhUtFHAgpocR8GC6QmwMSFpZs7h6Eziw3SpThFfczTDh5rW2krkqffa11UpX3XkeTTB2FvzZKWXqPY54Y6Rq4AQ5R8L/84'/1'/0'/1/*)"; + + /// An in-memory store whose writes can be made to fail on demand. + #[derive(Clone)] + struct FailSwitchStore { + inner: Arc, + fail_writes: Arc, + } + + impl FailSwitchStore { + fn new() -> Self { + Self { + inner: Arc::new(InMemoryStore::new()), + fail_writes: Arc::new(AtomicBool::new(false)), + } + } + } + + impl KVStore for FailSwitchStore { + fn read( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, + ) -> impl Future, 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 fail_writes = Arc::clone(&self.fail_writes); + let primary_namespace = primary_namespace.to_string(); + let secondary_namespace = secondary_namespace.to_string(); + let key = key.to_string(); + async move { + if fail_writes.load(Ordering::Acquire) { + return Err(io::Error::new(io::ErrorKind::Other, "writes disabled")); + } + KVStore::write(&*inner, &primary_namespace, &secondary_namespace, &key, buf).await + } + } + + 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, io::Error>> + 'static + Send { + KVStore::list(&*self.inner, primary_namespace, secondary_namespace) + } + } + + impl PaginatedKVStore for FailSwitchStore { + 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, + ) + } + } + + /// Constructs a `Wallet` around the given store, either creating a fresh BDK wallet or + /// loading the one the store already holds. + async fn new_test_wallet(store: Arc, load_existing: bool) -> Arc { + let logger = Arc::new(Logger::new_log_facade()); + let mut config = Config::default(); + config.network = Network::Regtest; + let config = Arc::new(config); + + let mut wallet_persister = + KVStoreWalletPersister::new(Arc::clone(&store), Arc::clone(&logger)); + let bdk_wallet = if load_existing { + BdkWallet::load() + .descriptor(KeychainKind::External, Some(EXTERNAL_DESCRIPTOR)) + .descriptor(KeychainKind::Internal, Some(INTERNAL_DESCRIPTOR)) + .extract_keys() + .check_network(Network::Regtest) + .load_wallet_async(&mut wallet_persister) + .await + .unwrap() + .unwrap() + } else { + BdkWallet::create(EXTERNAL_DESCRIPTOR, INTERNAL_DESCRIPTOR) + .network(Network::Regtest) + .create_wallet_async(&mut wallet_persister) + .await + .unwrap() + }; + + let fee_estimator = Arc::new(OnchainFeeEstimator::new()); + let broadcaster = Arc::new(Broadcaster::new(Arc::clone(&logger))); + let node_metrics = Arc::new(PersistedNodeMetrics::new(NodeMetrics::default())); + let (chain_source, _) = ChainSource::new_esplora( + "http://localhost:1".to_string(), + HashMap::new(), + EsploraSyncConfig::default(), + Arc::clone(&fee_estimator), + Arc::clone(&broadcaster), + Arc::clone(&store), + Arc::clone(&config), + Arc::clone(&logger), + node_metrics, + ) + .unwrap(); + let payment_store = Arc::new(PaymentStore::new( + Vec::new(), + PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE.to_string(), + PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE.to_string(), + Arc::clone(&store), + Arc::clone(&logger), + )); + let pending_payment_store = Arc::new(PendingPaymentStore::new( + Vec::new(), + PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE.to_string(), + PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE.to_string(), + Arc::clone(&store), + Arc::clone(&logger), + )); + let runtime = Arc::new(Runtime::new(Arc::clone(&logger)).unwrap()); + + Arc::new(Wallet::new( + bdk_wallet, + wallet_persister, + broadcaster, + fee_estimator, + Arc::new(chain_source), + payment_store, + runtime, + config, + logger, + pending_payment_store, + )) + } + + fn pooled_indices(wallet: &Wallet) -> Vec { + wallet.address_pool.lock().unwrap().available.iter().map(|(index, _)| *index).collect() + } + + #[tokio::test] + async fn refill_publishes_addresses_only_after_their_reveal_is_persisted() { + let fail_store = FailSwitchStore::new(); + let store: Arc = Arc::new(DynStoreWrapper(fail_store.clone())); + let wallet = new_test_wallet(Arc::clone(&store), false).await; + + wallet.initialize_address_pool().await.unwrap(); + assert_eq!(pooled_indices(&wallet).len(), ADDRESS_POOL_TARGET_SIZE); + + // Simulate a handout, then make wallet writes fail: the refill must not publish the + // address it revealed, as a crash would leave its script unwatched by incremental syncs. + wallet.address_pool.lock().unwrap().available.pop_front().unwrap(); + fail_store.fail_writes.store(true, Ordering::Release); + assert!(wallet.refill_address_pool().await.is_err()); + let unpersisted_index = ADDRESS_POOL_TARGET_SIZE as u32; + let indices = pooled_indices(&wallet); + assert_eq!(indices.len(), ADDRESS_POOL_TARGET_SIZE - 1); + assert!(!indices.contains(&unpersisted_index)); + + // Once persistence recovers, the next refill publishes the retained reveal without + // burning another derivation index. + fail_store.fail_writes.store(false, Ordering::Release); + wallet.refill_address_pool().await.unwrap(); + let indices = pooled_indices(&wallet); + assert_eq!(indices.len(), ADDRESS_POOL_TARGET_SIZE); + assert!(indices.contains(&unpersisted_index)); + let last_revealed = wallet.inner.lock().unwrap().derivation_index(KeychainKind::External); + assert_eq!(last_revealed, Some(unpersisted_index)); + } + + #[tokio::test] + async fn pool_reloads_across_restarts_without_burning_indices() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + + let (popped_address, indices_before) = { + let wallet = new_test_wallet(Arc::clone(&store), false).await; + wallet.initialize_address_pool().await.unwrap(); + // Simulate a handout and a completed refill before the restart. + let (_, popped_address) = + wallet.address_pool.lock().unwrap().available.pop_front().unwrap(); + wallet.refill_address_pool().await.unwrap(); + (popped_address, pooled_indices(&wallet)) + }; + + let wallet = new_test_wallet(Arc::clone(&store), true).await; + wallet.initialize_address_pool().await.unwrap(); + + // The pool is rebuilt from the persisted record: the restart neither reveals fresh + // indices (widening what incremental syncs must watch) nor re-hands-out the address + // popped before the restart. + assert_eq!(pooled_indices(&wallet), indices_before); + let last_revealed = wallet.inner.lock().unwrap().derivation_index(KeychainKind::External); + assert_eq!(last_revealed, Some(ADDRESS_POOL_TARGET_SIZE as u32)); + let pool = wallet.address_pool.lock().unwrap(); + assert!(!pool.available.iter().any(|(_, address)| *address == popped_address)); + } + + #[tokio::test] + async fn initialize_drops_pool_indices_the_wallet_never_revealed() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + { + let wallet = new_test_wallet(Arc::clone(&store), false).await; + wallet.initialize_address_pool().await.unwrap(); + } + + // Corrupt the persisted record with an index the wallet never revealed. + let logger = Arc::new(Logger::new_log_facade()); + let mut persister = KVStoreWalletPersister::new(Arc::clone(&store), logger); + persister.persist_address_pool(vec![5, 100]).await.unwrap(); + + let wallet = new_test_wallet(Arc::clone(&store), true).await; + wallet.initialize_address_pool().await.unwrap(); + + // Index 5 was revealed before the restart and is kept; the never-revealed index 100 + // must be dropped, as no sync path would watch its script. The initial refill then + // tops the pool back up with fresh reveals. + let indices = pooled_indices(&wallet); + assert_eq!(indices.len(), ADDRESS_POOL_TARGET_SIZE); + assert!(indices.contains(&5)); + assert!(!indices.contains(&100)); + } + + #[tokio::test] + async fn signer_provider_callbacks_fail_closed_when_pool_is_empty() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(Arc::clone(&store), false).await; + let logger = Arc::new(Logger::new_log_facade()); + let keys_manager = WalletKeysManager::new(&[7u8; 32], 42, 42, Arc::clone(&wallet), logger); + + // Before the pool is initialized it is empty: the sync callbacks must fail closed + // rather than hand out an address whose reveal was never persisted. + assert!(keys_manager.get_destination_script([0u8; 32]).is_err()); + assert!(keys_manager.get_shutdown_scriptpubkey().is_err()); + + wallet.initialize_address_pool().await.unwrap(); + assert!(keys_manager.get_destination_script([0u8; 32]).is_ok()); + assert!(keys_manager.get_shutdown_scriptpubkey().is_ok()); + } + + /// An in-memory store that snapshots its full contents after every completed write, letting + /// tests reload the wallet from any crash point. + #[derive(Clone)] + struct SnapshotStore { + data: Arc>>>, + snapshots: Arc>>>>, + } + + impl SnapshotStore { + fn new() -> Self { + Self { + data: Arc::new(Mutex::new(HashMap::new())), + snapshots: Arc::new(Mutex::new(Vec::new())), + } + } + + fn from_contents(data: HashMap<(String, String, String), Vec>) -> Self { + Self { data: Arc::new(Mutex::new(data)), snapshots: Arc::new(Mutex::new(Vec::new())) } + } + } + + impl KVStore for SnapshotStore { + fn read( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, + ) -> impl Future, io::Error>> + 'static + Send { + let res = self + .data + .lock() + .unwrap() + .get(&( + primary_namespace.to_string(), + secondary_namespace.to_string(), + key.to_string(), + )) + .cloned() + .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "not found")); + async move { res } + } + + fn write( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec, + ) -> impl Future> + 'static + Send { + let mut data = self.data.lock().unwrap(); + data.insert( + (primary_namespace.to_string(), secondary_namespace.to_string(), key.to_string()), + buf, + ); + self.snapshots.lock().unwrap().push(data.clone()); + async move { Ok(()) } + } + + fn remove( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, _lazy: bool, + ) -> impl Future> + 'static + Send { + let mut data = self.data.lock().unwrap(); + data.remove(&( + primary_namespace.to_string(), + secondary_namespace.to_string(), + key.to_string(), + )); + self.snapshots.lock().unwrap().push(data.clone()); + async move { Ok(()) } + } + + fn list( + &self, primary_namespace: &str, secondary_namespace: &str, + ) -> impl Future, io::Error>> + 'static + Send { + let keys = self + .data + .lock() + .unwrap() + .keys() + .filter(|(primary, secondary, _)| { + primary == primary_namespace && secondary == secondary_namespace + }) + .map(|(_, _, key)| key.clone()) + .collect::>(); + async move { Ok(keys) } + } + } + + impl PaginatedKVStore for SnapshotStore { + fn list_paginated( + &self, primary_namespace: &str, secondary_namespace: &str, + _page_token: Option, + ) -> impl Future> + 'static + Send { + let keys = self + .data + .lock() + .unwrap() + .keys() + .filter(|(primary, secondary, _)| { + primary == primary_namespace && secondary == secondary_namespace + }) + .map(|(_, _, key)| key.clone()) + .collect::>(); + async move { Ok(PaginatedListResponse { keys, next_page_token: None }) } + } + } + + #[tokio::test] + async fn pool_survives_a_crash_at_any_point_during_refill() { + let snapshot_store = SnapshotStore::new(); + let store: Arc = Arc::new(DynStoreWrapper(snapshot_store.clone())); + let wallet = new_test_wallet(Arc::clone(&store), false).await; + // Only replay crash points from wallet creation onwards; earlier snapshots hold a + // half-created wallet, which is the builder's concern rather than the pool's. + let baseline = snapshot_store.snapshots.lock().unwrap().len(); + + wallet.initialize_address_pool().await.unwrap(); + // Simulate a handout plus the refill it schedules. + wallet.address_pool.lock().unwrap().available.pop_front().unwrap(); + wallet.refill_address_pool().await.unwrap(); + let final_derivation = + wallet.inner.lock().unwrap().derivation_index(KeychainKind::External).unwrap(); + + // Reload the wallet from every intermediate store state. No crash point may leave the + // pool unfillable or burn indices: a reload revealing past `final_derivation` means some + // reveal was durable while absent from the pool record, stranding its index as + // revealed-but-unused forever. + let snapshots = snapshot_store.snapshots.lock().unwrap().clone(); + assert!(snapshots.len() > baseline); + for snapshot in snapshots.into_iter().skip(baseline) { + let store: Arc = + Arc::new(DynStoreWrapper(SnapshotStore::from_contents(snapshot))); + let wallet = new_test_wallet(Arc::clone(&store), true).await; + wallet.initialize_address_pool().await.unwrap(); + assert_eq!(pooled_indices(&wallet).len(), ADDRESS_POOL_TARGET_SIZE); + let derivation = + wallet.inner.lock().unwrap().derivation_index(KeychainKind::External).unwrap(); + assert!(derivation <= final_derivation); + } + } + + #[tokio::test] + async fn initialize_survives_an_undecodable_pool_record() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + { + let wallet = new_test_wallet(Arc::clone(&store), false).await; + wallet.initialize_address_pool().await.unwrap(); + } + + // Corrupt the record itself: the pool is a reconstructible cache, so an undecodable + // record must not prevent the node from starting. + KVStore::write( + &*store, + BDK_WALLET_ADDRESS_POOL_PRIMARY_NAMESPACE, + BDK_WALLET_ADDRESS_POOL_SECONDARY_NAMESPACE, + BDK_WALLET_ADDRESS_POOL_KEY, + vec![0x00, 0xff], + ) + .await + .unwrap(); + + let wallet = new_test_wallet(Arc::clone(&store), true).await; + wallet.initialize_address_pool().await.unwrap(); + assert_eq!(pooled_indices(&wallet).len(), ADDRESS_POOL_TARGET_SIZE); + } +} diff --git a/src/wallet/persist.rs b/src/wallet/persist.rs index 9d33a09f9..017e9e73c 100644 --- a/src/wallet/persist.rs +++ b/src/wallet/persist.rs @@ -11,15 +11,38 @@ use std::sync::Arc; use bdk_chain::Merge; use bdk_wallet::{AsyncWalletPersister, ChangeSet}; +use lightning::impl_writeable_tlv_based; +use lightning::util::persist::KVStore; +use lightning::util::ser::{Readable, Writeable}; use crate::io::utils::{ read_bdk_wallet_change_set, write_bdk_wallet_change_descriptor, write_bdk_wallet_descriptor, write_bdk_wallet_indexer, write_bdk_wallet_local_chain, write_bdk_wallet_network, write_bdk_wallet_tx_graph, }; +use crate::io::{ + BDK_WALLET_ADDRESS_POOL_KEY, BDK_WALLET_ADDRESS_POOL_PRIMARY_NAMESPACE, + BDK_WALLET_ADDRESS_POOL_SECONDARY_NAMESPACE, +}; use crate::logger::{log_error, LdkLogger, Logger}; use crate::types::DynStore; +/// The persisted derivation indices of the wallet's address pool. +/// +/// The record is advisory and reconstructible: it is written before the reveals it references +/// are necessarily persisted, so it may briefly lead the persisted [`ChangeSet::indexer`]. +/// Readers must validate each index against the wallet's revealed range before handing out any +/// of the addresses; dropped entries are re-derived to the same indices by the next refill. +/// +/// [`ChangeSet::indexer`]: bdk_wallet::ChangeSet::indexer +struct AddressPoolRecord { + indices: Vec, +} + +impl_writeable_tlv_based!(AddressPoolRecord, { + (0, indices, required_vec), +}); + pub(crate) struct KVStoreWalletPersister { latest_change_set: Option, pending_change_set: ChangeSet, @@ -187,6 +210,55 @@ impl KVStoreWalletPersister { let _ = std::mem::take(&mut self.pending_change_set); Ok(()) } + + /// Reads the persisted address-pool derivation indices, or an empty list if none were + /// persisted yet. + pub(super) async fn read_address_pool(&self) -> Result, std::io::Error> { + let reader = match KVStore::read( + &*self.kv_store, + BDK_WALLET_ADDRESS_POOL_PRIMARY_NAMESPACE, + BDK_WALLET_ADDRESS_POOL_SECONDARY_NAMESPACE, + BDK_WALLET_ADDRESS_POOL_KEY, + ) + .await + { + Ok(reader) => reader, + Err(e) if e.kind() == lightning::io::ErrorKind::NotFound => return Ok(Vec::new()), + Err(e) => return Err(e.into()), + }; + let record = match AddressPoolRecord::read(&mut &*reader) { + Ok(record) => record, + Err(e) => { + // The record is a reconstructible cache: a decode failure (corruption, or a + // future version's incompatible encoding) at worst costs the pool's indices, so + // degrade to an empty pool rather than failing the node's startup. + log_error!(self.logger, "Dropping undecodable address pool: {}", e); + return Ok(Vec::new()); + }, + }; + Ok(record.indices) + } + + /// Persists the address-pool derivation indices. + /// + /// See [`AddressPoolRecord`] for what readers may assume about the persisted indices. + pub(super) async fn persist_address_pool( + &mut self, indices: Vec, + ) -> Result<(), std::io::Error> { + let record = AddressPoolRecord { indices }; + KVStore::write( + &*self.kv_store, + BDK_WALLET_ADDRESS_POOL_PRIMARY_NAMESPACE, + BDK_WALLET_ADDRESS_POOL_SECONDARY_NAMESPACE, + BDK_WALLET_ADDRESS_POOL_KEY, + record.encode(), + ) + .await + .map_err(|e| { + log_error!(self.logger, "Failed to persist address pool: {}", e); + e.into() + }) + } } impl AsyncWalletPersister for KVStoreWalletPersister { diff --git a/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index 03287a409..f89fb9ac8 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -298,7 +298,8 @@ impl PaginatedKVStore for WalletPersistGatedStore { // 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. +// channel open still completes (served from the pre-persisted address pool), with the pool +// refill's persistence landing 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(); @@ -351,8 +352,8 @@ async fn channel_open_completes_while_wallet_persistence_is_stalled() { // 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. + // Reopen the gate and verify the pool refill triggered by the handouts eventually persists + // its newly revealed addresses. 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(()); @@ -363,7 +364,7 @@ async fn channel_open_completes_while_wallet_persistence_is_stalled() { }; tokio::time::timeout(Duration::from_secs(common::INTEROP_TIMEOUT_SECS), persisted) .await - .expect("timed out waiting for the deferred wallet persist"); + .expect("timed out waiting for the address-pool refill to persist"); // The channel and both nodes remain fully functional. wait_for_tx(&electrsd.client, funding_txo.txid).await; @@ -374,6 +375,66 @@ async fn channel_open_completes_while_wallet_persistence_is_stalled() { expect_channel_ready_event!(node_b, node_a.node_id()); } +// The address pool's derivation indices are persisted alongside the wallet: a restart reloads +// the pooled addresses instead of revealing fresh ones, so restarts don't burn derivation +// indices (each of which incremental chain syncs would have to watch forever). Rebuild node B +// from the same store, assert the rebuild performs no wallet writes, and verify a channel open +// is served from the reloaded pool. +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn address_pool_is_reloaded_on_restart() { + 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); + + // Trust node A with no reserve so unfunded node B accepts the channel; its wallet then sees + // no activity besides the address pool itself. + let mut config_b = random_config(); + config_b.node_config.anchor_channels_config.trusted_peers_no_reserve.push(node_a.node_id()); + let mut sync_config = EsploraSyncConfig::default(); + sync_config.background_sync_config = None; + let store = WalletPersistGatedStore::new(); + + setup_builder!(builder_b, config_b.node_config); + builder_b.set_chain_source_esplora(esplora_url.clone(), Some(sync_config)); + let node_b = builder_b.build_with_store(config_b.node_entropy.into(), store.clone()).unwrap(); + node_b.start().unwrap(); + node_b.stop().unwrap(); + drop(node_b); + + // Rebuilding from the same store must reload the persisted pool rather than revealing (and + // persisting) fresh addresses. + let wallet_writes_before = store.wallet_writes_completed.load(Ordering::Acquire); + setup_builder!(builder_b, config_b.node_config); + builder_b.set_chain_source_esplora(esplora_url, Some(sync_config)); + let node_b = builder_b.build_with_store(config_b.node_entropy.into(), store.clone()).unwrap(); + assert_eq!(store.wallet_writes_completed.load(Ordering::Acquire), wallet_writes_before); + node_b.start().unwrap(); + + // The shutdown and destination scripts for the channel open are handed out of the reloaded + // pool; an empty pool would fail the open. + let address_a = node_a.onchain_payment().new_address().unwrap(); + premine_and_distribute_funds( + &bitcoind.client, + &electrsd.client, + vec![address_a], + Amount::from_sat(5_000_000), + ) + .await; + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + let funding_txo = open_channel_no_wait(&node_a, &node_b, 500_000, None, false).await; + + 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(); From 338dc8fdf938be0e722311a7953a6932f24a1127 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Fri, 7 Aug 2026 10:30:21 -0500 Subject: [PATCH 3/5] f - Serve user-facing address requests from the address pool Handing out fresh reveals while pooled addresses sit unused fragments the wallet's revealed range: every request pushes the first on-chain use further past a growing unused tail, which a from-seed restore's full scan must step over. Serving all external handouts from the pool front instead consumes the oldest revealed index first, so on-chain use compacts the unused window back down to roughly the pool size. Unlike the sync signer callbacks, these callers may wait on persistence, so the handout is only returned once the rewritten pool record is durable, preserving the previous no-reuse-across-restart guarantee for user-facing addresses. On persistence failure the popped address returns to the pool, unhanded-out. Since every node now hands out the first derivation indices rather than minting past the pool, addresses funded early are covered by any same-seed node's initial pool reveal; the force-full-scan test's previously-unknown addresses must accordingly lie beyond the pool. Co-Authored-By: Claude Fable 5 --- src/wallet/mod.rs | 267 ++++++++++++++++++++++++++++++-- tests/integration_tests_rust.rs | 6 + 2 files changed, 262 insertions(+), 11 deletions(-) diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index 2e3d966a8..0881e3081 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -538,18 +538,47 @@ impl Wallet { Ok(tx) } + /// Returns a fresh address, served from the address pool so that all external handouts + /// consume the oldest revealed index first. + /// + /// Allocating strictly in reveal order keeps the window of revealed-but-unused scripts + /// compact: as soon as a handed-out address is used on-chain, everything before it no longer + /// counts towards a from-seed restore's full-scan stop gap. Minting a fresh index here + /// instead would strand the pooled indices as an ever-growing unused tail in front of every + /// address a restore must discover. + /// + /// Unlike [`Wallet::pop_pooled_address`], this may wait on persistence, so the handout is + /// made durable before the address is returned: the awaited refill rewrites the pool record + /// (no longer containing the popped index) before topping the pool back up, so a restart + /// never hands the returned address out again. On failure the address instead returns to + /// the pool unhanded-out, with a compensating record write covering the case where the + /// failed refill had already rewritten the record. pub(crate) async fn get_new_address(&self) -> Result { - let mut locked_persister = self.persister.lock().await; - let (address_info, change_set) = { - let mut locked_wallet = self.inner.lock().expect("lock"); - let address_info = locked_wallet.reveal_next_address(KeychainKind::External); - (address_info, locked_wallet.take_staged().unwrap_or_default()) + let (index, address) = loop { + if let Some(entry) = self.address_pool.lock().expect("lock").available.pop_front() { + break entry; + } + // Another caller may pop what this refill publishes before the re-check, so loop + // rather than assuming a successful refill leaves the pool non-empty. + self.refill_address_pool().await?; }; - locked_persister.persist_changeset(change_set).await.map_err(|e| { - log_error!(self.logger, "Failed to persist wallet: {}", e); - Error::PersistenceFailed - })?; - Ok(address_info.address) + + // Force the record rewrite: a failed handout's push-back can leave the pool over its + // target size, and an early-returning refill would then leave the just-popped index + // durably recorded, handing the address out again after a restart. + match self.refill_address_pool_inner(true).await { + Ok(()) => Ok(address), + Err(e) => { + // The address was never handed out, so return it for the next caller rather + // than leaving its index revealed but unreachable. + self.address_pool.lock().expect("lock").available.push_front((index, address)); + // The refill may have failed after rewriting the record, which then durably + // excludes the pushed-back index; rewrite it from the restored pool so a crash + // before the next successful refill doesn't strand the index outside the pool. + self.rewrite_pool_record().await; + Err(e) + }, + } } /// Loads the persisted address pool and tops it up to [`ADDRESS_POOL_TARGET_SIZE`]. @@ -621,9 +650,16 @@ impl Wallet { /// Tops the address pool up to [`ADDRESS_POOL_TARGET_SIZE`], publishing newly revealed /// addresses only after their reveal has been durably persisted. pub(crate) async fn refill_address_pool(&self) -> Result<(), Error> { + self.refill_address_pool_inner(false).await + } + + /// [`Wallet::refill_address_pool`], where `force_record_rewrite` makes the pool-record + /// rewrite unconditional: a pool at or over its target size otherwise skips it, which after + /// a pop would leave the popped index in the record. + async fn refill_address_pool_inner(&self, force_record_rewrite: bool) -> Result<(), Error> { let _refill_guard = self.address_pool_refill_lock.lock().await; - { + if !force_record_rewrite { let locked_pool = self.address_pool.lock().expect("lock"); if locked_pool.unpublished.is_empty() && locked_pool.available.len() >= ADDRESS_POOL_TARGET_SIZE @@ -680,6 +716,25 @@ impl Wallet { Ok(()) } + /// Best-effort rewrite of the pool record from the pool's current contents, used to + /// re-include a pushed-back index whose handout's record write succeeded before the handout + /// failed. Failures are only logged: the pool still covers the index in memory and the next + /// successful refill rewrites the record anyway, so only a crash before then strands the + /// index outside the pool. + async fn rewrite_pool_record(&self) { + let mut locked_persister = self.persister.lock().await; + let indices: Vec = { + let locked_pool = self.address_pool.lock().expect("lock"); + locked_pool + .available + .iter() + .chain(locked_pool.unpublished.iter()) + .map(|(index, _)| *index) + .collect() + }; + let _ = locked_persister.persist_address_pool(indices).await; + } + pub(crate) async fn get_new_internal_address(&self) -> Result { let mut locked_persister = self.persister.lock().await; let (address_info, change_set) = { @@ -2722,6 +2777,196 @@ mod tests { } } + #[tokio::test] + async fn get_new_address_pops_the_oldest_pooled_address_and_persists_the_dequeue() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(Arc::clone(&store), false).await; + wallet.initialize_address_pool().await.unwrap(); + + let (front_index, front_address) = + wallet.address_pool.lock().unwrap().available.front().cloned().unwrap(); + assert_eq!(front_index, 0); + + // The handout comes from the pool front (the oldest revealed index) rather than minting + // a fresh index past the pool's unused tail, keeping the window of revealed-but-unused + // scripts compact for a from-seed restore's full scan. + let address = wallet.get_new_address().await.unwrap(); + assert_eq!(address, front_address); + let indices = pooled_indices(&wallet); + assert_eq!(indices.len(), ADDRESS_POOL_TARGET_SIZE); + assert!(!indices.contains(&front_index)); + + // The dequeue must be durable before the address is returned: a wallet reloaded from + // the store may not pool (and later re-hand-out) the returned address. + let reloaded = new_test_wallet(Arc::clone(&store), true).await; + reloaded.initialize_address_pool().await.unwrap(); + let reloaded_indices = pooled_indices(&reloaded); + assert!(!reloaded_indices.contains(&front_index)); + assert_eq!(reloaded_indices, pooled_indices(&wallet)); + } + + #[tokio::test] + async fn get_new_address_fails_closed_and_returns_the_address_to_the_pool() { + let fail_store = FailSwitchStore::new(); + let store: Arc = Arc::new(DynStoreWrapper(fail_store.clone())); + let wallet = new_test_wallet(Arc::clone(&store), false).await; + wallet.initialize_address_pool().await.unwrap(); + + let (front_index, front_address) = + wallet.address_pool.lock().unwrap().available.front().cloned().unwrap(); + + // While persistence is unavailable no address is handed out, and the popped address + // returns to the pool front: its index is neither skipped nor left unreachable. + fail_store.fail_writes.store(true, Ordering::Release); + assert!(wallet.get_new_address().await.is_err()); + let (index, address) = + wallet.address_pool.lock().unwrap().available.front().cloned().unwrap(); + assert_eq!(index, front_index); + assert_eq!(address, front_address); + assert_eq!(pooled_indices(&wallet).len(), ADDRESS_POOL_TARGET_SIZE); + + // Once persistence recovers, the very address the failed call popped is handed out. + fail_store.fail_writes.store(false, Ordering::Release); + assert_eq!(wallet.get_new_address().await.unwrap(), front_address); + } + + #[tokio::test] + async fn get_new_address_refills_an_empty_pool_before_handing_out() { + let fail_store = FailSwitchStore::new(); + let store: Arc = Arc::new(DynStoreWrapper(fail_store.clone())); + let wallet = new_test_wallet(Arc::clone(&store), false).await; + + // With the pool empty and persistence down, the call must fail closed rather than hand + // out an address whose reveal isn't durable. + fail_store.fail_writes.store(true, Ordering::Release); + assert!(wallet.get_new_address().await.is_err()); + + // With persistence available it fills the pool inline and serves from it. + fail_store.fail_writes.store(false, Ordering::Release); + let address = wallet.get_new_address().await.unwrap(); + let expected = wallet.inner.lock().unwrap().peek_address(KeychainKind::External, 0).address; + assert_eq!(address, expected); + assert_eq!(pooled_indices(&wallet).len(), ADDRESS_POOL_TARGET_SIZE); + } + + #[tokio::test] + async fn get_new_address_never_reuses_across_restarts_after_an_overfull_pool() { + let fail_store = FailSwitchStore::new(); + let store: Arc = Arc::new(DynStoreWrapper(fail_store.clone())); + let wallet = new_test_wallet(Arc::clone(&store), false).await; + wallet.initialize_address_pool().await.unwrap(); + + // A failed handout returns the popped address to the pool while the refill retains its + // unpublished reveal; the next successful refill then records and publishes all + // seventeen indices, filling the pool past its target size. + fail_store.fail_writes.store(true, Ordering::Release); + assert!(wallet.get_new_address().await.is_err()); + fail_store.fail_writes.store(false, Ordering::Release); + wallet.refill_address_pool().await.unwrap(); + assert!(pooled_indices(&wallet).len() > ADDRESS_POOL_TARGET_SIZE); + + // Handing out from the overfull pool must still durably exclude the returned address + // from the pool record before returning: a wallet reloaded from the store may never + // hand it out again. + let address = wallet.get_new_address().await.unwrap(); + + let reloaded = new_test_wallet(Arc::clone(&store), true).await; + reloaded.initialize_address_pool().await.unwrap(); + let reloaded_pool = reloaded.address_pool.lock().unwrap(); + assert!(!reloaded_pool.available.iter().any(|(_, pooled)| *pooled == address)); + } + + /// An in-memory store that can fail all writes except the address-pool record's. + #[derive(Clone)] + struct RecordOnlyStore { + inner: Arc, + fail_non_record_writes: Arc, + } + + impl RecordOnlyStore { + fn new() -> Self { + Self { + inner: Arc::new(InMemoryStore::new()), + fail_non_record_writes: Arc::new(AtomicBool::new(false)), + } + } + } + + impl KVStore for RecordOnlyStore { + fn read( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, + ) -> impl Future, 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 fail_non_record_writes = Arc::clone(&self.fail_non_record_writes); + let primary_namespace = primary_namespace.to_string(); + let secondary_namespace = secondary_namespace.to_string(); + let key = key.to_string(); + async move { + if fail_non_record_writes.load(Ordering::Acquire) + && key != BDK_WALLET_ADDRESS_POOL_KEY + { + return Err(io::Error::new(io::ErrorKind::Other, "writes disabled")); + } + KVStore::write(&*inner, &primary_namespace, &secondary_namespace, &key, buf).await + } + } + + 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, io::Error>> + 'static + Send { + KVStore::list(&*self.inner, primary_namespace, secondary_namespace) + } + } + + impl PaginatedKVStore for RecordOnlyStore { + 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, + ) + } + } + + #[tokio::test] + async fn failed_get_new_address_leaves_the_pool_record_covering_the_pool() { + let record_store = RecordOnlyStore::new(); + let store: Arc = Arc::new(DynStoreWrapper(record_store.clone())); + let wallet = new_test_wallet(Arc::clone(&store), false).await; + wallet.initialize_address_pool().await.unwrap(); + let (front_index, _) = + wallet.address_pool.lock().unwrap().available.front().cloned().unwrap(); + + // Fail everything but the pool record: the handout's record write succeeds (durably + // excluding the popped index) while the reveal flush fails, so the call fails and the + // address goes back into the pool. Its index must not be stranded by that partial + // failure: a crash right here reloads the pool from the record, and a durably revealed + // index missing from it would never be pooled or handed out again. + record_store.fail_non_record_writes.store(true, Ordering::Release); + assert!(wallet.get_new_address().await.is_err()); + record_store.fail_non_record_writes.store(false, Ordering::Release); + + let reloaded = new_test_wallet(Arc::clone(&store), true).await; + reloaded.initialize_address_pool().await.unwrap(); + assert!(pooled_indices(&reloaded).contains(&front_index)); + } + #[tokio::test] async fn initialize_survives_an_undecodable_pool_record() { let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); diff --git a/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index f89fb9ac8..e804ffbed 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -1364,6 +1364,12 @@ async fn onchain_wallet_force_full_scan_rediscovers_esplora_funds() { let address_source_config = random_config(); let node_entropy = address_source_config.node_entropy; let address_source_node = setup_node(&chain_source, address_source_config); + // Skip past the address pool every node with this seed reveals (and thus watches) on its + // first start: the funded addresses must lie beyond it to be genuinely unknown to the stale + // node's incremental sync. + for _ in 0..16 { + address_source_node.onchain_payment().new_address().unwrap(); + } let addr_1 = address_source_node.onchain_payment().new_address().unwrap(); let addr_2 = address_source_node.onchain_payment().new_address().unwrap(); address_source_node.stop().unwrap(); From 6ed64e8ebbab267bf10399f38e551d95899ade44 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Fri, 7 Aug 2026 10:39:41 -0500 Subject: [PATCH 4/5] f - Abort in-flight address pool refills at shutdown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refills ran on the joined background task set, which shutdown waits on — a refill wedged on an unresponsive store holds up every stop for the per-task timeout. Spawn them on the cancellable set instead, which the node aborts at shutdown. Completed tasks still accumulate there until shutdown; the runtime rework in #997 is what reaps them continuously. Aborting a task mid-refill must not lose wallet state: the refill used to hold the taken change set in a local across its store writes, so an abort landing there dropped reveals that were already taken from the wallet's staged state — a later refill would then publish addresses no persisted wallet state covers, recreating the unwatched-script problem the pool exists to prevent. The refill now stages the taken change set with the persister in the same critical section that takes it from the wallet, so an abort at any await leaves the reveals pending for the next persist call to flush. Co-Authored-By: Claude Fable 5 --- src/wallet/mod.rs | 140 +++++++++++++++++++++++++++++++++++++++--- src/wallet/persist.rs | 15 +++++ 2 files changed, 146 insertions(+), 9 deletions(-) diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index 0881e3081..fc2f9bd66 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -637,8 +637,12 @@ impl Wallet { pub(crate) fn pop_pooled_address(self: &Arc) -> Option { let popped = self.address_pool.lock().expect("lock").available.pop_front(); + // Spawning cancellable lets shutdown abort an in-flight refill rather than wait on it. + // Aborting mid-refill (or dropping a refill spawned during shutdown) is safe: the reveals + // are staged with the persister in the same critical section that takes them from the + // wallet, and nothing is published whose persistence the refill did not see complete. let wallet = Arc::clone(self); - self.runtime.spawn_background_task(async move { + self.runtime.spawn_cancellable_background_task(async move { if let Err(e) = wallet.refill_address_pool().await { log_error!(wallet.logger, "Failed to refill the address pool: {}", e); } @@ -669,7 +673,7 @@ impl Wallet { } let mut locked_persister = self.persister.lock().await; - let (change_set, indices) = { + let indices = { let mut locked_wallet = self.inner.lock().expect("lock"); let mut locked_pool = self.address_pool.lock().expect("lock"); let needed = ADDRESS_POOL_TARGET_SIZE @@ -678,13 +682,17 @@ impl Wallet { let address_info = locked_wallet.reveal_next_address(KeychainKind::External); locked_pool.unpublished.push((address_info.index, address_info.address)); } - let indices: Vec = locked_pool + // Hand the reveals straight to the persister: this refill may run as a task the + // runtime aborts at shutdown, and holding the taken change set across an await + // would lose the reveals if the abort lands there — a later refill run would then + // publish addresses no persisted wallet state covers. + locked_persister.stage(locked_wallet.take_staged().unwrap_or_default()); + locked_pool .available .iter() .chain(locked_pool.unpublished.iter()) .map(|(index, _)| *index) - .collect(); - (locked_wallet.take_staged().unwrap_or_default(), indices) + .collect::>() }; // Persist the pool record before the reveals. A crash between the two writes then leaves @@ -694,10 +702,9 @@ impl Wallet { // Writing the record first also drops popped indices from it as early as possible, // narrowing the restart window in which a handed-out address is handed out again. let record_res = locked_persister.persist_address_pool(indices).await; - // Attempt the change-set persist even if the record write failed: the reveals were - // already taken from the wallet, so they must reach the persister's pending change set - // (either persisted now or retained for retry) to not be lost. - let change_set_res = locked_persister.persist_changeset(change_set).await; + // Flush the staged reveals even if the record write failed: persisted now or retained + // for retry, they must not be lost. + let change_set_res = locked_persister.persist_staged().await; record_res.map_err(|e| { log_error!(self.logger, "Failed to persist address pool: {}", e); Error::PersistenceFailed @@ -2777,6 +2784,121 @@ mod tests { } } + /// An in-memory store whose writes can be made to park until aborted, signalling when a + /// write has entered the gate. + #[derive(Clone)] + struct GatedStore { + inner: Arc, + gate_writes: Arc, + write_entered: Arc, + release: Arc, + } + + impl GatedStore { + fn new() -> Self { + Self { + inner: Arc::new(InMemoryStore::new()), + gate_writes: Arc::new(AtomicBool::new(false)), + write_entered: Arc::new(tokio::sync::Notify::new()), + release: Arc::new(tokio::sync::Notify::new()), + } + } + } + + impl KVStore for GatedStore { + fn read( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, + ) -> impl Future, 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 gate_writes = Arc::clone(&self.gate_writes); + let write_entered = Arc::clone(&self.write_entered); + let release = Arc::clone(&self.release); + let primary_namespace = primary_namespace.to_string(); + let secondary_namespace = secondary_namespace.to_string(); + let key = key.to_string(); + async move { + if gate_writes.load(Ordering::Acquire) { + write_entered.notify_one(); + release.notified().await; + } + KVStore::write(&*inner, &primary_namespace, &secondary_namespace, &key, buf).await + } + } + + 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, io::Error>> + 'static + Send { + KVStore::list(&*self.inner, primary_namespace, secondary_namespace) + } + } + + impl PaginatedKVStore for GatedStore { + 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, + ) + } + } + + #[tokio::test] + async fn aborting_a_refill_mid_persist_loses_no_reveals() { + let gated_store = GatedStore::new(); + let store: Arc = Arc::new(DynStoreWrapper(gated_store.clone())); + let wallet = new_test_wallet(Arc::clone(&store), false).await; + wallet.initialize_address_pool().await.unwrap(); + + // Simulate two handouts, then a refill that is aborted (as node shutdown aborts + // cancellable tasks) while parked on its first store write. + wallet.address_pool.lock().unwrap().available.pop_front().unwrap(); + wallet.address_pool.lock().unwrap().available.pop_front().unwrap(); + gated_store.gate_writes.store(true, Ordering::Release); + let refill_wallet = Arc::clone(&wallet); + let refill_task = tokio::spawn(async move { + let _ = refill_wallet.refill_address_pool().await; + }); + gated_store.write_entered.notified().await; + refill_task.abort(); + assert!(refill_task.await.unwrap_err().is_cancelled()); + gated_store.gate_writes.store(false, Ordering::Release); + + // The aborted refill had already revealed replacements and taken them out of the + // wallet's staged change set. Those reveals must survive the abort: everything a later + // refill publishes has to be covered by persisted wallet state, or a crash would leave + // handed-out scripts unwatched by incremental syncs. + wallet.refill_address_pool().await.unwrap(); + let indices = pooled_indices(&wallet); + assert_eq!(indices.len(), ADDRESS_POOL_TARGET_SIZE); + let max_pooled = *indices.iter().max().unwrap(); + + let reloaded = new_test_wallet(Arc::clone(&store), true).await; + let persisted_last_revealed = + reloaded.inner.lock().unwrap().derivation_index(KeychainKind::External).unwrap(); + assert!( + persisted_last_revealed >= max_pooled, + "pooled index {} exceeds the persisted last revealed index {}", + max_pooled, + persisted_last_revealed + ); + } + #[tokio::test] async fn get_new_address_pops_the_oldest_pooled_address_and_persists_the_dequeue() { let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); diff --git a/src/wallet/persist.rs b/src/wallet/persist.rs index 017e9e73c..70b5a5f0f 100644 --- a/src/wallet/persist.rs +++ b/src/wallet/persist.rs @@ -199,7 +199,22 @@ impl KVStoreWalletPersister { pub(super) async fn persist_changeset( &mut self, change_set: ChangeSet, ) -> Result<(), std::io::Error> { + self.stage(change_set); + self.persist_staged().await + } + + /// Merges the given change set into the pending one without persisting it, to be flushed by + /// [`Self::persist_staged`] or any later persist call. + /// + /// Staging is synchronous, letting callers that run as abortable tasks hand a change set + /// over without an intervening await point: once staged, an abort leaves it pending for the + /// next persist rather than dropping it. + pub(super) fn stage(&mut self, change_set: ChangeSet) { self.pending_change_set.merge(change_set); + } + + /// Persists the pending change set, retaining it for retry on failure. + pub(super) async fn persist_staged(&mut self) -> Result<(), std::io::Error> { Self::persist_inner( &mut self.latest_change_set, &self.kv_store, From 075a40c809748c477f347682452cc0f80218f29f Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Fri, 7 Aug 2026 10:45:35 -0500 Subject: [PATCH 5/5] f - Account for the address pool in full-scan stop gaps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The pool keeps a fixed number of addresses standing revealed-but-unused, and a wallet restored from seed alone has no record of them: they count against a full scan's stop gap exactly like a genuine unused gap, so a restore could stop scanning before reaching funds that lie past them — with the default gap of 20, two channel opens' worth of handouts sufficed. Extend the effective stop gap by the pool size so the pool's tail can never exhaust the configured gap on its own, and document the pool size as a public constant alongside the stop-gap bounds. Handed-out scripts that have yet to appear on-chain (e.g. shutdown scripts of open channels) still count against the configured gap, as they did before the pool existed. Co-Authored-By: Claude Fable 5 --- src/chain/electrum.rs | 9 ++++++--- src/chain/esplora.rs | 7 +++++-- src/config.rs | 22 ++++++++++++++++++++++ src/wallet/mod.rs | 18 ++++++++++-------- tests/integration_tests_rust.rs | 12 +++++++++--- 5 files changed, 52 insertions(+), 16 deletions(-) diff --git a/src/chain/electrum.rs b/src/chain/electrum.rs index 59fa23a6c..5ea82b56f 100644 --- a/src/chain/electrum.rs +++ b/src/chain/electrum.rs @@ -27,8 +27,8 @@ use lightning_transaction_sync::ElectrumSyncClient; use super::WalletSyncStatus; use crate::config::{ - clamp_full_scan_stop_gap, Config, ElectrumSyncConfig, MAX_FULL_SCAN_STOP_GAP, - MIN_FULL_SCAN_STOP_GAP, + clamp_full_scan_stop_gap, Config, ElectrumSyncConfig, ADDRESS_POOL_SIZE, + MAX_FULL_SCAN_STOP_GAP, MIN_FULL_SCAN_STOP_GAP, }; use crate::error::Error; use crate::fee_estimator::{ @@ -598,7 +598,10 @@ impl ElectrumRuntimeClient { bounded ); } - bounded as usize + // Extend the gap by the address pool size: the pool keeps that many addresses standing + // revealed-but-unused, which a scan restoring the wallet from seed alone would otherwise + // count against the configured gap. + (bounded as usize).saturating_add(ADDRESS_POOL_SIZE as usize) } async fn get_incremental_sync_wallet_update( diff --git a/src/chain/esplora.rs b/src/chain/esplora.rs index 21205bd25..5da02df8e 100644 --- a/src/chain/esplora.rs +++ b/src/chain/esplora.rs @@ -20,7 +20,7 @@ use lightning_transaction_sync::EsploraSyncClient; use super::WalletSyncStatus; use crate::config::{ - clamp_full_scan_stop_gap, Config, EsploraSyncConfig, BDK_CLIENT_CONCURRENCY, + clamp_full_scan_stop_gap, Config, EsploraSyncConfig, ADDRESS_POOL_SIZE, BDK_CLIENT_CONCURRENCY, MAX_FULL_SCAN_STOP_GAP, MIN_FULL_SCAN_STOP_GAP, }; use crate::fee_estimator::{ @@ -256,7 +256,10 @@ impl EsploraChainSource { bounded ); } - bounded as usize + // Extend the gap by the address pool size: the pool keeps that many addresses standing + // revealed-but-unused, which a scan restoring the wallet from seed alone would otherwise + // count against the configured gap. + (bounded as usize).saturating_add(ADDRESS_POOL_SIZE as usize) } pub(super) async fn sync_lightning_wallet( diff --git a/src/config.rs b/src/config.rs index 772c4bd80..c62f06bf0 100644 --- a/src/config.rs +++ b/src/config.rs @@ -79,6 +79,20 @@ pub const MIN_FULL_SCAN_STOP_GAP: u32 = 1; /// Values above 1000 are clamped to 1000 when a full scan runs. pub const MAX_FULL_SCAN_STOP_GAP: u32 = 1000; +/// The number of addresses the node keeps revealed and persisted ahead of use, from which it +/// serves fresh-address requests and channel destination and shutdown scripts. +/// +/// Pooled addresses are revealed-but-unused wallet scripts, and a wallet restored from seed +/// alone has no record of them. On-chain wallet full scans therefore extend the configured stop +/// gap (e.g. [`EsploraSyncConfig::full_scan_stop_gap`]) by this amount, so that the pool's +/// unused tail can never exhaust the gap on its own. +/// +/// After a restore from seed, the pool refills from the keychain's first indices before the +/// initial full scan runs, so it can serve addresses a previous installation of the wallet +/// already handed out — possibly even used ones. The cost is address reuse, not fund +/// visibility: the reveals keep the scripts watched and the scan discovers any prior use. +pub const ADDRESS_POOL_SIZE: u32 = 16; + // The number of concurrent requests made against the API provider. pub(crate) const BDK_CLIENT_CONCURRENCY: usize = 4; @@ -550,6 +564,10 @@ pub struct EsploraSyncConfig { /// ([`MAX_FULL_SCAN_STOP_GAP`]), inclusive. Values outside this range will be clamped to the /// nearest bound and a warning will be logged when the full scan runs. /// + /// The scan extends this value by [`ADDRESS_POOL_SIZE`] to account for the addresses the + /// node keeps revealed-but-unused ahead of use, which would otherwise count against the gap + /// when restoring a wallet from seed. + /// /// **Note:** Large values can cause many Esplora requests, hit server rate limits, /// take a long time to complete, or cause syncs to fail with /// [`SyncTimeoutsConfig::onchain_wallet_sync_timeout_secs`]. @@ -601,6 +619,10 @@ pub struct ElectrumSyncConfig { /// ([`MAX_FULL_SCAN_STOP_GAP`]), inclusive. Values outside this range will be clamped to the /// nearest bound and a warning will be logged when the full scan runs. /// + /// The scan extends this value by [`ADDRESS_POOL_SIZE`] to account for the addresses the + /// node keeps revealed-but-unused ahead of use, which would otherwise count against the gap + /// when restoring a wallet from seed. + /// /// **Note:** Large values can cause many Electrum requests, hit server rate limits, /// take a long time to complete, or cause syncs to fail with /// [`SyncTimeoutsConfig::onchain_wallet_sync_timeout_secs`]. diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index fc2f9bd66..ffe5d5374 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -53,7 +53,7 @@ use lightning::util::wallet_utils::{ use lightning_invoice::RawBolt11Invoice; use persist::KVStoreWalletPersister; -use crate::config::Config; +use crate::config::{Config, ADDRESS_POOL_SIZE}; use crate::fee_estimator::{ConfirmationTarget, FeeEstimator, OnchainFeeEstimator}; use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger}; use crate::payment::store::ConfirmationStatus; @@ -82,18 +82,20 @@ pub(crate) mod ser; const DUST_LIMIT_SATS: u64 = 546; /// The number of external addresses kept revealed, persisted, and ready for handout via -/// [`Wallet::pop_pooled_address`]. +/// [`Wallet::pop_pooled_address`] and [`Wallet::get_new_address`]. /// /// Each channel open consumes two pooled addresses (one for the destination script and one for /// the upfront shutdown script), and the pool is refilled after every handout, so this bounds /// how many channels can be opened while wallet persistence is unavailable rather than steady -/// state throughput. Pooled addresses are revealed-but-unused, so this value also widens -/// incremental chain syncs accordingly and must stay below the default full-scan stop gap -/// ([`DEFAULT_FULL_SCAN_STOP_GAP`]) lest a from-seed restore's full scan stop inside the pool's -/// unused tail. +/// state throughput. /// -/// [`DEFAULT_FULL_SCAN_STOP_GAP`]: crate::config::DEFAULT_FULL_SCAN_STOP_GAP -pub(crate) const ADDRESS_POOL_TARGET_SIZE: usize = 16; +/// Pooled addresses are revealed-but-unused, widening what incremental chain syncs must watch, +/// and a wallet restored from seed alone knows nothing of them: on-chain wallet full scans +/// extend their configured stop gap by this amount so the pool's unused tail can never exhaust +/// the gap on its own. Handed-out scripts that have yet to appear on-chain (e.g. the shutdown +/// scripts of open channels) still count against the configured gap, as they did before the +/// pool existed. +pub(crate) const ADDRESS_POOL_TARGET_SIZE: usize = ADDRESS_POOL_SIZE as usize; /// A pool of pre-revealed external addresses whose derivation indices are already persisted, /// allowing LDK's synchronous [`SignerProvider`] callbacks to obtain fresh addresses without diff --git a/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index e804ffbed..6b7cecb9c 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -32,7 +32,9 @@ use common::{ }; use electrsd::corepc_node::{self, Node as BitcoinD}; use electrsd::ElectrsD; -use ldk_node::config::{AsyncPaymentsRole, EsploraSyncConfig, DEFAULT_FULL_SCAN_STOP_GAP}; +use ldk_node::config::{ + AsyncPaymentsRole, EsploraSyncConfig, ADDRESS_POOL_SIZE, DEFAULT_FULL_SCAN_STOP_GAP, +}; use ldk_node::entropy::NodeEntropy; use ldk_node::liquidity::LSPS2ServiceConfig; use ldk_node::payment::{ @@ -1367,7 +1369,7 @@ async fn onchain_wallet_force_full_scan_rediscovers_esplora_funds() { // Skip past the address pool every node with this seed reveals (and thus watches) on its // first start: the funded addresses must lie beyond it to be genuinely unknown to the stale // node's incremental sync. - for _ in 0..16 { + for _ in 0..ADDRESS_POOL_SIZE { address_source_node.onchain_payment().new_address().unwrap(); } let addr_1 = address_source_node.onchain_payment().new_address().unwrap(); @@ -1446,12 +1448,16 @@ async fn do_onchain_wallet_full_scan_stop_gap_recovers_far_funds( chain_source: TestChainSource<'_>, bitcoind: &BitcoinD, electrsd: &ElectrsD, ) { let configured_stop_gap = DEFAULT_FULL_SCAN_STOP_GAP + 5; + // Full scans extend the stop gap by the address pool size, so the funded address must lie + // that much further out to stay beyond the default gap's (extended) reach, while remaining + // within the configured gap's. + let handout_count = DEFAULT_FULL_SCAN_STOP_GAP + ADDRESS_POOL_SIZE + 5; let address_source_config = random_config(); let node_entropy = address_source_config.node_entropy; let address_source_node = setup_node(&chain_source, address_source_config); let mut far_address = None; - for _ in 0..configured_stop_gap { + for _ in 0..handout_count { far_address = Some(address_source_node.onchain_payment().new_address().unwrap()); } address_source_node.stop().unwrap();