use crate::chain::Chain;
use crate::chain::NewHeader;
use crate::chaindef::Block;
use crate::chaindef::BlockHash;
use crate::chaindef::BlockHeader;
use crate::chaindef::OutPointHash;
use crate::chaindef::ScriptHash;
use crate::chaindef::{Transaction, TxIn};
use crate::config::Config;
use crate::daemon::p2pconnection::duration_to_seconds;
use crate::daemon::Daemon;
use crate::encode::compute_outpoint_hash;
use crate::indexes::headerindex::HeaderRow;
use crate::indexes::heightindex::HeightIndexRow;
use crate::indexes::inputindex::InputIndexRow;
use crate::indexes::outputindex::OutputIndexRow;
use crate::indexes::scripthashindex::OutputFlags;
use crate::indexes::scripthashindex::ScriptHashIndexRow;
use crate::indexes::unspentindex::UnspentIndexRow;
use crate::indexes::DBRow;
use crate::indexes::{
outputtokenindex::OutputTokenIndexRow, tokenoutputindex::TokenOutputIndexRow,
};
use crate::metrics;
use crate::metrics::Metrics;
use crate::undo::{BlockUndoer, StoreBlockUndoer};
use crate::signal::Waiter;
use crate::store;
use crate::store::db_decode;
use crate::store::db_encode;
use crate::store::DBContents;
use crate::store::DBStore;
use crate::store::Row;
use crate::store::METADATA_LAST_INDEXED_BLOCK;
use crate::store::META_CF;
use crate::thread;
use crate::util;
use crate::writebatch::SpentUpdateBatch;
use crate::writebatch::WriteBatch;
use anyhow::Context;
use anyhow::Result;
use bitcoin_hashes::hex::ToHex;
use bitcoin_hashes::Hash;
use rayon::prelude::*;
use std::collections::HashMap;
use std::collections::HashSet;
use std::sync::Arc;
use std::time::Duration;
use std::time::Instant;
use tokio::sync::Mutex;
struct Stats {
update_duration: prometheus::HistogramVec,
blocks: prometheus::IntCounter,
txns: prometheus::IntCounter,
vsize: prometheus::IntCounter,
height: prometheus::IntGauge,
}
impl Stats {
fn new(metrics: &Metrics) -> Stats {
Stats {
update_duration: metrics.histogram_vec(
"rostrum_update_duration",
"Indexing update duration (in seconds)",
&["step"],
metrics::default_duration_buckets(),
),
blocks: metrics.counter_int(prometheus::Opts::new(
"rostrum_index_blocks",
"# of indexed blocks",
)),
txns: metrics.counter_int(prometheus::Opts::new(
"rostrum_index_txns",
"# of indexed transactions",
)),
vsize: metrics.counter_int(prometheus::Opts::new(
"rostrum_index_vsize",
"# of indexed vbytes",
)),
height: metrics.gauge_int(prometheus::Opts::new(
"rostrum_index_height",
"Last indexed block's height",
)),
}
}
fn update(&self, block: &Block, height: usize) {
self.blocks.inc();
self.txns.inc_by(block.txdata.len() as u64);
for tx in &block.txdata {
self.vsize.inc_by(tx.size() as u64);
}
self.update_height(height);
}
fn update_height(&self, height: usize) {
self.height.set(height as i64);
}
fn start_timer(&self, step: &str) -> prometheus::HistogramTimer {
self.update_duration
.with_label_values(&[step])
.start_timer()
}
async fn observe_chain(&self, chain: &Chain) {
self.height.set(chain.height().await as i64)
}
}
#[cfg(bch)]
#[inline]
pub(crate) fn is_coinbase_input(input: &TxIn) -> bool {
use bitcoincash::Txid;
let null_hash: Txid = Txid::all_zeros();
input.previous_output.txid == null_hash
}
#[cfg(nexa)]
#[inline]
pub(crate) fn is_coinbase_input(_input: &TxIn) -> bool {
false
}
#[cfg(bch)]
#[inline]
fn get_outpoint_hash(input: &TxIn) -> [u8; 32] {
compute_outpoint_hash(&input.previous_output.txid, input.previous_output.vout).into_inner()
}
#[cfg(nexa)]
#[inline]
fn get_outpoint_hash(input: &TxIn) -> [u8; 32] {
input.previous_output.hash.into_inner()
}
#[cfg(nexa)]
fn index_outputs(writebatch: &WriteBatch, tx: &Transaction, height: u32) {
let txid = tx.txid();
let rows = tx.output.par_iter().enumerate().map(move |(i, output)| {
OutputIndexRow::new(
&txid,
&compute_outpoint_hash(&tx.txidem(), i as u32),
output.value,
i as u32,
height,
)
.to_row()
});
writebatch.insert(OutputIndexRow::CF, rows);
}
#[cfg(bch)]
fn index_outputs(writebatch: &WriteBatch, tx: &Transaction, height: u32) {
let txid = tx.txid();
let rows = tx.output.par_iter().enumerate().map(move |(i, output)| {
OutputIndexRow::new(
&txid,
&compute_outpoint_hash(&txid, i as u32),
output.value,
i as u32,
height,
)
.to_row()
});
writebatch.insert(OutputIndexRow::CF, rows);
}
#[cfg(nexa)]
fn index_scriptsig(writebatch: &WriteBatch, tx: &Transaction, height: u32) {
use crate::indexes::outtoscriptindex::OutToScripthashIndex;
let rows: Vec<(Row, Row)> = tx
.output
.par_iter()
.enumerate()
.map(move |(index, output)| {
let scripthash = ScriptHash::normalized_from_txout(output);
let oph = compute_outpoint_hash(&tx.txidem(), index as u32);
let scripthash_to_out = ScriptHashIndexRow::new_funding(
&scripthash,
&oph,
if output.has_token() {
OutputFlags::FundingHasTokens
} else {
OutputFlags::FundingNone
},
tx.txid().into_inner(),
index as u32,
height,
)
.to_row();
let out_to_scripthash =
OutToScripthashIndex::new(scripthash, oph, output.has_token()).to_row();
(scripthash_to_out, out_to_scripthash)
})
.collect();
let (from_scripthash, to_scripthash): (Vec<Row>, Vec<Row>) = rows.into_iter().unzip();
writebatch.insert(ScriptHashIndexRow::CF, from_scripthash);
writebatch.insert(OutToScripthashIndex::CF, to_scripthash);
}
#[cfg(bch)]
fn index_scriptsig(writebatch: &WriteBatch, tx: &Transaction, height: u32) {
use crate::indexes::outtoscriptindex::OutToScripthashIndex;
let txid = tx.txid();
let rows = tx
.output
.par_iter()
.enumerate()
.map(move |(index, output)| {
let scripthash = ScriptHash::from_script(&output.script_pubkey);
let oph = compute_outpoint_hash(&txid, index as u32);
let scripthash_to_out = ScriptHashIndexRow::new_funding(
&scripthash,
&oph,
if output.has_token() {
OutputFlags::FundingHasTokens
} else {
OutputFlags::FundingNone
},
tx.txid().into_inner(),
index as u32,
height,
)
.to_row();
let out_to_scripthash =
OutToScripthashIndex::new(scripthash, oph, output.has_token()).to_row();
(scripthash_to_out, out_to_scripthash)
})
.collect::<Vec<_>>();
let (from_scripthash, to_scripthash): (Vec<Row>, Vec<Row>) = rows.into_iter().unzip();
writebatch.insert(ScriptHashIndexRow::CF, from_scripthash);
writebatch.insert(OutToScripthashIndex::CF, to_scripthash);
}
#[cfg(nexa)]
fn index_token_outputs(writebatch: &WriteBatch, tx: &Transaction) {
use crate::nexa::token::parse_token_from_scriptpubkey;
tx.output.par_iter().enumerate().for_each(move |(i, out)| {
if !out.has_token() {
return;
}
if let Some((token, amount)) = parse_token_from_scriptpubkey(&out.script_pubkey) {
let outpoint_hash = &compute_outpoint_hash(&tx.txidem(), i as u32);
let out_token_row =
OutputTokenIndexRow::new(outpoint_hash, token.clone(), amount).to_row();
let token_out_row = TokenOutputIndexRow::new(token, outpoint_hash).to_row();
writebatch.insert(OutputTokenIndexRow::CF, rayon::iter::once(out_token_row));
writebatch.insert(TokenOutputIndexRow::CF, rayon::iter::once(token_out_row));
};
});
}
#[cfg(bch)]
fn index_token_outputs(writebatch: &WriteBatch, tx: &Transaction) {
tx.output.par_iter().enumerate().for_each(move |(i, out)| {
if let Some(t) = out.token.as_ref() {
let outpoint_hash = compute_outpoint_hash(&tx.txid(), i as u32);
let out_token_row = OutputTokenIndexRow::new(
&outpoint_hash,
t.id.clone(),
t.commitment.clone(),
t.amount,
)
.to_row();
let token_out_row =
TokenOutputIndexRow::new(t.id, t.commitment.clone(), &outpoint_hash).to_row();
writebatch.insert(OutputTokenIndexRow::CF, rayon::iter::once(out_token_row));
writebatch.insert(TokenOutputIndexRow::CF, rayon::iter::once(token_out_row));
}
});
}
fn index_unspent(
write: &WriteBatch,
spentupdatebatch: &SpentUpdateBatch,
tx: &Transaction,
height: u32,
) {
let txid = tx.txid();
for (vin, input) in tx.input.iter().enumerate() {
if is_coinbase_input(input) {
continue;
}
let outpoint = OutPointHash::from_inner(get_outpoint_hash(input));
spentupdatebatch.insert_spending_metadata(outpoint, txid, vin as u32, height, None);
}
#[cfg(nexa)]
let txid_or_txidem = tx.txidem();
#[cfg(bch)]
let txid_or_txidem = tx.txid();
let rows = tx.output.par_iter().enumerate().map(|(index, out)| {
let scripthash = ScriptHash::normalized_from_txout(out);
let oph = compute_outpoint_hash(&txid_or_txidem, index as u32);
UnspentIndexRow::new(&scripthash, &oph, out.has_token()).to_row()
});
write.insert(UnspentIndexRow::CF, rows);
}
pub fn index_transaction(
writebatch: &WriteBatch,
spentupdatebatch: &SpentUpdateBatch,
txn: &Transaction,
height: usize,
) {
let txid = txn.txid();
let inputs = txn
.input
.par_iter()
.enumerate()
.filter_map(move |(index, input)| {
if is_coinbase_input(input) {
None
} else {
let outpointhash = get_outpoint_hash(input);
Some(InputIndexRow::new(outpointhash, txid, index as u32).to_row())
}
});
writebatch.insert(InputIndexRow::CF, inputs);
index_outputs(writebatch, txn, height as u32);
index_unspent(writebatch, spentupdatebatch, txn, height as u32);
index_scriptsig(writebatch, txn, height as u32);
index_token_outputs(writebatch, txn);
writebatch.insert(
HeightIndexRow::CF,
rayon::iter::once(HeightIndexRow::new(&txid, height as u32).to_row()),
);
}
pub(crate) fn index_block(
writebatch: &WriteBatch,
spentupdatebatch: &SpentUpdateBatch,
block: &Block,
height: usize,
) {
let blockhash = block.block_hash();
let header_row = HeaderRow::new(&block.header).to_row();
block
.txdata
.par_iter()
.for_each(move |txn| index_transaction(writebatch, spentupdatebatch, txn, height));
writebatch.insert(HeaderRow::CF, rayon::iter::once(header_row));
writebatch.insert(META_CF, rayon::iter::once(last_indexed_block(&blockhash)));
}
pub(crate) fn last_indexed_block(blockhash: &BlockHash) -> Row {
Row {
key: METADATA_LAST_INDEXED_BLOCK.to_vec().into_boxed_slice(),
value: db_encode(blockhash.as_inner()).unwrap().into_boxed_slice(),
}
}
pub(crate) fn get_last_indexed_block(store: &DBStore) -> Option<BlockHash> {
assert!(store.contents == DBContents::ConfirmedIndex);
store
.get_blocking(META_CF, METADATA_LAST_INDEXED_BLOCK)
.map(|row| BlockHash::from_inner(db_decode(&row).expect("db error fetching blockhash")))
}
async fn find_header_recovery(
store: &DBStore,
daemon: &Daemon,
genesis: &BlockHash,
) -> Result<(Vec<BlockHeader>, HashSet<BlockHash>)> {
info!("Searching for a reacovery point");
let (_key, max_height) = store
.get(META_CF, METADATA_LAST_INDEXED_BLOCK.to_vec())
.await;
let max_height: BlockHash = match max_height {
Some(row) => BlockHash::from_inner(db_decode(&row).expect("db error fetching blockhash")),
None => bail!("Unable to find starting point for recovery"),
};
let headers: Vec<BlockHeader> = HeaderRow::read_headers_unsorted(store).await;
let headers: HashMap<BlockHash, BlockHeader> = headers
.into_par_iter()
.map(|h| (h.block_hash(), h))
.collect();
info!(
"recovery: Our tip is {}; our db has {} headers",
max_height.to_hex(),
headers.len()
);
let mut max_height = daemon.getblockheader(&max_height).await?.1;
let mut min_height: u64 = 0;
let mut highest_match: Option<u64> = None;
let mut highest_match_hash: Option<BlockHash> = None;
let has_all_parent = |header: &BlockHeader| {
let mut parent = header;
loop {
if parent.block_hash().eq(genesis) {
return true;
}
parent = match headers.get(&parent.prev_blockhash) {
Some(h) => h,
None => {
return false;
}
}
}
};
loop {
let mid = (min_height + max_height) / 2;
info!(
"recovery: searching for header at height {} (min: {}, max: {})",
mid, min_height, max_height
);
if mid == 0 {
break;
}
let header_at_mid: BlockHeader =
match daemon.getblockheaders(&[mid as usize]).await?.first() {
Some(h) => h.0.clone(),
None => bail!("Failed to get header from daemon"),
};
if headers.contains_key(&header_at_mid.block_hash()) && has_all_parent(&header_at_mid) {
highest_match = Some(mid);
highest_match_hash = Some(header_at_mid.block_hash());
min_height = mid + 1;
} else {
max_height = mid - 1;
}
if min_height > max_height {
break;
}
}
if highest_match.is_none() {
bail!("Found no recovery point")
};
info!(
"recovery: Highest match between us and daemon is {} ({})",
highest_match.unwrap(),
highest_match_hash.unwrap().to_hex()
);
let mut new_headers: Vec<BlockHeader> = vec![];
let mut parent = headers.get(&highest_match_hash.unwrap()).unwrap();
loop {
new_headers.push(parent.clone());
if parent.block_hash().eq(genesis) {
break;
}
parent = headers.get(&parent.prev_blockhash).unwrap();
}
new_headers.reverse();
let new_headers_hashes: HashSet<BlockHash> =
new_headers.iter().map(|h| h.block_hash()).collect();
let old_headers_hashes: HashSet<BlockHash> = headers.keys().cloned().collect();
let mut undo_blocks: HashSet<BlockHash> = old_headers_hashes
.difference(&new_headers_hashes)
.cloned()
.collect();
drop(new_headers_hashes);
drop(old_headers_hashes);
for _ in 0..100 {
if new_headers.len() == 1 {
break;
}
match new_headers.pop() {
Some(h) => {
undo_blocks.insert(h.block_hash());
}
None => break,
}
}
info!(
"recovery: New headers: {}, undo blocks: {}",
new_headers.len(),
undo_blocks.len()
);
Ok((new_headers, undo_blocks))
}
async fn read_indexed_headers(store: &DBStore) -> Result<Vec<BlockHeader>> {
let (latest_blockhash, headers) = match store.get_blocking(META_CF, METADATA_LAST_INDEXED_BLOCK)
{
Some(row) => {
let tip: BlockHash =
BlockHash::from_inner(db_decode(&row).expect("db error fetching blockhash"));
let headers = HeaderRow::headers_up_to_tip(store, &tip)
.await
.context("Headers missing from database")?;
(tip, headers)
}
None => (BlockHash::all_zeros(), vec![]),
};
info!("Latest indexed blockhash: {}", latest_blockhash.to_hex());
assert_eq!(
headers
.first()
.map(|h| h.prev_blockhash)
.unwrap_or_else(BlockHash::all_zeros),
BlockHash::all_zeros()
);
assert_eq!(
headers
.last()
.map(BlockHeader::block_hash)
.unwrap_or_else(BlockHash::all_zeros),
latest_blockhash
);
Ok(headers)
}
pub struct Index {
store: Arc<DBStore>,
daemon: Arc<Daemon>,
stats: Stats,
batch_size: usize,
chain: Chain,
low_memory: bool,
db_write_cache_entries: usize,
last_flush: Mutex<Instant>,
ignore_blocks_below: usize,
}
impl Index {
pub async fn load(
store: Arc<DBStore>,
mut chain: Chain,
daemon: Arc<Daemon>,
metrics: &Metrics,
config: &Config,
) -> Result<Self> {
let (headers, undo_blocks) = match read_indexed_headers(&store).await {
Ok(h) => (h, HashSet::default()),
Err(e) => {
warn!("Error loading headers; will attempt recovery: {}", e);
find_header_recovery(&store, &daemon, &chain.genesis_hash().await).await?
}
};
if !headers.is_empty() {
let tip = headers.last().unwrap().block_hash();
chain.load(headers, tip).await;
let block_undoer = StoreBlockUndoer::new(store.clone(), daemon.clone());
for u in undo_blocks {
let height = match daemon.getblockheader(&u).await {
Ok(header) => header.1,
Err(e) => {
bail!(
"Failed to undo block, failed to get header from daemon: {}",
e
);
}
};
if let Err(e) = block_undoer.undo_block(&u, height, &tip).await {
warn!("Failed to undo block: {}", e);
}
}
chain
.drop_last_headers(block_undoer, config.reindex_last_blocks as u64)
.await;
info!("Loaded {} headers from index", chain.height().await);
}
let stats = Stats::new(metrics);
stats.observe_chain(&chain).await;
Ok(Index {
store,
daemon,
stats,
batch_size: config.index_batch_size,
chain,
low_memory: config.low_memory,
db_write_cache_entries: config.db_write_cache_entries,
last_flush: Mutex::new(Instant::now()),
ignore_blocks_below: config.ignore_blocks_below,
})
}
pub fn chain(&self) -> &Chain {
&self.chain
}
pub async fn update(&self) -> Result<(Vec<BlockHeader>, BlockHeader)> {
let tip = self.daemon.getbestblockhash().await?;
let tip_height = self.daemon.getblockheader(&tip).await?.1;
let start = Instant::now();
let mut new_headers = self.daemon.get_new_headers(&self.chain, &tip).await?;
metrics::observe(
&self.stats.update_duration,
"new_headers",
duration_to_seconds(start.elapsed()),
);
let block_undoer = StoreBlockUndoer::new(self.store.clone(), self.daemon.clone());
if !new_headers.is_empty() {
let header_writebatch = WriteBatch::new();
for (header, _) in &new_headers {
let header_row = HeaderRow::new(header).to_row();
header_writebatch.insert(HeaderRow::CF, rayon::iter::once(header_row));
}
self.store.write_batch(&header_writebatch);
}
self.chain
.update(
block_undoer,
new_headers
.iter()
.map(|(h, height)| NewHeader::from_header_and_height(h.clone(), *height))
.collect(),
Some(tip_height),
)
.await;
let (blockhashes, indexed_headers): (Vec<BlockHash>, Vec<(BlockHeader, u64)>) =
if self.ignore_blocks_below > 0 {
let filtered: Vec<_> = new_headers
.iter()
.filter(|(_, height)| *height >= self.ignore_blocks_below as u64)
.cloned()
.collect();
let hashes: Vec<_> = filtered.iter().map(|(h, _)| h.block_hash()).collect();
(hashes, filtered)
} else {
let hashes: Vec<_> = new_headers.iter().map(|(h, _)| h.block_hash()).collect();
let indexed = std::mem::take(&mut new_headers);
(hashes, indexed)
};
let block_channel = util::Channel::<Option<Block>>::bounded(self.batch_size);
let block_sender = block_channel.sender();
let writer_channel = util::Channel::<Option<(WriteBatch, SpentUpdateBatch)>>::bounded(1);
let writer_sender = writer_channel.sender();
let daemon = self.daemon.clone();
let block_fetcher = thread::spawn_task("fetcher", async move {
for blockhash in blockhashes.iter() {
Waiter::shutdown_check()?;
block_sender
.send(Some(daemon.getblock(blockhash).await?))
.await
.context("failed sending blocks to be indexed")?;
}
block_sender
.send(None)
.await
.context("failed sending explicit end of stream")?;
Ok(())
});
let writer_store = Arc::clone(&self.store);
let writer = thread::spawn_task("index_writer", async move {
loop {
Waiter::shutdown_check()?;
let mut write_receiver = writer_channel.receiver().await;
let (write, spentupdate): (WriteBatch, SpentUpdateBatch) =
match tokio::time::timeout(Duration::from_secs(10), write_receiver.recv()).await
{
Ok(Some(Some((write, erase)))) => (write, erase),
Ok(Some(None)) => break,
Ok(None) =>
{
break
}
Err(_) =>
{
continue
}
};
writer_store.write_batch(&write);
writer_store.update_spends(&spentupdate);
}
Ok(())
});
let mut writebatch = WriteBatch::new();
let mut spentupdatebatch = SpentUpdateBatch::new();
loop {
Waiter::shutdown_check()?;
let mut block_receiver = block_channel.receiver().await;
let timer = self.stats.start_timer("fetch");
let block =
match tokio::time::timeout(Duration::from_secs(10), block_receiver.recv()).await {
Ok(Some(Some(block))) => block,
Ok(Some(None)) =>
{
break
}
Err(_) =>
{
continue
}
Ok(None) => break,
};
timer.observe_duration();
let blockhash = block.block_hash();
let height = self
.chain
.get_block_height(&blockhash)
.await
.unwrap_or_else(|| panic!("missing header for block {}", blockhash));
if height % 1000 == 0 && height != self.chain.height().await {
let chain_height = self.chain.height().await;
info!(
"Indexing block {height} out of {chain_height}. Progress: {:.2}%",
(height as f64 / chain_height as f64) * 100.
)
}
let timer = self.stats.start_timer("index");
index_block(&writebatch, &spentupdatebatch, &block, height as usize);
timer.observe_duration();
let timer = self.stats.start_timer("write");
if self.low_memory || writebatch.len() >= self.db_write_cache_entries {
if !self.low_memory {
debug!(
"Writing batch of {} entries to db at block {height}.",
self.db_write_cache_entries
);
}
writer_sender
.send(Some((writebatch, spentupdatebatch)))
.await
.context("failed to send batch for writing")?;
writebatch = WriteBatch::new();
spentupdatebatch = SpentUpdateBatch::new();
}
timer.observe_duration();
self.stats.update(&block, height as usize);
if self.last_flush.lock().await.elapsed() > Duration::from_secs(10 * 60) {
*self.last_flush.lock().await = Instant::now();
info!("Flushing index to disk...");
self.store.flush()?;
info!("Flushing done");
}
}
writer_sender
.send(Some((writebatch, spentupdatebatch)))
.await
.context("failed to send batch for writing")?;
writer_sender
.send(None)
.await
.context("failed to EOF to index writer")?;
let timer = self.stats.start_timer("flush");
timer.observe_duration();
if let Err(e) = block_fetcher.await {
warn!("failed to complete block fetcher: {}", e);
}
if let Err(e) = writer.await {
warn!("failed to complete index writer: {}", e);
}
let tip_header = self.chain.tip().await;
assert_eq!(&tip, &tip_header.block_hash());
self.stats.observe_chain(&self.chain).await;
Ok((
indexed_headers.into_iter().map(|(h, _)| h).collect(),
tip_header,
))
}
pub fn read_store(&self) -> &Arc<store::DBStore> {
&self.store
}
}