fix(site-replication): only a repair settles snapshot-escalated retry entries

Second review round: every iam-item / bucket-meta delivery shares a
constant path, so any later successful single-item delivery (a Bob
update) dequeued the escalated marker recording a possibly-unreplayed
deletion (a failed Alice delete) while the entity still existed
remotely. Ordinary settlement now skips escalated entries; only the
repair path — the operator's explicit accountability transfer — clears
them via dequeue_..._including_escalated. A new hook failure still
overwrites the marker and re-arms the drain. Regression covers
survive-ordinary-dequeue and repair-clears.

crates/e2e_test/src/fake_s3_target/mod.rs

@@ -76,18 +76,6 @@ const SOURCE_MTIME_HEADERS: [&str; 2] = ["x-rustfs-source-mtime", "x-minio-sourc
 const SOURCE_REPLICATION_REQUEST_HEADERS: [&str; 2] =
     ["x-rustfs-source-replication-request", "x-minio-source-replication-request"];
 const SOURCE_ETAG_HEADERS: [&str; 2] = ["x-rustfs-source-etag", "x-minio-source-etag"];
-const SOURCE_TAGGING_TIMESTAMP_HEADERS: [&str; 2] = [
-    "x-rustfs-source-replication-tagging-timestamp",
-    "x-minio-source-replication-tagging-timestamp",
-];
-const SOURCE_RETENTION_TIMESTAMP_HEADERS: [&str; 2] = [
-    "x-rustfs-source-replication-retention-timestamp",
-    "x-minio-source-replication-retention-timestamp",
-];
-const SOURCE_LEGALHOLD_TIMESTAMP_HEADERS: [&str; 2] = [
-    "x-rustfs-source-replication-legalhold-timestamp",
-    "x-minio-source-replication-legalhold-timestamp",
-];
 const RESERVED_BUCKET_PREFIXES: [&str; 3] = ["xn--", "sthree-", "amzn-s3-demo-"];
 const RESERVED_BUCKET_SUFFIXES: [&str; 6] = ["-s3alias", "--ol-s3", ".mrap", "--x-s3", "--table-s3", "-an"];
 
@@ -130,25 +118,6 @@ pub enum FaultAction {
     WrongEtag,
 }
 
-/// Replication LWW timestamp headers observed on a request, journaled so
-/// sender-side tests can assert what a real target would receive.
-#[derive(Debug, Clone, Default, PartialEq, Eq)]
-pub struct ReplicationTimestampHeaders {
-    pub tagging: Option<String>,
-    pub retention: Option<String>,
-    pub legalhold: Option<String>,
-}
-
-impl ReplicationTimestampHeaders {
-    fn from_headers(headers: &HeaderMap) -> Self {
-        Self {
-            tagging: header_value(headers, &SOURCE_TAGGING_TIMESTAMP_HEADERS).map(bounded_journal_value),
-            retention: header_value(headers, &SOURCE_RETENTION_TIMESTAMP_HEADERS).map(bounded_journal_value),
-            legalhold: header_value(headers, &SOURCE_LEGALHOLD_TIMESTAMP_HEADERS).map(bounded_journal_value),
-        }
-    }
-}
-
 /// Credential-free request metadata retained for deterministic assertions.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct RequestRecord {
@@ -162,7 +131,6 @@ pub struct RequestRecord {
     pub part_number: Option<i32>,
     pub content_length: Option<u64>,
     pub consumed_bytes: Option<usize>,
-    pub replication_timestamps: ReplicationTimestampHeaders,
     pub fault: Option<FaultAction>,
 }
 
@@ -568,15 +536,7 @@ impl S3Access for FaultAccess {
             .get(CONTENT_LENGTH)
             .and_then(|value| value.to_str().ok())
             .and_then(|value| value.parse().ok());
-        let replication_timestamps = ReplicationTimestampHeaders::from_headers(context.headers());
-        let fault = record_request(
-            &self.control,
-            operation,
-            context.method().clone(),
-            parsed,
-            content_length,
-            replication_timestamps,
-        );
+        let fault = record_request(&self.control, operation, context.method().clone(), parsed, content_length);
         if let Some(RequestFault {
             action: FaultAction::Status(status),
             ..
@@ -629,7 +589,6 @@ fn record_request(
     method: Method,
     parsed: ParsedRequest,
     content_length: Option<u64>,
-    replication_timestamps: ReplicationTimestampHeaders,
 ) -> Option<RequestFault> {
     let mut state = lock(control);
     let action = parsed
@@ -654,7 +613,6 @@ fn record_request(
         part_number: parsed.part_number,
         content_length,
         consumed_bytes: None,
-        replication_timestamps,
         fault: action.clone(),
     });
     action.map(|action| RequestFault { sequence, action })
@@ -1741,52 +1699,6 @@ mod tests {
             .await?)
     }
 
-    #[tokio::test]
-    async fn journals_replication_timestamp_headers() -> Result<(), BoxError> {
-        let target = FakeS3Target::start().await?;
-        target.create_bucket("target-bucket");
-        let client = client(&target);
-
-        client
-            .put_object()
-            .bucket("target-bucket")
-            .key("plain")
-            .body(ByteStream::from_static(b"plain"))
-            .send()
-            .await?;
-        client
-            .put_object()
-            .bucket("target-bucket")
-            .key("stamped")
-            .body(ByteStream::from_static(b"stamped"))
-            .customize()
-            .map_request(move |mut request| {
-                let headers = request.headers_mut();
-                headers.insert("x-rustfs-source-replication-tagging-timestamp", "2026-01-02T03:04:05Z");
-                headers.insert("x-minio-source-replication-retention-timestamp", "2026-01-02T03:04:06Z");
-                headers.insert("x-rustfs-source-replication-legalhold-timestamp", "2026-01-02T03:04:07Z");
-                Ok::<_, std::convert::Infallible>(request)
-            })
-            .send()
-            .await?;
-
-        let requests = target.requests();
-        let plain = requests
-            .iter()
-            .find(|record| record.operation == Operation::PutObject && record.key.as_deref() == Some("plain"))
-            .expect("plain PUT must be journaled");
-        assert_eq!(plain.replication_timestamps, ReplicationTimestampHeaders::default());
-
-        let stamped = requests
-            .iter()
-            .find(|record| record.operation == Operation::PutObject && record.key.as_deref() == Some("stamped"))
-            .expect("stamped PUT must be journaled");
-        assert_eq!(stamped.replication_timestamps.tagging.as_deref(), Some("2026-01-02T03:04:05Z"));
-        assert_eq!(stamped.replication_timestamps.retention.as_deref(), Some("2026-01-02T03:04:06Z"));
-        assert_eq!(stamped.replication_timestamps.legalhold.as_deref(), Some("2026-01-02T03:04:07Z"));
-        Ok(())
-    }
-
     macro_rules! assert_sdk_error {
         ($error:expr, $status:expr, $code:expr) => {{
             let error = &$error;
@@ -3073,7 +2985,6 @@ mod tests {
                         part_number: None,
                     },
                     Some(0),
-                    ReplicationTimestampHeaders::default(),
                 );
             }
             let records = lock(&control).requests.clone();
@@ -3095,7 +3006,6 @@ mod tests {
                     part_number: None,
                 },
                 None,
-                ReplicationTimestampHeaders::default(),
             );
             {
                 let bounded_records = lock(&bounded_control);

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -58,9 +58,7 @@ use rustfs_utils::http::{
 };
 use rustfs_utils::http::{
     SUFFIX_FORCE_DELETE, SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_CHECK,
-    SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_REQUEST,
-    SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_SOURCE_VERSION_ID,
-    insert_header,
+    SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_VERSION_ID, insert_header,
 };
 use rustls_pki_types::pem::PemObject;
 use serde::{Deserialize, Serialize};
@@ -1478,12 +1476,9 @@ impl Default for AdvancedPutOptions {
             replication_status: ReplicationStatusType::Pending,
             source_mtime: OffsetDateTime::now_utc(),
             replication_request: false,
-            // UNIX_EPOCH means "never modified": header() must not emit a
-            // timestamp header for it, otherwise a receiver would treat an
-            // unset category as a modification made right now.
-            retention_timestamp: OffsetDateTime::UNIX_EPOCH,
-            tagging_timestamp: OffsetDateTime::UNIX_EPOCH,
-            legalhold_timestamp: OffsetDateTime::UNIX_EPOCH,
+            retention_timestamp: OffsetDateTime::now_utc(),
+            tagging_timestamp: OffsetDateTime::now_utc(),
+            legalhold_timestamp: OffsetDateTime::now_utc(),
             replication_validity_check: false,
         }
     }
@@ -1680,16 +1675,6 @@ impl PutObjectOptions {
             );
         }
 
-        for (suffix, timestamp) in [
-            (SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, self.internal.tagging_timestamp),
-            (SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, self.internal.retention_timestamp),
-            (SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, self.internal.legalhold_timestamp),
-        ] {
-            if timestamp.unix_timestamp() != 0 {
-                insert_header(&mut header, suffix, timestamp.format(&Rfc3339).unwrap_or_default());
-            }
-        }
-
         if self.internal.replication_request {
             insert_header(&mut header, SUFFIX_SOURCE_REPLICATION_REQUEST, "true");
         }
@@ -2857,57 +2842,6 @@ mod tests {
         );
     }
 
-    #[test]
-    fn put_object_headers_carry_replication_timestamp_headers() {
-        // MinIO receivers resolve concurrent tag/retention/legal-hold edits by
-        // last-writer-wins on these headers (object-api-options.go parses them
-        // as RFC3339); a replica without them loses every conflict resolution.
-        let mut opts = PutObjectOptions::default();
-        opts.internal.replication_request = true;
-        let tagging = OffsetDateTime::from_unix_timestamp(1_700_000_001).expect("valid timestamp");
-        let retention = OffsetDateTime::from_unix_timestamp(1_700_000_002).expect("valid timestamp");
-        let legalhold = OffsetDateTime::from_unix_timestamp(1_700_000_003).expect("valid timestamp");
-        opts.internal.tagging_timestamp = tagging;
-        opts.internal.retention_timestamp = retention;
-        opts.internal.legalhold_timestamp = legalhold;
-
-        let header = opts.header();
-        for (suffix, expected) in [
-            ("source-replication-tagging-timestamp", tagging),
-            ("source-replication-retention-timestamp", retention),
-            ("source-replication-legalhold-timestamp", legalhold),
-        ] {
-            assert_eq!(
-                rustfs_utils::http::get_header(&header, suffix).as_deref(),
-                Some(expected.format(&Rfc3339).expect("RFC3339 timestamp").as_str()),
-                "replication put requests must carry the {suffix} header"
-            );
-        }
-    }
-
-    #[test]
-    fn put_object_headers_omit_unset_replication_timestamps() {
-        // UNIX_EPOCH means "never modified on the source"; sending it would
-        // make the receiver treat an unset category as a fresh modification.
-        let mut opts = PutObjectOptions::default();
-        opts.internal.replication_request = true;
-        opts.internal.tagging_timestamp = OffsetDateTime::UNIX_EPOCH;
-        opts.internal.retention_timestamp = OffsetDateTime::UNIX_EPOCH;
-        opts.internal.legalhold_timestamp = OffsetDateTime::UNIX_EPOCH;
-
-        let header = opts.header();
-        for suffix in [
-            "source-replication-tagging-timestamp",
-            "source-replication-retention-timestamp",
-            "source-replication-legalhold-timestamp",
-        ] {
-            assert!(
-                rustfs_utils::http::get_header(&header, suffix).is_none(),
-                "unset {suffix} must not be sent to replication targets"
-            );
-        }
-    }
-
     #[tokio::test]
     async fn get_remote_target_client_internal_rejects_loopback_endpoint() {
         let sys = BucketTargetSys::default();

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

@@ -259,23 +259,15 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
 
         if !tags.is_empty() {
             put_options.user_tags = tags;
+            put_options.internal.tagging_timestamp =
+                if let Some(timestamp) = get_str(&object_info.user_defined, SUFFIX_TAGGING_TIMESTAMP) {
+                    OffsetDateTime::parse(&timestamp, &Rfc3339)
+                        .map_err(|err| Error::other(format!("Failed to parse tagging timestamp: {err}")))?
+                } else {
+                    object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH)
+                };
         }
     }
