From a076ae404552055cc19b132a911de63c5f14922d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Tue, 11 Aug 2026 11:58:25 +0800 Subject: [PATCH] test(replication): pin the scanner existing-object compensation matrix (#5877) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P1-20 (rustfs/backlog#1675 B2, test-only). No prior test wrote objects BEFORE the replication rule arrived, leaving the scanner's existing-object resync pass — the only channel for such objects — without end-to-end coverage, and the enqueue truth table partially unpinned at unit level. e2e (both negative cells are contracts, asserted over multiple fast-scanner cycles next to a replicated control key that proves the scanner and the live path are running): - test_scanner_compensates_existing_objects_across_write_paths: plain PUT, CopyObject and Snowball auto-extract products written pre-rule all converge via scanner compensation; a null-version object (PUT before the bucket became versioned) is pinned as never compensated (the scanner heal gate skips nil-version objects). - test_scanner_never_compensates_when_existing_object_replication_disabled: ExistingObjectReplication=Disabled is a contract, not a delay — existing keys stay absent while post-rule writes replicate normally. Unit truth-table pins (crates/replication): - queue.rs: an empty replicate decision (Disabled existing-object, inbound REPLICA) skips heal queueing for every status; Completed without a resync decision skips. - operation.rs: existing-object resync without a reset replicates exactly the never-replicated (Empty) objects. Helper: put_bucket_replication_with_statuses parameterizes the previously hardcoded ExistingObjectReplication status; the nextest count comments are refreshed to the post-rebase totals. --- .config/nextest.toml | 4 +- .../src/replication_extension_test.rs | 247 +++++++++++++++++- crates/replication/src/operation.rs | 32 +++ crates/replication/src/queue.rs | 42 +++ 4 files changed, 322 insertions(+), 3 deletions(-) diff --git a/.config/nextest.toml b/.config/nextest.toml index 615c4c4fe..fdb877939 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -218,7 +218,7 @@ test-group = 'ecstore-serial-flaky' # the nightly profile derives its set as "the replication module MINUS this # allowlist", so any new replication test lands in nightly by default (never # silently unrun) until it is explicitly blessed as fast here. Keep the two -# regexes byte-identical. Count invariant: 20 here + 47 nightly = 67 total +# regexes byte-identical. Count invariant: 20 here + 49 nightly = 69 total # (authority: `cargo nextest list`; docs/testing/e2e-suite-inventory.md). # HISTORY (2026-07-11): the 20 fast tests were briefly pulled out of this lane # (#4724) because they set a loopback (127.0.0.1) replication target that the @@ -344,7 +344,7 @@ path = "junit.xml" # object_lambda) — too heavy for the merge budget; they run in ci-7's # nightly 4-node lane. # * replication_extension_test — repl-1 already splits it into the PR -# `e2e-smoke` (20 fast) and `e2e-repl-nightly` (47 slow) lanes and reserves +# `e2e-smoke` (20 fast) and `e2e-repl-nightly` (49 slow) lanes and reserves # it for those, so e2e-full does not double-run it. # * #[ignore]d tests — nextest skips them by default (no --run-ignored); the # manual-localhost:9000 reliant/policy tests are ci-13's migration. diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index bb02eb604..2734295f3 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -629,6 +629,17 @@ async fn put_bucket_replication_with_delete_statuses( target_arn: &str, delete_marker_status: &str, version_delete_status: Option<&str>, +) -> Result<(), Box> { + put_bucket_replication_with_statuses(env, bucket, target_arn, delete_marker_status, version_delete_status, "Enabled").await +} + +async fn put_bucket_replication_with_statuses( + env: &RustFSTestEnvironment, + bucket: &str, + target_arn: &str, + delete_marker_status: &str, + version_delete_status: Option<&str>, + existing_object_status: &str, ) -> Result<(), Box> { let delete_replication = version_delete_status .map(|status| format!("{status}")) @@ -645,7 +656,7 @@ async fn put_bucket_replication_with_delete_statuses( {delete_replication} - Enabled + {existing_object_status} {target_arn} @@ -8149,5 +8160,239 @@ async fn test_delayed_delete_marker_purge_exhaustion_persists_to_mrf_and_replays ); target.shutdown().await; + + Ok(()) +} + +// --- P1-20 (backlog#1675): scanner existing-object compensation matrix --- +// +// Every case below inverts the order used by the rest of this file: objects +// are written FIRST and the replication rule arrives afterwards, so the only +// channel that can move the pre-existing objects is the data scanner's +// existing-object resync pass. Negative cells ("never compensated") are +// contracts and are asserted over multiple scanner cycles, always next to a +// replicated control key that proves the scanner and the live path are +// running — an absent key on a dead scanner proves nothing. + +/// Envs + buckets only: versioning, the remote target, and the rule variant +/// are wired by each test (the null-version case must PUT before the source +/// bucket becomes versioned). The source runs with FAST_SCANNER_ENV so +/// existing keys are rescanned within seconds instead of 16 dir cycles. +async fn build_scanner_compensation_pair( + source_bucket: &str, + target_bucket: &str, +) -> Result<(RustFSTestEnvironment, RustFSTestEnvironment), Box> { + let mut source_env = RustFSTestEnvironment::new().await?; + let mut source_process_env = replication_fast_env(); + source_process_env.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV); + source_process_env.extend_from_slice(FAST_SCANNER_ENV); + source_env.start_rustfs_server_with_env(vec![], &source_process_env).await?; + + let mut target_env = RustFSTestEnvironment::new().await?; + target_env.start_rustfs_server_without_cleanup(vec![]).await?; + + source_env + .create_s3_client() + .create_bucket() + .bucket(source_bucket) + .send() + .await?; + target_env + .create_s3_client() + .create_bucket() + .bucket(target_bucket) + .send() + .await?; + + Ok((source_env, target_env)) +} + +/// P1-20: objects that already exist when a rule with +/// ExistingObjectReplication=Enabled arrives are compensated by the scanner's +/// existing-object resync pass, whatever wrote them — plain PUT, CopyObject, +/// or Snowball auto-extract. The pinned exception is a null-version object +/// (written before the bucket became versioned): the scanner heal gate skips +/// nil-version objects entirely (`scanner_folder.rs` heal_replication), so it +/// must NEVER be compensated. +#[tokio::test] +#[serial] +async fn test_scanner_compensates_existing_objects_across_write_paths() -> TestResult { + init_logging(); + let source_bucket = "scanner-comp-src"; + let target_bucket = "scanner-comp-dst"; + let (source_env, target_env) = build_scanner_compensation_pair(source_bucket, target_bucket).await?; + let source_client = source_env.create_s3_client(); + let target_client = target_env.create_s3_client(); + + // Null-version cell: PUT before versioning; the object keeps the nil + // version id forever. + let null_key = "pre-versioning-null.txt"; + source_client + .put_object() + .bucket(source_bucket) + .key(null_key) + .body(ByteStream::from_static(b"null version payload")) + .send() + .await?; + + enable_bucket_versioning(&source_env, source_bucket).await?; + enable_bucket_versioning(&target_env, target_bucket).await?; + + // Pre-existing objects from three write paths, all before any replication + // config exists (their replication status stays Empty). + let plain_key = "existing-plain.txt"; + let plain_payload = "existing plain payload"; + source_client + .put_object() + .bucket(source_bucket) + .key(plain_key) + .body(ByteStream::from_static(plain_payload.as_bytes())) + .send() + .await?; + + let copy_key = "existing-copy.txt"; + source_client + .copy_object() + .bucket(source_bucket) + .key(copy_key) + .copy_source(format!("{source_bucket}/{plain_key}")) + .send() + .await?; + + let member_key = "snowball/existing-member.txt"; + let member_payload: &[u8] = b"existing snowball member payload"; + let mut builder = tokio_tar::Builder::new(std::io::Cursor::new(Vec::new())); + let mut header = tokio_tar::Header::new_gnu(); + header.set_size(member_payload.len() as u64); + header.set_mode(0o644); + header.set_cksum(); + builder + .append_data(&mut header, member_key, std::io::Cursor::new(member_payload)) + .await?; + let archive = builder.into_inner().await?.into_inner(); + source_client + .put_object() + .bucket(source_bucket) + .key("existing-members.tar") + .metadata("Snowball-Auto-Extract", "true") + .body(ByteStream::from(archive)) + .send() + .await?; + // The extracted member must exist locally before the rule arrives, or it + // would replicate through the live path instead of the scanner. + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + loop { + if source_client + .head_object() + .bucket(source_bucket) + .key(member_key) + .send() + .await + .is_ok() + { + break; + } + if tokio::time::Instant::now() >= deadline { + return Err("snowball member was never extracted on the source".into()); + } + sleep(Duration::from_millis(200)).await; + } + + // Only now wire the remote target and the Enabled rule. + let target_arn = set_replication_target(&source_env, source_bucket, &target_env, target_bucket).await?; + put_bucket_replication(&source_env, source_bucket, &target_arn).await?; + + // Control key written after the rule replicates through the live path. + let control_key = "control-live.txt"; + let control_payload = "control live payload"; + source_client + .put_object() + .bucket(source_bucket) + .key(control_key) + .body(ByteStream::from_static(control_payload.as_bytes())) + .send() + .await?; + wait_for_replicated_object(&target_client, target_bucket, control_key, control_payload).await?; + + // Scanner compensation for each pre-existing write path. + wait_for_replicated_object(&target_client, target_bucket, plain_key, plain_payload).await?; + wait_for_replicated_object(&target_client, target_bucket, copy_key, plain_payload).await?; + wait_for_replicated_object(&target_client, target_bucket, member_key, std::str::from_utf8(member_payload)?).await?; + + // Null-version contract: with every sibling compensated (scanner proven + // live), the nil-version object must stay absent across further cycles. + assert_replication_key_absent(&target_client, target_bucket, null_key, Duration::from_secs(6)).await?; + + Ok(()) +} + +/// P1-20: ExistingObjectReplication=Disabled is a contract, not a delay — the +/// scanner must NEVER compensate objects that predate the rule, while objects +/// written after the rule replicate normally (the setting only gates the +/// existing-object resync path). +#[tokio::test] +#[serial] +async fn test_scanner_never_compensates_when_existing_object_replication_disabled() -> TestResult { + init_logging(); + let source_bucket = "scanner-disabled-src"; + let target_bucket = "scanner-disabled-dst"; + let (source_env, mut target_env) = build_scanner_compensation_pair(source_bucket, target_bucket).await?; + let source_client = source_env.create_s3_client(); + let target_client = target_env.create_s3_client(); + + enable_bucket_versioning(&source_env, source_bucket).await?; + enable_bucket_versioning(&target_env, target_bucket).await?; + + let existing_key = "existing-disabled.txt"; + source_client + .put_object() + .bucket(source_bucket) + .key(existing_key) + .body(ByteStream::from_static(b"existing disabled payload")) + .send() + .await?; + + let target_arn = set_replication_target(&source_env, source_bucket, &target_env, target_bucket).await?; + put_bucket_replication_with_statuses(&source_env, source_bucket, &target_arn, "Enabled", None, "Disabled").await?; + + // The live path is unaffected by the Disabled existing-object setting. + let control_key = "control-live.txt"; + let control_payload = "control live payload"; + source_client + .put_object() + .bucket(source_bucket) + .key(control_key) + .body(ByteStream::from_static(control_payload.as_bytes())) + .send() + .await?; + wait_for_replicated_object(&target_client, target_bucket, control_key, control_payload).await?; + + // Scanner-only witness. A live-path control key alone would let this test + // pass while the existing-object scanner is disabled or wedged, so make + // the scanner itself observable: an object whose replication FAILED while + // the target was down can only be re-driven by the data scanner's + // replication heal pass (see FAST_SCANNER_ENV), and that pass is NOT + // gated by ExistingObjectReplication. The witness lives in the same + // bucket and prefix as the pre-existing key, so a heal pass that reached + // it necessarily walked the pre-existing key in the same scan. + let witness_key = "scanner-witness.txt"; + let witness_payload = "scanner witness payload"; + target_env.stop_server(); + source_client + .put_object() + .bucket(source_bucket) + .key(witness_key) + .body(ByteStream::from_static(witness_payload.as_bytes())) + .send() + .await?; + wait_for_source_replication_status(&source_client, source_bucket, witness_key, "FAILED", false).await?; + target_env.restart_server_preserving_data(vec![], &[]).await?; + let target_client = target_env.create_s3_client(); + wait_for_replicated_object(&target_client, target_bucket, witness_key, witness_payload).await?; + + // The scanner demonstrably swept this bucket; the pre-existing key must + // still be absent, and stay absent over further cycles. + assert_replication_key_absent(&target_client, target_bucket, existing_key, Duration::from_secs(6)).await?; + Ok(()) } diff --git a/crates/replication/src/operation.rs b/crates/replication/src/operation.rs index b3476f771..a905ed992 100644 --- a/crates/replication/src/operation.rs +++ b/crates/replication/src/operation.rs @@ -635,6 +635,38 @@ mod tests { assert!(!should_use_existing_delete_replication_info(false, false)); } + /// P1-20 truth-table pin (rustfs/backlog#1675): without any reset in play + /// (no per-target reset header on the object, empty reset id on the + /// target) the existing-object resync decision compensates exactly the + /// never-replicated objects — Empty replicates, any recorded status does + /// not. + #[test] + fn resync_target_without_reset_replicates_only_empty_status() { + let user_defined = HashMap::new(); + let object = ReplicationResyncTargetObject { + mod_time: Some(OffsetDateTime::UNIX_EPOCH + Duration::seconds(10)), + user_defined: &user_defined, + }; + + for (status, expected) in [ + (ReplicationStatusType::Empty, true), + (ReplicationStatusType::Completed, false), + // "COMPLETE" on disk parses to this legacy variant, so objects + // written by older versions reach the decision through it. + (ReplicationStatusType::CompletedLegacy, false), + (ReplicationStatusType::Pending, false), + (ReplicationStatusType::Failed, false), + (ReplicationStatusType::Replica, false), + ] { + let label = format!("{status:?}"); + let decision = resync_target_for_object(&object, "arn:target", "", None, status); + assert_eq!( + decision.replicate, expected, + "existing-object resync without a reset must replicate only never-replicated objects (status {label})" + ); + } + } + #[test] fn resync_target_includes_object_at_reset_before_boundary() { let reset_before = OffsetDateTime::UNIX_EPOCH + Duration::seconds(30); diff --git a/crates/replication/src/queue.rs b/crates/replication/src/queue.rs index b652a6237..7e030bbde 100644 --- a/crates/replication/src/queue.rs +++ b/crates/replication/src/queue.rs @@ -360,6 +360,48 @@ mod tests { ); } + /// P1-20 truth-table pin (rustfs/backlog#1675): when no target replicates + /// — the decision is empty because ExistingObjectReplication is Disabled + /// for a never-replicated object, or because the object is an inbound + /// REPLICA (must_replicate returns an empty decision for those) — the + /// heal pass must skip entirely, whatever the recorded status says. The + /// scanner never compensates these objects. + #[test] + fn heal_queue_action_skips_when_no_target_replicates() { + for status in [ + ReplicationStatusType::Empty, + ReplicationStatusType::Failed, + ReplicationStatusType::Replica, + ] { + let mut roi = ReplicateObjectInfo { + bucket: "bucket".to_string(), + name: "object".to_string(), + replication_status: status, + dsc: ReplicateDecision::new(), + ..Default::default() + }; + + let action = replication_heal_queue_action(&mut roi); + + assert!( + matches!(action, ReplicationHealQueueAction::Skip), + "an empty replicate decision must skip heal queueing (status {:?})", + roi.replication_status + ); + } + } + + /// P1-20 truth-table pin: a Completed object with no resync decision has + /// nothing left to heal — the scanner must not requeue it. + #[test] + fn heal_queue_action_skips_completed_object_without_resync() { + let mut roi = replicate_object_info(ReplicationStatusType::Completed); + + let action = replication_heal_queue_action(&mut roi); + + assert!(matches!(action, ReplicationHealQueueAction::Skip)); + } + #[test] fn heal_queue_action_routes_failed_objects_to_heal_queue() { let mut roi = replicate_object_info(ReplicationStatusType::Failed);