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.
fix(site-replication): escalate snapshot-replayed retry entries instead of clearing them
2026-08-15 17:43:13 +00:00 · 2026-08-16 01:02:42 +08:00 · 2026-08-15 19:14:23 +08:00 · 2026-08-15 11:01:58 +08:00 · 2026-08-15 10:42:36 +08:00 · 2026-08-15 10:19:28 +08:00
8 changed files with 751 additions and 382 deletions
@@ -190,17 +190,6 @@ pub(crate) const GET_METADATA_CACHE_REASON_VERSION_SUSPENDED: &str = "version_su
 pub(crate) const GET_METADATA_CACHE_REASON_VERSIONED: &str = "versioned";
 pub(crate) const GET_METADATA_EARLY_STOP_REASON_CONFLICTING_METADATA: &str = "conflicting_metadata";
 pub(crate) const GET_METADATA_EARLY_STOP_REASON_DELETE_MARKER: &str = "delete_marker";
-pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY: &str = "data_read_inline_body_verify";
-pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_DELETED: &str = "data_read_inline_deleted";
-pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY: &str = "data_read_inline_geometry";
-pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH: &str = "data_read_inline_identity_mismatch";
-pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD: &str = "data_read_inline_missing_payload";
-pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD: &str = "data_read_inline_missing_shard";
-pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE: &str = "data_read_inline_not_inline";
-pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE: &str = "data_read_inline_part_shape";
-pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_REMOTE: &str = "data_read_inline_remote";
-pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE: &str = "data_read_inline_size";
-pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_TRANSFORMED: &str = "data_read_inline_transformed";
 pub(crate) const GET_METADATA_EARLY_STOP_REASON_ERROR: &str = "error";
 pub(crate) const GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM: &str = "insufficient_quorum";
 pub(crate) const GET_METADATA_EARLY_STOP_REASON_NOT_FOUND: &str = "not_found";
@@ -562,32 +551,6 @@ mod tests {
        assert_eq!(GET_METADATA_CACHE_REASON_VERSIONED, "versioned");
        assert_eq!(GET_METADATA_EARLY_STOP_REASON_CONFLICTING_METADATA, "conflicting_metadata");
        assert_eq!(GET_METADATA_EARLY_STOP_REASON_DELETE_MARKER, "delete_marker");
-        assert_eq!(
-            GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY,
-            "data_read_inline_body_verify"
-        );
-        assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_DELETED, "data_read_inline_deleted");
-        assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY, "data_read_inline_geometry");
-        assert_eq!(
-            GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH,
-            "data_read_inline_identity_mismatch"
-        );
-        assert_eq!(
-            GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD,
-            "data_read_inline_missing_payload"
-        );
-        assert_eq!(
-            GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD,
-            "data_read_inline_missing_shard"
-        );
-        assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE, "data_read_inline_not_inline");
-        assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE, "data_read_inline_part_shape");
-        assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_REMOTE, "data_read_inline_remote");
-        assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE, "data_read_inline_size");
-        assert_eq!(
-            GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_TRANSFORMED,
-            "data_read_inline_transformed"
-        );
        assert_eq!(GET_METADATA_EARLY_STOP_REASON_ERROR, "error");
        assert_eq!(GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM, "insufficient_quorum");
        assert_eq!(GET_METADATA_EARLY_STOP_REASON_NOT_FOUND, "not_found");
