From e3e5693e604a7045b63f977577fde49ff6dd6454 Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 9 Jul 2026 02:29:54 +0800 Subject: [PATCH] fix(ecstore): correct heal result drive-record reporting (#4555) Two report-only defects on the heal result-reporting surface (they do not affect the data path or heal decisions): - default_heal_result (set_disk/ops/heal.rs): the offline-disk branch pushed an Offline record but fell through into the unconditional push, emitting a second (Corrupt) record for the same disk. This grew before/after.drives to disk_count + offline_count and misaligned every entry after the first offline slot. Add `continue` after the offline push, drive `disk_len` and the loop from a single `self.disks` snapshot, and assert `errs.len() == disk_len`. - Sets::heal_format (core/sets.rs): the before/after drive lists were pre-filled with N default placeholders and then N real entries were pushed, yielding a 2N list whose healed status updates (indexed 0..N) landed on the blank placeholder half. Assign the lists directly from formats_to_drives_info (mirroring the set-level heal_format) so the healed updates hit the real entries. Add regression tests covering offline/online record alignment and the NoHealRequired and heal paths of the pool-level heal_format. Co-authored-by: heihutu --- crates/ecstore/src/core/sets.rs | 144 +++++++++++++++++++-- crates/ecstore/src/set_disk/ops/heal.rs | 165 +++++++++++++++++++++++- 2 files changed, 299 insertions(+), 10 deletions(-) diff --git a/crates/ecstore/src/core/sets.rs b/crates/ecstore/src/core/sets.rs index d73e504c6..cd99272cc 100644 --- a/crates/ecstore/src/core/sets.rs +++ b/crates/ecstore/src/core/sets.rs @@ -47,7 +47,7 @@ use rustfs_common::heal_channel::HealOpts; use rustfs_common::heal_channel::{DriveState, HealItemType}; use rustfs_filemeta::FileInfo; use rustfs_lock::NamespaceLockWrapper; -use rustfs_madmin::heal_commands::{HealDriveInfo, HealResultItem}; +use rustfs_madmin::heal_commands::HealResultItem; use rustfs_utils::{crc_hash, path::path_join_buf, sip_hash}; use std::{collections::HashMap, sync::Arc}; use tokio::sync::RwLock; @@ -868,14 +868,15 @@ impl crate::storage_api_contracts::heal::HealOperations for Sets { set_count: self.set_count, ..Default::default() }; + // One drive record per endpoint (`formats_to_drives_info` returns exactly + // N entries). Assign directly instead of pre-filling N defaults and then + // pushing N real entries: the old form produced a 2N-long list whose empty + // placeholder half received the healed uuid/Ok updates below (indexed by + // `i * set_drive_count + j`, i.e. 0..N), leaving the real drive entries + // never marked healed. Mirrors the set-level `heal_format`. let before_derives = formats_to_drives_info(&self.endpoints.endpoints, &formats, &errs); - res.before.drives = vec![HealDriveInfo::default(); before_derives.len()]; - res.after.drives = vec![HealDriveInfo::default(); before_derives.len()]; - - for v in before_derives.iter() { - res.before.drives.push(v.clone()); - res.after.drives.push(v.clone()); - } + res.before.drives = before_derives.clone(); + res.after.drives = before_derives; if count_errs(&errs, &DiskError::UnformattedDisk) == 0 { info!("disk formats success, NoHealRequired, errs: {:?}", errs); return Ok((res, Some(StorageError::NoHealRequired))); @@ -1318,4 +1319,131 @@ mod tests { .expect_err("abandoned-parts ownership should stay above the pool/set storage layers"); assert!(matches!(err, StorageError::NotImplemented)); } + + // Builds a single-set `Sets` over `SET_DRIVE_COUNT` local temp-dir disks, + // formatting the first `num_formatted` of them against a shared reference + // format and leaving the rest unformatted. Returns the live TempDir handles + // (must be kept alive), the reference format, and the assembled `Sets`. + // `disk_set` is intentionally empty: these tests only drive `heal_format` + // with `dry_run == true`, which never touches `disk_set`. + async fn setup_heal_format_sets(num_formatted: usize) -> (Vec, FormatV3, Sets) { + const SET_DRIVE_COUNT: usize = 3; + let ref_format = FormatV3::new(1, SET_DRIVE_COUNT); + + let mut dirs = Vec::with_capacity(SET_DRIVE_COUNT); + let mut endpoints = Vec::with_capacity(SET_DRIVE_COUNT); + for i in 0..SET_DRIVE_COUNT { + let dir = tempfile::tempdir().expect("tempdir should be created"); + let mut endpoint = + Endpoint::try_from(dir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse"); + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(i); + dirs.push(dir); + endpoints.push(endpoint); + } + + for (i, endpoint) in endpoints.iter().enumerate().take(num_formatted) { + let disk = new_disk( + endpoint, + &DiskOption { + cleanup: false, + health_check: false, + }, + ) + .await + .expect("disk should be created"); + let mut disk_format = ref_format.clone(); + disk_format.erasure.this = ref_format.erasure.sets[0][i]; + save_format_file(&Some(disk), &Some(disk_format)) + .await + .expect("format should be saved"); + } + + let sets = Sets { + id: ref_format.id, + disk_set: Vec::new(), + pool_idx: 0, + endpoints: PoolEndpoints { + legacy: false, + set_count: 1, + drives_per_set: SET_DRIVE_COUNT, + endpoints: Endpoints::from(endpoints), + cmd_line: String::new(), + platform: String::new(), + }, + format: ref_format.clone(), + parity_count: 1, + set_count: 1, + set_drive_count: SET_DRIVE_COUNT, + default_parity_count: 1, + distribution_algo: DistributionAlgoVersion::V1, + exit_signal: None, + ctx: bootstrap_ctx(), + }; + + (dirs, ref_format, sets) + } + + // Regression for #956 (NoHealRequired path): with every disk already + // formatted, `heal_format` reports exactly one drive record per disk + // (N = set_count * set_drive_count), each carrying a real endpoint. Before + // the fix the list was pre-filled with N empty placeholders and then N real + // entries were pushed, yielding a 2N list whose first half was blank. + #[tokio::test] + #[serial] + async fn heal_format_no_heal_required_reports_one_record_per_disk() { + let (_dirs, _ref_format, sets) = setup_heal_format_sets(3).await; + + let (res, err) = sets.heal_format(true).await.expect("heal_format should succeed"); + // All disks formatted -> NoHealRequired early return, still returns `res`. + assert!(matches!(err, Some(StorageError::NoHealRequired)), "expected NoHealRequired, got {err:?}"); + + assert_eq!(res.before.drives.len(), 3, "before drives must be N, not 2N"); + assert_eq!(res.after.drives.len(), 3, "after drives must be N, not 2N"); + for (i, d) in res.before.drives.iter().enumerate() { + assert!( + !d.endpoint.is_empty(), + "before drive {i} endpoint must not be empty (no placeholder rows)" + ); + assert_eq!(d.state, DriveState::Ok.to_string(), "formatted disk {i} must be Ok"); + } + for (i, d) in res.after.drives.iter().enumerate() { + assert!(!d.endpoint.is_empty(), "after drive {i} endpoint must not be empty (no placeholder rows)"); + } + } + + // Regression for #956 (heal path): with one unformatted disk the heal path is + // taken (past the NoHealRequired check). Even here the reported drive list is + // length N (never 2N) and index-aligned with the endpoints, so the healed + // status updates (indexed `i * set_drive_count + j`, 0..N) address the real + // drive entries rather than an empty placeholder half. `dry_run == true` keeps + // the assertion on the reported shape without mutating disks or global state. + #[tokio::test] + #[serial] + async fn heal_format_heal_path_reports_one_record_per_disk_aligned() { + // Disks 0 and 1 formatted (quorum), disk 2 unformatted. + let (_dirs, _ref_format, sets) = setup_heal_format_sets(2).await; + + let (res, err) = sets.heal_format(true).await.expect("heal_format should succeed"); + // Unformatted disk present -> heal path, not NoHealRequired. + assert!(err.is_none(), "expected heal path (no NoHealRequired), got {err:?}"); + + assert_eq!(res.before.drives.len(), 3, "before drives must be N, not 2N"); + assert_eq!(res.after.drives.len(), 3, "after drives must be N, not 2N"); + + // Every record maps to a real endpoint; the drive list is index-aligned + // with `formats_to_drives_info`, so no empty placeholder half remains. + for (i, d) in res.before.drives.iter().enumerate() { + assert!(!d.endpoint.is_empty(), "before drive {i} endpoint must not be empty"); + } + assert_eq!(res.before.drives[0].state, DriveState::Ok.to_string()); + assert_eq!(res.before.drives[1].state, DriveState::Ok.to_string()); + // The unformatted disk is reported Missing on its own (real) entry. + assert_eq!( + res.before.drives[2].state, + DriveState::Missing.to_string(), + "unformatted disk must be Missing on its real index, not on a placeholder" + ); + } } diff --git a/crates/ecstore/src/set_disk/ops/heal.rs b/crates/ecstore/src/set_disk/ops/heal.rs index 6eeea3e9a..81516a847 100644 --- a/crates/ecstore/src/set_disk/ops/heal.rs +++ b/crates/ecstore/src/set_disk/ops/heal.rs @@ -977,7 +977,12 @@ impl SetDisks { object: &str, version_id: &str, ) -> HealResultItem { - let disk_len = { self.disks.read().await.len() }; + // Take a single snapshot of the disk vector and drive both `disk_len` and + // the per-drive loop below from it, so the reported `disk_count` and the + // pushed drive records always agree (previously two independent + // `self.disks.read()` calls could observe different lengths). + let disks = self.disks.read().await; + let disk_len = disks.len(); let mut result = HealResultItem { heal_item_type: HealItemType::Object.to_string(), bucket: bucket.to_string(), @@ -996,7 +1001,11 @@ impl SetDisks { result.data_blocks = disk_len - result.parity_blocks; - for (index, disk) in self.disks.read().await.iter().enumerate() { + // `errs` is index-aligned with the disk vector; only the online path below + // indexes into it (the offline branch `continue`s before touching it). + debug_assert_eq!(errs.len(), disk_len, "errs length must match the disk count"); + + for (index, disk) in disks.iter().enumerate() { if disk.is_none() { result.before.drives.push(HealDriveInfo { uuid: "".to_string(), @@ -1009,6 +1018,10 @@ impl SetDisks { endpoint: self.set_endpoints[index].to_string(), state: DriveState::Offline.to_string(), }); + // Offline disks contribute exactly one record; without this the + // control flow fell through and pushed a second (Corrupt) record + // for the same disk, doubling the list and breaking index alignment. + continue; } let mut drive_state = DriveState::Corrupt; @@ -1211,3 +1224,151 @@ impl crate::storage_api_contracts::heal::HealOperations for SetDisks { Err(StorageError::NotImplemented) } } + +#[cfg(test)] +mod heal_result_report_tests { + use super::SetDisks; + use crate::disk::endpoint::Endpoint; + use crate::disk::error::DiskError; + use crate::disk::format::FormatV3; + use crate::disk::{DiskOption, DiskStore, new_disk}; + use rustfs_common::heal_channel::DriveState; + use rustfs_filemeta::FileInfo; + use std::sync::Arc; + use tempfile::TempDir; + use tokio::sync::RwLock; + + async fn real_disk() -> (TempDir, Endpoint, DiskStore) { + let dir = tempfile::tempdir().expect("tempdir should be created"); + let endpoint = + Endpoint::try_from(dir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse"); + let disk = new_disk( + &endpoint, + &DiskOption { + cleanup: false, + health_check: false, + }, + ) + .await + .expect("disk should be created"); + (dir, endpoint, disk) + } + + async fn set_disks_with( + disks: Vec>, + endpoints: Vec, + default_parity_count: usize, + ) -> Arc { + let set_drive_count = disks.len(); + SetDisks::new( + "test-owner".to_string(), + Arc::new(RwLock::new(disks)), + set_drive_count, + default_parity_count, + 0, + 0, + endpoints, + FormatV3::new(1, set_drive_count), + vec![], + ) + .await + } + + // Regression for #955: an offline disk must contribute exactly one drive + // record. Before the fix the offline branch fell through and pushed a second + // (Corrupt) record for the same disk, so `before/after.drives` grew to + // `disk_count + offline_count` and every entry after the first offline slot + // was misaligned relative to its disk index. + #[tokio::test] + async fn default_heal_result_reports_one_record_per_disk_and_stays_aligned() { + // index 0: online, no error -> Ok + // index 1: offline (None) -> Offline (single record) + // index 2: online, FileNotFound -> Missing + // index 3: online, DiskAccessDenied-> Corrupt + let (_d0, ep0, disk0) = real_disk().await; + let (_d2, ep2, disk2) = real_disk().await; + let (_d3, ep3, disk3) = real_disk().await; + let ep1 = Endpoint::try_from("http://127.0.0.1:9001/data").expect("endpoint should parse"); + + let disks = vec![Some(disk0), None, Some(disk2), Some(disk3)]; + let endpoints = vec![ep0, ep1, ep2, ep3]; + let set = set_disks_with(disks, endpoints, 1).await; + + let errs = vec![ + None, + Some(DiskError::DiskNotFound), + Some(DiskError::FileNotFound), + Some(DiskError::DiskAccessDenied), + ]; + + let result = set + .default_heal_result(FileInfo::default(), &errs, "bucket", "object", "") + .await; + + // Exactly one record per disk (not disk_count + offline_count). + assert_eq!(result.disk_count, 4); + assert_eq!(result.before.drives.len(), 4, "one before record per disk"); + assert_eq!(result.after.drives.len(), 4, "one after record per disk"); + + // Records stay index-aligned with the disk vector and set_endpoints. + let expected_states = [ + DriveState::Ok.to_string(), + DriveState::Offline.to_string(), + DriveState::Missing.to_string(), + DriveState::Corrupt.to_string(), + ]; + for (i, expected) in expected_states.iter().enumerate() { + assert_eq!(&result.before.drives[i].state, expected, "before state at {i}"); + assert_eq!(&result.after.drives[i].state, expected, "after state at {i}"); + assert_eq!( + result.before.drives[i].endpoint, + set.set_endpoints[i].to_string(), + "before endpoint aligned at {i}" + ); + assert_eq!( + result.after.drives[i].endpoint, + set.set_endpoints[i].to_string(), + "after endpoint aligned at {i}" + ); + } + + // The offline endpoint appears exactly once, never as a second Corrupt row. + let offline_ep = set.set_endpoints[1].to_string(); + assert_eq!( + result.before.drives.iter().filter(|d| d.endpoint == offline_ep).count(), + 1, + "offline disk must not produce a duplicate record" + ); + } + + // Two interleaved offline disks: assert every record still maps to its own + // set_endpoints[index] (no cumulative drift after the first offline slot). + #[tokio::test] + async fn default_heal_result_alignment_with_multiple_offline_disks() { + let (_d1, ep1, disk1) = real_disk().await; + let (_d3, ep3, disk3) = real_disk().await; + let ep0 = Endpoint::try_from("http://127.0.0.1:9000/data").expect("endpoint should parse"); + let ep2 = Endpoint::try_from("http://127.0.0.1:9002/data").expect("endpoint should parse"); + + // index 0 offline, 1 online, 2 offline, 3 online. + let disks = vec![None, Some(disk1), None, Some(disk3)]; + let endpoints = vec![ep0, ep1, ep2, ep3]; + let set = set_disks_with(disks, endpoints, 1).await; + + let errs = vec![Some(DiskError::DiskNotFound), None, Some(DiskError::DiskNotFound), None]; + + let result = set + .default_heal_result(FileInfo::default(), &errs, "bucket", "object", "") + .await; + + assert_eq!(result.before.drives.len(), 4); + assert_eq!(result.after.drives.len(), 4); + for i in 0..4 { + assert_eq!(result.before.drives[i].endpoint, set.set_endpoints[i].to_string(), "aligned at {i}"); + } + assert_eq!(result.before.drives[0].state, DriveState::Offline.to_string()); + assert_eq!(result.before.drives[1].state, DriveState::Ok.to_string()); + assert_eq!(result.before.drives[2].state, DriveState::Offline.to_string()); + assert_eq!(result.before.drives[3].state, DriveState::Ok.to_string()); + } +}