fix(guards): resolve main integration conflicts

Pure move produced by scripts/codemods/split_bucket_usecase.py: the struct, admission control and shared helpers stay in bucket_usecase.rs; the operations of each configuration family, their helpers and their tests move to bucket_usecase/<family>.rs, request fixtures to test_support.rs and the source-text guards to guards.rs. Empty stubs reserve the files the next batches add. No signature, body or test changes.

.config/make/tests.mak

@@ -62,8 +62,10 @@ script-tests: ## Run shell script tests
 	./scripts/check_embedded_secrets.sh --self-test
 	./scripts/check_s3s_footprint.sh --self-test
 	./scripts/test_check_test_count_ratchet.sh
+	./scripts/test_guard_source_set.sh
 	$(RUSTFS_PYTHON_BIN) ./scripts/check_test_wiring.py --self-test
 	$(RUSTFS_PYTHON_BIN) ./scripts/codemods/s3_error_to_s3_types.py --self-test
+	$(RUSTFS_PYTHON_BIN) ./scripts/codemods/split_bucket_usecase.py --self-test
 	$(RUSTFS_PYTHON_BIN) ./scripts/codemods/split_lifecycle_transition_api_test.py --self-test
 	$(RUSTFS_PYTHON_BIN) ./scripts/check_test_seams.py --self-test
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_e2e_binary.py
@@ -95,6 +97,7 @@ script-tests: ## Run shell script tests
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_junit_diff.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/s3-tests/test_report_compat.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/s3-tests/test_runner_tools.py
+	$(RUSTFS_PYTHON_BIN) ./scripts/s3-tests/test_api_coverage.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/s3-tests/test_reference_parity.py
 	bash -n ./scripts/validate_object_data_cache_cold_stampede.sh
 	$(RUSTFS_PYTHON_BIN) ./scripts/check_object_data_cache_follower_samples.py --self-test

docs/architecture/s3-compatibility-matrix.md

@@ -43,7 +43,7 @@ Counts ignore blank lines and comments; compute them from the files. The lifecyc
 
 `CopyObject` with `MetadataDirective=REPLACE` clears standard metadata fields that the request omits, including `Content-Type`; it does not retain the source type or infer a default. Clients requiring a MIME type on the copied object must send `Content-Type` with the replacement metadata. This contract is covered by `crates/e2e_test/src/copy_object_metadata_test.rs`. A client test that expects an implicit `application/octet-stream` does not match this behavior.
 
-The MinIO-style `metadata=true` listing extension returns user metadata names without the HTTP `x-amz-meta-` prefix. It is not the standard S3 `ListObjectsV2` response. Clients that expect canonical HTTP header names in `UserMetadata` must normalize the names at that boundary; ordinary HEAD/GET metadata is unaffected. See `rustfs/src/app/bucket_usecase.rs` and its serialization tests.
+The MinIO-style `metadata=true` listing extension returns user metadata names without the HTTP `x-amz-meta-` prefix. It is not the standard S3 `ListObjectsV2` response. Clients that expect canonical HTTP header names in `UserMetadata` must normalize the names at that boundary; ordinary HEAD/GET metadata is unaffected. See `rustfs/src/app/bucket_usecase/listing.rs` and its serialization tests.
 
 ## Conditional Bulk Deletion
 

rustfs/src/app/bucket_usecase/accelerate.rs

@@ -0,0 +1,15 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Bucket accelerate configuration use-case operations (reserved for the next batch).

rustfs/src/app/bucket_usecase/acl.rs

@@ -0,0 +1,15 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Bucket ACL use-case operations (reserved for the next batch).

rustfs/src/app/bucket_usecase/core.rs

@@ -0,0 +1,488 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Bucket create, delete, head and location use-case operations.
+
+use super::*;
+
+/// Answer a `CreateBucket` for a bucket that already exists.
+///
+/// A deployment owns one bucket namespace, so every requester that passed
+/// authorization belongs to the same owner as the bucket's creator; there is no
+/// second account that could trigger `BucketAlreadyExists`. That leaves the AWS
+/// account-owner semantics: the default region returns the legacy idempotent
+/// `200 OK`, any other region returns `BucketAlreadyOwnedByYou`.
+fn create_bucket_exists_response(region: &str) -> S3Result<S3Response<CreateBucketOutput>> {
+    if region == RUSTFS_REGION {
+        return Ok(S3Response::new(CreateBucketOutput::default()));
+    }
+
+    Err(S3Error::with_message(
+        S3ErrorCode::BucketAlreadyOwnedByYou,
+        "The requested bucket already exists.",
+    ))
+}
+
+impl DefaultBucketUsecase {
+    fn create_bucket_exists_result(&self) -> S3Result<S3Response<CreateBucketOutput>> {
+        let region = self
+            .global_region()
+            .map(|region| region.to_string())
+            .unwrap_or_else(|| RUSTFS_REGION.to_string());
+        create_bucket_exists_response(&region)
+    }
+
+    #[instrument(
+        level = "debug",
+        skip(self, req),
+        fields(start_time=?time::OffsetDateTime::now_utc())
+    )]
+    pub async fn execute_create_bucket(&self, req: S3Request<CreateBucketInput>) -> S3Result<S3Response<CreateBucketOutput>> {
+        rustfs_io_metrics::s3_http_metrics::observe_s3_http_operation(S3Operation::CreateBucket);
+        if let Some(location) = req
+            .input
+            .create_bucket_configuration
+            .as_ref()
+            .and_then(|config| config.location_constraint.as_ref())
+            .map(BucketLocationConstraint::as_str)
+            .filter(|location| !location.is_empty())
+        {
+            let location = if location == "EU" { "eu-west-1" } else { location };
+            let region = self.global_region();
+            if location != region.as_ref().map_or(RUSTFS_REGION, Region::as_str) {
+                return Err(S3Error::with_message(
+                    S3ErrorCode::InvalidLocationConstraint,
+                    "The specified location constraint does not match this server's region.",
+                ));
+            }
+        }
+
+        // An existing bucket needs no mutation. Answer before the mutation queue and the
+        // bucket's exclusive locks, which wait for every in-flight object write.
+        if req_info_ref(&req).is_ok()
+            && let Some(store) = self.object_store()
+            && store
+                .get_bucket_info(&req.input.bucket, &BucketOptions::default())
+                .await
+                .is_ok()
+        {
+            let helper = OperationHelper::new(&req, EventName::BucketCreated, S3Operation::CreateBucket);
+            let result = self.create_bucket_exists_result();
+            let _ = helper.complete(&result);
+            return result;
+        }
+        let usecase = self.clone();
+        await_bucket_usecase_on_fresh_task("bucket creation", move |mutation_permit| async move {
+            usecase.execute_create_bucket_inner(req, mutation_permit).await
+        })
+        .await
+    }
+
+    async fn execute_create_bucket_inner(
+        &self,
+        req: S3Request<CreateBucketInput>,
+        mutation_permit: OwnedSemaphorePermit,
+    ) -> S3Result<S3Response<CreateBucketOutput>> {
+        let helper = OperationHelper::new(&req, EventName::BucketCreated, S3Operation::CreateBucket);
+        let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
+        if req_info_ref(&req).is_err() {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Missing request info".to_string()));
+        }
+        let CreateBucketInput {
+            bucket,
+            object_lock_enabled_for_bucket,
+            ..
+        } = req.input;
+        let lock_enabled = object_lock_enabled_for_bucket.is_some_and(|v| v);
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        // Keep the local namespace mutation and its peer hook ordered across
+        // every node in this site. Otherwise a delete waiting for repair
+        // coordination can arrive after this create on remote sites.
+        let operation_bucket = bucket.clone();
+        let operation_store = store.clone();
+        let make_result = with_site_replication_bucket_mutation_lock(store, &bucket, move || async move {
+            let make_result = operation_store
+                .make_bucket(
+                    &operation_bucket,
+                    &MakeBucketOptions {
+                        force_create: false,
+                        lock_enabled,
+                        ..Default::default()
+                    },
+                )
+                .await;
+            if make_result.is_ok() {
+                crate::storage::invalidate_bucket_validation_cache(&operation_bucket);
+                if let Err(err) = site_replication_make_bucket_hook(&operation_bucket, lock_enabled).await {
+                    warn!(bucket = %operation_bucket, error = ?err, "site replication make bucket hook failed");
+                }
+            }
+            make_result
+        })
+        .await?;
+
+        // The storage/site mutation and its lock fences have finished. Keep
+        // the operation's in-flight slot while peers reload, without holding
+        // an execution slot needed by other creates and deletes.
+        drop(mutation_permit);
+
+        match make_result {
+            Ok(()) => {
+                // A peer may have cached this bucket as missing before the
+                // create. Refresh peer metadata after the namespace and
+                // metadata commit fences are released so the peer reload can
+                // acquire its bucket namespace read lock without an ABBA
+                // interleaving with creation.
+                notify_bucket_metadata_reload_with_context(
+                    bucket.clone(),
+                    "create bucket",
+                    request_context,
+                    false,
+                    self.context.clone(),
+                )
+                .await;
+            }
+            Err(StorageError::BucketExists(_)) => {
+                let result = self.create_bucket_exists_result();
+                let _ = helper.complete(&result);
+                return result;
+            }
+            Err(e) => return Err(ApiError::from(e).into()),
+        }
+
+        let output = CreateBucketOutput::default();
+        counter!("rustfs_create_bucket_total").increment(1);
+        let result = Ok(S3Response::new(output));
+        let _ = helper.complete(&result);
+        rustfs_scanner::record_dirty_usage_bucket_from_producer(
+            &bucket,
+            rustfs_scanner::SegmentInvalidationProducerIdentity::BucketMetadata,
+        );
+        result
+    }
+
+    #[instrument(level = "debug", skip(self, req))]
+    pub async fn execute_delete_bucket(&self, req: S3Request<DeleteBucketInput>) -> S3Result<S3Response<DeleteBucketOutput>> {
+        rustfs_io_metrics::s3_http_metrics::observe_s3_http_operation(S3Operation::DeleteBucket);
+        let usecase = self.clone();
+        await_bucket_usecase_on_fresh_task("bucket deletion", move |mutation_permit| async move {
+            let _mutation_permit = mutation_permit;
+            usecase.execute_delete_bucket_inner(req).await
+        })
+        .await
+    }
+
+    async fn execute_delete_bucket_inner(
+        &self,
+        mut req: S3Request<DeleteBucketInput>,
+    ) -> S3Result<S3Response<DeleteBucketOutput>> {
+        let helper = OperationHelper::new(&req, EventName::BucketRemoved, S3Operation::DeleteBucket);
+        let input = req.input.clone();
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        let force = match force_delete_header(&req.headers) {
+            Ok(value) => value.unwrap_or(false),
+            Err(_) => return Err(S3Error::with_message(S3ErrorCode::InvalidRequest, "Invalid force-delete header value")),
+        };
+
+        if force {
+            authorize_request(&mut req, Action::S3Action(S3Action::ForceDeleteBucketAction)).await?;
+        }
+
+        // Keep the local namespace mutation and its peer hook ordered across
+        // every node in this site so an older delete cannot overtake a new
+        // same-name make while it waits for repair coordination.
+        let operation_bucket = input.bucket.clone();
+        let operation_store = store.clone();
+        with_site_replication_bucket_mutation_lock(store, &input.bucket, move || async move {
+            let intent = prepare_site_replication_delete_bucket(&operation_bucket, force).await?;
+            let delete_result = operation_store
+                .delete_bucket(
+                    &operation_bucket,
+                    &DeleteBucketOptions {
+                        force,
+                        ..Default::default()
+                    },
+                )
+                .await;
+            match delete_result {
+                Ok(()) => {
+                    crate::storage::invalidate_bucket_validation_cache(&operation_bucket);
+                    if let Some(intent) = intent
+                        && let Err(err) = commit_site_replication_delete_bucket(&intent).await
+                    {
+                        warn!(bucket = %operation_bucket, error = ?err, "site replication delete bucket hook failed");
+                    }
+                    Ok::<(), S3Error>(())
+                }
+                Err(err) => {
+                    if let Some(intent) = intent {
+                        cancel_site_replication_delete_bucket(intent).await;
+                    }
+                    Err(S3Error::from(ApiError::from(err)))
+                }
+            }
+        })
+        .await??;
+
+        // Drop every cached object body for the now-deleted bucket so dead
+        // bytes do not sit resident until TTL. Covers both the normal and the
+        // force-delete path, which share this single delete_bucket call
+        // (ODC-28, backlog#1133).
+        let cache_adapter = current_object_data_cache_for_context(self.context.as_deref());
+        let _ = invalidate_object_data_cache_bucket_after_delete(&cache_adapter, &input.bucket).await;
+
+        // Re-evaluate lifecycle and replication after bucket removal.
+        rustfs_scanner::record_scanner_maintenance_change(&input.bucket);
+
+        // Notify peers to drop their cached metadata for the now-deleted bucket.
+        let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
+        notify_bucket_metadata_delete(input.bucket.clone(), request_context);
+
+        let result = Ok(S3Response::new(DeleteBucketOutput {}));
+        let _ = helper.complete(&result);
+        result
+    }
+
+    #[instrument(level = "debug", skip(self, req))]
+    pub async fn execute_head_bucket(&self, req: S3Request<HeadBucketInput>) -> S3Result<S3Response<HeadBucketOutput>> {
+        record_s3_op(S3Operation::HeadBucket);
+        let input = req.input;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&input.bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        Ok(S3Response::new(HeadBucketOutput::default()))
+    }
+
+    #[instrument(level = "debug", skip(self, req))]
+    pub async fn execute_get_bucket_location(
+        &self,
+        req: S3Request<GetBucketLocationInput>,
+    ) -> S3Result<S3Response<GetBucketLocationOutput>> {
+        let input = req.input;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&input.bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        let location_constraint = self
+            .global_region()
+            .filter(|region| region.as_str() != RUSTFS_REGION)
+            .map(|region| BucketLocationConstraint::from(region.to_string()));
+        Ok(S3Response::new(GetBucketLocationOutput { location_constraint }))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::app::bucket_usecase::test_support::{build_request, build_request_with_req_info, s3_op_total};
+    use http::Method;
+
+    #[test]
+    fn create_bucket_exists_response_returns_ok_in_default_region() {
+        let response = create_bucket_exists_response(RUSTFS_REGION).expect("recreate in us-east-1 should succeed");
+        assert_eq!(response.output.location, None);
+    }
+
+    #[test]
+    fn create_bucket_exists_response_returns_owned_by_you_outside_default_region() {
+        let err = create_bucket_exists_response("eu-west-1").expect_err("recreate outside us-east-1 should report ownership");
+        assert_eq!(err.code(), &S3ErrorCode::BucketAlreadyOwnedByYou);
+    }
+
+    #[tokio::test]
+    async fn execute_create_bucket_returns_internal_error_when_store_uninitialized() {
+        let input = CreateBucketInput::builder()
+            .bucket("test-bucket".to_string())
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::PUT);
+        let usecase = DefaultBucketUsecase::without_context();
+
+        let mut request = rustfs_io_metrics::s3_http_metrics::S3HttpRequestGuard::new("PUT");
+        let mut call = Box::pin(usecase.execute_create_bucket(req));
+        let err = std::future::poll_fn(|cx| request.in_scope(|| call.as_mut().poll(cx)))
+            .await
+            .expect_err("an uninitialized store must fail");
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+        assert_eq!(
+            request.in_scope(rustfs_io_metrics::s3_http_metrics::current_s3_http_operation_name),
+            Some("s3:CreateBucket")
+        );
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn execute_create_bucket_on_existing_bucket_does_not_wait_for_bucket_lifecycle_lock() {
+        crate::app::gating_test_env::run_large_stack_test("create-existing-bucket", || async {
+            let context = crate::app::gating_test_env::shared_gating_ambient().await;
+            let store = crate::app::gating_test_env::shared_gating_ecstore().await;
+            let bucket = format!("create-existing-{}", uuid::Uuid::new_v4().simple());
+            store
+                .make_bucket(&bucket, &MakeBucketOptions::default())
+                .await
+                .expect("create bucket");
+
+            // Object writes hold this lock shared; an exclusive guard exposes any lifecycle-lock wait.
+            let _lifecycle_guard = store
+                .lock_bucket_for_recursive_delete(&bucket)
+                .await
+                .expect("hold the bucket lifecycle lock");
+            let usecase = DefaultBucketUsecase::with_context(Some(context));
+
+            // The requester's owner flag must not change the duplicate response:
+            // one deployment has a single bucket owner.
+            for (requester_is_owner, location, expected_error) in [
+                (true, None, None),
+                (false, None, None),
+                (true, Some("ap-south-1"), Some(S3ErrorCode::InvalidLocationConstraint)),
+            ] {
+                let input = CreateBucketInput {
+                    bucket: bucket.clone(),
+                    create_bucket_configuration: location.map(|location| s3s_crate::dto::CreateBucketConfiguration {
+                        location_constraint: Some(BucketLocationConstraint::from(location.to_string())),
+                        ..Default::default()
+                    }),
+                    ..Default::default()
+                };
+                let req = build_request_with_req_info(
+                    input,
+                    Method::PUT,
+                    ReqInfo {
+                        is_owner: requester_is_owner,
+                        ..Default::default()
+                    },
+                );
+                let result = tokio::time::timeout(Duration::from_secs(2), usecase.execute_create_bucket(req))
+                    .await
+                    .expect("CreateBucket on an existing bucket must not wait for the bucket lifecycle lock");
+                match expected_error {
+                    Some(code) => assert_eq!(result.expect_err("invalid request must fail").code(), &code),
+                    None => {
+                        result.expect("recreating a bucket in us-east-1 returns the idempotent success");
+                    }
+                }
+            }
+
+            let req = build_request(
+                CreateBucketInput {
+                    bucket,
+                    ..Default::default()
+                },
+                Method::PUT,
+            );
+            let err = tokio::time::timeout(Duration::from_secs(2), usecase.execute_create_bucket(req))
+                .await
+                .expect("missing request info must fail before storage mutation")
+                .expect_err("an existing bucket must not bypass request-info validation");
+            assert_eq!(err.code(), &S3ErrorCode::InternalError);
+        });
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn execute_create_bucket_creates_missing_bucket() {
+        crate::app::gating_test_env::run_large_stack_test("create-missing-bucket", || async {
+            let context = crate::app::gating_test_env::shared_gating_ambient().await;
+            let store = crate::app::gating_test_env::shared_gating_ecstore().await;
+            let bucket = format!("create-missing-{}", uuid::Uuid::new_v4().simple());
+            let input = CreateBucketInput::builder().bucket(bucket.clone()).build().unwrap();
+            let req = build_request_with_req_info(
+                input,
+                Method::PUT,
+                ReqInfo {
+                    is_owner: true,
+                    ..Default::default()
+                },
+            );
+            let usecase = DefaultBucketUsecase::with_context(Some(context));
+            usecase.execute_create_bucket(req).await.expect("create a missing bucket");
+            let info = store
+                .get_bucket_info(&bucket, &BucketOptions::default())
+                .await
+                .expect("the missing-bucket lookup must fall through to real creation");
+            assert_eq!(info.name, bucket);
+        });
+    }
+
+    #[tokio::test]
+    async fn execute_delete_bucket_returns_internal_error_when_store_uninitialized() {
+        let input = DeleteBucketInput::builder()
+            .bucket("test-bucket".to_string())
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::DELETE);
+        let usecase = DefaultBucketUsecase::without_context();
+
+        let mut request = rustfs_io_metrics::s3_http_metrics::S3HttpRequestGuard::new("DELETE");
+        let mut call = Box::pin(usecase.execute_delete_bucket(req));
+        let err = std::future::poll_fn(|cx| request.in_scope(|| call.as_mut().poll(cx)))
+            .await
+            .expect_err("an uninitialized store must fail");
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+        assert_eq!(
+            request.in_scope(rustfs_io_metrics::s3_http_metrics::current_s3_http_operation_name),
+            Some("s3:DeleteBucket")
+        );
+    }
+
+    #[tokio::test]
+    async fn execute_head_bucket_returns_internal_error_when_store_uninitialized() {
+        let input = HeadBucketInput::builder().bucket("test-bucket".to_string()).build().unwrap();
+
+        let req = build_request(input, Method::HEAD);
+        let usecase = DefaultBucketUsecase::without_context();
+        let before = s3_op_total(S3Operation::HeadBucket);
+
+        let err = usecase.execute_head_bucket(req).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+        assert_eq!(s3_op_total(S3Operation::HeadBucket), before + 1);
+    }
+
+    #[tokio::test]
+    async fn execute_get_bucket_location_returns_internal_error_when_store_uninitialized() {
+        let input = GetBucketLocationInput::builder()
+            .bucket("test-bucket".to_string())
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::GET);
+        let usecase = DefaultBucketUsecase::without_context();
+
+        let err = usecase.execute_get_bucket_location(req).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+    }
+}

rustfs/src/app/bucket_usecase/cors.rs

@@ -0,0 +1,178 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Bucket CORS configuration use-case operations.
+
+use super::*;
+
+impl DefaultBucketUsecase {
+    #[instrument(level = "debug", skip(self, req))]
+    pub async fn execute_delete_bucket_cors(
+        &self,
+        req: S3Request<DeleteBucketCorsInput>,
+    ) -> S3Result<S3Response<DeleteBucketCorsOutput>> {
+        let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
+        let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
+        let DeleteBucketCorsInput { bucket, .. } = req.input;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        delete_bucket_config_for_incarnation(&bucket, BUCKET_CORS_CONFIG, expected_incarnation_id)
+            .await
+            .map_err(ApiError::from)?;
+
+        notify_bucket_metadata_reload(bucket.clone(), "delete bucket cors", request_context, false).await;
+
+        let item = sr_bucket_meta_item(bucket.clone(), "cors-config");
+        if let Err(err) = site_replication_bucket_meta_hook(item).await {
+            warn!(bucket = %bucket, error = ?err, "site replication bucket cors delete hook failed");
+        }
+
+        Ok(S3Response::new(DeleteBucketCorsOutput {}))
+    }
+
+    #[instrument(level = "debug", skip(self, req))]
+    pub async fn execute_get_bucket_cors(&self, req: S3Request<GetBucketCorsInput>) -> S3Result<PersistedCorsConfiguration> {
+        let GetBucketCorsInput { bucket, .. } = req.input;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        let cors_configuration = match metadata_sys::get_cors_config(&bucket).await {
+            Ok((config, _)) => config,
+            Err(err) => {
+                if err == StorageError::ConfigNotFound {
+                    return Err(S3Error::with_message(
+                        S3ErrorCode::NoSuchCORSConfiguration,
+                        "The CORS configuration does not exist".to_string(),
+                    ));
+                }
+                warn!(
+                    component = LOG_COMPONENT_APP,
+                    subsystem = LOG_SUBSYSTEM_BUCKET,
+                    event = "bucket_cors_config_load_failed",
+                    bucket = %bucket,
+                    error = ?err,
+                    "Failed to load bucket CORS configuration"
+                );
+                return Err(ApiError::from(err).into());
+            }
+        };
+
+        Ok(cors_configuration)
+    }
+
+    #[instrument(level = "debug", skip(self, req))]
+    pub async fn execute_put_bucket_cors(
+        &self,
+        req: S3Request<PutBucketCorsInput>,
+        prepared: S3Result<PersistedCorsConfiguration>,
+    ) -> S3Result<S3Response<PutBucketCorsOutput>> {
+        let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
+        let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
+        let PutBucketCorsInput { bucket, .. } = req.input;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        let cors_configuration = prepared?;
+        let data = serialize_config(&cors_configuration)?;
+        update_bucket_config_for_incarnation(&bucket, BUCKET_CORS_CONFIG, data, expected_incarnation_id)
+            .await
+            .map_err(ApiError::from)?;
+
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket cors", request_context, false).await;
+
+        let mut item = sr_bucket_meta_item(bucket.clone(), "cors-config");
+        item.cors =
+            Some(serialize_config(&cors_configuration).and_then(|bytes| String::from_utf8(bytes).map_err(to_internal_error))?);
+        if let Err(err) = site_replication_bucket_meta_hook(item).await {
+            warn!(bucket = %bucket, error = ?err, "site replication bucket cors hook failed");
+        }
+
+        Ok(S3Response::new(PutBucketCorsOutput::default()))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::app::bucket_usecase::test_support::build_request;
+    use http::Method;
+
+    #[tokio::test]
+    async fn execute_delete_bucket_cors_returns_internal_error_when_store_uninitialized() {
+        let input = DeleteBucketCorsInput::builder()
+            .bucket("test-bucket".to_string())
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::DELETE);
+        let usecase = DefaultBucketUsecase::without_context();
+
+        let err = usecase.execute_delete_bucket_cors(req).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+    }
+
+    #[tokio::test]
+    async fn execute_get_bucket_cors_returns_internal_error_when_store_uninitialized() {
+        let input = GetBucketCorsInput::builder()
+            .bucket("test-bucket".to_string())
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::GET);
+        let usecase = DefaultBucketUsecase::without_context();
+
+        let err = usecase.execute_get_bucket_cors(req).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+    }
+
+    #[tokio::test]
+    async fn execute_put_bucket_cors_returns_internal_error_when_store_uninitialized() {
+        let input = PutBucketCorsInput::builder()
+            .bucket("test-bucket".to_string())
+            .cors_configuration(CORSConfiguration::default())
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::PUT);
+        let usecase = DefaultBucketUsecase::without_context();
+
+        let err = usecase
+            .execute_put_bucket_cors(req, Ok(PersistedCorsConfiguration::default()))
+            .await
+            .unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+    }
+}

