db: rework SSTable corruption handling

This commit reworks SSTable corruption handling. Instead of Pebble
calling `Fatalf` in some cases, and the higher layer (CRDB) calling it
in other cases, we add a `DataCorruption` event. The default handler
calls `Fatalf` but a custom handler can be used for different
behavior. We rework the `MustExist` semantics to mark the not-exist
error as a corruption error and rely on the same reporting mechanism.

We implement this by adding a callback to report corruption to
`block.ReadEnv`. This was necessary because some of the information we
need (like the user key bounds) are only available at a higher level.

Author: RaduBerinde

compaction.go

@@ -2562,8 +2562,10 @@ func (d *DB) runCopyCompaction(
 
 		// NB: external files are always virtual.
 		var wrote uint64
-		err = d.fileCache.withVirtualReader(inputMeta.VirtualMeta(), func(r sstable.VirtualReader) error {
+		err = d.fileCache.withVirtualReader(ctx, block.NoReadEnv, inputMeta.VirtualMeta(), func(r sstable.VirtualReader, _ block.ReadEnv) error {
 			var err error
+			// TODO(radu): plumb a ReadEnv to CopySpan (it could use the buffer pool
+			// or update category stats).
 			wrote, err = sstable.CopySpan(ctx,
 				src, r.UnsafeReader(), d.opts.MakeReaderOptions(),
 				w, d.opts.MakeWriterOptions(c.outputLevel.level, d.TableFormat()),

error_test.go

@@ -16,6 +16,7 @@ import (
 	"time"
 
 	"github.com/cockroachdb/errors"
+	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/cockroachdb/pebble/vfs/errorfs"
@@ -271,7 +272,7 @@ func TestCorruptReadError(t *testing.T) {
 		opts := &Options{
 			DisableTableStats:  true,
 			FS:                 fs,
-			Logger:             panicLogger{},
+			Logger:             base.NoopLoggerAndTracer{},
 			FormatMajorVersion: formatVersion,
 		}
 		d, err := Open("", opts)

event.go

@@ -16,6 +16,8 @@ import (
 	"github.com/cockroachdb/pebble/internal/humanize"
 	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/manifest"
+	"github.com/cockroachdb/pebble/objstorage"
+	"github.com/cockroachdb/pebble/objstorage/remote"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/cockroachdb/redact"
 )
@@ -45,6 +47,35 @@ func formatFileNums(tables []TableInfo) string {
 	return buf.String()
 }
 
+// DataCorruptionInfo contains the information for a DataCorruption event.
+type DataCorruptionInfo struct {
+	// Path of the file that is corrupted. For remote files the path starts with
+	// "remote://".
+	Path     string
+	IsRemote bool
+	// Locator is only set when IsRemote is true (note that an empty Locator is
+	// valid even then).
+	Locator remote.Locator
+	// Bounds indicates the keyspace range that is affected.
+	Bounds base.UserKeyBounds
+	// Details of the error. See cockroachdb/error for how to format with or
+	// without redaction.
+	Details error
+}
+
+func (i DataCorruptionInfo) String() string {
+	return redact.StringWithoutMarkers(i)
+}
+
+// SafeFormat implements redact.SafeFormatter.
+func (i DataCorruptionInfo) SafeFormat(w redact.SafePrinter, _ rune) {
+	w.Printf("on-disk corruption: %s", redact.Safe(i.Path))
+	if i.IsRemote {
+		w.Printf(" (remote locator %q)", redact.Safe(i.Locator))
+	}
+	w.Printf("; bounds: %s; details: %+v", i.Bounds.String(), i.Details)
+}
+
 // LevelInfo contains info pertaining to a particular level.
 type LevelInfo struct {
 	Level  int
@@ -693,6 +724,10 @@ type EventListener struct {
 	// operation such as flush or compaction.
 	BackgroundError func(error)
 
+	// DataCorruption is invoked when an on-disk corruption is detected. It should
+	// not block, as it is called synchronously in read paths.
+	DataCorruption func(DataCorruptionInfo)
+
 	// CompactionBegin is invoked after the inputs to a compaction have been
 	// determined, but before the compaction has produced any output.
 	CompactionBegin func(CompactionInfo)
@@ -797,6 +832,15 @@ func (l *EventListener) EnsureDefaults(logger Logger) {
 			l.BackgroundError = func(error) {}
 		}
 	}
+	if l.DataCorruption == nil {
+		if logger != nil {
+			l.DataCorruption = func(info DataCorruptionInfo) {
+				logger.Fatalf("%s", info)
+			}
+		} else {
+			l.DataCorruption = func(info DataCorruptionInfo) {}
+		}
+	}
 	if l.CompactionBegin == nil {
 		l.CompactionBegin = func(info CompactionInfo) {}
 	}
@@ -873,6 +917,9 @@ func MakeLoggingEventListener(logger Logger) EventListener {
 		BackgroundError: func(err error) {
 			backgroundError(logger, err)
 		},
+		DataCorruption: func(info DataCorruptionInfo) {
+			logger.Errorf("%s", info)
+		},
 		CompactionBegin: func(info CompactionInfo) {
 			logger.Infof("%s", info)
 		},
@@ -948,6 +995,10 @@ func TeeEventListener(a, b EventListener) EventListener {
 			a.BackgroundError(err)
 			b.BackgroundError(err)
 		},
+		DataCorruption: func(info DataCorruptionInfo) {
+			a.DataCorruption(info)
+			b.DataCorruption(info)
+		},
 		CompactionBegin: func(info CompactionInfo) {
 			a.CompactionBegin(info)
 			b.CompactionBegin(info)
@@ -1098,3 +1149,35 @@ func (r *lowDiskSpaceReporter) findThreshold(
 	}
 	return threshold, ok
 }
+
+func (d *DB) reportSSTableCorruption(meta *manifest.TableMetadata, err error) {
+	if invariants.Enabled && err == nil {
+		panic("nil error")
+	}
+	objMeta, lookupErr := d.objProvider.Lookup(base.FileTypeTable, meta.FileBacking.DiskFileNum)
+	if lookupErr != nil {
+		// If the object is not known to the provider, it must be a local object
+		// that was missing when we opened the store. Remote objects have their
+		// metadata in a catalog, so even if the backing object is deleted, the
+		// DiskFileNum would still be known.
+		objMeta = objstorage.ObjectMetadata{DiskFileNum: meta.FileBacking.DiskFileNum, FileType: base.FileTypeTable}
+	}
+	path := d.objProvider.Path(objMeta)
+	if objMeta.IsRemote() {
+		// Remote path (which include the locator and full path) might not always be
+		// safe.
+		err = errors.WithHintf(err, "path: %s", path)
+	} else {
+		// Local paths are safe: they start with the store directory and the
+		// filename is generated by Pebble.
+		err = errors.WithHintf(err, "path: %s", redact.Safe(path))
+	}
+	info := DataCorruptionInfo{
+		Path:     path,
+		IsRemote: objMeta.IsRemote(),
+		Locator:  objMeta.Remote.Locator,
+		Bounds:   meta.UserKeyBounds(),
+		Details:  err,
+	}
+	d.opts.EventListener.DataCorruption(info)
+}

event_listener_test.go

@@ -10,6 +10,7 @@ import (
 	"fmt"
 	"reflect"
 	"runtime"
+	"slices"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -20,6 +21,7 @@ import (
 	"github.com/cockroachdb/datadriven"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/testutils"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/vfs"
@@ -350,12 +352,6 @@ func TestEventListenerRedact(t *testing.T) {
 	require.Equal(t, "[JOB 5] WAL delete error: unredacted error: ‹×›\n", log.String())
 }
 
-func TestEventListenerEnsureDefaultsBackgroundError(t *testing.T) {
-	e := EventListener{}
-	e.EnsureDefaults(nil)
-	e.BackgroundError(errors.New("an example error"))
-}
-
 func TestEventListenerEnsureDefaultsSetsAllCallbacks(t *testing.T) {
 	e := EventListener{}
 	e.EnsureDefaults(nil)
@@ -542,3 +538,76 @@ type mockDiskUsageFS struct {
 func (fs *mockDiskUsageFS) GetDiskUsage(path string) (vfs.DiskUsage, error) {
 	return fs.usage.Load().(vfs.DiskUsage), nil
 }
+
+func TestSSTCorruptionEvent(t *testing.T) {
+	for _, test := range []string{"missing-file", "missing-before-open", "meta-block-corruption", "data-block-corruption"} {
+		t.Run(test, func(t *testing.T) {
+			var mu sync.Mutex
+			var events []DataCorruptionInfo
+			fs := vfs.NewMem()
+			opts := &Options{
+				FS: fs,
+				// We use panicLogger to avoid errors being printed and make sure Fatalf
+				// is not called.
+				Logger: panicLogger{},
+				EventListener: &EventListener{
+					DataCorruption: func(info DataCorruptionInfo) {
+						mu.Lock()
+						defer mu.Unlock()
+						events = append(events, info)
+					},
+				},
+				DisableAutomaticCompactions: true,
+			}
+			d, err := Open("", opts)
+			require.NoError(t, err)
+			key := func(k int) []byte {
+				return []byte(fmt.Sprintf("key-%05d", k))
+			}
+
+			for i := 0; i < 100; i++ {
+				d.Set(key(i), []byte(fmt.Sprintf("value-%05d", i)), nil)
+			}
+			require.NoError(t, d.Flush())
+			require.NoError(t, d.Compact([]byte("a"), []byte("z"), false /* parallelize */))
+
+			// We expect a single sst file.
+			files := testutils.CheckErr(fs.List(""))
+			files = slices.DeleteFunc(files, func(name string) bool {
+				return !strings.HasSuffix(name, ".sst")
+			})
+			require.Lenf(t, files, 1, "expected a single sst file, got %v", files)
+			sstFileName := files[0]
+
+			switch test {
+			case "missing-file":
+				require.NoError(t, fs.Remove(sstFileName))
+			case "missing-before-open":
+				require.NoError(t, d.Close())
+				require.NoError(t, fs.Remove(sstFileName))
+				opts.DisableConsistencyCheck = true
+				d, err = Open("", opts)
+				require.NoError(t, err)
+			case "meta-block-corruption":
+				buf, err := fs.UnsafeGetFileDataBuffer(sstFileName)
+				require.NoError(t, err)
+				buf[len(buf)-100]++
+			case "data-block-corruption":
+				buf, err := fs.UnsafeGetFileDataBuffer(sstFileName)
+				require.NoError(t, err)
+				buf[20]++
+			default:
+				t.Fatalf("invalid test")
+			}
+			_, _, err = d.Get(key(5))
+			require.Error(t, err)
+			require.Greater(t, len(events), 0)
+			info := events[0]
+			require.Equal(t, info.Path, sstFileName)
+			require.False(t, info.IsRemote)
+			require.Equal(t, base.UserKeyBoundsInclusive(key(0), key(99)), info.Bounds)
+
+			d.Close()
+		})
+	}
+}