5363628790

pending completion

Build # 5363628790

Build Type

Pull #1002

github

Committed by

web-flow

Commit Message

Merge 64e8b2e32 into 5d1f922b3

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

Run Details

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

7904 of 10206 relevant lines covered (77.44%)

5065.64 hits per line

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

0.64

/crates/bitcoind_rpc/src/lib.rs

use std::collections::HashSet;

use bdk_chain::{
    bitcoin::{Block, Transaction, Txid},
    keychain::LocalUpdate,
    local_chain::CheckPoint,
    BlockId, ConfirmationHeightAnchor, ConfirmationTimeAnchor, TxGraph,
};
pub use bitcoincore_rpc;
use bitcoincore_rpc::{bitcoincore_rpc_json::GetBlockResult, Client, RpcApi};

#[derive(Debug, Clone)]
pub enum BitcoindRpcItem {
    Block {
        cp: CheckPoint,
        info: Box<GetBlockResult>,
        block: Box<Block>,
    },
    Mempool {
        cp: CheckPoint,
        txs: Vec<(Transaction, u64)>,
    },
}

pub fn confirmation_height_anchor(
    info: &GetBlockResult,
    _txid: Txid,
    _tx_pos: usize,
) -> ConfirmationHeightAnchor {
    ConfirmationHeightAnchor {
        anchor_block: BlockId {
            height: info.height as _,
            hash: info.hash,
        },
        confirmation_height: info.height as _,
    }
}

pub fn confirmation_time_anchor(
    info: &GetBlockResult,
    _txid: Txid,
    _tx_pos: usize,
) -> ConfirmationTimeAnchor {
    ConfirmationTimeAnchor {
        anchor_block: BlockId {
            height: info.height as _,
            hash: info.hash,
        },
        confirmation_height: info.height as _,
        confirmation_time: info.time as _,
    }
}

impl BitcoindRpcItem {
    pub fn is_mempool(&self) -> bool {
        matches!(self, Self::Mempool { .. })
    }

    pub fn into_update<K, A, F>(self, anchor: F) -> LocalUpdate<K, A>
    where
        A: Clone + Ord + PartialOrd,
        F: Fn(&GetBlockResult, Txid, usize) -> A,
    {
        match self {
            BitcoindRpcItem::Block { cp, info, block } => LocalUpdate {
                graph: {
                    let mut g = TxGraph::<A>::new(block.txdata);
                    for (tx_pos, &txid) in info.tx.iter().enumerate() {
                        let _ = g.insert_anchor(txid, anchor(&info, txid, tx_pos));
                    }
                    g
                },
                ..LocalUpdate::new(cp)
            },
            BitcoindRpcItem::Mempool { cp, txs } => LocalUpdate {
                graph: {
                    let mut last_seens = Vec::<(Txid, u64)>::with_capacity(txs.len());
                    let mut g = TxGraph::<A>::new(txs.into_iter().map(|(tx, last_seen)| {
                        last_seens.push((tx.txid(), last_seen));
                        tx
                    }));
                    for (txid, seen_at) in last_seens {
                        let _ = g.insert_seen_at(txid, seen_at);
                    }
                    g
                },
                ..LocalUpdate::new(cp)
            },
        }
    }
}

pub struct BitcoindRpcIter<'a> {
    client: &'a Client,
    fallback_height: u32,

    last_cp: Option<CheckPoint>,
    last_info: Option<GetBlockResult>,

    seen_txids: HashSet<Txid>,
}

impl<'a> Iterator for BitcoindRpcIter<'a> {
    type Item = Result<BitcoindRpcItem, bitcoincore_rpc::Error>;

    fn next(&mut self) -> Option<Self::Item> {
        self.next_emission().transpose()
    }
}

impl<'a> BitcoindRpcIter<'a> {
    pub fn new(client: &'a Client, fallback_height: u32, last_cp: Option<CheckPoint>) -> Self {
        Self {
            client,
            fallback_height,
            last_cp,
            last_info: None,
            seen_txids: HashSet::new(),
        }
    }

