fix(error): merge equivalent api message branches

Combine the MaxVersionsExceeded and internal IO message branches so Clippy no longer flags identical if blocks while preserving the existing response messages.

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

crates/heal/src/heal/manager.rs

@@ -20,7 +20,7 @@ use crate::heal::{
     task::{HealOptions, HealPriority, HealRequest, HealTask, HealTaskStatus, HealType, demote_to_debug_when},
 };
 use crate::{Error, Result};
-use metrics::{counter, gauge};
+use metrics::{counter, gauge, histogram};
 use rustfs_concurrency::WorkloadAdmissionSnapshotProvider;
 use rustfs_concurrency::workload::{ForegroundPressure, foreground_pressure};
 #[cfg(test)]
@@ -34,7 +34,7 @@ use std::sync::LazyLock;
 use std::{
     collections::{BinaryHeap, HashMap, HashSet},
     sync::{Arc, Mutex as StdMutex, MutexGuard as StdMutexGuard},
-    time::{Duration, SystemTime},
+    time::{Duration, Instant, SystemTime},
 };
 use tokio::{
     sync::{Mutex, Notify, RwLock},
@@ -181,6 +181,13 @@ fn lock_displaced_terminals(
     }
 }
 
+fn lock_admission_telemetry(registry: &StdMutex<HealAdmissionTelemetry>) -> StdMutexGuard<'_, HealAdmissionTelemetry> {
+    match registry.lock() {
+        Ok(guard) => guard,
+        Err(poisoned) => poisoned.into_inner(),
+    }
+}
+
 fn record_displaced_terminal(
     registry: &StdMutex<HashMap<String, Arc<CompletedHealStatus>>>,
     request: &HealRequest,
@@ -384,6 +391,61 @@ impl HealSourceCounts {
     }
 }
 
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub struct HealAdmissionTelemetry {
+    pub accepted: u64,
+    pub merged: u64,
+    pub full: u64,
+    pub dropped: u64,
+    pub duplicate: u64,
+    pub overlap_rejected: u64,
+    pub displaced: u64,
+    pub force_start: u64,
+    pub max_start_duration_micros: u64,
+    pub max_lock_phase_micros: u64,
+}
+
+impl HealAdmissionTelemetry {
+    fn record(&mut self, observation: HealAdmissionObservation) {
+        match observation.result {
+            HealAdmissionResult::Accepted => self.accepted = self.accepted.saturating_add(1),
+            HealAdmissionResult::Merged => self.merged = self.merged.saturating_add(1),
+            HealAdmissionResult::Full => self.full = self.full.saturating_add(1),
+            HealAdmissionResult::Dropped(_) => self.dropped = self.dropped.saturating_add(1),
+        }
+        if observation.context == "duplicate" {
+            self.duplicate = self.duplicate.saturating_add(1);
+        }
+        if observation.context == "overlap_rejected" {
+            self.overlap_rejected = self.overlap_rejected.saturating_add(1);
+        }
+        if observation.displaced {
+            self.displaced = self.displaced.saturating_add(1);
+        }
+        if observation.force_start {
+            self.force_start = self.force_start.saturating_add(1);
+        }
+        self.max_start_duration_micros = self
+            .max_start_duration_micros
+            .max(duration_micros_saturated(observation.start_duration));
+        self.max_lock_phase_micros = self
+            .max_lock_phase_micros
+            .max(duration_micros_saturated(observation.lock_phase));
+    }
+}
+
+#[derive(Debug, Clone, Copy)]
+struct HealAdmissionObservation {
+    source: HealRequestSource,
+    result: HealAdmissionResult,
+    context: &'static str,
+    force_start: bool,
+    displaced: bool,
+    start_duration: Duration,
+    lock_phase: Duration,
+}
+
 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
 #[serde(rename_all = "camelCase", deny_unknown_fields)]
 pub struct HealOperationsSnapshot {
@@ -396,12 +458,18 @@ pub struct HealOperationsSnapshot {
     pub queued_by_source: HealSourceCounts,
     pub active_by_source: HealSourceCounts,
     pub retrying_by_source: HealSourceCounts,
+    #[serde(default)]
+    pub admission: HealAdmissionTelemetry,
 }
 
 fn usize_to_u64_saturated(value: usize) -> u64 {
     u64::try_from(value).unwrap_or(u64::MAX)
 }
 
+fn duration_micros_saturated(duration: Duration) -> u64 {
+    u64::try_from(duration.as_micros()).unwrap_or(u64::MAX)
+}
+
 fn heal_type_matches_path(heal_type: &HealType, heal_path: &str) -> bool {
     let heal_path = heal_path.trim_matches('/');
     if heal_path.is_empty() || heal_path == LEGACY_ROOT_HEAL_PATH {
@@ -764,6 +832,9 @@ pub struct HealManager {
     notify: Arc<Notify>,
     /// Optional runtime workload snapshot provider used to protect foreground data-plane work.
     workload_provider: Option<WorkloadSnapshotProviderRef>,
+    /// Bounded, low-cardinality admission telemetry exposed through the
+    /// existing operations snapshot for cluster E2E assertions.
+    admission_telemetry: Arc<StdMutex<HealAdmissionTelemetry>>,
 }
 
 /// Where a task-id lookup resolved. The variants carry the resolved state
@@ -919,6 +990,33 @@ impl HealManager {
         .increment(1);
     }
 
+    fn record_admission_observation(&self, observation: HealAdmissionObservation) {
+        let result = observation.result.result_label().to_string();
+        let reason = observation.result.reason_label().to_string();
+        let source = observation.source.as_str().to_string();
+        let context = observation.context.to_string();
+        let force_start = observation.force_start.to_string();
+        histogram!(
+            "rustfs_heal_admission_start_duration_seconds",
+            "source" => source.clone(),
+            "result" => result.clone(),
+            "reason" => reason.clone(),
+            "context" => context.clone(),
+            "force_start" => force_start.clone()
+        )
+        .record(observation.start_duration.as_secs_f64());
+        histogram!(
+            "rustfs_heal_admission_lock_phase_seconds",
+            "source" => source,
+            "result" => result,
+            "reason" => reason,
+            "context" => context,
+            "force_start" => force_start
+        )
+        .record(observation.lock_phase.as_secs_f64());
+        lock_admission_telemetry(&self.admission_telemetry).record(observation);
+    }
+
     fn remove_mrf_repair_notice_targets_for_task(&self, task_id: &str) {
         let targets = lock_mrf_repair_notice_targets(&self.mrf_repair_notice_targets).remove(task_id);
         if let Some(targets) = targets {
@@ -1265,6 +1363,7 @@ impl HealManager {
             statistics: Arc::new(RwLock::new(HealStatistics::new())),
             notify: Arc::new(Notify::new()),
             workload_provider,
+            admission_telemetry: Arc::new(StdMutex::new(HealAdmissionTelemetry::default())),
         }
     }
 
@@ -1455,6 +1554,9 @@ impl HealManager {
         preserve_alias: bool,
         mrf_notice_target: Option<MrfRepairNoticeTarget>,
     ) -> Result<HealAdmissionReceipt> {
+        let admission_start = Instant::now();
+        let source = request.source;
+        let force_start = request.force_start;
         // HS-06 forceStart semantics (admin only): MinIO stops the old task
         // first and then starts the new one. Cancel any active admin task
         // overlapping this request's path before entering admission, so the
@@ -1505,6 +1607,7 @@ impl HealManager {
         // Match the scheduler's active -> queue order and keep retry ownership
         // in the same atomic view. Otherwise queue -> active and
         // active -> retrying transitions can slip between duplicate checks.
+        let lock_phase_start = Instant::now();
         let active_heals = self.active_heals.lock().await;
         #[cfg(test)]
         pause_duplicate_admission_after_active_lock(&request.id).await;
@@ -1539,7 +1642,17 @@ impl HealManager {
             drop(retrying_heals);
             drop(queue);
             drop(active_heals);
+            let lock_phase = lock_phase_start.elapsed();
             Self::record_admission_metric(request.source, admission, "duplicate");
+            self.record_admission_observation(HealAdmissionObservation {
+                source,
+                result: admission,
+                context: "duplicate",
+                force_start,
+                displaced: false,
+                start_duration: admission_start.elapsed(),
+                lock_phase,
+            });
 
             match admission {
                 HealAdmissionResult::Merged => {
@@ -1618,7 +1731,17 @@ impl HealManager {
                 drop(retrying_heals);
                 drop(queue);
                 drop(active_heals);
+                let lock_phase = lock_phase_start.elapsed();
                 Self::record_admission_metric(request.source, HealAdmissionResult::Dropped(reason), "overlap_rejected");
+                self.record_admission_observation(HealAdmissionObservation {
+                    source,
+                    result: HealAdmissionResult::Dropped(reason),
+                    context: "overlap_rejected",
+                    force_start,
+                    displaced: false,
+                    start_duration: admission_start.elapsed(),
+                    lock_phase,
+                });
                 warn!(
                     target: "rustfs::heal::manager",
                     event = EVENT_HEAL_QUEUE_ADMISSION,
@@ -1663,6 +1786,8 @@ impl HealManager {
         drop(retrying_heals);
         drop(queue);
         drop(active_heals);
+        let lock_phase = lock_phase_start.elapsed();
+        let displaced = displaced_terminal.is_some();
 
         if let (Some(displaced_task_id), Some(displaced_terminal)) = (displaced_task_id, displaced_terminal) {
             // The queue has already removed the displaced request, so the
@@ -1676,6 +1801,16 @@ impl HealManager {
             self.notify.notify_one();
         }
 
+        self.record_admission_observation(HealAdmissionObservation {
+            source,
+            result: admission,
+            context: "submit",
+            force_start,
+            displaced,
+            start_duration: admission_start.elapsed(),
+            lock_phase,
+        });
+
         Ok(HealAdmissionReceipt {
             result: admission,
             task_id,
@@ -2111,17 +2246,25 @@ impl HealManager {
         }
         publish_active_heal_count(&active_heals);
         publish_heal_queue_length(&queue);
+        let queue_length = usize_to_u64_saturated(queue.len());
+        let active_tasks = usize_to_u64_saturated(active_heals.len());
+        let retrying_tasks = usize_to_u64_saturated(retrying_heals.len());
+        drop(retrying_heals);
+        drop(queue);
+        drop(active_heals);
+        let admission = *lock_admission_telemetry(&self.admission_telemetry);
 
         HealOperationsSnapshot {
-            queue_length: usize_to_u64_saturated(queue.len()),
-            active_tasks: usize_to_u64_saturated(active_heals.len()),
-            retrying_tasks: usize_to_u64_saturated(retrying_heals.len()),
+            queue_length,
+            active_tasks,
+            retrying_tasks,
             queued_by_priority,
             active_by_priority,
             retrying_by_priority,
             queued_by_source,
             active_by_source,
             retrying_by_source,
+            admission,
         }
     }
 

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

@@ -2792,6 +2792,88 @@ async fn admin_force_start_cancels_overlapping_active_task_first() {
     );
 }
 
+#[tokio::test]
+async fn admission_snapshot_tracks_start_duplicate_force_start_and_displacement() {
+    let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+    let manager = Arc::new(HealManager::new(
+        storage,
+        Some(HealConfig {
+            queue_size: 1,
+            ..Default::default()
+        }),
+    ));
+
+    let mut paused = admin_prefix_request("bucket-a", "logs/");
+    paused.priority = HealPriority::Low;
+    let hook = Arc::new(DuplicateAdmissionTestHook {
+        request_id: paused.id.clone(),
+        active_lock_reached: Notify::new(),
+        active_lock_release: Notify::new(),
+    });
+    *DUPLICATE_ADMISSION_TEST_HOOK.lock().await = Some(hook.clone());
+
+    let submit_manager = Arc::clone(&manager);
+    let mut paused_submission = tokio::spawn(async move { submit_manager.submit_heal_request(paused).await });
+    tokio::time::timeout(Duration::from_secs(1), hook.active_lock_reached.notified())
+        .await
+        .expect("admission should reach the test-only lock phase hook");
+    assert!(
+        tokio::time::timeout(Duration::from_millis(10), &mut paused_submission)
+            .await
+            .is_err(),
+        "admission must wait while the lock-phase hook is held"
+    );
+    hook.active_lock_release.notify_one();
+    assert_eq!(
+        paused_submission
+            .await
+            .expect("paused admission task should join")
+            .expect("paused admission should succeed"),
+        HealAdmissionResult::Accepted
+    );
+    *DUPLICATE_ADMISSION_TEST_HOOK.lock().await = None;
+
+    let duplicate = admin_prefix_request("bucket-a", "logs/");
+    let duplicate_receipt = manager
+        .submit_heal_request_with_receipt(duplicate)
+        .await
+        .expect("duplicate admission should return a canonical receipt");
+    assert_eq!(duplicate_receipt.result, HealAdmissionResult::Merged);
+
+    let mut high = admin_prefix_request("bucket-b", "logs/");
+    high.priority = HealPriority::High;
+    assert_eq!(
+        manager
+            .submit_heal_request(high)
+            .await
+            .expect("higher priority admin request should displace queued low-priority work"),
+        HealAdmissionResult::Accepted
+    );
+
+    let mut forced = admin_prefix_request("bucket-c", "logs/");
+    forced.force_start = true;
+    assert_eq!(
+        manager
+            .submit_heal_request(forced)
+            .await
+            .expect("forceStart should keep explicit admission semantics"),
+        HealAdmissionResult::Accepted
+    );
+
+    let admission = manager.operations_snapshot().await.admission;
+    assert_eq!(admission.accepted, 3);
+    assert_eq!(admission.merged, 1);
+    assert_eq!(admission.full, 0);
+    assert_eq!(admission.dropped, 0);
+    assert_eq!(admission.duplicate, 1);
+    assert_eq!(admission.displaced, 1);
+    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"
+    );
+}
+
 #[tokio::test]
 async fn test_operations_snapshot_counts_active_by_source_and_priority() {
     let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);

crates/heal/src/heal/mod.rs

@@ -34,7 +34,7 @@ use storage_api::owner::{
 };
 
 pub use erasure_healer::ErasureSetHealer;
-pub use manager::{HealManager, HealOperationsSnapshot, HealPriorityCounts, HealSourceCounts};
+pub use manager::{HealAdmissionTelemetry, HealManager, HealOperationsSnapshot, HealPriorityCounts, HealSourceCounts};
 pub use resume::{CheckpointManager, ResumeCheckpoint, ResumeManager, ResumeState, ResumeUtils};
 pub use task::{HealOptions, HealPriority, HealRequest, HealTask, HealType};
 

crates/heal/src/lib.rs

@@ -19,7 +19,8 @@ pub mod heal;
 
 pub use error::{Error, Result};
 pub use heal::{
-    HealManager, HealOperationsSnapshot, HealOptions, HealPriority, HealPriorityCounts, HealRequest, HealSourceCounts, HealType,
+    HealAdmissionTelemetry, HealManager, HealOperationsSnapshot, HealOptions, HealPriority, HealPriorityCounts, HealRequest,
+    HealSourceCounts, HealType,
     channel::HealChannelProcessor,
     progress::{HealProgress, aggregate_heal_progress},
     resume::{ReplacementRecoveryRecord, ReplacementRecoveryState, ResumeUtils},

rustfs/src/admin/handlers/heal.rs

@@ -339,6 +339,19 @@ fn add_source_counts(total: &mut rustfs_heal::HealSourceCounts, next: rustfs_hea
     total.mrf = total.mrf.saturating_add(next.mrf);
 }
 
+fn add_admission_telemetry(total: &mut rustfs_heal::HealAdmissionTelemetry, next: rustfs_heal::HealAdmissionTelemetry) {
+    total.accepted = total.accepted.saturating_add(next.accepted);
+    total.merged = total.merged.saturating_add(next.merged);
+    total.full = total.full.saturating_add(next.full);
+    total.dropped = total.dropped.saturating_add(next.dropped);
+    total.duplicate = total.duplicate.saturating_add(next.duplicate);
+    total.overlap_rejected = total.overlap_rejected.saturating_add(next.overlap_rejected);
+    total.displaced = total.displaced.saturating_add(next.displaced);
+    total.force_start = total.force_start.saturating_add(next.force_start);
+    total.max_start_duration_micros = total.max_start_duration_micros.max(next.max_start_duration_micros);
+    total.max_lock_phase_micros = total.max_lock_phase_micros.max(next.max_lock_phase_micros);
+}
+
 fn add_operations(total: &mut rustfs_heal::HealOperationsSnapshot, next: rustfs_heal::HealOperationsSnapshot) {
     total.queue_length = total.queue_length.saturating_add(next.queue_length);
     total.active_tasks = total.active_tasks.saturating_add(next.active_tasks);
@@ -349,6 +362,7 @@ fn add_operations(total: &mut rustfs_heal::HealOperationsSnapshot, next: rustfs_
     add_source_counts(&mut total.queued_by_source, next.queued_by_source);
     add_source_counts(&mut total.active_by_source, next.active_by_source);
     add_source_counts(&mut total.retrying_by_source, next.retrying_by_source);
+    add_admission_telemetry(&mut total.admission, next.admission);
 }
 
 fn aggregate_cluster_heal_status(snapshots: Vec<NodeHealStatusSnapshot>) -> ClusterHealStatusSnapshot {
@@ -2307,6 +2321,10 @@ mod tests {
         assert!(json["healOperations"]["queuedBySource"]["admin"].is_u64());
         assert!(json["healOperations"]["queuedByPriority"]["low"].is_u64());
         assert!(json["healOperations"]["queuedByPriority"]["high"].is_u64());
+        assert!(json["healOperations"]["admission"]["accepted"].is_u64());
+        assert!(json["healOperations"]["admission"]["duplicate"].is_u64());
+        assert!(json["healOperations"]["admission"]["forceStart"].is_u64());
+        assert!(json["healOperations"]["admission"]["maxLockPhaseMicros"].is_u64());
         assert_eq!(json["state"], "active");
         assert_eq!(json["clusterStatusComplete"], true);
         assert!(json["progress"].is_null());
@@ -2486,6 +2504,18 @@ mod tests {
             queued_by_source: sources(value),
             active_by_source: sources(value),
             retrying_by_source: sources(value),
+            admission: rustfs_heal::HealAdmissionTelemetry {
+                accepted: value,
+                merged: value,
+                full: value,
+                dropped: value,
+                duplicate: value,
+                overlap_rejected: value,
+                displaced: value,
+                force_start: value,
+                max_start_duration_micros: value,
+                max_lock_phase_micros: value,
+            },
         };
         let progress = |value| NodeHealProgress {
             objects_scanned: value,

rustfs/src/error.rs

@@ -496,9 +496,9 @@ impl From<StorageError> for ApiError {
 
         let message = if matches!(&err, StorageError::QuotaExceeded { .. }) {
             err.to_string()
-        } else if matches!(&err, StorageError::MaxVersionsExceeded) {
-            ApiError::error_code_to_message(&code)
-        } else if code == S3ErrorCode::InternalError && matches!(&err, StorageError::Io(_)) {
+        } else if matches!(&err, StorageError::MaxVersionsExceeded)
+            || (code == S3ErrorCode::InternalError && matches!(&err, StorageError::Io(_)))
+        {
             ApiError::error_code_to_message(&code)
         } else if code == S3ErrorCode::InternalError {
             err.to_string()

rustfs/src/storage/rpc/node_service/heal.rs

@@ -726,6 +726,7 @@ mod tests {
         assert_eq!(decoded.info().bitrot_start_cycle, 9);
         assert_eq!(decoded.operations.queue_length, 2);
         assert_eq!(decoded.operations.queued_by_source.mrf, 0);
+        assert_eq!(decoded.operations.admission, rustfs_heal::HealAdmissionTelemetry::default());
         let progress = decoded.progress.expect("legacy progress should decode");
         assert_eq!(progress.objects_scanned, 7);
         assert!(!progress.baseline_known);