reth_era/
era1_file.rs

Help
1//! Represents a complete Era1 file
2//!
3//! The structure of an Era1 file follows the specification:
4//! `Version | block-tuple* | other-entries* | Accumulator | BlockIndex`
5//!
6//! See also <https://github.com/eth-clients/e2store-format-specs/blob/main/formats/era1.md>
7
8use crate::{
9    e2s_file::{E2StoreReader, E2StoreWriter},
10    e2s_types::{E2sError, Version},
11    era1_types::{BlockIndex, Era1Group, Era1Id, BLOCK_INDEX},
12    execution_types::{
13        self, Accumulator, BlockTuple, CompressedBody, CompressedHeader, CompressedReceipts,
14        TotalDifficulty, MAX_BLOCKS_PER_ERA1,
15    },
16};
17use alloy_primitives::BlockNumber;
18use std::{
19    fs::File,
20    io::{Read, Seek, Write},
21    path::Path,
22};
23
24/// Era1 file interface
25#[derive(Debug)]
26pub struct Era1File {
27    /// Version record, must be the first record in the file
28    pub version: Version,
29
30    /// Main content group of the Era1 file
31    pub group: Era1Group,
32
33    /// File identifier
34    pub id: Era1Id,
35}
36
37impl Era1File {
38    /// Create a new [`Era1File`]
39    pub fn new(group: Era1Group, id: Era1Id) -> Self {
40        Self { version: Version, group, id }
41    }
42
43    /// Get a block by its number, if present in this file
44    pub fn get_block_by_number(&self, number: BlockNumber) -> Option<&BlockTuple> {
45        let index = (number - self.group.block_index.starting_number) as usize;
46        (index < self.group.blocks.len()).then(|| &self.group.blocks[index])
47    }
48
49    /// Get the range of block numbers contained in this file
50    pub fn block_range(&self) -> std::ops::RangeInclusive<BlockNumber> {
51        let start = self.group.block_index.starting_number;
52        let end = start + (self.group.blocks.len() as u64) - 1;
53        start..=end
54    }
55
56    /// Check if this file contains a specific block number
57    pub fn contains_block(&self, number: BlockNumber) -> bool {
58        self.block_range().contains(&number)
59    }
60}
61/// Reader for Era1 files that builds on top of [`E2StoreReader`]
62#[derive(Debug)]
63pub struct Era1Reader<R: Read> {
64    reader: E2StoreReader<R>,
65}
66
67impl<R: Read + Seek> Era1Reader<R> {
68    /// Create a new [`Era1Reader`]
69    pub fn new(reader: R) -> Self {
70        Self { reader: E2StoreReader::new(reader) }
71    }
72
73    /// Reads and parses an Era1 file from the underlying reader, assembling all components
74    /// into a complete [`Era1File`] with an [`Era1Id`] that includes the provided network name.
75    pub fn read(&mut self, network_name: String) -> Result<Era1File, E2sError> {
76        // Validate version entry
77        let _version_entry = match self.reader.read_version()? {
78            Some(entry) if entry.is_version() => entry,
79            Some(_) => return Err(E2sError::Ssz("First entry is not a Version entry".to_string())),
80            None => return Err(E2sError::Ssz("Empty Era1 file".to_string())),
81        };
82
83        // Read all entries
84        let entries = self.reader.entries()?;
85
86        // Skip the first entry since it's the version
87        let entries = entries.into_iter().skip(1).collect::<Vec<_>>();
88
89        // Temporary storage for block components
90        let mut headers = Vec::new();
91        let mut bodies = Vec::new();
92        let mut receipts = Vec::new();
93        let mut difficulties = Vec::new();
94        let mut other_entries = Vec::new();
95        let mut accumulator = None;
96        let mut block_index = None;
97
98        // Process entries in the order they appear
99        for entry in entries {
100            match entry.entry_type {
101                execution_types::COMPRESSED_HEADER => {
102                    headers.push(CompressedHeader::from_entry(&entry)?);
103                }
104                execution_types::COMPRESSED_BODY => {
105                    bodies.push(CompressedBody::from_entry(&entry)?);
106                }
107                execution_types::COMPRESSED_RECEIPTS => {
108                    receipts.push(CompressedReceipts::from_entry(&entry)?);
109                }
110                execution_types::TOTAL_DIFFICULTY => {
111                    difficulties.push(TotalDifficulty::from_entry(&entry)?);
112                }
113                execution_types::ACCUMULATOR => {
114                    if accumulator.is_some() {
115                        return Err(E2sError::Ssz("Multiple accumulator entries found".to_string()));
116                    }
117                    accumulator = Some(Accumulator::from_entry(&entry)?);
118                }
119                BLOCK_INDEX => {
120                    if block_index.is_some() {
121                        return Err(E2sError::Ssz("Multiple block index entries found".to_string()));
122                    }
123                    block_index = Some(BlockIndex::from_entry(&entry)?);
124                }
125                _ => {
126                    other_entries.push(entry);
127                }
128            }
129        }
130
131        // Ensure we have matching counts for block components
132        if headers.len() != bodies.len() ||
133            headers.len() != receipts.len() ||
134            headers.len() != difficulties.len()
135        {
136            return Err(E2sError::Ssz(format!(
137            "Mismatched block component counts: headers={}, bodies={}, receipts={}, difficulties={}",
138            headers.len(), bodies.len(), receipts.len(), difficulties.len()
139        )));
140        }
141
142        let mut blocks = Vec::with_capacity(headers.len());
143        for i in 0..headers.len() {
144            blocks.push(BlockTuple::new(
145                headers[i].clone(),
146                bodies[i].clone(),
147                receipts[i].clone(),
148                difficulties[i].clone(),
149            ));
150        }
151
152        let accumulator = accumulator
153            .ok_or_else(|| E2sError::Ssz("Era1 file missing accumulator entry".to_string()))?;
154
155        let block_index = block_index
156            .ok_or_else(|| E2sError::Ssz("Era1 file missing block index entry".to_string()))?;
157
158        let mut group = Era1Group::new(blocks, accumulator, block_index.clone());
159
160        // Add other entries
161        for entry in other_entries {
162            group.add_entry(entry);
163        }
164
165        let id = Era1Id::new(
166            network_name,
167            block_index.starting_number,
168            block_index.offsets.len() as u32,
169        );
170
171        Ok(Era1File::new(group, id))
172    }
173}
174
175impl Era1Reader<File> {
176    /// Opens and reads an Era1 file from the given path
177    pub fn open<P: AsRef<Path>>(
178        path: P,
179        network_name: impl Into<String>,
180    ) -> Result<Era1File, E2sError> {
181        let file = File::open(path).map_err(E2sError::Io)?;
182        let mut reader = Self::new(file);
183        reader.read(network_name.into())
184    }
185}
186
187/// Writer for Era1 files that builds on top of [`E2StoreWriter`]
188#[derive(Debug)]
189pub struct Era1Writer<W: Write> {
190    writer: E2StoreWriter<W>,
191    has_written_version: bool,
192    has_written_blocks: bool,
193    has_written_accumulator: bool,
194    has_written_block_index: bool,
195}
196
197impl<W: Write> Era1Writer<W> {
198    /// Create a new [`Era1Writer`]
199    pub fn new(writer: W) -> Self {
200        Self {
201            writer: E2StoreWriter::new(writer),
202            has_written_version: false,
203            has_written_blocks: false,
204            has_written_accumulator: false,
205            has_written_block_index: false,
206        }
207    }
208
209    /// Write the version entry
210    pub fn write_version(&mut self) -> Result<(), E2sError> {
211        if self.has_written_version {
212            return Ok(());
213        }
214
215        self.writer.write_version()?;
216        self.has_written_version = true;
217        Ok(())
218    }
219
220    /// Write a complete [`Era1File`] to the underlying writer
221    pub fn write_era1_file(&mut self, era1_file: &Era1File) -> Result<(), E2sError> {
222        // Write version
223        self.write_version()?;
224
225        // Ensure blocks are written before other entries
226        if era1_file.group.blocks.len() > MAX_BLOCKS_PER_ERA1 {
227            return Err(E2sError::Ssz("Era1 file cannot contain more than 8192 blocks".to_string()));
228        }
229
230        // Write all blocks
231        for block in &era1_file.group.blocks {
232            self.write_block(block)?;
233        }
234
235        // Write other entries
236        for entry in &era1_file.group.other_entries {
237            self.writer.write_entry(entry)?;
238        }
239
240        // Write accumulator
241        self.write_accumulator(&era1_file.group.accumulator)?;
242
243        // Write block index
244        self.write_block_index(&era1_file.group.block_index)?;
245
246        // Flush the writer
247        self.writer.flush()?;
248
249        Ok(())
250    }
251
252    /// Write a single block tuple
253    pub fn write_block(
254        &mut self,
255        block_tuple: &crate::execution_types::BlockTuple,
256    ) -> Result<(), E2sError> {
257        if !self.has_written_version {
258            self.write_version()?;
259        }
260
261        if self.has_written_accumulator || self.has_written_block_index {
262            return Err(E2sError::Ssz(
263                "Cannot write blocks after accumulator or block index".to_string(),
264            ));
265        }
266
267        // Write header
268        let header_entry = block_tuple.header.to_entry();
269        self.writer.write_entry(&header_entry)?;
270
271        // Write body
272        let body_entry = block_tuple.body.to_entry();
273        self.writer.write_entry(&body_entry)?;
274
275        // Write receipts
276        let receipts_entry = block_tuple.receipts.to_entry();
277        self.writer.write_entry(&receipts_entry)?;
278
279        // Write difficulty
280        let difficulty_entry = block_tuple.total_difficulty.to_entry();
281        self.writer.write_entry(&difficulty_entry)?;
282
283        self.has_written_blocks = true;
284
285        Ok(())
286    }
287
288    /// Write the accumulator
289    pub fn write_accumulator(&mut self, accumulator: &Accumulator) -> Result<(), E2sError> {
290        if !self.has_written_version {
291            self.write_version()?;
292        }
293
294        if self.has_written_accumulator {
295            return Err(E2sError::Ssz("Accumulator already written".to_string()));
296        }
297
298        if self.has_written_block_index {
299            return Err(E2sError::Ssz("Cannot write accumulator after block index".to_string()));
300        }
301
302        let accumulator_entry = accumulator.to_entry();
303        self.writer.write_entry(&accumulator_entry)?;
304        self.has_written_accumulator = true;
305
306        Ok(())
307    }
308
309    /// Write the block index
310    pub fn write_block_index(&mut self, block_index: &BlockIndex) -> Result<(), E2sError> {
311        if !self.has_written_version {
312            self.write_version()?;
313        }
314
315        if self.has_written_block_index {
316            return Err(E2sError::Ssz("Block index already written".to_string()));
317        }
318
319        let block_index_entry = block_index.to_entry();
320        self.writer.write_entry(&block_index_entry)?;
321        self.has_written_block_index = true;
322
323        Ok(())
324    }
325
326    /// Flush any buffered data to the underlying writer
327    pub fn flush(&mut self) -> Result<(), E2sError> {
328        self.writer.flush()
329    }
330}
331
332impl Era1Writer<File> {
333    /// Creates a new file at the specified path and writes the [`Era1File`] to it
334    pub fn create<P: AsRef<Path>>(path: P, era1_file: &Era1File) -> Result<(), E2sError> {
335        let file = File::create(path).map_err(E2sError::Io)?;
336        let mut writer = Self::new(file);
337        writer.write_era1_file(era1_file)?;
338        Ok(())
339    }
340
341    /// Creates a new file in the specified directory with a filename derived from the
342    /// [`Era1File`]'s ID using the standardized Era1 file naming convention
343    pub fn create_with_id<P: AsRef<Path>>(
344        directory: P,
345        era1_file: &Era1File,
346    ) -> Result<(), E2sError> {
347        let filename = era1_file.id.to_file_name();
348        let path = directory.as_ref().join(filename);
349        Self::create(path, era1_file)
350    }
351}
352
353#[cfg(test)]
354mod tests {
355    use super::*;
356    use crate::execution_types::{
357        Accumulator, BlockTuple, CompressedBody, CompressedHeader, CompressedReceipts,
358        TotalDifficulty,
359    };
360    use alloy_primitives::{B256, U256};
361    use std::io::Cursor;
362    use tempfile::tempdir;
363
364    // Helper to create a sample block tuple for testing
365    fn create_test_block(number: BlockNumber, data_size: usize) -> BlockTuple {
366        let header_data = vec![(number % 256) as u8; data_size];
367        let header = CompressedHeader::new(header_data);
368
369        let body_data = vec![((number + 1) % 256) as u8; data_size * 2];
370        let body = CompressedBody::new(body_data);
371
372        let receipts_data = vec![((number + 2) % 256) as u8; data_size];
373        let receipts = CompressedReceipts::new(receipts_data);
374
375        let difficulty = TotalDifficulty::new(U256::from(number * 1000));
376
377        BlockTuple::new(header, body, receipts, difficulty)
378    }
379
380    // Helper to create a sample Era1File for testing
381    fn create_test_era1_file(
382        start_block: BlockNumber,
383        block_count: usize,
384        network: &str,
385    ) -> Era1File {
386        // Create blocks
387        let mut blocks = Vec::with_capacity(block_count);
388        for i in 0..block_count {
389            let block_num = start_block + i as u64;
390            blocks.push(create_test_block(block_num, 32));
391        }
392
393        let accumulator = Accumulator::new(B256::from([0xAA; 32]));
394
395        let mut offsets = Vec::with_capacity(block_count);
396        for i in 0..block_count {
397            offsets.push(i as i64 * 100);
398        }
399        let block_index = BlockIndex::new(start_block, offsets);
400        let group = Era1Group::new(blocks, accumulator, block_index);
401        let id = Era1Id::new(network, start_block, block_count as u32);
402
403        Era1File::new(group, id)
404    }
405
406    #[test]
407    fn test_era1_roundtrip_memory() -> Result<(), E2sError> {
408        // Create a test Era1File
409        let start_block = 1000;
410        let era1_file = create_test_era1_file(1000, 5, "testnet");
411
412        // Write to memory buffer
413        let mut buffer = Vec::new();
414        {
415            let mut writer = Era1Writer::new(&mut buffer);
416            writer.write_era1_file(&era1_file)?;
417        }
418
419        // Read back from memory buffer
420        let mut reader = Era1Reader::new(Cursor::new(&buffer));
421        let read_era1 = reader.read("testnet".to_string())?;
422
423        // Verify core properties
424        assert_eq!(read_era1.id.network_name, "testnet");
425        assert_eq!(read_era1.id.start_block, 1000);
426        assert_eq!(read_era1.id.block_count, 5);
427        assert_eq!(read_era1.group.blocks.len(), 5);
428
429        // Verify block properties
430        assert_eq!(read_era1.group.blocks[0].total_difficulty.value, U256::from(1000 * 1000));
431        assert_eq!(read_era1.group.blocks[1].total_difficulty.value, U256::from(1001 * 1000));
432
433        // Verify block data
434        assert_eq!(read_era1.group.blocks[0].header.data, vec![(start_block % 256) as u8; 32]);
435        assert_eq!(read_era1.group.blocks[0].body.data, vec![((start_block + 1) % 256) as u8; 64]);
436        assert_eq!(
437            read_era1.group.blocks[0].receipts.data,
438            vec![((start_block + 2) % 256) as u8; 32]
439        );
440
441        // Verify block access methods
442        assert!(read_era1.contains_block(1000));
443        assert!(read_era1.contains_block(1004));
444        assert!(!read_era1.contains_block(999));
445        assert!(!read_era1.contains_block(1005));
446
447        let block_1002 = read_era1.get_block_by_number(1002);
448        assert!(block_1002.is_some());
449        assert_eq!(block_1002.unwrap().header.data, vec![((start_block + 2) % 256) as u8; 32]);
450
451        Ok(())
452    }
453
454    #[test]
455    fn test_era1_roundtrip_file() -> Result<(), E2sError> {
456        // Create a temporary directory
457        let temp_dir = tempdir().expect("Failed to create temp directory");
458        let file_path = temp_dir.path().join("test_roundtrip.era1");
459
460        // Create and write `Era1File` to disk
461        let era1_file = create_test_era1_file(2000, 3, "mainnet");
462        Era1Writer::create(&file_path, &era1_file)?;
463
464        // Read it back
465        let read_era1 = Era1Reader::open(&file_path, "mainnet")?;
466
467        // Verify core properties
468        assert_eq!(read_era1.id.network_name, "mainnet");
469        assert_eq!(read_era1.id.start_block, 2000);
470        assert_eq!(read_era1.id.block_count, 3);
471        assert_eq!(read_era1.group.blocks.len(), 3);
472
473        // Verify blocks
474        for i in 0..3 {
475            let block_num = 2000 + i as u64;
476            let block = read_era1.get_block_by_number(block_num);
477            assert!(block.is_some());
478            assert_eq!(block.unwrap().header.data, vec![block_num as u8; 32]);
479        }
480
481        Ok(())
482    }
483}
reth_era/era1_file.rs

reth_era/
era1_file.rs
