diff --git a/src/builder.rs b/src/builder.rs index f117800996..3222f4b93f 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/chain/electrum.rs b/src/chain/electrum.rs index 59fa23a6ca..5ea82b56fc 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 21205bd252..5da02df8e5 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 772c4bd806..c62f06bf0b 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/io/mod.rs b/src/io/mod.rs index a01aa59a83..c70c68d96e 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 f8d9d521eb..ffe5d53742 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; @@ -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; @@ -81,10 +81,41 @@ 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`] 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, 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 +/// 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 +136,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, @@ -504,18 +540,208 @@ 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 (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?; + }; + + // 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`]. + /// + /// 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. + /// + /// 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(); + + // 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_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); + } + }); + + 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> { + 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 + { + return Ok(()); + } + } + let mut locked_persister = self.persister.lock().await; - let (address_info, change_set) = { + let indices = { 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 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)); + } + // 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_persister.persist_changeset(change_set).await.map_err(|e| { + + // 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; + // 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 + })?; + // 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 })?; - Ok(address_info.address) + + // 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(()) + } + + /// 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 { @@ -2090,15 +2316,19 @@ 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.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 { - 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.pop_pooled_address().ok_or_else(|| { + log_error!(self.logger, "Failed to retrieve a shutdown script: address pool empty"); })?; match address.witness_program() { @@ -2154,3 +2384,735 @@ 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); + } + } + + /// 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())); + 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())); + { + 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 9d33a09f93..70b5a5f0f5 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, @@ -176,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, @@ -187,6 +225,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 e401c82189..6b7cecb9c0 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,15 +24,17 @@ 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}; 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::{ @@ -216,6 +218,225 @@ 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 (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(); + 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 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(()); + 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 address-pool refill to 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()); +} + +// 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(); @@ -1145,6 +1366,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..ADDRESS_POOL_SIZE { + 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(); @@ -1221,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();