diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index 281bcf5e2..fb3c3eb57 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -2834,11 +2834,9 @@ impl Volume { if offset.is_zero() && size.is_deleted() { return Ok(()); } - // Deleted needles still occupy .dat space: count their size, don't read. - total_read += get_actual_size(size, version); - if size.is_deleted() { - return Ok(()); - } + let on_disk_size = Size(size.raw() as i32); + // compute the actual size of the needle in disk, including needle header, body and alignment padding. + total_read += get_actual_size(on_disk_size, version); let actual_offset = offset.to_actual_offset(); if actual_offset < 0 || actual_offset as u64 >= dat_size { broken.push(format!( @@ -2853,13 +2851,21 @@ impl Volume { ..Needle::default() }; let mut read_option = ReadOption::default(); - if let Err(e) = - self.read_needle_data_at_unlocked(&mut n, actual_offset, size, &mut read_option) - { + if let Err(e) = self.read_needle_data_at_unlocked( + &mut n, + actual_offset, + on_disk_size, + &mut read_option, + ) { broken.push(format!( "failed to read needle {} on volume {}: {}", needle_id.0, self.id.0, e )); + } else if size.is_deleted() && n.id != needle_id { + broken.push(format!( + "index key {} does not match needle's Id {} on volume {}", + needle_id.0, n.id.0, self.id.0 + )); } Ok(()) }); @@ -4419,12 +4425,10 @@ 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. +/// Byte offset just past the needle's on-disk record. Mirrors Go's +/// needleDiskEnd. pub(crate) 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) + offset.to_actual_offset() + get_actual_size(Size(size.raw() as i32), version) } fn size_mismatch_error(offset: i64, id: NeedleId, found: Size, expected: Size) -> VolumeError { @@ -6075,6 +6079,112 @@ mod tests { assert_eq!(count, 2, "both .idx rows are walked"); } + /// The .dat offset a local delete's .idx row points at: the physical + /// tombstone record. Mirrors the Go test helper localTombstoneOffset. + fn local_tombstone_offset(v: &Volume, id: u64) -> Offset { + let mut idx_file = File::open(v.file_name(".idx")).unwrap(); + let mut found = Offset::default(); + idx::walk_index_file(&mut idx_file, 0, |key, offset, size| { + if key == NeedleId(id) && !offset.is_zero() && size.is_deleted() { + found = offset; + } + Ok(()) + }) + .unwrap(); + assert!( + !found.is_zero(), + "expected a local deletion tombstone for needle {} in .idx", + id + ); + found + } + + #[test] + fn test_scrub_checks_local_deletion_tombstone() { + // Mirror of Go's TestScrubVolumeDataChecksLocalDeletionTombstone: a + // local delete's tombstone record is read and its needle id checked + // against the .idx key. + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let mut v = make_test_volume(dir); + + write_test_needle(&mut v, 1, b"needle data"); + v.delete_needle(&mut Needle { + id: NeedleId(1), + cookie: Cookie(0x12345678), + ..Needle::default() + }) + .unwrap(); + // A later record keeps the tombstone mid-file rather than at the tail. + write_test_needle(&mut v, 2, b"needle data"); + v.sync_to_disk().unwrap(); + + let tombstone_offset = local_tombstone_offset(&v, 1); + let (_count, broken) = v.scrub().unwrap(); + assert!( + broken.is_empty(), + "healthy local deletion tombstone must pass scrub, got {:?}", + broken + ); + + let mut id_bytes = [0u8; NEEDLE_ID_SIZE]; + NeedleId(99).to_bytes(&mut id_bytes); + let mut dat = OpenOptions::new() + .write(true) + .open(v.file_name(".dat")) + .unwrap(); + dat.seek(SeekFrom::Start( + tombstone_offset.to_actual_offset() as u64 + COOKIE_SIZE as u64, + )) + .unwrap(); + dat.write_all(&id_bytes).unwrap(); + dat.sync_all().unwrap(); + + let (_count, broken) = v.scrub().unwrap(); + assert!( + broken + .iter() + .any(|e| e.contains("does not match needle's Id")), + "scrub should report the corrupted tombstone's Id, got {:?}", + broken + ); + } + + #[test] + fn test_scrub_reports_truncated_local_deletion_tombstone() { + // Mirror of Go's TestScrubVolumeDataReportsTruncatedLocalDeletionTombstone. + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let mut v = make_test_volume(dir); + + write_test_needle(&mut v, 1, b"needle data"); + v.delete_needle(&mut Needle { + id: NeedleId(1), + cookie: Cookie(0x12345678), + ..Needle::default() + }) + .unwrap(); + v.sync_to_disk().unwrap(); + + let tombstone_offset = local_tombstone_offset(&v, 1); + let dat = OpenOptions::new() + .write(true) + .open(v.file_name(".dat")) + .unwrap(); + dat.set_len(tombstone_offset.to_actual_offset() as u64 + NEEDLE_HEADER_SIZE as u64) + .unwrap(); + dat.sync_all().unwrap(); + + let (_count, broken) = v.scrub().unwrap(); + assert!( + broken + .iter() + .any(|e| e.contains("failed to read needle 1 on volume 1")), + "scrub should report the truncated tombstone read, got {:?}", + broken + ); + } + #[test] fn test_scrub_index_flags_zero_size_idx_with_data() { // A populated .dat with an empty .idx is corruption — the pre-allocated diff --git a/weed/storage/volume_checking.go b/weed/storage/volume_checking.go index 5961fb118..8fdb6bef9 100644 --- a/weed/storage/volume_checking.go +++ b/weed/storage/volume_checking.go @@ -85,21 +85,15 @@ func (v *Volume) scrubVolumeData(idxFile *os.File, idxFileSize int64) (int64, [] if offset.IsZero() && size.IsDeleted() { return nil } + physicalSize := types.Size(size.Raw()) // compute the actual size of the needle in disk, including needle header, body and alignment padding. - actualSize := int64(needle.GetActualSize(size, version)) - - // TODO: Needle.ReadData() is currently broken for deleted files, which have a types.Size < 0. Fix - // so deleted needles get properly scrubbed as well. - // TODO: idx.WalkIndexFile() returns a size -1 (and actual size of 32 bytes) for deleted needles. We - // want to scrub deleted needles whenever possible. - if size.IsDeleted() { - totalRead += actualSize - return nil - } + actualSize := int64(needle.GetActualSize(physicalSize, version)) n := needle.Needle{} - if err := n.ReadData(v.DataBackend, offset.ToActualOffset(), size, version); err != nil { + if err := n.ReadData(v.DataBackend, offset.ToActualOffset(), physicalSize, version); err != nil { errs = append(errs, fmt.Errorf("failed to read needle %d on volume %d: %v", id, v.Id, err)) + } else if size.IsDeleted() && n.Id != id { + errs = append(errs, fmt.Errorf("index key %v does not match needle's Id %v on volume %d", id, n.Id, v.Id)) } totalRead += actualSize @@ -149,7 +143,7 @@ func CheckVolumeDataIntegrity(v *Volume, indexFile *os.File) (lastAppendAtNs uin return 0, nil } // The deeper-than-tail structural check (every (offset + actual size) - // fits inside .dat — issue #8928) lives in volume.load(): it reads + // fits inside .dat) lives in volume.load(): it reads // MaximumNeedleEnd from the needle map after the load walk, so we don't // need a redundant linear scan of the .idx here. @@ -159,7 +153,7 @@ func CheckVolumeDataIntegrity(v *Volume, indexFile *os.File) (lastAppendAtNs uin // 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 (issue #9688). + // load. tailEntryPos, err := findDatTailEntryOffset(v, indexFile, indexSize) if err != nil { return 0, fmt.Errorf("CheckVolumeDataIntegrity %s: %v", indexFile.Name(), err) @@ -218,14 +212,8 @@ func findDatTailEntryOffset(v *Volume, indexFile *os.File, indexSize int64) (int } // needleDiskEnd returns the 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. func needleDiskEnd(offset types.Offset, size types.Size, version needle.Version) int64 { - onDiskSize := size - if size.IsDeleted() { - onDiskSize = 0 - } - return offset.ToActualOffset() + needle.GetActualSize(onDiskSize, version) + return offset.ToActualOffset() + needle.GetActualSize(types.Size(size.Raw()), version) } func doCheckAndFixVolumeData(v *Volume, indexFile *os.File, indexOffset int64) (lastAppendAtNs uint64, err error) { diff --git a/weed/storage/volume_checking_test.go b/weed/storage/volume_checking_test.go index 6ef90b7c6..75e32d013 100644 --- a/weed/storage/volume_checking_test.go +++ b/weed/storage/volume_checking_test.go @@ -5,6 +5,7 @@ import ( "os" "reflect" "sort" + "strings" "testing" "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" @@ -65,6 +66,16 @@ func TestScrubVolumeData(t *testing.T) { fmt.Errorf("failed to read needle 45 on volume 0: EOF"), fmt.Errorf("failed to read needle 46 on volume 0: EOF"), fmt.Errorf("failed to read needle 48 on volume 0: EOF"), + fmt.Errorf("failed to read needle 20 on volume 0: EOF"), + fmt.Errorf("failed to read needle 31 on volume 0: EOF"), + fmt.Errorf("failed to read needle 25 on volume 0: EOF"), + fmt.Errorf("failed to read needle 11 on volume 0: EOF"), + fmt.Errorf("failed to read needle 33 on volume 0: EOF"), + fmt.Errorf("failed to read needle 45 on volume 0: EOF"), + fmt.Errorf("failed to read needle 18 on volume 0: EOF"), + fmt.Errorf("failed to read needle 29 on volume 0: EOF"), + fmt.Errorf("failed to read needle 35 on volume 0: EOF"), + fmt.Errorf("failed to read needle 46 on volume 0: EOF"), fmt.Errorf("data file for volume 0 is smaller (8) than the 27 needles it contains (942856)"), }, }, @@ -164,18 +175,78 @@ func TestScrubVolumeData_IgnoresOffset0Tombstone(t *testing.T) { } f.Close() - idxFile, err := os.OpenFile(v.FileName(".idx"), os.O_RDONLY, 0644) - if err != nil { - t.Fatalf("reopen .idx: %v", err) + if errs := scrubVolumeErrors(t, v); len(errs) != 0 { + t.Fatalf("offset-0 tombstone must not flag the volume, got %v", errs) } - defer idxFile.Close() - idxStat, err := idxFile.Stat() +} + +func TestScrubVolumeDataChecksLocalDeletionTombstone(t *testing.T) { + dir := t.TempDir() + v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) if err != nil { - t.Fatalf("stat .idx: %v", err) + t.Fatalf("volume creation: %v", err) + } + defer v.Close() + + const deletedID = uint64(1) + if _, _, _, err := v.writeNeedle2(newRandomNeedle(deletedID), true, false, false); err != nil { + t.Fatalf("write needle: %v", err) + } + if _, err := v.doDeleteRequest(newEmptyNeedle(deletedID)); err != nil { + t.Fatalf("delete needle: %v", err) + } + // A later record keeps the tombstone mid-file rather than at the tail. + if _, _, _, err := v.writeNeedle2(newRandomNeedle(2), true, false, false); err != nil { + t.Fatalf("write needle after deletion: %v", err) + } + syncVolumeFiles(t, v) + + tombstoneOffset := localTombstoneOffset(t, v, deletedID) + if errs := scrubVolumeErrors(t, v); len(errs) != 0 { + t.Fatalf("healthy local deletion tombstone must pass scrub, got %v", errs) } - if _, errs := v.scrubVolumeData(idxFile, idxStat.Size()); len(errs) != 0 { - t.Fatalf("offset-0 tombstone must not flag the volume, got %v", errs) + corruptedID := make([]byte, types.NeedleIdSize) + types.NeedleIdToBytes(corruptedID, types.Uint64ToNeedleId(99)) + if _, err := v.DataBackend.WriteAt(corruptedID, tombstoneOffset.ToActualOffset()+types.CookieSize); err != nil { + t.Fatalf("corrupt tombstone ID: %v", err) + } + if err := v.DataBackend.Sync(); err != nil { + t.Fatalf("sync corrupted .dat: %v", err) + } + + if errs := scrubVolumeErrors(t, v); !strings.Contains(fmt.Sprint(errs), "does not match needle's Id") { + t.Fatalf("scrub should report the corrupted tombstone's Id, got %v", errs) + } +} + +func TestScrubVolumeDataReportsTruncatedLocalDeletionTombstone(t *testing.T) { + dir := t.TempDir() + v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("volume creation: %v", err) + } + defer v.Close() + + const deletedID = uint64(1) + if _, _, _, err := v.writeNeedle2(newRandomNeedle(deletedID), true, false, false); err != nil { + t.Fatalf("write needle: %v", err) + } + if _, err := v.doDeleteRequest(newEmptyNeedle(deletedID)); err != nil { + t.Fatalf("delete needle: %v", err) + } + syncVolumeFiles(t, v) + + tombstoneOffset := localTombstoneOffset(t, v, deletedID) + if err := v.DataBackend.Truncate(tombstoneOffset.ToActualOffset() + int64(types.NeedleHeaderSize)); err != nil { + t.Fatalf("truncate tombstone: %v", err) + } + if err := v.DataBackend.Sync(); err != nil { + t.Fatalf("sync truncated .dat: %v", err) + } + + if errs := scrubVolumeErrors(t, v); !strings.Contains(fmt.Sprint(errs), "failed to read needle 1 on volume 1: EOF") { + t.Fatalf("scrub should report the truncated tombstone read, got %v", errs) } } @@ -231,7 +302,7 @@ func TestCheckVolumeDataIntegrityWithDeletionTombstone(t *testing.T) { } } -// TestCheckVolumeDataIntegritySortedIndex reproduces issue #9688: after the +// TestCheckVolumeDataIntegritySortedIndex covers the case where the // .idx is rebuilt sorted by key — what weed fix and other rebuilds emitted via // needle_map.AscendingVisit — the last file-position entry is the highest-key // needle, not the needle at the .dat tail. The integrity check compared that @@ -325,7 +396,7 @@ func TestCheckVolumeDataIntegrityVerifiesDeletionTail(t *testing.T) { } } -// TestVolumeLoadStaysWritableWithKeySortedIndex reproduces issue #9688 end to +// TestVolumeLoadStaysWritableWithKeySortedIndex exercises end to // end through the real volume load path: a volume whose .idx is sorted by key // (the on-disk state weed fix left behind) must reload writable, not flip to // read-only — and a live needle must still be readable afterward. @@ -361,7 +432,7 @@ func TestVolumeLoadStaysWritableWithKeySortedIndex(t *testing.T) { } defer reloaded.Close() if reloaded.noWriteOrDelete { - t.Fatal("volume flipped read-only after reload with a key-sorted .idx (issue #9688)") + t.Fatal("volume flipped read-only after reload with a key-sorted .idx") } // The surviving needle is still readable, and new writes still succeed. @@ -414,9 +485,60 @@ func rewriteIdxSortedByKey(t *testing.T, idxPath string) { } } +// localTombstoneOffset returns the .dat offset a local delete's .idx row +// points at: the physical tombstone record. +func localTombstoneOffset(t *testing.T, v *Volume, id uint64) types.Offset { + t.Helper() + idxFile, err := os.OpenFile(v.FileName(".idx"), os.O_RDONLY, 0644) + if err != nil { + t.Fatalf("open .idx: %v", err) + } + defer idxFile.Close() + + var offset types.Offset + if err := idx.WalkIndexFile(idxFile, 0, func(key types.NeedleId, o types.Offset, size types.Size) error { + if key == types.Uint64ToNeedleId(id) && !o.IsZero() && size.IsDeleted() { + offset = o + } + return nil + }); err != nil { + t.Fatalf("walk .idx: %v", err) + } + if offset.IsZero() { + t.Fatalf("expected a local deletion tombstone for needle %d in .idx", id) + } + return offset +} + +// scrubVolumeErrors runs a full scrub over the volume's current .idx. +func scrubVolumeErrors(t *testing.T, v *Volume) []error { + t.Helper() + idxFile, err := os.OpenFile(v.FileName(".idx"), os.O_RDONLY, 0644) + if err != nil { + t.Fatalf("open .idx: %v", err) + } + defer idxFile.Close() + idxStat, err := idxFile.Stat() + if err != nil { + t.Fatalf("stat .idx: %v", err) + } + _, errs := v.scrubVolumeData(idxFile, idxStat.Size()) + return errs +} + +func syncVolumeFiles(t *testing.T, v *Volume) { + t.Helper() + if err := v.DataBackend.Sync(); err != nil { + t.Fatalf("sync .dat: %v", err) + } + if err := v.nm.Sync(); err != nil { + t.Fatalf("sync .idx: %v", err) + } +} + // TestMaxNeedleEnd ensures the needle map's MaxNeedleEnd accumulator lets // volume.load() detect an .idx that references bytes past the end of the .dat -// — the deeper-than-tail corruption shape from issue #8928 that the existing +// — the deeper-than-tail corruption shape that the existing // last-10-entries scan cannot see. The check is populated by the load walk // and read by volume.load() to flip the volume read-only. func TestMaxNeedleEnd(t *testing.T) {