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,html5ever=error,selectors=error"
+      RUST_LOG: "debug,html5ever=error,selectors=error,discv5::service=info"
     steps:
       - checkout
       - checkout-submodules

e2store/src/era.rs

@@ -12,7 +12,7 @@ use std::{
     io::{Read, Write},
 };
 
-const SLOTS_PER_HISTORICAL_ROOT: usize = 8192;
+pub const SLOTS_PER_HISTORICAL_ROOT: usize = 8192;
 
 /// group := Version | block* | era-state | other-entries* | slot-index(block)? | slot-index(state)
 /// block := CompressedSignedBeaconBlock
@@ -47,22 +47,9 @@ impl Era {
 
         let slot_index_block = SlotIndexBlockEntry::try_from(&file.entries[entries_length - 2])?;
         let slot_index_state = SlotIndexStateEntry::try_from(&file.entries[entries_length - 1])?;
+        let slot_indexes = Era::get_block_slot_indexes(&slot_index_block);
 
-        // Iterate over the block entries. Skip the first and last 3 entries.
-        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>>();
-
+        // an era file should has 4 entries which are not blocks
         ensure!(
             slot_indexes.len() == entries_length - 4,
             "invalid slot index block: incorrect count"
@@ -102,21 +89,8 @@ impl Era {
     ) -> 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>>();
+        let block_index = SlotIndexBlockEntry::try_from(&file.entries[entries_length - 2])?;
+        let slot_indexes = Era::get_block_slot_indexes(&block_index);
 
         ensure!(
             slot_indexes.len() == entries_length - 4,
@@ -134,6 +108,22 @@ impl Era {
             }))
     }
 
