Add RBF replacement tracking with new persisted lookup table

Introduce a new lookup `ReplacedTransactionStore` that maps old/replaced
transaction IDs to their current replacement transaction IDs, enabling
reliable tracking of replaced transactions throughout the replacement chain.

Key changes:
- Add persisted storage for RBF replacement relationships
- Link transactions in replacement trees using payment IDs
- Remove entire replacement chains from persistence when any transaction
  in the tree is confirmed

Author: Camillarhi

src/builder.rs

@@ -59,6 +59,7 @@ use crate::io::utils::{
 use crate::io::vss_store::VssStoreBuilder;
 use crate::io::{
 	self, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
+	REPLACED_TX_PERSISTENCE_PRIMARY_NAMESPACE, REPLACED_TX_PERSISTENCE_SECONDARY_NAMESPACE,
 };
 use crate::liquidity::{
 	LSPS1ClientConfig, LSPS2ClientConfig, LSPS2ServiceConfig, LiquiditySourceBuilder,
@@ -71,7 +72,8 @@ use crate::runtime::Runtime;
 use crate::tx_broadcaster::TransactionBroadcaster;
 use crate::types::{
 	ChainMonitor, ChannelManager, DynStore, DynStoreWrapper, GossipSync, Graph, KeysManager,
-	MessageRouter, OnionMessenger, PaymentStore, PeerManager, Persister, SyncAndAsyncKVStore,
+	MessageRouter, OnionMessenger, PaymentStore, PeerManager, Persister, ReplacedTransactionStore,
+	SyncAndAsyncKVStore,
 };
 use crate::wallet::persist::KVStoreWalletPersister;
 use crate::wallet::Wallet;
@@ -1230,6 +1232,21 @@ fn build_with_store_internal(
 		},
 	};
 
+	let replaced_tx_store =
+		match io::utils::read_replaced_txs(Arc::clone(&kv_store), Arc::clone(&logger)) {
+			Ok(replaced_txs) => Arc::new(ReplacedTransactionStore::new(
+				replaced_txs,
+				REPLACED_TX_PERSISTENCE_PRIMARY_NAMESPACE.to_string(),
+				REPLACED_TX_PERSISTENCE_SECONDARY_NAMESPACE.to_string(),
+				Arc::clone(&kv_store),
+				Arc::clone(&logger),
+			)),
+			Err(e) => {
+				log_error!(logger, "Failed to read replaced transaction data from store: {}", e);
+				return Err(BuildError::ReadFailed);
+			},
+		};
+
 	let wallet = Arc::new(Wallet::new(
 		bdk_wallet,
 		wallet_persister,
@@ -1238,6 +1255,7 @@ fn build_with_store_internal(
 		Arc::clone(&payment_store),
 		Arc::clone(&config),
 		Arc::clone(&logger),
+		Arc::clone(&replaced_tx_store),
 	));
 
 	// Initialize the KeysManager

src/io/mod.rs

@@ -78,3 +78,7 @@ pub(crate) const BDK_WALLET_INDEXER_KEY: &str = "indexer";
 ///
 /// [`StaticInvoice`]: lightning::offers::static_invoice::StaticInvoice
 pub(crate) const STATIC_INVOICE_STORE_PRIMARY_NAMESPACE: &str = "static_invoices";
+
+/// The replaced transaction information will be persisted under this prefix.
+pub(crate) const REPLACED_TX_PERSISTENCE_PRIMARY_NAMESPACE: &str = "replaced_txs";
+pub(crate) const REPLACED_TX_PERSISTENCE_SECONDARY_NAMESPACE: &str = "";

src/io/utils.rs

@@ -45,6 +45,7 @@ use crate::io::{
 	NODE_METRICS_KEY, NODE_METRICS_PRIMARY_NAMESPACE, NODE_METRICS_SECONDARY_NAMESPACE,
 };
 use crate::logger::{log_error, LdkLogger, Logger};
+use crate::payment::ReplacedOnchainTransactionDetails;
 use crate::peer_store::PeerStore;
 use crate::types::{Broadcaster, DynStore, KeysManager, Sweeper};
 use crate::wallet::ser::{ChangeSetDeserWrapper, ChangeSetSerWrapper};
@@ -576,6 +577,38 @@ pub(crate) fn read_bdk_wallet_change_set(
 	Ok(Some(change_set))
 }
 
+/// Read previously persisted replaced transaction information from the store.
+pub(crate) fn read_replaced_txs<L: Deref>(
+	kv_store: Arc<DynStore>, logger: L,
+) -> Result<Vec<ReplacedOnchainTransactionDetails>, std::io::Error>
+where
+	L::Target: LdkLogger,
+{
+	let mut res = Vec::new();
+
+	for stored_key in KVStoreSync::list(
+		&*kv_store,
+		REPLACED_TX_PERSISTENCE_PRIMARY_NAMESPACE,
+		REPLACED_TX_PERSISTENCE_SECONDARY_NAMESPACE,
+	)? {
+		let mut reader = Cursor::new(KVStoreSync::read(
+			&*kv_store,
+			REPLACED_TX_PERSISTENCE_PRIMARY_NAMESPACE,
+			REPLACED_TX_PERSISTENCE_SECONDARY_NAMESPACE,
+			&stored_key,
+		)?);
+		let payment = ReplacedOnchainTransactionDetails::read(&mut reader).map_err(|e| {
+			log_error!(logger, "Failed to deserialize ReplacedOnchainTransactionDetails: {}", e);
+			std::io::Error::new(
+				std::io::ErrorKind::InvalidData,
+				"Failed to deserialize ReplacedOnchainTransactionDetails",
+			)
+		})?;
+		res.push(payment);
+	}
+	Ok(res)
+}
+
 #[cfg(test)]
 mod tests {
 	use super::read_or_generate_seed_file;

src/payment/mod.rs

@@ -11,13 +11,15 @@ pub(crate) mod asynchronous;
 mod bolt11;
 mod bolt12;
 mod onchain;
+mod replaced_transaction_store;
 mod spontaneous;
 pub(crate) mod store;
 mod unified_qr;
 
 pub use bolt11::Bolt11Payment;
 pub use bolt12::Bolt12Payment;
 pub use onchain::OnchainPayment;
+pub use replaced_transaction_store::ReplacedOnchainTransactionDetails;
 pub use spontaneous::SpontaneousPayment;
 pub use store::{
 	ConfirmationStatus, LSPFeeLimits, PaymentDetails, PaymentDirection, PaymentKind, PaymentStatus,

src/payment/replaced_transaction_store.rs

@@ -0,0 +1,104 @@
+// This file is Copyright its original authors, visible in version control history.
+//
+// This file is licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license <LICENSE-MIT or
+// http://opensource.org/licenses/MIT>, at your option. You may not use this file except in
+// accordance with one or both of these licenses.
+
+use bitcoin::Txid;
+use lightning::ln::channelmanager::PaymentId;
+use lightning::ln::msgs::DecodeError;
+use lightning::util::ser::{Readable, Writeable};
+use lightning::{_init_and_read_len_prefixed_tlv_fields, write_tlv_fields};
+
+use crate::data_store::{StorableObject, StorableObjectId, StorableObjectUpdate};
+
+/// Details of an on-chain transaction that has replaced a previous transaction (e.g., via RBF).
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct ReplacedOnchainTransactionDetails {
+	/// The new transaction ID.
+	pub new_txid: Txid,
+	/// The original transaction ID that was replaced.
+	pub original_txid: Txid,
+	/// The payment ID associated with the transaction.
+	pub payment_id: PaymentId,
+}
+
+impl ReplacedOnchainTransactionDetails {
+	pub(crate) fn new(new_txid: Txid, original_txid: Txid, payment_id: PaymentId) -> Self {
+		Self { new_txid, original_txid, payment_id }
+	}
+}
+
+impl Writeable for ReplacedOnchainTransactionDetails {
+	fn write<W: lightning::util::ser::Writer>(
+		&self, writer: &mut W,
+	) -> Result<(), lightning::io::Error> {
+		write_tlv_fields!(writer, {
+			(0, self.new_txid, required),
+			(2, self.original_txid, required),
+			(4, self.payment_id, required),
+		});
+		Ok(())
+	}
+}
+
+impl Readable for ReplacedOnchainTransactionDetails {
+	fn read<R: lightning::io::Read>(
+		reader: &mut R,
+	) -> Result<ReplacedOnchainTransactionDetails, DecodeError> {
+		_init_and_read_len_prefixed_tlv_fields!(reader, {
+			(0, new_txid, required),
+			(2, original_txid, required),
+			(4, payment_id, required),
+		});
+
+		let new_txid: Txid = new_txid.0.ok_or(DecodeError::InvalidValue)?;
+		let original_txid: Txid = original_txid.0.ok_or(DecodeError::InvalidValue)?;
+		let payment_id: PaymentId = payment_id.0.ok_or(DecodeError::InvalidValue)?;
+
+		Ok(ReplacedOnchainTransactionDetails { new_txid, original_txid, payment_id })
+	}
+}
+
+impl StorableObjectId for Txid {
+	fn encode_to_hex_str(&self) -> String {
+		self.to_string()
+	}
+}
+impl StorableObject for ReplacedOnchainTransactionDetails {
+	type Id = Txid;
+	type Update = ReplacedOnchainTransactionDetailsUpdate;
+
+	fn id(&self) -> Self::Id {
+		self.new_txid
+	}
+
+	fn update(&mut self, _update: &Self::Update) -> bool {
+		// We don't update, we delete on confirmation
+		false
+	}
+
+	fn to_update(&self) -> Self::Update {
+		self.into()
+	}
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub(crate) struct ReplacedOnchainTransactionDetailsUpdate {
+	pub id: Txid,
+}
+
+impl From<&ReplacedOnchainTransactionDetails> for ReplacedOnchainTransactionDetailsUpdate {
+	fn from(value: &ReplacedOnchainTransactionDetails) -> Self {
+		Self { id: value.new_txid }
+	}
+}
+
+impl StorableObjectUpdate<ReplacedOnchainTransactionDetails>
+	for ReplacedOnchainTransactionDetailsUpdate
+{
+	fn id(&self) -> <ReplacedOnchainTransactionDetails as StorableObject>::Id {
+		self.id
+	}
+}

src/types.rs

@@ -37,7 +37,7 @@ use crate::fee_estimator::OnchainFeeEstimator;
 use crate::gossip::RuntimeSpawner;
 use crate::logger::Logger;
 use crate::message_handler::NodeCustomMessageHandler;
-use crate::payment::PaymentDetails;
+use crate::payment::{PaymentDetails, ReplacedOnchainTransactionDetails};
 
 /// A supertrait that requires that a type implements both [`KVStore`] and [`KVStoreSync`] at the
 /// same time.
@@ -596,3 +596,6 @@ impl From<&(u64, Vec<u8>)> for CustomTlvRecord {
 		CustomTlvRecord { type_num: tlv.0, value: tlv.1.clone() }
 	}
 }
+
+pub(crate) type ReplacedTransactionStore =
+	DataStore<ReplacedOnchainTransactionDetails, Arc<Logger>>;

src/wallet/mod.rs

@@ -50,8 +50,10 @@ use crate::config::Config;
 use crate::fee_estimator::{ConfirmationTarget, FeeEstimator, OnchainFeeEstimator};
 use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger};
 use crate::payment::store::ConfirmationStatus;
-use crate::payment::{PaymentDetails, PaymentDirection, PaymentKind, PaymentStatus};
-use crate::types::{Broadcaster, PaymentStore};
+use crate::payment::{
+	PaymentDetails, PaymentDirection, PaymentKind, PaymentStatus, ReplacedOnchainTransactionDetails,
+};
+use crate::types::{Broadcaster, PaymentStore, ReplacedTransactionStore};
 use crate::Error;
 
 pub(crate) enum OnchainSendAmount {
@@ -72,18 +74,28 @@ pub(crate) struct Wallet {
 	payment_store: Arc<PaymentStore>,
 	config: Arc<Config>,
 	logger: Arc<Logger>,
+	replaced_tx_store: Arc<ReplacedTransactionStore>,
 }
 
 impl Wallet {
 	pub(crate) fn new(
 		wallet: bdk_wallet::PersistedWallet<KVStoreWalletPersister>,
 		wallet_persister: KVStoreWalletPersister, broadcaster: Arc<Broadcaster>,
 		fee_estimator: Arc<OnchainFeeEstimator>, payment_store: Arc<PaymentStore>,
-		config: Arc<Config>, logger: Arc<Logger>,
+		config: Arc<Config>, logger: Arc<Logger>, replaced_tx_store: Arc<ReplacedTransactionStore>,
 	) -> Self {
 		let inner = Mutex::new(wallet);
 		let persister = Mutex::new(wallet_persister);
-		Self { inner, persister, broadcaster, fee_estimator, payment_store, config, logger }
+		Self {
+			inner,
+			persister,
+			broadcaster,
+			fee_estimator,
+			payment_store,
+			config,
+			logger,
+			replaced_tx_store,
+		}
 	}
 
 	pub(crate) fn get_full_scan_request(&self) -> FullScanRequest<KeychainKind> {
@@ -229,9 +241,21 @@ impl Wallet {
 							..
 						} = payment.kind
 						{
+							let payment_id= payment.id;
 							if new_tip.height >= height + ANTI_REORG_DELAY - 1 {
 								payment.status = PaymentStatus::Succeeded;
 								self.payment_store.insert_or_update(payment)?;
+
+								// Remove any replaced transactions associated with this payment
+								let replaced_txids = self
+									.replaced_tx_store
+									.list_filter(|r| r.payment_id == payment_id)
+									.iter()
+									.map(|p| p.new_txid)
+									.collect::<Vec<Txid>>();
+								for replaced_txid in replaced_txids {
+									self.replaced_tx_store.remove(&replaced_txid)?;
+								}
 							}
 						}
 					}
@@ -252,47 +276,24 @@ impl Wallet {
 					);
 					self.payment_store.insert_or_update(payment)?;
 				},
-				WalletEvent::TxReplaced { txid, tx, conflicts } => {
+				WalletEvent::TxReplaced { txid, conflicts, .. } => {
 					let payment_id = self
 						.find_payment_by_txid(*txid)
 						.unwrap_or_else(|| PaymentId(txid.to_byte_array()));
 
-					if let Some(mut payment) = self.payment_store.get(&payment_id) {
-						if let PaymentKind::Onchain {
-							ref mut conflicting_txids,
-							txid: current_txid,
-							..
-						} = payment.kind
-						{
-							let existing_set: std::collections::HashSet<_> =
-								conflicting_txids.iter().collect();
-
-							let new_conflicts: Vec<_> = conflicts
-								.iter()
-								.map(|(_, conflict_txid)| *conflict_txid)
-								.filter(|conflict_txid| {
-									*conflict_txid != current_txid
-										&& !existing_set.contains(conflict_txid)
-								})
-								.collect();
-
-							conflicting_txids.extend(new_conflicts);
-						}
-						self.payment_store.insert_or_update(payment)?;
-					} else {
-						let conflicting_txids =
-							Some(conflicts.iter().map(|(_, txid)| *txid).collect());
+					// Collect all conflict txids
+					let conflict_txids: Vec<Txid> =
+						conflicts.iter().map(|(_, conflict_txid)| *conflict_txid).collect();
 
-						let payment = self.create_payment_from_tx(
-							locked_wallet,
+					for conflict_txid in conflict_txids {
+						// Update the replaced transaction store
+						let replaced_tx_details = ReplacedOnchainTransactionDetails::new(
+							conflict_txid,
 							*txid,
 							payment_id,
-							tx,
-							PaymentStatus::Pending,
-							ConfirmationStatus::Unconfirmed,
-							conflicting_txids,
 						);
-						self.payment_store.insert_or_update(payment)?;
+
+						self.replaced_tx_store.insert_or_update(replaced_tx_details)?;
 					}
 				},
 				WalletEvent::TxDropped { txid, tx } => {
@@ -962,16 +963,12 @@ impl Wallet {
 			return Some(direct_payment_id);
 		}
 
-		self.payment_store
-			.list_filter(|p| {
-				if let PaymentKind::Onchain { txid, conflicting_txids, .. } = &p.kind {
-					*txid == target_txid || conflicting_txids.contains(&target_txid)
-				} else {
-					false
-				}
-			})
-			.first()
-			.map(|p| p.id)
+		// Check if this txid is a replaced transaction
+		if let Some(replaced_details) = self.replaced_tx_store.get(&target_txid) {
+			return Some(replaced_details.payment_id);
+		}
+
+		None
 	}
 }