Merge branch 'main' into minor_improvements_precompile

Author: edg-l

.github/workflows/manual_docker_performance_publish.yaml

@@ -1,4 +1,4 @@
-name: Publish Docker
+name: Publish Performance tag Docker
 
 on:
   workflow_dispatch:

Cargo.lock

@@ -30,6 +30,7 @@ eyre.workspace = true
 tokio = { workspace = true, features = ["full"] }
 clap.workspace = true
 charming = { version = "0.4.0", features = ["ssr"] }
+rkyv.workspace = true
 
 jemallocator = { version = "0.5.4", optional = true }
 

cmd/ethrex_replay/Cargo.toml

@@ -1,22 +1,23 @@
 use ethrex_common::types::blobs_bundle;
 use ethrex_common::types::{Block, block_execution_witness::ExecutionWitnessResult};
+use eyre::Context;
+use rkyv::rancor::Error;
+use rkyv::{Archive, Deserialize as RDeserialize, Serialize as RSerialize};
 use serde::{Deserialize, Serialize};
 use serde_with::serde_as;
-use std::{
-    fs::File,
-    io::{BufReader, BufWriter},
-};
+use std::io::Write;
+use std::{fs::File, io::BufWriter};
 
 #[serde_as]
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, RSerialize, RDeserialize, Archive)]
 pub struct L2Fields {
     #[serde_as(as = "[_; 48]")]
     pub blob_commitment: blobs_bundle::Commitment,
     #[serde_as(as = "[_; 48]")]
     pub blob_proof: blobs_bundle::Proof,
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, RSerialize, RDeserialize, Archive)]
 pub struct Cache {
     pub blocks: Vec<Block>,
     pub witness: ExecutionWitnessResult,
@@ -36,14 +37,18 @@ impl Cache {
 }
 
 pub fn load_cache(file_name: &str) -> eyre::Result<Cache> {
-    let file = BufReader::new(File::open(file_name)?);
-    Ok(serde_json::from_reader(file)?)
+    let file_data = std::fs::read(file_name)?;
+    let cache =
+        rkyv::from_bytes::<Cache, Error>(&file_data).wrap_err("Failed to deserialize with rkyv")?;
+    Ok(cache)
 }
 
 pub fn write_cache(cache: &Cache, file_name: &str) -> eyre::Result<()> {
     if cache.blocks.is_empty() {
         return Err(eyre::Error::msg("cache can't be empty"));
     }
-    let file = BufWriter::new(File::create(file_name)?);
-    Ok(serde_json::to_writer_pretty(file, cache)?)
+    let mut file = BufWriter::new(File::create(file_name)?);
+    let bytes = rkyv::to_bytes::<Error>(cache).wrap_err("Failed to serialize with rkyv")?;
+    file.write_all(&bytes)
+        .wrap_err("Failed to write binary data")
 }

cmd/ethrex_replay/src/cache.rs

@@ -488,6 +488,103 @@ impl SubcommandProve {
     }
 }
 