rustfs/src/app/bucket_usecase/encryption.rs

@@ -0,0 +1,292 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Bucket default-encryption use-case operations.
+
+use super::*;
+
+/// Refuse a default-encryption configuration the write path could not honour
+/// as written. `bucket_default_write_sse` falls back to AES256 for any
+/// algorithm it does not know, so storing one would make GetBucketEncryption
+/// advertise a scheme no object is encrypted under. s3s parses `SSEAlgorithm`
+/// as an open string, so the schema check has to happen here.
+fn validate_bucket_encryption_configuration(config: &PersistedBucketEncryptionConfiguration) -> S3Result<()> {
+    if config.rules.is_empty() {
+        return Err(S3Error::with_message(
+            S3ErrorCode::MalformedXML,
+            "ServerSideEncryptionConfiguration must contain at least one Rule".to_string(),
+        ));
+    }
+    for rule in &config.rules {
+        let Some(by_default) = rule.apply_server_side_encryption_by_default.as_ref() else {
+            return Err(S3Error::with_message(
+                S3ErrorCode::MalformedXML,
+                "Rule must contain ApplyServerSideEncryptionByDefault".to_string(),
+            ));
+        };
+        let names_kms_key = by_default.kms_master_key_id.as_deref().is_some_and(|id| !id.is_empty());
+        match by_default.sse_algorithm.as_str() {
+            ServerSideEncryption::AWS_KMS => {}
+            ServerSideEncryption::AES256 if names_kms_key => {
+                return Err(S3Error::with_message(
+                    S3ErrorCode::InvalidArgument,
+                    "KMSMasterKeyID can only be specified when SSEAlgorithm is aws:kms".to_string(),
+                ));
+            }
+            ServerSideEncryption::AES256 => {}
+            other => {
+                return Err(S3Error::with_message(
+                    S3ErrorCode::MalformedXML,
+                    format!("SSEAlgorithm {other} is not supported; expected AES256 or aws:kms"),
+                ));
+            }
+        }
+    }
+    Ok(())
+}
+
+impl DefaultBucketUsecase {
+    pub async fn execute_delete_bucket_encryption(
+        &self,
+        req: S3Request<DeleteBucketEncryptionInput>,
+    ) -> S3Result<S3Response<DeleteBucketEncryptionOutput>> {
+        let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
+        let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
+        let DeleteBucketEncryptionInput { bucket, .. } = req.input;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        delete_bucket_config_for_incarnation(&bucket, BUCKET_SSECONFIG, expected_incarnation_id)
+            .await
+            .map_err(ApiError::from)?;
+
+        notify_bucket_metadata_reload(bucket.clone(), "delete bucket encryption", request_context, false).await;
+
+        let item = sr_bucket_meta_item(bucket.clone(), "sse-config");
+        if let Err(err) = site_replication_bucket_meta_hook(item).await {
+            warn!(bucket = %bucket, error = ?err, "site replication bucket encryption delete hook failed");
+        }
+
+        Ok(S3Response::with_status(DeleteBucketEncryptionOutput::default(), StatusCode::NO_CONTENT))
+    }
+
+    pub async fn execute_get_bucket_encryption(
+        &self,
+        req: S3Request<GetBucketEncryptionInput>,
+    ) -> S3Result<Option<PersistedBucketEncryptionConfiguration>> {
+        let GetBucketEncryptionInput { bucket, .. } = req.input;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        let server_side_encryption_configuration = match metadata_sys::get_sse_config(&bucket).await {
+            Ok((cfg, _)) => Some(cfg),
+            Err(err) => {
+                if err == StorageError::ConfigNotFound {
+                    return Err(s3_error!(ServerSideEncryptionConfigurationNotFoundError));
+                }
+                warn!(
+                    component = LOG_COMPONENT_APP,
+                    subsystem = LOG_SUBSYSTEM_BUCKET,
+                    event = "bucket_sse_config_load_failed",
+                    bucket = %bucket,
+                    error = ?err,
+                    "Failed to load bucket SSE configuration"
+                );
+                return Err(ApiError::from(err).into());
+            }
+        };
+
+        Ok(server_side_encryption_configuration)
+    }
+
+    pub async fn execute_put_bucket_encryption(
+        &self,
+        req: S3Request<PutBucketEncryptionInput>,
+        prepared: S3Result<PersistedBucketEncryptionConfiguration>,
+    ) -> S3Result<S3Response<PutBucketEncryptionOutput>> {
+        let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
+        let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
+        let PutBucketEncryptionInput { bucket, .. } = req.input;
+
+        let mut server_side_encryption_configuration = prepared?;
+        validate_bucket_encryption_configuration(&server_side_encryption_configuration)?;
+
+        // When SSE-KMS is set without a specific key ID, populate the default
+        // KMS key so that GetBucketEncryption responses include it. Clients like
+        // mc rely on the presence of KMSMasterKeyID to distinguish SSE-KMS from
+        // SSE-S3 in their display logic.
+        if let Some(rule) = server_side_encryption_configuration.rules.first_mut()
+            && let Some(ref mut by_default) = rule.apply_server_side_encryption_by_default
+            && by_default.sse_algorithm.as_str() == ServerSideEncryption::AWS_KMS
+            && by_default.kms_master_key_id.as_deref().is_none_or(str::is_empty)
+        {
+            let service = current_encryption_service()
+                .await
+                .ok_or_else(|| S3Error::with_message(S3ErrorCode::InternalError, "KMS service not initialized".to_string()))?;
+            let default_key = service
+                .get_default_key_id()
+                .filter(|key| !key.is_empty())
+                .ok_or_else(|| S3Error::with_message(S3ErrorCode::InternalError, "KMS default key not configured".to_string()))?;
+            by_default.kms_master_key_id = Some(default_key.clone());
+        }
+
+        info!("sse_config {:?}", &server_side_encryption_configuration);
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        let data = serialize_config(&server_side_encryption_configuration)?;
+        update_bucket_config_for_incarnation(&bucket, BUCKET_SSECONFIG, data, expected_incarnation_id)
+            .await
+            .map_err(ApiError::from)?;
+
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket encryption", request_context, false).await;
+
+        let mut item = sr_bucket_meta_item(bucket.clone(), "sse-config");
+        item.sse_config = Some(
+            serialize_config(&server_side_encryption_configuration)
+                .and_then(|bytes| String::from_utf8(bytes).map_err(to_internal_error))?,
+        );
+        if let Err(err) = site_replication_bucket_meta_hook(item).await {
+            warn!(bucket = %bucket, error = ?err, "site replication bucket encryption hook failed");
+        }
+        Ok(S3Response::new(PutBucketEncryptionOutput::default()))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::app::bucket_usecase::test_support::build_request;
+    use http::Method;
+    use rustfs_gateway_types::persistence::{
+        PersistedBucketEncryptionConfiguration as ServerSideEncryptionConfiguration,
+        PersistedBucketEncryptionRule as ServerSideEncryptionRule, PersistedEncryptionByDefault as ServerSideEncryptionByDefault,
+    };
+
+    fn sse_config(rules: Vec<ServerSideEncryptionRule>) -> ServerSideEncryptionConfiguration {
+        ServerSideEncryptionConfiguration { rules }
+    }
+
+    fn sse_rule(algorithm: &str, kms_key_id: Option<&str>) -> ServerSideEncryptionRule {
+        ServerSideEncryptionRule {
+            apply_server_side_encryption_by_default: Some(ServerSideEncryptionByDefault {
+                sse_algorithm: algorithm.to_string(),
+                kms_master_key_id: kms_key_id.map(|id| id.to_string()),
+            }),
+            blocked_encryption_types: None,
+            bucket_key_enabled: None,
+        }
+    }
+
+    /// The stored configuration must be one the write path honours as written:
+    /// only AES256 and aws:kms exist, and a key id belongs to aws:kms alone.
+
+    #[test]
+    fn put_bucket_encryption_refuses_configurations_the_write_path_cannot_honour() {
+        validate_bucket_encryption_configuration(&sse_config(vec![sse_rule("AES256", None)])).expect("AES256 is valid");
+        validate_bucket_encryption_configuration(&sse_config(vec![sse_rule("aws:kms", Some("bucket-key"))]))
+            .expect("aws:kms with a key is valid");
+        validate_bucket_encryption_configuration(&sse_config(vec![sse_rule("aws:kms", None)]))
+            .expect("aws:kms without a key is valid (the default key is filled in)");
+        validate_bucket_encryption_configuration(&sse_config(vec![sse_rule("AES256", Some(""))]))
+            .expect("an empty key id on AES256 is how some clients spell 'none'");
+
+        let unknown = validate_bucket_encryption_configuration(&sse_config(vec![sse_rule("AES128", None)]))
+            .expect_err("AES128 is not an algorithm this server encrypts with");
+        assert_eq!(*unknown.code(), S3ErrorCode::MalformedXML);
+
+        let misplaced = validate_bucket_encryption_configuration(&sse_config(vec![sse_rule("AES256", Some("bucket-key"))]))
+            .expect_err("a key id only makes sense for aws:kms");
+        assert_eq!(*misplaced.code(), S3ErrorCode::InvalidArgument);
+
+        let empty = validate_bucket_encryption_configuration(&sse_config(Vec::new())).expect_err("no rule, no default");
+        assert_eq!(*empty.code(), S3ErrorCode::MalformedXML);
+
+        let bare_rule = validate_bucket_encryption_configuration(&sse_config(vec![ServerSideEncryptionRule {
+            apply_server_side_encryption_by_default: None,
+            blocked_encryption_types: None,
+            bucket_key_enabled: None,
+        }]))
+        .expect_err("a rule without ApplyServerSideEncryptionByDefault configures nothing");
+        assert_eq!(*bare_rule.code(), S3ErrorCode::MalformedXML);
+
+        // A second rule is checked too, so a malformed one cannot hide behind a valid first rule.
+        let second_bad =
+            validate_bucket_encryption_configuration(&sse_config(vec![sse_rule("AES256", None), sse_rule("garbage", None)]))
+                .expect_err("every rule is validated");
+        assert_eq!(*second_bad.code(), S3ErrorCode::MalformedXML);
+    }
+
+    #[tokio::test]
+    async fn execute_get_bucket_encryption_returns_internal_error_when_store_uninitialized() {
+        let input = GetBucketEncryptionInput::builder()
+            .bucket("test-bucket".to_string())
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::GET);
+        let usecase = DefaultBucketUsecase::without_context();
+
+        let err = usecase.execute_get_bucket_encryption(req).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+    }
+
+    #[tokio::test]
+    async fn execute_put_bucket_encryption_returns_internal_error_when_store_uninitialized() {
+        // A well-formed rule, so the request reaches the store lookup instead
+        // of being refused by configuration validation first.
+        let input = PutBucketEncryptionInput::builder()
+            .bucket("test-bucket".to_string())
+            .server_side_encryption_configuration(s3s_crate::dto::ServerSideEncryptionConfiguration {
+                rules: vec![s3s_crate::dto::ServerSideEncryptionRule {
+                    apply_server_side_encryption_by_default: Some(s3s_crate::dto::ServerSideEncryptionByDefault {
+                        sse_algorithm: ServerSideEncryption::from_static(ServerSideEncryption::AES256),
+                        kms_master_key_id: None,
+                    }),
+                    blocked_encryption_types: None,
+                    bucket_key_enabled: None,
+                }],
+            })
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::PUT);
+        let prepared = crate::storage::persisted_from_s3s(&req.input.server_side_encryption_configuration);
+        let usecase = DefaultBucketUsecase::without_context();
+
+        let err = usecase.execute_put_bucket_encryption(req, prepared).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+    }
+}

rustfs/src/app/bucket_usecase/guards.rs

@@ -0,0 +1,120 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Source-text guards over the bucket use-case sources.
+
+/// Bucket usecase source: this file plus `bucket_usecase/`, where a split moves its methods.
+fn bucket_usecase_sources() -> crate::guard_sources::SourceSet {
+    crate::guard_sources::SourceSet::read("src/app/bucket_usecase.rs", "src/app/bucket_usecase")
+}
+
+/// Text after each exact `{marker}(` or `{marker}<` definition, in set order.
+fn method_definitions<'a>(source: &'a crate::guard_sources::SourceSet, marker: &str) -> Vec<&'a str> {
+    source
+        .files()
+        .iter()
+        .flat_map(|file| {
+            file.text
+                .match_indices(marker)
+                .map(move |(start, _)| &file.text[start + marker.len()..])
+                .filter(|rest| rest.starts_with(['(', '<']))
+        })
+        .collect()
+}
+
+/// The body of the one definition of `method` across the bucket usecase set, preferring the
+/// public form; it ends at the next method of the same indentation or at the end of its file.
+fn usecase_method_source<'a>(source: &'a crate::guard_sources::SourceSet, method: &str) -> &'a str {
+    let public = method_definitions(source, &format!("pub async fn {method}"));
+    let definitions = if public.is_empty() {
+        method_definitions(source, &format!("async fn {method}"))
+    } else {
+        public
+    };
+    assert_eq!(definitions.len(), 1, "{method} must be defined exactly once across {}", source.describe());
+    let rest = definitions[0];
+    let end = [rest.find("\n    pub async fn "), rest.find("\n    async fn ")]
+        .into_iter()
+        .flatten()
+        .min()
+        .unwrap_or(rest.len());
+    &rest[..end]
+}
+
+#[test]
+fn bucket_create_and_delete_use_admitted_fresh_tasks() {
+    let sources = bucket_usecase_sources();
+    let source = &sources;
+    for (method, operation, inner_call) in [
+        (
+            "execute_create_bucket",
+            "bucket creation",
+            "execute_create_bucket_inner(req, mutation_permit).await",
+        ),
+        ("execute_delete_bucket", "bucket deletion", "execute_delete_bucket_inner(req).await"),
+    ] {
+        let body = usecase_method_source(source, method);
+        assert!(
+            body.contains(&format!("await_bucket_usecase_on_fresh_task(\"{operation}\"")),
+            "{method} must use bounded detached-task admission"
+        );
+        assert!(body.contains(inner_call), "{method} must run its mutation inside the admitted task");
+    }
+    let create_body = usecase_method_source(source, "execute_create_bucket_inner");
+    assert!(
+        create_body.contains("notify_bucket_metadata_reload_with_context("),
+        "a successful bucket creation must refresh peer metadata caches"
+    );
+}
+
+#[test]
+fn bucket_metadata_config_changes_notify_peer_metadata_reload() {
+    let sources = bucket_usecase_sources();
+    let source = &sources;
+    for (method, operation, scanner_maintenance_change) in [
+        ("execute_delete_bucket_policy", "delete bucket policy", false),
+        ("execute_put_bucket_policy", "put bucket policy", false),
+        ("execute_delete_public_access_block", "delete public access block", false),
+        ("execute_put_public_access_block", "put public access block", false),
+        ("execute_delete_bucket_lifecycle", "delete bucket lifecycle", true),
+        ("execute_put_bucket_lifecycle_configuration", "put bucket lifecycle", true),
+        ("execute_put_bucket_versioning", "put bucket versioning", false),
+        ("execute_delete_bucket_tagging", "delete bucket tagging", false),
+        ("execute_put_bucket_tagging", "put bucket tagging", false),
+        ("execute_delete_bucket_replication", "delete bucket replication", true),
+        ("execute_put_bucket_replication", "put bucket replication", true),
+        ("execute_delete_bucket_cors", "delete bucket cors", false),
+        ("execute_put_bucket_cors", "put bucket cors", false),
+        ("execute_delete_bucket_encryption", "delete bucket encryption", false),
+        ("execute_put_bucket_encryption", "put bucket encryption", false),
+        ("execute_put_bucket_notification_configuration", "put bucket notification", false),
+    ] {
+        let body = usecase_method_source(source, method);
+        assert!(
+            body.contains("notify_bucket_metadata_reload("),
+            "{method} should notify peers to reload cached bucket metadata"
+        );
+        assert!(
+            body.contains(operation),
+            "{method} should identify the bucket metadata operation in reload logs"
+        );
+        let expected_reload = format!(
+            "notify_bucket_metadata_reload(bucket.clone(), \"{operation}\", request_context, {scanner_maintenance_change}).await;"
+        );
+        assert!(
+            body.contains(&expected_reload),
+            "{method} should propagate scanner_maintenance_change={scanner_maintenance_change}"
+        );
+    }
+}

rustfs/src/app/bucket_usecase/lifecycle.rs

