diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index efd57a012..6466c842f 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -255,6 +255,60 @@ async fn put_bucket_replication( Ok(()) } +async fn put_bucket_replication_rules( + env: &RustFSTestEnvironment, + bucket: &str, + target_arns: &[&str], +) -> Result<(), Box> { + let mut rules = String::new(); + for (idx, target_arn) in target_arns.iter().enumerate() { + rules.push_str(&format!( + r#" + + rule-{} + {} + Enabled + + Enabled + + + Enabled + + + {} + + "#, + idx + 1, + idx + 1, + target_arn + )); + } + + let body = format!( + r#" + {rules} +"# + ); + let url = format!("{}/{bucket}?replication", env.url); + let response = signed_request( + http::Method::PUT, + &url, + &env.access_key, + &env.secret_key, + Some(body.into_bytes()), + Some("application/xml"), + ) + .await?; + + if response.status() != StatusCode::OK { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + return Err(format!("put bucket replication with multiple rules failed: {status} {body}").into()); + } + + Ok(()) +} + async fn delete_bucket_replication( env: &RustFSTestEnvironment, bucket: &str, @@ -415,6 +469,33 @@ async fn admin_attach_policy_to_group( Ok(()) } +async fn wait_for_replicated_object( + client: &aws_sdk_s3::Client, + bucket: &str, + key: &str, + expected_body: &str, +) -> Result<(), Box> { + let deadline = tokio::time::Instant::now() + Duration::from_secs(30); + + loop { + match client.get_object().bucket(bucket).key(key).send().await { + Ok(output) => { + let body = output.body.collect().await?.into_bytes(); + let body = String::from_utf8(body.to_vec())?; + if body == expected_body { + return Ok(()); + } + return Err(format!("replicated object body mismatch: expected {expected_body}, got {body}").into()); + } + Err(_err) if tokio::time::Instant::now() < deadline => { + sleep(Duration::from_secs(1)).await; + continue; + } + Err(err) => return Err(err.into()), + } + } +} + async fn run_replication_check( env: &RustFSTestEnvironment, bucket: &str, @@ -1518,6 +1599,55 @@ async fn test_delete_bucket_replication_removes_remote_target() -> Result<(), Bo Ok(()) } +#[tokio::test] +#[serial] +async fn test_single_bucket_replication_fans_out_to_multiple_targets() -> Result<(), Box> { + init_logging(); + + let mut source_env = RustFSTestEnvironment::new().await?; + source_env.start_rustfs_server(vec![]).await?; + + let mut target_env_a = RustFSTestEnvironment::new().await?; + target_env_a.start_rustfs_server_without_cleanup(vec![]).await?; + + let mut target_env_b = RustFSTestEnvironment::new().await?; + target_env_b.start_rustfs_server_without_cleanup(vec![]).await?; + + let source_bucket = "replication-fanout-src"; + let target_bucket_a = "replication-fanout-dst-a"; + let target_bucket_b = "replication-fanout-dst-b"; + let object_key = "fanout.txt"; + let body = "payload-fanout"; + + let source_client = source_env.create_s3_client(); + let target_client_a = target_env_a.create_s3_client(); + let target_client_b = target_env_b.create_s3_client(); + + source_client.create_bucket().bucket(source_bucket).send().await?; + target_client_a.create_bucket().bucket(target_bucket_a).send().await?; + target_client_b.create_bucket().bucket(target_bucket_b).send().await?; + enable_bucket_versioning(&source_env, source_bucket).await?; + enable_bucket_versioning(&target_env_a, target_bucket_a).await?; + enable_bucket_versioning(&target_env_b, target_bucket_b).await?; + + let target_arn_a = set_replication_target(&source_env, source_bucket, &target_env_a, target_bucket_a).await?; + let target_arn_b = set_replication_target(&source_env, source_bucket, &target_env_b, target_bucket_b).await?; + put_bucket_replication_rules(&source_env, source_bucket, &[target_arn_a.as_str(), target_arn_b.as_str()]).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_a, target_bucket_a, object_key, body).await?; + wait_for_replicated_object(&target_client_b, target_bucket_b, 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/replication/config.rs b/crates/ecstore/src/bucket/replication/config.rs index e150488ee..3eea72b16 100644 --- a/crates/ecstore/src/bucket/replication/config.rs +++ b/crates/ecstore/src/bucket/replication/config.rs @@ -224,19 +224,98 @@ impl ReplicationConfigurationExt for ReplicationConfiguration { continue; } - if !self.role.is_empty() { - arns.push(self.role.clone()); // Use the legacy RoleArn when present - return arns; - } - - if !targets_map.contains(&rule.destination.bucket) { + if !rule.destination.bucket.is_empty() && !targets_map.contains(&rule.destination.bucket) { targets_map.insert(rule.destination.bucket.clone()); } } + if targets_map.is_empty() && !self.role.is_empty() { + arns.push(self.role.clone()); + return arns; + } + for arn in targets_map { arns.push(arn); } arns } } + +#[cfg(test)] +mod tests { + use super::*; + use s3s::dto::{DeleteMarkerReplication, Destination, ExistingObjectReplication, ReplicationRule}; + + fn replication_rule(id: &str, arn: &str) -> ReplicationRule { + ReplicationRule { + delete_marker_replication: Some(DeleteMarkerReplication::default()), + delete_replication: None, + destination: Destination { + bucket: arn.to_string(), + ..Default::default() + }, + existing_object_replication: Some(ExistingObjectReplication { + status: ExistingObjectReplicationStatus::from_static(ExistingObjectReplicationStatus::ENABLED), + }), + filter: None, + id: Some(id.to_string()), + prefix: Some(String::new()), + priority: Some(1), + source_selection_criteria: None, + status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED), + } + } + + #[test] + fn filter_target_arns_keeps_multiple_destinations_when_role_is_present() { + let config = ReplicationConfiguration { + role: "arn:legacy:target".to_string(), + rules: vec![ + replication_rule("rule-1", "arn:target:a"), + replication_rule("rule-2", "arn:target:b"), + ], + }; + + let arns = config.filter_target_arns(&ObjectOpts { + name: "object".to_string(), + op_type: ReplicationType::Object, + ..Default::default() + }); + + assert_eq!(arns.len(), 2); + assert!(arns.iter().any(|arn| arn == "arn:target:a")); + assert!(arns.iter().any(|arn| arn == "arn:target:b")); + } + + #[test] + fn filter_target_arns_falls_back_to_role_when_destination_is_empty() { + let config = ReplicationConfiguration { + role: "arn:legacy:target".to_string(), + rules: vec![ReplicationRule { + delete_marker_replication: Some(DeleteMarkerReplication::default()), + delete_replication: None, + destination: Destination { + bucket: String::new(), + ..Default::default() + }, + existing_object_replication: Some(ExistingObjectReplication { + status: ExistingObjectReplicationStatus::from_static(ExistingObjectReplicationStatus::ENABLED), + }), + filter: None, + id: Some("rule-1".to_string()), + prefix: Some(String::new()), + priority: Some(1), + source_selection_criteria: None, + status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED), + }], + }; + + let arns = config.filter_target_arns(&ObjectOpts { + name: "object".to_string(), + op_type: ReplicationType::Object, + ..Default::default() + }); + + assert_eq!(arns, vec!["arn:legacy:target".to_string()]); + } +}