diff --git a/.config/e2e-repl-nightly-selection.txt b/.config/e2e-repl-nightly-selection.txt index 88cbd3d94..f21b50aba 100644 --- a/.config/e2e-repl-nightly-selection.txt +++ b/.config/e2e-repl-nightly-selection.txt @@ -1 +1 @@ -sha256=fc653ffc00f85a109be82e494117f0f60930f1c105898a5758af3c1bd998b1e1 +sha256=e4ee3898b840a8374cb02c64342cc15f8d7ecd1becaf4b4a2a4657290f9e7ff1 diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index e636d9db5..c57ac9f14 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -6134,6 +6134,118 @@ async fn test_site_replication_allows_private_ca_https_with_ca_cert_pem_real_dua Ok(()) } +/// rustfs/backlog#2479: a site resync must count a replicated delete marker +/// as converged. The peer answers `HEAD ?versionId=` with a bodiless +/// 405, which the SDK surfaces without an error code; the resync worker used +/// to record that as `target service error` and fail the whole bucket. +#[tokio::test] +async fn test_site_replication_resync_replicates_delete_marker() -> Result<(), Box> { + init_logging(); + let process_env = [ + ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", "true"), + ("RUSTFS_REPL_RESYNC_POLL_MAX_MS", "100"), + ("RUST_LOG", "error"), + ]; + + let mut source_env = RustFSTestEnvironment::new().await?; + source_env.start_rustfs_server_with_env(vec![], &process_env).await?; + let mut target_env = RustFSTestEnvironment::new().await?; + target_env.start_rustfs_server_with_env(vec![], &process_env).await?; + + let bucket = "site-repl-resync-marker"; + let live_key = "live.bin"; + let gone_key = "gone.txt"; + let source_client = source_env.create_s3_client(); + let target_client = target_env.create_s3_client(); + + source_client.create_bucket().bucket(bucket).send().await?; + enable_bucket_versioning(&source_env, bucket).await?; + + let add_status = site_replication_add( + &source_env, + &[ + PeerSite { + name: "source-site".to_string(), + endpoint: source_env.url.clone(), + access_key: source_env.access_key.clone(), + secret_key: source_env.secret_key.clone(), + ..Default::default() + }, + PeerSite { + name: "target-site".to_string(), + endpoint: target_env.url.clone(), + access_key: target_env.access_key.clone(), + secret_key: target_env.secret_key.clone(), + ..Default::default() + }, + ], + ) + .await?; + assert!(add_status.success, "unexpected site add result: {:?}", add_status); + + let source_info = wait_for_site_replication_enabled(&source_env, 2).await?; + wait_for_site_replication_enabled(&target_env, 2).await?; + let remote_peer = source_info + .sites + .into_iter() + .find(|peer| peer.endpoint == target_env.url) + .ok_or("target peer missing from source site replication info")?; + wait_for_bucket_on_target(&target_client, bucket).await?; + wait_for_remote_target_arn(&source_env, bucket).await?; + + // One live object and one key whose latest version is a delete marker, + // both converged to the peer through live replication first so the + // resync re-drives objects the peer already holds. + source_client + .put_object() + .bucket(bucket) + .key(live_key) + .body(ByteStream::from(vec![b'l'; 4096])) + .send() + .await?; + source_client + .put_object() + .bucket(bucket) + .key(gone_key) + .body(ByteStream::from(vec![b'g'; 128])) + .send() + .await?; + let delete = source_client.delete_object().bucket(bucket).key(gone_key).send().await?; + assert_eq!(delete.delete_marker(), Some(true), "a versioned delete must create a delete marker"); + wait_for_object_on_target(&target_client, bucket, live_key).await?; + wait_for_target_delete_marker(&target_client, bucket, gone_key).await?; + + let started = site_replication_resync_op(&source_env, "start", &remote_peer).await?; + assert_eq!(started.status, "success", "unexpected start result: {:?}", started); + + let deadline = tokio::time::Instant::now() + Duration::from_secs(60); + let finished = loop { + let status = site_replication_resync_op(&source_env, "status", &remote_peer).await?; + match status.state.as_str() { + "completed" | "failed" => break status, + _ if tokio::time::Instant::now() < deadline => sleep(Duration::from_millis(250)).await, + _ => return Err(format!("site resync did not reach a terminal state in time: {status:?}").into()), + } + }; + let entry = finished + .buckets + .iter() + .find(|entry| entry.bucket == bucket) + .ok_or_else(|| format!("resync status lost the bucket: {finished:?}"))?; + assert_eq!( + (finished.state.as_str(), entry.status.as_str(), entry.failed_objects), + ("completed", "completed", 0), + "the delete marker must verify as replicated, not fail the bucket: {finished:?}" + ); + assert!( + entry.replicated_objects >= 2, + "the live object and the delete marker both count as replicated: {entry:?}" + ); + assert!(entry.err_detail.is_empty(), "unexpected bucket error: {entry:?}"); + + Ok(()) +} + /// rustfs/backlog#2489: an operator's bucket-level replication to a site that /// later becomes a peer keeps working through the site's add, resync and /// removal. Site replication wires its own same-name target next to the diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 6b615804c..2b680cbf0 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -1815,12 +1815,26 @@ async fn verify_resync_head_result( // (retryable/ambiguous) HEAD error leaves the outcome // unverified, so it must count as failed — not as a // blanket success (backlog#862 / #799 B13). + // A HEAD carries no body, so the SDK only synthesizes an error + // code for 404; a target answering the delete marker's version + // with 405 (RustFS, MinIO, AWS) leaves `code()` empty and the + // raw status is the only evidence (rustfs/backlog#2479). + // A marker *version purge* (`version_purge_status` set) asks + // for the opposite outcome: the marker must be gone from the + // target, so only absence verifies it and a 405 proves the + // marker is still there. let retryable = { let (is_not_found, code) = err .as_service_error() .map(|se| (se.is_not_found(), se.code())) .unwrap_or((false, None)); - is_retryable_delete_replication_head_error(is_not_found, code) + let not_found = is_not_found || has_raw_status(&err, 404); + if roi.version_purge_status.is_empty() { + let code = code.or_else(|| has_raw_status(&err, 405).then_some("405")); + is_retryable_delete_replication_head_error(not_found, code) + } else { + !not_found + } }; if retryable { st.failed_count += 1; @@ -6503,6 +6517,108 @@ mod tests { server.join().expect("test HTTP server should finish"); } + fn delete_marker_roi() -> ReplicateObjectInfo { + ReplicateObjectInfo { + bucket: "source".to_string(), + name: "gone.txt".to_string(), + version_id: Some(Uuid::new_v4()), + op_type: ReplicationType::ExistingObject, + replication_status: ReplicationStatusType::Pending, + delete_marker: true, + ..Default::default() + } + } + + /// A target answers `HEAD ?versionId=` with a bodiless + /// 405, so the SDK reports no error code. That is the marker having + /// propagated, not a failure (rustfs/backlog#2479). + #[tokio::test] + async fn resync_verification_accepts_bodiless_405_for_delete_marker() { + let (endpoint, server) = spawn_head_status_server(405); + let target = test_target_client(endpoint); + let roi = delete_marker_roi(); + let mut st = TargetReplicationResyncStatus::default(); + + let head_result = + head_object_for_worker(target.as_ref(), &target.bucket, &roi.name, roi.version_id.map(|v| v.to_string())).await; + assert_eq!( + head_result + .as_ref() + .err() + .and_then(|err| err.as_service_error()) + .and_then(|se| se.code()), + None, + "the fixture must reproduce the codeless 405 the SDK yields for a bodiless HEAD" + ); + let (size, err) = verify_resync_head_result(head_result, &roi, &mut st, &target).await; + + assert!(err.is_none(), "{err:?}"); + assert_eq!((size, st.replicated_count, st.failed_count), (0, 1, 0)); + server.join().expect("test HTTP server should finish"); + } + + fn marker_purge_roi(status: VersionPurgeStatusType) -> ReplicateObjectInfo { + ReplicateObjectInfo { + version_purge_status: status, + ..delete_marker_roi() + } + } + + /// A pending or failed marker-version purge is verified by absence: the + /// same bodiless 405 that proves a created marker propagated proves a + /// purged marker is still on the target, so it must stay failed. + #[tokio::test] + async fn resync_verification_keeps_marker_purge_failed_when_target_still_answers_405() { + for status in [VersionPurgeStatusType::Pending, VersionPurgeStatusType::Failed] { + let (endpoint, server) = spawn_head_status_server(405); + let target = test_target_client(endpoint); + let roi = marker_purge_roi(status.clone()); + let mut st = TargetReplicationResyncStatus::default(); + + let head_result = + head_object_for_worker(target.as_ref(), &target.bucket, &roi.name, roi.version_id.map(|v| v.to_string())).await; + let (size, err) = verify_resync_head_result(head_result, &roi, &mut st, &target).await; + + assert!(err.is_some(), "a 405 must not count as a completed purge ({status:?})"); + assert_eq!((size, st.replicated_count, st.failed_count), (0, 0, 1), "{status:?}"); + server.join().expect("test HTTP server should finish"); + } + } + + /// Absence on the target is what completes a marker-version purge. + #[tokio::test] + async fn resync_verification_accepts_marker_purge_when_target_answers_404() { + let (endpoint, server) = spawn_head_status_server(404); + let target = test_target_client(endpoint); + let roi = marker_purge_roi(VersionPurgeStatusType::Pending); + let mut st = TargetReplicationResyncStatus::default(); + + let head_result = + head_object_for_worker(target.as_ref(), &target.bucket, &roi.name, roi.version_id.map(|v| v.to_string())).await; + let (size, err) = verify_resync_head_result(head_result, &roi, &mut st, &target).await; + + assert!(err.is_none(), "{err:?}"); + assert_eq!((size, st.replicated_count, st.failed_count), (0, 1, 0)); + server.join().expect("test HTTP server should finish"); + } + + /// Ambiguous HEAD failures still leave the delete marker unverified. + #[tokio::test] + async fn resync_verification_still_fails_delete_marker_on_ambiguous_head_error() { + let (endpoint, server) = spawn_head_status_server(503); + let target = test_target_client(endpoint); + let roi = delete_marker_roi(); + let mut st = TargetReplicationResyncStatus::default(); + + let head_result = + head_object_for_worker(target.as_ref(), &target.bucket, &roi.name, roi.version_id.map(|v| v.to_string())).await; + let (size, err) = verify_resync_head_result(head_result, &roi, &mut st, &target).await; + + assert!(err.is_some(), "a 503 must not count as a propagated delete marker"); + assert_eq!((size, st.replicated_count, st.failed_count), (0, 0, 1)); + server.join().expect("test HTTP server should finish"); + } + /// Serves exactly `requests` connections, answering each from /// `respond(request_line)`, and returns the request lines it saw. Reads /// the whole request (headers plus `Content-Length` body, honoring