Enhance Object Version Management and Replication Status Handling (#1413)

crates/ecstore/src/disk/local.rs

@@ -21,6 +21,7 @@ use super::{
 };
 use super::{endpoint::Endpoint, error::DiskError, format::FormatV3};
 
+use crate::config::storageclass::DEFAULT_INLINE_BLOCK;
 use crate::data_usage::local_snapshot::ensure_data_usage_layout;
 use crate::disk::error::FileAccessDeniedWithContext;
 use crate::disk::error_conv::{to_access_error, to_file_error, to_unformatted_disk_error, to_volume_error};
@@ -489,9 +490,17 @@ impl LocalDisk {
         let file_dir = self.get_bucket_path(volume)?;
         let (data, _) = self.read_raw(volume, file_dir, file_path, opts.read_data).await?;
 
-        get_file_info(&data, volume, path, version_id, FileInfoOpts { data: opts.read_data })
-            .await
-            .map_err(|_e| DiskError::Unexpected)
+        get_file_info(
+            &data,
+            volume,
+            path,
+            version_id,
+            FileInfoOpts {
+                data: opts.read_data,
+                include_free_versions: false,
+            },
+        )
+        .map_err(|_e| DiskError::Unexpected)
     }
 
     // Batch metadata reading for multiple objects
@@ -2240,20 +2249,93 @@ impl DiskAPI for LocalDisk {
     #[tracing::instrument(level = "debug", skip(self))]
     async fn read_version(
         &self,
-        _org_volume: &str,
+        org_volume: &str,
         volume: &str,
         path: &str,
         version_id: &str,
         opts: &ReadOptions,
     ) -> Result<FileInfo> {
+        if !org_volume.is_empty() {
+            let org_volume_path = self.get_bucket_path(org_volume)?;
+            if !skip_access_checks(org_volume) {
+                access(&org_volume_path)
+                    .await
+                    .map_err(|e| to_access_error(e, DiskError::VolumeAccessDenied))?;
+            }
+        }
+
         let file_path = self.get_object_path(volume, path)?;
-        let file_dir = self.get_bucket_path(volume)?;
+        let volume_dir = self.get_bucket_path(volume)?;
+
+        check_path_length(file_path.to_string_lossy().as_ref())?;
 
         let read_data = opts.read_data;
 
-        let (data, _) = self.read_raw(volume, file_dir, file_path, read_data).await?;
+        let (data, _) = self
+            .read_raw(volume, volume_dir.clone(), file_path, read_data)
+            .await
+            .map_err(|e| {
+                if e == DiskError::FileNotFound && !version_id.is_empty() {
+                    DiskError::FileVersionNotFound
+                } else {
+                    e
+                }
+            })?;
 
-        let fi = get_file_info(&data, volume, path, version_id, FileInfoOpts { data: read_data }).await?;
+        let mut fi = get_file_info(
+            &data,
+            volume,
+            path,
+            version_id,
+            FileInfoOpts {
+                data: read_data,
+                include_free_versions: opts.incl_free_versions,
+            },
+        )?;
+
+        if opts.read_data {
+            if fi.data.as_ref().is_some_and(|d| !d.is_empty()) || fi.size == 0 {
+                if fi.inline_data() {
+                    return Ok(fi);
+                }
+
+                if fi.size == 0 || fi.version_id.is_none_or(|v| v.is_nil()) {
+                    fi.set_inline_data();
+                    return Ok(fi);
+                };
+                if let Some(part) = fi.parts.first() {
+                    let part_path = format!("part.{}", part.number);
+                    let part_path = path_join_buf(&[
+                        path,
+                        fi.data_dir.map_or("".to_string(), |dir| dir.to_string()).as_str(),
+                        part_path.as_str(),
+                    ]);
+                    let part_path = self.get_object_path(volume, part_path.as_str())?;
+                    if lstat(&part_path).await.is_err() {
+                        fi.set_inline_data();
+                        return Ok(fi);
+                    }
+                }
+
+                fi.data = None;
+            }
+
+            let inline = fi.transition_status.is_empty() && fi.data_dir.is_some() && fi.parts.len() == 1;
+            if inline && fi.shard_file_size(fi.parts[0].actual_size) < DEFAULT_INLINE_BLOCK as i64 {
+                let part_path = path_join_buf(&[
+                    path,
+                    fi.data_dir.map_or("".to_string(), |dir| dir.to_string()).as_str(),
+                    format!("part.{}", fi.parts[0].number).as_str(),
+                ]);
+                let part_path = self.get_object_path(volume, part_path.as_str())?;
+
+                let data = self.read_all_data(volume, volume_dir, part_path.clone()).await.map_err(|e| {
+                    warn!("read_version read_all_data {:?} failed: {e}", part_path);
+                    e
+                })?;
+                fi.data = Some(Bytes::from(data));
+            }
+        }
 
         Ok(fi)
     }

crates/ecstore/src/set_disk.rs

@@ -1488,17 +1488,7 @@ impl SetDisks {
                 if let Some(disk) = disk
                     && disk.is_online().await
                 {
-                    if version_id.is_empty() {
-                        match disk.read_xl(&bucket, &object, read_data).await {
-                            Ok(info) => {
-                                let fi = file_info_from_raw(info, &bucket, &object, read_data).await?;
-                                Ok(fi)
-                            }
-                            Err(err) => Err(err),
-                        }
-                    } else {
-                        disk.read_version(&org_bucket, &bucket, &object, &version_id, &opts).await
-                    }
+                    disk.read_version(&org_bucket, &bucket, &object, &version_id, &opts).await
                 } else {
                     Err(DiskError::DiskNotFound)
                 }
@@ -1626,7 +1616,7 @@ impl SetDisks {
         bucket: &str,
         object: &str,
         read_data: bool,
-        _incl_free_vers: bool,
+        incl_free_vers: bool,
     ) -> (Vec<FileInfo>, Vec<Option<DiskError>>) {
         let mut metadata_array = vec![None; fileinfos.len()];
         let mut meta_file_infos = vec![FileInfo::default(); fileinfos.len()];
@@ -1676,7 +1666,7 @@ impl SetDisks {
             ..Default::default()
         };
 
-        let finfo = match meta.into_fileinfo(bucket, object, "", true, true) {
+        let finfo = match meta.into_fileinfo(bucket, object, "", true, incl_free_vers, true) {
             Ok(res) => res,
             Err(err) => {
                 for item in errs.iter_mut() {
@@ -1703,7 +1693,7 @@ impl SetDisks {
 
         for (idx, meta_op) in metadata_array.iter().enumerate() {
             if let Some(meta) = meta_op {
-                match meta.into_fileinfo(bucket, object, vid.to_string().as_str(), read_data, true) {
+                match meta.into_fileinfo(bucket, object, vid.to_string().as_str(), read_data, incl_free_vers, true) {
                     Ok(res) => meta_file_infos[idx] = res,
                     Err(err) => errs[idx] = Some(err.into()),
                 }

crates/ecstore/src/store_api.rs

@@ -28,14 +28,15 @@ use http::{HeaderMap, HeaderValue};
 use rustfs_common::heal_channel::HealOpts;
 use rustfs_filemeta::{
     FileInfo, MetaCacheEntriesSorted, ObjectPartInfo, REPLICATION_RESET, REPLICATION_STATUS, ReplicateDecision, ReplicationState,
-    ReplicationStatusType, VersionPurgeStatusType, replication_statuses_map, version_purge_statuses_map,
+    ReplicationStatusType, RestoreStatusOps as _, VersionPurgeStatusType, parse_restore_obj_status, replication_statuses_map,
+    version_purge_statuses_map,
 };
 use rustfs_madmin::heal_commands::HealResultItem;
 use rustfs_rio::Checksum;
 use rustfs_rio::{DecompressReader, HashReader, LimitReader, WarpReader};
 use rustfs_utils::CompressionAlgorithm;
-use rustfs_utils::http::AMZ_STORAGE_CLASS;
 use rustfs_utils::http::headers::{AMZ_OBJECT_TAGGING, RESERVED_METADATA_PREFIX_LOWER};
+use rustfs_utils::http::{AMZ_BUCKET_REPLICATION_STATUS, AMZ_RESTORE, AMZ_STORAGE_CLASS};
 use rustfs_utils::path::decode_dir_object;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
@@ -756,7 +757,24 @@ impl ObjectInfo {
                 .ok()
         });
 
-        // TODO:ReplicationState
+        let replication_status_internal = fi
+            .replication_state_internal
+            .as_ref()
+            .and_then(|v| v.replication_status_internal.clone());
+        let version_purge_status_internal = fi
+            .replication_state_internal
+            .as_ref()
+            .and_then(|v| v.version_purge_status_internal.clone());
+
+        let mut replication_status = fi.replication_status();
+        if replication_status.is_empty()
+            && let Some(status) = fi.metadata.get(AMZ_BUCKET_REPLICATION_STATUS).cloned()
+            && status == ReplicationStatusType::Replica.as_str()
+        {
+            replication_status = ReplicationStatusType::Replica;
+        }
+
+        let version_purge_status = fi.version_purge_status();
 
         let transitioned_object = TransitionedObject {
             name: fi.transitioned_objname.clone(),
@@ -777,10 +795,24 @@ impl ObjectInfo {
         };
 
         // Extract storage class from metadata, default to STANDARD if not found
-        let storage_class = metadata
-            .get(AMZ_STORAGE_CLASS)
-            .cloned()
-            .or_else(|| Some(storageclass::STANDARD.to_string()));
+        let storage_class = if !fi.transition_tier.is_empty() {
+            Some(fi.transition_tier.clone())
+        } else {
+            fi.metadata
+                .get(AMZ_STORAGE_CLASS)
+                .cloned()
+                .or_else(|| Some(storageclass::STANDARD.to_string()))
+        };
+
+        let mut restore_ongoing = false;
+        let mut restore_expires = None;
+        if let Some(restore_status) = fi.metadata.get(AMZ_RESTORE).cloned() {
+            //
+            if let Ok(restore_status) = parse_restore_obj_status(&restore_status) {
+                restore_ongoing = restore_status.on_going();
+                restore_expires = restore_status.expiry();
+            }
+        }
 
         // Convert parts from rustfs_filemeta::ObjectPartInfo to store_api::ObjectPartInfo
         let parts = fi
@@ -798,6 +830,8 @@ impl ObjectInfo {
             })
             .collect();
 
+        // TODO: part checksums
+
         ObjectInfo {
             bucket: bucket.to_string(),
             name,
@@ -822,6 +856,12 @@ impl ObjectInfo {
             transitioned_object,
             checksum: fi.checksum.clone(),
             storage_class,
+            restore_ongoing,
+            restore_expires,
+            replication_status_internal,
+            replication_status,
+            version_purge_status_internal,
+            version_purge_status,
             ..Default::default()
         }
     }