storage: do not panic on corruption in SST iterator

In iterators created through NewPebbleSSTIterator, do not panic on corruption
errors at Close. The panic is important for iterators created over committed
engine state, but it's a problem when iterating over sstables from external
sources like backups.

Release justification: minor, low-risk bug fix.
Release note: None

pkg/storage/pebble_iterator.go

@@ -53,6 +53,12 @@ type pebbleIterator struct {
 	// iterator, but simply marks it as not inuse. Used by pebbleReadOnly.
 	reusable bool
 	inuse    bool
+	// Set to true if the underlying Pebble Iterator was created through
+	// pebble.NewExternalIter, and so the iterator is iterating over files
+	// external to the storage engine. This is used to avoid panicking on
+	// corruption errors that should be non-fatal if encountered from external
+	// sources of sstables.
+	external bool
 	// mvccDirIsReverse and mvccDone are used only for the methods implementing
 	// MVCCIterator. They are used to prevent the iterator from iterating into
 	// the lock table key space.
@@ -122,6 +128,7 @@ func newPebbleSSTIterator(
 		p.Close()
 		return nil, err
 	}
+	p.external = true
 	return p, nil
 }
 
@@ -920,7 +927,10 @@ func (p *pebbleIterator) destroy() {
 		// we panic on the error types we know to be NOT ephemeral.
 		//
 		// See cockroachdb/pebble#1811.
-		if err := p.iter.Close(); errors.Is(err, pebble.ErrCorruption) {
+		//
+		// NB: The panic is omitted if the error is encountered on an external
+		// iterator which is iterating over uncommitted sstables.
+		if err := p.iter.Close(); !p.external && errors.Is(err, pebble.ErrCorruption) {
 			panic(err)
 		}
 		p.iter = nil

pkg/storage/pebble_iterator_test.go

@@ -13,14 +13,21 @@ package storage
 import (
 	"context"
 	"io/fs"
+	"math/rand"
 	"os"
 	"path/filepath"
 	"strings"
 	"testing"
 
+	"github.com/cockroachdb/cockroach/pkg/clusterversion"
+	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble"
+	"github.com/cockroachdb/pebble/sstable"
+	"github.com/cockroachdb/pebble/vfs"
 	"github.com/stretchr/testify/require"
 )
 
@@ -75,3 +82,60 @@ func TestPebbleIterator_Corruption(t *testing.T) {
 	// Closing the iter results in a panic due to the corruption.
 	require.Panics(t, func() { iter.Close() })
 }
+
+func randStr(fill []byte, rng *rand.Rand) {
+	const letters = "abcdefghijklmnopqrstuvwxyz"
+	const lettersLen = len(letters)
+	for i := 0; i < len(fill); i++ {
+		fill[i] = letters[rng.Intn(lettersLen)]
+	}
+}
+
+func TestPebbleIterator_ExternalCorruption(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	version := clusterversion.ByKey(clusterversion.EnsurePebbleFormatVersionRangeKeys)
+	st := cluster.MakeTestingClusterSettingsWithVersions(version, version, true)
+	ctx := context.Background()
+	rng := rand.New(rand.NewSource(timeutil.Now().UnixNano()))
+	f := &MemFile{}
+	w := MakeBackupSSTWriter(ctx, st, f)
+
+	// Create an example sstable.
+	var rawValue [64]byte
+	for i := 0; i < 26; i++ {
+		const numVersions = 10
+		for j := 0; j < numVersions; j++ {
+			randStr(rawValue[:], rng)
+			v, err := EncodeMVCCValue(MVCCValue{Value: roachpb.MakeValueFromBytes(rawValue[:])})
+			require.NoError(t, err)
+			require.NoError(t, w.Put(pointKey(string(rune('a'+i)), numVersions-j), v))
+		}
+	}
+	require.NoError(t, w.Finish())
+
+	// Trash a random byte.
+	b := f.Bytes()
+	b[rng.Intn(len(b))]++
+
+	it, err := NewPebbleSSTIterator([][]sstable.ReadableFile{{vfs.NewMemFile(b)}},
+		IterOptions{UpperBound: roachpb.KeyMax}, false)
+
+	// We may error early, while opening the iterator.
+	if err != nil {
+		require.True(t, errors.Is(err, pebble.ErrCorruption))
+		return
+	}
+
+	it.SeekGE(NilKey)
+	valid, err := it.Valid()
+	for valid {
+		it.Next()
+		valid, err = it.Valid()
+	}
+	// Or we may error during iteration.
+	if err != nil {
+		require.True(t, errors.Is(err, pebble.ErrCorruption))
+	}
+	it.Close()
+}