Scanner (#156)

* feat: integrate CancellationToken for unified background services management

- Consolidate data scanner and auto heal cancellation tokens into single unified token
- Move GLOBAL_BACKGROUND_SERVICES_CANCEL_TOKEN to global.rs for centralized management
- Add graceful shutdown support to MRF heal routine with MinIO-compatible logic
- Implement heal_routine_with_cancel method preserving original healing logic
- Update main.rs to use unified background services shutdown mechanism
- Enhance error handling with proper ecstore Result types
- Fix clippy warnings for needless return statements
- Maintain backward compatibility while adding modern cancellation support

This change provides a cleaner architecture for background service lifecycle management
and ensures all healing services can be gracefully shut down through a single token.

Signed-off-by: junxiang Mu <1948535941@qq.com>

* fix: Refact heal and scanner design

Signed-off-by: junxiang Mu <1948535941@qq.com>

* refact: step 2

Signed-off-by: junxiang Mu <1948535941@qq.com>

* feat: refactor scanner module and add data usage statistics

- Move scanner code to scanner/ subdirectory for better organization
- Add data usage statistics collection and persistence
- Implement histogram support for size and version distribution
- Add global cancel token management for scanner operations
- Integrate scanner with ECStore for comprehensive data analysis
- Update error handling and improve test isolation
- Add data usage API endpoints and backend integration

Signed-off-by: junxiang Mu <1948535941@qq.com>

* Chore: fix ref and fix comment

Signed-off-by: junxiang Mu <1948535941@qq.com>

* fix: fix clippy

Signed-off-by: junxiang Mu <1948535941@qq.com>

---------

Signed-off-by: junxiang Mu <1948535941@qq.com>
Co-authored-by: dandan <dandan@dandandeMac-Studio.local>

Cargo.lock

@@ -7857,6 +7857,7 @@ dependencies = [
  "pin-project-lite",
  "reqwest",
  "rust-embed",
+ "rustfs-ahm",
  "rustfs-appauth",
  "rustfs-common",
  "rustfs-config",
@@ -7897,6 +7898,34 @@ dependencies = [
  "zip",
 ]
 
+[[package]]
+name = "rustfs-ahm"
+version = "0.0.3"
+dependencies = [
+ "anyhow",
+ "async-trait",
+ "bytes",
+ "futures",
+ "lazy_static",
+ "rmp-serde",
+ "rustfs-common",
+ "rustfs-ecstore",
+ "rustfs-filemeta",
+ "rustfs-lock",
+ "rustfs-madmin",
+ "rustfs-utils",
+ "serde",
+ "serde_json",
+ "thiserror 2.0.12",
+ "time",
+ "tokio",
+ "tokio-test",
+ "tokio-util",
+ "tracing",
+ "url",
+ "uuid",
+]
+
 [[package]]
 name = "rustfs-appauth"
 version = "0.0.5"

Cargo.toml

@@ -36,6 +36,7 @@ members = [
     "crates/utils", # Utility functions and helpers
     "crates/workers", # Worker thread pools and task scheduling
     "crates/zip", # ZIP file handling and compression
+    "crates/ahm",
 ]
 resolver = "2"
 
@@ -62,6 +63,7 @@ rustfs-filemeta = { path = "crates/filemeta" }
 rustfs-rio = { path = "crates/rio" }
 
 [workspace.dependencies]
+rustfs-ahm = { path = "crates/ahm", version = "0.0.3" }
 rustfs-s3select-api = { path = "crates/s3select-api", version = "0.0.5" }
 rustfs-appauth = { path = "crates/appauth", version = "0.0.5" }
 rustfs-common = { path = "crates/common", version = "0.0.5" }
@@ -261,6 +263,7 @@ winapi = { version = "0.3.9" }
 xxhash-rust = { version = "0.8.15", features = ["xxh64", "xxh3"] }
 zip = "2.4.2"
 zstd = "0.13.3"
+anyhow = "1.0.86"
 
 [profile.wasm-dev]
 inherits = "dev"

crates/ahm/Cargo.toml

@@ -0,0 +1,35 @@
+[package]
+name = "rustfs-ahm"
+version = "0.0.3"
+edition = "2021"
+authors = ["RustFS Team"]
+license = "Apache-2.0"
+description = "RustFS AHM (Automatic Health Management) Scanner"
+
+[dependencies]
+rustfs-ecstore = { workspace = true }
+rustfs-common = { workspace = true }
+rustfs-filemeta = { workspace = true }
+rustfs-madmin = { workspace = true }
+rustfs-utils = { workspace = true }
+tokio = { workspace = true, features = ["full"] }
+tokio-util = { workspace = true }
+tracing = { workspace = true }
+serde = { workspace = true, features = ["derive"] }
+serde_json = { workspace = true }
+thiserror = { workspace = true }
+bytes = { workspace = true }
+time = { workspace = true, features = ["serde"] }
+uuid = { workspace = true, features = ["v4", "serde"] }
+anyhow = { workspace = true }
+async-trait = { workspace = true }
+futures = { workspace = true }
+url = { workspace = true }
+rustfs-lock = { workspace = true }
+
+lazy_static = { workspace = true }
+
+[dev-dependencies]
+rmp-serde = { workspace = true }
+tokio-test = "0.4"
+serde_json = "1.0"

crates/ahm/src/error.rs

@@ -0,0 +1,45 @@
+// 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.
+
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum Error {
+    #[error("I/O error: {0}")]
+    Io(#[from] std::io::Error),
+
+    #[error("Storage error: {0}")]
+    Storage(#[from] rustfs_ecstore::error::Error),
+
+    #[error("Configuration error: {0}")]
+    Config(String),
+
+    #[error("Scanner error: {0}")]
+    Scanner(String),
+
+    #[error("Metrics error: {0}")]
+    Metrics(String),
+
+    #[error(transparent)]
+    Other(#[from] anyhow::Error),
+}
+
+pub type Result<T, E = Error> = std::result::Result<T, E>;
+
+// Implement conversion from ahm::Error to std::io::Error for use in main.rs
+impl From<Error> for std::io::Error {
+    fn from(err: Error) -> Self {
+        std::io::Error::other(err)
+    }
+}

crates/ahm/src/lib.rs

@@ -0,0 +1,54 @@
+// 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.
+
+use std::sync::OnceLock;
+use tokio_util::sync::CancellationToken;
+
+pub mod error;
+pub mod scanner;
+
+pub use error::{Error, Result};
+pub use scanner::{
+    load_data_usage_from_backend, store_data_usage_in_backend, BucketTargetUsageInfo, BucketUsageInfo, DataUsageInfo, Scanner,
+    ScannerMetrics,
+};
+
+// Global cancellation token for AHM services (scanner and other background tasks)
+static GLOBAL_AHM_SERVICES_CANCEL_TOKEN: OnceLock<CancellationToken> = OnceLock::new();
+
+/// Initialize the global AHM services cancellation token
+pub fn init_ahm_services_cancel_token(cancel_token: CancellationToken) -> Result<()> {
+    GLOBAL_AHM_SERVICES_CANCEL_TOKEN
+        .set(cancel_token)
+        .map_err(|_| Error::Config("AHM services cancel token already initialized".to_string()))
+}
+
+/// Get the global AHM services cancellation token
+pub fn get_ahm_services_cancel_token() -> Option<&'static CancellationToken> {
+    GLOBAL_AHM_SERVICES_CANCEL_TOKEN.get()
+}
+
+/// Create and initialize the global AHM services cancellation token
+pub fn create_ahm_services_cancel_token() -> CancellationToken {
+    let cancel_token = CancellationToken::new();
+    init_ahm_services_cancel_token(cancel_token.clone()).expect("AHM services cancel token already initialized");
+    cancel_token
+}
+
+/// Shutdown all AHM services gracefully
+pub fn shutdown_ahm_services() {
+    if let Some(cancel_token) = GLOBAL_AHM_SERVICES_CANCEL_TOKEN.get() {
+        cancel_token.cancel();
+    }
+}

crates/ahm/src/scanner/data_usage.rs

@@ -0,0 +1,671 @@
+// 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.
+
+use std::{collections::HashMap, sync::Arc, time::SystemTime};
+
+use rustfs_ecstore::{bucket::metadata_sys::get_replication_config, config::com::read_config, store::ECStore};
+use rustfs_utils::path::SLASH_SEPARATOR;
+use serde::{Deserialize, Serialize};
+use tracing::{error, info, warn};
+
+use crate::error::{Error, Result};
+
+// Data usage storage constants
+pub const DATA_USAGE_ROOT: &str = SLASH_SEPARATOR;
+const DATA_USAGE_OBJ_NAME: &str = ".usage.json";
+const DATA_USAGE_BLOOM_NAME: &str = ".bloomcycle.bin";
+pub const DATA_USAGE_CACHE_NAME: &str = ".usage-cache.bin";
+
+// Data usage storage paths
+lazy_static::lazy_static! {
+    pub static ref DATA_USAGE_BUCKET: String = format!("{}{}{}",
+        rustfs_ecstore::disk::RUSTFS_META_BUCKET,
+        SLASH_SEPARATOR,
+        rustfs_ecstore::disk::BUCKET_META_PREFIX
+    );
+    pub static ref DATA_USAGE_OBJ_NAME_PATH: String = format!("{}{}{}",
+        rustfs_ecstore::disk::BUCKET_META_PREFIX,
+        SLASH_SEPARATOR,
+        DATA_USAGE_OBJ_NAME
+    );
+    pub static ref DATA_USAGE_BLOOM_NAME_PATH: String = format!("{}{}{}",
+        rustfs_ecstore::disk::BUCKET_META_PREFIX,
+        SLASH_SEPARATOR,
+        DATA_USAGE_BLOOM_NAME
+    );
+}
+
+/// Bucket target usage info provides replication statistics
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+pub struct BucketTargetUsageInfo {
+    pub replication_pending_size: u64,
+    pub replication_failed_size: u64,
+    pub replicated_size: u64,
+    pub replica_size: u64,
+    pub replication_pending_count: u64,
+    pub replication_failed_count: u64,
+    pub replicated_count: u64,
+}
+
+/// Bucket usage info provides bucket-level statistics
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+pub struct BucketUsageInfo {
+    pub size: u64,
+    // Following five fields suffixed with V1 are here for backward compatibility
+    // Total Size for objects that have not yet been replicated
+    pub replication_pending_size_v1: u64,
+    // Total size for objects that have witness one or more failures and will be retried
+    pub replication_failed_size_v1: u64,
+    // Total size for objects that have been replicated to destination
+    pub replicated_size_v1: u64,
+    // Total number of objects pending replication
+    pub replication_pending_count_v1: u64,
+    // Total number of objects that failed replication
+    pub replication_failed_count_v1: u64,
+
+    pub objects_count: u64,
+    pub object_size_histogram: HashMap<String, u64>,
+    pub object_versions_histogram: HashMap<String, u64>,
+    pub versions_count: u64,
+    pub delete_markers_count: u64,
+    pub replica_size: u64,
+    pub replica_count: u64,
+    pub replication_info: HashMap<String, BucketTargetUsageInfo>,
+}
+
+/// DataUsageInfo represents data usage stats of the underlying storage
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+pub struct DataUsageInfo {
+    /// Total capacity
+    pub total_capacity: u64,
+    /// Total used capacity
+    pub total_used_capacity: u64,
+    /// Total free capacity
+    pub total_free_capacity: u64,
+
+    /// LastUpdate is the timestamp of when the data usage info was last updated
+    pub last_update: Option<SystemTime>,
+
+    /// Objects total count across all buckets
+    pub objects_total_count: u64,
+    /// Versions total count across all buckets
+    pub versions_total_count: u64,
+    /// Delete markers total count across all buckets
+    pub delete_markers_total_count: u64,
+    /// Objects total size across all buckets
+    pub objects_total_size: u64,
+    /// Replication info across all buckets
+    pub replication_info: HashMap<String, BucketTargetUsageInfo>,
+
+    /// Total number of buckets in this cluster
+    pub buckets_count: u64,
+    /// Buckets usage info provides following information across all buckets
+    pub buckets_usage: HashMap<String, BucketUsageInfo>,
+    /// Deprecated kept here for backward compatibility reasons
+    pub bucket_sizes: HashMap<String, u64>,
+}
+
+/// Size summary for a single object or group of objects
+#[derive(Debug, Default, Clone)]
+pub struct SizeSummary {
+    /// Total size
+    pub total_size: usize,
+    /// Number of versions
+    pub versions: usize,
+    /// Number of delete markers
+    pub delete_markers: usize,
+    /// Replicated size
+    pub replicated_size: usize,
+    /// Replicated count
+    pub replicated_count: usize,
+    /// Pending size
+    pub pending_size: usize,
+    /// Failed size
+    pub failed_size: usize,
+    /// Replica size
+    pub replica_size: usize,
+    /// Replica count
+    pub replica_count: usize,
+    /// Pending count
+    pub pending_count: usize,
+    /// Failed count
+    pub failed_count: usize,
+    /// Replication target stats
+    pub repl_target_stats: HashMap<String, ReplTargetSizeSummary>,
+}
+
+/// Replication target size summary
+#[derive(Debug, Default, Clone)]
+pub struct ReplTargetSizeSummary {
+    /// Replicated size
+    pub replicated_size: usize,
+    /// Replicated count
+    pub replicated_count: usize,
+    /// Pending size
+    pub pending_size: usize,
+    /// Failed size
+    pub failed_size: usize,
+    /// Pending count
+    pub pending_count: usize,
+    /// Failed count
+    pub failed_count: usize,
+}
+
+impl DataUsageInfo {
+    /// Create a new DataUsageInfo
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Add object metadata to data usage statistics
+    pub fn add_object(&mut self, object_path: &str, meta_object: &rustfs_filemeta::MetaObject) {
+        // This method is kept for backward compatibility
+        // For accurate version counting, use add_object_from_file_meta instead
+        let bucket_name = match self.extract_bucket_from_path(object_path) {
+            Ok(name) => name,
+            Err(_) => return,
+        };
+
+        // Update bucket statistics
+        if let Some(bucket_usage) = self.buckets_usage.get_mut(&bucket_name) {
+            bucket_usage.size += meta_object.size as u64;
+            bucket_usage.objects_count += 1;
+            bucket_usage.versions_count += 1; // Simplified: assume 1 version per object
+
+            // Update size histogram
+            let total_size = meta_object.size as u64;
+            let size_ranges = [
+                ("0-1KB", 0, 1024),
+                ("1KB-1MB", 1024, 1024 * 1024),
+                ("1MB-10MB", 1024 * 1024, 10 * 1024 * 1024),
+                ("10MB-100MB", 10 * 1024 * 1024, 100 * 1024 * 1024),
+                ("100MB-1GB", 100 * 1024 * 1024, 1024 * 1024 * 1024),
+                ("1GB+", 1024 * 1024 * 1024, u64::MAX),
+            ];
+
+            for (range_name, min_size, max_size) in size_ranges {
+                if total_size >= min_size && total_size < max_size {
+                    *bucket_usage.object_size_histogram.entry(range_name.to_string()).or_insert(0) += 1;
+                    break;
+                }
+            }
+
+            // Update version histogram (simplified - count as single version)
+            *bucket_usage
+                .object_versions_histogram
+                .entry("SINGLE_VERSION".to_string())
+                .or_insert(0) += 1;
+        } else {
+            // Create new bucket usage
+            let mut bucket_usage = BucketUsageInfo {
+                size: meta_object.size as u64,
+                objects_count: 1,
+                versions_count: 1,
+                ..Default::default()
+            };
+            bucket_usage.object_size_histogram.insert("0-1KB".to_string(), 1);
+            bucket_usage.object_versions_histogram.insert("SINGLE_VERSION".to_string(), 1);
+            self.buckets_usage.insert(bucket_name, bucket_usage);
+        }
+
+        // Update global statistics
+        self.objects_total_size += meta_object.size as u64;
+        self.objects_total_count += 1;
+        self.versions_total_count += 1;
+    }
+
+    /// Add object from FileMeta for accurate version counting
+    pub fn add_object_from_file_meta(&mut self, object_path: &str, file_meta: &rustfs_filemeta::FileMeta) {
+        let bucket_name = match self.extract_bucket_from_path(object_path) {
+            Ok(name) => name,
+            Err(_) => return,
+        };
+
+        // Calculate accurate statistics from all versions
+        let mut total_size = 0u64;
+        let mut versions_count = 0u64;
+        let mut delete_markers_count = 0u64;
+        let mut latest_object_size = 0u64;
+
+        // Process all versions to get accurate counts
+        for version in &file_meta.versions {
+            match rustfs_filemeta::FileMetaVersion::try_from(version.clone()) {
+                Ok(ver) => {
+                    if let Some(obj) = ver.object {
+                        total_size += obj.size as u64;
+                        versions_count += 1;
+                        latest_object_size = obj.size as u64; // Keep track of latest object size
+                    } else if ver.delete_marker.is_some() {
+                        delete_markers_count += 1;
+                    }
+                }
+                Err(_) => {
+                    // Skip invalid versions
+                    continue;
+                }
+            }
+        }
+
+        // Update bucket statistics
+        if let Some(bucket_usage) = self.buckets_usage.get_mut(&bucket_name) {
+            bucket_usage.size += total_size;
+            bucket_usage.objects_count += 1;
+            bucket_usage.versions_count += versions_count;
+            bucket_usage.delete_markers_count += delete_markers_count;
+
+            // Update size histogram based on latest object size
+            let size_ranges = [
+                ("0-1KB", 0, 1024),
+                ("1KB-1MB", 1024, 1024 * 1024),
+                ("1MB-10MB", 1024 * 1024, 10 * 1024 * 1024),
+                ("10MB-100MB", 10 * 1024 * 1024, 100 * 1024 * 1024),
+                ("100MB-1GB", 100 * 1024 * 1024, 1024 * 1024 * 1024),
+                ("1GB+", 1024 * 1024 * 1024, u64::MAX),
+            ];
+
+            for (range_name, min_size, max_size) in size_ranges {
+                if latest_object_size >= min_size && latest_object_size < max_size {
+                    *bucket_usage.object_size_histogram.entry(range_name.to_string()).or_insert(0) += 1;
+                    break;
+                }
+            }
+
+            // Update version histogram based on actual version count
+            let version_ranges = [
+                ("1", 1, 1),
+                ("2-5", 2, 5),
+                ("6-10", 6, 10),
+                ("11-50", 11, 50),
+                ("51-100", 51, 100),
+                ("100+", 101, usize::MAX),
+            ];
+
+            for (range_name, min_versions, max_versions) in version_ranges {
+                if versions_count as usize >= min_versions && versions_count as usize <= max_versions {
+                    *bucket_usage
+                        .object_versions_histogram
+                        .entry(range_name.to_string())
+                        .or_insert(0) += 1;
+                    break;
+                }
+            }
+        } else {
+            // Create new bucket usage
+            let mut bucket_usage = BucketUsageInfo {
+                size: total_size,
+                objects_count: 1,
+                versions_count,
+                delete_markers_count,
+                ..Default::default()
+            };
+
+            // Set size histogram
+            let size_ranges = [
+                ("0-1KB", 0, 1024),
+                ("1KB-1MB", 1024, 1024 * 1024),
+                ("1MB-10MB", 1024 * 1024, 10 * 1024 * 1024),
+                ("10MB-100MB", 10 * 1024 * 1024, 100 * 1024 * 1024),
+                ("100MB-1GB", 100 * 1024 * 1024, 1024 * 1024 * 1024),
+                ("1GB+", 1024 * 1024 * 1024, u64::MAX),
+            ];
+
+            for (range_name, min_size, max_size) in size_ranges {
+                if latest_object_size >= min_size && latest_object_size < max_size {
+                    bucket_usage.object_size_histogram.insert(range_name.to_string(), 1);
+                    break;
+                }
+            }
+
+            // Set version histogram
+            let version_ranges = [
+                ("1", 1, 1),
+                ("2-5", 2, 5),
+                ("6-10", 6, 10),
+                ("11-50", 11, 50),
+                ("51-100", 51, 100),
+                ("100+", 101, usize::MAX),
+            ];
+
+            for (range_name, min_versions, max_versions) in version_ranges {
+                if versions_count as usize >= min_versions && versions_count as usize <= max_versions {
+                    bucket_usage.object_versions_histogram.insert(range_name.to_string(), 1);
+                    break;
+                }
+            }
+
+            self.buckets_usage.insert(bucket_name, bucket_usage);
+            // Update buckets count when adding new bucket
+            self.buckets_count = self.buckets_usage.len() as u64;
+        }
+
+        // Update global statistics
+        self.objects_total_size += total_size;
+        self.objects_total_count += 1;
+        self.versions_total_count += versions_count;
+        self.delete_markers_total_count += delete_markers_count;
+    }
+
+    /// Extract bucket name from object path
+    fn extract_bucket_from_path(&self, object_path: &str) -> Result<String> {
+        let parts: Vec<&str> = object_path.split('/').collect();
+        if parts.is_empty() {
+            return Err(Error::Scanner("Invalid object path: empty".to_string()));
+        }
+        Ok(parts[0].to_string())
+    }
+
+    /// Update capacity information
+    pub fn update_capacity(&mut self, total: u64, used: u64, free: u64) {
+        self.total_capacity = total;
+        self.total_used_capacity = used;
+        self.total_free_capacity = free;
+        self.last_update = Some(SystemTime::now());
+    }
+
+    /// Add bucket usage info
+    pub fn add_bucket_usage(&mut self, bucket: String, usage: BucketUsageInfo) {
+        self.buckets_usage.insert(bucket.clone(), usage);
+        self.buckets_count = self.buckets_usage.len() as u64;
+        self.last_update = Some(SystemTime::now());
+    }
+
+    /// Get bucket usage info
+    pub fn get_bucket_usage(&self, bucket: &str) -> Option<&BucketUsageInfo> {
+        self.buckets_usage.get(bucket)
+    }
+
+    /// Calculate total statistics from all buckets
+    pub fn calculate_totals(&mut self) {
+        self.objects_total_count = 0;
+        self.versions_total_count = 0;
+        self.delete_markers_total_count = 0;
+        self.objects_total_size = 0;
+
+        for usage in self.buckets_usage.values() {
+            self.objects_total_count += usage.objects_count;
+            self.versions_total_count += usage.versions_count;
+            self.delete_markers_total_count += usage.delete_markers_count;
+            self.objects_total_size += usage.size;
+        }
+    }
+
+    /// Merge another DataUsageInfo into this one
+    pub fn merge(&mut self, other: &DataUsageInfo) {
+        // Merge bucket usage
+        for (bucket, usage) in &other.buckets_usage {
+            if let Some(existing) = self.buckets_usage.get_mut(bucket) {
+                existing.merge(usage);
+            } else {
+                self.buckets_usage.insert(bucket.clone(), usage.clone());
+            }
+        }
+
+        // Recalculate totals
+        self.calculate_totals();
+
+        // Ensure buckets_count stays consistent with buckets_usage
+        self.buckets_count = self.buckets_usage.len() as u64;
+
+        // Update last update time
+        if let Some(other_update) = other.last_update {
+            if self.last_update.is_none() || other_update > self.last_update.unwrap() {
+                self.last_update = Some(other_update);
+            }
+        }
+    }
+}
+
+impl BucketUsageInfo {
+    /// Create a new BucketUsageInfo
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Add size summary to this bucket usage
+    pub fn add_size_summary(&mut self, summary: &SizeSummary) {
+        self.size += summary.total_size as u64;
+        self.versions_count += summary.versions as u64;
+        self.delete_markers_count += summary.delete_markers as u64;
+        self.replica_size += summary.replica_size as u64;
+        self.replica_count += summary.replica_count as u64;
+    }
+
+    /// Merge another BucketUsageInfo into this one
+    pub fn merge(&mut self, other: &BucketUsageInfo) {
+        self.size += other.size;
+        self.objects_count += other.objects_count;
+        self.versions_count += other.versions_count;
+        self.delete_markers_count += other.delete_markers_count;
+        self.replica_size += other.replica_size;
+        self.replica_count += other.replica_count;
+
+        // Merge histograms
+        for (key, value) in &other.object_size_histogram {
+            *self.object_size_histogram.entry(key.clone()).or_insert(0) += value;
+        }
+
+        for (key, value) in &other.object_versions_histogram {
+            *self.object_versions_histogram.entry(key.clone()).or_insert(0) += value;
+        }
+
+        // Merge replication info
+        for (target, info) in &other.replication_info {
+            let entry = self.replication_info.entry(target.clone()).or_default();
+            entry.replicated_size += info.replicated_size;
+            entry.replica_size += info.replica_size;
+            entry.replication_pending_size += info.replication_pending_size;
+            entry.replication_failed_size += info.replication_failed_size;
+            entry.replication_pending_count += info.replication_pending_count;
+            entry.replication_failed_count += info.replication_failed_count;
+            entry.replicated_count += info.replicated_count;
+        }
+
+        // Merge backward compatibility fields
+        self.replication_pending_size_v1 += other.replication_pending_size_v1;
+        self.replication_failed_size_v1 += other.replication_failed_size_v1;
+        self.replicated_size_v1 += other.replicated_size_v1;
+        self.replication_pending_count_v1 += other.replication_pending_count_v1;
+        self.replication_failed_count_v1 += other.replication_failed_count_v1;
+    }
+}
+
+impl SizeSummary {
+    /// Create a new SizeSummary
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Add another SizeSummary to this one
+    pub fn add(&mut self, other: &SizeSummary) {
+        self.total_size += other.total_size;
+        self.versions += other.versions;
+        self.delete_markers += other.delete_markers;
+        self.replicated_size += other.replicated_size;
+        self.replicated_count += other.replicated_count;
+        self.pending_size += other.pending_size;
+        self.failed_size += other.failed_size;
+        self.replica_size += other.replica_size;
+        self.replica_count += other.replica_count;
+        self.pending_count += other.pending_count;
+        self.failed_count += other.failed_count;
+
+        // Merge replication target stats
+        for (target, stats) in &other.repl_target_stats {
+            let entry = self.repl_target_stats.entry(target.clone()).or_default();
+            entry.replicated_size += stats.replicated_size;
+            entry.replicated_count += stats.replicated_count;
+            entry.pending_size += stats.pending_size;
+            entry.failed_size += stats.failed_size;
+            entry.pending_count += stats.pending_count;
+            entry.failed_count += stats.failed_count;
+        }
+    }
+}
+
+/// Store data usage info to backend storage
+pub async fn store_data_usage_in_backend(data_usage_info: DataUsageInfo, store: Arc<ECStore>) -> Result<()> {
+    let data =
+        serde_json::to_vec(&data_usage_info).map_err(|e| Error::Config(format!("Failed to serialize data usage info: {e}")))?;
+
+    // Save to backend using the same mechanism as original code
+    rustfs_ecstore::config::com::save_config(store, &DATA_USAGE_OBJ_NAME_PATH, data)
+        .await
+        .map_err(Error::Storage)?;
+
+    Ok(())
+}
+
+/// Load data usage info from backend storage
+pub async fn load_data_usage_from_backend(store: Arc<ECStore>) -> Result<DataUsageInfo> {
+    let buf = match read_config(store, &DATA_USAGE_OBJ_NAME_PATH).await {
+        Ok(data) => data,
+        Err(e) => {
+            error!("Failed to read data usage info from backend: {}", e);
+            if e == rustfs_ecstore::error::Error::ConfigNotFound {
+                return Ok(DataUsageInfo::default());
+            }
+            return Err(Error::Storage(e));
+        }
+    };
+
+    let mut data_usage_info: DataUsageInfo =
+        serde_json::from_slice(&buf).map_err(|e| Error::Config(format!("Failed to deserialize data usage info: {e}")))?;
+
+    warn!("Loaded data usage info from backend {:?}", &data_usage_info);
+
+    // Handle backward compatibility like original code
+    if data_usage_info.buckets_usage.is_empty() {
+        data_usage_info.buckets_usage = data_usage_info
+            .bucket_sizes
+            .iter()
+            .map(|(bucket, &size)| {
+                (
+                    bucket.clone(),
+                    BucketUsageInfo {
+                        size,
+                        ..Default::default()
+                    },
+                )
+            })
+            .collect();
+    }
+
+    if data_usage_info.bucket_sizes.is_empty() {
+        data_usage_info.bucket_sizes = data_usage_info
+            .buckets_usage
+            .iter()
+            .map(|(bucket, bui)| (bucket.clone(), bui.size))
+            .collect();
+    }
+
+    for (bucket, bui) in &data_usage_info.buckets_usage {
+        if bui.replicated_size_v1 > 0
+            || bui.replication_failed_count_v1 > 0
+            || bui.replication_failed_size_v1 > 0
+            || bui.replication_pending_count_v1 > 0
+        {
+            if let Ok((cfg, _)) = get_replication_config(bucket).await {
+                if !cfg.role.is_empty() {
+                    data_usage_info.replication_info.insert(
+                        cfg.role.clone(),
+                        BucketTargetUsageInfo {
+                            replication_failed_size: bui.replication_failed_size_v1,
+                            replication_failed_count: bui.replication_failed_count_v1,
+                            replicated_size: bui.replicated_size_v1,
+                            replication_pending_count: bui.replication_pending_count_v1,
+                            replication_pending_size: bui.replication_pending_size_v1,
+                            ..Default::default()
+                        },
+                    );
+                }
+            }
+        }
+    }
+
+    Ok(data_usage_info)
+}
+
+/// Example function showing how to use AHM data usage functionality
+/// This demonstrates the integration pattern for DataUsageInfoHandler
+pub async fn example_data_usage_integration() -> Result<()> {
+    // Get the global storage instance
+    let Some(store) = rustfs_ecstore::new_object_layer_fn() else {
+        return Err(Error::Config("Storage not initialized".to_string()));
+    };
+
+    // Load data usage from backend (this replaces the original load_data_usage_from_backend)
+    let data_usage = load_data_usage_from_backend(store).await?;
+
+    info!(
+        "Loaded data usage info: {} buckets, {} total objects",
+        data_usage.buckets_count, data_usage.objects_total_count
+    );
+
+    // Example: Store updated data usage back to backend
+    // This would typically be called by the scanner after collecting new statistics
+    // store_data_usage_in_backend(data_usage, store).await?;
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_data_usage_info_creation() {
+        let mut info = DataUsageInfo::new();
+        info.update_capacity(1000, 500, 500);
+
+        assert_eq!(info.total_capacity, 1000);
+        assert_eq!(info.total_used_capacity, 500);
+        assert_eq!(info.total_free_capacity, 500);
+        assert!(info.last_update.is_some());
+    }
+
+    #[test]
+    fn test_bucket_usage_info_merge() {
+        let mut usage1 = BucketUsageInfo::new();
+        usage1.size = 100;
+        usage1.objects_count = 10;
+        usage1.versions_count = 5;
+
+        let mut usage2 = BucketUsageInfo::new();
+        usage2.size = 200;
+        usage2.objects_count = 20;
+        usage2.versions_count = 10;
+
+        usage1.merge(&usage2);
+
+        assert_eq!(usage1.size, 300);
+        assert_eq!(usage1.objects_count, 30);
+        assert_eq!(usage1.versions_count, 15);
+    }
+
+    #[test]
+    fn test_size_summary_add() {
+        let mut summary1 = SizeSummary::new();
+        summary1.total_size = 100;
+        summary1.versions = 5;
+
+        let mut summary2 = SizeSummary::new();
+        summary2.total_size = 200;
+        summary2.versions = 10;
+
+        summary1.add(&summary2);
+
+        assert_eq!(summary1.total_size, 300);
+        assert_eq!(summary1.versions, 15);
+    }
+}

crates/ahm/src/scanner/histogram.rs

@@ -0,0 +1,277 @@
+// 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.
+
+use std::collections::HashMap;
+
+/// Size interval for object size histogram
+#[derive(Debug, Clone)]
+pub struct SizeInterval {
+    pub start: u64,
+    pub end: u64,
+    pub name: &'static str,
+}
+
+/// Version interval for object versions histogram
+#[derive(Debug, Clone)]
+pub struct VersionInterval {
+    pub start: u64,
+    pub end: u64,
+    pub name: &'static str,
+}
+
+/// Object size histogram intervals
+pub const OBJECTS_HISTOGRAM_INTERVALS: &[SizeInterval] = &[
+    SizeInterval {
+        start: 0,
+        end: 1024 - 1,
+        name: "LESS_THAN_1_KiB",
+    },
+    SizeInterval {
+        start: 1024,
+        end: 1024 * 1024 - 1,
+        name: "1_KiB_TO_1_MiB",
+    },
+    SizeInterval {
+        start: 1024 * 1024,
+        end: 10 * 1024 * 1024 - 1,
+        name: "1_MiB_TO_10_MiB",
+    },
+    SizeInterval {
+        start: 10 * 1024 * 1024,
+        end: 64 * 1024 * 1024 - 1,
+        name: "10_MiB_TO_64_MiB",
+    },
+    SizeInterval {
+        start: 64 * 1024 * 1024,
+        end: 128 * 1024 * 1024 - 1,
+        name: "64_MiB_TO_128_MiB",
+    },
+    SizeInterval {
+        start: 128 * 1024 * 1024,
+        end: 512 * 1024 * 1024 - 1,
+        name: "128_MiB_TO_512_MiB",
+    },
+    SizeInterval {
+        start: 512 * 1024 * 1024,
+        end: u64::MAX,
+        name: "MORE_THAN_512_MiB",
+    },
+];
+
+/// Object version count histogram intervals
+pub const OBJECTS_VERSION_COUNT_INTERVALS: &[VersionInterval] = &[
+    VersionInterval {
+        start: 1,
+        end: 1,
+        name: "1_VERSION",
+    },
+    VersionInterval {
+        start: 2,
+        end: 10,
+        name: "2_TO_10_VERSIONS",
+    },
+    VersionInterval {
+        start: 11,
+        end: 100,
+        name: "11_TO_100_VERSIONS",
+    },
+    VersionInterval {
+        start: 101,
+        end: 1000,
+        name: "101_TO_1000_VERSIONS",
+    },
+    VersionInterval {
+        start: 1001,
+        end: u64::MAX,
+        name: "MORE_THAN_1000_VERSIONS",
+    },
+];
+
+/// Size histogram for object size distribution
+#[derive(Debug, Clone, Default)]
+pub struct SizeHistogram {
+    counts: Vec<u64>,
+}
+
+/// Versions histogram for object version count distribution
+#[derive(Debug, Clone, Default)]
+pub struct VersionsHistogram {
+    counts: Vec<u64>,
+}
+
+impl SizeHistogram {
+    /// Create a new size histogram
+    pub fn new() -> Self {
+        Self {
+            counts: vec![0; OBJECTS_HISTOGRAM_INTERVALS.len()],
+        }
+    }
+
+    /// Add a size to the histogram
+    pub fn add(&mut self, size: u64) {
+        for (idx, interval) in OBJECTS_HISTOGRAM_INTERVALS.iter().enumerate() {
+            if size >= interval.start && size <= interval.end {
+                self.counts[idx] += 1;
+                break;
+            }
+        }
+    }
+
+    /// Get the histogram as a map
+    pub fn to_map(&self) -> HashMap<String, u64> {
+        let mut result = HashMap::new();
+        for (idx, count) in self.counts.iter().enumerate() {
+            let interval = &OBJECTS_HISTOGRAM_INTERVALS[idx];
+            result.insert(interval.name.to_string(), *count);
+        }
+        result
+    }
+
+    /// Merge another histogram into this one
+    pub fn merge(&mut self, other: &SizeHistogram) {
+        for (idx, count) in other.counts.iter().enumerate() {
+            self.counts[idx] += count;
+        }
+    }
+
+    /// Get total count
+    pub fn total_count(&self) -> u64 {
+        self.counts.iter().sum()
+    }
+
+    /// Reset the histogram
+    pub fn reset(&mut self) {
+        for count in &mut self.counts {
+            *count = 0;
+        }
+    }
+}
+
+impl VersionsHistogram {
+    /// Create a new versions histogram
+    pub fn new() -> Self {
+        Self {
+            counts: vec![0; OBJECTS_VERSION_COUNT_INTERVALS.len()],
+        }
+    }
+
+    /// Add a version count to the histogram
+    pub fn add(&mut self, versions: u64) {
+        for (idx, interval) in OBJECTS_VERSION_COUNT_INTERVALS.iter().enumerate() {
+            if versions >= interval.start && versions <= interval.end {
+                self.counts[idx] += 1;
+                break;
+            }
+        }
+    }
+
+    /// Get the histogram as a map
+    pub fn to_map(&self) -> HashMap<String, u64> {
+        let mut result = HashMap::new();
+        for (idx, count) in self.counts.iter().enumerate() {
+            let interval = &OBJECTS_VERSION_COUNT_INTERVALS[idx];
+            result.insert(interval.name.to_string(), *count);
+        }
+        result
+    }
+
+    /// Merge another histogram into this one
+    pub fn merge(&mut self, other: &VersionsHistogram) {
+        for (idx, count) in other.counts.iter().enumerate() {
+            self.counts[idx] += count;
+        }
+    }
+
+    /// Get total count
+    pub fn total_count(&self) -> u64 {
+        self.counts.iter().sum()
+    }
+
+    /// Reset the histogram
+    pub fn reset(&mut self) {
+        for count in &mut self.counts {
+            *count = 0;
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_size_histogram() {
+        let mut histogram = SizeHistogram::new();
+
+        // Add some sizes
+        histogram.add(512); // LESS_THAN_1_KiB
+        histogram.add(1024); // 1_KiB_TO_1_MiB
+        histogram.add(1024 * 1024); // 1_MiB_TO_10_MiB
+        histogram.add(5 * 1024 * 1024); // 1_MiB_TO_10_MiB
+
+        let map = histogram.to_map();
+
+        assert_eq!(map.get("LESS_THAN_1_KiB"), Some(&1));
+        assert_eq!(map.get("1_KiB_TO_1_MiB"), Some(&1));
+        assert_eq!(map.get("1_MiB_TO_10_MiB"), Some(&2));
+        assert_eq!(map.get("10_MiB_TO_64_MiB"), Some(&0));
+    }
+
+    #[test]
+    fn test_versions_histogram() {
+        let mut histogram = VersionsHistogram::new();
+
+        // Add some version counts
+        histogram.add(1); // 1_VERSION
+        histogram.add(5); // 2_TO_10_VERSIONS
+        histogram.add(50); // 11_TO_100_VERSIONS
+        histogram.add(500); // 101_TO_1000_VERSIONS
+
+        let map = histogram.to_map();
+
+        assert_eq!(map.get("1_VERSION"), Some(&1));
+        assert_eq!(map.get("2_TO_10_VERSIONS"), Some(&1));
+        assert_eq!(map.get("11_TO_100_VERSIONS"), Some(&1));
+        assert_eq!(map.get("101_TO_1000_VERSIONS"), Some(&1));
+    }
+
+    #[test]
+    fn test_histogram_merge() {
+        let mut histogram1 = SizeHistogram::new();
+        histogram1.add(1024);
+        histogram1.add(1024 * 1024);
+
+        let mut histogram2 = SizeHistogram::new();
+        histogram2.add(1024);
+        histogram2.add(5 * 1024 * 1024);
+
+        histogram1.merge(&histogram2);
+
+        let map = histogram1.to_map();
+        assert_eq!(map.get("1_KiB_TO_1_MiB"), Some(&2)); // 1 from histogram1 + 1 from histogram2
+        assert_eq!(map.get("1_MiB_TO_10_MiB"), Some(&2)); // 1 from histogram1 + 1 from histogram2
+    }
+
+    #[test]
+    fn test_histogram_reset() {
+        let mut histogram = SizeHistogram::new();
+        histogram.add(1024);
+        histogram.add(1024 * 1024);
+
+        assert_eq!(histogram.total_count(), 2);
+
+        histogram.reset();
+        assert_eq!(histogram.total_count(), 0);
+    }
+}

crates/ahm/src/scanner/metrics.rs

@@ -0,0 +1,284 @@
+// 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.
+
+use std::{
+    collections::HashMap,
+    sync::atomic::{AtomicU64, Ordering},
+    time::{Duration, SystemTime},
+};
+
+use serde::{Deserialize, Serialize};
+use tracing::info;
+
+/// Scanner metrics
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+pub struct ScannerMetrics {
+    /// Total objects scanned since server start
+    pub objects_scanned: u64,
+    /// Total object versions scanned since server start
+    pub versions_scanned: u64,
+    /// Total directories scanned since server start
+    pub directories_scanned: u64,
+    /// Total bucket scans started since server start
+    pub bucket_scans_started: u64,
+    /// Total bucket scans finished since server start
+    pub bucket_scans_finished: u64,
+    /// Total objects with health issues found
+    pub objects_with_issues: u64,
+    /// Total heal tasks queued
+    pub heal_tasks_queued: u64,
+    /// Total heal tasks completed
+    pub heal_tasks_completed: u64,
+    /// Total heal tasks failed
+    pub heal_tasks_failed: u64,
+    /// Last scan activity time
+    pub last_activity: Option<SystemTime>,
+    /// Current scan cycle
+    pub current_cycle: u64,
+    /// Total scan cycles completed
+    pub total_cycles: u64,
+    /// Current scan duration
+    pub current_scan_duration: Option<Duration>,
+    /// Average scan duration
+    pub avg_scan_duration: Duration,
+    /// Objects scanned per second
+    pub objects_per_second: f64,
+    /// Buckets scanned per second
+    pub buckets_per_second: f64,
+    /// Storage metrics by bucket
+    pub bucket_metrics: HashMap<String, BucketMetrics>,
+    /// Disk metrics
+    pub disk_metrics: HashMap<String, DiskMetrics>,
+}
+
+/// Bucket-specific metrics
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+pub struct BucketMetrics {
+    /// Bucket name
+    pub bucket: String,
+    /// Total objects in bucket
+    pub total_objects: u64,
+    /// Total size of objects in bucket (bytes)
+    pub total_size: u64,
+    /// Objects with health issues
+    pub objects_with_issues: u64,
+    /// Last scan time
+    pub last_scan_time: Option<SystemTime>,
+    /// Scan duration
+    pub scan_duration: Option<Duration>,
+    /// Heal tasks queued for this bucket
+    pub heal_tasks_queued: u64,
+    /// Heal tasks completed for this bucket
+    pub heal_tasks_completed: u64,
+    /// Heal tasks failed for this bucket
+    pub heal_tasks_failed: u64,
+}
+
+/// Disk-specific metrics
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+pub struct DiskMetrics {
+    /// Disk path
+    pub disk_path: String,
+    /// Total disk space (bytes)
+    pub total_space: u64,
+    /// Used disk space (bytes)
+    pub used_space: u64,
+    /// Free disk space (bytes)
+    pub free_space: u64,
+    /// Objects scanned on this disk
+    pub objects_scanned: u64,
+    /// Objects with issues on this disk
+    pub objects_with_issues: u64,
+    /// Last scan time
+    pub last_scan_time: Option<SystemTime>,
+    /// Whether disk is online
+    pub is_online: bool,
+    /// Whether disk is being scanned
+    pub is_scanning: bool,
+}
+
+/// Thread-safe metrics collector
+pub struct MetricsCollector {
+    /// Atomic counters for real-time metrics
+    objects_scanned: AtomicU64,
+    versions_scanned: AtomicU64,
+    directories_scanned: AtomicU64,
+    bucket_scans_started: AtomicU64,
+    bucket_scans_finished: AtomicU64,
+    objects_with_issues: AtomicU64,
+    heal_tasks_queued: AtomicU64,
+    heal_tasks_completed: AtomicU64,
+    heal_tasks_failed: AtomicU64,
+    current_cycle: AtomicU64,
+    total_cycles: AtomicU64,
+}
+
+impl MetricsCollector {
+    /// Create a new metrics collector
+    pub fn new() -> Self {
+        Self {
+            objects_scanned: AtomicU64::new(0),
+            versions_scanned: AtomicU64::new(0),
+            directories_scanned: AtomicU64::new(0),
+            bucket_scans_started: AtomicU64::new(0),
+            bucket_scans_finished: AtomicU64::new(0),
+            objects_with_issues: AtomicU64::new(0),
+            heal_tasks_queued: AtomicU64::new(0),
+            heal_tasks_completed: AtomicU64::new(0),
+            heal_tasks_failed: AtomicU64::new(0),
+            current_cycle: AtomicU64::new(0),
+            total_cycles: AtomicU64::new(0),
+        }
+    }
+
+    /// Increment objects scanned count
+    pub fn increment_objects_scanned(&self, count: u64) {
+        self.objects_scanned.fetch_add(count, Ordering::Relaxed);
+    }
+
+    /// Increment versions scanned count
+    pub fn increment_versions_scanned(&self, count: u64) {
+        self.versions_scanned.fetch_add(count, Ordering::Relaxed);
+    }
+
+    /// Increment directories scanned count
+    pub fn increment_directories_scanned(&self, count: u64) {
+        self.directories_scanned.fetch_add(count, Ordering::Relaxed);
+    }
+
+    /// Increment bucket scans started count
+    pub fn increment_bucket_scans_started(&self, count: u64) {
+        self.bucket_scans_started.fetch_add(count, Ordering::Relaxed);
+    }
+
+    /// Increment bucket scans finished count
+    pub fn increment_bucket_scans_finished(&self, count: u64) {
+        self.bucket_scans_finished.fetch_add(count, Ordering::Relaxed);
+    }
+
+    /// Increment objects with issues count
+    pub fn increment_objects_with_issues(&self, count: u64) {
+        self.objects_with_issues.fetch_add(count, Ordering::Relaxed);
+    }
+
+    /// Increment heal tasks queued count
+    pub fn increment_heal_tasks_queued(&self, count: u64) {
+        self.heal_tasks_queued.fetch_add(count, Ordering::Relaxed);
+    }
+
+    /// Increment heal tasks completed count
+    pub fn increment_heal_tasks_completed(&self, count: u64) {
+        self.heal_tasks_completed.fetch_add(count, Ordering::Relaxed);
+    }
+
+    /// Increment heal tasks failed count
+    pub fn increment_heal_tasks_failed(&self, count: u64) {
+        self.heal_tasks_failed.fetch_add(count, Ordering::Relaxed);
+    }
+
+    /// Set current cycle
+    pub fn set_current_cycle(&self, cycle: u64) {
+        self.current_cycle.store(cycle, Ordering::Relaxed);
+    }
+
+    /// Increment total cycles
+    pub fn increment_total_cycles(&self) {
+        self.total_cycles.fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Get current metrics snapshot
+    pub fn get_metrics(&self) -> ScannerMetrics {
+        ScannerMetrics {
+            objects_scanned: self.objects_scanned.load(Ordering::Relaxed),
+            versions_scanned: self.versions_scanned.load(Ordering::Relaxed),
+            directories_scanned: self.directories_scanned.load(Ordering::Relaxed),
+            bucket_scans_started: self.bucket_scans_started.load(Ordering::Relaxed),
+            bucket_scans_finished: self.bucket_scans_finished.load(Ordering::Relaxed),
+            objects_with_issues: self.objects_with_issues.load(Ordering::Relaxed),
+            heal_tasks_queued: self.heal_tasks_queued.load(Ordering::Relaxed),
+            heal_tasks_completed: self.heal_tasks_completed.load(Ordering::Relaxed),
+            heal_tasks_failed: self.heal_tasks_failed.load(Ordering::Relaxed),
+            last_activity: Some(SystemTime::now()),
+            current_cycle: self.current_cycle.load(Ordering::Relaxed),
+            total_cycles: self.total_cycles.load(Ordering::Relaxed),
+            current_scan_duration: None,       // Will be set by scanner
+            avg_scan_duration: Duration::ZERO, // Will be calculated
+            objects_per_second: 0.0,           // Will be calculated
+            buckets_per_second: 0.0,           // Will be calculated
+            bucket_metrics: HashMap::new(),    // Will be populated by scanner
+            disk_metrics: HashMap::new(),      // Will be populated by scanner
+        }
+    }
+
+    /// Reset all metrics
+    pub fn reset(&self) {
+        self.objects_scanned.store(0, Ordering::Relaxed);
+        self.versions_scanned.store(0, Ordering::Relaxed);
+        self.directories_scanned.store(0, Ordering::Relaxed);
+        self.bucket_scans_started.store(0, Ordering::Relaxed);
+        self.bucket_scans_finished.store(0, Ordering::Relaxed);
+        self.objects_with_issues.store(0, Ordering::Relaxed);
+        self.heal_tasks_queued.store(0, Ordering::Relaxed);
+        self.heal_tasks_completed.store(0, Ordering::Relaxed);
+        self.heal_tasks_failed.store(0, Ordering::Relaxed);
+        self.current_cycle.store(0, Ordering::Relaxed);
+        self.total_cycles.store(0, Ordering::Relaxed);
+
+        info!("Scanner metrics reset");
+    }
+}
+
+impl Default for MetricsCollector {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_metrics_collector_creation() {
+        let collector = MetricsCollector::new();
+        let metrics = collector.get_metrics();
+        assert_eq!(metrics.objects_scanned, 0);
+        assert_eq!(metrics.versions_scanned, 0);
+    }
+
+    #[test]
+    fn test_metrics_increment() {
+        let collector = MetricsCollector::new();
+
+        collector.increment_objects_scanned(10);
+        collector.increment_versions_scanned(5);
+        collector.increment_objects_with_issues(2);
+
+        let metrics = collector.get_metrics();
+        assert_eq!(metrics.objects_scanned, 10);
+        assert_eq!(metrics.versions_scanned, 5);
+        assert_eq!(metrics.objects_with_issues, 2);
+    }
+
+    #[test]
+    fn test_metrics_reset() {
+        let collector = MetricsCollector::new();
+
+        collector.increment_objects_scanned(10);
+        collector.reset();
+
+        let metrics = collector.get_metrics();
+        assert_eq!(metrics.objects_scanned, 0);
+    }
+}

crates/ahm/src/scanner/mod.rs

@@ -0,0 +1,25 @@
+// 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.
+
+pub mod data_scanner;
+pub mod data_usage;
+pub mod histogram;
+pub mod metrics;
+
+// Re-export main types for convenience
+pub use data_scanner::Scanner;
+pub use data_usage::{
+    load_data_usage_from_backend, store_data_usage_in_backend, BucketTargetUsageInfo, BucketUsageInfo, DataUsageInfo,
+};
+pub use metrics::ScannerMetrics;

crates/ecstore/src/global.rs

@@ -30,6 +30,7 @@ use std::{
     time::SystemTime,
 };
 use tokio::sync::{OnceCell, RwLock};
+use tokio_util::sync::CancellationToken;
 use uuid::Uuid;
 
 pub const DISK_ASSUME_UNKNOWN_SIZE: u64 = 1 << 30;
@@ -66,6 +67,9 @@ pub static ref GLOBAL_NodeNamesHex: HashMap<String, ()> = HashMap::new();
 pub static ref GLOBAL_REGION: OnceLock<String> = OnceLock::new();
 }
 
+// Global cancellation token for background services (data scanner and auto heal)
+static GLOBAL_BACKGROUND_SERVICES_CANCEL_TOKEN: OnceLock<CancellationToken> = OnceLock::new();
+
 static GLOBAL_ACTIVE_CRED: OnceLock<Credentials> = OnceLock::new();
 
 pub fn init_global_action_cred(ak: Option<String>, sk: Option<String>) {
@@ -192,3 +196,27 @@ pub fn set_global_region(region: String) {
 pub fn get_global_region() -> Option<String> {
     GLOBAL_REGION.get().cloned()
 }
+
+/// Initialize the global background services cancellation token
+pub fn init_background_services_cancel_token(cancel_token: CancellationToken) -> Result<(), CancellationToken> {
+    GLOBAL_BACKGROUND_SERVICES_CANCEL_TOKEN.set(cancel_token)
+}
+
+/// Get the global background services cancellation token
+pub fn get_background_services_cancel_token() -> Option<&'static CancellationToken> {
+    GLOBAL_BACKGROUND_SERVICES_CANCEL_TOKEN.get()
+}
+
+/// Create and initialize the global background services cancellation token
+pub fn create_background_services_cancel_token() -> CancellationToken {
+    let cancel_token = CancellationToken::new();
+    init_background_services_cancel_token(cancel_token.clone()).expect("Background services cancel token already initialized");
+    cancel_token
+}
+
+/// Shutdown all background services gracefully
+pub fn shutdown_background_services() {
+    if let Some(cancel_token) = GLOBAL_BACKGROUND_SERVICES_CANCEL_TOKEN.get() {
+        cancel_token.cancel();
+    }
+}

crates/ecstore/src/heal/background_heal_ops.rs

@@ -24,6 +24,7 @@ use tokio::{
     },
     time::interval,
 };
+use tokio_util::sync::CancellationToken;
 use tracing::{error, info};
 use uuid::Uuid;
 
@@ -32,7 +33,7 @@ use super::{
     heal_ops::{HealSequence, new_bg_heal_sequence},
 };
 use crate::error::{Error, Result};
-use crate::global::GLOBAL_MRFState;
+use crate::global::{GLOBAL_MRFState, get_background_services_cancel_token};
 use crate::heal::error::ERR_RETRY_HEALING;
 use crate::heal::heal_commands::{HEAL_ITEM_BUCKET, HealScanMode};
 use crate::heal::heal_ops::{BG_HEALING_UUID, HealSource};
@@ -54,6 +55,13 @@ use crate::{
 pub static DEFAULT_MONITOR_NEW_DISK_INTERVAL: Duration = Duration::from_secs(10);
 
 pub async fn init_auto_heal() {
+    info!("Initializing auto heal background task");
+
+    let Some(cancel_token) = get_background_services_cancel_token() else {
+        error!("Background services cancel token not initialized");
+        return;
+    };
+
     init_background_healing().await;
     let v = env::var("_RUSTFS_AUTO_DRIVE_HEALING").unwrap_or("on".to_string());
     if v == "on" {
@@ -61,12 +69,16 @@ pub async fn init_auto_heal() {
         GLOBAL_BackgroundHealState
             .push_heal_local_disks(&get_local_disks_to_heal().await)
             .await;
-        spawn(async {
+
-            monitor_local_disks_and_heal().await;
+        let cancel_clone = cancel_token.clone();
+        spawn(async move {
+            monitor_local_disks_and_heal(cancel_clone).await;
         });
     }
-    spawn(async {
+
-        GLOBAL_MRFState.heal_routine().await;
+    let cancel_clone = cancel_token.clone();
+    spawn(async move {
+        GLOBAL_MRFState.heal_routine_with_cancel(cancel_clone).await;
     });
 }
 
@@ -108,50 +120,66 @@ pub async fn get_local_disks_to_heal() -> Vec<Endpoint> {
     disks_to_heal
 }
 
-async fn monitor_local_disks_and_heal() {
+async fn monitor_local_disks_and_heal(cancel_token: CancellationToken) {
+    info!("Auto heal monitor started");
     let mut interval = interval(DEFAULT_MONITOR_NEW_DISK_INTERVAL);
 
     loop {
-        interval.tick().await;
+        tokio::select! {
-        let heal_disks = GLOBAL_BackgroundHealState.get_heal_local_disk_endpoints().await;
+            _ = cancel_token.cancelled() => {
-        if heal_disks.is_empty() {
+                info!("Auto heal monitor received shutdown signal, exiting gracefully");
-            info!("heal local disks is empty");
+                break;
-            interval.reset();
+            }
-            continue;
+            _ = interval.tick() => {
-        }
+                let heal_disks = GLOBAL_BackgroundHealState.get_heal_local_disk_endpoints().await;
+                if heal_disks.is_empty() {
+                    info!("heal local disks is empty");
+                    interval.reset();
+                    continue;
+                }
 
-        info!("heal local disks: {:?}", heal_disks);
+                info!("heal local disks: {:?}", heal_disks);
 
-        let store = new_object_layer_fn().expect("errServerNotInitialized");
+                let store = new_object_layer_fn().expect("errServerNotInitialized");
-        if let (_result, Some(err)) = store.heal_format(false).await.expect("heal format failed") {
+                if let (_result, Some(err)) = store.heal_format(false).await.expect("heal format failed") {
-            error!("heal local disk format error: {}", err);
+                    error!("heal local disk format error: {}", err);
-            if err == Error::NoHealRequired {
+                    if err == Error::NoHealRequired {
-            } else {
+                    } else {
-                info!("heal format err: {}", err.to_string());
+                        info!("heal format err: {}", err.to_string());
+                        interval.reset();
+                        continue;
+                    }
+                }
+
+                let mut futures = Vec::new();
+                for disk in heal_disks.into_ref().iter() {
+                    let disk_clone = disk.clone();
+                    let cancel_clone = cancel_token.clone();
+                    futures.push(async move {
+                        let disk_for_cancel = disk_clone.clone();
+                        tokio::select! {
+                            _ = cancel_clone.cancelled() => {
+                                info!("Disk healing task cancelled for disk: {}", disk_for_cancel);
+                            }
+                            _ = async {
+                                GLOBAL_BackgroundHealState
+                                    .set_disk_healing_status(disk_clone.clone(), true)
+                                    .await;
+                                if heal_fresh_disk(&disk_clone).await.is_err() {
+                                    info!("heal_fresh_disk is err");
+                                    GLOBAL_BackgroundHealState
+                                        .set_disk_healing_status(disk_clone.clone(), false)
+                                        .await;
+                                }
+                                GLOBAL_BackgroundHealState.pop_heal_local_disks(&[disk_clone]).await;
+                            } => {}
+                        }
+                    });
+                }
+                let _ = join_all(futures).await;
                 interval.reset();
-                continue;
             }
         }
-
-        let mut futures = Vec::new();
-        for disk in heal_disks.into_ref().iter() {
-            let disk_clone = disk.clone();
-            futures.push(async move {
-                GLOBAL_BackgroundHealState
-                    .set_disk_healing_status(disk_clone.clone(), true)
-                    .await;
-                if heal_fresh_disk(&disk_clone).await.is_err() {
-                    info!("heal_fresh_disk is err");
-                    GLOBAL_BackgroundHealState
-                        .set_disk_healing_status(disk_clone.clone(), false)
-                        .await;
-                    return;
-                }
-                GLOBAL_BackgroundHealState.pop_heal_local_disks(&[disk_clone]).await;
-            });
-        }
-        let _ = join_all(futures).await;
-        interval.reset();
     }
 }
 

crates/ecstore/src/heal/data_scanner.rs

@@ -20,14 +20,13 @@ use std::{
     path::{Path, PathBuf},
     pin::Pin,
     sync::{
-        Arc, OnceLock,
+        Arc,
         atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering},
     },
     time::{Duration, SystemTime},
 };
 
 use time::{self, OffsetDateTime};
-use tokio_util::sync::CancellationToken;
 
 use super::{
     data_scanner_metric::{ScannerMetric, ScannerMetrics, globalScannerMetrics},
@@ -51,7 +50,7 @@ use crate::{
         metadata_sys,
     },
     event_notification::{EventArgs, send_event},
-    global::GLOBAL_LocalNodeName,
+    global::{GLOBAL_LocalNodeName, get_background_services_cancel_token},
     store_api::{ObjectOptions, ObjectToDelete, StorageAPI},
 };
 use crate::{
@@ -128,8 +127,6 @@ lazy_static! {
     pub static ref globalHealConfig: Arc<RwLock<Config>> = Arc::new(RwLock::new(Config::default()));
 }
 
-static GLOBAL_SCANNER_CANCEL_TOKEN: OnceLock<CancellationToken> = OnceLock::new();
-
 struct DynamicSleeper {
     factor: f64,
     max_sleep: Duration,
@@ -198,21 +195,18 @@ fn new_dynamic_sleeper(factor: f64, max_wait: Duration, is_scanner: bool) -> Dyn
 /// - Minimum sleep duration to avoid excessive CPU usage
 /// - Proper error handling and logging
 ///
-/// # Returns
-/// A CancellationToken that can be used to gracefully shutdown the scanner
-///
 /// # Architecture
 /// 1. Initialize with random seed for sleep intervals
 /// 2. Run scanner cycles in a loop
 /// 3. Use randomized sleep between cycles to avoid thundering herd
 /// 4. Ensure minimum sleep duration to prevent CPU thrashing
-pub async fn init_data_scanner() -> CancellationToken {
+pub async fn init_data_scanner() {
     info!("Initializing data scanner background task");
 
-    let cancel_token = CancellationToken::new();
+    let Some(cancel_token) = get_background_services_cancel_token() else {
-    GLOBAL_SCANNER_CANCEL_TOKEN
+        error!("Background services cancel token not initialized");
-        .set(cancel_token.clone())
+        return;
-        .expect("Scanner already initialized");
+    };
 
     let cancel_clone = cancel_token.clone();
     tokio::spawn(async move {
@@ -256,8 +250,6 @@ pub async fn init_data_scanner() -> CancellationToken {
 
         info!("Data scanner background task stopped gracefully");
     });
-
-    cancel_token
 }
 
 /// Run a single data scanner cycle
@@ -282,7 +274,7 @@ async fn run_data_scanner_cycle() {
     };
 
     // Check for cancellation before starting expensive operations
-    if let Some(token) = GLOBAL_SCANNER_CANCEL_TOKEN.get() {
+    if let Some(token) = get_background_services_cancel_token() {
         if token.is_cancelled() {
             debug!("Scanner cancelled before starting cycle");
             return;
@@ -397,9 +389,8 @@ async fn execute_namespace_scan(
     cycle: u64,
     scan_mode: HealScanMode,
 ) -> Result<()> {
-    let cancel_token = GLOBAL_SCANNER_CANCEL_TOKEN
+    let cancel_token =
-        .get()
+        get_background_services_cancel_token().ok_or_else(|| Error::other("Background services not initialized"))?;
-        .ok_or_else(|| Error::other("Scanner not initialized"))?;
 
     tokio::select! {
         result = store.ns_scanner(tx, cycle as usize, scan_mode) => {

crates/ecstore/src/heal/mrf.rs

@@ -25,7 +25,8 @@ use std::time::Duration;
 use tokio::sync::RwLock;
 use tokio::sync::mpsc::{Receiver, Sender};
 use tokio::time::sleep;
-use tracing::error;
+use tokio_util::sync::CancellationToken;
+use tracing::{error, info};
 use uuid::Uuid;
 
 pub const MRF_OPS_QUEUE_SIZE: u64 = 100000;
@@ -87,56 +88,96 @@ impl MRFState {
         let _ = self.tx.send(op).await;
     }
 
-    pub async fn heal_routine(&self) {
+    /// Enhanced heal routine with cancellation support
+    ///
+    /// This method implements the same healing logic as the original heal_routine,
+    /// but adds proper cancellation support via CancellationToken.
+    /// The core logic remains identical to maintain compatibility.
+    pub async fn heal_routine_with_cancel(&self, cancel_token: CancellationToken) {
+        info!("MRF heal routine started with cancellation support");
+
         loop {
-            // rx used only there,
+            tokio::select! {
-            if let Some(op) = self.rx.write().await.recv().await {
+                _ = cancel_token.cancelled() => {
-                if op.bucket == RUSTFS_META_BUCKET {
+                    info!("MRF heal routine received shutdown signal, exiting gracefully");
-                    for pattern in &*PATTERNS {
+                    break;
-                        if pattern.is_match(&op.object) {
+                }
-                            return;
+                op_result = async {
+                    let mut rx_guard = self.rx.write().await;
+                    rx_guard.recv().await
+                } => {
+                    if let Some(op) = op_result {
+                        // Special path filtering (original logic)
+                        if op.bucket == RUSTFS_META_BUCKET {
+                            for pattern in &*PATTERNS {
+                                if pattern.is_match(&op.object) {
+                                    continue; // Skip this operation, continue with next
+                                }
+                            }
                         }
-                    }
-                }
 
-                let now = Utc::now();
+                        // Network reconnection delay (original logic)
-                if now.sub(op.queued).num_seconds() < 1 {
+                        let now = Utc::now();
-                    sleep(Duration::from_secs(1)).await;
+                        if now.sub(op.queued).num_seconds() < 1 {
-                }
+                            tokio::select! {
+                                _ = cancel_token.cancelled() => {
+                                    info!("MRF heal routine cancelled during reconnection delay");
+                                    break;
+                                }
+                                _ = sleep(Duration::from_secs(1)) => {}
+                            }
+                        }
 
-                let scan_mode = if op.bitrot_scan { HEAL_DEEP_SCAN } else { HEAL_NORMAL_SCAN };
+                        // Core healing logic (original logic preserved)
-                if op.object.is_empty() {
+                        let scan_mode = if op.bitrot_scan { HEAL_DEEP_SCAN } else { HEAL_NORMAL_SCAN };
-                    if let Err(err) = heal_bucket(&op.bucket).await {
+
-                        error!("heal bucket failed, bucket: {}, err: {:?}", op.bucket, err);
+                        if op.object.is_empty() {
-                    }
+                            // Heal bucket (original logic)
-                } else if op.versions.is_empty() {
+                            if let Err(err) = heal_bucket(&op.bucket).await {
-                    if let Err(err) =
+                                error!("heal bucket failed, bucket: {}, err: {:?}", op.bucket, err);
-                        heal_object(&op.bucket, &op.object, &op.version_id.clone().unwrap_or_default(), scan_mode).await
+                            }
-                    {
+                        } else if op.versions.is_empty() {
-                        error!("heal object failed, bucket: {}, object: {}, err: {:?}", op.bucket, op.object, err);
+                            // Heal single object (original logic)
-                    }
-                } else {
-                    let vers = op.versions.len() / 16;
-                    if vers > 0 {
-                        for i in 0..vers {
-                            let start = i * 16;
-                            let end = start + 16;
                             if let Err(err) = heal_object(
                                 &op.bucket,
                                 &op.object,
-                                &Uuid::from_slice(&op.versions[start..end]).expect("").to_string(),
+                                &op.version_id.clone().unwrap_or_default(),
-                                scan_mode,
+                                scan_mode
-                            )
+                            ).await {
-                            .await
-                            {
                                 error!("heal object failed, bucket: {}, object: {}, err: {:?}", op.bucket, op.object, err);
                             }
+                        } else {
+                            // Heal multiple versions (original logic)
+                            let vers = op.versions.len() / 16;
+                            if vers > 0 {
+                                for i in 0..vers {
+                                    // Check for cancellation before each version
+                                    if cancel_token.is_cancelled() {
+                                        info!("MRF heal routine cancelled during version processing");
+                                        return;
+                                    }
+
+                                    let start = i * 16;
+                                    let end = start + 16;
+                                    if let Err(err) = heal_object(
+                                        &op.bucket,
+                                        &op.object,
+                                        &Uuid::from_slice(&op.versions[start..end]).expect("").to_string(),
+                                        scan_mode,
+                                    ).await {
+                                        error!("heal object failed, bucket: {}, object: {}, err: {:?}", op.bucket, op.object, err);
+                                    }
+                                }
+                            }
                         }
+                    } else {
+                        info!("MRF heal routine channel closed, exiting");
+                        break;
                     }
                 }
-            } else {
-                return;
             }
         }
+
+        info!("MRF heal routine stopped gracefully");
     }
 }

rustfs/Cargo.toml

@@ -34,6 +34,7 @@ path = "src/main.rs"
 workspace = true
 
 [dependencies]
+rustfs-ahm = { workspace = true }
 rustfs-zip = { workspace = true }
 rustfs-madmin = { workspace = true }
 rustfs-s3select-api = { workspace = true }

rustfs/src/admin/handlers.rs

@@ -31,6 +31,7 @@ use rustfs_ecstore::cmd::bucket_targets::{self, GLOBAL_Bucket_Target_Sys};
 use rustfs_ecstore::error::StorageError;
 use rustfs_ecstore::global::GLOBAL_ALlHealState;
 use rustfs_ecstore::global::get_global_action_cred;
+// use rustfs_ecstore::heal::data_usage::load_data_usage_from_backend;
 use rustfs_ecstore::heal::data_usage::load_data_usage_from_backend;
 use rustfs_ecstore::heal::heal_commands::HealOpts;
 use rustfs_ecstore::heal::heal_ops::new_heal_sequence;

rustfs/src/main.rs

@@ -42,18 +42,24 @@ use hyper_util::{
     service::TowerToHyperService,
 };
 use license::init_license;
+use rustfs_ahm::scanner::data_scanner::ScannerConfig;
+use rustfs_ahm::{Scanner, create_ahm_services_cancel_token, shutdown_ahm_services};
 use rustfs_common::globals::set_global_addr;
 use rustfs_config::{DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY, RUSTFS_TLS_CERT, RUSTFS_TLS_KEY};
 use rustfs_ecstore::bucket::metadata_sys::init_bucket_metadata_sys;
 use rustfs_ecstore::cmd::bucket_replication::init_bucket_replication_pool;
 use rustfs_ecstore::config as ecconfig;
 use rustfs_ecstore::config::GLOBAL_ConfigSys;
-use rustfs_ecstore::heal::background_heal_ops::init_auto_heal;
 use rustfs_ecstore::rpc::make_server;
 use rustfs_ecstore::store_api::BucketOptions;
 use rustfs_ecstore::{
-    StorageAPI, endpoints::EndpointServerPools, global::set_global_rustfs_port, heal::data_scanner::init_data_scanner,
+    StorageAPI,
-    notification_sys::new_global_notification_sys, set_global_endpoints, store::ECStore, store::init_local_disks,
+    endpoints::EndpointServerPools,
+    global::{set_global_rustfs_port, shutdown_background_services},
+    notification_sys::new_global_notification_sys,
+    set_global_endpoints,
+    store::ECStore,
+    store::init_local_disks,
     update_erasure_type,
 };
 use rustfs_iam::init_iam_sys;
@@ -442,13 +448,16 @@ async fn run(opt: config::Opt) -> Result<()> {
         Error::other(err)
     })?;
 
-    // init scanner
+    // init scanner and auto heal with unified cancellation token
-    let scanner_cancel_token = init_data_scanner().await;
+    // let _background_services_cancel_token = create_background_services_cancel_token();
-    // init auto heal
+    // init_data_scanner().await;
-    init_auto_heal().await;
+    // init_auto_heal().await;
+    let _ = create_ahm_services_cancel_token();
+    let scanner = Scanner::new(Some(ScannerConfig::default()));
+    scanner.start().await?;
+
     // init console configuration
     init_console_cfg(local_ip, server_port);
-
     print_server_info();
     init_bucket_replication_pool().await;
 
@@ -507,11 +516,11 @@ async fn run(opt: config::Opt) -> Result<()> {
     match wait_for_shutdown().await {
         #[cfg(unix)]
         ShutdownSignal::CtrlC | ShutdownSignal::Sigint | ShutdownSignal::Sigterm => {
-            handle_shutdown(&state_manager, &shutdown_tx, &scanner_cancel_token).await;
+            handle_shutdown(&state_manager, &shutdown_tx).await;
         }
         #[cfg(not(unix))]
         ShutdownSignal::CtrlC => {
-            handle_shutdown(&state_manager, &shutdown_tx, &scanner_cancel_token).await;
+            handle_shutdown(&state_manager, &shutdown_tx).await;
         }
     }
 
@@ -617,18 +626,18 @@ fn process_connection(
 }
 
 /// Handles the shutdown process of the server
-async fn handle_shutdown(
+async fn handle_shutdown(state_manager: &ServiceStateManager, shutdown_tx: &tokio::sync::broadcast::Sender<()>) {
-    state_manager: &ServiceStateManager,
-    shutdown_tx: &tokio::sync::broadcast::Sender<()>,
-    scanner_cancel_token: &tokio_util::sync::CancellationToken,
-) {
     info!("Shutdown signal received in main thread");
     // update the status to stopping first
     state_manager.update(ServiceState::Stopping);
 
-    // Stop data scanner gracefully
+    // Stop background services (data scanner and auto heal) gracefully
-    info!("Stopping data scanner...");
+    info!("Stopping background services (data scanner and auto heal)...");
-    scanner_cancel_token.cancel();
+    shutdown_background_services();
+
+    // Stop AHM services gracefully
+    info!("Stopping AHM services...");
+    shutdown_ahm_services();
 
     // Stop the notification system
     shutdown_event_notifier().await;