fix(replication): finish writeback for already purged versions

crates/e2e_test/src/replication_extension_test.rs

@@ -31,8 +31,8 @@ use aws_sdk_s3::error::ProvideErrorMetadata;
 use aws_sdk_s3::operation::list_object_versions::ListObjectVersionsOutput;
 use aws_sdk_s3::primitives::ByteStream;
 use aws_sdk_s3::types::{
-    BucketVersioningStatus, CompletedMultipartUpload, CompletedPart, DeleteMarkerEntry, ObjectVersion, ServerSideEncryption,
-    VersioningConfiguration,
+    BucketVersioningStatus, CompletedMultipartUpload, CompletedPart, Delete, DeleteMarkerEntry, ObjectIdentifier, ObjectVersion,
+    ServerSideEncryption, VersioningConfiguration,
 };
 use base64_simd::STANDARD as BASE64_STANDARD;
 use bytes::Bytes;
@@ -3729,6 +3729,8 @@ async fn test_bucket_replication_converges_delete_marker_and_version_purge() ->
     init_logging();
 
     let mut source_env = RustFSTestEnvironment::new().await?;
+    let source_log = format!("{}/source.log", source_env.temp_dir);
+    source_env.capture_log_path = Some(source_log.clone());
     let mut source_env_vars = replication_fast_env();
     source_env_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
     source_env.start_rustfs_server_with_env(vec![], &source_env_vars).await?;
@@ -3791,19 +3793,66 @@ async fn test_bucket_replication_converges_delete_marker_and_version_purge() ->
         "target did not preserve the source delete-marker version ID"
     );
 
-    source_client
-        .delete_object()
+    let wait_for_source_writeback = || async {
+        let metrics_url = format!("{}/{source_bucket}?replication-metrics", source_env.url);
+        timeout(Duration::from_secs(30), async {
+            loop {
+                let response = signed_request(
+                    http::Method::GET,
+                    &metrics_url,
+                    &source_env.access_key,
+                    &source_env.secret_key,
+                    None,
+                    None,
+                )
+                .await?;
+                let metrics: serde_json::Value = response.error_for_status()?.json().await?;
+                let queued = metrics["queued"]["curr"]["count"]
+                    .as_f64()
+                    .ok_or("missing replication queue count")?;
+                if metrics["provider_available"] == true && queued == 0.0 {
+                    return Ok::<(), Box<dyn Error + Send + Sync>>(());
+                }
+                sleep(Duration::from_millis(50)).await;
+            }
+        })
+        .await?
+    };
+    wait_for_source_writeback().await?;
+
+    let purged = source_client
+        .delete_objects()
         .bucket(source_bucket)
-        .key(object_key)
-        .version_id(&purged_version_id)
+        .delete(
+            Delete::builder()
+                .objects(
+                    ObjectIdentifier::builder()
+                        .key(object_key)
+                        .version_id(&purged_version_id)
+                        .build()?,
+                )
+                .objects(
+                    ObjectIdentifier::builder()
+                        .key(object_key)
+                        .version_id(delete_marker_version_id)
+                        .build()?,
+                )
+                .build()?,
+        )
         .send()
         .await?;
+    assert!(purged.errors().is_empty(), "batch version purge failed: {:?}", purged.errors());
+    assert_eq!(purged.deleted().len(), 2, "both the data version and marker must be purged");
     assert_replication_converged(&source_client, source_bucket, &target_client, target_bucket).await?;
     let target_state = list_replication_state(&target_client, target_bucket).await?;
     assert!(
         target_state.iter().all(|entry| entry.version_id != purged_version_id),
         "target retained the explicitly purged object version"
     );
