From bd9b5c25ff40dd5c8a7544d1a8ae5e6e37ff66da Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 12 Jul 2026 19:37:58 -0700 Subject: [PATCH] rust volume: verify the .dat ends at the last indexed needle (#10320) * rust volume: verify the .dat ends at the last indexed needle The Go loader quarantines a volume whose .dat extends past the last indexed needle - the leftover of a torn shutdown - but the Rust check only verified header fields of the trailing index entries and never compared file sizes, so a torn tail loaded clean and writable. Appends land at the raw file end, and past a misaligned tail the next needle sits at an offset the 8-byte-unit .idx encoding rounds down, pointing the index a few bytes before the needle. Replace the last-10-entries walk with the current Go shape: find the entry physically last in the .dat (append-ordered fast path, max-offset scan for key-sorted rebuilds), verify that needle - tombstones with their on-disk Size=0 - and require the file to end exactly at it, marking the volume read-only otherwise. Claude-Session: https://claude.ai/code/session_01XgGXMLknzaNgQzHMyo2Vhb * rust volume: buffer the .idx max-offset scan The slow path read one 16-byte entry per syscall; a BufReader batches the sequential scan like Go's WalkIndexFile does. Claude-Session: https://claude.ai/code/session_01XgGXMLknzaNgQzHMyo2Vhb --- seaweed-volume/src/storage/volume.rs | 442 ++++++++++++++++++++++----- 1 file changed, 358 insertions(+), 84 deletions(-) diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index 77b0ccc0f..9a8a5a6a9 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -1866,104 +1866,216 @@ impl Volume { // needle map's max_needle_end accumulator, so we don't pay for a // second linear scan of the .idx here. - // Check last 10 index entries (matching Go's CheckVolumeDataIntegrity). - // Go starts healthyIndexSize = indexSize and reduces on EOF. - // On success: break (err != ErrorSizeMismatch when err == nil). - // On EOF: set healthyIndexSize = position of corrupt entry, continue. - // On ErrorSizeMismatch: continue (try next entry). - // After loop: if healthyIndexSize < indexSize → error. + // Verify the needle physically last in the .dat (the highest offset) + // and that the .dat ends exactly at it. The .idx is not always in .dat + // append order: weed fix and other rebuilds could rewrite it sorted by + // key, which puts the highest-key needle last instead of the .dat-tail + // needle. Picking the last file-position entry there compares a + // mid-file needle's tail against the full .dat size and falsely flips + // the volume read-only on every load. Mirrors Go's + // findDatTailEntryOffset + doCheckAndFixVolumeData. let mut idx_file = File::open(&idx_path)?; - let max_entries = std::cmp::min(10, idx_size / NEEDLE_MAP_ENTRY_SIZE as i64); - let mut healthy_index_size: i64 = idx_size; + let tail_entry_pos = self.find_dat_tail_entry_offset(&mut idx_file, idx_size, version)?; + if tail_entry_pos < 0 { + // pre-allocated / all-zero index, nothing to verify against the .dat + return Ok(()); + } - for i in 1..=max_entries { - let entry_offset = idx_size - i * NEEDLE_MAP_ENTRY_SIZE as i64; - let mut buf = [0u8; NEEDLE_MAP_ENTRY_SIZE]; - idx_file.seek(SeekFrom::Start(entry_offset as u64))?; - idx_file.read_exact(&mut buf)?; + let mut buf = [0u8; NEEDLE_MAP_ENTRY_SIZE]; + idx_file.seek(SeekFrom::Start(tail_entry_pos as u64))?; + idx_file.read_exact(&mut buf)?; + let (key, offset, size) = idx_entry_from_bytes(&buf); + if offset.is_zero() { + return Ok(()); + } + if size.is_deleted() { + // Deletion tombstone: the .idx carries TombstoneFileSize (-1) but + // the appended needle in .dat carries Size=0, so verify with + // Size=0 — comparing against -1 would mismatch the on-disk header + // and send every volume with a trailing deletion read-only. + self.verify_deleted_needle_integrity(offset.to_actual_offset(), key, version) + } else { + self.verify_needle_integrity(offset.to_actual_offset(), key, size, version) + } + } - let (key, offset, size) = idx_entry_from_bytes(&buf); - if offset.is_zero() { - continue; - } - - let actual_offset = offset.to_actual_offset() as u64; - - // Read needle header at the offset - let mut header = [0u8; NEEDLE_HEADER_SIZE]; - match self.read_exact_at_backend(&mut header, actual_offset) { - Ok(()) => {} - Err(VolumeError::Io(e)) if e.kind() == io::ErrorKind::UnexpectedEof => { - // Match Go: on EOF, mark this entry as corrupt and continue - // checking earlier entries (healthyIndexSize tracks the boundary). - healthy_index_size = entry_offset; - continue; - } - Err(e) => return Err(e), - } - - let (_cookie, needle_id, needle_size) = Needle::parse_header(&header); - - // Verify the needle ID matches the index entry - if !key.is_empty() && needle_id != key { - return Err(VolumeError::Io(io::Error::new( - io::ErrorKind::InvalidData, - format!( - "index key {:?} does not match needle Id {:?} at offset {}", - key, needle_id, actual_offset - ), - ))); - } - - // For non-deleted entries, verify the size matches - if !size.is_deleted() && size.0 > 0 && needle_size.0 != size.0 { - // Try with MaxPossibleVolumeSize offset adjustment (Go parity) - let alt_offset = actual_offset + MAX_POSSIBLE_VOLUME_SIZE as u64; - let mut alt_header = [0u8; NEEDLE_HEADER_SIZE]; - if self - .read_exact_at_backend(&mut alt_header, alt_offset) - .is_ok() - { - let (_, _, alt_size) = Needle::parse_header(&alt_header); - if alt_size.0 == size.0 { - continue; - } - } - // Match Go: ErrorSizeMismatch breaks out of the loop - break; - } - - // If V3, try to read the append timestamp from the last verified entry. - // Go reads AppendAtNs from both live and deleted (tombstone) entries - // via verifyNeedleIntegrity and verifyDeletedNeedleIntegrity. - if version == VERSION_3 { - // For tombstones (deleted), body size on disk is 0. - // For live entries, body size is size.0. - let body_size = if size.is_deleted() { 0u64 } else { size.0 as u64 }; - let ts_offset = - actual_offset + NEEDLE_HEADER_SIZE as u64 + body_size + 4; // skip checksum - let mut ts_buf = [0u8; 8]; - if self.read_exact_at_backend(&mut ts_buf, ts_offset).is_ok() { - let ts = u64::from_be_bytes(ts_buf); - if ts > 0 { - self.last_append_at_ns = ts; - } + /// .idx file position of the entry whose needle is physically last in the + /// .dat (highest offset). The common case — an append-ordered .idx — is + /// resolved in O(1): the last entry's on-disk end equals the .dat size. A + /// key-sorted .idx (e.g. legacy weed fix output) falls back to a single + /// linear scan for the maximum offset. Returns -1 when no entry has a + /// non-zero offset (a pre-allocated / all-zero index). + fn find_dat_tail_entry_offset( + &self, + idx_file: &mut File, + idx_size: i64, + version: Version, + ) -> Result { + if let Ok(dat_size) = self.dat_file_size() { + if dat_size > 0 { + let last_pos = idx_size - NEEDLE_MAP_ENTRY_SIZE as i64; + let mut buf = [0u8; NEEDLE_MAP_ENTRY_SIZE]; + idx_file.seek(SeekFrom::Start(last_pos as u64))?; + idx_file.read_exact(&mut buf)?; + let (_, offset, size) = idx_entry_from_bytes(&buf); + if !offset.is_zero() && needle_disk_end(offset, size, version) == dat_size as i64 { + return Ok(last_pos); } } } - // Match Go: if healthyIndexSize < indexSize, trailing entries are corrupt - if healthy_index_size < idx_size { + // Slow path: scan the whole .idx for the entry at the maximum offset. + // Tombstone entries are not skipped: a trailing deletion is a real + // on-disk record and can be the needle physically last in the .dat. + let mut max_entry_pos: i64 = -1; + let mut max_actual_offset: i64 = -1; + let mut entry_pos: i64 = 0; + let mut buf = [0u8; NEEDLE_MAP_ENTRY_SIZE]; + idx_file.seek(SeekFrom::Start(0))?; + let mut reader = io::BufReader::new(idx_file); + while entry_pos < idx_size { + reader.read_exact(&mut buf)?; + let (_, offset, _) = idx_entry_from_bytes(&buf); + if !offset.is_zero() { + let ao = offset.to_actual_offset(); + if ao > max_actual_offset { + max_actual_offset = ao; + max_entry_pos = entry_pos; + } + } + entry_pos += NEEDLE_MAP_ENTRY_SIZE as i64; + } + Ok(max_entry_pos) + } + + /// Verify the live needle at the .dat tail: its header must match the + /// .idx entry and the .dat must end exactly at its on-disk end. Extra + /// bytes mean an unindexed trailing record (a torn append); appending + /// after one would place the next needle at an offset the 8-byte .idx + /// encoding may not represent, so the volume is quarantined read-only + /// instead. Mirrors Go's verifyNeedleIntegrity. + fn verify_needle_integrity( + &mut self, + actual_offset: i64, + key: NeedleId, + size: Size, + version: Version, + ) -> Result<(), VolumeError> { + let mut header = [0u8; NEEDLE_HEADER_SIZE]; + self.read_exact_at_backend(&mut header, actual_offset as u64)?; + let (_cookie, needle_id, needle_size) = Needle::parse_header(&header); + + let mut checked_offset = actual_offset; + let mut checked_id = needle_id; + if needle_size != size { + // Go parity: on a size mismatch, retry past the 32GB offset + // wrap-around before giving up. + let alt_offset = actual_offset + MAX_POSSIBLE_VOLUME_SIZE as i64; + let mut alt_header = [0u8; NEEDLE_HEADER_SIZE]; + self.read_exact_at_backend(&mut alt_header, alt_offset as u64) + .map_err(|_| size_mismatch_error(actual_offset, needle_id, needle_size, size))?; + let (_, alt_id, alt_size) = Needle::parse_header(&alt_header); + if alt_size != size { + return Err(size_mismatch_error(actual_offset, needle_id, needle_size, size)); + } + checked_offset = alt_offset; + checked_id = alt_id; + } + + if !key.is_empty() && checked_id != key { return Err(VolumeError::Io(io::Error::new( io::ErrorKind::InvalidData, format!( - "healthy index size {} is less than expected {}", - healthy_index_size, idx_size + "index key {:?} does not match needle Id {:?} at offset {}", + key, checked_id, checked_offset ), ))); } - Ok(()) + if version == VERSION_3 { + let ts_offset = + checked_offset as u64 + NEEDLE_HEADER_SIZE as u64 + size.0 as u64 + 4; // skip checksum + let mut ts_buf = [0u8; 8]; + self.read_exact_at_backend(&mut ts_buf, ts_offset)?; + let ts = u64::from_be_bytes(ts_buf); + if ts > 0 { + self.last_append_at_ns = ts; + } + } + + self.verify_dat_ends_at(checked_offset + get_actual_size(size, version)) + } + + /// Verify the deletion tombstone at the .dat tail. Tombstones are appended + /// with DataSize=0, so the on-disk header carries Size=0; verify with that + /// and then require the .dat to end exactly at the tombstone. Mirrors Go's + /// verifyDeletedNeedleIntegrity. + fn verify_deleted_needle_integrity( + &mut self, + actual_offset: i64, + key: NeedleId, + version: Version, + ) -> Result<(), VolumeError> { + let record_size = get_actual_size(Size(0), version) as usize; + let mut record = vec![0u8; record_size]; + self.read_exact_at_backend(&mut record, actual_offset as u64)?; + + let mut n = Needle::default(); + n.read_bytes(&record, actual_offset, Size(0), version) + .map_err(|e| { + VolumeError::Io(io::Error::new( + io::ErrorKind::InvalidData, + format!("read tombstone at offset {}: {}", actual_offset, e), + )) + })?; + if !key.is_empty() && n.id != key { + return Err(VolumeError::Io(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "index key {:?} does not match needle Id {:?} at offset {}", + key, n.id, actual_offset + ), + ))); + } + if n.append_at_ns > 0 { + self.last_append_at_ns = n.append_at_ns; + } + + self.verify_dat_ends_at(actual_offset + get_actual_size(Size(0), version)) + } + + /// The checked needle is the one physically last in the .dat, so the file + /// must end exactly at it; extra bytes mean an unindexed trailing record. + fn verify_dat_ends_at(&self, file_tail_offset: i64) -> Result<(), VolumeError> { + let file_size = self.dat_file_size().map_err(VolumeError::Io)? as i64; + if file_size == file_tail_offset { + return Ok(()); + } + let dat_path = self.file_name(".dat"); + if file_size > file_tail_offset { + warn!( + "data file {} actual {} bytes expected {} bytes!", + dat_path, file_size, file_tail_offset + ); + return Err(VolumeError::Io(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "data file {} actual {} bytes expected {} bytes", + dat_path, file_size, file_tail_offset + ), + ))); + } + warn!( + "data file {} has {} bytes, less than expected {} bytes!", + dat_path, file_size, file_tail_offset + ); + Err(VolumeError::Io(io::Error::new( + io::ErrorKind::UnexpectedEof, + format!( + "data file {} has {} bytes, less than expected {} bytes", + dat_path, file_size, file_tail_offset + ), + ))) } /// Open the on-disk .idx for scrubbing and apply Go openIndex's zero-size @@ -3421,6 +3533,24 @@ impl Volume { // ============================================================================ /// Generate volume file base name: dir/collection_id or dir/id +/// Byte offset just past the needle's on-disk record. Deletion tombstones +/// carry TombstoneFileSize (-1) in the .idx but are written with DataSize=0, +/// so their on-disk record is sized as 0. Mirrors Go's needleDiskEnd. +fn needle_disk_end(offset: Offset, size: Size, version: Version) -> i64 { + let on_disk_size = if size.is_deleted() { Size(0) } else { size }; + offset.to_actual_offset() + get_actual_size(on_disk_size, version) +} + +fn size_mismatch_error(offset: i64, id: NeedleId, found: Size, expected: Size) -> VolumeError { + VolumeError::Io(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "entry not found: offset {} found id {:?} size {}, expected size {}", + offset, id, found.0, expected.0 + ), + )) +} + pub fn volume_file_name(dir: &str, collection: &str, id: VolumeId) -> String { if collection.is_empty() { format!("{}/{}", dir, id.0) @@ -3881,6 +4011,150 @@ mod tests { ); } + fn write_three_needles(dir: &str) { + let mut v = make_test_volume(dir); + for i in 1..=3u64 { + let data = format!("data {}", i); + let mut n = Needle { + id: NeedleId(i), + cookie: Cookie(i as u32), + data: data.as_bytes().to_vec(), + data_size: data.len() as u32, + ..Needle::default() + }; + v.write_needle(&mut n, true).unwrap(); + } + v.sync_to_disk().unwrap(); + } + + fn reload_volume(dir: &str) -> Volume { + Volume::new( + dir, + dir, + "", + VolumeId(1), + NeedleMapKind::InMemory, + None, + None, + 0, + Version::current(), + ) + .unwrap() + } + + // A torn shutdown can leave bytes in the .dat past the last indexed + // needle. Appending after such a tail would place the next needle at an + // offset the 8-byte-unit .idx encoding may not represent, so the load + // check must quarantine the volume read-only, matching Go's + // verifyNeedleIntegrity "actual N bytes expected M bytes" behavior. + #[test] + fn test_integrity_detects_unindexed_dat_tail() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + write_three_needles(dir); + + let dat_path = volume_file_name(dir, "", VolumeId(1)) + ".dat"; + let mut f = OpenOptions::new().append(true).open(&dat_path).unwrap(); + f.write_all(&[0xFFu8; 100]).unwrap(); + f.sync_all().unwrap(); + drop(f); + + let v = reload_volume(dir); + assert!( + v.is_no_write_or_delete(), + "volume with unindexed .dat tail must load read-only" + ); + } + + // Same torn tail, but with a deletion tombstone as the last indexed + // record — the tombstone path must also verify the .dat ends at it. + #[test] + fn test_integrity_detects_tail_after_tombstone() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + { + let mut v = make_test_volume(dir); + let data = b"data 1".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.delete_needle(&mut Needle { + id: NeedleId(1), + cookie: Cookie(1), + ..Needle::default() + }) + .unwrap(); + v.sync_to_disk().unwrap(); + } + + let dat_path = volume_file_name(dir, "", VolumeId(1)) + ".dat"; + let mut f = OpenOptions::new().append(true).open(&dat_path).unwrap(); + f.write_all(&[0xFFu8; 64]).unwrap(); + f.sync_all().unwrap(); + drop(f); + + let v = reload_volume(dir); + assert!( + v.is_no_write_or_delete(), + "volume with unindexed tail after tombstone must load read-only" + ); + } + + // A key-sorted .idx (legacy weed fix output) puts the highest-key needle + // last by file position, not the .dat-tail needle. The check must find + // the true tail entry via the max-offset scan instead of falsely flipping + // a healthy volume read-only. + #[test] + fn test_integrity_key_sorted_idx_no_false_readonly() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + write_three_needles(dir); + + // Swap the last two .idx entries so the file-position-last entry + // references a mid-file needle. + let idx_path = volume_file_name(dir, "", VolumeId(1)) + ".idx"; + let content = fs::read(&idx_path).unwrap(); + assert_eq!(content.len(), 3 * NEEDLE_MAP_ENTRY_SIZE); + let mut swapped = content.clone(); + swapped[NEEDLE_MAP_ENTRY_SIZE..2 * NEEDLE_MAP_ENTRY_SIZE] + .copy_from_slice(&content[2 * NEEDLE_MAP_ENTRY_SIZE..]); + swapped[2 * NEEDLE_MAP_ENTRY_SIZE..] + .copy_from_slice(&content[NEEDLE_MAP_ENTRY_SIZE..2 * NEEDLE_MAP_ENTRY_SIZE]); + fs::write(&idx_path, swapped).unwrap(); + + let v = reload_volume(dir); + assert!( + !v.is_no_write_or_delete(), + "healthy volume with key-sorted .idx must not load read-only" + ); + } + + // A .dat truncated mid-needle (crash during append) must quarantine. + #[test] + fn test_integrity_detects_truncated_dat_tail() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + write_three_needles(dir); + + let dat_path = volume_file_name(dir, "", VolumeId(1)) + ".dat"; + let len = fs::metadata(&dat_path).unwrap().len(); + let f = OpenOptions::new().write(true).open(&dat_path).unwrap(); + f.set_len(len - 5).unwrap(); + f.sync_all().unwrap(); + drop(f); + + let v = reload_volume(dir); + assert!( + v.is_no_write_or_delete(), + "volume with truncated .dat tail must load read-only" + ); + } + #[test] fn test_scrub_empty_volume() { // Mirror of Go's TestScrubVolumeData "zero-size volume without index"