Apply RBF replacements to the wallet before sync

src/wallet/mod.rs

@@ -539,8 +539,13 @@ impl Wallet {
 					// Collect all conflict txids
 					let mut conflict_txids: Vec<Txid> =
 						conflicts.iter().map(|(_, conflict_txid)| *conflict_txid).collect();
+					if let Some(previous) = self.pending_payment_store.get(&payment_id).await? {
+						conflict_txids.extend(previous.conflicting_txids);
+					}
 
 					conflict_txids.push(txid);
+					conflict_txids.sort_unstable();
+					conflict_txids.dedup();
 					// The payment already exists in the store at this point: `bump_fee_rbf`
 					// updates the payment store with the replacement txid before the next sync
 					// cycle, and an id resolved through the candidate history comes from a
@@ -2084,6 +2089,13 @@ impl Wallet {
 			},
 		};
 
+		let mut conflicts = self
+			.pending_payment_store
+			.get(&payment_id)
+			.await?
+			.map(|p| p.conflicting_txids)
+			.unwrap_or_default();
+
 		let mut locked_persister = self.persister.lock().await;
 		let mut locked_wallet = self.inner.lock().expect("lock");
 
@@ -2260,8 +2272,26 @@ impl Wallet {
 			ConfirmationStatus::Unconfirmed,
 		);
 
+		conflicts.push(txid);
+		conflicts.sort_unstable();
+		conflicts.dedup();
 		let pending_payment_store =
-			self.create_pending_payment_from_tx(new_payment.clone(), Vec::new());
+			self.create_pending_payment_from_tx(new_payment.clone(), conflicts);
+		let seen_at = std::time::SystemTime::now()
+			.duration_since(std::time::UNIX_EPOCH)
+			.unwrap_or_default()
+			.as_secs();
+		let previous_seen_at = locked_wallet
+			.tx_details(txid)
+			.and_then(|details| match details.chain_position {
+				bdk_chain::ChainPosition::Unconfirmed { last_seen, .. } => last_seen,
+				_ => None,
+			})
+			.unwrap_or(0);
+		locked_wallet.apply_unconfirmed_txs([(
+			fee_bumped_tx.clone(),
+			seen_at.max(previous_seen_at.saturating_add(1)),
+		)]);
 		let change_set = locked_wallet.take_staged().unwrap_or_default();
 		drop(locked_wallet);
 		locked_persister.persist_changeset(change_set).await.map_err(|e| {

tests/integration_tests_rust.rs

@@ -4730,6 +4730,43 @@ async fn fs_store_persistence_backwards_compatibility() {
 	node_new.stop().unwrap();
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+async fn onchain_fee_bump_rbf_twice_before_sync() {
+	let (bitcoind, electrsd) = setup_bitcoind_and_electrsd();
+	let chain_source = random_chain_source(&bitcoind, &electrsd);
+	let (node_a, node_b) = setup_two_nodes(&chain_source, false, false);
+
+	let addr_a = node_a.onchain_payment().new_address().unwrap();
+	let addr_b = node_b.onchain_payment().new_address().unwrap();
+	premine_and_distribute_funds(
+		&bitcoind.client,
+		&electrsd.client,
+		vec![addr_a.clone(), addr_b],
+		Amount::from_sat(500_000),
+	)
+	.await;
+	node_b.sync_wallets().unwrap();
+	let txid = node_b.onchain_payment().send_to_address(&addr_a, 100_000, None).unwrap();
+	wait_for_tx(&electrsd.client, txid).await;
+	node_b.sync_wallets().unwrap();
+	let payment_id = PaymentId(txid.to_byte_array());
+	let first_fee = node_b.payment(&payment_id).unwrap().unwrap().fee_paid_msat.unwrap();
+
+	let first_replacement = node_b.onchain_payment().bump_fee_rbf(payment_id, None).unwrap();
+	let second_replacement = node_b.onchain_payment().bump_fee_rbf(payment_id, None).unwrap();
+	assert_ne!(first_replacement, second_replacement);
+	let payment = node_b.payment(&payment_id).unwrap().unwrap();
+	assert!(payment.fee_paid_msat.unwrap() > first_fee);
+	assert!(
+		matches!(payment.kind, PaymentKind::Onchain { txid, .. } if txid == second_replacement)
+	);
+	wait_for_tx(&electrsd.client, second_replacement).await;
+	node_b.sync_wallets().unwrap();
+	generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await;
+	node_b.sync_wallets().unwrap();
+	assert_eq!(node_b.payment(&payment_id).unwrap().unwrap().status, PaymentStatus::Succeeded);
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 async fn onchain_fee_bump_rbf() {
 	let (bitcoind, electrsd) = setup_bitcoind_and_electrsd();