From eb1a7a21d31f7929d0e0dd9bdd41f377bd5f70c7 Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Mon, 10 Aug 2026 14:00:28 +0200 Subject: [PATCH 1/2] Prevent wallet input reuse before sync Record wallet transactions before broadcast and reserve funding inputs until their transactions are durable. This prevents concurrent operations from selecting the same inputs. Dropped transactions remain recoverable through the existing rebroadcast path. Co-Authored-By: HAL 9000 --- src/chain/mod.rs | 8 +- src/event.rs | 91 ++++---- src/io/mod.rs | 7 + src/io/utils.rs | 13 ++ src/tx_broadcaster.rs | 21 +- src/wallet/mod.rs | 368 +++++++++++++++++++++++++++++--- src/wallet/persist.rs | 49 ++++- src/wallet/ser.rs | 30 +++ tests/integration_tests_rust.rs | 58 ++++- 9 files changed, 561 insertions(+), 84 deletions(-) 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 = From 2932580fd0883b01a3f944a400af0ef6811e7f41 Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Mon, 10 Aug 2026 14:03:12 +0200 Subject: [PATCH 2/2] Timestamp mempool evictions when observed Locally inserted transactions can be newer than Bitcoin Core's latest mempool timestamp. Reporting that stale timestamp for an eviction makes BDK ignore it and leaves the transaction's inputs unavailable. Use the later of the local observation time and Bitcoin Core's mempool time. This makes local transactions evictable without regressing nodes whose Bitcoin Core clock is ahead of the application clock. Co-Authored-By: HAL 9000 --- src/chain/bitcoind.rs | 74 +++++++++++++++++++++++++++++++++++++------ 1 file changed, 64 insertions(+), 10 deletions(-) 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)