feat: Add RustFS Scanner Module and Multiple Bug Fixes (#1579)

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -23,6 +23,7 @@ use crate::bucket::target::{self, BucketTarget, BucketTargets, Credentials};
 use crate::bucket::versioning_sys::BucketVersioningSys;
 use aws_credential_types::Credentials as SdkCredentials;
 use aws_sdk_s3::config::Region as SdkRegion;
+use aws_sdk_s3::error::ProvideErrorMetadata;
 use aws_sdk_s3::error::SdkError;
 use aws_sdk_s3::operation::complete_multipart_upload::CompleteMultipartUploadOutput;
 use aws_sdk_s3::operation::head_bucket::HeadBucketError;
@@ -1094,8 +1095,7 @@ pub struct TargetClient {
 
 impl TargetClient {
     pub fn to_url(&self) -> Url {
-        let scheme = if self.secure { "https" } else { "http" };
-        Url::parse(&format!("{scheme}://{}", self.endpoint)).unwrap()
+        Url::parse(&self.endpoint).unwrap()
     }
 
     pub async fn bucket_exists(&self, bucket: &str) -> Result<bool, S3ClientError> {
@@ -1104,9 +1104,17 @@ impl TargetClient {
             Err(e) => match e {
                 SdkError::ServiceError(oe) => match oe.into_err() {
                     HeadBucketError::NotFound(_) => Ok(false),
-                    other => Err(S3ClientError::new(format!(
-                        "failed to check bucket exists for bucket:{bucket} please check the bucket name and credentials, error:{other:?}"
-                    ))),
+                    other => {
+                        warn!(
+                            "failed to check bucket exists for bucket:{bucket} please check the bucket name and credentials, error:{:?}",
+                            other
+                        );
+                        let message = other.meta().meta();
+                        Err(S3ClientError::new(format!(
+                            "failed to check bucket exists for bucket:{bucket} please check the bucket name and credentials, error:{:?}",
+                            message
+                        )))
+                    }
                 },
                 SdkError::DispatchFailure(e) => Err(S3ClientError::new(format!(
                     "failed to dispatch bucket exists for bucket:{bucket} error:{e:?}"
@@ -1154,10 +1162,17 @@ impl TargetClient {
         body: ByteStream,
         opts: &PutObjectOptions,
     ) -> Result<(), S3ClientError> {
-        let headers = opts.header();
+        let mut headers = opts.header();
 
         let builder = self.client.put_object();
 
+        let version_id = opts.internal.source_version_id.clone();
+        if !version_id.is_empty()
+            && let Ok(header_value) = HeaderValue::from_str(&version_id)
+        {
+            headers.insert(RUSTFS_BUCKET_SOURCE_VERSION_ID, header_value);
+        }
+
         match builder
             .bucket(bucket)
             .key(object)
@@ -1185,9 +1200,33 @@ impl TargetClient {
         &self,
         bucket: &str,
         object: &str,
-        _opts: &PutObjectOptions,
+        opts: &PutObjectOptions,
     ) -> Result<String, S3ClientError> {
-        match self.client.create_multipart_upload().bucket(bucket).key(object).send().await {
+        let mut headers = HeaderMap::new();
+        let version_id = opts.internal.source_version_id.clone();
+        if !version_id.is_empty()
+            && let Ok(header_value) = HeaderValue::from_str(&version_id)
+        {
+            headers.insert(RUSTFS_BUCKET_SOURCE_VERSION_ID, header_value);
+        }
+
+        match self
+            .client
+            .create_multipart_upload()
+            .bucket(bucket)
+            .key(object)
+            .customize()
+            .map_request(move |mut req| {
+                for (k, v) in headers.clone().into_iter() {
+                    let key_str = k.unwrap().as_str().to_string();
+                    let value_str = v.to_str().unwrap_or("").to_string();
+                    req.headers_mut().insert(key_str, value_str);
+                }
+                Result::<_, aws_smithy_types::error::operation::BuildError>::Ok(req)
+            })
+            .send()
+            .await
+        {
             Ok(res) => Ok(res.upload_id.unwrap_or_default()),
             Err(e) => Err(e.into()),
         }

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -953,7 +953,7 @@ impl LifecycleOps for ObjectInfo {
         lifecycle::ObjectOpts {
             name: self.name.clone(),
             user_tags: self.user_tags.clone(),
-            version_id: self.version_id.map(|v| v.to_string()).unwrap_or_default(),
+            version_id: self.version_id.clone(),
             mod_time: self.mod_time,
             size: self.size as usize,
             is_latest: self.is_latest,
@@ -1067,7 +1067,7 @@ pub async fn eval_action_from_lifecycle(
     event
 }
 
-async fn apply_transition_rule(event: &lifecycle::Event, src: &LcEventSrc, oi: &ObjectInfo) -> bool {
+pub async fn apply_transition_rule(event: &lifecycle::Event, src: &LcEventSrc, oi: &ObjectInfo) -> bool {
     if oi.delete_marker || oi.is_dir {
         return false;
     }
@@ -1161,7 +1161,7 @@ pub async fn apply_expiry_on_non_transitioned_objects(
     true
 }
 
-async fn apply_expiry_rule(event: &lifecycle::Event, src: &LcEventSrc, oi: &ObjectInfo) -> bool {
+pub async fn apply_expiry_rule(event: &lifecycle::Event, src: &LcEventSrc, oi: &ObjectInfo) -> bool {
     let mut expiry_state = GLOBAL_ExpiryState.write().await;
     expiry_state.enqueue_by_days(oi, event, src).await;
     true

crates/ecstore/src/bucket/lifecycle/evaluator.rs

@@ -0,0 +1,191 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+
+use s3s::dto::{
+    BucketLifecycleConfiguration, ObjectLockConfiguration, ObjectLockEnabled, ObjectLockLegalHoldStatus, ObjectLockRetentionMode,
+};
+use time::OffsetDateTime;
+use tracing::info;
+
+use crate::bucket::lifecycle::lifecycle::{Event, Lifecycle, ObjectOpts};
+use crate::bucket::object_lock::ObjectLockStatusExt;
+use crate::bucket::object_lock::objectlock::{get_object_legalhold_meta, get_object_retention_meta, utc_now_ntp};
+use crate::bucket::replication::ReplicationConfig;
+use rustfs_common::metrics::IlmAction;
+
+/// Evaluator - evaluates lifecycle policy on objects for the given lifecycle
+/// configuration, lock retention configuration and replication configuration.
+pub struct Evaluator {
+    policy: Arc<BucketLifecycleConfiguration>,
+    lock_retention: Option<Arc<ObjectLockConfiguration>>,
+    repl_cfg: Option<Arc<ReplicationConfig>>,
+}
+
+impl Evaluator {
+    /// NewEvaluator - creates a new evaluator with the given lifecycle
+    pub fn new(policy: Arc<BucketLifecycleConfiguration>) -> Self {
+        Self {
+            policy,
+            lock_retention: None,
+            repl_cfg: None,
+        }
+    }
+
+    /// WithLockRetention - sets the lock retention configuration for the evaluator
+    pub fn with_lock_retention(mut self, lr: Option<Arc<ObjectLockConfiguration>>) -> Self {
+        self.lock_retention = lr;
+        self
+    }
+
+    /// WithReplicationConfig - sets the replication configuration for the evaluator
+    pub fn with_replication_config(mut self, rcfg: Option<Arc<ReplicationConfig>>) -> Self {
+        self.repl_cfg = rcfg;
+        self
+    }
+
+    /// IsPendingReplication checks if the object is pending replication.
+    pub fn is_pending_replication(&self, obj: &ObjectOpts) -> bool {
+        use crate::bucket::replication::ReplicationConfigurationExt;
+        if self.repl_cfg.is_none() {
+            return false;
+        }
+        if let Some(rcfg) = &self.repl_cfg
+            && rcfg
+                .config
+                .as_ref()
+                .is_some_and(|config| config.has_active_rules(obj.name.as_str(), true))
+            && !obj.version_purge_status.is_empty()
+        {
+            return true;
+        }
+        false
+    }
+
+    /// IsObjectLocked checks if it is appropriate to remove an
+    /// object according to locking configuration when this is lifecycle/ bucket quota asking.
+    /// (copied over from enforceRetentionForDeletion)
+    pub fn is_object_locked(&self, obj: &ObjectOpts) -> bool {
+        if self.lock_retention.as_ref().is_none_or(|v| {
+            v.object_lock_enabled
+                .as_ref()
+                .is_none_or(|v| v.as_str() != ObjectLockEnabled::ENABLED)
+        }) {
+            return false;
+        }
+
+        if obj.delete_marker {
+            return false;
+        }
+
+        let lhold = get_object_legalhold_meta(obj.user_defined.clone());
+        if lhold
+            .status
+            .is_some_and(|v| v.valid() && v.as_str() == ObjectLockLegalHoldStatus::ON)
+        {
+            return true;
+        }
+
+        let ret = get_object_retention_meta(obj.user_defined.clone());
+        if ret
+            .mode
+            .is_some_and(|v| matches!(v.as_str(), ObjectLockRetentionMode::COMPLIANCE | ObjectLockRetentionMode::GOVERNANCE))
+        {
+            let t = utc_now_ntp();
+            if let Some(retain_until) = ret.retain_until_date
+                && OffsetDateTime::from(retain_until).gt(&t)
+            {
+                return true;
+            }
+        }
+        false
+    }
+
+    /// eval will return a lifecycle event for each object in objs for a given time.
+    async fn eval_inner(&self, objs: &[ObjectOpts], now: OffsetDateTime) -> Vec<Event> {
+        let mut events = vec![Event::default(); objs.len()];
+        let mut newer_noncurrent_versions = 0;
+
+        'top_loop: {
+            for (i, obj) in objs.iter().enumerate() {
+                let mut event = self.policy.eval_inner(obj, now, newer_noncurrent_versions).await;
+                match event.action {
+                    IlmAction::DeleteAllVersionsAction | IlmAction::DelMarkerDeleteAllVersionsAction => {
+                        // Skip if bucket has object locking enabled; To prevent the
+                        // possibility of violating an object retention on one of the
+                        // noncurrent versions of this object.
+                        if self.lock_retention.as_ref().is_some_and(|v| {
+                            v.object_lock_enabled
+                                .as_ref()
+                                .is_some_and(|v| v.as_str() == ObjectLockEnabled::ENABLED)
+                        }) {
+                            event = Event::default();
+                        } else {
+                            // No need to evaluate remaining versions' lifecycle
+                            // events after DeleteAllVersionsAction*
+                            events[i] = event;
+
+                            info!("eval_inner: skipping remaining versions' lifecycle events after DeleteAllVersionsAction*");
+
+                            break 'top_loop;
+                        }
+                    }
+                    IlmAction::DeleteVersionAction | IlmAction::DeleteRestoredVersionAction => {
+                        // Defensive code, should never happen
+                        if obj.version_id.is_none_or(|v| v.is_nil()) {
+                            event.action = IlmAction::NoneAction;
+                        }
+                        if self.is_object_locked(obj) {
+                            event = Event::default();
+                        }
+
+                        if self.is_pending_replication(obj) {
+                            event = Event::default();
+                        }
+                    }
+                    _ => {}
+                }
+
+                if !obj.is_latest {
+                    match event.action {
+                        IlmAction::DeleteVersionAction => {
+                            // this noncurrent version will be expired, nothing to add
+                        }
+                        _ => {
+                            // this noncurrent version will be spared
+                            newer_noncurrent_versions += 1;
+                        }
+                    }
+                }
+                events[i] = event;
+            }
+        }
+        events
+    }
+
+    /// Eval will return a lifecycle event for each object in objs
+    pub async fn eval(&self, objs: &[ObjectOpts]) -> Result<Vec<Event>, std::io::Error> {
+        if objs.is_empty() {
+            return Ok(vec![]);
+        }
+        if objs.len() != objs[0].num_versions {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidInput,
+                format!("number of versions mismatch, expected {}, got {}", objs[0].num_versions, objs.len()),
+            ));
+        }
+        Ok(self.eval_inner(objs, OffsetDateTime::now_utc()).await)
+    }
+}

crates/ecstore/src/bucket/lifecycle/lifecycle.rs

@@ -19,28 +19,31 @@
 #![allow(clippy::all)]
 
 use crate::bucket::lifecycle::rule::TransitionOps;
+use crate::store_api::ObjectInfo;
+use rustfs_filemeta::{ReplicationStatusType, VersionPurgeStatusType};
 use s3s::dto::{
     BucketLifecycleConfiguration, ExpirationStatus, LifecycleExpiration, LifecycleRule, NoncurrentVersionTransition,
     ObjectLockConfiguration, ObjectLockEnabled, RestoreRequest, Transition,
 };
 use std::cmp::Ordering;
+use std::collections::HashMap;
 use std::env;
 use std::fmt::Display;
 use std::sync::Arc;
 use time::macros::{datetime, offset};
 use time::{self, Duration, OffsetDateTime};
 use tracing::info;
+use uuid::Uuid;
 
 pub const TRANSITION_COMPLETE: &str = "complete";
 pub const TRANSITION_PENDING: &str = "pending";
-
-const ERR_LIFECYCLE_TOO_MANY_RULES: &str = "Lifecycle configuration allows a maximum of 1000 rules";
 const ERR_LIFECYCLE_NO_RULE: &str = "Lifecycle configuration should have at least one rule";
 const ERR_LIFECYCLE_DUPLICATE_ID: &str = "Rule ID must be unique. Found same ID for more than one rule";
 const _ERR_XML_NOT_WELL_FORMED: &str =
     "The XML you provided was not well-formed or did not validate against our published schema";
 const ERR_LIFECYCLE_BUCKET_LOCKED: &str =
     "ExpiredObjectAllVersions element and DelMarkerExpiration action cannot be used on an retention bucket";
+const ERR_LIFECYCLE_TOO_MANY_RULES: &str = "Lifecycle configuration should have at most 1000 rules";
 
 pub use rustfs_common::metrics::IlmAction;
 
@@ -130,11 +133,11 @@ impl RuleValidate for LifecycleRule {
 pub trait Lifecycle {
     async fn has_transition(&self) -> bool;
     fn has_expiry(&self) -> bool;
-    async fn has_active_rules(&self, prefix: &str) -> bool;
+    fn has_active_rules(&self, prefix: &str) -> bool;
     async fn validate(&self, lr: &ObjectLockConfiguration) -> Result<(), std::io::Error>;
     async fn filter_rules(&self, obj: &ObjectOpts) -> Option<Vec<LifecycleRule>>;
     async fn eval(&self, obj: &ObjectOpts) -> Event;
-    async fn eval_inner(&self, obj: &ObjectOpts, now: OffsetDateTime) -> Event;
+    async fn eval_inner(&self, obj: &ObjectOpts, now: OffsetDateTime, newer_noncurrent_versions: usize) -> Event;
     //fn set_prediction_headers(&self, w: http.ResponseWriter, obj: ObjectOpts);
     async fn noncurrent_versions_expiration_limit(self: Arc<Self>, obj: &ObjectOpts) -> Event;
 }
@@ -159,7 +162,7 @@ impl Lifecycle for BucketLifecycleConfiguration {
         false
     }
 
-    async fn has_active_rules(&self, prefix: &str) -> bool {
+    fn has_active_rules(&self, prefix: &str) -> bool {
         if self.rules.len() == 0 {
             return false;
         }
@@ -273,10 +276,10 @@ impl Lifecycle for BucketLifecycleConfiguration {
     }
 
     async fn eval(&self, obj: &ObjectOpts) -> Event {
-        self.eval_inner(obj, OffsetDateTime::now_utc()).await
+        self.eval_inner(obj, OffsetDateTime::now_utc(), 0).await
     }
 
-    async fn eval_inner(&self, obj: &ObjectOpts, now: OffsetDateTime) -> Event {
+    async fn eval_inner(&self, obj: &ObjectOpts, now: OffsetDateTime, newer_noncurrent_versions: usize) -> Event {
         let mut events = Vec::<Event>::new();
         info!(
             "eval_inner: object={}, mod_time={:?}, now={:?}, is_latest={}, delete_marker={}",
@@ -435,10 +438,10 @@ impl Lifecycle for BucketLifecycleConfiguration {
                     obj.is_latest,
                     obj.delete_marker,
                     obj.version_id,
-                    (obj.is_latest || obj.version_id.is_empty()) && !obj.delete_marker
+                    (obj.is_latest || obj.version_id.is_none_or(|v| v.is_nil())) && !obj.delete_marker
                 );
                 // Allow expiration for latest objects OR non-versioned objects (empty version_id)
-                if (obj.is_latest || obj.version_id.is_empty()) && !obj.delete_marker {
+                if (obj.is_latest || obj.version_id.is_none_or(|v| v.is_nil())) && !obj.delete_marker {
                     info!("eval_inner: entering expiration check");
                     if let Some(ref expiration) = rule.expiration {
                         if let Some(ref date) = expiration.date {
@@ -658,7 +661,7 @@ pub struct ObjectOpts {
     pub user_tags: String,
     pub mod_time: Option<OffsetDateTime>,
     pub size: usize,
-    pub version_id: String,
+    pub version_id: Option<Uuid>,
     pub is_latest: bool,
     pub delete_marker: bool,
     pub num_versions: usize,
@@ -668,12 +671,37 @@ pub struct ObjectOpts {
     pub restore_expires: Option<OffsetDateTime>,
     pub versioned: bool,
     pub version_suspended: bool,
+    pub user_defined: HashMap<String, String>,
+    pub version_purge_status: VersionPurgeStatusType,
+    pub replication_status: ReplicationStatusType,
 }
 
 impl ObjectOpts {
     pub fn expired_object_deletemarker(&self) -> bool {
         self.delete_marker && self.num_versions == 1
     }
+
+    pub fn from_object_info(oi: &ObjectInfo) -> Self {
+        Self {
+            name: oi.name.clone(),
+            user_tags: oi.user_tags.clone(),
+            mod_time: oi.mod_time,
+            size: oi.size as usize,
+            version_id: oi.version_id.clone(),
+            is_latest: oi.is_latest,
+            delete_marker: oi.delete_marker,
+            num_versions: oi.num_versions,
+            successor_mod_time: oi.successor_mod_time,
+            transition_status: oi.transitioned_object.status.clone(),
+            restore_ongoing: oi.restore_ongoing,
+            restore_expires: oi.restore_expires,
+            versioned: false,
+            version_suspended: false,
+            user_defined: oi.user_defined.clone(),
+            version_purge_status: oi.version_purge_status.clone(),
+            replication_status: oi.replication_status.clone(),
+        }
+    }
 }
 
 #[derive(Debug, Clone)]

crates/ecstore/src/bucket/lifecycle/mod.rs

@@ -14,6 +14,7 @@
 
 pub mod bucket_lifecycle_audit;
 pub mod bucket_lifecycle_ops;
+pub mod evaluator;
 pub mod lifecycle;
 pub mod rule;
 pub mod tier_last_day_stats;

crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs

@@ -24,6 +24,7 @@ use crate::global::GLOBAL_TierConfigMgr;
 use sha2::{Digest, Sha256};
 use std::any::Any;
 use std::io::Write;
+use uuid::Uuid;
 use xxhash_rust::xxh64;
 
 static XXHASH_SEED: u64 = 0;
@@ -33,7 +34,7 @@ static XXHASH_SEED: u64 = 0;
 struct ObjSweeper {
     object: String,
     bucket: String,
-    version_id: String,
+    version_id: Option<Uuid>,
     versioned: bool,
     suspended: bool,
     transition_status: String,
@@ -53,8 +54,8 @@ impl ObjSweeper {
         })
     }
 
-    pub fn with_version(&mut self, vid: String) -> &Self {
-        self.version_id = vid;
+    pub fn with_version(&mut self, vid: Option<Uuid>) -> &Self {
+        self.version_id = vid.clone();
         self
     }
 
@@ -71,8 +72,8 @@ impl ObjSweeper {
             version_suspended: self.suspended,
             ..Default::default()
         };
-        if self.suspended && self.version_id == "" {
-            opts.version_id = String::from("");
+        if self.suspended && self.version_id.is_none_or(|v| v.is_nil()) {
+            opts.version_id = None;
         }
         opts
     }
@@ -93,7 +94,7 @@ impl ObjSweeper {
         if !self.versioned || self.suspended {
             // 1, 2.a, 2.b
             del_tier = true;
-        } else if self.versioned && self.version_id != "" {
+        } else if self.versioned && self.version_id.is_some_and(|v| !v.is_nil()) {
             // 3.a
             del_tier = true;
         }

crates/ecstore/src/bucket/metadata_sys.rs

@@ -180,6 +180,13 @@ pub async fn created_at(bucket: &str) -> Result<OffsetDateTime> {
     bucket_meta_sys.created_at(bucket).await
 }
 
+pub async fn list_bucket_targets(bucket: &str) -> Result<BucketTargets> {
+    let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
+    let bucket_meta_sys = bucket_meta_sys_lock.read().await;
+
+    bucket_meta_sys.get_bucket_targets_config(bucket).await
+}
+
 #[derive(Debug)]
 pub struct BucketMetadataSys {
     metadata_map: RwLock<HashMap<String, Arc<BucketMetadata>>>,

crates/ecstore/src/bucket/object_lock/mod.rs

@@ -15,7 +15,7 @@
 pub mod objectlock;
 pub mod objectlock_sys;
 
-use s3s::dto::{ObjectLockConfiguration, ObjectLockEnabled};
+use s3s::dto::{ObjectLockConfiguration, ObjectLockEnabled, ObjectLockLegalHoldStatus};
 
 pub trait ObjectLockApi {
     fn enabled(&self) -> bool;
@@ -28,3 +28,13 @@ impl ObjectLockApi for ObjectLockConfiguration {
             .is_some_and(|v| v.as_str() == ObjectLockEnabled::ENABLED)
     }
 }
+
+pub trait ObjectLockStatusExt {
+    fn valid(&self) -> bool;
+}
+
+impl ObjectLockStatusExt for ObjectLockLegalHoldStatus {
+    fn valid(&self) -> bool {
+        matches!(self.as_str(), ObjectLockLegalHoldStatus::ON | ObjectLockLegalHoldStatus::OFF)
+    }
+}

crates/ecstore/src/bucket/replication/replication_pool.rs

@@ -13,12 +13,15 @@
 // limitations under the License.
 
 use crate::StorageAPI;
+use crate::bucket::bucket_target_sys::BucketTargetSys;
+use crate::bucket::metadata_sys;
 use crate::bucket::replication::ResyncOpts;
 use crate::bucket::replication::ResyncStatusType;
 use crate::bucket::replication::replicate_delete;
 use crate::bucket::replication::replicate_object;
 use crate::bucket::replication::replication_resyncer::{
-    BucketReplicationResyncStatus, DeletedObjectReplicationInfo, ReplicationResyncer,
+    BucketReplicationResyncStatus, DeletedObjectReplicationInfo, ReplicationConfig, ReplicationResyncer,
+    get_heal_replicate_object_info,
 };
 use crate::bucket::replication::replication_state::ReplicationStats;
 use crate::config::com::read_config;
@@ -34,8 +37,10 @@ use rustfs_filemeta::ReplicationStatusType;
 use rustfs_filemeta::ReplicationType;
 use rustfs_filemeta::ReplicationWorkerOperation;
 use rustfs_filemeta::ResyncDecision;
+use rustfs_filemeta::VersionPurgeStatusType;
 use rustfs_filemeta::replication_statuses_map;
 use rustfs_filemeta::version_purge_statuses_map;
+use rustfs_filemeta::{REPLICATE_EXISTING, REPLICATE_HEAL, REPLICATE_HEAL_DELETE};
 use rustfs_utils::http::RESERVED_METADATA_PREFIX_LOWER;
 use std::any::Any;
 use std::sync::Arc;
@@ -1041,3 +1046,152 @@ pub async fn schedule_replication_delete(dv: DeletedObjectReplicationInfo) {
         }
     }
 }
+
+/// QueueReplicationHeal is a wrapper for queue_replication_heal_internal
+pub async fn queue_replication_heal(bucket: &str, oi: ObjectInfo, retry_count: u32) {
+    // ignore modtime zero objects
+    if oi.mod_time.is_none() || oi.mod_time == Some(OffsetDateTime::UNIX_EPOCH) {
+        return;
+    }
+
+    let rcfg = match metadata_sys::get_replication_config(bucket).await {
+        Ok((config, _)) => config,
+        Err(err) => {
+            warn!("Failed to get replication config for bucket {}: {}", bucket, err);
+
+            return;
+        }
+    };
+
+    let tgts = match BucketTargetSys::get().list_bucket_targets(bucket).await {
+        Ok(targets) => Some(targets),
+        Err(err) => {
+            warn!("Failed to list bucket targets for bucket {}: {}", bucket, err);
+            None
+        }
+    };
+
+    let rcfg_wrapper = ReplicationConfig::new(Some(rcfg), tgts);
+    queue_replication_heal_internal(bucket, oi, rcfg_wrapper, retry_count).await;
+}
+
+/// queue_replication_heal_internal enqueues objects that failed replication OR eligible for resyncing through
+/// an ongoing resync operation or via existing objects replication configuration setting.
+pub async fn queue_replication_heal_internal(
+    _bucket: &str,
+    oi: ObjectInfo,
+    rcfg: ReplicationConfig,
+    retry_count: u32,
+) -> ReplicateObjectInfo {
+    let mut roi = ReplicateObjectInfo::default();
+
+    // ignore modtime zero objects
+    if oi.mod_time.is_none() || oi.mod_time == Some(OffsetDateTime::UNIX_EPOCH) {
+        return roi;
+    }
+
+    if rcfg.config.is_none() || rcfg.remotes.is_none() {
+        return roi;
+    }
+
+    roi = get_heal_replicate_object_info(&oi, &rcfg).await;
+    roi.retry_count = retry_count;
+
+    if !roi.dsc.replicate_any() {
+        return roi;
+    }
+
+    // early return if replication already done, otherwise we need to determine if this
+    // version is an existing object that needs healing.
+    if roi.replication_status == ReplicationStatusType::Completed
+        && roi.version_purge_status.is_empty()
+        && !roi.existing_obj_resync.must_resync()
+    {
+        return roi;
+    }
+
+    if roi.delete_marker || !roi.version_purge_status.is_empty() {
+        let (version_id, dm_version_id) = if roi.version_purge_status.is_empty() {
+            (None, roi.version_id)
+        } else {
+            (roi.version_id, None)
+        };
+
+        let dv = DeletedObjectReplicationInfo {
+            delete_object: crate::store_api::DeletedObject {
+                object_name: roi.name.clone(),
+                delete_marker_version_id: dm_version_id,
+                version_id,
+                replication_state: roi.replication_state.clone(),
+                delete_marker_mtime: roi.mod_time,
+                delete_marker: roi.delete_marker,
+                ..Default::default()
+            },
+            bucket: roi.bucket.clone(),
+            op_type: ReplicationType::Heal,
+            event_type: REPLICATE_HEAL_DELETE.to_string(),
+            ..Default::default()
+        };
+
+        // heal delete marker replication failure or versioned delete replication failure
+        if roi.replication_status == ReplicationStatusType::Pending
+            || roi.replication_status == ReplicationStatusType::Failed
+            || roi.version_purge_status == VersionPurgeStatusType::Failed
+            || roi.version_purge_status == VersionPurgeStatusType::Pending
+        {
+            if let Some(pool) = GLOBAL_REPLICATION_POOL.get() {
+                pool.queue_replica_delete_task(dv).await;
+            }
+            return roi;
+        }
+
+        // if replication status is Complete on DeleteMarker and existing object resync required
+        let existing_obj_resync = roi.existing_obj_resync.clone();
+        if existing_obj_resync.must_resync()
+            && (roi.replication_status == ReplicationStatusType::Completed || roi.replication_status.is_empty())
+        {
+            queue_replicate_deletes_wrapper(dv, existing_obj_resync).await;
+            return roi;
+        }
+
+        return roi;
+    }
+
+    if roi.existing_obj_resync.must_resync() {
+        roi.op_type = ReplicationType::ExistingObject;
+    }
+
+    match roi.replication_status {
+        ReplicationStatusType::Pending | ReplicationStatusType::Failed => {
+            roi.event_type = REPLICATE_HEAL.to_string();
+            if let Some(pool) = GLOBAL_REPLICATION_POOL.get() {
+                pool.queue_replica_task(roi.clone()).await;
+            }
+            return roi;
+        }
+        _ => {}
+    }
+
+    if roi.existing_obj_resync.must_resync() {
+        roi.event_type = REPLICATE_EXISTING.to_string();
+        if let Some(pool) = GLOBAL_REPLICATION_POOL.get() {
+            pool.queue_replica_task(roi.clone()).await;
+        }
+    }
+
+    roi
+}
+
+/// Wrapper function for queueing replicate deletes with resync decision
+async fn queue_replicate_deletes_wrapper(doi: DeletedObjectReplicationInfo, existing_obj_resync: ResyncDecision) {
+    for (k, v) in existing_obj_resync.targets.iter() {
+        if v.replicate {
+            let mut dv = doi.clone();
+            dv.reset_id = v.reset_id.clone();
+            dv.target_arn = k.clone();
+            if let Some(pool) = GLOBAL_REPLICATION_POOL.get() {
+                pool.queue_replica_delete_task(dv).await;
+            }
+        }
+    }
+}

crates/ecstore/src/bucket/replication/replication_resyncer.rs

@@ -754,7 +754,7 @@ impl ReplicationWorkerOperation for DeletedObjectReplicationInfo {
     }
 }
 
-#[derive(Debug, Clone, Default)]
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
 pub struct ReplicationConfig {
     pub config: Option<ReplicationConfiguration>,
     pub remotes: Option<BucketTargets>,
@@ -2174,35 +2174,122 @@ fn is_standard_header(k: &str) -> bool {
     STANDARD_HEADERS.iter().any(|h| h.eq_ignore_ascii_case(k))
 }
 
+// Valid SSE replication headers mapping from internal to replication headers
+static VALID_SSE_REPLICATION_HEADERS: &[(&str, &str)] = &[
+    (
+        "X-Rustfs-Internal-Server-Side-Encryption-Sealed-Key",
+        "X-Rustfs-Replication-Server-Side-Encryption-Sealed-Key",
+    ),
+    (
+        "X-Rustfs-Internal-Server-Side-Encryption-Seal-Algorithm",
+        "X-Rustfs-Replication-Server-Side-Encryption-Seal-Algorithm",
+    ),
+    (
+        "X-Rustfs-Internal-Server-Side-Encryption-Iv",
+        "X-Rustfs-Replication-Server-Side-Encryption-Iv",
+    ),
+    ("X-Rustfs-Internal-Encrypted-Multipart", "X-Rustfs-Replication-Encrypted-Multipart"),
+    ("X-Rustfs-Internal-Actual-Object-Size", "X-Rustfs-Replication-Actual-Object-Size"),
+];
+
+const REPLICATION_SSEC_CHECKSUM_HEADER: &str = "X-Rustfs-Replication-Ssec-Crc";
+
+fn is_valid_sse_header(k: &str) -> Option<&str> {
+    VALID_SSE_REPLICATION_HEADERS
+        .iter()
+        .find(|(internal, _)| k.eq_ignore_ascii_case(internal))
+        .map(|(_, replication)| *replication)
+}
+
 fn put_replication_opts(sc: &str, object_info: &ObjectInfo) -> Result<(PutObjectOptions, bool)> {
+    use crate::config::storageclass::{RRS, STANDARD};
+    use base64::{Engine, engine::general_purpose::STANDARD as BASE64_STANDARD};
+    use rustfs_utils::http::{
+        AMZ_CHECKSUM_TYPE, AMZ_CHECKSUM_TYPE_FULL_OBJECT, AMZ_SERVER_SIDE_ENCRYPTION, AMZ_SERVER_SIDE_ENCRYPTION_KMS_ID,
+    };
+
     let mut meta = HashMap::new();
+    let is_ssec = is_ssec_encrypted(&object_info.user_defined);
 
+    // Process user-defined metadata
     for (k, v) in object_info.user_defined.iter() {
-        if strings_has_prefix_fold(k, RESERVED_METADATA_PREFIX) {
-            continue;
+        let has_valid_sse_header = is_valid_sse_header(k).is_some();
+
+        // In case of SSE-C objects copy the allowed internal headers as well
+        if !is_ssec || !has_valid_sse_header {
+            if strings_has_prefix_fold(k, RESERVED_METADATA_PREFIX) {
+                continue;
+            }
+            if is_standard_header(k) {
+                continue;
+            }
         }
 
-        if is_standard_header(k) {
-            continue;
+        if let Some(replication_header) = is_valid_sse_header(k) {
+            meta.insert(replication_header.to_string(), v.to_string());
+        } else {
+            meta.insert(k.to_string(), v.to_string());
         }
-
-        meta.insert(k.to_string(), v.to_string());
     }
 
-    let is_multipart = object_info.is_multipart();
+    let mut is_multipart = object_info.is_multipart();
+
+    // Handle checksum
+    if let Some(checksum_data) = &object_info.checksum
+        && !checksum_data.is_empty()
+    {
+        // Add encrypted CRC to metadata for SSE-C objects
+        if is_ssec {
+            let encoded = BASE64_STANDARD.encode(checksum_data);
+            meta.insert(REPLICATION_SSEC_CHECKSUM_HEADER.to_string(), encoded);
+        } else {
+            // Get checksum metadata for non-SSE-C objects
+            let (cs_meta, is_mp) = object_info.decrypt_checksums(0, &http::HeaderMap::new())?;
+            is_multipart = is_mp;
+
+            // Set object checksum metadata
+            for (k, v) in cs_meta.iter() {
+                if k != AMZ_CHECKSUM_TYPE {
+                    meta.insert(k.clone(), v.clone());
+                }
+            }
+
+            // For objects where checksum is full object, use the cheaper PutObject replication
+            if !object_info.is_multipart()
+                && cs_meta
+                    .get(AMZ_CHECKSUM_TYPE)
+                    .map(|v| v.as_str() == AMZ_CHECKSUM_TYPE_FULL_OBJECT)
+                    .unwrap_or(false)
+            {
+                is_multipart = false;
+            }
+        }
+    }
+
+    // Handle storage class default
+    let storage_class = if sc.is_empty() {
+        let obj_sc = object_info.storage_class.as_deref().unwrap_or_default();
+        if obj_sc == STANDARD || obj_sc == RRS {
+            obj_sc.to_string()
+        } else {
+            sc.to_string()
+        }
+    } else {
+        sc.to_string()
+    };
 
     let mut put_op = PutObjectOptions {
         user_metadata: meta,
         content_type: object_info.content_type.clone().unwrap_or_default(),
         content_encoding: object_info.content_encoding.clone().unwrap_or_default(),
         expires: object_info.expires.unwrap_or(OffsetDateTime::UNIX_EPOCH),
-        storage_class: sc.to_string(),
+        storage_class,
         internal: AdvancedPutOptions {
             source_version_id: object_info.version_id.map(|v| v.to_string()).unwrap_or_default(),
             source_etag: object_info.etag.clone().unwrap_or_default(),
             source_mtime: object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH),
-            replication_status: ReplicationStatusType::Pending,
-            replication_request: true,
+            replication_status: ReplicationStatusType::Replica, // Changed from Pending to Replica
+            replication_request: true, // always set this to distinguish between replication and normal PUT operation
             ..Default::default()
         },
         ..Default::default()
@@ -2213,36 +2300,43 @@ fn put_replication_opts(sc: &str, object_info: &ObjectInfo) -> Result<(PutObject
 
         if !tags.is_empty() {
             put_op.user_tags = tags;
+            // set tag timestamp in opts
             put_op.internal.tagging_timestamp = if let Some(ts) = object_info
                 .user_defined
-                .get(&format!("{RESERVED_METADATA_PREFIX}tagging-timestamp"))
+                .get(&format!("{RESERVED_METADATA_PREFIX_LOWER}tagging-timestamp"))
             {
-                OffsetDateTime::parse(ts, &Rfc3339).unwrap_or(OffsetDateTime::UNIX_EPOCH)
+                OffsetDateTime::parse(ts, &Rfc3339)
+                    .map_err(|e| Error::other(format!("Failed to parse tagging timestamp: {}", e)))?
             } else {
                 object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH)
             };
         }
     }
 
-    if let Some(lang) = object_info.user_defined.lookup(headers::CONTENT_LANGUAGE) {
+    // Use case-insensitive lookup for headers
+    let lk_map = object_info.user_defined.clone();
+
+    if let Some(lang) = lk_map.lookup(headers::CONTENT_LANGUAGE) {
         put_op.content_language = lang.to_string();
     }
 
-    if let Some(cd) = object_info.user_defined.lookup(headers::CONTENT_DISPOSITION) {
+    if let Some(cd) = lk_map.lookup(headers::CONTENT_DISPOSITION) {
         put_op.content_disposition = cd.to_string();
     }
 
-    if let Some(v) = object_info.user_defined.lookup(headers::CACHE_CONTROL) {
+    if let Some(v) = lk_map.lookup(headers::CACHE_CONTROL) {
         put_op.cache_control = v.to_string();
     }
 
-    if let Some(v) = object_info.user_defined.lookup(headers::AMZ_OBJECT_LOCK_MODE) {
+    if let Some(v) = lk_map.lookup(headers::AMZ_OBJECT_LOCK_MODE) {
         let mode = v.to_string().to_uppercase();
         put_op.mode = Some(aws_sdk_s3::types::ObjectLockRetentionMode::from(mode.as_str()));
     }
 
-    if let Some(v) = object_info.user_defined.lookup(headers::AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE) {
-        put_op.retain_until_date = OffsetDateTime::parse(v, &Rfc3339).unwrap_or(OffsetDateTime::UNIX_EPOCH);
+    if let Some(v) = lk_map.lookup(headers::AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE) {
+        put_op.retain_until_date =
+            OffsetDateTime::parse(v, &Rfc3339).map_err(|e| Error::other(format!("Failed to parse retain until date: {}", e)))?;
+        // set retention timestamp in opts
         put_op.internal.retention_timestamp = if let Some(v) = object_info
             .user_defined
             .get(&format!("{RESERVED_METADATA_PREFIX_LOWER}objectlock-retention-timestamp"))
@@ -2253,9 +2347,10 @@ fn put_replication_opts(sc: &str, object_info: &ObjectInfo) -> Result<(PutObject
         };
     }
 
-    if let Some(v) = object_info.user_defined.lookup(headers::AMZ_OBJECT_LOCK_LEGAL_HOLD) {
+    if let Some(v) = lk_map.lookup(headers::AMZ_OBJECT_LOCK_LEGAL_HOLD) {
         let hold = v.to_uppercase();
         put_op.legalhold = Some(ObjectLockLegalHoldStatus::from(hold.as_str()));
+        // set legalhold timestamp in opts
         put_op.internal.legalhold_timestamp = if let Some(v) = object_info
             .user_defined
             .get(&format!("{RESERVED_METADATA_PREFIX_LOWER}objectlock-legalhold-timestamp"))
@@ -2266,7 +2361,34 @@ fn put_replication_opts(sc: &str, object_info: &ObjectInfo) -> Result<(PutObject
         };
     }
 
-    // TODO: is encrypted
+    // Handle SSE-S3 encryption
+    if object_info
+        .user_defined
+        .get(AMZ_SERVER_SIDE_ENCRYPTION)
+        .map(|v| v.eq_ignore_ascii_case("AES256"))
+        .unwrap_or(false)
+    {
+        // SSE-S3 detected - set ServerSideEncryption
+        // Note: This requires the PutObjectOptions to support SSE
+        // TODO: Implement SSE-S3 support in PutObjectOptions if not already present
+    }
+
+    // Handle SSE-KMS encryption
+    if object_info.user_defined.contains_key(AMZ_SERVER_SIDE_ENCRYPTION_KMS_ID) {
+        // SSE-KMS detected
+        // If KMS key ID replication is enabled (as by default)
+        // we include the object's KMS key ID. In any case, we
+        // always set the SSE-KMS header. If no KMS key ID is
+        // specified, MinIO is supposed to use whatever default
+        // config applies on the site or bucket.
+        // TODO: Implement SSE-KMS support with key ID replication
+        // let key_id = if kms::replicate_key_id() {
+        //     object_info.kms_key_id()
+        // } else {
+        //     None
+        // };
+        // TODO: Set SSE-KMS encryption in put_op
+    }
 
     Ok((put_op, is_multipart))
 }

crates/ecstore/src/disk/disk_store.rs

@@ -15,7 +15,8 @@
 use crate::disk::{
     CheckPartsResp, DeleteOptions, DiskAPI, DiskError, DiskInfo, DiskInfoOptions, DiskLocation, Endpoint, Error,
     FileInfoVersions, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, Result, UpdateMetadataOpts, VolumeInfo,
-    WalkDirOptions, local::LocalDisk,
+    WalkDirOptions,
+    local::{LocalDisk, ScanGuard},
 };
 use bytes::Bytes;
 use rustfs_filemeta::{FileInfo, ObjectPartInfo, RawFileInfo};
@@ -259,6 +260,7 @@ impl LocalDiskWrapper {
                     let test_obj = format!("health-check-{}", Uuid::new_v4());
                     if Self::perform_health_check(disk.clone(), &TEST_BUCKET, &test_obj, &TEST_DATA, true, CHECK_TIMEOUT_DURATION).await.is_err() && health.swap_ok_to_faulty() {
                         // Health check failed, disk is considered faulty
+                        warn!("health check: failed, disk is considered faulty");
 
                         health.increment_waiting(); // Balance the increment from failed operation
 
@@ -429,7 +431,7 @@ impl LocalDiskWrapper {
     {
         // Check if disk is faulty
         if self.health.is_faulty() {
-            warn!("disk {} health is faulty, returning error", self.to_string());
+            warn!("local disk {} health is faulty, returning error", self.to_string());
             return Err(DiskError::FaultyDisk);
         }
 
@@ -476,6 +478,15 @@ impl LocalDiskWrapper {
 
 #[async_trait::async_trait]
 impl DiskAPI for LocalDiskWrapper {
+    async fn read_metadata(&self, volume: &str, path: &str) -> Result<Bytes> {
+        self.track_disk_health(|| async { self.disk.read_metadata(volume, path).await }, Duration::ZERO)
+            .await
+    }
+
+    fn start_scan(&self) -> ScanGuard {
+        self.disk.start_scan()
+    }
+
     fn to_string(&self) -> String {
         self.disk.to_string()
     }

crates/ecstore/src/disk/local.rs

@@ -82,7 +82,7 @@ pub struct LocalDisk {
     pub format_info: RwLock<FormatInfo>,
     pub endpoint: Endpoint,
     pub disk_info_cache: Arc<Cache<DiskInfo>>,
-    pub scanning: AtomicU32,
+    pub scanning: Arc<AtomicU32>,
     pub rotational: bool,
     pub fstype: String,
     pub major: u64,
@@ -209,7 +209,7 @@ impl LocalDisk {
             format_path,
             format_info: RwLock::new(format_info),
             disk_info_cache: Arc::new(cache),
-            scanning: AtomicU32::new(0),
+            scanning: Arc::new(AtomicU32::new(0)),
             rotational: Default::default(),
             fstype: Default::default(),
             minor: Default::default(),
@@ -664,6 +664,7 @@ impl LocalDisk {
                 match self.read_metadata_with_dmtime(meta_path).await {
                     Ok(res) => Ok(res),
                     Err(err) => {
+                        warn!("read_raw: error: {:?}", err);
                         if err == Error::FileNotFound
                             && !skip_access_checks(volume_dir.as_ref().to_string_lossy().to_string().as_str())
                             && let Err(e) = access(volume_dir.as_ref()).await
@@ -687,20 +688,6 @@ impl LocalDisk {
         Ok((buf, mtime))
     }
 
-    async fn read_metadata(&self, file_path: impl AsRef<Path>) -> Result<Vec<u8>> {
-        // Try to use cached file content reading for better performance, with safe fallback
-        let path = file_path.as_ref().to_path_buf();
-
-        // First, try the cache
-        if let Ok(bytes) = get_global_file_cache().get_file_content(path.clone()).await {
-            return Ok(bytes.to_vec());
-        }
-
-        // Fallback to direct read if cache fails
-        let (data, _) = self.read_metadata_with_dmtime(file_path.as_ref()).await?;
-        Ok(data)
-    }
-
     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())?;
 
@@ -1052,7 +1039,7 @@ impl LocalDisk {
 
             if entry.ends_with(STORAGE_FORMAT_FILE) {
                 let metadata = self
-                    .read_metadata(self.get_object_path(bucket, format!("{}/{}", &current, &entry).as_str())?)
+                    .read_metadata(bucket, format!("{}/{}", &current, &entry).as_str())
                     .await?;
 
                 let entry = entry.strip_suffix(STORAGE_FORMAT_FILE).unwrap_or_default().to_owned();
@@ -1068,7 +1055,7 @@ impl LocalDisk {
 
                 out.write_obj(&MetaCacheEntry {
                     name: name.clone(),
-                    metadata,
+                    metadata: metadata.to_vec(),
                     ..Default::default()
                 })
                 .await?;
@@ -1135,14 +1122,14 @@ impl LocalDisk {
 
             let fname = format!("{}/{}", &meta.name, STORAGE_FORMAT_FILE);
 
-            match self.read_metadata(self.get_object_path(&opts.bucket, fname.as_str())?).await {
+            match self.read_metadata(&opts.bucket, fname.as_str()).await {
                 Ok(res) => {
                     if is_dir_obj {
                         meta.name = meta.name.trim_end_matches(GLOBAL_DIR_SUFFIX_WITH_SLASH).to_owned();
                         meta.name.push_str(SLASH_SEPARATOR_STR);
                     }
 
-                    meta.metadata = res;
+                    meta.metadata = res.to_vec();
 
                     out.write_obj(&meta).await?;
 
@@ -1189,6 +1176,14 @@ impl LocalDisk {
     }
 }
 
+pub struct ScanGuard(pub Arc<AtomicU32>);
+
+impl Drop for ScanGuard {
+    fn drop(&mut self) {
+        self.0.fetch_sub(1, Ordering::Release);
+    }
+}
+
 fn is_root_path(path: impl AsRef<Path>) -> bool {
     path.as_ref().components().count() == 1 && path.as_ref().has_root()
 }
@@ -1579,6 +1574,8 @@ impl DiskAPI for LocalDisk {
                 .as_str(),
             )?;
 
+            info!("check_parts: part_path: {:?}", &part_path);
+
             match lstat(&part_path).await {
                 Ok(st) => {
                     if st.is_dir() {
@@ -1593,6 +1590,8 @@ impl DiskAPI for LocalDisk {
                     resp.results[i] = CHECK_PART_SUCCESS;
                 }
                 Err(err) => {
+                    info!("check_parts: failed to stat file: {:?}, error: {:?}", &part_path, &err);
+
                     let e: DiskError = to_file_error(err).into();
 
                     if e == DiskError::FileNotFound {
@@ -1882,19 +1881,20 @@ impl DiskAPI for LocalDisk {
         let mut objs_returned = 0;
 
         if opts.base_dir.ends_with(SLASH_SEPARATOR_STR) {
-            let fpath = self.get_object_path(
-                &opts.bucket,
-                path_join_buf(&[
-                    format!("{}{}", opts.base_dir.trim_end_matches(SLASH_SEPARATOR_STR), GLOBAL_DIR_SUFFIX).as_str(),
-                    STORAGE_FORMAT_FILE,
-                ])
-                .as_str(),
-            )?;
-
-            if let Ok(data) = self.read_metadata(fpath).await {
+            if let Ok(data) = self
+                .read_metadata(
+                    &opts.bucket,
+                    path_join_buf(&[
+                        format!("{}{}", opts.base_dir.trim_end_matches(SLASH_SEPARATOR_STR), GLOBAL_DIR_SUFFIX).as_str(),
+                        STORAGE_FORMAT_FILE,
+                    ])
+                    .as_str(),
+                )
+                .await
+            {
                 let meta = MetaCacheEntry {
                     name: opts.base_dir.clone(),
-                    metadata: data,
+                    metadata: data.to_vec(),
                     ..Default::default()
                 };
                 out.write_obj(&meta).await?;
@@ -2541,7 +2541,7 @@ impl DiskAPI for LocalDisk {
         info.rotational = self.rotational;
         info.mount_path = self.path().to_str().unwrap().to_string();
         info.endpoint = self.endpoint.to_string();
-        info.scanning = self.scanning.load(Ordering::SeqCst) == 1;
+        info.scanning = self.scanning.load(Ordering::Acquire) == 1;
 
         if info.id.is_none() {
             info.id = self.get_disk_id().await.unwrap_or(None);
@@ -2549,6 +2549,29 @@ impl DiskAPI for LocalDisk {
 
         Ok(info)
     }
+    #[tracing::instrument(skip(self))]
+    fn start_scan(&self) -> ScanGuard {
+        self.scanning.fetch_add(1, Ordering::Release);
+        ScanGuard(Arc::clone(&self.scanning))
+    }
+
+    async fn read_metadata(&self, volume: &str, path: &str) -> Result<Bytes> {
+        // Try to use cached file content reading for better performance, with safe fallback
+        let file_path = self.get_object_path(volume, path)?;
+        // let file_path = file_path.join(Path::new(STORAGE_FORMAT_FILE));
+
+        // First, try the cache
+        if let Ok(bytes) = get_global_file_cache().get_file_content(file_path.clone()).await {
+            return Ok(bytes);
+        }
+
+        // Fallback to direct read if cache fails
+        let (data, _) = self.read_metadata_with_dmtime(&file_path).await.map_err(|e| {
+            error!("read_metadata: error: {:?}, file_path={}", e, file_path.to_string_lossy());
+            e
+        })?;
+        Ok(data.into())
+    }
 }
 
 async fn get_disk_info(drive_path: PathBuf) -> Result<(rustfs_utils::os::DiskInfo, bool)> {

crates/ecstore/src/disk/mod.rs

@@ -32,6 +32,7 @@ pub const STORAGE_FORMAT_FILE: &str = "xl.meta";
 pub const STORAGE_FORMAT_FILE_BACKUP: &str = "xl.meta.bkp";
 
 use crate::disk::disk_store::LocalDiskWrapper;
+use crate::disk::local::ScanGuard;
 use crate::rpc::RemoteDisk;
 use bytes::Bytes;
 use endpoint::Endpoint;
@@ -395,6 +396,20 @@ impl DiskAPI for Disk {
             Disk::Remote(remote_disk) => remote_disk.disk_info(opts).await,
         }
     }
+
+    fn start_scan(&self) -> ScanGuard {
+        match self {
+            Disk::Local(local_disk) => local_disk.start_scan(),
+            Disk::Remote(remote_disk) => remote_disk.start_scan(),
+        }
+    }
+
+    async fn read_metadata(&self, volume: &str, path: &str) -> Result<Bytes> {
+        match self {
+            Disk::Local(local_disk) => local_disk.read_metadata(volume, path).await,
+            Disk::Remote(remote_disk) => remote_disk.read_metadata(volume, path).await,
+        }
+    }
 }
 
 pub async fn new_disk(ep: &Endpoint, opt: &DiskOption) -> Result<DiskStore> {
@@ -458,6 +473,7 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
         opts: &ReadOptions,
     ) -> Result<FileInfo>;
     async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result<RawFileInfo>;
+    async fn read_metadata(&self, volume: &str, path: &str) -> Result<Bytes>;
     async fn rename_data(
         &self,
         src_volume: &str,
@@ -489,6 +505,7 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
     async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> Result<()>;
     async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes>;
     async fn disk_info(&self, opts: &DiskInfoOptions) -> Result<DiskInfo>;
+    fn start_scan(&self) -> ScanGuard;
 }
 
 #[derive(Debug, Default, Serialize, Deserialize)]
@@ -694,6 +711,10 @@ pub fn has_part_err(part_errs: &[usize]) -> bool {
     part_errs.iter().any(|err| *err != CHECK_PART_SUCCESS)
 }
 
+pub fn count_part_not_success(part_errs: &[usize]) -> usize {
+    part_errs.iter().filter(|err| **err != CHECK_PART_SUCCESS).count()
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/ecstore/src/pools.rs

@@ -442,11 +442,11 @@ impl PoolMeta {
     }
 }
 
-fn path2_bucket_object(name: &str) -> (String, String) {
+pub fn path2_bucket_object(name: &str) -> (String, String) {
     path2_bucket_object_with_base_path("", name)
 }
 
-fn path2_bucket_object_with_base_path(base_path: &str, path: &str) -> (String, String) {
+pub fn path2_bucket_object_with_base_path(base_path: &str, path: &str) -> (String, String) {
     // Trim the base path and leading slash
     let trimmed_path = path
         .strip_prefix(base_path)
@@ -454,7 +454,9 @@ fn path2_bucket_object_with_base_path(base_path: &str, path: &str) -> (String, S
         .strip_prefix(SLASH_SEPARATOR_STR)
         .unwrap_or(path);
     // Find the position of the first '/'
-    let pos = trimmed_path.find(SLASH_SEPARATOR_STR).unwrap_or(trimmed_path.len());
+    let Some(pos) = trimmed_path.find(SLASH_SEPARATOR_STR) else {
+        return (trimmed_path.to_string(), "".to_string());
+    };
     // Split into bucket and prefix
     let bucket = &trimmed_path[0..pos];
     let prefix = &trimmed_path[pos + 1..]; // +1 to skip the '/' character if it exists

crates/ecstore/src/rpc/remote_disk.rs

@@ -12,23 +12,19 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::{
-    disk::{
-        CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, FileInfoVersions,
-        FileReader, FileWriter, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, UpdateMetadataOpts, VolumeInfo,
-        WalkDirOptions,
-        disk_store::{
-            CHECK_EVERY, CHECK_TIMEOUT_DURATION, ENV_RUSTFS_DRIVE_ACTIVE_MONITORING, SKIP_IF_SUCCESS_BEFORE,
-            get_max_timeout_duration,
-        },
-        endpoint::Endpoint,
-        {
-            disk_store::DiskHealthTracker,
-            error::{DiskError, Error, Result},
-        },
+use crate::disk::{
+    CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, FileInfoVersions, FileReader,
+    FileWriter, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
+    disk_store::{
+        CHECK_EVERY, CHECK_TIMEOUT_DURATION, ENV_RUSTFS_DRIVE_ACTIVE_MONITORING, SKIP_IF_SUCCESS_BEFORE, get_max_timeout_duration,
     },
+    endpoint::Endpoint,
+};
+use crate::disk::{disk_store::DiskHealthTracker, error::DiskError, local::ScanGuard};
+use crate::rpc::client::{TonicInterceptor, gen_tonic_signature_interceptor, node_service_time_out_client};
+use crate::{
+    disk::error::{Error, Result},
     rpc::build_auth_headers,
-    rpc::client::{TonicInterceptor, gen_tonic_signature_interceptor, node_service_time_out_client},
 };
 use bytes::Bytes;
 use futures::lock::Mutex;
@@ -38,15 +34,18 @@ use rustfs_protos::proto_gen::node_service::RenamePartRequest;
 use rustfs_protos::proto_gen::node_service::{
     CheckPartsRequest, DeletePathsRequest, DeleteRequest, DeleteVersionRequest, DeleteVersionsRequest, DeleteVolumeRequest,
     DiskInfoRequest, ListDirRequest, ListVolumesRequest, MakeVolumeRequest, MakeVolumesRequest, ReadAllRequest,
-    ReadMultipleRequest, ReadPartsRequest, ReadVersionRequest, ReadXlRequest, RenameDataRequest, RenameFileRequest,
-    StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest, WriteMetadataRequest,
+    ReadMetadataRequest, ReadMultipleRequest, ReadPartsRequest, ReadVersionRequest, ReadXlRequest, RenameDataRequest,
+    RenameFileRequest, StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest, WriteMetadataRequest,
     node_service_client::NodeServiceClient,
 };
 use rustfs_rio::{HttpReader, HttpWriter};
 use rustfs_utils::string::parse_bool_with_default;
 use std::{
     path::PathBuf,
-    sync::{Arc, atomic::Ordering},
+    sync::{
+        Arc,
+        atomic::{AtomicU32, Ordering},
+    },
     time::Duration,
 };
 use tokio::time;
@@ -60,9 +59,8 @@ use uuid::Uuid;
 pub struct RemoteDisk {
     pub id: Mutex<Option<Uuid>>,
     pub addr: String,
-    pub url: url::Url,
-    pub root: PathBuf,
     endpoint: Endpoint,
+    pub scanning: Arc<AtomicU32>,
     /// Whether health checking is enabled
     health_check: bool,
     /// Health tracker for connection monitoring
@@ -73,8 +71,6 @@ pub struct RemoteDisk {
 
 impl RemoteDisk {
     pub async fn new(ep: &Endpoint, opt: &DiskOption) -> Result<Self> {
-        // let root = fs::canonicalize(ep.url.path()).await?;
-        let root = PathBuf::from(ep.get_file_path());
         let addr = if let Some(port) = ep.url.port() {
             format!("{}://{}:{}", ep.url.scheme(), ep.url.host_str().unwrap(), port)
         } else {
@@ -88,9 +84,8 @@ impl RemoteDisk {
         let disk = Self {
             id: Mutex::new(None),
             addr: addr.clone(),
-            url: ep.url.clone(),
-            root,
             endpoint: ep.clone(),
+            scanning: Arc::new(AtomicU32::new(0)),
             health_check: opt.health_check && env_health_check,
             health: Arc::new(DiskHealthTracker::new()),
             cancel_token: CancellationToken::new(),
@@ -227,7 +222,7 @@ impl RemoteDisk {
     {
         // Check if disk is faulty
         if self.health.is_faulty() {
-            warn!("disk {} health is faulty, returning error", self.to_string());
+            warn!("remote disk {} health is faulty, returning error", self.to_string());
             return Err(DiskError::FaultyDisk);
         }
 
@@ -313,7 +308,7 @@ impl DiskAPI for RemoteDisk {
 
     #[tracing::instrument(skip(self))]
     fn path(&self) -> PathBuf {
-        self.root.clone()
+        PathBuf::from(self.endpoint.get_file_path())
     }
 
     #[tracing::instrument(skip(self))]
@@ -740,6 +735,26 @@ impl DiskAPI for RemoteDisk {
         .await
     }
 
+    async fn read_metadata(&self, volume: &str, path: &str) -> Result<Bytes> {
+        let mut client = self
+            .get_client()
+            .await
+            .map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
+        let request = Request::new(ReadMetadataRequest {
+            volume: volume.to_string(),
+            path: path.to_string(),
+            disk: self.endpoint.to_string(),
+        });
+
+        let response = client.read_metadata(request).await?.into_inner();
+
+        if !response.success {
+            return Err(response.error.unwrap_or_default().into());
+        }
+
+        Ok(response.data)
+    }
+
     #[tracing::instrument(skip(self))]
     async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: &UpdateMetadataOpts) -> Result<()> {
         info!("update_metadata");
@@ -1360,6 +1375,12 @@ impl DiskAPI for RemoteDisk {
 
         Ok(disk_info)
     }
+
+    #[tracing::instrument(skip(self))]
+    fn start_scan(&self) -> ScanGuard {
+        self.scanning.fetch_add(1, Ordering::Relaxed);
+        ScanGuard(Arc::clone(&self.scanning))
+    }
 }
 
 #[cfg(test)]

crates/ecstore/src/sets.rs

@@ -42,6 +42,7 @@ use rustfs_common::{
     heal_channel::{DriveState, HealItemType},
 };
 use rustfs_filemeta::FileInfo;
+use rustfs_lock::FastLockGuard;
 use rustfs_madmin::heal_commands::{HealDriveInfo, HealResultItem};
 use rustfs_utils::{crc_hash, path::path_join_buf, sip_hash};
 use std::{collections::HashMap, sync::Arc};
@@ -363,6 +364,10 @@ impl ObjectIO for Sets {
 
 #[async_trait::async_trait]
 impl StorageAPI for Sets {
+    #[tracing::instrument(skip(self))]
+    async fn new_ns_lock(&self, bucket: &str, object: &str) -> Result<FastLockGuard> {
+        self.disk_set[0].new_ns_lock(bucket, object).await
+    }
     #[tracing::instrument(skip(self))]
     async fn backend_info(&self) -> rustfs_madmin::BackendInfo {
         unimplemented!()

crates/ecstore/src/store.rs

@@ -69,6 +69,7 @@ use rand::Rng as _;
 use rustfs_common::heal_channel::{HealItemType, HealOpts};
 use rustfs_common::{GLOBAL_LOCAL_NODE_NAME, GLOBAL_RUSTFS_HOST, GLOBAL_RUSTFS_PORT};
 use rustfs_filemeta::FileInfo;
+use rustfs_lock::FastLockGuard;
 use rustfs_madmin::heal_commands::HealResultItem;
 use rustfs_utils::path::{decode_dir_object, encode_dir_object, path_join_buf};
 use s3s::dto::{BucketVersioningStatus, ObjectLockConfiguration, ObjectLockEnabled, VersioningConfiguration};
@@ -1203,6 +1204,10 @@ lazy_static! {
 
 #[async_trait::async_trait]
 impl StorageAPI for ECStore {
+    #[instrument(skip(self))]
+    async fn new_ns_lock(&self, bucket: &str, object: &str) -> Result<FastLockGuard> {
+        self.pools[0].new_ns_lock(bucket, object).await
+    }
     #[instrument(skip(self))]
     async fn backend_info(&self) -> rustfs_madmin::BackendInfo {
         let (standard_sc_parity, rr_sc_parity) = {

crates/ecstore/src/store_api.rs

@@ -31,6 +31,7 @@ use rustfs_filemeta::{
     ReplicationStatusType, RestoreStatusOps as _, VersionPurgeStatusType, parse_restore_obj_status, replication_statuses_map,
     version_purge_statuses_map,
 };
+use rustfs_lock::FastLockGuard;
 use rustfs_madmin::heal_commands::HealResultItem;
 use rustfs_rio::Checksum;
 use rustfs_rio::{DecompressReader, HashReader, LimitReader, WarpReader};
@@ -1349,6 +1350,7 @@ pub trait ObjectIO: Send + Sync + Debug + 'static {
 #[allow(clippy::too_many_arguments)]
 pub trait StorageAPI: ObjectIO + Debug {
     // NewNSLock TODO:
+    async fn new_ns_lock(&self, bucket: &str, object: &str) -> Result<FastLockGuard>;
     // Shutdown TODO:
     // NSScanner TODO: