fix(chain): detect incoming transaction being replaced/canceled

crates/chain/src/indexed_tx_graph.rs

@@ -1,5 +1,6 @@
 //! Contains the [`IndexedTxGraph`] and associated types. Refer to the
 //! [`IndexedTxGraph`] documentation for more.
+use crate::collections::HashSet;
 use core::fmt::Debug;
 
 use alloc::{sync::Arc, vec::Vec};
@@ -178,7 +179,7 @@ where
 
         let mut tx_graph = tx_graph::ChangeSet::default();
         for (tx, anchors) in txs {
-            if self.index.is_tx_relevant(&tx) {
+            if self.index.is_tx_relevant(&tx) || self.graph.direct_conflicts(&tx).next().is_some() {
                 let txid = tx.compute_txid();
                 tx_graph.merge(self.graph.insert_tx(tx.clone()));
                 for anchor in anchors {
@@ -218,10 +219,37 @@ where
             indexer.merge(self.index.index_tx(tx));
         }
 
+        // It is possible that a spk-relevant transaction can be replaced with a non-spk-relevant
+        // transaction. Furthermore, the existence of the replacement transaction may mean that the
+        // original spk-relevant transaction is evicted. Therefore, we need to include the new
+        // non-spk-relevant transaction in the graph.
+        //
+        // `relevant_spends` keeps tracks of outpoints that are associated with spk-relevant
+        // transactions in `unconfirmed_txs`. This allows us to detect non-spk-relevant replacements
+        // when both the original and the replacement are introduced together.
+        let mut relevant_spends = HashSet::<OutPoint>::new();
+
+        let mut indexer_changeset = I::ChangeSet::default();
+        for (tx, _) in &txs {
+            indexer_changeset.merge(self.index.index_tx(tx));
+
+            if self.index.is_tx_relevant(tx) && !tx.is_coinbase() {
+                relevant_spends.extend(tx.input.iter().map(|txin| txin.previous_output));
+            }
+        }
+
         let graph = self.graph.batch_insert_unconfirmed(
             txs.into_iter()
-                .filter(|(tx, _)| self.index.is_tx_relevant(tx))
-                .map(|(tx, seen_at)| (tx.clone(), seen_at)),
+                .filter(|(tx, _)| {
+                    self.index.is_tx_relevant(tx)
+                        || tx
+                            .input
+                            .iter()
+                            .any(|txin| relevant_spends.contains(&txin.previous_output))
+                        || self.graph.direct_conflicts(tx).next().is_some()
+                })
+                .map(|(tx, seen_at)| (tx.clone(), seen_at))
+                .collect::<Vec<_>>(),
         );
 
         ChangeSet {
@@ -278,7 +306,8 @@ where
         let mut changeset = ChangeSet::<A, I::ChangeSet>::default();
         for (tx_pos, tx) in block.txdata.iter().enumerate() {
             changeset.indexer.merge(self.index.index_tx(tx));
-            if self.index.is_tx_relevant(tx) {
+
+            if self.index.is_tx_relevant(tx) || self.graph.direct_conflicts(tx).next().is_some() {
                 let txid = tx.compute_txid();
                 let anchor = TxPosInBlock {
                     block,

crates/chain/src/indexer/keychain_txout.rs

@@ -7,11 +7,13 @@ use crate::{
     spk_client::{FullScanRequestBuilder, SyncRequestBuilder},
     spk_iter::BIP32_MAX_INDEX,
     spk_txout::SpkTxOutIndex,
-    DescriptorExt, DescriptorId, Indexed, Indexer, KeychainIndexed, SpkIterator,
+    Anchor, CanonicalIter, ChainOracle, DescriptorExt, DescriptorId, Indexed, Indexer,
+    KeychainIndexed, SpkIterator,
 };
 use alloc::{borrow::ToOwned, vec::Vec};
 use bitcoin::{Amount, OutPoint, ScriptBuf, SignedAmount, Transaction, TxOut, Txid};
 use core::{
+    convert::Infallible,
     fmt::Debug,
     ops::{Bound, RangeBounds},
 };
@@ -879,6 +881,18 @@ pub trait SyncRequestBuilderExt<K> {
 
     /// Add [`Script`](bitcoin::Script)s that are revealed by the `indexer` but currently unused.
     fn unused_spks_from_indexer(self, indexer: &KeychainTxOutIndex<K>) -> Self;
+
+    /// Add [`OutPoint`]s which are spent by unconfirmed transactions.
+    ///
+    /// This allows the chain source to detect transactions which are cancelled/replaced.
+    fn unconfirmed_outpoints<A, C>(
+        self,
+        canonical_iter: CanonicalIter<A, C>,
+        indexer: &KeychainTxOutIndex<K>,
+    ) -> Self
+    where
+        A: Anchor,
+        C: ChainOracle<Error = Infallible>;
 }
 
 impl<K: Clone + Ord + core::fmt::Debug> SyncRequestBuilderExt<K> for SyncRequestBuilder<(K, u32)> {
@@ -892,6 +906,37 @@ impl<K: Clone + Ord + core::fmt::Debug> SyncRequestBuilderExt<K> for SyncRequest
     fn unused_spks_from_indexer(self, indexer: &KeychainTxOutIndex<K>) -> Self {
         self.spks_with_indexes(indexer.unused_spks())
     }
+
+    fn unconfirmed_outpoints<A, C>(
+        self,
+        canonical_iter: CanonicalIter<A, C>,
+        indexer: &KeychainTxOutIndex<K>,
+    ) -> Self
+    where
+        A: Anchor,
+        C: ChainOracle<Error = Infallible>,
+    {
+        self.outpoints(
+            canonical_iter
+                .filter_map(|r| {
+                    let (_, tx, reason) = r.expect("infallible");
+                    match reason {
+                        crate::CanonicalReason::ObservedIn { .. }
+                            if indexer.is_tx_relevant(&tx) =>
+                        {
+                            Some(tx)
+                        }
+                        _ => None,
+                    }
+                })
+                .flat_map(|tx| {
+                    tx.input
+                        .iter()
+                        .map(|txin| txin.previous_output)
+                        .collect::<Vec<_>>()
+                }),
+        )
+    }
 }
 
 /// Trait to extend [`FullScanRequestBuilder`].

crates/chain/tests/test_indexed_tx_graph.rs

@@ -3,12 +3,16 @@
 #[macro_use]
 mod common;
 
-use std::{collections::BTreeSet, sync::Arc};
+use std::{
+    collections::{BTreeMap, BTreeSet},
+    sync::Arc,
+};
 
 use bdk_chain::{
     indexed_tx_graph::{self, IndexedTxGraph},
     indexer::keychain_txout::KeychainTxOutIndex,
     local_chain::LocalChain,
+    spk_txout::SpkTxOutIndex,
     tx_graph, Balance, ChainPosition, ConfirmationBlockTime, DescriptorExt,
 };
 use bdk_testenv::{
@@ -99,6 +103,139 @@ fn insert_relevant_txs() {
     assert_eq!(graph.initial_changeset(), initial_changeset);
 }
 
+/// Ensure that [`IndexedTxGraph::batch_insert_relevant`] adds transactions that are direct
+/// conflicts with transactions in our graph but are not directly tracked by our indexer.
+#[test]
+fn insert_relevant_conflicting_txs() {
+    let (descriptor, _) = Descriptor::parse_descriptor(&Secp256k1::signing_only(), DESCRIPTORS[0])
+        .expect("must be valid");
+    let sender_spk = descriptor.at_derivation_index(0).unwrap().script_pubkey();
+    let receiver_spk = descriptor.at_derivation_index(9).unwrap().script_pubkey();
+
+    let sender_prev_tx = Transaction {
+        output: vec![TxOut {
+            value: Amount::from_sat(10_000),
+            script_pubkey: sender_spk,
+        }],
+        ..new_tx(0)
+    };
+
+    let send_tx = Transaction {
+        input: vec![TxIn {
+            previous_output: OutPoint::new(sender_prev_tx.compute_txid(), 0),
+            ..Default::default()
+        }],
+        output: vec![TxOut {
+            value: Amount::from_sat(10_000),
+            script_pubkey: receiver_spk.clone(),
+        }],
+        ..new_tx(1)
+    };
+
+    let undo_send_tx = Transaction {
+        input: vec![TxIn {
+            previous_output: OutPoint::new(sender_prev_tx.compute_txid(), 0),
+            ..Default::default()
+        }],
+        ..new_tx(2)
+    };
+
+    // Given: An empty receiver IndexedTxGraph.
+    // When: batch_insert_relevant_unconfirmed is called with [send_tx, undo_send_tx].
+    // Assert: Both txs are included in the changeset.
+    let mut graph = IndexedTxGraph::<ConfirmationBlockTime, SpkTxOutIndex<u32>>::new({
+        let mut index = SpkTxOutIndex::default();
+        let _ = index.insert_spk(0u32, receiver_spk.clone());
+        index
+    });
+    let txs = [send_tx.clone(), undo_send_tx.clone()];
+
+    let changeset = graph.batch_insert_relevant_unconfirmed(txs.iter().cloned().map(|tx| (tx, 0)));
+
+    let expected_changeset = indexed_tx_graph::ChangeSet {
+        tx_graph: tx_graph::ChangeSet::<ConfirmationBlockTime> {
+            txs: txs.iter().cloned().map(Arc::new).collect(),
+            last_seen: BTreeMap::from([
+                (send_tx.compute_txid(), 0_u64),
+                (undo_send_tx.compute_txid(), 0_u64),
+            ]),
+            ..Default::default()
+        },
+        indexer: (),
+    };
+    assert_eq!(changeset, expected_changeset);
+
+    // Given: An empty receiver IndexedTxGraph.
+    // When: batch_insert_relevant_unconfirmed is called with [undo_send_tx, send_tx].
+    // Assert: Both txs are included in the changeset.
+    let mut graph = IndexedTxGraph::<ConfirmationBlockTime, SpkTxOutIndex<u32>>::new({
+        let mut index = SpkTxOutIndex::default();
+        let _ = index.insert_spk(0u32, receiver_spk.clone());
+        index
+    });
+    let txs = [undo_send_tx.clone(), send_tx.clone()];
+
+    let changeset = graph.batch_insert_relevant_unconfirmed(txs.iter().cloned().map(|tx| (tx, 0)));
+
+    let expected_changeset = indexed_tx_graph::ChangeSet {
+        tx_graph: tx_graph::ChangeSet::<ConfirmationBlockTime> {
+            txs: txs.iter().cloned().map(Arc::new).collect(),
+            last_seen: BTreeMap::from([
+                (undo_send_tx.compute_txid(), 0_u64),
+                (send_tx.compute_txid(), 0_u64),
+            ]),
+            ..Default::default()
+        },
+        indexer: (),
+    };
+    assert_eq!(changeset, expected_changeset);
+
+    // Given: A receiver IndexedTxGraph which already contains send_tx.
+    // When: batch_insert_relevant_unconfirmed is called with undo_send_tx.
+    // Assert: undo_send_tx is included in the returned changeset.
+    let mut graph = IndexedTxGraph::<ConfirmationBlockTime, SpkTxOutIndex<u32>>::new({
+        let mut index = SpkTxOutIndex::default();
+        let _ = index.insert_spk(0u32, receiver_spk.clone());
+        index
+    });
+    let _ = graph.insert_tx(send_tx.clone());
+    let txs = [undo_send_tx.clone()];
+
+    let changeset = graph.batch_insert_relevant_unconfirmed(txs.iter().cloned().map(|tx| (tx, 0)));
+
+    let expected_changeset = indexed_tx_graph::ChangeSet {
+        tx_graph: tx_graph::ChangeSet::<ConfirmationBlockTime> {
+            txs: txs.iter().cloned().map(Arc::new).collect(),
+            last_seen: BTreeMap::from([(undo_send_tx.compute_txid(), 0_u64)]),
+            ..Default::default()
+        },
+        indexer: (),
+    };
+    assert_eq!(changeset, expected_changeset);
+
+    // Given: A receiver IndexedTxGraph which already contains send_tx.
+    // When: batch_insert_relevant is called with undo_send_tx.
+    // Assert: undo_send_tx is included in the returned changeset.
+    let mut graph = IndexedTxGraph::<ConfirmationBlockTime, SpkTxOutIndex<u32>>::new({
+        let mut index = SpkTxOutIndex::default();
+        let _ = index.insert_spk(0u32, receiver_spk);
+        index
+    });
+    let _ = graph.insert_tx(send_tx);
+    let txs = [undo_send_tx.clone()];
+
+    let changeset = graph.batch_insert_relevant(txs.iter().cloned().map(|tx| (tx, None)));
+
+    let expected_changeset = indexed_tx_graph::ChangeSet {
+        tx_graph: tx_graph::ChangeSet::<ConfirmationBlockTime> {
+            txs: txs.iter().cloned().map(Arc::new).collect(),
+            ..Default::default()
+        },
+        indexer: (),
+    };
+    assert_eq!(changeset, expected_changeset);
+}
+
 /// Ensure consistency IndexedTxGraph list_* and balance methods. These methods lists
 /// relevant txouts and utxos from the information fetched from a ChainOracle (here a LocalChain).
 ///

crates/wallet/src/wallet/mod.rs

@@ -2445,13 +2445,39 @@ impl Wallet {
     /// This is the first step when performing a spk-based wallet partial sync, the returned
     /// [`SyncRequest`] collects all revealed script pubkeys from the wallet keychain needed to
     /// start a blockchain sync with a spk based blockchain client.
+    #[deprecated(
+        since = "1.1.0",
+        note = "start_sync_with_revealed_spks could not detect receiving transactions being replaced. Use start_sync instead"
+    )]
     pub fn start_sync_with_revealed_spks(&self) -> SyncRequestBuilder<(KeychainKind, u32)> {
         use bdk_chain::keychain_txout::SyncRequestBuilderExt;
         SyncRequest::builder()
             .chain_tip(self.chain.tip())
             .revealed_spks_from_indexer(&self.indexed_graph.index, ..)
     }
 
+    /// Create a [`SyncRequest`] for this wallet.
+    ///
+    /// This assembles a request which initiates a sync against a spk-based chain source. This
+    /// request contains all revealed script pubkeys and unconfirmed spends.
+    ///
+    /// This request can detect when transactions get cancelled/replaced.
+    pub fn start_sync(&self) -> SyncRequestBuilder<(KeychainKind, u32)> {
+        use bdk_chain::keychain_txout::SyncRequestBuilderExt;
+
+        let chain = &self.chain;
+        let tx_graph = &self.indexed_graph.graph();
+        let tx_index = &self.indexed_graph.index;
+
+        SyncRequest::builder()
+            .chain_tip(chain.tip())
+            .revealed_spks_from_indexer(tx_index, ..)
+            .unconfirmed_outpoints(
+                tx_graph.canonical_iter(chain, chain.tip().block_id()),
+                tx_index,
+            )
+    }
+
     /// Create a [`FullScanRequest] for this wallet.
     ///
     /// This is the first step when performing a spk-based wallet full scan, the returned