db: propagate *PhysicalBlobFile to reportCorruption

If corruption is encountered while reading a blob file, propagate the blob
file's *PhysicalBlobFile metadata struct to DB.reportCorruption. This fixes a
nil pointer that would previously result while handling blob file corruption.

Fixes #5127.

Author: jbowens

blob_rewrite.go

@@ -181,12 +181,12 @@ func (c *blobFileRewriteCompaction) Execute(jobID JobID, d *DB) error {
 			// Now that we have the manifest lock, check if the blob file is
 			// still current. If not, we bubble up ErrCancelledCompaction.
 			v := d.mu.versions.currentVersion()
-			currentDiskFile, ok := v.BlobFiles.Lookup(c.input.FileID)
+			currentDiskFile, ok := v.BlobFiles.LookupPhysical(c.input.FileID)
 			if !ok {
 				return versionUpdate{}, errors.Wrapf(ErrCancelledCompaction,
 					"blob file %s became unreferenced", c.input.FileID)
 			}
-			currentDiskFileNum := currentDiskFile.DiskFileNum()
+			currentDiskFileNum := currentDiskFile.FileNum
 			// Assert that the current version's disk file number for the blob
 			// matches the one we rewrote. This compaction should be the only
 			// rewrite compaction running for this blob file.
@@ -391,7 +391,7 @@ func newBlobFileRewriter(
 	sstables []*manifest.TableMetadata,
 	inputBlob manifest.BlobFileMetadata,
 ) *blobFileRewriter {
-	rw := blob.NewFileRewriter(inputBlob.FileID, inputBlob.Physical.FileNum, fc, readEnv, outputFileNum, w, opts)
+	rw := blob.NewFileRewriter(inputBlob.FileID, inputBlob.Physical, fc, readEnv, outputFileNum, w, opts)
 	return &blobFileRewriter{
 		fc:        fc,
 		readEnv:   readEnv,

blob_rewrite_test.go

@@ -173,7 +173,7 @@ func TestBlobRewrite(t *testing.T) {
 					objStore,
 					&base.LoggerWithNoopTracer{Logger: base.DefaultLogger},
 					sstable.ReaderOptions{},
-					func(any, error) error { return nil },
+					func(base.ObjectInfo, error) error { return nil },
 				)
 				var sstables []*manifest.TableMetadata
 				for _, sstFileNum := range sstableFileNums {
@@ -421,6 +421,10 @@ type constantFileMapping base.DiskFileNum
 // Assert that (*inputFileMapping) implements base.BlobFileMapping.
 var _ base.BlobFileMapping = constantFileMapping(0)
 
-func (m constantFileMapping) Lookup(fileID base.BlobFileID) (base.DiskFile, bool) {
-	return base.DiskFileNum(m), true
+func (m constantFileMapping) Lookup(fileID base.BlobFileID) (base.ObjectInfo, bool) {
+	return base.ObjectInfoLiteral{
+		FileType:    base.FileTypeBlob,
+		DiskFileNum: base.DiskFileNum(m),
+		Bounds:      base.UserKeyBounds{},
+	}, true
 }

checkpoint.go

@@ -326,11 +326,11 @@ func (d *DB) Checkpoint(
 						includedBlobFiles[ref.FileID] = struct{}{}
 
 						// Map the BlobFileID to a DiskFileNum in the current version.
-						diskFile, ok := current.BlobFiles.Lookup(ref.FileID)
+						obj, ok := current.BlobFiles.Lookup(ref.FileID)
 						if !ok {
 							return errors.Errorf("blob file %s not found", ref.FileID)
 						}
-						ckErr = copyFile(base.FileTypeBlob, diskFile.DiskFileNum())
+						ckErr = copyFile(obj.FileInfo())
 						if ckErr != nil {
 							return ckErr
 						}

db.go

@@ -633,7 +633,11 @@ func (d *DB) getInternal(key []byte, b *Batch, s *Snapshot) ([]byte, io.Closer,
 		}
 		return nil, nil, ErrNotFound
 	}
-	return i.Value(), i, nil
+	val, err := i.ValueAndErr()
+	if err != nil {
+		return nil, nil, errors.CombineErrors(err, i.Close())
+	}
+	return val, i, nil
 }
 
 // Set sets the value for the given key. It overwrites any previous value

event.go

@@ -1343,24 +1343,11 @@ func (r *lowDiskSpaceReporter) findThreshold(
 
 // 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:
-		return d.reportFileCorruption(base.FileTypeTable, meta.TableBacking.DiskFileNum, meta.UserKeyBounds(), err)
-	case *manifest.PhysicalBlobFile:
-		// TODO(jackson): Add bounds for blob files.
-		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,
-) error {
+func (d *DB) reportCorruption(meta base.ObjectInfo, err error) error {
 	if invariants.Enabled && !IsCorruptionError(err) {
 		panic("not a corruption error")
 	}
+	fileType, fileNum := meta.FileInfo()
 
 	objMeta, lookupErr := d.objProvider.Lookup(fileType, fileNum)
 	if lookupErr != nil {
@@ -1384,7 +1371,7 @@ func (d *DB) reportFileCorruption(
 		Path:     path,
 		IsRemote: objMeta.IsRemote(),
 		Locator:  objMeta.Remote.Locator,
-		Bounds:   userKeyBounds,
+		Bounds:   meta.UserKeyBounds(),
 		Details:  err,
 	}
 	d.opts.EventListener.DataCorruption(info)

event_listener_test.go

@@ -613,3 +613,80 @@ func TestSSTCorruptionEvent(t *testing.T) {
 		})
 	}
 }
+
+func TestBlobCorruptionEvent(t *testing.T) {
+	for _, test := range []string{"missing-file"} {
+		t.Run(test, func(t *testing.T) {
+			var mu sync.Mutex
+			var events []DataCorruptionInfo
+			fs := vfs.NewMem()
+			opts := &Options{
+				FS:                 fs,
+				Logger:             testLogger{t},
+				FormatMajorVersion: FormatValueSeparation,
+				EventListener: &EventListener{
+					DataCorruption: func(info DataCorruptionInfo) {
+						mu.Lock()
+						defer mu.Unlock()
+						events = append(events, info)
+					},
+				},
+				DisableAutomaticCompactions: true,
+			}
+			opts.Experimental.ValueSeparationPolicy = func() ValueSeparationPolicy {
+				return ValueSeparationPolicy{
+					Enabled:               true,
+					MinimumSize:           1,
+					MaxBlobReferenceDepth: 10,
+				}
+			}
+			d, err := Open("", opts)
+			require.NoError(t, err)
+
+			key := func(k int) []byte {
+				return []byte(fmt.Sprintf("key-%05d", k))
+			}
+
+			// Create large values to ensure they get separated into blob files
+			largeValue := make([]byte, 1000) // 1KB values
+			for i := range largeValue {
+				largeValue[i] = byte('a' + (i % 26))
+			}
+
+			for i := 0; i < 10; i++ {
+				require.NoError(t, d.Set(key(i), largeValue, nil))
+			}
+			require.NoError(t, d.Flush())
+
+			// We expect a blob file to be created.
+			files := testutils.CheckErr(fs.List(""))
+			blobFiles := slices.DeleteFunc(files, func(name string) bool {
+				return !strings.HasSuffix(name, ".blob")
+			})
+			require.Greaterf(t, len(blobFiles), 0, "expected at least one blob file, got %v", files)
+			for _, blobFileName := range blobFiles {
+				switch test {
+				case "missing-file":
+					require.NoError(t, fs.Remove(blobFileName))
+				default:
+					t.Fatalf("invalid test")
+				}
+			}
+
+			// Try to read a value that should be in the blob file
+			_, _, 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.False(t, info.IsRemote)
+			// Note: Blob files don't currently have user key bounds, so they're empty
+			require.Equal(t, base.UserKeyBounds{}, info.Bounds)
+			require.Equal(t, info, *infoInError)
+
+			d.Close()
+		})
+	}
+}

file_cache.go

@@ -118,10 +118,11 @@ type fileCacheHandle struct {
 	iterCount         atomic.Int32
 	sstStatsCollector block.CategoryStatsCollector
 
-	// reportCorruptionFn is used for block.ReadEnv.ReportCorruptionFn. It expects
-	// the first argument to be a `*TableMetadata`. It returns an error that
-	// contains more details.
-	reportCorruptionFn func(any, error) error
+	// reportCorruptionFn is used for block.ReadEnv.ReportCorruptionFn. The
+	// first argument must implement the ObjectInfo interface. Typically callers
+	// use *TableMetadata or *PhysicalBlobFile to satisfy this interface.
+	// reportCorruptionFn returns an error that contains more details.
+	reportCorruptionFn func(base.ObjectInfo, error) error
 
 	// This struct is only populated in race builds.
 	raceMu struct {
@@ -146,7 +147,7 @@ func (c *FileCache) newHandle(
 	objProvider objstorage.Provider,
 	loggerAndTracer LoggerAndTracer,
 	readerOpts sstable.ReaderOptions,
-	reportCorruptionFn func(any, error) error,
+	reportCorruptionFn func(base.ObjectInfo, error) error,
 ) *fileCacheHandle {
 	c.Ref()
 
@@ -268,17 +269,17 @@ func (h *fileCacheHandle) findOrCreateTable(
 // the given blob file. If a corruption error is encountered,
 // reportCorruptionFn() is called.
 func (h *fileCacheHandle) findOrCreateBlob(
-	ctx context.Context, fileNum base.DiskFileNum,
+	ctx context.Context, info base.ObjectInfo,
 ) (genericcache.ValueRef[fileCacheKey, fileCacheValue], error) {
+	ftyp, fileNum := info.FileInfo()
 	key := fileCacheKey{
 		handle:   h,
 		fileNum:  fileNum,
-		fileType: base.FileTypeBlob,
+		fileType: ftyp,
 	}
 	valRef, err := h.fileCache.c.FindOrCreate(ctx, key)
-	// TODO(jackson): Propagate a blob metadata object here.
 	if err != nil && IsCorruptionError(err) {
-		err = h.reportCorruptionFn(nil, err)
+		err = h.reportCorruptionFn(info, err)
 	}
 	return valRef, err
 }
@@ -388,9 +389,9 @@ func (h *fileCacheHandle) IterCount() int64 {
 
 // GetValueReader returns a blob.ValueReader for blob file identified by fileNum.
 func (h *fileCacheHandle) GetValueReader(
-	ctx context.Context, fileNum base.DiskFileNum,
+	ctx context.Context, diskFile base.ObjectInfo,
 ) (r blob.ValueReader, closeFunc func(), err error) {
-	ref, err := h.findOrCreateBlob(ctx, fileNum)
+	ref, err := h.findOrCreateBlob(ctx, diskFile)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -803,7 +804,7 @@ func SetupBlobReaderProvider(
 		objProvider,
 		opts.LoggerAndTracer,
 		readOpts,
-		func(any, error) error { return nil },
+		func(base.ObjectInfo, error) error { return nil },
 	)
 
 	return provider, cleanup, nil

file_cache_test.go

@@ -179,11 +179,17 @@ func newFileCacheTest(
 	}
 }
 
-func (t *fileCacheTest) fileByIdx(i int) (base.DiskFileNum, base.FileType) {
+func (t *fileCacheTest) fileByIdx(i int) base.ObjectInfo {
 	if i < fileCacheTestNumTables {
-		return base.DiskFileNum(i), base.FileTypeTable
+		return base.ObjectInfoLiteral{
+			FileType:    base.FileTypeTable,
+			DiskFileNum: base.DiskFileNum(i),
+		}
+	}
+	return base.ObjectInfoLiteral{
+		FileType:    base.FileTypeBlob,
+		DiskFileNum: base.DiskFileNum(i),
 	}
-	return base.DiskFileNum(i), base.FileTypeBlob
 }
 
 func (t *fileCacheTest) cleanup() {
@@ -192,7 +198,7 @@ func (t *fileCacheTest) cleanup() {
 	t.blockCache.Unref()
 }
 
-func noopCorruptionFn(_ any, err error) error { return err }
+func noopCorruptionFn(_ base.ObjectInfo, 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.
@@ -727,9 +733,10 @@ func testFileCacheFrequentlyUsedInternal(t *testing.T, rangeIter bool) {
 
 	for i := 0; i < N; i++ {
 		for _, j := range [...]int{pinned0, i % fileCacheTestNumFiles, pinned1} {
-			fn, typ := fct.fileByIdx(j)
-			if typ == base.FileTypeBlob {
-				_, closeFunc, err := h.GetValueReader(context.Background(), fn)
+			obj := fct.fileByIdx(j)
+			ftyp, fn := obj.FileInfo()
+			if ftyp == base.FileTypeBlob {
+				_, closeFunc, err := h.GetValueReader(context.Background(), obj)
 				if err != nil {
 					t.Fatalf("i=%d, j=%d: get value reader: %v", i, j, err)
 				}
@@ -787,9 +794,10 @@ func TestSharedFileCacheFrequentlyUsed(t *testing.T) {
 
 	for i := 0; i < N; i++ {
 		for _, j := range [...]int{pinned0, i % fileCacheTestNumFiles, pinned1} {
-			fn, typ := fct.fileByIdx(j)
-			if typ == base.FileTypeBlob {
-				_, closeFunc, err := h1.GetValueReader(context.Background(), fn)
+			obj := fct.fileByIdx(j)
+			ftyp, fn := obj.FileInfo()
+			if ftyp == base.FileTypeBlob {
+				_, closeFunc, err := h1.GetValueReader(context.Background(), obj)
 				if err != nil {
 					t.Fatalf("i=%d, j=%d: get value reader: %v", i, j, err)
 				}
@@ -847,9 +855,10 @@ func testFileCacheEvictionsInternal(t *testing.T, rangeIter bool) {
 
 	rng := rand.New(rand.NewPCG(2, 2))
 	for i := 0; i < N; i++ {
-		fn, typ := fct.fileByIdx(rng.IntN(fileCacheTestNumFiles))
-		if typ == base.FileTypeBlob {
-			_, closeFunc, err := h.GetValueReader(context.Background(), fn)
+		obj := fct.fileByIdx(rng.IntN(fileCacheTestNumFiles))
+		ftyp, fn := obj.FileInfo()
+		if ftyp == base.FileTypeBlob {
+			_, closeFunc, err := h.GetValueReader(context.Background(), obj)
 			if err != nil {
 				t.Fatalf("i=%d, fn=%d: get value reader: %v", i, fn, err)
 			}
@@ -873,7 +882,9 @@ func testFileCacheEvictionsInternal(t *testing.T, rangeIter bool) {
 			}
 		}
 
-		h.Evict(fct.fileByIdx(int(lo + rng.Uint64N(hi-lo))))
+		obj = fct.fileByIdx(int(lo + rng.Uint64N(hi-lo)))
+		ftyp, fn = obj.FileInfo()
+		h.Evict(fn, ftyp)
 	}
 
 	sumEvicted, nEvicted := 0, 0
@@ -923,13 +934,14 @@ func TestSharedFileCacheEvictions(t *testing.T) {
 	rng := rand.New(rand.NewPCG(0, 0))
 	for i := 0; i < N; i++ {
 		j := rng.IntN(fileCacheTestNumFiles)
-		fn, typ := fct.fileByIdx(j)
-		if typ == base.FileTypeBlob {
-			_, closeFunc1, err := h1.GetValueReader(context.Background(), fn)
+		obj := fct.fileByIdx(j)
+		ftyp, fn := obj.FileInfo()
+		if ftyp == base.FileTypeBlob {
+			_, closeFunc1, err := h1.GetValueReader(context.Background(), obj)
 			if err != nil {
 				t.Fatalf("i=%d, fn=%d: get value reader: %v", i, fn, err)
 			}
-			_, closeFunc2, err := h2.GetValueReader(context.Background(), fn)
+			_, closeFunc2, err := h2.GetValueReader(context.Background(), obj)
 			if err != nil {
 				t.Fatalf("i=%d, fn=%d: get value reader: %v", i, fn, err)
 			}
@@ -955,8 +967,12 @@ func TestSharedFileCacheEvictions(t *testing.T) {
 			}
 		}
 
-		h1.Evict(fct.fileByIdx(int(lo + rng.Uint64N(hi-lo))))
-		h2.Evict(fct.fileByIdx(int(lo + rng.Uint64N(hi-lo))))
+		obj = fct.fileByIdx(int(lo + rng.Uint64N(hi-lo)))
+		ftyp, fn = obj.FileInfo()
+		h1.Evict(fn, ftyp)
+		obj = fct.fileByIdx(int(lo + rng.Uint64N(hi-lo)))
+		ftyp, fn = obj.FileInfo()
+		h2.Evict(fn, ftyp)
 	}
 
 	check := func(fs *fileCacheTestFS, h *fileCacheHandle) (float64, float64, float64) {
@@ -1089,13 +1105,14 @@ func TestFileCacheRetryAfterFailure(t *testing.T) {
 		h, fs := fct.newTestHandle()
 
 		fs.setOpenError(true /* enabled */)
-		_, _, err := h.GetValueReader(ctx, fileCacheTestNumTables)
+		obj := fct.fileByIdx(fileCacheTestNumTables)
+		_, _, err := h.GetValueReader(ctx, obj)
 		if err == nil {
 			t.Fatalf("expected failure, but found success")
 		}
 		require.Equal(t, "pebble: backing file 000200 error: injected error", err.Error())
 		fs.setOpenError(false /* enabled */)
-		_, closeFunc, err := h.GetValueReader(ctx, fileCacheTestNumTables)
+		_, closeFunc, err := h.GetValueReader(ctx, obj)
 		require.NoError(t, err)
 		closeFunc()
 		fs.validateAndCloseHandle(t, h, nil)