Deal with zero-length segments

Signed-off-by: Fabian Reinartz <[email protected]>

Author: Fabian Reinartz

repair_test.go

@@ -76,7 +76,7 @@ func TestRepairBadIndexVersion(t *testing.T) {
 	}
 
 	// On DB opening all blocks in the base dir should be repaired.
-	db, _ := Open("testdata/repair_index_version", nil, nil, nil)
+	db, err := Open("testdata/repair_index_version", nil, nil, nil)
 	if err != nil {
 		t.Fatal(err)
 	}

wal.go

@@ -1214,6 +1214,9 @@ func (r *walReader) decodeDeletes(flag byte, b []byte, res *[]Stone) error {
 
 // MigrateWAL rewrites the deprecated write ahead log into the new format.
 func MigrateWAL(logger log.Logger, dir string) (err error) {
+	if logger == nil {
+		logger = log.NewNopLogger()
+	}
 	// Detect whether we still have the old WAL.
 	fns, err := sequenceFiles(dir)
 	if err != nil && !os.IsNotExist(err) {
@@ -1222,21 +1225,23 @@ func MigrateWAL(logger log.Logger, dir string) (err error) {
 	if len(fns) == 0 {
 		return nil // No WAL at all yet.
 	}
-	// Check header of first segment.
+	// Check header of first segment to see whether we are still dealing with an
+	// old WAL.
 	f, err := os.Open(fns[0])
 	if err != nil {
 		return errors.Wrap(err, "check first existing segment")
 	}
 	defer f.Close()
 
 	var hdr [4]byte
-	if n, err := f.Read(hdr[:]); err != nil {
+	if _, err := f.Read(hdr[:]); err != nil && err != io.EOF {
 		return errors.Wrap(err, "read header from first segment")
-	} else if n != 4 {
-		return errors.New("could not read full header from segment")
 	}
+	// If we cannot read the magic header for segments of the old WAL, abort.
+	// Either it's migrated already or there's a corruption issue with which
+	// we cannot deal here anyway. Subsequent attempts to open the WAL will error in that case.
 	if binary.BigEndian.Uint32(hdr[:]) != WALMagic {
-		return nil // Not the old WAL anymore.
+		return nil
 	}
 
 	level.Info(logger).Log("msg", "migrating WAL format")

wal_test.go

@@ -434,6 +434,23 @@ func TestWALRestoreCorrupted(t *testing.T) {
 	}
 }
 
+func TestMigrateWAL_Empty(t *testing.T) {
+	// The migration proecedure must properly deal with a zero-length segment,
+	// which is valid in the new format.
+	dir, err := ioutil.TempDir("", "walmigrate")
+	testutil.Ok(t, err)
+	defer os.RemoveAll(dir)
+
+	wdir := path.Join(dir, "wal")
+
+	// Initialize empty WAL.
+	w, err := wal.New(nil, nil, wdir)
+	testutil.Ok(t, err)
+	testutil.Ok(t, w.Close())
+
+	testutil.Ok(t, MigrateWAL(nil, wdir))
+}
+
 func TestMigrateWAL_Fuzz(t *testing.T) {
 	dir, err := ioutil.TempDir("", "walmigrate")
 	testutil.Ok(t, err)