diff --git a/AGENTS.md b/AGENTS.md index e9b176f9..b68e91e1 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 witness primitives: SigV4 signer, object-key layout, endpoint parsing, signed PUT, and the retrying witness-object GET fetcher over either the signed S3 API or an unsigned Cloudflare custom domain; consumed by mega-reth's uploaders (write) and both binaries' R2 witness sources (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). @@ -172,7 +172,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/rpc_middleware.rs` | Concurrent execution of inbound JSON-RPC batch entries | diff --git a/Cargo.lock b/Cargo.lock index 96e8d4fb..6bd286ab 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1911,7 +1911,6 @@ dependencies = [ "mega-evm", "metrics", "metrics-derive", - "metrics-exporter-prometheus", "metrics-util", "op-alloy-consensus", "op-alloy-network", @@ -5682,10 +5681,10 @@ dependencies = [ "bincode 2.0.1", "clap", "eyre", - "fastrand", "futures", "jsonrpsee", "kanal", + "metrics-exporter-prometheus", "op-alloy-network", "op-alloy-rpc-types", "reqwest", @@ -5814,7 +5813,6 @@ dependencies = [ "jsonrpsee", "jsonrpsee-types", "metrics", - "metrics-exporter-prometheus", "op-alloy-rpc-types", "redb", "revm", diff --git a/README.md b/README.md index 65f36168..04792fcc 100644 --- a/README.md +++ b/README.md @@ -347,7 +347,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 a1e01edf..1cd63f35 100644 --- a/bin/debug-trace-server/Cargo.toml +++ b/bin/debug-trace-server/Cargo.toml @@ -29,7 +29,7 @@ revm.workspace = true revm-inspectors.workspace = true # stateless -stateless-common = { path = "../../crates/stateless-common" } +stateless-common = { path = "../../crates/stateless-common", features = ["prometheus-exporter"] } stateless-core = { path = "../../crates/stateless-core" } stateless-db = { path = "../../crates/stateless-db" } stateless-r2 = { path = "../../crates/stateless-r2" } @@ -46,7 +46,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 1f929dbe..661776c9 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/main.rs b/bin/debug-trace-server/src/main.rs index 8b6712d2..4a2e42de 100644 --- a/bin/debug-trace-server/src/main.rs +++ b/bin/debug-trace-server/src/main.rs @@ -853,7 +853,7 @@ async fn main() -> Result<()> { .rpc_per_attempt_timeout_ms .map(std::time::Duration::from_millis) .unwrap_or(rpc_defaults.per_attempt_timeout); - let rpc_retry = rpc_defaults.rpc_retry.clone(); + let rpc_retry = rpc_defaults.rpc_retry; let rpc_config = RpcClientConfig { data_max_concurrent_requests: args.data_max_concurrent_requests, witness_max_concurrent_requests: args.witness_max_concurrent_requests, @@ -1264,14 +1264,8 @@ async fn init_validator_db( rpc_client.get_header(BlockId::latest(), false).await }; - let anchor = BlockMeta { - block_number: header.number, - block_hash: header.hash, - post_state_root: header.state_root, - post_withdrawals_root: header - .withdrawals_root - .ok_or_else(|| eyre::eyre!("Block {} is missing withdrawals_root", header.hash))?, - }; + let anchor = BlockMeta::try_from_header(&header) + .ok_or_else(|| eyre::eyre!("Block {} is missing withdrawals_root", header.hash))?; ChainStore::reset_to_anchor(&*db, &anchor) .map_err(|e| eyre::eyre!("Failed to reset anchor: {}", e))?; diff --git a/bin/debug-trace-server/src/metrics.rs b/bin/debug-trace-server/src/metrics.rs index c12191b4..4fc647a8 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, gauge, 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. @@ -1026,15 +1025,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 51beca81..fabb866a 100644 --- a/bin/stateless-validator/Cargo.toml +++ b/bin/stateless-validator/Cargo.toml @@ -27,7 +27,7 @@ op-alloy-rpc-types.workspace = true revm = { workspace = true, features = ["serde"] } # stateless -stateless-common = { path = "../../crates/stateless-common" } +stateless-common = { path = "../../crates/stateless-common", features = ["prometheus-exporter"] } stateless-core = { path = "../../crates/stateless-core" } stateless-db = { path = "../../crates/stateless-db" } stateless-r2 = { path = "../../crates/stateless-r2" } @@ -36,7 +36,6 @@ stateless-r2 = { path = "../../crates/stateless-r2" } clap = { workspace = true, features = ["env"] } eyre.workspace = true metrics.workspace = true -metrics-exporter-prometheus.workspace = true redb.workspace = true serde_json.workspace = true thiserror.workspace = true diff --git a/bin/stateless-validator/src/app.rs b/bin/stateless-validator/src/app.rs index 90a2ac3e..0f07afd7 100644 --- a/bin/stateless-validator/src/app.rs +++ b/bin/stateless-validator/src/app.rs @@ -15,7 +15,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)] @@ -291,7 +291,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()?; @@ -356,7 +356,7 @@ pub async fn run() -> Result<()> { .r2_connect_timeout_ms .map_or(stateless_r2::fetch::DEFAULT_CONNECT_TIMEOUT, Duration::from_millis), }; - let transport = build_r2_transport(&args, timeouts, rpc_config.rpc_retry.clone())?; + let transport = build_r2_transport(&args, timeouts, rpc_config.rpc_retry)?; Some(Arc::new(R2WitnessClient::new(transport))) } }; @@ -392,14 +392,8 @@ pub async fn run() -> Result<()> { // surfaces as "no forward progress" rather than an arbitrarily bounded retry error. let header = client.get_header(BlockId::Hash(block_hash.into()), true).await; - let anchor = BlockMeta { - block_number: header.number, - block_hash: header.hash, - post_state_root: header.state_root, - post_withdrawals_root: header - .withdrawals_root - .ok_or_else(|| eyre::eyre!("Block {} is missing withdrawals_root", block_hash))?, - }; + let anchor = BlockMeta::try_from_header(&header) + .ok_or_else(|| eyre::eyre!("Block {} is missing withdrawals_root", block_hash))?; validator_db.reset_to_anchor(&anchor)?; info!( @@ -434,13 +428,12 @@ 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, pipeline_config, ) .await; diff --git a/bin/stateless-validator/src/chain_sync.rs b/bin/stateless-validator/src/chain_sync.rs index 82cf2529..b91436fb 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) @@ -313,15 +306,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 6be652b4..2e68ac6d 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; @@ -127,15 +129,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(); @@ -228,7 +222,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/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..97eac419 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, pipeline_config: PipelineConfig, ) -> Result<()> { - let report_validation = report_validation_endpoint.is_some(); + let report_validation = client.reports_validation(); 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..0b5c0327 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,7 @@ mod tests { post_state_root: B256::from([2u8; 32]), post_withdrawals_root: B256::from([3u8; 32]), }; - store.set_anchor_block(&tip).unwrap(); + 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 c69347f1..b42f9dbd 100644 --- a/bin/stateless-validator/tests/integration.rs +++ b/bin/stateless-validator/tests/integration.rs @@ -12,9 +12,7 @@ use alloy_primitives::{B256, BlockHash}; use alloy_rpc_types_eth::Block; use clap::Parser; use jsonrpsee::server::ServerConfigBuilder; -use jsonrpsee_types::error::{ - CALL_EXECUTION_FAILED_CODE, ErrorObject, ErrorObjectOwned, INVALID_PARAMS_CODE, -}; +use jsonrpsee_types::error::{CALL_EXECUTION_FAILED_CODE, ErrorObject, ErrorObjectOwned}; use stateless_common::{RpcClient, RpcClientConfig, WitnessRequestKeys, encode_witness_response}; use stateless_core::{ BisectResolver, ChainStore, ContractStore, PipelineConfig, db::BlockMeta, @@ -332,6 +330,36 @@ 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 = parse_hex_u64(hex_number); + 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> { @@ -357,19 +385,9 @@ fn setup_test_db(fx: &TestFixtures) -> eyre::Result<(Arc, tempfile: let temp_dir = tempfile::tempdir()?; let db = ValidatorDB::new(temp_dir.path().join(VALIDATOR_DB_FILENAME))?; - let (block_num, block_hash) = fx.min_block(); - let block = &fx.blocks[&block_hash]; - let withdrawals_root = block - .header - .withdrawals_root + let (_, block_hash) = fx.min_block(); + let anchor = BlockMeta::try_from_header(&fx.blocks[&block_hash].header) .ok_or_else(|| eyre::eyre!("Block {block_hash} missing withdrawals_root"))?; - - let anchor = BlockMeta { - block_number: block_num, - block_hash, - post_state_root: block.header.state_root, - post_withdrawals_root: withdrawals_root, - }; db.reset_to_anchor(&anchor)?; Ok((Arc::new(db), temp_dir)) @@ -383,39 +401,16 @@ async fn setup_mock_rpc_server( serve_with_config(cfg, state, |module| { 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 = parse_hex_u64(&hex_number); - - 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()?; + 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()?; + Ok::<_, ErrorObject<'static>>(shape_block(ctx.block_by_hash(hash)?, full_block)) }) .unwrap(); @@ -428,45 +423,21 @@ async fn setup_mock_rpc_server( module .register_method("eth_getHeaderByNumber", |params, ctx, _| { - let (hex_number,): (String,) = params.parse().unwrap(); - let block_number = parse_hex_u64(&hex_number); - - 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()?; + 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()?; + 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()?; let code = ctx.fixtures.contracts.get(&hash).cloned().unwrap_or_default(); Ok::<_, ErrorObject<'static>>(code.original_bytes()) @@ -475,9 +446,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()?; let block_hash = BlockHash::from(keys.block_hash.0); let salt_witness = @@ -562,11 +531,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); @@ -628,16 +593,9 @@ async fn run_end_block_slice( cfg.concurrent_workers = 1; cfg.sync_target = Some(max_block_number); - let result = run_with_signals( - client, - None, - Arc::clone(&validator_db), - contract_cache, - chain_spec, - Some(url.clone()), - cfg, - ) - .await; + let result = + run_with_signals(client, None, Arc::clone(&validator_db), contract_cache, chain_spec, cfg) + .await; handle.stop().unwrap(); diff --git a/crates/stateless-common/Cargo.toml b/crates/stateless-common/Cargo.toml index 16a6130e..bb2ea816 100644 --- a/crates/stateless-common/Cargo.toml +++ b/crates/stateless-common/Cargo.toml @@ -35,8 +35,8 @@ base64.workspace = true bincode.workspace = true clap.workspace = true eyre.workspace = true -fastrand = { workspace = true, features = ["std"] } futures.workspace = true +metrics-exporter-prometheus = { workspace = true, optional = true } # Also enables gzip/brotli on the shared reqwest 0.12 client (Cargo feature unification) for witness/data fetches. reqwest = { workspace = true, features = ["gzip", "brotli"] } rolling-file.workspace = true @@ -54,6 +54,11 @@ kanal.workspace = true stateless-test-utils = { path = "../stateless-test-utils", features = ["mock-rpc"] } tokio-util.workspace = true +[features] +# The Prometheus exporter installer; on only in the binaries, so library consumers +# (mega-reth) do not build the exporter for a function they never call. +prometheus-exporter = ["dep:metrics-exporter-prometheus"] + [[bench]] harness = false name = "witness_zstd_level" diff --git a/crates/stateless-common/src/lib.rs b/crates/stateless-common/src/lib.rs index aeed85f8..2cf8294b 100644 --- a/crates/stateless-common/src/lib.rs +++ b/crates/stateless-common/src/lib.rs @@ -3,9 +3,17 @@ pub mod metrics; pub use metrics::{RpcMethod, RpcMetrics}; pub mod rpc_client; pub use rpc_client::{ - BackoffPolicy, CodeFetchError, RpcClient, RpcClientConfig, RpcDeadlineExceeded, - SetValidatedBlocksResponse, WitnessRequestKeys, + CodeFetchError, RpcClient, RpcClientConfig, RpcDeadlineExceeded, SetValidatedBlocksResponse, + WitnessRequestKeys, }; +/// Exponential-backoff policy used by [`RpcClient`]'s round-level retry loop: `initial` is the +/// first sleep duration; each round doubles it up to `max`. +/// +/// The same pair paces the R2 GET loop, so the type is defined in `stateless-r2` (which must +/// stay free of upward dependencies) and re-exported here under the name this crate's API +/// uses; the retry loop steps it through +/// [`RetryPacing::schedule`](stateless_r2::fetch::RetryPacing::schedule). +pub use stateless_r2::fetch::RetryPacing as BackoffPolicy; pub mod witness_encoding; pub use witness_encoding::{ WITNESS_RESPONSE_VERSION_PREFIX, WITNESS_ZSTD_LEVEL, WitnessDecodingError, diff --git a/crates/stateless-common/src/metrics.rs b/crates/stateless-common/src/metrics.rs index 20d61a99..1065f151 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 — behind the +//! `prometheus-exporter` feature — the shared exporter installer +//! (`install_prometheus_exporter`). 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. +#[cfg(feature = "prometheus-exporter")] +pub fn install_prometheus_exporter( + addr: std::net::SocketAddr, + bucket_specs: &[(&str, &[f64])], +) -> eyre::Result<()> { + use metrics_exporter_prometheus::{Matcher, PrometheusBuilder}; + + 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/r2_witness.rs b/crates/stateless-common/src/r2_witness.rs index 157ad2c7..7f344591 100644 --- a/crates/stateless-common/src/r2_witness.rs +++ b/crates/stateless-common/src/r2_witness.rs @@ -12,7 +12,7 @@ use std::time::Instant; use alloy_primitives::B256; use stateless_r2::{ - fetch::{CfAccessCredentials, FetchTimeouts, R2GetError, R2ObjectFetcher, RetryPacing}, + fetch::{CfAccessCredentials, FetchTimeouts, R2GetError, R2ObjectFetcher}, keys, }; use tokio::task::JoinError; @@ -129,12 +129,6 @@ pub struct R2WitnessTransport { max_concurrent_requests: Option, } -/// The fetcher's pacing view of a [`BackoffPolicy`] — the adapter-layer conversion that -/// keeps `stateless-r2` free of a dependency on this workspace's backoff type. -fn pacing(backoff: &BackoffPolicy) -> RetryPacing { - RetryPacing { initial: backoff.initial, max: backoff.max } -} - impl R2WitnessTransport { /// Builds a transport from an R2 endpoint origin, bucket, and bucket-scoped S3 /// credentials. @@ -158,7 +152,7 @@ impl R2WitnessTransport { access_key_id, secret_access_key, timeouts, - pacing(&retry_backoff), + retry_backoff, max_concurrent_requests, ) .map_err(|e| eyre::eyre!(e))?; @@ -183,7 +177,7 @@ impl R2WitnessTransport { domain, access, timeouts, - pacing(&retry_backoff), + retry_backoff, max_concurrent_requests, connections, ) diff --git a/crates/stateless-common/src/rpc_client.rs b/crates/stateless-common/src/rpc_client.rs index dfe2f38e..6a480bc5 100644 --- a/crates/stateless-common/src/rpc_client.rs +++ b/crates/stateless-common/src/rpc_client.rs @@ -56,31 +56,12 @@ use tokio::sync::Semaphore; use tracing::{instrument, trace, warn}; use crate::{ + BackoffPolicy, metrics::{RpcAttemptOutcome, RpcMethod, RpcMetrics}, witness_encoding::{decode_witness_response, decode_witness_response_light}, witness_size::WitnessSizeBreakdown, }; -/// Exponential-backoff policy used by [`RpcClient`]'s round-level 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. -#[derive(Debug, Clone)] -pub struct BackoffPolicy { - /// First retry sleep. Each subsequent retry doubles up to `max`. - pub initial: Duration, - /// Upper bound on any single retry sleep. - pub max: Duration, -} - -impl BackoffPolicy { - /// Creates a new policy with the given `initial` and `max` sleep durations. - pub const fn new(initial: Duration, max: Duration) -> Self { - Self { initial, max } - } -} - /// Error returned by the `_with_deadline` RPC methods when a caller-supplied /// deadline elapses before any provider succeeds. /// @@ -420,6 +401,11 @@ impl RpcClient { self.witness_providers.len() } + /// Returns whether a validation report endpoint is configured. + pub fn reports_validation(&self) -> bool { + self.report_provider.is_some() + } + /// Returns the credential-stripped `{idx}:{host}` metric/log label of the witness /// endpoint at `idx`, or `None` when out of range. Use this instead of the raw /// configured URL wherever an endpoint identity is logged. @@ -1151,9 +1137,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(); // Records the logical-call deadline give-up (once) and builds the typed error. Called @@ -1393,11 +1377,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 { @@ -1419,7 +1399,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; } } diff --git a/crates/stateless-core/src/chain_spec.rs b/crates/stateless-core/src/chain_spec.rs index adba1fc3..48298172 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,14 @@ impl MegaethGenesisHardforks { } /// Convert the MegaETH genesis hardforks into a vector of hardforks and their conditions. + /// + /// Fork selection never reads this order: mega-evm resolves the active fork through + /// [`MegaHardforks::mega_fork_activation`], a name-keyed lookup on [`ChainHardforks`], so + /// insertion order cannot change which hardfork params a block gets. What the literal + /// must agree with is mega-evm's own [`MegaHardfork`] declaration, in membership and + /// order: the chain-spec tests pin it against `MegaHardfork::VARIANTS`, so a fork the + /// pinned mega-evm supports cannot go unscheduled unnoticed. Add a new fork here at its + /// declaration position, with its genesis field above. pub fn into_vec(self) -> Vec<(Box, ForkCondition)> { vec![ (MegaHardfork::MiniRex.boxed(), self.mini_rex_time.map(ForkCondition::Timestamp)), @@ -303,25 +295,12 @@ 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; + use std::{ + format, + string::{String, ToString}, + }; use alloy_serde::OtherFields; @@ -382,6 +361,56 @@ mod tests { assert_eq!(spec.hardforks.fork(MegaHardfork::MiniRex), ForkCondition::Timestamp(3)); } + /// Pins [`MegaethGenesisHardforks::into_vec`], end-to-end through `from_genesis`, against + /// mega-evm's own `MegaHardfork` declaration: every variant the pinned mega-evm declares + /// comes out scheduled, in declaration order. A variant `into_vec` lacks, duplicates, or + /// misplaces fails here, whichever end of the ladder it sits at. + #[test] + fn test_mega_hardforks_iterate_in_activation_order() { + let mut genesis = Genesis::default(); + for (index, fork) in MegaHardfork::VARIANTS.iter().enumerate() { + let activation = index as u64 + 1; + if *fork == MegaHardfork::Rex5 { + // `from_genesis` refuses a scheduled Rex5 without its bootstrap seeds. + schedule_valid_rex5(&mut genesis, activation); + } else { + genesis + .config + .extra_fields + .insert_value(genesis_time_field(*fork), activation) + .unwrap(); + } + } + // Likewise required alongside `rex6Time`. + genesis.config.extra_fields.insert_value("rex6MinRotationDelay".to_string(), 7200).unwrap(); + let spec = ChainSpec::from_genesis(genesis); + + let expected: Vec<&str> = MegaHardfork::VARIANTS.iter().map(|fork| fork.name()).collect(); + // `expected` is the complete MegaETH membership, so filtering by it drops only the + // Optimism/Ethereum forks `from_genesis` merges in after the MegaETH ones; a MegaETH + // fork `into_vec` omits is missing from `mega_order`, not filtered out of it. + let mega_order: Vec<&str> = spec + .hardforks + .forks_iter() + .map(|(hardfork, _)| hardfork.name()) + .filter(|name| expected.contains(name)) + .collect(); + assert_eq!( + mega_order, expected, + "the scheduled MegaETH forks must match mega-evm's `MegaHardfork` declaration \ + exactly, in order — a new variant needs its `MegaethGenesisHardforks` field and \ + its `into_vec` entry at the declaration position; fix those, not this test" + ); + } + + /// The genesis `config` field that schedules `fork`, under the naming every MegaETH fork + /// follows: the variant name with its first letter lowercased, plus `Time` + /// (`MiniRex1` → `miniRex1Time`). + fn genesis_time_field(fork: MegaHardfork) -> String { + let (head, tail) = fork.name().split_at(1); + format!("{}{tail}Time", head.to_ascii_lowercase()) + } + #[test] fn test_extract_from_json() { let genesis_info = r#" @@ -668,49 +697,44 @@ mod tests { let _ = ChainSpec::from_genesis(genesis); } - /// Every fork in the canonical [`mega_mainnet_hardforks()`] ladder must be scheduled by + /// Every fork the pinned mega-evm declares (`MegaHardfork::VARIANTS`) must be scheduled by /// the shipped mainnet genesis file, minus an explicit not-yet-scheduled allowlist. /// /// The schema tests above all run on synthetic genesis JSON, so none of them can notice /// the shipped data file missing a fork — the shape of a real production failure: the /// pinned mega-evm already supports the fork, genesis never schedules it, chain-spec /// loading stays green, and every block past the activation timestamp fails to replay. - /// A fork added to the ladder without its genesis field fails here at test time instead - /// of at the activation boundary. + /// A mega-evm bump that brings a new fork without its genesis field fails here at test + /// time instead of at the activation boundary. #[cfg(feature = "std")] #[test] fn mainnet_genesis_schedules_every_canonical_hardfork() { use stateless_test_utils::fixtures::TestFixtures; - // Ladder forks mainnet has deliberately not scheduled yet — empty today, every fork - // through Rex6 is live. An entry here must actually be unscheduled: once its genesis - // field lands, the assertion below demands the entry's removal, so the allowlist - // cannot rot into shadowing the check. - const NOT_YET_SCHEDULED: &[&str] = &[]; + // Forks mainnet has deliberately not scheduled yet — empty today, every fork through + // Rex6 is live. An entry here must actually be unscheduled: once its genesis field + // lands, the assertion below demands the entry's removal, so the allowlist cannot rot + // into shadowing the check. + const NOT_YET_SCHEDULED: &[MegaHardfork] = &[]; let genesis = TestFixtures::mainnet_shared().load_genesis().expect("mainnet genesis"); let spec = ChainSpec::from_genesis(genesis); - let scheduled: Vec<(&str, ForkCondition)> = - spec.hardforks.forks_iter().map(|(fork, condition)| (fork.name(), condition)).collect(); - - let ladder = mega_mainnet_hardforks(); - for (fork, _) in ladder.forks_iter() { - let activation = - scheduled.iter().find(|(name, _)| *name == fork.name()).map(|(_, c)| *c); - if NOT_YET_SCHEDULED.contains(&fork.name()) { + + for fork in MegaHardfork::VARIANTS { + // The same name-keyed lookup fork selection resolves through at runtime. + let activation = spec.hardforks.fork(*fork); + if NOT_YET_SCHEDULED.contains(fork) { assert_eq!( activation, - None, - "{} is scheduled now — remove it from NOT_YET_SCHEDULED", - fork.name() + ForkCondition::Never, + "{fork} is scheduled now — remove it from NOT_YET_SCHEDULED" ); } else { assert!( - matches!(activation, Some(ForkCondition::Timestamp(_))), - "mainnet genesis does not schedule {} (activation: {activation:?}); add \ + matches!(activation, ForkCondition::Timestamp(_)), + "mainnet genesis does not schedule {fork} (activation: {activation:?}); add \ its genesis field, or allowlist it in NOT_YET_SCHEDULED if it is \ - genuinely not scheduled yet", - fork.name() + genuinely not scheduled yet" ); } } diff --git a/crates/stateless-core/src/db.rs b/crates/stateless-core/src/db.rs index f75f2626..6c250aa0 100644 --- a/crates/stateless-core/src/db.rs +++ b/crates/stateless-core/src/db.rs @@ -29,6 +29,33 @@ 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; anchor initialization, which must instead *reject* such a + /// header, goes through [`Self::try_from_header`]. + pub fn from_header(header: &alloy_rpc_types_eth::Header) -> Self { + Self::with_withdrawals_root(header, header.withdrawals_root.unwrap_or_default()) + } + + /// Strict [`Self::from_header`]: `None` when the header carries no `withdrawals_root`. + pub fn try_from_header(header: &alloy_rpc_types_eth::Header) -> Option { + header.withdrawals_root.map(|root| Self::with_withdrawals_root(header, root)) + } + + fn with_withdrawals_root( + header: &alloy_rpc_types_eth::Header, + post_withdrawals_root: B256, + ) -> Self { + Self { + block_number: header.number, + block_hash: header.hash, + post_state_root: header.state_root, + post_withdrawals_root, + } + } +} + /// Errors returned by persistence trait methods. /// /// This is the single typed error at the library/binary boundary: every @@ -107,7 +134,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<()>; @@ -136,6 +165,44 @@ mod tests { } impl core::error::Error for TestErr {} + /// An RPC header for block `number` whose hash and roots are distinguishable bytes. + fn rpc_header(number: u64, withdrawals_root: Option) -> alloy_rpc_types_eth::Header { + alloy_rpc_types_eth::Header { + hash: BlockHash::from([0xAA; 32]), + inner: alloy_consensus::Header { + number, + state_root: B256::from([0xBB; 32]), + withdrawals_root, + ..Default::default() + }, + ..Default::default() + } + } + + #[test] + fn from_header_projects_the_sealed_block_meta() { + let root = B256::from([0xCC; 32]); + let header = rpc_header(7, Some(root)); + let meta = BlockMeta::from_header(&header); + assert_eq!( + meta, + BlockMeta { + block_number: 7, + block_hash: BlockHash::from([0xAA; 32]), + post_state_root: B256::from([0xBB; 32]), + post_withdrawals_root: root, + } + ); + assert_eq!(BlockMeta::try_from_header(&header), Some(meta)); + } + + #[test] + fn missing_withdrawals_root_defaults_in_from_header_and_rejects_in_try_from_header() { + let header = rpc_header(7, None); + assert_eq!(BlockMeta::from_header(&header).post_withdrawals_root, B256::ZERO); + assert_eq!(BlockMeta::try_from_header(&header), None); + } + #[test] fn test_block_meta_equality() { let a = BlockMeta { diff --git a/crates/stateless-core/src/evm_database.rs b/crates/stateless-core/src/evm_database.rs index abdf2f3e..0689707f 100644 --- a/crates/stateless-core/src/evm_database.rs +++ b/crates/stateless-core/src/evm_database.rs @@ -5,6 +5,7 @@ //! validation. use std::{ + collections::BTreeMap, format, string::{String, ToString}, vec::Vec, @@ -210,23 +211,24 @@ impl WitnessExternalEnv { /// * `salt_witness` - The SALT witness containing bucket metadata /// * `block_number` - The block number for validation checks /// - /// # Returns - /// - /// Returns `Ok(WitnessExternalEnv)` if all metadata is valid, or an error if: - /// - Any metadata key has a `None` value (malformed witness) - /// - Metadata cannot be parsed as `BucketMeta` (corrupt witness) - /// /// # Errors /// - /// This method enforces strict witness validation and will fail if: - /// - A metadata key is present but has no value - /// - A metadata value cannot be deserialized into valid `BucketMeta` + /// Fails on a malformed witness: a metadata key with no value, or a metadata value that + /// does not deserialize into a valid `BucketMeta`. pub fn new( 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::, _>>()?; @@ -252,21 +254,13 @@ impl WitnessExternalEnv { Ok((bucket_id, meta.capacity)) } - /// Creates a new external environment provider from a LightWitness. - /// - /// This is the fast version of `new()` that works with `LightWitness` - /// for improved deserialization performance. + /// [`Self::new`] over a [`LightWitness`]: the same metadata scan, the light witness + /// only being cheaper to decode. pub fn from_light_witness( 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) } } diff --git a/crates/stateless-core/src/executor.rs b/crates/stateless-core/src/executor.rs index 71a6e9a3..96ef17cd 100644 --- a/crates/stateless-core/src/executor.rs +++ b/crates/stateless-core/src/executor.rs @@ -61,7 +61,7 @@ use revm::{ State, 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}; @@ -69,7 +69,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}, @@ -263,10 +263,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, @@ -286,7 +282,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) diff --git a/crates/stateless-core/src/light_witness.rs b/crates/stateless-core/src/light_witness.rs index 035e200f..973a03e6 100644 --- a/crates/stateless-core/src/light_witness.rs +++ b/crates/stateless-core/src/light_witness.rs @@ -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/fetcher.rs b/crates/stateless-core/src/pipeline/fetcher.rs index b66393b7..0e96c898 100644 --- a/crates/stateless-core/src/pipeline/fetcher.rs +++ b/crates/stateless-core/src/pipeline/fetcher.rs @@ -14,17 +14,18 @@ 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. +/// `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_to_block: HashMap, - /// Mirror of `task_to_block.values()` for O(1) block-in-flight lookup. - in_flight_blocks: HashSet, + /// In-flight block → its task id. Keyed by block so the per-completion paths + /// (`recover_gaps`, success, failure) are O(1); the id exists only to map a panicked + /// task's `JoinError` (which carries nothing else) back to its block, a scan on that + /// rare path. + in_flight: HashMap, /// Successful blocks, waiting for `base_block` to catch up. sent: HashSet, /// Blocks awaiting retry. The RPC client retries transient errors internally, so failures @@ -41,8 +42,7 @@ impl FetcherState { base_block: start_block, next_block: start_block, tasks: JoinSet::new(), - task_to_block: HashMap::new(), - in_flight_blocks: HashSet::new(), + in_flight: HashMap::new(), sent: HashSet::new(), failed: HashSet::new(), } @@ -74,8 +74,7 @@ impl FetcherState { let span = info_span!("fetch_block", block_number = bn); 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); + self.in_flight.insert(bn, handle.id()); } fn spawn_next(&mut self, fetcher: &Arc) { @@ -91,23 +90,22 @@ impl FetcherState { Some(bn) } - fn on_success(&mut self, id: Id, bn: u64) { - self.task_to_block.remove(&id); - self.in_flight_blocks.remove(&bn); + fn on_success(&mut self, bn: u64) { + self.in_flight.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); + fn on_failure(&mut self, bn: u64) { + self.in_flight.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`). + /// Re-enqueues the panicked task's block, found by scanning `in_flight` for its id. + /// Returns `None` if the id is unknown (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); + let bn = self.in_flight.iter().find_map(|(&bn, &task)| (task == id).then_some(bn))?; + self.in_flight.remove(&bn); self.failed.insert(bn); Some(bn) } @@ -129,8 +127,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.in_flight.contains_key(&bn) { self.failed.insert(bn); recovered += 1; @@ -270,19 +268,19 @@ pub async fn block_fetcher( }; match joined { - Some(Ok((id, (bn, Ok(item))))) => { - state.on_success(id, bn); + Some(Ok((_, (bn, Ok(item))))) => { + state.on_success(bn); if tx.send(item).await.is_err() { info!("Channel closed, stopping"); return Ok(()); } debug!(block_number = bn, "Block sent to pipeline"); } - Some(Ok((id, (bn, Err(e))))) => { + Some(Ok((_, (bn, Err(e))))) => { // RPC client handles transient errors; anything here is deterministic // (integrity check fails, etc.). Re-enqueue — next attempt rotates // round-robin to a different provider. - state.on_failure(id, bn); + state.on_failure(bn); warn!(block_number = bn, error = %e, "Block fetch failed, re-enqueueing"); } Some(Err(join_err)) => { @@ -333,8 +331,8 @@ mod tests { } } - /// Spawns a dummy task so we can obtain a real `task::Id` to drive `on_failure` / - /// `on_success`, which expect an id that came out of the `JoinSet`. + /// Spawns a dummy task so we can obtain a real `task::Id` to record in `in_flight`, + /// as the fetcher does for a task that came out of the `JoinSet`. async fn fresh_task_id(tasks: &mut JoinSet<(u64, Result<()>)>, bn: u64) -> Id { let handle = tasks.spawn(async move { (bn, Ok(())) }); handle.id() @@ -344,12 +342,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.in_flight.insert(100, id); - state.on_failure(id, 100); + state.on_failure(100); assert!(state.failed.contains(&100)); - assert!(!state.in_flight_blocks.contains(&100)); + assert!(!state.in_flight.contains_key(&100)); assert_eq!(state.pop_failed(), Some(100)); assert!(!state.failed.contains(&100)); @@ -382,7 +379,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.in_flight.insert(101, id); state.failed.insert(102); assert_eq!(state.recover_gaps(), 0); @@ -395,23 +393,21 @@ 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); + state.in_flight.insert(100, id); 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)); + assert!(!state.in_flight.contains_key(&100)); } #[tokio::test] async fn on_panic_unknown_id_returns_none() { - // An id that was never recorded in `task_to_block` — e.g. a stale id from a prior + // An id that was never recorded in `in_flight` — e.g. a stale id from a prior // cycle. `on_panic` must not touch any state in that case. let mut state = FetcherState::::new(100); let id = fresh_task_id(&mut state.tasks, 100).await; - // Note: we did NOT populate task_to_block with this id. + // Note: we did NOT populate in_flight with this id. assert!(state.on_panic(id).is_none()); assert!(state.failed.is_empty()); } diff --git a/crates/stateless-core/src/pipeline/mod.rs b/crates/stateless-core/src/pipeline/mod.rs index d7d67c24..1603713d 100644 --- a/crates/stateless-core/src/pipeline/mod.rs +++ b/crates/stateless-core/src/pipeline/mod.rs @@ -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..deaeca59 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()) diff --git a/crates/stateless-r2/src/fetch.rs b/crates/stateless-r2/src/fetch.rs index 55494887..f7bfc6ff 100644 --- a/crates/stateless-r2/src/fetch.rs +++ b/crates/stateless-r2/src/fetch.rs @@ -176,12 +176,12 @@ impl std::error::Error for R2GetError { } } -/// Retry pacing for retryable GET failures: first sleep `initial` (with up to 50% jitter), +/// Retry pacing for retryable failures: first sleep `initial` (with up to 50% jitter), /// doubling up to `max`. /// -/// A plain pair rather than a reference to any binary's backoff-policy type, so this crate -/// stays free of upward dependencies; callers build it from whatever flags govern their -/// retry pacing. +/// Shared by this crate's GET loop and the RPC client's round-robin loop (which re-exports +/// it as its backoff policy); it lives here because this crate must stay free of upward +/// dependencies. Stepped through [`Self::schedule`]. #[derive(Clone, Copy, Debug)] pub struct RetryPacing { /// First inter-attempt sleep. @@ -190,6 +190,47 @@ pub struct RetryPacing { pub max: Duration, } +impl RetryPacing { + /// Creates a pacing with the given `initial` and `max` sleep durations. + pub const fn new(initial: Duration, max: Duration) -> Self { + Self { initial, max } + } + + /// Starts executing this pacing, 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 [`RetryPacing`] — the arithmetic every retry loop that paces this +/// way needs, in one place. +/// +/// Owns the three invariants those loops rely 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 pacing cannot turn a retry loop into a busy-loop. +/// +/// Deadline handling stays with the caller: whether an overrunning sleep is clamped or +/// gives up is policy, not arithmetic. +#[derive(Clone, 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 + } +} + /// A successfully fetched object body plus the time the fetch spent queued on the /// concurrency cap. /// @@ -791,8 +832,7 @@ impl R2ObjectFetcher { on_retry: impl Fn(), ) -> Result { let key = keys::block_object_key(number, hash); - let max_backoff_ms = self.pacing.max.as_millis() as u64; - let mut backoff_ms = self.pacing.initial.as_millis() as u64; + let mut backoff = self.pacing.schedule(); let mut attempt = 0usize; let mut queue_wait = Duration::ZERO; @@ -838,10 +878,7 @@ impl R2ObjectFetcher { if !e.is_retryable() || attempt >= max_attempts { return Err(e); } - // Jittered doubling; `.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); + let sleep_ms = backoff.next_sleep_ms(); if deadline .is_some_and(|d| Instant::now() + Duration::from_millis(sleep_ms) >= d) { @@ -853,7 +890,6 @@ impl R2ObjectFetcher { "R2 witness GET failed, backing off", ); tokio::time::sleep(Duration::from_millis(sleep_ms)).await; - backoff_ms = (backoff_ms * 2).min(max_backoff_ms); } } }