5398026791

pending completion

Build # 5398026791

Build Type

Pull #1002

github

Committed by

web-flow

Commit Message

Merge 8272aa35d into 26ade1172

Pull Request Pull Request #1002: Implement linked-list `LocalChain` and add rpc-chain module/example

Run Details

681 of 681 new or added lines in 9 files covered. (100.0%)

7903 of 10220 relevant lines covered (77.33%)

5078.37 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/crates/esplora/src/blocking_ext.rs

use bdk_chain::bitcoin::{BlockHash, OutPoint, Script, Txid};
use bdk_chain::collections::BTreeMap;
use bdk_chain::local_chain::CheckPoint;
use bdk_chain::{keychain::LocalUpdate, ConfirmationTimeAnchor};
use bdk_chain::{BlockId, TxGraph};
use esplora_client::{Error, OutputStatus, TxStatus};

/// Trait to extend [`esplora_client::BlockingClient`] functionality.
///
/// Refer to [crate-level documentation] for more.
///
/// [crate-level documentation]: crate
pub trait EsploraExt {
    /// Scan the blockchain (via esplora) for the data specified and returns a
    /// [`LocalUpdate<K, ConfirmationTimeAnchor>`].
    ///
    /// - `local_chain`: the most recent block hashes present locally
    /// - `keychain_spks`: keychains that we want to scan transactions for
    /// - `txids`: transactions for which we want updated [`ConfirmationTimeAnchor`]s
    /// - `outpoints`: transactions associated with these outpoints (residing, spending) that we
    ///     want to included in the update
    ///
    /// The scan for each keychain stops after a gap of `stop_gap` script pubkeys with no associated
    /// transactions. `parallel_requests` specifies the max number of HTTP requests to make in
    /// parallel.
    #[allow(clippy::result_large_err)] // FIXME
    fn scan<K: Ord + Clone>(
        &self,
        prev_tip: Option<CheckPoint>,
        keychain_spks: BTreeMap<K, impl IntoIterator<Item = (u32, Script)>>,
        txids: impl IntoIterator<Item = Txid>,
        outpoints: impl IntoIterator<Item = OutPoint>,
        stop_gap: usize,
        parallel_requests: usize,
    ) -> Result<LocalUpdate<K, ConfirmationTimeAnchor>, Error>;

    /// Convenience method to call [`scan`] without requiring a keychain.
    ///
    /// [`scan`]: EsploraExt::scan
    #[allow(clippy::result_large_err)] // FIXME
    fn scan_without_keychain(
        &self,
        prev_tip: Option<CheckPoint>,
        misc_spks: impl IntoIterator<Item = Script>,
        txids: impl IntoIterator<Item = Txid>,
        outpoints: impl IntoIterator<Item = OutPoint>,
        parallel_requests: usize,
    ) -> Result<LocalUpdate<(), ConfirmationTimeAnchor>, Error> {
        self.scan(
            prev_tip,
            [(
                (),
                misc_spks
                    .into_iter()
                    .enumerate()
                    .map(|(i, spk)| (i as u32, spk)),
            )]
            .into(),
            txids,
            outpoints,
            usize::MAX,
            parallel_requests,
        )
    }
}

