refactor: apply simplify-review cleanups across core, common, and the binaries

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                                    |

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                                                           |

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

bin/debug-trace-server/src/chain_sync.rs

@@ -157,12 +157,7 @@ impl BlockFetcher for TraceFetcher {
     async fn latest_block_meta(&self) -> Result<BlockMeta> {
         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<TraceProcessedBlock, Self::Error> {
-        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 })
     }
 }

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();

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<B256, Bytecode>, Vec<B256>)> {
-            Ok((HashMap::default(), vec![]))
-        }
-        fn add_contracts(&self, _: &[(B256, Bytecode)]) -> StoreResult<()> {
-            Ok(())
-        }
-    }
-
     impl ChainStore for StubBlockStore {
         fn get_canonical_tip(&self) -> StoreResult<Option<BlockMeta>> {
             self.tip_reads.fetch_add(1, Ordering::Relaxed);

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"] }

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;

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<RpcClient>,
     /// `Some` ⇒ fetch witnesses directly from R2; `None` ⇒ RPC.
     pub r2_witness: Option<Arc<R2WitnessClient>>,
-    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<u64> {
         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<BlockMeta> {
         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<B256> =
-            iter_code_hashes(&task.salt_witness.kvs).collect::<HashSet<_>>().into_iter().collect();
+        let codehashes: Vec<B256> = 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() {

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]),
+        }
+    }
+}

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,
     ];

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);
                 }
             }
         };