feat!: Improve spk-based syncing flow

* Change `TxUpdate::seen_ats` to be a `HashSet<(Txid, u64)>` so we can
  introduce multiple timestamps per tx. This is useful to introduce both
  `first_seen` and `last_seen` timestamps to `TxGraph`. This is also a
  better API for chain-sources as they can just insert timestamps into
  the field without checking previous values.
* Change sync/full-scan flow to have the request structure introduce the
  sync time instead of introducing the timestamp when apply the
  `TxUpdate`. This simplifies the `apply_update{_at}` logic and makes
  `evicted_at` easier to reason about (in the future).

Author: evanlinjin

crates/chain/benches/canonicalization.rs

@@ -133,8 +133,9 @@ pub fn many_conflicting_unconfirmed(c: &mut Criterion) {
                 ..new_tx(i)
             };
             let mut update = TxUpdate::default();
+            update.seen_ats = [(tx.compute_txid(), i as u64)].into();
             update.txs = vec![Arc::new(tx)];
-            let _ = tx_graph.apply_update_at(update, Some(i as u64));
+            let _ = tx_graph.apply_update(update);
         }
     }));
     c.bench_function("many_conflicting_unconfirmed::list_canonical_txs", {
@@ -168,8 +169,9 @@ pub fn many_chained_unconfirmed(c: &mut Criterion) {
             };
             let txid = tx.compute_txid();
             let mut update = TxUpdate::default();
+            update.seen_ats = [(txid, i as u64)].into();
             update.txs = vec![Arc::new(tx)];
-            let _ = tx_graph.apply_update_at(update, Some(i as u64));
+            let _ = tx_graph.apply_update(update);
             // Store the next prevout.
             previous_output = OutPoint::new(txid, 0);
         }

crates/chain/src/indexed_tx_graph.rs

@@ -91,37 +91,12 @@ where
     /// Apply an `update` directly.
     ///
     /// `update` is a [`tx_graph::TxUpdate<A>`] and the resultant changes is returned as [`ChangeSet`].
-    #[cfg(feature = "std")]
-    #[cfg_attr(docsrs, doc(cfg(feature = "std")))]
     pub fn apply_update(&mut self, update: tx_graph::TxUpdate<A>) -> ChangeSet<A, I::ChangeSet> {
         let tx_graph = self.graph.apply_update(update);
         let indexer = self.index_tx_graph_changeset(&tx_graph);
         ChangeSet { tx_graph, indexer }
     }
 
-    /// Apply the given `update` with an optional `seen_at` timestamp.
-    ///
-    /// `seen_at` represents when the update is seen (in unix seconds). It is used to determine the
-    /// `last_seen`s for all transactions in the update which have no corresponding anchor(s). The
-    /// `last_seen` value is used internally to determine precedence of conflicting unconfirmed
-    /// transactions (where the transaction with the lower `last_seen` value is omitted from the
-    /// canonical history).
-    ///
-    /// Not setting a `seen_at` value means unconfirmed transactions introduced by this update will
-    /// not be part of the canonical history of transactions.
-    ///
-    /// Use [`apply_update`](IndexedTxGraph::apply_update) to have the `seen_at` value automatically
-    /// set to the current time.
-    pub fn apply_update_at(
-        &mut self,
-        update: tx_graph::TxUpdate<A>,
-        seen_at: Option<u64>,
-    ) -> ChangeSet<A, I::ChangeSet> {
-        let tx_graph = self.graph.apply_update_at(update, seen_at);
-        let indexer = self.index_tx_graph_changeset(&tx_graph);
-        ChangeSet { tx_graph, indexer }
-    }
-
     /// Insert a floating `txout` of given `outpoint`.
     pub fn insert_txout(&mut self, outpoint: OutPoint, txout: TxOut) -> ChangeSet<A, I::ChangeSet> {
         let graph = self.graph.insert_txout(outpoint, txout);

crates/chain/src/tx_graph.rs

@@ -129,7 +129,7 @@ impl<A: Ord> From<TxGraph<A>> for TxUpdate<A> {
 impl<A: Anchor> From<TxUpdate<A>> for TxGraph<A> {
     fn from(update: TxUpdate<A>) -> Self {
         let mut graph = TxGraph::<A>::default();
-        let _ = graph.apply_update_at(update, None);
+        let _ = graph.apply_update(update);
         graph
     }
 }
@@ -719,52 +719,20 @@ impl<A: Anchor> TxGraph<A> {
     ///
     /// The returned [`ChangeSet`] is the set difference between `update` and `self` (transactions that
     /// exist in `update` but not in `self`).
-    #[cfg(feature = "std")]
-    #[cfg_attr(docsrs, doc(cfg(feature = "std")))]
     pub fn apply_update(&mut self, update: TxUpdate<A>) -> ChangeSet<A> {
-        use std::time::*;
-        let now = SystemTime::now()
-            .duration_since(UNIX_EPOCH)
-            .expect("current time must be greater than epoch anchor");
-        self.apply_update_at(update, Some(now.as_secs()))
-    }
-
-    /// Extends this graph with the given `update` alongside an optional `seen_at` timestamp.
-    ///
-    /// `seen_at` represents when the update is seen (in unix seconds). It is used to determine the
-    /// `last_seen`s for all transactions in the update which have no corresponding anchor(s). The
-    /// `last_seen` value is used internally to determine precedence of conflicting unconfirmed
-    /// transactions (where the transaction with the lower `last_seen` value is omitted from the
-    /// canonical history).
-    ///
-    /// Not setting a `seen_at` value means unconfirmed transactions introduced by this update will
-    /// not be part of the canonical history of transactions.
-    ///
-    /// Use [`apply_update`](TxGraph::apply_update) to have the `seen_at` value automatically set
-    /// to the current time.
-    pub fn apply_update_at(&mut self, update: TxUpdate<A>, seen_at: Option<u64>) -> ChangeSet<A> {
         let mut changeset = ChangeSet::<A>::default();
-        let mut unanchored_txs = HashSet::<Txid>::new();
         for tx in update.txs {
-            if unanchored_txs.insert(tx.compute_txid()) {
-                changeset.merge(self.insert_tx(tx));
-            }
+            changeset.merge(self.insert_tx(tx));
         }
         for (outpoint, txout) in update.txouts {
             changeset.merge(self.insert_txout(outpoint, txout));
         }
         for (anchor, txid) in update.anchors {
-            unanchored_txs.remove(&txid);
             changeset.merge(self.insert_anchor(txid, anchor));
         }
         for (txid, seen_at) in update.seen_ats {
             changeset.merge(self.insert_seen_at(txid, seen_at));
         }
-        if let Some(seen_at) = seen_at {
-            for txid in unanchored_txs {
-                changeset.merge(self.insert_seen_at(txid, seen_at));
-            }
-        }
         changeset
     }
 

crates/chain/tests/test_tx_graph.rs

@@ -94,7 +94,7 @@ fn insert_txouts() {
             // Insert partials transactions.
             update.txouts.insert(*outpoint, txout.clone());
             // Mark them unconfirmed.
-            update.seen_ats.insert(outpoint.txid, unconf_seen_at);
+            update.seen_ats.insert((outpoint.txid, unconf_seen_at));
         }
 
         // Insert the full transaction.
@@ -1289,7 +1289,7 @@ fn tx_graph_update_conversion() {
 
     for (test_name, update) in test_cases {
         let mut tx_graph = TxGraph::<ConfirmationBlockTime>::default();
-        let _ = tx_graph.apply_update_at(update.clone(), None);
+        let _ = tx_graph.apply_update(update.clone());
         let update_from_tx_graph: TxUpdate<ConfirmationBlockTime> = tx_graph.into();
 
         assert_eq!(

crates/core/src/spk_client.rs

@@ -87,19 +87,13 @@ impl SyncProgress {
 }
 
 /// Builds a [`SyncRequest`].
+///
+/// Construct with [`SyncRequest::builder`].
 #[must_use]
 pub struct SyncRequestBuilder<I = ()> {
     inner: SyncRequest<I>,
 }
 
-impl<I> Default for SyncRequestBuilder<I> {
-    fn default() -> Self {
-        Self {
-            inner: Default::default(),
-        }
-    }
-}
-
 impl SyncRequestBuilder<()> {
     /// Add [`Script`]s that will be synced against.
     pub fn spks(self, spks: impl IntoIterator<Item = ScriptBuf>) -> Self {
@@ -210,6 +204,7 @@ impl<I> SyncRequestBuilder<I> {
 /// ```
 #[must_use]
 pub struct SyncRequest<I = ()> {
+    start_time: u64,
     chain_tip: Option<CheckPoint>,
     spks: VecDeque<(I, ScriptBuf)>,
     spks_consumed: usize,
@@ -220,35 +215,56 @@ pub struct SyncRequest<I = ()> {
     inspect: Box<InspectSync<I>>,
 }
 
-impl<I> Default for SyncRequest<I> {
-    fn default() -> Self {
-        Self {
-            chain_tip: None,
-            spks: VecDeque::new(),
-            spks_consumed: 0,
-            txids: VecDeque::new(),
-            txids_consumed: 0,
-            outpoints: VecDeque::new(),
-            outpoints_consumed: 0,
-            inspect: Box::new(|_, _| {}),
-        }
-    }
-}
-
 impl<I> From<SyncRequestBuilder<I>> for SyncRequest<I> {
     fn from(builder: SyncRequestBuilder<I>) -> Self {
         builder.inner
     }
 }
 
 impl<I> SyncRequest<I> {
-    /// Start building a [`SyncRequest`].
-    pub fn builder() -> SyncRequestBuilder<I> {
+    /// Start building [`SyncRequest`] with a given `start_time`.
+    ///
+    /// `start_time` specifies the start time of sync. Chain sources can use this value to set
+    /// [`TxUpdate::seen_ats`](crate::TxUpdate::seen_ats) for mempool transactions. A transaction
+    /// without any `seen_ats` is assumed to be unseen in the mempool.
+    ///
+    /// Use [`SyncRequest::builder`] to use the current timestamp as `start_time` (this requires
+    /// `feature = "std"`).
+    pub fn builder_at(start_time: u64) -> SyncRequestBuilder<I> {
         SyncRequestBuilder {
-            inner: Default::default(),
+            inner: Self {
+                start_time,
+                chain_tip: None,
+                spks: VecDeque::new(),
+                spks_consumed: 0,
+                txids: VecDeque::new(),
+                txids_consumed: 0,
+                outpoints: VecDeque::new(),
+                outpoints_consumed: 0,
+                inspect: Box::new(|_, _| ()),
+            },
         }
     }
 
+    /// Start building [`SyncRequest`] with the current timestamp as the `start_time`.
+    ///
+    /// Use [`SyncRequest::builder_at`] to manually set the `start_time`, or if `feature = "std"`
+    /// is not available.
+    #[cfg(feature = "std")]
+    #[cfg_attr(docsrs, doc(cfg(feature = "std")))]
+    pub fn builder() -> SyncRequestBuilder<I> {
+        let start_time = std::time::UNIX_EPOCH
+            .elapsed()
+            .expect("failed to get current timestamp")
+            .as_secs();
+        Self::builder_at(start_time)
+    }
+
+    /// When the sync-request was initiated.
+    pub fn start_time(&self) -> u64 {
+        self.start_time
+    }
+
     /// Get the [`SyncProgress`] of this request.
     pub fn progress(&self) -> SyncProgress {
         SyncProgress {
@@ -339,19 +355,13 @@ impl<A> Default for SyncResponse<A> {
 }
 
 /// Builds a [`FullScanRequest`].
+///
+/// Construct with [`FullScanRequest::builder`].
 #[must_use]
 pub struct FullScanRequestBuilder<K> {
     inner: FullScanRequest<K>,
 }
 
-impl<K> Default for FullScanRequestBuilder<K> {
-    fn default() -> Self {
-        Self {
-            inner: Default::default(),
-        }
-    }
-}
-
 impl<K: Ord> FullScanRequestBuilder<K> {
     /// Set the initial chain tip for the full scan request.
     ///
@@ -397,6 +407,7 @@ impl<K: Ord> FullScanRequestBuilder<K> {
 /// [`chain_tip`](FullScanRequestBuilder::chain_tip) (if provided).
 #[must_use]
 pub struct FullScanRequest<K> {
+    start_time: u64,
     chain_tip: Option<CheckPoint>,
     spks_by_keychain: BTreeMap<K, Box<dyn Iterator<Item = Indexed<ScriptBuf>> + Send>>,
     inspect: Box<InspectFullScan<K>>,
@@ -408,22 +419,43 @@ impl<K> From<FullScanRequestBuilder<K>> for FullScanRequest<K> {
     }
 }
 
-impl<K> Default for FullScanRequest<K> {
-    fn default() -> Self {
-        Self {
-            chain_tip: None,
-            spks_by_keychain: Default::default(),
-            inspect: Box::new(|_, _, _| {}),
+impl<K: Ord + Clone> FullScanRequest<K> {
+    /// Start building a [`FullScanRequest`] with a given `start_time`.
+    ///
+    /// `start_time` specifies the start time of sync. Chain sources can use this value to set
+    /// [`TxUpdate::seen_ats`](crate::TxUpdate::seen_ats) for mempool transactions. A transaction
+    /// without any `seen_ats` is assumed to be unseen in the mempool.
+    ///
+    /// Use [`FullScanRequest::builder`] to use the current timestamp as `start_time` (this
+    /// requires `feature = "std`).
+    pub fn builder_at(start_time: u64) -> FullScanRequestBuilder<K> {
+        FullScanRequestBuilder {
+            inner: Self {
+                start_time,
+                chain_tip: None,
+                spks_by_keychain: BTreeMap::new(),
+                inspect: Box::new(|_, _, _| ()),
+            },
         }
     }
-}
 
-impl<K: Ord + Clone> FullScanRequest<K> {
-    /// Start building a [`FullScanRequest`].
+    /// Start building a [`FullScanRequest`] with the current timestamp as the `start_time`.
+    ///
+    /// Use [`FullScanRequest::builder_at`] to manually set the `start_time`, or if `feature =
+    /// "std"` is not available.
+    #[cfg(feature = "std")]
+    #[cfg_attr(docsrs, doc(cfg(feature = "std")))]
     pub fn builder() -> FullScanRequestBuilder<K> {
-        FullScanRequestBuilder {
-            inner: Self::default(),
-        }
+        let start_time = std::time::UNIX_EPOCH
+            .elapsed()
+            .expect("failed to get current timestamp")
+            .as_secs();
+        Self::builder_at(start_time)
+    }
+
+    /// When the full-scan-request was initiated.
+    pub fn start_time(&self) -> u64 {
+        self.start_time
     }
 
     /// Get the chain tip [`CheckPoint`] of this request (if any).

crates/core/src/tx_update.rs

@@ -1,4 +1,4 @@
-use crate::collections::{BTreeMap, BTreeSet, HashMap};
+use crate::collections::{BTreeMap, BTreeSet, HashSet};
 use alloc::{sync::Arc, vec::Vec};
 use bitcoin::{OutPoint, Transaction, TxOut, Txid};
 
@@ -24,16 +24,26 @@ pub struct TxUpdate<A = ()> {
     /// Full transactions. These are transactions that were determined to be relevant to the wallet
     /// given the request.
     pub txs: Vec<Arc<Transaction>>,
+
     /// Floating txouts. These are `TxOut`s that exist but the whole transaction wasn't included in
     /// `txs` since only knowing about the output is important. These are often used to help determine
     /// the fee of a wallet transaction.
     pub txouts: BTreeMap<OutPoint, TxOut>,
+
     /// Transaction anchors. Anchors tells us a position in the chain where a transaction was
     /// confirmed.
     pub anchors: BTreeSet<(A, Txid)>,
-    /// Seen at times for transactions. This records when a transaction was most recently seen in
-    /// the user's mempool for the sake of tie-breaking other conflicting transactions.
-    pub seen_ats: HashMap<Txid, u64>,
+
+    /// When transactions were seen in the mempool.
+    ///
+    /// An unconfirmed transaction can only be canonical with a `seen_at` value. It is the
+    /// responsibility of the chain-source to include the `seen_at` values for unconfirmed
+    /// (unanchored) transactions.
+    ///
+    /// [`FullScanRequest::start_time`](crate::spk_client::FullScanRequest::start_time) or
+    /// [`SyncRequest::start_time`](crate::spk_client::SyncRequest::start_time) can be used to
+    /// provide the `seen_at` value.
+    pub seen_ats: HashSet<(Txid, u64)>,
 }
 
 impl<A> Default for TxUpdate<A> {