diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index c5023d64c..1c0e64517 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -6975,3 +6975,87 @@ async fn test_replication_put_and_create_multipart_carry_source_version_id_query target.shutdown().await; Ok(()) } + +/// P1-20: inbound replicas never cascade. An object replicated A->B carries +/// x-amz-replication-status=REPLICA on B; `must_replicate` returns an empty +/// decision for replicas, so even an ExistingObjectReplication=Enabled rule +/// configured on B AFTER the replica landed (making it an "existing object" +/// for that rule) must never push it onward — while B's own native objects +/// flow to the onward bucket, proving B's outbound replication and scanner +/// are live. +#[tokio::test] +#[serial] +async fn test_scanner_never_cascades_inbound_replicas() -> TestResult { + init_logging(); + + let mut env_a = RustFSTestEnvironment::new().await?; + let mut env_a_vars = replication_fast_env(); + env_a_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV); + env_a.start_rustfs_server_with_env(vec![], &env_a_vars).await?; + + // B becomes a replication source itself, driven by its scanner. + let mut env_b = RustFSTestEnvironment::new().await?; + let mut env_b_vars = replication_fast_env(); + env_b_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV); + env_b_vars.extend_from_slice(FAST_SCANNER_ENV); + env_b.start_rustfs_server_with_env(vec![], &env_b_vars).await?; + + let bucket_a = "cascade-a-src"; + let bucket_b = "cascade-b-mid"; + let bucket_c = "cascade-a-third"; + let client_a = env_a.create_s3_client(); + let client_b = env_b.create_s3_client(); + client_a.create_bucket().bucket(bucket_a).send().await?; + client_b.create_bucket().bucket(bucket_b).send().await?; + client_a.create_bucket().bucket(bucket_c).send().await?; + enable_bucket_versioning(&env_a, bucket_a).await?; + enable_bucket_versioning(&env_b, bucket_b).await?; + enable_bucket_versioning(&env_a, bucket_c).await?; + + // A -> B first: the replica lands on B before B has any outbound rule. + let arn_ab = set_replication_target(&env_a, bucket_a, &env_b, bucket_b).await?; + put_bucket_replication(&env_a, bucket_a, &arn_ab).await?; + + let replica_key = "replica-object.txt"; + let replica_payload = "replica payload"; + client_a + .put_object() + .bucket(bucket_a) + .key(replica_key) + .body(ByteStream::from_static(replica_payload.as_bytes())) + .send() + .await?; + wait_for_replicated_object(&client_b, bucket_b, replica_key, replica_payload).await?; + + // Precondition for the anti-cascade contract: the inbound copy must carry + // REPLICA status on B. If this fails, the break is in inbound status + // stamping, not in the scanner guard. + let inbound = client_b.head_object().bucket(bucket_b).key(replica_key).send().await?; + assert_eq!( + inbound.replication_status().map(|status| status.as_str()), + Some("REPLICA"), + "inbound replica must be stamped REPLICA on the target" + ); + + // Now wire B's outbound rule; the replica is an "existing object" for it. + let arn_bc = set_replication_target(&env_b, bucket_b, &env_a, bucket_c).await?; + put_bucket_replication(&env_b, bucket_b, &arn_bc).await?; + + // B's own native object flows onward through the live path. + let native_key = "native-control.txt"; + let native_payload = "native control payload"; + client_b + .put_object() + .bucket(bucket_b) + .key(native_key) + .body(ByteStream::from_static(native_payload.as_bytes())) + .send() + .await?; + wait_for_replicated_object(&client_a, bucket_c, native_key, native_payload).await?; + + // With B's outbound proven, the inbound replica must stay put across + // multiple fast-scanner cycles. + assert_replication_key_absent(&client_a, bucket_c, replica_key, Duration::from_secs(6)).await?; + + Ok(()) +} diff --git a/rustfs/src/storage/options.rs b/rustfs/src/storage/options.rs index b1df0f4db..e7fdd7120 100644 --- a/rustfs/src/storage/options.rs +++ b/rustfs/src/storage/options.rs @@ -450,6 +450,18 @@ fn apply_replica_status_from_headers(headers: &HeaderMap, opts: &mu .is_some_and(|status| status.eq_ignore_ascii_case(ReplicationStatusType::Replica.as_str())) { opts.set_replica_status(ReplicationStatusType::Replica); + // Persist REPLICA into the object metadata as well (mirrors the + // Snowball inbound path). The read side derives + // ObjectInfo::replication_status from this key, and must_replicate's + // anti-loop guard consumes it — without it, HEAD/GET report no + // status and the scanner's existing-object pass cascades the replica + // onward. `opts.delete_replication` alone only reaches delete flows. + opts.user_defined + .retain(|key, _| !key.eq_ignore_ascii_case(AMZ_BUCKET_REPLICATION_STATUS)); + opts.user_defined.insert( + AMZ_BUCKET_REPLICATION_STATUS.to_string(), + ReplicationStatusType::Replica.as_str().to_string(), + ); } } @@ -1479,10 +1491,25 @@ mod tests { let opts = put_opts_from_headers(&headers, HashMap::new()).expect("replica status header should be ignored"); assert_eq!(opts.delete_marker_replication_status(), ReplicationStatusType::Empty); + assert!( + !opts + .user_defined + .keys() + .any(|key| key.eq_ignore_ascii_case(AMZ_BUCKET_REPLICATION_STATUS)), + "an unauthorized client must not forge a persisted REPLICA status" + ); let authorized = put_opts_from_headers_with_replication_authorization(&headers, HashMap::new(), true) .expect("authorized replica status header should parse"); assert_eq!(authorized.delete_marker_replication_status(), ReplicationStatusType::Replica); + // The status must reach the object metadata: the read side derives + // ObjectInfo::replication_status from this key and the scanner's + // anti-cascade guard consumes it. + assert_eq!( + authorized.user_defined.get(AMZ_BUCKET_REPLICATION_STATUS).map(String::as_str), + Some(ReplicationStatusType::Replica.as_str()), + "authorized inbound replication must persist REPLICA into object metadata" + ); } #[tokio::test] @@ -1522,10 +1549,27 @@ mod tests { let opts = get_complete_multipart_upload_opts(&headers).expect("replica status header should be ignored"); assert_eq!(opts.delete_marker_replication_status(), ReplicationStatusType::Empty); + assert!( + !opts + .user_defined + .keys() + .any(|key| key.eq_ignore_ascii_case(AMZ_BUCKET_REPLICATION_STATUS)), + "an unauthorized client must not forge a persisted REPLICA status" + ); let authorized = get_complete_multipart_upload_opts_with_replication_authorization(&headers, true) .expect("authorized replica status header should parse"); assert_eq!(authorized.delete_marker_replication_status(), ReplicationStatusType::Replica); + // For multipart the on-disk stamp actually comes from the SAME header + // at initiate time (create-multipart builds its options through + // put_opts_with_replication_authorization and persists user_defined + // into the upload's metadata); the completion-time insert asserted + // here feeds the anti-cascade must_replicate check on completion. + assert_eq!( + authorized.user_defined.get(AMZ_BUCKET_REPLICATION_STATUS).map(String::as_str), + Some(ReplicationStatusType::Replica.as_str()), + "authorized inbound multipart completion must carry REPLICA in its metadata" + ); } #[test]