rewrite real_meta

ecstore/src/disk/local.rs

@@ -18,6 +18,7 @@ use crate::disk::error::{
 use crate::disk::os::{check_path_length, is_empty_dir};
 use crate::disk::{LocalFileReader, LocalFileWriter, STORAGE_FORMAT_FILE};
 use crate::error::{Error, Result};
+use crate::file_meta::read_xl_meta_no_data;
 use crate::global::{GLOBAL_IsErasureSD, GLOBAL_RootDiskThreshold};
 use crate::heal::data_scanner::{has_active_rules, scan_data_folder, ScannerItem, ShouldSleepFn, SizeSummary};
 use crate::heal::data_scanner_metric::{ScannerMetric, ScannerMetrics};
@@ -452,7 +453,6 @@ impl LocalDisk {
         Ok(data)
     }
 
-    // FIXME: read_metadata only suport
     async fn read_metadata_with_dmtime(&self, file_path: impl AsRef<Path>) -> Result<(Vec<u8>, Option<OffsetDateTime>)> {
         check_path_length(file_path.as_ref().to_string_lossy().as_ref())?;
 
@@ -471,18 +471,15 @@ impl LocalDisk {
         }
 
         let size = meta.len() as usize;
-        let mut bytes = Vec::new();
-        bytes.try_reserve_exact(size)?;
 
-        // FIXME: real xl no data
+        let data = read_xl_meta_no_data(&mut f, size).await?;
-        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))
+        Ok((data, modtime))
     }
 
     async fn read_all_data(&self, volume: &str, volume_dir: impl AsRef<Path>, file_path: impl AsRef<Path>) -> Result<Vec<u8>> {
@@ -1475,7 +1472,7 @@ impl DiskAPI for LocalDisk {
         let mut xlmeta = FileMeta::new();
 
         if let Some(dst_buf) = has_dst_buf.as_ref() {
-            if FileMeta::is_xl_format(dst_buf) {
+            if FileMeta::is_xl2_v1_format(dst_buf) {
                 if let Ok(nmeta) = FileMeta::load(dst_buf) {
                     xlmeta = nmeta
                 }
@@ -1723,7 +1720,7 @@ impl DiskAPI for LocalDisk {
                     }
                 })?;
 
-            if !FileMeta::is_xl_format(buf.as_slice()) {
+            if !FileMeta::is_xl2_v1_format(buf.as_slice()) {
                 return Err(Error::new(DiskError::FileVersionNotFound));
             }
 

ecstore/src/file_meta.rs

@@ -3,10 +3,11 @@ use rmp::Marker;
 use serde::{Deserialize, Serialize};
 use std::cmp::Ordering;
 use std::fmt::Display;
-use std::io::{Read, Write};
+use std::io::{self, Read, Write};
 use std::{collections::HashMap, io::Cursor};
 use time::OffsetDateTime;
-use tracing::warn;
+use tokio::io::AsyncRead;
+use tracing::{error, warn};
 use uuid::Uuid;
 use xxhash_rust::xxh64;
 
@@ -35,6 +36,9 @@ const XL_FLAG_FREE_VERSION: u8 = 1 << 0;
 // const XL_FLAG_USES_DATA_DIR: u8 = 1 << 1;
 const _XL_FLAG_INLINE_DATA: u8 = 1 << 2;
 
+const META_DATA_READ_DEFAULT: usize = 4 << 10;
+const MSGP_UINT32_SIZE: usize = 5;
+
 #[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
 pub struct FileMeta {
     pub versions: Vec<FileMetaShallowVersion>,
@@ -51,8 +55,9 @@ impl FileMeta {
         }
     }
 
-    pub fn is_xl_format(buf: &[u8]) -> bool {
+    // isXL2V1Format
-        !matches!(Self::read_xl_file_header(buf), Err(_e))
+    pub fn is_xl2_v1_format(buf: &[u8]) -> bool {
+        !matches!(Self::check_xl2_v1(buf), Err(_e))
     }
 
     pub fn load(buf: &[u8]) -> Result<FileMeta> {
@@ -62,9 +67,10 @@ impl FileMeta {
         Ok(xl)
     }
 
-    // read_xl_file_header 读xl文件头，返回后续内容，版本信息
+    // check_xl2_v1 读xl文件头，返回后续内容，版本信息
+    // checkXL2V1
     #[tracing::instrument]
-    pub fn read_xl_file_header(buf: &[u8]) -> Result<(&[u8], u16, u16)> {
+    pub fn check_xl2_v1(buf: &[u8]) -> Result<(&[u8], u16, u16)> {
         if buf.len() < 8 {
             return Err(Error::msg("xl file header not exists"));
         }
@@ -81,12 +87,22 @@ impl FileMeta {
 
         Ok((&buf[8..], major, minor))
     }
+
+    // 固定u32
+    pub fn read_bytes_header(buf: &[u8]) -> Result<(u32, &[u8])> {
+        let (mut size_buf, _) = buf.split_at(5);
+
+        //  取meta数据，buf = crc + data
+        let bin_len = rmp::decode::read_bin_len(&mut size_buf)?;
+
+        Ok((bin_len, &buf[5..]))
+    }
     #[tracing::instrument]
     pub fn unmarshal_msg(&mut self, buf: &[u8]) -> Result<u64> {
         let i = buf.len() as u64;
 
         // check version, buf = buf[8..]
-        let (buf, _, _) = Self::read_xl_file_header(buf)?;
+        let (buf, _, _) = Self::check_xl2_v1(buf)?;
 
         let (mut size_buf, buf) = buf.split_at(5);
 
@@ -1994,9 +2010,96 @@ async fn get_file_info(buf: &[u8], volume: &str, path: &str, version_id: &str, o
 
     Ok(fi)
 }
+
+async fn read_more<R: AsyncRead + Unpin>(
+    reader: &mut R,
+    buf: &mut Vec<u8>,
+    total_size: usize,
+    read_size: usize,
+    has_full: bool,
+) -> Result<()> {
+    use tokio::io::AsyncReadExt;
+    let has = buf.len();
+
+    if has >= read_size {
+        return Ok(());
+    }
+
+    if has_full || read_size > total_size {
+        return Err(Error::new(io::Error::new(io::ErrorKind::UnexpectedEof, "Unexpected EOF")));
+    }
+
+    let extra = read_size - has;
+    if buf.capacity() >= read_size {
+        // Extend the buffer if we have enough space.
+        buf.resize(read_size, 0);
+    } else {
+        buf.extend(vec![0u8; extra]);
+    }
+
+    reader.read_exact(&mut buf[has..]).await?;
+    Ok(())
+}
+
+pub async fn read_xl_meta_no_data<R: AsyncRead + Unpin>(reader: &mut R, size: usize) -> Result<Vec<u8>> {
+    use tokio::io::AsyncReadExt;
+
+    let mut initial = size;
+    let mut has_full = true;
+
+    if initial > META_DATA_READ_DEFAULT {
+        initial = META_DATA_READ_DEFAULT;
+        has_full = false;
+    }
+
+    let mut buf = vec![0u8; initial];
+    reader.read_exact(&mut buf).await?;
+
+    let (tmp_buf, major, minor) = FileMeta::check_xl2_v1(&buf)?;
+
+    match major {
+        1 => match minor {
+            0 => {
+                read_more(reader, &mut buf, size, size, has_full).await?;
+                Ok(buf)
+            }
+            1..=3 => {
+                let (sz, tmp_buf) = FileMeta::read_bytes_header(tmp_buf)?;
+                let mut want = sz as usize + (buf.len() - tmp_buf.len());
+
+                if minor < 2 {
+                    read_more(reader, &mut buf, size, want, has_full).await?;
+                    return Ok(buf[..want].to_vec());
+                }
+
+                let want_max = usize::min(want + MSGP_UINT32_SIZE, size);
+                read_more(reader, &mut buf, size, want_max, has_full).await?;
+
+                if buf.len() < want {
+                    error!("read_xl_meta_no_data buffer too small (length: {}, needed: {})", &buf.len(), want);
+                    return Err(Error::new(DiskError::FileCorrupt));
+                }
+
+                let tmp = &buf[want..];
+                let crc_size = 5;
+                let other_size = tmp.len() - crc_size;
+
+                want += tmp.len() - other_size;
+
+                Ok(buf[..want].to_vec())
+            }
+            _ => Err(Error::new(io::Error::new(io::ErrorKind::InvalidData, "Unknown minor metadata version"))),
+        },
+        _ => Err(Error::new(io::Error::new(io::ErrorKind::InvalidData, "Unknown major metadata version"))),
+    }
+}
 #[cfg(test)]
 mod test {
 
+    use std::fs;
+    use std::fs::File;
+    use std::os::unix::fs::MetadataExt;
+
     use super::*;
 
     #[test]
@@ -2012,15 +2115,11 @@ mod test {
             fm.add_version(fi).unwrap();
         }
 
-        // println!("fm:{:?}", &fm);
-
         let buff = fm.marshal_msg().unwrap();
 
         let mut newfm = FileMeta::default();
         newfm.unmarshal_msg(&buff).unwrap();
 
-        // println!("newone:{:?}", newone);
-
         assert_eq!(fm, newfm)
     }
 
@@ -2107,3 +2206,44 @@ mod test {
         assert_eq!(obj.version_id, vid);
     }
 }
+
+#[tokio::test]
+async fn test_read_xl_meta_no_data() {
+    use tokio::fs;
+    use tokio::fs::File;
+    use tokio::io::AsyncWriteExt;
+
+    let mut fm = FileMeta::new();
+
+    let (m, n) = (3, 2);
+
+    for i in 0..5 {
+        let mut fi = FileInfo::new(i.to_string().as_str(), m, n);
+        fi.mod_time = Some(OffsetDateTime::now_utc());
+
+        fm.add_version(fi).unwrap();
+    }
+
+    let mut buff = fm.marshal_msg().unwrap();
+
+    buff.resize(buff.len() + 100, 0);
+
+    let filepath = "./test_xl.meta";
+
+    let mut file = File::create(filepath).await.unwrap();
+    // 写入字符串
+    file.write_all(&buff).await.unwrap();
+
+    let mut f = File::open(filepath).await.unwrap();
+
+    let stat = f.metadata().await.unwrap();
+
+    let data = read_xl_meta_no_data(&mut f, stat.len() as usize).await.unwrap();
+
+    let mut newfm = FileMeta::default();
+    newfm.unmarshal_msg(&data).unwrap();
+
+    fs::remove_file(filepath).await.unwrap();
+
+    assert_eq!(fm, newfm)
+}

ecstore/src/lib.rs

@@ -1,5 +1,6 @@
 pub mod admin_server_info;
 pub mod bitrot;
+pub mod bucket;
 pub mod cache_value;
 mod chunk_stream;
 pub mod config;
@@ -9,25 +10,23 @@ pub mod endpoints;
 pub mod erasure;
 pub mod error;
 mod file_meta;
+pub mod file_meta_inline;
 pub mod global;
 pub mod heal;
 pub mod metrics_realtime;
 pub mod notification_sys;
 pub mod peer;
 pub mod peer_rest_client;
+pub mod pools;
 mod quorum;
 pub mod set_disk;
 mod sets;
 pub mod store;
 pub mod store_api;
-mod store_init;
-pub mod utils;
-
-pub mod bucket;
-pub mod file_meta_inline;
-
-pub mod pools;
 pub mod store_err;
+mod store_init;
+mod store_list_objects;
+pub mod utils;
 pub mod xhttp;
 
 pub use global::new_object_layer_fn;

ecstore/src/notification_sys.rs

@@ -46,6 +46,11 @@ pub struct NotificationPeerErr {
 }
 
 impl NotificationSys {
+    pub fn rest_client_from_hash(&self, s:&str) ->Option<PeerRestClient>{
+        
+
+        None
+    }
     pub async fn delete_policy(&self) -> Vec<NotificationPeerErr> {
         unimplemented!()
     }

ecstore/src/store.rs

@@ -24,19 +24,19 @@ use crate::store_err::{
 };
 use crate::store_init::ec_drives_no_config;
 use crate::utils::crypto::base64_decode;
-use crate::utils::path::{base_dir_from_prefix, decode_dir_object, encode_dir_object, SLASH_SEPARATOR};
+use crate::utils::path::{decode_dir_object, encode_dir_object, SLASH_SEPARATOR};
 use crate::utils::xml;
 use crate::{
     bucket::metadata::BucketMetadata,
-    disk::{error::DiskError, new_disk, DiskOption, DiskStore, WalkDirOptions, BUCKET_META_PREFIX, RUSTFS_META_BUCKET},
+    disk::{error::DiskError, new_disk, DiskOption, DiskStore, BUCKET_META_PREFIX, RUSTFS_META_BUCKET},
     endpoints::EndpointServerPools,
     error::{Error, Result},
     peer::S3PeerSys,
     sets::Sets,
     store_api::{
         BucketInfo, BucketOptions, CompletePart, DeleteBucketOptions, DeletedObject, GetObjectReader, HTTPRangeSpec,
-        ListObjectsInfo, ListObjectsV2Info, MakeBucketOptions, MultipartUploadResult, ObjectInfo, ObjectOptions, ObjectToDelete,
+        ListObjectsV2Info, MakeBucketOptions, MultipartUploadResult, ObjectInfo, ObjectOptions, ObjectToDelete, PartInfo,
-        PartInfo, PutObjReader, StorageAPI,
+        PutObjReader, StorageAPI,
     },
     store_init, utils,
 };
@@ -52,11 +52,7 @@ use std::cmp::Ordering;
 use std::process::exit;
 use std::slice::Iter;
 use std::time::SystemTime;
-use std::{
+use std::{collections::HashMap, sync::Arc, time::Duration};
-    collections::{HashMap, HashSet},
-    sync::Arc,
-    time::Duration,
-};
 use time::OffsetDateTime;
 use tokio::select;
 use tokio::sync::mpsc::Sender;
@@ -265,102 +261,104 @@ impl ECStore {
         self.pools.len() == 1
     }
 
-    pub async fn list_path(&self, opts: &ListPathOptions, delimiter: &str) -> Result<ListObjectsInfo> {
+    // define in store_list_objects.rs
-        // if opts.prefix.ends_with(SLASH_SEPARATOR) {
+    // pub async fn list_path(&self, opts: &ListPathOptions, delimiter: &str) -> Result<ListObjectsInfo> {
-        //     return Err(Error::msg("eof"));
+    //     // if opts.prefix.ends_with(SLASH_SEPARATOR) {
-        // }
+    //     //     return Err(Error::msg("eof"));
+    //     // }
 
-        let mut opts = opts.clone();
+    //     let mut opts = opts.clone();
 
-        if opts.base_dir.is_empty() {
+    //     if opts.base_dir.is_empty() {
-            opts.base_dir = base_dir_from_prefix(&opts.prefix);
+    //         opts.base_dir = base_dir_from_prefix(&opts.prefix);
-        }
+    //     }
 
-        let objects = self.list_merged(&opts, delimiter).await?;
+    //     let objects = self.list_merged(&opts, delimiter).await?;
 
-        let info = ListObjectsInfo {
+    //     let info = ListObjectsInfo {
-            objects,
+    //         objects,
-            ..Default::default()
+    //         ..Default::default()
-        };
+    //     };
-        Ok(info)
+    //     Ok(info)
-    }
+    // }
 
     // 读所有
-    async fn list_merged(&self, opts: &ListPathOptions, delimiter: &str) -> Result<Vec<ObjectInfo>> {
+    // define in store_list_objects.rs
-        let walk_opts = WalkDirOptions {
+    // async fn list_merged(&self, opts: &ListPathOptions, delimiter: &str) -> Result<Vec<ObjectInfo>> {
-            bucket: opts.bucket.clone(),
+    //     let walk_opts = WalkDirOptions {
-            base_dir: opts.base_dir.clone(),
+    //         bucket: opts.bucket.clone(),
-            ..Default::default()
+    //         base_dir: opts.base_dir.clone(),
-        };
+    //         ..Default::default()
+    //     };
 
-        // let (mut wr, mut rd) = tokio::io::duplex(1024);
+    //     // let (mut wr, mut rd) = tokio::io::duplex(1024);
 
-        let mut futures = Vec::new();
+    //     let mut futures = Vec::new();
 
-        for sets in self.pools.iter() {
+    //     for sets in self.pools.iter() {
-            for set in sets.disk_set.iter() {
+    //         for set in sets.disk_set.iter() {
-                futures.push(set.walk_dir(&walk_opts));
+    //             futures.push(set.walk_dir(&walk_opts));
-            }
+    //         }
-        }
+    //     }
 
-        let results = join_all(futures).await;
+    //     let results = join_all(futures).await;
 
-        // let mut errs = Vec::new();
+    //     // let mut errs = Vec::new();
-        let mut ress = Vec::new();
+    //     let mut ress = Vec::new();
-        let mut uniq = HashSet::new();
+    //     let mut uniq = HashSet::new();
 
-        for (disks_ress, _disks_errs) in results {
+    //     for (disks_ress, _disks_errs) in results {
-            for disks_res in disks_ress.iter() {
+    //         for disks_res in disks_ress.iter() {
-                if disks_res.is_none() {
+    //             if disks_res.is_none() {
-                    // TODO handle errs
+    //                 // TODO handle errs
-                    continue;
+    //                 continue;
-                }
+    //             }
-                let entrys = disks_res.as_ref().unwrap();
+    //             let entrys = disks_res.as_ref().unwrap();
 
-                for entry in entrys {
+    //             for entry in entrys {
-                    // warn!("lst_merged entry---- {}", &entry.name);
+    //                 // warn!("lst_merged entry---- {}", &entry.name);
 
-                    if !opts.prefix.is_empty() && !entry.name.starts_with(&opts.prefix) {
+    //                 if !opts.prefix.is_empty() && !entry.name.starts_with(&opts.prefix) {
-                        continue;
+    //                     continue;
-                    }
+    //                 }
 
-                    if !uniq.contains(&entry.name) {
+    //                 if !uniq.contains(&entry.name) {
-                        uniq.insert(entry.name.clone());
+    //                     uniq.insert(entry.name.clone());
-                        // TODO: 过滤
+    //                     // TODO: 过滤
 
-                        if opts.limit > 0 && ress.len() as i32 >= opts.limit {
+    //                     if opts.limit > 0 && ress.len() as i32 >= opts.limit {
-                            return Ok(ress);
+    //                         return Ok(ress);
-                        }
+    //                     }
 
-                        if entry.is_object() {
+    //                     if entry.is_object() {
-                            if !delimiter.is_empty() {
+    //                         if !delimiter.is_empty() {
-                                // entry.name.trim_start_matches(pat)
+    //                             // entry.name.trim_start_matches(pat)
-                            }
+    //                         }
 
-                            let fi = entry.to_fileinfo(&opts.bucket)?;
+    //                         let fi = entry.to_fileinfo(&opts.bucket)?;
-                            if let Some(f) = fi {
+    //                         if let Some(f) = fi {
-                                ress.push(f.to_object_info(&opts.bucket, &entry.name, false));
+    //                             ress.push(f.to_object_info(&opts.bucket, &entry.name, false));
-                            }
+    //                         }
-                            continue;
+    //                         continue;
-                        }
+    //                     }
 
-                        if entry.is_dir() {
+    //                     if entry.is_dir() {
-                            ress.push(ObjectInfo {
+    //                         ress.push(ObjectInfo {
-                                is_dir: true,
+    //                             is_dir: true,
-                                bucket: opts.bucket.clone(),
+    //                             bucket: opts.bucket.clone(),
-                                name: entry.name.clone(),
+    //                             name: entry.name.clone(),
-                                ..Default::default()
+    //                             ..Default::default()
-                            });
+    //                         });
-                        }
+    //                     }
-                    }
+    //                 }
-                }
+    //             }
-            }
+    //         }
-        }
+    //     }
 
-        // warn!("list_merged errs {:?}", errs);
+    //     // warn!("list_merged errs {:?}", errs);
 
-        Ok(ress)
+    //     Ok(ress)
-    }
+    // }
 
     async fn delete_all(&self, bucket: &str, prefix: &str) -> Result<()> {
         let mut futures = Vec::new();
@@ -1520,29 +1518,32 @@ impl StorageAPI for ECStore {
         continuation_token: &str,
         delimiter: &str,
         max_keys: i32,
-        _fetch_owner: bool,
+        fetch_owner: bool,
-        _start_after: &str,
+        start_after: &str,
     ) -> Result<ListObjectsV2Info> {
-        let opts = ListPathOptions {
+        self.inner_list_objects_v2(&bucket, &prefix, &continuation_token, &delimiter, max_keys, fetch_owner, start_after)
-            bucket: bucket.to_string(),
+            .await
-            limit: max_keys,
-            prefix: prefix.to_owned(),
-            ..Default::default()
-        };
 
-        let info = self.list_path(&opts, delimiter).await?;
+        // let opts = ListPathOptions {
+        //     bucket: bucket.to_string(),
+        //     limit: max_keys,
+        //     prefix: prefix.to_owned(),
+        //     ..Default::default()
+        // };
 
-        // warn!("list_objects_v2 info {:?}", info);
+        // let info = self.list_path(&opts, delimiter).await?;
 
-        let v2 = ListObjectsV2Info {
+        // // warn!("list_objects_v2 info {:?}", info);
-            is_truncated: info.is_truncated,
-            continuation_token: continuation_token.to_owned(),
-            next_continuation_token: info.next_marker,
-            objects: info.objects,
-            prefixes: info.prefixes,
-        };
 
-        Ok(v2)
+        // let v2 = ListObjectsV2Info {
+        //     is_truncated: info.is_truncated,
+        //     continuation_token: continuation_token.to_owned(),
+        //     next_continuation_token: info.next_marker,
+        //     objects: info.objects,
+        //     prefixes: info.prefixes,
+        // };
+
+        // Ok(v2)
     }
     async fn list_object_versions(
         &self,
@@ -2238,7 +2239,7 @@ fn check_bucket_and_object_names(bucket: &str, object: &str) -> Result<()> {
     Ok(())
 }
 
-fn check_list_objs_args(bucket: &str, prefix: &str, _marker: &str) -> Result<()> {
+pub fn check_list_objs_args(bucket: &str, prefix: &str, _marker: &str) -> Result<()> {
     if !is_meta_bucketname(bucket) && check_valid_bucket_name_strict(bucket).is_err() {
         return Err(Error::new(StorageError::BucketNameInvalid(bucket.to_string())));
     }

ecstore/src/store_list_objects.rs

@@ -101,6 +101,7 @@ impl ListPathOptions {
 }
 
 impl ECStore {
+    #[allow(clippy::too_many_arguments)]
     pub async fn inner_list_objects_v2(
         &self,
         bucket: &str,
@@ -120,7 +121,7 @@ impl ECStore {
         };
 
         self.list_objects_generic(bucket, prefix, marker, delimiter, max_keys).await?;
-
+        // FIXME:TODO:
         unimplemented!()
     }
 
@@ -145,6 +146,8 @@ impl ECStore {
 
         let merged = self.list_path(&opts).await?;
 
+        // FIXME:TODO:
+
         todo!()
     }
 
@@ -159,10 +162,8 @@ impl ECStore {
             o.marker = "".to_owned();
         }
 
-        if !o.marker.is_empty() && !o.prefix.is_empty() {
+        if !o.marker.is_empty() && !o.prefix.is_empty() && !o.marker.starts_with(&o.prefix) {
-            if !o.marker.starts_with(&o.prefix) {
+            return Err(Error::new(std::io::Error::from(ErrorKind::UnexpectedEof)));
-                return Err(Error::new(std::io::Error::from(ErrorKind::UnexpectedEof)));
-            }
         }
 
         if o.limit == 0 {
@@ -194,6 +195,7 @@ impl ECStore {
             o.create = false;
         }
 
+        // FIXME:TODO:
         todo!()
 
         // let mut opts = opts.clone();