Merge PR 'release utxos from failed splices' (#4973)

### lightning/src/ln/splicing_tests.rs
@@ -12,7 +12,7 @@
 use crate::chain::chaininterface::{FundingPurpose, TransactionType, FEERATE_FLOOR_SATS_PER_KW};
 use crate::chain::channelmonitor::{ANTI_REORG_DELAY, LATENCY_GRACE_PERIOD_BLOCKS};
 use crate::chain::transaction::OutPoint;
-use crate::chain::{ChannelMonitorUpdateStatus, Confirm};
+use crate::chain::{ChannelMonitorUpdateStatus, ClaimId, Confirm};
 use crate::events::{
 	ClosureReason, Event, FundingInfo, HTLCHandlingFailureType, NegotiationFailureReason,
 };
@@ -41,6 +41,7 @@ use crate::util::config::UserConfig;
 use crate::util::errors::APIError;
 use crate::util::ser::Writeable;
 use crate::util::test_channel_signer::SignerOp;
+use crate::util::test_utils::{TestLogger, TestWalletSource};
 use crate::util::wallet_utils::{
 	CoinSelection, CoinSelectionSourceSync, ConfirmedUtxo, Input, WalletSourceSync, WalletSync,
 };
@@ -16805,3 +16806,97 @@ fn test_stale_template_from_dropped_candidate_releases_only_its_own_input() {
 	assert!(nodes[0].node.get_and_clear_pending_events().is_empty());
 	let _stfu = get_event_msg!(nodes[0], MessageSendEvent::SendStfu, node_id_1);
 }
+
+/// Initiates a splice-in funded through `wallet`, then has the acceptor abort it during
+/// interactive-tx construction. Handles the resulting `DiscardFunding` event by releasing the
+/// discarded inputs back to `wallet`. Returns the contribution that was discarded.
+#[cfg(test)]
+fn fail_splice_in_with_tx_abort<'a, 'b, 'c, 'd>(
+	initiator: &'a Node<'b, 'c, 'd>, acceptor: &'a Node<'b, 'c, 'd>, channel_id: ChannelId,
+	wallet: &WalletSync<Arc<TestWalletSource>, &'d TestLogger>, value_added: Amount,
+) -> FundingContribution {
+	let node_id_initiator = initiator.node.get_our_node_id();
+	let node_id_acceptor = acceptor.node.get_our_node_id();
+	let feerate = FeeRate::from_sat_per_kwu(FEERATE_FLOOR_SATS_PER_KW as u64);
+
+	let funding_template = initiator.node.splice_channel(&channel_id, &node_id_acceptor).unwrap();
+	let funding_contribution =
+		funding_template.splice_in_sync(value_added, feerate, FeeRate::MAX, wallet).unwrap();
+	initiator
+		.node
+		.funding_contributed(&channel_id, &node_id_acceptor, funding_contribution.clone(), None)
+		.unwrap();
+
+	let _ = complete_splice_handshake(initiator, acceptor);
+	let tx_add_input =
+		get_event_msg!(initiator, MessageSendEvent::SendTxAddInput, node_id_acceptor);
+	acceptor.node.handle_tx_add_input(node_id_initiator, &tx_add_input);
+	let _ = get_event_msg!(acceptor, MessageSendEvent::SendTxComplete, node_id_initiator);
+
+	let tx_abort = msgs::TxAbort { channel_id, data: Vec::new() };
+	initiator.node.handle_tx_abort(node_id_acceptor, &tx_abort);
+	let (discarded_inputs, _) = expect_failed_rbf_events(
+		initiator,
+		&channel_id,
+		&funding_contribution,
+		NegotiationFailureReason::CounterpartyAborted { msg: UntrustedString(String::new()) },
+	);
+	let contributed_inputs =
+		funding_contribution.inputs().iter().map(|input| input.outpoint()).collect::<Vec<_>>();
+	assert_eq!(discarded_inputs, contributed_inputs);
+	wallet.release_utxos(&discarded_inputs);
+
+	let tx_abort = get_event_msg!(initiator, MessageSendEvent::SendTxAbort, node_id_acceptor);
+	acceptor.node.handle_tx_abort(node_id_initiator, &tx_abort);
+	let _ = get_event_msg!(acceptor, MessageSendEvent::SendTxAbort, node_id_initiator);
+	assert!(acceptor.node.get_and_clear_pending_events().is_empty());
+
+	funding_contribution
+}
+
+#[test]
+fn splice_abort_releases_wallet_utxos() {
+	let chanmon_cfgs = create_chanmon_cfgs(2);
+	let node_cfgs = create_node_cfgs(2, &chanmon_cfgs);
+	let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]);
+	let nodes = create_network(2, &node_cfgs, &node_chanmgrs);
+
+	let initiator = &nodes[0];
+	let acceptor = &nodes[1];
+
+	let (_, _, channel_id, _) =
+		create_announced_chan_between_nodes_with_value(&nodes, 0, 1, 100_000, 0);
+
+	// A single wallet UTXO, so every attempt below must be funded from the same one.
+	provide_utxo_reserves(&nodes, 1, Amount::from_sat(100_000));
+
+	// A single long-lived wallet, as a node would hold across all splices and claims.
+	let wallet = WalletSync::new(Arc::clone(&initiator.wallet_source), initiator.logger);
+	let splice_in_value = Amount::from_sat(50_000);
+
+	// Each aborted attempt releases its input via `DiscardFunding`, so the next attempt can
+	// select the same UTXO again.
+	let first =
+		fail_splice_in_with_tx_abort(initiator, acceptor, channel_id, &wallet, splice_in_value);
+	let second =
+		fail_splice_in_with_tx_abort(initiator, acceptor, channel_id, &wallet, splice_in_value);
+	assert_eq!(first.inputs().len(), 1);
+	assert_eq!(first.inputs(), second.inputs());
+
+	// The UTXO is likewise available to a claim, as the bump transaction handler would request
+	// it.
+	let claim_output = TxOut {
+		value: splice_in_value,
+		script_pubkey: initiator.wallet_source.get_change_script().unwrap(),
+	};
+	let claim_selection = wallet
+		.select_confirmed_utxos(
+			Some(ClaimId([42; 32])),
+			Vec::new(),
+			&[claim_output],
+			FEERATE_FLOOR_SATS_PER_KW,
+			u64::MAX,
+		)
+		.unwrap();
+	assert_eq!(claim_selection.confirmed_utxos, first.inputs());
+}

