diff --git a/src/chain/bitcoind.rs b/src/chain/bitcoind.rs index f7589671b8..c5aa1a82ad 100644 --- a/src/chain/bitcoind.rs +++ b/src/chain/bitcoind.rs @@ -1274,18 +1274,18 @@ impl BitcoindClient { &self, bdk_unconfirmed_txids: Vec, ) -> Result, BitcoindClientError> { match self { - BitcoindClient::Rpc { latest_mempool_timestamp, mempool_entries_cache, .. } => { + BitcoindClient::Rpc { mempool_entries_cache, latest_mempool_timestamp, .. } => { Self::get_evicted_mempool_txids_and_timestamp_inner( - latest_mempool_timestamp, mempool_entries_cache, + latest_mempool_timestamp, bdk_unconfirmed_txids, ) .await }, - BitcoindClient::Rest { latest_mempool_timestamp, mempool_entries_cache, .. } => { + BitcoindClient::Rest { mempool_entries_cache, latest_mempool_timestamp, .. } => { Self::get_evicted_mempool_txids_and_timestamp_inner( - latest_mempool_timestamp, mempool_entries_cache, + latest_mempool_timestamp, bdk_unconfirmed_txids, ) .await @@ -1294,16 +1294,17 @@ impl BitcoindClient { } async fn get_evicted_mempool_txids_and_timestamp_inner( - latest_mempool_timestamp: &AtomicU64, mempool_entries_cache: &tokio::sync::Mutex>, - bdk_unconfirmed_txids: Vec, + latest_mempool_timestamp: &AtomicU64, bdk_unconfirmed_txids: Vec, ) -> Result, BitcoindClientError> { - let latest_mempool_timestamp = latest_mempool_timestamp.load(Ordering::Relaxed); let mempool_entries_cache = mempool_entries_cache.lock().await; + let observed_at = + SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs(); + let evicted_at = observed_at.max(latest_mempool_timestamp.load(Ordering::Relaxed)); let evicted_txids = bdk_unconfirmed_txids .into_iter() .filter(|txid| !mempool_entries_cache.contains_key(txid)) - .map(|txid| (txid, latest_mempool_timestamp)) + .map(|txid| (txid, evicted_at)) .collect(); Ok(evicted_txids) } @@ -1588,6 +1589,10 @@ impl std::error::Error for BitcoindClientError {} #[cfg(test)] mod tests { + use std::collections::HashMap; + use std::sync::atomic::{AtomicU64, Ordering}; + use std::time::{SystemTime, UNIX_EPOCH}; + use bitcoin::hashes::Hash; use bitcoin::{FeeRate, OutPoint, ScriptBuf, Transaction, TxIn, TxOut, Txid, Witness}; use lightning_block_sync::http::JsonResponse; @@ -1597,10 +1602,59 @@ mod tests { use serde_json::json; use crate::chain::bitcoind::{ - FeeResponse, GetMempoolEntryResponse, GetRawMempoolResponse, GetRawTransactionResponse, - MempoolMinFeeResponse, + BitcoindClient, FeeResponse, GetMempoolEntryResponse, GetRawMempoolResponse, + GetRawTransactionResponse, MempoolMinFeeResponse, }; + #[tokio::test] + async fn eviction_uses_absence_observation_time() { + let txid = Txid::all_zeros(); + let mempool_entries = tokio::sync::Mutex::new(HashMap::new()); + let latest_mempool_timestamp = AtomicU64::new(0); + let before = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs(); + + let evicted = BitcoindClient::get_evicted_mempool_txids_and_timestamp_inner( + &mempool_entries, + &latest_mempool_timestamp, + vec![txid], + ) + .await + .unwrap(); + + assert_eq!(evicted.len(), 1); + assert_eq!(evicted[0].0, txid); + assert!(evicted[0].1 >= before, "eviction timestamp must record when absence was observed"); + } + + #[tokio::test] + async fn eviction_preserves_newer_mempool_time() { + let txid = Txid::from_byte_array([1; 32]); + let client = BitcoindClient::new_rpc( + "127.0.0.1".to_string(), + 18443, + "user".to_string(), + "password".to_string(), + ); + let observed_at = + SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs(); + let newer_mempool_time = observed_at.saturating_add(60); + match &client { + BitcoindClient::Rpc { latest_mempool_timestamp, .. } => { + latest_mempool_timestamp.store(newer_mempool_time, Ordering::Relaxed); + }, + BitcoindClient::Rest { .. } => unreachable!(), + } + + let evicted = client.get_evicted_mempool_txids_and_timestamp(vec![txid]).await.unwrap(); + + assert_eq!(evicted.len(), 1); + assert_eq!(evicted[0].0, txid); + assert_eq!( + evicted[0].1, newer_mempool_time, + "eviction timestamp must not precede Bitcoin Core's mempool time" + ); + } + prop_compose! { fn arbitrary_witness()( witness_elements in vec(vec(any::(), 0..100), 0..20) diff --git a/src/chain/mod.rs b/src/chain/mod.rs index 0f96c409f8..7242116740 100644 --- a/src/chain/mod.rs +++ b/src/chain/mod.rs @@ -504,15 +504,15 @@ impl ChainSource { return; } Some(next_package) = receiver.recv() => { - // Classify funding broadcasts into payment records before sending. If - // classification fails we skip the broadcast, since broadcasting a tx we - // failed to record would leave it on-chain without a payment. + // Prepare wallet transactions and classify funding broadcasts before sending. + // If either fails, broadcasting could race another spend or leave an on-chain + // transaction without a payment record. let package = match self.tx_broadcaster.classify_package(next_package).await { Ok(package) => package, Err(e) => { log_error!( tx_bcast_logger, - "Skipping broadcast: failed to persist payment records: {:?}", + "Skipping broadcast: failed to prepare transaction: {:?}", e, ); continue; diff --git a/src/event.rs b/src/event.rs index 8e6a98dc91..8e069fa174 100644 --- a/src/event.rs +++ b/src/event.rs @@ -715,6 +715,7 @@ where .await; match funding_transaction { Ok(final_tx) => { + let final_tx_for_cancel = final_tx.clone(); let needs_manual_broadcast = self .liquidity_source .lsps2_service() @@ -745,27 +746,39 @@ where match result { Ok(()) => {}, - Err(APIError::APIMisuseError { err }) => { - log_error!( - self.logger, - "Encountered APIMisuseError, this should never happen: {}", - err - ); - debug_assert!(false, "APIMisuseError: {}", err); - }, - Err(APIError::ChannelUnavailable { err }) => { - log_error!( - self.logger, - "Failed to process funding transaction as channel went away before we could fund it: {}", - err - ) - }, Err(err) => { - log_error!( - self.logger, - "Failed to process funding transaction: {:?}", - err - ) + if let Err(e) = self.wallet.cancel_tx(final_tx_for_cancel).await { + log_error!( + self.logger, + "Failed to release funding inputs: {}", + e + ); + return Err(ReplayEvent()); + } + match err { + APIError::APIMisuseError { err } => { + log_error!( + self.logger, + "Encountered APIMisuseError, this should never happen: {}", + err + ); + debug_assert!(false, "APIMisuseError: {}", err); + }, + APIError::ChannelUnavailable { err } => { + log_error!( + self.logger, + "Failed to process funding transaction as channel went away before we could fund it: {}", + err + ) + }, + err => { + log_error!( + self.logger, + "Failed to process funding transaction: {:?}", + err + ) + }, + } }, } }, @@ -1737,27 +1750,33 @@ where } }, LdkEvent::DiscardFunding { channel_id, funding_info } => { - if let FundingInfo::Contribution { inputs: _, outputs } = funding_info { + let tx = match funding_info { + FundingInfo::Tx { transaction } => Some(transaction), + FundingInfo::Contribution { inputs: _, outputs } => { + Some(bitcoin::Transaction { + version: bitcoin::transaction::Version::TWO, + lock_time: bitcoin::absolute::LockTime::ZERO, + input: vec![], + output: outputs + .into_iter() + .map(|script_pubkey| bitcoin::TxOut { + value: bitcoin::Amount::ZERO, + script_pubkey, + }) + .collect(), + }) + }, + FundingInfo::OutPoint { .. } => None, + }; + + if let Some(tx) = tx { log_info!( self.logger, - "Reclaiming unused addresses from channel {} funding", + "Reclaiming unused wallet state from channel {} funding", channel_id, ); - - let tx = bitcoin::Transaction { - version: bitcoin::transaction::Version::TWO, - lock_time: bitcoin::absolute::LockTime::ZERO, - input: vec![], - output: outputs - .into_iter() - .map(|script_pubkey| bitcoin::TxOut { - value: bitcoin::Amount::ZERO, - script_pubkey, - }) - .collect(), - }; if let Err(e) = self.wallet.cancel_tx(tx).await { - log_error!(self.logger, "Failed reclaiming unused addresses: {}", e); + 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 a01aa59a83..b11a125f21 100644 --- a/src/io/mod.rs +++ b/src/io/mod.rs @@ -73,6 +73,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 4657688f51..467da42888 100644 --- a/src/io/utils.rs +++ b/src/io/utils.rs @@ -18,6 +18,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; @@ -581,6 +582,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, @@ -623,6 +633,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/tx_broadcaster.rs b/src/tx_broadcaster.rs index 782112dadb..36041d1eeb 100644 --- a/src/tx_broadcaster.rs +++ b/src/tx_broadcaster.rs @@ -133,15 +133,30 @@ where self.queue_receiver.lock().await } - /// Classifies a queued package into payment records and returns the package ready for the - /// chain client. Returns `Err` if any classification fails; callers must not broadcast the - /// package in that case, since a crash would leave the transaction on-chain without a record. + /// Prepares a queued package in the wallet, classifies it into payment records, and returns the + /// package ready for the chain client. Returns `Err` if preparation or classification fails; + /// callers must not broadcast the package in that case. pub(crate) async fn classify_package( &self, package: BroadcastPackage, ) -> Result { let wallet_opt = self.wallet.lock().expect("lock").as_ref().and_then(Weak::upgrade); if let Some(wallet) = wallet_opt { for (tx, tx_type) in package.transactions() { + let should_broadcast = match tx_type { + Some(LdkTransactionType::Funding { .. }) => { + wallet.prepare_funding_broadcast(tx).await? + }, + None => wallet.prepare_unclassified_broadcast(tx).await?, + _ => true, + }; + if !should_broadcast { + log_error!( + self.logger, + "Skipping broadcast of {} because an input is no longer available", + tx.compute_txid(), + ); + return Err(Error::WalletOperationFailed); + } if let Some(tx_type) = tx_type { wallet.classify_broadcast(tx, tx_type).await?; } diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index f8d9d521eb..2b0cb44834 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}; +use std::time::{SystemTime, UNIX_EPOCH}; use bdk_chain::spk_client::{FullScanRequest, SyncRequest}; use bdk_wallet::descriptor::ExtendedDescriptor; @@ -123,6 +124,27 @@ impl Wallet { self.inner.lock().expect("lock").start_full_scan().build() } + fn next_seen_at(wallet: &PersistedWallet, tx: &Transaction) -> u64 { + let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs(); + let graph = wallet.tx_graph(); + let txid = tx.compute_txid(); + let latest_timestamp = graph + .direct_conflicts(tx) + .map(|(_, conflict_txid)| conflict_txid) + .chain(std::iter::once(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() } @@ -332,8 +354,9 @@ impl Wallet { .iter() .filter_map(|txid| { locked_wallet + .tx_graph() .get_tx(*txid) - .map(|tx| tx.tx_node.tx.as_ref().clone()) + .map(|tx| tx.as_ref().clone()) }) .collect() }; @@ -461,7 +484,7 @@ impl Wallet { ) -> Result { let fee_rate = 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); @@ -489,18 +512,21 @@ 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 + })?; + for txin in &tx.input { + locked_wallet.lock_outpoint(txin.previous_output); + } + + (tx, 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 - })?; - Ok(tx) } @@ -550,6 +576,9 @@ impl Wallet { fn cancel_tx_inner( locked_wallet: &mut PersistedWallet, tx: Transaction, ) { + for txin in tx.input { + locked_wallet.unlock_outpoint(txin.previous_output); + } for txout in tx.output { if let Some((keychain, index)) = locked_wallet.derivation_of_spk(txout.script_pubkey) { // This mirrors the removed BDK helper: it only frees superficial usage marks. @@ -561,32 +590,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 (balance, locked_amount_sats) = { + let locked_wallet = self.inner.lock().expect("lock"); + let locked_amount_sats = Self::locked_unspent_value(&locked_wallet); + (locked_wallet.balance(), locked_amount_sats) + }; // 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 balance.confirmed.to_sat() > locked_amount_sats { 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) + self.get_balances_inner(balance, total_anchor_channels_reserve_sats, locked_amount_sats) } fn get_balances_inner( - &self, balance: Balance, total_anchor_channels_reserve_sats: u64, + &self, balance: Balance, total_anchor_channels_reserve_sats: u64, locked_amount_sats: u64, ) -> Result<(u64, u64), Error> { 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)) } + fn locked_unspent_value(wallet: &PersistedWallet) -> u64 { + wallet + .list_unspent() + .filter(|output| wallet.is_outpoint_locked(output.outpoint)) + .map(|output| output.txout.value.to_sat()) + .sum() + } + pub(crate) fn get_spendable_amount_sats( &self, total_anchor_channels_reserve_sats: u64, ) -> Result { @@ -645,8 +690,11 @@ impl Wallet { shared_input: Option<&Input>, ) -> 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 locked_amount_sats = Self::locked_unspent_value(locked_wallet); + let spendable_amount_sats = self + .get_balances_inner(balance, cur_anchor_reserve_sats, locked_amount_sats) + .map(|(_, s)| s) + .unwrap_or(0); if spendable_amount_sats == 0 { log_error!( @@ -765,7 +813,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, 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. @@ -831,8 +879,9 @@ impl Wallet { cur_anchor_reserve_sats, } => { let balance = locked_wallet.balance(); + let locked_amount_sats = Self::locked_unspent_value(&locked_wallet); let spendable_amount_sats = self - .get_balances_inner(balance, cur_anchor_reserve_sats) + .get_balances_inner(balance, cur_anchor_reserve_sats, locked_amount_sats) .map(|(_, s)| s) .unwrap_or(0); let tx_fee_sats = locked_wallet @@ -858,8 +907,9 @@ impl Wallet { }, OnchainSendAmount::AllRetainingReserve { cur_anchor_reserve_sats } => { let balance = locked_wallet.balance(); + let locked_amount_sats = Self::locked_unspent_value(&locked_wallet); let spendable_amount_sats = self - .get_balances_inner(balance, cur_anchor_reserve_sats) + .get_balances_inner(balance, cur_anchor_reserve_sats, locked_amount_sats) .map(|(_, s)| s) .unwrap_or(0); let (sent, received) = locked_wallet.sent_and_received(&psbt.unsigned_tx); @@ -888,18 +938,24 @@ 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 + })?; + let seen_at = Self::next_seen_at(&locked_wallet, &tx); + let events = locked_wallet.apply_unconfirmed_txs_events([(tx.clone(), seen_at)]); + let change_set = locked_wallet.take_staged().unwrap_or_default(); + (tx, events, change_set) }; + self.update_payment_store(events).await.map_err(|e| { + log_error!(self.logger, "Failed to record on-chain payment: {}", e); + Error::PersistenceFailed + })?; 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(); self.broadcaster.broadcast_unclassified_transaction(tx); @@ -1030,8 +1086,10 @@ impl Wallet { .filter(|t| t.chain_position.is_confirmed()) .map(|t| t.tx_node.txid) .collect(); - let unspent_confirmed_utxos = - locked_wallet.list_unspent().filter(|u| confirmed_txs.contains(&u.outpoint.txid)); + let unspent_confirmed_utxos = locked_wallet.list_unspent().filter(|u| { + confirmed_txs.contains(&u.outpoint.txid) + && !locked_wallet.is_outpoint_locked(u.outpoint) + }); for u in unspent_confirmed_utxos { let script_pubkey = u.txout.script_pubkey; @@ -1200,6 +1258,103 @@ impl Wallet { Ok(tx) } + /// Makes a channel-funding transaction durable and releases its temporary input locks before + /// broadcasting it. + pub(crate) async fn prepare_funding_broadcast(&self, tx: &Transaction) -> Result { + self.prepare_wallet_broadcast(tx, true).await + } + + /// Restores a wallet transaction evicted from BDK's canonical graph before rebroadcasting it. + pub(crate) async fn prepare_unclassified_broadcast( + &self, tx: &Transaction, + ) -> Result { + self.prepare_wallet_broadcast(tx, false).await + } + + async fn prepare_wallet_broadcast( + &self, tx: &Transaction, is_funding: bool, + ) -> Result { + let mut locked_persister = self.persister.lock().await; + let (should_broadcast, change_set) = { + let mut locked_wallet = self.inner.lock().expect("lock"); + let txid = tx.compute_txid(); + let is_canonical = locked_wallet.get_tx(txid).is_some(); + let was_known = locked_wallet.tx_graph().get_tx(txid).is_some(); + + if !is_canonical { + let (has_canonical_conflict, has_confirmed_conflict) = locked_wallet + .tx_graph() + .direct_conflicts(tx) + .map(|(_, conflict_txid)| conflict_txid) + .fold((false, false), |(has_conflict, has_confirmed), conflict_txid| { + match locked_wallet.get_tx(conflict_txid) { + Some(conflict) => { + (true, has_confirmed || conflict.chain_position.is_confirmed()) + }, + None => (has_conflict, has_confirmed), + } + }); + let is_current_outbound_tx = has_canonical_conflict + && !is_funding && !self + .pending_payment_store + .list_filter(|payment| { + payment.details.direction == PaymentDirection::Outbound + && payment.details.status == PaymentStatus::Pending + && matches!( + payment.details.kind, + PaymentKind::Onchain { + txid: current_txid, + status: ConfirmationStatus::Unconfirmed, + .. + } if current_txid == txid + ) + }) + .is_empty(); + let unavailable_conflict = + has_canonical_conflict && (!is_current_outbound_tx || has_confirmed_conflict); + let has_locked_input = tx + .input + .iter() + .any(|txin| locked_wallet.is_outpoint_locked(txin.previous_output)); + let owns_initial_funding_locks = is_funding && !was_known; + + if unavailable_conflict || (has_locked_input && !owns_initial_funding_locks) { + if owns_initial_funding_locks { + for txin in &tx.input { + locked_wallet.unlock_outpoint(txin.previous_output); + } + } + let change_set = locked_wallet.take_staged().unwrap_or_default(); + (false, change_set) + } else { + let seen_at = Self::next_seen_at(&locked_wallet, tx); + locked_wallet.apply_unconfirmed_txs([(tx.clone(), seen_at)]); + if is_funding { + for txin in &tx.input { + locked_wallet.unlock_outpoint(txin.previous_output); + } + } + let change_set = locked_wallet.take_staged().unwrap_or_default(); + (true, change_set) + } + } else { + if is_funding { + for txin in &tx.input { + locked_wallet.unlock_outpoint(txin.previous_output); + } + } + let change_set = locked_wallet.take_staged().unwrap_or_default(); + (true, change_set) + } + }; + + locked_persister.persist_changeset(change_set).await.map_err(|e| { + log_error!(self.logger, "Failed to persist transaction before broadcast: {}", e); + Error::PersistenceFailed + })?; + Ok(should_broadcast) + } + /// Classifies an on-chain broadcast handed to the broadcaster by LDK, recording a payment for it /// before it is sent when it affects this node's wallet. pub(crate) async fn classify_broadcast( @@ -1730,8 +1885,11 @@ 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 locked_amount_sats = Self::locked_unspent_value(&locked_wallet); + let spendable_amount_sats = self + .get_balances_inner(balance, cur_anchor_reserve_sats, locked_amount_sats) + .map(|(_, s)| s) + .unwrap_or(0); if spendable_amount_sats < additional_fee_sats { log_error!( self.logger, @@ -1768,6 +1926,8 @@ impl Wallet { })?; let new_txid = fee_bumped_tx.compute_txid(); + let seen_at = Self::next_seen_at(&locked_wallet, &fee_bumped_tx); + locked_wallet.apply_unconfirmed_txs([(fee_bumped_tx.clone(), seen_at)]); let new_payment = self.create_payment_from_tx( &locked_wallet, @@ -2154,3 +2314,155 @@ fn ldk_to_bdk_satisfaction_weight(ldk_satisfaction_weight: u64) -> Weight { .saturating_sub(EMPTY_SCRIPT_SIG_WEIGHT + EMPTY_WITNESS_COUNT_WEIGHT), ) } + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + use std::sync::Arc; + + use bdk_wallet::{KeychainKind, Wallet as BdkWallet}; + use bitcoin::hashes::Hash; + use bitcoin::transaction::Version; + use bitcoin::{Amount, Network, OutPoint, ScriptBuf, Sequence, Transaction, TxIn, TxOut, Txid}; + use lightning::ln::channelmanager::PaymentId; + + use super::{KVStoreWalletPersister, Wallet}; + use crate::chain::ChainSource; + use crate::config::{Config, EsploraSyncConfig}; + use crate::fee_estimator::OnchainFeeEstimator; + use crate::io::test_utils::InMemoryStore; + use crate::logger::Logger; + use crate::payment::{ + ConfirmationStatus, PaymentDetails, PaymentDirection, PaymentKind, PaymentStatus, + PendingPaymentDetails, + }; + use crate::runtime::Runtime; + use crate::tx_broadcaster::TransactionBroadcaster; + use crate::types::{DynStore, DynStoreWrapper, PaymentStore, PendingPaymentStore}; + use crate::{NodeMetrics, PersistedNodeMetrics}; + + const EXTERNAL_DESCRIPTOR: &str = "wpkh(tprv8ZgxMBicQKsPdy6LMhUtFHAgpocR8GC6QmwMSFpZs7h6Eziw3SpThFfczTDh5rW2krkqffa11UpX3XkeTTB2FvzZKWXqPY54Y6Rq4AQ5R8L/84'/1'/0'/0/*)"; + const INTERNAL_DESCRIPTOR: &str = "wpkh(tprv8ZgxMBicQKsPdy6LMhUtFHAgpocR8GC6QmwMSFpZs7h6Eziw3SpThFfczTDh5rW2krkqffa11UpX3XkeTTB2FvzZKWXqPY54Y6Rq4AQ5R8L/84'/1'/0'/1/*)"; + + #[tokio::test] + async fn restores_current_rbf_transaction_before_broadcast() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(Logger::new_log_facade()); + let broadcaster = Arc::new(TransactionBroadcaster::new(Arc::clone(&logger))); + let fee_estimator = Arc::new(OnchainFeeEstimator::new()); + let mut config = Config::default(); + config.network = Network::Regtest; + let config = Arc::new(config); + let node_metrics = Arc::new(PersistedNodeMetrics::new(NodeMetrics::default())); + let (chain_source, _) = ChainSource::new_esplora( + "http://127.0.0.1:1".to_string(), + HashMap::new(), + EsploraSyncConfig::default(), + Arc::clone(&fee_estimator), + Arc::clone(&broadcaster), + Arc::clone(&store), + Arc::clone(&config), + Arc::clone(&logger), + node_metrics, + ) + .expect("valid Esplora URL"); + + let payment_store = Arc::new(PaymentStore::new( + Vec::new(), + "payments".to_string(), + String::new(), + Arc::clone(&store), + Arc::clone(&logger), + )); + let pending_payment_store = Arc::new(PendingPaymentStore::new( + Vec::new(), + "pending_payments".to_string(), + String::new(), + Arc::clone(&store), + Arc::clone(&logger), + )); + + let mut persister = KVStoreWalletPersister::new(Arc::clone(&store), Arc::clone(&logger)); + let mut bdk_wallet = BdkWallet::create(EXTERNAL_DESCRIPTOR, INTERNAL_DESCRIPTOR) + .network(Network::Regtest) + .create_wallet_async(&mut persister) + .await + .unwrap(); + let wallet_script = + bdk_wallet.next_unused_address(KeychainKind::External).address.script_pubkey(); + let funding_tx = Transaction { + version: Version::TWO, + lock_time: bitcoin::absolute::LockTime::ZERO, + input: vec![TxIn { + previous_output: OutPoint::new(Txid::from_byte_array([42; 32]), 0), + ..TxIn::default() + }], + output: vec![TxOut { value: Amount::from_sat(100_000), script_pubkey: wallet_script }], + }; + let funding_outpoint = OutPoint::new(funding_tx.compute_txid(), 0); + let spend = |value_sat| Transaction { + version: Version::TWO, + lock_time: bitcoin::absolute::LockTime::ZERO, + input: vec![TxIn { + previous_output: funding_outpoint, + sequence: Sequence::ENABLE_RBF_NO_LOCKTIME, + ..TxIn::default() + }], + output: vec![TxOut { + value: Amount::from_sat(value_sat), + script_pubkey: ScriptBuf::new(), + }], + }; + let original_tx = spend(90_000); + let replacement_tx = spend(89_000); + let original_txid = original_tx.compute_txid(); + let replacement_txid = replacement_tx.compute_txid(); + + bdk_wallet.apply_unconfirmed_txs([ + (funding_tx, 1), + (original_tx, 2), + (replacement_tx.clone(), 3), + ]); + bdk_wallet.apply_evicted_txs([(replacement_txid, 4)]); + assert!(bdk_wallet.get_tx(original_txid).is_some()); + assert!(bdk_wallet.get_tx(replacement_txid).is_none()); + + let payment_id = PaymentId(original_txid.to_byte_array()); + let payment = PaymentDetails::new( + payment_id, + PaymentKind::Onchain { + txid: replacement_txid, + status: ConfirmationStatus::Unconfirmed, + tx_type: None, + }, + Some(89_000_000), + Some(11_000_000), + PaymentDirection::Outbound, + PaymentStatus::Pending, + ); + payment_store.insert(payment.clone()).await.unwrap(); + pending_payment_store + .insert(PendingPaymentDetails::new(payment, vec![original_txid], Vec::new())) + .await + .unwrap(); + + let wallet = Wallet::new( + bdk_wallet, + persister, + broadcaster, + fee_estimator, + Arc::new(chain_source), + payment_store, + Arc::new(Runtime::new(Arc::clone(&logger)).unwrap()), + config, + logger, + pending_payment_store, + ); + + assert!( + wallet.prepare_unclassified_broadcast(&replacement_tx).await.unwrap(), + "the current RBF replacement must be restored before broadcast" + ); + assert!(wallet.inner.lock().unwrap().get_tx(replacement_txid).is_some()); + } +} diff --git a/src/wallet/persist.rs b/src/wallet/persist.rs index 9d33a09f93..27f2f4b18a 100644 --- a/src/wallet/persist.rs +++ b/src/wallet/persist.rs @@ -14,8 +14,8 @@ use bdk_wallet::{AsyncWalletPersister, ChangeSet}; 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::logger::{log_error, LdkLogger, Logger}; use crate::types::DynStore; @@ -160,6 +160,19 @@ impl KVStoreWalletPersister { .await?; } + // Persist transaction graph changes before releasing locks 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()); + 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( @@ -223,7 +236,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}; @@ -341,4 +355,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), Some(&false)); + } } 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 df477588f1..236d8aba26 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -453,11 +453,6 @@ 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. - 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; open_channel(&nodes[3], &nodes[4], 1_000_000, true, &electrsd).await; @@ -509,6 +504,49 @@ async fn multi_hop_sending() { expect_payment_successful_event!(nodes[0], payment_id, Some(fee_paid_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(); + let esplora_url = format!("http://{}", electrsd.esplora_url.as_ref().unwrap()); + let config = random_config(); + let mut sync_config = EsploraSyncConfig::default(); + sync_config.background_sync_config = None; + setup_builder!(builder, config.node_config); + builder.set_chain_source_esplora(esplora_url, Some(sync_config)); + let node = builder.build(config.node_entropy.into()).unwrap(); + node.start().unwrap(); + + 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(); + + for _ in 0..50 { + let mempool = bitcoind.client.get_raw_mempool().unwrap().into_model().unwrap(); + if mempool.0.contains(&first_txid) && mempool.0.contains(&second_txid) { + return; + } + tokio::time::sleep(Duration::from_millis(100)).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(); @@ -3860,17 +3898,17 @@ 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("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(); - let original_fee = original_payment.fee_paid_msat.unwrap(); // Non-existent payment id let fake_txid =