bitrot(1)

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

Cargo.lock

@@ -276,6 +276,15 @@ version = "2.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b048fb63fd8b5923fc5aa7b340d8e156aec7ec02f0c78fa8a6ddc2613f6f71de"
 
+[[package]]
+name = "blake2"
+version = "0.10.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "46502ad458c9a52b69d4d4d32775c788b7a1b85e8bc9d482d92250fc0e3f8efe"
+dependencies = [
+ "digest",
+]
+
 [[package]]
 name = "block-buffer"
 version = "0.10.4"
@@ -479,6 +488,7 @@ dependencies = [
  "async-trait",
  "backon",
  "base64-simd",
+ "blake2",
  "byteorder",
  "bytes",
  "common",
@@ -486,6 +496,7 @@ dependencies = [
  "futures",
  "glob",
  "hex-simd",
+ "highway",
  "http",
  "lazy_static",
  "lock",
@@ -796,6 +807,12 @@ dependencies = [
  "vsimd",
 ]
 
+[[package]]
+name = "highway"
+version = "1.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c706f1711006204c2ba8fb1a7bd55f689bbf7feca9ff40325206b5e140cff6df"
+
 [[package]]
 name = "hmac"
 version = "0.12.1"

ecstore/Cargo.toml

@@ -11,6 +11,7 @@ rust-version.workspace = true
 [dependencies]
 async-trait.workspace = true
 backon.workspace = true
+blake2 = "0.10.6"
 bytes.workspace = true
 common.workspace = true
 glob = "0.3.1"
@@ -23,6 +24,7 @@ serde_json.workspace = true
 tracing-error.workspace = true
 s3s.workspace = true
 http.workspace = true
+highway = "1.2.0"
 url.workspace = true
 uuid = { version = "1.8.0", features = ["v4", "fast-rng", "serde"] }
 reed-solomon-erasure = { version = "6.0.0", features = ["simd-accel"] }

ecstore/src/bitrot.rs

@@ -0,0 +1,167 @@
+use std::collections::HashMap;
+
+use blake2::{Blake2b, Blake2b512};
+use highway::{HighwayHash, HighwayHasher, Key};
+use lazy_static::lazy_static;
+use sha2::{Digest, Sha256};
+
+use crate::{
+    disk::DiskStore,
+    erasure::{ReadAt, Write},
+    error::Result,
+    store_api::BitrotAlgorithm,
+};
+
+lazy_static! {
+    static ref BITROT_ALGORITHMS: HashMap<BitrotAlgorithm, &'static str> = {
+        let mut m = HashMap::new();
+        m.insert(BitrotAlgorithm::SHA256, "sha256");
+        m.insert(BitrotAlgorithm::BLAKE2b512, "blake2b");
+        m.insert(BitrotAlgorithm::HighwayHash256, "highwayhash256");
+        m.insert(BitrotAlgorithm::HighwayHash256S, "highwayhash256S");
+        m
+    };
+}
+
+// const MAGIC_HIGHWAY_HASH256_KEY: &[u8] = &[
+//     0x4b, 0xe7, 0x34, 0xfa, 0x8e, 0x23, 0x8a, 0xcd, 0x26, 0x3e, 0x83, 0xe6, 0xbb, 0x96, 0x85, 0x52, 0x04, 0x0f, 0x93, 0x5d, 0xa3,
+//     0x9f, 0x44, 0x14, 0x97, 0xe0, 0x9d, 0x13, 0x22, 0xde, 0x36, 0xa0,
+// ];
+const MAGIC_HIGHWAY_HASH256_KEY: &[u64; 4] = &[3, 4, 2, 1];
+
+pub enum Hasher {
+    SHA256(Sha256),
+    HighwayHash256(HighwayHasher),
+    BLAKE2b512(Blake2b512),
+}
+
+impl Hasher {
+    pub fn update(&mut self, data: impl AsRef<[u8]>) {
+        match self {
+            Hasher::SHA256(core_wrapper) => {
+                core_wrapper.update(data);
+            }
+            Hasher::HighwayHash256(highway_hasher) => {
+                highway_hasher.append(data.as_ref());
+            }
+            Hasher::BLAKE2b512(core_wrapper) => {
+                core_wrapper.update(data);
+            }
+        }
+    }
+
+    pub fn finalize(self) -> Vec<u8> {
+        match self {
+            Hasher::SHA256(core_wrapper) => core_wrapper.finalize().to_vec(),
+            Hasher::HighwayHash256(highway_hasher) => highway_hasher
+                .finalize256()
+                .iter()
+                .flat_map(|&n| n.to_le_bytes()) // 使用小端字节序转换
+                .collect(),
+            Hasher::BLAKE2b512(core_wrapper) => core_wrapper.finalize().to_vec(),
+        }
+    }
+}
+
+impl BitrotAlgorithm {
+    pub fn new(&self) -> Hasher {
+        match self {
+            BitrotAlgorithm::SHA256 => Hasher::SHA256(Sha256::new()),
+            BitrotAlgorithm::HighwayHash256 | BitrotAlgorithm::HighwayHash256S => {
+                let key = Key(*MAGIC_HIGHWAY_HASH256_KEY);
+                Hasher::HighwayHash256(HighwayHasher::new(key))
+            }
+            BitrotAlgorithm::BLAKE2b512 => Hasher::BLAKE2b512(Blake2b512::new()),
+        }
+    }
+
+    pub fn available(&self) -> bool {
+        BITROT_ALGORITHMS.get(self).is_some()
+    }
+
+    pub fn string(&self) -> String {
+        BITROT_ALGORITHMS.get(self).map_or("".to_string(), |s| s.to_string())
+    }
+}
+
+pub struct BitrotVerifier {
+    algorithm: BitrotAlgorithm,
+    sum: Vec<u8>,
+}
+
+impl BitrotVerifier {
+    pub fn new(algorithm: BitrotAlgorithm, checksum: &[u8]) -> BitrotVerifier {
+        BitrotVerifier {
+            algorithm,
+            sum: checksum.to_vec(),
+        }
+    }
+}
+
+pub fn bitrot_algorithm_from_string(s: &str) -> BitrotAlgorithm {
+    for (k, v) in BITROT_ALGORITHMS.iter() {
+        if *v == s {
+            return k.clone();
+        }
+    }
+
+    BitrotAlgorithm::HighwayHash256S
+}
+
+type BitrotWriter = Box<dyn Write>;
+
+pub fn new_bitrot_writer(
+    disk: DiskStore,
+    orig_volume: &str,
+    volume: &str,
+    file_path: &str,
+    length: usize,
+    algo: BitrotAlgorithm,
+    shard_size: usize,
+) -> BitrotWriter {
+    todo!()
+}
+
+pub struct WholeBitrotWriter {
+    disk: DiskStore,
+    volume: String,
+    file_path: String,
+    shard_size: usize,
+    hash: Hasher,
+}
+
+impl WholeBitrotWriter {
+    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,
+            hash: algo.new(),
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl Write for WholeBitrotWriter {
+    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?;
+        self.hash.update(buf);
+
+        Ok(())
+    }
+}
+
+pub struct WholeBitrotReader {
+    disk: DiskStore,
+    volume: String,
+    file_path: String,
+}
+
+#[async_trait::async_trait]
+impl ReadAt for WholeBitrotReader {
+    async fn read_at(&mut self, offset: usize, length: usize) -> Result<(Vec<u8>, usize)> {
+        todo!()
+    }
+}

ecstore/src/disk/local.rs

@@ -2,9 +2,9 @@ use super::error::{is_sys_err_io, is_sys_err_not_empty, is_sys_err_too_many_file
 use super::os::is_root_disk;
 use super::{endpoint::Endpoint, error::DiskError, format::FormatV3};
 use super::{
-    os, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskMetrics, FileInfoVersions, FileReader, FileWriter,
-    Info, MetaCacheEntry, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, UpdateMetadataOpts, VolumeInfo,
-    WalkDirOptions,
+    os, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskMetrics, FileInfoVersions,
+    FileReader, FileWriter, Info, MetaCacheEntry, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp,
+    UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
 };
 use crate::cache_value::cache::{Cache, Opts};
 use crate::disk::error::{
@@ -15,16 +15,16 @@ use crate::disk::os::check_path_length;
 use crate::disk::{LocalFileReader, LocalFileWriter, STORAGE_FORMAT_FILE};
 use crate::error::{Error, Result};
 use crate::global::{GLOBAL_IsErasureSD, GLOBAL_RootDiskThreshold};
+use crate::store_api::BitrotAlgorithm;
 use crate::utils::fs::{lstat, O_APPEND, O_CREATE, O_RDONLY, O_WRONLY};
-use crate::utils::path::{clean, has_suffix, SLASH_SEPARATOR};
 use crate::utils::os::get_info;
+use crate::utils::path::{clean, has_suffix, SLASH_SEPARATOR};
 use crate::{
     file_meta::FileMeta,
     store_api::{FileInfo, RawFileInfo},
     utils,
 };
 use path_absolutize::Absolutize;
-use tokio::runtime::Runtime;
 use std::fmt::Debug;
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::Arc;
@@ -36,6 +36,7 @@ use std::{
 use time::OffsetDateTime;
 use tokio::fs::{self, File};
 use tokio::io::{AsyncReadExt, AsyncWriteExt, ErrorKind};
+use tokio::runtime::Runtime;
 use tokio::sync::RwLock;
 use tracing::{error, warn};
 use uuid::Uuid;
@@ -149,19 +150,17 @@ impl LocalDisk {
                         if root {
                             return Err(Error::new(DiskError::DriveIsRoot));
                         }
-        
+
                         // disk_info.healing =
                         Ok(disk_info)
-                    },
-                    Err(err) => {
-                        Err(err)
                     }
+                    Err(err) => Err(err),
                 }
             })
         };
