feat(chain)!: IndexedTxGraph::expected_unconfirmed_spk_txids

Make this method work when the indexer is `KeychainTxOutIndex`. We
reintroduce the ability to get the internal `SpkTxOutIndex` from
`KeychainTxOutIndex` so that `SpkTxOutIndex::relevant_spks_of_tx` is
callable from `KeychainTxOutIndex`.

This commit renames `iter_spks_with_expected_txids` to
`expected_unconfirmed_spk_txids` for `TxGraph`, `IndexedTxGraph` and
`SyncRequestBuilder`. Docs are also improved to explain how these
methods are useful.

Remove unused `SyncRequestBuilder` methods.

Author: evanlinjin

crates/bitcoind_rpc/tests/test_emitter.rs

@@ -808,7 +808,7 @@ fn test_expect_tx_evicted() -> anyhow::Result<()> {
 
     // We evict the expected txs that are missing from mempool.
     let exp_txids = graph
-        .iter_spks_with_expected_txids(&chain, ..)
+        .expected_unconfirmed_spk_txids(&chain, chain_tip, ..)?
         .collect::<Vec<_>>();
     assert_eq!(exp_txids, vec![(txid_1, spk)]);
     let mempool = emitter

crates/chain/src/indexed_tx_graph.rs

@@ -1,7 +1,6 @@
 //! Contains the [`IndexedTxGraph`] and associated types. Refer to the
 //! [`IndexedTxGraph`] documentation for more.
 
-use core::fmt;
 use core::ops::RangeBounds;
 
 use alloc::{sync::Arc, vec::Vec};
@@ -345,24 +344,50 @@ where
 impl<A, I> IndexedTxGraph<A, SpkTxOutIndex<I>>
 where
     A: Anchor,
-    I: fmt::Debug + Clone + Ord,
+    I: core::fmt::Debug + Clone + Ord,
 {
-    /// Returns an iterator over unconfirmed transactions and their associated script pubkeys,
-    /// filtered within the specified `range`.
+    /// Iterate over unconfirmed txids that we expect to exist in a chain source's spk history
+    /// response.
     ///
-    /// This function delegates the transaction filtering to [`TxGraph::iter_spks_with_expected_txids`],
-    /// using the [`SpkTxOutIndex`] stored in [`IndexedTxGraph`]. The [`TxGraph`] internally scans
-    /// for unconfirmed transactions relevant to the indexed outputs.
-    pub fn iter_spks_with_expected_txids<'a, O>(
+    /// This is used to fill [`SyncRequestBuilder::expected_unconfirmed_spk_txids`](bdk_core::spk_client::SyncRequestBuilder::expected_unconfirmed_spk_txids).
+    ///
+    /// The spk range can be contrained with `range`.
+    pub fn expected_unconfirmed_spk_txids<'a, O>(
         &'a self,
         chain: &'a O,
+        chain_tip: BlockId,
         range: impl RangeBounds<I> + 'a,
-    ) -> impl Iterator<Item = (Txid, ScriptBuf)> + 'a
+    ) -> Result<impl Iterator<Item = (Txid, ScriptBuf)> + 'a, O::Error>
+    where
+        O: ChainOracle,
+    {
+        self.graph
+            .expected_unconfirmed_spk_txids(chain, chain_tip, &self.index, range)
+    }
+}
+
+impl<A, K> IndexedTxGraph<A, crate::keychain_txout::KeychainTxOutIndex<K>>
+where
+    A: Anchor,
+    K: core::fmt::Debug + Clone + Ord,
+{
+    /// Iterate over unconfirmed txids that we expect to exist in a chain source's spk history
+    /// response.
+    ///
+    /// This is used to fill [`SyncRequestBuilder::expected_unconfirmed_spk_txids`](bdk_core::spk_client::SyncRequestBuilder::expected_unconfirmed_spk_txids).
+    ///
+    /// The spk range can be contrained with `range`.
+    pub fn expected_unconfirmed_spk_txids<'a, O>(
+        &'a self,
+        chain: &'a O,
+        chain_tip: BlockId,
+        range: impl RangeBounds<(K, u32)> + 'a,
+    ) -> Result<impl Iterator<Item = (Txid, ScriptBuf)> + 'a, O::Error>
     where
