db: embed DataCorruptionInfo into corruption errors

We now add the `DataCorruptionInfo` to corruption errors and allow
users to extract it using `ExtractDataCorruptionInfo()`.

Author: RaduBerinde

event.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/cockroachdb/crlib/crtime"
 	"github.com/cockroachdb/errors"
+	errorsjoin "github.com/cockroachdb/errors/join"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/humanize"
 	"github.com/cockroachdb/pebble/internal/invariants"
@@ -1144,23 +1145,25 @@ func (r *lowDiskSpaceReporter) findThreshold(
 	return threshold, ok
 }
 
-func (d *DB) reportCorruption(meta any, err error) {
+// reportCorruption reports a corruption of a TableMetadata or BlobFileMetadata
+// to the event listener and also adds a DataCorruptionInfo payload to the error.
+func (d *DB) reportCorruption(meta any, err error) error {
 	switch meta := meta.(type) {
 	case *manifest.TableMetadata:
-		d.reportFileCorruption(base.FileTypeTable, meta.FileBacking.DiskFileNum, meta.UserKeyBounds(), err)
+		return d.reportFileCorruption(base.FileTypeTable, meta.FileBacking.DiskFileNum, meta.UserKeyBounds(), err)
 	case *manifest.BlobFileMetadata:
 		// TODO(jackson): Add bounds for blob files.
-		d.reportFileCorruption(base.FileTypeBlob, meta.FileNum, base.UserKeyBounds{}, err)
+		return d.reportFileCorruption(base.FileTypeBlob, meta.FileNum, base.UserKeyBounds{}, err)
 	default:
 		panic(fmt.Sprintf("unknown metadata type: %T", meta))
 	}
 }
 
 func (d *DB) reportFileCorruption(
 	fileType base.FileType, fileNum base.DiskFileNum, userKeyBounds base.UserKeyBounds, err error,
-) {
-	if invariants.Enabled && err == nil {
-		panic("nil error")
+) error {
+	if invariants.Enabled && !IsCorruptionError(err) {
+		panic("not a corruption error")
 	}
 
 	objMeta, lookupErr := d.objProvider.Lookup(fileType, fileNum)
@@ -1189,4 +1192,25 @@ func (d *DB) reportFileCorruption(
 		Details:  err,
 	}
 	d.opts.EventListener.DataCorruption(info)
+	// We don't use errors.Join() because that also annotates with this stack
+	// trace which would not be useful.
+	return errorsjoin.Join(err, &corruptionDetailError{info: info})
+}
+
+type corruptionDetailError struct {
+	info DataCorruptionInfo
+}
+
+func (e *corruptionDetailError) Error() string {
+	return "<corruption detail carrier>"
+}
+
+// ExtractDataCorruptionInfo extracts the DataCorruptionInfo details from a
+// corruption error. Returns nil if there is no such detail.
+func ExtractDataCorruptionInfo(err error) *DataCorruptionInfo {
+	var e *corruptionDetailError
+	if errors.As(err, &e) {
+		return &e.info
+	}
+	return nil
 }

event_listener_test.go

@@ -590,11 +590,15 @@ func TestSSTCorruptionEvent(t *testing.T) {
 			}
 			_, _, err = d.Get(key(5))
 			require.Error(t, err)
+			require.True(t, IsCorruptionError(err))
+			infoInError := ExtractDataCorruptionInfo(err)
+			require.NotNil(t, infoInError)
 			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)
+			require.Equal(t, info, *infoInError)
 
 			d.Close()
 		})

file_cache.go

