add rename_part

ecstore/src/disk/error.rs

@@ -252,6 +252,142 @@ pub fn ioerr_to_err(e: io::Error) -> Error {
     }
 }
 
-pub fn os_is_not_exist(e: io::Error) -> bool {
+pub fn is_sys_err_no_space(e: &io::Error) -> bool {
+    if let Some(no) = e.raw_os_error() {
+        return no == 28;
+    }
+    false
+}
+
+pub fn is_sys_err_invalid_arg(e: &io::Error) -> bool {
+    if let Some(no) = e.raw_os_error() {
+        return no == 22;
+    }
+    false
+}
+
+pub fn is_sys_err_io(e: &io::Error) -> bool {
+    if let Some(no) = e.raw_os_error() {
+        return no == 5;
+    }
+    false
+}
+
+pub fn is_sys_err_is_dir(e: &io::Error) -> bool {
+    if let Some(no) = e.raw_os_error() {
+        return no == 21;
+    }
+    false
+}
+
+pub fn is_sys_err_not_dir(e: &io::Error) -> bool {
+    if let Some(no) = e.raw_os_error() {
+        return no == 20;
+    }
+    false
+}
+
+pub fn is_sys_err_too_long(e: &io::Error) -> bool {
+    if let Some(no) = e.raw_os_error() {
+        return no == 63;
+    }
+    false
+}
+
+pub fn is_sys_err_too_many_symlinks(e: &io::Error) -> bool {
+    if let Some(no) = e.raw_os_error() {
+        return no == 62;
+    }
+    false
+}
+
+pub fn is_sys_err_not_empty(e: &io::Error) -> bool {
+    if let Some(no) = e.raw_os_error() {
+        if no == 66 {
+            return true;
+        }
+
+        if cfg!(target_os = "solaris") && no == 17 {
+            return true;
+        }
+
+        if cfg!(target_os = "windows") && no == 145 {
+            return true;
+        }
+    }
+    false
+}
+
+pub fn is_sys_err_path_not_found(e: &io::Error) -> bool {
+    if let Some(no) = e.raw_os_error() {
+        if cfg!(target_os = "windows") {
+            if no == 3 {
+                return true;
+            }
+        } else {
+            if no == 2 {
+                return true;
+            }
+        }
+    }
+    false
+}
+
+pub fn is_sys_err_handle_invalid(e: &io::Error) -> bool {
+    if let Some(no) = e.raw_os_error() {
+        if cfg!(target_os = "windows") {
+            if no == 6 {
+                return true;
+            }
+        } else {
+            return false;
+        }
+    }
+    false
+}
+
+pub fn is_sys_err_cross_device(e: &io::Error) -> bool {
+    if let Some(no) = e.raw_os_error() {
+        return no == 18;
+    }
+    false
+}
+
+pub fn is_sys_err_too_many_files(e: &io::Error) -> bool {
+    if let Some(no) = e.raw_os_error() {
+        return no == 23 || no == 24;
+    }
+    false
+}
+
+pub fn os_is_not_exist(e: &io::Error) -> bool {
     e.kind() == ErrorKind::NotFound
 }
+
+pub fn os_is_permission(e: &io::Error) -> bool {
+    if e.kind() == ErrorKind::PermissionDenied {
+        return true;
+    }
+    if let Some(no) = e.raw_os_error() {
+        if no == 30 {
+            return true;
+        }
+    }
+
+    false
+}
+
+pub fn os_is_exist(e: &io::Error) -> bool {
+    e.kind() == ErrorKind::AlreadyExists
+}
+
+// map_err_not_exists
+pub fn map_err_not_exists(e: io::Error) -> Error {
+    if os_is_not_exist(&e) {
+        return Error::new(DiskError::VolumeNotEmpty);
+    } else if is_sys_err_io(&e) {
+        return Error::new(DiskError::FaultyDisk);
+    }
+
+    Error::new(e)
+}

ecstore/src/disk/local.rs

@@ -1,11 +1,14 @@
-use super::error::os_is_not_exist;
+use super::error::{is_sys_err_io, is_sys_err_not_empty, os_is_not_exist};
 use super::{endpoint::Endpoint, error::DiskError, format::FormatV3};
 use super::{
     os, DeleteOptions, DiskAPI, DiskLocation, FileInfoVersions, FileReader, FileWriter, MetaCacheEntry, ReadMultipleReq,
     ReadMultipleResp, ReadOptions, RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
 };
+use crate::disk::error::{is_sys_err_not_dir, map_err_not_exists};
+use crate::disk::os::check_path_length;
 use crate::disk::{LocalFileReader, LocalFileWriter, STORAGE_FORMAT_FILE};
-use crate::utils::path::SLASH_SEPARATOR;
+use crate::utils::fs::lstat;
+use crate::utils::path::{has_suffix, SLASH_SEPARATOR};
 use crate::{
     error::{Error, Result},
     file_meta::FileMeta,
@@ -22,6 +25,7 @@ use time::OffsetDateTime;
 use tokio::fs::{self, File};
 use tokio::io::ErrorKind;
 use tokio::sync::Mutex;
+use tower::layer::util;
 use tracing::{debug, warn};
 use uuid::Uuid;
 
@@ -620,15 +624,71 @@ impl DiskAPI for LocalDisk {
 
         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)?;
+    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)?;
         if !skip_access_checks(src_volume) {
-            check_volume_exists(&src_volume_path).await?;
+            utils::fs::access(&src_volume_dir).await.map_err(map_err_not_exists)?
         }
         if !skip_access_checks(dst_volume) {
-            let vol_path = self.get_bucket_path(dst_volume)?;
-            check_volume_exists(&vol_path).await?;
+            utils::fs::access(&dst_volume_dir).await.map_err(map_err_not_exists)?
+        }
+
+        let src_is_dir = has_suffix(&src_path, SLASH_SEPARATOR);
+        let dst_is_dir = has_suffix(&dst_path, SLASH_SEPARATOR);
+
+        if !(src_is_dir && dst_is_dir || !src_is_dir && !dst_is_dir) {
+            return Err(Error::from(DiskError::FileAccessDenied));
+        }
+
+        let src_file_path = src_volume_dir.join(Path::new(src_path));
+        let dst_file_path = dst_volume_dir.join(Path::new(dst_path));
+
+        check_path_length(&src_file_path.to_string_lossy().to_string())?;
+        check_path_length(&dst_file_path.to_string_lossy().to_string())?;
+
+        if src_is_dir {
+            let meta_op = match lstat(&src_file_path).await {
+                Ok(meta) => Some(meta),
+                Err(e) => {
+                    if is_sys_err_io(&e) {
+                        return Err(Error::new(DiskError::FaultyDisk));
+                    }
+
+                    if !os_is_not_exist(&e) {
+                        return Err(Error::new(e));
+                    }
+                    None
+                }
+            };
+
+            if let Some(meta) = meta_op {
+                if !meta.is_dir() {
+                    return Err(Error::new(DiskError::FileAccessDenied));
+                }
+            }
+
+            if let Err(e) = utils::fs::remove(&dst_file_path).await {
+                if is_sys_err_not_empty(&e) || is_sys_err_not_dir(&e) {
+                    return Err(Error::new(DiskError::FileAccessDenied));
+                } else if is_sys_err_io(&e) {
+                    return Err(Error::new(DiskError::FaultyDisk));
+                }
+
+                return Err(Error::new(e));
+            }
+        }
+
+        unimplemented!()
+    }
+    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)?;
+        let dst_volume_path = self.get_bucket_path(dst_volume)?;
+        if !skip_access_checks(src_volume) {
+            utils::fs::access(&src_volume_path).await.map_err(map_err_not_exists)?;
+        }
+        if !skip_access_checks(dst_volume) {
+            utils::fs::access(&dst_volume_path).await.map_err(map_err_not_exists)?;
         }
 
         let srcp = self.get_object_path(src_volume, src_path)?;

