From 20d52deaaa8fd6f8a6f6cd01dacc54f9e810814d Mon Sep 17 00:00:00 2001 From: Randall Naar Date: Fri, 31 Jul 2026 15:55:41 -0400 Subject: [PATCH] Added elements support to /block-template. --- .gitignore | 1 + README.md | 15 +- src/daemon.rs | 110 ++- src/new_index/block_template.rs | 860 ++++++++++++++++++ src/new_index/mod.rs | 4 +- src/new_index/query.rs | 97 +- src/rest.rs | 36 +- .../liquid_dynafed_getnewblockhex.hex | 1 + tests/rest.rs | 33 + 9 files changed, 1052 insertions(+), 105 deletions(-) create mode 100644 src/new_index/block_template.rs create mode 100644 tests/fixtures/liquid_dynafed_getnewblockhex.hex diff --git a/.gitignore b/.gitignore index 9e717b212..3805efb66 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,5 @@ target +.DS_Store *db/ *.log *.sublime* diff --git a/README.md b/README.md index a0514fa81..ad00e5874 100644 --- a/README.md +++ b/README.md @@ -84,9 +84,18 @@ See `$ cargo run --release --bin electrs -- --help` for the full list of options ### Mining-related HTTP endpoints `GET /block-template` is available only with `--enable-mining-rest`. It proxies -the daemon's `getblocktemplate` template-mode response and caches successful -responses for 15 seconds, invalidating early when electrs indexes a new tip. -Callers that require fresher templates should account for this cache behavior. +the daemon's `getblocktemplate` response unchanged on Bitcoin-compatible chains. +On Liquid, it instead decodes the complete proposal returned by +`getnewblockhex` and projects the recoverable header, transaction, fee, +coinbase, and witness-commitment data into the same response shape. Fields that +have no equivalent mining semantics for signed dynafed blocks use compatibility +defaults or are omitted. The Liquid response is intended for template inspection +and distribution, not block reconstruction or federation signing. + +Successful responses are cached for 15 seconds and invalidated early when +electrs indexes a new tip. Cache misses are coalesced into one daemon request. +Responses use `Cache-Control: no-store`, so downstream caches do not extend the +internal lifetime. ## License diff --git a/src/daemon.rs b/src/daemon.rs index cb2cc2553..df4885843 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -222,6 +222,40 @@ pub trait CookieGetter: Send + Sync { fn get(&self) -> Result>; } +#[derive(Clone)] +struct ConnectionConfig { + addr: SocketAddr, + fallback: Option, + cookie_getter: Arc, + signal: Waiter, + max_age: Option, +} + +impl ConnectionConfig { + fn connect(&self) -> Result { + Connection::new( + self.addr, + self.fallback, + Arc::clone(&self.cookie_getter), + self.signal.clone(), + self.max_age, + ) + } + + fn connect_once(&self) -> Result { + let (conn, active_addr) = tcp_connect_once(self.addr, self.fallback)?; + Connection::from_stream( + conn, + active_addr, + self.addr, + self.fallback, + Arc::clone(&self.cookie_getter), + self.signal.clone(), + self.max_age, + ) + } +} + struct Connection { tx: TcpStream, rx: Lines>, @@ -516,7 +550,9 @@ pub struct Daemon { daemon_dir: PathBuf, blocks_dir: PathBuf, network: Network, + connection_config: ConnectionConfig, conn: Mutex, + block_template_conn: Mutex>, message_id: Counter, // for monotonic JSONRPC 'id' signal: Waiter, conn_max_age: Option, @@ -542,17 +578,21 @@ impl Daemon { metrics: &Metrics, conn_max_age: Option, ) -> Result { + let connection_config = ConnectionConfig { + addr: daemon_rpc_addr, + fallback: daemon_rpc_fallback_addr, + cookie_getter, + signal: signal.clone(), + max_age: conn_max_age, + }; + let conn = connection_config.connect()?; let daemon = Daemon { daemon_dir: daemon_dir.clone(), blocks_dir: blocks_dir.clone(), network, - conn: Mutex::new(Connection::new( - daemon_rpc_addr, - daemon_rpc_fallback_addr, - cookie_getter, - signal.clone(), - conn_max_age, - )?), + connection_config, + conn: Mutex::new(conn), + block_template_conn: Mutex::new(None), message_id: Counter::new(), signal: signal.clone(), conn_max_age, @@ -616,7 +656,9 @@ impl Daemon { daemon_dir: self.daemon_dir.clone(), blocks_dir: self.blocks_dir.clone(), network: self.network, + connection_config: self.connection_config.clone(), conn: Mutex::new(self.conn.lock().unwrap().reconnect()?), + block_template_conn: Mutex::new(None), message_id: Counter::new(), signal: self.signal.clone(), conn_max_age: self.conn_max_age, @@ -664,8 +706,12 @@ impl Daemon { } #[trace] - fn call_jsonrpc(&self, method: &str, request: &Value) -> Result { - let mut conn = self.conn.lock().unwrap(); + fn call_jsonrpc_on_connection( + &self, + method: &str, + request: &Value, + conn: &mut Connection, + ) -> Result { // Proactively recycle connections older than the configured max age. Re-establishing // the TCP connection lets a fronting load balancer (e.g. a Kubernetes ClusterSetIP) // re-select a backend, so a long-lived connection does not stay pinned to a stale @@ -711,6 +757,12 @@ impl Daemon { Ok(result) } + #[trace] + fn call_jsonrpc(&self, method: &str, request: &Value) -> Result { + let mut conn = self.conn.lock().unwrap(); + self.call_jsonrpc_on_connection(method, request, &mut conn) + } + #[trace(method = %method)] fn handle_request(&self, method: &str, params: &Value) -> Result { let id = self.message_id.next(); @@ -746,9 +798,32 @@ impl Daemon { self.retry_request(method, ¶ms) } + /// Perform a block-template RPC on a connection isolated from singleton RPC users. + /// Connection and warmup failures are returned to the REST caller without retrying. #[trace] - fn request_no_retry(&self, method: &str, params: Value) -> Result { - self.handle_request(method, ¶ms) + fn request_block_template_no_retry(&self, method: &str, params: Value) -> Result { + let id = self.message_id.next(); + let req = json!({"method": method, "params": params, "id": id}); + let reply = { + let mut slot = self.block_template_conn.lock().unwrap(); + if slot.is_none() { + *slot = Some(self.connection_config.connect_once()?); + } + let result = self.call_jsonrpc_on_connection( + method, + &req, + slot.as_mut() + .expect("block-template connection initialized"), + ); + if matches!( + result.as_ref().err().map(|err| err.kind()), + Some(ErrorKind::Connection(_)) + ) { + *slot = None; + } + result? + }; + parse_jsonrpc_reply(reply, method, id) } #[trace] @@ -938,9 +1013,20 @@ impl Daemon { Ok(serde_json::from_value(res).chain_err(|| "invalid getrawmempool reply")?) } + #[cfg(not(feature = "liquid"))] #[trace] pub fn getblocktemplate(&self, rules: &[&str]) -> Result { - self.request_no_retry("getblocktemplate", json!([{ "rules": rules }])) + self.request_block_template_no_retry("getblocktemplate", json!([{ "rules": rules }])) + } + + #[cfg(feature = "liquid")] + #[trace] + pub fn getnewblockhex(&self) -> Result { + let value = self.request_block_template_no_retry("getnewblockhex", json!([]))?; + value + .as_str() + .map(str::to_owned) + .chain_err(|| "non-string getnewblockhex response") } #[trace] diff --git a/src/new_index/block_template.rs b/src/new_index/block_template.rs new file mode 100644 index 000000000..4981fbfde --- /dev/null +++ b/src/new_index/block_template.rs @@ -0,0 +1,860 @@ +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use hyper::body::Bytes as BodyBytes; +#[cfg(any(not(feature = "liquid"), test))] +use serde_json::Value; +use tokio::sync::{watch, Mutex}; + +use crate::chain::{BlockHash, Network}; +use crate::daemon::Daemon; +use crate::errors::*; +use crate::new_index::ChainQuery; + +pub const GETBLOCKTEMPLATE_TTL: u64 = 15; // seconds + +pub struct BlockTemplateCache { + inner: Arc, +} + +struct BlockTemplateCacheInner { + entry: Mutex, + updated: watch::Sender, +} + +enum BlockTemplateCacheEntry { + Empty, + Fetching, + Ready(CachedBlockTemplate), + Failed(CachedBlockTemplateError), +} + +struct CachedBlockTemplate { + template_tip: BlockHash, + observed_tip: BlockHash, + fetched_at: Instant, + generation: u64, + body: BodyBytes, + // Keep the complete Elements proposal available for a future internal consumer. + _source_block_hex: Option, +} + +struct CachedBlockTemplateError { + generation: u64, + error: CachedFetchError, +} + +enum CachedFetchError { + Connection(String), + Rpc(i64, String, String), + Other(String), +} + +struct FetchedBlockTemplate { + tip: BlockHash, + body: BodyBytes, + source_block_hex: Option, +} + +#[derive(Copy, Clone)] +struct ChainTipState { + current: BlockHash, + template_is_on_current_chain: bool, +} + +type TipProbe = Arc ChainTipState + Send + Sync>; + +impl CachedBlockTemplate { + fn is_valid(&self, tip: ChainTipState, ttl: Duration) -> bool { + if self.fetched_at.elapsed() >= ttl { + return false; + } + + // A template can legitimately be one block ahead of electrs while the index catches up. + // In that case it remains valid only while the locally observed tip is unchanged and the + // template tip is not an older block already known on the current chain. + self.template_tip == tip.current + || (!tip.template_is_on_current_chain && self.observed_tip == tip.current) + } +} + +impl CachedFetchError { + fn from_error(error: &Error) -> Self { + match error.kind() { + ErrorKind::Connection(message) => Self::Connection(message.clone()), + ErrorKind::RpcError(code, message, method) => { + Self::Rpc(*code, message.clone(), method.clone()) + } + _ => Self::Other(error.to_string()), + } + } + + fn to_error(&self) -> Error { + match self { + Self::Connection(message) => ErrorKind::Connection(message.clone()).into(), + Self::Rpc(code, message, method) => { + ErrorKind::RpcError(*code, message.clone(), method.clone()).into() + } + Self::Other(message) => message.clone().into(), + } + } +} + +impl BlockTemplateCache { + pub fn new() -> Self { + let (updated, _) = watch::channel(0); + Self { + inner: Arc::new(BlockTemplateCacheInner { + entry: Mutex::new(BlockTemplateCacheEntry::Empty), + updated, + }), + } + } + + pub async fn get( + &self, + chain: Arc, + daemon: Arc, + network: Network, + ) -> Result { + let tip_probe: TipProbe = Arc::new(move |template_tip| { + let current = chain.best_hash(); + ChainTipState { + current, + template_is_on_current_chain: template_tip == ¤t + || chain.blockid_by_hash(template_tip).is_some(), + } + }); + self.get_or_fetch_with_tip_probe( + Duration::from_secs(GETBLOCKTEMPLATE_TTL), + tip_probe, + move || fetch_block_template(&daemon, network), + ) + .await + } + + async fn get_or_fetch_with_tip_probe( + &self, + ttl: Duration, + tip_probe: TipProbe, + fetch: F, + ) -> Result + where + F: FnOnce() -> Result + Send + 'static, + { + let mut fetch = Some(fetch); + // A completion with a different generation happened after this request began, so it is + // the single-flight result this caller should receive even if the tip changed meanwhile. + let request_generation = *self.inner.updated.borrow(); + + loop { + let mut updated = self.inner.updated.subscribe(); + let mut entry = self.inner.entry.lock().await; + match &*entry { + BlockTemplateCacheEntry::Ready(cached) => { + if cached.generation != request_generation + || cached.is_valid(tip_probe(&cached.template_tip), ttl) + { + return Ok(cached.body.clone()); + } + } + BlockTemplateCacheEntry::Failed(failure) + if failure.generation != request_generation => + { + return Err(failure.error.to_error()); + } + BlockTemplateCacheEntry::Fetching => { + drop(entry); + let _ = updated.changed().await; + continue; + } + BlockTemplateCacheEntry::Empty | BlockTemplateCacheEntry::Failed(_) => {} + } + + let fetch = fetch + .take() + .expect("block-template fetch was already started by this request"); + *entry = BlockTemplateCacheEntry::Fetching; + + let inner = Arc::clone(&self.inner); + let tip_probe = Arc::clone(&tip_probe); + // The cache owns the worker so disconnecting the request that observed the miss does + // not cancel the shared fetch and cause a second daemon call. + tokio::spawn(async move { + let result = match tokio::task::spawn_blocking(fetch).await { + Ok(result) => result, + Err(error) => Err(format!("block-template worker failed: {error}").into()), + }; + let generation = (*inner.updated.borrow()).wrapping_add(1); + let completed = match result { + Ok(fetched) => { + let observed_tip = tip_probe(&fetched.tip).current; + BlockTemplateCacheEntry::Ready(CachedBlockTemplate { + template_tip: fetched.tip, + observed_tip, + fetched_at: Instant::now(), + generation, + body: fetched.body, + _source_block_hex: fetched.source_block_hex, + }) + } + Err(error) => BlockTemplateCacheEntry::Failed(CachedBlockTemplateError { + generation, + error: CachedFetchError::from_error(&error), + }), + }; + + *inner.entry.lock().await = completed; + inner.updated.send_replace(generation); + }); + + drop(entry); + let _ = updated.changed().await; + } + } + + #[cfg(test)] + async fn get_or_fetch( + &self, + current_tip: BlockHash, + ttl: Duration, + fetch: F, + ) -> Result + where + F: FnOnce() -> Result + Send + 'static, + { + let tip_probe: TipProbe = Arc::new(move |template_tip| ChainTipState { + current: current_tip, + template_is_on_current_chain: template_tip == ¤t_tip, + }); + self.get_or_fetch_with_tip_probe(ttl, tip_probe, fetch) + .await + } +} + +#[cfg(any(not(feature = "liquid"), test))] +fn json_template_from_value(value: Value) -> Result { + let previousblockhash = value + .get("previousblockhash") + .and_then(Value::as_str) + .chain_err(|| "getblocktemplate response missing previousblockhash")?; + let tip = previousblockhash + .parse::() + .chain_err(|| "invalid getblocktemplate previousblockhash")?; + let body = BodyBytes::from( + serde_json::to_string(&value).chain_err(|| "failed to serialize getblocktemplate")?, + ); + Ok(FetchedBlockTemplate { + tip, + body, + source_block_hex: None, + }) +} + +#[cfg(not(feature = "liquid"))] +fn fetch_block_template(daemon: &Daemon, network: Network) -> Result { + json_template_from_value(daemon.getblocktemplate(block_template_rules(network))?) +} + +#[cfg(not(feature = "liquid"))] +fn block_template_rules(network: Network) -> &'static [&'static str] { + match network { + Network::Signet => &["segwit", "signet"], + _ => &["segwit"], + } +} + +#[cfg(feature = "liquid")] +fn fetch_block_template(daemon: &Daemon, network: Network) -> Result { + elements_template_from_hex(daemon.getnewblockhex()?, network) +} + +#[cfg(feature = "liquid")] +use std::collections::{BTreeMap, HashMap}; + +#[cfg(feature = "liquid")] +use bitcoin::hex::{DisplayHex, FromHex}; + +#[cfg(feature = "liquid")] +use crate::chain::{AssetId, Block, Transaction, Txid}; + +#[cfg(feature = "liquid")] +#[derive(Debug, Deserialize, Serialize)] +struct BlockTemplateResponse { + capabilities: Vec, + version: u32, + rules: Vec, + vbavailable: BTreeMap, + vbrequired: u32, + previousblockhash: String, + transactions: Vec, + coinbaseaux: BTreeMap, + coinbasevalue: u64, + target: String, + mutable: Vec, + noncerange: String, + curtime: u32, + bits: String, + height: u32, + #[serde(skip_serializing_if = "Option::is_none")] + default_witness_commitment: Option, +} + +#[cfg(feature = "liquid")] +#[derive(Debug, Deserialize, Serialize)] +struct BlockTemplateTransaction { + data: String, + txid: String, + hash: String, + depends: Vec, + fee: u64, + weight: usize, +} + +#[cfg(feature = "liquid")] +fn elements_template_from_hex(raw_hex: String, network: Network) -> Result { + let bytes = Vec::from_hex(&raw_hex).chain_err(|| "invalid getnewblockhex block hex")?; + let block: Block = elements::encode::deserialize(&bytes) + .chain_err(|| "failed to deserialize getnewblockhex Elements block")?; + let response = BlockTemplateResponse::from_block(&block, *network.native_asset())?; + let body = BodyBytes::from( + serde_json::to_string(&response) + .chain_err(|| "failed to serialize Elements block template")?, + ); + Ok(FetchedBlockTemplate { + tip: block.header.prev_blockhash, + body, + source_block_hex: Some(BodyBytes::from(raw_hex)), + }) +} + +#[cfg(feature = "liquid")] +impl BlockTemplateResponse { + fn from_block(block: &Block, policy_asset: AssetId) -> Result { + let coinbase = block + .txdata + .first() + .chain_err(|| "getnewblockhex block has no transactions")?; + if !coinbase.is_coinbase() { + bail!("getnewblockhex transaction zero is not coinbase") + } + if block.txdata.iter().skip(1).any(Transaction::is_coinbase) { + bail!("getnewblockhex block contains multiple coinbase transactions") + } + + let mut indexes = HashMap::::new(); + let mut transactions = Vec::with_capacity(block.txdata.len().saturating_sub(1)); + for (index, tx) in block.txdata.iter().enumerate() { + let txid = tx.txid(); + if index == 0 { + indexes.insert(txid, index); + continue; + } + let depends = tx + .input + .iter() + .filter_map(|input| indexes.get(&input.previous_output.txid).copied()) + .filter(|dependency| *dependency != 0) + .collect(); + transactions.push(BlockTemplateTransaction { + data: elements::encode::serialize_hex(tx), + txid: txid.to_string(), + hash: tx.wtxid().to_string(), + depends, + fee: tx.fee_in(policy_asset), + weight: tx.weight(), + }); + indexes.insert(txid, index); + } + + Ok(Self { + capabilities: vec![], + version: block.header.version, + rules: vec![], + vbavailable: BTreeMap::new(), + vbrequired: 0, + previousblockhash: block.header.prev_blockhash.to_string(), + transactions, + coinbaseaux: BTreeMap::new(), + coinbasevalue: coinbase_value(coinbase, policy_asset)?, + target: "0".repeat(64), + mutable: vec![], + noncerange: "00000000ffffffff".to_string(), + curtime: block.header.time, + bits: "00000000".to_string(), + height: block.header.height, + default_witness_commitment: witness_commitment(coinbase), + }) + } +} + +#[cfg(feature = "liquid")] +fn coinbase_value(coinbase: &Transaction, policy_asset: AssetId) -> Result { + coinbase.output.iter().try_fold(0u64, |total, output| { + if output.asset.explicit() != Some(policy_asset) { + return Ok(total); + } + let value = output + .value + .explicit() + .chain_err(|| "non-explicit policy-asset coinbase value")?; + total + .checked_add(value) + .chain_err(|| "policy-asset coinbase value overflow") + }) +} + +#[cfg(feature = "liquid")] +fn witness_commitment(coinbase: &Transaction) -> Option { + const PREFIX: &[u8] = &[0x6a, 0x24, 0xaa, 0x21, 0xa9, 0xed]; + coinbase.output.iter().rev().find_map(|output| { + let script = output.script_pubkey.as_bytes(); + (script.len() >= 38 && script.starts_with(PREFIX)).then(|| script.to_lower_hex_string()) + }) +} + +#[cfg(test)] +mod tests { + #[cfg(feature = "liquid")] + use std::convert::TryInto; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + + use super::*; + use serde_json::json; + use tokio::sync::Barrier; + + fn hash(byte: u8) -> BlockHash { + format!("{:02x}", byte) + .repeat(32) + .parse() + .expect("valid block hash") + } + + fn fetched(tip: BlockHash, body: &'static str) -> FetchedBlockTemplate { + FetchedBlockTemplate { + tip, + body: BodyBytes::from_static(body.as_bytes()), + source_block_hex: None, + } + } + + #[test] + fn json_template_preserves_unknown_fields() { + let value = json!({ + "previousblockhash": hash(1).to_string(), + "future-field": { "kept": true } + }); + let fetched = json_template_from_value(value.clone()).unwrap(); + assert_eq!( + serde_json::from_slice::(&fetched.body).unwrap(), + value + ); + } + + #[tokio::test] + async fn cache_hits_until_expiry_or_tip_change() { + let cache = BlockTemplateCache::new(); + let calls = Arc::new(AtomicUsize::new(0)); + let tip = hash(2); + + let body = cache + .get_or_fetch(tip, Duration::from_secs(60), { + let calls = Arc::clone(&calls); + move || { + calls.fetch_add(1, Ordering::SeqCst); + Ok(fetched(tip, "first")) + } + }) + .await + .unwrap(); + assert_eq!(&body[..], b"first"); + + let body = cache + .get_or_fetch(tip, Duration::from_secs(60), { + let calls = Arc::clone(&calls); + move || { + calls.fetch_add(1, Ordering::SeqCst); + Ok(fetched(tip, "unexpected")) + } + }) + .await + .unwrap(); + assert_eq!(&body[..], b"first"); + assert_eq!(calls.load(Ordering::SeqCst), 1); + + cache + .get_or_fetch(tip, Duration::ZERO, { + let calls = Arc::clone(&calls); + move || { + calls.fetch_add(1, Ordering::SeqCst); + Ok(fetched(tip, "expired")) + } + }) + .await + .unwrap(); + cache + .get_or_fetch(hash(3), Duration::from_secs(60), { + let calls = Arc::clone(&calls); + move || { + calls.fetch_add(1, Ordering::SeqCst); + Ok(fetched(hash(3), "new-tip")) + } + }) + .await + .unwrap(); + assert_eq!(calls.load(Ordering::SeqCst), 3); + } + + #[tokio::test] + async fn failed_fetch_is_not_cached() { + let cache = BlockTemplateCache::new(); + let calls = Arc::new(AtomicUsize::new(0)); + let tip = hash(4); + + let result = cache + .get_or_fetch(tip, Duration::from_secs(60), { + let calls = Arc::clone(&calls); + move || { + calls.fetch_add(1, Ordering::SeqCst); + bail!("fetch failed") + } + }) + .await; + assert!(result.is_err()); + + let body = cache + .get_or_fetch(tip, Duration::from_secs(60), { + let calls = Arc::clone(&calls); + move || { + calls.fetch_add(1, Ordering::SeqCst); + Ok(fetched(tip, "recovered")) + } + }) + .await + .unwrap(); + assert_eq!(&body[..], b"recovered"); + assert_eq!(calls.load(Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn daemon_ahead_template_is_cached_until_index_catches_up() { + let cache = BlockTemplateCache::new(); + let calls = Arc::new(AtomicUsize::new(0)); + let observed_tip = hash(5); + let template_tip = hash(6); + + cache + .get_or_fetch(observed_tip, Duration::from_secs(60), { + let calls = Arc::clone(&calls); + move || { + calls.fetch_add(1, Ordering::SeqCst); + Ok(fetched(template_tip, "new-tip-template")) + } + }) + .await + .unwrap(); + let body = cache + .get_or_fetch(observed_tip, Duration::from_secs(60), { + let calls = Arc::clone(&calls); + move || { + calls.fetch_add(1, Ordering::SeqCst); + Ok(fetched(template_tip, "unexpected")) + } + }) + .await + .unwrap(); + + assert_eq!(&body[..], b"new-tip-template"); + assert_eq!(calls.load(Ordering::SeqCst), 1); + + let body = cache + .get_or_fetch(template_tip, Duration::from_secs(60), { + let calls = Arc::clone(&calls); + move || { + calls.fetch_add(1, Ordering::SeqCst); + Ok(fetched(template_tip, "unexpected")) + } + }) + .await + .unwrap(); + assert_eq!(&body[..], b"new-tip-template"); + assert_eq!(calls.load(Ordering::SeqCst), 1); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn concurrent_miss_is_single_flight() { + let cache = Arc::new(BlockTemplateCache::new()); + let calls = Arc::new(AtomicUsize::new(0)); + let ready = Arc::new(Barrier::new(5)); + let tip = hash(7); + let mut tasks = vec![]; + for _ in 0..4 { + let cache = Arc::clone(&cache); + let calls = Arc::clone(&calls); + let ready = Arc::clone(&ready); + tasks.push(tokio::spawn(async move { + ready.wait().await; + cache + .get_or_fetch(tip, Duration::from_secs(60), move || { + calls.fetch_add(1, Ordering::SeqCst); + std::thread::sleep(Duration::from_millis(25)); + Ok(fetched(tip, "shared")) + }) + .await + .unwrap() + })); + } + ready.wait().await; + for task in tasks { + assert_eq!(&task.await.unwrap()[..], b"shared"); + } + assert_eq!(calls.load(Ordering::SeqCst), 1); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn concurrent_failed_miss_is_single_flight() { + let cache = Arc::new(BlockTemplateCache::new()); + let calls = Arc::new(AtomicUsize::new(0)); + let ready = Arc::new(Barrier::new(5)); + let tip = hash(8); + let mut tasks = vec![]; + for _ in 0..4 { + let cache = Arc::clone(&cache); + let calls = Arc::clone(&calls); + let ready = Arc::clone(&ready); + tasks.push(tokio::spawn(async move { + ready.wait().await; + cache + .get_or_fetch(tip, Duration::from_secs(60), move || { + calls.fetch_add(1, Ordering::SeqCst); + std::thread::sleep(Duration::from_millis(25)); + bail!("shared failure") + }) + .await + .unwrap_err() + .to_string() + })); + } + ready.wait().await; + for task in tasks { + assert_eq!(task.await.unwrap(), "shared failure"); + } + assert_eq!(calls.load(Ordering::SeqCst), 1); + + let body = cache + .get_or_fetch(tip, Duration::from_secs(60), { + let calls = Arc::clone(&calls); + move || { + calls.fetch_add(1, Ordering::SeqCst); + Ok(fetched(tip, "recovered")) + } + }) + .await + .unwrap(); + assert_eq!(&body[..], b"recovered"); + assert_eq!(calls.load(Ordering::SeqCst), 2); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn cancelled_request_does_not_cancel_shared_fetch() { + let cache = Arc::new(BlockTemplateCache::new()); + let calls = Arc::new(AtomicUsize::new(0)); + let tip = hash(9); + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + + let leader = { + let cache = Arc::clone(&cache); + let calls = Arc::clone(&calls); + tokio::spawn(async move { + cache + .get_or_fetch(tip, Duration::from_secs(60), move || { + calls.fetch_add(1, Ordering::SeqCst); + let _ = started_tx.send(()); + std::thread::sleep(Duration::from_millis(25)); + Ok(fetched(tip, "shared")) + }) + .await + }) + }; + started_rx.await.unwrap(); + leader.abort(); + + let body = cache + .get_or_fetch(tip, Duration::from_secs(60), { + let calls = Arc::clone(&calls); + move || { + calls.fetch_add(1, Ordering::SeqCst); + Ok(fetched(tip, "unexpected")) + } + }) + .await + .unwrap(); + assert_eq!(&body[..], b"shared"); + assert_eq!(calls.load(Ordering::SeqCst), 1); + } + + #[cfg(feature = "liquid")] + fn dynafed_fixture() -> (String, Block) { + // A witness-bearing dynafed block from rust-elements' block_decoder test, + // used here as representative getnewblockhex output. + let raw = include_str!("../../tests/fixtures/liquid_dynafed_getnewblockhex.hex") + .trim() + .to_string(); + let bytes = Vec::from_hex(&raw).unwrap(); + let block = elements::encode::deserialize(&bytes).unwrap(); + (raw, block) + } + + #[cfg(feature = "liquid")] + #[test] + fn projects_dynafed_getnewblockhex_fixture() { + let (raw, block) = dynafed_fixture(); + let fetched = elements_template_from_hex(raw.clone(), Network::LiquidRegtest).unwrap(); + let response: Value = serde_json::from_slice(&fetched.body).unwrap(); + let raw_bytes = Vec::from_hex(&raw).unwrap(); + + assert_eq!( + u32::from_le_bytes(raw_bytes[..4].try_into().unwrap()), + 0xa000_0000 + ); + assert_eq!(block.header.version, 0x2000_0000); + assert_eq!(response["version"].as_u64(), Some(0x2000_0000)); + assert_eq!(response["height"].as_u64(), Some(7)); + assert_eq!(response["curtime"].as_u64(), Some(block.header.time.into())); + assert_eq!( + response["previousblockhash"].as_str(), + Some(block.header.prev_blockhash.to_string().as_str()) + ); + assert_eq!(response["transactions"].as_array().unwrap().len(), 0); + assert_eq!(response["coinbasevalue"].as_u64(), Some(0)); + assert_eq!(response["capabilities"], json!([])); + assert_eq!(response["rules"], json!([])); + assert_eq!(response["vbavailable"], json!({})); + assert_eq!(response["vbrequired"].as_u64(), Some(0)); + assert_eq!(response["coinbaseaux"], json!({})); + assert_eq!(response["mutable"], json!([])); + assert_eq!(response["bits"].as_str(), Some("00000000")); + assert_eq!(response["target"], json!("0".repeat(64))); + assert_eq!(response["noncerange"].as_str(), Some("00000000ffffffff")); + assert_eq!( + response["default_witness_commitment"].as_str(), + Some("6a24aa21a9ed94f15ed3a62165e4a0b99699cc28b48e19cb5bc1b1f47155db62d63f1e047d45") + ); + for omitted in [ + "longpollid", + "mintime", + "sigoplimit", + "sizelimit", + "weightlimit", + ] { + assert!(response.get(omitted).is_none(), "unexpected {omitted}"); + } + assert_eq!(fetched.source_block_hex.unwrap().as_ref(), raw.as_bytes()); + } + + #[cfg(feature = "liquid")] + #[test] + fn projects_fees_dependencies_and_witness_transactions() { + use elements::{LockTime, OutPoint, TxIn, TxInWitness, TxOut}; + + let (_, mut block) = dynafed_fixture(); + let policy_asset = *Network::LiquidRegtest.native_asset(); + let external_txid: Txid = "11".repeat(32).parse().unwrap(); + let parent = Transaction { + version: 2, + lock_time: LockTime::ZERO, + input: vec![TxIn { + previous_output: OutPoint { + txid: external_txid, + vout: 0, + }, + ..TxIn::default() + }], + output: vec![TxOut::new_fee(5, policy_asset)], + }; + let child = Transaction { + version: 2, + lock_time: LockTime::ZERO, + input: vec![TxIn { + previous_output: OutPoint { + txid: parent.txid(), + vout: 0, + }, + witness: TxInWitness { + script_witness: vec![vec![1, 2, 3]], + ..TxInWitness::default() + }, + ..TxIn::default() + }], + output: vec![TxOut::new_fee(7, policy_asset)], + }; + block.txdata.push(parent.clone()); + block.txdata.push(child.clone()); + + let response = BlockTemplateResponse::from_block(&block, policy_asset).unwrap(); + assert_eq!(response.transactions.len(), 2); + assert_eq!(response.transactions[0].depends, Vec::::new()); + assert_eq!(response.transactions[1].depends, vec![1]); + assert_eq!(response.transactions[0].fee, 5); + assert_eq!(response.transactions[1].fee, 7); + assert_eq!(response.transactions[0].txid, parent.txid().to_string()); + assert_eq!(response.transactions[1].hash, child.wtxid().to_string()); + assert_ne!(response.transactions[1].hash, response.transactions[1].txid); + assert_eq!(response.transactions[1].weight, child.weight()); + assert_eq!( + response.transactions[1].data, + elements::encode::serialize_hex(&child) + ); + let response = serde_json::to_value(&response).unwrap(); + assert!(response["transactions"] + .as_array() + .unwrap() + .iter() + .all(|tx| tx.get("sigops").is_none())); + } + + #[cfg(feature = "liquid")] + #[test] + fn validates_coinbase_and_malformed_source() { + use elements::confidential; + + let (_, mut block) = dynafed_fixture(); + let coinbase = block.txdata.first_mut().unwrap(); + let fixture_asset = coinbase.output[0].asset.explicit().unwrap(); + coinbase.output[0].value = confidential::Value::Explicit(42); + assert_eq!( + BlockTemplateResponse::from_block(&block, fixture_asset) + .unwrap() + .coinbasevalue, + 42 + ); + + let mut empty = block.clone(); + empty.txdata.clear(); + assert!(BlockTemplateResponse::from_block(&empty, fixture_asset).is_err()); + + let mut duplicate = block.clone(); + duplicate.txdata.push(duplicate.txdata[0].clone()); + assert!(BlockTemplateResponse::from_block(&duplicate, fixture_asset).is_err()); + assert!(elements_template_from_hex("not hex".to_string(), Network::LiquidRegtest).is_err()); + assert!(elements_template_from_hex("00".to_string(), Network::LiquidRegtest).is_err()); + } + + #[cfg(feature = "liquid")] + #[test] + fn bitcoin_and_elements_bodies_share_the_response_contract() { + let (raw, _) = dynafed_fixture(); + let elements = elements_template_from_hex(raw, Network::LiquidRegtest).unwrap(); + let elements_value: Value = serde_json::from_slice(&elements.body).unwrap(); + let _: BlockTemplateResponse = serde_json::from_value(elements_value.clone()).unwrap(); + + let mut bitcoin_value = elements_value; + bitcoin_value["previousblockhash"] = json!(hash(9).to_string()); + bitcoin_value["bitcoin-only-field"] = json!("preserved"); + let bitcoin = json_template_from_value(bitcoin_value.clone()).unwrap(); + let parsed: Value = serde_json::from_slice(&bitcoin.body).unwrap(); + let _: BlockTemplateResponse = serde_json::from_value(parsed.clone()).unwrap(); + assert_eq!(parsed, bitcoin_value); + } +} diff --git a/src/new_index/mod.rs b/src/new_index/mod.rs index 723dd27ab..80cef99c7 100644 --- a/src/new_index/mod.rs +++ b/src/new_index/mod.rs @@ -1,3 +1,4 @@ +mod block_template; pub mod db; pub mod db_metrics; mod fetch; @@ -10,7 +11,8 @@ pub mod zmq; pub use self::db::{DBRow, DB}; pub use self::fetch::{BlockEntry, FetchFrom}; pub use self::mempool::Mempool; -pub use self::query::{Query, GETBLOCKTEMPLATE_TTL}; +pub use self::block_template::GETBLOCKTEMPLATE_TTL; +pub use self::query::Query; pub use self::schema::{ compute_script_hash, parse_hash, ChainQuery, FundingInfo, GetAmountVal, Indexer, ScriptStats, SpendingInfo, SpendingInput, Store, TxHistoryInfo, TxHistoryKey, TxHistoryRow, Utxo, diff --git a/src/new_index/query.rs b/src/new_index/query.rs index 20cc0969c..86dada56e 100644 --- a/src/new_index/query.rs +++ b/src/new_index/query.rs @@ -1,18 +1,17 @@ use std::collections::{BTreeSet, HashMap}; -use std::sync::{Arc, Mutex, RwLock, RwLockReadGuard}; +use std::sync::{Arc, RwLock, RwLockReadGuard}; use std::time::{Duration, Instant}; use crate::chain::{Network, OutPoint, Transaction, TxOut, Txid}; use crate::config::Config; use crate::daemon::{Daemon, SubmitPackageResult}; use crate::errors::*; +use crate::new_index::block_template::BlockTemplateCache; use crate::new_index::{ChainQuery, Mempool, ScriptStats, SpendingInput, Utxo}; use crate::util::{is_spendable, BlockId, Bytes, TransactionStatus}; use electrs_macros::trace; use hyper::body::Bytes as BodyBytes; -use serde_json::Value; -use std::str::FromStr; #[cfg(feature = "liquid")] use crate::{ @@ -21,7 +20,6 @@ use crate::{ }; const FEE_ESTIMATES_TTL: u64 = 60; // seconds -pub const GETBLOCKTEMPLATE_TTL: u64 = 15; // seconds const CONF_TARGETS: [u16; 28] = [ 1u16, 2u16, 3u16, 4u16, 5u16, 6u16, 7u16, 8u16, 9u16, 10u16, 11u16, 12u16, 13u16, 14u16, 15u16, @@ -35,17 +33,11 @@ pub struct Query { config: Arc, cached_estimates: RwLock<(HashMap, Option)>, cached_relayfee: RwLock>, - cached_block_template: Mutex>, + cached_block_template: BlockTemplateCache, #[cfg(feature = "liquid")] asset_db: Option>>, } -struct CachedBlockTemplate { - tip: crate::chain::BlockHash, - fetched_at: Instant, - body: BodyBytes, -} - impl Query { #[cfg(not(feature = "liquid"))] pub fn new( @@ -61,7 +53,7 @@ impl Query { config, cached_estimates: RwLock::new((HashMap::new(), None)), cached_relayfee: RwLock::new(None), - cached_block_template: Mutex::new(None), + cached_block_template: BlockTemplateCache::new(), } } @@ -114,40 +106,14 @@ impl Query { } #[trace] - pub fn getblocktemplate(&self) -> Result { - let tip = self.chain.best_hash(); - - let mut cache = self.cached_block_template.lock().unwrap(); - if let Some(cached) = cache.as_ref() { - if cached.tip == tip - && cached.fetched_at.elapsed() < Duration::from_secs(GETBLOCKTEMPLATE_TTL) - { - return Ok(cached.body.clone()); - } - } - - let value = self - .daemon - .getblocktemplate(block_template_rules(self.config.network_type))?; - let body = BodyBytes::from( - serde_json::to_string(&value).chain_err(|| "failed to serialize getblocktemplate")?, - ); - - match block_template_tip(&value) { - Ok(tip) => { - *cache = Some(CachedBlockTemplate { - tip, - fetched_at: Instant::now(), - body: body.clone(), - }); - } - Err(err) => { - warn!("not caching getblocktemplate response: {}", err); - *cache = None; - } - } - - Ok(body) + pub async fn getblocktemplate(&self) -> Result { + self.cached_block_template + .get( + Arc::clone(&self.chain), + Arc::clone(&self.daemon), + self.config.network_type, + ) + .await } #[trace] @@ -327,7 +293,7 @@ impl Query { asset_db, cached_estimates: RwLock::new((HashMap::new(), None)), cached_relayfee: RwLock::new(None), - cached_block_template: Mutex::new(None), + cached_block_template: BlockTemplateCache::new(), } } @@ -361,40 +327,3 @@ impl Query { Ok((total_num, results)) } } - -#[cfg(not(feature = "liquid"))] -fn block_template_rules(network: Network) -> &'static [&'static str] { - match network { - Network::Signet => &["segwit", "signet"], - _ => &["segwit"], - } -} - -#[cfg(feature = "liquid")] -fn block_template_rules(_network: Network) -> &'static [&'static str] { - &["segwit"] -} - -fn block_template_tip(value: &Value) -> Result { - let previousblockhash = value - .get("previousblockhash") - .and_then(|value| value.as_str()) - .chain_err(|| "getblocktemplate response missing previousblockhash")?; - crate::chain::BlockHash::from_str(previousblockhash) - .chain_err(|| "invalid getblocktemplate previousblockhash") -} - -#[cfg(test)] -mod tests { - use serde_json::json; - - #[test] - fn block_template_tip_parses_previousblockhash() { - let hash = "0000000000000000000000000000000000000000000000000000000000000000"; - let value = json!({ "previousblockhash": hash }); - assert_eq!(super::block_template_tip(&value).unwrap().to_string(), hash); - - assert!(super::block_template_tip(&json!({})).is_err()); - assert!(super::block_template_tip(&json!({ "previousblockhash": "not a hash" })).is_err()); - } -} diff --git a/src/rest.rs b/src/rest.rs index 850a3a79a..05723314b 100644 --- a/src/rest.rs +++ b/src/rest.rs @@ -533,7 +533,7 @@ fn spawn_conn( let resp_result = match body_result { Ok(Ok(collected)) => { - handle_request(method, uri, collected.to_bytes(), &query, &config) + handle_request(method, uri, collected.to_bytes(), &query, &config).await } // Inner Err by http_body_util::Limited, either a LengthLimitError or an error by the underlying body stream Ok(Err(e)) if e.is::() => Err(HttpError( @@ -672,7 +672,7 @@ impl Handle { } #[trace] -fn handle_request( +async fn handle_request( method: Method, uri: hyper::Uri, body: Bytes, @@ -1156,7 +1156,7 @@ fn handle_request( "mining REST endpoints are disabled".to_string(), )); } - getblocktemplate_response(query.getblocktemplate()) + getblocktemplate_response(query.getblocktemplate().await) } #[cfg(feature = "liquid")] @@ -1366,14 +1366,16 @@ fn getblocktemplate_response( if let errors::ErrorKind::Connection(message) = err.kind() { return text_response_no_store(StatusCode::BAD_GATEWAY, message.clone()); } - Err(HttpError::from(err)) + text_response_no_store(StatusCode::BAD_GATEWAY, err.to_string()) } } } fn getblocktemplate_rpc_error(err: &errors::Error) -> Option<(i64, String)> { match err.kind() { - errors::ErrorKind::RpcError(code, message, method) if method == "getblocktemplate" => { + errors::ErrorKind::RpcError(code, message, method) + if method == "getblocktemplate" || method == "getnewblockhex" => + { Some((*code, message.clone())) } _ => None, @@ -1628,6 +1630,17 @@ mod tests { errors::ErrorKind::RpcError(-5, "Block not found".to_string(), "getblock".to_string()) .into(); assert_eq!(super::getblocktemplate_rpc_error(&other_method), None); + + let elements_err: errors::Error = errors::ErrorKind::RpcError( + -28, + "warming up".to_string(), + "getnewblockhex".to_string(), + ) + .into(); + assert_eq!( + super::getblocktemplate_rpc_error(&elements_err), + Some((-28, "warming up".to_string())) + ); } #[tokio::test] @@ -1665,5 +1678,18 @@ mod tests { ); let body = response.into_body().collect().await.unwrap().to_bytes(); assert_eq!(&body[..], b"daemon unavailable"); + + let malformed: errors::Error = "invalid getnewblockhex block hex".into(); + let response = super::getblocktemplate_response(Err(malformed)).unwrap(); + assert_eq!(response.status(), StatusCode::BAD_GATEWAY); + assert_eq!( + response + .headers() + .get("cache-control") + .and_then(|value| value.to_str().ok()), + Some("no-store") + ); + let body = response.into_body().collect().await.unwrap().to_bytes(); + assert_eq!(&body[..], b"invalid getnewblockhex block hex"); } } diff --git a/tests/fixtures/liquid_dynafed_getnewblockhex.hex b/tests/fixtures/liquid_dynafed_getnewblockhex.hex new file mode 100644 index 000000000..de3c0b6a0 --- /dev/null +++ b/tests/fixtures/liquid_dynafed_getnewblockhex.hex @@ -0,0 +1 @@ +000000a0da9d569617d1d65c3390a01c18c4fa7c4d0f4738b6fc2b5c5faf2e8a463abbaa46eb9123808e1e2ff75e9472fa0f0589b53b7518a69d3d6fcb9228ed345734ea06b9c45d070000000122002057c555a91edf9552282d88624d1473c275e64b7218870eb8fb0335b442976b8d02010000fbee9cea00d8efdc49cfbec328537e0d7032194de6ebf3cf42e5c05bb89a08b100040047304402206f55bc871387a9840489d47624b02995e774e3b70fed56d1eb43a9a53d4fd3e102201e1cbfbbd1079f5bea3bc216882d3fefbf6f27aa761820d3a88f12e5a5ea7ff001483045022100c072816f6561e73ee6c0ae32d55c3eec4da73b035425e4eb05ab50772591b4360220311bf295010094a489d9b280d9dafb724d776a1d99b9ede31c4b59bc2095c5c30169522103cadff18e928133df2e670a3715c4e7a81d357de36ddaa5016628e70a3e6a452f21021f0d8638c413ef7769cd711ce84c8f192f5a85f0fd6d8e63ddb4d2cf6740b23b210296db75c11ea3a292a372f6c94f5013eaeb379f701857a702f3b83f88da21be6f53ae010200000001010000000000000000000000000000000000000000000000000000000000000000ffffffff03570101ffffffff020137c495f58d698979ff9124e8c7455fe79b13ddb96afa25c45894eb059868a8c001000000000000000000016a0137c495f58d698979ff9124e8c7455fe79b13ddb96afa25c45894eb059868a8c001000000000000000000266a24aa21a9ed94f15ed3a62165e4a0b99699cc28b48e19cb5bc1b1f47155db62d63f1e047d4500000000000001200000000000000000000000000000000000000000000000000000000000000000000000 diff --git a/tests/rest.rs b/tests/rest.rs index 982462dc9..91100cf9c 100644 --- a/tests/rest.rs +++ b/tests/rest.rs @@ -459,6 +459,10 @@ fn test_rest_mempool() -> Result<()> { fn test_rest_getblocktemplate() -> Result<()> { let (rest_handle, rest_addr, mut tester) = common::init_rest_tester().unwrap(); + let address = tester.newaddress()?; + let pending_txid = tester + .send(&address, "0.001 BTC".parse().unwrap())? + .to_string(); let tip = tester.get_best_block_hash()?; let response = get(rest_addr, "/block-template")?; assert_eq!( @@ -477,6 +481,35 @@ fn test_rest_getblocktemplate() -> Result<()> { assert!(template["version"].is_i64() || template["version"].is_u64()); assert!(template["rules"].is_array()); assert!(template["bits"].is_string()); + assert!(template["transactions"] + .as_array() + .unwrap() + .iter() + .any(|tx| tx["txid"].as_str() == Some(pending_txid.as_str()))); + + #[cfg(feature = "liquid")] + { + assert_eq!(template["capabilities"], serde_json::json!([])); + assert_eq!(template["rules"], serde_json::json!([])); + assert_eq!(template["vbavailable"], serde_json::json!({})); + assert_eq!(template["vbrequired"].as_u64(), Some(0)); + assert_eq!(template["bits"].as_str(), Some("00000000")); + assert_eq!(template["target"], serde_json::json!("0".repeat(64))); + assert_eq!( + template["noncerange"].as_str(), + Some("00000000ffffffff") + ); + assert!(template.get("longpollid").is_none()); + assert!(template.get("mintime").is_none()); + assert!(template.get("sigoplimit").is_none()); + assert!(template.get("sizelimit").is_none()); + assert!(template.get("weightlimit").is_none()); + assert!(template["transactions"] + .as_array() + .unwrap() + .iter() + .all(|tx| tx.get("sigops").is_none())); + } let cached_template = get_json(rest_addr, "/block-template")?; assert_eq!(cached_template, template);