Merge bitcoindevkit/bdk#1307: Bump version to 1.0.0-alpha.5

d597f4c761 Bump version to 1.0.0-alpha.5 (Steve Myers)

Pull request description:

  ### Description

  Bump version to 1.0.0-alpha.5

  bdk_chain to 0.9.0
  bdk_bitcoind_rpc to 0.4.0
  bdk_electrum to 0.7.0
  bdk_esplora to 0.7.0
  bdk_file_store to 0.5.0

  ### Checklists

  #### All Submissions:

  * [x] I've signed all my commits
  * [x] I followed the [contribution guidelines](https://github.com/bitcoindevkit/bdk/blob/master/CONTRIBUTING.md)
  * [x] I ran `cargo fmt` and `cargo clippy` before committing

ACKs for top commit:
  danielabrozzoni:
    ACK d597f4c761

Tree-SHA512: 36ba783083441ddd8d209ede35ff46832fd2532f588ddbff0e4e23fb336aff398f8ca612e2bfd5fa214d0f4c14a4dea9147b34724734cc43b752789765e7a447

.github/workflows/cont_integration.yml

@@ -30,7 +30,7 @@ jobs:
       - name: Pin dependencies for MSRV
         if: matrix.rust.version == '1.63.0'
         run: |
-          cargo update -p zip --precise "0.6.2"
+          cargo update -p zstd-sys --precise "2.0.8+zstd.1.5.5"
           cargo update -p time --precise "0.3.20"
           cargo update -p jobserver --precise "0.1.26"
           cargo update -p home --precise "0.5.5"

Cargo.toml

@@ -15,6 +15,7 @@ members = [
     "example-crates/wallet_electrum",
     "example-crates/wallet_esplora_blocking",
     "example-crates/wallet_esplora_async",
+    "example-crates/wallet_rpc",
     "nursery/tmp_plan",
     "nursery/coin_select"
 ]

README.md

@@ -48,6 +48,8 @@ The project is split up into several crates in the `/crates` directory:
 Fully working examples of how to use these components are in `/example-crates`:
 - [`example_cli`](./example-crates/example_cli): Library used by the `example_*` crates. Provides utilities for syncing, showing the balance, generating addresses and creating transactions without using the bdk `Wallet`.
 - [`example_electrum`](./example-crates/example_electrum): A command line Bitcoin wallet application built on top of `example_cli` and the `electrum` crate. It shows the power of the bdk tools (`chain` + `file_store` + `electrum`), without depending on the main `bdk` library.
+- [`example_esplora`](./example-crates/example_esplora): A command line Bitcoin wallet application built on top of `example_cli` and the `esplora` crate. It shows the power of the bdk tools (`chain` + `file_store` + `esplora`), without depending on the main `bdk` library.
+- [`example_bitcoind_rpc_polling`](./example-crates/example_bitcoind_rpc_polling): A command line Bitcoin wallet application built on top of `example_cli` and the `bitcoind_rpc` crate. It shows the power of the bdk tools (`chain` + `file_store` + `bitcoind_rpc`), without depending on the main `bdk` library.
 - [`wallet_esplora_blocking`](./example-crates/wallet_esplora_blocking): Uses the `Wallet` to sync and spend using the Esplora blocking interface.
 - [`wallet_esplora_async`](./example-crates/wallet_esplora_async): Uses the `Wallet` to sync and spend using the Esplora asynchronous interface.
 - [`wallet_electrum`](./example-crates/wallet_electrum): Uses the `Wallet` to sync and spend using Electrum.

crates/bdk/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "bdk"
 homepage = "https://bitcoindevkit.org"
-version = "1.0.0-alpha.4"
+version = "1.0.0-alpha.5"
 repository = "https://github.com/bitcoindevkit/bdk"
 documentation = "https://docs.rs/bdk"
 description = "A modern, lightweight, descriptor-based wallet library"
@@ -18,7 +18,7 @@ miniscript = { version = "10.0.0", features = ["serde"], default-features = fals
 bitcoin = { version = "0.30.0", features = ["serde", "base64", "rand-std"], default-features = false }
 serde = { version = "^1.0", features = ["derive"] }
 serde_json = { version = "^1.0" }
-bdk_chain = { path = "../chain", version = "0.8.0", features = ["miniscript", "serde"], default-features = false }
+bdk_chain = { path = "../chain", version = "0.9.0", features = ["miniscript", "serde"], default-features = false }
 
 # Optional dependencies
 bip39 = { version = "2.0", optional = true }

crates/bdk/src/descriptor/checksum.rs

@@ -42,22 +42,16 @@ fn poly_mod(mut c: u64, val: u64) -> u64 {
     c
 }
 
-/// Computes the checksum bytes of a descriptor.
-/// `exclude_hash = true` ignores all data after the first '#' (inclusive).
-pub(crate) fn calc_checksum_bytes_internal(
-    mut desc: &str,
-    exclude_hash: bool,
-) -> Result<[u8; 8], DescriptorError> {
+/// Compute the checksum bytes of a descriptor, excludes any existing checksum in the descriptor string from the calculation
+pub fn calc_checksum_bytes(mut desc: &str) -> Result<[u8; 8], DescriptorError> {
     let mut c = 1;
     let mut cls = 0;
     let mut clscount = 0;
 
     let mut original_checksum = None;
-    if exclude_hash {
-        if let Some(split) = desc.split_once('#') {
-            desc = split.0;
-            original_checksum = Some(split.1);
-        }
+    if let Some(split) = desc.split_once('#') {
+        desc = split.0;
+        original_checksum = Some(split.1);
     }
 
     for ch in desc.as_bytes() {
@@ -95,39 +89,10 @@ pub(crate) fn calc_checksum_bytes_internal(
     Ok(checksum)
 }
 
-/// Compute the checksum bytes of a descriptor, excludes any existing checksum in the descriptor string from the calculation
-pub fn calc_checksum_bytes(desc: &str) -> Result<[u8; 8], DescriptorError> {
-    calc_checksum_bytes_internal(desc, true)
-}
-
 /// Compute the checksum of a descriptor, excludes any existing checksum in the descriptor string from the calculation
 pub fn calc_checksum(desc: &str) -> Result<String, DescriptorError> {
     // unsafe is okay here as the checksum only uses bytes in `CHECKSUM_CHARSET`
-    calc_checksum_bytes_internal(desc, true)
-        .map(|b| unsafe { String::from_utf8_unchecked(b.to_vec()) })
-}
-
-// TODO in release 0.25.0, remove get_checksum_bytes and get_checksum
-// TODO in release 0.25.0, consolidate calc_checksum_bytes_internal into calc_checksum_bytes
-
-/// Compute the checksum bytes of a descriptor
-#[deprecated(
-    since = "0.24.0",
-    note = "Use new `calc_checksum_bytes` function which excludes any existing checksum in the descriptor string before calculating the checksum hash bytes. See https://github.com/bitcoindevkit/bdk/pull/765."
-)]
-pub fn get_checksum_bytes(desc: &str) -> Result<[u8; 8], DescriptorError> {
-    calc_checksum_bytes_internal(desc, false)
-}
-
-/// Compute the checksum of a descriptor
-#[deprecated(
-    since = "0.24.0",
-    note = "Use new `calc_checksum` function which excludes any existing checksum in the descriptor string before calculating the checksum hash. See https://github.com/bitcoindevkit/bdk/pull/765."
-)]
-pub fn get_checksum(desc: &str) -> Result<String, DescriptorError> {
-    // unsafe is okay here as the checksum only uses bytes in `CHECKSUM_CHARSET`
-    calc_checksum_bytes_internal(desc, false)
-        .map(|b| unsafe { String::from_utf8_unchecked(b.to_vec()) })
+    calc_checksum_bytes(desc).map(|b| unsafe { String::from_utf8_unchecked(b.to_vec()) })
 }
 
 #[cfg(test)]

crates/bdk/src/wallet/coin_selection.rs

@@ -100,6 +100,7 @@
 //! # Ok::<(), anyhow::Error>(())
 //! ```
 
+use crate::chain::collections::HashSet;
 use crate::types::FeeRate;
 use crate::wallet::utils::IsDust;
 use crate::Utxo;
@@ -107,6 +108,7 @@ use crate::WeightedUtxo;
 
 use alloc::vec::Vec;
 use bitcoin::consensus::encode::serialize;
+use bitcoin::OutPoint;
 use bitcoin::{Script, Weight};
 
 use core::convert::TryInto;
@@ -711,6 +713,25 @@ impl BranchAndBoundCoinSelection {
     }
 }
 
+/// Remove duplicate UTXOs.
+///
+/// If a UTXO appears in both `required` and `optional`, the appearance in `required` is kept.
+pub(crate) fn filter_duplicates<I>(required: I, optional: I) -> (I, I)
+where
+    I: IntoIterator<Item = WeightedUtxo> + FromIterator<WeightedUtxo>,
+{
+    let mut visited = HashSet::<OutPoint>::new();
+    let required = required
+        .into_iter()
+        .filter(|utxo| visited.insert(utxo.utxo.outpoint()))
+        .collect::<I>();
+    let optional = optional
+        .into_iter()
+        .filter(|utxo| visited.insert(utxo.utxo.outpoint()))
+        .collect::<I>();
+    (required, optional)
+}
+
 #[cfg(test)]
 mod test {
     use assert_matches::assert_matches;
@@ -721,6 +742,7 @@ mod test {
 
     use super::*;
     use crate::types::*;
+    use crate::wallet::coin_selection::filter_duplicates;
     use crate::wallet::Vbytes;
 
     use rand::rngs::StdRng;
@@ -799,13 +821,14 @@ mod test {
 
     fn generate_random_utxos(rng: &mut StdRng, utxos_number: usize) -> Vec<WeightedUtxo> {
         let mut res = Vec::new();
-        for _ in 0..utxos_number {
+        for i in 0..utxos_number {
             res.push(WeightedUtxo {
                 satisfaction_weight: P2WPKH_SATISFACTION_SIZE,
                 utxo: Utxo::Local(LocalOutput {
-                    outpoint: OutPoint::from_str(
-                        "ebd9813ecebc57ff8f30797de7c205e3c7498ca950ea4341ee51a685ff2fa30a:0",
-                    )
+                    outpoint: OutPoint::from_str(&format!(
+                        "ebd9813ecebc57ff8f30797de7c205e3c7498ca950ea4341ee51a685ff2fa30a:{}",
+                        i
+                    ))
                     .unwrap(),
                     txout: TxOut {
                         value: rng.gen_range(0..200000000),
@@ -829,24 +852,26 @@ mod test {
     }
 
     fn generate_same_value_utxos(utxos_value: u64, utxos_number: usize) -> Vec<WeightedUtxo> {
-        let utxo = WeightedUtxo {
-            satisfaction_weight: P2WPKH_SATISFACTION_SIZE,
-            utxo: Utxo::Local(LocalOutput {
-                outpoint: OutPoint::from_str(
-                    "ebd9813ecebc57ff8f30797de7c205e3c7498ca950ea4341ee51a685ff2fa30a:0",
-                )
-                .unwrap(),
-                txout: TxOut {
-                    value: utxos_value,
-                    script_pubkey: ScriptBuf::new(),
-                },
-                keychain: KeychainKind::External,
-                is_spent: false,
-                derivation_index: 42,
-                confirmation_time: ConfirmationTime::Unconfirmed { last_seen: 0 },
-            }),
-        };
-        vec![utxo; utxos_number]
+        (0..utxos_number)
+            .map(|i| WeightedUtxo {
+                satisfaction_weight: P2WPKH_SATISFACTION_SIZE,
+                utxo: Utxo::Local(LocalOutput {
+                    outpoint: OutPoint::from_str(&format!(
+                        "ebd9813ecebc57ff8f30797de7c205e3c7498ca950ea4341ee51a685ff2fa30a:{}",
+                        i
+                    ))
+                    .unwrap(),
+                    txout: TxOut {
+                        value: utxos_value,
+                        script_pubkey: ScriptBuf::new(),
+                    },
+                    keychain: KeychainKind::External,
+                    is_spent: false,
+                    derivation_index: 42,
+                    confirmation_time: ConfirmationTime::Unconfirmed { last_seen: 0 },
+                }),
+            })
+            .collect()
     }
 
     fn sum_random_utxos(mut rng: &mut StdRng, utxos: &mut Vec<WeightedUtxo>) -> u64 {
@@ -1478,4 +1503,95 @@ mod test {
             })
         );
     }
+
+    #[test]
+    fn test_filter_duplicates() {
+        fn utxo(txid: &str, value: u64) -> WeightedUtxo {
+            WeightedUtxo {
+                satisfaction_weight: 0,
+                utxo: Utxo::Local(LocalOutput {
+                    outpoint: OutPoint::new(bitcoin::hashes::Hash::hash(txid.as_bytes()), 0),
+                    txout: TxOut {
+                        value,
+                        script_pubkey: ScriptBuf::new(),
+                    },
+                    keychain: KeychainKind::External,
+                    is_spent: false,
+                    derivation_index: 0,
+                    confirmation_time: ConfirmationTime::Confirmed {
+                        height: 12345,
+                        time: 12345,
+                    },
+                }),
+            }
+        }
+
+        fn to_utxo_vec(utxos: &[(&str, u64)]) -> Vec<WeightedUtxo> {
+            let mut v = utxos
+                .iter()
+                .map(|&(txid, value)| utxo(txid, value))
+                .collect::<Vec<_>>();
+            v.sort_by_key(|u| u.utxo.outpoint());
+            v
+        }
+
+        struct TestCase<'a> {
+            name: &'a str,
+            required: &'a [(&'a str, u64)],
+            optional: &'a [(&'a str, u64)],
+            exp_required: &'a [(&'a str, u64)],
+            exp_optional: &'a [(&'a str, u64)],
+        }
+
+        let test_cases = [
+            TestCase {
+                name: "no_duplicates",
+                required: &[("A", 1000), ("B", 2100)],
+                optional: &[("C", 1000)],
+                exp_required: &[("A", 1000), ("B", 2100)],
+                exp_optional: &[("C", 1000)],
+            },
+            TestCase {
+                name: "duplicate_required_utxos",
+                required: &[("A", 3000), ("B", 1200), ("C", 1234), ("A", 3000)],
+                optional: &[("D", 2100)],
+                exp_required: &[("A", 3000), ("B", 1200), ("C", 1234)],
+                exp_optional: &[("D", 2100)],
+            },
+            TestCase {
+                name: "duplicate_optional_utxos",
+                required: &[("A", 3000), ("B", 1200)],
+                optional: &[("C", 5000), ("D", 1300), ("C", 5000)],
+                exp_required: &[("A", 3000), ("B", 1200)],
+                exp_optional: &[("C", 5000), ("D", 1300)],
+            },
+            TestCase {
+                name: "duplicate_across_required_and_optional_utxos",
+                required: &[("A", 3000), ("B", 1200), ("C", 2100)],
+                optional: &[("A", 3000), ("D", 1200), ("E", 5000)],
+                exp_required: &[("A", 3000), ("B", 1200), ("C", 2100)],
+                exp_optional: &[("D", 1200), ("E", 5000)],
+            },
+        ];
+
+        for (i, t) in test_cases.into_iter().enumerate() {
+            println!("Case {}: {}", i, t.name);
+            let (required, optional) =
+                filter_duplicates(to_utxo_vec(t.required), to_utxo_vec(t.optional));
+            assert_eq!(
+                required,
+                to_utxo_vec(t.exp_required),
+                "[{}:{}] unexpected `required` result",
+                i,
+                t.name
+            );
+            assert_eq!(
+                optional,
+                to_utxo_vec(t.exp_optional),
+                "[{}:{}] unexpected `optional` result",
+                i,
+                t.name
+            );
+        }
+    }
 }

