Apply RBF replacements to the wallet before sync

src/wallet/mod.rs

@@ -2260,17 +2260,48 @@ impl Wallet {
 			ConfirmationStatus::Unconfirmed,
 		);
 
-		let pending_payment_store =
-			self.create_pending_payment_from_tx(new_payment.clone(), Vec::new());
+		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);
+		// BDK keeps the conflicting tx with the latest last-seen and breaks ties by txid, so a bump
+		// in the same second as the replaced round must still be seen strictly after it.
+		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| {
 			log_error!(self.logger, "Failed to persist wallet after fee bump of {}: {}", txid, e);
 			Error::PersistenceFailed
 		})?;
 
+		// Wallet sync maps a replaced transaction to its payment with `find_payment_by_txid`. From
+		// the second bump on, the replaced `txid` matches only through the pending-store entry's
+		// `conflicting_txids`. A non-empty list in an update replaces the stored one, so extend it
+		// rather than writing just `txid`.
+		let mut conflicting_txids = self
+			.pending_payment_store
+			.get(&payment_id)
+			.await?
+			.map(|pending| pending.conflicting_txids)
+			.unwrap_or_default();
+		if !conflicting_txids.contains(&txid) {
+			conflicting_txids.push(txid);
+		}
+		let pending_payment =
+			self.create_pending_payment_from_tx(new_payment.clone(), conflicting_txids);
+
 		self.payment_store.insert_or_update(new_payment).await?;
-		self.pending_payment_store.insert_or_update(pending_payment_store).await?;
+		self.pending_payment_store.insert_or_update(pending_payment).await?;
 
 		self.broadcaster.broadcast_unclassified_transaction(fee_bumped_tx);
 
@@ -3972,6 +4003,100 @@ mod tests {
 		assert_eq!(wallet.find_payment_by_txid(txid2).await.unwrap(), Some(payment_id));
 	}
 
+	/// A second bump must add the replaced txid to the pending-store entry's `conflicting_txids`.
+	/// The payment id is derived from the first txid and the payment's txid is now the newest
+	/// one, so wallet sync's `TxReplaced` event for an earlier replacement resolves the payment
+	/// only through that list. Without it, the event finds no payment and is skipped.
+	#[allow(deprecated)]
+	#[tokio::test]
+	async fn bump_fee_rbf_keeps_replaced_txid_mapped() {
+		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(InMemoryStore::new()));
+		let wallet = new_test_wallet(store, false).await;
+
+		// A confirmed output funds the wallet...
+		{
+			let mut locked_wallet = wallet.inner.lock().unwrap();
+			let funding_tx = Transaction {
+				version: bitcoin::transaction::Version::TWO,
+				lock_time: LockTime::ZERO,
+				input: Vec::new(),
+				output: vec![TxOut {
+					value: Amount::from_sat(200_000),
+					script_pubkey: locked_wallet
+						.reveal_next_address(KeychainKind::External)
+						.address
+						.script_pubkey(),
+				}],
+			};
+			let funding_txid = funding_tx.compute_txid();
+			let block_id = BlockId {
+				height: locked_wallet.latest_checkpoint().height() + 1,
+				hash: bitcoin::BlockHash::from_byte_array([42; 32]),
+			};
+			let mut tx_update = TxUpdate::default();
+			tx_update.txs = vec![Arc::new(funding_tx)];
+			tx_update.anchors =
+				[(ConfirmationBlockTime { block_id, confirmation_time: 1 }, funding_txid)].into();
+			let chain = CheckPoint::from_block_ids([
+				locked_wallet.latest_checkpoint().block_id(),
+				block_id,
+			])
+			.unwrap();
+			locked_wallet
+				.apply_update(Update { tx_update, chain: Some(chain), ..Default::default() })
+				.unwrap();
+		}
+
+		// ...which the wallet spends in a replaceable payment sitting in the mempool. This
+		// transaction (B) has already replaced the original (A).
+		let tx_b = {
+			let mut locked_wallet = wallet.inner.lock().unwrap();
+			let recipient = ScriptBuf::new_p2wpkh(&WPubkeyHash::from_byte_array([0x42u8; 20]));
+			let mut builder = locked_wallet.build_tx();
+			builder
+				.add_recipient(recipient, Amount::from_sat(100_000))
+				.fee_rate(FeeRate::from_sat_per_kwu(500));
+			let mut psbt = builder.finish().unwrap();
+			assert!(locked_wallet.sign(&mut psbt, SignOptions::default()).unwrap());
+			let tx = psbt.extract_tx().unwrap();
+			locked_wallet.apply_unconfirmed_txs([(tx.clone(), 1)]);
+			tx
+		};
+		let txid_b = tx_b.compute_txid();
+
+		// The stores hold what the first bump (A -> B) and its sync left behind: the payment id
+		// is derived from A, the payment's txid is B, and `conflicting_txids` holds A.
+		let txid_a = Txid::from_byte_array([0xaa; 32]);
+		let payment_id = PaymentId(txid_a.to_byte_array());
+		let details = PaymentDetails::new(
+			payment_id,
+			PaymentKind::Onchain {
+				txid: txid_b,
+				status: ConfirmationStatus::Unconfirmed,
+				tx_type: None,
+			},
+			Some(100_000_000),
+			Some(1_000),
+			PaymentDirection::Outbound,
+			PaymentStatus::Pending,
+		);
+		wallet.payment_store.insert_or_update(details.clone()).await.unwrap();
+		let entry = PendingPaymentDetails::new(details, vec![txid_a], Vec::new());
+		wallet.pending_payment_store.insert_or_update(entry).await.unwrap();
+
+		// Bump again: B -> C.
+		let txid_c = wallet.bump_fee_rbf(payment_id, None, 0).await.unwrap();
+
+		let payment = wallet.payment_store.get(&payment_id).await.unwrap().unwrap();
+		assert!(matches!(payment.kind, PaymentKind::Onchain { txid, .. } if txid == txid_c));
+
+		// `PaymentId(txid_b)` is not the payment id and B is no longer the payment's txid, so B
+		// must be in `conflicting_txids` for wallet sync to find the payment.
+		let entry = wallet.pending_payment_store.get(&payment_id).await.unwrap().unwrap();
+		assert_eq!(entry.conflicting_txids, vec![txid_a, txid_b]);
+		assert_eq!(wallet.find_payment_by_txid(txid_b).await.unwrap(), Some(payment_id));
+	}
+
 	/// Removing a payment must also drop its pending-store entry. The entry indexes the
 	/// payment's txids (current, conflicting, and candidates), so leaving it behind keeps
 	/// resolving those txids to the removed record — routing later wallet events to a payment

tests/integration_tests_rust.rs

@@ -4667,6 +4667,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();