@@ -0,0 +1,588 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Bucket lifecycle configuration use-case operations.
+
+use super::*;
+
+const ERR_LIFECYCLE_RULE_STATUS: &str = "Rule status must be either Enabled or Disabled";
+
+fn assign_lifecycle_rule_ids(rules: &mut [LifecycleRule]) {
+    let mut rule_ids: HashSet<String> = HashSet::new();
+
+    for rule in rules.iter() {
+        if let Some(id) = rule.id.as_ref() {
+            rule_ids.insert(id.to_string());
+        }
+    }
+
+    for (idx, rule) in rules.iter_mut().enumerate() {
+        if rule.id.is_none() {
+            let mut suffix = 0usize;
+            let mut generated_id = format!("rule-{}", idx);
+
+            while rule_ids.contains(&generated_id) {
+                suffix += 1;
+                generated_id = format!("rule-{idx}-{suffix}");
+            }
+
+            rule_ids.insert(generated_id.clone());
+            rule.id = Some(generated_id);
+        }
+    }
+}
+
+fn validate_lifecycle_rule_status(rules: &[LifecycleRule]) -> std::result::Result<(), &'static str> {
+    for rule in rules {
+        if rule.status != ExpirationStatus::from_static(ExpirationStatus::ENABLED)
+            && rule.status != ExpirationStatus::from_static(ExpirationStatus::DISABLED)
+        {
+            return Err(ERR_LIFECYCLE_RULE_STATUS);
+        }
+    }
+
+    Ok(())
+}
+
+/// Map a lifecycle validation failure onto the S3 error the client should see.
+///
+/// The validator reports a schema-shape violation (a `Filter` with more than
+/// one predicate, a one-member `And`) with
+/// [`LIFECYCLE_MALFORMED_XML_ERROR_KIND`]; AWS answers those with
+/// `MalformedXML`. Everything else is a value the schema allows but S3 refuses,
+/// which stays `InvalidArgument` — the code this path has always returned
+/// (backlog#2201).
+fn lifecycle_validation_error(err: &std::io::Error) -> S3Error {
+    if err.kind() == LIFECYCLE_MALFORMED_XML_ERROR_KIND {
+        return S3Error::with_message(S3ErrorCode::MalformedXML, format!("Malformed XML: {err}"));
+    }
+    s3_error!(InvalidArgument, "{err}")
+}
+
+fn lifecycle_has_transition_rules(config: &PersistedLifecycleConfiguration) -> bool {
+    config.rules.iter().any(|rule| {
+        rule.status == "Enabled"
+            && (rule.transitions.as_ref().is_some_and(|transitions| {
+                transitions.iter().any(|transition| {
+                    transition
+                        .storage_class
+                        .as_ref()
+                        .is_some_and(|storage_class| !storage_class.as_str().is_empty())
+                })
+            }) || rule.noncurrent_version_transitions.as_ref().is_some_and(|transitions| {
+                transitions.iter().any(|transition| {
+                    transition
+                        .storage_class
+                        .as_ref()
+                        .is_some_and(|storage_class| !storage_class.as_str().is_empty())
+                })
+            }))
+    })
+}
+
+fn lifecycle_has_expiry_rules(config: &PersistedLifecycleConfiguration) -> bool {
+    config.rules.iter().any(|rule| {
+        rule.status == "Enabled"
+            && (rule.expiration.is_some() || rule.del_marker_expiration.is_some() || rule.noncurrent_version_expiration.is_some())
+    })
+}
+
+/// Status-independent presence of the expiry subset that site replication
+/// propagates (`replicateILMExpiry`): expiration / noncurrent-version
+/// expiration only. Distinct from [`lifecycle_has_expiry_rules`], which
+/// filters on ENABLED for scanner scheduling — editing a Disabled expiry rule
+/// must still advance the replication axis. Del-marker expiration and
+/// abort-multipart are site-local and never travel.
+fn lifecycle_rules_have_expiry(config: &PersistedLifecycleConfiguration) -> bool {
+    config
+        .rules
+        .iter()
+        .any(|rule| rule.expiration.is_some() || rule.noncurrent_version_expiration.is_some())
+}
+
+fn lifecycle_has_abort_multipart_rules(config: &PersistedLifecycleConfiguration) -> bool {
+    config
+        .rules
+        .iter()
+        .any(|rule| rule.status == "Enabled" && rule.abort_incomplete_multipart_upload.is_some())
+}
+
+impl DefaultBucketUsecase {
+    #[instrument(level = "debug", skip(self, req))]
+    pub async fn execute_delete_bucket_lifecycle(
+        &self,
+        req: S3Request<DeleteBucketLifecycleInput>,
+    ) -> S3Result<S3Response<DeleteBucketLifecycleOutput>> {
+        let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
+        let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
+        let DeleteBucketLifecycleInput { bucket, .. } = req.input;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        delete_bucket_config_for_incarnation(&bucket, BUCKET_LIFECYCLE_CONFIG, expected_incarnation_id)
+            .await
+            .map_err(ApiError::from)?;
+
+        notify_bucket_metadata_reload(bucket.clone(), "delete bucket lifecycle", request_context, true).await;
+
+        let item = sr_bucket_meta_item(bucket.clone(), "lc-config");
+        if let Err(err) = site_replication_bucket_meta_hook(item).await {
+            warn!(bucket = %bucket, error = ?err, "site replication bucket lifecycle delete hook failed");
+        }
+
+        Ok(S3Response::new(DeleteBucketLifecycleOutput::default()))
+    }
+
+    #[instrument(level = "debug", skip(self, req))]
+    pub async fn execute_get_bucket_lifecycle_configuration(
+        &self,
+        req: S3Request<GetBucketLifecycleConfigurationInput>,
+    ) -> S3Result<PersistedLifecycleConfiguration> {
+        let GetBucketLifecycleConfigurationInput { bucket, .. } = req.input;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        let config = match metadata_sys::get_lifecycle_config(&bucket).await {
+            Ok((cfg, _)) => cfg,
+            Err(StorageError::ConfigNotFound) => {
+                return Err(s3_error!(NoSuchLifecycleConfiguration));
+            }
+            // An unreadable stored configuration is not an absent one.
+            Err(err) => return Err(ApiError::from(err).into()),
+        };
+
+        Ok(config)
+    }
+
+    #[instrument(level = "debug", skip(self, req))]
+    pub async fn execute_put_bucket_lifecycle_configuration(
+        &self,
+        req: S3Request<PutBucketLifecycleConfigurationInput>,
+        prepared_configuration: std::io::Result<Option<PersistedLifecycleConfiguration>>,
+    ) -> S3Result<S3Response<PutBucketLifecycleConfigurationOutput>> {
+        let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
+        let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
+        let PutBucketLifecycleConfigurationInput {
+            bucket,
+            lifecycle_configuration,
+            ..
+        } = req.input;
+
+        let Some(mut wire_cfg) = lifecycle_configuration else { return Err(s3_error!(InvalidArgument)) };
+        assign_lifecycle_rule_ids(&mut wire_cfg.rules);
+        if let Err(err) = validate_lifecycle_rule_status(&wire_cfg.rules) {
+            return Err(S3Error::with_message(S3ErrorCode::MalformedXML, format!("Malformed XML: {err}")));
+        }
+
+        let rcfg = match metadata_sys::get_object_lock_config(&bucket).await {
+            Ok((cfg, _)) => cfg,
+            Err(StorageError::ConfigNotFound) => ObjectLockConfiguration::default(),
+            Err(err) => {
+                warn!(
+                    component = LOG_COMPONENT_APP,
+                    subsystem = LOG_SUBSYSTEM_BUCKET,
+                    event = "bucket_object_lock_config_load_failed",
+                    bucket = %bucket,
+                    error = ?err,
+                    "Failed to load bucket object lock configuration"
+                );
+                return Err(ApiError::from(err).into());
+            }
+        };
+
+        // Keep conversion failures behind the request-boundary status and Object Lock checks.
+        let Some(mut input_cfg) = prepared_configuration.map_err(to_internal_error)? else {
+            return Err(rustfs_s3_types::s3_error!(InvalidArgument).into());
+        };
+        if input_cfg.rules.len() != wire_cfg.rules.len() {
+            return Err(S3Error::with_message(
+                S3ErrorCode::InternalError,
+                "lifecycle conversion changed the rule count",
+            ));
+        }
+        for (rule, wire_rule) in input_cfg.rules.iter_mut().zip(&wire_cfg.rules) {
+            rule.id.clone_from(&wire_rule.id);
+        }
+
+        if let Err(err) = validate_lifecycle_config(&input_cfg, &rcfg).await {
+            return Err(lifecycle_validation_error(&err));
+        }
+
+        if let Err(err) = validate_transition_tier(&input_cfg).await {
+            return Err(s3_error!(InvalidArgument, "{err}"));
+        }
+
+        // Stamp the expiry axis only when the expiry subset can have changed
+        // (MinIO: HasExpiry() || expiryRuleRemoved). Site-replication peers
+        // judge lc-config staleness on this axis; a transition-only edit that
+        // advanced it would let this site's stale expiry subset shadow — and
+        // roll back — a newer peer expiry edit fleet-wide.
+        let previous_expiry_updated_at = match metadata_sys::get_lifecycle_config(&bucket).await {
+            Ok((previous, _)) => {
+                if lifecycle_rules_have_expiry(&input_cfg) || lifecycle_rules_have_expiry(&previous) {
+                    Some(
+                        time::OffsetDateTime::now_utc()
+                            .format(&time::format_description::well_known::Rfc3339)
+                            .map_err(to_internal_error)?,
+                    )
+                } else {
+                    previous.expiry_updated_at
+                }
+            }
+            // No previous config (or unreadable): stamping is the
+            // conservative pre-existing behavior.
+            Err(_) => {
+                if lifecycle_rules_have_expiry(&input_cfg) {
+                    Some(
+                        time::OffsetDateTime::now_utc()
+                            .format(&time::format_description::well_known::Rfc3339)
+                            .map_err(to_internal_error)?,
+                    )
+                } else {
+                    None
+                }
+            }
+        };
+        input_cfg.expiry_updated_at = previous_expiry_updated_at;
+        let data = serialize_config(&input_cfg)?;
+        update_bucket_config_for_incarnation(&bucket, BUCKET_LIFECYCLE_CONFIG, data, expected_incarnation_id)
+            .await
+            .map_err(ApiError::from)?;
+
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket lifecycle", request_context, true).await;
+
+        let mut item = sr_bucket_meta_item(bucket.clone(), "lc-config");
+        item.expiry_lc_config =
+            Some(serialize_config(&input_cfg).and_then(|bytes| String::from_utf8(bytes).map_err(to_internal_error))?);
+        // The item travels with the expiry axis, not the wall clock: a site
+        // whose expiry knowledge is old (or absent — UNIX_EPOCH) must not
+        // out-rank newer peer expiry state at the receivers.
+        item.expiry_updated_at = input_cfg
+            .expiry_updated_at
+            .as_deref()
+            .map(|value| time::OffsetDateTime::parse(value, &time::format_description::well_known::Rfc3339))
+            .transpose()
+            .map_err(to_internal_error)?
+            .or(Some(time::OffsetDateTime::UNIX_EPOCH));
+        if let Err(err) = site_replication_bucket_meta_hook(item).await {
+            warn!(bucket = %bucket, error = ?err, "site replication bucket lifecycle hook failed");
+        }
+
+        if lifecycle_has_transition_rules(&input_cfg)
+            && let Some(store) = self.object_store()
+        {
+            let bucket_name = bucket.clone();
+            let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
+            spawn_background_with_context(request_context, async move {
+                if let Err(err) = enqueue_transition_for_existing_objects(store, &bucket_name).await {
+                    warn!(bucket = %bucket_name, error = ?err, "failed to enqueue transition for existing objects");
+                }
+            });
+        }
+
+        if lifecycle_has_expiry_rules(&input_cfg)
+            && let Some(store) = self.object_store()
+        {
+            let bucket_name = bucket.clone();
+            let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
+            spawn_background_with_context(request_context, async move {
+                if let Err(err) = enqueue_expiry_for_existing_objects(store, &bucket_name).await {
+                    warn!(bucket = %bucket_name, error = ?err, "failed to enqueue expiry for existing objects");
+                }
+            });
+        }
+
+        if lifecycle_has_abort_multipart_rules(&input_cfg)
+            && let Some(store) = self.object_store()
+        {
+            let bucket_name = bucket.clone();
+            let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
+            spawn_background_with_context(request_context, async move {
+                let deleted = run_stale_multipart_upload_cleanup_once(store).await;
+                debug!(bucket = %bucket_name, deleted, "completed lifecycle abort multipart cleanup trigger");
+            });
+        }
+
+        Ok(S3Response::new(PutBucketLifecycleConfigurationOutput::default()))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::app::bucket_usecase::test_support::build_request;
+    use http::Method;
+
+    #[test]
+    fn normalize_lifecycle_rules_generate_rule_ids_for_missing_values() {
+        let mut rules = vec![
+            LifecycleRule {
+                status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                expiration: Some(LifecycleExpiration {
+                    days: Some(30),
+                    ..Default::default()
+                }),
+                abort_incomplete_multipart_upload: None,
+                del_marker_expiration: None,
+                filter: None,
+                id: None,
+                noncurrent_version_expiration: None,
+                noncurrent_version_transitions: None,
+                prefix: None,
+                transitions: None,
+            },
+            LifecycleRule {
+                status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                expiration: Some(LifecycleExpiration {
+                    days: Some(60),
+                    ..Default::default()
+                }),
+                abort_incomplete_multipart_upload: None,
+                del_marker_expiration: None,
+                filter: None,
+                id: Some("rule-1".to_string()),
+                noncurrent_version_expiration: None,
+                noncurrent_version_transitions: None,
+                prefix: None,
+                transitions: None,
+            },
+            LifecycleRule {
+                status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                expiration: Some(LifecycleExpiration {
+                    days: Some(90),
+                    ..Default::default()
+                }),
+                abort_incomplete_multipart_upload: None,
+                del_marker_expiration: None,
+                filter: None,
+                id: None,
+                noncurrent_version_expiration: None,
+                noncurrent_version_transitions: None,
+                prefix: None,
+                transitions: None,
+            },
+        ];
+
+        assign_lifecycle_rule_ids(&mut rules);
+
+        assert_eq!(rules[0].id.as_deref(), Some("rule-0"));
+        assert_eq!(rules[1].id.as_deref(), Some("rule-1"));
+        assert_eq!(rules[2].id.as_deref(), Some("rule-2"));
+    }
+
+    #[tokio::test]
+    async fn put_bucket_lifecycle_validation_errors_keep_their_s3_code() {
+        // The PUT path answers a schema-shape violation with MalformedXML and a
+        // rejected value with InvalidArgument. Both categories are produced by
+        // the real validator here, so the mapping cannot drift from it
+        // (backlog#2201).
+        let malformed = validate_lifecycle_config(
+            &rustfs_gateway_types::persistence::PersistedLifecycleConfiguration {
+                expiry_updated_at: None,
+                rules: vec![rustfs_gateway_types::persistence::PersistedLifecycleRule {
+                    status: "Enabled".to_owned(),
+                    expiration: Some(rustfs_gateway_types::persistence::PersistedLifecycleExpiration {
+                        days: Some(1),
+                        ..Default::default()
+                    }),
+                    abort_incomplete_multipart_upload: None,
+                    del_marker_expiration: None,
+                    filter: Some(rustfs_gateway_types::persistence::PersistedLifecycleFilter {
+                        and: Some(rustfs_gateway_types::persistence::PersistedLifecycleAnd {
+                            prefix: Some("logs/".to_string()),
+                            ..Default::default()
+                        }),
+                        ..Default::default()
+                    }),
+                    id: Some("one-member-and".to_string()),
+                    noncurrent_version_expiration: None,
+                    noncurrent_version_transitions: None,
+                    prefix: None,
+                    transitions: None,
+                }],
+            },
+            &ObjectLockConfiguration::default(),
+        )
+        .await
+        .expect_err("a Filter And with one predicate is a schema violation");
+        assert_eq!(*lifecycle_validation_error(&malformed).code(), S3ErrorCode::MalformedXML);
+
+        let invalid_value = validate_lifecycle_config(
+            &rustfs_gateway_types::persistence::PersistedLifecycleConfiguration {
+                expiry_updated_at: None,
+                rules: vec![rustfs_gateway_types::persistence::PersistedLifecycleRule {
+                    status: "Enabled".to_owned(),
+                    expiration: None,
+                    abort_incomplete_multipart_upload: None,
+                    del_marker_expiration: None,
+                    filter: None,
+                    id: Some("negative-count".to_string()),
+                    noncurrent_version_expiration: Some(
+                        rustfs_gateway_types::persistence::PersistedNoncurrentVersionExpiration {
+                            noncurrent_days: Some(30),
+                            newer_noncurrent_versions: Some(-1),
+                        },
+                    ),
+                    noncurrent_version_transitions: None,
+                    prefix: None,
+                    transitions: None,
+                }],
+            },
+            &ObjectLockConfiguration::default(),
+        )
+        .await
+        .expect_err("a negative retention count is rejected");
+        assert_eq!(*lifecycle_validation_error(&invalid_value).code(), S3ErrorCode::InvalidArgument);
+    }
+
+    #[test]
+    fn validate_lifecycle_rule_status_rejects_invalid_status() {
+        let rules = vec![LifecycleRule {
+            status: ExpirationStatus::from_static("enabled"),
+            expiration: Some(LifecycleExpiration {
+                days: Some(30),
+                ..Default::default()
+            }),
+            abort_incomplete_multipart_upload: None,
+            del_marker_expiration: None,
+            filter: None,
+            id: None,
+            noncurrent_version_expiration: None,
+            noncurrent_version_transitions: None,
+            prefix: None,
+            transitions: None,
+        }];
+
+        assert_eq!(validate_lifecycle_rule_status(&rules).unwrap_err(), ERR_LIFECYCLE_RULE_STATUS);
+    }
+
+    #[test]
+    fn lifecycle_has_transition_rules_ignores_disabled_rules() {
+        let config = BucketLifecycleConfiguration {
+            expiry_updated_at: None,
+            rules: vec![LifecycleRule {
+                status: ExpirationStatus::from_static(ExpirationStatus::DISABLED),
+                expiration: None,
+                abort_incomplete_multipart_upload: None,
+                del_marker_expiration: None,
+                filter: None,
+                id: Some("disabled-transition".to_string()),
+                noncurrent_version_expiration: None,
+                noncurrent_version_transitions: None,
+                prefix: None,
+                transitions: Some(vec![Transition {
+                    days: Some(1),
+                    date: None,
+                    storage_class: Some(TransitionStorageClass::from_static("WARM")),
+                }]),
+            }],
+        };
+
+        assert!(!lifecycle_has_transition_rules(
+            &crate::storage::s3_compat::lifecycle::lifecycle_configuration_from_s3s(config).unwrap()
+        ));
+    }
+
+    #[test]
+    fn lifecycle_has_transition_rules_accepts_enabled_noncurrent_transitions() {
+        let config = BucketLifecycleConfiguration {
+            expiry_updated_at: None,
+            rules: vec![LifecycleRule {
+                status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                expiration: None,
+                abort_incomplete_multipart_upload: None,
+                del_marker_expiration: None,
+                filter: None,
+                id: Some("enabled-noncurrent-transition".to_string()),
+                noncurrent_version_expiration: None,
+                noncurrent_version_transitions: Some(vec![NoncurrentVersionTransition {
+                    noncurrent_days: Some(1),
+                    newer_noncurrent_versions: None,
+                    storage_class: Some(TransitionStorageClass::from_static("WARM")),
+                }]),
+                prefix: None,
+                transitions: None,
+            }],
+        };
+
+        assert!(lifecycle_has_transition_rules(
+            &crate::storage::s3_compat::lifecycle::lifecycle_configuration_from_s3s(config).unwrap()
+        ));
+    }
+
+    #[test]
+    fn lifecycle_has_abort_multipart_rules_accepts_enabled_abort_rule() {
+        let config = BucketLifecycleConfiguration {
+            expiry_updated_at: None,
+            rules: vec![LifecycleRule {
+                status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                expiration: None,
+                abort_incomplete_multipart_upload: Some(s3s_crate::dto::AbortIncompleteMultipartUpload {
+                    days_after_initiation: Some(0),
+                }),
+                del_marker_expiration: None,
+                filter: None,
+                id: Some("enabled-abort".to_string()),
+                noncurrent_version_expiration: None,
+                noncurrent_version_transitions: None,
+                prefix: None,
+                transitions: None,
+            }],
+        };
+
+        assert!(lifecycle_has_abort_multipart_rules(
+            &crate::storage::s3_compat::lifecycle::lifecycle_configuration_from_s3s(config.clone()).unwrap()
+        ));
+        assert!(!lifecycle_has_expiry_rules(
+            &crate::storage::s3_compat::lifecycle::lifecycle_configuration_from_s3s(config).unwrap()
+        ));
+    }
+
+    #[tokio::test]
+    async fn execute_put_bucket_lifecycle_configuration_rejects_missing_configuration() {
+        let input = PutBucketLifecycleConfigurationInput::builder()
+            .bucket("test-bucket".to_string())
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::PUT);
+        let usecase = DefaultBucketUsecase::without_context();
+
+        let prepared = req
+            .input
+            .lifecycle_configuration
+            .clone()
+            .map(crate::storage::s3_compat::lifecycle::lifecycle_configuration_from_s3s)
+            .transpose();
+        let err = usecase
+            .execute_put_bucket_lifecycle_configuration(req, prepared)
+            .await
+            .unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+    }
+}

rustfs/src/app/bucket_usecase/logging.rs

@@ -0,0 +1,15 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Bucket logging configuration use-case operations (reserved for the next batch).

rustfs/src/app/bucket_usecase/notification.rs

@@ -0,0 +1,493 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Bucket notification configuration use-case operations.
+
+use super::*;
+
+fn is_valid_notification_filter_value(value: &str) -> bool {
+    if value.len() > 1024 || value.contains('\\') {
+        return false;
+    }
+    !value.split('/').any(|segment| segment == "." || segment == "..")
+}
+
+fn invalid_filter_value_message(cfg_scope: &str, value: &str) -> String {
+    format!("invalid notification filter value (len={}) ({cfg_scope})", value.len())
+}
+
+fn invalid_filter_name_message(cfg_scope: &str, name: &str) -> String {
+    format!(
+        "invalid notification filter name (len={}) (only 'prefix'/'suffix' are supported) ({cfg_scope})",
+        name.len()
+    )
+}
+
+fn validate_notification_filter_rules(
+    filter: Option<&NotificationConfigurationFilter>,
+    cfg_kind: &str,
+    cfg_id: Option<&str>,
+) -> S3Result<()> {
+    let Some(filter) = filter else {
+        return Ok(());
+    };
+    let Some(s3key_filter) = filter.key.as_ref() else {
+        return Ok(());
+    };
+    let Some(rules) = s3key_filter.filter_rules.as_ref() else {
+        return Ok(());
+    };
+
+    let mut has_prefix = false;
+    let mut has_suffix = false;
+    let cfg_scope = cfg_id.map_or_else(|| cfg_kind.to_string(), |id| format!("{cfg_kind} id={id}"));
+
+    for rule in rules {
+        let Some(name) = rule.name.as_ref() else {
+            return Err(s3_error!(InvalidArgument, "invalid notification filter rule: missing Name ({cfg_scope})"));
+        };
+        let Some(value) = rule.value.as_ref() else {
+            return Err(s3_error!(
+                InvalidArgument,
+                "invalid notification filter rule: missing Value ({cfg_scope})"
+            ));
+        };
+
+        if !is_valid_notification_filter_value(value) {
+            return Err(s3_error!(InvalidArgument, "{}", invalid_filter_value_message(&cfg_scope, value)));
+        }
+
+        if name.as_str().eq_ignore_ascii_case("prefix") {
+            if has_prefix {
+                return Err(s3_error!(InvalidArgument, "duplicate notification filter name 'prefix' ({cfg_scope})"));
+            }
+            has_prefix = true;
+        } else if name.as_str().eq_ignore_ascii_case("suffix") {
+            if has_suffix {
+                return Err(s3_error!(InvalidArgument, "duplicate notification filter name 'suffix' ({cfg_scope})"));
+            }
+            has_suffix = true;
+        } else {
+            return Err(s3_error!(InvalidArgument, "{}", invalid_filter_name_message(&cfg_scope, name.as_str())));
+        }
+    }
+
+    Ok(())
+}
+
+fn validate_notification_configuration_filters(notification_configuration: &NotificationConfiguration) -> S3Result<()> {
+    if let Some(queue_configs) = notification_configuration.queue_configurations.as_ref() {
+        for cfg in queue_configs {
+            validate_notification_filter_rules(cfg.filter.as_ref(), "QueueConfiguration", cfg.id.as_deref())?;
+        }
+    }
+    if let Some(topic_configs) = notification_configuration.topic_configurations.as_ref() {
+        for cfg in topic_configs {
+            validate_notification_filter_rules(cfg.filter.as_ref(), "TopicConfiguration", cfg.id.as_deref())?;
+        }
+    }
+    if let Some(lambda_configs) = notification_configuration.lambda_function_configurations.as_ref() {
+        for cfg in lambda_configs {
+            validate_notification_filter_rules(cfg.filter.as_ref(), "LambdaFunctionConfiguration", cfg.id.as_deref())?;
+        }
+    }
+    Ok(())
+}
+
+fn parse_notification_target_id(arn_str: &str) -> Result<TargetID, TargetIDError> {
+    ARN::parse(arn_str)
+        .map(|arn| arn.target_id)
+        .map_err(|e| TargetIDError::InvalidFormat(e.to_string()))
+}
+
+type NotificationEventRule = (Vec<EventName>, String, String, Vec<TargetID>);
+
+/// Builds the notify runtime rules for a bucket notification configuration
+/// without touching the store or the runtime rule state.
+fn build_notification_event_rules(
+    notification_configuration: &NotificationConfiguration,
+) -> S3Result<Vec<NotificationEventRule>> {
+    let mut event_rules = Vec::new();
+    let invalid_arn = |e: TargetIDError| {
+        S3Error::with_message(S3ErrorCode::InvalidArgument, format!("Invalid ARN in notification configuration: {e}"))
+    };
+
+    process_queue_configurations(
+        &mut event_rules,
+        notification_configuration.queue_configurations.clone(),
+        parse_notification_target_id,
+    )
+    .map_err(invalid_arn)?;
+    process_topic_configurations(
+        &mut event_rules,
+        notification_configuration.topic_configurations.clone(),
+        parse_notification_target_id,
+    )
+    .map_err(invalid_arn)?;
+    process_lambda_configurations(
+        &mut event_rules,
+        notification_configuration.lambda_function_configurations.clone(),
+        parse_notification_target_id,
+    )
+    .map_err(invalid_arn)?;
+
+    Ok(event_rules)
+}
+
+fn resolve_notification_region(global_region: Option<Region>, request_region: Option<Region>) -> String {
+    global_region
+        .or(request_region)
+        .map(|region| region.to_string())
+        .unwrap_or_else(|| RUSTFS_REGION.to_string())
+}
+
+impl DefaultBucketUsecase {
+    pub async fn execute_get_bucket_notification_configuration(
+        &self,
+        req: S3Request<GetBucketNotificationConfigurationInput>,
+    ) -> S3Result<Option<PersistedNotificationConfiguration>> {
+        let GetBucketNotificationConfigurationInput { bucket, .. } = req.input;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        let has_notification_config = match metadata_sys::get_notification_config(&bucket).await {
+            Ok(config) => config,
+            // An unreadable config is not "no notifications configured".
+            Err(err) if crate::storage_api::error::is_unreadable_config_error(&err) => {
+                return Err(ApiError::from(err).into());
+            }
+            Err(err) => {
+                warn!(
+                    component = LOG_COMPONENT_APP,
+                    subsystem = LOG_SUBSYSTEM_BUCKET,
+                    event = "bucket_notification_config_load_failed",
+                    bucket = %bucket,
+                    error = ?err,
+                    "Failed to load bucket notification configuration"
+                );
+                None
+            }
+        };
+
+        Ok(has_notification_config)
+    }
+
+    pub async fn execute_put_bucket_notification_configuration(
+        &self,
+        req: S3Request<PutBucketNotificationConfigurationInput>,
+        prepared: S3Result<PersistedNotificationConfiguration>,
+    ) -> S3Result<S3Response<PutBucketNotificationConfigurationOutput>> {
+        let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
+        let request_region = req.region.clone();
+        let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
+
+        let PutBucketNotificationConfigurationInput {
+            bucket,
+            notification_configuration,
+            ..
+        } = req.input;
+
+        validate_notification_configuration_filters(&notification_configuration)?;
+        let persisted_notification_configuration = prepared?;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        let region = resolve_notification_region(self.global_region(), request_region);
+        let notify = current_notify_interface_for_context(self.context.as_deref());
+        let event_rules = build_notification_event_rules(&notification_configuration)?;
+
+        // Reject the request before the store write so a failure cannot leave a
+        // persisted configuration that the notify runtime refused to activate.
+        notify
+            .validate_event_specific_rules(&bucket, region.as_str(), &event_rules)
+            .await
+            .map_err(|e| s3_error!(InternalError, "Failed to add rules: {e}"))?;
+
+        let data = serialize_config(&persisted_notification_configuration)?;
+        update_bucket_config_for_incarnation(&bucket, BUCKET_NOTIFICATION_CONFIG, data, expected_incarnation_id)
+            .await
+            .map_err(ApiError::from)?;
+
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket notification", request_context, false).await;
+
+        notify
+            .clear_bucket_notification_rules(&bucket)
+            .await
+            .map_err(|e| s3_error!(InternalError, "Failed to clear rules: {e}"))?;
+        warn!("notify event rules: {:?}", &event_rules);
+        notify
+            .add_event_specific_rules(&bucket, region.as_str(), &event_rules)
+            .await
+            .map_err(|e| s3_error!(InternalError, "Failed to add rules: {e}"))?;
+
+        Ok(S3Response::new(PutBucketNotificationConfigurationOutput {}))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::app::bucket_usecase::test_support::build_request;
+    use http::Method;
+
+    #[test]
+    fn resolve_notification_region_prefers_global_region() {
+        let binding = resolve_notification_region(Some("us-east-1".parse().unwrap()), Some("ap-southeast-1".parse().unwrap()));
+        assert_eq!(binding, "us-east-1");
+    }
+
+    #[test]
+    fn resolve_notification_region_falls_back_to_request_region() {
+        let binding = resolve_notification_region(None, Some("ap-southeast-1".parse().unwrap()));
+        assert_eq!(binding, "ap-southeast-1");
+    }
+
+    #[test]
+    fn resolve_notification_region_defaults_value() {
+        let binding = resolve_notification_region(None, None);
+        assert_eq!(binding, RUSTFS_REGION);
+    }
+
+    #[test]
+    fn validate_notification_configuration_filters_rejects_invalid_filter_name() {
+        let raw_name = "unsupported".repeat(100);
+        let cfg = NotificationConfiguration {
+            queue_configurations: Some(vec![QueueConfiguration {
+                id: Some("q1".to_string()),
+                queue_arn: "arn:rustfs:sqs:us-east-1:1:webhook".to_string(),
+                events: vec!["s3:ObjectCreated:*".to_string().into()],
+                filter: Some(NotificationConfigurationFilter {
+                    key: Some(S3KeyFilter {
+                        filter_rules: Some(vec![FilterRule {
+                            name: Some(FilterRuleName::from(raw_name.clone())),
+                            value: Some("uploads/".to_string()),
+                        }]),
+                    }),
+                }),
+            }]),
+            ..Default::default()
+        };
+
+        let err = validate_notification_configuration_filters(&cfg).unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+        let msg = err.message().unwrap_or_default();
+        assert!(msg.contains("len="), "error message should include summarized length");
+        assert!(!msg.contains(&raw_name), "error message should not echo full raw filter name");
+    }
+
+    #[test]
+    fn validate_notification_configuration_filters_accepts_case_insensitive_filter_names() {
+        let cfg = NotificationConfiguration {
+            queue_configurations: Some(vec![QueueConfiguration {
+                id: Some("q1".to_string()),
+                queue_arn: "arn:rustfs:sqs:us-east-1:1:webhook".to_string(),
+                events: vec!["s3:ObjectCreated:*".to_string().into()],
+                filter: Some(NotificationConfigurationFilter {
+                    key: Some(S3KeyFilter {
+                        filter_rules: Some(vec![
+                            FilterRule {
+                                name: Some(FilterRuleName::from("Prefix".to_string())),
+                                value: Some("uploads/".to_string()),
+                            },
+                            FilterRule {
+                                name: Some(FilterRuleName::from("Suffix".to_string())),
+                                value: Some(".csv".to_string()),
+                            },
+                        ]),
+                    }),
+                }),
+            }]),
+            ..Default::default()
+        };
+
+        validate_notification_configuration_filters(&cfg).expect("capitalized filter names should be accepted");
+    }
+
+    #[test]
+    fn validate_notification_configuration_filters_rejects_duplicate_prefix_rules() {
+        let cfg = NotificationConfiguration {
+            queue_configurations: Some(vec![QueueConfiguration {
+                id: Some("q1".to_string()),
+                queue_arn: "arn:rustfs:sqs:us-east-1:1:webhook".to_string(),
+                events: vec!["s3:ObjectCreated:*".to_string().into()],
+                filter: Some(NotificationConfigurationFilter {
+                    key: Some(S3KeyFilter {
+                        filter_rules: Some(vec![
+                            FilterRule {
+                                name: Some(FilterRuleName::from_static(FilterRuleName::PREFIX)),
+                                value: Some("uploads/".to_string()),
+                            },
+                            FilterRule {
+                                name: Some(FilterRuleName::from_static(FilterRuleName::PREFIX)),
+                                value: Some("images/".to_string()),
+                            },
+                        ]),
+                    }),
+                }),
+            }]),
+            ..Default::default()
+        };
+
+        let err = validate_notification_configuration_filters(&cfg).unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+    }
+
+    #[test]
+    fn validate_notification_configuration_filters_rejects_invalid_filter_value() {
+        let cfg = NotificationConfiguration {
+            queue_configurations: Some(vec![QueueConfiguration {
+                id: Some("q1".to_string()),
+                queue_arn: "arn:rustfs:sqs:us-east-1:1:webhook".to_string(),
+                events: vec!["s3:ObjectCreated:*".to_string().into()],
+                filter: Some(NotificationConfigurationFilter {
+                    key: Some(S3KeyFilter {
+                        filter_rules: Some(vec![FilterRule {
+                            name: Some(FilterRuleName::from_static(FilterRuleName::SUFFIX)),
+                            value: Some("../secret".to_string()),
+                        }]),
+                    }),
+                }),
+            }]),
+            ..Default::default()
+        };
+
+        let err = validate_notification_configuration_filters(&cfg).unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+        let msg = err.message().unwrap_or_default();
+        assert!(msg.contains("len="), "error message should include summarized length");
+        assert!(!msg.contains("../secret"), "error message should not echo full raw filter value");
+    }
+
+    #[test]
+    fn validate_notification_configuration_filters_rejects_missing_filter_name() {
+        let cfg = NotificationConfiguration {
+            queue_configurations: Some(vec![QueueConfiguration {
+                id: Some("q1".to_string()),
+                queue_arn: "arn:rustfs:sqs:us-east-1:1:webhook".to_string(),
+                events: vec!["s3:ObjectCreated:*".to_string().into()],
+                filter: Some(NotificationConfigurationFilter {
+                    key: Some(S3KeyFilter {
+                        filter_rules: Some(vec![FilterRule {
+                            name: None,
+                            value: Some("uploads/".to_string()),
+                        }]),
+                    }),
+                }),
+            }]),
+            ..Default::default()
+        };
+
+        let err = validate_notification_configuration_filters(&cfg).unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+    }
+
+    #[test]
+    fn validate_notification_configuration_filters_rejects_missing_filter_value() {
+        let cfg = NotificationConfiguration {
+            queue_configurations: Some(vec![QueueConfiguration {
+                id: Some("q1".to_string()),
+                queue_arn: "arn:rustfs:sqs:us-east-1:1:webhook".to_string(),
+                events: vec!["s3:ObjectCreated:*".to_string().into()],
+                filter: Some(NotificationConfigurationFilter {
+                    key: Some(S3KeyFilter {
+                        filter_rules: Some(vec![FilterRule {
+                            name: Some(FilterRuleName::from_static(FilterRuleName::PREFIX)),
+                            value: None,
+                        }]),
+                    }),
+                }),
+            }]),
+            ..Default::default()
+        };
+
+        let err = validate_notification_configuration_filters(&cfg).unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+    }
+
+    #[test]
+    fn validate_notification_configuration_filters_rejects_duplicate_suffix_rules() {
+        let cfg = NotificationConfiguration {
+            queue_configurations: Some(vec![QueueConfiguration {
+                id: Some("q1".to_string()),
+                queue_arn: "arn:rustfs:sqs:us-east-1:1:webhook".to_string(),
+                events: vec!["s3:ObjectCreated:*".to_string().into()],
+                filter: Some(NotificationConfigurationFilter {
+                    key: Some(S3KeyFilter {
+                        filter_rules: Some(vec![
+                            FilterRule {
+                                name: Some(FilterRuleName::from_static(FilterRuleName::SUFFIX)),
+                                value: Some(".csv".to_string()),
+                            },
+                            FilterRule {
+                                name: Some(FilterRuleName::from_static(FilterRuleName::SUFFIX)),
+                                value: Some(".log".to_string()),
+                            },
+                        ]),
+                    }),
+                }),
+            }]),
+            ..Default::default()
+        };
+
+        let err = validate_notification_configuration_filters(&cfg).unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+    }
+
+    #[tokio::test]
+    async fn execute_put_bucket_notification_configuration_rejects_invalid_filter_before_store_lookup() {
+        let input = PutBucketNotificationConfigurationInput::builder()
+            .bucket("test-bucket".to_string())
+            .notification_configuration(NotificationConfiguration {
+                queue_configurations: Some(vec![QueueConfiguration {
+                    id: Some("q1".to_string()),
+                    queue_arn: "arn:rustfs:sqs:us-east-1:1:webhook".to_string(),
+                    events: vec!["s3:ObjectCreated:*".to_string().into()],
+                    filter: Some(NotificationConfigurationFilter {
+                        key: Some(S3KeyFilter {
+                            filter_rules: Some(vec![FilterRule {
+                                name: Some(FilterRuleName::from("unsupported".to_string())),
+                                value: Some("uploads/".to_string()),
+                            }]),
+                        }),
+                    }),
+                }]),
+                ..Default::default()
+            })
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::PUT);
+        let usecase = DefaultBucketUsecase::without_context();
+
+        let err = usecase
+            .execute_put_bucket_notification_configuration(req, Ok(PersistedNotificationConfiguration::default()))
+            .await
+            .unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+    }
+}

