bitrot test(1)

Signed-off-by: junxiang Mu <1948535941@qq.com>

Cargo.lock

@@ -517,6 +517,7 @@ dependencies = [
  "serde_json",
  "sha2",
  "siphasher",
+ "tempfile",
  "thiserror",
  "time",
  "tokio",

Cargo.toml

@@ -53,6 +53,7 @@ s3s = { git = "https://github.com/Nugine/s3s.git", rev = "0caf79822ae1f2e7a7fad3
 s3s-policy = { git = "https://github.com/Nugine/s3s.git", rev = "0caf79822ae1f2e7a7fad3c4a093c94dd2e2c33f" }
 serde = { version = "1.0.210", features = ["derive"] }
 serde_json = "1.0.128"
+tempfile = "3.13.0"
 thiserror = "1.0.64"
 time = { version = "0.3.36", features = [
     "std",

ecstore/Cargo.toml

@@ -43,6 +43,7 @@ base64-simd = "0.8.0"
 sha2 = { version = "0.10.8", features = ["asm"] }
 hex-simd = "0.8.0"
 path-clean = "1.0.1"
+tempfile.workspace = true
 tokio = { workspace = true, features = ["io-util", "sync"] }
 tokio-stream = "0.1.15"
 tonic.workspace = true

ecstore/src/bitrot.rs

@@ -1,7 +1,6 @@
 use std::{any::Any, collections::HashMap, io::Cursor};
 
 use blake2::Blake2b512;
-use hex_simd::{decode_to_vec, encode_to_string};
 use highway::{HighwayHash, HighwayHasher, Key};
 use lazy_static::lazy_static;
 use sha2::{digest::core_api::BlockSizeUser, Digest, Sha256};
@@ -30,6 +29,7 @@ lazy_static! {
 // ];
 const MAGIC_HIGHWAY_HASH256_KEY: &[u64; 4] = &[3, 4, 2, 1];
 
+#[derive(Clone)]
 pub enum Hasher {
     SHA256(Sha256),
     HighwayHash256(HighwayHasher),
@@ -85,8 +85,8 @@ impl Hasher {
             Hasher::HighwayHash256(highway_hasher) => {
                 let key = Key(*MAGIC_HIGHWAY_HASH256_KEY);
                 *highway_hasher = HighwayHasher::new(key);
-            } ,
+            }
-            Hasher::BLAKE2b512(core_wrapper) =>core_wrapper.reset(),
+            Hasher::BLAKE2b512(core_wrapper) => core_wrapper.reset(),
         }
     }
 }
@@ -165,6 +165,14 @@ pub fn new_bitrot_reader(
     Box::new(WholeBitrotReader::new(disk, bucket, file_path, algo, till_offset, sum))
 }
 
+pub fn bitrot_writer_sum(w: &BitrotWriter) -> Vec<u8> {
+    if let Some(w) = w.as_any().downcast_ref::<WholeBitrotWriter>() {
+        return w.hash.clone().finalize();
+    }
+
+    Vec::new()
+}
+
 pub fn bitrot_shard_file_size(size: i64, _shard_size: i64, algo: BitrotAlgorithm) -> i64 {
     if algo != BitrotAlgorithm::HighwayHash256S {
         return size;
@@ -195,21 +203,17 @@ pub struct WholeBitrotWriter {
     disk: DiskStore,
     volume: String,
     file_path: String,
-    shard_size: usize,
+    _shard_size: usize,
-    hash: Hasher,
+    pub hash: Hasher,
 }
 
 impl WholeBitrotWriter {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     pub fn new(disk: DiskStore, volume: &str, file_path: &str, algo: BitrotAlgorithm, shard_size: usize) -> Self {
         WholeBitrotWriter {
             disk,
             volume: volume.to_string(),
             file_path: file_path.to_string(),
-            shard_size,
+            _shard_size: shard_size,
             hash: algo.new(),
         }
     }
@@ -217,6 +221,10 @@ impl WholeBitrotWriter {
 
 #[async_trait::async_trait]
 impl Write for WholeBitrotWriter {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
     async fn write(&mut self, buf: &[u8]) -> Result<()> {
         let mut file = self.disk.append_file(&self.volume, &self.file_path).await?;
         let _ = file.write(buf).await?;
@@ -230,7 +238,7 @@ pub struct WholeBitrotReader {
     disk: DiskStore,
     volume: String,
     file_path: String,
-    verifier: BitrotVerifier,
+    _verifier: BitrotVerifier,
     till_offset: usize,
     buf: Option<Vec<u8>>,
 }
@@ -241,7 +249,7 @@ impl WholeBitrotReader {
             disk,
             volume: volume.to_string(),
             file_path: file_path.to_string(),
-            verifier: BitrotVerifier::new(algo, sum),
+            _verifier: BitrotVerifier::new(algo, sum),
             till_offset,
             buf: None,
         }
@@ -270,39 +278,108 @@ impl ReadAt for WholeBitrotReader {
     }
 }
 
-#[test]
+#[cfg(test)]
-fn bitrot_self_test() -> Result<()> {
+mod test {
-    let mut checksums = HashMap::new();
+    use std::{collections::HashMap, fs};
-    checksums.insert(BitrotAlgorithm::SHA256, "a7677ff19e0182e4d52e3a3db727804abc82a5818749336369552e54b838b004");
-    checksums.insert(BitrotAlgorithm::BLAKE2b512, "e519b7d84b1c3c917985f544773a35cf265dcab10948be3550320d156bab612124a5ae2ae5a8c73c0eea360f68b0e28136f26e858756dbfe7375a7389f26c669");
-    checksums.insert(BitrotAlgorithm::HighwayHash256, "c81c2386a1f565e805513d630d4e50ff26d11269b21c221cf50fc6c29d6ff75b");
-    checksums.insert(BitrotAlgorithm::HighwayHash256S, "c81c2386a1f565e805513d630d4e50ff26d11269b21c221cf50fc6c29d6ff75b");
 
-    let iter = [BitrotAlgorithm::SHA256, BitrotAlgorithm::BLAKE2b512, BitrotAlgorithm:: HighwayHash256];
+    use hex_simd::decode_to_vec;
+    use tempfile::TempDir;
 
-    for algo in iter.iter() {
+    use crate::{
-        if !algo.available() || *algo != BitrotAlgorithm::HighwayHash256 {
+        bitrot::{new_bitrot_writer, BITROT_ALGORITHMS},
-            continue;
+        disk::{endpoint::Endpoint, error::DiskError, new_disk, DiskOption},
-        }
+        error::{Error, Result},
-        let checksum = decode_to_vec(checksums.get(algo).unwrap()).unwrap();
+        store_api::BitrotAlgorithm,
+    };
 
-        let mut h = algo.new();
+    use super::{bitrot_writer_sum, new_bitrot_reader};
-        let mut msg = Vec::with_capacity(h.size() * h.block_size());
-        let mut sum = Vec::with_capacity(h.size());
 
-        for i in (0..h.size()*h.block_size()).step_by(h.size()) {
+    #[test]
-            h.update(&msg);
+    fn bitrot_self_test() -> Result<()> {
-            sum = h.finalize();
+        let mut checksums = HashMap::new();
-            msg.extend(sum.clone());
+        checksums.insert(
-            h = algo.new();
+            BitrotAlgorithm::SHA256,
+            "a7677ff19e0182e4d52e3a3db727804abc82a5818749336369552e54b838b004",
+        );
+        checksums.insert(BitrotAlgorithm::BLAKE2b512, "e519b7d84b1c3c917985f544773a35cf265dcab10948be3550320d156bab612124a5ae2ae5a8c73c0eea360f68b0e28136f26e858756dbfe7375a7389f26c669");
+        checksums.insert(
+            BitrotAlgorithm::HighwayHash256,
+            "c81c2386a1f565e805513d630d4e50ff26d11269b21c221cf50fc6c29d6ff75b",
+        );
+        checksums.insert(
+            BitrotAlgorithm::HighwayHash256S,
+            "c81c2386a1f565e805513d630d4e50ff26d11269b21c221cf50fc6c29d6ff75b",
+        );
+
+        let iter = [
+            BitrotAlgorithm::SHA256,
+            BitrotAlgorithm::BLAKE2b512,
+            BitrotAlgorithm::HighwayHash256,
+        ];
+
+        for algo in iter.iter() {
+            if !algo.available() || *algo != BitrotAlgorithm::HighwayHash256 {
+                continue;
+            }
+            let checksum = decode_to_vec(checksums.get(algo).unwrap()).unwrap();
+
+            let mut h = algo.new();
+            let mut msg = Vec::with_capacity(h.size() * h.block_size());
+            let mut sum = Vec::with_capacity(h.size());
+
+            for _ in (0..h.size() * h.block_size()).step_by(h.size()) {
+                h.update(&msg);
+                sum = h.finalize();
+                msg.extend(sum.clone());
+                h = algo.new();
+            }
+
+            if checksum != sum {
+                println!("failed: {:?}, expect: {:?}, actual: {:?}", algo, checksum, sum);
+                return Err(Error::new(DiskError::FileCorrupt));
+            }
+            println!("success: {:?}", algo);
         }
 
-        if checksum != sum {
+        Ok(())
-            println!("failed: {:?}, expect: {:?}, actual: {:?}", algo, checksum, sum);
-            return Err(Error::new(DiskError::FileCorrupt));
-        }
-        println!("success: {:?}", algo);
     }
 
-    Ok(())
+    #[tokio::test]
+    async fn test_all_bitrot_algorithms() -> Result<()> {
+        for algo in BITROT_ALGORITHMS.keys() {
+            if *algo == BitrotAlgorithm::HighwayHash256S {
+                continue;
+            }
+            test_bitrot_reader_writer_algo(algo.clone()).await?;
+        }
+
+        Ok(())
+    }
+
+    async fn test_bitrot_reader_writer_algo(algo: BitrotAlgorithm) -> Result<()> {
+        let temp_dir = TempDir::new().unwrap().path().to_string_lossy().to_string();
+        fs::create_dir_all(&temp_dir)?;
+        let volume = "testvol";
+        let file_path = "testfile";
+
+        let ep = Endpoint::try_from(temp_dir.as_str())?;
+        let opt = DiskOption::default();
+        let disk = new_disk(&ep, &opt).await?;
+        let _ = disk.make_volume(volume).await?;
+        let mut writer = new_bitrot_writer(disk.clone(), "", volume, file_path, 35, algo.clone(), 10);
+
+        let _ = writer.write(b"aaaaaaaaaa").await?;
+        let _ = writer.write(b"aaaaaaaaaa").await?;
+        let _ = writer.write(b"aaaaaaaaaa").await?;
+        let _ = writer.write(b"aaaaa").await?;
+
+        let mut reader = new_bitrot_reader(disk, b"", volume, file_path, 35, algo, &bitrot_writer_sum(&writer), 10);
+        let read_len = 10;
+        (_, _) = reader.read_at(0, read_len).await?;
+        (_, _) = reader.read_at(0, read_len).await?;
+        (_, _) = reader.read_at(0, read_len).await?;
+        (_, _) = reader.read_at(0, read_len / 2).await?;
+
+        Ok(())
+    }
 }

ecstore/src/disk/mod.rs

@@ -26,7 +26,7 @@ use protos::proto_gen::node_service::{
     node_service_client::NodeServiceClient, ReadAtRequest, ReadAtResponse, WriteRequest, WriteResponse,
 };
 use serde::{Deserialize, Serialize};
-use std::{cmp::Ordering, collections::HashMap, fmt::Debug, io::SeekFrom, path::PathBuf, sync::Arc, usize};
+use std::{any::Any, cmp::Ordering, collections::HashMap, fmt::Debug, io::SeekFrom, path::PathBuf, sync::Arc, usize};
 use time::OffsetDateTime;
 use tokio::{
     fs::File,
@@ -646,6 +646,10 @@ pub enum FileWriter {
 
 #[async_trait::async_trait]
 impl Write for FileWriter {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
     async fn write(&mut self, buf: &[u8]) -> Result<()> {
         match self {
             Self::Local(local_file_writer) => local_file_writer.write(buf).await,
@@ -666,6 +670,10 @@ impl LocalFileWriter {
 
 #[async_trait::async_trait]
 impl Write for LocalFileWriter {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
     async fn write(&mut self, buf: &[u8]) -> Result<()> {
         let _ = self.inner.write(buf).await?;
         self.inner.flush().await?;
@@ -714,6 +722,10 @@ impl RemoteFileWriter {
 
 #[async_trait::async_trait]
 impl Write for RemoteFileWriter {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
     async fn write(&mut self, buf: &[u8]) -> Result<()> {
         let request = WriteRequest {
             disk: self.root.to_string_lossy().to_string(),

ecstore/src/erasure.rs

@@ -4,6 +4,7 @@ use bytes::Bytes;
 use futures::future::join_all;
 use futures::{Stream, StreamExt};
 use reed_solomon_erasure::galois_8::ReedSolomon;
+use std::any::Any;
 use std::fmt::Debug;
 use tokio::io::AsyncWriteExt;
 use tokio::io::DuplexStream;
@@ -325,6 +326,7 @@ impl Erasure {
 
 #[async_trait::async_trait]
 pub trait Write {
+    fn as_any(&self) -> &dyn Any;
     async fn write(&mut self, buf: &[u8]) -> Result<()>;
 }
 

ecstore/src/utils/os/linux.rs

@@ -1,79 +1,107 @@
-use nix::sys::{
+use nix::sys::stat::{self, stat};
-    stat::{major, minor, stat},
+use nix::sys::statfs::{self, statfs, FsType};
-    statfs::{statfs, FsType},
+use std::io::{Error, ErrorKind};
-};
+use std::path::Path;
 
-use crate::{
+use crate::{disk::Info, error::Result};
-    disk::Info,
-    error::{Error, Result},
-};
 
-use lazy_static::lazy_static;
+/// returns total and free bytes available in a directory, e.g. `/`.
-use std::collections::HashMap;
+pub fn get_info(p: impl AsRef<Path>, _first_time: bool) -> std::io::Result<Info> {
+    let stat_fs = statfs(p.as_ref())?;
 
-lazy_static! {
+    let bsize = stat_fs.block_size() as u64;
-    static ref FS_TYPE_TO_STRING_MAP: HashMap<&'static str, &'static str> = {
+    let bfree = stat_fs.blocks_free() as u64;
-        let mut m = HashMap::new();
+    let bavail = stat_fs.blocks_available() as u64;
-        m.insert("1021994", "TMPFS");
+    let blocks = stat_fs.blocks() as u64;
-        m.insert("137d", "EXT");
+
-        m.insert("4244", "HFS");
+    let reserved = match bfree.checked_sub(bavail) {
-        m.insert("4d44", "MSDOS");
+        Some(reserved) => reserved,
-        m.insert("52654973", "REISERFS");
+        None => {
-        m.insert("5346544e", "NTFS");
+            return Err(Error::new(
-        m.insert("58465342", "XFS");
+                ErrorKind::Other,
-        m.insert("61756673", "AUFS");
+                format!(
-        m.insert("6969", "NFS");
+                    "detected f_bavail space ({}) > f_bfree space ({}), fs corruption at ({}). please run 'fsck'",
-        m.insert("ef51", "EXT2OLD");
+                    bavail,
-        m.insert("ef53", "EXT4");
+                    bfree,
-        m.insert("f15f", "ecryptfs");
+                    p.as_ref().display()
-        m.insert("794c7630", "overlayfs");
+                ),
-        m.insert("2fc12fc1", "zfs");
+            ))
-        m.insert("ff534d42", "cifs");
+        }
-        m.insert("53464846", "wslfs");
-        m
     };
-}
 
-fn get_fs_type(ftype: FsType) -> String {
+    let total = match blocks.checked_sub(reserved) {
-    let binding = format!("{:?}", ftype);
+        Some(total) => total * bsize,
-    let fs_type_hex = binding.as_str();
+        None => {
-    match FS_TYPE_TO_STRING_MAP.get(fs_type_hex) {
+            return Err(Error::new(
-        Some(fs_type_string) => fs_type_string.to_string(),
+                ErrorKind::Other,
-        None => "UNKNOWN".to_string(),
+                format!(
-    }
+                    "detected reserved space ({}) > blocks space ({}), fs corruption at ({}). please run 'fsck'",
-}
+                    reserved,
+                    blocks,
+                    p.as_ref().display()
+                ),
+            ))
+        }
+    };
+
+    let free = bavail * bsize;
+    let used = match total.checked_sub(free) {
+        Some(used) => used,
+        None => {
+            return Err(Error::new(
+                ErrorKind::Other,
+                format!(
+                    "detected free space ({}) > total drive space ({}), fs corruption at ({}). please run 'fsck'",
+                    free,
+                    total,
+                    p.as_ref().display()
+                ),
+            ))
+        }
+    };
+
+    let st = stat(p.as_ref())?;
+
+    Ok(Info {
+        total,
+        free,
+        used,
+        files: stat_fs.files(),
+        ffree: stat_fs.files_free(),
+        fstype: get_fs_type(stat_fs.filesystem_type()).to_string(),
+
+        major: stat::major(st.st_dev),
+        minor: stat::minor(st.st_dev),
 
-pub fn get_info(path: &str, first_time: bool) -> Result<Info> {
-    let statfs = statfs(path)?;
-    let reserved_blocks = statfs.blocks_free() - statfs.blocks_available();
-    let mut info = Info {
-        total: statfs.block_size() as u64 * (statfs.blocks() - reserved_blocks),
-        free: statfs.blocks() as u64 * statfs.blocks_available(),
-        files: statfs.files(),
-        ffree: statfs.files_free(),
-        fstype: get_fs_type(statfs.filesystem_type()),
         ..Default::default()
-    };
+    })
+}
 
-    let stat = stat(path)?;
+/// returns the filesystem type of the underlying mounted filesystem
-    let dev_id = stat.st_dev as u64;
+///
-    info.major = major(dev_id);
+/// TODO The following mapping could not find the corresponding constant in `nix`:
-    info.minor = minor(dev_id);
+///
+/// "137d" => "EXT",
+/// "4244" => "HFS",
+/// "5346544e" => "NTFS",
+/// "61756673" => "AUFS",
+/// "ef51" => "EXT2OLD",
+/// "2fc12fc1" => "zfs",
+/// "ff534d42" => "cifs",
+/// "53464846" => "wslfs",
+fn get_fs_type(fs_type: FsType) -> &'static str {
+    match fs_type {
+        statfs::TMPFS_MAGIC => "TMPFS",
+        statfs::MSDOS_SUPER_MAGIC => "MSDOS",
+        statfs::XFS_SUPER_MAGIC => "XFS",
+        statfs::NFS_SUPER_MAGIC => "NFS",
+        statfs::EXT4_SUPER_MAGIC => "EXT4",
+        statfs::ECRYPTFS_SUPER_MAGIC => "ecryptfs",
+        statfs::OVERLAYFS_SUPER_MAGIC => "overlayfs",
+        statfs::REISERFS_SUPER_MAGIC => "REISERFS",
 
-    if info.free > info.total {
+        _ => "UNKNOWN",
-        return Err(Error::from_string(format!(
-            "detected free space {} > total drive space {}, fs corruption at {}. please run 'fsck'",
-            info.free, info.total, path
-        )));
     }
-
-    info.used = info.total - info.free;
-
-    if first_time {
-        // todo
-    }
-
-    Ok(info)
 }
 
 pub fn same_disk(disk1: &str, disk2: &str) -> Result<bool> {