+#[derive(Subcommand)]
+enum SubcommandCache {
+    #[command(about = "Cache a single block.")]
+    Block {
+        #[arg(help = "Block to use. Uses the latest if not specified.")]
+        block: Option<usize>,
+        #[arg(long, env = "RPC_URL", required = true)]
+        rpc_url: Url,
+        #[arg(
+            long,
+            help = "Name of the network or genesis file. Supported: mainnet, holesky, sepolia, hoodi. Default: mainnet",
+            value_parser = clap::value_parser!(Network),
+            default_value_t = Network::default(),
+        )]
+        network: Network,
+    },
+    #[command(about = "Cache multiple blocks.")]
+    Blocks {
+        #[arg(help = "List of blocks to execute.", num_args = 1.., value_delimiter = ',')]
+        blocks: Vec<u64>,
+        #[arg(long, env = "RPC_URL", required = true)]
+        rpc_url: Url,
+        #[arg(
+            long,
+            help = "Name of the network or genesis file. Supported: mainnet, holesky, sepolia, hoodi. Default: mainnet",
+            value_parser = clap::value_parser!(Network),
+            default_value_t = Network::default(),
+        )]
+        network: Network,
+    },
+    #[command(about = "Cache a range of blocks")]
+    BlockRange {
+        #[arg(help = "Starting block. (Inclusive)")]
+        start: usize,
+        #[arg(help = "Ending block. (Inclusive)")]
+        end: usize,
+        #[arg(long, env = "RPC_URL", required = true)]
+        rpc_url: Url,
+        #[arg(
+            long,
+            help = "Name of the network or genesis file. Supported: mainnet, holesky, sepolia, hoodi. Default: mainnet",
+            value_parser = clap::value_parser!(Network),
+            default_value_t = Network::default(),
+        )]
+        network: Network,
+    },
+}
+
+impl SubcommandCache {
+    pub async fn run(self) -> eyre::Result<()> {
+        match self {
+            SubcommandCache::Block {
+                block,
+                rpc_url,
+                network,
+            } => {
+                let eth_client = EthClient::new(rpc_url.as_ref())?;
+                let block_identifier = or_latest(block)?;
+                let _ = get_blockdata(eth_client, network.clone(), block_identifier).await?;
+                if let Some(block_number) = block {
+                    info!("Block {block_number} data cached successfully.");
+                } else {
+                    info!("Latest block data cached successfully.");
+                }
+            }
+            SubcommandCache::Blocks {
+                mut blocks,
+                rpc_url,
+                network,
+            } => {
+                blocks.sort();
+                let eth_client = EthClient::new(rpc_url.as_ref())?;
+                for block_number in blocks {
+                    let _ = get_blockdata(
+                        eth_client.clone(),
+                        network.clone(),
+                        BlockIdentifier::Number(block_number),
+                    )
+                    .await?;
+                }
+                info!("Blocks data cached successfully.");
+            }
+            SubcommandCache::BlockRange {
+                start,
+                end,
+                rpc_url,
+                network,
+            } => {
+                let eth_client = EthClient::new(rpc_url.as_ref())?;
+                let _ = get_rangedata(eth_client, network, start, end).await?;
+                info!("Block from {start} to {end} data cached successfully.");
+            }
+        }
+        Ok(())
+    }
+}
+
 #[derive(Subcommand)]
 enum EthrexReplayCommand {
     #[command(
@@ -516,6 +613,11 @@ enum EthrexReplayCommand {
         )]
         network: Network,
     },
+    #[command(
+        subcommand,
+        about = "Store the state prior to the execution of the block"
+    )]
+    Cache(SubcommandCache),
 }
 
 pub async fn start() -> eyre::Result<()> {
@@ -539,6 +641,7 @@ pub async fn start() -> eyre::Result<()> {
             let cache = get_rangedata(eth_client, network, start, end).await?;
             plot(cache).await?;
         }
+        EthrexReplayCommand::Cache(cmd) => cmd.run().await?,
     };
     Ok(())
 }

cmd/ethrex_replay/src/cli.rs

