volume: repair torn tails on canDelete volumes; reject damaged 1-3 byte records (#11876)

- Go/Rust: run the unindexed .dat tail repair on noWriteCanDelete volumes
  too — deletes still append tombstone records, so the same crash window
  applies and they previously degraded to fully read-only.
- Go readTailNeedle: reject a positive size of 1-3 for v2+ as corrupted
  before ReadBytes slices the body's 4-byte data-size field and panics;
  a torn last record is now cut instead of crashing the load.
- Tests: TestVolumeLoadRepairsTailOnCanDeleteVolume and
  TestVolumeLoadCutsDamagedSmallSizeTailOnCanDeleteVolume (Go), and the
  Rust mirrors volume_load_repairs_tail_on_can_delete_volume /
  volume_load_cuts_damaged_small_size_tail.

Generated with [Devin](https://devin.ai)

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

seaweed-volume/src/storage/volume.rs

@@ -1732,7 +1732,9 @@ impl Volume {
             // offset. Mirrors Go running repairUnindexedDatTail inside
             // CheckVolumeDataIntegrity ahead of the needle map load. Only a
             // volume about to load writable is repaired.
-            if !self.has_remote_file() && self.no_write_flags() == (false, false) {
+            // A no_write_can_delete volume still appends tombstone records on
+            // delete, so it tears tails the same way and gets the same repair.
+            if !self.has_remote_file() && !self.no_write_flags().0 {
                 // An empty .idx gets the same repair: every check below
                 // skips an empty index on its own, so a tail the repair
                 // cannot explain there fails the load like in Go rather

seaweed-volume/src/storage/volume_tail_repair.rs

@@ -1145,4 +1145,71 @@ mod tests {
         assert_eq!(fs::metadata(&dat_path).unwrap().len(), dat_before);
         assert_eq!(fs::metadata(&idx_path).unwrap().len(), idx_before);
     }
+
+    /// A canDelete volume still appends a tombstone record on every delete,
+    /// so it can tear a tail the same way: it is repaired and keeps taking
+    /// deletes. Mirrors Go's TestVolumeLoadRepairsTailOnCanDeleteVolume.
+    #[test]
+    fn volume_load_repairs_tail_on_can_delete_volume() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap().to_string();
+        let mut last_data = Vec::new();
+        {
+            let v = open_volume(&dir);
+            for id in [1u64, 2, 99] {
+                let mut n = new_needle(id);
+                v.write_needle(&mut n, true, false).unwrap();
+                last_data = n.data.clone();
+            }
+            v.set_read_only_persist(true, true).unwrap();
+        }
+        let idx_path = format!("{dir}/1.idx");
+        let idx_len = fs::metadata(&idx_path).unwrap().len();
+        OpenOptions::new()
+            .write(true)
+            .open(&idx_path)
+            .unwrap()
+            .set_len(idx_len - NEEDLE_MAP_ENTRY_SIZE as u64)
+            .unwrap();
+
+        let v = open_volume(&dir);
+        assert!(
+            !v.is_no_write_or_delete() && v.is_no_write_can_delete(),
+            "canDelete volume with a torn tail did not load canDelete after the repair"
+        );
+        assert_readable(&v, 99, &last_data);
+    }
+
+    /// A torn record whose header advertises a positive size smaller than
+    /// the body's own 4-byte data-size field is damaged, not decodable: the
+    /// decoder's range check rejects it, and as the last record it is cut.
+    /// Mirrors Go's TestVolumeLoadCutsDamagedSmallSizeTailOnCanDeleteVolume.
+    #[test]
+    fn volume_load_cuts_damaged_small_size_tail() {
+        let f = build_fixture(&writes(&[1, 2, 3]), 1);
+        let mut record = vec![0u8; get_actual_size(Size(2), VERSION_3) as usize];
+        Cookie(0xabcd).to_bytes(&mut record[0..COOKIE_SIZE]);
+        NeedleId(7).to_bytes(&mut record[COOKIE_SIZE..COOKIE_SIZE + NEEDLE_ID_SIZE]);
+        Size(2).to_bytes(
+            &mut record[COOKIE_SIZE + NEEDLE_ID_SIZE..NEEDLE_HEADER_SIZE],
+        );
+        OpenOptions::new()
+            .append(true)
+            .open(f.dat_path())
+            .unwrap()
+            .write_all(&record)
+            .unwrap();
+
+        let v = open_volume(&f.dir);
+        assert!(
+            !v.is_read_only(),
+            "damaged tail record loaded the volume read-only"
+        );
+        assert_not_readable(&v, 7);
+        assert_eq!(
+            f.dat_size(),
+            f.end,
+            "damaged tail was not cut"
+        );
+    }
 }

weed/storage/volume_checking.go

@@ -145,7 +145,7 @@ func CheckVolumeDataIntegrity(v *Volume, indexFile *os.File) (lastAppendAtNs uin
 		return 0, fmt.Errorf("verifyIndexFileIntegrity %s failed: %v", indexFile.Name(), err)
 	}
 	if indexSize == 0 {
-		if v.noWriteOrDelete || v.noWriteCanDelete {
+		if v.noWriteOrDelete {
 			return 0, nil
 		}
 		// An empty index can hide writes a crash stranded in the .dat: the
@@ -194,7 +194,7 @@ func CheckVolumeDataIntegrity(v *Volume, indexFile *os.File) (lastAppendAtNs uin
 	// A repair that fails for any other reason than an unexplained tail (a
 	// read, write, truncate or sync error) fails the check, so the volume
 	// loads read-only instead of writable on a repair that may not be on disk.
-	if !v.noWriteOrDelete && !v.noWriteCanDelete {
+	if !v.noWriteOrDelete {
 		reindexed, _, err := repairUnindexedDatTail(v, indexFile, indexSize, tailEntryPos)
 		if err != nil {
 			if !errors.Is(err, errTailNotTorn) {
@@ -428,6 +428,11 @@ func appendIndexEntries(indexFile indexAppender, entries []byte, at int64) error
 // readTailNeedle reads and verifies the whole record of a needle of the given
 // on-disk size at offset: header, body, CRC and timestamp.
 func readTailNeedle(datFile backend.BackendStorageFile, version needle.Version, at int64, size types.Size) (*needle.Needle, error) {
+	if size > 0 && size < 4 && version != needle.Version1 {
+		// a v2+ body always starts with its 4-byte data-size field; size 1-3
+		// is damaged bytes, not a record — decoding it slices out of range
+		return nil, fmt.Errorf("record at %d has size %d: %w", at, size, needle.ErrorCorrupted)
+	}
 	blob, err := needle.ReadNeedleBlob(datFile, at, size, version)
 	if err != nil {
 		return nil, err

weed/storage/volume_checking_tail_repair_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/seaweedfs/seaweedfs/weed/util"
 )
 
 // A torn tail is what a crash between the two appends of a write leaves: the
@@ -548,6 +549,88 @@ func TestVolumeLoadLeavesReadOnlyVolumeTailAlone(t *testing.T) {
 	}
 }
 
+// A canDelete volume still appends a tombstone record on every delete, so it
+// can tear a tail the same way: it is repaired and keeps taking deletes.
+func TestVolumeLoadRepairsTailOnCanDeleteVolume(t *testing.T) {
+	dir := t.TempDir()
+	v := newTornTailTestVolume(t, dir)
+	var last *needle.Needle
+	for _, id := range []uint64{1, 2, 99} {
+		last = newRandomNeedle(id)
+		if _, _, _, err := v.writeNeedle2(last, true, false, false); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := v.PersistReadOnly(true, true); err != nil {
+		t.Fatal(err)
+	}
+	idxName := v.FileName(".idx")
+	v.Close()
+	if err := os.Truncate(idxName, fileSize(t, idxName)-types.NeedleMapEntrySize); err != nil {
+		t.Fatal(err)
+	}
+
+	v = newTornTailTestVolume(t, dir)
+	defer v.Close()
+	if v.noWriteOrDelete || !v.noWriteCanDelete {
+		t.Fatal("canDelete volume with a torn tail did not load canDelete after the repair")
+	}
+	assertReadable(t, v, last)
+}
+
+// A torn record whose header advertises a positive size smaller than the
+// body's own 4-byte data-size field is damaged, not decodable — reading it
+// used to panic on the slice. As the last record it is cut, and a canDelete
+// volume keeps taking deletes after the repair.
+func TestVolumeLoadCutsDamagedSmallSizeTailOnCanDeleteVolume(t *testing.T) {
+	dir := t.TempDir()
+	v := newTornTailTestVolume(t, dir)
+	var last *needle.Needle
+	for _, id := range []uint64{1, 2, 99} {
+		last = newRandomNeedle(id)
+		if _, _, _, err := v.writeNeedle2(last, true, false, false); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := v.PersistReadOnly(true, true); err != nil {
+		t.Fatal(err)
+	}
+	datName, idxName := v.FileName(".dat"), v.FileName(".idx")
+	version := v.Version()
+	v.Close()
+	if err := os.Truncate(idxName, fileSize(t, idxName)-types.NeedleMapEntrySize); err != nil {
+		t.Fatal(err)
+	}
+	datBefore := fileSize(t, datName)
+
+	// a complete-looking record whose header claims a 2-byte body
+	damagedSize := types.Size(2)
+	record := make([]byte, needle.GetActualSize(damagedSize, version))
+	util.Uint32toBytes(record[0:types.CookieSize], 0xabcd)
+	util.Uint64toBytes(record[types.CookieSize:types.CookieSize+types.NeedleIdSize], uint64(types.Uint64ToNeedleId(7)))
+	util.Uint32toBytes(record[types.CookieSize+types.NeedleIdSize:types.NeedleHeaderSize], uint32(damagedSize))
+	dat, err := os.OpenFile(datName, os.O_WRONLY|os.O_APPEND, 0644)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := dat.Write(record); err != nil {
+		t.Fatal(err)
+	}
+	if err := dat.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	v = newTornTailTestVolume(t, dir)
+	defer v.Close()
+	if v.noWriteOrDelete || !v.noWriteCanDelete {
+		t.Fatal("canDelete volume with a damaged-size tail did not load canDelete after the repair")
+	}
+	assertReadable(t, v, last)
+	if got := fileSize(t, datName); got != datBefore {
+		t.Fatalf("damaged tail was not cut: .dat is %d bytes, want %d", got, datBefore)
+	}
+}
+
 // An .idx that ends in a partial entry keeps the existing behaviour:
 // the volume does not load writable and neither file is changed.
 func TestVolumeLoadLeavesPartialIdxEntryAlone(t *testing.T) {