-        O: ChainOracle<Error = core::convert::Infallible>,
+        O: ChainOracle,
     {
         self.graph
-            .iter_spks_with_expected_txids(chain, &self.index, range)
+            .expected_unconfirmed_spk_txids(chain, chain_tip, &self.index, range)
     }
 }
 

crates/chain/src/indexer/keychain_txout.rs

@@ -7,13 +7,11 @@ use crate::{
     spk_client::{FullScanRequestBuilder, SyncRequestBuilder},
     spk_iter::BIP32_MAX_INDEX,
     spk_txout::SpkTxOutIndex,
-    Anchor, CanonicalIter, CanonicalReason, ChainOracle, DescriptorExt, DescriptorId, Indexed,
-    Indexer, KeychainIndexed, SpkIterator,
+    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},
 };
@@ -138,6 +136,12 @@ impl<K> Default for KeychainTxOutIndex<K> {
     }
 }
 
+impl<K> AsRef<SpkTxOutIndex<(K, u32)>> for KeychainTxOutIndex<K> {
+    fn as_ref(&self) -> &SpkTxOutIndex<(K, u32)> {
+        self.inner()
+    }
+}
+
 impl<K: Clone + Ord + Debug> Indexer for KeychainTxOutIndex<K> {
     type ChangeSet = ChangeSet;
 
@@ -202,6 +206,11 @@ impl<K> KeychainTxOutIndex<K> {
             lookahead,
         }
     }
+
+    /// Get a reference to the internal [`SpkTxOutIndex`].
+    pub fn inner(&self) -> &SpkTxOutIndex<(K, u32)> {
+        &self.inner
+    }
 }
 
 /// Methods that are *re-exposed* from the internal [`SpkTxOutIndex`].
@@ -881,20 +890,6 @@ 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 unconfirmed txids and their associated spks.
-    ///
-    /// We expect that the chain source should include these txids in their spk histories. If not,
-    /// the transaction has been evicted for some reason and we will inform the receiving
-    /// structures in the response.
-    fn check_unconfirmed_statuses<A, C>(
-        self,
-        indexer: &KeychainTxOutIndex<K>,
-        canonical_iter: CanonicalIter<A, C>,
-    ) -> Self
-    where
-        A: Anchor,
-        C: ChainOracle<Error = Infallible>;
 }
 
 impl<K: Clone + Ord + core::fmt::Debug> SyncRequestBuilderExt<K> for SyncRequestBuilder<(K, u32)> {
@@ -908,29 +903,6 @@ 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 check_unconfirmed_statuses<A, C>(
-        self,
-        indexer: &KeychainTxOutIndex<K>,
-        canonical_iter: CanonicalIter<A, C>,
-    ) -> Self
-    where
-        A: Anchor,
-        C: ChainOracle<Error = Infallible>,
-    {
-        self.expected_txids_of_spk(
-            canonical_iter
-                .map(|res| res.expect("infallible"))
-                .filter(|(_, _, reason)| matches!(reason, CanonicalReason::ObservedIn { .. }))
-                .flat_map(|(txid, tx, _)| {
-                    indexer
-                        .inner
-                        .relevant_spks_of_tx(tx.as_ref())
-                        .into_iter()
-                        .map(move |spk| (txid, spk))
-                }),
-        )
-    }
 }
 
 /// Trait to extend [`FullScanRequestBuilder`].

crates/chain/src/indexer/spk_txout.rs

@@ -42,6 +42,12 @@ pub struct SpkTxOutIndex<I> {
     spk_txouts: BTreeSet<(I, OutPoint)>,
 }
 