@@ -32,22 +32,15 @@ use crate::diagnostics::get::{
    GET_METADATA_CACHE_REASON_NOT_READ_DATA, GET_METADATA_CACHE_REASON_PART_NUMBER,
    GET_METADATA_CACHE_REASON_RAW_DATA_MOVEMENT_READ, GET_METADATA_CACHE_REASON_USABLE, GET_METADATA_CACHE_REASON_VERSION_ID,
    GET_METADATA_CACHE_REASON_VERSION_SUSPENDED, GET_METADATA_CACHE_REASON_VERSIONED,
-    GET_METADATA_EARLY_STOP_REASON_CONFLICTING_METADATA, GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY,
-    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_DELETED, GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY,
-    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH,
-    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD,
-    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD, GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE,
-    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE, GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_REMOTE,
-    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE, GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_TRANSFORMED,
-    GET_METADATA_EARLY_STOP_REASON_DELETE_MARKER, GET_METADATA_EARLY_STOP_REASON_ERROR,
-    GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM, GET_METADATA_EARLY_STOP_REASON_NOT_FOUND,
-    GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST, GET_METADATA_EARLY_STOP_REASON_VALID_QUORUM,
-    GET_METADATA_EARLY_STOP_REASON_VERSION_MATCH_QUORUM, GET_METADATA_EARLY_STOP_REASON_VERSION_NOT_FOUND,
-    GET_METADATA_RESPONSE_CORRUPT, GET_METADATA_RESPONSE_DISK_NOT_FOUND, GET_METADATA_RESPONSE_ERROR,
-    GET_METADATA_RESPONSE_IGNORED, GET_METADATA_RESPONSE_NOT_FOUND, GET_METADATA_RESPONSE_TIMEOUT, GET_METADATA_RESPONSE_VALID,
-    GET_METADATA_RESPONSE_VERSION_NOT_FOUND, GET_OBJECT_PATH_CODEC_STREAMING, GET_OBJECT_PATH_DIRECT_MEMORY,
-    GET_OBJECT_PATH_INTERNAL_META, GET_OBJECT_PATH_LEGACY_DUPLEX, GET_OBJECT_PATH_SET_DISK, GET_STAGE_DECODE,
-    GET_STAGE_METADATA_CACHE_LOOKUP, GET_STAGE_METADATA_RESOLVE, GET_STAGE_RANGE, GET_STAGE_READER_SETUP,
+    GET_METADATA_EARLY_STOP_REASON_CONFLICTING_METADATA, GET_METADATA_EARLY_STOP_REASON_DELETE_MARKER,
+    GET_METADATA_EARLY_STOP_REASON_ERROR, GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM,
+    GET_METADATA_EARLY_STOP_REASON_NOT_FOUND, GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST,
+    GET_METADATA_EARLY_STOP_REASON_VALID_QUORUM, GET_METADATA_EARLY_STOP_REASON_VERSION_MATCH_QUORUM,
+    GET_METADATA_EARLY_STOP_REASON_VERSION_NOT_FOUND, GET_METADATA_RESPONSE_CORRUPT, GET_METADATA_RESPONSE_DISK_NOT_FOUND,
+    GET_METADATA_RESPONSE_ERROR, GET_METADATA_RESPONSE_IGNORED, GET_METADATA_RESPONSE_NOT_FOUND, GET_METADATA_RESPONSE_TIMEOUT,
+    GET_METADATA_RESPONSE_VALID, GET_METADATA_RESPONSE_VERSION_NOT_FOUND, GET_OBJECT_PATH_CODEC_STREAMING,
+    GET_OBJECT_PATH_DIRECT_MEMORY, GET_OBJECT_PATH_INTERNAL_META, GET_OBJECT_PATH_LEGACY_DUPLEX, GET_OBJECT_PATH_SET_DISK,
+    GET_STAGE_DECODE, GET_STAGE_METADATA_CACHE_LOOKUP, GET_STAGE_METADATA_RESOLVE, GET_STAGE_RANGE, GET_STAGE_READER_SETUP,
    GET_STAGE_READER_SETUP_DROP_PENDING, GET_STAGE_READER_SETUP_SCHEDULE, GET_STAGE_READER_SETUP_WAIT_QUORUM,
    GET_STAGE_READER_TASK_BITROT_READER_INIT, GET_STAGE_READER_TASK_FILE_OPEN, GET_STAGE_READER_TASK_READER_CONSTRUCTION,
    GetObjectFailureReason, classify_disk_error, get_stage_timer_if_enabled, record_get_object_pipeline_failure,
@@ -659,48 +652,36 @@ pub(in crate::set_disk) fn metadata_early_stop_candidate_matches(left: &FileInfo
        && left.erasure.distribution == right.erasure.distribution
 }

-pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason(
+pub(in crate::set_disk) async fn data_read_early_stop_inline_body_verified(
    bucket: &str,
    object: &str,
    candidate: &FileInfo,
    parts_metadata: &[FileInfo],
    disks: &[Option<DiskStore>],
-) -> Option<&'static str> {
-    if !candidate.inline_data() {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE);
-    }
-    if candidate.is_compressed()
+) -> bool {
+    if !candidate.inline_data()
+        || candidate.is_compressed()
        || candidate
            .metadata
            .keys()
            .any(|key| rustfs_utils::http::is_object_encryption_marker(key))
+        || candidate.is_remote()
+        || candidate.deleted
+        || candidate.size <= 0
+        || candidate.parts.len() != 1
+        || !candidate.has_valid_erasure_geometry()
    {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_TRANSFORMED);
-    }
-    if candidate.is_remote() {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_REMOTE);
-    }
-    if candidate.deleted {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_DELETED);
-    }
-    if candidate.size <= 0 {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
-    }
-    if candidate.parts.len() != 1 {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
-    }
-    if !candidate.has_valid_erasure_geometry() {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
+        return false;
    }

    let Ok(object_size) = usize::try_from(candidate.size) else {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
+        return false;
    };
    if candidate.parts.first().is_none_or(|part| part.size != object_size) {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
+        return false;
    }
    if !can_try_inline_data_shards_direct(object_size, candidate.erasure.block_size) {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
+        return false;
    }

    let Ok(erasure) = coding::Erasure::try_new_with_options(
@@ -709,18 +690,18 @@ pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason(
        candidate.erasure.block_size,
        candidate.uses_legacy_checksum,
    ) else {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
+        return false;
    };
-    let data_files =
-        match collect_inline_data_shard_fileinfos_by_index_or_reason(parts_metadata, candidate, erasure.data_shards, |index| {
+    let Some(data_files) =
+        collect_inline_data_shard_fileinfos_by_index(parts_metadata, candidate, erasure.data_shards, |index| {
            disks.get(index).is_some_and(Option::is_some)
-        }) {
-            Ok(data_files) => data_files,
-            Err(reason) => return Some(reason),
-        };
+        })
+    else {
+        return false;
+    };

    let Some(part) = candidate.parts.first() else {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
+        return false;
    };
    let checksum_info = candidate.erasure.get_checksum_info(part.number);
    let checksum_algo = if candidate.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
@@ -740,13 +721,12 @@ pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason(
    let Ok(mut readers) =
        build_inline_bitrot_readers_from_refs(&data_files, bucket, object, read_length, shard_size, &checksum_algo, false).await
    else {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY);
+        return false;
    };

-    match try_read_inline_data_shards_direct(&mut readers, erasure.data_shards, read_length, object_size).await {
-        Some(body) if body.len() == object_size => None,
-        _ => Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
-    }
+    try_read_inline_data_shards_direct(&mut readers, erasure.data_shards, read_length, object_size)
+        .await
+        .is_some_and(|body| body.len() == object_size)
 }

 pub(in crate::set_disk) fn classify_metadata_response_error(err: &DiskError) -> &'static str {