rustfs/src/app/bucket_usecase/object_lock.rs

@@ -0,0 +1,15 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Bucket object-lock configuration use-case operations (reserved for the next batch).

rustfs/src/app/bucket_usecase/policy.rs

@@ -0,0 +1,294 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Bucket policy and policy status use-case operations.
+
+use super::*;
+use rustfs_policy::policy::Validator;
+
+impl DefaultBucketUsecase {
+    pub async fn execute_delete_bucket_policy(
+        &self,
+        req: S3Request<DeleteBucketPolicyInput>,
+    ) -> S3Result<S3Response<DeleteBucketPolicyOutput>> {
+        record_s3_op(S3Operation::DeleteBucketPolicy);
+        let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
+        let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
+        let DeleteBucketPolicyInput { bucket, .. } = req.input;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        delete_bucket_config_for_incarnation(&bucket, BUCKET_POLICY_CONFIG, expected_incarnation_id)
+            .await
+            .map_err(ApiError::from)?;
+
+        notify_bucket_metadata_reload(bucket.clone(), "delete bucket policy", request_context, false).await;
+
+        let item = sr_bucket_meta_item(bucket.clone(), "policy");
+        if let Err(err) = site_replication_bucket_meta_hook(item).await {
+            warn!(bucket = %bucket, error = ?err, "site replication bucket policy delete hook failed");
+        }
+
+        Ok(S3Response::new(DeleteBucketPolicyOutput {}))
+    }
+
+    pub async fn execute_get_bucket_policy(
+        &self,
+        req: S3Request<GetBucketPolicyInput>,
+    ) -> S3Result<S3Response<GetBucketPolicyOutput>> {
+        let GetBucketPolicyInput { bucket, .. } = req.input;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        let (policy_str, _) = match metadata_sys::get_bucket_policy_raw(&bucket).await {
+            Ok(res) => res,
+            Err(err) => {
+                if StorageError::ConfigNotFound == err {
+                    return Err(s3_error!(NoSuchBucketPolicy));
+                }
+                return Err(S3Error::with_message(S3ErrorCode::InternalError, err.to_string()));
+            }
+        };
+
+        Ok(S3Response::new(GetBucketPolicyOutput {
+            policy: Some(policy_str),
+        }))
+    }
+
+    pub async fn execute_get_bucket_policy_status(
+        &self,
+        req: S3Request<GetBucketPolicyStatusInput>,
+    ) -> S3Result<S3Response<GetBucketPolicyStatusOutput>> {
+        let GetBucketPolicyStatusInput { bucket, .. } = req.input;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
+        let client_info = req.extensions.get::<ClientInfo>();
+        let conditions = get_condition_values_with_client_info(
+            &req.headers,
+            &rustfs_credentials::Credentials::default(),
+            None,
+            None,
+            remote_addr,
+            client_info,
+        );
+
+        let read_allowed = PolicySys::try_is_allowed(&BucketPolicyArgs {
+            bucket: &bucket,
+            action: Action::S3Action(S3Action::ListBucketAction),
+            is_owner: false,
+            account: "",
+            groups: &None,
+            conditions: &conditions,
+            object: "",
+        })
+        .await
+        .map_err(ApiError::from)?;
+
+        let write_allowed = PolicySys::try_is_allowed(&BucketPolicyArgs {
+            bucket: &bucket,
+            action: Action::S3Action(S3Action::PutObjectAction),
+            is_owner: false,
+            account: "",
+            groups: &None,
+            conditions: &conditions,
+            object: "",
+        })
+        .await
+        .map_err(ApiError::from)?;
+
+        let mut is_public = read_allowed || write_allowed;
+        let ignore_public_acls = match metadata_sys::get_public_access_block_config(&bucket).await {
+            Ok((config, _)) => config.ignore_public_acls.unwrap_or(false),
+            Err(_) => false,
+        };
+
+        let _ = ignore_public_acls;
+
+        let policy_public = match metadata_sys::get_bucket_policy(&bucket).await {
+            Ok((cfg, _)) => cfg.statements.iter().any(|statement| {
+                matches!(statement.effect, Effect::Allow)
+                    && statement.principal.is_match("*")
+                    && statement.conditions.is_empty()
+                    && statement.actions.is_match(&Action::S3Action(S3Action::ListBucketAction))
+            }),
+            Err(err) => {
+                if err == StorageError::ConfigNotFound {
+                    false
+                } else {
+                    return Err(ApiError::from(err).into());
+                }
+            }
+        };
+
+        if policy_public {
+            is_public = true;
+        }
+
+        let output = GetBucketPolicyStatusOutput {
+            policy_status: Some(PolicyStatus {
+                is_public: Some(is_public),
+            }),
+        };
+
+        Ok(S3Response::new(output))
+    }
+
+    pub async fn execute_put_bucket_policy(
+        &self,
+        req: S3Request<PutBucketPolicyInput>,
+    ) -> S3Result<S3Response<PutBucketPolicyOutput>> {
+        record_s3_op(S3Operation::PutBucketPolicy);
+        let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
+        let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
+        let PutBucketPolicyInput { bucket, policy, .. } = req.input;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        let cfg: BucketPolicy =
+            serde_json::from_str(&policy).map_err(|e| s3_error!(InvalidArgument, "parse policy failed {:?}", e))?;
+
+        if let Err(err) = cfg.is_valid() {
+            warn!("put_bucket_policy err input {:?}, {:?}", &policy, err);
+            return Err(s3_error!(MalformedPolicy));
+        }
+
+        let is_public_policy = cfg
+            .statements
+            .iter()
+            .any(|statement| matches!(statement.effect, Effect::Allow) && statement.principal.is_match("*"));
+
+        if is_public_policy {
+            match metadata_sys::get_public_access_block_config(&bucket).await {
+                Ok((config, _)) => {
+                    if config.block_public_policy.unwrap_or(false) {
+                        return Err(s3_error!(AccessDenied, "Access Denied"));
+                    }
+                }
+                Err(err) => {
+                    if err != StorageError::ConfigNotFound {
+                        warn!(
+                            component = LOG_COMPONENT_APP,
+                            subsystem = LOG_SUBSYSTEM_BUCKET,
+                            event = "bucket_public_access_block_load_failed",
+                            bucket = %bucket,
+                            error = ?err,
+                            "Failed to load bucket public access block configuration"
+                        );
+                        return Err(ApiError::from(err).into());
+                    }
+                }
+            }
+        }
+
+        let data = policy.as_bytes().to_vec();
+
+        update_bucket_config_for_incarnation(&bucket, BUCKET_POLICY_CONFIG, data, expected_incarnation_id)
+            .await
+            .map_err(ApiError::from)?;
+
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket policy", request_context, false).await;
+
+        let mut item = sr_bucket_meta_item(bucket.clone(), "policy");
+        item.policy = Some(serde_json::from_str(&policy).map_err(|e| s3_error!(InvalidArgument, "parse policy failed {:?}", e))?);
+        if let Err(err) = site_replication_bucket_meta_hook(item).await {
+            warn!(bucket = %bucket, error = ?err, "site replication bucket policy hook failed");
+        }
+
+        Ok(S3Response::new(PutBucketPolicyOutput {}))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::app::bucket_usecase::test_support::{build_request, s3_op_total};
+    use http::Method;
+
+    #[tokio::test]
+    async fn execute_delete_bucket_policy_records_s3_operation_before_store_lookup() {
+        let input = DeleteBucketPolicyInput::builder()
+            .bucket("test-bucket".to_string())
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::DELETE);
+        let usecase = DefaultBucketUsecase::without_context();
+        let before = s3_op_total(S3Operation::DeleteBucketPolicy);
+
+        let err = usecase.execute_delete_bucket_policy(req).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+        assert_eq!(s3_op_total(S3Operation::DeleteBucketPolicy), before + 1);
+    }
+
+    #[tokio::test]
+    async fn execute_get_bucket_policy_returns_internal_error_when_store_uninitialized() {
+        let input = GetBucketPolicyInput::builder()
+            .bucket("test-bucket".to_string())
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::GET);
+        let usecase = DefaultBucketUsecase::without_context();
+
+        let err = usecase.execute_get_bucket_policy(req).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+    }
+
+    #[tokio::test]
+    async fn execute_put_bucket_policy_returns_internal_error_when_store_uninitialized() {
+        let input = PutBucketPolicyInput::builder()
+            .bucket("test-bucket".to_string())
+            .policy("{}".to_string())
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::PUT);
+        let usecase = DefaultBucketUsecase::without_context();
+        let before = s3_op_total(S3Operation::PutBucketPolicy);
+
+        let err = usecase.execute_put_bucket_policy(req).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+        assert_eq!(s3_op_total(S3Operation::PutBucketPolicy), before + 1);
+    }
+}

rustfs/src/app/bucket_usecase/public_access_block.rs

@@ -0,0 +1,161 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Bucket public access block use-case operations.
+
+use super::*;
+
+impl DefaultBucketUsecase {
+    #[instrument(level = "debug", skip(self, req))]
+    pub async fn execute_delete_public_access_block(
+        &self,
+        req: S3Request<DeletePublicAccessBlockInput>,
+    ) -> S3Result<S3Response<DeletePublicAccessBlockOutput>> {
+        let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
+        let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
+        let DeletePublicAccessBlockInput { bucket, .. } = req.input;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        delete_bucket_config_for_incarnation(&bucket, BUCKET_PUBLIC_ACCESS_BLOCK_CONFIG, expected_incarnation_id)
+            .await
+            .map_err(ApiError::from)?;
+
+        notify_bucket_metadata_reload(bucket.clone(), "delete public access block", request_context, false).await;
+
+        Ok(S3Response::with_status(DeletePublicAccessBlockOutput::default(), StatusCode::NO_CONTENT))
+    }
+
+    #[instrument(level = "debug", skip(self, req))]
+    pub async fn execute_get_public_access_block(
+        &self,
+        req: S3Request<GetPublicAccessBlockInput>,
+    ) -> S3Result<PersistedPublicAccessBlockConfiguration> {
+        let GetPublicAccessBlockInput { bucket, .. } = req.input;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        let config = match metadata_sys::get_public_access_block_config(&bucket).await {
+            Ok((config, _)) => config,
+            Err(err) => {
+                if err == StorageError::ConfigNotFound {
+                    return Err(S3Error::with_message(
+                        S3ErrorCode::Custom("NoSuchPublicAccessBlockConfiguration".into()),
+                        "Public access block configuration does not exist".to_string(),
+                    ));
+                }
+                return Err(ApiError::from(err).into());
+            }
+        };
+
+        Ok(config)
+    }
+
+    #[instrument(level = "debug", skip(self, req))]
+    pub async fn execute_put_public_access_block(
+        &self,
+        req: S3Request<PutPublicAccessBlockInput>,
+        prepared: S3Result<PersistedPublicAccessBlockConfiguration>,
+    ) -> S3Result<S3Response<PutPublicAccessBlockOutput>> {
+        let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
+        let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
+        let PutPublicAccessBlockInput { bucket, .. } = req.input;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        let public_access_block_configuration = prepared?;
+        let data = serialize_config(&public_access_block_configuration)?;
+        update_bucket_config_for_incarnation(&bucket, BUCKET_PUBLIC_ACCESS_BLOCK_CONFIG, data, expected_incarnation_id)
+            .await
+            .map_err(ApiError::from)?;
+
+        notify_bucket_metadata_reload(bucket.clone(), "put public access block", request_context, false).await;
+
+        Ok(S3Response::new(PutPublicAccessBlockOutput::default()))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::app::bucket_usecase::test_support::build_request;
+    use http::Method;
+
+    #[tokio::test]
+    async fn execute_delete_public_access_block_returns_internal_error_when_store_uninitialized() {
+        let input = DeletePublicAccessBlockInput::builder()
+            .bucket("test-bucket".to_string())
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::DELETE);
+        let usecase = DefaultBucketUsecase::without_context();
+
+        let err = usecase.execute_delete_public_access_block(req).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+    }
+
+    #[tokio::test]
+    async fn execute_get_public_access_block_returns_internal_error_when_store_uninitialized() {
+        let input = GetPublicAccessBlockInput::builder()
+            .bucket("test-bucket".to_string())
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::GET);
+        let usecase = DefaultBucketUsecase::without_context();
+
+        let err = usecase.execute_get_public_access_block(req).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+    }
+
+    #[tokio::test]
+    async fn execute_put_public_access_block_returns_internal_error_when_store_uninitialized() {
+        let input = PutPublicAccessBlockInput::builder()
+            .bucket("test-bucket".to_string())
+            .public_access_block_configuration(PublicAccessBlockConfiguration::default())
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::PUT);
+        let usecase = DefaultBucketUsecase::without_context();
+
+        let err = usecase
+            .execute_put_public_access_block(req, Ok(PersistedPublicAccessBlockConfiguration::default()))
+            .await
+            .unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+    }
+}

rustfs/src/app/bucket_usecase/replication.rs