+impl<I> AsRef<SpkTxOutIndex<I>> for SpkTxOutIndex<I> {
+    fn as_ref(&self) -> &SpkTxOutIndex<I> {
+        self
+    }
+}
+
 impl<I> Default for SpkTxOutIndex<I> {
     fn default() -> Self {
         Self {

crates/chain/src/tx_graph.rs

@@ -95,7 +95,6 @@ use crate::spk_txout::SpkTxOutIndex;
 use crate::BlockId;
 use crate::CanonicalIter;
 use crate::CanonicalReason;
-use crate::Indexer;
 use crate::ObservedIn;
 use crate::{Anchor, Balance, ChainOracle, ChainPosition, FullTxOut, Merge};
 use alloc::collections::vec_deque::VecDeque;
@@ -1164,48 +1163,46 @@ impl<A: Anchor> TxGraph<A> {
             .expect("oracle is infallible")
     }
 
-    /// Returns an iterator over unconfirmed transactions and their associated script pubkeys,
-    /// filtered within the specified `range`.
+    /// Iterate over unconfirmed txids that we expect to exist in a chain source's spk history
+    /// response.
     ///
-    /// This function scans the transaction graph for unconfirmed transactions relevant to the
-    /// provided [`SpkTxOutIndex`], determining which transactions should be considered based on
-    /// indexed outputs.
-    pub fn iter_spks_with_expected_txids<'a, C, I>(
+    /// This is used to fill [`SyncRequestBuilder::expected_unconfirmed_spk_txids`](bdk_core::spk_client::SyncRequestBuilder::expected_unconfirmed_spk_txids).
+    ///
+    /// The spk range can be contrained with `range`.
+    pub fn expected_unconfirmed_spk_txids<'a, C, I>(
         &'a self,
         chain: &'a C,
-        indexer: &'a SpkTxOutIndex<I>,
+        chain_tip: BlockId,
+        indexer: &'a impl AsRef<SpkTxOutIndex<I>>,
         range: impl RangeBounds<I> + 'a,
-    ) -> impl Iterator<Item = (Txid, ScriptBuf)> + 'a
+    ) -> Result<impl Iterator<Item = (Txid, ScriptBuf)> + 'a, C::Error>
     where
-        C: ChainOracle<Error = core::convert::Infallible>,
-        I: fmt::Debug + Clone + Ord,
+        C: ChainOracle,
+        I: fmt::Debug + Clone + Ord + 'a,
     {
-        let chain_tip = chain.get_chain_tip().unwrap();
-
-        self.list_canonical_txs(chain, chain_tip)
-            .filter(|c| !c.chain_position.is_confirmed() && indexer.is_tx_relevant(&c.tx_node))
-            .flat_map(move |c| {
-                let txid = c.tx_node.txid;
-                let spks = c
-                    .tx_node
-                    .input
+        let mut spk_txs = vec![];
+        for res in self.try_list_canonical_txs(chain, chain_tip) {
+            let canonical_tx = res?;
+            if canonical_tx.chain_position.is_confirmed() {
+                continue;
+            }
+            let txid = canonical_tx.tx_node.txid;
+            let tx = canonical_tx.tx_node.tx;
+            let outpoints = tx.input.iter().map(|txin| txin.previous_output).chain(
+                tx.output
                     .iter()
-                    .map(|txin| txin.previous_output)
-                    .chain(
-                        c.tx_node
-                            .output
-                            .iter()
-                            .enumerate()
-                            .map(move |(vout, _)| OutPoint::new(txid, vout as u32)),
-                    )
-                    .flat_map(|op| match indexer.txout(op) {
-                        Some((i, txo)) if range.contains(i) => Some(txo.script_pubkey.clone()),
-                        _ => None,
-                    })
-                    .collect::<Vec<_>>();
-
-                core::iter::repeat(txid).zip(spks)
-            })
+                    .enumerate()
+                    .map(|(vout, _)| OutPoint::new(txid, vout as u32)),
+            );
+            for op in outpoints {
+                if let Some((index, txo)) = indexer.as_ref().txout(op) {
+                    if range.contains(index) {
+                        spk_txs.push((txid, txo.script_pubkey.clone()));
+                    }
+                }
+            }
+        }
+        Ok(spk_txs.into_iter())
     }
 }
 

crates/core/src/spk_client.rs

