From ad4a32762dec27690c22c6650dcf1ecf8154cb35 Mon Sep 17 00:00:00 2001 From: weisd Date: Wed, 25 Sep 2024 11:49:12 +0800 Subject: [PATCH] add rename_part --- ecstore/src/disk/error.rs | 138 ++++++++++++++++++++++++++++++++++++- ecstore/src/disk/local.rs | 76 +++++++++++++++++--- ecstore/src/disk/mod.rs | 5 +- ecstore/src/disk/os.rs | 117 ++++++++++++++++++++++++++++++- ecstore/src/disk/remote.rs | 11 ++- ecstore/src/utils/fs.rs | 21 ++++++ ecstore/src/utils/path.rs | 6 ++ 7 files changed, 358 insertions(+), 16 deletions(-) diff --git a/ecstore/src/disk/error.rs b/ecstore/src/disk/error.rs index 24864213e..145b4d195 100644 --- a/ecstore/src/disk/error.rs +++ b/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) +} diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index 9a85464dc..768e3ac75 100644 --- a/ecstore/src/disk/local.rs +++ b/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) -> 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)?; diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs index 3ec7f52df..cd8e9af7f 100644 --- a/ecstore/src/disk/mod.rs +++ b/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; // 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) -> 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) -> Result<()>; async fn read_all(&self, volume: &str, path: &str) -> Result; - } pub struct UpdateMetadataOpts { diff --git a/ecstore/src/disk/os.rs b/ecstore/src/disk/os.rs index 016e5e3c6..e9a100bb6 100644 --- a/ecstore/src/disk/os.rs +++ b/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, count: usize) -> Result, + dst_file_path: impl AsRef, + base_dir: impl AsRef, +) -> 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, + dst_file_path: impl AsRef, + base_dir: impl AsRef, +) -> 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, base_dir: impl AsRef) -> 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, base_dir: impl AsRef) -> 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() {} +} diff --git a/ecstore/src/disk/remote.rs b/ecstore/src/disk/remote.rs index b3360f13d..51763c4ae 100644 --- a/ecstore/src/disk/remote.rs +++ b/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, + ) -> 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?; diff --git a/ecstore/src/utils/fs.rs b/ecstore/src/utils/fs.rs index 41c7e4867..0ff3b320a 100644 --- a/ecstore/src/utils/fs.rs +++ b/ecstore/src/utils/fs.rs @@ -49,6 +49,27 @@ pub async fn access(path: impl AsRef) -> io::Result<()> { Ok(()) } +pub async fn lstat(path: impl AsRef) -> io::Result { + fs::metadata(path).await +} + pub async fn make_dir_all(path: impl AsRef) -> io::Result<()> { fs::create_dir_all(path.as_ref()).await } + +pub async fn remove(path: impl AsRef) -> 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) -> io::Result<()> { + fs::create_dir(path.as_ref()).await +} + +pub async fn rename(from: impl AsRef, to: impl AsRef) -> io::Result<()> { + fs::rename(from, to).await +} diff --git a/ecstore/src/utils/path.rs b/ecstore/src/utils/path.rs index 79bfc6ced..49a7dc9e4 100644 --- a/ecstore/src/utils/path.rs +++ b/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() +}