@@ -0,0 +1,912 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Bucket replication configuration use-case operations.
+
+use super::*;
+
+fn validate_replication_config_targets(targets: &BucketTargets, config: &PersistedReplicationConfiguration) -> S3Result<()> {
+    let configured_arns = targets
+        .targets
+        .iter()
+        .filter(|target| target.target_type == BucketTargetType::ReplicationService)
+        .map(|target| target.arn.as_str());
+
+    match validate_replication_config_target_arns(configured_arns, config) {
+        Ok(()) => Ok(()),
+        Err(err) => {
+            let message = match err {
+                ReplicationTargetValidationError::RoleWithMultipleDestinations => {
+                    "replication config with Role cannot define multiple destination targets"
+                }
+                ReplicationTargetValidationError::StaleTarget => "replication config has a stale target",
+            };
+            Err(S3Error::with_message(S3ErrorCode::InvalidRequest, message))
+        }
+    }
+}
+
+fn validate_replication_config_capabilities(config: &PersistedReplicationConfiguration) -> S3Result<()> {
+    if let Err(err) = validate_replication_config_structure(config) {
+        return Err(S3Error::with_message(S3ErrorCode::InvalidRequest, err.message()));
+    }
+    if let Some(field) = invalid_replication_config_status_field(config) {
+        return Err(S3Error::with_message(
+            S3ErrorCode::InvalidRequest,
+            format!("replication field {field} has an invalid status"),
+        ));
+    }
+    if let Some(field) = unsupported_replication_config_field(config) {
+        return Err(S3Error::with_message(
+            S3ErrorCode::InvalidRequest,
+            format!("replication field {field} is not supported by this RustFS version"),
+        ));
+    }
+    Ok(())
+}
+
+async fn validate_bucket_replication_update(bucket: &str, config: &PersistedReplicationConfiguration) -> S3Result<()> {
+    if !BucketVersioningSys::enabled(bucket).await {
+        return Err(s3_error!(
+            InvalidRequest,
+            "bucket versioning must be enabled before replication can be configured"
+        ));
+    }
+
+    let targets = metadata_sys::get_bucket_targets_config(bucket)
+        .await
+        .map_err(|err| match err {
+            StorageError::ConfigNotFound => {
+                S3Error::with_message(S3ErrorCode::InvalidRequest, "replication target configuration not found".to_string())
+            }
+            other => ApiError::from(other).into(),
+        })?;
+
+    validate_replication_config_targets(&targets, config)
+}
+
+/// Defense in depth for site-replication-managed buckets (issue #1948): an S3
+/// PutBucketReplication replaces the operator-authored rules but must not wipe
+/// the rules the reconciler derived for the current remote peers
+/// (`site_peer_deployment_ids`) — until its next pass (600s period) every
+/// peer link on this bucket would be silently dead. The same merge also drops
+/// incoming impostors of those rules. An empty peer set (site replication
+/// disabled) keeps the verbatim overwrite semantics: rule ids are not
+/// reserved, so an operator's own `site-repl-*` rule is ordinary state there.
+/// `contract` is what the peers were probed to support; the merged config is
+/// what gets broadcast, so it is built the way every peer will merge it.
+fn merge_user_replication_config_update(
+    incoming: PersistedReplicationConfiguration,
+    existing: Option<PersistedReplicationConfiguration>,
+    site_peer_deployment_ids: &HashSet<String>,
+    contract: OperatorRuleContract,
+) -> PersistedReplicationConfiguration {
+    if site_peer_deployment_ids.is_empty() {
+        return incoming;
+    }
+    // `incoming` passed structure validation, so it holds at least one rule;
+    // `None` is only reachable when every incoming rule impersonates a
+    // reconciler rule, and then the stored reconciler rules are what remains.
+    merge_user_replication_config(Some(incoming.clone()), existing, site_peer_deployment_ids, contract).unwrap_or(incoming)
+}
+
+/// Split of an S3 DeleteBucketReplication on the stored config (issue #1948):
+/// the operator-authored rules are removed, the rules the reconciler derived
+/// for the current remote peers survive (`None` means nothing survives and
+/// the config is deleted), and the returned ARNs are the ones whose bucket
+/// targets may be garbage-collected — never an ARN a surviving reconciler
+/// rule still points at.
+fn split_replication_config_for_user_delete(
+    config: PersistedReplicationConfiguration,
+    site_peer_deployment_ids: &HashSet<String>,
+    contract: OperatorRuleContract,
+) -> (Option<PersistedReplicationConfiguration>, HashSet<String>) {
+    let mut removable_arns = replication_target_arns(&config);
+    let remaining = merge_user_replication_config(None, Some(config), site_peer_deployment_ids, contract);
+    if let Some(remaining) = remaining.as_ref() {
+        for rule in &remaining.rules {
+            removable_arns.remove(rule.destination.bucket.trim());
+        }
+    }
+    (remaining, removable_arns)
+}
+
+async fn replication_targets_without_arns(
+    bucket: &str,
+    target_arns: &HashSet<String>,
+) -> S3Result<Option<(BucketTargets, usize)>> {
+    if target_arns.is_empty() {
+        return Ok(None);
+    }
+
+    let mut targets = match metadata_sys::get_bucket_targets_config(bucket).await {
+        Ok(targets) => targets,
+        Err(StorageError::ConfigNotFound) => return Ok(None),
+        Err(err) => return Err(ApiError::from(err).into()),
+    };
+
+    let removed = remove_replication_targets_from_config_targets(&mut targets, target_arns);
+    if removed == 0 {
+        return Ok(None);
+    }
+
+    Ok(Some((targets, removed)))
+}
+
+fn remove_replication_targets_from_config_targets(targets: &mut BucketTargets, target_arns: &HashSet<String>) -> usize {
+    let original_len = targets.targets.len();
+    targets.targets.retain(|target| {
+        !should_remove_replication_target(
+            target.arn.as_str(),
+            target.target_type == BucketTargetType::ReplicationService,
+            target_arns,
+        )
+    });
+
+    original_len - targets.targets.len()
+}
+
+async fn write_replication_targets_after_config_delete(
+    bucket: &str,
+    targets: &BucketTargets,
+    removed: usize,
+    expected_incarnation_id: Option<uuid::Uuid>,
+) -> S3Result<()> {
+    let json_targets = serde_json::to_vec(&targets).map_err(to_internal_error)?;
+    update_bucket_config_for_incarnation(bucket, BUCKET_TARGETS_FILE, json_targets, expected_incarnation_id)
+        .await
+        .map_err(ApiError::from)?;
+    info!(bucket = %bucket, removed, "removed replication remote targets referenced by deleted bucket replication config");
+
+    Ok(())
+}
+
+async fn restore_replication_config_after_target_cleanup_failure(
+    bucket: &str,
+    config: &PersistedReplicationConfiguration,
+    cleanup_err: S3Error,
+    expected_incarnation_id: Option<uuid::Uuid>,
+) -> S3Error {
+    match serialize(config) {
+        Ok(data) => {
+            if let Err(restore_err) =
+                update_bucket_config_for_incarnation(bucket, BUCKET_REPLICATION_CONFIG, data, expected_incarnation_id).await
+            {
+                error!(
+                    bucket = %bucket,
+                    error = ?restore_err,
+                    cleanup_error = ?cleanup_err,
+                    "failed to restore bucket replication config after target cleanup failure"
+                );
+            }
+        }
+        Err(restore_err) => {
+            error!(
+                bucket = %bucket,
+                error = ?restore_err,
+                cleanup_error = ?cleanup_err,
+                "failed to serialize bucket replication config for restore after target cleanup failure"
+            );
+        }
+    }
+
+    cleanup_err
+}
+
+impl DefaultBucketUsecase {
+    /// `site_peers` is the set of remote site-replication peer deployment ids
+    /// (empty when site replication is disabled). The interface layer reads it
+    /// from the persisted state and fails closed on a read error, so this
+    /// usecase stays a pure function of its inputs (layer rule: app never
+    /// imports interface).
+    pub async fn execute_delete_bucket_replication(
+        &self,
+        req: S3Request<DeleteBucketReplicationInput>,
+        site_peers: HashSet<String>,
+        contract: OperatorRuleContract,
+    ) -> S3Result<S3Response<DeleteBucketReplicationOutput>> {
+        let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
+        let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
+        let DeleteBucketReplicationInput { bucket, .. } = req.input;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+        let targets_guard = lock_bucket_targets_metadata(&bucket).await;
+        let replication_config = match metadata_sys::get_replication_config(&bucket).await {
+            Ok((config, _)) => Some(config),
+            Err(StorageError::ConfigNotFound) => None,
+            Err(err) => return Err(ApiError::from(err).into()),
+        };
+        let (remaining_config, updated_targets) = if let Some(config) = replication_config.as_ref() {
+            let (remaining, removable_arns) = split_replication_config_for_user_delete(config.clone(), &site_peers, contract);
+            let targets = replication_targets_without_arns(&bucket, &removable_arns).await?;
+            (remaining, targets)
+        } else {
+            (None, None)
+        };
+
+        match remaining_config {
+            // Site-replication rules and the targets backing them survive the
+            // S3 delete (issue #1948); only the operator-authored rules go.
+            Some(remaining) => {
+                let data = serialize_config(&remaining)?;
+                update_bucket_config_for_incarnation(&bucket, BUCKET_REPLICATION_CONFIG, data, expected_incarnation_id)
+                    .await
+                    .map_err(ApiError::from)?;
+            }
+            None => {
+                delete_bucket_config_for_incarnation(&bucket, BUCKET_REPLICATION_CONFIG, expected_incarnation_id)
+                    .await
+                    .map_err(ApiError::from)?;
+            }
+        }
+        if let Some((targets, removed)) = updated_targets
+            && let Err(err) =
+                write_replication_targets_after_config_delete(&bucket, &targets, removed, expected_incarnation_id).await
+        {
+            if let Some(config) = replication_config.as_ref() {
+                return Err(restore_replication_config_after_target_cleanup_failure(
+                    &bucket,
+                    config,
+                    err,
+                    expected_incarnation_id,
+                )
+                .await);
+            }
+            return Err(err);
+        }
+        drop(targets_guard);
+
+        notify_bucket_metadata_reload(bucket.clone(), "delete bucket replication", request_context, true).await;
+
+        let item = sr_bucket_meta_item(bucket.clone(), "replication-config");
+        if let Err(err) = site_replication_bucket_meta_hook(item).await {
+            warn!(bucket = %bucket, error = ?err, "site replication bucket replication-config delete hook failed");
+        }
+
+        info!(bucket = %bucket, "deleted bucket replication config");
+
+        Ok(S3Response::new(DeleteBucketReplicationOutput::default()))
+    }
+
+    /// Answers the stored configuration in the engine's shape; the s3s edge
+    /// (`storage::ecfs`) converts it into the `GetBucketReplication` output.
+    pub async fn execute_get_bucket_replication(
+        &self,
+        req: S3Request<GetBucketReplicationInput>,
+    ) -> S3Result<PersistedReplicationConfiguration> {
+        let GetBucketReplicationInput { bucket, .. } = req.input;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        let replication_configuration = match metadata_sys::get_replication_config(&bucket).await {
+            Ok((cfg, _created)) => cfg,
+            Err(StorageError::ConfigNotFound) => {
+                return Err(S3Error::with_message(
+                    S3ErrorCode::ReplicationConfigurationNotFoundError,
+                    "replication not found".to_string(),
+                ));
+            }
+            Err(err) => {
+                warn!(
+                    component = LOG_COMPONENT_APP,
+                    subsystem = LOG_SUBSYSTEM_BUCKET,
+                    event = "bucket_replication_config_load_failed",
+                    bucket = %bucket,
+                    error = ?err,
+                    "Failed to load bucket replication configuration"
+                );
+                return Err(ApiError::from(err).into());
+            }
+        };
+
+        Ok(replication_configuration)
+    }
+
+    /// See [`Self::execute_delete_bucket_replication`] for `site_peers` and
+    /// `contract`. `replication_configuration` is the request body in the
+    /// engine's shape: the s3s edge (`storage::ecfs`) converts it, and this use
+    /// case never reads `req.input.replication_configuration`.
+    pub async fn execute_put_bucket_replication(
+        &self,
+        req: S3Request<PutBucketReplicationInput>,
+        replication_configuration: PersistedReplicationConfiguration,
+        site_peers: HashSet<String>,
+        contract: OperatorRuleContract,
+    ) -> S3Result<S3Response<PutBucketReplicationOutput>> {
+        let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
+        let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
+        let PutBucketReplicationInput { bucket, .. } = req.input;
+        info!(bucket = %bucket, "updating bucket replication config");
+
+        validate_replication_config_capabilities(&replication_configuration)?;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        let targets_guard = lock_bucket_targets_metadata(&bucket).await;
+        validate_bucket_replication_update(&bucket, &replication_configuration).await?;
+        let existing_config = match metadata_sys::get_replication_config(&bucket).await {
+            Ok((config, _)) => Some(config),
+            Err(StorageError::ConfigNotFound) => None,
+            Err(err) => return Err(ApiError::from(err).into()),
+        };
+        let replication_configuration =
+            merge_user_replication_config_update(replication_configuration, existing_config, &site_peers, contract);
+        let data = serialize_config(&replication_configuration)?;
+        update_bucket_config_for_incarnation(&bucket, BUCKET_REPLICATION_CONFIG, data, expected_incarnation_id)
+            .await
+            .map_err(ApiError::from)?;
+        drop(targets_guard);
+
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket replication", request_context, true).await;
+
+        let mut item = sr_bucket_meta_item(bucket.clone(), "replication-config");
+        item.replication_config = Some(
+            serialize_config(&replication_configuration).and_then(|bytes| String::from_utf8(bytes).map_err(to_internal_error))?,
+        );
+        if let Err(err) = site_replication_bucket_meta_hook(item).await {
+            warn!(bucket = %bucket, error = ?err, "site replication bucket replication-config hook failed");
+        }
+
+        Ok(S3Response::new(PutBucketReplicationOutput::default()))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::app::bucket_usecase::test_support::build_request;
+    use http::Method;
+    use rustfs_gateway_types::dto::Status;
+    use rustfs_gateway_types::persistence::{
+        PersistedEncryptionConfiguration, PersistedReplicationDestination, PersistedReplicationRule,
+    };
+
+    fn replication_rule_for_target(arn: &str) -> PersistedReplicationRule {
+        PersistedReplicationRule {
+            delete_marker_replication: None,
+            delete_replication: None,
+            destination: PersistedReplicationDestination {
+                bucket: arn.to_string(),
+                ..Default::default()
+            },
+            existing_object_replication: None,
+            filter: None,
+            id: Some("rule-1".to_string()),
+            prefix: None,
+            priority: Some(1),
+            source_selection_criteria: None,
+            status: Status::ENABLED.to_string(),
+        }
+    }
+
+    #[test]
+    fn replication_target_arns_use_role_when_present() {
+        let role = "arn:rustfs:replication:us-east-1:source:bucket";
+        let destination = "arn:rustfs:replication:us-east-1:target:bucket";
+        let config = PersistedReplicationConfiguration {
+            role: format!(" {role} "),
+            rules: vec![replication_rule_for_target(destination)],
+        };
+
+        let arns = replication_target_arns(&config);
+
+        assert!(arns.contains(role));
+        assert!(!arns.contains(destination));
+    }
+
+    #[test]
+    fn replication_target_arns_use_rule_destinations_without_role() {
+        let destination = "arn:rustfs:replication:us-east-1:target:bucket";
+        let config = PersistedReplicationConfiguration {
+            role: String::new(),
+            rules: vec![replication_rule_for_target(destination)],
+        };
+
+        let arns = replication_target_arns(&config);
+
+        assert!(arns.contains(destination));
+    }
+
+    fn replication_rule_with_id(arn: &str, id: &str, priority: i32) -> PersistedReplicationRule {
+        let mut rule = replication_rule_for_target(arn);
+        rule.id = Some(id.to_string());
+        rule.priority = Some(priority);
+        rule
+    }
+
+    fn site_peers(deployment_ids: &[&str]) -> HashSet<String> {
+        deployment_ids.iter().map(|id| id.to_string()).collect()
+    }
+
+    #[test]
+    fn put_replication_merge_preserves_site_replication_rules() {
+        let existing = PersistedReplicationConfiguration {
+            role: String::new(),
+            rules: vec![
+                replication_rule_with_id("arn:rustfs:replication::peer-dep:bucket", "site-repl-peer-dep", 1),
+                replication_rule_with_id("arn:rustfs:replication:us-east-1:old:bucket", "old-user-rule", 2),
+            ],
+        };
+        let incoming = PersistedReplicationConfiguration {
+            role: String::new(),
+            rules: vec![
+                replication_rule_with_id("arn:rustfs:replication:us-east-1:new:bucket", "new-user-rule", 1),
+                replication_rule_with_id("arn:rustfs:replication::forged-dep:bucket", "site-repl-peer-dep", 2),
+                replication_rule_with_id("arn:rustfs:replication::other-dep:bucket", "site-repl-other", 3),
+            ],
+        };
+
+        let merged = merge_user_replication_config_update(
+            incoming,
+            Some(existing),
+            &site_peers(&["peer-dep"]),
+            OperatorRuleContract::Derived,
+        );
+
+        let rules: Vec<_> = merged
+            .rules
+            .iter()
+            .map(|rule| (rule.id.as_deref().unwrap_or_default(), rule.destination.bucket.as_str()))
+            .collect();
+        assert_eq!(
+            rules,
+            vec![
+                ("new-user-rule", "arn:rustfs:replication:us-east-1:new:bucket"),
+                ("site-repl-other", "arn:rustfs:replication::other-dep:bucket"),
+                ("site-repl-peer-dep", "arn:rustfs:replication::peer-dep:bucket"),
+            ],
+            "user rules replaced, the reconciler rule for the current peer kept over the incoming impostor, \
+             a site-repl-* id that names no current peer is ordinary operator state"
+        );
+    }
+
+    // Rule ids do not reserve `site-repl-*`: outside site replication an
+    // owner's `site-repl-user` rule is ordinary state, so PUT stores it
+    // verbatim and DELETE removes it and garbage-collects its target.
+    #[test]
+    fn put_then_delete_replication_without_site_replication_treats_site_repl_id_as_user_rule() {
+        let user_arn = "arn:minio:replication:us-east-1:2f1c-remote:bucket";
+        let incoming = PersistedReplicationConfiguration {
+            role: String::new(),
+            rules: vec![replication_rule_with_id(user_arn, "site-repl-user", 1)],
+        };
+
+        let stored = merge_user_replication_config_update(incoming.clone(), None, &HashSet::new(), OperatorRuleContract::Derived);
+        assert_eq!(stored, incoming, "PUT on a non-site-replication bucket is verbatim");
+
+        let (remaining, removable) =
+            split_replication_config_for_user_delete(stored, &HashSet::new(), OperatorRuleContract::Derived);
+        assert!(remaining.is_none(), "DELETE must remove the operator's site-repl-* rule");
+        assert_eq!(removable, HashSet::from([user_arn.to_string()]));
+    }
+
+    // Under site replication only a rule the reconciler would derive — id
+    // `site-repl-<peer>` for a current peer, destination ARN naming the same
+    // peer — is reconciler-owned. Everything else is operator state.
+    #[test]
+    fn delete_replication_split_keeps_only_reconciler_derived_rules() {
+        let peer_arn = "arn:rustfs:replication::peer-dep:bucket";
+        let user_arn = "arn:minio:replication:us-east-1:2f1c-remote:bucket";
+        let config = PersistedReplicationConfiguration {
+            role: String::new(),
+            rules: vec![
+                replication_rule_with_id(user_arn, "site-repl-user", 1),
+                replication_rule_with_id(user_arn, "site-repl-peer-dep", 2),
+                replication_rule_with_id("arn:rustfs:replication::gone-dep:bucket", "site-repl-gone-dep", 3),
+                replication_rule_with_id(peer_arn, "site-repl-peer-dep", 4),
+            ],
+        };
+
+        let (remaining, removable) =
+            split_replication_config_for_user_delete(config, &site_peers(&["peer-dep"]), OperatorRuleContract::Derived);
+
+        let remaining = remaining.expect("the reconciler-derived rule must survive");
+        assert_eq!(remaining.rules.len(), 1);
+        assert_eq!(remaining.rules[0].destination.bucket, peer_arn);
+        assert_eq!(
+            removable,
+            HashSet::from([user_arn.to_string(), "arn:rustfs:replication::gone-dep:bucket".to_string()]),
+            "targets of operator rules and of a removed peer are garbage-collected"
+        );
+    }
+
+    #[test]
+    fn put_replication_merge_returns_incoming_verbatim_without_site_rules() {
+        let existing = PersistedReplicationConfiguration {
+            role: String::new(),
+            rules: vec![replication_rule_with_id(
+                "arn:rustfs:replication:us-east-1:old:bucket",
+                "old-user-rule",
+                7,
+            )],
+        };
+        let incoming = PersistedReplicationConfiguration {
+            role: String::new(),
+            rules: vec![replication_rule_with_id(
+                "arn:rustfs:replication:us-east-1:new:bucket",
+                "new-user-rule",
+                5,
+            )],
+        };
+
+        let merged = merge_user_replication_config_update(
+            incoming.clone(),
+            Some(existing),
+            &HashSet::new(),
+            OperatorRuleContract::Derived,
+        );
+
+        assert_eq!(merged.role, incoming.role);
+        assert_eq!(merged.rules, incoming.rules, "non-SR buckets keep the verbatim overwrite semantics");
+    }
+
+    #[test]
+    fn delete_replication_split_keeps_site_rules_and_their_targets() {
+        let sr_arn = "arn:rustfs:replication::peer-dep:bucket";
+        let user_arn = "arn:rustfs:replication:us-east-1:user:bucket";
+        let config = PersistedReplicationConfiguration {
+            role: String::new(),
+            rules: vec![
+                replication_rule_with_id(user_arn, "user-rule", 1),
+                replication_rule_with_id(sr_arn, "site-repl-peer-dep", 2),
+            ],
+        };
+
+        let (remaining, removable) =
+            split_replication_config_for_user_delete(config, &site_peers(&["peer-dep"]), OperatorRuleContract::Derived);
+
+        let remaining = remaining.expect("site-replication rules must survive a user delete");
+        let ids: Vec<_> = remaining
+            .rules
+            .iter()
+            .map(|rule| rule.id.as_deref().unwrap_or_default())
+            .collect();
+        assert_eq!(ids, vec!["site-repl-peer-dep"]);
+        assert_eq!(removable, HashSet::from([user_arn.to_string()]));
+    }
+
+    #[test]
+    fn delete_replication_split_protects_targets_shared_with_site_rules() {
+        let sr_arn = "arn:rustfs:replication::peer-dep:bucket";
+        let config = PersistedReplicationConfiguration {
+            role: String::new(),
+            rules: vec![
+                replication_rule_with_id(sr_arn, "user-rule-on-sr-target", 1),
+                replication_rule_with_id(sr_arn, "site-repl-peer-dep", 2),
+            ],
+        };
+
+        let (remaining, removable) =
+            split_replication_config_for_user_delete(config, &site_peers(&["peer-dep"]), OperatorRuleContract::Derived);
+
+        assert!(remaining.is_some());
+        assert!(
+            removable.is_empty(),
+            "a target still referenced by a surviving site-replication rule must not be removed"
+        );
+    }
+
+    #[test]
+    fn delete_replication_split_removes_everything_without_site_rules() {
+        let user_arn = "arn:rustfs:replication:us-east-1:user:bucket";
+        let config = PersistedReplicationConfiguration {
+            role: String::new(),
+            rules: vec![replication_rule_with_id(user_arn, "user-rule", 1)],
+        };
+
+        let (remaining, removable) =
+            split_replication_config_for_user_delete(config, &site_peers(&["peer-dep"]), OperatorRuleContract::Derived);
+
+        assert!(remaining.is_none(), "without site-replication rules the whole config is deleted");
+        assert_eq!(removable, HashSet::from([user_arn.to_string()]));
+    }
+
+    fn replication_targets_with_arn(arns: &[&str]) -> BucketTargets {
+        BucketTargets {
+            targets: arns
+                .iter()
+                .map(|arn| BucketTarget {
+                    arn: (*arn).to_string(),
+                    target_type: BucketTargetType::ReplicationService,
+                    ..Default::default()
+                })
+                .collect(),
+        }
+    }
+
+    #[test]
+    fn validate_replication_config_targets_accepts_matching_destination_arns() {
+        let arn = "arn:rustfs:replication:us-east-1:target:bucket";
+        let targets = replication_targets_with_arn(&[arn]);
+        let config = PersistedReplicationConfiguration {
+            role: String::new(),
+            rules: vec![replication_rule_for_target(arn)],
+        };
+
+        validate_replication_config_targets(&targets, &config).expect("matching target should pass validation");
+    }
+
+    #[test]
+    fn validate_replication_config_targets_rejects_stale_destination_arns() {
+        let targets = replication_targets_with_arn(&["arn:rustfs:replication:us-east-1:target:bucket-a"]);
+        let config = PersistedReplicationConfiguration {
+            role: String::new(),
+            rules: vec![replication_rule_for_target(
+                "arn:rustfs:replication:us-east-1:target:bucket-b",
+            )],
+        };
+
+        let err = validate_replication_config_targets(&targets, &config).expect_err("stale target should fail validation");
+        assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
+    }
+
+    #[test]
+    fn validate_replication_config_targets_accepts_matching_role_arn() {
+        let arn = "arn:rustfs:replication:us-east-1:role-target:bucket";
+        let targets = replication_targets_with_arn(&[arn]);
+        let config = PersistedReplicationConfiguration {
+            role: format!(" {arn} "),
+            rules: vec![replication_rule_for_target("arn:rustfs:replication:us-east-1:ignored:bucket")],
+        };
+
+        validate_replication_config_targets(&targets, &config).expect("matching role ARN should pass validation");
+    }
+
+    #[test]
+    fn validate_replication_config_targets_rejects_role_with_multiple_destinations() {
+        let role = "arn:rustfs:replication:us-east-1:role-target:bucket";
+        let targets = replication_targets_with_arn(&[role]);
+        let config = PersistedReplicationConfiguration {
+            role: role.to_string(),
+            rules: vec![
+                replication_rule_for_target("arn:rustfs:replication:us-east-1:target-a:bucket"),
+                replication_rule_for_target("arn:rustfs:replication:us-east-1:target-b:bucket"),
+            ],
+        };
+
+        let err = validate_replication_config_targets(&targets, &config)
+            .expect_err("role plus multiple destinations should be rejected");
+        assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
+    }
+
+    #[test]
+    fn validate_replication_config_targets_trims_destination_arns() {
+        let arn = "arn:rustfs:replication:us-east-1:target:bucket";
+        let targets = replication_targets_with_arn(&[arn]);
+        let config = PersistedReplicationConfiguration {
+            role: String::new(),
+            rules: vec![replication_rule_for_target(
+                " arn:rustfs:replication:us-east-1:target:bucket ",
+            )],
+        };
+
+        validate_replication_config_targets(&targets, &config).expect("trimmed destination ARN should match configured target");
+    }
+
+    #[test]
+    fn validate_replication_config_targets_ignores_disabled_rules() {
+        let targets = replication_targets_with_arn(&[]);
+        let mut rule = replication_rule_for_target("arn:rustfs:replication:us-east-1:stale:bucket");
+        rule.status = Status::DISABLED.to_string();
+        let config = PersistedReplicationConfiguration {
+            role: String::new(),
+            rules: vec![rule],
+        };
+
+        validate_replication_config_targets(&targets, &config).expect("disabled rules should not require live targets");
+    }
+
+    #[test]
+    fn validate_replication_config_capabilities_names_unsupported_field() {
+        let mut rule = replication_rule_for_target("arn:rustfs:replication:us-east-1:target:bucket");
+        let destination_key_id = "arn:aws:kms:us-east-1:123456789012:key/opaque-key-id";
+        rule.destination.encryption_configuration = Some(PersistedEncryptionConfiguration {
+            replica_kms_key_id: Some(destination_key_id.to_string()),
+        });
+        let config = PersistedReplicationConfiguration {
+            role: String::new(),
+            rules: vec![rule],
+        };
+
+        let err = validate_replication_config_capabilities(&config)
+            .expect_err("destination encryption must be rejected until the execution path supports it");
+
+        assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
+        assert!(
+            err.to_string()
+                .contains("Destination.EncryptionConfiguration is not supported")
+        );
+        assert!(!err.to_string().contains(destination_key_id));
+    }
+
+    #[test]
+    fn validate_replication_config_capabilities_rejects_structural_defects_before_write() {
+        let mut first = replication_rule_for_target("arn:rustfs:replication:us-east-1:target:bucket");
+        first.priority = Some(1);
+        let mut second = replication_rule_for_target("arn:rustfs:replication:us-east-1:target:bucket");
+        second.priority = Some(1);
+        let config = PersistedReplicationConfiguration {
+            role: String::new(),
+            rules: vec![first, second],
+        };
+
+        let err = validate_replication_config_capabilities(&config)
+            .expect_err("duplicate rule priorities must be rejected before persistence");
+
+        assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
+        assert!(err.to_string().contains("Priority must be unique"));
+    }
+
+    #[test]
+    fn validate_replication_config_capabilities_rejects_invalid_status_before_write() {
+        let mut rule = replication_rule_for_target("arn:rustfs:replication:us-east-1:target:bucket");
+        rule.status = "Invalid".to_owned();
+        let config = PersistedReplicationConfiguration {
+            role: String::new(),
+            rules: vec![rule],
+        };
+
+        let err = validate_replication_config_capabilities(&config)
+            .expect_err("an invalid string-backed replication status must be rejected before persistence");
+
+        assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
+        assert!(err.to_string().contains("Rule.Status has an invalid status"));
+    }
+
+    #[test]
+    fn remove_replication_targets_from_config_targets_only_removes_referenced_replication_targets() {
+        let removed_arn = "arn:rustfs:replication:us-east-1:removed:bucket";
+        let kept_replication_arn = "arn:rustfs:replication:us-east-1:kept:bucket";
+        let kept_ilm_arn = "arn:rustfs:ilm:us-east-1:kept:bucket";
+        let mut targets = BucketTargets {
+            targets: vec![
+                BucketTarget {
+                    arn: removed_arn.to_string(),
+                    target_type: BucketTargetType::ReplicationService,
+                    ..Default::default()
+                },
+                BucketTarget {
+                    arn: kept_replication_arn.to_string(),
+                    target_type: BucketTargetType::ReplicationService,
+                    ..Default::default()
+                },
+                BucketTarget {
+                    arn: kept_ilm_arn.to_string(),
+                    target_type: BucketTargetType::IlmService,
+                    ..Default::default()
+                },
+            ],
+        };
+        let target_arns = HashSet::from([removed_arn.to_string(), kept_ilm_arn.to_string()]);
+
+        let removed = remove_replication_targets_from_config_targets(&mut targets, &target_arns);
+
+        assert_eq!(removed, 1);
+        let remaining_arns = targets
+            .targets
+            .iter()
+            .map(|target| target.arn.as_str())
+            .collect::<HashSet<_>>();
+        assert!(!remaining_arns.contains(removed_arn));
+        assert!(remaining_arns.contains(kept_replication_arn));
+        assert!(remaining_arns.contains(kept_ilm_arn));
+    }
+
+    #[tokio::test]
+    async fn execute_delete_bucket_replication_returns_internal_error_when_store_uninitialized() {
+        let input = DeleteBucketReplicationInput::builder()
+            .bucket("test-bucket".to_string())
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::DELETE);
+        let usecase = DefaultBucketUsecase::without_context();
+
+        let err = usecase
+            .execute_delete_bucket_replication(req, HashSet::new(), OperatorRuleContract::Derived)
+            .await
+            .unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+    }
+
+    #[tokio::test]
+    async fn execute_get_bucket_replication_returns_internal_error_when_store_uninitialized() {
+        let input = GetBucketReplicationInput::builder()
+            .bucket("test-bucket".to_string())
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::GET);
+        let usecase = DefaultBucketUsecase::without_context();
+
+        let err = usecase.execute_get_bucket_replication(req).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+    }
+
+    #[tokio::test]
+    async fn execute_put_bucket_replication_returns_internal_error_when_store_uninitialized() {
+        // The config must clear the structural/capability validators so the
+        // request actually reaches the store lookup this test pins.
+        let configuration = PersistedReplicationConfiguration {
+            role: "arn:aws:iam::123456789012:role/test".to_string(),
+            rules: vec![replication_rule_for_target("arn:rustfs:replication:us-east-1:target:bucket")],
+        };
+        let input = PutBucketReplicationInput::builder()
+            .bucket("test-bucket".to_string())
+            .replication_configuration(Default::default())
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::PUT);
+        let usecase = DefaultBucketUsecase::without_context();
+
+        let err = usecase
+            .execute_put_bucket_replication(req, configuration, HashSet::new(), OperatorRuleContract::Derived)
+            .await
+            .unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+    }
+
+    #[tokio::test]
+    async fn execute_put_bucket_replication_rejects_unsupported_fields_before_store_or_metadata_write() {
+        let mut rule = replication_rule_for_target("arn:rustfs:replication:us-east-1:target:bucket");
+        rule.destination.account = Some("123456789012".to_string());
+        let configuration = PersistedReplicationConfiguration {
+            role: String::new(),
+            rules: vec![rule],
+        };
+        let input = PutBucketReplicationInput::builder()
+            .bucket("test-bucket".to_string())
+            .replication_configuration(Default::default())
+            .build()
+            .unwrap();
+
+        let err = DefaultBucketUsecase::without_context()
+            .execute_put_bucket_replication(
+                build_request(input, Method::PUT),
+                configuration,
+                HashSet::new(),
+                OperatorRuleContract::Derived,
+            )
+            .await
+            .expect_err("unsupported fields must be rejected before store access");
+
+        assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
+        assert_eq!(
+            err.message(),
+            Some("replication field Destination.Account is not supported by this RustFS version")
+        );
+    }
+}