impl EsploraExt for esplora_client::BlockingClient {
    fn scan<K: Ord + Clone>(
        &self,
        prev_tip: Option<CheckPoint>,
        keychain_spks: BTreeMap<K, impl IntoIterator<Item = (u32, Script)>>,
        txids: impl IntoIterator<Item = Txid>,
        outpoints: impl IntoIterator<Item = OutPoint>,
        stop_gap: usize,
        parallel_requests: usize,
    ) -> Result<LocalUpdate<K, ConfirmationTimeAnchor>, Error> {
        let parallel_requests = Ord::max(parallel_requests, 1);

        let (tip, _) = construct_update_tip(self, prev_tip)?;
        let mut make_anchor = crate::confirmation_time_anchor_maker(&tip);
        let mut update = LocalUpdate::<K, ConfirmationTimeAnchor>::new(tip);

        for (keychain, spks) in keychain_spks {
            let mut spks = spks.into_iter();
            let mut last_active_index = None;
            let mut empty_scripts = 0;
            type IndexWithTxs = (u32, Vec<esplora_client::Tx>);

            loop {
                let handles = (0..parallel_requests)
                    .filter_map(
                        |_| -> Option<std::thread::JoinHandle<Result<IndexWithTxs, _>>> {
                            let (index, script) = spks.next()?;
                            let client = self.clone();
                            Some(std::thread::spawn(move || {
                                let mut related_txs = client.scripthash_txs(&script, None)?;

                                let n_confirmed =
                                    related_txs.iter().filter(|tx| tx.status.confirmed).count();
                                // esplora pages on 25 confirmed transactions. If there are 25 or more we
                                // keep requesting to see if there's more.
                                if n_confirmed >= 25 {
                                    loop {
                                        let new_related_txs = client.scripthash_txs(
                                            &script,
                                            Some(related_txs.last().unwrap().txid),
                                        )?;
                                        let n = new_related_txs.len();
                                        related_txs.extend(new_related_txs);
                                        // we've reached the end
                                        if n < 25 {
                                            break;
                                        }
                                    }
                                }

                                Result::<_, esplora_client::Error>::Ok((index, related_txs))
                            }))
                        },
                    )
                    .collect::<Vec<_>>();

                let n_handles = handles.len();

                for handle in handles {
                    let (index, related_txs) = handle.join().unwrap()?; // TODO: don't unwrap
                    if related_txs.is_empty() {
                        empty_scripts += 1;
                    } else {
                        last_active_index = Some(index);
                        empty_scripts = 0;
                    }
                    for tx in related_txs {
                        let anchor = make_anchor(&tx.status);
                        let _ = update.graph.insert_tx(tx.to_tx());
                        if let Some(anchor) = anchor {
                            let _ = update.graph.insert_anchor(tx.txid, anchor);
                        }
                    }
                }

                if n_handles == 0 || empty_scripts >= stop_gap {
                    break;
                }
            }

            if let Some(last_active_index) = last_active_index {
                update.keychain.insert(keychain, last_active_index);
            }
        }

        for txid in txids.into_iter() {
            if update.graph.get_tx(txid).is_none() {
                match self.get_tx(&txid)? {
                    Some(tx) => {
                        let _ = update.graph.insert_tx(tx);
                    }
                    None => continue,
                }
            }
            match self.get_tx_status(&txid)? {
                tx_status if tx_status.confirmed => {
                    if let Some(anchor) = make_anchor(&tx_status) {
                        let _ = update.graph.insert_anchor(txid, anchor);
                    }
                }
                _ => continue,
            }
        }

        for op in outpoints.into_iter() {
            let mut op_txs = Vec::with_capacity(2);
            if let (
                Some(tx),
                tx_status @ TxStatus {
                    confirmed: true, ..
                },
            ) = (self.get_tx(&op.txid)?, self.get_tx_status(&op.txid)?)
            {
                op_txs.push((tx, tx_status));
                if let Some(OutputStatus {
                    txid: Some(txid),
                    status: Some(spend_status),
                    ..
                }) = self.get_output_status(&op.txid, op.vout as _)?
                {
                    if let Some(spend_tx) = self.get_tx(&txid)? {
                        op_txs.push((spend_tx, spend_status));
                    }
                }
            }

            for (tx, status) in op_txs {
                let txid = tx.txid();
                let anchor = make_anchor(&status);

                let _ = update.graph.insert_tx(tx);
                if let Some(anchor) = anchor {
                    let _ = update.graph.insert_anchor(txid, anchor);
                }
            }
        }

        // If a reorg occured during the update, anchors may be wrong. We handle this by scrapping
        // all anchors, reconstructing checkpoints and reconstructing anchors.
        while self.get_block_hash(update.tip.height())? != update.tip.hash() {
            let (new_tip, _) = construct_update_tip(self, Some(update.tip.clone()))?;
            make_anchor = crate::confirmation_time_anchor_maker(&new_tip);

            // Reconstruct graph with only transactions (no anchors).
            update.graph = TxGraph::new(update.graph.full_txs().map(|n| n.tx.clone()));
            update.tip = new_tip;

            // Re-fetch anchors.
            let anchors = update
                .graph
                .full_txs()
                .filter_map(|n| match self.get_tx_status(&n.txid) {
                    Err(err) => Some(Err(err)),
                    Ok(status) if status.confirmed => make_anchor(&status).map(|a| Ok((n.txid, a))),
                    _ => None,
                })
                .collect::<Result<Vec<_>, _>>()?;
            for (txid, anchor) in anchors {
                let _ = update.graph.insert_anchor(txid, anchor);
            }
        }

        Ok(update)
    }
}

