mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-08-19 05:36:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(scrub): align Rust INDEX scrub to Go's idx.CheckIndexFile (#10142)

Browse Source 
* feat(idx): add check_index_file mirroring Go idx.CheckIndexFile

Index-only structural check: walk the on-disk index, sort by (offset, size),
flag overlapping needles, and verify the file is a whole number of entries.
No data-file reads.

Claude-Session: https://claude.ai/code/session_015EE9Sc9EvNp8BCVva4RKdo

* refactor(ec): use idx::check_index_file in EcVolume::scrub_index

Drops the inline walk/sort/overlap copy. Walks a private fd so the structural
scan never moves the shared ecx_file cursor (read positionally elsewhere).

Claude-Session: https://claude.ai/code/session_015EE9Sc9EvNp8BCVva4RKdo

* fix(scrub): make Volume::scrub_index an index-only check on the on-disk .idx

INDEX mode walked the deduped in-memory map and read .dat headers — more
than the cheap-INDEX contract allows, yet missing Go's overlap and
size-multiple structural checks. Route it through idx::check_index_file so
it matches Go's Volume.ScrubIndex and the INDEX<LOCAL<FULL cost tiering holds.
Ports openIndex's zero-size-index guard (a populated .dat with an empty .idx
is corruption) and takes the data-file read lock for a consistent snapshot.

