fix(heal): correct progress accounting (#6382)

* fix(heal): correct progress accounting

* fix(heal): atomically persist page progress

* fix(heal): preserve terminal progress counters

* fix(heal): make resume handoff crash safe

* fix(heal): preserve resumable bucket checkpoints

* fix(heal): satisfy checkpoint outcome lint

* fix(heal): preserve progress status across nodes

* fix(heal): stabilize progress generations

* style: restore rebalance formatting

* test(heal): cover cross-set baseline generation

---------

Signed-off-by: houseme <housemecn@gmail.com>
Co-authored-by: overtrue <anzhengchao@gmail.com>
Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: heihutu <heihutu@gmail.com>

rustfs/src/admin/handlers/heal.rs

@@ -20,7 +20,7 @@ use crate::admin::storage_api::bucket::utils::is_valid_object_prefix;
 use crate::server::ADMIN_PREFIX;
 use crate::server::RemoteAddr;
 use crate::storage::rpc::node_service::heal::{
-    HealControlCoordinator, NodeHealProgress, NodeHealStatusSnapshot, capture_node_heal_status, decode_node_heal_status,
+    HealControlCoordinator, NodeHealStatusSnapshot, capture_node_heal_status, decode_node_heal_status,
     decode_node_replacement_recovery_status, heal_control_coordinator, heal_topology_fingerprint,
 };
 use bytes::Bytes;
@@ -298,14 +298,7 @@ fn background_heal_runtime_state(
     }
 }
 
-#[derive(Debug, Serialize)]
-#[serde(rename_all = "camelCase")]
-struct BackgroundHealProgress {
-    objects_scanned: u64,
-    objects_healed: u64,
-    objects_failed: u64,
-    bytes_processed: u64,
-}
+type BackgroundHealProgress = rustfs_heal::HealProgress;
 
 #[derive(Debug)]
 struct ClusterHealStatusSnapshot {
@@ -344,17 +337,10 @@ fn add_operations(total: &mut rustfs_heal::HealOperationsSnapshot, next: rustfs_
     add_source_counts(&mut total.retrying_by_source, next.retrying_by_source);
 }
 
-fn add_progress(total: &mut BackgroundHealProgress, next: NodeHealProgress) {
-    total.objects_scanned = total.objects_scanned.saturating_add(next.objects_scanned);
-    total.objects_healed = total.objects_healed.saturating_add(next.objects_healed);
-    total.objects_failed = total.objects_failed.saturating_add(next.objects_failed);
-    total.bytes_processed = total.bytes_processed.saturating_add(next.bytes_processed);
-}
-
 fn aggregate_cluster_heal_status(snapshots: Vec<NodeHealStatusSnapshot>) -> ClusterHealStatusSnapshot {
     let mut info = BackgroundHealInfo::default();
     let mut operations = rustfs_heal::HealOperationsSnapshot::default();
-    let mut progress = None;
+    let mut progress = Vec::new();
     let mut any_services_enabled = false;
     let mut any_initialized = false;
 
@@ -371,18 +357,12 @@ fn aggregate_cluster_heal_status(snapshots: Vec<NodeHealStatusSnapshot>) -> Clus
         }
         add_operations(&mut operations, snapshot.operations);
         if let Some(next) = snapshot.progress {
-            add_progress(
-                progress.get_or_insert(BackgroundHealProgress {
-                    objects_scanned: 0,
-                    objects_healed: 0,
-                    objects_failed: 0,
-                    bytes_processed: 0,
-                }),
-                next,
-            );
+            progress.push(next);
         }
     }
 
