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

A permanent version delete whose replication keeps failing stays in
xl.meta as a PENDING purge, hidden from listings, until every target
confirms it. Once the operator removes the replication configuration or
the rule naming that target nothing ever confirms it: the heal path
derived its delete decision from the configuration (the decision string
is not persisted) and skipped the version forever, so DeleteBucket
answered BucketNotEmpty for a residue the client could neither list nor
remove (rustfs/backlog#2340).

Owe a version purge to the targets its purge state names, let the heal
path through without a configuration, and have the delete worker settle
a target the configuration no longer names as abandoned: the purge is
reported complete locally through the normal writeback, the replica on
the former target is left alone, and the event
replication_purge_abandoned plus a counter are the record.

crates/e2e_test/src/replication_extension_test.rs

@@ -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>> {

crates/e2e_test/src/replication_target_matrix_test.rs

@@ -36,11 +36,12 @@ 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, put_bucket_replication_with_delete_statuses, 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, ObjectLockLegalHold,
@@ -640,6 +641,141 @@ async fn matrix_mint_own_version_ids_addresses_mutations_through_the_ledger() ->
     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() {

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,

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

@@ -22,7 +22,8 @@ use super::replication_filemeta_boundary::ReplicationGenerationSnapshot;
 use super::replication_filemeta_boundary::{
     REPLICATE_EXISTING, ReplicateDecision, ReplicateObjectInfo, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction,
     ReplicationState, ReplicationStatusType, ReplicationType, VersionPurgeStatusType, get_replication_state,
-    parse_replicate_decision, replication_statuses_map, target_reset_header, version_purge_statuses_map,
+    parse_replicate_decision, replicate_decision_for_admitted_targets, replication_statuses_map, target_reset_header,
+    version_purge_statuses_map,
 };
 use super::replication_lock_boundary::ReplicationLockTiming;
 use super::replication_logging::{EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_REPLICATION_RESYNC};
@@ -130,6 +131,8 @@ const EVENT_REPLICATION_DRIFTED_REPLICA_LOCATED: &str = "replication_drifted_rep
 const EVENT_REPLICATION_OBJECT_FAILED: &str = "replication_object_failed";
 const EVENT_REPLICATION_PURGE_OBJECT_LOCK_DENIED: &str = "replication_purge_object_lock_denied";
 const EVENT_REPLICATION_PURGE_REPLICA_UNRESOLVED: &str = "replication_purge_replica_unresolved";
+const EVENT_REPLICATION_PURGE_ABANDONED: &str = "replication_purge_abandoned";
+const METRIC_VERSION_PURGE_ABANDONED_TOTAL: &str = "rustfs_replication_version_purge_abandoned_total";
 const EVENT_REPLICATION_DRIFTED_REPLICA_METADATA_SYNCED: &str = "replication_drifted_replica_metadata_synced";
 const METRIC_VERSION_PURGE_REPLICA_TOTAL: &str = "rustfs_replication_version_purge_replica_total";
 
@@ -2052,6 +2055,26 @@ pub async fn get_heal_replicate_object_info(oi: &ObjectInfo, rcfg: &ReplicationC
 
     let target_statuses = replication_statuses_map(&oi.replication_status_internal.clone().unwrap_or_default());
     let target_purge_statuses = version_purge_statuses_map(&oi.version_purge_status_internal.clone().unwrap_or_default());
+    // A version purge is owed to the targets its purge state names, whatever
+    // the configuration says now: the decision string is not persisted, so a
+    // heal after restart (or after the configuration was removed or edited)
+    // would otherwise never revisit the purge and the hidden version would
+    // block DeleteBucket forever (rustfs/backlog#2340). The delete worker
+    // settles a target the configuration no longer names as abandoned.
+    let dsc = if delete_path && !dsc.replicate_any() {
+        let owed: Vec<String> = target_purge_statuses
+            .iter()
+            .filter(|(_, status)| matches!(status, VersionPurgeStatusType::Pending | VersionPurgeStatusType::Failed))
+            .map(|(arn, _)| arn.clone())
+            .collect();
+        if owed.is_empty() {
+            dsc
+        } else {
+            replicate_decision_for_admitted_targets(&owed)
+        }
+    } else {
+        dsc
+    };
     let existing_obj_resync = if delete_path && !has_stored_delete_decision && !delete_state.0 && !delete_state.1 {
         Default::default()
     } else {
@@ -2408,6 +2431,11 @@ pub(crate) async fn replicate_delete_with_outcome<S: ReplicationStorage>(
     };
 
     let purge_source = version_purge_source(&storage, &bucket, &dobj, &dsc).await.map(Arc::new);
+    let configured_arns = if is_version_delete_replication(&dobj.delete_object) {
+        configured_replication_arns(&bucket).await
+    } else {
+        None
+    };
 
     let mut join_set = JoinSet::new();
 
@@ -2429,6 +2457,17 @@ pub(crate) async fn replicate_delete_with_outcome<S: ReplicationStorage>(
             continue;
         }
 
+        // The bucket no longer replicates to this target: nothing can ever
+        // confirm the purge remotely, so finish it locally as abandoned.
+        if let Some(configured) = configured_arns.as_ref()
+            && !configured.contains(&tgt_entry.arn)
+        {
+            rinfos
+                .targets
+                .push(abandoned_purge_target_info(&bucket, &dobj, &tgt_entry.arn));
+            continue;
+        }
+
         // Get the remote target client
         let Some(tgt_client) = ReplicationTargetStore::remote_target_client(&bucket, &tgt_entry.arn).await else {
             debug!(
@@ -3199,6 +3238,56 @@ fn unavailable_delete_target_info(dobj: &DeletedObjectReplicationInfo, arn: &str
     rinfo
 }
 
+/// The target ARNs the bucket's replication configuration still names, or
+/// `None` when that cannot be decided right now (unreadable/invalid
+/// configuration): a purge is only abandoned on positive evidence. No
+/// configuration at all names no target.
+async fn configured_replication_arns(bucket: &str) -> Option<HashSet<String>> {
+    match get_replication_config(bucket).await {
+        Ok(Some(config)) => Some(config.configured_target_arns()),
+        Ok(None) => Some(HashSet::new()),
+        Err(_) => None,
+    }
+}
+
+/// Finish a version purge locally for a target the bucket no longer
+/// replicates to (the rule or the whole configuration was removed).
+///
+/// The source keeps a purged version in xl.meta, hidden from listings, until
+/// every target confirms the purge — and once the operator removed the
+/// target nothing ever will: the version stayed PENDING forever, blocking
+/// `DeleteBucket` with a residue the client could neither see nor remove
+/// (rustfs/backlog#2340). Reporting the purge as complete lets the normal
+/// writeback drop the version. The replica, if any, stays on the former
+/// target: that is the operator's data now, and this event is the record.
+fn abandoned_purge_target_info(bucket: &str, dobj: &DeletedObjectReplicationInfo, arn: &str) -> ReplicatedTargetInfo {
+    let mut rinfo = dobj
+        .delete_object
+        .replication_state
+        .clone()
+        .unwrap_or_default()
+        .target_state(arn);
+    rinfo.op_type = dobj.op_type;
+    if rinfo.version_purge_status == VersionPurgeStatusType::Complete {
+        return rinfo;
+    }
+    warn!(
+        event = EVENT_REPLICATION_PURGE_ABANDONED,
+        component = LOG_COMPONENT_ECSTORE,
+        subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+        bucket,
+        object = dobj.delete_object.object_name,
+        version_id = ?dobj.delete_object.version_id.or(dobj.delete_object.delete_marker_version_id),
+        arn,
+        reason = "target_not_configured",
+        "Replicated version purge abandoned: the bucket no longer replicates to this target, so the version is purged locally"
+    );
+    counter!(METRIC_VERSION_PURGE_ABANDONED_TOTAL).increment(1);
+    rinfo.version_purge_status = VersionPurgeStatusType::Complete;
+    rinfo.error = None;
+    rinfo
+}
+
 /// What the source still knows about a data version being purged, read once
 /// per delete: the version stays in xl.meta with a PENDING purge status until
 /// every target confirms, so its ETag and target-version ledger are available
@@ -6719,6 +6808,27 @@ mod tests {
         assert_eq!(server.join().expect("test HTTP server should finish").len(), 2);
     }
 
+    #[test]
+    fn abandoned_purge_completes_locally_and_keeps_a_finished_target_untouched() {
+        let arn = "arn:rustfs:replication::removed-target";
+        let dobj = version_purge_dobj(arn);
+        let rinfo = abandoned_purge_target_info("source", &dobj, arn);
+        assert_eq!(rinfo.version_purge_status, VersionPurgeStatusType::Complete);
+        assert_eq!(rinfo.arn, arn);
+        assert!(rinfo.error.is_none());
+
+        let mut finished = version_purge_dobj(arn);
+        finished
+            .delete_object
+            .replication_state
+            .as_mut()
+            .expect("purge state")
+            .purge_targets
+            .insert(arn.to_string(), VersionPurgeStatusType::Complete);
+        let rinfo = abandoned_purge_target_info("source", &finished, arn);
+        assert_eq!(rinfo.version_purge_status, VersionPurgeStatusType::Complete);
+    }
+
     #[tokio::test]
     async fn version_purge_refuses_a_corrupt_ledger() {
         let target = test_target_client("http://127.0.0.1:1".to_string());
@@ -7392,6 +7502,42 @@ mod tests {
         assert_eq!(roi.target_purge_statuses.get(role), Some(&VersionPurgeStatusType::Pending));
     }
 
+    /// The decision string is not persisted: after a restart, or once the
+    /// configuration is gone, a heal of a failed purge must still name the
+    /// targets the purge state records — that is what lets the delete worker
+    /// settle a removed target as abandoned instead of skipping forever.
+    #[tokio::test]
+    async fn heal_owes_a_failed_purge_to_the_targets_its_purge_state_names_without_a_configuration() {
+        let bucket = format!("heal-orphaned-purge-{}", Uuid::new_v4());
+        let arn = "arn:rustfs:replication:us-east-1:removed:bucket";
+        ReplicationVersioningStore::install_prefix_state_test_config(
+            &bucket,
+            VersioningConfiguration {
+                status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
+                ..Default::default()
+            },
+        );
+        let oi = ObjectInfo {
+            bucket,
+            name: "purge/orphaned.bin".to_string(),
+            version_id: Some(Uuid::new_v4()),
+            mod_time: Some(OffsetDateTime::now_utc()),
+            version_purge_status: VersionPurgeStatusType::Failed,
+            version_purge_status_internal: Some(format!("{arn}=FAILED;")),
+            ..Default::default()
+        };
+
+        let mut roi = get_heal_replicate_object_info(&oi, &ReplicationConfig::new(None, None))
+            .await
+            .expect("a purge without a configuration must still classify");
+
+        assert!(roi.dsc.targets_map.get(arn).is_some_and(|target| target.replicate), "{:?}", roi.dsc);
+        assert!(matches!(
+            super::super::replication_queue_boundary::replication_heal_queue_action(&mut roi),
+            super::super::replication_queue_boundary::ReplicationHealQueueAction::QueueDelete(_)
+        ));
+    }
+
     #[tokio::test]
     async fn heal_pending_purge_reads_one_versioning_generation() {
         let bucket = format!("heal-versioning-snapshot-{}", Uuid::new_v4());

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/scanner/src/scanner_folder/item_actions.rs

@@ -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);