fix(replication): replay the MRF journal periodically (#9139 )

* fix(replication): replay the MRF journal periodically The persisted MRF journal was replayed only at startup. Delete-marker purges whose watch window drained with targets unreachable, and replications that overflowed the worker queues, reach the journal while the server runs, so they waited for a restart to converge. Move the startup pass into replay_mrf_journal and rerun it every five minutes from a background loop. The existing recovery leader lock keeps one pass at a time across the cluster, including the startup pass. * fix(replication): space periodic MRF replay across nodes Every node runs the MRF replay loop and the recovery leader lock only serializes passes, so an N-node cluster replayed the same retained backlog N times per interval while a target stayed down. Record the start time of a pass that finds entries in a shared replay stamp and skip a periodic pass when any node started one within the interval. The startup pass still always runs.
fix(heal): treat SlowDown as a rate signal and bound page churn (#9144 )
2026-11-04 01:53:36 +00:00 · 2026-11-03 17:06:14 +00:00 · 2026-11-03 17:05:05 +00:00 · 2026-11-03 15:11:58 +00:00 · 2026-11-03 15:04:29 +00:00
24 changed files with 1212 additions and 748 deletions
@@ -26,6 +26,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - IAM snapshot/deletion replay, target-assigned delete-marker purges, timestamp ordering, and best-effort peer broadcast now close the control-plane gaps found by the R6 review ([#7195](https://github.com/rustfs/rustfs/pull/7195)).
 - A `set-remote-target?update=true&creds=true` request on a site-replication peer's bucket target is now rejected with `400 InvalidRequest`, naming the target and why it counts as peer-owned. It used to return success while keeping the stored endpoint, credentials, and TLS settings, so an endpoint edit from the Console's bucket replication form appeared to apply but did not. Change a peer's address with a site-replication edit instead; see "Changing a site's address" in `docs/operations/site-replication-operations.md`. The `sync`, `bandwidth`, `proxy`, and `path` update groups are unchanged.
 - Site-replication edits that omit the site name, such as `mc admin replicate update --deployment-id <ID> --endpoint <URL>`, now keep the stored name on the editing site and on every receiver instead of renaming the site after the new endpoint host. No persisted format changes.
+- The persisted replication MRF journal is now replayed every 5 minutes while the server runs, not only at startup. Delete-marker purges whose targets were unreachable, and replications that overflowed the worker queues, now converge within one replay interval after the target returns instead of waiting for a restart. A target that stays down is re-attempted once per retained entry per interval across the cluster: the start time of the last pass is shared through a small `config/replication/mrf.bin.replay` object, so nodes do not each replay once per interval. Older versions ignore that object; existing persisted formats are unchanged.
 - Upgrade and rollback: upgrade every node in one site consecutively and verify reconciliation before moving to the next site; do not intentionally run a site mixed-version. Target-version ledger keys are harmless on rollback, although old code cannot use their routing. Before rolling back past [#7307](https://github.com/rustfs/rustfs/pull/7307), drain or repair every pending version purge: older code can free a retained version's data directory before its remote purge is acknowledged. See `docs/operations/site-replication-operations.md`.

 ### Security
@@ -40,9 +40,9 @@ dependencies = [

 [[package]]
 name = "aes"
-version = "0.9.3"
+version = "0.9.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "35f0f96ce78e38c3dc6d8948aa8163d06385be74000f3c7a95bf1eef35d3ea32"
+checksum = "fc0fc36d7440d015b45468ba08a21e80e056ebbd014c611c6298bb7503e04bd1"
 dependencies = [
 "cipher 0.5.2",
 "cpubits",
@@ -57,7 +57,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7f2b8006a0c83f52b62ba44a97b58bf76fe2f70a329e588f67f89691d93d498f"
 dependencies = [
 "aead",
- "aes 0.9.3",
+ "aes 0.9.4",
 "cipher 0.5.2",
 "ctr",
 "ctutils",
@@ -772,9 +772,9 @@ dependencies = [

 [[package]]
 name = "async-executor"
-version = "1.14.0"
+version = "1.14.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c96bf972d85afc50bf5ab8fe2d54d1586b4e0b46c97c50a0c9e71e2f7bcd812a"
+checksum = "f426a31126c735126297bc5ea7a521263c9faff4f63ca415267ee625651c4f37"
 dependencies = [
 "async-task",
 "concurrent-queue",
@@ -800,9 +800,9 @@ dependencies = [

 [[package]]
 name = "async-lock"
-version = "3.4.2"
+version = "3.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "290f7f2596bd5b78a9fec8088ccd89180d7f9f55b94b0576823bbbdc72ee8311"
+checksum = "0876700b0d007ad25d5508865a4259c3af9af6d296bf663a0d586d0bc3369f6f"
 dependencies = [
 "event-listener",
 "event-listener-strategy",
@@ -2580,9 +2580,9 @@ dependencies = [

 [[package]]
 name = "crc32c"
-version = "0.6.8"
+version = "0.6.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3a47af21622d091a8f0fb295b88bc886ac74efcc613efc19f5d0b21de5c89e47"
+checksum = "7f132a176423ee852e8f0a35d43088d1850966b50745be818a88b76f64b1c6cb"
 dependencies = [
 "rustc_version",
 ]
@@ -2766,9 +2766,9 @@ dependencies = [

 [[package]]
 name = "ctutils"
-version = "0.4.7"
+version = "0.4.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a2376ffd6f96d8a9db59875b4b6b0237d6900b5baf64279714e05493ce64828a"
+checksum = "261af4dc7a06ea8a527a8717d8cbba3417530d6ea14e1edb3de6b67806be474f"
 dependencies = [
 "cmov",
 "subtle",
@@ -3906,9 +3906,9 @@ dependencies = [

 [[package]]
 name = "dial9"
-version = "0.5.6"
+version = "0.5.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "11185c62245c5ff61c753ec8ec3d56341a13fe4195344ecc100ce94f780533cc"
+checksum = "2f3298a54fd82af243c9f122595762f5b0e48bd3d1ea8fe3f3eb5e95c130e881"
 dependencies = [
 "bon",
 "dial9-core",
@@ -3923,9 +3923,9 @@ dependencies = [

 [[package]]
 name = "dial9-core"
-version = "0.5.6"
+version = "0.5.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "54f1dcb5d20bdc85b9beb7c2864540526c1394c00783e9f3544feed1cfe529e8"
+checksum = "ebc7907b149a394404e3075f57bfbc751c7292ef115027f13f4995ff3a8b360c"
 dependencies = [
 "ahash",
 "arc-swap",
@@ -3957,9 +3957,9 @@ dependencies = [

 [[package]]
 name = "dial9-metrique"
-version = "0.5.6"
+version = "0.5.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "56645773b4185f51a9528625a5aedcd4bbc5b5919ba68e7a6108cf076f635efc"
+checksum = "35f30e80d69214ed7c684d81afd8e52315b4acbfa5be9ebb84ccee02ebd06f44"
 dependencies = [
 "dial9-core",
 "dial9-trace-format",
@@ -3971,9 +3971,9 @@ dependencies = [

 [[package]]
 name = "dial9-perf-self-profile"
-version = "0.5.6"
+version = "0.5.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "00c8c96d18b28ec76adde2107f7d2d5f8bb53b0670fba31bc012841703bc49ca"
+checksum = "17f09434b7623ef8b9b107062941a8e471e6044d6c946fa4af899130223b7a28"
 dependencies = [
 "blazesym",
 "bon",
@@ -3990,9 +3990,9 @@ dependencies = [

 [[package]]
 name = "dial9-tokio-telemetry"
-version = "0.5.6"
+version = "0.5.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7ccc79f3683219471a8758b8ce459589167c55d72a649d27b89068a1a4cfffe9"
+checksum = "0428cba60fbcf37ab82666f3a7ef3b6d0671e90dfb6a352376813b471319f061"
 dependencies = [
 "bon",
 "bytes",
@@ -4015,9 +4015,9 @@ dependencies = [

 [[package]]
 name = "dial9-trace-format"
-version = "0.5.3"
+version = "0.5.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "44026339a4c6cbd4914c3eb85e653e741d5739f1e401695a039127ba49c19731"
+checksum = "af753ba562fab9b99024b74234fe3a93b01c32d6474db8ae4f2c3bb562958f8f"
 dependencies = [
 "ahash",
 "dial9-trace-format-derive",
@@ -4027,9 +4027,9 @@ dependencies = [

 [[package]]
 name = "dial9-trace-format-derive"
-version = "0.5.3"
+version = "0.5.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bbdbf5c76eb6824b0576e6891811caae93fc0bc96c23003556e3f752d069a84d"
+checksum = "e00786284c5d77b60aac60ee96797b8ac020306da8f02fecb0e86c5e82610aae"
 dependencies = [
 "proc-macro-crate",
 "proc-macro2",
@@ -5895,9 +5895,9 @@ checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682"

 [[package]]
 name = "jiff"
-version = "0.2.38"
+version = "0.2.39"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b2b005715dcbeb0089a3c0dab99f2ff1cc3b2525323552703d648585d342a383"
+checksum = "3ac703058abac89e5c39b0b879dce2b6b14a4ba36600c7e7e3d0c60a6883ef3c"
 dependencies = [
 "defmt 1.2.0",
 "jiff-core",
@@ -5912,9 +5912,9 @@ dependencies = [

 [[package]]
 name = "jiff-core"
-version = "0.1.1"
+version = "0.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5e52fe76043ccecc9005d2305ebaadf7d7fc0cc89ca6baa10a94d6bc68c7128c"
+checksum = "aee43309eb53bafb8c36d35d6e444769c8f8e6206a9fbf7546f5db9d7d4a0e93"
 dependencies = [
 "defmt 1.2.0",
 "log",
@@ -5922,9 +5922,9 @@ dependencies = [

 [[package]]
 name = "jiff-static"
-version = "0.2.38"
+version = "0.2.39"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2cc9817253cf7c7ee4684451bd327e88d6f3658014e54a29198625590650695c"
+checksum = "6dc32470c67eb7d0e9ccb74303fa925313dc04cecb1aa048beaa6f391e36a93d"
 dependencies = [
 "jiff-core",
 "proc-macro2",
@@ -6413,9 +6413,9 @@ dependencies = [

 [[package]]
 name = "lzma-rust2"
-version = "0.23.0"
+version = "0.24.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f1c46f3fcd2e888afdb01a07efc071cb21300e07c68d32c3e8b01b5c8cb1fcaa"
+checksum = "490750560fa7851fbbf5d3ab1a8750f09350ed361ed8aed85b1e1e40587812ee"
 dependencies = [
 "sha2 0.11.1",
 ]
@@ -8040,7 +8040,7 @@ version = "0.8.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "63d440a804ec8d6fafbb6b84471e013286658d373248927692ab3366686220ca"
 dependencies = [
- "aes 0.9.3",
+ "aes 0.9.4",
 "aes-gcm",
 "cbc 0.2.1",
 "der 0.8.3",
@@ -9353,7 +9353,7 @@ version = "0.65.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3bb77a4b259312b2abbc541bca1c13cb4e9b61c560a7fd894068c3b92e0c9dc3"
 dependencies = [
- "aes 0.9.3",
+ "aes 0.9.4",
 "aws-lc-rs",
 "bitflags 2.13.2",
 "block-padding 0.4.2",
@@ -12145,7 +12145,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d801accda99469cde6d73da741422610fdf6508a72d9a69d1b55cb241c720597"
 dependencies = [
 "aead",
- "aes 0.9.3",
+ "aes 0.9.4",
 "aes-gcm",
 "chacha20",
 "cipher 0.5.2",
@@ -14199,11 +14199,11 @@ dependencies = [

 [[package]]
 name = "zip"
-version = "9.2.2"
+version = "9.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e408c6cc4699dd017d53dc080cd51c2cea2b5354af57a1479bd265ed42911487"
+checksum = "eaeb7f16d2605b49aef586996ded29e65642fab7259a3077e537351df6d80426"
 dependencies = [
- "aes 0.9.3",
+ "aes 0.9.4",
 "bzip2",
 "constant_time_eq",
 "crc32fast",
@@ -235,7 +235,7 @@ proptest = "1"
 # Time and Date
 chrono = { version = "0.4.45" }
 humantime = "2.4.0"
-jiff = { version = "0.2.38" }
+jiff = { version = "0.2.39" }
 time = { version = "0.3.55" }

 # Database
@@ -357,13 +357,13 @@ winapi-util = "0.1.11"
 windows = { version = "0.62.2" }
 windows-sys = "0.61.2"
 xxhash-rust = { version = "0.8.19" }
-zip = "9.2.2"
+zip = "9.3.0"
 zstd = "0.14.3"

 # Observability and Metrics
 metrics = "0.24.6"
 metrics-util = "0.20"
-dial9 = { version = "0.5.6", default-features = false }
+dial9 = { version = "0.5.7", default-features = false }
 opentelemetry = { version = "0.33.1" }
 opentelemetry-appender-tracing = { version = "0.33.1" }
 opentelemetry-otlp = { version = "0.33.1" }
@@ -33,6 +33,7 @@ pub(crate) struct ReplicationMetadataStore;
 impl ReplicationMetadataStore {
    pub(crate) const MRF_REPLICATION_FILE: &'static str = "config/replication/mrf.bin";
    pub(crate) const MRF_REPLICATION_RECOVERY_LOCK: &'static str = "config/replication/mrf.bin.recovery";
+    pub(crate) const MRF_REPLICATION_REPLAY_STAMP: &'static str = "config/replication/mrf.bin.replay";
    pub(crate) const FORCE_DELETE_REPLICATION_FILE: &'static str = "config/replication/force-delete.bin";
    pub(crate) const FORCE_DELETE_REPLICATION_TRANSACTION_LOCK: &'static str = "config/replication/force-delete.bin.transaction";

@@ -116,6 +117,10 @@ mod tests {
            ReplicationMetadataStore::MRF_REPLICATION_RECOVERY_LOCK,
            "config/replication/mrf.bin.recovery"
        );
+        assert_eq!(
+            ReplicationMetadataStore::MRF_REPLICATION_REPLAY_STAMP,
+            "config/replication/mrf.bin.replay"
+        );
        assert_eq!(
            ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE,
            "config/replication/force-delete.bin"
@@ -86,6 +86,10 @@ const MRF_CORRUPT_FILE_PREFIX: &str = "config/replication/mrf.corrupt";
 const MRF_PENDING_CAP: usize = 200_000;
 const MRF_RETRY_INITIAL_DELAY: Duration = Duration::from_millis(100);
 const MRF_RETRY_MAX_DELAY: Duration = Duration::from_secs(5);
+/// Interval between runtime replays of the MRF journal, cluster-wide. A pass
+/// re-attempts every retained entry, so a target that stays down costs one
+/// attempt per entry per interval.
+const MRF_REPLAY_INTERVAL: Duration = Duration::from_secs(300);

 #[derive(Debug, Default)]
 pub struct DurableMrfBacklog {
@@ -615,7 +619,7 @@ async fn read_mrf_entries<S: ReplicationObjectIO>(storage: Arc<S>) -> Result<Vec
 }

 /// Acknowledge only the generation read by the recovery leader. Entries appended
-/// after that generation are retained as a suffix and replayed on the next startup.
+/// after that generation are retained as a suffix and replayed on the next pass.
 /// Lock order: recovery leader lock -> MRF journal object lock.
 async fn acknowledge_mrf_recovery<S: ReplicationStorage>(
    storage: Arc<S>,
@@ -667,7 +671,7 @@ async fn acknowledge_mrf_recovery<S: ReplicationStorage>(
    Err(EcstoreError::PreconditionFailed)
 }

-/// Acquires the MRF recovery leader lock for the startup replay.
+/// Acquires the MRF recovery leader lock for one journal replay pass.
 /// Returns `None` (after logging) when the lock cannot be created or another
 /// node is already processing the backlog.
 async fn acquire_mrf_recovery_guard<S: ReplicationStorage>(storage: &Arc<S>) -> Option<rustfs_lock::NamespaceLockGuard> {
@@ -1018,7 +1022,7 @@ fn admitted_mrf_replicate_object(oi: ObjectInfo, entry: &MrfReplicateEntry, op_t
 }

 /// Acknowledges the replayed MRF prefix and returns the retained backlog.
-/// On acknowledgement failure the backlog is preserved for the next startup and
+/// On acknowledgement failure the backlog is preserved for the next replay pass and
 /// re-read (falling back to the replayed snapshot) so the published summary stays accurate.
 async fn resolve_retained_mrf_entries<S: ReplicationStorage>(
    storage: &Arc<S>,
@@ -1033,7 +1037,7 @@ async fn resolve_retained_mrf_entries<S: ReplicationStorage>(
                component = LOG_COMPONENT_ECSTORE,
                subsystem = LOG_SUBSYSTEM_REPLICATION,
                error = %error,
-                "Failed to acknowledge the MRF recovery prefix; preserving it for the next startup"
+                "Failed to acknowledge the MRF recovery prefix; preserving it for the next replay pass"
            );
            match read_mrf_entries(storage.clone()).await {
                Ok(current) => current,
@@ -1051,6 +1055,120 @@ async fn resolve_retained_mrf_entries<S: ReplicationStorage>(
    }
 }

+fn unix_millis_now() -> i64 {
+    i64::try_from(OffsetDateTime::now_utc().unix_timestamp_nanos() / 1_000_000).unwrap_or(i64::MAX)
+}
+
+/// Whether a periodic pass may run, given the start time of the last pass on
+/// any node. A stamp more than one interval ahead of this node's clock is skew
+/// or a bad value, so it does not suppress replay.
+fn mrf_replay_due(last_started_ms: Option<i64>, now_ms: i64, interval: Duration) -> bool {
+    let Some(last_started_ms) = last_started_ms else {
+        return true;
+    };
+    let interval_ms = i64::try_from(interval.as_millis()).unwrap_or(i64::MAX);
+    now_ms.saturating_sub(last_started_ms) >= interval_ms || last_started_ms.saturating_sub(now_ms) >= interval_ms
+}
+
+async fn read_mrf_replay_stamp<S: ReplicationStorage>(storage: &Arc<S>) -> Option<i64> {
+    let data = ReplicationConfigStore::read(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_REPLAY_STAMP)
+        .await
+        .ok()?;
+    std::str::from_utf8(&data).ok()?.trim().parse().ok()
+}
+
+async fn write_mrf_replay_stamp<S: ReplicationStorage>(storage: &Arc<S>, started_ms: i64) {
+    if let Err(error) = ReplicationConfigStore::save(
+        storage.clone(),
+        ReplicationMetadataStore::MRF_REPLICATION_REPLAY_STAMP,
+        started_ms.to_string().into_bytes(),
+    )
+    .await
+    {
+        warn!(
+            component = LOG_COMPONENT_ECSTORE,
+            subsystem = LOG_SUBSYSTEM_REPLICATION,
+            error = %error,
+            "Failed to record the MRF replay start time"
+        );
+    }
+}
+
+/// One MRF journal replay pass. The recovery leader reads the journal,
+/// re-drives every entry, and acknowledges the replayed prefix; entries that
+/// could not be admitted, and entries appended meanwhile, stay for the next pass.
+///
+/// Every node runs the periodic loop, and the leader lock only serializes
+/// passes. With `min_interval`, a pass is skipped when a pass on any node
+/// started within that interval, read from the shared replay stamp, so a
+/// cluster replays once per interval rather than once per node. The startup
+/// pass passes `None` and always runs. A pass that finds entries records its
+/// start time. Returns whether the journal was read.
+/// Lock order: recovery leader lock -> replay stamp object lock.
+async fn replay_mrf_journal<S: ReplicationStorage>(storage: &Arc<S>, min_interval: Option<Duration>) -> bool {
+    let Some(recovery_guard) = acquire_mrf_recovery_guard(storage).await else {
+        return false;
+    };
+    let started_ms = unix_millis_now();
+    if let Some(interval) = min_interval
+        && !mrf_replay_due(read_mrf_replay_stamp(storage).await, started_ms, interval)
+    {
+        return false;
+    }
+
+    let Some(entries) = load_mrf_recovery_entries(storage).await else {
+        return true;
+    };
+    if !entries.is_empty() {
+        write_mrf_replay_stamp(storage, started_ms).await;
+    }
+    set_durable_mrf_backlog_snapshot(durable_mrf_backlog_summary_from_entries(&entries));
+
+    let total = entries.len();
+    let mut queued_count = 0usize;
+    let mut retry_entries = Vec::new();
+
+    for entry in entries.iter() {
+        let Some(admission) = replay_mrf_entry(entry, storage, &mut retry_entries).await else {
+            continue;
+        };
+
+        if admission == ReplicationQueueAdmission::Missed {
+            retry_entries.push(entry.clone());
+        } else if admission == ReplicationQueueAdmission::Queued {
+            queued_count += 1;
+        }
+    }
+
+    let retained = resolve_retained_mrf_entries(storage, &recovery_guard, &entries, &retry_entries).await;
+    let retained_count = retained.len();
+    set_durable_mrf_backlog_snapshot(durable_mrf_backlog_summary_from_entries(&retained));
+
+    if queued_count > 0 {
+        info!(
+            component = LOG_COMPONENT_ECSTORE,
+            subsystem = LOG_SUBSYSTEM_REPLICATION,
+            recovered = queued_count,
+            total,
+            retained = retained_count,
+            "Replayed MRF entries admitted for retry"
+        );
+    }
+    true
+}
+
+/// Re-runs [`replay_mrf_journal`] every `period`. The recovery leader lock keeps
+/// one pass at a time across the cluster, including the startup pass, and the
+/// shared replay stamp spaces passes from different nodes by `period`.
+fn spawn_mrf_replay_loop<S: ReplicationStorage>(storage: Arc<S>, period: Duration) -> JoinHandle<()> {
+    tokio::spawn(async move {
+        loop {
+            tokio::time::sleep(period).await;
+            replay_mrf_journal(&storage, Some(period)).await;
+        }
+    })
+}
+
 #[derive(Debug, thiserror::Error)]
 #[error("replication resync {active_resync_id} is already active for {bucket}/{arn}")]
 struct ResyncActiveConflictError {
@@ -1171,6 +1289,7 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
        // Start background tasks
        pool.start_mrf_persister().await;
        pool.start_mrf_processor().await;
+        pool.start_mrf_replayer().await;
        pool.start_force_delete_processor().await;

        pool
@@ -1639,55 +1758,26 @@ impl<S: ReplicationStorage> ReplicationPool<S> {

    /// Starts the MRF processor — one-shot at startup.
    ///
-    /// Reads the on-disk MRF file, re-injects admitted entries as Heal operations, and
-    /// rewrites any entries that could not be admitted for a later startup retry.
+    /// Replays the on-disk MRF journal once; [`Self::start_mrf_replayer`] repeats
+    /// the pass while the server runs.
    async fn start_mrf_processor(&self) {
        let storage = self.storage.clone();
-
        let handle = tokio::spawn(async move {
-            let Some(recovery_guard) = acquire_mrf_recovery_guard(&storage).await else {
-                return;
-            };
-
-            let Some(entries) = load_mrf_recovery_entries(&storage).await else {
-                return;
-            };
-            set_durable_mrf_backlog_snapshot(durable_mrf_backlog_summary_from_entries(&entries));
-
-            let total = entries.len();
-            let mut queued_count = 0usize;
-            let mut retry_entries = Vec::new();
-
-            for entry in entries.iter() {
-                let Some(admission) = replay_mrf_entry(entry, &storage, &mut retry_entries).await else {
-                    continue;
-                };
-
-                if admission == ReplicationQueueAdmission::Missed {
-                    retry_entries.push(entry.clone());
-                } else if admission == ReplicationQueueAdmission::Queued {
-                    queued_count += 1;
-                }
-            }
-
-            let retained = resolve_retained_mrf_entries(&storage, &recovery_guard, &entries, &retry_entries).await;
-            let retained_count = retained.len();
-            set_durable_mrf_backlog_snapshot(durable_mrf_backlog_summary_from_entries(&retained));
-
-            if queued_count > 0 {
-                info!(
-                    component = LOG_COMPONENT_ECSTORE,
-                    subsystem = LOG_SUBSYSTEM_REPLICATION,
-                    recovered = queued_count,
-                    total,
-                    retained = retained_count,
-                    "Replayed MRF entries admitted for retry"
-                );
-            }
+            replay_mrf_journal(&storage, None).await;
        });
        self.task_handles.lock().await.push(handle);
    }

+    /// Starts the runtime MRF replayer — ongoing background task.
+    ///
+    /// Entries reach the journal while the server runs (worker-queue overflow,
+    /// delete-marker purge intents whose watch window drained with targets still
+    /// dirty). Replaying only at startup left them waiting for a restart.
+    async fn start_mrf_replayer(&self) {
+        let handle = spawn_mrf_replay_loop(self.storage.clone(), MRF_REPLAY_INTERVAL);
+        self.task_handles.lock().await.push(handle);
+    }
+
    async fn start_force_delete_processor(&self) {
        let storage = self.storage.clone();
        let handle =
@@ -1755,7 +1845,7 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
    /// writes them to the on-disk MRF file every flush interval (default 10s,
    /// overridable via `RUSTFS_REPL_MRF_FLUSH_INTERVAL_MS`) or when 1 000 new
    /// entries accumulate. Runtime writers append under the journal lock; only the
-    /// startup recovery leader may remove an acknowledged prefix.
+    /// recovery leader of a replay pass may remove an acknowledged prefix.
    async fn start_mrf_persister(&self) {
        let Some(mut rx) = self.mrf_save_rx.lock().await.take() else {
            return;
@@ -3422,6 +3512,7 @@ mod tests {
        read_count: AtomicUsize,
        delete_storage_calls: AtomicUsize,
        delete_error: StdMutex<EcstoreError>,
+        replay_stamp: StdMutex<Option<Vec<u8>>>,
        reported_size: StdMutex<Option<i64>>,
        stream_read_bytes: Arc<AtomicUsize>,
        write_count: AtomicUsize,
@@ -3489,6 +3580,24 @@ mod tests {
            _h: Self::HeaderMap,
            _opts: &Self::ObjectOptions,
        ) -> Result<Self::GetObjectReader, Self::Error> {
+            if object.ends_with(ReplicationMetadataStore::MRF_REPLICATION_REPLAY_STAMP) {
+                let stamp = self
+                    .shared
+                    .replay_stamp
+                    .lock()
+                    .expect("test replay stamp lock should not be poisoned")
+                    .clone()
+                    .ok_or(EcstoreError::FileNotFound)?;
+                return Ok(Self::GetObjectReader {
+                    object_info: ObjectInfo {
+                        size: i64::try_from(stamp.len()).expect("test stamp length should fit i64"),
+                        ..Default::default()
+                    },
+                    stream: Box::new(Cursor::new(stamp)),
+                    buffered_body: None,
+                    body_source: Default::default(),
+                });
+            }
            if !object.ends_with("/.replication/resync.bin")
                && !object.ends_with("config/replication/mrf.bin")
                && !object.ends_with("config/replication/force-delete.bin")
@@ -3552,6 +3661,16 @@ mod tests {
                let lock = self.new_ns_lock(bucket, object).await?;
                Some(lock.get_write_lock(Duration::from_secs(10)).await?)
            };
+            if object.ends_with(ReplicationMetadataStore::MRF_REPLICATION_REPLAY_STAMP) {
+                let mut stamp = Vec::new();
+                data.stream.read_to_end(&mut stamp).await.map_err(EcstoreError::from)?;
+                *self
+                    .shared
+                    .replay_stamp
+                    .lock()
+                    .expect("test replay stamp lock should not be poisoned") = Some(stamp);
+                return Ok(ObjectInfo::default());
+            }
            if opts.http_preconditions.is_some()
                && let Some(replacement) = self
                    .shared
@@ -4178,6 +4297,7 @@ mod tests {
            read_count: AtomicUsize::new(0),
            delete_storage_calls: AtomicUsize::new(0),
            delete_error: StdMutex::new(EcstoreError::NotImplemented),
+            replay_stamp: StdMutex::new(None),
            reported_size: StdMutex::new(None),
            stream_read_bytes: Arc::new(AtomicUsize::new(0)),
            write_count: AtomicUsize::new(0),
@@ -5409,6 +5529,7 @@ mod tests {
                read_count: AtomicUsize::new(0),
                delete_storage_calls: AtomicUsize::new(0),
                delete_error: StdMutex::new(EcstoreError::NotImplemented),
+                replay_stamp: StdMutex::new(None),
                reported_size: StdMutex::new(None),
                stream_read_bytes: Arc::new(AtomicUsize::new(0)),
                write_count: AtomicUsize::new(0),
@@ -6373,6 +6494,108 @@ mod tests {
        assert_eq!(retained.target_arns, entry.target_arns);
    }

+    #[tokio::test]
+    async fn mrf_replay_loop_drains_entries_persisted_after_startup() {
+        let shared = empty_resync_shared_state();
+        let storage = Arc::new(LoadResyncNodeStore::new("mrf-runtime-replay", shared.clone()));
+        // A force-delete intent from a superseded generation replays as
+        // Skipped, so a pass that reaches it acknowledges it off the journal.
+        let stale_intent = |object: &str| MrfReplicateEntry {
+            bucket: "mrf-runtime-replay".to_string(),
+            object: object.to_string(),
+            op: MrfOpKind::Delete,
+            force_delete_id: Some(uuid::Uuid::new_v4()),
+            force_delete_local_commit: false,
+            target_arns: vec!["arn:rustfs:replication:target-a".to_string()],
+            ..Default::default()
+        };
+        let journal_drained = || async {
+            tokio::time::timeout(Duration::from_secs(5), async {
+                while !shared.data.lock().expect("test data lock should not be poisoned").is_empty() {
+                    tokio::time::sleep(Duration::from_millis(10)).await;
+                }
+            })
+            .await
+            .is_ok()
+        };
+
+        let handle = spawn_mrf_replay_loop(storage, Duration::from_millis(20));
+        for object in ["first", "second"] {
+            *shared.data.lock().expect("test data lock should not be poisoned") =
+                encode_mrf_file(&[stale_intent(object)]).expect("MRF entry should encode");
+            assert!(
+                journal_drained().await,
+                "an MRF entry persisted while the server runs must be replayed without a restart ({object})"
+            );
+        }
+        handle.abort();
+    }
+
+    #[tokio::test]
+    async fn mrf_periodic_replay_runs_once_per_interval_across_nodes() {
+        let shared = empty_resync_shared_state();
+        let node_a = Arc::new(LoadResyncNodeStore::new("mrf-node-a", shared.clone()));
+        let node_b = Arc::new(LoadResyncNodeStore::new("mrf-node-b", shared.clone()));
+        let interval = Duration::from_secs(1);
+        let persist_backlog = |object: &str| {
+            let entry = MrfReplicateEntry {
+                bucket: "mrf-node-replay".to_string(),
+                object: object.to_string(),
+                op: MrfOpKind::Delete,
+                force_delete_id: Some(uuid::Uuid::new_v4()),
+                force_delete_local_commit: false,
+                target_arns: vec!["arn:rustfs:replication:target-a".to_string()],
+                ..Default::default()
+            };
+            *shared.data.lock().expect("test data lock should not be poisoned") =
+                encode_mrf_file(&[entry]).expect("MRF entry should encode");
+        };
+        let journal_is_empty = || shared.data.lock().expect("test data lock should not be poisoned").is_empty();
+
+        persist_backlog("first");
+        assert!(replay_mrf_journal(&node_a, Some(interval)).await, "node A's periodic pass should replay");
+        assert!(journal_is_empty(), "node A's pass should acknowledge the replayed entry");
+
+        // Node B's timer fires within the same interval with a backlog present.
+        persist_backlog("second");
+        assert!(
+            !replay_mrf_journal(&node_b, Some(interval)).await,
+            "a node must not replay within one interval of another node's pass"
+        );
+        assert!(!journal_is_empty(), "the backlog should wait for the next interval");
+
+        assert!(
+            replay_mrf_journal(&node_b, None).await,
+            "the startup pass should replay regardless of the shared stamp"
+        );
+        assert!(journal_is_empty(), "the startup pass should acknowledge the replayed entry");
+
+        persist_backlog("third");
+        tokio::time::sleep(interval).await;
+        assert!(
+            replay_mrf_journal(&node_a, Some(interval)).await,
+            "a periodic pass should replay once the interval has elapsed"
+        );
+        assert!(journal_is_empty(), "the due pass should acknowledge the replayed entry");
+    }
+
+    #[test]
+    fn mrf_replay_due_tolerates_missing_and_skewed_stamps() {
+        let interval = Duration::from_secs(300);
+        let now = 1_000_000_000;
+        assert!(mrf_replay_due(None, now, interval));
+        assert!(!mrf_replay_due(Some(now - 299_999), now, interval));
+        assert!(mrf_replay_due(Some(now - 300_000), now, interval));
+        assert!(
+            !mrf_replay_due(Some(now + 1_000), now, interval),
+            "small clock skew should still space passes"
+        );
+        assert!(
+            mrf_replay_due(Some(now + 300_000), now, interval),
+            "a stamp an interval ahead must not suppress replay"
+        );
+    }
+
    #[test]
    fn mrf_entry_delete_marker_roundtrip() {
        let dm_vid = Uuid::new_v4();
@@ -3015,7 +3015,7 @@ const DELETE_MARKER_PURGE_WATCH_INTERVAL: TokioDuration = TokioDuration::from_se
 /// If the marker disappears (deleted before or while the replica landed),
 /// purge the replicated marker from the targets, retrying failed targets on
 /// later rounds. When the window drains with targets still dirty, persist the
-/// purge intent as a durable MRF entry so the next startup replays it through
+/// purge intent as a durable MRF entry so the next MRF replay pass re-drives it through
 /// `purge_stale_delete_marker_targets`.
 async fn watch_and_purge_source_delete_marker<S: ReplicationStorage>(
    bucket: String,
@@ -3106,7 +3106,7 @@ async fn enqueue_delete_marker_purge_mrf(dobj: &DeletedObjectReplicationInfo, fa
 /// Heal callers retry for the full watch window because the startup MRF
 /// processor runs before bucket metadata (and thus target clients) finishes
 /// initializing — the first pass can see `target_client_missing` and a later
-/// round resolves the client; the replay loop is serial and startup-only, so
+/// round resolves the client; replay passes are serial background work, so
 /// blocking it for up to the window per dirty entry is acceptable. Live
 /// callers run on replication workers where a down target would pin a worker
 /// for the whole window, so they attempt once and lean on the durable intent
@@ -8408,7 +8408,7 @@ impl LocalDisk {
                },
            )
        {
-            multipart_source_state::put_active_multipart_source_head(admission.source().to_path_buf(), fm_bytes, cached_fi);
+            multipart_source_state::put_active_multipart_source_head(admission.source(), fm_bytes, cached_fi);
        }
        drop(namespace_owner);

@@ -1002,21 +1002,17 @@ pub(super) fn inspection_with_markers(

 type PreparedDeleteReceiptEntry = (DeleteRecoveryJournal, bool);

-static PREPARED_DELETE_RECEIPTS: std::sync::LazyLock<parking_lot::Mutex<HashMap<PathBuf, PreparedDeleteReceiptEntry>>> =
-    std::sync::LazyLock::new(|| parking_lot::Mutex::new(HashMap::new()));
+const PREPARED_DELETE_RECEIPTS_PER_SHARD: usize = 64;

-fn put_prepared_delete_receipt(rollback: PathBuf, entry: PreparedDeleteReceiptEntry) {
-    let mut map = PREPARED_DELETE_RECEIPTS.lock();
-    if map.len() >= 4096
-        && let Some(evict_key) = map.keys().next().cloned()
-    {
-        map.remove(&evict_key);
-    }
-    map.insert(rollback, entry);
+static PREPARED_DELETE_RECEIPTS: std::sync::LazyLock<os::ShardedPathCache<PreparedDeleteReceiptEntry>> =
+    std::sync::LazyLock::new(|| os::ShardedPathCache::new(PREPARED_DELETE_RECEIPTS_PER_SHARD));
+
+fn put_prepared_delete_receipt(rollback: impl AsRef<Path>, entry: PreparedDeleteReceiptEntry) {
+    PREPARED_DELETE_RECEIPTS.insert(rollback.as_ref(), entry);
 }

 fn take_prepared_delete_receipt(rollback: &Path) -> Option<PreparedDeleteReceiptEntry> {
-    PREPARED_DELETE_RECEIPTS.lock().remove(rollback)
+    PREPARED_DELETE_RECEIPTS.remove(rollback)
 }

 impl LocalDisk {
@@ -270,21 +270,17 @@ struct ActiveMultipartSourceEntry {
    parts: std::collections::HashMap<usize, (FileStamp, rustfs_filemeta::ObjectPartInfo)>,
 }

-static ACTIVE_MULTIPART_SOURCE_HEADS: std::sync::LazyLock<
-    parking_lot::Mutex<std::collections::HashMap<PathBuf, ActiveMultipartSourceEntry>>,
-> = std::sync::LazyLock::new(|| parking_lot::Mutex::new(std::collections::HashMap::new()));
+const ACTIVE_MULTIPART_SOURCE_HEADS_PER_SHARD: usize = 64;

-pub(super) fn put_active_multipart_source_head(source: PathBuf, raw: Bytes, fi: FileInfo) {
+static ACTIVE_MULTIPART_SOURCE_HEADS: std::sync::LazyLock<os::ShardedPathCache<ActiveMultipartSourceEntry>> =
+    std::sync::LazyLock::new(|| os::ShardedPathCache::new(ACTIVE_MULTIPART_SOURCE_HEADS_PER_SHARD));
+
+pub(super) fn put_active_multipart_source_head(source: impl AsRef<Path>, raw: Bytes, fi: FileInfo) {
+    let source = source.as_ref();
    let Some(head_stamp) = FileStamp::from_path(&source.join(STORAGE_FORMAT_FILE)) else {
        return;
    };
-    let mut map = ACTIVE_MULTIPART_SOURCE_HEADS.lock();
-    if map.len() >= 4096
-        && let Some(evict_key) = map.keys().next().cloned()
-    {
-        map.remove(&evict_key);
-    }
-    map.insert(
+    ACTIVE_MULTIPART_SOURCE_HEADS.insert(
        source,
        ActiveMultipartSourceEntry {
            raw,
@@ -304,56 +300,50 @@ fn stamp_matches_active_disk(source: &Path, expected_head_stamp: FileStamp) -> b
 }

 pub(super) fn get_active_multipart_source_head(source: &Path) -> Option<(Bytes, FileInfo)> {
-    let (raw, fi, head_stamp) = {
-        let map = ACTIVE_MULTIPART_SOURCE_HEADS.lock();
-        let entry = map.get(source)?;
-        (entry.raw.clone(), entry.fi.clone(), entry.head_stamp)
-    };
+    let (raw, fi, head_stamp) =
+        ACTIVE_MULTIPART_SOURCE_HEADS.with_get(source, |entry| (entry.raw.clone(), entry.fi.clone(), entry.head_stamp))?;
    if !stamp_matches_active_disk(source, head_stamp) {
-        ACTIVE_MULTIPART_SOURCE_HEADS.lock().remove(source);
+        ACTIVE_MULTIPART_SOURCE_HEADS.remove(source);
        return None;
    }
    Some((raw, fi))
 }

 pub(super) fn has_active_multipart_source_head(source: &Path) -> bool {
-    let Some(head_stamp) = ACTIVE_MULTIPART_SOURCE_HEADS.lock().get(source).map(|entry| entry.head_stamp) else {
+    let Some(head_stamp) = ACTIVE_MULTIPART_SOURCE_HEADS.with_get(source, |entry| entry.head_stamp) else {
        return false;
    };
    if !stamp_matches_active_disk(source, head_stamp) {
-        ACTIVE_MULTIPART_SOURCE_HEADS.lock().remove(source);
+        ACTIVE_MULTIPART_SOURCE_HEADS.remove(source);
        return false;
    }
    true
 }

 pub(super) fn has_clean_active_multipart_source(source: &Path) -> bool {
-    let Some((head_stamp, clean)) = ACTIVE_MULTIPART_SOURCE_HEADS
-        .lock()
-        .get(source)
-        .map(|entry| (entry.head_stamp, entry.unsettled_parts.is_empty()))
+    let Some((head_stamp, clean)) =
+        ACTIVE_MULTIPART_SOURCE_HEADS.with_get(source, |entry| (entry.head_stamp, entry.unsettled_parts.is_empty()))
    else {
        return false;
    };
    if !stamp_matches_active_disk(source, head_stamp) {
-        ACTIVE_MULTIPART_SOURCE_HEADS.lock().remove(source);
+        ACTIVE_MULTIPART_SOURCE_HEADS.remove(source);
        return false;
    }
    clean
 }

 pub(super) fn begin_active_multipart_part_transaction(source: &Path, part_number: usize) {
-    if let Some(entry) = ACTIVE_MULTIPART_SOURCE_HEADS.lock().get_mut(source) {
+    ACTIVE_MULTIPART_SOURCE_HEADS.with_get_mut(source, |entry| {
        entry.unsettled_parts.insert(part_number);
        entry.parts.remove(&part_number);
-    }
+    });
 }

 pub(super) fn record_active_multipart_part_committed(source: &Path, part: rustfs_filemeta::ObjectPartInfo) {
    let Some(data_dir) = ACTIVE_MULTIPART_SOURCE_HEADS
-        .lock()
-        .get(source)
-        .and_then(|entry| entry.fi.data_dir)
+        .with_get(source, |entry| entry.fi.data_dir)
+        .flatten()
        .filter(|dir| !dir.is_nil())
    else {
        return;
@@ -362,9 +352,9 @@ pub(super) fn record_active_multipart_part_committed(source: &Path, part: rustfs
    let Some(part_stamp) = FileStamp::from_path(&part_meta_path) else {
        return;
    };
-    if let Some(entry) = ACTIVE_MULTIPART_SOURCE_HEADS.lock().get_mut(source) {
+    ACTIVE_MULTIPART_SOURCE_HEADS.with_get_mut(source, |entry| {
        entry.parts.insert(part.number, (part_stamp, part));
-    }
+    });
 }

 pub(super) fn finish_active_multipart_part_transaction(
@@ -373,34 +363,29 @@ pub(super) fn finish_active_multipart_part_transaction(
    committed: bool,
    retired_existing_txn: bool,
 ) {
-    let mut map = ACTIVE_MULTIPART_SOURCE_HEADS.lock();
    let Some(part_number) = part_number.filter(|_| committed) else {
-        map.remove(source);
+        ACTIVE_MULTIPART_SOURCE_HEADS.remove(source);
        return;
    };
    if !retired_existing_txn {
        return;
    }
-    if let Some(entry) = map.get_mut(source)
-        && !entry.unsettled_parts.remove(&part_number)
-    {
-        map.remove(source);
-    }
+    ACTIVE_MULTIPART_SOURCE_HEADS.remove_if_mut(source, |entry| !entry.unsettled_parts.remove(&part_number));
 }

 pub(super) fn get_active_multipart_source_part(source: &Path, part_number: usize) -> Option<rustfs_filemeta::ObjectPartInfo> {
-    let (head_stamp, data_dir, expected_stamp, part) = {
-        let map = ACTIVE_MULTIPART_SOURCE_HEADS.lock();
-        let entry = map.get(source)?;
-        if !entry.unsettled_parts.is_empty() {
-            return None;
-        }
-        let data_dir = entry.fi.data_dir.filter(|dir| !dir.is_nil())?;
-        let (expected_stamp, part) = entry.parts.get(&part_number)?;
-        (entry.head_stamp, data_dir, *expected_stamp, part.clone())
-    };
+    let (head_stamp, data_dir, expected_stamp, part) = ACTIVE_MULTIPART_SOURCE_HEADS
+        .with_get(source, |entry| {
+            if !entry.unsettled_parts.is_empty() {
+                return None;
+            }
+            let data_dir = entry.fi.data_dir.filter(|dir| !dir.is_nil())?;
+            let (expected_stamp, part) = entry.parts.get(&part_number)?;
+            Some((entry.head_stamp, data_dir, *expected_stamp, part.clone()))
+        })
+        .flatten()?;
    if !stamp_matches_active_disk(source, head_stamp) {
-        ACTIVE_MULTIPART_SOURCE_HEADS.lock().remove(source);
+        ACTIVE_MULTIPART_SOURCE_HEADS.remove(source);
        return None;
    }
    let part_meta_path = source.join(data_dir.to_string()).join(format!("part.{part_number}.meta"));
@@ -411,11 +396,11 @@ pub(super) fn get_active_multipart_source_part(source: &Path, part_number: usize
 }

 pub(super) fn invalidate_active_multipart_source_head(source: &Path) {
-    ACTIVE_MULTIPART_SOURCE_HEADS.lock().remove(source);
+    ACTIVE_MULTIPART_SOURCE_HEADS.remove(source);
 }

 pub(super) fn invalidate_active_multipart_sources_under(prefix: &Path) {
-    ACTIVE_MULTIPART_SOURCE_HEADS.lock().retain(|key, _| !key.starts_with(prefix));
+    ACTIVE_MULTIPART_SOURCE_HEADS.retain_outside_prefix(prefix);
 }

 /// Canonical source mutations are admitted under SourceRW at the syscall,
@@ -1899,6 +1899,172 @@ impl<L> ShardedPathLockRegistry<L> {
    }
 }

+#[cfg(unix)]
+type PathCacheShardMap<V> = HashMap<Box<[u8]>, V, Xxh3SliceBuildHasher>;
+#[cfg(not(unix))]
+type PathCacheShardMap<V> = HashMap<PathBuf, V>;
+
+pub(crate) struct ShardedPathCache<V> {
+    shards: [Mutex<PathCacheShardMap<V>>; PATH_LOCK_SHARDS],
+    max_entries_per_shard: usize,
+}
+
+impl<V> ShardedPathCache<V> {
+    pub(crate) fn new(max_entries_per_shard: usize) -> Self {
+        Self {
+            shards: std::array::from_fn(|_| Mutex::new(HashMap::default())),
+            max_entries_per_shard: max_entries_per_shard.max(1),
+        }
+    }
+
+    #[cfg(unix)]
+    #[inline]
+    fn shard_index(key: &[u8]) -> usize {
+        let hash = xxhash_rust::xxh3::xxh3_64(key);
+        ((hash >> 32) as usize) & (PATH_LOCK_SHARDS - 1)
+    }
+
+    #[cfg(not(unix))]
+    #[inline]
+    fn shard_index(path: &Path) -> usize {
+        use std::hash::{Hash, Hasher};
+        let mut hasher = std::collections::hash_map::DefaultHasher::new();
+        path.hash(&mut hasher);
+        ((hasher.finish() >> 32) as usize) & (PATH_LOCK_SHARDS - 1)
+    }
+
+    #[inline]
+    pub(crate) fn insert(&self, path: &Path, value: V) {
+        #[cfg(unix)]
+        {
+            let key = canonical_unix_path_key(path);
+            let shard_idx = Self::shard_index(key.as_ref());
+            let mut shard = self.shards[shard_idx].lock();
+            if let Some(slot) = shard.get_mut(key.as_ref()) {
+                *slot = value;
+                return;
+            }
+            if shard.len() >= self.max_entries_per_shard
+                && let Some(evict_key) = shard.keys().next().cloned()
+            {
+                shard.remove(&evict_key);
+            }
+            shard.insert(key.into_owned().into_boxed_slice(), value);
+        }
+        #[cfg(not(unix))]
+        {
+            let shard_idx = Self::shard_index(path);
+            let mut shard = self.shards[shard_idx].lock();
+            if let Some(slot) = shard.get_mut(path) {
+                *slot = value;
+                return;
+            }
+            if shard.len() >= self.max_entries_per_shard
+                && let Some(evict_key) = shard.keys().next().cloned()
+            {
+                shard.remove(&evict_key);
+            }
+            shard.insert(path.to_path_buf(), value);
+        }
+    }
+
+    #[inline]
+    pub(crate) fn remove(&self, path: &Path) -> Option<V> {
+        #[cfg(unix)]
+        {
+            let key = canonical_unix_path_key(path);
+            let shard_idx = Self::shard_index(key.as_ref());
+            self.shards[shard_idx].lock().remove(key.as_ref())
+        }
+        #[cfg(not(unix))]
+        {
+            let shard_idx = Self::shard_index(path);
+            self.shards[shard_idx].lock().remove(path)
+        }
+    }
+
+    /// Runs `f` while holding the target shard mutex. `f` must not block, perform I/O, or re-enter `self`.
+    #[inline]
+    pub(crate) fn with_get<R>(&self, path: &Path, f: impl FnOnce(&V) -> R) -> Option<R> {
+        #[cfg(unix)]
+        {
+            let key = canonical_unix_path_key(path);
+            let shard_idx = Self::shard_index(key.as_ref());
+            let shard = self.shards[shard_idx].lock();
+            shard.get(key.as_ref()).map(f)
+        }
+        #[cfg(not(unix))]
+        {
+            let shard_idx = Self::shard_index(path);
+            let shard = self.shards[shard_idx].lock();
+            shard.get(path).map(f)
+        }
+    }
+
+    /// Runs `f` while holding the target shard mutex. `f` must not block, perform I/O, or re-enter `self`.
+    #[inline]
+    pub(crate) fn with_get_mut<R>(&self, path: &Path, f: impl FnOnce(&mut V) -> R) -> Option<R> {
+        #[cfg(unix)]
+        {
+            let key = canonical_unix_path_key(path);
+            let shard_idx = Self::shard_index(key.as_ref());
+            let mut shard = self.shards[shard_idx].lock();
+            shard.get_mut(key.as_ref()).map(f)
+        }
+        #[cfg(not(unix))]
+        {
+            let shard_idx = Self::shard_index(path);
+            let mut shard = self.shards[shard_idx].lock();
+            shard.get_mut(path).map(f)
+        }
+    }
+
+    /// Runs `should_remove` while holding the target shard mutex. `should_remove` must not block, perform I/O, or re-enter `self`.
+    #[inline]
+    pub(crate) fn remove_if_mut(&self, path: &Path, should_remove: impl FnOnce(&mut V) -> bool) {
+        #[cfg(unix)]
+        {
+            let key = canonical_unix_path_key(path);
+            let shard_idx = Self::shard_index(key.as_ref());
+            let mut shard = self.shards[shard_idx].lock();
+            if let Some(entry) = shard.get_mut(key.as_ref())
+                && should_remove(entry)
+            {
+                shard.remove(key.as_ref());
+            }
+        }
+        #[cfg(not(unix))]
+        {
+            let shard_idx = Self::shard_index(path);
+            let mut shard = self.shards[shard_idx].lock();
+            if let Some(entry) = shard.get_mut(path)
+                && should_remove(entry)
+            {
+                shard.remove(path);
+            }
+        }
+    }
+
+    pub(crate) fn retain_outside_prefix(&self, prefix: &Path) {
+        #[cfg(unix)]
+        {
+            let prefix_key = canonical_unix_path_key(prefix);
+            let prefix_bytes = prefix_key.as_ref();
+            for shard in &self.shards {
+                shard
+                    .lock()
+                    .retain(|key, _| !unix_canonical_key_starts_with(key, prefix_bytes));
+            }
+        }
+        #[cfg(not(unix))]
+        {
+            for shard in &self.shards {
+                shard.lock().retain(|key, _| !key.starts_with(prefix));
+            }
+        }
+    }
+}
+
 static FILE_SYNC_PERMITS: LazyLock<Semaphore> = LazyLock::new(|| Semaphore::new(global_file_sync_limit()));
 static DISK_FILE_SYNC_LIMITERS: LazyLock<ShardedPathLockRegistry<Semaphore>> = LazyLock::new(ShardedPathLockRegistry::new);

@@ -12437,6 +12603,59 @@ mod tests {
            );
        }
    }
+
+    #[test]
+    fn sharded_path_cache_normalizes_keys_reuses_slots_and_filters_prefixes() {
+        let cache: ShardedPathCache<usize> = ShardedPathCache::new(2);
+        let clean = Path::new("/tmp/sharded-cache-test/bucket/obj");
+        let non_canonical = Path::new("/tmp/sharded-cache-test//bucket/./obj/.");
+
+        cache.insert(clean, 10);
+        assert_eq!(cache.with_get(non_canonical, |v| *v), Some(10));
+
+        cache.with_get_mut(non_canonical, |v| *v += 5);
+        assert_eq!(cache.with_get(clean, |v| *v), Some(15));
+
+        cache.remove_if_mut(clean, |v| *v == 99);
+        assert_eq!(cache.with_get(clean, |v| *v), Some(15));
+        cache.remove_if_mut(non_canonical, |v| *v == 15);
+        assert_eq!(cache.with_get(clean, |v| *v), None);
+
+        cache.insert(Path::new("/tmp/sharded-cache-test/bucket/a"), 1);
+        cache.insert(Path::new("/tmp/sharded-cache-test/bucket/b"), 2);
+        cache.insert(Path::new("/tmp/sharded-cache-test/bucket-other/c"), 3);
+        cache.retain_outside_prefix(Path::new("/tmp/sharded-cache-test//bucket/."));
+        assert_eq!(cache.with_get(Path::new("/tmp/sharded-cache-test/bucket/a"), |v| *v), None);
+        assert_eq!(cache.with_get(Path::new("/tmp/sharded-cache-test/bucket/b"), |v| *v), None);
+        assert_eq!(cache.with_get(Path::new("/tmp/sharded-cache-test/bucket-other/c"), |v| *v), Some(3));
+
+        for i in 0..512 {
+            cache.insert(&PathBuf::from(format!("/tmp/sharded-cache-test/cap/{i}")), i);
+        }
+        for shard in &cache.shards {
+            assert!(shard.lock().len() <= 2);
+        }
+
+        #[cfg(unix)]
+        {
+            let overwrite_cache: ShardedPathCache<usize> = ShardedPathCache::new(2);
+            let target_shard = ShardedPathCache::<usize>::shard_index(b"/tmp/sharded-cache-test/same-0");
+            let mut same_shard_paths = vec![PathBuf::from("/tmp/sharded-cache-test/same-0")];
+            let mut candidate = 1usize;
+            while same_shard_paths.len() < 2 {
+                let p = PathBuf::from(format!("/tmp/sharded-cache-test/same-{candidate}"));
+                if ShardedPathCache::<usize>::shard_index(canonical_unix_path_key(&p).as_ref()) == target_shard {
+                    same_shard_paths.push(p);
+                }
+                candidate += 1;
+            }
+            overwrite_cache.insert(&same_shard_paths[0], 100);
+            overwrite_cache.insert(&same_shard_paths[1], 200);
+            overwrite_cache.insert(&same_shard_paths[0], 101);
+            assert_eq!(overwrite_cache.with_get(&same_shard_paths[0], |v| *v), Some(101));
+            assert_eq!(overwrite_cache.with_get(&same_shard_paths[1], |v| *v), Some(200));
+        }
+    }
 }

 #[cfg(all(test, target_os = "linux"))]
@@ -128,6 +128,20 @@ impl Error {
        Error::TransientSkip { message: message.into() }
    }

+    /// S3 `SlowDown` ("Please reduce your request rate") is a rate signal.
+    /// Callers must cut their own concurrency instead of retrying at full speed.
+    pub(crate) fn is_slow_down(&self) -> bool {
+        match self {
+            Error::Storage(error) => matches!(error, EcstoreError::SlowDown | EcstoreError::TooManyOpenFiles),
+            Error::Disk(error) => matches!(error, DiskError::TooManyOpenFiles),
+            Error::TransientSkip { message } | Error::Other(message) | Error::TaskExecutionFailed { message } => {
+                message.contains("Please reduce your request rate")
+            }
+            Error::Io(error) => error.kind() == std::io::ErrorKind::WouldBlock,
+            _ => false,
+        }
+    }
+
    /// Whether a heal operation can be retried without changing its inputs.
    pub(crate) fn is_recoverable_heal(&self) -> bool {
        match classify_heal_error(self) {
@@ -449,6 +463,22 @@ mod tests {
    use super::Error;
    use crate::heal::{DiskError, EcstoreError};

+    #[test]
+    fn slow_down_is_a_rate_signal_across_wrapped_shapes() {
+        assert!(Error::Storage(EcstoreError::SlowDown).is_slow_down());
+        assert!(Error::Storage(EcstoreError::TooManyOpenFiles).is_slow_down());
+        assert!(Error::Disk(DiskError::TooManyOpenFiles).is_slow_down());
+        assert!(Error::transient_skip("Please reduce your request rate").is_slow_down());
+        assert!(
+            Error::TaskExecutionFailed {
+                message: "Failed to recreate missing object b/o: Storage error: Please reduce your request rate".to_owned(),
+            }
+            .is_slow_down()
+        );
+        assert!(!Error::Storage(EcstoreError::FileNotFound).is_slow_down());
+        assert!(!Error::transient_skip("DiskNotFound").is_slow_down());
+    }
+
    #[test]
    fn pool_metadata_failures_keep_their_typed_recovery_classification() {
        use crate::heal::storage_api::owner::{EcstorePoolMetadataError, EcstorePoolMetadataFailure};
@@ -14,6 +14,7 @@

 use crate::error::{HealErrorClass, classify_heal_error};
 use crate::heal::{
+    pacing::HealPageRate,
    progress::{HealProgress, add_bytes, increment_counter},
    resume::{
        CheckpointManager, CheckpointObjectOutcome, CheckpointObjectOutcomeRecord, ReplacementTargetIdentity, ResumeManager,
@@ -128,12 +129,12 @@ pub struct ErasureSetHealer {
    replacement_target_identities: Option<Arc<[ReplacementTargetIdentity]>>,
    replacement_execution: Option<Arc<super::storage::ReplacementExecution>>,
    work_parent: Option<super::manager::HealWork>,
-    mainline_pacer: Option<Arc<super::pacing::MainlinePacer>>,
+    mainline_pacer: Option<super::pacing::ForegroundPacing>,
 }

 async fn acquire_page_permit(
    semaphore: Arc<Semaphore>,
-    pacer: Option<&super::pacing::MainlinePacer>,
+    rate: &HealPageRate,
    cancel: &tokio_util::sync::CancellationToken,
 ) -> Result<tokio::sync::OwnedSemaphorePermit> {
    let acquire = || async {
@@ -146,18 +147,16 @@ async fn acquire_page_permit(
    let mut paid_pause = false;
    loop {
        let permit = acquire().await?;
-        if let Some(pacer) = pacer {
-            // Keep the real permit on the low-pressure path. Every acquisition
-            // gets a fresh decision, including a waiter that queued a second
-            // time. One completed pause is a bounded minimum-progress grant.
-            match pacer.admission_decision() {
-                super::pacing::PacingDecision::Wait(pressure) if !paid_pause => {
-                    drop(permit);
-                    paid_pause = pacer.wait_after_admission(cancel, pressure).await?;
-                    continue;
-                }
-                _ => {}
+        // Keep the real permit on the low-pressure path. Every acquisition
+        // gets a fresh decision, including a waiter that queued a second
+        // time. One completed pause is a bounded minimum-progress grant.
+        match rate.admission_decision() {
+            super::pacing::PacingDecision::Wait(pressure) if !paid_pause => {
+                drop(permit);
+                paid_pause = rate.wait_after_admission(cancel, pressure).await?;
+                continue;
            }
+            _ => {}
        }
        return Ok(permit);
    }
@@ -166,7 +165,7 @@ async fn acquire_page_permit(
 #[cfg(test)]
 mod mainline_pacing_tests {
    use super::*;
-    use crate::heal::pacing::{MainlinePacer, TestPressure};
+    use crate::heal::pacing::{ForegroundPacing, HealPageRate, TestPressure};
    use rustfs_concurrency::WorkloadClass;
    use std::sync::atomic::Ordering;
    use tokio_util::sync::CancellationToken;
@@ -180,15 +179,15 @@ mod mainline_pacing_tests {
            .await
            .expect("existing object owns capacity");
        let provider = Arc::new(TestPressure::new(WorkloadClass::ForegroundRead, 0));
-        let pacer = Arc::new(
-            MainlinePacer::new(provider.clone(), 80, 80, Duration::from_millis(250), HealRequestSource::Admin).expect("pacer"),
-        );
+        let pacer =
+            ForegroundPacing::new(provider.clone(), 80, 80, Duration::from_millis(250), HealRequestSource::Admin).expect("pacer");
+        let rate = Arc::new(HealPageRate::new(1, Some(pacer)));
        let cancel = CancellationToken::new();
        let waiting = tokio::spawn({
            let semaphore = semaphore.clone();
-            let pacer = pacer.clone();
+            let rate = rate.clone();
            let cancel = cancel.clone();
-            async move { acquire_page_permit(semaphore, Some(&pacer), &cancel).await }
+            async move { acquire_page_permit(semaphore, &rate, &cancel).await }
        });
        tokio::task::yield_now().await;
        provider.active.store(100, Ordering::SeqCst);
@@ -200,12 +199,13 @@ mod mainline_pacing_tests {
        assert_eq!(semaphore.available_permits(), 1);
        let deadline = tokio::time::timeout(
            Duration::from_millis(10),
-            acquire_page_permit(semaphore.clone(), Some(&pacer), &CancellationToken::new()),
+            acquire_page_permit(semaphore.clone(), &rate, &CancellationToken::new()),
        )
        .await;
        assert!(deadline.is_err());
        assert_eq!(semaphore.available_permits(), 1, "deadline must release all permits");
-        let permit = acquire_page_permit(semaphore.clone(), None, &CancellationToken::new())
+        let unpaced = HealPageRate::new(1, None);
+        let permit = acquire_page_permit(semaphore.clone(), &unpaced, &CancellationToken::new())
            .await
            .expect("unpaced admission");
        assert_eq!(semaphore.available_permits(), 0, "disabling pacing cannot disable the hard cap");
@@ -221,10 +221,11 @@ mod mainline_pacing_tests {
            let occupied = semaphore.clone().acquire_owned().await.expect("queue both waiters");
            let provider = Arc::new(TestPressure::new(WorkloadClass::ForegroundRead, 0));
            let pause = Duration::from_millis(250);
-            let pacer = MainlinePacer::new(provider.clone(), 80, 80, pause, HealRequestSource::Admin).expect("pacer");
+            let pacer = ForegroundPacing::new(provider.clone(), 80, 80, pause, HealRequestSource::Admin).expect("pacer");
+            let rate = HealPageRate::new(1, Some(pacer));
            let cancel = CancellationToken::new();
-            let mut first = Box::pin(acquire_page_permit(semaphore.clone(), Some(&pacer), &cancel));
-            let mut second = Box::pin(acquire_page_permit(semaphore.clone(), Some(&pacer), &cancel));
+            let mut first = Box::pin(acquire_page_permit(semaphore.clone(), &rate, &cancel));
+            let mut second = Box::pin(acquire_page_permit(semaphore.clone(), &rate, &cancel));
            assert!(futures::poll!(first.as_mut()).is_pending());
            assert!(futures::poll!(second.as_mut()).is_pending());
            drop(occupied);
@@ -329,6 +330,20 @@ impl ErasureSetHealer {
        }
    }

+    /// Page fan-out for one whole erasure-set pass (stable across buckets).
+    fn pass_page_concurrency_limit(&self) -> usize {
+        if self.source == HealRequestSource::AutoHeal && self.replacement_task_id.is_some() && self.heal_opts.no_lock {
+            // Versions of the same object must not race without its namespace lock.
+            1
+        } else {
+            Self::effective_heal_page_object_concurrency_for_source(
+                self.source,
+                self.heal_opts.scan_mode,
+                self.replacement_task_id.is_some(),
+            )
+        }
+    }
+
    /// Map the shared heal error class onto this healer's per-object actions.
    ///
    /// `classify_heal_error` is the only classifier; this adapter decides what
@@ -373,7 +388,7 @@ impl ErasureSetHealer {
        }
    }

-    pub(crate) fn with_mainline_pacer(mut self, pacer: Option<Arc<super::pacing::MainlinePacer>>) -> Self {
+    pub(crate) fn with_mainline_pacer(mut self, pacer: Option<super::pacing::ForegroundPacing>) -> Self {
        self.mainline_pacer = pacer;
        self
    }
@@ -464,12 +479,7 @@ impl ErasureSetHealer {
            None
        };
        let buckets = admin_buckets.as_deref().unwrap_or(buckets);
-        if checkpoint_manager
-            .get_checkpoint()
-            .await
-            .admin
-            .is_some_and(|admin| admin.completed)
-        {
+        if checkpoint_manager.admin_completed().await {
            self.initialize_progress(buckets, &resume_manager.get_state().await).await;
            return Ok(());
        }
@@ -490,14 +500,7 @@ impl ErasureSetHealer {
        }

        if self.admin_task.is_some() {
-            let mut admin = checkpoint_manager
-                .get_checkpoint()
-                .await
-                .admin
-                .ok_or_else(|| Error::InvalidCheckpoint("Missing administrator outcome at completion".to_string()))?;
-            admin.completed = true;
-            admin.outcome.finish(None);
-            checkpoint_manager.set_admin(admin).await?;
+            checkpoint_manager.complete_admin().await?;
        } else {
            checkpoint_manager.cleanup().await?;
            resume_manager.cleanup().await?;
@@ -718,7 +721,10 @@ impl ErasureSetHealer {
            }
            return Err(Error::DanglingDeleteDeferred { retry_not_before });
        }
-        let checkpoint = checkpoint_manager.get_checkpoint().await;
+        // One rate controller for the whole pass: SlowDown cool-down and page
+        // fan-out persist across buckets instead of resetting at each one.
+        let page_rate = Arc::new(HealPageRate::new(self.pass_page_concurrency_limit(), self.mainline_pacer.clone()));
+        let (current_bucket_index, current_object_index) = checkpoint_manager.resume_position().await;

        debug!(
            target: "rustfs::heal::erasure_healer",
@@ -726,14 +732,17 @@ impl ErasureSetHealer {
            component = LOG_COMPONENT_HEAL,
            subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
            set_disk_id,
-            current_bucket_index = checkpoint.current_bucket_index,
-            current_object_index = checkpoint.current_object_index,
+            current_bucket_index,
+            current_object_index,
            state = "resuming",
            "Erasure set resumed"
        );

        // 2. initialize progress
        self.initialize_progress(buckets, &state).await;
+        // Snapshot before any baseline write: set_progress_baseline overwrites
+        // the saved generation, which would hide a real generation mismatch.
+        let checkpoint = checkpoint_manager.progress_snapshot().await;
        let (baseline_known, baseline_count, baseline_size, baseline_generation) = {
            let baseline = self.progress.read().await;
            (
@@ -760,17 +769,7 @@ impl ErasureSetHealer {
        let mut successful_objects = state.successful_objects;
        let mut failed_objects = state.failed_objects;
        let mut skipped_objects = state.skipped_objects;
-        let checkpoint_has_progress = checkpoint.baseline_known
-            || checkpoint.successful_objects > 0
-            || checkpoint.failed_object_count > 0
-            || checkpoint.skipped_object_count > 0
-            || checkpoint.skipped_new_versions > 0
-            || checkpoint.skipped_ilm_expired > 0
-            || checkpoint.processed_bytes > 0
-            || checkpoint.total_objects > 0
-            || checkpoint.total_bytes > 0
-            || checkpoint.baseline_generation.is_some()
-            || checkpoint.counter_unknown;
+        let checkpoint_has_progress = checkpoint.has_progress();
        let checkpoint_generation_mismatch = checkpoint.baseline_known && checkpoint.baseline_generation != baseline_generation;
        let mut restored_counter_unknown = state.counter_unknown || checkpoint.counter_unknown;
        if checkpoint_has_progress {
@@ -854,6 +853,7 @@ impl ErasureSetHealer {
                    resume_manager,
                    checkpoint_manager,
                    state.start_time,
+                    &page_rate,
                )
                .await;

@@ -1020,14 +1020,14 @@ impl ErasureSetHealer {
        if failed_objects > 0 || skipped_objects > 0 || failed_buckets > 0 {
            if self.replacement_task_id.is_some() {
                let state = resume_manager.get_state().await;
-                let checkpoint = checkpoint_manager.get_checkpoint().await;
+                let (grace_objects, retry_not_before, counter_unknown) = checkpoint_manager.dangling_delete_grace().await;
                if state.can_retry()
                    && failed_objects == 0
                    && failed_buckets == 0
                    && skipped_objects > 0
-                    && skipped_objects == checkpoint.dangling_delete_grace_objects
-                    && !checkpoint.counter_unknown
-                    && let Some(retry_not_before) = checkpoint.dangling_delete_retry_not_before
+                    && skipped_objects == grace_objects
+                    && !counter_unknown
+                    && let Some(retry_not_before) = retry_not_before
                {
                    // Publish the generation's rewind and wait first. If the
                    // checkpoint reset is interrupted, initialization reapplies
@@ -1167,14 +1167,9 @@ impl ErasureSetHealer {

        let object_key = format!("{RUSTFS_META_BUCKET}/{POOL_META_NAME}");
        let checkpoint_key = compose_key(&object_key, None);
-        let checkpoint = checkpoint_manager.get_checkpoint().await;
-        if checkpoint.processed_objects.contains(&checkpoint_key)
-            || checkpoint.failed_objects.contains(&checkpoint_key)
-            || checkpoint.skipped_objects.contains(&checkpoint_key)
-        {
+        if checkpoint_manager.is_object_settled(&checkpoint_key).await {
            return Ok(());
        }
-        drop(checkpoint);

        self.verify_replacement_identity_fence("pool metadata").await?;
        resume_manager
@@ -1428,7 +1423,7 @@ impl ErasureSetHealer {

    /// Heal one bucket's ordinary objects and report unresolved delete recovery separately.
    #[allow(clippy::too_many_arguments)]
-    #[tracing::instrument(skip(self, current_object_index, processed_objects, successful_objects, failed_objects, skipped_objects, resume_manager, checkpoint_manager), fields(bucket = %bucket, bucket_index = bucket_index))]
+    #[tracing::instrument(skip(self, current_object_index, processed_objects, successful_objects, failed_objects, skipped_objects, resume_manager, checkpoint_manager, page_rate), fields(bucket = %bucket, bucket_index = bucket_index))]
    async fn heal_bucket_with_resume(
        &self,
        bucket: &str,
@@ -1442,6 +1437,7 @@ impl ErasureSetHealer {
        resume_manager: &ResumeManager,
        checkpoint_manager: &CheckpointManager,
        started_at_secs: u64,
+        page_rate: &Arc<HealPageRate>,
    ) -> Result<bool> {
        debug!(
            target: "rustfs::heal::erasure_healer",
@@ -1491,17 +1487,6 @@ impl ErasureSetHealer {
        // detects a non-advancing cursor even though next_marker embeds a
        // transient cache id that changes between identical pages.
        let mut previous_page_last: Option<(String, Option<String>)> = None;
-        let page_concurrency_limit =
-            if self.source == HealRequestSource::AutoHeal && self.replacement_task_id.is_some() && self.heal_opts.no_lock {
-                // Versions of the same object must not race without its namespace lock.
-                1
-            } else {
-                Self::effective_heal_page_object_concurrency_for_source(
-                    self.source,
-                    self.heal_opts.scan_mode,
-                    self.replacement_task_id.is_some(),
-                )
-            };
        let in_flight = Arc::new(AtomicUsize::new(0));
        // Per-bucket sample caps for per-object warn! lines: a flapping rebuild
        // disk can fail/skip hundreds of thousands of versions in one sweep, so
@@ -1523,9 +1508,7 @@ impl ErasureSetHealer {
        let include_lifecycle_object_info = lifecycle_expiry_context.is_some();

        loop {
-            if let Some(pacer) = &self.mainline_pacer {
-                pacer.wait(&self.cancel_token).await?;
-            }
+            page_rate.wait_before_page(&self.cancel_token).await?;
            self.verify_replacement_identity_fence("page scan").await?;
            if let Some(expected) = incarnation {
                self.storage.validate_bucket_incarnation(bucket, Some(expected)).await?;
@@ -1547,9 +1530,8 @@ impl ErasureSetHealer {
                    .await?
            };
            let page_is_empty = objects.is_empty();
-            let checkpoint = checkpoint_manager.get_checkpoint().await;
            let page_resume_index = *current_object_index;
-            let semaphore = Arc::new(Semaphore::new(page_concurrency_limit));
+            let semaphore = Arc::new(Semaphore::new(page_rate.effective_concurrency()));
            let mut page_tasks = FuturesUnordered::new();
            let mut completed_in_page = 0usize;

@@ -1563,10 +1545,7 @@ impl ErasureSetHealer {

                // Per-version dedup identity — the single canonical key.
                let key = compose_key(&item.name, item.version_id.as_deref());
-                if checkpoint.processed_objects.contains(&key)
-                    || checkpoint.failed_objects.contains(&key)
-                    || checkpoint.skipped_objects.contains(&key)
-                {
+                if checkpoint_manager.is_object_settled(&key).await {
                    continue;
                }

@@ -1724,7 +1703,7 @@ impl ErasureSetHealer {
                let target_endpoints = self.target_endpoints.clone();
                let replacement_commit_evidence_required = self.replacement_task_id.is_some();
                let retry_replacement = replacement_commit_evidence_required && self.source == HealRequestSource::AutoHeal;
-                let mainline_pacer = self.mainline_pacer.clone();
+                let page_rate = page_rate.clone();

                let admin_identity = self.admin_identity(bucket, &object_name, version_id.as_deref(), incarnation);
                let admin_task_id = self.admin_task.as_ref().map(|task| task.id.clone());
@@ -1736,14 +1715,15 @@ impl ErasureSetHealer {
                    .map(|_| (dedup_key.clone(), object_name.clone(), version_id.clone()));
                let work = async move {
                    let mut retries = 0;
+                    let mut saw_slow_down = false;
                    let mut admin_identity = admin_identity;
                    let mut admin_task_id = admin_task_id;
                    loop {
-                        let permit = acquire_page_permit(semaphore.clone(), mainline_pacer.as_deref(), &cancel_token).await;
+                        let permit = acquire_page_permit(semaphore.clone(), &page_rate, &cancel_token).await;

                        let permit = match permit {
                            Ok(permit) => permit,
-                            Err(err) => return (dedup_key, object_name, version_id, (0, Err(err)), None),
+                            Err(err) => return (dedup_key, object_name, version_id, (0, Err(err), saw_slow_down), None),
                        };

                        let in_flight_guard = PageConcurrencyGuard::new(in_flight.clone(), set_label.clone());
@@ -1751,7 +1731,8 @@ impl ErasureSetHealer {
                        if let (Some(identity), Some(task_id)) = (admin_identity.take(), admin_task_id.take()) {
                            let (result, canonical) =
                                admin::heal_object(storage.as_ref(), &heal_opts, identity, &task_id, &cancel_token).await;
-                            return (dedup_key, object_name, version_id, result, canonical);
+                            let saw_slow_down = saw_slow_down || matches!(&result.1, Err(err) if err.is_slow_down());
+                            return (dedup_key, object_name, version_id, (result.0, result.1, saw_slow_down), canonical);
                        }

                        // Always go through heal_object. Genuine absence flows through
@@ -1853,18 +1834,23 @@ impl ErasureSetHealer {

                        drop(in_flight_guard);
                        drop(permit);
+                        if matches!(&result.1, Err(err) if err.is_slow_down()) {
+                            saw_slow_down = true;
+                        }
                        if retry_replacement
                            && retries < REPLACEMENT_OBJECT_MAX_RETRIES
                            && matches!(&result.1, Err(Error::TransientSkip { .. }))
                        {
                            retries += 1;
                            tokio::select! {
-                                _ = cancel_token.cancelled() => return (dedup_key, object_name, version_id, (0, Err(Error::TaskCancelled)), None),
+                                _ = cancel_token.cancelled() => {
+                                    return (dedup_key, object_name, version_id, (0, Err(Error::TaskCancelled), saw_slow_down), None);
+                                }
                                _ = tokio::time::sleep(Duration::from_millis(250 * u64::from(retries))) => {}
                            }
                            continue;
                        }
-                        return (dedup_key, object_name, version_id, result, None);
+                        return (dedup_key, object_name, version_id, (result.0, result.1, saw_slow_down), None);
                    }
                };
                page_tasks.push(async move {
@@ -1875,7 +1861,7 @@ impl ErasureSetHealer {
                        };
                        match result {
                            Ok(result) => result,
-                            Err(error) => (key, object, version, (0, Err(error)), None),
+                            Err(error) => (key, object, version, (0, Err(error), false), None),
                        }
                    } else {
                        work.await
@@ -1884,7 +1870,16 @@ impl ErasureSetHealer {
            }

            while let Some((key, object, version_id, result, canonical)) = page_tasks.next().await {
-                let (object_size, result) = result;
+                let (object_size, result, saw_slow_down) = result;
+                // Exactly one rate observation per finished object, including
+                // SlowDown that later succeeded on a replacement retry.
+                page_rate.observe_object_outcome(
+                    &match &result {
+                        Ok(_) => Ok(true),
+                        Err(err) => Err(err),
+                    },
+                    saw_slow_down,
+                );
                let mut telemetry_unknown = false;
                let checkpoint_outcome = match result {
                    Ok(true) => {
@@ -2482,6 +2477,7 @@ mod resume_loop_tests {
    use super::{
        ErasureSetHealer, NANOS_PER_SECOND, NEW_VERSION_SKIP_GRACE_SECS, should_skip_new_version, target_outcomes_complete,
    };
+    use crate::heal::pacing::HealPageRate;
    use crate::heal::progress::HealProgress;
    use crate::heal::resume::{
        CheckpointManager, CheckpointObjectOutcome, CheckpointObjectOutcomeRecord, RESUME_CHECKPOINT_FILE,
@@ -3063,6 +3059,7 @@ mod resume_loop_tests {
        let mut successful = 0u64;
        let mut failed = 0u64;
        let mut skipped = 0u64;
+        let page_rate = Arc::new(HealPageRate::new(env.healer.pass_page_concurrency_limit(), None));
        let result = env
            .healer
            .heal_bucket_with_resume(
@@ -3077,6 +3074,7 @@ mod resume_loop_tests {
                &env.resume,
                &env.checkpoint,
                state.start_time,
+                &page_rate,
            )
            .await;
        (processed, successful, failed, skipped, result.map(|_| ()))
@@ -3144,6 +3142,7 @@ mod resume_loop_tests {
        let mut skipped = 0;
        let started_at = env.resume.get_state().await.start_time;

+        let page_rate = Arc::new(HealPageRate::new(healer.pass_page_concurrency_limit(), None));
        let error = healer
            .heal_bucket_with_resume(
                "b",
@@ -3157,6 +3156,7 @@ mod resume_loop_tests {
                &env.resume,
                &env.checkpoint,
                started_at,
+                &page_rate,
            )
            .await
            .expect_err("a remounted target must not begin a new page scan");
@@ -34,7 +34,7 @@ pub(super) fn running_mainline_pacer(
    request: &HealRequest,
    config: &HealConfig,
    provider: &Option<WorkloadSnapshotProviderRef>,
-) -> Option<Arc<crate::heal::pacing::MainlinePacer>> {
+) -> Option<crate::heal::pacing::ForegroundPacing> {
    let pace_running = request.source == HealRequestSource::Admin
        || (request.source == HealRequestSource::AutoHeal
            && matches!(&request.heal_type, HealType::ErasureSet { .. })
@@ -43,14 +43,13 @@ pub(super) fn running_mainline_pacer(
        return None;
    }
    provider.as_ref().and_then(|provider| {
-        crate::heal::pacing::MainlinePacer::new(
+        crate::heal::pacing::ForegroundPacing::new(
            provider.clone(),
            config.mainline_read_utilization_high_percent,
            config.mainline_write_utilization_high_percent,
            config.mainline_max_sleep,
            request.source,
        )
-        .map(Arc::new)
    })
 }

@@ -7,8 +7,14 @@ use rustfs_concurrency::{
    workload::{ForegroundPressure, foreground_pressure},
 };
 use rustfs_heal_contracts::heal_channel::HealRequestSource;
-use std::{sync::Arc, time::Duration};
-use tokio::{sync::Mutex, time::Instant};
+use std::{
+    sync::{
+        Arc,
+        atomic::{AtomicU32, Ordering},
+    },
+    time::Duration,
+};
+use tokio::{sync::Mutex as AsyncMutex, time::Instant};
 use tokio_util::sync::CancellationToken;

 #[derive(Default)]
@@ -22,17 +28,19 @@ pub(crate) enum PacingDecision {
    Wait(Option<ForegroundPressure>),
 }

-/// Cooperative pacing for one heal execution, not a storage admission permit.
-pub(crate) struct MainlinePacer {
+/// Foreground watermarks for one heal execution. Constructed once per request;
+/// disabled when watermarks are zero or no workload provider is configured.
+#[derive(Clone)]
+pub(crate) struct ForegroundPacing {
    provider: Arc<dyn WorkloadAdmissionSnapshotProvider + Send + Sync>,
    read_high: usize,
    write_high: usize,
    pause: Duration,
    source: HealRequestSource,
-    state: Mutex<PacingState>,
+    state: Arc<AsyncMutex<PacingState>>,
 }

-impl MainlinePacer {
+impl ForegroundPacing {
    pub(crate) fn new(
        provider: Arc<dyn WorkloadAdmissionSnapshotProvider + Send + Sync>,
        read_high: usize,
@@ -49,7 +57,7 @@ impl MainlinePacer {
            write_high: write_high.min(100),
            pause: pause.min(Duration::from_secs(1)),
            source,
-            state: Mutex::new(PacingState::default()),
+            state: Arc::new(AsyncMutex::new(PacingState::default())),
        })
    }

@@ -125,6 +133,142 @@ impl MainlinePacer {
    }
 }

+/// Single owner of "how hard may this heal pass push storage?".
+///
+/// One instance spans the whole erasure-set scan (all buckets and pages):
+/// - foreground admission pressure (`ForegroundPacing`)
+/// - S3 `SlowDown` ("Please reduce your request rate") self-backpressure
+///
+/// Call `observe_object_outcome` exactly once per finished object so a single
+/// SlowDown cannot double-reduce concurrency.
+pub(crate) struct HealPageRate {
+    base_concurrency: usize,
+    /// 0 = full base concurrency; each SlowDown increases the reduction.
+    slow_down_step: AtomicU32,
+    slow_down_until: std::sync::Mutex<Option<Instant>>,
+    foreground: Option<ForegroundPacing>,
+}
+
+impl HealPageRate {
+    /// Always-on controller. Foreground yield is optional; SlowDown control is not.
+    pub(crate) fn new(base_concurrency: usize, foreground: Option<ForegroundPacing>) -> Self {
+        Self {
+            base_concurrency: base_concurrency.max(1),
+            slow_down_step: AtomicU32::new(0),
+            slow_down_until: std::sync::Mutex::new(None),
+            foreground,
+        }
+    }
+
+    /// Concurrency after applying the current SlowDown reduction. Always ≥ 1
+    /// so maintenance can still make progress under sustained rate pressure.
+    pub(crate) fn effective_concurrency(&self) -> usize {
+        let step = self.slow_down_step.load(Ordering::Relaxed).min(8);
+        (self.base_concurrency >> step).max(1)
+    }
+
+    /// Fold one finished object into the rate state. Call once per outcome.
+    /// `saw_slow_down` is true when any attempt of this object hit SlowDown,
+    /// including attempts that later succeeded on retry.
+    pub(crate) fn observe_object_outcome(&self, result: &std::result::Result<bool, &Error>, saw_slow_down: bool) {
+        if saw_slow_down || matches!(result, Err(err) if err.is_slow_down()) {
+            self.observe_slow_down();
+            return;
+        }
+        if result.is_ok() {
+            self.observe_success();
+        }
+    }
+
+    /// S3 `SlowDown` is a rate signal: cut fan-out and arm exponential cool-down.
+    fn observe_slow_down(&self) {
+        // fetch_add returns the previous step; +1 so the first SlowDown uses
+        // exponent 1 (2s), not exponent 0 (1s).
+        let step = self.slow_down_step.fetch_add(1, Ordering::Relaxed).saturating_add(1).min(8);
+        let delay = Self::backoff(step);
+        let ready = Instant::now() + delay;
+        if let Ok(mut until) = self.slow_down_until.lock()
+            && until.is_none_or(|current| current < ready)
+        {
+            *until = Some(ready);
+        }
+    }
+
+    /// Restore one concurrency step only after the SlowDown cool-down expires.
+    /// A same-page success must not undo the reduction while the pass is still
+    /// in the rate-limited window (the page semaphore is fixed until the next page).
+    fn observe_success(&self) {
+        let cooling = self
+            .slow_down_until
+            .lock()
+            .ok()
+            .and_then(|until| *until)
+            .is_some_and(|ready| ready > Instant::now());
+        if cooling {
+            return;
+        }
+        self.slow_down_step
+            .try_update(Ordering::Relaxed, Ordering::Relaxed, |step| Some(step.saturating_sub(1)))
+            .ok();
+    }
+
+    /// Wait out SlowDown cool-down, then the foreground pacer. Cancellation interrupts.
+    pub(crate) async fn wait_before_page(&self, cancel: &CancellationToken) -> Result<()> {
+        self.wait_slow_down(cancel).await?;
+        if let Some(foreground) = &self.foreground {
+            foreground.wait(cancel).await?;
+        }
+        Ok(())
+    }
+
+    /// Foreground admission decision while the caller owns page capacity.
+    pub(crate) fn admission_decision(&self) -> PacingDecision {
+        match &self.foreground {
+            Some(foreground) => foreground.admission_decision(),
+            None => PacingDecision::Ready,
+        }
+    }
+
+    /// Foreground wait after a rejected admission. Returns whether a pause was paid.
+    pub(crate) async fn wait_after_admission(
+        &self,
+        cancel: &CancellationToken,
+        observed: Option<ForegroundPressure>,
+    ) -> Result<bool> {
+        match &self.foreground {
+            Some(foreground) => foreground.wait_after_admission(cancel, observed).await,
+            None => Ok(false),
+        }
+    }
+
+    async fn wait_slow_down(&self, cancel: &CancellationToken) -> Result<()> {
+        let delay = self
+            .slow_down_until
+            .lock()
+            .ok()
+            .and_then(|until| *until)
+            .map(|ready| ready.saturating_duration_since(Instant::now()));
+        let Some(delay) = delay else {
+            return Ok(());
+        };
+        if delay.is_zero() {
+            return Ok(());
+        }
+        tokio::select! {
+            biased;
+            _ = cancel.cancelled() => Err(Error::TaskCancelled),
+            _ = tokio::time::sleep(delay) => Ok(()),
+        }
+    }
+
+    fn backoff(step: u32) -> Duration {
+        // step 1 => 2s, 2 => 4s, … capped at 30s — S3 clients are told to reduce
+        // rate with exponential backoff, and heal must do the same to itself.
+        let exponent = step.clamp(1, 5);
+        Duration::from_secs(2_u64.saturating_pow(exponent)).min(Duration::from_secs(30))
+    }
+}
+
 #[cfg(test)]
 pub(crate) struct TestPressure {
    pub(crate) active: std::sync::atomic::AtomicUsize,
@@ -164,12 +308,87 @@ mod tests {
    use rustfs_concurrency::WorkloadClass;
    use std::sync::atomic::Ordering;

+    #[tokio::test(start_paused = true)]
+    async fn heal_page_rate_first_slow_down_waits_two_seconds() {
+        let rate = HealPageRate::new(8, None);
+        rate.observe_object_outcome(&Err(&Error::transient_skip("Please reduce your request rate")), false);
+        let cancel = CancellationToken::new();
+        let before = Instant::now();
+        rate.wait_before_page(&cancel).await.expect("cool-down wait");
+        assert_eq!(Instant::now() - before, Duration::from_secs(2));
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn heal_page_rate_reduces_concurrency_and_backs_off() {
+        let rate = HealPageRate::new(8, None);
+        assert_eq!(rate.effective_concurrency(), 8);
+        rate.observe_object_outcome(&Err(&Error::transient_skip("Please reduce your request rate")), false);
+        assert_eq!(rate.effective_concurrency(), 4);
+        rate.observe_object_outcome(&Err(&Error::Storage(crate::heal::EcstoreError::SlowDown)), true);
+        assert_eq!(rate.effective_concurrency(), 2);
+        // Floor is one so maintenance never stops entirely.
+        for _ in 0..8 {
+            rate.observe_object_outcome(&Err(&Error::Storage(crate::heal::EcstoreError::SlowDown)), true);
+        }
+        assert_eq!(rate.effective_concurrency(), 1);
+
+        let cancel = CancellationToken::new();
+        assert!(rate.wait_before_page(&cancel).await.is_ok());
+
+        // After cool-down, successes restore concurrency one step at a time.
+        for _ in 0..12 {
+            rate.observe_object_outcome(&Ok(true), false);
+        }
+        assert!(rate.effective_concurrency() > 1);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn heal_page_rate_same_page_success_does_not_undo_slow_down() {
+        let rate = HealPageRate::new(8, None);
+        rate.observe_object_outcome(&Err(&Error::Storage(crate::heal::EcstoreError::SlowDown)), false);
+        assert_eq!(rate.effective_concurrency(), 4);
+        // Same-page successes while cool-down is armed must not restore fan-out.
+        for _ in 0..4 {
+            rate.observe_object_outcome(&Ok(true), false);
+        }
+        assert_eq!(rate.effective_concurrency(), 4);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn heal_page_rate_slow_down_then_retry_success_still_reduces() {
+        let rate = HealPageRate::new(8, None);
+        // Replacement retry: SlowDown on attempt 1, success on attempt 2.
+        rate.observe_object_outcome(&Ok(true), true);
+        assert_eq!(rate.effective_concurrency(), 4);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn heal_page_rate_wait_is_cancellable() {
+        let rate = HealPageRate::new(4, None);
+        rate.observe_object_outcome(&Err(&Error::Storage(crate::heal::EcstoreError::SlowDown)), false);
+        let cancel = CancellationToken::new();
+        cancel.cancel();
+        assert!(matches!(rate.wait_before_page(&cancel).await, Err(Error::TaskCancelled)));
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn heal_page_rate_still_waits_for_foreground_after_slow_down_clears() {
+        let provider = Arc::new(TestPressure::new(WorkloadClass::ForegroundWrite, 100));
+        let foreground = ForegroundPacing::new(provider, 0, 80, Duration::from_millis(250), HealRequestSource::AutoHeal)
+            .expect("foreground pacer");
+        let rate = HealPageRate::new(4, Some(foreground));
+        let cancel = CancellationToken::new();
+        let before = Instant::now();
+        rate.wait_before_page(&cancel).await.expect("bounded foreground pause");
+        assert_eq!(Instant::now() - before, Duration::from_millis(250));
+    }
+
    #[tokio::test(start_paused = true)]
    async fn running_mainline_hysteresis_uses_configured_watermarks_and_stable_low_window() {
        for class in [WorkloadClass::ForegroundRead, WorkloadClass::ForegroundWrite] {
            let provider = Arc::new(TestPressure::new(class, 0));
            let pause = Duration::from_millis(250);
-            let pacer = MainlinePacer::new(
+            let pacer = ForegroundPacing::new(
                provider.clone(),
                if class == WorkloadClass::ForegroundRead { 40 } else { 0 },
                if class == WorkloadClass::ForegroundWrite { 40 } else { 0 },
@@ -197,9 +416,9 @@ mod tests {
    #[tokio::test(start_paused = true)]
    async fn running_mainline_huge_pause_is_capped_and_disabled_classes_do_not_sleep() {
        let provider = Arc::new(TestPressure::new(WorkloadClass::ForegroundRead, 100));
-        assert!(MainlinePacer::new(provider.clone(), 0, 0, Duration::from_secs(1), HealRequestSource::Admin).is_none());
-        assert!(MainlinePacer::new(provider.clone(), 80, 80, Duration::ZERO, HealRequestSource::Admin).is_none());
-        let pacer = MainlinePacer::new(provider, 80, 80, Duration::from_secs(3600), HealRequestSource::Admin).expect("pacer");
+        assert!(ForegroundPacing::new(provider.clone(), 0, 0, Duration::from_secs(1), HealRequestSource::Admin).is_none());
+        assert!(ForegroundPacing::new(provider.clone(), 80, 80, Duration::ZERO, HealRequestSource::Admin).is_none());
+        let pacer = ForegroundPacing::new(provider, 80, 80, Duration::from_secs(3600), HealRequestSource::Admin).expect("pacer");
        let before = Instant::now();
        pacer.wait(&CancellationToken::new()).await.expect("hard-capped pause");
        assert_eq!(Instant::now() - before, Duration::from_secs(1));
@@ -209,7 +428,7 @@ mod tests {
    async fn running_mainline_waiters_cancel_and_task_latches_are_isolated() {
        let provider = Arc::new(TestPressure::new(WorkloadClass::ForegroundRead, 100));
        let paced = Arc::new(
-            MainlinePacer::new(provider.clone(), 80, 80, Duration::from_secs(1), HealRequestSource::Admin).expect("pacer"),
+            ForegroundPacing::new(provider.clone(), 80, 80, Duration::from_secs(1), HealRequestSource::Admin).expect("pacer"),
        );
        let cancel_first = CancellationToken::new();
        let first = tokio::spawn({
@@ -228,7 +447,7 @@ mod tests {
        cancel_second.cancel();
        assert!(matches!(second.await.expect("queued waiter"), Err(Error::TaskCancelled)));
        provider.active.store(0, Ordering::SeqCst);
-        let other_task = MainlinePacer::new(provider.clone(), 80, 80, Duration::from_secs(1), HealRequestSource::Admin)
+        let other_task = ForegroundPacing::new(provider.clone(), 80, 80, Duration::from_secs(1), HealRequestSource::Admin)
            .expect("independent task");
        let before = Instant::now();
        other_task
@@ -35,7 +35,9 @@ mod replacement;
 mod utils;

 pub(crate) use checkpoint::AdminErasureCheckpoint;
-pub use checkpoint::{CheckpointManager, CheckpointObjectOutcome, CheckpointObjectOutcomeRecord, ResumeCheckpoint};
+pub use checkpoint::{
+    CheckpointManager, CheckpointObjectOutcome, CheckpointObjectOutcomeRecord, CheckpointProgressSnapshot, ResumeCheckpoint,
+};
 pub(crate) use gc::ResumeGc;
 pub use handoff::{ReplacementHandoff, ReplacementHandoffLink, ReplacementHandoffPhase};
 pub use legacy_handoff::{LegacyReplacementApproval, LegacyReplacementSource};
@@ -152,6 +152,42 @@ impl AdminErasureCheckpoint {
    }
 }

+/// Lightweight progress view for resume decisions. Omits the per-page object
+/// ledgers so callers do not clone a million-object checkpoint just to read
+/// counters and positions.
+#[derive(Debug, Clone, Copy)]
+pub struct CheckpointProgressSnapshot {
+    pub current_bucket_index: usize,
+    pub current_object_index: usize,
+    pub successful_objects: u64,
+    pub failed_object_count: u64,
+    pub skipped_object_count: u64,
+    pub skipped_new_versions: u64,
+    pub skipped_ilm_expired: u64,
+    pub processed_bytes: u64,
+    pub total_objects: u64,
+    pub total_bytes: u64,
+    pub baseline_generation: Option<u64>,
+    pub baseline_known: bool,
+    pub counter_unknown: bool,
+}
+
+impl CheckpointProgressSnapshot {
+    pub fn has_progress(&self) -> bool {
+        self.baseline_known
+            || self.successful_objects > 0
+            || self.failed_object_count > 0
+            || self.skipped_object_count > 0
+            || self.skipped_new_versions > 0
+            || self.skipped_ilm_expired > 0
+            || self.processed_bytes > 0
+            || self.total_objects > 0
+            || self.total_bytes > 0
+            || self.baseline_generation.is_some()
+            || self.counter_unknown
+    }
+}
+
 /// resume checkpoint
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct ResumeCheckpoint {
@@ -715,6 +751,75 @@ impl CheckpointManager {
        self.checkpoint.read().await.clone()
    }

+    /// True when a page object is already settled in the current ledger.
+    /// Membership is checked under the lock without cloning the checkpoint,
+    /// which used to copy every page's dedup sets on the small-object path.
+    pub async fn is_object_settled(&self, key: &str) -> bool {
+        let checkpoint = self.checkpoint.read().await;
+        checkpoint.processed_objects.contains(key)
+            || checkpoint.failed_objects.contains(key)
+            || checkpoint.skipped_objects.contains(key)
+    }
+
+    /// True when the administrator checkpoint already finished.
+    pub async fn admin_completed(&self) -> bool {
+        self.checkpoint
+            .read()
+            .await
+            .admin
+            .as_ref()
+            .is_some_and(|admin| admin.completed)
+    }
+
+    /// Current resume position without cloning the object ledgers.
+    pub async fn resume_position(&self) -> (usize, usize) {
+        let checkpoint = self.checkpoint.read().await;
+        (checkpoint.current_bucket_index, checkpoint.current_object_index)
+    }
+
+    /// Dangling-delete grace summary without cloning the object ledgers.
+    pub async fn dangling_delete_grace(&self) -> (u64, Option<u64>, bool) {
+        let checkpoint = self.checkpoint.read().await;
+        (
+            checkpoint.dangling_delete_grace_objects,
+            checkpoint.dangling_delete_retry_not_before,
+            checkpoint.counter_unknown,
+        )
+    }
+
+    /// Aggregate progress counters without cloning the page dedup sets.
+    pub async fn progress_snapshot(&self) -> CheckpointProgressSnapshot {
+        let checkpoint = self.checkpoint.read().await;
+        CheckpointProgressSnapshot {
+            current_bucket_index: checkpoint.current_bucket_index,
+            current_object_index: checkpoint.current_object_index,
+            successful_objects: checkpoint.successful_objects,
+            failed_object_count: checkpoint.failed_object_count,
+            skipped_object_count: checkpoint.skipped_object_count,
+            skipped_new_versions: checkpoint.skipped_new_versions,
+            skipped_ilm_expired: checkpoint.skipped_ilm_expired,
+            processed_bytes: checkpoint.processed_bytes,
+            total_objects: checkpoint.total_objects,
+            total_bytes: checkpoint.total_bytes,
+            baseline_generation: checkpoint.baseline_generation,
+            baseline_known: checkpoint.baseline_known,
+            counter_unknown: checkpoint.counter_unknown,
+        }
+    }
+
+    /// Finish the administrator outcome in place and persist it.
+    pub async fn complete_admin(&self) -> Result<()> {
+        let mut checkpoint = self.checkpoint.write().await;
+        let admin = checkpoint
+            .admin
+            .as_mut()
+            .ok_or_else(|| Error::InvalidCheckpoint("Missing administrator outcome at completion".to_string()))?;
+        admin.completed = true;
+        admin.outcome.finish(None);
+        drop(checkpoint);
+        self.save_checkpoint().await
+    }
+
    /// update position
    pub async fn update_position(&self, bucket_index: usize, object_index: usize) -> Result<()> {
        let mut checkpoint = self.checkpoint.write().await;
@@ -485,7 +485,7 @@ pub struct HealTask {
    pub cancel_token: tokio_util::sync::CancellationToken,
    /// Storage layer interface
    pub storage: Arc<dyn HealStorageAPI>,
-    mainline_pacer: Option<Arc<super::pacing::MainlinePacer>>,
+    mainline_pacer: Option<super::pacing::ForegroundPacing>,
 }

 impl HealTask {
@@ -586,7 +586,7 @@ impl HealTask {
        task
    }

-    pub(crate) fn with_mainline_pacer(mut self, pacer: Option<Arc<super::pacing::MainlinePacer>>) -> Self {
+    pub(crate) fn with_mainline_pacer(mut self, pacer: Option<super::pacing::ForegroundPacing>) -> Self {
        self.mainline_pacer = pacer;
        self
    }
@@ -131,7 +131,10 @@ persist until the refresh settles.
 Control-plane retry begins on the 30-second drain, while heavyweight snapshots,
 pending topology operations, and bucket wiring are revisited on the 600-second
 pass. Object MRF entries are persisted every 10 seconds by default and target
-health is probed every 5 seconds. These are scheduling bounds, not delivery
+health is probed every 5 seconds. The persisted MRF journal is replayed at
+startup and then once every 5 minutes across the cluster, so a delete-marker purge or an overflowed
+replication whose target was down converges within one replay interval after
+the target returns. These are scheduling bounds, not delivery
 SLAs: network timeouts and the amount of queued work add to them.

 Objects that must be rediscovered by the scanner have this conservative upper
@@ -4522,45 +4522,6 @@ mod tests {
        assert!(options.ignore_errors);
    }

-    #[test]
-    fn resolve_put_object_extract_options_sees_hoisted_query_controls() {
-        // The AWS SDK presigner moves `x-amz-meta-snowball-*` into the query
-        // string. The request-hoist layer restores them as headers, and the
-        // extraction options must resolve from them before member headers drop
-        // the transport keys — otherwise members land under their bare names
-        // and can overwrite unrelated objects (rustfs/rustfs#9146).
-        let mut query = String::from("X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=AKID&X-Amz-Signature=deadbeef");
-        query.push_str("&x-amz-meta-snowball-auto-extract=true&x-amz-meta-snowball-prefix=batch/");
-        query.push_str("&x-amz-meta-snowball-ignore-dirs=true&x-amz-meta-minio-snowball-ignore-errors=true");
-        let mut req = http::Request::builder()
-            .method(http::Method::PUT)
-            .uri(format!("/bucket/key?{query}"))
-            .body(())
-            .expect("request");
-        crate::server::hoist_signed_query_params(&mut req).expect("hoist");
-
-        assert!(is_put_object_extract_requested(req.headers()));
-        let options = resolve_put_object_extract_options(req.headers()).unwrap();
-        assert_eq!(options.prefix.as_deref(), Some("batch"));
-        assert!(options.ignore_dirs);
-        assert!(options.ignore_errors);
-
-        // Member headers keep the archive's own metadata and drop the transport
-        // keys, so no member object stores the controls that drove extraction.
-        let member_headers = snowball_member_headers(req.headers());
-        assert!(!member_headers.contains_key(AMZ_SNOWBALL_EXTRACT));
-        assert!(!member_headers.contains_key("x-amz-meta-snowball-prefix"));
-    }
-
-    #[test]
-    fn extract_request_discovery_honors_an_explicit_false_header_over_query() {
-        // Header precedence: `x-amz-meta-snowball-auto-extract: false` must not
-        // fall through to a hoisted `true` of the same key.
-        let mut headers = HeaderMap::new();
-        headers.insert(AMZ_SNOWBALL_EXTRACT, HeaderValue::from_static("false"));
-        assert!(!is_put_object_extract_requested(&headers));
-    }
-
    #[test]
    fn resolve_put_object_extract_options_accepts_standard_headers() {
        let mut headers = HeaderMap::new();
@@ -972,58 +972,6 @@ fn is_request_signature_v2(header: &HeaderMap) -> bool {
    false
 }

-/// True when the request's query string is covered by its own signature, so a
-/// query parameter may be read as an authenticated request property.
-///
-/// Every SigV4 variant signs the canonical query string (AWS General Reference,
-/// "Create a signed AWS API request"; s3s-sigv4 verifies every parameter except
-/// the signature itself), so a SigV4 request's query parameters are as
-/// trustworthy as its headers — the AWS SDK presigners rely on this when they
-/// hoist `x-amz-*` headers into the query (`@smithy/signature-v4`
-/// `moveHeadersToQuery`).
-///
-/// SigV2 is ruled out first and must stay ruled out. Its string-to-sign covers
-/// headers plus a fixed subresource list rather than arbitrary query parameters,
-/// so a SigV2 presigned URL keeps validating after `X-Amz-Algorithm` and
-/// `X-Amz-Signature` markers are appended to it. Those markers would otherwise
-/// talk this predicate into trusting a query the SigV2 signature never covered,
-/// which is how a URL holder could turn on `x-amz-meta-snowball-auto-extract`
-/// and have a plain upload untarred.
-///
-/// After that, the request must carry SigV4 the way the verifier authenticates
-/// it: an `Authorization: AWS4-HMAC-SHA256` header (all header-signed variants,
-/// streaming included), or the complete presign marker set in the query. A bare
-/// `x-amz-credential` header — what
-/// [`is_request_presigned_signature_v4_with_query`] falls back on — is not
-/// proof of presigning and does not qualify. A forged marker set on a request
-/// with no signature at all (anonymous, JWT) is accepted here and is not a
-/// widening: such a request may already send the same `x-amz-*` headers, and
-/// the hoisted form is checked into the same place. It does not survive a
-/// verifier that rejects it either.
-///
-/// [`is_request_presigned_signature_v4_with_query`]: crate::auth::is_request_presigned_signature_v4_with_query
-pub(crate) fn query_string_is_signed(header: &HeaderMap, query: &str) -> bool {
-    if is_request_signature_v2(header) || is_request_presigned_signature_v2(header, Some(query)) {
-        return false;
-    }
-    if is_request_signature_v4(header) {
-        return true;
-    }
-    let mut algorithm = false;
-    let mut credential = false;
-    let mut signature = false;
-    for (name, value) in form_urlencoded::parse(query.as_bytes()) {
-        if name.eq_ignore_ascii_case("x-amz-algorithm") {
-            algorithm = value == SIGN_V4_ALGORITHM;
-        } else if name.eq_ignore_ascii_case("x-amz-credential") {
-            credential = !value.is_empty();
-        } else if name.eq_ignore_ascii_case("x-amz-signature") {
-            signature = !value.is_empty();
-        }
-    }
-    algorithm && credential && signature
-}
-
 pub(crate) fn is_request_presigned_signature_v4_with_query(header: &HeaderMap, query: Option<&str>) -> bool {
    if let Some(credential) = header.get(AMZ_CREDENTIAL) {
        return !credential.to_str().unwrap_or("").is_empty();
@@ -1296,47 +1244,22 @@ pub(crate) fn reject_unsigned_amz_headers_on_presigned_request(header: &HeaderMa
        if !name.starts_with("x-amz-") || PRESIGNED_UNSIGNED_AMZ_HEADER_ALLOWLIST.contains(&name) {
            continue;
        }
-        if signed.iter().any(|signed_name| signed_name == name) {
-            continue;
+        if !signed.iter().any(|signed_name| signed_name == name) {
+            warn!(
+                event = EVENT_PRESIGNED_UNSIGNED_AMZ_HEADER,
+                component = LOG_COMPONENT_AUTH,
+                subsystem = LOG_SUBSYSTEM_REQUEST,
+                reason = "unsigned_amz_header",
+                header = name,
+                "Presigned request rejected"
+            );
+            return Err(S3Error::with_message(S3ErrorCode::AccessDenied, UNSIGNED_HEADERS_MESSAGE.to_string()));
        }
-        // The query-hoist layer restores a signed `x-amz-*` query parameter as
-        // a header before this rule runs a second time at the access boundary.
-        // Such a header carries exactly the value the signature covers, so it
-        // is as signed as the parameter itself; a header the signer did not
-        // sign cannot match one.
-        if header_as_signed_query_value(header, query, name) {
-            continue;
-        }
-        warn!(
-            event = EVENT_PRESIGNED_UNSIGNED_AMZ_HEADER,
-            component = LOG_COMPONENT_AUTH,
-            subsystem = LOG_SUBSYSTEM_REQUEST,
-            reason = "unsigned_amz_header",
-            header = name,
-            "Presigned request rejected"
-        );
-        return Err(S3Error::with_message(S3ErrorCode::AccessDenied, UNSIGNED_HEADERS_MESSAGE.to_string()));
    }

    Ok(())
 }

-/// True when `name` is a single-valued header whose value is exactly the value
-/// of a same-named query parameter.
-///
-/// The SigV4 canonical query string covers every parameter except the signature
-/// itself, so a match proves the value the header carries is authorized. Only
-/// a single-valued header can match: the query-hoist layer never produces a
-/// multi-valued one, and a second value would be unsigned.
-fn header_as_signed_query_value(header: &HeaderMap, query: &str, name: &str) -> bool {
-    let mut values = header.get_all(name).iter();
-    let (Some(value), None) = (values.next(), values.next()) else {
-        return false;
-    };
-    form_urlencoded::parse(query.as_bytes())
-        .any(|(param, param_value)| param.eq_ignore_ascii_case(name) && param_value.as_bytes() == value.as_bytes())
-}
-
 /// Parse the RustFS presigned PutObject size capability after authentication.
 ///
 /// The query value is covered by SigV4 when it is present before presigning, but
@@ -26,8 +26,7 @@ use crate::server::{
        BodylessStatusFixLayer, ConditionalCorsLayer, DoubleSlashListBucketsCompatLayer, EmptyBodyContentLengthCompatLayer,
        ExternalRequestContextLayer, HeadRequestBodyFixLayer, IcebergRestErrorCompatLayer, ObjectAttributesEtagFixLayer,
        PublicHealthEndpointLayer, RedirectLayer, RequestContextLayer, RequestLoggingLayer, S3ErrorMessageCompatLayer,
-        SigV4HeaderGuardLayer, SignedQueryHoistLayer, StsQueryApiCompatLayer, VirtualHostStyleHintLayer,
-        redact_sensitive_uri_query,
+        SigV4HeaderGuardLayer, StsQueryApiCompatLayer, VirtualHostStyleHintLayer, redact_sensitive_uri_query,
    },
    rate_limit::{RateLimitLayer, api_rate_limit_layer_from_env, rate_limit_vh_domain_hosts},
    ssec_transport::SsecTransportLayer,
@@ -1650,7 +1649,6 @@ macro_rules! external_service_stack {
            .layer(PublicHealthEndpointLayer::new(server_ctx, readiness))
            .option_layer((!server_domains_configured && !is_console).then_some(VirtualHostStyleHintLayer))
            .layer(DoubleSlashListBucketsCompatLayer)
-            .layer(SignedQueryHoistLayer)
            .layer(SigV4HeaderGuardLayer)
            .service($service)
    }};
@@ -34,7 +34,6 @@ use crate::storage_api::server::layer::batch_diagnostic;
 use crate::storage_api::server::layer::request_context::{RequestContext, extract_request_id_from_headers, spawn_traced};
 use bytes::{Bytes, BytesMut};
 use futures::future::Either;
-use http::header::HeaderName;
 use http::{HeaderMap, HeaderValue, Method, Request as HttpRequest, Response, StatusCode, Uri};
 use http_body::Body;
 use http_body_util::{BodyExt, Full};
@@ -1860,130 +1859,6 @@ where
 /// rulings of [`crate::auth::reject_unsigned_amz_headers_on_sigv4_request`]
 /// must be applied here, in front of s3s. Rejections carry the same S3 error
 /// document the access layer would have produced.
-/// SigV4 query parameters that carry authentication state rather than a request
-/// property. They must not be restored as headers: the verifier reads them from
-/// the query, and `x-amz-security-token` in particular would otherwise look like
-/// an unsigned `x-amz-*` header to [`reject_unsigned_amz_headers_on_presigned_request`].
-///
-/// [`reject_unsigned_amz_headers_on_presigned_request`]: crate::auth::reject_unsigned_amz_headers_on_presigned_request
-const SIGV4_QUERY_AUTH_PARAMS: &[&str] = &[
-    "x-amz-algorithm",
-    "x-amz-credential",
-    "x-amz-date",
-    "x-amz-expires",
-    "x-amz-signedheaders",
-    "x-amz-signature",
-    "x-amz-security-token",
-    "x-amz-region-set",
-];
-
-/// Restores the `x-amz-*` request properties the SigV4 presigners hoist into
-/// the query string back into the request headers.
-///
-/// The AWS SDK presigners move every signable `x-amz-*` header into the query
-/// (`@smithy/signature-v4` `moveHeadersToQuery`) so an uploader does not have to
-/// resend them, and AWS S3 serves the request as if the headers were still
-/// there. RustFS dropped the hoisted form, so a presigned upload lost its user
-/// metadata (rustfs/rustfs#9142), its tagging, its storage class, its object
-/// lock settings and its copy source. Doing this once at the request boundary
-/// means every downstream reader — s3s input parsing,
-/// `extract_metadata_from_mime`, the snowball extract trigger and its
-/// transport-key filter — sees an ordinary header and needs no query knowledge.
-///
-/// Only a query the SigV4 signature covers is honored: on SigV2 the query is
-/// not signed, so honoring it would let the holder of a presigned URL add
-/// object properties the presigner never authorized (for example
-/// `x-amz-meta-snowball-auto-extract`, which triggers tar extraction).
-///
-/// An existing header always wins, so this never overwrites what the client
-/// sent and the unsigned-header guard keeps rejecting anything the signer did
-/// not sign. Keys and values that could not have arrived as a header are
-/// rejected rather than silently dropped or stored: the read path rebuilds them
-/// as header fields, where a control character would fail every `HeadObject` of
-/// that object.
-pub fn hoist_signed_query_params<B>(req: &mut HttpRequest<B>) -> Result<(), S3Error> {
-    let Some(query) = req.uri().query().map(str::to_owned) else {
-        return Ok(());
-    };
-    if !crate::auth::query_string_is_signed(req.headers(), &query) {
-        return Ok(());
-    }
-    let mut seen = std::collections::HashSet::new();
-    for (name, value) in url::form_urlencoded::parse(query.as_bytes()) {
-        // Query parameter names keep whatever case the signer used (`X-Amz-Meta-User-Id`
-        // is what a hoist produces), so match the prefix the way `HeaderName` would.
-        if !rustfs_utils::http::metadata_compat::starts_with_ignore_ascii_case(&name, "x-amz-")
-            || SIGV4_QUERY_AUTH_PARAMS.iter().any(|param| name.eq_ignore_ascii_case(param))
-        {
-            continue;
-        }
-        if !seen.insert(name.to_ascii_lowercase()) {
-            return Err(S3Error::with_message(
-                S3ErrorCode::InvalidArgument,
-                "Duplicate x-amz request parameters in the request query string".to_owned(),
-            ));
-        }
-        let header_name = HeaderName::from_bytes(name.as_bytes()).map_err(|_| {
-            S3Error::with_message(S3ErrorCode::InvalidArgument, "Invalid x-amz request parameter name".to_owned())
-        })?;
-        let header_value = HeaderValue::from_bytes(value.as_bytes()).map_err(|_| {
-            S3Error::with_message(
-                S3ErrorCode::InvalidArgument,
-                "x-amz request parameter values must be valid HTTP header values".to_owned(),
-            )
-        })?;
-        if req.headers().contains_key(&header_name) {
-            continue;
-        }
-        req.headers_mut().insert(header_name, header_value);
-    }
-    Ok(())
-}
-
-/// Restores hoisted `x-amz-*` query parameters as request headers before s3s
-/// routes and deserializes the request. See [`hoist_signed_query_params`].
-#[derive(Clone, Default)]
-pub struct SignedQueryHoistLayer;
-
-impl<S> Layer<S> for SignedQueryHoistLayer {
-    type Service = SignedQueryHoistService<S>;
-
-    fn layer(&self, inner: S) -> Self::Service {
-        SignedQueryHoistService { inner }
-    }
-}
-
-#[derive(Clone)]
-pub struct SignedQueryHoistService<S> {
-    inner: S,
-}
-
-impl<S, ReqBody, RestBody, GrpcBody> Service<HttpRequest<ReqBody>> for SignedQueryHoistService<S>
-where
-    S: Service<HttpRequest<ReqBody>, Response = Response<HybridBody<RestBody, GrpcBody>>> + Clone + Send + 'static,
-    S::Future: Send + 'static,
-    ReqBody: Send + 'static,
-    RestBody: From<Bytes> + Send + 'static,
-    GrpcBody: Send + 'static,
-{
-    type Response = Response<HybridBody<RestBody, GrpcBody>>;
-    type Error = S::Error;
-    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
-
-    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        self.inner.poll_ready(cx)
-    }
-
-    fn call(&mut self, mut req: HttpRequest<ReqBody>) -> Self::Future {
-        if let Err(error) = hoist_signed_query_params(&mut req) {
-            let version = req.version();
-            return Box::pin(async move { Ok(sigv4_header_guard_rejection(version, error)) });
-        }
-        let mut inner = self.inner.clone();
-        Box::pin(async move { inner.call(req).await })
-    }
-}
-
 #[derive(Clone, Default)]
 pub struct SigV4HeaderGuardLayer;

@@ -4243,223 +4118,6 @@ mod tests {
        assert_eq!(calls.load(Ordering::SeqCst), 2);
    }

-    use crate::auth::reject_unsigned_amz_headers_on_presigned_request;
-
-    /// A SigV4 presigned query: the presigner moved the request's `x-amz-*`
-    /// headers into it and every parameter is covered by the signature.
-    fn presigned_query(extra: &str) -> String {
-        let mut query =
-            String::from("X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=AKID%2F20260101%2Fus-east-1%2Fs3%2Faws4_request");
-        query.push_str("&X-Amz-Date=20260101T000000Z&X-Amz-Expires=300&X-Amz-SignedHeaders=host&X-Amz-Signature=deadbeef");
-        query.push('&');
-        query.push_str(extra);
-        query
-    }
-
-    fn hoist(uri_query: &str) -> Result<http::Request<()>, S3Error> {
-        let mut req = Request::builder()
-            .method(Method::PUT)
-            .uri(format!("/bucket/key?{uri_query}"))
-            .body(())
-            .expect("request");
-        hoist_signed_query_params(&mut req)?;
-        Ok(req)
-    }
-
-    #[test]
-    fn hoist_restores_presigned_query_params_as_headers() {
-        let req = hoist(&presigned_query("x-amz-meta-user-id=u1&x-amz-tagging=project%3Drustfs")).expect("hoist");
-        assert_eq!(req.headers().get("x-amz-meta-user-id").and_then(|v| v.to_str().ok()), Some("u1"));
-        assert_eq!(req.headers().get("x-amz-tagging").and_then(|v| v.to_str().ok()), Some("project=rustfs"));
-    }
-
-    #[test]
-    fn hoist_normalizes_parameter_names_like_header_names() {
-        let req = hoist(&presigned_query("X-Amz-Meta-User-Id=u1")).expect("hoist");
-        assert_eq!(req.headers().get("x-amz-meta-user-id").and_then(|v| v.to_str().ok()), Some("u1"));
-        // `HeaderName` is lowercase, so the stored key is the normalized one.
-        assert_eq!(req.headers().keys().map(|name| name.as_str()).collect::<Vec<_>>(), ["x-amz-meta-user-id"]);
-    }
-
-    #[test]
-    fn hoist_never_overwrites_a_header_the_client_sent() {
-        let mut req = Request::builder()
-            .method(Method::PUT)
-            .uri(format!("/bucket/key?{}", presigned_query("x-amz-meta-user-id=from-query")))
-            .header("x-amz-meta-user-id", "from-header")
-            .body(())
-            .expect("request");
-        hoist_signed_query_params(&mut req).expect("hoist");
-        assert_eq!(req.headers().get("x-amz-meta-user-id").and_then(|v| v.to_str().ok()), Some("from-header"));
-    }
-
-    #[test]
-    fn hoist_skips_sigv4_auth_params_and_non_amz_parameters() {
-        let req = hoist(&presigned_query("versionId=3&x-amz-security-token=token")).expect("hoist");
-        assert!(req.headers().get("x-amz-algorithm").is_none());
-        assert!(req.headers().get("x-amz-credential").is_none());
-        assert!(req.headers().get("x-amz-signature").is_none());
-        assert!(req.headers().get("x-amz-security-token").is_none());
-        assert!(req.headers().get("versionId").is_none());
-    }
-
-    #[test]
-    fn hoist_ignores_a_sigv2_upload_that_only_claims_sigv4_markers() {
-        // The SigV2 string-to-sign does not cover arbitrary query parameters,
-        // so appending the SigV4 presign markers leaves the original signature
-        // valid while making the query look signed. A URL holder must not be
-        // able to turn on archive extraction that way.
-        let mut query = String::from("AWSAccessKeyId=AKID&Expires=1893456000&Signature=deadbeef");
-        query.push_str("&X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=AKID&X-Amz-Signature=deadbeef");
-        query.push_str("&x-amz-meta-snowball-auto-extract=true&x-amz-meta-user-id=u1");
-        let req = hoist(&query).expect("hoist");
-        assert!(req.headers().get("x-amz-meta-snowball-auto-extract").is_none());
-        assert!(req.headers().get("x-amz-meta-user-id").is_none());
-    }
-
-    #[test]
-    fn hoist_requires_the_complete_presign_marker_set() {
-        // A lone marker is not a presigned request; `X-Amz-Signature` alone is
-        // what the unsigned-header guard keys on, not proof of presigning.
-        let req = hoist("X-Amz-Signature=deadbeef&x-amz-meta-user-id=u1").expect("hoist");
-        assert!(req.headers().get("x-amz-meta-user-id").is_none());
-    }
-
-    #[test]
-    fn hoist_preserves_an_explicit_false_extraction_header() {
-        // Header precedence: a client that says `false` in a header must not be
-        // overruled by the hoisted form of the same key.
-        let mut req = Request::builder()
-            .method(Method::PUT)
-            .uri(format!("/bucket/key?{}", presigned_query("x-amz-meta-snowball-auto-extract=true")))
-            .header("x-amz-meta-snowball-auto-extract", "false")
-            .body(())
-            .expect("request");
-        hoist_signed_query_params(&mut req).expect("hoist");
-        assert_eq!(
-            req.headers()
-                .get("x-amz-meta-snowball-auto-extract")
-                .and_then(|v| v.to_str().ok()),
-            Some("false")
-        );
-    }
-
-    #[test]
-    fn hoist_ignores_a_query_the_signature_does_not_cover() {
-        // SigV2 signs headers plus a fixed subresource list, not arbitrary
-        // query parameters; honoring them would widen a presigned upload.
-        let req = hoist("AWSAccessKeyId=AKID&Signature=deadbeef&x-amz-meta-user-id=u1").expect("hoist");
-        assert!(req.headers().get("x-amz-meta-user-id").is_none());
-
-        // Anonymous: no signature at all.
-        let req = hoist("x-amz-meta-user-id=u1").expect("hoist");
-        assert!(req.headers().get("x-amz-meta-user-id").is_none());
-    }
-
-    #[test]
-    fn hoist_restores_params_on_a_header_signed_sigv4_request() {
-        let mut req = Request::builder()
-            .method(Method::PUT)
-            .uri("/bucket/key?x-amz-meta-user-id=u1")
-            .header(
-                "authorization",
-                "AWS4-HMAC-SHA256 Credential=AKID/20260101/us-east-1/s3/aws4_request, SignedHeaders=host, Signature=deadbeef",
-            )
-            .body(())
-            .expect("request");
-        hoist_signed_query_params(&mut req).expect("hoist");
-        assert_eq!(req.headers().get("x-amz-meta-user-id").and_then(|v| v.to_str().ok()), Some("u1"));
-    }
-
-    #[test]
-    fn hoist_rejects_names_and_values_that_could_not_be_headers() {
-        // A stored control character would make the read path fail every
-        // HeadObject of that object.
-        let err = hoist(&presigned_query("x-amz-meta-user-id=a%0d%0ab")).expect_err("control char");
-        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
-
-        let err = hoist(&presigned_query("x-amz-meta-bad%20key=v")).expect_err("bad name");
-        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
-    }
-
-    #[test]
-    fn hoist_rejects_duplicate_parameters_after_name_normalization() {
-        let err = hoist(&presigned_query("x-amz-meta-User-Id=1&x-amz-meta-user-id=2")).expect_err("duplicate");
-        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
-    }
-
-    #[test]
-    fn hoisted_metadata_matches_the_header_path_exactly() {
-        let req = hoist(&presigned_query("x-amz-meta-user-id=u1&x-amz-meta-content-type=custom")).expect("hoist");
-        let mut from_query = HeaderMap::new();
-        for name in ["x-amz-meta-user-id", "x-amz-meta-content-type"] {
-            from_query.insert(HeaderName::from_static(name), req.headers().get(name).expect("hoisted header").clone());
-        }
-
-        let mut headers = HeaderMap::new();
-        headers.insert("x-amz-meta-user-id", HeaderValue::from_static("u1"));
-        headers.insert("x-amz-meta-content-type", HeaderValue::from_static("custom"));
-        assert_eq!(from_query, headers);
-    }
-
-    #[tokio::test]
-    async fn signed_query_hoist_layer_rejects_a_control_character_value() {
-        let inner = CountingHybridService::default();
-        let calls = inner.calls();
-        let mut service = SignedQueryHoistLayer.layer(inner);
-
-        let response = service
-            .call(
-                Request::builder()
-                    .method(Method::PUT)
-                    .uri(format!("/bucket/key?{}", presigned_query("x-amz-meta-user-id=a%0d%0ab")))
-                    .body(Full::<Bytes>::from(Bytes::new()))
-                    .expect("request"),
-            )
-            .await
-            .expect("hoist response");
-
-        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
-        assert_eq!(calls.load(Ordering::SeqCst), 0);
-        let body = BodyExt::collect(response.into_body()).await.expect("body").to_bytes();
-        let body = String::from_utf8(body.to_vec()).expect("utf8 body");
-        assert!(body.contains("<Code>InvalidArgument</Code>"), "body: {body}");
-    }
-
-    #[tokio::test]
-    async fn sigv4_header_guard_accepts_a_header_that_matches_a_signed_query_value() {
-        // The hoist layer restores `?x-amz-meta-user-id=u1` as a header before
-        // the access boundary re-runs this rule; the value it carries is the one
-        // the signature covers.
-        let uri = format!("/bucket/key?{}", presigned_query("x-amz-meta-user-id=u1"));
-        let mut headers = HeaderMap::new();
-        headers.insert("x-amz-meta-user-id", HeaderValue::from_static("u1"));
-        assert!(reject_unsigned_amz_headers_on_presigned_request(&headers, Some(uri.split('?').nth(1).unwrap())).is_ok());
-    }
-
-    #[tokio::test]
-    async fn sigv4_header_guard_still_rejects_an_unsigned_header() {
-        let uri = format!("/bucket/key?{}", presigned_query("x-amz-meta-user-id=u1"));
-        let query = uri.split('?').nth(1).unwrap();
-
-        // Same name, different value: the signer did not authorize it.
-        let mut headers = HeaderMap::new();
-        headers.insert("x-amz-meta-user-id", HeaderValue::from_static("evil"));
-        assert!(reject_unsigned_amz_headers_on_presigned_request(&headers, Some(query)).is_err());
-
-        // No such parameter at all.
-        let mut headers = HeaderMap::new();
-        headers.insert("x-amz-meta-other", HeaderValue::from_static("v"));
-        assert!(reject_unsigned_amz_headers_on_presigned_request(&headers, Some(query)).is_err());
-
-        // Multi-valued: the hoist layer never produces one, and the extra value
-        // is unsigned.
-        let mut headers = HeaderMap::new();
-        headers.append("x-amz-meta-user-id", HeaderValue::from_static("u1"));
-        headers.append("x-amz-meta-user-id", HeaderValue::from_static("evil"));
-        assert!(reject_unsigned_amz_headers_on_presigned_request(&headers, Some(query)).is_err());
-    }
-
    #[tokio::test]
    async fn virtual_host_style_hint_layer_short_circuits_unroutable_put() {
        let inner = CountingHybridService::default();
@@ -72,7 +72,7 @@ pub(crate) use health::{
 pub(crate) use http::HeaderMapCarrier;
 pub(crate) use http::active_http_requests;
 pub use layer::s3_http_request_guard;
-pub(crate) use layer::{RequestContextLayer, hoist_signed_query_params, is_sts_query_request};
+pub(crate) use layer::{RequestContextLayer, is_sts_query_request};
 pub(crate) use module_switch::{
    MODULE_SWITCHES_SIGNAL_SUBSYSTEM, ModuleSwitchSnapshot, ModuleSwitchSource, PersistedModuleSwitches,
    current_module_switch_snapshot, refresh_persisted_module_switches_from, refresh_persisted_module_switches_from_store,
@@ -90,8 +90,83 @@ static PUT_FILE_AUTH_STRICT: LazyLock<bool> = LazyLock::new(|| {
        rustfs_config::DEFAULT_INTERNODE_RPC_BODY_DIGEST_STRICT,
    )
 });
-static PUT_FILE_TARGET_LOCKS: LazyLock<parking_lot::Mutex<HashMap<String, Weak<Mutex<()>>>>> =
-    LazyLock::new(|| parking_lot::Mutex::new(HashMap::new()));
+const PUT_FILE_TARGET_LOCK_SHARDS: usize = 64;
+const PUT_FILE_TARGET_LOCK_INITIAL_GC_THRESHOLD: usize = 64;
+
+#[derive(Clone, Copy, Default)]
+struct Xxh3TargetKeyHasher(u64);
+
+impl std::hash::Hasher for Xxh3TargetKeyHasher {
+    #[inline]
+    fn write(&mut self, bytes: &[u8]) {
+        self.0 = xxhash_rust::xxh3::xxh3_64(bytes);
+    }
+
+    #[inline]
+    fn write_usize(&mut self, _i: usize) {}
+
+    #[inline]
+    fn finish(&self) -> u64 {
+        self.0
+    }
+}
+
+type Xxh3TargetKeyBuildHasher = std::hash::BuildHasherDefault<Xxh3TargetKeyHasher>;
+
+struct PutFileTargetLockShard {
+    entries: HashMap<Box<[u8]>, Weak<Mutex<()>>, Xxh3TargetKeyBuildHasher>,
+    next_gc_threshold: usize,
+}
+
+impl Default for PutFileTargetLockShard {
+    fn default() -> Self {
+        Self {
+            entries: HashMap::default(),
+            next_gc_threshold: PUT_FILE_TARGET_LOCK_INITIAL_GC_THRESHOLD,
+        }
+    }
+}
+
+struct ShardedPutFileTargetLockRegistry {
+    shards: [parking_lot::Mutex<PutFileTargetLockShard>; PUT_FILE_TARGET_LOCK_SHARDS],
+}
+
+impl ShardedPutFileTargetLockRegistry {
+    fn new() -> Self {
+        Self {
+            shards: std::array::from_fn(|_| parking_lot::Mutex::new(PutFileTargetLockShard::default())),
+        }
+    }
+
+    #[inline]
+    fn get_or_create(&self, key: &[u8]) -> Arc<Mutex<()>> {
+        let hash = xxhash_rust::xxh3::xxh3_64(key);
+        let shard_idx = ((hash >> 32) as usize) & (PUT_FILE_TARGET_LOCK_SHARDS - 1);
+        let mut shard = self.shards[shard_idx].lock();
+        if let Some(existing) = shard.entries.get(key).and_then(Weak::upgrade) {
+            return existing;
+        }
+        let created = Arc::new(Mutex::new(()));
+        let weak = Arc::downgrade(&created);
+        if let Some(slot) = shard.entries.get_mut(key) {
+            *slot = weak;
+            return created;
+        }
+        if shard.entries.len() >= shard.next_gc_threshold {
+            shard.entries.retain(|_, lock| lock.strong_count() > 0);
+            shard.next_gc_threshold = shard
+                .entries
+                .len()
+                .saturating_add(1)
+                .saturating_mul(2)
+                .max(PUT_FILE_TARGET_LOCK_INITIAL_GC_THRESHOLD);
+        }
+        shard.entries.insert(key.into(), weak);
+        created
+    }
+}
+
+static PUT_FILE_TARGET_LOCKS: LazyLock<ShardedPutFileTargetLockRegistry> = LazyLock::new(ShardedPutFileTargetLockRegistry::new);

 macro_rules! log_internode_rpc_response_failure {
    ($status:expr, $rpc_path:expr, $method:expr, $operation:expr, $reason:expr, $result:expr, Some(($context_key:expr, $context_value:expr)), Some($error_text:expr)) => {{
@@ -1437,16 +1512,31 @@ where
 }

 fn put_file_target_lock(disk: &DiskStore, query: &PutFileQuery) -> Arc<Mutex<()>> {
-    let key = format!("{:p}\0{}\0{}", Arc::as_ptr(disk), query.volume, query.path);
-    let mut locks = PUT_FILE_TARGET_LOCKS.lock();
-    locks.retain(|_, lock| lock.strong_count() > 0);
-    if let Some(lock) = locks.get(&key).and_then(Weak::upgrade) {
-        return lock;
+    let disk_addr = (Arc::as_ptr(disk) as usize).to_ne_bytes();
+    let vol_bytes = query.volume.as_bytes();
+    let vol_len = (vol_bytes.len() as u64).to_ne_bytes();
+    let path_bytes = query.path.as_bytes();
+    let total_len = disk_addr.len() + vol_len.len() + vol_bytes.len() + path_bytes.len();
+
+    let mut stack_buf = [0u8; 512];
+    if total_len <= stack_buf.len() {
+        let mut pos = 0;
+        stack_buf[pos..pos + disk_addr.len()].copy_from_slice(&disk_addr);
+        pos += disk_addr.len();
+        stack_buf[pos..pos + vol_len.len()].copy_from_slice(&vol_len);
+        pos += vol_len.len();
+        stack_buf[pos..pos + vol_bytes.len()].copy_from_slice(vol_bytes);
+        pos += vol_bytes.len();
+        stack_buf[pos..pos + path_bytes.len()].copy_from_slice(path_bytes);
+        return PUT_FILE_TARGET_LOCKS.get_or_create(&stack_buf[..total_len]);
    }

-    let lock = Arc::new(Mutex::new(()));
-    locks.insert(key, Arc::downgrade(&lock));
-    lock
+    let mut heap_key = Vec::with_capacity(total_len);
+    heap_key.extend_from_slice(&disk_addr);
+    heap_key.extend_from_slice(&vol_len);
+    heap_key.extend_from_slice(vol_bytes);
+    heap_key.extend_from_slice(path_bytes);
+    PUT_FILE_TARGET_LOCKS.get_or_create(&heap_key)
 }

 async fn remove_put_file_staging(disk: &DiskStore, volume: &str, path: &str) -> Result<(), BoxError> {
@@ -3747,4 +3837,51 @@ mod tests {
            assert_eq!(data, expected);
        }
    }
+
+    #[tokio::test]
+    async fn sharded_put_file_target_lock_separates_tuples_and_handles_long_keys() {
+        let (disk_a, _dir_a) = new_put_file_test_disk().await;
+        let (disk_b, _dir_b) = new_put_file_test_disk().await;
+
+        let make_query = |volume: &str, path: &str| PutFileQuery {
+            disk: "disk-a".to_string(),
+            volume: volume.to_string(),
+            path: path.to_string(),
+            append: false,
+            size: 0,
+            put_file_auth: None,
+            put_file_nonce: None,
+            put_file_server_epoch: None,
+        };
+
+        let q1 = make_query("ab", "c");
+        let q2 = make_query("a", "bc");
+        let lock_q1_a = put_file_target_lock(&disk_a, &q1);
+        let lock_q1_b = put_file_target_lock(&disk_a, &q1);
+        let lock_q2 = put_file_target_lock(&disk_a, &q2);
+        let lock_other_disk = put_file_target_lock(&disk_b, &q1);
+        assert!(std::sync::Arc::ptr_eq(&lock_q1_a, &lock_q1_b));
+        assert!(!std::sync::Arc::ptr_eq(&lock_q1_a, &lock_q2));
+        assert!(!std::sync::Arc::ptr_eq(&lock_q1_a, &lock_other_disk));
+
+        let long_path = "segment/".repeat(80);
+        let q_long = make_query("bucket", &long_path);
+        let lock_long_a = put_file_target_lock(&disk_a, &q_long);
+        let lock_long_b = put_file_target_lock(&disk_a, &q_long);
+        assert!(std::sync::Arc::ptr_eq(&lock_long_a, &lock_long_b));
+
+        let registry = super::ShardedPutFileTargetLockRegistry::new();
+        let held = registry.get_or_create(b"persistent-target");
+        drop(registry.get_or_create(b"reused-target"));
+        drop(registry.get_or_create(b"reused-target"));
+        for i in 0..(super::PUT_FILE_TARGET_LOCK_SHARDS * super::PUT_FILE_TARGET_LOCK_INITIAL_GC_THRESHOLD * 3) {
+            let key = format!("ephemeral-{i}");
+            drop(registry.get_or_create(key.as_bytes()));
+        }
+        assert!(std::sync::Arc::ptr_eq(&held, &registry.get_or_create(b"persistent-target")));
+        for shard in &registry.shards {
+            let guard = shard.lock();
+            assert!(guard.entries.len() <= guard.next_gc_threshold);
+        }
+    }
 }
Author	SHA1	Message	Date
唐小鸭	116a5299b2	fix(replication): replay the MRF journal periodically (#9139 ) * fix(replication): replay the MRF journal periodically The persisted MRF journal was replayed only at startup. Delete-marker purges whose watch window drained with targets unreachable, and replications that overflowed the worker queues, reach the journal while the server runs, so they waited for a restart to converge. Move the startup pass into replay_mrf_journal and rerun it every five minutes from a background loop. The existing recovery leader lock keeps one pass at a time across the cluster, including the startup pass. * fix(replication): space periodic MRF replay across nodes Every node runs the MRF replay loop and the recovery leader lock only serializes passes, so an N-node cluster replayed the same retained backlog N times per interval while a target stayed down. Record the start time of a pass that finds entries in a shared replay stamp and skip a periodic pass when any node started one within the interval. The startup pass still always runs.	2026-11-03 17:06:14 +00:00
Hauser Zhi	fbc54f7212	fix(heal): treat SlowDown as a rate signal and bound page churn (#9144 ) * fix(heal): treat SlowDown as a rate signal and bound page churn * refactor(heal): unify page rate control and drop checkpoint clones * refactor(heal): one pass-wide rate controller, single SlowDown observe * fix(heal): address SlowDown backoff and checkpoint baseline review	2026-11-03 17:05:05 +00:00
Ramakrishna Chilaka	3aa492f5ee	perf(ecstore,rpc): shard hot-path delete, multipart, and RPC lock tables (#9143 ) Replace the single-mutex PREPARED_DELETE_RECEIPTS and ACTIVE_MULTIPART_SOURCE_HEADS maps in rustfs-ecstore with a 64-shard ShardedPathCache backed by SWAR canonical byte keys and xxh3_64 hashing. Replace PUT_FILE_TARGET_LOCKS in rustfs internode RPC with a 64-shard ShardedPutFileTargetLockRegistry using stack-buffered binary keys, in-place dead-Weak slot reuse, and amortized garbage collection.	2026-11-03 15:11:58 +00:00
Hauser Zhi	00a81b4048	chore(deps): refresh crate requirements and lockfile (#9140 ) Run cargo update and cargo upgrade twice to bump compatible requirements (jiff, zip, dial9) and refresh transitive lockfile entries.	2026-11-03 15:04:29 +00:00