-        
+
         let cache = Cache::new(Box::new(update_fn), Duration::from_secs(1), Opts::default());
-        
+
         // TODO: DIRECT suport
         // TODD: DiskInfo
         let mut disk = Self {
@@ -650,6 +649,22 @@ impl LocalDisk {
     fn get_metrics(&self) -> DiskMetrics {
         DiskMetrics::default()
     }
+
+    async fn bitrot_verify(
+        &self,
+        part_path: &PathBuf,
+        part_size: u64,
+        algo: BitrotAlgorithm,
+        sum: &[u8],
+        shard_size: u64,
+    ) -> Result<()> {
+        let file = utils::fs::open_file(part_path, O_CREATE | O_WRONLY)
+            .await
+            .map_err(os_err_to_file_err)?;
+
+
+        todo!()
+    }
 }
 
 fn is_root_path(path: impl AsRef<Path>) -> bool {
@@ -846,6 +861,30 @@ impl DiskAPI for LocalDisk {
         Ok(())
     }
 
+    async fn verify_file(&self, volume: &str, path: &str, fi: FileInfo) -> Result<CheckPartsResp> {
+        let volume_dir = self.get_bucket_path(volume)?;
+        if !skip_access_checks(volume) {
+            if let Err(e) = utils::fs::access(&volume_dir).await {
+                return Err(convert_access_error(e, DiskError::VolumeAccessDenied));
+            }
+        }
+
+        let mut resp = CheckPartsResp {
+            results: Vec::with_capacity(fi.parts.len()),
+        };
+
+        let erasure = fi.erasure;
+        fi.parts.iter().enumerate().for_each(|(i, part)| {
+            let checksum_info = erasure.get_checksum_info(part.number);
+            let part_path = Path::new(&volume_dir)
+                .join(path)
+                .join(fi.data_dir.map_or("".to_string(), |dir| dir.to_string()))
+                .join(format!("part.{}", part.number));
+
+            self.bi
+        });
+    }
+
     async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Vec<u8>) -> Result<()> {
         let src_volume_dir = self.get_bucket_path(src_volume)?;
         let dst_volume_dir = self.get_bucket_path(dst_volume)?;

ecstore/src/disk/mod.rs

@@ -127,6 +127,7 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
     // CheckParts
     async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()>;
     // VerifyFile
+    async fn verify_file(&self, volume: &str, path: &str, fi: FileInfo) -> Result<CheckPartsResp>;
     // StatInfoFile
     // ReadParts
     async fn read_multiple(&self, req: ReadMultipleReq) -> Result<Vec<ReadMultipleResp>>;
@@ -136,6 +137,10 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
     async fn disk_info(&self, opts: &DiskInfoOptions) -> Result<DiskInfo>;
 }
 
+pub struct CheckPartsResp {
+    pub results: Vec<usize>,
+}
+
 #[derive(Debug, Serialize, Deserialize)]
 pub struct UpdateMetadataOpts {
     pub no_persistence: bool,

ecstore/src/disk/remote.rs

@@ -15,9 +15,7 @@ use tracing::info;
 use uuid::Uuid;
 
 use super::{
-    endpoint::Endpoint, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, FileInfoVersions,
-    FileReader, FileWriter, MetaCacheEntry, ReadMultipleReq, ReadMultipleResp, ReadOptions, RemoteFileReader, RemoteFileWriter,
-    RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
+    endpoint::Endpoint, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, FileInfoVersions, FileReader, FileWriter, MetaCacheEntry, ReadMultipleReq, ReadMultipleResp, ReadOptions, RemoteFileReader, RemoteFileWriter, RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions
 };
 use crate::{
     disk::error::DiskError,
@@ -163,6 +161,11 @@ impl DiskAPI for RemoteDisk {
 
         Ok(())
     }
+
+    async fn verify_file(&self, volume: &str, path: &str, fi: FileInfo) -> Result<CheckPartsResp> {
+        unimplemented!()
+    }
+
     async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Vec<u8>) -> Result<()> {
         info!("rename_part");
         let mut client = node_service_time_out_client(&self.addr)

ecstore/src/lib.rs

@@ -1,3 +1,4 @@
+pub mod bitrot;
 pub mod cache_value;
 mod chunk_stream;
 mod config;

ecstore/src/store_api.rs

@@ -12,6 +12,7 @@ use http::HeaderMap;
 use rmp_serde::Serializer;
 use s3s::dto::StreamingBlob;
 use serde::{Deserialize, Serialize};
+use sha2::Sha256;
 use time::OffsetDateTime;
 use uuid::Uuid;
 
@@ -240,6 +241,18 @@ pub struct ErasureInfo {
     pub checksums: Vec<ChecksumInfo>,
 }
 
+impl ErasureInfo {
+    pub fn get_checksum_info(&self, part_number: usize) -> ChecksumInfo {
+        for sum in &self.checksums {
+            if sum.part_number == part_number {
+                return sum.clone();
+            }
+        }
+
+        ChecksumInfo {algorithm: BitrotAlgorithm::HighwayHash256S, ..Default::default()}
+    }
+}
+
 #[derive(Serialize, Deserialize, Debug, PartialEq, Default, Clone)]
 // ChecksumInfo - carries checksums of individual scattered parts per disk.
 pub struct ChecksumInfo {
@@ -248,7 +261,7 @@ pub struct ChecksumInfo {
     pub hash: Vec<u8>,
 }
 
-#[derive(Serialize, Deserialize, Debug, PartialEq, Default, Clone)]
+#[derive(Serialize, Deserialize, Debug, PartialEq, Default, Clone, Eq, Hash)]
 // BitrotAlgorithm specifies a algorithm used for bitrot protection.
 pub enum BitrotAlgorithm {
     // SHA256 represents the SHA-256 hash function

ecstore/src/utils/path.rs

@@ -1,3 +1,5 @@
+use std::path::PathBuf;
+
 const GLOBAL_DIR_SUFFIX: &str = "__XLDIR__";
 
 pub const SLASH_SEPARATOR: &str = "/";
@@ -50,6 +52,16 @@ pub fn has_profix(s: &str, prefix: &str) -> bool {
     s.starts_with(prefix)
 }
 
+pub fn path_join(elem: &[PathBuf]) -> PathBuf {
+    let mut joined_path = PathBuf::new();
+    
+    for path in elem {
+        joined_path.push(path);
+    }
+
+    joined_path
+}
+
 pub struct LazyBuf {
     s: String,
     buf: Option<Vec<u8>>,