From 312cfe5ae17facbb765b87870e05168452a24ad6 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 4 Aug 2026 16:58:25 -0700 Subject: [PATCH] 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 | 56 ++++++++++++++++++ weed/shell/command_volume_merge.go | 6 +- weed/shell/command_volume_merge_test.go | 77 +++++++++++++++++++++++++ weed/storage/volume_write.go | 11 ++++ weed/storage/volume_write_test.go | 51 +++++++++++++++- 5 files changed, 196 insertions(+), 5 deletions(-) diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index 6c3b0cbe2..3c00f1866 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/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(); diff --git a/weed/shell/command_volume_merge.go b/weed/shell/command_volume_merge.go index c115af20d..5605ae632 100644 --- a/weed/shell/command_volume_merge.go +++ b/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) { diff --git a/weed/shell/command_volume_merge_test.go b/weed/shell/command_volume_merge_test.go index d1caac273..4e4a1ee7e 100644 --- a/weed/shell/command_volume_merge_test.go +++ b/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{ diff --git a/weed/storage/volume_write.go b/weed/storage/volume_write.go index 3a8ae5f92..070d9bd81 100644 --- a/weed/storage/volume_write.go +++ b/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()) } diff --git a/weed/storage/volume_write_test.go b/weed/storage/volume_write_test.go index dc5f5fdb5..b060d6728 100644 --- a/weed/storage/volume_write_test.go +++ b/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) + } +}