### lightning/src/util/wallet_utils.rs
@@ -466,19 +466,29 @@ pub trait WalletSource {
 /// that would avoid conflicting double spends. If not enough UTXOs are available to do so,
 /// conflicting double spends may happen.
 ///
+/// Reservations of selected UTXOs are tracked in memory, so this wrapper is intended for a
+/// [`WalletSource`] whose wallet does not reserve UTXOs on its own. Wallets that do should
+/// implement [`CoinSelectionSource`] directly.
+///
+/// Selected UTXOs stay reserved until released with [`Wallet::release_utxos`]. This must be
+/// called with the inputs from [`Event::DiscardFunding`] once a transaction funded through this
+/// wallet will no longer be broadcast, such as when a splice negotiation fails.
+///
 /// For a synchronous version of this wrapper, see [`WalletSync`].
 ///
 /// This is not exported to bindings users as async is only supported in Rust.
+///
+/// [`Event::DiscardFunding`]: crate::events::Event::DiscardFunding
 // Note that updates to documentation on this struct should be copied to the synchronous version.
 pub struct Wallet<W: Deref + MaybeSync + MaybeSend, L: Logger + MaybeSync + MaybeSend>
 where
 	W::Target: WalletSource + MaybeSend,
 {
 	source: W,
 	logger: L,
-	// TODO: Do we care about cleaning this up once the UTXOs have a confirmed spend? We can do so
-	// by checking whether any UTXOs that exist in the map are no longer returned in
-	// `list_confirmed_utxos`.
+	// UTXOs previously selected, keyed to the claim they were selected for. Entries are only
+	// removed by `release_utxos`. Entries for spent UTXOs are harmless since they are never
+	// listed by `list_confirmed_utxos` again.
 	locked_utxos: Mutex<HashMap<OutPoint, Option<ClaimId>>>,
 }
 
@@ -492,6 +502,19 @@ where
 		Self { source, logger, locked_utxos: Mutex::new(new_hash_map()) }
 	}
 