+    let progress = rustfs_heal::aggregate_heal_progress(progress);
+
     let state = if operations.queue_length > 0 || operations.active_tasks > 0 || operations.retrying_tasks > 0 {
         HealRuntimeState::Active
     } else if any_initialized {
@@ -2201,10 +2181,19 @@ mod tests {
         };
 
         let progress = BackgroundHealProgress {
+            kind: rustfs_heal::heal::progress::HealProgressKind::ObjectSweep,
             objects_scanned: 7,
             objects_healed: 3,
             objects_failed: 1,
+            skipped_objects: 3,
+            objects_total_count: 10,
+            objects_total_size: 8192,
             bytes_processed: 4096,
+            progress_percentage: 50.0,
+            progress_state: rustfs_heal::heal::progress::HealProgressState::Running,
+            baseline_generation: Some(42),
+            baseline_known: true,
+            ..Default::default()
         };
 
         let encoded = encode_background_heal_status(
@@ -2220,7 +2209,14 @@ mod tests {
         assert_eq!(json["progress"]["objectsScanned"], 7);
         assert_eq!(json["progress"]["objectsHealed"], 3);
         assert_eq!(json["progress"]["objectsFailed"], 1);
+        assert_eq!(json["progress"]["skippedObjects"], 3);
+        assert_eq!(json["progress"]["objectsTotalCount"], 10);
+        assert_eq!(json["progress"]["objectsTotalSize"], 8192);
         assert_eq!(json["progress"]["bytesProcessed"], 4096);
+        assert_eq!(json["progress"]["progressState"], "running");
+        assert_eq!(json["progress"]["baselineGeneration"], 42);
+        assert_eq!(json["progress"]["baselineKnown"], true);
+        assert_eq!(json["progress"]["counterUnknown"], false);
     }
 
     #[test]
@@ -2242,10 +2238,18 @@ mod tests {
                 ..Default::default()
             },
             Some(NodeHealProgress {
+                kind: rustfs_heal::heal::progress::HealProgressKind::ObjectSweep,
                 objects_scanned: 3,
                 objects_healed: 1,
                 objects_failed: 0,
+                skipped_objects: 2,
+                objects_total_count: 6,
+                objects_total_size: 400,
                 bytes_processed: 100,
+                progress_state: rustfs_heal::heal::progress::HealProgressState::Running,
+                baseline_generation: Some(9),
+                baseline_known: true,
+                ..Default::default()
             }),
         );
         let peer = NodeHealStatusSnapshot::for_test(
@@ -2265,10 +2269,17 @@ mod tests {
                 ..Default::default()
             },
             Some(NodeHealProgress {
+                kind: rustfs_heal::heal::progress::HealProgressKind::ObjectSweep,
                 objects_scanned: 5,
                 objects_healed: 4,
                 objects_failed: 1,
+                objects_total_count: 4,
+                objects_total_size: 1600,
                 bytes_processed: 900,
+                progress_state: rustfs_heal::heal::progress::HealProgressState::Running,
+                baseline_generation: Some(9),
+                baseline_known: true,
+                ..Default::default()
             }),
         );
 
@@ -2287,7 +2298,15 @@ mod tests {
         assert_eq!(progress.objects_scanned, 8);
         assert_eq!(progress.objects_healed, 5);
         assert_eq!(progress.objects_failed, 1);
+        assert_eq!(progress.skipped_objects, 2);
+        assert_eq!(progress.objects_total_count, 10);
+        assert_eq!(progress.objects_total_size, 2000);
         assert_eq!(progress.bytes_processed, 1000);
+        assert_eq!(progress.progress_percentage, 50.0);
+        assert_eq!(progress.progress_state, rustfs_heal::heal::progress::HealProgressState::Running);
+        assert_eq!(progress.baseline_generation, Some(9));
+        assert!(progress.baseline_known);
+        assert!(!progress.counter_unknown);
 
         assert_eq!(peer_first.state, HealRuntimeState::Active);
         assert_eq!(peer_first.operations, local_first.operations);
@@ -2327,6 +2346,7 @@ mod tests {
             objects_healed: value,
             objects_failed: value,
             bytes_processed: value,
+            ..Default::default()
         };
         let saturated = NodeHealStatusSnapshot::for_test(
             true,

rustfs/src/storage/rpc/node_service.rs

@@ -1912,7 +1912,7 @@ impl Node for NodeService {
 
     async fn background_heal_status(
         &self,
-        _request: Request<BackgroundHealStatusRequest>,
+        request: Request<BackgroundHealStatusRequest>,
     ) -> Result<Response<BackgroundHealStatusResponse>, Status> {
         if self.resolve_object_store().is_none() {
             return Ok(Response::new(BackgroundHealStatusResponse {
@@ -1922,7 +1922,7 @@ impl Node for NodeService {
             }));
         }
         let snapshot = heal::capture_node_heal_status(rustfs_scanner::scanner::BackgroundHealInfo::default()).await;
-        match heal::encode_node_heal_status(&snapshot) {
+        match heal::encode_node_heal_status(&snapshot, request.into_inner().protocol_version) {
             Ok(bg_heal_state) => Ok(Response::new(BackgroundHealStatusResponse {
                 success: true,
                 bg_heal_state: bg_heal_state.into(),

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

@@ -25,7 +25,8 @@ use std::io::Cursor;
 
 use super::super::encode_msgpack_map;
 
-const NODE_HEAL_STATUS_VERSION: u8 = 1;
+const NODE_HEAL_STATUS_PREVIOUS_VERSION: u8 = 1;
+const NODE_HEAL_STATUS_VERSION: u8 = 2;
 const NODE_HEAL_STATUS_MAX_SIZE: usize = 64 * 1024;
 const NODE_REPLACEMENT_RECOVERY_STATUS_VERSION: u8 = 1;
 const NODE_REPLACEMENT_RECOVERY_STATUS_MAX_SIZE: usize = 64 * 1024;
@@ -172,13 +173,38 @@ pub(crate) fn heal_topology_fingerprint(endpoint_pools: &EndpointServerPools) ->
     Ok(hex_simd::encode_to_string(hasher.finalize(), hex_simd::AsciiCase::Lower))
 }
 
-#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub(crate) type NodeHealProgress = rustfs_heal::HealProgress;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase", deny_unknown_fields)]
-pub(crate) struct NodeHealProgress {
-    pub objects_scanned: u64,
-    pub objects_healed: u64,
-    pub objects_failed: u64,
-    pub bytes_processed: u64,
+struct NodeHealProgressV1 {
+    objects_scanned: u64,
+    objects_healed: u64,
+    objects_failed: u64,
+    bytes_processed: u64,
+}
+
+impl From<&NodeHealProgress> for NodeHealProgressV1 {
+    fn from(progress: &NodeHealProgress) -> Self {
+        Self {
+            objects_scanned: progress.objects_scanned,
+            objects_healed: progress.objects_healed,
+            objects_failed: progress.objects_failed,
+            bytes_processed: progress.bytes_processed,
+        }
+    }
+}
+
+impl From<NodeHealProgressV1> for NodeHealProgress {
+    fn from(progress: NodeHealProgressV1) -> Self {
+        Self {
+            objects_scanned: progress.objects_scanned,
+            objects_healed: progress.objects_healed,
+            objects_failed: progress.objects_failed,
+            bytes_processed: progress.bytes_processed,
+            ..Default::default()
+        }
+    }
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -220,6 +246,48 @@ pub(crate) struct NodeHealStatusSnapshot {
     pub progress: Option<NodeHealProgress>,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+struct NodeHealStatusSnapshotV1 {
+    version: u8,
+    services_enabled: bool,
+    initialized: bool,
+    info: NodeHealInfo,
+    operations: HealOperationsSnapshot,
+    progress: Option<NodeHealProgressV1>,
+}
+
+impl From<&NodeHealStatusSnapshot> for NodeHealStatusSnapshotV1 {
+    fn from(snapshot: &NodeHealStatusSnapshot) -> Self {
+        Self {
+            version: NODE_HEAL_STATUS_PREVIOUS_VERSION,
+            services_enabled: snapshot.services_enabled,
+            initialized: snapshot.initialized,
+            info: snapshot.info.clone(),
+            operations: snapshot.operations,
+            progress: snapshot.progress.as_ref().map(NodeHealProgressV1::from),
+        }
+    }
+}
+
+impl From<NodeHealStatusSnapshotV1> for NodeHealStatusSnapshot {
+    fn from(snapshot: NodeHealStatusSnapshotV1) -> Self {
+        Self {
+            version: snapshot.version,
+            services_enabled: snapshot.services_enabled,
+            initialized: snapshot.initialized,
+            info: snapshot.info,
+            operations: snapshot.operations,
+            progress: snapshot.progress.map(NodeHealProgress::from),
+        }
+    }
+}
+
+#[derive(Deserialize)]
+struct NodeHealStatusVersion {
+    version: u8,
+}
+
 impl NodeHealStatusSnapshot {
     #[cfg(test)]
     pub(crate) fn for_test(
@@ -249,14 +317,7 @@ impl NodeHealStatusSnapshot {
 }
 
 pub(crate) async fn capture_node_heal_status(info: BackgroundHealInfo) -> NodeHealStatusSnapshot {
-    let progress = rustfs_heal::current_heal_progress_snapshot()
-        .await
-        .map(|progress| NodeHealProgress {
-            objects_scanned: progress.objects_scanned,
-            objects_healed: progress.objects_healed,
-            objects_failed: progress.objects_failed,
-            bytes_processed: progress.bytes_processed,
-        });
+    let progress = rustfs_heal::current_heal_progress_snapshot().await;
 
     NodeHealStatusSnapshot {
         version: NODE_HEAL_STATUS_VERSION,
@@ -268,22 +329,43 @@ pub(crate) async fn capture_node_heal_status(info: BackgroundHealInfo) -> NodeHe
     }
 }
 
-pub(crate) fn encode_node_heal_status(snapshot: &NodeHealStatusSnapshot) -> Result<Vec<u8>, String> {
-    encode_msgpack_map(snapshot).map_err(|err| format!("failed to encode node heal status: {err}"))
+pub(crate) fn encode_node_heal_status(snapshot: &NodeHealStatusSnapshot, protocol_version: u32) -> Result<Vec<u8>, String> {
+    let encoded = if protocol_version < rustfs_protos::BACKGROUND_HEAL_STATUS_PROTOCOL_VERSION {
+        encode_msgpack_map(&NodeHealStatusSnapshotV1::from(snapshot))
+    } else {
+        let mut snapshot = snapshot.clone();
+        snapshot.version = NODE_HEAL_STATUS_VERSION;
+        encode_msgpack_map(&snapshot)
+    };
+    encoded.map_err(|err| format!("failed to encode node heal status: {err}"))
 }
 
 pub(crate) fn decode_node_heal_status(data: &[u8]) -> Result<NodeHealStatusSnapshot, String> {
     if data.len() > NODE_HEAL_STATUS_MAX_SIZE {
         return Err("node heal status exceeds size limit".to_string());
     }
-    let mut deserializer = Deserializer::new(Cursor::new(data));
-    let snapshot = NodeHealStatusSnapshot::deserialize(&mut deserializer)
-        .map_err(|err| format!("failed to decode node heal status: {err}"))?;
+    let decode_version = || {
+        let mut deserializer = Deserializer::new(Cursor::new(data));
+        NodeHealStatusVersion::deserialize(&mut deserializer)
+            .map(|version| (version, deserializer))
+            .map_err(|err| format!("failed to decode node heal status: {err}"))
+    };
+    let (version, deserializer) = decode_version()?;
     if usize::try_from(deserializer.get_ref().position()).ok() != Some(data.len()) {
         return Err("node heal status contains trailing data".to_string());
     }
-    if snapshot.version != NODE_HEAL_STATUS_VERSION {
-        return Err(format!("unsupported node heal status version: {}", snapshot.version));
+
+    let mut deserializer = Deserializer::new(Cursor::new(data));
+    let snapshot = match version.version {
+        NODE_HEAL_STATUS_PREVIOUS_VERSION => {
+            NodeHealStatusSnapshotV1::deserialize(&mut deserializer).map(NodeHealStatusSnapshot::from)
+        }
+        NODE_HEAL_STATUS_VERSION => NodeHealStatusSnapshot::deserialize(&mut deserializer),
+        version => return Err(format!("unsupported node heal status version: {version}")),
+    }
+    .map_err(|err| format!("failed to decode node heal status: {err}"))?;
+    if usize::try_from(deserializer.get_ref().position()).ok() != Some(data.len()) {
+        return Err("node heal status contains trailing data".to_string());
     }
     Ok(snapshot)
 }
@@ -387,9 +469,10 @@ pub(crate) fn decode_node_replacement_recovery_status(data: &[u8]) -> Result<Nod
 #[cfg(test)]
 mod tests {
     use super::{
-        NODE_HEAL_STATUS_MAX_SIZE, NODE_HEAL_STATUS_VERSION, NodeHealProgress, NodeHealStatusSnapshot,
-        NodeReplacementRecoveryStatusSnapshot, decode_node_heal_status, decode_node_replacement_recovery_status,
-        encode_node_heal_status, encode_node_replacement_recovery_status, heal_control_coordinator, heal_topology_fingerprint,
+        NODE_HEAL_STATUS_MAX_SIZE, NODE_HEAL_STATUS_PREVIOUS_VERSION, NODE_HEAL_STATUS_VERSION, NodeHealProgress,
+        NodeHealStatusSnapshot, NodeReplacementRecoveryStatusSnapshot, decode_node_heal_status,
+        decode_node_replacement_recovery_status, encode_node_heal_status, encode_node_replacement_recovery_status,
+        heal_control_coordinator, heal_topology_fingerprint,
     };
     use crate::storage::storage_api::{
         Endpoint,
@@ -533,20 +616,72 @@ mod tests {
                 ..Default::default()
             },
             Some(NodeHealProgress {
+                kind: rustfs_heal::heal::progress::HealProgressKind::ObjectSweep,
                 objects_scanned: 7,
                 objects_healed: 5,
                 objects_failed: 1,
+                skipped_objects: 1,
+                objects_total_count: 10,
+                objects_total_size: 2048,
                 bytes_processed: 1024,
+                progress_percentage: 50.0,
+                progress_state: rustfs_heal::heal::progress::HealProgressState::Running,
+                baseline_generation: Some(42),
+                baseline_known: true,
+                ..Default::default()
             }),
         );
 
-        let encoded = encode_node_heal_status(&snapshot).expect("snapshot should encode");
+        let encoded = encode_node_heal_status(&snapshot, rustfs_protos::BACKGROUND_HEAL_STATUS_PROTOCOL_VERSION)
+            .expect("snapshot should encode");
         let decoded = decode_node_heal_status(&encoded).expect("snapshot should decode");
         assert_eq!(decoded.version, NODE_HEAL_STATUS_VERSION);
         assert_eq!(decoded.operations.queue_length, 2);
         assert_eq!(decoded.progress, snapshot.progress);
     }
 
+    #[test]
+    fn node_heal_status_v1_encoding_preserves_rolling_compatibility() {
+        let snapshot = NodeHealStatusSnapshot::for_test(
+            true,
+            true,
+            BackgroundHealInfo::default(),
+            HealOperationsSnapshot::default(),
+            Some(NodeHealProgress {
+                objects_scanned: 7,
+                objects_healed: 5,
+                objects_failed: 1,
+                skipped_objects: 3,
+                bytes_processed: 1024,
+                progress_state: rustfs_heal::heal::progress::HealProgressState::Running,
+                baseline_known: true,
+                ..Default::default()
+            }),
+        );
+
+        let encoded =
+            encode_node_heal_status(&snapshot, u32::from(NODE_HEAL_STATUS_PREVIOUS_VERSION)).expect("v1 snapshot should encode");
+        let wire: serde_json::Value = rmp_serde::from_slice(&encoded).expect("v1 snapshot should decode as JSON");
+        let progress = wire["progress"].as_object().expect("v1 progress should be a map");
+        assert_eq!(wire["version"], NODE_HEAL_STATUS_PREVIOUS_VERSION);
+        assert_eq!(progress.len(), 4);
+        assert_eq!(progress["objectsScanned"], 7);
+        assert!(!progress.contains_key("skippedObjects"));
+
+        let decoded = decode_node_heal_status(&encoded).expect("v1 snapshot should decode");
+        let progress = decoded.progress.as_ref().expect("v1 progress should be present");
+        assert_eq!(decoded.version, NODE_HEAL_STATUS_PREVIOUS_VERSION);
+        assert_eq!(progress.objects_scanned, 7);
+        assert_eq!(progress.progress_state, rustfs_heal::heal::progress::HealProgressState::Unknown);
+        assert!(!progress.baseline_known);
+
+        let upgraded = encode_node_heal_status(&decoded, rustfs_protos::BACKGROUND_HEAL_STATUS_PROTOCOL_VERSION)
+            .expect("decoded v1 snapshot should upgrade to v2");
+        let upgraded = decode_node_heal_status(&upgraded).expect("upgraded snapshot should decode");
+        assert_eq!(upgraded.version, NODE_HEAL_STATUS_VERSION);
+        assert_eq!(upgraded.progress.as_ref(), Some(progress));
+    }
+
     #[test]
     fn node_heal_status_rejects_unknown_version() {
         let mut snapshot = NodeHealStatusSnapshot::for_test(
@@ -558,7 +693,7 @@ mod tests {
         );
         snapshot.version += 1;
 
-        let encoded = encode_node_heal_status(&snapshot).expect("snapshot should encode");
+        let encoded = rmp_serde::to_vec_named(&snapshot).expect("snapshot should encode");
         let err = decode_node_heal_status(&encoded).expect_err("unknown version should fail closed");
         assert!(err.contains("unsupported node heal status version"));
     }
@@ -579,13 +714,22 @@ mod tests {
                 "activeBySource": {"scanner": 0, "admin": 1, "autoHeal": 0, "internal": 0, "readRepair": 0},
                 "retryingBySource": {"scanner": 0, "admin": 0, "autoHeal": 0, "internal": 0, "readRepair": 0}
             },
-            "progress": null
+            "progress": {
+                "objectsScanned": 7,
+                "objectsHealed": 5,
+                "objectsFailed": 1,
+                "bytesProcessed": 1024
+            }
         });
         let encoded = rmp_serde::to_vec_named(&fixture).expect("fixture should encode");
         let decoded = decode_node_heal_status(&encoded).expect("fixed v1 fixture should decode");
         assert_eq!(decoded.info().bitrot_start_cycle, 9);
         assert_eq!(decoded.operations.queue_length, 2);
         assert_eq!(decoded.operations.queued_by_source.mrf, 0);
+        let progress = decoded.progress.expect("legacy progress should decode");
+        assert_eq!(progress.objects_scanned, 7);
+        assert!(!progress.baseline_known);
+        assert_eq!(progress.progress_state, rustfs_heal::heal::progress::HealProgressState::Unknown);
     }
 
     #[test]
@@ -631,7 +775,8 @@ mod tests {
             HealOperationsSnapshot::default(),
             None,
         );
-        let encoded = encode_node_heal_status(&snapshot).expect("snapshot should encode");
+        let encoded = encode_node_heal_status(&snapshot, rustfs_protos::BACKGROUND_HEAL_STATUS_PROTOCOL_VERSION)
+            .expect("snapshot should encode");
         let encoded_json: serde_json::Value = rmp_serde::from_slice(&encoded).expect("encoded snapshot should decode as JSON");
         assert_eq!(encoded_json["info"]["bitrotStartTime"], serde_json::json!("2023-11-14T22:13:20.123456Z"));
     }