perf(get,heal): fix GET hot-path overhead and heal checkpoint scaling (#4237)

perf(get,heal): land verified fixes for backlog #800-#804

Five fixes from the GET-path performance audit and scanner/heal
completeness audit (rustfs/backlog#800..#804), each verified locally:

- backlog#800 (heal checkpoint O(N^2)): ResumeCheckpoint object sets are
  now HashSet (Vec::contains was O(n) per healed object; 1.5ms at n=1M
  vs 11ns measured), and per-object checkpoint/resume-state persistence
  is batched (1000 mutations / 5s) instead of rewriting the whole file
  per object. complete_page() prunes the sets at page boundaries so
  memory stays bounded; positions still persist unconditionally and
  legacy Vec-format checkpoints still deserialize.

- backlog#801 (DiskInfo.healing never set): erasure-set heal now writes
  a healing marker (.rustfs.sys/healing.bin) on the disks it rebuilds
  (endpoints plumbed via HealRequest/HealTask.heal_endpoints from the
  auto disk scanner) and clears it on success. LocalDisk::disk_info
  surfaces the marker, so scanner heal coordination, lock selection and
  admin/metrics healing counts see the rebuild.

- backlog#802 (cache probe after data read): new GetObjectBodyCacheHook
  in ecstore lets the app-layer object data cache serve the body inside
  get_object_reader, after metadata quorum resolution (etag known) but
  before the erasure shard read/decode. Previously a hit still paid the
  full disk read. Hook is None/no-op when the cache is disabled.

- backlog#803 (GET hot-path redundant work): ObjectInfo is cloned for
  event notification only when an event will actually be built (GET and
  HEAD paths; events are currently suppressed so the clone was pure
  waste); get_opts/put_opts/del_opts resolve bucket versioning with one
  metadata-sys lookup instead of two; skip_verify_bitrot and
  get_lock_acquire_timeout env reads are cached via OnceLock; the
  io-priority metric is no longer double-counted; GetObject input fields
  are cloned selectively instead of cloning the whole input.

- backlog#804 (disk permit starvation): the disk-read permit wait is now
  bounded (RUSTFS_OBJECT_DISK_PERMIT_WAIT_TIMEOUT, default 5s, 0 =
  previous unbounded behavior); on timeout the GET proceeds without a
  permit and the bypass is counted. DiskReadPermitReader also releases
  the permit at body EOF instead of holding it until the client drops
  the stream.

Verification: make pre-commit; cargo clippy -D warnings on the four
changed crates; full rustfs lib suite (2096 tests) green; rustfs-heal
lib suite green with new unit tests for checkpoint pruning/legacy
format/throttle, permit EOF release, and the cache hook (hit + SSE
skip). The heal_integration_test and one set_disk listing test fail
identically on unmodified main (pre-existing global-state ordering
flakes, verified via git stash A/B).

Co-authored-by: houseme <housemecn@gmail.com>

crates/config/src/constants/object.rs

@@ -67,6 +67,21 @@ pub const DEFAULT_OBJECT_MEDIUM_CONCURRENCY_THRESHOLD: usize = 4;
 /// Default is set to 64 concurrent reads.
 pub const DEFAULT_OBJECT_MAX_CONCURRENT_DISK_READS: usize = 64;
 
+/// Environment variable for the disk read permit wait timeout (seconds).
+/// - Purpose: Bound how long a GET waits for a disk read permit before proceeding without one.
+/// - Unit: seconds (u64). `0` waits indefinitely.
+/// - Example: `export RUSTFS_OBJECT_DISK_PERMIT_WAIT_TIMEOUT=5`
+pub const ENV_OBJECT_DISK_PERMIT_WAIT_TIMEOUT: &str = "RUSTFS_OBJECT_DISK_PERMIT_WAIT_TIMEOUT";
+
+/// Maximum time a GET request waits for a disk read permit (seconds).
+///
+/// Permits are held for the whole response body transfer, so slow clients can
+/// occupy all of them while the disks sit idle. Instead of stalling until the
+/// request-level timeout fires, a GET that waits longer than this proceeds
+/// without a permit (degraded pass-through) and the bypass is counted in
+/// metrics/logs. Set to 0 to wait indefinitely (previous behavior).
+pub const DEFAULT_OBJECT_DISK_PERMIT_WAIT_TIMEOUT: u64 = 5;
+
 /// Skip bitrot hash verification on GetObject reads.
 ///
 /// When enabled, GetObject reads skip the per-shard hash

crates/ecstore/src/api/mod.rs

@@ -86,8 +86,8 @@ pub mod disk {
     pub use crate::disk::{
         BATCH_READ_VERSION_MAX_ITEMS, BUCKET_META_PREFIX, BatchReadVersionItem, BatchReadVersionReq, BatchReadVersionResp,
         CheckPartsResp, DeleteOptions, Disk, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, DiskStore,
-        FileInfoVersions, FileReader, FileWriter, RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions,
-        RenameDataResp, STORAGE_FORMAT_FILE, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, new_disk,
+        FileInfoVersions, FileReader, FileWriter, HEALING_MARKER_PATH, RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp,
+        ReadOptions, RenameDataResp, STORAGE_FORMAT_FILE, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, new_disk,
         validate_batch_read_version_item_count,
     };
     pub use crate::disk::{endpoint, error, error_reduce};
@@ -145,8 +145,8 @@ pub mod notification {
 
 pub mod object {
     pub use crate::object_api::{
-        BLOCK_SIZE_V2, ERASURE_ALGORITHM, GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader, RangedDecompressReader,
-        StreamConsumer,
+        BLOCK_SIZE_V2, ERASURE_ALGORITHM, GetObjectBodyCacheHook, GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader,
+        RangedDecompressReader, StreamConsumer, register_get_object_body_cache_hook,
     };
 }
 

crates/ecstore/src/diagnostics/get.rs

@@ -22,6 +22,7 @@ pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE: &str = "codec_st
 pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE: &str = "codec_streaming_rustfs_engine";
 pub(crate) const GET_OBJECT_PATH_EMPTY: &str = "empty";
 pub(crate) const GET_OBJECT_PATH_DIRECT_MEMORY: &str = "direct_memory";
+pub(crate) const GET_OBJECT_PATH_BODY_CACHE: &str = "body_cache";
 pub(crate) const GET_OBJECT_PATH_INLINE_DIRECT: &str = "inline_direct";
 pub(crate) const GET_OBJECT_PATH_LEGACY_DUPLEX: &str = "legacy_duplex";
 pub(crate) const GET_OBJECT_PATH_REMOTE_TRANSITION: &str = "remote_transition";

crates/ecstore/src/disk/local.rs

@@ -864,6 +864,14 @@ impl LocalDisk {
                                 Vec::new()
                             }
                         };
+                        // An erasure-set heal drops a marker on the disks it is
+                        // rebuilding (see rustfs-heal); surface it so scanner
+                        // coordination, lock selection and admin/metrics see
+                        // the rebuild. Refreshed with this cache (~1s).
+                        let healing =
+                            tokio::fs::try_exists(root.join(super::RUSTFS_META_BUCKET).join(super::HEALING_MARKER_PATH))
+                                .await
+                                .unwrap_or(false);
                         let disk_info = DiskInfo {
                             total: info.total,
                             free: info.free,
@@ -876,13 +884,13 @@ impl LocalDisk {
                             root_disk: is_root_disk,
                             physical_device_ids,
                             id: disk_id,
+                            healing,
                             ..Default::default()
                         };
                         // if root {
                         //     return Err(Error::new(DiskError::DriveIsRoot));
                         // }
 
-                        // disk_info.healing =
                         Ok(disk_info)
                     }
                     Err(err) => Err(err.into()),

crates/ecstore/src/disk/mod.rs

@@ -33,6 +33,9 @@ pub const RUSTFS_META_TMP_BUCKET: &str = ".rustfs.sys/tmp";
 pub const RUSTFS_META_TMP_DELETED_BUCKET: &str = ".rustfs.sys/tmp/.trash";
 pub const BUCKET_META_PREFIX: &str = "buckets";
 pub const FORMAT_CONFIG_FILE: &str = "format.json";
+/// Per-disk marker present while an erasure-set heal is rebuilding this disk.
+/// `LocalDisk::disk_info` reports `healing = true` while the file exists.
+pub const HEALING_MARKER_PATH: &str = "healing.bin";
 pub const STORAGE_FORMAT_FILE: &str = "xl.meta";
 pub const STORAGE_FORMAT_FILE_BACKUP: &str = "xl.meta.bkp";
 

crates/ecstore/src/object_api/body_cache_hook.rs

@@ -0,0 +1,49 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Hook that lets the application layer serve a GET body from its object data
+//! cache after metadata resolution but before the erasure data read.
+//!
+//! The cache itself lives above ecstore (it needs app-level config, metrics
+//! and invalidation), but the lookup must happen inside `get_object_reader`:
+//! probing earlier would require a second metadata fan-out, and probing later
+//! (after the reader is built) means a hit no longer saves any disk I/O.
+
+use crate::object_api::ObjectInfo;
+use bytes::Bytes;
+use std::sync::{Arc, OnceLock};
+
+/// Serves full-object GET bodies from a cache keyed by object identity.
+///
+/// Implementations must validate identity (etag/version/size) against the
+/// provided `ObjectInfo`, which reflects the just-resolved metadata quorum,
+/// and must return `None` for anything they cannot serve byte-identically
+/// (encrypted objects, remote/transitioned objects, size mismatches, ...).
+#[async_trait::async_trait]
+pub trait GetObjectBodyCacheHook: Send + Sync + 'static {
+    async fn lookup(&self, bucket: &str, object: &str, info: &ObjectInfo) -> Option<Bytes>;
+}
+
+static GET_OBJECT_BODY_CACHE_HOOK: OnceLock<Arc<dyn GetObjectBodyCacheHook>> = OnceLock::new();
+
+/// Register the process-wide GET body cache hook. First registration wins;
+/// later calls are ignored so tests and re-inits cannot swap the hook midway.
+pub fn register_get_object_body_cache_hook(hook: Arc<dyn GetObjectBodyCacheHook>) {
+    let _ = GET_OBJECT_BODY_CACHE_HOOK.set(hook);
+}
+
+/// The registered hook, if any.
+pub(crate) fn get_object_body_cache_hook() -> Option<&'static Arc<dyn GetObjectBodyCacheHook>> {
+    GET_OBJECT_BODY_CACHE_HOOK.get()
+}

crates/ecstore/src/object_api/mod.rs

@@ -52,8 +52,11 @@ use uuid::Uuid;
 pub const ERASURE_ALGORITHM: &str = "rs-vandermonde";
 pub const BLOCK_SIZE_V2: usize = 1024 * 1024; // 1M
 
+mod body_cache_hook;
 mod readers;
 mod types;
 
+pub(crate) use body_cache_hook::get_object_body_cache_hook;
+pub use body_cache_hook::{GetObjectBodyCacheHook, register_get_object_body_cache_hook};
 pub use readers::*;
 pub use types::*;

crates/ecstore/src/set_disk/mod.rs

@@ -27,7 +27,7 @@ use crate::client::{object_api_utils::get_raw_etag, transition_api::ReaderImpl};
 use crate::cluster::rpc::heal_bucket_local_on_disks;
 use crate::data_usage::record_compression_total_memory;
 use crate::diagnostics::get::{
-    GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART, GET_OBJECT_PATH_CODEC_STREAMING,
+    GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART, GET_OBJECT_PATH_BODY_CACHE, GET_OBJECT_PATH_CODEC_STREAMING,
     GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE, GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE, GET_OBJECT_PATH_DIRECT_MEMORY,
     GET_OBJECT_PATH_EMPTY, GET_OBJECT_PATH_INLINE_DIRECT, GET_OBJECT_PATH_LEGACY_DUPLEX, GET_OBJECT_PATH_REMOTE_TRANSITION,
     GET_OBJECT_PATH_SET_DISK, GET_STAGE_DECODE, GET_STAGE_EMIT, GET_STAGE_INLINE_PREPARE, GET_STAGE_LOCK_ACQUIRE,
@@ -51,6 +51,7 @@ use crate::error::{Error, Result, is_err_version_not_found};
 use crate::error::{GenericError, ObjectApiError, is_err_object_not_found};
 use crate::io_support::bitrot::{create_bitrot_reader, create_bitrot_reader_from_bytes, create_bitrot_writer};
 use crate::object_api::ObjectOptions;
+use crate::object_api::get_object_body_cache_hook;
 use crate::runtime::sources as runtime_sources;
 use crate::services::batch_processor::AsyncBatchProcessor;
 use crate::storage_api_contracts::{
@@ -453,11 +454,27 @@ mod write;
 
 /// Get lock acquire timeout from environment variable RUSTFS_LOCK_ACQUIRE_TIMEOUT (in seconds)
 /// Defaults to 30 seconds if not set or invalid
+/// Lock acquisition timeout. Cached: this is consulted on every object
+/// lock acquisition and `std::env::var` takes a process-global lock. In test
+/// builds the env var is read directly so `temp_env` overrides take effect.
 pub fn get_lock_acquire_timeout() -> Duration {
-    Duration::from_secs(rustfs_utils::get_env_u64(
-        rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT,
-        rustfs_config::DEFAULT_OBJECT_LOCK_ACQUIRE_TIMEOUT,
-    ))
+    #[cfg(test)]
+    {
+        Duration::from_secs(rustfs_utils::get_env_u64(
+            rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT,
+            rustfs_config::DEFAULT_OBJECT_LOCK_ACQUIRE_TIMEOUT,
+        ))
+    }
+    #[cfg(not(test))]
+    {
+        static CACHED: OnceLock<Duration> = OnceLock::new();
+        *CACHED.get_or_init(|| {
+            Duration::from_secs(rustfs_utils::get_env_u64(
+                rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT,
+                rustfs_config::DEFAULT_OBJECT_LOCK_ACQUIRE_TIMEOUT,
+            ))
+        })
+    }
 }
 
 pub fn is_object_lock_diag_enabled() -> bool {
@@ -2332,6 +2349,27 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
             ));
         }
 
+        // App-layer object data cache probe: metadata (etag/size) is resolved
+        // but no data shards have been read yet, so a hit skips the erasure
+        // read, bitrot verify and decode entirely. The hook validates object
+        // identity and rejects anything it cannot serve byte-identically.
+        if range.is_none()
+            && opts.part_number.is_none()
+            && let Some(hook) = get_object_body_cache_hook()
+            && let Some(body) = hook.lookup(bucket, object, &object_info).await
+        {
+            record_get_object_reader_path_observation(GET_OBJECT_PATH_BODY_CACHE, object_class, size_bucket);
+            let reader = GetObjectReader {
+                stream: Box::new(Cursor::new(body.clone())),
+                object_info,
+                buffered_body: Some(body),
+            };
+            if lock_optimization_enabled {
+                release_materialized_read_lock(bucket, object, read_lock_guard.take());
+            }
+            return Ok(reader);
+        }
+
         if is_get_small_object_direct_memory_eligible(&range, &object_info, &fi, opts) {
             let object_size = usize::try_from(object_info.size)
                 .map_err(|_| to_object_err(Error::other("direct-memory GET object size is invalid"), vec![bucket, object]))?;

crates/heal/src/heal/erasure_healer.rs

@@ -677,9 +677,7 @@ impl ErasureSetHealer {
             }
 
             *current_object_index = global_obj_idx;
-            checkpoint_manager
-                .update_position(bucket_index, *current_object_index)
-                .await?;
+            checkpoint_manager.complete_page(bucket_index, *current_object_index).await?;
             gauge!(
                 "rustfs_heal_page_concurrency_current",
                 "set" => set_disk_id.to_string()

crates/heal/src/heal/manager.rs

@@ -2371,6 +2371,7 @@ impl HealManager {
                                 HealPriority::Low,
                             );
                             req.source = HealRequestSource::AutoHeal;
+                            req.heal_endpoints = vec![ep.to_string()];
                             let config = config.read().await;
                             let mut queue = heal_queue.lock().await;
                             let admission = Self::admit_request_to_queue(&mut queue, req, &config, "auto_scan");

crates/heal/src/heal/mod.rs

@@ -24,9 +24,10 @@ pub mod task;
 pub mod utils;
 
 use storage_api::owner::{
-    ECSTORE_BUCKET_META_PREFIX, ECSTORE_DATA_USAGE_CACHE_NAME, ECSTORE_RUSTFS_META_BUCKET, EcstoreDeleteOptions, EcstoreDiskAPI,
-    EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskResult, EcstoreDiskStore, EcstoreEndpoint, EcstoreErrorType,
-    EcstoreStorageError, EcstoreStore, ObjectIO, ObjectOperations, ecstore_local_disk_map_read,
+    ECSTORE_BUCKET_META_PREFIX, ECSTORE_DATA_USAGE_CACHE_NAME, ECSTORE_HEALING_MARKER_PATH, ECSTORE_RUSTFS_META_BUCKET,
+    EcstoreDeleteOptions, EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskResult, EcstoreDiskStore,
+    EcstoreEndpoint, EcstoreErrorType, EcstoreStorageError, EcstoreStore, ObjectIO, ObjectOperations,
+    ecstore_local_disk_map_read,
 };
 #[cfg(test)]
 use storage_api::owner::{EcstoreDiskOption, ecstore_new_disk};
@@ -69,6 +70,64 @@ pub async fn clear_unclean_shutdown_markers() {
     }
 }
 
+/// Per-disk healing marker path (inside `RUSTFS_META_BUCKET`), mirrored from
+/// ecstore so both sides agree on where `DiskInfo.healing` is derived from.
+pub(crate) const HEALING_MARKER_PATH: &str = ECSTORE_HEALING_MARKER_PATH;
+
+/// Write the healing marker on the local disks matching `endpoints` so their
+/// `DiskInfo.healing` reports true while the erasure-set heal rebuilds them.
+pub(crate) async fn set_healing_markers(endpoints: &[String], set_disk_id: &str) {
+    apply_healing_markers(endpoints, Some(set_disk_id)).await;
+}
+
+/// Remove the healing markers written by [`set_healing_markers`].
+pub(crate) async fn clear_healing_markers(endpoints: &[String]) {
+    apply_healing_markers(endpoints, None).await;
+}
+
+async fn apply_healing_markers(endpoints: &[String], set_disk_id: Option<&str>) {
+    if endpoints.is_empty() {
+        return;
+    }
+    let local_disk_map = local_disk_map_read().await;
+    for disk in local_disk_map.values().flatten() {
+        let endpoint = EcstoreDiskAPI::endpoint(disk.as_ref()).to_string();
+        if !endpoints.iter().any(|candidate| candidate == &endpoint) {
+            continue;
+        }
+        let result = match set_disk_id {
+            Some(set_disk_id) => {
+                EcstoreDiskAPI::write_all(
+                    disk.as_ref(),
+                    RUSTFS_META_BUCKET,
+                    HEALING_MARKER_PATH,
+                    EcstoreDiskBytes::copy_from_slice(set_disk_id.as_bytes()),
+                )
+                .await
+            }
+            None => match EcstoreDiskAPI::delete(
+                disk.as_ref(),
+                RUSTFS_META_BUCKET,
+                HEALING_MARKER_PATH,
+                EcstoreDeleteOptions::default(),
+            )
+            .await
+            {
+                Err(DiskError::FileNotFound) => Ok(()),
+                other => other,
+            },
+        };
+        if let Err(err) = result {
+            tracing::warn!(
+                endpoint = %endpoint,
+                action = if set_disk_id.is_some() { "set" } else { "clear" },
+                error = ?err,
+                "failed to update healing marker"
+            );
+        }
+    }
+}
+
 pub(crate) type DiskError = EcstoreDiskError;
 pub(crate) type DiskResult<T> = EcstoreDiskResult<T>;
 pub(crate) type DiskStore = EcstoreDiskStore;

crates/heal/src/heal/resume.rs

@@ -14,9 +14,10 @@
 
 use crate::{Error, Result};
 use serde::{Deserialize, Serialize};
+use std::collections::HashSet;
 use std::path::Path;
-use std::sync::Arc;
-use std::time::{SystemTime, UNIX_EPOCH};
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 use tokio::sync::RwLock;
 use tracing::{debug, warn};
 use uuid::Uuid;
@@ -33,6 +34,39 @@ const RESUME_STATE_FILE: &str = "ahm_resume_state.json";
 const RESUME_PROGRESS_FILE: &str = "ahm_progress.json";
 const RESUME_CHECKPOINT_FILE: &str = "ahm_checkpoint.json";
 
+/// Persistence throttle for per-object bookkeeping: flush after this many
+/// buffered mutations or once the interval elapses, whichever comes first.
+/// Object heal is idempotent, so a crash re-heals at most one throttle window.
+const PERSIST_EVERY_MUTATIONS: usize = 1000;
+const PERSIST_INTERVAL: Duration = Duration::from_secs(5);
+
+/// Tracks buffered mutations between persisted snapshots.
+#[derive(Debug)]
+struct PersistThrottle {
+    pending: usize,
+    last_save: Instant,
+}
+
+impl PersistThrottle {
+    fn new() -> Self {
+        Self {
+            pending: 0,
+            last_save: Instant::now(),
+        }
+    }
+
+    /// Record one mutation; returns true when the batch should be flushed.
+    fn record(&mut self) -> bool {
+        self.pending += 1;
+        self.pending >= PERSIST_EVERY_MUTATIONS || self.last_save.elapsed() >= PERSIST_INTERVAL
+    }
+
+    fn mark_saved(&mut self) {
+        self.pending = 0;
+        self.last_save = Instant::now();
+    }
+}
+
 /// Helper function to convert Path to &str, returning an error if conversion fails
 fn path_to_str(path: &Path) -> Result<&str> {
     path.to_str()
@@ -168,6 +202,7 @@ impl ResumeState {
 pub struct ResumeManager {
     disk: DiskStore,
     state: Arc<RwLock<ResumeState>>,
+    throttle: Mutex<PersistThrottle>,
 }
 
 impl ResumeManager {
@@ -183,6 +218,7 @@ impl ResumeManager {
         let manager = Self {
             disk,
             state: Arc::new(RwLock::new(state)),
+            throttle: Mutex::new(PersistThrottle::new()),
         };
 
         // save initial state
@@ -210,6 +246,7 @@ impl ResumeManager {
         Ok(Self {
             disk,
             state: Arc::new(RwLock::new(state)),
+            throttle: Mutex::new(PersistThrottle::new()),
         })
     }
 
@@ -235,15 +272,31 @@ impl ResumeManager {
         let mut state = self.state.write().await;
         state.update_progress(processed, successful, failed, skipped);
         drop(state);
-        self.save_state().await
+        self.save_state_throttled().await
     }
 
-    /// set current item
+    /// Set current item. Called once per healed object, so persistence is
+    /// throttled: the in-memory state always updates, but the snapshot is only
+    /// written every `PERSIST_EVERY_MUTATIONS` calls or `PERSIST_INTERVAL`.
     pub async fn set_current_item(&self, bucket: Option<String>, object: Option<String>) -> Result<()> {
         let mut state = self.state.write().await;
         state.set_current_item(bucket, object);
         drop(state);
-        self.save_state().await
+        let should_save = self.throttle.lock().map(|mut throttle| throttle.record()).unwrap_or(true);
+        if !should_save {
+            return Ok(());
+        }
+        self.save_state_throttled().await
+    }
+
+    async fn save_state_throttled(&self) -> Result<()> {
+        let result = self.save_state().await;
+        if result.is_ok()
+            && let Ok(mut throttle) = self.throttle.lock()
+        {
+            throttle.mark_saved();
+        }
+        result
     }
 
     /// complete bucket
@@ -251,7 +304,7 @@ impl ResumeManager {
         let mut state = self.state.write().await;
         state.complete_bucket(bucket);
         drop(state);
-        self.save_state().await
+        self.save_state_throttled().await
     }
 
     /// mark task completed
@@ -259,7 +312,7 @@ impl ResumeManager {
         let mut state = self.state.write().await;
         state.mark_completed();
         drop(state);
-        self.save_state().await
+        self.save_state_throttled().await
     }
 
     /// set error message
@@ -376,12 +429,15 @@ pub struct ResumeCheckpoint {
     pub current_bucket_index: usize,
     /// current object index
     pub current_object_index: usize,
-    /// processed objects
-    pub processed_objects: Vec<String>,
+    /// Objects healed since the last completed page. HashSet: with the
+    /// previous Vec the per-object `contains` was O(n) and made large-bucket
+    /// heals O(N²). Only spans the in-flight page — completed pages are
+    /// covered by `current_object_index`, so `complete_page` prunes the sets.
+    pub processed_objects: HashSet<String>,
     /// failed objects
-    pub failed_objects: Vec<String>,
+    pub failed_objects: HashSet<String>,
     /// skipped objects
-    pub skipped_objects: Vec<String>,
+    pub skipped_objects: HashSet<String>,
 }
 
 impl ResumeCheckpoint {
@@ -391,9 +447,9 @@ impl ResumeCheckpoint {
             checkpoint_time: SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs(),
             current_bucket_index: 0,
             current_object_index: 0,
-            processed_objects: Vec::new(),
-            failed_objects: Vec::new(),
-            skipped_objects: Vec::new(),
+            processed_objects: HashSet::new(),
+            failed_objects: HashSet::new(),
+            skipped_objects: HashSet::new(),
         }
     }
 
@@ -404,21 +460,25 @@ impl ResumeCheckpoint {
     }
 
     pub fn add_processed_object(&mut self, object: String) {
-        if !self.processed_objects.contains(&object) {
-            self.processed_objects.push(object);
-        }
+        self.processed_objects.insert(object);
     }
 
     pub fn add_failed_object(&mut self, object: String) {
-        if !self.failed_objects.contains(&object) {
-            self.failed_objects.push(object);
-        }
+        self.failed_objects.insert(object);
     }
 
     pub fn add_skipped_object(&mut self, object: String) {
-        if !self.skipped_objects.contains(&object) {
-            self.skipped_objects.push(object);
-        }
+        self.skipped_objects.insert(object);
+    }
+
+    /// Advance past a fully-processed page: objects below `object_index` are
+    /// skipped by position on resume, so the per-object sets no longer need
+    /// their entries and would otherwise grow with the whole bucket.
+    pub fn complete_page(&mut self, bucket_index: usize, object_index: usize) {
+        self.update_position(bucket_index, object_index);
+        self.processed_objects.clear();
+        self.skipped_objects.clear();
+        self.failed_objects.clear();
     }
 }
 
@@ -426,6 +486,7 @@ impl ResumeCheckpoint {
 pub struct CheckpointManager {
     disk: DiskStore,
     checkpoint: Arc<RwLock<ResumeCheckpoint>>,
+    throttle: Mutex<PersistThrottle>,
 }
 
 impl CheckpointManager {
@@ -435,6 +496,7 @@ impl CheckpointManager {
         let manager = Self {
             disk,
             checkpoint: Arc::new(RwLock::new(checkpoint)),
+            throttle: Mutex::new(PersistThrottle::new()),
         };
 
         // save initial checkpoint
@@ -462,6 +524,7 @@ impl CheckpointManager {
         Ok(Self {
             disk,
             checkpoint: Arc::new(RwLock::new(checkpoint)),
+            throttle: Mutex::new(PersistThrottle::new()),
         })
     }
 
@@ -487,31 +550,59 @@ impl CheckpointManager {
         let mut checkpoint = self.checkpoint.write().await;
         checkpoint.update_position(bucket_index, object_index);
         drop(checkpoint);
-        self.save_checkpoint().await
+        self.save_checkpoint_throttled().await
     }
 
-    /// add processed object
+    /// Advance past a completed page and prune the per-object sets, then persist.
+    pub async fn complete_page(&self, bucket_index: usize, object_index: usize) -> Result<()> {
+        let mut checkpoint = self.checkpoint.write().await;
+        checkpoint.complete_page(bucket_index, object_index);
+        drop(checkpoint);
+        self.save_checkpoint_throttled().await
+    }
+
+    /// Add a processed object. Called once per healed object, so persistence
+    /// is batched (`PERSIST_EVERY_MUTATIONS` / `PERSIST_INTERVAL`); positions
+    /// and page boundaries still persist unconditionally.
     pub async fn add_processed_object(&self, object: String) -> Result<()> {
         let mut checkpoint = self.checkpoint.write().await;
         checkpoint.add_processed_object(object);
         drop(checkpoint);
-        self.save_checkpoint().await
+        self.save_checkpoint_if_due().await
     }
 
-    /// add failed object
+    /// add failed object (batched, see `add_processed_object`)
     pub async fn add_failed_object(&self, object: String) -> Result<()> {
         let mut checkpoint = self.checkpoint.write().await;
         checkpoint.add_failed_object(object);
         drop(checkpoint);
-        self.save_checkpoint().await
+        self.save_checkpoint_if_due().await
     }
 
-    /// add skipped object
+    /// add skipped object (batched, see `add_processed_object`)
     pub async fn add_skipped_object(&self, object: String) -> Result<()> {
         let mut checkpoint = self.checkpoint.write().await;
         checkpoint.add_skipped_object(object);
         drop(checkpoint);
-        self.save_checkpoint().await
+        self.save_checkpoint_if_due().await
+    }
+
+    async fn save_checkpoint_if_due(&self) -> Result<()> {
+        let should_save = self.throttle.lock().map(|mut throttle| throttle.record()).unwrap_or(true);
+        if !should_save {
+            return Ok(());
+        }
+        self.save_checkpoint_throttled().await
+    }
+
+    async fn save_checkpoint_throttled(&self) -> Result<()> {
+        let result = self.save_checkpoint().await;
+        if result.is_ok()
+            && let Ok(mut throttle) = self.throttle.lock()
+        {
+            throttle.mark_saved();
+        }
+        result
     }
 
     /// cleanup checkpoint
@@ -732,6 +823,55 @@ mod tests {
         assert!(state.completed_buckets.contains(&"bucket1".to_string()));
     }
 
+    #[test]
+    fn test_checkpoint_object_sets_dedupe_and_prune() {
+        let mut checkpoint = ResumeCheckpoint::new("task".to_string());
+        checkpoint.add_processed_object("bucket/a".to_string());
+        checkpoint.add_processed_object("bucket/a".to_string());
+        checkpoint.add_skipped_object("bucket/b".to_string());
+        checkpoint.add_failed_object("bucket/c".to_string());
+        assert_eq!(checkpoint.processed_objects.len(), 1);
+        assert!(checkpoint.processed_objects.contains("bucket/a"));
+
+        checkpoint.complete_page(2, 2000);
+        assert_eq!(checkpoint.current_bucket_index, 2);
+        assert_eq!(checkpoint.current_object_index, 2000);
+        assert!(checkpoint.processed_objects.is_empty());
+        assert!(checkpoint.skipped_objects.is_empty());
+        assert!(checkpoint.failed_objects.is_empty());
+    }
+
+    #[test]
+    fn test_checkpoint_loads_legacy_vec_format() {
+        // Checkpoints written before the HashSet migration stored the object
+        // lists as JSON arrays (possibly with duplicates); they must still load.
+        let legacy = r#"{
+            "task_id": "t1",
+            "checkpoint_time": 1700000000,
+            "current_bucket_index": 1,
+            "current_object_index": 42,
+            "processed_objects": ["a", "b", "a"],
+            "failed_objects": [],
+            "skipped_objects": ["c"]
+        }"#;
+        let checkpoint: ResumeCheckpoint = serde_json::from_str(legacy).unwrap();
+        assert_eq!(checkpoint.current_object_index, 42);
+        assert_eq!(checkpoint.processed_objects.len(), 2);
+        assert!(checkpoint.processed_objects.contains("a"));
+        assert!(checkpoint.skipped_objects.contains("c"));
+    }
+
+    #[test]
+    fn test_persist_throttle_batches_until_threshold() {
+        let mut throttle = PersistThrottle::new();
+        for _ in 0..PERSIST_EVERY_MUTATIONS - 1 {
+            assert!(!throttle.record(), "must not flush below the mutation threshold");
+        }
+        assert!(throttle.record(), "must flush at the mutation threshold");
+        throttle.mark_saved();
+        assert!(!throttle.record(), "counter must reset after a save");
+    }
+
     #[tokio::test]
     async fn test_resume_utils() {
         let task_id1 = ResumeUtils::generate_task_id();

crates/heal/src/heal/storage_api.rs

@@ -17,7 +17,8 @@ pub(crate) use rustfs_ecstore::api::disk::endpoint::Endpoint as EcstoreEndpoint;
 pub(crate) use rustfs_ecstore::api::disk::error::{DiskError as EcstoreDiskError, Result as EcstoreDiskResult};
 pub(crate) use rustfs_ecstore::api::disk::{
     BUCKET_META_PREFIX as ECSTORE_BUCKET_META_PREFIX, Bytes as EcstoreDiskBytes, DeleteOptions as EcstoreDeleteOptions,
-    DiskAPI as EcstoreDiskAPI, DiskStore as EcstoreDiskStore, RUSTFS_META_BUCKET as ECSTORE_RUSTFS_META_BUCKET,
+    DiskAPI as EcstoreDiskAPI, DiskStore as EcstoreDiskStore, HEALING_MARKER_PATH as ECSTORE_HEALING_MARKER_PATH,
+    RUSTFS_META_BUCKET as ECSTORE_RUSTFS_META_BUCKET,
 };
 #[cfg(test)]
 pub(crate) use rustfs_ecstore::api::disk::{DiskOption as EcstoreDiskOption, new_disk as ecstore_new_disk};
@@ -30,9 +31,9 @@ pub(crate) mod owner {
     pub(crate) use super::storage_contracts::{ObjectIO, ObjectOperations};
 
     pub(crate) use super::{
-        ECSTORE_BUCKET_META_PREFIX, ECSTORE_DATA_USAGE_CACHE_NAME, ECSTORE_RUSTFS_META_BUCKET, EcstoreDeleteOptions,
-        EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskResult, EcstoreDiskStore, EcstoreEndpoint,
-        EcstoreErrorType, EcstoreStorageError, EcstoreStore, ecstore_local_disk_map_read,
+        ECSTORE_BUCKET_META_PREFIX, ECSTORE_DATA_USAGE_CACHE_NAME, ECSTORE_HEALING_MARKER_PATH, ECSTORE_RUSTFS_META_BUCKET,
+        EcstoreDeleteOptions, EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskResult, EcstoreDiskStore,
+        EcstoreEndpoint, EcstoreErrorType, EcstoreStorageError, EcstoreStore, ecstore_local_disk_map_read,
     };
 
     #[cfg(test)]

crates/heal/src/heal/task.rs

@@ -198,6 +198,11 @@ pub struct HealRequest {
     pub force_start: bool,
     /// Number of recoverable retry attempts already scheduled for this request.
     pub retry_attempts: u32,
+    /// Endpoints of the disks being rebuilt by an erasure-set heal. Used to
+    /// write per-disk healing markers so `DiskInfo.healing` reflects reality;
+    /// empty when the trigger doesn't know the specific disks (admin API,
+    /// unclean-shutdown verification).
+    pub heal_endpoints: Vec<String>,
     /// Created time
     pub created_at: SystemTime,
     /// Queue admission time used for scheduler delay metrics
@@ -215,6 +220,7 @@ impl HealRequest {
             source: HealRequestSource::Internal,
             force_start: false,
             retry_attempts: 0,
+            heal_endpoints: Vec::new(),
             created_at: now,
             enqueued_at: now,
         }
@@ -267,6 +273,8 @@ pub struct HealTask {
     pub source: HealRequestSource,
     /// Number of recoverable retry attempts already scheduled for this task.
     pub retry_attempts: u32,
+    /// Endpoints of the disks being rebuilt (see `HealRequest::heal_endpoints`).
+    pub heal_endpoints: Vec<String>,
     /// Task status
     pub status: Arc<RwLock<HealTaskStatus>>,
     /// Progress tracking
@@ -298,6 +306,7 @@ impl HealTask {
             priority: request.priority,
             source: request.source,
             retry_attempts: request.retry_attempts,
+            heal_endpoints: request.heal_endpoints,
             status: Arc::new(RwLock::new(HealTaskStatus::Pending)),
             progress: Arc::new(RwLock::new(HealProgress::new())),
             result_items: Arc::new(RwLock::new(Vec::new())),
@@ -320,6 +329,7 @@ impl HealTask {
             source: self.source,
             force_start: false,
             retry_attempts: self.retry_attempts.saturating_add(1),
+            heal_endpoints: self.heal_endpoints.clone(),
             created_at: self.created_at,
             enqueued_at: SystemTime::now(),
         }
@@ -2038,6 +2048,10 @@ impl HealTask {
             progress.update_progress(1, 4, 0, 0);
         }
 
+        // The rebuilt disks are formatted now: mark them as healing so
+        // DiskInfo.healing reflects the rebuild until it completes.
+        super::set_healing_markers(&self.heal_endpoints, &set_disk_id).await;
+
         // Step 2: Get disk for resume functionality
         debug!(
             target: "rustfs::heal::task",
@@ -2152,6 +2166,12 @@ impl HealTask {
         );
         let result = erasure_healer.heal_erasure_set(&buckets, &set_disk_id).await;
 
+        // Keep the markers on failure: the resume state also persists, and the
+        // next run of this set heal re-marks and eventually clears them.
+        if result.is_ok() {
+            super::clear_healing_markers(&self.heal_endpoints).await;
+        }
+
         {
             let mut progress = self.progress.write().await;
             progress.update_progress(4, 4, 0, 0);

rustfs/src/app/context/global.rs

@@ -79,6 +79,9 @@ pub struct AppContext {
 impl AppContext {
     pub fn new(object_store: Arc<ECStore>, iam: Arc<dyn IamInterface>, kms: Arc<dyn KmsInterface>) -> Self {
         let object_data_cache = ObjectDataCacheAdapter::from_env_or_disabled();
+        // Let ecstore probe this cache inside get_object_reader, after
+        // metadata resolution but before the erasure data read (backlog#802).
+        crate::app::object_data_cache::register_object_data_cache_body_hook(Arc::clone(&object_data_cache));
 
         Self {
             object_store,

rustfs/src/app/object_data_cache/hook.rs

@@ -0,0 +1,143 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! ecstore-facing GET body cache hook.
+//!
+//! Registered into ecstore so the cache probe runs inside `get_object_reader`
+//! right after metadata resolution: probing earlier needs a second metadata
+//! fan-out, probing later (after the reader is built) means a hit no longer
+//! saves the erasure read/decode.
+
+use crate::app::object_data_cache::{
+    GetObjectBodyCacheLookup, GetObjectBodyCacheRequest, ObjectDataCacheAdapter, build_get_object_body_cache_plan,
+    lookup_get_object_body_cache_hit,
+};
+use crate::app::storage_api::object_usecase::StorageObjectInfo;
+use crate::storage::sse::contains_managed_encryption_metadata;
+use crate::storage::storage_api::ecstore_bucket::lifecycle::bucket_lifecycle_ops::LifecycleOps as _;
+use crate::storage::storage_api::ecstore_object::{GetObjectBodyCacheHook, register_get_object_body_cache_hook};
+use bytes::Bytes;
+use rustfs_utils::http::headers::SSEC_ALGORITHM_HEADER;
+use std::sync::Arc;
+
+/// Adapter-backed implementation of ecstore's GET body cache hook.
+pub(crate) struct ObjectDataCacheBodyHook {
+    adapter: Arc<ObjectDataCacheAdapter>,
+}
+
+/// Registers the body-cache hook into ecstore. No-op for a disabled cache so
+/// the hot path keeps a single `None` branch when the feature is off.
+pub(crate) fn register_object_data_cache_body_hook(adapter: Arc<ObjectDataCacheAdapter>) {
+    if adapter.is_disabled() {
+        return;
+    }
+    register_get_object_body_cache_hook(Arc::new(ObjectDataCacheBodyHook { adapter }));
+}
+
+fn object_metadata_indicates_encryption(metadata: &std::collections::HashMap<String, String>) -> bool {
+    // SSE-C or managed SSE bodies are decrypted on the normal read path, so
+    // the cached plaintext identity used by the planner would not match what
+    // ecstore returns here (pre-decryption). Skip them entirely.
+    metadata.contains_key(SSEC_ALGORITHM_HEADER) || contains_managed_encryption_metadata(metadata)
+}
+
+#[async_trait::async_trait]
+impl GetObjectBodyCacheHook for ObjectDataCacheBodyHook {
+    async fn lookup(&self, bucket: &str, object: &str, info: &StorageObjectInfo) -> Option<Bytes> {
+        if info.is_remote() || object_metadata_indicates_encryption(&info.user_defined) {
+            return None;
+        }
+        let response_content_length = info.get_actual_size().ok()?;
+        let request = GetObjectBodyCacheRequest {
+            bucket,
+            key: object,
+            info,
+            response_content_length,
+            has_range: false,
+            part_number: None,
+            encryption_applied: false,
+        };
+        let plan = build_get_object_body_cache_plan(&self.adapter, request);
+        match lookup_get_object_body_cache_hit(&self.adapter, &plan).await {
+            GetObjectBodyCacheLookup::Hit(bytes) => Some(bytes),
+            GetObjectBodyCacheLookup::Disabled | GetObjectBodyCacheLookup::Skip | GetObjectBodyCacheLookup::Miss => None,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use rustfs_object_data_cache::{ObjectDataCacheConfig, ObjectDataCacheMode};
+
+    fn hit_only_adapter() -> Arc<ObjectDataCacheAdapter> {
+        Arc::new(
+            ObjectDataCacheAdapter::new(ObjectDataCacheConfig {
+                mode: ObjectDataCacheMode::FillBufferedOnly,
+                max_bytes: 4 * 1024 * 1024,
+                ..ObjectDataCacheConfig::default()
+            })
+            .expect("adapter"),
+        )
+    }
+
+    fn plain_info(size: i64) -> StorageObjectInfo {
+        StorageObjectInfo {
+            bucket: "b".to_string(),
+            name: "k".to_string(),
+            etag: Some("etag-1".to_string()),
+            size,
+            actual_size: size,
+            ..Default::default()
+        }
+    }
+
+    #[tokio::test]
+    async fn hook_lookup_returns_cached_body_after_fill() {
+        let adapter = hit_only_adapter();
+        let info = plain_info(5);
+        let request = GetObjectBodyCacheRequest {
+            bucket: "b",
+            key: "k",
+            info: &info,
+            response_content_length: 5,
+            has_range: false,
+            part_number: None,
+            encryption_applied: false,
+        };
+        let plan = build_get_object_body_cache_plan(&adapter, request);
+        let body = Bytes::from_static(b"hello");
+        let _ = crate::app::object_data_cache::fill_get_object_body_cache_from_buffered_body(&adapter, &plan, &body).await;
+
+        let hook = ObjectDataCacheBodyHook {
+            adapter: Arc::clone(&adapter),
+        };
+        let hit = hook.lookup("b", "k", &info).await.expect("cache hit");
+        assert_eq!(hit, body);
+    }
+
+    #[tokio::test]
+    async fn hook_lookup_skips_encrypted_objects() {
+        let adapter = hit_only_adapter();
+        let mut info = plain_info(5);
+        info.user_defined = std::sync::Arc::new(
+            [(SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string())]
+                .into_iter()
+                .collect(),
+        );
+
+        let hook = ObjectDataCacheBodyHook { adapter };
+        assert!(hook.lookup("b", "k", &info).await.is_none());
+    }
+}

rustfs/src/app/object_data_cache/mod.rs

@@ -16,6 +16,7 @@
 
 mod adapter;
 mod body;
+mod hook;
 mod invalidation;
 mod planner;
 
@@ -24,6 +25,7 @@ pub(crate) use body::{
     GetObjectBodyCacheLookup, fill_get_object_body_cache_from_buffered_body, fill_get_object_body_cache_from_materialized_body,
     lookup_get_object_body_cache_hit,
 };
+pub(crate) use hook::register_object_data_cache_body_hook;
 pub(crate) use invalidation::{
     invalidate_object_data_cache_after_complete_multipart_success, invalidate_object_data_cache_after_copy_success,
     invalidate_object_data_cache_after_delete_success, invalidate_object_data_cache_after_put_success,

rustfs/src/app/object_usecase.rs

@@ -330,7 +330,6 @@ struct GetObjectRequestContext {
 
 struct GetObjectReadSetup {
     info: ObjectInfo,
-    event_info: ObjectInfo,
     final_stream: DynReader,
     buffered_body: Option<Bytes>,
     rs: Option<HTTPRangeSpec>,
@@ -361,7 +360,7 @@ struct GetObjectStrategyContext {
 
 struct GetObjectOutputContext {
     output: GetObjectOutput,
-    event_info: ObjectInfo,
+    event_info: Option<ObjectInfo>,
     response_content_length: i64,
     optimal_buffer_size: usize,
 }
@@ -522,7 +521,7 @@ pin_project! {
     struct DiskReadPermitReader<R> {
         #[pin]
         inner: R,
-        _disk_permit: OwnedSemaphorePermit,
+        disk_permit: Option<OwnedSemaphorePermit>,
     }
 }
 
@@ -530,7 +529,7 @@ impl<R> DiskReadPermitReader<R> {
     fn new(inner: R, disk_permit: OwnedSemaphorePermit) -> Self {
         Self {
             inner,
-            _disk_permit: disk_permit,
+            disk_permit: Some(disk_permit),
         }
     }
 }
@@ -540,7 +539,16 @@ where
     R: AsyncRead,
 {
     fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
-        self.project().inner.poll_read(cx, buf)
+        let this = self.project();
+        let had_capacity = buf.remaining() > 0;
+        let filled_before = buf.filled().len();
+        let poll = this.inner.poll_read(cx, buf);
+        // EOF: no more disk reads can happen through this stream, so release
+        // the permit instead of holding it until the client drops the body.
+        if had_capacity && matches!(poll, Poll::Ready(Ok(()))) && buf.filled().len() == filled_before {
+            this.disk_permit.take();
+        }
+        poll
     }
 }
 
@@ -2208,6 +2216,18 @@ impl DefaultObjectUsecase {
         })
     }
 
+    /// How long a GET waits for a disk read permit before degrading to a
+    /// permit-less read. Cached: consulted per GET. Zero disables the bound.
+    fn disk_permit_wait_timeout() -> Duration {
+        static CACHED: std::sync::OnceLock<Duration> = std::sync::OnceLock::new();
+        *CACHED.get_or_init(|| {
+            Duration::from_secs(rustfs_utils::get_env_u64(
+                rustfs_config::ENV_OBJECT_DISK_PERMIT_WAIT_TIMEOUT,
+                rustfs_config::DEFAULT_OBJECT_DISK_PERMIT_WAIT_TIMEOUT,
+            ))
+        })
+    }
+
     async fn acquire_get_object_io_planning(
         manager: &ConcurrencyManager,
         wrapper: &RequestTimeoutWrapper,
@@ -2216,10 +2236,32 @@ impl DefaultObjectUsecase {
         key: &str,
     ) -> S3Result<GetObjectIoPlanning> {
         let permit_wait_start = std::time::Instant::now();
-        let disk_permit = manager
-            .acquire_owned_disk_read_permit()
-            .await
-            .map_err(|_| s3_error!(InternalError, "disk read semaphore closed"))?;
+        let permit_wait_timeout = Self::disk_permit_wait_timeout();
+        // Permits are held for the whole body transfer, so slow clients can
+        // pin all of them while disks are idle. Bound the wait and degrade to
+        // a permit-less read instead of stalling into the request timeout.
+        let disk_permit = if permit_wait_timeout.is_zero() {
+            Some(
+                manager
+                    .acquire_owned_disk_read_permit()
+                    .await
+                    .map_err(|_| s3_error!(InternalError, "disk read semaphore closed"))?,
+            )
+        } else {
+            match tokio::time::timeout(permit_wait_timeout, manager.acquire_owned_disk_read_permit()).await {
+                Ok(permit) => Some(permit.map_err(|_| s3_error!(InternalError, "disk read semaphore closed"))?),
+                Err(_) => {
+                    metrics::counter!("rustfs.get_object.disk_permit.bypass.total").increment(1);
+                    warn!(
+                        bucket = %bucket,
+                        key = %key,
+                        wait_ms = permit_wait_start.elapsed().as_millis() as u64,
+                        "GetObject proceeding without disk read permit after bounded wait"
+                    );
+                    None
+                }
+            }
+        };
         let permit_wait_duration = permit_wait_start.elapsed();
 
         Self::ensure_get_object_not_timed_out(
@@ -2253,7 +2295,7 @@ impl DefaultObjectUsecase {
         Self::ensure_get_object_not_timed_out(wrapper, timeout_config, bucket, key, GetObjectTimeoutStage::BeforeRead)?;
 
         Ok(GetObjectIoPlanning {
-            disk_permit: Some(disk_permit),
+            disk_permit,
             permit_wait_duration,
             queue_status,
             queue_utilization,
@@ -2261,14 +2303,12 @@ impl DefaultObjectUsecase {
     }
 
     async fn prepare_get_object_request_context(req: &S3Request<GetObjectInput>) -> S3Result<GetObjectRequestContext> {
-        let GetObjectInput {
-            bucket,
-            key,
-            version_id,
-            part_number,
-            range,
-            ..
-        } = req.input.clone();
+        // Clone only the fields this path needs instead of the whole input.
+        let bucket = req.input.bucket.clone();
+        let key = req.input.key.clone();
+        let version_id = req.input.version_id.clone();
+        let part_number = req.input.part_number;
+        let range = req.input.range;
 
         validate_object_key(&key, "GET")?;
 
@@ -2420,7 +2460,6 @@ impl DefaultObjectUsecase {
             );
         }
 
-        let event_info = info.clone();
         let content_type = if let Some(content_type) = &info.content_type {
             match ContentType::from_str(content_type) {
                 Ok(res) => Some(res),
@@ -2510,7 +2549,6 @@ impl DefaultObjectUsecase {
 
         Ok(GetObjectReadSetup {
             info,
-            event_info,
             final_stream,
             buffered_body,
             rs,
@@ -2595,8 +2633,6 @@ impl DefaultObjectUsecase {
                 request_size = response_content_length,
                 "I/O priority assigned (based on actual request size)"
             );
-
-            rustfs_io_metrics::record_io_priority_assignment(io_priority.as_str());
         }
 
         rustfs_io_metrics::record_get_object_io_state(
@@ -3609,11 +3645,15 @@ impl DefaultObjectUsecase {
         bucket: &str,
         method: &hyper::Method,
         headers: &HeaderMap,
-        event_info: ObjectInfo,
+        event_info: Option<ObjectInfo>,
         version_id_for_event: String,
         output: GetObjectOutput,
     ) -> S3Result<S3Response<GetObjectOutput>> {
-        let helper = helper.object(event_info).version_id(version_id_for_event);
+        let helper = match event_info {
+            Some(event_info) => helper.object(event_info),
+            None => helper,
+        };
+        let helper = helper.version_id(version_id_for_event);
         let response = wrap_response_with_cors(bucket, method, headers, output).await;
         let result = Ok(response);
         let _ = helper.complete(&result);
@@ -3627,7 +3667,7 @@ impl DefaultObjectUsecase {
         bucket: &str,
         key: &str,
         info: ObjectInfo,
-        event_info: ObjectInfo,
+        event_info: Option<ObjectInfo>,
         final_stream: DynReader,
         buffered_body: Option<Bytes>,
         rs: Option<HTTPRangeSpec>,
@@ -3831,7 +3871,6 @@ impl DefaultObjectUsecase {
 
         let GetObjectReadSetup {
             info,
-            event_info,
             final_stream,
             buffered_body,
             rs,
@@ -3852,6 +3891,10 @@ impl DefaultObjectUsecase {
             final_stream
         };
 
+        // Clone ObjectInfo for event notification only when an event will
+        // actually be built — the clone is expensive for multipart objects.
+        let event_info = helper.wants_object_info().then(|| info.clone());
+
         let output_build_start = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now);
         let output_context = self
             .build_get_object_output_context(
@@ -5225,7 +5268,9 @@ impl DefaultObjectUsecase {
 
         // Compute x-amz-expiration header from lifecycle prediction (before info is partially moved)
         let expiration_header = resolve_put_object_expiration(&bucket, &info).await;
-        let event_info = info.clone();
+        // Clone ObjectInfo for event notification only when an event will
+        // actually be built — the clone is expensive for multipart objects.
+        let event_info = helper.wants_object_info().then(|| info.clone());
         let content_type = {
             if let Some(content_type) = &info.content_type {
                 match ContentType::from_str(content_type) {
@@ -5344,7 +5389,10 @@ impl DefaultObjectUsecase {
         };
 
         let version_id = req.input.version_id.clone().unwrap_or_default();
-        helper = helper.object(event_info).version_id(version_id);
+        if let Some(event_info) = event_info {
+            helper = helper.object(event_info);
+        }
+        helper = helper.version_id(version_id);
 
         // NOTE ON CORS:
         // Bucket-level CORS headers are intentionally applied only for object retrieval
@@ -7401,6 +7449,26 @@ mod tests {
         assert_eq!(body, b"hello");
     }
 
+    #[tokio::test]
+    async fn disk_read_permit_reader_releases_permit_at_eof() {
+        use tokio::io::AsyncReadExt;
+
+        let semaphore = Arc::new(tokio::sync::Semaphore::new(1));
+        let permit = semaphore.clone().acquire_owned().await.expect("acquire permit");
+        assert_eq!(semaphore.available_permits(), 0);
+
+        let mut reader = DiskReadPermitReader::new(std::io::Cursor::new(b"hello".to_vec()), permit);
+        let mut body = Vec::new();
+        reader.read_to_end(&mut body).await.expect("read body");
+        assert_eq!(body, b"hello");
+
+        // The reader is still alive (client hasn't dropped the body), but EOF
+        // was observed, so the permit must already be back in the semaphore.
+        assert_eq!(semaphore.available_permits(), 1);
+        drop(reader);
+        assert_eq!(semaphore.available_permits(), 1);
+    }
+
     #[tokio::test]
     async fn pooled_buffer_reader_keeps_buffer_alive_until_consumed() {
         use tokio::io::AsyncReadExt;
@@ -7783,7 +7851,7 @@ mod tests {
                 "test-bucket",
                 "path/raw",
                 info.clone(),
-                info,
+                Some(info),
                 wrap_reader(tokio::io::empty()),
                 None,
                 None,

rustfs/src/storage/helper.rs

@@ -237,6 +237,13 @@ impl OperationHelper {
         }))
     }
 
+    /// True when a pending event notification still needs the final ObjectInfo.
+    /// Callers can use this to avoid cloning ObjectInfo when the event chain is
+    /// disabled or suppressed.
+    pub fn wants_object_info(&self) -> bool {
+        matches!(self, Self::Enabled(state) if state.event_builder.is_some())
+    }
+
     /// Sets the ObjectInfo for event notification.
     pub fn object(mut self, object_info: ObjectInfo) -> Self {
         if let Self::Enabled(state) = &mut self

rustfs/src/storage/options.rs

@@ -45,7 +45,44 @@ use crate::auth::AuthType;
 use crate::auth::get_query_param;
 use crate::auth::get_request_auth_type_with_query;
 use crate::auth::is_request_presigned_signature_v4_with_query;
+use crate::storage::storage_api::ecstore_bucket::versioning::VersioningApi as _;
 use crate::storage::storage_api::options_consumer::StorageObjectOptions as ObjectOptions;
+use s3s::dto::VersioningConfiguration;
+
+/// Fetch the bucket's versioning configuration once so callers can derive
+/// enabled/suspended state without repeated metadata-sys lookups per request.
+async fn bucket_versioning_config(bucket: &str) -> VersioningConfiguration {
+    match BucketVersioningSys::get(bucket).await {
+        Ok(cfg) => cfg,
+        Err(err) => {
+            tracing::warn!("{:?}", err);
+            VersioningConfiguration::default()
+        }
+    }
+}
+
+/// Whether GET should skip per-shard bitrot verification. Read once: the env
+/// flag is consulted on every GET and `std::env::var` takes a process-global
+/// lock. In tests the env is read directly so `temp_env` overrides apply.
+fn get_skip_verify_bitrot() -> bool {
+    #[cfg(test)]
+    {
+        rustfs_utils::get_env_bool(
+            rustfs_config::ENV_OBJECT_GET_SKIP_BITROT_VERIFY,
+            rustfs_config::DEFAULT_OBJECT_GET_SKIP_BITROT_VERIFY,
+        )
+    }
+    #[cfg(not(test))]
+    {
+        static CACHED: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
+        *CACHED.get_or_init(|| {
+            rustfs_utils::get_env_bool(
+                rustfs_config::ENV_OBJECT_GET_SKIP_BITROT_VERIFY,
+                rustfs_config::DEFAULT_OBJECT_GET_SKIP_BITROT_VERIFY,
+            )
+        })
+    }
+}
 
 /// Creates options for deleting an object in a bucket.
 pub async fn del_opts(
@@ -55,8 +92,9 @@ pub async fn del_opts(
     headers: &HeaderMap<HeaderValue>,
     metadata: HashMap<String, String>,
 ) -> Result<ObjectOptions> {
-    let versioned = BucketVersioningSys::prefix_enabled(bucket, object).await;
-    let version_suspended = BucketVersioningSys::suspended(bucket).await;
+    let versioning_cfg = bucket_versioning_config(bucket).await;
+    let versioned = versioning_cfg.prefix_enabled(object);
+    let version_suspended = versioning_cfg.suspended();
 
     let vid = if vid.is_none() {
         get_header(headers, SUFFIX_SOURCE_VERSION_ID).map(|s| s.into_owned())
@@ -120,8 +158,9 @@ pub async fn get_opts(
     part_num: Option<usize>,
     headers: &HeaderMap<HeaderValue>,
 ) -> Result<ObjectOptions> {
-    let versioned = BucketVersioningSys::prefix_enabled(bucket, object).await;
-    let version_suspended = BucketVersioningSys::prefix_suspended(bucket, object).await;
+    let versioning_cfg = bucket_versioning_config(bucket).await;
+    let versioned = versioning_cfg.prefix_enabled(object);
+    let version_suspended = versioning_cfg.prefix_suspended(object);
 
     let vid = vid.map(|v| v.as_str().trim().to_owned());
 
@@ -159,10 +198,7 @@ pub async fn get_opts(
 
     // Optionally skip per-shard bitrot hash verification on reads to save CPU.
     // Background scanner still performs full integrity checks asynchronously.
-    opts.skip_verify_bitrot = rustfs_utils::get_env_bool(
-        rustfs_config::ENV_OBJECT_GET_SKIP_BITROT_VERIFY,
-        rustfs_config::DEFAULT_OBJECT_GET_SKIP_BITROT_VERIFY,
-    );
+    opts.skip_verify_bitrot = get_skip_verify_bitrot();
 
     fill_conditional_writes_opts_from_header(headers, &mut opts)?;
 
@@ -213,8 +249,9 @@ pub async fn put_opts(
     headers: &HeaderMap<HeaderValue>,
     metadata: HashMap<String, String>,
 ) -> Result<ObjectOptions> {
-    let versioned = BucketVersioningSys::prefix_enabled(bucket, object).await;
-    let version_suspended = BucketVersioningSys::prefix_suspended(bucket, object).await;
+    let versioning_cfg = bucket_versioning_config(bucket).await;
+    let versioned = versioning_cfg.prefix_enabled(object);
+    let version_suspended = versioning_cfg.prefix_suspended(object);
 
     let vid = if vid.is_none() {
         get_header(headers, SUFFIX_SOURCE_VERSION_ID).map(|s| s.into_owned())

rustfs/src/storage/sse.rs

@@ -2208,7 +2208,7 @@ pub fn mark_encrypted_multipart_metadata(metadata: &mut HashMap<String, String>)
     metadata.insert(MINIO_INTERNAL_ENCRYPTION_MULTIPART_HEADER.to_string(), String::new());
 }
 
-fn contains_managed_encryption_metadata(metadata: &HashMap<String, String>) -> bool {
+pub(crate) fn contains_managed_encryption_metadata(metadata: &HashMap<String, String>) -> bool {
     metadata.contains_key(INTERNAL_ENCRYPTION_KEY_HEADER)
         || metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER)
         || metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER)

rustfs/src/storage/storage_api.rs

@@ -431,6 +431,10 @@ pub(crate) mod ecstore_rpc {
     };
 }
 
+pub(crate) mod ecstore_object {
+    pub(crate) use rustfs_ecstore::api::object::{GetObjectBodyCacheHook, register_get_object_body_cache_hook};
+}
+
 pub(crate) mod ecstore_set_disk {
     pub(crate) use rustfs_ecstore::api::set_disk::{DEFAULT_READ_BUFFER_SIZE, get_lock_acquire_timeout, is_valid_storage_class};
 }