Merge #1839: Introduce evicted-at/last-evicted timestamps

0a20724 feat(examples): Update example crates to use `expected_spk_txids` (志宇)
1f8fc17 feat(core)!: Remove redundant `SyncRequest` methods (志宇)
f42d5a8 feat(esplora): Handle spks with expected txids (志宇)
3ab4994 feat(electrum): Handle spks with expected txids (志宇)
64e4100 feat(chain): Add `TxGraph` methods that handle expected spk txids (志宇)
b38569f feat(core): Add expected txids to `SyncRequest` spks (Wei Chen)
a2dfcb9 feat(chain)!: Change `TxGraph` to understand evicted-at & last-evicted (志宇)
ae0336b feat(core): Add `TxUpdate::evicted_ats` (志宇)

Pull request description:

  Partially Fixes #1740.
  Replaces #1765.
  Replaces #1811.

  ### Description

  This PR allows the receiving structures (`bdk_chain`, `bdk_wallet`) to detect and evict incoming transactions that are double spent (cancelled).

  We add a new field to `TxUpdate` (`TxUpdate::evicted_ats`), which in turn, updates the `last_evicted` timestamps that are tracked/persisted by `TxGraph`. This is similar to how `TxUpdate::seen_ats` update the `last_seen` timestamp in `TxGraph`. Transactions with a `last_evicted` timestamp higher than their `last_seen` timestamp are considered evicted.

  `SpkWithExpectedTxids` is introduced in `SpkClient` to track expected `Txid`s for each `spk`. During a sync, if any `Txid`s from `SpkWithExpectedTxids` are not in the current history of an `spk` obtained from the chain source, those `Txid`s are considered missing. Support for `SpkWithExpectedTxids` has been added to both `bdk_electrum` and `bdk_esplora` chain source crates.

  The canonicalization algorithm is updated to disregard transactions with a `last_evicted` timestamp greater than or equal to their `last_seen` timestamp, except in cases where transitivity rules apply.

  ### Notes to the reviewers

  This PR does not fix #1740 for block-by-block chain source (such as `bdk_bitcoind_rpc`). This work is done in a separate PR (#1857).

  ### Changelog notice

  * Add `TxUpdate::evicted_ats` which tracks transactions that have been replaced and are no longer present in mempool.
  * Change `TxGraph` to track `last_evicted` timestamps. This is when a transaction is last marked as missing from the mempool.
  * The canonicalization algorithm now disregards transactions with a `last_evicted` timestamp greater than or equal to it's `last_seen` timestamp, except when a canonical descendant exists due to rules of transitivity.
  * Add `SpkWithExpectedTxids` in `spk_client` which keeps track of expected `Txid`s for each `spk`.
  * Change `bdk_electrum` and `bdk_esplora` to understand `SpkWithExpectedTxids`.
  * Add `SyncRequestBuilder::expected_txids_of_spk` method which adds an association between `txid`s and `spk`s.

  ### 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

  #### New Features:

  * [x] I've added tests for the new feature
  * [x] I've added docs for the new feature

  #### Bugfixes:

  * [x] This pull request breaks the existing API
  * [x] I've added tests to reproduce the issue which are now passing
  * [x] I'm linking the issue being fixed by this PR

ACKs for top commit:
  LLFourn:
    ACK 0a20724

Tree-SHA512: 29ef964e4597aa9f4cf02e9997b0d17cb91ec2f0f1187b0e9ade3709636b873c2a7cbe803facbc4686a3050a2abeb3e9cc40f9308f8cded9c9353734dcc5755b

Author: evanlinjin

.github/workflows/cont_integration.yml

@@ -96,7 +96,7 @@ jobs:
       - name: Check esplora
         working-directory: ./crates/esplora
         # TODO "--target thumbv6m-none-eabi" should work but currently does not
-        run: cargo check --no-default-features --features miniscript/no-std,bdk_chain/hashbrown
+        run: cargo check --no-default-features --features bdk_chain/hashbrown
 
   check-wasm:
     needs: prepare
@@ -128,7 +128,7 @@ jobs:
         run: cargo check --target wasm32-unknown-unknown --no-default-features --features miniscript/no-std,bdk_chain/hashbrown
       - name: Check esplora
         working-directory: ./crates/esplora
