From 74171bd673e868d93023a6c1200b3f4b411b3c11 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Sun, 23 Aug 2026 10:13:59 +0800 Subject: [PATCH] fix(replication): pass site peer ids into the bucket usecase from the interface layer The review fix made the bucket usecase read the site-replication peer set through the admin handlers, an app->interface import the layer guard rejects. The S3 handlers (interface) now read the peer set and pass it in, so the usecase stays a pure function of its inputs; a state-read failure still fails the edit closed, just one layer up. --- rustfs/src/app/bucket_usecase.rs | 20 ++++++++++++++------ rustfs/src/storage/ecfs.rs | 21 +++++++++++++++++++-- 2 files changed, 33 insertions(+), 8 deletions(-) diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index 65bcd5206..415666e91 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -66,7 +66,6 @@ use super::storage_api::bucket_usecase::{ }; use crate::admin::handlers::site_replication::{ site_replication_bucket_meta_hook, site_replication_delete_bucket_hook, site_replication_make_bucket_hook, - site_replication_remote_peer_deployment_ids, }; use crate::app::object_data_cache::invalidate_object_data_cache_bucket_after_delete; use crate::app::runtime_sources::{ @@ -1624,9 +1623,15 @@ impl DefaultBucketUsecase { Ok(S3Response::new(DeleteBucketPolicyOutput {})) } + /// `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, + site_peers: HashSet, ) -> S3Result> { let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?; let request_context = req.extensions.get::().cloned(); @@ -1647,7 +1652,6 @@ impl DefaultBucketUsecase { Err(err) => return Err(ApiError::from(err).into()), }; let (remaining_config, updated_targets) = if let Some(config) = replication_config.as_ref() { - let site_peers = site_replication_remote_peer_deployment_ids().await?; let (remaining, removable_arns) = split_replication_config_for_user_delete(config.clone(), &site_peers); let targets = replication_targets_without_arns(&bucket, &removable_arns).await?; (remaining, targets) @@ -2516,9 +2520,11 @@ impl DefaultBucketUsecase { Ok(S3Response::new(PutBucketCorsOutput::default())) } + /// See [`Self::execute_delete_bucket_replication`] for `site_peers`. pub async fn execute_put_bucket_replication( &self, req: S3Request, + site_peers: HashSet, ) -> S3Result> { let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?; let request_context = req.extensions.get::().cloned(); @@ -2547,7 +2553,6 @@ impl DefaultBucketUsecase { Err(StorageError::ConfigNotFound) => None, Err(err) => return Err(ApiError::from(err).into()), }; - let site_peers = site_replication_remote_peer_deployment_ids().await?; let replication_configuration = merge_user_replication_config_update(replication_configuration, existing_config, &site_peers); let data = serialize_config(&replication_configuration)?; @@ -3695,7 +3700,10 @@ mod tests { let req = build_request(input, Method::DELETE); let usecase = DefaultBucketUsecase::without_context(); - let err = usecase.execute_delete_bucket_replication(req).await.unwrap_err(); + let err = usecase + .execute_delete_bucket_replication(req, HashSet::new()) + .await + .unwrap_err(); assert_eq!(err.code(), &S3ErrorCode::InternalError); } @@ -4781,7 +4789,7 @@ mod tests { let req = build_request(input, Method::PUT); let usecase = DefaultBucketUsecase::without_context(); - let err = usecase.execute_put_bucket_replication(req).await.unwrap_err(); + let err = usecase.execute_put_bucket_replication(req, HashSet::new()).await.unwrap_err(); assert_eq!(err.code(), &S3ErrorCode::InternalError); } @@ -4799,7 +4807,7 @@ mod tests { .unwrap(); let err = DefaultBucketUsecase::without_context() - .execute_put_bucket_replication(build_request(input, Method::PUT)) + .execute_put_bucket_replication(build_request(input, Method::PUT), HashSet::new()) .await .expect_err("unsupported fields must be rejected before store access"); diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index 0d2db854d..f2e258c27 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -80,6 +80,21 @@ async fn site_replication_gate_enabled() -> S3Result { crate::admin::handlers::site_replication::site_replication_enabled().await } +/// Remote site-replication peer deployment ids handed to the bucket usecase +/// so an S3 replication-config edit keeps exactly the reconciler-owned rules +/// (issue #1948). Read here, in the interface layer, because the usecase must +/// not import the admin handlers (layer guard); a state-read failure +/// propagates so the edit fails closed. +async fn site_replication_peer_deployment_ids_for_edit() -> S3Result> { + // While the gate override is in effect the test exercises the deny/allow + // branch, not the peer set; there is no persisted state to read. + #[cfg(test)] + if SITE_REPLICATION_GATE_TEST_OVERRIDE.load(std::sync::atomic::Ordering::SeqCst) != 0 { + return Ok(std::collections::HashSet::new()); + } + crate::admin::handlers::site_replication::site_replication_remote_peer_deployment_ids().await +} + /// MinIO `ErrReplicationDenyEditError`. fn replication_deny_edit_error() -> S3Error { let mut err = S3Error::with_message( @@ -549,8 +564,9 @@ impl S3 for FS { req: S3Request, ) -> S3Result> { deny_replication_config_edit_for_non_owner(&req).await?; + let site_peers = site_replication_peer_deployment_ids_for_edit().await?; let usecase = s3_api::bucket_usecase_for(self); - usecase.execute_delete_bucket_replication(req).await + usecase.execute_delete_bucket_replication(req, site_peers).await } #[instrument(level = "debug", skip(self))] @@ -1403,8 +1419,9 @@ impl S3 for FS { req: S3Request, ) -> S3Result> { deny_replication_config_edit_for_non_owner(&req).await?; + let site_peers = site_replication_peer_deployment_ids_for_edit().await?; let usecase = s3_api::bucket_usecase_for(self); - usecase.execute_put_bucket_replication(req).await + usecase.execute_put_bucket_replication(req, site_peers).await } async fn put_bucket_request_payment(