From fef0bd2ef4ee0c0480d2004d5ae5c856895f9b9e Mon Sep 17 00:00:00 2001 From: "liquan.eth" Date: Sat, 1 Aug 2026 15:52:28 +0800 Subject: [PATCH 1/2] refactor: apply simplify-review cleanups across core, common, and the binaries Co-Authored-By: Claude Fable 5 --- AGENTS.md | 4 +- Cargo.lock | 4 +- README.md | 2 +- bin/debug-trace-server/Cargo.toml | 1 - bin/debug-trace-server/src/chain_sync.rs | 14 +-- bin/debug-trace-server/src/metrics.rs | 13 +- bin/debug-trace-server/src/server_db.rs | 9 -- bin/stateless-validator/Cargo.toml | 2 - bin/stateless-validator/src/app.rs | 14 +-- bin/stateless-validator/src/chain_sync.rs | 28 +---- bin/stateless-validator/src/lib.rs | 21 +++- bin/stateless-validator/src/metrics.rs | 17 +-- bin/stateless-validator/src/r2_witness.rs | 14 +-- .../src/{workers.rs => runner.rs} | 10 +- bin/stateless-validator/src/validator_db.rs | 25 +--- bin/stateless-validator/tests/integration.rs | 114 ++++++++---------- crates/stateless-common/Cargo.toml | 1 + crates/stateless-common/src/lib.rs | 4 +- crates/stateless-common/src/metrics.rs | 29 ++++- crates/stateless-common/src/rpc_client.rs | 104 +++++++++++----- crates/stateless-core/src/chain_spec.rs | 48 ++------ crates/stateless-core/src/data_types.rs | 29 ++++- crates/stateless-core/src/db.rs | 19 ++- crates/stateless-core/src/evm_database.rs | 49 +++++--- crates/stateless-core/src/executor.rs | 92 +++----------- crates/stateless-core/src/light_witness.rs | 9 +- .../stateless-core/src/pipeline/advancer.rs | 38 ++++-- crates/stateless-core/src/pipeline/fetcher.rs | 27 ++--- crates/stateless-core/src/pipeline/mod.rs | 22 +--- crates/stateless-core/src/pipeline/tests.rs | 34 ++---- 30 files changed, 362 insertions(+), 435 deletions(-) rename bin/stateless-validator/src/{workers.rs => runner.rs} (97%) diff --git a/AGENTS.md b/AGENTS.md index c73406eb..87c26ac2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -41,7 +41,7 @@ The project uses nightly `2026-02-03` toolchain (edition 2024, rust-version 1.95 | `stateless-common` | `crates/stateless-common` | RPC client, metrics/logging utilities, witness size estimation | | `stateless-test-utils` | `crates/stateless-test-utils` | Test fixtures (blocks, witnesses, contracts) and env-var lock for integration tests | | `stateless-r2` | `crates/stateless-r2` | Shared R2 (S3) witness primitives: SigV4 signer, object-key layout, endpoint parsing, signed PUT; consumed by mega-reth's uploaders (write) and the validator's R2 witness source (read) | -| `stateless-validator` | `bin/stateless-validator` | Main binary: chain sync, parallel validation workers (`app.rs` / `workers.rs` / `main.rs`) | +| `stateless-validator` | `bin/stateless-validator` | Main binary: chain sync, parallel validation workers (`app.rs` / `runner.rs` / `main.rs`) | | `debug-trace-server` | `bin/debug-trace-server` | Standalone RPC server for debug/trace methods | Additional directories: `test_data/` (integration test fixtures including genesis config), `audits/` (security audit reports). @@ -128,7 +128,7 @@ The background chain-sync prefetch routes by freshness against the last observed | `crates/stateless-db/src/{lib,tables,helpers,serialize,cache}.rs` | Shared redb tables, helpers, serialization, and `ContractCache` | | `crates/stateless-common/src/rpc_client.rs` | RPC client for blocks, witnesses, and bytecode | | `crates/stateless-common/src/metrics.rs` | RpcMethod, RpcMetrics, RpcClientConfig | -| `bin/stateless-validator/src/{main,app,workers,chain_sync,validator_db,metrics}.rs` | Thin entry, CLI/startup wiring, pipeline+reporter, fetcher/processor, DB | +| `bin/stateless-validator/src/{main,app,runner,chain_sync,validator_db,metrics}.rs` | Thin entry, CLI/startup wiring, pipeline+reporter, fetcher/processor, DB | | `bin/debug-trace-server/src/chain_sync.rs` | TraceFetcher, TraceProcessor, TraceHooks | | `bin/debug-trace-server/src/rpc_service.rs` | RPC method definitions and handlers | | `bin/debug-trace-server/src/data_provider.rs` | Block data fetching with single-flight coalescing | diff --git a/Cargo.lock b/Cargo.lock index 21dc5230..5ad54709 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1909,7 +1909,6 @@ dependencies = [ "mega-evm", "metrics", "metrics-derive", - "metrics-exporter-prometheus", "op-alloy-network", "op-alloy-rpc-types", "pin-project-lite", @@ -5666,6 +5665,7 @@ dependencies = [ "futures", "jsonrpsee", "kanal", + "metrics-exporter-prometheus", "op-alloy-network", "op-alloy-rpc-types", "reqwest", @@ -5785,11 +5785,9 @@ dependencies = [ "chrono", "clap", "eyre", - "fastrand", "jsonrpsee", "jsonrpsee-types", "metrics", - "metrics-exporter-prometheus", "op-alloy-rpc-types", "redb", "reqwest", diff --git a/README.md b/README.md index 272eba9e..912bc794 100644 --- a/README.md +++ b/README.md @@ -257,7 +257,7 @@ The pipeline is configured via `PipelineConfig` and customized through trait imp | `crates/stateless-common/src/metrics.rs` | `RpcMethod`, `RpcMetrics`, `RpcClientConfig` | | `crates/stateless-common/src/witness_size.rs` | `WitnessSizeBreakdown` + `estimate_witness_size` for RPC and trace-server metrics | | `crates/stateless-test-utils/src/fixtures.rs` | `TestFixtures` loader (blocks, SALT/MPT witnesses, contracts, genesis) | -| `bin/stateless-validator/src/{main,app,workers,chain_sync,validator_db,metrics}.rs` | Thin entry, CLI/startup wiring, pipeline+reporter, fetcher/processor, DB | +| `bin/stateless-validator/src/{main,app,runner,chain_sync,validator_db,metrics}.rs` | Thin entry, CLI/startup wiring, pipeline+reporter, fetcher/processor, DB | | `bin/debug-trace-server/src/chain_sync.rs` | `TraceFetcher`, `TraceProcessor`, `TraceHooks` | | `bin/debug-trace-server/src/rpc_service.rs` | RPC method definitions and handlers | | `bin/debug-trace-server/src/data_provider.rs` | Block data fetching with single-flight coalescing | diff --git a/bin/debug-trace-server/Cargo.toml b/bin/debug-trace-server/Cargo.toml index d0346a8f..3f1ba08f 100644 --- a/bin/debug-trace-server/Cargo.toml +++ b/bin/debug-trace-server/Cargo.toml @@ -44,7 +44,6 @@ jsonrpsee.workspace = true libc.workspace = true metrics.workspace = true metrics-derive.workspace = true -metrics-exporter-prometheus.workspace = true pin-project-lite.workspace = true quick_cache.workspace = true rayon.workspace = true diff --git a/bin/debug-trace-server/src/chain_sync.rs b/bin/debug-trace-server/src/chain_sync.rs index 5f4b7c5b..57b0487f 100644 --- a/bin/debug-trace-server/src/chain_sync.rs +++ b/bin/debug-trace-server/src/chain_sync.rs @@ -157,12 +157,7 @@ impl BlockFetcher for TraceFetcher { async fn latest_block_meta(&self) -> Result { let header = self.rpc_client.get_header(BlockId::Number(BlockNumberOrTag::Latest), false).await; - Ok(BlockMeta { - block_number: header.number, - block_hash: header.hash, - post_state_root: header.state_root, - post_withdrawals_root: header.withdrawals_root.unwrap_or_default(), - }) + Ok(BlockMeta::from_header(&header)) } } @@ -207,12 +202,7 @@ impl BlockProcessor for TraceProcessor { &self, (block, witness): Self::Input, ) -> std::result::Result { - let meta = BlockMeta { - block_number: block.header.number, - block_hash: block.header.hash, - post_state_root: block.header.state_root, - post_withdrawals_root: block.header.withdrawals_root.unwrap_or_default(), - }; + let meta = BlockMeta::from_header(&block.header); Ok(TraceProcessedBlock { block, witness, meta }) } } diff --git a/bin/debug-trace-server/src/metrics.rs b/bin/debug-trace-server/src/metrics.rs index 9b0a11dd..cfeb52dd 100644 --- a/bin/debug-trace-server/src/metrics.rs +++ b/bin/debug-trace-server/src/metrics.rs @@ -10,10 +10,9 @@ use std::net::SocketAddr; use eyre::Result; use metrics::{Counter, Gauge, Histogram, counter, histogram}; use metrics_derive::Metrics; -use metrics_exporter_prometheus::{Matcher, PrometheusBuilder}; pub use stateless_common::{ DEFAULT_METRICS_PORT, - metrics::{BYTE_BUCKETS, REORG_DEPTH_BUCKETS}, + metrics::{BYTE_BUCKETS, REORG_DEPTH_BUCKETS, install_prometheus_exporter}, }; /// Prefix for timed RPC method aliases. @@ -609,15 +608,7 @@ const BUCKET_SPECS: &[(&str, &[f64])] = &[ /// Initializes the Prometheus metrics exporter. pub fn init_metrics(addr: SocketAddr) -> Result<()> { - let builder = BUCKET_SPECS.iter().fold(PrometheusBuilder::new(), |b, &(name, buckets)| { - b.set_buckets_for_metric(Matcher::Full(name.to_owned()), buckets) - .expect("valid bucket config") - }); - - builder - .with_http_listener(addr) - .install() - .map_err(|e| eyre::eyre!("Failed to install metrics exporter: {}", e))?; + install_prometheus_exporter(addr, BUCKET_SPECS)?; // Pre-register all metrics pre_register_all_metrics(); diff --git a/bin/debug-trace-server/src/server_db.rs b/bin/debug-trace-server/src/server_db.rs index deb7d39f..41ac890e 100644 --- a/bin/debug-trace-server/src/server_db.rs +++ b/bin/debug-trace-server/src/server_db.rs @@ -308,15 +308,6 @@ pub(crate) mod test_support { pub tip_reads: AtomicUsize, } - impl ContractStore for StubBlockStore { - fn get_contracts(&self, _: &[B256]) -> StoreResult<(HashMap, Vec)> { - Ok((HashMap::default(), vec![])) - } - fn add_contracts(&self, _: &[(B256, Bytecode)]) -> StoreResult<()> { - Ok(()) - } - } - impl ChainStore for StubBlockStore { fn get_canonical_tip(&self) -> StoreResult> { self.tip_reads.fetch_add(1, Ordering::Relaxed); diff --git a/bin/stateless-validator/Cargo.toml b/bin/stateless-validator/Cargo.toml index 68491818..e8470ca6 100644 --- a/bin/stateless-validator/Cargo.toml +++ b/bin/stateless-validator/Cargo.toml @@ -37,9 +37,7 @@ bytes.workspace = true chrono = { workspace = true, features = ["clock"] } clap = { workspace = true, features = ["env"] } eyre.workspace = true -fastrand = { workspace = true, features = ["std"] } metrics.workspace = true -metrics-exporter-prometheus.workspace = true redb.workspace = true # rustls-tls gives the R2 witness client HTTPS without a system TLS backend. reqwest = { workspace = true, features = ["rustls-tls"] } diff --git a/bin/stateless-validator/src/app.rs b/bin/stateless-validator/src/app.rs index 9a396883..fedfd351 100644 --- a/bin/stateless-validator/src/app.rs +++ b/bin/stateless-validator/src/app.rs @@ -12,7 +12,7 @@ use stateless_core::{ChainStore, ContractStore, chain_spec::ChainSpec, db::Block use stateless_db::ContractCache; use tracing::{info, warn}; -use crate::{metrics, r2_witness::R2WitnessClient, validator_db::ValidatorDB, workers}; +use crate::{metrics, r2_witness::R2WitnessClient, runner, validator_db::ValidatorDB}; /// Where the validator sources witnesses from. #[derive(ValueEnum, Clone, Debug, PartialEq, Eq, Default)] @@ -235,7 +235,7 @@ pub struct CommandLineArgs { /// /// Parses CLI args, initializes tracing and metrics, constructs the RPC client and /// validator DB, loads or initializes the chain spec + anchor, then hands off to -/// [`workers::run_with_signals`]. +/// [`runner::run_with_signals`]. pub async fn run() -> Result<()> { let args = CommandLineArgs::parse(); let _log_guard = args.log.init_tracing()?; @@ -275,8 +275,6 @@ pub async fn run() -> Result<()> { ..rpc_defaults } .with_metrics(Arc::new(metrics::ValidatorMetrics)); - // In R2 mode the RpcClient's witness providers are never used, but its constructor requires - // a non-empty list — hand it the data endpoints as a placeholder. let data_apis: Vec<&str> = args.rpc_endpoint.iter().map(String::as_str).collect(); let r2_witness = match args.witness_source { WitnessSource::Rpc => { @@ -312,8 +310,10 @@ pub async fn run() -> Result<()> { } }; + // In R2 mode the client carries no witness providers: witnesses come straight from R2, + // and an accidental witness RPC call fails loudly instead of hitting the data endpoints. let witness_apis: Vec<&str> = if r2_witness.is_some() { - data_apis.clone() + Vec::new() } else { args.witness_endpoint.iter().map(String::as_str).collect() }; @@ -385,13 +385,13 @@ pub async fn run() -> Result<()> { info!(end_block = end, "Validating up to end block, then stopping"); } - let result = workers::run_with_signals( + let result = runner::run_with_signals( client, r2_witness, validator_db, contract_cache, chain_spec, - args.report_validation_endpoint, + args.report_validation_endpoint.is_some(), pipeline_config, ) .await; diff --git a/bin/stateless-validator/src/chain_sync.rs b/bin/stateless-validator/src/chain_sync.rs index 82cf2529..7010e2b0 100644 --- a/bin/stateless-validator/src/chain_sync.rs +++ b/bin/stateless-validator/src/chain_sync.rs @@ -4,7 +4,7 @@ //! [`ValidatorHooks`] (metrics integration) for the shared pipeline in //! [`stateless_core::pipeline::run_pipeline`]. -use std::{collections::HashSet, sync::Arc}; +use std::sync::Arc; use alloy_primitives::{B256, BlockHash, BlockNumber}; use alloy_rpc_types_eth::{Block, BlockId}; @@ -15,7 +15,7 @@ use salt::SaltWitness; use stateless_common::{CodeFetchError, RpcClient}; use stateless_core::{ chain_spec::ChainSpec, - data_types::iter_code_hashes, + data_types::collect_code_hashes, db::BlockMeta, executor::validate_block, pipeline::{BlockFetcher, BlockProcessor, ErrorAction, PipelineHooks, ProcessedBlock}, @@ -36,7 +36,6 @@ pub struct ValidatorFetcher { pub rpc_client: Arc, /// `Some` ⇒ fetch witnesses directly from R2; `None` ⇒ RPC. pub r2_witness: Option>, - pub on_remote_height: fn(u64), } impl BlockFetcher for ValidatorFetcher { @@ -63,7 +62,7 @@ impl BlockFetcher for ValidatorFetcher { async fn latest_block_number(&self) -> Result { let n = self.rpc_client.get_latest_block_number().await; - (self.on_remote_height)(n); + metrics::set_remote_chain_height(n); Ok(n) } @@ -73,12 +72,7 @@ impl BlockFetcher for ValidatorFetcher { async fn latest_block_meta(&self) -> Result { let header = self.rpc_client.get_header(BlockId::latest(), false).await; - Ok(BlockMeta { - block_number: header.number, - block_hash: header.hash, - post_state_root: header.state_root, - post_withdrawals_root: header.withdrawals_root.unwrap_or_default(), - }) + Ok(BlockMeta::from_header(&header)) } } @@ -186,8 +180,7 @@ impl BlockProcessor for ValidatorProcessor { // Resolve contract codes via the shared three-tier chain. Memory/disk hits // are trusted; the RPC tier verifies each bytecode's hash inside `get_codes`. - let codehashes: Vec = - iter_code_hashes(&task.salt_witness.kvs).collect::>().into_iter().collect(); + let codehashes: Vec = collect_code_hashes(&task.salt_witness.kvs); let (mut contracts, missing_contracts) = self .contract_cache .get(&codehashes) @@ -241,7 +234,6 @@ impl BlockProcessor for ValidatorProcessor { task.salt_witness, task.mpt_witness, &contracts, - None, ) }) .await @@ -313,15 +305,7 @@ mod tests { use stateless_core::pipeline::ProcessedBlock; use super::*; - - fn make_block_meta(num: u64) -> BlockMeta { - BlockMeta { - block_number: num, - block_hash: BlockHash::from([num as u8; 32]), - post_state_root: B256::from([(num.wrapping_add(100)) as u8; 32]), - post_withdrawals_root: B256::from([(num.wrapping_add(200)) as u8; 32]), - } - } + use crate::test_support::make_block_meta; #[test] fn test_verify_continuity_success() { diff --git a/bin/stateless-validator/src/lib.rs b/bin/stateless-validator/src/lib.rs index a5a8ea2b..22d2a11b 100644 --- a/bin/stateless-validator/src/lib.rs +++ b/bin/stateless-validator/src/lib.rs @@ -7,13 +7,30 @@ pub(crate) mod app; pub(crate) mod chain_sync; pub(crate) mod metrics; pub(crate) mod r2_witness; +pub(crate) mod runner; pub(crate) mod validator_db; -pub(crate) mod workers; pub use app::{ CommandLineArgs, VALIDATOR_DB_FILENAME, WitnessSource, load_or_create_chain_spec, run, }; pub use chain_sync::{ValidationTask, ValidatorFetcher, ValidatorHooks, ValidatorProcessor}; pub use r2_witness::{R2WitnessClient, R2WitnessError}; +pub use runner::run_with_signals; pub use validator_db::ValidatorDB; -pub use workers::run_with_signals; + +/// Fixtures shared by the unit tests of several modules. +#[cfg(test)] +pub(crate) mod test_support { + use alloy_primitives::{B256, BlockHash}; + use stateless_core::db::BlockMeta; + + /// A deterministic `BlockMeta` derived from `num` alone. + pub(crate) fn make_block_meta(num: u64) -> BlockMeta { + BlockMeta { + block_number: num, + block_hash: BlockHash::from([num as u8; 32]), + post_state_root: B256::from([(num.wrapping_add(100)) as u8; 32]), + post_withdrawals_root: B256::from([(num.wrapping_add(200)) as u8; 32]), + } + } +} diff --git a/bin/stateless-validator/src/metrics.rs b/bin/stateless-validator/src/metrics.rs index d3b141bb..f35b71ea 100644 --- a/bin/stateless-validator/src/metrics.rs +++ b/bin/stateless-validator/src/metrics.rs @@ -10,10 +10,12 @@ use std::{ use eyre::Result; use metrics::{counter, describe_counter, describe_gauge, describe_histogram, gauge, histogram}; -use metrics_exporter_prometheus::{Matcher, PrometheusBuilder}; pub use stateless_common::{ DEFAULT_METRICS_PORT, WitnessSizeBreakdown, - metrics::{BYTE_BUCKETS, REORG_DEPTH_BUCKETS, RpcAttemptOutcome, RpcMethod, RpcMetrics}, + metrics::{ + BYTE_BUCKETS, REORG_DEPTH_BUCKETS, RpcAttemptOutcome, RpcMethod, RpcMetrics, + install_prometheus_exporter, + }, }; use tracing::info; @@ -124,15 +126,7 @@ const BUCKET_SPECS: &[(&str, &[f64])] = &[ /// Initialize the Prometheus metrics exporter at the given address. pub fn init_metrics(addr: SocketAddr) -> Result<()> { - let builder = BUCKET_SPECS.iter().fold(PrometheusBuilder::new(), |b, &(name, buckets)| { - b.set_buckets_for_metric(Matcher::Full(name.to_owned()), buckets) - .expect("valid bucket config") - }); - - builder - .with_http_listener(addr) - .install() - .map_err(|e| eyre::eyre!("Failed to install Prometheus exporter: {}", e))?; + install_prometheus_exporter(addr, BUCKET_SPECS)?; register_metric_descriptions(); init_rpc_method_counters(); @@ -215,7 +209,6 @@ fn init_rpc_method_counters() { RpcMethod::EthGetBlock, RpcMethod::EthBlockNumber, RpcMethod::EthGetHeader, - RpcMethod::EthGetTransactionByHash, RpcMethod::MegaGetBlockWitness, RpcMethod::MegaSetValidatedBlocks, ]; diff --git a/bin/stateless-validator/src/r2_witness.rs b/bin/stateless-validator/src/r2_witness.rs index 83df0164..8c5173b0 100644 --- a/bin/stateless-validator/src/r2_witness.rs +++ b/bin/stateless-validator/src/r2_witness.rs @@ -221,8 +221,7 @@ impl R2WitnessClient { // self-imposed, and folded in it would masquerade as R2 slowness. let mut queue_wait = Duration::ZERO; let key = keys::block_object_key(number, hash); - let max_backoff_ms = self.retry_backoff.max.as_millis() as u64; - let mut backoff_ms = self.retry_backoff.initial.as_millis() as u64; + let mut backoff = self.retry_backoff.schedule(); let mut attempt = 0usize; let bytes = loop { @@ -242,18 +241,15 @@ impl R2WitnessClient { return Err(e); } metrics::on_r2_witness_retry(); - // Jittered doubling, mirroring the RPC retry loop: jitter keeps parallel - // validators (several typically slice a block range) from retrying in - // lockstep through a shared R2 brownout, and `.max(1)` keeps a - // zero-duration policy from busy-looping. - let jitter_ms = fastrand::u64(0..=backoff_ms / 2); - let sleep_ms = (backoff_ms + jitter_ms).min(max_backoff_ms).max(1); + // The shared jittered-doubling schedule (`BackoffPolicy::schedule`): jitter + // keeps parallel validators (several typically slice a block range) from + // retrying in lockstep through a shared R2 brownout. + let sleep_ms = backoff.next_sleep_ms(); warn!( number, %key, attempt, sleep_ms, error = %e, "R2 witness GET failed, backing off", ); tokio::time::sleep(Duration::from_millis(sleep_ms)).await; - backoff_ms = (backoff_ms * 2).min(max_backoff_ms); } } }; diff --git a/bin/stateless-validator/src/workers.rs b/bin/stateless-validator/src/runner.rs similarity index 97% rename from bin/stateless-validator/src/workers.rs rename to bin/stateless-validator/src/runner.rs index 8abd80ec..30b06a7e 100644 --- a/bin/stateless-validator/src/workers.rs +++ b/bin/stateless-validator/src/runner.rs @@ -21,7 +21,6 @@ use tracing::{debug, error, info, warn}; use crate::{ chain_sync::{ValidatorFetcher, ValidatorHooks, ValidatorProcessor}, - metrics, r2_witness::R2WitnessClient, validator_db::ValidatorDB, }; @@ -44,10 +43,9 @@ pub async fn run_with_signals( validator_db: Arc, contract_cache: Arc, chain_spec: Arc, - report_validation_endpoint: Option, + report_validation: bool, pipeline_config: PipelineConfig, ) -> Result<()> { - let report_validation = report_validation_endpoint.is_some(); let config = Arc::new(pipeline_config); let is_slice_run = config.sync_target.is_some(); info!( @@ -62,11 +60,7 @@ pub async fn run_with_signals( let mut sigterm = signal::unix::signal(signal::unix::SignalKind::terminate()) .map_err(|e| eyre::eyre!("Failed to register SIGTERM handler: {e}"))?; - let fetcher = Arc::new(ValidatorFetcher { - rpc_client: client.clone(), - r2_witness, - on_remote_height: metrics::set_remote_chain_height, - }); + let fetcher = Arc::new(ValidatorFetcher { rpc_client: client.clone(), r2_witness }); let processor = Arc::new(ValidatorProcessor { chain_spec, contract_cache, rpc_client: client.clone() }); let hooks = Arc::new(ValidatorHooks); diff --git a/bin/stateless-validator/src/validator_db.rs b/bin/stateless-validator/src/validator_db.rs index 0545b68b..66e2780d 100644 --- a/bin/stateless-validator/src/validator_db.rs +++ b/bin/stateless-validator/src/validator_db.rs @@ -58,18 +58,6 @@ impl ValidatorDB { Ok(Self { database, max_chain_length }) } - - #[cfg(test)] - fn set_anchor_block(&self, tip: &BlockMeta) -> StoreResult<()> { - use stateless_db::block_meta_to_tuple; - let write_txn = self.database.begin_write().store_err()?; - { - let mut table = write_txn.open_table(ANCHOR_BLOCK).store_err()?; - table.insert("anchor", block_meta_to_tuple(tip)).store_err()?; - } - write_txn.commit().store_err()?; - Ok(()) - } } impl ContractStore for ValidatorDB { @@ -158,6 +146,7 @@ mod tests { use stateless_db::ContractCache; use super::*; + use crate::test_support::make_block_meta; fn temp_store() -> (tempfile::TempDir, ValidatorDB) { let dir = tempfile::tempdir().unwrap(); @@ -165,15 +154,6 @@ mod tests { (dir, store) } - fn make_block_meta(number: u64) -> BlockMeta { - BlockMeta { - block_number: number, - block_hash: BlockHash::from([number as u8; 32]), - post_state_root: B256::from([(number + 100) as u8; 32]), - post_withdrawals_root: B256::from([(number + 200) as u8; 32]), - } - } - #[test] fn test_anchor_block_roundtrip() { let (_dir, store) = temp_store(); @@ -186,7 +166,8 @@ mod tests { post_state_root: B256::from([2u8; 32]), post_withdrawals_root: B256::from([3u8; 32]), }; - store.set_anchor_block(&tip).unwrap(); + // The production anchor write path — no test-only shortcut around the helper layer. + store.reset_to_anchor(&tip).unwrap(); let loaded = ChainStore::get_anchor(&store).unwrap().unwrap(); assert_eq!(loaded, tip); diff --git a/bin/stateless-validator/tests/integration.rs b/bin/stateless-validator/tests/integration.rs index 2d0a9b86..a37119e4 100644 --- a/bin/stateless-validator/tests/integration.rs +++ b/bin/stateless-validator/tests/integration.rs @@ -233,12 +233,48 @@ impl MockServerState { reject_reports: Arc::default(), } } + + /// Fixture block for a `0x…` hex block number, or the RPC error the handlers return + /// for unknown blocks. Shared by the by-number block and header handlers. + fn block_by_number_hex( + &self, + hex_number: &str, + ) -> Result<&Block, ErrorObject<'static>> { + let block_number = + u64::from_str_radix(hex_number.trim_start_matches("0x"), 16).unwrap_or(0); + self.fixtures + .block_numbers + .get(&block_number) + .and_then(|hash| self.fixtures.blocks.get(hash)) + .ok_or_else(|| { + make_rpc_error( + CALL_EXECUTION_FAILED_CODE, + format!("Block {block_number} not found"), + ) + }) + } + + /// Fixture block for a block hash, or the RPC error the handlers return for unknown + /// blocks. Shared by the by-hash block and header handlers. + fn block_by_hash( + &self, + hash: B256, + ) -> Result<&Block, ErrorObject<'static>> { + self.fixtures.blocks.get(&BlockHash::from(hash.0)).ok_or_else(|| { + make_rpc_error(CALL_EXECUTION_FAILED_CODE, format!("Block {hash} not found")) + }) + } } fn make_rpc_error(code: i32, msg: String) -> ErrorObject<'static> { ErrorObject::owned(code, msg, None::<()>) } +/// Invalid-params RPC error for a failed `params.parse()`. +fn invalid_params(e: impl std::fmt::Display) -> ErrorObject<'static> { + make_rpc_error(INVALID_PARAMS_CODE, format!("Invalid params: {e}")) +} + fn shape_block( block: &Block, full_block: bool, @@ -284,39 +320,17 @@ async fn setup_mock_rpc_server( module .register_method("eth_getBlockByNumber", |params, ctx, _| { - let (hex_number, full_block): (String, bool) = params - .parse() - .map_err(|e| make_rpc_error(INVALID_PARAMS_CODE, format!("Invalid params: {e}")))?; - let block_number = u64::from_str_radix(&hex_number[2..], 16).unwrap_or(0); - - let block = ctx - .fixtures - .block_numbers - .get(&block_number) - .and_then(|hash| ctx.fixtures.blocks.get(hash)) - .ok_or_else(|| { - make_rpc_error( - CALL_EXECUTION_FAILED_CODE, - format!("Block {block_number} not found"), - ) - })?; - + let (hex_number, full_block): (String, bool) = + params.parse().map_err(invalid_params)?; + let block = ctx.block_by_number_hex(&hex_number)?; Ok::<_, ErrorObject<'static>>(shape_block(block, full_block)) }) .unwrap(); module .register_method("eth_getBlockByHash", |params, ctx, _| { - let (hash, full_block): (B256, bool) = params - .parse() - .map_err(|e| make_rpc_error(INVALID_PARAMS_CODE, format!("Invalid params: {e}")))?; - - let block_hash = BlockHash::from(hash.0); - let block = ctx.fixtures.blocks.get(&block_hash).ok_or_else(|| { - make_rpc_error(CALL_EXECUTION_FAILED_CODE, format!("Block {hash} not found")) - })?; - - Ok::<_, ErrorObject<'static>>(shape_block(block, full_block)) + let (hash, full_block): (B256, bool) = params.parse().map_err(invalid_params)?; + Ok::<_, ErrorObject<'static>>(shape_block(ctx.block_by_hash(hash)?, full_block)) }) .unwrap(); @@ -329,45 +343,21 @@ async fn setup_mock_rpc_server( module .register_method("eth_getHeaderByNumber", |params, ctx, _| { - let (hex_number,): (String,) = params.parse().unwrap(); - let block_number = u64::from_str_radix(&hex_number[2..], 16).unwrap_or(0); - - let block = ctx - .fixtures - .block_numbers - .get(&block_number) - .and_then(|hash| ctx.fixtures.blocks.get(hash)) - .ok_or_else(|| { - make_rpc_error( - CALL_EXECUTION_FAILED_CODE, - format!("Block {block_number} not found"), - ) - })?; - - Ok::<_, ErrorObject<'static>>(block.header.clone()) + let (hex_number,): (String,) = params.parse().map_err(invalid_params)?; + Ok::<_, ErrorObject<'static>>(ctx.block_by_number_hex(&hex_number)?.header.clone()) }) .unwrap(); module .register_method("eth_getHeaderByHash", |params, ctx, _| { - let (hash,): (B256,) = params - .parse() - .map_err(|e| make_rpc_error(INVALID_PARAMS_CODE, format!("Invalid params: {e}")))?; - - let block_hash = BlockHash::from(hash.0); - let block = ctx.fixtures.blocks.get(&block_hash).ok_or_else(|| { - make_rpc_error(CALL_EXECUTION_FAILED_CODE, format!("Block {hash} not found")) - })?; - - Ok::<_, ErrorObject<'static>>(block.header.clone()) + let (hash,): (B256,) = params.parse().map_err(invalid_params)?; + Ok::<_, ErrorObject<'static>>(ctx.block_by_hash(hash)?.header.clone()) }) .unwrap(); module .register_method("eth_getCodeByHash", |params, ctx, _| { - let (hash,): (B256,) = params - .parse() - .map_err(|e| make_rpc_error(INVALID_PARAMS_CODE, format!("Invalid params: {e}")))?; + let (hash,): (B256,) = params.parse().map_err(invalid_params)?; let code = ctx.fixtures.contracts.get(&hash).cloned().unwrap_or_default(); Ok::<_, ErrorObject<'static>>(code.original_bytes()) @@ -376,9 +366,7 @@ async fn setup_mock_rpc_server( module .register_method("mega_getBlockWitness", |params, ctx, _| { - let (keys,): (WitnessRequestKeys,) = params - .parse() - .map_err(|e| make_rpc_error(INVALID_PARAMS_CODE, format!("Invalid params: {e}")))?; + let (keys,): (WitnessRequestKeys,) = params.parse().map_err(invalid_params)?; let block_hash = BlockHash::from(keys.block_hash.0); let salt_witness = @@ -461,11 +449,7 @@ async fn integration_test() { let config = Arc::new(cfg); let shutdown = CancellationToken::new(); - let fetcher = Arc::new(ValidatorFetcher { - rpc_client: client.clone(), - r2_witness: None, - on_remote_height: |_| {}, - }); + let fetcher = Arc::new(ValidatorFetcher { rpc_client: client.clone(), r2_witness: None }); let processor = Arc::new(ValidatorProcessor { chain_spec, contract_cache, rpc_client: client }); let hooks = Arc::new(ValidatorHooks); @@ -533,7 +517,7 @@ async fn run_end_block_slice( Arc::clone(&validator_db), contract_cache, chain_spec, - Some(url.clone()), + true, cfg, ) .await; diff --git a/crates/stateless-common/Cargo.toml b/crates/stateless-common/Cargo.toml index 143e2664..260a048c 100644 --- a/crates/stateless-common/Cargo.toml +++ b/crates/stateless-common/Cargo.toml @@ -35,6 +35,7 @@ clap.workspace = true eyre.workspace = true fastrand = { workspace = true, features = ["std"] } futures.workspace = true +metrics-exporter-prometheus.workspace = true # Enables gzip/brotli on alloy-provider's reqwest 0.12 (Cargo feature unification) for witness/data fetches; not referenced in code. reqwest = { workspace = true, features = ["gzip", "brotli"] } rolling-file.workspace = true diff --git a/crates/stateless-common/src/lib.rs b/crates/stateless-common/src/lib.rs index 1c3356fc..254f53b5 100644 --- a/crates/stateless-common/src/lib.rs +++ b/crates/stateless-common/src/lib.rs @@ -3,8 +3,8 @@ pub mod metrics; pub use metrics::{RpcMethod, RpcMetrics}; pub mod rpc_client; pub use rpc_client::{ - BackoffPolicy, CodeFetchError, RpcClient, RpcClientConfig, RpcDeadlineExceeded, - SetValidatedBlocksResponse, WitnessRequestKeys, + BackoffPolicy, BackoffSchedule, CodeFetchError, RpcClient, RpcClientConfig, + RpcDeadlineExceeded, SetValidatedBlocksResponse, WitnessRequestKeys, }; pub mod witness_encoding; pub use witness_encoding::{ diff --git a/crates/stateless-common/src/metrics.rs b/crates/stateless-common/src/metrics.rs index 278e5c30..753d5e60 100644 --- a/crates/stateless-common/src/metrics.rs +++ b/crates/stateless-common/src/metrics.rs @@ -1,10 +1,35 @@ //! RPC metrics types shared by both binaries. //! -//! Provides [`RpcMethod`] for identifying RPC calls and [`RpcMetrics`] as a -//! callback trait for tracking RPC performance. +//! Provides [`RpcMethod`] for identifying RPC calls, [`RpcMetrics`] as a +//! callback trait for tracking RPC performance, and the shared Prometheus +//! exporter installer ([`install_prometheus_exporter`]). + +use std::net::SocketAddr; + +use metrics_exporter_prometheus::{Matcher, PrometheusBuilder}; use crate::witness_size::WitnessSizeBreakdown; +/// Installs the Prometheus exporter with an HTTP listener on `addr`, applying the given +/// per-metric histogram buckets (`(metric_name, buckets)` pairs) before install. +/// +/// Shared by both binaries; each keeps its own metric names, descriptions, and +/// pre-registration after this returns. +pub fn install_prometheus_exporter( + addr: SocketAddr, + bucket_specs: &[(&str, &[f64])], +) -> eyre::Result<()> { + let builder = bucket_specs.iter().fold(PrometheusBuilder::new(), |b, &(name, buckets)| { + b.set_buckets_for_metric(Matcher::Full(name.to_owned()), buckets) + .expect("valid bucket config") + }); + + builder + .with_http_listener(addr) + .install() + .map_err(|e| eyre::eyre!("Failed to install Prometheus exporter: {e}")) +} + /// Byte-size histogram buckets: 1 KB, 10 KB, 50 KB, 200 KB, 1 MB, 5 MB, 20 MB. pub const BYTE_BUCKETS: &[f64] = &[1_024.0, 10_240.0, 51_200.0, 204_800.0, 1_048_576.0, 5_242_880.0, 20_971_520.0]; diff --git a/crates/stateless-common/src/rpc_client.rs b/crates/stateless-common/src/rpc_client.rs index 893c06bb..542d9f8b 100644 --- a/crates/stateless-common/src/rpc_client.rs +++ b/crates/stateless-common/src/rpc_client.rs @@ -55,11 +55,11 @@ use crate::{ witness_size::WitnessSizeBreakdown, }; -/// Exponential-backoff policy used by [`RpcClient`]'s round-level retry loop. +/// Exponential-backoff policy used by [`RpcClient`]'s round-level retry loop and the +/// validator's R2 witness retry loop. /// -/// `initial` is the first sleep duration; each round doubles it up to `max`. -/// The loop itself lives in [`round_robin_with_backoff`]; this type only describes -/// the sleep schedule. +/// `initial` is the first sleep duration; each retry doubles it up to `max`. Retry loops +/// execute the schedule via [`Self::schedule`], which owns the jitter/cap/floor invariants. #[derive(Debug, Clone)] pub struct BackoffPolicy { /// First retry sleep. Each subsequent retry doubles up to `max`. @@ -73,6 +73,36 @@ impl BackoffPolicy { pub const fn new(initial: Duration, max: Duration) -> Self { Self { initial, max } } + + /// Starts executing the schedule from `initial`. + pub fn schedule(&self) -> BackoffSchedule { + BackoffSchedule { + current_ms: self.initial.as_millis() as u64, + max_ms: self.max.as_millis() as u64, + } + } +} + +/// Stepping state for a [`BackoffPolicy`]'s sleep schedule. +/// +/// Owns the three invariants every consumer relies on: up to 50% random jitter per sleep +/// (keeps parallel clients from retrying in lockstep through a shared outage), the `max` +/// cap, and a 1 ms floor so a zero-duration policy cannot busy-loop a retry loop. +#[derive(Debug)] +pub struct BackoffSchedule { + current_ms: u64, + max_ms: u64, +} + +impl BackoffSchedule { + /// Returns the next sleep in milliseconds (jittered, capped, floored at 1 ms) and + /// advances the doubling state. + pub fn next_sleep_ms(&mut self) -> u64 { + let jitter_ms = fastrand::u64(0..=self.current_ms / 2); + let sleep_ms = (self.current_ms + jitter_ms).min(self.max_ms).max(1); + self.current_ms = (self.current_ms * 2).min(self.max_ms); + sleep_ms + } } /// Error returned by the `_with_deadline` RPC methods when a caller-supplied @@ -266,7 +296,10 @@ impl RpcClient { /// # Arguments /// * `data_apis` - HTTP URLs of the standard JSON-RPC endpoints for blocks and contract data /// (tried in order, non-empty) - /// * `witness_apis` - HTTP URLs of the witness RPC endpoints (tried in order, non-empty) + /// * `witness_apis` - HTTP URLs of the witness RPC endpoints (tried in order). May be empty + /// when the deployment sources witnesses elsewhere (e.g. the validator's R2 witness mode); a + /// witness call issued with no providers configured panics (see `witness_round_robin`) + /// instead of silently retrying against the wrong endpoints. /// * `config` - Configuration controlling verification, retry, and concurrency behavior /// * `report_api` - Optional HTTP URL of the endpoint for reporting validated blocks pub fn new_with_config( @@ -278,9 +311,6 @@ impl RpcClient { if data_apis.is_empty() { return Err(eyre!("At least one data API URL must be provided")); } - if witness_apis.is_empty() { - return Err(eyre!("At least one witness API URL must be provided")); - } let data_providers = data_apis .iter() @@ -471,10 +501,18 @@ impl RpcClient { self.call_with_deadline(RpcMethod::EthGetBlock, deadline, move |provider| { Box::pin(async move { let block = do_get_block_unchecked(&provider, block_id, full_txs).await?; - if verify { - verify_block_integrity(&block)?; + if !verify { + return Ok(block); } - Ok(block) + // Per-tx ECDSA recovery + re-encoding over a full block is CPU-bound — run it + // on the blocking pool (like the witness decode in `fetch_witness_with`) + // instead of pinning an async runtime worker for the duration. + tokio::task::spawn_blocking(move || { + verify_block_integrity(&block)?; + Ok(block) + }) + .await + .context("block verification task panicked")? }) }) .await @@ -699,6 +737,11 @@ impl RpcClient { decode: fn(&str) -> std::result::Result, trace_msg: &'static str, ) -> std::result::Result { + assert!( + !self.witness_providers.is_empty(), + "witness call issued with no witness providers configured — this client was \ + constructed for a deployment that sources witnesses elsewhere (R2 witness mode)" + ); assert!( !providers.is_empty() && providers.end <= self.witness_providers.len(), "witness provider range ({providers:?}) must select at least one of {} providers", @@ -984,9 +1027,7 @@ where const WARN_AT_ROUND: u32 = 3; let n = providers.len(); - let max_backoff_ms = policy.max.as_millis() as u64; - let initial_backoff_ms = policy.initial.as_millis() as u64; - let mut round_backoff_ms = initial_backoff_ms; + let mut backoff = policy.schedule(); let mut round = 0u32; let call_start = Instant::now(); @@ -1117,11 +1158,7 @@ where // `last_err` is always `Some` here: `n >= 1` is enforced by the `RpcClient` // constructor and we only reach this point after `n` iterations that each set it. let last_err = last_err.expect("last_err set when every provider failed this round"); - let jitter_ms = fastrand::u64(0..=round_backoff_ms / 2); - // `.max(1)` prevents a hot-spin loop if a caller constructs a zero-backoff policy - // (`BackoffPolicy::new(Duration::ZERO, Duration::ZERO)`): the computed sleep would - // otherwise be `0` and the retry loop would busy-wait on every round. - let mut sleep_ms = (round_backoff_ms + jitter_ms).min(max_backoff_ms).max(1); + let mut sleep_ms = backoff.next_sleep_ms(); // Clamp the sleep so it doesn't overshoot the caller's deadline — if no time is // left we bail immediately rather than sleeping past the deadline and then bailing. if let Some(d) = deadline { @@ -1141,7 +1178,6 @@ where "All providers failed this round, backing off", ); tokio::time::sleep(std::time::Duration::from_millis(sleep_ms)).await; - round_backoff_ms = (round_backoff_ms * 2).min(max_backoff_ms); round += 1; } } @@ -1305,13 +1341,19 @@ fn verify_block_integrity(block: &Block) -> Result<()> { // Verify transaction hashes and transactions root if let BlockTransactions::Full(ref transactions) = block.transactions { + // Encode each envelope exactly once: keccak of the encoding is the tx hash, and the + // same bytes feed the ordered trie for the transactions-root check. + let mut encoded_txs: Vec> = Vec::with_capacity(transactions.len()); for tx in transactions { - let tx_envelope = tx.inner.clone().into_inner(); + let tx_envelope = tx.inner.inner.inner(); + let mut encoded = Vec::with_capacity(tx_envelope.encode_2718_len()); + tx_envelope.encode_2718(&mut encoded); + let computed_hash = alloy_primitives::keccak256(&encoded); ensure!( - tx_envelope.trie_hash() == *tx_envelope.hash(), + computed_hash == *tx_envelope.hash(), "Transaction hash mismatch: expected {:?}, computed {:?}", tx_envelope.hash(), - tx_envelope.trie_hash() + computed_hash ); let recovered = tx_envelope @@ -1324,10 +1366,11 @@ fn verify_block_integrity(block: &Block) -> Result<()> { tx.from(), recovered ); + encoded_txs.push(encoded); } - let computed_tx_root = ordered_trie_root_with_encoder(transactions, |tx, buf| { - tx.inner.clone().into_inner().encode_2718(buf) + let computed_tx_root = ordered_trie_root_with_encoder(&encoded_txs, |tx_bytes, buf| { + buf.extend_from_slice(tx_bytes) }); ensure!( computed_tx_root == block.header.transactions_root, @@ -1520,11 +1563,12 @@ mod tests { .to_string() .contains("At least one data API") ); - assert!( - RpcClient::new(&[LOCALHOST_A], &[]) - .unwrap_err() - .to_string() - .contains("At least one witness API") + // An empty witness list is a legal configuration (the validator's R2 witness mode); + // witness calls on such a client panic instead (see `witness_round_robin`). + assert_eq!( + RpcClient::new(&[LOCALHOST_A], &[]).unwrap().witness_provider_count(), + 0, + "empty witness list must construct" ); for endpoints in [&[LOCALHOST_B][..], &[LOCALHOST_B, "http://localhost:8547"]] { diff --git a/crates/stateless-core/src/chain_spec.rs b/crates/stateless-core/src/chain_spec.rs index 11413124..900184f6 100644 --- a/crates/stateless-core/src/chain_spec.rs +++ b/crates/stateless-core/src/chain_spec.rs @@ -15,9 +15,6 @@ use mega_evm::{ use reth_ethereum_forks::ChainHardforks; use reth_optimism_chainspec::OpChainSpec; -/// Default blob gas price update fraction for Cancun (from EIP-4844) -pub const BLOB_GASPRICE_UPDATE_FRACTION: u64 = 3338477; - /// Chain specification for the Optimism network. /// /// Defines when various Ethereum and Optimism hardforks are activated. @@ -74,10 +71,8 @@ impl ChainSpec { /// Ordering rules: /// - [`OpChainSpec`] already yields Optimism/Ethereum hardforks in the correct order, so they /// do not require reordering. - /// - MegaETH hardforks are extracted from the genesis `extra_fields` and explicitly ordered to - /// match the canonical sequence defined by [`mega_mainnet_hardforks()`]. Any remaining, - /// unknown MegaETH hardforks are preserved and appended after the known ones so nothing is - /// dropped. + /// - MegaETH hardforks are extracted from the genesis `extra_fields`; + /// [`MegaethGenesisHardforks::into_vec`] yields them in canonical activation order. /// - The MegaETH set is then merged with the Optimism/Ethereum set to build a single /// [`ChainHardforks`] that drives fork activation. /// @@ -120,7 +115,7 @@ impl ChainSpec { ); } - let mut megaeth_hardforks = megaeth_hardforks.into_vec(); + let megaeth_hardforks = megaeth_hardforks.into_vec(); // Rex5 SequencerRegistry bootstrap, required iff `rex5Time` is scheduled. Parsed from // the same flat schema mega-reth uses (`rex5InitialSequencer` / `rex5InitialAdmin` as @@ -167,20 +162,9 @@ impl ChainSpec { .map(|(f, b)| (dyn_clone::clone_box(f), b)) .collect(); - let hardfork_order = mega_mainnet_hardforks(); - let mut all_hardforks = Vec::with_capacity(op_hardforks.len() + megaeth_hardforks.len()); - for (order, _) in hardfork_order.forks_iter() { - if let Some(mega_hardfork_index) = - megaeth_hardforks.iter().position(|(hardfork, _)| **hardfork == *order) - { - all_hardforks.push(megaeth_hardforks.remove(mega_hardfork_index)); - } - } - - // append the remaining unknown hardforks to ensure we don't filter any out - all_hardforks.append(&mut megaeth_hardforks); - - // we merge megaeth_hardforks with op_hardforks + // `into_vec` yields the MegaETH hardforks already in canonical activation order, + // so the merge is a straight concatenation. + let mut all_hardforks = megaeth_hardforks; all_hardforks.append(&mut op_hardforks); Self { @@ -227,6 +211,10 @@ impl MegaethGenesisHardforks { } /// Convert the MegaETH genesis hardforks into a vector of hardforks and their conditions. + /// + /// The literal below is the single source of the canonical MegaETH activation order — + /// [`ChainSpec::from_genesis`] merges it as-is, so new hardforks must be inserted at + /// their activation position. pub fn into_vec(self) -> Vec<(Box, ForkCondition)> { vec![ (MegaHardfork::MiniRex.boxed(), self.mini_rex_time.map(ForkCondition::Timestamp)), @@ -303,22 +291,6 @@ impl MegaethGenesisSequencerRegistryRex6Config { } } -/// Build a fresh `ChainHardforks` describing MegaETH's canonical hardfork sequence. -pub fn mega_mainnet_hardforks() -> ChainHardforks { - ChainHardforks::new(vec![ - (MegaHardfork::MiniRex.boxed(), ForkCondition::Timestamp(0)), - (MegaHardfork::MiniRex1.boxed(), ForkCondition::Timestamp(0)), - (MegaHardfork::MiniRex2.boxed(), ForkCondition::Timestamp(0)), - (MegaHardfork::Rex.boxed(), ForkCondition::Timestamp(0)), - (MegaHardfork::Rex1.boxed(), ForkCondition::Timestamp(0)), - (MegaHardfork::Rex2.boxed(), ForkCondition::Timestamp(0)), - (MegaHardfork::Rex3.boxed(), ForkCondition::Timestamp(0)), - (MegaHardfork::Rex4.boxed(), ForkCondition::Timestamp(0)), - (MegaHardfork::Rex5.boxed(), ForkCondition::Timestamp(0)), - (MegaHardfork::Rex6.boxed(), ForkCondition::Timestamp(0)), - ]) -} - #[cfg(test)] mod tests { use std::string::ToString; diff --git a/crates/stateless-core/src/data_types.rs b/crates/stateless-core/src/data_types.rs index 47905aa3..50abec53 100644 --- a/crates/stateless-core/src/data_types.rs +++ b/crates/stateless-core/src/data_types.rs @@ -26,7 +26,7 @@ use std::{collections::BTreeMap, vec::Vec}; pub use alloy_primitives::Bytes; -use alloy_primitives::{Address, B256, U256}; +use alloy_primitives::{Address, B256, FixedBytes, U256}; use revm::primitives::KECCAK_EMPTY; use salt::{SaltKey, SaltValue}; @@ -67,14 +67,33 @@ impl PlainKey { /// - Unknown: preserved raw bytes from decode pub fn encode(&self) -> Vec { match self { - PlainKey::Account(addr) => addr.as_slice().to_vec(), - PlainKey::Storage(addr, slot) => { - addr.concat_const::(*slot).as_slice().to_vec() - } + PlainKey::Account(addr) => Self::account_key_bytes(addr).to_vec(), + PlainKey::Storage(addr, slot) => Self::storage_key_bytes(*addr, *slot).to_vec(), PlainKey::Unknown(data) => data.clone(), } } + /// Encoding of an account key — the raw address bytes. + /// + /// Same bytes as `PlainKey::Account(address).encode()` without the heap allocation, + /// for per-state-read hot paths. + #[inline] + pub(crate) fn account_key_bytes(address: &Address) -> &[u8] { + address.as_slice() + } + + /// Stack-allocated encoding of a storage-slot key — address (20) ++ slot (32). + /// + /// Same bytes as `PlainKey::Storage(address, slot).encode()` without the heap + /// allocation, for per-state-read hot paths. + #[inline] + pub(crate) fn storage_key_bytes( + address: Address, + slot: B256, + ) -> FixedBytes { + address.concat_const::(slot) + } + /// Decodes a byte slice into a PlainKey. /// /// Returns `PlainKey::Unknown` if the buffer length is neither 20 (account) diff --git a/crates/stateless-core/src/db.rs b/crates/stateless-core/src/db.rs index f75f2626..8df3769f 100644 --- a/crates/stateless-core/src/db.rs +++ b/crates/stateless-core/src/db.rs @@ -29,6 +29,21 @@ pub struct BlockMeta { pub post_withdrawals_root: B256, } +impl BlockMeta { + /// Projects an RPC header into the meta of the block it seals — a header's roots are that + /// block's post-state. A missing `withdrawals_root` defaults to zero, the tip-observation + /// policy both binaries use; callers that must instead *reject* such headers (e.g. anchor + /// initialization from an operator-supplied hash) build the meta explicitly. + pub fn from_header(header: &alloy_rpc_types_eth::Header) -> Self { + Self { + block_number: header.number, + block_hash: header.hash, + post_state_root: header.state_root, + post_withdrawals_root: header.withdrawals_root.unwrap_or_default(), + } + } +} + /// Errors returned by persistence trait methods. /// /// This is the single typed error at the library/binary boundary: every @@ -107,7 +122,9 @@ pub trait ContractStore: Send + Sync { /// pipeline's [`ReorgResolver`](crate::pipeline::ReorgResolver) seam, which each scenario supplies. /// History-owning stores additionally implement /// [`DivergenceLookups`](crate::pipeline::DivergenceLookups) so the pipeline can bisect them. -pub trait ChainStore: ContractStore { +/// Deliberately independent of [`ContractStore`]: a chain-cursor store (e.g. an embedder whose +/// bytecode integrity is enforced at ingest) need not stub contract persistence. +pub trait ChainStore: Send + Sync { fn get_canonical_tip(&self) -> StoreResult>; fn get_anchor(&self) -> StoreResult>; fn advance_chain(&self, blocks: &[BlockMeta]) -> StoreResult<()>; diff --git a/crates/stateless-core/src/evm_database.rs b/crates/stateless-core/src/evm_database.rs index abdf2f3e..1618a1ec 100644 --- a/crates/stateless-core/src/evm_database.rs +++ b/crates/stateless-core/src/evm_database.rs @@ -5,9 +5,9 @@ //! validation. use std::{ + collections::BTreeMap, format, string::{String, ToString}, - vec::Vec, }; use alloy_consensus::Header; @@ -78,10 +78,16 @@ where W: StateReader, W::Error: core::fmt::Display, { - /// Get value from witness for the given plain key - fn plain_value(&self, plain_key: &[u8]) -> Result>, WitnessDatabaseError> { + /// Get the witness entry for the given plain key. + /// + /// Returns the whole `SaltValue` (an inline array) instead of going through salt's + /// `plain_value()`, which heap-allocates a copy of the value bytes on every read — + /// this lookup runs once per unique account/slot touched during block replay. + /// Callers decode from `.value()` in place. + fn find(&self, plain_key: &[u8]) -> Result, WitnessDatabaseError> { EphemeralSaltState::new(self.witness) - .plain_value(plain_key) + .find(plain_key) + .map(|found| found.map(|(_, salt_value)| salt_value)) .map_err(|e| WitnessDatabaseError(e.to_string())) } } @@ -97,9 +103,9 @@ where fn basic_ref(&self, address: Address) -> Result, Self::Error> { trace!(?address, "basic_ref"); - let raw_value = self.plain_value(&PlainKey::Account(address).encode())?; + let salt_value = self.find(PlainKey::account_key_bytes(&address))?; - match raw_value.and_then(|v| match PlainValue::decode(&v) { + match salt_value.and_then(|v| match PlainValue::decode(v.value()) { PlainValue::Account(acc) => Some(acc), _ => None, }) { @@ -133,10 +139,11 @@ where fn storage_ref(&self, address: Address, index: U256) -> Result { trace!(?address, index = %format_args!("{:#x}", index), "storage_ref"); - let raw_value = self.plain_value(&PlainKey::Storage(address, index.into()).encode())?; + let salt_value = + self.find(PlainKey::storage_key_bytes(address, index.into()).as_slice())?; - Ok(raw_value - .and_then(|v| match PlainValue::decode(&v) { + Ok(salt_value + .and_then(|v| match PlainValue::decode(v.value()) { PlainValue::Storage(value) => Some(value), _ => None, }) @@ -225,8 +232,16 @@ impl WitnessExternalEnv { salt_witness: &SaltWitness, block_number: BlockNumber, ) -> Result { - let bucket_capacities = salt_witness - .kvs + Self::from_metadata_kvs(&salt_witness.kvs, block_number) + } + + /// Shared constructor body: scans the metadata key range of a witness's `kvs` map and + /// collects the bucket capacities (both witness types expose the same map layout). + fn from_metadata_kvs( + kvs: &BTreeMap>, + block_number: BlockNumber, + ) -> Result { + let bucket_capacities = kvs .range(METADATA_KEYS_RANGE) .map(|(key, value)| Self::parse_metadata_entry(key, value)) .collect::, _>>()?; @@ -260,13 +275,7 @@ impl WitnessExternalEnv { light_witness: &LightWitness, block_number: BlockNumber, ) -> Result { - let bucket_capacities = light_witness - .kvs - .range(METADATA_KEYS_RANGE) - .map(|(key, value)| Self::parse_metadata_entry(key, value)) - .collect::, _>>()?; - - Ok(Self { block_number, bucket_capacities }) + Self::from_metadata_kvs(&light_witness.kvs, block_number) } } @@ -295,11 +304,11 @@ impl SaltEnv for WitnessExternalEnv { } fn bucket_id_for_account(account: Address) -> BucketId { - hasher::bucket_id(&PlainKey::Account(account).encode()) + hasher::bucket_id(PlainKey::account_key_bytes(&account)) } fn bucket_id_for_slot(address: Address, key: U256) -> BucketId { - hasher::bucket_id(&PlainKey::Storage(address, key.into()).encode()) + hasher::bucket_id(PlainKey::storage_key_bytes(address, key.into()).as_slice()) } } diff --git a/crates/stateless-core/src/executor.rs b/crates/stateless-core/src/executor.rs index daee20fd..f2371d81 100644 --- a/crates/stateless-core/src/executor.rs +++ b/crates/stateless-core/src/executor.rs @@ -28,9 +28,9 @@ //! The module integrates with the Salt witness system for state reconstruction //! and uses Revm for transaction execution. -use std::{boxed::Box, collections::BTreeMap, fmt::Debug, vec::Vec}; #[cfg(feature = "std")] -use std::{io::Write, time::Instant}; +use std::time::Instant; +use std::{boxed::Box, collections::BTreeMap, fmt::Debug, vec::Vec}; use alloy_consensus::{TxReceipt, proofs::calculate_receipt_root, transaction::Recovered}; use alloy_eips::eip2718::Encodable2718; @@ -50,13 +50,11 @@ use mega_evm::{ }; use op_alloy_consensus::OpTxEnvelope; use op_alloy_rpc_types::Transaction as OpTransaction; -#[cfg(feature = "std")] -use revm::inspector::inspectors::TracerEip3155; use revm::{ DatabaseRef, context::{BlockEnv, CfgEnv}, database::states::{BundleAccount, StateBuilder, bundle_state::BundleRetention}, - primitives::{B256, KECCAK_EMPTY, U256}, + primitives::{B256, KECCAK_EMPTY, U256, eip4844::BLOB_BASE_FEE_UPDATE_FRACTION_CANCUN}, state::Bytecode, }; use salt::{EphemeralSaltState, SaltValue, SaltWitness, StateRoot, StateUpdates, Witness}; @@ -64,7 +62,7 @@ use thiserror::Error; use tracing::debug; use crate::{ - chain_spec::{BLOB_GASPRICE_UPDATE_FRACTION, ChainSpec}, + chain_spec::ChainSpec, data_types::{Account, PlainKey, PlainValue}, evm_database::{WitnessDatabase, WitnessDatabaseError, WitnessExternalEnv}, withdrawals::{self, ADDRESS_L2_TO_L1_MESSAGE_PASSER, MptWitness}, @@ -258,10 +256,6 @@ impl ValidationOptions { /// - Chain configuration with appropriate spec ID for the block number /// - Block environment with gas limits, timestamps, and fee parameters /// - Blob gas pricing if excess blob gas is present in the header -/// -/// Creates an EVM environment from a block header and chain specification. -/// -/// This function sets up the configuration and block environment needed for EVM execution. pub fn create_evm_env( header: &alloy_consensus::Header, chain_spec: &ChainSpec, @@ -281,7 +275,8 @@ pub fn create_evm_env( }; if let Some(excess_blob_gas) = header.excess_blob_gas { - block_env.set_blob_excess_gas_and_price(excess_blob_gas, BLOB_GASPRICE_UPDATE_FRACTION); + block_env + .set_blob_excess_gas_and_price(excess_blob_gas, BLOB_BASE_FEE_UPDATE_FRACTION_CANCUN); } EvmEnv::new(cfg_env, block_env) @@ -373,7 +368,6 @@ pub fn replay_block( block: &B, db: &DB, env_oracle: ENV, - #[cfg(feature = "std")] trace_writer: Option>, ) -> Result<(HashMap, BlockExecutionOutput), ValidationError> where B: BlockInput, @@ -417,28 +411,9 @@ where block_limits, ); - // Plain execution path, shared by the non-tracer std branch and the no_std build. - // Extracted as a closure so the body lives in one place — any future change to the - // non-tracer path only needs to be made here. - let run_plain = |state: &mut _, ctx, env| { - let executor = executor_factory.create_executor(state, ctx, env); - execute_transactions(executor, block.txs_recovered()) - }; - - #[cfg(feature = "std")] - let (receipts_root, logs_bloom, gas_used) = if let Some(writer) = trace_writer { - let executor = executor_factory.create_executor_with_inspector( - &mut state, - execution_context, - evm_env, - TracerEip3155::new(writer), - ); - execute_transactions(executor, block.txs_recovered())? - } else { - run_plain(&mut state, execution_context, evm_env)? - }; - #[cfg(not(feature = "std"))] - let (receipts_root, logs_bloom, gas_used) = run_plain(&mut state, execution_context, evm_env)?; + let executor = executor_factory.create_executor(&mut state, execution_context, evm_env); + let (receipts_root, logs_bloom, gas_used) = + execute_transactions(executor, block.txs_recovered())?; // Merge transitions into bundle_state state.merge_transitions(BundleRetention::PlainState); @@ -496,8 +471,7 @@ where let logs_bloom = execution_result.receipts.iter().fold(Bloom::ZERO, |acc, receipt| acc | receipt.bloom()); - // Gas used is the cumulative gas used of the last receipt - let gas_used = execution_result.receipts.last().map(|r| r.cumulative_gas_used()).unwrap_or(0); + let gas_used = execution_result.gas_used; let receipts_root = calculate_receipt_root(&execution_result.receipts); @@ -667,7 +641,6 @@ fn verify_and_replay( block: &B, salt_witness: SaltWitness, contracts: &HashMap, - #[cfg(feature = "std")] writer: Option>, ) -> Result { let header = block.consensus_header(); @@ -685,14 +658,7 @@ fn verify_and_replay( // Replay block transactions let ((accounts, output), block_replay_time) = timed(|| { let witness_db = WitnessDatabase { header, witness: &witness, contracts }; - replay_block( - chain_spec, - block, - &witness_db, - ext_env, - #[cfg(feature = "std")] - writer, - ) + replay_block(chain_spec, block, &witness_db, ext_env) })?; let stats = ValidationStats { @@ -722,8 +688,6 @@ fn verify_and_replay( /// * `salt_witness` - The salt witness data needed for state validation /// * `mpt_witness` - The MPT witness data for withdrawal verification /// * `contracts` - Contract bytecode cache for transaction execution -/// * `writer` - Optional writer for EIP-3155 trace output. When provided, enables step-by-step EVM -/// execution tracing in EIP-3155 format. /// /// # Returns /// @@ -735,7 +699,6 @@ pub fn validate_block( salt_witness: SaltWitness, mpt_witness: MptWitness, contracts: &HashMap, - #[cfg(feature = "std")] writer: Option>, ) -> Result { // A block carrying only transaction hashes can't be replayed — fail fast before paying // the witness proof verification. `replay_block` re-checks for direct callers. @@ -745,14 +708,8 @@ pub fn validate_block( let header = block.consensus_header(); // Verify the witness proof and replay the block's transactions over it - let VerifiedReplay { witness, accounts, output, mut stats } = verify_and_replay( - chain_spec, - block, - salt_witness, - contracts, - #[cfg(feature = "std")] - writer, - )?; + let VerifiedReplay { witness, accounts, output, mut stats } = + verify_and_replay(chain_spec, block, salt_witness, contracts)?; // Extract and hash storage updates (only changed values) let withdrawal_storage = withdrawal_storage(&accounts); @@ -812,7 +769,6 @@ pub fn validate_block_deriving_updates( mpt_witness: MptWitness, contracts: &HashMap, options: ValidationOptions, - #[cfg(feature = "std")] writer: Option>, ) -> Result<(StateUpdates, ValidationStats), ValidationError> { // A block carrying only transaction hashes can't be replayed — fail fast before paying // the witness proof verification. `replay_block` re-checks for direct callers. @@ -841,14 +797,8 @@ pub fn validate_block_deriving_updates( } // Verify the witness proof and replay the block's transactions over it - let VerifiedReplay { witness, accounts, output, mut stats } = verify_and_replay( - chain_spec, - block, - salt_witness, - contracts, - #[cfg(feature = "std")] - writer, - )?; + let VerifiedReplay { witness, accounts, output, mut stats } = + verify_and_replay(chain_spec, block, salt_witness, contracts)?; // Check the header's claims (withdrawals root, receipts root, logs bloom, gas used) // before the more expensive state-update derivation. @@ -890,8 +840,6 @@ mod tests { fx.mpt_witness(&hash), &fx.contracts, options, - #[cfg(feature = "std")] - None, ) } @@ -902,15 +850,7 @@ mod tests { salt_witness: SaltWitness, hash: B256, ) -> Result { - validate_block( - &chain_spec(), - block, - salt_witness, - fx.mpt_witness(&hash), - &fx.contracts, - #[cfg(feature = "std")] - None, - ) + validate_block(&chain_spec(), block, salt_witness, fx.mpt_witness(&hash), &fx.contracts) } /// The fixture witness for `hash` with the first byte of one witnessed (non-metadata) diff --git a/crates/stateless-core/src/light_witness.rs b/crates/stateless-core/src/light_witness.rs index 035e200f..244b450e 100644 --- a/crates/stateless-core/src/light_witness.rs +++ b/crates/stateless-core/src/light_witness.rs @@ -180,7 +180,7 @@ impl StateReader for LightWitness { fn metadata(&self, bucket_id: BucketId) -> Result { let metadata_key = bucket_metadata_key(bucket_id); match self.kvs.get(&metadata_key) { - Some(Some(salt_value)) => BucketMeta::try_from(salt_value.clone()) + Some(Some(salt_value)) => BucketMeta::try_from(salt_value) .map_err(|_| LightWitnessError { message: "Failed to decode metadata" }), // A well-formed witness never maps a metadata key to a deletion, // but witness bytes are network input (and the light decode @@ -293,13 +293,6 @@ impl StateReader for LightWitnessExecutor { } } -impl LightWitnessExecutor { - /// Get the underlying kvs map - pub fn kvs(&self) -> &BTreeMap> { - &self.light_witness.kvs - } -} - #[cfg(test)] mod tests { // `std` is the `alloc` alias in no_std builds, where the prelude carries diff --git a/crates/stateless-core/src/pipeline/advancer.rs b/crates/stateless-core/src/pipeline/advancer.rs index 360da60c..8f6ed103 100644 --- a/crates/stateless-core/src/pipeline/advancer.rs +++ b/crates/stateless-core/src/pipeline/advancer.rs @@ -1,6 +1,6 @@ //! Advancer stage: reorders processed blocks, detects reorgs, persists progress. -use std::collections::BTreeMap; +use std::{collections::BTreeMap, sync::Arc}; use eyre::Result; use tokio_util::sync::CancellationToken; @@ -78,8 +78,8 @@ where /// and advances the canonical chain. pub(crate) async fn chain_advancer( fetcher: &F, - store: &S, - hooks: &H, + store: Arc, + hooks: Arc, resolver: &R, result_rx: kanal::Receiver>, initial_tip: BlockMeta, @@ -87,7 +87,7 @@ pub(crate) async fn chain_advancer( ) -> Result where F: BlockFetcher, - S: ChainStore, + S: ChainStore + 'static, H: PipelineHooks, R: ReorgResolver, { @@ -127,7 +127,6 @@ where buffer.insert(item.block_number(), item); batch.clear(); - metas.clear(); while let Some(item) = buffer.remove(&next_expected) { if item.parent_hash() != current_tip.block_hash { debug!( @@ -139,7 +138,7 @@ where // Strategy is scenario-supplied (see `ReorgResolver`); `Floor` rolls back, // `Fatal`/`Retry` end the cycle. - let rollback_to = match resolver.resolve(fetcher, store, persisted_tip).await? { + let rollback_to = match resolver.resolve(fetcher, &store, persisted_tip).await? { ReorgResolution::Floor(floor) => { debug!(block = next_expected, floor, "Resolved reorg floor"); floor @@ -179,13 +178,34 @@ where } current_tip = item.to_block_meta(); next_expected += 1; - metas.push(current_tip.clone()); batch.push(item); } if !batch.is_empty() { - hooks.pre_advance(&batch)?; - store.advance_chain(&metas)?; + // The hooks' pre-advance persistence and the store commit are synchronous, + // potentially multi-ms disk work — run them off the async runtime so they can't + // stall other tasks (the trace server shares this runtime with its RPC handlers). + let advance_store = store.clone(); + let advance_hooks = hooks.clone(); + let owned_batch = std::mem::take(&mut batch); + let mut owned_metas = std::mem::take(&mut metas); + let advanced = tokio::task::spawn_blocking(move || { + owned_metas.clear(); + owned_metas.extend(owned_batch.iter().map(|item| item.to_block_meta())); + advance_hooks.pre_advance(&owned_batch)?; + advance_store.advance_chain(&owned_metas)?; + Ok::<_, eyre::Report>((owned_batch, owned_metas)) + }) + .await; + (batch, metas) = match advanced { + Ok(bufs) => bufs?, + // A panic in the store/hooks must propagate unchanged, exactly as it did + // when these calls ran inline on this task. + Err(join_err) => match join_err.try_into_panic() { + Ok(payload) => std::panic::resume_unwind(payload), + Err(join_err) => return Err(join_err.into()), + }, + }; persisted_tip = current_tip.block_number; debug!( tip = current_tip.block_number, diff --git a/crates/stateless-core/src/pipeline/fetcher.rs b/crates/stateless-core/src/pipeline/fetcher.rs index b66393b7..6d0582ce 100644 --- a/crates/stateless-core/src/pipeline/fetcher.rs +++ b/crates/stateless-core/src/pipeline/fetcher.rs @@ -14,17 +14,17 @@ use tracing::{Instrument, debug, error, info, info_span, warn}; use crate::pipeline::{config::PipelineConfig, traits::BlockFetcher}; /// Invariant: every block in `[base_block, next_block)` is in exactly one of -/// `in_flight_blocks`, `sent`, or `failed`. All mutations go through the methods below. +/// `task_to_block` (in flight), `sent`, or `failed`. All mutations go through the methods below. struct FetcherState { /// Lowest block not yet sent downstream. base_block: u64, /// Next block to spawn fresh. next_block: u64, tasks: JoinSet<(u64, Result)>, - /// Task id → block, for panic recovery (`JoinError` only carries the id). + /// Task id → block, for panic recovery (`JoinError` only carries the id) and + /// block-in-flight lookups (`recover_gaps` scans the values; the set is bounded by + /// `max_in_flight`, so a linear scan is negligible next to the awaited fetches). task_to_block: HashMap, - /// Mirror of `task_to_block.values()` for O(1) block-in-flight lookup. - in_flight_blocks: HashSet, /// Successful blocks, waiting for `base_block` to catch up. sent: HashSet, /// Blocks awaiting retry. The RPC client retries transient errors internally, so failures @@ -42,7 +42,6 @@ impl FetcherState { next_block: start_block, tasks: JoinSet::new(), task_to_block: HashMap::new(), - in_flight_blocks: HashSet::new(), sent: HashSet::new(), failed: HashSet::new(), } @@ -75,7 +74,6 @@ impl FetcherState { let handle = self.tasks.spawn(async move { (bn, fetcher.fetch(bn).await) }.instrument(span)); self.task_to_block.insert(handle.id(), bn); - self.in_flight_blocks.insert(bn); } fn spawn_next(&mut self, fetcher: &Arc) { @@ -93,21 +91,18 @@ impl FetcherState { fn on_success(&mut self, id: Id, bn: u64) { self.task_to_block.remove(&id); - self.in_flight_blocks.remove(&bn); self.sent.insert(bn); } fn on_failure(&mut self, id: Id, bn: u64) { self.task_to_block.remove(&id); - self.in_flight_blocks.remove(&bn); self.failed.insert(bn); } /// Re-enqueues the panicked task's block. Returns `None` if the id is unknown - /// (shouldn't happen — would leak the block from `in_flight_blocks`). + /// (shouldn't happen — `recover_gaps` would pick the block up). fn on_panic(&mut self, id: Id) -> Option { let bn = self.task_to_block.remove(&id)?; - self.in_flight_blocks.remove(&bn); self.failed.insert(bn); Some(bn) } @@ -129,8 +124,8 @@ impl FetcherState { let mut recovered = 0; for bn in self.base_block..self.next_block { if !self.sent.contains(&bn) && - !self.in_flight_blocks.contains(&bn) && - !self.failed.contains(&bn) + !self.failed.contains(&bn) && + !self.task_to_block.values().any(|&in_flight| in_flight == bn) { self.failed.insert(bn); recovered += 1; @@ -344,12 +339,11 @@ mod tests { async fn on_failure_re_enqueues_for_immediate_retry() { let mut state = FetcherState::::new(100); let id = fresh_task_id(&mut state.tasks, 100).await; - state.in_flight_blocks.insert(100); state.task_to_block.insert(id, 100); state.on_failure(id, 100); assert!(state.failed.contains(&100)); - assert!(!state.in_flight_blocks.contains(&100)); + assert!(!state.task_to_block.contains_key(&id)); assert_eq!(state.pop_failed(), Some(100)); assert!(!state.failed.contains(&100)); @@ -382,7 +376,8 @@ mod tests { let mut state = FetcherState::::new(100); state.next_block = 103; state.sent.insert(100); - state.in_flight_blocks.insert(101); + let id = fresh_task_id(&mut state.tasks, 101).await; + state.task_to_block.insert(id, 101); state.failed.insert(102); assert_eq!(state.recover_gaps(), 0); @@ -395,13 +390,11 @@ mod tests { async fn on_panic_re_enqueues_known_task() { let mut state = FetcherState::::new(100); let id = fresh_task_id(&mut state.tasks, 100).await; - state.in_flight_blocks.insert(100); state.task_to_block.insert(id, 100); let bn = state.on_panic(id); assert_eq!(bn, Some(100)); assert!(state.failed.contains(&100)); - assert!(!state.in_flight_blocks.contains(&100)); assert!(!state.task_to_block.contains_key(&id)); } diff --git a/crates/stateless-core/src/pipeline/mod.rs b/crates/stateless-core/src/pipeline/mod.rs index d7d67c24..092b2262 100644 --- a/crates/stateless-core/src/pipeline/mod.rs +++ b/crates/stateless-core/src/pipeline/mod.rs @@ -93,8 +93,8 @@ where let outcome = chain_advancer( &*fetcher, - &*store, - &*hooks, + store.clone(), + hooks.clone(), &resolver, result_rx, initial_tip, @@ -105,7 +105,7 @@ where fetcher_shutdown.cancel(); await_handles(fetcher_handle, worker_handles, config.await_handles_timeout).await; - let transient_reason: String = match outcome { + match outcome { Ok(PipelineOutcome::Shutdown) => { info!("Shutting down"); return Ok(()); @@ -129,27 +129,16 @@ where } Ok(PipelineOutcome::Retry(msg)) => { warn!(reason = %msg, "Cycle ended with retry signal"); - msg } Err(e) => { // Any `Err` at this level is unexpected (every intentional transient/fatal // case returns `Ok(PipelineOutcome::..)`). Log and fall into the same // stale-detect + sleep + continue recovery path as `Retry`. error!(error = %e, "Cycle ended with unexpected error"); - e.to_string() } - }; + } - if handle_transient_restart( - transient_reason, - &*fetcher, - &*store, - &*hooks, - &config, - &shutdown, - ) - .await? - { + if handle_transient_restart(&*fetcher, &*store, &*hooks, &config, &shutdown).await? { return Ok(()); } } @@ -162,7 +151,6 @@ where /// returns `Ok(false)` so the outer loop `continue`s. Propagates `Err` only for store / /// hook failures the caller can't meaningfully recover from. async fn handle_transient_restart( - _reason: String, fetcher: &F, store: &S, hooks: &H, diff --git a/crates/stateless-core/src/pipeline/tests.rs b/crates/stateless-core/src/pipeline/tests.rs index a1511c5e..ae5dcd9e 100644 --- a/crates/stateless-core/src/pipeline/tests.rs +++ b/crates/stateless-core/src/pipeline/tests.rs @@ -2,7 +2,6 @@ use std::{sync::Arc, time::Duration}; use alloy_primitives::{B256, BlockHash, BlockNumber, map::HashMap}; use eyre::{Result, anyhow}; -use revm::state::Bytecode; use tokio_util::sync::CancellationToken; use super::{ @@ -82,15 +81,6 @@ impl MockStore { } } -impl crate::ContractStore for MockStore { - fn get_contracts(&self, _: &[B256]) -> StoreResult<(HashMap, Vec)> { - Ok((HashMap::default(), vec![])) - } - fn add_contracts(&self, _: &[(B256, Bytecode)]) -> StoreResult<()> { - Ok(()) - } -} - impl ChainStore for MockStore { fn get_canonical_tip(&self) -> StoreResult> { Ok(self.chain.lock().unwrap().values().next_back().cloned()) @@ -185,7 +175,7 @@ async fn run_advancer( tip: BlockMeta, rpc_hashes: HashMap, blocks: Vec, -) -> (Result, MockStore) { +) -> (Result, Arc) { run_advancer_with_resolver(tip, rpc_hashes, blocks, &BisectResolver).await } @@ -194,10 +184,10 @@ async fn run_advancer_with_resolver>( rpc_hashes: HashMap, blocks: Vec, resolver: &R, -) -> (Result, MockStore) { - let store = MockStore::new(tip.clone()); +) -> (Result, Arc) { + let store = Arc::new(MockStore::new(tip.clone())); let fetcher = MockFetcher { hashes: rpc_hashes }; - let hooks = NoopHooks; + let hooks = Arc::new(NoopHooks); let (tx, rx) = kanal::bounded(16); { @@ -220,7 +210,8 @@ async fn run_advancer_with_resolver>( } let result = - chain_advancer(&fetcher, &store, &hooks, resolver, rx, tip, CancellationToken::new()).await; + chain_advancer(&fetcher, store.clone(), hooks, resolver, rx, tip, CancellationToken::new()) + .await; (result, store) } @@ -284,9 +275,9 @@ impl PipelineHooks for BadBlockHooks { /// but specialized — generalizing the original would cascade through ~6 helper types for /// a single test case. async fn run_bad_block_advancer(tip: BlockMeta, blocks: Vec) -> Result { - let store = MockStore::new(tip.clone()); + let store = Arc::new(MockStore::new(tip.clone())); let fetcher = MockFetcher { hashes: HashMap::default() }; - let hooks = BadBlockHooks; + let hooks = Arc::new(BadBlockHooks); let (tx, rx) = kanal::bounded::< std::result::Result, ErrorAction)>, >(16); @@ -298,8 +289,7 @@ async fn run_bad_block_advancer(tip: BlockMeta, blocks: Vec) -> Result } } - chain_advancer(&fetcher, &store, &hooks, &BisectResolver, rx, tip, CancellationToken::new()) - .await + chain_advancer(&fetcher, store, hooks, &BisectResolver, rx, tip, CancellationToken::new()).await } /// Covers the `verify_continuity` → Fatal branch in `chain_advancer` (`advancer.rs:109`). @@ -354,9 +344,9 @@ async fn test_chain_advancer_transient_error_returns_retry_outcome() { #[tokio::test] async fn test_chain_advancer_shutdown() { let tip = make_tip(10); - let store = MockStore::new(tip.clone()); + let store = Arc::new(MockStore::new(tip.clone())); let fetcher = MockFetcher { hashes: HashMap::default() }; - let hooks = NoopHooks; + let hooks = Arc::new(NoopHooks); let (_tx, rx) = kanal::bounded::< std::result::Result, ErrorAction)>, >(16); @@ -364,7 +354,7 @@ async fn test_chain_advancer_shutdown() { shutdown.cancel(); let outcome = - chain_advancer(&fetcher, &store, &hooks, &BisectResolver, rx, tip, shutdown).await.unwrap(); + chain_advancer(&fetcher, store, hooks, &BisectResolver, rx, tip, shutdown).await.unwrap(); assert!(matches!(outcome, PipelineOutcome::Shutdown)); } From 9549fffffb35c78c461680e15c3766f6a30758fb Mon Sep 17 00:00:00 2001 From: "liquan.eth" Date: Sun, 2 Aug 2026 15:45:35 +0800 Subject: [PATCH 2/2] refactor(core): build the metas batch inside the advance closure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit /simplify pass over the PR: the `metas` recycling was a vestigial half-collapse — a loop-level buffer threaded through mem::take, a tuple return, and a destructuring assignment purely to reuse one small Vec allocation per batch, next to multi-ms blocking disk work. Build it locally inside the spawn_blocking closure and return only the batch; the advanced count reads batch.len(), which it always equalled. Co-Authored-By: Claude Fable 5 --- crates/stateless-core/src/pipeline/advancer.rs | 16 +++++++--------- 1 file changed, 7 insertions(+), 9 deletions(-) diff --git a/crates/stateless-core/src/pipeline/advancer.rs b/crates/stateless-core/src/pipeline/advancer.rs index 8f6ed103..0ec5c01f 100644 --- a/crates/stateless-core/src/pipeline/advancer.rs +++ b/crates/stateless-core/src/pipeline/advancer.rs @@ -103,7 +103,6 @@ where // Reused across iterations to avoid per-iteration allocations; typical batch // size is small (<= `concurrent_workers`) and stable. let mut batch: Vec = Vec::new(); - let mut metas: Vec = Vec::new(); loop { let item = tokio::select! { @@ -188,17 +187,16 @@ where let advance_store = store.clone(); let advance_hooks = hooks.clone(); let owned_batch = std::mem::take(&mut batch); - let mut owned_metas = std::mem::take(&mut metas); let advanced = tokio::task::spawn_blocking(move || { - owned_metas.clear(); - owned_metas.extend(owned_batch.iter().map(|item| item.to_block_meta())); + let metas: Vec = + owned_batch.iter().map(|item| item.to_block_meta()).collect(); advance_hooks.pre_advance(&owned_batch)?; - advance_store.advance_chain(&owned_metas)?; - Ok::<_, eyre::Report>((owned_batch, owned_metas)) + advance_store.advance_chain(&metas)?; + Ok::<_, eyre::Report>(owned_batch) }) .await; - (batch, metas) = match advanced { - Ok(bufs) => bufs?, + batch = match advanced { + Ok(buf) => buf?, // A panic in the store/hooks must propagate unchanged, exactly as it did // when these calls ran inline on this task. Err(join_err) => match join_err.try_into_panic() { @@ -209,7 +207,7 @@ where persisted_tip = current_tip.block_number; debug!( tip = current_tip.block_number, - advanced = metas.len(), + advanced = batch.len(), buffered = buffer.len(), "Chain advanced" );