Repository: petermattis/pebble
Pull request #107 (2019-05-02 12:52:02)

Fix replaying of the WAL when preallocation is active
Previously, replaying of a preallocated WAL would encounter the error:
`pebble/record: block appears to be zeroed`. This occurred because the
tail of the log was all zeroes. Now we treat the zeroed records error
similar to EOF.


Patch:
diff --git a/internal/record/record.go b/internal/record/record.go
index 1bd6970..71c4fdb 100644
--- a/internal/record/record.go
+++ b/internal/record/record.go
@@ -138,9 +138,8 @@ var (
 	// ErrNoLastRecord is returned if LastRecordOffset is called and there is no previous record.
 	ErrNoLastRecord = errors.New("pebble/record: no last record exists")
 
-	// ErrInvalidLogNum is returned if the log number in a record differs from
-	// the one expected.
-	ErrInvalidLogNum = errors.New("pebble/record: invalid log number")
+	// ErrZeroedChunk is returned if a chunk is encountered that is zeroed.
+	ErrZeroedChunk = errors.New("pebble/record: zeroed chunk")
 )
 
 // Reader reads records from an underlying io.Reader.
@@ -196,19 +195,16 @@ func (r *Reader) nextChunk(wantFirst bool) error {
 					r.end = r.n
 					continue
 				}
-				if wantFirst || r.recovering {
+				if r.recovering {
 					// Skip the rest of the block, if it looks like it is all
-					// zeroes. This is common if the record file was created via mmap.
+					// zeroes. This is common with WAL preallocation.
 					//
-					// Set r.err to be an error so r.Recover actually recovers.
+					// Set r.err to be an error so r.recover actually recovers.
 					r.err = errors.New("pebble/record: block appears to be zeroed")
-					if !r.recovering {
-						return r.err
-					}
 					r.recover()
 					continue
 				}
-				return errors.New("pebble/record: invalid chunk")
+				return ErrZeroedChunk
 			}
 
 			headerSize := legacyHeaderSize
@@ -220,7 +216,8 @@ func (r *Reader) nextChunk(wantFirst bool) error {
 
 				logNum := binary.LittleEndian.Uint32(r.buf[r.end+7 : r.end+11])
 				if logNum != r.logNum {
-					return ErrInvalidLogNum
+					// Treat a record from a previous instance of the log as EOF.
+					return io.EOF
 				}
 
 				chunkType -= (recyclableFullChunkType - 1)
diff --git a/open.go b/open.go
index bace5d4..adf0cf5 100644
--- a/open.go
+++ b/open.go
@@ -232,14 +232,16 @@ func (d *DB) replayWAL(
 	)
 	for {
 		r, err := rr.Next()
-		if err == io.EOF {
-			break
+		if err == nil {
+			_, err = io.Copy(&buf, r)
 		}
 		if err != nil {
-			return 0, err
-		}
-		_, err = io.Copy(&buf, r)
-		if err != nil {
+			// It is common to encounter a zeroed chunk due to WAL preallocation. We
+			// need to distinguish this from EOF in order to recognize that the
+			// record was truncated, but want to otherwise treat it like EOF.
+			if err == io.EOF || err == record.ErrZeroedChunk {
+				break
+			}
 			return 0, err
 		}
 


Test patch:
diff --git a/internal/record/record_test.go b/internal/record/record_test.go
index 82741bf..16021a5 100644
--- a/internal/record/record_test.go
+++ b/internal/record/record_test.go
@@ -812,8 +812,8 @@ func TestInvalidLogNum(t *testing.T) {
 
 	{
 		r := NewReader(bytes.NewReader(buf.Bytes()), 2)
-		if _, err := r.Next(); err != ErrInvalidLogNum {
-			t.Fatalf("expected %s, but found %s\n", ErrInvalidLogNum, err)
+		if _, err := r.Next(); err != io.EOF {
+			t.Fatalf("expected %s, but found %s\n", io.EOF, err)
 		}
 	}
 }
diff --git a/open_test.go b/open_test.go
index 4e4aad9..ee11c24 100644
--- a/open_test.go
+++ b/open_test.go
@@ -5,6 +5,8 @@
 package pebble
 
 import (
+	"io/ioutil"
+	"os"
 	"path/filepath"
 	"reflect"
 	"sort"
@@ -75,9 +77,9 @@ func TestNewDBFilenames(t *testing.T) {
 	}
 }
 
-func TestOpenCloseOpenClose(t *testing.T) {
+func testOpenCloseOpenClose(t *testing.T, fs storage.Storage, root string) {
 	opts := &db.Options{
-		Storage: storage.NewMem(),
+		Storage: fs,
 	}
 
 	for _, startFromEmpty := range []bool{false, true} {
@@ -86,6 +88,7 @@ func TestOpenCloseOpenClose(t *testing.T) {
 			if startFromEmpty {
 				dirname = "startFromEmpty" + strconv.Itoa(length)
 			}
+			dirname = filepath.Join(root, dirname)
 
 			got, xxx := []byte(nil), ""
 			if length >= 0 {
@@ -113,12 +116,6 @@ func TestOpenCloseOpenClose(t *testing.T) {
 				continue
 			}
 
-			// TODO(peter): make the second Open recover (without a fatal "corrupt
-			// log file" error) even if the d0 database was not closed but the xxx
-			// value is large enough to write a partial record. Writing to the
-			// database should not corrupt it even if the writer process was killed
-			// part-way through.
-
 			d1, err := Open(dirname, opts)
 			if err != nil {
 				t.Errorf("sfe=%t, length=%d: Open #1: %v",
@@ -165,6 +162,31 @@ func TestOpenCloseOpenClose(t *testing.T) {
 	}
 }
 
+func TestOpenCloseOpenClose(t *testing.T) {
+	for _, fstype := range []string{"disk", "mem"} {
+		t.Run(fstype, func(t *testing.T) {
+			var fs storage.Storage
+			var dir string
+			switch fstype {
+			case "disk":
+				var err error
+				dir, err = ioutil.TempDir("", "open-close")
+				if err != nil {
+					t.Fatal(err)
+				}
+				defer func() {
+					_ = os.RemoveAll(dir)
+				}()
+				fs = storage.Default
+			case "mem":
+				dir = ""
+				fs = storage.NewMem()
+			}
+			testOpenCloseOpenClose(t, fs, dir)
+		})
+	}
+}
+
 func TestOpenOptionsCheck(t *testing.T) {
 	mem := storage.NewMem()
 	opts := &db.Options{Storage: mem}