feat(era_manager): add era support + pre-fetching

.circleci/config.yml

@@ -214,7 +214,7 @@ jobs:
       tag: 1.79.0
     environment:
       RUSTFLAGS: "-D warnings"
-      RUST_LOG: "debug"
+      RUST_LOG: "debug,html5ever=error,selectors=error"
     steps:
       - checkout
       - checkout-submodules

e2store/Cargo.toml

@@ -26,4 +26,4 @@ surf = { version = "2.3.2", default-features = false, features = ["h1-client-rus
 [dev-dependencies]
 rstest = "0.18.2"
 tempfile = "3.3.0"
-
+tokio = { version = "1.14.0", features = ["full"] }

e2store/src/era.rs

@@ -49,12 +49,28 @@ impl Era {
         let slot_index_state = SlotIndexStateEntry::try_from(&file.entries[entries_length - 1])?;
 
         // Iterate over the block entries. Skip the first and last 3 entries.
-        let mut next_slot = slot_index_block.slot_index.starting_slot;
-        for idx in 1..entries_length - 3 {
-            let entry = &file.entries[idx];
-            let fork = get_beacon_fork(next_slot);
+        let slot_indexes = slot_index_block
+            .slot_index
+            .indices
+            .iter()
+            .enumerate()
+            .filter_map(|(i, index)| {
+                if *index != 0 {
+                    Some(slot_index_block.slot_index.starting_slot + i as u64)
+                } else {
+                    None
+                }
+            })
+            .collect::<Vec<u64>>();
+
+        ensure!(
+            slot_indexes.len() == entries_length - 4,
+            "invalid slot index block: incorrect count"
+        );
+        for (index, slot) in slot_indexes.into_iter().enumerate() {
+            let entry = &file.entries[index + 1];
+            let fork = get_beacon_fork(slot);
             let beacon_block = CompressedSignedBeaconBlock::try_from(entry, fork)?;
-            next_slot = beacon_block.block.slot() + 1;
             blocks.push(beacon_block);
         }
         let fork = get_beacon_fork(slot_index_state.slot_index.starting_slot);
@@ -80,6 +96,44 @@ impl Era {
         Ok(era_state.state)
     }
 
+    /// Iterate over beacon blocks.
+    pub fn iter_blocks(
+        raw_era: Vec<u8>,
+    ) -> anyhow::Result<impl Iterator<Item = anyhow::Result<CompressedSignedBeaconBlock>>> {
+        let file = E2StoreMemory::deserialize(&raw_era)?;
+        let entries_length = file.entries.len();
+        let block_index =
+            SlotIndexBlockEntry::try_from(&file.entries[entries_length - 2])?.slot_index;
+
+        let slot_indexes = block_index
+            .indices
+            .iter()
+            .enumerate()
+            .filter_map(|(i, index)| {
+                if *index != 0 {
+                    Some(block_index.starting_slot + i as u64)
+                } else {
+                    None
+                }
+            })
+            .collect::<Vec<u64>>();
+
+        ensure!(
+            slot_indexes.len() == entries_length - 4,
+            "invalid slot index block: incorrect count"
+        );
+
+        Ok(slot_indexes
+            .into_iter()
+            .enumerate()
+            .map(move |(index, slot)| {
+                let entry: Entry = file.entries[index + 1].clone();
+                let fork = get_beacon_fork(slot);
+                let beacon_block = CompressedSignedBeaconBlock::try_from(&entry, fork)?;
+                Ok(beacon_block)
+            }))
+    }
+
     #[allow(dead_code)]
     fn write(&self) -> anyhow::Result<Vec<u8>> {
         let mut entries: Vec<Entry> = vec![];
@@ -102,6 +156,17 @@ impl Era {
         file.write(&mut buf)?;
         Ok(buf)
     }
+
+    pub fn contains(&self, block_number: u64) -> bool {
+        if self.blocks.is_empty() {
+            return false;
+        }
+        let first_block_number = self.blocks[0].block.execution_block_number();
+        let last_block_number = self.blocks[self.blocks.len() - 1]
+            .block
+            .execution_block_number();
+        (first_block_number..=last_block_number).contains(&block_number)
+    }
 }
 
 #[derive(Clone, PartialEq, Debug)]
@@ -110,7 +175,7 @@ pub struct CompressedSignedBeaconBlock {
 }
 
 impl CompressedSignedBeaconBlock {
-    fn try_from(entry: &Entry, fork: ForkName) -> Result<Self, anyhow::Error> {
+    pub fn try_from(entry: &Entry, fork: ForkName) -> Result<Self, anyhow::Error> {
         ensure!(
             entry.header.type_ == 0x01,
             "invalid compressed signed beacon block entry: incorrect header type"
@@ -257,7 +322,7 @@ impl TryFrom<Entry> for SlotIndexBlock {
 // slot-index := starting-slot | index | index | index ... | count
 #[derive(Clone, Eq, PartialEq, Debug)]
 pub struct SlotIndexStateEntry {
-    slot_index: SlotIndexState,
+    pub slot_index: SlotIndexState,
 }
 
 impl TryFrom<&Entry> for SlotIndexStateEntry {
@@ -304,7 +369,7 @@ impl TryInto<Entry> for SlotIndexStateEntry {
 
 #[derive(Clone, Eq, PartialEq, Debug)]
 pub struct SlotIndexState {
-    starting_slot: u64,
+    pub starting_slot: u64,
     indices: [u64; 1],
     count: u64,
 }
@@ -326,7 +391,7 @@ impl TryFrom<Entry> for SlotIndexState {
     }
 }
 
-fn get_beacon_fork(slot_index: u64) -> ForkName {
+pub fn get_beacon_fork(slot_index: u64) -> ForkName {
     if slot_index < 4_636_672 {
         panic!("e2store/era doesn't support this fork");
     } else if (4_636_672..6_209_536).contains(&slot_index) {

e2store/src/utils.rs

@@ -1,12 +1,12 @@
-use std::io;
+use std::{collections::HashMap, io};
 
 use anyhow::{anyhow, ensure, Error};
 use rand::{seq::SliceRandom, thread_rng};
 use scraper::{Html, Selector};
 use surf::Client;
 
 const ERA1_DIR_URL: &str = "https://era1.ethportal.net/";
-const ERA1_FILE_COUNT: usize = 1897;
+pub const ERA1_FILE_COUNT: usize = 1897;
 
 /// Fetches era1 files hosted on era1.ethportal.net and shuffles them
 pub async fn get_shuffled_era1_files(http_client: &Client) -> anyhow::Result<Vec<String>> {
@@ -47,3 +47,54 @@ pub fn underlying_io_error_kind(error: &Error) -> Option<io::ErrorKind> {
     }
     None
 }
+
+const ERA_DIR_URL: &str = "https://mainnet.era.nimbus.team/";
+
+/// Fetches era file download links
+pub async fn get_era_file_download_links(
+    http_client: &Client,
+) -> anyhow::Result<HashMap<u64, String>> {
+    let index_html = http_client
+        .get(ERA_DIR_URL)
+        .recv_string()
+        .await
+        .map_err(|e| anyhow!("{e}"))?;
+    let index_html = Html::parse_document(&index_html);
+    let selector = Selector::parse("a[href*='mainnet-']").expect("to be able to parse selector");
+    let era_files: HashMap<u64, String> = index_html
+        .select(&selector)
+        .map(|element| {
+            let href = element
+                .value()
+                .attr("href")
+                .expect("to be able to get href");
+            let epoch_index = href
+                .split('-')
+                .nth(1)
+                .expect("to be able to get epoch")
+                .parse::<u64>()
+                .expect("to be able to parse epoch");
+            (epoch_index, format!("{ERA_DIR_URL}{href}"))
+        })
+        .collect();
+    Ok(era_files)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn test_get_shuffled_era1_files() {
+        let http_client = Client::new();
+        let era1_files = get_shuffled_era1_files(&http_client).await.unwrap();
+        assert_eq!(era1_files.len(), ERA1_FILE_COUNT);
+    }
+
+    #[tokio::test]
+    async fn test_get_era_file_download_links() {
+        let http_client = Client::new();
+        let era_files = get_era_file_download_links(&http_client).await.unwrap();
+        assert!(!era_files.is_empty());
+    }
+}

ethportal-api/src/types/consensus/beacon_block.rs

@@ -174,6 +174,17 @@ impl SignedBeaconBlock {
             SignedBeaconBlock::Deneb(block) => block.message.slot,
         }
     }
+
+    /// Returns execution block number.
+    pub fn execution_block_number(&self) -> u64 {
+        match self {
+            SignedBeaconBlock::Bellatrix(block) => {
+                block.message.body.execution_payload.block_number
+            }
+            SignedBeaconBlock::Capella(block) => block.message.body.execution_payload.block_number,
+            SignedBeaconBlock::Deneb(block) => block.message.body.execution_payload.block_number,
+        }
+    }
 }
 
 #[cfg(test)]

ethportal-api/src/types/consensus/execution_payload.rs

@@ -123,6 +123,30 @@ impl ExecutionPayloadBellatrix {
 
         proof_hashes
     }
+
+    pub fn transaction_root(&self) -> B256 {
+        self.transactions.tree_hash_root()
+    }
+}
+
+impl ExecutionPayloadCapella {
+    pub fn transaction_root(&self) -> B256 {
+        self.transactions.tree_hash_root()
+    }
+
+    pub fn withdrawals_root(&self) -> B256 {
+        self.withdrawals.tree_hash_root()
+    }
+}
+
+impl ExecutionPayloadDeneb {
+    pub fn transaction_root(&self) -> B256 {
+        self.transactions.tree_hash_root()
+    }
+
+    pub fn withdrawals_root(&self) -> B256 {
+        self.withdrawals.tree_hash_root()
+    }
 }
 
 #[derive(Debug, PartialEq, Eq, Hash, Clone, Serialize, Deserialize, Encode, Decode, TreeHash)]

portal-bridge/src/bridge/state.rs

@@ -2,7 +2,7 @@ use std::{path::PathBuf, sync::Arc};
 
 use alloy_rlp::Decodable;
 use anyhow::anyhow;
-use e2store::{era1::Era1, utils::get_shuffled_era1_files};
+use e2store::utils::get_shuffled_era1_files;
 use eth_trie::{decode_node, node::Node, RootWithTrieDiff};
 use ethportal_api::{
     jsonrpsee::http_client::HttpClient,
@@ -24,6 +24,7 @@ use trin_execution::{
         create_account_content_key, create_account_content_value, create_contract_content_key,
         create_contract_content_value, create_storage_content_key, create_storage_content_value,
     },
+    era::manager::EraManager,
     execution::State,
     spec_id::get_spec_block_number,
     storage::utils::setup_temp_dir,
@@ -32,7 +33,7 @@ use trin_execution::{
     utils::full_nibble_path_to_address_hash,
 };
 use trin_metrics::bridge::BridgeMetricsReporter;
-use trin_validation::{constants::EPOCH_SIZE, oracle::HeaderOracle};
+use trin_validation::oracle::HeaderOracle;
 
 use crate::{
     bridge::history::SERVE_BLOCK_TIMEOUT,
@@ -102,8 +103,6 @@ impl StateBridge {
 
     async fn launch_state(&self, last_block: u64) -> anyhow::Result<()> {
         info!("Gossiping state data from block 0 to {last_block}");
-        let mut current_epoch_index = u64::MAX;
-        let mut current_raw_era1 = vec![];
         let temp_directory = setup_temp_dir()?;
 
         // Enable contract storage changes caching required for gossiping the storage trie
@@ -112,38 +111,21 @@ impl StateBridge {
             block_to_trace: BlockToTrace::None,
         };
         let mut state = State::new(Some(temp_directory.path().to_path_buf()), state_config)?;
-        for block_index in 0..=last_block {
-            info!("Gossipping state for block at height: {block_index}");
-            let epoch_index = block_index / EPOCH_SIZE;
-            // make sure we have the current era1 file loaded
-            if current_epoch_index != epoch_index {
-                let era1_path = self
-                    .era1_files
-                    .iter()
-                    .find(|file| file.contains(&format!("mainnet-{epoch_index:05}-")))
-                    .expect("to be able to find era1 file");
-                let raw_era1 = self
-                    .http_client
-                    .get(era1_path.clone())
-                    .recv_bytes()
-                    .await
-                    .unwrap_or_else(|err| {
-                        panic!("unable to read era1 file at path: {era1_path:?} : {err}")
-                    });
-                current_epoch_index = epoch_index;
-                current_raw_era1 = raw_era1;
-            }
+        let mut era_manager = EraManager::new().await?;
+        for block_number in 0..=last_block {
+            info!("Gossipping state for block at height: {block_number}");
+
+            let block = era_manager.get_block_by_number(block_number).await?;
 
             // process block
-            let block_tuple = Era1::get_tuple_by_index(&current_raw_era1, block_index % EPOCH_SIZE);
             let RootWithTrieDiff {
                 root: root_hash,
                 trie_diff: changed_nodes,
-            } = match block_index == 0 {
+            } = match block_number == 0 {
                 true => state
                     .initialize_genesis()
                     .map_err(|e| anyhow!("unable to create genesis state: {e}"))?,
-                false => state.process_block(&block_tuple)?,
+                false => state.process_block(block)?,
             };
 
             let walk_diff = TrieWalker::new(root_hash, changed_nodes);
@@ -153,7 +135,7 @@ impl StateBridge {
                 let account_proof = walk_diff.get_proof(*node);
 
                 // gossip the account
-                self.gossip_account(&account_proof, block_tuple.header.header.hash())
+                self.gossip_account(&account_proof, block.header.hash())
                     .await?;
 
                 let Some(encoded_last_node) = account_proof.proof.last() else {
@@ -183,7 +165,7 @@ impl StateBridge {
                     self.gossip_contract_bytecode(
                         address_hash,
                         &account_proof,
-                        block_tuple.header.header.hash(),
+                        block.header.hash(),
                         account.code_hash,
                         code,
                     )
@@ -202,7 +184,7 @@ impl StateBridge {
                         &account_proof,
                         &storage_proof,
                         address_hash,
-                        block_tuple.header.header.hash(),
+                        block.header.hash(),
                     )
                     .await?;
                 }

trin-execution/Cargo.toml

@@ -25,6 +25,7 @@ hashbrown = "0.14.0"
 lazy_static = "1.4.0"
 parking_lot = "0.11.2"
 prometheus_exporter = "0.8.4"
+rayon = "1.10.0"
 revm = { version = "8.0.0", features = ["std", "secp256k1", "serde-json"], default-features = false }
 revm-inspectors = { git = "https://github.com/paradigmxyz/evm-inspectors", rev = "848d568" }
 revm-primitives = { version = "3.1.0", features = ["std", "serde"], default-features = false }