diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index d0949c30f..fced05f36 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -17,6 +17,7 @@ use crate::{ }; use bytes::Bytes; use path_absolutize::Absolutize; +use std::io::Read; use std::{ fs::Metadata, path::{Path, PathBuf}, @@ -405,6 +406,65 @@ impl LocalDisk { Ok(()) } + + async fn write_all_meta(&self, volume: &str, path: &str, buf: &[u8], sync: bool) -> Result<()> { + 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())?; + + let tmp_volume_dir = self.get_bucket_path(super::RUSTFS_META_TMP_BUCKET)?; + let tmp_file_path = tmp_volume_dir.join(Path::new(Uuid::new_v4().to_string().as_str())); + + self.write_all_internal(&tmp_file_path, buf, sync, tmp_volume_dir).await?; + + os::rename_all(tmp_file_path, file_path, volume_dir).await + } + + // write_all_private + pub async fn write_all_private( + &self, + volume: &str, + path: &str, + buf: &[u8], + sync: bool, + skip_parent: impl AsRef, + ) -> Result<()> { + 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, buf, sync, skip_parent).await + } + + pub async fn write_all_internal( + &self, + p: impl AsRef, + data: impl AsRef<[u8]>, + sync: bool, + base_dir: 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?; + + fs::write(&p, data).await?; + Ok(()) + } + + async fn open_file(&self, path: impl AsRef, mode: usize, skip_parent: impl AsRef) -> Result { + let mut skip_parent = skip_parent.as_ref(); + if skip_parent.as_os_str().is_empty() { + skip_parent = self.root.as_path(); + } + + if let Some(parent) = path.as_ref().parent() { + os::make_dir_all(parent, skip_parent).await?; + } + + unimplemented!() + } } fn is_root_path(path: impl AsRef) -> bool { @@ -433,14 +493,6 @@ pub async fn read_file_exists(path: impl AsRef) -> Result<(Vec, Option Ok((data, meta)) } -pub async fn write_all_internal(p: impl AsRef, 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) -> Result<(Vec, Metadata)> { let p = path.as_ref(); let meta = read_file_metadata(&path).await?; @@ -584,18 +636,25 @@ impl DiskAPI for LocalDisk { } #[must_use] - async fn read_all(&self, volume: &str, path: &str) -> Result { + async fn read_all(&self, volume: &str, path: &str) -> Result> { // TOFIX: let p = self.get_object_path(volume, path)?; let (data, _) = read_file_all(&p).await?; - Ok(Bytes::from(data)) + Ok(data) } async fn write_all(&self, volume: &str, path: &str, data: Vec) -> Result<()> { - let p = self.get_object_path(volume, path)?; + 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(); + } - write_all_internal(p, data).await?; + 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(()) } @@ -978,7 +1037,8 @@ impl DiskAPI for LocalDisk { let fm_data = meta.marshal_msg()?; // 写入xl.meta - write_all_internal(&src_file_path, fm_data).await?; + self.write_all(src_volume, format!("{}/{}", &src_path, super::STORAGE_FORMAT_FILE).as_str(), fm_data) + .await?; let no_inline = src_data_path.has_root() && fi.data.is_none() && fi.size > 0; if no_inline { @@ -1035,7 +1095,7 @@ impl DiskAPI for LocalDisk { if let Err(e) = utils::fs::access(&p).await { if os_is_not_exist(&e) { - os::make_dir_all(&p).await?; + os::make_dir_all(&p, self.root.as_path()).await?; } } @@ -1102,14 +1162,15 @@ impl DiskAPI for LocalDisk { Ok(()) } - async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, _opts: UpdateMetadataOpts) -> Result<()> { - if let Some(metadata) = fi.metadata { + async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: UpdateMetadataOpts) -> Result<()> { + if let Some(metadata) = &fi.metadata { 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.read_all(&volume, format!("{}/{}", &path, super::STORAGE_FORMAT_FILE).as_str()) + let buf = self + .read_all(&volume, format!("{}/{}", &path, super::STORAGE_FORMAT_FILE).as_str()) .await .map_err(|e| { if DiskError::FileNotFound.is(&e) && fi.version_id.is_some() { @@ -1119,6 +1180,25 @@ impl DiskAPI for LocalDisk { } })?; + if !FileMeta::is_xl_format(buf.as_slice()) { + return Err(Error::new(DiskError::FileVersionNotFound)); + } + + let xl_meta = FileMeta::load(buf.as_slice())?; + + xl_meta.update_object_version(fi)?; + + let wbuf = xl_meta.marshal_msg()?; + + return self + .write_all_meta( + &volume, + format!("{}/{}", path, super::STORAGE_FORMAT_FILE).as_str(), + &wbuf, + !opts.no_persistence, + ) + .await; + // FIXME: } @@ -1144,7 +1224,8 @@ impl DiskAPI for LocalDisk { let fm_data = meta.marshal_msg()?; - write_all_internal(p, fm_data).await?; + self.write_all(volume, format!("{}/{}", path, super::STORAGE_FORMAT_FILE).as_str(), fm_data) + .await?; return Ok(()); } diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs index a59d720d5..9d12f6e65 100644 --- a/ecstore/src/disk/mod.rs +++ b/ecstore/src/disk/mod.rs @@ -126,7 +126,7 @@ pub trait DiskAPI: Debug + Send + Sync + 'static { async fn read_multiple(&self, req: ReadMultipleReq) -> Result>; // CleanAbandonedData async fn write_all(&self, volume: &str, path: &str, data: Vec) -> Result<()>; - async fn read_all(&self, volume: &str, path: &str) -> 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 3bf4d10e4..680428cb8 100644 --- a/ecstore/src/disk/os.rs +++ b/ecstore/src/disk/os.rs @@ -52,10 +52,18 @@ pub fn check_path_length(path_name: &str) -> Result<()> { Ok(()) } -pub async fn make_dir_all(path: impl AsRef) -> Result<()> { +pub async fn make_dir_all(path: impl AsRef, base_dir: impl AsRef) -> Result<()> { check_path_length(path.as_ref().to_string_lossy().to_string().as_str())?; - utils::fs::make_dir_all(path.as_ref()).map_err(os_err_to_file_err).await?; + if let Err(e) = reliable_mkdir_all(path.as_ref(), base_dir.as_ref()).await { + if is_sys_err_not_dir(&e) { + return Err(Error::new(DiskError::FileAccessDenied)); + } else if is_sys_err_path_not_found(&e) { + return Err(Error::new(DiskError::FileAccessDenied)); + } + + return Err(os_err_to_file_err(e)); + } Ok(()) } @@ -181,7 +189,7 @@ pub async fn os_mkdir_all(dir_path: impl AsRef, base_dir: impl AsRef if let Some(parent) = dir_path.as_ref().parent() { // 不支持递归，直接create_dir_all了 - if let Err(e) = fs::create_dir_all(&parent).await { + if let Err(e) = utils::fs::make_dir_all(&parent).await { if os_is_exist(&e) { return Ok(()); } diff --git a/ecstore/src/disk/remote.rs b/ecstore/src/disk/remote.rs index e035ad9c5..7cd11d9e6 100644 --- a/ecstore/src/disk/remote.rs +++ b/ecstore/src/disk/remote.rs @@ -143,7 +143,7 @@ impl DiskAPI for RemoteDisk { Ok(()) } - async fn read_all(&self, volume: &str, path: &str) -> Result { + async fn read_all(&self, volume: &str, path: &str) -> Result> { info!("read_all"); let mut client = self.get_client_v2().await?; let request = Request::new(ReadAllRequest { @@ -160,7 +160,7 @@ impl DiskAPI for RemoteDisk { return Err(DiskError::FileNotFound.into()); } - Ok(Bytes::from(response.data)) + Ok(response.data) } async fn write_all(&self, volume: &str, path: &str, data: Vec) -> Result<()> { diff --git a/ecstore/src/store_init.rs b/ecstore/src/store_init.rs index 941aa4adf..ebd28c881 100644 --- a/ecstore/src/store_init.rs +++ b/ecstore/src/store_init.rs @@ -232,7 +232,7 @@ pub async fn load_format_erasure(disk: &DiskStore, _heal: bool) -> Result e, })?; - let fm = FormatV3::try_from(data.as_ref())?; + let fm = FormatV3::try_from(data.as_slice())?; // TODO: heal diff --git a/ecstore/src/utils/fs.rs b/ecstore/src/utils/fs.rs index 0ff3b320a..9d9f54c2b 100644 --- a/ecstore/src/utils/fs.rs +++ b/ecstore/src/utils/fs.rs @@ -1,6 +1,9 @@ use std::{fs::Metadata, path::Path}; -use tokio::{fs, io}; +use tokio::{ + fs::{self, File}, + io, +}; #[cfg(not(windows))] pub fn same_file(f1: &Metadata, f2: &Metadata) -> bool { @@ -44,6 +47,51 @@ pub fn same_file(f1: &Metadata, f2: &Metadata) -> bool { true } +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 async fn open_file(path: impl AsRef, mode: FileMode) -> io::Result { + let mut opts = fs::OpenOptions::new(); + + match mode & (O_RDONLY | O_WRONLY | O_RDWR) { + O_RDONLY => { + opts.read(true); + } + O_WRONLY => { + opts.write(true); + } + O_RDWR => { + opts.read(true); + opts.write(true); + } + _ => (), + }; + + if mode & O_CREAT != 0 { + opts.write(true); + } + + if mode & O_APPEND != 0 { + opts.write(true); + } + + // FIXME: TODO + + opts.open(path.as_ref()).await +} + pub async fn access(path: impl AsRef) -> io::Result<()> { fs::metadata(path).await?; Ok(())