@@ -106,8 +106,9 @@ type fileCacheHandle struct {
 	sstStatsCollector block.CategoryStatsCollector
 
 	// reportCorruptionFn is used for block.ReadEnv.ReportCorruptionFn. It expects
-	// the first argument to be a `*TableMetadata`.
-	reportCorruptionFn func(any, error)
+	// the first argument to be a `*TableMetadata`. It returns an error that
+	// contains more details.
+	reportCorruptionFn func(any, error) error
 
 	// This struct is only populated in race builds.
 	raceMu struct {
@@ -132,7 +133,7 @@ func (c *FileCache) newHandle(
 	objProvider objstorage.Provider,
 	loggerAndTracer LoggerAndTracer,
 	readerOpts sstable.ReaderOptions,
-	reportCorruptionFn func(any, error),
+	reportCorruptionFn func(any, error) error,
 ) *fileCacheHandle {
 	c.Ref()
 
@@ -241,7 +242,7 @@ func (h *fileCacheHandle) findOrCreateTable(
 	}
 	valRef, err := h.fileCache.c.FindOrCreate(ctx, key)
 	if err != nil && IsCorruptionError(err) {
-		h.reportCorruptionFn(meta, err)
+		err = h.reportCorruptionFn(meta, err)
 	}
 	return valRef, err
 }

file_cache_test.go

@@ -192,7 +192,7 @@ func (t *fileCacheTest) cleanup() {
 	t.blockCache.Unref()
 }
 
-func noopCorruptionFn(any, error) {}
+func noopCorruptionFn(_ any, err error) error { return err }
 
 // newTestHandle creates a filesystem with a set of test tables and an
 // associated file cache handle. The caller must close the handle.

sstable/block/block.go

@@ -377,8 +377,9 @@ type ReadEnv struct {
 
 	// ReportCorruptionFn is called with ReportCorruptionArg and the error
 	// whenever an SSTable corruption is detected. The argument is used to avoid
-	// allocating a separate function for each object.
-	ReportCorruptionFn  func(opaque any, err error)
+	// allocating a separate function for each object. It returns an error with
+	// more details.
+	ReportCorruptionFn  func(opaque any, err error) error
 	ReportCorruptionArg any
 }
 
@@ -406,10 +407,11 @@ func (env *ReadEnv) BlockRead(blockLength uint64, readDuration time.Duration) {
 
 // maybeReportCorruption calls the ReportCorruptionFn if the given error
 // indicates corruption.
-func (env *ReadEnv) maybeReportCorruption(err error) {
+func (env *ReadEnv) maybeReportCorruption(err error) error {
 	if env.ReportCorruptionFn != nil && base.IsCorruptionError(err) {
-		env.ReportCorruptionFn(env.ReportCorruptionArg, err)
+		return env.ReportCorruptionFn(env.ReportCorruptionArg, err)
 	}
+	return err
 }
 
 // A Reader reads blocks from a single file, handling caching, checksum
@@ -471,8 +473,7 @@ func (r *Reader) Read(
 		}
 		value, err := r.doRead(ctx, env, readHandle, bh, initBlockMetadataFn)
 		if err != nil {
-			env.maybeReportCorruption(err)
-			return BufferHandle{}, err
+			return BufferHandle{}, env.maybeReportCorruption(err)
 		}
 		return value.MakeHandle(), nil
 	}
@@ -490,8 +491,7 @@ func (r *Reader) Read(
 		// to report corruption errors separately, since the ReportCorruptionArg
 		// could be different. In particular, we might read the same physical block
 		// (e.g. an index block) for two different virtual tables.
-		env.maybeReportCorruption(err)
-		return BufferHandle{}, err
+		return BufferHandle{}, env.maybeReportCorruption(err)
 	}
 
 	if cv != nil {
@@ -507,8 +507,7 @@ func (r *Reader) Read(
 	value, err := r.doRead(ctx, env, readHandle, bh, initBlockMetadataFn)
 	if err != nil {
 		crh.SetReadError(err)
-		env.maybeReportCorruption(err)
-		return BufferHandle{}, err
+		return BufferHandle{}, env.maybeReportCorruption(err)
 	}
 	crh.SetReadValue(value.v)
 	return value.MakeHandle(), nil