diff --git a/CHANGELOG.md b/CHANGELOG.md index 2682bb7137..ba7dee8cd7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -35,6 +35,9 @@ `Config::manually_handle_unknown_bolt11_payments` to `true`. Otherwise matching inbound HTLCs will be failed back instead of emitting `Event::PaymentClaimable`. +- `BalanceDetails::spendable_onchain_balance_sats` no longer includes wallet outputs selected + as inputs of a channel funding transaction or splice that has yet to confirm. They remain + part of `total_onchain_balance_sats` until the transaction is seen by a wallet sync. (#1037) ## Feature and API updates - `Node::splice_out_max` withdraws the maximum available channel balance to an on-chain address. @@ -88,6 +91,10 @@ current chain tip now aborts with a new `BuildError::ChainTipFetchFailed` variant instead of silently pinning the wallet birthday to genesis, which would have forced a full-history rescan once the chain source became reachable again. (#884) +- On-chain sends and fee bumps are now known to the wallet and the payment store as soon as they + are made rather than only after the next wallet sync. Previously, a second send made before + that sync could spend the same inputs, and a second `OnchainPayment::bump_fee_rbf` of the same + payment failed. (#1037) # 0.7.0 - Dec. 3, 2025 This seventh minor release introduces numerous new features, bug fixes, and API improvements. In particular, it adds support for channel Splicing, Async Payments, as well as sourcing chain data from a Bitcoin Core REST backend. diff --git a/src/event.rs b/src/event.rs index 88501ae6ef..6ab2ed89e6 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,36 @@ 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 { + 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 } => ( + inputs, + 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/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/lib.rs b/src/lib.rs index f47c62e538..411111a75b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -387,6 +387,34 @@ impl Node { }, ))?; + // 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 { + 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); + referenced.extend(inputs); + } + self.runtime.block_on(self.wallet.unlock_unreferenced_outpoints(|outpoint| { + has_untracked_funding || referenced.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( @@ -1957,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 77dced3e20..5773db92ff 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; @@ -17,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; @@ -43,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::{ @@ -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); @@ -517,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 @@ -525,6 +565,130 @@ impl Wallet { Ok(()) } + /// The transactions of this node's own on-chain sends that have yet to confirm, each with the + /// transactions it replaced. + async fn pending_send_txids(&self) -> Vec<(Txid, 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, pending.conflicting_txids().to_vec())), + _ => 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. The exception is a conflict the + /// send replaced, given along with it: a chain source that has yet to see a fee bump still + /// reports the transaction the bump replaced. + fn restore_evicted_sends(wallet: &mut bdk_wallet::Wallet, sends: &[(Txid, Vec)]) { + // 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 evicted = sends.iter().filter(|(txid, _)| !in_view.contains(txid)); + let returning = evicted.into_iter().find_map(|(txid, replaced)| { + let tx = graph.get_tx(*txid)?; + let displaces_a_conflict = graph.direct_conflicts(&tx).any(|(_, conflict_txid)| { + in_view.contains(&conflict_txid) && !replaced.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)]); + } + } + + /// 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)>, @@ -534,12 +698,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") @@ -1744,7 +1910,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); @@ -1776,17 +1942,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) } @@ -2011,9 +2184,56 @@ 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 + } + + /// 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( + &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) @@ -2035,27 +2255,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)) @@ -2106,9 +2347,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!( @@ -2189,6 +2431,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; @@ -2245,7 +2490,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. @@ -2307,9 +2552,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 @@ -2334,9 +2578,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); @@ -2369,19 +2612,40 @@ 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(); + // The wallet is written first: a crash in between then leaves a transaction the next sync + // evicts, rather than a payment record for a transaction the wallet does not know. + let recorded = async { + locked_persister.persist_changeset(change_set).await.map_err(|e| { + log_error!(self.logger, "Failed to persist wallet: {}", e); + Error::PersistenceFailed + })?; + self.update_payment_store(events).await.map_err(|e| { + log_error!(self.logger, "Failed to update payment store: {}", 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 { @@ -2432,8 +2696,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 { @@ -2481,24 +2759,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) } @@ -2511,6 +2791,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)); @@ -3121,12 +3403,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) @@ -3996,9 +4274,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, @@ -4040,6 +4319,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 +4344,45 @@ 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; + // 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, conflicting_txids).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); @@ -4078,6 +4392,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, @@ -4324,7 +4658,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; @@ -13129,4 +13467,606 @@ 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, Vec::new()), (parent_txid, Vec::new())], + ); + + 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())); + } + + /// A payment record without its transaction in the wallet could neither be bumped nor would a + /// sync ever resolve it, whereas a transaction without a record is evicted by the next sync. + #[tokio::test] + async fn a_crash_during_a_send_leaves_no_payment_without_its_transaction() { + let snapshot_store = SnapshotStore::new(); + let store: Arc = Arc::new(DynStoreWrapper(snapshot_store.clone())); + let wallet = new_test_wallet(store, false).await; + insert_confirmed_tx(&wallet, wallet_paying_tx(&wallet, 1), 1); + let baseline = snapshot_store.snapshots.lock().unwrap().len(); + + let address = bitcoin::Address::p2wpkh( + &bitcoin::CompressedPublicKey::from_slice(&[2; 33]).unwrap(), + Network::Regtest, + ); + let amount = OnchainSendAmount::ExactRetainingReserve { + amount_sats: 50_000, + cur_anchor_reserve_sats: 0, + }; + let fee_rate = Some(FeeRate::from_sat_per_kwu(500)); + let txid = wallet.send_to_address(&address, amount, fee_rate).await.unwrap(); + + let snapshots = snapshot_store.snapshots.lock().unwrap().clone(); + assert!(snapshots.len() > baseline); + let mut recorded_at_some_point = false; + for snapshot in snapshots.into_iter().skip(baseline) { + let is_recorded = snapshot.keys().any(|(primary_namespace, _, _)| { + primary_namespace == PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE + || primary_namespace == PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE + }); + recorded_at_some_point |= is_recorded; + let store: Arc = + Arc::new(DynStoreWrapper(SnapshotStore::from_contents(snapshot))); + let wallet = new_test_wallet(store, true).await; + assert!( + !is_recorded || is_in_wallet_view(&wallet, txid), + "a crash point holds the payment record of a transaction the wallet lacks" + ); + } + assert!(recorded_at_some_point); + } + + /// A chain source that has yet to see a fee bump still reports the transaction it replaced. + /// That must not take the wallet and the payment back to the replaced transaction. + #[tokio::test] + async fn an_evicted_fee_bump_returns_over_the_transaction_it_replaced() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(store, false).await; + let coin = wallet_paying_tx(&wallet, 1); + let original = tx_spending(&wallet, OutPoint::new(coin.compute_txid(), 0)); + let mut bump = original.clone(); + bump.output[0].value = Amount::from_sat(89_000); + let (original_txid, bump_txid) = (original.compute_txid(), bump.compute_txid()); + insert_confirmed_tx(&wallet, coin, 1); + wallet.inner.lock().unwrap().apply_unconfirmed_txs([(original, 100), (bump, 101)]); + let payment_id = PaymentId(original_txid.to_byte_array()); + let payment = PaymentDetails::new( + payment_id, + PaymentKind::Onchain { + txid: bump_txid, + status: ConfirmationStatus::Unconfirmed, + tx_type: None, + }, + Some(89_000_000), + Some(2_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.clone(), vec![original_txid], Vec::new())) + .await + .unwrap(); + + let mut tx_update = bdk_chain::TxUpdate::default(); + tx_update.seen_ats = [(original_txid, 102)].into(); + tx_update.evicted_ats = [(bump_txid, 102)].into(); + wallet.apply_update(Update { tx_update, ..Default::default() }).await.unwrap(); + + assert!(is_in_wallet_view(&wallet, bump_txid)); + assert!(!is_in_wallet_view(&wallet, original_txid)); + let entry = wallet.payment_stores.pending_payment(&payment_id).await.unwrap().unwrap(); + assert_eq!(entry.details().unwrap().kind, payment.kind); + } + + /// 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)); + } + + /// 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)); + } + + /// 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] + 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); + } + + /// 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()); + } } 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/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/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); diff --git a/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index 691668c56a..1fd74cd5ca 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; @@ -1633,6 +1631,222 @@ 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 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" + ); +} + +/// Logged by a node as it starts, for every wallet output it unlocks as no channel refers to it. +const UNLOCKED_UNREFERENCED_OUTPUT: &str = "no channel refers to"; + +/// Funds `node_a` with two coins and opens a channel from it to `node_b`, returning `node_a`'s id +/// for the channel once it is ready. +async fn open_channel_keeping_a_coin( + bitcoind: &BitcoinD, electrsd: &ElectrsD, node_a: &TestNode, node_b: &TestNode, +) -> UserChannelId { + let addresses_a = (0..2).map(|_| node_a.onchain_payment().new_address().unwrap()).collect(); + premine_and_distribute_funds( + &bitcoind.client, + &electrsd.client, + addresses_a, + Amount::from_sat(1_000_000), + ) + .await; + node_a.sync_wallets().unwrap(); + + open_channel(node_a, node_b, 500_000, false, electrsd).await; + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + let user_channel_id_a = expect_channel_ready_event!(node_a, node_b.node_id()); + expect_channel_ready_event!(node_b, node_a.node_id()); + user_channel_id_a +} + +/// The inputs of a splice LDK still tracks when the node starts stay locked. Pinned to Esplora so +/// that node A's wallet syncs only on demand: it has not seen the splice transaction, and the +/// locks are all that keeps its inputs from being spent again. +#[cfg(feature = "chain-esplora")] +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn inputs_of_a_pending_splice_stay_locked_across_restart() { + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + let chain_source = TestChainSource::Esplora(&electrsd); + let logs_a = Arc::new(CollectingLogWriter::new()); + let mut config_a = random_config(); + config_a.store_type = TestStoreType::Sqlite; + config_a.log_writer = TestLogWriter::Custom(logs_a.clone()); + let node_a = setup_node(&chain_source, config_a.clone()); + let node_b = setup_node(&chain_source, random_config()); + let user_channel_id_a = + open_channel_keeping_a_coin(&bitcoind, &electrsd, &node_a, &node_b).await; + + let spendable_before_sats = node_a.list_balances().spendable_onchain_balance_sats; + node_a.splice_in(&user_channel_id_a, node_b.node_id(), 300_000, None).unwrap(); + let splice_txo = expect_splice_negotiated_event!(node_a, node_b.node_id()); + wait_for_tx(&electrsd.client, splice_txo.txid).await; + let spendable_locked_sats = node_a.list_balances().spendable_onchain_balance_sats; + assert!(spendable_locked_sats <= spendable_before_sats - 300_000); + + node_a.stop().unwrap(); + drop(node_a); + let node_a = setup_node(&chain_source, config_a); + + assert_eq!(logs_a.count(UNLOCKED_UNREFERENCED_OUTPUT), 0); + assert_eq!(node_a.list_balances().spendable_onchain_balance_sats, spendable_locked_sats); + + node_a.stop().unwrap(); + node_b.stop().unwrap(); +} + +/// The inputs of a splice LDK no longer tracks when the node starts are unlocked. The splice is +/// initiated while disconnected, so LDK holds its contribution without negotiating or persisting +/// it, and drops it across the restart. +#[cfg(feature = "chain-esplora")] +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn inputs_of_a_splice_lost_across_restart_are_unlocked() { + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + let chain_source = TestChainSource::Esplora(&electrsd); + let logs_a = Arc::new(CollectingLogWriter::new()); + let mut config_a = random_config(); + config_a.store_type = TestStoreType::Sqlite; + config_a.log_writer = TestLogWriter::Custom(logs_a.clone()); + let node_a = setup_node(&chain_source, config_a.clone()); + let node_b = setup_node(&chain_source, random_config()); + let user_channel_id_a = + open_channel_keeping_a_coin(&bitcoind, &electrsd, &node_a, &node_b).await; + + let spendable_before_sats = node_a.list_balances().spendable_onchain_balance_sats; + node_a.disconnect(node_b.node_id()).unwrap(); + node_a.splice_in(&user_channel_id_a, node_b.node_id(), 300_000, None).unwrap(); + let spendable_locked_sats = node_a.list_balances().spendable_onchain_balance_sats; + assert!(spendable_locked_sats <= spendable_before_sats - 300_000); + + node_a.stop().unwrap(); + drop(node_a); + let node_a = setup_node(&chain_source, config_a); + + // LDK also reports the contribution it dropped, which returns its inputs as well. The log + // line tells that they were returned before that report was handled. + assert!(logs_a.count(UNLOCKED_UNREFERENCED_OUTPUT) > 0); + assert_eq!(node_a.list_balances().spendable_onchain_balance_sats, spendable_before_sats); + + node_a.stop().unwrap(); + node_b.stop().unwrap(); +} + +/// A node that has an outbound channel awaiting confirmation without a record of the inputs its +/// funding transaction spends cannot tell that channel's locks from leftovers, and keeps every +/// lock when it starts. Pinned to Esplora so that node A's wallet syncs only on demand. +#[cfg(feature = "chain-esplora")] +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn locks_are_kept_across_restart_while_a_pending_funding_has_no_recorded_inputs() { + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + let chain_source = TestChainSource::Esplora(&electrsd); + let logs_a = Arc::new(CollectingLogWriter::new()); + let mut config_a = random_config(); + config_a.log_writer = TestLogWriter::Custom(logs_a.clone()); + let store_a = TestSyncStore::new(config_a.node_config.storage_dir_path.clone().into()); + let node_a = setup_node_with_store(&chain_source, config_a.clone(), store_a.clone()); + let node_b = setup_node(&chain_source, random_config()); + + let address_a = node_a.onchain_payment().new_address().unwrap(); + premine_and_distribute_funds( + &bitcoind.client, + &electrsd.client, + vec![address_a], + Amount::from_sat(1_000_000), + ) + .await; + node_a.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; + // The wallet's only coin is locked, and the funding transaction's change unknown to it. + assert_eq!(node_a.list_balances().spendable_onchain_balance_sats, 0); + + node_a.stop().unwrap(); + drop(node_a); + let facts_keys = KVStore::list(&store_a, "channel_tx_facts", "").await.unwrap(); + assert!(!facts_keys.is_empty(), "opening the channel recorded its funding"); + for key in facts_keys { + KVStore::remove(&store_a, "channel_tx_facts", "", &key, false).await.unwrap(); + } + let node_a = setup_node_with_store(&chain_source, config_a, store_a); + + assert_eq!(logs_a.count(UNLOCKED_UNREFERENCED_OUTPUT), 0); + assert_eq!(node_a.list_balances().spendable_onchain_balance_sats, 0); + + node_a.stop().unwrap(); + node_b.stop().unwrap(); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn split_underpaid_bolt11_payment() { let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); @@ -4706,7 +4920,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 @@ -6917,16 +7131,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; @@ -8012,6 +8232,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 +8295,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 =