mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-26 16:28:15 +00:00
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 <heihutu@gmail.com>
This commit is contained in:
houseme
@@ -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<tempfile::TempDir>, 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"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<Option<DiskStore>>,
|
||||
endpoints: Vec<Endpoint>,
|
||||
default_parity_count: usize,
|
||||
) -> Arc<SetDisks> {
|
||||
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());
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user