frost/bdk
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

bitcoind_rpc!: bring back CheckPoints to Emitter

Browse Source 
* `bdk_chain` dependency is added. In the future, we will introduce a
  separate `bdk_core` crate to contain shared types.
* replace `Emitter::new` with `from_height` and `from_checkpoint`
  * `from_height` emits from the given start height
  * `from_checkpoint` uses the provided cp to find agreement point
* introduce logic that ensures emitted blocks can connect with
  receiver's `LocalChain`
* in our rpc example, we can now `expect()` chain updates to always
  since we are using checkpoints and receiving blocks in order
This commit is contained in:
志宇
2023-10-07 00:56:01 +08:00
parent 57590e0a1f
commit 5f34df8489
4 changed files with 84 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
crates/bitcoind_rpc/src/lib.rs
View File

@@ -9,8 +9,7 @@
//! mempool.
#![warn(missing_docs)]
use std::collections::BTreeMap;
use bdk_chain::{local_chain::CheckPoint, BlockId};
use bitcoin::{block::Header, Block, BlockHash, Transaction};
pub use bitcoincore_rpc;
use bitcoincore_rpc::bitcoincore_rpc_json;
@@ -24,7 +23,7 @@ pub struct Emitter<'c, C> {
client: &'c C,
start_height: u32,
emitted_blocks: BTreeMap<u32, BlockHash>,
last_cp: Option<CheckPoint>,
last_block: Option<bitcoincore_rpc_json::GetBlockResult>,
/// The latest first-seen epoch of emitted mempool transactions. This is used to determine
@@ -37,14 +36,29 @@ pub struct Emitter<'c, C> {
}
impl<'c, C: bitcoincore_rpc::RpcApi> Emitter<'c, C> {
/// Constructs a new [`Emitter`] with the provided [`bitcoincore_rpc::Client`].
/// Construct a new [`Emitter`] with the given RPC `client` and `start_height`.
///
/// `start_height` is the block height to start emitting blocks from.
pub fn new(client: &'c C, start_height: u32) -> Self {
pub fn from_height(client: &'c C, start_height: u32) -> Self {
Self {
client,
start_height,
emitted_blocks: BTreeMap::new(),
last_cp: None,
last_block: None,
last_mempool_time: 0,
last_mempool_tip: None,
}
}
/// Construct a new [`Emitter`] with the given RPC `client` and `checkpoint`.
///
/// `checkpoint` is used to find the latest block which is still part of the best chain. The
/// [`Emitter`] will emit blocks starting right above this block.
pub fn from_checkpoint(client: &'c C, checkpoint: CheckPoint) -> Self {
Self {
client,
start_height: 0,
last_cp: Some(checkpoint),
last_block: None,
last_mempool_time: 0,
last_mempool_tip: None,
@@ -114,7 +128,7 @@ impl<'c, C: bitcoincore_rpc::RpcApi> Emitter<'c, C> {
.collect::<Result<Vec<_>, _>>()?;
self.last_mempool_time = latest_time;
self.last_mempool_tip = self.emitted_blocks.iter().last().map(|(&height, _)| height);
self.last_mempool_tip = self.last_cp.as_ref().map(|cp| cp.height());
Ok(txs_to_emit)
}
@@ -135,7 +149,7 @@ enum PollResponse {
NoMoreBlocks,
/// Fetched block is not in the best chain.
BlockNotInBestChain,
AgreementFound(bitcoincore_rpc_json::GetBlockResult),
AgreementFound(bitcoincore_rpc_json::GetBlockResult, CheckPoint),
AgreementPointNotFound,
}
@@ -146,7 +160,10 @@ where
let client = emitter.client;
if let Some(last_res) = &emitter.last_block {
assert!(!emitter.emitted_blocks.is_empty());
assert!(
emitter.last_cp.is_some(),
"must not have block result without last cp"
);
let next_hash = match last_res.nextblockhash {
None => return Ok(PollResponse::NoMoreBlocks),
@@ -160,7 +177,7 @@ where
return Ok(PollResponse::Block(res));
}
if emitter.emitted_blocks.is_empty() {
if emitter.last_cp.is_none() {
let hash = client.get_block_hash(emitter.start_height as _)?;
let res = client.get_block_info(&hash)?;
@@ -170,15 +187,15 @@ where
return Ok(PollResponse::Block(res));
}
for (&_, hash) in emitter.emitted_blocks.iter().rev() {
let res = client.get_block_info(hash)?;
for cp in emitter.last_cp.iter().flat_map(CheckPoint::iter) {
let res = client.get_block_info(&cp.hash())?;
if res.confirmations < 0 {
// block is not in best chain
continue;
}
// agreement point found
return Ok(PollResponse::AgreementFound(res));
return Ok(PollResponse::AgreementFound(res, cp));
}
Ok(PollResponse::AgreementPointNotFound)
@@ -196,9 +213,28 @@ where
match poll_once(emitter)? {
PollResponse::Block(res) => {
let height = res.height as u32;
let item = get_item(&res.hash)?;
assert_eq!(emitter.emitted_blocks.insert(height, res.hash), None);
let hash = res.hash;
let item = get_item(&hash)?;
let this_id = BlockId { height, hash };
let prev_id = res.previousblockhash.map(|prev_hash| BlockId {
height: height - 1,
hash: prev_hash,
});
match (&mut emitter.last_cp, prev_id) {
(Some(cp), _) => *cp = cp.clone().push(this_id).expect("must push"),
(last_cp, None) => *last_cp = Some(CheckPoint::new(this_id)),
// When the receiver constructs a local_chain update from a block, the previous
// checkpoint is also included in the update. We need to reflect this state in
// `Emitter::last_cp` as well.
(last_cp, Some(prev_id)) => {
*last_cp = Some(CheckPoint::new(prev_id).push(this_id).expect("must push"))
}
}
emitter.last_block = Some(res);
return Ok(Some((height, item)));
}
PollResponse::NoMoreBlocks => {
@@ -209,11 +245,11 @@ where
emitter.last_block = None;
continue;
}
PollResponse::AgreementFound(res) => {
PollResponse::AgreementFound(res, cp) => {
let agreement_h = res.height as u32;
// get rid of evicted blocks
emitter.emitted_blocks.split_off(&(agreement_h + 1));
emitter.last_cp = Some(cp);
// The tip during the last mempool emission needs to in the best chain, we reduce
// it if it is not.
@@ -226,7 +262,11 @@ where
continue;
}
PollResponse::AgreementPointNotFound => {
emitter.emitted_blocks.clear();
// We want to clear `last_cp` and set `start_height` to the first checkpoint's
// height. This way, the first checkpoint in `LocalChain` can be replaced.
if let Some(last_cp) = emitter.last_cp.take() {
emitter.start_height = last_cp.height();
}
emitter.last_block = None;
continue;
}