Fix wallet UTXO reuse for funding transactions and onchain spends

src/chain/bitcoind.rs

@@ -1302,18 +1302,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
@@ -1322,16 +1322,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)
 	}
@@ -1616,8 +1617,10 @@ impl std::error::Error for BitcoindClientError {}
 
 #[cfg(test)]
 mod tests {
+	use std::collections::HashMap;
+	use std::sync::atomic::{AtomicU64, Ordering};
 	use std::sync::Mutex;
-	use std::time::Duration;
+	use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 	use bitcoin::hashes::Hash;
 	use bitcoin::{FeeRate, OutPoint, ScriptBuf, Transaction, TxIn, TxOut, Txid, Witness};
@@ -1628,7 +1631,7 @@ mod tests {
 	use serde_json::json;
 
 	use crate::chain::bitcoind::{
- acquire_initial_wallet_sync_guard, FeeResponse, GetMempoolEntryResponse,
+ acquire_initial_wallet_sync_guard, BitcoindClient, FeeResponse, GetMempoolEntryResponse,
  GetRawMempoolResponse, GetRawTransactionResponse, MempoolMinFeeResponse,
 	};
 	use crate::chain::{WalletSyncGuard, WalletSyncStatus};
@@ -1659,6 +1662,55 @@ mod tests {
  acquired_guard.complete(Ok(()));
 	}
 
+	#[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

@@ -523,15 +523,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

@@ -538,6 +538,28 @@ impl Future for EventFuture {
 	}
 }
 
+fn discarded_funding_transaction(funding_info: FundingInfo) -> Option<bitcoin::Transaction> {
+	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<L: Deref + Clone + Sync + Send + 'static>
 where
 	L::Target: LdkLogger,
@@ -748,6 +770,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()
@@ -778,27 +801,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
+ )
+ },
+ }
  },
  }
  },
@@ -1941,27 +1976,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());
  }
  }
@@ -2230,13 +2252,36 @@ mod tests {
 	use std::sync::atomic::{AtomicU16, Ordering};
 	use std::time::Duration;
 
+	use bitcoin::hashes::Hash;
 	use lightning::util::test_utils::TestLogger;
 
 	use super::*;
 	use crate::io::test_utils::InMemoryStore;
 	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::<Vec<_>>(), inputs,);
+ assert_eq!(
+ tx.output.iter().map(|txout| txout.script_pubkey.clone()).collect::<Vec<_>>(),
+ outputs,
+ );
+	}
+
 	#[test]
 	fn lsps2_payment_metadata_decodes_total_fee_limit() {
  let metadata = PaymentMetadata {

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

@@ -19,6 +19,7 @@ use bdk_chain::local_chain::ChangeSet as BdkLocalChainChangeSet;
 use bdk_chain::miniscript::{Descriptor, DescriptorPublicKey};
 use bdk_chain::tx_graph::ChangeSet as BdkTxGraphChangeSet;
 use bdk_chain::ConfirmationBlockTime;
+use bdk_wallet::locked_outpoints::ChangeSet as BdkLockedOutpointsChangeSet;
 use bdk_wallet::ChangeSet as BdkWalletChangeSet;
 use bitcoin::Network;
 use lightning::ln::msgs::DecodeError;
@@ -715,6 +716,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,
@@ -757,6 +767,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?;
  }

src/chain/bitcoind.rs

@@ -1302,18 +1302,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
@@ -1322,16 +1322,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)
 	}
@@ -1616,8 +1617,10 @@ impl std::error::Error for BitcoindClientError {}
 
 #[cfg(test)]
 mod tests {
+	use std::collections::HashMap;
+	use std::sync::atomic::{AtomicU64, Ordering};
 	use std::sync::Mutex;
-	use std::time::Duration;
+	use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 	use bitcoin::hashes::Hash;
 	use bitcoin::{FeeRate, OutPoint, ScriptBuf, Transaction, TxIn, TxOut, Txid, Witness};
@@ -1628,7 +1631,7 @@ mod tests {
 	use serde_json::json;
 
 	use crate::chain::bitcoind::{
- acquire_initial_wallet_sync_guard, FeeResponse, GetMempoolEntryResponse,
+ acquire_initial_wallet_sync_guard, BitcoindClient, FeeResponse, GetMempoolEntryResponse,
  GetRawMempoolResponse, GetRawTransactionResponse, MempoolMinFeeResponse,
 	};
 	use crate::chain::{WalletSyncGuard, WalletSyncStatus};
@@ -1659,6 +1662,55 @@ mod tests {
  acquired_guard.complete(Ok(()));
 	}
 
+	#[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

@@ -523,15 +523,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

@@ -538,6 +538,28 @@ impl Future for EventFuture {
 	}
 }
 
+fn discarded_funding_transaction(funding_info: FundingInfo) -> Option<bitcoin::Transaction> {
+	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<L: Deref + Clone + Sync + Send + 'static>
 where
 	L::Target: LdkLogger,
@@ -748,6 +770,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()
@@ -778,27 +801,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
+ )
+ },
+ }
  },
  }
  },
