pebble: add missing cleanup in a NewExternalIter error case

If creating a point or range key iterator fails, we leak some
iterators that were already opened. This immediately leads to a
BufferPool-related panic when closing the `Iterator`.

This change adds an enhancement to a test which reproduces the panic
and patches up the error paths.

Informs github.com/cockroachdb/cockroach/issues/141606

Author: RaduBerinde

external_iterator.go

@@ -178,13 +178,21 @@ func createExternalPointIter(
 			pointIter, err = r.NewPointIter(
 				ctx, transforms, it.opts.LowerBound, it.opts.UpperBound, nil, /* BlockPropertiesFilterer */
 				sstable.NeverUseFilterBlock, readEnv, sstable.MakeTrivialReaderProvider(r))
-			if err != nil {
-				return nil, err
+			if err == nil {
+				rangeDelIter, err = r.NewRawRangeDelIter(ctx, sstable.FragmentIterTransforms{
+					SyntheticSeqNum: sstable.SyntheticSeqNum(seqNum),
+				}, readEnv)
 			}
-			rangeDelIter, err = r.NewRawRangeDelIter(ctx, sstable.FragmentIterTransforms{
-				SyntheticSeqNum: sstable.SyntheticSeqNum(seqNum),
-			}, readEnv)
 			if err != nil {
+				if pointIter != nil {
+					pointIter.Close()
+				}
+				for i := range mlevels {
+					mlevels[i].iter.Close()
+					if mlevels[i].rangeDelIter != nil {
+						mlevels[i].rangeDelIter.Close()
+					}
+				}
 				return nil, err
 			}
 			mlevels = append(mlevels, mergingIterLevel{
@@ -241,9 +249,14 @@ func finishInitializingExternal(ctx context.Context, it *Iterator) error {
 				for _, r := range readers {
 					transforms := sstable.FragmentIterTransforms{SyntheticSeqNum: sstable.SyntheticSeqNum(seqNum)}
 					seqNum--
-					if rki, err := r.NewRawRangeKeyIter(ctx, transforms, readEnv); err != nil {
+					rki, err := r.NewRawRangeKeyIter(ctx, transforms, readEnv)
+					if err != nil {
+						for _, iter := range rangeKeyIters {
+							iter.Close()
+						}
 						return err
-					} else if rki != nil {
+					}
+					if rki != nil {
 						rangeKeyIters = append(rangeKeyIters, rki)
 					}
 				}

external_iterator_test.go

@@ -8,9 +8,11 @@ import (
 	"bytes"
 	"fmt"
 	"math/rand/v2"
+	"slices"
 	"testing"
 
 	"github.com/cockroachdb/datadriven"
+	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
 	"github.com/cockroachdb/pebble/sstable"
@@ -61,6 +63,7 @@ func TestExternalIterator(t *testing.T) {
 					}
 				}
 			}
+			testExternalIteratorInitError(t, o, &opts, files)
 			it, err := NewExternalIter(o, &opts, files)
 			require.NoError(t, err)
 			return runIterCmd(td, it, true /* close iter */)
@@ -70,6 +73,45 @@ func TestExternalIterator(t *testing.T) {
 	})
 }
 
+// testExternalIteratorInitError tests error handling paths inside
+// NewExternalIter by injecting errors when reading files.
+//
+// See github.com/cockroachdb/cockroach/issues/141606 where an error during
+// initialization caused NewExternalIter to panic.
+func testExternalIteratorInitError(
+	t *testing.T, o *Options, iterOpts *IterOptions, files [][]sstable.ReadableFile,
+) {
+	files = slices.Clone(files)
+	for i := range files {
+		files[i] = slices.Clone(files[i])
+		for j := range files[i] {
+			files[i][j] = &flakyFile{ReadableFile: files[i][j]}
+		}
+	}
+
+	for iter := 0; iter < 100; iter++ {
+		it, err := NewExternalIter(o, iterOpts, files)
+		if err != nil {
+			require.Contains(t, err.Error(), "flaky file")
+		} else {
+			it.Close()
+		}
+	}
+}
+
+type flakyFile struct {
+	sstable.ReadableFile
+}
+
+func (ff *flakyFile) ReadAt(p []byte, off int64) (n int, err error) {
+	if rand.IntN(10) == 0 {
+		return 0, errors.New("flaky file")
+	}
+	return ff.ReadableFile.ReadAt(p, off)
+}
+
+func (ff *flakyFile) Close() error { return nil }
+
 func BenchmarkExternalIter_NonOverlapping_Scan(b *testing.B) {
 	ks := testkeys.Alpha(6)
 	opts := &Options{Comparer: testkeys.Comparer}