@@ -2489,7 +2469,6 @@ impl SetDisks {
        let mut next_fanout_index = 0usize;
        let mut scheduled_count = 0usize;
        let mut force_full_wait = false;
-        let mut final_miss_reason_override = None;
        let spawn_read_version =
            |join_set: &mut JoinSet<(usize, disk::error::Result<FileInfo>, Duration)>, index: usize, disk: Option<DiskStore>| {
                let task_opts = opts;
@@ -2562,29 +2541,17 @@ impl SetDisks {
                    .or_else(|| accumulator.version_early_stop_decision())
            {
                let should_return_early = if read_data {
-                    match accumulator.candidate.as_ref() {
-                        Some(candidate) => match data_read_early_stop_inline_body_miss_reason(
-                            bucket.as_ref(),
-                            object.as_ref(),
-                            candidate,
-                            &ress,
-                            disks,
-                        )
-                        .await
-                        {
-                            None => true,
-                            Some(reason) => {
-                                force_full_wait = true;
-                                final_miss_reason_override = Some(reason);
-                                false
-                            }
-                        },
-                        None => {
-                            force_full_wait = true;
-                            final_miss_reason_override = Some(GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM);
-                            false
+                    let allow_data_read_early_stop = match accumulator.candidate.as_ref() {
+                        Some(candidate) => {
+                            data_read_early_stop_inline_body_verified(bucket.as_ref(), object.as_ref(), candidate, &ress, disks)
+                                .await
                        }
+                        None => false,
+                    };
+                    if !allow_data_read_early_stop {
+                        force_full_wait = true;
                    }
+                    allow_data_read_early_stop
                } else {
                    true
                };
@@ -2646,12 +2613,7 @@ impl SetDisks {
            }
        }

-        let accumulator_miss_reason = accumulator.final_miss_reason();
-        let final_miss_reason = match (final_miss_reason_override, accumulator_miss_reason) {
-            (Some(reason), GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM) => reason,
-            _ => accumulator_miss_reason,
-        };
-        rustfs_io_metrics::record_get_object_metadata_early_stop_miss(metrics_path, final_miss_reason);
+        rustfs_io_metrics::record_get_object_metadata_early_stop_miss(metrics_path, accumulator.final_miss_reason());
        rustfs_io_metrics::record_get_object_metadata_early_stop_saved_responses(metrics_path, 0);
        rustfs_io_metrics::record_get_object_metadata_fanout_lifecycle(metrics_path, scheduled_count, scheduled_count, 0);
        let diagnostics = MetadataFanoutDiagnostics::new(fanout_start.elapsed(), observations);
@@ -6105,126 +6067,11 @@ mod tests {
            .clone();

        assert!(
-            data_read_early_stop_inline_body_miss_reason(bucket, object, &candidate, &parts_metadata, &disks)
-                .await
-                .is_none(),
+            data_read_early_stop_inline_body_verified(bucket, object, &candidate, &parts_metadata, &disks).await,
            "legacy inline metadata must use the legacy bitrot shard sizing and checksum algorithm"
        );
    }

-    #[tokio::test]
-    async fn data_read_early_stop_reports_inline_miss_reasons() {
-        let bucket = "inline-data-get-miss-reason-bucket";
-        let object = "inline-data-get-miss-reason-object";
-        let payload = b"verified inline payload";
-        let (_dirs, disks) = call_counter_local_disks(bucket, 4).await;
-        let files = inline_metadata_fanout_fileinfos_with_mode(bucket, object, payload, false).await;
-        let distribution = files
-            .first()
-            .map(|file| file.erasure.distribution.clone())
-            .expect("fixture should include metadata");
-        let order = bounded_metadata_fanout_order(bucket, object, 4, 2);
-        let mut parts_metadata = vec![FileInfo::default(); 4];
-        for disk_index in order.into_iter().take(3) {
-            let block_index = distribution
-                .get(disk_index)
-                .copied()
-                .expect("fixture distribution should cover every disk");
-            parts_metadata[disk_index] = files
-                .get(block_index.checked_sub(1).expect("erasure block indexes are one-based"))
-                .expect("fixture should include every distributed shard")
-                .clone();
-        }
-        let candidate = parts_metadata
-            .iter()
-            .find(|file| file.name == object)
-            .expect("fixture should include observed metadata")
-            .clone();
-        let data_disk = distribution
-            .iter()
-            .position(|block_index| *block_index == 1)
-            .expect("fixture distribution should include first data shard");
-
-        assert_eq!(
-            data_read_early_stop_inline_body_miss_reason(bucket, object, &candidate, &parts_metadata, &disks).await,
-            None
-        );
-
-        let mut not_inline = candidate.clone();
-        rustfs_utils::http::remove_str(&mut not_inline.metadata, rustfs_utils::http::SUFFIX_INLINE_DATA);
-        assert_eq!(
-            data_read_early_stop_inline_body_miss_reason(bucket, object, &not_inline, &parts_metadata, &disks).await,
-            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE)
-        );
-
-        let mut transformed = candidate.clone();
-        rustfs_utils::http::insert_str(&mut transformed.metadata, rustfs_utils::http::SUFFIX_COMPRESSION, "zstd".to_string());
-        assert_eq!(
-            data_read_early_stop_inline_body_miss_reason(bucket, object, &transformed, &parts_metadata, &disks).await,
-            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_TRANSFORMED)
-        );
-
-        let mut deleted = candidate.clone();
-        deleted.deleted = true;
-        assert_eq!(
-            data_read_early_stop_inline_body_miss_reason(bucket, object, &deleted, &parts_metadata, &disks).await,
-            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_DELETED)
-        );
-
-        let mut zero_size = candidate.clone();
-        zero_size.size = 0;
-        assert_eq!(
-            data_read_early_stop_inline_body_miss_reason(bucket, object, &zero_size, &parts_metadata, &disks).await,
-            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE)
-        );
-
-        let mut multipart = candidate.clone();
-        multipart.parts.push(multipart.parts[0].clone());
-        assert_eq!(
-            data_read_early_stop_inline_body_miss_reason(bucket, object, &multipart, &parts_metadata, &disks).await,
-            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE)
-        );
-
-        let mut invalid_geometry = candidate.clone();
-        invalid_geometry.erasure.data_blocks = 0;
-        assert_eq!(
-            data_read_early_stop_inline_body_miss_reason(bucket, object, &invalid_geometry, &parts_metadata, &disks).await,
-            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY)
-        );
-
-        let mut missing_shard = parts_metadata.clone();
-        missing_shard[data_disk] = FileInfo::default();
-        assert_eq!(
-            data_read_early_stop_inline_body_miss_reason(bucket, object, &candidate, &missing_shard, &disks).await,
-            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD)
-        );
-
-        let mut missing_payload = parts_metadata.clone();
-        missing_payload[data_disk].data = None;
-        assert_eq!(
-            data_read_early_stop_inline_body_miss_reason(bucket, object, &candidate, &missing_payload, &disks).await,
-            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD)
-        );
-
-        let mut identity_mismatch = parts_metadata.clone();
-        identity_mismatch[data_disk].version_id = Some(Uuid::new_v4());
-        assert_eq!(
-            data_read_early_stop_inline_body_miss_reason(bucket, object, &candidate, &identity_mismatch, &disks).await,
-            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH)
-        );
-
-        let mut corrupt = parts_metadata.clone();
-        if let Some(data) = corrupt[data_disk].data.as_mut() {
-            let mut corrupt_data = data.to_vec();
-            corrupt_data[0] ^= 0x01;
-            *data = Bytes::from(corrupt_data);
-        }
-        assert_eq!(
-            data_read_early_stop_inline_body_miss_reason(bucket, object, &candidate, &corrupt, &disks).await,
-            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY)
-        );
-    }
-
    #[test]
    #[serial_test::serial]
    fn metadata_fanout_lifecycle_records_real_early_stop_abort() {
@@ -6314,7 +6161,7 @@ mod tests {
                &[
                    ("path", GET_OBJECT_PATH_INTERNAL_META),
                    ("decision", "miss"),
-                    ("reason", GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
+                    ("reason", GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM),
                ],
            ),
            1,
@@ -6326,7 +6173,7 @@ mod tests {
                &[
                    ("path", GET_OBJECT_PATH_LEGACY_DUPLEX),
                    ("decision", "miss"),
-                    ("reason", GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
+                    ("reason", GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM),
                ],
            ),
            0,
@@ -59,10 +59,7 @@ use crate::client::{object_api_utils::get_raw_etag, transition_api::ReaderImpl};
 use crate::cluster::rpc::heal_bucket_local_on_disks;
 use crate::data_usage::record_compression_total_memory;
 use crate::diagnostics::get::{
-    GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART, GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY,
-    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH,
-    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD,
-    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD, GET_OBJECT_PATH_BODY_CACHE, GET_OBJECT_PATH_CODEC_STREAMING,
+    GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART, GET_OBJECT_PATH_BODY_CACHE, GET_OBJECT_PATH_CODEC_STREAMING,
    GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE, GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE, GET_OBJECT_PATH_DIRECT_MEMORY,
    GET_OBJECT_PATH_EMPTY, GET_OBJECT_PATH_INLINE_DIRECT, GET_OBJECT_PATH_INTERNAL_META, GET_OBJECT_PATH_LEGACY_DUPLEX,
    GET_OBJECT_PATH_REMOTE_TRANSITION, GET_OBJECT_PATH_SET_DISK, GET_STAGE_DECODE, GET_STAGE_EMIT, GET_STAGE_INLINE_PREPARE,
@@ -3866,20 +3863,11 @@ fn inline_erasure_shard_file_offset(
 }

 fn collect_inline_data_shard_fileinfos_by_index<'a>(
-    parts_metadata: &'a [FileInfo],
-    fi: &FileInfo,
-    data_shards: usize,
-    disk_is_online: impl FnMut(usize) -> bool,
-) -> Option<Vec<&'a FileInfo>> {
-    collect_inline_data_shard_fileinfos_by_index_or_reason(parts_metadata, fi, data_shards, disk_is_online).ok()
-}
-
-fn collect_inline_data_shard_fileinfos_by_index_or_reason<'a>(
    parts_metadata: &'a [FileInfo],
    fi: &FileInfo,
    data_shards: usize,
    mut disk_is_online: impl FnMut(usize) -> bool,
-) -> std::result::Result<Vec<&'a FileInfo>, &'static str> {
+) -> Option<Vec<&'a FileInfo>> {
    let distribution = &fi.erasure.distribution;
    let mut data_files = vec![None; data_shards];

@@ -3887,35 +3875,27 @@ fn collect_inline_data_shard_fileinfos_by_index_or_reason<'a>(
        if !disk_is_online(disk_index) {
            continue;
        }
-        let Some(&block_index) = distribution.get(disk_index) else {
-            return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
-        };
+        let block_index = *distribution.get(disk_index)?;
        if block_index == 0 || block_index > data_shards {
            continue;
        }
-        if file_info.name.is_empty() {
-            return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD);
-        }
        if file_info.erasure.index != block_index {
-            return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH);
+            continue;
        }
        if !file_info.has_valid_erasure_geometry() {
-            return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
+            continue;
        }
        if !core::io_primitives::metadata_early_stop_candidate_matches(file_info, fi) {
-            return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH);
+            continue;
        }
        if file_info.data.as_ref().is_none_or(|data| data.is_empty()) {
-            return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD);
+            continue;
        }

        data_files[block_index - 1] = Some(file_info);
    }