+	/// Releases the given UTXOs so they may be selected again.
+	///
+	/// Call this with the inputs from [`Event::DiscardFunding`] once a transaction funded through
+	/// this wallet will no longer be broadcast, such as when a splice negotiation fails.
+	///
+	/// [`Event::DiscardFunding`]: crate::events::Event::DiscardFunding
+	pub fn release_utxos(&self, outpoints: &[OutPoint]) {
+		let mut locked_utxos = self.locked_utxos.lock().unwrap();
+		for outpoint in outpoints {
+			locked_utxos.remove(outpoint);
+		}
+	}
+
 	/// Performs coin selection on the set of UTXOs obtained from
 	/// [`WalletSource::list_confirmed_utxos`]. Its algorithm can be described as "smallest
 	/// above-dust-after-spend first", with a slight twist: we may skip UTXOs that are above dust at
@@ -521,6 +544,7 @@ where
 		let mut selected_amount;
 		let mut total_fees;
 		let mut selected_utxos;
+		let mut prev_locks = Vec::new();
 		{
 			let mut locked_utxos = self.locked_utxos.lock().unwrap();
 			let mut eligible_utxos = utxos
@@ -623,46 +647,62 @@ where
 				total_fees -= fee_to_spend_utxo;
 			}
 			for (utxo, _) in &selected_utxos {
-				locked_utxos.insert(utxo.outpoint, claim_id);
+				let prev = locked_utxos.insert(utxo.outpoint, claim_id);
+				if prev != Some(claim_id) {
+					prev_locks.push((utxo.outpoint, prev));
+				}
 			}
 		}
 
-		let remaining_amount = selected_amount - target_amount_sat - total_fees;
-		let change_script = self.source.get_change_script().await?;
-		let change_output_fee = fee_for_weight(
-			target_feerate_sat_per_1000_weight,
-			(8 /* value */ + change_script.consensus_encode(&mut sink()).unwrap() as u64)
-				* WITNESS_SCALE_FACTOR as u64,
-		);
-		let change_output_amount =
-			Amount::from_sat(remaining_amount.to_sat().saturating_sub(change_output_fee));
-		let change_output = if change_output_amount < change_script.minimal_non_dust() {
-			log_debug!(self.logger, "Coin selection attempt did not yield change output");
-			None
-		} else {
-			Some(TxOut { script_pubkey: change_script, value: change_output_amount })
-		};
+		let selection = async {
+			let remaining_amount = selected_amount - target_amount_sat - total_fees;
+			let change_script = self.source.get_change_script().await?;
+			let change_output_fee = fee_for_weight(
+				target_feerate_sat_per_1000_weight,
+				(8 /* value */ + change_script.consensus_encode(&mut sink()).unwrap() as u64)
+					* WITNESS_SCALE_FACTOR as u64,
+			);
+			let change_output_amount =
+				Amount::from_sat(remaining_amount.to_sat().saturating_sub(change_output_fee));
+			let change_output = if change_output_amount < change_script.minimal_non_dust() {
+				log_debug!(self.logger, "Coin selection attempt did not yield change output");
+				None
+			} else {
+				Some(TxOut { script_pubkey: change_script, value: change_output_amount })
+			};
 
-		let mut confirmed_utxos = Vec::with_capacity(selected_utxos.len());
-		for (utxo, _) in selected_utxos {
-			let prevtx = self.source.get_prevtx(utxo.outpoint).await?;
-			let prevtx_id = prevtx.compute_txid();
-			if prevtx_id != utxo.outpoint.txid
-				|| prevtx.output.get(utxo.outpoint.vout as usize).is_none()
-			{
-				log_error!(
-					self.logger,
-					"Tx {} from wallet source doesn't contain output referenced by outpoint: {}",
-					prevtx_id,
-					utxo.outpoint,
-				);
-				return Err(());
+			let mut confirmed_utxos = Vec::with_capacity(selected_utxos.len());
+			for (utxo, _) in selected_utxos {
+				let prevtx = self.source.get_prevtx(utxo.outpoint).await?;
+				let prevtx_id = prevtx.compute_txid();
+				if prevtx_id != utxo.outpoint.txid
+					|| prevtx.output.get(utxo.outpoint.vout as usize).is_none()
+				{
+					log_error!(
+						self.logger,
+						"Tx {} from wallet source doesn't contain output referenced by outpoint: {}",
+						prevtx_id,
+						utxo.outpoint,
+					);
+					return Err(());
+				}
+
+				confirmed_utxos.push(ConfirmedUtxo { utxo, prevtx });
 			}
 
-			confirmed_utxos.push(ConfirmedUtxo { utxo, prevtx });
+			Ok(CoinSelection { confirmed_utxos, change_output })
 		}
-
-		Ok(CoinSelection { confirmed_utxos, change_output })
+		.await;
+		if selection.is_err() {
+			let mut locked_utxos = self.locked_utxos.lock().unwrap();
+			for (outpoint, prev) in prev_locks {
+				match prev {
+					Some(prev_claim_id) => locked_utxos.insert(outpoint, prev_claim_id),
+					None => locked_utxos.remove(&outpoint),
+				};
+			}
+		}
+		selection
 	}
 }
 
