mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-07-19 06:22:31 +00:00
Lisandro Pin
39e9294907
Fixes scrubbing for pre-allocated volumes with zero-size indeces by reworking the validation code to allow zero-size indeces on zero-size volumes.
256 lines
8.6 KiB
Go
256 lines
8.6 KiB
Go
package storage
|
|
|
|
import (
|
|
"fmt"
|
|
"os"
|
|
"reflect"
|
|
"testing"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/backend"
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/types"
|
|
)
|
|
|
|
func TestScrubVolumeData(t *testing.T) {
|
|
testCases := []struct {
|
|
name string
|
|
dataPath string
|
|
indexPath string
|
|
version needle.Version
|
|
want int64
|
|
wantErrs []error
|
|
}{
|
|
{
|
|
name: "healthy volume",
|
|
dataPath: "./test_files/healthy_volume.dat",
|
|
indexPath: "./test_files/healthy_volume.idx",
|
|
version: needle.Version3,
|
|
want: 27,
|
|
wantErrs: []error{},
|
|
},
|
|
{
|
|
name: "zero-size volume without index",
|
|
dataPath: "./test_files/empty_volume.dat",
|
|
indexPath: "./test_files/empty_volume.idx",
|
|
version: needle.Version3,
|
|
want: 0,
|
|
wantErrs: []error{},
|
|
},
|
|
{
|
|
name: "zero-size volume with index",
|
|
dataPath: "./test_files/empty_volume.dat",
|
|
indexPath: "./test_files/healthy_volume.idx",
|
|
version: needle.Version3,
|
|
want: 27,
|
|
wantErrs: []error{
|
|
fmt.Errorf("failed to read needle 3 on volume 0: EOF"),
|
|
fmt.Errorf("failed to read needle 4 on volume 0: EOF"),
|
|
fmt.Errorf("failed to read needle 8 on volume 0: EOF"),
|
|
fmt.Errorf("failed to read needle 9 on volume 0: EOF"),
|
|
fmt.Errorf("failed to read needle 11 on volume 0: EOF"),
|
|
fmt.Errorf("failed to read needle 18 on volume 0: EOF"),
|
|
fmt.Errorf("failed to read needle 20 on volume 0: EOF"),
|
|
fmt.Errorf("failed to read needle 21 on volume 0: EOF"),
|
|
fmt.Errorf("failed to read needle 25 on volume 0: EOF"),
|
|
fmt.Errorf("failed to read needle 29 on volume 0: EOF"),
|
|
fmt.Errorf("failed to read needle 31 on volume 0: EOF"),
|
|
fmt.Errorf("failed to read needle 33 on volume 0: EOF"),
|
|
fmt.Errorf("failed to read needle 35 on volume 0: EOF"),
|
|
fmt.Errorf("failed to read needle 39 on volume 0: EOF"),
|
|
fmt.Errorf("failed to read needle 45 on volume 0: EOF"),
|
|
fmt.Errorf("failed to read needle 46 on volume 0: EOF"),
|
|
fmt.Errorf("failed to read needle 48 on volume 0: EOF"),
|
|
fmt.Errorf("data file for volume 0 is smaller (8) than the 27 needles it contains (942856)"),
|
|
},
|
|
},
|
|
{
|
|
name: "volume without index",
|
|
dataPath: "./test_files/healthy_volume.dat",
|
|
indexPath: "./test_files/empty_volume.idx",
|
|
version: needle.Version3,
|
|
want: 0,
|
|
wantErrs: []error{
|
|
fmt.Errorf("data file size for volume 0 (942856) doesn't match the size for 0 needles read (8)"),
|
|
},
|
|
},
|
|
{
|
|
name: "bitrot volume",
|
|
dataPath: "./test_files/bitrot_volume.dat",
|
|
indexPath: "./test_files/bitrot_volume.idx",
|
|
version: needle.Version3,
|
|
want: 27,
|
|
wantErrs: []error{
|
|
fmt.Errorf("failed to read needle 3 on volume 0: invalid CRC for needle 3 (got 0b243a0d, want 4af853fb), data on disk corrupted: needle data corrupted"),
|
|
fmt.Errorf("failed to read needle 48 on volume 0: invalid CRC for needle 30 (got 3c40e8d5, want 5077fea1), data on disk corrupted: needle data corrupted"),
|
|
fmt.Errorf("data file size for volume 0 (942864) doesn't match the size for 27 needles read (942856)"),
|
|
},
|
|
},
|
|
}
|
|
|
|
for _, tc := range testCases {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
datFile, err := os.OpenFile(tc.dataPath, os.O_RDONLY, 0)
|
|
if err != nil {
|
|
t.Fatalf("failed to open data file: %v", err)
|
|
}
|
|
defer datFile.Close()
|
|
|
|
idxFile, err := os.OpenFile(tc.indexPath, os.O_RDONLY, 0)
|
|
if err != nil {
|
|
t.Fatalf("failed to open index file: %v", err)
|
|
}
|
|
defer idxFile.Close()
|
|
|
|
idxStat, err := idxFile.Stat()
|
|
if err != nil {
|
|
t.Fatalf("failed to stat index file: %v", err)
|
|
}
|
|
|
|
v := Volume{
|
|
DataBackend: backend.NewDiskFile(datFile),
|
|
volumeInfo: &volume_server_pb.VolumeInfo{
|
|
Version: uint32(tc.version),
|
|
},
|
|
}
|
|
|
|
got, gotErrs := v.scrubVolumeData(idxFile, idxStat.Size())
|
|
|
|
if got != tc.want {
|
|
t.Errorf("expected %d files processed, got %d", tc.want, got)
|
|
}
|
|
if !reflect.DeepEqual(gotErrs, tc.wantErrs) {
|
|
t.Errorf("expected errors %v, got %v", tc.wantErrs, gotErrs)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
// TestCheckVolumeDataIntegrityWithDeletionTombstone guards the size argument
|
|
// passed to ReadData when verifying a trailing deletion tombstone. The .idx
|
|
// entry carries TombstoneFileSize (-1) but the appended tombstone needle in
|
|
// .dat carries Size=0. Forwarding the .idx size into ReadData fails the size
|
|
// check, activates the 32GB 4-byte-offset wrap-around retry, and surfaces EOF;
|
|
// CheckVolumeDataIntegrity then treats the volume as corrupt and the loader
|
|
// marks every such volume read-only on startup.
|
|
func TestCheckVolumeDataIntegrityWithDeletionTombstone(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()
|
|
|
|
for i := 1; i <= 3; i++ {
|
|
n := newRandomNeedle(uint64(i))
|
|
if _, _, _, err := v.writeNeedle2(n, true, false); err != nil {
|
|
t.Fatalf("write needle %d: %v", i, err)
|
|
}
|
|
}
|
|
// Delete a live needle so a tombstone (Size=0 on disk, -1 in idx) is the
|
|
// last entry the integrity check examines.
|
|
if _, err := v.doDeleteRequest(newEmptyNeedle(2)); err != nil {
|
|
t.Fatalf("delete needle: %v", err)
|
|
}
|
|
if err := v.DataBackend.Sync(); err != nil {
|
|
t.Fatalf("sync .dat: %v", err)
|
|
}
|
|
if err := v.nm.Sync(); err != nil {
|
|
t.Fatalf("sync .idx: %v", err)
|
|
}
|
|
|
|
idxFile, err := os.OpenFile(v.FileName(".idx"), os.O_RDONLY, 0644)
|
|
if err != nil {
|
|
t.Fatalf("open .idx: %v", err)
|
|
}
|
|
defer idxFile.Close()
|
|
|
|
lastAppendAtNs, err := CheckVolumeDataIntegrity(v, idxFile)
|
|
if err != nil {
|
|
t.Fatalf("CheckVolumeDataIntegrity should succeed with a trailing deletion tombstone: %v", err)
|
|
}
|
|
// V3 tombstones still record AppendAtNs; a zero return means ReadData
|
|
// bailed out before populating the needle and would mask future regressions
|
|
// that quietly skip the tombstone body.
|
|
if lastAppendAtNs == 0 {
|
|
t.Errorf("expected non-zero lastAppendAtNs for V3 deletion tombstone, got 0")
|
|
}
|
|
}
|
|
|
|
// TestMaxNeedleEnd ensures the needle map's MaxNeedleEnd accumulator lets
|
|
// volume.load() detect an .idx that references bytes past the end of the .dat
|
|
// — the deeper-than-tail corruption shape from issue #8928 that the existing
|
|
// last-10-entries scan cannot see. The check is populated by the load walk
|
|
// and read by volume.load() to flip the volume read-only.
|
|
func TestMaxNeedleEnd(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()
|
|
|
|
// A handful of healthy needles establishes a baseline .dat/.idx.
|
|
for i := 1; i <= 4; i++ {
|
|
n := newRandomNeedle(uint64(i))
|
|
if _, _, _, err := v.writeNeedle2(n, true, false); err != nil {
|
|
t.Fatalf("write needle %d: %v", i, err)
|
|
}
|
|
}
|
|
if err := v.DataBackend.Sync(); err != nil {
|
|
t.Fatalf("sync .dat: %v", err)
|
|
}
|
|
if err := v.nm.Sync(); err != nil {
|
|
t.Fatalf("sync .idx: %v", err)
|
|
}
|
|
|
|
datSize, _, err := v.DataBackend.GetStat()
|
|
if err != nil {
|
|
t.Fatalf("stat .dat: %v", err)
|
|
}
|
|
|
|
// Sanity: a fresh load over the healthy .idx puts MaxNeedleEnd inside
|
|
// the .dat.
|
|
idxFile, err := os.OpenFile(v.FileName(".idx"), os.O_RDONLY, 0644)
|
|
if err != nil {
|
|
t.Fatalf("open .idx: %v", err)
|
|
}
|
|
healthyNm, err := LoadCompactNeedleMap(idxFile, v.Version())
|
|
idxFile.Close()
|
|
if err != nil {
|
|
t.Fatalf("load healthy .idx: %v", err)
|
|
}
|
|
healthyEnd := healthyNm.MaxNeedleEnd()
|
|
if healthyEnd <= 0 || healthyEnd > datSize {
|
|
t.Fatalf("healthy volume should have MaxNeedleEnd (%d) in (0, dat_size=%d]", healthyEnd, datSize)
|
|
}
|
|
|
|
// Inject a dangling entry by appending to the .idx, then reload. The
|
|
// walk should observe MaxNeedleEnd past dat_size — exactly the signal
|
|
// volume.load uses to mark the volume read-only.
|
|
bogusOffset := types.ToOffset(datSize + 4*1024*1024)
|
|
if err := v.nm.Put(types.Uint64ToNeedleId(9999), bogusOffset, types.Size(1024)); err != nil {
|
|
t.Fatalf("inject dangling idx entry: %v", err)
|
|
}
|
|
if err := v.nm.Sync(); err != nil {
|
|
t.Fatalf("sync .idx after inject: %v", err)
|
|
}
|
|
|
|
idxFile, err = os.OpenFile(v.FileName(".idx"), os.O_RDONLY, 0644)
|
|
if err != nil {
|
|
t.Fatalf("reopen .idx: %v", err)
|
|
}
|
|
defer idxFile.Close()
|
|
badNm, err := LoadCompactNeedleMap(idxFile, v.Version())
|
|
if err != nil {
|
|
t.Fatalf("reload .idx after inject: %v", err)
|
|
}
|
|
badEnd := badNm.MaxNeedleEnd()
|
|
if badEnd <= datSize {
|
|
t.Fatalf("after dangling-entry inject MaxNeedleEnd (%d) should exceed dat_size (%d)", badEnd, datSize)
|
|
}
|
|
}
|