From 00324e693612c47316d8c8909b083ff3d58039ef Mon Sep 17 00:00:00 2001 From: cxymds Date: Sun, 2 Aug 2026 22:34:00 +0800 Subject: [PATCH] test(e2e): add replication acceptance matrix (#5642) --- .../src/replication_extension_test.rs | 376 +++++++++++++++++- 1 file changed, 371 insertions(+), 5 deletions(-) diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index 6b654415c..ce4bc633d 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -734,15 +734,19 @@ async fn get_bucket_replication( } async fn enable_bucket_versioning(env: &RustFSTestEnvironment, bucket: &str) -> Result<(), Box> { + set_bucket_versioning(env, bucket, BucketVersioningStatus::Enabled).await +} + +async fn set_bucket_versioning( + env: &RustFSTestEnvironment, + bucket: &str, + status: BucketVersioningStatus, +) -> Result<(), Box> { let client = env.create_s3_client(); client .put_bucket_versioning() .bucket(bucket) - .versioning_configuration( - VersioningConfiguration::builder() - .status(BucketVersioningStatus::Enabled) - .build(), - ) + .versioning_configuration(VersioningConfiguration::builder().status(status).build()) .send() .await?; Ok(()) @@ -1269,6 +1273,48 @@ async fn assert_replication_converged( } } +async fn wait_for_replication_state( + client: &Client, + bucket: &str, + description: &str, + predicate: F, +) -> Result, Box> +where + F: Fn(&[ReplicatedVersion]) -> bool, +{ + let deadline = tokio::time::Instant::now() + Duration::from_secs(30); + loop { + let state = list_replication_state(client, bucket).await?; + if predicate(&state) { + return Ok(state); + } + if tokio::time::Instant::now() >= deadline { + return Err(format!("{description}; last target state: {state:?}").into()); + } + sleep(Duration::from_millis(250)).await; + } +} + +async fn assert_replication_key_absent( + client: &Client, + bucket: &str, + key: &str, + observation: Duration, +) -> Result<(), Box> { + let deadline = tokio::time::Instant::now() + observation; + loop { + let state = list_replication_state(client, bucket).await?; + assert!( + state.iter().all(|entry| entry.key != key), + "unexpected replicated key {bucket}/{key}: {state:?}" + ); + if tokio::time::Instant::now() >= deadline { + return Ok(()); + } + sleep(Duration::from_millis(250)).await; + } +} + async fn get_version_body( client: &Client, bucket: &str, @@ -3624,6 +3670,326 @@ async fn test_bucket_replication_disabled_delete_marker_does_not_propagate() -> Ok(()) } +/// Bounded executable slice for backlog#1620. It deliberately uses real +/// source and target RustFS processes and leaves the full MinIO +/// interoperability profile for a runner that provisions MinIO credentials +/// and a reachable endpoint. +#[tokio::test] +#[serial] +async fn test_bucket_replication_acceptance_matrix_local_dual_targets() -> TestResult { + init_logging(); + + let mut source_env = RustFSTestEnvironment::new().await?; + let mut source_env_vars = replication_fast_env(); + source_env_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV); + source_env.start_rustfs_server_with_env(vec![], &source_env_vars).await?; + + let mut target_env_a = RustFSTestEnvironment::new().await?; + target_env_a + .start_rustfs_server_without_cleanup_with_env(&source_env_vars) + .await?; + let mut target_env_b = RustFSTestEnvironment::new().await?; + target_env_b + .start_rustfs_server_without_cleanup_with_env(&source_env_vars) + .await?; + + let source_bucket = "replication-acceptance-src"; + let target_bucket_a = "replication-acceptance-dst-a"; + let target_bucket_b = "replication-acceptance-dst-b"; + 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_a_arn = set_replication_target(&source_env, source_bucket, &target_env_a, target_bucket_a).await?; + let target_b_arn = set_replication_target(&source_env, source_bucket, &target_env_b, target_bucket_b).await?; + let body = format!( + r#" + + + matrix-prefix + 100 + Enabled + prefix/ + Enabled + Enabled + Enabled + Enabled + {target_a_arn} + + + matrix-both-prefix + 100 + Enabled + both/ + Enabled + Enabled + Enabled + {target_a_arn} + + + matrix-tag + 100 + Enabled + routetagged + Disabled + Enabled + Enabled + {target_b_arn} + + + matrix-disabled + 100 + Disabled + disabled/ + Enabled + Enabled + Enabled + {target_b_arn} + + + matrix-priority-high + 200 + Enabled + priority/ + Disabled + Enabled + Enabled + {target_a_arn} + + + matrix-priority-low + 100 + Enabled + priority/ + Enabled + Enabled + Enabled + {target_a_arn} + +"# + ); + let url = format!("{}/{source_bucket}?replication", source_env.url); + let response = signed_request( + http::Method::PUT, + &url, + &source_env.access_key, + &source_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 replication acceptance matrix failed: {status} {body}").into()); + } + + let saved_config = get_bucket_replication(&source_env, source_bucket).await?.text().await?; + for expected in [ + "matrix-prefix", + "matrix-tag", + "matrix-disabled", + "matrix-priority-high", + "Priority>200", + "Disabled", + "route", + ] { + assert!(saved_config.contains(expected), "replication config omitted {expected}: {saved_config}"); + } + + let version_one = source_client + .put_object() + .bucket(source_bucket) + .key("prefix/versions.txt") + .body(ByteStream::from_static(b"version-one")) + .send() + .await?; + let version_one_id = version_one + .version_id() + .ok_or("first matrix PUT omitted version ID")? + .to_string(); + let version_two = source_client + .put_object() + .bucket(source_bucket) + .key("prefix/versions.txt") + .body(ByteStream::from_static(b"version-two")) + .send() + .await?; + let version_two_id = version_two + .version_id() + .ok_or("second matrix PUT omitted version ID")? + .to_string(); + wait_for_replication_state(&target_client_a, target_bucket_a, "prefix object did not replicate", |state| { + state + .iter() + .any(|entry| entry.key == "prefix/versions.txt" && entry.version_id == version_two_id) + }) + .await?; + + let delete_marker = source_client + .delete_object() + .bucket(source_bucket) + .key("prefix/versions.txt") + .send() + .await?; + let delete_marker_id = delete_marker + .version_id() + .ok_or("matrix DELETE omitted marker version ID")? + .to_string(); + wait_for_replication_state(&target_client_a, target_bucket_a, "enabled delete marker did not replicate", |state| { + state + .iter() + .any(|entry| entry.key == "prefix/versions.txt" && entry.delete_marker && entry.version_id == delete_marker_id) + }) + .await?; + + source_client + .delete_object() + .bucket(source_bucket) + .key("prefix/versions.txt") + .version_id(&version_one_id) + .send() + .await?; + wait_for_replication_state(&target_client_a, target_bucket_a, "enabled version purge did not replicate", |state| { + state.iter().all(|entry| entry.version_id != version_one_id) + }) + .await?; + + source_client + .put_object() + .bucket(source_bucket) + .key("priority/object.txt") + .body(ByteStream::from_static(b"priority winner")) + .send() + .await?; + wait_for_user_get_object(&target_client_a, target_bucket_a, "priority/object.txt").await?; + source_client + .delete_object() + .bucket(source_bucket) + .key("priority/object.txt") + .send() + .await?; + sleep(Duration::from_secs(3)).await; + let priority_state = list_replication_state(&target_client_a, target_bucket_a).await?; + assert!( + priority_state + .iter() + .any(|entry| entry.key == "priority/object.txt" && !entry.delete_marker), + "priority rule did not retain the object version: {priority_state:?}" + ); + assert!( + priority_state + .iter() + .all(|entry| !(entry.key == "priority/object.txt" && entry.delete_marker)), + "lower-priority delete-marker rule overrode the higher-priority disabled rule: {priority_state:?}" + ); + + source_client + .put_object() + .bucket(source_bucket) + .key("tagged/object.txt") + .tagging("route=tagged") + .body(ByteStream::from_static(b"tag filter")) + .send() + .await?; + wait_for_user_get_object(&target_client_b, target_bucket_b, "tagged/object.txt").await?; + source_client + .put_object() + .bucket(source_bucket) + .key("tagged/no-match.txt") + .body(ByteStream::from_static(b"not tagged")) + .send() + .await?; + assert_replication_key_absent(&target_client_b, target_bucket_b, "tagged/no-match.txt", Duration::from_secs(3)).await?; + + source_client + .put_object() + .bucket(source_bucket) + .key("both/object.txt") + .tagging("route=tagged") + .body(ByteStream::from_static(b"mixed targets")) + .send() + .await?; + tokio::try_join!( + wait_for_user_get_object(&target_client_a, target_bucket_a, "both/object.txt"), + wait_for_user_get_object(&target_client_b, target_bucket_b, "both/object.txt"), + )?; + + source_client + .put_object() + .bucket(source_bucket) + .key("disabled/object.txt") + .body(ByteStream::from_static(b"disabled")) + .send() + .await?; + assert_replication_key_absent(&target_client_a, target_bucket_a, "disabled/object.txt", Duration::from_secs(3)).await?; + assert_replication_key_absent(&target_client_b, target_bucket_b, "disabled/object.txt", Duration::from_secs(3)).await?; + + source_client + .delete_object() + .bucket(source_bucket) + .key("tagged/object.txt") + .send() + .await?; + sleep(Duration::from_secs(3)).await; + let tagged_state = list_replication_state(&target_client_b, target_bucket_b).await?; + assert!( + tagged_state + .iter() + .any(|entry| entry.key == "tagged/object.txt" && !entry.delete_marker), + "tag rule should retain the replicated data version: {tagged_state:?}" + ); + assert!( + tagged_state + .iter() + .all(|entry| !(entry.key == "tagged/object.txt" && entry.delete_marker)), + "tag rule with disabled delete-marker replication created a marker: {tagged_state:?}" + ); + + set_bucket_versioning(&source_env, source_bucket, BucketVersioningStatus::Suspended).await?; + set_bucket_versioning(&target_env_a, target_bucket_a, BucketVersioningStatus::Suspended).await?; + let null_put = source_client + .put_object() + .bucket(source_bucket) + .key("prefix/null.txt") + .body(ByteStream::from_static(b"null version")) + .send() + .await?; + assert!(null_put.version_id().is_none(), "suspended source PUT must create a null version"); + wait_for_replication_state(&target_client_a, target_bucket_a, "null version did not replicate", |state| { + state + .iter() + .any(|entry| entry.key == "prefix/null.txt" && entry.version_id == "null" && !entry.delete_marker) + }) + .await?; + let null_delete = source_client + .delete_object() + .bucket(source_bucket) + .key("prefix/null.txt") + .send() + .await?; + assert!( + null_delete.version_id().is_none(), + "suspended source DELETE must create a null delete marker" + ); + wait_for_replication_state(&target_client_a, target_bucket_a, "null delete marker did not replicate", |state| { + state + .iter() + .any(|entry| entry.key == "prefix/null.txt" && entry.version_id == "null" && entry.delete_marker) + }) + .await?; + + Ok(()) +} + #[tokio::test] #[serial] async fn test_single_bucket_multipart_replication_fans_out_to_multiple_targets() -> Result<(), Box> {