-        run: cargo check --target wasm32-unknown-unknown --no-default-features --features miniscript/no-std,bdk_chain/hashbrown,async
+        run: cargo check --target wasm32-unknown-unknown --no-default-features --features bdk_core/hashbrown,async
 
   fmt:
     needs: prepare

crates/chain/src/indexed_tx_graph.rs

@@ -1,13 +1,18 @@
 //! Contains the [`IndexedTxGraph`] and associated types. Refer to the
 //! [`IndexedTxGraph`] documentation for more.
-use core::fmt::Debug;
+use core::{
+    convert::Infallible,
+    fmt::{self, Debug},
+    ops::RangeBounds,
+};
 
 use alloc::{sync::Arc, vec::Vec};
-use bitcoin::{Block, OutPoint, Transaction, TxOut, Txid};
+use bitcoin::{Block, OutPoint, ScriptBuf, Transaction, TxOut, Txid};
 
 use crate::{
+    spk_txout::SpkTxOutIndex,
     tx_graph::{self, TxGraph},
-    Anchor, BlockId, Indexer, Merge, TxPosInBlock,
+    Anchor, BlockId, ChainOracle, Indexer, Merge, TxPosInBlock,
 };
 
 /// The [`IndexedTxGraph`] combines a [`TxGraph`] and an [`Indexer`] implementation.
@@ -127,6 +132,19 @@ where
         self.graph.insert_seen_at(txid, seen_at).into()
     }
 
+    /// Inserts the given `evicted_at` for `txid`.
+    ///
+    /// The `evicted_at` timestamp represents the last known time when the transaction was observed
+    /// to be missing from the mempool. If `txid` was previously recorded with an earlier
+    /// `evicted_at` value, it is updated only if the new value is greater.
+    pub fn insert_evicted_at(&mut self, txid: Txid, evicted_at: u64) -> ChangeSet<A, I::ChangeSet> {
+        let tx_graph = self.graph.insert_evicted_at(txid, evicted_at);
+        ChangeSet {
+            tx_graph,
+            ..Default::default()
+        }
+    }
+
     /// Batch insert transactions, filtering out those that are irrelevant.
     ///
     /// Relevancy is determined by the [`Indexer::is_tx_relevant`] implementation of `I`. Irrelevant
@@ -301,6 +319,58 @@ where
     }
 }
 
+impl<A, X> IndexedTxGraph<A, X>
+where
+    A: Anchor,
+{
+    /// List txids that are expected to exist under the given spks.
+    ///
+    /// This is used to fill [`SyncRequestBuilder::expected_spk_txids`](bdk_core::spk_client::SyncRequestBuilder::expected_spk_txids).
+    ///
+    /// The spk index range can be contrained with `range`.
+    ///
+    /// # Error
+    ///
+    /// If the [`ChainOracle`] implementation (`chain`) fails, an error will be returned with the
+    /// returned item.
+    ///
+    /// If the [`ChainOracle`] is infallible,
+    /// [`list_expected_spk_txids`](Self::list_expected_spk_txids) can be used instead.
+    pub fn try_list_expected_spk_txids<'a, C, I>(
+        &'a self,
+        chain: &'a C,
+        chain_tip: BlockId,
+        spk_index_range: impl RangeBounds<I> + 'a,
+    ) -> impl Iterator<Item = Result<(ScriptBuf, Txid), C::Error>> + 'a
+    where
+        C: ChainOracle,
+        X: AsRef<SpkTxOutIndex<I>> + 'a,
+        I: fmt::Debug + Clone + Ord + 'a,
+    {
+        self.graph
+            .try_list_expected_spk_txids(chain, chain_tip, &self.index, spk_index_range)
+    }
+
+    /// List txids that are expected to exist under the given spks.
+    ///
+    /// This is the infallible version of
+    /// [`try_list_expected_spk_txids`](Self::try_list_expected_spk_txids).
+    pub fn list_expected_spk_txids<'a, C, I>(
+        &'a self,
+        chain: &'a C,
+        chain_tip: BlockId,
+        spk_index_range: impl RangeBounds<I> + 'a,
+    ) -> impl Iterator<Item = (ScriptBuf, Txid)> + 'a
+    where
+        C: ChainOracle<Error = Infallible>,
+        X: AsRef<SpkTxOutIndex<I>> + 'a,
+        I: fmt::Debug + Clone + Ord + 'a,
+    {
+        self.try_list_expected_spk_txids(chain, chain_tip, spk_index_range)
+            .map(|r| r.expect("infallible"))
+    }
+}
+
 impl<A, I> AsRef<TxGraph<A>> for IndexedTxGraph<A, I> {
     fn as_ref(&self) -> &TxGraph<A> {
         &self.graph

crates/chain/src/indexer/keychain_txout.rs

@@ -136,6 +136,12 @@ impl<K> Default for KeychainTxOutIndex<K> {
     }
 }
 
+impl<K> AsRef<SpkTxOutIndex<(K, u32)>> for KeychainTxOutIndex<K> {
+    fn as_ref(&self) -> &SpkTxOutIndex<(K, u32)> {
+        &self.inner
+    }
+}
+
 impl<K: Clone + Ord + Debug> Indexer for KeychainTxOutIndex<K> {
     type ChangeSet = ChangeSet;
 
@@ -200,6 +206,11 @@ impl<K> KeychainTxOutIndex<K> {
             lookahead,
         }
     }
+
+    /// Get a reference to the internal [`SpkTxOutIndex`].
+    pub fn inner(&self) -> &SpkTxOutIndex<(K, u32)> {
+        &self.inner
+    }
 }
 
 /// Methods that are *re-exposed* from the internal [`SpkTxOutIndex`].

