diff --git a/crates/config/src/constants/capacity.rs b/crates/config/src/constants/capacity.rs
index f9650242b..7afb50590 100644
--- a/crates/config/src/constants/capacity.rs
+++ b/crates/config/src/constants/capacity.rs
@@ -62,32 +62,32 @@ pub const ENV_CAPACITY_STALL_TIMEOUT: &str = "RUSTFS_CAPACITY_STALL_TIMEOUT";
 // ============================================================================
 
 /// Scheduled update interval in seconds
-/// Default: 300 seconds (5 minutes)
-pub const DEFAULT_SCHEDULED_UPDATE_INTERVAL_SECS: u64 = 300;
+/// Default: 120 seconds (2 minutes)
+pub const DEFAULT_SCHEDULED_UPDATE_INTERVAL_SECS: u64 = 120;
 
 /// Write trigger delay in seconds
-/// Default: 10 seconds
-pub const DEFAULT_WRITE_TRIGGER_DELAY_SECS: u64 = 10;
+/// Default: 5 seconds
+pub const DEFAULT_WRITE_TRIGGER_DELAY_SECS: u64 = 5;
 
 /// Write frequency threshold (writes per minute)
-/// Default: 10 writes/minute
-pub const DEFAULT_WRITE_FREQUENCY_THRESHOLD: usize = 10;
+/// Default: 5 writes/minute
+pub const DEFAULT_WRITE_FREQUENCY_THRESHOLD: usize = 5;
 
 /// Fast update threshold in seconds
-/// Default: 60 seconds
-pub const DEFAULT_FAST_UPDATE_THRESHOLD_SECS: u64 = 60;
+/// Default: 30 seconds
+pub const DEFAULT_FAST_UPDATE_THRESHOLD_SECS: u64 = 30;
 
 /// Maximum files threshold for sampling
-/// Default: 1,000,000 files
-pub const DEFAULT_MAX_FILES_THRESHOLD: usize = 1_000_000;
+/// Default: 200,000 files
+pub const DEFAULT_MAX_FILES_THRESHOLD: usize = 200_000;
 
 /// Statistics timeout in seconds
-/// Default: 5 seconds
-pub const DEFAULT_STAT_TIMEOUT_SECS: u64 = 5;
+/// Default: 3 seconds
+pub const DEFAULT_STAT_TIMEOUT_SECS: u64 = 3;
 
 /// Sampling rate (1 in every N files)
-/// Default: 100
-pub const DEFAULT_SAMPLE_RATE: usize = 100;
+/// Default: 200
+pub const DEFAULT_SAMPLE_RATE: usize = 200;
 
 /// Follow symbolic links during capacity calculation
 /// Default: false (disabled for safety)
@@ -102,16 +102,16 @@ pub const DEFAULT_CAPACITY_MAX_SYMLINK_DEPTH: u8 = 3;
 pub const DEFAULT_CAPACITY_ENABLE_DYNAMIC_TIMEOUT: bool = true;
 
 /// Minimum capacity calculation timeout in seconds
-/// Default: 5 seconds
-pub const DEFAULT_CAPACITY_MIN_TIMEOUT_SECS: u64 = 5;
+/// Default: 2 seconds
+pub const DEFAULT_CAPACITY_MIN_TIMEOUT_SECS: u64 = 2;
 
 /// Maximum capacity calculation timeout in seconds
-/// Default: 60 seconds
-pub const DEFAULT_CAPACITY_MAX_TIMEOUT_SECS: u64 = 60;
+/// Default: 15 seconds
+pub const DEFAULT_CAPACITY_MAX_TIMEOUT_SECS: u64 = 15;
 
 /// Progress stall detection timeout in seconds
-/// Default: 1 second (no progress for 1 second = stall)
-pub const DEFAULT_CAPACITY_STALL_TIMEOUT_SECS: u64 = 1;
+/// Default: 20 seconds
+pub const DEFAULT_CAPACITY_STALL_TIMEOUT_SECS: u64 = 20;
 
 // ============================================================================
 // Tests
@@ -140,16 +140,16 @@ mod tests {
 
     #[test]
     fn test_default_values() {
-        assert_eq!(DEFAULT_SCHEDULED_UPDATE_INTERVAL_SECS, 300);
-        assert_eq!(DEFAULT_WRITE_TRIGGER_DELAY_SECS, 10);
-        assert_eq!(DEFAULT_WRITE_FREQUENCY_THRESHOLD, 10);
-        assert_eq!(DEFAULT_FAST_UPDATE_THRESHOLD_SECS, 60);
-        assert_eq!(DEFAULT_MAX_FILES_THRESHOLD, 1_000_000);
-        assert_eq!(DEFAULT_STAT_TIMEOUT_SECS, 5);
-        assert_eq!(DEFAULT_SAMPLE_RATE, 100);
+        assert_eq!(DEFAULT_SCHEDULED_UPDATE_INTERVAL_SECS, 120);
+        assert_eq!(DEFAULT_WRITE_TRIGGER_DELAY_SECS, 5);
+        assert_eq!(DEFAULT_WRITE_FREQUENCY_THRESHOLD, 5);
+        assert_eq!(DEFAULT_FAST_UPDATE_THRESHOLD_SECS, 30);
+        assert_eq!(DEFAULT_MAX_FILES_THRESHOLD, 200_000);
+        assert_eq!(DEFAULT_STAT_TIMEOUT_SECS, 3);
+        assert_eq!(DEFAULT_SAMPLE_RATE, 200);
         assert_eq!(DEFAULT_CAPACITY_MAX_SYMLINK_DEPTH, 3);
-        assert_eq!(DEFAULT_CAPACITY_MIN_TIMEOUT_SECS, 5);
-        assert_eq!(DEFAULT_CAPACITY_MAX_TIMEOUT_SECS, 60);
-        assert_eq!(DEFAULT_CAPACITY_STALL_TIMEOUT_SECS, 1);
+        assert_eq!(DEFAULT_CAPACITY_MIN_TIMEOUT_SECS, 2);
+        assert_eq!(DEFAULT_CAPACITY_MAX_TIMEOUT_SECS, 15);
+        assert_eq!(DEFAULT_CAPACITY_STALL_TIMEOUT_SECS, 20);
     }
 }
diff --git a/crates/io-metrics/src/capacity_metrics.rs b/crates/io-metrics/src/capacity_metrics.rs
new file mode 100644
index 000000000..070d67cc9
--- /dev/null
+++ b/crates/io-metrics/src/capacity_metrics.rs
@@ -0,0 +1,92 @@
+// 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.
+
+//! Capacity metrics recording helpers.
+
+use metrics::{counter, gauge, histogram};
+use std::time::Duration;
+
+/// Record capacity cache hit.
+#[inline(always)]
+pub fn record_capacity_cache_hit() {
+    counter!("rustfs.capacity.cache.hits").increment(1);
+}
+
+/// Record capacity cache miss.
+#[inline(always)]
+pub fn record_capacity_cache_miss() {
+    counter!("rustfs.capacity.cache.misses").increment(1);
+}
+
+/// Record current capacity gauge.
+#[inline(always)]
+pub fn record_capacity_current_bytes(used_bytes: u64) {
+    gauge!("rustfs.capacity.current").set(used_bytes as f64);
+}
+
+/// Record capacity update completion.
+#[inline(always)]
+pub fn record_capacity_update_completed(source: &str, duration: Duration, used_bytes: u64, is_estimated: bool) {
+    counter!("rustfs.capacity.update.total", "source" => source.to_string()).increment(1);
+    histogram!("rustfs.capacity.update.duration.seconds", "source" => source.to_string()).record(duration.as_secs_f64());
+    histogram!("rustfs.capacity.update.bytes", "source" => source.to_string()).record(used_bytes as f64);
+    counter!("rustfs.capacity.update.estimated.total", "source" => source.to_string(), "estimated" => is_estimated.to_string())
+        .increment(1);
+}
+
+/// Record failed capacity update.
+#[inline(always)]
+pub fn record_capacity_update_failed(source: &str) {
+    counter!("rustfs.capacity.update.failures", "source" => source.to_string()).increment(1);
+}
+
+/// Record capacity write activity.
+#[inline(always)]
+pub fn record_capacity_write_operation(write_frequency: usize) {
+    counter!("rustfs.capacity.write.operations").increment(1);
+    gauge!("rustfs.capacity.write.frequency").set(write_frequency as f64);
+}
+
+/// Record symlink accounting.
+#[inline(always)]
+pub fn record_capacity_symlink(size_bytes: u64) {
+    counter!("rustfs.capacity.symlinks.encountered").increment(1);
+    histogram!("rustfs.capacity.symlinks.size.bytes").record(size_bytes as f64);
+}
+
+/// Record timeout fallback event.
+#[inline(always)]
+pub fn record_capacity_timeout_fallback() {
+    counter!("rustfs.capacity.timeout.fallback").increment(1);
+}
+
+/// Record stall detection event.
+#[inline(always)]
+pub fn record_capacity_stall_detected() {
+    counter!("rustfs.capacity.timeout.stall").increment(1);
+}
+
+/// Record dynamic timeout usage.
+#[inline(always)]
+pub fn record_capacity_dynamic_timeout(timeout: Duration) {
+    counter!("rustfs.capacity.timeout.dynamic").increment(1);
+    histogram!("rustfs.capacity.timeout.dynamic.seconds").record(timeout.as_secs_f64());
+}
+
+/// Record scan sampling outcome.
+#[inline(always)]
+pub fn record_capacity_scan_sampling(sampled_count: usize, estimated: bool) {
+    histogram!("rustfs.capacity.scan.sampled.count").record(sampled_count as f64);
+    counter!("rustfs.capacity.scan.estimated.total", "estimated" => estimated.to_string()).increment(1);
+}
diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs
index faacff974..3a0f1ce46 100644
--- a/crates/io-metrics/src/lib.rs
+++ b/crates/io-metrics/src/lib.rs
@@ -52,6 +52,7 @@ pub mod adaptive_ttl;
 pub mod autotuner;
 pub mod backpressure_metrics;
 pub mod cache_config;
+pub mod capacity_metrics;
 pub mod collector;
 pub mod config;
 pub mod deadlock_metrics;
@@ -71,6 +72,13 @@ pub use adaptive_ttl::{
     record_ttl_expiration,
 };
 