-    data_files
-        .into_iter()
-        .collect::<Option<Vec<_>>>()
-        .ok_or(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD)
+    data_files.into_iter().collect()
 }

 impl SetDisks {
@@ -135,7 +135,10 @@ impl FileMeta {
        let i = buf.len() as u64;

        // check version, buf = buf[8..]
-        let (buf, _, _) = Self::check_xl2_v1(buf)?;
+        let (buf, _, _) = Self::check_xl2_v1(buf).map_err(|e| {
+            error!("failed to check XL2 v1 format: {}", e);
+            e
+        })?;

        if buf.len() < 5 {
            error!(
@@ -102,7 +102,7 @@ bytes.workspace = true
 hex-simd.workspace = true

 [dev-dependencies]
-tracing-subscriber = { workspace = true, features = ["json", "env-filter", "time"] }
+tracing-subscriber = { workspace = true, features = ["env-filter", "time"] }
 serial_test = { workspace = true }
 temp-env = { workspace = true }
 tempfile = { workspace = true }
@@ -65,7 +65,6 @@ const LOG_SUBSYSTEM_FOLDER: &str = "folder";
 const LOG_SUBSYSTEM_LIFECYCLE: &str = "lifecycle";
 const LOG_SUBSYSTEM_HEAL: &str = "heal";
 const EVENT_SCANNER_FOLDER_STATE: &str = "scanner_folder_state";
-const EVENT_SCANNER_METADATA_CORRUPT: &str = "scanner_metadata_corrupt";
 const EVENT_SCANNER_LIFECYCLE_ACTION: &str = "scanner_lifecycle_action";
 const EVENT_SCANNER_HEAL_ADMISSION: &str = "scanner_heal_admission";
 const EVENT_SCANNER_ALERT_STATE: &str = "scanner_alert_state";
@@ -2155,34 +2154,17 @@ impl FolderScanner {
                            self.record_failed(&item.path);

                            if should_log_failed_object(into.failed_objects) {
-                                if let GetSizeFailureAction::HealMetadata { object } = &failure_action {
-                                    error!(
-                                        target: "rustfs::scanner::folder",
-                                        event = EVENT_SCANNER_METADATA_CORRUPT,
-                                        component = LOG_COMPONENT_SCANNER,
-                                        subsystem = LOG_SUBSYSTEM_FOLDER,
-                                        drive = %self.local_disk.path().display(),
-                                        bucket = %item.bucket,
-                                        object = %object,
-                                        metadata_path = %item.path,
-                                        failed_objects = into.failed_objects,
-                                        state = "metadata_corrupt",
-                                        error = %e,
-                                        "Scanner detected corrupt object metadata"
-                                    );
-                                } else {
-                                    warn!(
-                                        target: "rustfs::scanner::folder",
-                                        event = EVENT_SCANNER_FOLDER_STATE,
-                                        component = LOG_COMPONENT_SCANNER,
-                                        subsystem = LOG_SUBSYSTEM_FOLDER,
-                                        path = %item.path,
-                                        failed_objects = into.failed_objects,
-                                        state = "get_size_failed",
-                                        error = %e,
-                                        "Scanner folder failed to get object size"
-                                    );
-                                }
+                                warn!(
+                                    target: "rustfs::scanner::folder",
+                                    event = EVENT_SCANNER_FOLDER_STATE,
+                                    component = LOG_COMPONENT_SCANNER,
+                                    subsystem = LOG_SUBSYSTEM_FOLDER,
+                                    path = %item.path,
+                                    failed_objects = into.failed_objects,
+                                    state = "get_size_failed",
+                                    error = %e,
+                                    "Scanner folder failed to get object size"
+                                );
                            }
                        }

@@ -3072,59 +3054,12 @@ mod tests {
    use crate::{DiskOption, Endpoint, STORAGE_FORMAT_FILE, TierStats, new_disk, storageclass};
    use rustfs_filemeta::{FileInfo, FileMeta};
    use serial_test::serial;
-    use std::io::Write;
    #[cfg(unix)]
    use std::os::unix::fs::{PermissionsExt, symlink};
-    use std::sync::Mutex;
    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
    use temp_env::{with_var, with_var_unset};
-    use tracing_subscriber::fmt::MakeWriter;
    use uuid::Uuid;

-    #[derive(Clone, Default)]
-    struct CapturedLogs {
-        buffer: Arc<Mutex<Vec<u8>>>,
-    }
-
-    struct CapturedLogWriter {
-        buffer: Arc<Mutex<Vec<u8>>>,
-    }
-
-    impl CapturedLogs {
-        fn contents(&self) -> String {
-            let buffer = self
-                .buffer
-                .lock()
-                .expect("captured logs mutex should not be poisoned")
-                .clone();
-            String::from_utf8(buffer).expect("captured logs should be valid UTF-8")
-        }
-    }
-
-    impl Write for CapturedLogWriter {
-        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
-            self.buffer
-                .lock()
-                .expect("captured logs mutex should not be poisoned")
-                .extend_from_slice(buf);
-            Ok(buf.len())
-        }
-
-        fn flush(&mut self) -> std::io::Result<()> {
-            Ok(())
-        }
-    }
-
-    impl<'a> MakeWriter<'a> for CapturedLogs {
-        type Writer = CapturedLogWriter;
-
-        fn make_writer(&'a self) -> Self::Writer {
-            CapturedLogWriter {
-                buffer: Arc::clone(&self.buffer),
-            }
-        }
-    }
-
    #[test]
    fn scanner_size_summary_application_saturates_usage_counters() {
        let target = "arn:minio:replication::target".to_string();
@@ -4607,19 +4542,9 @@ mod tests {
        assert!(budget.entries_visited() >= 1);
    }

-    #[tokio::test(flavor = "current_thread")]
+    #[tokio::test]
    #[serial]
    async fn test_scan_folder_corrupt_xl_meta_stops_erasure_data_dir_descent() {
-        let logs = CapturedLogs::default();
-        let subscriber = tracing_subscriber::fmt()
-            .json()
-            .with_max_level(tracing::Level::ERROR)
-            .with_writer(logs.clone())
-            .with_ansi(false)
-            .without_time()
-            .finish();
-        let _subscriber_guard = tracing::subscriber::set_default(subscriber);
-
        let (mut scanner, temp_dir) = build_test_scanner().await;
        let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());

@@ -4671,30 +4596,6 @@ mod tests {
        assert!(!budget.budget_elapsed());
        assert_eq!(budget.reason(), None);

-        let captured = logs.contents();
-        assert!(
-            !captured.contains("failed to check XL2 v1 format"),
-            "the context-free filemeta parser error must not be emitted"
-        );
-        let events = captured
-            .lines()
-            .map(|line| serde_json::from_str::<serde_json::Value>(line).expect("captured scanner log should be valid JSON"))
-            .filter(|line| line["fields"]["event"] == EVENT_SCANNER_METADATA_CORRUPT)
-            .collect::<Vec<_>>();
-        assert_eq!(
-            events.len(),
-            1,
-            "one corrupt metadata observation must emit one scanner-owned diagnostic event"
-        );
-        let fields = &events[0]["fields"];
-        assert_eq!(fields["component"], LOG_COMPONENT_SCANNER);
-        assert_eq!(fields["subsystem"], LOG_SUBSYSTEM_FOLDER);
-        assert_eq!(fields["drive"], temp_dir.to_string_lossy().as_ref());
-        assert_eq!(fields["bucket"], "bucket");
-        assert_eq!(fields["object"], "object");
-        assert_eq!(fields["metadata_path"], metadata_path.to_string_lossy().as_ref());
-        assert_eq!(fields["state"], "metadata_corrupt");
-
        let retry_budget = ScannerCycleBudget::new_with_progress_tracking(
            &parent,
            crate::scanner_budget::ScannerCycleBudgetConfig {
@@ -3849,6 +3849,17 @@ impl ScannerIODisk for Disk {
        let fivs = match meta.get_file_info_versions(item.bucket.as_str(), item.object_path().as_str(), false) {
            Ok(versions) => versions,
            Err(e) => {
+                error!(
+                    target: "rustfs::scanner::io",
+                    event = EVENT_SCANNER_DISK_BUCKET_STATE,
+                    component = LOG_COMPONENT_SCANNER,
+                    subsystem = LOG_SUBSYSTEM_IO,
+                    bucket = %item.bucket,
+                    object = %item.object_path(),
+                    state = "file_info_versions_failed",
+                    error = %e,
+                    "Scanner disk bucket failed to resolve file info versions"
+                );
                return Err(scanner_metadata_corrupt_error(
                    format!("failed to resolve file info versions: {e}"),
                    &item.bucket,
@@ -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"),
Author	SHA1	Message	Date
唐小鸭	af802756c5	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.	2026-08-16 01:02:42 +08:00
唐小鸭	971addca6e	fix(site-replication): escalate snapshot-replayed retry entries instead of clearing them Review: the bootstrap-plan snapshot cannot replay deletions — a deleted IAM entity or absent bucket config produces no task, so clearing the collapsed iam-item / bucket-meta entry after a successful snapshot resend silently lost a failed delete and the peer kept stale state permanently. The drain now keeps those entries until remote absence is proven: after a successful snapshot resend the entry is escalated (failed=true, marker last_error) so it stays operator-visible in retry_stats, and classification skips marked entries so the once-per-failure-episode snapshot is not re-sent daily. A newer hook failure overwrites the marker and re-arms the drain; a later full delivery or a manual repair settles the entry. Escalation is conditional on the snapshot timestamp, preserving the earlier review's in-flight-failure guarantee. Bucket ops and peer edits keep auto-settle — their replays are faithful.	2026-08-15 19:14:23 +08:00
唐小鸭	2c7d1f1f9f	fix(site-replication): harden the retry drain against review findings Adversarial review of the drain surfaced one real race and three cheap hardenings: - Conditional settlement for collapsed (constant-path) iam-item / bucket-meta entries: the snapshot resend proves delivery of the state as of plan-build time, so a hook failure stamped during the delivery window (a newer local commit the snapshot did not contain) must survive the snapshot's success instead of being cleared — previously the peer would silently diverge until the next same-path failure. (The operator repair path shares this collapse but keeps its existing unconditional settle; the drain runs every tick and needed the guard first.) - Re-check the pending_* gates on the freshly loaded state: an endpoint refresh can commit its pending marker mid-tick without the lifecycle guard. - Do not upsert retry events for peers that already left the state — remove_sites pruned their entries and they can never drain again. - Correct the mutual-exclusion comment: the repair-execution lock does NOT close the dry-run -> execute window (dry-run takes no lock); that window fails safe via the preflight token, which hashes the replayable entries. New tests pin the (Some settled, None failed) generation-settle combination the peer-edit drain depends on, and the snapshot-relative conditional settlement.	2026-08-15 11:01:58 +08:00
唐小鸭	5328e8b958	feat(site-replication): drain the retry queue from the reconcile tick The retry queue recorded every failed peer delivery but had no consumer — missed IAM/bucket metadata updates drifted until an operator ran a manual repair (backlog#1675 P1-3). The 600s reconcile tick now drains the queue behind the existing lifecycle guard and pending_* gates. Replay discipline (pinned by the red-light tests): - IAM / bucket-meta entries collapse per (peer, path) and their bodies are not persisted, so the only faithful replay is the current bootstrap-plan snapshot (SiteReplicationRepairTask reuse) sent to the affected peer. - make-with-versioning / configure-replication ops are re-derived from the CURRENT plan for their bucket — never the recorded path, whose query can carry an expired one-shot bootstrap token; an op whose bucket left the plan is provably stale and is settled. - Peer edits are re-sent as the current peer records under a freshly allocated generation inside one state transaction — the recorded generation is stale by definition and the receiver would fence it. - Destructive bucket ops (delete-bucket / force-delete-bucket) and internal: marker records (pending-endpoint-refresh backup store) are never background-replayed. - Exponential backoff (600s * 2^(n-1), 24h ceiling) gates each attempt; an unreachable peer's transport failure re-queues its events so the backoff still advances. Concurrency: the drain takes the repair-execution config lock — the operator repair preflight token hashes the replayable retry events, so settling them between dry-run and execute would strand the operator on a stale preflight. Lock order matches repair (lifecycle guard -> repair execution lock -> state object lock); success/failure settlement reuses the generation-fenced upsert/settle semantics from #5882/#6097. Scope note: the drain settles 'delivered once and failed' entries. A hook that never fired (crash between local commit and send) leaves no entry; a low-frequency plan-diff catch-all remains follow-up work, and manual repair stays authoritative.	2026-08-15 10:42:36 +08:00
唐小鸭	1e16e06f8a	test(site-replication): pin the background retry-drain discipline Red-light evidence for backlog#1675 P1-3: the retry queue has no background consumer — every failed peer delivery waits for a manual repair. The new tests specify the drain rules before the drain exists: - classification: IAM / bucket-meta entries collapse per (peer, path) with no persisted body, so only a bootstrap-plan snapshot resend is a faithful replay; make-with-versioning / configure-replication are re-derivable per bucket; peer edits are re-sent under a fresh generation; destructive bucket ops and internal: marker records (pending-endpoint-refresh backup store) are never background-replayed - exponential backoff (600s * 2^(n-1), 24h ceiling) gates every attempt, otherwise a dead peer's entries hit the failed threshold within 30 minutes of reconcile ticks - the actionable subset respects classification, current peer membership and backoff All fail against the placeholder implementations (no consumer).	2026-08-15 10:19:28 +08:00