[Volume] Scrub local deletion tombstones during FULL scrub (#11396)

* fix 11388

* fix(volume): scrub validates local deletion tombstones

TombstoneFileSize (-1) is an .idx-only sentinel; the physical record it
points at carries a zero-sized body. Normalize deleted index sizes to 0
via onDiskSize before computing disk usage and calling ReadData, so
corrupted or truncated tombstone records are detected instead of
skipped. Offset-zero entries (remote logical deletes, no .dat record)
remain skipped, and the physical needle id is checked against the index
key. Mirror the behavior in the Rust volume server.

* fix(volume): scrub preserves physical size of deleted non-tombstone entries

Size.Raw()/raw() already encodes the index-to-disk mapping: tombstone
(-1) -> 0, other negative sizes -> their absolute value (the offset then
points at the original record, per the ReadDeleted path). Use it instead
of mapping every deleted size to 0.

---------

Co-authored-by: Chris Lu <chris.lu@gmail.com>

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

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) {

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) {