crates/chain/src/indexer/spk_txout.rs

@@ -54,6 +54,12 @@ impl<I> Default for SpkTxOutIndex<I> {
     }
 }
 
+impl<I> AsRef<SpkTxOutIndex<I>> for SpkTxOutIndex<I> {
+    fn as_ref(&self) -> &SpkTxOutIndex<I> {
+        self
+    }
+}
+
 impl<I: Clone + Ord + core::fmt::Debug> Indexer for SpkTxOutIndex<I> {
     type ChangeSet = ();
 
@@ -334,4 +340,24 @@ impl<I: Clone + Ord + core::fmt::Debug> SpkTxOutIndex<I> {
             .any(|output| self.spk_indices.contains_key(&output.script_pubkey));
         input_matches || output_matches
     }
+
+    /// Find relevant script pubkeys associated with a transaction for tracking and validation.
+    ///
+    /// Returns a set of script pubkeys from [`SpkTxOutIndex`] that are relevant to the outputs and
+    /// previous outputs of a given transaction. Inputs are only considered relevant if the parent
+    /// transactions have been scanned.
+    pub fn relevant_spks_of_tx(&self, tx: &Transaction) -> BTreeSet<(I, ScriptBuf)> {
+        let spks_from_inputs = tx.input.iter().filter_map(|txin| {
+            self.txouts
+                .get(&txin.previous_output)
+                .cloned()
+                .map(|(i, prev_txo)| (i, prev_txo.script_pubkey))
+        });
+        let spks_from_outputs = tx
+            .output
+            .iter()
+            .filter_map(|txout| self.spk_indices.get_key_value(&txout.script_pubkey))
+            .map(|(spk, i)| (i.clone(), spk.clone()));
+        spks_from_inputs.chain(spks_from_outputs).collect()
+    }
 }

crates/chain/src/rusqlite_impl.rs

@@ -264,12 +264,20 @@ impl tx_graph::ChangeSet<ConfirmationBlockTime> {
         format!("{add_confirmation_time_column}; {extract_confirmation_time_from_anchor_column}; {drop_anchor_column}")
     }
 
+    /// Get v2 of sqlite [tx_graph::ChangeSet] schema
+    pub fn schema_v2() -> String {
+        format!(
+            "ALTER TABLE {} ADD COLUMN last_evicted INTEGER",
+            Self::TXS_TABLE_NAME,
+        )
+    }
+
     /// Initialize sqlite tables.
     pub fn init_sqlite_tables(db_tx: &rusqlite::Transaction) -> rusqlite::Result<()> {
         migrate_schema(
             db_tx,
             Self::SCHEMA_NAME,
-            &[&Self::schema_v0(), &Self::schema_v1()],
+            &[&Self::schema_v0(), &Self::schema_v1(), &Self::schema_v2()],
         )
     }
 
