fix(replication): schedule replication for CopyObject and snowball extracted objects (#5753)

* test(replication): expect CopyObject and snowball extract to schedule replication

Red-phase TDD tests for P0-6: CopyObject never consults the bucket
replication config (no pending stamp, no schedule, and the destination
inherits the source's stale replication status metadata wholesale), and
snowball auto-extract members are never scheduled either.

- usecase white-box: observe MUST_REPLICATE_OBJECT_CALLS for
  execute_copy_object (currently 0, must be 1) and
  execute_put_object_extract (currently 0, must be 2 for a two-member
  archive), plus stale replication-status metadata cleanup assertions
  (MinIO filterReplicationStatusMetadata parity).
- e2e: CopyObject destination and snowball-extracted members must appear
  on the remote replication target and reach COMPLETED on the source.

Red evidence (before fix):
  copy_object_computes_replication_decision_and_strips_stale_status
    assertion failed: left: 0, right: 1
  put_object_extract_computes_replication_decision_per_entry
    assertion failed: left: 0, right: 2

* fix(replication): schedule replication for CopyObject and snowball extracted objects

CopyObject and snowball auto-extract never consulted the bucket
replication config: no PENDING stamp, no post-commit schedule, and no
scanner-heal backstop (heal only re-drives Pending/Failed objects, and
these objects carried no status at all). Worse, the copy path cloned the
source metadata wholesale, so a destination object inherited the
source's replication bookkeeping and could present a fake
COMPLETED/REPLICA state.

Mirroring the PUT path (single immutable decision drives both the
pending metadata and the post-commit schedule, rustfs/backlog#1320):

- execute_copy_object: strip the source's replication status metadata
  (internal replication/replica status + timestamps under both
  compatibility prefixes, plus x-amz-replication-status) for
  non-inbound requests — MinIO filterReplicationStatusMetadata parity;
  the cleanup runs before the decision so an inherited REPLICA status
  cannot suppress it. Then compute must_replicate_object once, stamp
  PENDING when it replicates, and schedule after the copy commits and
  the self-copy lock guard is released. Inbound replica writes keep
  their authorized metadata and are declined inside
  must_replicate_object, so replicas are never re-scheduled outbound.

- execute_put_object_extract: same stamp + schedule per extracted
  member object (MinIO PutObjectExtract parity).

- execute_put_object dispatch: an authorized inbound replication PUT is
  stored verbatim instead of being re-dispatched into the extract path.
  Extracted members keep x-amz-meta-snowball-auto-extract in their user
  metadata and the replication client replays stored metadata as
  headers, so the target used to try to untar each member's own bytes,
  permanently failing replication for non-archive members (surfaced by
  the new snowball e2e test).

Green evidence:
- copy_object_computes_replication_decision_and_strips_stale_status,
  put_object_extract_computes_replication_decision_per_entry (red: 0
  decisions; green: 1 and 2), plus the existing PUT/object-lock
  decision-count tests stay green.
- e2e test_copy_object_replicates_to_target and
  test_snowball_extract_replicates_members_to_target pass against two
  live instances.

crates/e2e_test/src/replication_extension_test.rs

@@ -2436,6 +2436,107 @@ async fn build_replication_pair(
     Ok((source_env, target_env, source_bucket.to_string()))
 }
 
+/// P0-6: CopyObject creates a new object on the destination key, so it must be
+/// scheduled for bucket replication exactly like PutObject (MinIO
+/// CopyObjectHandler parity). Before the fix the copy path never consulted the
+/// replication config: the destination object stayed local forever (its status
+/// metadata was inherited wholesale from the source, so the scanner heal pass
+/// skipped it too — no PENDING/FAILED marker meant nothing to re-drive).
+#[tokio::test]
+#[serial]
+async fn test_copy_object_replicates_to_target() -> TestResult {
+    init_logging();
+
+    let (source_env, target_env, source_bucket) = build_replication_pair(true).await?;
+    let source_client = source_env.create_s3_client();
+    let target_client = target_env.create_s3_client();
+    let target_bucket = "replication-check-dst";
+
+    let src_key = "copy-repl-source.txt";
+    let dst_key = "copy-repl-destination.txt";
+    let payload = b"copy object replication payload".to_vec();
+
+    source_client
+        .put_object()
+        .bucket(&source_bucket)
+        .key(src_key)
+        .body(ByteStream::from(payload.clone()))
+        .send()
+        .await?;
+    assert_eq!(wait_for_object_on_target(&target_client, target_bucket, src_key).await?, payload);
+    // Wait for the source object's terminal COMPLETED status so the copy below
+    // starts from metadata that carries a stale terminal replication state; the
+    // copy must not inherit it (MinIO filterReplicationStatusMetadata parity)
+    // and must drive its own PENDING -> COMPLETED cycle.
+    wait_for_source_replication_status(&source_client, &source_bucket, src_key, "COMPLETED", false).await?;
+
+    source_client
+        .copy_object()
+        .bucket(&source_bucket)
+        .key(dst_key)
+        .copy_source(format!("{source_bucket}/{src_key}"))
+        .send()
+        .await?;
+
+    assert_eq!(
+        wait_for_object_on_target(&target_client, target_bucket, dst_key).await?,
+        payload,
+        "CopyObject destination must replicate to the remote target"
+    );
+    wait_for_source_replication_status(&source_client, &source_bucket, dst_key, "COMPLETED", false).await?;
+
+    Ok(())
+}
+
+/// P0-6 companion: snowball auto-extract writes each archive member as an
+/// independent object; every member must replicate to the remote target like a
+/// regular PUT (MinIO PutObjectExtract parity).
+#[tokio::test]
+#[serial]
+async fn test_snowball_extract_replicates_members_to_target() -> TestResult {
+    init_logging();
+
+    let (source_env, target_env, source_bucket) = build_replication_pair(true).await?;
+    let source_client = source_env.create_s3_client();
+    let target_client = target_env.create_s3_client();
+    let target_bucket = "replication-check-dst";
+
+    let members: [(&str, &[u8]); 2] = [
+        ("snowball/member-one.txt", b"first member payload"),
+        ("snowball/member-two.txt", b"second member payload"),
+    ];
+
+    let mut builder = tokio_tar::Builder::new(std::io::Cursor::new(Vec::new()));
+    for (path, data) in members {
+        let mut header = tokio_tar::Header::new_gnu();
+        header.set_size(data.len() as u64);
+        header.set_mode(0o644);
+        header.set_cksum();
+        builder.append_data(&mut header, path, std::io::Cursor::new(data)).await?;
+    }
+    let archive = builder.into_inner().await?.into_inner();
+
+    source_client
+        .put_object()
+        .bucket(&source_bucket)
+        .key("members.tar")
+        .metadata("Snowball-Auto-Extract", "true")
+        .body(ByteStream::from(archive))
+        .send()
+        .await?;
+
+    for (key, data) in members {
+        assert_eq!(
+            wait_for_object_on_target(&target_client, target_bucket, key).await?,
+            data,
+            "snowball-extracted member {key} must replicate to the remote target"
+        );
+        wait_for_source_replication_status(&source_client, &source_bucket, key, "COMPLETED", false).await?;
+    }
+
+    Ok(())
+}
+
 #[tokio::test]
 #[serial]
 async fn test_replication_check_succeeds_with_remote_target() -> Result<(), Box<dyn Error + Send + Sync>> {

rustfs/src/app/lifecycle_transition_api_test.rs

@@ -2430,6 +2430,192 @@ async fn put_object_computes_replication_decision_exactly_once() {
     );
 }
 
+/// CopyObject creates a new object on the destination key, so it must join
+/// bucket replication exactly like PutObject: compute one replication decision
+/// that drives both the persisted pending marker and the post-commit schedule
+/// (MinIO CopyObjectHandler parity). A count of 0 means a copied object is
+/// never scheduled for replication and — because the copy path clones the
+/// source metadata wholesale — the destination silently inherits the source's
+/// replication bookkeeping, faking a COMPLETED/REPLICA state for an object
+/// that never replicated. The second half of this test pins the metadata
+/// cleanup (MinIO filterReplicationStatusMetadata parity).
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[serial]
+#[ignore = "global-state usecase integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
+async fn copy_object_computes_replication_decision_and_strips_stale_status() {
+    use super::storage_api::object_usecase::bucket::replication::MUST_REPLICATE_OBJECT_CALLS;
+    use rustfs_utils::http::{
+        SUFFIX_REPLICA_STATUS, SUFFIX_REPLICA_TIMESTAMP, SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, get_str,
+        insert_str,
+    };
+    use std::collections::HashMap;
+    use std::sync::atomic::Ordering;
+
+    let (_disk_paths, ecstore) = setup_test_env().await;
+    let usecase = DefaultObjectUsecase::from_global();
+
+    let bucket = format!("test-copy-repl-{}", &Uuid::new_v4().simple().to_string()[..8]);
+    let src_object = "copy/source.txt";
+    let dst_object = "copy/destination.txt";
+    let payload = b"copy replication decision payload";
+
+    create_test_bucket(&ecstore, bucket.as_str()).await;
+
+    // Seed the source object with stale replication bookkeeping, as if it had
+    // already replicated elsewhere (internal status/timestamp under both
+    // compatibility prefixes, replica state, and the surfaced x-amz header).
+    let mut stale_metadata = HashMap::new();
+    insert_str(
+        &mut stale_metadata,
+        SUFFIX_REPLICATION_STATUS,
+        "arn:minio:replication::stale:dst=COMPLETED;".to_string(),
+    );
+    insert_str(&mut stale_metadata, SUFFIX_REPLICATION_TIMESTAMP, "2024-01-01T00:00:00Z".to_string());
+    insert_str(&mut stale_metadata, SUFFIX_REPLICA_STATUS, "REPLICA".to_string());
+    insert_str(&mut stale_metadata, SUFFIX_REPLICA_TIMESTAMP, "2024-01-01T00:00:00Z".to_string());
+    stale_metadata.insert(AMZ_BUCKET_REPLICATION_STATUS.to_string(), "COMPLETED".to_string());
+    let mut reader = PutObjReader::from_vec(payload.to_vec());
+    (*ecstore)
+        .put_object(
+            bucket.as_str(),
+            src_object,
+            &mut reader,
+            &ObjectOptions {
+                user_defined: stale_metadata,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("Failed to upload source object with stale replication metadata");
+
+    MUST_REPLICATE_OBJECT_CALLS.store(0, Ordering::SeqCst);
+
+    let copy_input = CopyObjectInput::builder()
+        .copy_source(CopySource::Bucket {
+            bucket: bucket.clone().into(),
+            key: src_object.into(),
+            version_id: None,
+        })
+        .bucket(bucket.clone())
+        .key(dst_object.to_string())
+        .build()
+        .unwrap();
+    Box::pin(usecase.execute_copy_object(build_request(copy_input, Method::PUT)))
+        .await
+        .expect("Failed to copy object through usecase");
+
+    assert_eq!(
+        MUST_REPLICATE_OBJECT_CALLS.load(Ordering::SeqCst),
+        1,
+        "CopyObject must compute the replication decision exactly once; 0 means the copied \
+         object never enters bucket replication (P0-6, MinIO CopyObjectHandler parity)"
+    );
+
+    let copied = ecstore
+        .get_object_info(bucket.as_str(), dst_object, &ObjectOptions::default())
+        .await
+        .expect("Failed to read copied destination object info");
+    assert!(
+        get_str(&copied.user_defined, SUFFIX_REPLICATION_STATUS).is_none(),
+        "destination must not inherit the source's replication status; got {:?}",
+        get_str(&copied.user_defined, SUFFIX_REPLICATION_STATUS)
+    );
+    assert!(
+        get_str(&copied.user_defined, SUFFIX_REPLICATION_TIMESTAMP).is_none(),
+        "destination must not inherit the source's replication timestamp"
+    );
+    assert!(
+        get_str(&copied.user_defined, SUFFIX_REPLICA_STATUS).is_none(),
+        "destination must not inherit the source's replica status"
+    );
+    assert!(
+        !copied
+            .user_defined
+            .keys()
+            .any(|k| k.eq_ignore_ascii_case(AMZ_BUCKET_REPLICATION_STATUS)),
+        "destination must not inherit the surfaced x-amz-replication-status header key"
+    );
+}
+
+/// Snowball auto-extract writes each archive member as an independent object,
+/// so each member must join bucket replication like a regular PUT (MinIO
+/// PutObjectExtract parity). The archive holds two file entries; a count other
+/// than 2 means extracted objects bypass the replication decision entirely and
+/// stay local forever.
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[serial]
+#[ignore = "global-state usecase integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
+async fn put_object_extract_computes_replication_decision_per_entry() {
+    use super::storage_api::object_usecase::bucket::replication::MUST_REPLICATE_OBJECT_CALLS;
+    use super::storage_api::test::ReqInfo;
+    use super::storage_api::test::bucket::metadata::BUCKET_POLICY_CONFIG;
+    use std::io::Cursor;
+    use std::sync::atomic::Ordering;
+
+    let (_disk_paths, ecstore) = setup_test_env().await;
+    let usecase = DefaultObjectUsecase::from_global();
+
+    let bucket = format!("test-extract-repl-{}", &Uuid::new_v4().simple().to_string()[..8]);
+    create_test_bucket(&ecstore, bucket.as_str()).await;
+
+    // The extract path re-authorizes every extracted entry internally. The test
+    // harness has no IAM system, so allow anonymous PutObject via bucket policy.
+    let policy_json = serde_json::json!({
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Effect": "Allow",
+                "Principal": "*",
+                "Action": ["s3:PutObject"],
+                "Resource": [format!("arn:aws:s3:::{bucket}/*")]
+            }
+        ]
+    })
+    .to_string();
+    metadata_sys::update(bucket.as_str(), BUCKET_POLICY_CONFIG, policy_json.into_bytes())
+        .await
+        .expect("Failed to install anonymous put bucket policy");
+
+    let mut builder = tokio_tar::Builder::new(Cursor::new(Vec::new()));
+    for (path, data) in [
+        ("member-one.txt", b"first member payload".as_slice()),
+        ("nested/member-two.txt", b"second member payload".as_slice()),
+    ] {
+        let mut header = tokio_tar::Header::new_gnu();
+        header.set_size(data.len() as u64);
+        header.set_mode(0o644);
+        header.set_cksum();
+        builder
+            .append_data(&mut header, path, Cursor::new(data))
+            .await
+            .expect("Failed to append archive member");
+    }
+    let archive = builder.into_inner().await.expect("Failed to finish archive").into_inner();
+
+    MUST_REPLICATE_OBJECT_CALLS.store(0, Ordering::SeqCst);
+
+    let input = PutObjectInput::builder()
+        .bucket(bucket.clone())
+        .key("archive.tar".to_string())
+        .body(Some(streaming_blob_from_bytes(&archive)))
+        .content_length(Some(archive.len() as i64))
+        .build()
+        .unwrap();
+    let mut req = build_request(input, Method::PUT);
+    req.extensions.insert(ReqInfo::default());
+    Box::pin(usecase.execute_put_object_extract(req))
+        .await
+        .expect("Failed to extract archive through usecase");
+
+    assert_eq!(
+        MUST_REPLICATE_OBJECT_CALLS.load(Ordering::SeqCst),
+        2,
+        "snowball auto-extract must compute one replication decision per extracted member \
+         object; 0 means extracted objects bypass bucket replication (P0-6 companion, MinIO \
+         PutObjectExtract parity)"
+    );
+}
+
 /// The object-lock handlers do not go through the object PUT path, so they must schedule
 /// replication themselves. Without it a retention or legal hold applied after upload stays
 /// local and the replica keeps its previous, unprotected state (a WORM object that is still

rustfs/src/app/object_usecase.rs

@@ -136,8 +136,9 @@ use rustfs_targets::{EventName, get_request_host, get_request_port, get_request_
 use rustfs_utils::CompressionAlgorithm;
 use rustfs_utils::http::{
     AMZ_BUCKET_REPLICATION_STATUS, AMZ_CHECKSUM_MODE, AMZ_CHECKSUM_TYPE, AMZ_WEBSITE_REDIRECT_LOCATION, CONTENT_TYPE,
-    SUFFIX_ACTUAL_SIZE, SUFFIX_COMPRESSION, SUFFIX_COMPRESSION_SIZE, SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP,
-    SUFFIX_RESTORE_OPERATION_ID,
+    SUFFIX_ACTUAL_SIZE, SUFFIX_COMPRESSION, SUFFIX_COMPRESSION_SIZE, SUFFIX_REPLICA_STATUS, SUFFIX_REPLICA_TIMESTAMP,
+    SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, SUFFIX_RESTORE_OPERATION_ID, SUFFIX_SOURCE_REPLICATION_REQUEST,
+    get_header,
     headers::{
         AMZ_CONTENT_SHA256, AMZ_DECODED_CONTENT_LENGTH, AMZ_MINIO_SNOWBALL_IGNORE_DIRS, AMZ_MINIO_SNOWBALL_IGNORE_ERRORS,
         AMZ_MINIO_SNOWBALL_PREFIX, AMZ_OBJECT_LOCK_LEGAL_HOLD, AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE,
@@ -4828,7 +4829,15 @@ impl DefaultObjectUsecase {
         {
             return Err(s3_error!(InvalidStorageClass));
         }
-        if is_put_object_extract_requested(&req.headers) {
+        // An authorized inbound replication PUT must store the replica verbatim.
+        // A snowball-extracted member object keeps `x-amz-meta-snowball-auto-extract`
+        // in its user metadata, and the replication client replays stored metadata
+        // as headers — re-dispatching that PUT into the extract path would try to
+        // untar the member's own bytes (failing replication for any non-archive
+        // member) instead of writing the replica.
+        let inbound_replication_put = replication_request_authorized(&req)
+            && get_header(&req.headers, SUFFIX_SOURCE_REPLICATION_REQUEST).as_deref() == Some("true");
+        if is_put_object_extract_requested(&req.headers) && !inbound_replication_put {
             return Box::pin(self.execute_put_object_extract(req)).await;
         }
 
@@ -6723,6 +6732,43 @@ impl DefaultObjectUsecase {
             user_defined.insert(k, v);
         }
 
+        // The source object's replication bookkeeping (internal status/timestamp,
+        // replica state, and the surfaced x-amz-replication-status) describes the
+        // SOURCE's replication history; carried onto the destination it fakes a
+        // COMPLETED/REPLICA state for an object that never replicated (MinIO
+        // filterReplicationStatusMetadata parity). Inbound replica writes are
+        // exempt: the authorized replication request owns these keys (see
+        // copy_dst_opts_with_replication_authorization above).
+        if !dst_opts.replication_request {
+            user_defined.retain(|k, _| !k.eq_ignore_ascii_case(AMZ_BUCKET_REPLICATION_STATUS));
+            remove_str(&mut user_defined, SUFFIX_REPLICATION_STATUS);
+            remove_str(&mut user_defined, SUFFIX_REPLICATION_TIMESTAMP);
+            remove_str(&mut user_defined, SUFFIX_REPLICA_STATUS);
+            remove_str(&mut user_defined, SUFFIX_REPLICA_TIMESTAMP);
+        }
+
+        // Compute the replication decision exactly once per copy. The same
+        // immutable `dsc` drives both the pending metadata written below and the
+        // post-commit schedule (see the reuse site after copy_object), so a
+        // replication-config hot update cannot split the two phases — same
+        // contract as the PUT path (https://github.com/rustfs/backlog/issues/1320).
+        // `must_replicate_object` itself declines inbound replica writes
+        // (replication_request / REPLICA status), so replicas are never
+        // re-scheduled outbound.
+        let dsc = must_replicate_object(
+            &bucket,
+            &key,
+            &user_defined,
+            "".to_string(),
+            dst_opts.delete_marker_replication_status(),
+            dst_opts.clone(),
+        )
+        .await;
+        if dsc.replicate_any() {
+            insert_str(&mut user_defined, SUFFIX_REPLICATION_TIMESTAMP, jiff::Zoned::now().to_string());
+            insert_str(&mut user_defined, SUFFIX_REPLICATION_STATUS, dsc.pending_status().unwrap_or_default());
+        }
+
         src_info.user_defined = Arc::new(user_defined);
 
         self.check_bucket_quota(&bucket, QuotaOperation::CopyObject, src_info.size as u64)
@@ -6737,6 +6783,13 @@ impl DefaultObjectUsecase {
             .map_err(ApiError::from)?;
         drop(_self_copy_lock_guard);
 
+        // Reuse the single pre-commit replication decision (see `dsc` above) so
+        // the persisted pending marker and the schedule always agree, mirroring
+        // the PUT path.
+        if dsc.replicate_any() {
+            schedule_object_replication(oi.clone(), store.clone(), dsc).await;
+        }
+
         maybe_enqueue_transition_immediate(&oi, LcEventSrc::S3CopyObject).await;
         let _ = invalidate_object_data_cache_after_copy_success(&cache_adapter, &bucket, &key).await;
 
@@ -8607,6 +8660,31 @@ impl DefaultObjectUsecase {
             }
             hrd = write_plan.apply(hrd, actual_size).map_err(ApiError::from)?;
             opts.user_defined.extend(metadata);
+
+            // Each extracted member is an independent user write and joins
+            // bucket replication like a regular PUT (MinIO PutObjectExtract
+            // parity). One immutable decision drives both the pending metadata
+            // and the post-commit schedule below, same contract as the PUT path
+            // (https://github.com/rustfs/backlog/issues/1320); inbound replica
+            // writes are declined inside `must_replicate_object`.
+            let dsc = must_replicate_object(
+                &bucket,
+                &fpath,
+                &opts.user_defined,
+                "".to_string(),
+                opts.delete_marker_replication_status(),
+                opts.clone(),
+            )
+            .await;
+            if dsc.replicate_any() {
+                insert_str(&mut opts.user_defined, SUFFIX_REPLICATION_TIMESTAMP, jiff::Zoned::now().to_string());
+                insert_str(
+                    &mut opts.user_defined,
+                    SUFFIX_REPLICATION_STATUS,
+                    dsc.pending_status().unwrap_or_default(),
+                );
+            }
+
             let mut reader = PutObjReader::new(hrd);
             let cache_adapter = self.object_data_cache();
             let _ = invalidate_object_data_cache_before_mutation(&cache_adapter, &bucket, &fpath).await;
@@ -8640,6 +8718,13 @@ impl DefaultObjectUsecase {
                 }
             }
             let _ = invalidate_object_data_cache_after_put_success(&cache_adapter, &bucket, &fpath).await;
+
+            // Reuse the per-entry pre-commit decision (see `dsc` above) so the
+            // persisted pending marker and the schedule always agree.
+            if dsc.replicate_any() {
+                schedule_object_replication(obj_info.clone(), store.clone(), dsc).await;
+            }
+
             if !wrote_any_entry {
                 rustfs_scanner::record_dirty_usage_bucket(&bucket);
                 wrote_any_entry = true;