mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-26 05:56:50 +00:00
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.
This commit is contained in:
@@ -66,7 +66,6 @@ use super::storage_api::bucket_usecase::{
|
|||||||
};
|
};
|
||||||
use crate::admin::handlers::site_replication::{
|
use crate::admin::handlers::site_replication::{
|
||||||
site_replication_bucket_meta_hook, site_replication_delete_bucket_hook, site_replication_make_bucket_hook,
|
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::object_data_cache::invalidate_object_data_cache_bucket_after_delete;
|
||||||
use crate::app::runtime_sources::{
|
use crate::app::runtime_sources::{
|
||||||
@@ -1624,9 +1623,15 @@ impl DefaultBucketUsecase {
|
|||||||
Ok(S3Response::new(DeleteBucketPolicyOutput {}))
|
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(
|
pub async fn execute_delete_bucket_replication(
|
||||||
&self,
|
&self,
|
||||||
req: S3Request<DeleteBucketReplicationInput>,
|
req: S3Request<DeleteBucketReplicationInput>,
|
||||||
|
site_peers: HashSet<String>,
|
||||||
) -> S3Result<S3Response<DeleteBucketReplicationOutput>> {
|
) -> S3Result<S3Response<DeleteBucketReplicationOutput>> {
|
||||||
let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
|
let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
|
||||||
let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
|
let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
|
||||||
@@ -1647,7 +1652,6 @@ impl DefaultBucketUsecase {
|
|||||||
Err(err) => return Err(ApiError::from(err).into()),
|
Err(err) => return Err(ApiError::from(err).into()),
|
||||||
};
|
};
|
||||||
let (remaining_config, updated_targets) = if let Some(config) = replication_config.as_ref() {
|
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 (remaining, removable_arns) = split_replication_config_for_user_delete(config.clone(), &site_peers);
|
||||||
let targets = replication_targets_without_arns(&bucket, &removable_arns).await?;
|
let targets = replication_targets_without_arns(&bucket, &removable_arns).await?;
|
||||||
(remaining, targets)
|
(remaining, targets)
|
||||||
@@ -2516,9 +2520,11 @@ impl DefaultBucketUsecase {
|
|||||||
Ok(S3Response::new(PutBucketCorsOutput::default()))
|
Ok(S3Response::new(PutBucketCorsOutput::default()))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// See [`Self::execute_delete_bucket_replication`] for `site_peers`.
|
||||||
pub async fn execute_put_bucket_replication(
|
pub async fn execute_put_bucket_replication(
|
||||||
&self,
|
&self,
|
||||||
req: S3Request<PutBucketReplicationInput>,
|
req: S3Request<PutBucketReplicationInput>,
|
||||||
|
site_peers: HashSet<String>,
|
||||||
) -> S3Result<S3Response<PutBucketReplicationOutput>> {
|
) -> S3Result<S3Response<PutBucketReplicationOutput>> {
|
||||||
let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
|
let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
|
||||||
let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
|
let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
|
||||||
@@ -2547,7 +2553,6 @@ impl DefaultBucketUsecase {
|
|||||||
Err(StorageError::ConfigNotFound) => None,
|
Err(StorageError::ConfigNotFound) => None,
|
||||||
Err(err) => return Err(ApiError::from(err).into()),
|
Err(err) => return Err(ApiError::from(err).into()),
|
||||||
};
|
};
|
||||||
let site_peers = site_replication_remote_peer_deployment_ids().await?;
|
|
||||||
let replication_configuration =
|
let replication_configuration =
|
||||||
merge_user_replication_config_update(replication_configuration, existing_config, &site_peers);
|
merge_user_replication_config_update(replication_configuration, existing_config, &site_peers);
|
||||||
let data = serialize_config(&replication_configuration)?;
|
let data = serialize_config(&replication_configuration)?;
|
||||||
@@ -3695,7 +3700,10 @@ mod tests {
|
|||||||
let req = build_request(input, Method::DELETE);
|
let req = build_request(input, Method::DELETE);
|
||||||
let usecase = DefaultBucketUsecase::without_context();
|
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);
|
assert_eq!(err.code(), &S3ErrorCode::InternalError);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -4781,7 +4789,7 @@ mod tests {
|
|||||||
let req = build_request(input, Method::PUT);
|
let req = build_request(input, Method::PUT);
|
||||||
let usecase = DefaultBucketUsecase::without_context();
|
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);
|
assert_eq!(err.code(), &S3ErrorCode::InternalError);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -4799,7 +4807,7 @@ mod tests {
|
|||||||
.unwrap();
|
.unwrap();
|
||||||
|
|
||||||
let err = DefaultBucketUsecase::without_context()
|
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
|
.await
|
||||||
.expect_err("unsupported fields must be rejected before store access");
|
.expect_err("unsupported fields must be rejected before store access");
|
||||||
|
|
||||||
|
|||||||
@@ -80,6 +80,21 @@ async fn site_replication_gate_enabled() -> S3Result<bool> {
|
|||||||
crate::admin::handlers::site_replication::site_replication_enabled().await
|
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<std::collections::HashSet<String>> {
|
||||||
|
// 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`.
|
/// MinIO `ErrReplicationDenyEditError`.
|
||||||
fn replication_deny_edit_error() -> S3Error {
|
fn replication_deny_edit_error() -> S3Error {
|
||||||
let mut err = S3Error::with_message(
|
let mut err = S3Error::with_message(
|
||||||
@@ -549,8 +564,9 @@ impl S3 for FS {
|
|||||||
req: S3Request<DeleteBucketReplicationInput>,
|
req: S3Request<DeleteBucketReplicationInput>,
|
||||||
) -> S3Result<S3Response<DeleteBucketReplicationOutput>> {
|
) -> S3Result<S3Response<DeleteBucketReplicationOutput>> {
|
||||||
deny_replication_config_edit_for_non_owner(&req).await?;
|
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);
|
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))]
|
#[instrument(level = "debug", skip(self))]
|
||||||
@@ -1403,8 +1419,9 @@ impl S3 for FS {
|
|||||||
req: S3Request<PutBucketReplicationInput>,
|
req: S3Request<PutBucketReplicationInput>,
|
||||||
) -> S3Result<S3Response<PutBucketReplicationOutput>> {
|
) -> S3Result<S3Response<PutBucketReplicationOutput>> {
|
||||||
deny_replication_config_edit_for_non_owner(&req).await?;
|
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);
|
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(
|
async fn put_bucket_request_payment(
|
||||||
|
|||||||
Reference in New Issue
Block a user