From 334184b005b3f621f2ee3118a1171371a9575376 Mon Sep 17 00:00:00 2001 From: weisd Date: Mon, 27 Apr 2026 17:27:33 +0800 Subject: [PATCH] fix(replication): prevent target state loss across buckets (#2704) --- .../src/replication_extension_test.rs | 85 ++++++++++++ .../ecstore/src/bucket/bucket_target_sys.rs | 122 +++++++++--------- rustfs/src/admin/handlers/replication.rs | 16 +-- rustfs/src/app/bucket_usecase.rs | 121 ++++++++++++++++- 4 files changed, 271 insertions(+), 73 deletions(-) diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index 6466c842f..a362678eb 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -21,6 +21,7 @@ use aws_sdk_s3::types::{BucketVersioningStatus, VersioningConfiguration}; use aws_sdk_s3::{Client, Config}; use http::header::{CONTENT_TYPE, HOST}; use reqwest::StatusCode; +use rustfs_ecstore::bucket::bucket_target_sys::BucketTargetSys; use rustfs_madmin::{ AddServiceAccountReq, ListServiceAccountsResp, PeerInfo, PeerSite, ReplicateAddStatus, ReplicateEditStatus, ReplicateRemoveStatus, SRRemoveReq, SRResyncOpStatus, SRStatusInfo, SiteReplicationInfo, SyncStatus, @@ -1648,6 +1649,90 @@ async fn test_single_bucket_replication_fans_out_to_multiple_targets() -> Result Ok(()) } +#[tokio::test] +#[serial] +async fn test_sequential_bucket_replication_succeeds_for_multiple_buckets() -> Result<(), Box> { + init_logging(); + + let mut source_env = RustFSTestEnvironment::new().await?; + source_env.start_rustfs_server(vec![]).await?; + + let mut target_env = RustFSTestEnvironment::new().await?; + target_env.start_rustfs_server_without_cleanup(vec![]).await?; + + let source_client = source_env.create_s3_client(); + let target_client = target_env.create_s3_client(); + + for idx in 1..=5 { + let source_bucket = format!("replication-multi-src-{idx}"); + let target_bucket = format!("replication-multi-dst-{idx}"); + let object_key = format!("probe-{idx}.txt"); + let body = format!("payload-{idx}"); + + source_client.create_bucket().bucket(&source_bucket).send().await?; + target_client.create_bucket().bucket(&target_bucket).send().await?; + enable_bucket_versioning(&source_env, &source_bucket).await?; + enable_bucket_versioning(&target_env, &target_bucket).await?; + + let target_arn = set_replication_target(&source_env, &source_bucket, &target_env, &target_bucket).await?; + put_bucket_replication(&source_env, &source_bucket, &target_arn).await?; + + source_client + .put_object() + .bucket(&source_bucket) + .key(&object_key) + .body(ByteStream::from(body.clone().into_bytes())) + .send() + .await?; + + wait_for_replicated_object(&target_client, &target_bucket, &object_key, &body).await?; + } + + Ok(()) +} + +#[tokio::test] +#[serial] +async fn test_replication_recovers_after_runtime_target_cache_is_cleared() -> Result<(), Box> { + init_logging(); + + let mut source_env = RustFSTestEnvironment::new().await?; + source_env.start_rustfs_server(vec![]).await?; + + let mut target_env = RustFSTestEnvironment::new().await?; + target_env.start_rustfs_server_without_cleanup(vec![]).await?; + + let source_bucket = "replication-refresh-src"; + let target_bucket = "replication-refresh-dst"; + let object_key = "probe-refresh.txt"; + let body = "payload-refresh"; + + let source_client = source_env.create_s3_client(); + let target_client = target_env.create_s3_client(); + + source_client.create_bucket().bucket(source_bucket).send().await?; + target_client.create_bucket().bucket(target_bucket).send().await?; + enable_bucket_versioning(&source_env, source_bucket).await?; + enable_bucket_versioning(&target_env, target_bucket).await?; + + let target_arn = set_replication_target(&source_env, source_bucket, &target_env, target_bucket).await?; + put_bucket_replication(&source_env, source_bucket, &target_arn).await?; + + BucketTargetSys::get().delete(source_bucket).await; + + source_client + .put_object() + .bucket(source_bucket) + .key(object_key) + .body(ByteStream::from(body.as_bytes().to_vec())) + .send() + .await?; + + wait_for_replicated_object(&target_client, target_bucket, object_key, body).await?; + + Ok(()) +} + #[tokio::test] #[serial] async fn test_site_replication_resync_start_cancel_restart_real_dual_node() -> Result<(), Box> { diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index de049204f..707acb33b 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/crates/ecstore/src/bucket/bucket_target_sys.rs @@ -395,8 +395,27 @@ impl BucketTargetSys { } } - pub async fn set_target(&self, bucket: &str, target: &BucketTarget, update: bool) -> Result<(), BucketTargetError> { - if !target.target_type.is_valid() && !update { + pub async fn set_target( + &self, + bucket: &str, + target: &BucketTarget, + update: bool, + ) -> Result { + self.validate_target(bucket, target).await?; + + let mut bucket_targets = match self.list_bucket_targets(bucket).await { + Ok(targets) => targets, + Err(BucketTargetError::BucketRemoteTargetNotFound { .. }) => BucketTargets::default(), + Err(err) => return Err(err), + }; + + Self::upsert_target_entry(&mut bucket_targets.targets, target, update)?; + + Ok(bucket_targets) + } + + pub async fn validate_target(&self, bucket: &str, target: &BucketTarget) -> Result<(), BucketTargetError> { + if !target.target_type.is_valid() { return Err(BucketTargetError::BucketRemoteArnTypeInvalid { bucket: bucket.to_string(), }); @@ -450,52 +469,44 @@ impl BucketTargetSys { } } - { - let mut targets_map = self.targets_map.write().await; - let bucket_targets = targets_map.entry(bucket.to_string()).or_insert_with(Vec::new); - let mut found = false; + Ok(()) + } - for (idx, existing_target) in bucket_targets.iter().enumerate() { - if existing_target.target_type.to_string() == target.target_type.to_string() { - if existing_target.arn == target.arn { - if !update { - return Err(BucketTargetError::BucketRemoteAlreadyExists { - bucket: existing_target.target_bucket.clone(), - }); - } - bucket_targets[idx] = target.clone(); - found = true; - break; - } - if existing_target.endpoint == target.endpoint { + fn upsert_target_entry( + bucket_targets: &mut Vec, + target: &BucketTarget, + update: bool, + ) -> Result<(), BucketTargetError> { + let mut found = false; + + for (idx, existing_target) in bucket_targets.iter().enumerate() { + if existing_target.target_type.to_string() == target.target_type.to_string() { + if existing_target.arn == target.arn { + if !update { return Err(BucketTargetError::BucketRemoteAlreadyExists { bucket: existing_target.target_bucket.clone(), }); } + bucket_targets[idx] = target.clone(); + found = true; + break; + } + if existing_target.endpoint == target.endpoint { + return Err(BucketTargetError::BucketRemoteAlreadyExists { + bucket: existing_target.target_bucket.clone(), + }); } } - - if !found && !update { - bucket_targets.push(target.clone()); - } } - { - let mut arn_remotes_map = self.arn_remotes_map.write().await; - arn_remotes_map.insert( - target.arn.clone(), - ArnTarget { - client: Some(Arc::new(target_client)), - last_refresh: OffsetDateTime::now_utc(), - }, - ); + if !found && !update { + bucket_targets.push(target.clone()); } - self.update_bandwidth_limit(bucket, &target.arn, target.bandwidth_limit); Ok(()) } - pub async fn remove_target(&self, bucket: &str, arn_str: &str) -> Result<(), BucketTargetError> { + pub async fn remove_target(&self, bucket: &str, arn_str: &str) -> Result { if arn_str.is_empty() { return Err(BucketTargetError::BucketRemoteArnInvalid { bucket: bucket.to_string(), @@ -524,33 +535,16 @@ impl BucketTargetSys { } } - { - let mut targets_map = self.targets_map.write().await; + let targets = self.list_bucket_targets(bucket).await?; + let new_targets: Vec = targets.targets.iter().filter(|t| t.arn != arn_str).cloned().collect(); - let Some(targets) = targets_map.get(bucket) else { - return Err(BucketTargetError::BucketRemoteTargetNotFound { - bucket: bucket.to_string(), - }); - }; - - let new_targets: Vec = targets.iter().filter(|t| t.arn != arn_str).cloned().collect(); - - if new_targets.len() == targets.len() { - return Err(BucketTargetError::BucketRemoteTargetNotFound { - bucket: bucket.to_string(), - }); - } - - targets_map.insert(bucket.to_string(), new_targets); + if new_targets.len() == targets.targets.len() { + return Err(BucketTargetError::BucketRemoteTargetNotFound { + bucket: bucket.to_string(), + }); } - { - self.arn_remotes_map.write().await.remove(arn_str); - } - - self.update_bandwidth_limit(bucket, arn_str, 0); - - Ok(()) + Ok(BucketTargets { targets: new_targets }) } pub async fn mark_refresh_in_progress(&self, bucket: &str, arn: &str) { @@ -603,7 +597,7 @@ impl BucketTargetSys { if let Some(last_refresh) = last_refresh { let now = OffsetDateTime::now_utc(); - if now - last_refresh > Duration::from_secs(60 * 5) { + if now - last_refresh < Duration::from_secs(60 * 5) { return None; } } @@ -619,6 +613,16 @@ impl BucketTargetSys { } }; + let cli = self + .arn_remotes_map + .read() + .await + .get(arn) + .and_then(|target| target.client.clone()); + if cli.is_some() { + return cli; + } + self.inc_arn_errs(bucket, arn).await; None } diff --git a/rustfs/src/admin/handlers/replication.rs b/rustfs/src/admin/handlers/replication.rs index 1958c2e24..0d9618ab3 100644 --- a/rustfs/src/admin/handlers/replication.rs +++ b/rustfs/src/admin/handlers/replication.rs @@ -288,15 +288,10 @@ impl Operation for SetRemoteTargetHandler { let arn = remote_target.arn.clone(); - bucket_target_sys + let targets = bucket_target_sys .set_target(bucket, &remote_target, update) .await .map_err(map_bucket_target_error)?; - - let targets = bucket_target_sys.list_bucket_targets(bucket).await.map_err(|e| { - error!("Failed to list bucket targets: {}", e); - S3Error::with_message(S3ErrorCode::InternalError, "Failed to list bucket targets".to_string()) - })?; let json_targets = serde_json::to_vec(&targets).map_err(|e| { error!("Serialization error: {}", e); S3Error::with_message(S3ErrorCode::InternalError, "Failed to serialize targets".to_string()) @@ -308,6 +303,7 @@ impl Operation for SetRemoteTargetHandler { error!("Failed to update bucket targets: {}", e); S3Error::with_message(S3ErrorCode::InternalError, format!("Failed to update bucket targets: {e}")) })?; + bucket_target_sys.update_all_targets(bucket, Some(&targets)).await; let arn_str = serde_json::to_string(&arn).unwrap_or_default(); @@ -405,12 +401,7 @@ impl Operation for RemoveRemoteTargetHandler { let sys = BucketTargetSys::get(); - sys.remove_target(bucket, arn_str).await.map_err(map_bucket_target_error)?; - - let targets = sys.list_bucket_targets(bucket).await.map_err(|e| { - error!("Failed to list bucket targets: {}", e); - S3Error::with_message(S3ErrorCode::InternalError, "Failed to list bucket targets".to_string()) - })?; + let targets = sys.remove_target(bucket, arn_str).await.map_err(map_bucket_target_error)?; let json_targets = serde_json::to_vec(&targets).map_err(|e| { error!("Serialization error: {}", e); @@ -423,6 +414,7 @@ impl Operation for RemoveRemoteTargetHandler { error!("Failed to update bucket targets: {}", e); S3Error::with_message(S3ErrorCode::InternalError, format!("Failed to update bucket targets: {e}")) })?; + sys.update_all_targets(bucket, Some(&targets)).await; Ok(S3Response::new((StatusCode::NO_CONTENT, Body::from("".to_string())))) } diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index 8753007aa..0a0aa4b3c 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -46,7 +46,7 @@ use rustfs_ecstore::bucket::{ metadata_sys, object_lock::ObjectLockApi, policy_sys::PolicySys, - target::BucketTargetType, + target::{BucketTargetType, BucketTargets}, utils::serialize, versioning::VersioningApi, versioning_sys::BucketVersioningSys, @@ -119,6 +119,59 @@ fn replication_target_arns(config: &ReplicationConfiguration) -> HashSet arns } +fn validate_replication_config_targets(targets: &BucketTargets, config: &ReplicationConfiguration) -> S3Result<()> { + let configured_arns = targets + .targets + .iter() + .filter(|target| target.target_type == BucketTargetType::ReplicationService) + .map(|target| target.arn.as_str()) + .collect::>(); + + for rule in &config.rules { + if rule.status == ReplicationRuleStatus::from_static(ReplicationRuleStatus::DISABLED) { + continue; + } + + let configured_arn = if config.role.trim().is_empty() { + rule.destination.bucket.trim() + } else { + config.role.trim() + }; + + if !configured_arn.is_empty() && configured_arns.contains(configured_arn) { + continue; + } + + return Err(s3_error!( + InvalidRequest, + "replication config with rule ID {} has a stale target", + rule.id.clone().unwrap_or_default() + )); + } + + Ok(()) +} + +async fn validate_bucket_replication_update(bucket: &str, config: &ReplicationConfiguration) -> 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) +} + async fn remove_replication_targets_for_config(bucket: &str, config: &ReplicationConfiguration) -> S3Result<()> { let target_arns = replication_target_arns(config); if target_arns.is_empty() { @@ -1615,7 +1668,7 @@ impl DefaultBucketUsecase { .await .map_err(ApiError::from)?; - // TODO: check enable, versioning enable + validate_bucket_replication_update(&bucket, &replication_configuration).await?; let data = serialize_config(&replication_configuration)?; metadata_sys::update(&bucket, BUCKET_REPLICATION_CONFIG, data) .await @@ -1979,6 +2032,70 @@ mod tests { assert!(arns.contains(destination)); } + fn replication_targets_with_arn(arns: &[&str]) -> BucketTargets { + BucketTargets { + targets: arns + .iter() + .map(|arn| rustfs_ecstore::bucket::target::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 = ReplicationConfiguration { + 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 = ReplicationConfiguration { + 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 = ReplicationConfiguration { + role: arn.to_string(), + 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_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 = ReplicationRuleStatus::from_static(ReplicationRuleStatus::DISABLED); + let config = ReplicationConfiguration { + role: String::new(), + rules: vec![rule], + }; + + validate_replication_config_targets(&targets, &config).expect("disabled rules should not require live targets"); + } + #[test] fn versioning_configuration_has_object_lock_incompatible_settings_rejects_suspended() { let config = VersioningConfiguration {