Compress snapshot archive within the validator to reduce system dependencies, and default to zstd compression (bp #12085) (#12086)

* Compress snapshot archive within the validator to reduce system dependencies

(cherry picked from commit d3750b47d2)

* Default snapshot compression to zstd instead of bzip2 for quicker snapshot generation

(cherry picked from commit 9ade73841f)

Co-authored-by: Michael Vines <mvines@gmail.com>

ledger/src/snapshot_utils.rs

@@ -21,9 +21,9 @@ use std::{
     cmp::Ordering,
     fmt,
     fs::{self, File},
-    io::{BufReader, BufWriter, Error as IOError, ErrorKind, Read, Seek, SeekFrom, Write},
+    io::{self, BufReader, BufWriter, Error as IOError, ErrorKind, Read, Seek, SeekFrom, Write},
     path::{Path, PathBuf},
-    process::ExitStatus,
+    process::{self, ExitStatus},
     str::FromStr,
     sync::Arc,
 };
@@ -194,12 +194,12 @@ pub fn package_snapshot<P: AsRef<Path>, Q: AsRef<Path>>(
     Ok(package)
 }
 
-fn get_compression_ext(compression: &CompressionType) -> (&'static str, &'static str) {
+fn get_compression_ext(compression: &CompressionType) -> &'static str {
     match compression {
-        CompressionType::Bzip2 => ("bzip2", ".tar.bz2"),
+        CompressionType::Bzip2 => ".tar.bz2",
-        CompressionType::Gzip => ("gzip", ".tar.gz"),
+        CompressionType::Gzip => ".tar.gz",
-        CompressionType::Zstd => ("zstd", ".tar.zst"),
+        CompressionType::Zstd => ".tar.zst",
-        CompressionType::NoCompression => ("", ".tar"),
+        CompressionType::NoCompression => ".tar",
     }
 }
 
@@ -258,40 +258,73 @@ pub fn archive_snapshot_package(snapshot_package: &AccountsPackage) -> Result<()
 
     // Write version file
     {
-        let mut f = std::fs::File::create(staging_version_file)?;
+        let mut f = fs::File::create(staging_version_file)?;
         f.write_all(snapshot_package.snapshot_version.as_str().as_bytes())?;
     }
 
-    let (compression_option, file_ext) = get_compression_ext(&snapshot_package.compression);
+    let file_ext = get_compression_ext(&snapshot_package.compression);
-    let archive_options = "cfhS";
 
     // Tar the staging directory into the archive at `archive_path`
-    let archive_file = format!("new_state{}", file_ext);
+    //
-    let archive_path = tar_dir.join(archive_file);
+    // system `tar` program is used for -S (sparse file support)
-    let args = vec![
+    let archive_path = tar_dir.join(format!("new_state{}", file_ext));
-        archive_options,
+
-        archive_path.to_str().unwrap(),
+    let mut tar = process::Command::new("tar")
-        "--use-compress-program",
+        .args(&[
-        compression_option,
+            "chS",
             "-C",
             staging_dir.path().to_str().unwrap(),
             TAR_ACCOUNTS_DIR,
             TAR_SNAPSHOTS_DIR,
             TAR_VERSION_FILE,
-    ];
+        ])
+        .stdin(process::Stdio::null())
+        .stdout(process::Stdio::piped())
+        .stderr(process::Stdio::inherit())
+        .spawn()?;
 
-    let output = std::process::Command::new("tar").args(&args).output()?;
+    match &mut tar.stdout {
-    if !output.status.success() {
+        None => {
-        warn!("tar command failed with exit code: {}", output.status);
+            return Err(SnapshotError::IO(IOError::new(
-        use std::str::from_utf8;
+                ErrorKind::Other,
-        info!("tar stdout: {}", from_utf8(&output.stdout).unwrap_or("?"));
+                "tar stdout unavailable".to_string(),
-        info!("tar stderr: {}", from_utf8(&output.stderr).unwrap_or("?"));
+            )));
+        }
+        Some(tar_output) => {
+            let mut archive_file = fs::File::create(&archive_path)?;
 
-        return Err(SnapshotError::ArchiveGenerationFailure(output.status));
+            match snapshot_package.compression {
+                CompressionType::Bzip2 => {
+                    let mut encoder =
+                        bzip2::write::BzEncoder::new(archive_file, bzip2::Compression::Best);
+                    io::copy(tar_output, &mut encoder)?;
+                    let _ = encoder.finish()?;
+                }
+                CompressionType::Gzip => {
+                    let mut encoder =
+                        flate2::write::GzEncoder::new(archive_file, flate2::Compression::default());
+                    io::copy(tar_output, &mut encoder)?;
+                    let _ = encoder.finish()?;
+                }
+                CompressionType::NoCompression => {
+                    io::copy(tar_output, &mut archive_file)?;
+                }
+                CompressionType::Zstd => {
+                    let mut encoder = zstd::stream::Encoder::new(archive_file, 0)?;
+                    io::copy(tar_output, &mut encoder)?;
+                    let _ = encoder.finish()?;
+                }
+            };
+        }
     }
 
-    // Once everything is successful, overwrite the previous tarball so that other validators
+    let tar_exit_status = tar.wait()?;
-    // can fetch this newly packaged snapshot
+    if !tar_exit_status.success() {
+        warn!("tar command failed with exit code: {}", tar_exit_status);
+        return Err(SnapshotError::ArchiveGenerationFailure(tar_exit_status));
+    }
+
+    // Atomically move the archive into position for other validators to find
     let metadata = fs::metadata(&archive_path)?;
     fs::rename(&archive_path, &snapshot_package.tar_output_file)?;
 
@@ -588,7 +621,7 @@ pub fn get_snapshot_archive_path<P: AsRef<Path>>(
         "snapshot-{}-{}{}",
         snapshot_hash.0,
         snapshot_hash.1,
-        get_compression_ext(compression).1,
+        get_compression_ext(compression),
     ))
 }
 

validator/src/main.rs

@@ -1053,7 +1053,7 @@ pub fn main() {
         exit(1);
     });
 
-    let mut snapshot_compression = CompressionType::Bzip2;
+    let mut snapshot_compression = CompressionType::Zstd;
     if let Ok(compression_str) = value_t!(matches, "snapshot_compression", String) {
         match compression_str.as_str() {
             "bz2" => snapshot_compression = CompressionType::Bzip2,