diff --git a/crates/common/src/metrics.rs b/crates/common/src/metrics.rs
index 00521b55b..9a6198d35 100644
--- a/crates/common/src/metrics.rs
+++ b/crates/common/src/metrics.rs
@@ -781,6 +781,19 @@ struct ScannerBucketDriveResultValue {
     last_seen: u64,
 }
 
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+struct ScannerActiveBucketDriveKey {
+    source: String,
+    bucket: String,
+    drive: String,
+}
+
+#[derive(Clone, Copy, Debug)]
+struct ScannerActiveBucketDriveValue {
+    count: u64,
+    started_at: Timestamp,
+}
+
 // ---------------------------------------------------------------------------
 // Metrics
 // ---------------------------------------------------------------------------
@@ -813,6 +826,7 @@ pub struct Metrics {
     scanner_set_scans_active: AtomicU64,
     scanner_disk_bucket_scan_states: Mutex<HashMap<ScannerDiskBucketScanKey, ScannerDiskBucketScanState>>,
     scanner_bucket_drive_results: Mutex<ScannerBucketDriveResults>,
+    scanner_active_bucket_drive_scans: Mutex<HashMap<ScannerActiveBucketDriveKey, ScannerActiveBucketDriveValue>>,
     scanner_bucket_drive_result_clock: AtomicU64,
     current_scan_cycle_bucket_drive_results_start: Mutex<HashMap<ScannerBucketDriveResultKey, u64>>,
     last_scan_cycle_bucket_drive_results: Mutex<Vec<ScannerBucketDriveResultSnapshot>>,
@@ -1045,6 +1059,15 @@ pub struct ScannerBucketDriveResultSnapshot {
     pub count: u64,
 }
 
+#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
+pub struct ScannerActiveBucketDriveSnapshot {
+    pub source: String,
+    pub bucket: String,
+    pub drive: String,
+    pub count: u64,
+    pub age_seconds: u64,
+}
+
 #[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
 pub struct ScannerReplicationRepairSnapshot {
     pub source: String,
@@ -1387,6 +1410,8 @@ pub struct ScannerRuntimeDetailsReport {
     pub current_cycle_bucket_drive_results: Vec<ScannerBucketDriveResultSnapshot>,
     #[serde(default)]
     pub last_cycle_bucket_drive_results: Vec<ScannerBucketDriveResultSnapshot>,
+    #[serde(default)]
+    pub active_bucket_drive_scans: Vec<ScannerActiveBucketDriveSnapshot>,
 }
 
 impl CurrentCycle {
@@ -1746,7 +1771,7 @@ pub fn emit_scan_cycle_deferred(duration: Duration) {
     metrics::counter!(OTEL_SCANNER_CYCLES, "result" => SCAN_CYCLE_RESULT_DEFERRED_LABEL).increment(1);
 }
 
-pub fn emit_scan_bucket_drive_complete(success: bool, bucket: &str, disk: &str, duration: Duration) {
+pub fn emit_scan_bucket_drive_complete(_source: ScannerWorkSource, success: bool, bucket: &str, disk: &str, duration: Duration) {
     let result = if success { "success" } else { "error" };
     global_metrics().record_scanner_bucket_drive_result(bucket, disk, result);
     metrics::counter!(
@@ -1764,7 +1789,7 @@ pub fn emit_scan_bucket_drive_complete(success: bool, bucket: &str, disk: &str,
     .record(duration.as_secs_f64());
 }
 
-pub fn emit_scan_bucket_drive_partial(bucket: &str, disk: &str, duration: Duration) {
+pub fn emit_scan_bucket_drive_partial(_source: ScannerWorkSource, bucket: &str, disk: &str, duration: Duration) {
     global_metrics().record_scanner_bucket_drive_result(bucket, disk, SCAN_CYCLE_RESULT_PARTIAL_LABEL);
     metrics::counter!(
         OTEL_SCANNER_BUCKETS_SCANNED,
@@ -1817,6 +1842,7 @@ impl Metrics {
             scanner_set_scans_active: AtomicU64::new(0),
             scanner_disk_bucket_scan_states: Mutex::new(HashMap::new()),
             scanner_bucket_drive_results: Mutex::new(ScannerBucketDriveResults::default()),
+            scanner_active_bucket_drive_scans: Mutex::new(HashMap::new()),
             scanner_bucket_drive_result_clock: AtomicU64::new(0),
             current_scan_cycle_bucket_drive_results_start: Mutex::new(HashMap::new()),
             last_scan_cycle_bucket_drive_results: Mutex::new(Vec::new()),
@@ -2308,8 +2334,45 @@ impl Metrics {
         }
     }
 
-    pub fn record_scan_bucket_drive_start(&self) {
+    pub fn record_scan_bucket_drive_start(&self, source: ScannerWorkSource, bucket: &str, drive: &str) {
         self.operations[Metric::ScanBucketDriveStart as usize].fetch_add(1, Ordering::Relaxed);
+        if bucket.is_empty() || drive.is_empty() {
+            return;
+        }
+        let key = ScannerActiveBucketDriveKey {
+            source: source.as_str().to_string(),
+            bucket: bucket.to_string(),
+            drive: drive.to_string(),
+        };
+        let mut active = self
+            .scanner_active_bucket_drive_scans
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        active
+            .entry(key)
+            .and_modify(|value| value.count = value.count.saturating_add(1))
+            .or_insert(ScannerActiveBucketDriveValue {
+                count: 1,
+                started_at: Timestamp::now(),
+            });
+    }
+
+    pub fn record_scan_bucket_drive_end(&self, source: ScannerWorkSource, bucket: &str, drive: &str) {
+        let key = ScannerActiveBucketDriveKey {
+            source: source.as_str().to_string(),
+            bucket: bucket.to_string(),
+            drive: drive.to_string(),
+        };
+        let mut active = self
+            .scanner_active_bucket_drive_scans
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        if let Some(value) = active.get_mut(&key) {
+            value.count = value.count.saturating_sub(1);
+            if value.count == 0 {
+                active.remove(&key);
+            }
+        }
     }
 
     pub fn record_scan_bucket_drive_failure(&self) {
@@ -2782,6 +2845,26 @@ impl Metrics {
         } else {
             Vec::new()
         };
+        let now = Timestamp::now();
+        let mut active_bucket_drive_scans = self
+            .scanner_active_bucket_drive_scans
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .iter()
+            .map(|(key, value)| ScannerActiveBucketDriveSnapshot {
+                source: key.source.clone(),
+                bucket: key.bucket.clone(),
+                drive: key.drive.clone(),
+                count: value.count,
+                age_seconds: timestamp_elapsed_seconds_since(now, value.started_at),
+            })
+            .collect::<Vec<_>>();
+        active_bucket_drive_scans.sort_by(|left, right| {
+            left.source
+                .cmp(&right.source)
+                .then_with(|| left.bucket.cmp(&right.bucket))
+                .then_with(|| left.drive.cmp(&right.drive))
+        });
         ScannerRuntimeDetailsReport {
             disk_bucket_scan_states: self.scanner_disk_bucket_scan_state_snapshots(),
             bucket_drive_results: self.scanner_bucket_drive_result_counter_snapshots(),
@@ -2791,6 +2874,7 @@ impl Metrics {
                 .lock()
                 .unwrap_or_else(|poisoned| poisoned.into_inner())
                 .clone(),
+            active_bucket_drive_scans,
         }
     }
 
@@ -4371,7 +4455,7 @@ mod tests {
     #[tokio::test]
     async fn report_includes_bucket_drive_scan_starts() {
         let metrics = Metrics::new();
-        metrics.record_scan_bucket_drive_start();
+        metrics.record_scan_bucket_drive_start(ScannerWorkSource::Usage, "bucket-a", "/mnt/data/1");
         metrics.record_scan_bucket_drive_failure();
 
         let report = metrics.report().await;
@@ -4380,6 +4464,27 @@ mod tests {
         assert_eq!(report.life_time_ops.get("scan_bucket_drive_failure"), Some(&1));
     }
 
+    #[tokio::test]
+    async fn active_bucket_drive_snapshot_is_structured_and_retired_on_end() {
+        let metrics = Metrics::new();
+        metrics.record_scan_bucket_drive_start(ScannerWorkSource::Usage, "bucket-a", "/mnt/data/1");
+        metrics.record_scan_bucket_drive_start(ScannerWorkSource::Usage, "bucket-a", "/mnt/data/1");
+        let active = metrics.scanner_runtime_details_report().active_bucket_drive_scans;
+        assert_eq!(active.len(), 1);
+        assert_eq!(active[0].source, ScannerWorkSource::Usage.as_str());
+        assert_eq!(active[0].bucket, "bucket-a");
+        assert_eq!(active[0].drive, "/mnt/data/1");
+        assert_eq!(active[0].count, 2);
+
+        metrics.record_scan_bucket_drive_end(ScannerWorkSource::Usage, "bucket-a", "/mnt/data/1");
+        assert_eq!(metrics.scanner_runtime_details_report().active_bucket_drive_scans[0].count, 1);
+        metrics.record_scan_bucket_drive_end(ScannerWorkSource::Usage, "bucket-a", "/mnt/data/1");
+        assert!(metrics.scanner_runtime_details_report().active_bucket_drive_scans.is_empty());
+
+        metrics.record_scan_bucket_drive_start(ScannerWorkSource::Usage, "", "/mnt/data/1");
+        assert!(metrics.scanner_runtime_details_report().active_bucket_drive_scans.is_empty());
+    }
+
     #[tokio::test]
     async fn report_includes_structured_bucket_drive_results() {
         let metrics = Metrics::new();
diff --git a/crates/obs/src/metrics/collectors/ilm.rs b/crates/obs/src/metrics/collectors/ilm.rs
index d9a3d55a3..4c04bd390 100644
--- a/crates/obs/src/metrics/collectors/ilm.rs
+++ b/crates/obs/src/metrics/collectors/ilm.rs
@@ -54,12 +54,37 @@ pub(crate) struct IlmActionTaskStats {
     pub(crate) value: u64,
 }
 
+#[derive(Debug, Clone, Default)]
+pub(crate) struct IlmQueueTaskStats {
+    pub(crate) action: String,
+    pub(crate) state: String,
+    pub(crate) value: u64,
+}
+
+#[derive(Debug, Clone, Default)]
+pub(crate) struct IlmTaskEventStats {
+    pub(crate) action: String,
+    pub(crate) result: String,
+    pub(crate) value: u64,
+}
+
+#[derive(Debug, Clone, Default)]
+pub(crate) struct IlmBackpressureStats {
+    pub(crate) action: String,
+    pub(crate) reason: String,
+    pub(crate) value: u64,
+}
+
 /// ILM statistics with runtime-local node identity and bounded action/state details.
 #[derive(Debug, Clone, Default)]
 pub(crate) struct IlmRuntimeStats {
     pub(crate) server: String,
     pub(crate) stats: IlmStats,
     pub(crate) action_tasks: Vec<IlmActionTaskStats>,
+    pub(crate) queue_tasks: Vec<IlmQueueTaskStats>,
+    pub(crate) task_events: Vec<IlmTaskEventStats>,
+    pub(crate) backpressure: Vec<IlmBackpressureStats>,
+    pub(crate) versions_scanned: u64,
 }
 
 fn is_live_action_task_state(state: &str) -> bool {
@@ -112,6 +137,30 @@ pub(crate) fn collect_ilm_runtime_metrics(stats: &IlmRuntimeStats) -> Vec<Promet
             }),
     );
 
+    metrics.extend(stats.queue_tasks.iter().map(|task| {
+        PrometheusMetric::from_descriptor(&ILM_TASKS_MD, task.value as f64)
+            .with_label_owned(SERVER_LABEL, stats.server.clone())
+            .with_label_owned(ACTION_LABEL, task.action.clone())
+            .with_label_owned(QUEUE_STATE_LABEL, task.state.clone())
+    }));
+    metrics.extend(stats.task_events.iter().map(|event| {
+        PrometheusMetric::from_descriptor(&ILM_TASK_EVENTS_MD, event.value as f64)
+            .with_label_owned(SERVER_LABEL, stats.server.clone())
+            .with_label_owned(ACTION_LABEL, event.action.clone())
+            .with_label_owned(RESULT_LABEL, event.result.clone())
+    }));
+    metrics.extend(stats.backpressure.iter().map(|event| {
+        PrometheusMetric::from_descriptor(&ILM_QUEUE_BACKPRESSURE_MD, event.value as f64)
+            .with_label_owned(SERVER_LABEL, stats.server.clone())
+            .with_label_owned(ACTION_LABEL, event.action.clone())
+            .with_label_owned(REASON_LABEL, event.reason.clone())
+    }));
+    metrics.push(
+        PrometheusMetric::from_descriptor(&ILM_VERSIONS_SCANNED_BY_SERVER_MD, stats.versions_scanned as f64)
+            .with_label_owned(SERVER_LABEL, stats.server.clone())
+            .with_label_owned(SOURCE_LABEL, "lifecycle".to_string()),
+    );
+
     metrics
 }
 
@@ -135,6 +184,22 @@ mod tests {
         let runtime_stats = IlmRuntimeStats {
             server: "node1:9000".to_string(),
             stats,
+            queue_tasks: vec![IlmQueueTaskStats {
+                action: "transition".to_string(),
+                state: "pending".to_string(),
+                value: 8,
+            }],
+            task_events: vec![IlmTaskEventStats {
+                action: "transition".to_string(),
+                result: "completed".to_string(),
+                value: 7,
+            }],
+            backpressure: vec![IlmBackpressureStats {
+                action: "transition".to_string(),
+                reason: "queue_full".to_string(),
+                value: 2,
+            }],
+            versions_scanned: 1000000,
             action_tasks: vec![
                 IlmActionTaskStats {
                     action: "expiry".to_string(),
@@ -156,7 +221,7 @@ mod tests {
 
         let metrics = collect_ilm_runtime_metrics(&runtime_stats);
 
-        assert_eq!(metrics.len(), 11);
+        assert_eq!(metrics.len(), 15);
 
         let pending = metrics.iter().find(|m| m.value == 100.0);
         assert!(pending.is_some());
@@ -178,6 +243,44 @@ mod tests {
         });
         assert!(transition_timeout.is_none());
 
+        let transition_queue = metrics.iter().find(|m| {
+            m.name == ILM_TASKS_MD.get_full_metric_name()
+                && m.labels
+                    .iter()
+                    .any(|(name, value)| *name == ACTION_LABEL && value.as_ref() == "transition")
+                && m.labels
+                    .iter()
+                    .any(|(name, value)| *name == QUEUE_STATE_LABEL && value.as_ref() == "pending")
+        });
+        assert_eq!(transition_queue.map(|metric| metric.value), Some(8.0));
+
+        let completed = metrics.iter().find(|m| {
+            m.name == ILM_TASK_EVENTS_MD.get_full_metric_name()
+                && m.labels
+                    .iter()
+                    .any(|(name, value)| *name == RESULT_LABEL && value.as_ref() == "completed")
+        });
+        assert_eq!(completed.map(|metric| metric.value), Some(7.0));
+
+        let backpressure = metrics.iter().find(|m| {
+            m.name == ILM_QUEUE_BACKPRESSURE_MD.get_full_metric_name()
+                && m.labels
+                    .iter()
+                    .any(|(name, value)| *name == REASON_LABEL && value.as_ref() == "queue_full")
+        });
+        assert_eq!(backpressure.map(|metric| metric.value), Some(2.0));
+
+        let version_detail = metrics.iter().find(|m| {
+            m.name == ILM_VERSIONS_SCANNED_BY_SERVER_MD.get_full_metric_name()
+                && m.labels
+                    .iter()
+                    .any(|(name, value)| *name == SERVER_LABEL && value.as_ref() == "node1:9000")
+                && m.labels
+                    .iter()
+                    .any(|(name, value)| *name == SOURCE_LABEL && value.as_ref() == "lifecycle")
+        });
+        assert_eq!(version_detail.map(|metric| metric.value), Some(1000000.0));
+
         let transition_active = metrics.iter().find(|m| {
             m.name == ILM_ACTION_TASKS_MD.get_full_metric_name()
                 && m.labels
diff --git a/crates/obs/src/metrics/collectors/mod.rs b/crates/obs/src/metrics/collectors/mod.rs
index 1a356025f..67df19035 100644
--- a/crates/obs/src/metrics/collectors/mod.rs
+++ b/crates/obs/src/metrics/collectors/mod.rs
@@ -59,9 +59,12 @@ pub use cluster_iam::{IamStats, collect_iam_metrics};
 pub use cluster_usage::{BucketUsageStats, ClusterUsageStats, collect_bucket_usage_metrics, collect_cluster_usage_metrics};
 pub use compression::{CompressionClusterStats, collect_compression_cluster_metrics};
 pub use dial9::{Dial9Stats, collect_current_dial9_metrics, collect_dial9_metrics, is_dial9_enabled};
-pub(crate) use ilm::{IlmActionTaskStats, IlmRuntimeStats, collect_ilm_runtime_metrics};
+pub(crate) use ilm::{
+    IlmActionTaskStats, IlmBackpressureStats, IlmQueueTaskStats, IlmRuntimeStats, IlmTaskEventStats, collect_ilm_runtime_metrics,
+};
 pub use ilm::{IlmStats, collect_ilm_metrics};
 pub use node::{DiskStats, collect_node_metrics};
+pub(crate) use notification::collect_notification_runtime_metrics;
 pub use notification::{NotificationStats, collect_notification_metrics};
 pub(crate) use notification_target::{NotificationTargetRuntimeStats, collect_notification_target_runtime_metrics};
 pub use notification_target::{NotificationTargetStats, collect_notification_target_metrics};
diff --git a/crates/obs/src/metrics/collectors/notification.rs b/crates/obs/src/metrics/collectors/notification.rs
index 7a10e0efd..e7c2f4177 100644
--- a/crates/obs/src/metrics/collectors/notification.rs
+++ b/crates/obs/src/metrics/collectors/notification.rs
@@ -19,9 +19,12 @@
 
 use crate::metrics::report::PrometheusMetric;
 use crate::metrics::schema::cluster_notification::{
-    NOTIFICATION_CURRENT_SEND_IN_PROGRESS_MD, NOTIFICATION_EVENTS_ERRORS_TOTAL_MD, NOTIFICATION_EVENTS_SENT_TOTAL_MD,
-    NOTIFICATION_EVENTS_SKIPPED_TOTAL_MD,
+    NOTIFICATION_CURRENT_SEND_IN_PROGRESS_BY_SERVER_MD, NOTIFICATION_CURRENT_SEND_IN_PROGRESS_MD,
+    NOTIFICATION_EVENTS_ERRORS_TOTAL_BY_SERVER_MD, NOTIFICATION_EVENTS_ERRORS_TOTAL_MD,
+    NOTIFICATION_EVENTS_SENT_TOTAL_BY_SERVER_MD, NOTIFICATION_EVENTS_SENT_TOTAL_MD,
+    NOTIFICATION_EVENTS_SKIPPED_TOTAL_BY_SERVER_MD, NOTIFICATION_EVENTS_SKIPPED_TOTAL_MD, SERVER,
 };
+use std::borrow::Cow;
 
 /// Notification statistics.
 #[derive(Debug, Clone, Default)]
@@ -49,6 +52,30 @@ pub fn collect_notification_metrics(stats: &NotificationStats) -> Vec<Prometheus
     ]
 }
 
+/// Collects the legacy aggregate metrics and node-local runtime siblings.
+pub(crate) fn collect_notification_runtime_metrics(stats: &NotificationStats, server: &str) -> Vec<PrometheusMetric> {
+    let mut metrics = collect_notification_metrics(stats);
+    if server.is_empty() {
+        return metrics;
+    }
+
+    let server_label: Cow<'static, str> = Cow::Owned(server.to_string());
+    metrics.extend([
+        PrometheusMetric::from_descriptor(
+            &NOTIFICATION_CURRENT_SEND_IN_PROGRESS_BY_SERVER_MD,
+            stats.current_send_in_progress as f64,
+        )
+        .with_label(SERVER, server_label.clone()),
+        PrometheusMetric::from_descriptor(&NOTIFICATION_EVENTS_ERRORS_TOTAL_BY_SERVER_MD, stats.events_errors_total as f64)
+            .with_label(SERVER, server_label.clone()),
+        PrometheusMetric::from_descriptor(&NOTIFICATION_EVENTS_SENT_TOTAL_BY_SERVER_MD, stats.events_sent_total as f64)
+            .with_label(SERVER, server_label.clone()),
+        PrometheusMetric::from_descriptor(&NOTIFICATION_EVENTS_SKIPPED_TOTAL_BY_SERVER_MD, stats.events_skipped_total as f64)
+            .with_label(SERVER, server_label),
+    ]);
+    metrics
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -86,4 +113,32 @@ mod tests {
             assert!(metric.labels.is_empty());
         }
     }
+
+    #[test]
+    fn runtime_metrics_keep_aggregate_and_add_server_siblings() {
+        let stats = NotificationStats {
+            current_send_in_progress: 5,
+            events_errors_total: 10,
+            events_sent_total: 100,
+            events_skipped_total: 2,
+        };
+
+        let metrics = collect_notification_runtime_metrics(&stats, "node1:9000");
+        assert_eq!(metrics.len(), 8);
+        assert_eq!(metrics.iter().filter(|metric| metric.labels.is_empty()).count(), 4);
+        assert_eq!(metrics.iter().filter(|metric| metric.labels.len() == 1).count(), 4);
+        assert!(metrics.iter().filter(|metric| metric.labels.len() == 1).all(|metric| {
+            metric
+                .labels
+                .iter()
+                .any(|(name, value)| *name == SERVER && value == "node1:9000")
+        }));
+    }
+
+    #[test]
+    fn runtime_metrics_do_not_publish_empty_server_series() {
+        let metrics = collect_notification_runtime_metrics(&NotificationStats::default(), "");
+        assert_eq!(metrics.len(), 4);
+        assert!(metrics.iter().all(|metric| metric.labels.is_empty()));
+    }
 }
diff --git a/crates/obs/src/metrics/collectors/scanner.rs b/crates/obs/src/metrics/collectors/scanner.rs
index 56f1db556..1e0221908 100644
--- a/crates/obs/src/metrics/collectors/scanner.rs
+++ b/crates/obs/src/metrics/collectors/scanner.rs
@@ -184,6 +184,15 @@ pub struct ScannerBucketDriveResultStats {
     pub count: u64,
 }
 
+#[derive(Debug, Clone, Default)]
+pub struct ScannerActiveBucketDriveStats {
+    pub source: String,
+    pub bucket: String,
+    pub drive: String,
+    pub count: u64,
+    pub age_seconds: u64,
+}
+
 /// Scanner statistics with runtime-local node identity and bounded source/result details.
 #[derive(Debug, Clone, Default)]
 pub(crate) struct ScannerRuntimeStats {
@@ -195,6 +204,7 @@ pub(crate) struct ScannerRuntimeStats {
     pub(crate) bucket_drive_results: Vec<ScannerBucketDriveResultStats>,
     pub(crate) current_cycle_bucket_drive_results: Vec<ScannerBucketDriveResultStats>,
     pub(crate) last_cycle_bucket_drive_results: Vec<ScannerBucketDriveResultStats>,
+    pub(crate) active_bucket_drive_scans: Vec<ScannerActiveBucketDriveStats>,
 }
 
 /// Collects scanner metrics from the given stats.
@@ -452,6 +462,23 @@ fn collect_scanner_metrics_with_runtime(stats: &ScannerStats, runtime: Option<&S
             &runtime.last_cycle_bucket_drive_results,
             Some("last"),
         );
+        for active in &runtime.active_bucket_drive_scans {
+            let labels = |metric: PrometheusMetric| {
+                metric
+                    .with_label_owned(SERVER_LABEL, runtime.server.clone())
+                    .with_label_owned(SOURCE_LABEL, active.source.clone())
+                    .with_label_owned(BUCKET_LABEL, active.bucket.clone())
+                    .with_label_owned(DRIVE_LABEL, active.drive.clone())
+            };
+            metrics.push(labels(PrometheusMetric::from_descriptor(
+                &SCANNER_ACTIVE_BUCKET_DRIVE_SCANS_MD,
+                active.count as f64,
+            )));
+            metrics.push(labels(PrometheusMetric::from_descriptor(
+                &SCANNER_ACTIVE_BUCKET_DRIVE_SCAN_AGE_SECONDS_MD,
+                active.age_seconds as f64,
+            )));
+        }
     }
 
     metrics
@@ -566,6 +593,13 @@ mod tests {
                 result: "error".to_string(),
                 count: 2,
             }],
+            active_bucket_drive_scans: vec![ScannerActiveBucketDriveStats {
+                source: "usage".to_string(),
+                bucket: "photos".to_string(),
+                drive: "/data1".to_string(),
+                count: 2,
+                age_seconds: 7,
+            }],
             stats: ScannerStats {
                 bucket_scans_finished: 100,
                 bucket_scans_started: 100,
@@ -642,7 +676,7 @@ mod tests {
         let metrics = collect_scanner_runtime_metrics(&stats);
         report_metrics(&metrics);
 
-        assert_eq!(metrics.len(), 90);
+        assert_eq!(metrics.len(), 92);
 
         let objects = metrics.iter().find(|m| m.value == 1000000.0);
         assert!(objects.is_some());
@@ -656,6 +690,35 @@ mod tests {
         assert_eq!(active_paths.map(|m| m.value), Some(4.0));
         assert_eq!(active_paths.map(|m| m.labels.len()), Some(0));
 
+        let active_bucket_drive = metrics
+            .iter()
+            .find(|m| m.name == SCANNER_ACTIVE_BUCKET_DRIVE_SCANS_MD.get_full_metric_name())
+            .expect("active bucket-drive metric");
+        assert_eq!(active_bucket_drive.value, 2.0);
+        assert!(
+            active_bucket_drive
+                .labels
+                .iter()
+                .any(|(name, value)| *name == SOURCE_LABEL && value == "usage")
+        );
+        assert!(
+            active_bucket_drive
+                .labels
+                .iter()
+                .any(|(name, value)| *name == BUCKET_LABEL && value == "photos")
+        );
+        assert!(
+            active_bucket_drive
+                .labels
+                .iter()
+                .any(|(name, value)| *name == DRIVE_LABEL && value == "/data1")
+        );
+        let active_age = metrics
+            .iter()
+            .find(|m| m.name == SCANNER_ACTIVE_BUCKET_DRIVE_SCAN_AGE_SECONDS_MD.get_full_metric_name())
+            .expect("active bucket-drive age metric");
+        assert_eq!(active_age.value, 7.0);
+
         let bucket_drive_result = metrics
             .iter()
             .find(|m| m.name == SCANNER_BUCKET_DRIVE_RESULT_TOTAL_MD.get_full_metric_name());
diff --git a/crates/obs/src/metrics/collectors/system_drive.rs b/crates/obs/src/metrics/collectors/system_drive.rs
index 8306ef175..c38797bfa 100644
--- a/crates/obs/src/metrics/collectors/system_drive.rs
+++ b/crates/obs/src/metrics/collectors/system_drive.rs
@@ -60,6 +60,10 @@ pub struct DriveDetailedStats {
     pub api_latency_micros: Option<u64>,
     /// Health status (1=healthy, 0=unhealthy)
     pub health: u8,
+    /// Total successful write operations when backed by a real disk metric.
+    pub writes_total: Option<u64>,
+    /// Total successful delete operations when backed by a real disk metric.
+    pub deletes_total: Option<u64>,
     /// Reads per second when backed by a real iostat sample
     pub reads_per_sec: Option<f64>,
     /// Kilobytes read per second when backed by a real iostat sample
@@ -282,6 +286,12 @@ pub(crate) fn collect_drive_runtime_detailed_metrics(stats: &[DriveRuntimeDetail
         if let Some(value) = stat.stats.perc_util {
             push_drive_metric(&mut metrics, &DRIVE_PERC_UTIL_MD, value, server_label, drive_label);
         }
+        if let Some(value) = stat.stats.writes_total {
+            push_drive_metric(&mut metrics, &DRIVE_WRITES_TOTAL_MD, value as f64, server_label, drive_label);
+        }
+        if let Some(value) = stat.stats.deletes_total {
+            push_drive_metric(&mut metrics, &DRIVE_DELETES_TOTAL_MD, value as f64, server_label, drive_label);
+        }
         if let Some(labels) = &topology_labels {
             if let Some(disk_id) = stat.disk_id.as_ref().filter(|disk_id| !disk_id.is_empty()) {
                 metrics.push(
@@ -449,6 +459,8 @@ mod tests {
                 waiting_io: Some(3),
                 api_latency_micros: Some(1500),
                 health: 1,
+                writes_total: Some(11),
+                deletes_total: Some(4),
                 reads_per_sec: Some(100.0),
                 reads_kb_per_sec: Some(1024.0),
                 reads_await: Some(5.5),
@@ -462,7 +474,7 @@ mod tests {
         let metrics = collect_drive_runtime_detailed_metrics(&stats);
         report_metrics(&metrics);
 
-        assert_eq!(metrics.len(), 34);
+        assert_eq!(metrics.len(), 36);
 
         // Verify total bytes metric
         let total_bytes_name = DRIVE_TOTAL_BYTES_MD.get_full_metric_name();
@@ -503,6 +515,8 @@ mod tests {
                 API_LABEL,
             ],
         );
+        assert_metric_label_keys(&metrics, &DRIVE_WRITES_TOTAL_MD, 11.0, &[SERVER_LABEL, DRIVE_LABEL]);
+        assert_metric_label_keys(&metrics, &DRIVE_DELETES_TOTAL_MD, 4.0, &[SERVER_LABEL, DRIVE_LABEL]);
     }
 
     #[test]
@@ -524,6 +538,8 @@ mod tests {
             waiting_io: None,
             api_latency_micros: None,
             health: 1,
+            writes_total: None,
+            deletes_total: None,
             reads_per_sec: None,
             reads_kb_per_sec: None,
             reads_await: None,
diff --git a/crates/obs/src/metrics/scheduler.rs b/crates/obs/src/metrics/scheduler.rs
index a7e4f8325..088799d5e 100644
--- a/crates/obs/src/metrics/scheduler.rs
+++ b/crates/obs/src/metrics/scheduler.rs
@@ -62,7 +62,7 @@ use crate::metrics::collectors::{
     collect_memory_metrics,
     collect_network_metrics,
     collect_node_metrics,
-    collect_notification_metrics,
+    collect_notification_runtime_metrics,
     collect_notification_target_runtime_metrics,
     collect_process_attributes,
     collect_process_cpu_metrics,
@@ -120,12 +120,13 @@ use crate::metrics::schema::notification_target::{
 };
 use crate::metrics::schema::scanner::{
     BUCKET_LABEL as SCANNER_BUCKET_LABEL, CYCLE_SCOPE_LABEL as SCANNER_CYCLE_SCOPE_LABEL, DRIVE_LABEL as SCANNER_DRIVE_LABEL,
-    RESULT_LABEL as SCANNER_RESULT_LABEL, SCANNER_BUCKET_DRIVE_RESULT_TOTAL_MD, SCANNER_CYCLE_BUCKET_DRIVE_RESULT_MD,
+    RESULT_LABEL as SCANNER_RESULT_LABEL, SCANNER_ACTIVE_BUCKET_DRIVE_SCAN_AGE_SECONDS_MD, SCANNER_ACTIVE_BUCKET_DRIVE_SCANS_MD,
+    SCANNER_BUCKET_DRIVE_RESULT_TOTAL_MD, SCANNER_CYCLE_BUCKET_DRIVE_RESULT_MD, SOURCE_LABEL as SCANNER_SOURCE_LABEL,
 };
 use crate::metrics::schema::system_drive::{
-    API_LABEL as DRIVE_API_LABEL, DISK_ID_LABEL, DRIVE_API_CALLS_MD, DRIVE_API_LATENCY_BY_API_MD, DRIVE_HEALING_MD,
-    DRIVE_INDEX_LABEL, DRIVE_INFO_MD, DRIVE_LABEL, DRIVE_OFFLINE_DURATION_SECONDS_MD, DRIVE_RUNTIME_STATE_MD, DRIVE_SCANNING_MD,
-    POOL_INDEX_LABEL, SET_INDEX_LABEL, STATE_LABEL as DRIVE_STATE_LABEL,
+    API_LABEL as DRIVE_API_LABEL, DISK_ID_LABEL, DRIVE_API_CALLS_MD, DRIVE_API_LATENCY_BY_API_MD, DRIVE_DELETES_TOTAL_MD,
+    DRIVE_HEALING_MD, DRIVE_INDEX_LABEL, DRIVE_INFO_MD, DRIVE_LABEL, DRIVE_OFFLINE_DURATION_SECONDS_MD, DRIVE_RUNTIME_STATE_MD,
+    DRIVE_SCANNING_MD, DRIVE_WRITES_TOTAL_MD, POOL_INDEX_LABEL, SET_INDEX_LABEL, STATE_LABEL as DRIVE_STATE_LABEL,
 };
 use crate::metrics::schema::system_process::{PROCESS_EXECUTABLE_NAME_LABEL, PROCESS_PID_LABEL};
 use crate::metrics::stats_collector::{
@@ -303,15 +304,33 @@ type AuditTargetKey = (String, String); // (server, target_id)
 type NotificationLegacyTargetKey = (String, String); // (target_id, target_type)
 type NotificationTargetKey = (String, String, String); // (server, target_id, target_type)
 type DriveTopologyKey = (String, String, String, String, String); // (server, drive, pool, set, drive_index)
+type DriveBasicKey = (String, String); // (server, drive)
 type DriveTopologyApiKey = (String, String, String, String, String, String); // (server, drive, pool, set, drive_index, api)
 type DriveInfoKey = (String, String, String, String, String, String); // (server, drive, pool, set, drive_index, disk_id)
 type ScannerCycleBucketDriveResultKey = (String, String, String, String, String); // (server, cycle_scope, bucket, drive, result)
 type ScannerBucketDriveResultKey = (String, String, String, String); // (server, bucket, drive, result)
+type ScannerActiveBucketDriveKey = (String, String, String, String); // (server, source, bucket, drive)
 
 fn drive_info_live_keys(stats: &[DriveRuntimeDetailedStats]) -> HashSet<DriveInfoKey> {
     stats.iter().filter_map(drive_info_key).collect()
 }
 
+fn drive_basic_live_keys(stats: &[DriveRuntimeDetailedStats]) -> HashSet<DriveBasicKey> {
+    stats
+        .iter()
+        .map(|stat| (stat.stats.server.clone(), stat.stats.drive.clone()))
+        .collect()
+}
+
+fn retire_drive_basic_metric_series(key: &DriveBasicKey) -> usize {
+    let labels = [
+        (SERVER_LABEL, Cow::Owned(key.0.clone())),
+        (DRIVE_LABEL, Cow::Owned(key.1.clone())),
+    ];
+    retire_metric_series(&DRIVE_WRITES_TOTAL_MD.get_full_metric_name(), &labels)
+        + retire_metric_series(&DRIVE_DELETES_TOTAL_MD.get_full_metric_name(), &labels)
+}
+
 fn drive_topology_live_keys(stats: &[DriveRuntimeDetailedStats]) -> HashSet<DriveTopologyKey> {
     stats.iter().filter_map(drive_topology_key).collect()
 }
@@ -469,6 +488,25 @@ fn retire_scanner_bucket_drive_result_metric_series(key: &ScannerBucketDriveResu
     retire_metric_series(&SCANNER_BUCKET_DRIVE_RESULT_TOTAL_MD.get_full_metric_name(), &labels)
 }
 
+fn scanner_active_bucket_drive_live_keys(stats: &ScannerRuntimeStats) -> HashSet<ScannerActiveBucketDriveKey> {
+    stats
+        .active_bucket_drive_scans
+        .iter()
+        .map(|active| (stats.server.clone(), active.source.clone(), active.bucket.clone(), active.drive.clone()))
+        .collect()
+}
+
+fn retire_scanner_active_bucket_drive_metric_series(key: &ScannerActiveBucketDriveKey) -> usize {
+    let labels = [
+        (SERVER_LABEL, Cow::Owned(key.0.clone())),
+        (SCANNER_SOURCE_LABEL, Cow::Owned(key.1.clone())),
+        (SCANNER_BUCKET_LABEL, Cow::Owned(key.2.clone())),
+        (SCANNER_DRIVE_LABEL, Cow::Owned(key.3.clone())),
+    ];
+    retire_metric_series(&SCANNER_ACTIVE_BUCKET_DRIVE_SCANS_MD.get_full_metric_name(), &labels)
+        + retire_metric_series(&SCANNER_ACTIVE_BUCKET_DRIVE_SCAN_AGE_SECONDS_MD.get_full_metric_name(), &labels)
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Default)]
 pub struct MetricsRuntimeCollectorHealthSnapshot {
     pub healthy_collectors: u8,
@@ -1841,6 +1879,7 @@ pub fn init_metrics_runtime(token: CancellationToken) {
     let token_clone = token.clone();
     tokio::spawn(async move {
         let mut interval = metrics_interval(node_interval, Duration::ZERO);
+        let mut prev_drive_basic_keys: HashSet<DriveBasicKey> = HashSet::new();
         let mut prev_drive_info_keys: HashSet<DriveInfoKey> = HashSet::new();
         let mut prev_drive_topology_keys: HashSet<DriveTopologyKey> = HashSet::new();
         let mut prev_drive_topology_api_keys: HashSet<DriveTopologyApiKey> = HashSet::new();
@@ -1851,6 +1890,7 @@ pub fn init_metrics_runtime(token: CancellationToken) {
                     run_metrics_collector_tick(health, MetricsCollectorTaskId::NodeDiskStats, "node_disk_stats", async {
                         let (disk_stats, drive_stats, drive_counts) = collect_disk_and_system_drive_runtime_stats().await;
                         let current_drive_info_keys = drive_info_live_keys(&drive_stats);
+                        let current_drive_basic_keys = drive_basic_live_keys(&drive_stats);
                         let current_drive_topology_keys = drive_topology_live_keys(&drive_stats);
                         let current_drive_topology_api_keys = drive_topology_api_live_keys(&drive_stats);
                         let retire_drive_info_keys = if has_seen_drive_info_snapshot {
@@ -1858,6 +1898,11 @@ pub fn init_metrics_runtime(token: CancellationToken) {
                         } else {
                             Vec::new()
                         };
+                        let retire_drive_basic_keys = if has_seen_drive_info_snapshot {
+                            prev_drive_basic_keys.difference(&current_drive_basic_keys).cloned().collect::<Vec<_>>()
+                        } else {
+                            Vec::new()
+                        };
                         let retire_drive_topology_keys = if has_seen_drive_info_snapshot {
                             prev_drive_topology_keys.difference(&current_drive_topology_keys).cloned().collect::<Vec<_>>()
                         } else {
@@ -1872,6 +1917,7 @@ pub fn init_metrics_runtime(token: CancellationToken) {
                             Vec::new()
                         };
                         prev_drive_info_keys = current_drive_info_keys;
+                        prev_drive_basic_keys = current_drive_basic_keys;
                         prev_drive_topology_keys = current_drive_topology_keys;
                         prev_drive_topology_api_keys = current_drive_topology_api_keys;
                         has_seen_drive_info_snapshot = true;
@@ -1882,6 +1928,9 @@ pub fn init_metrics_runtime(token: CancellationToken) {
                         for key in retire_drive_info_keys {
                             let _ = retire_drive_info_metric_series(&key);
                         }
+                        for key in retire_drive_basic_keys {
+                            let _ = retire_drive_basic_metric_series(&key);
+                        }
                         for key in retire_drive_topology_keys {
                             let _ = retire_drive_topology_metric_series(&key);
                         }
@@ -2106,14 +2155,14 @@ pub fn init_metrics_runtime(token: CancellationToken) {
                 _ = interval.tick() => {
                     run_metrics_collector_tick(health, MetricsCollectorTaskId::NotificationStats, "notification_stats", async {
                         let snapshot = notification_metrics_snapshot();
-                        let mut metrics = collect_notification_metrics(&NotificationStats {
+                        let server = current_local_node_identity();
+                        let mut metrics = collect_notification_runtime_metrics(&NotificationStats {
                             current_send_in_progress: snapshot.current_send_in_progress,
                             events_errors_total: snapshot.events_errors_total,
                             events_sent_total: snapshot.events_sent_total,
                             events_skipped_total: snapshot.events_skipped_total,
-                        });
+                        }, &server);
 
-                        let server = current_local_node_identity();
                         let target_stats = notification_target_metrics().await
                             .into_iter()
                             .map(|snapshot| NotificationTargetRuntimeStats {
@@ -2173,6 +2222,7 @@ pub fn init_metrics_runtime(token: CancellationToken) {
         let mut has_seen_scanner_snapshot = false;
         let mut prev_scanner_cycle_bucket_drive_result_keys: HashSet<ScannerCycleBucketDriveResultKey> = HashSet::new();
         let mut prev_scanner_bucket_drive_result_keys: HashSet<ScannerBucketDriveResultKey> = HashSet::new();
+        let mut prev_scanner_active_bucket_drive_keys: HashSet<ScannerActiveBucketDriveKey> = HashSet::new();
         loop {
             tokio::select! {
                 _ = interval.tick() => {
@@ -2189,9 +2239,11 @@ pub fn init_metrics_runtime(token: CancellationToken) {
 
                             let mut retire_scanner_cycle_bucket_drive_result_keys = Vec::new();
                             let mut retire_scanner_bucket_drive_result_keys = Vec::new();
+                            let mut retire_scanner_active_bucket_drive_keys = Vec::new();
                             if let Some(stats) = collect_scanner_runtime_metric_stats().await {
                                 let current_cycle_keys = scanner_cycle_bucket_drive_result_live_keys(&stats);
                                 let current_keys = scanner_bucket_drive_result_live_keys(&stats);
+                                let current_active_keys = scanner_active_bucket_drive_live_keys(&stats);
                                 if has_seen_scanner_snapshot {
                                     retire_scanner_cycle_bucket_drive_result_keys = prev_scanner_cycle_bucket_drive_result_keys
                                         .difference(&current_cycle_keys)
@@ -2201,9 +2253,14 @@ pub fn init_metrics_runtime(token: CancellationToken) {
                                         .difference(&current_keys)
                                         .cloned()
                                         .collect();
+                                    retire_scanner_active_bucket_drive_keys = prev_scanner_active_bucket_drive_keys
+                                        .difference(&current_active_keys)
+                                        .cloned()
+                                        .collect();
                                 }
                                 prev_scanner_cycle_bucket_drive_result_keys = current_cycle_keys;
                                 prev_scanner_bucket_drive_result_keys = current_keys;
+                                prev_scanner_active_bucket_drive_keys = current_active_keys;
                                 has_seen_scanner_snapshot = true;
                                 metrics.extend(collect_scanner_runtime_metrics(&stats));
                             }
@@ -2217,6 +2274,9 @@ pub fn init_metrics_runtime(token: CancellationToken) {
                             for key in retire_scanner_bucket_drive_result_keys {
                                 let _ = retire_scanner_bucket_drive_result_metric_series(&key);
                             }
+                            for key in retire_scanner_active_bucket_drive_keys {
+                                let _ = retire_scanner_active_bucket_drive_metric_series(&key);
+                            }
                         },
                     ).await;
                 }
@@ -2495,6 +2555,7 @@ fn collect_system_monitoring_metrics(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::metrics::collectors::scanner::ScannerActiveBucketDriveStats;
     use std::collections::{HashMap, HashSet};
     use std::time::Duration;
     use tokio::time::Instant;
@@ -2723,6 +2784,30 @@ mod tests {
         assert!(current.contains(&("server-a".to_string(), "logs".to_string(), "/data1".to_string(), "success".to_string(),)));
     }
 
+    #[test]
+    fn scanner_active_bucket_drive_keys_detect_completed_scans() {
+        let previous = scanner_active_bucket_drive_live_keys(&ScannerRuntimeStats {
+            server: "server-a".to_string(),
+            active_bucket_drive_scans: vec![ScannerActiveBucketDriveStats {
+                source: "usage".to_string(),
+                bucket: "photos".to_string(),
+                drive: "/data1".to_string(),
+                count: 1,
+                age_seconds: 3,
+            }],
+            ..Default::default()
+        });
+        let current = scanner_active_bucket_drive_live_keys(&ScannerRuntimeStats {
+            server: "server-a".to_string(),
+            ..Default::default()
+        });
+        assert!(
+            previous
+                .difference(&current)
+                .any(|key| key == &("server-a".to_string(), "usage".to_string(), "photos".to_string(), "/data1".to_string()))
+        );
+    }
+
     #[test]
     fn replication_proxy_bucket_keys_detect_removed_buckets() {
         let previous = repl_proxy_bucket_live_keys(&[BucketReplicationRuntimeStats {
diff --git a/crates/obs/src/metrics/schema/cluster_notification.rs b/crates/obs/src/metrics/schema/cluster_notification.rs
index 8be9a9807..1182fb884 100644
--- a/crates/obs/src/metrics/schema/cluster_notification.rs
+++ b/crates/obs/src/metrics/schema/cluster_notification.rs
@@ -15,6 +15,10 @@
 use crate::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems};
 use std::sync::LazyLock;
 
+pub const SERVER: &str = "server";
+
+const SERVER_LABELS: [&str; 1] = [SERVER];
+
 pub static NOTIFICATION_CURRENT_SEND_IN_PROGRESS_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
     new_gauge_md(
         MetricName::NotificationCurrentSendInProgress,
@@ -24,6 +28,15 @@ pub static NOTIFICATION_CURRENT_SEND_IN_PROGRESS_MD: LazyLock<MetricDescriptor>
     )
 });
 
+pub static NOTIFICATION_CURRENT_SEND_IN_PROGRESS_BY_SERVER_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_gauge_md(
+        MetricName::Custom("current_send_in_progress_by_server".to_string()),
+        "Number of concurrent async Send calls active to all targets by server",
+        &SERVER_LABELS,
+        subsystems::NOTIFICATION,
+    )
+});
+
 pub static NOTIFICATION_EVENTS_ERRORS_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
     new_counter_md(
         MetricName::NotificationEventsErrorsTotal,
@@ -33,6 +46,15 @@ pub static NOTIFICATION_EVENTS_ERRORS_TOTAL_MD: LazyLock<MetricDescriptor> = Laz
     )
 });
 
+pub static NOTIFICATION_EVENTS_ERRORS_TOTAL_BY_SERVER_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_counter_md(
+        MetricName::Custom("events_errors_total_by_server".to_string()),
+        "Events that failed to be sent to the targets by server",
+        &SERVER_LABELS,
+        subsystems::NOTIFICATION,
+    )
+});
+
 pub static NOTIFICATION_EVENTS_SENT_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
     new_counter_md(
         MetricName::NotificationEventsSentTotal,
@@ -42,6 +64,15 @@ pub static NOTIFICATION_EVENTS_SENT_TOTAL_MD: LazyLock<MetricDescriptor> = LazyL
     )
 });
 
+pub static NOTIFICATION_EVENTS_SENT_TOTAL_BY_SERVER_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_counter_md(
+        MetricName::Custom("events_sent_total_by_server".to_string()),
+        "Total number of events sent to the targets by server",
+        &SERVER_LABELS,
+        subsystems::NOTIFICATION,
+    )
+});
+
 pub static NOTIFICATION_EVENTS_SKIPPED_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
     new_counter_md(
         MetricName::NotificationEventsSkippedTotal,
@@ -50,3 +81,12 @@ pub static NOTIFICATION_EVENTS_SKIPPED_TOTAL_MD: LazyLock<MetricDescriptor> = La
         subsystems::NOTIFICATION,
     )
 });
+
+pub static NOTIFICATION_EVENTS_SKIPPED_TOTAL_BY_SERVER_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_counter_md(
+        MetricName::Custom("events_skipped_total_by_server".to_string()),
+        "Notification dispatch attempts skipped before delivery by server",
+        &SERVER_LABELS,
+        subsystems::NOTIFICATION,
+    )
+});
diff --git a/crates/obs/src/metrics/schema/entry/metric_name.rs b/crates/obs/src/metrics/schema/entry/metric_name.rs
index d9ec407b3..228bc6d92 100644
--- a/crates/obs/src/metrics/schema/entry/metric_name.rs
+++ b/crates/obs/src/metrics/schema/entry/metric_name.rs
@@ -371,6 +371,8 @@ pub enum MetricName {
     DriveWaitingIO,
     DriveAPILatencyMicros,
     DriveHealth,
+    DriveWritesTotal,
+    DriveDeletesTotal,
 
     DriveOfflineCount,
     DriveOnlineCount,
@@ -780,6 +782,8 @@ impl MetricName {
             Self::DriveWaitingIO => "waiting_io".to_string(),
             Self::DriveAPILatencyMicros => "api_latency_micros".to_string(),
             Self::DriveHealth => "health".to_string(),
+            Self::DriveWritesTotal => "writes_total".to_string(),
+            Self::DriveDeletesTotal => "deletes_total".to_string(),
 
             Self::DriveOfflineCount => "offline_count".to_string(),
             Self::DriveOnlineCount => "online_count".to_string(),
diff --git a/crates/obs/src/metrics/schema/ilm.rs b/crates/obs/src/metrics/schema/ilm.rs
index ac0538654..e38bcd862 100644
--- a/crates/obs/src/metrics/schema/ilm.rs
+++ b/crates/obs/src/metrics/schema/ilm.rs
@@ -18,6 +18,10 @@ use std::sync::LazyLock;
 pub const SERVER_LABEL: &str = "server";
 pub const ACTION_LABEL: &str = "action";
 pub const STATE_LABEL: &str = "state";
+pub const QUEUE_STATE_LABEL: &str = "queue_state";
+pub const RESULT_LABEL: &str = "result";
+pub const REASON_LABEL: &str = "reason";
+pub const SOURCE_LABEL: &str = "source";
 
 pub static ILM_ACTION_TASKS_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
     new_gauge_md(
@@ -28,6 +32,33 @@ pub static ILM_ACTION_TASKS_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
     )
 });
 
+pub static ILM_TASKS_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_gauge_md(
+        MetricName::Custom("tasks".to_string()),
+        "Current ILM task counts by server, action, and queue state",
+        &[SERVER_LABEL, ACTION_LABEL, QUEUE_STATE_LABEL],
+        subsystems::ILM,
+    )
+});
+
+pub static ILM_TASK_EVENTS_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_counter_md(
+        MetricName::Custom("task_events_total".to_string()),
+        "ILM task events by server, action, and result",
+        &[SERVER_LABEL, ACTION_LABEL, RESULT_LABEL],
+        subsystems::ILM,
+    )
+});
+
+pub static ILM_QUEUE_BACKPRESSURE_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_counter_md(
+        MetricName::Custom("queue_backpressure_total".to_string()),
+        "ILM queue backpressure events by server, action, and reason",
+        &[SERVER_LABEL, ACTION_LABEL, REASON_LABEL],
+        subsystems::ILM,
+    )
+});
+
 pub static ILM_EXPIRY_PENDING_TASKS_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
     new_gauge_md(
         MetricName::IlmExpiryPendingTasks,
@@ -108,3 +139,12 @@ pub static ILM_VERSIONS_SCANNED_MD: LazyLock<MetricDescriptor> = LazyLock::new(|
         subsystems::ILM,
     )
 });
+
+pub static ILM_VERSIONS_SCANNED_BY_SERVER_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_counter_md(
+        MetricName::Custom("versions_scanned_by_server".to_string()),
+        "ILM lifecycle-checked object versions by server and source",
+        &[SERVER_LABEL, SOURCE_LABEL],
+        subsystems::ILM,
+    )
+});
diff --git a/crates/obs/src/metrics/schema/scanner.rs b/crates/obs/src/metrics/schema/scanner.rs
index 53cb9548b..3a498d9f4 100644
--- a/crates/obs/src/metrics/schema/scanner.rs
+++ b/crates/obs/src/metrics/schema/scanner.rs
@@ -59,6 +59,24 @@ pub static SCANNER_CYCLE_BUCKET_DRIVE_RESULT_MD: LazyLock<MetricDescriptor> = La
     )
 });
 
+pub static SCANNER_ACTIVE_BUCKET_DRIVE_SCANS_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_gauge_md(
+        MetricName::Custom("active_bucket_drive_scans".to_string()),
+        "Current active scanner bucket-drive scans by server, source, bucket, and drive",
+        &[SERVER_LABEL, SOURCE_LABEL, BUCKET_LABEL, DRIVE_LABEL],
+        subsystems::SCANNER,
+    )
+});
+
+pub static SCANNER_ACTIVE_BUCKET_DRIVE_SCAN_AGE_SECONDS_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_gauge_md(
+        MetricName::Custom("active_bucket_drive_scan_age_seconds".to_string()),
+        "Age of the oldest active scanner bucket-drive scan by server, source, bucket, and drive",
+        &[SERVER_LABEL, SOURCE_LABEL, BUCKET_LABEL, DRIVE_LABEL],
+        subsystems::SCANNER,
+    )
+});
+
 pub static SCANNER_BUCKET_SCANS_FINISHED_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
     new_counter_md(
         MetricName::ScannerBucketScansFinished,
diff --git a/crates/obs/src/metrics/schema/system_drive.rs b/crates/obs/src/metrics/schema/system_drive.rs
index e64ee0b9d..f67449281 100644
--- a/crates/obs/src/metrics/schema/system_drive.rs
+++ b/crates/obs/src/metrics/schema/system_drive.rs
@@ -259,6 +259,24 @@ pub static DRIVE_HEALTH_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
     )
 });
 
+pub static DRIVE_WRITES_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_counter_md(
+        MetricName::DriveWritesTotal,
+        "Total successful write operations on a drive",
+        &ALL_DRIVE_LABELS[..],
+        subsystems::SYSTEM_DRIVE,
+    )
+});
+
+pub static DRIVE_DELETES_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_counter_md(
+        MetricName::DriveDeletesTotal,
+        "Total successful delete operations on a drive",
+        &ALL_DRIVE_LABELS[..],
+        subsystems::SYSTEM_DRIVE,
+    )
+});
+
 pub static DRIVE_OFFLINE_COUNT_MD: LazyLock<MetricDescriptor> =
     LazyLock::new(|| new_gauge_md(MetricName::DriveOfflineCount, "Count of offline drives", &[], subsystems::SYSTEM_DRIVE));
 
diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs
index cf5e77025..e80e3a199 100644
--- a/crates/obs/src/metrics/stats_collector.rs
+++ b/crates/obs/src/metrics/stats_collector.rs
@@ -18,15 +18,15 @@
 //! RustFS internal sources (storage layer, bucket monitor, system info)
 //! and convert them to the Stats structs used by collectors.
 
-use crate::metrics::collectors::scanner::{ScannerBucketDriveResultStats, ScannerSourceWorkStats};
+use crate::metrics::collectors::scanner::{ScannerActiveBucketDriveStats, ScannerBucketDriveResultStats, ScannerSourceWorkStats};
 use crate::metrics::collectors::{
     ApiRequestMetricSupport, ApiRequestStats, BucketReplicationBacklogStats, BucketReplicationBandwidthStats,
     BucketReplicationRuntimeStats, BucketReplicationStats, BucketReplicationTargetBacklogStats, BucketReplicationTargetFlowStats,
     BucketReplicationTargetStats, BucketStats, BucketUsageStats, ClusterConfigStats, ClusterHealthStats, ClusterStats,
     ClusterUsageStats, CompressionClusterStats, CpuStats, DiskStats, DriveCountStats, DriveDetailedStats,
-    DriveRuntimeDetailedStats, ErasureSetStats, HostNetworkStats, IamStats, IlmActionTaskStats, IlmRuntimeStats, IlmStats,
-    MemoryStats, NetworkStats, ProcessStats, ProcessStatusType, ReplicationStats, ResourceStats, ScannerRuntimeStats,
-    ScannerStats,
+    DriveRuntimeDetailedStats, ErasureSetStats, HostNetworkStats, IamStats, IlmActionTaskStats, IlmBackpressureStats,
+    IlmQueueTaskStats, IlmRuntimeStats, IlmStats, IlmTaskEventStats, MemoryStats, NetworkStats, ProcessStats, ProcessStatusType,
+    ReplicationStats, ResourceStats, ScannerRuntimeStats, ScannerStats,
 };
 use crate::metrics::runtime_sources::{ObsIlmRuntimeSnapshot, bucket_monitor_handle, iam_metrics_snapshot, ilm_runtime_snapshot};
 use crate::metrics::{
@@ -38,7 +38,10 @@ use crate::metrics::{
 use crate::node_identity::current_local_node_identity;
 use jiff::Timestamp;
 use rustfs_common::heal_channel::HealScanMode;
-use rustfs_common::metrics::{ScannerBucketDriveResultSnapshot, ScannerMetricsReport, ScannerSourceWorkSnapshot, global_metrics};
+use rustfs_common::metrics::{
+    ScannerActiveBucketDriveSnapshot, ScannerBucketDriveResultSnapshot, ScannerMetricsReport, ScannerSourceWorkSnapshot,
+    global_metrics,
+};
 use rustfs_io_metrics::internode_metrics::global_internode_metrics;
 use rustfs_io_metrics::{
     ProcessResourceSnapshot, ProcessSampler, ProcessStatusSnapshot, ProcessSystemSnapshot, s3_op_metrics_snapshot,
@@ -835,6 +838,8 @@ pub(crate) async fn collect_disk_and_system_drive_runtime_stats()
                         drive_api_latency_micros(metrics.last_minute.values().map(|action| (action.count, action.acc_time)))
                     }),
                     health: if is_online { 1 } else { 0 },
+                    writes_total: disk.metrics.as_ref().map(|metrics| metrics.total_writes),
+                    deletes_total: disk.metrics.as_ref().map(|metrics| metrics.total_deletes),
                     reads_per_sec: None,
                     reads_kb_per_sec: None,
                     reads_await: None,
@@ -1275,6 +1280,110 @@ fn ilm_action_task_stats(ilm: &ObsIlmRuntimeSnapshot) -> Vec<IlmActionTaskStats>
     ]
 }
 
+fn ilm_queue_task_stats(metrics: &ScannerMetricsReport) -> Vec<IlmQueueTaskStats> {
+    let expiry = &metrics.lifecycle_expiry;
+    let transition = &metrics.lifecycle_transition;
+    vec![
+        IlmQueueTaskStats {
+            action: "expiry".to_string(),
+            state: "pending".to_string(),
+            value: expiry.current_queued,
+        },
+        IlmQueueTaskStats {
+            action: "expiry".to_string(),
+            state: "active".to_string(),
+            value: expiry.current_active,
+        },
+        IlmQueueTaskStats {
+            action: "transition".to_string(),
+            state: "pending".to_string(),
+            value: transition.current_queued,
+        },
+        IlmQueueTaskStats {
+            action: "transition".to_string(),
+            state: "active".to_string(),
+            value: transition.current_active,
+        },
+        IlmQueueTaskStats {
+            action: "transition".to_string(),
+            state: "compensation_running".to_string(),
+            value: transition.compensation_running,
+        },
+    ]
+}
+
+fn ilm_task_event_stats(metrics: &ScannerMetricsReport) -> Vec<IlmTaskEventStats> {
+    let expiry = &metrics.lifecycle_expiry;
+    let transition = &metrics.lifecycle_transition;
+    vec![
+        IlmTaskEventStats {
+            action: "expiry".to_string(),
+            result: "queued".to_string(),
+            value: expiry.scanner_queued,
+        },
+        IlmTaskEventStats {
+            action: "expiry".to_string(),
+            result: "missed".to_string(),
+            value: expiry.scanner_missed,
+        },
+        IlmTaskEventStats {
+            action: "expiry".to_string(),
+            result: "blocked".to_string(),
+            value: expiry.scanner_blocked,
+        },
+        IlmTaskEventStats {
+            action: "expiry".to_string(),
+            result: "not_enqueued".to_string(),
+            value: expiry.scanner_not_enqueued,
+        },
+        IlmTaskEventStats {
+            action: "expiry".to_string(),
+            result: "failed".to_string(),
+            value: expiry.delete_failed,
+        },
+        IlmTaskEventStats {
+            action: "transition".to_string(),
+            result: "queued".to_string(),
+            value: transition.scanner_queued,
+        },
+        IlmTaskEventStats {
+            action: "transition".to_string(),
+            result: "missed".to_string(),
+            value: transition.scanner_missed,
+        },
+        IlmTaskEventStats {
+            action: "transition".to_string(),
+            result: "completed".to_string(),
+            value: transition.completed,
+        },
+        IlmTaskEventStats {
+            action: "transition".to_string(),
+            result: "failed".to_string(),
+            value: transition.failed,
+        },
+    ]
+}
+
+fn ilm_backpressure_stats(metrics: &ScannerMetricsReport) -> Vec<IlmBackpressureStats> {
+    vec![
+        IlmBackpressureStats {
+            action: "expiry".to_string(),
+            reason: "queue_missed".to_string(),
+            value: metrics.lifecycle_expiry.queue_missed,
+        },
+        IlmBackpressureStats {
+            action: "transition".to_string(),
+            reason: "queue_full".to_string(),
+            value: metrics.lifecycle_transition.queue_full,
+        },
+        IlmBackpressureStats {
+            action: "transition".to_string(),
+            reason: "send_timeout".to_string(),
+            value: metrics.lifecycle_transition.queue_send_timeout,
+        },
+    ]
+}
+
 /// Collect ILM metrics from the current lifecycle runtime state.
 pub async fn collect_ilm_metric_stats() -> Option<IlmStats> {
     collect_ilm_runtime_metric_stats().await.map(|stats| stats.stats)
@@ -1288,6 +1397,10 @@ pub(crate) async fn collect_ilm_runtime_metric_stats() -> Option<IlmRuntimeStats
     Some(IlmRuntimeStats {
         server: current_local_node_identity(),
         action_tasks: ilm_action_task_stats(&ilm),
+        queue_tasks: ilm_queue_task_stats(&metrics),
+        task_events: ilm_task_event_stats(&metrics),
+        backpressure: ilm_backpressure_stats(&metrics),
+        versions_scanned,
         stats: IlmStats {
             expiry_pending_tasks: ilm.expiry_pending_tasks,
             transition_active_tasks: ilm.transition_active_tasks,
@@ -1377,6 +1490,27 @@ fn scanner_bucket_drive_result_stats(results: &[ScannerBucketDriveResultSnapshot
     stats
 }
 
+fn scanner_active_bucket_drive_stats(results: &[ScannerActiveBucketDriveSnapshot]) -> Vec<ScannerActiveBucketDriveStats> {
+    let mut stats = results
+        .iter()
+        .filter(|result| !result.source.is_empty() && !result.bucket.is_empty() && !result.drive.is_empty() && result.count > 0)
+        .map(|result| ScannerActiveBucketDriveStats {
+            source: result.source.clone(),
+            bucket: result.bucket.clone(),
+            drive: result.drive.clone(),
+            count: result.count,
+            age_seconds: result.age_seconds,
+        })
+        .collect::<Vec<_>>();
+    stats.sort_by(|left, right| {
+        left.source
+            .cmp(&right.source)
+            .then_with(|| left.bucket.cmp(&right.bucket))
+            .then_with(|| left.drive.cmp(&right.drive))
+    });
+    stats
+}
+
 pub async fn collect_scanner_metric_stats() -> Option<ScannerStats> {
     collect_scanner_runtime_metric_stats().await.map(|stats| stats.stats)
 }
@@ -1418,6 +1552,7 @@ pub(crate) async fn collect_scanner_runtime_metric_stats() -> Option<ScannerRunt
             &runtime_details.current_cycle_bucket_drive_results,
         ),
         last_cycle_bucket_drive_results: scanner_bucket_drive_result_stats(&runtime_details.last_cycle_bucket_drive_results),
+        active_bucket_drive_scans: scanner_active_bucket_drive_stats(&runtime_details.active_bucket_drive_scans),
         stats: ScannerStats {
             bucket_scans_finished,
             bucket_scans_started,
@@ -1984,6 +2119,72 @@ mod tests {
         assert_eq!(scanner_lifecycle_checked_versions(&report), 37);
     }
 
+    #[test]
+    fn ilm_detail_stats_keep_expiry_and_transition_results_separate() {
+        let report = ScannerMetricsReport {
+            lifecycle_expiry: rustfs_common::metrics::ScannerLifecycleExpirySnapshot {
+                current_queued: 2,
+                current_active: 1,
+                scanner_queued: 10,
+                scanner_missed: 3,
+                delete_failed: 4,
+                ..Default::default()
+            },
+            lifecycle_transition: rustfs_common::metrics::ScannerLifecycleTransitionSnapshot {
+                current_queued: 5,
+                current_active: 6,
+                queue_full: 7,
+                queue_send_timeout: 8,
+                scanner_queued: 11,
+                completed: 12,
+                failed: 13,
+                ..Default::default()
+            },
+            ..Default::default()
+        };
+
+        let queues = ilm_queue_task_stats(&report);
+        assert!(
+            queues
+                .iter()
+                .any(|task| task.action == "expiry" && task.state == "pending" && task.value == 2)
+        );
+        assert!(
+            queues
+                .iter()
+                .any(|task| task.action == "transition" && task.state == "active" && task.value == 6)
+        );
+
+        let events = ilm_task_event_stats(&report);
+        assert!(
+            events
+                .iter()
+                .any(|event| event.action == "expiry" && event.result == "failed" && event.value == 4)
+        );
+        assert!(
+            events
+                .iter()
+                .any(|event| event.action == "transition" && event.result == "completed" && event.value == 12)
+        );
+        assert!(
+            events
+                .iter()
+                .any(|event| event.action == "transition" && event.result == "failed" && event.value == 13)
+        );
+
+        let backpressure = ilm_backpressure_stats(&report);
+        assert!(
+            backpressure
+                .iter()
+                .any(|event| event.action == "transition" && event.reason == "queue_full" && event.value == 7)
+        );
+        assert!(
+            backpressure
+                .iter()
+                .any(|event| event.action == "transition" && event.reason == "send_timeout" && event.value == 8)
+        );
+    }
+
     #[test]
     fn scanner_source_work_stats_sorts_and_skips_empty_source() {
         let stats = scanner_source_work_stats(&[
diff --git a/crates/scanner/src/scanner_io/guards.rs b/crates/scanner/src/scanner_io/guards.rs
index 270348f2d..3c93f759e 100644
--- a/crates/scanner/src/scanner_io/guards.rs
+++ b/crates/scanner/src/scanner_io/guards.rs
@@ -147,11 +147,19 @@ impl Drop for DiskBucketScanActiveGuard {
 
 pub(super) struct BucketDriveFailureGuard {
     failed: bool,
+    source: rustfs_common::metrics::ScannerWorkSource,
+    bucket: String,
+    drive: String,
 }
 
 impl BucketDriveFailureGuard {
-    pub(super) fn new() -> Self {
-        Self { failed: true }
+    pub(super) fn new(source: rustfs_common::metrics::ScannerWorkSource, bucket: &str, drive: &str) -> Self {
+        Self {
+            failed: true,
+            source,
+            bucket: bucket.to_string(),
+            drive: drive.to_string(),
+        }
     }
 
     pub(super) fn mark_not_failed(&mut self) {
@@ -161,6 +169,7 @@ impl BucketDriveFailureGuard {
 
 impl Drop for BucketDriveFailureGuard {
     fn drop(&mut self) {
+        global_metrics().record_scan_bucket_drive_end(self.source, &self.bucket, &self.drive);
         if self.failed {
             global_metrics().record_scan_bucket_drive_failure();
         }
@@ -272,3 +281,28 @@ pub(super) fn record_set_scan_failure(first_err: &mut Option<Error>, err: Error)
 pub(super) fn scanner_task_join_error(stage: &str, err: tokio::task::JoinError) -> Error {
     Error::other(format!("{stage} task join failed: {err}"))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use rustfs_common::metrics::{ScannerWorkSource, global_metrics};
+
+    #[test]
+    fn bucket_drive_failure_guard_retires_active_scan_on_drop() {
+        let source = ScannerWorkSource::Usage;
+        let bucket = "__guard_active_lifecycle_test__";
+        let drive = "/__guard_active_lifecycle_test__";
+        global_metrics().record_scan_bucket_drive_start(source, bucket, drive);
+        {
+            let mut guard = BucketDriveFailureGuard::new(source, bucket, drive);
+            guard.mark_not_failed();
+        }
+        assert!(
+            !global_metrics()
+                .scanner_runtime_details_report()
+                .active_bucket_drive_scans
+                .iter()
+                .any(|active| active.source == source.as_str() && active.bucket == bucket && active.drive == drive)
+        );
+    }
+}
diff --git a/crates/scanner/src/scanner_io/io_disk.rs b/crates/scanner/src/scanner_io/io_disk.rs
index ab5dee51f..560af841c 100644
--- a/crates/scanner/src/scanner_io/io_disk.rs
+++ b/crates/scanner/src/scanner_io/io_disk.rs
@@ -147,8 +147,12 @@ impl ScannerIODisk for Disk {
         let drive_start = std::time::Instant::now();
         let bucket = cache.info.name.clone();
         let disk_path = self.path().to_string_lossy().to_string();
-        global_metrics().record_scan_bucket_drive_start();
-        let mut failure_guard = BucketDriveFailureGuard::new();
+        let source = match scan_mode {
+            HealScanMode::Deep => rustfs_common::metrics::ScannerWorkSource::Bitrot,
+            HealScanMode::Normal | HealScanMode::Unknown => rustfs_common::metrics::ScannerWorkSource::Usage,
+        };
+        global_metrics().record_scan_bucket_drive_start(source, &bucket, &disk_path);
+        let mut failure_guard = BucketDriveFailureGuard::new(source, &bucket, &disk_path);
         let _guard = self.start_scan();
 
         let mut cache = cache;
@@ -196,32 +200,32 @@ impl ScannerIODisk for Disk {
         match result {
             Ok(mut data_usage_info) => {
                 done_drive();
-                emit_scan_bucket_drive_complete(true, &bucket, &disk_path, drive_start.elapsed());
+                emit_scan_bucket_drive_complete(source, true, &bucket, &disk_path, drive_start.elapsed());
                 data_usage_info.info.last_update = Some(SystemTime::now());
                 failure_guard.mark_not_failed();
                 Ok(ScannerDiskScanOutcome::Complete(data_usage_info))
             }
             Err(ScannerError::PartialCache(mut partial_cache)) => {
                 done_drive();
-                emit_scan_bucket_drive_partial(&bucket, &disk_path, drive_start.elapsed());
+                emit_scan_bucket_drive_partial(source, &bucket, &disk_path, drive_start.elapsed());
                 partial_cache.info.last_update.get_or_insert_with(SystemTime::now);
                 failure_guard.mark_not_failed();
                 Ok(ScannerDiskScanOutcome::Partial(*partial_cache))
             }
             Err(ScannerError::NamespaceNotFoundCache(mut partial_cache)) => {
                 done_drive();
-                emit_scan_bucket_drive_partial(&bucket, &disk_path, drive_start.elapsed());
+                emit_scan_bucket_drive_partial(source, &bucket, &disk_path, drive_start.elapsed());
                 partial_cache.info.last_update.get_or_insert_with(SystemTime::now);
                 failure_guard.mark_not_failed();
                 Ok(ScannerDiskScanOutcome::NamespaceNotFound(*partial_cache))
             }
             Err(e) => {
                 if ctx.is_cancelled() {
-                    emit_scan_bucket_drive_partial(&bucket, &disk_path, drive_start.elapsed());
+                    emit_scan_bucket_drive_partial(source, &bucket, &disk_path, drive_start.elapsed());
                     failure_guard.mark_not_failed();
                 } else {
                     done_drive();
-                    emit_scan_bucket_drive_complete(false, &bucket, &disk_path, drive_start.elapsed());
+                    emit_scan_bucket_drive_complete(source, false, &bucket, &disk_path, drive_start.elapsed());
                 }
                 Err(StorageError::other(format!("Failed to scan data folder: {e}")))
             }