fix(replication): close the pre-stable convergence gaps from backlog#2367 (#7626)

* fix(replication): total-order rule sort and honor V1 top-level Prefix Rule matching had two defects from the pre-GA replication audit (rustfs/backlog#2367 C-1 and C-2): - The actionable-rule sort compared same-destination rules by priority but answered Equal for any other pair, which is not a total order; the standard library sort panics on such comparators once a slice exceeds the insertion-sort threshold, so an object matching more than 20 enabled rules across two or more targets could panic the PUT or DELETE task. Rules now sort by priority descending with destination and id as tie-breakers, and filter_target_arns preserves that order instead of draining a HashSet. - A V1 rule written without a <Filter> carries its prefix at the top level; that field was never read, so <Prefix>logs/</Prefix> matched every object. ReplicationRuleExt::prefix now falls back to it, with a <Filter> keeping precedence. The existing prefix fixtures were built this way and had been asserting nothing. * fix(admin): advertise data-usage and listen capabilities to rc The rc client gated `rc du` and `rc watch` on a pinned contract that matched server versions by the string prefix `1.0.0-rc.`; a server that reports `1.0.0` no longer matches, and the dynamic `advertised` list did not carry either name, so `rc du` against a GA server fails with an unsupported-capability error (rustfs/backlog#2367 E-2). Advertise `admin.data-usage` from the admin route inventory like the IAM entries, and `listen_notification` for the bucket `?events=` extension route the admin router dispatches. The client merges advertised entries ahead of its pinned contract, so no version sniffing is needed. * fix(site-replication): stop notifying the local site on remove and rotate The pending-remove and pending-rotation notification loops skipped the local site by endpoint only, while finalization identifies it by deployment id or endpoint. The reconcile tick resolves the local peer from the node's own listen address (and a handler from the request Host), so `remove --all` dialed the site's registered endpoint, waited out the request timeout against the lifecycle lock it was holding, and answered `Partial: failed to notify 1 peer(s)` for a removal that had succeeded (rustfs/backlog#2367 A-4, backlog#2195 item 3). Both loops now iterate the peers still awaiting notification through one helper that applies the finalization identity. * fix(site-replication): promote and settle IAM retries without a tick of slack Two retry-queue behaviours kept an IAM change from converging for ten to twenty minutes after a peer came back (rustfs/backlog#2367 A-1 and A-3, backlog#2305): - The lightweight 30-second pass filtered its reachability probe to bucket ops, so a backed-off IAM or bucket-metadata snapshot waited for the 600-second tick to notice the peer. It now probes every backed-off class and still replays only bounded bucket ops; promotion is a state flip the heavyweight tick acts on. - Backoffs are multiples of the tick interval, so a failure stamped δ seconds after a tick was 600 − δ old at the next tick and slipped a whole extra interval. The heavyweight drain now evaluates backoff halfway to its next tick. - An IAM entry first created by a non-deletion failure (the add bootstrap's snapshot send, the drain's own replay, an import-iam schedule) was never stamped `deletions_recorded`, so a later recorded deletion could not settle it and it escalated to the marker only `replicate repair` clears. Entries created by this binary now start recorded; a row persisted by an older binary keeps the escalation semantics. * fix(site-replication): reload peer node caches after bucket wiring writes Every S3 bucket-config write ends by asking the other nodes of the cluster to reload the bucket's metadata; the site-replication writers never did. On a multi-node site the node that ran the pairing (or applied a peer's bucket-meta item) rewrote the bucket targets and the derived replication rules on disk, while every other node kept serving its cached copy for up to the 15-minute refresh. A `resync start` routed to such a node reported every freshly wired bucket as `Config not found` and a bucket whose operator target the pairing had replaced as `recorded remote target no longer exists` (rustfs/backlog#2367 A-5, backlog#2195 item 2; functional SITE-105). Add one best-effort reload helper in the site-replication hooks and call it after the bucket setup, versioning, peer bucket-meta apply, removed-peer cleanup, make-with-versioning, and endpoint-refresh writes; the ensure helpers now report whether they wrote so unchanged passes stay silent. The resync manifest and start now read the persisted wiring instead of the node-local cache, matching the target read the start path already did. The new four-node e2e pairs two clusters and starts a resync through a non-coordinator node right after pairing; it also covers an IAM user created on a non-coordinator node converging to the peer site. * test(e2e): cover delete-marker replication from a multi-node source The functional suite reported delete markers created on a 3-node source never reaching the target (rustfs/backlog#2195 item 4, REP-105). The report was a probe defect, but the shape had no coverage: the existing delete-marker e2e runs a single-node source. Pin it against a four-node source replicating to a four-node peer and to a single-node target, with the write and the delete issued through different nodes. * ci(e2e): refresh the distributed selection for the new replication cases Four distributed cases were added (two site-replication, two delete-marker replication). The linux digest is derived from the last CI listing of the lane (34 cases, matching the previous pin) plus the four new names; the darwin digest is the local listing, which selects the same 38 cases.
2026-09-10 14:16:01 +00:00 · 2026-09-10 16:43:13 +08:00
parent 8f150d1d8e
commit aeaba86d73
12 changed files with 744 additions and 102 deletions
@@ -1,2 +1,2 @@
-sha256-linux=4696a43b167ac608b3b8677027c9fe9fdac3396d37c8cca11dce531c720ac6d2
-sha256-darwin=9785867929047dfd8c6f768e0d2b1e0a8fdba85216f4a4139093b1619d03ff07
+sha256-linux=4a399af73f3a954702967b80649958ae127f4e14560194e193de37e3eac5a17a
+sha256-darwin=4a399af73f3a954702967b80649958ae127f4e14560194e193de37e3eac5a17a
@@ -27,6 +27,7 @@ mod extra_test;
 mod harness;
 mod object_lock_test;
 mod observability_test;
+mod replication_delete_marker_test;
 mod replication_quota_test;
 mod s3_basic_test;
 mod s3_during_data_movement_test;
@@ -0,0 +1,137 @@
+// Copyright 2026 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Functional REP-105 (rustfs/backlog#2195 item 4): a delete marker created
+//! on a multi-node source cluster must replicate to the bucket-replication
+//! target. Objects converged in seconds while delete markers did not arrive
+//! within 180 s on the shared 3-node functional environment; the single-node
+//! e2e never saw it.
+
+use super::harness::{
+    DistCluster, DistLayout, TestResult, enable_versioning, put_bucket_replication, put_object, set_remote_target, unique_bucket,
+    wait_for_replicated_bytes, wait_until,
+};
+use crate::common::{FAST_DATA_USAGE_SCANNER_ENV, RustFSTestEnvironment, init_logging, replication_fast_env, signed_request};
+use crate::replication_extension_test::LOOPBACK_REPLICATION_TARGET_ENV;
+use aws_sdk_s3::Client;
+use http::{Method, StatusCode};
+use std::time::Duration;
+
+async fn target_has_delete_marker(client: &Client, bucket: &str, key: &str) -> TestResult<bool> {
+    let versions = client.list_object_versions().bucket(bucket).prefix(key).send().await?;
+    Ok(versions.delete_markers().iter().any(|marker| marker.key() == Some(key)))
+}
+
+async fn delete_marker_replicates(
+    source: &DistCluster,
+    source_bucket: &str,
+    target_client: &Client,
+    target_bucket: &str,
+) -> TestResult {
+    let key = "delete-marker/object.bin";
+    let body = b"delete marker replication payload".to_vec();
+    // Write through one node, delete through another: behind a load
+    // balancer consecutive requests land on different nodes.
+    put_object(&source.client(1)?, source_bucket, key, body.clone()).await?;
+    wait_for_replicated_bytes(target_client, target_bucket, key, &body, Duration::from_secs(60)).await?;
+
+    let delete = source
+        .client(2)?
+        .delete_object()
+        .bucket(source_bucket)
+        .key(key)
+        .send()
+        .await?;
+    assert_eq!(
+        delete.delete_marker(),
+        Some(true),
+        "a versioned DELETE without versionId must create a marker"
+    );
+
+    wait_until(
+        Duration::from_secs(90),
+        || async { target_has_delete_marker(target_client, target_bucket, key).await },
+        "delete marker replicated to the target bucket",
+    )
+    .await
+}
+
+#[tokio::test]
+async fn four_node_bucket_replication_replicates_delete_marker_to_peer_cluster() -> TestResult {
+    init_logging();
+    let (source, target) = DistCluster::start_replication_pair().await?;
+    let source_bucket = unique_bucket("dm-src");
+    let target_bucket = unique_bucket("dm-dst");
+    source.create_bucket(&source_bucket).await?;
+    target.create_bucket(&target_bucket).await?;
+    enable_versioning(&source.client(0)?, &source_bucket).await?;
+    enable_versioning(&target.client(0)?, &target_bucket).await?;
+
+    let arn = set_remote_target(&source.cluster, &source_bucket, &target.cluster, &target_bucket).await?;
+    put_bucket_replication(&source.cluster, &source_bucket, &arn).await?;
+
+    delete_marker_replicates(&source, &source_bucket, &target.client(0)?, &target_bucket).await
+}
+
+/// The functional environment replicates from a 3-node site to a single-node
+/// target; keep that shape as its own case.
+#[tokio::test]
+async fn four_node_bucket_replication_replicates_delete_marker_to_single_node_target() -> TestResult {
+    init_logging();
+    let mut extra: Vec<(&str, &str)> = replication_fast_env();
+    extra.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
+    extra.extend_from_slice(FAST_DATA_USAGE_SCANNER_ENV);
+    let source = DistCluster::start_with_env(DistLayout::FourNodeFourDisk, &extra).await?;
+    let mut target = RustFSTestEnvironment::new().await?;
+    target.start_rustfs_server_without_cleanup(vec![]).await?;
+
+    let source_bucket = unique_bucket("dm-src");
+    let target_bucket = unique_bucket("dm-dst");
+    source.create_bucket(&source_bucket).await?;
+    let target_client = target.create_s3_client();
+    target_client.create_bucket().bucket(&target_bucket).send().await?;
+    enable_versioning(&source.client(0)?, &source_bucket).await?;
+    enable_versioning(&target_client, &target_bucket).await?;
+
+    let body = serde_json::json!({
+        "endpoint": target.address,
+        "credentials": { "accessKey": target.access_key, "secretKey": target.secret_key },
+        "targetbucket": target_bucket,
+        "secure": false,
+        "type": "replication"
+    });
+    let url = format!(
+        "{}/rustfs/admin/v3/set-remote-target?bucket={}",
+        source.cluster.nodes[0].url,
+        urlencoding::encode(&source_bucket)
+    );
+    let response = signed_request(
+        Method::PUT,
+        &url,
+        &source.cluster.access_key,
+        &source.cluster.secret_key,
+        Some(body.to_string().into_bytes()),
+        Some("application/json"),
+    )
+    .await?;
+    if response.status() != StatusCode::OK {
+        let status = response.status();
+        let body = response.text().await.unwrap_or_default();
+        return Err(format!("set remote target failed: {status} {body}").into());
+    }
+    let arn: String = serde_json::from_slice(&response.bytes().await?)?;
+    put_bucket_replication(&source.cluster, &source_bucket, &arn).await?;
+
+    delete_marker_replicates(&source, &source_bucket, &target_client, &target_bucket).await
+}
@@ -126,3 +126,165 @@ async fn four_node_site_replication_replicates_object_to_peer_site() -> TestResu
    wait_for_replicated_bytes(&site_a.client(3)?, &bucket, reverse_key, &reverse_body, Duration::from_secs(60)).await?;
    Ok(())
 }
