fix: 去除anyhow提供的错误处理，统一使用自定义的Error和Result

ecstore/src/disk/error.rs

@@ -0,0 +1,99 @@
+use crate::error::{Error, Result, StdError};
+
+#[derive(Debug, thiserror::Error)]
+pub enum DiskError {
+    #[error("file not found")]
+    FileNotFound,
+
+    #[error("file version not found")]
+    FileVersionNotFound,
+
+    #[error("disk not found")]
+    DiskNotFound,
+
+    #[error("disk access denied")]
+    FileAccessDenied,
+
+    #[error("InconsistentDisk")]
+    InconsistentDisk,
+
+    #[error("volume already exists")]
+    VolumeExists,
+
+    #[error("unformatted disk error")]
+    UnformattedDisk,
+
+    #[error("unsupport disk")]
+    UnsupportedDisk,
+
+    #[error("disk not a dir")]
+    DiskNotDir,
+
+    #[error("volume not found")]
+    VolumeNotFound,
+}
+
+impl DiskError {
+    pub fn check_disk_fatal_errs(errs: &Vec<Option<Error>>) -> Result<()> {
+        if Self::count_errs(errs, &DiskError::UnsupportedDisk) == errs.len() {
+            return Err(DiskError::UnsupportedDisk.into());
+        }
+
+        if Self::count_errs(errs, &DiskError::FileAccessDenied) == errs.len() {
+            return Err(DiskError::FileAccessDenied.into());
+        }
+
+        if Self::count_errs(errs, &DiskError::DiskNotDir) == errs.len() {
+            return Err(DiskError::DiskNotDir.into());
+        }
+
+        Ok(())
+    }
+
+    pub fn count_errs(errs: &Vec<Option<Error>>, err: &DiskError) -> usize {
+        return errs
+            .iter()
+            .filter(|&e| {
+                if e.is_some() {
+                    let e = e.as_ref().unwrap();
+                    let cast = e.downcast_ref::<DiskError>();
+                    if cast.is_some() {
+                        let cast = cast.unwrap();
+                        return cast == err;
+                    }
+                }
+                false
+            })
+            .count();
+    }
+
+    pub fn quorum_unformatted_disks(errs: &Vec<Option<Error>>) -> bool {
+        Self::count_errs(errs, &DiskError::UnformattedDisk) >= (errs.len() / 2) + 1
+    }
+
+    pub fn is(&self, err: &Error) -> bool {
+        if let Some(e) = err.downcast_ref::<DiskError>() {
+            e == self
+        } else {
+            false
+        }
+    }
+
+    pub fn is_err<T: std::error::Error + ?Sized>(err: &T, disk_err: &DiskError) -> bool {
+        return false;
+        // let cast = err.
+        // if cast.is_none() {
+        //     return false;
+        // }
+
+        // let e = cast.unwrap();
+
+        // e == disk_err
+    }
+}
+
+impl PartialEq for DiskError {
+    fn eq(&self, other: &Self) -> bool {
+        core::mem::discriminant(self) == core::mem::discriminant(other)
+    }
+}

ecstore/src/disk/format.rs

