Fix wallet UTXO reuse for funding transactions and onchain spends

src/chain/bitcoind.rs

@@ -1274,18 +1274,18 @@ impl BitcoindClient {
 		&self, bdk_unconfirmed_txids: Vec<Txid>,
 	) -> Result<Vec<(Txid, u64)>, 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<HashMap<Txid, MempoolEntry>>,
-		bdk_unconfirmed_txids: Vec<Txid>,
+		latest_mempool_timestamp: &AtomicU64, bdk_unconfirmed_txids: Vec<Txid>,
 	) -> Result<Vec<(Txid, u64)>, 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::<u8>(), 0..100), 0..20)

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;

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());
 					}
 				}

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

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))
 }

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<BroadcastPackage, Error> {
 		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?;
 				}