@@ -794,7 +834,7 @@ where
 		&'a self, outpoint: OutPoint,
 	) -> impl Future<Output = Result<Transaction, ()>> + MaybeSend + 'a {
 		let prevtx = self.0.get_prevtx(outpoint);
-		Box::pin(async move { prevtx })
+		async move { prevtx }
 	}
 
 	fn get_change_script<'a>(
@@ -816,7 +856,17 @@ where
 /// UTXOs that would avoid conflicting double spends. If not enough UTXOs are available to do so,
 /// conflicting double spends may happen.
 ///
+/// Reservations of selected UTXOs are tracked in memory, so this wrapper is intended for a
+/// [`WalletSourceSync`] whose wallet does not reserve UTXOs on its own. Wallets that do should
+/// implement [`CoinSelectionSourceSync`] directly.
+///
+/// Selected UTXOs stay reserved until released with [`WalletSync::release_utxos`]. This must be
+/// called with the inputs from [`Event::DiscardFunding`] once a transaction funded through this
+/// wallet will no longer be broadcast, such as when a splice negotiation fails.
+///
 /// For an asynchronous version of this wrapper, see [`Wallet`].
+///
+/// [`Event::DiscardFunding`]: crate::events::Event::DiscardFunding
 // Note that updates to documentation on this struct should be copied to the asynchronous version.
 pub struct WalletSync<W: Deref + MaybeSync + MaybeSend, L: Logger + MaybeSync + MaybeSend>
 where
@@ -833,6 +883,16 @@ where
 	pub fn new(source: W, logger: L) -> Self {
 		Self { wallet: Wallet::new(WalletSourceSyncWrapper(source), logger) }
 	}
+
+	/// Releases the given UTXOs so they may be selected again.
+	///
+	/// Call this with the inputs from [`Event::DiscardFunding`] once a transaction funded through
+	/// this wallet will no longer be broadcast, such as when a splice negotiation fails.
+	///
+	/// [`Event::DiscardFunding`]: crate::events::Event::DiscardFunding
+	pub fn release_utxos(&self, outpoints: &[OutPoint]) {
+		self.wallet.release_utxos(outpoints)
+	}
 }
 
 impl<W: Deref + MaybeSync + MaybeSend, L: Logger + MaybeSync + MaybeSend> CoinSelectionSourceSync