@@ -1941,27 +1976,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());
  }
  }
@@ -2230,13 +2252,36 @@ mod tests {
 	use std::sync::atomic::{AtomicU16, Ordering};
 	use std::time::Duration;
 
+	use bitcoin::hashes::Hash;
 	use lightning::util::test_utils::TestLogger;
 
 	use super::*;
 	use crate::io::test_utils::InMemoryStore;
 	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::<Vec<_>>(), inputs,);
+ assert_eq!(
+ tx.output.iter().map(|txout| txout.script_pubkey.clone()).collect::<Vec<_>>(),
+ outputs,
+ );
+	}
+
 	#[test]
 	fn lsps2_payment_metadata_decodes_total_fee_limit() {
  let metadata = PaymentMetadata {

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

@@ -19,6 +19,7 @@ use bdk_chain::local_chain::ChangeSet as BdkLocalChainChangeSet;
 use bdk_chain::miniscript::{Descriptor, DescriptorPublicKey};
 use bdk_chain::tx_graph::ChangeSet as BdkTxGraphChangeSet;
 use bdk_chain::ConfirmationBlockTime;
+use bdk_wallet::locked_outpoints::ChangeSet as BdkLockedOutpointsChangeSet;
 use bdk_wallet::ChangeSet as BdkWalletChangeSet;
 use bitcoin::Network;
 use lightning::ln::msgs::DecodeError;
@@ -715,6 +716,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,
@@ -757,6 +767,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?;
  }

src/chain/bitcoind.rs

@@ -1302,18 +1302,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
@@ -1322,16 +1322,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)
 	}
@@ -1616,8 +1617,10 @@ impl std::error::Error for BitcoindClientError {}
 
 #[cfg(test)]
 mod tests {
+	use std::collections::HashMap;
+	use std::sync::atomic::{AtomicU64, Ordering};
 	use std::sync::Mutex;
-	use std::time::Duration;
+	use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 	use bitcoin::hashes::Hash;
 	use bitcoin::{FeeRate, OutPoint, ScriptBuf, Transaction, TxIn, TxOut, Txid, Witness};
@@ -1628,7 +1631,7 @@ mod tests {
 	use serde_json::json;
 
 	use crate::chain::bitcoind::{
- acquire_initial_wallet_sync_guard, FeeResponse, GetMempoolEntryResponse,
+ acquire_initial_wallet_sync_guard, BitcoindClient, FeeResponse, GetMempoolEntryResponse,
  GetRawMempoolResponse, GetRawTransactionResponse, MempoolMinFeeResponse,
 	};
 	use crate::chain::{WalletSyncGuard, WalletSyncStatus};
@@ -1659,6 +1662,55 @@ mod tests {
  acquired_guard.complete(Ok(()));
 	}
 
+	#[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

@@ -523,15 +523,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

@@ -538,6 +538,28 @@ impl Future for EventFuture {
 	}
 }
 
+fn discarded_funding_transaction(funding_info: FundingInfo) -> Option<bitcoin::Transaction> {
+	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<L: Deref + Clone + Sync + Send + 'static>
 where
 	L::Target: LdkLogger,
@@ -748,6 +770,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()
@@ -778,27 +801,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
+ )
+ },
+ }
  },
  }
  },
@@ -1941,27 +1976,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());
  }
  }
@@ -2230,13 +2252,36 @@ mod tests {
 	use std::sync::atomic::{AtomicU16, Ordering};
 	use std::time::Duration;
 
+	use bitcoin::hashes::Hash;
 	use lightning::util::test_utils::TestLogger;
 
 	use super::*;
 	use crate::io::test_utils::InMemoryStore;
 	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::<Vec<_>>(), inputs,);
