}
impl<'c, C: bitcoincore_rpc::RpcApi> Emitter<'c, C> {
- /// Construct a new [`Emitter`] with the given RPC `client`, `last_cp` and `start_height`.
+ /// Construct a new [`Emitter`].
///
- /// * `last_cp` is the check point used to find the latest block which is still part of the best
- /// chain.
- /// * `start_height` is the block height to start emitting blocks from.
+ /// `last_cp` informs the emitter of the chain we are starting off with. This way, the emitter
+ /// can start emission from a block that connects to the original chain.
+ ///
+ /// `start_height` starts emission from a given height (if there are no conflicts with the
+ /// original chain).
pub fn new(client: &'c C, last_cp: CheckPoint, start_height: u32) -> Self {
Self {
client,
}
/// Emit the next block height and header (if any).
- pub fn next_header(&mut self) -> Result<Option<(u32, Header)>, bitcoincore_rpc::Error> {
- poll(self, |hash| self.client.get_block_header(hash))
+ pub fn next_header(&mut self) -> Result<Option<BlockEvent<Header>>, bitcoincore_rpc::Error> {
+ Ok(poll(self, |hash| self.client.get_block_header(hash))?
+ .map(|(checkpoint, block)| BlockEvent { block, checkpoint }))
}
/// Emit the next block height and block (if any).
- pub fn next_block(&mut self) -> Result<Option<(u32, Block)>, bitcoincore_rpc::Error> {
- poll(self, |hash| self.client.get_block(hash))
+ pub fn next_block(&mut self) -> Result<Option<BlockEvent<Block>>, bitcoincore_rpc::Error> {
+ Ok(poll(self, |hash| self.client.get_block(hash))?
+ .map(|(checkpoint, block)| BlockEvent { block, checkpoint }))
+ }
+}
+
+/// A newly emitted block from [`Emitter`].
+#[derive(Debug)]
+pub struct BlockEvent<B> {
+ /// Either a full [`Block`] or [`Header`] of the new block.
+ pub block: B,
+
+ /// The checkpoint of the new block.
+ ///
+ /// A [`CheckPoint`] is a node of a linked list of [`BlockId`]s. This checkpoint is linked to
+ /// all [`BlockId`]s originally passed in [`Emitter::new`] as well as emitted blocks since then.
+ /// These blocks are guaranteed to be of the same chain.
+ ///
+ /// This is important as BDK structures require block-to-apply to be connected with another
+ /// block in the original chain.
+ pub checkpoint: CheckPoint,
+}
+
+impl<B> BlockEvent<B> {
+ /// The block height of this new block.
+ pub fn block_height(&self) -> u32 {
+ self.checkpoint.height()
+ }
+
+ /// The block hash of this new block.
+ pub fn block_hash(&self) -> BlockHash {
+ self.checkpoint.hash()
+ }
+
+ /// The [`BlockId`] of a previous block that this block connects to.
+ ///
+ /// This either returns a [`BlockId`] of a previously emitted block or from the chain we started
+ /// with (passed in as `last_cp` in [`Emitter::new`]).
+ ///
+ /// This value is derived from [`BlockEvent::checkpoint`].
+ pub fn connected_to(&self) -> BlockId {
+ match self.checkpoint.prev() {
+ Some(prev_cp) => prev_cp.block_id(),
+ // there is no previous checkpoint, so just connect with itself
+ None => self.checkpoint.block_id(),
+ }
}
}
fn poll<C, V, F>(
emitter: &mut Emitter<C>,
get_item: F,
-) -> Result<Option<(u32, V)>, bitcoincore_rpc::Error>
+) -> Result<Option<(CheckPoint, V)>, bitcoincore_rpc::Error>
where
C: bitcoincore_rpc::RpcApi,
F: Fn(&BlockHash) -> Result<V, bitcoincore_rpc::Error>,
let hash = res.hash;
let item = get_item(&hash)?;
- emitter.last_cp = emitter
+ let new_cp = emitter
.last_cp
.clone()
.push(BlockId { height, hash })
.expect("must push");
+ emitter.last_cp = new_cp.clone();
emitter.last_block = Some(res);
- return Ok(Some((height, item)));
+ return Ok(Some((new_cp, item)));
}
PollResponse::NoMoreBlocks => {
emitter.last_block = None;
}
}
-fn block_to_chain_update(block: &bitcoin::Block, height: u32) -> local_chain::Update {
- let this_id = BlockId {
- height,
- hash: block.block_hash(),
- };
- let tip = if block.header.prev_blockhash == BlockHash::all_zeros() {
- CheckPoint::new(this_id)
- } else {
- CheckPoint::new(BlockId {
- height: height - 1,
- hash: block.header.prev_blockhash,
- })
- .extend(core::iter::once(this_id))
- .expect("must construct checkpoint")
- };
-
- local_chain::Update {
- tip,
- introduce_older_blocks: false,
- }
-}
-
/// Ensure that blocks are emitted in order even after reorg.
///
/// 1. Mine 101 blocks.
// see if the emitter outputs the right blocks
println!("first sync:");
- while let Some((height, block)) = emitter.next_block()? {
+ while let Some(emission) = emitter.next_block()? {
+ let height = emission.block_height();
+ let hash = emission.block_hash();
assert_eq!(
- block.block_hash(),
+ emission.block_hash(),
exp_hashes[height as usize],
"emitted block hash is unexpected"
);
- let chain_update = block_to_chain_update(&block, height);
assert_eq!(
- local_chain.apply_update(chain_update)?,
- BTreeMap::from([(height, Some(block.block_hash()))]),
+ local_chain.apply_update(local_chain::Update {
+ tip: emission.checkpoint,
+ introduce_older_blocks: false,
+ })?,
+ BTreeMap::from([(height, Some(hash))]),
"chain update changeset is unexpected",
);
}
// see if the emitter outputs the right blocks
println!("after reorg:");
let mut exp_height = exp_hashes.len() - reorged_blocks.len();
- while let Some((height, block)) = emitter.next_block()? {
+ while let Some(emission) = emitter.next_block()? {
+ let height = emission.block_height();
+ let hash = emission.block_hash();
assert_eq!(
height, exp_height as u32,
"emitted block has unexpected height"
);
assert_eq!(
- block.block_hash(),
- exp_hashes[height as usize],
+ hash, exp_hashes[height as usize],
"emitted block is unexpected"
);
- let chain_update = block_to_chain_update(&block, height);
assert_eq!(
- local_chain.apply_update(chain_update)?,
+ local_chain.apply_update(local_chain::Update {
+ tip: emission.checkpoint,
+ introduce_older_blocks: false,
+ })?,
if exp_height == exp_hashes.len() - reorged_blocks.len() {
- core::iter::once((height, Some(block.block_hash())))
+ core::iter::once((height, Some(hash)))
.chain((height + 1..exp_hashes.len() as u32).map(|h| (h, None)))
.collect::<bdk_chain::local_chain::ChangeSet>()
} else {
- BTreeMap::from([(height, Some(block.block_hash()))])
+ BTreeMap::from([(height, Some(hash))])
},
"chain update changeset is unexpected",
);
let emitter = &mut Emitter::new(&env.client, chain.tip(), 0);
- while let Some((height, block)) = emitter.next_block()? {
- let _ = chain.apply_update(block_to_chain_update(&block, height))?;
- let indexed_additions = indexed_tx_graph.apply_block_relevant(block, height);
+ while let Some(emission) = emitter.next_block()? {
+ let height = emission.block_height();
+ let _ = chain.apply_update(local_chain::Update {
+ tip: emission.checkpoint,
+ introduce_older_blocks: false,
+ })?;
+ let indexed_additions = indexed_tx_graph.apply_block_relevant(emission.block, height);
assert!(indexed_additions.is_empty());
}
// must receive mined block which will confirm the transactions.
{
- let (height, block) = emitter.next_block()?.expect("must get mined block");
- let _ = chain
- .apply_update(CheckPoint::from_header(&block.header, height).into_update(false))?;
- let indexed_additions = indexed_tx_graph.apply_block_relevant(block, height);
+ let emission = emitter.next_block()?.expect("must get mined block");
+ let height = emission.block_height();
+ let _ = chain.apply_update(local_chain::Update {
+ tip: emission.checkpoint,
+ introduce_older_blocks: false,
+ })?;
+ let indexed_additions = indexed_tx_graph.apply_block_relevant(emission.block, height);
assert!(indexed_additions.graph.txs.is_empty());
assert!(indexed_additions.graph.txouts.is_empty());
assert_eq!(indexed_additions.graph.anchors, exp_anchors);
for reorg_count in 1..=10 {
let replaced_blocks = env.reorg_empty_blocks(reorg_count)?;
- let (height, next_header) = emitter.next_header()?.expect("must emit block after reorg");
+ let next_emission = emitter.next_header()?.expect("must emit block after reorg");
assert_eq!(
- (height as usize, next_header.block_hash()),
+ (
+ next_emission.block_height() as usize,
+ next_emission.block_hash()
+ ),
replaced_blocks[0],
"block emitted after reorg should be at the reorg height"
);
where
C: bitcoincore_rpc::RpcApi,
{
- while let Some((height, block)) = emitter.next_block()? {
- process_block(recv_chain, recv_graph, block, height)?;
+ while let Some(emission) = emitter.next_block()? {
+ let height = emission.block_height();
+ process_block(recv_chain, recv_graph, emission.block, height)?;
}
Ok(())
}
// At this point, the emitter has seen all mempool transactions. It should only re-emit those
// that have introduction heights less than the emitter's last-emitted block tip.
- while let Some((height, _)) = emitter.next_header()? {
+ while let Some(emission) = emitter.next_header()? {
+ let height = emission.block_height();
// We call `mempool()` twice.
// The second call (at height `h`) should skip the tx introduced at height `h`.
for try_index in 0..2 {
.collect::<BTreeMap<_, _>>());
// `next_header` emits the replacement block of the reorg
- if let Some((height, _)) = emitter.next_header()? {
+ if let Some(emission) = emitter.next_header()? {
+ let height = emission.block_height();
println!("\t- replacement height: {}", height);
// the mempool emission (that follows the first block emission after reorg) should only
env.mine_blocks(PREMINE_COUNT, None)?;
// emit block 99a
- let (_, block_header_99a) = emitter.next_header()?.expect("block 99a header");
+ let block_header_99a = emitter.next_header()?.expect("block 99a header").block;
let block_hash_99a = block_header_99a.block_hash();
let block_hash_98a = block_header_99a.prev_blockhash;
// emit block 100a
- let (_, block_header_100a) = emitter.next_header()?.expect("block 100a header");
+ let block_header_100a = emitter.next_header()?.expect("block 100a header").block;
let block_hash_100a = block_header_100a.block_hash();
// get hash for block 101a
env.mine_blocks(3, None)?;
// emit block header 99b
- let (_, block_header_99b) = emitter.next_header()?.expect("block 99b header");
+ let block_header_99b = emitter.next_header()?.expect("block 99b header").block;
let block_hash_99b = block_header_99b.block_hash();
let block_hash_98b = block_header_99b.prev_blockhash;
use bdk_chain::{
bitcoin::{constants::genesis_block, Block, Transaction},
indexed_tx_graph, keychain,
- local_chain::{self, CheckPoint, LocalChain},
+ local_chain::{self, LocalChain},
ConfirmationTimeHeightAnchor, IndexedTxGraph,
};
use example_cli::{
#[derive(Debug)]
enum Emission {
- Block { height: u32, block: Block },
+ Block(bdk_bitcoind_rpc::BlockEvent<Block>),
Mempool(Vec<(Transaction, u64)>),
Tip(u32),
}
let mut last_db_commit = Instant::now();
let mut last_print = Instant::now();
- while let Some((height, block)) = emitter.next_block()? {
+ while let Some(emission) = emitter.next_block()? {
+ let height = emission.block_height();
+
let mut chain = chain.lock().unwrap();
let mut graph = graph.lock().unwrap();
let mut db = db.lock().unwrap();
- let chain_update =
- CheckPoint::from_header(&block.header, height).into_update(false);
let chain_changeset = chain
- .apply_update(chain_update)
+ .apply_update(local_chain::Update {
+ tip: emission.checkpoint,
+ introduce_older_blocks: false,
+ })
.expect("must always apply as we receive blocks in order from emitter");
- let graph_changeset = graph.apply_block_relevant(block, height);
+ let graph_changeset = graph.apply_block_relevant(emission.block, height);
db.stage((chain_changeset, graph_changeset));
// commit staged db changes in intervals
loop {
match emitter.next_block()? {
- Some((height, block)) => {
+ Some(block_emission) => {
+ let height = block_emission.block_height();
if sigterm_flag.load(Ordering::Acquire) {
break;
}
block_count = rpc_client.get_block_count()? as u32;
tx.send(Emission::Tip(block_count))?;
}
- tx.send(Emission::Block { height, block })?;
+ tx.send(Emission::Block(block_emission))?;
}
None => {
if await_flag(&sigterm_flag, MEMPOOL_EMIT_DELAY) {
let mut chain = chain.lock().unwrap();
let changeset = match emission {
- Emission::Block { height, block } => {
- let chain_update =
- CheckPoint::from_header(&block.header, height).into_update(false);
+ Emission::Block(block_emission) => {
+ let height = block_emission.block_height();
+ let chain_update = local_chain::Update {
+ tip: block_emission.checkpoint,
+ introduce_older_blocks: false,
+ };
let chain_changeset = chain
.apply_update(chain_update)
.expect("must always apply as we receive blocks in order from emitter");
- let graph_changeset = graph.apply_block_relevant(block, height);
+ let graph_changeset =
+ graph.apply_block_relevant(block_emission.block, height);
(chain_changeset, graph_changeset)
}
Emission::Mempool(mempool_txs) => {
projects / bdk / commitdiff