seaweedfs/weed/storage/volume_idx_rebuild_test.go

package storage

import (
	"bytes"
	"os"
	"path/filepath"
	"testing"

	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
	"github.com/seaweedfs/seaweedfs/weed/storage/types"
)

// Pointing -dir.idx at a directory with no .idx used to abort the whole volume
// server in checkIdxFile; the index is derivable from the .dat, so it must be
// rebuilt in place instead.
func TestLoad_MovedIdxDirectory_RebuildsIdx(t *testing.T) {
	root := t.TempDir()
	dataDir := filepath.Join(root, "data")
	oldIdxDir := filepath.Join(root, "idxA")
	newIdxDir := filepath.Join(root, "idxB")
	for _, dir := range []string{dataDir, oldIdxDir, newIdxDir} {
		if err := os.MkdirAll(dir, 0755); err != nil {
			t.Fatalf("mkdir %s: %v", dir, err)
		}
	}

	v, err := NewVolume(dataDir, oldIdxDir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
	if err != nil {
		t.Fatalf("create volume: %v", err)
	}
	for id := uint64(1); id <= 3; id++ {
		if _, _, _, err := v.writeNeedle2(newRandomNeedle(id), true, false, false); err != nil {
			t.Fatalf("seed write %d: %v", id, err)
		}
	}
	if _, err := v.deleteNeedle2(newRandomNeedle(2)); err != nil {
		t.Fatalf("seed delete: %v", err)
	}
	wantCount, wantDeleted := v.nm.FileCount(), v.nm.DeletedCount()
	v.Close()

	if _, err := os.Stat(filepath.Join(oldIdxDir, "1.idx")); err != nil {
		t.Fatalf("seeded idx should live in the old idx dir: %v", err)
	}

	v2, err := NewVolume(dataDir, newIdxDir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
	if err != nil {
		t.Fatalf("reload against the new idx dir: %v", err)
	}
	defer v2.Close()

	if _, err := os.Stat(filepath.Join(newIdxDir, "1.idx")); err != nil {
		t.Fatalf("idx not rebuilt in the new idx dir: %v", err)
	}
	if got := v2.nm.FileCount(); got != wantCount {
		t.Errorf("file count = %d, want %d", got, wantCount)
	}
	if got := v2.nm.DeletedCount(); got != wantDeleted {
		t.Errorf("deleted count = %d, want %d", got, wantDeleted)
	}
	old, err := os.ReadFile(filepath.Join(oldIdxDir, "1.idx"))
	if err != nil {
		t.Fatalf("read the seeded idx: %v", err)
	}
	rebuilt, err := os.ReadFile(filepath.Join(newIdxDir, "1.idx"))
	if err != nil {
		t.Fatalf("read the rebuilt idx: %v", err)
	}
	if !bytes.Equal(old, rebuilt) {
		t.Errorf("rebuilt idx (%d bytes) differs from the one the server wrote (%d bytes)", len(rebuilt), len(old))
	}
	if v2.noWriteOrDelete {
		t.Errorf("volume marked read-only after the rebuild")
	}
}

// A .dat padded with zeros must not be indexed as needle 0 rows: the walk stops
// where the records do.
func TestRebuildIdx_StopsAtZeroPaddedDatTail(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("create volume: %v", err)
	}
	if _, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, false, false); err != nil {
		t.Fatalf("seed write: %v", err)
	}
	v.Close()

	base := VolumeFileName(dir, "", 1)
	seeded, err := os.ReadFile(base + ".idx")
	if err != nil {
		t.Fatalf("read the seeded idx: %v", err)
	}
	datSize, err := os.Stat(base + ".dat")
	if err != nil {
		t.Fatalf("stat dat: %v", err)
	}
	if err := os.Truncate(base+".dat", datSize.Size()+4096); err != nil {
		t.Fatalf("pad dat: %v", err)
	}
	if err := os.Remove(base + ".idx"); err != nil {
		t.Fatalf("drop idx: %v", err)
	}

	v2, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
	if err != nil {
		t.Fatalf("reload: %v", err)
	}
	defer v2.Close()

	rebuilt, err := os.ReadFile(base + ".idx")
	if err != nil {
		t.Fatalf("read the rebuilt idx: %v", err)
	}
	if !bytes.Equal(seeded, rebuilt) {
		t.Errorf("rebuilt idx has %d bytes, want the %d the server wrote", len(rebuilt), len(seeded))
	}
}

