reth_cli_commands/download/
extract.rs

1use super::{
2    fetch::{ArchiveFetcher, DownloadedArchive},
3    progress::{
4        ArchiveExtractionProgress, ArchiveExtractionProgressHandle, DownloadProgress,
5        DownloadRequestLimiter, ProgressReader, SharedProgressReader,
6    },
7    session::DownloadSession,
8    MAX_DOWNLOAD_RETRIES, RETRY_BACKOFF_SECS,
9};
10use eyre::{Result, WrapErr};
11use lz4::Decoder;
12use reqwest::blocking::Client as BlockingClient;
13use reth_cli_util::cancellation::CancellationToken;
14use reth_fs_util as fs;
15use std::{
16    io::Read,
17    path::{Component, Path, PathBuf},
18    sync::{
19        atomic::{AtomicBool, Ordering},
20        Arc,
21    },
22    thread,
23    time::{Duration, Instant},
24};
25use tar::Archive;
26use tracing::{info, warn};
27use url::Url;
28use zstd::stream::read::Decoder as ZstdDecoder;
29
30const EXTENSION_TAR_LZ4: &str = ".tar.lz4";
31const EXTENSION_TAR_ZSTD: &str = ".tar.zst";
32const STREAMING_EXTRACTION_PROGRESS_MIN_FILE_SIZE: u64 = 64 * 1024 * 1024;
33const EXTRACTION_PROGRESS_POLL_INTERVAL: Duration = Duration::from_millis(100);
34
35/// Supported compression formats for snapshots
36#[derive(Debug, Clone, Copy)]
37pub(crate) enum CompressionFormat {
38    /// LZ4-compressed tar archive.
39    Lz4,
40    /// Zstandard-compressed tar archive.
41    Zstd,
42}
43
44impl CompressionFormat {
45    /// Detect compression format from file extension
46    pub(crate) fn from_url(url: &str) -> Result<Self> {
47        let path =
48            Url::parse(url).map(|u| u.path().to_string()).unwrap_or_else(|_| url.to_string());
49
50        if path.ends_with(EXTENSION_TAR_LZ4) {
51            Ok(Self::Lz4)
52        } else if path.ends_with(EXTENSION_TAR_ZSTD) {
53            Ok(Self::Zstd)
54        } else {
55            Err(eyre::eyre!(
56                "Unsupported file format. Expected .tar.lz4 or .tar.zst, got: {}",
57                path
58            ))
59        }
60    }
61}
62
63/// Extracts a compressed tar archive to the target directory with progress tracking.
64fn extract_archive<R: Read>(
65    reader: R,
66    total_size: u64,
67    format: CompressionFormat,
68    target_dir: &Path,
69    static_files_dir: Option<&Path>,
70    cancel_token: CancellationToken,
71) -> Result<()> {
72    let progress_reader = ProgressReader::new(reader, total_size, cancel_token);
73
74    match format {
75        CompressionFormat::Lz4 => {
76            let decoder = Decoder::new(progress_reader)?;
77            unpack_archive(Archive::new(decoder), target_dir, static_files_dir, None)?;
78        }
79        CompressionFormat::Zstd => {
80            let decoder = ZstdDecoder::new(progress_reader)?;
81            unpack_archive(Archive::new(decoder), target_dir, static_files_dir, None)?;
82        }
83    }
84
85    println!();
86    Ok(())
87}
88
89/// Extracts a compressed tar archive without progress tracking.
90pub(crate) fn extract_archive_raw<R: Read>(
91    reader: R,
92    format: CompressionFormat,
93    target_dir: &Path,
94    static_files_dir: Option<&Path>,
95    progress: Option<&mut ArchiveExtractionProgress>,
96) -> Result<()> {
97    match format {
98        CompressionFormat::Lz4 => {
99            unpack_archive(
100                Archive::new(Decoder::new(reader)?),
101                target_dir,
102                static_files_dir,
103                progress,
104            )?;
105        }
106        CompressionFormat::Zstd => {
107            unpack_archive(
108                Archive::new(ZstdDecoder::new(reader)?),
109                target_dir,
110                static_files_dir,
111                progress,
112            )?;
113        }
114    }
115
116    Ok(())
117}
118
119fn unpack_archive<R: Read>(
120    mut archive: Archive<R>,
121    target_dir: &Path,
122    static_files_dir: Option<&Path>,
123    mut progress: Option<&mut ArchiveExtractionProgress>,
124) -> Result<()> {
125    // Every entry goes through `unpack_entry` so link entries are rejected; `Archive::unpack`
126    // would create them.
127    let entries = archive.entries().wrap_err_with(|| {
128        format!("failed to read archive entries for `{}`", target_dir.display())
129    })?;
130
131    for entry in entries {
132        let mut entry = entry.wrap_err_with(|| {
133            format!("failed to read archive entry for `{}`", target_dir.display())
134        })?;
135        extract_entry_with_progress(
136            &mut entry,
137            target_dir,
138            static_files_dir,
139            progress.as_deref_mut(),
140        )?;
141    }
142
143    Ok(())
144}
145
146/// Returns the path within the static files directory, accepting tar's optional `./` prefix.
147pub(crate) fn static_file_relative_path(path: &Path) -> Option<&Path> {
148    path.strip_prefix(".").unwrap_or(path).strip_prefix("static_files").ok()
149}
150
151/// Fails if any existing component of `relative_path` below `root` is a symlink.
152///
153/// Components that do not exist yet are ignored. `root` itself is trusted and not checked. This
154/// is a point-in-time check: it does not stop a link from being created right after it returns.
155pub(crate) fn ensure_no_symlinks(root: &Path, relative_path: &Path) -> Result<()> {
156    let mut current = root.to_path_buf();
157    for part in relative_path.components() {
158        current.push(part);
159        match std::fs::symlink_metadata(&current) {
160            Ok(metadata) => eyre::ensure!(
161                !metadata.file_type().is_symlink(),
162                "Archive path `{}` contains a symlink",
163                current.display()
164            ),
165            Err(err) if err.kind() == std::io::ErrorKind::NotFound => {}
166            Err(err) => return Err(err.into()),
167        }
168    }
169    Ok(())
170}
171
172/// Extracts static files beneath their configured root without allowing archive paths or links
173/// to escape it.
174fn unpack_static_file<R: Read>(
175    entry: &mut tar::Entry<'_, R>,
176    static_files_dir: &Path,
177    relative_path: &Path,
178) -> Result<()> {
179    eyre::ensure!(
180        relative_path
181            .components()
182            .all(|part| matches!(part, Component::Normal(_) | Component::CurDir)),
183        "Invalid static file archive path: {}",
184        relative_path.display()
185    );
186    let entry_type = entry.header().entry_type();
187    eyre::ensure!(
188        entry_type.is_file() || entry_type.is_dir(),
189        "Unsupported static file archive entry"
190    );
191    fs::create_dir_all(static_files_dir)?;
192    let root = static_files_dir.to_path_buf();
193    let dest = root.join(relative_path);
194    // Check each existing ancestor before creating directories or replacing an output.
195    ensure_no_symlinks(&root, relative_path)?;
196    if relative_path.as_os_str().is_empty() {
197        return Ok(())
198    }
199    if let Some(parent) = dest.parent() {
200        fs::create_dir_all(parent)?;
201    }
202    entry.unpack(dest)?;
203    Ok(())
204}
205
206fn unpack_entry<R: Read>(
207    entry: &mut tar::Entry<'_, R>,
208    target_dir: &Path,
209    static_files_dir: Option<&Path>,
210) -> Result<()> {
211    // Links planted by an archive could redirect later extraction, or retry cleanup running
212    // concurrently for another archive, outside the datadir. Snapshots never contain links.
213    let entry_type = entry.header().entry_type();
214    eyre::ensure!(
215        !entry_type.is_symlink() && !entry_type.is_hard_link(),
216        "Archive links are unsupported: `{}`",
217        entry.path()?.display()
218    );
219    let path = entry.path()?.into_owned();
220    if let Some(static_files_dir) = static_files_dir &&
221        let Some(relative_path) = static_file_relative_path(&path)
222    {
223        unpack_static_file(entry, static_files_dir, relative_path)
224    } else {
225        entry.unpack_in(target_dir)?;
226        Ok(())
227    }
228}
229
230fn extract_entry_with_progress<R: Read>(
231    entry: &mut tar::Entry<'_, R>,
232    target_dir: &Path,
233    static_files_dir: Option<&Path>,
234    progress: Option<&mut ArchiveExtractionProgress>,
235) -> Result<()> {
236    let size = entry.header().entry_size().unwrap_or(0);
237    let entry_type = entry.header().entry_type();
238
239    if !entry_type.is_file() || size == 0 {
240        unpack_entry(entry, target_dir, static_files_dir).wrap_err_with(|| {
241            format!("failed to extract archive into `{}`", target_dir.display())
242        })?;
243        return Ok(())
244    }
245
246    if size < STREAMING_EXTRACTION_PROGRESS_MIN_FILE_SIZE {
247        unpack_entry(entry, target_dir, static_files_dir).wrap_err_with(|| {
248            format!("failed to extract archive into `{}`", target_dir.display())
249        })?;
250        if let Some(progress) = progress {
251            progress.record_extracted(size);
252        }
253        return Ok(())
254    }
255
256    let Some(progress_handle) = progress.as_ref().and_then(|progress| progress.handle()) else {
257        unpack_entry(entry, target_dir, static_files_dir).wrap_err_with(|| {
258            format!("failed to extract archive into `{}`", target_dir.display())
259        })?;
260        return Ok(())
261    };
262
263    let Some(entry_path) = entry_destination_path(entry, target_dir, static_files_dir)? else {
264        unpack_entry(entry, target_dir, static_files_dir).wrap_err_with(|| {
265            format!("failed to extract archive into `{}`", target_dir.display())
266        })?;
267        return Ok(())
268    };
269
270    let stop = Arc::new(AtomicBool::new(false));
271    let monitor = spawn_extraction_progress_monitor(entry_path, progress_handle, Arc::clone(&stop));
272    let unpack_result = unpack_entry(entry, target_dir, static_files_dir)
273        .wrap_err_with(|| format!("failed to extract archive into `{}`", target_dir.display()));
274    stop.store(true, Ordering::Relaxed);
275
276    let monitor_result = monitor.join();
277    unpack_result?;
278
279    monitor_result.map_err(|_| eyre::eyre!("extraction progress monitor panicked"))?;
280    Ok(())
281}
282
283fn entry_destination_path<R: Read>(
284    entry: &tar::Entry<'_, R>,
285    target_dir: &Path,
286    static_files_dir: Option<&Path>,
287) -> Result<Option<PathBuf>> {
288    let mut file_dst = target_dir.to_path_buf();
289    let path = entry.path().wrap_err("invalid path in archive entry")?;
290    let path = if let Some(static_files_dir) = static_files_dir &&
291        let Some(relative_path) = static_file_relative_path(&path)
292    {
293        file_dst = static_files_dir.to_path_buf();
294        relative_path
295    } else {
296        path.as_ref()
297    };
298
299    for part in path.components() {
300        match part {
301            Component::Prefix(..) | Component::RootDir | Component::CurDir => continue,
302            Component::ParentDir => return Ok(None),
303            Component::Normal(part) => file_dst.push(part),
304        }
305    }
306
307    if file_dst == target_dir {
308        return Ok(None)
309    }
310
311    Ok(Some(file_dst))
312}
313
314fn spawn_extraction_progress_monitor(
315    entry_path: PathBuf,
316    progress: ArchiveExtractionProgressHandle,
317    stop: Arc<AtomicBool>,
318) -> thread::JoinHandle<()> {
319    thread::spawn(move || {
320        let mut extracted = 0_u64;
321
322        loop {
323            record_extracted_file_bytes(&entry_path, &progress, &mut extracted);
324            if stop.load(Ordering::Relaxed) {
325                break;
326            }
327            thread::sleep(EXTRACTION_PROGRESS_POLL_INTERVAL);
328        }
329    })
330}
331
332fn record_extracted_file_bytes(
333    entry_path: &Path,
334    progress: &ArchiveExtractionProgressHandle,
335    extracted: &mut u64,
336) {
337    let Ok(meta) = fs::metadata(entry_path) else { return };
338    let len = meta.len();
339    if len > *extracted {
340        progress.record_extracted(len - *extracted);
341        *extracted = len;
342    }
343}
344
345/// Extracts a snapshot from a local file.
346fn extract_from_file(
347    path: &Path,
348    format: CompressionFormat,
349    target_dir: &Path,
350    static_files_dir: Option<&Path>,
351) -> Result<()> {
352    let file = std::fs::File::open(path)?;
353    let total_size = file.metadata()?.len();
354    info!(target: "reth::cli",
355        file = %path.display(),
356        size = %DownloadProgress::format_size(total_size),
357        "Extracting local archive"
358    );
359    let start = Instant::now();
360    extract_archive(
361        file,
362        total_size,
363        format,
364        target_dir,
365        static_files_dir,
366        CancellationToken::new(),
367    )?;
368    info!(target: "reth::cli",
369        file = %path.display(),
370        elapsed = %DownloadProgress::format_duration(start.elapsed()),
371        "Local extraction complete"
372    );
373    Ok(())
374}
375
376/// Streams a remote archive directly into the extractor without writing to disk.
377///
378/// On failure, retries from scratch up to [`MAX_DOWNLOAD_RETRIES`] times.
379pub(crate) fn streaming_download_and_extract(
380    url: &str,
381    format: CompressionFormat,
382    target_dir: &Path,
383    static_files_dir: Option<&Path>,
384    session: &DownloadSession,
385) -> Result<()> {
386    if let Some(path) = archive_file_url_path(url)? {
387        let size = path.metadata()?.len();
388        extract_from_file(&path, format, target_dir, static_files_dir)?;
389        session.record_archive_output_complete(size);
390        return Ok(())
391    }
392
393    let shared = session.progress();
394    let quiet = session.progress().is_some();
395    let mut last_error: Option<eyre::Error> = None;
396
397    for attempt in 1..=MAX_DOWNLOAD_RETRIES {
398        if attempt > 1 {
399            info!(target: "reth::cli",
400                url = %url,
401                attempt,
402                max = MAX_DOWNLOAD_RETRIES,
403                "Retrying streaming download from scratch"
404            );
405        }
406
407        let client = BlockingClient::builder().connect_timeout(Duration::from_secs(30)).build()?;
408        let _request_permit = session
409            .request_limiter()
410            .map(|limiter| limiter.acquire(session.progress(), session.cancel_token()))
411            .transpose()?;
412
413        let response = match client.get(url).send().and_then(|r| r.error_for_status()) {
414            Ok(r) => r,
415            Err(error) => {
416                let err = eyre::Error::from(error);
417                if attempt < MAX_DOWNLOAD_RETRIES {
418                    warn!(target: "reth::cli",
419                        url = %url,
420                        attempt,
421                        max = MAX_DOWNLOAD_RETRIES,
422                        err = %err,
423                        "Streaming request failed, retrying"
424                    );
425                }
426                last_error = Some(err);
427                if attempt < MAX_DOWNLOAD_RETRIES {
428                    std::thread::sleep(
429                        session.retry_delay(Duration::from_secs(RETRY_BACKOFF_SECS)),
430                    );
431                }
432                continue;
433            }
434        };
435
436        if !quiet && let Some(size) = response.content_length() {
437            info!(target: "reth::cli",
438                url = %url,
439                size = %DownloadProgress::format_size(size),
440                "Streaming archive"
441            );
442        }
443
444        let result = if let Some(progress) = shared {
445            let reader = SharedProgressReader { inner: response, progress: Arc::clone(progress) };
446            extract_archive_raw(reader, format, target_dir, static_files_dir, None)
447        } else {
448            let total_size = response.content_length().unwrap_or(0);
449            extract_archive(
450                response,
451                total_size,
452                format,
453                target_dir,
454                static_files_dir,
455                session.cancel_token().clone(),
456            )
457        };
458
459        match result {
460            Ok(()) => return Ok(()),
461            Err(error) => {
462                if attempt < MAX_DOWNLOAD_RETRIES {
463                    warn!(target: "reth::cli",
464                        url = %url,
465                        attempt,
466                        max = MAX_DOWNLOAD_RETRIES,
467                        err = %error,
468                        "Streaming extraction failed, retrying"
469                    );
470                }
471                last_error = Some(error);
472                if attempt < MAX_DOWNLOAD_RETRIES {
473                    std::thread::sleep(
474                        session.retry_delay(Duration::from_secs(RETRY_BACKOFF_SECS)),
475                    );
476                }
477            }
478        }
479    }
480
481    Err(last_error.unwrap_or_else(|| {
482        eyre::eyre!("Streaming download failed after {MAX_DOWNLOAD_RETRIES} attempts")
483    }))
484}
485
486/// Resolves a `file://` archive URL to its local path.
487fn archive_file_url_path(url: &str) -> Result<Option<PathBuf>> {
488    let Ok(parsed) = Url::parse(url) else { return Ok(None) };
489    if parsed.scheme() != "file" {
490        return Ok(None)
491    }
492
493    parsed
494        .to_file_path()
495        .map(Some)
496        .map_err(|_| eyre::eyre!("Invalid file:// archive URL path: {url}"))
497}
498
499/// Fetches the snapshot from a remote URL with resume support, then extracts it.
500fn download_and_extract(
501    url: &str,
502    format: CompressionFormat,
503    target_dir: &Path,
504    static_files_dir: Option<&Path>,
505    session: DownloadSession,
506) -> Result<()> {
507    let quiet = session.progress().is_some();
508    let fetcher = ArchiveFetcher::new(url.to_string(), target_dir, session.clone());
509    let DownloadedArchive { path: downloaded_path, size: total_size } = fetcher.download(None)?;
510
511    let file_name =
512        downloaded_path.file_name().map(|f| f.to_string_lossy().to_string()).unwrap_or_default();
513
514    if !quiet {
515        info!(target: "reth::cli",
516            file = %file_name,
517            size = %DownloadProgress::format_size(total_size),
518            "Extracting archive"
519        );
520    }
521    let file = fs::open(&downloaded_path)?;
522
523    if quiet {
524        extract_archive_raw(file, format, target_dir, static_files_dir, None)?;
525    } else {
526        extract_archive(
527            file,
528            total_size,
529            format,
530            target_dir,
531            static_files_dir,
532            session.cancel_token().clone(),
533        )?;
534        info!(target: "reth::cli",
535            file = %file_name,
536            "Extraction complete"
537        );
538    }
539
540    fetcher.cleanup_downloaded_files();
541    session.record_archive_output_complete(total_size);
542
543    Ok(())
544}
545
546/// Downloads and extracts a snapshot, blocking until finished.
547///
548/// Supports `file://` URLs for local files and HTTP(S) URLs for remote downloads.
549/// When `resumable` is true, downloads to a `.part` file first with HTTP Range resume
550/// support. Otherwise streams directly into the extractor.
551fn blocking_download_and_extract(
552    url: &str,
553    target_dir: &Path,
554    static_files_dir: Option<&Path>,
555    resumable: bool,
556    request_limiter: Option<Arc<DownloadRequestLimiter>>,
557    cancel_token: CancellationToken,
558    retry_backoff: Option<Duration>,
559) -> Result<()> {
560    let format = CompressionFormat::from_url(url)?;
561
562    if let Ok(parsed_url) = Url::parse(url) &&
563        parsed_url.scheme() == "file"
564    {
565        let session = DownloadSession::new(None, request_limiter, cancel_token)
566            .with_retry_backoff(retry_backoff);
567        let file_path = parsed_url
568            .to_file_path()
569            .map_err(|_| eyre::eyre!("Invalid file:// URL path: {}", url))?;
570        let result = extract_from_file(&file_path, format, target_dir, static_files_dir);
571        if result.is_ok() {
572            session.record_archive_output_complete(file_path.metadata()?.len());
573        }
574        result
575    } else if let Some(request_limiter) = request_limiter {
576        download_and_extract(
577            url,
578            format,
579            target_dir,
580            static_files_dir,
581            DownloadSession::new(None, Some(request_limiter), cancel_token)
582                .with_retry_backoff(retry_backoff),
583        )
584    } else if resumable {
585        let session =
586            DownloadSession::new(None, Some(DownloadRequestLimiter::new(1)), cancel_token)
587                .with_retry_backoff(retry_backoff);
588        download_and_extract(url, format, target_dir, static_files_dir, session)
589    } else {
590        let session =
591            DownloadSession::new(None, None, cancel_token).with_retry_backoff(retry_backoff);
592        let result =
593            streaming_download_and_extract(url, format, target_dir, static_files_dir, &session);
594        if result.is_ok() {
595            session.record_archive_output_complete(0);
596        }
597        result
598    }
599}
600
601/// Downloads and extracts a snapshot archive asynchronously.
602///
603/// Download progress is reported using a local progress bar.
604/// When `resumable` is true, uses two-phase download with `.part` files.
605pub(crate) async fn stream_and_extract(
606    url: &str,
607    target_dir: &Path,
608    static_files_dir: Option<&Path>,
609    resumable: bool,
610    request_limiter: Option<Arc<DownloadRequestLimiter>>,
611    cancel_token: CancellationToken,
612    retry_backoff: Option<Duration>,
613) -> Result<()> {
614    let target_dir = target_dir.to_path_buf();
615    let static_files_dir = static_files_dir.map(Path::to_path_buf);
616    let url = url.to_string();
617    reth_tasks::spawn_blocking(move || {
618        blocking_download_and_extract(
619            &url,
620            &target_dir,
621            static_files_dir.as_deref(),
622            resumable,
623            request_limiter,
624            cancel_token,
625            retry_backoff,
626        )
627    })
628    .await??;
629
630    Ok(())
631}
632
633#[cfg(test)]
634mod tests {
635    use super::*;
636
637    #[test]
638    fn custom_static_root_cannot_be_redirected_by_archive_symlinks() {
639        for custom_path in ["custom", "custom/nested"] {
640            let target = tempfile::tempdir().unwrap();
641            let outside = tempfile::tempdir().unwrap();
642            fs::create_dir_all(outside.path().join("nested")).unwrap();
643            for path in ["headers", "nested/headers"] {
644                fs::write(outside.path().join(path), b"keep").unwrap();
645            }
646            let custom = target.path().join(custom_path);
647            let mut archive = tar::Builder::new(Vec::new());
648            let mut link = tar::Header::new_gnu();
649            link.set_entry_type(tar::EntryType::Symlink);
650            link.set_size(0);
651            link.set_mode(0o777);
652            archive.append_link(&mut link, "custom", outside.path()).unwrap();
653            let mut file = tar::Header::new_gnu();
654            file.set_size(4);
655            file.set_mode(0o644);
656            file.set_cksum();
657            archive.append_data(&mut file, "static_files/headers", b"data".as_slice()).unwrap();
658            let tar = archive.into_inner().unwrap();
659            let err =
660                unpack_archive(Archive::new(tar.as_slice()), target.path(), Some(&custom), None)
661                    .unwrap_err();
662            assert!(format!("{err:#}").contains("Archive links are unsupported"));
663            let outputs = [super::super::manifest::OutputFileChecksum {
664                path: "static_files/headers".into(),
665                size: 4,
666                blake3: String::new(),
667            }];
668            super::super::verify::OutputVerifier::new(target.path(), Some(&custom))
669                .cleanup(&outputs);
670            for path in ["headers", "nested/headers"] {
671                assert_eq!(fs::read(outside.path().join(path)).unwrap(), b"keep");
672            }
673        }
674    }
675
676    #[test]
677    fn link_entries_are_rejected_without_custom_static_dir() {
678        for entry_type in [tar::EntryType::Symlink, tar::EntryType::Link] {
679            for with_progress in [false, true] {
680                let root = tempfile::tempdir().unwrap();
681                let target = root.path().join("target");
682                let outside = root.path().join("outside");
683                fs::create_dir_all(&target).unwrap();
684                fs::create_dir_all(&outside).unwrap();
685                fs::write(outside.join("file"), b"keep").unwrap();
686
687                let mut archive = tar::Builder::new(Vec::new());
688                let mut link = tar::Header::new_gnu();
689                link.set_entry_type(entry_type);
690                link.set_size(0);
691                link.set_mode(0o777);
692                // A symlink to the outside directory would redirect the entry below, a hard link
693                // would alias the outside file.
694                let link_target =
695                    if entry_type.is_symlink() { outside.clone() } else { outside.join("file") };
696                archive.append_link(&mut link, "db", link_target).unwrap();
697                let mut file = tar::Header::new_gnu();
698                file.set_size(4);
699                file.set_mode(0o644);
700                file.set_cksum();
701                archive.append_data(&mut file, "db/file", b"data".as_slice()).unwrap();
702                let tar = archive.into_inner().unwrap();
703
704                let mut progress = ArchiveExtractionProgress::new(None);
705                let progress = with_progress.then_some(&mut progress);
706                let err = unpack_archive(Archive::new(tar.as_slice()), &target, None, progress)
707                    .unwrap_err();
708                assert!(format!("{err:#}").contains("Archive links are unsupported"));
709
710                assert!(
711                    !target.join("db").exists() && target.join("db").symlink_metadata().is_err()
712                );
713                assert_eq!(fs::read(outside.join("file")).unwrap(), b"keep");
714                assert_eq!(fs::read_dir(&outside).unwrap().count(), 1);
715            }
716        }
717    }
718
719    #[test]
720    fn remap_static_files_in_both_compression_formats() {
721        let mut archive = tar::Builder::new(Vec::new());
722        let mut directory = tar::Header::new_gnu();
723        directory.set_entry_type(tar::EntryType::Directory);
724        directory.set_size(0);
725        directory.set_mode(0o755);
726        directory.set_cksum();
727        archive.append_data(&mut directory, "./static_files/", std::io::empty()).unwrap();
728        for path in ["./static_files/nested/headers", "db/data"] {
729            let mut header = tar::Header::new_gnu();
730            header.set_size(4);
731            header.set_mode(0o644);
732            header.set_cksum();
733            archive.append_data(&mut header, path, b"data".as_slice()).unwrap();
734        }
735        let tar = archive.into_inner().unwrap();
736        for format in [CompressionFormat::Lz4, CompressionFormat::Zstd] {
737            let bytes = match format {
738                CompressionFormat::Lz4 => {
739                    let mut encoder = lz4::EncoderBuilder::new().build(Vec::new()).unwrap();
740                    std::io::copy(&mut tar.as_slice(), &mut encoder).unwrap();
741                    let (bytes, result) = encoder.finish();
742                    result.unwrap();
743                    bytes
744                }
745                CompressionFormat::Zstd => zstd::encode_all(tar.as_slice(), 0).unwrap(),
746            };
747            let target = tempfile::tempdir().unwrap();
748            let custom = tempfile::tempdir().unwrap();
749            extract_archive_raw(bytes.as_slice(), format, target.path(), Some(custom.path()), None)
750                .unwrap();
751            assert_eq!(fs::read(custom.path().join("nested/headers")).unwrap(), b"data");
752            assert_eq!(fs::read(target.path().join("db/data")).unwrap(), b"data");
753            assert!(!target.path().join("static_files").exists());
754        }
755    }
756
757    #[test]
758    fn test_compression_format_detection() {
759        assert!(matches!(
760            CompressionFormat::from_url("https://example.com/snapshot.tar.lz4"),
761            Ok(CompressionFormat::Lz4)
762        ));
763        assert!(matches!(
764            CompressionFormat::from_url("https://example.com/snapshot.tar.zst"),
765            Ok(CompressionFormat::Zstd)
766        ));
767        assert!(matches!(
768            CompressionFormat::from_url("file:///path/to/snapshot.tar.lz4"),
769            Ok(CompressionFormat::Lz4)
770        ));
771        assert!(matches!(
772            CompressionFormat::from_url("file:///path/to/snapshot.tar.zst"),
773            Ok(CompressionFormat::Zstd)
774        ));
775        assert!(CompressionFormat::from_url("https://example.com/snapshot.tar.gz").is_err());
776    }
777}
reth_cli_commands/download/extract.rs

reth_cli_commands/download/
extract.rs