Claude-Session: https://claude.ai/code/session_015EE9Sc9EvNp8BCVva4RKdo
This commit is contained in:
Chris Lu
2026-06-30 00:00:24 -07:00
committed by GitHub
parent 0e293c9b0a
commit 72009c607b
3 changed files with 192 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files
seaweed-volume/src/storage/erasure_coding/ec_volume.rs
+15 -82
View File
@@ -708,97 +708,30 @@ impl EcVolume {
/// Matches Go's `(ev *EcVolume) ScrubIndex()` → `idx.CheckIndexFile()`.
/// Returns (entry_count, errors).
pub fn scrub_index(&self) -> (u64, Vec<String>) {
let ecx_file = match self.ecx_file.as_ref() {
Some(f) => f,
None => {
return (
0,
vec![format!(
"no ECX file associated with EC volume {}",
self.volume_id.0
)],
)
}
};
if self.ecx_file_size == 0 {
if self.ecx_file.is_none() {
return (
0,
vec![format!(
"zero-size ECX file for EC volume {}",
"no ECX file associated with EC volume {}",
self.volume_id.0
)],
);
}
let entry_count = self.ecx_file_size as usize / NEEDLE_MAP_ENTRY_SIZE;
let mut entries: Vec<(usize, NeedleId, i64, Size)> = Vec::with_capacity(entry_count);
let mut errs: Vec<String> = Vec::new();
let mut entry_buf = [0u8; NEEDLE_MAP_ENTRY_SIZE];
// Walk all entries
for i in 0..entry_count {
let file_offset = (i * NEEDLE_MAP_ENTRY_SIZE) as u64;
#[cfg(unix)]
{
use std::os::unix::fs::FileExt;
if let Err(e) = ecx_file.read_exact_at(&mut entry_buf, file_offset) {
errs.push(format!("read ecx entry {}: {}", i, e));
continue;
}
}
let (key, offset, size) = idx_entry_from_bytes(&entry_buf);
entries.push((i, key, offset.to_actual_offset(), size));
if self.ecx_file_size == 0 {
return (
0,
vec![format!("zero-size ECX file for EC volume {}", self.volume_id.0)],
);
}
// Sort by offset, then size
entries.sort_by(|a, b| a.2.cmp(&b.2).then(a.3 .0.cmp(&b.3 .0)));
// Check for overlapping needles
for i in 1..entries.len() {
let (idx, id, offset, size) = entries[i];
let (_, last_id, last_offset, last_size) = entries[i - 1];
let actual_size =
crate::storage::needle::needle::get_actual_size(size, self.version);
let end = if actual_size != 0 {
offset + actual_size - 1
} else {
offset
};
let last_actual_size =
crate::storage::needle::needle::get_actual_size(last_size, self.version);
let last_end = if last_actual_size != 0 {
last_offset + last_actual_size - 1
} else {
last_offset
};
if offset <= last_end {
errs.push(format!(
"needle {} (#{}) at [{}-{}] overlaps needle {} at [{}-{}]",
id.0,
idx + 1,
offset,
end,
last_id.0,
last_offset,
last_end
));
}
}
// Verify file size matches entry count
let expected_size = entry_count as i64 * NEEDLE_MAP_ENTRY_SIZE as i64;
if expected_size != self.ecx_file_size {
errs.push(format!(
"expected an index file of size {}, got {}",
expected_size, self.ecx_file_size
));
}
(entries.len() as u64, errs)
// Walk a private fd so the structural scan never moves the shared
// ecx_file cursor (the cached handle is read positionally elsewhere).
let ecx_path = self.ecx_file_name();
let mut ecx_file = match File::open(&ecx_path) {
Ok(f) => f,
Err(e) => return (0, vec![format!("open ECX file {}: {}", ecx_path, e)]),
};
crate::storage::idx::check_index_file(&mut ecx_file, self.ecx_file_size, self.version)
}
// ---- Deletion ----
seaweed-volume/src/storage/idx/mod.rs
+109
View File
@@ -2,6 +2,7 @@
//!
//! Each entry: NeedleId(8) + Offset(5) + Size(4) = 17 bytes.
use crate::storage::needle::needle::get_actual_size;
use crate::storage::types::*;
use std::io::{self, Read, Seek, SeekFrom};
@@ -36,6 +37,67 @@ where
}
}
/// Verify the integrity of an .idx/.ecx index file: walk every entry, sort by
/// (offset, size), flag overlapping needles, and check the file is a whole
/// number of entries. Mirrors Go's `idx.CheckIndexFile`. No data-file reads.
pub fn check_index_file<R: Read + Seek>(
reader: &mut R,
idx_file_size: i64,
version: Version,
) -> (u64, Vec<String>) {
let mut errs = Vec::new();
// (walk index, id, actual offset, size)
let mut entries: Vec<(usize, NeedleId, i64, Size)> = Vec::new();
let mut i = 0usize;
if let Err(e) = walk_index_file(reader, 0, |id, offset, size| {
entries.push((i, id, offset.to_actual_offset(), size));
i += 1;
Ok(())
}) {
errs.push(format!("walk index file: {}", e));
}
entries.sort_by(|a, b| a.2.cmp(&b.2).then(a.3 .0.cmp(&b.3 .0)));
for j in 1..entries.len() {
let (index, id, offset, size) = entries[j];
let (_, last_id, last_offset, last_size) = entries[j - 1];
let end = match get_actual_size(size, version) {
0 => offset,
s => offset + s - 1,
};
let last_end = match get_actual_size(last_size, version) {
0 => last_offset,
s => last_offset + s - 1,
};
if offset <= last_end {
errs.push(format!(
"needle {} (#{}) at [{}-{}] overlaps needle {} at [{}-{}]",
id.0,
index + 1,
offset,
end,
last_id.0,
last_offset,
last_end
));
}
}
let count = entries.len() as u64;
let expected = count as i64 * NEEDLE_MAP_ENTRY_SIZE as i64;
if expected != idx_file_size {
errs.push(format!(
"expected an index file of size {}, got {}",
idx_file_size, expected
));
}
(count, errs)
}
/// Write a single index entry to a writer.
pub fn write_index_entry<W: io::Write>(
writer: &mut W,
@@ -96,6 +158,53 @@ mod tests {
assert_eq!(count, 0);
}
fn idx_bytes(entries: &[(NeedleId, Offset, Size)]) -> Vec<u8> {
let mut data = Vec::new();
for (key, offset, size) in entries {
let mut buf = [0u8; NEEDLE_MAP_ENTRY_SIZE];
idx_entry_to_bytes(&mut buf, *key, *offset, *size);
data.extend_from_slice(&buf);
}
data
}
#[test]
fn test_check_index_file_clean() {
let data = idx_bytes(&[
(NeedleId(1), Offset::from_actual_offset(0), Size(50)),
(NeedleId(2), Offset::from_actual_offset(100_000), Size(50)),
]);
let size = data.len() as i64;
let (count, errs) = check_index_file(&mut Cursor::new(data), size, Version(3));
assert_eq!(count, 2);
assert!(errs.is_empty(), "{:?}", errs);
}
#[test]
fn test_check_index_file_detects_overlap() {
let data = idx_bytes(&[
(NeedleId(1), Offset::from_actual_offset(0), Size(100)),
(NeedleId(2), Offset::from_actual_offset(8), Size(100)),
]);
let size = data.len() as i64;
let (count, errs) = check_index_file(&mut Cursor::new(data), size, Version(3));
assert_eq!(count, 2);
assert_eq!(errs.len(), 1, "{:?}", errs);
assert!(errs[0].contains("overlaps"), "{:?}", errs);
}
#[test]
fn test_check_index_file_detects_size_mismatch() {
let data = idx_bytes(&[(NeedleId(1), Offset::from_actual_offset(0), Size(50))]);
let claimed = data.len() as i64 + 5;
let (_count, errs) = check_index_file(&mut Cursor::new(data), claimed, Version(3));
assert!(
errs.iter().any(|e| e.contains("expected an index file")),
"{:?}",
errs
);
}
#[test]
fn test_write_index_entry() {
let mut buf = Vec::new();
seaweed-volume/src/storage/volume.rs
+68 -51
View File
@@ -1966,66 +1966,51 @@ impl Volume {
Ok(())
}
/// Scrub the volume index by verifying each needle map entry against the dat file.
/// For each entry, reads only the 16-byte needle header at the given offset to verify:
/// correct needle ID, correct cookie (non-zero), and valid size.
/// Does NOT read/verify the full needle data or CRC.
/// Returns (files_checked, broken_needles) tuple.
/// Scrub the volume index: verify the on-disk .idx for overlapping needles
/// and a size that is a whole number of entries. Mirrors Go's Volume.ScrubIndex
/// → idx.CheckIndexFile. Index-only; does not read the .dat.
pub fn scrub_index(&self) -> Result<(u64, Vec<String>), VolumeError> {
if self.dat_file.is_none() && self.remote_dat_file.is_none() {
return Err(VolumeError::NotFound);
}
let nm = self.nm_or_not_found()?;
let dat_size = self.dat_file_size().map_err(VolumeError::Io)?;
let _guard = self.data_file_access_control.read_lock();
let mut files_checked: u64 = 0;
let mut broken = Vec::new();
let idx_path = self.file_name(".idx");
let mut idx_file = match File::open(&idx_path) {
Ok(f) => f,
Err(e) => return Ok((0, vec![format!("open index file {}: {}", idx_path, e)])),
};
let idx_file_size = match idx_file.metadata() {
Ok(m) => m.len() as i64,
Err(e) => return Ok((0, vec![format!("stat index file {}: {}", idx_path, e)])),
};
for (needle_id, nv) in nm.iter_entries() {
if nv.offset.is_zero() || nv.size.is_deleted() {
continue;
}
let offset = nv.offset.to_actual_offset();
if offset < 0 || offset as u64 >= dat_size {
broken.push(format!(
"needle {} offset {} out of range (dat_size={})",
needle_id.0, offset, dat_size
));
continue;
}
// Read only the 16-byte needle header to verify ID, cookie, and size
let mut header_buf = [0u8; NEEDLE_HEADER_SIZE];
match self.read_exact_at_backend(&mut header_buf, offset as u64) {
Ok(()) => {
let (cookie, id, size) = Needle::parse_header(&header_buf);
if id != needle_id {
broken.push(format!(
"needle {} header id mismatch: expected {}, got {}",
needle_id.0, needle_id.0, id.0
));
} else if cookie.0 == 0 {
broken.push(format!(
"needle {} has zero cookie at offset {}",
needle_id.0, offset
));
} else if size.0 <= 0 && !nv.size.is_deleted() {
broken.push(format!(
"needle {} has invalid size {} at offset {}",
needle_id.0, size.0, offset
));
}
// A zero-size index is only legal for a pre-allocated volume without
// data (e.g. after volume.grow); a populated .dat with an empty index
// is corruption the scrub must catch.
if idx_file_size == 0 {
match self.dat_file_size() {
Ok(dat_size) if dat_size > SUPER_BLOCK_SIZE as u64 => {
return Ok((
0,
vec![format!(
"zero-size IDX file for volume {} with store size {}",
self.id.0, dat_size
)],
));
}
Ok(_) => {}
Err(e) => {
broken.push(format!("needle {} read header error: {}", needle_id.0, e));
return Ok((
0,
vec![format!("stat data file for volume {}: {}", self.id.0, e)],
))
}
}
files_checked += 1;
}
Ok((files_checked, broken))
Ok(crate::storage::idx::check_index_file(
&mut idx_file,
idx_file_size,
self.version(),
))
}
/// Scrub the volume by reading and verifying all needles.
@@ -3914,6 +3899,38 @@ mod tests {
);
}
#[test]
fn test_scrub_index_flags_zero_size_idx_with_data() {
// A populated .dat with an empty .idx is corruption — the pre-allocated
// exception only covers a superblock-only .dat.
let tmp = TempDir::new().unwrap();
let dir = tmp.path().to_str().unwrap();
let mut v = make_test_volume(dir);
let data = b"needle data".to_vec();
let mut n = Needle {
id: NeedleId(1),
cookie: Cookie(1),
data: data.clone(),
data_size: data.len() as u32,
..Needle::default()
};
v.write_needle(&mut n, true).unwrap();
v.sync_to_disk().unwrap();
assert!(v.dat_file_size().unwrap() > SUPER_BLOCK_SIZE as u64);
// Truncate the .idx to zero while the .dat keeps its needle.
std::fs::File::create(v.file_name(".idx")).unwrap();
let (count, broken) = v.scrub_index().unwrap();
assert_eq!(count, 0);
assert!(
broken.iter().any(|e| e.contains("zero-size IDX file")),
"expected zero-size IDX error, got {:?}",
broken
);
}
#[test]
fn test_scrub_healthy_volume() {
// Mirror of Go's TestScrubVolumeData "healthy volume" case: a volume