refactor: behavior-preserving cleanups and unused-API removals

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                               |

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                                                           |

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

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/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))?;
 

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

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

@@ -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

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;

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)
@@ -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() {

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