rustfs/src/app/bucket_usecase/request_payment.rs

@@ -0,0 +1,15 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Bucket request-payment configuration use-case operations (reserved for the next batch).

rustfs/src/app/bucket_usecase/tagging.rs

@@ -0,0 +1,247 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Bucket tagging use-case operations.
+
+use super::*;
+
+impl DefaultBucketUsecase {
+    #[instrument(level = "debug", skip(self, req))]
+    pub async fn execute_delete_bucket_tagging(
+        &self,
+        req: S3Request<DeleteBucketTaggingInput>,
+    ) -> S3Result<S3Response<DeleteBucketTaggingOutput>> {
+        let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
+        let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
+        let DeleteBucketTaggingInput { bucket, .. } = req.input;
+
+        metadata_sys::update_s3_tagging_config(&bucket, None, expected_incarnation_id)
+            .await
+            .map_err(ApiError::from)?;
+
+        notify_bucket_metadata_reload(bucket.clone(), "delete bucket tagging", request_context, false).await;
+
+        let item = sr_bucket_meta_item(bucket.clone(), "tags");
+        if let Err(err) = site_replication_bucket_meta_hook(item).await {
+            warn!(bucket = %bucket, error = ?err, "site replication bucket tagging delete hook failed");
+        }
+
+        rustfs_scanner::record_dirty_usage_bucket_from_producer(
+            &bucket,
+            rustfs_scanner::SegmentInvalidationProducerIdentity::BucketMetadata,
+        );
+        Ok(S3Response::new(DeleteBucketTaggingOutput {}))
+    }
+
+    #[instrument(level = "debug", skip(self, req))]
+    pub async fn execute_get_bucket_tagging(&self, req: S3Request<GetBucketTaggingInput>) -> S3Result<PersistedTagging> {
+        let GetBucketTaggingInput { bucket, .. } = req.input;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        let tagging = match metadata_sys::get_tagging_config(&bucket).await {
+            Ok((tags, _)) => tags,
+            Err(err) => {
+                if err == StorageError::ConfigNotFound {
+                    return Err(S3Error::with_message(S3ErrorCode::NoSuchTagSet, "The TagSet does not exist".to_string()));
+                }
+                warn!(
+                    component = LOG_COMPONENT_APP,
+                    subsystem = LOG_SUBSYSTEM_BUCKET,
+                    event = "bucket_tagging_config_load_failed",
+                    bucket = %bucket,
+                    error = ?err,
+                    "Failed to load bucket tagging configuration"
+                );
+                return Err(ApiError::from(err).into());
+            }
+        };
+
+        Ok(tagging)
+    }
+
+    #[instrument(level = "debug", skip(self, req))]
+    pub async fn execute_put_bucket_tagging(
+        &self,
+        req: S3Request<PutBucketTaggingInput>,
+        tagging: PersistedTagging,
+    ) -> S3Result<S3Response<PutBucketTaggingOutput>> {
+        let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
+        let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
+        let PutBucketTaggingInput { bucket, .. } = req.input;
+        let s3_tag_set: Vec<s3s_crate::dto::Tag> = tagging
+            .tag_set
+            .iter()
+            .map(|tag| s3s_crate::dto::Tag {
+                key: tag.key.clone(),
+                value: tag.value.clone(),
+            })
+            .collect();
+        crate::app::storage_api::bucket_usecase::s3_api::tagging::validate_bucket_tag_set(&s3_tag_set)?;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        metadata_sys::update_s3_tagging_config(&bucket, Some(tagging.clone()), expected_incarnation_id)
+            .await
+            .map_err(ApiError::from)?;
+
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket tagging", request_context, false).await;
+
+        let mut item = sr_bucket_meta_item(bucket.clone(), "tags");
+        item.tags = Some(serialize_config(&tagging).and_then(|bytes| String::from_utf8(bytes).map_err(to_internal_error))?);
+        if let Err(err) = site_replication_bucket_meta_hook(item).await {
+            warn!(bucket = %bucket, error = ?err, "site replication bucket tagging hook failed");
+        }
+
+        rustfs_scanner::record_dirty_usage_bucket_from_producer(
+            &bucket,
+            rustfs_scanner::SegmentInvalidationProducerIdentity::BucketMetadata,
+        );
+        Ok(S3Response::new(PutBucketTaggingOutput::default()))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::app::bucket_usecase::test_support::build_request;
+    use http::Method;
+
+    #[tokio::test]
+    async fn execute_get_bucket_tagging_returns_internal_error_when_store_uninitialized() {
+        let input = GetBucketTaggingInput::builder()
+            .bucket("test-bucket".to_string())
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::GET);
+        let usecase = DefaultBucketUsecase::without_context();
+
+        let err = usecase.execute_get_bucket_tagging(req).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn put_bucket_tagging_rejects_invalid_shapes_before_storage_mutation() {
+        crate::app::gating_test_env::run_large_stack_test("bucket-tag-validation", || async {
+            use crate::app::runtime_sources::{AppContext, ServerContextSlot};
+            use crate::app::storage_api::test::ReqInfo;
+            use crate::app::storage_api::test::ecfs::FS;
+            use s3s_crate::access::S3Access as _;
+            let ambient = crate::app::gating_test_env::shared_gating_ambient().await;
+            let store = crate::app::gating_test_env::shared_gating_ecstore().await;
+            let context = Arc::new(AppContext::new(store.clone(), ambient.iam(), ambient.kms()));
+            let slot = ServerContextSlot::new();
+            assert!(slot.install(context.clone()));
+            let fs = FS::with_server_ctx(slot.clone());
+            let usecase = DefaultBucketUsecase::with_context(Some(context));
+            let bucket = format!("tag-validation-{}", uuid::Uuid::new_v4().simple());
+            store.make_bucket(&bucket, &MakeBucketOptions::default()).await.unwrap();
+            store
+                .update_bucket_metadata_config(
+                    &bucket,
+                    BUCKET_POLICY_CONFIG,
+                    serde_json::to_vec(
+                        &serde_json::json!({"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"AWS":"*"},
+                    "Action":"s3:PutBucketTagging","Resource":format!("arn:aws:s3:::{bucket}")}]}),
+                    )
+                    .unwrap(),
+                )
+                .await
+                .unwrap();
+            let tag = s3s_crate::dto::Tag {
+                key: Some("team".to_owned()),
+                value: Some("storage".to_owned()),
+            };
+            for (tag_set, valid) in [
+                (vec![tag.clone()], true),
+                (
+                    vec![s3s_crate::dto::Tag {
+                        key: Some("team".to_owned()),
+                        value: None,
+                    }],
+                    false,
+                ),
+                (vec![tag.clone(), tag], false),
+            ] {
+                let persisted = PersistedTagging {
+                    tag_set: tag_set
+                        .iter()
+                        .map(|tag| rustfs_gateway_types::cors_tagging::PersistedTag {
+                            key: tag.key.clone(),
+                            value: tag.value.clone(),
+                        })
+                        .collect(),
+                };
+                let input = PutBucketTaggingInput::builder()
+                    .bucket(bucket.clone())
+                    .tagging(s3s_crate::dto::Tagging { tag_set })
+                    .build()
+                    .unwrap();
+                let mut req = build_request(input, Method::PUT);
+                req.extensions.insert(ReqInfo::default());
+                req.extensions.insert(slot.clone());
+                fs.put_bucket_tagging(&mut req).await.expect("permission allows tag updates");
+                let result = usecase.execute_put_bucket_tagging(req, persisted).await;
+                if valid {
+                    result.expect("valid tags are persisted");
+                } else {
+                    assert_eq!(result.unwrap_err().code(), &S3ErrorCode::InvalidTag);
+                }
+                assert_eq!(
+                    store.get_bucket_tags_for_policy(&bucket).await.unwrap(),
+                    HashMap::from([("team".to_owned(), "storage".to_owned())])
+                );
+            }
+        });
+    }
+
+    #[tokio::test]
+    async fn execute_put_bucket_tagging_returns_internal_error_when_store_uninitialized() {
+        let input = PutBucketTaggingInput::builder()
+            .bucket("test-bucket".to_string())
+            .tagging(Tagging {
+                tag_set: vec![Tag {
+                    key: Some("env".to_string()),
+                    value: Some("prod".to_string()),
+                }],
+            })
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::PUT);
+        let usecase = DefaultBucketUsecase::without_context();
+
+        let err = usecase
+            .execute_put_bucket_tagging(req, PersistedTagging::default())
+            .await
+            .unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+    }
+}

rustfs/src/app/bucket_usecase/test_support.rs

@@ -0,0 +1,46 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Request fixtures shared by the bucket use-case tests.
+
+use super::*;
+use http::{Extensions, HeaderMap, Method, Uri};
+
+pub(super) fn s3_op_total(op: S3Operation) -> u64 {
+    rustfs_io_metrics::s3_op_metrics_snapshot()
+        .into_iter()
+        .find(|snapshot| snapshot.op == op.as_str())
+        .map(|snapshot| snapshot.total)
+        .unwrap_or_default()
+}
+
+pub(super) fn build_request<T>(input: T, method: Method) -> S3Request<T> {
+    S3Request {
+        input,
+        method,
+        uri: Uri::from_static("/"),
+        headers: HeaderMap::new(),
+        extensions: Extensions::new(),
+        credentials: None,
+        region: None,
+        service: None,
+        trailing_headers: None,
+    }
+}
+
+pub(super) fn build_request_with_req_info<T>(input: T, method: Method, req_info: ReqInfo) -> S3Request<T> {
+    let mut req = build_request(input, method);
+    req.extensions.insert(req_info);
+    req
+}

rustfs/src/app/bucket_usecase/versioning.rs

@@ -0,0 +1,193 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Bucket versioning use-case operations.
+
+use super::*;
+use crate::app::storage_api::bucket_usecase::bucket::{ObjectLockConfigExt as _, VersioningConfigExt as _};
+
+fn versioning_configuration_has_object_lock_incompatible_settings(config: &PersistedVersioningConfiguration) -> bool {
+    config.suspended()
+        || config.exclude_folders.unwrap_or(false)
+        || config
+            .excluded_prefixes
+            .as_ref()
+            .is_some_and(|excluded_prefixes| !excluded_prefixes.is_empty())
+}
+
+async fn validate_bucket_versioning_update(bucket: &str, config: &PersistedVersioningConfiguration) -> S3Result<()> {
+    if config
+        .status
+        .as_ref()
+        .is_some_and(|status| !matches!(status.as_str(), "Enabled" | "Suspended"))
+    {
+        return Err(S3Error::with_message(
+            S3ErrorCode::InvalidArgument,
+            "bucket versioning configuration has an invalid status",
+        ));
+    }
+    match metadata_sys::get_object_lock_config(bucket).await {
+        Ok((object_lock_config, _)) => {
+            if object_lock_config.enabled() && versioning_configuration_has_object_lock_incompatible_settings(config) {
+                return Err(S3Error::with_message(
+                    S3ErrorCode::InvalidBucketState,
+                    "An Object Lock configuration is present on this bucket, versioning cannot be suspended.".to_string(),
+                ));
+            }
+        }
+        Err(StorageError::ConfigNotFound) => {}
+        Err(err) => return Err(ApiError::from(err).into()),
+    }
+    // AWS S3 and MinIO both refuse to suspend versioning while a replication
+    // configuration exists: suspension would start minting null versions that
+    // the replication engine (versioned by contract) can never converge.
+    if config.suspended() {
+        match metadata_sys::get_replication_config(bucket).await {
+            Ok(_) => {
+                return Err(S3Error::with_message(
+                    S3ErrorCode::InvalidBucketState,
+                    "A replication configuration is present on this bucket, bucket wide versioning cannot be suspended."
+                        .to_string(),
+                ));
+            }
+            Err(StorageError::ConfigNotFound) => {}
+            Err(err) => return Err(ApiError::from(err).into()),
+        }
+    }
+
+    Ok(())
+}
+
+impl DefaultBucketUsecase {
+    #[instrument(level = "debug", skip(self, req))]
+    pub async fn execute_get_bucket_versioning(
+        &self,
+        req: S3Request<GetBucketVersioningInput>,
+    ) -> S3Result<PersistedVersioningConfiguration> {
+        let GetBucketVersioningInput { bucket, .. } = req.input;
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        BucketVersioningSys::get(&bucket)
+            .await
+            .map_err(|err| ApiError::from(err).into())
+    }
+
+    #[instrument(level = "debug", skip(self, req))]
+    pub async fn execute_put_bucket_versioning(
+        &self,
+        req: S3Request<PutBucketVersioningInput>,
+        prepared: S3Result<PersistedVersioningConfiguration>,
+    ) -> S3Result<S3Response<PutBucketVersioningOutput>> {
+        let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
+        let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
+        let PutBucketVersioningInput { bucket, .. } = req.input;
+
+        let versioning_configuration = prepared?;
+        validate_bucket_versioning_update(&bucket, &versioning_configuration).await?;
+
+        let data = serialize_config(&versioning_configuration)?;
+
+        update_bucket_config_for_incarnation(&bucket, BUCKET_VERSIONING_CONFIG, data, expected_incarnation_id)
+            .await
+            .map_err(ApiError::from)?;
+
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket versioning", request_context, false).await;
+
+        let mut item = sr_bucket_meta_item(bucket.clone(), "version-config");
+        item.versioning = Some(
+            serialize_config(&versioning_configuration).and_then(|bytes| String::from_utf8(bytes).map_err(to_internal_error))?,
+        );
+        if let Err(err) = site_replication_bucket_meta_hook(item).await {
+            warn!(bucket = %bucket, error = ?err, "site replication bucket versioning hook failed");
+        }
+
+        rustfs_scanner::record_dirty_usage_bucket_from_producer(
+            &bucket,
+            rustfs_scanner::SegmentInvalidationProducerIdentity::BucketMetadata,
+        );
+        Ok(S3Response::new(PutBucketVersioningOutput {}))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::app::bucket_usecase::test_support::build_request;
+    use http::Method;
+
+    #[tokio::test]
+    async fn validate_bucket_versioning_update_rejects_invalid_status_before_metadata_lookup() {
+        let config = PersistedVersioningConfiguration {
+            status: Some("Invalid".to_owned()),
+            ..Default::default()
+        };
+
+        let err = validate_bucket_versioning_update("unregistered-test-bucket", &config)
+            .await
+            .expect_err("an invalid string-backed versioning status must be rejected before persistence");
+
+        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+    }
+
+    #[test]
+    fn versioning_configuration_has_object_lock_incompatible_settings_rejects_suspended() {
+        let config = PersistedVersioningConfiguration {
+            status: Some("Suspended".to_owned()),
+            ..Default::default()
+        };
+
+        assert!(versioning_configuration_has_object_lock_incompatible_settings(&config));
+    }
+
+    #[test]
+    fn versioning_configuration_has_object_lock_incompatible_settings_rejects_exclude_folders() {
+        let config = PersistedVersioningConfiguration {
+            exclude_folders: Some(true),
+            ..Default::default()
+        };
+
+        assert!(versioning_configuration_has_object_lock_incompatible_settings(&config));
+    }
+
+    #[test]
+    fn versioning_configuration_has_object_lock_incompatible_settings_rejects_excluded_prefixes() {
+        let config = PersistedVersioningConfiguration {
+            excluded_prefixes: Some(vec![Some("archive/".to_string())]),
+            ..Default::default()
+        };
+
+        assert!(versioning_configuration_has_object_lock_incompatible_settings(&config));
+    }
+
+    #[tokio::test]
+    async fn execute_get_bucket_versioning_returns_internal_error_when_store_uninitialized() {
+        let input = GetBucketVersioningInput::builder()
+            .bucket("test-bucket".to_string())
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::GET);
+        let usecase = DefaultBucketUsecase::without_context();
+
+        let err = usecase.execute_get_bucket_versioning(req).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+    }
+}

rustfs/src/app/bucket_usecase/website.rs

@@ -0,0 +1,15 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Bucket website configuration use-case operations (reserved for the next batch).

rustfs/src/guard_sources.rs