@@ -977,3 +1037,99 @@ impl<T: CoinSelectionSourceSync> CoinSelectionSource for CoinSelectionSourceSync
 		async move { psbt }
 	}
 }
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use crate::util::test_utils::{TestLogger, TestWalletSource};
+
+	use bitcoin::absolute::LockTime;
+	use bitcoin::secp256k1::SecretKey;
+	use bitcoin::transaction::Version;
+	use bitcoin::TxIn;
+	use core::sync::atomic::{AtomicBool, Ordering};
+
+	/// A wallet source whose `get_prevtx` fails while `fail_prevtx` is set.
+	struct FailingPrevtxSource {
+		inner: TestWalletSource,
+		fail_prevtx: AtomicBool,
+	}
+
+	impl WalletSourceSync for FailingPrevtxSource {
+		fn list_confirmed_utxos(&self) -> Result<Vec<Utxo>, ()> {
+			self.inner.list_confirmed_utxos()
+		}
+		fn get_prevtx(&self, outpoint: OutPoint) -> Result<Transaction, ()> {
+			if self.fail_prevtx.load(Ordering::Acquire) {
+				return Err(());
+			}
+			self.inner.get_prevtx(outpoint)
+		}
+		fn get_change_script(&self) -> Result<ScriptBuf, ()> {
+			self.inner.get_change_script()
+		}
+		fn sign_psbt(&self, psbt: Psbt) -> Result<Transaction, ()> {
+			self.inner.sign_psbt(psbt)
+		}
+	}
+
+	/// Returns a source holding a single UTXO, along with that UTXO's outpoint.
+	fn single_utxo_source() -> (FailingPrevtxSource, OutPoint) {
+		let inner = TestWalletSource::new(SecretKey::from_slice(&[1; 32]).unwrap());
+		let prevtx = Transaction {
+			version: Version::TWO,
+			lock_time: LockTime::ZERO,
+			input: vec![TxIn::default()],
+			output: vec![TxOut {
+				value: Amount::from_sat(100_000),
+				script_pubkey: inner.get_change_script().unwrap(),
+			}],
+		};
+		let outpoint = OutPoint { txid: prevtx.compute_txid(), vout: 0 };
+		inner.add_utxo(prevtx, 0);
+		(FailingPrevtxSource { inner, fail_prevtx: AtomicBool::new(false) }, outpoint)
+	}
+
+	fn select(
+		wallet: &WalletSync<&FailingPrevtxSource, &TestLogger>, claim_id: Option<ClaimId>,
+	) -> Result<CoinSelection, ()> {
+		let must_pay_to = [TxOut {
+			value: Amount::from_sat(50_000),
+			script_pubkey: wallet.wallet.source.0.get_change_script().unwrap(),
+		}];
+		wallet.select_confirmed_utxos(claim_id, Vec::new(), &must_pay_to, 253, u64::MAX)
+	}
+
+	#[test]
+	fn failed_selection_releases_newly_locked_utxos() {
+		let (source, outpoint) = single_utxo_source();
+		let logger = TestLogger::new();
+		let wallet = WalletSync::new(&source, &logger);
+
+		source.fail_prevtx.store(true, Ordering::Release);
+		assert!(select(&wallet, None).is_err());
+
+		source.fail_prevtx.store(false, Ordering::Release);
+		let selection = select(&wallet, None).unwrap();
+		assert_eq!(selection.confirmed_utxos[0].outpoint(), outpoint);
+	}
+
+	#[test]
+	fn failed_forced_selection_restores_previous_claim() {
+		let (source, outpoint) = single_utxo_source();
+		let logger = TestLogger::new();
+		let wallet = WalletSync::new(&source, &logger);
+		let claim_a = Some(ClaimId([1; 32]));
+		let claim_b = Some(ClaimId([2; 32]));
+
+		select(&wallet, claim_a).unwrap();
+
+		// With the only UTXO locked to claim A, claim B reaches it only by forcing a conflicting
+		// spend, which then fails at `get_prevtx`.
+		source.fail_prevtx.store(true, Ordering::Release);
+		assert!(select(&wallet, claim_b).is_err());
+
+		let locked_utxos = wallet.wallet.locked_utxos.lock().unwrap();
+		assert_eq!(locked_utxos.get(&outpoint), Some(&claim_a));
+	}
+}

### pending_changelog/wallet-release-utxos.txt
@@ -0,0 +1,6 @@
+# API Updates
+
+ * `Wallet` and `WalletSync` keep every UTXO they select reserved until it is released with the
+   new `Wallet::release_utxos` and `WalletSync::release_utxos`. Users of these wrappers must call
+   them with the inputs from `Event::DiscardFunding` once a transaction funded through the wallet
+   will no longer be broadcast, such as when a splice negotiation fails.