/// Constructs a new checkpoint tip that can "connect" to our previous checkpoint history. We return
/// the new checkpoint tip alongside the height of agreement between the two histories (if any).
#[allow(clippy::result_large_err)]
fn construct_update_tip(
    client: &esplora_client::BlockingClient,
    prev_tip: Option<CheckPoint>,
) -> Result<(CheckPoint, Option<u32>), Error> {
    let new_tip_height = client.get_height()?;

    // If esplora returns a tip height that is lower than our previous tip, then checkpoints do not
    // need updating. We just return the previous tip and use that as the point of agreement.
    if let Some(prev_tip) = prev_tip.as_ref() {
        if new_tip_height < prev_tip.height() {
            return Ok((prev_tip.clone(), Some(prev_tip.height())));
        }
    }

    // Grab latest blocks from esplora atomically first. We assume that deeper blocks cannot be
    // reorged. This ensures that our checkpoint history is consistent.
    let mut new_blocks = {
        let heights = (0..new_tip_height).rev();
        let hashes = client
            .get_blocks(Some(new_tip_height))?
            .into_iter()
            .map(|b| b.id);
        heights.zip(hashes).collect::<BTreeMap<u32, BlockHash>>()
    };

    let mut agreement_cp = Option::<CheckPoint>::None;

    for cp in prev_tip.iter().flat_map(CheckPoint::iter) {
        let cp_block = cp.block_id();

        // We check esplora blocks cached in `new_blocks` first, keeping the checkpoint history
        // consistent even during reorgs.
        let hash = match new_blocks.get(&cp_block.height) {
            Some(&hash) => hash,
            None => {
                assert!(
                    new_tip_height >= cp_block.height,
                    "already checked that esplora's tip cannot be smaller"
                );
                let hash = client.get_block_hash(cp_block.height)?;
                new_blocks.insert(cp_block.height, hash);
                hash
            }
        };

        if hash == cp_block.hash {
            agreement_cp = Some(cp);
            break;
        }
    }

    let agreement_height = agreement_cp.as_ref().map(CheckPoint::height);

    let new_tip = new_blocks
        .into_iter()
        // Prune `new_blocks` to only include blocks that are actually new.
        .filter(|(height, _)| Some(*height) > agreement_height)
        .map(|(height, hash)| BlockId { height, hash })
        .fold(agreement_cp, |prev_cp, block| {
            Some(match prev_cp {
                Some(cp) => cp.extend(block).expect("must extend cp"),
                None => CheckPoint::new(block),
            })
        })
        .expect("must have at least one checkpoint");

    Ok((new_tip, agreement_height))
}

