Enhance onchain transaction management

bindings/ldk_node.udl

@@ -276,6 +276,8 @@ interface OnchainPayment {
 	Txid send_to_address([ByRef]Address address, u64 amount_sats, FeeRate? fee_rate);
 	[Throws=NodeError]
 	Txid send_all_to_address([ByRef]Address address, boolean retain_reserve, FeeRate? fee_rate);
+	[Throws=NodeError]
+	Txid bump_fee_rbf(PaymentId payment_id);
 };
 
 interface FeeRate {

src/payment/onchain.rs

@@ -10,6 +10,7 @@
 use std::sync::{Arc, RwLock};
 
 use bitcoin::{Address, Txid};
+use lightning::ln::channelmanager::PaymentId;
 
 use crate::config::Config;
 use crate::error::Error;
@@ -120,4 +121,18 @@ impl OnchainPayment {
 		let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate);
 		self.wallet.send_to_address(address, send_amount, fee_rate_opt)
 	}
+
+	/// Attempt to bump the fee of an unconfirmed transaction using Replace-by-Fee (RBF).
+	///
+	/// This creates a new transaction that replaces the original one, increasing the fee by the
+	/// specified increment to improve its chances of confirmation. The original transaction must
+	/// be signaling RBF replaceability for this to succeed.
+	///
+	/// The new transaction will have the same outputs as the original but with a
+	/// higher fee, resulting in faster confirmation potential.
+	///
+	/// Returns the Txid of the new replacement transaction if successful.
+	pub fn bump_fee_rbf(&self, payment_id: PaymentId) -> Result<Txid, Error> {
+		self.wallet.bump_fee_rbf(payment_id)
+	}
 }

src/payment/pending_payment_store.rs

@@ -84,10 +84,11 @@ impl StorableObjectUpdate<PendingPaymentDetails> for PendingPaymentDetailsUpdate
 
 impl From<&PendingPaymentDetails> for PendingPaymentDetailsUpdate {
 	fn from(value: &PendingPaymentDetails) -> Self {
-		Self {
-			id: value.id(),
-			payment_update: Some(value.details.to_update()),
-			conflicting_txids: Some(value.conflicting_txids.clone()),
-		}
+		let conflicting_txids = if value.conflicting_txids.is_empty() {
+			None
+		} else {
+			Some(value.conflicting_txids.clone())
+		};
+		Self { id: value.id(), payment_update: Some(value.details.to_update()), conflicting_txids }
 	}
 }

src/payment/store.rs

@@ -291,6 +291,15 @@ impl StorableObject for PaymentDetails {
 			}
 		}
 
+		if let Some(tx_id) = update.txid {
+			match self.kind {
+				PaymentKind::Onchain { ref mut txid, .. } => {
+					update_if_necessary!(*txid, tx_id);
+				},
+				_ => {},
+			}
+		}
+
 		if updated {
 			self.latest_update_timestamp = SystemTime::now()
 				.duration_since(UNIX_EPOCH)
@@ -540,6 +549,7 @@ pub(crate) struct PaymentDetailsUpdate {
 	pub direction: Option<PaymentDirection>,
 	pub status: Option<PaymentStatus>,
 	pub confirmation_status: Option<ConfirmationStatus>,
+	pub txid: Option<Txid>,
 }
 
 impl PaymentDetailsUpdate {
@@ -555,6 +565,7 @@ impl PaymentDetailsUpdate {
 			direction: None,
 			status: None,
 			confirmation_status: None,
+			txid: None,
 		}
 	}
 }
@@ -570,9 +581,9 @@ impl From<&PaymentDetails> for PaymentDetailsUpdate {
 			_ => (None, None, None),
 		};
 