// A .dat whose last append was torn mid-body must not gain an index row that
// points past the end of the file.
func TestRebuildIdx_SkipsTruncatedDatTail(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("create volume: %v", err)
	}
	if _, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, false, false); err != nil {
		t.Fatalf("seed write: %v", err)
	}
	base := VolumeFileName(dir, "", 1)
	kept, err := os.ReadFile(base + ".idx")
	if err != nil {
		t.Fatalf("read the seeded idx: %v", err)
	}
	if _, _, _, err := v.writeNeedle2(newRandomNeedle(2), true, false, false); err != nil {
		t.Fatalf("seed torn write: %v", err)
	}
	v.Close()

	// Chop the second needle's body, leaving its header intact.
	datSize, err := os.Stat(base + ".dat")
	if err != nil {
		t.Fatalf("stat dat: %v", err)
	}
	if err := os.Truncate(base+".dat", datSize.Size()-8); err != nil {
		t.Fatalf("tear dat: %v", err)
	}
	if err := os.Remove(base + ".idx"); err != nil {
		t.Fatalf("drop idx: %v", err)
	}

	v2, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
	if err != nil {
		t.Fatalf("reload: %v", err)
	}
	defer v2.Close()

	rebuilt, err := os.ReadFile(base + ".idx")
	if err != nil {
		t.Fatalf("read the rebuilt idx: %v", err)
	}
	if !bytes.Equal(kept, rebuilt) {
		t.Errorf("rebuilt idx has %d bytes, want the %d covering only the intact needle", len(rebuilt), len(kept))
	}
	if maxEnd := v2.nm.MaxNeedleEnd(); maxEnd > datSize.Size()-8 {
		t.Errorf("rebuilt idx reaches %d, past the %d-byte .dat", maxEnd, datSize.Size()-8)
	}
}

// A corrupt header carrying a negative size advances the .dat walk backwards,
// which cycles forever between it and the record before it.
func TestRebuildIdx_StopsAtNegativeSizeHeader(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("create volume: %v", err)
	}
	if _, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, false, false); err != nil {
		t.Fatalf("seed write: %v", err)
	}
	base := VolumeFileName(dir, "", 1)
	kept, err := os.ReadFile(base + ".idx")
	if err != nil {
		t.Fatalf("read the seeded idx: %v", err)
	}
	if _, _, _, err := v.writeNeedle2(newRandomNeedle(2), true, false, false); err != nil {
		t.Fatalf("seed second write: %v", err)
	}
	nv, ok := v.nm.Get(2)
	if !ok {
		t.Fatalf("second needle missing from the index")
	}
	corruptAt := nv.Offset.ToActualOffset()
	v.Close()

	// Overwrite the second needle's size field with a negative int32.
	dat, err := os.OpenFile(base+".dat", os.O_WRONLY, 0644)
	if err != nil {
		t.Fatalf("open dat: %v", err)
	}
	if _, err := dat.WriteAt([]byte{0xff, 0xff, 0xf0, 0x00}, corruptAt+types.CookieSize+types.NeedleIdSize); err != nil {
		t.Fatalf("corrupt size field: %v", err)
	}
	dat.Close()
	if err := os.Remove(base + ".idx"); err != nil {
		t.Fatalf("drop idx: %v", err)
	}

	// Pre-fix the rebuild walked backwards from here and never terminated.
	v2, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
	if err != nil {
		t.Fatalf("reload: %v", err)
	}
	defer v2.Close()

	rebuilt, err := os.ReadFile(base + ".idx")
	if err != nil {
		t.Fatalf("read the rebuilt idx: %v", err)
	}
	if !bytes.Equal(kept, rebuilt) {
		t.Errorf("rebuilt idx has %d bytes, want the %d covering only the intact needle", len(rebuilt), len(kept))
	}
}

// A rebuild that cannot write skips just this volume; it used to call
// glog.Fatalf and take the whole volume server down with it.
func TestRebuildIdx_UnwritableIdxDirSkipsOnlyThisVolume(t *testing.T) {
	if os.Geteuid() == 0 {
		t.Skip("root ignores directory permissions")
	}
	root := t.TempDir()
	dataDir := filepath.Join(root, "data")
	idxDir := filepath.Join(root, "idx")
	for _, dir := range []string{dataDir, idxDir} {
		if err := os.MkdirAll(dir, 0755); err != nil {
			t.Fatalf("mkdir %s: %v", dir, err)
		}
	}

	v, err := NewVolume(dataDir, dataDir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
	if err != nil {
		t.Fatalf("create volume: %v", err)
	}
	if _, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, false, false); err != nil {
		t.Fatalf("seed write: %v", err)
	}
	v.Close()
	if err := os.Remove(VolumeFileName(dataDir, "", 1) + ".idx"); err != nil {
		t.Fatalf("drop idx: %v", err)
	}
	if err := os.Chmod(idxDir, 0555); err != nil {
		t.Fatalf("seal idx dir: %v", err)
	}
	defer os.Chmod(idxDir, 0755)

	if _, err := NewVolume(dataDir, idxDir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0); err == nil {
		t.Errorf("expected a load error for an unwritable idx dir")
	}
}