Merge branch 'fix/replication-check-ledger-probe' into fix/replication-orphaned-purge-lifecycle

Same one-line change as rustfs/rustfs#7366 (main is red with it under -D warnings); carried here so the stacked PRs' merge commits compile until that fix lands.

crates/e2e_test/src/fake_s3_target/mod.rs

@@ -34,12 +34,14 @@ use s3s::dto::{
     AbortMultipartUploadInput, AbortMultipartUploadOutput, CommonPrefix, CompleteMultipartUploadInput,
     CompleteMultipartUploadOutput, CreateMultipartUploadInput, CreateMultipartUploadOutput, DeleteMarkerEntry, DeleteObjectInput,
     DeleteObjectOutput, DeleteObjectTaggingInput, DeleteObjectTaggingOutput, ETag, GetBucketVersioningInput,
-    GetBucketVersioningOutput, GetObjectInput, GetObjectLockConfigurationInput, GetObjectLockConfigurationOutput,
-    GetObjectOutput, GetObjectTaggingInput, GetObjectTaggingOutput, HeadBucketInput, HeadBucketOutput, HeadObjectInput,
+    GetBucketVersioningOutput, GetObjectInput, GetObjectLegalHoldInput, GetObjectLegalHoldOutput,
+    GetObjectLockConfigurationInput, GetObjectLockConfigurationOutput, GetObjectOutput, GetObjectRetentionInput,
+    GetObjectRetentionOutput, GetObjectTaggingInput, GetObjectTaggingOutput, HeadBucketInput, HeadBucketOutput, HeadObjectInput,
     HeadObjectOutput, ListObjectVersionsInput, ListObjectVersionsOutput, ListObjectsV2Input, ListObjectsV2Output, Object,
-    ObjectLockConfiguration, ObjectLockEnabled, ObjectStorageClass, ObjectVersionId, PutObjectInput, PutObjectOutput,
-    PutObjectTaggingInput, PutObjectTaggingOutput, Range, StreamingBlob, Tag, TagSet, Timestamp, TimestampFormat,
-    UploadPartInput, UploadPartOutput,
+    ObjectLockConfiguration, ObjectLockEnabled, ObjectLockLegalHold, ObjectLockLegalHoldStatus, ObjectLockMode,
+    ObjectLockRetention, ObjectLockRetentionMode, ObjectStorageClass, ObjectVersionId, PutObjectInput, PutObjectLegalHoldInput,
+    PutObjectLegalHoldOutput, PutObjectOutput, PutObjectRetentionInput, PutObjectRetentionOutput, PutObjectTaggingInput,
+    PutObjectTaggingOutput, Range, StreamingBlob, Tag, TagSet, Timestamp, TimestampFormat, UploadPartInput, UploadPartOutput,
 };
 use s3s::service::{S3Service, S3ServiceBuilder};
 use s3s::validation::{AwsNameValidation, NameValidation};