-		let confirmation_status = match value.kind {
-			PaymentKind::Onchain { status, .. } => Some(status),
-			_ => None,
+		let (confirmation_status, txid) = match &value.kind {
+			PaymentKind::Onchain { status, txid, .. } => (Some(*status), Some(*txid)),
+			_ => (None, None),
 		};
 
 		let counterparty_skimmed_fee_msat = match value.kind {
@@ -593,6 +604,7 @@ impl From<&PaymentDetails> for PaymentDetailsUpdate {
 			direction: Some(value.direction),
 			status: Some(value.status),
 			confirmation_status,
+			txid,
 		}
 	}
 }

src/wallet/mod.rs

@@ -12,6 +12,7 @@ use std::sync::{Arc, Mutex};
 
 use bdk_chain::spk_client::{FullScanRequest, SyncRequest};
 use bdk_wallet::descriptor::ExtendedDescriptor;
+use bdk_wallet::error::{BuildFeeBumpError, CreateTxError};
 use bdk_wallet::event::WalletEvent;
 #[allow(deprecated)]
 use bdk_wallet::SignOptions;
@@ -29,7 +30,10 @@ use bitcoin::{
 	Address, Amount, FeeRate, OutPoint, ScriptBuf, Transaction, TxOut, Txid, WPubkeyHash, Weight,
 	WitnessProgram, WitnessVersion,
 };
-use lightning::chain::chaininterface::BroadcasterInterface;
+
+use lightning::chain::chaininterface::{
+	BroadcasterInterface, INCREMENTAL_RELAY_FEE_SAT_PER_1000_WEIGHT,
+};
 use lightning::chain::channelmonitor::ANTI_REORG_DELAY;
 use lightning::chain::{BestBlock, Listen};
 use lightning::events::bump_transaction::{Input, Utxo, WalletSource};
