Repository: cockroachdb/pebble
Pull request #3742 (2024-07-10 16:43:03)

metamorphic: limit aggregate injected I/O latency
Limit the amount of I/O latency injected into any individual run of the metamorphic test to improve worst-case test runtimes.

Informs #3681.
Informs #3733.

cockroach-teamcity: 

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


Patch:
diff --git a/metamorphic/meta.go b/metamorphic/meta.go
index 649e7a4e..30d17b40 100644
--- a/metamorphic/meta.go
+++ b/metamorphic/meta.go
@@ -492,13 +492,16 @@ func RunOnce(t TestingT, runDir string, seed uint64, historyPath string, rOpts .
 	} else {
 		opts.Cleaner = base.ArchiveCleaner{}
 	}
-	// Wrap the filesystem with a VFS that will inject random latency if
-	// the test options require it.
+	// Wrap the filesystem with a VFS that will inject random latency if the
+	// test options require it. We cap the overlal injected latency to ten
+	// seconds to avoid excessive test run times when paired with small target
+	// file sizes, block sizes, etc.
 	if testOpts.ioLatencyProbability > 0 {
 		opts.FS = errorfs.Wrap(opts.FS, errorfs.RandomLatency(
 			errorfs.Randomly(testOpts.ioLatencyProbability, testOpts.ioLatencySeed),
 			testOpts.ioLatencyMean,
 			testOpts.ioLatencySeed,
+			10*time.Second,
 		))
 	}
 
