Add target limit to sync requests (#1818)

crates/core/src/spk_client.rs

@@ -2,7 +2,7 @@
 use crate::{
     alloc::{boxed::Box, collections::VecDeque, vec::Vec},
     collections::{BTreeMap, HashMap, HashSet},
-    CheckPoint, ConfirmationBlockTime, Indexed,
+    BlockId, CheckPoint, ConfirmationBlockTime, Indexed,
 };
 use bitcoin::{BlockHash, OutPoint, Script, ScriptBuf, Txid};
 
@@ -132,6 +132,14 @@ impl<I, D> SyncRequestBuilder<I, D> {
         self
     }
 
+    /// Set the target chain tip for the sync request.
+    ///
+    /// The sync will not fetch chain data beyond this target.
+    pub fn target(mut self, target: BlockId) -> Self {
+        self.inner.target = Some(target);
+        self
+    }
+
     /// Add [`Script`]s coupled with associated indexes that will be synced against.
     ///
     /// # Example
@@ -259,6 +267,7 @@ impl<I, D> SyncRequestBuilder<I, D> {
 pub struct SyncRequest<I = (), D = BlockHash> {
     start_time: u64,
     chain_tip: Option<CheckPoint<D>>,
+    target: Option<BlockId>,
     spks: VecDeque<(I, ScriptBuf)>,
     spks_consumed: usize,
     spk_expected_txids: HashMap<ScriptBuf, HashSet<Txid>>,
@@ -289,6 +298,7 @@ impl<I, D> SyncRequest<I, D> {
             inner: Self {
                 start_time,
                 chain_tip: None,
+                target: None,
                 spks: VecDeque::new(),
                 spks_consumed: 0,
                 spk_expected_txids: HashMap::new(),
@@ -337,6 +347,11 @@ impl<I, D> SyncRequest<I, D> {
         self.chain_tip.clone()
     }
 
+    /// Get the target chain tip [`BlockId`] of this request (if any).
+    pub fn target(&self) -> Option<BlockId> {
+        self.target
+    }
+
     /// Advances the sync request and returns the next [`ScriptBuf`] with corresponding [`Txid`]
     /// history.
     ///
@@ -444,6 +459,14 @@ impl<K: Ord, D> FullScanRequestBuilder<K, D> {
         self
     }
 
+    /// Set the target chain tip for the full scan request.
+    ///
+    /// The sync will not fetch chain data beyond this target.
+    pub fn target(mut self, target: BlockId) -> Self {
+        self.inner.target = Some(target);
+        self
+    }
+
     /// Set the spk iterator for a given `keychain`.
     pub fn spks_for_keychain(
         mut self,
@@ -495,6 +518,7 @@ impl<K: Ord, D> FullScanRequestBuilder<K, D> {
 pub struct FullScanRequest<K, D = BlockHash> {
     start_time: u64,
     chain_tip: Option<CheckPoint<D>>,
+    target: Option<BlockId>,
     spks_by_keychain: BTreeMap<K, Box<dyn Iterator<Item = Indexed<ScriptBuf>> + Send>>,
     last_revealed: BTreeMap<K, u32>,
     inspect: Box<InspectFullScan<K>>,
@@ -520,6 +544,7 @@ impl<K: Ord + Clone, D> FullScanRequest<K, D> {
             inner: Self {
                 start_time,
                 chain_tip: None,
+                target: None,
                 spks_by_keychain: BTreeMap::new(),
                 last_revealed: BTreeMap::new(),
                 inspect: Box::new(|_, _, _| ()),
@@ -551,6 +576,11 @@ impl<K: Ord + Clone, D> FullScanRequest<K, D> {
         self.chain_tip.clone()
     }
 
+    /// Get the target chain tip [`BlockId`] of this request (if any).
+    pub fn target(&self) -> Option<BlockId> {
+        self.target
+    }
+
     /// List all keychains contained in this request.
     pub fn keychains(&self) -> Vec<K> {
         self.spks_by_keychain.keys().cloned().collect()

crates/electrum/src/bdk_electrum_client.rs

@@ -132,7 +132,11 @@
         let start_time = request.start_time();
 
         let tip_and_latest_blocks = match request.chain_tip() {
-            Some(chain_tip) => Some(fetch_tip_and_latest_blocks(&self.inner, chain_tip)?),
+            Some(chain_tip) => Some(fetch_tip_and_latest_blocks(
+                &self.inner,
+                chain_tip,
+                request.target(),
+            )?),
             None => None,
         };
 
@@ -150,6 +154,7 @@
                 spks,
                 stop_gap,
                 last_revealed,
+                request.target(),
                 batch_size,
                 &mut pending_anchors,
             )? {
@@ -215,9 +220,10 @@
     ) -> Result<SyncResponse, Error> {
         let mut request: SyncRequest<I> = request.into();
         let start_time = request.start_time();
+        let target = request.target();
 
         let tip_and_latest_blocks = match request.chain_tip() {
-            Some(chain_tip) => Some(fetch_tip_and_latest_blocks(&self.inner, chain_tip)?),
+            Some(chain_tip) => Some(fetch_tip_and_latest_blocks(&self.inner, chain_tip, target)?),
             None => None,
         };
 
@@ -232,19 +238,22 @@
                 .map(|(i, spk)| (i as u32, spk)),
             usize::MAX,
             None,
+            target,
             batch_size,
             &mut pending_anchors,
         )?;
         self.populate_with_txids(
             start_time,
             &mut tx_update,
             request.iter_txids(),
+            target,
             &mut pending_anchors,
         )?;
         self.populate_with_outpoints(
             start_time,
             &mut tx_update,
             request.iter_outpoints(),
+            target,
             &mut pending_anchors,
         )?;
 
@@ -288,6 +297,7 @@
         mut spks_with_expected_txids: impl Iterator<Item = (u32, SpkWithExpectedTxids)>,
         stop_gap: usize,
         last_revealed: Option<u32>,
+        target: Option<BlockId>,
         batch_size: usize,
         pending_anchors: &mut Vec<(Txid, usize)>,
     ) -> Result<Option<u32>, Error> {
@@ -335,7 +345,11 @@
                     match tx_res.height.try_into() {
                         // Returned heights 0 & -1 are reserved for unconfirmed txs.
                         Ok(height) if height > 0 => {
-                            pending_anchors.push((tx_res.tx_hash, height));
+                            if target.map_or(true, |t| height as u32 <= t.height) {
+                                pending_anchors.push((tx_res.tx_hash, height));
+                            } else {
+                                tx_update.seen_ats.insert((tx_res.tx_hash, start_time));
+                            }
                         }
                         _ => {
                             tx_update.seen_ats.insert((tx_res.tx_hash, start_time));
@@ -355,6 +369,7 @@
         start_time: u64,
         tx_update: &mut TxUpdate<ConfirmationBlockTime>,
         outpoints: impl IntoIterator<Item = OutPoint>,
+        target: Option<BlockId>,
         pending_anchors: &mut Vec<(Txid, usize)>,
     ) -> Result<(), Error> {
         // Collect valid outpoints with their corresponding `spk` and `tx`.
@@ -398,7 +413,11 @@
                         match res.height.try_into() {
                             // Returned heights 0 & -1 are reserved for unconfirmed txs.
                             Ok(height) if height > 0 => {
-                                pending_anchors.push((res.tx_hash, height));
+                                if target.map_or(true, |t| height as u32 <= t.height) {
+                                    pending_anchors.push((res.tx_hash, height));
+                                } else {
+                                    tx_update.seen_ats.insert((res.tx_hash, start_time));
+                                }
                             }
                             _ => {
                                 tx_update.seen_ats.insert((res.tx_hash, start_time));
@@ -420,7 +439,11 @@
                         match res.height.try_into() {
                             // Returned heights 0 & -1 are reserved for unconfirmed txs.
                             Ok(height) if height > 0 => {
-                                pending_anchors.push((res.tx_hash, height));
+                                if target.map_or(true, |t| height as u32 <= t.height) {
+                                    pending_anchors.push((res.tx_hash, height));
+                                } else {
+                                    tx_update.seen_ats.insert((res.tx_hash, start_time));
+                                }
                             }
                             _ => {
                                 tx_update.seen_ats.insert((res.tx_hash, start_time));
@@ -440,6 +463,7 @@
         start_time: u64,
         tx_update: &mut TxUpdate<ConfirmationBlockTime>,
         txids: impl IntoIterator<Item = Txid>,
+        target: Option<BlockId>,
         pending_anchors: &mut Vec<(Txid, usize)>,
     ) -> Result<(), Error> {
         let mut txs = Vec::<(Txid, Arc<Transaction>)>::new();
@@ -501,7 +525,11 @@
                 match res.height.try_into() {
                     // Returned heights 0 & -1 are reserved for unconfirmed txs.
                     Ok(height) if height > 0 => {
-                        pending_anchors.push((tx.0, height));
+                        if target.map_or(true, |t| height as u32 <= t.height) {
+                            pending_anchors.push((tx.0, height));
+                        } else {
+                            tx_update.seen_ats.insert((res.tx_hash, start_time));
+                        }
                     }
                     _ => {
                         tx_update.seen_ats.insert((res.tx_hash, start_time));
@@ -652,9 +680,16 @@
 fn fetch_tip_and_latest_blocks(
     client: &impl ElectrumApi,
     prev_tip: CheckPoint<BlockHash>,
+    target: Option<BlockId>,
 ) -> Result<(CheckPoint<BlockHash>, BTreeMap<u32, BlockHash>), Error> {
     let HeaderNotification { height, .. } = client.block_headers_subscribe()?;
-    let new_tip_height = height as u32;
+    let mut new_tip_height = height as u32;
+
+    if let Some(t) = target {
+        if new_tip_height > t.height {
+            new_tip_height = t.height;
+        }
+    }
 
     // If electrum returns a tip height that is lower than our previous tip, then checkpoints do
     // not need updating. We just return the previous tip and use that as the point of agreement.
@@ -871,4 +906,48 @@
 
         Ok(())
     }
+
+    #[cfg(feature = "default")]
+    #[test]
+    fn test_target_limit() -> anyhow::Result<()> {
+        use bdk_chain::BlockId;
+
+        let env = TestEnv::new()?;
+        let client = electrum_client::Client::new(env.electrsd.electrum_url.as_str()).unwrap();
+        let electrum_client = BdkElectrumClient::new(client);
+
+        let addr = env.rpc_client().get_new_address(None, None)?.address()?.assume_checked();
+        let spk = addr.script_pubkey();
+
+        // Send funds to address.
+        let txid = env.send(&addr, Amount::from_sat(50_000))?;
+
+        let target_height: u32 = env.rpc_client().get_block_count()?.into_model().0 as u32;
+        let target_hash = electrum_client.inner.block_header(target_height as _)?.block_hash();
+
+        // Mine a block to confirm the transaction.
+        env.mine_blocks(1, None)?;
+        env.wait_until_electrum_sees_block(Duration::from_secs(6))?;
+
+        let bogus_genesis = constants::genesis_block(Network::Testnet).block_hash();
+        let cp = CheckPoint::new(0, bogus_genesis);
+
+        let target = BlockId {
+            height: target_height,
+            hash: target_hash,
+        };
+
+        let req = SyncRequest::builder()
+            .chain_tip(cp)
+            .spks([spk])
+            .target(target)
+            .build();
+
+        let res = electrum_client.sync(req, 10, false)?;
+
+        assert!(res.tx_update.anchors.is_empty(), "anchors should be empty");
+        assert!(res.tx_update.seen_ats.iter().any(|(t, _)| *t == txid), "tx should be in seen_ats");
+
+        Ok(())
+    }
 }