+
+async fn node_admin(
+    cluster: &crate::common::RustFSTestClusterEnvironment,
+    node_idx: usize,
+    method: Method,
+    path_and_query: &str,
+    body: Option<String>,
+) -> TestResult<(StatusCode, String)> {
+    crate::common::admin_request(
+        &cluster.nodes[node_idx].url,
+        method,
+        path_and_query,
+        body,
+        &cluster.access_key,
+        &cluster.secret_key,
+    )
+    .await
+}
+
+/// Pair two clusters through site A's first node and wait until both report
+/// the two-site topology as enabled.
+async fn pair_sites(site_a: &DistCluster, site_b: &DistCluster) -> TestResult {
+    let sites = vec![
+        PeerSite {
+            name: "site-a".to_string(),
+            endpoint: site_a.cluster.nodes[0].url.clone(),
+            access_key: site_a.cluster.access_key.clone(),
+            secret_key: site_a.cluster.secret_key.clone(),
+            ..Default::default()
+        },
+        PeerSite {
+            name: "site-b".to_string(),
+            endpoint: site_b.cluster.nodes[0].url.clone(),
+            access_key: site_b.cluster.access_key.clone(),
+            secret_key: site_b.cluster.secret_key.clone(),
+            ..Default::default()
+        },
+    ];
+    let add_status = site_replication_add(&site_a.cluster, &sites).await?;
+    assert!(
+        add_status.success && add_status.err_detail.is_empty() && add_status.initial_sync_error_message.is_empty(),
+        "site replication add reported failure: {add_status:?}"
+    );
+    wait_for_site_replication_enabled(&site_a.cluster).await?;
+    wait_for_site_replication_enabled(&site_b.cluster).await?;
+    Ok(())
+}
+
+async fn list_users_contains(
+    cluster: &crate::common::RustFSTestClusterEnvironment,
+    node_idx: usize,
+    access_key: &str,
+) -> TestResult<bool> {
+    let (status, body) = node_admin(cluster, node_idx, Method::GET, "/rustfs/admin/v3/list-users", None).await?;
+    if !status.is_success() {
+        return Err(format!("list-users on node {node_idx} failed: {status} {body}").into());
+    }
+    let users: serde_json::Value = serde_json::from_str(&body)?;
+    Ok(users.get(access_key).is_some())
+}
+
+/// backlog#2367 A-7 / functional SITE-102: an IAM change handled by a node
+/// other than the one that ran `site-replication/add` must still reach the
+/// peer site. Behind a load balancer every admin call may land on a
+/// different node, so the coordinator node is not special.
+#[tokio::test]
+async fn four_node_site_replication_converges_iam_user_created_on_a_non_coordinator_node() -> TestResult {
+    init_logging();
+    let (site_a, site_b) = DistCluster::start_replication_pair().await?;
+    pair_sites(&site_a, &site_b).await?;
+
+    let user = format!("siteuser-{}", &uuid::Uuid::new_v4().simple().to_string()[..8]);
+    let body = serde_json::json!({ "secretKey": "siteuser-secret-key-1234", "status": "enabled" }).to_string();
+    let (status, response) = node_admin(
+        &site_a.cluster,
+        1,
+        Method::PUT,
+        &format!("/rustfs/admin/v3/add-user?accessKey={user}"),
+        Some(body),
+    )
+    .await?;
+    assert!(status.is_success(), "add-user on site A node 1 failed: {status} {response}");
+
+    let site_b_cluster = &site_b.cluster;
+    let user_ref = user.as_str();
+    wait_until(
+        Duration::from_secs(90),
+        || async move { list_users_contains(site_b_cluster, 0, user_ref).await },
+        "user created on site A node 1 visible on site B",
+    )
+    .await?;
+    assert!(
+        list_users_contains(&site_a.cluster, 2, &user).await?,
+        "the user must be visible on every site A node"
+    );
+    Ok(())
+}
+
+/// backlog#2367 A-5 / functional SITE-105: a resync started right after
+/// pairing must not report buckets as failed. The bucket carrying an
+/// operator-configured bucket-replication target to the peer (the shape the
+/// functional suite leaves behind) and a plain versioned bucket are both
+/// wired by the pairing itself.
+#[tokio::test]
+async fn four_node_site_replication_resync_start_right_after_pairing_reports_no_failed_bucket() -> TestResult {
+    init_logging();
+    let (site_a, site_b) = DistCluster::start_replication_pair().await?;
+
+    let pre_src = unique_bucket("pre-src");
+    let pre_dst = unique_bucket("pre-dst");
+    let plain = unique_bucket("plain");
+    site_a.create_bucket(&pre_src).await?;
+    site_b.create_bucket(&pre_dst).await?;
+    site_a.create_bucket(&plain).await?;
+    enable_versioning(&site_a.client(0)?, &pre_src).await?;
+    enable_versioning(&site_b.client(0)?, &pre_dst).await?;
+    enable_versioning(&site_a.client(0)?, &plain).await?;
+    let arn = super::harness::set_remote_target(&site_a.cluster, &pre_src, &site_b.cluster, &pre_dst).await?;
+    super::harness::put_bucket_replication(&site_a.cluster, &pre_src, &arn).await?;
+
+    pair_sites(&site_a, &site_b).await?;
+
+    let (status, info) = node_admin(&site_a.cluster, 1, Method::GET, "/rustfs/admin/v3/site-replication/info", None).await?;
+    assert!(status.is_success(), "site-replication/info failed: {status} {info}");
+    let info: serde_json::Value = serde_json::from_str(&info)?;
+    let peer = info["sites"]
+        .as_array()
+        .and_then(|sites| sites.iter().find(|site| site["name"] == "site-b"))
+        .cloned()
+        .ok_or_else(|| format!("site-b peer missing from info: {info}"))?;
+
+    // Through a non-coordinator node, like a load-balanced admin call.
+    let (status, response) = node_admin(
+        &site_a.cluster,
+        1,
+        Method::PUT,
+        "/rustfs/admin/v3/site-replication/resync/op?operation=start",
+        Some(peer.to_string()),
+    )
+    .await?;
+    assert!(status.is_success(), "resync start failed: {status} {response}");
+    let resync: rustfs_madmin::SRResyncOpStatus = serde_json::from_str(&response)?;
+    let failed: Vec<String> = resync
+        .buckets
+        .iter()
+        .filter(|bucket| bucket.status == "failed")
+        .map(|bucket| format!("{}: {}", bucket.bucket, bucket.err_detail))
+        .collect();
+    assert!(
+        failed.is_empty(),
+        "resync right after pairing reported failed buckets: {failed:?} (status={}, detail={})",
+        resync.status,
+        resync.err_detail
+    );
+    assert!(
+        resync.buckets.iter().any(|bucket| bucket.bucket == pre_src)
+            && resync.buckets.iter().any(|bucket| bucket.bucket == plain),
+        "both buckets must be part of the resync: {:?}",
+        resync.buckets
+    );
+    Ok(())
+}
@@ -730,12 +730,16 @@ impl ReplicationConfigurationExt for ReplicationConfiguration {
            }
        }