+ assert_eq!(
+ tx.output.iter().map(|txout| txout.script_pubkey.clone()).collect::<Vec<_>>(),
+ outputs,
+ );
+	}
+
 	#[test]
 	fn lsps2_payment_metadata_decodes_total_fee_limit() {
  let metadata = PaymentMetadata {

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

@@ -19,6 +19,7 @@ use bdk_chain::local_chain::ChangeSet as BdkLocalChainChangeSet;
 use bdk_chain::miniscript::{Descriptor, DescriptorPublicKey};
 use bdk_chain::tx_graph::ChangeSet as BdkTxGraphChangeSet;
 use bdk_chain::ConfirmationBlockTime;
+use bdk_wallet::locked_outpoints::ChangeSet as BdkLockedOutpointsChangeSet;
 use bdk_wallet::ChangeSet as BdkWalletChangeSet;
 use bitcoin::Network;
 use lightning::ln::msgs::DecodeError;
@@ -715,6 +716,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,
@@ -757,6 +767,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?;
  }

src/chain/bitcoind.rs

@@ -1302,18 +1302,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
@@ -1322,16 +1322,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)
 	}
@@ -1616,8 +1617,10 @@ impl std::error::Error for BitcoindClientError {}
 
 #[cfg(test)]
 mod tests {
+	use std::collections::HashMap;
+	use std::sync::atomic::{AtomicU64, Ordering};
 	use std::sync::Mutex;
-	use std::time::Duration;
+	use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 	use bitcoin::hashes::Hash;
 	use bitcoin::{FeeRate, OutPoint, ScriptBuf, Transaction, TxIn, TxOut, Txid, Witness};
@@ -1628,7 +1631,7 @@ mod tests {
 	use serde_json::json;
 
 	use crate::chain::bitcoind::{
- acquire_initial_wallet_sync_guard, FeeResponse, GetMempoolEntryResponse,
+ acquire_initial_wallet_sync_guard, BitcoindClient, FeeResponse, GetMempoolEntryResponse,
  GetRawMempoolResponse, GetRawTransactionResponse, MempoolMinFeeResponse,
 	};
 	use crate::chain::{WalletSyncGuard, WalletSyncStatus};
@@ -1659,6 +1662,55 @@ mod tests {
  acquired_guard.complete(Ok(()));
 	}
 
+	#[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

@@ -523,15 +523,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

@@ -538,6 +538,28 @@ impl Future for EventFuture {
 	}
 }
 
+fn discarded_funding_transaction(funding_info: FundingInfo) -> Option<bitcoin::Transaction> {
+	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<L: Deref + Clone + Sync + Send + 'static>
 where
 	L::Target: LdkLogger,
@@ -748,6 +770,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()
@@ -778,27 +801,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
+ )
+ },
+ }
  },
  }
  },
@@ -1941,27 +1976,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());
  }
  }
@@ -2230,13 +2252,36 @@ mod tests {
 	use std::sync::atomic::{AtomicU16, Ordering};
 	use std::time::Duration;
 
+	use bitcoin::hashes::Hash;
 	use lightning::util::test_utils::TestLogger;
 
 	use super::*;
 	use crate::io::test_utils::InMemoryStore;
 	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::<Vec<_>>(), inputs,);
+ assert_eq!(
+ tx.output.iter().map(|txout| txout.script_pubkey.clone()).collect::<Vec<_>>(),
+ outputs,
+ );
+	}
+
 	#[test]
 	fn lsps2_payment_metadata_decodes_total_fee_limit() {
  let metadata = PaymentMetadata {

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

@@ -19,6 +19,7 @@ use bdk_chain::local_chain::ChangeSet as BdkLocalChainChangeSet;
 use bdk_chain::miniscript::{Descriptor, DescriptorPublicKey};
 use bdk_chain::tx_graph::ChangeSet as BdkTxGraphChangeSet;
 use bdk_chain::ConfirmationBlockTime;
+use bdk_wallet::locked_outpoints::ChangeSet as BdkLockedOutpointsChangeSet;
 use bdk_wallet::ChangeSet as BdkWalletChangeSet;
 use bitcoin::Network;
 use lightning::ln::msgs::DecodeError;
@@ -715,6 +716,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,
@@ -757,6 +767,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?;
  }

src/chain/bitcoind.rs

@@ -1302,18 +1302,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
@@ -1322,16 +1322,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)
 	}
