fix(volume): return an error instead of panicking on a corrupt needle size (#11393)

ReadNeedleBodyBytes sliced the needle body with the size from the needle
header without checking it. A corrupted .dat header carrying size -1 still
gets a positive body length (16 bytes on v3), so vacuum compaction read
that body and panicked with "slice bounds out of range [:-1]".

Writers never put a negative size in a .dat header: a delete appends a
size-0 record, and TombstoneFileSize only lives in the .idx. Reject a size
that is negative or leaves no room for the checksum/timestamp tail with an
error wrapping ErrorCorrupted. ScanVolumeFileFrom already logs body read
errors and moves on, so compaction now skips the record like any other
corrupt needle.

Fixes #6763

weed/storage/needle/needle_read.go

@@ -224,8 +224,15 @@ func (n *Needle) ReadNeedleBody(r backend.BackendStorageFile, version Version, o
 
 func (n *Needle) ReadNeedleBodyBytes(needleBody []byte, version Version) (err error) {
 
-	if len(needleBody) <= 0 {
-		return nil
+	// n.Size comes from the on-disk header, so a corrupted header can carry a
+	// negative size or one the body cannot hold along with its tail.
+	tailSize := NeedleChecksumSize
+	if version == Version3 {
+		tailSize += TimestampSize
+	}
+	if n.Size < 0 || int64(n.Size)+int64(tailSize) > int64(len(needleBody)) {
+		stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorIndexOutOfRange).Inc()
+		return fmt.Errorf("needle %v size %d out of range for body length %d: %w", n.Id, n.Size, len(needleBody), ErrorCorrupted)
 	}
 	switch version {
 	case Version1:

weed/storage/needle/needle_read_test.go

@@ -0,0 +1,87 @@
+package needle
+
+import (
+	"bytes"
+	"errors"
+	"testing"
+
+	. "github.com/seaweedfs/seaweedfs/weed/storage/types"
+)
+
+// readNeedleBodyBytes runs ReadNeedleBodyBytes and turns a panic into a test
+// failure, so a regression reports which case broke instead of killing the run.
+func readNeedleBodyBytes(t *testing.T, n *Needle, body []byte, version Version) (err error) {
+	t.Helper()
+	defer func() {
+		if r := recover(); r != nil {
+			t.Fatalf("ReadNeedleBodyBytes panicked for size %d, body length %d: %v", n.Size, len(body), r)
+		}
+	}()
+	return n.ReadNeedleBodyBytes(body, version)
+}
+
+// A corrupted .dat header can carry a size that does not fit the body read for
+// it. Vacuum used to panic on it with "slice bounds out of range [:-1]" (#6763).
+func TestReadNeedleBodyBytesRejectsCorruptSize(t *testing.T) {
+	for _, version := range []Version{Version1, Version2, Version3} {
+		t.Run(versionString(version), func(t *testing.T) {
+			// A size of -1 is the case from #6763: its body length is still
+			// positive, so the scan reads a body and hands it over.
+			bodyLength := NeedleBodyLength(-1, version)
+			if bodyLength <= 0 {
+				t.Fatalf("expected a positive body length for size -1, got %d", bodyLength)
+			}
+
+			cases := []struct {
+				name string
+				size Size
+				body int
+			}{
+				{"size -1", -1, int(bodyLength)},
+				{"size -12", -12, 32},
+				{"size larger than body", 64, 32},
+				{"no room for the tail", 32, 32},
+				{"empty body", 0, 0},
+				{"empty body with data size", 1, 0},
+			}
+			for _, c := range cases {
+				t.Run(c.name, func(t *testing.T) {
+					n := &Needle{Size: c.size}
+					err := readNeedleBodyBytes(t, n, make([]byte, c.body), version)
+					if !errors.Is(err, ErrorCorrupted) {
+						t.Fatalf("expected an error wrapping ErrorCorrupted, got %v", err)
+					}
+				})
+			}
+		})
+	}
+}
+
+// The size guard must still accept every record the writer produces,
+// including the size-0 record a delete appends.
+func TestReadNeedleBodyBytesWrittenNeedles(t *testing.T) {
+	for _, version := range []Version{Version1, Version2, Version3} {
+		t.Run(versionString(version), func(t *testing.T) {
+			for _, data := range [][]byte{nil, []byte("hello seaweed")} {
+				written := &Needle{Id: 7, Cookie: 9, Data: data, Checksum: NewCRC(data), AppendAtNs: 42}
+				buf := new(bytes.Buffer)
+				if _, _, err := writeNeedleByVersion(version, written, 0, buf); err != nil {
+					t.Fatalf("write needle: %v", err)
+				}
+
+				n := new(Needle)
+				n.ParseNeedleHeader(buf.Bytes())
+				body := buf.Bytes()[NeedleHeaderSize:]
+				if int64(len(body)) != NeedleBodyLength(n.Size, version) {
+					t.Fatalf("body length %d, want %d", len(body), NeedleBodyLength(n.Size, version))
+				}
+				if err := readNeedleBodyBytes(t, n, body, version); err != nil {
+					t.Fatalf("read %d-byte needle: %v", len(data), err)
+				}
+				if !bytes.Equal(n.Data, data) {
+					t.Fatalf("data %q, want %q", n.Data, data)
+				}
+			}
+		})
+	}
+}

weed/storage/volume_vacuum_test.go

@@ -408,6 +408,68 @@ func TestCompactByIndex_ConcurrentWriteDoesNotFailIntegrityCheck(t *testing.T) {
 	}
 }
 
+// TestCompactByVolumeData_SkipsNegativeSizeHeader covers issue #6763: a .dat
+// record whose header carries size -1 made compaction panic with "slice bounds
+// out of range [:-1]". The corrupt record must be skipped and the needles on
+// both sides of it kept.
+func TestCompactByVolumeData_SkipsNegativeSizeHeader(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()
+
+	if _, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, false, false); err != nil {
+		t.Fatalf("write needle 1: %v", err)
+	}
+
+	// Size -1 still yields a positive body length, so the scan reads a body
+	// for this record and parses it.
+	datSize, _, err := v.DataBackend.GetStat()
+	if err != nil {
+		t.Fatalf("stat .dat: %v", err)
+	}
+	corrupt := make([]byte, needle.GetActualSize(-1, v.Version()))
+	types.NeedleIdToBytes(corrupt[types.CookieSize:types.CookieSize+types.NeedleIdSize], types.Uint64ToNeedleId(99))
+	types.SizeToBytes(corrupt[types.CookieSize+types.NeedleIdSize:types.NeedleHeaderSize], -1)
+	if _, err := v.DataBackend.WriteAt(corrupt, datSize); err != nil {
+		t.Fatalf("append corrupt record: %v", err)
+	}
+
+	if _, _, _, err := v.writeNeedle2(newRandomNeedle(2), true, false, false); err != nil {
+		t.Fatalf("write needle 2: %v", err)
+	}
+
+	if err := v.CompactByVolumeData(nil); err != nil {
+		t.Fatalf("CompactByVolumeData: %v", err)
+	}
+
+	cpx, err := os.Open(filepath.Join(dir, "1.cpx"))
+	if err != nil {
+		t.Fatalf("open .cpx: %v", err)
+	}
+	defer cpx.Close()
+	kept := map[types.NeedleId]bool{}
+	if err := idx.WalkIndexFile(cpx, 0, func(key types.NeedleId, _ types.Offset, size types.Size) error {
+		if size.IsValid() {
+			kept[key] = true
+		}
+		return nil
+	}); err != nil {
+		t.Fatalf("walk .cpx: %v", err)
+	}
+	for _, id := range []uint64{1, 2} {
+		if !kept[types.Uint64ToNeedleId(id)] {
+			t.Errorf("needle %d missing from the compacted index", id)
+		}
+	}
+	if kept[types.Uint64ToNeedleId(99)] {
+		t.Errorf("corrupt record 99 should not be in the compacted index")
+	}
+}
+
 // TestExceedsExpectedCompactedSize guards the copy-phase integrity check
 // against regressing into double-subtracting skipped bytes: expectedLiveBytes
 // already excludes needles dropped as unreadable (they return before being