diff --git a/sstable/reader.go b/sstable/reader.go
index 51c24f4e..4acbd7f4 100644
--- a/sstable/reader.go
+++ b/sstable/reader.go
@@ -221,23 +221,22 @@ func (r *Reader) newIterWithBlockPropertyFiltersAndContext(
 	// NB: pebble.tableCache wraps the returned iterator with one which performs
 	// reference counting on the Reader, preventing the Reader from being closed
 	// until the final iterator closes.
+	var res Iterator
+	var err error
 	if r.Properties.IndexType == twoLevelIndex {
-		i := twoLevelIterPool.Get().(*twoLevelIterator)
-		err := i.init(ctx, r, vState, transforms, lower, upper, filterer, useFilterBlock,
+		res, err = newTwoLevelIterator(ctx, r, vState, transforms, lower, upper, filterer, useFilterBlock,
+			stats, categoryAndQoS, statsCollector, rp, nil /* bufferPool */)
+	} else {
+		res, err = newSingleLevelIterator(
+			ctx, r, vState, transforms, lower, upper, filterer, useFilterBlock,
 			stats, categoryAndQoS, statsCollector, rp, nil /* bufferPool */)
-		if err != nil {
-			return nil, err
-		}
-		return i, nil
 	}
-
-	i := singleLevelIterPool.Get().(*singleLevelIterator)
-	err := i.init(ctx, r, vState, transforms, lower, upper, filterer, useFilterBlock,
-		stats, categoryAndQoS, statsCollector, rp, nil /* bufferPool */)
 	if err != nil {
+		// Note: we don't want to return res here - it will be a nil
+		// single/twoLevelIterator, not a nil Iterator.
 		return nil, err
 	}
-	return i, nil
+	return res, nil
 }
 
 // NewIter returns an iterator for the contents of the table. If an error
@@ -275,8 +274,7 @@ func (r *Reader) newCompactionIter(
 		transforms.HideObsoletePoints = true
 	}
 	if r.Properties.IndexType == twoLevelIndex {
-		i := twoLevelIterPool.Get().(*twoLevelIterator)
-		err := i.init(
+		i, err := newTwoLevelIterator(
 			context.Background(),
 			r, vState, transforms, nil /* lower */, nil /* upper */, nil,
 			false /* useFilter */, nil /* stats */, categoryAndQoS, statsCollector, rp, bufferPool,
@@ -287,8 +285,7 @@ func (r *Reader) newCompactionIter(
 		i.setupForCompaction()
 		return &twoLevelCompactionIterator{twoLevelIterator: i}, nil
 	}
-	i := singleLevelIterPool.Get().(*singleLevelIterator)
-	err := i.init(
+	i, err := newSingleLevelIterator(
 		context.Background(), r, vState, transforms, nil /* lower */, nil, /* upper */
 		nil, false /* useFilter */, nil /* stats */, categoryAndQoS, statsCollector, rp, bufferPool,
 	)
diff --git a/sstable/reader_iter_single_lvl.go b/sstable/reader_iter_single_lvl.go
index 22f8c066..c0efbb61 100644
--- a/sstable/reader_iter_single_lvl.go
+++ b/sstable/reader_iter_single_lvl.go
@@ -174,14 +174,13 @@ type singleLevelIterator struct {
 // singleLevelIterator implements the base.InternalIterator interface.
 var _ base.InternalIterator = (*singleLevelIterator)(nil)
 
-// init initializes a singleLevelIterator for reading from the table. It is
-// synonmous with Reader.NewIter, but allows for reusing of the iterator
-// between different Readers.
+// newSingleLevelIterator reads the index block and creates and initializes a
+// singleLevelIterator.
 //
-// Note that lower, upper passed into init has nothing to do with virtual sstable
-// bounds. If the virtualState passed in is not nil, then virtual sstable bounds
-// will be enforced.
-func (i *singleLevelIterator) init(
+// Note that lower, upper are iterator bounds and are separate from virtual
+// sstable bounds. If the virtualState passed in is not nil, then virtual
+// sstable bounds will be enforced.
+func newSingleLevelIterator(
 	ctx context.Context,
 	r *Reader,
 	v *virtualState,
@@ -194,22 +193,41 @@ func (i *singleLevelIterator) init(
 	statsCollector *CategoryStatsCollector,
 	rp ReaderProvider,
 	bufferPool *block.BufferPool,
-) error {
+) (*singleLevelIterator, error) {
 	if r.err != nil {
-		return r.err
+		return nil, r.err
 	}
-	i.iterStats.init(categoryAndQoS, statsCollector)
-	i.indexFilterRH = objstorageprovider.UsePreallocatedReadHandle(
-		ctx, r.readable, objstorage.ReadBeforeForIndexAndFilter, &i.indexFilterRHPrealloc)
+	i := singleLevelIterPool.Get().(*singleLevelIterator)
+	i.init(
+		ctx, r, v, transforms, lower, upper, filterer, useFilter,
+		stats, categoryAndQoS, statsCollector, rp, bufferPool,
+	)
 	indexH, err := r.readIndex(ctx, i.indexFilterRH, stats, &i.iterStats)
-	if err != nil {
-		return err
+	if err == nil {
+		err = i.index.InitHandle(i.cmp, r.Split, indexH, transforms)
 	}
-	if v != nil {
-		i.vState = v
-		i.endKeyInclusive, lower, upper = v.constrainBounds(lower, upper, false /* endInclusive */)
+	if err != nil {
+		_ = i.Close()
+		return nil, err
 	}
+	return i, nil
+}
 
+// init initializes the singleLevelIterator struct. It does not read the index.
+func (i *singleLevelIterator) init(
+	ctx context.Context,
+	r *Reader,
+	v *virtualState,
+	transforms IterTransforms,
+	lower, upper []byte,
+	filterer *BlockPropertiesFilterer,
+	useFilter bool,
+	stats *base.InternalIteratorStats,
+	categoryAndQoS CategoryAndQoS,
+	statsCollector *CategoryStatsCollector,
+	rp ReaderProvider,
+	bufferPool *block.BufferPool,
+) {
 	i.inPool = false
 	i.ctx = ctx
 	i.lower = lower
@@ -221,13 +239,18 @@ func (i *singleLevelIterator) init(
 	i.stats = stats
 	i.transforms = transforms
 	i.bufferPool = bufferPool
-	err = i.index.InitHandle(i.cmp, r.Split, indexH, transforms)
-	if err != nil {
-		// blockIter.Close releases indexH and always returns a nil error
-		_ = i.index.Close()
-		return err
+	if v != nil {
+		i.vState = v
+		i.endKeyInclusive, i.lower, i.upper = v.constrainBounds(lower, upper, false /* endInclusive */)
 	}
-	i.dataRH = objstorageprovider.UsePreallocatedReadHandle(ctx, r.readable, objstorage.NoReadBefore, &i.dataRHPrealloc)
+
+	i.iterStats.init(categoryAndQoS, statsCollector)
+
+	i.indexFilterRH = objstorageprovider.UsePreallocatedReadHandle(
+		ctx, r.readable, objstorage.ReadBeforeForIndexAndFilter, &i.indexFilterRHPrealloc)
+	i.dataRH = objstorageprovider.UsePreallocatedReadHandle(
+		ctx, r.readable, objstorage.NoReadBefore, &i.dataRHPrealloc)
+
 	if r.tableFormat >= TableFormatPebblev3 {
 		if r.Properties.NumValueBlocks > 0 {
 			// NB: we cannot avoid this ~248 byte allocation, since valueBlockReader
@@ -250,7 +273,6 @@ func (i *singleLevelIterator) init(
 		}
 		i.data.SetHasValuePrefix(true)
 	}
-	return nil
 }
 
 // Helper function to check if keys returned from iterator are within virtual bounds.
diff --git a/sstable/reader_iter_two_lvl.go b/sstable/reader_iter_two_lvl.go
index 5818b1ec..cd2c9551 100644
--- a/sstable/reader_iter_two_lvl.go
+++ b/sstable/reader_iter_two_lvl.go
@@ -12,8 +12,6 @@ import (
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/treeprinter"
-	"github.com/cockroachdb/pebble/objstorage"
-	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider/objiotracing"
 	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/sstable/rowblk"
@@ -42,7 +40,7 @@ func (i *twoLevelIterator) loadIndex(dir int8) loadBlockResult {
 	v := i.topLevelIndex.Value()
 	bhp, err := decodeBlockHandleWithProperties(v.InPlaceValue())
 	if err != nil {
-		i.err = base.CorruptionErrorf("pebble/table: corrupt top level index entry")
+		i.err = base.CorruptionErrorf("pebble/table: corrupt top level index entry (%v)", err)
 		return loadBlockFailed
 	}
 	if i.bpfs != nil {
@@ -62,20 +60,20 @@ func (i *twoLevelIterator) loadIndex(dir int8) loadBlockResult {
 	ctx := objiotracing.WithBlockType(i.ctx, objiotracing.MetadataBlock)
 	indexBlock, err := i.reader.readBlock(
 		ctx, bhp.Handle, nil /* transform */, i.indexFilterRH, i.stats, &i.iterStats, i.bufferPool)
+	if err == nil {
+		err = i.index.InitHandle(i.cmp, i.reader.Split, indexBlock, i.transforms)
+	}
 	if err != nil {
 		i.err = err
 		return loadBlockFailed
 	}
-	if i.err = i.index.InitHandle(i.cmp, i.reader.Split, indexBlock, i.transforms); i.err == nil {
-		return loadBlockOK
-	}
-	return loadBlockFailed
+	return loadBlockOK
 }
 
 // resolveMaybeExcluded is invoked when the block-property filterer has found
 // that an index block is excluded according to its properties but only if its
 // bounds fall within the filter's current bounds. This function consults the
-// apprioriate bound, depending on the iteration direction, and returns either
+// appropriate bound, depending on the iteration direction, and returns either
 // `blockIntersects` or `blockExcluded`.
 func (i *twoLevelIterator) resolveMaybeExcluded(dir int8) intersectsResult {
 	// This iterator is configured with a bound-limited block property filter.
@@ -134,10 +132,13 @@ func (i *twoLevelIterator) resolveMaybeExcluded(dir int8) intersectsResult {
 	return blockIntersects
 }
 
-// Note that lower, upper passed into init has nothing to do with virtual sstable
-// bounds. If the virtualState passed in is not nil, then virtual sstable bounds
-// will be enforced.
-func (i *twoLevelIterator) init(
+// newTwoLevelIterator reads the top-level index block and creates and
+// initializes a two level iterator.
+//
+// Note that lower, upper are iterator bounds and are separate from virtual
+// sstable bounds. If the virtualState passed in is not nil, then virtual
+// sstable bounds will be enforced.
+func newTwoLevelIterator(
 	ctx context.Context,
 	r *Reader,
 	v *virtualState,
@@ -150,57 +151,23 @@ func (i *twoLevelIterator) init(
 	statsCollector *CategoryStatsCollector,
 	rp ReaderProvider,
 	bufferPool *block.BufferPool,
-) error {
+) (*twoLevelIterator, error) {
 	if r.err != nil {
-		return r.err
+		return nil, r.err
 	}
-	i.iterStats.init(categoryAndQoS, statsCollector)
-	i.indexFilterRH = objstorageprovider.UsePreallocatedReadHandle(
-		ctx, r.readable, objstorage.ReadBeforeForIndexAndFilter, &i.indexFilterRHPrealloc)
+	i := twoLevelIterPool.Get().(*twoLevelIterator)
+	i.singleLevelIterator.init(ctx, r, v, transforms, lower, upper, filterer, useFilter,
+		stats, categoryAndQoS, statsCollector, rp, bufferPool)
+
 	topLevelIndexH, err := r.readIndex(ctx, i.indexFilterRH, stats, &i.iterStats)
-	if err != nil {
-		return err
-	}
-	if v != nil {
-		i.vState = v
-		// Note that upper is exclusive here.
-		i.endKeyInclusive, lower, upper = v.constrainBounds(lower, upper, false /* endInclusive */)
+	if err == nil {
+		err = i.topLevelIndex.InitHandle(i.cmp, i.reader.Split, topLevelIndexH, transforms)
 	}
-
-	i.inPool = false
-	i.ctx = ctx
-	i.lower = lower
-	i.upper = upper
-	i.bpfs = filterer
-	i.useFilter = useFilter
-	i.reader = r
-	i.cmp = r.Compare
-	i.stats = stats
-	i.transforms = transforms
-	i.bufferPool = bufferPool
-	err = i.topLevelIndex.InitHandle(i.cmp, i.reader.Split, topLevelIndexH, transforms)
 	if err != nil {
-		// blockIter.Close releases topLevelIndexH and always returns a nil error
-		_ = i.topLevelIndex.Close()
-		return err
-	}
-	i.dataRH = objstorageprovider.UsePreallocatedReadHandle(
-		ctx, r.readable, objstorage.NoReadBefore, &i.dataRHPrealloc)
-	if r.tableFormat >= TableFormatPebblev3 {
-		if r.Properties.NumValueBlocks > 0 {
-			i.vbReader = &valueBlockReader{
-				bpOpen: i,
-				rp:     rp,
-				vbih:   r.valueBIH,
-				stats:  stats,
-			}
-			i.data.SetGetLazyValue(i.vbReader.getLazyValueForPrefixAndValueHandle)
-			i.vbRH = objstorageprovider.UsePreallocatedReadHandle(
-				ctx, r.readable, objstorage.NoReadBefore, &i.vbRHPrealloc)
-		}
-		i.data.SetHasValuePrefix(true)
+		_ = i.Close()
+		return nil, err
 	}
-	return nil
+	return i, nil
 }
 
 func (i *twoLevelIterator) String() string {
diff --git a/vfs/errorfs/latency.go b/vfs/errorfs/latency.go
index 1e5a9dda..de033730 100644
--- a/vfs/errorfs/latency.go
+++ b/vfs/errorfs/latency.go
@@ -8,6 +8,7 @@ import (
 	"math/rand"
 	"strconv"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/cockroachdb/errors"
@@ -19,7 +20,10 @@ import (
 // amount of latency injected follows an exponential distribution with the
 // provided mean. Latency injected is derived from the provided seed and is
 // deterministic with respect to each file's path.
-func RandomLatency(pred Predicate, mean time.Duration, seed int64) Injector {
+//
+// If limit is nonzero, total latency injected over the lifetime of the Injector
+// is capped to limit.
+func RandomLatency(pred Predicate, mean time.Duration, seed int64, limit time.Duration) Injector {
 	rl := &randomLatency{
 		predicate: pred,
 		mean:      mean,
@@ -47,13 +51,18 @@ func parseRandomLatency(p *Parser, s *dsl.Scanner) Injector {
 	if tok.Kind != token.RPAREN {
 		panic(errors.Errorf("errorfs: unexpected token %s; expected %s", tok.String(), token.RPAREN))
 	}
-	return RandomLatency(pred, dur, seed)
+	return RandomLatency(pred, dur, seed, 0 /* no limit */)
 }
 
 type randomLatency struct {
 	predicate Predicate
-	// p defines the probability of an error being injected.
+	// mean is the mean duration injected each operation.
 	mean time.Duration
+	// limit configures a limit on total latency injected over the lifetime of
+	// the Injector.
+	limit time.Duration
+	// agg is the aggregate latency injected over the lifetime of the Injector.
+	agg atomic.Int64
 	keyedPrng
 }
 
@@ -75,6 +84,16 @@ func (rl *randomLatency) MaybeError(op Op) error {
 		// enough that causes a test timeout.
 		dur = time.Duration(min(prng.ExpFloat64(), 20.0) * float64(rl.mean))
 	})
+
+	if v := time.Duration(rl.agg.Add(int64(dur))); v-dur > rl.limit {
+		// We'd already exceeded the limit before adding dur. Don't inject
+		// anything.
+		return nil
+	} else if v > rl.limit {
+		// We're about to exceed the limit. Cap the duration.
+		dur -= v - rl.limit
+	}
+
 	time.Sleep(dur)
 	return nil
 }


Test patch:
diff --git a/db_test.go b/db_test.go
index 4864b652..a864ed82 100644
--- a/db_test.go
+++ b/db_test.go
@@ -2214,7 +2214,8 @@ func TestDeterminism(t *testing.T) {
 						if cmdArg.Key == "io-latency" {
 							prevMkfs := mkfs
 							mkfs = func() vfs.FS {
-								return errorfs.Wrap(prevMkfs(), errorfs.RandomLatency(errorfs.Randomly(p, 0), mean, 0))
+								return errorfs.Wrap(prevMkfs(), errorfs.RandomLatency(
+									errorfs.Randomly(p, 0), mean, 0 /* seed */, 0 /* no limit */))
 							}
 						} else if cmdArg.Key == "step-latency" {
 							beforeStep = func() {
diff --git a/error_test.go b/error_test.go
index 2766ab8f..c152c92c 100644
--- a/error_test.go
+++ b/error_test.go
@@ -447,7 +447,7 @@ func TestDBCompactionCrash(t *testing.T) {
 		// really slow. See https://github.com/golang/go/issues/61042,
 		// https://github.com/golang/go/issues/44343.
 		if runtime.GOOS != "windows" {
-			fs = errorfs.Wrap(fs, errorfs.RandomLatency(nil, 20*time.Microsecond, seed))
+			fs = errorfs.Wrap(fs, errorfs.RandomLatency(nil, 20*time.Microsecond, seed, 0 /* no limit */))
 		}
 		rng := rand.New(rand.NewSource(seed))
 		maxConcurrentCompactions := rng.Intn(3) + 2
diff --git a/metrics_test.go b/metrics_test.go
index c649968d..9b1d3a17 100644
--- a/metrics_test.go
+++ b/metrics_test.go
@@ -442,7 +442,8 @@ func TestMetricsWAmpDisableWAL(t *testing.T) {
 // Metrics.WAL.BytesWritten metric is always nondecreasing.
 // It's a regression test for issue #3505.
 func TestMetricsWALBytesWrittenMonotonicity(t *testing.T) {
-	fs := errorfs.Wrap(vfs.NewMem(), errorfs.RandomLatency(nil, 100*time.Microsecond, time.Now().UnixNano()))
+	fs := errorfs.Wrap(vfs.NewMem(), errorfs.RandomLatency(
+		nil, 100*time.Microsecond, time.Now().UnixNano(), 0 /* no limit */))
 	d, err := Open("", &Options{
 		FS: fs,
 		// Use a tiny memtable size so that we get frequent flushes. While a
diff --git a/sstable/reader_iter_test.go b/sstable/reader_iter_test.go
new file mode 100644
index 00000000..43357247
--- /dev/null
+++ b/sstable/reader_iter_test.go
@@ -0,0 +1,89 @@
+// Copyright 2024 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 sstable
+
+import (
+	"context"
+	"math/rand"
+	"testing"
+
+	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/testutils"
+	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
+	"github.com/cockroachdb/pebble/sstable/block"
+	"github.com/cockroachdb/pebble/vfs"
+	"github.com/stretchr/testify/require"
+)
+
+// TestIteratorErrorOnInit tests the path where creation of an iterator fails
+// when reading the index block.
+func TestIteratorErrorOnInit(t *testing.T) {
+	mem := vfs.NewMem()
+
+	f0, err := mem.Create("test.sst", vfs.WriteCategoryUnspecified)
+	require.NoError(t, err)
+	writerOpts := WriterOptions{
+		Comparer:   base.DefaultComparer,
+		MergerName: base.DefaultMerger.Name,
+	}
+	w := NewWriter(objstorageprovider.NewFileWritable(f0), writerOpts)
+	require.NoError(t, w.Set([]byte("test"), nil))
+	require.NoError(t, w.Close())
+
+	f1 := testutils.CheckErr(mem.Open("test.sst"))
+	r, err := newReader(f1, ReaderOptions{
+		Comparer: base.DefaultComparer,
+		Merger:   base.DefaultMerger,
+	})
+	require.NoError(t, err)
+	defer r.Close()
+
+	// Swap the readable in the reader.
+	bad := testutils.CheckErr(mem.Create("bad.sst", vfs.WriteCategoryUnspecified))
+	require.NoError(t, bad.Close())
+	bad = testutils.CheckErr(mem.Open("bad.sst"))
+	saveReadable := r.readable
+	r.readable = testutils.CheckErr(NewSimpleReadable(bad))
+
+	var pool block.BufferPool
+	pool.Init(5)
+
+	var stats base.InternalIteratorStats
+	for k := 0; k < 20; k++ {
+		if rand.Intn(2) == 0 {
+			_, err := newSingleLevelIterator(
+				context.Background(),
+				r,
+				nil, /* v */
+				NoTransforms,
+				nil /* lower */, nil, /* upper */
+				nil /* filterer */, false, /* useFilter */
+				&stats,
+				CategoryAndQoS{},
+				nil, /* statsCollector */
+				TrivialReaderProvider{Reader: r},
+				&pool,
+			)
+			require.Error(t, err)
+		} else {
+			_, err := newTwoLevelIterator(
+				context.Background(),
+				r,
+				nil, /* v */
+				NoTransforms,
+				nil /* lower */, nil, /* upper */
+				nil /* filterer */, false, /* useFilter */
+				&stats,
+				CategoryAndQoS{},
+				nil, /* statsCollector */
+				TrivialReaderProvider{Reader: r},
+				&pool,
+			)
+			require.Error(t, err)
+		}
+	}
+	require.NoError(t, r.readable.Close())
+	r.readable = saveReadable
+}
diff --git a/wal/failover_manager_test.go b/wal/failover_manager_test.go
index 8af0dcd2..770af4a9 100644
--- a/wal/failover_manager_test.go
+++ b/wal/failover_manager_test.go
@@ -573,7 +573,8 @@ func TestFailoverManager_Quiesce(t *testing.T) {
 	memFS := vfs.NewMem()
 	require.NoError(t, memFS.MkdirAll("primary", os.ModePerm))
 	require.NoError(t, memFS.MkdirAll("secondary", os.ModePerm))
-	fs := errorfs.Wrap(memFS, errorfs.RandomLatency(errorfs.Randomly(0.50, seed), 10*time.Millisecond, seed))
+	fs := errorfs.Wrap(memFS, errorfs.RandomLatency(
+		errorfs.Randomly(0.50, seed), 10*time.Millisecond, seed, 0 /* no limit */))
 
 	var m failoverManager
 	require.NoError(t, m.init(Options{