diff --git a/.config/nextest.toml b/.config/nextest.toml index ee5e72657..3073dae55 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -218,7 +218,7 @@ test-group = 'ecstore-serial-flaky' # the nightly profile derives its set as "the replication module MINUS this # allowlist", so any new replication test lands in nightly by default (never # silently unrun) until it is explicitly blessed as fast here. Keep the two -# regexes byte-identical. Count invariant: 20 here + 36 nightly = 56 total +# regexes byte-identical. Count invariant: 20 here + 43 nightly = 63 total # (authority: `cargo nextest list`; docs/testing/e2e-suite-inventory.md). # HISTORY (2026-07-11): the 20 fast tests were briefly pulled out of this lane # (#4724) because they set a loopback (127.0.0.1) replication target that the @@ -344,7 +344,7 @@ path = "junit.xml" # object_lambda) — too heavy for the merge budget; they run in ci-7's # nightly 4-node lane. # * replication_extension_test — repl-1 already splits it into the PR -# `e2e-smoke` (20 fast) and `e2e-repl-nightly` (27 slow) lanes and reserves +# `e2e-smoke` (20 fast) and `e2e-repl-nightly` (43 slow) lanes and reserves # it for those, so e2e-full does not double-run it. # * #[ignore]d tests — nextest skips them by default (no --run-ignored); the # manual-localhost:9000 reliant/policy tests are ci-13's migration. diff --git a/crates/e2e_test/src/fake_s3_target/mod.rs b/crates/e2e_test/src/fake_s3_target/mod.rs index ba47db6f0..c40058e8d 100644 --- a/crates/e2e_test/src/fake_s3_target/mod.rs +++ b/crates/e2e_test/src/fake_s3_target/mod.rs @@ -141,6 +141,7 @@ struct ControlState { #[derive(Default)] struct StoreState { + assign_own_version_ids: bool, buckets: HashMap, uploads: HashMap, total_bytes: usize, @@ -383,6 +384,12 @@ impl FakeS3Target { .is_some_and(|version| !version.delete_marker) } + /// Make the target mint its own version ids instead of mirroring the + /// forwarded source version id — models a generic S3 service. + pub fn assign_own_version_ids(&self, enabled: bool) { + lock(&self.backend.store).assign_own_version_ids = enabled; + } + /// Queue `times` copies of a fault for one operation. pub fn inject(&self, operation: Operation, action: FaultAction, times: usize) { if times == 0 { @@ -433,6 +440,25 @@ impl FakeS3Target { lock(&self.control).requests.drain(..).collect() } + /// Stored versions for one key as `(version_id, is_delete_marker)`, oldest + /// first. Empty when the bucket or key does not exist. Lets purge tests + /// assert on the target's actual state instead of inferring it from the + /// request journal (a versioned DELETE is a silent no-op for missing ids). + pub fn stored_versions(&self, bucket: &str, key: &str) -> Vec<(String, bool)> { + let state = lock(&self.backend.store); + state + .buckets + .get(bucket) + .and_then(|bucket_state| bucket_state.objects.get(key)) + .map(|versions| { + versions + .iter() + .map(|version| (version.version_id.clone(), version.delete_marker)) + .collect() + }) + .unwrap_or_default() + } + pub async fn shutdown(mut self) { let _ = self.shutdown.send(true); if let Some(task) = self.task.take() { @@ -689,10 +715,17 @@ fn validate_retained_identifier(value: String, field: &str) -> S3Result } } -fn new_version_id(headers: &HeaderMap) -> S3Result { +/// `assign_own` models a target that mints its own version ids (a generic S3 +/// service): the forwarded source-version-id header is validated but NOT +/// mirrored into the stored version. +fn new_version_id(headers: &HeaderMap, assign_own: bool) -> S3Result { let Some(value) = header_value(headers, &SOURCE_VERSION_ID_HEADERS) else { return Ok(Uuid::new_v4().to_string()); }; + if assign_own { + validate_retained_identifier(value.trim().to_owned(), "source version ID")?; + return Ok(Uuid::new_v4().to_string()); + } let value = validate_retained_identifier(value.trim().to_owned(), "source version ID")?; let version_id = Uuid::parse_str(&value).map_err(|_| s3s::s3_error!(InvalidArgument, "source version ID must be a UUID"))?; Ok(version_id.to_string()) @@ -1078,7 +1111,8 @@ impl S3 for FakeBackend { let input = req.input; let body = collect_stream(input.body, input.content_length, fault.as_ref(), &self.control).await?; validate_stored_metadata(&input.content_type, &input.metadata)?; - let version_id = new_version_id(&headers)?; + let assign_own = lock(&self.store).assign_own_version_ids; + let version_id = new_version_id(&headers, assign_own)?; let e_tag = match source_etag(&headers)? { Some(value) => value, None => { @@ -1212,7 +1246,9 @@ impl S3 for FakeBackend { )); } - let version_id = new_version_id(&headers)?; + // `state` is the live store guard: read the flag from it. Re-locking + // would self-deadlock (the store mutex is not reentrant). + let version_id = new_version_id(&headers, state.assign_own_version_ids)?; upsert_version( &mut state, &input.bucket, @@ -1252,12 +1288,15 @@ impl S3 for FakeBackend { ensure_upload_budget(&state)?; validate_stored_metadata(&input.content_type, &input.metadata)?; let upload_id = Uuid::new_v4().to_string(); + // Read the flag before the mutable borrow of `state.uploads` below + // (and never re-lock the store: the mutex is not reentrant). + let version_id = new_version_id(&headers, state.assign_own_version_ids)?; state.uploads.insert( upload_id.clone(), MultipartState { bucket: input.bucket.clone(), key: input.key.clone(), - version_id: new_version_id(&headers)?, + version_id, content_type: input.content_type, metadata: input.metadata, parts: BTreeMap::new(), diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index 7ca4688e5..bd3e72223 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -18,6 +18,7 @@ use crate::common::{ }; use crate::fake_s3_target::{ FAKE_ACCESS_KEY, FAKE_SECRET_KEY, FakeS3Target, FaultAction as FakeTargetFault, Operation as FakeTargetOperation, + RequestRecord, }; use crate::kms::common::{create_key_with_specific_id, sse_customer_key_md5_base64}; use crate::storage_api::replication_extension::BucketTargetSys; @@ -7487,3 +7488,335 @@ async fn test_scanner_never_cascades_inbound_replicas() -> TestResult { Ok(()) } + +// --- P1-21 (backlog#1675): delayed delete-marker purge failure handling --- +// +// The fixtures below wire a versioned source bucket to a FakeS3Target with the +// default replication shape: DeleteMarkerReplication=Enabled and +// DeleteReplication omitted. With version-delete replication unconfigured, +// purging the source marker version emits no replication event, and the data +// scanner cannot see a source version that is gone — the delayed purge watcher +// spawned by the marker replication is the ONLY channel that can remove the +// replicated marker from the target. + +const DELAYED_PURGE_KEY: &str = "doc.txt"; + +fn delayed_purge_process_env() -> Vec<(&'static str, &'static str)> { + let mut env = replication_fast_env(); + env.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV); + env.extend_from_slice(&[("NO_PROXY", "127.0.0.1,localhost"), ("HTTP_PROXY", ""), ("HTTPS_PROXY", "")]); + env +} + +async fn start_delayed_purge_fixture( + source_bucket: &str, + target_bucket: &str, +) -> Result<(FakeS3Target, RustFSTestEnvironment, Client), Box> { + let target = FakeS3Target::start().await?; + target.create_bucket(target_bucket); + + let mut source_env = RustFSTestEnvironment::new().await?; + source_env + .start_rustfs_server_with_env(vec![], &delayed_purge_process_env()) + .await?; + + let source_client = source_env.create_s3_client(); + source_client.create_bucket().bucket(source_bucket).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, + secure: false, + skip_tls_verify: false, + ca_cert_pem: None, + }, + ) + .await?; + put_bucket_replication(&source_env, source_bucket, &target_arn).await?; + + Ok((target, source_env, source_client)) +} + +/// PUT an object, stack a delete marker on it, and wait until the fake target +/// stores the marker replica. Returns the source marker version id — the fake +/// target mirrors it because delete replication forwards +/// `x-*-source-version-id`. +/// +/// Timing budget for callers: the delayed purge watcher only observes the +/// source for ~4s after the marker replication completes, so the source-side +/// marker-version DELETE must be issued promptly after this returns (the +/// 100ms journal poll below keeps the detection latency small). +async fn replicate_delete_marker( + target: &FakeS3Target, + target_bucket: &str, + source_client: &Client, + source_bucket: &str, +) -> Result> { + source_client + .put_object() + .bucket(source_bucket) + .key(DELAYED_PURGE_KEY) + .body(ByteStream::from_static(b"delayed purge payload")) + .send() + .await?; + + let delete = source_client + .delete_object() + .bucket(source_bucket) + .key(DELAYED_PURGE_KEY) + .send() + .await?; + assert_eq!(delete.delete_marker(), Some(true), "unversioned DELETE must create a marker"); + let marker_version = delete + .version_id() + .ok_or("source DELETE omitted the marker version ID")? + .to_string(); + + // Wait for ANY delete marker: a target that mints its own version ids + // does not mirror the source one. + let deadline = tokio::time::Instant::now() + Duration::from_secs(30); + loop { + let replicated = target + .stored_versions(target_bucket, DELAYED_PURGE_KEY) + .iter() + .any(|(_, delete_marker)| *delete_marker); + if replicated { + return Ok(marker_version); + } + if tokio::time::Instant::now() >= deadline { + return Err( + format!("fake target never stored the replicated delete marker; journal: {:?}", target.requests()).into(), + ); + } + sleep(Duration::from_millis(100)).await; + } +} + +/// Journal records of purge attempts: target DELETE calls addressing the marker +/// version explicitly. The marker-creation replica DELETE carries no +/// `versionId` query, so the version id is an exact discriminator. +fn delayed_purge_attempts(target: &FakeS3Target, marker_version: &str) -> Vec { + target + .requests() + .into_iter() + .filter(|record| { + record.operation == FakeTargetOperation::DeleteObject + && record.key.as_deref() == Some(DELAYED_PURGE_KEY) + && record.version_id.as_deref() == Some(marker_version) + }) + .collect() +} + +async fn wait_for_target_marker_purged( + target: &FakeS3Target, + target_bucket: &str, + max_wait: Duration, +) -> Result<(), Box> { + let deadline = tokio::time::Instant::now() + max_wait; + loop { + let marker_present = target + .stored_versions(target_bucket, DELAYED_PURGE_KEY) + .iter() + .any(|(_, delete_marker)| *delete_marker); + if !marker_present { + return Ok(()); + } + if tokio::time::Instant::now() >= deadline { + return Err(format!( + "target delete marker was never purged; target state: {:?}", + target.stored_versions(target_bucket, DELAYED_PURGE_KEY) + ) + .into()); + } + sleep(Duration::from_millis(200)).await; + } +} + +/// P1-21: the delayed purge's single target DELETE currently swallows failures +/// (`let _ =`), so one transient target error strands the replicated marker on +/// the target forever. Contract under test: a failed purge attempt is retried +/// within the watch window and converges once the fault clears. +#[tokio::test] +#[serial] +async fn test_delayed_delete_marker_purge_retries_after_transient_target_failure() -> TestResult { + init_logging(); + let source_bucket = "delayed-purge-retry-src"; + let target_bucket = "delayed-purge-retry-dst"; + let (target, _source_env, source_client) = start_delayed_purge_fixture(source_bucket, target_bucket).await?; + + let marker_version = replicate_delete_marker(&target, target_bucket, &source_client, source_bucket).await?; + + // Four scripted failures. Fault budget accounting (each journal record + // consumes one fault, including the SDK's own per-request retries): + // deleting the marker version fans out over the version-purge replication + // channel (initial attempt + its fast in-memory MRF retries) plus the + // delayed purge watcher's single pre-fix attempt — three target DELETE calls + // in total today, empirically (see the exhaustion test's journal). Four + // faults outlast all of them, so only a delayed-purge retry in a later + // watch round can converge. If the SDK retry configuration ever changes, + // re-derive this budget from a fresh journal capture. + target.inject( + FakeTargetOperation::DeleteObject, + FakeTargetFault::Status(StatusCode::SERVICE_UNAVAILABLE), + 4, + ); + + // Purge the marker at the source. The watcher spawned when the marker + // replication completed moments ago observes the source marker vanish + // within its watch window and drives the target purge. + source_client + .delete_object() + .bucket(source_bucket) + .key(DELAYED_PURGE_KEY) + .version_id(&marker_version) + .send() + .await?; + + // Tight window on purpose: a fixed delayed purge retries on 1s rounds and + // converges within ~5s, while any straggling backoff retry from the other + // channels would land later and must not be what turns this test green. + wait_for_target_marker_purged(&target, target_bucket, Duration::from_secs(15)).await?; + + let attempts = delayed_purge_attempts(&target, &marker_version); + assert!( + attempts.len() >= 2, + "expected the faulted purge attempt plus at least one retry, got: {attempts:?}" + ); + assert!( + attempts.iter().any(|record| record.fault.is_none()), + "expected a clean purge attempt after the fault script drained, got: {attempts:?}" + ); + + target.shutdown().await; + Ok(()) +} + +/// P1-21 review follow-up: the watcher must purge the version the TARGET +/// assigned to the replicated marker, not one derived from the source uuid. +/// A target that mints its own version ids answers a source-derived purge +/// with an idempotent 204, which used to look like success and strand the +/// real marker on the target forever. +#[tokio::test] +#[serial] +async fn test_delayed_delete_marker_purge_uses_target_assigned_version() -> TestResult { + init_logging(); + let source_bucket = "delayed-purge-mint-src"; + let target_bucket = "delayed-purge-mint-dst"; + let (target, _source_env, source_client) = start_delayed_purge_fixture(source_bucket, target_bucket).await?; + // The target ignores the forwarded source-version-id header and mints its + // own ids for both the object and the replicated delete marker. + target.assign_own_version_ids(true); + + let marker_version = replicate_delete_marker(&target, target_bucket, &source_client, source_bucket).await?; + + source_client + .delete_object() + .bucket(source_bucket) + .key(DELAYED_PURGE_KEY) + .version_id(&marker_version) + .send() + .await?; + + // The replicated marker carries a target-minted version id, so nothing + // but the recorded mapping can address it. + wait_for_target_marker_purged(&target, target_bucket, Duration::from_secs(25)).await?; + + target.shutdown().await; + Ok(()) +} + +/// P1-21: when every watch-window purge attempt fails, the purge intent must +/// survive as a durable MRF entry and replay on the next startup; once the +/// replayed purge succeeds, the entry must be acknowledged instead of being +/// retained as Missed forever. +#[tokio::test] +#[serial] +async fn test_delayed_delete_marker_purge_exhaustion_persists_to_mrf_and_replays_on_restart() -> TestResult { + init_logging(); + let source_bucket = "delayed-purge-mrf-src"; + let target_bucket = "delayed-purge-mrf-dst"; + let (target, mut source_env, source_client) = start_delayed_purge_fixture(source_bucket, target_bucket).await?; + + let marker_version = replicate_delete_marker(&target, target_bucket, &source_client, source_bucket).await?; + + // Outlast the whole watch window: every in-process purge attempt fails. + target.inject( + FakeTargetOperation::DeleteObject, + FakeTargetFault::Status(StatusCode::SERVICE_UNAVAILABLE), + 64, + ); + + source_client + .delete_object() + .bucket(source_bucket) + .key(DELAYED_PURGE_KEY) + .version_id(&marker_version) + .send() + .await?; + + // Let the watch window drain before restarting. The wall-clock length is + // not 5x1s: every faulted attempt embeds the SDK's own per-request 503 + // retries (a few seconds each), so instead of a fixed sleep, wait until + // the faulted attempts stop arriving (the watcher exhausted its rounds and + // persisted the purge intent), then give the MRF persister its 100ms + // flush interval. + let deadline = tokio::time::Instant::now() + Duration::from_secs(60); + let mut last_seen = delayed_purge_attempts(&target, &marker_version).len(); + let mut quiet_since = tokio::time::Instant::now(); + loop { + sleep(Duration::from_millis(500)).await; + let seen = delayed_purge_attempts(&target, &marker_version).len(); + if seen != last_seen { + last_seen = seen; + quiet_since = tokio::time::Instant::now(); + } + if last_seen > 0 && quiet_since.elapsed() >= Duration::from_secs(5) { + break; + } + if tokio::time::Instant::now() >= deadline { + return Err(format!("purge attempts never quiesced (saw {last_seen}); journal: {:?}", target.requests()).into()); + } + } + sleep(Duration::from_secs(1)).await; + + let marker_survives_faults = target + .stored_versions(target_bucket, DELAYED_PURGE_KEY) + .iter() + .any(|(_, delete_marker)| *delete_marker); + assert!(marker_survives_faults, "scripted faults must have blocked every in-process purge attempt"); + + target.clear_faults(); + let attempts_before_restart = delayed_purge_attempts(&target, &marker_version).len(); + + // Startup MRF replay must re-drive the purge and clean the target. + source_env + .restart_server_preserving_data(vec![], &delayed_purge_process_env()) + .await?; + wait_for_target_marker_purged(&target, target_bucket, Duration::from_secs(30)).await?; + let attempts_after_replay = delayed_purge_attempts(&target, &marker_version).len(); + assert!( + attempts_after_replay > attempts_before_restart, + "the restart replay must have issued the purge DELETE" + ); + + // The successful replay must acknowledge the MRF entry: another restart may + // not re-drive the purge again. + source_env + .restart_server_preserving_data(vec![], &delayed_purge_process_env()) + .await?; + sleep(Duration::from_secs(5)).await; + assert_eq!( + delayed_purge_attempts(&target, &marker_version).len(), + attempts_after_replay, + "acknowledged purge-intent MRF entries must not replay again" + ); + + target.shutdown().await; + Ok(()) +} diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index 07bbb1515..76d807d75 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -2567,6 +2567,12 @@ pub trait ReplicationPoolTrait: std::fmt::Debug { async fn queue_replica_task(&self, ri: ReplicateObjectInfo) -> ReplicationQueueAdmission; async fn queue_replica_delete_task(&self, ri: DeletedObjectReplicationInfo) -> ReplicationQueueAdmission; async fn queue_replica_delete_batch(&self, deletes: &[DeletedObjectReplicationInfo]) -> ReplicationBatchAdmission; + /// Persist one entry straight to the durable MRF journal, bypassing the + /// live worker queues. For failures whose source state is already gone — + /// e.g. exhausted delete-marker purges — where only a startup replay can + /// retry, and live re-dispatch would loop unboundedly against a down + /// target. + async fn persist_mrf_entry(&self, entry: MrfReplicateEntry) -> ReplicationQueueAdmission; async fn resize(&self, priority: ReplicationPriority, max_workers: usize, max_l_workers: usize); async fn get_bucket_resync_status(&self, bucket: &str) -> Result; async fn cancel_bucket_resync(&self, opts: ResyncOpts) -> Result<(), EcstoreError>; @@ -2607,6 +2613,10 @@ impl ReplicationPoolTrait for ReplicationPool { self.queue_replica_delete_batch(deletes).await } + async fn persist_mrf_entry(&self, entry: MrfReplicateEntry) -> ReplicationQueueAdmission { + self.queue_mrf_save_admission(entry, "delete_marker_purge").await + } + async fn resize(&self, priority: ReplicationPriority, max_workers: usize, max_l_workers: usize) { self.resize(priority, max_workers, max_l_workers).await; } diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 9350f9c0d..a7a2b6a7a 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -18,9 +18,10 @@ use super::replication_config_store::ReplicationConfigStore; use super::replication_error_boundary::{Result, is_err_object_not_found, is_err_version_not_found}; use super::replication_event_sink::{EventArgs, send_event, send_local_event}; use super::replication_filemeta_boundary::{ - NULL_VERSION_ID, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, 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, + MrfReplicateEntry, NULL_VERSION_ID, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, ReplicateDecision, ReplicateObjectInfo, + ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction, ReplicationState, ReplicationStatusType, ReplicationType, + ReplicationWorkerOperation, VersionPurgeStatusType, get_replication_state, parse_replicate_decision, + 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}; @@ -33,7 +34,7 @@ use super::replication_object_decision_boundary::{ is_retryable_delete_replication_head_error, is_version_delete_replication, replication_etags_match, replication_multipart_complete_actual_size, replication_multipart_part_plan, should_retry_delete_marker_purge, }; -use super::replication_queue_boundary::DeletedObjectReplicationInfo; +use super::replication_queue_boundary::{DeletedObjectReplicationInfo, ReplicationQueueAdmission}; use super::replication_resync_boundary::ResyncStatusType; use super::replication_resync_boundary::{ BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus, encode_resync_file, is_version_id_mismatch, @@ -63,6 +64,7 @@ use futures::stream::StreamExt; use http::HeaderMap; use http_body::Frame; use http_body_util::StreamBody; +use metrics::counter; #[cfg(test)] use rmp_serde; use rustfs_s3_types::EventName; @@ -100,6 +102,9 @@ const EVENT_REPLICATION_FORCE_DELETE_SKIPPED: &str = "replication_force_delete_s const EVENT_RESYNC_TASK_FAILED: &str = "replication_resync_task_failed"; const EVENT_RESYNC_TARGET_OPERATION_FAILED: &str = "replication_resync_target_operation_failed"; const EVENT_RESYNC_RUNTIME_CHANNEL_FAILED: &str = "replication_resync_runtime_channel_failed"; +const EVENT_DELETE_MARKER_PURGE_FAILED: &str = "replication_delete_marker_purge_failed"; +const EVENT_DELETE_MARKER_PURGE_MRF: &str = "replication_delete_marker_purge_mrf"; +const METRIC_DELETE_MARKER_PURGE_TOTAL: &str = "rustfs_replication_delete_marker_purge_total"; const REPLICATION_TARGET_OFFLINE_ERROR_MARKERS: &[&str] = &[ "dispatch failure", "timeouterror", @@ -1271,7 +1276,12 @@ pub(crate) async fn replicate_delete_with_outcome( reason = "source_version_missing", "Skipping stale delete-marker replication" ); - return true; + // The marker is gone at the source, but a replica of it may + // already exist on the targets (a live race, or an MRF + // purge-intent replay landing here on purpose). Purge instead + // of just skipping; the result decides whether an MRF replay + // may acknowledge the entry. + return purge_stale_delete_marker_targets(&bucket, &dobj).await; } Err(err) => { source_state_verified = false; @@ -1485,29 +1495,6 @@ pub(crate) async fn replicate_delete_with_outcome( let is_version_purge = is_version_delete_replication(&dobj.delete_object); let requires_delayed_purge = should_retry_delete_marker_purge(&dobj.delete_object); - if requires_delayed_purge { - let bucket_clone = bucket.clone(); - let dobj_clone = dobj.clone(); - let dsc_clone = dsc.clone(); - let storage_clone = storage.clone(); - tokio::spawn(async move { - for _ in 0..5 { - if let Some(delete_marker_version_id) = dobj_clone.delete_object.delete_marker_version_id - && source_delete_marker_missing( - &*storage_clone, - &bucket_clone, - &dobj_clone.delete_object.object_name, - delete_marker_version_id, - ) - .await - { - replicate_delete_marker_purge_to_targets(&bucket_clone, &dobj_clone, &dsc_clone).await; - break; - } - tokio::time::sleep(TokioDuration::from_secs(1)).await; - } - }); - } let (replication_status, prev_status) = if !is_version_purge { ( @@ -1550,6 +1537,24 @@ pub(crate) async fn replicate_delete_with_outcome( drs.replication_timestamp = Some(OffsetDateTime::now_utc()); } + if requires_delayed_purge { + // Hand the watcher the MERGED replication state: `drs` folds this + // round's per-target results into the previous state, including the + // version ids the targets assigned to the markers they just created. + // Spawning with the pre-merge `dobj` made the purge fall back to a + // source-derived id, which a target that mints its own ids answers + // with an idempotent 204 — the intent was then dropped while the + // real marker stayed behind. + let bucket_clone = bucket.clone(); + let mut dobj_clone = dobj.clone(); + dobj_clone.delete_object.replication_state = Some(drs.clone()); + let dsc_clone = dsc.clone(); + let storage_clone = storage.clone(); + tokio::spawn(async move { + watch_and_purge_source_delete_marker(bucket_clone, dobj_clone, dsc_clone, storage_clone).await; + }); + } + let event_name = if replication_status == ReplicationStatusType::Completed { EventName::ObjectReplicationComplete.to_string() } else { @@ -1608,12 +1613,36 @@ pub(crate) async fn replicate_delete_with_outcome( } }; + replicate_delete_outcome( + expected_targets, + rinfos.targets.len(), + state_persisted, + source_state_verified, + &replication_status, + ) +} + +/// Whether a delete replication fully succeeded — the MRF replay acknowledges +/// (drops) an entry exactly when this returns true. +/// +/// The delayed purge is deliberately NOT an input: holding the outcome hostage +/// to it (`&& !requires_delayed_purge`) forced `false` for every delete-marker +/// entry and retained them all in the durable MRF journal forever. Purge +/// failures persist their own purge-intent entry instead +/// (`watch_and_purge_source_delete_marker`), and replays of those entries +/// report purge success through `purge_stale_delete_marker_targets`. +fn replicate_delete_outcome( + expected_targets: usize, + replicated_targets: usize, + state_persisted: bool, + source_state_verified: bool, + replication_status: &ReplicationStatusType, +) -> bool { expected_targets > 0 - && rinfos.targets.len() == expected_targets + && replicated_targets == expected_targets && state_persisted && source_state_verified - && !requires_delayed_purge - && replication_status == ReplicationStatusType::Completed + && *replication_status == ReplicationStatusType::Completed } async fn source_delete_marker_missing( @@ -1663,48 +1692,286 @@ fn delete_marker_purge_version_id( }) } -async fn replicate_delete_marker_purge_to_targets(bucket: &str, dobj: &DeletedObjectReplicationInfo, dsc: &ReplicateDecision) { +/// One purge pass over the eligible targets. Returns the ARNs that must be +/// retried: the remote DELETE failed, or the target client was unavailable +/// (e.g. a runtime cache miss). Inconsistent recorded version mappings are a +/// deliberate refusal — retrying cannot make guessing a version id safe — so +/// they are logged and excluded from the retry set. +async fn replicate_delete_marker_purge_to_targets( + bucket: &str, + dobj: &DeletedObjectReplicationInfo, + dsc: &ReplicateDecision, + retry_arns: Option<&[String]>, +) -> Vec { let Some(delete_marker_version_id) = dobj.delete_object.delete_marker_version_id else { - return; + return Vec::new(); }; + let target_arns = dobj.admitted_target_arns(); + let mut failed_arns = Vec::new(); for tgt_entry in dsc.targets_map.values() { if !tgt_entry.replicate { continue; } - let target_arns = dobj.admitted_target_arns(); if !target_arns.is_empty() && !target_arns.iter().any(|arn| arn == &tgt_entry.arn) { continue; } - let Some(tgt_client) = ReplicationTargetStore::remote_target_client(bucket, &tgt_entry.arn).await else { + if let Some(retry_arns) = retry_arns + && !retry_arns.iter().any(|arn| arn == &tgt_entry.arn) + { continue; - }; - + } + // Decide the version first: refusing to guess is a per-target + // FAILURE, not a silent skip. Reporting it as success would let the + // watcher and the MRF replay drop the purge intent while the marker + // is still on the target — the leak stays visible instead (the + // entry is retained and keeps warning) until an operator repairs + // the metadata. let Some(purge_version_id) = delete_marker_purge_version_id( dobj.delete_object.replication_state.as_ref(), &tgt_entry.arn, delete_marker_version_id, ) else { warn!( + event = EVENT_DELETE_MARKER_PURGE_FAILED, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, bucket, object = dobj.delete_object.object_name, arn = tgt_entry.arn, - "Skipping delete-marker purge: recorded target version metadata is inconsistent" + reason = "recorded_target_version_inconsistent", + "Delete-marker purge refused: recorded target version metadata is inconsistent" ); + counter!(METRIC_DELETE_MARKER_PURGE_TOTAL, "state" => "refused").increment(1); + failed_arns.push(tgt_entry.arn.clone()); continue; }; - let _ = tgt_client + let Some(tgt_client) = ReplicationTargetStore::remote_target_client(bucket, &tgt_entry.arn).await else { + warn!( + event = EVENT_DELETE_MARKER_PURGE_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket, + object = dobj.delete_object.object_name, + arn = tgt_entry.arn, + reason = "target_client_missing", + "Delete-marker purge attempt failed" + ); + counter!(METRIC_DELETE_MARKER_PURGE_TOTAL, "state" => "failed").increment(1); + failed_arns.push(tgt_entry.arn.clone()); + continue; + }; + + match tgt_client .remove_object( &tgt_client.bucket, &dobj.delete_object.object_name, purge_version_id, replication_delete_marker_purge_remove_options(dobj.delete_object.delete_marker_mtime), ) - .await; + .await + { + Ok(_) => { + counter!(METRIC_DELETE_MARKER_PURGE_TOTAL, "state" => "purged").increment(1); + } + // The marker version is already gone on the target: the purge goal + // is met. Strict S3 targets 404 here (RustFS/MinIO answer 204); + // treating it as a failure would retain the intent entry forever. + Err(error) if matches!(error.code.as_deref(), Some("NoSuchKey" | "NoSuchVersion")) => { + counter!(METRIC_DELETE_MARKER_PURGE_TOTAL, "state" => "purged").increment(1); + } + Err(error) => { + warn!( + event = EVENT_DELETE_MARKER_PURGE_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket, + object = dobj.delete_object.object_name, + arn = tgt_entry.arn, + error = %error, + reason = "target_delete_failed", + "Delete-marker purge attempt failed" + ); + counter!(METRIC_DELETE_MARKER_PURGE_TOTAL, "state" => "failed").increment(1); + mark_replication_target_offline_if_needed(&tgt_client, &error).await; + failed_arns.push(tgt_entry.arn.clone()); + } + } } + failed_arns +} + +const DELETE_MARKER_PURGE_WATCH_ROUNDS: usize = 5; +const DELETE_MARKER_PURGE_WATCH_INTERVAL: TokioDuration = TokioDuration::from_secs(1); + +/// Watch the source delete marker for a short window after its replication. +/// +/// KNOWN NON-DURABLE WINDOW: this task is detached, so a process exit inside +/// the watch window loses an intent that has not been persisted yet. The +/// window predates this code (the previous implementation had no durable +/// channel at all, and no replay half either), so nothing regresses — closing +/// it needs a write-ahead intent recorded before the parent delete is +/// acknowledged, which is tracked as follow-up rather than done here: every +/// delete-marker replication would pay a journal write for a purge that +/// almost never happens. +/// +/// If the marker disappears (deleted before or while the replica landed), +/// purge the replicated marker from the targets, retrying failed targets on +/// later rounds. When the window drains with targets still dirty, persist the +/// purge intent as a durable MRF entry so the next startup replays it through +/// `purge_stale_delete_marker_targets`. +async fn watch_and_purge_source_delete_marker( + bucket: String, + dobj: DeletedObjectReplicationInfo, + dsc: ReplicateDecision, + storage: Arc, +) { + let Some(delete_marker_version_id) = dobj.delete_object.delete_marker_version_id else { + return; + }; + + // `pending` is None until the source marker is observed missing; after the + // first purge pass it holds the targets that still need a successful purge. + let mut pending: Option> = None; + for round in 0..DELETE_MARKER_PURGE_WATCH_ROUNDS { + pending = match pending.take() { + None => { + if source_delete_marker_missing(&*storage, &bucket, &dobj.delete_object.object_name, delete_marker_version_id) + .await + { + Some(replicate_delete_marker_purge_to_targets(&bucket, &dobj, &dsc, None).await) + } else { + None + } + } + Some(failed_arns) => Some(replicate_delete_marker_purge_to_targets(&bucket, &dobj, &dsc, Some(&failed_arns)).await), + }; + if matches!(pending.as_deref(), Some([])) { + return; + } + if round + 1 < DELETE_MARKER_PURGE_WATCH_ROUNDS { + tokio::time::sleep(DELETE_MARKER_PURGE_WATCH_INTERVAL).await; + } + } + if let Some(failed_arns) = pending.filter(|failed_arns| !failed_arns.is_empty()) { + enqueue_delete_marker_purge_mrf(&dobj, failed_arns).await; + } +} + +/// Shape an exhausted purge intent as a marker-creation delete entry. Replay +/// reconstructs it with `delete_marker: true`, finds the source marker gone, +/// and funnels into the stale-marker branch of `replicate_delete_with_outcome` +/// — which re-runs the purge without touching source state and reports purge +/// success as the replay outcome. +fn delete_marker_purge_mrf_entry(dobj: &DeletedObjectReplicationInfo, failed_arns: Vec) -> MrfReplicateEntry { + let mut entry = dobj.to_mrf_entry(); + entry.delete_marker = true; + entry.version_id = None; + entry.retry_count = 0; + entry.target_arns = failed_arns; + entry +} + +async fn enqueue_delete_marker_purge_mrf(dobj: &DeletedObjectReplicationInfo, failed_arns: Vec) { + let arns = failed_arns.join(","); + let miss_reason = match runtime_sources::replication_pool() { + None => Some("replication_pool_unavailable"), + Some(pool) => match pool.persist_mrf_entry(delete_marker_purge_mrf_entry(dobj, failed_arns)).await { + ReplicationQueueAdmission::Queued => None, + _ => Some("mrf_save_unavailable"), + }, + }; + match miss_reason { + None => { + warn!( + event = EVENT_DELETE_MARKER_PURGE_MRF, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = dobj.bucket, + object = dobj.delete_object.object_name, + arns, + state = "queued", + "Delete-marker purge exhausted its watch window; intent persisted to the MRF journal" + ); + counter!(METRIC_DELETE_MARKER_PURGE_TOTAL, "state" => "mrf_queued").increment(1); + } + Some(reason) => { + warn!( + event = EVENT_DELETE_MARKER_PURGE_MRF, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = dobj.bucket, + object = dobj.delete_object.object_name, + arns, + state = "missed", + reason, + "Delete-marker purge intent could not be persisted for retry" + ); + counter!(METRIC_DELETE_MARKER_PURGE_TOTAL, "state" => "mrf_missed").increment(1); + } + } +} + +/// The marker vanished at the source while its replication was still pending +/// (a live race), or this is an MRF purge-intent replay. Any marker already +/// replicated to a target must still be purged; run bounded retry passes and +/// report the result so an MRF replay only acknowledges the entry once every +/// target is clean. Live callers persist a fresh purge intent on failure; +/// replay callers (`ReplicationType::Heal`) rely on Missed retention instead, +/// so the journal does not accumulate duplicate entries. +/// +/// Heal callers retry for the full watch window because the startup MRF +/// processor runs before bucket metadata (and thus target clients) finishes +/// initializing — the first pass can see `target_client_missing` and a later +/// round resolves the client; the replay loop is serial and startup-only, so +/// blocking it for up to the window per dirty entry is acceptable. Live +/// callers run on replication workers where a down target would pin a worker +/// for the whole window, so they attempt once and lean on the durable intent +/// entry instead. +async fn purge_stale_delete_marker_targets(bucket: &str, dobj: &DeletedObjectReplicationInfo) -> bool { + let decision_str = dobj + .delete_object + .replication_state + .as_ref() + .map(|state| state.replicate_decision_str.clone()) + .unwrap_or_default(); + let dsc = match parse_replicate_decision(bucket, &decision_str) { + Ok(dsc) => dsc, + Err(error) => { + warn!( + event = EVENT_DELETE_MARKER_PURGE_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket, + object = dobj.delete_object.object_name, + error = %error, + reason = "replicate_decision_parse_failed", + "Delete-marker purge attempt failed" + ); + return false; + } + }; + let rounds = if dobj.op_type == ReplicationType::Heal { + DELETE_MARKER_PURGE_WATCH_ROUNDS + } else { + 1 + }; + let mut failed_arns = replicate_delete_marker_purge_to_targets(bucket, dobj, &dsc, None).await; + for _ in 1..rounds { + if failed_arns.is_empty() { + break; + } + tokio::time::sleep(DELETE_MARKER_PURGE_WATCH_INTERVAL).await; + failed_arns = replicate_delete_marker_purge_to_targets(bucket, dobj, &dsc, Some(&failed_arns)).await; + } + if failed_arns.is_empty() { + return true; + } + if dobj.op_type != ReplicationType::Heal { + enqueue_delete_marker_purge_mrf(dobj, failed_arns).await; + } + false } async fn replicate_force_delete_to_targets(dobj: &DeletedObjectReplicationInfo, storage: Arc) -> bool { @@ -3218,6 +3485,7 @@ async fn replicate_object_with_multipart(ctx: MultipartR #[cfg(test)] mod tests { + use super::super::replication_filemeta_boundary::ReplicateTargetDecision; use super::super::replication_target_boundary::{BucketTarget, BucketTargets}; use super::*; use s3s::dto::{ @@ -3581,6 +3849,101 @@ mod tests { ); } + /// P1-21 regression guard for the outcome formula. A fully successful + /// delete-marker replication must acknowledge its MRF entry: the formula + /// once carried `&& !requires_delayed_purge`, which pinned every + /// delete-marker entry to Missed and retained the whole backlog forever. + /// (Deterministically staging a marker-creation entry in the durable + /// journal from e2e would require saturating the worker queues, so the + /// formula is pinned here instead; the purge-intent replay half is pinned + /// by the delayed-purge e2e pair.) + #[test] + fn test_replicate_delete_outcome_is_not_held_hostage_by_the_delayed_purge() { + assert!( + replicate_delete_outcome(1, 1, true, true, &ReplicationStatusType::Completed), + "a completed delete-marker replication must be acknowledgeable even though a delayed purge watch is pending" + ); + assert!(!replicate_delete_outcome(0, 0, true, true, &ReplicationStatusType::Completed)); + assert!(!replicate_delete_outcome(2, 1, true, true, &ReplicationStatusType::Completed)); + assert!(!replicate_delete_outcome(1, 1, false, true, &ReplicationStatusType::Completed)); + assert!(!replicate_delete_outcome(1, 1, true, false, &ReplicationStatusType::Completed)); + assert!(!replicate_delete_outcome(1, 1, true, true, &ReplicationStatusType::Failed)); + } + + /// P1-21 review follow-up: a target whose recorded marker version is + /// inconsistent must be reported as a per-target FAILURE. Treating the + /// refusal as success let the watcher and the MRF replay drop the purge + /// intent while the marker was still on the target. + #[tokio::test] + async fn test_delete_marker_purge_reports_corrupt_recorded_version_as_failure() { + let arn = format!("arn:rustfs:replication:us-east-1:corrupt:{}", Uuid::new_v4()); + let mut dsc = ReplicateDecision::new(); + dsc.set(ReplicateTargetDecision::new(arn.clone(), true, false)); + + let mut state = ReplicationState { + target_delete_marker_version_ids_corrupt: true, + ..Default::default() + }; + state.targets.insert(arn.clone(), ReplicationStatusType::Completed); + + let dobj = DeletedObjectReplicationInfo { + delete_object: ReplicationDeletedObject { + object_name: "doc.txt".to_string(), + delete_marker: true, + delete_marker_version_id: Some(Uuid::new_v4()), + replication_state: Some(state), + ..Default::default() + }, + bucket: "bucket-a".to_string(), + ..Default::default() + }; + + // No target client is registered: the refusal must be decided from + // the recorded metadata alone, before any remote call is attempted. + let failed = replicate_delete_marker_purge_to_targets("bucket-a", &dobj, &dsc, None).await; + + assert_eq!( + failed, + vec![arn], + "a refused purge must stay in the failed set so the intent is never acknowledged" + ); + } + + #[test] + fn test_delete_marker_purge_mrf_entry_replays_through_the_stale_marker_branch() { + let delete_marker_version_id = Uuid::new_v4(); + let dobj = DeletedObjectReplicationInfo { + delete_object: ReplicationDeletedObject { + object_name: "doc.txt".to_string(), + // A version-purge flavored source event: the entry must still + // be reshaped as a marker-creation delete so replay funnels + // into the stale-marker branch instead of re-running the full + // delete replication (whose source-state stamping would fail + // against the already-purged version). + delete_marker: false, + version_id: Some(Uuid::new_v4()), + delete_marker_version_id: Some(delete_marker_version_id), + ..Default::default() + }, + bucket: "bucket-a".to_string(), + ..Default::default() + }; + + let entry = delete_marker_purge_mrf_entry(&dobj, vec!["arn:a".to_string()]); + + assert!(entry.delete_marker, "purge intents must replay as marker-creation deletes"); + assert_eq!(entry.version_id, None, "the purged data version must not leak into the replay"); + assert_eq!(entry.delete_marker_version_id, Some(delete_marker_version_id)); + assert_eq!( + entry.target_arns, + vec!["arn:a".to_string()], + "only the targets whose purge failed may be retried" + ); + assert_eq!(entry.retry_count, 0); + assert_eq!(entry.bucket, "bucket-a"); + assert_eq!(entry.object, "doc.txt"); + } + #[test] fn test_is_retryable_delete_replication_head_error_allows_delete_marker_head_responses() { assert!(