5165327676

pending completion

Build # 5165327676

Build Type

Pull #1002

github

Committed by

web-flow

Commit Message

Merge afb523dc8 into 8641847e6

Pull Request Pull Request #1002: Implement linked-list `LocalChain`

Run Details

1213 of 1213 new or added lines in 14 files covered. (100.0%)

7770 of 9890 relevant lines covered (78.56%)

5181.52 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::BlockId;
use bdk_chain::{keychain::LocalUpdate, ConfirmationTimeAnchor};
use esplora_client::{Error, OutputStatus, TxStatus};

use crate::map_confirmation_time_anchor;

/// 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,
        local_chain: &BTreeMap<u32, BlockHash>,
        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,
        local_chain: &BTreeMap<u32, BlockHash>,
        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(
            local_chain,
            [(
                (),
                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,
        local_chain: &BTreeMap<u32, BlockHash>,
        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 (mut update, tip_at_start) = loop {
            let mut update = LocalUpdate::<K, ConfirmationTimeAnchor>::default();

            for (&height, &original_hash) in local_chain.iter().rev() {
                let update_block_id = BlockId {
                    height,
                    hash: self.get_block_hash(height)?,
                };
                let _ = update
                    .chain
                    .insert_block(update_block_id)
                    .expect("cannot repeat height here");
                if update_block_id.hash == original_hash {
                    break;
                }
            }

            let tip_at_start = BlockId {
                height: self.get_height()?,
                hash: self.get_tip_hash()?,
            };

            if update.chain.insert_block(tip_at_start).is_ok() {
                break (update, tip_at_start);
            }
        };

        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 = map_confirmation_time_anchor(&tx.status, tip_at_start);

                        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 @ TxStatus {
                    confirmed: true, ..
                } => {
                    if let Some(anchor) = map_confirmation_time_anchor(&tx_status, tip_at_start) {
                        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 = map_confirmation_time_anchor(&status, tip_at_start);

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

        if tip_at_start.hash != self.get_block_hash(tip_at_start.height)? {
            // A reorg occurred, so let's find out where all the txids we found are now in the chain
            let txids_found = update
                .graph
                .full_txs()
                .map(|tx_node| tx_node.txid)
                .collect::<Vec<_>>();
            update.chain = EsploraExt::scan_without_keychain(
                self,
                local_chain,
                [],
                txids_found,
                [],
                parallel_requests,
            )?
            .chain;
        }

        Ok(update)
    }
}

1	use bdk_chain::bitcoin::{BlockHash, OutPoint, Script, Txid};
2	use bdk_chain::collections::BTreeMap;
3	use bdk_chain::BlockId;
4	use bdk_chain::{keychain::LocalUpdate, ConfirmationTimeAnchor};
5	use esplora_client::{Error, OutputStatus, TxStatus};
6
7	use crate::map_confirmation_time_anchor;
8
9	/// Trait to extend [`esplora_client::BlockingClient`] functionality.
10	///
11	/// Refer to [crate-level documentation] for more.
12	///
13	/// [crate-level documentation]: crate
14	pub trait EsploraExt {
15	/// Scan the blockchain (via esplora) for the data specified and returns a
16	/// [`LocalUpdate<K, ConfirmationTimeAnchor>`].
17	///
18	/// - `local_chain`: the most recent block hashes present locally
19	/// - `keychain_spks`: keychains that we want to scan transactions for
20	/// - `txids`: transactions for which we want updated [`ConfirmationTimeAnchor`]s
21	/// - `outpoints`: transactions associated with these outpoints (residing, spending) that we
22	/// want to included in the update
23	///
24	/// The scan for each keychain stops after a gap of `stop_gap` script pubkeys with no associated
25	/// transactions. `parallel_requests` specifies the max number of HTTP requests to make in
26	/// parallel.
27	#[allow(clippy::result_large_err)] // FIXME
28	fn scan<K: Ord + Clone>(
29	&self,
30	local_chain: &BTreeMap<u32, BlockHash>,
31	keychain_spks: BTreeMap<K, impl IntoIterator<Item = (u32, Script)>>,
32	txids: impl IntoIterator<Item = Txid>,
33	outpoints: impl IntoIterator<Item = OutPoint>,
34	stop_gap: usize,
35	parallel_requests: usize,
36	) -> Result<LocalUpdate<K, ConfirmationTimeAnchor>, Error>;
37
38	/// Convenience method to call [`scan`] without requiring a keychain.
39	///
40	/// [`scan`]: EsploraExt::scan
41	#[allow(clippy::result_large_err)] // FIXME
42	fn scan_without_keychain(	×
43	&self,	×
44	local_chain: &BTreeMap<u32, BlockHash>,	×
45	misc_spks: impl IntoIterator<Item = Script>,	×
46	txids: impl IntoIterator<Item = Txid>,	×
47	outpoints: impl IntoIterator<Item = OutPoint>,	×
48	parallel_requests: usize,	×
49	) -> Result<LocalUpdate<(), ConfirmationTimeAnchor>, Error> {	×
50	self.scan(	×
51	local_chain,	×
52	[(	×
53	(),	×
54	misc_spks	×
55	.into_iter()	×
56	.enumerate()	×
57	.map(\|(i, spk)\| (i as u32, spk)),	×
58	)]	×
59	.into(),	×
60	txids,	×
61	outpoints,	×
62	usize::MAX,	×
63	parallel_requests,	×
64	)	×
65	}	×
66	}
67
68	impl EsploraExt for esplora_client::BlockingClient {
69	fn scan<K: Ord + Clone>(	×
70	&self,	×
71	local_chain: &BTreeMap<u32, BlockHash>,	×
72	keychain_spks: BTreeMap<K, impl IntoIterator<Item = (u32, Script)>>,	×
73	txids: impl IntoIterator<Item = Txid>,	×
74	outpoints: impl IntoIterator<Item = OutPoint>,	×
75	stop_gap: usize,	×
76	parallel_requests: usize,	×
77	) -> Result<LocalUpdate<K, ConfirmationTimeAnchor>, Error> {	×
78	let parallel_requests = Ord::max(parallel_requests, 1);	×
79
80	let (mut update, tip_at_start) = loop {	×
81	let mut update = LocalUpdate::<K, ConfirmationTimeAnchor>::default();	×
82
83	for (&height, &original_hash) in local_chain.iter().rev() {	×
84	let update_block_id = BlockId {	×
85	height,	×
86	hash: self.get_block_hash(height)?,	×
87	};
88	let _ = update	×
89	.chain	×
90	.insert_block(update_block_id)	×
91	.expect("cannot repeat height here");	×
92	if update_block_id.hash == original_hash {	×
93	break;	×
94	}	×
95	}
96
97	let tip_at_start = BlockId {	×
98	height: self.get_height()?,	×
99	hash: self.get_tip_hash()?,	×
100	};
101
102	if update.chain.insert_block(tip_at_start).is_ok() {	×
103	break (update, tip_at_start);	×
104	}	×
105	};
106
107	for (keychain, spks) in keychain_spks {	×
108	let mut spks = spks.into_iter();	×
109	let mut last_active_index = None;	×
110	let mut empty_scripts = 0;	×
111	type IndexWithTxs = (u32, Vec<esplora_client::Tx>);
112
113	loop {	×
114	let handles = (0..parallel_requests)	×
115	.filter_map(	×
116	\|_\| -> Option<std::thread::JoinHandle<Result<IndexWithTxs, _>>> {	×
117	let (index, script) = spks.next()?;	×
118	let client = self.clone();	×
119	Some(std::thread::spawn(move \|\| {	×
120	let mut related_txs = client.scripthash_txs(&script, None)?;	×
121
122	let n_confirmed =	×
123	related_txs.iter().filter(\|tx\| tx.status.confirmed).count();	×
124	// esplora pages on 25 confirmed transactions. If there are 25 or more we	×
125	// keep requesting to see if there's more.	×
126	if n_confirmed >= 25 {	×
127	loop {
128	let new_related_txs = client.scripthash_txs(	×
129	&script,	×
130	Some(related_txs.last().unwrap().txid),	×
131	)?;	×
132	let n = new_related_txs.len();	×
133	related_txs.extend(new_related_txs);	×
134	// we've reached the end	×
135	if n < 25 {	×
136	break;	×
137	}	×
138	}
139	}	×
140
141	Result::<_, esplora_client::Error>::Ok((index, related_txs))	×
142	}))	×
143	},	×
144	)	×
145	.collect::<Vec<_>>();	×
146		×
147	let n_handles = handles.len();	×
148
149	for handle in handles {	×
150	let (index, related_txs) = handle.join().unwrap()?; // TODO: don't unwrap	×
151	if related_txs.is_empty() {	×
152	empty_scripts += 1;	×
153	} else {	×
154	last_active_index = Some(index);	×
155	empty_scripts = 0;	×
156	}	×
157	for tx in related_txs {	×
158	let anchor = map_confirmation_time_anchor(&tx.status, tip_at_start);	×
159		×
160	let _ = update.graph.insert_tx(tx.to_tx());	×
161	if let Some(anchor) = anchor {	×
162	let _ = update.graph.insert_anchor(tx.txid, anchor);	×
163	}	×
164	}
165	}
166
167	if n_handles == 0 \|\| empty_scripts >= stop_gap {	×
168	break;	×
169	}	×
170	}
171
172	if let Some(last_active_index) = last_active_index {	×
173	update.keychain.insert(keychain, last_active_index);	×
174	}	×
175	}
176
177	for txid in txids.into_iter() {	×
178	if update.graph.get_tx(txid).is_none() {	×
179	match self.get_tx(&txid)? {	×
180	Some(tx) => {	×
181	let _ = update.graph.insert_tx(tx);	×
182	}	×
183	None => continue,	×
184	}
185	}	×
186	match self.get_tx_status(&txid)? {	×
187	tx_status @ TxStatus {	×
188	confirmed: true, ..
189	} => {
190	if let Some(anchor) = map_confirmation_time_anchor(&tx_status, tip_at_start) {	×
191	let _ = update.graph.insert_anchor(txid, anchor);	×
192	}	×
193	}
194	_ => continue,	×
195	}
196	}
197
198	for op in outpoints.into_iter() {	×
199	let mut op_txs = Vec::with_capacity(2);	×
200	if let (
201	Some(tx),	×
202	tx_status @ TxStatus {	×
203	confirmed: true, ..
204	},
205	) = (self.get_tx(&op.txid)?, self.get_tx_status(&op.txid)?)	×
206	{
207	op_txs.push((tx, tx_status));	×
208	if let Some(OutputStatus {
209	txid: Some(txid),	×
210	status: Some(spend_status),	×
211	..
212	}) = self.get_output_status(&op.txid, op.vout as _)?	×
213	{
214	if let Some(spend_tx) = self.get_tx(&txid)? {	×
215	op_txs.push((spend_tx, spend_status));	×
216	}	×
217	}	×
218	}	×
219
220	for (tx, status) in op_txs {	×
221	let txid = tx.txid();	×
222	let anchor = map_confirmation_time_anchor(&status, tip_at_start);	×
223		×
224	let _ = update.graph.insert_tx(tx);	×
225	if let Some(anchor) = anchor {	×
226	let _ = update.graph.insert_anchor(txid, anchor);	×
227	}	×
228	}
229	}
230
231	if tip_at_start.hash != self.get_block_hash(tip_at_start.height)? {	×
232	// A reorg occurred, so let's find out where all the txids we found are now in the chain
233	let txids_found = update	×
234	.graph	×
235	.full_txs()	×
236	.map(\|tx_node\| tx_node.txid)	×
237	.collect::<Vec<_>>();	×
238	update.chain = EsploraExt::scan_without_keychain(	×
239	self,	×
240	local_chain,	×
241	[],	×
242	txids_found,	×
243	[],	×
244	parallel_requests,	×
245	)?	×
246	.chain;
247	}	×
248
249	Ok(update)	×
250	}	×
251	}

bitcoindevkit / bdk / 5165327676

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