@@ -1616,8 +1617,10 @@ impl std::error::Error for BitcoindClientError {}
 
 #[cfg(test)]
 mod tests {
+	use std::collections::HashMap;
+	use std::sync::atomic::{AtomicU64, Ordering};
 	use std::sync::Mutex;
-	use std::time::Duration;
+	use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 	use bitcoin::hashes::Hash;
 	use bitcoin::{FeeRate, OutPoint, ScriptBuf, Transaction, TxIn, TxOut, Txid, Witness};
@@ -1628,7 +1631,7 @@ mod tests {
 	use serde_json::json;
 
 	use crate::chain::bitcoind::{
- acquire_initial_wallet_sync_guard, FeeResponse, GetMempoolEntryResponse,
+ acquire_initial_wallet_sync_guard, BitcoindClient, FeeResponse, GetMempoolEntryResponse,
  GetRawMempoolResponse, GetRawTransactionResponse, MempoolMinFeeResponse,
 	};
 	use crate::chain::{WalletSyncGuard, WalletSyncStatus};
@@ -1659,6 +1662,55 @@ mod tests {
  acquired_guard.complete(Ok(()));
 	}
 
+	#[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

@@ -523,15 +523,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

@@ -538,6 +538,28 @@ impl Future for EventFuture {
 	}
 }
 
+fn discarded_funding_transaction(funding_info: FundingInfo) -> Option<bitcoin::Transaction> {
+	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<L: Deref + Clone + Sync + Send + 'static>
 where
 	L::Target: LdkLogger,
@@ -748,6 +770,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()
@@ -778,27 +801,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
+ )
+ },
+ }
  },
  }
  },
@@ -1941,27 +1976,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());
  }
  }
@@ -2230,13 +2252,36 @@ mod tests {
 	use std::sync::atomic::{AtomicU16, Ordering};
 	use std::time::Duration;
 
+	use bitcoin::hashes::Hash;
 	use lightning::util::test_utils::TestLogger;
 
 	use super::*;
 	use crate::io::test_utils::InMemoryStore;
 	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::<Vec<_>>(), inputs,);
+ assert_eq!(
+ tx.output.iter().map(|txout| txout.script_pubkey.clone()).collect::<Vec<_>>(),
+ outputs,
+ );
+	}
+
 	#[test]
 	fn lsps2_payment_metadata_decodes_total_fee_limit() {
  let metadata = PaymentMetadata {

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

@@ -19,6 +19,7 @@ use bdk_chain::local_chain::ChangeSet as BdkLocalChainChangeSet;
 use bdk_chain::miniscript::{Descriptor, DescriptorPublicKey};
 use bdk_chain::tx_graph::ChangeSet as BdkTxGraphChangeSet;
 use bdk_chain::ConfirmationBlockTime;
+use bdk_wallet::locked_outpoints::ChangeSet as BdkLockedOutpointsChangeSet;
 use bdk_wallet::ChangeSet as BdkWalletChangeSet;
 use bitcoin::Network;
 use lightning::ln::msgs::DecodeError;
@@ -715,6 +716,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,
@@ -757,6 +767,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?;
  }

src/chain/bitcoind.rs

@@ -1302,18 +1302,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
@@ -1322,16 +1322,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)
 	}
@@ -1616,8 +1617,10 @@ impl std::error::Error for BitcoindClientError {}
 
 #[cfg(test)]
 mod tests {
+	use std::collections::HashMap;
+	use std::sync::atomic::{AtomicU64, Ordering};
 	use std::sync::Mutex;
-	use std::time::Duration;
+	use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 	use bitcoin::hashes::Hash;
 	use bitcoin::{FeeRate, OutPoint, ScriptBuf, Transaction, TxIn, TxOut, Txid, Witness};
@@ -1628,7 +1631,7 @@ mod tests {
 	use serde_json::json;
 
 	use crate::chain::bitcoind::{
- acquire_initial_wallet_sync_guard, FeeResponse, GetMempoolEntryResponse,
+ acquire_initial_wallet_sync_guard, BitcoindClient, FeeResponse, GetMempoolEntryResponse,
  GetRawMempoolResponse, GetRawTransactionResponse, MempoolMinFeeResponse,
 	};
 	use crate::chain::{WalletSyncGuard, WalletSyncStatus};
@@ -1659,6 +1662,55 @@ mod tests {
  acquired_guard.complete(Ok(()));
 	}
 
+	#[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

@@ -523,15 +523,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

@@ -538,6 +538,28 @@ impl Future for EventFuture {
 	}
 }
 
+fn discarded_funding_transaction(funding_info: FundingInfo) -> Option<bitcoin::Transaction> {
+	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<L: Deref + Clone + Sync + Send + 'static>
 where
 	L::Target: LdkLogger,
@@ -748,6 +770,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()
@@ -778,27 +801,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
+ )
+ },
+ }
  },
  }
  },
@@ -1941,27 +1976,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());
  }
  }
