fix(replication): retry, persist and replay failed delete-marker purges (#5864)

* test(replication): pin delayed delete-marker purge failure handling (red)

P1-21 (rustfs/backlog#1675 B2): two failing e2e tests that pin the missing
failure handling of the delayed delete-marker purge:

- test_delayed_delete_marker_purge_retries_after_transient_target_failure:
  four scripted 503s outlast every existing channel (version-purge
  replication + its in-process MRF fast retries + the watcher's single
  attempt = 3 target DELETEs, all faulted in the recorded run); the
  replicated marker is stranded on the target forever.
- test_delayed_delete_marker_purge_exhaustion_persists_to_mrf_and_replays_on_restart:
  exhausted purge intents never reach the durable MRF journal, so a restart
  replays nothing (recorded run: 3 faulted attempts, zero post-restart).

Red-light evidence (current main):
- Test A: FAILED, journal shows 3x DeleteObject fault=Status(503), no clean
  attempt, target marker still present after 15s.
- Test B: FAILED after 468s, same 3 faulted attempts, no purge DELETE after
  restart, marker still present.

Test infra: FakeS3Target::stored_versions() exposes per-key version state so
purge tests assert target state instead of inferring it from the journal;
nextest count comments 36->38 nightly / 56->58 total.

* fix(replication): retry, persist and replay failed delete-marker purges

P1-21 (rustfs/backlog#1675 B2). The delayed delete-marker purge was
fire-and-forget: the target DELETE discarded its result (`let _ =`), a
missing target client was silently skipped, and nothing recorded the intent
— one transient target error stranded the replicated marker on the target
forever. Separately, `replicate_delete_with_outcome` held its outcome
hostage to `!requires_delayed_purge`, pinning every delete-marker MRF entry
to Missed so the durable backlog retained them permanently.

Changes:
- `replicate_delete_marker_purge_to_targets` now reports per-target
  results (warn + metrics on failure, including `target_client_missing`),
  supports retrying only the failed targets, and treats a target-side
  NoSuchKey/NoSuchVersion as purge success (strict-404 targets must not
  retain the intent forever).
- The delayed watcher (`watch_and_purge_source_delete_marker`) retries
  failed targets across its 5x1s watch window; on exhaustion it persists
  the purge intent to the durable MRF journal via the new
  `ReplicationPoolTrait::persist_mrf_entry` (journal-only on purpose: live
  re-dispatch would loop unboundedly against a down target). Intent entries
  are shaped as marker-creation deletes so replay funnels into the stale-
  marker branch.
- The stale-marker branch (source marker already gone) now purges the
  targets instead of silently returning success — closing a latent leak —
  and reports the purge result as the replay outcome. Heal callers retry
  for the full window (the startup MRF processor runs before target
  clients initialize); live callers attempt once and fall back to a fresh
  durable intent, so a down target cannot pin a replication worker.
- The outcome formula (extracted as `replicate_delete_outcome` and pinned
  by a unit test) no longer includes the delayed purge, so successfully
  replayed delete-marker entries are acknowledged instead of retained
  forever.

Verification: red -> green e2e pair (transient-failure retry; exhaustion ->
durable MRF -> restart replay -> second-restart zero-replay ack) plus unit
tests; `make pre-commit`, logging guardrails, clippy (ecstore + e2e_test)
all clean; full ecstore lib suite 3729 passed (3 pre-existing local-DNS
kubernetes endpoint failures reproduce without this change).

Adversarial validation (7 roles): no blocking findings after adding the
outcome-formula guard test. Known residuals recorded in the PR: watcher
shutdown window (intent not yet persisted), rolling-downgrade replay acks
without purging (equals pre-fix behavior), and replay falling back to the
source version id on targets that mint their own version ids (P1-19).

* chore(test): refresh the nextest replication count invariant

The e2e-smoke/e2e-repl-nightly split comment is descriptive metadata
(authority: `cargo nextest list`); refresh it to this branch's
post-rebase total.

* fix(replication): purge the marker version the target actually assigned

Review follow-up (#5864), two real defects:

- The delayed purge watcher was spawned with the pre-merge `dobj`, so the
  per-target marker version ids this round recorded were invisible to it.
  Against a target that mints its own ids the purge fell back to a
  source-derived id, the target answered the versioned DELETE with an
  idempotent 204, and that "success" cleared the retry set while the real
  marker stayed behind. The watcher now receives the merged replication
  state (`drs`), which folds this round's target-assigned ids in.
- A target whose recorded version metadata is inconsistent was skipped
  without entering `failed_arns`, so an empty result made both the watcher
  and the MRF replay treat a purge that issued no DELETE as successful and
  drop the intent. The refusal is now a per-target failure (own metric
  label): the leak stays visible and the intent is retained instead of
  being acknowledged. The version decision also moved ahead of the client
  lookup, so the refusal is decided from metadata alone.

Tests: a new e2e drives a fake target with `assign_own_version_ids`, which
ignores the forwarded source-version header for both objects and delete
markers, and asserts the replicated marker is really gone; a unit test
pins the corrupt-metadata refusal as a failed outcome without any target
client registered. The detached-watcher shutdown window is documented at
the watcher as a known non-durable window with the write-ahead follow-up
spelled out.

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<String, BucketState>,
     uploads: HashMap<String, MultipartState>,
     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<String>
     }
 }
 
-fn new_version_id(headers: &HeaderMap) -> S3Result<String> {
+/// `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<String> {
     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(),