pub mod manager;
pub mod p2pconnection;
pub mod p2pmessage;
pub mod rpcconnection;
use crate::chain::Chain;
use crate::chaindef::BlockHash;
use crate::chaindef::Transaction;
use crate::config::Config;
use crate::daemon;
use crate::daemon::manager::BitcoindRPCConnectionManager;
use crate::errors::ConnectionError;
use crate::metrics;
use crate::signal::NetworkNotifier;
use bitcoin_hashes::Hash;
use bitcoincash::consensus::encode::{deserialize, serialize};
use bitcoincash::hash_types::Txid;
use bitcoincash::hashes::hex::FromHex;
use bitcoincash::network::constants::Network;
use serde_json::{from_value, Value};
use std::collections::HashSet;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
use crate::cache::BlockTxIDsCache;
use crate::chaindef::Block;
use crate::chaindef::BlockHeader;
use crate::metrics::Metrics;
#[cfg(nexa)]
use crate::nexa::hash_types::TxIdem;
use crate::rpc::encoding::{blockhash_to_hex, hex_to_blockhash, hex_to_txid, txid_to_hex};
use crate::signal::Waiter;
use anyhow::{Context, Result};
use self::manager::BitcoindP2PConnectionManager;
fn header_from_value(value: Value) -> Result<BlockHeader> {
let header_hex = value
.as_str()
.context(format!("non-string header: {}", value))?;
let header_bytes = hex::decode(header_hex).context("non-hex header")?;
deserialize(&header_bytes).context(format!("failed to parse header {}", header_hex))
}
fn block_from_value(value: Value) -> Result<Block> {
let block_hex = value.as_str().context("non-string block")?;
let block_bytes = hex::decode(block_hex).context("non-hex block")?;
deserialize(&block_bytes).context(format!("failed to parse block {}", block_hex))
}
fn tx_from_value(value: Value) -> Result<Transaction> {
let tx_hex = value.as_str().context("non-string tx")?;
let tx_bytes = hex::decode(tx_hex).context("non-hex tx")?;
deserialize(&tx_bytes).context(format!("failed to parse tx {}", tx_hex))
}
#[derive(Serialize, Deserialize, Debug)]
struct BlockchainInfo {
chain: String,
blocks: u32,
headers: u32,
verificationprogress: f64,
bestblockhash: String,
pruned: bool,
initialblockdownload: bool,
}
#[derive(Serialize, Deserialize, Debug)]
struct NetworkInfo {
version: u64,
subversion: String,
relayfee: f64, }
pub trait CookieGetter: Send + Sync {
fn get(&self) -> Result<Vec<u8>>;
}
pub struct Daemon {
daemon_dir: PathBuf,
network: Network,
rpc_pool: bb8::Pool<BitcoindRPCConnectionManager>,
p2p_pool: bb8::Pool<BitcoindP2PConnectionManager>,
blocktxids_cache: Arc<BlockTxIDsCache>,
latency: prometheus::HistogramVec,
size: prometheus::HistogramVec,
}
impl Daemon {
#[allow(clippy::too_many_arguments)]
pub async fn new(
config: &Config,
network_notifier: Arc<NetworkNotifier>,
metrics: &Metrics,
) -> Result<Daemon> {
let rpc_manager = daemon::manager::BitcoindRPCConnectionManager::new(
config.daemon_rpc_addr,
Arc::clone(&config.cookie_getter),
);
let p2p_manager = daemon::manager::BitcoindP2PConnectionManager::new(
config.network_type,
config.daemon_p2p_addr,
network_notifier,
metrics,
);
let pool_size = config.daemon_rpc_connections;
let rpc_pool = bb8::Pool::builder()
.max_size(pool_size)
.min_idle(Some(pool_size))
.test_on_check_out(false)
.build(rpc_manager)
.await?;
let p2p_pool = bb8::Pool::builder()
.max_size(1)
.min_idle(Some(1))
.test_on_check_out(false)
.build_unchecked(p2p_manager);
let blocktxids_cache = Arc::new(BlockTxIDsCache::new(
config.blocktxids_cache_size as u64,
metrics,
));
let daemon = Daemon {
daemon_dir: config.daemon_dir.to_path_buf(),
network: config.network_type,
rpc_pool,
p2p_pool,
blocktxids_cache,
latency: metrics.histogram_vec(
"rostrum_daemon_rpc",
"Bitcoind RPC latency (in seconds)",
&["method"],
metrics::default_duration_buckets(),
),
size: metrics.histogram_vec(
"rostrum_daemon_bytes",
"Bitcoind RPC size (in bytes)",
&["method", "dir"],
metrics::default_size_buckets(),
),
};
let network_info = daemon.getnetworkinfo().await?;
info!("{:?}", network_info);
let blockchain_info = daemon.getblockchaininfo().await?;
info!("{:?}", blockchain_info);
if blockchain_info.pruned {
bail!("pruned node is not supported (use '-prune=0' bitcoind flag)".to_owned())
}
Ok(daemon)
}
pub fn reconnect(&self) -> Result<Daemon> {
Ok(Daemon {
daemon_dir: self.daemon_dir.clone(),
network: self.network,
rpc_pool: self.rpc_pool.clone(),
p2p_pool: self.p2p_pool.clone(),
blocktxids_cache: Arc::clone(&self.blocktxids_cache),
latency: self.latency.clone(),
size: self.size.clone(),
})
}
async fn retry_request_batch(
&self,
method: &str,
params_list: &[Option<Value>],
) -> Result<Vec<Value>> {
let mut connection = self.rpc_pool.get().await?;
loop {
Waiter::shutdown_check()?;
match connection.multi_request(method, params_list).await {
Err(e) => {
if let Some(e) = e.downcast_ref::<ConnectionError>() {
warn!("Full node connection broken: {}", e.msg);
drop(connection);
Waiter::wait_or_shutdown(Duration::from_secs(1)).await?;
connection = self.rpc_pool.get().await?;
continue;
}
return Err(e);
}
Ok(result) => return Ok(result),
}
}
}
async fn retry_request(&self, method: &str, params: &Option<Value>) -> Result<Value> {
let mut connection = self.rpc_pool.get().await?;
loop {
match connection.request(method, params).await {
Err(e) => {
if let Some(e) = e.downcast_ref::<ConnectionError>() {
warn!("Full node connection broken: {}", e.msg);
drop(connection);
Waiter::wait_or_shutdown(Duration::from_secs(1)).await?;
connection = self.rpc_pool.get().await?;
continue;
}
return Err(e);
}
Ok(result) => return Ok(result),
}
}
}
pub async fn request(&self, method: &str, params: &Option<Value>) -> Result<Value> {
self.retry_request(method, params).await
}
pub async fn multi_request(
&self,
method: &str,
params_list: &[Option<Value>],
) -> Result<Vec<Value>> {
self.retry_request_batch(method, params_list).await
}
async fn getblockchaininfo(&self) -> Result<BlockchainInfo> {
let info: Value = self.request("getblockchaininfo", &None).await?;
from_value(info).context("invalid blockchain info")
}
async fn getnetworkinfo(&self) -> Result<NetworkInfo> {
let info: Value = self.request("getnetworkinfo", &None).await?;
from_value(info).context("invalid network info")
}
pub async fn get_subversion(&self) -> Result<String> {
Ok(self.getnetworkinfo().await?.subversion)
}
pub async fn get_relayfee(&self) -> Result<f64> {
Ok(self.getnetworkinfo().await?.relayfee)
}
pub async fn getbestblockhash(&self) -> Result<BlockHash> {
hex_to_blockhash(&self.request("getbestblockhash", &None).await?)
.context("invalid blockhash")
}
#[cfg(nexa)]
pub async fn getblockheader(&self, blockhash: &BlockHash) -> Result<(BlockHeader, u64)> {
let header = header_from_value(
self.request(
"getblockheader",
&Some(json!([
blockhash_to_hex(blockhash),
false,
])),
)
.await?,
)?;
let height = header.height();
Ok((header, height))
}
#[cfg(bch)]
pub async fn getblockheader(&self, blockhash: &BlockHash) -> Result<(BlockHeader, u64)> {
use bitcoincash::TxMerkleNode;
let h = self
.request(
"getblockheader",
&Some(json!([blockhash_to_hex(blockhash), true,])),
)
.await?;
let version = h
.get("version")
.context("version missing")?
.as_i64()
.context("version is not numeric")? as i32;
let prev_blockhash = BlockHash::from_hex(
h.get("previousblockhash")
.unwrap_or(&json!(BlockHash::all_zeros()))
.as_str()
.context("previousblockhash not a string")?,
)
.context("invalid prev_blockhash")?;
let merkle_root = TxMerkleNode::from_hex(
h.get("merkleroot")
.context("merkleroot missing")?
.as_str()
.context("merkleroot not a hex")?,
)
.context("invalid merkleroot")?;
let time = h
.get("time")
.context("time missing")?
.as_u64()
.context("time is not numeric")? as u32;
let bits = h
.get("bits")
.context("bits missing")?
.as_str()
.context("bits is not a string")?;
let bits = u32::from_str_radix(bits, 16).context("invalid bits")?;
let nonce = h
.get("nonce")
.context("nonce missing")?
.as_u64()
.context("nonce is not numeric")? as u32;
let header = BlockHeader {
version,
prev_blockhash,
merkle_root,
time,
bits,
nonce,
};
debug_assert!(&header.block_hash() == blockhash);
let height = h
.get("height")
.context("height missing for header")?
.as_u64()
.context("height was not numerical")?;
Ok((header, height))
}
pub async fn getblockheaders(&self, heights: &[usize]) -> Result<Vec<(BlockHeader, u64)>> {
let heights_json: Vec<Option<Value>> = heights
.iter()
.map(|height| Some(json!([height, false])))
.collect();
let result = self
.multi_request("getblockheader", &heights_json)
.await?
.into_iter()
.zip(heights.iter())
.map(|(response, height)| {
let header = header_from_value(response)?;
Ok((header, *height as u64))
})
.collect();
result
}
pub async fn getblock(&self, blockhash: &BlockHash) -> Result<Block> {
let block = block_from_value(
self.request(
"getblock",
&Some(json!([blockhash_to_hex(blockhash), 0,])),
)
.await?,
)?;
assert_eq!(block.block_hash(), *blockhash);
Ok(block)
}
async fn load_blocktxids(&self, blockhash: &BlockHash) -> Result<Vec<Txid>> {
let block = self
.request(
"getblock",
&Some(json!([blockhash_to_hex(blockhash), 1])),
)
.await?;
block
.get("tx")
.or_else(|| block.get("txid")) .context("block missing txids")?
.as_array()
.context("invalid block txids")?
.iter()
.map(hex_to_txid)
.collect::<Result<Vec<Txid>>>()
}
pub async fn getblocktxids(&self, blockhash: &BlockHash) -> Result<Vec<Txid>> {
self.blocktxids_cache
.get_or_else(
blockhash,
async move { self.load_blocktxids(blockhash).await },
)
.await
}
pub async fn gettransaction(
&self,
txid: &Txid,
blockhash: Option<BlockHash>,
) -> Result<Transaction> {
let tx = self
.gettransaction_raw(txid, blockhash.as_ref(), false)
.await?;
tx_from_value(tx)
}
pub async fn gettransaction_raw(
&self,
txid: &Txid,
blockhash: Option<&BlockHash>,
verbose: bool,
) -> Result<Value> {
let txid = txid_to_hex(txid);
if let Some(h) = blockhash {
return self
.request(
"getrawtransaction",
&Some(json!([txid, verbose, blockhash_to_hex(h)])),
)
.await;
}
self.request("getrawtransaction", &Some(json!([txid])))
.await
}
fn txids_to_hashset(&self, txids: Value) -> Result<HashSet<Txid>> {
let mut result = HashSet::new();
for value in txids.as_array().context("non-array result")? {
result.insert(hex_to_txid(value).context("invalid txid")?);
}
Ok(result)
}
#[cfg(nexa)]
pub async fn getmempooltxids(&self) -> Result<HashSet<Txid>> {
let txids: Value = self
.request(
"getrawtxpool",
&Some(json!([ false, "id"])),
)
.await?;
self.txids_to_hashset(txids)
}
#[cfg(bch)]
pub async fn getmempooltxids(&self) -> Result<HashSet<Txid>> {
let txids: Value = self
.request("getrawmempool", &Some(json!([ false])))
.await?;
self.txids_to_hashset(txids)
}
#[cfg(bch)]
pub async fn broadcast(&self, tx: &Transaction) -> Result<Txid> {
let tx = hex::encode(serialize(tx));
let txid = self
.request("sendrawtransaction", &Some(json!([tx])))
.await?;
Txid::from_hex(txid.as_str().context("non-string txid")?).context("failed to parse txid")
}
#[cfg(nexa)]
pub async fn broadcast(&self, tx: &Transaction) -> Result<TxIdem> {
let tx = hex::encode(serialize(tx));
let txidem = self
.request("sendrawtransaction", &Some(json!([tx])))
.await?;
TxIdem::from_hex(txidem.as_str().context("non-string txidem")?)
.context("failed to parse txidem")
}
pub async fn broadcast_p2p(&self, tx: Transaction) -> Result<()> {
self.p2p_pool.get().await?.send_tx(tx).await
}
async fn get_all_headers(&self, tip: &BlockHash) -> Result<Vec<(BlockHeader, u64)>> {
let info: Value = self
.request("getblockheader", &Some(json!([blockhash_to_hex(tip)])))
.await?;
let tip_height = info
.get("height")
.expect("missing height")
.as_u64()
.expect("non-numeric height") as usize;
let all_heights: Vec<usize> = (0..=tip_height).collect();
let chunk_size = 100_000;
let mut result = vec![];
let null_hash = BlockHash::all_zeros();
for heights in all_heights.chunks(chunk_size) {
Waiter::shutdown_check()?;
info!("Downloading {} block headers", heights.len());
let mut headers = self.getblockheaders(heights).await?;
assert!(headers.len() == heights.len());
result.append(&mut headers);
}
let mut blockhash = null_hash;
let mut prev_header: Option<BlockHeader> = None;
#[allow(clippy::clone_on_copy)]
for (header, _height) in &result {
assert_eq!(
header.prev_blockhash,
blockhash,
"Prev block hex: {}, this block hex: {}",
hex::encode(serialize(&prev_header.unwrap())),
hex::encode(serialize(&header)),
);
blockhash = header.block_hash();
prev_header = Some(header.clone());
}
assert_eq!(blockhash, *tip);
Ok(result)
}
pub async fn get_new_headers(
&self,
chain: &Chain,
bestblockhash: &BlockHash,
) -> Result<Vec<(BlockHeader, u64)>> {
if chain.height().await == 0 {
info!("Fetching all headers");
return self.get_all_headers(bestblockhash).await;
}
info!(
"Downloading new block headers ({} already indexed) from {}",
chain.height().await,
bestblockhash,
);
let mut new_headers: Vec<(BlockHeader, u64)> = vec![];
let null_hash = BlockHash::all_zeros();
let mut blockhash = *bestblockhash;
while blockhash != null_hash {
Waiter::shutdown_check()?;
if new_headers.len().is_multiple_of(1000) {
info!(
"Downloading headers progress: {} fetched... ",
new_headers.len()
);
}
if chain.contains(&blockhash).await {
break;
}
let header = self
.getblockheader(&blockhash)
.await
.context(format!("failed to get {} header", blockhash))?;
blockhash = header.0.prev_blockhash;
new_headers.push(header);
}
trace!("downloaded {} block headers", new_headers.len());
new_headers.reverse(); Ok(new_headers)
}
pub async fn get_genesis(&self) -> Result<Block> {
let hash = self.request("getblockhash", &Some(json!([0]))).await?;
let response = self.request("getblock", &Some(json!([hash, 0]))).await?;
block_from_value(response)
}
pub async fn get_block_count(&self) -> Result<u64> {
self.request("getblockcount", &None)
.await?
.as_u64()
.context("expected blockheight")
}
pub async fn p2p_keepalive(&self) {
let parent = self;
let _ = tokio::time::timeout(Duration::from_millis(50), parent.p2p_pool.get()).await;
}
}