diff --git a/src/chain/bitcoind.rs b/src/chain/bitcoind.rs index f7589671b..c5aa1a82a 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 0f96c409f..724211674 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 be54969c7..7a542ffb7 100644 --- a/src/event.rs +++ b/src/event.rs @@ -538,6 +538,28 @@ impl Future for EventFuture { } } +fn discarded_funding_transaction(funding_info: FundingInfo) -> Option { + 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: inputs + .into_iter() + .map(|previous_output| bitcoin::TxIn { + previous_output, + ..bitcoin::TxIn::default() + }) + .collect(), + output: outputs + .into_iter() + .map(|script_pubkey| bitcoin::TxOut { value: bitcoin::Amount::ZERO, script_pubkey }) + .collect(), + }), + FundingInfo::OutPoint { .. } => None, + } +} + pub(crate) struct EventHandler where L::Target: LdkLogger, @@ -740,6 +762,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() @@ -770,27 +793,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 + ) + }, + } }, } }, @@ -1976,27 +2011,14 @@ where } }, LdkEvent::DiscardFunding { channel_id, funding_info } => { - if let FundingInfo::Contribution { inputs: _, outputs } = funding_info { + if let Some(tx) = discarded_funding_transaction(funding_info) { 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()); } } @@ -2265,6 +2287,7 @@ mod tests { use std::sync::atomic::{AtomicU16, Ordering}; use std::time::Duration; + use bitcoin::hashes::Hash; use lightning::util::test_utils::TestLogger; use super::*; @@ -2272,6 +2295,28 @@ mod tests { use crate::payment::store::LSPS2Parameters; use crate::types::DynStoreWrapper; + #[test] + fn discarded_contribution_preserves_inputs_and_outputs() { + let inputs = vec![ + OutPoint::new(bitcoin::Txid::from_byte_array([1; 32]), 2), + OutPoint::new(bitcoin::Txid::from_byte_array([3; 32]), 4), + ]; + let outputs = + vec![bitcoin::ScriptBuf::from_bytes(vec![5]), bitcoin::ScriptBuf::from_bytes(vec![6])]; + + let tx = discarded_funding_transaction(FundingInfo::Contribution { + inputs: inputs.clone(), + outputs: outputs.clone(), + }) + .unwrap(); + + assert_eq!(tx.input.iter().map(|txin| txin.previous_output).collect::>(), inputs,); + assert_eq!( + tx.output.iter().map(|txout| txout.script_pubkey.clone()).collect::>(), + outputs, + ); + } + #[test] fn lsps2_payment_metadata_decodes_total_fee_limit() { let metadata = PaymentMetadata { diff --git a/src/io/mod.rs b/src/io/mod.rs index c70c68d96..55acab932 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 4657688f5..467da4288 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 782112dad..36041d1ee 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 df95c11ec..f567587ca 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; @@ -203,6 +204,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() } @@ -455,8 +477,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() }; @@ -602,7 +625,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); @@ -630,18 +653,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) } @@ -858,6 +884,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. @@ -869,32 +898,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 { @@ -953,8 +998,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!( @@ -1073,7 +1121,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. @@ -1139,8 +1187,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 @@ -1166,8 +1215,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); @@ -1196,18 +1246,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); @@ -1308,24 +1364,26 @@ impl Wallet { return Err(()); } + // Keep selected wallet inputs unavailable until LDK either broadcasts a transaction + // spending them or returns them through `DiscardFunding`. + for txin in unsigned_tx.input.iter().filter(|txin| { + must_spend.iter().all(|input| input.outpoint != txin.previous_output) + }) { + locked_wallet.lock_outpoint(txin.previous_output); + } + 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 - }; + let change_set = locked_wallet.take_staged().unwrap_or_default(); (CoinSelection { confirmed_utxos, change_output }, change_set) }; - 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); - })?; - } + locked_persister.persist_changeset(change_set).await.map_err(|e| { + log_error!(self.logger, "Failed to persist wallet: {}", e); + })?; Ok(coin_selection) } @@ -1338,8 +1396,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; @@ -1508,6 +1568,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( @@ -2188,8 +2345,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, @@ -2226,6 +2386,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, @@ -2677,10 +2839,10 @@ mod tests { use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; - use bdk_chain::{BlockId, ConfirmationBlockTime}; + use bdk_chain::{BlockId, CheckPoint, ConfirmationBlockTime, TxUpdate}; use bdk_wallet::Wallet as BdkWallet; use bitcoin::hashes::Hash; - use bitcoin::Network; + use bitcoin::{Network, TxIn}; use lightning::io; use lightning::util::persist::{KVStore, PageToken, PaginatedKVStore, PaginatedListResponse}; @@ -2842,6 +3004,132 @@ mod tests { )) } + #[tokio::test] + async fn splice_coin_selection_locks_inputs_until_cancelled() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(Arc::clone(&store), false).await; + let (funding_tx, block_id) = { + let mut locked_wallet = wallet.inner.lock().unwrap(); + let outputs = (0..2) + .map(|_| TxOut { + value: Amount::from_sat(100_000), + script_pubkey: locked_wallet + .reveal_next_address(KeychainKind::External) + .address + .script_pubkey(), + }) + .collect(); + let funding_tx = Transaction { + version: bitcoin::transaction::Version::TWO, + lock_time: LockTime::ZERO, + input: Vec::new(), + output: outputs, + }; + let block_id = BlockId { + height: locked_wallet.latest_checkpoint().height() + 1, + hash: bitcoin::BlockHash::from_byte_array([42; 32]), + }; + (funding_tx, block_id) + }; + let funding_txid = funding_tx.compute_txid(); + 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([ + wallet.inner.lock().unwrap().latest_checkpoint().block_id(), + block_id, + ]) + .unwrap(); + wallet + .apply_update(Update { tx_update, chain: Some(chain), ..Default::default() }) + .await + .unwrap(); + + let payment = TxOut { + value: Amount::from_sat(50_000), + script_pubkey: ScriptBuf::new_p2wpkh(&WPubkeyHash::from_slice(&[1; 20]).unwrap()), + }; + let fee_rate = FeeRate::from_sat_per_kwu(250); + let selection = + Wallet::select_confirmed_utxos(&wallet, Vec::new(), &[payment.clone()], fee_rate) + .await + .unwrap(); + let selected_outpoints = selection + .confirmed_utxos + .iter() + .cloned() + .map(ConfirmedUtxo::into_utxo) + .map(|utxo| utxo.outpoint) + .collect::>(); + assert!(!selected_outpoints.is_empty()); + assert!( + selected_outpoints.iter().all(|outpoint| wallet + .inner + .lock() + .unwrap() + .is_outpoint_locked(*outpoint)), + "splice coin selection must lock selected wallet inputs", + ); + drop(wallet); + + let reloaded = new_test_wallet(Arc::clone(&store), true).await; + assert!( + selected_outpoints.iter().all(|outpoint| reloaded + .inner + .lock() + .unwrap() + .is_outpoint_locked(*outpoint)), + "splice input locks must survive a wallet reload", + ); + let second_selection = + Wallet::select_confirmed_utxos(&reloaded, Vec::new(), &[payment.clone()], fee_rate) + .await + .unwrap(); + let second_outpoints = second_selection + .confirmed_utxos + .into_iter() + .map(ConfirmedUtxo::into_utxo) + .map(|utxo| utxo.outpoint) + .collect::>(); + assert!( + selected_outpoints.iter().all(|outpoint| !second_outpoints.contains(outpoint)), + "subsequent splice coin selection must not reuse locked inputs", + ); + + let cancelled_tx = Transaction { + version: bitcoin::transaction::Version::TWO, + lock_time: LockTime::ZERO, + input: selected_outpoints + .iter() + .map(|outpoint| TxIn { previous_output: *outpoint, ..TxIn::default() }) + .collect(), + output: selection.change_output.into_iter().collect(), + }; + reloaded.cancel_tx(cancelled_tx).await.unwrap(); + drop(reloaded); + let reloaded = new_test_wallet(store, true).await; + assert!( + selected_outpoints.iter().all(|outpoint| !reloaded + .inner + .lock() + .unwrap() + .is_outpoint_locked(*outpoint)), + "discarded splice inputs must be unlocked persistently", + ); + let replacement_selection = + Wallet::select_confirmed_utxos(&reloaded, Vec::new(), &[payment], fee_rate) + .await + .unwrap(); + let replacement_outpoints = replacement_selection + .confirmed_utxos + .into_iter() + .map(ConfirmedUtxo::into_utxo) + .map(|utxo| utxo.outpoint) + .collect::>(); + assert_eq!(replacement_outpoints, selected_outpoints); + } + fn pooled_indices(wallet: &Wallet) -> Vec { wallet.address_pool.lock().unwrap().available.iter().map(|(index, _)| *index).collect() } @@ -4214,4 +4502,127 @@ mod tests { PaymentKind::Onchain { tx_type: Some(TransactionType::InteractiveFunding { .. }), .. } )); } + + #[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(Broadcaster::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(), + PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE.to_string(), + PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE.to_string(), + Arc::clone(&store), + Arc::clone(&logger), + )); + let pending_payment_store = Arc::new(PendingPaymentStore::new( + Vec::new(), + PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE.to_string(), + PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE.to_string(), + Arc::clone(&store), + Arc::clone(&logger), + )); + + let 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: bitcoin::transaction::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: bitcoin::transaction::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, + Vec::new(), + 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 6384d0fce..e2f430e5d 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,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( @@ -312,7 +325,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 +444,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 c6a707bcd..80b7d2516 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 c1e973091..55af77cf1 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -676,11 +676,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; @@ -732,6 +727,49 @@ async fn multi_hop_sending() { expect_payment_successful_event!(nodes[0], outbound_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(); @@ -4370,17 +4408,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 =