Add stream digest

Cargo.toml

@@ -24,13 +24,12 @@ num_enum = "0.5.6"
 rand = "0.8"
 rand_chacha = "0.3"
 roaring = "0.9"
-serde = "1.0.136"
-serde_derive = "1.0.136"
+serde = { version = "1", features = ["derive"] }
+serde_yaml = "0.9"
 size-display = "0.1.4"
 thinp = { git = "https://github.com/jthornber/thin-provisioning-tools.git", branch = "main" }
 # thinp = { path = "../thinp-for-dm-archive/" }
 threadpool = "1.0"
-toml = "0.5.8"
 udev = "0.7"
 walkdir = "2"
 zstd = "0.11"

src/config.rs

@@ -1,6 +1,6 @@
 use anyhow::{Context, Result};
 use chrono::prelude::*;
-use serde_derive::{Deserialize, Serialize};
+use serde::{Deserialize, Serialize};
 use std::fs;
 use std::io::Write;
 use std::path::{Path, PathBuf};
@@ -18,35 +18,36 @@ pub struct Config {
 pub fn read_config<P: AsRef<Path>>(root: P) -> Result<Config> {
     let mut p = PathBuf::new();
     p.push(root);
-    p.push("dm-archive.toml");
+    p.push("dm-archive.yaml");
     let input = fs::read_to_string(p).context("couldn't read config file")?;
-    let config: Config = toml::from_str(&input).context("couldn't parse config file")?;
+    let config: Config = serde_yaml::from_str(&input).context("couldn't parse config file")?;
     Ok(config)
 }
 
 //-----------------------------------------
 
-#[derive(Debug, Deserialize, Serialize)]
+#[derive(Debug, Deserialize, Serialize, PartialEq)]
 pub struct StreamConfig {
     pub name: Option<String>,
     pub source_path: String,
-    pub pack_time: toml::value::Datetime,
+    pub pack_time: String,
     pub size: u64,
     pub mapped_size: u64,
     pub packed_size: u64,
     pub thin_id: Option<u32>,
+    pub source_sig: Option<String>,
 }
 
 fn stream_cfg_path(stream_id: &str) -> PathBuf {
-    ["streams", stream_id, "config.toml"].iter().collect()
+    ["streams", stream_id, "config.yaml"].iter().collect()
 }
 
 pub fn read_stream_config(stream_id: &str) -> Result<StreamConfig> {
     let p = stream_cfg_path(stream_id);
     let input =
         fs::read_to_string(&p).with_context(|| format!("couldn't read stream config '{:?}", &p))?;
     let config: StreamConfig =
-        toml::from_str(&input).context("couldn't parse stream config file")?;
+        serde_yaml::from_str(&input).context("couldn't parse stream config file")?;
     Ok(config)
 }
 
@@ -57,19 +58,42 @@ pub fn write_stream_config(stream_id: &str, cfg: &StreamConfig) -> Result<()> {
         .write(true)
         .create(true)
         .open(p)?;
-    let toml = toml::to_string(cfg).unwrap();
-    output.write_all(toml.as_bytes())?;
+    let yaml = serde_yaml::to_string(cfg).unwrap();
+    output.write_all(yaml.as_bytes())?;
     Ok(())
 }
 
-pub fn now() -> toml::value::Datetime {
+pub fn now() -> String {
     let dt = Utc::now();
-    let str = dt.to_rfc3339();
-    str.parse::<toml::value::Datetime>().unwrap()
+    dt.to_rfc3339()
 }
 
-pub fn to_date_time(t: &toml::value::Datetime) -> chrono::DateTime<FixedOffset> {
-    DateTime::parse_from_rfc3339(&t.to_string()).unwrap()
+pub fn to_date_time(t: &str) -> chrono::DateTime<FixedOffset> {
+    DateTime::parse_from_rfc3339(t).unwrap()
 }
 
 //-----------------------------------------
+
+#[cfg(test)]
+mod config_tests {
+
+    use super::*;
+
+    #[test]
+    fn test_simple() {
+        let config = StreamConfig {
+            name: Some(String::from("test_file")),
+            source_path: String::from("/home/some_user/test_file"),
+            pack_time: String::from("2023-11-14T22:06:02.101221624+00:00"),
+            size: u64::MAX,
+            mapped_size: u64::MAX,
+            packed_size: u64::MAX,
+            thin_id: None,
+            source_sig: None,
+        };
+
+        let ser = serde_yaml::to_string(&config).unwrap();
+        let des_config: StreamConfig = serde_yaml::from_str(&ser).unwrap();
+        assert!(config == des_config);
+    }
+}

src/create.rs

@@ -31,7 +31,7 @@ fn write_config(
 ) -> Result<()> {
     let mut p = PathBuf::new();
     p.push(root);
-    p.push("dm-archive.toml");
+    p.push("dm-archive.yaml");
 
     let mut output = OpenOptions::new()
         .read(false)
@@ -47,7 +47,7 @@ fn write_config(
         data_cache_size_meg,
     };
 
-    write!(output, "{}", &toml::to_string(&config).unwrap())?;
+    write!(output, "{}", &serde_yaml::to_string(&config).unwrap())?;
     Ok(())
 }
 

src/main.rs

@@ -1,5 +1,5 @@
 use anyhow::Result;
-use clap::{command, Arg, ArgMatches, Command};
+use clap::{command, Arg, ArgMatches, Command, SubCommand};
 use std::env;
 use std::process::exit;
 use std::sync::Arc;
@@ -64,6 +64,17 @@ fn main_() -> Result<()> {
         .value_name("JSON")
         .takes_value(false);
 
+    let validate_operations = SubCommand::with_name("validate")
+        .about("Validate operations")
+        .arg_required_else_help(true)
+        .subcommand(
+            SubCommand::with_name("stream")
+                .help("Validates an individual stream against stored b2sum digest")
+                .about("Validates an individual stream against stored b2sum digest")
+                .arg(stream_arg.clone())
+                .arg(archive_arg.clone()),
+        );
+
     let matches = command!()
         .arg(json)
         .propagate_version(true)
@@ -184,6 +195,7 @@ fn main_() -> Result<()> {
                 .about("lists the streams in the archive")
                 .arg(archive_arg.clone()),
         )
+        .subcommand(validate_operations)
         .get_matches();
 
     let report = mk_report(&matches);
@@ -211,6 +223,12 @@ fn main_() -> Result<()> {
         Some(("dump-stream", sub_matches)) => {
             dump_stream::run(sub_matches, output)?;
         }
+        Some(("validate", sub_matches)) => match sub_matches.subcommand() {
+            Some(("stream", args)) => {
+                unpack::run_validate_stream(args, report)?;
+            }
+            _ => unreachable!("Exhausted list of validate sub commands"),
+        },
         _ => unreachable!("Exhausted list of subcommands and subcommand_required prevents 'None'"),
     }
 

src/pack.rs

@@ -1,4 +1,5 @@
 use anyhow::{anyhow, Context, Result};
+use blake2::{Blake2b512, Digest};
 use byteorder::{LittleEndian, ReadBytesExt};
 use chrono::prelude::*;
 use clap::ArgMatches;
@@ -389,6 +390,17 @@ struct Packer {
     hash_cache_size_meg: usize,
 }
 
+fn unmapped_digest_add(digest: &mut Blake2b512, len: u64) {
+    let buf = [0; 4096];
+
+    let mut remaining = len;
+    while remaining > 0 {
+        let hash_len = std::cmp::min(buf.len() as u64, remaining);
+        digest.update(&buf[0..hash_len as usize]);
+        remaining -= hash_len;
+    }
+}
+
 impl Packer {
     #[allow(clippy::too_many_arguments)]
     fn new(
@@ -427,6 +439,8 @@ impl Packer {
             .context("couldn't open data slab file")?;
         let data_size = data_file.get_file_size();
 
+        let mut input_digest = Blake2b512::new();
+
         let hashes_size = {
             let hashes_file = hashes_file.lock().unwrap();
             hashes_file.get_file_size()
@@ -465,6 +479,7 @@ impl Packer {
             match chunk? {
                 Chunk::Mapped(buffer) => {
                     let len = buffer.len();
+                    input_digest.update(&buffer);
                     splitter.next_data(buffer, &mut handler)?;
                     total_read += len as u64;
                     self.output
@@ -473,6 +488,7 @@ impl Packer {
                 }
                 Chunk::Unmapped(len) => {
                     assert!(len > 0);
+                    unmapped_digest_add(&mut input_digest, len);
                     splitter.next_break(&mut handler)?;
                     handler.handle_gap(len)?;
                 }
@@ -484,6 +500,7 @@ impl Packer {
         }
 
         splitter.complete(&mut handler)?;
+        let input_hash_sig = format!("{:02x}", input_digest.finalize());
         self.output.report.progress(100);
         let end_time: DateTime<Utc> = Utc::now();
         let elapsed = end_time - start_time;
@@ -555,6 +572,7 @@ impl Packer {
             mapped_size: self.mapped_size,
             packed_size: data_written + hashes_written + stream_written,
             thin_id: self.thin_id,
+            source_sig: Some(input_hash_sig),
         };
         config::write_stream_config(&stream_id, &cfg)?;
 

src/paths.rs

@@ -19,7 +19,7 @@ pub fn stream_path(stream: &str) -> PathBuf {
 }
 
 pub fn stream_config(stream: &str) -> PathBuf {
-    ["streams", stream, "config.toml"].iter().collect()
+    ["streams", stream, "config.yaml"].iter().collect()
 }
 
 //------------------------------