fix(replication): count a bodiless 405 as a replicated delete marker (#7756)

* fix(replication): accept a bodiless 405 as a replicated delete marker during resync A resync verifies each delete marker with `HEAD ?versionId=<marker>` on the target. S3 targets (RustFS, MinIO, AWS) answer that with 405 and no body, and the SDK only synthesizes an error code for 404, so the error arrived with `code() == None`. `is_retryable_delete_replication_head_error` then treated it as an ambiguous failure: every delete marker counted as a failed object with `target service error`, and a site resync over any bucket that holds a delete marker reported the whole bucket as failed (rustfs/backlog#2479, SITE-105 on rc.6 and nightly). Use the raw HTTP status the way the 404 path already does: a 405 without a code confirms the marker propagated. Ambiguous statuses still fail. - Unit tests drive `verify_resync_head_result` against a scripted target answering 405 (accepted) and 503 (still failed). - e2e `test_site_replication_resync_replicates_delete_marker` joins two sites, converges a live object and a delete marker, and requires the site resync to complete with zero failed objects; the repl-nightly selection digest is refreshed for the new case. * fix(replication): verify marker-version purges by absence during resync A delete-marker resync entry with a pending or failed version purge asks the target to remove the marker, so the bodiless 405 that proves a created marker propagated proves the purge did not happen. Accept the 405-as-success mapping only for marker creation (empty version_purge_status); a purge counts as replicated only when the target answers not found, and any other HEAD outcome stays failed. Unit tests drive both purge statuses against the 405 fixture and the 404 fixture.
2026-09-19 01:54:50 +08:00 · 2026-09-14 11:41:35 +00:00
parent 36227adc96
commit 1ef3974bad
3 changed files with 230 additions and 2 deletions
@@ -1 +1 @@
-sha256=fc653ffc00f85a109be82e494117f0f60930f1c105898a5758af3c1bd998b1e1
+sha256=e4ee3898b840a8374cb02c64342cc15f8d7ecd1becaf4b4a2a4657290f9e7ff1
@@ -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=<marker>` 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<dyn Error + Send + Sync>> {
+    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
@@ -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=<delete marker>` 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