    fn next_emission(&mut self) -> Result<Option<BitcoindRpcItem>, bitcoincore_rpc::Error> {
        let client = self.client;

        'main_loop: loop {
            match (&mut self.last_cp, &mut self.last_info) {
                (last_cp @ None, last_info @ None) => {
                    // get first item at fallback_height
                    let info = client
                        .get_block_info(&client.get_block_hash(self.fallback_height as _)?)?;
                    let block = self.client.get_block(&info.hash)?;
                    let cp = CheckPoint::new(BlockId {
                        height: info.height as _,
                        hash: info.hash,
                    });
                    *last_info = Some(info.clone());
                    *last_cp = Some(cp.clone());
                    return Ok(Some(BitcoindRpcItem::Block {
                        cp,
                        info: Box::new(info),
                        block: Box::new(block),
                    }));
                }
                (last_cp @ Some(_), last_info @ None) => {
                    'cp_loop: for cp in last_cp.clone().iter().flat_map(CheckPoint::iter) {
                        let cp_block = cp.block_id();

                        let info = client.get_block_info(&cp_block.hash)?;
                        if info.confirmations < 0 {
                            // block is not in the main chain
                            continue 'cp_loop;
                        }

                        // agreement
                        *last_cp = Some(cp);
                        *last_info = Some(info);
                        continue 'main_loop;
                    }

                    // no point of agreement found
                    // next loop will emit block @ fallback height
                    *last_cp = None;
                    *last_info = None;
                }
                (Some(last_cp), last_info @ Some(_)) => {
                    // find next block
                    match last_info.as_ref().unwrap().nextblockhash {
                        Some(next_hash) => {
                            let info = self.client.get_block_info(&next_hash)?;

                            if info.confirmations < 0 {
                                *last_info = None;
                                continue 'main_loop;
                            }

                            let block = self.client.get_block(&info.hash)?;
                            let cp = last_cp
                                .clone()
                                .extend(BlockId {
                                    height: info.height as _,
                                    hash: info.hash,
                                })
                                .expect("must extend from checkpoint");

                            *last_cp = cp.clone();
                            *last_info = Some(info.clone());

                            return Ok(Some(BitcoindRpcItem::Block {
                                cp,
                                info: Box::new(info),
                                block: Box::new(block),
                            }));
                        }
                        None => {
                            // emit from mempool!
                            let mempool_txs = client
                                .get_raw_mempool()?
                                .into_iter()
                                .filter(|&txid| self.seen_txids.insert(txid))
                                .map(
                                    |txid| -> Result<(Transaction, u64), bitcoincore_rpc::Error> {
                                        let first_seen = client
                                            .get_mempool_entry(&txid)
                                            .map(|entry| entry.time)?;
                                        let tx = client.get_raw_transaction(&txid, None)?;
                                        Ok((tx, first_seen))
                                    },
                                )
                                .collect::<Result<Vec<_>, _>>()?;

                            // remove last info...
                            *last_info = None;

                            return Ok(Some(BitcoindRpcItem::Mempool {
                                txs: mempool_txs,
                                cp: last_cp.clone(),
                            }));
                        }
                    }
                }
                (None, Some(info)) => unreachable!("got info with no checkpoint? info={:#?}", info),
            }
        }
    }
}

pub trait BitcoindRpcErrorExt {
    fn is_not_found_error(&self) -> bool;
}

impl BitcoindRpcErrorExt for bitcoincore_rpc::Error {
    fn is_not_found_error(&self) -> bool {
        if let bitcoincore_rpc::Error::JsonRpc(bitcoincore_rpc::jsonrpc::Error::Rpc(rpc_err)) = self
        {
            rpc_err.code == -5
        } else {
            false
        }
    }
}