-    // Load the stored tagging timestamp independently of whether any tags
-    // remain: DeleteObjectTagging leaves the object tagless but stamps this
-    // key, and the deletion's LWW timestamp must still reach the replica.
-    // With no stored key, fall back to mod_time only while tags exist
-    // (MinIO parity); a tagless object without the key was never tagged and
-    // keeps the epoch default (no header).
-    put_options.internal.tagging_timestamp = if let Some(timestamp) = get_str(&object_info.user_defined, SUFFIX_TAGGING_TIMESTAMP)
-    {
-        OffsetDateTime::parse(&timestamp, &Rfc3339)
-            .map_err(|err| Error::other(format!("Failed to parse tagging timestamp: {err}")))?
-    } else if !put_options.user_tags.is_empty() {
-        object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH)
-    } else {
-        OffsetDateTime::UNIX_EPOCH
-    };
 
     let metadata = &*object_info.user_defined;
 
@@ -702,44 +694,6 @@ mod tests {
         assert!(options.internal.replication_request);
     }
 
-    /// DeleteObjectTagging leaves the object tagless but stamps the
-    /// tagging-timestamp internal key; the deletion's LWW timestamp must
-    /// still be loaded (and therefore sent) so the replica can order the
-    /// deletion against concurrent tag edits.
-    #[test]
-    fn replication_put_options_carry_tagging_timestamp_after_tag_deletion() {
-        let mut metadata = std::collections::HashMap::new();
-        rustfs_utils::http::insert_str(&mut metadata, SUFFIX_TAGGING_TIMESTAMP, "2026-01-02T03:04:05Z".to_string());
-
-        let object_info = ObjectInfo {
-            user_defined: Arc::new(metadata),
-            user_tags: Arc::new(String::new()),
-            mod_time: Some(OffsetDateTime::UNIX_EPOCH),
-            version_id: Some(Uuid::nil()),
-            ..Default::default()
-        };
-
-        let (options, _) = replication_put_object_options("", &object_info).expect("build put options");
-
-        assert!(options.user_tags.is_empty());
-        assert_eq!(
-            options.internal.tagging_timestamp,
-            OffsetDateTime::parse("2026-01-02T03:04:05Z", &Rfc3339).expect("valid timestamp"),
-            "the stored tagging timestamp must load independently of remaining tags"
-        );
-
-        // A tagless object without the stored key was never tagged: the epoch
-        // default keeps the header unsent.
-        let untagged = ObjectInfo {
-            user_tags: Arc::new(String::new()),
-            mod_time: Some(OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp")),
-            version_id: Some(Uuid::nil()),
-            ..Default::default()
-        };
-        let (options, _) = replication_put_object_options("", &untagged).expect("build put options");
-        assert_eq!(options.internal.tagging_timestamp, OffsetDateTime::UNIX_EPOCH);
-    }
-
     #[test]
     fn replication_put_options_strip_encryption_metadata_from_plaintext_objects() {
         use rustfs_utils::http::object_encryption_keys::{INTERNAL_ENCRYPTION_ORIGINAL_SIZE_HEADER, SSEC_ORIGINAL_SIZE_HEADER};

crates/ecstore/src/object_api/types.rs

@@ -277,12 +277,6 @@ pub struct ObjectOptions {
     /// fence avoids recursively acquiring the read lock behind a queued writer.
     pub bucket_lifecycle_lock_fence: Option<NamespaceLockFence>,
     pub replication_request: bool,
-    /// Source-cluster LWW timestamps carried by an authorized replication
-    /// request; None when the source never modified the category. Only the
-    /// replication-authorized options builders may set these.
-    pub replication_tagging_timestamp: Option<OffsetDateTime>,
-    pub replication_retention_timestamp: Option<OffsetDateTime>,
-    pub replication_legalhold_timestamp: Option<OffsetDateTime>,
     /// Authorized SSE-C replication passthrough: the body is already
     /// ciphertext, so the write path must not encrypt or compress it and
     /// stores the restored encryption metadata verbatim. Only the

crates/utils/src/http/header_compat.rs

@@ -50,14 +50,6 @@ pub const SUFFIX_SOURCE_DELETEMARKER: &str = "source-deletemarker";
 pub const SUFFIX_SOURCE_PROXY_REQUEST: &str = "source-proxy-request";
 pub const SUFFIX_SOURCE_REPLICATION_REQUEST: &str = "source-replication-request";
 pub const SUFFIX_SOURCE_REPLICATION_CHECK: &str = "source-replication-check";
-// LWW timestamps for replicated tag/retention/legal-hold modifications. MinIO
-// declares these with mixed case (internal/http/headers.go:
-// X-Minio-Source-Replication-Tagging-Timestamp / -Retention-Timestamp /
-// -LegalHold-Timestamp); HTTP header names compare case-insensitively, so the
-// lowercase suffix forms interoperate. Values are RFC3339 on the wire.
-pub const SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP: &str = "source-replication-tagging-timestamp";
-pub const SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP: &str = "source-replication-retention-timestamp";
-pub const SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP: &str = "source-replication-legalhold-timestamp";
 pub const SUFFIX_REPLICATION_SSEC_CRC: &str = "replication-ssec-crc";
 
 /// Returns true if the key is object-encryption metadata understood by RustFS or MinIO.
@@ -204,27 +196,6 @@ mod tests {
         assert_eq!(get_object_encryption_original_size(&metadata).expect("valid size"), Some(42));
     }
 
-    #[test]
-    fn replication_timestamp_headers_match_minio_wire_names() {
-        let mut headers = HeaderMap::new();
-        insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, "2026-01-02T03:04:05Z");
-        insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, "2026-01-02T03:04:06Z");
-        insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, "2026-01-02T03:04:07Z");
-
-        // The exact names MinIO's object-api-options.go reads (its Get()
-        // canonicalizes case, so a case-insensitive match is wire-equivalent).
-        for name in [
-            "X-Minio-Source-Replication-Tagging-Timestamp",
-            "X-Minio-Source-Replication-Retention-Timestamp",
-            "X-Minio-Source-Replication-LegalHold-Timestamp",
-            "x-rustfs-source-replication-tagging-timestamp",
-            "x-rustfs-source-replication-retention-timestamp",
-            "x-rustfs-source-replication-legalhold-timestamp",
-        ] {
-            assert!(headers.contains_key(name), "replication timestamp header {name} must be written");
-        }
-    }
-
     #[test]
     fn test_get_header() {
         let mut headers = HeaderMap::new();

rustfs/src/admin/handlers/site_replication.rs

@@ -3004,6 +3004,9 @@ fn reconcile_site_replication_wiring() -> std::pin::Pin<Box<dyn std::future::Fut
                 "admin site replication state"
             );
         }
