Merge commit 'b8f6aac1172a537248bdc7fc44da72e67cb3b9a2' into overtrue/scanner-ssd-baseline-20261024

crates/ecstore/src/set_disk/ops/object.rs

@@ -24505,7 +24505,20 @@ mod delete_objects_lock_gating_tests {
         for disk in &disks {
             disk.make_volume(bucket).await.expect("create fixture volume");
         }
-        put_plain_object(&set, bucket, object).await;
+        // This fixture inspects a specific drive and then takes it offline.
+        // A quorum ACK alone does not mean that drive's rename has completed.
+        let mut reader = PutObjReader::from_vec(vec![3u8; 1024]);
+        set.put_object(
+            bucket,
+            object,
+            &mut reader,
+            &ObjectOptions {
+                write_completion: crate::object_api::WriteCompletion::TailDrained,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("fixture rename fan-out should finish before injecting a drive fault");
         let disk_info = disks[0]
             .read_version(bucket, bucket, object, "", &crate::disk::ReadOptions::default())
             .await

crates/heal/src/heal/manager.rs

@@ -506,7 +506,27 @@ impl HealAdmissionTelemetry {
             .max(duration_micros_saturated(observation.start_duration));
         self.max_lock_phase_micros = self
             .max_lock_phase_micros
-            .max(duration_micros_saturated(observation.lock_phase));
+            .max(duration_micros_saturated(observation.lock_timing.phase));
+    }
+}
+
+#[derive(Debug, Clone, Copy)]
+struct HealAdmissionLockTiming {
+    phase: Duration,
+    wait: Duration,
+    hold: Duration,
+}
+
+impl HealAdmissionLockTiming {
+    // Called after the last outer admission guard is released. forceStart's
+    // cancellation gap stays in phase, outside the accumulated guard intervals.
+    fn finish(phase_started: Instant, acquired: Instant, wait: Duration, previous_hold: Duration) -> Self {
+        let released = Instant::now();
+        Self {
+            phase: released.duration_since(phase_started),
+            wait,
+            hold: previous_hold + released.duration_since(acquired),
+        }
     }
 }
 
@@ -518,7 +538,7 @@ struct HealAdmissionObservation {
     force_start: bool,
     displaced: bool,
     start_duration: Duration,
-    lock_phase: Duration,
+    lock_timing: HealAdmissionLockTiming,
 }
 
 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
@@ -1217,7 +1237,10 @@ impl HealManager {
             "context" => context,
             "force_start" => force_start
         )
-        .record(observation.lock_phase.as_secs_f64());
+        .record(observation.lock_timing.phase.as_secs_f64());
+        // Static keys keep the added observations bounded and avoid per-admission label allocation.
+        histogram!("rustfs_heal_admission_active_lock_wait_seconds").record(observation.lock_timing.wait.as_secs_f64());
+        histogram!("rustfs_heal_admission_active_lock_hold_seconds").record(observation.lock_timing.hold.as_secs_f64());
         lock_admission_telemetry(&self.admission_telemetry).record(observation);
     }
 
@@ -2039,6 +2062,9 @@ impl HealManager {
         // active -> retrying transitions can slip between duplicate checks.
         let lock_phase_start = Instant::now();
         let mut active_heals = self.active_heals.lock().await;
+        let mut active_lock_acquired = Instant::now();
+        let mut active_lock_wait = active_lock_acquired.duration_since(lock_phase_start);
+        let mut active_lock_hold = Duration::ZERO;
         if self.cancel_token.is_cancelled() {
             return Err(Error::Other("Heal manager is stopping".to_string()));
         }
@@ -2088,7 +2114,8 @@ impl HealManager {
             drop(retrying_heals);
             drop(queue);
             drop(active_heals);
-            let lock_phase = lock_phase_start.elapsed();
+            let lock_timing =
+                HealAdmissionLockTiming::finish(lock_phase_start, active_lock_acquired, active_lock_wait, active_lock_hold);
             Self::record_admission_metric(request.source, admission, "duplicate");
             self.record_admission_observation(HealAdmissionObservation {
                 source,
@@ -2097,7 +2124,7 @@ impl HealManager {
                 force_start,
                 displaced: false,
                 start_duration: admission_start.elapsed(),
-                lock_phase,
+                lock_timing,
             });
             debug!(
                 target: "rustfs::heal::manager",
@@ -2138,6 +2165,7 @@ impl HealManager {
             drop(retrying_heals);
             drop(queue);
             drop(active_heals);
+            active_lock_hold += active_lock_acquired.elapsed();
             for task_id in &overlapping {
                 match self.cancel_task(task_id).await {
                     Ok(_) => info!(
@@ -2155,7 +2183,10 @@ impl HealManager {
                 }
             }
             durable_owners.retain(|owner| !overlapping.contains(&owner.id));
+            let reacquire_start = Instant::now();
             active_heals = self.active_heals.lock().await;
+            active_lock_acquired = Instant::now();
+            active_lock_wait += active_lock_acquired.duration_since(reacquire_start);
             queue = self.heal_queue.lock().await;
             retrying_heals = self.retrying_heals.lock().await;
         }
@@ -2238,7 +2269,8 @@ impl HealManager {
             drop(retrying_heals);
             drop(queue);
             drop(active_heals);
-            let lock_phase = lock_phase_start.elapsed();
+            let lock_timing =
+                HealAdmissionLockTiming::finish(lock_phase_start, active_lock_acquired, active_lock_wait, active_lock_hold);
             Self::record_admission_metric(request.source, admission, "duplicate");
             self.record_admission_observation(HealAdmissionObservation {
                 source,
@@ -2247,7 +2279,7 @@ impl HealManager {
                 force_start,
                 displaced: false,
                 start_duration: admission_start.elapsed(),
-                lock_phase,
+                lock_timing,
             });
 
             match admission {
@@ -2294,7 +2326,8 @@ impl HealManager {
             drop(retrying_heals);
             drop(queue);
             drop(active_heals);
-            let lock_phase = lock_phase_start.elapsed();
+            let lock_timing =
+                HealAdmissionLockTiming::finish(lock_phase_start, active_lock_acquired, active_lock_wait, active_lock_hold);
             Self::record_admission_metric(request.source, HealAdmissionResult::Dropped(reason), "overlap_rejected");
             self.record_admission_observation(HealAdmissionObservation {
                 source,
@@ -2303,7 +2336,7 @@ impl HealManager {
                 force_start,
                 displaced: false,
                 start_duration: admission_start.elapsed(),
-                lock_phase,
+                lock_timing,
             });
             warn!(
                 target: "rustfs::heal::manager",
@@ -2332,7 +2365,8 @@ impl HealManager {
                 drop(retrying_heals);
                 drop(queue);
                 drop(active_heals);
-                let lock_phase = lock_phase_start.elapsed();
+                let lock_timing =
+                    HealAdmissionLockTiming::finish(lock_phase_start, active_lock_acquired, active_lock_wait, active_lock_hold);
                 let admission = HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped);
                 Self::record_admission_metric(source, admission, "disposition_cache");
                 self.record_admission_observation(HealAdmissionObservation {
@@ -2342,7 +2376,7 @@ impl HealManager {
                     force_start,
                     displaced: false,
                     start_duration: admission_start.elapsed(),
-                    lock_phase,
+                    lock_timing,
                 });
                 debug!(
                     target: "rustfs::heal::manager",
@@ -2412,7 +2446,8 @@ impl HealManager {
         drop(retrying_heals);
         drop(queue);
         drop(active_heals);
-        let lock_phase = lock_phase_start.elapsed();
+        let lock_timing =
+            HealAdmissionLockTiming::finish(lock_phase_start, active_lock_acquired, active_lock_wait, active_lock_hold);
         let displaced = displaced_terminal.is_some();
 
         if let (Some(displaced_task_id), Some(displaced_terminal)) = (displaced_task_id, displaced_terminal) {
@@ -2434,7 +2469,7 @@ impl HealManager {
             force_start,
             displaced,
             start_duration: admission_start.elapsed(),
-            lock_phase,
+            lock_timing,
         });
 
         Ok(HealAdmissionReceipt {

crates/heal/src/heal/manager/tests.rs

@@ -27,6 +27,7 @@ use std::sync::Mutex as StdMutex;
 use tempfile::TempDir;
 
 pub(super) mod admin_overlap;
+mod admission_timing;
 mod root_recovery;
 use uuid::Uuid;
 mod auto_scan;
@@ -4177,7 +4178,7 @@ async fn admission_snapshot_tracks_start_duplicate_force_start_and_displacement(
     assert_eq!(admission.force_start, 1);
     assert!(
         admission.max_lock_phase_micros > 0,
-        "snapshot should expose a measurable queue/admission lock phase for p95-style external aggregation"
+        "snapshot should expose a measurable maximum queue/admission lock phase"
     );
 }
 

crates/heal/src/heal/manager/tests/admission_timing.rs

@@ -0,0 +1,213 @@
+// 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 super::*;
+use std::future::Future;
+
+const PHASE: &str = "rustfs_heal_admission_lock_phase_seconds";
+const WAIT: &str = "rustfs_heal_admission_active_lock_wait_seconds";
+const HOLD: &str = "rustfs_heal_admission_active_lock_hold_seconds";
+
+struct AdmissionRecorder {
+    manager: Arc<HealManager>,
+    samples: Arc<StdMutex<Vec<(&'static str, f64)>>>,
+}
+
+impl AdmissionRecorder {
+    fn new(manager: &Arc<HealManager>) -> Self {
+        Self {
+            manager: Arc::clone(manager),
+            samples: Arc::default(),
+        }
+    }
+}
+
+struct AdmissionHistogram {
+    name: &'static str,
+    manager: Arc<HealManager>,
+    samples: Arc<StdMutex<Vec<(&'static str, f64)>>>,
+}
+
+impl metrics::HistogramFn for AdmissionHistogram {
+    fn record(&self, value: f64) {
+        assert!(self.manager.active_heals.try_lock().is_ok(), "recording must follow active lock release");
+        assert!(self.manager.heal_queue.try_lock().is_ok(), "recording must follow queue lock release");
+        assert!(self.manager.retrying_heals.try_lock().is_ok(), "recording must follow retry lock release");
+        self.samples.lock().expect("admission samples").push((self.name, value));
+    }
+}
+
+impl metrics::Recorder for AdmissionRecorder {
+    fn describe_counter(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
+    fn describe_gauge(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
+    fn describe_histogram(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
+    fn register_counter(&self, _: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Counter {
+        metrics::Counter::noop()
+    }
+    fn register_gauge(&self, _: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Gauge {
+        metrics::Gauge::noop()
+    }
+    fn register_histogram(&self, key: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Histogram {
+        let name = match key.name() {
+            PHASE => PHASE,
+            WAIT => WAIT,
+            HOLD => HOLD,
+            _ => return metrics::Histogram::noop(),
+        };
+        if name != PHASE {
+            assert_eq!(key.labels().count(), 0, "new lock histograms have static, unlabelled keys");
+        }
+        metrics::Histogram::from_arc(Arc::new(AdmissionHistogram {
+            name,
+            manager: Arc::clone(&self.manager),
+            samples: Arc::clone(&self.samples),
+        }))
+    }
+}
+
+async fn recorded<F: Future>(recorder: &AdmissionRecorder, future: F) -> F::Output {
+    tokio::pin!(future);
+    futures::future::poll_fn(|cx| metrics::with_local_recorder(recorder, || future.as_mut().poll(cx))).await
+}
+
+fn timing_samples(recorder: &AdmissionRecorder) -> (f64, f64, f64) {
+    let samples = recorder.samples.lock().expect("admission samples");
+    let sample = |name| {
+        let values: Vec<_> = samples
+            .iter()
+            .filter(|(key, _)| *key == name)
+            .map(|(_, value)| *value)
+            .collect();
+        assert_eq!(values.len(), 1, "exactly one {name} observation per receipt");
+        values[0]
+    };
+    (sample(PHASE), sample(WAIT), sample(HOLD))
+}
+
+fn assert_partition(recorder: &AdmissionRecorder, minimum_wait: Duration, minimum_hold: Duration) {
+    let (phase, wait, hold) = timing_samples(recorder);
+    assert!(wait >= minimum_wait.as_secs_f64(), "initial active-lock contention belongs to wait");
+    assert!(hold >= minimum_hold.as_secs_f64(), "nested-lock contention belongs to active-lock hold");
+    assert!(
+        (phase - (wait + hold)).abs() <= phase * f64::EPSILON * 4.0,
+        "wait and hold must partition the existing phase without double-counting contention"
+    );
+}
+
+#[tokio::test]
+async fn admission_timing_separates_initial_active_lock_wait() {
+    let manager = Arc::new(HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None));
+    let recorder = AdmissionRecorder::new(&manager);
+    let active = manager.active_heals.lock().await;
+    let admission = recorded(&recorder, manager.submit_heal_request(internal_prefix_request("bucket-a", "logs/")));
+    tokio::pin!(admission);
+    assert!(futures::poll!(&mut admission).is_pending(), "admission must reach the held active lock");
+    let stalled = Instant::now();
+    tokio::task::yield_now().await;
+    let minimum_wait = stalled.elapsed();
+    drop(active);
+    assert_eq!(admission.await.expect("admit after active lock release"), HealAdmissionResult::Accepted);
+    assert!(manager.active_heals.try_lock().is_ok(), "receipt must release active ownership");
+    assert_partition(&recorder, minimum_wait, Duration::ZERO);
+}
+
+#[tokio::test]
+async fn admission_timing_includes_queue_wait_while_active_lock_is_held() {
+    let manager = Arc::new(HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None));
+    let recorder = AdmissionRecorder::new(&manager);
+    let queue = manager.heal_queue.lock().await;
+    let admission = recorded(&recorder, manager.submit_heal_request(internal_prefix_request("bucket-a", "logs/")));
+    tokio::pin!(admission);
+    assert!(futures::poll!(&mut admission).is_pending(), "admission must reach the held queue lock");
+    assert!(manager.active_heals.try_lock().is_err(), "admission owns active while waiting for queue");
+    let stalled = Instant::now();
+    tokio::task::yield_now().await;
+    let minimum_hold = stalled.elapsed();
+    drop(queue);
+    assert_eq!(admission.await.expect("admit after queue lock release"), HealAdmissionResult::Accepted);
+    assert!(manager.active_heals.try_lock().is_ok(), "receipt must release active ownership");
+    assert_partition(&recorder, Duration::ZERO, minimum_hold);
+}
+
+#[tokio::test]
+async fn admission_timing_force_start_excludes_cancel_gap_and_sums_reacquisition() {
+    let manager = Arc::new(manager_with_policy(HealOverlapPolicy::Merge));
+    let old_id = insert_active_task(&manager, admin_prefix_request("bucket-a", "logs/")).await;
+    let recorder = AdmissionRecorder::new(&manager);
+    let mut replacement = admin_prefix_request("bucket-a", "logs/");
+    replacement.force_start = true;
+    let queue = manager.heal_queue.lock().await;
+    let admission = recorded(&recorder, manager.submit_heal_request_with_receipt(replacement));
+    tokio::pin!(admission);
+    assert!(futures::poll!(&mut admission).is_pending());
+    assert!(
+        manager.active_heals.try_lock().is_err(),
+        "initial interval owns active while waiting for queue"
+    );
+    let first_held_at = Instant::now();
+    let aliases = manager.task_aliases.lock().await;
+    let first_hold = first_held_at.elapsed();
+    drop(queue);
+    assert!(
+        futures::poll!(&mut admission).is_pending(),
+        "cancellation must wait for canonical alias lookup"
+    );
+    assert!(manager.active_heals.try_lock().is_ok());
+    assert!(manager.heal_queue.try_lock().is_ok());
+    assert!(manager.retrying_heals.try_lock().is_ok());
+    let unlocked_at = Instant::now();
+
+    // Queue cancellation first, then a real competing owner. Tokio's FIFO
+    // mutex hands that owner the lock before forceStart can reacquire it.
+    let active = manager.active_heals.lock().await;
+    let cancellation_gap = unlocked_at.elapsed();
+    drop(aliases);
+    assert!(futures::poll!(&mut admission).is_pending(), "cancellation is queued for active");
+    let next_active = manager.active_heals.lock();
+    tokio::pin!(next_active);
+    assert!(futures::poll!(&mut next_active).is_pending());
+    drop(active);
+    assert!(
+        futures::poll!(&mut admission).is_pending(),
+        "forceStart must queue behind the competing owner"
+    );
+    let active = next_active.await;
+    assert!(
+        manager.completed_heals.lock().await.contains_key(&old_id),
+        "cancellation finished before reacquisition"
+    );
+    let reacquiring_at = Instant::now();
+    let queue = manager.heal_queue.lock().await;
+    assert!(manager.retrying_heals.try_lock().is_ok());
+    let second_wait = reacquiring_at.elapsed();
+    drop(active);
+    assert!(futures::poll!(&mut admission).is_pending(), "reacquired active waits for queue");
+    assert!(manager.active_heals.try_lock().is_err());
+    let second_held_at = Instant::now();
+    assert!(recorder.samples.lock().expect("samples").is_empty());
+    let second_hold = second_held_at.elapsed();
+    drop(queue);
+    assert!(admission.await.expect("force-start receipt").result.is_admitted());
+
+    let (phase, wait, hold) = timing_samples(&recorder);
+    assert!(wait >= second_wait.as_secs_f64(), "second acquisition contention must be counted as wait");
+    assert!(
+        hold >= (first_hold + second_hold).as_secs_f64(),
+        "both actual held intervals must be counted"
+    );
+    assert!(
+        phase - (wait + hold) >= cancellation_gap.as_secs_f64(),
+        "the unlocked cancellation gap belongs only to the unchanged total phase"
+    );
+}

crates/scanner/src/scanner_folder.rs

@@ -760,6 +760,7 @@ pub struct FolderScanner {
     checkpoint_depth: usize,
     last_update: SystemTime,
     checkpoint_objects: u64,
+    object_metadata_visits: metrics::Counter,
     last_checkpoint_objects: u64,
     last_checkpoint_at: Instant,
     last_checkpoint_build_time: Duration,
@@ -2026,6 +2027,7 @@ impl FolderScanner {
                 };
                 wait_for_scanner_io(&ctx, self.sleeper.admit_disk_io(1, budgeted_bytes)).await?;
 
+                self.object_metadata_visits.increment(1);
                 let sz = match self
                     .local_disk
                     .get_size_with_tier_names(item.clone(), &self.tier_registry.names)
@@ -3002,6 +3004,11 @@ pub(crate) async fn scan_data_folder_scoped(
         global_metrics().record_scanner_checkpoint_used();
     }
     let bucket_name = cache.info.name.clone();
+    let object_metadata_visits = metrics::counter!(
+        "rustfs_scanner_object_metadata_visits_total",
+        "bucket" => bucket_name.clone(),
+        "disk" => base_path.clone()
+    );
     let next_cycle = cache.info.next_cycle;
     let resume_position_on_entry = cache.has_scan_position();
     let had_scan_checkpoint = cache.info.scan_checkpoint.is_some();
@@ -3042,6 +3049,7 @@ pub(crate) async fn scan_data_folder_scoped(
         checkpoint_depth: 0,
         last_update: SystemTime::UNIX_EPOCH,
         checkpoint_objects: 0,
+        object_metadata_visits,
         last_checkpoint_objects: 0,
         last_checkpoint_at: Instant::now(),
         last_checkpoint_build_time: Duration::ZERO,

crates/scanner/src/scanner_folder/tests.rs

@@ -29,6 +29,7 @@ use std::sync::Mutex;
 pub(super) mod checkpoint_fixture;
 pub(super) mod enumeration_restart;
 mod incremental_enumeration;
+mod metadata_visits;
 
 /// Reset the process-global alert cooldown map; test-only.
 fn reset_alert_cooldowns() {
@@ -347,6 +348,7 @@ async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) {
         checkpoint_depth: 0,
         last_update: SystemTime::UNIX_EPOCH,
         checkpoint_objects: 0,
+        object_metadata_visits: metrics::Counter::noop(),
         last_checkpoint_objects: 0,
         last_checkpoint_at: Instant::now(),
         last_checkpoint_build_time: Duration::ZERO,
@@ -2356,6 +2358,7 @@ async fn scan_folder_charges_each_enumerated_entry_to_the_disk_budget() {
 #[serial]
 async fn scan_folder_budgets_payload_entries_without_metadata_io() {
     let (mut scanner, temp_dir) = build_test_scanner().await;
+    let visits = metadata_visits::observe(&mut scanner);
     let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
     let bucket_dir = temp_dir.join("bucket");
     tokio::fs::create_dir_all(&bucket_dir).await.expect("create bucket");
@@ -2384,6 +2387,7 @@ async fn scan_folder_budgets_payload_entries_without_metadata_io() {
         )
         .await
         .expect("scan payload entries across a directory prefetch boundary");
+    assert_eq!(visits.load(Ordering::Relaxed), 0, "payload enumeration is not a metadata visit");
 
     let status = crate::sleeper::scanner_disk_io_budget_status()
         .into_iter()
@@ -2755,6 +2759,7 @@ async fn test_scan_folder_corrupt_xl_meta_stops_erasure_data_dir_descent() {
     let _subscriber_guard = tracing::subscriber::set_default(subscriber);
 
     let (mut scanner, temp_dir) = build_test_scanner().await;
+    let visits = metadata_visits::observe(&mut scanner);
     // Canonicalize for the "drive" field comparison (scanner resolves symlinks).
     let canonical_temp_dir = std::fs::canonicalize(&temp_dir).unwrap_or_else(|_| temp_dir.clone());
     let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
@@ -2797,6 +2802,7 @@ async fn test_scan_folder_corrupt_xl_meta_stops_erasure_data_dir_descent() {
         .expect("failed metadata must not make scanner descend into erasure data directories");
 
     assert_eq!(into.failed_objects, 1);
+    assert_eq!(visits.load(Ordering::Relaxed), 1, "failed metadata reads still consume a visit");
     assert!(
         scanner
             .new_cache
@@ -3164,6 +3170,7 @@ async fn scoped_root_scan_reuses_clean_top_level_entries_and_rescans_dirty_entri
 
 async fn scan_hot_cold_segment_fixture(scoped: bool) -> (DataUsageEntry, Vec<String>) {
     let (mut scanner, temp_dir) = build_test_scanner().await;
+    let visits = metadata_visits::observe(&mut scanner);
     let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
     write_test_object_metadata(&temp_dir, "bucket", "cold/object").await;
     write_test_object_metadata(&temp_dir, "bucket", "hot/object").await;
@@ -3205,6 +3212,11 @@ async fn scan_hot_cold_segment_fixture(scoped: bool) -> (DataUsageEntry, Vec<Str
         .scan_folder(CancellationToken::new(), folder, &mut root)
         .await
         .expect("segment fixture scan should finish");
+    assert_eq!(
+        visits.load(Ordering::Relaxed),
+        if scoped { 1 } else { 2 },
+        "reused cold cache entries must not count as metadata visits"
+    );
     let root = scanner
         .new_cache
         .size_recursive("bucket")

crates/scanner/src/scanner_folder/tests/metadata_visits.rs

@@ -0,0 +1,88 @@
+// Copyright 2026 RustFS Team
+// Licensed under the Apache License, Version 2.0.
+
+use super::*;
+use std::sync::atomic::AtomicU64;
+
+const VISITS: &str = "rustfs_scanner_object_metadata_visits_total";
+
+pub(super) fn observe(scanner: &mut FolderScanner) -> Arc<AtomicU64> {
+    let count = Arc::new(AtomicU64::new(0));
+    scanner.object_metadata_visits = metrics::Counter::from_arc(count.clone());
+    count
+}
+
+#[derive(Default)]
+struct VisitRecorder(Mutex<HashMap<metrics::Key, Arc<AtomicU64>>>);
+
+impl metrics::Recorder for VisitRecorder {
+    fn describe_counter(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
+    fn describe_gauge(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
+    fn describe_histogram(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
+    fn register_counter(&self, key: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Counter {
+        if key.name() != VISITS {
+            return metrics::Counter::noop();
+        }
+        metrics::Counter::from_arc(self.0.lock().expect("visit recorder").entry(key.clone()).or_default().clone())
+    }
+    fn register_gauge(&self, _: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Gauge {
+        metrics::Gauge::noop()
+    }
+    fn register_histogram(&self, _: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Histogram {
+        metrics::Histogram::noop()
+    }
+}
+
+#[tokio::test]
+#[serial]
+async fn real_scan_registers_bucket_disk_counter_and_accumulates_visits() {
+    let (scanner, temp_dir) = build_test_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(temp_dir.clone()),
+    };
+    write_test_object_metadata(&temp_dir, "bucket", "first").await;
+    write_test_object_metadata(&temp_dir, "bucket", "second").await;
+    let recorder = VisitRecorder::default();
+    let disk = scanner.local_disk.clone();
+    for (cancelled, expected) in [(true, 0), (false, 2), (false, 4)] {
+        let parent = CancellationToken::new();
+        if cancelled {
+            parent.cancel();
+        }
+        let budget = ScannerCycleBudget::new(&parent, Default::default());
+        let cache = DataUsageCache {
+            info: crate::data_usage_define::DataUsageCacheInfo {
+                name: "bucket".to_string(),
+                ..Default::default()
+            },
+            ..Default::default()
+        };
+        let scan = scan_data_folder(
+            parent,
+            budget,
+            vec![disk.clone()],
+            disk.clone(),
+            cache,
+            None,
+            HealScanMode::Normal,
+            DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest),
+        );
+        let mut scan = std::pin::pin!(scan);
+        let result = std::future::poll_fn(|cx| metrics::with_local_recorder(&recorder, || scan.as_mut().poll(cx))).await;
+        if cancelled {
+            assert!(result.is_err(), "cancelled scan must still fail");
+        } else {
+            result.expect("real bucket scanner should finish");
+        }
+        let rows = recorder.0.lock().expect("visit recorder");
+        let key = metrics::Key::from_parts(
+            VISITS,
+            vec![
+                metrics::Label::new("bucket", "bucket"),
+                metrics::Label::new("disk", disk.path().to_string_lossy().into_owned()),
+            ],
+        );
+        assert_eq!(rows.len(), 1, "only bucket/disk identity, no object labels");
+        assert_eq!(rows[&key].load(Ordering::Relaxed), expected);
+    }
+}

docs/operations/scanner-benchmark-runbook.md

@@ -215,6 +215,26 @@ and physical disk writes are outside this counter. Keep the same definition on
 both builds; retain disk counters separately. Neither counter measures cache
 clone allocations, and an absent series is not a measured zero.
 
+`walk_objects` can use the aligned window delta of
+`rustfs_scanner_object_metadata_visits_total`, summed across the declared
+`bucket` and `disk` identities. Each visit starts after scanner I/O admission,
+immediately before reading an `xl.meta` entry. Failed or cancelled reads still
+count; payload entries, cached summaries, and paths skipped before admission do
+not. Re-reading an object counts again, including on another drive. This measures
+scanner metadata visits, not unique S3 objects, versions, successful repairs, or
+physical disk operations. The counter handle is registered once per bucket/drive
+walker and reused without per-object label allocation; recording still has a cost.
+
+For a workload with distinct hot and cold buckets, `cold_walk_objects` selects
+only the cold bucket identities declared in the independent workload manifest.
+Do not infer coldness from the observed counts or multiply bucket-walk counts by
+cached object totals. This metric cannot separate hot and cold prefixes within
+one bucket; that attribution remains unverified. Warm up the required series,
+retain node/drive identities, reject resets or missing window boundaries, and
+use the same visit definition on both builds. Bucket labels follow the existing
+bucket/drive scanner metrics and have bucket-scale cardinality; no key or version
+is a label. An absent or dropped series is not zero.
+
 Each comparison records a `w22` section with clone, encode, and save bytes per
 walked object, clone/encode and save/encode byte ratios, and candidate changes.
 These are traffic amplification indicators, not allocation attribution or an

docs/operations/scanner-runtime-controls.md

@@ -433,6 +433,18 @@ Example risk-acceptance body (substitute values from the node-local listing):
 
 A process restart restores lifecycle state only when the checksummed sidecar matches the committed checkpoint. Otherwise the durable intent remains visible and generation-unknown; no record is deleted or certified, and other durable intents on that node replay normally. For source-bound unversioned objects, restore expected content from a trusted canonical source with protected shard integrity enabled, then request a targeted recheck so the exact intent can obtain a receipt. For a legacy record without a source generation, the operator must first establish which bucket generation the responsibility belongs to; if that cannot be established, keep it generation-unknown or explicitly accept the risk. For a versioned object, a protected rewrite creates a new version and does not resolve a responsibility for the old version; preserve the source and only retire that exact version when the intended data is backed up and an authoritative absence proof is appropriate. A successful receipt may discharge only the matching responsibility; an unverified result remains retained and becomes held again. If no trusted copy or expected checksum is available, keep the responsibility held or explicitly accept the risk with the node-local admin operation. The protected-copy migration in the [shard-integrity audit workflow](shard-integrity-audit.md) writes a new key and preserves its source; completing that migration alone does not prove or discharge an intent for the original key. Never remove MRF journal files manually.
 
+### Admission lock timing
+
+The OTLP exporter records these node-local histograms for completed manager admissions that return a receipt, including merged and rejected work. Cancellation and error exits without a receipt do not emit these observations.
+
+| Metric | Meaning |
+|---|---|
+| `rustfs_heal_admission_lock_phase_seconds` | Existing total from immediately before waiting for `active_heals` until immediately after releasing it, with the existing source/result/context labels. |
+| `rustfs_heal_admission_active_lock_wait_seconds` | Sum of waits to acquire the outermost admission `active_heals` guard, including forceStart reacquisition. |
+| `rustfs_heal_admission_active_lock_hold_seconds` | Sum of the outer admission guard's actual held intervals, including waits for the nested queue and retrying mutexes. |
+
+The two new histograms aggregate all observed admission sources and results without per-request labels. An ordinary uninterrupted admission partitions the phase into wait and hold. Admin forceStart releases these guards while cancelling overlaps, then reacquires them: the cancellation gap is included only in the original total phase, and the second acquisition contributes to wait. Cancellation subroutine work is outside these outer admission guard intervals. Recording happens after all three guards are released. They measure manager admission ownership, not namespace/object-lock waiting during repair. The status field `maxLockPhaseMicros` remains a manager-lifetime maximum, not a p95 sample. Keep same-window histogram counts and bucket resolution when deriving percentiles; missing observations do not mean zero latency. These metrics do not establish performance acceptance by themselves.
+
 ## Heal runtime controls
 
 Heal knobs are environment-only and read by `HealConfig::default` (`crates/heal/src/heal/manager.rs`), the MRF queue (`crates/heal/src/heal/mrf_queue.rs`), or the erasure-set healer (`crates/heal/src/heal/erasure_healer.rs`). The admin `heal` config subsystem accepts only `bitrot_cycle` (`HEAL_KEYS`), which is documented in the scanner table above. Constants live in `crates/config/src/constants/heal.rs` unless another file is named.