1	use std::collections::HashSet;	1✔
2
3	use bdk_chain::{
4	bitcoin::{Block, Transaction, Txid},
5	keychain::LocalUpdate,
6	local_chain::CheckPoint,
7	BlockId, ConfirmationHeightAnchor, ConfirmationTimeAnchor, TxGraph,
8	};
9	pub use bitcoincore_rpc;
10	use bitcoincore_rpc::{bitcoincore_rpc_json::GetBlockResult, Client, RpcApi};
11
12	#[derive(Debug, Clone)]	×
13	pub enum BitcoindRpcItem {
14	Block {
15	cp: CheckPoint,
16	info: Box<GetBlockResult>,
17	block: Box<Block>,
18	},
19	Mempool {
20	cp: CheckPoint,
21	txs: Vec<(Transaction, u64)>,
22	},
23	}
24
25	pub fn confirmation_height_anchor(	×
26	info: &GetBlockResult,	×
27	_txid: Txid,	×
28	_tx_pos: usize,	×
29	) -> ConfirmationHeightAnchor {	×
30	ConfirmationHeightAnchor {	×
31	anchor_block: BlockId {	×
32	height: info.height as _,	×
33	hash: info.hash,	×
34	},	×
35	confirmation_height: info.height as _,	×
36	}	×
37	}	×
38
39	pub fn confirmation_time_anchor(	×
40	info: &GetBlockResult,	×
41	_txid: Txid,	×
42	_tx_pos: usize,	×
43	) -> ConfirmationTimeAnchor {	×
44	ConfirmationTimeAnchor {	×
45	anchor_block: BlockId {	×
46	height: info.height as _,	×
47	hash: info.hash,	×
48	},	×
49	confirmation_height: info.height as _,	×
50	confirmation_time: info.time as _,	×
51	}	×
52	}	×
53
54	impl BitcoindRpcItem {
55	pub fn is_mempool(&self) -> bool {	×
56	matches!(self, Self::Mempool { .. })	×
57	}	×
58
59	pub fn into_update<K, A, F>(self, anchor: F) -> LocalUpdate<K, A>	×
60	where	×
61	A: Clone + Ord + PartialOrd,	×
62	F: Fn(&GetBlockResult, Txid, usize) -> A,	×
63	{	×
64	match self {	×
65	BitcoindRpcItem::Block { cp, info, block } => LocalUpdate {	×
66	graph: {	×
67	let mut g = TxGraph::<A>::new(block.txdata);	×
68	for (tx_pos, &txid) in info.tx.iter().enumerate() {	×
69	let _ = g.insert_anchor(txid, anchor(&info, txid, tx_pos));	×
70	}	×
71	g	×
72	},	×
73	..LocalUpdate::new(cp)	×
74	},
75	BitcoindRpcItem::Mempool { cp, txs } => LocalUpdate {	×
76	graph: {	×
77	let mut last_seens = Vec::<(Txid, u64)>::with_capacity(txs.len());	×
78	let mut g = TxGraph::<A>::new(txs.into_iter().map(\|(tx, last_seen)\| {	×
79	last_seens.push((tx.txid(), last_seen));	×
80	tx	×
81	}));	×
82	for (txid, seen_at) in last_seens {	×
83	let _ = g.insert_seen_at(txid, seen_at);	×
84	}	×
85	g	×
86	},	×
87	..LocalUpdate::new(cp)	×
88	},
89	}
90	}	×
91	}
92
93	pub struct BitcoindRpcIter<'a> {
94	client: &'a Client,
95	fallback_height: u32,
96
97	last_cp: Option<CheckPoint>,
98	last_info: Option<GetBlockResult>,
99
100	seen_txids: HashSet<Txid>,
101	}
102
103	impl<'a> Iterator for BitcoindRpcIter<'a> {
104	type Item = Result<BitcoindRpcItem, bitcoincore_rpc::Error>;
105
106	fn next(&mut self) -> Option<Self::Item> {	×
107	self.next_emission().transpose()	×
108	}	×
109	}
110
111	impl<'a> BitcoindRpcIter<'a> {
112	pub fn new(client: &'a Client, fallback_height: u32, last_cp: Option<CheckPoint>) -> Self {	×
113	Self {	×
114	client,	×
115	fallback_height,	×
116	last_cp,	×
117	last_info: None,	×
118	seen_txids: HashSet::new(),	×
119	}	×
120	}	×
121
122	fn next_emission(&mut self) -> Result<Option<BitcoindRpcItem>, bitcoincore_rpc::Error> {	×
123	let client = self.client;	×
124
125	'main_loop: loop {
126	match (&mut self.last_cp, &mut self.last_info) {	×
127	(last_cp @ None, last_info @ None) => {	×
128	// get first item at fallback_height
129	let info = client	×
130	.get_block_info(&client.get_block_hash(self.fallback_height as _)?)?;	×
131	let block = self.client.get_block(&info.hash)?;	×
132	let cp = CheckPoint::new(BlockId {	×
133	height: info.height as _,	×
134	hash: info.hash,	×
135	});	×
136	*last_info = Some(info.clone());	×
137	*last_cp = Some(cp.clone());	×
138	return Ok(Some(BitcoindRpcItem::Block {	×
139	cp,	×
140	info: Box::new(info),	×
141	block: Box::new(block),	×
142	}));	×
143	}
144	(last_cp @ Some(_), last_info @ None) => {	×
145	'cp_loop: for cp in last_cp.clone().iter().flat_map(CheckPoint::iter) {	×
146	let cp_block = cp.block_id();	×
147
148	let info = client.get_block_info(&cp_block.hash)?;	×
149	if info.confirmations < 0 {	×
150	// block is not in the main chain
151	continue 'cp_loop;	×
152	}	×
153		×
154	// agreement	×
155	*last_cp = Some(cp);	×
156	*last_info = Some(info);	×
157	continue 'main_loop;	×
158	}
159
160	// no point of agreement found
161	// next loop will emit block @ fallback height
162	*last_cp = None;	×
163	*last_info = None;	×
164	}
165	(Some(last_cp), last_info @ Some(_)) => {	×
166	// find next block	×
167	match last_info.as_ref().unwrap().nextblockhash {	×
168	Some(next_hash) => {	×
169	let info = self.client.get_block_info(&next_hash)?;	×
170
171	if info.confirmations < 0 {	×
172	*last_info = None;	×
173	continue 'main_loop;	×
174	}	×
175
176	let block = self.client.get_block(&info.hash)?;	×
177	let cp = last_cp	×
178	.clone()	×
179	.extend(BlockId {	×
180	height: info.height as _,	×
181	hash: info.hash,	×
182	})	×
183	.expect("must extend from checkpoint");	×
184		×
185	*last_cp = cp.clone();	×
186	*last_info = Some(info.clone());	×
187		×
188	return Ok(Some(BitcoindRpcItem::Block {	×
189	cp,	×
190	info: Box::new(info),	×
191	block: Box::new(block),	×
192	}));	×
193	}
194	None => {
195	// emit from mempool!
196	let mempool_txs = client	×
197	.get_raw_mempool()?	×
198	.into_iter()	×
199	.filter(\|&txid\| self.seen_txids.insert(txid))	×
200	.map(	×
201	\|txid\| -> Result<(Transaction, u64), bitcoincore_rpc::Error> {	×
202	let first_seen = client	×
203	.get_mempool_entry(&txid)	×
204	.map(\|entry\| entry.time)?;	×
205	let tx = client.get_raw_transaction(&txid, None)?;	×
206	Ok((tx, first_seen))	×
207	},	×
208	)	×
209	.collect::<Result<Vec<_>, _>>()?;	×
210
211	// remove last info...
212	*last_info = None;	×
213		×
214	return Ok(Some(BitcoindRpcItem::Mempool {	×
215	txs: mempool_txs,	×
216	cp: last_cp.clone(),	×
217	}));	×
218	}
219	}
220	}
221	(None, Some(info)) => unreachable!("got info with no checkpoint? info={:#?}", info),	×
222	}
223	}
224	}	×
225	}
226
227	pub trait BitcoindRpcErrorExt {
228	fn is_not_found_error(&self) -> bool;
229	}
230
231	impl BitcoindRpcErrorExt for bitcoincore_rpc::Error {
232	fn is_not_found_error(&self) -> bool {
233	if let bitcoincore_rpc::Error::JsonRpc(bitcoincore_rpc::jsonrpc::Error::Rpc(rpc_err)) = self	×
234	{
235	rpc_err.code == -5	×
236	} else {
237	false	×
238	}
239	}	×
240	}

bitcoindevkit / bdk / 5363628790

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