Fix: add support for additional AWS S3 storage classes and validation logic (#487)

* Fix: add pagination fields to S3 response

* Fix: add support for additional AWS S3 storage classes and validation logic

* Fix: improve handling of optional fields in S3 response

---------

Co-authored-by: DamonXue <damonxue2@gmail.com>

.vscode/launch.json

@@ -85,6 +85,19 @@
             "sourceLanguages": [
                 "rust"
             ],
+        },
+        {
+            "name": "Debug executable target/debug/test",
+            "type":  "lldb",
+            "request": "launch",
+            "program": "${workspaceFolder}/target/debug/deps/lifecycle_integration_test-5eb7590b8f3bea55",
+            "args": [],
+            "cwd": "${workspaceFolder}",
+            //"stopAtEntry": false,
+            //"preLaunchTask": "cargo build",
+            "sourceLanguages": [
+                "rust"
+            ],
         }
     ]
 }

Cargo.toml

@@ -97,7 +97,7 @@ async-recursion = "1.1.1"
 async-trait = "0.1.89"
 async-compression = { version = "0.4.19" }
 atomic_enum = "0.3.0"
-aws-config = { version = "1.8.5" }
+aws-config = { version = "1.8.6" }
 aws-sdk-s3 = "1.101.0"
 axum = "0.8.4"
 base64-simd = "0.8.0"
@@ -106,22 +106,22 @@ brotli = "8.0.2"
 bytes = { version = "1.10.1", features = ["serde"] }
 bytesize = "2.0.1"
 byteorder = "1.5.0"
-cfg-if = "1.0.1"
-crc-fast = "1.4.0"
+cfg-if = "1.0.3"
+crc-fast = "1.5.0"
 chacha20poly1305 = { version = "0.10.1" }
 chrono = { version = "0.4.41", features = ["serde"] }
-clap = { version = "4.5.45", features = ["derive", "env"] }
+clap = { version = "4.5.46", features = ["derive", "env"] }
 const-str = { version = "0.6.4", features = ["std", "proc"] }
 crc32fast = "1.5.0"
 criterion = { version = "0.7", features = ["html_reports"] }
 dashmap = "6.1.0"
 datafusion = "46.0.1"
 derive_builder = "0.20.2"
-enumset = "1.1.9"
+enumset = "1.1.10"
 flatbuffers = "25.2.10"
 flate2 = "1.1.2"
 flexi_logger = { version = "0.31.2", features = ["trc", "dont_minimize_extra_stacks"] }
-form_urlencoded = "1.2.1"
+form_urlencoded = "1.2.2"
 futures = "0.3.31"
 futures-core = "0.3.31"
 futures-util = "0.3.31"
@@ -175,15 +175,15 @@ path-absolutize = "3.1.1"
 path-clean = "1.0.1"
 blake3 = { version = "1.8.2" }
 pbkdf2 = "0.12.2"
-percent-encoding = "2.3.1"
+percent-encoding = "2.3.2"
 pin-project-lite = "0.2.16"
 prost = "0.14.1"
 pretty_assertions = "1.4.1"
-quick-xml = "0.38.1"
+quick-xml = "0.38.3"
 rand = "0.9.2"
 rdkafka = { version = "0.38.0", features = ["tokio"] }
 reed-solomon-simd = { version = "3.0.1" }
