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,discv5::service=info"
     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

@@ -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,14 +47,17 @@ 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 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);
+        // an era file has 4 entries which are not blocks
+        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 +83,47 @@ 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])?;
+        let slot_indexes = Era::get_block_slot_indexes(&block_index);
+
+        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)
+            }))
+    }
+
+    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![];
@@ -102,6 +146,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 +165,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 +312,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 +359,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 +381,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,32 +1,65 @@
-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 ERA_DIR_URL: &str = "https://mainnet.era.nimbus.team/";
 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>> {
+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>() {
+            return Some(io_error.kind());
+        }
+    }
+    None
+}
+
+pub async fn download_era_links(
+    http_client: &Client,
+    url: &str,
+) -> anyhow::Result<HashMap<u64, String>> {
     let index_html = http_client
-        .get(ERA1_DIR_URL)
+        .get(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
+    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");
-            format!("{ERA1_DIR_URL}{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!("{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!(
@@ -35,15 +68,36 @@ pub async fn get_shuffled_era1_files(http_client: &Client) -> anyhow::Result<Vec
             era1_files.len()
         )
     );
+    ensure!(
+        (0..ERA1_FILE_COUNT).all(|epoch| era1_files.contains_key(&(epoch as u64))),
+        "Epoch indices are not starting from zero or not consecutive",
+    );
+    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)
 }
 
-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>() {
-            return Some(io_error.kind());
-        }
+#[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_files(&http_client).await.unwrap();
+        assert!(!era_files.is_empty());
     }
-    None
 }

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)]