ecstore/src/disk/mod.rs

@@ -2,8 +2,8 @@ pub mod endpoint;
 pub mod error;
 pub mod format;
 mod local;
-mod remote;
 pub mod os;
+mod remote;
 
 pub const RUSTFS_META_BUCKET: &str = ".rustfs.sys";
 pub const RUSTFS_META_MULTIPART_BUCKET: &str = ".rustfs.sys/multipart";
@@ -117,7 +117,7 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
     async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: usize) -> Result<FileWriter>;
     // ReadFileStream
     async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()>;
-    // RenamePart
+    async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Vec<u8>) -> Result<()>;
     // CheckParts
     async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()>;
     // VerifyFile
@@ -127,7 +127,6 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
     // CleanAbandonedData
     async fn write_all(&self, volume: &str, path: &str, data: Vec<u8>) -> Result<()>;
     async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes>;
-    
 }
 
 pub struct UpdateMetadataOpts {

ecstore/src/disk/os.rs

@@ -1,16 +1,20 @@
-use std::path::Path;
+use std::{
+    io,
+    path::{Component, Path},
+};
 
 use futures::TryFutureExt;
 use tokio::fs;
 
 use crate::{
+    disk::error::{is_sys_err_not_dir, is_sys_err_path_not_found, os_is_not_exist},
     error::{Error, Result},
     utils,
 };
 
-use super::error::{ioerr_to_err, DiskError};
+use super::error::{ioerr_to_err, os_is_exist, DiskError};
 
-fn check_path_length(path_name: &str) -> Result<()> {
+pub fn check_path_length(path_name: &str) -> Result<()> {
     // Apple OS X path length is limited to 1016
     if cfg!(target_os = "macos") && path_name.len() > 1016 {
         return Err(Error::new(DiskError::FileNameTooLong));
@@ -91,3 +95,110 @@ pub async fn read_dir(path: impl AsRef<Path>, count: usize) -> Result<Vec<String
 
     Ok(volumes)
 }
+
+pub async fn rename_all(
+    src_file_path: impl AsRef<Path>,
+    dst_file_path: impl AsRef<Path>,
+    base_dir: impl AsRef<Path>,
+) -> Result<()> {
+    reliable_rename(src_file_path, dst_file_path, base_dir).await.map_err(|e| {
+        if is_sys_err_not_dir(&e) || !os_is_not_exist(&e) {
+            Error::new(DiskError::FileAccessDenied)
+        } else if is_sys_err_path_not_found(&e) {
+            Error::new(DiskError::FileAccessDenied)
+        } else if os_is_not_exist(&e) {
+            Error::new(DiskError::FileNotFound)
+        } else if os_is_exist(&e) {
+            Error::new(DiskError::IsNotRegular)
+        } else {
+            Error::new(e)
+        }
+    })?;
+
+    Ok(())
+}
+
+pub async fn reliable_rename(
+    src_file_path: impl AsRef<Path>,
+    dst_file_path: impl AsRef<Path>,
+    base_dir: impl AsRef<Path>,
+) -> io::Result<()> {
+    if let Some(parent) = dst_file_path.as_ref().parent() {
+        reliable_mkdir_all(parent, base_dir.as_ref()).await?;
+    }
+
+    let mut i = 0;
+    loop {
+        if let Err(e) = utils::fs::rename(src_file_path.as_ref(), dst_file_path.as_ref()).await {
+            if os_is_not_exist(&e) && i == 0 {
+                i += 1;
+                continue;
+            }
+
+            return Err(e);
+        }
+
+        break;
+    }
+
+    Ok(())
+}
+
+pub async fn reliable_mkdir_all(path: impl AsRef<Path>, base_dir: impl AsRef<Path>) -> io::Result<()> {
+    let mut i = 0;
+
+    let mut base_dir = base_dir.as_ref().clone();
+    loop {
+        if let Err(e) = os_mkdir_all(path.as_ref(), base_dir).await {
+            if os_is_not_exist(&e) && i == 0 {
+                i += 1;
+
+                if let Some(base_parent) = base_dir.parent() {
+                    if let Some(c) = base_parent.components().next() {
+                        if c != Component::RootDir {
+                            base_dir = base_parent
+                        }
+                    }
+                }
+                continue;
+            }
+
+            return Err(e);
+        }
+
+        break;
+    }
+
+    Ok(())
+}
+
+pub async fn os_mkdir_all(dir_path: impl AsRef<Path>, base_dir: impl AsRef<Path>) -> io::Result<()> {
+    if !base_dir.as_ref().to_string_lossy().is_empty() {
+        if base_dir.as_ref().starts_with(dir_path.as_ref()) {
+            return Ok(());
+        }
+    }
+
+    if let Some(parent) = dir_path.as_ref().parent() {
+        Box::pin(os_mkdir_all(parent, base_dir)).await?;
+    }
+
+    if let Err(e) = utils::fs::mkdir(dir_path.as_ref()).await {
+        if os_is_exist(&e) {
+            return Ok(());
+        }
+
+        return Err(e);
+    }
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod test {
+
+    use super::*;
+
+    #[tokio::test]
+    async fn test_make_dir() {}
+}

ecstore/src/disk/remote.rs

@@ -201,7 +201,16 @@ impl DiskAPI for RemoteDisk {
 
         Ok(())
     }
-
+    async fn rename_part(
+        &self,
+        _src_volume: &str,
+        _src_path: &str,
+        _dst_volume: &str,
+        _dst_path: &str,
+        _meta: Vec<u8>,
+    ) -> Result<()> {
+        unimplemented!()
+    }
     async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()> {
         info!("rename_file");
         let mut client = self.get_client_v2().await?;

ecstore/src/utils/fs.rs

@@ -49,6 +49,27 @@ pub async fn access(path: impl AsRef<Path>) -> io::Result<()> {
     Ok(())
 }
 
+pub async fn lstat(path: impl AsRef<Path>) -> io::Result<Metadata> {
+    fs::metadata(path).await
+}
+
 pub async fn make_dir_all(path: impl AsRef<Path>) -> io::Result<()> {
     fs::create_dir_all(path.as_ref()).await
 }
+
+pub async fn remove(path: impl AsRef<Path>) -> io::Result<()> {
+    let meta = fs::metadata(path.as_ref()).await?;
+    if meta.is_dir() {
+        fs::remove_dir(path.as_ref()).await
+    } else {
+        fs::remove_file(path.as_ref()).await
+    }
+}
+
+pub async fn mkdir(path: impl AsRef<Path>) -> io::Result<()> {
+    fs::create_dir(path.as_ref()).await
+}
+
+pub async fn rename(from: impl AsRef<Path>, to: impl AsRef<Path>) -> io::Result<()> {
+    fs::rename(from, to).await
+}

ecstore/src/utils/path.rs

@@ -1,3 +1,5 @@
+use std::path::Path;
+
 const GLOBAL_DIR_SUFFIX: &str = "__XLDIR__";
 
 pub const SLASH_SEPARATOR: &str = "/";
@@ -37,3 +39,7 @@ pub fn retain_slash(s: &str) -> String {
         format!("{}{}", s, SLASH_SEPARATOR)
     }
 }
+
+pub fn join(p1: &str, p2: &str) -> String {
+    Path::new(p1).join(Path::new(p2)).to_string_lossy().to_string()
+}