diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index e2c2dc2fa..f0f467927 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/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> { diff --git a/rustfs/src/app/lifecycle_transition_api_test.rs b/rustfs/src/app/lifecycle_transition_api_test.rs index a4e58667d..bdf50a5e3 100644 --- a/rustfs/src/app/lifecycle_transition_api_test.rs +++ b/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 diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 8226bc33b..a9e9126b3 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/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;