From 82410d6ddbeb5656d6b5d1fd8bcd552f6a1732eb Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Thu, 12 Nov 2026 10:22:33 +0100 Subject: [PATCH 1/5] Persist BDK wallet outpoint locks The wallet persister wrote every part of a BDK change set except the locked outpoints, so a lock placed on a wallet output was forgotten on restart. Store the locked outpoints under their own key and restore them when the wallet is loaded. Released locks are dropped from the stored set so it does not grow with every lock ever taken. Co-Authored-By: HAL 9000 --- src/io/mod.rs | 7 ++++++ src/io/utils.rs | 13 +++++++++++ src/wallet/persist.rs | 52 ++++++++++++++++++++++++++++++++++++++++--- src/wallet/ser.rs | 30 +++++++++++++++++++++++++ 4 files changed, 99 insertions(+), 3 deletions(-) diff --git a/src/io/mod.rs b/src/io/mod.rs index 4d229e8b0d..4b4d51612f 100644 --- a/src/io/mod.rs +++ b/src/io/mod.rs @@ -90,6 +90,13 @@ pub(crate) const BDK_WALLET_TX_GRAPH_PRIMARY_NAMESPACE: &str = "bdk_wallet"; pub(crate) const BDK_WALLET_TX_GRAPH_SECONDARY_NAMESPACE: &str = ""; pub(crate) const BDK_WALLET_TX_GRAPH_KEY: &str = "tx_graph"; +/// The BDK wallet's [`ChangeSet::locked_outpoints`] will be persisted under this key. +/// +/// [`ChangeSet::locked_outpoints`]: bdk_wallet::ChangeSet::locked_outpoints +pub(crate) const BDK_WALLET_LOCKED_OUTPOINTS_PRIMARY_NAMESPACE: &str = "bdk_wallet"; +pub(crate) const BDK_WALLET_LOCKED_OUTPOINTS_SECONDARY_NAMESPACE: &str = ""; +pub(crate) const BDK_WALLET_LOCKED_OUTPOINTS_KEY: &str = "locked_outpoints"; + /// The BDK wallet's [`ChangeSet::indexer`] will be persisted under this key. /// /// [`ChangeSet::indexer`]: bdk_wallet::ChangeSet::indexer diff --git a/src/io/utils.rs b/src/io/utils.rs index 31f3166597..5b67a399be 100644 --- a/src/io/utils.rs +++ b/src/io/utils.rs @@ -19,6 +19,7 @@ use bdk_chain::local_chain::ChangeSet as BdkLocalChainChangeSet; use bdk_chain::miniscript::{Descriptor, DescriptorPublicKey}; use bdk_chain::tx_graph::ChangeSet as BdkTxGraphChangeSet; use bdk_chain::ConfirmationBlockTime; +use bdk_wallet::locked_outpoints::ChangeSet as BdkLockedOutpointsChangeSet; use bdk_wallet::ChangeSet as BdkWalletChangeSet; use bitcoin::Network; use lightning::ln::msgs::DecodeError; @@ -721,6 +722,15 @@ impl_read_write_change_set_type!( BDK_WALLET_TX_GRAPH_KEY ); +impl_read_write_change_set_type!( + read_bdk_wallet_locked_outpoints, + write_bdk_wallet_locked_outpoints, + BdkLockedOutpointsChangeSet, + BDK_WALLET_LOCKED_OUTPOINTS_PRIMARY_NAMESPACE, + BDK_WALLET_LOCKED_OUTPOINTS_SECONDARY_NAMESPACE, + BDK_WALLET_LOCKED_OUTPOINTS_KEY +); + impl_read_write_change_set_type!( read_bdk_wallet_indexer, write_bdk_wallet_indexer, @@ -763,6 +773,9 @@ pub(crate) async fn read_bdk_wallet_change_set( read_bdk_wallet_tx_graph(&*kv_store, logger) .await? .map(|tx_graph| change_set.tx_graph = tx_graph); + read_bdk_wallet_locked_outpoints(&*kv_store, logger) + .await? + .map(|locked_outpoints| change_set.locked_outpoints = locked_outpoints); read_bdk_wallet_indexer(&*kv_store, logger).await?.map(|indexer| change_set.indexer = indexer); Ok(Some(change_set)) } diff --git a/src/wallet/persist.rs b/src/wallet/persist.rs index 6384d0fcee..584bafa98f 100644 --- a/src/wallet/persist.rs +++ b/src/wallet/persist.rs @@ -17,8 +17,8 @@ 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, + write_bdk_wallet_indexer, write_bdk_wallet_local_chain, write_bdk_wallet_locked_outpoints, + write_bdk_wallet_network, write_bdk_wallet_tx_graph, }; use crate::io::{ BDK_WALLET_ADDRESS_POOL_KEY, BDK_WALLET_ADDRESS_POOL_PRIMARY_NAMESPACE, @@ -183,6 +183,22 @@ impl KVStoreWalletPersister { .await?; } + // Written after the transaction graph so that an interrupted write remains conservative: an + // input may stay locked, but can never become available before its spending transaction is + // durable. + if !change_set.locked_outpoints.is_empty() { + latest_change_set.locked_outpoints.merge(change_set.locked_outpoints.clone()); + // Released locks carry no information once merged, so they are dropped rather than + // accumulated in the persisted set. + latest_change_set.locked_outpoints.outpoints.retain(|_, is_locked| *is_locked); + write_bdk_wallet_locked_outpoints( + &latest_change_set.locked_outpoints, + &*kv_store, + Arc::clone(&logger), + ) + .await?; + } + if !change_set.local_chain.is_empty() { latest_change_set.local_chain.merge(change_set.local_chain.clone()); write_bdk_wallet_local_chain( @@ -312,7 +328,8 @@ mod tests { use std::time::Duration; use bdk_wallet::{AsyncWalletPersister, ChangeSet, Wallet as BdkWallet}; - use bitcoin::Network; + use bitcoin::hashes::Hash; + use bitcoin::{Network, OutPoint, Txid}; use lightning::io; use lightning::util::persist::{KVStore, PageToken, PaginatedKVStore, PaginatedListResponse}; @@ -430,4 +447,33 @@ mod tests { let reloaded = AsyncWalletPersister::initialize(&mut reloaded_persister).await.unwrap(); assert_eq!(reloaded.network, Some(Network::Regtest)); } + + #[tokio::test] + async fn persists_locked_outpoints() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(Logger::new_log_facade()); + let mut persister = KVStoreWalletPersister::new(Arc::clone(&store), Arc::clone(&logger)); + AsyncWalletPersister::initialize(&mut persister).await.unwrap(); + + let mut wallet = BdkWallet::create(EXTERNAL_DESCRIPTOR, INTERNAL_DESCRIPTOR) + .network(Network::Regtest) + .create_wallet_no_persist() + .unwrap(); + persister.persist_changeset(wallet.take_staged().unwrap()).await.unwrap(); + + let outpoint = OutPoint::new(Txid::all_zeros(), 42); + wallet.lock_outpoint(outpoint); + persister.persist_changeset(wallet.take_staged().unwrap()).await.unwrap(); + + let mut reloaded_persister = + KVStoreWalletPersister::new(Arc::clone(&store), Arc::clone(&logger)); + let reloaded = AsyncWalletPersister::initialize(&mut reloaded_persister).await.unwrap(); + assert_eq!(reloaded.locked_outpoints.outpoints.get(&outpoint), Some(&true)); + + wallet.unlock_outpoint(outpoint); + persister.persist_changeset(wallet.take_staged().unwrap()).await.unwrap(); + let mut reloaded_persister = KVStoreWalletPersister::new(store, logger); + let reloaded = AsyncWalletPersister::initialize(&mut reloaded_persister).await.unwrap(); + assert_eq!(reloaded.locked_outpoints.outpoints.get(&outpoint), None); + } } diff --git a/src/wallet/ser.rs b/src/wallet/ser.rs index c6a707bcdf..80b7d25168 100644 --- a/src/wallet/ser.rs +++ b/src/wallet/ser.rs @@ -16,6 +16,7 @@ use bdk_chain::tx_graph::ChangeSet as BdkTxGraphChangeSet; use bdk_chain::DescriptorId; use bdk_wallet::descriptor::Descriptor; use bdk_wallet::keys::DescriptorPublicKey; +use bdk_wallet::locked_outpoints::ChangeSet as BdkLockedOutpointsChangeSet; use bitcoin::hashes::sha256::Hash as Sha256Hash; use bitcoin::p2p::Magic; use bitcoin::{BlockHash, Network, OutPoint, Transaction, TxOut, Txid}; @@ -304,6 +305,35 @@ impl Readable for ChangeSetDeserWrapper { } } +impl<'a> Writeable for ChangeSetSerWrapper<'a, BdkLockedOutpointsChangeSet> { + fn write(&self, writer: &mut W) -> Result<(), lightning::io::Error> { + CHANGESET_SERIALIZATION_VERSION.write(writer)?; + + encode_tlv_stream!(writer, { + (0, self.0.outpoints, required), + }); + Ok(()) + } +} + +impl Readable for ChangeSetDeserWrapper { + fn read(reader: &mut R) -> Result { + let version: u8 = Readable::read(reader)?; + if version != CHANGESET_SERIALIZATION_VERSION { + return Err(DecodeError::UnknownVersion); + } + + let mut outpoints = RequiredWrapper(None); + decode_tlv_stream!(reader, { + (0, outpoints, required), + }); + + Ok(Self(BdkLockedOutpointsChangeSet { + outpoints: outpoints.0.expect("required outpoints TLV field should be present"), + })) + } +} + impl<'a> Writeable for ChangeSetSerWrapper<'a, BTreeMap> { fn write(&self, writer: &mut W) -> Result<(), lightning::io::Error> { let len = BigSize(self.0.len() as u64); From ee24b80812011e38cecf25f054969ee01e544317 Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Thu, 12 Nov 2026 10:31:06 +0100 Subject: [PATCH 2/5] Apply wallet sends to the wallet before broadcast The wallet learned of its own sends and fee bumps only from the next chain sync. Until then a second send selected the same inputs, and a second fee bump of the same payment failed as the wallet did not know the first replacement. Apply such a transaction to the wallet when it is created, and record its payment the way a sync would. A chain source that has not seen the transaction yet reports it as gone, so a sync returns a pending send to the wallet unless that would overrule what else the chain source reported. Fixes #1117. Co-Authored-By: HAL 9000 --- src/wallet/mod.rs | 288 +++++++++++++++++++++++++++++--- tests/integration_tests_rust.rs | 81 ++++++++- 2 files changed, 343 insertions(+), 26 deletions(-) diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index 77dced3e20..07cae04e62 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -10,6 +10,7 @@ use std::future::Future; use std::ops::Deref; use std::str::FromStr; use std::sync::{Arc, Mutex, OnceLock}; +use std::time::{SystemTime, UNIX_EPOCH}; use bdk_chain::spk_client::{FullScanRequest, SyncRequest}; use bdk_chain::ChainPosition; @@ -467,6 +468,35 @@ impl Wallet { self.inner.lock().expect("lock").start_full_scan().build() } + /// A seen-at time under which `tx` takes precedence over everything it conflicts with. + /// + /// BDK keeps, of two conflicting unconfirmed transactions, the one last seen later, and drops a + /// transaction evicted at or after the time it was last seen. A transaction keeps its + /// ancestors with it, so a descendant of a conflict counts like the conflict itself. The + /// current time alone does not settle any of this: each may carry the same second. + fn next_seen_at(wallet: &bdk_wallet::Wallet, tx: &Transaction) -> u64 { + let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs(); + let graph = wallet.tx_graph(); + let latest_timestamp = graph + .direct_conflicts(tx) + .flat_map(|(_, conflict_txid)| { + std::iter::once(conflict_txid) + .chain(graph.walk_descendants(conflict_txid, |_, txid| Some(txid))) + }) + .chain(std::iter::once(tx.compute_txid())) + .flat_map(|txid| { + [ + graph.get_tx_node(txid).and_then(|node| node.last_seen), + graph.get_last_evicted(txid), + ] + .into_iter() + .flatten() + }) + .max() + .unwrap_or(0); + now.max(latest_timestamp.saturating_add(1)) + } + pub(crate) fn get_incremental_sync_request(&self) -> SyncRequest<(KeychainKind, u32)> { self.inner.lock().expect("lock").start_sync_with_revealed_spks().build() } @@ -502,9 +532,15 @@ impl Wallet { pub(crate) async fn apply_update(&self, update: impl Into) -> Result<(), Error> { let mut locked_persister = self.persister.lock().await; + let pending_sends = self.pending_send_txids().await; let events = { let mut locked_wallet = self.inner.lock().expect("lock"); - match locked_wallet.apply_update_events(update) { + let applied = locked_wallet.events_helper(|wallet| { + wallet.apply_update(update)?; + Self::restore_evicted_sends(wallet, &pending_sends); + Ok::<_, bdk_chain::local_chain::CannotConnectError>(()) + }); + match applied { Ok(events) => events, Err(e) => { log_error!(self.logger, "Sync failed due to chain connection error: {}", e); @@ -525,6 +561,63 @@ impl Wallet { Ok(()) } + /// The transactions of this node's own on-chain sends that have yet to confirm. + async fn pending_send_txids(&self) -> Vec { + self.payment_stores + .pending_payments(|pending| { + pending.details().is_some_and(|details| { + details.direction == PaymentDirection::Outbound + && details.status == PaymentStatus::Pending + }) + }) + .await + .iter() + .filter_map(|pending| match pending.details()?.kind { + // A transaction of a channel carries a type, and is LDK's to rebroadcast. + PaymentKind::Onchain { + txid, + status: ConfirmationStatus::Unconfirmed, + tx_type: None, + } => Some(txid), + _ => None, + }) + .collect() + } + + /// Returns the given sends to the wallet's view if a chain source update evicted them. + /// + /// A chain source evicts an unconfirmed transaction it does not report, which includes one it + /// has yet to see. The inputs of a send the node only just made would then be available for + /// another spend until a later update reports the send. + /// + /// A send only returns if that changes nothing else in the wallet's view: returning it over + /// a conflict, or along with a parent that is itself gone from the view, would put the + /// wallet's own guess above what the chain source reported. + fn restore_evicted_sends(wallet: &mut bdk_wallet::Wallet, txids: &[Txid]) { + // One send at a time: a send returns only once its parent has, and what the view holds + // changes with every send returned. + loop { + let in_view: HashSet = wallet.transactions().map(|tx| tx.tx_node.txid).collect(); + let graph = wallet.tx_graph(); + let returning = txids.iter().filter(|txid| !in_view.contains(*txid)).find_map(|txid| { + let tx = graph.get_tx(*txid)?; + let displaces_a_conflict = graph + .direct_conflicts(&tx) + .any(|(_, conflict_txid)| in_view.contains(&conflict_txid)); + let returns_a_parent = tx.input.iter().any(|txin| { + let parent_txid = txin.previous_output.txid; + graph.get_tx(parent_txid).is_some() && !in_view.contains(&parent_txid) + }); + (!displaces_a_conflict && !returns_a_parent).then_some(tx) + }); + let Some(tx) = returning else { + break; + }; + let seen_at = Self::next_seen_at(wallet, &tx); + wallet.apply_unconfirmed_txs([(tx, seen_at)]); + } + } + #[cfg(feature = "chain-bitcoind")] pub(crate) async fn apply_mempool_txs( &self, unconfirmed_txs: Vec<(Transaction, u64)>, evicted_txids: Vec<(Txid, u64)>, @@ -534,12 +627,14 @@ impl Wallet { } let mut locked_persister = self.persister.lock().await; + let pending_sends = self.pending_send_txids().await; let events = { let mut locked_wallet = self.inner.lock().expect("lock"); locked_wallet .events_helper(|wallet| -> Result<(), std::convert::Infallible> { wallet.apply_unconfirmed_txs(unconfirmed_txs); wallet.apply_evicted_txs(evicted_txids); + Self::restore_evicted_sends(wallet, &pending_sends); Ok(()) }) .expect("applying mempool updates cannot fail") @@ -2245,7 +2340,7 @@ impl Wallet { fee_rate.unwrap_or_else(|| self.fee_estimator.estimate_fee_rate(confirmation_target)); let mut locked_persister = self.persister.lock().await; - let (psbt, change_set) = { + let (tx, seen_at, events, change_set) = { let mut locked_wallet = self.inner.lock().expect("lock"); // Prepare the tx_builder. We properly check the reserve requirements (again) further down. @@ -2369,19 +2464,38 @@ impl Wallet { }, } - (psbt, locked_wallet.take_staged().unwrap_or_default()) + let tx = psbt.extract_tx().map_err(|e| { + log_error!(self.logger, "Failed to extract transaction: {}", e); + e + })?; + // The wallet learns of the transaction here rather than from the next sync, so that + // nothing built before then spends the same inputs. + let seen_at = Self::next_seen_at(&locked_wallet, &tx); + let events = locked_wallet.apply_unconfirmed_txs_events([(tx.clone(), seen_at)]); + (tx, seen_at, events, locked_wallet.take_staged().unwrap_or_default()) }; - locked_persister.persist_changeset(change_set).await.map_err(|e| { - log_error!(self.logger, "Failed to persist wallet: {}", e); - Error::PersistenceFailed - })?; - - let tx = psbt.extract_tx().map_err(|e| { - log_error!(self.logger, "Failed to extract transaction: {}", e); - e - })?; - let txid = tx.compute_txid(); + let recorded = async { + self.update_payment_store(events).await.map_err(|e| { + log_error!(self.logger, "Failed to update payment store: {}", e); + Error::PersistenceFailed + })?; + locked_persister.persist_changeset(change_set).await.map_err(|e| { + log_error!(self.logger, "Failed to persist wallet: {}", e); + Error::PersistenceFailed + }) + } + .await; + if let Err(e) = recorded { + // The send is reported as failed, so nothing may broadcast the transaction later on: + // it leaves the wallet's view, and its payment record with it. + self.inner.lock().expect("lock").apply_evicted_txs([(txid, seen_at)]); + if let Err(e) = self.remove_payment(&PaymentId(txid.to_byte_array())).await { + log_error!(self.logger, "Failed to remove the payment of a failed send: {}", e); + } + return Err(e); + } + self.broadcaster.broadcast(tx); match send_amount { @@ -4040,6 +4154,11 @@ impl Wallet { let new_txid = fee_bumped_tx.compute_txid(); + // See `send_to_address`: a further bump or send before the next sync builds on the + // replacement rather than on the transaction it replaces. + let seen_at = Self::next_seen_at(&locked_wallet, &fee_bumped_tx); + locked_wallet.apply_unconfirmed_txs([(fee_bumped_tx.clone(), seen_at)]); + let change_set = locked_wallet.take_staged().unwrap_or_default(); drop(locked_wallet); @@ -4060,15 +4179,33 @@ impl Wallet { ) }; - locked_persister.persist_changeset(change_set).await.map_err(|e| { - log_error!(self.logger, "Failed to persist wallet after fee bump of {}: {}", txid, e); - Error::PersistenceFailed - })?; + let recorded = async { + locked_persister.persist_changeset(change_set).await.map_err(|e| { + log_error!( + self.logger, + "Failed to persist wallet after fee bump of {}: {}", + txid, + e + ); + Error::PersistenceFailed + })?; - // Taken after the persister, the order wallet sync takes the two locks in. - let stores = self.payment_stores.lock().await; - stores.insert_or_update_payment(new_payment.clone()).await?; - self.upsert_pending_payment(&stores, new_payment, Vec::new()).await?; + // Taken after the persister, the order wallet sync takes the two locks in. + let stores = self.payment_stores.lock().await; + stores.insert_or_update_payment(new_payment.clone()).await?; + self.upsert_pending_payment(&stores, new_payment, Vec::new()).await + } + .await; + if let Err(e) = recorded { + // The bump is reported as failed and the replacement is not broadcast, so the wallet + // and the payment's record go back to the transaction it would have replaced. + self.inner.lock().expect("lock").apply_evicted_txs([(new_txid, seen_at)]); + let stores = self.payment_stores.lock().await; + if let Err(e) = stores.insert_or_update_payment(payment).await { + log_error!(self.logger, "Failed to restore the payment of a failed bump: {}", e); + } + return Err(e); + } self.broadcaster.broadcast(fee_bumped_tx); @@ -13129,4 +13266,113 @@ mod tests { assert_eq!(inputs[0].outpoint(), OutPoint { txid: confirmed_txid, vout: 0 }); assert_eq!(inputs[0].output().value, Amount::from_sat(200_000)); } + + /// Records `txid` as the transaction of a pending send of this node. + async fn record_pending_send(wallet: &Wallet, txid: Txid) { + let payment = PaymentDetails::new( + PaymentId(txid.to_byte_array()), + PaymentKind::Onchain { txid, status: ConfirmationStatus::Unconfirmed, tx_type: None }, + Some(89_000_000), + Some(1_000_000), + PaymentDirection::Outbound, + PaymentStatus::Pending, + ); + wallet.payment_stores.payment_store().insert(payment.clone()).await.unwrap(); + wallet + .payment_stores + .pending_payment_store() + .insert(PendingPaymentDetails::new(payment, Vec::new(), Vec::new())) + .await + .unwrap(); + } + + /// Lets a wallet sync report `txid` as gone from the chain source's mempool. + async fn sync_evicting(wallet: &Wallet, txid: Txid, evicted_at: u64) { + let mut tx_update = bdk_chain::TxUpdate::default(); + tx_update.evicted_ats = [(txid, evicted_at)].into(); + wallet.apply_update(Update { tx_update, ..Default::default() }).await.unwrap(); + } + + fn is_in_wallet_view(wallet: &Wallet, txid: Txid) -> bool { + wallet.inner.lock().unwrap().get_tx(txid).is_some() + } + + /// A chain source that has yet to see a send the node just made reports it as gone, which + /// must not free the send's inputs for another spend. + #[tokio::test] + async fn a_sync_evicting_a_pending_send_returns_it_to_the_wallet() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(store, false).await; + let coin = wallet_paying_tx(&wallet, 1); + let other_coin = wallet_paying_tx(&wallet, 2); + let send = tx_spending(&wallet, OutPoint::new(coin.compute_txid(), 0)); + let unrelated = tx_spending(&wallet, OutPoint::new(other_coin.compute_txid(), 0)); + insert_confirmed_tx(&wallet, coin, 1); + insert_confirmed_tx(&wallet, other_coin, 2); + insert_unconfirmed_tx(&wallet, send.clone()); + insert_unconfirmed_tx(&wallet, unrelated.clone()); + record_pending_send(&wallet, send.compute_txid()).await; + + sync_evicting(&wallet, send.compute_txid(), 101).await; + sync_evicting(&wallet, unrelated.compute_txid(), 101).await; + + assert!(is_in_wallet_view(&wallet, send.compute_txid())); + assert!(!is_in_wallet_view(&wallet, unrelated.compute_txid())); + } + + /// A send spending the change of another returns with it, whichever of the two is looked at + /// first. + #[tokio::test] + async fn evicted_sends_return_along_with_the_sends_they_spend() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(store, false).await; + let coin = wallet_paying_tx(&wallet, 1); + let parent = tx_spending(&wallet, OutPoint::new(coin.compute_txid(), 0)); + let child = tx_spending(&wallet, OutPoint::new(parent.compute_txid(), 0)); + let (parent_txid, child_txid) = (parent.compute_txid(), child.compute_txid()); + insert_confirmed_tx(&wallet, coin, 1); + insert_unconfirmed_tx(&wallet, parent); + insert_unconfirmed_tx(&wallet, child); + evict_tx(&wallet, parent_txid); + evict_tx(&wallet, child_txid); + assert!(!is_in_wallet_view(&wallet, parent_txid)); + + Wallet::restore_evicted_sends( + &mut wallet.inner.lock().unwrap(), + &[child_txid, parent_txid], + ); + + assert!(is_in_wallet_view(&wallet, parent_txid)); + assert!(is_in_wallet_view(&wallet, child_txid)); + } + + /// Returning an evicted send must not overrule what the chain source reported about other + /// transactions: neither one the send conflicts with, nor the replacement of its parent. + #[tokio::test] + async fn an_evicted_send_does_not_return_over_other_transactions() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(store, false).await; + let coin = wallet_paying_tx(&wallet, 1); + let parent = tx_spending(&wallet, OutPoint::new(coin.compute_txid(), 0)); + let mut parent_replacement = parent.clone(); + parent_replacement.output[0].value = Amount::from_sat(89_000); + let child = tx_spending(&wallet, OutPoint::new(parent.compute_txid(), 0)); + insert_confirmed_tx(&wallet, coin, 1); + wallet.inner.lock().unwrap().apply_unconfirmed_txs([ + (parent.clone(), 100), + (child.clone(), 100), + (parent_replacement.clone(), 101), + ]); + for tx in [&parent, &child] { + record_pending_send(&wallet, tx.compute_txid()).await; + } + assert!(!is_in_wallet_view(&wallet, parent.compute_txid())); + assert!(!is_in_wallet_view(&wallet, child.compute_txid())); + + sync_evicting(&wallet, child.compute_txid(), 102).await; + + assert!(is_in_wallet_view(&wallet, parent_replacement.compute_txid())); + assert!(!is_in_wallet_view(&wallet, parent.compute_txid())); + assert!(!is_in_wallet_view(&wallet, child.compute_txid())); + } } diff --git a/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index 691668c56a..891372ee05 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -1633,6 +1633,38 @@ async fn detailed_forwarded_payment_tracking() { assert_eq!(outbound_stats.total_outbound_amount_msat, amount_msat); } +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn back_to_back_onchain_sends_before_sync() { + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + // Without background sync, so that the wallet does not learn of the first send in between. + let node = setup_node(&TestChainSource::Esplora(&electrsd), random_config()); + + let funding_address = node.onchain_payment().new_address().unwrap(); + premine_and_distribute_funds( + &bitcoind.client, + &electrsd.client, + vec![funding_address], + Amount::from_sat(500_000), + ) + .await; + node.sync_wallets().unwrap(); + + let first_address = bitcoind.client.new_address().unwrap(); + let second_address = bitcoind.client.new_address().unwrap(); + let first_txid = node.onchain_payment().send_to_address(&first_address, 100_000, None).unwrap(); + let second_txid = + node.onchain_payment().send_to_address(&second_address, 100_000, None).unwrap(); + + wait_for_tx(&electrsd.client, first_txid).await; + wait_for_tx(&electrsd.client, second_txid).await; + + let mempool = bitcoind.client.get_raw_mempool().unwrap().into_model().unwrap(); + assert!( + mempool.0.contains(&first_txid) && mempool.0.contains(&second_txid), + "both back-to-back transactions must coexist in the mempool" + ); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn split_underpaid_bolt11_payment() { let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); @@ -8012,6 +8044,43 @@ async fn fs_store_persistence_backwards_compatibility() { node_new.stop().unwrap(); } +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn onchain_fee_bump_rbf_twice_before_sync() { + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + let chain_source = random_chain_source(&bitcoind, &electrsd); + let (node_a, node_b) = setup_two_nodes(&chain_source, false, false); + + let addr_a = node_a.onchain_payment().new_address().unwrap(); + let addr_b = node_b.onchain_payment().new_address().unwrap(); + premine_and_distribute_funds( + &bitcoind.client, + &electrsd.client, + vec![addr_a.clone(), addr_b], + Amount::from_sat(500_000), + ) + .await; + node_b.sync_wallets().unwrap(); + let txid = node_b.onchain_payment().send_to_address(&addr_a, 100_000, None).unwrap(); + wait_for_tx(&electrsd.client, txid).await; + node_b.sync_wallets().unwrap(); + let payment_id = PaymentId(txid.to_byte_array()); + let first_fee = node_b.payment(&payment_id).unwrap().unwrap().fee_paid_msat.unwrap(); + + let first_replacement = node_b.onchain_payment().bump_fee_rbf(payment_id, None).unwrap(); + let second_replacement = node_b.onchain_payment().bump_fee_rbf(payment_id, None).unwrap(); + assert_ne!(first_replacement, second_replacement); + let payment = node_b.payment(&payment_id).unwrap().unwrap(); + assert!(payment.fee_paid_msat.unwrap() > first_fee); + assert!( + matches!(payment.kind, PaymentKind::Onchain { txid, .. } if txid == second_replacement) + ); + wait_for_tx(&electrsd.client, second_replacement).await; + node_b.sync_wallets().unwrap(); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + node_b.sync_wallets().unwrap(); + assert_eq!(node_b.payment(&payment_id).unwrap().unwrap().status, PaymentStatus::Succeeded); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn onchain_fee_bump_rbf() { let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); @@ -8038,17 +8107,19 @@ async fn onchain_fee_bump_rbf() { let amount_to_send_sats = 100_000; let txid = node_b.onchain_payment().send_to_address(&addr_a, amount_to_send_sats, None).unwrap(); + let payment_id = PaymentId(txid.to_byte_array()); + let original_payment = node_b + .payment(&payment_id) + .expect("payment lookup must succeed") + .expect("outbound payment must be recorded before wallet sync"); + let original_fee = original_payment.fee_paid_msat.unwrap(); + wait_for_tx(&electrsd.client, txid).await; // Give the chain source time to index the unconfirmed transaction before syncing. // Without this, Esplora may not yet have the tx, causing sync to miss it and // leaving the BDK wallet graph empty. tokio::time::sleep(std::time::Duration::from_secs(5)).await; node_a.sync_wallets().unwrap(); - node_b.sync_wallets().unwrap(); - - let payment_id = PaymentId(txid.to_byte_array()); - let original_payment = node_b.payment(&payment_id).unwrap().unwrap(); - let original_fee = original_payment.fee_paid_msat.unwrap(); // Non-existent payment id let fake_txid = From 8036ea8b80813630e67e0fc87bb0ceb441d8edeb Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Tue, 22 Sep 2026 00:01:35 -0700 Subject: [PATCH 3/5] Keep replaced txid mapped across fee bumps Bumping the same on-chain payment twice left the first replacement unmapped. The payment id is derived from the original txid and each bump sets the payment's txid to the newest one, so an earlier replacement is found only through the pending-store entry's conflicting txids. The bump never added the replaced txid to that list, and since the bump applies the replacement to the wallet itself, no later sync reports the replacement to fill it in. The bump now records the replaced txid in the pending-store entry, so a replaced round that confirms or is reported by sync still resolves to the payment. Co-Authored-By: HAL 9000 --- src/wallet/mod.rs | 107 +++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 106 insertions(+), 1 deletion(-) diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index 07cae04e62..bf51cde0cd 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -4192,8 +4192,20 @@ impl Wallet { // Taken after the persister, the order wallet sync takes the two locks in. let stores = self.payment_stores.lock().await; + // Wallet sync maps a replaced transaction to its payment with `find_payment_by_txid`. + // From the second bump on, the replaced `txid` matches only through the pending-store + // entry's `conflicting_txids`. A non-empty list in an update replaces the stored one, + // so extend it rather than writing just `txid`. + let mut conflicting_txids = stores + .pending_payment(&payment.id) + .await? + .map(|pending| pending.conflicting_txids) + .unwrap_or_default(); + if !conflicting_txids.contains(&txid) { + conflicting_txids.push(txid); + } stores.insert_or_update_payment(new_payment.clone()).await?; - self.upsert_pending_payment(&stores, new_payment, Vec::new()).await + self.upsert_pending_payment(&stores, new_payment, conflicting_txids).await } .await; if let Err(e) = recorded { @@ -13375,4 +13387,97 @@ mod tests { assert!(!is_in_wallet_view(&wallet, parent.compute_txid())); assert!(!is_in_wallet_view(&wallet, child.compute_txid())); } + + /// A second bump must add the replaced txid to the pending-store entry's `conflicting_txids`. + /// The payment id is derived from the first txid and the payment's txid is the newest one, + /// so wallet sync's `TxReplaced` event for an earlier replacement resolves the payment + /// only through that list. Without it, the event finds no payment and is skipped. + #[allow(deprecated)] + #[tokio::test] + async fn bump_fee_rbf_keeps_replaced_txid_mapped() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(store, false).await; + wallet.set_channel_liveness(TestLiveness::empty()); + + // A confirmed output funds the wallet... + { + let mut locked_wallet = wallet.inner.lock().unwrap(); + let funding_tx = Transaction { + version: bitcoin::transaction::Version::TWO, + lock_time: LockTime::ZERO, + input: Vec::new(), + output: vec![TxOut { + value: Amount::from_sat(200_000), + script_pubkey: locked_wallet + .reveal_next_address(KeychainKind::External) + .address + .script_pubkey(), + }], + }; + let funding_txid = funding_tx.compute_txid(); + let block_id = BlockId { + height: locked_wallet.latest_checkpoint().height() + 1, + hash: bitcoin::BlockHash::from_byte_array([42; 32]), + }; + let mut tx_update = TxUpdate::default(); + tx_update.txs = vec![Arc::new(funding_tx)]; + tx_update.anchors = + [(ConfirmationBlockTime { block_id, confirmation_time: 1 }, funding_txid)].into(); + let chain = CheckPoint::from_block_ids([ + locked_wallet.latest_checkpoint().block_id(), + block_id, + ]) + .unwrap(); + locked_wallet + .apply_update(Update { tx_update, chain: Some(chain), ..Default::default() }) + .unwrap(); + } + + // ...which the wallet spends in a replaceable payment sitting in the mempool. This + // transaction (B) has already replaced the original (A). + let tx_b = { + let mut locked_wallet = wallet.inner.lock().unwrap(); + let recipient = ScriptBuf::new_p2wpkh(&WPubkeyHash::from_byte_array([0x42u8; 20])); + let mut builder = locked_wallet.build_tx(); + builder + .add_recipient(recipient, Amount::from_sat(100_000)) + .fee_rate(FeeRate::from_sat_per_kwu(500)); + let tx = builder.finish().unwrap().unsigned_tx; + locked_wallet.apply_unconfirmed_txs([(tx.clone(), 1)]); + tx + }; + let txid_b = tx_b.compute_txid(); + + // The stores hold what the first bump (A -> B) and its sync left behind: the payment id + // is derived from A, the payment's txid is B, and `conflicting_txids` holds A. + let txid_a = Txid::from_byte_array([0xaa; 32]); + let payment_id = PaymentId(txid_a.to_byte_array()); + let details = PaymentDetails::new( + payment_id, + PaymentKind::Onchain { + txid: txid_b, + status: ConfirmationStatus::Unconfirmed, + tx_type: None, + }, + Some(100_000_000), + Some(1_000), + PaymentDirection::Outbound, + PaymentStatus::Pending, + ); + wallet.payment_stores.payment_store().insert(details.clone()).await.unwrap(); + let entry = PendingPaymentDetails::new(details, vec![txid_a], Vec::new()); + wallet.payment_stores.pending_payment_store().insert(entry).await.unwrap(); + + // Bump again: B -> C. + let txid_c = wallet.bump_fee_rbf(payment_id, None, 0).await.unwrap(); + + let payment = wallet.payment_stores.payment(&payment_id).await.unwrap().unwrap(); + assert!(matches!(payment.kind, PaymentKind::Onchain { txid, .. } if txid == txid_c)); + + // `PaymentId(txid_b)` is not the payment id and the payment's txid is C, so B must be in + // `conflicting_txids` for wallet sync to find the payment. + let entry = wallet.payment_stores.pending_payment(&payment_id).await.unwrap().unwrap(); + assert_eq!(entry.conflicting_txids, vec![txid_a, txid_b]); + assert_eq!(wallet.find_payment_by_txid(txid_b).await.unwrap(), Some(payment_id)); + } } From b0103ad0a8013208168feac2a7b0da84e7fbc5c9 Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Thu, 12 Nov 2026 10:41:48 +0100 Subject: [PATCH 4/5] Lock channel funding inputs until confirmed The wallet learns of a channel funding transaction only from a chain sync, as LDK broadcasts it once the counterparty signed. Until then nothing kept a second channel open or an on-chain send from selecting the same inputs, leaving one of the two transactions unable to confirm. Lock the inputs of a funding transaction when it is built, and keep them out of coin selection and the spendable balance. The locks are returned when LDK discards the transaction or never takes it, and dropped once the transaction confirms. They are recorded with what the node knows about the transaction, so that a startup sweep can return the locks of a channel that is gone, as LDK does not report every transaction it lets go. Co-Authored-By: HAL 9000 --- src/event.rs | 86 ++++++-- src/lib.rs | 21 ++ src/wallet/mod.rs | 359 +++++++++++++++++++++++++++++--- src/wallet/provenance.rs | 43 +++- tests/integration_tests_rust.rs | 58 +++++- 5 files changed, 508 insertions(+), 59 deletions(-) diff --git a/src/event.rs b/src/event.rs index 88501ae6ef..b93cf4d0f2 100644 --- a/src/event.rs +++ b/src/event.rs @@ -1054,6 +1054,23 @@ where .await; match funding_transaction { Ok(final_tx) => { + // LDK reports a transaction it took and then lets go, but not one it + // never took. For the latter the wallet state held for the transaction + // is returned here. + let release_unused_funding = |tx: bitcoin::Transaction| async move { + let inputs = + tx.input.into_iter().map(|txin| txin.previous_output).collect(); + if let Err(e) = + self.wallet.release_inputs_and_outputs(inputs, tx.output).await + { + log_error!( + self.logger, + "Failed to release the inputs of an unused funding transaction: {}", + e + ); + } + }; + // Record what the transaction is before handing it to LDK, which is what // authorizes either party to broadcast it. LDK identifies the funding // output by the same script and value, and derives the channel's id from @@ -1074,17 +1091,22 @@ where counterparty_node_id, channel_id: ChannelId::v1_from_funding_outpoint(funding_txo), }; - let facts = ChannelTxFacts::new(txid).with_outputs( - &channel, - Some(UserChannelId(user_channel_id)), - ChannelOutputRole::Funding, - [vout as u32], - ); + let facts = ChannelTxFacts::new(txid) + .with_outputs( + &channel, + Some(UserChannelId(user_channel_id)), + ChannelOutputRole::Funding, + [vout as u32], + ) + .with_locked_inputs( + final_tx.input.iter().map(|txin| txin.previous_output), + ); match self.wallet.record_own_funding_tx_facts(facts).await { Ok(FactsRecordOutcome::Recorded) => {}, // Admitted however full the store is, the report can only be - // refused as too large a record, which one funding output does - // not make. Replaying would meet the same refusal, so the + // refused as too large a record, which one funding output and + // the inputs of a transaction do not make. Replaying would meet + // the same refusal, so the // channel is funded with a transaction this node will report // without a classification. Ok(FactsRecordOutcome::Incomplete) => { @@ -1101,6 +1123,8 @@ where temporary_channel_id, e, ); + // The replayed event builds another transaction. + release_unused_funding(final_tx).await; return Err(ReplayEvent()); }, } @@ -1112,6 +1136,7 @@ where ); } + let unused_tx = final_tx.clone(); let needs_manual_broadcast = self .liquidity_source .lsps2_service() @@ -1140,6 +1165,9 @@ where ) }; + if result.is_err() { + release_unused_funding(unused_tx).await; + } match result { Ok(()) => {}, Err(APIError::APIMisuseError { err }) => { @@ -2532,24 +2560,38 @@ where return Err(ReplayEvent()); } - // TODO(#1037): once inputs are locked at coin selection, `inputs` are locks this - // event returns: unlock them here. - if let FundingInfo::Contribution { inputs: _, outputs } = funding_info { + // TODO(#1037): once inputs are locked at coin selection, the `inputs` of a + // contribution are locks this event returns: unlock them here. + let (inputs, outputs) = match funding_info { + FundingInfo::Tx { transaction } => ( + transaction.input.into_iter().map(|txin| txin.previous_output).collect(), + transaction.output, + ), + FundingInfo::Contribution { inputs: _, outputs } => ( + Vec::new(), + outputs + .into_iter() + .map(|script_pubkey| bitcoin::TxOut { + value: bitcoin::Amount::ZERO, + script_pubkey, + }) + .collect(), + ), + // The funding transaction of a channel LDK left to this node to broadcast, or + // a splice round this node contributed nothing to. The event does not say what + // the transaction spent, but the wallet recorded that for one it built. + FundingInfo::OutPoint { outpoint } => { + (self.wallet.locked_funding_inputs(&outpoint.txid).await, Vec::new()) + }, + }; + if !inputs.is_empty() || !outputs.is_empty() { log_info!( self.logger, - "Reclaiming unused addresses from channel {} funding", + "Reclaiming unused wallet state from channel {} funding", channel_id, ); - - let outputs = outputs - .into_iter() - .map(|script_pubkey| bitcoin::TxOut { - value: bitcoin::Amount::ZERO, - script_pubkey, - }) - .collect(); - if let Err(e) = self.wallet.release_outputs(outputs).await { - log_error!(self.logger, "Failed reclaiming unused addresses: {}", e); + if let Err(e) = self.wallet.release_inputs_and_outputs(inputs, outputs).await { + log_error!(self.logger, "Failed reclaiming unused wallet state: {}", e); return Err(ReplayEvent()); } } diff --git a/src/lib.rs b/src/lib.rs index f47c62e538..c199560966 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -387,6 +387,27 @@ impl Node { }, ))?; + // See `Wallet::unlock_unreferenced_outpoints`. The locks still needed are those the wallet + // recorded for the funding transaction of a channel that is still around. Without that + // record for a channel awaiting its funding's confirmation, its locks cannot be told apart + // from leftovers, and every lock is kept. An output this releases although a broadcast + // transaction spends it is spent again in the wallet's view by the next sync. + let mut funding_inputs = std::collections::HashSet::new(); + let mut has_untracked_funding = false; + for channel in channels.iter().filter(|channel| channel.is_outbound) { + let Some(funding_txo) = channel.funding_txo else { + continue; + }; + let inputs = + self.runtime.block_on(self.wallet.locked_funding_inputs(&funding_txo.txid)); + has_untracked_funding |= inputs.is_empty() + && channel.confirmations.map_or(true, |confirmations| confirmations == 0); + funding_inputs.extend(inputs); + } + self.runtime.block_on(self.wallet.unlock_unreferenced_outpoints(|outpoint| { + has_untracked_funding || funding_inputs.contains(outpoint) + }))?; + // Synchronize Lightning chain state before peers or event processing can act on stale // confirmations. Each restored ChannelMonitor may have been persisted at a different tip. self.runtime.block_on(self.chain_source.initialize_wallets( diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index bf51cde0cd..2c08f41cda 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -18,7 +18,7 @@ use bdk_wallet::descriptor::ExtendedDescriptor; use bdk_wallet::error::{BuildFeeBumpError, CreateTxError}; use bdk_wallet::signer::SignersContainer; use bdk_wallet::{ - Balance, FinalizeInputOutcome, KeychainKind, LocalOutput, PersistedWallet, SignOptions, Update, + FinalizeInputOutcome, KeychainKind, LocalOutput, PersistedWallet, SignOptions, Update, WalletEvent, }; use bitcoin::address::NetworkUnchecked; @@ -553,7 +553,11 @@ impl Wallet { Error::PersistenceFailed })?; - let change_set = self.inner.lock().expect("lock").take_staged().unwrap_or_default(); + let change_set = { + let mut locked_wallet = self.inner.lock().expect("lock"); + Self::unlock_confirmed_spends(&mut locked_wallet); + locked_wallet.take_staged().unwrap_or_default() + }; locked_persister.persist_changeset(change_set).await.map_err(|e| { log_error!(self.logger, "Failed to persist wallet: {}", e); Error::PersistenceFailed @@ -618,6 +622,69 @@ impl Wallet { } } + /// Unlocks the outpoints a confirmed transaction spends. + /// + /// A lock outlives a spend that is only unconfirmed: a chain source that stops reporting the + /// spending transaction evicts it from the wallet, and the lock is then all that keeps its + /// inputs from being selected again. + fn unlock_confirmed_spends(wallet: &mut PersistedWallet) { + if wallet.list_locked_outpoints().next().is_none() { + return; + } + let confirmed: HashSet = wallet + .transactions() + .filter(|tx| tx.chain_position.is_confirmed()) + .map(|tx| tx.tx_node.txid) + .collect(); + let spent: Vec = wallet + .list_locked_outpoints() + .filter(|outpoint| { + wallet.tx_graph().outspends(*outpoint).iter().any(|txid| confirmed.contains(txid)) + }) + .collect(); + for outpoint in spent { + wallet.unlock_outpoint(outpoint); + } + } + + /// The wallet outputs locked for the channel funding transaction `txid`, as recorded when the + /// wallet built it. Empty if nothing is on record. + pub(crate) async fn locked_funding_inputs(&self, txid: &Txid) -> Vec { + self.channel_tx_facts(txid).await.map(|facts| facts.locked_inputs).unwrap_or_default() + } + + /// Unlocks every outpoint `is_referenced` does not vouch for. + /// + /// The wallet locks the inputs of a channel funding transaction or splice contribution when + /// it selects them, as it learns of the transaction spending them only from a later sync. A + /// lock is returned when LDK discards the transaction or never takes it + /// ([`Self::release_inputs_and_outputs`]), and dropped once the transaction confirms. + /// + /// LDK does not report every transaction it lets go, though: not one it lost to a restart, + /// and not a withheld funding transaction whose channel closed. This returns what such a lock + /// holds. + pub(crate) async fn unlock_unreferenced_outpoints( + &self, is_referenced: impl Fn(&OutPoint) -> bool, + ) -> Result<(), Error> { + let mut locked_persister = self.persister.lock().await; + let change_set = { + let mut locked_wallet = self.inner.lock().expect("lock"); + let unreferenced: Vec = locked_wallet + .list_locked_outpoints() + .filter(|outpoint| !is_referenced(outpoint)) + .collect(); + for outpoint in unreferenced { + log_info!(self.logger, "Unlocking wallet output {} no channel refers to", outpoint); + locked_wallet.unlock_outpoint(outpoint); + } + locked_wallet.take_staged().unwrap_or_default() + }; + locked_persister.persist_changeset(change_set).await.map_err(|e| { + log_error!(self.logger, "Failed to persist wallet: {}", e); + Error::PersistenceFailed + }) + } + #[cfg(feature = "chain-bitcoind")] pub(crate) async fn apply_mempool_txs( &self, unconfirmed_txs: Vec<(Transaction, u64)>, evicted_txids: Vec<(Txid, u64)>, @@ -1839,7 +1906,7 @@ impl Wallet { let fee_rate = fee_rate.unwrap_or_else(|| self.fee_estimator.estimate_fee_rate(confirmation_target)); let mut locked_persister = self.persister.lock().await; - let (psbt, change_set) = { + let (tx, change_set) = { let mut locked_wallet = self.inner.lock().expect("lock"); let mut tx_builder = locked_wallet.build_tx(); tx_builder.add_recipient(output_script, amount).fee_rate(fee_rate).nlocktime(locktime); @@ -1871,17 +1938,24 @@ impl Wallet { }, } - (psbt, locked_wallet.take_staged().unwrap_or_default()) - }; - locked_persister.persist_changeset(change_set).await.map_err(|e| { - log_error!(self.logger, "Failed to persist wallet: {}", e); - Error::PersistenceFailed - })?; + let tx = psbt.extract_tx().map_err(|e| { + log_error!(self.logger, "Failed to extract transaction: {}", e); + e + })?; - let tx = psbt.extract_tx().map_err(|e| { - log_error!(self.logger, "Failed to extract transaction: {}", e); - e - })?; + // See `unlock_unreferenced_outpoints` for what a lock is for and what returns it. + // + // A funding transaction an LSPS2 service withholds until its payment is claimed is + // locked as well, and nothing reports it if the payment never is: its inputs then stay + // locked for as long as the channel exists. + for txin in &tx.input { + locked_wallet.lock_outpoint(txin.previous_output); + } + + (tx, locked_wallet.take_staged().unwrap_or_default()) + }; + let locked = tx.input.iter().map(|txin| txin.previous_output); + self.persist_new_locks(&mut locked_persister, change_set, locked).await?; Ok(tx) } @@ -2106,9 +2180,36 @@ impl Wallet { /// Frees the wallet addresses paid by `outputs` for reuse, for a transaction that will not be /// made after all. pub(crate) async fn release_outputs(&self, outputs: Vec) -> Result<(), Error> { + self.release_inputs_and_outputs(Vec::new(), outputs).await + } + + /// Persists a change set that locks `newly_locked`. If that fails the caller does not hand + /// out what the locks were taken for, and nothing would ever return them, so they are dropped. + async fn persist_new_locks( + &self, persister: &mut KVStoreWalletPersister, change_set: bdk_wallet::ChangeSet, + newly_locked: impl Iterator, + ) -> Result<(), Error> { + persister.persist_changeset(change_set).await.map_err(|e| { + log_error!(self.logger, "Failed to persist wallet: {}", e); + let mut locked_wallet = self.inner.lock().expect("lock"); + for outpoint in newly_locked { + locked_wallet.unlock_outpoint(outpoint); + } + Error::PersistenceFailed + }) + } + + /// Returns what the wallet held for a transaction that will not be made after all: unlocks + /// `inputs` and frees the wallet addresses paid by `outputs` for reuse. + pub(crate) async fn release_inputs_and_outputs( + &self, inputs: Vec, outputs: Vec, + ) -> Result<(), Error> { let mut locked_persister = self.persister.lock().await; let change_set = { let mut locked_wallet = self.inner.lock().expect("lock"); + for outpoint in inputs { + locked_wallet.unlock_outpoint(outpoint); + } for txout in outputs { if let Some((keychain, index)) = locked_wallet.derivation_of_spk(txout.script_pubkey) @@ -2130,27 +2231,48 @@ impl Wallet { pub(crate) fn get_balances( &self, total_anchor_channels_reserve_sats: u64, ) -> Result<(u64, u64), Error> { - let balance = self.inner.lock().expect("lock").balance(); + let locked_wallet = self.inner.lock().expect("lock"); + let balances = self.get_balances_inner(&locked_wallet, total_anchor_channels_reserve_sats); + #[cfg(debug_assertions)] + let has_confirmed_funds = locked_wallet.balance().confirmed != Amount::ZERO; + drop(locked_wallet); // Make sure `list_confirmed_utxos` returns at least one `Utxo` we could use to spend/bump // Anchors if we have any confirmed amounts. #[cfg(debug_assertions)] - if balance.confirmed != Amount::ZERO { + if has_confirmed_funds { debug_assert!( self.list_confirmed_utxos_inner().map_or(false, |v| !v.is_empty()), "Confirmed amounts should always be available for Anchor spending" ); } - self.get_balances_inner(balance, total_anchor_channels_reserve_sats) + balances } fn get_balances_inner( - &self, balance: Balance, total_anchor_channels_reserve_sats: u64, + &self, wallet: &PersistedWallet, + total_anchor_channels_reserve_sats: u64, ) -> Result<(u64, u64), Error> { + let balance = wallet.balance(); + // Locked outputs still belong to the wallet, but are promised to a pending funding. Only + // those the trusted spendable balance counts are taken off it: confirmed outputs, and + // unconfirmed change. + let locked_amount_sats: u64 = wallet + .list_locked_unspent() + .filter_map(|outpoint| wallet.get_utxo(outpoint)) + .filter(|utxo| { + utxo.chain_position.is_confirmed() || utxo.keychain == KeychainKind::Internal + }) + .map(|utxo| utxo.txout.value.to_sat()) + .sum(); let (total, spendable) = ( balance.total().to_sat(), - balance.trusted_spendable().to_sat().saturating_sub(total_anchor_channels_reserve_sats), + balance + .trusted_spendable() + .to_sat() + .saturating_sub(total_anchor_channels_reserve_sats) + .saturating_sub(locked_amount_sats), ); Ok((total, spendable)) @@ -2201,9 +2323,10 @@ impl Wallet { &self, locked_wallet: &mut PersistedWallet, drain_script: ScriptBuf, cur_anchor_reserve_sats: u64, fee_rate: FeeRate, ) -> Result<(u64, Psbt), Error> { - let balance = locked_wallet.balance(); - let spendable_amount_sats = - self.get_balances_inner(balance, cur_anchor_reserve_sats).map(|(_, s)| s).unwrap_or(0); + let spendable_amount_sats = self + .get_balances_inner(&locked_wallet, cur_anchor_reserve_sats) + .map(|(_, s)| s) + .unwrap_or(0); if spendable_amount_sats == 0 { log_error!( @@ -2284,6 +2407,9 @@ impl Wallet { let tip_height = locked_wallet.latest_checkpoint().height(); let mut inputs = Vec::new(); for utxo in locked_wallet.list_unspent() { + if locked_wallet.is_outpoint_locked(utxo.outpoint) { + continue; + } let Some(confirmation_height) = utxo.chain_position.confirmation_height_upper_bound() else { continue; @@ -2402,9 +2528,8 @@ impl Wallet { amount_sats, cur_anchor_reserve_sats, } => { - let balance = locked_wallet.balance(); let spendable_amount_sats = self - .get_balances_inner(balance, cur_anchor_reserve_sats) + .get_balances_inner(&locked_wallet, cur_anchor_reserve_sats) .map(|(_, s)| s) .unwrap_or(0); let tx_fee_sats = locked_wallet @@ -2429,9 +2554,8 @@ impl Wallet { } }, OnchainSendAmount::AllRetainingReserve { cur_anchor_reserve_sats } => { - let balance = locked_wallet.balance(); let spendable_amount_sats = self - .get_balances_inner(balance, cur_anchor_reserve_sats) + .get_balances_inner(&locked_wallet, cur_anchor_reserve_sats) .map(|(_, s)| s) .unwrap_or(0); let (sent, received) = locked_wallet.sent_and_received(&psbt.unsigned_tx); @@ -2625,6 +2749,8 @@ impl Wallet { .filter(|t| t.chain_position.is_confirmed()) .map(|t| t.tx_node.txid) .collect(); + // Locked outputs are listed as well: bumping the anchor of a closing channel may not wait + // for a funding that is still being negotiated, and takes its inputs if it has to. let unspent_confirmed_utxos = locked_wallet.list_unspent().filter(|u| confirmed_txs.contains(&u.outpoint.txid)); @@ -4110,9 +4236,10 @@ impl Wallet { })? .to_sat(); let additional_fee_sats = replacement_fee_sats.saturating_sub(old_fee_sats); - let balance = locked_wallet.balance(); - let spendable_amount_sats = - self.get_balances_inner(balance, cur_anchor_reserve_sats).map(|(_, s)| s).unwrap_or(0); + let spendable_amount_sats = self + .get_balances_inner(&locked_wallet, cur_anchor_reserve_sats) + .map(|(_, s)| s) + .unwrap_or(0); if spendable_amount_sats < additional_fee_sats { log_error!( self.logger, @@ -4473,7 +4600,11 @@ impl Listen for Wallet { return; } - let change_set = self.inner.lock().expect("lock").take_staged().unwrap_or_default(); + let change_set = { + let mut locked_wallet = self.inner.lock().expect("lock"); + Self::unlock_confirmed_spends(&mut locked_wallet); + locked_wallet.take_staged().unwrap_or_default() + }; if let Err(e) = locked_persister.persist_changeset(change_set).await { log_error!(self.logger, "Failed to persist on-chain wallet: {}", e); return; @@ -13480,4 +13611,174 @@ mod tests { assert_eq!(entry.conflicting_txids, vec![txid_a, txid_b]); assert_eq!(wallet.find_payment_by_txid(txid_b).await.unwrap(), Some(payment_id)); } + + /// Builds a channel funding transaction of `amount_sat` from the wallet's coins. + async fn test_funding_tx(wallet: &Wallet, amount_sat: u64) -> Result { + let funding_script = ScriptBuf::new_p2wsh(&bitcoin::WScriptHash::from_byte_array([7; 32])); + wallet + .create_funding_transaction( + funding_script, + Amount::from_sat(amount_sat), + ConfirmationTarget::ChannelFunding, + LockTime::ZERO, + None, + ) + .await + } + + fn locked_outpoint_count(wallet: &Wallet) -> usize { + wallet.inner.lock().unwrap().list_locked_outpoints().count() + } + + fn spent_outpoints(tx: &Transaction) -> Vec { + tx.input.iter().map(|txin| txin.previous_output).collect() + } + + fn all_locked(wallet: &Wallet, outpoints: &[OutPoint]) -> bool { + let locked_wallet = wallet.inner.lock().unwrap(); + outpoints.iter().all(|outpoint| locked_wallet.is_outpoint_locked(*outpoint)) + } + + fn none_locked(wallet: &Wallet, outpoints: &[OutPoint]) -> bool { + let locked_wallet = wallet.inner.lock().unwrap(); + outpoints.iter().all(|outpoint| !locked_wallet.is_outpoint_locked(*outpoint)) + } + + /// The wallet has not seen a funding transaction it built, so only the locks on its inputs + /// keep a second funding transaction from spending them too. + #[tokio::test] + async fn funding_transactions_built_back_to_back_spend_distinct_inputs() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(Arc::clone(&store), false).await; + insert_confirmed_tx(&wallet, wallet_paying_tx(&wallet, 1), 1); + insert_confirmed_tx(&wallet, wallet_paying_tx(&wallet, 2), 2); + + let first = spent_outpoints(&test_funding_tx(&wallet, 50_000).await.unwrap()); + let second = spent_outpoints(&test_funding_tx(&wallet, 50_000).await.unwrap()); + assert!(first.iter().all(|outpoint| !second.contains(outpoint))); + assert!(all_locked(&wallet, &first) && all_locked(&wallet, &second)); + + // Both coins are spoken for. + assert!(test_funding_tx(&wallet, 50_000).await.is_err()); + assert_eq!(wallet.get_balances(0).unwrap(), (180_000, 0)); + assert_eq!(wallet.list_confirmed_utxos_inner().unwrap().len(), 2); + assert!(wallet.list_confirmed_funding_inputs().unwrap().is_empty()); + + drop(wallet); + let wallet = new_test_wallet(store, true).await; + assert!(all_locked(&wallet, &first) && all_locked(&wallet, &second)); + } + + #[tokio::test] + async fn an_unused_funding_transaction_returns_its_inputs() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(Arc::clone(&store), false).await; + insert_confirmed_tx(&wallet, wallet_paying_tx(&wallet, 1), 1); + + let unused = test_funding_tx(&wallet, 50_000).await.unwrap(); + let inputs = spent_outpoints(&unused); + wallet.release_inputs_and_outputs(inputs.clone(), unused.output).await.unwrap(); + + drop(wallet); + let wallet = new_test_wallet(store, true).await; + assert!(none_locked(&wallet, &inputs)); + assert_eq!(spent_outpoints(&test_funding_tx(&wallet, 50_000).await.unwrap()), inputs); + } + + /// A funding transaction seen unconfirmed can still be evicted, which would free its inputs + /// were they not locked. Its confirmation is what makes the locks redundant. + #[tokio::test] + async fn funding_input_locks_last_until_the_funding_confirms() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(store, false).await; + insert_confirmed_tx(&wallet, wallet_paying_tx(&wallet, 1), 1); + + let funding_tx = test_funding_tx(&wallet, 50_000).await.unwrap(); + let funding_txid = funding_tx.compute_txid(); + let inputs = spent_outpoints(&funding_tx); + + let mut tx_update = bdk_chain::TxUpdate::default(); + tx_update.txs = vec![Arc::new(funding_tx)]; + tx_update.seen_ats = [(funding_txid, 100)].into(); + wallet.apply_update(Update { tx_update, ..Default::default() }).await.unwrap(); + assert!(all_locked(&wallet, &inputs)); + + sync_evicting(&wallet, funding_txid, 101).await; + assert!(all_locked(&wallet, &inputs)); + assert!(test_funding_tx(&wallet, 50_000).await.is_err()); + + let block = block_id_at(2); + let chain = wallet.inner.lock().unwrap().latest_checkpoint().insert(block); + let mut tx_update = bdk_chain::TxUpdate::default(); + tx_update.anchors = + [(ConfirmationBlockTime { block_id: block, confirmation_time: 100 }, funding_txid)] + .into(); + wallet + .apply_update(Update { tx_update, chain: Some(chain), ..Default::default() }) + .await + .unwrap(); + assert!(none_locked(&wallet, &inputs)); + } + + #[tokio::test] + async fn only_unreferenced_locks_are_unlocked() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(Arc::clone(&store), false).await; + insert_confirmed_tx(&wallet, wallet_paying_tx(&wallet, 1), 1); + insert_confirmed_tx(&wallet, wallet_paying_tx(&wallet, 2), 2); + let kept = spent_outpoints(&test_funding_tx(&wallet, 50_000).await.unwrap()); + let dropped = spent_outpoints(&test_funding_tx(&wallet, 50_000).await.unwrap()); + + wallet.unlock_unreferenced_outpoints(|outpoint| kept.contains(outpoint)).await.unwrap(); + + drop(wallet); + let wallet = new_test_wallet(store, true).await; + assert!(all_locked(&wallet, &kept)); + assert!(none_locked(&wallet, &dropped)); + } + + /// A funding transaction that could not be persisted is not handed out, so nobody would ever + /// return the locks on its inputs. + #[tokio::test] + async fn a_funding_transaction_that_fails_to_persist_holds_no_locks() { + let fail_store = + FailSwitchStore::failing_only(crate::io::BDK_WALLET_LOCKED_OUTPOINTS_PRIMARY_NAMESPACE); + let store: Arc = Arc::new(DynStoreWrapper(fail_store.clone())); + let wallet = new_test_wallet(Arc::clone(&store), false).await; + insert_confirmed_tx(&wallet, wallet_paying_tx(&wallet, 1), 1); + + fail_store.fail_writes.store(true, Ordering::Release); + assert!(matches!(test_funding_tx(&wallet, 50_000).await, Err(Error::PersistenceFailed))); + assert_eq!(locked_outpoint_count(&wallet), 0); + + // The persister retries what it failed to write along with the next change, which must + // not make the locks durable after all. + fail_store.fail_writes.store(false, Ordering::Release); + wallet.release_outputs(Vec::new()).await.unwrap(); + drop(wallet); + let wallet = new_test_wallet(store, true).await; + assert_eq!(locked_outpoint_count(&wallet), 0); + assert!(test_funding_tx(&wallet, 50_000).await.is_ok()); + } + + /// A funding transaction LDK lets go without saying what it spent is looked up in what the + /// wallet recorded when it built the transaction. + #[tokio::test] + async fn locked_funding_inputs_are_read_back_from_the_recorded_facts() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(store, false).await; + insert_confirmed_tx(&wallet, wallet_paying_tx(&wallet, 1), 1); + let funding_tx = test_funding_tx(&wallet, 50_000).await.unwrap(); + let txid = funding_tx.compute_txid(); + let inputs = spent_outpoints(&funding_tx); + assert!(wallet.locked_funding_inputs(&txid).await.is_empty()); + + let facts = ChannelTxFacts::new(txid).with_locked_inputs(inputs.clone()); + assert_eq!( + wallet.record_own_funding_tx_facts(facts).await.unwrap(), + FactsRecordOutcome::Recorded + ); + + assert_eq!(wallet.locked_funding_inputs(&txid).await, inputs); + } } diff --git a/src/wallet/provenance.rs b/src/wallet/provenance.rs index 445f0d37b3..819e5ac498 100644 --- a/src/wallet/provenance.rs +++ b/src/wallet/provenance.rs @@ -20,7 +20,7 @@ use std::sync::{Arc, Mutex, Weak}; use bitcoin::hashes::Hash; use bitcoin::secp256k1::PublicKey; -use bitcoin::{Sequence, Transaction, Txid}; +use bitcoin::{OutPoint, Sequence, Transaction, Txid}; use lightning::ln::channelmanager::PaymentId; use lightning::ln::types::ChannelId; use lightning::util::persist::PageToken; @@ -136,6 +136,10 @@ pub(crate) struct ChannelTxFacts { /// This node's share of an interactive-funding candidate, and the funding record it belongs /// to. pub local_figures: Option, + /// The wallet outputs locked for this transaction when the wallet built it to fund a channel. + /// They tell which locks the transaction accounts for, and are what to unlock should it never + /// be made. + pub locked_inputs: Vec, /// The chain tip this node was at when it last learned something new about the transaction. /// It dates the record for retention; it is not a fact about the transaction, and so is the /// one part of a record a later report may move. @@ -148,6 +152,7 @@ impl_writeable_tlv_based!(ChannelTxFacts, { (4, self_role, option), (6, local_figures, option), (8, recorded_at_height, required), + (10, locked_inputs, optional_vec), }); impl ChannelTxFacts { @@ -158,10 +163,17 @@ impl ChannelTxFacts { outputs: Vec::new(), self_role: None, local_figures: None, + locked_inputs: Vec::new(), recorded_at_height: 0, } } + /// Records `inputs` as the wallet outputs locked for this transaction. + pub(crate) fn with_locked_inputs(mut self, inputs: impl IntoIterator) -> Self { + self.locked_inputs.extend(inputs); + self + } + /// Dates these facts at the chain tip the node is at while reporting them. pub(crate) fn reported_at_height(mut self, height: u32) -> Self { self.recorded_at_height = height; @@ -322,6 +334,12 @@ impl ChannelTxFacts { _ => {}, } + // One report covers all of them: the wallet builds a funding transaction once. + if self.locked_inputs.is_empty() && !incoming.locked_inputs.is_empty() { + self.locked_inputs = incoming.locked_inputs.clone(); + changed = true; + } + if !changed { return Ok(None); } @@ -940,6 +958,7 @@ mod tests { .with_outputs(&channel, Some(UserChannelId(42)), ChannelOutputRole::Funding, [0]) .with_outputs(&channel, None, ChannelOutputRole::Anchor, [1]) .with_outputs(&channel, None, ChannelOutputRole::Htlc, [2, 3]) + .with_locked_inputs([OutPoint::new(test_txid(8), 1), OutPoint::new(test_txid(9), 0)]) .with_outputs(&channel, None, ChannelOutputRole::Spendable, [4]) .with_self_role(TransactionType::InteractiveFunding { channels: vec![channel.clone()], @@ -1116,6 +1135,28 @@ mod tests { ); } + /// The channel's own report of its funding output arrives after the wallet's, and carries no + /// inputs: it must not take the recorded ones away. + #[test] + fn locked_inputs_survive_a_later_report_without_them() { + let channel = test_channel(1); + let locked = vec![OutPoint::new(test_txid(8), 1)]; + let built = ChannelTxFacts::new(test_txid(7)).with_locked_inputs(locked.clone()); + let pending = ChannelTxFacts::new(test_txid(7)).with_outputs( + &channel, + Some(UserChannelId(42)), + ChannelOutputRole::Funding, + [0], + ); + + let merged = built.clone().merged_with(&pending).unwrap().unwrap(); + assert_eq!(merged.locked_inputs, locked); + assert_eq!(merged.outputs.len(), 1); + + let merged = pending.merged_with(&built).unwrap().unwrap(); + assert_eq!(merged.locked_inputs, locked); + } + #[test] fn a_second_role_for_one_output_is_rejected() { let channel = test_channel(1); diff --git a/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index 891372ee05..5511f4a395 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -1368,10 +1368,8 @@ async fn multi_hop_sending() { open_channel(&nodes[0], &nodes[1], 100_000, true, &electrsd).await; open_channel(&nodes[1], &nodes[2], 1_000_000, true, &electrsd).await; - // We need to sync wallets in-between back-to-back channel opens from the same node so BDK - // wallet picks up on the broadcast funding tx and doesn't double-spend itself. - // - // TODO: Remove once fixed in BDK. + // The wallet learns of a funding transaction's change only from a sync, and the second + // channel is funded from the change of the first. nodes[1].sync_wallets().unwrap(); open_channel(&nodes[1], &nodes[3], 1_000_000, true, &electrsd).await; open_channel(&nodes[2], &nodes[4], 1_000_000, true, &electrsd).await; @@ -1665,6 +1663,46 @@ async fn back_to_back_onchain_sends_before_sync() { ); } +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn back_to_back_channel_opens_before_sync() { + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + // Without background sync, so that the wallet does not learn of the first funding in between. + let chain_source = TestChainSource::Esplora(&electrsd); + let node_a = setup_node(&chain_source, random_config()); + let node_b = setup_node(&chain_source, random_config()); + + let funding_address = node_a.onchain_payment().new_address().unwrap(); + premine_and_distribute_funds( + &bitcoind.client, + &electrsd.client, + vec![funding_address], + Amount::from_sat(500_000), + ) + .await; + node_a.sync_wallets().unwrap(); + + let first_funding_txo = open_channel_no_wait(&node_a, &node_b, 100_000, None, false).await; + + // The wallet's only coin is spent by the first channel's funding transaction, and its change + // is unknown to the wallet until the next sync. + let node_b_address = node_b.listening_addresses().unwrap().first().unwrap().clone(); + assert_eq!( + node_a.open_channel(node_b.node_id(), node_b_address, 100_000, None, None, None), + Err(NodeError::InsufficientFunds) + ); + + wait_for_tx(&electrsd.client, first_funding_txo.txid).await; + node_a.sync_wallets().unwrap(); + let second_funding_txo = open_channel_no_wait(&node_a, &node_b, 100_000, None, false).await; + wait_for_tx(&electrsd.client, second_funding_txo.txid).await; + + let mempool = bitcoind.client.get_raw_mempool().unwrap().into_model().unwrap(); + assert!( + mempool.0.contains(&first_funding_txo.txid) && mempool.0.contains(&second_funding_txo.txid), + "both funding transactions must coexist in the mempool" + ); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn split_underpaid_bolt11_payment() { let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); @@ -4738,7 +4776,7 @@ const CLOSED_CHANNEL_PAYMENT_RESOLVED: &str = "of closed channel"; /// tracks once it promoted a splice round to the channel's funding, however it does. const PROMOTED_ROUND_PAYMENT_RESOLVED: &str = "once splice round"; /// Logged by a node as it returns the addresses of a contribution LDK discarded to the wallet. -const RECLAIMED_ADDRESSES: &str = "Reclaiming unused addresses from channel"; +const RECLAIMED_ADDRESSES: &str = "Reclaiming unused wallet state from channel"; /// Logged by a node once it has decided the funding payments of a closed channel by the rounds the /// channel's monitor watches, at `ChannelClosed` and for a round LDK discards after the close. /// Unlike [`CLOSED_CHANNEL_PAYMENT_RESOLVED`], logged whatever was found, so also when no payment @@ -6949,16 +6987,22 @@ async fn do_lsps2_client_service_integration(client_trusts_lsp: bool) { let payer_node = payer_builder.build(payer_config.node_entropy.into()).unwrap(); payer_node.start().unwrap(); - let service_addr = service_node.onchain_payment().new_address().unwrap(); + // The service opens four JIT channels below without syncing in between. Each needs a coin of + // its own: the wallet learns of a funding transaction's change only from a sync, and the + // inputs of a funding transaction withheld for an unclaimed payment stay locked. + let mut funded_addrs: Vec<_> = + (0..4).map(|_| service_node.onchain_payment().new_address().unwrap()).collect(); let client_addr = client_node.onchain_payment().new_address().unwrap(); let payer_addr = payer_node.onchain_payment().new_address().unwrap(); + funded_addrs.extend([client_addr, payer_addr]); + let premine_amount_sat = 10_000_000; premine_and_distribute_funds( &bitcoind.client, &electrsd.client, - vec![service_addr, client_addr, payer_addr], + funded_addrs, Amount::from_sat(premine_amount_sat), ) .await; From 00136a43bd8707d2467e3925d1b3ff672489bbaa Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Thu, 12 Nov 2026 10:49:05 +0100 Subject: [PATCH 5/5] Lock splice inputs during negotiation The inputs the wallet selects for a splice are handed to LDK, which negotiates the transaction spending them with the counterparty. The wallet sees nothing of that negotiation, so a concurrent splice, channel open or on-chain send could select the same inputs. Lock the inputs of a splice contribution when they are selected, the way channel funding inputs are. They are returned when LDK discards the contribution, dropped once the splice transaction confirms, and kept across a restart for as long as LDK tracks the round they were contributed to. Inputs LDK would refuse to announce are left out of the selection, as it refuses them only after they are locked. Co-Authored-By: HAL 9000 --- src/event.rs | 6 +- src/lib.rs | 30 ++++-- src/wallet/mod.rs | 232 ++++++++++++++++++++++++++++++++++++++++++---- 3 files changed, 238 insertions(+), 30 deletions(-) diff --git a/src/event.rs b/src/event.rs index b93cf4d0f2..6ab2ed89e6 100644 --- a/src/event.rs +++ b/src/event.rs @@ -2560,15 +2560,13 @@ where return Err(ReplayEvent()); } - // TODO(#1037): once inputs are locked at coin selection, the `inputs` of a - // contribution are locks this event returns: unlock them here. let (inputs, outputs) = match funding_info { FundingInfo::Tx { transaction } => ( transaction.input.into_iter().map(|txin| txin.previous_output).collect(), transaction.output, ), - FundingInfo::Contribution { inputs: _, outputs } => ( - Vec::new(), + FundingInfo::Contribution { inputs, outputs } => ( + inputs, outputs .into_iter() .map(|script_pubkey| bitcoin::TxOut { diff --git a/src/lib.rs b/src/lib.rs index c199560966..411111a75b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -387,12 +387,19 @@ impl Node { }, ))?; - // See `Wallet::unlock_unreferenced_outpoints`. The locks still needed are those the wallet - // recorded for the funding transaction of a channel that is still around. Without that - // record for a channel awaiting its funding's confirmation, its locks cannot be told apart - // from leftovers, and every lock is kept. An output this releases although a broadcast - // transaction spends it is spent again in the wallet's view by the next sync. - let mut funding_inputs = std::collections::HashSet::new(); + // See `Wallet::unlock_unreferenced_outpoints`. The locks still needed are those of a splice + // round LDK still tracks, which lists the inputs this node contributed to it, and those + // the wallet recorded for the funding transaction of a channel that is still around. + // Without that record for a channel awaiting its funding's confirmation, its locks cannot + // be told apart from leftovers, and every lock is kept. An output this releases although a + // broadcast transaction spends it is spent again in the wallet's view by the next sync. + let mut referenced: std::collections::HashSet = channels + .iter() + .filter_map(|channel| channel.splice_details.as_ref()) + .flat_map(|splice| splice.candidates.iter()) + .filter_map(|candidate| candidate.contribution.as_ref()) + .flat_map(|contribution| contribution.inputs().iter().map(|input| input.outpoint())) + .collect(); let mut has_untracked_funding = false; for channel in channels.iter().filter(|channel| channel.is_outbound) { let Some(funding_txo) = channel.funding_txo else { @@ -402,10 +409,10 @@ impl Node { self.runtime.block_on(self.wallet.locked_funding_inputs(&funding_txo.txid)); has_untracked_funding |= inputs.is_empty() && channel.confirmations.map_or(true, |confirmations| confirmations == 0); - funding_inputs.extend(inputs); + referenced.extend(inputs); } self.runtime.block_on(self.wallet.unlock_unreferenced_outpoints(|outpoint| { - has_untracked_funding || funding_inputs.contains(outpoint) + has_untracked_funding || referenced.contains(outpoint) }))?; // Synchronize Lightning chain state before peers or event processing can act on stale @@ -1978,6 +1985,13 @@ impl Node { reserve_output.clone(), feerate, ); + let built = built.and_then(|contribution| { + // These inputs were not chosen by the wallet's coin selection, which would have locked + // them. LDK reports them once it lets the contribution go. + let outpoints = contribution.inputs().iter().map(|input| input.outpoint()).collect(); + self.runtime.block_on(self.wallet.lock_outpoints(outpoints))?; + Ok(contribution) + }); match built { Ok(contribution) => Ok(contribution), Err(e) => { diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index 2c08f41cda..5cf4e46268 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -44,7 +44,7 @@ use lightning::chain::{BlockLocator, ClaimId, Listen}; use lightning::ln::channel_state::{SpliceCandidateDetails, SpliceCandidateStatus, SpliceDetails}; use lightning::ln::channelmanager::PaymentId; use lightning::ln::inbound_payment::ExpandedKey; -use lightning::ln::msgs::UnsignedGossipMessage; +use lightning::ln::msgs::{TxAddInput, UnsignedGossipMessage}; use lightning::ln::script::ShutdownScript; use lightning::ln::types::ChannelId; use lightning::sign::{ @@ -2183,6 +2183,26 @@ impl Wallet { self.release_inputs_and_outputs(Vec::new(), outputs).await } + /// Locks `outpoints` against selection for another transaction, for inputs chosen without + /// [`Self::select_confirmed_utxos`]. + pub(crate) async fn lock_outpoints(&self, outpoints: Vec) -> Result<(), Error> { + let mut locked_persister = self.persister.lock().await; + let (newly_locked, change_set) = { + let mut locked_wallet = self.inner.lock().expect("lock"); + // A fee bump of a splice carries the inputs of the round it replaces, which hold + // that round's locks already. + let newly_locked: Vec = outpoints + .into_iter() + .filter(|outpoint| !locked_wallet.is_outpoint_locked(*outpoint)) + .collect(); + for outpoint in &newly_locked { + locked_wallet.lock_outpoint(*outpoint); + } + (newly_locked, locked_wallet.take_staged().unwrap_or_default()) + }; + self.persist_new_locks(&mut locked_persister, change_set, newly_locked.into_iter()).await + } + /// Persists a change set that locks `newly_locked`. If that fails the caller does not hand /// out what the locks were taken for, and nothing would ever return them, so they are dropped. async fn persist_new_locks( @@ -2670,8 +2690,22 @@ impl Wallet { ExtendedDescriptor::Wpkh(_) )); + // LDK refuses a contribution with an input it could not announce to the counterparty, + // and does so only once the selection is made and its inputs are locked. + let unannounceable: Vec = locked_wallet + .list_unspent() + .filter(|utxo| { + locked_wallet + .tx_graph() + .get_tx(utxo.outpoint.txid) + .is_some_and(|prevtx| !fits_tx_add_input(&prevtx)) + }) + .map(|utxo| utxo.outpoint) + .collect(); + let mut tx_builder = locked_wallet.build_tx(); tx_builder.only_witness_utxo(); + tx_builder.unspendable(unannounceable); for input in &must_spend { let psbt_input = psbt::Input { @@ -2719,24 +2753,26 @@ impl Wallet { return Err(()); } + // See `unlock_unreferenced_outpoints` for what a lock is for and what returns it. LDK + // reports nothing if the caller fails to build a contribution from the selection, + // which leaves these locks to that sweep. + for utxo in &confirmed_utxos { + locked_wallet.lock_outpoint(utxo.outpoint()); + } + let change_output = unsigned_tx .output .into_iter() .find(|txout| must_pay_to.iter().all(|output| output != txout)); - let change_set = if change_output.is_some() { - Some(locked_wallet.take_staged().unwrap_or_default()) - } else { - None - }; - (CoinSelection { confirmed_utxos, change_output }, change_set) + ( + CoinSelection { confirmed_utxos, change_output }, + locked_wallet.take_staged().unwrap_or_default(), + ) }; - if let Some(change_set) = change_set { - locked_persister.persist_changeset(change_set).await.map_err(|e| { - log_error!(self.logger, "Failed to persist wallet: {}", e); - })?; - } + let locked = coin_selection.confirmed_utxos.iter().map(|utxo| utxo.outpoint()); + self.persist_new_locks(&mut locked_persister, change_set, locked).await.map_err(|_| ())?; Ok(coin_selection) } @@ -3361,12 +3397,8 @@ impl Wallet { let payment_id = entry.id(); let (abandoned, remaining): (Vec, Vec) = { let locked_wallet = self.inner.lock().expect("lock"); - // TODO(#1037): the graph learns a round LDK broadcast from wallet sync alone - // today, so this check only adds what a sync has already seen to `tracked_rounds`. - // It catches every broadcast round by itself, whichever caller — the startup - // sweep or a live event — runs the drop, only once the `InteractiveFunding` - // broadcast arm applies the round to the graph, which #1037 does not do: it - // prepares only `Funding`-typed packages. + // The graph learns a round LDK broadcast from wallet sync alone, so this check only + // adds what a sync has already seen to `tracked_rounds`. entry.candidates().iter().cloned().partition(|candidate| { candidate.awaiting_broadcast && !tracked_rounds.contains(&candidate.txid) @@ -4354,6 +4386,26 @@ impl Wallet { } } +/// Whether a `tx_add_input` message carrying `prevtx` fits a Lightning message. +/// +/// The message announces an input of an interactively built transaction along with the whole +/// transaction the input spends from. Mirrors the check LDK applies to a funding contribution. +fn fits_tx_add_input(prevtx: &Transaction) -> bool { + use lightning::util::ser::Writeable; + const MESSAGE_TYPE_LEN: usize = 2; + const MAX_MESSAGE_LEN: usize = u16::MAX as usize; + let message_without_prevtx = TxAddInput { + channel_id: ChannelId([0; 32]), + serial_id: 0, + prevtx: None, + prevtx_out: 0, + sequence: 0, + shared_input_txid: None, + }; + MESSAGE_TYPE_LEN + message_without_prevtx.serialized_length() + prevtx.serialized_length() + <= MAX_MESSAGE_LEN +} + struct LocalStakeAggregate { amount_msat: Option, fee_paid_msat: Option, @@ -13737,6 +13789,82 @@ mod tests { assert!(none_locked(&wallet, &dropped)); } + /// LDK holds a splice coin selection through a negotiation the wallet sees nothing of, so only + /// the locks keep a second selection from choosing the same inputs. + #[tokio::test] + async fn splice_coin_selection_locks_inputs_until_discarded() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(Arc::clone(&store), false).await; + insert_confirmed_tx(&wallet, wallet_paying_tx(&wallet, 1), 1); + insert_confirmed_tx(&wallet, wallet_paying_tx(&wallet, 2), 2); + + let payment = TxOut { + value: Amount::from_sat(50_000), + script_pubkey: ScriptBuf::new_p2wpkh(&WPubkeyHash::from_byte_array([1; 20])), + }; + let fee_rate = FeeRate::from_sat_per_kwu(250); + let selected_outpoints = |selection: &CoinSelection| -> Vec { + selection.confirmed_utxos.iter().map(|utxo| utxo.outpoint()).collect() + }; + + let first = + Wallet::select_confirmed_utxos(&wallet, Vec::new(), &[payment.clone()], fee_rate) + .await + .unwrap(); + let first_outpoints = selected_outpoints(&first); + assert!(!first_outpoints.is_empty()); + + drop(wallet); + let wallet = new_test_wallet(Arc::clone(&store), true).await; + assert!(all_locked(&wallet, &first_outpoints)); + let second = + Wallet::select_confirmed_utxos(&wallet, Vec::new(), &[payment.clone()], fee_rate) + .await + .unwrap(); + assert!(selected_outpoints(&second) + .iter() + .all(|outpoint| !first_outpoints.contains(outpoint))); + + // What a `DiscardFunding` event returns of the first selection. + wallet + .release_inputs_and_outputs( + first_outpoints.clone(), + first.change_output.into_iter().collect(), + ) + .await + .unwrap(); + + drop(wallet); + let wallet = new_test_wallet(store, true).await; + assert!(none_locked(&wallet, &first_outpoints)); + let third = Wallet::select_confirmed_utxos(&wallet, Vec::new(), &[payment], fee_rate) + .await + .unwrap(); + assert_eq!(selected_outpoints(&third), first_outpoints); + } + + #[tokio::test] + async fn outpoints_locked_by_hand_are_kept_out_of_coin_selection() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(Arc::clone(&store), false).await; + insert_confirmed_tx(&wallet, wallet_paying_tx(&wallet, 1), 1); + let inputs: Vec = wallet + .list_confirmed_funding_inputs() + .unwrap() + .iter() + .map(|input| input.outpoint()) + .collect(); + assert_eq!(inputs.len(), 1); + + wallet.lock_outpoints(inputs.clone()).await.unwrap(); + + drop(wallet); + let wallet = new_test_wallet(store, true).await; + assert!(all_locked(&wallet, &inputs)); + assert!(wallet.list_confirmed_funding_inputs().unwrap().is_empty()); + assert!(test_funding_tx(&wallet, 50_000).await.is_err()); + } + /// A funding transaction that could not be persisted is not handed out, so nobody would ever /// return the locks on its inputs. #[tokio::test] @@ -13781,4 +13909,72 @@ mod tests { assert_eq!(wallet.locked_funding_inputs(&txid).await, inputs); } + + /// LDK checks a contribution's inputs only after coin selection locked them, and nothing + /// would return the locks of a selection it then refuses. + #[tokio::test] + async fn splice_coin_selection_skips_inputs_too_large_to_announce() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(store, false).await; + let mut oversized = wallet_paying_tx(&wallet, 1); + oversized.output.push(TxOut { + value: Amount::ZERO, + script_pubkey: ScriptBuf::from_bytes(vec![0x6a; 70_000]), + }); + assert!(!fits_tx_add_input(&oversized)); + let oversized_outpoint = OutPoint::new(oversized.compute_txid(), 0); + insert_confirmed_tx(&wallet, oversized, 1); + + let payment = TxOut { + value: Amount::from_sat(50_000), + script_pubkey: ScriptBuf::new_p2wpkh(&WPubkeyHash::from_byte_array([1; 20])), + }; + let fee_rate = FeeRate::from_sat_per_kwu(250); + assert!(Wallet::select_confirmed_utxos(&wallet, Vec::new(), &[payment.clone()], fee_rate) + .await + .is_err()); + assert_eq!(locked_outpoint_count(&wallet), 0); + + let coin = wallet_paying_tx(&wallet, 2); + assert!(fits_tx_add_input(&coin)); + let coin_outpoint = OutPoint::new(coin.compute_txid(), 0); + insert_confirmed_tx(&wallet, coin, 2); + let selection = Wallet::select_confirmed_utxos(&wallet, Vec::new(), &[payment], fee_rate) + .await + .unwrap(); + let selected: Vec = + selection.confirmed_utxos.iter().map(|utxo| utxo.outpoint()).collect(); + assert_eq!(selected, vec![coin_outpoint]); + assert!(none_locked(&wallet, &[oversized_outpoint])); + } + + /// See `a_funding_transaction_that_fails_to_persist_holds_no_locks`. + #[tokio::test] + async fn a_splice_coin_selection_that_fails_to_persist_holds_no_locks() { + let fail_store = + FailSwitchStore::failing_only(crate::io::BDK_WALLET_LOCKED_OUTPOINTS_PRIMARY_NAMESPACE); + let store: Arc = Arc::new(DynStoreWrapper(fail_store.clone())); + let wallet = new_test_wallet(Arc::clone(&store), false).await; + insert_confirmed_tx(&wallet, wallet_paying_tx(&wallet, 1), 1); + let payment = TxOut { + value: Amount::from_sat(50_000), + script_pubkey: ScriptBuf::new_p2wpkh(&WPubkeyHash::from_byte_array([1; 20])), + }; + let fee_rate = FeeRate::from_sat_per_kwu(250); + + fail_store.fail_writes.store(true, Ordering::Release); + assert!(Wallet::select_confirmed_utxos(&wallet, Vec::new(), &[payment.clone()], fee_rate) + .await + .is_err()); + assert_eq!(locked_outpoint_count(&wallet), 0); + + fail_store.fail_writes.store(false, Ordering::Release); + wallet.release_outputs(Vec::new()).await.unwrap(); + drop(wallet); + let wallet = new_test_wallet(store, true).await; + assert_eq!(locked_outpoint_count(&wallet), 0); + assert!(Wallet::select_confirmed_utxos(&wallet, Vec::new(), &[payment], fee_rate) + .await + .is_ok()); + } }