+    fn get_block_slot_indexes(slot_index_block_entry: &SlotIndexBlockEntry) -> Vec<u64> {
+        slot_index_block_entry
+            .slot_index
+            .indices
+            .iter()
+            .enumerate()
+            .filter_map(|(i, index)| {
+                if *index != 0 {
+                    Some(slot_index_block_entry.slot_index.starting_slot + i as u64)
+                } else {
+                    None
+                }
+            })
+            .collect::<Vec<u64>>()
+    }
+
     #[allow(dead_code)]
     fn write(&self) -> anyhow::Result<Vec<u8>> {
         let mut entries: Vec<Entry> = vec![];

e2store/src/utils.rs

@@ -5,40 +5,10 @@ use rand::{seq::SliceRandom, thread_rng};
 use scraper::{Html, Selector};
 use surf::Client;
 
+const ERA_DIR_URL: &str = "https://mainnet.era.nimbus.team/";
 const ERA1_DIR_URL: &str = "https://era1.ethportal.net/";
 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>> {
-    let index_html = http_client
-        .get(ERA1_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 mut era1_files: Vec<String> = index_html
-        .select(&selector)
-        .map(|element| {
-            let href = element
-                .value()
-                .attr("href")
-                .expect("to be able to get href");
-            format!("{ERA1_DIR_URL}{href}")
-        })
-        .collect();
-    ensure!(
-        era1_files.len() == ERA1_FILE_COUNT,
-        format!(
-            "invalid era1 source, not enough era1 files found: expected {}, found {}",
-            ERA1_FILE_COUNT,
-            era1_files.len()
-        )
-    );
-    era1_files.shuffle(&mut thread_rng());
-    Ok(era1_files)
-}
-
 pub fn underlying_io_error_kind(error: &Error) -> Option<io::ErrorKind> {
     for cause in error.chain() {
         if let Some(io_error) = cause.downcast_ref::<io::Error>() {
@@ -48,14 +18,12 @@ 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(
+pub async fn download_era_links(
     http_client: &Client,
+    url: &str,
 ) -> anyhow::Result<HashMap<u64, String>> {
     let index_html = http_client
-        .get(ERA_DIR_URL)
+        .get(url)
         .recv_string()
         .await
         .map_err(|e| anyhow!("{e}"))?;
@@ -74,12 +42,43 @@ pub async fn get_era_file_download_links(
                 .expect("to be able to get epoch")
                 .parse::<u64>()
                 .expect("to be able to parse epoch");
-            (epoch_index, format!("{ERA_DIR_URL}{href}"))
+            (epoch_index, format!("{url}{href}"))
         })
         .collect();
     Ok(era_files)
 }
 
+pub async fn get_era_files(http_client: &Client) -> anyhow::Result<HashMap<u64, String>> {
+    let era_files = download_era_links(http_client, ERA_DIR_URL).await?;
+    ensure!(!era_files.is_empty(), "No era files found at {ERA_DIR_URL}");
+    ensure!(
+        (0..era_files.len()).all(|epoch| era_files.contains_key(&(epoch as u64))),
+        "Epoch indices are not starting from zero or not consecutive",
+    );
+    Ok(era_files)
+}
+
+pub async fn get_era1_files(http_client: &Client) -> anyhow::Result<HashMap<u64, String>> {
+    let era1_files = download_era_links(http_client, ERA1_DIR_URL).await?;
+    ensure!(
+        era1_files.len() == ERA1_FILE_COUNT,
+        format!(
+            "invalid era1 source, not enough era1 files found: expected {}, found {}",
+            ERA1_FILE_COUNT,
+            era1_files.len()
+        )
+    );
+    Ok(era1_files)
+}
+
+/// 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>> {
+    let era1_files = get_era1_files(http_client).await?;
+    let mut era1_files: Vec<String> = era1_files.into_values().collect();
+    era1_files.shuffle(&mut thread_rng());
+    Ok(era1_files)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -94,7 +93,7 @@ mod tests {
     #[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();
+        let era_files = get_era_files(&http_client).await.unwrap();
         assert!(!era_files.is_empty());
     }
 }

portal-bridge/src/bridge/state.rs

@@ -111,11 +111,16 @@ impl StateBridge {
             block_to_trace: BlockToTrace::None,
         };
         let mut state = State::new(Some(temp_directory.path().to_path_buf()), state_config)?;
-        let mut era_manager = EraManager::new().await?;
-        for block_number in 0..=last_block {
+        let starting_block = 0;
+        let mut era_manager = EraManager::new(starting_block).await?;
+        for block_number in starting_block..=last_block {
             info!("Gossipping state for block at height: {block_number}");
 
-            let block = era_manager.get_block_by_number(block_number).await?;
+            let block = if block_number == starting_block {
+                era_manager.get_current_block().await?
+            } else {
+                era_manager.get_next_block().await?
+            };
 
             // process block
             let RootWithTrieDiff {

trin-execution/src/era/binary_search.rs

@@ -1,10 +1,13 @@
 use e2store::{
     e2store::types::{Entry, Header as E2StoreHeader},
-    era::{get_beacon_fork, CompressedSignedBeaconBlock, Era},
-    utils::get_era_file_download_links,
+    era::{get_beacon_fork, CompressedSignedBeaconBlock, Era, SLOTS_PER_HISTORICAL_ROOT},
+    utils::get_era_files,
 };
+use revm_primitives::SpecId;
 use surf::Client;
 
+use crate::spec_id::get_spec_block_number;
+
 use super::{
     constants::FIRST_ERA_EPOCH_WITH_EXECUTION_PAYLOAD,
     types::ProcessedEra,
@@ -25,14 +28,20 @@ impl EraBinarySearch {
         http_client: Client,
         block_number: u64,
     ) -> anyhow::Result<ProcessedEra> {
-        let era_links = get_era_file_download_links(&http_client).await?;
-        let last_epoch_index = era_links.len() - 1;
+        if block_number < get_spec_block_number(SpecId::MERGE) {
+            return Err(anyhow::anyhow!(
+                "Block number is too low to be in any era file"
+            ));
+        }
+
+        let era_links = get_era_files(&http_client).await?;
         let mut start_epoch_index = FIRST_ERA_EPOCH_WITH_EXECUTION_PAYLOAD;
         let mut end_epoch_index =
             *era_links.keys().max().expect("Getting max shouldn't fail") as u64;
+        let last_epoch_index = end_epoch_index;
 
         while start_epoch_index <= end_epoch_index {
-            let mid = end_epoch_index + (start_epoch_index - end_epoch_index) / 2;
+            let mid = (start_epoch_index + end_epoch_index) / 2;
             let mid_block = EraBinarySearch::download_first_beacon_block_from_era(
                 mid,
                 era_links[&mid].clone(),
@@ -41,7 +50,16 @@ impl EraBinarySearch {
             .await?;
             let mid_block_number = mid_block.block.execution_block_number();
 
+            // this is an edge case where the block number is in the last era file, we can't check
+            // mid plus 1 so we just manually check the last era file
             if mid + 1 > last_epoch_index as u64 {
+                let era_to_check =
+                    download_raw_era(era_links[&(mid)].clone(), http_client.clone()).await?;
+
+                let decoded_era = Era::deserialize(&era_to_check)?;
+                if decoded_era.contains(block_number) {
+                    return process_era_file(era_to_check, mid);
+                }
                 return Err(anyhow::anyhow!("Block not found in any era file"));
             }
 
@@ -101,11 +119,15 @@ impl EraBinarySearch {
             .expect("to be able to download compressed beacon block");
         let entry = Entry::deserialize(&compressed_beacon_block)?;
 
-        let slot_index = (era_index - 1) * 8192;
+        let slot_index = Self::start_slot_index(era_index);
         let fork = get_beacon_fork(slot_index);
 
         let beacon_block = CompressedSignedBeaconBlock::try_from(&entry, fork)?;
 
         Ok(beacon_block)
     }
+
+    fn start_slot_index(era_index: u64) -> u64 {
+        (era_index - 1) * SLOTS_PER_HISTORICAL_ROOT as u64
+    }
 }