perf(memory): add reclaim signals and cache controls (#2689)

crates/config/src/constants/runtime.rs

@@ -59,9 +59,28 @@ pub const DEFAULT_RUNTIME_DIAL9_ROTATION_COUNT: usize = 10;
 pub const DEFAULT_RUNTIME_DIAL9_SAMPLING_RATE: f64 = 1.0; // 100% sampling
 // Note: S3 bucket/prefix have no default; absence means upload is disabled (modeled as Option<String>)
 
+// Allocator reclaim configuration
+pub const ENV_ALLOCATOR_RECLAIM_ENABLED: &str = "RUSTFS_ALLOCATOR_RECLAIM_ENABLED";
+pub const ENV_ALLOCATOR_RECLAIM_INTERVAL_SECS: &str = "RUSTFS_ALLOCATOR_RECLAIM_INTERVAL_SECS";
+pub const ENV_ALLOCATOR_RECLAIM_FORCE: &str = "RUSTFS_ALLOCATOR_RECLAIM_FORCE";
+pub const ENV_ALLOCATOR_RECLAIM_IDLE_INTERVALS: &str = "RUSTFS_ALLOCATOR_RECLAIM_IDLE_INTERVALS";
+pub const DEFAULT_ALLOCATOR_RECLAIM_ENABLED: bool = false;
+pub const DEFAULT_ALLOCATOR_RECLAIM_INTERVAL_SECS: u64 = 30;
+pub const DEFAULT_ALLOCATOR_RECLAIM_FORCE: bool = true;
+pub const DEFAULT_ALLOCATOR_RECLAIM_IDLE_INTERVALS: u64 = 3;
+
+// File page-cache reclaim configuration
+pub const ENV_OBJECT_FILE_CACHE_RECLAIM_WRITE_ENABLE: &str = "RUSTFS_OBJECT_FILE_CACHE_RECLAIM_WRITE_ENABLE";
+pub const ENV_OBJECT_FILE_CACHE_RECLAIM_READ_ENABLE: &str = "RUSTFS_OBJECT_FILE_CACHE_RECLAIM_READ_ENABLE";
+pub const ENV_OBJECT_FILE_CACHE_RECLAIM_THRESHOLD: &str = "RUSTFS_OBJECT_FILE_CACHE_RECLAIM_THRESHOLD";
+pub const DEFAULT_OBJECT_FILE_CACHE_RECLAIM_WRITE_ENABLE: bool = false;
+pub const DEFAULT_OBJECT_FILE_CACHE_RECLAIM_READ_ENABLE: bool = false;
+pub const DEFAULT_OBJECT_FILE_CACHE_RECLAIM_THRESHOLD: usize = 4 * 1024 * 1024;
+
 /// Threshold for small object seek support in megabytes.
 ///
 /// When an object is smaller than this size, rustfs will provide seek support.
 ///
 /// Default is set to 10MB.
+pub const ENV_OBJECT_SEEK_SUPPORT_THRESHOLD: &str = "RUSTFS_OBJECT_SEEK_SUPPORT_THRESHOLD";
 pub const DEFAULT_OBJECT_SEEK_SUPPORT_THRESHOLD: usize = 10 * 1024 * 1024;

crates/ecstore/Cargo.toml

@@ -100,6 +100,7 @@ pin-project-lite.workspace = true
 md-5.workspace = true
 memmap2 = { workspace = true }
 libc.workspace = true
+rustix = { workspace = true }
 rustfs-madmin.workspace = true
 rustfs-workers.workspace = true
 reqwest = { workspace = true }

crates/ecstore/src/disk/local.rs

@@ -31,6 +31,7 @@ use crate::disk::{
 use crate::erasure_coding::bitrot_verify;
 use crate::global::{GLOBAL_IsErasureSD, GLOBAL_RootDiskThreshold};
 use bytes::Bytes;
+use metrics::counter;
 use parking_lot::RwLock as ParkingLotRwLock;
 use rustfs_filemeta::{
     Cache, FileInfo, FileInfoOpts, FileMeta, MetaCacheEntry, MetacacheWriter, ObjectPartInfo, Opts, RawFileInfo, UpdateFn,
@@ -79,6 +80,238 @@ pub enum InternalBuf<'a> {
     Owned(Bytes),
 }
 
+struct FileCacheReclaimWriter {
+    inner: File,
+    reclaim_len: usize,
+    reclaim_on_shutdown: bool,
+    reclaimed: bool,
+}
+
+struct FileCacheReclaimReader {
+    inner: File,
+    reclaim_offset: u64,
+    reclaim_len: usize,
+    reclaim_on_drop: bool,
+    reclaimed: bool,
+}
+
+fn record_file_cache_reclaim_success(kind: &'static str, reclaim_len: usize, started: std::time::Instant) {
+    counter!("rustfs_page_cache_reclaim_requests_total", "kind" => kind.to_string(), "result" => "ok".to_string()).increment(1);
+    counter!("rustfs_page_cache_reclaim_bytes_total", "kind" => kind.to_string()).increment(reclaim_len as u64);
+    metrics::histogram!("rustfs_page_cache_reclaim_duration_seconds", "kind" => kind.to_string())
+        .record(started.elapsed().as_secs_f64());
+}
+
+fn record_file_cache_reclaim_error(kind: &'static str) {
+    counter!("rustfs_page_cache_reclaim_requests_total", "kind" => kind.to_string(), "result" => "err".to_string()).increment(1);
+}
+
+impl FileCacheReclaimReader {
+    fn new(inner: File, reclaim_offset: u64, reclaim_len: usize, reclaim_on_drop: bool) -> Self {
+        #[cfg(target_os = "macos")]
+        if reclaim_on_drop {
+            let _ = set_fd_nocache(&inner);
+        }
+
+        Self {
+            inner,
+            reclaim_offset,
+            reclaim_len,
+            reclaim_on_drop,
+            reclaimed: false,
+        }
+    }
+
+    #[cfg(target_os = "linux")]
+    fn reclaim_file_cache(&mut self) -> std::io::Result<()> {
+        use core::num::NonZeroU64;
+        use rustix::fs::{Advice, fadvise};
+
+        if !self.reclaim_on_drop || self.reclaimed || self.reclaim_len == 0 {
+            return Ok(());
+        }
+
+        let started = std::time::Instant::now();
+        let reclaim_len =
+            NonZeroU64::new(self.reclaim_len as u64).expect("reclaim_len is guaranteed non-zero by the early return");
+        fadvise(&self.inner, self.reclaim_offset, Some(reclaim_len), Advice::DontNeed).map_err(std::io::Error::from)?;
+
+        self.reclaimed = true;
+        record_file_cache_reclaim_success("read", self.reclaim_len, started);
+        Ok(())
+    }
+
+    #[cfg(not(target_os = "linux"))]
+    fn reclaim_file_cache(&mut self) -> std::io::Result<()> {
+        Ok(())
+    }
+}
+
+#[cfg(target_os = "macos")]
+#[allow(unsafe_code)]
+fn set_fd_nocache(file: &File) -> std::io::Result<()> {
+    use std::os::fd::AsRawFd;
+
+    // SAFETY: `fcntl` is called on a valid file descriptor owned by `file`.
+    let ret = unsafe { libc::fcntl(file.as_raw_fd(), libc::F_NOCACHE, 1) };
+    if ret == -1 {
+        return Err(std::io::Error::last_os_error());
+    }
+    Ok(())
+}
+
+#[cfg(target_os = "macos")]
+#[allow(unsafe_code)]
+fn set_std_fd_nocache(file: &std::fs::File) -> std::io::Result<()> {
+    use std::os::fd::AsRawFd;
+
+    // SAFETY: `fcntl` is called on a valid file descriptor owned by `file`.
+    let ret = unsafe { libc::fcntl(file.as_raw_fd(), libc::F_NOCACHE, 1) };
+    if ret == -1 {
+        return Err(std::io::Error::last_os_error());
+    }
+    Ok(())
+}
+
+impl Drop for FileCacheReclaimReader {
+    fn drop(&mut self) {
+        if let Err(err) = self.reclaim_file_cache() {
+            record_file_cache_reclaim_error("read");
+            debug!(error = ?err, reclaim_offset = self.reclaim_offset, reclaim_len = self.reclaim_len, "failed to reclaim file cache after read");
+        }
+    }
+}
+
+impl tokio::io::AsyncRead for FileCacheReclaimReader {
+    fn poll_read(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        buf: &mut tokio::io::ReadBuf<'_>,
+    ) -> std::task::Poll<std::io::Result<()>> {
+        std::pin::Pin::new(&mut self.inner).poll_read(cx, buf)
+    }
+}
+
+impl FileCacheReclaimWriter {
+    fn new(inner: File, reclaim_len: usize, reclaim_on_shutdown: bool) -> Self {
+        #[cfg(target_os = "macos")]
+        if reclaim_on_shutdown {
+            let _ = set_fd_nocache(&inner);
+        }
+
+        Self {
+            inner,
+            reclaim_len,
+            reclaim_on_shutdown,
+            reclaimed: false,
+        }
+    }
+
+    #[cfg(target_os = "linux")]
+    fn reclaim_file_cache(&mut self) -> std::io::Result<()> {
+        use core::num::NonZeroU64;
+        use rustix::fs::{Advice, fadvise};
+
+        if !self.reclaim_on_shutdown || self.reclaimed || self.reclaim_len == 0 {
+            return Ok(());
+        }
+
+        let started = std::time::Instant::now();
+        let reclaim_len =
+            NonZeroU64::new(self.reclaim_len as u64).expect("reclaim_len is guaranteed non-zero by the early return");
+        fadvise(&self.inner, 0, Some(reclaim_len), Advice::DontNeed).map_err(std::io::Error::from)?;
+
+        self.reclaimed = true;
+        record_file_cache_reclaim_success("write", self.reclaim_len, started);
+        Ok(())
+    }
+
+    #[cfg(not(target_os = "linux"))]
+    fn reclaim_file_cache(&mut self) -> std::io::Result<()> {
+        Ok(())
+    }
+}
+
+impl AsyncWrite for FileCacheReclaimWriter {
+    fn poll_write(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        buf: &[u8],
+    ) -> std::task::Poll<std::io::Result<usize>> {
+        std::pin::Pin::new(&mut self.inner).poll_write(cx, buf)
+    }
+
+    fn poll_flush(mut self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> std::task::Poll<std::io::Result<()>> {
+        std::pin::Pin::new(&mut self.inner).poll_flush(cx)
+    }
+
+    fn poll_shutdown(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<std::io::Result<()>> {
+        match std::pin::Pin::new(&mut self.inner).poll_shutdown(cx) {
+            std::task::Poll::Ready(Ok(())) => {
+                if let Err(err) = self.reclaim_file_cache() {
+                    record_file_cache_reclaim_error("write");
+                    debug!(error = ?err, reclaim_len = self.reclaim_len, "failed to reclaim file cache after write");
+                }
+                std::task::Poll::Ready(Ok(()))
+            }
+            other => other,
+        }
+    }
+
+    fn poll_write_vectored(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        bufs: &[std::io::IoSlice<'_>],
+    ) -> std::task::Poll<std::io::Result<usize>> {
+        std::pin::Pin::new(&mut self.inner).poll_write_vectored(cx, bufs)
+    }
+
+    fn is_write_vectored(&self) -> bool {
+        self.inner.is_write_vectored()
+    }
+}
+
+fn should_reclaim_file_cache_after_write(file_size: i64) -> bool {
+    if file_size <= 0 {
+        return false;
+    }
+
+    if !rustfs_utils::get_env_bool(
+        rustfs_config::ENV_OBJECT_FILE_CACHE_RECLAIM_WRITE_ENABLE,
+        rustfs_config::DEFAULT_OBJECT_FILE_CACHE_RECLAIM_WRITE_ENABLE,
+    ) {
+        return false;
+    }
+
+    let threshold = rustfs_utils::get_env_usize(
+        rustfs_config::ENV_OBJECT_FILE_CACHE_RECLAIM_THRESHOLD,
+        rustfs_config::DEFAULT_OBJECT_FILE_CACHE_RECLAIM_THRESHOLD,
+    );
+    file_size as usize >= threshold
+}
+
+fn should_reclaim_file_cache_after_read(length: usize) -> bool {
+    if length == 0 {
+        return false;
+    }
+
+    if !rustfs_utils::get_env_bool(
+        rustfs_config::ENV_OBJECT_FILE_CACHE_RECLAIM_READ_ENABLE,
+        rustfs_config::DEFAULT_OBJECT_FILE_CACHE_RECLAIM_READ_ENABLE,
+    ) {
+        return false;
+    }
+
+    let threshold = rustfs_utils::get_env_usize(
+        rustfs_config::ENV_OBJECT_FILE_CACHE_RECLAIM_THRESHOLD,
+        rustfs_config::DEFAULT_OBJECT_FILE_CACHE_RECLAIM_THRESHOLD,
+    );
+    length >= threshold
+}
+
 pub struct LocalDisk {
     pub root: PathBuf,
     pub format_path: PathBuf,
@@ -1847,8 +2080,9 @@ impl DiskAPI for LocalDisk {
         let f = super::fs::open_file(&file_path, O_CREATE | O_WRONLY)
             .await
             .map_err(to_file_error)?;
+        let reclaim_on_shutdown = should_reclaim_file_cache_after_write(_file_size);
 
-        Ok(Box::new(f))
+        Ok(Box::new(FileCacheReclaimWriter::new(f, _file_size.max(0) as usize, reclaim_on_shutdown)))
 
         // Ok(())
     }
@@ -1920,7 +2154,8 @@ impl DiskAPI for LocalDisk {
             f.seek(SeekFrom::Start(offset as u64)).await?;
         }
 
-        Ok(Box::new(f))
+        let reclaim_on_drop = should_reclaim_file_cache_after_read(length);
+        Ok(Box::new(FileCacheReclaimReader::new(f, offset as u64, length, reclaim_on_drop)))
     }
 
     /// Zero-copy file read using memory mapping (Unix) or efficient read (non-Unix).
@@ -1965,9 +2200,15 @@ impl DiskAPI for LocalDisk {
             use memmap2::MmapOptions;
             let file_path_clone = file_path.clone();
 
+            let should_reclaim_after_read = should_reclaim_file_cache_after_read(length);
             let bytes = tokio::task::spawn_blocking(move || {
                 let file = std::fs::File::open(&file_path_clone).map_err(DiskError::from)?;
 
+                #[cfg(target_os = "macos")]
+                if should_reclaim_after_read {
+                    let _ = set_std_fd_nocache(&file);
+                }
+
                 // mmap offsets on Unix must be page-size aligned. Align the
                 // mapping down to the nearest page boundary, then slice out the
                 // originally requested logical range.
@@ -1995,7 +2236,21 @@ impl DiskAPI for LocalDisk {
                 let end = logical_offset
                     .checked_add(length)
                     .ok_or_else(|| DiskError::other("mmap slice length overflow"))?;
-                Ok::<Bytes, DiskError>(Bytes::copy_from_slice(&mmap[logical_offset..end]))
+                let bytes = Bytes::copy_from_slice(&mmap[logical_offset..end]);
+
+                #[cfg(target_os = "linux")]
+                if should_reclaim_after_read {
+                    use core::num::NonZeroU64;
+                    use rustix::fs::{Advice, fadvise};
+
+                    let reclaim_len =
+                        NonZeroU64::new(map_len as u64).ok_or_else(|| DiskError::other("mmap reclaim length overflow"))?;
+                    fadvise(&file, aligned_offset, Some(reclaim_len), Advice::DontNeed)
+                        .map_err(std::io::Error::from)
+                        .map_err(DiskError::from)?;
+                }
+
+                Ok::<Bytes, DiskError>(bytes)
             })
             .await
             .map_err(DiskError::from)??;
@@ -3372,4 +3627,32 @@ mod test {
             assert_eq!(normalize_path_components("C:\\a\\..\\b"), PathBuf::from("C:\\b"));
         }
     }
+
+    #[test]
+    fn should_reclaim_file_cache_after_write_respects_env_and_threshold() {
+        temp_env::with_var_unset(rustfs_config::ENV_OBJECT_FILE_CACHE_RECLAIM_WRITE_ENABLE, || {
+            assert!(!should_reclaim_file_cache_after_write(8 * 1024 * 1024));
+        });
+
+        temp_env::with_var(rustfs_config::ENV_OBJECT_FILE_CACHE_RECLAIM_WRITE_ENABLE, Some("true"), || {
+            temp_env::with_var(rustfs_config::ENV_OBJECT_FILE_CACHE_RECLAIM_THRESHOLD, Some("4194304"), || {
+                assert!(should_reclaim_file_cache_after_write(8 * 1024 * 1024));
+                assert!(!should_reclaim_file_cache_after_write(1024));
+            });
+        });
+    }
+
+    #[test]
+    fn should_reclaim_file_cache_after_read_respects_env_and_threshold() {
+        temp_env::with_var_unset(rustfs_config::ENV_OBJECT_FILE_CACHE_RECLAIM_READ_ENABLE, || {
+            assert!(!should_reclaim_file_cache_after_read(8 * 1024 * 1024));
+        });
+
+        temp_env::with_var(rustfs_config::ENV_OBJECT_FILE_CACHE_RECLAIM_READ_ENABLE, Some("true"), || {
+            temp_env::with_var(rustfs_config::ENV_OBJECT_FILE_CACHE_RECLAIM_THRESHOLD, Some("4194304"), || {
+                assert!(should_reclaim_file_cache_after_read(8 * 1024 * 1024));
+                assert!(!should_reclaim_file_cache_after_read(1024));
+            });
+        });
+    }
 }

crates/ecstore/src/erasure_coding/encode.rs

@@ -27,7 +27,7 @@ use tokio::sync::mpsc;
 use tracing::error;
 
 const ENV_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES: &str = "RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES";
-const DEFAULT_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES: usize = 32 * 1024 * 1024;
+const DEFAULT_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES: usize = 8 * 1024 * 1024;
 const DEFAULT_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BLOCKS: usize = 8;
 
 fn encode_channel_capacity(expanded_block_bytes: usize, max_inflight_bytes: usize) -> usize {
@@ -40,6 +40,16 @@ fn encode_channel_capacity(expanded_block_bytes: usize, max_inflight_bytes: usiz
         .clamp(1, DEFAULT_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BLOCKS)
 }
 
+fn queued_block_bytes(block: &[Bytes]) -> usize {
+    block.iter().map(Bytes::len).sum()
+}
+
+async fn drain_queued_inflight_bytes(rx: &mut mpsc::Receiver<Vec<Bytes>>) {
+    while let Some(block) = rx.recv().await {
+        rustfs_io_metrics::remove_ec_encode_inflight_bytes(queued_block_bytes(&block));
+    }
+}
+
 pub(crate) struct MultiWriter<'a> {
     writers: &'a mut [Option<BitrotWriterWrapper>],
     write_quorum: usize,
@@ -217,7 +227,10 @@ impl Erasure {
                     Ok(n) if n > 0 => {
                         total += n;
                         let res = self.encode_data(&buf[..n])?;
+                        let queued_bytes = queued_block_bytes(&res);
+                        rustfs_io_metrics::add_ec_encode_inflight_bytes(queued_bytes);
                         if let Err(err) = tx.send(res).await {
+                            rustfs_io_metrics::remove_ec_encode_inflight_bytes(queued_bytes);
                             return Err(std::io::Error::other(format!("Failed to send encoded data : {err}")));
                         }
                     }
@@ -250,6 +263,8 @@ impl Erasure {
             if block.is_empty() {
                 break;
             }
+            let queued_bytes = queued_block_bytes(&block);
+            rustfs_io_metrics::remove_ec_encode_inflight_bytes(queued_bytes);
             if let Err(err) = writers.write(block).await {
                 write_err = Some(err);
                 break;
@@ -259,6 +274,7 @@ impl Erasure {
         if let Some(err) = write_err {
             task.abort();
             let _ = task.await;
+            drain_queued_inflight_bytes(&mut rx).await;
             if let Err(shutdown_err) = writers.shutdown().await {
                 error!("failed to shutdown erasure writers after write error: {:?}", shutdown_err);
             }

crates/heal/src/heal/manager.rs

@@ -237,6 +237,14 @@ impl PriorityHealQueue {
     }
 }
 
+fn publish_active_heal_count(active_heals: &HashMap<String, Arc<HealTask>>) {
+    crate::set_heal_active_tasks(active_heals.len());
+}
+
+fn publish_heal_queue_length(queue: &PriorityHealQueue) {
+    crate::set_heal_queue_length(queue.len());
+}
+
 /// Heal config
 #[derive(Debug, Clone)]
 pub struct HealConfig {
@@ -420,6 +428,8 @@ impl HealManager {
             }
         }
         active_heals.clear();
+        publish_active_heal_count(&active_heals);
+        crate::set_heal_queue_length(0);
 
         // update state
         let mut state = self.state.write().await;
@@ -435,6 +445,7 @@ impl HealManager {
         let mut queue = self.heal_queue.lock().await;
 
         let queue_len = queue.len();
+        publish_heal_queue_length(&queue);
         let queue_capacity = config.queue_size;
 
         if queue.contains_key(&request) {
@@ -505,6 +516,7 @@ impl HealManager {
 
         let push_outcome = queue.push(request);
         debug_assert_eq!(push_outcome, QueuePushOutcome::Accepted);
+        publish_heal_queue_length(&queue);
 
         // Log queue statistics periodically (when adding high/urgent priority items)
         if matches!(priority, HealPriority::High | HealPriority::Urgent) {
@@ -544,7 +556,9 @@ impl HealManager {
 
     /// Get task progress
     pub async fn get_active_tasks_count(&self) -> usize {
-        self.active_heals.lock().await.len()
+        let active_heals = self.active_heals.lock().await;
+        publish_active_heal_count(&active_heals);
+        active_heals.len()
     }
 
     pub async fn get_task_progress(&self, task_id: &str) -> Result<HealProgress> {
@@ -564,6 +578,7 @@ impl HealManager {
         if let Some(task) = active_heals.get(task_id) {
             task.cancel().await?;
             active_heals.remove(task_id);
+            publish_active_heal_count(&active_heals);
             info!("Cancelled heal task: {}", task_id);
             Ok(())
         } else {
@@ -581,12 +596,14 @@ impl HealManager {
     /// Get active task count
     pub async fn get_active_task_count(&self) -> usize {
         let active_heals = self.active_heals.lock().await;
+        publish_active_heal_count(&active_heals);
         active_heals.len()
     }
 
     /// Get queue length
     pub async fn get_queue_length(&self) -> usize {
         let queue = self.heal_queue.lock().await;
+        publish_heal_queue_length(&queue);
         queue.len()
     }
 
@@ -731,6 +748,7 @@ impl HealManager {
                             );
                             let mut queue = heal_queue.lock().await;
                             if matches!(queue.push(req), QueuePushOutcome::Accepted) {
+                                publish_heal_queue_length(&queue);
                                 let config = config.read().await;
                                 if config.event_driven_scheduler_enable {
                                     notify.notify_one();
@@ -757,6 +775,7 @@ impl HealManager {
     ) {
         let config = config.read().await;
         let mut active_heals_guard = active_heals.lock().await;
+        publish_active_heal_count(&active_heals_guard);
 
         // Check if new heal tasks can be started
         let active_count = active_heals_guard.len();
@@ -769,6 +788,7 @@ impl HealManager {
 
         let mut queue = heal_queue.lock().await;
         let queue_len = queue.len();
+        publish_heal_queue_length(&queue);
 
         if queue_len == 0 {
             return;
@@ -804,6 +824,7 @@ impl HealManager {
                 let task = Arc::new(HealTask::from_request(request, storage.clone()));
                 let task_id = task.id.clone();
                 active_heals_guard.insert(task_id.clone(), task.clone());
+                publish_active_heal_count(&active_heals_guard);
                 update_task_running_metric_for_task(&active_heals_guard, task.as_ref());
                 let active_heals_clone = active_heals.clone();
                 let statistics_clone = statistics.clone();
@@ -828,6 +849,7 @@ impl HealManager {
                     }
                     let mut active_heals_guard = active_heals_clone.lock().await;
                     if let Some(completed_task) = active_heals_guard.remove(&task_id) {
+                        publish_active_heal_count(&active_heals_guard);
                         update_task_running_metric_for_task(&active_heals_guard, completed_task.as_ref());
                         // update statistics
                         let mut stats = statistics_clone.write().await;
@@ -853,6 +875,8 @@ impl HealManager {
         let mut stats = statistics.write().await;
         stats.total_tasks += tasks_started as u64;
         stats.update_running_tasks(active_heals_guard.len() as u64);
+        publish_active_heal_count(&active_heals_guard);
+        publish_heal_queue_length(&queue);
 
         // Log queue status if items remain
         if !queue.is_empty() {

crates/heal/src/lib.rs

@@ -17,6 +17,7 @@ pub mod heal;
 
 pub use error::{Error, Result};
 pub use heal::{HealManager, HealOptions, HealPriority, HealRequest, HealType, channel::HealChannelProcessor};
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, OnceLock};
 use tokio_util::sync::CancellationToken;
 use tracing::{error, info};
@@ -55,6 +56,8 @@ static GLOBAL_HEAL_MANAGER: OnceLock<Arc<HealManager>> = OnceLock::new();
 
 /// Global heal channel processor instance
 static GLOBAL_HEAL_CHANNEL_PROCESSOR: OnceLock<Arc<tokio::sync::Mutex<HealChannelProcessor>>> = OnceLock::new();
+static GLOBAL_HEAL_ACTIVE_TASKS: AtomicU64 = AtomicU64::new(0);
+static GLOBAL_HEAL_QUEUE_LENGTH: AtomicU64 = AtomicU64::new(0);
 
 /// Initialize and start heal manager with channel processor
 pub async fn init_heal_manager(
@@ -107,3 +110,19 @@ pub fn get_heal_manager() -> Option<&'static Arc<HealManager>> {
 pub fn get_heal_channel_processor() -> Option<&'static Arc<tokio::sync::Mutex<HealChannelProcessor>>> {
     GLOBAL_HEAL_CHANNEL_PROCESSOR.get()
 }
+
+pub fn current_heal_active_tasks() -> u64 {
+    GLOBAL_HEAL_ACTIVE_TASKS.load(Ordering::Relaxed)
+}
+
+pub fn current_heal_queue_length() -> u64 {
+    GLOBAL_HEAL_QUEUE_LENGTH.load(Ordering::Relaxed)
+}
+
+pub(crate) fn set_heal_active_tasks(count: usize) {
+    GLOBAL_HEAL_ACTIVE_TASKS.store(count as u64, Ordering::Relaxed);
+}
+
+pub(crate) fn set_heal_queue_length(count: usize) {
+    GLOBAL_HEAL_QUEUE_LENGTH.store(count as u64, Ordering::Relaxed);
+}

crates/io-metrics/src/lib.rs

@@ -49,6 +49,8 @@
 #[macro_use]
 extern crate metrics;
 
+use std::sync::atomic::{AtomicU64, Ordering};
+
 // Public modules
 pub mod adaptive_ttl;
 pub mod autotuner;
@@ -137,6 +139,43 @@ pub use config::{
 pub use collector::MetricsCollector;
 pub use performance::PerformanceMetrics;
 
+static EC_ENCODE_INFLIGHT_BYTES: AtomicU64 = AtomicU64::new(0);
+static GET_OBJECT_BUFFERED_BYTES: AtomicU64 = AtomicU64::new(0);
+
+fn saturating_sub_atomic(counter: &AtomicU64, bytes: u64) -> u64 {
+    let mut current = counter.load(Ordering::Relaxed);
+    loop {
+        let next = current.saturating_sub(bytes);
+        match counter.compare_exchange_weak(current, next, Ordering::Relaxed, Ordering::Relaxed) {
+            Ok(_) => return next,
+            Err(actual) => current = actual,
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum TrackedMemoryGauge {
+    GetObjectBufferedBytes,
+}
+
+/// Drop-based guard for tracked in-memory payloads.
+#[derive(Debug)]
+pub struct MemoryGaugeGuard {
+    gauge: TrackedMemoryGauge,
+    bytes: u64,
+}
+
+impl Drop for MemoryGaugeGuard {
+    fn drop(&mut self) {
+        match self.gauge {
+            TrackedMemoryGauge::GetObjectBufferedBytes => {
+                let next = saturating_sub_atomic(&GET_OBJECT_BUFFERED_BYTES, self.bytes);
+                gauge!("rustfs_get_object_buffered_bytes_current").set(next as f64);
+            }
+        }
+    }
+}
+
 /// Record GetObject request start.
 #[inline(always)]
 pub fn record_get_object_request_start(concurrent_requests: usize) {
@@ -544,6 +583,85 @@ pub fn record_memory_usage(used_bytes: u64, total_bytes: u64) {
     }
 }
 
+/// Record process-level memory split metrics.
+#[inline(always)]
+pub fn record_process_memory_split(resident_bytes: u64, virtual_bytes: u64) {
+    gauge!("rustfs_memory_process_resident_bytes").set(resident_bytes as f64);
+    gauge!("rustfs_memory_process_virtual_bytes").set(virtual_bytes as f64);
+}
+
+/// Record cgroup memory split metrics when available.
+#[inline(always)]
+pub fn record_cgroup_memory_split(
+    current_bytes: Option<u64>,
+    limit_bytes: Option<u64>,
+    anon_bytes: Option<u64>,
+    file_bytes: Option<u64>,
+    active_file_bytes: Option<u64>,
+    inactive_file_bytes: Option<u64>,
+) {
+    if let Some(current_bytes) = current_bytes {
+        gauge!("rustfs_memory_cgroup_current_bytes").set(current_bytes as f64);
+    }
+    if let Some(limit_bytes) = limit_bytes {
+        gauge!("rustfs_memory_cgroup_limit_bytes").set(limit_bytes as f64);
+    }
+    if let Some(anon_bytes) = anon_bytes {
+        gauge!("rustfs_memory_cgroup_anon_bytes").set(anon_bytes as f64);
+    }
+    if let Some(file_bytes) = file_bytes {
+        gauge!("rustfs_memory_cgroup_file_bytes").set(file_bytes as f64);
+    }
+    if let Some(active_file_bytes) = active_file_bytes {
+        gauge!("rustfs_memory_cgroup_active_file_bytes").set(active_file_bytes as f64);
+    }
+    if let Some(inactive_file_bytes) = inactive_file_bytes {
+        gauge!("rustfs_memory_cgroup_inactive_file_bytes").set(inactive_file_bytes as f64);
+    }
+}
+
+/// Track encoded bytes currently queued between erasure encode and disk writers.
+#[inline(always)]
+pub fn add_ec_encode_inflight_bytes(bytes: usize) {
+    let next = EC_ENCODE_INFLIGHT_BYTES.fetch_add(bytes as u64, Ordering::Relaxed) + bytes as u64;
+    gauge!("rustfs_ec_encode_inflight_bytes_current").set(next as f64);
+}
+
+/// Remove encoded bytes from the tracked erasure encode in-flight gauge.
+#[inline(always)]
+pub fn remove_ec_encode_inflight_bytes(bytes: usize) {
+    let next = saturating_sub_atomic(&EC_ENCODE_INFLIGHT_BYTES, bytes as u64);
+    gauge!("rustfs_ec_encode_inflight_bytes_current").set(next as f64);
+}
+
+/// Return the current tracked EC encode in-flight bytes.
+#[inline(always)]
+pub fn current_ec_encode_inflight_bytes() -> u64 {
+    EC_ENCODE_INFLIGHT_BYTES.load(Ordering::Relaxed)
+}
+
+/// Track whole-object buffering on the GET path.
+#[inline(always)]
+pub fn track_get_object_buffered_bytes(bytes: usize) -> Option<MemoryGaugeGuard> {
+    if bytes == 0 {
+        return None;
+    }
+
+    let next = GET_OBJECT_BUFFERED_BYTES.fetch_add(bytes as u64, Ordering::Relaxed) + bytes as u64;
+    gauge!("rustfs_get_object_buffered_bytes_current").set(next as f64);
+
+    Some(MemoryGaugeGuard {
+        gauge: TrackedMemoryGauge::GetObjectBufferedBytes,
+        bytes: bytes as u64,
+    })
+}
+
+/// Return the current tracked GET whole-buffered bytes.
+#[inline(always)]
+pub fn current_get_object_buffered_bytes() -> u64 {
+    GET_OBJECT_BUFFERED_BYTES.load(Ordering::Relaxed)
+}
+
 /// Record CPU usage.
 ///
 /// # Arguments
@@ -758,6 +876,48 @@ mod tests {
         record_memory_usage(2 * 1024 * 1024 * 1024, 8 * 1024 * 1024 * 1024);
     }
 
+    #[test]
+    fn test_record_process_memory_split() {
+        record_process_memory_split(1024, 2048);
+        record_process_memory_split(4096, 8192);
+    }
+
+    #[test]
+    fn test_record_cgroup_memory_split() {
+        record_cgroup_memory_split(Some(1), Some(2), Some(3), Some(4), Some(5), Some(6));
+        record_cgroup_memory_split(None, None, None, None, None, None);
+    }
+
+    #[test]
+    fn test_ec_encode_inflight_bytes_tracking() {
+        EC_ENCODE_INFLIGHT_BYTES.store(0, Ordering::Relaxed);
+        add_ec_encode_inflight_bytes(1024);
+        add_ec_encode_inflight_bytes(2048);
+        remove_ec_encode_inflight_bytes(1024);
+        remove_ec_encode_inflight_bytes(2048);
+        remove_ec_encode_inflight_bytes(4096);
+        assert_eq!(current_ec_encode_inflight_bytes(), 0);
+    }
+
+    #[test]
+    fn test_get_object_buffered_bytes_guard() {
+        GET_OBJECT_BUFFERED_BYTES.store(0, Ordering::Relaxed);
+        drop(track_get_object_buffered_bytes(1024));
+        let guard = track_get_object_buffered_bytes(2048);
+        drop(guard);
+        assert_eq!(current_get_object_buffered_bytes(), 0);
+    }
+
+    #[test]
+    fn test_get_object_buffered_bytes_guard_saturates_on_underflow() {
+        GET_OBJECT_BUFFERED_BYTES.store(1024, Ordering::Relaxed);
+        drop(MemoryGaugeGuard {
+            gauge: TrackedMemoryGauge::GetObjectBufferedBytes,
+            bytes: 2048,
+        });
+        assert_eq!(current_get_object_buffered_bytes(), 0);
+    }
+
     #[test]
     fn test_record_cpu_usage() {
         record_cpu_usage(25.5);

crates/scanner/src/lib.rs

@@ -32,3 +32,26 @@ pub use data_usage_define::*;
 pub use error::ScannerError;
 pub use scanner::init_data_scanner;
 pub use sleeper::{DynamicSleeper, SCANNER_IDLE_MODE, SCANNER_SLEEPER};
+use std::sync::atomic::{AtomicU64, Ordering};
+
+static SCANNER_ACTIVE_WORK_UNITS: AtomicU64 = AtomicU64::new(0);
+
+pub fn current_scanner_activity() -> u64 {
+    SCANNER_ACTIVE_WORK_UNITS.load(Ordering::Relaxed)
+}
+
+pub(crate) struct ScannerActivityGuard;
+
+impl ScannerActivityGuard {
+    pub(crate) fn new() -> Self {
+        SCANNER_ACTIVE_WORK_UNITS.fetch_add(1, Ordering::Relaxed);
+        Self
+    }
+}
+
+impl Drop for ScannerActivityGuard {
+    fn drop(&mut self) {
+        let _ = SCANNER_ACTIVE_WORK_UNITS
+            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| Some(current.saturating_sub(1)));
+    }
+}

crates/scanner/src/scanner.rs

@@ -18,7 +18,7 @@ use crate::data_usage_define::{BACKGROUND_HEAL_INFO_PATH, DATA_USAGE_BLOOM_NAME_
 use crate::scanner_folder::data_usage_update_dir_cycles;
 use crate::scanner_io::ScannerIO;
 use crate::sleeper::SCANNER_SLEEPER;
-use crate::{DataUsageInfo, ScannerError};
+use crate::{DataUsageInfo, ScannerActivityGuard, ScannerError};
 use chrono::{DateTime, Utc};
 use rustfs_common::heal_channel::HealScanMode;
 use rustfs_common::metrics::{CurrentCycle, Metric, Metrics, emit_scan_cycle_complete, global_metrics};
@@ -183,6 +183,7 @@ fn get_lock_acquire_timeout() -> Duration {
 
 #[instrument(skip_all)]
 async fn run_data_scanner_cycle(ctx: &CancellationToken, storeapi: &Arc<ECStore>, cycle_info: &mut CurrentCycle) {
+    let _activity_guard = ScannerActivityGuard::new();
     SCANNER_SLEEPER.refresh_from_env();
     info!("Start run data scanner cycle");
     cycle_info.current = cycle_info.next;
@@ -324,6 +325,7 @@ pub async fn store_data_usage_in_backend(
     let mut attempts = 1u32;
 
     while let Some(data_usage_info) = receiver.recv().await {
+        let _activity_guard = ScannerActivityGuard::new();
         if ctx.is_cancelled() {
             break;
         }