+        // Highest priority first, like MinIO's `FilterActionableRules`. The
+        // tie-breakers make this a total order: a comparator that only
+        // orders same-destination pairs is not transitive, and the standard
+        // library sort panics on such inputs past its insertion-sort
+        // threshold (backlog#2367 C-1).
        rules.sort_by(|a, b| {
-            if a.destination == b.destination {
-                b.priority.cmp(&a.priority)
-            } else {
-                std::cmp::Ordering::Equal
-            }
+            b.priority
+                .cmp(&a.priority)
+                .then_with(|| a.destination.bucket.cmp(&b.destination.bucket))
+                .then_with(|| a.id.cmp(&b.id))
        });

        rules
@@ -813,24 +817,19 @@ impl ReplicationConfigurationExt for ReplicationConfiguration {
            return vec![role.to_string()];
        }

-        let mut arns = Vec::new();
-        let mut targets_map: HashSet<String> = HashSet::new();
-        let rules = self.filter_actionable_rules(obj);
-
-        for rule in rules {
+        // Rule order (priority descending) is the ARN order: callers that
+        // iterate targets see the highest-priority destination first.
+        let mut arns: Vec<String> = Vec::new();
+        for rule in self.filter_actionable_rules(obj) {
            if rule.status == ReplicationRuleStatus::from_static(ReplicationRuleStatus::DISABLED) {
                continue;
            }

            let arn = rule.destination.bucket.trim();
-            if !arn.is_empty() && !targets_map.contains(arn) {
-                targets_map.insert(arn.to_string());
+            if !arn.is_empty() && !arns.iter().any(|seen| seen == arn) {
+                arns.push(arn.to_string());
            }
        }
-
-        for arn in targets_map {
-            arns.push(arn);
-        }
        arns
    }

@@ -1908,6 +1907,84 @@ mod tests {
        assert_eq!(decisions, vec![(target_a.to_string(), false), (target_b.to_string(), true)]);
    }

+    // backlog#2367 C-1: the actionable-rule sort must be a total order. A
+    // comparator that answers `Equal` for different destinations but orders
+    // same-destination rules by priority is not transitive, and the standard
+    // library sort panics on such inputs once the slice is past the
+    // insertion-sort threshold (> 20 rules).
+    #[test]
+    fn actionable_rule_sort_is_a_total_order_across_destinations() {
+        let targets = ["arn:target:a", "arn:target:b", "arn:target:c"];
+        let mut seed: u64 = 0x2367;
+        for _ in 0..200 {
+            let rule_count = 21 + (seed % 200) as usize;
+            let rules = (0..rule_count)
+                .map(|index| {
+                    seed = seed.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407);
+                    let target = targets[(seed >> 33) as usize % targets.len()];
+                    delete_marker_rule(&format!("r{index}"), target, "", index as i32, true)
+                })
+                .collect();
+            let config = ReplicationConfiguration {
+                role: String::new(),
+                rules,
+            };
+            let ordered = config.filter_actionable_rules(&ObjectOpts {
+                name: "logs/app.log".to_string(),
+                op_type: ReplicationType::Object,
+                ..Default::default()
+            });
+            assert_eq!(ordered.len(), rule_count);
+            assert!(
+                ordered.windows(2).all(|pair| pair[0].priority >= pair[1].priority),
+                "actionable rules must be ordered by descending priority"
+            );
+        }
+    }
+
+    // backlog#2367 C-2: a V1 rule carries its prefix at the top level (no
+    // <Filter>). Ignoring it made `<Prefix>logs/</Prefix>` match every object.
+    #[test]
+    fn top_level_rule_prefix_scopes_matching_without_a_filter() {
+        let arn = "arn:target:a";
+        let config = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![delete_marker_rule("v1-prefix", arn, "logs/", 1, true)],
+        };
+        assert_eq!(config.rules[0].prefix(), "logs/");
+
+        let matching = config.filter_actionable_rules(&ObjectOpts {
+            name: "logs/app.log".to_string(),
+            op_type: ReplicationType::Object,
+            ..Default::default()
+        });
+        assert_eq!(matching.len(), 1);
+
+        let outside = config.filter_actionable_rules(&ObjectOpts {
+            name: "data/app.log".to_string(),
+            op_type: ReplicationType::Object,
+            ..Default::default()
+        });
+        assert!(outside.is_empty(), "an object outside the V1 prefix must not match: {outside:?}");
+        assert!(
+            config
+                .filter_target_arns(&ObjectOpts {
+                    name: "data/app.log".to_string(),
+                    op_type: ReplicationType::Object,
+                    ..Default::default()
+                })
+                .is_empty()
+        );
+
+        // A <Filter> still wins over the deprecated top-level element.
+        let mut filtered = delete_marker_rule("filtered", arn, "logs/", 1, true);
+        filtered.filter = Some(s3s::dto::ReplicationRuleFilter {
+            prefix: Some("photos/".to_string()),
+            ..Default::default()
+        });
+        assert_eq!(filtered.prefix(), "photos/");
+    }
+
    #[test]
    fn force_delete_targets_use_overlapping_rules_and_highest_priority_switch() {
        let target_a = "arn:target:a";
@@ -22,6 +22,10 @@ pub trait ReplicationRuleExt {
 }

 impl ReplicationRuleExt for ReplicationRule {
+    /// The rule's key prefix: `Filter.Prefix`, else `Filter.And.Prefix`, else
+    /// the deprecated top-level `Prefix` of a V1 rule written without a
+    /// `<Filter>` (backlog#2367 C-2). A rule that carries both keeps AWS's
+    /// precedence: the `<Filter>` is authoritative.
    fn prefix(&self) -> &str {
        if let Some(filter) = &self.filter {
            if let Some(prefix) = &filter.prefix {
@@ -32,7 +36,7 @@ impl ReplicationRuleExt for ReplicationRule {
                ""
            }
        } else {
-            ""
+            self.prefix.as_deref().unwrap_or("")
        }
    }

@@ -3853,6 +3853,30 @@ fn pending_remote_peer_ids(peers: &BTreeMap<String, PeerInfo>, local_peer: &Peer
        .collect()
 }

+/// The peers a pending remove / rotation still has to notify: every remote
+/// peer that has not acked, with the local site excluded by the same
+/// deployment-id-or-endpoint identity [`pending_remote_peer_ids`] finalizes
+/// on. The tick-driven `local_peer` carries the node's own listen address
+/// rather than the registered site endpoint (and a handler's carries the
+/// request `Host`, which behind a load balancer differs too), so an
+/// endpoint-only check dialed the site itself, timed out against the
+/// lifecycle lock this very request holds, and reported the operation as
+/// `Partial` (backlog#2367 A-4).
+fn pending_peers_awaiting_notification<'a>(
+    peers: &'a BTreeMap<String, PeerInfo>,
+    local_peer: &PeerInfo,
+    acked_deployment_ids: &BTreeSet<String>,
+) -> Vec<&'a PeerInfo> {
+    peers
+        .values()
+        .filter(|peer| {
+            peer.deployment_id != local_peer.deployment_id
+                && !same_identity_endpoint(&peer.endpoint, &local_peer.endpoint)
+                && !acked_deployment_ids.contains(&peer.deployment_id)
+        })
+        .collect()
+}
+
 fn pending_all_remote_peers_acked(
    peers: &BTreeMap<String, PeerInfo>,
    local_peer: &PeerInfo,
@@ -4068,12 +4092,7 @@ async fn drive_pending_rotation(pending: &PendingRotation, local_peer: &PeerInfo
    };

    let mut peer_errors = Vec::new();
-    for peer in pending.peers.values() {
-        if same_identity_endpoint(&peer.endpoint, &local_peer.endpoint)
-            || pending.acked_deployment_ids.contains(&peer.deployment_id)
-        {
-            continue;
-        }
+    for peer in pending_peers_awaiting_notification(&pending.peers, local_peer, &pending.acked_deployment_ids) {
        // A superseded join returns BEFORE `apply_iam`, so a no-op answer
        // means the peer never installed the new secret. Acking it would
        // finalize a rotation half the mesh cannot authenticate against
@@ -4376,12 +4395,9 @@ async fn drive_pending_remove(pending_remove: &PendingRemove, local_peer: &PeerI
    if secret_candidates.is_empty() {
        peer_errors.push("site replication service account secret unavailable".to_string());
    } else {
-        for peer in pending_remove.original_peers.values() {
-            if same_identity_endpoint(&peer.endpoint, &local_peer.endpoint)
-                || pending_remove.acked_deployment_ids.contains(&peer.deployment_id)
-            {
-                continue;
-            }
+        for peer in
+            pending_peers_awaiting_notification(&pending_remove.original_peers, local_peer, &pending_remove.acked_deployment_ids)
+        {
            if let Err(err) = PeerAdminRequest::put(
                &runtime_peer_connection(peer)?,
                SITE_REPLICATION_PEER_REMOVE_PATH,
@@ -4931,7 +4947,7 @@ async fn ensure_site_replication_bucket_targets(bucket: &str) -> S3Result<()> {
        return Ok(());
    };
    let config = bucket_replication_config_for_target_refresh(bucket).await?;
-    ensure_site_replication_bucket_targets_with_runtime(
+    let written = ensure_site_replication_bucket_targets_with_runtime(
        bucket,
        &runtime.state,
        &runtime.local_peer,
@@ -4939,7 +4955,11 @@ async fn ensure_site_replication_bucket_targets(bucket: &str) -> S3Result<()> {
        &runtime.service_account_secret_key,
        expected_incarnation_id,
    )
-    .await
+    .await?;
+    if written {
+        reload_bucket_metadata_on_peers(bucket, "site_replication_bucket_targets", false).await;
+    }
+    Ok(())
 }

 async fn ensure_site_replication_bucket_setup(bucket: &str) -> S3Result<bool> {
@@ -5009,6 +5029,9 @@ async fn cleanup_removed_site_replication_bucket(bucket: &str, removed_deploymen
        Err(err) => return Err(ApiError::from(err).into()),
    }

+    if removed > 0 {
+        reload_bucket_metadata_on_peers(bucket, "site_replication_bucket_cleanup", true).await;
+    }
    Ok(removed)
 }

@@ -5325,7 +5348,7 @@ async fn refresh_bucket_targets_after_endpoint_edit(pending_id: &str, service_ac
        let local_peer = current_local_runtime_peer(&target_state);
        let _targets_guard = lock_bucket_targets_metadata(&bucket.name).await;
        let replication_config = bucket_replication_config_for_target_refresh(&bucket.name).await?;
-        ensure_site_replication_bucket_targets_with_runtime(
+        let written = ensure_site_replication_bucket_targets_with_runtime(
            &bucket.name,
            &target_state,
            &local_peer,
@@ -5334,6 +5357,9 @@ async fn refresh_bucket_targets_after_endpoint_edit(pending_id: &str, service_ac
            expected_incarnation_id,
        )
        .await?;
+        if written {
+            reload_bucket_metadata_on_peers(&bucket.name, "site_replication_endpoint_refresh", false).await;
+        }

        rewritten.push(bucket.name.clone());

@@ -5377,16 +5403,12 @@ async fn site_bucket_resync_manifest_entry(bucket: &str, peer: &PeerInfo, now: O
        ..Default::default()
    };
    let _targets_guard = lock_bucket_targets_metadata(bucket).await;
-    let (config, _) = match metadata_sys::get_replication_config(bucket).await {
-        Ok(config) => config,
-        Err(err) => {
-            entry.status = "failed".to_string();
-            entry.err_detail = summarize_peer_error_detail(&err.to_string());
-            return entry;
-        }
-    };
-    let targets = match metadata_sys::list_bucket_targets(bucket).await {
-        Ok(targets) => targets,
+    // Read what is persisted, not this node's cache: the wiring may have
+    // been written by another node moments ago (`start_site_bucket_resync`
+    // already reads its targets from disk), and an operator resync must see
+    // the same records the drive will use.
+    let (config, targets) = match site_bucket_resync_persisted_wiring(bucket).await {
+        Ok(wiring) => wiring,
        Err(err) => {
            entry.status = "failed".to_string();
            entry.err_detail = summarize_peer_error_detail(&err.to_string());
@@ -5417,6 +5439,15 @@ async fn site_bucket_resync_manifest_entry(bucket: &str, peer: &PeerInfo, now: O
    entry
 }

+/// The persisted replication configuration and bucket targets, bypassing the
+/// node-local metadata cache. `ConfigNotFound` surfaces for a bucket without
+/// a replication configuration, matching the cached read's error.
+async fn site_bucket_resync_persisted_wiring(bucket: &str) -> Result<(ReplicationConfiguration, BucketTargets), StorageError> {
+    let metadata = metadata_sys::get_config_from_disk(bucket).await?;
+    let config = metadata.replication_config.ok_or(StorageError::ConfigNotFound)?;
+    Ok((config, metadata.bucket_target_config.unwrap_or_default()))
+}
+
 async fn start_site_bucket_resync(bucket: &str, target_arn: &str, resync_id: &str) -> ResyncBucketStatus {
    let mut bucket_status = ResyncBucketStatus {
        bucket: bucket.to_string(),
@@ -5439,17 +5470,8 @@ async fn start_site_bucket_resync(bucket: &str, target_arn: &str, resync_id: &st
        }
    };

-    let (config, _) = match metadata_sys::get_replication_config(bucket).await {
-        Ok(config) => config,
-        Err(err) => {
-            bucket_status.status = "failed".to_string();
-            bucket_status.err_detail = err.to_string();
-            return bucket_status;
-        }
-    };
-
-    let targets = match metadata_sys::list_bucket_targets_from_disk(bucket).await {
-        Ok(targets) => targets,
+    let (config, targets) = match site_bucket_resync_persisted_wiring(bucket).await {
+        Ok(wiring) => wiring,
        Err(err) => {
            bucket_status.status = "failed".to_string();
            bucket_status.err_detail = err.to_string();
@@ -6046,6 +6068,10 @@ async fn apply_bucket_meta_item(item: SRBucketMeta) -> S3Result<()> {
    drop(lifecycle_guard);
    drop(targets_guard);

+    if !skip_config_write {
+        reload_bucket_metadata_on_peers(&item.bucket, "site_replication_bucket_meta", item.r#type == "lc-config").await;
+    }
+
    if item.r#type == "replication-config" {
        // Rebuild the local outbound rules too: a site that joined an already-replicated
        // bucket receives this item before it has any `site-repl-*` rule of its own.
@@ -7432,6 +7458,7 @@ impl Operation for SRPeerBucketOpsHandler {
                )
                .await
                .map_err(ApiError::from)?;
+                reload_bucket_metadata_on_peers(&bucket, "site_replication_make_bucket", false).await;
            }
            "configure-replication" => {
                store
@@ -15434,4 +15461,54 @@ mod tests {
            "no replicated config write may bypass the source stamp"
        );
    }
+
+    /// backlog#2367 A-4: `remove --all` notified "the peer" at the site's own
+    /// registered endpoint. The tick-driven local peer carries the node's
+    /// listen address, so an endpoint-only self check let the loop dial the
+    /// site itself and report `Partial: failed to notify 1 peer(s)`.
+    #[test]
+    fn pending_notifications_skip_the_local_site_by_deployment_id() {
+        let local_registered = PeerInfo {
+            deployment_id: "site-b".to_string(),
+            ..peer("site-b", "http://site-b.example.com:9000")
+        };
+        let remote = PeerInfo {
+            deployment_id: "site-a".to_string(),
+            ..peer("site-a", "http://site-a.example.com:9000")
+        };
+        let acked = PeerInfo {
+            deployment_id: "site-c".to_string(),
+            ..peer("site-c", "http://site-c.example.com:9000")
+        };
+        let peers = BTreeMap::from([
+            (local_registered.deployment_id.clone(), local_registered.clone()),
+            (remote.deployment_id.clone(), remote),
+            (acked.deployment_id.clone(), acked.clone()),
+        ]);
+        let acked_ids = BTreeSet::from([acked.deployment_id]);
+
+        // The tick resolves the local peer from its own listen address.
+        let local_from_tick = PeerInfo {
+            deployment_id: "site-b".to_string(),
+            ..peer("site-b", "http://127.0.0.1:9000")
+        };
+        let to_notify: Vec<&str> = pending_peers_awaiting_notification(&peers, &local_from_tick, &acked_ids)
+            .iter()
+            .map(|peer| peer.deployment_id.as_str())
+            .collect();
+        assert_eq!(to_notify, vec!["site-a"], "the local site and the acked peer are never dialed");
+
+        // Identity stays consistent with what finalization waits for.
+        assert_eq!(
+            pending_remote_peer_ids(&peers, &local_from_tick),
+            BTreeSet::from(["site-a".to_string(), "site-c".to_string()])
+        );
+
+        // A handler-resolved local peer (registered endpoint) agrees.
+        let to_notify: Vec<&str> = pending_peers_awaiting_notification(&peers, &local_registered, &acked_ids)
+            .iter()
+            .map(|peer| peer.deployment_id.as_str())
+            .collect();
+        assert_eq!(to_notify, vec!["site-a"]);
+    }
 }
@@ -73,6 +73,7 @@ const SITE_REPLICATION_RESYNC_ROUTE: &str = "/rustfs/admin/v3/site-replication/r
 const SITE_REPLICATION_REPAIR_ROUTE: &str = "/rustfs/admin/v3/site-replication/repair";
 const SITE_REPLICATION_REPAIR_STATUS_ROUTE: &str = "/rustfs/admin/v3/site-replication/repair/status";
 const IAM_POLICY_ATTACH_ROUTE: &str = "/rustfs/admin/v3/idp/builtin/policy/attach";
+const DATA_USAGE_INFO_ROUTE: &str = "/rustfs/admin/v3/datausageinfo";
 const IAM_POLICY_DETACH_ROUTE: &str = "/rustfs/admin/v3/idp/builtin/policy/detach";
 const IAM_POLICY_ENTITIES_ROUTE: &str = "/rustfs/admin/v3/idp/builtin/policy-entities";
 const IAM_ACCESS_KEYS_BULK_ROUTE: &str = "/rustfs/admin/v3/list-access-keys-bulk";
@@ -1077,12 +1078,25 @@ fn advertised_admin_capabilities() -> Vec<AdvertisedAdminCapability> {
        ("admin.account.mfa", HttpMethod::Get, ACCOUNT_MFA_ROUTE),
        ("admin.mfa.challenge", HttpMethod::Get, MFA_CHALLENGE_ROUTE),
        ("admin.user.mfa", HttpMethod::Get, USER_MFA_ROUTE),
+        // `rc du` is gated on this name. Before it was advertised the client
+        // inferred it from a `1.0.0-rc.` version prefix, which no longer
+        // matches once the server reports `1.0.0` (backlog#2367 E-2).
+        ("admin.data-usage", HttpMethod::Get, DATA_USAGE_INFO_ROUTE),
    ]
    .into_iter()
    .map(|(name, method, route)| AdvertisedAdminCapability {
        name,
        status: admin_route_capability(method, route),
    })
+    .chain(std::iter::once(AdvertisedAdminCapability {
+        // `rc watch` streams `GET /{bucket}?events=`, a misc extension route
+        // dispatched by `admin::router` rather than an admin policy route,
+        // so its status is not an inventory lookup (same version-prefix
+        // inference on the client as `admin.data-usage`).
+        name: "listen_notification",
+        status: CapabilityStatus::supported()
+            .with_reason("bucket listen notification (?events=) is dispatched by the admin router"),
+    }))
    .collect()
 }

@@ -1258,6 +1272,9 @@ mod tests {
            "admin.iam.access-keys-bulk",
            "admin.iam.access-keys-bulk.ldap",
            "admin.iam.access-keys-bulk.openid",
+            // rc pinned these two by version prefix until 1.0.0 (backlog#2367 E-2).
+            "admin.data-usage",
+            "listen_notification",
        ];
        for name in expected_supported {
            let entry = response
@@ -1630,6 +1630,44 @@ pub(crate) fn build_site_replication_config(
    }
 }

+/// Reload `bucket`'s metadata on every other node of this site after a
+/// site-replication write. Every S3 bucket-config write does this
+/// (`app::bucket_usecase::notify_bucket_metadata_reload`); the
+/// site-replication writers did not, so on a multi-node site a node other
+/// than the one that applied the write served the previous targets and
+/// rules for up to the 15-minute refresh — a `resync start` routed to such a
+/// node reported every freshly wired bucket as `Config not found` or
+/// `recorded remote target no longer exists` (backlog#2367 A-5, backlog#2195
+/// item 2). Best effort like the S3 path: the write is durable and the
+/// refresh loop is the fallback, so an unreachable node must not fail the
+/// operation that already committed.
+pub(crate) async fn reload_bucket_metadata_on_peers(bucket: &str, operation: &'static str, scanner_maintenance_change: bool) {
+    if scanner_maintenance_change {
+        rustfs_scanner::record_scanner_maintenance_change(bucket);
+    }
+    let Some(notification_sys) = crate::admin::runtime_sources::current_notification_system() else {
+        return;
+    };
+    let result = if scanner_maintenance_change {
+        notification_sys.load_bucket_metadata_for_scanner_maintenance(bucket).await
+    } else {
+        notification_sys.load_bucket_metadata(bucket).await
+    };
+    if let Err(err) = result {
+        warn!(
+            event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+            component = LOG_COMPONENT_ADMIN,
+            subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+            bucket = %bucket,
+            operation,
+            result = "peer_metadata_reload_failed",
+            error = %err,
+            "admin site replication state"
+        );
+    }
+}
+
+/// Returns whether the bucket targets were rewritten.
 pub(crate) async fn ensure_site_replication_bucket_targets_with_runtime(
    bucket: &str,
    state: &SiteReplicationState,
@@ -1637,7 +1675,7 @@ pub(crate) async fn ensure_site_replication_bucket_targets_with_runtime(
    config: Option<&ReplicationConfiguration>,
    service_account_secret_key: &str,
    expected_incarnation_id: Uuid,
-) -> S3Result<()> {
+) -> S3Result<bool> {
    let existing = match metadata_sys::list_bucket_targets(bucket).await {
        Ok(targets) => targets,
        Err(StorageError::ConfigNotFound) => BucketTargets::default(),
@@ -1649,7 +1687,7 @@ pub(crate) async fn ensure_site_replication_bucket_targets_with_runtime(
    let updated =
        reconcile_site_replication_bucket_targets(existing, bucket, state, local_peer, config, service_account_secret_key)?;
    if updated.targets.is_empty() {
-        return Ok(());
+        return Ok(false);
    }

    let json_targets = serde_json::to_vec(&updated)
@@ -1658,12 +1696,12 @@ pub(crate) async fn ensure_site_replication_bucket_targets_with_runtime(
    // client — noticeable now that startup reconciles all buckets, not just the one bucket
    // an operation touched.
    if json_targets == existing_json {
-        return Ok(());
+        return Ok(false);
    }
    metadata_sys::update_if_incarnation(bucket, BUCKET_TARGETS_FILE, json_targets, expected_incarnation_id)
        .await
        .map_err(ApiError::from)?;
-    Ok(())
+    Ok(true)
 }

 pub(crate) async fn bucket_replication_config_for_target_refresh(bucket: &str) -> S3Result<Option<ReplicationConfiguration>> {
@@ -1674,13 +1712,14 @@ pub(crate) async fn bucket_replication_config_for_target_refresh(bucket: &str) -
    }
 }

+/// Returns whether the replication configuration was rewritten.
 pub(crate) async fn ensure_site_replication_bucket_replication_config_with_runtime(
    bucket: &str,
    state: &SiteReplicationState,
    local_peer: &PeerInfo,
    service_account_secret_key: &str,
    expected_incarnation_id: Uuid,
-) -> S3Result<()> {
+) -> S3Result<bool> {
    let existing = match metadata_sys::get_replication_config(bucket).await {
        Ok((existing, _)) => Some(existing),
        Err(StorageError::ConfigNotFound) => None,
@@ -1689,7 +1728,7 @@ pub(crate) async fn ensure_site_replication_bucket_replication_config_with_runti

    let Some(desired) = build_site_replication_config(bucket, state, local_peer, service_account_secret_key, existing.as_ref())?
    else {
-        return Ok(());
+        return Ok(false);
    };

    // Derived rules are state owned by this site: rebuild them from the current peer
@@ -1721,7 +1760,7 @@ pub(crate) async fn ensure_site_replication_bucket_replication_config_with_runti
    };

    if rules == existing_rules && role == existing_role {
-        return Ok(());
+        return Ok(false);
    }

    let data = serialize(&ReplicationConfiguration { role, rules })
@@ -1730,7 +1769,7 @@ pub(crate) async fn ensure_site_replication_bucket_replication_config_with_runti
        .await
        .map_err(ApiError::from)?;

-    Ok(())
+    Ok(true)
 }

 pub(crate) async fn ensure_site_replication_bucket_setup_with_runtime(
@@ -1748,9 +1787,9 @@ pub(crate) async fn ensure_site_replication_bucket_setup_with_runtime_for_incarn
    runtime: &SiteReplicationRuntime,
    expected_incarnation_id: Uuid,
 ) -> S3Result<()> {
-    let _targets_guard = lock_bucket_targets_metadata(bucket).await;
+    let targets_guard = lock_bucket_targets_metadata(bucket).await;
    let config = bucket_replication_config_for_target_refresh(bucket).await?;
-    ensure_site_replication_bucket_targets_with_runtime(
+    let targets_written = ensure_site_replication_bucket_targets_with_runtime(
        bucket,
        &runtime.state,
        &runtime.local_peer,
@@ -1759,7 +1798,7 @@ pub(crate) async fn ensure_site_replication_bucket_setup_with_runtime_for_incarn
        expected_incarnation_id,
    )
    .await?;
-    ensure_site_replication_bucket_replication_config_with_runtime(
+    let config_written = ensure_site_replication_bucket_replication_config_with_runtime(
        bucket,
        &runtime.state,
        &runtime.local_peer,
@@ -1767,6 +1806,10 @@ pub(crate) async fn ensure_site_replication_bucket_setup_with_runtime_for_incarn
        expected_incarnation_id,
    )
    .await?;
+    drop(targets_guard);
+    if targets_written || config_written {
+        reload_bucket_metadata_on_peers(bucket, "site_replication_bucket_setup", config_written).await;
+    }
    Ok(())
 }

@@ -1791,6 +1834,7 @@ pub(crate) async fn ensure_site_replication_bucket_versioning(bucket: &str) -> S
    metadata_sys::update_if_incarnation(bucket, BUCKET_VERSIONING_CONFIG, bucket_versioning_xml()?, expected_incarnation_id)
        .await
        .map_err(ApiError::from)?;
+    reload_bucket_metadata_on_peers(bucket, "site_replication_bucket_versioning", false).await;

    Ok(())
 }
@@ -52,7 +52,11 @@ pub(crate) struct SiteReplicationRetryEvent {
    /// deletion body (if it was a deletion) recorded in
    /// [`SiteReplicationState::iam_deletion_replays`]. Only then may a
    /// successful deletion replay plus a stable snapshot resend settle the
-    /// entry; a legacy entry (or one degraded by record overflow) keeps the
+    /// entry. Every entry this binary creates starts recorded: the IAM
+    /// change hook records deletion bodies, and the other creators (the add
+    /// bootstrap's snapshot send, the drain's own replay) never carry a
+    /// deletion. A legacy entry persisted by a binary that predates recording
+    /// (serde default `false`), or one degraded by record overflow, keeps the
    /// escalation semantics because an unrecorded deletion may hide in it.
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
    pub(crate) deletions_recorded: bool,
@@ -308,7 +312,12 @@ fn push_site_replication_retry_event(
        updated_at: Some(OffsetDateTime::now_utc()),
        edit_generation: generation,
        peer_unreachable,
-        deletions_recorded: false,
+        // See the field doc: only a row persisted by an older binary is
+        // unrecorded. Stamping at creation is what lets an entry first
+        // created by the bootstrap snapshot send settle after a later
+        // deletion is replayed, instead of escalating forever
+        // (backlog#2367 A-3).
+        deletions_recorded: true,
    });
    Ok(evicted)
 }
@@ -598,22 +607,7 @@ pub(crate) fn record_failed_iam_delivery(
    item: &SRIAMItem,
    error: &str,
 ) -> S3Result<()> {
-    let existed = state
-        .retry_queue
-        .iter()
-        .any(|event| retry_event_matches(event, peer, SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH));
    upsert_site_replication_retry_event(&mut state.retry_queue, peer, SITE_REPLICATION_PEER_IAM_ITEM_WIRE_PATH, error, None)?;
-    if !existed
-        && let Some(event) = state
-            .retry_queue
-            .iter_mut()
-            .find(|event| retry_event_matches(event, peer, SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH))
-    {
-        // Fresh entry: every failure it will ever collapse goes through this
-        // recording path, so a deletion replay plus a stable snapshot resend
-        // can later settle it instead of escalating.
-        event.deletions_recorded = true;
-    }

    let Some(entity) = iam_item_deletion_entity(item) else {
        return Ok(());
@@ -1418,6 +1412,33 @@ pub(crate) fn site_replication_retry_backoff_elapsed(event: &SiteReplicationRetr
    now.unix_timestamp().saturating_sub(updated_at.unix_timestamp()) >= delay
 }

+/// Backoff evaluation time for the heavyweight tick: halfway to the next
+/// tick. Backoffs are multiples of the tick interval, so an entry stamped δ
+/// seconds after a tick is `600 − δ` old at the next one and slipped a whole
+/// extra interval for every δ > 0 — a first replay landed at T+1200 rather
+/// than T+600 (backlog#2367 A-1). Evaluating at the midpoint bounds the slip
+/// to half an interval either way; timestamps written back stay real time.
+pub(crate) fn heavyweight_retry_drain_horizon(now: OffsetDateTime) -> OffsetDateTime {
+    let half_interval = crate::site_replication_reconcile::RECONCILE_INTERVAL / 2;
+    now + time::Duration::seconds(i64::try_from(half_interval.as_secs()).unwrap_or(i64::MAX))
+}
+
+/// What the lightweight 30-second pass may act on. It replays bounded bucket
+/// ops only, but probes every backed-off class: promotion is a state flip
+/// the heavyweight tick then replays, so an IAM or bucket-metadata snapshot
+/// owed to a peer that came back is resent at the next tick instead of
+/// after its own backoff has fully elapsed (backlog#2367 A-1).
+pub(crate) fn lightweight_retry_drain_partition(
+    state: &SiteReplicationState,
+    now: OffsetDateTime,
+) -> (Vec<SiteReplicationRetryEvent>, Vec<SiteReplicationRetryEvent>) {
+    let mut actionable = actionable_site_replication_retry_events(state, now);
+    actionable.retain(|event| {
+        classify_site_replication_retry_event(event).is_some_and(|action| is_lightweight_retry_drain_action(&action))
+    });
+    (actionable, deferred_site_replication_retry_events(state, now))
+}
+
 /// The subset of the retry queue the background drain is allowed to touch.
 pub(crate) fn actionable_site_replication_retry_events(
    state: &SiteReplicationState,
@@ -1666,14 +1687,7 @@ async fn drain_site_replication_retry_queue_lightweight_inner() -> S3Result<()>
        return Ok(());
    }
    let now = OffsetDateTime::now_utc();
-    let mut actionable = actionable_site_replication_retry_events(&runtime.state, now);
-    let mut deferred = deferred_site_replication_retry_events(&runtime.state, now);
-    actionable.retain(|event| {
-        classify_site_replication_retry_event(event).is_some_and(|action| is_lightweight_retry_drain_action(&action))
-    });
-    deferred.retain(|event| {
-        classify_site_replication_retry_event(event).is_some_and(|action| is_lightweight_retry_drain_action(&action))
-    });
+    let (actionable, deferred) = lightweight_retry_drain_partition(&runtime.state, now);
    if actionable.is_empty() && deferred.is_empty() {
        return Ok(());
    }
@@ -1697,10 +1711,7 @@ async fn drain_site_replication_retry_queue_lightweight_inner() -> S3Result<()>
            return Ok(());
        }
        let now = OffsetDateTime::now_utc();
-        let mut actionable = actionable_site_replication_retry_events(&runtime.state, now);
-        actionable.retain(|event| {
-            classify_site_replication_retry_event(event).is_some_and(|action| is_lightweight_retry_drain_action(&action))
-        });
+        let (actionable, _) = lightweight_retry_drain_partition(&runtime.state, now);
        if actionable.is_empty() {
            return Ok(());
        }
@@ -1717,9 +1728,9 @@ pub(crate) async fn drain_site_replication_retry_queue_inner() -> S3Result<()> {
    // The alert must fire even when nothing is drainable this tick —
    // escalated markers are exactly the entries the drain skips.
    log_site_replication_retry_liabilities(&runtime.state);
-    let now = OffsetDateTime::now_utc();
-    let actionable = actionable_site_replication_retry_events(&runtime.state, now);
-    let deferred = deferred_site_replication_retry_events(&runtime.state, now);
+    let horizon = heavyweight_retry_drain_horizon(OffsetDateTime::now_utc());
+    let actionable = actionable_site_replication_retry_events(&runtime.state, horizon);
+    let deferred = deferred_site_replication_retry_events(&runtime.state, horizon);
    if actionable.is_empty() && deferred.is_empty() {
        return Ok(());
    }
@@ -1764,8 +1775,8 @@ pub(crate) async fn drain_site_replication_retry_queue_inner() -> S3Result<()> {
        {
            return Ok(());
        }
-        let now = OffsetDateTime::now_utc();
-        let actionable = actionable_site_replication_retry_events(&runtime.state, now);
+        let horizon = heavyweight_retry_drain_horizon(OffsetDateTime::now_utc());
+        let actionable = actionable_site_replication_retry_events(&runtime.state, horizon);
        if actionable.is_empty() {
            return Ok(());
        }
@@ -845,7 +845,8 @@ fn test_record_failed_iam_delivery_records_deletions_and_flags_entry() {
    record_failed_iam_delivery(&mut state, &target, &policy_delete_item("readonly"), "peer offline").expect("record failure");
    assert_eq!(state.iam_deletion_replays.len(), 2);

-    // A legacy entry (created without recording) is never stamped.
+    // A legacy entry (persisted by a binary that predates recording, so it
+    // deserialized with the `false` default) is never stamped.
    let legacy = PeerInfo {
        deployment_id: "legacy-dep".to_string(),
        ..peer("legacy", "https://legacy.example.com")
@@ -859,6 +860,12 @@ fn test_record_failed_iam_delivery_records_deletions_and_flags_entry() {
        None,
    )
    .expect("upsert retry event");
+    state
+        .retry_queue
+        .iter_mut()
+        .find(|event| event.peer_deployment_id == legacy.deployment_id)
+        .expect("legacy entry")
+        .deletions_recorded = false;
    record_failed_iam_delivery(&mut state, &legacy, &user_delete_item("bob"), "peer offline").expect("record failure");
    let legacy_event = state
        .retry_queue
@@ -871,6 +878,43 @@ fn test_record_failed_iam_delivery_records_deletions_and_flags_entry() {
    );
 }

+/// backlog#2367 A-3: an entry first created by a non-deletion failure — the
+/// add bootstrap's snapshot send, or the drain's own replay — hides no
+/// unrecorded deletion, so a deletion recorded later plus a stable snapshot
+/// resend must settle it instead of escalating it to the permanent marker
+/// that only `replicate repair` clears.
+#[test]
+fn test_bootstrap_created_iam_entry_settles_after_deletion_replay() {
+    let target = PeerInfo {
+        deployment_id: "remote-dep".to_string(),
+        ..peer("remote", "https://remote.example.com")
+    };
+    let mut state = deletion_replay_state(&target);
+    upsert_site_replication_retry_event(
+        &mut state.retry_queue,
+        &target,
+        SITE_REPLICATION_PEER_IAM_ITEM_WIRE_PATH,
+        "peer request to https://remote.example.com failed (connect): connection refused",
+        None,
+    )
+    .expect("bootstrap send failure");
+    assert!(state.retry_queue[0].deletions_recorded, "a fresh entry carries no unrecorded deletion");
+
+    record_failed_iam_delivery(&mut state, &target, &user_delete_item("alice"), "peer offline").expect("record failure");
+    assert_eq!(state.retry_queue.len(), 1, "the hook failure collapses into the bootstrap entry");
+    assert!(state.retry_queue[0].deletions_recorded);
+    assert_eq!(state.iam_deletion_replays.len(), 1);
+
+    let observed = state.retry_queue[0].clone();
+    let replayed: Vec<String> = state.iam_deletion_replays.iter().map(|record| record.id.clone()).collect();
+    assert!(
+        settle_replayed_iam_retry_events(&mut state, &target, &observed, &replayed),
+        "the replayed deletion plus the snapshot resend settle the entry"
+    );
+    assert!(state.retry_queue.is_empty(), "no escalation marker may remain: {:?}", state.retry_queue);
+    assert!(state.iam_deletion_replays.is_empty());
+}
+
 /// Overflowing the per-peer record cap degrades the entry back to the
 /// escalation semantics: the record set is no longer complete, so a replay
 /// can no longer prove the peer converged.
@@ -1626,6 +1670,74 @@ fn test_deferred_retry_events_do_not_probe_fresh_application_failures() {
    assert!(actionable_site_replication_retry_events(&state, now).is_empty());
 }

+/// backlog#2367 A-1: the lightweight pass replays bucket ops only, but
+/// probes every backed-off class so a recovered peer's IAM snapshot is
+/// promoted within 30 seconds instead of waiting for the heavyweight tick
+/// to notice it.
+#[test]
+fn test_lightweight_partition_probes_snapshot_entries_but_replays_bucket_ops_only() {
+    let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp");
+    let mut state = SiteReplicationState::default();
+    state
+        .peers
+        .insert("remote".to_string(), peer("remote", "https://remote.example.com"));
+
+    let bucket_make = "/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=make-with-versioning";
+    let mut iam_unreachable = drain_event(
+        "remote",
+        SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH,
+        3,
+        Some(now - time::Duration::seconds(30)),
+    );
+    iam_unreachable.peer_unreachable = true;
+    let mut bucket_unreachable = drain_event("remote", bucket_make, 3, Some(now - time::Duration::seconds(30)));
+    bucket_unreachable.peer_unreachable = true;
+    state.retry_queue = vec![
+        iam_unreachable,
+        bucket_unreachable,
+        // Already promoted (or never stamped): due now.
+        drain_event("remote", SITE_REPLICATION_RETRY_BUCKET_METADATA_SNAPSHOT_PATH, 1, None),
+        drain_event("remote", bucket_make, 1, None),
+    ];
+
+    let (actionable, deferred) = lightweight_retry_drain_partition(&state, now);
+    let deferred_paths: Vec<&str> = deferred.iter().map(|event| event.path.as_str()).collect();
+    assert!(
+        deferred_paths.contains(&SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH),
+        "the backed-off IAM snapshot must be probed by the lightweight pass: {deferred_paths:?}"
+    );
+    assert!(deferred_paths.contains(&bucket_make));
+    assert_eq!(
+        actionable.iter().map(|event| event.path.as_str()).collect::<Vec<_>>(),
+        vec![bucket_make],
+        "only the bounded bucket op is replayed by the lightweight pass"
+    );
+}
+
+/// backlog#2367 A-1: the heavyweight tick evaluates backoff halfway to its
+/// next tick. A first failure stamped one second after a tick is 599 s old
+/// at the next tick; without the horizon it slipped to the tick after.
+#[test]
+fn test_heavyweight_horizon_absorbs_tick_phase() {
+    let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp");
+    let horizon = heavyweight_retry_drain_horizon(now);
+    assert_eq!(horizon - now, time::Duration::seconds(300));
+
+    let elapsed_at_horizon = |secs_ago: i64| {
+        site_replication_retry_backoff_elapsed(
+            &drain_event("remote", "/p", 1, Some(now - time::Duration::seconds(secs_ago))),
+            horizon,
+        )
+    };
+    // Stamped just after the previous tick: due at this tick, not the next.
+    assert!(elapsed_at_horizon(599));
+    // Due before the next tick's midpoint: drained now rather than a whole
+    // interval late.
+    assert!(elapsed_at_horizon(301));
+    // Due after the midpoint: waits for the next tick.
+    assert!(!elapsed_at_horizon(299));
+}
+
 /// 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)`
@@ -32,7 +32,7 @@ use tokio::time::Instant;
 use tokio_util::sync::CancellationToken;
 use tracing::warn;

-const RECONCILE_INTERVAL: Duration = Duration::from_secs(600);
+pub(crate) const RECONCILE_INTERVAL: Duration = Duration::from_secs(600);
 pub(crate) const RETRY_DRAIN_INTERVAL: Duration = Duration::from_secs(30);

 /// A reconciler reports its own failures; the outcome carries no value because neither