crates/bdk/src/wallet/mod.rs

@@ -23,7 +23,9 @@ pub use bdk_chain::keychain::Balance;
 use bdk_chain::{
     indexed_tx_graph,
     keychain::{self, KeychainTxOutIndex},
-    local_chain::{self, CannotConnectError, CheckPoint, CheckPointIter, LocalChain},
+    local_chain::{
+        self, ApplyHeaderError, CannotConnectError, CheckPoint, CheckPointIter, LocalChain,
+    },
     tx_graph::{CanonicalTx, TxGraph},
     Append, BlockId, ChainPosition, ConfirmationTime, ConfirmationTimeHeightAnchor, FullTxOut,
     IndexedTxGraph, Persist, PersistBackend,
@@ -31,8 +33,8 @@ use bdk_chain::{
 use bitcoin::secp256k1::{All, Secp256k1};
 use bitcoin::sighash::{EcdsaSighashType, TapSighashType};
 use bitcoin::{
-    absolute, Address, Network, OutPoint, Script, ScriptBuf, Sequence, Transaction, TxOut, Txid,
-    Weight, Witness,
+    absolute, Address, Block, Network, OutPoint, Script, ScriptBuf, Sequence, Transaction, TxOut,
+    Txid, Weight, Witness,
 };
 use bitcoin::{consensus::encode::serialize, BlockHash};
 use bitcoin::{constants::genesis_block, psbt};
@@ -438,6 +440,55 @@ pub enum InsertTxError {
     },
 }
 
+impl fmt::Display for InsertTxError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            InsertTxError::ConfirmationHeightCannotBeGreaterThanTip {
+                tip_height,
+                tx_height,
+            } => {
+                write!(f, "cannot insert tx with confirmation height ({}) higher than internal tip height ({})", tx_height, tip_height)
+            }
+        }
+    }
+}
+
+#[cfg(feature = "std")]
+impl std::error::Error for InsertTxError {}
+
+/// An error that may occur when applying a block to [`Wallet`].
+#[derive(Debug)]
+pub enum ApplyBlockError {
+    /// Occurs when the update chain cannot connect with original chain.
+    CannotConnect(CannotConnectError),
+    /// Occurs when the `connected_to` hash does not match the hash derived from `block`.
+    UnexpectedConnectedToHash {
+        /// Block hash of `connected_to`.
+        connected_to_hash: BlockHash,
+        /// Expected block hash of `connected_to`, as derived from `block`.
+        expected_hash: BlockHash,
+    },
+}
+
+impl fmt::Display for ApplyBlockError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            ApplyBlockError::CannotConnect(err) => err.fmt(f),
+            ApplyBlockError::UnexpectedConnectedToHash {
+                expected_hash: block_hash,
+                connected_to_hash: checkpoint_hash,
+            } => write!(
+                f,
+                "`connected_to` hash {} differs from the expected hash {} (which is derived from `block`)",
+                checkpoint_hash, block_hash
+            ),
+        }
+    }
+}
+
+#[cfg(feature = "std")]
+impl std::error::Error for ApplyBlockError {}
+
 impl<D> Wallet<D> {
     /// Initialize an empty [`Wallet`].
     pub fn new<E: IntoWalletDescriptor>(
@@ -941,12 +992,11 @@ impl<D> Wallet<D> {
         })
     }
 
-    /// Computes total input value going from script pubkeys in the index (sent) and the total output
-    /// value going to script pubkeys in the index (received) in `tx`.
+    /// Compute the `tx`'s sent and received amounts (in satoshis).
     ///
-    /// For the `sent` to be computed correctly, the outputs being spent must have already been
-    /// scanned by the index. Calculating received just uses the [`Transaction`] outputs directly,
-    /// so it will be correct even if it has not been scanned.
+    /// This method returns a tuple `(sent, received)`. Sent is the sum of the txin amounts
+    /// that spend from previous txouts tracked by this wallet. Received is the summation
+    /// of this tx's outputs that send to script pubkeys tracked by this wallet.
     ///
     /// # Examples
     ///
@@ -1496,6 +1546,9 @@ impl<D> Wallet<D> {
             }
         };
 