+    assert!(
+        target_state.iter().all(|entry| entry.version_id != delete_marker_version_id),
+        "target retained the explicitly purged delete marker"
+    );
     assert!(
         target_state
             .iter()
@@ -3819,6 +3868,19 @@ async fn test_bucket_replication_converges_delete_marker_and_version_purge() ->
         .await?;
     assert_eq!(retained.body.collect().await?.into_bytes().as_ref(), b"versioned replication payload v2");
 
+    // Wait for the source workers to finish their final writeback, which runs
+    // after the remote purge has already become visible (rustfs#9121).
+    wait_for_source_writeback().await?;
+    let logs = fs::read_to_string(source_log).await?;
+    let writeback_failures: Vec<_> = logs
+        .lines()
+        .filter(|line| line.contains("apply_replication_delete_state"))
+        .collect();
+    assert!(
+        writeback_failures.is_empty(),
+        "a completed batch purge must not report a source writeback failure: {writeback_failures:?}"
+    );
+
     Ok(())
 }
 

crates/ecstore/src/bucket/replication/replication_pool.rs

@@ -3421,6 +3421,7 @@ mod tests {
         allow_first_read: Notify,
         read_count: AtomicUsize,
         delete_storage_calls: AtomicUsize,
+        delete_error: StdMutex<EcstoreError>,
         reported_size: StdMutex<Option<i64>>,
         stream_read_bytes: Arc<AtomicUsize>,
         write_count: AtomicUsize,
@@ -3672,7 +3673,12 @@ mod tests {
             _opts: Self::ObjectOptions,
         ) -> Result<Self::ObjectInfo, Self::Error> {
             self.shared.delete_storage_calls.fetch_add(1, Ordering::SeqCst);
-            Err(EcstoreError::NotImplemented)
+            Err(self
+                .shared
+                .delete_error
+                .lock()
+                .expect("delete error lock should not be poisoned")
+                .clone())
         }
 
         async fn delete_objects(
@@ -4171,6 +4177,7 @@ mod tests {
             allow_first_read: Notify::new(),
             read_count: AtomicUsize::new(0),
             delete_storage_calls: AtomicUsize::new(0),
+            delete_error: StdMutex::new(EcstoreError::NotImplemented),
             reported_size: StdMutex::new(None),
             stream_read_bytes: Arc::new(AtomicUsize::new(0)),
             write_count: AtomicUsize::new(0),
@@ -4221,6 +4228,103 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn delete_worker_already_purged_source_completes_without_retry() {
+        let arn = "arn:test";
+        for delete_marker_version in [false, true] {
+            for version_id in [Uuid::new_v4(), Uuid::nil()] {
+                for error in [
+                    EcstoreError::VersionNotFound("source".to_string(), "purged".to_string(), version_id.to_string()),
+                    EcstoreError::ObjectNotFound("source".to_string(), "purged".to_string()),
+                    EcstoreError::FileVersionNotFound,
+                    EcstoreError::FileNotFound,
+                ] {
+                    let shared = empty_resync_shared_state();
+                    *shared.delete_error.lock().expect("delete error lock should not be poisoned") = error;
+                    let storage = Arc::new(LoadResyncNodeStore::new("purged-source", shared.clone()));
+                    let delete = DeletedObjectReplicationInfo {
+                        bucket: "source".to_string(),
+                        delete_object: ReplicationDeletedObject {
+                            object_name: "purged".to_string(),
+                            version_id: (!delete_marker_version).then_some(version_id),
+                            delete_marker_version_id: delete_marker_version.then_some(version_id),
+                            replication_state: Some(ReplicationState {
+                                replicate_decision_str: test_replicate_decision(&[arn]).to_string(),
+                                purge_targets: HashMap::from([(arn.to_string(), VersionPurgeStatusType::Complete)]),
+                                ..Default::default()
+                            }),
+                            ..Default::default()
+                        },
+                        op_type: ReplicationType::Delete,
+                        ..Default::default()
+                    };
+
+                    assert!(
+                        replicate_delete_with_outcome(delete, storage).await,
+                        "a confirmed target purge must complete when the source is already absent (marker={delete_marker_version}, version={version_id})"
+                    );
+                    assert_eq!(
+                        shared.delete_storage_calls.load(Ordering::SeqCst),
+                        2 + usize::from(!delete_marker_version && !version_id.is_nil()),
+                        "the worker must lock and attempt writeback, reading a non-null data version first"
+                    );
+                }
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn delete_worker_missing_source_does_not_hide_unfinished_replication() {
+        let arn = "arn:test";
+        for (version_purge, target_status, error) in [
+            (
+                true,
+                VersionPurgeStatusType::Pending,
+                EcstoreError::ObjectNotFound("source".to_string(), "purged".to_string()),
+            ),
+            (true, VersionPurgeStatusType::Complete, EcstoreError::ErasureWriteQuorum),
+            (
+                false,
+                VersionPurgeStatusType::Empty,
+                EcstoreError::ObjectNotFound("source".to_string(), "purged".to_string()),
+            ),
+        ] {
+            let shared = empty_resync_shared_state();
+            *shared.delete_error.lock().expect("delete error lock should not be poisoned") = error;
+            let storage = Arc::new(LoadResyncNodeStore::new("unfinished-delete", shared));
+            let delete = DeletedObjectReplicationInfo {
+                bucket: "source".to_string(),
+                delete_object: ReplicationDeletedObject {
+                    object_name: "purged".to_string(),
+                    version_id: version_purge.then(Uuid::new_v4),
+                    delete_marker: !version_purge,
+                    replication_state: Some(ReplicationState {
+                        replicate_decision_str: test_replicate_decision(&[arn]).to_string(),
+                        targets: if version_purge {
+                            HashMap::new()
+                        } else {
+                            HashMap::from([(arn.to_string(), ReplicationStatusType::Completed)])
+                        },
+                        purge_targets: if version_purge {
+                            HashMap::from([(arn.to_string(), target_status)])
+                        } else {
+                            HashMap::new()
+                        },
+                        ..Default::default()
+                    }),
+                    ..Default::default()
+                },
+                op_type: ReplicationType::Delete,
+                ..Default::default()
+            };
+
+            assert!(
+                !replicate_delete_with_outcome(delete, storage).await,
+                "unfinished target work, source I/O failures, and marker creation must remain retryable"
+            );
+        }
+    }
+
     #[tokio::test]
     async fn delete_worker_preserves_persisted_admission_without_live_config() {
         let arn = "arn:test";
@@ -5304,6 +5408,7 @@ mod tests {
                 allow_first_read: Notify::new(),
                 read_count: AtomicUsize::new(0),
                 delete_storage_calls: AtomicUsize::new(0),
+                delete_error: StdMutex::new(EcstoreError::NotImplemented),
                 reported_size: StdMutex::new(None),
                 stream_read_bytes: Arc::new(AtomicUsize::new(0)),
                 write_count: AtomicUsize::new(0),

crates/ecstore/src/bucket/replication/replication_resyncer.rs

@@ -2809,17 +2809,22 @@ pub(crate) async fn replicate_delete_with_outcome<S: ReplicationStorage>(
             true
         }
         Err(e) => {
-            error!(
-                event = EVENT_RESYNC_TARGET_OPERATION_FAILED,
-                component = LOG_COMPONENT_ECSTORE,
-                subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
-                bucket = %bucket,
-                arn = %dobj.target_arn,
-                object = %dobj.delete_object.object_name,
-                operation = "apply_replication_delete_state",
-                error = %e,
-                "Replication target operation failed"
-            );
+            // A batch purge or another worker may have removed this version
+            // already. Remote failures still control the final outcome below.
+            let source_already_purged = is_version_purge && (is_err_object_not_found(&e) || is_err_version_not_found(&e));
+            if !source_already_purged {
+                error!(
+                    event = EVENT_RESYNC_TARGET_OPERATION_FAILED,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+                    bucket = %bucket,
+                    arn = %dobj.target_arn,
+                    object = %dobj.delete_object.object_name,
+                    operation = "apply_replication_delete_state",
+                    error = %e,
+                    "Replication target operation failed"
+                );
+            }
             send_event(EventArgs {
                 event_name,
                 bucket_name: bucket.clone(),
@@ -2832,7 +2837,7 @@ pub(crate) async fn replicate_delete_with_outcome<S: ReplicationStorage>(
                 },
                 ..Default::default()
             });
-            false
+            source_already_purged
         }
     };