-regex = { version = "1.11.1" }
+regex = { version = "1.11.2" }
 reqwest = { version = "0.12.23", default-features = false, features = [
     "rustls-tls",
     "charset",
@@ -193,7 +193,7 @@ reqwest = { version = "0.12.23", default-features = false, features = [
     "json",
     "blocking",
 ] }
-rmcp = { version = "0.5.0" }
+rmcp = { version = "0.6.1" }
 rmp = "0.8.14"
 rmp-serde = "1.3.0"
 rsa = "0.9.8"
@@ -211,20 +211,20 @@ serde_urlencoded = "0.7.1"
 serial_test = "3.2.0"
 sha1 = "0.10.6"
 sha2 = "0.10.9"
-shadow-rs = { version = "1.2.1", default-features = false }
+shadow-rs = { version = "1.3.0", default-features = false }
 siphasher = "1.0.1"
 smallvec = { version = "1.15.1", features = ["serde"] }
-snafu = "0.8.6"
+snafu = "0.8.8"
 snap = "1.1.1"
 socket2 = "0.6.0"
 strum = { version = "0.27.2", features = ["derive"] }
 sysinfo = "0.37.0"
 sysctl = "0.6.0"
-tempfile = "3.20.0"
+tempfile = "3.21.0"
 temp-env = "0.3.6"
 test-case = "3.3.1"
-thiserror = "2.0.15"
-time = { version = "0.3.41", features = [
+thiserror = "2.0.16"
+time = { version = "0.3.42", features = [
     "std",
     "parsing",
     "formatting",
@@ -246,9 +246,9 @@ tracing = "0.1.41"
 tracing-core = "0.1.34"
 tracing-error = "0.2.1"
 tracing-opentelemetry = "0.31.0"
-tracing-subscriber = { version = "0.3.19", features = ["env-filter", "time"] }
+tracing-subscriber = { version = "0.3.20", features = ["env-filter", "time"] }
 transform-stream = "0.3.1"
-url = "2.5.4"
+url = "2.5.7"
 urlencoding = "2.1.3"
 uuid = { version = "1.18.0", features = [
     "v4",

crates/ahm/Cargo.toml

@@ -22,6 +22,7 @@ tokio = { workspace = true, features = ["full"] }
 tokio-util = { workspace = true }
 tracing = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
+time.workspace = true
 serde_json = { workspace = true }
 thiserror = { workspace = true }
 uuid = { workspace = true, features = ["v4", "serde"] }

crates/ahm/src/scanner/data_scanner.rs

@@ -19,7 +19,7 @@ use std::{
 };
 
 use ecstore::{
-    disk::{DiskAPI, DiskStore, WalkDirOptions},
+    disk::{Disk, DiskAPI, DiskStore, WalkDirOptions},
     set_disk::SetDisks,
 };
 use rustfs_ecstore::{self as ecstore, StorageAPI, data_usage::store_data_usage_in_backend};
@@ -38,9 +38,11 @@ use crate::{
 };
 
 use rustfs_common::data_usage::DataUsageInfo;
+use rustfs_common::data_usage::SizeSummary;
 use rustfs_common::metrics::{Metric, Metrics, globalMetrics};
+use rustfs_ecstore::bucket::versioning::VersioningApi;
+use rustfs_ecstore::bucket::versioning_sys::BucketVersioningSys;
 use rustfs_ecstore::cmd::bucket_targets::VersioningConfig;
-
 use rustfs_ecstore::disk::RUSTFS_META_BUCKET;
 
 /// Custom scan mode enum for AHM scanner
@@ -1282,10 +1284,81 @@ impl Scanner {
                     } else {
                         // Apply lifecycle actions
                         if let Some(lifecycle_config) = &lifecycle_config {
-                            let mut scanner_item =
-                                ScannerItem::new(bucket.to_string(), Some(lifecycle_config.clone()), versioning_config.clone());
-                            if let Err(e) = scanner_item.apply_actions(&entry.name, entry.clone()).await {
-                                error!("Failed to apply lifecycle actions for {}/{}: {}", bucket, entry.name, e);
+                            if let Disk::Local(_local_disk) = &**disk {
+                                let vcfg = BucketVersioningSys::get(bucket).await.ok();
+
+                                let mut scanner_item = ScannerItem {
+                                    bucket: bucket.to_string(),
+                                    object_name: entry.name.clone(),
+                                    lifecycle: Some(lifecycle_config.clone()),
+                                    versioning: versioning_config.clone(),
+                                };
+                                //ScannerItem::new(bucket.to_string(), Some(lifecycle_config.clone()), versioning_config.clone());
+                                let fivs = match entry.clone().file_info_versions(&scanner_item.bucket) {
+                                    Ok(fivs) => fivs,
+                                    Err(_err) => {
+                                        stop_fn();
+                                        return Err(Error::other("skip this file"));
+                                    }
+                                };
+                                let mut size_s = SizeSummary::default();
+                                let obj_infos = match scanner_item.apply_versions_actions(&fivs.versions).await {
+                                    Ok(obj_infos) => obj_infos,
+                                    Err(_err) => {
+                                        stop_fn();
+                                        return Err(Error::other("skip this file"));
+                                    }
+                                };
+
+                                let versioned = if let Some(vcfg) = vcfg.as_ref() {
+                                    vcfg.versioned(&scanner_item.object_name)
+                                } else {
+                                    false
+                                };
+
+                                #[allow(unused_assignments)]
+                                let mut obj_deleted = false;
+                                for info in obj_infos.iter() {
+                                    let sz: i64;
+                                    (obj_deleted, sz) = scanner_item.apply_actions(info, &mut size_s).await;
+
+                                    if obj_deleted {
+                                        break;
+                                    }
+
+                                    let actual_sz = match info.get_actual_size() {
+                                        Ok(size) => size,
+                                        Err(_) => continue,
+                                    };
+
+                                    if info.delete_marker {
+                                        size_s.delete_markers += 1;
+                                    }
+
+                                    if info.version_id.is_some() && sz == actual_sz {
+                                        size_s.versions += 1;
+                                    }
+
+                                    size_s.total_size += sz as usize;
+
+                                    if info.delete_marker {
+                                        continue;
+                                    }
+                                }
+
+                                for free_version in fivs.free_versions.iter() {
+                                    let _obj_info = rustfs_ecstore::store_api::ObjectInfo::from_file_info(
+                                        free_version,
+                                        &scanner_item.bucket,
+                                        &scanner_item.object_name,
+                                        versioned,
+                                    );
+                                }
+
+                                // todo: global trace
+                                /*if obj_deleted {
+                                    return Err(Error::other(ERR_IGNORE_FILE_CONTRIB).into());
+                                }*/
                             }
                         }
 

crates/ahm/src/scanner/lifecycle.rs

@@ -13,66 +13,175 @@
 // limitations under the License.
 
 use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use time::OffsetDateTime;
 
+use crate::error::Result;
+use rustfs_common::data_usage::SizeSummary;
 use rustfs_common::metrics::IlmAction;
-use rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc;
-use rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_ops::{apply_lifecycle_action, eval_action_from_lifecycle};
+use rustfs_ecstore::bucket::lifecycle::{
+    bucket_lifecycle_audit::LcEventSrc,
+    bucket_lifecycle_ops::{GLOBAL_ExpiryState, apply_lifecycle_action, eval_action_from_lifecycle},
+    lifecycle,
+    lifecycle::Lifecycle,
+};
 use rustfs_ecstore::bucket::metadata_sys::get_object_lock_config;
+use rustfs_ecstore::bucket::object_lock::objectlock_sys::{BucketObjectLockSys, enforce_retention_for_deletion};
+use rustfs_ecstore::bucket::versioning::VersioningApi;
+use rustfs_ecstore::bucket::versioning_sys::BucketVersioningSys;
 use rustfs_ecstore::cmd::bucket_targets::VersioningConfig;
-use rustfs_ecstore::store_api::ObjectInfo;
-use rustfs_filemeta::FileMetaVersion;
-use rustfs_filemeta::metacache::MetaCacheEntry;
+use rustfs_ecstore::store_api::{ObjectInfo, ObjectToDelete};
+use rustfs_filemeta::FileInfo;
 use s3s::dto::BucketLifecycleConfiguration as LifecycleConfig;
 use tracing::info;
 
+static SCANNER_EXCESS_OBJECT_VERSIONS: AtomicU64 = AtomicU64::new(100);
+static SCANNER_EXCESS_OBJECT_VERSIONS_TOTAL_SIZE: AtomicU64 = AtomicU64::new(1024 * 1024 * 1024 * 1024); // 1 TB
+
 #[derive(Clone)]
 pub struct ScannerItem {
-    bucket: String,
-    lifecycle: Option<Arc<LifecycleConfig>>,
-    versioning: Option<Arc<VersioningConfig>>,
+    pub bucket: String,
+    pub object_name: String,
+    pub lifecycle: Option<Arc<LifecycleConfig>>,
+    pub versioning: Option<Arc<VersioningConfig>>,
 }
 
 impl ScannerItem {
     pub fn new(bucket: String, lifecycle: Option<Arc<LifecycleConfig>>, versioning: Option<Arc<VersioningConfig>>) -> Self {
         Self {
             bucket,
+            object_name: "".to_string(),
             lifecycle,
             versioning,
         }
     }
 
-    pub async fn apply_actions(&mut self, object: &str, mut meta: MetaCacheEntry) -> anyhow::Result<()> {
-        info!("apply_actions called for object: {}", object);
-        if self.lifecycle.is_none() {
-            info!("No lifecycle config for object: {}", object);
-            return Ok(());
+    pub async fn apply_versions_actions(&self, fivs: &[FileInfo]) -> Result<Vec<ObjectInfo>> {
+        let obj_infos = self.apply_newer_noncurrent_version_limit(fivs).await?;
+        if obj_infos.len() >= SCANNER_EXCESS_OBJECT_VERSIONS.load(Ordering::SeqCst) as usize {
+            // todo
         }
-        info!("Lifecycle config exists for object: {}", object);
 
-        let file_meta = match meta.xl_meta() {
-            Ok(meta) => meta,
-            Err(e) => {
-                tracing::error!("Failed to get xl_meta for {}: {}", object, e);
-                return Ok(());
+        let mut cumulative_size = 0;
+        for obj_info in obj_infos.iter() {
+            cumulative_size += obj_info.size;
+        }
+
+        if cumulative_size >= SCANNER_EXCESS_OBJECT_VERSIONS_TOTAL_SIZE.load(Ordering::SeqCst) as i64 {
+            //todo
+        }
+
+        Ok(obj_infos)
+    }
+
+    pub async fn apply_newer_noncurrent_version_limit(&self, fivs: &[FileInfo]) -> Result<Vec<ObjectInfo>> {
+        let lock_enabled = if let Some(rcfg) = BucketObjectLockSys::get(&self.bucket).await {
+            rcfg.mode.is_some()
+        } else {
+            false
+        };
+        let _vcfg = BucketVersioningSys::get(&self.bucket).await?;
+
+        let versioned = match BucketVersioningSys::get(&self.bucket).await {
+            Ok(vcfg) => vcfg.versioned(&self.object_name),
+            Err(_) => false,
+        };
+        let mut object_infos = Vec::with_capacity(fivs.len());
+
+        if self.lifecycle.is_none() {
+            for info in fivs.iter() {
+                object_infos.push(ObjectInfo::from_file_info(info, &self.bucket, &self.object_name, versioned));
             }
-        };
+            return Ok(object_infos);
+        }
 
-        let latest_version = file_meta.versions.first().cloned().unwrap_or_default();
-        let file_meta_version = FileMetaVersion::try_from(latest_version.meta.as_slice()).unwrap_or_default();
+        let event = self
+            .lifecycle
+            .as_ref()
+            .expect("lifecycle err.")
+            .clone()
+            .noncurrent_versions_expiration_limit(&lifecycle::ObjectOpts {
+                name: self.object_name.clone(),
+                ..Default::default()
+            })
+            .await;
+        let lim = event.newer_noncurrent_versions;
+        if lim == 0 || fivs.len() <= lim + 1 {
+            for fi in fivs.iter() {
+                object_infos.push(ObjectInfo::from_file_info(fi, &self.bucket, &self.object_name, versioned));
+            }
+            return Ok(object_infos);
+        }
 
-        let obj_info = ObjectInfo {
-            bucket: self.bucket.clone(),
-            name: object.to_string(),
-            version_id: latest_version.header.version_id,
-            mod_time: latest_version.header.mod_time,
-            size: file_meta_version.object.as_ref().map_or(0, |o| o.size),
-            user_defined: serde_json::from_slice(file_meta.data.as_slice()).unwrap_or_default(),
-            ..Default::default()
-        };
+        let overflow_versions = &fivs[lim + 1..];
+        for fi in fivs[..lim + 1].iter() {
+            object_infos.push(ObjectInfo::from_file_info(fi, &self.bucket, &self.object_name, versioned));
+        }
 
-        self.apply_lifecycle(&obj_info).await;
+        let mut to_del = Vec::<ObjectToDelete>::with_capacity(overflow_versions.len());
+        for fi in overflow_versions.iter() {
+            let obj = ObjectInfo::from_file_info(fi, &self.bucket, &self.object_name, versioned);
+            if lock_enabled && enforce_retention_for_deletion(&obj) {
+                //if enforce_retention_for_deletion(&obj) {
+                /*if self.debug {
+                    if obj.version_id.is_some() {
+                        info!("lifecycle: {} v({}) is locked, not deleting\n", obj.name, obj.version_id.expect("err"));
+                    } else {
+                        info!("lifecycle: {} is locked, not deleting\n", obj.name);
+                    }
+                }*/
+                object_infos.push(obj);
+                continue;
+            }
 
-        Ok(())
+            if OffsetDateTime::now_utc().unix_timestamp()
+                < lifecycle::expected_expiry_time(obj.successor_mod_time.expect("err"), event.noncurrent_days as i32)
+                    .unix_timestamp()
+            {
+                object_infos.push(obj);
+                continue;
+            }
+
+            to_del.push(ObjectToDelete {
+                object_name: obj.name,
+                version_id: obj.version_id,
+            });
+        }
+
+        if !to_del.is_empty() {
+            let mut expiry_state = GLOBAL_ExpiryState.write().await;
+            expiry_state.enqueue_by_newer_noncurrent(&self.bucket, to_del, event).await;
+        }
+
+        Ok(object_infos)
+    }
+
+    pub async fn apply_actions(&mut self, oi: &ObjectInfo, _size_s: &mut SizeSummary) -> (bool, i64) {
+        let (action, _size) = self.apply_lifecycle(oi).await;
+
+        info!(
+            "apply_actions {} {} {:?} {:?}",
+            oi.bucket.clone(),
+            oi.name.clone(),
+            oi.version_id.clone(),
+            oi.user_defined.clone()
+        );
+
+        // Create a mutable clone if you need to modify fields
+        /*let mut oi = oi.clone();
+        oi.replication_status = ReplicationStatusType::from(
+            oi.user_defined
+                .get("x-amz-bucket-replication-status")
+                .unwrap_or(&"PENDING".to_string()),
+        );
+        info!("apply status is: {:?}", oi.replication_status);
+        self.heal_replication(&oi, _size_s).await;*/
+
+        if action.delete_all() {
+            return (true, 0);
+        }
+
+        (false, oi.size)
     }
 
     async fn apply_lifecycle(&mut self, oi: &ObjectInfo) -> (IlmAction, i64) {

crates/ahm/tests/lifecycle_integration_test.rs

@@ -19,17 +19,22 @@ use rustfs_ecstore::{
     disk::endpoint::Endpoint,
     endpoints::{EndpointServerPools, Endpoints, PoolEndpoints},
     store::ECStore,
-    store_api::{ObjectIO, ObjectOptions, PutObjReader, StorageAPI},
+    store_api::{MakeBucketOptions, ObjectIO, ObjectOptions, PutObjReader, StorageAPI},
+    tier::tier::TierConfigMgr,
+    tier::tier_config::{TierConfig, TierMinIO, TierType},
 };
 use serial_test::serial;
 use std::sync::Once;
 use std::sync::OnceLock;
 use std::{path::PathBuf, sync::Arc, time::Duration};
 use tokio::fs;
-use tracing::info;
+use tokio::sync::RwLock;
+use tracing::warn;
+use tracing::{debug, info};
 
 static GLOBAL_ENV: OnceLock<(Vec<PathBuf>, Arc<ECStore>)> = OnceLock::new();
 static INIT: Once = Once::new();
+static GLOBAL_TIER_CONFIG_MGR: OnceLock<Arc<RwLock<TierConfigMgr>>> = OnceLock::new();
 
 fn init_tracing() {
     INIT.call_once(|| {
@@ -113,6 +118,8 @@ async fn setup_test_env() -> (Vec<PathBuf>, Arc<ECStore>) {
     // Store in global once lock
     let _ = GLOBAL_ENV.set((disk_paths.clone(), ecstore.clone()));
 
+    let _ = GLOBAL_TIER_CONFIG_MGR.set(TierConfigMgr::new());
+
     (disk_paths, ecstore)
 }
 
@@ -125,6 +132,22 @@ async fn create_test_bucket(ecstore: &Arc<ECStore>, bucket_name: &str) {
     info!("Created test bucket: {}", bucket_name);
 }
 
+/// Test helper: Create a test lock bucket
+async fn create_test_lock_bucket(ecstore: &Arc<ECStore>, bucket_name: &str) {
+    (**ecstore)
+        .make_bucket(
+            bucket_name,
+            &MakeBucketOptions {
+                lock_enabled: true,
+                versioning_enabled: true,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("Failed to create test bucket");
+    info!("Created test bucket: {}", bucket_name);
+}
+
 /// Test helper: Upload test object
 async fn upload_test_object(ecstore: &Arc<ECStore>, bucket: &str, object: &str, data: &[u8]) {
     let mut reader = PutObjReader::from_vec(data.to_vec());
@@ -158,11 +181,121 @@ async fn set_bucket_lifecycle(bucket_name: &str) -> Result<(), Box<dyn std::erro
     Ok(())
 }
 
+/// Test helper: Set bucket lifecycle configuration
+async fn set_bucket_lifecycle_deletemarker(bucket_name: &str) -> Result<(), Box<dyn std::error::Error>> {
+    // Create a simple lifecycle configuration XML with 0 days expiry for immediate testing
+    let lifecycle_xml = r#"<?xml version="1.0" encoding="UTF-8"?>
+<LifecycleConfiguration>
+    <Rule>
+        <ID>test-rule</ID>
+        <Status>Enabled</Status>
+        <Filter>
+            <Prefix>test/</Prefix>
+        </Filter>
+        <Expiration>
+            <Days>0</Days>
+            <ExpiredObjectDeleteMarker>true</ExpiredObjectDeleteMarker>
+        </Expiration>
+    </Rule>
+</LifecycleConfiguration>"#;
+
+    metadata_sys::update(bucket_name, BUCKET_LIFECYCLE_CONFIG, lifecycle_xml.as_bytes().to_vec()).await?;
+
+    Ok(())
+}
+
+#[allow(dead_code)]
+async fn set_bucket_lifecycle_transition(bucket_name: &str) -> Result<(), Box<dyn std::error::Error>> {
+    // Create a simple lifecycle configuration XML with 0 days expiry for immediate testing
+    let lifecycle_xml = r#"<?xml version="1.0" encoding="UTF-8"?>
+<LifecycleConfiguration>
+    <Rule>
+        <ID>test-rule</ID>
+        <Status>Enabled</Status>
+        <Filter>
+            <Prefix>test/</Prefix>
+        </Filter>
+        <Transition>
+          <Days>0</Days>
+          <StorageClass>COLDTIER</StorageClass>
+        </Transition>
+    </Rule>
+    <Rule>
+        <ID>test-rule2</ID>
+        <Status>Desabled</Status>
+        <Filter>
+            <Prefix>test/</Prefix>
+        </Filter>
+        <NoncurrentVersionTransition>
+          <NoncurrentDays>0</NoncurrentDays>
+          <StorageClass>COLDTIER</StorageClass>
+        </NoncurrentVersionTransition>
+    </Rule>
+</LifecycleConfiguration>"#;
+
+    metadata_sys::update(bucket_name, BUCKET_LIFECYCLE_CONFIG, lifecycle_xml.as_bytes().to_vec()).await?;
+
+    Ok(())
+}
+
+/// Test helper: Create a test tier
+#[allow(dead_code)]
+async fn create_test_tier() {
+    let args = TierConfig {
+        version: "v1".to_string(),
+        tier_type: TierType::MinIO,
+        name: "COLDTIER".to_string(),
+        s3: None,
+        rustfs: None,
+        minio: Some(TierMinIO {
+            access_key: "minioadmin".to_string(),
+            secret_key: "minioadmin".to_string(),
+            bucket: "mblock2".to_string(),
+            endpoint: "http://127.0.0.1:9020".to_string(),
+            prefix: "mypre3/".to_string(),
+            region: "".to_string(),
+            ..Default::default()
+        }),
+    };
+    let mut tier_config_mgr = GLOBAL_TIER_CONFIG_MGR.get().unwrap().write().await;
+    if let Err(err) = tier_config_mgr.add(args, false).await {
+        warn!("tier_config_mgr add failed, e: {:?}", err);
+        panic!("tier add failed. {err}");
+    }
+    if let Err(e) = tier_config_mgr.save().await {
+        warn!("tier_config_mgr save failed, e: {:?}", e);
+        panic!("tier save failed");
+    }
+    info!("Created test tier: {}", "COLDTIER");
+}
+
 /// Test helper: Check if object exists
 async fn object_exists(ecstore: &Arc<ECStore>, bucket: &str, object: &str) -> bool {
     ((**ecstore).get_object_info(bucket, object, &ObjectOptions::default()).await).is_ok()
 }
 
+/// Test helper: Check if object exists
+#[allow(dead_code)]
+async fn object_is_delete_marker(ecstore: &Arc<ECStore>, bucket: &str, object: &str) -> bool {
+    if let Ok(oi) = (**ecstore).get_object_info(bucket, object, &ObjectOptions::default()).await {
+        debug!("oi: {:?}", oi);
+        oi.delete_marker
+    } else {
+        panic!("object_is_delete_marker is error");
+    }
+}
+
+/// Test helper: Check if object exists
+#[allow(dead_code)]
+async fn object_is_transitioned(ecstore: &Arc<ECStore>, bucket: &str, object: &str) -> bool {
+    if let Ok(oi) = (**ecstore).get_object_info(bucket, object, &ObjectOptions::default()).await {
+        info!("oi: {:?}", oi);
+        !oi.transitioned_object.status.is_empty()
+    } else {
+        panic!("object_is_transitioned is error");
+    }
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 #[serial]
 async fn test_lifecycle_expiry_basic() {
@@ -221,11 +354,105 @@ async fn test_lifecycle_expiry_basic() {
     // Wait a bit more for background workers to process expiry tasks
     tokio::time::sleep(Duration::from_secs(5)).await;
 
-    // Check if object has been expired (deleted)
-    let object_still_exists = object_exists(&ecstore, bucket_name, object_name).await;
-    println!("Object exists after lifecycle processing: {object_still_exists}");
+    // Check if object has been expired (delete_marker)
+    let check_result = object_exists(&ecstore, bucket_name, object_name).await;
+    println!("Object is_delete_marker after lifecycle processing: {check_result}");
 
-    if object_still_exists {
+    if check_result {
+        println!("❌ Object was not deleted by lifecycle processing");
+    } else {
+        println!("✅ Object was successfully deleted by lifecycle processing");
+        // Let's try to get object info to see its details
+        match ecstore
+            .get_object_info(bucket_name, object_name, &rustfs_ecstore::store_api::ObjectOptions::default())
+            .await
+        {
+            Ok(obj_info) => {
+                println!(
+                    "Object info: name={}, size={}, mod_time={:?}",
+                    obj_info.name, obj_info.size, obj_info.mod_time
+                );
+            }
+            Err(e) => {
+                println!("Error getting object info: {e:?}");
+            }
+        }
+    }
+
+    assert!(!check_result);
+    println!("✅ Object successfully expired");
+
+    // Stop scanner
+    let _ = scanner.stop().await;
+    println!("✅ Scanner stopped");
+
+    println!("Lifecycle expiry basic test completed");
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn test_lifecycle_expiry_deletemarker() {
+    let (_disk_paths, ecstore) = setup_test_env().await;
+
+    // Create test bucket and object
+    let bucket_name = "test-lifecycle-bucket";
+    let object_name = "test/object.txt"; // Match the lifecycle rule prefix "test/"
+    let test_data = b"Hello, this is test data for lifecycle expiry!";
+
+    create_test_lock_bucket(&ecstore, bucket_name).await;
+    upload_test_object(&ecstore, bucket_name, object_name, test_data).await;
+
+    // Verify object exists initially
+    assert!(object_exists(&ecstore, bucket_name, object_name).await);
+    println!("✅ Object exists before lifecycle processing");
+
+    // Set lifecycle configuration with very short expiry (0 days = immediate expiry)
+    set_bucket_lifecycle_deletemarker(bucket_name)
+        .await
+        .expect("Failed to set lifecycle configuration");
+    println!("✅ Lifecycle configuration set for bucket: {bucket_name}");
+
+    // Verify lifecycle configuration was set
+    match rustfs_ecstore::bucket::metadata_sys::get(bucket_name).await {
+        Ok(bucket_meta) => {
+            assert!(bucket_meta.lifecycle_config.is_some());
+            println!("✅ Bucket metadata retrieved successfully");
+        }
+        Err(e) => {
+            println!("❌ Error retrieving bucket metadata: {e:?}");
+        }
+    }
+
+    // Create scanner with very short intervals for testing
+    let scanner_config = ScannerConfig {
+        scan_interval: Duration::from_millis(100),
+        deep_scan_interval: Duration::from_millis(500),
+        max_concurrent_scans: 1,
+        ..Default::default()
+    };
+
+    let scanner = Scanner::new(Some(scanner_config), None);
+
+    // Start scanner
+    scanner.start().await.expect("Failed to start scanner");
+    println!("✅ Scanner started");
+
+    // Wait for scanner to process lifecycle rules
+    tokio::time::sleep(Duration::from_secs(2)).await;
+
+    // Manually trigger a scan cycle to ensure lifecycle processing
+    scanner.scan_cycle().await.expect("Failed to trigger scan cycle");
+    println!("✅ Manual scan cycle completed");
+
+    // Wait a bit more for background workers to process expiry tasks
+    tokio::time::sleep(Duration::from_secs(5)).await;
+
+    // Check if object has been expired (deleted)
+    //let check_result = object_is_delete_marker(&ecstore, bucket_name, object_name).await;
+    let check_result = object_exists(&ecstore, bucket_name, object_name).await;
+    println!("Object exists after lifecycle processing: {check_result}");
+
+    if !check_result {
         println!("❌ Object was not deleted by lifecycle processing");
         // Let's try to get object info to see its details
         match ecstore
@@ -246,7 +473,7 @@ async fn test_lifecycle_expiry_basic() {
         println!("✅ Object was successfully deleted by lifecycle processing");
     }
 
-    assert!(!object_still_exists);
+    assert!(check_result);
     println!("✅ Object successfully expired");
 
     // Stop scanner
@@ -255,3 +482,100 @@ async fn test_lifecycle_expiry_basic() {
 
     println!("Lifecycle expiry basic test completed");
 }
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn test_lifecycle_transition_basic() {
+    let (_disk_paths, ecstore) = setup_test_env().await;
+
+    //create_test_tier().await;
+
+    // Create test bucket and object
+    let bucket_name = "test-lifecycle-bucket";
+    let object_name = "test/object.txt"; // Match the lifecycle rule prefix "test/"
+    let test_data = b"Hello, this is test data for lifecycle expiry!";
+
+    create_test_bucket(&ecstore, bucket_name).await;
+    upload_test_object(&ecstore, bucket_name, object_name, test_data).await;
+
+    // Verify object exists initially
+    assert!(object_exists(&ecstore, bucket_name, object_name).await);
+    println!("✅ Object exists before lifecycle processing");
+
+    // Set lifecycle configuration with very short expiry (0 days = immediate expiry)
+    /*set_bucket_lifecycle_transition(bucket_name)
+        .await
+        .expect("Failed to set lifecycle configuration");
+    println!("✅ Lifecycle configuration set for bucket: {bucket_name}");
+
+    // Verify lifecycle configuration was set
+    match rustfs_ecstore::bucket::metadata_sys::get(bucket_name).await {
+        Ok(bucket_meta) => {
+            assert!(bucket_meta.lifecycle_config.is_some());
+            println!("✅ Bucket metadata retrieved successfully");
+        }
+        Err(e) => {
+            println!("❌ Error retrieving bucket metadata: {e:?}");
+        }
+    }*/
+
+    // Create scanner with very short intervals for testing
+    let scanner_config = ScannerConfig {
+        scan_interval: Duration::from_millis(100),
+        deep_scan_interval: Duration::from_millis(500),
+        max_concurrent_scans: 1,
+        ..Default::default()
+    };
+
+    let scanner = Scanner::new(Some(scanner_config), None);
+
+    // Start scanner
+    scanner.start().await.expect("Failed to start scanner");
+    println!("✅ Scanner started");
+
+    // Wait for scanner to process lifecycle rules
+    tokio::time::sleep(Duration::from_secs(2)).await;
+
+    // Manually trigger a scan cycle to ensure lifecycle processing
+    scanner.scan_cycle().await.expect("Failed to trigger scan cycle");
+    println!("✅ Manual scan cycle completed");
+
+    // Wait a bit more for background workers to process expiry tasks
+    tokio::time::sleep(Duration::from_secs(5)).await;
+
+    // Check if object has been expired (deleted)
+    //let check_result = object_is_transitioned(&ecstore, bucket_name, object_name).await;
+    let check_result = object_exists(&ecstore, bucket_name, object_name).await;
+    println!("Object exists after lifecycle processing: {check_result}");
+
+    if check_result {
+        println!("✅ Object was not deleted by lifecycle processing");
+        // Let's try to get object info to see its details
+        match ecstore
+            .get_object_info(bucket_name, object_name, &rustfs_ecstore::store_api::ObjectOptions::default())
+            .await
+        {
+            Ok(obj_info) => {
+                println!(
+                    "Object info: name={}, size={}, mod_time={:?}",
+                    obj_info.name, obj_info.size, obj_info.mod_time
+                );
+                println!("Object info: transitioned_object={:?}", obj_info.transitioned_object);
+            }
+            Err(e) => {
+                println!("Error getting object info: {e:?}");
+            }
+        }
+    } else {
+        println!("❌ Object was deleted by lifecycle processing");
+    }
+
+    assert!(check_result);
+    println!("✅ Object successfully transitioned");
+
+    // Stop scanner
+    let _ = scanner.stop().await;
+    println!("✅ Scanner stopped");
+
+    println!("Lifecycle transition basic test completed");
+}

crates/config/src/constants/app.rs

@@ -124,7 +124,7 @@ pub const DEFAULT_LOG_FILENAME: &str = "rustfs";
 /// This is the default log filename for OBS.
 /// It is used to store the logs of the application.
 /// Default value: rustfs.log
-pub const DEFAULT_OBS_LOG_FILENAME: &str = concat!(DEFAULT_LOG_FILENAME, ".log");
+pub const DEFAULT_OBS_LOG_FILENAME: &str = concat!(DEFAULT_LOG_FILENAME, ".");
 
 /// Default sink file log file for rustfs
 /// This is the default sink file log file for rustfs.

crates/config/src/notify/mqtt.rs

@@ -27,7 +27,7 @@ pub const MQTT_QUEUE_LIMIT: &str = "queue_limit";
 
 /// A list of all valid configuration keys for an MQTT target.
 pub const NOTIFY_MQTT_KEYS: &[&str] = &[
-    ENABLE_KEY, // "enable" is a common key
+    ENABLE_KEY,
     MQTT_BROKER,
     MQTT_TOPIC,
     MQTT_QOS,

crates/config/src/notify/webhook.rs

@@ -24,7 +24,7 @@ pub const WEBHOOK_CLIENT_KEY: &str = "client_key";
 
 /// A list of all valid configuration keys for a webhook target.
 pub const NOTIFY_WEBHOOK_KEYS: &[&str] = &[
-    ENABLE_KEY, // "enable" is a common key
+    ENABLE_KEY,
     WEBHOOK_ENDPOINT,
     WEBHOOK_AUTH_TOKEN,
     WEBHOOK_QUEUE_LIMIT,

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

@@ -1,4 +1,3 @@
-#![allow(unused_imports)]
 // Copyright 2024 RustFS Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
@@ -12,6 +11,7 @@
 // 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.
+#![allow(unused_imports)]
 #![allow(unused_variables)]
 #![allow(unused_mut)]
 #![allow(unused_assignments)]
@@ -39,7 +39,7 @@ use time::OffsetDateTime;
 use tokio::select;
 use tokio::sync::mpsc::{Receiver, Sender};
 use tokio::sync::{RwLock, mpsc};
-use tracing::{error, info};
+use tracing::{debug, error, info};
 use uuid::Uuid;
 use xxhash_rust::xxh64;
 
@@ -587,7 +587,7 @@ impl TransitionState {
 pub async fn init_background_expiry(api: Arc<ECStore>) {
     let mut workers = num_cpus::get() / 2;
     //globalILMConfig.getExpirationWorkers()
-    if let Ok(env_expiration_workers) = env::var("_RUSTFS_EXPIRATION_WORKERS") {
+    if let Ok(env_expiration_workers) = env::var("_RUSTFS_ILM_EXPIRATION_WORKERS") {
         if let Ok(num_expirations) = env_expiration_workers.parse::<usize>() {
             workers = num_expirations;
         }
@@ -945,10 +945,13 @@ pub async fn apply_expiry_on_non_transitioned_objects(
 
     // let time_ilm = ScannerMetrics::time_ilm(lc_event.action.clone());
 
+    //debug!("lc_event.action: {:?}", lc_event.action);
+    //debug!("opts: {:?}", opts);
     let mut dobj = api
         .delete_object(&oi.bucket, &encode_dir_object(&oi.name), opts)
         .await
         .unwrap();
+    //debug!("dobj: {:?}", dobj);
     if dobj.name.is_empty() {
         dobj = oi.clone();
     }

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

@@ -25,6 +25,7 @@ use s3s::dto::{
 use std::cmp::Ordering;
 use std::env;
 use std::fmt::Display;
+use std::sync::Arc;
 use time::macros::{datetime, offset};
 use time::{self, Duration, OffsetDateTime};
 use tracing::info;
@@ -138,7 +139,7 @@ pub trait Lifecycle {
     async fn eval(&self, obj: &ObjectOpts) -> Event;
     async fn eval_inner(&self, obj: &ObjectOpts, now: OffsetDateTime) -> Event;
     //fn set_prediction_headers(&self, w: http.ResponseWriter, obj: ObjectOpts);
-    async fn noncurrent_versions_expiration_limit(&self, obj: &ObjectOpts) -> Event;
+    async fn noncurrent_versions_expiration_limit(self: Arc<Self>, obj: &ObjectOpts) -> Event;
 }
 
 #[async_trait::async_trait]
@@ -322,9 +323,7 @@ impl Lifecycle for BucketLifecycleConfiguration {
                             });
                             break;
                         }
-                    }
 
-                    if let Some(expiration) = rule.expiration.as_ref() {
                         if let Some(days) = expiration.days {
                             let expected_expiry = expected_expiry_time(obj.mod_time.expect("err!"), days /*, date*/);
                             if now.unix_timestamp() == 0 || now.unix_timestamp() > expected_expiry.unix_timestamp() {
@@ -538,7 +537,7 @@ impl Lifecycle for BucketLifecycleConfiguration {
         Event::default()
     }
 
-    async fn noncurrent_versions_expiration_limit(&self, obj: &ObjectOpts) -> Event {
+    async fn noncurrent_versions_expiration_limit(self: Arc<Self>, obj: &ObjectOpts) -> Event {
         if let Some(filter_rules) = self.filter_rules(obj).await {
             for rule in filter_rules.iter() {
                 if let Some(ref noncurrent_version_expiration) = rule.noncurrent_version_expiration {
@@ -626,7 +625,7 @@ pub fn expected_expiry_time(mod_time: OffsetDateTime, days: i32) -> OffsetDateTi
         .to_offset(offset!(-0:00:00))
         .saturating_add(Duration::days(days as i64));
     let mut hour = 3600;
-    if let Ok(env_ilm_hour) = env::var("_RUSTFS_ILM_HOUR") {
+    if let Ok(env_ilm_hour) = env::var("_RUSTFS_ILM_PROCESS_TIME") {
         if let Ok(num_hour) = env_ilm_hour.parse::<usize>() {
             hour = num_hour;
         }

crates/ecstore/src/config/storageclass.rs

@@ -36,6 +36,17 @@ pub fn default_parity_count(drive: usize) -> usize {
 pub const RRS: &str = "REDUCED_REDUNDANCY";
 pub const STANDARD: &str = "STANDARD";
 
+// AWS S3 Storage Classes
+pub const DEEP_ARCHIVE: &str = "DEEP_ARCHIVE";
+pub const EXPRESS_ONEZONE: &str = "EXPRESS_ONEZONE";
+pub const GLACIER: &str = "GLACIER";
+pub const GLACIER_IR: &str = "GLACIER_IR";
+pub const INTELLIGENT_TIERING: &str = "INTELLIGENT_TIERING";
+pub const ONEZONE_IA: &str = "ONEZONE_IA";
+pub const OUTPOSTS: &str = "OUTPOSTS";
+pub const SNOW: &str = "SNOW";
+pub const STANDARD_IA: &str = "STANDARD_IA";
+
 // Standard constants for config info storage class
 pub const CLASS_STANDARD: &str = "standard";
 pub const CLASS_RRS: &str = "rrs";
@@ -115,6 +126,15 @@ impl Config {
                     None
                 }
             }
+            // All these storage classes use standard parity configuration
+            STANDARD | DEEP_ARCHIVE | EXPRESS_ONEZONE | GLACIER | GLACIER_IR | INTELLIGENT_TIERING | ONEZONE_IA | OUTPOSTS
+            | SNOW | STANDARD_IA => {
+                if self.initialized {
+                    Some(self.standard.parity)
+                } else {
+                    None
+                }
+            }
             _ => {
                 if self.initialized {
                     Some(self.standard.parity)

crates/ecstore/src/set_disk.rs

@@ -4778,10 +4778,18 @@ impl StorageAPI for SetDisks {
 
         let part_number_marker = part_number_marker.unwrap_or_default();
 
+        // Extract storage class from metadata, default to STANDARD if not found
+        let storage_class = fi
+            .metadata
+            .get(rustfs_filemeta::headers::AMZ_STORAGE_CLASS)
+            .cloned()
+            .unwrap_or_else(|| storageclass::STANDARD.to_string());
+
         let mut ret = ListPartsInfo {
             bucket: bucket.to_owned(),
             object: object.to_owned(),
             upload_id: upload_id.to_owned(),
+            storage_class,
             max_parts,
             part_number_marker,
             user_defined: fi.metadata.clone(),
@@ -6039,6 +6047,40 @@ pub fn should_prevent_write(oi: &ObjectInfo, if_none_match: Option<String>, if_m
     }
 }
 
+/// Validates if the given storage class is supported
+pub fn is_valid_storage_class(storage_class: &str) -> bool {
+    matches!(
+        storage_class,
+        storageclass::STANDARD
+            | storageclass::RRS
+            | storageclass::DEEP_ARCHIVE
+            | storageclass::EXPRESS_ONEZONE
+            | storageclass::GLACIER
+            | storageclass::GLACIER_IR
+            | storageclass::INTELLIGENT_TIERING
+            | storageclass::ONEZONE_IA
+            | storageclass::OUTPOSTS
+            | storageclass::SNOW
+            | storageclass::STANDARD_IA
+    )
+}
+
+/// Returns true if the storage class is a cold storage tier that requires special handling
+pub fn is_cold_storage_class(storage_class: &str) -> bool {
+    matches!(
+        storage_class,
+        storageclass::DEEP_ARCHIVE | storageclass::GLACIER | storageclass::GLACIER_IR
+    )
+}
+
+/// Returns true if the storage class is an infrequent access tier
+pub fn is_infrequent_access_class(storage_class: &str) -> bool {
+    matches!(
+        storage_class,
+        storageclass::ONEZONE_IA | storageclass::STANDARD_IA | storageclass::INTELLIGENT_TIERING
+    )
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -6528,4 +6570,53 @@ mod tests {
         let if_match = None;
         assert!(!should_prevent_write(&oi, if_none_match, if_match));
     }
+
+    #[test]
+    fn test_is_valid_storage_class() {
+        // Test valid storage classes
+        assert!(is_valid_storage_class(storageclass::STANDARD));
+        assert!(is_valid_storage_class(storageclass::RRS));
+        assert!(is_valid_storage_class(storageclass::DEEP_ARCHIVE));
+        assert!(is_valid_storage_class(storageclass::EXPRESS_ONEZONE));
+        assert!(is_valid_storage_class(storageclass::GLACIER));
+        assert!(is_valid_storage_class(storageclass::GLACIER_IR));
+        assert!(is_valid_storage_class(storageclass::INTELLIGENT_TIERING));
+        assert!(is_valid_storage_class(storageclass::ONEZONE_IA));
+        assert!(is_valid_storage_class(storageclass::OUTPOSTS));
+        assert!(is_valid_storage_class(storageclass::SNOW));
+        assert!(is_valid_storage_class(storageclass::STANDARD_IA));
+
+        // Test invalid storage classes
+        assert!(!is_valid_storage_class("INVALID"));
+        assert!(!is_valid_storage_class(""));
+        assert!(!is_valid_storage_class("standard")); // lowercase
+    }
+
+    #[test]
+    fn test_is_cold_storage_class() {
+        // Test cold storage classes
+        assert!(is_cold_storage_class(storageclass::DEEP_ARCHIVE));
+        assert!(is_cold_storage_class(storageclass::GLACIER));
+        assert!(is_cold_storage_class(storageclass::GLACIER_IR));
+
+        // Test non-cold storage classes
+        assert!(!is_cold_storage_class(storageclass::STANDARD));
+        assert!(!is_cold_storage_class(storageclass::RRS));
+        assert!(!is_cold_storage_class(storageclass::STANDARD_IA));
+        assert!(!is_cold_storage_class(storageclass::EXPRESS_ONEZONE));
+    }
+
+    #[test]
+    fn test_is_infrequent_access_class() {
+        // Test infrequent access classes
+        assert!(is_infrequent_access_class(storageclass::ONEZONE_IA));
+        assert!(is_infrequent_access_class(storageclass::STANDARD_IA));
+        assert!(is_infrequent_access_class(storageclass::INTELLIGENT_TIERING));
+
+        // Test frequent access classes
+        assert!(!is_infrequent_access_class(storageclass::STANDARD));
+        assert!(!is_infrequent_access_class(storageclass::RRS));
+        assert!(!is_infrequent_access_class(storageclass::DEEP_ARCHIVE));
+        assert!(!is_infrequent_access_class(storageclass::EXPRESS_ONEZONE));
+    }
 }

crates/ecstore/src/store_list_objects.rs

@@ -1003,7 +1003,7 @@ async fn gather_results(
             }
         }
 
-        if !opts.incl_deleted && entry.is_object() && entry.is_latest_delete_marker() && entry.is_object_dir() {
+        if !opts.incl_deleted && entry.is_object() && entry.is_latest_delete_marker() && !entry.is_object_dir() {
             continue;
         }
 

crates/filemeta/src/filemeta.rs

@@ -112,6 +112,39 @@ impl FileMeta {
         Ok((&buf[8..], major, minor))
     }
 
+    // Returns (meta, inline_data)
+    pub fn is_indexed_meta(buf: &[u8]) -> Result<(&[u8], &[u8])> {
+        let (buf, major, minor) = Self::check_xl2_v1(buf)?;
+        if major != 1 || minor < 3 {
+            return Ok((&[], &[]));
+        }
+
+        let (mut size_buf, buf) = buf.split_at(5);
+
+        // Get meta data, buf = crc + data
+        let bin_len = rmp::decode::read_bin_len(&mut size_buf)?;
+
+        if buf.len() < bin_len as usize {
+            return Ok((&[], &[]));
+        }
+        let (meta, buf) = buf.split_at(bin_len as usize);
+
+        if buf.len() < 5 {
+            return Err(Error::other("insufficient data for CRC"));
+        }
+        let (mut crc_buf, inline_data) = buf.split_at(5);
+
+        // crc check
+        let crc = rmp::decode::read_u32(&mut crc_buf)?;
+        let meta_crc = xxh64::xxh64(meta, XXHASH_SEED) as u32;
+
+        if crc != meta_crc {
+            return Err(Error::other("xl file crc check failed"));
+        }
+
+        Ok((meta, inline_data))
+    }
+
     // Fixed u32
     pub fn read_bytes_header(buf: &[u8]) -> Result<(u32, &[u8])> {
         let (mut size_buf, _) = buf.split_at(5);
@@ -289,6 +322,7 @@ impl FileMeta {
 
         let offset = wr.len();
 
+        // xl header
         rmp::encode::write_uint8(&mut wr, XL_HEADER_VERSION)?;
         rmp::encode::write_uint8(&mut wr, XL_META_VERSION)?;
 
@@ -540,6 +574,15 @@ impl FileMeta {
             }
         }
 
+        let mut update_version = fi.mark_deleted;
+        /*if fi.version_purge_status().is_empty()
+        {
+            update_version = fi.mark_deleted;
+        }*/
+        if fi.transition_status == TRANSITION_COMPLETE {
+            update_version = false;
+        }
+
         for (i, ver) in self.versions.iter().enumerate() {
             if ver.header.version_id != fi.version_id {
                 continue;
@@ -557,54 +600,73 @@ impl FileMeta {
                     return Ok(None);
                 }
                 VersionType::Object => {
-                    let v = self.get_idx(i)?;
+                    if update_version && !fi.deleted {
+                        let v = self.get_idx(i)?;
 
-                    self.versions.remove(i);
+                        self.versions.remove(i);
 
-                    let a = v.object.map(|v| v.data_dir).unwrap_or_default();
-                    return Ok(a);
+                        let a = v.object.map(|v| v.data_dir).unwrap_or_default();
+                        return Ok(a);
+                    }
                 }
             }
         }
 
+        let mut found_index = None;
         for (i, version) in self.versions.iter().enumerate() {
-            if version.header.version_type != VersionType::Object || version.header.version_id != fi.version_id {
-                continue;
-            }
-
-            let mut ver = self.get_idx(i)?;
-
-            if fi.expire_restored {
-                ver.object.as_mut().unwrap().remove_restore_hdrs();
-                let _ = self.set_idx(i, ver.clone());
-            } else if fi.transition_status == TRANSITION_COMPLETE {
-                ver.object.as_mut().unwrap().set_transition(fi);
-                ver.object.as_mut().unwrap().reset_inline_data();
-                self.set_idx(i, ver.clone())?;
-            } else {
-                let vers = self.versions[i + 1..].to_vec();
-                self.versions.extend(vers.iter().cloned());
-                let (free_version, to_free) = ver.object.as_ref().unwrap().init_free_version(fi);
-                if to_free {
-                    self.add_version_filemata(free_version)?;
-                }
+            if version.header.version_type == VersionType::Object && version.header.version_id == fi.version_id {
+                found_index = Some(i);
+                break;
             }
+        }
 
+        let Some(i) = found_index else {
             if fi.deleted {
                 self.add_version_filemata(ventry)?;
-            }
-            if self.shared_data_dir_count(ver.object.as_ref().unwrap().version_id, ver.object.as_ref().unwrap().data_dir) > 0 {
                 return Ok(None);
             }
-            return Ok(ver.object.as_ref().unwrap().data_dir);
+            return Err(Error::FileVersionNotFound);
+        };
+
+        let mut ver = self.get_idx(i)?;
+
+        let Some(obj) = &mut ver.object else {
+            if fi.deleted {
+                self.add_version_filemata(ventry)?;
+                return Ok(None);
+            }
+            return Err(Error::FileVersionNotFound);
+        };
+
+        let obj_version_id = obj.version_id;
+        let obj_data_dir = obj.data_dir;
+
+        if fi.expire_restored {
+            obj.remove_restore_hdrs();
+            self.set_idx(i, ver)?;
+        } else if fi.transition_status == TRANSITION_COMPLETE {
+            obj.set_transition(fi);
+            obj.reset_inline_data();
+            self.set_idx(i, ver)?;
+        } else {
+            self.versions.remove(i);
+
+            let (free_version, to_free) = obj.init_free_version(fi);
+
+            if to_free {
+                self.add_version_filemata(free_version)?;
+            }
         }
 
         if fi.deleted {
             self.add_version_filemata(ventry)?;
+        }
+
+        if self.shared_data_dir_count(obj_version_id, obj_data_dir) > 0 {
             return Ok(None);
         }
 
-        Err(Error::FileVersionNotFound)
+        Ok(obj_data_dir)
     }
 
     pub fn into_fileinfo(

crates/filemeta/src/metacache.rs

@@ -112,8 +112,8 @@ impl MetaCacheEntry {
             return false;
         }
 
-        match FileMeta::check_xl2_v1(&self.metadata) {
-            Ok((meta, _, _)) => {
+        match FileMeta::is_indexed_meta(&self.metadata) {
+            Ok((meta, _inline_data)) => {
                 if !meta.is_empty() {
                     return FileMeta::is_latest_delete_marker(meta);
                 }

crates/mcp/src/server.rs

@@ -15,7 +15,7 @@
 use anyhow::Result;
 use rmcp::{
     ErrorData, RoleServer, ServerHandler,
-    handler::server::{router::tool::ToolRouter, tool::Parameters},
+    handler::server::{router::tool::ToolRouter, wrapper::Parameters},
     model::{Implementation, ProtocolVersion, ServerCapabilities, ServerInfo, ToolsCapability},
     service::{NotificationContext, RequestContext},
     tool, tool_handler, tool_router,

crates/notify/src/global.rs

@@ -162,13 +162,13 @@ impl Notifier {
         &self,
         bucket_name: &str,
         region: &str,
-        event_rules: &[(Vec<EventName>, &str, &str, Vec<TargetID>)],
+        event_rules: &[(Vec<EventName>, String, String, Vec<TargetID>)],
     ) -> Result<(), NotificationError> {
         let mut bucket_config = BucketNotificationConfig::new(region);
 
         for (event_names, prefix, suffix, target_ids) in event_rules {
             // Use `new_pattern` to construct a matching pattern
-            let pattern = crate::rules::pattern::new_pattern(Some(prefix), Some(suffix));
+            let pattern = crate::rules::pattern::new_pattern(Some(prefix.as_str()), Some(suffix.as_str()));
 
             for target_id in target_ids {
                 bucket_config.add_rule(event_names, pattern.clone(), target_id.clone());
@@ -186,4 +186,25 @@ impl Notifier {
             .load_bucket_notification_config(bucket_name, &bucket_config)
             .await
     }
+
+    /// Clear all notification rules for the specified bucket.
+    /// # Parameter
+    /// - `bucket_name`: The name of the target bucket.
+    /// # Return value
+    /// Returns `Result<(), NotificationError>`, Ok on success, and an error on failure.
+    /// # Using
+    /// This function allows you to clear all notification rules for a specific bucket.
+    /// This is useful when you want to reset the notification configuration for a bucket.
+    ///
+    pub async fn clear_bucket_notification_rules(&self, bucket_name: &str) -> Result<(), NotificationError> {
+        // Get global NotificationSystem instance
+        let notification_sys = match notification_system() {
+            Some(sys) => sys,
+            None => return Err(NotificationError::ServerNotInitialized),
+        };
+
+        // Clear configuration
+        notification_sys.remove_bucket_notification_config(bucket_name).await;
+        Ok(())
+    }
 }

crates/obs/src/config.rs

@@ -17,14 +17,13 @@ use rustfs_config::observability::{
     DEFAULT_SINKS_FILE_FLUSH_THRESHOLD, DEFAULT_SINKS_KAFKA_BATCH_SIZE, DEFAULT_SINKS_KAFKA_BATCH_TIMEOUT_MS,
     DEFAULT_SINKS_KAFKA_BROKERS, DEFAULT_SINKS_KAFKA_TOPIC, DEFAULT_SINKS_WEBHOOK_AUTH_TOKEN, DEFAULT_SINKS_WEBHOOK_ENDPOINT,
     DEFAULT_SINKS_WEBHOOK_MAX_RETRIES, DEFAULT_SINKS_WEBHOOK_RETRY_DELAY_MS, ENV_AUDIT_LOGGER_QUEUE_CAPACITY, ENV_OBS_ENDPOINT,
-    ENV_OBS_ENVIRONMENT, ENV_OBS_LOCAL_LOGGING_ENABLED, ENV_OBS_LOG_FILENAME, ENV_OBS_LOG_KEEP_FILES,
+    ENV_OBS_ENVIRONMENT, ENV_OBS_LOCAL_LOGGING_ENABLED, ENV_OBS_LOG_DIRECTORY, ENV_OBS_LOG_FILENAME, ENV_OBS_LOG_KEEP_FILES,
     ENV_OBS_LOG_ROTATION_SIZE_MB, ENV_OBS_LOG_ROTATION_TIME, ENV_OBS_LOGGER_LEVEL, ENV_OBS_METER_INTERVAL, ENV_OBS_SAMPLE_RATIO,
-    ENV_OBS_SERVICE_NAME, ENV_OBS_SERVICE_VERSION, ENV_SINKS_FILE_BUFFER_SIZE, ENV_SINKS_FILE_FLUSH_INTERVAL_MS,
-    ENV_SINKS_FILE_FLUSH_THRESHOLD, ENV_SINKS_FILE_PATH, ENV_SINKS_KAFKA_BATCH_SIZE, ENV_SINKS_KAFKA_BATCH_TIMEOUT_MS,
-    ENV_SINKS_KAFKA_BROKERS, ENV_SINKS_KAFKA_TOPIC, ENV_SINKS_WEBHOOK_AUTH_TOKEN, ENV_SINKS_WEBHOOK_ENDPOINT,
-    ENV_SINKS_WEBHOOK_MAX_RETRIES, ENV_SINKS_WEBHOOK_RETRY_DELAY_MS,
+    ENV_OBS_SERVICE_NAME, ENV_OBS_SERVICE_VERSION, ENV_OBS_USE_STDOUT, ENV_SINKS_FILE_BUFFER_SIZE,
+    ENV_SINKS_FILE_FLUSH_INTERVAL_MS, ENV_SINKS_FILE_FLUSH_THRESHOLD, ENV_SINKS_FILE_PATH, ENV_SINKS_KAFKA_BATCH_SIZE,
+    ENV_SINKS_KAFKA_BATCH_TIMEOUT_MS, ENV_SINKS_KAFKA_BROKERS, ENV_SINKS_KAFKA_TOPIC, ENV_SINKS_WEBHOOK_AUTH_TOKEN,
+    ENV_SINKS_WEBHOOK_ENDPOINT, ENV_SINKS_WEBHOOK_MAX_RETRIES, ENV_SINKS_WEBHOOK_RETRY_DELAY_MS,
 };
-use rustfs_config::observability::{ENV_OBS_LOG_DIRECTORY, ENV_OBS_USE_STDOUT};
 use rustfs_config::{
     APP_NAME, DEFAULT_LOG_KEEP_FILES, DEFAULT_LOG_LEVEL, DEFAULT_LOG_ROTATION_SIZE_MB, DEFAULT_LOG_ROTATION_TIME,
     DEFAULT_OBS_LOG_FILENAME, ENVIRONMENT, METER_INTERVAL, SAMPLE_RATIO, SERVICE_VERSION, USE_STDOUT,

crates/obs/src/sinks/mod.rs

@@ -79,7 +79,10 @@ pub async fn create_sinks(config: &AppConfig) -> Vec<Arc<dyn Sink>> {
             SinkConfig::File(file_config) => {
                 tracing::debug!("FileSink: Using path: {}", file_config.path);
                 match file::FileSink::new(
-                    format!("{}/{}", file_config.path.clone(), rustfs_config::DEFAULT_SINK_FILE_LOG_FILE),
+                    std::path::Path::new(&file_config.path)
+                        .join(rustfs_config::DEFAULT_SINK_FILE_LOG_FILE)
+                        .to_string_lossy()
+                        .to_string(),
                     file_config
                         .buffer_size
                         .unwrap_or(rustfs_config::observability::DEFAULT_SINKS_FILE_BUFFER_SIZE),

rustfs/Cargo.toml

@@ -103,6 +103,7 @@ tower-http = { workspace = true, features = [
     "cors",
     "catch-panic",
 ] }
+url = { workspace = true }
 urlencoding = { workspace = true }
 uuid = { workspace = true }
 zip = { workspace = true }

rustfs/src/admin/handlers/event.rs

@@ -27,15 +27,14 @@ use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, header::C
 use serde::{Deserialize, Serialize};
 use serde_urlencoded::from_bytes;
 use std::collections::HashMap;
+use std::future::Future;
+use std::io::{Error, ErrorKind};
+use std::net::SocketAddr;
+use std::path::Path;
+use tokio::net::lookup_host;
+use tokio::time::{Duration, sleep};
 use tracing::{debug, error, info, warn};
-
-#[derive(Debug, Deserialize)]
-struct TargetQuery {
-    #[serde(rename = "targetType")]
-    target_type: String,
-    #[serde(rename = "targetName")]
-    target_name: String,
-}
+use url::Url;
 
 #[derive(Debug, Deserialize)]
 struct BucketQuery {
@@ -43,19 +42,104 @@ struct BucketQuery {
     bucket_name: String,
 }
 
-/// Set (create or update) a notification target
-pub struct SetNotificationTarget {}
-#[async_trait::async_trait]
-impl Operation for SetNotificationTarget {
-    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        // 1. Analyze query parameters
-        let query: TargetQuery = from_bytes(req.uri.query().unwrap_or("").as_bytes())
-            .map_err(|e| s3_error!(InvalidArgument, "invalid query parameters: {}", e))?;
+#[derive(Debug, Deserialize)]
+pub struct KeyValue {
+    pub key: String,
+    pub value: String,
+}
 
-        let target_type = query.target_type.to_lowercase();
-        if target_type != *NOTIFY_WEBHOOK_SUB_SYS && target_type != *NOTIFY_MQTT_SUB_SYS {
-            return Err(s3_error!(InvalidArgument, "unsupported target type: {}", query.target_type));
+#[derive(Debug, Deserialize)]
+pub struct NotificationTargetBody {
+    pub key_values: Vec<KeyValue>,
+}
+
+#[derive(Serialize, Debug)]
+struct NotificationEndpoint {
+    account_id: String,
+    service: String,
+    status: String,
+}
+
+#[derive(Serialize, Debug)]
+struct NotificationEndpointsResponse {
+    notification_endpoints: Vec<NotificationEndpoint>,
+}
+
+async fn retry_with_backoff<F, Fut, T>(mut operation: F, max_attempts: usize, base_delay: Duration) -> Result<T, Error>
+where
+    F: FnMut() -> Fut,
+    Fut: Future<Output = Result<T, Error>>,
+{
+    assert!(max_attempts > 0, "max_attempts must be greater than 0");
+    let mut attempts = 0;
+    let mut delay = base_delay;
+    let mut last_err = None;
+
+    while attempts < max_attempts {
+        match operation().await {
+            Ok(result) => return Ok(result),
+            Err(e) => {
+                last_err = Some(e);
+                attempts += 1;
+                if attempts < max_attempts {
+                    warn!(
+                        "Retry attempt {}/{} failed: {}. Retrying in {:?}",
+                        attempts,
+                        max_attempts,
+                        last_err.as_ref().unwrap(),
+                        delay
+                    );
+                    sleep(delay).await;
+                    delay = delay.saturating_mul(2);
+                }
+            }
         }
+    }
+    Err(last_err.unwrap_or_else(|| Error::other("retry_with_backoff: unknown error")))
+}
+
+async fn retry_metadata(path: &str) -> Result<(), Error> {
+    retry_with_backoff(|| async { tokio::fs::metadata(path).await.map(|_| ()) }, 3, Duration::from_millis(100)).await
+}
+
+async fn validate_queue_dir(queue_dir: &str) -> S3Result<()> {
+    if !queue_dir.is_empty() {
+        if !Path::new(queue_dir).is_absolute() {
+            return Err(s3_error!(InvalidArgument, "queue_dir must be absolute path"));
+        }
+
+        if let Err(e) = retry_metadata(queue_dir).await {
+            match e.kind() {
+                ErrorKind::NotFound => {
+                    return Err(s3_error!(InvalidArgument, "queue_dir does not exist"));
+                }
+                ErrorKind::PermissionDenied => {
+                    return Err(s3_error!(InvalidArgument, "queue_dir exists but permission denied"));
+                }
+                _ => {
+                    return Err(s3_error!(InvalidArgument, "failed to access queue_dir: {}", e));
+                }
+            }
+        }
+    }
+
+    Ok(())
+}
+
+fn validate_cert_key_pair(cert: &Option<String>, key: &Option<String>) -> S3Result<()> {
+    if cert.is_some() != key.is_some() {
+        return Err(s3_error!(InvalidArgument, "client_cert and client_key must be specified as a pair"));
+    }
+    Ok(())
+}
+
+/// Set (create or update) a notification target
+pub struct NotificationTarget {}
+#[async_trait::async_trait]
+impl Operation for NotificationTarget {
+    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        // 1. Analyze query parameters
+        let (target_type, target_name) = extract_target_params(&params)?;
 
         // 2. Permission verification
         let Some(input_cred) = &req.credentials else {
@@ -82,25 +166,119 @@ impl Operation for SetNotificationTarget {
             kvs_map.insert(ENABLE_KEY.to_string(), EnableState::On.to_string());
         }
 
-        let kvs = rustfs_ecstore::config::KVS(
-            kvs_map
-                .into_iter()
-                .map(|(key, value)| rustfs_ecstore::config::KV {
-                    key,
-                    value,
-                    hidden_if_empty: false, // Set a default value
-                })
-                .collect(),
-        );
+        // 1. Get the allowed key range
+        let allowed_keys: std::collections::HashSet<&str> = match target_type {
+            NOTIFY_WEBHOOK_SUB_SYS => rustfs_config::notify::NOTIFY_WEBHOOK_KEYS.iter().cloned().collect(),
+            NOTIFY_MQTT_SUB_SYS => rustfs_config::notify::NOTIFY_MQTT_KEYS.iter().cloned().collect(),
+            _ => unreachable!(),
+        };
+
+        let notification_body: NotificationTargetBody = serde_json::from_slice(&body)
+            .map_err(|e| s3_error!(InvalidArgument, "invalid json body for target config: {}", e))?;
+
+        // 2. Filter and verify keys, and splice target_name
+        let mut kvs_vec = Vec::new();
+        let mut endpoint_val = None;
+        let mut queue_dir_val = None;
+        let mut client_cert_val = None;
+        let mut client_key_val = None;
+        let mut qos_val = None;
+
+        for kv in notification_body.key_values.iter() {
+            if !allowed_keys.contains(kv.key.as_str()) {
+                return Err(s3_error!(
+                    InvalidArgument,
+                    "key '{}' not allowed for target type '{}'",
+                    kv.key,
+                    target_type
+                ));
+            }
+            if kv.key == "endpoint" {
+                endpoint_val = Some(kv.value.clone());
+            }
+            if kv.key == "queue_dir" {
+                queue_dir_val = Some(kv.value.clone());
+            }
+            if kv.key == "client_cert" {
+                client_cert_val = Some(kv.value.clone());
+            }
+            if kv.key == "client_key" {
+                client_key_val = Some(kv.value.clone());
+            }
+            if kv.key == "qos" {
+                qos_val = Some(kv.value.clone());
+            }
+
+            kvs_vec.push(rustfs_ecstore::config::KV {
+                key: kv.key.clone(),
+                value: kv.value.clone(),
+                hidden_if_empty: false,
+            });
+        }
+
+        if target_type == NOTIFY_WEBHOOK_SUB_SYS {
+            let endpoint = endpoint_val
+                .clone()
+                .ok_or_else(|| s3_error!(InvalidArgument, "endpoint is required"))?;
+            let url = Url::parse(&endpoint).map_err(|e| s3_error!(InvalidArgument, "invalid endpoint url: {}", e))?;
+            let host = url
+                .host_str()
+                .ok_or_else(|| s3_error!(InvalidArgument, "endpoint missing host"))?;
+            let port = url
+                .port_or_known_default()
+                .ok_or_else(|| s3_error!(InvalidArgument, "endpoint missing port"))?;
+            let addr = format!("{host}:{port}");
+            // First, try to parse as SocketAddr (IP:port)
+            if addr.parse::<SocketAddr>().is_err() {
+                // If not an IP:port, try DNS resolution
+                if lookup_host(&addr).await.is_err() {
+                    return Err(s3_error!(InvalidArgument, "invalid or unresolvable endpoint address"));
+                }
+            }
+            if let Some(queue_dir) = queue_dir_val.clone() {
+                validate_queue_dir(&queue_dir).await?;
+            }
+            validate_cert_key_pair(&client_cert_val, &client_key_val)?;
+        }
+
+        if target_type == NOTIFY_MQTT_SUB_SYS {
+            let endpoint = endpoint_val.ok_or_else(|| s3_error!(InvalidArgument, "endpoint is required"))?;
+            let url = Url::parse(&endpoint).map_err(|e| s3_error!(InvalidArgument, "invalid endpoint url: {}", e))?;
+            match url.scheme() {
+                "tcp" | "ssl" | "ws" | "wss" | "mqtt" | "mqtts" => {}
+                _ => return Err(s3_error!(InvalidArgument, "unsupported broker url scheme")),
+            }
+            if let Some(queue_dir) = queue_dir_val {
+                validate_queue_dir(&queue_dir).await?;
+                if let Some(qos) = qos_val {
+                    match qos.parse::<u8>() {
+                        Ok(qos_int) if qos_int == 1 || qos_int == 2 => {}
+                        Ok(0) => {
+                            return Err(s3_error!(InvalidArgument, "qos should be 1 or 2 if queue_dir is set"));
+                        }
+                        _ => {
+                            return Err(s3_error!(InvalidArgument, "qos must be an integer 0, 1, or 2"));
+                        }
+                    }
+                }
+            }
+        }
+
+        // 3. Add ENABLE_KEY
+        kvs_vec.push(rustfs_ecstore::config::KV {
+            key: ENABLE_KEY.to_string(),
+            value: EnableState::On.to_string(),
+            hidden_if_empty: false,
+        });
+
+        let kvs = rustfs_ecstore::config::KVS(kvs_vec);
 
         // 5. Call notification system to set target configuration
-        info!("Setting target config for type '{}', name '{}'", &query.target_type, &query.target_name);
-        ns.set_target_config(&query.target_type, &query.target_name, kvs)
-            .await
-            .map_err(|e| {
-                error!("failed to set target config: {}", e);
-                S3Error::with_message(S3ErrorCode::InternalError, format!("failed to set target config: {e}"))
-            })?;
+        info!("Setting target config for type '{}', name '{}'", target_type, target_name);
+        ns.set_target_config(target_type, target_name, kvs).await.map_err(|e| {
+            error!("failed to set target config: {}", e);
+            S3Error::with_message(S3ErrorCode::InternalError, format!("failed to set target config: {e}"))
+        })?;
 
         let mut header = HeaderMap::new();
         header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
@@ -131,20 +309,68 @@ impl Operation for ListNotificationTargets {
         // 3. Get the list of activity targets
         let active_targets = ns.get_active_targets().await;
 
+        debug!("ListNotificationTargets call found {} active targets", active_targets.len());
+        let mut notification_endpoints = Vec::new();
+        for target_id in active_targets.iter() {
+            notification_endpoints.push(NotificationEndpoint {
+                account_id: target_id.id.clone(),
+                service: target_id.name.to_string(),
+                status: "online".to_string(),
+            });
+        }
+
+        let response = NotificationEndpointsResponse { notification_endpoints };
+
+        // 4. Serialize and return the result
+        let data = serde_json::to_vec(&response).map_err(|e| {
+            error!("Failed to serialize notification targets response: {:?}", response);
+            S3Error::with_message(S3ErrorCode::InternalError, format!("failed to serialize targets: {e}"))
+        })?;
+        debug!("ListNotificationTargets call end, response data length: {}", data.len(),);
+        let mut header = HeaderMap::new();
+        header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
+        Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), header))
+    }
+}
+
+/// Get a list of notification targets for all activities
+pub struct ListTargetsArns {}
+#[async_trait::async_trait]
+impl Operation for ListTargetsArns {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        debug!("ListTargetsArns call start request params: {:?}", req.uri.query());
+
+        // 1. Permission verification
+        let Some(input_cred) = &req.credentials else {
+            return Err(s3_error!(InvalidRequest, "credentials not found"));
+        };
+        let (_cred, _owner) =
+            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+
+        // 2. Get notification system instance
+        let Some(ns) = rustfs_notify::global::notification_system() else {
+            return Err(s3_error!(InternalError, "notification system not initialized"));
+        };
+
+        // 3. Get the list of activity targets
+        let active_targets = ns.get_active_targets().await;
+
+        debug!("ListTargetsArns call found {} active targets", active_targets.len());
+
         let region = match req.region.clone() {
             Some(region) => region,
             None => return Err(s3_error!(InvalidRequest, "region not found")),
         };
         let mut data_target_arn_list = Vec::new();
+
         for target_id in active_targets.iter() {
-            let target_arn = target_id.to_arn(&region);
-            data_target_arn_list.push(target_arn.to_string());
+            data_target_arn_list.push(target_id.to_arn(&region).to_string());
         }
 
         // 4. Serialize and return the result
         let data = serde_json::to_vec(&data_target_arn_list)
             .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("failed to serialize targets: {e}")))?;
-        debug!("ListNotificationTargets call end, response data length: {}", data.len(),);
+        debug!("ListTargetsArns call end, response data length: {}", data.len(),);
         let mut header = HeaderMap::new();
         header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
         Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), header))
@@ -155,10 +381,9 @@ impl Operation for ListNotificationTargets {
 pub struct RemoveNotificationTarget {}
 #[async_trait::async_trait]
 impl Operation for RemoveNotificationTarget {
-    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         // 1. Analyze query parameters
-        let query: TargetQuery = from_bytes(req.uri.query().unwrap_or("").as_bytes())
-            .map_err(|e| s3_error!(InvalidArgument, "invalid query parameters: {}", e))?;
+        let (target_type, target_name) = extract_target_params(&params)?;
 
         // 2. Permission verification
         let Some(input_cred) = &req.credentials else {
@@ -173,13 +398,11 @@ impl Operation for RemoveNotificationTarget {
         };
 
         // 4. Call notification system to remove target configuration
-        info!("Removing target config for type '{}', name '{}'", &query.target_type, &query.target_name);
-        ns.remove_target_config(&query.target_type, &query.target_name)
-            .await
-            .map_err(|e| {
-                error!("failed to remove target config: {}", e);
-                S3Error::with_message(S3ErrorCode::InternalError, format!("failed to remove target config: {e}"))
-            })?;
+        info!("Removing target config for type '{}', name '{}'", target_type, target_name);
+        ns.remove_target_config(target_type, target_name).await.map_err(|e| {
+            error!("failed to remove target config: {}", e);
+            S3Error::with_message(S3ErrorCode::InternalError, format!("failed to remove target config: {e}"))
+        })?;
 
         let mut header = HeaderMap::new();
         header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
@@ -188,6 +411,22 @@ impl Operation for RemoveNotificationTarget {
     }
 }
 
+fn extract_param<'a>(params: &'a Params<'_, '_>, key: &str) -> S3Result<&'a str> {
+    params
+        .get(key)
+        .ok_or_else(|| s3_error!(InvalidArgument, "missing required parameter: '{}'", key))
+}
+
+fn extract_target_params<'a>(params: &'a Params<'_, '_>) -> S3Result<(&'a str, &'a str)> {
+    let target_type = extract_param(params, "target_type")?;
+    if target_type != NOTIFY_WEBHOOK_SUB_SYS && target_type != NOTIFY_MQTT_SUB_SYS {
+        return Err(s3_error!(InvalidArgument, "unsupported target type: '{}'", target_type));
+    }
+
+    let target_name = extract_param(params, "target_name")?;
+    Ok((target_type, target_name))
+}
+
 /// Set notification rules for buckets
 pub struct SetBucketNotification {}
 #[async_trait::async_trait]

rustfs/src/admin/mod.rs

@@ -20,16 +20,15 @@ pub mod utils;
 
 // use ecstore::global::{is_dist_erasure, is_erasure};
 use handlers::{
-    bucket_meta, group, policies, pools, rebalance,
+    GetReplicationMetricsHandler, ListRemoteTargetHandler, RemoveRemoteTargetHandler, SetRemoteTargetHandler, bucket_meta,
+    event::{
+        GetBucketNotification, ListNotificationTargets, NotificationTarget, RemoveBucketNotification, RemoveNotificationTarget,
+        SetBucketNotification,
+    },
+    group, policies, pools, rebalance,
     service_account::{AddServiceAccount, DeleteServiceAccount, InfoServiceAccount, ListServiceAccount, UpdateServiceAccount},
     sts, tier, user,
 };
-
-use crate::admin::handlers::event::{
-    GetBucketNotification, ListNotificationTargets, RemoveBucketNotification, RemoveNotificationTarget, SetBucketNotification,
-    SetNotificationTarget,
-};
-use handlers::{GetReplicationMetricsHandler, ListRemoteTargetHandler, RemoveRemoteTargetHandler, SetRemoteTargetHandler};
 use hyper::Method;
 use router::{AdminOperation, S3Router};
 use rpc::register_rpc_route;
@@ -371,14 +370,14 @@ fn register_user_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<()>
 
     r.insert(
         Method::GET,
-        format!("{}{}", ADMIN_PREFIX, "/v3/target-list").as_str(),
+        format!("{}{}", ADMIN_PREFIX, "/v3/target/list").as_str(),
         AdminOperation(&ListNotificationTargets {}),
     )?;
 
     r.insert(
-        Method::POST,
-        format!("{}{}", ADMIN_PREFIX, "/v3/target-set").as_str(),
-        AdminOperation(&SetNotificationTarget {}),
+        Method::PUT,
+        format!("{}{}", ADMIN_PREFIX, "/v3/target/{target_type}/{target_name}").as_str(),
+        AdminOperation(&NotificationTarget {}),
     )?;
 
     // Remove notification target
@@ -388,9 +387,15 @@ fn register_user_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<()>
     // * `target_name` - A unique name for a Target, such as "1".
     r.insert(
         Method::DELETE,
-        format!("{}{}", ADMIN_PREFIX, "/v3/target-remove").as_str(),
+        format!("{}{}", ADMIN_PREFIX, "/v3/target/{target_type}/{target_name}/reset").as_str(),
         AdminOperation(&RemoveNotificationTarget {}),
     )?;
+    // arns
+    r.insert(
+        Method::GET,
+        format!("{}{}", ADMIN_PREFIX, "/v3/target/arns").as_str(),
+        AdminOperation(&ListNotificationTargets {}),
+    )?;
 
     r.insert(
         Method::POST,

rustfs/src/server/http.rs

@@ -144,9 +144,9 @@ pub async fn start_http_server(
             for domain in &opt.server_domains {
                 domain_sets.insert(domain.to_string());
                 if let Some((host, _)) = domain.split_once(':') {
-                    domain_sets.insert(format!("{}:{}", host, server_port));
+                    domain_sets.insert(format!("{host}:{server_port}"));
                 } else {
-                    domain_sets.insert(format!("{}:{}", domain, server_port));
+                    domain_sets.insert(format!("{domain}:{server_port}"));
                 }
             }
 

rustfs/src/storage/ecfs.rs

@@ -29,12 +29,6 @@ use chrono::Utc;
 use datafusion::arrow::csv::WriterBuilder as CsvWriterBuilder;
 use datafusion::arrow::json::WriterBuilder as JsonWriterBuilder;
 use datafusion::arrow::json::writer::JsonArray;
-use rustfs_ecstore::set_disk::MAX_PARTS_COUNT;
-use rustfs_s3select_api::object_store::bytes_stream;
-use rustfs_s3select_api::query::Context;
-use rustfs_s3select_api::query::Query;
-use rustfs_s3select_query::get_global_db;
-
 // use rustfs_ecstore::store_api::RESERVED_METADATA_PREFIX;
 use futures::StreamExt;
 use http::HeaderMap;
@@ -63,7 +57,8 @@ use rustfs_ecstore::compress::MIN_COMPRESSIBLE_SIZE;
 use rustfs_ecstore::compress::is_compressible;
 use rustfs_ecstore::error::StorageError;
 use rustfs_ecstore::new_object_layer_fn;
-use rustfs_ecstore::set_disk::DEFAULT_READ_BUFFER_SIZE;
+use rustfs_ecstore::set_disk::MAX_PARTS_COUNT;
+use rustfs_ecstore::set_disk::{DEFAULT_READ_BUFFER_SIZE, is_valid_storage_class};
 use rustfs_ecstore::store_api::BucketOptions;
 use rustfs_ecstore::store_api::CompletePart;
 use rustfs_ecstore::store_api::DeleteBucketOptions;
@@ -77,6 +72,7 @@ use rustfs_ecstore::store_api::PutObjReader;
 use rustfs_ecstore::store_api::StorageAPI;
 use rustfs_filemeta::headers::RESERVED_METADATA_PREFIX_LOWER;
 use rustfs_filemeta::headers::{AMZ_DECODED_CONTENT_LENGTH, AMZ_OBJECT_TAGGING};
+use rustfs_notify::global::notifier_instance;
 use rustfs_policy::auth;
 use rustfs_policy::policy::action::Action;
 use rustfs_policy::policy::action::S3Action;
@@ -86,7 +82,12 @@ use rustfs_rio::EtagReader;
 use rustfs_rio::HashReader;
 use rustfs_rio::Reader;
 use rustfs_rio::WarpReader;
+use rustfs_s3select_api::object_store::bytes_stream;
+use rustfs_s3select_api::query::Context;
+use rustfs_s3select_api::query::Query;
+use rustfs_s3select_query::get_global_db;
 use rustfs_targets::EventName;
+use rustfs_targets::arn::{TargetID, TargetIDError};
 use rustfs_utils::CompressionAlgorithm;
 use rustfs_utils::path::path_join_buf;
 use rustfs_zip::CompressionFormat;
@@ -262,7 +263,7 @@ impl FS {
 
                 // Asynchronous call will not block the response of the current request
                 tokio::spawn(async move {
-                    rustfs_notify::global::notifier_instance().notify(event_args).await;
+                    notifier_instance().notify(event_args).await;
                 });
             }
         }
@@ -290,6 +291,7 @@ impl FS {
         Ok(S3Response::new(output))
     }
 }
+
 #[async_trait::async_trait]
 impl S3 for FS {
     #[tracing::instrument(
@@ -335,7 +337,7 @@ impl S3 for FS {
 
         // Asynchronous call will not block the response of the current request
         tokio::spawn(async move {
-            rustfs_notify::global::notifier_instance().notify(event_args).await;
+            notifier_instance().notify(event_args).await;
         });
 
         Ok(S3Response::new(output))
@@ -481,7 +483,7 @@ impl S3 for FS {
 
         // Asynchronous call will not block the response of the current request
         tokio::spawn(async move {
-            rustfs_notify::global::notifier_instance().notify(event_args).await;
+            notifier_instance().notify(event_args).await;
         });
 
         Ok(S3Response::new(output))
@@ -681,7 +683,7 @@ impl S3 for FS {
 
         // Asynchronous call will not block the response of the current request
         tokio::spawn(async move {
-            rustfs_notify::global::notifier_instance().notify(event_args).await;
+            notifier_instance().notify(event_args).await;
         });
 
         Ok(S3Response::new(DeleteBucketOutput {}))
@@ -756,7 +758,7 @@ impl S3 for FS {
 
         // Asynchronous call will not block the response of the current request
         tokio::spawn(async move {
-            rustfs_notify::global::notifier_instance().notify(event_args).await;
+            notifier_instance().notify(event_args).await;
         });
 
         Ok(S3Response::new(output))
@@ -841,7 +843,7 @@ impl S3 for FS {
                     host: rustfs_utils::get_request_host(&req.headers),
                     user_agent: rustfs_utils::get_request_user_agent(&req.headers),
                 };
-                rustfs_notify::global::notifier_instance().notify(event_args).await;
+                notifier_instance().notify(event_args).await;
             }
         });
 
@@ -961,11 +963,11 @@ impl S3 for FS {
             }
         }
 
-        let mut content_length = info.size as i64;
+        let mut content_length = info.size;
 
         let content_range = if let Some(rs) = rs {
             let total_size = info.get_actual_size().map_err(ApiError::from)?;
-            let (start, length) = rs.get_offset_length(total_size as i64).map_err(ApiError::from)?;
+            let (start, length) = rs.get_offset_length(total_size).map_err(ApiError::from)?;
             content_length = length;
             Some(format!("bytes {}-{}/{}", start, start as i64 + length - 1, total_size))
         } else {
@@ -1006,7 +1008,7 @@ impl S3 for FS {
 
         // Asynchronous call will not block the response of the current request
         tokio::spawn(async move {
-            rustfs_notify::global::notifier_instance().notify(event_args).await;
+            notifier_instance().notify(event_args).await;
         });
 
         Ok(S3Response::new(output))
@@ -1128,7 +1130,7 @@ impl S3 for FS {
 
         // Asynchronous call will not block the response of the current request
         tokio::spawn(async move {
-            rustfs_notify::global::notifier_instance().notify(event_args).await;
+            notifier_instance().notify(event_args).await;
         });
 
         Ok(S3Response::new(output))
@@ -1383,8 +1385,7 @@ impl S3 for FS {
         let input = req.input;
 
         if let Some(ref storage_class) = input.storage_class {
-            let is_valid = ["STANDARD", "REDUCED_REDUNDANCY"].contains(&storage_class.as_str());
-            if !is_valid {
+            if !is_valid_storage_class(storage_class.as_str()) {
                 return Err(s3_error!(InvalidStorageClass));
             }
         }
@@ -1512,7 +1513,7 @@ impl S3 for FS {
 
         // Asynchronous call will not block the response of the current request
         tokio::spawn(async move {
-            rustfs_notify::global::notifier_instance().notify(event_args).await;
+            notifier_instance().notify(event_args).await;
         });
 
         Ok(S3Response::new(output))
@@ -1528,9 +1529,17 @@ impl S3 for FS {
             key,
             tagging,
             version_id,
+            storage_class,
             ..
         } = req.input.clone();
 
+        // Validate storage class if provided
+        if let Some(ref storage_class) = storage_class {
+            if !is_valid_storage_class(storage_class.as_str()) {
+                return Err(s3_error!(InvalidStorageClass));
+            }
+        }
+
         // mc cp step 3
 
         // debug!("create_multipart_upload meta {:?}", &metadata);
@@ -1590,7 +1599,7 @@ impl S3 for FS {
 
         // Asynchronous call will not block the response of the current request
         tokio::spawn(async move {
-            rustfs_notify::global::notifier_instance().notify(event_args).await;
+            notifier_instance().notify(event_args).await;
         });
 
         Ok(S3Response::new(output))
@@ -1893,6 +1902,20 @@ impl S3 for FS {
                     })
                     .collect(),
             ),
+            owner: Some(RUSTFS_OWNER.to_owned()),
+            initiator: Some(Initiator {
+                id: RUSTFS_OWNER.id.clone(),
+                display_name: RUSTFS_OWNER.display_name.clone(),
+            }),
+            is_truncated: Some(res.is_truncated),
+            next_part_number_marker: res.next_part_number_marker.try_into().ok(),
+            max_parts: res.max_parts.try_into().ok(),
+            part_number_marker: res.part_number_marker.try_into().ok(),
+            storage_class: if res.storage_class.is_empty() {
+                None
+            } else {
+                Some(res.storage_class.into())
+            },
             ..Default::default()
         };
         Ok(S3Response::new(output))
@@ -2147,7 +2170,7 @@ impl S3 for FS {
 
         // Asynchronous call will not block the response of the current request
         tokio::spawn(async move {
-            rustfs_notify::global::notifier_instance().notify(event_args).await;
+            notifier_instance().notify(event_args).await;
         });
 
         Ok(S3Response::new(PutObjectTaggingOutput { version_id: None }))
@@ -2214,7 +2237,7 @@ impl S3 for FS {
 
         // Asynchronous call will not block the response of the current request
         tokio::spawn(async move {
-            rustfs_notify::global::notifier_instance().notify(event_args).await;
+            notifier_instance().notify(event_args).await;
         });
 
         Ok(S3Response::new(DeleteObjectTaggingOutput { version_id: None }))
@@ -2791,20 +2814,56 @@ impl S3 for FS {
             return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
         };
 
+        //  Verify that the bucket exists
         store
             .get_bucket_info(&bucket, &BucketOptions::default())
             .await
             .map_err(ApiError::from)?;
 
+        //  Persist the new notification configuration
         let data = try_!(serialize(&notification_configuration));
-
         metadata_sys::update(&bucket, BUCKET_NOTIFICATION_CONFIG, data)
             .await
             .map_err(ApiError::from)?;
 
-        // TODO: event notice add rule
+        // Determine region (BucketInfo has no region field) -> use global region or default
+        let region = rustfs_ecstore::global::get_global_region().unwrap_or_else(|| req.region.clone().unwrap_or_default());
 
-        Ok(S3Response::new(PutBucketNotificationConfigurationOutput::default()))
+        // Purge old rules and resolve new rules in parallel
+        let clear_rules = notifier_instance().clear_bucket_notification_rules(&bucket);
+        let parse_rules = async {
+            let mut event_rules = Vec::new();
+
+            process_queue_configurations(
+                &mut event_rules,
+                notification_configuration.queue_configurations.clone(),
+                TargetID::from_str,
+            );
+            process_topic_configurations(
+                &mut event_rules,
+                notification_configuration.topic_configurations.clone(),
+                TargetID::from_str,
+            );
+            process_lambda_configurations(
+                &mut event_rules,
+                notification_configuration.lambda_function_configurations.clone(),
+                TargetID::from_str,
+            );
+
+            event_rules
+        };
+
+        let (clear_result, event_rules) = tokio::join!(clear_rules, parse_rules);
+
+        clear_result.map_err(|e| s3_error!(InternalError, "Failed to clear rules: {e}"))?;
+
+        // Add a new notification rule
+        notifier_instance()
+            .add_event_specific_rules(&bucket, &region, &event_rules)
+            .await
+            .map_err(|e| s3_error!(InternalError, "Failed to add rules: {e}"))?;
+
+        Ok(S3Response::new(PutBucketNotificationConfigurationOutput {}))
     }
 
     async fn get_bucket_acl(&self, req: S3Request<GetBucketAclInput>) -> S3Result<S3Response<GetBucketAclOutput>> {
@@ -2951,7 +3010,7 @@ impl S3 for FS {
 
         // Asynchronous call will not block the response of the current request
         tokio::spawn(async move {
-            rustfs_notify::global::notifier_instance().notify(event_args).await;
+            notifier_instance().notify(event_args).await;
         });
 
         Ok(S3Response::new(output))
@@ -3129,7 +3188,7 @@ impl S3 for FS {
 
         // Asynchronous call will not block the response of the current request
         tokio::spawn(async move {
-            rustfs_notify::global::notifier_instance().notify(event_args).await;
+            notifier_instance().notify(event_args).await;
         });
 
         Ok(S3Response::new(output))
@@ -3208,7 +3267,7 @@ impl S3 for FS {
 
         // Asynchronous call will not block the response of the current request
         tokio::spawn(async move {
-            rustfs_notify::global::notifier_instance().notify(event_args).await;
+            notifier_instance().notify(event_args).await;
         });
 
         Ok(S3Response::new(output))
@@ -3269,7 +3328,7 @@ impl S3 for FS {
 
         // Asynchronous call will not block the response of the current request
         tokio::spawn(async move {
-            rustfs_notify::global::notifier_instance().notify(event_args).await;
+            notifier_instance().notify(event_args).await;
         });
 
         Ok(S3Response::new(output))
@@ -3344,13 +3403,91 @@ impl S3 for FS {
 
         // Asynchronous call will not block the response of the current request
         tokio::spawn(async move {
-            rustfs_notify::global::notifier_instance().notify(event_args).await;
+            notifier_instance().notify(event_args).await;
         });
 
         Ok(S3Response::new(output))
     }
 }
 
+/// Auxiliary functions: extract prefixes and suffixes
+fn extract_prefix_suffix(filter: Option<&NotificationConfigurationFilter>) -> (String, String) {
+    if let Some(filter) = filter {
+        if let Some(filter_rules) = &filter.key {
+            let mut prefix = String::new();
+            let mut suffix = String::new();
+            if let Some(rules) = &filter_rules.filter_rules {
+                for rule in rules {
+                    if let (Some(name), Some(value)) = (rule.name.as_ref(), rule.value.as_ref()) {
+                        match name.as_str() {
+                            "prefix" => prefix = value.clone(),
+                            "suffix" => suffix = value.clone(),
+                            _ => {}
+                        }
+                    }
+                }
+            }
+            return (prefix, suffix);
+        }
+    }
+    (String::new(), String::new())
+}
+
+/// Auxiliary functions: Handle configuration
+fn process_queue_configurations<F>(
+    event_rules: &mut Vec<(Vec<EventName>, String, String, Vec<TargetID>)>,
+    configurations: Option<Vec<QueueConfiguration>>,
+    target_id_parser: F,
+) where
+    F: Fn(&str) -> Result<TargetID, TargetIDError>,
+{
+    if let Some(configs) = configurations {
+        for cfg in configs {
+            let events = cfg.events.iter().filter_map(|e| EventName::parse(e.as_ref()).ok()).collect();
+            let (prefix, suffix) = extract_prefix_suffix(cfg.filter.as_ref());
+            let target_ids = vec![target_id_parser(&cfg.queue_arn).ok()].into_iter().flatten().collect();
+            event_rules.push((events, prefix, suffix, target_ids));
+        }
+    }
+}
+
+fn process_topic_configurations<F>(
+    event_rules: &mut Vec<(Vec<EventName>, String, String, Vec<TargetID>)>,
+    configurations: Option<Vec<TopicConfiguration>>,
+    target_id_parser: F,
+) where
+    F: Fn(&str) -> Result<TargetID, TargetIDError>,
+{
+    if let Some(configs) = configurations {
+        for cfg in configs {
+            let events = cfg.events.iter().filter_map(|e| EventName::parse(e.as_ref()).ok()).collect();
+            let (prefix, suffix) = extract_prefix_suffix(cfg.filter.as_ref());
+            let target_ids = vec![target_id_parser(&cfg.topic_arn).ok()].into_iter().flatten().collect();
+            event_rules.push((events, prefix, suffix, target_ids));
+        }
+    }
+}
+
+fn process_lambda_configurations<F>(
+    event_rules: &mut Vec<(Vec<EventName>, String, String, Vec<TargetID>)>,
+    configurations: Option<Vec<LambdaFunctionConfiguration>>,
+    target_id_parser: F,
+) where
+    F: Fn(&str) -> Result<TargetID, TargetIDError>,
+{
+    if let Some(configs) = configurations {
+        for cfg in configs {
+            let events = cfg.events.iter().filter_map(|e| EventName::parse(e.as_ref()).ok()).collect();
+            let (prefix, suffix) = extract_prefix_suffix(cfg.filter.as_ref());
+            let target_ids = vec![target_id_parser(&cfg.lambda_function_arn).ok()]
+                .into_iter()
+                .flatten()
+                .collect();
+            event_rules.push((events, prefix, suffix, target_ids));
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

scripts/run.sh

@@ -45,7 +45,7 @@ export RUSTFS_VOLUMES="./target/volume/test{1...4}"
 # export RUSTFS_VOLUMES="./target/volume/test"
 export RUSTFS_ADDRESS=":9000"
 export RUSTFS_CONSOLE_ENABLE=true
-export RUSTFS_CONSOLE_ADDRESS=":9001"
+# export RUSTFS_CONSOLE_ADDRESS=":9001"
 # export RUSTFS_SERVER_DOMAINS="localhost:9000"
 # HTTPS certificate directory
 # export RUSTFS_TLS_PATH="./deploy/certs"