@@ -0,0 +1,222 @@
+use super::error::DiskError;
+use crate::error::{Error, Result};
+use serde::{Deserialize, Serialize};
+use serde_json::Error as JsonError;
+use uuid::Uuid;
+
+#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
+pub enum FormatMetaVersion {
+    #[serde(rename = "1")]
+    V1,
+
+    #[serde(other)]
+    Unknown,
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
+pub enum FormatBackend {
+    #[serde(rename = "xl")]
+    Erasure,
+    #[serde(rename = "xl-single")]
+    ErasureSingle,
+
+    #[serde(other)]
+    Unknown,
+}
+
+/// Represents the V3 backend disk structure version
+/// under `.rustfs.sys` and actual data namespace.
+///
+/// FormatErasureV3 - structure holds format config version '3'.
+///
+/// The V3 format to support "large bucket" support where a bucket
+/// can span multiple erasure sets.
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct FormatErasureV3 {
+    /// Version of 'xl' format.
+    pub version: FormatErasureVersion,
+
+    /// This field carries assigned disk uuid.
+    pub this: Uuid,
+
+    /// Sets field carries the input disk order generated the first
+    /// time when fresh disks were supplied, it is a two dimensional
+    /// array second dimension represents list of disks used per set.
+    pub sets: Vec<Vec<Uuid>>,
+
+    /// Distribution algorithm represents the hashing algorithm
+    /// to pick the right set index for an object.
+    #[serde(rename = "distributionAlgo")]
+    pub distribution_algo: DistributionAlgoVersion,
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
+pub enum FormatErasureVersion {
+    #[serde(rename = "1")]
+    V1,
+    #[serde(rename = "2")]
+    V2,
+    #[serde(rename = "3")]
+    V3,
+
+    #[serde(other)]
+    Unknown,
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
+pub enum DistributionAlgoVersion {
+    #[serde(rename = "CRCMOD")]
+    V1,
+    #[serde(rename = "SIPMOD")]
+    V2,
+    #[serde(rename = "SIPMOD+PARITY")]
+    V3,
+}
+
+/// format.json currently has the format:
+///
+/// ```json
+/// {
+///   "version": "1",
+///   "format": "XXXXX",
+///   "id": "XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX",
+///   "XXXXX": {
+//
+///   }
+/// }
+/// ```
+///
+/// Ideally we will never have a situation where we will have to change the
+/// fields of this struct and deal with related migration.
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct FormatV3 {
+    /// Version of the format config.
+    pub version: FormatMetaVersion,
+
+    /// Format indicates the backend format type, supports two values 'xl' and 'xl-single'.
+    pub format: FormatBackend,
+
+    /// ID is the identifier for the rustfs deployment
+    pub id: Uuid,
+
+    #[serde(rename = "xl")]
+    pub erasure: FormatErasureV3,
+    // /// DiskInfo is an extended type which returns current
+    // /// disk usage per path.
+    // #[serde(skip)]
+    // pub disk_info: Option<data_types::DeskInfo>,
+}
+
+impl TryFrom<&[u8]> for FormatV3 {
+    type Error = JsonError;
+
+    fn try_from(data: &[u8]) -> Result<Self, JsonError> {
+        serde_json::from_slice(data)
+    }
+}
+
+impl TryFrom<&str> for FormatV3 {
+    type Error = JsonError;
+
+    fn try_from(data: &str) -> Result<Self, JsonError> {
+        serde_json::from_str(data)
+    }
+}
+
+impl FormatV3 {
+    /// Create a new format config with the given number of sets and set length.
+    pub fn new(num_sets: usize, set_len: usize) -> Self {
+        let format = if set_len == 1 {
+            FormatBackend::ErasureSingle
+        } else {
+            FormatBackend::Erasure
+        };
+
+        let erasure = FormatErasureV3 {
+            version: FormatErasureVersion::V3,
+            this: Uuid::nil(),
+            sets: (0..num_sets)
+                .map(|_| (0..set_len).map(|_| Uuid::new_v4()).collect())
+                .collect(),
+            distribution_algo: DistributionAlgoVersion::V3,
+        };
+
+        Self {
+            version: FormatMetaVersion::V1,
+            format,
+            id: Uuid::new_v4(),
+            erasure,
+            // disk_info: None,
+        }
+    }
+
+    /// Returns the number of drives in the erasure set.
+    pub fn drives(&self) -> usize {
+        self.erasure.sets.iter().map(|v| v.len()).sum()
+    }
+
+    pub fn to_json(&self) -> Result<String, JsonError> {
+        serde_json::to_string(self)
+    }
+
+    /// returns the i,j'th position of the input `diskID` against the reference
+    ///
+    /// format, after successful validation.
+    ///   - i'th position is the set index
+    ///   - j'th position is the disk index in the current set
+    pub fn find_disk_index_by_disk_id(&self, disk_id: Uuid) -> Result<(usize, usize)> {
+        if disk_id == Uuid::nil() {
+            return Err(Error::from(DiskError::DiskNotFound));
+        }
+        if disk_id == Uuid::max() {
+            return Err(Error::msg("disk offline"));
+        }
+
+        for (i, set) in self.erasure.sets.iter().enumerate() {
+            for (j, d) in set.iter().enumerate() {
+                if disk_id.eq(d) {
+                    return Ok((i, j));
+                }
+            }
+        }
+
+        Err(Error::msg(format!("disk id not found {}", disk_id)))
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[test]
+    fn test_format_v1() {
+        let format = FormatV3::new(1, 4);
+
+        let str = serde_json::to_string(&format);
+        println!("{:?}", str);
+
+        let data = r#"
+        {
+            "version": "1",
+            "format": "xl",
+            "id": "321b3874-987d-4c15-8fa5-757c956b1243",
+            "xl": {
+                "version": "1",
+                "this": null,
+                "sets": [
+                    [
+                        "8ab9a908-f869-4f1f-8e42-eb067ffa7eb5",
+                        "c26315da-05cf-4778-a9ea-b44ea09f58c5",
+                        "fb87a891-18d3-44cf-a46f-bcc15093a038",
+                        "356a925c-57b9-4313-88b3-053edf1104dc"
+                    ]
+                ],
+                "distributionAlgo": "CRCMOD"
+            }
+        }"#;
+
+        let p = FormatV3::try_from(data);
+
+        println!("{:?}", p);
+    }
+}

ecstore/src/disk/local.rs

@@ -0,0 +1,905 @@
+use super::{error::DiskError, format::FormatV3};
+use bytes::Bytes;
+use futures::future::join_all;
+use path_absolutize::Absolutize;
+use std::{
+    fs::Metadata,
+    io::{self, SeekFrom},
+    path::{Path, PathBuf},
+    sync::Arc,
+};
+use time::OffsetDateTime;
+use tokio::io::{AsyncReadExt, ErrorKind};
+use tokio::{
+    fs::{self, File},
+    io::AsyncSeekExt,
+};
+use tracing::{debug, warn};
+use uuid::Uuid;
+
+use crate::{
+    disk_api::{
+        DeleteOptions, DiskAPI, FileReader, FileWriter, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp,
+        VolumeInfo,
+    },
+    endpoint::{Endpoint, Endpoints},
+    erasure::ReadAt,
+    error::{Error, Result},
+    file_meta::FileMeta,
+    store_api::{FileInfo, RawFileInfo},
+    utils,
+};
+
+pub type DiskStore = Arc<Box<dyn DiskAPI>>;
+
+pub struct DiskOption {
+    pub cleanup: bool,
+    pub health_check: bool,
+}
+
+pub async fn new_disk(ep: &Endpoint, opt: &DiskOption) -> Result<DiskStore> {
+    if ep.is_local {
+        let s = LocalDisk::new(ep, opt.cleanup).await?;
+        Ok(Arc::new(Box::new(s)))
+    } else {
+        let _ = opt.health_check;
+        unimplemented!()
+        // Ok(Disk::Remote(RemoteDisk::new(ep, opt.health_check)?))
+    }
+}
+
+pub async fn init_disks(eps: &Endpoints, opt: &DiskOption) -> (Vec<Option<DiskStore>>, Vec<Option<Error>>) {
+    let mut futures = Vec::with_capacity(eps.as_ref().len());
+
+    for ep in eps.as_ref().iter() {
+        futures.push(new_disk(ep, opt));
+    }
+
+    let mut res = Vec::with_capacity(eps.as_ref().len());
+    let mut errors = Vec::with_capacity(eps.as_ref().len());
+
+    let results = join_all(futures).await;
+    for result in results {
+        match result {
+            Ok(s) => {
+                res.push(Some(s));
+                errors.push(None);
+            }
+            Err(e) => {
+                res.push(None);
+                errors.push(Some(e));
+            }
+        }
+    }
+
+    (res, errors)
+}
+
+// pub async fn load_format(&self, heal: bool) -> Result<FormatV3> {
+//     unimplemented!()
+// }
+
+#[derive(Debug)]
+pub struct LocalDisk {
+    pub root: PathBuf,
+    pub id: Uuid,
+    pub format_data: Vec<u8>,
+    pub format_meta: Option<Metadata>,
+    pub format_path: PathBuf,
+    // pub format_legacy: bool, // drop
+    pub format_last_check: OffsetDateTime,
+}
+
+impl LocalDisk {
+    pub async fn new(ep: &Endpoint, cleanup: bool) -> Result<Self> {
+        let root = fs::canonicalize(ep.url.path()).await?;
+
+        if cleanup {
+            // TODO: 删除tmp数据
+        }
+
+        let format_path = Path::new(super::RUSTFS_META_BUCKET)
+            .join(Path::new(super::FORMAT_CONFIG_FILE))
+            .absolutize_virtually(&root)?
+            .into_owned();
+
+        let (format_data, format_meta) = read_file_exists(&format_path).await?;
+
+        let mut id = Uuid::nil();
+        // let mut format_legacy = false;
+        let mut format_last_check = OffsetDateTime::UNIX_EPOCH;
+
+        if !format_data.is_empty() {
+            let s = format_data.as_slice();
+            let fm = FormatV3::try_from(s)?;
+            let (set_idx, disk_idx) = fm.find_disk_index_by_disk_id(fm.erasure.this)?;
+
+            if Some(set_idx) != ep.set_idx || Some(disk_idx) != ep.disk_idx {
+                return Err(Error::from(DiskError::InconsistentDisk));
+            }
+
+            id = fm.erasure.this;
+            // format_legacy = fm.erasure.distribution_algo == DistributionAlgoVersion::V1;
+            format_last_check = OffsetDateTime::now_utc();
+        }
+
+        let disk = Self {
+            root,
+            id,
+            format_meta,
+            format_data: format_data,
+            format_path,
+            // format_legacy,
+            format_last_check,
+        };
+
+        disk.make_meta_volumes().await?;
+
+        Ok(disk)
+    }
+
+    async fn make_meta_volumes(&self) -> Result<()> {
+        let buckets = format!("{}/{}", super::RUSTFS_META_BUCKET, super::BUCKET_META_PREFIX);
+        let multipart = format!("{}/{}", super::RUSTFS_META_BUCKET, "multipart");
+        let config = format!("{}/{}", super::RUSTFS_META_BUCKET, "config");
+        let tmp = format!("{}/{}", super::RUSTFS_META_BUCKET, "tmp");
+        let defaults = vec![buckets.as_str(), multipart.as_str(), config.as_str(), tmp.as_str()];
+
+        self.make_volumes(defaults).await
+    }
+
+    pub fn resolve_abs_path(&self, path: impl AsRef<Path>) -> Result<PathBuf> {
+        Ok(path.as_ref().absolutize_virtually(&self.root)?.into_owned())
+    }
+
+    pub fn get_object_path(&self, bucket: &str, key: &str) -> Result<PathBuf> {
+        let dir = Path::new(&bucket);
+        let file_path = Path::new(&key);
+        self.resolve_abs_path(dir.join(file_path))
+    }
+
+    pub fn get_bucket_path(&self, bucket: &str) -> Result<PathBuf> {
+        let dir = Path::new(&bucket);
+        self.resolve_abs_path(dir)
+    }
+
+    // /// Write to the filesystem atomically.
+    // /// This is done by first writing to a temporary location and then moving the file.
+    // pub(crate) async fn prepare_file_write<'a>(&self, path: &'a PathBuf) -> Result<FileWriter<'a>> {
+    //     let tmp_path = self.get_object_path(RUSTFS_META_TMP_BUCKET, Uuid::new_v4().to_string().as_str())?;
+
+    //     debug!("prepare_file_write tmp_path:{:?}, path:{:?}", &tmp_path, &path);
+
+    //     let file = File::create(&tmp_path).await?;
+    //     let writer = BufWriter::new(file);
+    //     Ok(FileWriter {
+    //         tmp_path,
+    //         dest_path: path,
+    //         writer,
+    //         clean_tmp: true,
+    //     })
+    // }
+
+    pub async fn rename_all(&self, src_data_path: &PathBuf, dst_data_path: &PathBuf, skip: &PathBuf) -> Result<()> {
+        if !skip.starts_with(&src_data_path) {
+            fs::create_dir_all(dst_data_path.parent().unwrap_or(Path::new("/"))).await?;
+        }
+
+        debug!(
+            "rename_all from \n {:?} \n to \n {:?} \n skip:{:?}",
+            &src_data_path, &dst_data_path, &skip
+        );
+
+        fs::rename(&src_data_path, &dst_data_path).await?;
+
+        Ok(())
+    }
+
+    pub async fn delete_file(&self, base_path: &PathBuf, delete_path: &PathBuf, recursive: bool, _immediate: bool) -> Result<()> {
+        debug!("delete_file {:?}", &delete_path);
+
+        if is_root_path(base_path) || is_root_path(delete_path) {
+            return Ok(());
+        }
+
+        if !delete_path.starts_with(base_path) || base_path == delete_path {
+            return Ok(());
+        }
+
+        if recursive {
+            let trash_path = self.get_object_path(super::RUSTFS_META_TMP_DELETED_BUCKET, Uuid::new_v4().to_string().as_str())?;
+            // fs::create_dir_all(&trash_path).await?;
+            fs::rename(&delete_path, &trash_path).await.map_err(|err| {
+                // 使用文件路径自定义错误信息
+                io::Error::new(
+                    err.kind(),
+                    format!("Failed to rename file '{:?}' to '{:?}': {}", &delete_path, &trash_path, err),
+                )
+            })?;
+
+            // TODO: immediate
+
+            return Ok(());
+        }
+
+        if delete_path.is_dir() {
+            fs::remove_dir(delete_path).await?;
+        } else {
+            fs::remove_file(delete_path).await?;
+        }
+
+        if let Some(dir_path) = delete_path.parent() {
+            Box::pin(self.delete_file(base_path, &PathBuf::from(dir_path), false, false)).await?;
+        }
+
+        Ok(())
+    }
+
+    /// read xl.meta raw data
+    async fn read_raw(
+        &self,
+        bucket: &str,
+        volume_dir: impl AsRef<Path>,
+        path: impl AsRef<Path>,
+        read_data: bool,
+    ) -> Result<(Vec<u8>, OffsetDateTime)> {
+        let meta_path = path.as_ref().join(Path::new(super::STORAGE_FORMAT_FILE));
+        if read_data {
+            self.read_all_data(bucket, volume_dir, meta_path).await
+        } else {
+            self.read_metadata_with_dmtime(meta_path).await
+        }
+    }
+
+    async fn read_metadata_with_dmtime(&self, path: impl AsRef<Path>) -> Result<(Vec<u8>, OffsetDateTime)> {
+        let (data, meta) = read_file_all(path).await?;
+
+        let modtime = match meta.modified() {
+            Ok(md) => OffsetDateTime::from(md),
+            Err(_) => return Err(Error::msg("Not supported modified on this platform")),
+        };
+
+        Ok((data, modtime))
+    }
+
+    async fn read_all_data(
+        &self,
+        _bucket: &str,
+        _volume_dir: impl AsRef<Path>,
+        path: impl AsRef<Path>,
+    ) -> Result<(Vec<u8>, OffsetDateTime)> {
+        let (data, meta) = read_file_all(path).await?;
+
+        let modtime = match meta.modified() {
+            Ok(md) => OffsetDateTime::from(md),
+            Err(_) => return Err(Error::msg("Not supported modified on this platform")),
+        };
+
+        Ok((data, modtime))
+    }
+}
+
+fn is_root_path(path: &PathBuf) -> bool {
+    path.components().count() == 1 && path.has_root()
+}
+
+// 过滤 std::io::ErrorKind::NotFound
+pub async fn read_file_exists(path: impl AsRef<Path>) -> Result<(Vec<u8>, Option<Metadata>)> {
+    let p = path.as_ref();
+    let (data, meta) = match read_file_all(&p).await {
+        Ok((data, meta)) => (data, Some(meta)),
+        Err(e) => {
+            if DiskError::FileNotFound.is(&e) {
+                (Vec::new(), None)
+            } else {
+                return Err(e);
+            }
+        }
+    };
+
+    // let mut data = Vec::new();
+    // if meta.is_some() {
+    //     data = fs::read(&p).await?;
+    // }
+
+    Ok((data, meta))
+}
+
+pub async fn write_all_internal(p: impl AsRef<Path>, data: impl AsRef<[u8]>) -> Result<()> {
+    // create top dir if not exists
+    fs::create_dir_all(&p.as_ref().parent().unwrap_or_else(|| Path::new("."))).await?;
+
+    fs::write(&p, data).await?;
+    Ok(())
+}
+
+pub async fn read_file_all(path: impl AsRef<Path>) -> Result<(Vec<u8>, Metadata)> {
+    let p = path.as_ref();
+    let meta = read_file_metadata(&path).await?;
+
+    let data = fs::read(&p).await?;
+
+    Ok((data, meta))
+}
+
+pub async fn read_file_metadata(p: impl AsRef<Path>) -> Result<Metadata> {
+    let meta = fs::metadata(&p).await.map_err(|e| match e.kind() {
+        ErrorKind::NotFound => Error::from(DiskError::FileNotFound),
+        ErrorKind::PermissionDenied => Error::from(DiskError::FileAccessDenied),
+        _ => Error::from(e),
+    })?;
+
+    Ok(meta)
+}
+
+pub async fn check_volume_exists(p: impl AsRef<Path>) -> Result<()> {
+    fs::metadata(&p).await.map_err(|e| match e.kind() {
+        ErrorKind::NotFound => Error::from(DiskError::VolumeNotFound),
+        ErrorKind::PermissionDenied => Error::from(DiskError::FileAccessDenied),
+        _ => Error::from(e),
+    })?;
+    Ok(())
+}
+
+fn skip_access_checks(p: impl AsRef<str>) -> bool {
+    let vols = vec![
+        super::RUSTFS_META_TMP_DELETED_BUCKET,
+        super::RUSTFS_META_TMP_BUCKET,
+        super::RUSTFS_META_MULTIPART_BUCKET,
+        super::RUSTFS_META_BUCKET,
+    ];
+
+    for v in vols.iter() {
+        if p.as_ref().starts_with(v) {
+            return true;
+        }
+    }
+
+    false
+}
+
+#[async_trait::async_trait]
+impl DiskAPI for LocalDisk {
+    fn is_local(&self) -> bool {
+        true
+    }
+
+    fn id(&self) -> Uuid {
+        self.id
+    }
+
+    #[must_use]
+    async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes> {
+        let p = self.get_object_path(&volume, &path)?;
+        let (data, _) = read_file_all(&p).await?;
+
+        Ok(Bytes::from(data))
+    }
+
+    async fn write_all(&self, volume: &str, path: &str, data: Vec<u8>) -> Result<()> {
+        let p = self.get_object_path(&volume, &path)?;
+
+        write_all_internal(p, data).await?;
+
+        Ok(())
+    }
+
+    async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> {
+        let vol_path = self.get_bucket_path(&volume)?;
+        if !skip_access_checks(&volume) {
+            check_volume_exists(&vol_path).await?;
+        }
+
+        let fpath = self.get_object_path(&volume, &path)?;
+
+        self.delete_file(&vol_path, &fpath, opt.recursive, opt.immediate).await?;
+
+        // if opt.recursive {
+        //     let trash_path = self.get_object_path(RUSTFS_META_TMP_DELETED_BUCKET, Uuid::new_v4().to_string().as_str())?;
+        //     fs::create_dir_all(&trash_path).await?;
+        //     fs::rename(&fpath, &trash_path).await?;
+
+        //     // TODO: immediate
+
+        //     return Ok(());
+        // }
+
+        // fs::remove_file(fpath).await?;
+
+        Ok(())
+    }
+
+    async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()> {
+        let src_volume_path = self.get_bucket_path(&src_volume)?;
+        if !skip_access_checks(&src_volume) {
+            check_volume_exists(&src_volume_path).await?;
+        }
+        if !skip_access_checks(&dst_volume) {
+            let vol_path = self.get_bucket_path(&dst_volume)?;
+            check_volume_exists(&vol_path).await?;
+        }
+
+        let srcp = self.get_object_path(&src_volume, &src_path)?;
+        let dstp = self.get_object_path(&dst_volume, &dst_path)?;
+
+        let src_is_dir = srcp.is_dir();
+        let dst_is_dir = dstp.is_dir();
+        if !(src_is_dir && dst_is_dir || !src_is_dir && !dst_is_dir) {
+            return Err(Error::from(DiskError::FileAccessDenied));
+        }
+
+        // TODO: check path length
+
+        if src_is_dir {
+            // TODO: remove dst_dir
+        }
+
+        fs::create_dir_all(dstp.parent().unwrap_or_else(|| Path::new("."))).await?;
+
+        let mut idx = 0;
+        loop {
+            if let Err(e) = fs::rename(&srcp, &dstp).await {
+                if e.kind() == ErrorKind::NotFound && idx == 0 {
+                    idx += 1;
+                    continue;
+                }
+            };
+
+            break;
+        }
+
+        if let Some(dir_path) = srcp.parent() {
+            self.delete_file(&src_volume_path, &PathBuf::from(dir_path), false, false)
+                .await?;
+        }
+
+        Ok(())
+    }
+
+    async fn create_file(&self, _origvolume: &str, volume: &str, path: &str, _file_size: usize) -> Result<FileWriter> {
+        let fpath = self.get_object_path(volume, path)?;
+
+        debug!("CreateFile fpath: {:?}", fpath);
+
+        if let Some(_dir_path) = fpath.parent() {
+            fs::create_dir_all(&_dir_path).await?;
+        }
+
+        let file = File::create(&fpath).await?;
+
+        Ok(FileWriter::new(file))
+
+        // let mut writer = BufWriter::new(file);
+
+        // io::copy(&mut r, &mut writer).await?;
+
+        // Ok(())
+    }
+    // async fn append_file(&self, volume: &str, path: &str, mut r: DuplexStream) -> Result<File> {
+    async fn append_file(&self, volume: &str, path: &str) -> Result<FileWriter> {
+        let p = self.get_object_path(&volume, &path)?;
+
+        if let Some(dir_path) = p.parent() {
+            fs::create_dir_all(&dir_path).await?;
+        }
+
+        // debug!("append_file open {} {:?}", self.id(), &p);
+
+        let file = File::options()
+            .read(true)
+            .create(true)
+            .write(true)
+            .append(true)
+            .open(&p)
+            .await?;
+
+        Ok(FileWriter::new(file))
+
+        // let mut writer = BufWriter::new(file);
+
+        // io::copy(&mut r, &mut writer).await?;
+
+        // debug!("append_file end {} {}", self.id(), path);
+        // io::copy(&mut r, &mut file).await?;
+
+        // Ok(())
+    }
+    async fn read_file(&self, volume: &str, path: &str) -> Result<FileReader> {
+        let p = self.get_object_path(&volume, &path)?;
+
+        debug!("read_file {:?}", &p);
+        let file = File::options().read(true).open(&p).await?;
+
+        Ok(FileReader::new(file))
+
+        // file.seek(SeekFrom::Start(offset as u64)).await?;
+
+        // let mut buffer = vec![0; length];
+
+        // let bytes_read = file.read(&mut buffer).await?;
+
+        // buffer.truncate(bytes_read);
+
+        // Ok((buffer, bytes_read))
+    }
+    async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: usize) -> Result<Vec<String>> {
+        let p = self.get_bucket_path(&volume)?;
+
+        let mut entries = fs::read_dir(&p).await?;
+
+        let mut volumes = Vec::new();
+
+        while let Some(entry) = entries.next_entry().await? {
+            if let Ok(metadata) = entry.metadata().await {
+                // if !metadata.is_dir() {
+                //     continue;
+                // }
+
+                let name = entry.file_name().to_string_lossy().to_string();
+
+                // let created = match metadata.created() {
+                //     Ok(md) => OffsetDateTime::from(md),
+                //     Err(_) => return Err(Error::msg("Not supported created on this platform")),
+                // };
+
+                volumes.push(name);
+            }
+        }
+
+        Ok(volumes)
+    }
+    async fn walk_dir(&self) -> Result<Vec<FileInfo>> {
+        unimplemented!()
+    }
+    async fn rename_data(
+        &self,
+        src_volume: &str,
+        src_path: &str,
+        fi: FileInfo,
+        dst_volume: &str,
+        dst_path: &str,
+    ) -> Result<RenameDataResp> {
+        let src_volume_path = self.get_bucket_path(&src_volume)?;
+        if !skip_access_checks(&src_volume) {
+            check_volume_exists(&src_volume_path).await?;
+        }
+
+        let dst_volume_path = self.get_bucket_path(&dst_volume)?;
+        if !skip_access_checks(&dst_volume) {
+            check_volume_exists(&dst_volume_path).await?;
+        }
+
+        let src_file_path =
+            self.get_object_path(&src_volume, format!("{}/{}", &src_path, super::STORAGE_FORMAT_FILE).as_str())?;
+        let dst_file_path =
+            self.get_object_path(&dst_volume, format!("{}/{}", &dst_path, super::STORAGE_FORMAT_FILE).as_str())?;
+
+        let (src_data_path, dst_data_path) = {
+            let mut data_dir = String::new();
+            if !fi.is_remote() {
+                data_dir = utils::path::retain_slash(fi.data_dir.to_string().as_str());
+            }
+
+            if !data_dir.is_empty() {
+                let src_data_path = self.get_object_path(
+                    &src_volume,
+                    utils::path::retain_slash(format!("{}/{}", &src_path, data_dir).as_str()).as_str(),
+                )?;
+                let dst_data_path = self.get_object_path(&dst_volume, format!("{}/{}", &dst_path, data_dir).as_str())?;
+
+                (src_data_path, dst_data_path)
+            } else {
+                (PathBuf::new(), PathBuf::new())
+            }
+        };
+
+        // let curreng_data_path = self.get_object_path(&dst_volume, &dst_path);
+
+        let mut meta = FileMeta::new();
+
+        let (dst_buf, _) = read_file_exists(&dst_file_path).await?;
+
+        let mut skip_parent = dst_volume_path;
+        if !&dst_buf.is_empty() {
+            skip_parent = PathBuf::from(&dst_file_path.parent().unwrap_or(Path::new("/")));
+        }
+
+        if !dst_buf.is_empty() {
+            meta = match FileMeta::unmarshal(&dst_buf) {
+                Ok(m) => m,
+                Err(_) => FileMeta::new(),
+            }
+            // xl.load
+            // meta.from(dst_buf);
+        }
+
+        meta.add_version(fi.clone())?;
+
+        let fm_data = meta.marshal_msg()?;
+
+        fs::write(&src_file_path, fm_data).await?;
+
+        let no_inline = src_data_path.has_root() && fi.data.is_none() && fi.size > 0;
+        if no_inline {
+            self.rename_all(&src_data_path, &dst_data_path, &skip_parent).await?;
+        }
+
+        self.rename_all(&src_file_path, &dst_file_path, &skip_parent).await?;
+
+        if src_volume != super::RUSTFS_META_MULTIPART_BUCKET {
+            fs::remove_dir(&src_file_path.parent().unwrap()).await?;
+        } else {
+            self.delete_file(&src_volume_path, &PathBuf::from(src_file_path.parent().unwrap()), true, false)
+                .await?;
+        }
+
+        Ok(RenameDataResp {
+            old_data_dir: fi.data_dir.to_string(),
+        })
+    }
+
+    async fn make_volumes(&self, volumes: Vec<&str>) -> Result<()> {
+        for vol in volumes {
+            if let Err(e) = self.make_volume(vol).await {
+                match &e.downcast_ref::<DiskError>() {
+                    Some(DiskError::VolumeExists) => Ok(()),
+                    Some(_) => Err(e),
+                    None => Err(e),
+                }?;
+            }
+            // TODO: health check
+        }
+        Ok(())
+    }
+    async fn make_volume(&self, volume: &str) -> Result<()> {
+        let p = self.get_bucket_path(&volume)?;
+        match File::open(&p).await {
+            Ok(_) => (),
+            Err(e) => match e.kind() {
+                ErrorKind::NotFound => {
+                    fs::create_dir_all(&p).await?;
+                    return Ok(());
+                }
+                _ => return Err(Error::from(e)),
+            },
+        }
+
+        Err(Error::from(DiskError::VolumeExists))
+    }
+    async fn list_volumes(&self) -> Result<Vec<VolumeInfo>> {
+        let mut entries = fs::read_dir(&self.root).await?;
+
+        let mut volumes = Vec::new();
+
+        while let Some(entry) = entries.next_entry().await? {
+            if let Ok(metadata) = entry.metadata().await {
+                // if !metadata.is_dir() {
+                //     continue;
+                // }
+
+                let name = entry.file_name().to_string_lossy().to_string();
+
+                let created = match metadata.created() {
+                    Ok(md) => OffsetDateTime::from(md),
+                    Err(_) => return Err(Error::msg("Not supported created on this platform")),
+                };
+
+                volumes.push(VolumeInfo { name, created });
+            }
+        }
+
+        Ok(volumes)
+    }
+    async fn stat_volume(&self, volume: &str) -> Result<VolumeInfo> {
+        let p = self.get_bucket_path(volume)?;
+
+        let m = read_file_metadata(&p).await?;
+        let modtime = match m.modified() {
+            Ok(md) => OffsetDateTime::from(md),
+            Err(_) => return Err(Error::msg("Not supported modified on this platform")),
+        };
+
+        Ok(VolumeInfo {
+            name: volume.to_string(),
+            created: modtime,
+        })
+    }
+
+    async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> {
+        let p = self.get_object_path(&volume, format!("{}/{}", path, super::STORAGE_FORMAT_FILE).as_str())?;
+
+        warn!("write_metadata {:?} {:?}", &p, &fi);
+
+        let mut meta = FileMeta::new();
+        if !fi.fresh {
+            let (buf, _) = read_file_exists(&p).await?;
+            if !buf.is_empty() {
+                meta = FileMeta::unmarshal(&buf)?;
+            }
+        }
+
+        meta.add_version(fi)?;
+
+        let fm_data = meta.marshal_msg()?;
+
+        write_all_internal(p, fm_data).await?;
+
+        return Ok(());
+    }
+
+    async fn read_version(
+        &self,
+        _org_volume: &str,
+        volume: &str,
+        path: &str,
+        version_id: &str,
+        opts: &ReadOptions,
+    ) -> Result<FileInfo> {
+        let file_path = self.get_object_path(volume, path)?;
+        let file_dir = self.get_bucket_path(volume)?;
+
+        let read_data = opts.read_data;
+
+        let (data, _) = self.read_raw(volume, file_dir, file_path, read_data).await?;
+
+        let meta = FileMeta::unmarshal(&data)?;
+
+        let fi = meta.into_fileinfo(volume, path, version_id, false, true)?;
+        Ok(fi)
+    }
+    async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result<RawFileInfo> {
+        let file_path = self.get_object_path(volume, path)?;
+        let file_dir = self.get_bucket_path(volume)?;
+
+        let (buf, _) = self.read_raw(volume, file_dir, file_path, read_data).await?;
+
+        Ok(RawFileInfo { buf })
+    }
+    async fn read_multiple(&self, req: ReadMultipleReq) -> Result<Vec<ReadMultipleResp>> {
+        let mut results = Vec::new();
+        let mut found = 0;
+
+        for v in req.files.iter() {
+            let fpath = self.get_object_path(&req.bucket, format!("{}/{}", &req.prefix, v).as_str())?;
+            let mut res = ReadMultipleResp {
+                bucket: req.bucket.clone(),
+                prefix: req.prefix.clone(),
+                file: v.clone(),
+                ..Default::default()
+            };
+
+            // if req.metadata_only {}
+            match read_file_all(&fpath).await {
+                Ok((data, meta)) => {
+                    found += 1;
+
+                    if req.max_size > 0 && data.len() > req.max_size {
+                        res.exists = true;
+                        res.error = format!("max size ({}) exceeded: {}", req.max_size, data.len());
+                        results.push(res);
+                        break;
+                    }
+
+                    res.exists = true;
+                    res.data = data;
+                    res.mod_time = match meta.modified() {
+                        Ok(md) => OffsetDateTime::from(md),
+                        Err(_) => return Err(Error::msg("Not supported modified on this platform")),
+                    };
+                    results.push(res);
+
+                    if req.max_results > 0 && found >= req.max_results {
+                        break;
+                    }
+                }
+                Err(e) => {
+                    if !(DiskError::FileNotFound.is(&e) || DiskError::VolumeNotFound.is(&e)) {
+                        res.exists = true;
+                        res.error = e.to_string();
+                    }
+
+                    if req.abort404 && !res.exists {
+                        results.push(res);
+                        break;
+                    }
+
+                    results.push(res);
+                }
+            }
+        }
+
+        Ok(results)
+    }
+
+    async fn delete_volume(&self, volume: &str) -> Result<()> {
+        let p = self.get_bucket_path(&volume)?;
+
+        fs::remove_dir_all(&p).await?;
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod test {
+
+    use super::*;
+
+    #[tokio::test]
+    async fn test_skip_access_checks() {
+        // let arr = Vec::new();
+
+        let vols = vec![
+            super::super::RUSTFS_META_TMP_DELETED_BUCKET,
+            super::super::RUSTFS_META_TMP_BUCKET,
+            super::super::RUSTFS_META_MULTIPART_BUCKET,
+            super::super::RUSTFS_META_BUCKET,
+        ];
+
+        let paths: Vec<_> = vols.iter().map(|v| Path::new(v).join("test")).collect();
+
+        for p in paths.iter() {
+            assert!(skip_access_checks(p.to_str().unwrap()));
+        }
+    }
+
+    #[tokio::test]
+    async fn test_make_volume() {
+        let p = "./testv";
+        fs::create_dir_all(&p).await.unwrap();
+
+        let ep = match Endpoint::try_from(p) {
+            Ok(e) => e,
+            Err(e) => {
+                println!("{e}");
+                return;
+            }
+        };
+
+        let disk = LocalDisk::new(&ep, false).await.unwrap();
+
+        let tmpp = disk
+            .resolve_abs_path(Path::new(super::super::RUSTFS_META_TMP_DELETED_BUCKET))
+            .unwrap();
+
+        println!("ppp :{:?}", &tmpp);
+
+        let volumes = vec!["a", "b", "c"];
+
+        disk.make_volumes(volumes.clone()).await.unwrap();
+
+        disk.make_volumes(volumes.clone()).await.unwrap();
+
+        fs::remove_dir_all(&p).await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_delete_volume() {
+        let p = "./testv";
+        fs::create_dir_all(&p).await.unwrap();
+
+        let ep = match Endpoint::try_from(p) {
+            Ok(e) => e,
+            Err(e) => {
+                println!("{e}");
+                return;
+            }
+        };
+
+        let disk = LocalDisk::new(&ep, false).await.unwrap();
+
+        let tmpp = disk
+            .resolve_abs_path(Path::new(super::super::RUSTFS_META_TMP_DELETED_BUCKET))
+            .unwrap();
+
+        println!("ppp :{:?}", &tmpp);
+
+        let volumes = vec!["a", "b", "c"];
+
+        disk.make_volumes(volumes.clone()).await.unwrap();
+
+        disk.delete_volume("a").await.unwrap();
+
+        fs::remove_dir_all(&p).await.unwrap();
+    }
+}

ecstore/src/disk/mod.rs

@@ -0,0 +1,13 @@
+pub mod error;
+pub mod format;
+mod local;
+
+pub use local::{init_disks, DiskOption, DiskStore};
+
+pub const RUSTFS_META_BUCKET: &str = ".rustfs.sys";
+pub const RUSTFS_META_MULTIPART_BUCKET: &str = ".rustfs.sys/multipart";
+pub const RUSTFS_META_TMP_BUCKET: &str = ".rustfs.sys/tmp";
+pub const RUSTFS_META_TMP_DELETED_BUCKET: &str = ".rustfs.sys/tmp/.trash";
+pub const BUCKET_META_PREFIX: &str = "buckets";
+pub const FORMAT_CONFIG_FILE: &str = "format.json";
+const STORAGE_FORMAT_FILE: &str = "xl.meta";