fix(admin): authorize cross-user ListServiceAccount with ListServiceAccount (#2640)

Signed-off-by: 唐小鸭 <tangtang1251@qq.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: loverustfs <hello@rustfs.com>

crates/e2e_test/src/list_object_versions_regression_test.rs

@@ -19,6 +19,7 @@
 mod tests {
     use crate::common::{RustFSTestEnvironment, init_logging};
     use aws_sdk_s3::Client;
+    use aws_sdk_s3::primitives::ByteStream;
     use aws_sdk_s3::types::{BucketVersioningStatus, VersioningConfiguration};
     use serial_test::serial;
     use tracing::info;
@@ -179,4 +180,84 @@ mod tests {
             "Delete marker should no longer be latest after the second put"
         );
     }
+
+    #[tokio::test]
+    #[serial]
+    async fn test_list_object_versions_prefix_with_marker_object_returns_children() {
+        init_logging();
+        info!("🧪 TEST: ListObjectVersions returns prefix children when a marker object also exists");
+
+        let mut env = RustFSTestEnvironment::new().await.expect("Failed to create test environment");
+        env.start_rustfs_server(vec![]).await.expect("Failed to start RustFS");
+
+        let client = create_s3_client(&env);
+        let bucket = "test-list-versions-prefix-marker";
+        let marker_key = "data01";
+        let child_keys = [
+            "data01/meta/dump-2026-04-08-053205.json.gz",
+            "data01/meta/dump-2026-04-08-063209.json.gz",
+        ];
+
+        client
+            .create_bucket()
+            .bucket(bucket)
+            .send()
+            .await
+            .expect("Failed to create bucket");
+
+        client
+            .put_bucket_versioning()
+            .bucket(bucket)
+            .versioning_configuration(
+                VersioningConfiguration::builder()
+                    .status(BucketVersioningStatus::Suspended)
+                    .build(),
+            )
+            .send()
+            .await
+            .expect("Failed to suspend versioning");
+
+        client
+            .put_object()
+            .bucket(bucket)
+            .key(marker_key)
+            .body(ByteStream::from_static(b""))
+            .send()
+            .await
+            .expect("Failed to put marker object");
+
+        for key in child_keys {
+            client
+                .put_object()
+                .bucket(bucket)
+                .key(key)
+                .body(ByteStream::from_static(b"payload"))
+                .send()
+                .await
+                .expect("Failed to put child object");
+        }
+
+        let listing = client
+            .list_object_versions()
+            .bucket(bucket)
+            .prefix("data01/")
+            .send()
+            .await
+            .expect("Failed to list object versions by prefix");
+
+        let version_keys: Vec<_> = listing.versions().iter().filter_map(|version| version.key()).collect();
+
+        assert_eq!(
+            version_keys.len(),
+            child_keys.len(),
+            "ListObjectVersions with a trailing slash prefix should include child objects even when the marker object exists"
+        );
+
+        for key in child_keys {
+            assert!(
+                version_keys.contains(&key),
+                "ListObjectVersions(prefix=data01/) should include child object {key}"
+            );
+        }
+    }
 }

crates/e2e_test/src/reliant/grpc_lock_client.rs

