From 29f47fda75ff85622adf885b4a8de686ef122b91 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Mon, 7 Sep 2026 16:35:58 +0800 Subject: [PATCH] fix(replication): abandon purges to targets the bucket no longer names 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. --- .../src/replication_extension_test.rs | 2 +- .../src/replication_target_matrix_test.rs | 142 ++++++++++++++++- .../bucket/replication/replication_pool.rs | 14 +- .../replication/replication_resyncer.rs | 148 +++++++++++++++++- crates/replication/src/config.rs | 17 ++ .../src/scanner_folder/item_actions.rs | 13 +- 6 files changed, 325 insertions(+), 11 deletions(-) diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index 92f6e166a..eb3a3d493 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/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> { diff --git a/crates/e2e_test/src/replication_target_matrix_test.rs b/crates/e2e_test/src/replication_target_matrix_test.rs index 6d2d80e70..2a3cb973e 100644 --- a/crates/e2e_test/src/replication_target_matrix_test.rs +++ b/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> { let versions = target.stored_versions(target_bucket, key); match versions.as_slice() { diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index 34c85ac25..ac502602d 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/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, diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 9e1895468..a6b762e88 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/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 = 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( }; 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( 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> { + 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()); diff --git a/crates/replication/src/config.rs b/crates/replication/src/config.rs index e413434cf..c7dba3508 100644 --- a/crates/replication/src/config.rs +++ b/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; fn filter_force_delete_target_arns(&self, prefix: &str) -> Vec; + /// 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; 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 { + 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 { let role = self.role.trim(); if !role.is_empty() { diff --git a/crates/scanner/src/scanner_folder/item_actions.rs b/crates/scanner/src/scanner_folder/item_actions.rs index 5120f8181..d216aa02b 100644 --- a/crates/scanner/src/scanner_folder/item_actions.rs +++ b/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);