Repository: cockroachdb/pebble
Pull request #1214 (2021-08-03 15:26:32)

crl-release-20.2: db: write manifest before creating WAL during Open 
CockroachDB 20.2 backport of #1212.

----

Write out the new manifest file before creating a new WAL file during
Open. Previously, a crash between the creation of the WAL and the
MANIFEST could leave the second most recent WAL with an unclean tail.
The unclean tail would be considered corruption on a subsequent Open.

See cockroachdb/cockroach#68319.

cockroach-teamcity: 

<!-- Reviewable:start -->
This change is [<img src="https://reviewable.io/review_button.png" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/cockroachdb/pebble/1214)
<!-- Reviewable:end -->


Patch:
diff --git a/internal/errorfs/errorfs.go b/internal/errorfs/errorfs.go
index e294b3cf..b0cc291c 100644
--- a/internal/errorfs/errorfs.go
+++ b/internal/errorfs/errorfs.go
@@ -5,6 +5,7 @@
 package errorfs
 
 import (
+	"fmt"
 	"io"
 	"math/rand"
 	"os"
@@ -19,14 +20,73 @@ import (
 // ErrInjected is an error artifically injected for testing fs error paths.
 var ErrInjected = errors.New("injected error")
 
-// Op is an enum describing the type of operation performed.
+// Op is an enum describing the type of operation.
 type Op int
 
 const (
-	// OpRead describes read operations.
-	OpRead Op = iota
-	// OpWrite describes write operations.
-	OpWrite
+	// OpCreate describes a create file operation.
+	OpCreate Op = iota
+	// OpLink describes a hardlink operation.
+	OpLink
+	// OpOpen describes a file open operation.
+	OpOpen
+	// OpOpenDir describes a directory open operation.
+	OpOpenDir
+	// OpRemove describes a remove file operation.
+	OpRemove
+	// OpRemoveAll describes a recursive remove operation.
+	OpRemoveAll
+	// OpRename describes a rename operation.
+	OpRename
+	// OpReuseForRewrite describes a reuse for rewriting operation.
+	OpReuseForRewrite
+	// OpMkdirAll describes a make directory including parents operation.
+	OpMkdirAll
+	// OpLock describes a lock file operation.
+	OpLock
+	// OpList describes a list directory operation.
+	OpList
+	// OpStat describes a path-based stat operation.
+	OpStat
+	// OpGetFreeSpace describes a disk usage operation.
+	OpGetFreeSpace
+	// OpFileClose describes a close file operation.
+	OpFileClose
+	// OpFileRead describes a file read operation.
+	OpFileRead
+	// OpFileReadAt describes a file seek read operation.
+	OpFileReadAt
+	// OpFileWrite describes a file write operation.
+	OpFileWrite
+	// OpFileStat describes a file stat operation.
+	OpFileStat
+	// OpFileSync describes a file sync operation.
+	OpFileSync
+	// OpFileFlush describes a file flush operation.
+	OpFileFlush
+)
+
+// OpKind returns the operation's kind.
+func (o Op) OpKind() OpKind {
+	switch o {
+	case OpOpen, OpOpenDir, OpList, OpStat, OpGetFreeSpace, OpFileRead, OpFileReadAt, OpFileStat:
+		return OpKindRead
+	case OpCreate, OpLink, OpRemove, OpRemoveAll, OpRename, OpReuseForRewrite, OpMkdirAll, OpLock, OpFileClose, OpFileWrite, OpFileSync, OpFileFlush:
+		return OpKindWrite
+	default:
+		panic(fmt.Sprintf("unrecognized op %v\n", o))
+	}
+}
+
+// OpKind is an enum describing whether an operation is a read or write
+// operation.
+type OpKind int
+
+const (
+	// OpKindRead describes read operations.
+	OpKindRead OpKind = iota
+	// OpKindWrite describes write operations.
+	OpKindWrite
 )
 
 // OnIndex constructs an injector that returns an error on
@@ -48,7 +108,7 @@ func (ii *InjectIndex) Index() int32 { return atomic.LoadInt32(&ii.index) }
 func (ii *InjectIndex) SetIndex(v int32) { atomic.StoreInt32(&ii.index, v) }
 
 // MaybeError implements the Injector interface.
-func (ii *InjectIndex) MaybeError(op Op) error {
+func (ii *InjectIndex) MaybeError(_ Op, _ string) error {
 	if atomic.AddInt32(&ii.index, -1) == -1 {
 		return errors.WithStack(ErrInjected)
 	}
@@ -59,13 +119,13 @@ func (ii *InjectIndex) MaybeError(op Op) error {
 // probability when passed op. It may be passed to Wrap to inject an error
 // into an ErrFS with the provided probability. p should be within the range
 // [0.0,1.0].
-func WithProbability(op Op, p float64) Injector {
+func WithProbability(op OpKind, p float64) Injector {
 	mu := new(sync.Mutex)
 	rnd := rand.New(rand.NewSource(time.Now().UnixNano()))
-	return InjectorFunc(func(currOp Op) error {
+	return InjectorFunc(func(currOp Op, _ string) error {
 		mu.Lock()
 		defer mu.Unlock()
-		if currOp == op && rnd.Float64() < p {
+		if currOp.OpKind() == op && rnd.Float64() < p {
 			return errors.WithStack(ErrInjected)
 		}
 		return nil
@@ -74,14 +134,18 @@ func WithProbability(op Op, p float64) Injector {
 
 // InjectorFunc implements the Injector interface for a function with
 // MaybeError's signature.
-type InjectorFunc func(Op) error
+type InjectorFunc func(Op, string) error
 
 // MaybeError implements the Injector interface.
-func (f InjectorFunc) MaybeError(op Op) error { return f(op) }
+func (f InjectorFunc) MaybeError(op Op, path string) error { return f(op, path) }
 
 // Injector injects errors into FS operations.
 type Injector interface {
-	MaybeError(Op) error
+	// MaybeError is invoked by an errorfs before an operation is executed. It
+	// is passed an enum indicating the type of operation and a path of the
+	// subject file or directory. If the operation takes two paths (eg,
+	// Rename, Link), the original source path is provided.
+	MaybeError(op Op, path string) error
 }
 
 // FS implements vfs.FS, injecting errors into
@@ -119,19 +183,19 @@ func (fs *FS) Unwrap() vfs.FS {
 
 // Create implements FS.Create.
 func (fs *FS) Create(name string) (vfs.File, error) {
-	if err := fs.inj.MaybeError(OpWrite); err != nil {
+	if err := fs.inj.MaybeError(OpCreate, name); err != nil {
 		return nil, err
 	}
 	f, err := fs.fs.Create(name)
 	if err != nil {
 		return nil, err
 	}
-	return &errorFile{f, fs.inj}, nil
+	return &errorFile{name, f, fs.inj}, nil
 }
 
 // Link implements FS.Link.
 func (fs *FS) Link(oldname, newname string) error {
-	if err := fs.inj.MaybeError(OpWrite); err != nil {
+	if err := fs.inj.MaybeError(OpLink, oldname); err != nil {
 		return err
 	}
 	return fs.fs.Link(oldname, newname)
@@ -139,14 +203,14 @@ func (fs *FS) Link(oldname, newname string) error {
 
 // Open implements FS.Open.
 func (fs *FS) Open(name string, opts ...vfs.OpenOption) (vfs.File, error) {
-	if err := fs.inj.MaybeError(OpRead); err != nil {
+	if err := fs.inj.MaybeError(OpOpen, name); err != nil {
 		return nil, err
 	}
 	f, err := fs.fs.Open(name)
 	if err != nil {
 		return nil, err
 	}
-	ef := &errorFile{f, fs.inj}
+	ef := &errorFile{name, f, fs.inj}
 	for _, opt := range opts {
 		opt.Apply(ef)
 	}
@@ -155,19 +219,19 @@ func (fs *FS) Open(name string, opts ...vfs.OpenOption) (vfs.File, error) {
 
 // OpenDir implements FS.OpenDir.
 func (fs *FS) OpenDir(name string) (vfs.File, error) {
-	if err := fs.inj.MaybeError(OpRead); err != nil {
+	if err := fs.inj.MaybeError(OpOpenDir, name); err != nil {
 		return nil, err
 	}
 	f, err := fs.fs.OpenDir(name)
 	if err != nil {
 		return nil, err
 	}
-	return &errorFile{f, fs.inj}, nil
+	return &errorFile{name, f, fs.inj}, nil
 }
 
 // GetFreeSpace implements FS.GetFreeSpace.
 func (fs *FS) GetFreeSpace(path string) (uint64, error) {
-	if err := fs.inj.MaybeError(OpRead); err != nil {
+	if err := fs.inj.MaybeError(OpGetFreeSpace, path); err != nil {
 		return 0, err
 	}
 	return fs.fs.GetFreeSpace(path)
@@ -194,7 +258,7 @@ func (fs *FS) Remove(name string) error {
 		return nil
 	}
 
-	if err := fs.inj.MaybeError(OpWrite); err != nil {
+	if err := fs.inj.MaybeError(OpRemove, name); err != nil {
 		return err
 	}
 	return fs.fs.Remove(name)
@@ -202,7 +266,7 @@ func (fs *FS) Remove(name string) error {
 
 // RemoveAll implements FS.RemoveAll.
 func (fs *FS) RemoveAll(fullname string) error {
-	if err := fs.inj.MaybeError(OpWrite); err != nil {
+	if err := fs.inj.MaybeError(OpRemoveAll, fullname); err != nil {
 		return err
 	}
 	return fs.fs.RemoveAll(fullname)
@@ -210,7 +274,7 @@ func (fs *FS) RemoveAll(fullname string) error {
 
 // Rename implements FS.Rename.
 func (fs *FS) Rename(oldname, newname string) error {
-	if err := fs.inj.MaybeError(OpWrite); err != nil {
+	if err := fs.inj.MaybeError(OpRename, oldname); err != nil {
 		return err
 	}
 	return fs.fs.Rename(oldname, newname)
@@ -218,7 +282,7 @@ func (fs *FS) Rename(oldname, newname string) error {
 
 // ReuseForWrite implements FS.ReuseForWrite.
 func (fs *FS) ReuseForWrite(oldname, newname string) (vfs.File, error) {
-	if err := fs.inj.MaybeError(OpWrite); err != nil {
+	if err := fs.inj.MaybeError(OpReuseForRewrite, oldname); err != nil {
 		return nil, err
 	}
 	return fs.fs.ReuseForWrite(oldname, newname)
@@ -226,7 +290,7 @@ func (fs *FS) ReuseForWrite(oldname, newname string) (vfs.File, error) {
 
 // MkdirAll implements FS.MkdirAll.
 func (fs *FS) MkdirAll(dir string, perm os.FileMode) error {
-	if err := fs.inj.MaybeError(OpWrite); err != nil {
+	if err := fs.inj.MaybeError(OpMkdirAll, dir); err != nil {
 		return err
 	}
 	return fs.fs.MkdirAll(dir, perm)
@@ -234,7 +298,7 @@ func (fs *FS) MkdirAll(dir string, perm os.FileMode) error {
 
 // Lock implements FS.Lock.
 func (fs *FS) Lock(name string) (io.Closer, error) {
-	if err := fs.inj.MaybeError(OpWrite); err != nil {
+	if err := fs.inj.MaybeError(OpLock, name); err != nil {
 		return nil, err
 	}
 	return fs.fs.Lock(name)
@@ -242,7 +306,7 @@ func (fs *FS) Lock(name string) (io.Closer, error) {
 
 // List implements FS.List.
 func (fs *FS) List(dir string) ([]string, error) {
-	if err := fs.inj.MaybeError(OpRead); err != nil {
+	if err := fs.inj.MaybeError(OpList, dir); err != nil {
 		return nil, err
 	}
 	return fs.fs.List(dir)
@@ -250,7 +314,7 @@ func (fs *FS) List(dir string) ([]string, error) {
 
 // Stat implements FS.Stat.
 func (fs *FS) Stat(name string) (os.FileInfo, error) {
-	if err := fs.inj.MaybeError(OpRead); err != nil {
+	if err := fs.inj.MaybeError(OpStat, name); err != nil {
 		return nil, err
 	}
 	return fs.fs.Stat(name)
@@ -259,6 +323,7 @@ func (fs *FS) Stat(name string) (os.FileInfo, error) {
 // errorFile implements vfs.File. The interface is implemented on the pointer
 // type to allow pointer equality comparisons.
 type errorFile struct {
+	path string
 	file vfs.File
 	inj  Injector
 }
@@ -270,35 +335,35 @@ func (f *errorFile) Close() error {
 }
 
 func (f *errorFile) Read(p []byte) (int, error) {
-	if err := f.inj.MaybeError(OpRead); err != nil {
+	if err := f.inj.MaybeError(OpFileRead, f.path); err != nil {
 		return 0, err
 	}
 	return f.file.Read(p)
 }
 
 func (f *errorFile) ReadAt(p []byte, off int64) (int, error) {
-	if err := f.inj.MaybeError(OpRead); err != nil {
+	if err := f.inj.MaybeError(OpFileReadAt, f.path); err != nil {
 		return 0, err
 	}
 	return f.file.ReadAt(p, off)
 }
 
 func (f *errorFile) Write(p []byte) (int, error) {
-	if err := f.inj.MaybeError(OpWrite); err != nil {
+	if err := f.inj.MaybeError(OpFileWrite, f.path); err != nil {
 		return 0, err
 	}
 	return f.file.Write(p)
 }
 
 func (f *errorFile) Stat() (os.FileInfo, error) {
-	if err := f.inj.MaybeError(OpRead); err != nil {
+	if err := f.inj.MaybeError(OpFileStat, f.path); err != nil {
 		return nil, err
 	}
 	return f.file.Stat()
 }
 
 func (f *errorFile) Sync() error {
-	if err := f.inj.MaybeError(OpWrite); err != nil {
+	if err := f.inj.MaybeError(OpFileSync, f.path); err != nil {
 		return err
 	}
 	return f.file.Sync()
diff --git a/open.go b/open.go
index 3f9a6bc6..a5cbf2c8 100644
--- a/open.go
+++ b/open.go
@@ -283,6 +283,23 @@ func Open(dirname string, opts *Options) (db *DB, _ error) {
 	if !d.opts.ReadOnly {
 		// Create an empty .log file.
 		newLogNum := d.mu.versions.getNextFileNum()
+
+		// This logic is slightly different than RocksDB's. Specifically, RocksDB
+		// sets MinUnflushedLogNum to max-recovered-log-num + 1. We set it to the
+		// newLogNum. There should be no difference in using either value.
+		ve.MinUnflushedLogNum = newLogNum
+
+		// Create the manifest with the updated MinUnflushedLogNum before
+		// creating the new log file. If we created the log file first, a
+		// crash before the manifest is synced could leave two WALs with
+		// unclean tails.
+		d.mu.versions.logLock()
+		if err := d.mu.versions.logAndApply(jobID, &ve, newFileMetrics(ve.NewFiles), d.dataDir, func() []compactionInfo {
+			return nil
+		}); err != nil {
+			return nil, err
+		}
+
 		newLogName := base.MakeFilename(opts.FS, d.walDirname, fileTypeLog, newLogNum)
 		d.mu.log.queue = append(d.mu.log.queue, newLogNum)
 		logFile, err := opts.FS.Create(newLogName)
@@ -308,17 +325,6 @@ func Open(dirname string, opts *Options) (db *DB, _ error) {
 		d.mu.log.LogWriter = record.NewLogWriter(logFile, newLogNum)
 		d.mu.log.LogWriter.SetMinSyncInterval(d.opts.WALMinSyncInterval)
 		d.mu.versions.metrics.WAL.Files++
-
-		// This logic is slightly different than RocksDB's. Specifically, RocksDB
-		// sets MinUnflushedLogNum to max-recovered-log-num + 1. We set it to the
-		// newLogNum. There should be no difference in using either value.
-		ve.MinUnflushedLogNum = newLogNum
-		d.mu.versions.logLock()
-		if err := d.mu.versions.logAndApply(jobID, &ve, newFileMetrics(ve.NewFiles), d.dataDir, func() []compactionInfo {
-			return nil
-		}); err != nil {
-			return nil, err
-		}
 	}
 	d.updateReadStateLocked(d.opts.DebugCheck)
 


Test patch:
diff --git a/error_test.go b/error_test.go
index 6b5eb4d9..2616eddc 100644
--- a/error_test.go
+++ b/error_test.go
@@ -335,8 +335,8 @@ func TestDBWALRotationCrash(t *testing.T) {
 	memfs := vfs.NewStrictMem()
 
 	var index int32
-	inj := errorfs.InjectorFunc(func(op errorfs.Op) error {
-		if op == errorfs.OpWrite && atomic.AddInt32(&index, -1) == -1 {
+	inj := errorfs.InjectorFunc(func(op errorfs.Op, _ string) error {
+		if op.OpKind() == errorfs.OpKindWrite && atomic.AddInt32(&index, -1) == -1 {
 			memfs.SetIgnoreSyncs(true)
 		}
 		return nil
diff --git a/internal/metamorphic/meta_test.go b/internal/metamorphic/meta_test.go
index cabfe818..80b14617 100644
--- a/internal/metamorphic/meta_test.go
+++ b/internal/metamorphic/meta_test.go
@@ -99,7 +99,7 @@ func testMetaRun(t *testing.T, runDir string) {
 	}
 	// Wrap the filesystem with one that will inject errors into read
 	// operations with *errorRate probability.
-	opts.FS = errorfs.Wrap(opts.FS, errorfs.WithProbability(errorfs.OpRead, *errorRate))
+	opts.FS = errorfs.Wrap(opts.FS, errorfs.WithProbability(errorfs.OpKindRead, *errorRate))
 
 	if opts.WALDir != "" {
 		opts.WALDir = opts.FS.PathJoin(runDir, opts.WALDir)
diff --git a/open_test.go b/open_test.go
index 92823a27..acae554d 100644
--- a/open_test.go
+++ b/open_test.go
@@ -14,9 +14,11 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync/atomic"
 	"testing"
 
 	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/errorfs"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/kr/pretty"
 	"github.com/stretchr/testify/require"
@@ -629,6 +631,110 @@ func TestTwoWALReplayPermissive(t *testing.T) {
 	require.NoError(t, d.Close())
 }
 
+// TestCrashOpenCrashAfterWALCreation tests a database that exits
+// ungracefully, begins recovery, creates the new WAL but promptly exits
+// ungracefully again.
+//
+// This sequence has the potential to be problematic with the strict_wal_tail
+// behavior because the first crash's WAL has an unclean tail. By the time the
+// new WAL is created, the current manifest's MinUnflushedLogNum must be
+// higher than the previous WAL.
+func TestCrashOpenCrashAfterWALCreation(t *testing.T) {
+	fs := vfs.NewStrictMem()
+
+	getLogs := func() (logs []string) {
+		ls, err := fs.List("")
+		require.NoError(t, err)
+		for _, name := range ls {
+			if filepath.Ext(name) == ".log" {
+				logs = append(logs, name)
+			}
+		}
+		return logs
+	}
+
+	{
+		d, err := Open("", &Options{FS: fs})
+		require.NoError(t, err)
+		require.NoError(t, d.Set([]byte("abc"), nil, Sync))
+
+		// Ignore syncs during close to simulate a crash. This will leave the WAL
+		// without an EOF trailer. It won't be an 'unclean tail' yet since the
+		// log file was not recycled, but we'll fix that down below.
+		fs.SetIgnoreSyncs(true)
+		require.NoError(t, d.Close())
+		fs.ResetToSyncedState()
+		fs.SetIgnoreSyncs(false)
+	}
+
+	// There should be one WAL.
+	logs := getLogs()
+	if len(logs) != 1 {
+		t.Fatalf("expected one log file, found %d", len(logs))
+	}
+
+	// The one WAL file doesn't have an EOF trailer, but since it wasn't
+	// recycled it won't have garbage at the end. Rewrite it so that it has
+	// the same contents it currently has, followed by garbage.
+	{
+		f, err := fs.Open(logs[0])
+		require.NoError(t, err)
+		b, err := ioutil.ReadAll(f)
+		require.NoError(t, err)
+		require.NoError(t, f.Close())
+		f, err = fs.Create(logs[0])
+		_, err = f.Write(b)
+		require.NoError(t, err)
+		_, err = f.Write([]byte{0xde, 0xad, 0xbe, 0xef})
+		require.NoError(t, err)
+		require.NoError(t, f.Sync())
+		require.NoError(t, f.Close())
+		dir, err := fs.OpenDir("")
+		require.NoError(t, err)
+		require.NoError(t, dir.Sync())
+		require.NoError(t, dir.Close())
+	}
+
+	// Open the database again (with syncs respected again). Wrap the
+	// filesystem with an errorfs that will turn off syncs after a new .log
+	// file is created and after a subsequent directory sync occurs. This
+	// simulates a crash after the new log file is created and synced.
+	{
+		var atomicWALCreated, atomicDirSynced uint32
+		d, err := Open("", &Options{
+			FS: errorfs.Wrap(fs, errorfs.InjectorFunc(func(op errorfs.Op, path string) error {
+				if atomic.LoadUint32(&atomicDirSynced) == 1 {
+					fs.SetIgnoreSyncs(true)
+				}
+				if op == errorfs.OpCreate && filepath.Ext(path) == ".log" {
+					atomic.StoreUint32(&atomicWALCreated, 1)
+				}
+				// Record when there's a sync of the data directory after the
+				// WAL was created. The data directory will have an empty
+				// path because that's what we passed into Open.
+				if op == errorfs.OpFileSync && path == "" && atomic.LoadUint32(&atomicWALCreated) == 1 {
+					atomic.StoreUint32(&atomicDirSynced, 1)
+				}
+				return nil
+			})),
+		})
+		require.NoError(t, err)
+		require.NoError(t, d.Close())
+	}
+
+	fs.ResetToSyncedState()
+	fs.SetIgnoreSyncs(false)
+
+	if n := len(getLogs()); n != 2 {
+		t.Fatalf("expected two logs, found %d\n", n)
+	}
+
+	// Finally, open the database with syncs enabled.
+	d, err := Open("", &Options{FS: fs})
+	require.NoError(t, err)
+	require.NoError(t, d.Close())
+}
+
 // TestOpenWALReplayReadOnlySeqNums tests opening a database:
 // * in read-only mode
 // * with multiple unflushed log files that must replayed
diff --git a/testdata/checkpoint b/testdata/checkpoint
index 09102381..4faaa8b8 100644
--- a/testdata/checkpoint
+++ b/testdata/checkpoint
@@ -10,9 +10,9 @@ sync: db/CURRENT.000001.dbtmp
 close: db/CURRENT.000001.dbtmp
 rename: db/CURRENT.000001.dbtmp -> db/CURRENT
 sync: db
+sync: db/MANIFEST-000001
 create: db/000002.log
 sync: db
-sync: db/MANIFEST-000001
 create: db/OPTIONS-000003
 sync: db/OPTIONS-000003
 close: db/OPTIONS-000003
diff --git a/testdata/cleaner b/testdata/cleaner
index 67ff6700..eb61ca39 100644
--- a/testdata/cleaner
+++ b/testdata/cleaner
@@ -12,9 +12,9 @@ sync: db/CURRENT.000001.dbtmp
 close: db/CURRENT.000001.dbtmp
 rename: db/CURRENT.000001.dbtmp -> db/CURRENT
 sync: db
+sync: db/MANIFEST-000001
 create: wal/000002.log
 sync: wal
-sync: db/MANIFEST-000001
 create: db/OPTIONS-000003
 sync: db/OPTIONS-000003
 close: db/OPTIONS-000003
diff --git a/testdata/event_listener b/testdata/event_listener
index 5a38783d..5ee14c81 100644
--- a/testdata/event_listener
+++ b/testdata/event_listener
@@ -13,9 +13,6 @@ close: db/CURRENT.000001.dbtmp
 rename: db/CURRENT.000001.dbtmp -> db/CURRENT
 sync: db
 [JOB 1] MANIFEST created 000001
-create: wal/000002.log
-sync: wal
-[JOB 1] WAL created 000002
 create: db/MANIFEST-000003
 close: db/MANIFEST-000001
 sync: db/MANIFEST-000003
@@ -25,6 +22,9 @@ close: db/CURRENT.000003.dbtmp
 rename: db/CURRENT.000003.dbtmp -> db/CURRENT
 sync: db
 [JOB 1] MANIFEST created 000003
+create: wal/000002.log
+sync: wal
+[JOB 1] WAL created 000002
 create: db/OPTIONS-000004
 sync: db/OPTIONS-000004
 close: db/OPTIONS-000004