diff --git a/crates/core/src/spk_client.rs b/crates/core/src/spk_client.rs index 64fe3dd8c..feb33ff95 100644 --- a/crates/core/src/spk_client.rs +++ b/crates/core/src/spk_client.rs @@ -2,13 +2,12 @@ use crate::{ alloc::{boxed::Box, collections::VecDeque, vec::Vec}, collections::{BTreeMap, HashMap, HashSet}, - CheckPoint, ConfirmationBlockTime, Indexed, + CheckPoint, ConfirmationBlockTime, Indexed, TxUpdate, TxUpdateCursor, }; use bitcoin::{BlockHash, OutPoint, Script, ScriptBuf, Txid}; -type InspectSync = dyn FnMut(SyncItem, SyncProgress) + Send + 'static; - -type InspectFullScan = dyn FnMut(K, u32, &Script) + Send + 'static; +type OnSyncEvent = dyn for<'a> FnMut(SyncRequestEvent<'a, I, A>) + Send + 'static; +type OnFullScanEvent = dyn for<'a> FnMut(FullScanRequestEvent<'a, K, A>) + Send + 'static; /// An item reported to the [`inspect`](SyncRequestBuilder::inspect) closure of [`SyncRequest`]. #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] @@ -86,6 +85,117 @@ impl SyncProgress { } } +/// Progress of a [`FullScanRequest`] (total unknown until `stop_gap` terminates the scan). +#[derive(Debug, Clone, Default)] +pub struct FullScanProgress { + /// Script pubkeys consumed across all keychains. + pub keychain_spks_consumed: usize, + /// Consecutive unused script pubkeys past the last-revealed index. + pub consecutive_unused: usize, + /// Transactions discovered so far. + pub txs_discovered: usize, +} + +/// Events emitted during [`SyncRequest`] execution. +#[derive(Debug)] +pub enum SyncRequestEvent<'a, I, A = ConfirmationBlockTime> { + /// A sync item is about to be fetched from the chain source. + ItemStarted(SyncItem<'a, I>, SyncProgress), + /// Incremental transaction data safe to pass to + /// [`TxGraph::apply_update`](../../bdk_chain/tx_graph/struct.TxGraph.html#method. + /// apply_update). + /// + /// On Electrum, confirmed transactions may lack anchors until [`AnchorsResolved`]. + PartialUpdate(TxUpdate), + /// Anchor-only update after Electrum resolves confirmation proofs. + AnchorsResolved(TxUpdate), +} + +/// Events emitted during [`FullScanRequest`] execution. +#[derive(Debug)] +pub enum FullScanRequestEvent<'a, K, A = ConfirmationBlockTime> { + /// A script pubkey is about to be scanned. + SpkStarted { + /// Keychain being scanned. + keychain: K, + /// Derivation index of the script pubkey. + index: u32, + /// Script pubkey. + script: &'a Script, + /// Scan progress so far. + progress: FullScanProgress, + }, + /// Incremental transaction data safe to pass to + /// [`TxGraph::apply_update`](../../bdk_chain/tx_graph/struct.TxGraph.html#method. + /// apply_update). + /// + /// On Electrum, confirmed transactions may lack anchors until [`AnchorsResolved`]. + PartialUpdate(TxUpdate), + /// Anchor-only update after Electrum resolves confirmation proofs. + AnchorsResolved(TxUpdate), +} + +/// Backend-facing sink for emitting [`SyncRequestEvent`]s. +/// +/// Constructed via [`SyncRequest::event_sink`]. Intended for official chain sources. +pub struct SyncRequestEventSink<'a, I, D = BlockHash> { + request: &'a mut SyncRequest, +} + +impl<'a, I, D> SyncRequestEventSink<'a, I, D> { + /// Emit a [`SyncRequestEvent::PartialUpdate`] if `update` is non-empty. + pub fn emit_partial_update(&mut self, update: TxUpdate) { + if !update.is_empty() { + (self.request.on_event)(SyncRequestEvent::PartialUpdate(update)); + } + } + + /// Emit a [`SyncRequestEvent::AnchorsResolved`] if `update` is non-empty. + pub fn emit_anchors_resolved(&mut self, update: TxUpdate) { + if !update.is_empty() { + (self.request.on_event)(SyncRequestEvent::AnchorsResolved(update)); + } + } +} + +/// Backend-facing sink for emitting [`FullScanRequestEvent`]s. +/// +/// Constructed via [`FullScanRequest::event_sink`]. Intended for official chain sources. +pub struct FullScanRequestEventSink<'a, K, D = BlockHash> { + request: &'a mut FullScanRequest, +} + +impl<'a, K: Ord + Clone, D> FullScanRequestEventSink<'a, K, D> { + /// Emit a [`FullScanRequestEvent::PartialUpdate`] if `update` is non-empty. + pub fn emit_partial_update(&mut self, update: TxUpdate) { + if !update.is_empty() { + (self.request.on_event)(FullScanRequestEvent::PartialUpdate(update)); + } + } + + /// Emit a [`FullScanRequestEvent::AnchorsResolved`] if `update` is non-empty. + pub fn emit_anchors_resolved(&mut self, update: TxUpdate) { + if !update.is_empty() { + (self.request.on_event)(FullScanRequestEvent::AnchorsResolved(update)); + } + } + + /// Update full-scan progress counters (called by chain sources between batches). + pub fn set_full_scan_progress(&mut self, progress: FullScanProgress) { + self.request.full_scan_progress = progress; + } + + /// Access the current full-scan progress. + pub fn full_scan_progress(&self) -> &FullScanProgress { + &self.request.full_scan_progress + } + + /// Mutable access to full-scan progress. + pub fn full_scan_progress_mut(&mut self) -> &mut FullScanProgress { + &mut self.request.full_scan_progress + } +} + /// [`Script`] with expected [`Txid`] histories. #[derive(Debug, Clone)] pub struct SpkWithExpectedTxids { @@ -202,13 +312,31 @@ impl SyncRequestBuilder { self } + /// Register a callback for sync events including progress and partial transaction updates. + /// + /// When partial updates are applied during sync, [`SyncResponse::tx_update`] at the end + /// contains only undrained remainder (empty when everything was streamed). + pub fn on_event(mut self, on_event: F) -> Self + where + F: for<'a> FnMut(SyncRequestEvent<'a, I>) + Send + 'static, + { + self.inner.emit_partial_updates = true; + self.inner.on_event = Box::new(on_event); + self + } + /// Set the closure that will inspect every sync item visited. - pub fn inspect(mut self, inspect: F) -> Self + /// + /// This is sugar over [`Self::on_event`] for progress-only callbacks. + pub fn inspect(self, mut inspect: F) -> Self where F: FnMut(SyncItem, SyncProgress) + Send + 'static, { - self.inner.inspect = Box::new(inspect); - self + self.on_event(move |event| { + if let SyncRequestEvent::ItemStarted(item, progress) = event { + inspect(item, progress); + } + }) } /// Build the [`SyncRequest`]. @@ -266,7 +394,10 @@ pub struct SyncRequest { txids_consumed: usize, outpoints: VecDeque, outpoints_consumed: usize, - inspect: Box>, + on_event: Box>, + /// Whether to emit partial [`TxUpdate`]s during sync (set by + /// [`SyncRequestBuilder::on_event`]). + pub emit_partial_updates: bool, } impl From> for SyncRequest { @@ -296,7 +427,8 @@ impl SyncRequest { txids_consumed: 0, outpoints: VecDeque::new(), outpoints_consumed: 0, - inspect: Box::new(|_, _| ()), + on_event: Box::new(|_| ()), + emit_partial_updates: false, }, } } @@ -393,9 +525,46 @@ impl SyncRequest { SyncIter::::new(self) } + /// Returns a sink for emitting sync events from chain sources. + pub fn event_sink(&mut self) -> SyncRequestEventSink<'_, I, D> { + SyncRequestEventSink { request: self } + } + + /// Emit a partial [`TxUpdate`] if [`Self::emit_partial_updates`] is enabled. + pub fn try_emit_partial_update( + &mut self, + cursor: &mut Option>, + tx_update: &mut TxUpdate, + ) { + if !self.emit_partial_updates { + return; + } + let cursor = cursor.get_or_insert_with(TxUpdateCursor::new); + let delta = tx_update.drain_since(cursor); + if !delta.is_empty() { + (self.on_event)(SyncRequestEvent::PartialUpdate(delta)); + } + } + + /// Emit an anchor-only [`TxUpdate`] if [`Self::emit_partial_updates`] is enabled. + pub fn try_emit_anchors_resolved( + &mut self, + cursor: &mut Option>, + tx_update: &mut TxUpdate, + ) { + if !self.emit_partial_updates { + return; + } + let cursor = cursor.get_or_insert_with(TxUpdateCursor::new); + let delta = tx_update.drain_since(cursor); + if !delta.is_empty() { + (self.on_event)(SyncRequestEvent::AnchorsResolved(delta)); + } + } + fn _call_inspect(&mut self, item: SyncItem) { let progress = self.progress(); - (*self.inspect)(item, progress); + (self.on_event)(SyncRequestEvent::ItemStarted(item, progress)); } } @@ -468,13 +637,37 @@ impl FullScanRequestBuilder { self } + /// Register a callback for full-scan events including progress and partial transaction updates. + /// + /// When partial updates are applied during the scan, [`FullScanResponse::tx_update`] at the end + /// contains only undrained remainder (empty when everything was streamed). + pub fn on_event(mut self, on_event: F) -> Self + where + F: for<'a> FnMut(FullScanRequestEvent<'a, K>) + Send + 'static, + { + self.inner.emit_partial_updates = true; + self.inner.on_event = Box::new(on_event); + self + } + /// Set the closure that will inspect every sync item visited. - pub fn inspect(mut self, inspect: F) -> Self + /// + /// This is sugar over [`Self::on_event`] for progress-only callbacks. + pub fn inspect(self, mut inspect: F) -> Self where F: FnMut(K, u32, &Script) + Send + 'static, { - self.inner.inspect = Box::new(inspect); - self + self.on_event(move |event| { + if let FullScanRequestEvent::SpkStarted { + keychain, + index, + script, + .. + } = event + { + inspect(keychain, index, script); + } + }) } /// Build the [`FullScanRequest`]. @@ -497,7 +690,11 @@ pub struct FullScanRequest { chain_tip: Option>, spks_by_keychain: BTreeMap> + Send>>, last_revealed: BTreeMap, - inspect: Box>, + full_scan_progress: FullScanProgress, + on_event: Box>, + /// Whether to emit partial [`TxUpdate`]s during full scan (set by + /// [`FullScanRequestBuilder::on_event`]). + pub emit_partial_updates: bool, } impl From> for FullScanRequest { @@ -522,7 +719,9 @@ impl FullScanRequest { chain_tip: None, spks_by_keychain: BTreeMap::new(), last_revealed: BTreeMap::new(), - inspect: Box::new(|_, _, _| ()), + full_scan_progress: FullScanProgress::default(), + on_event: Box::new(|_| ()), + emit_partial_updates: false, }, } } @@ -572,12 +771,48 @@ impl FullScanRequest { /// Iterate over indexed [`ScriptBuf`]s contained in this request of the given `keychain`. pub fn iter_spks(&mut self, keychain: K) -> impl Iterator> + '_ { - let spks = self.spks_by_keychain.get_mut(&keychain); - let inspect = &mut self.inspect; KeychainSpkIter { - keychain, - spks, - inspect, + keychain: keychain.clone(), + spks: self.spks_by_keychain.get_mut(&keychain), + full_scan_progress: &mut self.full_scan_progress, + on_event: &mut self.on_event, + } + } + + /// Returns a sink for emitting full-scan events from chain sources. + pub fn event_sink(&mut self) -> FullScanRequestEventSink<'_, K, D> { + FullScanRequestEventSink { request: self } + } + + /// Emit a partial [`TxUpdate`] if [`Self::emit_partial_updates`] is enabled. + pub fn try_emit_partial_update( + &mut self, + cursor: &mut Option>, + tx_update: &mut TxUpdate, + ) { + if !self.emit_partial_updates { + return; + } + let cursor = cursor.get_or_insert_with(TxUpdateCursor::new); + let delta = tx_update.drain_since(cursor); + if !delta.is_empty() { + (self.on_event)(FullScanRequestEvent::PartialUpdate(delta)); + } + } + + /// Emit an anchor-only [`TxUpdate`] if [`Self::emit_partial_updates`] is enabled. + pub fn try_emit_anchors_resolved( + &mut self, + cursor: &mut Option>, + tx_update: &mut TxUpdate, + ) { + if !self.emit_partial_updates { + return; + } + let cursor = cursor.get_or_insert_with(TxUpdateCursor::new); + let delta = tx_update.drain_since(cursor); + if !delta.is_empty() { + (self.on_event)(FullScanRequestEvent::AnchorsResolved(delta)); } } } @@ -619,7 +854,8 @@ impl FullScanResponse { struct KeychainSpkIter<'r, K> { keychain: K, spks: Option<&'r mut Box> + Send>>, - inspect: &'r mut Box>, + full_scan_progress: &'r mut FullScanProgress, + on_event: &'r mut Box>, } impl Iterator for KeychainSpkIter<'_, K> { @@ -627,7 +863,17 @@ impl Iterator for KeychainSpkIter<'_, K> { fn next(&mut self) -> Option { let (i, spk) = self.spks.as_mut()?.next()?; - (*self.inspect)(self.keychain.clone(), i, &spk); + self.full_scan_progress.keychain_spks_consumed = self + .full_scan_progress + .keychain_spks_consumed + .saturating_add(1); + let progress = self.full_scan_progress.clone(); + (self.on_event)(FullScanRequestEvent::SpkStarted { + keychain: self.keychain.clone(), + index: i, + script: spk.as_script(), + progress, + }); Some((i, spk)) } } diff --git a/crates/core/src/tx_update.rs b/crates/core/src/tx_update.rs index f92cf3e92..5ea11aae4 100644 --- a/crates/core/src/tx_update.rs +++ b/crates/core/src/tx_update.rs @@ -2,6 +2,39 @@ use crate::collections::{BTreeMap, BTreeSet, HashSet}; use alloc::{sync::Arc, vec::Vec}; use bitcoin::{OutPoint, Transaction, TxOut, Txid}; +/// Tracks fields already drained from a working [`TxUpdate`] via [`TxUpdate::drain_since`]. +/// +/// Create one cursor per sync/full_scan working buffer. Backends pass the same cursor to each +/// [`TxUpdate::drain_since`] call so partial events do not overlap. After all emissions, the +/// working buffer contains only undrained remainder (empty when fully streamed). +#[derive(Debug, Clone)] +pub struct TxUpdateCursor { + txs_len: usize, + txouts: BTreeSet, + anchors: BTreeSet<(A, Txid)>, + seen_ats: HashSet<(Txid, u64)>, + evicted_ats: HashSet<(Txid, u64)>, +} + +impl TxUpdateCursor { + /// Create a cursor at the start of a new sync/full_scan working buffer. + pub fn new() -> Self { + Self { + txs_len: 0, + txouts: BTreeSet::new(), + anchors: BTreeSet::new(), + seen_ats: HashSet::new(), + evicted_ats: HashSet::new(), + } + } +} + +impl Default for TxUpdateCursor { + fn default() -> Self { + Self::new() + } +} + /// Data object used to communicate updates about relevant transactions from some chain data source /// to the core model (usually a `bdk_chain::TxGraph`). /// @@ -82,7 +115,7 @@ impl TxUpdate { } } -impl TxUpdate { +impl TxUpdate { /// Transforms the [`TxUpdate`] to have `anchors` (`A`) of another type (`A2`). /// /// This takes in a closure with signature `FnMut(A) -> A2` which is called for each anchor to @@ -109,4 +142,72 @@ impl TxUpdate { self.seen_ats.extend(other.seen_ats); self.evicted_ats.extend(other.evicted_ats); } + + /// Destructively drain newly accumulated data since the last call. + /// + /// Returns the delta and advances `cursor`. Whatever remains in `self` at sync end becomes + /// `SyncResponse::tx_update` / `FullScanResponse::tx_update`. If everything was streamed via + /// events, the remainder is empty and applying it at the end is a no-op. + pub fn drain_since(&mut self, cursor: &mut TxUpdateCursor) -> TxUpdate { + let mut delta = TxUpdate::default(); + + if self.txs.len() > cursor.txs_len { + delta.txs = self.txs.drain(cursor.txs_len..).collect(); + cursor.txs_len = self.txs.len(); + } + + let new_outpoints: Vec = self + .txouts + .keys() + .filter(|op| !cursor.txouts.contains(op)) + .cloned() + .collect(); + for op in new_outpoints { + if let Some(txout) = self.txouts.remove(&op) { + delta.txouts.insert(op, txout); + cursor.txouts.insert(op); + } + } + + let new_anchors: Vec<(A, Txid)> = self + .anchors + .iter() + .filter(|entry| !cursor.anchors.contains(entry)) + .cloned() + .collect(); + for entry in new_anchors { + if self.anchors.remove(&entry) { + delta.anchors.insert(entry.clone()); + cursor.anchors.insert(entry); + } + } + + let new_seen: Vec<(Txid, u64)> = self + .seen_ats + .iter() + .filter(|entry| !cursor.seen_ats.contains(*entry)) + .cloned() + .collect(); + for entry in new_seen { + if self.seen_ats.remove(&entry) { + delta.seen_ats.insert(entry); + cursor.seen_ats.insert(entry); + } + } + + let new_evicted: Vec<(Txid, u64)> = self + .evicted_ats + .iter() + .filter(|entry| !cursor.evicted_ats.contains(*entry)) + .cloned() + .collect(); + for entry in new_evicted { + if self.evicted_ats.remove(&entry) { + delta.evicted_ats.insert(entry); + cursor.evicted_ats.insert(entry); + } + } + + delta + } } diff --git a/crates/core/tests/test_spk_client.rs b/crates/core/tests/test_spk_client.rs index 38766ac31..f7c12acae 100644 --- a/crates/core/tests/test_spk_client.rs +++ b/crates/core/tests/test_spk_client.rs @@ -1,4 +1,10 @@ -use bdk_core::spk_client::{FullScanResponse, SyncResponse}; +use std::sync::{Arc, Mutex}; + +use bdk_core::spk_client::{ + FullScanRequest, FullScanRequestEvent, FullScanResponse, SyncRequest, SyncRequestEvent, + SyncResponse, +}; +use bitcoin::{BlockHash, ScriptBuf}; #[test] fn test_empty() { @@ -11,3 +17,87 @@ fn test_empty() { "Default `SyncResponse` must be empty" ); } + +#[test] +fn sync_on_event_fires_item_started() { + let events = Arc::new(Mutex::new(Vec::new())); + let events2 = events.clone(); + + let request = SyncRequest::builder_at(0) + .spks([ScriptBuf::new(), ScriptBuf::new()]) + .on_event(move |event| { + if let SyncRequestEvent::ItemStarted(_item, progress) = event { + events2.lock().unwrap().push(progress.consumed()); + } + }) + .build(); + + let mut request = request; + let _ = request.next_spk_with_expected_txids(); + let _ = request.next_spk_with_expected_txids(); + + let recorded = events.lock().unwrap(); + assert_eq!(recorded.as_slice(), &[1, 2]); +} + +#[test] +fn sync_inspect_sugar_fires_item_started() { + let events = Arc::new(Mutex::new(0usize)); + let events2 = events.clone(); + + let mut request = SyncRequest::builder_at(0) + .spks([ScriptBuf::new()]) + .inspect(move |_, progress| { + *events2.lock().unwrap() = progress.consumed(); + }) + .build(); + + let _ = request.next_spk_with_expected_txids(); + assert_eq!(*events.lock().unwrap(), 1); +} + +#[test] +fn full_scan_on_event_fires_spk_started() { + let events = Arc::new(Mutex::new(Vec::new())); + let events2 = events.clone(); + + let mut request: FullScanRequest = FullScanRequest::builder_at(0) + .spks_for_keychain(0u8, [(0u32, ScriptBuf::new()), (1u32, ScriptBuf::new())]) + .on_event(move |event| { + if let FullScanRequestEvent::SpkStarted { + index, progress, .. + } = event + { + events2 + .lock() + .unwrap() + .push((index, progress.keychain_spks_consumed)); + } + }) + .build(); + + let _ = request.iter_spks(0).collect::>(); + + let recorded = events.lock().unwrap(); + assert_eq!(recorded.len(), 2); + assert_eq!(recorded[0], (0, 1)); + assert_eq!(recorded[1], (1, 2)); +} + +#[test] +fn full_scan_inspect_sugar_fires_spk_started() { + let events = Arc::new(Mutex::new(Vec::new())); + let events2 = events.clone(); + + let mut request: FullScanRequest = FullScanRequest::builder_at(0) + .spks_for_keychain(0u8, [(5u32, ScriptBuf::new())]) + .inspect(move |keychain, index, _| { + events2.lock().unwrap().push((keychain, index)); + }) + .build(); + + let _ = request.iter_spks(0).collect::>(); + + let recorded = events.lock().unwrap(); + assert_eq!(recorded.as_slice(), &[(0u8, 5u32)]); +} diff --git a/crates/core/tests/test_tx_update.rs b/crates/core/tests/test_tx_update.rs index d5d2dbc3b..467939e09 100644 --- a/crates/core/tests/test_tx_update.rs +++ b/crates/core/tests/test_tx_update.rs @@ -1,4 +1,16 @@ -use bdk_core::TxUpdate; +use std::sync::Arc; + +use bdk_core::{TxUpdate, TxUpdateCursor}; +use bitcoin::{absolute::LockTime, transaction::Version, OutPoint, Transaction, TxOut}; + +fn dummy_tx() -> Arc { + Arc::new(Transaction { + version: Version::ONE, + lock_time: LockTime::ZERO, + input: vec![], + output: vec![], + }) +} #[test] fn test_empty() { @@ -7,3 +19,73 @@ fn test_empty() { "Default `TxUpdate` must be empty" ); } + +#[test] +fn drain_since_returns_only_new_data() { + let tx1 = dummy_tx(); + let tx2 = dummy_tx(); + let id1 = tx1.compute_txid(); + let id2 = tx2.compute_txid(); + + let mut update = TxUpdate::default(); + let mut cursor = TxUpdateCursor::new(); + + update.txs.push(tx1.clone()); + update.seen_ats.insert((id1, 1)); + + let d1 = update.drain_since(&mut cursor); + assert_eq!(d1.txs.len(), 1); + assert!(d1.seen_ats.contains(&(id1, 1))); + assert!(update.is_empty()); + + update.txs.push(tx2.clone()); + update.anchors.insert(((), id2)); + + let d2 = update.drain_since(&mut cursor); + assert_eq!(d2.txs.len(), 1); + assert!(d2.anchors.contains(&((), id2))); + assert!(update.is_empty()); +} + +#[test] +fn drain_since_partitions_match_extend_roundtrip() { + let tx = dummy_tx(); + let id = tx.compute_txid(); + let op = OutPoint { txid: id, vout: 0 }; + + let mut full = TxUpdate::<()>::default(); + full.txs.push(tx); + full.txouts.insert(op, TxOut::NULL); + full.seen_ats.insert((id, 42)); + full.evicted_ats.insert((id, 99)); + + let mut working = full.clone(); + let mut cursor = TxUpdateCursor::new(); + let d1 = working.drain_since(&mut cursor); + let d2 = working.drain_since(&mut cursor); + + assert!(working.is_empty()); + assert!(d2.is_empty()); + + let mut reconstructed = TxUpdate::default(); + reconstructed.extend(d1); + assert_eq!(reconstructed.txs.len(), full.txs.len()); + assert_eq!(reconstructed.txouts, full.txouts); + assert_eq!(reconstructed.seen_ats, full.seen_ats); + assert_eq!(reconstructed.evicted_ats, full.evicted_ats); +} + +#[test] +fn final_response_empty_after_full_drain() { + let tx = dummy_tx(); + let mut update = TxUpdate::<()>::default(); + update.txs.push(tx); + + let mut cursor = TxUpdateCursor::new(); + let _ = update.drain_since(&mut cursor); + + assert!( + update.is_empty(), + "final tx_update must be empty after full drain" + ); +} diff --git a/crates/electrum/src/bdk_electrum_client.rs b/crates/electrum/src/bdk_electrum_client.rs index a7c943150..af4dd773a 100644 --- a/crates/electrum/src/bdk_electrum_client.rs +++ b/crates/electrum/src/bdk_electrum_client.rs @@ -8,12 +8,16 @@ use bdk_core::{ spk_client::{ FullScanRequest, FullScanResponse, SpkWithExpectedTxids, SyncRequest, SyncResponse, }, - BlockId, CheckPoint, ConfirmationBlockTime, TxUpdate, + BlockId, CheckPoint, ConfirmationBlockTime, TxUpdate, TxUpdateCursor, }; use electrum_client::{ElectrumApi, Error, HeaderNotification}; use std::convert::TryInto; use std::sync::{Arc, Mutex}; +use crate::sync_emit::{ + maybe_drain_and_emit_anchors, maybe_drain_and_emit_partial, PartialUpdateSink, +}; + /// We include a chain suffix of a certain length for the purpose of robustness. const CHAIN_SUFFIX_LENGTH: u32 = 8; @@ -139,19 +143,36 @@ impl BdkElectrumClient { let mut tx_update = TxUpdate::::default(); let mut last_active_indices = BTreeMap::::default(); let mut pending_anchors = Vec::new(); - for keychain in request.keychains() { + + let keychains = request.keychains(); + let mut spk_batches = Vec::with_capacity(keychains.len()); + for keychain in keychains { let last_revealed = request.last_revealed(&keychain); - let spks = request + let spks: Vec<_> = request .iter_spks(keychain.clone()) - .map(|(spk_i, spk)| (spk_i, SpkWithExpectedTxids::from(spk))); + .map(|(spk_i, spk)| (spk_i, SpkWithExpectedTxids::from(spk))) + .collect(); + spk_batches.push((keychain, last_revealed, spks)); + } + + let emit_partial = request.emit_partial_updates; + let mut cursor = emit_partial.then(TxUpdateCursor::new); + let mut full_scan_sink = emit_partial.then(|| request.event_sink()); + let mut partial_sink: Option<&mut dyn PartialUpdateSink> = full_scan_sink + .as_mut() + .map(|s| s as &mut dyn PartialUpdateSink); + + for (keychain, last_revealed, spks) in spk_batches { if let Some(last_active_index) = self.populate_with_spks( start_time, &mut tx_update, - spks, + spks.into_iter(), stop_gap, last_revealed, batch_size, &mut pending_anchors, + &mut partial_sink, + &mut cursor, )? { last_active_indices.insert(keychain, last_active_index); } @@ -160,6 +181,7 @@ impl BdkElectrumClient { // Fetch previous `TxOut`s for fee calculation if flag is enabled. if fetch_prev_txouts { self.fetch_prev_txout(&mut tx_update)?; + maybe_drain_and_emit_partial(&mut partial_sink, &mut cursor, &mut tx_update); } if !pending_anchors.is_empty() { @@ -167,6 +189,7 @@ impl BdkElectrumClient { for (txid, anchor) in anchors { tx_update.anchors.insert((anchor, txid)); } + maybe_drain_and_emit_anchors(&mut partial_sink, &mut cursor, &mut tx_update); } let chain_update = match tip_and_latest_blocks { @@ -223,34 +246,55 @@ impl BdkElectrumClient { let mut tx_update = TxUpdate::::default(); let mut pending_anchors = Vec::new(); + + let spks: Vec<_> = request + .iter_spks_with_expected_txids() + .enumerate() + .map(|(i, spk)| (i as u32, spk)) + .collect(); + let txids: Vec<_> = request.iter_txids().collect(); + let outpoints: Vec<_> = request.iter_outpoints().collect(); + + let emit_partial = request.emit_partial_updates; + let mut cursor = emit_partial.then(TxUpdateCursor::new); + let mut sync_sink = emit_partial.then(|| request.event_sink()); + let mut partial_sink: Option<&mut dyn PartialUpdateSink> = + sync_sink.as_mut().map(|s| s as &mut dyn PartialUpdateSink); + self.populate_with_spks( start_time, &mut tx_update, - request - .iter_spks_with_expected_txids() - .enumerate() - .map(|(i, spk)| (i as u32, spk)), + spks.into_iter(), usize::MAX, None, batch_size, &mut pending_anchors, + &mut partial_sink, + &mut cursor, )?; + self.populate_with_txids( start_time, &mut tx_update, - request.iter_txids(), + txids, &mut pending_anchors, + &mut partial_sink, + &mut cursor, )?; + self.populate_with_outpoints( start_time, &mut tx_update, - request.iter_outpoints(), + outpoints, &mut pending_anchors, + &mut partial_sink, + &mut cursor, )?; // Fetch previous `TxOut`s for fee calculation if flag is enabled. if fetch_prev_txouts { self.fetch_prev_txout(&mut tx_update)?; + maybe_drain_and_emit_partial(&mut partial_sink, &mut cursor, &mut tx_update); } if !pending_anchors.is_empty() { @@ -258,6 +302,7 @@ impl BdkElectrumClient { for (txid, anchor) in anchors { tx_update.anchors.insert((anchor, txid)); } + maybe_drain_and_emit_anchors(&mut partial_sink, &mut cursor, &mut tx_update); } let chain_update = match tip_and_latest_blocks { @@ -290,6 +335,8 @@ impl BdkElectrumClient { last_revealed: Option, batch_size: usize, pending_anchors: &mut Vec<(Txid, usize)>, + partial_sink: &mut Option<&mut dyn PartialUpdateSink>, + cursor: &mut Option>, ) -> Result, Error> { let mut unused_spk_count = 0_usize; let mut last_active_index = Option::::None; @@ -315,6 +362,7 @@ impl BdkElectrumClient { } else if beyond_revealed { unused_spk_count = unused_spk_count.saturating_add(1); if unused_spk_count >= stop_gap { + maybe_drain_and_emit_partial(partial_sink, cursor, tx_update); return Ok(last_active_index); } } @@ -343,6 +391,8 @@ impl BdkElectrumClient { } } } + + maybe_drain_and_emit_partial(partial_sink, cursor, tx_update); } } @@ -356,6 +406,8 @@ impl BdkElectrumClient { tx_update: &mut TxUpdate, outpoints: impl IntoIterator, pending_anchors: &mut Vec<(Txid, usize)>, + partial_sink: &mut Option<&mut dyn PartialUpdateSink>, + cursor: &mut Option>, ) -> Result<(), Error> { // Collect valid outpoints with their corresponding `spk` and `tx`. let mut ops_spks_txs = Vec::new(); @@ -431,6 +483,7 @@ impl BdkElectrumClient { } } + maybe_drain_and_emit_partial(partial_sink, cursor, tx_update); Ok(()) } @@ -441,6 +494,8 @@ impl BdkElectrumClient { tx_update: &mut TxUpdate, txids: impl IntoIterator, pending_anchors: &mut Vec<(Txid, usize)>, + partial_sink: &mut Option<&mut dyn PartialUpdateSink>, + cursor: &mut Option>, ) -> Result<(), Error> { let mut txs = Vec::<(Txid, Arc)>::new(); let mut scripts = Vec::new(); @@ -512,6 +567,7 @@ impl BdkElectrumClient { tx_update.txs.push(tx.1); } + maybe_drain_and_emit_partial(partial_sink, cursor, tx_update); Ok(()) } diff --git a/crates/electrum/src/lib.rs b/crates/electrum/src/lib.rs index 7af8b847d..c50ec2a40 100644 --- a/crates/electrum/src/lib.rs +++ b/crates/electrum/src/lib.rs @@ -17,6 +17,7 @@ #![warn(missing_docs)] mod bdk_electrum_client; +mod sync_emit; pub use bdk_electrum_client::*; pub use bdk_core; diff --git a/crates/electrum/src/sync_emit.rs b/crates/electrum/src/sync_emit.rs new file mode 100644 index 000000000..f75a10e6b --- /dev/null +++ b/crates/electrum/src/sync_emit.rs @@ -0,0 +1,68 @@ +use bdk_core::{ + spk_client::{FullScanRequestEventSink, SyncRequestEventSink}, + ConfirmationBlockTime, TxUpdate, TxUpdateCursor, +}; + +/// Sink for emitting incremental [`TxUpdate`]s during sync or full scan. +pub(crate) trait PartialUpdateSink { + fn emit_partial(&mut self, update: TxUpdate); + fn emit_anchors(&mut self, update: TxUpdate); +} + +impl PartialUpdateSink for SyncRequestEventSink<'_, I, D> { + fn emit_partial(&mut self, update: TxUpdate) { + self.emit_partial_update(update); + } + + fn emit_anchors(&mut self, update: TxUpdate) { + self.emit_anchors_resolved(update); + } +} + +impl PartialUpdateSink for FullScanRequestEventSink<'_, K, D> { + fn emit_partial(&mut self, update: TxUpdate) { + self.emit_partial_update(update); + } + + fn emit_anchors(&mut self, update: TxUpdate) { + self.emit_anchors_resolved(update); + } +} + +pub(crate) fn drain_and_emit_partial( + sink: &mut dyn PartialUpdateSink, + cursor: &mut TxUpdateCursor, + tx_update: &mut TxUpdate, +) { + let delta = tx_update.drain_since(cursor); + sink.emit_partial(delta); +} + +pub(crate) fn drain_and_emit_anchors( + sink: &mut dyn PartialUpdateSink, + cursor: &mut TxUpdateCursor, + tx_update: &mut TxUpdate, +) { + let delta = tx_update.drain_since(cursor); + sink.emit_anchors(delta); +} + +pub(crate) fn maybe_drain_and_emit_partial( + sink: &mut Option<&mut dyn PartialUpdateSink>, + cursor: &mut Option>, + tx_update: &mut TxUpdate, +) { + if let (Some(sink), Some(cursor)) = (sink.as_deref_mut(), cursor.as_mut()) { + drain_and_emit_partial(sink, cursor, tx_update); + } +} + +pub(crate) fn maybe_drain_and_emit_anchors( + sink: &mut Option<&mut dyn PartialUpdateSink>, + cursor: &mut Option>, + tx_update: &mut TxUpdate, +) { + if let (Some(sink), Some(cursor)) = (sink.as_deref_mut(), cursor.as_mut()) { + drain_and_emit_anchors(sink, cursor, tx_update); + } +} diff --git a/crates/electrum/tests/test_electrum.rs b/crates/electrum/tests/test_electrum.rs index c569b065d..3af5f4fc2 100644 --- a/crates/electrum/tests/test_electrum.rs +++ b/crates/electrum/tests/test_electrum.rs @@ -1,7 +1,7 @@ use bdk_chain::{ bitcoin::{hashes::Hash, Address, Amount, ScriptBuf, WScriptHash}, local_chain::LocalChain, - spk_client::{FullScanRequest, SyncRequest, SyncResponse}, + spk_client::{FullScanRequest, SyncRequest, SyncRequestEvent, SyncResponse}, spk_txout::SpkTxOutIndex, Balance, ConfirmationBlockTime, IndexedTxGraph, Indexer, Merge, TxGraph, }; @@ -9,6 +9,7 @@ use bdk_core::bitcoin::{ key::{Secp256k1, UntweakedPublicKey}, Denomination, }; +use bdk_core::TxUpdate; use bdk_electrum::BdkElectrumClient; use bdk_testenv::{ anyhow, @@ -19,6 +20,7 @@ use core::time::Duration; use electrum_client::ElectrumApi; use std::collections::{BTreeSet, HashSet}; use std::str::FromStr; +use std::sync::{Arc, Mutex}; // Batch size for `sync_with_electrum`. const BATCH_SIZE: usize = 5; @@ -393,6 +395,153 @@ pub fn test_update_tx_graph_without_keychain() -> anyhow::Result<()> { Ok(()) } +/// Sync with `.on_event()` streams [`PartialUpdate`](SyncRequestEvent::PartialUpdate) events +/// before returning, and the final [`SyncResponse::tx_update`] is empty when all data was drained. +#[test] +pub fn sync_streams_partial_updates() -> anyhow::Result<()> { + let env = TestEnv::new()?; + let electrum_client = electrum_client::Client::new(env.electrsd.electrum_url.as_str())?; + let client = BdkElectrumClient::new(electrum_client); + + let receive_address0 = + Address::from_str("bcrt1qc6fweuf4xjvz4x3gx3t9e0fh4hvqyu2qw4wvxm")?.assume_checked(); + let receive_address1 = + Address::from_str("bcrt1qfjg5lv3dvc9az8patec8fjddrs4aqtauadnagr")?.assume_checked(); + + let misc_spks = [ + receive_address0.script_pubkey(), + receive_address1.script_pubkey(), + ]; + + let _block_hashes = env.mine_blocks(101, None)?; + let txid1 = env + .bitcoind + .client + .send_to_address(&receive_address1, Amount::from_sat(10000))? + .txid()?; + let txid2 = env + .bitcoind + .client + .send_to_address(&receive_address0, Amount::from_sat(20000))? + .txid()?; + env.mine_blocks(1, None)?; + env.wait_until_electrum_sees_block(Duration::from_secs(6))?; + + let cp_tip = env.make_checkpoint_tip(); + + let partial_count = Arc::new(Mutex::new(0usize)); + let streamed_txs = Arc::new(Mutex::new(BTreeSet::new())); + let streamed_update = Arc::new(Mutex::new(TxUpdate::default())); + let partial_count2 = partial_count.clone(); + let streamed_txs2 = streamed_txs.clone(); + let streamed_update2 = streamed_update.clone(); + + let sync_update = { + let request = SyncRequest::builder() + .chain_tip(cp_tip) + .spks(misc_spks) + .on_event(move |event| { + if let SyncRequestEvent::PartialUpdate(delta) = event { + *partial_count2.lock().unwrap() += 1; + for tx in &delta.txs { + streamed_txs2.lock().unwrap().insert(tx.compute_txid()); + } + streamed_update2.lock().unwrap().extend(delta); + } + }); + client.sync(request, 1, true)? + }; + + assert!( + *partial_count.lock().unwrap() > 0, + "must emit partial updates during sync" + ); + assert!( + sync_update.tx_update.is_empty(), + "final tx_update must be empty when events drained all data" + ); + assert_eq!( + *streamed_txs.lock().unwrap(), + [txid1, txid2].into(), + "streamed txs must include all discovered transactions" + ); + + let mut graph = TxGraph::::default(); + let _ = graph.apply_update(streamed_update.lock().unwrap().clone()); + assert_eq!( + graph + .full_txs() + .map(|tx| tx.compute_txid()) + .collect::>(), + [txid1, txid2].into(), + "reconstructed update from events must match full sync" + ); + + Ok(()) +} + +/// Electrum defers anchor resolution: confirmed txs in +/// [`PartialUpdate`](SyncRequestEvent::PartialUpdate) lack anchors until +/// [`AnchorsResolved`](SyncRequestEvent::AnchorsResolved). +#[test] +pub fn sync_defers_anchors_until_resolved() -> anyhow::Result<()> { + let env = TestEnv::new()?; + let electrum_client = electrum_client::Client::new(env.electrsd.electrum_url.as_str())?; + let client = BdkElectrumClient::new(electrum_client); + + let receive_address = + Address::from_str("bcrt1qc6fweuf4xjvz4x3gx3t9e0fh4hvqyu2qw4wvxm")?.assume_checked(); + let spk = receive_address.script_pubkey(); + + let _block_hashes = env.mine_blocks(101, None)?; + let txid = env + .bitcoind + .client + .send_to_address(&receive_address, Amount::from_sat(10000))? + .txid()?; + env.mine_blocks(1, None)?; + env.wait_until_electrum_sees_block(Duration::from_secs(6))?; + + let partial_anchor_txids = Arc::new(Mutex::new(BTreeSet::new())); + let resolved_anchor_txids = Arc::new(Mutex::new(BTreeSet::new())); + let partial_anchor_txids2 = partial_anchor_txids.clone(); + let resolved_anchor_txids2 = resolved_anchor_txids.clone(); + + let sync_update = { + let request = SyncRequest::builder() + .spks([spk]) + .on_event(move |event| match event { + SyncRequestEvent::PartialUpdate(delta) => { + for (_, txid) in &delta.anchors { + partial_anchor_txids2.lock().unwrap().insert(*txid); + } + } + SyncRequestEvent::AnchorsResolved(delta) => { + for (_, txid) in &delta.anchors { + resolved_anchor_txids2.lock().unwrap().insert(*txid); + } + } + _ => {} + }); + client.sync(request, 1, true)? + }; + + assert!( + partial_anchor_txids.lock().unwrap().is_empty(), + "confirmed txs in partial updates must not include anchors" + ); + assert!( + resolved_anchor_txids.lock().unwrap().contains(&txid), + "AnchorsResolved must add anchor for confirmed tx" + ); + assert!( + sync_update.tx_update.is_empty(), + "final tx_update must be empty when events drained all data" + ); + + Ok(()) +} + /// Test the bounds of the address scan depending on the `stop_gap`. #[test] pub fn test_update_tx_graph_stop_gap() -> anyhow::Result<()> { diff --git a/crates/esplora/src/async_ext.rs b/crates/esplora/src/async_ext.rs index 2b3828952..b59678682 100644 --- a/crates/esplora/src/async_ext.rs +++ b/crates/esplora/src/async_ext.rs @@ -5,11 +5,12 @@ use bdk_core::spk_client::{ }; use bdk_core::{ bitcoin::{BlockHash, OutPoint, Txid}, - BlockId, CheckPoint, ConfirmationBlockTime, Indexed, TxUpdate, + BlockId, CheckPoint, ConfirmationBlockTime, Indexed, TxUpdate, TxUpdateCursor, }; use esplora_client::Sleeper; use futures::{stream::FuturesOrdered, TryStreamExt}; +use crate::fetch_emit::EmitPartial; use crate::{insert_anchor_or_seen_at_from_status, insert_prevouts}; /// [`esplora_client::Error`] @@ -66,7 +67,6 @@ where ) -> Result, Error> { let mut request: FullScanRequest = request.into(); let start_time = request.start_time(); - let keychains = request.keychains(); let chain_tip = request.chain_tip(); let latest_blocks = if chain_tip.is_some() { @@ -78,23 +78,36 @@ where let mut tx_update = TxUpdate::::default(); let mut inserted_txs = HashSet::::new(); let mut last_active_indices = BTreeMap::::new(); + + let keychains = request.keychains(); + let mut spk_batches = Vec::with_capacity(keychains.len()); for keychain in keychains { let last_revealed = request.last_revealed(&keychain); - let keychain_spks = request + let spks: Vec<_> = request .iter_spks(keychain.clone()) - .map(|(spk_i, spk)| (spk_i, spk.into())); - let (update, last_active_index) = fetch_txs_with_keychain_spks( + .map(|(spk_i, spk)| (spk_i, spk.into())) + .collect(); + spk_batches.push((keychain, last_revealed, spks)); + } + + let emit_partial = request.emit_partial_updates; + let mut cursor = emit_partial.then(TxUpdateCursor::new); + + for (keychain, last_revealed, spks) in spk_batches { + if let Some(last_active_index) = fetch_txs_with_keychain_spks( self, start_time, &mut inserted_txs, - keychain_spks, + &mut tx_update, + spks.into_iter(), stop_gap, last_revealed, parallel_requests, + &mut request, + &mut cursor, ) - .await?; - tx_update.extend(update); - if let Some(last_active_index) = last_active_index { + .await? + { last_active_indices.insert(keychain, last_active_index); } } @@ -130,36 +143,47 @@ where let mut tx_update = TxUpdate::::default(); let mut inserted_txs = HashSet::::new(); - tx_update.extend( - fetch_txs_with_spks( - self, - start_time, - &mut inserted_txs, - request.iter_spks_with_expected_txids(), - parallel_requests, - ) - .await?, - ); - tx_update.extend( - fetch_txs_with_txids( - self, - start_time, - &mut inserted_txs, - request.iter_txids(), - parallel_requests, - ) - .await?, - ); - tx_update.extend( - fetch_txs_with_outpoints( - self, - start_time, - &mut inserted_txs, - request.iter_outpoints(), - parallel_requests, - ) - .await?, - ); + + let spks: Vec<_> = request.iter_spks_with_expected_txids().collect(); + let txids: Vec<_> = request.iter_txids().collect(); + let outpoints: Vec<_> = request.iter_outpoints().collect(); + + let emit_partial = request.emit_partial_updates; + let mut cursor = emit_partial.then(TxUpdateCursor::new); + + fetch_txs_with_spks( + self, + start_time, + &mut inserted_txs, + &mut tx_update, + spks, + parallel_requests, + &mut request, + &mut cursor, + ) + .await?; + fetch_txs_with_txids( + self, + start_time, + &mut inserted_txs, + &mut tx_update, + txids, + parallel_requests, + &mut request, + &mut cursor, + ) + .await?; + fetch_txs_with_outpoints( + self, + start_time, + &mut inserted_txs, + &mut tx_update, + outpoints, + parallel_requests, + &mut request, + &mut cursor, + ) + .await?; let chain_update = match (chain_tip, latest_blocks) { (Some(chain_tip), Some(latest_blocks)) => { @@ -301,26 +325,28 @@ async fn chain_update( /// script pubkey that contains a non-empty transaction history. /// /// Refer to [crate-level docs](crate) for more. -async fn fetch_txs_with_keychain_spks( +#[allow(clippy::too_many_arguments)] +async fn fetch_txs_with_keychain_spks( client: &esplora_client::AsyncClient, start_time: u64, inserted_txs: &mut HashSet, + tx_update: &mut TxUpdate, mut keychain_spks: I, stop_gap: usize, last_revealed: Option, parallel_requests: usize, -) -> Result<(TxUpdate, Option), Error> + emitter: &mut E, + cursor: &mut Option>, +) -> Result, Error> where I: Iterator> + Send, + E: EmitPartial, S: Sleeper + Clone + Send + Sync, { type TxsOfSpkIndex = (u32, Vec, HashSet); - let mut update = TxUpdate::::default(); let mut last_active_index = Option::::None; - // Use consecutive_unused so unused count drives stop gap. let mut consecutive_unused = 0usize; - // Treat stop_gap = 0 as 1 while preserving original semantics for other values. let gap_limit = stop_gap.max(1); loop { @@ -367,22 +393,29 @@ where for tx in txs { if inserted_txs.insert(tx.txid) { - update.txs.push(tx.to_tx().into()); + tx_update.txs.push(tx.to_tx().into()); } - insert_anchor_or_seen_at_from_status(&mut update, start_time, tx.txid, tx.status); - insert_prevouts(&mut update, tx.vin); + insert_anchor_or_seen_at_from_status( + &mut *tx_update, + start_time, + tx.txid, + tx.status, + ); + insert_prevouts(&mut *tx_update, tx.vin); } - update + tx_update .evicted_ats .extend(evicted.into_iter().map(|txid| (txid, start_time))); } + emitter.emit_partial(cursor, tx_update); + if consecutive_unused >= gap_limit { break; } } - Ok((update, last_active_index)) + Ok(last_active_index) } /// Fetch transactions and associated [`ConfirmationBlockTime`]s by scanning `spks` @@ -393,29 +426,37 @@ where /// HTTP requests to make in parallel. /// /// Refer to [crate-level docs](crate) for more. -async fn fetch_txs_with_spks( +#[allow(clippy::too_many_arguments)] +async fn fetch_txs_with_spks( client: &esplora_client::AsyncClient, start_time: u64, inserted_txs: &mut HashSet, + tx_update: &mut TxUpdate, spks: I, parallel_requests: usize, -) -> Result, Error> + emitter: &mut E, + cursor: &mut Option>, +) -> Result<(), Error> where I: IntoIterator + Send, I::IntoIter: Send, + E: EmitPartial, S: Sleeper + Clone + Send + Sync, { fetch_txs_with_keychain_spks( client, start_time, inserted_txs, + tx_update, spks.into_iter().enumerate().map(|(i, spk)| (i as u32, spk)), usize::MAX, None, parallel_requests, + emitter, + cursor, ) .await - .map(|(update, _)| update) + .map(|_| ()) } /// Fetch transactions and associated [`ConfirmationBlockTime`]s by scanning `txids` @@ -424,20 +465,23 @@ where /// `parallel_requests` specifies the maximum number of HTTP requests to make in parallel. /// /// Refer to [crate-level docs](crate) for more. -async fn fetch_txs_with_txids( +#[allow(clippy::too_many_arguments)] +async fn fetch_txs_with_txids( client: &esplora_client::AsyncClient, start_time: u64, inserted_txs: &mut HashSet, + tx_update: &mut TxUpdate, txids: I, parallel_requests: usize, -) -> Result, Error> + emitter: &mut E, + cursor: &mut Option>, +) -> Result<(), Error> where I: IntoIterator + Send, I::IntoIter: Send, + E: EmitPartial, S: Sleeper + Clone + Send + Sync, { - let mut update = TxUpdate::::default(); - // Only fetch for non-inserted txs. let mut txids = txids .into_iter() .filter(|txid| !inserted_txs.contains(txid)) @@ -460,14 +504,21 @@ where for (txid, tx_info) in handles.try_collect::>().await? { if let Some(tx_info) = tx_info { if inserted_txs.insert(txid) { - update.txs.push(tx_info.to_tx().into()); + tx_update.txs.push(tx_info.to_tx().into()); } - insert_anchor_or_seen_at_from_status(&mut update, start_time, txid, tx_info.status); - insert_prevouts(&mut update, tx_info.vin); + insert_anchor_or_seen_at_from_status( + &mut *tx_update, + start_time, + txid, + tx_info.status, + ); + insert_prevouts(&mut *tx_update, tx_info.vin); } } } - Ok(update) + + emitter.emit_partial(cursor, tx_update); + Ok(()) } /// Fetch transactions and [`ConfirmationBlockTime`]s that contain and spend the provided @@ -476,35 +527,37 @@ where /// `parallel_requests` specifies the maximum number of HTTP requests to make in parallel. /// /// Refer to [crate-level docs](crate) for more. -async fn fetch_txs_with_outpoints( +#[allow(clippy::too_many_arguments)] +async fn fetch_txs_with_outpoints( client: &esplora_client::AsyncClient, start_time: u64, inserted_txs: &mut HashSet, + tx_update: &mut TxUpdate, outpoints: I, parallel_requests: usize, -) -> Result, Error> + emitter: &mut E, + cursor: &mut Option>, +) -> Result<(), Error> where I: IntoIterator + Send, I::IntoIter: Send, + E: EmitPartial, S: Sleeper + Clone + Send + Sync, { let outpoints = outpoints.into_iter().collect::>(); - let mut update = TxUpdate::::default(); - // make sure txs exists in graph and tx statuses are updated - // TODO: We should maintain a tx cache (like we do with Electrum). - update.extend( - fetch_txs_with_txids( - client, - start_time, - inserted_txs, - outpoints.iter().copied().map(|op| op.txid), - parallel_requests, - ) - .await?, - ); + fetch_txs_with_txids( + client, + start_time, + inserted_txs, + tx_update, + outpoints.iter().copied().map(|op| op.txid), + parallel_requests, + emitter, + cursor, + ) + .await?; - // get outpoint spend-statuses let mut outpoints = outpoints.into_iter(); let mut missing_txs = HashSet::::with_capacity(outpoints.len()); loop { @@ -531,7 +584,7 @@ where } if let Some(spend_status) = op_status.status { insert_anchor_or_seen_at_from_status( - &mut update, + &mut *tx_update, start_time, spend_txid, spend_status, @@ -540,17 +593,17 @@ where } } - update.extend( - fetch_txs_with_txids( - client, - start_time, - inserted_txs, - missing_txs, - parallel_requests, - ) - .await?, - ); - Ok(update) + fetch_txs_with_txids( + client, + start_time, + inserted_txs, + tx_update, + missing_txs, + parallel_requests, + emitter, + cursor, + ) + .await } #[cfg(test)] diff --git a/crates/esplora/src/blocking_ext.rs b/crates/esplora/src/blocking_ext.rs index 225b574ea..5a1b4933f 100644 --- a/crates/esplora/src/blocking_ext.rs +++ b/crates/esplora/src/blocking_ext.rs @@ -4,11 +4,12 @@ use bdk_core::spk_client::{ }; use bdk_core::{ bitcoin::{BlockHash, OutPoint, Txid}, - BlockId, CheckPoint, ConfirmationBlockTime, Indexed, TxUpdate, + BlockId, CheckPoint, ConfirmationBlockTime, Indexed, TxUpdate, TxUpdateCursor, }; use esplora_client::{OutputStatus, Tx}; use std::thread::JoinHandle; +use crate::fetch_emit::EmitPartial; use crate::{insert_anchor_or_seen_at_from_status, insert_prevouts}; /// [`esplora_client::Error`] @@ -68,22 +69,34 @@ impl EsploraExt for esplora_client::BlockingClient { let mut tx_update = TxUpdate::default(); let mut inserted_txs = HashSet::::new(); let mut last_active_indices = BTreeMap::::new(); - for keychain in request.keychains() { + + let keychains = request.keychains(); + let mut spk_batches = Vec::with_capacity(keychains.len()); + for keychain in keychains { let last_revealed = request.last_revealed(&keychain); - let keychain_spks = request + let spks: Vec<_> = request .iter_spks(keychain.clone()) - .map(|(spk_i, spk)| (spk_i, spk.into())); - let (update, last_active_index) = fetch_txs_with_keychain_spks( + .map(|(spk_i, spk)| (spk_i, spk.into())) + .collect(); + spk_batches.push((keychain, last_revealed, spks)); + } + + let emit_partial = request.emit_partial_updates; + let mut cursor = emit_partial.then(TxUpdateCursor::new); + + for (keychain, last_revealed, spks) in spk_batches { + if let Some(last_active_index) = fetch_txs_with_keychain_spks( self, start_time, &mut inserted_txs, - keychain_spks, + &mut tx_update, + spks.into_iter(), stop_gap, last_revealed, parallel_requests, - )?; - tx_update.extend(update); - if let Some(last_active_index) = last_active_index { + &mut request, + &mut cursor, + )? { last_active_indices.insert(keychain, last_active_index); } } @@ -122,27 +135,44 @@ impl EsploraExt for esplora_client::BlockingClient { let mut tx_update = TxUpdate::::default(); let mut inserted_txs = HashSet::::new(); - tx_update.extend(fetch_txs_with_spks( + + let spks: Vec<_> = request.iter_spks_with_expected_txids().collect(); + let txids: Vec<_> = request.iter_txids().collect(); + let outpoints: Vec<_> = request.iter_outpoints().collect(); + + let emit_partial = request.emit_partial_updates; + let mut cursor = emit_partial.then(TxUpdateCursor::new); + + fetch_txs_with_spks( self, start_time, &mut inserted_txs, - request.iter_spks_with_expected_txids(), + &mut tx_update, + spks, parallel_requests, - )?); - tx_update.extend(fetch_txs_with_txids( + &mut request, + &mut cursor, + )?; + fetch_txs_with_txids( self, start_time, &mut inserted_txs, - request.iter_txids(), + &mut tx_update, + txids, parallel_requests, - )?); - tx_update.extend(fetch_txs_with_outpoints( + &mut request, + &mut cursor, + )?; + fetch_txs_with_outpoints( self, start_time, &mut inserted_txs, - request.iter_outpoints(), + &mut tx_update, + outpoints, parallel_requests, - )?); + &mut request, + &mut cursor, + )?; let chain_update = match (chain_tip, latest_blocks) { (Some(chain_tip), Some(latest_blocks)) => Some(chain_update( @@ -273,22 +303,26 @@ fn chain_update( Ok(tip) } -fn fetch_txs_with_keychain_spks>>( +#[allow(clippy::too_many_arguments)] +fn fetch_txs_with_keychain_spks( client: &esplora_client::BlockingClient, start_time: u64, inserted_txs: &mut HashSet, + tx_update: &mut TxUpdate, mut keychain_spks: I, stop_gap: usize, last_revealed: Option, parallel_requests: usize, -) -> Result<(TxUpdate, Option), Error> { + emitter: &mut E, + cursor: &mut Option>, +) -> Result, Error> +where + I: Iterator>, +{ type TxsOfSpkIndex = (u32, Vec, HashSet); - let mut update = TxUpdate::::default(); let mut last_active_index = Option::::None; - // Use consecutive_unused so unused count drives stop gap. let mut consecutive_unused = 0usize; - // Treat stop_gap = 0 as 1 while preserving original semantics for other values. let gap_limit = stop_gap.max(1); loop { @@ -336,22 +370,29 @@ fn fetch_txs_with_keychain_spks for tx in txs { if inserted_txs.insert(tx.txid) { - update.txs.push(tx.to_tx().into()); + tx_update.txs.push(tx.to_tx().into()); } - insert_anchor_or_seen_at_from_status(&mut update, start_time, tx.txid, tx.status); - insert_prevouts(&mut update, tx.vin); + insert_anchor_or_seen_at_from_status( + &mut *tx_update, + start_time, + tx.txid, + tx.status, + ); + insert_prevouts(&mut *tx_update, tx.vin); } - update + tx_update .evicted_ats .extend(evicted.into_iter().map(|txid| (txid, start_time))); } + emitter.emit_partial(cursor, tx_update); + if consecutive_unused >= gap_limit { break; } } - Ok((update, last_active_index)) + Ok(last_active_index) } /// Fetch transactions and associated [`ConfirmationBlockTime`]s by scanning `spks` @@ -362,23 +403,33 @@ fn fetch_txs_with_keychain_spks /// requests to make in parallel. /// /// Refer to [crate-level docs](crate) for more. -fn fetch_txs_with_spks>( +#[allow(clippy::too_many_arguments)] +fn fetch_txs_with_spks( client: &esplora_client::BlockingClient, start_time: u64, inserted_txs: &mut HashSet, + tx_update: &mut TxUpdate, spks: I, parallel_requests: usize, -) -> Result, Error> { + emitter: &mut E, + cursor: &mut Option>, +) -> Result<(), Error> +where + I: IntoIterator, +{ fetch_txs_with_keychain_spks( client, start_time, inserted_txs, + tx_update, spks.into_iter().enumerate().map(|(i, spk)| (i as u32, spk)), usize::MAX, None, parallel_requests, + emitter, + cursor, ) - .map(|(update, _)| update) + .map(|_| ()) } /// Fetch transactions and associated [`ConfirmationBlockTime`]s by scanning `txids` @@ -387,15 +438,20 @@ fn fetch_txs_with_spks>( /// `parallel_requests` specifies the maximum number of HTTP requests to make in parallel. /// /// Refer to [crate-level docs](crate) for more. -fn fetch_txs_with_txids>( +#[allow(clippy::too_many_arguments)] +fn fetch_txs_with_txids( client: &esplora_client::BlockingClient, start_time: u64, inserted_txs: &mut HashSet, + tx_update: &mut TxUpdate, txids: I, parallel_requests: usize, -) -> Result, Error> { - let mut update = TxUpdate::::default(); - // Only fetch for non-inserted txs. + emitter: &mut E, + cursor: &mut Option>, +) -> Result<(), Error> +where + I: IntoIterator, +{ let mut txids = txids .into_iter() .filter(|txid| !inserted_txs.contains(txid)) @@ -424,14 +480,21 @@ fn fetch_txs_with_txids>( let (txid, tx_info) = handle.join().expect("thread must not panic")?; if let Some(tx_info) = tx_info { if inserted_txs.insert(txid) { - update.txs.push(tx_info.to_tx().into()); + tx_update.txs.push(tx_info.to_tx().into()); } - insert_anchor_or_seen_at_from_status(&mut update, start_time, txid, tx_info.status); - insert_prevouts(&mut update, tx_info.vin); + insert_anchor_or_seen_at_from_status( + &mut *tx_update, + start_time, + txid, + tx_info.status, + ); + insert_prevouts(&mut *tx_update, tx_info.vin); } } } - Ok(update) + + emitter.emit_partial(cursor, tx_update); + Ok(()) } /// Fetch transactions and [`ConfirmationBlockTime`]s that contain and spend the provided @@ -440,27 +503,33 @@ fn fetch_txs_with_txids>( /// `parallel_requests` specifies the maximum number of HTTP requests to make in parallel. /// /// Refer to [crate-level docs](crate) for more. -fn fetch_txs_with_outpoints>( +#[allow(clippy::too_many_arguments)] +fn fetch_txs_with_outpoints( client: &esplora_client::BlockingClient, start_time: u64, inserted_txs: &mut HashSet, + tx_update: &mut TxUpdate, outpoints: I, parallel_requests: usize, -) -> Result, Error> { + emitter: &mut E, + cursor: &mut Option>, +) -> Result<(), Error> +where + I: IntoIterator, +{ let outpoints = outpoints.into_iter().collect::>(); - let mut update = TxUpdate::::default(); - // make sure txs exists in graph and tx statuses are updated - // TODO: We should maintain a tx cache (like we do with Electrum). - update.extend(fetch_txs_with_txids( + fetch_txs_with_txids( client, start_time, inserted_txs, + tx_update, outpoints.iter().map(|op| op.txid), parallel_requests, - )?); + emitter, + cursor, + )?; - // get outpoint spend-statuses let mut outpoints = outpoints.into_iter(); let mut missing_txs = HashSet::::with_capacity(outpoints.len()); loop { @@ -492,7 +561,7 @@ fn fetch_txs_with_outpoints>( } if let Some(spend_status) = op_status.status { insert_anchor_or_seen_at_from_status( - &mut update, + &mut *tx_update, start_time, spend_txid, spend_status, @@ -502,14 +571,16 @@ fn fetch_txs_with_outpoints>( } } - update.extend(fetch_txs_with_txids( + fetch_txs_with_txids( client, start_time, inserted_txs, + tx_update, missing_txs, parallel_requests, - )?); - Ok(update) + emitter, + cursor, + ) } #[cfg(test)] diff --git a/crates/esplora/src/fetch_emit.rs b/crates/esplora/src/fetch_emit.rs new file mode 100644 index 000000000..18023151c --- /dev/null +++ b/crates/esplora/src/fetch_emit.rs @@ -0,0 +1,33 @@ +use bdk_core::{ + spk_client::{FullScanRequest, SyncRequest}, + ConfirmationBlockTime, TxUpdate, TxUpdateCursor, +}; + +/// Emit partial [`TxUpdate`]s during Esplora fetch loops. +pub(crate) trait EmitPartial { + fn emit_partial( + &mut self, + cursor: &mut Option>, + tx_update: &mut TxUpdate, + ); +} + +impl EmitPartial for SyncRequest { + fn emit_partial( + &mut self, + cursor: &mut Option>, + tx_update: &mut TxUpdate, + ) { + self.try_emit_partial_update(cursor, tx_update); + } +} + +impl EmitPartial for FullScanRequest { + fn emit_partial( + &mut self, + cursor: &mut Option>, + tx_update: &mut TxUpdate, + ) { + self.try_emit_partial_update(cursor, tx_update); + } +} diff --git a/crates/esplora/src/lib.rs b/crates/esplora/src/lib.rs index 60b4f1eb3..6c4090d73 100644 --- a/crates/esplora/src/lib.rs +++ b/crates/esplora/src/lib.rs @@ -27,6 +27,8 @@ use esplora_client::TxStatus; pub use esplora_client; +mod fetch_emit; + #[cfg(feature = "blocking")] mod blocking_ext; #[cfg(feature = "blocking")] diff --git a/crates/esplora/tests/async_ext.rs b/crates/esplora/tests/async_ext.rs index 2941e2a66..0716b6a96 100644 --- a/crates/esplora/tests/async_ext.rs +++ b/crates/esplora/tests/async_ext.rs @@ -1,6 +1,6 @@ use bdk_chain::bitcoin::{Address, Amount}; use bdk_chain::local_chain::LocalChain; -use bdk_chain::spk_client::{FullScanRequest, SyncRequest}; +use bdk_chain::spk_client::{FullScanRequest, SyncRequest, SyncRequestEvent}; use bdk_chain::spk_txout::SpkTxOutIndex; use bdk_chain::{ConfirmationBlockTime, IndexedTxGraph, TxGraph}; use bdk_esplora::EsploraAsyncExt; @@ -9,6 +9,7 @@ use bdk_testenv::{anyhow, TestEnv}; use esplora_client::{self, Builder}; use std::collections::{BTreeSet, HashSet}; use std::str::FromStr; +use std::sync::{Arc, Mutex}; use std::thread::sleep; use std::time::Duration; @@ -249,6 +250,78 @@ pub async fn test_update_tx_graph_without_keychain() -> anyhow::Result<()> { Ok(()) } +/// Async sync with `.on_event()` streams partial updates before returning. +#[tokio::test] +pub async fn sync_streams_partial_updates() -> anyhow::Result<()> { + let env = TestEnv::new()?; + let base_url = format!("http://{}", &env.electrsd.esplora_url.clone().unwrap()); + let client = Builder::new(base_url.as_str()).build_async()?; + + let receive_address0 = + Address::from_str("bcrt1qc6fweuf4xjvz4x3gx3t9e0fh4hvqyu2qw4wvxm")?.assume_checked(); + let receive_address1 = + Address::from_str("bcrt1qfjg5lv3dvc9az8patec8fjddrs4aqtauadnagr")?.assume_checked(); + + let misc_spks = [ + receive_address0.script_pubkey(), + receive_address1.script_pubkey(), + ]; + + let _block_hashes = env.mine_blocks(101, None)?; + let txid1 = env + .bitcoind + .client + .send_to_address(&receive_address1, Amount::from_sat(10000))? + .txid()?; + let txid2 = env + .bitcoind + .client + .send_to_address(&receive_address0, Amount::from_sat(20000))? + .txid()?; + let _block_hashes = env.mine_blocks(1, None)?; + while client.get_height().await.unwrap() < 102 { + sleep(Duration::from_millis(10)) + } + + let cp_tip = env.make_checkpoint_tip(); + + let partial_count = Arc::new(Mutex::new(0usize)); + let streamed_txs = Arc::new(Mutex::new(BTreeSet::new())); + let partial_count2 = partial_count.clone(); + let streamed_txs2 = streamed_txs.clone(); + + let sync_update = { + let request = SyncRequest::builder() + .chain_tip(cp_tip) + .spks(misc_spks) + .on_event(move |event| { + if let SyncRequestEvent::PartialUpdate(delta) = event { + *partial_count2.lock().unwrap() += 1; + for tx in &delta.txs { + streamed_txs2.lock().unwrap().insert(tx.compute_txid()); + } + } + }); + client.sync(request, 1).await? + }; + + assert!( + *partial_count.lock().unwrap() > 0, + "must emit partial updates during sync" + ); + assert!( + sync_update.tx_update.is_empty(), + "final tx_update must be empty when events drained all data" + ); + assert_eq!( + *streamed_txs.lock().unwrap(), + [txid1, txid2].into(), + "streamed txs must include all discovered transactions" + ); + + Ok(()) +} + /// Test the bounds of the address scan depending on the `stop_gap`. #[tokio::test] pub async fn test_async_update_tx_graph_stop_gap() -> anyhow::Result<()> { diff --git a/crates/esplora/tests/blocking_ext.rs b/crates/esplora/tests/blocking_ext.rs index 8db59947f..611df0e47 100644 --- a/crates/esplora/tests/blocking_ext.rs +++ b/crates/esplora/tests/blocking_ext.rs @@ -1,6 +1,6 @@ use bdk_chain::bitcoin::{Address, Amount}; use bdk_chain::local_chain::LocalChain; -use bdk_chain::spk_client::{FullScanRequest, SyncRequest}; +use bdk_chain::spk_client::{FullScanRequest, SyncRequest, SyncRequestEvent}; use bdk_chain::spk_txout::SpkTxOutIndex; use bdk_chain::{ConfirmationBlockTime, IndexedTxGraph, TxGraph}; use bdk_esplora::EsploraExt; @@ -9,6 +9,7 @@ use bdk_testenv::{anyhow, TestEnv}; use esplora_client::{self, Builder}; use std::collections::{BTreeSet, HashSet}; use std::str::FromStr; +use std::sync::{Arc, Mutex}; use std::thread::sleep; use std::time::Duration; @@ -247,6 +248,78 @@ pub fn test_update_tx_graph_without_keychain() -> anyhow::Result<()> { Ok(()) } +/// Blocking sync with `.on_event()` streams partial updates before returning. +#[test] +pub fn sync_streams_partial_updates() -> anyhow::Result<()> { + let env = TestEnv::new()?; + let base_url = format!("http://{}", &env.electrsd.esplora_url.clone().unwrap()); + let client = Builder::new(base_url.as_str()).build_blocking(); + + let receive_address0 = + Address::from_str("bcrt1qc6fweuf4xjvz4x3gx3t9e0fh4hvqyu2qw4wvxm")?.assume_checked(); + let receive_address1 = + Address::from_str("bcrt1qfjg5lv3dvc9az8patec8fjddrs4aqtauadnagr")?.assume_checked(); + + let misc_spks = [ + receive_address0.script_pubkey(), + receive_address1.script_pubkey(), + ]; + + let _block_hashes = env.mine_blocks(101, None)?; + let txid1 = env + .bitcoind + .client + .send_to_address(&receive_address1, Amount::from_sat(10000))? + .txid()?; + let txid2 = env + .bitcoind + .client + .send_to_address(&receive_address0, Amount::from_sat(20000))? + .txid()?; + let _block_hashes = env.mine_blocks(1, None)?; + while client.get_height().unwrap() < 102 { + sleep(Duration::from_millis(10)) + } + + let cp_tip = env.make_checkpoint_tip(); + + let partial_count = Arc::new(Mutex::new(0usize)); + let streamed_txs = Arc::new(Mutex::new(BTreeSet::new())); + let partial_count2 = partial_count.clone(); + let streamed_txs2 = streamed_txs.clone(); + + let sync_update = { + let request = SyncRequest::builder() + .chain_tip(cp_tip) + .spks(misc_spks) + .on_event(move |event| { + if let SyncRequestEvent::PartialUpdate(delta) = event { + *partial_count2.lock().unwrap() += 1; + for tx in &delta.txs { + streamed_txs2.lock().unwrap().insert(tx.compute_txid()); + } + } + }); + client.sync(request, 1)? + }; + + assert!( + *partial_count.lock().unwrap() > 0, + "must emit partial updates during sync" + ); + assert!( + sync_update.tx_update.is_empty(), + "final tx_update must be empty when events drained all data" + ); + assert_eq!( + *streamed_txs.lock().unwrap(), + [txid1, txid2].into(), + "streamed txs must include all discovered transactions" + ); + + Ok(()) +} + /// Test the bounds of the address scan depending on the `stop_gap`. #[test] pub fn test_update_tx_graph_stop_gap() -> anyhow::Result<()> {