Add Restore snapshot #102

mithril-network/mithril-client/Cargo.toml

@@ -19,6 +19,8 @@ reqwest = { version = "0.11", features = ["json", "stream"] }
 tokio = { version = "1", features = ["full"] }
 async-trait = "0.1.52"
 futures = "0.3"
+tar = "0.4.38"
+flate2 = "1.0.23"
 
 #[dev-dependencies]
 mockall = "0.11.0"

mithril-network/mithril-client/src/aggregator.rs

@@ -1,12 +1,14 @@
 use async_trait::async_trait;
+use flate2::read::GzDecoder;
 use futures::StreamExt;
 use log::debug;
 use reqwest::{self, StatusCode};
 use std::env;
 use std::fs;
-use std::io::Write;
+use std::io::{self, Write};
 use std::path;
 use std::sync::Arc;
+use tar::Archive;
 
 use crate::entities::*;
 
@@ -25,6 +27,9 @@ pub trait AggregatorHandler {
 
     /// Download snapshot
     async fn download_snapshot(&self, digest: String, location: String) -> Result<String, String>;
+
+    /// Unarchive snapshot
+    async fn unarchive_snapshot(&self, digest: String) -> Result<String, String>;
 }
 
 /// AggregatorHTTPClient is a http client for an aggregator
@@ -89,12 +94,7 @@ impl AggregatorHandler for AggregatorHTTPClient {
         match response {
             Ok(response) => match response.status() {
                 StatusCode::OK => {
-                    let local_path = env::current_dir()
-                        .map_err(|e| format!("current dir not available: {}", e))?
-                        .join(path::Path::new(&format!(
-                            "data/{}/{}/snapshot.archive",
-                            self.config.network, digest
-                        )));
+                    let local_path = archive_file_path(digest, self.config.network.clone())?;
                     fs::create_dir_all(&local_path.parent().unwrap())
                         .map_err(|e| format!("can't create snapshot dir: {}", e))?;
                     let mut local_file = fs::File::create(&local_path)
@@ -110,11 +110,13 @@ impl AggregatorHandler for AggregatorHTTPClient {
                             .write_all(&chunk)
                             .map_err(|e| format!("can't write to snapshot file: {}", e))?;
                         bytes_downloaded += chunk.len() as u64;
-                        debug!(
-                            "Downloaded {}% - {} Bytes",
+                        print!(
+                            "Downloaded {}% - {}/{} Bytes\r",
                             100 * bytes_downloaded / bytes_total,
-                            bytes_downloaded
+                            bytes_downloaded,
+                            bytes_total
                         );
+                        io::stdout().flush().ok().expect("Could not flush stdout");
                     }
                     Ok(local_path.into_os_string().into_string().unwrap())
                 }
@@ -124,6 +126,36 @@ impl AggregatorHandler for AggregatorHTTPClient {
             Err(err) => Err(err.to_string()),
         }
     }
+
+    /// Unarchive snapshot
+    async fn unarchive_snapshot(&self, digest: String) -> Result<String, String> {
+        debug!("Restore snapshot {}", digest);
+        println!("Restoring...");
+        let local_path = archive_file_path(digest, self.config.network.clone())?;
+        let snapshot_file_tar_gz = fs::File::open(local_path.clone())
+            .map_err(|e| format!("can't open snapshot file: {}", e))?;
+        let snapshot_file_tar = GzDecoder::new(snapshot_file_tar_gz);
+        let unarchive_dir_path = local_path
+            .clone()
+            .parent()
+            .unwrap()
+            .join(path::Path::new("db"));
+        let mut snapshot_archive = Archive::new(snapshot_file_tar);
+        snapshot_archive
+            .unpack(&unarchive_dir_path)
+            .map_err(|e| format!("can't unpack snapshot archive: {}", e))?;
+        Ok(unarchive_dir_path.into_os_string().into_string().unwrap())
+    }
+}
+
+/// Computes local archive filepath
+fn archive_file_path(digest: String, network: String) -> Result<path::PathBuf, String> {
+    Ok(env::current_dir()
+        .map_err(|e| format!("current dir not available: {}", e))?
+        .join(path::Path::new(&format!(
+            "data/{}/{}/snapshot.archive.tar.gz",
+            network, digest
+        ))))
 }
 
 #[cfg(test)]
@@ -240,7 +272,7 @@ mod tests {
         let url_path = "/download";
         let (server, config) = setup_test();
         let data_expected = "1234567890".repeat(1024).to_string();
-        let _download_mock = server.mock(|when, then| {
+        server.mock(|when, then| {
             when.path(url_path.to_string());
             then.status(200).body(&data_expected);
         });

mithril-network/mithril-client/src/aggregator_fake.rs

@@ -37,4 +37,9 @@ impl AggregatorHandler for AggregatorHandlerFake {
     async fn download_snapshot(&self, digest: String, location: String) -> Result<String, String> {
         unimplemented!("Download snapshot {} at {}", digest, location);
     }
+
+    /// Unarchive snapshot
+    async fn unarchive_snapshot(&self, digest: String) -> Result<String, String> {
+        unimplemented!("Unarchive snapshot {}", digest);
+    }
 }

mithril-network/mithril-client/src/client.rs

@@ -98,9 +98,17 @@ where
     }
 
     /// Restore a snapshot by hash
-    pub async fn restore_snapshot(&self, hash: String) -> Result<(), String> {
-        debug!("Restore snapshot {}", hash);
-        Ok(())
+    pub async fn restore_snapshot(&self, digest: String) -> Result<String, String> {
+        debug!("Restore snapshot {}", digest);
+        match &self.aggregator_handler {
+            Some(aggregator_handler) => {
+                match aggregator_handler.unarchive_snapshot(digest.clone()).await {
+                    Ok(to) => Ok(to),
+                    Err(err) => Err(err),
+                }
+            }
+            None => Err(errors::MISSING_AGGREGATOR_HANDLER.to_string()),
+        }
     }
 }
 

mithril-network/mithril-client/src/main.rs

@@ -128,9 +128,22 @@ async fn main() {
             ),
             Err(err) => pretty_print_error(err),
         },
-        Commands::Restore { digest } => {
-            client.restore_snapshot(digest.to_string()).await.unwrap();
-        }
+        Commands::Restore { digest } => match client.restore_snapshot(digest.to_string()).await {
+            Ok(to) => {
+                println!(
+                    r###"Unarchive success {}
+to {}
+
+Restore a Cardano Node with:
+
+docker run -v cardano-node-ipc:/ipc -v cardano-node-data:/data --mount type=bind,source="{}",target=/data/db/ -e NETWORK=testnet inputoutput/cardano-node
+
+"###,
+                    digest, to, to
+                )
+            }
+            Err(err) => pretty_print_error(err),
+        },
     }
 }