+// Capacity metrics exports
+pub use capacity_metrics::{
+    record_capacity_cache_hit, record_capacity_cache_miss, record_capacity_current_bytes, record_capacity_dynamic_timeout,
+    record_capacity_scan_sampling, record_capacity_stall_detected, record_capacity_symlink, record_capacity_timeout_fallback,
+    record_capacity_update_completed, record_capacity_update_failed, record_capacity_write_operation,
+};
+
 // I/O metrics exports
 pub use io_metrics::{
     IoSchedulerStats, record_bandwidth_observation, record_buffer_size_adjustment, record_io_priority_decision,
diff --git a/rustfs/src/app/admin_usecase.rs b/rustfs/src/app/admin_usecase.rs
index 7e2090ee3..8b4d5117f 100644
--- a/rustfs/src/app/admin_usecase.rs
+++ b/rustfs/src/app/admin_usecase.rs
@@ -16,10 +16,9 @@
 
 use crate::app::context::{AppContext, get_global_app_context};
 use crate::capacity::capacity_manager::{
-    DataSource, get_capacity_manager, get_enable_dynamic_timeout, get_follow_symlinks, get_max_files_threshold,
+    CapacityUpdate, DataSource, get_capacity_manager, get_enable_dynamic_timeout, get_follow_symlinks, get_max_files_threshold,
     get_max_symlink_depth, get_max_timeout, get_min_timeout, get_sample_rate, get_stall_timeout, get_stat_timeout,
 };
-use crate::capacity::capacity_metrics::get_capacity_metrics;
 use crate::error::ApiError;
 use rustfs_common::data_usage::DataUsageInfo;
 use rustfs_ecstore::admin_server_info::get_server_info;
@@ -28,6 +27,10 @@ use rustfs_ecstore::endpoints::EndpointServerPools;
 use rustfs_ecstore::new_object_layer_fn;
 use rustfs_ecstore::pools::{PoolStatus, get_total_usable_capacity, get_total_usable_capacity_free};
 use rustfs_ecstore::store_api::StorageAPI;
+use rustfs_io_metrics::{
+    record_capacity_dynamic_timeout, record_capacity_scan_sampling, record_capacity_stall_detected, record_capacity_symlink,
+    record_capacity_timeout_fallback,
+};
 use rustfs_madmin::{InfoMessage, StorageInfo};
 use s3s::S3ErrorCode;
 use std::collections::HashSet;
@@ -44,6 +47,31 @@ pub struct QueryServerInfoRequest {
     pub include_pools: bool,
 }
 
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub(crate) struct CapacityScanResult {
+    pub used_bytes: u64,
+    pub file_count: usize,
+    pub sampled_count: usize,
+    pub is_estimated: bool,
+    pub scan_duration: Duration,
+    pub had_partial_errors: bool,
+}
+
+impl CapacityScanResult {
+    fn with_partial_errors(mut self) -> Self {
+        self.had_partial_errors = true;
+        self
+    }
+
+    pub(crate) fn to_capacity_update(self) -> CapacityUpdate {
+        if self.is_estimated {
+            CapacityUpdate::estimated(self.used_bytes, self.file_count)
+        } else {
+            CapacityUpdate::exact(self.used_bytes, self.file_count)
+        }
+    }
+}
+
 pub struct QueryServerInfoResponse {
     pub info: InfoMessage,
 }
@@ -69,47 +97,66 @@ pub struct QueryPoolStatusRequest {
 /// Calculate actual used capacity of all data directories
 pub(crate) async fn calculate_data_dir_used_capacity(
     disks: &[rustfs_madmin::Disk],
-) -> Result<u64, Box<dyn std::error::Error + Send + Sync>> {
+) -> Result<CapacityScanResult, Box<dyn std::error::Error + Send + Sync>> {
+    let start = Instant::now();
     let mut total_used = 0u64;
+    let mut total_files = 0usize;
+    let mut total_sampled = 0usize;
     let mut has_failure = false;
     let mut has_success = false;
+    let mut is_estimated = false;
 
     for disk in disks {
         let path = Path::new(&disk.drive_path);
 
-        // Check if path exists
         if !path.exists() {
             warn!("Data directory does not exist: {}", disk.drive_path);
             has_failure = true;
             continue;
         }
 
-        // Asynchronously calculate directory size
         match get_dir_size_async(path).await {
-            Ok(size) => {
-                debug!("Data directory {} size: {} bytes", disk.drive_path, size);
-                total_used += size;
+            Ok(scan) => {
+                debug!(
+                    "Data directory {} size: {} bytes, files={}, sampled={}, estimated={}",
+                    disk.drive_path, scan.used_bytes, scan.file_count, scan.sampled_count, scan.is_estimated
+                );
+                total_used += scan.used_bytes;
+                total_files += scan.file_count;
+                total_sampled += scan.sampled_count;
+                is_estimated |= scan.is_estimated;
+                has_failure |= scan.had_partial_errors;
                 has_success = true;
             }
             Err(e) => {
                 warn!("Failed to get size for directory {}: {:?}", disk.drive_path, e);
                 has_failure = true;
-                // Continue with other directories
             }
         }
     }
 
-    // If all directories failed, return error to trigger fallback
     if !has_success {
         return Err("All directories failed to calculate size".into());
     }
 
-    // Log warning if there were some failures
     if has_failure {
         warn!("Some directories failed to calculate size, result may be incomplete");
     }
 
-    Ok(total_used)
+    let mut result = CapacityScanResult {
+        used_bytes: total_used,
+        file_count: total_files,
+        sampled_count: total_sampled,
+        is_estimated,
+        scan_duration: start.elapsed(),
+        had_partial_errors: false,
+    };
+
+    if has_failure {
+        result = result.with_partial_errors();
+    }
+
+    Ok(result)
 }
 
 // ============================================================================
@@ -159,11 +206,7 @@ impl SymlinkTracker {
         self.visited.insert(path);
         self.symlink_count += 1;
         self.symlink_size += size;
-
-        // Record to metrics
-        if let Ok(metrics) = std::panic::catch_unwind(std::panic::AssertUnwindSafe(get_capacity_metrics)) {
-            metrics.record_symlink(size);
-        }
+        record_capacity_symlink(size);
     }
 
     /// Get symlink statistics
@@ -269,11 +312,8 @@ impl ProgressMonitor {
                 files_processed, elapsed, dynamic_timeout
             );
 
-            // Record timeout to metrics
-            if let Ok(metrics) = std::panic::catch_unwind(std::panic::AssertUnwindSafe(get_capacity_metrics))
-                && self.used_dynamic_timeout
-            {
-                metrics.record_dynamic_timeout();
+            if self.enable_dynamic_timeout {
+                record_capacity_dynamic_timeout(dynamic_timeout);
             }
 
             return Err(std::io::Error::new(
@@ -294,10 +334,7 @@ impl ProgressMonitor {
                     self.stall_timeout, files_processed
                 );
 
-                // Record stall to metrics
-                if let Ok(metrics) = std::panic::catch_unwind(std::panic::AssertUnwindSafe(get_capacity_metrics)) {
-                    metrics.record_stall_detected();
-                }
+                record_capacity_stall_detected();
 
                 return Err(std::io::Error::new(
                     std::io::ErrorKind::TimedOut,
@@ -314,17 +351,14 @@ impl ProgressMonitor {
 
     /// Record timeout fallback to sampling
     fn record_timeout_fallback(&self) {
-        if let Ok(metrics) = std::panic::catch_unwind(std::panic::AssertUnwindSafe(get_capacity_metrics)) {
-            metrics.record_timeout_fallback();
-        }
+        record_capacity_timeout_fallback();
     }
 }
 
 /// Asynchronously get directory size with enhanced symlink handling and dynamic timeout
-async fn get_dir_size_async(path: &Path) -> Result<u64, std::io::Error> {
+async fn get_dir_size_async(path: &Path) -> Result<CapacityScanResult, std::io::Error> {
     let path = path.to_path_buf();
 
-    // Get configuration values
     let max_files_threshold = get_max_files_threshold();
     let base_timeout = get_stat_timeout();
     let min_timeout = get_min_timeout();
@@ -335,7 +369,6 @@ async fn get_dir_size_async(path: &Path) -> Result<u64, std::io::Error> {
     let follow_symlinks = get_follow_symlinks();
     let max_symlink_depth = get_max_symlink_depth();
 
-    // Ensure sample_rate is never zero to avoid panics in is_multiple_of
     let effective_sample_rate = if sample_rate == 0 {
         warn!("Invalid sampling configuration: sample_rate=0. Clamping to 1 to avoid panic.");
         1
@@ -343,7 +376,6 @@ async fn get_dir_size_async(path: &Path) -> Result<u64, std::io::Error> {
         sample_rate
     };
 
-    // Check if path exists before traversing
     if !path.exists() {
         return Err(std::io::Error::new(
             std::io::ErrorKind::NotFound,
@@ -351,15 +383,14 @@ async fn get_dir_size_async(path: &Path) -> Result<u64, std::io::Error> {
         ));
     }
 
-    // Use tokio::task::spawn_blocking to avoid blocking the async runtime
     tokio::task::spawn_blocking(move || {
         let start_time = Instant::now();
-        let mut total_size = 0u64;
+        let mut exact_prefix_bytes = 0u64;
+        let mut overflow_sampled_bytes = 0u64;
         let mut file_count = 0usize;
-        let mut sampled_size = 0u64;
         let mut sampled_count = 0usize;
+        let mut had_partial_errors = false;
 
-        // Initialize symlink tracker and progress monitor
         let mut symlink_tracker = if follow_symlinks {
             Some(SymlinkTracker::new(max_symlink_depth))
         } else {
@@ -369,7 +400,6 @@ async fn get_dir_size_async(path: &Path) -> Result<u64, std::io::Error> {
         let mut progress_monitor =
             ProgressMonitor::new(base_timeout, min_timeout, max_timeout, stall_timeout, enable_dynamic_timeout);
 
-        // Build WalkDir with appropriate settings
         let mut walker_builder = WalkDir::new(&path);
         if !follow_symlinks {
             walker_builder = walker_builder.follow_links(false);
@@ -377,79 +407,87 @@ async fn get_dir_size_async(path: &Path) -> Result<u64, std::io::Error> {
         let walker = walker_builder.into_iter();
 
         for entry_result in walker {
-            // Propagate traversal errors instead of silently dropping them
             let entry = match entry_result {
                 Ok(entry) => entry,
                 Err(err) => {
                     warn!("Failed to traverse directory entry under {:?}: {}", path, err);
-                    return Err(std::io::Error::other(err.to_string()));
-                }
-            };
-
-            // Get file metadata
-            let metadata = match entry.metadata() {
-                Ok(meta) => meta,
-                Err(err) => {
-                    warn!("Failed to get metadata for {:?}: {}", entry.path(), err);
+                    had_partial_errors = true;
+                    continue;
+                }
+            };
+
+            let metadata = match entry.metadata() {
+                Ok(meta) => meta,
+                Err(err) => {
+                    warn!("Failed to get metadata for {:?}: {}", entry.path(), err);
+                    had_partial_errors = true;
                     continue;
                 }
             };
 
-            // Handle symlinks if enabled
             if metadata.is_symlink() {
                 if let Some(ref mut tracker) = symlink_tracker
                     && let Ok(target) = std::fs::read_link(entry.path())
                     && tracker.should_follow(&target, 0)
                 {
                     tracker.record_symlink(target, metadata.len());
-                    // Don't count symlink size itself, only target
-                    continue;
                 }
-                // If not following symlinks, skip
                 continue;
             }
 
-            // Only count file sizes, ignore directories
             if !metadata.is_file() {
                 continue;
             }
 
             file_count += 1;
+            let exact_count = file_count.min(max_files_threshold);
+            let avg_size = if exact_count > 0 {
+                exact_prefix_bytes / exact_count as u64
+            } else {
+                0
+            };
 
-            // Update progress and check for timeout/stall
-            let avg_size = if file_count > 0 { total_size / file_count as u64 } else { 0 };
             if let Err(e) = progress_monitor.update_and_check_timeout(file_count, avg_size) {
-                // Timeout or stall detected
                 if sampled_count > 0 {
-                    info!("Timeout/stall at {} files, using sampled estimate", file_count);
+                    let overflow_count = file_count.saturating_sub(max_files_threshold);
+                    let estimated_overflow = overflow_sampled_bytes.saturating_mul(overflow_count as u64) / sampled_count as u64;
+                    let estimated_total = exact_prefix_bytes.saturating_add(estimated_overflow);
+                    info!(
+                        "Timeout/stall at {} files, using sampled estimate: exact_prefix={} overflow_estimate={} sampled={}",
+                        file_count, exact_prefix_bytes, estimated_overflow, sampled_count
+                    );
                     progress_monitor.record_timeout_fallback();
-                    return Ok(sampled_size * file_count as u64 / sampled_count as u64);
+                    record_capacity_scan_sampling(sampled_count, true);
+                    return Ok(CapacityScanResult {
+                        used_bytes: estimated_total,
+                        file_count,
+                        sampled_count,
+                        is_estimated: true,
+                        scan_duration: start_time.elapsed(),
+                        had_partial_errors,
+                    });
                 }
                 return Err(e);
             }
 
-            // When file count exceeds threshold, enable sampling
-            if file_count > max_files_threshold {
-                // Sampling: count 1 in every effective_sample_rate files
-                if file_count.is_multiple_of(effective_sample_rate) {
-                    sampled_size += metadata.len();
+            if file_count <= max_files_threshold {
+                exact_prefix_bytes += metadata.len();
+            } else {
+                let overflow_index = file_count - max_files_threshold;
+                if overflow_index.is_multiple_of(effective_sample_rate) {
+                    overflow_sampled_bytes += metadata.len();
                     sampled_count += 1;
                 }
 
-                // Log progress every 100k files
                 if file_count.is_multiple_of(100_000) {
                     debug!(
-                        "Processed {} files, sampled {} files, size: {} bytes",
-                        file_count, sampled_count, sampled_size
+                        "Processed {} files, exact_prefix_bytes={}, sampled_overflow={} files/{} bytes",
+                        file_count, exact_prefix_bytes, sampled_count, overflow_sampled_bytes
                     );
                 }
-            } else {
-                // Below threshold, full statistics
-                total_size += metadata.len();
             }
         }
 
-        // Report symlink statistics if tracking was enabled
         if let Some(tracker) = symlink_tracker {
             let (count, size) = tracker.get_stats();
             if count > 0 {
@@ -457,22 +495,68 @@ async fn get_dir_size_async(path: &Path) -> Result<u64, std::io::Error> {
             }
         }
 
-        // If sampling was enabled, return estimated value
         if file_count > max_files_threshold && sampled_count > 0 {
-            let estimated_size = sampled_size * file_count as u64 / sampled_count as u64;
+            let overflow_count = file_count - max_files_threshold;
+            let estimated_overflow = overflow_sampled_bytes.saturating_mul(overflow_count as u64) / sampled_count as u64;
+            let estimated_size = exact_prefix_bytes.saturating_add(estimated_overflow);
             info!(
-                "Large directory detected: {} files, estimated size: {} bytes (sampled {}/{} files)",
-                file_count, estimated_size, sampled_count, file_count
+                "Large directory detected: {} files, estimated size: {} bytes (exact prefix: {}, sampled overflow {}/{})",
+                file_count, estimated_size, exact_prefix_bytes, sampled_count, overflow_count
             );
-            Ok(estimated_size)
+            record_capacity_scan_sampling(sampled_count, true);
+            Ok(CapacityScanResult {
+                used_bytes: estimated_size,
+                file_count,
+                sampled_count,
+                is_estimated: true,
+                scan_duration: start_time.elapsed(),
+                had_partial_errors,
+            })
+        } else if file_count > max_files_threshold {
+            // sampled_count == 0: too few overflow files to reach the sample rate threshold.
+            // Fall back to estimating the overflow using the average file size from the exact
+            // prefix so that overflow files are not silently dropped from the total.
+            let overflow_count = file_count - max_files_threshold;
+            // Use the actual number of files counted in the exact prefix, not the threshold
+            // value, to avoid a divide-by-zero or incorrect average when fewer files were
+            // processed than max_files_threshold.
+            let exact_prefix_count = file_count.min(max_files_threshold) as u64;
+            let avg_prefix_size = if exact_prefix_count > 0 {
+                exact_prefix_bytes / exact_prefix_count
+            } else {
+                0
+            };
+            let estimated_overflow = avg_prefix_size.saturating_mul(overflow_count as u64);
+            let estimated_size = exact_prefix_bytes.saturating_add(estimated_overflow);
+            info!(
+                "Large directory detected: {} files, estimated size: {} bytes (no overflow samples, used prefix average {} bytes/file)",
+                file_count, estimated_size, avg_prefix_size
+            );
+            record_capacity_scan_sampling(0, true);
+            Ok(CapacityScanResult {
+                used_bytes: estimated_size,
+                file_count,
+                sampled_count: 0,
+                is_estimated: true,
+                scan_duration: start_time.elapsed(),
+                had_partial_errors,
+            })
         } else {
+            record_capacity_scan_sampling(0, false);
             debug!(
                 "Directory size calculation completed: {} files, {} bytes, took {:?}",
                 file_count,
-                total_size,
+                exact_prefix_bytes,
                 start_time.elapsed()
             );
-            Ok(total_size)
+            Ok(CapacityScanResult {
+                used_bytes: exact_prefix_bytes,
+                file_count,
+                sampled_count,
+                is_estimated: false,
+                scan_duration: start_time.elapsed(),
+                had_partial_errors,
+            })
         }
     })
     .await
@@ -616,47 +700,65 @@ impl DefaultAdminUsecase {
             if cache_age < fast_update_threshold {
                 info.total_used_capacity = cached.total_used;
                 debug!(
-                    "Using cached capacity: {} bytes (age: {:?}, source: {:?})",
-                    cached.total_used, cache_age, cached.source
+                    "Using cached capacity: {} bytes (age: {:?}, source: {:?}, files={}, estimated={})",
+                    cached.total_used, cache_age, cached.source, cached.file_count, cached.is_estimated
                 );
             } else {
                 // Cache is stale, check if we need fast update
                 let needs_update = capacity_manager.needs_fast_update().await;
+                let should_block = capacity_manager.should_block_on_refresh(cache_age);
 
-                if needs_update {
-                    // Fast update needed (recent writes or high frequency)
+                if needs_update && should_block {
                     let start = Instant::now();
-                    match calculate_data_dir_used_capacity(&storage_info.disks).await {
-                        Ok(used_capacity) => {
-                            info.total_used_capacity = used_capacity;
-                            capacity_manager
-                                .update_capacity(used_capacity, DataSource::WriteTriggered)
-                                .await;
+                    match capacity_manager
+                        .refresh_or_join(DataSource::WriteTriggered, || async {
+                            calculate_data_dir_used_capacity(&storage_info.disks)
+                                .await
+                                .map(|scan| scan.to_capacity_update())
+                                .map_err(|e| e.to_string())
+                        })
+                        .await
+                    {
+                        Ok(update) => {
+                            info.total_used_capacity = update.total_used;
 
                             let elapsed = start.elapsed();
-                            debug!("Fast capacity update completed in {:?}", elapsed);
+                            debug!(
+                                "Foreground capacity refresh completed in {:?} (files={}, estimated={})",
+                                elapsed, update.file_count, update.is_estimated
+                            );
                         }
                         Err(e) => {
-                            warn!("Fast capacity update failed: {:?}, using cached value", e);
+                            warn!("Foreground capacity refresh failed: {}, using cached value", e);
                             info.total_used_capacity = cached.total_used;
                         }
                     }
                 } else {
-                    // Use stale cache and trigger background update (if not already in progress)
                     info.total_used_capacity = cached.total_used;
-                    debug!("Using stale cache, background update will be triggered if not already in progress");
+                    debug!(
+                        "Using stale cached capacity: {} bytes (age: {:?}, source: {:?}, files={}, estimated={}, needs_update={}, blocking={})",
+                        cached.total_used,
+                        cache_age,
+                        cached.source,
+                        cached.file_count,
+                        cached.is_estimated,
+                        needs_update,
+                        should_block
+                    );
 
-                    // Trigger background update only if not already in progress (prevent thundering herd)
-                    if capacity_manager.try_start_background_update() {
-                        let disks = storage_info.disks.clone();
-                        let manager = capacity_manager.clone();
-                        tokio::spawn(async move {
-                            if let Ok(new_capacity) = calculate_data_dir_used_capacity(&disks).await {
-                                manager.update_capacity(new_capacity, DataSource::Scheduled).await;
-                                debug!("Background capacity update completed: {} bytes", new_capacity);
-                            }
-                            manager.complete_background_update();
-                        });
+                    let disks = storage_info.disks.clone();
+                    let manager = capacity_manager.clone();
+                    if manager
+                        .clone()
+                        .spawn_refresh_if_needed(DataSource::Scheduled, move || async move {
+                            calculate_data_dir_used_capacity(&disks)
+                                .await
+                                .map(|scan| scan.to_capacity_update())
+                                .map_err(|e| e.to_string())
+                        })
+                        .await
+                    {
+                        debug!("Background capacity update started");
                     } else {
                         debug!("Background update already in progress, skipping spawn");
                     }
@@ -665,21 +767,33 @@ impl DefaultAdminUsecase {
         } else {
             // No cache, perform initial calculation
             let start = Instant::now();
-            match calculate_data_dir_used_capacity(&storage_info.disks).await {
-                Ok(used_capacity) => {
-                    info.total_used_capacity = used_capacity;
-                    capacity_manager.update_capacity(used_capacity, DataSource::RealTime).await;
+            match capacity_manager
+                .refresh_or_join(DataSource::RealTime, || async {
+                    calculate_data_dir_used_capacity(&storage_info.disks)
+                        .await
+                        .map(|scan| scan.to_capacity_update())
+                        .map_err(|e| e.to_string())
+                })
+                .await
+            {
+                Ok(update) => {
+                    info.total_used_capacity = update.total_used;
 
                     let elapsed = start.elapsed();
-                    info!("Initial capacity calculation completed: {} bytes in {:?}", used_capacity, elapsed);
+                    info!(
+                        "Initial capacity calculation completed: {} bytes in {:?} (files={}, estimated={})",
+                        update.total_used, elapsed, update.file_count, update.is_estimated
+                    );
                 }
                 Err(e) => {
                     warn!(
-                        "Failed to calculate data directory used capacity: {:?}, falling back to disk used capacity",
+                        "Failed to calculate data directory used capacity: {}, falling back to disk used capacity",
                         e
                     );
-                    // Fallback: use disk used capacity
                     info.total_used_capacity = info.total_capacity.saturating_sub(info.total_free_capacity);
+                    capacity_manager
+                        .update_capacity(CapacityUpdate::fallback(info.total_used_capacity), DataSource::Fallback)
+                        .await;
                 }
             }
         }
@@ -805,7 +919,8 @@ mod tests {
 
         let temp_dir = TempDir::new().unwrap();
         let size = get_dir_size_async(temp_dir.path()).await.unwrap();
-        assert_eq!(size, 0);
+        assert_eq!(size.used_bytes, 0);
+        assert_eq!(size.file_count, 0);
     }
 
     #[tokio::test]
@@ -820,7 +935,8 @@ mod tests {
         file.write_all(b"Hello, World!").unwrap();
 
         let size = get_dir_size_async(temp_dir.path()).await.unwrap();
-        assert_eq!(size, 13);
+        assert_eq!(size.used_bytes, 13);
+        assert_eq!(size.file_count, 1);
     }
 
     #[tokio::test]
@@ -839,7 +955,8 @@ mod tests {
         }
 
         let size = get_dir_size_async(temp_dir.path()).await.unwrap();
-        assert_eq!(size, 40); // 10 files * 4 bytes
+        assert_eq!(size.used_bytes, 40); // 10 files * 4 bytes
+        assert_eq!(size.file_count, 10);
     }
 
     #[tokio::test]
@@ -863,7 +980,8 @@ mod tests {
         f2.write_all(b"content2").unwrap();
 
         let size = get_dir_size_async(temp_dir.path()).await.unwrap();
-        assert_eq!(size, 16); // "content1" (8) + "content2" (8)
+        assert_eq!(size.used_bytes, 16); // "content1" (8) + "content2" (8)
+        assert_eq!(size.file_count, 2);
     }
 
     #[tokio::test]
diff --git a/rustfs/src/capacity/capacity_integration.rs b/rustfs/src/capacity/capacity_integration.rs
index cce4ecb98..82dcffb29 100644
--- a/rustfs/src/capacity/capacity_integration.rs
+++ b/rustfs/src/capacity/capacity_integration.rs
@@ -15,9 +15,8 @@
 //! Capacity management integration for application startup
 
 use crate::capacity::capacity_manager::{DataSource, get_capacity_manager, start_background_task};
-use crate::capacity::capacity_metrics::{get_capacity_metrics, start_metrics_logging};
 use rustfs_ecstore::disk::DiskAPI;
-use std::time::Duration;
+use rustfs_io_metrics::{record_capacity_cache_hit, record_capacity_cache_miss};
 use tracing::{info, warn};
 
 /// Initialize capacity management system
@@ -50,11 +49,6 @@ pub async fn init_capacity_management() {
     info!("Starting background capacity update task...");
     start_background_task(disk_refs).await;
 
-    // Start metrics logging (log every 10 minutes)
-    let metrics_interval = Duration::from_secs(600);
-    info!("Starting metrics logging task (interval: {:?})...", metrics_interval);
-    start_metrics_logging(metrics_interval).await;
-
     info!("Capacity management system initialized successfully");
 }
 
@@ -62,11 +56,10 @@ pub async fn init_capacity_management() {
 #[allow(dead_code)]
 pub async fn get_capacity_with_metrics() -> Option<(u64, String)> {
     let manager = get_capacity_manager();
-    let metrics = get_capacity_metrics();
 
     // Check cache
     if let Some(cached) = manager.get_capacity().await {
-        metrics.record_cache_hit();
+        record_capacity_cache_hit();
 
         let source = match cached.source {
             DataSource::RealTime => "real-time",
@@ -78,19 +71,21 @@ pub async fn get_capacity_with_metrics() -> Option<(u64, String)> {
         return Some((cached.total_used, source.to_string()));
     }
 
-    metrics.record_cache_miss();
+    record_capacity_cache_miss();
     None
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::capacity::capacity_manager::{DataSource, get_capacity_manager};
+    use crate::capacity::capacity_manager::{CapacityUpdate, DataSource, get_capacity_manager};
 
     #[tokio::test]
     async fn test_get_capacity_with_metrics() {
         let manager = get_capacity_manager();
-        manager.update_capacity(1000, DataSource::RealTime).await;
+        manager
+            .update_capacity(CapacityUpdate::exact(1000, 0), DataSource::RealTime)
+            .await;
 
         let result = get_capacity_with_metrics().await;
         assert!(result.is_some());
diff --git a/rustfs/src/capacity/capacity_manager.rs b/rustfs/src/capacity/capacity_manager.rs
index 3db201717..8f1096215 100644
--- a/rustfs/src/capacity/capacity_manager.rs
+++ b/rustfs/src/capacity/capacity_manager.rs
@@ -15,7 +15,6 @@
 //! Hybrid Capacity Manager for efficient capacity statistics
 
 use crate::app::admin_usecase::calculate_data_dir_used_capacity;
-use metrics::{counter, gauge};
 use rustfs_config::{
     DEFAULT_CAPACITY_ENABLE_DYNAMIC_TIMEOUT, DEFAULT_CAPACITY_FOLLOW_SYMLINKS, DEFAULT_CAPACITY_MAX_SYMLINK_DEPTH,
     DEFAULT_CAPACITY_MAX_TIMEOUT_SECS, DEFAULT_CAPACITY_MIN_TIMEOUT_SECS, DEFAULT_CAPACITY_STALL_TIMEOUT_SECS,
@@ -26,12 +25,13 @@ use rustfs_config::{
     ENV_CAPACITY_SAMPLE_RATE, ENV_CAPACITY_SCHEDULED_INTERVAL, ENV_CAPACITY_STALL_TIMEOUT, ENV_CAPACITY_STAT_TIMEOUT,
     ENV_CAPACITY_WRITE_FREQUENCY_THRESHOLD, ENV_CAPACITY_WRITE_TRIGGER_DELAY,
 };
+use rustfs_io_metrics::{record_capacity_current_bytes, record_capacity_update_completed, record_capacity_write_operation};
 use rustfs_utils::{get_env_bool, get_env_u64, get_env_usize};
+use std::future::Future;
 use std::sync::Arc;
-use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::{Duration, Instant};
-use tokio::sync::RwLock;
-use tracing::{debug, error, info, warn};
+use tokio::sync::{Mutex, RwLock, watch};
+use tracing::{debug, info, warn};
 
 // ============================================================================
 // Configuration Functions
@@ -279,15 +279,53 @@ pub struct CachedCapacity {
     /// Last update time
     pub last_update: Instant,
     /// File count (optional)
-    #[allow(dead_code)]
     pub file_count: usize,
     /// Whether it's an estimated value
-    #[allow(dead_code)]
     pub is_estimated: bool,
     /// Data source
     pub source: DataSource,
 }
 
+/// Structured capacity update payload.
+#[derive(Clone, Debug)]
+pub struct CapacityUpdate {
+    /// Total used capacity in bytes.
+    pub total_used: u64,
+    /// Number of files observed during scan.
+    pub file_count: usize,
+    /// Whether the value is estimated instead of exact.
+    pub is_estimated: bool,
+}
+
+impl CapacityUpdate {
+    /// Create an exact capacity update.
+    pub fn exact(total_used: u64, file_count: usize) -> Self {
+        Self {
+            total_used,
+            file_count,
+            is_estimated: false,
+        }
+    }
+
+    /// Create an estimated capacity update.
+    pub fn estimated(total_used: u64, file_count: usize) -> Self {
+        Self {
+            total_used,
+            file_count,
+            is_estimated: true,
+        }
+    }
+
+    /// Create a fallback capacity update.
+    pub fn fallback(total_used: u64) -> Self {
+        Self {
+            total_used,
+            file_count: 0,
+            is_estimated: true,
+        }
+    }
+}
+
 #[derive(Clone, Debug, PartialEq, Copy, Eq)]
 pub enum DataSource {
     /// Real-time statistics
@@ -301,6 +339,17 @@ pub enum DataSource {
     Fallback,
 }
 
+impl DataSource {
+    fn as_metric_label(self) -> &'static str {
+        match self {
+            Self::RealTime => "realtime",
+            Self::Scheduled => "scheduled",
+            Self::WriteTriggered => "write_triggered",
+            Self::Fallback => "fallback",
+        }
+    }
+}
+
 /// Write record for tracking write operations
 #[derive(Debug)]
 pub struct WriteRecord {
@@ -354,6 +403,25 @@ impl HybridStrategyConfig {
 // Hybrid Capacity Manager
 // ============================================================================
 
+struct RefreshState {
+    running: bool,
+    /// Sender for the current refresh cycle. Joiners subscribe to this before releasing the
+    /// mutex so they cannot miss the completion notification. A new channel is created at the
+    /// start of every refresh cycle so stale subscribers from previous cycles are not confused
+    /// by results that were already published.
+    result_tx: watch::Sender<Option<Result<CapacityUpdate, String>>>,
+}
+
+impl Default for RefreshState {
+    fn default() -> Self {
+        let (tx, _) = watch::channel(None);
+        Self {
+            running: false,
+            result_tx: tx,
+        }
+    }
+}
+
 /// Hybrid capacity manager
 pub struct HybridCapacityManager {
     /// Capacity cache
@@ -362,11 +430,17 @@ pub struct HybridCapacityManager {
     write_record: Arc<RwLock<WriteRecord>>,
     /// Configuration
     config: HybridStrategyConfig,
-    /// Background update in progress flag
-    update_in_progress: Arc<AtomicBool>,
+    /// Shared singleflight refresh state
+    refresh_state: Arc<Mutex<RefreshState>>,
 }
 
 impl HybridCapacityManager {
+    fn max_stale_age(&self) -> Duration {
+        self.config
+            .scheduled_update_interval
+            .max(self.config.fast_update_threshold.checked_mul(3).unwrap_or(Duration::MAX))
+    }
+
     /// Create a new hybrid capacity manager
     pub fn new(config: HybridStrategyConfig) -> Self {
         Self {
@@ -377,7 +451,7 @@ impl HybridCapacityManager {
                 write_window: Vec::new(),
             })),
             config,
-            update_in_progress: Arc::new(AtomicBool::new(false)),
+            refresh_state: Arc::new(Mutex::new(RefreshState::default())),
         }
     }
 
@@ -393,25 +467,23 @@ impl HybridCapacityManager {
     }
 
     /// Update capacity
-    pub async fn update_capacity(&self, capacity: u64, source: DataSource) {
+    pub async fn update_capacity(&self, update: CapacityUpdate, source: DataSource) {
+        let start = Instant::now();
         let mut cache = self.cache.write().await;
         *cache = Some(CachedCapacity {
-            total_used: capacity,
+            total_used: update.total_used,
             last_update: Instant::now(),
-            file_count: 0,
-            is_estimated: false,
+            file_count: update.file_count,
+            is_estimated: update.is_estimated,
             source,
         });
 
-        debug!("Capacity updated: {} bytes, source: {:?}", capacity, source);
-        // Update metrics
-        gauge!("rustfs.capacity.current").set(capacity as f64);
-        match source {
-            DataSource::RealTime => counter!("rustfs.capacity.update.realtime").increment(1),
-            DataSource::Scheduled => counter!("rustfs.capacity.update.scheduled").increment(1),
-            DataSource::WriteTriggered => counter!("rustfs.capacity.update.write_triggered").increment(1),
-            DataSource::Fallback => counter!("rustfs.capacity.update.fallback").increment(1),
-        }
+        debug!(
+            "Capacity updated: {} bytes, files={}, estimated={}, source: {:?}",
+            update.total_used, update.file_count, update.is_estimated, source
+        );
+        record_capacity_current_bytes(update.total_used);
+        record_capacity_update_completed(source.as_metric_label(), start.elapsed(), update.total_used, update.is_estimated);
     }
 
     /// Record write operation
@@ -432,8 +504,7 @@ impl HybridCapacityManager {
             record.write_window.push(now);
         }
 
-        counter!("rustfs.capacity.write.operations").increment(1);
-        gauge!("rustfs.capacity.write.frequency").set(record.write_window.len() as f64);
+        record_capacity_write_operation(record.write_window.len());
         debug!(
             "Write operation recorded: total writes = {}, recent writes = {}",
             record.write_count,
@@ -490,21 +561,110 @@ impl HybridCapacityManager {
         record.write_window.len()
     }
 
+    /// Run a singleflight refresh. Callers either join an existing in-flight refresh or become the leader.
+    ///
+    /// Joiners subscribe to the watch channel *before* releasing the mutex, which guarantees
+    /// they cannot miss the completion notification even if the leader finishes very quickly.
+    pub async fn refresh_or_join<F, Fut>(&self, source: DataSource, refresh_fn: F) -> Result<CapacityUpdate, String>
+    where
+        F: FnOnce() -> Fut,
+        Fut: Future<Output = Result<CapacityUpdate, String>>,
+    {
+        let maybe_rx = {
+            let mut state = self.refresh_state.lock().await;
+            if state.running {
+                // Subscribe while holding the lock so the send that completes the current
+                // refresh cycle cannot happen before we are subscribed.
+                Some(state.result_tx.subscribe())
+            } else {
+                // Become the leader. Create a fresh channel so that joiners from a previous
+                // cycle cannot observe the result that was published for the new cycle.
+                let (tx, _) = watch::channel(None);
+                state.result_tx = tx;
+                state.running = true;
+                None
+            }
+        };
+
+        if let Some(mut result_rx) = maybe_rx {
+            // Wait until the leader publishes Some(result). Because we subscribed before
+            // releasing the mutex, we cannot miss the notification.
+            if result_rx.wait_for(|v| v.is_some()).await.is_err() {
+                // The leader's sender was dropped (e.g. due to a panic) without publishing
+                // a result. Surface a clear error rather than silently returning the default.
+                return Err("capacity refresh leader exited without publishing a result".to_string());
+            }
+            return result_rx
+                .borrow()
+                .as_ref()
+                .cloned()
+                .unwrap_or_else(|| Err("capacity refresh completed without a result".to_string()));
+        }
+
+        let result = refresh_fn().await;
+        if let Ok(update) = &result {
+            self.update_capacity(update.clone(), source).await;
+        }
+
+        {
+            let mut state = self.refresh_state.lock().await;
+            state.running = false;
+            let _ = state.result_tx.send(Some(result.clone()));
+        }
+
+        result
+    }
+
+    /// Start a background refresh if one is not already in flight.
+    pub async fn spawn_refresh_if_needed<F, Fut>(self: Arc<Self>, source: DataSource, refresh_fn: F) -> bool
+    where
+        F: FnOnce() -> Fut + Send + 'static,
+        Fut: Future<Output = Result<CapacityUpdate, String>> + Send + 'static,
+    {
+        let should_spawn = {
+            let mut state = self.refresh_state.lock().await;
+            if state.running {
+                false
+            } else {
+                let (tx, _) = watch::channel(None);
+                state.result_tx = tx;
+                state.running = true;
+                true
+            }
+        };
+
+        if !should_spawn {
+            return false;
+        }
+
+        tokio::spawn(async move {
+            let result = refresh_fn().await;
+            if let Ok(update) = &result {
+                self.update_capacity(update.clone(), source).await;
+            }
+
+            let mut state = self.refresh_state.lock().await;
+            state.running = false;
+            let _ = state.result_tx.send(Some(result));
+        });
+
+        true
+    }
+
     /// Get config
     pub fn get_config(&self) -> &HybridStrategyConfig {
         &self.config
     }
 
-    /// Try to start a background update, returns true if update was started (false if already in progress)
-    pub fn try_start_background_update(&self) -> bool {
-        self.update_in_progress
-            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
-            .is_ok()
+    /// Check if the cache is too stale to keep serving without a foreground refresh.
+    pub fn should_block_on_refresh(&self, cache_age: Duration) -> bool {
+        cache_age >= self.max_stale_age()
     }
 
-    /// Mark background update as complete
-    pub fn complete_background_update(&self) {
-        self.update_in_progress.store(false, Ordering::Release);
+    /// Return whether a refresh is currently in flight.
+    #[cfg(test)]
+    pub async fn refresh_in_progress(&self) -> bool {
+        self.refresh_state.lock().await.running
     }
 }
 
@@ -526,7 +686,9 @@ pub fn get_capacity_manager() -> Arc<HybridCapacityManager> {
 /// # Example
 /// ```no_run
 /// let manager = create_isolated_manager(HybridStrategyConfig::default());
-/// manager.update_capacity(1000, DataSource::RealTime).await;
+/// manager
+///     .update_capacity(CapacityUpdate::exact(1000, 0), DataSource::RealTime)
+///     .await;
 /// ```
 #[cfg(test)]
 #[allow(dead_code)]
@@ -553,17 +715,22 @@ pub async fn start_background_task(disks: Vec<rustfs_madmin::Disk>) {
 
             info!("Starting scheduled capacity update");
             let start = Instant::now();
+            let manager = manager.clone();
+            let disks = disks.clone();
+            let started = manager
+                .clone()
+                .spawn_refresh_if_needed(DataSource::Scheduled, move || async move {
+                    calculate_data_dir_used_capacity(&disks)
+                        .await
+                        .map(|scan| scan.to_capacity_update())
+                        .map_err(|e| e.to_string())
+                })
+                .await;
 
-            // Import the calculate function
-            match calculate_data_dir_used_capacity(&disks).await {
-                Ok(new_capacity) => {
-                    let elapsed = start.elapsed();
-                    info!("Scheduled update completed: {} bytes in {:?}", new_capacity, elapsed);
-                    manager.update_capacity(new_capacity, DataSource::Scheduled).await;
-                }
-                Err(e) => {
-                    error!("Scheduled update failed: {:?}", e);
-                }
+            if started {
+                debug!("Scheduled capacity refresh started in {:?}", start.elapsed());
+            } else {
+                debug!("Scheduled capacity refresh skipped because another refresh is already in progress");
             }
         }
     });
@@ -586,49 +753,49 @@ mod tests {
     #[serial]
     fn test_get_scheduled_update_interval() {
         let interval = get_scheduled_update_interval();
-        assert_eq!(interval, Duration::from_secs(300));
+        assert_eq!(interval, Duration::from_secs(120));
     }
 
     #[test]
     #[serial]
     fn test_get_write_trigger_delay() {
         let delay = get_write_trigger_delay();
-        assert_eq!(delay, Duration::from_secs(10));
+        assert_eq!(delay, Duration::from_secs(5));
     }
 
     #[test]
     #[serial]
     fn test_get_write_frequency_threshold() {
         let threshold = get_write_frequency_threshold();
-        assert_eq!(threshold, 10);
+        assert_eq!(threshold, 5);
     }
 
     #[test]
     #[serial]
     fn test_get_fast_update_threshold() {
         let threshold = get_fast_update_threshold();
-        assert_eq!(threshold, Duration::from_secs(60));
+        assert_eq!(threshold, Duration::from_secs(30));
     }
 
     #[test]
     #[serial]
     fn test_get_max_files_threshold() {
         let threshold = get_max_files_threshold();
-        assert_eq!(threshold, 1_000_000);
+        assert_eq!(threshold, 200_000);
     }
 
     #[test]
     #[serial]
     fn test_get_stat_timeout() {
         let timeout = get_stat_timeout();
-        assert_eq!(timeout, Duration::from_secs(5));
+        assert_eq!(timeout, Duration::from_secs(3));
     }
 
     #[test]
     #[serial]
     fn test_get_sample_rate() {
         let rate = get_sample_rate();
-        assert_eq!(rate, 100);
+        assert_eq!(rate, 200);
     }
 
     #[test]
@@ -708,7 +875,9 @@ mod tests {
     async fn test_update_capacity() {
         let manager = HybridCapacityManager::from_env();
 
-        manager.update_capacity(1000, DataSource::RealTime).await;
+        manager
+            .update_capacity(CapacityUpdate::exact(1000, 0), DataSource::RealTime)
+            .await;
 
         let cached = manager.get_capacity().await;
         assert!(cached.is_some());
@@ -735,7 +904,9 @@ mod tests {
         assert!(!manager.needs_fast_update().await);
 
         // Update cache
-        manager.update_capacity(1000, DataSource::RealTime).await;
+        manager
+            .update_capacity(CapacityUpdate::exact(1000, 0), DataSource::RealTime)
+            .await;
 
         // Fresh cache, should not need update
         assert!(!manager.needs_fast_update().await);
@@ -747,10 +918,10 @@ mod tests {
         let config = HybridStrategyConfig::from_env();
 
         // Check default values
-        assert_eq!(config.scheduled_update_interval, Duration::from_secs(300));
-        assert_eq!(config.write_trigger_delay, Duration::from_secs(10));
-        assert_eq!(config.write_frequency_threshold, 10);
-        assert_eq!(config.fast_update_threshold, Duration::from_secs(60));
+        assert_eq!(config.scheduled_update_interval, Duration::from_secs(120));
+        assert_eq!(config.write_trigger_delay, Duration::from_secs(5));
+        assert_eq!(config.write_frequency_threshold, 5);
+        assert_eq!(config.fast_update_threshold, Duration::from_secs(30));
         assert!(config.enable_smart_update);
         assert!(config.enable_write_trigger);
     }
diff --git a/rustfs/src/capacity/capacity_manager_test.rs b/rustfs/src/capacity/capacity_manager_test.rs
index 16a8412a8..33158b1f9 100644
--- a/rustfs/src/capacity/capacity_manager_test.rs
+++ b/rustfs/src/capacity/capacity_manager_test.rs
@@ -16,9 +16,10 @@
 
 #[cfg(test)]
 mod tests {
-    use crate::capacity::capacity_manager::{DataSource, HybridCapacityManager, HybridStrategyConfig};
+    use crate::capacity::capacity_manager::{CapacityUpdate, DataSource, HybridCapacityManager, HybridStrategyConfig};
     use serial_test::serial;
     use std::sync::Arc;
+    use std::sync::atomic::{AtomicUsize, Ordering};
     use std::time::Duration;
     use tokio::time::sleep;
 
@@ -33,17 +34,17 @@ mod tests {
     async fn test_capacity_update_and_retrieval() {
         let manager = HybridCapacityManager::from_env();
 
-        // Initially no cache
         assert!(manager.get_capacity().await.is_none());
 
-        // Update capacity
-        manager.update_capacity(1000, DataSource::RealTime).await;
+        manager
+            .update_capacity(CapacityUpdate::exact(1000, 10), DataSource::RealTime)
+            .await;
 
-        // Retrieve cached value
         let cached = manager.get_capacity().await;
         assert!(cached.is_some());
         let cached = cached.unwrap();
         assert_eq!(cached.total_used, 1000);
+        assert_eq!(cached.file_count, 10);
         assert_eq!(cached.source, DataSource::RealTime);
         assert!(!cached.is_estimated);
     }
@@ -52,7 +53,6 @@ mod tests {
     async fn test_write_operation_recording() {
         let manager = HybridCapacityManager::from_env();
 
-        // Record multiple write operations
         manager.record_write_operation().await;
         manager.record_write_operation().await;
         manager.record_write_operation().await;
@@ -65,23 +65,17 @@ mod tests {
     async fn test_fast_update_detection() {
         let manager = HybridCapacityManager::from_env();
 
-        // No cache, should not need fast update
         assert!(!manager.needs_fast_update().await);
 
-        // Update cache
-        manager.update_capacity(1000, DataSource::RealTime).await;
+        manager
+            .update_capacity(CapacityUpdate::exact(1000, 1), DataSource::RealTime)
+            .await;
 
-        // Fresh cache, should not need fast update
         assert!(!manager.needs_fast_update().await);
 
-        // Record write operation
         manager.record_write_operation().await;
-
-        // Wait for cache to become stale
         sleep(Duration::from_millis(100)).await;
 
-        // Now cache is stale and there's recent write
-        // Note: This might not trigger due to timing, so we just check it doesn't panic
         let _needs_update = manager.needs_fast_update().await;
     }
 
@@ -89,22 +83,19 @@ mod tests {
     async fn test_cache_age_tracking() {
         let manager = HybridCapacityManager::from_env();
 
-        // No cache, age should be None
         assert!(manager.get_cache_age().await.is_none());
 
-        // Update cache
-        manager.update_capacity(1000, DataSource::RealTime).await;
+        manager
+            .update_capacity(CapacityUpdate::exact(1000, 1), DataSource::RealTime)
+            .await;
 
-        // Check cache age
         let age = manager.get_cache_age().await;
         assert!(age.is_some());
         let age = age.unwrap();
         assert!(age < Duration::from_secs(1));
 
-        // Wait a bit
         sleep(Duration::from_millis(100)).await;
 
-        // Check age again
         let age = manager.get_cache_age().await.unwrap();
         assert!(age >= Duration::from_millis(100));
     }
@@ -113,7 +104,6 @@ mod tests {
     async fn test_data_source_tracking() {
         let manager = HybridCapacityManager::from_env();
 
-        // Test different data sources
         let sources = vec![
             DataSource::RealTime,
             DataSource::Scheduled,
@@ -122,7 +112,7 @@ mod tests {
         ];
 
         for source in sources {
-            manager.update_capacity(1000, source).await;
+            manager.update_capacity(CapacityUpdate::exact(1000, 1), source).await;
             let cached = manager.get_capacity().await.unwrap();
             assert_eq!(cached.source, source);
         }
@@ -132,11 +122,10 @@ mod tests {
     async fn test_config_from_env() {
         let config = HybridStrategyConfig::from_env();
 
-        // Check default values
-        assert_eq!(config.scheduled_update_interval, Duration::from_secs(300));
-        assert_eq!(config.write_trigger_delay, Duration::from_secs(10));
-        assert_eq!(config.write_frequency_threshold, 10);
-        assert_eq!(config.fast_update_threshold, Duration::from_secs(60));
+        assert_eq!(config.scheduled_update_interval, Duration::from_secs(120));
+        assert_eq!(config.write_trigger_delay, Duration::from_secs(5));
+        assert_eq!(config.write_frequency_threshold, 5);
+        assert_eq!(config.fast_update_threshold, Duration::from_secs(30));
         assert!(config.enable_smart_update);
         assert!(config.enable_write_trigger);
     }
@@ -145,42 +134,34 @@ mod tests {
     async fn test_write_frequency_window() {
         let manager = HybridCapacityManager::from_env();
 
-        // Record many write operations
         for _ in 0..20 {
             manager.record_write_operation().await;
         }
 
-        // Check frequency (should be 20 since all are within 1 minute)
         let frequency = manager.get_write_frequency().await;
         assert_eq!(frequency, 20);
-
-        // Note: In a real test, we would wait for the window to expire
-        // and verify that old writes are removed
     }
 
     #[tokio::test]
     #[serial]
     async fn test_concurrent_access() {
         let manager = Arc::new(HybridCapacityManager::from_env());
-
-        // Simulate concurrent updates
         let mut handles = vec![];
 
         for i in 0..10 {
             let mgr = manager.clone();
             let handle = tokio::spawn(async move {
-                mgr.update_capacity(i as u64 * 100, DataSource::RealTime).await;
+                mgr.update_capacity(CapacityUpdate::exact(i as u64 * 100, i), DataSource::RealTime)
+                    .await;
                 mgr.record_write_operation().await;
             });
             handles.push(handle);
         }
 
-        // Wait for all tasks to complete
         for handle in handles {
             handle.await.unwrap();
         }
 
-        // Verify final state
         let cached = manager.get_capacity().await;
         assert!(cached.is_some());
 
@@ -192,21 +173,95 @@ mod tests {
     #[serial]
     async fn test_performance_overhead() {
         let manager = Arc::new(HybridCapacityManager::from_env());
-
-        // Measure time for 1000 operations
         let start = std::time::Instant::now();
 
         for i in 0..1000 {
-            manager.update_capacity(i as u64, DataSource::RealTime).await;
+            manager
+                .update_capacity(CapacityUpdate::exact(i as u64, i), DataSource::RealTime)
+                .await;
             manager.record_write_operation().await;
             let _ = manager.get_capacity().await;
         }
 
         let elapsed = start.elapsed();
-
-        // Should complete in less than 1 second
         assert!(elapsed < Duration::from_secs(1));
 
         println!("1000 operations completed in {:?}", elapsed);
     }
+
+    #[tokio::test]
+    async fn test_refresh_or_join_singleflight() {
+        let manager = Arc::new(HybridCapacityManager::from_env());
+        let calls = Arc::new(AtomicUsize::new(0));
+
+        let mgr1 = manager.clone();
+        let calls1 = calls.clone();
+        let first = tokio::spawn(async move {
+            mgr1.refresh_or_join(DataSource::Scheduled, move || async move {
+                calls1.fetch_add(1, Ordering::SeqCst);
+                sleep(Duration::from_millis(50)).await;
+                Ok(CapacityUpdate::exact(2048, 8))
+            })
+            .await
+        });
+
+        sleep(Duration::from_millis(10)).await;
+
+        let mgr2 = manager.clone();
+        let calls2 = calls.clone();
+        let second = tokio::spawn(async move {
+            mgr2.refresh_or_join(DataSource::WriteTriggered, move || async move {
+                calls2.fetch_add(1, Ordering::SeqCst);
+                Ok(CapacityUpdate::exact(4096, 16))
+            })
+            .await
+        });
+
+        let first = first.await.unwrap().unwrap();
+        let second = second.await.unwrap().unwrap();
+
+        assert_eq!(calls.load(Ordering::SeqCst), 1);
+        assert_eq!(first.total_used, 2048);
+        assert_eq!(second.total_used, 2048);
+        let cached = manager.get_capacity().await.unwrap();
+        assert_eq!(cached.total_used, 2048);
+        assert_eq!(cached.file_count, 8);
+    }
+
+    #[tokio::test]
+    async fn test_spawn_refresh_if_needed_deduplicates_background_refresh() {
+        let manager = Arc::new(HybridCapacityManager::from_env());
+        let calls = Arc::new(AtomicUsize::new(0));
+
+        let first_manager = manager.clone();
+        let first_calls = calls.clone();
+        let started = first_manager
+            .clone()
+            .spawn_refresh_if_needed(DataSource::Scheduled, move || async move {
+                first_calls.fetch_add(1, Ordering::SeqCst);
+                sleep(Duration::from_millis(50)).await;
+                Ok(CapacityUpdate::estimated(8192, 32))
+            })
+            .await;
+        assert!(started);
+
+        let second_manager = manager.clone();
+        let second_calls = calls.clone();
+        let started = second_manager
+            .clone()
+            .spawn_refresh_if_needed(DataSource::Scheduled, move || async move {
+                second_calls.fetch_add(1, Ordering::SeqCst);
+                Ok(CapacityUpdate::exact(1, 1))
+            })
+            .await;
+        assert!(!started);
+
+        sleep(Duration::from_millis(100)).await;
+
+        assert_eq!(calls.load(Ordering::SeqCst), 1);
+        assert!(!manager.refresh_in_progress().await);
+        let cached = manager.get_capacity().await.unwrap();
+        assert_eq!(cached.total_used, 8192);
+        assert!(cached.is_estimated);
+    }
 }
diff --git a/rustfs/src/capacity/capacity_metrics.rs b/rustfs/src/capacity/capacity_metrics.rs
deleted file mode 100644
index 0a6deda81..000000000
--- a/rustfs/src/capacity/capacity_metrics.rs
+++ /dev/null
@@ -1,465 +0,0 @@
-// 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.
-
-//! Capacity Metrics for monitoring
-
-use metrics::{counter, gauge, histogram};
-use std::sync::Arc;
-use std::sync::atomic::{AtomicU64, Ordering};
-use std::time::Duration;
-use tracing::info;
-
-// ============================================================================
-// Metric Name Constants (following existing naming convention)
-// ============================================================================
-
-/// Cache hit counter
-const CAPACITY_CACHE_HIT: &str = "rustfs.capacity.cache.hits";
-
-/// Cache miss counter
-const CAPACITY_CACHE_MISS: &str = "rustfs.capacity.cache.misses";
-
-/// Cache hit rate gauge
-const CAPACITY_CACHE_HIT_RATE: &str = "rustfs.capacity.cache.hit_rate";
-
-/// Cache hits total gauge
-const CAPACITY_CACHE_HITS_TOTAL: &str = "rustfs.capacity.cache.hits_total";
-
-/// Cache misses total gauge
-const CAPACITY_CACHE_MISSES_TOTAL: &str = "rustfs.capacity.cache.misses_total";
-
-/// Scheduled update counter
-const CAPACITY_UPDATE_SCHEDULED: &str = "rustfs.capacity.update.scheduled";
-
-/// Write-triggered update counter
-const CAPACITY_UPDATE_WRITE_TRIGGERED: &str = "rustfs.capacity.update.write_triggered";
-
-/// Update failure counter
-const CAPACITY_UPDATE_FAILURES: &str = "rustfs.capacity.update.failures";
-
-/// Current capacity in bytes gauge
-#[allow(dead_code)]
-const CAPACITY_CURRENT_BYTES: &str = "rustfs.capacity.current";
-
-/// Write operations counter
-const CAPACITY_WRITE_OPERATIONS: &str = "rustfs.capacity.write.operations";
-
-/// Write frequency gauge
-#[allow(dead_code)]
-const CAPACITY_WRITE_FREQUENCY: &str = "rustfs.capacity.write.frequency";
-
-/// Update duration in microseconds histogram
-const CAPACITY_UPDATE_DURATION_US: &str = "rustfs.capacity.update.duration_us";
-
-/// Scheduled updates total gauge
-const CAPACITY_UPDATE_SCHEDULED_TOTAL: &str = "rustfs.capacity.update.scheduled_total";
-
-/// Write-triggered updates total gauge
-const CAPACITY_UPDATE_WRITE_TRIGGERED_TOTAL: &str = "rustfs.capacity.update.write_triggered_total";
-
-/// Update failures total gauge
-const CAPACITY_UPDATE_FAILURES_TOTAL: &str = "rustfs.capacity.update.failures_total";
-
-/// Symlinks encountered counter
-const CAPACITY_SYMLINKS_ENCOUNTERED: &str = "rustfs.capacity.symlinks.encountered";
-
-/// Symlinks total size gauge
-const CAPACITY_SYMLINKS_SIZE: &str = "rustfs.capacity.symlinks.total_size";
-
-/// Symlinks count gauge
-const CAPACITY_SYMLINKS_COUNT: &str = "rustfs.capacity.symlinks.count";
-
-/// Dynamic timeout counter
-const CAPACITY_TIMEOUT_DYNAMIC: &str = "rustfs.capacity.timeout.dynamic";
-
-/// Timeout fallback counter
-const CAPACITY_TIMEOUT_FALLBACK: &str = "rustfs.capacity.timeout.fallback";
-
-/// Stall detected counter
-const CAPACITY_TIMEOUT_STALL: &str = "rustfs.capacity.timeout.stall";
-
-/// Dynamic timeout total gauge
-const CAPACITY_TIMEOUT_DYNAMIC_TOTAL: &str = "rustfs.capacity.timeout.dynamic_total";
-
-/// Timeout fallback total gauge
-const CAPACITY_TIMEOUT_FALLBACK_TOTAL: &str = "rustfs.capacity.timeout.fallback_total";
-
-/// Stall detected total gauge
-const CAPACITY_TIMEOUT_STALL_TOTAL: &str = "rustfs.capacity.timeout.stall_total";
-
-// ============================================================================
-// Capacity Metrics
-// ============================================================================
-
-/// Capacity metrics for monitoring
-#[derive(Debug, Default)]
-pub struct CapacityMetrics {
-    /// Cache hit count
-    pub cache_hits: AtomicU64,
-    /// Cache miss count
-    pub cache_misses: AtomicU64,
-    /// Scheduled update count
-    pub scheduled_updates: AtomicU64,
-    /// Write triggered update count
-    pub write_triggered_updates: AtomicU64,
-    /// Update failure count
-    pub update_failures: AtomicU64,
-    /// Total update duration in microseconds
-    pub total_update_duration_us: AtomicU64,
-    /// Update count for average calculation
-    pub update_count: AtomicU64,
-    /// Symlink count encountered during capacity calculation
-    pub symlink_count: AtomicU64,
-    /// Total size of symlink targets
-    pub symlink_size: AtomicU64,
-    /// Dynamic timeout usage count
-    pub dynamic_timeout_count: AtomicU64,
-    /// Timeout fallback to sampling count
-    pub timeout_fallback_count: AtomicU64,
-    /// Stall detection count
-    pub stall_detected_count: AtomicU64,
-}
-
-impl CapacityMetrics {
-    /// Create new metrics
-    pub fn new() -> Self {
-        Self::default()
-    }
-
-    /// Record cache hit
-    pub fn record_cache_hit(&self) {
-        self.cache_hits.fetch_add(1, Ordering::Relaxed);
-        counter!(CAPACITY_CACHE_HIT).increment(1);
-    }
-
-    /// Record cache miss
-    pub fn record_cache_miss(&self) {
-        self.cache_misses.fetch_add(1, Ordering::Relaxed);
-        counter!(CAPACITY_CACHE_MISS).increment(1);
-    }
-
-    /// Record scheduled update
-    #[allow(dead_code)]
-    pub fn record_scheduled_update(&self) {
-        self.scheduled_updates.fetch_add(1, Ordering::Relaxed);
-        counter!(CAPACITY_UPDATE_SCHEDULED).increment(1);
-    }
-
-    /// Record write triggered update
-    #[allow(dead_code)]
-    pub fn record_write_triggered_update(&self) {
-        self.write_triggered_updates.fetch_add(1, Ordering::Relaxed);
-        counter!(CAPACITY_UPDATE_WRITE_TRIGGERED).increment(1);
-    }
-
-    /// Record update failure
-    #[allow(dead_code)]
-    pub fn record_update_failure(&self) {
-        self.update_failures.fetch_add(1, Ordering::Relaxed);
-        counter!(CAPACITY_UPDATE_FAILURES).increment(1);
-    }
-
-    /// Record write operation
-    #[allow(dead_code)]
-    pub fn record_write_operation(&self) {
-        counter!(CAPACITY_WRITE_OPERATIONS).increment(1);
-    }
-
-    /// Record symlink encountered
-    pub fn record_symlink(&self, size: u64) {
-        self.symlink_count.fetch_add(1, Ordering::Relaxed);
-        let total_size = self.symlink_size.fetch_add(size, Ordering::Relaxed) + size;
-        counter!(CAPACITY_SYMLINKS_ENCOUNTERED).increment(1);
-        gauge!(CAPACITY_SYMLINKS_SIZE).set(total_size as f64);
-    }
-
-    /// Record dynamic timeout usage
-    pub fn record_dynamic_timeout(&self) {
-        self.dynamic_timeout_count.fetch_add(1, Ordering::Relaxed);
-        counter!(CAPACITY_TIMEOUT_DYNAMIC).increment(1);
-    }
-
-    /// Record timeout fallback to sampling
-    pub fn record_timeout_fallback(&self) {
-        self.timeout_fallback_count.fetch_add(1, Ordering::Relaxed);
-        counter!(CAPACITY_TIMEOUT_FALLBACK).increment(1);
-    }
-
-    /// Record stall detection
-    pub fn record_stall_detected(&self) {
-        self.stall_detected_count.fetch_add(1, Ordering::Relaxed);
-        counter!(CAPACITY_TIMEOUT_STALL).increment(1);
-    }
-
-    /// Get symlink statistics
-    #[allow(dead_code)]
-    pub fn get_symlink_stats(&self) -> (u64, u64) {
-        (self.symlink_count.load(Ordering::Relaxed), self.symlink_size.load(Ordering::Relaxed))
-    }
-
-    /// Get timeout statistics
-    #[allow(dead_code)]
-    pub fn get_timeout_stats(&self) -> (u64, u64, u64) {
-        (
-            self.dynamic_timeout_count.load(Ordering::Relaxed),
-            self.timeout_fallback_count.load(Ordering::Relaxed),
-            self.stall_detected_count.load(Ordering::Relaxed),
-        )
-    }
-
-    /// Record update duration
-    #[allow(dead_code)]
-    pub fn record_update_duration(&self, duration: Duration) {
-        let duration_us = duration.as_micros() as u64;
-        self.total_update_duration_us.fetch_add(duration_us, Ordering::Relaxed);
-        self.update_count.fetch_add(1, Ordering::Relaxed);
-
-        histogram!(CAPACITY_UPDATE_DURATION_US).record(duration_us as f64);
-    }
-
-    /// Get cache hit rate
-    pub fn get_cache_hit_rate(&self) -> f64 {
-        let hits = self.cache_hits.load(Ordering::Relaxed);
-        let misses = self.cache_misses.load(Ordering::Relaxed);
-        let total = hits + misses;
-        if total == 0 { 0.0 } else { hits as f64 / total as f64 }
-    }
-
-    /// Get average update duration
-    pub fn get_avg_update_duration(&self) -> Duration {
-        let total_us = self.total_update_duration_us.load(Ordering::Relaxed);
-        let count = self.update_count.load(Ordering::Relaxed);
-        if count == 0 {
-            Duration::from_secs(0)
-        } else {
-            Duration::from_micros(total_us / count)
-        }
-    }
-
-    /// Get metrics summary
-    pub fn get_summary(&self) -> MetricsSummary {
-        MetricsSummary {
-            cache_hits: self.cache_hits.load(Ordering::Relaxed),
-            cache_misses: self.cache_misses.load(Ordering::Relaxed),
-            cache_hit_rate: self.get_cache_hit_rate(),
-            scheduled_updates: self.scheduled_updates.load(Ordering::Relaxed),
-            write_triggered_updates: self.write_triggered_updates.load(Ordering::Relaxed),
-            update_failures: self.update_failures.load(Ordering::Relaxed),
-            avg_update_duration: self.get_avg_update_duration(),
-            symlink_count: self.symlink_count.load(Ordering::Relaxed),
-            symlink_size: self.symlink_size.load(Ordering::Relaxed),
-            dynamic_timeout_count: self.dynamic_timeout_count.load(Ordering::Relaxed),
-            timeout_fallback_count: self.timeout_fallback_count.load(Ordering::Relaxed),
-            stall_detected_count: self.stall_detected_count.load(Ordering::Relaxed),
-        }
-    }
-
-    /// Log metrics summary
-    pub fn log_summary(&self) {
-        let summary = self.get_summary();
-
-        // Update gauges for current values using constant names
-        gauge!(CAPACITY_CACHE_HIT_RATE).set(summary.cache_hit_rate);
-        gauge!(CAPACITY_CACHE_HITS_TOTAL).set(summary.cache_hits as f64);
-        gauge!(CAPACITY_CACHE_MISSES_TOTAL).set(summary.cache_misses as f64);
-        gauge!(CAPACITY_UPDATE_SCHEDULED_TOTAL).set(summary.scheduled_updates as f64);
-        gauge!(CAPACITY_UPDATE_WRITE_TRIGGERED_TOTAL).set(summary.write_triggered_updates as f64);
-        gauge!(CAPACITY_UPDATE_FAILURES_TOTAL).set(summary.update_failures as f64);
-        gauge!(CAPACITY_SYMLINKS_COUNT).set(summary.symlink_count as f64);
-        gauge!(CAPACITY_SYMLINKS_SIZE).set(summary.symlink_size as f64);
-        gauge!(CAPACITY_TIMEOUT_DYNAMIC_TOTAL).set(summary.dynamic_timeout_count as f64);
-        gauge!(CAPACITY_TIMEOUT_FALLBACK_TOTAL).set(summary.timeout_fallback_count as f64);
-        gauge!(CAPACITY_TIMEOUT_STALL_TOTAL).set(summary.stall_detected_count as f64);
-
-        info!(
-            "Capacity Metrics: cache_hit_rate={:.2}%, cache_hits={}, cache_misses={}, scheduled_updates={}, write_triggered_updates={}, update_failures={}, avg_update_duration={:?}, symlinks={}, symlink_size={}, dynamic_timeouts={}, timeout_fallbacks={}, stalls={}",
-            summary.cache_hit_rate * 100.0,
-            summary.cache_hits,
-            summary.cache_misses,
-            summary.scheduled_updates,
-            summary.write_triggered_updates,
-            summary.update_failures,
-            summary.avg_update_duration,
-            summary.symlink_count,
-            summary.symlink_size,
-            summary.dynamic_timeout_count,
-            summary.timeout_fallback_count,
-            summary.stall_detected_count
-        );
-    }
-}
-
-/// Metrics summary
-#[derive(Debug, Clone)]
-pub struct MetricsSummary {
-    pub cache_hits: u64,
-    pub cache_misses: u64,
-    pub cache_hit_rate: f64,
-    pub scheduled_updates: u64,
-    pub write_triggered_updates: u64,
-    pub update_failures: u64,
-    pub avg_update_duration: Duration,
-    pub symlink_count: u64,
-    pub symlink_size: u64,
-    pub dynamic_timeout_count: u64,
-    pub timeout_fallback_count: u64,
-    pub stall_detected_count: u64,
-}
-
-/// Global metrics instance
-static CAPACITY_METRICS: std::sync::OnceLock<Arc<CapacityMetrics>> = std::sync::OnceLock::new();
-
-/// Get global metrics
-pub fn get_capacity_metrics() -> Arc<CapacityMetrics> {
-    CAPACITY_METRICS.get_or_init(|| Arc::new(CapacityMetrics::new())).clone()
-}
-
-/// Start metrics logging task
-pub async fn start_metrics_logging(interval: Duration) {
-    let metrics = get_capacity_metrics();
-
-    tokio::spawn(async move {
-        let mut timer = tokio::time::interval(interval);
-
-        loop {
-            timer.tick().await;
-            metrics.log_summary();
-        }
-    });
-}
-
-/// Record a write operation globally
-#[allow(dead_code)]
-pub fn record_global_write_operation() {
-    let metrics = get_capacity_metrics();
-    metrics.record_write_operation();
-}
-
-/// Record cache hit globally
-#[allow(dead_code)]
-pub fn record_global_cache_hit() {
-    let metrics = get_capacity_metrics();
-    metrics.record_cache_hit();
-}
-
-/// Record cache miss globally
-#[allow(dead_code)]
-pub fn record_global_cache_miss() {
-    let metrics = get_capacity_metrics();
-    metrics.record_cache_miss();
-}
-
-// ============================================================================
-// Tests
-// ============================================================================
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_metrics_creation() {
-        let metrics = CapacityMetrics::new();
-        assert_eq!(metrics.cache_hits.load(Ordering::Relaxed), 0);
-        assert_eq!(metrics.cache_misses.load(Ordering::Relaxed), 0);
-    }
-
-    #[test]
-    fn test_record_cache_hit() {
-        let metrics = CapacityMetrics::new();
-        metrics.record_cache_hit();
-        metrics.record_cache_hit();
-        assert_eq!(metrics.cache_hits.load(Ordering::Relaxed), 2);
-    }
-
-    #[test]
-    fn test_cache_hit_rate() {
-        let metrics = CapacityMetrics::new();
-        metrics.record_cache_hit();
-        metrics.record_cache_hit();
-        metrics.record_cache_miss();
-
-        let rate = metrics.get_cache_hit_rate();
-        assert!((rate - 0.6666666666666666).abs() < 0.0001);
-    }
-
-    #[test]
-    fn test_avg_update_duration() {
-        let metrics = CapacityMetrics::new();
-        metrics.record_update_duration(Duration::from_millis(100));
-        metrics.record_update_duration(Duration::from_millis(200));
-
-        let avg = metrics.get_avg_update_duration();
-        assert_eq!(avg, Duration::from_millis(150));
-    }
-
-    #[test]
-    fn test_get_summary() {
-        let metrics = CapacityMetrics::new();
-        metrics.record_cache_hit();
-        metrics.record_scheduled_update();
-        metrics.record_update_duration(Duration::from_millis(100));
-
-        let summary = metrics.get_summary();
-        assert_eq!(summary.cache_hits, 1);
-        assert_eq!(summary.scheduled_updates, 1);
-        assert_eq!(summary.avg_update_duration, Duration::from_millis(100));
-        assert_eq!(summary.symlink_count, 0);
-        assert_eq!(summary.dynamic_timeout_count, 0);
-    }
-
-    #[test]
-    fn test_record_write_operation() {
-        let metrics = CapacityMetrics::new();
-        metrics.record_write_operation();
-        metrics.record_write_operation();
-        // This test just ensures the method doesn't panic
-        assert_eq!(metrics.write_triggered_updates.load(Ordering::Relaxed), 0);
-    }
-
-    #[test]
-    fn test_record_symlink() {
-        let metrics = CapacityMetrics::new();
-        metrics.record_symlink(1024);
-        metrics.record_symlink(2048);
-
-        let (count, size) = metrics.get_symlink_stats();
-        assert_eq!(count, 2);
-        assert_eq!(size, 3072);
-    }
-
-    #[test]
-    fn test_record_dynamic_timeout() {
-        let metrics = CapacityMetrics::new();
-        metrics.record_dynamic_timeout();
-        metrics.record_dynamic_timeout();
-
-        let (dynamic, fallback, stalls) = metrics.get_timeout_stats();
-        assert_eq!(dynamic, 2);
-        assert_eq!(fallback, 0);
-        assert_eq!(stalls, 0);
-    }
-
-    #[test]
-    fn test_record_timeout_fallback() {
-        let metrics = CapacityMetrics::new();
-        metrics.record_timeout_fallback();
-        metrics.record_stall_detected();
-
-        let (dynamic, fallback, stalls) = metrics.get_timeout_stats();
-        assert_eq!(dynamic, 0);
-        assert_eq!(fallback, 1);
-        assert_eq!(stalls, 1);
-    }
-}
diff --git a/rustfs/src/capacity/mod.rs b/rustfs/src/capacity/mod.rs
index 536621da3..10e0c7749 100644
--- a/rustfs/src/capacity/mod.rs
+++ b/rustfs/src/capacity/mod.rs
@@ -18,24 +18,24 @@
 //! - Scheduled background updates (configurable interval)
 //! - Write-triggered updates for high-frequency write scenarios
 //! - Configurable caching thresholds and smart update strategies
-//! - Comprehensive metrics collection for monitoring
+//! - Capacity metrics emitted through `rustfs-io-metrics`
 //!
 //! ## Configuration
 //!
 //! All configuration is via environment variables (see `rustfs_config`):
-//! - `RUSTFS_CAPACITY_SCHEDULED_INTERVAL` - Update interval in seconds (default: 300)
-//! - `RUSTFS_CAPACITY_WRITE_TRIGGER_DELAY` - Write trigger delay (default: 10s)
-//! - `RUSTFS_CAPACITY_WRITE_FREQUENCY_THRESHOLD` - Write frequency threshold (default: 10 writes/min)
-//! - `RUSTFS_CAPACITY_FAST_UPDATE_THRESHOLD` - Fast update threshold (default: 60s)
-//! - `RUSTFS_CAPACITY_MAX_FILES_THRESHOLD` - Max files before sampling (default: 1,000,000)
-//! - `RUSTFS_CAPACITY_STAT_TIMEOUT` - Stat operation timeout (default: 5s)
-//! - `RUSTFS_CAPACITY_SAMPLE_RATE` - Sampling rate for metrics (default: 100)
+//! - `RUSTFS_CAPACITY_SCHEDULED_INTERVAL` - Update interval in seconds (default: 120)
+//! - `RUSTFS_CAPACITY_WRITE_TRIGGER_DELAY` - Write trigger delay (default: 5s)
+//! - `RUSTFS_CAPACITY_WRITE_FREQUENCY_THRESHOLD` - Write frequency threshold (default: 5 writes/min)
+//! - `RUSTFS_CAPACITY_FAST_UPDATE_THRESHOLD` - Fast update threshold (default: 30s)
+//! - `RUSTFS_CAPACITY_MAX_FILES_THRESHOLD` - Max files before sampling (default: 200,000)
+//! - `RUSTFS_CAPACITY_STAT_TIMEOUT` - Stat operation timeout (default: 3s)
+//! - `RUSTFS_CAPACITY_SAMPLE_RATE` - Sampling rate for metrics (default: 200)
 //! - `RUSTFS_CAPACITY_FOLLOW_SYMLINKS` - Follow symlinks during traversal (default: false)
-//! - `RUSTFS_CAPACITY_MAX_SYMLINK_DEPTH` - Max symlink depth (default: 8)
-//! - `RUSTFS_CAPACITY_ENABLE_DYNAMIC_TIMEOUT` - Enable dynamic timeout (default: false)
-//! - `RUSTFS_CAPACITY_MIN_TIMEOUT` - Minimum timeout (default: 1s)
-//! - `RUSTFS_CAPACITY_MAX_TIMEOUT` - Maximum timeout (default: 300s)
-//! - `RUSTFS_CAPACITY_STALL_TIMEOUT` - Stall detection timeout (default: 30s)
+//! - `RUSTFS_CAPACITY_MAX_SYMLINK_DEPTH` - Max symlink depth (default: 3)
+//! - `RUSTFS_CAPACITY_ENABLE_DYNAMIC_TIMEOUT` - Enable dynamic timeout (default: true)
+//! - `RUSTFS_CAPACITY_MIN_TIMEOUT` - Minimum timeout (default: 2s)
+//! - `RUSTFS_CAPACITY_MAX_TIMEOUT` - Maximum timeout (default: 15s)
+//! - `RUSTFS_CAPACITY_STALL_TIMEOUT` - Stall detection timeout (default: 20s)
 //!
 //! ## Architecture
 //!
@@ -45,16 +45,8 @@
 //! 3. **Cached responses**: Returns cached data when fresh
 //! 4. **Timeout protection**: Dynamic timeouts prevent hangs on large directories
 //!
-//! ## Metrics
-//!
-//! Metrics are automatically recorded via the `metrics` crate and accessible
-//! through the `rustfs-metrics` collection system. Key metrics include:
-//! - `rustfs.capacity.cache.{hits,misses}` - Cache hit/miss tracking
-//! - `rustfs.capacity.current` - Current capacity in bytes
-//! - `rustfs.capacity.write.operations` - Write operation count
-//! - `rustfs.capacity.update.{scheduled,write_triggered,failures}` - Update statistics
-//! - `rustfs.capacity.symlinks.*` - Symlink tracking statistics
-//! - `rustfs.capacity.timeout.*` - Timeout and stall detection
+//! Capacity metrics flow through the existing observability pipeline via the `metrics`
+//! crate and `rustfs-io-metrics`; this module does not expose a Prometheus HTTP endpoint.
 //!
 //! ## Testing
 //!
@@ -73,6 +65,5 @@ pub mod capacity_integration;
 pub mod capacity_manager;
 #[cfg(test)]
 mod capacity_manager_test;
-pub mod capacity_metrics;
 #[cfg(test)]
 mod write_trigger_test;
diff --git a/rustfs/src/capacity/write_trigger_test.rs b/rustfs/src/capacity/write_trigger_test.rs
index a7d07e14f..1a606e005 100644
--- a/rustfs/src/capacity/write_trigger_test.rs
+++ b/rustfs/src/capacity/write_trigger_test.rs
@@ -16,10 +16,7 @@
 
 #[cfg(test)]
 mod tests {
-    use crate::capacity::capacity_manager::{DataSource, HybridCapacityManager};
-    use crate::capacity::capacity_metrics::{
-        CapacityMetrics, get_capacity_metrics, record_global_cache_hit, record_global_cache_miss, record_global_write_operation,
-    };
+    use crate::capacity::capacity_manager::{CapacityUpdate, DataSource, HybridCapacityManager};
     use serial_test::serial;
     use std::time::Duration;
 
@@ -27,93 +24,45 @@ mod tests {
     #[serial]
     async fn test_write_trigger_integration() {
         let manager = HybridCapacityManager::from_env();
-        let metrics = CapacityMetrics::new();
 
-        // Record write operations
         manager.record_write_operation().await;
         manager.record_write_operation().await;
         manager.record_write_operation().await;
 
-        // Check write frequency
         let frequency = manager.get_write_frequency().await;
         assert_eq!(frequency, 3);
-
-        // Check metrics
-        let summary = metrics.get_summary();
-        assert_eq!(summary.write_triggered_updates, 0); // Not triggered yet
     }
 
     #[tokio::test]
     #[serial]
     async fn test_write_trigger_with_capacity_update() {
         let manager = HybridCapacityManager::from_env();
-        let metrics = CapacityMetrics::new();
 
-        // Simulate write-triggered update by calling metrics directly
-        metrics.record_write_triggered_update();
+        manager
+            .update_capacity(CapacityUpdate::exact(1000, 4), DataSource::WriteTriggered)
+            .await;
 
-        // Check metrics
-        let summary = metrics.get_summary();
-        assert_eq!(summary.write_triggered_updates, 1);
-
-        // Also test manager update
-        manager.update_capacity(1000, DataSource::WriteTriggered).await;
-
-        // Check capacity
         let cached = manager.get_capacity().await;
         assert!(cached.is_some());
-        assert_eq!(cached.unwrap().total_used, 1000);
-    }
-
-    #[tokio::test]
-    #[serial]
-    async fn test_metrics_recording() {
-        let metrics = CapacityMetrics::new();
-
-        // Record various operations
-        metrics.record_cache_hit();
-        metrics.record_cache_hit();
-        metrics.record_cache_miss();
-
-        metrics.record_scheduled_update();
-        metrics.record_write_triggered_update();
-
-        metrics.record_update_duration(Duration::from_millis(100));
-        metrics.record_update_duration(Duration::from_millis(200));
-
-        // Check summary
-        let summary = metrics.get_summary();
-        assert_eq!(summary.cache_hits, 2);
-        assert_eq!(summary.cache_misses, 1);
-        assert_eq!(summary.scheduled_updates, 1);
-        assert_eq!(summary.write_triggered_updates, 1);
-        assert_eq!(summary.avg_update_duration, Duration::from_millis(150));
-
-        // Check hit rate
-        let hit_rate = metrics.get_cache_hit_rate();
-        assert!((hit_rate - 0.6666666666666666).abs() < 0.0001);
+        let cached = cached.unwrap();
+        assert_eq!(cached.total_used, 1000);
+        assert_eq!(cached.file_count, 4);
+        assert_eq!(cached.source, DataSource::WriteTriggered);
     }
 
     #[tokio::test]
     async fn test_write_frequency_tracking() {
         let manager = HybridCapacityManager::from_env();
 
-        // Initial state
         assert_eq!(manager.get_write_frequency().await, 0);
 
-        // Record writes
         for _ in 0..5 {
             manager.record_write_operation().await;
         }
 
-        // Check frequency
         assert_eq!(manager.get_write_frequency().await, 5);
 
-        // Wait for window to expire (60 seconds)
-        // In real tests, we'd use a shorter window
         tokio::time::sleep(Duration::from_millis(10)).await;
-
-        // Frequency should still be 5 (window not expired)
         assert_eq!(manager.get_write_frequency().await, 5);
     }
 
@@ -121,37 +70,18 @@ mod tests {
     async fn test_needs_fast_update() {
         let manager = HybridCapacityManager::from_env();
 
-        // No cache, should not need update
         assert!(!manager.needs_fast_update().await);
 
-        // Update cache
-        manager.update_capacity(1000, DataSource::Scheduled).await;
+        manager
+            .update_capacity(CapacityUpdate::exact(1000, 1), DataSource::Scheduled)
+            .await;
 
-        // Fresh cache, should not need update
         assert!(!manager.needs_fast_update().await);
 
-        // Record write operation
         manager.record_write_operation().await;
 
-        // With recent write, should need fast update
-        // (depending on configuration, this may or may not trigger)
         let needs_update = manager.needs_fast_update().await;
-        // Just ensure it doesn't panic
         #[allow(clippy::overly_complex_bool_expr)]
         let _ = needs_update || !needs_update;
     }
-
-    #[test]
-    #[serial]
-    fn test_global_metrics_functions() {
-        // Test global functions don't panic
-        let before = get_capacity_metrics().cache_hits.load(std::sync::atomic::Ordering::Relaxed);
-
-        record_global_write_operation();
-        record_global_cache_hit();
-        record_global_cache_miss();
-
-        let metrics = get_capacity_metrics();
-        assert!(metrics.cache_hits.load(std::sync::atomic::Ordering::Relaxed) > before);
-    }
 }