@@ -2230,13 +2252,36 @@ mod tests {
 	use std::sync::atomic::{AtomicU16, Ordering};
 	use std::time::Duration;
 
+	use bitcoin::hashes::Hash;
 	use lightning::util::test_utils::TestLogger;
 
 	use super::*;
 	use crate::io::test_utils::InMemoryStore;
 	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::<Vec<_>>(), inputs,);
+ assert_eq!(
+ tx.output.iter().map(|txout| txout.script_pubkey.clone()).collect::<Vec<_>>(),
+ outputs,
+ );
+	}
+
 	#[test]
 	fn lsps2_payment_metadata_decodes_total_fee_limit() {
  let metadata = PaymentMetadata {

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

@@ -19,6 +19,7 @@ use bdk_chain::local_chain::ChangeSet as BdkLocalChainChangeSet;
 use bdk_chain::miniscript::{Descriptor, DescriptorPublicKey};
 use bdk_chain::tx_graph::ChangeSet as BdkTxGraphChangeSet;
 use bdk_chain::ConfirmationBlockTime;
+use bdk_wallet::locked_outpoints::ChangeSet as BdkLockedOutpointsChangeSet;
 use bdk_wallet::ChangeSet as BdkWalletChangeSet;
 use bitcoin::Network;
 use lightning::ln::msgs::DecodeError;
@@ -715,6 +716,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,
@@ -757,6 +767,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?;
  }

src/chain/bitcoind.rs

@@ -1302,18 +1302,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
@@ -1322,16 +1322,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)
 	}
@@ -1616,8 +1617,10 @@ impl std::error::Error for BitcoindClientError {}
 
 #[cfg(test)]
 mod tests {
+	use std::collections::HashMap;
+	use std::sync::atomic::{AtomicU64, Ordering};
 	use std::sync::Mutex;
-	use std::time::Duration;
+	use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 	use bitcoin::hashes::Hash;
 	use bitcoin::{FeeRate, OutPoint, ScriptBuf, Transaction, TxIn, TxOut, Txid, Witness};
@@ -1628,7 +1631,7 @@ mod tests {
 	use serde_json::json;
 
 	use crate::chain::bitcoind::{
- acquire_initial_wallet_sync_guard, FeeResponse, GetMempoolEntryResponse,
+ acquire_initial_wallet_sync_guard, BitcoindClient, FeeResponse, GetMempoolEntryResponse,
  GetRawMempoolResponse, GetRawTransactionResponse, MempoolMinFeeResponse,
 	};
 	use crate::chain::{WalletSyncGuard, WalletSyncStatus};
@@ -1659,6 +1662,55 @@ mod tests {
  acquired_guard.complete(Ok(()));
 	}
 
+	#[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

@@ -523,15 +523,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

@@ -538,6 +538,28 @@ impl Future for EventFuture {
 	}
 }
 
+fn discarded_funding_transaction(funding_info: FundingInfo) -> Option<bitcoin::Transaction> {
+	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<L: Deref + Clone + Sync + Send + 'static>
 where
 	L::Target: LdkLogger,
@@ -748,6 +770,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()
@@ -778,27 +801,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
+ )
+ },
+ }
  },
  }
  },
@@ -1941,27 +1976,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());
  }
  }
@@ -2230,13 +2252,36 @@ mod tests {
 	use std::sync::atomic::{AtomicU16, Ordering};
 	use std::time::Duration;
 
+	use bitcoin::hashes::Hash;
 	use lightning::util::test_utils::TestLogger;
 
 	use super::*;
 	use crate::io::test_utils::InMemoryStore;
 	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::<Vec<_>>(), inputs,);
+ assert_eq!(
+ tx.output.iter().map(|txout| txout.script_pubkey.clone()).collect::<Vec<_>>(),
+ outputs,
+ );
+	}
+
 	#[test]
 	fn lsps2_payment_metadata_decodes_total_fee_limit() {
  let metadata = PaymentMetadata {

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

@@ -19,6 +19,7 @@ use bdk_chain::local_chain::ChangeSet as BdkLocalChainChangeSet;
 use bdk_chain::miniscript::{Descriptor, DescriptorPublicKey};
 use bdk_chain::tx_graph::ChangeSet as BdkTxGraphChangeSet;
 use bdk_chain::ConfirmationBlockTime;
+use bdk_wallet::locked_outpoints::ChangeSet as BdkLockedOutpointsChangeSet;
 use bdk_wallet::ChangeSet as BdkWalletChangeSet;
 use bitcoin::Network;
 use lightning::ln::msgs::DecodeError;
@@ -715,6 +716,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,
@@ -757,6 +767,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?;
  }