vfs/errorfs: add facilities for error injection in datadriven tests

We frequently use the errorfs package to inject errors into filesystem
operations to test error code paths. This commit builds off this package to
support encoding error injection conditions within datadriven tests through
parsing a small lisp-like DSL. Future work will build off this to test handling
of I/O errors during iteration.

Informs cockroachdb#1115.
Informs cockroachdb#2994.

compaction_test.go

@@ -2900,7 +2900,7 @@ func TestCompactionErrorCleanup(t *testing.T) {
 	mem := vfs.NewMem()
 	ii := errorfs.OnIndex(math.MaxInt32) // start disabled
 	opts := (&Options{
-		FS:     errorfs.Wrap(mem, ii),
+		FS:     errorfs.Wrap(mem, errorfs.ErrInjected.If(ii)),
 		Levels: make([]LevelOptions, numLevels),
 		EventListener: &EventListener{
 			TableCreated: func(info TableCreateInfo) {
@@ -3286,8 +3286,8 @@ func TestFlushError(t *testing.T) {
 	// Error the first five times we try to write a sstable.
 	var errorOps atomic.Int32
 	errorOps.Store(3)
-	fs := errorfs.Wrap(vfs.NewMem(), errorfs.InjectorFunc(func(op errorfs.Op, path string) error {
-		if op == errorfs.OpCreate && filepath.Ext(path) == ".sst" && errorOps.Add(-1) >= 0 {
+	fs := errorfs.Wrap(vfs.NewMem(), errorfs.InjectorFunc(func(op errorfs.Op) error {
+		if op.Kind == errorfs.OpCreate && filepath.Ext(op.Path) == ".sst" && errorOps.Add(-1) >= 0 {
 			return errorfs.ErrInjected
 		}
 		return nil
@@ -3704,19 +3704,24 @@ type createManifestErrorInjector struct {
 	enabled atomic.Bool
 }
 
+// TODO(jackson): Replace the createManifestErrorInjector with the composition
+// of primitives defined in errorfs. This may require additional primitives.
+
+func (i *createManifestErrorInjector) String() string { return "MANIFEST-Creates" }
+
 // enable enables error injection for the vfs.FS.
 func (i *createManifestErrorInjector) enable() {
 	i.enabled.Store(true)
 }
 
 // MaybeError implements errorfs.Injector.
-func (i *createManifestErrorInjector) MaybeError(op errorfs.Op, path string) error {
+func (i *createManifestErrorInjector) MaybeError(op errorfs.Op) error {
 	if !i.enabled.Load() {
 		return nil
 	}
 	// This necessitates having a MaxManifestSize of 1, to reliably induce
 	// logAndApply errors.
-	if strings.Contains(path, "MANIFEST") && op == errorfs.OpCreate {
+	if strings.Contains(op.Path, "MANIFEST") && op.Kind == errorfs.OpCreate {
 		return errorfs.ErrInjected
 	}
 	return nil
@@ -3884,6 +3889,11 @@ type WriteErrorInjector struct {
 	enabled atomic.Bool
 }
 
+// TODO(jackson): Replace WriteErrorInjector with use of primitives in errorfs,
+// adding new primitives as necessary.
+
+func (i *WriteErrorInjector) String() string { return "FileWrites(ErrInjected)" }
+
 // enable enables error injection for the vfs.FS.
 func (i *WriteErrorInjector) enable() {
 	i.enabled.Store(true)
@@ -3895,12 +3905,12 @@ func (i *WriteErrorInjector) disable() {
 }
 
 // MaybeError implements errorfs.Injector.
-func (i *WriteErrorInjector) MaybeError(op errorfs.Op, path string) error {
+func (i *WriteErrorInjector) MaybeError(op errorfs.Op) error {
 	if !i.enabled.Load() {
 		return nil
 	}
 	// Fail any future write.
-	if op == errorfs.OpFileWrite {
+	if op.Kind == errorfs.OpFileWrite {
 		return errorfs.ErrInjected
 	}
 	return nil

data_test.go

@@ -18,6 +18,7 @@ import (
 
 	"github.com/cockroachdb/datadriven"
 	"github.com/cockroachdb/errors"
+	"github.com/cockroachdb/pebble/bloom"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/humanize"
 	"github.com/cockroachdb/pebble/internal/keyspan"
@@ -29,6 +30,7 @@ import (
 	"github.com/cockroachdb/pebble/objstorage/remote"
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/vfs"
+	"github.com/cockroachdb/pebble/vfs/errorfs"
 	"github.com/stretchr/testify/require"
 )
 
@@ -1447,3 +1449,75 @@ func runLSMCmd(td *datadriven.TestData, d *DB) string {
 	}
 	return d.mu.versions.currentVersion().String()
 }
+
+func parseDBOptionsArgs(opts *Options, args []datadriven.CmdArg) error {
+	for _, cmdArg := range args {
+		switch cmdArg.Key {
+		case "inject-errors":
+			injs := make([]errorfs.Injector, len(cmdArg.Vals))
+			for i := 0; i < len(cmdArg.Vals); i++ {
+				inj, err := errorfs.ParseInjectorFromDSL(cmdArg.Vals[i])
+				if err != nil {
+					return err
+				}
+				injs[i] = inj
+			}
+			opts.FS = errorfs.Wrap(opts.FS, errorfs.Any(injs...))
+		case "enable-table-stats":
+			enable, err := strconv.ParseBool(cmdArg.Vals[0])
+			if err != nil {
+				return errors.Errorf("%s: could not parse %q as bool: %s", cmdArg.Key, cmdArg.Vals[0], err)
+			}
+			opts.private.disableTableStats = !enable
+		case "format-major-version":
+			v, err := strconv.Atoi(cmdArg.Vals[0])
+			if err != nil {
+				return err
+			}
+			// Override the DB version.
+			opts.FormatMajorVersion = FormatMajorVersion(v)
+		case "block-size":
+			v, err := strconv.Atoi(cmdArg.Vals[0])
+			if err != nil {
+				return err
+			}
+			for i := range opts.Levels {
+				opts.Levels[i].BlockSize = v
+			}
+		case "index-block-size":
+			v, err := strconv.Atoi(cmdArg.Vals[0])
+			if err != nil {
+				return err
+			}
+			for i := range opts.Levels {
+				opts.Levels[i].IndexBlockSize = v
+			}
+		case "target-file-size":
+			v, err := strconv.Atoi(cmdArg.Vals[0])
+			if err != nil {
+				return err
+			}
+			for i := range opts.Levels {
+				opts.Levels[i].TargetFileSize = int64(v)
+			}
+		case "bloom-bits-per-key":
+			v, err := strconv.Atoi(cmdArg.Vals[0])
+			if err != nil {
+				return err
+			}
+			fp := bloom.FilterPolicy(v)
+			opts.Filters = map[string]FilterPolicy{fp.Name(): fp}
+			for i := range opts.Levels {
+				opts.Levels[i].FilterPolicy = fp
+			}
+		case "merger":
+			switch cmdArg.Vals[0] {
+			case "appender":
+				opts.Merger = base.DefaultMerger
+			default:
+				return errors.Newf("unrecognized Merger %q\n", cmdArg.Vals[0])
+			}
+		}
+	}
+	return nil
+}

error_test.go

@@ -136,7 +136,7 @@ func TestErrors(t *testing.T) {
 
 	errorCounts := make(map[string]int)
 	for i := int32(0); ; i++ {
-		fs := errorfs.Wrap(vfs.NewMem(), errorfs.OnIndex(i))
+		fs := errorfs.Wrap(vfs.NewMem(), errorfs.ErrInjected.If(errorfs.OnIndex(i)))
 		err := run(fs)
 		if err == nil {
 			t.Logf("success %d\n", i)
@@ -166,8 +166,8 @@ func TestErrors(t *testing.T) {
 func TestRequireReadError(t *testing.T) {
 	run := func(formatVersion FormatMajorVersion, index int32) (err error) {
 		// Perform setup with error injection disabled as it involves writes/background ops.
-		inj := errorfs.OnIndex(-1)
-		fs := errorfs.Wrap(vfs.NewMem(), inj)
+		ii := errorfs.OnIndex(-1)
+		fs := errorfs.Wrap(vfs.NewMem(), errorfs.ErrInjected.If(ii))
 		opts := &Options{
 			FS:                 fs,
 			Logger:             panicLogger{},
@@ -210,7 +210,7 @@ func TestRequireReadError(t *testing.T) {
 		}
 
 		// Now perform foreground ops with error injection enabled.
-		inj.SetIndex(index)
+		ii.SetIndex(index)
 		iter, _ := d.NewIter(nil)
 		if err := iter.Error(); err != nil {
 			return err
@@ -237,7 +237,7 @@ func TestRequireReadError(t *testing.T) {
 		// Reaching here implies all read operations succeeded. This
 		// should only happen when we reached a large enough index at
 		// which `errorfs.FS` did not return any error.
-		if i := inj.Index(); i < 0 {
+		if i := ii.Index(); i < 0 {
 			t.Errorf("FS error injected %d ops ago went unreported", -i)
 		}
 		if numFound != 2 {
@@ -367,8 +367,8 @@ func TestDBWALRotationCrash(t *testing.T) {
 	memfs := vfs.NewStrictMem()
 
 	var index atomic.Int32
-	inj := errorfs.InjectorFunc(func(op errorfs.Op, _ string) error {
-		if op.OpKind() == errorfs.OpKindWrite && index.Add(-1) == -1 {
+	inj := errorfs.InjectorFunc(func(op errorfs.Op) error {
+		if op.Kind.ReadOrWrite() == errorfs.OpIsWrite && index.Add(-1) == -1 {
 			memfs.SetIgnoreSyncs(true)
 		}
 		return nil

ingest_test.go

@@ -386,7 +386,7 @@ func TestIngestLinkFallback(t *testing.T) {
 	src, err := mem.Create("source")
 	require.NoError(t, err)
 
-	opts := &Options{FS: errorfs.Wrap(mem, errorfs.OnIndex(1))}
+	opts := &Options{FS: errorfs.Wrap(mem, errorfs.ErrInjected.If(errorfs.OnIndex(1)))}
 	opts.EnsureDefaults().WithFSDefaults()
 	objSettings := objstorageprovider.DefaultSettings(opts.FS, "")
 	// Prevent the provider from listing the dir (where we may get an injected error).
@@ -2182,9 +2182,9 @@ func TestIngestError(t *testing.T) {
 		require.NoError(t, w.Set([]byte("d"), nil))
 		require.NoError(t, w.Close())
 
-		inj := errorfs.OnIndex(-1)
+		ii := errorfs.OnIndex(-1)
 		d, err := Open("", &Options{
-			FS:                    errorfs.Wrap(mem, inj),
+			FS:                    errorfs.Wrap(mem, errorfs.ErrInjected.If(ii)),
 			Logger:                panicLogger{},
 			L0CompactionThreshold: 8,
 		})
@@ -2211,7 +2211,7 @@ func TestIngestError(t *testing.T) {
 				}
 			}()
 
-			inj.SetIndex(i)
+			ii.SetIndex(i)
 			err1 := d.Ingest([]string{"ext0"})
 			err2 := d.Ingest([]string{"ext1"})
 			err := firstError(err1, err2)
@@ -2225,7 +2225,7 @@ func TestIngestError(t *testing.T) {
 
 		// If the injector's index is non-negative, the i-th filesystem
 		// operation was never executed.
-		if inj.Index() >= 0 {
+		if ii.Index() >= 0 {
 			break
 		}
 	}
@@ -3225,11 +3225,11 @@ func TestIngestValidation(t *testing.T) {
 			cLoc:        corruptionLocationNone,
 			wantErr:     errorfs.ErrInjected,
 			wantErrType: errReportLocationBackgroundError,
-			errorfsInjector: errorfs.InjectorFunc(func(op errorfs.Op, path string) error {
+			errorfsInjector: errorfs.InjectorFunc(func(op errorfs.Op) error {
 				// Inject an error on the first read-at operation on an sstable
 				// (excluding the read on the sstable before ingestion has
 				// linked it in).
-				if path != "ext" && op != errorfs.OpFileReadAt || filepath.Ext(path) != ".sst" {
+				if op.Path != "ext" && op.Kind != errorfs.OpFileReadAt || filepath.Ext(op.Path) != ".sst" {
 					return nil
 				}
 				if errfsCounter.Add(1) == 1 {

iterator_histories_test.go

@@ -13,8 +13,6 @@ import (
 
 	"github.com/cockroachdb/datadriven"
 	"github.com/cockroachdb/errors"
-	"github.com/cockroachdb/pebble/bloom"
-	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/sstable"
@@ -41,70 +39,22 @@ func TestIterHistories(t *testing.T) {
 			iters[name] = it
 			return it
 		}
+		var opts *Options
 		parseOpts := func(td *datadriven.TestData) (*Options, error) {
-			opts := &Options{
+			opts = &Options{
 				FS:                 vfs.NewMem(),
 				Comparer:           testkeys.Comparer,
 				FormatMajorVersion: FormatRangeKeys,
 				BlockPropertyCollectors: []func() BlockPropertyCollector{
 					sstable.NewTestKeysBlockPropertyCollector,
 				},
 			}
+
 			opts.DisableAutomaticCompactions = true
 			opts.EnsureDefaults()
 			opts.WithFSDefaults()
-
-			for _, cmdArg := range td.CmdArgs {
-				switch cmdArg.Key {
-				case "format-major-version":
-					v, err := strconv.Atoi(cmdArg.Vals[0])
-					if err != nil {
-						return nil, err
-					}
-					// Override the DB version.
-					opts.FormatMajorVersion = FormatMajorVersion(v)
-				case "block-size":
-					v, err := strconv.Atoi(cmdArg.Vals[0])
-					if err != nil {
-						return nil, err
-					}
-					for i := range opts.Levels {
-						opts.Levels[i].BlockSize = v
-					}
-				case "index-block-size":
-					v, err := strconv.Atoi(cmdArg.Vals[0])
-					if err != nil {
-						return nil, err
-					}
-					for i := range opts.Levels {
-						opts.Levels[i].IndexBlockSize = v
-					}
-				case "target-file-size":
-					v, err := strconv.Atoi(cmdArg.Vals[0])
-					if err != nil {
-						return nil, err
-					}
-					for i := range opts.Levels {
-						opts.Levels[i].TargetFileSize = int64(v)
-					}
-				case "bloom-bits-per-key":
-					v, err := strconv.Atoi(cmdArg.Vals[0])
-					if err != nil {
-						return nil, err
-					}
-					fp := bloom.FilterPolicy(v)
-					opts.Filters = map[string]FilterPolicy{fp.Name(): fp}
-					for i := range opts.Levels {
-						opts.Levels[i].FilterPolicy = fp
-					}
-				case "merger":
-					switch cmdArg.Vals[0] {
-					case "appender":
-						opts.Merger = base.DefaultMerger
-					default:
-						return nil, errors.Newf("unrecognized Merger %q\n", cmdArg.Vals[0])
-					}
-				}
+			if err := parseDBOptionsArgs(opts, td.CmdArgs); err != nil {
+				return nil, err
 			}
 			return opts, nil
 		}
@@ -128,10 +78,11 @@ func TestIterHistories(t *testing.T) {
 		datadriven.RunTest(t, path, func(t *testing.T, td *datadriven.TestData) string {
 			switch td.Cmd {
 			case "define":
+				var err error
 				if err := cleanup(); err != nil {
 					return err.Error()
 				}
-				opts, err := parseOpts(td)
+				opts, err = parseOpts(td)
 				if err != nil {
 					return err.Error()
 				}
@@ -140,12 +91,23 @@ func TestIterHistories(t *testing.T) {
 					return err.Error()
 				}
 				return runLSMCmd(td, d)
-
+			case "reopen":
+				var err error
+				if err := cleanup(); err != nil {
+					return err.Error()
+				}
+				if err := parseDBOptionsArgs(opts, td.CmdArgs); err != nil {
+					return err.Error()
+				}
+				d, err = Open("", opts)
+				require.NoError(t, err)
+				return ""
 			case "reset":
+				var err error
 				if err := cleanup(); err != nil {
 					return err.Error()
 				}
-				opts, err := parseOpts(td)
+				opts, err = parseOpts(td)
 				if err != nil {
 					return err.Error()
 				}