@@ -280,24 +288,28 @@ impl tx_graph::ChangeSet<ConfirmationBlockTime> {
         let mut changeset = Self::default();
 
         let mut statement = db_tx.prepare(&format!(
-            "SELECT txid, raw_tx, last_seen FROM {}",
+            "SELECT txid, raw_tx, last_seen, last_evicted FROM {}",
             Self::TXS_TABLE_NAME,
         ))?;
         let row_iter = statement.query_map([], |row| {
             Ok((
                 row.get::<_, Impl<bitcoin::Txid>>("txid")?,
                 row.get::<_, Option<Impl<bitcoin::Transaction>>>("raw_tx")?,
                 row.get::<_, Option<u64>>("last_seen")?,
+                row.get::<_, Option<u64>>("last_evicted")?,
             ))
         })?;
         for row in row_iter {
-            let (Impl(txid), tx, last_seen) = row?;
+            let (Impl(txid), tx, last_seen, last_evicted) = row?;
             if let Some(Impl(tx)) = tx {
                 changeset.txs.insert(Arc::new(tx));
             }
             if let Some(last_seen) = last_seen {
                 changeset.last_seen.insert(txid, last_seen);
             }
+            if let Some(last_evicted) = last_evicted {
+                changeset.last_evicted.insert(txid, last_evicted);
+            }
         }
 
         let mut statement = db_tx.prepare(&format!(
@@ -377,6 +389,19 @@ impl tx_graph::ChangeSet<ConfirmationBlockTime> {
             })?;
         }
 
+        let mut statement = db_tx
+            .prepare_cached(&format!(
+                "INSERT INTO {}(txid, last_evicted) VALUES(:txid, :last_evicted) ON CONFLICT(txid) DO UPDATE SET last_evicted=:last_evicted",
+                Self::TXS_TABLE_NAME,
+            ))?;
+        for (&txid, &last_evicted) in &self.last_evicted {
+            let checked_time = last_evicted.to_sql()?;
+            statement.execute(named_params! {
+                ":txid": Impl(txid),
+                ":last_evicted": Some(checked_time),
+            })?;
+        }
+
         let mut statement = db_tx.prepare_cached(&format!(
             "REPLACE INTO {}(txid, vout, value, script) VALUES(:txid, :vout, :value, :script)",
             Self::TXOUTS_TABLE_NAME,
@@ -628,7 +653,7 @@ mod test {
     }
 
     #[test]
-    fn v0_to_v1_schema_migration_is_backward_compatible() -> anyhow::Result<()> {
+    fn v0_to_v2_schema_migration_is_backward_compatible() -> anyhow::Result<()> {
         type ChangeSet = tx_graph::ChangeSet<ConfirmationBlockTime>;
         let mut conn = rusqlite::Connection::open_in_memory()?;
 
@@ -697,13 +722,17 @@ mod test {
             }
         }
 
-        // Apply v1 sqlite schema to tables with data
+        // Apply v1 & v2 sqlite schema to tables with data
         {
             let db_tx = conn.transaction()?;
             migrate_schema(
                 &db_tx,
                 ChangeSet::SCHEMA_NAME,
-                &[&ChangeSet::schema_v0(), &ChangeSet::schema_v1()],
+                &[
+                    &ChangeSet::schema_v0(),
+                    &ChangeSet::schema_v1(),
+                    &ChangeSet::schema_v2(),
+                ],
             )?;
             db_tx.commit()?;
         }
@@ -718,4 +747,43 @@ mod test {
 
         Ok(())
     }
+
+    #[test]
+    fn can_persist_last_evicted() -> anyhow::Result<()> {
+        use bitcoin::hashes::Hash;
+
+        type ChangeSet = tx_graph::ChangeSet<ConfirmationBlockTime>;
+        let mut conn = rusqlite::Connection::open_in_memory()?;
+
+        // Init tables
+        {
+            let db_tx = conn.transaction()?;
+            ChangeSet::init_sqlite_tables(&db_tx)?;
+            db_tx.commit()?;
+        }
+
+        let txid = bitcoin::Txid::all_zeros();
+        let last_evicted = 100;
+
+        // Persist `last_evicted`
+        {
+            let changeset = ChangeSet {
+                last_evicted: [(txid, last_evicted)].into(),
+                ..Default::default()
+            };
+            let db_tx = conn.transaction()?;
+            changeset.persist_to_sqlite(&db_tx)?;
+            db_tx.commit()?;
+        }
+
+        // Load from sqlite should succeed
+        {
+            let db_tx = conn.transaction()?;
+            let changeset = ChangeSet::from_sqlite(&db_tx)?;
+            db_tx.commit()?;
+            assert_eq!(changeset.last_evicted.get(&txid), Some(&last_evicted));
+        }
+
+        Ok(())
+    }
 }