reth_cli_commands/download/
manifest_cmd.rs

1use crate::download::manifest::{generate_manifest, SnapshotManifest};
2use clap::Parser;
3use eyre::{Result, WrapErr};
4use reth_db::{mdbx::DatabaseArguments, open_db_read_only, tables, Database};
5use reth_db_api::transaction::DbTx;
6use reth_primitives_traits::FastInstant as Instant;
7use reth_stages_types::StageId;
8use reth_static_file_types::DEFAULT_BLOCKS_PER_STATIC_FILE;
9use std::path::PathBuf;
10use tracing::{info, warn};
11
12/// Generate modular chunk archives and a snapshot manifest from a source datadir.
13///
14/// Archive naming convention:
15///   - Chunked: `{component}-{start}-{end}.tar.zst` (e.g. `transactions-0-499999.tar.zst`)
16#[derive(Debug, Parser)]
17pub struct SnapshotManifestCommand {
18    /// Source datadir containing static files.
19    #[arg(long, short = 'd')]
20    source_datadir: PathBuf,
21
22    /// Optional base URL where archives will be hosted.
23    #[arg(long)]
24    base_url: Option<String>,
25
26    /// Output directory where chunk archives and manifest.json are written.
27    #[arg(long, short = 'o')]
28    output_dir: PathBuf,
29
30    /// Block number this snapshot was taken at.
31    ///
32    /// If omitted, this is inferred from the source datadir's `Finish` stage checkpoint.
33    #[arg(long)]
34    block: Option<u64>,
35
36    /// Chain ID.
37    #[arg(long, default_value = "1")]
38    chain_id: u64,
39
40    /// Blocks per archive file for chunked components.
41    ///
42    /// If omitted, this is inferred from header static file ranges in the source datadir.
43    #[arg(long, value_parser = clap::value_parser!(u64).range(1..))]
44    blocks_per_file: Option<u64>,
45
46    /// Path to the manifest of a previously published snapshot of the same chain.
47    ///
48    /// Chunks sealed in both snapshots are hashed against it instead of being packaged again,
49    /// and the new manifest references their published archives. Fails if any sealed chunk's
50    /// files differ from the previous manifest.
51    #[arg(long)]
52    previous_manifest: Option<PathBuf>,
53}
54
55impl SnapshotManifestCommand {
56    /// Packages snapshot archives and writes the manifest file.
57    pub fn execute(self) -> Result<()> {
58        let block = match self.block {
59            Some(block) => block,
60            None => infer_snapshot_block(&self.source_datadir)?,
61        };
62        let blocks_per_file = match self.blocks_per_file {
63            Some(blocks_per_file) => blocks_per_file,
64            None => infer_blocks_per_file(&self.source_datadir)?,
65        };
66
67        info!(target: "reth::cli",
68            dir = ?self.source_datadir,
69            output = ?self.output_dir,
70            block,
71            blocks_per_file,
72            "Packaging modular snapshot archives"
73        );
74        let previous = self
75            .previous_manifest
76            .as_ref()
77            .map(|path| -> Result<SnapshotManifest> {
78                let json = reth_fs_util::read_to_string(path)?;
79                serde_json::from_str(&json)
80                    .wrap_err_with(|| format!("Invalid previous manifest: {}", path.display()))
81            })
82            .transpose()?;
83
84        let start = Instant::now();
85        let manifest = generate_manifest(
86            &self.source_datadir,
87            &self.output_dir,
88            self.base_url.as_deref(),
89            block,
90            self.chain_id,
91            blocks_per_file,
92            previous.as_ref(),
93        )?;
94
95        let num_components = manifest.components.len();
96        let json = serde_json::to_string_pretty(&manifest)?;
97        let output = self.output_dir.join("manifest.json");
98        reth_fs_util::write(&output, &json)?;
99        info!(target: "reth::cli",
100            path = ?output,
101            components = num_components,
102            block = manifest.block,
103            elapsed = ?start.elapsed(),
104            "Manifest written"
105        );
106
107        Ok(())
108    }
109}
110
111/// Infers the snapshot block from the source datadir.
112fn infer_snapshot_block(source_datadir: &std::path::Path) -> Result<u64> {
113    if let Ok(block) = infer_snapshot_block_from_db(source_datadir) {
114        return Ok(block);
115    }
116
117    let block = infer_snapshot_block_from_headers(source_datadir)?;
118    warn!(
119        target: "reth::cli",
120        block,
121        "Could not read Finish stage checkpoint from source DB, using header static-file tip"
122    );
123    Ok(block)
124}
125
126/// Reads the snapshot block from the source database Finish stage checkpoint.
127fn infer_snapshot_block_from_db(source_datadir: &std::path::Path) -> Result<u64> {
128    let candidates = [source_datadir.join("db"), source_datadir.to_path_buf()];
129
130    for db_path in candidates {
131        if !db_path.exists() {
132            continue;
133        }
134
135        let db = match open_db_read_only(&db_path, DatabaseArguments::default()) {
136            Ok(db) => db,
137            Err(_) => continue,
138        };
139
140        let tx = db.tx()?;
141        if let Some(checkpoint) = tx.get::<tables::StageCheckpoints>(StageId::Finish.to_string())? {
142            return Ok(checkpoint.block_number);
143        }
144    }
145
146    eyre::bail!(
147        "Could not infer --block from source DB (Finish checkpoint missing); pass --block manually"
148    );
149}
150
151/// Infers the snapshot block from the highest header static-file range.
152fn infer_snapshot_block_from_headers(source_datadir: &std::path::Path) -> Result<u64> {
153    let max_end = header_ranges(source_datadir)?
154        .into_iter()
155        .map(|(_, end)| end)
156        .max()
157        .ok_or_else(|| eyre::eyre!("No header static files found to infer --block"))?;
158    Ok(max_end)
159}
160
161/// Infers the static-file block span from header file ranges.
162fn infer_blocks_per_file(source_datadir: &std::path::Path) -> Result<u64> {
163    let mut inferred = None;
164    for (start, end) in header_ranges(source_datadir)? {
165        let span = end.saturating_sub(start).saturating_add(1);
166        if span == 0 {
167            continue;
168        }
169
170        if let Some(existing) = inferred {
171            if existing != span {
172                eyre::bail!(
173                    "Inconsistent header static file ranges; pass --blocks-per-file manually"
174                );
175            }
176        } else {
177            inferred = Some(span);
178        }
179    }
180
181    inferred.ok_or_else(|| {
182        eyre::eyre!(
183            "Could not infer --blocks-per-file from header static files; pass it manually (default is {DEFAULT_BLOCKS_PER_STATIC_FILE})"
184        )
185    })
186}
187
188/// Collects header static-file ranges from the source datadir.
189fn header_ranges(source_datadir: &std::path::Path) -> Result<Vec<(u64, u64)>> {
190    let static_files_dir = source_datadir.join("static_files");
191    let static_files_dir =
192        if static_files_dir.exists() { static_files_dir } else { source_datadir.to_path_buf() };
193
194    let entries = std::fs::read_dir(&static_files_dir).wrap_err_with(|| {
195        format!("Failed to read static files directory: {}", static_files_dir.display())
196    })?;
197
198    let mut ranges = Vec::new();
199    for entry in entries {
200        let entry = entry?;
201        let file_name = entry.file_name();
202        let file_name = file_name.to_string_lossy();
203        if let Some(range) = parse_headers_range(&file_name) {
204            ranges.push(range);
205        }
206    }
207
208    Ok(ranges)
209}
210
211/// Parses the block range from a header static-file name.
212fn parse_headers_range(file_name: &str) -> Option<(u64, u64)> {
213    let remainder = file_name.strip_prefix("static_file_headers_")?;
214    let (start, end_with_suffix) = remainder.split_once('_')?;
215
216    let start = start.parse::<u64>().ok()?;
217    let end_digits: String = end_with_suffix.chars().take_while(|ch| ch.is_ascii_digit()).collect();
218    let end = end_digits.parse::<u64>().ok()?;
219
220    Some((start, end))
221}
222
223#[cfg(test)]
224mod tests {
225    use super::*;
226    use tempfile::tempdir;
227
228    #[test]
229    fn parse_headers_range_works_with_suffixes() {
230        assert_eq!(parse_headers_range("static_file_headers_0_499999"), Some((0, 499_999)));
231        assert_eq!(
232            parse_headers_range("static_file_headers_500000_999999.jar"),
233            Some((500_000, 999_999))
234        );
235        assert_eq!(parse_headers_range("static_file_transactions_0_499999"), None);
236    }
237
238    #[test]
239    fn infer_blocks_per_file_from_header_ranges() {
240        let dir = tempdir().unwrap();
241        let sf = dir.path().join("static_files");
242        std::fs::create_dir_all(&sf).unwrap();
243        std::fs::write(sf.join("static_file_headers_0_499999"), []).unwrap();
244        std::fs::write(sf.join("static_file_headers_500000_999999.jar"), []).unwrap();
245
246        assert_eq!(infer_blocks_per_file(dir.path()).unwrap(), 500_000);
247    }
248
249    #[test]
250    fn infer_snapshot_block_from_headers_uses_max_end() {
251        let dir = tempdir().unwrap();
252        let sf = dir.path().join("static_files");
253        std::fs::create_dir_all(&sf).unwrap();
254        std::fs::write(sf.join("static_file_headers_0_499999"), []).unwrap();
255        std::fs::write(sf.join("static_file_headers_500000_999999"), []).unwrap();
256
257        assert_eq!(infer_snapshot_block_from_headers(dir.path()).unwrap(), 999_999);
258    }
259}
reth_cli_commands/download/manifest_cmd.rs

reth_cli_commands/download/
manifest_cmd.rs