@@ -120,11 +120,6 @@ impl LockClient for GrpcLockClient {
             .map_err(|e| LockError::internal(e.to_string()))?
             .into_inner();
 
-        // Check for explicit error first
-        if let Some(error_info) = resp.error_info {
-            return Err(LockError::internal(error_info));
-        }
-
         // Check if the lock acquisition was successful
         if resp.success {
             Ok(LockResponse::success(
@@ -134,7 +129,8 @@ impl LockClient for GrpcLockClient {
         } else {
             // Lock acquisition failed
             Ok(LockResponse::failure(
-                "Lock acquisition failed on remote server".to_string(),
+                resp.error_info
+                    .unwrap_or_else(|| "Lock acquisition failed on remote server".to_string()),
                 std::time::Duration::ZERO,
             ))
         }

crates/e2e_test/src/reliant/grpc_lock_server.rs

@@ -50,6 +50,18 @@ fn lock_result_from_error(error: impl Into<String>) -> GenerallyLockResult {
     }
 }
 
+fn lock_result_from_release(lock_id: &rustfs_lock::LockId, success: bool) -> GenerallyLockResult {
+    if success {
+        GenerallyLockResult {
+            success: true,
+            error_info: None,
+            lock_info: None,
+        }
+    } else {
+        lock_result_from_error(format!("lock not found for release: {lock_id}"))
+    }
+}
+
 /// Minimal NodeService implementation that only supports Lock RPCs
 /// Used for testing distributed lock scenarios with real gRPC
 #[derive(Debug)]
@@ -102,7 +114,7 @@ impl NodeService for MinimalLockNodeService {
                 let lock_info_json = result.lock_info.as_ref().and_then(|info| serde_json::to_string(info).ok());
                 Ok(Response::new(GenerallyLockResponse {
                     success: result.success,
-                    error_info: None,
+                    error_info: result.error,
                     lock_info: lock_info_json,
                 }))
             }
@@ -131,11 +143,14 @@ impl NodeService for MinimalLockNodeService {
         };
 
         match self.lock_client.release(&args.lock_id).await {
-            Ok(success) => Ok(Response::new(GenerallyLockResponse {
-                success,
-                error_info: None,
-                lock_info: None,
-            })),
+            Ok(success) => {
+                let result = lock_result_from_release(&args.lock_id, success);
+                Ok(Response::new(GenerallyLockResponse {
+                    success: result.success,
+                    error_info: result.error_info,
+                    lock_info: None,
+                }))
+            }
             Err(err) => Ok(Response::new(GenerallyLockResponse {
                 success: false,
                 error_info: Some(format!(
@@ -161,11 +176,14 @@ impl NodeService for MinimalLockNodeService {
         };
 
         match self.lock_client.force_release(&args.lock_id).await {
-            Ok(success) => Ok(Response::new(GenerallyLockResponse {
-                success,
-                error_info: None,
-                lock_info: None,
-            })),
+            Ok(success) => {
+                let result = lock_result_from_release(&args.lock_id, success);
+                Ok(Response::new(GenerallyLockResponse {
+                    success: result.success,
+                    error_info: result.error_info,
+                    lock_info: None,
+                }))
+            }
             Err(err) => Ok(Response::new(GenerallyLockResponse {
                 success: false,
                 error_info: Some(format!(
@@ -271,10 +289,9 @@ impl NodeService for MinimalLockNodeService {
                 Ok(batch_results) => {
                     for (result_idx, success) in batch_results.into_iter().enumerate() {
                         if let Some(request_idx) = valid_indices.get(result_idx) {
-                            results[*request_idx] = GenerallyLockResult {
-                                success,
-                                error_info: None,
-                                lock_info: None,
+                            results[*request_idx] = match lock_ids.get(result_idx) {
+                                Some(lock_id) => lock_result_from_release(lock_id, success),
+                                None => lock_result_from_error(format!("unlock response index out of range: {result_idx}")),
                             };
                         }
                     }

crates/e2e_test/src/reliant/lock.rs

@@ -275,6 +275,41 @@ async fn test_grpc_lock_client_batch_acquire_and_release() {
     handle.abort();
 }
 
+#[tokio::test]
+async fn test_grpc_lock_client_uses_request_lock_id_and_reports_missing_unlock() {
+    let manager = Arc::new(GlobalLockManager::new());
+    let local_client: Arc<dyn rustfs_lock::LockClient> = Arc::new(LocalClient::with_manager(manager));
+
+    let (addr, handle) = spawn_lock_server(local_client).await.expect("Failed to spawn server");
+    tokio::time::sleep(Duration::from_millis(100)).await;
+
+    let grpc_client = GrpcLockClient::new(addr);
+    let request = LockRequest::new(test_resource(), LockType::Exclusive, "owner-a").with_acquire_timeout(Duration::from_secs(2));
+
+    let response = grpc_client.acquire_lock(&request).await.expect("gRPC acquire should succeed");
+    let lock_info = response.lock_info.expect("gRPC acquire should include lock info");
+    assert_eq!(lock_info.id, request.lock_id);
+
+    assert!(
+        grpc_client
+            .release(&request.lock_id)
+            .await
+            .expect("gRPC release should succeed"),
+        "release should find the request lock id"
+    );
+
+    let missing_release = grpc_client
+        .release(&request.lock_id)
+        .await
+        .expect_err("second release should report missing lock");
+    assert!(
+        missing_release.to_string().contains("lock not found for release"),
+        "missing release should preserve server error, got: {missing_release}"
+    );
+
+    handle.abort();
+}
+
 #[tokio::test]
 async fn test_distributed_lock_4_nodes_grpc_read_write_quorum_split_with_two_failed_nodes() {
     let manager1 = Arc::new(GlobalLockManager::new());

crates/ecstore/src/disk/local.rs

@@ -2072,6 +2072,7 @@ impl DiskAPI for LocalDisk {
 
         let mut objs_returned = 0;
 
+        let mut skip_current_dir_object = false;
         if opts.base_dir.ends_with(SLASH_SEPARATOR) {
             if let Ok(data) = self
                 .read_metadata(
@@ -2098,7 +2099,7 @@ impl DiskAPI for LocalDisk {
                 if let Ok(meta) = tokio::fs::metadata(fpath).await
                     && meta.is_file()
                 {
-                    return Err(DiskError::FileNotFound);
+                    skip_current_dir_object = true;
                 }
             }
         }
@@ -2109,7 +2110,7 @@ impl DiskAPI for LocalDisk {
             &opts,
             &mut out,
             &mut objs_returned,
-            false,
+            skip_current_dir_object,
         )
         .await?;
 

crates/ecstore/src/rpc/remote_disk.rs

@@ -886,7 +886,7 @@ impl DiskAPI for RemoteDisk {
 
         self.execute_with_timeout_for_op(
             "write_metadata",
-            || async {
+            move || async move {
                 let disk = self.disk_ref().await;
                 let mut client = self
                     .get_client()
@@ -896,8 +896,8 @@ impl DiskAPI for RemoteDisk {
                     disk,
                     volume: volume.to_string(),
                     path: path.to_string(),
-                    file_info: file_info.clone(),
-                    file_info_bin: file_info_bin.clone(),
+                    file_info,
+                    file_info_bin: file_info_bin.into(),
                 });
 
                 let response = client.write_metadata(request).await?.into_inner();
@@ -951,7 +951,7 @@ impl DiskAPI for RemoteDisk {
 
         self.execute_with_timeout_for_op(
             "update_metadata",
-            || async {
+            move || async move {
                 let disk = self.disk_ref().await;
                 let mut client = self
                     .get_client()
@@ -961,10 +961,10 @@ impl DiskAPI for RemoteDisk {
                     disk,
                     volume: volume.to_string(),
                     path: path.to_string(),
-                    file_info: file_info.clone(),
-                    opts: opts_str.clone(),
-                    file_info_bin: file_info_bin.clone(),
-                    opts_bin: opts_bin.clone(),
+                    file_info,
+                    opts: opts_str,
+                    file_info_bin: file_info_bin.into(),
+                    opts_bin: opts_bin.into(),
                 });
 
                 let response = client.update_metadata(request).await?.into_inner();
@@ -994,7 +994,7 @@ impl DiskAPI for RemoteDisk {
         let opts_bin = encode_msgpack(opts)?;
 
         self.execute_with_timeout(
-            || async {
+            move || async {
                 let disk = self.disk_ref().await;
                 let mut client = self
                     .get_client()
@@ -1005,8 +1005,8 @@ impl DiskAPI for RemoteDisk {
                     volume: volume.to_string(),
                     path: path.to_string(),
                     version_id: version_id.to_string(),
-                    opts: opts_str.clone(),
-                    opts_bin: opts_bin.clone(),
+                    opts: opts_str,
+                    opts_bin: opts_bin.into(),
                 });
 
                 let response = client.read_version(request).await?.into_inner();
@@ -1480,7 +1480,7 @@ impl DiskAPI for RemoteDisk {
                 let request = Request::new(ReadMultipleRequest {
                     disk,
                     read_multiple_req,
-                    read_multiple_req_bin,
+                    read_multiple_req_bin: read_multiple_req_bin.into(),
                 });
 
                 let response = client.read_multiple(request).await?.into_inner();

crates/ecstore/src/rpc/remote_locker.rs

@@ -190,11 +190,6 @@ impl LockClient for RemoteClient {
             Err(err) => return Ok(Self::rpc_failure_response(request, &err)),
         };
 
-        // Check for explicit error first
-        if let Some(error_info) = resp.error_info {
-            return Err(LockError::internal(error_info));
-        }
-
         // Check if the lock acquisition was successful
         if resp.success {
             Ok(LockResponse::success(
@@ -204,7 +199,8 @@ impl LockClient for RemoteClient {
         } else {
             // Lock acquisition failed
             Ok(LockResponse::failure(
-                "Lock acquisition failed on remote server".to_string(),
+                resp.error_info
+                    .unwrap_or_else(|| "Lock acquisition failed on remote server".to_string()),
                 std::time::Duration::ZERO,
             ))
         }

crates/ecstore/src/set_disk.rs

@@ -18,6 +18,7 @@
 use crate::batch_processor::{AsyncBatchProcessor, get_global_processors};
 use crate::bitrot::{create_bitrot_reader, create_bitrot_writer};
 use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE;
+use crate::bucket::object_lock::objectlock_sys::check_retention_for_modification;
 use crate::bucket::replication::check_replicate_delete;
 use crate::bucket::versioning::VersioningApi;
 use crate::bucket::versioning_sys::BucketVersioningSys;
@@ -1343,6 +1344,22 @@ impl BucketOperations for SetDisks {
     }
 }
 
+fn check_object_lock_retention_update(bucket: &str, object: &str, obj_info: &ObjectInfo, opts: &ObjectOptions) -> Result<()> {
+    if let Some(retention) = &opts.object_lock_retention
+        && check_retention_for_modification(
+            &obj_info.user_defined,
+            retention.mode.as_deref(),
+            retention.retain_until,
+            retention.bypass_governance,
+        )
+        .is_some()
+    {
+        return Err(StorageError::PrefixAccessDenied(bucket.to_string(), object.to_string()));
+    }
+
+    Ok(())
+}
+
 #[async_trait::async_trait]
 impl ObjectOperations for SetDisks {
     #[tracing::instrument(skip(self))]
@@ -1989,6 +2006,8 @@ impl ObjectOperations for SetDisks {
 
         let obj_info = ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended);
 
+        check_object_lock_retention_update(bucket, object, &obj_info, opts)?;
+
         for (k, v) in obj_info.user_defined {
             fi.metadata.insert(k, v);
         }
@@ -5483,6 +5502,74 @@ mod tests {
         assert!(rendered.contains("object"), "{rendered}");
     }
 
+    #[test]
+    fn test_check_object_lock_retention_update_blocks_compliance_shorten() {
+        let now = OffsetDateTime::now_utc();
+        let existing_until = now + Duration::from_secs(60 * 60 * 24 * 60);
+        let requested_until = now + Duration::from_secs(60 * 60 * 24);
+
+        let mut user_defined = HashMap::new();
+        user_defined.insert(
+            X_AMZ_OBJECT_LOCK_MODE.as_str().to_string(),
+            s3s::dto::ObjectLockRetentionMode::COMPLIANCE.to_string(),
+        );
+        user_defined.insert(
+            X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE.as_str().to_string(),
+            existing_until.format(&time::format_description::well_known::Rfc3339).unwrap(),
+        );
+
+        let obj_info = ObjectInfo {
+            user_defined,
+            ..Default::default()
+        };
+        let opts = ObjectOptions {
+            object_lock_retention: Some(crate::store_api::ObjectLockRetentionOptions {
+                mode: Some(s3s::dto::ObjectLockRetentionMode::COMPLIANCE.to_string()),
+                retain_until: Some(requested_until),
+                bypass_governance: true,
+            }),
+            ..Default::default()
+        };
+
+        let err = check_object_lock_retention_update("bucket", "object", &obj_info, &opts)
+            .expect_err("COMPLIANCE shortening must be blocked");
+
+        assert!(matches!(err, StorageError::PrefixAccessDenied(_, _)));
+    }
+
+    #[test]
+    fn test_check_object_lock_retention_update_allows_governance_shorten_with_bypass() {
+        let now = OffsetDateTime::now_utc();
+        let existing_until = now + Duration::from_secs(60 * 60 * 24 * 60);
+        let requested_until = now + Duration::from_secs(60 * 60 * 24);
+
+        let mut user_defined = HashMap::new();
+        user_defined.insert(
+            X_AMZ_OBJECT_LOCK_MODE.as_str().to_string(),
+            s3s::dto::ObjectLockRetentionMode::GOVERNANCE.to_string(),
+        );
+        user_defined.insert(
+            X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE.as_str().to_string(),
+            existing_until.format(&time::format_description::well_known::Rfc3339).unwrap(),
+        );
+
+        let obj_info = ObjectInfo {
+            user_defined,
+            ..Default::default()
+        };
+        let opts = ObjectOptions {
+            object_lock_retention: Some(crate::store_api::ObjectLockRetentionOptions {
+                mode: Some(s3s::dto::ObjectLockRetentionMode::GOVERNANCE.to_string()),
+                retain_until: Some(requested_until),
+                bypass_governance: true,
+            }),
+            ..Default::default()
+        };
+
+        check_object_lock_retention_update("bucket", "object", &obj_info, &opts)
+            .expect("GOVERNANCE shortening with bypass should remain allowed");
+    }
+
     #[test]
     fn test_should_prevent_write() {
         let oi = ObjectInfo {

crates/ecstore/src/store_api/types.rs

@@ -43,6 +43,13 @@ impl HTTPPreconditions {
     }
 }
 
+#[derive(Debug, Default, Clone)]
+pub struct ObjectLockRetentionOptions {
+    pub mode: Option<String>,
+    pub retain_until: Option<OffsetDateTime>,
+    pub bypass_governance: bool,
+}
+
 #[derive(Debug, Default, Clone)]
 pub struct ObjectOptions {
     // Use the maximum parity (N/2), used when saving server configuration files
@@ -79,6 +86,7 @@ pub struct ObjectOptions {
     pub lifecycle_audit_event: LcAuditEvent,
 
     pub eval_metadata: Option<HashMap<String, String>>,
+    pub object_lock_retention: Option<ObjectLockRetentionOptions>,
 
     pub want_checksum: Option<Checksum>,
     pub skip_verify_bitrot: bool,

crates/ecstore/tests/protobuf_bytes_regression_test.rs

@@ -0,0 +1,36 @@
+// Copyright (c) RustFS contributors
+// SPDX-License-Identifier: Apache-2.0
+
+use bytes::Bytes;
+use rustfs_protos::proto_gen::node_service::{
+    ReadMultipleRequest, ReadMultipleResponse, ReadVersionResponse, ReadXlResponse, UpdateMetadataRequest, WriteMetadataRequest,
+};
+
+fn expect_bytes(_: &Bytes) {}
+
+#[test]
+fn protobuf_bytes_fields_use_bytes_consistently() {
+    let update = UpdateMetadataRequest::default();
+    expect_bytes(&update.file_info_bin);
+    expect_bytes(&update.opts_bin);
+
+    let write = WriteMetadataRequest::default();
+    expect_bytes(&write.file_info_bin);
+
+    let version = ReadVersionResponse::default();
+    expect_bytes(&version.file_info_bin);
+
+    let read_xl = ReadXlResponse::default();
+    expect_bytes(&read_xl.raw_file_info_bin);
+
+    let read_multiple = ReadMultipleRequest::default();
+    expect_bytes(&read_multiple.read_multiple_req_bin);
+
+    let read_multiple_response = ReadMultipleResponse::default();
+    let first = read_multiple_response
+        .read_multiple_resps_bin
+        .first()
+        .cloned()
+        .unwrap_or_default();
+    expect_bytes(&first);
+}

crates/lock/src/distributed_lock.rs

@@ -22,13 +22,15 @@ use futures::future::join_all;
 use rustfs_io_metrics::{
     record_read_lock_held_acquire, record_read_lock_held_release, record_write_lock_held_acquire, record_write_lock_held_release,
 };
-use std::sync::{Arc, LazyLock};
+use std::sync::Arc;
 use std::time::Duration;
-use tokio::sync::mpsc;
 use tokio::task::JoinSet;
-use tracing::warn;
+use tracing::{debug, warn};
 use uuid::Uuid;
 
+const UNLOCK_RETRY_ATTEMPTS: usize = 3;
+const UNLOCK_RETRY_BACKOFF: Duration = Duration::from_millis(100);
+
 /// Generate a new aggregate lock ID for multiple client locks
 fn generate_aggregate_lock_id(resource: &ObjectKey) -> LockId {
     LockId {
@@ -37,45 +39,6 @@ fn generate_aggregate_lock_id(resource: &ObjectKey) -> LockId {
     }
 }
 
-#[derive(Debug, Clone)]
-struct UnlockJob {
-    /// Entries to release: each (LockId, client) pair will be released independently.
-    entries: Vec<(LockId, Arc<dyn LockClient>)>,
-}
-
-#[derive(Debug)]
-struct UnlockRuntime {
-    tx: mpsc::Sender<UnlockJob>,
-}
-
-// Global unlock runtime with background worker
-static UNLOCK_RUNTIME: LazyLock<UnlockRuntime> = LazyLock::new(|| {
-    // Larger buffer to reduce contention during bursts
-    let (tx, mut rx) = mpsc::channel::<UnlockJob>(8192);
-
-    // Spawn background worker when first used; assumes a Tokio runtime is available
-    tokio::spawn(async move {
-        while let Some(job) = rx.recv().await {
-            // Best-effort release across all (LockId, client) entries.
-            let results = join_all(
-                job.entries
-                    .into_iter()
-                    .map(|(lock_id, client)| async move { client.release(&lock_id).await.unwrap_or(false) }),
-            )
-            .await;
-            let any_ok = results.into_iter().any(|released| released);
-
-            if !any_ok {
-                tracing::warn!("DistributedLockGuard background release failed for one or more entries");
-            } else {
-                tracing::debug!("DistributedLockGuard background released one or more entries");
-            }
-        }
-    });
-
-    UnlockRuntime { tx }
-});
-
 /// A RAII guard for distributed locks that releases the lock asynchronously when dropped.
 #[derive(Debug)]
 pub struct DistributedLockGuard {
@@ -126,47 +89,22 @@ impl DistributedLockGuard {
     }
 
     /// Manually release the lock early.
-    /// This sends a release job to the background worker and then disarms the guard
+    /// This spawns a background release task and then disarms the guard
     /// to prevent double-release on drop.
-    /// Returns true if the lock was released (or was already released), false otherwise.
+    /// Returns true if release was scheduled or the guard was already disarmed.
     pub fn release(&mut self) -> bool {
         if self.disarmed {
             // Lock was already released, return true to indicate lock is in released state
             return true;
         }
 
-        let job = UnlockJob {
-            entries: self.entries.clone(),
-        };
-
-        // Try a non-blocking send to avoid panics
-        let success = if let Err(err) = UNLOCK_RUNTIME.tx.try_send(job) {
-            // Channel full or closed; best-effort fallback: spawn a detached task
-            let entries = self.entries.clone();
-            tracing::warn!(
-                "DistributedLockGuard channel send failed ({}), spawning fallback unlock task for {} entries",
-                err,
-                entries.len()
-            );
-
-            // If runtime is not available, this will panic; but in RustFS we are inside Tokio contexts.
-            let handle = tokio::spawn(async move {
-                let futures_iter = entries
-                    .into_iter()
-                    .map(|(lock_id, client)| async move { client.release(&lock_id).await.unwrap_or(false) });
-                let _ = join_all(futures_iter).await;
-            });
-            // Explicitly drop the JoinHandle to acknowledge detaching the task.
-            drop(handle);
-            true // Consider it successful even if we had to use fallback
-        } else {
-            true
-        };
+        let entries = self.entries.clone();
+        DistributedLock::spawn_release_cleanup(entries, "distributed_lock_guard_release");
 
         // Disarm to prevent double-release on drop
         self.disarmed = true;
         record_lock_held_release(self.lock_type);
-        success
+        true
     }
 }
 
@@ -349,22 +287,64 @@ impl DistributedLock {
         pending
     }
 
-    async fn release_entries(entries: &[(LockId, Arc<dyn LockClient>)], context: &'static str) {
-        let release_results = join_all(
-            entries
-                .iter()
-                .map(|(lock_id, client)| async move { (lock_id, client.release(lock_id).await) }),
-        )
-        .await;
+    async fn release_entries(entries: Vec<(LockId, Arc<dyn LockClient>)>, context: &'static str) {
+        let mut pending = entries;
 
-        for (lock_id, result) in release_results {
-            match result {
-                Ok(true) | Ok(false) => {}
-                Err(err) => {
-                    tracing::warn!("{context}: failed to release lock {} on client: {}", lock_id, err);
+        for attempt in 1..=UNLOCK_RETRY_ATTEMPTS {
+            let release_results = join_all(pending.into_iter().map(|(lock_id, client)| async move {
+                match client.release(&lock_id).await {
+                    Ok(true) => None,
+                    Ok(false) => {
+                        warn!(%lock_id, attempt, context, "distributed unlock did not find lock on client");
+                        Some((lock_id, client))
+                    }
+                    Err(err) => {
+                        warn!(%lock_id, attempt, context, "distributed unlock failed on client: {}", err);
+                        Some((lock_id, client))
+                    }
                 }
+            }))
+            .await;
+
+            pending = release_results.into_iter().flatten().collect();
+            if pending.is_empty() {
+                debug!(attempt, context, "distributed unlock completed");
+                return;
+            }
+
+            if attempt < UNLOCK_RETRY_ATTEMPTS {
+                tokio::time::sleep(UNLOCK_RETRY_BACKOFF * attempt as u32).await;
             }
         }
+
+        warn!(
+            remaining = pending.len(),
+            attempts = UNLOCK_RETRY_ATTEMPTS,
+            context,
+            "distributed unlock left unreleased entries after retry"
+        );
+    }
+
+    fn spawn_release_cleanup(entries: Vec<(LockId, Arc<dyn LockClient>)>, context: &'static str) {
+        if entries.is_empty() {
+            return;
+        }
+
+        if let Ok(handle) = tokio::runtime::Handle::try_current() {
+            let join_handle = handle.spawn(async move {
+                Self::release_entries(entries, context).await;
+            });
+            drop(join_handle);
+            return;
+        }
+
+        let join_handle = std::thread::spawn(move || match tokio::runtime::Builder::new_current_thread().enable_all().build() {
+            Ok(runtime) => runtime.block_on(async move {
+                Self::release_entries(entries, context).await;
+            }),
+            Err(err) => warn!(context, "failed to create fallback unlock runtime: {}", err),
+        });
+        drop(join_handle);
     }
 
     fn spawn_pending_cleanup(
@@ -387,9 +367,7 @@ impl DistributedLock {
                             continue;
                         };
 
-                        if let Err(err) = client.release(&lock_id).await {
-                            tracing::warn!("{context}: failed to cleanup late lock {} on client {}: {}", lock_id, idx, err);
-                        }
+                        Self::release_entries(vec![(lock_id, client.clone())], context).await;
                     }
                     Ok((idx, Ok(resp))) => {
                         tracing::debug!(
@@ -506,7 +484,7 @@ impl DistributedLock {
 
             if individual_locks.len() + pending.len() < required_quorum {
                 let rollback_count = individual_locks.len();
-                Self::release_entries(&individual_locks, "distributed_lock_quorum_rollback").await;
+                Self::spawn_release_cleanup(individual_locks.clone(), "distributed_lock_quorum_rollback");
                 if !pending.is_empty() {
                     Self::spawn_pending_cleanup(
                         pending,
@@ -525,7 +503,7 @@ impl DistributedLock {
         }
 
         let rollback_count = individual_locks.len();
-        Self::release_entries(&individual_locks, "distributed_lock_quorum_rollback").await;
+        Self::spawn_release_cleanup(individual_locks.clone(), "distributed_lock_quorum_rollback");
         let resp = LockResponse::failure(
             format!("Failed to acquire quorum: {rollback_count}/{required_quorum} required"),
             Duration::ZERO,

crates/lock/src/namespace/tests.rs

@@ -16,7 +16,10 @@ use super::*;
 use crate::client::{ClientFactory, local::LocalClient};
 use crate::types::LockType;
 use crate::{GlobalLockManager, LockError, LockInfo, LockResponse, LockStats};
-use std::sync::Arc;
+use std::sync::{
+    Arc,
+    atomic::{AtomicUsize, Ordering},
+};
 use std::time::Duration;
 
 #[derive(Debug, Default)]
@@ -107,6 +110,75 @@ impl crate::client::LockClient for DelayedClient {
     }
 }
 
+#[derive(Debug)]
+struct FlakyReleaseClient {
+    inner: LocalClient,
+    failed_releases_remaining: AtomicUsize,
+    release_attempts: AtomicUsize,
+}
+
+impl FlakyReleaseClient {
+    fn new(manager: Arc<GlobalLockManager>, failed_releases: usize) -> Self {
+        Self {
+            inner: LocalClient::with_manager(manager),
+            failed_releases_remaining: AtomicUsize::new(failed_releases),
+            release_attempts: AtomicUsize::new(0),
+        }
+    }
+
+    fn release_attempts(&self) -> usize {
+        self.release_attempts.load(Ordering::SeqCst)
+    }
+}
+
+#[async_trait::async_trait]
+impl crate::client::LockClient for FlakyReleaseClient {
+    async fn acquire_lock(&self, request: &LockRequest) -> crate::Result<LockResponse> {
+        self.inner.acquire_lock(request).await
+    }
+
+    async fn release(&self, lock_id: &LockId) -> crate::Result<bool> {
+        self.release_attempts.fetch_add(1, Ordering::SeqCst);
+        if self
+            .failed_releases_remaining
+            .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |remaining| remaining.checked_sub(1))
+            .is_ok()
+        {
+            return Ok(false);
+        }
+
+        self.inner.release(lock_id).await
+    }
+
+    async fn refresh(&self, lock_id: &LockId) -> crate::Result<bool> {
+        self.inner.refresh(lock_id).await
+    }
+
+    async fn force_release(&self, lock_id: &LockId) -> crate::Result<bool> {
+        self.inner.force_release(lock_id).await
+    }
+
+    async fn check_status(&self, lock_id: &LockId) -> crate::Result<Option<LockInfo>> {
+        self.inner.check_status(lock_id).await
+    }
+
+    async fn get_stats(&self) -> crate::Result<LockStats> {
+        self.inner.get_stats().await
+    }
+
+    async fn close(&self) -> crate::Result<()> {
+        self.inner.close().await
+    }
+
+    async fn is_online(&self) -> bool {
+        true
+    }
+
+    async fn is_local(&self) -> bool {
+        true
+    }
+}
+
 fn create_test_object_key(bucket: &str, object: &str) -> ObjectKey {
     ObjectKey {
         bucket: Arc::from(bucket),
@@ -115,6 +187,41 @@ fn create_test_object_key(bucket: &str, object: &str) -> ObjectKey {
     }
 }
 
+async fn wait_until_all_managers_can_write(managers: &[Arc<GlobalLockManager>], resource: ObjectKey) {
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
+
+    loop {
+        let mut guards = Vec::with_capacity(managers.len());
+        let mut all_available = true;
+
+        for (idx, manager) in managers.iter().enumerate() {
+            let local_lock = NamespaceLock::with_local_manager(format!("probe-node-{idx}"), manager.clone());
+            match local_lock
+                .get_write_lock(resource.clone(), "probe-owner", Duration::from_millis(20))
+                .await
+            {
+                Ok(guard) => guards.push(guard),
+                Err(_) => {
+                    all_available = false;
+                    break;
+                }
+            }
+        }
+
+        drop(guards);
+
+        if all_available {
+            return;
+        }
+
+        assert!(
+            tokio::time::Instant::now() < deadline,
+            "distributed lock was not released on all simulated nodes"
+        );
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+}
+
 #[tokio::test]
 async fn test_namespace_lock_new() {
     let client = ClientFactory::create_local();
@@ -174,6 +281,24 @@ async fn test_lock_client_default_batch_acquire_and_release() {
     assert_eq!(released, vec![true, true]);
 }
 
+#[tokio::test]
+async fn test_local_client_uses_request_lock_id_for_release() {
+    let manager = Arc::new(GlobalLockManager::new());
+    let client = LocalClient::with_manager(manager);
+    let resource = create_test_object_key("bucket", "object");
+    let request = LockRequest::new(resource.clone(), LockType::Exclusive, "owner-a").with_acquire_timeout(Duration::from_secs(1));
+
+    let response = client.acquire_lock(&request).await.unwrap();
+    let lock_info = response.lock_info.expect("successful acquire should return lock info");
+    assert_eq!(lock_info.id, request.lock_id);
+
+    assert!(client.release(&request.lock_id).await.unwrap());
+
+    let second_request = LockRequest::new(resource, LockType::Exclusive, "owner-b").with_acquire_timeout(Duration::from_secs(1));
+    let second_response = client.acquire_lock(&second_request).await.unwrap();
+    assert!(second_response.success);
+}
+
 #[tokio::test]
 async fn test_namespace_lock_get_resource_key() {
     let client = ClientFactory::create_local();
@@ -488,6 +613,97 @@ async fn test_namespace_lock_distributed_with_clients_and_quorum() {
     drop(guard_b);
 }
 
+#[tokio::test]
+async fn test_namespace_lock_distributed_eight_node_write_releases_all_nodes() {
+    let managers = (0..8).map(|_| Arc::new(GlobalLockManager::new())).collect::<Vec<_>>();
+    let clients = managers
+        .iter()
+        .map(|manager| Arc::new(LocalClient::with_manager(manager.clone())) as Arc<dyn LockClient>)
+        .collect::<Vec<_>>();
+
+    let lock = NamespaceLock::with_clients_and_quorum("eight-node".to_string(), clients, 5);
+    let resource = create_test_object_key("bucket", "object-eight-node");
+
+    let mut guard = lock
+        .get_write_lock(resource.clone(), "owner-a", Duration::from_secs(1))
+        .await
+        .expect("owner-a should acquire write lock across eight simulated nodes");
+
+    let err = lock
+        .get_write_lock(resource.clone(), "owner-b", Duration::from_millis(100))
+        .await
+        .expect_err("owner-b should not acquire while owner-a holds all node locks");
+    let err_str = err.to_string();
+    assert!(
+        err_str.contains("required 5") && err_str.contains("achieved"),
+        "expected 8-node quorum failure below required write quorum, got: {err}"
+    );
+
+    assert!(guard.release(), "distributed guard should enqueue release");
+    wait_until_all_managers_can_write(&managers, resource).await;
+}
+
+#[tokio::test]
+async fn test_namespace_lock_distributed_unlock_retries_release_false() {
+    let managers = (0..3).map(|_| Arc::new(GlobalLockManager::new())).collect::<Vec<_>>();
+    let flaky_clients = managers
+        .iter()
+        .map(|manager| Arc::new(FlakyReleaseClient::new(manager.clone(), 1)))
+        .collect::<Vec<_>>();
+    let clients = flaky_clients
+        .iter()
+        .map(|client| client.clone() as Arc<dyn LockClient>)
+        .collect::<Vec<_>>();
+
+    let lock = NamespaceLock::with_clients("flaky-release".to_string(), clients);
+    let resource = create_test_object_key("bucket", "object-flaky-release");
+
+    let mut guard = lock
+        .get_write_lock(resource.clone(), "owner-a", Duration::from_secs(1))
+        .await
+        .expect("owner-a should acquire write lock before flaky release");
+
+    assert!(guard.release(), "distributed guard should enqueue release");
+    wait_until_all_managers_can_write(&managers, resource).await;
+
+    assert!(
+        flaky_clients.iter().all(|client| client.release_attempts() >= 2),
+        "each simulated node should be retried after an initial false release"
+    );
+}
+
+#[test]
+fn test_namespace_lock_distributed_drop_without_runtime_does_not_panic() {
+    let (manager, resource, guard) = {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime should be created");
+        runtime.block_on(async {
+            let manager = Arc::new(GlobalLockManager::new());
+            let resource = create_test_object_key("bucket", "object-drop-no-runtime");
+            let lock = NamespaceLock::with_clients(
+                "drop-no-runtime".to_string(),
+                vec![Arc::new(LocalClient::with_manager(manager.clone()))],
+            );
+            let guard = lock
+                .get_write_lock(resource.clone(), "owner-a", Duration::from_secs(1))
+                .await
+                .expect("lock should be acquired");
+            (manager, resource, guard)
+        })
+    };
+
+    let drop_result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| drop(guard)));
+    assert!(drop_result.is_ok(), "dropping distributed guard without runtime should not panic");
+
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .expect("test runtime should be created");
+    runtime.block_on(wait_until_all_managers_can_write(&[manager], resource));
+}
+
 #[tokio::test]
 async fn test_namespace_lock_distributed_read_lock_succeeds_with_two_nodes_one_offline() {
     let manager = Arc::new(GlobalLockManager::new());
@@ -568,6 +784,39 @@ async fn test_namespace_lock_distributed_quorum_failure_rolls_back_successful_no
     drop(guard2);
 }
 
+#[tokio::test]
+async fn test_namespace_lock_distributed_quorum_rollback_retries_release_false() {
+    let managers = (0..2).map(|_| Arc::new(GlobalLockManager::new())).collect::<Vec<_>>();
+    let flaky_clients = managers
+        .iter()
+        .map(|manager| Arc::new(FlakyReleaseClient::new(manager.clone(), 1)))
+        .collect::<Vec<_>>();
+    let clients = vec![
+        flaky_clients[0].clone() as Arc<dyn LockClient>,
+        flaky_clients[1].clone() as Arc<dyn LockClient>,
+        Arc::new(FailingClient) as Arc<dyn LockClient>,
+    ];
+    let resource = create_test_object_key("bucket", "object-rollback-retry");
+    let lock = NamespaceLock::with_clients_and_quorum("rollback-retry".to_string(), clients, 3);
+
+    let err = lock
+        .get_write_lock(resource.clone(), "owner-a", Duration::from_millis(100))
+        .await
+        .expect_err("write lock should fail when quorum requires the offline node");
+
+    let err_str = err.to_string().to_lowercase();
+    assert!(
+        err_str.contains("quorum") || err_str.contains("not reached"),
+        "expected quorum error, got: {err}"
+    );
+    wait_until_all_managers_can_write(&managers, resource).await;
+
+    assert!(
+        flaky_clients.iter().all(|client| client.release_attempts() >= 2),
+        "rollback should retry node releases that initially returned false"
+    );
+}
+
 #[tokio::test]
 async fn test_namespace_lock_distributed_even_node_read_write_quorum_split() {
     let manager1 = Arc::new(GlobalLockManager::new());

crates/notify/src/integration.rs

@@ -15,7 +15,12 @@
 use crate::notification_system_subscriber::NotificationSystemSubscriberView;
 use crate::notifier::TargetList;
 use crate::{
-    Event, error::NotificationError, notifier::EventNotifier, registry::TargetRegistry, rules::BucketNotificationConfig, stream,
+    Event,
+    error::NotificationError,
+    notifier::EventNotifier,
+    registry::TargetRegistry,
+    rules::{BucketNotificationConfig, ParseConfigError},
+    stream,
 };
 use hashbrown::HashMap;
 use rustfs_config::notify::{
@@ -33,7 +38,7 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::time::{Duration, Instant};
 use tokio::sync::{RwLock, Semaphore, broadcast, mpsc};
-use tracing::{debug, error, info, warn};
+use tracing::{debug, info, warn};
 
 const MAX_RECENT_LIVE_EVENTS: usize = 1024;
 
@@ -309,7 +314,10 @@ impl NotificationSystem {
 
         let config = {
             let guard = self.config.read().await;
-            debug!("Initializing notification system with config: {:?}", *guard);
+            debug!(
+                subsystem_count = guard.0.len(),
+                "Initializing notification system with configuration summary"
+            );
             guard.clone()
         };
 
@@ -517,7 +525,10 @@ impl NotificationSystem {
                 if !changed {
                     info!("Target {} of type {} not found, no changes made.", target_name, target_type);
                 }
-                debug!("Config after remove: {:?}", config);
+                debug!(
+                    subsystem_count = config.0.len(),
+                    "Target config removal processed and configuration summary updated"
+                );
                 changed
             })
             .await;
@@ -593,7 +604,6 @@ impl NotificationSystem {
         bucket: &str,
         cfg: &BucketNotificationConfig,
     ) -> Result<(), NotificationError> {
-        self.subscriber_view.apply_bucket_config(bucket, cfg);
         let arn_list = self.notifier.get_arn_list(&cfg.region).await;
         if arn_list.is_empty() {
             return Err(NotificationError::Configuration("No targets configured".to_string()));
@@ -602,13 +612,18 @@ impl NotificationSystem {
         // Validate the configuration against the available ARNs
         if let Err(e) = cfg.validate(&cfg.region, &arn_list) {
             debug!("Bucket notification config validation region:{} failed: {}", &cfg.region, e);
-            if !e.to_string().contains("ARN not found") {
+            if !matches!(e, ParseConfigError::ArnNotFound(_)) {
                 return Err(NotificationError::BucketNotification(e.to_string()));
-            } else {
-                error!("config validate failed, err: {}", e);
             }
+            warn!(
+                bucket = %bucket,
+                region = %cfg.region,
+                error = %e,
+                "Bucket notification config references missing target ARN; keeping compatibility and loading remaining rules"
+            );
         }
 
+        self.subscriber_view.apply_bucket_config(bucket, cfg);
         let rules_map = cfg.get_rules_map();
         self.notifier.add_rules_map(bucket, rules_map.clone()).await;
         info!("Loaded notification config for bucket: {}", bucket);

crates/protos/src/generated/flatbuffers_generated/models.rs

@@ -1,55 +1,46 @@
 // automatically generated by the FlatBuffers compiler, do not modify
-
 // @generated
 
-use core::cmp::Ordering;
-use core::mem;
-
-extern crate flatbuffers;
-use self::flatbuffers::{EndianScalar, Follow};
+extern crate alloc;
 
 #[allow(unused_imports, dead_code)]
 pub mod models {
 
-    use core::cmp::Ordering;
-    use core::mem;
-
-    extern crate flatbuffers;
-    use self::flatbuffers::{EndianScalar, Follow};
+    extern crate alloc;
 
     pub enum PingBodyOffset {}
     #[derive(Copy, Clone, PartialEq)]
 
     pub struct PingBody<'a> {
-        pub _tab: flatbuffers::Table<'a>,
+        pub _tab: ::flatbuffers::Table<'a>,
     }
 
-    impl<'a> flatbuffers::Follow<'a> for PingBody<'a> {
+    impl<'a> ::flatbuffers::Follow<'a> for PingBody<'a> {
         type Inner = PingBody<'a>;
         #[inline]
         unsafe fn follow(buf: &'a [u8], loc: usize) -> Self::Inner {
             Self {
-                _tab: unsafe { flatbuffers::Table::new(buf, loc) },
+                _tab: unsafe { ::flatbuffers::Table::new(buf, loc) },
             }
         }
     }
 
     impl<'a> PingBody<'a> {
-        pub const VT_PAYLOAD: flatbuffers::VOffsetT = 4;
+        pub const VT_PAYLOAD: ::flatbuffers::VOffsetT = 4;
 
         pub const fn get_fully_qualified_name() -> &'static str {
             "models.PingBody"
         }
 
         #[inline]
-        pub unsafe fn init_from_table(table: flatbuffers::Table<'a>) -> Self {
+        pub unsafe fn init_from_table(table: ::flatbuffers::Table<'a>) -> Self {
             PingBody { _tab: table }
         }
         #[allow(unused_mut)]
-        pub fn create<'bldr: 'args, 'args: 'mut_bldr, 'mut_bldr, A: flatbuffers::Allocator + 'bldr>(
-            _fbb: &'mut_bldr mut flatbuffers::FlatBufferBuilder<'bldr, A>,
+        pub fn create<'bldr: 'args, 'args: 'mut_bldr, 'mut_bldr, A: ::flatbuffers::Allocator + 'bldr>(
+            _fbb: &'mut_bldr mut ::flatbuffers::FlatBufferBuilder<'bldr, A>,
             args: &'args PingBodyArgs<'args>,
-        ) -> flatbuffers::WIPOffset<PingBody<'bldr>> {
+        ) -> ::flatbuffers::WIPOffset<PingBody<'bldr>> {
             let mut builder = PingBodyBuilder::new(_fbb);
             if let Some(x) = args.payload {
                 builder.add_payload(x);
@@ -58,29 +49,28 @@ pub mod models {
         }
 
         #[inline]
-        pub fn payload(&self) -> Option<flatbuffers::Vector<'a, u8>> {
+        pub fn payload(&self) -> Option<::flatbuffers::Vector<'a, u8>> {
             // Safety:
             // Created from valid Table for this object
             // which contains a valid value in this slot
             unsafe {
                 self._tab
-                    .get::<flatbuffers::ForwardsUOffset<flatbuffers::Vector<'a, u8>>>(PingBody::VT_PAYLOAD, None)
+                    .get::<::flatbuffers::ForwardsUOffset<::flatbuffers::Vector<'a, u8>>>(PingBody::VT_PAYLOAD, None)
             }
         }
     }
 
-    impl flatbuffers::Verifiable for PingBody<'_> {
+    impl ::flatbuffers::Verifiable for PingBody<'_> {
         #[inline]
-        fn run_verifier(v: &mut flatbuffers::Verifier, pos: usize) -> Result<(), flatbuffers::InvalidFlatbuffer> {
-            use self::flatbuffers::Verifiable;
+        fn run_verifier(v: &mut ::flatbuffers::Verifier, pos: usize) -> Result<(), ::flatbuffers::InvalidFlatbuffer> {
             v.visit_table(pos)?
-                .visit_field::<flatbuffers::ForwardsUOffset<flatbuffers::Vector<'_, u8>>>("payload", Self::VT_PAYLOAD, false)?
+                .visit_field::<::flatbuffers::ForwardsUOffset<::flatbuffers::Vector<'_, u8>>>("payload", Self::VT_PAYLOAD, false)?
                 .finish();
             Ok(())
         }
     }
     pub struct PingBodyArgs<'a> {
-        pub payload: Option<flatbuffers::WIPOffset<flatbuffers::Vector<'a, u8>>>,
+        pub payload: Option<::flatbuffers::WIPOffset<::flatbuffers::Vector<'a, u8>>>,
     }
     impl<'a> Default for PingBodyArgs<'a> {
         #[inline]
@@ -89,18 +79,18 @@ pub mod models {
         }
     }
 
-    pub struct PingBodyBuilder<'a: 'b, 'b, A: flatbuffers::Allocator + 'a> {
-        fbb_: &'b mut flatbuffers::FlatBufferBuilder<'a, A>,
-        start_: flatbuffers::WIPOffset<flatbuffers::TableUnfinishedWIPOffset>,
+    pub struct PingBodyBuilder<'a: 'b, 'b, A: ::flatbuffers::Allocator + 'a> {
+        fbb_: &'b mut ::flatbuffers::FlatBufferBuilder<'a, A>,
+        start_: ::flatbuffers::WIPOffset<::flatbuffers::TableUnfinishedWIPOffset>,
     }
-    impl<'a: 'b, 'b, A: flatbuffers::Allocator + 'a> PingBodyBuilder<'a, 'b, A> {
+    impl<'a: 'b, 'b, A: ::flatbuffers::Allocator + 'a> PingBodyBuilder<'a, 'b, A> {
         #[inline]
-        pub fn add_payload(&mut self, payload: flatbuffers::WIPOffset<flatbuffers::Vector<'b, u8>>) {
+        pub fn add_payload(&mut self, payload: ::flatbuffers::WIPOffset<::flatbuffers::Vector<'b, u8>>) {
             self.fbb_
-                .push_slot_always::<flatbuffers::WIPOffset<_>>(PingBody::VT_PAYLOAD, payload);
+                .push_slot_always::<::flatbuffers::WIPOffset<_>>(PingBody::VT_PAYLOAD, payload);
         }
         #[inline]
-        pub fn new(_fbb: &'b mut flatbuffers::FlatBufferBuilder<'a, A>) -> PingBodyBuilder<'a, 'b, A> {
+        pub fn new(_fbb: &'b mut ::flatbuffers::FlatBufferBuilder<'a, A>) -> PingBodyBuilder<'a, 'b, A> {
             let start = _fbb.start_table();
             PingBodyBuilder {
                 fbb_: _fbb,
@@ -108,14 +98,14 @@ pub mod models {
             }
         }
         #[inline]
-        pub fn finish(self) -> flatbuffers::WIPOffset<PingBody<'a>> {
+        pub fn finish(self) -> ::flatbuffers::WIPOffset<PingBody<'a>> {
             let o = self.fbb_.end_table(self.start_);
-            flatbuffers::WIPOffset::new(o.value())
+            ::flatbuffers::WIPOffset::new(o.value())
         }
     }
 
-    impl core::fmt::Debug for PingBody<'_> {
-        fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+    impl ::core::fmt::Debug for PingBody<'_> {
+        fn fmt(&self, f: &mut ::core::fmt::Formatter<'_>) -> ::core::fmt::Result {
             let mut ds = f.debug_struct("PingBody");
             ds.field("payload", &self.payload());
             ds.finish()

crates/protos/src/generated/proto_gen/node_service.rs

@@ -467,10 +467,10 @@ pub struct UpdateMetadataRequest {
     pub file_info: ::prost::alloc::string::String,
     #[prost(string, tag = "5")]
     pub opts: ::prost::alloc::string::String,
-    #[prost(bytes = "vec", tag = "6")]
-    pub file_info_bin: ::prost::alloc::vec::Vec<u8>,
-    #[prost(bytes = "vec", tag = "7")]
-    pub opts_bin: ::prost::alloc::vec::Vec<u8>,
+    #[prost(bytes = "bytes", tag = "6")]
+    pub file_info_bin: ::prost::bytes::Bytes,
+    #[prost(bytes = "bytes", tag = "7")]
+    pub opts_bin: ::prost::bytes::Bytes,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct UpdateMetadataResponse {
@@ -490,8 +490,8 @@ pub struct WriteMetadataRequest {
     pub path: ::prost::alloc::string::String,
     #[prost(string, tag = "4")]
     pub file_info: ::prost::alloc::string::String,
-    #[prost(bytes = "vec", tag = "5")]
-    pub file_info_bin: ::prost::alloc::vec::Vec<u8>,
+    #[prost(bytes = "bytes", tag = "5")]
+    pub file_info_bin: ::prost::bytes::Bytes,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct WriteMetadataResponse {
@@ -512,8 +512,8 @@ pub struct ReadVersionRequest {
     pub version_id: ::prost::alloc::string::String,
     #[prost(string, tag = "5")]
     pub opts: ::prost::alloc::string::String,
-    #[prost(bytes = "vec", tag = "6")]
-    pub opts_bin: ::prost::alloc::vec::Vec<u8>,
+    #[prost(bytes = "bytes", tag = "6")]
+    pub opts_bin: ::prost::bytes::Bytes,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct ReadVersionResponse {
@@ -523,8 +523,8 @@ pub struct ReadVersionResponse {
     pub file_info: ::prost::alloc::string::String,
     #[prost(message, optional, tag = "3")]
     pub error: ::core::option::Option<Error>,
-    #[prost(bytes = "vec", tag = "4")]
-    pub file_info_bin: ::prost::alloc::vec::Vec<u8>,
+    #[prost(bytes = "bytes", tag = "4")]
+    pub file_info_bin: ::prost::bytes::Bytes,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct ReadXlRequest {
@@ -545,8 +545,8 @@ pub struct ReadXlResponse {
     pub raw_file_info: ::prost::alloc::string::String,
     #[prost(message, optional, tag = "3")]
     pub error: ::core::option::Option<Error>,
-    #[prost(bytes = "vec", tag = "4")]
-    pub raw_file_info_bin: ::prost::alloc::vec::Vec<u8>,
+    #[prost(bytes = "bytes", tag = "4")]
+    pub raw_file_info_bin: ::prost::bytes::Bytes,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct DeleteVersionRequest {
@@ -598,8 +598,8 @@ pub struct ReadMultipleRequest {
     pub disk: ::prost::alloc::string::String,
     #[prost(string, tag = "2")]
     pub read_multiple_req: ::prost::alloc::string::String,
-    #[prost(bytes = "vec", tag = "3")]
-    pub read_multiple_req_bin: ::prost::alloc::vec::Vec<u8>,
+    #[prost(bytes = "bytes", tag = "3")]
+    pub read_multiple_req_bin: ::prost::bytes::Bytes,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct ReadMultipleResponse {
@@ -609,8 +609,8 @@ pub struct ReadMultipleResponse {
     pub read_multiple_resps: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
     #[prost(message, optional, tag = "3")]
     pub error: ::core::option::Option<Error>,
-    #[prost(bytes = "vec", repeated, tag = "4")]
-    pub read_multiple_resps_bin: ::prost::alloc::vec::Vec<::prost::alloc::vec::Vec<u8>>,
+    #[prost(bytes = "bytes", repeated, tag = "4")]
+    pub read_multiple_resps_bin: ::prost::alloc::vec::Vec<::prost::bytes::Bytes>,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct DeleteVolumeRequest {
@@ -673,7 +673,7 @@ pub struct GenerallyLockResult {
     #[prost(string, optional, tag = "3")]
     pub lock_info: ::core::option::Option<::prost::alloc::string::String>,
 }
-#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+#[derive(Clone, PartialEq, ::prost::Message)]
 pub struct BatchGenerallyLockResponse {
     #[prost(message, repeated, tag = "1")]
     pub results: ::prost::alloc::vec::Vec<GenerallyLockResult>,
@@ -816,26 +816,6 @@ pub struct GetMetricsResponse {
     pub error_info: ::core::option::Option<::prost::alloc::string::String>,
 }
 #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct GetLiveEventsRequest {
-    #[prost(uint64, tag = "1")]
-    pub after_sequence: u64,
-    #[prost(uint32, tag = "2")]
-    pub limit: u32,
-}
-#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct GetLiveEventsResponse {
-    #[prost(bool, tag = "1")]
-    pub success: bool,
-    #[prost(bytes = "bytes", tag = "2")]
-    pub events: ::prost::bytes::Bytes,
-    #[prost(uint64, tag = "3")]
-    pub next_sequence: u64,
-    #[prost(bool, tag = "4")]
-    pub truncated: bool,
-    #[prost(string, optional, tag = "5")]
-    pub error_info: ::core::option::Option<::prost::alloc::string::String>,
-}
-#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct GetProcInfoRequest {}
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct GetProcInfoResponse {
@@ -1130,6 +1110,26 @@ pub struct LoadTransitionTierConfigResponse {
     #[prost(string, optional, tag = "2")]
     pub error_info: ::core::option::Option<::prost::alloc::string::String>,
 }
+#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct GetLiveEventsRequest {
+    #[prost(uint64, tag = "1")]
+    pub after_sequence: u64,
+    #[prost(uint32, tag = "2")]
+    pub limit: u32,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct GetLiveEventsResponse {
+    #[prost(bool, tag = "1")]
+    pub success: bool,
+    #[prost(bytes = "bytes", tag = "2")]
+    pub events: ::prost::bytes::Bytes,
+    #[prost(uint64, tag = "3")]
+    pub next_sequence: u64,
+    #[prost(bool, tag = "4")]
+    pub truncated: bool,
+    #[prost(string, optional, tag = "5")]
+    pub error_info: ::core::option::Option<::prost::alloc::string::String>,
+}
 /// Generated client implementations.
 pub mod node_service_client {
     #![allow(unused_variables, dead_code, missing_docs, clippy::wildcard_imports, clippy::let_unit_value)]
@@ -1991,21 +1991,6 @@ pub mod node_service_client {
                 .insert(GrpcMethod::new("node_service.NodeService", "GetMetrics"));
             self.inner.unary(req, path, codec).await
         }
-        pub async fn get_live_events(
-            &mut self,
-            request: impl tonic::IntoRequest<super::GetLiveEventsRequest>,
-        ) -> std::result::Result<tonic::Response<super::GetLiveEventsResponse>, tonic::Status> {
-            self.inner
-                .ready()
-                .await
-                .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?;
-            let codec = tonic_prost::ProstCodec::default();
-            let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/GetLiveEvents");
-            let mut req = request.into_request();
-            req.extensions_mut()
-                .insert(GrpcMethod::new("node_service.NodeService", "GetLiveEvents"));
-            self.inner.unary(req, path, codec).await
-        }
         pub async fn get_proc_info(
             &mut self,
             request: impl tonic::IntoRequest<super::GetProcInfoRequest>,
@@ -2383,6 +2368,21 @@ pub mod node_service_client {
                 .insert(GrpcMethod::new("node_service.NodeService", "LoadTransitionTierConfig"));
             self.inner.unary(req, path, codec).await
         }
+        pub async fn get_live_events(
+            &mut self,
+            request: impl tonic::IntoRequest<super::GetLiveEventsRequest>,
+        ) -> std::result::Result<tonic::Response<super::GetLiveEventsResponse>, tonic::Status> {
+            self.inner
+                .ready()
+                .await
+                .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?;
+            let codec = tonic_prost::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/GetLiveEvents");
+            let mut req = request.into_request();
+            req.extensions_mut()
+                .insert(GrpcMethod::new("node_service.NodeService", "GetLiveEvents"));
+            self.inner.unary(req, path, codec).await
+        }
     }
 }
 /// Generated server implementations.
@@ -2614,10 +2614,6 @@ pub mod node_service_server {
             &self,
             request: tonic::Request<super::GetMetricsRequest>,
         ) -> std::result::Result<tonic::Response<super::GetMetricsResponse>, tonic::Status>;
-        async fn get_live_events(
-            &self,
-            request: tonic::Request<super::GetLiveEventsRequest>,
-        ) -> std::result::Result<tonic::Response<super::GetLiveEventsResponse>, tonic::Status>;
         async fn get_proc_info(
             &self,
             request: tonic::Request<super::GetProcInfoRequest>,
@@ -2720,6 +2716,10 @@ pub mod node_service_server {
             &self,
             request: tonic::Request<super::LoadTransitionTierConfigRequest>,
         ) -> std::result::Result<tonic::Response<super::LoadTransitionTierConfigResponse>, tonic::Status>;
+        async fn get_live_events(
+            &self,
+            request: tonic::Request<super::GetLiveEventsRequest>,
+        ) -> std::result::Result<tonic::Response<super::GetLiveEventsResponse>, tonic::Status>;
     }
     #[derive(Debug)]
     pub struct NodeServiceServer<T> {
@@ -4250,34 +4250,6 @@ pub mod node_service_server {
                     };
                     Box::pin(fut)
                 }
-                "/node_service.NodeService/GetLiveEvents" => {
-                    #[allow(non_camel_case_types)]
-                    struct GetLiveEventsSvc<T: NodeService>(pub Arc<T>);
-                    impl<T: NodeService> tonic::server::UnaryService<super::GetLiveEventsRequest> for GetLiveEventsSvc<T> {
-                        type Response = super::GetLiveEventsResponse;
-                        type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
-                        fn call(&mut self, request: tonic::Request<super::GetLiveEventsRequest>) -> Self::Future {
-                            let inner = Arc::clone(&self.0);
-                            let fut = async move { <T as NodeService>::get_live_events(&inner, request).await };
-                            Box::pin(fut)
-                        }
-                    }
-                    let accept_compression_encodings = self.accept_compression_encodings;
-                    let send_compression_encodings = self.send_compression_encodings;
-                    let max_decoding_message_size = self.max_decoding_message_size;
-                    let max_encoding_message_size = self.max_encoding_message_size;
-                    let inner = self.inner.clone();
-                    let fut = async move {
-                        let method = GetLiveEventsSvc(inner);
-                        let codec = tonic_prost::ProstCodec::default();
-                        let mut grpc = tonic::server::Grpc::new(codec)
-                            .apply_compression_config(accept_compression_encodings, send_compression_encodings)
-                            .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size);
-                        let res = grpc.unary(method, req).await;
-                        Ok(res)
-                    };
-                    Box::pin(fut)
-                }
                 "/node_service.NodeService/GetProcInfo" => {
                     #[allow(non_camel_case_types)]
                     struct GetProcInfoSvc<T: NodeService>(pub Arc<T>);
@@ -4980,6 +4952,34 @@ pub mod node_service_server {
                     };
                     Box::pin(fut)
                 }
+                "/node_service.NodeService/GetLiveEvents" => {
+                    #[allow(non_camel_case_types)]
+                    struct GetLiveEventsSvc<T: NodeService>(pub Arc<T>);
+                    impl<T: NodeService> tonic::server::UnaryService<super::GetLiveEventsRequest> for GetLiveEventsSvc<T> {
+                        type Response = super::GetLiveEventsResponse;
+                        type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
+                        fn call(&mut self, request: tonic::Request<super::GetLiveEventsRequest>) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move { <T as NodeService>::get_live_events(&inner, request).await };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = GetLiveEventsSvc(inner);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(accept_compression_encodings, send_compression_encodings)
+                            .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size);
+                        let res = grpc.unary(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
                 _ => Box::pin(async move {
                     let mut response = http::Response::new(tonic::body::Body::default());
                     let headers = response.headers_mut();

crates/targets/src/config/loader.rs

@@ -28,6 +28,33 @@ pub fn collect_target_configs(
     collect_target_configs_from_env(config, route_prefix, target_type, valid_fields, std::env::vars())
 }
 
+fn is_sensitive_target_field(field_name: &str) -> bool {
+    let field_name = field_name.to_ascii_lowercase();
+    field_name.contains("password")
+        || field_name.contains("secret")
+        || field_name.contains("token")
+        || field_name.contains("credential")
+        || field_name.contains("private_key")
+        || field_name.contains("client_key")
+        || field_name.contains("access_key")
+        || field_name.contains("auth")
+}
+
+fn redact_target_field_value(field_name: &str, value: &str) -> String {
+    if is_sensitive_target_field(field_name) && !value.is_empty() {
+        return "***redacted***".to_string();
+    }
+    value.to_string()
+}
+
+fn redacted_target_config(config: &KVS) -> Vec<(String, String)> {
+    config
+        .0
+        .iter()
+        .map(|kv| (kv.key.clone(), redact_target_field_value(&kv.key, &kv.value)))
+        .collect()
+}
+
 pub fn collect_env_target_instance_ids(route_prefix: &str, target_type: &str, valid_fields: &HashSet<String>) -> HashSet<String> {
     collect_env_target_instance_ids_from_env(route_prefix, target_type, valid_fields, std::env::vars())
 }
@@ -104,7 +131,7 @@ where
         debug!(
             instance_id = %if instance_id == DEFAULT_DELIMITER { DEFAULT_DELIMITER } else { &instance_id },
             %field_name,
-            %value,
+            value = %redact_target_field_value(&field_name, value),
             "Parsed target environment override"
         );
         env_overrides
@@ -137,7 +164,10 @@ where
             merged_config.extend(env_instance_cfg.clone());
         }
 
-        debug!(instance_id = %id, ?merged_config, "Merged target configuration");
+        if tracing::enabled!(tracing::Level::DEBUG) {
+            let redacted_config = redacted_target_config(&merged_config);
+            debug!(instance_id = %id, ?redacted_config, "Merged target configuration");
+        }
         if is_target_enabled(&merged_config) {
             merged_configs.push((id, merged_config));
         }
@@ -148,7 +178,10 @@ where
 
 #[cfg(test)]
 mod tests {
-    use super::{collect_env_target_instance_ids_from_env, collect_target_configs_from_env};
+    use super::{
+        collect_env_target_instance_ids_from_env, collect_target_configs_from_env, redact_target_field_value,
+        redacted_target_config,
+    };
     use rustfs_config::notify::NOTIFY_ROUTE_PREFIX;
     use rustfs_config::{ENABLE_KEY, WEBHOOK_ENDPOINT, WEBHOOK_QUEUE_LIMIT};
     use rustfs_ecstore::config::{Config, KVS};
@@ -236,4 +269,40 @@ mod tests {
 
         assert_eq!(ids, HashSet::from(["primary".to_string()]));
     }
+
+    #[test]
+    fn redact_target_field_value_redacts_sensitive_fields() {
+        assert_eq!(redact_target_field_value("password", "secret"), "***redacted***");
+        assert_eq!(redact_target_field_value("auth_token", "token"), "***redacted***");
+        assert_eq!(redact_target_field_value("credentials_file", "/tmp/creds"), "***redacted***");
+    }
+
+    #[test]
+    fn redact_target_field_value_keeps_non_sensitive_fields() {
+        assert_eq!(redact_target_field_value("endpoint", "https://example.com"), "https://example.com");
+        assert_eq!(redact_target_field_value("queue_limit", "1000"), "1000");
+    }
+
+    #[test]
+    fn redacted_target_config_masks_sensitive_values_without_mutating_shape() {
+        let mut config = KVS::new();
+        config.insert("endpoint".to_string(), "https://example.com/hook".to_string());
+        config.insert("password".to_string(), "super-secret".to_string());
+        config.insert("client_key".to_string(), "private-key".to_string());
+        config.insert("auth_token".to_string(), "bearer-token".to_string());
+        config.insert("empty_secret".to_string(), String::new());
+
+        let redacted = redacted_target_config(&config);
+
+        assert_eq!(
+            redacted,
+            vec![
+                ("endpoint".to_string(), "https://example.com/hook".to_string()),
+                ("password".to_string(), "***redacted***".to_string()),
+                ("client_key".to_string(), "***redacted***".to_string()),
+                ("auth_token".to_string(), "***redacted***".to_string()),
+                ("empty_secret".to_string(), String::new()),
+            ]
+        );
+    }
 }

crates/targets/src/target/mqtt.rs

@@ -672,7 +672,7 @@ where
             bucket = %meta.bucket_name,
             object = %meta.object_name,
             event = %meta.event_name,
-            preview = %meta.best_effort_preview(&body, 256),
+            payload_len = body.len(),
             "Sending MQTT payload"
         );
 

crates/targets/src/target/webhook.rs

@@ -288,7 +288,7 @@ where
             bucket = %meta.bucket_name,
             object = %meta.object_name,
             event = %meta.event_name,
-            preview = %meta.best_effort_preview(&body, 256),
+            payload_len = body.len(),
             "Sending webhook payload"
         );
 

rustfs/src/admin/handlers/service_account.rs

@@ -925,11 +925,13 @@ impl Operation for ListServiceAccount {
         };
 
         let target_account = if query.user.as_ref().is_some_and(|v| v != &cred.access_key) {
+            // Cross-user listing must be authorized by ListServiceAccounts, matching the
+            // sibling InfoServiceAccount/InfoAccessKey/ListAccessKeysBulk handlers.
             if !iam_store
                 .is_allowed(&Args {
                     account: &cred.access_key,
                     groups: &cred.groups,
-                    action: Action::AdminAction(AdminAction::UpdateServiceAccountAdminAction),
+                    action: Action::AdminAction(AdminAction::ListServiceAccountsAdminAction),
                     bucket: "",
                     conditions: &get_condition_values(
                         &req.headers,

rustfs/src/server/http.rs

@@ -22,8 +22,8 @@ use crate::server::{
     compress::{CompressionConfig, PathAwareCompressionPredicate, PathCategoryInjectionLayer},
     hybrid::hybrid,
     layer::{
-        AdminChunkedContentLengthCompatLayer, ConditionalCorsLayer, ObjectAttributesEtagFixLayer, RedirectLayer,
-        RequestContextLayer, S3ErrorMessageCompatLayer,
+        AdminChunkedContentLengthCompatLayer, BodylessStatusFixLayer, ConditionalCorsLayer, ObjectAttributesEtagFixLayer,
+        RedirectLayer, RequestContextLayer, S3ErrorMessageCompatLayer,
     },
     tls_material::{TlsAcceptorHolder, TlsHandshakeFailureKind, TlsMaterialSnapshot, spawn_reload_loop},
 };
@@ -593,6 +593,7 @@ fn process_connection(
         // 15. ObjectAttributesEtagFixLayer           — ETag fix for GetObjectAttributes
         // 16. ConditionalCorsLayer                   — S3 API CORS
         // 17. RedirectLayer                          — console redirect (conditional)
+        // 18. BodylessStatusFixLayer                 — clears body for 1xx/204/205/304 responses
         // ─────────────────────────────────────────────────────────────
         let hybrid_service = ServiceBuilder::new()
             // NOTE: Both extension types are intentionally inserted to maintain compatibility:
@@ -735,6 +736,12 @@ fn process_connection(
             // Bucket-level CORS takes precedence when configured (handled in router.rs for OPTIONS, and in ecfs.rs for actual requests)
             .layer(ConditionalCorsLayer::new())
             .option_layer(if is_console { Some(RedirectLayer) } else { None })
+            // Must run first on responses: clear the body and remove
+            // Content-Length, Content-Type, and Transfer-Encoding for statuses
+            // that MUST NOT carry a body (1xx/204/304). Kept innermost so all
+            // other response-transforming layers see the already-bodyless
+            // response and so no layer (e.g. CORS) re-adds body headers afterward.
+            .layer(BodylessStatusFixLayer)
             .service(service);
 
         let hybrid_service = TowerToHyperService::new(hybrid_service);

rustfs/src/server/layer.rs

@@ -399,6 +399,91 @@ where
     }
 }
 
+/// Tower middleware that strips the body (and body-describing headers) from
+/// responses whose HTTP status code MUST NOT carry a body per RFC 9110 §6.4.1
+/// and §15 (1xx, 204, 205, 304).
+///
+/// The inner s3s layer serializes every `S3Error` — including 304 `NotModified`
+/// preconditions — as an XML body. Returning that body for a 304 is a protocol
+/// violation: hyper's HTTP/1.1 encoder forces the body to zero length but
+/// preserves the response, while the HTTP/2 path fills in `content-length`
+/// from the body's size hint and writes DATA frames after a HEADERS frame that
+/// should have carried END_STREAM. h2 clients (curl, browsers) and proxies see
+/// the malformed response as a connection-level failure — in the wild this
+/// surfaces as `GOAWAY error=0` on h2 and as an upstream-disconnect 5xx from
+/// reverse proxies like ngrok (`ERR_NGROK_3004`).
+#[derive(Clone)]
+pub struct BodylessStatusFixLayer;
+
+impl<S> Layer<S> for BodylessStatusFixLayer {
+    type Service = BodylessStatusFixService<S>;
+
+    fn layer(&self, inner: S) -> Self::Service {
+        BodylessStatusFixService { inner }
+    }
+}
+
+#[derive(Clone)]
+pub struct BodylessStatusFixService<S> {
+    inner: S,
+}
+
+impl<S, ReqBody, RestBody, GrpcBody> Service<HttpRequest<ReqBody>> for BodylessStatusFixService<S>
+where
+    S: Service<HttpRequest<ReqBody>, Response = Response<HybridBody<RestBody, GrpcBody>>> + Clone + Send + 'static,
+    S::Future: Send + 'static,
+    S::Error: Send + 'static,
+    ReqBody: Send + 'static,
+    RestBody: Body<Data = Bytes> + From<Bytes> + Send + 'static,
+    RestBody::Error: Into<S::Error> + Send + 'static,
+    GrpcBody: Send + 'static,
+{
+    type Response = Response<HybridBody<RestBody, GrpcBody>>;
+    type Error = S::Error;
+    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    fn call(&mut self, req: HttpRequest<ReqBody>) -> Self::Future {
+        let mut inner = self.inner.clone();
+
+        Box::pin(async move {
+            let response = inner.call(req).await?;
+            let (mut parts, body) = response.into_parts();
+
+            if !is_bodyless_status(parts.status) {
+                return Ok(Response::from_parts(parts, body));
+            }
+
+            let response = match body {
+                HybridBody::Rest { .. } => {
+                    parts.headers.remove(http::header::CONTENT_LENGTH);
+                    parts.headers.remove(http::header::CONTENT_TYPE);
+                    parts.headers.remove(http::header::TRANSFER_ENCODING);
+                    Response::from_parts(
+                        parts,
+                        HybridBody::Rest {
+                            rest_body: RestBody::from(Bytes::new()),
+                        },
+                    )
+                }
+                HybridBody::Grpc { grpc_body } => Response::from_parts(parts, HybridBody::Grpc { grpc_body }),
+            };
+
+            Ok(response)
+        })
+    }
+}
+
+fn is_bodyless_status(status: StatusCode) -> bool {
+    status.is_informational()
+        || status == StatusCode::NO_CONTENT
+        || status == StatusCode::RESET_CONTENT
+        || status == StatusCode::NOT_MODIFIED
+}
+
 fn is_xml_response(headers: &HeaderMap) -> bool {
     let is_xml = headers
         .get(http::header::CONTENT_TYPE)
@@ -1052,6 +1137,138 @@ mod tests {
         assert!(response_headers.get(cors::response::ACCESS_CONTROL_MAX_AGE).is_none());
     }
 
+    mod bodyless_status_fix {
+        use super::*;
+        use crate::server::hybrid::HybridBody;
+        use http_body_util::Empty;
+
+        // The production service takes `Request<Incoming>`, but `Incoming` can't be
+        // constructed in unit tests. `BodylessStatusFixService` doesn't inspect the
+        // request body, so parameterising over an arbitrary `B` is safe here.
+        #[derive(Clone)]
+        struct FixedResponse {
+            status: StatusCode,
+            body: Bytes,
+            content_type: Option<&'static str>,
+        }
+
+        impl<B: Send + 'static> Service<Request<B>> for FixedResponse {
+            type Response = Response<HybridBody<Full<Bytes>, Empty<Bytes>>>;
+            type Error = Infallible;
+            type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
+
+            fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+                Poll::Ready(Ok(()))
+            }
+
+            fn call(&mut self, _req: Request<B>) -> Self::Future {
+                let this = self.clone();
+                Box::pin(async move {
+                    let body = this.body.clone();
+                    let len = body.len();
+                    let mut builder = Response::builder().status(this.status);
+                    builder = builder.header(http::header::CONTENT_LENGTH, len.to_string());
+                    if let Some(ct) = this.content_type {
+                        builder = builder.header(http::header::CONTENT_TYPE, ct);
+                    }
+                    builder = builder.header(http::header::ETAG, "\"abc123\"");
+                    Ok(builder
+                        .body(HybridBody::Rest {
+                            rest_body: Full::from(body),
+                        })
+                        .expect("build response"))
+                })
+            }
+        }
+
+        fn empty_request() -> Request<()> {
+            Request::builder().uri("/").body(()).expect("request")
+        }
+
+        async fn collect_body<B: Body<Data = Bytes>>(body: B) -> Bytes
+        where
+            B::Error: std::fmt::Debug,
+        {
+            BodyExt::collect(body).await.expect("collect body").to_bytes()
+        }
+
+        #[tokio::test]
+        async fn strips_body_and_content_headers_for_304() {
+            let mut svc = BodylessStatusFixLayer.layer(FixedResponse {
+                status: StatusCode::NOT_MODIFIED,
+                body: Bytes::from_static(b"<Error><Code>NotModified</Code></Error>"),
+                content_type: Some("application/xml"),
+            });
+
+            let res = svc.call(empty_request()).await.expect("service call");
+            let (parts, body) = res.into_parts();
+
+            assert_eq!(parts.status, StatusCode::NOT_MODIFIED);
+            assert!(parts.headers.get(http::header::CONTENT_LENGTH).is_none());
+            assert!(parts.headers.get(http::header::CONTENT_TYPE).is_none());
+            assert_eq!(parts.headers.get(http::header::ETAG).unwrap(), "\"abc123\"");
+
+            let bytes = collect_body(body).await;
+            assert!(bytes.is_empty(), "304 response body must be empty");
+        }
+
+        #[tokio::test]
+        async fn strips_body_for_204() {
+            let mut svc = BodylessStatusFixLayer.layer(FixedResponse {
+                status: StatusCode::NO_CONTENT,
+                body: Bytes::from_static(b"unexpected"),
+                content_type: None,
+            });
+
+            let res = svc.call(empty_request()).await.expect("service call");
+            let (parts, body) = res.into_parts();
+
+            assert_eq!(parts.status, StatusCode::NO_CONTENT);
+            assert!(parts.headers.get(http::header::CONTENT_LENGTH).is_none());
+
+            let bytes = collect_body(body).await;
+            assert!(bytes.is_empty());
+        }
+
+        #[tokio::test]
+        async fn preserves_body_for_200() {
+            let payload = Bytes::from_static(b"hello");
+            let mut svc = BodylessStatusFixLayer.layer(FixedResponse {
+                status: StatusCode::OK,
+                body: payload.clone(),
+                content_type: Some("text/plain"),
+            });
+
+            let res = svc.call(empty_request()).await.expect("service call");
+            let (parts, body) = res.into_parts();
+
+            assert_eq!(parts.status, StatusCode::OK);
+            assert_eq!(parts.headers.get(http::header::CONTENT_TYPE).unwrap(), "text/plain");
+            assert_eq!(
+                parts.headers.get(http::header::CONTENT_LENGTH).unwrap(),
+                payload.len().to_string().as_str()
+            );
+
+            let bytes = collect_body(body).await;
+            assert_eq!(bytes, payload);
+        }
+
+        #[test]
+        fn is_bodyless_status_matches_rfc9110_statuses() {
+            assert!(is_bodyless_status(StatusCode::CONTINUE));
+            assert!(is_bodyless_status(StatusCode::SWITCHING_PROTOCOLS));
+            assert!(is_bodyless_status(StatusCode::NO_CONTENT));
+            assert!(is_bodyless_status(StatusCode::RESET_CONTENT));
+            assert!(is_bodyless_status(StatusCode::NOT_MODIFIED));
+
+            assert!(!is_bodyless_status(StatusCode::OK));
+            assert!(!is_bodyless_status(StatusCode::PARTIAL_CONTENT));
+            assert!(!is_bodyless_status(StatusCode::NOT_FOUND));
+            assert!(!is_bodyless_status(StatusCode::PRECONDITION_FAILED));
+            assert!(!is_bodyless_status(StatusCode::INTERNAL_SERVER_ERROR));
+        }
+    }
+
     #[test]
     fn test_apply_bucket_cors_result_replaces_existing_cors_headers() {
         let mut response_headers = HeaderMap::new();

rustfs/src/storage/ecfs.rs

@@ -38,7 +38,7 @@ use rustfs_ecstore::{
     },
     error::{StorageError, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found},
     new_object_layer_fn,
-    store_api::{BucketOperations, BucketOptions, ObjectOperations, ObjectOptions},
+    store_api::{BucketOperations, BucketOptions, ObjectLockRetentionOptions, ObjectOperations, ObjectOptions},
 };
 use rustfs_s3_common::{S3Operation, record_s3_op};
 use rustfs_targets::EventName;
@@ -1284,44 +1284,27 @@ impl S3 for FS {
             .as_ref()
             .and_then(|r| r.retain_until_date.as_ref())
             .map(|d| OffsetDateTime::from(d.clone()));
-        let new_mode = retention.as_ref().and_then(|r| r.mode.as_ref()).map(|mode| mode.as_str());
+        let new_mode = retention
+            .as_ref()
+            .and_then(|r| r.mode.as_ref())
+            .map(|mode| mode.as_str().to_string());
 
-        // TODO(security): Known TOCTOU race condition (fix in future PR).
-        //
-        // There is a time-of-check-time-of-use (TOCTOU) window between the retention
-        // check below (using get_object_info + check_retention_for_modification) and
-        // the actual update performed later in put_object_metadata.
-        //
-        // In theory:
-        //   * Thread A reads retention mode = GOVERNANCE and checks the bypass header.
-        //   * Thread B updates retention to COMPLIANCE mode.
-        //   * Thread A then proceeds to modify retention, still assuming GOVERNANCE,
-        //     and effectively bypasses what is now COMPLIANCE mode.
-        //
-        // This would violate the S3 spec, which states that COMPLIANCE-mode retention
-        // cannot be modified even with a bypass header.
-        //
-        // Possible fixes (to be implemented in a future change):
-        //   1. Pass the expected retention mode down to the storage layer and verify
-        //      it has not changed immediately before the update.
-        //   2. Use optimistic concurrency (e.g., version/etag) so that the update
-        //      fails if the object changed between check and update.
-        //   3. Perform the retention check inside the same lock/transaction scope as
-        //      the metadata update within the storage layer.
-        //
-        // Current mitigation: the storage layer provides a fast_lock_manager, which
-        // offers some protection, but it does not fully eliminate this race.
+        let bypass_governance = has_bypass_governance_header(&req.headers);
+        // Keep the early check for existing response behavior; put_object_metadata
+        // repeats the same check after taking the metadata write lock.
         let check_opts: ObjectOptions = get_opts(&bucket, &key, version_id.clone(), None, &req.headers)
             .await
             .map_err(ApiError::from)?;
 
-        if let Ok(existing_obj_info) = store.get_object_info(&bucket, &key, &check_opts).await {
-            let bypass_governance = has_bypass_governance_header(&req.headers);
-            if let Some(block_reason) =
-                check_retention_for_modification(&existing_obj_info.user_defined, new_mode, new_retain_until, bypass_governance)
-            {
-                return Err(S3Error::with_message(S3ErrorCode::AccessDenied, block_reason.error_message()));
-            }
+        if let Ok(existing_obj_info) = store.get_object_info(&bucket, &key, &check_opts).await
+            && let Some(block_reason) = check_retention_for_modification(
+                &existing_obj_info.user_defined,
+                new_mode.as_deref(),
+                new_retain_until,
+                bypass_governance,
+            )
+        {
+            return Err(S3Error::with_message(S3ErrorCode::AccessDenied, block_reason.error_message()));
         }
 
         let eval_metadata = parse_object_lock_retention(retention)?;
@@ -1330,10 +1313,15 @@ impl S3 for FS {
             .await
             .map_err(ApiError::from)?;
         opts.eval_metadata = Some(eval_metadata);
+        opts.object_lock_retention = Some(ObjectLockRetentionOptions {
+            mode: new_mode,
+            retain_until: new_retain_until,
+            bypass_governance,
+        });
 
         let object_info = store.put_object_metadata(&bucket, &key, &opts).await.map_err(|e| {
             error!("put_object_metadata failed, {}", e.to_string());
-            s3_error!(InternalError, "{}", e.to_string())
+            S3Error::from(ApiError::from(e))
         })?;
 
         let output = PutObjectRetentionOutput {

rustfs/src/storage/rpc/disk.rs

@@ -131,6 +131,7 @@ impl NodeService {
                         .iter()
                         .filter_map(|json_str| serde_json::from_str::<ReadMultipleResp>(json_str).ok())
                         .filter_map(|resp| encode_msgpack(&resp, "ReadMultipleResp").ok())
+                        .map(Into::into)
                         .collect();
 
                     Ok(Response::new(ReadMultipleResponse {
@@ -279,19 +280,19 @@ impl NodeService {
                         (Ok(raw_file_info), Ok(raw_file_info_bin)) => Ok(Response::new(ReadXlResponse {
                             success: true,
                             raw_file_info,
-                            raw_file_info_bin,
+                            raw_file_info_bin: raw_file_info_bin.into(),
                             error: None,
                         })),
                         (Err(err), _) => Ok(Response::new(ReadXlResponse {
                             success: false,
                             raw_file_info: String::new(),
-                            raw_file_info_bin: Vec::new(),
+                            raw_file_info_bin: Vec::new().into(),
                             error: Some(DiskError::other(format!("encode data failed: {err}")).into()),
                         })),
                         (_, Err(err)) => Ok(Response::new(ReadXlResponse {
                             success: false,
                             raw_file_info: String::new(),
-                            raw_file_info_bin: Vec::new(),
+                            raw_file_info_bin: Vec::new().into(),
                             error: Some(DiskError::other(format!("encode data failed: {err}")).into()),
                         })),
                     }
@@ -299,7 +300,7 @@ impl NodeService {
                 Err(err) => Ok(Response::new(ReadXlResponse {
                     success: false,
                     raw_file_info: String::new(),
-                    raw_file_info_bin: Vec::new(),
+                    raw_file_info_bin: Vec::new().into(),
                     error: Some(err.into()),
                 })),
             }
@@ -307,7 +308,7 @@ impl NodeService {
             Ok(Response::new(ReadXlResponse {
                 success: false,
                 raw_file_info: String::new(),
-                raw_file_info_bin: Vec::new(),
+                raw_file_info_bin: Vec::new().into(),
                 error: Some(DiskError::other("can not find disk".to_string()).into()),
             }))
         }
@@ -325,7 +326,7 @@ impl NodeService {
                     return Ok(Response::new(ReadVersionResponse {
                         success: false,
                         file_info: String::new(),
-                        file_info_bin: Vec::new(),
+                        file_info_bin: Vec::new().into(),
                         error: Some(DiskError::other(format!("decode ReadOptions failed: {err}")).into()),
                     }));
                 }
@@ -341,19 +342,19 @@ impl NodeService {
                         (Ok(file_info), Ok(file_info_bin)) => Ok(Response::new(ReadVersionResponse {
                             success: true,
                             file_info,
-                            file_info_bin,
+                            file_info_bin: file_info_bin.into(),
                             error: None,
                         })),
                         (Err(err), _) => Ok(Response::new(ReadVersionResponse {
                             success: false,
                             file_info: String::new(),
-                            file_info_bin: Vec::new(),
+                            file_info_bin: Vec::new().into(),
                             error: Some(DiskError::other(format!("encode data failed: {err}")).into()),
                         })),
                         (_, Err(err)) => Ok(Response::new(ReadVersionResponse {
                             success: false,
                             file_info: String::new(),
-                            file_info_bin: Vec::new(),
+                            file_info_bin: Vec::new().into(),
                             error: Some(DiskError::other(format!("encode data failed: {err}")).into()),
                         })),
                     }
@@ -361,7 +362,7 @@ impl NodeService {
                 Err(err) => Ok(Response::new(ReadVersionResponse {
                     success: false,
                     file_info: String::new(),
-                    file_info_bin: Vec::new(),
+                    file_info_bin: Vec::new().into(),
                     error: Some(err.into()),
                 })),
             }
@@ -369,7 +370,7 @@ impl NodeService {
             Ok(Response::new(ReadVersionResponse {
                 success: false,
                 file_info: String::new(),
-                file_info_bin: Vec::new(),
+                file_info_bin: Vec::new().into(),
                 error: Some(DiskError::other("can not find disk".to_string()).into()),
             }))
         }

rustfs/src/storage/rpc/lock.rs

@@ -30,6 +30,18 @@ fn lock_result_from_error(error: impl Into<String>) -> GenerallyLockResult {
     }
 }
 
+fn lock_result_from_release(lock_id: &rustfs_lock::LockId, success: bool) -> GenerallyLockResult {
+    if success {
+        GenerallyLockResult {
+            success: true,
+            error_info: None,
+            lock_info: None,
+        }
+    } else {
+        lock_result_from_error(format!("lock not found for release: {lock_id}"))
+    }
+}
+
 impl NodeService {
     pub(super) async fn handle_refresh(
         &self,
@@ -71,12 +83,15 @@ impl NodeService {
         };
 
         let lock_client = self.get_lock_client()?;
-        match lock_client.release(&args.lock_id).await {
-            Ok(_) => Ok(Response::new(GenerallyLockResponse {
-                success: true,
-                error_info: None,
-                lock_info: None,
-            })),
+        match lock_client.force_release(&args.lock_id).await {
+            Ok(success) => {
+                let result = lock_result_from_release(&args.lock_id, success);
+                Ok(Response::new(GenerallyLockResponse {
+                    success: result.success,
+                    error_info: result.error_info,
+                    lock_info: None,
+                }))
+            }
             Err(err) => Ok(Response::new(GenerallyLockResponse {
                 success: false,
                 error_info: Some(format!(
@@ -106,11 +121,14 @@ impl NodeService {
 
         let lock_client = self.get_lock_client()?;
         match lock_client.release(&args.lock_id).await {
-            Ok(_) => Ok(Response::new(GenerallyLockResponse {
-                success: true,
-                error_info: None,
-                lock_info: None,
-            })),
+            Ok(success) => {
+                let result = lock_result_from_release(&args.lock_id, success);
+                Ok(Response::new(GenerallyLockResponse {
+                    success: result.success,
+                    error_info: result.error_info,
+                    lock_info: None,
+                }))
+            }
             Err(err) => Ok(Response::new(GenerallyLockResponse {
                 success: false,
                 error_info: Some(format!(
@@ -146,7 +164,7 @@ impl NodeService {
                 let lock_info_json = result.lock_info.as_ref().and_then(|info| serde_json::to_string(info).ok());
                 Ok(Response::new(GenerallyLockResponse {
                     success: result.success,
-                    error_info: None,
+                    error_info: result.error,
                     lock_info: lock_info_json,
                 }))
             }
@@ -230,10 +248,9 @@ impl NodeService {
                 Ok(batch_results) => {
                     for (result_idx, success) in batch_results.into_iter().enumerate() {
                         if let Some(request_idx) = valid_indices.get(result_idx) {
-                            results[*request_idx] = GenerallyLockResult {
-                                success,
-                                error_info: None,
-                                lock_info: None,
+                            results[*request_idx] = match lock_ids.get(result_idx) {
+                                Some(lock_id) => lock_result_from_release(lock_id, success),
+                                None => lock_result_from_error(format!("unlock response index out of range: {result_idx}")),
                             };
                         }
                     }
@@ -249,3 +266,38 @@ impl NodeService {
         Ok(Response::new(BatchGenerallyLockResponse { results }))
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn test_lock_id() -> rustfs_lock::LockId {
+        rustfs_lock::LockRequest::new(rustfs_lock::ObjectKey::new("bucket", "object"), rustfs_lock::LockType::Exclusive, "owner")
+            .lock_id
+    }
+
+    #[test]
+    fn lock_result_from_release_reports_missing_lock() {
+        let lock_id = test_lock_id();
+        let result = lock_result_from_release(&lock_id, false);
+
+        assert!(!result.success);
+        assert!(result.lock_info.is_none());
+        assert!(
+            result
+                .error_info
+                .expect("missing release should include error")
+                .contains("lock not found for release")
+        );
+    }
+
+    #[test]
+    fn lock_result_from_response_preserves_lock_failure_error() {
+        let response = rustfs_lock::LockResponse::failure("lock conflict", std::time::Duration::ZERO);
+        let result = lock_result_from_response(response);
+
+        assert!(!result.success);
+        assert_eq!(result.error_info.as_deref(), Some("lock conflict"));
+        assert!(result.lock_info.is_none());
+    }
+}

rustfs/src/storage/rpc/node_service.rs

@@ -1389,8 +1389,8 @@ mod tests {
             path: "test-path".to_string(),
             file_info: "{}".to_string(),
             opts: "{}".to_string(),
-            file_info_bin: Vec::new(),
-            opts_bin: Vec::new(),
+            file_info_bin: Vec::new().into(),
+            opts_bin: Vec::new().into(),
         });
 
         let response = service.update_metadata(request).await;
@@ -1411,8 +1411,8 @@ mod tests {
             path: "test-path".to_string(),
             file_info: "invalid json".to_string(),
             opts: "{}".to_string(),
-            file_info_bin: Vec::new(),
-            opts_bin: Vec::new(),
+            file_info_bin: Vec::new().into(),
+            opts_bin: Vec::new().into(),
         });
 
         let response = service.update_metadata(request).await;
@@ -1433,8 +1433,8 @@ mod tests {
             path: "test-path".to_string(),
             file_info: "{}".to_string(),
             opts: "invalid json".to_string(),
-            file_info_bin: Vec::new(),
-            opts_bin: Vec::new(),
+            file_info_bin: Vec::new().into(),
+            opts_bin: Vec::new().into(),
         });
 
         let response = service.update_metadata(request).await;
@@ -1454,7 +1454,7 @@ mod tests {
             volume: "test-volume".to_string(),
             path: "test-path".to_string(),
             file_info: "{}".to_string(),
-            file_info_bin: Vec::new(),
+            file_info_bin: Vec::new().into(),
         });
 
         let response = service.write_metadata(request).await;
@@ -1474,7 +1474,7 @@ mod tests {
             volume: "test-volume".to_string(),
             path: "test-path".to_string(),
             file_info: "invalid json".to_string(),
-            file_info_bin: Vec::new(),
+            file_info_bin: Vec::new().into(),
         });
 
         let response = service.write_metadata(request).await;
@@ -1495,7 +1495,7 @@ mod tests {
             path: "test-path".to_string(),
             version_id: "version1".to_string(),
             opts: "{}".to_string(),
-            opts_bin: Vec::new(),
+            opts_bin: Vec::new().into(),
         });
 
         let response = service.read_version(request).await;
@@ -1517,7 +1517,7 @@ mod tests {
             path: "test-path".to_string(),
             version_id: "version1".to_string(),
             opts: "invalid json".to_string(),
-            opts_bin: Vec::new(),
+            opts_bin: Vec::new().into(),
         });
 
         let response = service.read_version(request).await;
@@ -1675,7 +1675,7 @@ mod tests {
         let request = Request::new(ReadMultipleRequest {
             disk: "invalid-disk-path".to_string(),
             read_multiple_req: "{}".to_string(),
-            read_multiple_req_bin: Vec::new(),
+            read_multiple_req_bin: Vec::new().into(),
         });
 
         let response = service.read_multiple(request).await;
@@ -1694,7 +1694,7 @@ mod tests {
         let request = Request::new(ReadMultipleRequest {
             disk: "invalid-disk-path".to_string(),
             read_multiple_req: "invalid json".to_string(),
-            read_multiple_req_bin: Vec::new(),
+            read_multiple_req_bin: Vec::new().into(),
         });
 
         let response = service.read_multiple(request).await;

scripts/run_object_batch_bench_abc.sh

@@ -34,6 +34,7 @@ SAMPLES=20000
 
 # optional hooks
 APPLY_CMD=""
+APPLY_CMD_ARR=()
 APPLY_WAIT_SECS=20
 
 EXTRA_ARGS=()
@@ -74,7 +75,7 @@ s3bench options:
 
 Hooks:
   --apply-cmd                 Optional command to apply/restart RustFS after profile env switch.
-                              Runs via: eval "$APPLY_CMD"
+                              Executed directly (no shell eval), e.g. "bash scripts/restart.sh"
   --apply-wait-secs           Wait time after apply cmd (default: 20)
 
 Extra:
@@ -110,6 +111,23 @@ validate_positive_int() {
   fi
 }
 
+parse_apply_cmd() {
+  local raw="$1"
+
+  if [[ "$raw" == *';'* || "$raw" == *'&&'* || "$raw" == *'||'* || "$raw" == *'|'* || "$raw" == *'<'* || "$raw" == *'>'* || "$raw" == *'`'* || "$raw" == *'$'* ]]; then
+    echo "ERROR: --apply-cmd does not allow shell operators or expansions; pass a plain command and args only" >&2
+    exit 1
+  fi
+
+  IFS=$' \t\n' read -r -a APPLY_CMD_ARR <<< "$raw"
+  if [[ "${#APPLY_CMD_ARR[@]}" -eq 0 ]]; then
+    echo "ERROR: --apply-cmd must not be empty" >&2
+    exit 1
+  fi
+
+  require_cmd "${APPLY_CMD_ARR[0]}"
+}
+
 parse_args() {
   while [[ $# -gt 0 ]]; do
     case "$1" in
@@ -171,6 +189,9 @@ validate_args() {
   if [[ "$TOOL" == "s3bench" ]]; then
     validate_positive_int "$SAMPLES" "--samples"
   fi
+  if [[ -n "$APPLY_CMD" ]]; then
+    parse_apply_cmd "$APPLY_CMD"
+  fi
 }
 
 setup_out_root() {
@@ -238,15 +259,17 @@ sizes_for_group() {
 
 run_apply_hook_if_needed() {
   local group="$1"
-  if [[ -z "$APPLY_CMD" ]]; then
+  if [[ "${#APPLY_CMD_ARR[@]}" -eq 0 ]]; then
     return
   fi
   echo "[${group}] running apply command..."
   if [[ "$DRY_RUN" == "true" ]]; then
-    echo "[DRY-RUN] eval \"$APPLY_CMD\""
+    printf '[DRY-RUN] '
+    printf '%q ' "${APPLY_CMD_ARR[@]}"
+    printf '\n'
     echo "[DRY-RUN] sleep $APPLY_WAIT_SECS"
   else
-    eval "$APPLY_CMD"
+    "${APPLY_CMD_ARR[@]}"
     echo "[${group}] waiting ${APPLY_WAIT_SECS}s for service readiness..."
     sleep "$APPLY_WAIT_SECS"
   fi