Fix volume.merge corrupting every needle it copies (#10565)

* Give volume.merge the needle size the target actually indexes by

needleBlobFromNeedle returned the size Append reports, which is
Size(n.DataSize) - payload bytes only. The .dat header, the needle map and
WriteNeedleBlobRequest.Size all use n.Size, which additionally covers the
flags, name, mime and lastModified fields.

Every needle volume.merge copied therefore landed with a too-small size. The
target indexed it at that length, so every later read failed the header check
in ReadBytes with a size mismatch, and on v3 the fresh AppendAtNs stamp landed
NeedleHeaderSize+DataSize+NeedleChecksumSize into the blob - exactly on the
flags byte - overwriting flags, name size, mime size and the first mime bytes
with the top of a timestamp. Needles came back with flags 0x18, no name, no
mime and a phantom TTL parsed from two arbitrary timestamp bytes; the ones
that decoded as expired 404 and vacuum would drop them. Since merge rebuilds
every replica from the merged copy, no clean replica survives.

Return n.Size, which Append fills in as it serializes, matching what the
normal write path stores via nm.Put.

* Reject needle blobs whose size disagrees with their own header

WriteNeedleBlob trusts the caller's size for two destructive things: it is
what goes into the needle map, and it is where the v3 AppendAtNs stamp is
written inside the caller's buffer. A caller passing the payload-only DataSize
convention corrupts both, and nothing surfaces until the needle is read back -
by which point every replica may already have been rebuilt from it.

Parse the blob's own header and refuse the write when the two disagree.
Mirrored in the Rust volume server.

seaweed-volume/src/storage/volume.rs

@@ -2739,6 +2739,29 @@ impl Volume {
             return Err(VolumeError::ReadOnly);
         }
 
+        // size indexes the needle and places the v3 append timestamp, so a caller using
+        // the payload-only DataSize corrupts both, silently until the needle is read back.
+        if needle_blob.len() < NEEDLE_HEADER_SIZE {
+            return Err(VolumeError::Io(io::Error::new(
+                io::ErrorKind::InvalidData,
+                format!(
+                    "needle {} blob of {} bytes is shorter than a needle header",
+                    needle_id.0,
+                    needle_blob.len()
+                ),
+            )));
+        }
+        let (_, _, header_size) = Needle::parse_header(needle_blob);
+        if header_size != size {
+            return Err(VolumeError::Io(io::Error::new(
+                io::ErrorKind::InvalidData,
+                format!(
+                    "needle {} size {} does not match its blob header size {}",
+                    needle_id.0, size.0, header_size.0
+                ),
+            )));
+        }
+
         // Dedup check: if the same needle already exists with matching content, skip the write.
         // Matches Go's WriteNeedleBlob which reads existing needle and compares cookie+checksum+data.
         if let Some(nm) = &self.nm {
@@ -4445,6 +4468,39 @@ mod tests {
         assert!(matches!(err, VolumeError::CookieMismatch(_)));
     }
 
+    // A size disagreeing with the blob's own header indexes the needle at the wrong
+    // length and, on v3, stamps the append timestamp into the middle of the needle.
+    #[test]
+    fn test_write_needle_blob_rejects_size_mismatch() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let mut v = make_test_volume(dir);
+
+        let mut n = Needle {
+            id: NeedleId(1),
+            cookie: Cookie(0x12345678),
+            data: b"the merged payload".to_vec(),
+            data_size: 18,
+            ..Needle::default()
+        };
+        n.checksum = CRC::new(&n.data);
+        let (offset, _, _) = v.write_needle(&mut n, true).unwrap();
+        let blob = v.read_needle_blob(offset as i64, n.size).unwrap();
+
+        let dat_size_before = v.dat_file_size().unwrap();
+
+        // Size(n.data_size) is what append reports, and what a caller following the
+        // payload-size convention would send.
+        let err = v
+            .write_needle_blob_and_index(NeedleId(2), &blob, Size(n.data_size as i32))
+            .unwrap_err();
+        assert!(matches!(err, VolumeError::Io(_)), "got {err:?}");
+        assert_eq!(v.dat_file_size().unwrap(), dat_size_before);
+
+        v.write_needle_blob_and_index(NeedleId(2), &blob, n.size)
+            .unwrap();
+    }
+
     #[test]
     fn test_volume_destroy() {
         let tmp = TempDir::new().unwrap();

weed/shell/command_volume_merge.go

@@ -423,7 +423,9 @@ func needleBlobFromNeedle(n *needle.Needle, version needle.Version) ([]byte, typ
 	memFile := newMemoryBackendFile()
 	defer memFile.Close()
 
-	_, size, actualSize, err := n.Append(memFile, version)
+	// Append reports Size(n.DataSize); the .dat header and the needle map both use
+	// n.Size, which Append fills in as it serializes.
+	_, _, actualSize, err := n.Append(memFile, version)
 	if err != nil {
 		return nil, 0, err
 	}
@@ -433,7 +435,7 @@ func needleBlobFromNeedle(n *needle.Needle, version needle.Version) ([]byte, typ
 	if err != nil && err != io.EOF {
 		return nil, 0, err
 	}
-	return buf[:read], size, nil
+	return buf[:read], n.Size, nil
 }
 
 func allocateMergeVolumeOnThirdLocation(grpcDialOption grpc.DialOption, allLocations []location, replicas []*VolumeReplica, info *master_pb.VolumeInformationMessage, replicaPlacement *super_block.ReplicaPlacement) (pb.ServerAddress, error) {

weed/shell/command_volume_merge_test.go

@@ -1,11 +1,14 @@
 package shell
 
 import (
+	"bytes"
 	"reflect"
 	"testing"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
 )
 
 type sliceNeedleStream struct {
@@ -455,6 +458,80 @@ func TestMergeWorkflowValidation(t *testing.T) {
 	}
 }
 
+// The size returned alongside the blob travels in WriteNeedleBlobRequest.Size, and
+// the target stores it in .idx and uses it to place the v3 append timestamp. Feed a
+// needle through the whole path and read it back the way a GET does.
+func TestNeedleBlobFromNeedleRoundTrip(t *testing.T) {
+	version := needle.GetCurrentVersion()
+	modified := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
+
+	source := &needle.Needle{
+		Id:           types.Uint64ToNeedleId(42),
+		Cookie:       types.Cookie(0x12345678),
+		Data:         []byte("the merged payload"),
+		Name:         []byte("report.txt"),
+		Mime:         []byte("text/plain"),
+		LastModified: uint64(modified.Unix()),
+		AppendAtNs:   uint64(modified.UnixNano()),
+	}
+	source.Checksum = needle.NewCRC(source.Data)
+	source.SetHasName()
+	source.SetHasMime()
+	source.SetHasLastModifiedDate()
+
+	blob, size, err := needleBlobFromNeedle(source, version)
+	if err != nil {
+		t.Fatalf("serialize needle: %v", err)
+	}
+
+	var header needle.Needle
+	header.ParseNeedleHeader(blob)
+	if size != header.Size {
+		t.Fatalf("needleBlobFromNeedle returned size %d, but the blob's own header says %d", size, header.Size)
+	}
+
+	// Replay what the target volume server does with the blob and the size.
+	dat := newMemoryBackendFile()
+	defer dat.Close()
+	appendAtNs := uint64(time.Date(2026, 6, 7, 8, 9, 10, 0, time.UTC).UnixNano())
+	offset, err := needle.WriteNeedleBlob(dat, blob, size, appendAtNs, version)
+	if err != nil {
+		t.Fatalf("write needle blob: %v", err)
+	}
+
+	// ... and what a GET does with the size the target recorded in .idx.
+	stored, err := needle.ReadNeedleBlob(dat, int64(offset), size, version)
+	if err != nil {
+		t.Fatalf("read needle blob back: %v", err)
+	}
+	var got needle.Needle
+	if err = got.ReadBytes(stored, int64(offset), size, version); err != nil {
+		t.Fatalf("read merged needle: %v", err)
+	}
+
+	if !bytes.Equal(got.Data, source.Data) {
+		t.Errorf("data: got %q, want %q", got.Data, source.Data)
+	}
+	if got.Flags != source.Flags {
+		t.Errorf("flags: got %#x, want %#x", got.Flags, source.Flags)
+	}
+	if !bytes.Equal(got.Name, source.Name) {
+		t.Errorf("name: got %q, want %q", got.Name, source.Name)
+	}
+	if !bytes.Equal(got.Mime, source.Mime) {
+		t.Errorf("mime: got %q, want %q", got.Mime, source.Mime)
+	}
+	if got.LastModified != source.LastModified {
+		t.Errorf("lastModified: got %d, want %d", got.LastModified, source.LastModified)
+	}
+	if got.HasTtl() {
+		t.Errorf("merged needle grew a phantom ttl %v", got.Ttl)
+	}
+	if got.AppendAtNs != appendAtNs {
+		t.Errorf("appendAtNs: got %d, want %d", got.AppendAtNs, appendAtNs)
+	}
+}
+
 // TestMergeEdgeCaseHandling validates that the merge handles known edge cases
 func TestMergeEdgeCaseHandling(t *testing.T) {
 	edgeCases := map[string]bool{

weed/storage/volume_write.go

@@ -378,6 +378,17 @@ func (v *Volume) WriteNeedleBlob(needleId NeedleId, needleBlob []byte, size Size
 		return fmt.Errorf("volume %d is read only", v.Id)
 	}
 
+	// size indexes the needle and places the v3 append timestamp, so a caller using
+	// the payload-only DataSize corrupts both, silently until the needle is read back.
+	if len(needleBlob) < NeedleHeaderSize {
+		return fmt.Errorf("needle %d blob of %d bytes is shorter than a needle header", needleId, len(needleBlob))
+	}
+	var blobHeader needle.Needle
+	blobHeader.ParseNeedleHeader(needleBlob)
+	if blobHeader.Size != size {
+		return fmt.Errorf("needle %d size %d does not match its blob header size %d", needleId, size, blobHeader.Size)
+	}
+
 	if MaxPossibleVolumeSize < v.nm.ContentSize()+uint64(len(needleBlob)) {
 		return fmt.Errorf("volume size limit %d exceeded! current size is %d", MaxPossibleVolumeSize, v.nm.ContentSize())
 	}

weed/storage/volume_write_test.go

@@ -9,9 +9,11 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
+	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"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"
 )
 
 func TestSearchVolumesWithDeletedNeedles(t *testing.T) {
@@ -176,11 +178,12 @@ func TestWriteNeedleBlobRejectedOnReadOnlyVolume(t *testing.T) {
 	if err != nil {
 		t.Fatalf("volume creation: %v", err)
 	}
-	offset, size, _, err := v.writeNeedle2(newRandomNeedle(1), true, false)
+	n := newRandomNeedle(1)
+	offset, _, _, err := v.writeNeedle2(n, true, false)
 	if err != nil {
 		t.Fatalf("write needle: %v", err)
 	}
-	blob, err := v.ReadNeedleBlob(int64(offset), size)
+	blob, err := v.ReadNeedleBlob(int64(offset), n.Size)
 	if err != nil {
 		t.Fatalf("read needle blob: %v", err)
 	}
@@ -198,7 +201,7 @@ func TestWriteNeedleBlobRejectedOnReadOnlyVolume(t *testing.T) {
 
 	datSizeBefore, _, _ := v.DataBackend.GetStat()
 
-	err = v.WriteNeedleBlob(types.Uint64ToNeedleId(2), blob, size)
+	err = v.WriteNeedleBlob(types.Uint64ToNeedleId(2), blob, n.Size)
 	if err == nil {
 		t.Fatalf("expected WriteNeedleBlob to be rejected on a read-only volume")
 	}
@@ -211,3 +214,45 @@ func TestWriteNeedleBlobRejectedOnReadOnlyVolume(t *testing.T) {
 		t.Errorf("read-only volume .dat grew from %d to %d, leaving an unindexed needle", datSizeBefore, datSizeAfter)
 	}
 }
+
+// A size disagreeing with the blob's own header indexes the needle at the wrong
+// length and, on v3, stamps the append timestamp into the middle of the needle.
+func TestWriteNeedleBlobRejectsSizeMismatch(t *testing.T) {
+	dir := t.TempDir()
+	location := NewDiskLocation(dir, 10, util.MinFreeSpace{}, dir, "", nil, stats.DefaultDiskIOProbeConfig())
+	defer location.Close()
+
+	v, err := NewVolume(dir, dir, "", 7, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+	if err != nil {
+		t.Fatalf("volume creation: %v", err)
+	}
+	defer v.Close()
+	location.SetVolume(7, v)
+
+	n := newRandomNeedle(1)
+	offset, _, _, err := v.writeNeedle2(n, true, false)
+	if err != nil {
+		t.Fatalf("write needle: %v", err)
+	}
+	blob, err := v.ReadNeedleBlob(int64(offset), n.Size)
+	if err != nil {
+		t.Fatalf("read needle blob: %v", err)
+	}
+
+	datSizeBefore, _, _ := v.DataBackend.GetStat()
+
+	// types.Size(n.DataSize) is what needle.Append reports, and what a caller
+	// following the payload-size convention would send.
+	if err = v.WriteNeedleBlob(types.Uint64ToNeedleId(2), blob, types.Size(n.DataSize)); err == nil {
+		t.Fatal("expected WriteNeedleBlob to reject a size that disagrees with the blob header")
+	}
+
+	datSizeAfter, _, _ := v.DataBackend.GetStat()
+	if datSizeAfter != datSizeBefore {
+		t.Errorf(".dat grew from %d to %d on a rejected blob", datSizeBefore, datSizeAfter)
+	}
+
+	if err = v.WriteNeedleBlob(types.Uint64ToNeedleId(2), blob, n.Size); err != nil {
+		t.Fatalf("write needle blob with the header size: %v", err)
+	}
+}