feat(trin-execution): replace EraManager with E2HSManager (#1855)

Author: KolbyML

Cargo.lock

@@ -27,12 +27,14 @@ futures.workspace = true
 humanize-duration.workspace = true
 reqwest.workspace = true
 rstest.workspace = true
+revm-primitives.workspace = true
 serde_yaml.workspace = true
 ssz_types.workspace = true
 tempfile.workspace = true
 tokio.workspace = true
 tracing.workspace = true
 tree_hash.workspace = true
+trin-evm.workspace = true
 trin-execution.workspace = true
 trin-utils.workspace = true
 trin-validation.workspace = true

bin/e2hs-writer/Cargo.toml

@@ -8,10 +8,11 @@ use ethportal_api::consensus::constants::SLOTS_PER_HISTORICAL_ROOT;
 use reqwest::Client;
 use revm_primitives::hardfork::SpecId;
 use trin_evm::spec_id::get_spec_block_number;
+use trin_execution::e2hs::utils::download_raw_e2store_file;
 
-use super::{constants::FIRST_ERA_EPOCH_WITH_EXECUTION_PAYLOAD, utils::download_raw_era};
+const FIRST_ERA_EPOCH_WITH_EXECUTION_PAYLOAD: u64 = 574;
 
-pub struct EraBinarySearch {}
+pub struct EraBinarySearch;
 
 impl EraBinarySearch {
     /// For era files there isn't really a good way to know which era file has the block we want,
@@ -50,7 +51,7 @@ impl EraBinarySearch {
             // mid plus 1 so we just manually check the last era file
             if mid + 1 > last_epoch_index {
                 let era_to_check =
-                    download_raw_era(era_links[&(mid)].clone(), http_client.clone()).await?;
+                    download_raw_e2store_file(era_links[&mid].clone(), http_client.clone()).await?;
 
                 let decoded_era = Era::deserialize(&era_to_check)?;
                 if decoded_era.contains(block_number) {
@@ -76,7 +77,7 @@ impl EraBinarySearch {
                 || (mid_block_number..mid_plus_one_block_number).contains(&block_number)
             {
                 let era_to_check =
-                    download_raw_era(era_links[&(mid)].clone(), http_client.clone()).await?;
+                    download_raw_e2store_file(era_links[&mid].clone(), http_client.clone()).await?;
 
                 let decoded_era = Era::deserialize(&era_to_check)?;
                 if decoded_era.contains(block_number) {

bin/trin-execution/src/era/binary_search.rs renamed to bin/e2hs-writer/src/subcommands/single_generator/era_binary_search.rs

@@ -1,4 +1,5 @@
 mod e2hs_builder;
+mod era_binary_search;
 mod provider;
 
 use std::time::Instant;

bin/e2hs-writer/src/subcommands/single_generator/mod.rs

@@ -20,7 +20,8 @@ use reqwest::{
     Client,
 };
 use tracing::info;
-use trin_execution::era::binary_search::EraBinarySearch;
+
+use crate::subcommands::single_generator::era_binary_search::EraBinarySearch;
 
 pub enum EraSource {
     // processed era1 file

bin/e2hs-writer/src/subcommands/single_generator/provider.rs

@@ -263,7 +263,7 @@ impl StateBridge {
         trin_execution: &mut TrinExecution,
     ) -> anyhow::Result<()> {
         let block_hash = trin_execution
-            .era_manager
+            .e2hs_manager
             .lock()
             .await
             .last_fetched_block()

bin/portal-bridge/src/bridge/state.rs

@@ -9,7 +9,7 @@ Currently the main priority is executing to the head of the chain so we can goss
 
 ## How to run
 ```bash
-cargo run -p trin-execution
+cargo run --release -p trin-execution
 ```
 
 ### Want to get a trace of the EVM's execution?

bin/trin-execution/README.md

@@ -137,7 +137,7 @@ mod tests {
     };
     use ethportal_api::consensus::{beacon_block::SignedBeaconBlock, fork::ForkName};
 
-    use crate::era::beacon::ProcessBeaconBlock;
+    use crate::e2hs::beacon::ProcessBeaconBlock;
 
     #[tokio::test]
     async fn process_beacon_block() {

bin/trin-execution/src/era/beacon.rs renamed to bin/trin-execution/src/e2hs/beacon.rs

@@ -0,0 +1,142 @@
+use std::collections::HashMap;
+
+use anyhow::{anyhow, bail, ensure};
+use e2store::{e2hs::E2HSMemory, utils::get_e2hs_files};
+use reqwest::{
+    header::{HeaderMap, HeaderValue, CONTENT_TYPE},
+    Client,
+};
+use tokio::task::JoinHandle;
+use tracing::info;
+
+use super::{
+    types::{ProcessedBlock, ProcessedE2HS},
+    utils::download_and_processed_e2hs_file,
+};
+
+pub struct E2HSManager {
+    next_block_number: u64,
+    current_e2hs: Option<ProcessedE2HS>,
+    next_e2hs: Option<JoinHandle<anyhow::Result<ProcessedE2HS>>>,
+    http_client: Client,
+    e2hs_files: HashMap<u64, String>,
+}
+
+impl E2HSManager {
+    pub async fn new(next_block_number: u64) -> anyhow::Result<Self> {
+        let http_client = Client::builder()
+            .default_headers(HeaderMap::from_iter([(
+                CONTENT_TYPE,
+                HeaderValue::from_static("application/xml"),
+            )]))
+            .build()?;
+        let e2hs_files = get_e2hs_files(&http_client).await?;
+
+        let mut e2hs_manager = Self {
+            next_block_number,
+            current_e2hs: None,
+            next_e2hs: None,
+            http_client,
+            e2hs_files,
+        };
+
+        // initialize the next e2hs file
+        // The first time using e2hs_manager we need to find the current e2hs file we are starting
+        // from If the block is from a e2hs file we will need to binary search to find which
+        // e2hs file contains the block we want
+        let e2hs_files = e2hs_manager.e2hs_files.clone();
+        let http_client = e2hs_manager.http_client.clone();
+        e2hs_manager.next_e2hs = Some(tokio::spawn(async move {
+            Self::fetch_processed_e2hs_by_block_number(e2hs_files, http_client, next_block_number)
+                .await
+        }));
+
+        Ok(e2hs_manager)
+    }
+
+    pub fn next_block_number(&self) -> u64 {
+        self.next_block_number
+    }
+
+    pub async fn last_fetched_block(&self) -> anyhow::Result<&ProcessedBlock> {
+        let Some(current_e2hs) = &self.current_e2hs else {
+            panic!("current_e2hs should always be present, perhaps it wasn't initialized in E2HSManager::new()?");
+        };
+        ensure!(
+            self.next_block_number > 0,
+            "next_block_number should be greater than 0"
+        );
+        current_e2hs
+            .get_block(self.next_block_number - 1)
+            .ok_or_else(|| {
+                anyhow!(
+                    "No block found for next_block_number: {}",
+                    self.next_block_number - 1
+                )
+            })
+    }
+
+    pub async fn get_next_block(&mut self) -> anyhow::Result<&ProcessedBlock> {
+        let processed_e2hs = match &self.current_e2hs {
+            Some(processed_e2hs) if processed_e2hs.contains_block(self.next_block_number) => {
+                self.current_e2hs.as_ref().expect("current_e2hs to be some")
+            }
+            _ => {
+                let current_e2hs = self.get_next_processed_e2hs().await?;
+                self.current_e2hs.insert(current_e2hs)
+            }
+        };
+        ensure!(
+            processed_e2hs.contains_block(self.next_block_number),
+            "Block not found in e2hs file"
+        );
+        let block = processed_e2hs
+            .get_block(self.next_block_number)
+            .ok_or_else(|| {
+                anyhow!(
+                    "No block found for next_block_number: {}",
+                    self.next_block_number
+                )
+            })?;
+        self.next_block_number += 1;
+
+        Ok(block)
+    }
+
+    /// gets the next e2hs file and processes it, and starts fetching the next one in the background
+    async fn get_next_processed_e2hs(&mut self) -> anyhow::Result<ProcessedE2HS> {
+        info!("Fetching the next e2hs file");
+
+        // Since E2HSManager is initialized in new(), next_e2hs should never be none
+        let new_current_e2hs = match self.next_e2hs.take() {
+            Some(e2hs_handle) => e2hs_handle.await??,
+            None => bail!("The next_e2hs handle can't be None"),
+        };
+
+        // Download the next e2hs file
+        let next_e2hs_index = new_current_e2hs.index + 1;
+        let next_e2hs_path = self.e2hs_files.get(&next_e2hs_index).cloned();
+        let http_client = self.http_client.clone();
+        let join_handle = tokio::spawn(async move {
+            let Some(next_e2hs_path) = next_e2hs_path else {
+                bail!("Unable to find next e2hs file's path: index {next_e2hs_index}");
+            };
+            download_and_processed_e2hs_file(next_e2hs_path, http_client.clone()).await
+        });
+        self.next_e2hs = Some(join_handle);
+
+        Ok(new_current_e2hs)
+    }
+
+    async fn fetch_processed_e2hs_by_block_number(
+        e2hs_files: HashMap<u64, String>,
+        http_client: Client,
+        block_number: u64,
+    ) -> anyhow::Result<ProcessedE2HS> {
+        let e2hs_index = E2HSMemory::index_from_block_number(block_number);
+        let Some(e2hs_path) = e2hs_files.get(&e2hs_index).cloned() else {
+            bail!("Unable to find e2hs file's path during initialization: index {e2hs_index}");
+        };
+        download_and_processed_e2hs_file(e2hs_path, http_client.clone()).await
+    }
+}

bin/trin-execution/src/e2hs/manager.rs

@@ -1,6 +1,4 @@
 pub mod beacon;
-pub mod binary_search;
-pub mod constants;
 pub mod manager;
 pub mod types;
 pub mod utils;