Repository: cockroachdb/pebble
Pull request #606 (2020-04-01 21:14:14)

internal/metamorphic: add error rate
Add  an -error-rate flag to the metamorphic tests to inject errors
into read operations at the provided rate.

petermattis: 

<!-- 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/606)
<!-- Reviewable:end -->


Patch:
diff --git a/internal/errorfs/errorfs.go b/internal/errorfs/errorfs.go
index a5eed066..fea3374b 100644
--- a/internal/errorfs/errorfs.go
+++ b/internal/errorfs/errorfs.go
@@ -6,8 +6,11 @@ package errorfs
 
 import (
 	"io"
+	"math/rand"
 	"os"
+	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/vfs"
@@ -16,6 +19,16 @@ 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.
+type Op int
+
+const (
+	// OpRead describes read operations.
+	OpRead Op = iota
+	// OpWrite describes write operations.
+	OpWrite
+)
+
 // OnIndex constructs an injector that returns an error on
 // the (n+1)-th invocation of its MaybeError function. It
 // may be passed to Wrap to inject an error into an FS.
@@ -35,16 +48,37 @@ 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() error {
+func (ii *InjectIndex) MaybeError(op Op) error {
 	if atomic.AddInt32(&ii.index, -1) == -1 {
 		return errors.WithStack(ErrInjected)
 	}
 	return nil
 }
 
+// WithProbability returns a function that returns an error with the provided
+// 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 {
+	mu := new(sync.Mutex)
+	rnd := rand.New(rand.NewSource(time.Now().UnixNano()))
+	return injectorFunc(func(currOp Op) error {
+		mu.Lock()
+		defer mu.Unlock()
+		if currOp == op && rnd.Float64() < p {
+			return errors.WithStack(ErrInjected)
+		}
+		return nil
+	})
+}
+
+type injectorFunc func(Op) error
+
+func (f injectorFunc) MaybeError(op Op) error { return f(op) }
+
 // Injector injects errors into FS operations.
 type Injector interface {
-	MaybeError() error
+	MaybeError(Op) error
 }
 
 // FS implements vfs.FS, injecting errors into
@@ -71,24 +105,30 @@ func Wrap(fs vfs.FS, inj Injector) *FS {
 // deciding when to inject errors. If an error is injected, the file
 // propagates the error instead of shadowing the operation.
 func WrapFile(f vfs.File, inj Injector) vfs.File {
-	return errorFile{file: f, inj: inj}
+	return &errorFile{file: f, inj: inj}
+}
+
+// Unwrap returns the FS implementation underlying fs.
+// See pebble/vfs.Root.
+func (fs *FS) Unwrap() vfs.FS {
+	return fs.fs
 }
 
 // Create implements FS.Create.
 func (fs *FS) Create(name string) (vfs.File, error) {
-	if err := fs.inj.MaybeError(); err != nil {
+	if err := fs.inj.MaybeError(OpWrite); 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{f, fs.inj}, nil
 }
 
 // Link implements FS.Link.
 func (fs *FS) Link(oldname, newname string) error {
-	if err := fs.inj.MaybeError(); err != nil {
+	if err := fs.inj.MaybeError(OpWrite); err != nil {
 		return err
 	}
 	return fs.fs.Link(oldname, newname)
@@ -96,14 +136,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(); err != nil {
+	if err := fs.inj.MaybeError(OpRead); err != nil {
 		return nil, err
 	}
 	f, err := fs.fs.Open(name)
 	if err != nil {
 		return nil, err
 	}
-	ef := errorFile{f, fs.inj}
+	ef := &errorFile{f, fs.inj}
 	for _, opt := range opts {
 		opt.Apply(ef)
 	}
@@ -112,14 +152,14 @@ 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(); err != nil {
+	if err := fs.inj.MaybeError(OpRead); 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{f, fs.inj}, nil
 }
 
 // PathBase implements FS.PathBase.
@@ -143,7 +183,7 @@ func (fs *FS) Remove(name string) error {
 		return nil
 	}
 
-	if err := fs.inj.MaybeError(); err != nil {
+	if err := fs.inj.MaybeError(OpWrite); err != nil {
 		return err
 	}
 	return fs.fs.Remove(name)
@@ -151,7 +191,7 @@ func (fs *FS) Remove(name string) error {
 
 // RemoveAll implements FS.RemoveAll.
 func (fs *FS) RemoveAll(fullname string) error {
-	if err := fs.inj.MaybeError(); err != nil {
+	if err := fs.inj.MaybeError(OpWrite); err != nil {
 		return err
 	}
 	return fs.fs.RemoveAll(fullname)
@@ -159,7 +199,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(); err != nil {
+	if err := fs.inj.MaybeError(OpWrite); err != nil {
 		return err
 	}
 	return fs.fs.Rename(oldname, newname)
@@ -167,7 +207,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(); err != nil {
+	if err := fs.inj.MaybeError(OpWrite); err != nil {
 		return nil, err
 	}
 	return fs.fs.ReuseForWrite(oldname, newname)
@@ -175,7 +215,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(); err != nil {
+	if err := fs.inj.MaybeError(OpWrite); err != nil {
 		return err
 	}
 	return fs.fs.MkdirAll(dir, perm)
@@ -183,7 +223,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(); err != nil {
+	if err := fs.inj.MaybeError(OpWrite); err != nil {
 		return nil, err
 	}
 	return fs.fs.Lock(name)
@@ -191,7 +231,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(); err != nil {
+	if err := fs.inj.MaybeError(OpRead); err != nil {
 		return nil, err
 	}
 	return fs.fs.List(dir)
@@ -199,53 +239,55 @@ 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(); err != nil {
+	if err := fs.inj.MaybeError(OpRead); err != nil {
 		return nil, err
 	}
 	return fs.fs.Stat(name)
 }
 
+// errorFile implements vfs.File. The interface is implemented on the pointer
+// type to allow pointer equality comparisons.
 type errorFile struct {
 	file vfs.File
 	inj  Injector
 }
 
-func (f errorFile) Close() error {
+func (f *errorFile) Close() error {
 	// We don't inject errors during close as those calls should never fail in
 	// practice.
 	return f.file.Close()
 }
 
-func (f errorFile) Read(p []byte) (int, error) {
-	if err := f.inj.MaybeError(); err != nil {
+func (f *errorFile) Read(p []byte) (int, error) {
+	if err := f.inj.MaybeError(OpRead); 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(); err != nil {
+func (f *errorFile) ReadAt(p []byte, off int64) (int, error) {
+	if err := f.inj.MaybeError(OpRead); err != nil {
 		return 0, err
 	}
 	return f.file.ReadAt(p, off)
 }
 
-func (f errorFile) Write(p []byte) (int, error) {
-	if err := f.inj.MaybeError(); err != nil {
+func (f *errorFile) Write(p []byte) (int, error) {
+	if err := f.inj.MaybeError(OpWrite); err != nil {
 		return 0, err
 	}
 	return f.file.Write(p)
 }
 
-func (f errorFile) Stat() (os.FileInfo, error) {
-	if err := f.inj.MaybeError(); err != nil {
+func (f *errorFile) Stat() (os.FileInfo, error) {
+	if err := f.inj.MaybeError(OpRead); err != nil {
 		return nil, err
 	}
 	return f.file.Stat()
 }
 
-func (f errorFile) Sync() error {
-	if err := f.inj.MaybeError(); err != nil {
+func (f *errorFile) Sync() error {
+	if err := f.inj.MaybeError(OpWrite); err != nil {
 		return err
 	}
 	return f.file.Sync()
diff --git a/internal/metamorphic/ops.go b/internal/metamorphic/ops.go
index 477be541..63f213af 100644
--- a/internal/metamorphic/ops.go
+++ b/internal/metamorphic/ops.go
@@ -6,13 +6,16 @@ package metamorphic
 
 import (
 	"fmt"
+	"io"
 	"os"
 	"strings"
 
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble"
+	"github.com/cockroachdb/pebble/internal/errorfs"
 	"github.com/cockroachdb/pebble/internal/private"
 	"github.com/cockroachdb/pebble/sstable"
+	"github.com/cockroachdb/pebble/vfs"
 )
 
 // op defines the interface for a single operation, such as creating a batch,
@@ -31,7 +34,7 @@ type initOp struct {
 
 func (o *initOp) run(t *test, h *history) {
 	t.batches = make([]*pebble.Batch, o.batchSlots)
-	t.iters = make([]*pebble.Iterator, o.iterSlots)
+	t.iters = make([]*retryableIter, o.iterSlots)
 	t.snapshots = make([]*pebble.Snapshot, o.snapshotSlots)
 	h.Recordf("%s", o)
 }
@@ -229,14 +232,17 @@ func (o *ingestOp) run(t *test, h *history) {
 		err = firstError(err, b.Close())
 	}
 
-	err = firstError(err, t.db.Ingest(paths))
+	err = firstError(err, withRetries(func() error {
+		return t.db.Ingest(paths)
+	}))
 
 	h.Recordf("%s // %v", o, err)
 }
 
 func (o *ingestOp) build(t *test, h *history, b *pebble.Batch, i int) (string, error) {
-	path := t.opts.FS.PathJoin(t.tmpDir, fmt.Sprintf("ext%d", i))
-	f, err := t.opts.FS.Create(path)
+	rootFS := vfs.Root(t.opts.FS)
+	path := rootFS.PathJoin(t.tmpDir, fmt.Sprintf("ext%d", i))
+	f, err := rootFS.Create(path)
 	if err != nil {
 		return "", err
 	}
@@ -407,7 +413,12 @@ type getOp struct {
 
 func (o *getOp) run(t *test, h *history) {
 	r := t.getReader(o.readerID)
-	val, closer, err := r.Get(o.key)
+	var val []byte
+	var closer io.Closer
+	err := withRetries(func() (err error) {
+		val, closer, err = r.Get(o.key)
+		return err
+	})
 	h.Recordf("%s // [%q] %v", o, val, err)
 	if closer != nil {
 		closer.Close()
@@ -428,12 +439,20 @@ type newIterOp struct {
 
 func (o *newIterOp) run(t *test, h *history) {
 	r := t.getReader(o.readerID)
-	i := r.NewIter(&pebble.IterOptions{
-		LowerBound: o.lower,
-		UpperBound: o.upper,
-	})
+	var i *pebble.Iterator
+	for {
+		i = r.NewIter(&pebble.IterOptions{
+			LowerBound: o.lower,
+			UpperBound: o.upper,
+		})
+		if err := i.Error(); !errors.Is(err, errorfs.ErrInjected) {
+			break
+		}
+		// close this iter and retry NewIter
+		_ = i.Close()
+	}
 	t.setIter(o.iterID, i)
-	h.Recordf("%s", o)
+	h.Recordf("%s // %v", o, i.Error())
 }
 
 func (o *newIterOp) String() string {
diff --git a/internal/metamorphic/retryable.go b/internal/metamorphic/retryable.go
new file mode 100644
index 00000000..ecbd0e20
--- /dev/null
+++ b/internal/metamorphic/retryable.go
@@ -0,0 +1,118 @@
+// Copyright 2020 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package metamorphic
+
+import (
+	"github.com/cockroachdb/errors"
+	"github.com/cockroachdb/pebble"
+	"github.com/cockroachdb/pebble/internal/errorfs"
+)
+
+// withRetries executes fn, retrying it whenever an errorfs.ErrInjected error
+// is returned.  It returns the first nil or non-errorfs.ErrInjected error
+// returned by fn.
+func withRetries(fn func() error) error {
+	for {
+		if err := fn(); !errors.Is(err, errorfs.ErrInjected) {
+			return err
+		}
+	}
+}
+
+// retryableIter holds an iterator and the state necessary to reset it to its
+// state after the last successful operation. This allows us to retry failed
+// iterator operations by running them again on a non-error iterator with the
+// same pre-operation state.
+type retryableIter struct {
+	iter    *pebble.Iterator
+	lastKey []byte
+}
+
+func (i *retryableIter) needRetry() bool {
+	return errors.Is(i.iter.Error(), errorfs.ErrInjected)
+}
+
+func (i *retryableIter) withRetry(fn func()) {
+	for {
+		fn()
+		if !i.needRetry() {
+			break
+		}
+		for i.needRetry() {
+			i.iter.SeekGE(i.lastKey)
+		}
+	}
+
+	i.lastKey = i.lastKey[:0]
+	if i.iter.Valid() {
+		i.lastKey = append(i.lastKey, i.iter.Key()...)
+	}
+}
+
+func (i *retryableIter) Close() error {
+	return i.iter.Close()
+}
+
+func (i *retryableIter) Error() error {
+	return i.iter.Error()
+}
+
+func (i *retryableIter) First() bool {
+	var valid bool
+	i.withRetry(func() { valid = i.iter.First() })
+	return valid
+}
+
+func (i *retryableIter) Key() []byte {
+	return i.iter.Key()
+}
+
+func (i *retryableIter) Last() bool {
+	var valid bool
+	i.withRetry(func() { valid = i.iter.Last() })
+	return valid
+}
+
+func (i *retryableIter) Next() bool {
+	var valid bool
+	i.withRetry(func() { valid = i.iter.Next() })
+	return valid
+}
+
+func (i *retryableIter) Prev() bool {
+	var valid bool
+	i.withRetry(func() { valid = i.iter.Prev() })
+	return valid
+}
+
+func (i *retryableIter) SeekGE(key []byte) bool {
+	var valid bool
+	i.withRetry(func() { valid = i.iter.SeekGE(key) })
+	return valid
+}
+
+func (i *retryableIter) SeekLT(key []byte) bool {
+	var valid bool
+	i.withRetry(func() { valid = i.iter.SeekLT(key) })
+	return valid
+}
+
+func (i *retryableIter) SeekPrefixGE(key []byte) bool {
+	var valid bool
+	i.withRetry(func() { valid = i.iter.SeekPrefixGE(key) })
+	return valid
+}
+
+func (i *retryableIter) SetBounds(lower, upper []byte) {
+	i.iter.SetBounds(lower, upper)
+}
+
+func (i *retryableIter) Valid() bool {
+	return i.iter.Valid()
+}
+
+func (i *retryableIter) Value() []byte {
+	return i.iter.Value()
+}
diff --git a/vfs/vfs.go b/vfs/vfs.go
index 52856391..a1c0b569 100644
--- a/vfs/vfs.go
+++ b/vfs/vfs.go
@@ -263,3 +263,20 @@ func LinkOrCopy(fs FS, oldname, newname string) error {
 	}
 	return Copy(fs, oldname, newname)
 }
+
+// Root returns the base FS implementation, unwrapping all nested FSs that
+// expose an Unwrap method.
+func Root(fs FS) FS {
+	type unwrapper interface {
+		Unwrap() FS
+	}
+
+	for {
+		u, ok := fs.(unwrapper)
+		if !ok {
+			break
+		}
+		fs = u.Unwrap()
+	}
+	return fs
+}


Test patch:
diff --git a/internal/metamorphic/meta_test.go b/internal/metamorphic/meta_test.go
index 34e78897..2b2d64f0 100644
--- a/internal/metamorphic/meta_test.go
+++ b/internal/metamorphic/meta_test.go
@@ -19,6 +19,7 @@ import (
 
 	"github.com/cockroachdb/pebble"
 	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/errorfs"
 	"github.com/cockroachdb/pebble/internal/randvar"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/pmezard/go-difflib/difflib"
@@ -41,6 +42,9 @@ var (
 		"the directory storing test state")
 	disk = flag.Bool("disk", false,
 		"whether to use an in-mem DB or on-disk (in-mem is significantly faster)")
+	// TODO: default error rate to a non-zero value
+	errorRate = flag.Float64("error-rate", 0.0,
+		"rate of errors injected into filesystem operations (0 ≤ r < 1)")
 	failRE = flag.String("fail", "",
 		"fail the test if the supplied regular expression matches the output")
 	traceFile = flag.String("trace-file", "",
@@ -91,6 +95,10 @@ func testMetaRun(t *testing.T, runDir string) {
 			opts.FS = vfs.NewMem()
 		}
 	}
+	// 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))
+
 	if opts.WALDir != "" {
 		opts.WALDir = opts.FS.PathJoin(runDir, opts.WALDir)
 	}
diff --git a/internal/metamorphic/test.go b/internal/metamorphic/test.go
index 4adfcb99..26d2769b 100644
--- a/internal/metamorphic/test.go
+++ b/internal/metamorphic/test.go
@@ -10,7 +10,9 @@ import (
 	"os"
 	"strings"
 
+	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble"
+	"github.com/cockroachdb/pebble/internal/errorfs"
 	"github.com/cockroachdb/pebble/vfs"
 )
 
@@ -29,7 +31,7 @@ type test struct {
 	// The slots for the batches, iterators, and snapshots. These are read and
 	// written by the ops to pass state from one op to another.
 	batches   []*pebble.Batch
-	iters     []*pebble.Iterator
+	iters     []*retryableIter
 	snapshots []*pebble.Snapshot
 }
 
@@ -49,7 +51,13 @@ func (t *test) init(h *history, dir string, testOpts *testOptions) error {
 	t.opts = testOpts.opts.EnsureDefaults()
 	t.opts.Logger = h.Logger()
 	t.opts.EventListener = pebble.MakeLoggingEventListener(t.opts.Logger)
-	t.opts.DebugCheck = pebble.DebugCheckLevels
+	t.opts.DebugCheck = func(db *pebble.DB) error {
+		// Wrap the ordinary DebugCheckLevels with retrying
+		// of injected errors.
+		return withRetries(func() error {
+			return pebble.DebugCheckLevels(db)
+		})
+	}
 
 	defer t.opts.Cache.Unref()
 
@@ -59,7 +67,7 @@ func (t *test) init(h *history, dir string, testOpts *testOptions) error {
 	// difference between in-memory and on-disk which causes different code paths
 	// and timings to be exercised.
 	maybeExit := func(err error) {
-		if err == nil {
+		if err == nil || errors.Is(err, errorfs.ErrInjected) {
 			return
 		}
 		t.maybeSaveData()
@@ -106,7 +114,12 @@ func (t *test) init(h *history, dir string, testOpts *testOptions) error {
 		maybeExit(info.Err)
 	}
 
-	db, err := pebble.Open(dir, t.opts)
+	var db *pebble.DB
+	var err error
+	err = withRetries(func() error {
+		db, err = pebble.Open(dir, t.opts)
+		return err
+	})
 	if err != nil {
 		return err
 	}
@@ -153,26 +166,29 @@ func (t *test) restartDB() error {
 		return nil
 	}
 	t.opts.Cache.Ref()
-	fs := t.opts.FS.(*vfs.MemFS)
+	fs := vfs.Root(t.opts.FS).(*vfs.MemFS)
 	fs.SetIgnoreSyncs(true)
 	if err := t.db.Close(); err != nil {
 		return err
 	}
 	fs.ResetToSyncedState()
 	fs.SetIgnoreSyncs(false)
-	var err error
-	t.db, err = pebble.Open(t.dir, t.opts)
+	err := withRetries(func() (err error) {
+		t.db, err = pebble.Open(t.dir, t.opts)
+		return err
+	})
 	t.opts.Cache.Unref()
 	return err
 }
 
 // If an in-memory FS is being used, save the contents to disk.
 func (t *test) maybeSaveData() {
-	if t.opts.FS == vfs.Default {
+	rootFS := vfs.Root(t.opts.FS)
+	if rootFS == vfs.Default {
 		return
 	}
 	_ = os.RemoveAll(t.dir)
-	if _, err := vfs.Clone(t.opts.FS, vfs.Default, t.dir, t.dir); err != nil {
+	if _, err := vfs.Clone(rootFS, vfs.Default, t.dir, t.dir); err != nil {
 		t.opts.Logger.Infof("unable to clone: %s: %v", t.dir, err)
 	}
 }
@@ -197,7 +213,7 @@ func (t *test) setIter(id objID, i *pebble.Iterator) {
 	if id.tag() != iterTag {
 		panic(fmt.Sprintf("invalid iter ID: %s", id))
 	}
-	t.iters[id.slot()] = i
+	t.iters[id.slot()] = &retryableIter{iter: i, lastKey: nil}
 }
 
 func (t *test) setSnapshot(id objID, s *pebble.Snapshot) {
@@ -241,7 +257,7 @@ func (t *test) getCloser(id objID) io.Closer {
 	panic(fmt.Sprintf("cannot close ID: %s", id))
 }
 
-func (t *test) getIter(id objID) *pebble.Iterator {
+func (t *test) getIter(id objID) *retryableIter {
 	if id.tag() != iterTag {
 		panic(fmt.Sprintf("invalid iter ID: %s", id))
 	}