|
| 1 | +//! This crate is used for emitting blockchain data from the `bitcoind` RPC interface. It does not |
| 2 | +//! use the wallet RPC API, so this crate can be used with wallet-disabled Bitcoin Core nodes. |
| 3 | +//! |
| 4 | +//! [`Emitter`] is the main structure which sources blockchain data from [`bitcoincore_rpc::Client`]. |
| 5 | +//! |
| 6 | +//! To only get block updates (exclude mempool transactions), the caller can use |
| 7 | +//! [`Emitter::next_block`] or/and [`Emitter::next_header`] until it returns `Ok(None)` (which means |
| 8 | +//! the chain tip is reached). A separate method, [`Emitter::mempool`] can be used to emit the whole |
| 9 | +//! mempool. |
| 10 | +#![warn(missing_docs)] |
| 11 | + |
| 12 | +use bdk_chain::{local_chain::CheckPoint, BlockId}; |
| 13 | +use bitcoin::{block::Header, Block, BlockHash, Transaction}; |
| 14 | +pub use bitcoincore_rpc; |
| 15 | +use bitcoincore_rpc::bitcoincore_rpc_json; |
| 16 | + |
| 17 | +/// A structure that emits data sourced from [`bitcoincore_rpc::Client`]. |
| 18 | +/// |
| 19 | +/// Refer to [module-level documentation] for more. |
| 20 | +/// |
| 21 | +/// [module-level documentation]: crate |
| 22 | +pub struct Emitter<'c, C> { |
| 23 | + client: &'c C, |
| 24 | + start_height: u32, |
| 25 | + |
| 26 | + /// The checkpoint of the last-emitted block that is in the best chain. If it is later found |
| 27 | + /// that the block is no longer in the best chain, it will be popped off from here. |
| 28 | + last_cp: Option<CheckPoint>, |
| 29 | + |
| 30 | + /// The block result returned from rpc of the last-emitted block. As this result contains the |
| 31 | + /// next block's block hash (which we use to fetch the next block), we set this to `None` |
| 32 | + /// whenever there are no more blocks, or the next block is no longer in the best chain. This |
| 33 | + /// gives us an opportunity to re-fetch this result. |
| 34 | + last_block: Option<bitcoincore_rpc_json::GetBlockResult>, |
| 35 | + |
| 36 | + /// The latest first-seen epoch of emitted mempool transactions. This is used to determine |
| 37 | + /// whether a mempool transaction is already emitted. |
| 38 | + last_mempool_time: usize, |
| 39 | + |
| 40 | + /// The last emitted block during our last mempool emission. This is used to determine whether |
| 41 | + /// there has been a reorg since our last mempool emission. |
| 42 | + last_mempool_tip: Option<u32>, |
| 43 | +} |
| 44 | + |
| 45 | +impl<'c, C: bitcoincore_rpc::RpcApi> Emitter<'c, C> { |
| 46 | + /// Construct a new [`Emitter`] with the given RPC `client` and `start_height`. |
| 47 | + /// |
| 48 | + /// `start_height` is the block height to start emitting blocks from. |
| 49 | + pub fn from_height(client: &'c C, start_height: u32) -> Self { |
| 50 | + Self { |
| 51 | + client, |
| 52 | + start_height, |
| 53 | + last_cp: None, |
| 54 | + last_block: None, |
| 55 | + last_mempool_time: 0, |
| 56 | + last_mempool_tip: None, |
| 57 | + } |
| 58 | + } |
| 59 | + |
| 60 | + /// Construct a new [`Emitter`] with the given RPC `client` and `checkpoint`. |
| 61 | + /// |
| 62 | + /// `checkpoint` is used to find the latest block which is still part of the best chain. The |
| 63 | + /// [`Emitter`] will emit blocks starting right above this block. |
| 64 | + pub fn from_checkpoint(client: &'c C, checkpoint: CheckPoint) -> Self { |
| 65 | + Self { |
| 66 | + client, |
| 67 | + start_height: 0, |
| 68 | + last_cp: Some(checkpoint), |
| 69 | + last_block: None, |
| 70 | + last_mempool_time: 0, |
| 71 | + last_mempool_tip: None, |
| 72 | + } |
| 73 | + } |
| 74 | + |
| 75 | + /// Emit mempool transactions, alongside their first-seen unix timestamps. |
| 76 | + /// |
| 77 | + /// This method emits each transaction only once, unless we cannot guarantee the transaction's |
| 78 | + /// ancestors are already emitted. |
| 79 | + /// |
| 80 | + /// To understand why, consider a receiver which filters transactions based on whether it |
| 81 | + /// alters the UTXO set of tracked script pubkeys. If an emitted mempool transaction spends a |
| 82 | + /// tracked UTXO which is confirmed at height `h`, but the receiver has only seen up to block |
| 83 | + /// of height `h-1`, we want to re-emit this transaction until the receiver has seen the block |
| 84 | + /// at height `h`. |
| 85 | + pub fn mempool(&mut self) -> Result<Vec<(Transaction, u64)>, bitcoincore_rpc::Error> { |
| 86 | + let client = self.client; |
| 87 | + |
| 88 | + // This is the emitted tip height during the last mempool emission. |
| 89 | + let prev_mempool_tip = self |
| 90 | + .last_mempool_tip |
| 91 | + // We use `start_height - 1` as we cannot guarantee that the block at |
| 92 | + // `start_height` has been emitted. |
| 93 | + .unwrap_or(self.start_height.saturating_sub(1)); |
| 94 | + |
| 95 | + // Mempool txs come with a timestamp of when the tx is introduced to the mempool. We keep |
| 96 | + // track of the latest mempool tx's timestamp to determine whether we have seen a tx |
| 97 | + // before. `prev_mempool_time` is the previous timestamp and `last_time` records what will |
| 98 | + // be the new latest timestamp. |
| 99 | + let prev_mempool_time = self.last_mempool_time; |
| 100 | + let mut latest_time = prev_mempool_time; |
| 101 | + |
| 102 | + let txs_to_emit = client |
| 103 | + .get_raw_mempool_verbose()? |
| 104 | + .into_iter() |
| 105 | + .filter_map({ |
| 106 | + let latest_time = &mut latest_time; |
| 107 | + move |(txid, tx_entry)| -> Option<Result<_, bitcoincore_rpc::Error>> { |
| 108 | + let tx_time = tx_entry.time as usize; |
| 109 | + if tx_time > *latest_time { |
| 110 | + *latest_time = tx_time; |
| 111 | + } |
| 112 | + |
| 113 | + // Avoid emitting transactions that are already emitted if we can guarantee |
| 114 | + // blocks containing ancestors are already emitted. The bitcoind rpc interface |
| 115 | + // provides us with the block height that the tx is introduced to the mempool. |
| 116 | + // If we have already emitted the block of height, we can assume that all |
| 117 | + // ancestor txs have been processed by the receiver. |
| 118 | + let is_already_emitted = tx_time <= prev_mempool_time; |
| 119 | + let is_within_height = tx_entry.height <= prev_mempool_tip as _; |
| 120 | + if is_already_emitted && is_within_height { |
| 121 | + return None; |
| 122 | + } |
| 123 | + |
| 124 | + let tx = match client.get_raw_transaction(&txid, None) { |
| 125 | + Ok(tx) => tx, |
| 126 | + // the tx is confirmed or evicted since `get_raw_mempool_verbose` |
| 127 | + Err(err) if err.is_not_found_error() => return None, |
| 128 | + Err(err) => return Some(Err(err)), |
| 129 | + }; |
| 130 | + |
| 131 | + Some(Ok((tx, tx_time as u64))) |
| 132 | + } |
| 133 | + }) |
| 134 | + .collect::<Result<Vec<_>, _>>()?; |
| 135 | + |
| 136 | + self.last_mempool_time = latest_time; |
| 137 | + self.last_mempool_tip = self.last_cp.as_ref().map(|cp| cp.height()); |
| 138 | + |
| 139 | + Ok(txs_to_emit) |
| 140 | + } |
| 141 | + |
| 142 | + /// Emit the next block height and header (if any). |
| 143 | + pub fn next_header(&mut self) -> Result<Option<(u32, Header)>, bitcoincore_rpc::Error> { |
| 144 | + poll(self, |hash| self.client.get_block_header(hash)) |
| 145 | + } |
| 146 | + |
| 147 | + /// Emit the next block height and block (if any). |
| 148 | + pub fn next_block(&mut self) -> Result<Option<(u32, Block)>, bitcoincore_rpc::Error> { |
| 149 | + poll(self, |hash| self.client.get_block(hash)) |
| 150 | + } |
| 151 | +} |
| 152 | + |
| 153 | +enum PollResponse { |
| 154 | + Block(bitcoincore_rpc_json::GetBlockResult), |
| 155 | + NoMoreBlocks, |
| 156 | + /// Fetched block is not in the best chain. |
| 157 | + BlockNotInBestChain, |
| 158 | + AgreementFound(bitcoincore_rpc_json::GetBlockResult, CheckPoint), |
| 159 | + AgreementPointNotFound, |
| 160 | +} |
| 161 | + |
| 162 | +fn poll_once<C>(emitter: &Emitter<C>) -> Result<PollResponse, bitcoincore_rpc::Error> |
| 163 | +where |
| 164 | + C: bitcoincore_rpc::RpcApi, |
| 165 | +{ |
| 166 | + let client = emitter.client; |
| 167 | + |
| 168 | + if let Some(last_res) = &emitter.last_block { |
| 169 | + assert!( |
| 170 | + emitter.last_cp.is_some(), |
| 171 | + "must not have block result without last cp" |
| 172 | + ); |
| 173 | + |
| 174 | + let next_hash = match last_res.nextblockhash { |
| 175 | + None => return Ok(PollResponse::NoMoreBlocks), |
| 176 | + Some(next_hash) => next_hash, |
| 177 | + }; |
| 178 | + |
| 179 | + let res = client.get_block_info(&next_hash)?; |
| 180 | + if res.confirmations < 0 { |
| 181 | + return Ok(PollResponse::BlockNotInBestChain); |
| 182 | + } |
| 183 | + return Ok(PollResponse::Block(res)); |
| 184 | + } |
| 185 | + |
| 186 | + if emitter.last_cp.is_none() { |
| 187 | + let hash = client.get_block_hash(emitter.start_height as _)?; |
| 188 | + |
| 189 | + let res = client.get_block_info(&hash)?; |
| 190 | + if res.confirmations < 0 { |
| 191 | + return Ok(PollResponse::BlockNotInBestChain); |
| 192 | + } |
| 193 | + return Ok(PollResponse::Block(res)); |
| 194 | + } |
| 195 | + |
| 196 | + for cp in emitter.last_cp.iter().flat_map(CheckPoint::iter) { |
| 197 | + let res = client.get_block_info(&cp.hash())?; |
| 198 | + if res.confirmations < 0 { |
| 199 | + // block is not in best chain |
| 200 | + continue; |
| 201 | + } |
| 202 | + |
| 203 | + // agreement point found |
| 204 | + return Ok(PollResponse::AgreementFound(res, cp)); |
| 205 | + } |
| 206 | + |
| 207 | + Ok(PollResponse::AgreementPointNotFound) |
| 208 | +} |
| 209 | + |
| 210 | +fn poll<C, V, F>( |
| 211 | + emitter: &mut Emitter<C>, |
| 212 | + get_item: F, |
| 213 | +) -> Result<Option<(u32, V)>, bitcoincore_rpc::Error> |
| 214 | +where |
| 215 | + C: bitcoincore_rpc::RpcApi, |
| 216 | + F: Fn(&BlockHash) -> Result<V, bitcoincore_rpc::Error>, |
| 217 | +{ |
| 218 | + loop { |
| 219 | + match poll_once(emitter)? { |
| 220 | + PollResponse::Block(res) => { |
| 221 | + let height = res.height as u32; |
| 222 | + let hash = res.hash; |
| 223 | + let item = get_item(&hash)?; |
| 224 | + |
| 225 | + let this_id = BlockId { height, hash }; |
| 226 | + let prev_id = res.previousblockhash.map(|prev_hash| BlockId { |
| 227 | + height: height - 1, |
| 228 | + hash: prev_hash, |
| 229 | + }); |
| 230 | + |
| 231 | + match (&mut emitter.last_cp, prev_id) { |
| 232 | + (Some(cp), _) => *cp = cp.clone().push(this_id).expect("must push"), |
| 233 | + (last_cp, None) => *last_cp = Some(CheckPoint::new(this_id)), |
| 234 | + // When the receiver constructs a local_chain update from a block, the previous |
| 235 | + // checkpoint is also included in the update. We need to reflect this state in |
| 236 | + // `Emitter::last_cp` as well. |
| 237 | + (last_cp, Some(prev_id)) => { |
| 238 | + *last_cp = Some(CheckPoint::new(prev_id).push(this_id).expect("must push")) |
| 239 | + } |
| 240 | + } |
| 241 | + |
| 242 | + emitter.last_block = Some(res); |
| 243 | + |
| 244 | + return Ok(Some((height, item))); |
| 245 | + } |
| 246 | + PollResponse::NoMoreBlocks => { |
| 247 | + emitter.last_block = None; |
| 248 | + return Ok(None); |
| 249 | + } |
| 250 | + PollResponse::BlockNotInBestChain => { |
| 251 | + emitter.last_block = None; |
| 252 | + continue; |
| 253 | + } |
| 254 | + PollResponse::AgreementFound(res, cp) => { |
| 255 | + let agreement_h = res.height as u32; |
| 256 | + |
| 257 | + // get rid of evicted blocks |
| 258 | + emitter.last_cp = Some(cp); |
| 259 | + |
| 260 | + // The tip during the last mempool emission needs to in the best chain, we reduce |
| 261 | + // it if it is not. |
| 262 | + if let Some(h) = emitter.last_mempool_tip.as_mut() { |
| 263 | + if *h > agreement_h { |
| 264 | + *h = agreement_h; |
| 265 | + } |
| 266 | + } |
| 267 | + emitter.last_block = Some(res); |
| 268 | + continue; |
| 269 | + } |
| 270 | + PollResponse::AgreementPointNotFound => { |
| 271 | + // We want to clear `last_cp` and set `start_height` to the first checkpoint's |
| 272 | + // height. This way, the first checkpoint in `LocalChain` can be replaced. |
| 273 | + if let Some(last_cp) = emitter.last_cp.take() { |
| 274 | + emitter.start_height = last_cp.height(); |
| 275 | + } |
| 276 | + emitter.last_block = None; |
| 277 | + continue; |
| 278 | + } |
| 279 | + } |
| 280 | + } |
| 281 | +} |
| 282 | + |
| 283 | +/// Extends [`bitcoincore_rpc::Error`]. |
| 284 | +pub trait BitcoindRpcErrorExt { |
| 285 | + /// Returns whether the error is a "not found" error. |
| 286 | + /// |
| 287 | + /// This is useful since [`Emitter`] emits [`Result<_, bitcoincore_rpc::Error>`]s as |
| 288 | + /// [`Iterator::Item`]. |
| 289 | + fn is_not_found_error(&self) -> bool; |
| 290 | +} |
| 291 | + |
| 292 | +impl BitcoindRpcErrorExt for bitcoincore_rpc::Error { |
| 293 | + fn is_not_found_error(&self) -> bool { |
| 294 | + if let bitcoincore_rpc::Error::JsonRpc(bitcoincore_rpc::jsonrpc::Error::Rpc(rpc_err)) = self |
| 295 | + { |
| 296 | + rpc_err.code == -5 |
| 297 | + } else { |
| 298 | + false |
| 299 | + } |
| 300 | + } |
| 301 | +} |
0 commit comments