@@ -32,7 +32,7 @@ pub async fn get_blockdata(
 
     let chain_config = network.get_genesis()?.config;
 
-    let file_name = format!("cache_{network}_{requested_block_number}.json");
+    let file_name = format!("cache_{network}_{requested_block_number}.bin");
 
     if let Ok(cache) = load_cache(&file_name).inspect_err(|e| warn!("Failed to load cache: {e}")) {
         info!("Getting block {requested_block_number} data from cache");
@@ -194,7 +194,7 @@ pub async fn get_rangedata(
 ) -> eyre::Result<Cache> {
     let chain_config = network.get_genesis()?.config;
 
-    let file_name = format!("cache_{network}_{from}-{to}.json");
+    let file_name = format!("cache_{network}_{from}-{to}.bin");
 
     if let Ok(cache) = load_cache(&file_name) {
         info!("Getting block range data from cache");
@@ -215,7 +215,7 @@ pub async fn get_batchdata(
     chain_config: ChainConfig,
     batch_number: u64,
 ) -> eyre::Result<Cache> {
-    let file_name = format!("cache_batch_{batch_number}.json");
+    let file_name = format!("cache_batch_{batch_number}.bin");
     if let Ok(cache) = load_cache(&file_name) {
         info!("Getting batch data from cache");
         return Ok(cache);

cmd/ethrex_replay/src/fetcher.rs

@@ -15,7 +15,7 @@ serde.workspace = true
 ethrex-common.workspace = true
 
 
-prometheus = { version = "0.13.4", optional = true }
+prometheus = { version = "0.13.4", features = ["process"], optional = true }
 axum = { workspace = true, optional = true }
 tracing-subscriber.workspace = true
 

crates/blockchain/metrics/Cargo.toml

@@ -2,7 +2,10 @@ use axum::{Router, routing::get};
 
 use crate::profiling::gather_profiling_metrics;
 
-use crate::{MetricsApiError, metrics_blocks::METRICS_BLOCKS, metrics_transactions::METRICS_TX};
+use crate::{
+    MetricsApiError, metrics_blocks::METRICS_BLOCKS, metrics_process::METRICS_PROCESS,
+    metrics_transactions::METRICS_TX,
+};
 
 pub async fn start_prometheus_metrics_api(
     address: String,
@@ -47,5 +50,11 @@ pub(crate) async fn get_metrics() -> String {
         }
     }
 
+    ret_string.push('\n');
+    match METRICS_PROCESS.gather_metrics() {
+        Ok(s) => ret_string.push_str(&s),
+        Err(_) => tracing::error!("Failed to register METRICS_PROCESS"),
+    };
+
     ret_string
 }

crates/blockchain/metrics/api.rs

@@ -0,0 +1,51 @@
+use prometheus::{Encoder, Registry, TextEncoder};
+use std::sync::LazyLock;
+
+use crate::MetricsError;
+
+pub static METRICS_PROCESS: LazyLock<MetricsProcess> = LazyLock::new(MetricsProcess::default);
+
+#[derive(Debug, Clone)]
+pub struct MetricsProcess;
+
+impl Default for MetricsProcess {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl MetricsProcess {
+    pub fn new() -> Self {
+        MetricsProcess
+    }
+
+    /// The Process collector gathers standard process metrics (CPU time, RSS, VSZ, FDs, threads, start_time).
+    /// But it only works on Linux. This is an initial implementation.
+    pub fn gather_metrics(&self) -> Result<String, MetricsError> {
+        let r = Registry::new();
+
+        // Register Prometheus' built-in Linux process metrics
+        #[cfg(target_os = "linux")]
+        {
+            use prometheus::process_collector::ProcessCollector;
+            r.register(Box::new(ProcessCollector::for_self()))
+                .map_err(|e| {
+                    MetricsError::PrometheusErr(format!(
+                        "Failed to register process collector: {}",
+                        e
+                    ))
+                })?;
+        }
+
+        let encoder = TextEncoder::new();
+        let metric_families = r.gather();
+
+        let mut buffer = Vec::new();
+        encoder
+            .encode(&metric_families, &mut buffer)
+            .map_err(|e| MetricsError::PrometheusErr(e.to_string()))?;
+
+        let res = String::from_utf8(buffer)?;
+        Ok(res)
+    }
+}

crates/blockchain/metrics/metrics_process.rs

@@ -4,6 +4,8 @@ pub mod api;
 pub mod l2;
 #[cfg(any(feature = "api", feature = "metrics"))]
 pub mod metrics_blocks;
+#[cfg(any(feature = "api", feature = "metrics"))]
+pub mod metrics_process;
 #[cfg(any(feature = "api", feature = "transactions"))]
 pub mod metrics_transactions;
 #[cfg(feature = "api")]