reth_era_downloader/
fs.rs

1use crate::{EraMeta, BLOCKS_PER_FILE};
2use alloy_primitives::{hex, hex::ToHexExt, BlockNumber};
3use eyre::{eyre, OptionExt};
4use futures_util::{stream, Stream};
5use reth_era::common::file_ops::EraFileType;
6use reth_fs_util as fs;
7use sha2::{Digest, Sha256};
8use std::{fmt::Debug, fs::DirEntry, io, io::BufRead, path::Path, str::FromStr};
9
10/// Creates a new ordered asynchronous [`Stream`] of ERA1 files read from `dir`.
11pub fn read_dir(
12    dir: impl AsRef<Path> + Send + Sync + 'static,
13    start_from: BlockNumber,
14) -> eyre::Result<impl Stream<Item = eyre::Result<EraLocalMeta>> + Send + Sync + 'static + Unpin> {
15    let mut checksums = None;
16
17    // read all the files in the given dir and also pick up the checksums file
18    let entries = sorted_era_entries(
19        dir,
20        |ty| matches!(ty, EraFileType::Era1 | EraFileType::Ere),
21        |path| {
22            if path.file_name() == Some("checksums.txt".as_ref()) {
23                let reader = io::BufReader::new(fs::open(path)?);
24                checksums = Some(reader.lines());
25            }
26            Ok(())
27        },
28    )?;
29    let mut checksums = checksums.ok_or_eyre("Missing file `checksums.txt` in the `dir`")?;
30
31    let start_index = start_from as usize / BLOCKS_PER_FILE;
32    for _ in 0..start_index {
33        // skip the first entries in the checksums iterator so that both iters align
34        checksums.next().transpose()?.ok_or_eyre("Got less checksums than ERA files")?;
35    }
36
37    Ok(stream::iter(entries.into_iter().skip_while(move |(n, _)| *n < start_index).map(
38        move |(_, path)| {
39            let expected_checksum =
40                checksums.next().transpose()?.ok_or_eyre("Got less checksums than ERA files")?;
41            let expected_checksum = hex::decode(expected_checksum)?;
42
43            let mut hasher = Sha256::new();
44            let mut reader = io::BufReader::new(fs::open(&path)?);
45
46            io::copy(&mut reader, &mut hasher)?;
47            let actual_checksum = hasher.finalize().to_vec();
48
49            if actual_checksum != expected_checksum {
50                return Err(eyre!(
51                    "Checksum mismatch, got: {}, expected: {}",
52                    actual_checksum.encode_hex(),
53                    expected_checksum.encode_hex()
54                ));
55            }
56
57            Ok(EraLocalMeta::new(path))
58        },
59    )))
60}
61
62/// Creates a new ordered asynchronous [`Stream`] of consensus `.era` files read from `dir`.
63///
64/// Unlike [`read_dir`], consensus `.era` files ship no `checksums.txt`, and their filenames encode
65/// an era (slot) number rather than a block number. Files are streamed in ascending era order; the
66/// import pipeline filters out blocks already present, so no block-level `start_from` skipping is
67/// done here.
68pub fn read_era_dir(
69    dir: impl AsRef<Path> + Send + Sync + 'static,
70) -> eyre::Result<impl Stream<Item = eyre::Result<EraLocalMeta>> + Send + Sync + 'static + Unpin> {
71    let entries = sorted_era_entries(dir, |ty| ty == EraFileType::Era, |_| Ok(()))?;
72
73    Ok(stream::iter(entries.into_iter().map(|(_, path)| Ok(EraLocalMeta::new(path)))))
74}
75
76/// Scans `dir` for ERA files whose type satisfies `accept`, returning them sorted by the number
77/// parsed from the `<network>-<number>-...<ext>` filename.
78///
79/// Files that don't match `accept` are passed to `on_other`, letting callers pick up sidecar files
80/// such as `checksums.txt`.
81fn sorted_era_entries(
82    dir: impl AsRef<Path>,
83    accept: impl Fn(EraFileType) -> bool,
84    mut on_other: impl FnMut(&Path) -> eyre::Result<()>,
85) -> eyre::Result<Vec<(usize, Box<Path>)>> {
86    let mut entries = fs::read_dir(dir)?
87        .filter_map(|entry| parse_era_entry(entry, &accept, &mut on_other).transpose())
88        .collect::<eyre::Result<Vec<_>>>()?;
89
90    entries.sort_by_key(|(number, _)| *number);
91
92    Ok(entries)
93}
94
95/// Parses one directory entry, returning `Some((number, path))` for an accepted ERA file.
96///
97/// Non-matching entries are forwarded to `on_other` (e.g. to pick up `checksums.txt`).
98fn parse_era_entry(
99    entry: io::Result<DirEntry>,
100    accept: &impl Fn(EraFileType) -> bool,
101    on_other: &mut impl FnMut(&Path) -> eyre::Result<()>,
102) -> eyre::Result<Option<(usize, Box<Path>)>> {
103    let path = entry?.path();
104
105    if let Some(name) = path.file_name().and_then(|name| name.to_str()) &&
106        EraFileType::from_filename(name).is_some_and(accept)
107    {
108        let parts = name.split('-').collect::<Vec<_>>();
109
110        if parts.len() >= 3 {
111            let number = usize::from_str(parts[1])?;
112
113            return Ok(Some((number, path.into_boxed_path())));
114        }
115    } else {
116        on_other(&path)?;
117    }
118
119    Ok(None)
120}
121
122/// Contains information about an ERA file that is on the local file-system and is read-only.
123#[derive(Debug)]
124pub struct EraLocalMeta {
125    path: Box<Path>,
126}
127
128impl EraLocalMeta {
129    const fn new(path: Box<Path>) -> Self {
130        Self { path }
131    }
132}
133
134impl<T: AsRef<Path>> PartialEq<T> for EraLocalMeta {
135    fn eq(&self, other: &T) -> bool {
136        self.as_ref().eq(other.as_ref())
137    }
138}
139
140impl AsRef<Path> for EraLocalMeta {
141    fn as_ref(&self) -> &Path {
142        self.path.as_ref()
143    }
144}
145
146impl EraMeta for EraLocalMeta {
147    /// A no-op.
148    fn mark_as_processed(&self) -> eyre::Result<()> {
149        Ok(())
150    }
151
152    fn path(&self) -> &Path {
153        &self.path
154    }
155}
reth_era_downloader/fs.rs

reth_era_downloader/
fs.rs