+        let (required_utxos, optional_utxos) =
+            coin_selection::filter_duplicates(required_utxos, optional_utxos);
+
         let coin_selection = coin_selection.coin_select(
             required_utxos,
             optional_utxos,
@@ -2329,7 +2382,7 @@ impl<D> Wallet<D> {
         self.persist.commit().map(|c| c.is_some())
     }
 
-    /// Returns the changes that will be staged with the next call to [`commit`].
+    /// Returns the changes that will be committed with the next call to [`commit`].
     ///
     /// [`commit`]: Self::commit
     pub fn staged(&self) -> &ChangeSet
@@ -2353,6 +2406,86 @@ impl<D> Wallet<D> {
     pub fn local_chain(&self) -> &LocalChain {
         &self.chain
     }
+
+    /// Introduces a `block` of `height` to the wallet, and tries to connect it to the
+    /// `prev_blockhash` of the block's header.
+    ///
+    /// This is a convenience method that is equivalent to calling [`apply_block_connected_to`]
+    /// with `prev_blockhash` and `height-1` as the `connected_to` parameter.
+    ///
+    /// [`apply_block_connected_to`]: Self::apply_block_connected_to
+    pub fn apply_block(&mut self, block: &Block, height: u32) -> Result<(), CannotConnectError>
+    where
+        D: PersistBackend<ChangeSet>,
+    {
+        let connected_to = match height.checked_sub(1) {
+            Some(prev_height) => BlockId {
+                height: prev_height,
+                hash: block.header.prev_blockhash,
+            },
+            None => BlockId {
+                height,
+                hash: block.block_hash(),
+            },
+        };
+        self.apply_block_connected_to(block, height, connected_to)
+            .map_err(|err| match err {
+                ApplyHeaderError::InconsistentBlocks => {
+                    unreachable!("connected_to is derived from the block so must be consistent")
+                }
+                ApplyHeaderError::CannotConnect(err) => err,
+            })
+    }
+
+    /// Applies relevant transactions from `block` of `height` to the wallet, and connects the
+    /// block to the internal chain.
+    ///
+    /// The `connected_to` parameter informs the wallet how this block connects to the internal
+    /// [`LocalChain`]. Relevant transactions are filtered from the `block` and inserted into the
+    /// internal [`TxGraph`].
+    pub fn apply_block_connected_to(
+        &mut self,
+        block: &Block,
+        height: u32,
+        connected_to: BlockId,
+    ) -> Result<(), ApplyHeaderError>
+    where
+        D: PersistBackend<ChangeSet>,
+    {
+        let mut changeset = ChangeSet::default();
+        changeset.append(
+            self.chain
+                .apply_header_connected_to(&block.header, height, connected_to)?
+                .into(),
+        );
+        changeset.append(
+            self.indexed_graph
+                .apply_block_relevant(block, height)
+                .into(),
+        );
+        self.persist.stage(changeset);
+        Ok(())
+    }
+
+    /// Apply relevant unconfirmed transactions to the wallet.
+    ///
+    /// Transactions that are not relevant are filtered out.
+    ///
+    /// This method takes in an iterator of `(tx, last_seen)` where `last_seen` is the timestamp of
+    /// when the transaction was last seen in the mempool. This is used for conflict resolution
+    /// when there is conflicting unconfirmed transactions. The transaction with the later
+    /// `last_seen` is prioritied.
+    pub fn apply_unconfirmed_txs<'t>(
+        &mut self,
+        unconfirmed_txs: impl IntoIterator<Item = (&'t Transaction, u64)>,
+    ) where
+        D: PersistBackend<ChangeSet>,
+    {
+        let indexed_graph_changeset = self
+            .indexed_graph
+            .batch_insert_relevant_unconfirmed(unconfirmed_txs);
+        self.persist.stage(ChangeSet::from(indexed_graph_changeset));
+    }
 }
 
 impl<D> AsRef<bdk_chain::tx_graph::TxGraph<ConfirmationTimeHeightAnchor>> for Wallet<D> {

crates/bitcoind_rpc/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "bdk_bitcoind_rpc"
-version = "0.3.0"
+version = "0.4.0"
 edition = "2021"
 rust-version = "1.63"
 homepage = "https://bitcoindevkit.org"
@@ -16,7 +16,7 @@ readme = "README.md"
 # For no-std, remember to enable the bitcoin/no-std feature
 bitcoin = { version = "0.30", default-features = false }
 bitcoincore-rpc = { version = "0.17" }
-bdk_chain = { path = "../chain", version = "0.8", default-features = false }
+bdk_chain = { path = "../chain", version = "0.9", default-features = false }
 
 [dev-dependencies]
 bitcoind = { version = "0.33", features = ["25_0"] }

crates/bitcoind_rpc/src/lib.rs

@@ -43,11 +43,13 @@ pub struct Emitter<'c, C> {
 }
 
 impl<'c, C: bitcoincore_rpc::RpcApi> Emitter<'c, C> {
-    /// Construct a new [`Emitter`] with the given RPC `client`, `last_cp` and `start_height`.
+    /// Construct a new [`Emitter`].
     ///
-    /// * `last_cp` is the check point used to find the latest block which is still part of the best
-    ///   chain.
-    /// * `start_height` is the block height to start emitting blocks from.
+    /// `last_cp` informs the emitter of the chain we are starting off with. This way, the emitter
+    /// can start emission from a block that connects to the original chain.
+    ///
+    /// `start_height` starts emission from a given height (if there are no conflicts with the
+    /// original chain).
     pub fn new(client: &'c C, last_cp: CheckPoint, start_height: u32) -> Self {
         Self {
             client,
@@ -127,13 +129,58 @@ impl<'c, C: bitcoincore_rpc::RpcApi> Emitter<'c, C> {
     }
 
     /// Emit the next block height and header (if any).
-    pub fn next_header(&mut self) -> Result<Option<(u32, Header)>, bitcoincore_rpc::Error> {
-        poll(self, |hash| self.client.get_block_header(hash))
+    pub fn next_header(&mut self) -> Result<Option<BlockEvent<Header>>, bitcoincore_rpc::Error> {
+        Ok(poll(self, |hash| self.client.get_block_header(hash))?
+            .map(|(checkpoint, block)| BlockEvent { block, checkpoint }))
     }
 
     /// Emit the next block height and block (if any).
-    pub fn next_block(&mut self) -> Result<Option<(u32, Block)>, bitcoincore_rpc::Error> {
-        poll(self, |hash| self.client.get_block(hash))
+    pub fn next_block(&mut self) -> Result<Option<BlockEvent<Block>>, bitcoincore_rpc::Error> {
+        Ok(poll(self, |hash| self.client.get_block(hash))?
+            .map(|(checkpoint, block)| BlockEvent { block, checkpoint }))
+    }
+}
+
+/// A newly emitted block from [`Emitter`].
+#[derive(Debug)]
+pub struct BlockEvent<B> {
+    /// Either a full [`Block`] or [`Header`] of the new block.
+    pub block: B,
+
+    /// The checkpoint of the new block.
+    ///
+    /// A [`CheckPoint`] is a node of a linked list of [`BlockId`]s. This checkpoint is linked to
+    /// all [`BlockId`]s originally passed in [`Emitter::new`] as well as emitted blocks since then.
+    /// These blocks are guaranteed to be of the same chain.
+    ///
+    /// This is important as BDK structures require block-to-apply to be connected with another
+    /// block in the original chain.
+    pub checkpoint: CheckPoint,
+}
+
+impl<B> BlockEvent<B> {
+    /// The block height of this new block.
+    pub fn block_height(&self) -> u32 {
+        self.checkpoint.height()
+    }
+
+    /// The block hash of this new block.
+    pub fn block_hash(&self) -> BlockHash {
+        self.checkpoint.hash()
+    }
+
+    /// The [`BlockId`] of a previous block that this block connects to.
+    ///
+    /// This either returns a [`BlockId`] of a previously emitted block or from the chain we started
+    /// with (passed in as `last_cp` in [`Emitter::new`]).
+    ///
+    /// This value is derived from [`BlockEvent::checkpoint`].
+    pub fn connected_to(&self) -> BlockId {
+        match self.checkpoint.prev() {
+            Some(prev_cp) => prev_cp.block_id(),
+            // there is no previous checkpoint, so just connect with itself
+            None => self.checkpoint.block_id(),
+        }
     }
 }
 
@@ -203,7 +250,7 @@ where
 fn poll<C, V, F>(
     emitter: &mut Emitter<C>,
     get_item: F,
-) -> Result<Option<(u32, V)>, bitcoincore_rpc::Error>
+) -> Result<Option<(CheckPoint, V)>, bitcoincore_rpc::Error>
 where
     C: bitcoincore_rpc::RpcApi,
     F: Fn(&BlockHash) -> Result<V, bitcoincore_rpc::Error>,
@@ -215,13 +262,14 @@ where
                 let hash = res.hash;
                 let item = get_item(&hash)?;
 
-                emitter.last_cp = emitter
+                let new_cp = emitter
                     .last_cp
                     .clone()
                     .push(BlockId { height, hash })
                     .expect("must push");
+                emitter.last_cp = new_cp.clone();
                 emitter.last_block = Some(res);
-                return Ok(Some((height, item)));
+                return Ok(Some((new_cp, item)));
             }
             PollResponse::NoMoreBlocks => {
                 emitter.last_block = None;

crates/bitcoind_rpc/tests/test_emitter.rs

@@ -157,28 +157,6 @@ impl TestEnv {
     }
 }
 
-fn block_to_chain_update(block: &bitcoin::Block, height: u32) -> local_chain::Update {
-    let this_id = BlockId {
-        height,
-        hash: block.block_hash(),
-    };
-    let tip = if block.header.prev_blockhash == BlockHash::all_zeros() {
-        CheckPoint::new(this_id)
-    } else {
-        CheckPoint::new(BlockId {
-            height: height - 1,
-            hash: block.header.prev_blockhash,
-        })
-        .extend(core::iter::once(this_id))
-        .expect("must construct checkpoint")
-    };
-
-    local_chain::Update {
-        tip,
-        introduce_older_blocks: false,
-    }
-}
-
 /// Ensure that blocks are emitted in order even after reorg.
 ///
 /// 1. Mine 101 blocks.
@@ -200,17 +178,21 @@ pub fn test_sync_local_chain() -> anyhow::Result<()> {
 
     // see if the emitter outputs the right blocks
     println!("first sync:");
-    while let Some((height, block)) = emitter.next_block()? {
+    while let Some(emission) = emitter.next_block()? {
+        let height = emission.block_height();
+        let hash = emission.block_hash();
         assert_eq!(
-            block.block_hash(),
+            emission.block_hash(),
             exp_hashes[height as usize],
             "emitted block hash is unexpected"
         );
 
-        let chain_update = block_to_chain_update(&block, height);
         assert_eq!(
-            local_chain.apply_update(chain_update)?,
-            BTreeMap::from([(height, Some(block.block_hash()))]),
+            local_chain.apply_update(local_chain::Update {
+                tip: emission.checkpoint,
+                introduce_older_blocks: false,
+            })?,
+            BTreeMap::from([(height, Some(hash))]),
             "chain update changeset is unexpected",
         );
     }
@@ -237,27 +219,30 @@ pub fn test_sync_local_chain() -> anyhow::Result<()> {
     // see if the emitter outputs the right blocks
     println!("after reorg:");
     let mut exp_height = exp_hashes.len() - reorged_blocks.len();
-    while let Some((height, block)) = emitter.next_block()? {
+    while let Some(emission) = emitter.next_block()? {
+        let height = emission.block_height();
+        let hash = emission.block_hash();
         assert_eq!(
             height, exp_height as u32,
             "emitted block has unexpected height"
         );
 
         assert_eq!(
-            block.block_hash(),
-            exp_hashes[height as usize],
+            hash, exp_hashes[height as usize],
             "emitted block is unexpected"
         );
 
-        let chain_update = block_to_chain_update(&block, height);
         assert_eq!(
-            local_chain.apply_update(chain_update)?,
+            local_chain.apply_update(local_chain::Update {
+                tip: emission.checkpoint,
+                introduce_older_blocks: false,
+            })?,
             if exp_height == exp_hashes.len() - reorged_blocks.len() {
-                core::iter::once((height, Some(block.block_hash())))
+                core::iter::once((height, Some(hash)))
                     .chain((height + 1..exp_hashes.len() as u32).map(|h| (h, None)))
                     .collect::<bdk_chain::local_chain::ChangeSet>()
             } else {
-                BTreeMap::from([(height, Some(block.block_hash()))])
+                BTreeMap::from([(height, Some(hash))])
             },
             "chain update changeset is unexpected",
         );
@@ -307,9 +292,13 @@ fn test_into_tx_graph() -> anyhow::Result<()> {
 
     let emitter = &mut Emitter::new(&env.client, chain.tip(), 0);
 
-    while let Some((height, block)) = emitter.next_block()? {
-        let _ = chain.apply_update(block_to_chain_update(&block, height))?;
-        let indexed_additions = indexed_tx_graph.apply_block_relevant(block, height);
+    while let Some(emission) = emitter.next_block()? {
+        let height = emission.block_height();
+        let _ = chain.apply_update(local_chain::Update {
+            tip: emission.checkpoint,
+            introduce_older_blocks: false,
+        })?;
+        let indexed_additions = indexed_tx_graph.apply_block_relevant(&emission.block, height);
         assert!(indexed_additions.is_empty());
     }
 
@@ -367,10 +356,13 @@ fn test_into_tx_graph() -> anyhow::Result<()> {
 
     // must receive mined block which will confirm the transactions.
     {
-        let (height, block) = emitter.next_block()?.expect("must get mined block");
-        let _ = chain
-            .apply_update(CheckPoint::from_header(&block.header, height).into_update(false))?;
-        let indexed_additions = indexed_tx_graph.apply_block_relevant(block, height);
+        let emission = emitter.next_block()?.expect("must get mined block");
+        let height = emission.block_height();
+        let _ = chain.apply_update(local_chain::Update {
+            tip: emission.checkpoint,
+            introduce_older_blocks: false,
+        })?;
+        let indexed_additions = indexed_tx_graph.apply_block_relevant(&emission.block, height);
         assert!(indexed_additions.graph.txs.is_empty());
         assert!(indexed_additions.graph.txouts.is_empty());
         assert_eq!(indexed_additions.graph.anchors, exp_anchors);
@@ -407,9 +399,12 @@ fn ensure_block_emitted_after_reorg_is_at_reorg_height() -> anyhow::Result<()> {
 
     for reorg_count in 1..=10 {
         let replaced_blocks = env.reorg_empty_blocks(reorg_count)?;
-        let (height, next_header) = emitter.next_header()?.expect("must emit block after reorg");
+        let next_emission = emitter.next_header()?.expect("must emit block after reorg");
         assert_eq!(
-            (height as usize, next_header.block_hash()),
+            (
+                next_emission.block_height() as usize,
+                next_emission.block_hash()
+            ),
             replaced_blocks[0],
             "block emitted after reorg should be at the reorg height"
         );
@@ -439,8 +434,9 @@ fn sync_from_emitter<C>(
 where
     C: bitcoincore_rpc::RpcApi,
 {
-    while let Some((height, block)) = emitter.next_block()? {
-        process_block(recv_chain, recv_graph, block, height)?;
+    while let Some(emission) = emitter.next_block()? {
+        let height = emission.block_height();
+        process_block(recv_chain, recv_graph, emission.block, height)?;
     }
     Ok(())
 }
@@ -660,7 +656,8 @@ fn mempool_re_emits_if_tx_introduction_height_not_reached() -> anyhow::Result<()
 
     // At this point, the emitter has seen all mempool transactions. It should only re-emit those
     // that have introduction heights less than the emitter's last-emitted block tip.
-    while let Some((height, _)) = emitter.next_header()? {
+    while let Some(emission) = emitter.next_header()? {
+        let height = emission.block_height();
         // We call `mempool()` twice.
         // The second call (at height `h`) should skip the tx introduced at height `h`.
         for try_index in 0..2 {
@@ -754,7 +751,8 @@ fn mempool_during_reorg() -> anyhow::Result<()> {
             .collect::<BTreeMap<_, _>>());
 
         // `next_header` emits the replacement block of the reorg
-        if let Some((height, _)) = emitter.next_header()? {
+        if let Some(emission) = emitter.next_header()? {
+            let height = emission.block_height();
             println!("\t- replacement height: {}", height);
 
             // the mempool emission (that follows the first block emission after reorg) should only
@@ -835,12 +833,12 @@ fn no_agreement_point() -> anyhow::Result<()> {
     env.mine_blocks(PREMINE_COUNT, None)?;
 
     // emit block 99a
-    let (_, block_header_99a) = emitter.next_header()?.expect("block 99a header");
+    let block_header_99a = emitter.next_header()?.expect("block 99a header").block;
     let block_hash_99a = block_header_99a.block_hash();
     let block_hash_98a = block_header_99a.prev_blockhash;
 
     // emit block 100a
-    let (_, block_header_100a) = emitter.next_header()?.expect("block 100a header");
+    let block_header_100a = emitter.next_header()?.expect("block 100a header").block;
     let block_hash_100a = block_header_100a.block_hash();
 
     // get hash for block 101a
@@ -855,7 +853,7 @@ fn no_agreement_point() -> anyhow::Result<()> {
     env.mine_blocks(3, None)?;
 
     // emit block header 99b
-    let (_, block_header_99b) = emitter.next_header()?.expect("block 99b header");
+    let block_header_99b = emitter.next_header()?.expect("block 99b header").block;
     let block_hash_99b = block_header_99b.block_hash();
     let block_hash_98b = block_header_99b.prev_blockhash;
 

crates/chain/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "bdk_chain"
-version = "0.8.0"
+version = "0.9.0"
 edition = "2021"
 rust-version = "1.63"
 homepage = "https://bitcoindevkit.org"

crates/chain/src/chain_data.rs

@@ -81,7 +81,7 @@ impl From<ChainPosition<ConfirmationTimeHeightAnchor>> for ConfirmationTime {
                 height: a.confirmation_height,
                 time: a.confirmation_time,
             },
-            ChainPosition::Unconfirmed(_) => Self::Unconfirmed { last_seen: 0 },
+            ChainPosition::Unconfirmed(last_seen) => Self::Unconfirmed { last_seen },
         }
     }
 }

crates/chain/src/indexed_tx_graph.rs

@@ -224,20 +224,26 @@ where
     /// Irrelevant transactions in `txs` will be ignored.
     pub fn apply_block_relevant(
         &mut self,
-        block: Block,
+        block: &Block,
         height: u32,
     ) -> ChangeSet<A, I::ChangeSet> {
         let block_id = BlockId {
             hash: block.block_hash(),
             height,
         };
-        let txs = block.txdata.iter().enumerate().map(|(tx_pos, tx)| {
-            (
-                tx,
-                core::iter::once(A::from_block_position(&block, block_id, tx_pos)),
-            )
-        });
-        self.batch_insert_relevant(txs)
+        let mut changeset = ChangeSet::<A, I::ChangeSet>::default();
+        for (tx_pos, tx) in block.txdata.iter().enumerate() {
+            changeset.indexer.append(self.index.index_tx(tx));
+            if self.index.is_tx_relevant(tx) {
+                let txid = tx.txid();
+                let anchor = A::from_block_position(block, block_id, tx_pos);
+                changeset.graph.append(self.graph.insert_tx(tx.clone()));
+                changeset
+                    .graph
+                    .append(self.graph.insert_anchor(txid, anchor));
+            }
+        }
+        changeset
     }
 
     /// Batch insert all transactions of the given `block` of `height`.

crates/chain/src/keychain.rs

@@ -20,7 +20,7 @@ pub use txout_index::*;
 /// Represents updates to the derivation index of a [`KeychainTxOutIndex`].
 /// It maps each keychain `K` to its last revealed index.
 ///
-/// It can be applied to [`KeychainTxOutIndex`] with [`apply_changeset`]. [`ChangeSet] are
+/// It can be applied to [`KeychainTxOutIndex`] with [`apply_changeset`]. [`ChangeSet`]s are
 /// monotone in that they will never decrease the revealed derivation index.
 ///
 /// [`KeychainTxOutIndex`]: crate::keychain::KeychainTxOutIndex

crates/chain/src/keychain/txout_index.rs

@@ -13,7 +13,7 @@ use core::{
 
 use crate::Append;
 
-const DEFAULT_LOOKAHEAD: u32 = 1_000;
+const DEFAULT_LOOKAHEAD: u32 = 25;
 
 /// [`KeychainTxOutIndex`] controls how script pubkeys are revealed for multiple keychains, and
 /// indexes [`TxOut`]s with them.

crates/chain/src/local_chain.rs

@@ -5,6 +5,7 @@ use core::convert::Infallible;
 use crate::collections::BTreeMap;
 use crate::{BlockId, ChainOracle};
 use alloc::sync::Arc;
+use bitcoin::block::Header;
 use bitcoin::BlockHash;
 
 /// The [`ChangeSet`] represents changes to [`LocalChain`].
@@ -39,6 +40,28 @@ impl CheckPoint {
         Self(Arc::new(CPInner { block, prev: None }))
     }
 
+    /// Construct a checkpoint from a list of [`BlockId`]s in ascending height order.
+    ///
+    /// # Errors
+    ///
+    /// This method will error if any of the follow occurs:
+    ///
+    /// - The `blocks` iterator is empty, in which case, the error will be `None`.
+    /// - The `blocks` iterator is not in ascending height order.
+    /// - The `blocks` iterator contains multiple [`BlockId`]s of the same height.
+    ///
+    /// The error type is the last successful checkpoint constructed (if any).
+    pub fn from_block_ids(
+        block_ids: impl IntoIterator<Item = BlockId>,
+    ) -> Result<Self, Option<Self>> {
+        let mut blocks = block_ids.into_iter();
+        let mut acc = CheckPoint::new(blocks.next().ok_or(None)?);
+        for id in blocks {
+            acc = acc.push(id).map_err(Some)?;
+        }
+        Ok(acc)
+    }
+
     /// Construct a checkpoint from the given `header` and block `height`.
     ///
     /// If `header` is of the genesis block, the checkpoint won't have a [`prev`] node. Otherwise,
@@ -347,6 +370,95 @@ impl LocalChain {
         Ok(changeset)
     }
 
+    /// Update the chain with a given [`Header`] at `height` which you claim is connected to a existing block in the chain.
+    ///
+    /// This is useful when you have a block header that you want to record as part of the chain but
+    /// don't necessarily know that the `prev_blockhash` is in the chain.
+    ///
+    /// This will usually insert two new [`BlockId`]s into the chain: the header's block and the
+    /// header's `prev_blockhash` block. `connected_to` must already be in the chain but is allowed
+    /// to be `prev_blockhash` (in which case only one new block id will be inserted).
+    /// To be successful, `connected_to` must be chosen carefully so that `LocalChain`'s [update
+    /// rules][`apply_update`] are satisfied.
+    ///
+    /// # Errors
+    ///
+    /// [`ApplyHeaderError::InconsistentBlocks`] occurs if the `connected_to` block and the
+    /// [`Header`] is inconsistent. For example, if the `connected_to` block is the same height as
+    /// `header` or `prev_blockhash`, but has a different block hash. Or if the `connected_to`
+    /// height is greater than the header's `height`.
+    ///
+    /// [`ApplyHeaderError::CannotConnect`] occurs if the internal call to [`apply_update`] fails.
+    ///
+    /// [`apply_update`]: Self::apply_update
+    pub fn apply_header_connected_to(
+        &mut self,
+        header: &Header,
+        height: u32,
+        connected_to: BlockId,
+    ) -> Result<ChangeSet, ApplyHeaderError> {
+        let this = BlockId {
+            height,
+            hash: header.block_hash(),
+        };
+        let prev = height.checked_sub(1).map(|prev_height| BlockId {
+            height: prev_height,
+            hash: header.prev_blockhash,
+        });
+        let conn = match connected_to {
+            // `connected_to` can be ignored if same as `this` or `prev` (duplicate)
+            conn if conn == this || Some(conn) == prev => None,
+            // this occurs if:
+            // - `connected_to` height is the same as `prev`, but different hash
+            // - `connected_to` height is the same as `this`, but different hash
+            // - `connected_to` height is greater than `this` (this is not allowed)
+            conn if conn.height >= height.saturating_sub(1) => {
+                return Err(ApplyHeaderError::InconsistentBlocks)
+            }
+            conn => Some(conn),
+        };
+
+        let update = Update {
+            tip: CheckPoint::from_block_ids([conn, prev, Some(this)].into_iter().flatten())
+                .expect("block ids must be in order"),
+            introduce_older_blocks: false,
+        };
+
+        self.apply_update(update)
+            .map_err(ApplyHeaderError::CannotConnect)
+    }
+
+    /// Update the chain with a given [`Header`] connecting it with the previous block.
+    ///
+    /// This is a convenience method to call [`apply_header_connected_to`] with the `connected_to`
+    /// parameter being `height-1:prev_blockhash`. If there is no previous block (i.e. genesis), we
+    /// use the current block as `connected_to`.
+    ///
+    /// [`apply_header_connected_to`]: LocalChain::apply_header_connected_to
+    pub fn apply_header(
+        &mut self,
+        header: &Header,
+        height: u32,
+    ) -> Result<ChangeSet, CannotConnectError> {
+        let connected_to = match height.checked_sub(1) {
+            Some(prev_height) => BlockId {
+                height: prev_height,
+                hash: header.prev_blockhash,
+            },
+            None => BlockId {
+                height,
+                hash: header.block_hash(),
+            },
+        };
+        self.apply_header_connected_to(header, height, connected_to)
+            .map_err(|err| match err {
+                ApplyHeaderError::InconsistentBlocks => {
+                    unreachable!("connected_to is derived from the block so is always consistent")
+                }
+                ApplyHeaderError::CannotConnect(err) => err,
+            })
+    }
+
     /// Apply the given `changeset`.
     pub fn apply_changeset(&mut self, changeset: &ChangeSet) -> Result<(), MissingGenesisError> {
         if let Some(start_height) = changeset.keys().next().cloned() {
@@ -557,6 +669,30 @@ impl core::fmt::Display for CannotConnectError {
 #[cfg(feature = "std")]
 impl std::error::Error for CannotConnectError {}
 
+/// The error type for [`LocalChain::apply_header_connected_to`].
+#[derive(Debug, Clone, PartialEq)]
+pub enum ApplyHeaderError {
+    /// Occurs when `connected_to` block conflicts with either the current block or previous block.
+    InconsistentBlocks,
+    /// Occurs when the update cannot connect with the original chain.
+    CannotConnect(CannotConnectError),
+}
+
+impl core::fmt::Display for ApplyHeaderError {
+    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+        match self {
+            ApplyHeaderError::InconsistentBlocks => write!(
+                f,
+                "the `connected_to` block conflicts with either the current or previous block"
+            ),
+            ApplyHeaderError::CannotConnect(err) => core::fmt::Display::fmt(err, f),
+        }
+    }
+}
+
+#[cfg(feature = "std")]
+impl std::error::Error for ApplyHeaderError {}
+
 fn merge_chains(
     original_tip: CheckPoint,
     update_tip: CheckPoint,

crates/chain/src/persist.rs

@@ -55,6 +55,18 @@ where
             // if written successfully, take and return `self.stage`
             .map(|_| Some(core::mem::take(&mut self.stage)))
     }
+
+    /// Stages a new changeset and commits it (along with any other previously staged changes) to
+    /// the persistence backend
+    ///
+    /// Convience method for calling [`stage`] and then [`commit`].
+    ///
+    /// [`stage`]: Self::stage
+    /// [`commit`]: Self::commit
+    pub fn stage_and_commit(&mut self, changeset: C) -> Result<Option<C>, B::WriteError> {
+        self.stage(changeset);
+        self.commit()
+    }
 }
 
 /// A persistence backend for [`Persist`].

crates/chain/src/spk_iter.rs

@@ -43,18 +43,24 @@ impl<D> SpkIterator<D>
 where
     D: Borrow<Descriptor<DescriptorPublicKey>>,
 {
-    /// Creates a new script pubkey iterator starting at 0 from a descriptor.
+    /// Create a new script pubkey iterator from `descriptor`.
+    ///
+    /// This iterates from derivation index 0 and stops at index 0x7FFFFFFF (as specified in
+    /// BIP-32). Non-wildcard descriptors will only return one script pubkey at derivation index 0.
+    ///
+    /// Use [`new_with_range`](SpkIterator::new_with_range) to create an iterator with a specified
+    /// derivation index range.
     pub fn new(descriptor: D) -> Self {
         SpkIterator::new_with_range(descriptor, 0..=BIP32_MAX_INDEX)
     }
 
-    // Creates a new script pubkey iterator from a descriptor with a given range.
-    // If the descriptor doesn't have a wildcard, we shorten whichever range you pass in
-    // to have length <= 1. This means that if you pass in 0..0 or 0..1 the range will
-    // remain the same, but if you pass in 0..10, we'll shorten it to 0..1
-    // Also note that if the descriptor doesn't have a wildcard, passing in a range starting
-    // from n > 0, will return an empty iterator.
-    pub(crate) fn new_with_range<R>(descriptor: D, range: R) -> Self
+    /// Create a new script pubkey iterator from `descriptor` and a given `range`.
+    ///
+    /// Non-wildcard descriptors will only emit a single script pubkey (at derivation index 0).
+    /// Wildcard descriptors have an end-bound of 0x7FFFFFFF (inclusive).
+    ///
+    /// Refer to [`new`](SpkIterator::new) for more.
+    pub fn new_with_range<R>(descriptor: D, range: R) -> Self
     where
         R: RangeBounds<u32>,
     {
@@ -73,13 +79,6 @@ where
         // Because `end` is exclusive, we want the maximum value to be BIP32_MAX_INDEX + 1.
         end = end.min(BIP32_MAX_INDEX + 1);
 
-        if !descriptor.borrow().has_wildcard() {
-            // The length of the range should be at most 1
-            if end != start {
-                end = start + 1;
-            }
-        }
-
         Self {
             next_index: start,
             end,
@@ -250,6 +249,14 @@ mod test {
             SpkIterator::new_with_range(&no_wildcard_descriptor, 1..=2).next(),
             None
         );
+        assert_eq!(
+            SpkIterator::new_with_range(&no_wildcard_descriptor, 10..11).next(),
+            None
+        );
+        assert_eq!(
+            SpkIterator::new_with_range(&no_wildcard_descriptor, 10..=10).next(),
+            None
+        );
     }
 
     // The following dummy traits were created to test if SpkIterator is working properly.

crates/chain/src/tx_graph.rs

@@ -337,7 +337,7 @@ impl<A> TxGraph<A> {
 
     /// The transactions spending from this output.
     ///
-    /// `TxGraph` allows conflicting transactions within the graph. Obviously the transactions in
+    /// [`TxGraph`] allows conflicting transactions within the graph. Obviously the transactions in
     /// the returned set will never be in the same active-chain.
     pub fn outspends(&self, outpoint: OutPoint) -> &HashSet<Txid> {
         self.spends.get(&outpoint).unwrap_or(&self.empty_outspends)

crates/chain/tests/test_local_chain.rs

@@ -1,7 +1,11 @@
-use bdk_chain::local_chain::{
-    AlterCheckPointError, CannotConnectError, ChangeSet, LocalChain, MissingGenesisError, Update,
+use bdk_chain::{
+    local_chain::{
+        AlterCheckPointError, ApplyHeaderError, CannotConnectError, ChangeSet, CheckPoint,
+        LocalChain, MissingGenesisError, Update,
+    },
+    BlockId,
 };
-use bitcoin::BlockHash;
+use bitcoin::{block::Header, hashes::Hash, BlockHash};
 
 #[macro_use]
 mod common;
@@ -288,6 +292,27 @@ fn update_local_chain() {
                 ],
             },
         },
+        // Allow update that is shorter than original chain
+        //        | 0 | 1 | 2 | 3 | 4 | 5
+        // chain  | A       C   D   E   F
+        // update | A       C   D'
+        TestLocalChain {
+            name: "allow update that is shorter than original chain",
+            chain: local_chain![(0, h!("_")), (2, h!("C")), (3, h!("D")), (4, h!("E")), (5, h!("F"))],
+            update: chain_update![(0, h!("_")), (2, h!("C")), (3, h!("D'"))],
+            exp: ExpectedResult::Ok {
+                changeset: &[
+                    (3, Some(h!("D'"))),
+                    (4, None),
+                    (5, None),
+                ],
+                init_changeset: &[
+                    (0, Some(h!("_"))),
+                    (2, Some(h!("C"))),
+                    (3, Some(h!("D'"))),
+                ],
+            },
+        },
     ]
     .into_iter()
     .for_each(TestLocalChain::run);
@@ -423,3 +448,234 @@ fn local_chain_disconnect_from() {
         );
     }
 }
+
+#[test]
+fn checkpoint_from_block_ids() {
+    struct TestCase<'a> {
+        name: &'a str,
+        blocks: &'a [(u32, BlockHash)],
+        exp_result: Result<(), Option<(u32, BlockHash)>>,
+    }
+
+    let test_cases = [
+        TestCase {
+            name: "in_order",
+            blocks: &[(0, h!("A")), (1, h!("B")), (3, h!("D"))],
+            exp_result: Ok(()),
+        },
+        TestCase {
+            name: "with_duplicates",
+            blocks: &[(1, h!("B")), (2, h!("C")), (2, h!("C'"))],
+            exp_result: Err(Some((2, h!("C")))),
+        },
+        TestCase {
+            name: "not_in_order",
+            blocks: &[(1, h!("B")), (3, h!("D")), (2, h!("C"))],
+            exp_result: Err(Some((3, h!("D")))),
+        },
+        TestCase {
+            name: "empty",
+            blocks: &[],
+            exp_result: Err(None),
+        },
+        TestCase {
+            name: "single",
+            blocks: &[(21, h!("million"))],
+            exp_result: Ok(()),
+        },
+    ];
+
+    for (i, t) in test_cases.into_iter().enumerate() {
+        println!("running test case {}: '{}'", i, t.name);
+        let result = CheckPoint::from_block_ids(
+            t.blocks
+                .iter()
+                .map(|&(height, hash)| BlockId { height, hash }),
+        );
+        match t.exp_result {
+            Ok(_) => {
+                assert!(result.is_ok(), "[{}:{}] should be Ok", i, t.name);
+                let result_vec = {
+                    let mut v = result
+                        .unwrap()
+                        .into_iter()
+                        .map(|cp| (cp.height(), cp.hash()))
+                        .collect::<Vec<_>>();
+                    v.reverse();
+                    v
+                };
+                assert_eq!(
+                    &result_vec, t.blocks,
+                    "[{}:{}] not equal to original block ids",
+                    i, t.name
+                );
+            }
+            Err(exp_last) => {
+                assert!(result.is_err(), "[{}:{}] should be Err", i, t.name);
+                let err = result.unwrap_err();
+                assert_eq!(
+                    err.as_ref()
+                        .map(|last_cp| (last_cp.height(), last_cp.hash())),
+                    exp_last,
+                    "[{}:{}] error's last cp height should be {:?}, got {:?}",
+                    i,
+                    t.name,
+                    exp_last,
+                    err
+                );
+            }
+        }
+    }
+}
+
+#[test]
+fn local_chain_apply_header_connected_to() {
+    fn header_from_prev_blockhash(prev_blockhash: BlockHash) -> Header {
+        Header {
+            version: bitcoin::block::Version::default(),
+            prev_blockhash,
+            merkle_root: bitcoin::hash_types::TxMerkleNode::all_zeros(),
+            time: 0,
+            bits: bitcoin::CompactTarget::default(),
+            nonce: 0,
+        }
+    }
+
+    struct TestCase {
+        name: &'static str,
+        chain: LocalChain,
+        header: Header,
+        height: u32,
+        connected_to: BlockId,
+        exp_result: Result<Vec<(u32, Option<BlockHash>)>, ApplyHeaderError>,
+    }
+
+    let test_cases = [
+        {
+            let header = header_from_prev_blockhash(h!("A"));
+            let hash = header.block_hash();
+            let height = 2;
+            let connected_to = BlockId { height, hash };
+            TestCase {
+                name: "connected_to_self_header_applied_to_self",
+                chain: local_chain![(0, h!("_")), (height, hash)],
+                header,
+                height,
+                connected_to,
+                exp_result: Ok(vec![]),
+            }
+        },
+        {
+            let prev_hash = h!("A");
+            let prev_height = 1;
+            let header = header_from_prev_blockhash(prev_hash);
+            let hash = header.block_hash();
+            let height = prev_height + 1;
+            let connected_to = BlockId {
+                height: prev_height,
+                hash: prev_hash,
+            };
+            TestCase {
+                name: "connected_to_prev_header_applied_to_self",
+                chain: local_chain![(0, h!("_")), (prev_height, prev_hash)],
+                header,
+                height,
+                connected_to,
+                exp_result: Ok(vec![(height, Some(hash))]),
+            }
+        },
+        {
+            let header = header_from_prev_blockhash(BlockHash::all_zeros());
+            let hash = header.block_hash();
+            let height = 0;
+            let connected_to = BlockId { height, hash };
+            TestCase {
+                name: "genesis_applied_to_self",
+                chain: local_chain![(0, hash)],
+                header,
+                height,
+                connected_to,
+                exp_result: Ok(vec![]),
+            }
+        },
+        {
+            let header = header_from_prev_blockhash(h!("Z"));
+            let height = 10;
+            let hash = header.block_hash();
+            let prev_height = height - 1;
+            let prev_hash = header.prev_blockhash;
+            TestCase {
+                name: "connect_at_connected_to",
+                chain: local_chain![(0, h!("_")), (2, h!("B")), (3, h!("C"))],
+                header,
+                height: 10,
+                connected_to: BlockId {
+                    height: 3,
+                    hash: h!("C"),
+                },
+                exp_result: Ok(vec![(prev_height, Some(prev_hash)), (height, Some(hash))]),
+            }
+        },
+        {
+            let prev_hash = h!("A");
+            let prev_height = 1;
+            let header = header_from_prev_blockhash(prev_hash);
+            let connected_to = BlockId {
+                height: prev_height,
+                hash: h!("not_prev_hash"),
+            };
+            TestCase {
+                name: "inconsistent_prev_hash",
+                chain: local_chain![(0, h!("_")), (prev_height, h!("not_prev_hash"))],
+                header,
+                height: prev_height + 1,
+                connected_to,
+                exp_result: Err(ApplyHeaderError::InconsistentBlocks),
+            }
+        },
+        {
+            let prev_hash = h!("A");
+            let prev_height = 1;
+            let header = header_from_prev_blockhash(prev_hash);
+            let height = prev_height + 1;
+            let connected_to = BlockId {
+                height,
+                hash: h!("not_current_hash"),
+            };
+            TestCase {
+                name: "inconsistent_current_block",
+                chain: local_chain![(0, h!("_")), (height, h!("not_current_hash"))],
+                header,
+                height,
+                connected_to,
+                exp_result: Err(ApplyHeaderError::InconsistentBlocks),
+            }
+        },
+        {
+            let header = header_from_prev_blockhash(h!("B"));
+            let height = 3;
+            let connected_to = BlockId {
+                height: 4,
+                hash: h!("D"),
+            };
+            TestCase {
+                name: "connected_to_is_greater",
+                chain: local_chain![(0, h!("_")), (2, h!("B"))],
+                header,
+                height,
+                connected_to,
+                exp_result: Err(ApplyHeaderError::InconsistentBlocks),
+            }
+        },
+    ];
+
+    for (i, t) in test_cases.into_iter().enumerate() {
+        println!("running test case {}: '{}'", i, t.name);
+        let mut chain = t.chain;
+        let result = chain.apply_header_connected_to(&t.header, t.height, t.connected_to);
+        let exp_result = t
+            .exp_result
+            .map(|cs| cs.iter().cloned().collect::<ChangeSet>());
+        assert_eq!(result, exp_result, "[{}:{}] unexpected result", i, t.name);
+    }
+}

crates/electrum/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "bdk_electrum"
-version = "0.6.0"
+version = "0.7.0"
 edition = "2021"
 homepage = "https://bitcoindevkit.org"
 repository = "https://github.com/bitcoindevkit/bdk"
@@ -12,6 +12,6 @@ readme = "README.md"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-bdk_chain = { path = "../chain", version = "0.8.0", default-features = false }
+bdk_chain = { path = "../chain", version = "0.9.0", default-features = false }
 electrum-client = { version = "0.18" }
 #rustls = { version = "=0.21.1", optional = true, features = ["dangerous_configuration"] }

crates/electrum/src/electrum_ext.rs

@@ -56,12 +56,14 @@ impl RelevantTxids {
         Ok(graph)
     }
 
-    /// Finalizes [`RelevantTxids`] with `new_txs` and anchors of type
-    /// [`ConfirmationTimeHeightAnchor`].
+    /// Finalizes the update by fetching `missing` txids from the `client`, where the
+    /// resulting [`TxGraph`] has anchors of type [`ConfirmationTimeHeightAnchor`].
+    ///
+    /// Refer to [`RelevantTxids`] for more details.
     ///
     /// **Note:** The confirmation time might not be precisely correct if there has been a reorg.
-    /// Electrum's API intends that we use the merkle proof API, we should change `bdk_electrum` to
-    /// use it.
+    // Electrum's API intends that we use the merkle proof API, we should change `bdk_electrum` to
+    // use it.
     pub fn into_confirmation_time_tx_graph(
         self,
         client: &Client,
@@ -177,7 +179,7 @@ pub trait ElectrumExt {
     ) -> Result<ElectrumUpdate, Error>;
 }
 
-impl ElectrumExt for Client {
+impl<A: ElectrumApi> ElectrumExt for A {
     fn full_scan<K: Ord + Clone>(
         &self,
         prev_tip: CheckPoint,
@@ -301,7 +303,7 @@ impl ElectrumExt for Client {
 
 /// Return a [`CheckPoint`] of the latest tip, that connects with `prev_tip`.
 fn construct_update_tip(
-    client: &Client,
+    client: &impl ElectrumApi,
     prev_tip: CheckPoint,
 ) -> Result<(CheckPoint, Option<u32>), Error> {
     let HeaderNotification { height, .. } = client.block_headers_subscribe()?;
@@ -415,7 +417,7 @@ fn determine_tx_anchor(
 }
 
 fn populate_with_outpoints(
-    client: &Client,
+    client: &impl ElectrumApi,
     cps: &BTreeMap<u32, CheckPoint>,
     relevant_txids: &mut RelevantTxids,
     outpoints: impl IntoIterator<Item = OutPoint>,
@@ -476,7 +478,7 @@ fn populate_with_outpoints(
 }
 
 fn populate_with_txids(
-    client: &Client,
+    client: &impl ElectrumApi,
     cps: &BTreeMap<u32, CheckPoint>,
     relevant_txids: &mut RelevantTxids,
     txids: impl IntoIterator<Item = Txid>,
@@ -512,7 +514,7 @@ fn populate_with_txids(
 }
 
 fn populate_with_spks<I: Ord + Clone>(
-    client: &Client,
+    client: &impl ElectrumApi,
     cps: &BTreeMap<u32, CheckPoint>,
     relevant_txids: &mut RelevantTxids,
     spks: &mut impl Iterator<Item = (I, ScriptBuf)>,

crates/esplora/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "bdk_esplora"
-version = "0.6.0"
+version = "0.7.0"
 edition = "2021"
 homepage = "https://bitcoindevkit.org"
 repository = "https://github.com/bitcoindevkit/bdk"
@@ -12,7 +12,7 @@ readme = "README.md"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-bdk_chain = { path = "../chain", version = "0.8.0", default-features = false }
+bdk_chain = { path = "../chain", version = "0.9.0", default-features = false }
 esplora-client = { version = "0.6.0", default-features = false }
 async-trait = { version = "0.1.66", optional = true }
 futures = { version = "0.3.26", optional = true }

crates/esplora/README.md

@@ -30,7 +30,7 @@ use bdk_esplora::EsploraExt;
 // use bdk_esplora::EsploraAsyncExt;
 ```
 
-For full examples, refer to [`example-crates/wallet_esplora`](https://github.com/bitcoindevkit/bdk/tree/master/example-crates/wallet_esplora) (blocking) and [`example-crates/wallet_esplora_async`](https://github.com/bitcoindevkit/bdk/tree/master/example-crates/wallet_esplora_async).
+For full examples, refer to [`example-crates/wallet_esplora_blocking`](https://github.com/bitcoindevkit/bdk/tree/master/example-crates/wallet_esplora_blocking) and [`example-crates/wallet_esplora_async`](https://github.com/bitcoindevkit/bdk/tree/master/example-crates/wallet_esplora_async).
 
 [`esplora-client`]: https://docs.rs/esplora-client/
 [`bdk_chain`]: https://docs.rs/bdk-chain/

crates/esplora/src/async_ext.rs

@@ -2,14 +2,14 @@ use async_trait::async_trait;
 use bdk_chain::collections::btree_map;
 use bdk_chain::{
     bitcoin::{BlockHash, OutPoint, ScriptBuf, Txid},
-    collections::{BTreeMap, BTreeSet},
+    collections::BTreeMap,
     local_chain::{self, CheckPoint},
     BlockId, ConfirmationTimeHeightAnchor, TxGraph,
 };
 use esplora_client::{Error, TxStatus};
 use futures::{stream::FuturesOrdered, TryStreamExt};
 
-use crate::{anchor_from_status, ASSUME_FINAL_DEPTH};
+use crate::anchor_from_status;
 
 /// Trait to extend the functionality of [`esplora_client::AsyncClient`].
 ///
@@ -19,13 +19,19 @@ use crate::{anchor_from_status, ASSUME_FINAL_DEPTH};
 #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
 #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
 pub trait EsploraAsyncExt {
-    /// Prepare an [`LocalChain`] update with blocks fetched from Esplora.
+    /// Prepare a [`LocalChain`] update with blocks fetched from Esplora.
     ///
     /// * `local_tip` is the previous tip of [`LocalChain::tip`].
     /// * `request_heights` is the block heights that we are interested in fetching from Esplora.
     ///
     /// The result of this method can be applied to [`LocalChain::apply_update`].
     ///
+    /// ## Consistency
+    ///
+    /// The chain update returned is guaranteed to be consistent as long as there is not a *large* re-org
+    /// during the call. The size of re-org we can tollerate is server dependent but will be at
+    /// least 10.
+    ///
     /// [`LocalChain`]: bdk_chain::local_chain::LocalChain
     /// [`LocalChain::tip`]: bdk_chain::local_chain::LocalChain::tip
     /// [`LocalChain::apply_update`]: bdk_chain::local_chain::LocalChain::apply_update
@@ -85,21 +91,22 @@ impl EsploraAsyncExt for esplora_client::AsyncClient {
         local_tip: CheckPoint,
         request_heights: impl IntoIterator<IntoIter = impl Iterator<Item = u32> + Send> + Send,
     ) -> Result<local_chain::Update, Error> {
-        let request_heights = request_heights.into_iter().collect::<BTreeSet<_>>();
-        let new_tip_height = self.get_height().await?;
+        // Fetch latest N (server dependent) blocks from Esplora. The server guarantees these are
+        // consistent.
+        let mut fetched_blocks = self
+            .get_blocks(None)
+            .await?
+            .into_iter()
+            .map(|b| (b.time.height, b.id))
+            .collect::<BTreeMap<u32, BlockHash>>();
+        let new_tip_height = fetched_blocks
+            .keys()
+            .last()
+            .copied()
+            .expect("must have atleast one block");
 
-        // atomically fetch blocks from esplora
-        let mut fetched_blocks = {
-            let heights = (0..=new_tip_height).rev();
-            let hashes = self
-                .get_blocks(Some(new_tip_height))
-                .await?
-                .into_iter()
-                .map(|b| b.id);
-            heights.zip(hashes).collect::<BTreeMap<u32, BlockHash>>()
-        };
-
-        // fetch heights that the caller is interested in
+        // Fetch blocks of heights that the caller is interested in, skipping blocks that are
+        // already fetched when constructing `fetched_blocks`.
         for height in request_heights {
             // do not fetch blocks higher than remote tip
             if height > new_tip_height {
@@ -107,81 +114,37 @@ impl EsploraAsyncExt for esplora_client::AsyncClient {
             }
             // only fetch what is missing
             if let btree_map::Entry::Vacant(entry) = fetched_blocks.entry(height) {
-                let hash = self.get_block_hash(height).await?;
-                entry.insert(hash);
+                // ❗The return value of `get_block_hash` is not strictly guaranteed to be consistent
+                // with the chain at the time of `get_blocks` above (there could have been a deep
+                // re-org). Since `get_blocks` returns 10 (or so) blocks we are assuming that it's
+                // not possible to have a re-org deeper than that.
+                entry.insert(self.get_block_hash(height).await?);
             }
         }
 
-        // find the earliest point of agreement between local chain and fetched chain
-        let earliest_agreement_cp = {
-            let mut earliest_agreement_cp = Option::<CheckPoint>::None;
-
-            let local_tip_height = local_tip.height();
-            for local_cp in local_tip.iter() {
-                let local_block = local_cp.block_id();
-
-                // the updated hash (block hash at this height after the update), can either be:
-                // 1. a block that already existed in `fetched_blocks`
-                // 2. a block that exists locally and at least has a depth of ASSUME_FINAL_DEPTH
-                // 3. otherwise we can freshly fetch the block from remote, which is safe as it
-                //    is guaranteed that this would be at or below ASSUME_FINAL_DEPTH from the
-                //    remote tip
-                let updated_hash = match fetched_blocks.entry(local_block.height) {
-                    btree_map::Entry::Occupied(entry) => *entry.get(),
-                    btree_map::Entry::Vacant(entry) => *entry.insert(
-                        if local_tip_height - local_block.height >= ASSUME_FINAL_DEPTH {
-                            local_block.hash
-                        } else {
-                            self.get_block_hash(local_block.height).await?
-                        },
-                    ),
-                };
-
-                // since we may introduce blocks below the point of agreement, we cannot break
-                // here unconditionally - we only break if we guarantee there are no new heights
-                // below our current local checkpoint
-                if local_block.hash == updated_hash {
-                    earliest_agreement_cp = Some(local_cp);
-
-                    let first_new_height = *fetched_blocks
-                        .keys()
-                        .next()
-                        .expect("must have at least one new block");
-                    if first_new_height >= local_block.height {
-                        break;
-                    }
-                }
+        // Ensure `fetched_blocks` can create an update that connects with the original chain by
+        // finding a "Point of Agreement".
+        for (height, local_hash) in local_tip.iter().map(|cp| (cp.height(), cp.hash())) {
+            if height > new_tip_height {
+                continue;
             }
 
-            earliest_agreement_cp
-        };
-
-        let tip = {
-            // first checkpoint to use for the update chain
-            let first_cp = match earliest_agreement_cp {
-                Some(cp) => cp,
-                None => {
-                    let (&height, &hash) = fetched_blocks
-                        .iter()
-                        .next()
-                        .expect("must have at least one new block");
-                    CheckPoint::new(BlockId { height, hash })
+            let fetched_hash = match fetched_blocks.entry(height) {
+                btree_map::Entry::Occupied(entry) => *entry.get(),
+                btree_map::Entry::Vacant(entry) => {
+                    *entry.insert(self.get_block_hash(height).await?)
                 }
             };
-            // transform fetched chain into the update chain
-            fetched_blocks
-                // we exclude anything at or below the first cp of the update chain otherwise
-                // building the chain will fail
-                .split_off(&(first_cp.height() + 1))
-                .into_iter()
-                .map(|(height, hash)| BlockId { height, hash })
-                .fold(first_cp, |prev_cp, block| {
-                    prev_cp.push(block).expect("must extend checkpoint")
-                })
-        };
+
+            // We have found point of agreement so the update will connect!
+            if fetched_hash == local_hash {
+                break;
+            }
+        }
 
         Ok(local_chain::Update {
-            tip,
+            tip: CheckPoint::from_block_ids(fetched_blocks.into_iter().map(BlockId::from))
+                .expect("must be in height order"),
             introduce_older_blocks: true,
         })
     }

crates/esplora/src/blocking_ext.rs

@@ -1,7 +1,7 @@
 use std::thread::JoinHandle;
 
 use bdk_chain::collections::btree_map;
-use bdk_chain::collections::{BTreeMap, BTreeSet};
+use bdk_chain::collections::BTreeMap;
 use bdk_chain::{
     bitcoin::{BlockHash, OutPoint, ScriptBuf, Txid},
     local_chain::{self, CheckPoint},
@@ -9,7 +9,7 @@ use bdk_chain::{
 };
 use esplora_client::{Error, TxStatus};
 
-use crate::{anchor_from_status, ASSUME_FINAL_DEPTH};
+use crate::anchor_from_status;
 
 /// Trait to extend the functionality of [`esplora_client::BlockingClient`].
 ///
@@ -17,13 +17,19 @@ use crate::{anchor_from_status, ASSUME_FINAL_DEPTH};
 ///
 /// [crate-level documentation]: crate
 pub trait EsploraExt {
-    /// Prepare an [`LocalChain`] update with blocks fetched from Esplora.
+    /// Prepare a [`LocalChain`] update with blocks fetched from Esplora.
     ///
     /// * `local_tip` is the previous tip of [`LocalChain::tip`].
     /// * `request_heights` is the block heights that we are interested in fetching from Esplora.
     ///
     /// The result of this method can be applied to [`LocalChain::apply_update`].
     ///
+    /// ## Consistency
+    ///
+    /// The chain update returned is guaranteed to be consistent as long as there is not a *large* re-org
+    /// during the call. The size of re-org we can tollerate is server dependent but will be at
+    /// least 10.
+    ///
     /// [`LocalChain`]: bdk_chain::local_chain::LocalChain
     /// [`LocalChain::tip`]: bdk_chain::local_chain::LocalChain::tip
     /// [`LocalChain::apply_update`]: bdk_chain::local_chain::LocalChain::apply_update
@@ -78,20 +84,21 @@ impl EsploraExt for esplora_client::BlockingClient {
         local_tip: CheckPoint,
         request_heights: impl IntoIterator<Item = u32>,
     ) -> Result<local_chain::Update, Error> {
-        let request_heights = request_heights.into_iter().collect::<BTreeSet<_>>();
-        let new_tip_height = self.get_height()?;
+        // Fetch latest N (server dependent) blocks from Esplora. The server guarantees these are
+        // consistent.
+        let mut fetched_blocks = self
+            .get_blocks(None)?
+            .into_iter()
+            .map(|b| (b.time.height, b.id))
+            .collect::<BTreeMap<u32, BlockHash>>();
+        let new_tip_height = fetched_blocks
+            .keys()
+            .last()
+            .copied()
+            .expect("must atleast have one block");
 
-        // atomically fetch blocks from esplora
-        let mut fetched_blocks = {
-            let heights = (0..=new_tip_height).rev();
-            let hashes = self
-                .get_blocks(Some(new_tip_height))?
-                .into_iter()
-                .map(|b| b.id);
-            heights.zip(hashes).collect::<BTreeMap<u32, BlockHash>>()
-        };
-
-        // fetch heights that the caller is interested in
+        // Fetch blocks of heights that the caller is interested in, skipping blocks that are
+        // already fetched when constructing `fetched_blocks`.
         for height in request_heights {
             // do not fetch blocks higher than remote tip
             if height > new_tip_height {
@@ -99,81 +106,35 @@ impl EsploraExt for esplora_client::BlockingClient {
             }
             // only fetch what is missing
             if let btree_map::Entry::Vacant(entry) = fetched_blocks.entry(height) {
-                let hash = self.get_block_hash(height)?;
-                entry.insert(hash);
+                // ❗The return value of `get_block_hash` is not strictly guaranteed to be consistent
+                // with the chain at the time of `get_blocks` above (there could have been a deep
+                // re-org). Since `get_blocks` returns 10 (or so) blocks we are assuming that it's
+                // not possible to have a re-org deeper than that.
+                entry.insert(self.get_block_hash(height)?);
             }
         }
 
-        // find the earliest point of agreement between local chain and fetched chain
-        let earliest_agreement_cp = {
-            let mut earliest_agreement_cp = Option::<CheckPoint>::None;
-
-            let local_tip_height = local_tip.height();
-            for local_cp in local_tip.iter() {
-                let local_block = local_cp.block_id();
-
-                // the updated hash (block hash at this height after the update), can either be:
-                // 1. a block that already existed in `fetched_blocks`
-                // 2. a block that exists locally and at least has a depth of ASSUME_FINAL_DEPTH
-                // 3. otherwise we can freshly fetch the block from remote, which is safe as it
-                //    is guaranteed that this would be at or below ASSUME_FINAL_DEPTH from the
-                //    remote tip
-                let updated_hash = match fetched_blocks.entry(local_block.height) {
-                    btree_map::Entry::Occupied(entry) => *entry.get(),
-                    btree_map::Entry::Vacant(entry) => *entry.insert(
-                        if local_tip_height - local_block.height >= ASSUME_FINAL_DEPTH {
-                            local_block.hash
-                        } else {
-                            self.get_block_hash(local_block.height)?
-                        },
-                    ),
-                };
-
-                // since we may introduce blocks below the point of agreement, we cannot break
-                // here unconditionally - we only break if we guarantee there are no new heights
-                // below our current local checkpoint
-                if local_block.hash == updated_hash {
-                    earliest_agreement_cp = Some(local_cp);
-
-                    let first_new_height = *fetched_blocks
-                        .keys()
-                        .next()
-                        .expect("must have at least one new block");
-                    if first_new_height >= local_block.height {
-                        break;
-                    }
-                }
+        // Ensure `fetched_blocks` can create an update that connects with the original chain by
+        // finding a "Point of Agreement".
+        for (height, local_hash) in local_tip.iter().map(|cp| (cp.height(), cp.hash())) {
+            if height > new_tip_height {
+                continue;
             }
 
-            earliest_agreement_cp
-        };
-
-        let tip = {
-            // first checkpoint to use for the update chain
-            let first_cp = match earliest_agreement_cp {
-                Some(cp) => cp,
-                None => {
-                    let (&height, &hash) = fetched_blocks
-                        .iter()
-                        .next()
-                        .expect("must have at least one new block");
-                    CheckPoint::new(BlockId { height, hash })
-                }
+            let fetched_hash = match fetched_blocks.entry(height) {
+                btree_map::Entry::Occupied(entry) => *entry.get(),
+                btree_map::Entry::Vacant(entry) => *entry.insert(self.get_block_hash(height)?),
             };
-            // transform fetched chain into the update chain
-            fetched_blocks
-                // we exclude anything at or below the first cp of the update chain otherwise
-                // building the chain will fail
-                .split_off(&(first_cp.height() + 1))
-                .into_iter()
-                .map(|(height, hash)| BlockId { height, hash })
-                .fold(first_cp, |prev_cp, block| {
-                    prev_cp.push(block).expect("must extend checkpoint")
-                })
-        };
+
+            // We have found point of agreement so the update will connect!
+            if fetched_hash == local_hash {
+                break;
+            }
+        }
 
         Ok(local_chain::Update {
-            tip,
+            tip: CheckPoint::from_block_ids(fetched_blocks.into_iter().map(BlockId::from))
+                .expect("must be in height order"),
             introduce_older_blocks: true,
         })
     }

crates/esplora/src/lib.rs

@@ -31,8 +31,6 @@ mod async_ext;
 #[cfg(feature = "async")]
 pub use async_ext::*;
 
-const ASSUME_FINAL_DEPTH: u32 = 15;
-
 fn anchor_from_status(status: &TxStatus) -> Option<ConfirmationTimeHeightAnchor> {
     if let TxStatus {
         block_height: Some(height),

crates/esplora/tests/blocking_ext.rs

@@ -1,15 +1,31 @@
+use bdk_chain::local_chain::LocalChain;
+use bdk_chain::BlockId;
 use bdk_esplora::EsploraExt;
 use electrsd::bitcoind::bitcoincore_rpc::RpcApi;
 use electrsd::bitcoind::{self, anyhow, BitcoinD};
 use electrsd::{Conf, ElectrsD};
 use esplora_client::{self, BlockingClient, Builder};
-use std::collections::{BTreeMap, HashSet};
+use std::collections::{BTreeMap, BTreeSet, HashSet};
 use std::str::FromStr;
 use std::thread::sleep;
 use std::time::Duration;
 
 use bdk_chain::bitcoin::{Address, Amount, BlockHash, Txid};
 
+macro_rules! h {
+    ($index:literal) => {{
+        bdk_chain::bitcoin::hashes::Hash::hash($index.as_bytes())
+    }};
+}
+
+macro_rules! local_chain {
+    [ $(($height:expr, $block_hash:expr)), * ] => {{
+        #[allow(unused_mut)]
+        bdk_chain::local_chain::LocalChain::from_blocks([$(($height, $block_hash).into()),*].into_iter().collect())
+            .expect("chain must have genesis block")
+    }};
+}
+
 struct TestEnv {
     bitcoind: BitcoinD,
     #[allow(dead_code)]
@@ -39,6 +55,20 @@ impl TestEnv {
         })
     }
 
+    fn reset_electrsd(mut self) -> anyhow::Result<Self> {
+        let mut electrs_conf = Conf::default();
+        electrs_conf.http_enabled = true;
+        let electrs_exe =
+            electrsd::downloaded_exe_path().expect("electrs version feature must be enabled");
+        let electrsd = ElectrsD::with_conf(electrs_exe, &self.bitcoind, &electrs_conf)?;
+
+        let base_url = format!("http://{}", &electrsd.esplora_url.clone().unwrap());
+        let client = Builder::new(base_url.as_str()).build_blocking()?;
+        self.electrsd = electrsd;
+        self.client = client;
+        Ok(self)
+    }
+
     fn mine_blocks(
         &self,
         count: usize,
@@ -202,3 +232,180 @@ pub fn test_update_tx_graph_gap_limit() -> anyhow::Result<()> {
 
     Ok(())
 }
+
+#[test]
+fn update_local_chain() -> anyhow::Result<()> {
+    const TIP_HEIGHT: u32 = 50;
+
+    let env = TestEnv::new()?;
+    let blocks = {
+        let bitcoind_client = &env.bitcoind.client;
+        assert_eq!(bitcoind_client.get_block_count()?, 1);
+        [
+            (0, bitcoind_client.get_block_hash(0)?),
+            (1, bitcoind_client.get_block_hash(1)?),
+        ]
+        .into_iter()
+        .chain((2..).zip(env.mine_blocks((TIP_HEIGHT - 1) as usize, None)?))
+        .collect::<BTreeMap<_, _>>()
+    };
+    // so new blocks can be seen by Electrs
+    let env = env.reset_electrsd()?;
+
+    struct TestCase {
+        name: &'static str,
+        chain: LocalChain,
+        request_heights: &'static [u32],
+        exp_update_heights: &'static [u32],
+    }
+
+    let test_cases = [
+        TestCase {
+            name: "request_later_blocks",
+            chain: local_chain![(0, blocks[&0]), (21, blocks[&21])],
+            request_heights: &[22, 25, 28],
+            exp_update_heights: &[21, 22, 25, 28],
+        },
+        TestCase {
+            name: "request_prev_blocks",
+            chain: local_chain![(0, blocks[&0]), (1, blocks[&1]), (5, blocks[&5])],
+            request_heights: &[4],
+            exp_update_heights: &[4, 5],
+        },
+        TestCase {
+            name: "request_prev_blocks_2",
+            chain: local_chain![(0, blocks[&0]), (1, blocks[&1]), (10, blocks[&10])],
+            request_heights: &[4, 6],
+            exp_update_heights: &[4, 6, 10],
+        },
+        TestCase {
+            name: "request_later_and_prev_blocks",
+            chain: local_chain![(0, blocks[&0]), (7, blocks[&7]), (11, blocks[&11])],
+            request_heights: &[8, 9, 15],
+            exp_update_heights: &[8, 9, 11, 15],
+        },
+        TestCase {
+            name: "request_tip_only",
+            chain: local_chain![(0, blocks[&0]), (5, blocks[&5]), (49, blocks[&49])],
+            request_heights: &[TIP_HEIGHT],
+            exp_update_heights: &[49],
+        },
+        TestCase {
+            name: "request_nothing",
+            chain: local_chain![(0, blocks[&0]), (13, blocks[&13]), (23, blocks[&23])],
+            request_heights: &[],
+            exp_update_heights: &[23],
+        },
+        TestCase {
+            name: "request_nothing_during_reorg",
+            chain: local_chain![(0, blocks[&0]), (13, blocks[&13]), (23, h!("23"))],
+            request_heights: &[],
+            exp_update_heights: &[13, 23],
+        },
+        TestCase {
+            name: "request_nothing_during_reorg_2",
+            chain: local_chain![
+                (0, blocks[&0]),
+                (21, blocks[&21]),
+                (22, h!("22")),
+                (23, h!("23"))
+            ],
+            request_heights: &[],
+            exp_update_heights: &[21, 22, 23],
+        },
+        TestCase {
+            name: "request_prev_blocks_during_reorg",
+            chain: local_chain![
+                (0, blocks[&0]),
+                (21, blocks[&21]),
+                (22, h!("22")),
+                (23, h!("23"))
+            ],
+            request_heights: &[17, 20],
+            exp_update_heights: &[17, 20, 21, 22, 23],
+        },
+        TestCase {
+            name: "request_later_blocks_during_reorg",
+            chain: local_chain![
+                (0, blocks[&0]),
+                (9, blocks[&9]),
+                (22, h!("22")),
+                (23, h!("23"))
+            ],
+            request_heights: &[25, 27],
+            exp_update_heights: &[9, 22, 23, 25, 27],
+        },
+        TestCase {
+            name: "request_later_blocks_during_reorg_2",
+            chain: local_chain![(0, blocks[&0]), (9, h!("9"))],
+            request_heights: &[10],
+            exp_update_heights: &[0, 9, 10],
+        },
+        TestCase {
+            name: "request_later_and_prev_blocks_during_reorg",
+            chain: local_chain![(0, blocks[&0]), (1, blocks[&1]), (9, h!("9"))],
+            request_heights: &[8, 11],
+            exp_update_heights: &[1, 8, 9, 11],
+        },
+    ];
+
+    for (i, t) in test_cases.into_iter().enumerate() {
+        println!("Case {}: {}", i, t.name);
+        let mut chain = t.chain;
+
+        let update = env
+            .client
+            .update_local_chain(chain.tip(), t.request_heights.iter().copied())
+            .map_err(|err| {
+                anyhow::format_err!("[{}:{}] `update_local_chain` failed: {}", i, t.name, err)
+            })?;
+
+        let update_blocks = update
+            .tip
+            .iter()
+            .map(|cp| cp.block_id())
+            .collect::<BTreeSet<_>>();
+
+        let exp_update_blocks = t
+            .exp_update_heights
+            .iter()
+            .map(|&height| {
+                let hash = blocks[&height];
+                BlockId { height, hash }
+            })
+            .chain(
+                // Electrs Esplora `get_block` call fetches 10 blocks which is included in the
+                // update
+                blocks
+                    .range(TIP_HEIGHT - 9..)
+                    .map(|(&height, &hash)| BlockId { height, hash }),
+            )
+            .collect::<BTreeSet<_>>();
+
+        assert_eq!(
+            update_blocks, exp_update_blocks,
+            "[{}:{}] unexpected update",
+            i, t.name
+        );
+
+        let _ = chain
+            .apply_update(update)
+            .unwrap_or_else(|err| panic!("[{}:{}] update failed to apply: {}", i, t.name, err));
+
+        // all requested heights must exist in the final chain
+        for height in t.request_heights {
+            let exp_blockhash = blocks.get(height).expect("block must exist in bitcoind");
+            assert_eq!(
+                chain.blocks().get(height),
+                Some(exp_blockhash),
+                "[{}:{}] block {}:{} must exist in final chain",
+                i,
+                t.name,
+                height,
+                exp_blockhash
+            );
+        }
+    }
+
+    Ok(())
+}

crates/file_store/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "bdk_file_store"
-version = "0.4.0"
+version = "0.5.0"
 edition = "2021"
 license = "MIT OR Apache-2.0"
 repository = "https://github.com/bitcoindevkit/bdk"
@@ -11,7 +11,7 @@ authors = ["Bitcoin Dev Kit Developers"]
 readme = "README.md"
 
 [dependencies]
-bdk_chain = { path = "../chain", version = "0.8.0", features = [ "serde", "miniscript" ] }
+bdk_chain = { path = "../chain", version = "0.9.0", features = [ "serde", "miniscript" ] }
 bincode = { version = "1" }
 serde = { version = "1", features = ["derive"] }
 

crates/file_store/src/entry_iter.rs

@@ -1,7 +1,7 @@
 use bincode::Options;
 use std::{
     fs::File,
-    io::{self, Seek},
+    io::{self, BufReader, Seek},
     marker::PhantomData,
 };
 
@@ -14,8 +14,9 @@ use crate::bincode_options;
 ///
 /// [`next`]: Self::next
 pub struct EntryIter<'t, T> {
-    db_file: Option<&'t mut File>,
-
+    /// Buffered reader around the file
+    db_file: BufReader<&'t mut File>,
+    finished: bool,
     /// The file position for the first read of `db_file`.
     start_pos: Option<u64>,
     types: PhantomData<T>,
@@ -24,8 +25,9 @@ pub struct EntryIter<'t, T> {
 impl<'t, T> EntryIter<'t, T> {
     pub fn new(start_pos: u64, db_file: &'t mut File) -> Self {
         Self {
-            db_file: Some(db_file),
+            db_file: BufReader::new(db_file),
             start_pos: Some(start_pos),
+            finished: false,
             types: PhantomData,
         }
     }
@@ -38,44 +40,44 @@ where
     type Item = Result<T, IterError>;
 
     fn next(&mut self) -> Option<Self::Item> {
-        // closure which reads a single entry starting from `self.pos`
-        let read_one = |f: &mut File, start_pos: Option<u64>| -> Result<Option<T>, IterError> {
-            let pos = match start_pos {
-                Some(pos) => f.seek(io::SeekFrom::Start(pos))?,
-                None => f.stream_position()?,
-            };
+        if self.finished {
+            return None;
+        }
+        (|| {
+            if let Some(start) = self.start_pos.take() {
+                self.db_file.seek(io::SeekFrom::Start(start))?;
+            }
 
-            match bincode_options().deserialize_from(&*f) {
-                Ok(changeset) => {
-                    f.stream_position()?;
-                    Ok(Some(changeset))
-                }
+            let pos_before_read = self.db_file.stream_position()?;
+            match bincode_options().deserialize_from(&mut self.db_file) {
+                Ok(changeset) => Ok(Some(changeset)),
                 Err(e) => {
+                    self.finished = true;
+                    let pos_after_read = self.db_file.stream_position()?;
+                    // allow unexpected EOF if 0 bytes were read
                     if let bincode::ErrorKind::Io(inner) = &*e {
-                        if inner.kind() == io::ErrorKind::UnexpectedEof {
-                            let eof = f.seek(io::SeekFrom::End(0))?;
-                            if pos == eof {
-                                return Ok(None);
-                            }
+                        if inner.kind() == io::ErrorKind::UnexpectedEof
+                            && pos_after_read == pos_before_read
+                        {
+                            return Ok(None);
                         }
                     }
-                    f.seek(io::SeekFrom::Start(pos))?;
+                    self.db_file.seek(io::SeekFrom::Start(pos_before_read))?;
                     Err(IterError::Bincode(*e))
                 }
             }
-        };
-
-        let result = read_one(self.db_file.as_mut()?, self.start_pos.take());
-        if result.is_err() {
-            self.db_file = None;
-        }
-        result.transpose()
+        })()
+        .transpose()
     }
 }
 
-impl From<io::Error> for IterError {
-    fn from(value: io::Error) -> Self {
-        IterError::Io(value)
+impl<'t, T> Drop for EntryIter<'t, T> {
+    fn drop(&mut self) {
+        // This syncs the underlying file's offset with the buffer's position. This way, we
+        // maintain the correct position to start the next read/write.
+        if let Ok(pos) = self.db_file.stream_position() {
+            let _ = self.db_file.get_mut().seek(io::SeekFrom::Start(pos));
+        }
     }
 }
 
@@ -97,4 +99,10 @@ impl core::fmt::Display for IterError {
     }
 }
 
+impl From<io::Error> for IterError {
+    fn from(value: io::Error) -> Self {
+        IterError::Io(value)
+    }
+}
+
 impl std::error::Error for IterError {}

crates/file_store/src/lib.rs

@@ -13,14 +13,14 @@ pub(crate) fn bincode_options() -> impl bincode::Options {
 
 /// Error that occurs due to problems encountered with the file.
 #[derive(Debug)]
-pub enum FileError<'a> {
+pub enum FileError {
     /// IO error, this may mean that the file is too short.
     Io(io::Error),
     /// Magic bytes do not match what is expected.
-    InvalidMagicBytes { got: Vec<u8>, expected: &'a [u8] },
+    InvalidMagicBytes { got: Vec<u8>, expected: Vec<u8> },
 }
 
-impl<'a> core::fmt::Display for FileError<'a> {
+impl core::fmt::Display for FileError {
     fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
         match self {
             Self::Io(e) => write!(f, "io error trying to read file: {}", e),
@@ -33,10 +33,10 @@ impl<'a> core::fmt::Display for FileError<'a> {
     }
 }
 
-impl<'a> From<io::Error> for FileError<'a> {
+impl From<io::Error> for FileError {
     fn from(value: io::Error) -> Self {
         Self::Io(value)
     }
 }
 
-impl<'a> std::error::Error for FileError<'a> {}
+impl std::error::Error for FileError {}

crates/file_store/src/store.rs

@@ -15,13 +15,13 @@ use crate::{bincode_options, EntryIter, FileError, IterError};
 ///
 /// The changesets are the results of altering a tracker implementation (`T`).
 #[derive(Debug)]
-pub struct Store<'a, C> {
-    magic: &'a [u8],
+pub struct Store<C> {
+    magic_len: usize,
     db_file: File,
     marker: PhantomData<C>,
 }
 
-impl<'a, C> PersistBackend<C> for Store<'a, C>
+impl<C> PersistBackend<C> for Store<C>
 where
     C: Append + serde::Serialize + serde::de::DeserializeOwned,
 {
@@ -38,7 +38,7 @@ where
     }
 }
 
-impl<'a, C> Store<'a, C>
+impl<C> Store<C>
 where
     C: Append + serde::Serialize + serde::de::DeserializeOwned,
 {
@@ -48,7 +48,7 @@ where
     /// the `Store` in the future with [`open`].
     ///
     /// [`open`]: Store::open
-    pub fn create_new<P>(magic: &'a [u8], file_path: P) -> Result<Self, FileError>
+    pub fn create_new<P>(magic: &[u8], file_path: P) -> Result<Self, FileError>
     where
         P: AsRef<Path>,
     {
@@ -67,7 +67,7 @@ where
             .open(file_path)?;
         f.write_all(magic)?;
         Ok(Self {
-            magic,
+            magic_len: magic.len(),
             db_file: f,
             marker: Default::default(),
         })
@@ -83,7 +83,7 @@ where
     /// [`FileError::InvalidMagicBytes`] error variant will be returned.
     ///
     /// [`create_new`]: Store::create_new
-    pub fn open<P>(magic: &'a [u8], file_path: P) -> Result<Self, FileError>
+    pub fn open<P>(magic: &[u8], file_path: P) -> Result<Self, FileError>
     where
         P: AsRef<Path>,
     {
@@ -94,12 +94,12 @@ where
         if magic_buf != magic {
             return Err(FileError::InvalidMagicBytes {
                 got: magic_buf,
-                expected: magic,
+                expected: magic.to_vec(),
             });
         }
 
         Ok(Self {
-            magic,
+            magic_len: magic.len(),
             db_file: f,
             marker: Default::default(),
         })
@@ -111,7 +111,7 @@ where
     ///
     /// [`open`]: Store::open
     /// [`create_new`]: Store::create_new
-    pub fn open_or_create_new<P>(magic: &'a [u8], file_path: P) -> Result<Self, FileError>
+    pub fn open_or_create_new<P>(magic: &[u8], file_path: P) -> Result<Self, FileError>
     where
         P: AsRef<Path>,
     {
@@ -132,14 +132,14 @@ where
     /// always iterate over all entries until `None` is returned if you want your next write to go
     /// at the end; otherwise, you will write over existing entries.
     pub fn iter_changesets(&mut self) -> EntryIter<C> {
-        EntryIter::new(self.magic.len() as u64, &mut self.db_file)
+        EntryIter::new(self.magic_len as u64, &mut self.db_file)
     }
 
     /// Loads all the changesets that have been stored as one giant changeset.
     ///
-    /// This function returns a tuple of the aggregate changeset and a result that indicates
-    /// whether an error occurred while reading or deserializing one of the entries. If so the
-    /// changeset will consist of all of those it was able to read.
+    /// This function returns the aggregate changeset, or `None` if nothing was persisted.
+    /// If reading or deserializing any of the entries fails, an error is returned that
+    /// consists of all those it was able to read.
     ///
     /// You should usually check the error. In many applications, it may make sense to do a full
     /// wallet scan with a stop-gap after getting an error, since it is likely that one of the
@@ -219,6 +219,7 @@ mod test {
 
     use bincode::DefaultOptions;
     use std::{
+        collections::BTreeSet,
         io::{Read, Write},
         vec::Vec,
     };
@@ -228,7 +229,7 @@ mod test {
     const TEST_MAGIC_BYTES: [u8; TEST_MAGIC_BYTES_LEN] =
         [98, 100, 107, 102, 115, 49, 49, 49, 49, 49, 49, 49];
 
-    type TestChangeSet = Vec<String>;
+    type TestChangeSet = BTreeSet<String>;
 
     #[derive(Debug)]
     struct TestTracker;
@@ -253,7 +254,7 @@ mod test {
     fn open_or_create_new() {
         let temp_dir = tempfile::tempdir().unwrap();
         let file_path = temp_dir.path().join("db_file");
-        let changeset = vec!["hello".to_string(), "world".to_string()];
+        let changeset = BTreeSet::from(["hello".to_string(), "world".to_string()]);
 
         {
             let mut db = Store::<TestChangeSet>::open_or_create_new(&TEST_MAGIC_BYTES, &file_path)
@@ -304,7 +305,7 @@ mod test {
         let mut data = [255_u8; 2000];
         data[..TEST_MAGIC_BYTES_LEN].copy_from_slice(&TEST_MAGIC_BYTES);
 
-        let changeset = vec!["one".into(), "two".into(), "three!".into()];
+        let changeset = TestChangeSet::from(["one".into(), "two".into(), "three!".into()]);
 
         let mut file = NamedTempFile::new().unwrap();
         file.write_all(&data).expect("should write");
@@ -340,4 +341,119 @@ mod test {
 
         assert_eq!(got_bytes, expected_bytes);
     }
+
+    #[test]
+    fn last_write_is_short() {
+        let temp_dir = tempfile::tempdir().unwrap();
+
+        let changesets = [
+            TestChangeSet::from(["1".into()]),
+            TestChangeSet::from(["2".into(), "3".into()]),
+            TestChangeSet::from(["4".into(), "5".into(), "6".into()]),
+        ];
+        let last_changeset = TestChangeSet::from(["7".into(), "8".into(), "9".into()]);
+        let last_changeset_bytes = bincode_options().serialize(&last_changeset).unwrap();
+
+        for short_write_len in 1..last_changeset_bytes.len() - 1 {
+            let file_path = temp_dir.path().join(format!("{}.dat", short_write_len));
+            println!("Test file: {:?}", file_path);
+
+            // simulate creating a file, writing data where the last write is incomplete
+            {
+                let mut db =
+                    Store::<TestChangeSet>::create_new(&TEST_MAGIC_BYTES, &file_path).unwrap();
+                for changeset in &changesets {
+                    db.append_changeset(changeset).unwrap();
+                }
+                // this is the incomplete write
+                db.db_file
+                    .write_all(&last_changeset_bytes[..short_write_len])
+                    .unwrap();
+            }
+
+            // load file again and aggregate changesets
+            // write the last changeset again (this time it succeeds)
+            {
+                let mut db = Store::<TestChangeSet>::open(&TEST_MAGIC_BYTES, &file_path).unwrap();
+                let err = db
+                    .aggregate_changesets()
+                    .expect_err("should return error as last read is short");
+                assert_eq!(
+                    err.changeset,
+                    changesets.iter().cloned().reduce(|mut acc, cs| {
+                        Append::append(&mut acc, cs);
+                        acc
+                    }),
+                    "should recover all changesets that are written in full",
+                );
+                db.db_file.write_all(&last_changeset_bytes).unwrap();
+            }
+
+            // load file again - this time we should successfully aggregate all changesets
+            {
+                let mut db = Store::<TestChangeSet>::open(&TEST_MAGIC_BYTES, &file_path).unwrap();
+                let aggregated_changesets = db
+                    .aggregate_changesets()
+                    .expect("aggregating all changesets should succeed");
+                assert_eq!(
+                    aggregated_changesets,
+                    changesets
+                        .iter()
+                        .cloned()
+                        .chain(core::iter::once(last_changeset.clone()))
+                        .reduce(|mut acc, cs| {
+                            Append::append(&mut acc, cs);
+                            acc
+                        }),
+                    "should recover all changesets",
+                );
+            }
+        }
+    }
+
+    #[test]
+    fn write_after_short_read() {
+        let temp_dir = tempfile::tempdir().unwrap();
+
+        let changesets = (0..20)
+            .map(|n| TestChangeSet::from([format!("{}", n)]))
+            .collect::<Vec<_>>();
+        let last_changeset = TestChangeSet::from(["last".into()]);
+
+        for read_count in 0..changesets.len() {
+            let file_path = temp_dir.path().join(format!("{}.dat", read_count));
+            println!("Test file: {:?}", file_path);
+
+            // First, we create the file with all the changesets!
+            let mut db = Store::<TestChangeSet>::create_new(&TEST_MAGIC_BYTES, &file_path).unwrap();
+            for changeset in &changesets {
+                db.append_changeset(changeset).unwrap();
+            }
+            drop(db);
+
+            // We re-open the file and read `read_count` number of changesets.
+            let mut db = Store::<TestChangeSet>::open(&TEST_MAGIC_BYTES, &file_path).unwrap();
+            let mut exp_aggregation = db
+                .iter_changesets()
+                .take(read_count)
+                .map(|r| r.expect("must read valid changeset"))
+                .fold(TestChangeSet::default(), |mut acc, v| {
+                    Append::append(&mut acc, v);
+                    acc
+                });
+            // We write after a short read.
+            db.write_changes(&last_changeset)
+                .expect("last write must succeed");
+            Append::append(&mut exp_aggregation, last_changeset.clone());
+            drop(db);
+
+            // We open the file again and check whether aggregate changeset is expected.
+            let aggregation = Store::<TestChangeSet>::open(&TEST_MAGIC_BYTES, &file_path)
+                .unwrap()
+                .aggregate_changesets()
+                .expect("must aggregate changesets")
+                .unwrap_or_default();
+            assert_eq!(aggregation, exp_aggregation);
+        }
+    }
 }

example-crates/example_bitcoind_rpc_polling/README.md

@@ -0,0 +1,68 @@
+# Example RPC CLI
+
+### Simple Regtest Test
+
+1. Start local regtest bitcoind.
+   ```
+    mkdir -p /tmp/regtest/bitcoind
+    bitcoind -regtest -server -fallbackfee=0.0002 -rpcuser=<your-rpc-username> -rpcpassword=<your-rpc-password> -datadir=/tmp/regtest/bitcoind -daemon
+   ```
+2. Create a test bitcoind wallet and set bitcoind env.
+   ```
+   bitcoin-cli -datadir=/tmp/regtest/bitcoind -regtest -rpcuser=<your-rpc-username> -rpcpassword=<your-rpc-password> -named createwallet wallet_name="test"
+   export RPC_URL=127.0.0.1:18443
+   export RPC_USER=<your-rpc-username>
+   export RPC_PASS=<your-rpc-password>
+   ```
+3. Get test bitcoind wallet info.
+   ```
+   bitcoin-cli -rpcwallet="test" -rpcuser=<your-rpc-username> -rpcpassword=<your-rpc-password> -datadir=/tmp/regtest/bitcoind -regtest getwalletinfo
+   ```
+4. Get new test bitcoind wallet address.
+   ```
+   BITCOIND_ADDRESS=$(bitcoin-cli -rpcwallet="test" -datadir=/tmp/regtest/bitcoind -regtest -rpcuser=<your-rpc-username> -rpcpassword=<your-rpc-password> getnewaddress)
+   echo $BITCOIND_ADDRESS
+   ```
+5. Generate 101 blocks with reward to test bitcoind wallet address.
+   ```
+   bitcoin-cli -datadir=/tmp/regtest/bitcoind -regtest -rpcuser=<your-rpc-username> -rpcpassword=<your-rpc-password> generatetoaddress 101 $BITCOIND_ADDRESS
+   ```
+6. Verify test bitcoind wallet balance.
+   ```
+   bitcoin-cli -rpcwallet="test" -datadir=/tmp/regtest/bitcoind -regtest -rpcuser=<your-rpc-username> -rpcpassword=<your-rpc-password> getbalances
+   ```
+7. Set descriptor env and get address from RPC CLI wallet.
+   ```
+   export DESCRIPTOR="wpkh(tprv8ZgxMBicQKsPfK9BTf82oQkHhawtZv19CorqQKPFeaHDMA4dXYX6eWsJGNJ7VTQXWmoHdrfjCYuDijcRmNFwSKcVhswzqs4fugE8turndGc/1/*)"
+   cargo run -- --network regtest address next
+   ```
+8. Send 5 test bitcoin to RPC CLI wallet.
+   ```
+   bitcoin-cli -rpcwallet="test" -datadir=/tmp/regtest/bitcoind -regtest -rpcuser=<your-rpc-username> -rpcpassword=<your-rpc-password> sendtoaddress <address> 5
+   ```
+9. Sync blockchain with RPC CLI wallet.
+   ```
+   cargo run -- --network regtest sync
+   <CNTRL-C to stop syncing>
+   ```
+10. Get RPC CLI wallet unconfirmed balances.
+   ```
+   cargo run -- --network regtest balance
+   ```
+11. Generate 1 block with reward to test bitcoind wallet address.
+   ```
+   bitcoin-cli -datadir=/tmp/regtest/bitcoind -rpcuser=<your-rpc-username> -rpcpassword=<your-rpc-password> -regtest generatetoaddress 10 $BITCOIND_ADDRESS
+   ```
+12. Sync the blockchain with RPC CLI wallet.
+   ```
+   cargo run -- --network regtest sync
+   <CNTRL-C to stop syncing>
+   ```
+13. Get RPC CLI wallet confirmed balances.
+   ```
+   cargo run -- --network regtest balance
+   ```
+14. Get RPC CLI wallet transactions.
+   ```
+   cargo run -- --network regtest txout list
+   ```

example-crates/example_bitcoind_rpc_polling/src/main.rs

@@ -14,7 +14,7 @@ use bdk_bitcoind_rpc::{
 use bdk_chain::{
     bitcoin::{constants::genesis_block, Block, Transaction},
     indexed_tx_graph, keychain,
-    local_chain::{self, CheckPoint, LocalChain},
+    local_chain::{self, LocalChain},
     ConfirmationTimeHeightAnchor, IndexedTxGraph,
 };
 use example_cli::{
@@ -42,7 +42,7 @@ type ChangeSet = (
 
 #[derive(Debug)]
 enum Emission {
-    Block { height: u32, block: Block },
+    Block(bdk_bitcoind_rpc::BlockEvent<Block>),
     Mempool(Vec<(Transaction, u64)>),
     Tip(u32),
 }
@@ -147,7 +147,7 @@ fn main() -> anyhow::Result<()> {
     let rpc_cmd = match args.command {
         example_cli::Commands::ChainSpecific(rpc_cmd) => rpc_cmd,
         general_cmd => {
-            let res = example_cli::handle_commands(
+            return example_cli::handle_commands(
                 &graph,
                 &db,
                 &chain,
@@ -160,8 +160,6 @@ fn main() -> anyhow::Result<()> {
                 },
                 general_cmd,
             );
-            db.lock().unwrap().commit()?;
-            return res;
         }
     };
 
@@ -178,17 +176,20 @@ fn main() -> anyhow::Result<()> {
             let mut last_db_commit = Instant::now();
             let mut last_print = Instant::now();
 
-            while let Some((height, block)) = emitter.next_block()? {
+            while let Some(emission) = emitter.next_block()? {
+                let height = emission.block_height();
+
                 let mut chain = chain.lock().unwrap();
                 let mut graph = graph.lock().unwrap();
                 let mut db = db.lock().unwrap();
 
-                let chain_update =
-                    CheckPoint::from_header(&block.header, height).into_update(false);
                 let chain_changeset = chain
-                    .apply_update(chain_update)
+                    .apply_update(local_chain::Update {
+                        tip: emission.checkpoint,
+                        introduce_older_blocks: false,
+                    })
                     .expect("must always apply as we receive blocks in order from emitter");
-                let graph_changeset = graph.apply_block_relevant(block, height);
+                let graph_changeset = graph.apply_block_relevant(&emission.block, height);
                 db.stage((chain_changeset, graph_changeset));
 
                 // commit staged db changes in intervals
@@ -256,7 +257,8 @@ fn main() -> anyhow::Result<()> {
 
                 loop {
                     match emitter.next_block()? {
-                        Some((height, block)) => {
+                        Some(block_emission) => {
+                            let height = block_emission.block_height();
                             if sigterm_flag.load(Ordering::Acquire) {
                                 break;
                             }
@@ -264,7 +266,7 @@ fn main() -> anyhow::Result<()> {
                                 block_count = rpc_client.get_block_count()? as u32;
                                 tx.send(Emission::Tip(block_count))?;
                             }
-                            tx.send(Emission::Block { height, block })?;
+                            tx.send(Emission::Block(block_emission))?;
                         }
                         None => {
                             if await_flag(&sigterm_flag, MEMPOOL_EMIT_DELAY) {
@@ -293,13 +295,17 @@ fn main() -> anyhow::Result<()> {
                 let mut chain = chain.lock().unwrap();
 
                 let changeset = match emission {
-                    Emission::Block { height, block } => {
-                        let chain_update =
-                            CheckPoint::from_header(&block.header, height).into_update(false);
+                    Emission::Block(block_emission) => {
+                        let height = block_emission.block_height();
+                        let chain_update = local_chain::Update {
+                            tip: block_emission.checkpoint,
+                            introduce_older_blocks: false,
+                        };
                         let chain_changeset = chain
                             .apply_update(chain_update)
                             .expect("must always apply as we receive blocks in order from emitter");
-                        let graph_changeset = graph.apply_block_relevant(block, height);
+                        let graph_changeset =
+                            graph.apply_block_relevant(&block_emission.block, height);
                         (chain_changeset, graph_changeset)
                     }
                     Emission::Mempool(mempool_txs) => {

example-crates/example_cli/src/lib.rs

@@ -29,7 +29,7 @@ pub type KeychainChangeSet<A> = (
     local_chain::ChangeSet,
     indexed_tx_graph::ChangeSet<A, keychain::ChangeSet<Keychain>>,
 );
-pub type Database<'m, C> = Persist<Store<'m, C>, C>;
+pub type Database<C> = Persist<Store<C>, C>;
 
 #[derive(Parser)]
 #[clap(author, version, about, long_about = None)]
@@ -73,7 +73,9 @@ pub enum Commands<CS: clap::Subcommand, S: clap::Args> {
     },
     /// Send coins to an address.
     Send {
+        /// Amount to send in satoshis
         value: u64,
+        /// Destination address
         address: Address<address::NetworkUnchecked>,
         #[clap(short, default_value = "bnb")]
         coin_select: CoinSelectionAlgo,
@@ -144,14 +146,17 @@ pub enum AddressCmd {
     New,
     /// List all addresses
     List {
+        /// List change addresses
         #[clap(long)]
         change: bool,
     },
+    /// Get last revealed address index for each keychain.
     Index,
 }
 
 #[derive(Subcommand, Debug, Clone)]
 pub enum TxOutCmd {
+    /// List transaction outputs.
     List {
         /// Return only spent outputs.
         #[clap(short, long)]
@@ -457,11 +462,10 @@ where
 
                     let ((spk_i, spk), index_changeset) = spk_chooser(index, &Keychain::External);
                     let db = &mut *db.lock().unwrap();
-                    db.stage(C::from((
+                    db.stage_and_commit(C::from((
                         local_chain::ChangeSet::default(),
                         indexed_tx_graph::ChangeSet::from(index_changeset),
-                    )));
-                    db.commit()?;
+                    )))?;
                     let addr =
                         Address::from_script(spk, network).context("failed to derive address")?;
                     println!("[address @ {}] {}", spk_i, addr);
@@ -601,11 +605,10 @@ where
                     // If we're unable to persist this, then we don't want to broadcast.
                     {
                         let db = &mut *db.lock().unwrap();
-                        db.stage(C::from((
+                        db.stage_and_commit(C::from((
                             local_chain::ChangeSet::default(),
                             indexed_tx_graph::ChangeSet::from(index_changeset),
-                        )));
-                        db.commit()?;
+                        )))?;
                     }
 
                     // We don't want other callers/threads to use this address while we're using it
@@ -627,10 +630,10 @@ where
                     // We know the tx is at least unconfirmed now. Note if persisting here fails,
                     // it's not a big deal since we can always find it again form
                     // blockchain.
-                    db.lock().unwrap().stage(C::from((
+                    db.lock().unwrap().stage_and_commit(C::from((
                         local_chain::ChangeSet::default(),
                         keychain_changeset,
-                    )));
+                    )))?;
                     Ok(())
                 }
                 Err(e) => {
@@ -646,14 +649,14 @@ where
 }
 
 #[allow(clippy::type_complexity)]
-pub fn init<'m, CS: clap::Subcommand, S: clap::Args, C>(
-    db_magic: &'m [u8],
+pub fn init<CS: clap::Subcommand, S: clap::Args, C>(
+    db_magic: &[u8],
     db_default_path: &str,
 ) -> anyhow::Result<(
     Args<CS, S>,
     KeyMap,
     KeychainTxOutIndex<Keychain>,
-    Mutex<Database<'m, C>>,
+    Mutex<Database<C>>,
     C,
 )>
 where
@@ -681,7 +684,7 @@ where
         index.add_keychain(Keychain::Internal, internal_descriptor);
     }
 
-    let mut db_backend = match Store::<'m, C>::open_or_create_new(db_magic, &args.db_path) {
+    let mut db_backend = match Store::<C>::open_or_create_new(db_magic, &args.db_path) {
         Ok(db_backend) => db_backend,
         // we cannot return `err` directly as it has lifetime `'m`
         Err(err) => return Err(anyhow::anyhow!("failed to init db backend: {:?}", err)),

example-crates/example_electrum/src/main.rs

@@ -122,7 +122,7 @@ fn main() -> anyhow::Result<()> {
     let electrum_cmd = match &args.command {
         example_cli::Commands::ChainSpecific(electrum_cmd) => electrum_cmd,
         general_cmd => {
-            let res = example_cli::handle_commands(
+            return example_cli::handle_commands(
                 &graph,
                 &db,
                 &chain,
@@ -135,9 +135,6 @@ fn main() -> anyhow::Result<()> {
                 },
                 general_cmd.clone(),
             );
-
-            db.lock().unwrap().commit()?;
-            return res;
         }
     };
 

example-crates/example_esplora/src/main.rs

@@ -125,7 +125,7 @@ fn main() -> anyhow::Result<()> {
         example_cli::Commands::ChainSpecific(esplora_cmd) => esplora_cmd,
         // These are general commands handled by example_cli. Execute the cmd and return.
         general_cmd => {
-            let res = example_cli::handle_commands(
+            return example_cli::handle_commands(
                 &graph,
                 &db,
                 &chain,
@@ -140,9 +140,6 @@ fn main() -> anyhow::Result<()> {
                 },
                 general_cmd.clone(),
             );
-
-            db.lock().unwrap().commit()?;
-            return res;
         }
     };
 

example-crates/wallet_rpc/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "wallet_rpc"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+bdk = { path = "../../crates/bdk" }
+bdk_file_store = { path = "../../crates/file_store" }
+bdk_bitcoind_rpc = { path = "../../crates/bitcoind_rpc" }
+
+anyhow = "1"
+clap = { version = "3.2.25", features = ["derive", "env"] }
+ctrlc = "2.0.1"

example-crates/wallet_rpc/README.md

@@ -0,0 +1,45 @@
+# Wallet RPC Example 
+
+```
+$ cargo run --bin wallet_rpc -- --help
+
+wallet_rpc 0.1.0
+Bitcoind RPC example usign `bdk::Wallet`
+
+USAGE:
+    wallet_rpc [OPTIONS] <DESCRIPTOR> [CHANGE_DESCRIPTOR]
+
+ARGS:
+    <DESCRIPTOR>           Wallet descriptor [env: DESCRIPTOR=]
+    <CHANGE_DESCRIPTOR>    Wallet change descriptor [env: CHANGE_DESCRIPTOR=]
+
+OPTIONS:
+        --db-path <DB_PATH>
+            Where to store wallet data [env: BDK_DB_PATH=] [default: .bdk_wallet_rpc_example.db]
+
+    -h, --help
+            Print help information
+
+        --network <NETWORK>
+            Bitcoin network to connect to [env: BITCOIN_NETWORK=] [default: testnet]
+
+        --rpc-cookie <RPC_COOKIE>
+            RPC auth cookie file [env: RPC_COOKIE=]
+
+        --rpc-pass <RPC_PASS>
+            RPC auth password [env: RPC_PASS=]
+
+        --rpc-user <RPC_USER>
+            RPC auth username [env: RPC_USER=]
+
+        --start-height <START_HEIGHT>
+            Earliest block height to start sync from [env: START_HEIGHT=] [default: 481824]
+
+        --url <URL>
+            RPC URL [env: RPC_URL=] [default: 127.0.0.1:8332]
+
+    -V, --version
+            Print version information
+
+```
+

example-crates/wallet_rpc/src/main.rs

@@ -0,0 +1,182 @@
+use bdk::{
+    bitcoin::{Block, Network, Transaction},
+    wallet::Wallet,
+};
+use bdk_bitcoind_rpc::{
+    bitcoincore_rpc::{Auth, Client, RpcApi},
+    Emitter,
+};
+use bdk_file_store::Store;
+use clap::{self, Parser};
+use std::{path::PathBuf, sync::mpsc::sync_channel, thread::spawn, time::Instant};
+
+const DB_MAGIC: &str = "bdk-rpc-wallet-example";
+
+/// Bitcoind RPC example usign `bdk::Wallet`.
+///
+/// This syncs the chain block-by-block and prints the current balance, transaction count and UTXO
+/// count.
+#[derive(Parser, Debug)]
+#[clap(author, version, about, long_about = None)]
+#[clap(propagate_version = true)]
+pub struct Args {
+    /// Wallet descriptor
+    #[clap(env = "DESCRIPTOR")]
+    pub descriptor: String,
+    /// Wallet change descriptor
+    #[clap(env = "CHANGE_DESCRIPTOR")]
+    pub change_descriptor: Option<String>,
+    /// Earliest block height to start sync from
+    #[clap(env = "START_HEIGHT", long, default_value = "481824")]
+    pub start_height: u32,
+    /// Bitcoin network to connect to
+    #[clap(env = "BITCOIN_NETWORK", long, default_value = "testnet")]
+    pub network: Network,
+    /// Where to store wallet data
+    #[clap(
+        env = "BDK_DB_PATH",
+        long,
+        default_value = ".bdk_wallet_rpc_example.db"
+    )]
+    pub db_path: PathBuf,
+
+    /// RPC URL
+    #[clap(env = "RPC_URL", long, default_value = "127.0.0.1:8332")]
+    pub url: String,
+    /// RPC auth cookie file
+    #[clap(env = "RPC_COOKIE", long)]
+    pub rpc_cookie: Option<PathBuf>,
+    /// RPC auth username
+    #[clap(env = "RPC_USER", long)]
+    pub rpc_user: Option<String>,
+    /// RPC auth password
+    #[clap(env = "RPC_PASS", long)]
+    pub rpc_pass: Option<String>,
+}
+
+impl Args {
+    fn client(&self) -> anyhow::Result<Client> {
+        Ok(Client::new(
+            &self.url,
+            match (&self.rpc_cookie, &self.rpc_user, &self.rpc_pass) {
+                (None, None, None) => Auth::None,
+                (Some(path), _, _) => Auth::CookieFile(path.clone()),
+                (_, Some(user), Some(pass)) => Auth::UserPass(user.clone(), pass.clone()),
+                (_, Some(_), None) => panic!("rpc auth: missing rpc_pass"),
+                (_, None, Some(_)) => panic!("rpc auth: missing rpc_user"),
+            },
+        )?)
+    }
+}
+
+#[derive(Debug)]
+enum Emission {
+    SigTerm,
+    Block(bdk_bitcoind_rpc::BlockEvent<Block>),
+    Mempool(Vec<(Transaction, u64)>),
+}
+
+fn main() -> anyhow::Result<()> {
+    let args = Args::parse();
+
+    let rpc_client = args.client()?;
+    println!(
+        "Connected to Bitcoin Core RPC at {:?}",
+        rpc_client.get_blockchain_info().unwrap()
+    );
+
+    let start_load_wallet = Instant::now();
+    let mut wallet = Wallet::new_or_load(
+        &args.descriptor,
+        args.change_descriptor.as_ref(),
+        Store::<bdk::wallet::ChangeSet>::open_or_create_new(DB_MAGIC.as_bytes(), args.db_path)?,
+        args.network,
+    )?;
+    println!(
+        "Loaded wallet in {}s",
+        start_load_wallet.elapsed().as_secs_f32()
+    );
+
+    let balance = wallet.get_balance();
+    println!("Wallet balance before syncing: {} sats", balance.total());
+
+    let wallet_tip = wallet.latest_checkpoint();
+    println!(
+        "Wallet tip: {} at height {}",
+        wallet_tip.hash(),
+        wallet_tip.height()
+    );
+
+    let (sender, receiver) = sync_channel::<Emission>(21);
+
+    let signal_sender = sender.clone();
+    ctrlc::set_handler(move || {
+        signal_sender
+            .send(Emission::SigTerm)
+            .expect("failed to send sigterm")
+    });
+
+    let emitter_tip = wallet_tip.clone();
+    spawn(move || -> Result<(), anyhow::Error> {
+        let mut emitter = Emitter::new(&rpc_client, emitter_tip, args.start_height);
+        while let Some(emission) = emitter.next_block()? {
+            sender.send(Emission::Block(emission))?;
+        }
+        sender.send(Emission::Mempool(emitter.mempool()?))?;
+        Ok(())
+    });
+
+    let mut blocks_received = 0_usize;
+    for emission in receiver {
+        match emission {
+            Emission::SigTerm => {
+                println!("Sigterm received, exiting...");
+                break;
+            }
+            Emission::Block(block_emission) => {
+                blocks_received += 1;
+                let height = block_emission.block_height();
+                let hash = block_emission.block_hash();
+                let connected_to = block_emission.connected_to();
+                let start_apply_block = Instant::now();
+                wallet.apply_block_connected_to(&block_emission.block, height, connected_to)?;
+                wallet.commit()?;
+                let elapsed = start_apply_block.elapsed().as_secs_f32();
+                println!(
+                    "Applied block {} at height {} in {}s",
+                    hash, height, elapsed
+                );
+            }
+            Emission::Mempool(mempool_emission) => {
+                let start_apply_mempool = Instant::now();
+                wallet.apply_unconfirmed_txs(mempool_emission.iter().map(|(tx, time)| (tx, *time)));
+                wallet.commit()?;
+                println!(
+                    "Applied unconfirmed transactions in {}s",
+                    start_apply_mempool.elapsed().as_secs_f32()
+                );
+                break;
+            }
+        }
+    }
+    let wallet_tip_end = wallet.latest_checkpoint();
+    let balance = wallet.get_balance();
+    println!(
+        "Synced {} blocks in {}s",
+        blocks_received,
+        start_load_wallet.elapsed().as_secs_f32(),
+    );
+    println!(
+        "Wallet tip is '{}:{}'",
+        wallet_tip_end.height(),
+        wallet_tip_end.hash()
+    );
+    println!("Wallet balance is {} sats", balance.total());
+    println!(
+        "Wallet has {} transactions and {} utxos",
+        wallet.transactions().count(),
+        wallet.list_unspent().count()
+    );
+
+    Ok(())
+}