Skip to content

feat: stream transactions and progress during sync #2242

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
ANAMASGARD wants to merge 1 commit into from
Closed

feat: stream transactions and progress during sync #2242

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
290 changes: 268 additions & 22 deletions crates/core/src/spk_client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<I> = dyn FnMut(SyncItem<I>, SyncProgress) + Send + 'static;

type InspectFullScan<K> = dyn FnMut(K, u32, &Script) + Send + 'static;
type OnSyncEvent<I, A> = dyn for<'a> FnMut(SyncRequestEvent<'a, I, A>) + Send + 'static;
type OnFullScanEvent<K, A> = 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)]
Expand Down Expand Up @@ -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<A>),
/// Anchor-only update after Electrum resolves confirmation proofs.
AnchorsResolved(TxUpdate<A>),
}

/// 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<A>),
/// Anchor-only update after Electrum resolves confirmation proofs.
AnchorsResolved(TxUpdate<A>),
}

/// 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<I, D>,
}

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<ConfirmationBlockTime>) {
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<ConfirmationBlockTime>) {
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<K, D>,
}

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<ConfirmationBlockTime>) {
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<ConfirmationBlockTime>) {
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 {
Expand Down Expand Up @@ -202,13 +312,31 @@ impl<I, D> SyncRequestBuilder<I, D> {
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<F>(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<F>(mut self, inspect: F) -> Self
///
/// This is sugar over [`Self::on_event`] for progress-only callbacks.
pub fn inspect<F>(self, mut inspect: F) -> Self
where
F: FnMut(SyncItem<I>, 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`].
Expand Down Expand Up @@ -266,7 +394,10 @@ pub struct SyncRequest<I = (), D = BlockHash> {
txids_consumed: usize,
outpoints: VecDeque<OutPoint>,
outpoints_consumed: usize,
inspect: Box<InspectSync<I>>,
on_event: Box<OnSyncEvent<I, ConfirmationBlockTime>>,
/// Whether to emit partial [`TxUpdate`]s during sync (set by
/// [`SyncRequestBuilder::on_event`]).
pub emit_partial_updates: bool,
}

impl<I, D> From<SyncRequestBuilder<I, D>> for SyncRequest<I, D> {
Expand Down Expand Up @@ -296,7 +427,8 @@ impl<I, D> SyncRequest<I, D> {
txids_consumed: 0,
outpoints: VecDeque::new(),
outpoints_consumed: 0,
inspect: Box::new(|_, _| ()),
on_event: Box::new(|_| ()),
emit_partial_updates: false,
},
}
}
Expand Down Expand Up @@ -393,9 +525,46 @@ impl<I, D> SyncRequest<I, D> {
SyncIter::<I, D, OutPoint>::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<TxUpdateCursor<ConfirmationBlockTime>>,
tx_update: &mut TxUpdate<ConfirmationBlockTime>,
) {
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<TxUpdateCursor<ConfirmationBlockTime>>,
tx_update: &mut TxUpdate<ConfirmationBlockTime>,
) {
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<I>) {
let progress = self.progress();
(*self.inspect)(item, progress);
(self.on_event)(SyncRequestEvent::ItemStarted(item, progress));
}
}

Expand Down Expand Up @@ -468,13 +637,37 @@ impl<K: Ord, D> FullScanRequestBuilder<K, D> {
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<F>(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<F>(mut self, inspect: F) -> Self
///
/// This is sugar over [`Self::on_event`] for progress-only callbacks.
pub fn inspect<F>(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`].
Expand All @@ -497,7 +690,11 @@ pub struct FullScanRequest<K, D = BlockHash> {
chain_tip: Option<CheckPoint<D>>,
spks_by_keychain: BTreeMap<K, Box<dyn Iterator<Item = Indexed<ScriptBuf>> + Send>>,
last_revealed: BTreeMap<K, u32>,
inspect: Box<InspectFullScan<K>>,
full_scan_progress: FullScanProgress,
on_event: Box<OnFullScanEvent<K, ConfirmationBlockTime>>,
/// Whether to emit partial [`TxUpdate`]s during full scan (set by
/// [`FullScanRequestBuilder::on_event`]).
pub emit_partial_updates: bool,
}

impl<K, D> From<FullScanRequestBuilder<K, D>> for FullScanRequest<K, D> {
Expand All @@ -522,7 +719,9 @@ impl<K: Ord + Clone, D> FullScanRequest<K, D> {
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,
},
}
}
Expand Down Expand Up @@ -572,12 +771,48 @@ impl<K: Ord + Clone, D> FullScanRequest<K, D> {

/// Iterate over indexed [`ScriptBuf`]s contained in this request of the given `keychain`.
pub fn iter_spks(&mut self, keychain: K) -> impl Iterator<Item = Indexed<ScriptBuf>> + '_ {
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<TxUpdateCursor<ConfirmationBlockTime>>,
tx_update: &mut TxUpdate<ConfirmationBlockTime>,
) {
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<TxUpdateCursor<ConfirmationBlockTime>>,
tx_update: &mut TxUpdate<ConfirmationBlockTime>,
) {
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));
}
}
}
Expand Down Expand Up @@ -619,15 +854,26 @@ impl<K, A> FullScanResponse<K, A> {
struct KeychainSpkIter<'r, K> {
keychain: K,
spks: Option<&'r mut Box<dyn Iterator<Item = Indexed<ScriptBuf>> + Send>>,
inspect: &'r mut Box<InspectFullScan<K>>,
full_scan_progress: &'r mut FullScanProgress,
on_event: &'r mut Box<OnFullScanEvent<K, ConfirmationBlockTime>>,
}

impl<K: Ord + Clone> Iterator for KeychainSpkIter<'_, K> {
type Item = Indexed<ScriptBuf>;

fn next(&mut self) -> Option<Self::Item> {
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))
}
}
Expand Down
Loading