volume: fix 5-byte index offset corruption in makeupDiff (#11411)

* volume: encode all offset bytes when makeupDiff replays a write

makeupDiff patched only bytes 8:12 of the index entry, so under the
5BytesOffset build the fifth byte kept the old offset's high bits and the
replayed needle's index pointed 32 GiB-aligned ranges away from its body.
A later vacuum then dropped the entry as unreadable. Rebuild the entry with
needle_map.ToBytes, the same encoder the tombstone branch just below uses.

* volume: test makeupDiff replay across a 32 GiB offset boundary

Sparse-file test: truncate the .dat to 64 GiB after one write, compact,
write a second needle, commit, and assert the index offset matches the
.compacted size and the needle stays readable through a second vacuum.
Only runs under -tags=5BytesOffset.

weed/storage/volume_vacuum.go

@@ -541,7 +541,7 @@ func (v *Volume) makeupDiff(newDatFileName, newIdxFileName, oldDatFileName, oldI
 				return fmt.Errorf("ReadNeedleBlob %s key %d offset %d size %d failed: %w", oldDatFile.Name(), key, increIdxEntry.offset.ToActualOffset(), increIdxEntry.size, err)
 			}
 			dstDatBackend.Write(needleBytes)
-			util.Uint32toBytes(idxEntryBytes[8:12], uint32(offset/NeedlePaddingSize))
+			idxEntryBytes = needle_map.ToBytes(key, ToOffset(offset), increIdxEntry.size)
 		} else { //deleted needle
 			//fakeDelNeedle's default Data field is nil
 			fakeDelNeedle := new(needle.Needle)

weed/storage/volume_vacuum_5byte_offset_test.go

@@ -0,0 +1,90 @@
+//go:build 5BytesOffset
+
+package storage
+
+import (
+	"bytes"
+	"os"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/storage/idx"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+)
+
+// A write replayed by makeupDiff must record all offset bytes, including when
+// the new offset lands in a different 32 GiB range than the old one.
+func TestConcurrentWriteCrossesOffsetBoundary(t *testing.T) {
+	dir := t.TempDir()
+	v, err := NewVolume(dir, dir, "", 784, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.Version3, 0, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer v.Close()
+	write := func(key uint64, size int) *needle.Needle {
+		n := &needle.Needle{Id: types.Uint64ToNeedleId(key), Data: bytes.Repeat([]byte("x"), size)}
+		n.Checksum = needle.NewCRC(n.Data)
+		if _, _, _, err := v.writeNeedle2(n, true, false, false); err != nil {
+			t.Fatal(err)
+		}
+		return n
+	}
+	write(1, 32)
+	if err := v.DataBackend.Truncate(64 << 30); err != nil {
+		t.Fatal(err)
+	}
+	if err := v.CompactByIndex(nil); err != nil {
+		t.Fatal(err)
+	}
+	written := write(342511246, 113)
+	before := newEmptyNeedle(uint64(written.Id))
+	if _, err := v.readNeedle(before, nil, nil); err != nil || !bytes.Equal(before.Data, written.Data) {
+		t.Fatalf("read before commit: %v", err)
+	}
+	cpd, err := os.Stat(v.FileName(".cpd"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	expectedOffset := cpd.Size()
+	if err := v.CommitCompact(); err != nil {
+		t.Fatalf("commit: %v", err)
+	}
+	indexFile, err := os.Open(v.FileName(".idx"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer indexFile.Close()
+	var actualOffset int64
+	if err := idx.WalkIndexFile(indexFile, 0, func(key types.NeedleId, offset types.Offset, _ types.Size) error {
+		if key == written.Id {
+			actualOffset = offset.ToActualOffset()
+		}
+		return nil
+	}); err != nil {
+		t.Fatal(err)
+	}
+	after := newEmptyNeedle(uint64(written.Id))
+	_, readErr := v.readNeedle(after, nil, nil)
+	t.Logf("stored offset=%d, correct offset=%d, excess=%d, post-commit read=%v", actualOffset, expectedOffset, actualOffset-expectedOffset, readErr)
+	if actualOffset != expectedOffset || readErr != nil || !bytes.Equal(after.Data, written.Data) {
+		physical := new(needle.Needle)
+		if err := physical.ReadData(v.DataBackend, expectedOffset, written.Size, v.Version()); err == nil && bytes.Equal(physical.Data, written.Data) {
+			t.Logf("body intact at offset %d despite the bad index entry", expectedOffset)
+		}
+		t.Fatalf("replayed write lost: stored offset=%d, want=%d, read=%v", actualOffset, expectedOffset, readErr)
+	}
+
+	if err := v.CompactByIndex(nil); err != nil {
+		t.Fatalf("second compact: %v", err)
+	}
+	if err := v.CommitCompact(); err != nil {
+		t.Fatalf("second commit: %v", err)
+	}
+	if _, err := v.readNeedle(newEmptyNeedle(uint64(written.Id)), nil, nil); err != nil {
+		t.Fatalf("needle lost after second vacuum: %v", err)
+	}
+	if _, present := v.nm.Get(written.Id); !present {
+		t.Fatal("needle missing from index after second vacuum")
+	}
+}