feat(pd): add --archive-url flag to pd testnet join

Support ingesting a remote URL for downloading and extracting a .tar.gz
file, that unpacks into a rocksdb dir for pd. Not yet supported is any
notion of copying a genesis file out of the archive. Should we default
to that behavior? Make it optional?

crates/bin/pd/Cargo.toml

@@ -92,7 +92,7 @@ rand                             = { workspace = true }
 rand_chacha                      = { workspace = true }
 rand_core                        = { workspace = true, features = ["getrandom"] }
 regex                            = { workspace = true }
-reqwest                          = { version = "0.11", features = ["json"] }
+reqwest                          = { version = "0.11", features = ["json", "stream"] }
 rocksdb                          = { workspace = true }
 serde                            = { workspace = true, features = ["derive"] }
 serde_json                       = { workspace = true }

crates/bin/pd/src/cli.rs

@@ -201,6 +201,13 @@ pub enum TestnetCommand {
             default_value = "https://rpc.testnet.penumbra.zone"
         )]
         node: Url,
+
+        /// Optional URL of archived node state in .tar.gz format. The archive will be
+        /// downloaded and extracted locally, allowing the node to join a network at a block height
+        /// higher than 0.
+        #[clap(long)]
+        archive_url: Option<Url>,
+
         /// Human-readable name to identify node on network
         // Default: 'node-#'
         #[clap(long, env = "PENUMBRA_PD_TM_MONIKER")]

crates/bin/pd/src/main.rs

@@ -1,8 +1,13 @@
 #![allow(clippy::clone_on_copy)]
 #![deny(clippy::unwrap_used)]
 #![recursion_limit = "512"]
+use flate2::read::GzDecoder;
+use flate2::write::GzEncoder;
+use flate2::Compression;
 use std::error::Error;
+use std::fs::File;
 use std::io::IsTerminal as _;
+use std::io::Write;
 
 use console_subscriber::ConsoleLayer;
 use metrics_tracing_context::{MetricsLayer, TracingContextLayer};
@@ -25,6 +30,7 @@ use rand::Rng;
 use rand_core::OsRng;
 use tendermint_config::net::Address as TendermintAddress;
 use tokio::runtime;
+use tokio_stream::StreamExt;
 use tower_http::cors::CorsLayer;
 use tracing_subscriber::{prelude::*, EnvFilter};
 use url::Url;
@@ -255,6 +261,7 @@ async fn main() -> anyhow::Result<()> {
             tn_cmd:
                 TestnetCommand::Join {
                     node,
+                    archive_url,
                     moniker,
                     external_address,
                     tendermint_rpc_bind,
@@ -290,14 +297,63 @@ async fn main() -> anyhow::Result<()> {
             // Join the target testnet, looking up network info and writing
             // local configs for pd and tendermint.
             testnet_join(
-                output_dir,
+                output_dir.clone(),
                 node,
                 &node_name,
                 external_address,
                 tendermint_rpc_bind,
                 tendermint_p2p_bind,
             )
             .await?;
+
+            // Download and extract archive URL, if set.
+            if let Some(archive_url) = archive_url {
+                tracing::info!(%archive_url, "downloading compressed node state");
+
+                // Download. Adapted from https://rust-lang-nursery.github.io/rust-cookbook/web/clients/download.html
+                // Here we inspect HEAD so we can infer filename.
+                let response = reqwest::get(archive_url).await?;
+
+                let fname = response
+                    .url()
+                    .path_segments()
+                    .and_then(|segments| segments.last())
+                    .and_then(|name| if name.is_empty() { None } else { Some(name) })
+                    .unwrap_or("pd-node-state-archive.tar.gz");
+
+                let archive_filepath = output_dir.join(fname);
+                let mut download_opts = std::fs::OpenOptions::new();
+                download_opts
+                    .create_new(true)
+                    .read(true)
+                    .write(true)
+                    .truncate(true);
+                let mut archive_file = download_opts.open(&archive_filepath)?;
+
+                // Download via stream, in case file is too large to shove into RAM.
+                let mut stream = response.bytes_stream();
+                while let Some(chunk_result) = stream.next().await {
+                    let chunk = chunk_result?;
+                    archive_file.write_all(&chunk)?;
+                }
+                archive_file.flush()?;
+                tracing::info!("download complete: {}", archive_filepath.display());
+
+                // Extract
+                // Open downloaded file for writing
+                let mut unpack_opts = std::fs::OpenOptions::new();
+                unpack_opts.read(true);
+                let f = unpack_opts.open(&archive_filepath)?;
+                let tar = GzDecoder::new(f);
+                let mut archive = tar::Archive::new(tar);
+                // This dir-path building is duplicated in the config gen code.
+                let pd_home = output_dir.join("node0").join("pd");
+                archive
+                    .unpack(&pd_home)
+                    .context("failed to extract tar.gz archive")?;
+                tracing::info!("archived node state unpacked to {}", pd_home.display());
+                // TODO: use "migrate" tarball, clobber genesis file from migration.
+            }
         }
 
         RootCommand::Testnet {
@@ -426,10 +482,6 @@ async fn main() -> anyhow::Result<()> {
                     );
                     anyhow::bail!("refusing to overwrite existing archive");
                 }
-                use flate2::write::GzEncoder;
-                use flate2::Compression;
-                use std::fs::File;
-
                 tracing::info!(
                     "creating archive {} -> {}",
                     dst_rocksdb_dir.display(),