storage: panic on iterator close when encountering corruption

In cockroachdb#84449, the behavior of the `pebbleIterator` was altered to ignore
all errors when closing an iterator. This was done to avoid panic-ing
when the iterator encounters an ephemeral (i.e. retriable) error that
remains pinned to the iterator until it is closed (see cockroachdb#84396 for the
motivation).

Partially address the TODO added in cockroachdb#84449 by panic-ing on errors that
are known to be fatal (such as corruption). Addressing the TODO in its
entirety would likely require enumerating known ephemeral errors and
allowing such errors to be ignored on close, as well as implementing
cockroachdb/pebble#1811 to avoid pinning errors to the iterator even
after they have been handled.

Fix cockroachdb#84479.

Release note: None.

pkg/storage/BUILD.bazel

@@ -120,6 +120,7 @@ go_test(
         "mvcc_test.go",
         "mvcc_value_test.go",
         "pebble_file_registry_test.go",
+        "pebble_iterator_test.go",
         "pebble_mvcc_scanner_test.go",
         "pebble_test.go",
         "read_as_of_iterator_test.go",

pkg/storage/pebble_iterator.go

@@ -899,17 +899,20 @@ func (p *pebbleIterator) destroy() {
 		// surfaced through Valid(), but wants to close the iterator (eg,
 		// potentially through a defer) and so we don't want to re-surface the
 		// error.
-		_ = p.iter.Close()
-
-		// TODO(jackson): In addition to errors accumulated during iteration,
-		// Close also returns errors encountered during the act of closing the
-		// iterator. Currently, these errors are swallowed. The error returned
-		// by iter.Close() may be an ephemeral error, or it may a misuse of the
+		//
+		// TODO(jackson): In addition to errors accumulated during iteration, Close
+		// also returns errors encountered during the act of closing the iterator.
+		// Currently, most of these errors are swallowed. The error returned by
+		// iter.Close() may be an ephemeral error, or it may a misuse of the
 		// Iterator or corruption. Only swallow ephemeral errors (eg,
-		// DeadlineExceeded, etc), panic-ing on Close errors that are not known
-		// to be ephemeral/retriable.
+		// DeadlineExceeded, etc), panic-ing on Close errors that are not known to
+		// be ephemeral/retriable. While these ephemeral error types are enumerated,
+		// 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) {
+			panic(err)
+		}
 		p.iter = nil
 	}
 	// Reset all fields except for the key and option buffers. Holding onto their

pkg/storage/pebble_iterator_test.go

@@ -0,0 +1,77 @@
+// Copyright 2022 The Cockroach Authors.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+package storage
+
+import (
+	"context"
+	"io/fs"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/errors"
+	"github.com/cockroachdb/pebble"
+	"github.com/stretchr/testify/require"
+)
+
+func TestPebbleIterator_Corruption(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	// Create a Pebble DB that can be used to back a pebbleIterator.
+	dir := t.TempDir()
+	dataDir := filepath.Join(dir, "data")
+	p, err := Open(context.Background(), Filesystem(dataDir))
+	require.NoError(t, err)
+	defer p.Close()
+
+	// Insert some data into the DB and flush to create an SST.
+	ek := engineKey("foo", 0)
+	require.NoError(t, p.PutEngineKey(ek, nil))
+	require.NoError(t, p.Flush())
+
+	// Corrupt the SSTs in the DB.
+	err = filepath.Walk(dataDir, func(path string, info fs.FileInfo, err error) error {
+		if err != nil {
+			return err
+		}
+		if !strings.HasSuffix(info.Name(), ".sst") {
+			return nil
+		}
+		file, err := os.OpenFile(path, os.O_WRONLY, 0600)
+		if err != nil {
+			return err
+		}
+		_, err = file.WriteAt([]byte("uh oh"), 0)
+		if err != nil {
+			return err
+		}
+		_ = file.Close()
+		return nil
+	})
+	require.NoError(t, err)
+
+	// Construct a pebbleIterator over the DB.
+	iterOpts := IterOptions{
+		LowerBound: []byte("a"),
+		UpperBound: []byte("z"),
+	}
+	iter := newPebbleIterator(p.db, iterOpts, StandardDurability, false /* range keys */)
+
+	// Seeking into the table catches the corruption.
+	ok, err := iter.SeekEngineKeyGE(ek)
+	require.False(t, ok)
+	require.True(t, errors.Is(err, pebble.ErrCorruption))
+
+	// Closing the iter results in a panic due to the corruption.
+	require.Panics(t, func() { iter.Close() })
+}