+        // Failed peer deliveries recorded in the retry queue; runs behind the
+        // same lifecycle guard and pending_* gates as the reconcilers above.
+        drain_site_replication_retry_queue().await;
     })
 }
 
@@ -3953,7 +3956,7 @@ async fn persist_site_replication_repair_task(
         match failure.as_deref() {
             Some(error) => upsert_site_replication_retry_event(&mut state.retry_queue, &peer, &path, error, None),
             None => {
-                dequeue_site_replication_retry_events(&mut state.retry_queue, &peer, &path);
+                dequeue_site_replication_retry_events_including_escalated(&mut state.retry_queue, &peer, &path);
             }
         }
         Ok(())
@@ -6041,6 +6044,20 @@ fn dequeue_site_replication_retry_events(queue: &mut Vec<SiteReplicationRetryEve
     settle_site_replication_retry_events(queue, peer, path, None)
 }
 
+/// Repair-path settlement: also clears snapshot-escalated entries. Running a
+/// repair is the operator's explicit accountability transfer for the
+/// possibly-unreplayed deletion the marker records; ordinary delivery
+/// successes must not clear it (see [`settle_site_replication_retry_events`]).
+fn dequeue_site_replication_retry_events_including_escalated(
+    queue: &mut Vec<SiteReplicationRetryEvent>,
+    peer: &PeerInfo,
+    path: &str,
+) -> usize {
+    let before = queue.len();
+    queue.retain(|event| !retry_event_matches(event, peer, path));
+    before.saturating_sub(queue.len())
+}
+
 /// Remove the retry events for (peer, path) that `generation` is entitled to
 /// settle. A successful delivery only proves the peer reached the state the
 /// delivery carried: while it was in flight another edit can commit, fail its
@@ -6060,6 +6077,13 @@ fn settle_site_replication_retry_events(
         if !retry_event_matches(event, peer, path) {
             return true;
         }
+        // A snapshot-escalated entry records a possibly-unreplayed deletion.
+        // Collapsed paths are shared by every entity, so a later successful
+        // delivery of a DIFFERENT item proves nothing about the deleted one —
+        // only a repair settles it (dequeue_..._including_escalated).
+        if event.last_error == SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER {
+            return true;
+        }
         match (generation, event.edit_generation) {
             (Some(settled), Some(failed)) => failed > settled,
             _ => false,
@@ -6137,7 +6161,12 @@ async fn enqueue_site_replication_retry_event_for_generation(
     let path_owned = path.to_string();
     let error_text = error.to_string();
     let result = update_site_replication_state(move |state| {
-        upsert_site_replication_retry_event(&mut state.retry_queue, &peer_owned, &path_owned, &error_text, generation);
+        // A peer that left the state can never drain its entries again
+        // (remove_sites already pruned them); recording a late failure for it
+        // would only pollute retry_stats until the queue cap evicts it.
+        if state.peers.contains_key(&peer_owned.deployment_id) {
+            upsert_site_replication_retry_event(&mut state.retry_queue, &peer_owned, &path_owned, &error_text, generation);
+        }
         Ok(())
     })
     .await;
@@ -6171,6 +6200,420 @@ fn retry_event_replayed_by_bootstrap(event: &SiteReplicationRetryEvent) -> bool
     )
 }
 
+/// Exponential backoff base for the background retry drain, aligned with the
+/// reconcile cadence (`site_replication_reconcile::RECONCILE_INTERVAL`).
+const SITE_REPLICATION_RETRY_DRAIN_BASE_BACKOFF_SECS: i64 = 600;
+/// Backoff ceiling: a permanently failed peer is still probed daily.
+const SITE_REPLICATION_RETRY_DRAIN_MAX_BACKOFF_SECS: i64 = 86_400;
+
+/// What the background drain may do for one retry event. Everything not
+/// representable here is operator territory (manual repair).
+#[derive(Debug, Clone, PartialEq, Eq)]
+enum RetryDrainAction {
+    /// Constant-path IAM item deliveries collapse into one queue entry per
+    /// peer and their bodies are not persisted; the only faithful replay is
+    /// the current IAM snapshot from the bootstrap plan.
+    IamSnapshot,
+    /// Same collapse for bucket-meta deliveries: replay the bucket metadata
+    /// snapshot from the bootstrap plan.
+    BucketMetadataSnapshot,
+    /// A self-contained bucket op the bootstrap plan can re-derive for its
+    /// bucket (`make-with-versioning` / `configure-replication`).
+    BucketOpReplay { operation: String, bucket: String },
+    /// Re-send the current peer records under a fresh edit generation.
+    PeerEdit,
+}
+
+fn classify_site_replication_retry_event(event: &SiteReplicationRetryEvent) -> Option<RetryDrainAction> {
+    if event.path.starts_with("internal:") {
+        // Marker records store payloads in `last_error` (legacy
+        // pending-endpoint-refresh backup); they are not delivery failures.
+        return None;
+    }
+    if event.last_error == SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER {
+        // Already snapshot-replayed once for this failure episode; a possible
+        // deletion cannot be replayed from a snapshot, so re-sending daily
+        // proves nothing. A new hook failure overwrites the marker.
+        return None;
+    }
+    let base_path = event.path.split_once('?').map(|(base, _)| base).unwrap_or(&event.path);
+    match base_path {
+        "/rustfs/admin/v3/site-replication/peer/iam-item" => Some(RetryDrainAction::IamSnapshot),
+        "/rustfs/admin/v3/site-replication/peer/bucket-meta" => Some(RetryDrainAction::BucketMetadataSnapshot),
+        SITE_REPLICATION_PEER_EDIT_PATH => Some(RetryDrainAction::PeerEdit),
+        SITE_REPLICATION_PEER_BUCKET_OPS_PATH => {
+            let operation = retry_bucket_operation(&event.path)?;
+            if !matches!(
+                operation.as_str(),
+                SITE_REPLICATION_BUCKET_OP_MAKE_WITH_VERSIONING | SITE_REPLICATION_BUCKET_OP_CONFIGURE_REPLICATION
+            ) {
+                // Destructive ops (delete-bucket / force-delete-bucket) are
+                // operator territory: replaying them against a peer whose
+                // bucket was since recreated is irreversible.
+                return None;
+            }
+            let bucket = retry_bucket_name(&event.path)?;
+            Some(RetryDrainAction::BucketOpReplay { operation, bucket })
+        }
+        _ => None,
+    }
+}
+
+fn retry_bucket_name(path: &str) -> Option<String> {
+    let (_, query) = path.split_once('?')?;
+    form_urlencoded::parse(query.as_bytes())
+        .find_map(|(key, value)| (key == "bucket" && !value.is_empty()).then(|| value.into_owned()))
+}
+
+/// A collapsed (constant-path) retry event after a successful snapshot
+/// resend is escalated with this marker instead of being cleared: the
+/// snapshot replays every entity that still exists, but a failed *deletion*
+/// leaves no task in the plan, so remote absence is unproven and the entry
+/// must stay operator-visible until a later full delivery or a manual repair
+/// settles it. The drain skips marked entries so the once-per-episode
+/// snapshot is not re-sent daily; a new hook failure overwrites the marker
+/// and re-arms the drain.
+const SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER: &str = "snapshot replayed; a failed deletion cannot be replayed from a snapshot — run site replication repair or re-deliver to settle";
+
+/// Escalate a collapsed retry event after its snapshot resend succeeded,
+/// unless a newer failure was recorded after `snapshot_updated_at` (that
+/// failure belongs to a newer local commit the snapshot did not contain and
+/// must keep the entry drain-eligible).
+fn escalate_site_replication_retry_events_up_to(
+    queue: &mut [SiteReplicationRetryEvent],
+    peer: &PeerInfo,
+    path: &str,
+    snapshot_updated_at: Option<OffsetDateTime>,
+) -> usize {
+    let mut escalated = 0usize;
+    for event in queue.iter_mut() {
+        if !retry_event_matches(event, peer, path) {
+            continue;
+        }
+        let newer_failure_recorded = match (event.updated_at, snapshot_updated_at) {
+            (Some(current), Some(seen)) => current > seen,
+            (Some(_), None) => true,
+            (None, _) => false,
+        };
+        if newer_failure_recorded {
+            continue;
+        }
+        event.failed = true;
+        event.retry_count = event.retry_count.max(SITE_REPLICATION_RETRY_FAILED_AFTER);
+        event.last_error = SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER.to_string();
+        escalated += 1;
+    }
+    escalated
+}
+
+async fn escalate_site_replication_retry_event_up_to(peer: &PeerInfo, path: &str, snapshot_updated_at: Option<OffsetDateTime>) {
+    let peer_owned = peer.clone();
+    let path_owned = path.to_string();
+    let result = update_site_replication_state(move |state| {
+        escalate_site_replication_retry_events_up_to(&mut state.retry_queue, &peer_owned, &path_owned, snapshot_updated_at);
+        Ok(())
+    })
+    .await;
+
+    if let Err(err) = result {
+        warn!(
+            component = LOG_COMPONENT_ADMIN,
+            subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+            event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+            peer = %peer.endpoint,
+            deployment_id = %peer.deployment_id,
+            path,
+            error = ?err,
+            "failed to escalate site replication retry event"
+        );
+    }
+}
+
+/// Whether the drain may attempt this event now.
+fn site_replication_retry_backoff_elapsed(event: &SiteReplicationRetryEvent, now: OffsetDateTime) -> bool {
+    let Some(updated_at) = event.updated_at else {
+        return true;
+    };
+    // 600 * 2^8 already exceeds the daily ceiling; capping the shift keeps
+    // the arithmetic overflow-free for any persisted retry_count.
+    let exponent = event.retry_count.saturating_sub(1).min(8);
+    let delay = (SITE_REPLICATION_RETRY_DRAIN_BASE_BACKOFF_SECS << exponent).min(SITE_REPLICATION_RETRY_DRAIN_MAX_BACKOFF_SECS);
+    now.unix_timestamp().saturating_sub(updated_at.unix_timestamp()) >= delay
+}
+
+/// The subset of the retry queue the background drain is allowed to touch.
+fn actionable_site_replication_retry_events(state: &SiteReplicationState, now: OffsetDateTime) -> Vec<SiteReplicationRetryEvent> {
+    state
+        .retry_queue
+        .iter()
+        .filter(|event| classify_site_replication_retry_event(event).is_some())
+        .filter(|event| state.peers.contains_key(&event.peer_deployment_id))
+        .filter(|event| site_replication_retry_backoff_elapsed(event, now))
+        .cloned()
+        .collect()
+}
+
+/// Background consumer for the retry queue, run from the reconcile tick.
+///
+/// Scope: this settles "delivered once and failed" entries whose replay is
+/// faithful (bucket ops, peer edits). Collapsed iam-item / bucket-meta
+/// entries are snapshot-resent and then *escalated*, not cleared — a failed
+/// deletion leaves no task in the snapshot, so remote absence stays unproven
+/// until a later delivery or a manual repair. A hook that never fired (crash
+/// between the local commit and the send) leaves no entry at all, so the
+/// drain is not a full cross-site diff-heal; manual repair remains the
+/// authoritative catch-all.
+async fn drain_site_replication_retry_queue() {
+    if let Err(err) = drain_site_replication_retry_queue_inner().await {
+        warn!(
+            event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+            component = LOG_COMPONENT_ADMIN,
+            subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+            result = "retry_drain_failed",
+            error = ?err,
+            "admin site replication state"
+        );
+    }
+}
+
+async fn drain_site_replication_retry_queue_inner() -> S3Result<()> {
+    let Some(runtime) = runtime_site_replication_targets().await? else {
+        return Ok(());
+    };
+    let actionable = actionable_site_replication_retry_events(&runtime.state, OffsetDateTime::now_utc());
+    if actionable.is_empty() {
+        return Ok(());
+    }
+    let Some(store) = current_object_store_handle() else {
+        return Ok(());
+    };
+    if runtime.state.pending_endpoint_refresh.is_some()
+        || runtime.state.pending_remove.is_some()
+        || runtime.state.pending_rotation.is_some()
+    {
+        // The tick-level gate ran before the reconcilers; a multi-step flow
+        // (endpoint refresh commits its pending marker without the lifecycle
+        // guard) may have started since. Re-check on the fresh state.
+        return Ok(());
+    }
+    // Serialize against operator repair execution. This does NOT close the
+    // dry-run -> execute window (dry-run takes no lock): a drain settling a
+    // replayable bucket-op entry in that window changes the preflight token
+    // and execute fails safe with "preflight is stale" — the operator
+    // re-runs the dry-run. Lock order matches repair: lifecycle guard (held
+    // by the reconcile tick) -> repair execution lock -> state object lock
+    // inside the send bookkeeping. An operator repair holding the lock makes
+    // this tick skip after the lock-acquire timeout.
+    with_config_object_write_lock(store, SITE_REPLICATION_REPAIR_EXECUTION_LOCK_PATH.to_string(), move || async move {
+        drain_site_replication_retry_queue_locked(runtime, actionable).await
+    })
+    .await
+    .map_err(ApiError::from)?
+}
+
+async fn drain_site_replication_retry_queue_locked(
+    runtime: SiteReplicationRuntime,
+    events: Vec<SiteReplicationRetryEvent>,
+) -> S3Result<()> {
+    let needs_plan = events
+        .iter()
+        .any(|event| !matches!(classify_site_replication_retry_event(event), Some(RetryDrainAction::PeerEdit)));
+    // The plan is a full local snapshot (buckets + IAM); build it once per
+    // tick and only when a snapshot resend is actually due.
+    let plan = if needs_plan {
+        let info = build_sr_info(&runtime.state, &runtime.local_peer).await?;
+        Some(site_replication_bootstrap_plan(&info)?)
+    } else {
+        None
+    };
+
+    let mut events_by_peer: BTreeMap<String, Vec<SiteReplicationRetryEvent>> = BTreeMap::new();
+    for event in events {
+        events_by_peer
+            .entry(event.peer_deployment_id.clone())
+            .or_default()
+            .push(event);
+    }
+
+    let mut settled = 0usize;
+    let mut failures = 0usize;
+    for (deployment_id, peer_events) in events_by_peer {
+        let Some(peer) = runtime.state.peers.get(&deployment_id) else {
+            continue;
+        };
+        if deployment_id == runtime.local_peer.deployment_id
+            || same_identity_endpoint(&peer.endpoint, &runtime.local_peer.endpoint)
+        {
+            continue;
+        }
+        let transport = match PeerTransport::for_runtime_peer(peer).await {
+            Ok(transport) => transport,
+            Err(err) => {
+                // Record the attempt so backoff advances for an unreachable
+                // peer instead of re-dialing it every tick.
+                for event in &peer_events {
+                    enqueue_site_replication_retry_event(peer, &event.path, &err).await;
+                }
+                failures += peer_events.len();
+                continue;
+            }
+        };
+        for event in peer_events {
+            let Some(action) = classify_site_replication_retry_event(&event) else {
+                continue;
+            };
+            match drain_one_site_replication_retry_event(&runtime, peer, &transport, &event, action, plan.as_ref()).await {
+                Ok(true) => settled += 1,
+                Ok(false) => {}
+                Err(_) => failures += 1,
+            }
+        }
+    }
+
+    if settled > 0 || failures > 0 {
+        info!(
+            event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+            component = LOG_COMPONENT_ADMIN,
+            subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+            result = "retry_drain_settled",
+            settled,
+            failures,
+            "admin site replication state"
+        );
+    }
+    Ok(())
+}
+
+/// Replay one retry event against its peer. Returns `Ok(true)` when the
+/// event was settled (delivered, or provably stale), `Ok(false)` when it was
+/// skipped, and `Err` after a failed delivery (already re-queued with an
+/// incremented retry count).
+async fn drain_one_site_replication_retry_event(
+    runtime: &SiteReplicationRuntime,
+    peer: &PeerInfo,
+    transport: &PeerTransport,
+    event: &SiteReplicationRetryEvent,
+    action: RetryDrainAction,
+    plan: Option<&SiteReplicationBootstrapPlan>,
+) -> S3Result<bool> {
+    let access_key = &runtime.state.service_account_access_key;
+    let secret_key = &runtime.service_account_secret_key;
+    match action {
+        RetryDrainAction::IamSnapshot | RetryDrainAction::BucketMetadataSnapshot => {
+            let Some(plan) = plan else {
+                return Ok(false);
+            };
+            let tasks: Vec<SiteReplicationRepairTask<'_>> = match action {
+                RetryDrainAction::IamSnapshot => plan.iam_items.iter().map(SiteReplicationRepairTask::Iam).collect(),
+                _ => plan
+                    .bucket_items
+                    .iter()
+                    .map(SiteReplicationRepairTask::BucketMetadata)
+                    .collect(),
+            };
+            for task in &tasks {
+                if let Err(err) = task.send(transport, access_key, secret_key).await {
+                    enqueue_site_replication_retry_event(peer, &event.path, &err).await;
+                    return Err(err);
+                }
+            }
+            // The snapshot replays every entity that still exists, but a
+            // failed *deletion* leaves no task in the plan — remote absence
+            // is unproven, so escalate (operator-visible, drain-idle) instead
+            // of clearing. Conditional on the snapshot timestamp: a hook
+            // failure recorded while this snapshot was in flight belongs to a
+            // newer commit and keeps the entry drain-eligible.
+            escalate_site_replication_retry_event_up_to(peer, &event.path, event.updated_at).await;
+            Ok(true)
+        }
+        RetryDrainAction::BucketOpReplay { operation, bucket } => {
+            let Some(plan) = plan else {
+                return Ok(false);
+            };
+            // Replay from the CURRENT plan, never the recorded path: the
+            // recorded query can carry an expired one-shot bootstrap token or
+            // a stale createdAt.
+            let make_op = operation == SITE_REPLICATION_BUCKET_OP_MAKE_WITH_VERSIONING;
+            let paths = if make_op {
+                &plan.bucket_make_ops
+            } else {
+                &plan.bucket_configure_ops
+            };
+            let tasks: Vec<SiteReplicationRepairTask<'_>> = paths
+                .iter()
+                .filter(|path| retry_bucket_name(path).as_deref() == Some(bucket.as_str()))
+                .map(|path| {
+                    if make_op {
+                        SiteReplicationRepairTask::BucketMake(path)
+                    } else {
+                        SiteReplicationRepairTask::Replication(path)
+                    }
+                })
+                .collect();
+            if tasks.is_empty() {
+                // The bucket left the plan (deleted, or replication no longer
+                // configured): the recorded intent is stale, settle it.
+                dequeue_site_replication_retry_event(peer, &event.path).await;
+                return Ok(true);
+            }
+            for task in &tasks {
+                if let Err(err) = task.send(transport, access_key, secret_key).await {
+                    enqueue_site_replication_retry_event(peer, &event.path, &err).await;
+                    return Err(err);
+                }
+            }
+            dequeue_site_replication_retry_event(peer, &event.path).await;
+            Ok(true)
+        }
+        RetryDrainAction::PeerEdit => {
+            // The recorded generation is stale by definition — the receiver
+            // fences it. Allocate a fresh generation and re-send the current
+            // peer records (a superset of the failed body; the receiver
+            // upserts), all inside one state transaction so the fence and the
+            // bodies agree.
+            let target_id = peer.deployment_id.clone();
+            let (generation, bodies) = update_site_replication_state(move |state| {
+                if !state.peers.contains_key(&target_id) {
+                    return Ok((None, Vec::new()));
+                }
+                Ok((Some(next_peer_edit_generation(state)), state.peers.values().cloned().collect::<Vec<_>>()))
+            })
+            .await?;
+            let Some(generation) = generation else {
+                // Peer left between the snapshot and now; the queue entry was
+                // already pruned by remove_sites.
+                return Ok(false);
+            };
+            let local_deployment_id = Some(runtime.local_peer.deployment_id.as_str()).filter(|id| !id.is_empty());
+            let edit_path = peer_edit_path_with_fence(local_deployment_id, generation);
+            let delivery_fence = local_deployment_id.is_some().then_some(generation);
+            for body in &bodies {
+                if let Err(err) = send_peer_admin_request_with_client(
+                    &transport.client,
+                    &transport.connection,
+                    &edit_path,
+                    access_key,
+                    secret_key,
+                    body,
+                )
+                .await
+                {
+                    enqueue_site_replication_retry_event_for_generation(
+                        peer,
+                        SITE_REPLICATION_PEER_EDIT_PATH,
+                        &err,
+                        delivery_fence,
+                    )
+                    .await;
+                    return Err(err);
+                }
+            }
+            dequeue_site_replication_retry_event_for_generation(peer, SITE_REPLICATION_PEER_EDIT_PATH, delivery_fence).await;
+            Ok(true)
+        }
+    }
+}
+
 /// Remove a retry event for (peer, path) from the queue on successful delivery.
 /// This is a no-op (load + no-op persist skipped) when no matching entry exists,
 /// avoiding unnecessary I/O on the common path.
@@ -11427,6 +11870,213 @@ mod tests {
         assert!(target_state.peers["remote"].skip_tls_verify);
     }
 
+    fn drain_event(peer: &str, path: &str, retry_count: u32, updated_at: Option<OffsetDateTime>) -> SiteReplicationRetryEvent {
+        SiteReplicationRetryEvent {
+            id: format!("evt-{peer}"),
+            peer_deployment_id: peer.to_string(),
+            peer_endpoint: format!("https://{peer}.example.com"),
+            path: path.to_string(),
+            retry_count,
+            failed: retry_count >= SITE_REPLICATION_RETRY_FAILED_AFTER,
+            last_error: "remote-operation-failed".to_string(),
+            updated_at,
+            edit_generation: None,
+        }
+    }
+
+    /// P1-3 red-light: the drain must only ever act on deliveries it can
+    /// replay faithfully. IAM / bucket-meta entries collapse per (peer, path)
+    /// with no body persisted — only a snapshot resend is truthful; bucket
+    /// makes/replication configs are re-derivable; destructive bucket ops and
+    /// `internal:` marker records (the pending-endpoint-refresh backup store)
+    /// are never background-replayed.
+    #[test]
+    fn test_classify_site_replication_retry_event_actions() {
+        let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp");
+        let classify = |path: &str| classify_site_replication_retry_event(&drain_event("remote", path, 1, Some(now)));
+
+        assert_eq!(
+            classify("/rustfs/admin/v3/site-replication/peer/iam-item"),
+            Some(RetryDrainAction::IamSnapshot)
+        );
+        assert_eq!(
+            classify("/rustfs/admin/v3/site-replication/peer/bucket-meta"),
+            Some(RetryDrainAction::BucketMetadataSnapshot)
+        );
+        assert_eq!(classify(SITE_REPLICATION_PEER_EDIT_PATH), Some(RetryDrainAction::PeerEdit));
+        assert_eq!(
+            classify(
+                "/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=make-with-versioning&createdAt=1"
+            ),
+            Some(RetryDrainAction::BucketOpReplay {
+                operation: SITE_REPLICATION_BUCKET_OP_MAKE_WITH_VERSIONING.to_string(),
+                bucket: "photos".to_string(),
+            })
+        );
+        assert_eq!(
+            classify("/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=configure-replication"),
+            Some(RetryDrainAction::BucketOpReplay {
+                operation: SITE_REPLICATION_BUCKET_OP_CONFIGURE_REPLICATION.to_string(),
+                bucket: "photos".to_string(),
+            })
+        );
+        // Destructive ops are operator territory: replaying a bucket delete
+        // against a peer whose bucket was since recreated is irreversible.
+        assert_eq!(
+            classify("/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=delete-bucket"),
+            None
+        );
+        assert_eq!(
+            classify("/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=force-delete-bucket"),
+            None
+        );
+        // `internal:` records store payloads in `last_error`, not failures.
+        assert_eq!(classify(SITE_REPLICATION_ENDPOINT_REFRESH_RETRY_PATH), None);
+        assert_eq!(classify("internal:some-future-marker"), None);
+        assert_eq!(classify("/rustfs/admin/v3/site-replication/peer/unknown"), None);
+    }
+
+    /// Exponential backoff gates every attempt: without it a dead peer's
+    /// entries hit `failed` (retry_count >= 3) within 30 minutes of reconcile
+    /// ticks and the retry stats lose their signal.
+    #[test]
+    fn test_site_replication_retry_backoff_schedule() {
+        let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp");
+        let at = |secs_ago: i64| Some(now - time::Duration::seconds(secs_ago));
+        let elapsed = |retry_count: u32, secs_ago: i64| {
+            site_replication_retry_backoff_elapsed(&drain_event("remote", "/p", retry_count, at(secs_ago)), now)
+        };
+
+        // No record of when it failed: attempt now.
+        assert!(site_replication_retry_backoff_elapsed(&drain_event("remote", "/p", 1, None), now));
+        // First failure: one reconcile interval.
+        assert!(!elapsed(1, 599));
+        assert!(elapsed(1, 601));
+        // Third failure: 600 * 2^2 = 2400s.
+        assert!(!elapsed(3, 1200));
+        assert!(elapsed(3, 2401));
+        // Ceiling: a long-dead peer is still probed daily, never less often.
+        assert!(!elapsed(30, 86_000));
+        assert!(elapsed(30, 86_401));
+    }
+
+    /// The actionable subset respects classification, peer membership and
+    /// backoff; everything else stays untouched in the queue.
+    #[test]
+    fn test_actionable_site_replication_retry_events_filters() {
+        let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp");
+        let old = Some(now - time::Duration::seconds(700));
+        let mut state = SiteReplicationState::default();
+        state
+            .peers
+            .insert("remote".to_string(), peer("remote", "https://remote.example.com"));
+
+        state.retry_queue = vec![
+            // Eligible: known peer, replayable, past backoff.
+            drain_event("remote", "/rustfs/admin/v3/site-replication/peer/iam-item", 1, old),
+            // Not yet due.
+            drain_event("remote", "/rustfs/admin/v3/site-replication/peer/bucket-meta", 2, Some(now)),
+            // Unknown peer (removed since the failure was recorded).
+            drain_event("gone", "/rustfs/admin/v3/site-replication/peer/iam-item", 1, old),
+            // Marker record, not a delivery failure.
+            drain_event("remote", SITE_REPLICATION_ENDPOINT_REFRESH_RETRY_PATH, 0, old),
+            // Destructive op: operator-only.
+            drain_event(
+                "remote",
+                "/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=delete-bucket",
+                1,
+                old,
+            ),
+        ];
+
+        let actionable = actionable_site_replication_retry_events(&state, now);
+        assert_eq!(actionable.len(), 1, "only the due, replayable, known-peer event is actionable");
+        assert_eq!(actionable[0].path, "/rustfs/admin/v3/site-replication/peer/iam-item");
+    }
+
+    /// The drain settles a peer-edit success under a freshly allocated
+    /// generation; legacy queue entries carry `edit_generation: None` and
+    /// must be cleared by that generation-scoped settlement (`(Some, None)`
+    /// falls through to removal), or the drain would spin on them forever.
+    #[test]
+    fn test_settle_clears_legacy_none_generation_event_for_generation_scoped_success() {
+        let target = peer("remote", "https://remote.example.com");
+        let mut queue = vec![drain_event("remote", SITE_REPLICATION_PEER_EDIT_PATH, 1, None)];
+        assert!(queue[0].edit_generation.is_none());
+
+        let settled = settle_site_replication_retry_events(&mut queue, &target, SITE_REPLICATION_PEER_EDIT_PATH, Some(42));
+
+        assert_eq!(settled, 1, "a legacy None-generation event must settle under a newer generation");
+        assert!(queue.is_empty());
+    }
+
+    /// A successful snapshot resend cannot prove a failed *deletion* was
+    /// replayed, so the collapsed entry is escalated (operator-visible,
+    /// drain-idle) instead of cleared — unless a newer failure was stamped
+    /// during the delivery window, which keeps the entry drain-eligible.
+    #[test]
+    fn test_escalate_up_to_marks_snapshot_replayed_and_keeps_newer_failures() {
+        let target = peer("remote", "https://remote.example.com");
+        let path = "/rustfs/admin/v3/site-replication/peer/iam-item";
+        let snapshot_at = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp");
+
+        // Failure re-stamped after the snapshot: untouched, still eligible.
+        let mut queue = vec![drain_event("remote", path, 2, Some(snapshot_at + time::Duration::seconds(5)))];
+        assert_eq!(
+            escalate_site_replication_retry_events_up_to(&mut queue, &target, path, Some(snapshot_at)),
+            0
+        );
+        assert!(!queue[0].failed);
+        assert!(
+            classify_site_replication_retry_event(&queue[0]).is_some(),
+            "a newer failure must stay drain-eligible"
+        );
+
+        // Unchanged since the snapshot: escalated, kept, drain-idle.
+        let mut queue = vec![drain_event("remote", path, 2, Some(snapshot_at))];
+        assert_eq!(
+            escalate_site_replication_retry_events_up_to(&mut queue, &target, path, Some(snapshot_at)),
+            1
+        );
+        assert_eq!(queue.len(), 1, "the entry must survive until remote absence is proven");
+        assert!(queue[0].failed);
+        assert_eq!(queue[0].last_error, SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER);
+        assert!(
+            classify_site_replication_retry_event(&queue[0]).is_none(),
+            "a snapshot-replayed entry must not be re-sent daily"
+        );
+        // Ordinary success dequeues must not clear the marker: collapsed
+        // paths are shared by every entity, so a successful Bob update
+        // proves nothing about a failed Alice deletion (second review
+        // round).
+        assert_eq!(dequeue_site_replication_retry_events(&mut queue, &target, path), 0);
+        assert_eq!(queue.len(), 1, "an escalated entry must survive an ordinary delivery success");
+        // Only a repair — the operator's accountability transfer — settles it.
+        assert_eq!(dequeue_site_replication_retry_events_including_escalated(&mut queue, &target, path), 1);
+        assert!(queue.is_empty());
+
+        // A later hook failure overwrites the marker and re-arms the drain.
+        let mut queue = vec![drain_event("remote", path, 2, Some(snapshot_at))];
+        escalate_site_replication_retry_events_up_to(&mut queue, &target, path, Some(snapshot_at));
+        upsert_site_replication_retry_event(&mut queue, &target, path, "peer offline", None);
+        assert!(classify_site_replication_retry_event(&queue[0]).is_some());
+
+        // Legacy entry without a timestamp: escalated.
+        let mut queue = vec![drain_event("remote", path, 2, None)];
+        assert_eq!(
+            escalate_site_replication_retry_events_up_to(&mut queue, &target, path, Some(snapshot_at)),
+            1
+        );
+
+        // Other (peer, path) entries are untouched.
+        let mut queue = vec![drain_event("other", path, 2, Some(snapshot_at))];
+        assert_eq!(
+            escalate_site_replication_retry_events_up_to(&mut queue, &target, path, Some(snapshot_at)),
+            0
+        );
+        assert!(!queue[0].failed);
+    }
+
     #[test]
     fn test_pending_endpoint_refresh_retry_summary_redacts_pem() {
         let pem = "-----BEGIN CERTIFICATE-----\nsecret-marker\n-----END CERTIFICATE-----";
@@ -16272,17 +16922,31 @@ mod tests {
     async fn test_retry_event_persist_must_not_wipe_concurrent_locked_rmw() {
         publish_ready_iam_context().await;
 
+        const ROUNDS: usize = 8;
         let seed = SiteReplicationState {
             pending_rotation: Some(PendingRotation {
                 id: "rot-1".to_string(),
                 access_key: "svc-account".to_string(),
                 ..Default::default()
             }),
+            // Retry events are only recorded for current peers; seed them so
+            // the concurrency assertion below exercises the persist path.
+            peers: (0..ROUNDS)
+                .map(|round| {
+                    let deployment_id = format!("peer-{round}-deployment");
+                    (
+                        deployment_id.clone(),
+                        PeerInfo {
+                            endpoint: format!("https://peer-{round}.example:9000"),
+                            deployment_id,
+                            ..Default::default()
+                        },
+                    )
+                })
+                .collect(),
             ..Default::default()
         };
         save_site_replication_state(&seed).await.expect("seed state");
-
-        const ROUNDS: usize = 8;
         for round in 0..ROUNDS {
             let peer = PeerInfo {
                 endpoint: format!("https://peer-{round}.example:9000"),

rustfs/src/storage/ecfs.rs

@@ -45,7 +45,7 @@ use rustfs_targets::EventName;
 use rustfs_utils::http::headers::{
     AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER,
 };
-use rustfs_utils::http::{SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, SUFFIX_TAGGING_TIMESTAMP, insert_str};
+use rustfs_utils::http::{SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, insert_str};
 use s3s::{S3, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, dto::*, s3_error};
 use std::collections::HashMap;
 use std::fmt::Debug;
@@ -461,11 +461,6 @@ impl S3 for FS {
             let mut eval_metadata = HashMap::new();
             insert_str(&mut eval_metadata, SUFFIX_REPLICATION_TIMESTAMP, jiff::Zoned::now().to_string());
             insert_str(&mut eval_metadata, SUFFIX_REPLICATION_STATUS, dsc.pending_status().unwrap_or_default());
-            insert_str(
-                &mut eval_metadata,
-                SUFFIX_TAGGING_TIMESTAMP,
-                OffsetDateTime::now_utc().format(&Rfc3339).unwrap_or_default(),
-            );
             opts.eval_metadata = Some(eval_metadata);
         }
 
@@ -1650,11 +1645,6 @@ impl S3 for FS {
             let mut eval_metadata = HashMap::new();
             insert_str(&mut eval_metadata, SUFFIX_REPLICATION_TIMESTAMP, jiff::Zoned::now().to_string());
             insert_str(&mut eval_metadata, SUFFIX_REPLICATION_STATUS, dsc.pending_status().unwrap_or_default());
-            insert_str(
-                &mut eval_metadata,
-                SUFFIX_TAGGING_TIMESTAMP,
-                OffsetDateTime::now_utc().format(&Rfc3339).unwrap_or_default(),
-            );
             opts.eval_metadata = Some(eval_metadata);
         }
 

rustfs/src/storage/options.rs

@@ -17,13 +17,11 @@ use crate::storage::storage_api::options_consumer::contract::{object::HTTPPrecon
 use http::header::{IF_MATCH, IF_NONE_MATCH};
 use http::{HeaderMap, HeaderValue};
 use rustfs_utils::http::{
-    AMZ_BUCKET_REPLICATION_STATUS, SUFFIX_FORCE_DELETE, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP,
-    SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC,
-    SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP,
-    SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP,
-    SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_SOURCE_VERSION_ID, SUFFIX_TAGGING_TIMESTAMP, get_header,
+    AMZ_BUCKET_REPLICATION_STATUS, SUFFIX_FORCE_DELETE, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC,
+    SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_REQUEST,
+    SUFFIX_SOURCE_VERSION_ID, get_header,
     header_compat::{MINIO_ENCRYPTION_PREFIX, RUSTFS_ENCRYPTION_PREFIX},
-    insert_header_map, insert_str,
+    insert_header_map,
     metadata_compat::{MINIO_INTERNAL_PREFIX, RUSTFS_INTERNAL_PREFIX},
 };
 use rustfs_utils::http::{
@@ -424,9 +422,6 @@ pub fn get_complete_multipart_upload_opts_with_replication_authorization(
         preserve_etag,
         ..Default::default()
     };
-    if replication_request {
-        apply_replication_timestamps_from_headers(headers, &mut opts);
-    }
     apply_replica_status_from_headers(headers, &mut opts, replication_request_authorized);
 
     fill_conditional_writes_opts_from_header(headers, &mut opts)?;
@@ -484,7 +479,6 @@ pub fn put_opts_from_headers_with_replication_authorization(
         if let Some(crc) = get_header(headers, SUFFIX_REPLICATION_SSEC_CRC) {
             insert_header_map(&mut opts.user_defined, SUFFIX_REPLICATION_SSEC_CRC, crc.into_owned());
         }
-        apply_replication_timestamps_from_headers(headers, &mut opts);
     }
     Ok(opts)
 }
@@ -510,47 +504,6 @@ fn replication_source_mtime(headers: &HeaderMap<HeaderValue>) -> Option<time::Of
     }
 }
 
-/// Parses one replication LWW timestamp header. Invalid values are dropped
-/// with a warning (same tolerance as [`replication_source_mtime`]) so a
-/// malformed source header cannot wedge the replication queue.
-fn replication_timestamp_header(headers: &HeaderMap<HeaderValue>, suffix: &str) -> Option<time::OffsetDateTime> {
-    let value = get_header(headers, suffix)?;
-    let value = value.trim();
-    match time::OffsetDateTime::parse(value, &time::format_description::well_known::Rfc3339) {
-        Ok(timestamp) => Some(timestamp),
-        Err(err) => {
-            tracing::warn!("Invalid {} value '{}' (replication request=true): {}", suffix, value, err);
-            None
-        }
-    }
-}
-
-/// Callers must gate on an authorized replication request: these headers are
-/// trusted source-cluster state, not client input.
-fn apply_replication_timestamps_from_headers(headers: &HeaderMap<HeaderValue>, opts: &mut ObjectOptions) {
-    opts.replication_tagging_timestamp = replication_timestamp_header(headers, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP);
-    opts.replication_retention_timestamp = replication_timestamp_header(headers, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP);
-    opts.replication_legalhold_timestamp = replication_timestamp_header(headers, SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP);
-
-    // Persist into the internal metadata keys so a later outbound replication
-    // pass (replication_target_boundary) reads the source's modification
-    // times instead of falling back to mod_time.
-    // TODO(P1-6): receiver-side LWW is still missing — when the stored
-    // per-category timestamp is newer than the inbound one, the existing
-    // tags/retention/legal-hold should win instead of being overwritten.
-    for (timestamp, suffix) in [
-        (opts.replication_tagging_timestamp, SUFFIX_TAGGING_TIMESTAMP),
-        (opts.replication_retention_timestamp, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP),
-        (opts.replication_legalhold_timestamp, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP),
-    ] {
-        if let Some(timestamp) = timestamp
-            && let Ok(value) = timestamp.format(&time::format_description::well_known::Rfc3339)
-        {
-            insert_str(&mut opts.user_defined, suffix, value);
-        }
-    }
-}
-
 fn apply_replica_status_from_headers(headers: &HeaderMap<HeaderValue>, opts: &mut ObjectOptions, authorized: bool) {
     if !authorized {
         return;
@@ -710,13 +663,6 @@ fn is_reserved_user_metadata_key(key: &str) -> bool {
         || starts_with_ignore_ascii_case(key, MINIO_INTERNAL_PREFIX)
         || starts_with_ignore_ascii_case(key, RUSTFS_ENCRYPTION_PREFIX)
         || starts_with_ignore_ascii_case(key, MINIO_ENCRYPTION_PREFIX)
-        // Replication transport names (source-replication timestamps,
-        // source-mtime/-etag/-version-id, ...). A bare stored key with one of
-        // these names is forwarded verbatim by the outbound replication
-        // header builder on a server-authorized request, so the receiver
-        // would persist attacker-chosen values as trusted internal LWW state.
-        || starts_with_ignore_ascii_case(key, "x-rustfs-source-")
-        || starts_with_ignore_ascii_case(key, "x-minio-source-")
 }
 
 fn stored_user_metadata_key(key: &str) -> String {
@@ -1653,136 +1599,6 @@ mod tests {
         assert!(opts_invalid.mod_time.is_none());
     }
 
-    /// A client PUT must not materialize the replication transport names as
-    /// bare stored user-metadata keys: the outbound replication header
-    /// builder forwards user metadata verbatim on a server-authorized
-    /// request, so a bare `x-rustfs-source-replication-*-timestamp` key would
-    /// deliver an attacker-chosen value into the replica's trusted internal
-    /// LWW state (for a tagless object nothing later overwrites it).
-    #[test]
-    fn test_replication_transport_names_cannot_be_forged_via_user_metadata() {
-        let mut headers = HeaderMap::new();
-        for name in [
-            "x-amz-meta-x-rustfs-source-replication-tagging-timestamp",
-            "x-amz-meta-x-minio-source-replication-legalhold-timestamp",
-            "x-rustfs-meta-x-rustfs-source-replication-retention-timestamp",
-            "x-amz-meta-x-rustfs-source-mtime",
-        ] {
-            headers.insert(
-                http::header::HeaderName::from_static(name),
-                HeaderValue::from_static("2026-01-02T03:04:05Z"),
-            );
-        }
-
-        let metadata = extract_metadata(&headers);
-
-        for forged in [
-            "x-rustfs-source-replication-tagging-timestamp",
-            "x-minio-source-replication-legalhold-timestamp",
-            "x-rustfs-source-replication-retention-timestamp",
-            "x-rustfs-source-mtime",
-        ] {
-            assert!(
-                metadata.get(forged).is_none(),
-                "{forged} must not be storable as a bare user-metadata key"
-            );
-        }
-        // The values survive, namespaced back under the user-metadata prefix.
-        assert_eq!(
-            metadata
-                .get("x-amz-meta-x-rustfs-source-replication-tagging-timestamp")
-                .map(String::as_str),
-            Some("2026-01-02T03:04:05Z")
-        );
-    }
-
-    #[test]
-    fn test_put_opts_from_headers_gates_replication_timestamp_persistence_on_authorization() {
-        // Sender-side LWW state (replication_target_boundary.rs) is read back
-        // from these internal metadata keys, so an authorized replication PUT
-        // must persist the inbound timestamp headers; an unauthorized client
-        // must not be able to forge them.
-        let mut headers = HeaderMap::new();
-        insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_REQUEST, "true");
-        insert_header(&mut headers, "source-replication-tagging-timestamp", "2026-01-02T03:04:05Z");
-        insert_header(&mut headers, "source-replication-retention-timestamp", "2026-01-02T03:04:06Z");
-        insert_header(&mut headers, "source-replication-legalhold-timestamp", "2026-01-02T03:04:07Z");
-
-        let untrusted = put_opts_from_headers(&headers, HashMap::new()).expect("ordinary PUT options should be created");
-        for suffix in [
-            "tagging-timestamp",
-            "objectlock-retention-timestamp",
-            "objectlock-legalhold-timestamp",
-        ] {
-            assert!(
-                rustfs_utils::http::get_str(&untrusted.user_defined, suffix).is_none(),
-                "unauthorized clients must not persist the {suffix} internal key"
-            );
-        }
-        assert!(untrusted.replication_tagging_timestamp.is_none());
-        assert!(untrusted.replication_retention_timestamp.is_none());
-        assert!(untrusted.replication_legalhold_timestamp.is_none());
-
-        let trusted = put_opts_from_headers_with_replication_authorization(&headers, HashMap::new(), true)
-            .expect("authorized replication request should parse");
-        let parse = |value: &str| {
-            time::OffsetDateTime::parse(value, &time::format_description::well_known::Rfc3339).expect("valid RFC3339")
-        };
-        assert_eq!(trusted.replication_tagging_timestamp, Some(parse("2026-01-02T03:04:05Z")));
-        assert_eq!(trusted.replication_retention_timestamp, Some(parse("2026-01-02T03:04:06Z")));
-        assert_eq!(trusted.replication_legalhold_timestamp, Some(parse("2026-01-02T03:04:07Z")));
-        for (suffix, expected) in [
-            ("tagging-timestamp", "2026-01-02T03:04:05Z"),
-            ("objectlock-retention-timestamp", "2026-01-02T03:04:06Z"),
-            ("objectlock-legalhold-timestamp", "2026-01-02T03:04:07Z"),
-        ] {
-            assert_eq!(
-                rustfs_utils::http::get_str(&trusted.user_defined, suffix).as_deref(),
-                Some(expected),
-                "authorized replication must persist the {suffix} internal key"
-            );
-        }
-    }
-
-    #[test]
-    fn test_complete_multipart_opts_persist_replication_timestamps_when_authorized() {
-        let mut headers = HeaderMap::new();
-        insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_REQUEST, "true");
-        insert_header(&mut headers, "replication-actual-object-size", "1");
-        insert_header(&mut headers, "source-replication-tagging-timestamp", "2026-01-02T03:04:05Z");
-        insert_header(&mut headers, "source-replication-retention-timestamp", "2026-01-02T03:04:06Z");
-        insert_header(&mut headers, "source-replication-legalhold-timestamp", "2026-01-02T03:04:07Z");
-
-        let untrusted = get_complete_multipart_upload_opts(&headers).expect("ordinary multipart options should be created");
-        for suffix in [
-            "tagging-timestamp",
-            "objectlock-retention-timestamp",
-            "objectlock-legalhold-timestamp",
-        ] {
-            assert!(
-                rustfs_utils::http::get_str(&untrusted.user_defined, suffix).is_none(),
-                "unauthorized multipart completes must not persist the {suffix} internal key"
-            );
-        }
-
-        let trusted = get_complete_multipart_upload_opts_with_replication_authorization(&headers, true)
-            .expect("authorized multipart complete should parse");
-        for (suffix, expected) in [
-            ("tagging-timestamp", "2026-01-02T03:04:05Z"),
-            ("objectlock-retention-timestamp", "2026-01-02T03:04:06Z"),
-            ("objectlock-legalhold-timestamp", "2026-01-02T03:04:07Z"),
-        ] {
-            assert_eq!(
-                rustfs_utils::http::get_str(&trusted.user_defined, suffix).as_deref(),
-                Some(expected),
-                "authorized multipart completes must persist the {suffix} internal key"
-            );
-        }
-        assert!(trusted.replication_tagging_timestamp.is_some());
-        assert!(trusted.replication_retention_timestamp.is_some());
-        assert!(trusted.replication_legalhold_timestamp.is_some());
-    }
-
     #[test]
     fn test_put_opts_from_headers_with_replica_status() {
         let mut headers = HeaderMap::new();