@@ -170,7 +170,7 @@ impl<I> SyncRequestBuilder<I> {
     /// Add transactions that are expected to exist under a given spk.
     ///
     /// This is useful for detecting a malicious replacement of an incoming transaction.
-    pub fn expected_txids_of_spk(
+    pub fn expected_unconfirmed_spk_txids(
         mut self,
         txs: impl IntoIterator<Item = (Txid, ScriptBuf)>,
     ) -> Self {

crates/electrum/tests/test_electrum.rs

@@ -135,10 +135,11 @@ pub fn detect_receive_tx_cancel() -> anyhow::Result<()> {
     let sync_request = SyncRequest::builder()
         .chain_tip(chain.tip())
         .revealed_spks_from_indexer(&graph.index, ..)
-        .check_unconfirmed_statuses(
-            &graph.index,
-            graph.graph().canonical_iter(&chain, chain.tip().block_id()),
-        );
+        .expected_unconfirmed_spk_txids(graph.expected_unconfirmed_spk_txids(
+            &chain,
+            chain.tip().block_id(),
+            ..,
+        )?);
     let sync_response = client.sync(sync_request, BATCH_SIZE, true)?;
     assert!(
         sync_response
@@ -163,10 +164,11 @@ pub fn detect_receive_tx_cancel() -> anyhow::Result<()> {
     let sync_request = SyncRequest::builder()
         .chain_tip(chain.tip())
         .revealed_spks_from_indexer(&graph.index, ..)
-        .check_unconfirmed_statuses(
-            &graph.index,
-            graph.graph().canonical_iter(&chain, chain.tip().block_id()),
-        );
+        .expected_unconfirmed_spk_txids(graph.expected_unconfirmed_spk_txids(
+            &chain,
+            chain.tip().block_id(),
+            ..,
+        )?);
     let sync_response = client.sync(sync_request, BATCH_SIZE, true)?;
     assert!(
         sync_response.tx_update.evicted.contains(&send_txid),

crates/esplora/tests/async_ext.rs

@@ -94,10 +94,11 @@ pub async fn detect_receive_tx_cancel() -> anyhow::Result<()> {
     let sync_request = SyncRequest::builder()
         .chain_tip(chain.tip())
         .revealed_spks_from_indexer(&graph.index, ..)
-        .check_unconfirmed_statuses(
-            &graph.index,
-            graph.graph().canonical_iter(&chain, chain.tip().block_id()),
-        );
+        .expected_unconfirmed_spk_txids(graph.expected_unconfirmed_spk_txids(
+            &chain,
+            chain.tip().block_id(),
+            ..,
+        )?);
     let sync_response = client.sync(sync_request, 1).await?;
     assert!(
         sync_response
@@ -122,10 +123,11 @@ pub async fn detect_receive_tx_cancel() -> anyhow::Result<()> {
     let sync_request = SyncRequest::builder()
         .chain_tip(chain.tip())
         .revealed_spks_from_indexer(&graph.index, ..)
-        .check_unconfirmed_statuses(
-            &graph.index,
-            graph.graph().canonical_iter(&chain, chain.tip().block_id()),
-        );
+        .expected_unconfirmed_spk_txids(graph.expected_unconfirmed_spk_txids(
+            &chain,
+            chain.tip().block_id(),
+            ..,
+        )?);
     let sync_response = client.sync(sync_request, 1).await?;
     assert!(
         sync_response.tx_update.evicted.contains(&send_txid),

crates/esplora/tests/blocking_ext.rs

@@ -94,10 +94,11 @@ pub fn detect_receive_tx_cancel() -> anyhow::Result<()> {
     let sync_request = SyncRequest::builder()
         .chain_tip(chain.tip())
         .revealed_spks_from_indexer(&graph.index, ..)
-        .check_unconfirmed_statuses(
-            &graph.index,
-            graph.graph().canonical_iter(&chain, chain.tip().block_id()),
-        );
+        .expected_unconfirmed_spk_txids(graph.expected_unconfirmed_spk_txids(
+            &chain,
+            chain.tip().block_id(),
+            ..,
+        )?);
     let sync_response = client.sync(sync_request, 1)?;
     assert!(
         sync_response
@@ -122,10 +123,11 @@ pub fn detect_receive_tx_cancel() -> anyhow::Result<()> {
     let sync_request = SyncRequest::builder()
         .chain_tip(chain.tip())
         .revealed_spks_from_indexer(&graph.index, ..)
-        .check_unconfirmed_statuses(
-            &graph.index,
-            graph.graph().canonical_iter(&chain, chain.tip().block_id()),
-        );
+        .expected_unconfirmed_spk_txids(graph.expected_unconfirmed_spk_txids(
+            &chain,
+            chain.tip().block_id(),
+            ..,
+        )?);
     let sync_response = client.sync(sync_request, 1)?;
     assert!(
         sync_response.tx_update.evicted.contains(&send_txid),