@@ -244,31 +248,54 @@ impl Wallet {
 					self.pending_payment_store.insert_or_update(pending_payment)?;
 				},
 				WalletEvent::ChainTipChanged { new_tip, .. } => {
-					// Get all payments that are Pending with Confirmed status
+					// Get all on-chain payments that are Pending
 					let pending_payments: Vec<PendingPaymentDetails> =
 						self.pending_payment_store.list_filter(|p| {
 							p.details.status == PaymentStatus::Pending
-								&& matches!(
-									p.details.kind,
-									PaymentKind::Onchain {
-										status: ConfirmationStatus::Confirmed { .. },
-										..
-									}
-								)
+								&& matches!(p.details.kind, PaymentKind::Onchain { .. })
 						});
 
+					let mut unconfirmed_outbound_txids: Vec<Txid> = Vec::new();
+
 					for mut payment in pending_payments {
-						if let PaymentKind::Onchain {
-							status: ConfirmationStatus::Confirmed { height, .. },
-							..
-						} = payment.details.kind
-						{
-							let payment_id = payment.details.id;
-							if new_tip.height >= height + ANTI_REORG_DELAY - 1 {
-								payment.details.status = PaymentStatus::Succeeded;
-								self.payment_store.insert_or_update(payment.details)?;
-								self.pending_payment_store.remove(&payment_id)?;
-							}
+						match payment.details.kind {
+							PaymentKind::Onchain {
+								status: ConfirmationStatus::Confirmed { height, .. },
+								..
+							} => {
+								let payment_id = payment.details.id;
+								if new_tip.height >= height + ANTI_REORG_DELAY - 1 {
+									payment.details.status = PaymentStatus::Succeeded;
+									self.payment_store.insert_or_update(payment.details)?;
+									self.pending_payment_store.remove(&payment_id)?;
+								}
+							},
+							PaymentKind::Onchain {
+								txid,
+								status: ConfirmationStatus::Unconfirmed,
+							} if payment.details.direction == PaymentDirection::Outbound => {
+								unconfirmed_outbound_txids.push(txid);
+							},
+							_ => {},
+						}
+					}
+
+					if !unconfirmed_outbound_txids.is_empty() {
+						let txs_to_broadcast: Vec<Transaction> = unconfirmed_outbound_txids
+							.iter()
+							.filter_map(|txid| {
+								locked_wallet.tx_details(*txid).map(|d| (*d.tx).clone())
+							})
+							.collect();
+
+						if !txs_to_broadcast.is_empty() {
+							let tx_refs: Vec<&Transaction> = txs_to_broadcast.iter().collect();
+							self.broadcaster.broadcast_transactions(&tx_refs);
+							log_info!(
+								self.logger,
+								"Rebroadcast {} unconfirmed transactions on chain tip change",
+								txs_to_broadcast.len()
+							);
 						}
 					}
 				},
@@ -290,7 +317,7 @@ impl Wallet {
 					self.payment_store.insert_or_update(payment)?;
 					self.pending_payment_store.insert_or_update(pending_payment)?;
 				},
-				WalletEvent::TxReplaced { txid, conflicts, tx, .. } => {
+				WalletEvent::TxReplaced { txid, conflicts, .. } => {
 					let payment_id = self
 						.find_payment_by_txid(txid)
 						.unwrap_or_else(|| PaymentId(txid.to_byte_array()));
@@ -299,14 +326,9 @@ impl Wallet {
 					let conflict_txids: Vec<Txid> =
 						conflicts.iter().map(|(_, conflict_txid)| *conflict_txid).collect();
 
-					let payment = self.create_payment_from_tx(
-						locked_wallet,
-						txid,
-						payment_id,
-						&tx,
-						PaymentStatus::Pending,
-						ConfirmationStatus::Unconfirmed,
-					);
+					// We fetch payment details here since the replacement has updated the stored state
+					let payment =
+						self.payment_store.get(&payment_id).ok_or(Error::InvalidPaymentId)?;
 					let pending_payment_details = self
 						.create_pending_payment_from_tx(payment.clone(), conflict_txids.clone());
 
@@ -978,6 +1000,152 @@ impl Wallet {
 
 		None
 	}
+
+	#[allow(deprecated)]
+	pub(crate) fn bump_fee_rbf(&self, payment_id: PaymentId) -> Result<Txid, Error> {
+		let payment = self.payment_store.get(&payment_id).ok_or(Error::InvalidPaymentId)?;
+
+		let mut locked_wallet = self.inner.lock().unwrap();
+
+		if let PaymentKind::Onchain { status, .. } = &payment.kind {
+			match status {
+				ConfirmationStatus::Confirmed { .. } => {
+					log_error!(
+						self.logger,
+						"Transaction {} is already confirmed and cannot be fee bumped",
+						payment_id
+					);
+					return Err(Error::InvalidPaymentId);
+				},
+				ConfirmationStatus::Unconfirmed => {},
+			}
+		}
+
+		if payment.direction != PaymentDirection::Outbound {
+			log_error!(self.logger, "Transaction {} is not an outbound payment", payment_id);
+			return Err(Error::InvalidPaymentId);
+		}
+
+		let txid = match &payment.kind {
+			PaymentKind::Onchain { txid, .. } => *txid,
+			_ => return Err(Error::InvalidPaymentId),
+		};
+
+		let old_tx =
+			locked_wallet.tx_details(txid).ok_or(Error::InvalidPaymentId)?.tx.deref().clone();
+
+		let old_fee_rate = locked_wallet.calculate_fee_rate(&old_tx).map_err(|e| {
+			log_error!(self.logger, "Failed to calculate fee rate of transaction {}: {}", txid, e);
+			Error::InvalidPaymentId
+		})?;
+		let old_fee_rate_sat_per_kwu = old_fee_rate.to_sat_per_kwu();
+
+		// BIP 125 requires the replacement to pay a higher fee rate than the original.
+		// The minimum increase is the incremental relay fee.
+		let min_required_fee_rate_sat_per_kwu =
+			old_fee_rate_sat_per_kwu + INCREMENTAL_RELAY_FEE_SAT_PER_1000_WEIGHT as u64;
+
+		let confirmation_target = ConfirmationTarget::OnchainPayment;
+		let estimated_fee_rate = self.fee_estimator.estimate_fee_rate(confirmation_target);
+
+		// Use the higher of minimum RBF requirement or current network estimate
+		let final_fee_rate_sat_per_kwu =
+			min_required_fee_rate_sat_per_kwu.max(estimated_fee_rate.to_sat_per_kwu());
+		let final_fee_rate = FeeRate::from_sat_per_kwu(final_fee_rate_sat_per_kwu);
+
+		let mut psbt = {
+			let mut builder = locked_wallet.build_fee_bump(txid).map_err(|e| {
+				log_error!(self.logger, "BDK fee bump failed for {}: {:?}", txid, e);
+				match e {
+					BuildFeeBumpError::TransactionNotFound(_) => Error::InvalidPaymentId,
+					BuildFeeBumpError::TransactionConfirmed(_) => {
+						log_error!(self.logger, "Payment {} is already confirmed", payment_id);
+						Error::InvalidPaymentId
+					},
+					BuildFeeBumpError::IrreplaceableTransaction(_) => {
+						Error::OnchainTxCreationFailed
+					},
+					BuildFeeBumpError::FeeRateUnavailable => Error::FeerateEstimationUpdateFailed,
+					BuildFeeBumpError::UnknownUtxo(_) => Error::OnchainTxCreationFailed,
+					BuildFeeBumpError::InvalidOutputIndex(_) => Error::OnchainTxCreationFailed,
+				}
+			})?;
+
+			builder.fee_rate(final_fee_rate);
+
+			match builder.finish() {
+				Ok(psbt) => Ok(psbt),
+				Err(CreateTxError::FeeRateTooLow { required: required_fee_rate }) => {
+					log_info!(self.logger, "BDK requires higher fee rate: {}", required_fee_rate);
+
+					let mut builder = locked_wallet.build_fee_bump(txid).map_err(|e| {
+						log_error!(self.logger, "BDK fee bump retry failed for {}: {:?}", txid, e);
+						Error::InvalidFeeRate
+					})?;
+
+					builder.fee_rate(required_fee_rate);
+					builder.finish().map_err(|e| {
+						log_error!(
+							self.logger,
+							"Failed to finish PSBT with required fee rate: {:?}",
+							e
+						);
+						Error::InvalidFeeRate
+					})
+				},
+				Err(e) => {
+					log_error!(self.logger, "Failed to create fee bump PSBT: {:?}", e);
+					Err(Error::InvalidFeeRate)
+				},
+			}?
+		};
+
+		match locked_wallet.sign(&mut psbt, SignOptions::default()) {
+			Ok(finalized) => {
+				if !finalized {
+					return Err(Error::OnchainTxCreationFailed);
+				}
+			},
+			Err(err) => {
+				log_error!(self.logger, "Failed to create transaction: {}", err);
+				return Err(err.into());
+			},
+		}
+
+		let mut locked_persister = self.persister.lock().unwrap();
+		locked_wallet.persist(&mut locked_persister).map_err(|e| {
+			log_error!(self.logger, "Failed to persist wallet: {}", e);
+			Error::PersistenceFailed
+		})?;
+
+		let fee_bumped_tx = psbt.extract_tx().map_err(|e| {
+			log_error!(self.logger, "Failed to extract transaction: {}", e);
+			e
+		})?;
+
+		let new_txid = fee_bumped_tx.compute_txid();
+
+		self.broadcaster.broadcast_transactions(&[&fee_bumped_tx]);
+
+		let new_payment = self.create_payment_from_tx(
+			&locked_wallet,
+			new_txid,
+			payment.id,
+			&fee_bumped_tx,
+			PaymentStatus::Pending,
+			ConfirmationStatus::Unconfirmed,
+		);
+
+		let pending_payment_store =
+			self.create_pending_payment_from_tx(new_payment.clone(), Vec::new());
+
+		self.pending_payment_store.insert_or_update(pending_payment_store)?;
+		self.payment_store.insert_or_update(new_payment)?;
+
+		log_info!(self.logger, "RBF successful: replaced {} with {}", txid, new_txid);
+
+		Ok(new_txid)
+	}
 }
 
 impl Listen for Wallet {