@@ -0,0 +1,287 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Test-only source reader for text guards that must survive a mechanical file split.
+//!
+//! A text guard asserts that some literal appears in production source. When it reads one file
+//! with `include_str!`, moving the guarded function into another file either breaks the guard or,
+//! worse, makes it stop seeing the code it guards. A guard built on [`SourceSet`] instead names
+//! the file it reads today plus the sibling directory a split moves that file's code into, and
+//! reads every `.rs` file of both at test time.
+//!
+//! The named file is required: a missing file panics with its path, so a guard never passes on
+//! absent input. The directory is optional while the split has not happened; once it exists,
+//! every `.rs` file under it (recursively) is read in byte order of its manifest-relative path,
+//! after the named file.
+
+use std::path::{Path, PathBuf};
+
+/// One source file of a [`SourceSet`].
+pub(crate) struct SourceFile {
+    /// Path relative to the `rustfs` package root, with `/` separators.
+    pub(crate) path: String,
+    pub(crate) text: String,
+}
+
+/// The ordered files a text guard reads: one required file, then the split directory.
+pub(crate) struct SourceSet {
+    files: Vec<SourceFile>,
+}
+
+impl SourceSet {
+    /// Reads `file` and every `.rs` file under `split_dir`, both relative to the `rustfs`
+    /// package root.
+    ///
+    /// # Panics
+    ///
+    /// Panics, naming the path, when `file` is missing or unreadable, when `split_dir` exists but
+    /// is not a directory, or when a file under `split_dir` cannot be read.
+    pub(crate) fn read(file: &str, split_dir: &str) -> Self {
+        Self::read_under(Path::new(env!("CARGO_MANIFEST_DIR")), file, split_dir)
+    }
+
+    fn read_under(root: &Path, file: &str, split_dir: &str) -> Self {
+        let required = root.join(file);
+        let text = std::fs::read_to_string(&required)
+            .unwrap_or_else(|err| panic!("guard source {} must exist and be readable: {err}", required.display()));
+        let mut files = vec![SourceFile {
+            path: file.to_owned(),
+            text,
+        }];
+
+        let dir = root.join(split_dir);
+        if dir.exists() {
+            assert!(dir.is_dir(), "guard split path {} must be a directory", dir.display());
+            let mut paths = Vec::new();
+            collect_rust_files(&dir, &mut paths);
+            let mut split_files: Vec<SourceFile> = paths
+                .into_iter()
+                .map(|path| {
+                    let text = std::fs::read_to_string(&path)
+                        .unwrap_or_else(|err| panic!("guard source {} must be readable: {err}", path.display()));
+                    let relative = path
+                        .strip_prefix(root)
+                        .unwrap_or_else(|_| panic!("guard source {} must be under {}", path.display(), root.display()));
+                    let path = relative
+                        .components()
+                        .map(|component| component.as_os_str().to_string_lossy())
+                        .collect::<Vec<_>>()
+                        .join("/");
+                    SourceFile { path, text }
+                })
+                .collect();
+            split_files.sort_by(|left, right| left.path.cmp(&right.path));
+            files.extend(split_files);
+        }
+
+        Self { files }
+    }
+
+    /// Every file of the set, the required file first.
+    pub(crate) fn files(&self) -> &[SourceFile] {
+        &self.files
+    }
+
+    /// Whether any one file of the set contains `needle`.
+    pub(crate) fn contains(&self, needle: &str) -> bool {
+        self.files.iter().any(|file| file.text.contains(needle))
+    }
+
+    /// Non-overlapping occurrences of `needle`, summed over the files; a match never spans two
+    /// files.
+    pub(crate) fn count(&self, needle: &str) -> usize {
+        self.files.iter().map(|file| file.text.matches(needle).count()).sum()
+    }
+
+    /// The text after `marker` in the one file that contains it.
+    ///
+    /// # Panics
+    ///
+    /// Panics unless `marker` occurs exactly once across the whole set.
+    pub(crate) fn after_unique_marker(&self, marker: &str) -> &str {
+        let occurrences = self.count(marker);
+        assert_eq!(occurrences, 1, "marker {marker:?} must occur exactly once across {}", self.describe());
+        self.files
+            .iter()
+            .find_map(|file| file.text.split_once(marker).map(|(_, rest)| rest))
+            .unwrap_or_else(|| unreachable!("a counted marker is found"))
+    }
+
+    /// Manifest-relative paths of the set, for panic messages.
+    pub(crate) fn describe(&self) -> String {
+        self.files
+            .iter()
+            .map(|file| file.path.as_str())
+            .collect::<Vec<_>>()
+            .join(", ")
+    }
+}
+
+fn collect_rust_files(dir: &Path, out: &mut Vec<PathBuf>) {
+    let entries =
+        std::fs::read_dir(dir).unwrap_or_else(|err| panic!("guard split directory {} must be readable: {err}", dir.display()));
+    for entry in entries {
+        let entry = entry.unwrap_or_else(|err| panic!("guard split directory {} must be listable: {err}", dir.display()));
+        let path = entry.path();
+        if path.is_dir() {
+            collect_rust_files(&path, out);
+        } else if path.extension().is_some_and(|extension| extension == "rs") {
+            out.push(path);
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::SourceSet;
+    use std::fs;
+    use std::path::Path;
+
+    fn write(root: &Path, relative: &str, text: &str) {
+        let path = root.join(relative);
+        fs::create_dir_all(path.parent().expect("fixture path has a parent")).expect("create fixture dir");
+        fs::write(path, text).expect("write fixture");
+    }
+
+    fn panic_message(run: impl FnOnce() + std::panic::UnwindSafe) -> String {
+        let payload = std::panic::catch_unwind(run).expect_err("the guard input must be rejected");
+        payload
+            .downcast_ref::<String>()
+            .cloned()
+            .or_else(|| payload.downcast_ref::<&str>().map(|message| (*message).to_owned()))
+            .unwrap_or_default()
+    }
+
+    #[test]
+    fn absent_split_directory_reads_only_the_required_file() {
+        let root = tempfile::tempdir().expect("tempdir");
+        write(root.path(), "src/app/usecase.rs", "guarded();\n");
+
+        let set = SourceSet::read_under(root.path(), "src/app/usecase.rs", "src/app/usecase");
+
+        assert_eq!(set.describe(), "src/app/usecase.rs");
+        assert!(set.contains("guarded();"));
+        assert_eq!(set.count("guarded();"), 1);
+    }
+
+    #[test]
+    fn split_directory_files_follow_the_required_file_in_path_order() {
+        let root = tempfile::tempdir().expect("tempdir");
+        write(root.path(), "src/app/usecase.rs", "guarded();\n");
+        write(root.path(), "src/app/usecase/zeta.rs", "guarded();\n");
+        write(root.path(), "src/app/usecase/alpha.rs", "guarded();\nguarded();\n");
+        write(root.path(), "src/app/usecase/tests/nested.rs", "guarded();\n");
+        write(root.path(), "src/app/usecase/notes.txt", "guarded();\n");
+
+        let set = SourceSet::read_under(root.path(), "src/app/usecase.rs", "src/app/usecase");
+
+        assert_eq!(
+            set.describe(),
+            "src/app/usecase.rs, src/app/usecase/alpha.rs, src/app/usecase/tests/nested.rs, src/app/usecase/zeta.rs"
+        );
+        assert_eq!(set.count("guarded();"), 5, "non-.rs files are not source");
+    }
+
+    #[test]
+    fn missing_required_file_fails_even_when_the_split_directory_exists() {
+        let root = tempfile::tempdir().expect("tempdir");
+        write(root.path(), "src/app/usecase/moved.rs", "guarded();\n");
+        let root_path = root.path().to_path_buf();
+
+        let message = panic_message(move || {
+            SourceSet::read_under(&root_path, "src/app/usecase.rs", "src/app/usecase");
+        });
+
+        assert!(message.contains("src/app/usecase.rs"), "panic must name the missing file: {message}");
+    }
+
+    #[test]
+    fn split_path_that_is_a_file_fails() {
+        let root = tempfile::tempdir().expect("tempdir");
+        write(root.path(), "src/app/usecase.rs", "guarded();\n");
+        write(root.path(), "src/app/usecase", "not a directory\n");
+        let root_path = root.path().to_path_buf();
+
+        let message = panic_message(move || {
+            SourceSet::read_under(&root_path, "src/app/usecase.rs", "src/app/usecase");
+        });
+
+        assert!(message.contains("must be a directory"), "{message}");
+    }
+
+    #[test]
+    fn count_does_not_join_a_needle_across_two_files() {
+        let root = tempfile::tempdir().expect("tempdir");
+        write(root.path(), "src/app/usecase.rs", "guard");
+        write(root.path(), "src/app/usecase/tail.rs", "ed();\n");
+
+        let set = SourceSet::read_under(root.path(), "src/app/usecase.rs", "src/app/usecase");
+
+        assert_eq!(set.count("guarded();"), 0);
+        assert!(!set.contains("guarded();"));
+    }
+
+    #[test]
+    fn unique_marker_absent_from_the_set_fails() {
+        let root = tempfile::tempdir().expect("tempdir");
+        write(root.path(), "src/app/usecase.rs", "fn other() {}\n");
+        let set = SourceSet::read_under(root.path(), "src/app/usecase.rs", "src/app/usecase");
+
+        let message = panic_message(move || {
+            set.after_unique_marker("fn guarded");
+        });
+
+        assert!(message.contains("exactly once"), "{message}");
+    }
+
+    #[test]
+    fn unique_marker_in_two_files_fails() {
+        let root = tempfile::tempdir().expect("tempdir");
+        write(root.path(), "src/app/usecase.rs", "fn guarded() {}\n");
+        write(root.path(), "src/app/usecase/copy.rs", "fn guarded() {}\n");
+        let set = SourceSet::read_under(root.path(), "src/app/usecase.rs", "src/app/usecase");
+
+        let message = panic_message(move || {
+            set.after_unique_marker("fn guarded");
+        });
+
+        assert!(message.contains("src/app/usecase/copy.rs"), "panic must name the set: {message}");
+    }
+
+    #[test]
+    fn unique_marker_twice_in_one_file_fails() {
+        let root = tempfile::tempdir().expect("tempdir");
+        write(root.path(), "src/app/usecase.rs", "fn guarded() {}\nfn guarded() {}\n");
+        let set = SourceSet::read_under(root.path(), "src/app/usecase.rs", "src/app/usecase");
+
+        let message = panic_message(move || {
+            set.after_unique_marker("fn guarded");
+        });
+
+        assert!(message.contains("exactly once"), "{message}");
+    }
+
+    #[test]
+    fn unique_marker_returns_the_rest_of_its_own_file_only() {
+        let root = tempfile::tempdir().expect("tempdir");
+        write(root.path(), "src/app/usecase.rs", "fn head() {}\n");
+        write(root.path(), "src/app/usecase/moved.rs", "fn guarded() { body(); }\n");
+        write(root.path(), "src/app/usecase/next.rs", "fn later() { other(); }\n");
+        let set = SourceSet::read_under(root.path(), "src/app/usecase.rs", "src/app/usecase");
+
+        let rest = set.after_unique_marker("fn guarded");
+
+        assert_eq!(rest, "() { body(); }\n");
+    }
+}

rustfs/src/lib.rs

@@ -89,6 +89,8 @@ pub mod diagnose;
 pub mod embedded;
 pub mod error;
 pub mod foreground_observation;
+#[cfg(test)]
+pub(crate) mod guard_sources;
 pub mod init;
 pub mod inspect;
 pub(crate) mod kms_deletion_gate;

rustfs/src/site_replication/tests.rs

@@ -1334,11 +1334,11 @@ fn test_delete_bucket_broadcast_fences_target_membership_through_delivery() {
         "a failed destructive delivery must remain visible for operator repair"
     );
 
-    let usecase = include_str!("../app/bucket_usecase.rs");
+    let usecase = crate::guard_sources::SourceSet::read("src/app/bucket_usecase.rs", "src/app/bucket_usecase");
     let delete = usecase
-        .split("async fn execute_delete_bucket_inner")
-        .nth(1)
-        .and_then(|rest| rest.split("pub async fn execute_head_bucket").next())
+        .after_unique_marker("async fn execute_delete_bucket_inner")
+        .split("pub async fn execute_head_bucket")
+        .next()
         .expect("delete bucket usecase");
     assert!(
         delete

rustfs/src/storage/ecfs.rs

@@ -2276,13 +2276,13 @@ impl S3 for FS {
 
 /// Every mutating S3 operation that has no bucket notification event reaches the audit
 /// stream only through `audit_only_operation`; a new such operation, or a dropped
-/// wrapper, must fail here rather than silently go unaudited.
+/// wrapper, must fail here rather than silently go unaudited. The wrappers are looked
+/// up in this file and in `legacy_edge/`, where a split of this file moves them.
 #[cfg(test)]
 mod audit_coverage_tests {
+    use crate::guard_sources::SourceSet;
     use rustfs_s3_ops::S3Operation;
 
-    const SOURCE: &str = include_str!("ecfs.rs");
-
     fn is_read_only(op: S3Operation) -> bool {
         let name = op.as_str().trim_start_matches("s3:");
         ["Get", "Head", "List", "Select"]
@@ -2300,9 +2300,15 @@ mod audit_coverage_tests {
         expected.push(S3Operation::AbortMultipartUpload);
         assert_eq!(expected.len(), 22, "an operation was added or removed: {expected:?}");
 
+        let source = SourceSet::read("src/storage/ecfs.rs", "src/storage/legacy_edge");
         for op in expected {
             let wrapper = format!("audit_only_operation(req, S3Operation::{op:?},");
-            assert!(SOURCE.contains(&wrapper), "{} is not run through audit_only_operation", op.as_str());
+            assert!(
+                source.contains(&wrapper),
+                "{} is not run through audit_only_operation in {}",
+                op.as_str(),
+                source.describe()
+            );
         }
     }
 }

rustfs/src/table_catalog/tests.rs

@@ -14478,31 +14478,31 @@ fn object_mutation_entrypoints_call_reserved_prefix_guard() {
 
 #[test]
 fn multipart_mutation_entrypoints_call_reserved_prefix_guard() {
-    let source = include_str!("../app/multipart_usecase.rs");
+    let source = crate::guard_sources::SourceSet::read("src/app/multipart_usecase.rs", "src/app/multipart_usecase");
 
     assert_eq!(
-        source
-            .matches("validate_table_catalog_object_mutation(&bucket, &key).await?;")
-            .count(),
-        4
+        source.count("validate_table_catalog_object_mutation(&bucket, &key).await?;"),
+        4,
+        "reserved-prefix guard calls across {}",
+        source.describe()
     );
 }
 
 #[test]
 fn object_metadata_mutation_entrypoints_call_reserved_prefix_guard() {
-    let source = include_str!("../storage/ecfs.rs");
+    let source = crate::guard_sources::SourceSet::read("src/storage/ecfs.rs", "src/storage/legacy_edge");
 
     assert_eq!(
-        source
-            .matches("validate_table_catalog_object_mutation(&bucket, &object).await?;")
-            .count(),
-        2
+        source.count("validate_table_catalog_object_mutation(&bucket, &object).await?;"),
+        2,
+        "reserved-prefix guard calls across {}",
+        source.describe()
     );
     assert_eq!(
-        source
-            .matches("validate_table_catalog_object_mutation(&bucket, &key).await?;")
-            .count(),
-        2
+        source.count("validate_table_catalog_object_mutation(&bucket, &key).await?;"),
+        2,
+        "reserved-prefix guard calls across {}",
+        source.describe()
     );
 }
 

scripts/README.md

@@ -38,6 +38,7 @@ their issue closes.
 | `check_no_tokio_io_uring.sh` | ci-gate | Keeps tokio's io-uring backend disabled | ci.yml Quick Checks |
 | `check_s3s_footprint.sh` | ci-gate | Exact, lower-only baseline on the direct s3s surface ahead of the gateway migration (reads `.config/s3s-footprint-baseline.txt`; `s3_error_lines` leaves out lines whose file imports the macro from `rustfs-s3-types`, rule in the script header); `--mode allowlist` fails on any Rust file referencing s3s outside `.config/s3s-edge-allowlist.txt`, `--dry-run` lists them | ci.yml Quick Checks; `make pre-commit`; `make script-tests` (`--self-test`) |
 | `check_test_count_ratchet.sh` | ci-gate | Per-crate test-count floor and `#[ignore]` ceiling against `.config/test-count-baseline.txt`; no crate may fall below its tests row or rise above its ignored row (extra tests and new crates pass with a note) | ci.yml Quick Checks; `make pre-commit` |
+| `guard_source_set.sh` | ci-gate (library) | Sourced helper that resolves a text guard's input as a required file plus the directory a mechanical split moves its code into (rustfs/backlog#2765); a missing required file fails | Sourced by `check_architecture_migration_rules.sh` and `check_logging_guardrails.sh`; contract tests `test_guard_source_set.sh` (`make script-tests`) |
 | `check_unsafe_code_allowances.sh` | ci-gate | Unsafe-code allowance ledger guard | ci.yml Quick Checks |
 | `layer-dependency-baseline.txt` | ci-gate (data) | Committed baseline consumed by `check_layer_dependencies.sh` | [Architecture guard troubleshooting](../docs/operations/architecture-guard-troubleshooting.md) |
 | `static.sh` | ci-gate | Static-build helper executed inside image builds | `Dockerfile.source`, `Dockerfile.decommission-local` |
@@ -167,7 +168,7 @@ their issue closes.
 
 | Entry | Status | Purpose | Wiring / docs |
 |---|---|---|---|
-| `codemods/` | dev-tool | Scripted, idempotent source migrations for the gateway migration (rustfs/backlog#2734); `s3_error_to_s3_types.py` moves s3s error-type paths onto `rustfs-s3-types` per crate closure and refuses the s3s edge allowlist | `make script-tests` (`--self-test`); usage in each script's docstring |
+| `codemods/` | dev-tool | Scripted, idempotent source migrations for the gateway migration (rustfs/backlog#2734); `s3_error_to_s3_types.py` moves s3s error-type paths onto `rustfs-s3-types` per crate closure and refuses the s3s edge allowlist; `split_bucket_usecase.py` splits `rustfs/src/app/bucket_usecase.rs` into one file per configuration family (preparatory refactor P2, rustfs/backlog#2765) and stops on any item its tables do not assign | `make script-tests` (`--self-test`); usage in each script's docstring |
 | `fuzz/` | ci-gate | Unified cargo-fuzz runner and helpers for the `fuzz/` sub-workspace | fuzz.yml; `fuzz/README.md` |
 | `release/` | ci-gate | Release creation and DEB/RPM version-normalization helpers | build.yml; package.yml; manual-main-packages.yml |
 | `s3-tests/` | ci-gate | ceph/s3-tests compatibility harness (allow-lists, patches, report tooling) | ci.yml; e2e-s3tests.yml; `scripts/s3-tests/README.md` |

scripts/check_architecture_migration_rules.sh

@@ -118,6 +118,28 @@ require_source_contains "docs/architecture/ecstore-api-facade-inventory.md" "Do
 TMP_DIR="$(mktemp -d)"
 trap 'rm -rf "$TMP_DIR"' EXIT
 
+# Rules about one of the large files below read that file plus the sibling
+# directory a mechanical split moves its code into (rustfs/backlog#2765), so a
+# split neither breaks nor silently blinds them. The *_SPLIT_SOURCES arrays hold
+# only the directory part, for rules that select the file itself by glob.
+# shellcheck source=scripts/guard_source_set.sh
+source "${ROOT_DIR}/scripts/guard_source_set.sh"
+guard_source_set "$ROOT_DIR" rustfs/src/storage/ecfs.rs rustfs/src/storage/legacy_edge ||
+  report_failure "ECFS guard source set is unreadable"
+ECFS_GUARD_SOURCES=("${GUARD_SOURCES[@]}")
+guard_source_set "$ROOT_DIR" rustfs/src/app/bucket_usecase.rs rustfs/src/app/bucket_usecase ||
+  report_failure "bucket usecase guard source set is unreadable"
+BUCKET_USECASE_GUARD_SOURCES=("${GUARD_SOURCES[@]}")
+BUCKET_USECASE_SPLIT_SOURCES=("${GUARD_SOURCES[@]:1}")
+guard_source_set "$ROOT_DIR" rustfs/src/app/multipart_usecase.rs rustfs/src/app/multipart_usecase ||
+  report_failure "multipart usecase guard source set is unreadable"
+MULTIPART_USECASE_GUARD_SOURCES=("${GUARD_SOURCES[@]}")
+MULTIPART_USECASE_SPLIT_SOURCES=("${GUARD_SOURCES[@]:1}")
+guard_source_set "$ROOT_DIR" rustfs/src/app/lifecycle_transition_api_test.rs rustfs/src/app/lifecycle_transition_api_test ||
+  report_failure "lifecycle transition API test guard source set is unreadable"
+LIFECYCLE_TRANSITION_TEST_GUARD_SOURCES=("${GUARD_SOURCES[@]}")
+LIFECYCLE_TRANSITION_TEST_SPLIT_SOURCES=("${GUARD_SOURCES[@]:1}")
+
 PR_TYPES_FILE="${TMP_DIR}/pr_types.txt"
 PR_TYPE_HITS_FILE="${TMP_DIR}/pr_type_hits.txt"
 SOURCE_MARKERS_FILE="${TMP_DIR}/source_markers.txt"
@@ -2157,7 +2179,7 @@ fi
   cd "$ROOT_DIR"
   rg -n --with-filename 'crate::app::context::(?:\{[^}]*\b(?:resolve_object_store_handle|resolve_buffer_config|resolve_internode_metrics|resolve_local_node_name|resolve_action_credentials|resolve_notify_interface|resolve_performance_metrics|resolve_encryption_service|resolve_region|resolve_ready_iam_handle)\b|(?:resolve_object_store_handle|resolve_buffer_config|resolve_internode_metrics|resolve_local_node_name|resolve_action_credentials|resolve_notify_interface|resolve_performance_metrics|resolve_encryption_service|resolve_region|resolve_ready_iam_handle)\b)' \
     rustfs/src/storage/access.rs \
-    rustfs/src/storage/ecfs.rs \
+    "${ECFS_GUARD_SOURCES[@]}" \
     rustfs/src/storage/ecfs_extend.rs \
     rustfs/src/storage/helper.rs \
     rustfs/src/storage/concurrency/manager.rs \
@@ -2175,7 +2197,7 @@ fi
 (
   cd "$ROOT_DIR"
   rg -n --with-filename 'crate::app::(?:bucket_usecase|multipart_usecase|object_usecase|object\b)|Default(?:Bucket|Multipart|Object)Usecase::from_global\(\)' \
-    rustfs/src/storage/ecfs.rs \
+    "${ECFS_GUARD_SOURCES[@]}" \
     --glob '*.rs' || true
 ) >"$RUSTFS_STORAGE_ECFS_USECASE_BYPASS_HITS_FILE"
 
@@ -2228,7 +2250,10 @@ fi
 
 (
   cd "$ROOT_DIR"
-  rg -n --with-filename 'use crate::storage::\*;' rustfs/src/app --glob '*_usecase.rs' --glob 'object/*.rs' || true
+  # A glob containing a slash is anchored at the search root, so object files
+  # need the leading **/ to match under rustfs/src/app.
+  rg -n --with-filename 'use crate::storage::\*;' rustfs/src/app --glob '*_usecase.rs' --glob '**/object/*.rs' \
+    "${BUCKET_USECASE_SPLIT_SOURCES[@]}" "${MULTIPART_USECASE_SPLIT_SOURCES[@]}" || true
 ) >"$RUSTFS_APP_USECASE_STORAGE_WILDCARD_HITS_FILE"
 
 if [[ -s "$RUSTFS_APP_USECASE_STORAGE_WILDCARD_HITS_FILE" ]]; then
@@ -2246,7 +2271,8 @@ fi
 
 (
   cd "$ROOT_DIR"
-  rg -n --with-filename 'crate::storage::s3_api::|use crate::storage::s3_api|super::s3_api::|use super::s3_api' rustfs/src/app --glob '*_usecase.rs' --glob 'object/*.rs' || true
+  rg -n --with-filename 'crate::storage::s3_api::|use crate::storage::s3_api|super::s3_api::|use super::s3_api' rustfs/src/app --glob '*_usecase.rs' --glob '**/object/*.rs' \
+    "${BUCKET_USECASE_SPLIT_SOURCES[@]}" "${MULTIPART_USECASE_SPLIT_SOURCES[@]}" || true
 ) >"$RUSTFS_APP_USECASE_S3_API_BYPASS_HITS_FILE"
 
 if [[ -s "$RUSTFS_APP_USECASE_S3_API_BYPASS_HITS_FILE" ]]; then
@@ -2258,13 +2284,14 @@ fi
   {
     rg -n --with-filename \
       '(use crate::storage::(access|helper|options|request_context|sse|timeout_wrapper|head_prefix|concurrency|ecfs)|crate::storage::sse::EncryptionKeyKind|use crate::storage::\{|use crate::storage::[A-Z])' \
-      rustfs/src/app/select_object.rs rustfs/src/app/*_usecase.rs rustfs/src/app/object || true
+      rustfs/src/app/select_object.rs rustfs/src/app/*_usecase.rs rustfs/src/app/object \
+      "${BUCKET_USECASE_SPLIT_SOURCES[@]}" "${MULTIPART_USECASE_SPLIT_SOURCES[@]}" || true
     rg -n --with-filename \
       'use super::(?:\{[^}]*\b(?:DynReader|HashReader|WriteEncryption|WritePlan|DecryptReader|EncryptReader|HardLimitReader|boxed_reader|wrap_reader|compression_metadata_value|is_disk_compressible|MIN_DISK_COMPRESSIBLE_SIZE|get_lock_acquire_timeout|is_valid_storage_class|StorageError|DiskError|is_all_buckets_not_found|is_err_bucket_not_found|is_err_object_not_found|is_err_version_not_found)\b|(?:object_api_utils::to_s3s_etag|storageclass|StorageError|DiskError|DynReader|HashReader|WriteEncryption|WritePlan|DecryptReader|EncryptReader|HardLimitReader|boxed_reader|wrap_reader|compression_metadata_value|is_disk_compressible|MIN_DISK_COMPRESSIBLE_SIZE|get_lock_acquire_timeout|is_valid_storage_class|is_all_buckets_not_found|is_err_bucket_not_found|is_err_object_not_found|is_err_version_not_found)\b)' \
-      rustfs/src/app/bucket_usecase.rs rustfs/src/app/object_usecase.rs rustfs/src/app/object rustfs/src/app/multipart_usecase.rs rustfs/src/app/lifecycle_transition_api_test.rs rustfs/src/app/lifecycle_transition_api_test || true
+      "${BUCKET_USECASE_GUARD_SOURCES[@]}" rustfs/src/app/object_usecase.rs rustfs/src/app/object "${MULTIPART_USECASE_GUARD_SOURCES[@]}" "${LIFECYCLE_TRANSITION_TEST_GUARD_SOURCES[@]}" || true
     rg -n --with-filename \
       'use super::(?:\{[^}]*\b(?:AppObjectLockConfigExt|AppReplicationConfigExt|AppVersioningConfigExt|predict_lifecycle_expiration|validate_restore_request|bucket_target_sys|lifecycle|metadata|metadata_sys|object_lock|policy_sys|quota|replication|tagging|target|utils|versioning_sys|transition_api|ObjectInfo|ObjectOptions)\b|(?:AppObjectLockConfigExt|AppReplicationConfigExt|AppVersioningConfigExt|predict_lifecycle_expiration|validate_restore_request|bucket_target_sys|lifecycle|metadata|metadata_sys|object_lock|policy_sys|quota|replication|tagging|target|utils|versioning_sys|transition_api|ObjectInfo|ObjectOptions)\b)|super::(?:lifecycle|metadata_sys|object_lock|quota|replication|tagging|target|utils|versioning_sys|transition_api)::|super::super::(?:metadata_sys|lifecycle|target)::' \
-      rustfs/src/app/bucket_usecase.rs rustfs/src/app/object_usecase.rs rustfs/src/app/object rustfs/src/app/multipart_usecase.rs rustfs/src/app/lifecycle_transition_api_test.rs rustfs/src/app/lifecycle_transition_api_test rustfs/src/app/capacity_dirty_scope_test.rs rustfs/src/app/context.rs rustfs/src/app/context/handles.rs rustfs/src/app/context/interfaces.rs rustfs/src/app/context/runtime_sources.rs || true
+      "${BUCKET_USECASE_GUARD_SOURCES[@]}" rustfs/src/app/object_usecase.rs rustfs/src/app/object "${MULTIPART_USECASE_GUARD_SOURCES[@]}" "${LIFECYCLE_TRANSITION_TEST_GUARD_SOURCES[@]}" rustfs/src/app/capacity_dirty_scope_test.rs rustfs/src/app/context.rs rustfs/src/app/context/handles.rs rustfs/src/app/context/interfaces.rs rustfs/src/app/context/runtime_sources.rs || true
   }
 ) >"$RUSTFS_APP_USECASE_STORAGE_API_BYPASS_HITS_FILE"
 
@@ -2277,6 +2304,7 @@ fi
   rg -n --with-filename \
     '(use crate::storage::(access|request_context|ecfs)|use crate::storage::\{[^}]*Storage(ObjectInfo|ObjectOptions|PutObjReader)|crate::storage::StorageObjectOptions)' \
     rustfs/src/app/*_test.rs \
+    "${LIFECYCLE_TRANSITION_TEST_SPLIT_SOURCES[@]}" \
     rustfs/src/admin/router.rs \
     rustfs/src/admin/console.rs \
     rustfs/src/admin/handlers/heal.rs \
@@ -4561,7 +4589,7 @@ fi
   cd "$ROOT_DIR"
   rg -n --with-filename 'crate::storage::core_storage_compat' \
     rustfs/src/storage/mod.rs \
-    rustfs/src/storage/ecfs.rs \
+    "${ECFS_GUARD_SOURCES[@]}" \
     rustfs/src/storage/ecfs_extend.rs \
     rustfs/src/storage/access.rs \
     rustfs/src/storage/head_prefix.rs \
@@ -4577,7 +4605,8 @@ fi
   cd "$ROOT_DIR"
   {
     rg -n --with-filename 'crate::app::usecase_storage_compat' \
-      rustfs/src/app/*.rs || true
+      rustfs/src/app/*.rs "${BUCKET_USECASE_SPLIT_SOURCES[@]}" "${MULTIPART_USECASE_SPLIT_SOURCES[@]}" \
+      "${LIFECYCLE_TRANSITION_TEST_SPLIT_SOURCES[@]}" || true
     rg -n --with-filename 'crate::admin::router_storage_compat' \
       rustfs/src/admin/router.rs || true
     rg -n --with-filename 'crate::storage::core_storage_compat' \

scripts/check_layer_dependencies.sh

@@ -22,7 +22,8 @@ classify_source_layer() {
     printf 'composition'
   elif [[ "$file" == rustfs/src/app/* ]]; then
     printf 'app'
-  elif [[ "$file" == rustfs/src/admin/* ]] || [[ "$file" == rustfs/src/storage/ecfs.rs ]] || [[ "$file" == rustfs/src/storage/s3_api/* ]] ||
+  elif [[ "$file" == rustfs/src/admin/* ]] || [[ "$file" == rustfs/src/storage/ecfs.rs ]] ||
+    [[ "$file" == rustfs/src/storage/legacy_edge/* ]] || [[ "$file" == rustfs/src/storage/s3_api/* ]] ||
     [[ "$file" == rustfs/src/storage/s3_compat/* ]]; then
     printf 'interface'
   elif [[ "$file" == rustfs/src/* ]]; then
@@ -55,6 +56,7 @@ classify_target_layer() {
     storage)
       storage_path="${path#storage::}"
       if [[ "$storage_path" == "ecfs" ]] || [[ "$storage_path" == ecfs::* ]] ||
+        [[ "$storage_path" == "legacy_edge" ]] || [[ "$storage_path" == legacy_edge::* ]] ||
         [[ "$storage_path" == "s3_api" ]] || [[ "$storage_path" == s3_api::* ]] ||
         [[ "$storage_path" == "s3_compat" ]] || [[ "$storage_path" == s3_compat::* ]]; then
         printf 'interface'
@@ -137,6 +139,60 @@ run_layer_model_self_tests() {
   assert_dependency_direction 'allowed' "$s3_compat_source" "$app_target"
   assert_dependency_direction 'reverse' "$app_source" "$s3_compat_target"
   assert_dependency_direction 'reverse' "$storage_source" "$s3_compat_target"
+
+  # A mechanical split moves ecfs.rs code into storage/legacy_edge/ and the
+  # usecase files' code into same-named directories; each keeps its layer.
+  assert_source_layer 'interface' rustfs/src/storage/ecfs.rs
+  assert_source_layer 'interface' rustfs/src/storage/legacy_edge/bucket_core.rs
+  assert_source_layer 'interface' rustfs/src/storage/legacy_edge/tests/object_meta.rs
+  assert_source_layer 'infra' rustfs/src/storage/legacy_edge.rs
+  assert_source_layer 'infra' rustfs/src/storage/legacy_edge_helper.rs
+  assert_source_layer 'app' rustfs/src/app/bucket_usecase/cors.rs
+  assert_source_layer 'app' rustfs/src/app/multipart_usecase/complete.rs
+  assert_target_layer 'interface' storage::legacy_edge
+  assert_target_layer 'interface' storage::legacy_edge::object_meta
+  assert_target_layer 'infra' storage::legacy_edge_helper
+  assert_dependency_direction 'allowed' "$(classify_source_layer rustfs/src/storage/legacy_edge/bucket_core.rs)" "$app_target"
+  assert_dependency_direction 'reverse' "$app_source" "$(classify_target_layer storage::legacy_edge::bucket_core)"
+  assert_dependency_direction 'reverse' "$storage_source" "$(classify_target_layer storage::legacy_edge)"
+  assert_dedicated_test_file 'yes' rustfs/src/app/lifecycle_transition_api_test.rs
+  assert_dedicated_test_file 'yes' rustfs/src/app/lifecycle_transition_api_test/put.rs
+  assert_dedicated_test_file 'no' rustfs/src/app/lifecycle_transition_api_test_support.rs
+  assert_dedicated_test_file 'no' rustfs/src/app/lifecycle_transition_api.rs
+  assert_dedicated_test_file 'no' rustfs/src/app/bucket_usecase/cors.rs
+  assert_dedicated_test_file 'no' rustfs/src/storage/legacy_edge/bucket_core.rs
+}
+
+assert_source_layer() {
+  local expected="$1" file="$2" actual
+
+  actual="$(classify_source_layer "$file")"
+  if [[ "$actual" != "$expected" ]]; then
+    printf 'Layer dependency guard self-test failed: %s classified as %s (expected %s)\n' "$file" "$actual" "$expected" >&2
+    exit 1
+  fi
+}
+
+assert_target_layer() {
+  local expected="$1" path="$2" actual
+
+  actual="$(classify_target_layer "$path")"
+  if [[ "$actual" != "$expected" ]]; then
+    printf 'Layer dependency guard self-test failed: crate::%s classified as %s (expected %s)\n' "$path" "$actual" "$expected" >&2
+    exit 1
+  fi
+}
+
+assert_dedicated_test_file() {
+  local expected="$1" file="$2" actual='no'
+
+  if is_dedicated_test_file "$file"; then
+    actual='yes'
+  fi
+  if [[ "$actual" != "$expected" ]]; then
+    printf 'Layer dependency guard self-test failed: %s dedicated test file = %s (expected %s)\n' "$file" "$actual" "$expected" >&2
+    exit 1
+  fi
 }
 
 normalize_import_group_item() {
@@ -251,11 +307,20 @@ normalize_import_path() {
   printf '%s\n' "$path"
 }
 
+# The guard is file-scoped: dedicated test modules are excluded, while inline
+# #[cfg(test)] imports remain subject to the source file's layer. The
+# lifecycle_transition_api_test/ directory is where a mechanical split of
+# lifecycle_transition_api_test.rs moves its tests (rustfs/backlog#2765).
+is_dedicated_test_file() {
+  local file="$1"
+
+  [[ "$file" == *_test.rs ]] || [[ "$file" == */tests/* ]] ||
+    [[ "$file" == rustfs/src/app/lifecycle_transition_api_test/* ]]
+}
+
 emit_crate_use_statements() {
   (cd "$ROOT_DIR" && rg --files -g '*.rs' rustfs/src | while IFS= read -r file; do
-    # The guard is file-scoped: dedicated test modules are excluded, while
-    # inline #[cfg(test)] imports remain subject to the source file's layer.
-    if [[ "$file" == *_test.rs ]] || [[ "$file" == */tests/* ]]; then
+    if is_dedicated_test_file "$file"; then
       continue
     fi
     perl -0777 -ne '

scripts/check_logging_guardrails.sh

@@ -1094,11 +1094,29 @@ trace_hot_spans=(
   "rustfs/src/app/object:execute_get_object"
 )
 
+# A hot span in a file that is due for a mechanical split (rustfs/backlog#2765)
+# is also looked up in the directory the split moves its functions into; the
+# named file itself stays required.
+# shellcheck source=scripts/guard_source_set.sh
+source "${repo_root}/scripts/guard_source_set.sh"
+
 for hot_span in "${trace_hot_spans[@]}"; do
   file="${hot_span%%:*}"
   function="${hot_span##*:}"
+  hot_span_sources=("$file")
+  split_dir=""
+  case "$file" in
+    rustfs/src/app/bucket_usecase.rs) split_dir="rustfs/src/app/bucket_usecase" ;;
+  esac
+  if [[ -n "$split_dir" ]]; then
+    if ! guard_source_set "$repo_root" "$file" "$split_dir"; then
+      echo "❌ logging guardrail violation: $file::$function guard source is missing" >&2
+      exit 1
+    fi
+    hot_span_sources=("${GUARD_SOURCES[@]}")
+  fi
   trace_span_pattern="#\\[(tracing::)?instrument\\([^]]*level = \\\"trace\\\"[^]]*\\)\\]([[:space:]]+#\\[[^]]+\\])*[[:space:]]+(pub(\\([^)]*\\))?[[:space:]]+)?(super[[:space:]]+)?async fn ${function}\\b"
-  if ! rg -U "$trace_span_pattern" "$file" >/dev/null; then
+  if ! rg -U "$trace_span_pattern" "${hot_span_sources[@]}" >/dev/null; then
     echo "❌ logging guardrail violation: $file::$function must remain TRACE-only" >&2
     exit 1
   fi

scripts/guard_source_set.sh

@@ -0,0 +1,51 @@
+#!/usr/bin/env bash
+#
+# Source-set helper for text guards that must survive a mechanical file split.
+# Sourced by check_architecture_migration_rules.sh and check_logging_guardrails.sh;
+# contract tests: scripts/test_guard_source_set.sh.
+#
+# A guard that greps one named file stops seeing the code it guards once that
+# file is split into a sibling directory. guard_source_set resolves the set of
+# files such a guard is about instead:
+#
+#   guard_source_set <root> <required-file> [<split-dir>...]
+#
+# On success it sets the global array GUARD_SOURCES to <required-file> followed
+# by every `.rs` file under each <split-dir> that exists, in byte order of path
+# (LC_ALL=C), all relative to <root>. An absent <split-dir> contributes nothing.
+# It returns 1 with a message on stderr when <required-file> is not a regular
+# file (a guard whose input is missing must fail, not pass) or when a
+# <split-dir> exists but is not a directory. On failure GUARD_SOURCES still
+# holds <required-file>, so a caller that records the failure and continues
+# hands the missing path to its search tool rather than nothing.
+
+guard_source_set() {
+  local root="$1" required="$2" split_dir path listing
+  shift 2
+  GUARD_SOURCES=("$required")
+
+  if [[ ! -f "${root}/${required}" ]]; then
+    printf 'guard source %s is missing\n' "$required" >&2
+    return 1
+  fi
+
+  for split_dir in "$@"; do
+    if [[ ! -e "${root}/${split_dir}" ]]; then
+      continue
+    fi
+    if [[ ! -d "${root}/${split_dir}" ]]; then
+      printf 'guard split path %s is not a directory\n' "$split_dir" >&2
+      return 1
+    fi
+    if ! listing="$(set -o pipefail; cd "$root" && find "$split_dir" -type f -name '*.rs' | LC_ALL=C sort)"; then
+      printf 'guard split directory %s cannot be listed\n' "$split_dir" >&2
+      return 1
+    fi
+    while IFS= read -r path; do
+      if [[ -n "$path" ]]; then
+        GUARD_SOURCES+=("$path")
+      fi
+    done <<<"$listing"
+  done
+  return 0
+}

scripts/s3-tests/README.md

@@ -509,7 +509,9 @@ pass/fail table in the job summary.
   incomplete parameterized cases.
 - `api_coverage.py` — quantifies S3 API surface coverage by comparing the
   s3s `S3` trait (at the revision pinned in Cargo.toml) against the methods
-  RustFS overrides in `impl S3 for FS`:
+  RustFS overrides in `impl S3 for FS`, found exactly once across
+  `rustfs/src/storage/ecfs.rs` and `rustfs/src/storage/legacy_edge/`
+  (contract tests: `test_api_coverage.py`, run by `make script-tests`):
 
   ```bash
   python3 scripts/s3-tests/api_coverage.py            # summary + missing ops

scripts/s3-tests/api_coverage.py

@@ -17,7 +17,8 @@
 
 Compares the operations defined by the s3s `S3` trait (at the revision pinned
 in the workspace Cargo.toml) against the methods RustFS overrides in
-`impl S3 for FS` (rustfs/src/storage/ecfs.rs). Trait methods that are not
+`impl S3 for FS` (rustfs/src/storage/ecfs.rs, or the rustfs/src/storage/legacy_edge/
+directory a mechanical split of that file moves code into). Trait methods that are not
 overridden fall back to the s3s default implementation, which returns
 NotImplemented — so "overridden" is a faithful proxy for "implemented".
 
@@ -36,6 +37,8 @@ import sys
 import tempfile
 
 ECFS_PATH = "rustfs/src/storage/ecfs.rs"
+ECFS_SPLIT_DIR = "rustfs/src/storage/legacy_edge"
+IMPL_MARKER = "impl S3 for FS {"
 TRAIT_RELPATH = "crates/s3s/src/s3_trait.rs"
 
 
@@ -97,14 +100,28 @@ def trait_operations(trait_source: pathlib.Path) -> dict[str, str]:
     return ops
 
 
-def implemented_operations(ecfs_source: pathlib.Path) -> set[str]:
-    src = ecfs_source.read_text(encoding="utf-8")
-    marker = "impl S3 for FS {"
-    start = src.find(marker)
-    if start < 0:
-        raise RuntimeError(f"`{marker}` not found in {ecfs_source}")
-    block = extract_block(src, start)
-    return set(re.findall(r"async fn (\w+)", block))
+def ecfs_sources(root: pathlib.Path) -> list[pathlib.Path]:
+    """Return ecfs.rs (required) followed by every .rs file under the split directory, in path order."""
+    required = root / ECFS_PATH
+    if not required.is_file():
+        raise RuntimeError(f"{required} is missing")
+    split_dir = root / ECFS_SPLIT_DIR
+    if not split_dir.exists():
+        return [required]
+    if not split_dir.is_dir():
+        raise RuntimeError(f"{split_dir} is not a directory")
+    return [required, *sorted(split_dir.rglob("*.rs"), key=lambda path: path.relative_to(root).as_posix())]
+
+
+def implemented_operations(root: pathlib.Path) -> set[str]:
+    """Return the methods of the one `impl S3 for FS` block across the ECFS source set."""
+    blocks = []
+    for source in ecfs_sources(root):
+        src = source.read_text(encoding="utf-8")
+        blocks += [extract_block(src, match.start()) for match in re.finditer(re.escape(IMPL_MARKER), src)]
+    if len(blocks) != 1:
+        raise RuntimeError(f"`{IMPL_MARKER}` must occur exactly once under {ECFS_PATH} and {ECFS_SPLIT_DIR}, found {len(blocks)}")
+    return set(re.findall(r"async fn (\w+)", blocks[0]))
 
 
 def main() -> int:
@@ -122,7 +139,7 @@ def main() -> int:
     trait_source = args.s3s_trait or locate_trait_source(git_url, rev)
 
     all_ops = trait_operations(trait_source)
-    impl_ops = implemented_operations(root / ECFS_PATH)
+    impl_ops = implemented_operations(root)
 
     stale = sorted(impl_ops - all_ops.keys())
     implemented = sorted(op for op in all_ops if op in impl_ops)

scripts/s3-tests/test_api_coverage.py

@@ -0,0 +1,93 @@
+#!/usr/bin/env python3
+# Copyright 2024 RustFS Team
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Contract tests for how api_coverage.py finds `impl S3 for FS` across the ECFS source set."""
+
+from __future__ import annotations
+
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+
+import api_coverage  # noqa: E402
+
+IMPL = "impl S3 for FS {\n    async fn get_object(&self) {}\n    async fn put_object(&self) {}\n}\n"
+
+
+class ImplementedOperationsTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.root = Path(self._tmp.name)
+
+    def tearDown(self) -> None:
+        self._tmp.cleanup()
+
+    def write(self, relative: str, text: str) -> None:
+        path = self.root / relative
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(text, encoding="utf-8")
+
+    def test_reads_the_impl_from_ecfs_when_the_split_directory_is_absent(self) -> None:
+        self.write(api_coverage.ECFS_PATH, "struct FS;\n" + IMPL)
+
+        self.assertEqual(api_coverage.implemented_operations(self.root), {"get_object", "put_object"})
+
+    def test_finds_the_impl_after_it_moves_into_the_split_directory(self) -> None:
+        self.write(api_coverage.ECFS_PATH, "struct FS;\n")
+        self.write(f"{api_coverage.ECFS_SPLIT_DIR}/s3_impl.rs", IMPL)
+        self.write(f"{api_coverage.ECFS_SPLIT_DIR}/bucket_core.rs", "impl FS {\n    async fn helper(&self) {}\n}\n")
+
+        self.assertEqual(api_coverage.implemented_operations(self.root), {"get_object", "put_object"})
+
+    def test_missing_ecfs_fails_even_when_the_split_directory_holds_the_impl(self) -> None:
+        self.write(f"{api_coverage.ECFS_SPLIT_DIR}/s3_impl.rs", IMPL)
+
+        with self.assertRaisesRegex(RuntimeError, "ecfs.rs is missing"):
+            api_coverage.implemented_operations(self.root)
+
+    def test_an_impl_in_both_ecfs_and_the_split_directory_fails(self) -> None:
+        self.write(api_coverage.ECFS_PATH, IMPL)
+        self.write(f"{api_coverage.ECFS_SPLIT_DIR}/s3_impl.rs", IMPL)
+
+        with self.assertRaisesRegex(RuntimeError, "exactly once.*found 2"):
+            api_coverage.implemented_operations(self.root)
+
+    def test_no_impl_anywhere_fails(self) -> None:
+        self.write(api_coverage.ECFS_PATH, "struct FS;\n")
+        self.write(f"{api_coverage.ECFS_SPLIT_DIR}/bucket_core.rs", "impl FS {}\n")
+
+        with self.assertRaisesRegex(RuntimeError, "found 0"):
+            api_coverage.implemented_operations(self.root)
+
+    def test_an_impl_in_a_non_rust_file_of_the_split_directory_is_not_source(self) -> None:
+        self.write(api_coverage.ECFS_PATH, "struct FS;\n")
+        self.write(f"{api_coverage.ECFS_SPLIT_DIR}/notes.md", IMPL)
+
+        with self.assertRaisesRegex(RuntimeError, "found 0"):
+            api_coverage.implemented_operations(self.root)
+
+    def test_a_split_path_that_is_a_file_fails(self) -> None:
+        self.write(api_coverage.ECFS_PATH, IMPL)
+        self.write(api_coverage.ECFS_SPLIT_DIR, "stray\n")
+
+        with self.assertRaisesRegex(RuntimeError, "not a directory"):
+            api_coverage.implemented_operations(self.root)
+
+
+if __name__ == "__main__":
+    unittest.main()

scripts/test_guard_source_set.sh

@@ -0,0 +1,153 @@
+#!/usr/bin/env bash
+# Contract tests for scripts/guard_source_set.sh, the source-set helper that
+# lets text guards read a file plus the directory a mechanical split moves its
+# code into (rustfs/backlog#2765).
+#
+# Each case builds a throwaway tree; nothing in the real workspace is read.
+
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
+# shellcheck source=scripts/guard_source_set.sh
+source "${GUARD_SOURCE_SET_SCRIPT:-$SCRIPT_DIR/guard_source_set.sh}"
+TMP_ROOT="$(mktemp -d)"
+trap 'chmod -R u+rwx "$TMP_ROOT" 2>/dev/null || true; rm -rf "$TMP_ROOT"' EXIT
+
+passed=0
+skipped=0
+case_no=0
+FX=""
+STATUS=0
+ERR=""
+
+fail() {
+  echo "FAIL: $1" >&2
+  exit 1
+}
+
+new_case() {
+  case_no=$((case_no + 1))
+  FX="$TMP_ROOT/case$case_no"
+  mkdir -p "$FX"
+}
+
+put() {
+  mkdir -p "$(dirname "$FX/$1")"
+  printf 'fn guarded() {}\n' >"$FX/$1"
+}
+
+# run_set <args...>: call guard_source_set on the case tree, capturing status and stderr.
+run_set() {
+  local err_file="$FX/.stderr"
+  STATUS=0
+  guard_source_set "$FX" "$@" 2>"$err_file" || STATUS=$?
+  ERR="$(cat "$err_file")"
+}
+
+sources_are() {
+  local expected actual
+  expected="$(printf '%s\n' "$@")"
+  actual="$(printf '%s\n' "${GUARD_SOURCES[@]}")"
+  [[ "$actual" == "$expected" ]] || fail "case $case_no: sources were [$actual], expected [$expected]"
+}
+
+ok() {
+  passed=$((passed + 1))
+}
+
+# --- positive cases ---------------------------------------------------------
+
+# An absent split directory contributes nothing.
+new_case
+put src/usecase.rs
+run_set src/usecase.rs src/usecase
+[[ "$STATUS" == 0 ]] || fail "absent split dir: status $STATUS ($ERR)"
+sources_are src/usecase.rs
+ok
+
+# A present split directory adds its .rs files, recursively, in byte order,
+# after the required file; non-.rs files are not source.
+new_case
+put src/usecase.rs
+put src/usecase/zeta.rs
+put src/usecase/B.rs
+put src/usecase/a.rs
+put src/usecase/tests/nested.rs
+printf 'not rust\n' >"$FX/src/usecase/notes.md"
+run_set src/usecase.rs src/usecase
+[[ "$STATUS" == 0 ]] || fail "present split dir: status $STATUS ($ERR)"
+sources_are src/usecase.rs src/usecase/B.rs src/usecase/a.rs src/usecase/tests/nested.rs src/usecase/zeta.rs
+ok
+
+# Several split directories are read in argument order; an empty one adds nothing.
+new_case
+put src/ecfs.rs
+put src/legacy_edge/b.rs
+mkdir -p "$FX/src/empty"
+put src/other/a.rs
+run_set src/ecfs.rs src/legacy_edge src/empty src/missing src/other
+[[ "$STATUS" == 0 ]] || fail "several split dirs: status $STATUS ($ERR)"
+sources_are src/ecfs.rs src/legacy_edge/b.rs src/other/a.rs
+ok
+
+# --- negative cases ---------------------------------------------------------
+
+# A missing required file fails and names the file.
+new_case
+run_set src/usecase.rs src/usecase
+[[ "$STATUS" == 1 ]] || fail "missing required file: status $STATUS"
+[[ "$ERR" == *"src/usecase.rs"* ]] || fail "missing required file: stderr does not name it: $ERR"
+sources_are src/usecase.rs
+ok
+
+# A missing required file fails even when the split directory holds the code:
+# the split must keep the file, and a guard must not pass on half its input.
+new_case
+put src/usecase/moved.rs
+run_set src/usecase.rs src/usecase
+[[ "$STATUS" == 1 ]] || fail "missing required file with split dir: status $STATUS"
+[[ "$ERR" == *"is missing"* ]] || fail "missing required file with split dir: $ERR"
+ok
+
+# A required path that is a directory is not a source file.
+new_case
+mkdir -p "$FX/src/usecase.rs"
+run_set src/usecase.rs src/usecase
+[[ "$STATUS" == 1 ]] || fail "required directory: status $STATUS"
+ok
+
+# A split path that exists as a regular file fails rather than being skipped.
+new_case
+put src/usecase.rs
+printf 'stray\n' >"$FX/src/usecase"
+run_set src/usecase.rs src/usecase
+[[ "$STATUS" == 1 ]] || fail "split path is a file: status $STATUS"
+[[ "$ERR" == *"not a directory"* ]] || fail "split path is a file: $ERR"
+ok
+
+# A later split path that is a file still fails after an earlier valid one.
+new_case
+put src/usecase.rs
+put src/usecase/a.rs
+printf 'stray\n' >"$FX/src/second"
+run_set src/usecase.rs src/usecase src/second
+[[ "$STATUS" == 1 ]] || fail "second split path is a file: status $STATUS"
+ok
+
+# An unlistable split directory fails rather than contributing nothing.
+new_case
+put src/usecase.rs
+put src/usecase/hidden/a.rs
+chmod 000 "$FX/src/usecase/hidden"
+if [[ "$(id -u)" == 0 ]]; then
+  echo "SKIP: unlistable split directory case needs a non-root user (root ignores directory permissions)"
+  skipped=$((skipped + 1))
+else
+  run_set src/usecase.rs src/usecase
+  [[ "$STATUS" == 1 ]] || fail "unlistable split dir: status $STATUS"
+  [[ "$ERR" == *"cannot be listed"* ]] || fail "unlistable split dir: $ERR"
+  ok
+fi
+chmod 755 "$FX/src/usecase/hidden"
+
+echo "guard_source_set contract tests: $passed passed, $skipped skipped"