fix(volume): derive needle body tail bound from the version layout (#11395)

* fix(volume): derive needle body tail bound from the version layout

The size guard in ReadNeedleBodyBytes computed the tail length as
checksum, plus timestamp only for Version3. Forks and future on-disk
formats whose tail carries more fields would silently under-check and
still panic in readNeedleTail on a truncated body. Derive the tail from
NeedleBodyLength minus data and padding so the bound stays exact for
every version.

Iterate IsSupportedVersion in the new tests instead of hardcoding
v1-v3 so downstream formats get covered automatically, and skip
versions the build cannot write rather than failing on them.

* test: skip needle write only on the unsupported-version error

A blanket skip would hide a real writer regression. Skip the version
subtest only when the writer reports the version is not supported in
this build (the error text differs between builds), and fail on any
other write error.

weed/storage/needle/needle_read.go

@@ -225,12 +225,11 @@ func (n *Needle) ReadNeedleBody(r backend.BackendStorageFile, version Version, o
 func (n *Needle) ReadNeedleBodyBytes(needleBody []byte, version Version) (err error) {
 
 	// 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)) {
+	// negative size or one the body cannot hold along with its tail. Deriving
+	// the tail from the version's own layout keeps the bound exact for every
+	// on-disk format.
+	tailSize := NeedleBodyLength(n.Size, version) - int64(n.Size) - int64(PaddingLength(n.Size, version))
+	if n.Size < 0 || int64(n.Size)+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)
 	}

weed/storage/needle/needle_read_test.go

@@ -3,6 +3,7 @@ package needle
 import (
 	"bytes"
 	"errors"
+	"strings"
 	"testing"
 
 	. "github.com/seaweedfs/seaweedfs/weed/storage/types"
@@ -23,7 +24,7 @@ func readNeedleBodyBytes(t *testing.T, n *Needle, body []byte, version 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} {
+	for version := Version1; IsSupportedVersion(version); version++ {
 		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.
@@ -60,12 +61,18 @@ func TestReadNeedleBodyBytesRejectsCorruptSize(t *testing.T) {
 // 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} {
+	for version := Version1; IsSupportedVersion(version); version++ {
 		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 {
+					// Some builds can read a version they cannot write; skip
+					// only that recognized case so a real writer regression
+					// still fails the test.
+					if strings.Contains(strings.ToLower(err.Error()), "unsupported version") {
+						t.Skipf("version %d is not writable in this build: %v", version, err)
+					}
 					t.Fatalf("write needle: %v", err)
 				}