@@ -127,6 +129,10 @@ pub enum Operation {
     GetObjectTagging,
     PutObjectTagging,
     DeleteObjectTagging,
+    GetObjectRetention,
+    PutObjectRetention,
+    GetObjectLegalHold,
+    PutObjectLegalHold,
     ListObjectVersions,
     ListObjectsV2,
     CreateMultipartUpload,
@@ -501,6 +507,10 @@ struct StoreState {
     /// PutObject carrying any `x-amz-object-lock-*` header must also carry
     /// `Content-MD5` or an `x-amz-checksum-*` header.
     require_checksum_for_object_lock: bool,
+    /// Models Wasabi (rustfs/backlog#2340): a version-addressed DELETE of a
+    /// version id the target never had answers 404 `NoSuchVersion` instead of
+    /// the idempotent 204 RustFS/MinIO give.
+    reject_unknown_version_deletes: bool,
     limits: StoreLimits,
     buckets: HashMap<String, BucketState>,
     uploads: HashMap<String, MultipartState>,
@@ -565,6 +575,41 @@ struct ObjectVersion {
     /// SSE-C passthrough transport headers stored with the version (RustFS
     /// target behavior); empty when the drop mode discarded them.
     replication_sse_headers: Vec<(String, String)>,
+    /// Object Lock state of the version: retention (mode, retain-until) from
+    /// the PUT / CreateMultipartUpload headers or PutObjectRetention, and the
+    /// legal hold flag; replayed on HEAD.
+    lock: VersionLock,
+}
+
+#[derive(Clone, Default)]
+struct VersionLock {
+    retention: Option<(String, Timestamp)>,
+    /// `None` until a legal hold status was ever set; like S3, HEAD then
+    /// reports nothing, while an explicit OFF is reported as `OFF`.
+    legal_hold: Option<bool>,
+}
+
+impl VersionLock {
+    fn from_headers(
+        mode: Option<ObjectLockMode>,
+        retain_until: Option<Timestamp>,
+        legal_hold: Option<ObjectLockLegalHoldStatus>,
+    ) -> Self {
+        Self {
+            retention: mode.zip(retain_until).map(|(mode, until)| (mode.as_str().to_string(), until)),
+            legal_hold: legal_hold.map(|status| status.as_str().eq_ignore_ascii_case("ON")),
+        }
+    }
+
+    fn legal_hold_status(&self) -> Option<ObjectLockLegalHoldStatus> {
+        self.legal_hold.map(|on| {
+            ObjectLockLegalHoldStatus::from_static(if on {
+                ObjectLockLegalHoldStatus::ON
+            } else {
+                ObjectLockLegalHoldStatus::OFF
+            })
+        })
+    }
 }
 
 #[derive(Clone)]
@@ -576,6 +621,7 @@ struct MultipartState {
     metadata: Option<HashMap<String, String>>,
     standard_headers: StandardHeaders,
     replication_sse_headers: Vec<(String, String)>,
+    lock: VersionLock,
     parts: BTreeMap<i32, MultipartPart>,
 }
 
@@ -845,6 +891,7 @@ impl FakeS3Target {
             standard_headers: seed.standard_headers.clone(),
             tags: Vec::new(),
             replication_sse_headers: Vec::new(),
+            lock: VersionLock::default(),
         };
         upsert_version(&mut state, bucket, key.into(), version).expect("seed object must fit the storage budget");
         e_tag
@@ -918,6 +965,12 @@ impl FakeS3Target {
     /// PutObject that carries Object Lock parameters (AWS S3 / MinIO rule,
     /// rustfs#7082). `Content-MD5`, when present, is always verified against
     /// the body regardless of this mode.
+    /// Wasabi-like mode: DELETE of an unknown version id answers 404
+    /// `NoSuchVersion` (the default 204 models RustFS/MinIO).
+    pub fn reject_unknown_version_deletes(&self, enabled: bool) {
+        lock(&self.backend.store).reject_unknown_version_deletes = enabled;
+    }
+
     pub fn require_checksum_for_object_lock(&self, enabled: bool) {
         lock(&self.backend.store).require_checksum_for_object_lock = enabled;
     }
@@ -1152,6 +1205,10 @@ fn operation_from_s3_name(name: &str) -> Operation {
         "GetObjectTagging" => Operation::GetObjectTagging,
         "PutObjectTagging" => Operation::PutObjectTagging,
         "DeleteObjectTagging" => Operation::DeleteObjectTagging,
+        "GetObjectRetention" => Operation::GetObjectRetention,
+        "PutObjectRetention" => Operation::PutObjectRetention,
+        "GetObjectLegalHold" => Operation::GetObjectLegalHold,
+        "PutObjectLegalHold" => Operation::PutObjectLegalHold,
         "ListObjectsV2" => Operation::ListObjectsV2,
         "CreateMultipartUpload" => Operation::CreateMultipartUpload,
         "UploadPart" => Operation::UploadPart,
@@ -1289,6 +1346,18 @@ fn parse_request(method: &Method, uri: &Uri) -> ParsedRequest {
         (&Method::DELETE, true) if query.contains_key("tagging") && only_query_keys(&["tagging", "versionId"]) => {
             Operation::DeleteObjectTagging
         }
+        (&Method::GET, true) if query.contains_key("retention") && only_query_keys(&["retention", "versionId"]) => {
+            Operation::GetObjectRetention
+        }
+        (&Method::PUT, true) if query.contains_key("retention") && only_query_keys(&["retention", "versionId"]) => {
+            Operation::PutObjectRetention
+        }
+        (&Method::GET, true) if query.contains_key("legal-hold") && only_query_keys(&["legal-hold", "versionId"]) => {
+            Operation::GetObjectLegalHold
+        }
+        (&Method::PUT, true) if query.contains_key("legal-hold") && only_query_keys(&["legal-hold", "versionId"]) => {
+            Operation::PutObjectLegalHold
+        }
         // A replication PUT addresses the source version via `?versionId=`.
         (&Method::PUT, true) if only_query_keys(&["versionId"]) => Operation::PutObject,
         (&Method::GET, true) if only_query_keys(&["versionId"]) => Operation::GetObject,
@@ -1842,6 +1911,28 @@ fn set_version_tags(
     Ok(resolved)
 }
 
+fn update_version_lock(
+    state: &mut StoreState,
+    bucket: &str,
+    key: &str,
+    version_id: Option<&str>,
+    update: impl FnOnce(&mut VersionLock),
+) -> S3Result<String> {
+    let resolved = find_version(state, bucket, key, version_id)?.version_id;
+    let version = state
+        .buckets
+        .get_mut(bucket)
+        .expect("bucket existence checked by find_version")
+        .objects
+        .get_mut(key)
+        .expect("key existence checked by find_version")
+        .iter_mut()
+        .find(|version| version.version_id == resolved)
+        .expect("version existence checked by find_version");
+    update(&mut version.lock);
+    Ok(resolved)
+}
+
 /// Whether version ids are surfaced for this bucket. Unknown buckets report
 /// `true`; the caller's lookup raises `NoSuchBucket` first.
 fn bucket_versioned(state: &StoreState, bucket: &str) -> bool {
@@ -2281,6 +2372,11 @@ impl S3 for FakeBackend {
             standard_headers,
             tags: Vec::new(),
             replication_sse_headers: captured_replication_sse_headers(&headers, drop_unlisted),
+            lock: VersionLock::from_headers(
+                input.object_lock_mode,
+                input.object_lock_retain_until_date,
+                input.object_lock_legal_hold_status,
+            ),
         };
         upsert_version(&mut lock(&self.store), &input.bucket, input.key, version)?;
         Ok(apply_response_fault(
@@ -2339,6 +2435,13 @@ impl S3 for FakeBackend {
             last_modified: Some(version.last_modified.clone()),
             version_id: versioned.then_some(version.version_id),
             sse_customer_algorithm,
+            object_lock_mode: version
+                .lock
+                .retention
+                .as_ref()
+                .map(|(mode, _)| ObjectLockMode::from(mode.clone())),
+            object_lock_retain_until_date: version.lock.retention.as_ref().map(|(_, until)| until.clone()),
+            object_lock_legal_hold_status: version.lock.legal_hold_status(),
             ..Default::default()
         });
         response.status = served.status;
@@ -2373,6 +2476,13 @@ impl S3 for FakeBackend {
             last_modified: Some(version.last_modified.clone()),
             version_id: versioned.then_some(version.version_id),
             sse_customer_algorithm,
+            object_lock_mode: version
+                .lock
+                .retention
+                .as_ref()
+                .map(|(mode, _)| ObjectLockMode::from(mode.clone())),
+            object_lock_retain_until_date: version.lock.retention.as_ref().map(|(_, until)| until.clone()),
+            object_lock_legal_hold_status: version.lock.legal_hold_status(),
             ..Default::default()
         });
         response.status = served.status;
@@ -2432,6 +2542,82 @@ impl S3 for FakeBackend {
         ))
     }
 
+    async fn get_object_retention(
+        &self,
+        req: S3Request<GetObjectRetentionInput>,
+    ) -> S3Result<S3Response<GetObjectRetentionOutput>> {
+        let fault = request_fault(&req);
+        apply_non_body_fault(fault.as_ref(), &self.control).await?;
+        let input = req.input;
+        let version = find_version(&lock(&self.store), &input.bucket, &input.key, input.version_id.as_deref())?;
+        Ok(apply_response_fault(
+            S3Response::new(GetObjectRetentionOutput {
+                retention: version.lock.retention.map(|(mode, until)| ObjectLockRetention {
+                    mode: Some(ObjectLockRetentionMode::from(mode)),
+                    retain_until_date: Some(until),
+                }),
+            }),
+            fault.as_ref(),
+        ))
+    }
+
+    async fn put_object_retention(
+        &self,
+        req: S3Request<PutObjectRetentionInput>,
+    ) -> S3Result<S3Response<PutObjectRetentionOutput>> {
+        let fault = request_fault(&req);
+        apply_non_body_fault(fault.as_ref(), &self.control).await?;
+        let input = req.input;
+        let retention = input
+            .retention
+            .and_then(|retention| retention.mode.zip(retention.retain_until_date))
+            .map(|(mode, until)| (mode.as_str().to_string(), until));
+        update_version_lock(&mut lock(&self.store), &input.bucket, &input.key, input.version_id.as_deref(), |lock| {
+            lock.retention = retention;
+        })?;
+        Ok(apply_response_fault(S3Response::new(PutObjectRetentionOutput::default()), fault.as_ref()))
+    }
+
+    async fn get_object_legal_hold(
+        &self,
+        req: S3Request<GetObjectLegalHoldInput>,
+    ) -> S3Result<S3Response<GetObjectLegalHoldOutput>> {
+        let fault = request_fault(&req);
+        apply_non_body_fault(fault.as_ref(), &self.control).await?;
+        let input = req.input;
+        let version = find_version(&lock(&self.store), &input.bucket, &input.key, input.version_id.as_deref())?;
+        Ok(apply_response_fault(
+            S3Response::new(GetObjectLegalHoldOutput {
+                legal_hold: Some(ObjectLockLegalHold {
+                    status: Some(
+                        version
+                            .lock
+                            .legal_hold_status()
+                            .unwrap_or_else(|| ObjectLockLegalHoldStatus::from_static(ObjectLockLegalHoldStatus::OFF)),
+                    ),
+                }),
+            }),
+            fault.as_ref(),
+        ))
+    }
+
+    async fn put_object_legal_hold(
+        &self,
+        req: S3Request<PutObjectLegalHoldInput>,
+    ) -> S3Result<S3Response<PutObjectLegalHoldOutput>> {
+        let fault = request_fault(&req);
+        apply_non_body_fault(fault.as_ref(), &self.control).await?;
+        let input = req.input;
+        let legal_hold_on = input
+            .legal_hold
+            .and_then(|hold| hold.status)
+            .is_some_and(|status| status.as_str().eq_ignore_ascii_case("ON"));
+        update_version_lock(&mut lock(&self.store), &input.bucket, &input.key, input.version_id.as_deref(), |lock| {
+            lock.legal_hold = Some(legal_hold_on);
+        })?;
+        Ok(apply_response_fault(S3Response::new(PutObjectLegalHoldOutput::default()), fault.as_ref()))
+    }
+
     async fn delete_object_tagging(
         &self,
         req: S3Request<DeleteObjectTaggingInput>,
@@ -2485,6 +2671,7 @@ impl S3 for FakeBackend {
             return Ok(apply_response_fault(S3Response::new(DeleteObjectOutput::default()), fault.as_ref()));
         }
         if let Some(version_id) = input.version_id {
+            let reject_unknown = state.reject_unknown_version_deletes;
             let (removed_bytes, removed_versions, delete_marker, remove_key) = {
                 let Some(versions) = state
                     .buckets
@@ -2493,6 +2680,9 @@ impl S3 for FakeBackend {
                     .objects
                     .get_mut(&input.key)
                 else {
+                    if reject_unknown {
+                        return Err(s3s::s3_error!(NoSuchVersion, "The specified version does not exist."));
+                    }
                     return Ok(apply_response_fault(
                         S3Response::new(DeleteObjectOutput {
                             version_id: Some(version_id),
@@ -2501,6 +2691,9 @@ impl S3 for FakeBackend {
                         fault.as_ref(),
                     ));
                 };
+                if reject_unknown && !versions.iter().any(|version| version.version_id == version_id) {
+                    return Err(s3s::s3_error!(NoSuchVersion, "The specified version does not exist."));
+                }
                 let mut removed_bytes = 0usize;
                 let mut removed_versions = 0usize;
                 let mut delete_marker = None;
@@ -2554,6 +2747,7 @@ impl S3 for FakeBackend {
                 standard_headers: StandardHeaders::default(),
                 tags: Vec::new(),
                 replication_sse_headers: Vec::new(),
+                lock: VersionLock::default(),
             },
         )?;
         Ok(apply_response_fault(
@@ -2608,6 +2802,11 @@ impl S3 for FakeBackend {
                 metadata: input.metadata,
                 standard_headers,
                 replication_sse_headers: captured_replication_sse_headers(&headers, drop_unlisted),
+                lock: VersionLock::from_headers(
+                    input.object_lock_mode,
+                    input.object_lock_retain_until_date,
+                    input.object_lock_legal_hold_status,
+                ),
                 parts: BTreeMap::new(),
             },
         );
@@ -2748,6 +2947,7 @@ impl S3 for FakeBackend {
                     metadata: upload.metadata.clone(),
                     standard_headers: upload.standard_headers.clone(),
                     replication_sse_headers: upload.replication_sse_headers.clone(),
+                    lock: upload.lock.clone(),
                     parts: BTreeMap::new(),
                 },
                 selected,
@@ -2778,6 +2978,7 @@ impl S3 for FakeBackend {
             standard_headers: upload.standard_headers,
             tags: Vec::new(),
             replication_sse_headers: upload.replication_sse_headers,
+            lock: upload.lock,
         };
         let mut state = lock(&self.store);
         let versioned = bucket_versioned(&state, &input.bucket);
@@ -4609,6 +4810,7 @@ mod tests {
                         metadata: None,
                         standard_headers: StandardHeaders::default(),
                         replication_sse_headers: Vec::new(),
+                        lock: VersionLock::default(),
                         parts: BTreeMap::new(),
                     },
                 );

crates/e2e_test/src/replication_extension_test.rs

@@ -512,7 +512,7 @@ pub(crate) async fn put_bucket_replication(
     put_bucket_replication_with_delete_statuses(env, bucket, target_arn, "Enabled", None).await
 }
 
-async fn put_bucket_replication_with_delete_statuses(
+pub(crate) async fn put_bucket_replication_with_delete_statuses(
     env: &RustFSTestEnvironment,
     bucket: &str,
     target_arn: &str,
@@ -627,7 +627,7 @@ async fn put_bucket_replication_rules(
     Ok(())
 }
 
-async fn delete_bucket_replication(
+pub(crate) async fn delete_bucket_replication(
     env: &RustFSTestEnvironment,
     bucket: &str,
 ) -> Result<reqwest::Response, Box<dyn Error + Send + Sync>> {
@@ -9289,11 +9289,13 @@ async fn test_replication_check_flags_version_minting_target() -> TestResult {
         fidelity["Code"], "BucketRemoteTargetVersionMismatch",
         "the failure must carry a machine-readable code: {payload}"
     );
-    // The probe PUT itself succeeded (fidelity is judged from its response);
-    // the later mutation phases are pointless against a drifting target and
-    // must be skipped, but cleanup still runs.
+    // The probe PUT itself succeeded (fidelity is judged from its response).
+    // The mutation phases address the id the target assigned — the ledger
+    // the worker records per object (rustfs/backlog#2340) — so they run and
+    // pass on a drifting target, and cleanup uses the same id.
     assert_eq!(target_report["Phases"]["Put"]["Status"], "OK", "{payload}");
-    assert_eq!(target_report["Phases"]["DeleteMarker"]["Status"], "SKIPPED", "{payload}");
+    assert_eq!(target_report["Phases"]["DeleteMarker"]["Status"], "OK", "{payload}");
+    assert_eq!(target_report["Phases"]["VersionDelete"]["Status"], "OK", "{payload}");
     assert_eq!(target_report["Phases"]["Cleanup"]["Status"], "OK", "{payload}");
 
     // The probe PUT must carry the source version as `?versionId=` — the

crates/e2e_test/src/replication_target_matrix_test.rs

@@ -36,14 +36,16 @@ use crate::fake_s3_target::{BucketMode, FAKE_ACCESS_KEY, FAKE_SECRET_KEY};
 use crate::fake_s3_target::{FakeS3Target, FaultAction as FakeTargetFault, Operation as FakeTargetOperation, RequestRecord};
 use crate::on_demand_migration::common::{OdmEnvOptions, OdmTestEnv, fake_source_client};
 use crate::replication_extension_test::{
-    LOOPBACK_REPLICATION_TARGET_ENV, ReplicationTargetOptions, enable_bucket_versioning, get_replication_reset_status,
-    put_bucket_replication, set_replication_target_with_options, start_bucket_replication_reset,
+    LOOPBACK_REPLICATION_TARGET_ENV, ReplicationTargetOptions, delete_bucket_replication, enable_bucket_versioning,
+    get_replication_reset_status, put_bucket_replication, put_bucket_replication_with_delete_statuses,
+    set_replication_target_with_options, start_bucket_replication_reset,
 };
 use aws_sdk_s3::Client;
+use aws_sdk_s3::error::ProvideErrorMetadata;
 use aws_sdk_s3::primitives::{ByteStream, DateTime};
 use aws_sdk_s3::types::{
-    Checksum, ChecksumAlgorithm, CompletedMultipartUpload, CompletedPart, ObjectAttributes, ObjectLockLegalHoldStatus,
-    ObjectLockMode,
+    Checksum, ChecksumAlgorithm, CompletedMultipartUpload, CompletedPart, ObjectAttributes, ObjectLockLegalHold,
+    ObjectLockLegalHoldStatus, ObjectLockMode, ObjectLockRetention, ObjectLockRetentionMode, Tag, Tagging,
 };
 use bytes::Bytes;
 use std::error::Error;
@@ -66,7 +68,10 @@ enum TargetMode {
     /// Object Lock parameters must carry `Content-MD5` or `x-amz-checksum-*`.
     RequireChecksumWithObjectLock,
     /// AWS S3 / Wasabi / Impossible Cloud: mints its own version ids
-    /// (rustfs/backlog#2085). Data must still land.
+    /// (rustfs/backlog#2085) and, like Wasabi, answers NoSuchVersion to a
+    /// DELETE of an id it never had (rustfs/backlog#2340). Data must still
+    /// land, and every version-addressed mutation must resolve the replica
+    /// through the target-version ledger.
     MintOwnVersionIds,
 }
 
@@ -83,7 +88,10 @@ impl TargetMode {
             TargetMode::Baseline => {}
             TargetMode::RejectAwsChunked => target.reject_aws_chunked_uploads(true),
             TargetMode::RequireChecksumWithObjectLock => target.require_checksum_for_object_lock(true),
-            TargetMode::MintOwnVersionIds => target.assign_own_version_ids(true),
+            TargetMode::MintOwnVersionIds => {
+                target.assign_own_version_ids(true);
+                target.reject_unknown_version_deletes(true);
+            }
         }
     }
 
@@ -384,6 +392,431 @@ async fn matrix_mint_own_version_ids_redrive_does_not_duplicate() -> TestResult
     Ok(())
 }
 
+/// rustfs/backlog#2340 (target-version ledger): on a target that mints its own
+/// version ids and answers NoSuchVersion to an unknown id (the Wasabi shape),
+/// every version-addressed mutation must land on the version the target
+/// assigned, which the replication PUT recorded on the source:
+/// - a tag update changes the existing target version, no new version;
+/// - a retention extension and legal hold ON/OFF change that version too;
+/// - a permanent delete of the older of two same-content generations removes
+///   exactly that replica and keeps the live one (content identity alone
+///   could not tell them apart).
+#[tokio::test]
+async fn matrix_mint_own_version_ids_addresses_mutations_through_the_ledger() -> TestResult {
+    init_logging();
+
+    let target = FakeS3Target::start().await?;
+    let target_bucket = "matrix-mint-own-ledger-dst".to_string();
+    target.create_bucket_with_object_lock(target_bucket.clone());
+    TargetMode::MintOwnVersionIds.apply(&target);
+
+    let mut env_vars = replication_fast_env();
+    env_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
+    env_vars.extend_from_slice(&[
+        ("NO_PROXY", "127.0.0.1,localhost"),
+        ("HTTP_PROXY", ""),
+        ("HTTPS_PROXY", ""),
+        // The scanner heal pass retries a purge the first attempt lost.
+        ("RUSTFS_SCANNER_CYCLE", "1"),
+        ("RUSTFS_SCANNER_START_DELAY_SECS", "1"),
+    ]);
+    let env = OdmTestEnv::start_with(OdmEnvOptions {
+        env: env_vars,
+        ..OdmEnvOptions::default()
+    })
+    .await?;
+    let source_env = &env.rustfs;
+
+    let source_bucket = "matrix-mint-own-ledger-src";
+    let source_client = source_env.create_s3_client();
+    source_client
+        .create_bucket()
+        .bucket(source_bucket)
+        .object_lock_enabled_for_bucket(true)
+        .send()
+        .await?;
+    enable_bucket_versioning(source_env, source_bucket).await?;
+    let target_arn = set_replication_target_with_options(
+        source_env,
+        source_bucket,
+        ReplicationTargetOptions {
+            endpoint: &target.address(),
+            access_key: FAKE_ACCESS_KEY,
+            secret_key: FAKE_SECRET_KEY,
+            target_bucket: &target_bucket,
+            secure: false,
+            skip_tls_verify: false,
+            ca_cert_pem: None,
+        },
+    )
+    .await?;
+    put_bucket_replication_with_delete_statuses(source_env, source_bucket, &target_arn, "Enabled", Some("Enabled")).await?;
+    let target_client = fake_source_client(&target);
+
+    // Tag update on an existing version.
+    let tag_key = "ledger/tags.bin";
+    let tagged = source_client
+        .put_object()
+        .bucket(source_bucket)
+        .key(tag_key)
+        .body(ByteStream::from(payload(4 * 1024, 0x01)))
+        .send()
+        .await?;
+    let tag_source_version = tagged.version_id().ok_or("source PUT returned no version id")?.to_string();
+    assert_eq!(
+        wait_for_terminal_replication_status(&source_client, source_bucket, tag_key).await?,
+        "COMPLETED"
+    );
+    let tag_target_version = single_target_version(&target, &target_bucket, tag_key)?;
+    source_client
+        .put_object_tagging()
+        .bucket(source_bucket)
+        .key(tag_key)
+        .version_id(&tag_source_version)
+        .tagging(
+            Tagging::builder()
+                .tag_set(Tag::builder().key("phase").value("after").build()?)
+                .build()?,
+        )
+        .send()
+        .await?;
+    wait_until("tag update on the existing target version", || async {
+        let tags = target_client
+            .get_object_tagging()
+            .bucket(&target_bucket)
+            .key(tag_key)
+            .version_id(&tag_target_version)
+            .send()
+            .await?;
+        Ok(tags
+            .tag_set()
+            .iter()
+            .any(|tag| tag.key() == "phase" && tag.value() == "after"))
+    })
+    .await?;
+    assert_stable_single_version(&target, &target_bucket, tag_key, &tag_target_version).await?;
+
+    // Retention extension and legal hold on an existing version.
+    let lock_key = "ledger/lock.bin";
+    let locked = source_client
+        .put_object()
+        .bucket(source_bucket)
+        .key(lock_key)
+        .body(ByteStream::from(payload(4 * 1024, 0x02)))
+        .object_lock_mode(ObjectLockMode::Governance)
+        .object_lock_retain_until_date(retain_until())
+        .send()
+        .await?;
+    let lock_source_version = locked.version_id().ok_or("source PUT returned no version id")?.to_string();
+    assert_eq!(
+        wait_for_terminal_replication_status(&source_client, source_bucket, lock_key).await?,
+        "COMPLETED"
+    );
+    let lock_target_version = single_target_version(&target, &target_bucket, lock_key)?;
+    let extended = DateTime::from_secs(retain_until().secs() + 86_400);
+    source_client
+        .put_object_retention()
+        .bucket(source_bucket)
+        .key(lock_key)
+        .version_id(&lock_source_version)
+        .retention(
+            ObjectLockRetention::builder()
+                .mode(ObjectLockRetentionMode::Governance)
+                .retain_until_date(extended)
+                .build(),
+        )
+        .send()
+        .await?;
+    source_client
+        .put_object_legal_hold()
+        .bucket(source_bucket)
+        .key(lock_key)
+        .version_id(&lock_source_version)
+        .legal_hold(ObjectLockLegalHold::builder().status(ObjectLockLegalHoldStatus::On).build())
+        .send()
+        .await?;
+    wait_until("retention extension and legal hold on the existing target version", || async {
+        let head = target_client
+            .head_object()
+            .bucket(&target_bucket)
+            .key(lock_key)
+            .version_id(&lock_target_version)
+            .send()
+            .await?;
+        Ok(head.object_lock_retain_until_date().map(|date| date.secs()) == Some(extended.secs())
+            && head.object_lock_legal_hold_status() == Some(&ObjectLockLegalHoldStatus::On))
+    })
+    .await?;
+    source_client
+        .put_object_legal_hold()
+        .bucket(source_bucket)
+        .key(lock_key)
+        .version_id(&lock_source_version)
+        .legal_hold(ObjectLockLegalHold::builder().status(ObjectLockLegalHoldStatus::Off).build())
+        .send()
+        .await?;
+    wait_until("legal hold removal on the existing target version", || async {
+        let head = target_client
+            .head_object()
+            .bucket(&target_bucket)
+            .key(lock_key)
+            .version_id(&lock_target_version)
+            .send()
+            .await?;
+        Ok(head.object_lock_legal_hold_status() == Some(&ObjectLockLegalHoldStatus::Off))
+    })
+    .await?;
+    assert_stable_single_version(&target, &target_bucket, lock_key, &lock_target_version).await?;
+
+    // Permanent delete of the older of two same-content generations.
+    let generations_key = "ledger/generations.bin";
+    let body = payload(4 * 1024, 0x03);
+    let older = source_client
+        .put_object()
+        .bucket(source_bucket)
+        .key(generations_key)
+        .body(ByteStream::from(body.clone()))
+        .send()
+        .await?;
+    let older_version = older.version_id().ok_or("source PUT returned no version id")?.to_string();
+    assert_eq!(
+        wait_for_terminal_replication_status(&source_client, source_bucket, generations_key).await?,
+        "COMPLETED"
+    );
+    let older_replica = single_target_version(&target, &target_bucket, generations_key)?;
+    source_client
+        .put_object()
+        .bucket(source_bucket)
+        .key(generations_key)
+        .body(ByteStream::from(body))
+        .send()
+        .await?;
+    assert_eq!(
+        wait_for_terminal_replication_status(&source_client, source_bucket, generations_key).await?,
+        "COMPLETED"
+    );
+    wait_until("both generations replicated", || async {
+        Ok(target.stored_versions(&target_bucket, generations_key).len() == 2)
+    })
+    .await?;
+    let newer_replica = target
+        .stored_versions(&target_bucket, generations_key)
+        .into_iter()
+        .map(|(version_id, _)| version_id)
+        .find(|version_id| version_id != &older_replica)
+        .ok_or("the second generation must have its own target version")?;
+
+    source_client
+        .delete_object()
+        .bucket(source_bucket)
+        .key(generations_key)
+        .version_id(&older_version)
+        .send()
+        .await?;
+    wait_until("permanent delete of the older generation's replica", || async {
+        let versions: Vec<String> = target
+            .stored_versions(&target_bucket, generations_key)
+            .into_iter()
+            .map(|(version_id, _)| version_id)
+            .collect();
+        Ok(versions == [newer_replica.clone()])
+    })
+    .await?;
+    assert_stable_single_version(&target, &target_bucket, generations_key, &newer_replica).await?;
+
+    // No mutation above may have gone out as a re-PUT: one upload per key.
+    for key in [tag_key, lock_key] {
+        let puts = target
+            .requests()
+            .iter()
+            .filter(|record| record.key.as_deref() == Some(key) && record.operation == FakeTargetOperation::PutObject)
+            .count();
+        assert_eq!(
+            puts, 1,
+            "{key}: a metadata update must not re-PUT the object on a target that mints its own ids"
+        );
+    }
+
+    target.shutdown().await;
+    Ok(())
+}
+
+/// rustfs/backlog#2340 (pending purge lifecycle): a permanent delete whose
+/// replication keeps failing leaves the version in xl.meta as a PENDING purge,
+/// hidden from listings. Once the bucket's replication configuration is
+/// removed nothing can ever confirm that purge remotely, so the delete worker
+/// must settle it locally (abandoned, with the replica left on the former
+/// target) — otherwise the bucket stays `BucketNotEmpty` forever with a
+/// residue the client cannot see.
+#[tokio::test]
+async fn matrix_removed_replication_config_abandons_pending_purge() -> TestResult {
+    init_logging();
+
+    let target = FakeS3Target::start().await?;
+    let target_bucket = "matrix-abandoned-purge-dst".to_string();
+    target.create_bucket_with_object_lock(target_bucket.clone());
+    TargetMode::MintOwnVersionIds.apply(&target);
+
+    let mut env_vars = replication_fast_env();
+    env_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
+    env_vars.extend_from_slice(&[
+        ("NO_PROXY", "127.0.0.1,localhost"),
+        ("HTTP_PROXY", ""),
+        ("HTTPS_PROXY", ""),
+        // The scanner heal pass is what revisits a pending purge.
+        ("RUSTFS_SCANNER_CYCLE", "1"),
+        ("RUSTFS_SCANNER_START_DELAY_SECS", "1"),
+    ]);
+    let env = OdmTestEnv::start_with(OdmEnvOptions {
+        env: env_vars,
+        ..OdmEnvOptions::default()
+    })
+    .await?;
+    let source_env = &env.rustfs;
+
+    let source_bucket = "matrix-abandoned-purge-src";
+    let source_client = source_env.create_s3_client();
+    source_client
+        .create_bucket()
+        .bucket(source_bucket)
+        .object_lock_enabled_for_bucket(true)
+        .send()
+        .await?;
+    enable_bucket_versioning(source_env, source_bucket).await?;
+    let target_arn = set_replication_target_with_options(
+        source_env,
+        source_bucket,
+        ReplicationTargetOptions {
+            endpoint: &target.address(),
+            access_key: FAKE_ACCESS_KEY,
+            secret_key: FAKE_SECRET_KEY,
+            target_bucket: &target_bucket,
+            secure: false,
+            skip_tls_verify: false,
+            ca_cert_pem: None,
+        },
+    )
+    .await?;
+    put_bucket_replication_with_delete_statuses(source_env, source_bucket, &target_arn, "Enabled", Some("Enabled")).await?;
+
+    let key = "purge/orphaned.bin";
+    let put = source_client
+        .put_object()
+        .bucket(source_bucket)
+        .key(key)
+        .body(ByteStream::from(payload(4 * 1024, 0x07)))
+        .send()
+        .await?;
+    let source_version = put.version_id().ok_or("source PUT returned no version id")?.to_string();
+    assert_eq!(
+        wait_for_terminal_replication_status(&source_client, source_bucket, key).await?,
+        "COMPLETED"
+    );
+    let replica = single_target_version(&target, &target_bucket, key)?;
+
+    // The target refuses every purge: the version stays a pending purge.
+    // More refusals than any scanner cycle can consume within the test.
+    target.inject_for_key(FakeTargetOperation::DeleteObject, key, FakeTargetFault::ResponseStatus(503), 4_000);
+    source_client
+        .delete_object()
+        .bucket(source_bucket)
+        .key(key)
+        .version_id(&source_version)
+        .send()
+        .await?;
+    wait_until("the refused purge to reach the target at least once", || async {
+        Ok(target.count_requests(FakeTargetOperation::DeleteObject, key) >= 1)
+    })
+    .await?;
+    let listed = source_client.list_object_versions().bucket(source_bucket).send().await?;
+    assert!(
+        listed.versions().is_empty() && listed.delete_markers().is_empty(),
+        "a pending purge is hidden from listings: {listed:?}"
+    );
+    let blocked = source_client.delete_bucket().bucket(source_bucket).send().await;
+    assert!(
+        blocked
+            .as_ref()
+            .err()
+            .and_then(|err| err.as_service_error())
+            .is_some_and(|err| err.code() == Some("BucketNotEmpty")),
+        "the hidden pending purge must block DeleteBucket while the target is still configured: {blocked:?}"
+    );
+
+    // Removing the replication configuration orphans the purge; the scanner
+    // heal pass must settle it locally so the bucket becomes deletable.
+    let response = delete_bucket_replication(source_env, source_bucket).await?;
+    assert!(response.status().is_success(), "DeleteBucketReplication: {}", response.status());
+    wait_until("DeleteBucket to succeed once the orphaned purge is abandoned", || async {
+        match source_client.delete_bucket().bucket(source_bucket).send().await {
+            Ok(_) => Ok(true),
+            Err(err) if err.as_service_error().is_some_and(|err| err.code() == Some("BucketNotEmpty")) => Ok(false),
+            Err(err) => Err(err.into()),
+        }
+    })
+    .await?;
+    // Abandoned means abandoned: the replica stays on the former target and,
+    // once the attempts in flight at removal time have drained, no further
+    // purge attempts are sent to it.
+    assert_eq!(
+        single_target_version(&target, &target_bucket, key)?,
+        replica,
+        "an abandoned purge must not touch the replica on the former target"
+    );
+    sleep(Duration::from_secs(3)).await;
+    let settled = target.count_requests(FakeTargetOperation::DeleteObject, key);
+    sleep(Duration::from_secs(3)).await;
+    assert_eq!(
+        target.count_requests(FakeTargetOperation::DeleteObject, key),
+        settled,
+        "purge attempts must stop once the target is no longer configured"
+    );
+
+    target.shutdown().await;
+    Ok(())
+}
+
+fn single_target_version(target: &FakeS3Target, target_bucket: &str, key: &str) -> Result<String, Box<dyn Error + Send + Sync>> {
+    let versions = target.stored_versions(target_bucket, key);
+    match versions.as_slice() {
+        [(version_id, false)] => Ok(version_id.clone()),
+        other => Err(format!("{key}: expected exactly one live target version, got {other:?}").into()),
+    }
+}
+
+/// The target keeps holding exactly `version_id` for a few scanner cycles: a
+/// re-driven PUT or a wrong delete would show up here.
+async fn assert_stable_single_version(target: &FakeS3Target, target_bucket: &str, key: &str, version_id: &str) -> TestResult {
+    for _ in 0..8 {
+        let versions = target.stored_versions(target_bucket, key);
+        if versions.len() != 1 || versions[0].0 != version_id {
+            return Err(
+                format!("{key}: target versions drifted from the single expected replica {version_id}: {versions:?}").into(),
+            );
+        }
+        sleep(Duration::from_millis(500)).await;
+    }
+    Ok(())
+}
+
+async fn wait_until<F, Fut>(what: &str, mut probe: F) -> TestResult
+where
+    F: FnMut() -> Fut,
+    Fut: std::future::Future<Output = Result<bool, Box<dyn Error + Send + Sync>>>,
+{
+    let wait = async {
+        loop {
+            if probe().await? {
+                return Ok::<_, Box<dyn Error + Send + Sync>>(());
+            }
+            sleep(Duration::from_millis(250)).await;
+        }
+    };
+    timeout(Duration::from_secs(90), wait)
+        .await
+        .map_err(|_| format!("{what} did not happen within 90 seconds"))?
+}
+
 /// Wait until `key` is COMPLETED on the source and, for the observation
 /// window after that, the target still holds exactly one live version of it.
 async fn wait_for_replication_status_and_single_version(

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -33,6 +33,8 @@ use aws_sdk_s3::operation::get_object::{GetObjectError, GetObjectOutput};
 use aws_sdk_s3::operation::get_object_tagging::{GetObjectTaggingError, GetObjectTaggingOutput};
 use aws_sdk_s3::operation::head_bucket::HeadBucketError;
 use aws_sdk_s3::operation::head_object::HeadObjectError;
+use aws_sdk_s3::operation::put_object_legal_hold::{PutObjectLegalHoldError, PutObjectLegalHoldOutput};
+use aws_sdk_s3::operation::put_object_retention::{PutObjectRetentionError, PutObjectRetentionOutput};
 use aws_sdk_s3::operation::put_object_tagging::{PutObjectTaggingError, PutObjectTaggingOutput};
 use aws_sdk_s3::operation::upload_part::UploadPartOutput;
 use aws_sdk_s3::primitives::ByteStream;
@@ -42,6 +44,7 @@ use aws_sdk_s3::types::{
     ChecksumAlgorithm, ChecksumMode, CompletedMultipartUpload, CompletedPart, ObjectLockLegalHoldStatus, ObjectLockRetentionMode,
     ServerSideEncryption,
 };
+use aws_sdk_s3::types::{ObjectLockLegalHold, ObjectLockRetention};
 use aws_sdk_s3::{Client as S3Client, operation::head_object::HeadObjectOutput};
 use aws_smithy_runtime_api::client::orchestrator::HttpRequest;
 use futures::{StreamExt, stream};
@@ -139,9 +142,12 @@ fn same_replication_service(edited: &BucketTarget, previous: &BucketTarget) -> b
         && access_key(edited) == access_key(previous)
 }
 
-/// Page size and page budget for [`TargetClient::find_version_by_etag`].
+/// Page size and page budget for [`TargetClient::locate_replica_by_etag`].
 const FIND_VERSION_BY_ETAG_PAGE_SIZE: i32 = 1000;
 const FIND_VERSION_BY_ETAG_MAX_PAGES: usize = 8;
+/// Candidate cap for [`TargetClient::replica_candidates_by_etag`]: more than
+/// this many same-content versions of one key is ambiguity by any measure.
+const FIND_VERSION_BY_ETAG_MAX_MATCHES: usize = 16;
 pub type GetObjectSdkError = Box<SdkError<GetObjectError>>;
 pub type GetObjectTaggingSdkError = Box<SdkError<GetObjectTaggingError>>;
 pub type PutObjectTaggingSdkError = Box<SdkError<PutObjectTaggingError>>;
@@ -1363,6 +1369,7 @@ fn generate_arn(t: &BucketTarget, depl_id: &str) -> String {
     arn.to_string()
 }
 
+#[derive(Debug, Clone)]
 pub struct RemoveObjectOptions {
     pub force_delete: bool,
     pub governance_bypass: bool,
@@ -1971,22 +1978,29 @@ impl TargetClient {
             .map_err(Box::new)
     }
 
-    /// Locate a replica by content identity on a target that mints its own
-    /// version ids: page `ListObjectVersions` under the exact key and return
-    /// the newest live version whose ETag matches `source_etag`. Delete
-    /// markers and prefix siblings never match. Bounded to
-    /// [`FIND_VERSION_BY_ETAG_MAX_PAGES`] pages so a key with a very deep
-    /// history cannot turn one convergence check into an unbounded scan; a
-    /// replica beyond that window reads as missing, which only costs a
+    /// Candidate replicas by content identity on a target that mints its own
+    /// version ids: page `ListObjectVersions` under the exact key and report
+    /// the live versions whose ETag matches `source_etag`, newest first.
+    /// Delete markers and prefix siblings never match. Bounded to
+    /// [`FIND_VERSION_BY_ETAG_MAX_PAGES`] pages and
+    /// [`FIND_VERSION_BY_ETAG_MAX_MATCHES`] candidates so a key with a very
+    /// deep history cannot turn one convergence check into an unbounded scan;
+    /// a replica beyond that window reads as missing, which only costs a
     /// re-PUT (today's behaviour), never a lost object.
-    pub async fn find_version_by_etag(
+    ///
+    /// Content identity is not version identity: two source generations with
+    /// the same bytes have the same ETag. Callers drop the candidates other
+    /// source versions already claim through their ledgers and refuse an
+    /// [`ReplicaLocation::Ambiguous`] remainder before mutating or deleting.
+    pub async fn replica_candidates_by_etag(
         &self,
         bucket: &str,
         object: &str,
         source_etag: &str,
-    ) -> Result<Option<String>, Box<SdkError<aws_sdk_s3::operation::list_object_versions::ListObjectVersionsError>>> {
+    ) -> Result<Vec<String>, Box<SdkError<aws_sdk_s3::operation::list_object_versions::ListObjectVersionsError>>> {
         let mut key_marker: Option<String> = None;
         let mut version_id_marker: Option<String> = None;
+        let mut matches: Vec<String> = Vec::new();
         for _ in 0..FIND_VERSION_BY_ETAG_MAX_PAGES {
             let page = self
                 .client
@@ -1999,32 +2013,88 @@ impl TargetClient {
                 .send()
                 .await
                 .map_err(Box::new)?;
-            if let Some(version) = page.versions().iter().find(|version| {
-                version.key() == Some(object)
-                    && version.version_id().is_some_and(|id| !id.is_empty())
-                    && replication_etags_match(Some(source_etag), version.e_tag())
-            }) {
-                return Ok(version.version_id().map(str::to_string));
-            }
-            // Every listed key is >= the prefix; once the listing moved past
-            // the exact key there is nothing left to find.
-            if page
-                .versions()
-                .iter()
-                .any(|version| version.key().is_some_and(|key| key > object))
+            matches.extend(
+                page.versions()
+                    .iter()
+                    .filter(|version| {
+                        version.key() == Some(object)
+                            && version.version_id().is_some_and(|id| !id.is_empty())
+                            && replication_etags_match(Some(source_etag), version.e_tag())
+                    })
+                    .filter_map(|version| version.version_id().map(str::to_string)),
+            );
+            // A listing that moved past the exact key (every listed key is >=
+            // the prefix), ended, or already filled the candidate cap decides.
+            if matches.len() >= FIND_VERSION_BY_ETAG_MAX_MATCHES
+                || page
+                    .versions()
+                    .iter()
+                    .any(|version| version.key().is_some_and(|key| key > object))
+                || !page.is_truncated().unwrap_or(false)
             {
-                return Ok(None);
-            }
-            if !page.is_truncated().unwrap_or(false) {
-                return Ok(None);
+                break;
             }
             key_marker = page.next_key_marker().map(str::to_string);
             version_id_marker = page.next_version_id_marker().map(str::to_string);
             if key_marker.is_none() {
-                return Ok(None);
+                break;
             }
         }
-        Ok(None)
+        matches.truncate(FIND_VERSION_BY_ETAG_MAX_MATCHES);
+        Ok(matches)
+    }
+
+    /// PutObjectRetention against a replica version on a target that does not
+    /// take retention through the replication PUT's own headers (it mints its
+    /// own version ids, so a re-PUT would create another version instead of
+    /// updating this one). Anti-loop marker always added.
+    pub async fn put_object_retention(
+        &self,
+        bucket: &str,
+        object: &str,
+        version_id: Option<String>,
+        mode: ObjectLockRetentionMode,
+        retain_until: aws_sdk_s3::primitives::DateTime,
+    ) -> Result<PutObjectRetentionOutput, Box<SdkError<PutObjectRetentionError>>> {
+        let headers = proxy_outbound_headers(HeaderMap::new());
+        self.client
+            .put_object_retention()
+            .bucket(bucket)
+            .key(object)
+            .set_version_id(resolve_read_api_version_id(version_id))
+            .retention(
+                ObjectLockRetention::builder()
+                    .mode(mode)
+                    .retain_until_date(retain_until)
+                    .build(),
+            )
+            .customize()
+            .map_request(move |req| apply_extra_headers(req, &headers))
+            .send()
+            .await
+            .map_err(Box::new)
+    }
+
+    /// PutObjectLegalHold counterpart of [`Self::put_object_retention`].
+    pub async fn put_object_legal_hold(
+        &self,
+        bucket: &str,
+        object: &str,
+        version_id: Option<String>,
+        status: ObjectLockLegalHoldStatus,
+    ) -> Result<PutObjectLegalHoldOutput, Box<SdkError<PutObjectLegalHoldError>>> {
+        let headers = proxy_outbound_headers(HeaderMap::new());
+        self.client
+            .put_object_legal_hold()
+            .bucket(bucket)
+            .key(object)
+            .set_version_id(resolve_read_api_version_id(version_id))
+            .legal_hold(ObjectLockLegalHold::builder().status(status).build())
+            .customize()
+            .map_request(move |req| apply_extra_headers(req, &headers))
+            .send()
+            .await
+            .map_err(Box::new)
     }
 
     /// HEAD used by the read-proxy path (GET/HEAD of an object not yet
@@ -2478,6 +2548,45 @@ impl TargetClient {
     }
 }
 
+/// Where a replica stands on a target that mints its own version ids, by
+/// content identity (exact key + ETag) after the candidates other source
+/// versions claim were removed. See
+/// [`TargetClient::replica_candidates_by_etag`].
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ReplicaLocation {
+    /// No live version under the key carries the source ETag.
+    Missing,
+    /// Exactly one live version carries it: safe to address.
+    Unique(String),
+    /// More than one live version carries it (same bytes replicated for
+    /// several source generations). `newest` is the most recently listed
+    /// one — good enough to prove the replica exists, never good enough to
+    /// pick which one to mutate or delete.
+    Ambiguous { newest: String },
+}
+
+impl ReplicaLocation {
+    /// `matches` newest first, as the target listed them.
+    pub fn from_matches(mut matches: Vec<String>) -> Self {
+        match matches.len() {
+            0 => Self::Missing,
+            1 => Self::Unique(matches.remove(0)),
+            _ => Self::Ambiguous {
+                newest: matches.remove(0),
+            },
+        }
+    }
+
+    /// The version to read for existence/ETag checks, where an ambiguous
+    /// match is still a located replica.
+    pub fn any_version_id(&self) -> Option<&str> {
+        match self {
+            Self::Missing => None,
+            Self::Unique(version_id) | Self::Ambiguous { newest: version_id } => Some(version_id),
+        }
+    }
+}
+
 #[derive(Debug)]
 pub enum BucketTargetError {
     BucketRemoteTargetNotFound {

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

@@ -3079,7 +3079,11 @@ pub async fn queue_replication_heal(bucket: &str, oi: ObjectInfo, retry_count: u
     }
 
     let rcfg = match ReplicationMetadataStore::optional_replication_config(bucket).await {
-        Ok(Some(config)) => config,
+        Ok(Some(config)) => Some(config),
+        // A bucket without a configuration still owes its pending purges an
+        // answer: the delete worker finishes them locally as abandoned, which
+        // is what makes the bucket deletable again (rustfs/backlog#2340).
+        Ok(None) if !oi.version_purge_status.is_empty() => None,
         Ok(None) => return ReplicationQueueAdmission::Skipped,
         Err(err) => {
             debug!(
@@ -3129,7 +3133,7 @@ pub async fn queue_replication_heal(bucket: &str, oi: ObjectInfo, retry_count: u
         }
     };
 
-    let rcfg_wrapper = ReplicationConfig::new(Some(rcfg), tgts);
+    let rcfg_wrapper = ReplicationConfig::new(rcfg, tgts);
     queue_replication_heal_internal(bucket, oi, rcfg_wrapper, retry_count)
         .await
         .admission
@@ -3175,7 +3179,11 @@ pub(crate) async fn queue_replication_heal_internal(
         };
     }
 
-    if rcfg.config.is_none() || rcfg.remotes.is_none() {
+    // Without a configuration or targets there is nothing to replicate —
+    // except a version purge the bucket still owes: its stored decision names
+    // the targets, and the delete worker settles the ones no longer
+    // configured as abandoned (rustfs/backlog#2340).
+    if (rcfg.config.is_none() || rcfg.remotes.is_none()) && oi.version_purge_status.is_empty() {
         return ReplicationHealQueueResult {
             object_info: roi,
             admission: ReplicationQueueAdmission::Skipped,
@@ -3220,12 +3228,15 @@ pub(crate) async fn queue_replication_heal_internal(
         }
         ReplicationHealQueueAction::QueueDelete(dv) => {
             // A purge the peer denied under object lock cannot succeed until
-            // the lock lapses (#6850); requeuing it every heal cycle only
+            // the lock lapses (#6850), and one whose replica cannot be told
+            // apart on a target that mints its own version ids cannot
+            // succeed until the ledger or an operator resolves it
+            // (rustfs/backlog#2340); requeuing either every heal cycle only
             // burns bandwidth and failure counters. The backoff expires on
             // its own, so the purge is probed again — and converges — once
-            // the retention window has a chance of being over.
+            // the condition has a chance of being over.
             if super::replication_object_decision_boundary::is_version_delete_replication(&dv.delete_object)
-                && super::replication_resyncer::object_lock_denied_purge_backoff_active(&dv)
+                && super::replication_resyncer::purge_backoff_active(&dv)
             {
                 return ReplicationHealQueueResult {
                     object_info: roi,

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

@@ -38,7 +38,7 @@ use time::format_description::well_known::Rfc3339;
 
 pub(crate) use crate::bucket::bucket_target_sys::{
     AdvancedPutOptions, HeadObjectSdkError, PutObjectOptions, PutObjectPartOptions, RemotePutObjectResponse, RemoveObjectOptions,
-    S3ClientError, TargetClient, resolve_read_api_version_id,
+    ReplicaLocation, S3ClientError, TargetClient, resolve_read_api_version_id,
 };
 #[cfg(test)]
 pub(crate) use crate::bucket::target::BucketTarget;

crates/filemeta/src/replication.rs

@@ -442,6 +442,10 @@ pub struct ReplicatedTargetInfo {
     pub error: Option<String>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub target_delete_marker_version_id: Option<String>,
+    /// Kept in step with the replication crate's copy: the id a target that
+    /// mints its own version ids assigned to this object version.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub target_version_id: Option<String>,
 }
 
 impl ReplicatedTargetInfo {

crates/replication/src/config.rs

@@ -122,6 +122,10 @@ pub trait ReplicationConfigurationExt {
     fn has_active_rules(&self, prefix: &str, recursive: bool) -> bool;
     fn filter_target_arns(&self, obj: &ObjectOpts) -> Vec<String>;
     fn filter_force_delete_target_arns(&self, prefix: &str) -> Vec<String>;
+    /// Every target ARN the configuration still names, whatever the rule's
+    /// status, prefix or filter: the set a pending replication delete may
+    /// still be owed to. A target outside it was removed by the operator.
+    fn configured_target_arns(&self) -> HashSet<String>;
     fn filter_target_replication_decisions(&self, obj: &ObjectOpts) -> Vec<(String, bool)> {
         self.filter_target_arns(obj)
             .into_iter()
@@ -772,6 +776,19 @@ impl ReplicationConfigurationExt for ReplicationConfiguration {
     }
 
     /// Filter target ARNs and return a slice of the distinct values in the config
+    fn configured_target_arns(&self) -> HashSet<String> {
+        let role = self.role.trim();
+        if !role.is_empty() {
+            return HashSet::from([role.to_string()]);
+        }
+        self.rules
+            .iter()
+            .map(|rule| rule.destination.bucket.trim())
+            .filter(|arn| !arn.is_empty())
+            .map(str::to_string)
+            .collect()
+    }
+
     fn filter_target_arns(&self, obj: &ObjectOpts) -> Vec<String> {
         let role = self.role.trim();
         if !role.is_empty() {

crates/replication/src/filemeta.rs

@@ -438,6 +438,12 @@ pub struct ReplicatedTargetInfo {
     /// Version the target assigned to the delete marker it just created.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub target_delete_marker_version_id: Option<String>,
+    /// Version the target assigned to this object version when it differs
+    /// from the source id (a target that mints its own ids). Persisted as the
+    /// per-target ledger every later version-addressed mutation resolves
+    /// through; `None` on targets that adopt the source id.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub target_version_id: Option<String>,
 }
 
 impl ReplicatedTargetInfo {

crates/scanner/src/scanner_folder/item_actions.rs

@@ -15,7 +15,7 @@
 use super::*;
 #[cfg(test)]
 use rustfs_filemeta::MetadataResolutionParams;
-use sha2::{Digest as _, Sha256};
+use sha2::Sha256;
 
 /// Cached folder information for scanning
 #[derive(Clone, Debug)]
@@ -1036,12 +1036,19 @@ impl ScannerItem {
             return;
         }
 
-        let Some(replication) = self.replication.clone() else {
-            return;
+        let replication = match self.replication.clone() {
+            Some(replication) => (*replication).clone(),
+            // No active rules or targets, but a purge the bucket still owes
+            // must reach the heal path: the delete worker settles it against
+            // the current configuration (abandoned when the target is gone,
+            // rustfs/backlog#2340) so the hidden version stops blocking
+            // DeleteBucket.
+            None if !oi.version_purge_status.is_empty() => ReplicationConfig::new(None, None),
+            None => return,
         };
 
         let done_replication = Metrics::time(Metric::CheckReplication);
-        let replication_result = queue_replication_heal(&oi.bucket, oi.clone(), (*replication).clone(), 0).await;
+        let replication_result = queue_replication_heal(&oi.bucket, oi.clone(), replication, 0).await;
         done_replication();
         let roi = replication_result.object_info;
         record_scanner_replication_admission(global_metrics(), &roi, replication_result.admission);

crates/utils/src/http/metadata_compat.rs

@@ -93,6 +93,10 @@ pub const SUFFIX_TIER_SKIP_FV_ID: &str = "tier-skip-fvid";
 
 /// Per-target delete-marker version ids are stored one key per target ARN.
 pub const SUFFIX_REPLICATION_DELETE_MARKER_VERSION_ARN_PREFIX: &str = "replication-delete-marker-version-";
+/// Per-target data-version ids, one key per target ARN: the version a
+/// replication target that mints its own ids assigned to this object version
+/// (rustfs/backlog#2340). Absent on targets that adopt the source id.
+pub const SUFFIX_REPLICATION_TARGET_VERSION_ARN_PREFIX: &str = "replication-target-version-";
 
 // On-demand migration provenance. Written by the migration write-back onto
 // every pulled object so operators and later tooling can tell a migrated
@@ -314,6 +318,17 @@ pub fn strip_internal_prefix_preserving_case(key: &str) -> Option<&str> {
 /// Reads the bounded per-target delete-marker version map in one metadata scan.
 /// The boolean is set when matching metadata is malformed or compatibility keys disagree.
 pub fn target_delete_marker_versions(map: &HashMap<String, String>) -> (HashMap<String, String>, bool) {
+    internal_versions_by_arn(map, SUFFIX_REPLICATION_DELETE_MARKER_VERSION_ARN_PREFIX)
+}
+
+/// Reads the bounded per-target data-version ledger (the id each drifting
+/// target assigned to this object version) in one metadata scan. Same
+/// bounds and corruption reporting as [`target_delete_marker_versions`].
+pub fn replication_target_versions(map: &HashMap<String, String>) -> (HashMap<String, String>, bool) {
+    internal_versions_by_arn(map, SUFFIX_REPLICATION_TARGET_VERSION_ARN_PREFIX)
+}
+
+fn internal_versions_by_arn(map: &HashMap<String, String>, arn_prefix: &str) -> (HashMap<String, String>, bool) {
     const MAX_ENTRIES: usize = 1_000;
     const MAX_ARN_LEN: usize = 1_024;
     const MAX_VERSION_ID_LEN: usize = 1_024;
@@ -324,13 +339,13 @@ pub fn target_delete_marker_versions(map: &HashMap<String, String>) -> (HashMap<
         let Some(suffix) = strip_internal_prefix_preserving_case(key) else {
             continue;
         };
-        let Some(prefix) = suffix.get(..SUFFIX_REPLICATION_DELETE_MARKER_VERSION_ARN_PREFIX.len()) else {
+        let Some(prefix) = suffix.get(..arn_prefix.len()) else {
             continue;
         };
-        if !prefix.eq_ignore_ascii_case(SUFFIX_REPLICATION_DELETE_MARKER_VERSION_ARN_PREFIX) {
+        if !prefix.eq_ignore_ascii_case(arn_prefix) {
             continue;
         }
-        let arn = &suffix[SUFFIX_REPLICATION_DELETE_MARKER_VERSION_ARN_PREFIX.len()..];
+        let arn = &suffix[arn_prefix.len()..];
         if !arn.starts_with("arn:") || arn.len() > MAX_ARN_LEN || value.is_empty() || value.len() > MAX_VERSION_ID_LEN {
             corrupt = true;
             continue;
@@ -703,6 +718,37 @@ mod tests {
         assert!(corrupt);
     }
 
+    #[test]
+    fn replication_target_versions_are_keyed_apart_from_delete_marker_versions() {
+        let arn = "arn:rustfs:replication::target";
+        let mut metadata = HashMap::new();
+        insert_str(
+            &mut metadata,
+            &format!("{SUFFIX_REPLICATION_TARGET_VERSION_ARN_PREFIX}{arn}"),
+            "data-version".to_string(),
+        );
+        insert_str(
+            &mut metadata,
+            &format!("{SUFFIX_REPLICATION_DELETE_MARKER_VERSION_ARN_PREFIX}{arn}"),
+            "marker-version".to_string(),
+        );
+
+        let (data_versions, corrupt) = replication_target_versions(&metadata);
+        assert!(!corrupt);
+        assert_eq!(data_versions.get(arn).map(String::as_str), Some("data-version"));
+        let (marker_versions, corrupt) = target_delete_marker_versions(&metadata);
+        assert!(!corrupt);
+        assert_eq!(marker_versions.get(arn).map(String::as_str), Some("marker-version"));
+
+        metadata.insert(
+            format!("{MINIO_INTERNAL_PREFIX}{SUFFIX_REPLICATION_TARGET_VERSION_ARN_PREFIX}{arn}"),
+            "other-version".to_string(),
+        );
+        let (data_versions, corrupt) = replication_target_versions(&metadata);
+        assert!(data_versions.is_empty());
+        assert!(corrupt);
+    }
+
     #[test]
     fn target_delete_marker_versions_bound_distinct_entries_during_scan() {
         let metadata = (0..=1_000)

docs/operations/replication-check.md

@@ -62,4 +62,4 @@ The route returns HTTP 200 with JSON after all configured targets have been chec
 
 `VersionFidelity` pins the version-identity contract on both write paths. The probe PUT carries a source version id (header plus `?versionId=` query, the exact shape live replication uses) and the target must answer with the same id; a second probe repeats the check through CreateMultipartUpload -> UploadPart -> CompleteMultipartUpload, where the target fixes the version at initiate and only reports it on completion. A target can adopt PutObject ids and still mint its own for multipart; the failure message names the path that drifted.
 
-A target that mints its own version ids breaks every version-addressed operation that follows (version deletes, heal re-drives). The phase therefore fails with `"Code": "BucketRemoteTargetVersionMismatch"`, the later mutation phases are skipped, and cleanup still removes the probe via the version id the target actually assigned.
+A target that mints its own version ids never answers to the source version id. The phase therefore fails with `"Code": "BucketRemoteTargetVersionMismatch"` and the target result is `FAILED`. Replication to such a target still converges: the replication worker records the id the target assigned to each object version on the source (the target-version ledger, internal metadata key `replication-target-version-<arn>`) and addresses version deletes, tag and Object Lock updates through it. The `DeleteMarker` and `VersionDelete` phases probe exactly that path — they address the id the target assigned to the probe object, not the source id — so on a drifting target they report whether ledger-addressed purges work against this endpoint, and cleanup removes the probe via the same id. They are `SKIPPED` only when the probe `Put` itself failed or reported no version id.

docs/operations/replication-outbound-transport.md

@@ -18,7 +18,7 @@
 | Rejects or mis-stores `aws-chunked` bodies (SeaweedFS 3.97) | Handled by the plain-payload default above. | Outbound target matrix, `RejectAwsChunked` mode |
 | Requires `Content-MD5` or `x-amz-checksum-*` on a PutObject with Object Lock parameters (AWS S3, MinIO, Impossible Cloud, most compatible stores) | Satisfied: a locked single PUT carries `Content-MD5` derived from the source ETag (plaintext objects whose ETag is the MD5 of the wire bytes) or an SDK CRC32 checksum (multipart-layout ETags, managed SSE, SSE-C passthrough — this one is an `aws-chunked` trailer, so a target that also rejects that framing cannot take such objects). Releases before this fix (`1.0.0-rc.5`) need `RUSTFS_REPLICATION_STREAMING_CHECKSUMS=true` as a workaround. | Outbound target matrix, `RequireChecksumWithObjectLock` mode |
 | Stores `x-amz-checksum-*` from a PutObject and returns it on `HEAD ?ChecksumMode=ENABLED` (AWS S3, Wasabi, RustFS) | Satisfied for single-part objects: the replica answers with the source's checksum. Before this fix (`1.0.0-rc.5`) the checksum left the source as `x-amz-meta-<algorithm>` user metadata and no replica carried it (rustfs/backlog#2340). | Outbound target matrix, `Checksummed` shape |
-| Mints its own version ids (AWS S3, Wasabi, Impossible Cloud) | Data lands; version-addressed convergence does not. See rustfs/backlog#2085 and `docs/operations/replication-check.md` (VersionFidelity). | `replication-check`, outbound target matrix, `MintOwnVersionIds` mode |
+| Mints its own version ids (AWS S3, Wasabi, Impossible Cloud) | Data lands, and later mutations converge through the target-version ledger: the id the target assigned is recorded on the source version (internal key `replication-target-version-<arn>`) and version deletes, tag and Object Lock updates address it (tag/retention/legal-hold through the metadata-only APIs, never a re-PUT). A replica written before the ledger existed is located by exact key + ETag, minus the candidates other generations of the key already claim; an ambiguous remainder is refused with a one-hour backoff (`replication_purge_replica_unresolved`). `NoSuchVersion` on a version-addressed DELETE counts as purged. See rustfs/backlog#2340, rustfs/backlog#2085 and `docs/operations/replication-check.md` (VersionFidelity). | `replication-check`, outbound target matrix, `MintOwnVersionIds` mode (also models Wasabi's 404 `NoSuchVersion` on unknown ids) |
 | Returns an ETag that is not the content MD5 without announcing SSE | Every single-part object fails ETag verification. Set `RUSTFS_REPLICATION_REPLICA_ETAG_VERIFY=false`. | Replication status FAILED with `replica etag mismatch` |
 
 ## Environment knobs

rustfs/src/admin/router.rs

@@ -2245,12 +2245,13 @@ async fn execute_replication_probe(result: &mut ReplicationCheckTargetStatus, op
     match operations.put().await {
         Ok(outcome) => {
             result.phases.put = ReplicationCheckPhaseStatus::passed();
-            // P1-19 version-identity contract: replication only converges on
-            // targets that adopt the source version id — version-addressed
-            // deletes and heal re-drives never match a minted id. Judge it
-            // from the probe PUT's own response; on mismatch the later
-            // mutation phases are pointless (they address by version id), but
-            // cleanup still runs against whatever id the target assigned.
+            // P1-19 version-identity contract: a target that mints its own
+            // version ids never answers to the source id. Judge it from the
+            // probe PUT's own response. The mutation phases below still run
+            // on such a target: they address the id the target assigned —
+            // the same ledger the replication worker records per object
+            // (rustfs/backlog#2340) — so they report whether version-
+            // addressed deletes can converge there at all.
             match version_fidelity_error("PutObject", &outcome) {
                 None => result.phases.version_fidelity = ReplicationCheckPhaseStatus::passed(),
                 Some(error) => {
@@ -2325,7 +2326,12 @@ async fn execute_replication_probe(result: &mut ReplicationCheckTargetStatus, op
         }
     }
 
-    if result.phases.put.status == "OK" && result.phases.version_fidelity.status == "OK" {
+    // DeleteMarker / VersionDelete address `probe_version_id`, the id the
+    // target actually assigned, so they run on a drifting target too. What
+    // they cannot prove there is the source-id addressing VersionFidelity
+    // already failed; what they do prove is that the ledger-addressed purge
+    // path the worker uses on such a target works against this endpoint.
+    if result.phases.put.status == "OK" && probe_version_id.is_some() {
         match operations.create_delete_marker(probe_version_id.as_deref()).await {
             Ok(version_id) => {
                 delete_marker_version_id = version_id;
@@ -3894,6 +3900,8 @@ mod tests {
         version_delete_error: Option<&'static str>,
         cleanup_error: Option<&'static str>,
         calls: Vec<&'static str>,
+        /// Version ids the delete-marker and version-delete phases addressed.
+        mutation_ids: Vec<Option<String>>,
         cleanup_ids: Vec<Option<String>>,
     }
 
@@ -3938,16 +3946,18 @@ mod tests {
             }
         }
 
-        async fn create_delete_marker(&mut self, _version_id: Option<&str>) -> Result<Option<String>, S3ClientError> {
+        async fn create_delete_marker(&mut self, version_id: Option<&str>) -> Result<Option<String>, S3ClientError> {
             self.calls.push("delete-marker");
+            self.mutation_ids.push(version_id.map(ToOwned::to_owned));
             match self.delete_marker_error {
                 Some(code) => Err(scripted_probe_error(code)),
                 None => Ok(Some("marker-version".to_string())),
             }
         }
 
-        async fn delete_version(&mut self, _version_id: Option<&str>) -> Result<(), S3ClientError> {
+        async fn delete_version(&mut self, version_id: Option<&str>) -> Result<(), S3ClientError> {
             self.calls.push("version-delete");
+            self.mutation_ids.push(version_id.map(ToOwned::to_owned));
             match self.version_delete_error {
                 Some(code) => Err(scripted_probe_error(code)),
                 None => Ok(()),
@@ -3968,12 +3978,12 @@ mod tests {
     }
 
     /// P1-19: a target that mints its own version ids must fail the
-    /// VersionFidelity phase with the machine-readable mismatch code, skip
-    /// the version-addressed mutation phases (they cannot mean anything on a
-    /// drifting target), and still clean up using the id the target actually
-    /// assigned — the source-derived id would never match.
+    /// VersionFidelity phase with the machine-readable mismatch code. The
+    /// mutation phases still run, addressing the id the target assigned
+    /// (the ledger the worker records per object, rustfs/backlog#2340), and
+    /// cleanup uses that id too — the source-derived id would never match.
     #[tokio::test]
-    async fn replication_probe_flags_version_minting_target() {
+    async fn replication_probe_flags_version_minting_target_and_probes_mutations_by_assigned_id() {
         let mut result = replication_check_target("arn:a", "OK", None);
         let mut operations = ScriptedReplicationProbe {
             minted_version_id: Some("target-minted-version"),
@@ -3982,16 +3992,52 @@ mod tests {
 
         execute_replication_probe(&mut result, &mut operations).await;
 
-        assert_eq!(operations.calls, ["put", "cleanup"]);
+        assert_eq!(operations.calls, ["put", "delete-marker", "version-delete", "cleanup"]);
         assert_eq!(result.status, "FAILED");
         assert_eq!(result.phases.put.status, "OK");
         assert_eq!(result.phases.version_fidelity.status, "FAILED");
         assert_eq!(result.phases.version_fidelity.code, Some(REPLICATION_CHECK_CODE_VERSION_MISMATCH));
-        assert_eq!(result.phases.delete_marker.status, "SKIPPED");
-        assert_eq!(result.phases.version_delete.status, "SKIPPED");
+        assert_eq!(result.phases.delete_marker.status, "OK");
+        assert_eq!(result.phases.version_delete.status, "OK");
+        assert_eq!(
+            operations.mutation_ids,
+            [
+                Some("target-minted-version".to_string()),
+                Some("target-minted-version".to_string())
+            ],
+            "both mutation phases must address the id the target assigned"
+        );
         assert_eq!(result.phases.ssec_passthrough.status, "SKIPPED");
         assert_eq!(result.phases.cleanup.status, "OK");
-        assert_eq!(operations.cleanup_ids, [Some("target-minted-version".to_string()), None, None, None]);
+        assert_eq!(
+            operations.cleanup_ids,
+            [
+                Some("target-minted-version".to_string()),
+                None,
+                None,
+                Some("marker-version".to_string())
+            ]
+        );
+    }
+
+    /// A drifting target that also refuses the ledger-addressed delete keeps
+    /// the phase-level evidence: VersionDelete fails on its own, apart from
+    /// the identity verdict.
+    #[tokio::test]
+    async fn replication_probe_reports_version_delete_failure_on_a_drifting_target() {
+        let mut result = replication_check_target("arn:a", "OK", None);
+        let mut operations = ScriptedReplicationProbe {
+            minted_version_id: Some("target-minted-version"),
+            version_delete_error: Some("AccessDenied"),
+            ..Default::default()
+        };
+
+        execute_replication_probe(&mut result, &mut operations).await;
+
+        assert_eq!(result.phases.version_fidelity.status, "FAILED");
+        assert_eq!(result.phases.delete_marker.status, "OK");
+        assert_eq!(result.phases.version_delete.status, "FAILED");
+        assert_eq!(result.phases.cleanup.status, "OK");
     }
 
     #[tokio::test]