1	use bdk_chain::bitcoin::{BlockHash, OutPoint, Script, Txid};
2	use bdk_chain::collections::BTreeMap;
3	use bdk_chain::local_chain::CheckPoint;
4	use bdk_chain::{keychain::LocalUpdate, ConfirmationTimeAnchor};
5	use bdk_chain::{BlockId, TxGraph};
6	use esplora_client::{Error, OutputStatus, TxStatus};
7
8	/// Trait to extend [`esplora_client::BlockingClient`] functionality.
9	///
10	/// Refer to [crate-level documentation] for more.
11	///
12	/// [crate-level documentation]: crate
13	pub trait EsploraExt {
14	/// Scan the blockchain (via esplora) for the data specified and returns a
15	/// [`LocalUpdate<K, ConfirmationTimeAnchor>`].
16	///
17	/// - `local_chain`: the most recent block hashes present locally
18	/// - `keychain_spks`: keychains that we want to scan transactions for
19	/// - `txids`: transactions for which we want updated [`ConfirmationTimeAnchor`]s
20	/// - `outpoints`: transactions associated with these outpoints (residing, spending) that we
21	/// want to included in the update
22	///
23	/// The scan for each keychain stops after a gap of `stop_gap` script pubkeys with no associated
24	/// transactions. `parallel_requests` specifies the max number of HTTP requests to make in
25	/// parallel.
26	#[allow(clippy::result_large_err)] // FIXME
27	fn scan<K: Ord + Clone>(
28	&self,
29	prev_tip: Option<CheckPoint>,
30	keychain_spks: BTreeMap<K, impl IntoIterator<Item = (u32, Script)>>,
31	txids: impl IntoIterator<Item = Txid>,
32	outpoints: impl IntoIterator<Item = OutPoint>,
33	stop_gap: usize,
34	parallel_requests: usize,
35	) -> Result<LocalUpdate<K, ConfirmationTimeAnchor>, Error>;
36
37	/// Convenience method to call [`scan`] without requiring a keychain.
38	///
39	/// [`scan`]: EsploraExt::scan
40	#[allow(clippy::result_large_err)] // FIXME
41	fn scan_without_keychain(	×
42	&self,	×
43	prev_tip: Option<CheckPoint>,	×
44	misc_spks: impl IntoIterator<Item = Script>,	×
45	txids: impl IntoIterator<Item = Txid>,	×
46	outpoints: impl IntoIterator<Item = OutPoint>,	×
47	parallel_requests: usize,	×
48	) -> Result<LocalUpdate<(), ConfirmationTimeAnchor>, Error> {	×
49	self.scan(	×
50	prev_tip,	×
51	[(	×
52	(),	×
53	misc_spks	×
54	.into_iter()	×
55	.enumerate()	×
56	.map(\|(i, spk)\| (i as u32, spk)),	×
57	)]	×
58	.into(),	×
59	txids,	×
60	outpoints,	×
61	usize::MAX,	×
62	parallel_requests,	×
63	)	×
64	}	×
65	}
66
67	impl EsploraExt for esplora_client::BlockingClient {
68	fn scan<K: Ord + Clone>(	×
69	&self,	×
70	prev_tip: Option<CheckPoint>,	×
71	keychain_spks: BTreeMap<K, impl IntoIterator<Item = (u32, Script)>>,	×
72	txids: impl IntoIterator<Item = Txid>,	×
73	outpoints: impl IntoIterator<Item = OutPoint>,	×
74	stop_gap: usize,	×
75	parallel_requests: usize,	×
76	) -> Result<LocalUpdate<K, ConfirmationTimeAnchor>, Error> {	×
77	let parallel_requests = Ord::max(parallel_requests, 1);	×
78
79	let (tip, _) = construct_update_tip(self, prev_tip)?;	×
80	let mut make_anchor = crate::confirmation_time_anchor_maker(&tip);	×
81	let mut update = LocalUpdate::<K, ConfirmationTimeAnchor>::new(tip);	×
82
83	for (keychain, spks) in keychain_spks {	×
84	let mut spks = spks.into_iter();	×
85	let mut last_active_index = None;	×
86	let mut empty_scripts = 0;	×
87	type IndexWithTxs = (u32, Vec<esplora_client::Tx>);
88
89	loop {	×
90	let handles = (0..parallel_requests)	×
91	.filter_map(	×
92	\|_\| -> Option<std::thread::JoinHandle<Result<IndexWithTxs, _>>> {	×
93	let (index, script) = spks.next()?;	×
94	let client = self.clone();	×
95	Some(std::thread::spawn(move \|\| {	×
96	let mut related_txs = client.scripthash_txs(&script, None)?;	×
97
98	let n_confirmed =	×
99	related_txs.iter().filter(\|tx\| tx.status.confirmed).count();	×
100	// esplora pages on 25 confirmed transactions. If there are 25 or more we	×
101	// keep requesting to see if there's more.	×
102	if n_confirmed >= 25 {	×
103	loop {
104	let new_related_txs = client.scripthash_txs(	×
105	&script,	×
106	Some(related_txs.last().unwrap().txid),	×
107	)?;	×
108	let n = new_related_txs.len();	×
109	related_txs.extend(new_related_txs);	×
110	// we've reached the end	×
111	if n < 25 {	×
112	break;	×
113	}	×
114	}
115	}	×
116
117	Result::<_, esplora_client::Error>::Ok((index, related_txs))	×
118	}))	×
119	},	×
120	)	×
121	.collect::<Vec<_>>();	×
122		×
123	let n_handles = handles.len();	×
124
125	for handle in handles {	×
126	let (index, related_txs) = handle.join().unwrap()?; // TODO: don't unwrap	×
127	if related_txs.is_empty() {	×
128	empty_scripts += 1;	×
129	} else {	×
130	last_active_index = Some(index);	×
131	empty_scripts = 0;	×
132	}	×
133	for tx in related_txs {	×
134	let anchor = make_anchor(&tx.status);	×
135	let _ = update.graph.insert_tx(tx.to_tx());	×
136	if let Some(anchor) = anchor {	×
137	let _ = update.graph.insert_anchor(tx.txid, anchor);	×
138	}	×
139	}
140	}
141
142	if n_handles == 0 \|\| empty_scripts >= stop_gap {	×
143	break;	×
144	}	×
145	}
146
147	if let Some(last_active_index) = last_active_index {	×
148	update.keychain.insert(keychain, last_active_index);	×
149	}	×
150	}
151
152	for txid in txids.into_iter() {	×
153	if update.graph.get_tx(txid).is_none() {	×
154	match self.get_tx(&txid)? {	×
155	Some(tx) => {	×
156	let _ = update.graph.insert_tx(tx);	×
157	}	×
158	None => continue,	×
159	}
160	}	×
161	match self.get_tx_status(&txid)? {	×
162	tx_status if tx_status.confirmed => {	×
163	if let Some(anchor) = make_anchor(&tx_status) {	×
164	let _ = update.graph.insert_anchor(txid, anchor);	×
165	}	×
166	}
167	_ => continue,	×
168	}
169	}
170
171	for op in outpoints.into_iter() {	×
172	let mut op_txs = Vec::with_capacity(2);	×
173	if let (
174	Some(tx),	×
175	tx_status @ TxStatus {	×
176	confirmed: true, ..
177	},
178	) = (self.get_tx(&op.txid)?, self.get_tx_status(&op.txid)?)	×
179	{
180	op_txs.push((tx, tx_status));	×
181	if let Some(OutputStatus {
182	txid: Some(txid),	×
183	status: Some(spend_status),	×
184	..
185	}) = self.get_output_status(&op.txid, op.vout as _)?	×
186	{
187	if let Some(spend_tx) = self.get_tx(&txid)? {	×
188	op_txs.push((spend_tx, spend_status));	×
189	}	×
190	}	×
191	}	×
192
193	for (tx, status) in op_txs {	×
194	let txid = tx.txid();	×
195	let anchor = make_anchor(&status);	×
196		×
197	let _ = update.graph.insert_tx(tx);	×
198	if let Some(anchor) = anchor {	×
199	let _ = update.graph.insert_anchor(txid, anchor);	×
200	}	×
201	}
202	}
203
204	// If a reorg occured during the update, anchors may be wrong. We handle this by scrapping
205	// all anchors, reconstructing checkpoints and reconstructing anchors.
206	while self.get_block_hash(update.tip.height())? != update.tip.hash() {	×
207	let (new_tip, _) = construct_update_tip(self, Some(update.tip.clone()))?;	×
208	make_anchor = crate::confirmation_time_anchor_maker(&new_tip);	×
209		×
210	// Reconstruct graph with only transactions (no anchors).	×
211	update.graph = TxGraph::new(update.graph.full_txs().map(\|n\| n.tx.clone()));	×
212	update.tip = new_tip;	×
213
214	// Re-fetch anchors.
215	let anchors = update	×
216	.graph	×
217	.full_txs()	×
218	.filter_map(\|n\| match self.get_tx_status(&n.txid) {	×
219	Err(err) => Some(Err(err)),	×
220	Ok(status) if status.confirmed => make_anchor(&status).map(\|a\| Ok((n.txid, a))),	×
221	_ => None,	×
222	})	×
223	.collect::<Result<Vec<_>, _>>()?;	×
224	for (txid, anchor) in anchors {	×
225	let _ = update.graph.insert_anchor(txid, anchor);	×
226	}	×
227	}
228
229	Ok(update)	×
230	}	×
231	}
232
233	/// Constructs a new checkpoint tip that can "connect" to our previous checkpoint history. We return
234	/// the new checkpoint tip alongside the height of agreement between the two histories (if any).
235	#[allow(clippy::result_large_err)]
236	fn construct_update_tip(	×
237	client: &esplora_client::BlockingClient,	×
238	prev_tip: Option<CheckPoint>,	×
239	) -> Result<(CheckPoint, Option<u32>), Error> {	×
240	let new_tip_height = client.get_height()?;	×
241
242	// If esplora returns a tip height that is lower than our previous tip, then checkpoints do not
243	// need updating. We just return the previous tip and use that as the point of agreement.
244	if let Some(prev_tip) = prev_tip.as_ref() {	×
245	if new_tip_height < prev_tip.height() {	×
246	return Ok((prev_tip.clone(), Some(prev_tip.height())));	×
247	}	×
248	}	×
249
250	// Grab latest blocks from esplora atomically first. We assume that deeper blocks cannot be
251	// reorged. This ensures that our checkpoint history is consistent.
252	let mut new_blocks = {	×
253	let heights = (0..new_tip_height).rev();	×
254	let hashes = client	×
255	.get_blocks(Some(new_tip_height))?	×
256	.into_iter()	×
257	.map(\|b\| b.id);	×
258	heights.zip(hashes).collect::<BTreeMap<u32, BlockHash>>()	×
259	};	×
260		×
261	let mut agreement_cp = Option::<CheckPoint>::None;	×
262
263	for cp in prev_tip.iter().flat_map(CheckPoint::iter) {	×
264	let cp_block = cp.block_id();	×
265
266	// We check esplora blocks cached in `new_blocks` first, keeping the checkpoint history
267	// consistent even during reorgs.
268	let hash = match new_blocks.get(&cp_block.height) {	×
269	Some(&hash) => hash,	×
270	None => {
271	assert!(	×
272	new_tip_height >= cp_block.height,	×
273	"already checked that esplora's tip cannot be smaller"	×
274	);
275	let hash = client.get_block_hash(cp_block.height)?;	×
276	new_blocks.insert(cp_block.height, hash);	×
277	hash	×
278	}
279	};
280
281	if hash == cp_block.hash {	×
282	agreement_cp = Some(cp);	×
283	break;	×
284	}	×
285	}
286
287	let agreement_height = agreement_cp.as_ref().map(CheckPoint::height);	×
288		×
289	let new_tip = new_blocks	×
290	.into_iter()	×
291	// Prune `new_blocks` to only include blocks that are actually new.	×
292	.filter(\|(height, _)\| Some(*height) > agreement_height)	×
293	.map(\|(height, hash)\| BlockId { height, hash })	×
294	.fold(agreement_cp, \|prev_cp, block\| {	×
295	Some(match prev_cp {	×
296	Some(cp) => cp.extend(block).expect("must extend cp"),	×
297	None => CheckPoint::new(block),	×
298	})
299	})	×
300	.expect("must have at least one checkpoint");	×
301		×
302	Ok((new_tip, agreement_height))	×
303	}	×

bitcoindevkit / bdk / 5398026791

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous