diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index b758b9e9e..80f2344ce 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -1,13 +1,16 @@ -use super::error::{is_sys_err_io, is_sys_err_not_empty, os_is_not_exist}; +use super::error::{is_sys_err_io, is_sys_err_not_empty, is_sys_err_too_many_files, os_is_not_exist, os_is_permission}; 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::{convert_access_error, is_sys_err_not_dir, map_err_not_exists, os_err_to_file_err}; +use crate::disk::error::{ + convert_access_error, is_sys_err_handle_invalid, is_sys_err_invalid_arg, is_sys_err_is_dir, is_sys_err_not_dir, + map_err_not_exists, os_err_to_file_err, +}; use crate::disk::os::check_path_length; use crate::disk::{LocalFileReader, LocalFileWriter, STORAGE_FORMAT_FILE}; -use crate::utils::fs::lstat; +use crate::utils::fs::{lstat, O_RDONLY}; use crate::utils::path::{has_suffix, SLASH_SEPARATOR}; use crate::{ error::{Error, Result}, @@ -22,9 +25,9 @@ use std::{ fs::Metadata, path::{Path, PathBuf}, }; -use time::OffsetDateTime; +use time::{util, OffsetDateTime}; use tokio::fs::{self, File}; -use tokio::io::ErrorKind; +use tokio::io::{self, AsyncReadExt, AsyncWriteExt, ErrorKind}; use tokio::sync::Mutex; use tracing::{debug, warn}; use uuid::Uuid; @@ -117,10 +120,6 @@ impl LocalDisk { Ok(disk) } - // fn check_path_length(_path_name: &str) -> Result<()> { - // unimplemented!() - // } - fn is_valid_volname(volname: &str) -> bool { if volname.len() < 3 { return false; @@ -324,54 +323,108 @@ impl LocalDisk { volume_dir: impl AsRef, path: impl AsRef, read_data: bool, - ) -> Result<(Vec, OffsetDateTime)> { + ) -> Result<(Vec, Option)> { 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 + self.read_all_data_with_dmtime(bucket, volume_dir, meta_path).await } else { - self.read_metadata_with_dmtime(meta_path).await + self.read_all_data_with_dmtime(bucket, volume_dir, meta_path).await + // FIXME: read_metadata only suport + // self.read_metadata_with_dmtime(meta_path).await } } - async fn read_metadata_with_dmtime(&self, path: impl AsRef) -> Result<(Vec, OffsetDateTime)> { - let (data, meta) = read_file_all(path).await?; + async fn read_metadata_with_dmtime(&self, file_path: impl AsRef) -> Result<(Vec, Option)> { + check_path_length(&file_path.as_ref().to_string_lossy().to_string())?; - let modtime = match meta.modified() { - Ok(md) => OffsetDateTime::from(md), - Err(_) => return Err(Error::msg("Not supported modified on this platform")), - }; + let f = utils::fs::open_file(file_path, O_RDONLY).await?; - Ok((data, modtime)) + let meta = f.metadata().await?; + + if meta.is_dir() { + return Err(Error::new(DiskError::FileNotFound)); + } + + // FIXME: + unimplemented!() } - async fn read_all_data( - &self, - _bucket: &str, - _volume_dir: impl AsRef, - path: impl AsRef, - ) -> Result<(Vec, OffsetDateTime)> { - let (data, meta) = read_file_all(path).await?; + async fn read_all_data(&self, volume: &str, volume_dir: impl AsRef, file_path: impl AsRef) -> Result> { + // TODO: timeout suport + let (data, _) = self.read_all_data_with_dmtime(volume, volume_dir, file_path).await?; + Ok(data) + } - let modtime = match meta.modified() { - Ok(md) => OffsetDateTime::from(md), - Err(_) => return Err(Error::msg("Not supported modified on this platform")), + async fn read_all_data_with_dmtime( + &self, + volume: &str, + volume_dir: impl AsRef, + file_path: impl AsRef, + ) -> Result<(Vec, Option)> { + let mut f = match utils::fs::open_file(file_path.as_ref(), utils::fs::O_RDONLY).await { + Ok(f) => f, + Err(e) => { + if os_is_not_exist(&e) { + if !skip_access_checks(volume) { + if let Err(er) = utils::fs::access(volume_dir.as_ref()).await { + if os_is_not_exist(&er) { + return Err(Error::new(DiskError::VolumeNotFound)); + } + } + } + + return Err(Error::new(DiskError::FileNotFound)); + } else if os_is_permission(&e) { + return Err(Error::new(DiskError::FileAccessDenied)); + } else if is_sys_err_not_dir(&e) || is_sys_err_is_dir(&e) { + return Err(Error::new(DiskError::FileNotFound)); + } else if is_sys_err_handle_invalid(&e) { + return Err(Error::new(DiskError::FileNotFound)); + } else if is_sys_err_io(&e) { + return Err(Error::new(DiskError::FaultyDisk)); + } else if is_sys_err_too_many_files(&e) { + return Err(Error::new(DiskError::TooManyOpenFiles)); + } else if is_sys_err_invalid_arg(&e) { + if let Ok(meta) = utils::fs::lstat(file_path.as_ref()).await { + if meta.is_dir() { + return Err(Error::new(DiskError::FileNotFound)); + } + } + return Err(Error::new(DiskError::UnsupportedDisk)); + } + + return Err(Error::new(e)); + } }; - Ok((data, modtime)) + let meta = f.metadata().await.map_err(os_err_to_file_err)?; + + if meta.is_dir() { + return Err(Error::new(DiskError::FileNotFound)); + } + + let size = meta.len() as usize; + let mut bytes = Vec::new(); + bytes.try_reserve_exact(size)?; + + f.read_to_end(&mut bytes).await.map_err(os_err_to_file_err)?; + + let modtime = match meta.modified() { + Ok(md) => Some(OffsetDateTime::from(md)), + Err(_) => None, + }; + + Ok((bytes, modtime)) } async fn delete_versions_internal(&self, volume: &str, path: &str, fis: &Vec) -> Result<()> { let volume_dir = self.get_bucket_path(volume)?; let xlpath = self.get_object_path(volume, format!("{}/{}", path, super::STORAGE_FORMAT_FILE).as_str())?; - let (data, _) = match self.read_all_data(volume, volume_dir.as_path(), &xlpath).await { - Ok(res) => res, - Err(_err) => { - // TODO: check if not found return err - - (Vec::new(), OffsetDateTime::UNIX_EPOCH) - } - }; + let data = self + .read_all_data(volume, volume_dir.as_path(), &xlpath) + .await + .unwrap_or_default(); if data.is_empty() { return Err(Error::new(DiskError::FileNotFound)); @@ -401,8 +454,16 @@ impl LocalDisk { // 更新xl.meta let buf = fm.marshal_msg()?; - self.write_all(volume, format!("{}/{}", path, super::STORAGE_FORMAT_FILE).as_str(), buf) - .await?; + let volume_dir = self.get_bucket_path(&volume)?; + + self.write_all_private( + volume, + format!("{}/{}", path, super::STORAGE_FORMAT_FILE).as_str(), + &buf, + true, + volume_dir, + ) + .await?; Ok(()) } @@ -420,7 +481,21 @@ impl LocalDisk { os::rename_all(tmp_file_path, file_path, volume_dir).await } - // write_all_private + // write_all_public for trail + async fn write_all_public(&self, volume: &str, path: &str, data: Vec) -> Result<()> { + if volume == super::RUSTFS_META_BUCKET && path == super::FORMAT_CONFIG_FILE { + let mut format_info = self.format_info.lock().await; + format_info.data = data.clone(); + } + + let volume_dir = self.get_bucket_path(&volume)?; + + self.write_all_private(&volume, &path, &data, true, volume_dir).await?; + + Ok(()) + } + + // write_all_private with check_path_length pub async fn write_all_private( &self, volume: &str, @@ -435,21 +510,27 @@ impl LocalDisk { self.write_all_internal(file_path, buf, sync, skip_parent).await } - + // write_all_internal do write file pub async fn write_all_internal( &self, - p: impl AsRef, + file_path: impl AsRef, data: impl AsRef<[u8]>, sync: bool, - base_dir: impl AsRef, + skip_parent: impl AsRef, ) -> Result<()> { - if sync { - } else { - } - // create top dir if not exists - fs::create_dir_all(&p.as_ref().parent().unwrap_or_else(|| Path::new("."))).await?; + let flags = utils::fs::O_CREATE | utils::fs::O_WRONLY | utils::fs::O_TRUNC; + + let mut f = { + if sync { + // TODO: suport sync + self.open_file(file_path.as_ref(), flags, skip_parent.as_ref()).await? + } else { + self.open_file(file_path.as_ref(), flags, skip_parent.as_ref()).await? + } + }; + + f.write_all(data.as_ref()).await?; - fs::write(&p, data).await?; Ok(()) } @@ -463,7 +544,23 @@ impl LocalDisk { os::make_dir_all(parent, skip_parent).await?; } - unimplemented!() + let f = utils::fs::open_file(path.as_ref(), mode).await.map_err(|e| { + if is_sys_err_io(&e) { + Error::new(DiskError::IsNotRegular) + } else if os_is_permission(&e) { + Error::new(DiskError::FileAccessDenied) + } else if is_sys_err_not_dir(&e) { + Error::new(DiskError::FileAccessDenied) + } else if is_sys_err_io(&e) { + Error::new(DiskError::FaultyDisk) + } else if is_sys_err_too_many_files(&e) { + Error::new(DiskError::TooManyOpenFiles) + } else { + Error::new(e) + } + })?; + + Ok(f) } } @@ -645,18 +742,7 @@ impl DiskAPI for LocalDisk { } async fn write_all(&self, volume: &str, path: &str, data: Vec) -> Result<()> { - if volume == super::RUSTFS_META_BUCKET && path == super::FORMAT_CONFIG_FILE { - let mut format_info = self.format_info.lock().await; - format_info.data = data.clone(); - } - - let volume_dir = self.get_bucket_path(&volume)?; - let file_path = volume_dir.join(Path::new(&path)); - check_path_length(&file_path.to_string_lossy().to_string())?; - - self.write_all_internal(file_path, data, true, volume_dir).await?; - - Ok(()) + self.write_all_public(volume, path, data).await } async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> { @@ -947,10 +1033,7 @@ impl DiskAPI for LocalDisk { let (fdata, _) = match self.read_metadata_with_dmtime(&fpath).await { Ok(res) => res, - Err(_) => { - // TODO: check err - (Vec::new(), OffsetDateTime::UNIX_EPOCH) - } + Err(_) => (Vec::new(), None), }; meta.metadata = fdata; @@ -1259,12 +1342,16 @@ impl DiskAPI for LocalDisk { } async fn delete_version( &self, - _volume: &str, - _path: &str, - _fi: FileInfo, - _force_del_marker: bool, - _opts: DeleteOptions, + volume: &str, + path: &str, + fi: FileInfo, + force_del_marker: bool, + opts: DeleteOptions, ) -> Result { + let volume_dir = self.get_bucket_path(&volume)?; + + // self.read_all_data(bucket, volume_dir, path); + unimplemented!() } async fn delete_versions( diff --git a/ecstore/src/utils/fs.rs b/ecstore/src/utils/fs.rs index 9d9f54c2b..afc13c5a3 100644 --- a/ecstore/src/utils/fs.rs +++ b/ecstore/src/utils/fs.rs @@ -49,18 +49,25 @@ pub fn same_file(f1: &Metadata, f2: &Metadata) -> bool { type FileMode = usize; -const O_RDONLY: FileMode = 0x00000; -const O_WRONLY: FileMode = 0x00001; -const O_RDWR: FileMode = 0x00002; -const O_CREAT: FileMode = 0x00040; -const O_EXCL: FileMode = 0x00080; -const O_NOCTTY: FileMode = 0x00100; -const O_TRUNC: FileMode = 0x00200; -const O_NONBLOCK: FileMode = 0x00800; -const O_APPEND: FileMode = 0x00400; -const O_SYNC: FileMode = 0x01000; -const O_ASYNC: FileMode = 0x02000; -const O_CLOEXEC: FileMode = 0x80000; +pub const O_RDONLY: FileMode = 0x00000; +pub const O_WRONLY: FileMode = 0x00001; +pub const O_RDWR: FileMode = 0x00002; +pub const O_CREATE: FileMode = 0x00040; +// pub const O_EXCL: FileMode = 0x00080; +// pub const O_NOCTTY: FileMode = 0x00100; +pub const O_TRUNC: FileMode = 0x00200; +// pub const O_NONBLOCK: FileMode = 0x00800; +pub const O_APPEND: FileMode = 0x00400; +// pub const O_SYNC: FileMode = 0x01000; +// pub const O_ASYNC: FileMode = 0x02000; +// pub const O_CLOEXEC: FileMode = 0x80000; + +// read: bool, +// write: bool, +// append: bool, +// truncate: bool, +// create: bool, +// create_new: bool, pub async fn open_file(path: impl AsRef, mode: FileMode) -> io::Result { let mut opts = fs::OpenOptions::new(); @@ -79,15 +86,17 @@ pub async fn open_file(path: impl AsRef, mode: FileMode) -> io::Result (), }; - if mode & O_CREAT != 0 { - opts.write(true); + if mode & O_CREATE != 0 { + opts.create(true); } if mode & O_APPEND != 0 { - opts.write(true); + opts.append(true); } - // FIXME: TODO + if mode & O_TRUNC != 0 { + opts.truncate(true); + } opts.open(path.as_ref()).await }