From 849ea9a1228adb9ae270c8cffee4c55c495e95b1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Mon, 6 Jul 2026 22:15:02 +0800 Subject: [PATCH] fix(site-replication): align IAM and bucket metadata replication (#4318) --- crates/ecstore/src/bucket/quota/checker.rs | 7 +-- rustfs/src/admin/handlers/quota.rs | 54 ++++++++++++++++++- rustfs/src/admin/handlers/site_replication.rs | 21 ++++++-- rustfs/src/admin/handlers/user.rs | 34 +++++++++++- rustfs/src/app/object_usecase.rs | 6 +++ rustfs/src/storage/ecfs.rs | 27 +++++++++- 6 files changed, 137 insertions(+), 12 deletions(-) diff --git a/crates/ecstore/src/bucket/quota/checker.rs b/crates/ecstore/src/bucket/quota/checker.rs index 6f1f05845..63b2e9587 100644 --- a/crates/ecstore/src/bucket/quota/checker.rs +++ b/crates/ecstore/src/bucket/quota/checker.rs @@ -19,6 +19,7 @@ use rustfs_common::metrics::Metric; use rustfs_config::QUOTA_CONFIG_FILE; use std::sync::Arc; use std::time::Instant; +use time::OffsetDateTime; use tokio::sync::RwLock; use tracing::{debug, warn}; @@ -137,18 +138,18 @@ impl QuotaChecker { Ok(quota) } - pub async fn set_quota_config(&mut self, bucket: &str, quota: BucketQuota) -> Result<(), QuotaError> { + pub async fn set_quota_config(&mut self, bucket: &str, quota: BucketQuota) -> Result { let json_data = serde_json::to_vec("a).map_err(|e| QuotaError::InvalidConfig { reason: format!("Failed to serialize quota config: {}", e), })?; let start_time = Instant::now(); - update(bucket, QUOTA_CONFIG_FILE, json_data) + let updated_at = update(bucket, QUOTA_CONFIG_FILE, json_data) .await .map_err(QuotaError::StorageError)?; rustfs_common::metrics::Metrics::inc_time(Metric::QuotaSync, start_time.elapsed()); - Ok(()) + Ok(updated_at) } pub async fn get_quota_stats(&self, bucket: &str) -> Result<(BucketQuota, Option), QuotaError> { diff --git a/rustfs/src/admin/handlers/quota.rs b/rustfs/src/admin/handlers/quota.rs index bb6ca3c02..1a91e27ab 100644 --- a/rustfs/src/admin/handlers/quota.rs +++ b/rustfs/src/admin/handlers/quota.rs @@ -15,6 +15,7 @@ //! Quota admin handlers for HTTP API use crate::admin::auth::{validate_admin_request, validate_admin_request_with_bucket}; +use crate::admin::handlers::site_replication::site_replication_bucket_meta_hook; use crate::admin::router::{AdminOperation, Operation, S3Router}; use crate::admin::runtime_sources::{current_bucket_metadata_handle, current_object_store_handle}; use crate::admin::storage_api::bucket::metadata_sys::BucketMetadataSys; @@ -24,6 +25,7 @@ use crate::auth::{check_key_valid, get_session_token}; use crate::server::ADMIN_PREFIX; use hyper::{Method, StatusCode}; use matchit::Params; +use rustfs_madmin::{SITE_REPL_API_VERSION, SRBucketMeta}; use rustfs_policy::policy::action::{Action, AdminAction, S3Action}; use s3s::{Body, S3Request, S3Response, S3Result, s3_error}; use serde::{Deserialize, Serialize}; @@ -284,11 +286,35 @@ impl Operation for SetBucketQuotaHandler { .ok_or_else(|| s3_error!(InternalError, "{}", rustfs_config::QUOTA_METADATA_SYSTEM_ERROR_MSG))?; let mut quota_checker = QuotaChecker::new(metadata_sys_lock.clone()); - quota_checker + let updated_at = quota_checker .set_quota_config(&bucket, quota.clone()) .await .map_err(|e| s3_error!(InternalError, "failed to set quota: {}", e))?; + if let Err(err) = site_replication_bucket_meta_hook(SRBucketMeta { + bucket: bucket.clone(), + r#type: "quota-config".to_string(), + quota: Some( + serde_json::to_value("a).map_err(|e| s3_error!(InternalError, "failed to encode quota payload: {}", e))?, + ), + updated_at: Some(updated_at), + api_version: Some(SITE_REPL_API_VERSION.to_string()), + ..Default::default() + }) + .await + { + warn!( + event = EVENT_ADMIN_QUOTA_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_QUOTA, + action = "set_bucket_quota", + bucket = %bucket, + state = "site_replication_hook_failed", + error = ?err, + "admin quota state" + ); + } + // Get real-time usage from data usage system let current_usage = current_usage_from_context(&bucket).await; @@ -427,11 +453,35 @@ impl Operation for ClearBucketQuotaHandler { // Clear quota (set to None) let quota = BucketQuota::new(None); - quota_checker + let updated_at = quota_checker .set_quota_config(&bucket, quota.clone()) .await .map_err(|e| s3_error!(InternalError, "failed to clear quota: {}", e))?; + if let Err(err) = site_replication_bucket_meta_hook(SRBucketMeta { + bucket: bucket.clone(), + r#type: "quota-config".to_string(), + quota: Some( + serde_json::to_value("a).map_err(|e| s3_error!(InternalError, "failed to encode quota payload: {}", e))?, + ), + updated_at: Some(updated_at), + api_version: Some(SITE_REPL_API_VERSION.to_string()), + ..Default::default() + }) + .await + { + warn!( + event = EVENT_ADMIN_QUOTA_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_QUOTA, + action = "clear_bucket_quota", + bucket = %bucket, + state = "site_replication_hook_failed", + error = ?err, + "admin quota state" + ); + } + info!( event = EVENT_ADMIN_QUOTA_STATE, component = LOG_COMPONENT_ADMIN, diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index 6a397552a..006de4cb5 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/rustfs/src/admin/handlers/site_replication.rs @@ -4648,16 +4648,29 @@ async fn apply_iam_item(item: SRIAMItem) -> S3Result<()> { let Some(user) = item.iam_user else { return Err(s3_error!(InvalidRequest, "iamUser is required")); }; + if let Some(local) = iam_sys.get_user(&user.access_key).await + && is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at) + { + return Ok(()); + } if user.is_delete_req { iam_sys.delete_user(&user.access_key, true).await.map_err(ApiError::from)?; } else { let Some(user_req) = user.user_req else { return Err(s3_error!(InvalidRequest, "userReq is required")); }; - iam_sys - .create_user(&user.access_key, &user_req) - .await - .map_err(ApiError::from)?; + let is_status_only_update = user_req.secret_key.is_empty() && user_req.policy.is_none(); + if is_status_only_update { + iam_sys + .set_user_status(&user.access_key, user_req.status) + .await + .map_err(ApiError::from)?; + } else { + iam_sys + .create_user(&user.access_key, &user_req) + .await + .map_err(ApiError::from)?; + } } Ok(()) } diff --git a/rustfs/src/admin/handlers/user.rs b/rustfs/src/admin/handlers/user.rs index 1cdb6eeea..3a576ef7a 100644 --- a/rustfs/src/admin/handlers/user.rs +++ b/rustfs/src/admin/handlers/user.rs @@ -360,11 +360,41 @@ impl Operation for SetUserStatus { return Err(s3_error!(InternalError, "iam is not initialized")); }; - iam_store - .set_user_status(ak, status) + let updated_at = iam_store + .set_user_status(ak, status.clone()) .await .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("failed to set user status: {e}")))?; + if let Err(err) = site_replication_iam_change_hook(SRIAMItem { + r#type: "iam-user".to_string(), + iam_user: Some(SRIAMUser { + access_key: ak.to_string(), + is_delete_req: false, + user_req: Some(AddOrUpdateUserReq { + secret_key: String::new(), + policy: None, + status: status.clone(), + }), + api_version: Some(SITE_REPL_API_VERSION.to_string()), + }), + updated_at: Some(updated_at), + api_version: Some(SITE_REPL_API_VERSION.to_string()), + ..Default::default() + }) + .await + { + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_USER, + event = EVENT_ADMIN_USER_STATE, + access_key = %ak, + action = "set_user_status", + result = "site_replication_hook_failed", + error = ?err, + "admin user state" + ); + } + let mut header = HeaderMap::new(); header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value")); header.insert(CONTENT_LENGTH, "0".parse().expect("valid header value")); diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index d92b990e9..931b65a99 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -5581,6 +5581,12 @@ impl DefaultObjectUsecase { { response.headers.insert(header_name, header_value); } + if info.replication_status != ReplicationStatusType::Empty + && let Ok(header_name) = http::HeaderName::from_bytes(AMZ_BUCKET_REPLICATION_STATUS.to_ascii_lowercase().as_bytes()) + && let Ok(header_value) = HeaderValue::from_str(info.replication_status.as_str()) + { + response.headers.insert(header_name, header_value); + } let result = Ok(response); let _ = helper.complete(&result); diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index 0f97d59fb..92b6798e7 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -21,6 +21,7 @@ use super::{ update_bucket_metadata_config, }; use super::{StorageReplicationConfigExt as _, StorageVersioningConfigExt as _}; +use crate::admin::handlers::site_replication::site_replication_bucket_meta_hook; use crate::error::ApiError; use crate::storage::access::has_bypass_governance_header; use crate::storage::helper::OperationHelper; @@ -38,6 +39,7 @@ use crate::table_catalog; use http::StatusCode; use metrics::{counter, histogram}; use rustfs_io_metrics::record_s3_op; +use rustfs_madmin::{SITE_REPL_API_VERSION, SRBucketMeta}; use rustfs_s3_ops::S3Operation; use rustfs_targets::EventName; use rustfs_utils::http::headers::{ @@ -1325,8 +1327,10 @@ impl S3 for FS { }; let data = serialize(&input_cfg).map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, format!("{err}")))?; + let object_lock_config = + String::from_utf8(data.clone()).map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, format!("{err}")))?; - update_bucket_metadata_config(&bucket, OBJECT_LOCK_CONFIG, data) + let updated_at = update_bucket_metadata_config(&bucket, OBJECT_LOCK_CONFIG, data) .await .map_err(ApiError::from)?; @@ -1345,6 +1349,27 @@ impl S3 for FS { .map_err(ApiError::from)?; } + if let Err(err) = site_replication_bucket_meta_hook(SRBucketMeta { + bucket: bucket.clone(), + r#type: "object-lock-config".to_string(), + object_lock_config: Some(object_lock_config), + updated_at: Some(updated_at), + api_version: Some(SITE_REPL_API_VERSION.to_string()), + ..Default::default() + }) + .await + { + warn!( + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_OBJECT_LOCK, + event = "put_object_lock_configuration", + bucket = %bucket, + result = "site_replication_hook_failed", + error = ?err, + "storage object lock state" + ); + } + rustfs_scanner::record_dirty_usage_bucket(&bucket); Ok(S3Response::new(PutObjectLockConfigurationOutput::default())) }