db: fix obsolete file metric underflows

Previously a race existed in the calculation of obsolete file metrics resulting
in underflow. The values within versionSet.metrics were reset and recalculated
to reflect the set of files in versionSet.obsoleteTables whenever new files
were added to obsoleteFiles. Additionally, the cleanup manager invoked a
callback to decrease versionSet.metrics whenever a table was deleted.

The recalculation of versionSet.metrics could reset the metrics to less than
the sum of outstanding pending deletes. When the cleanup manager eventually
deleted the pending tables, these metrics would underflow.

This commit fixes the bug by maintaining separate stats for all files that have
been enqueued for the cleanup manager and all files that have been successfully
deleted. The volume of outstanding, pending deletions is the difference between
the two.

For now, there's an additional wart that the set of files that are sitting in
versionSet.obsoleteTables are still separately tracked.

Informs #4811.

Author: jbowens

db.go

@@ -459,10 +459,16 @@ type DB struct {
 			noOngoingFlushStartTime time.Time
 		}
 
-		// Non-zero when file cleaning is disabled. The disabled count acts as a
-		// reference count to prohibit file cleaning. See
-		// DB.{disable,Enable}FileDeletions().
-		disableFileDeletions int
+		fileDeletions struct {
+			// Non-zero when file cleaning is disabled. The disabled count acts
+			// as a reference count to prohibit file cleaning. See
+			// DB.{disable,enable}FileDeletions().
+			disableCount int
+			// queuedStats holds cumulative stats for tables that have been
+			// queued for deletion by the cleanup manager. These stats are
+			// monotonically increasing for the *DB's lifetime.
+			queuedStats obsoleteTableStats
+		}
 
 		snapshots struct {
 			// The list of active snapshots.
@@ -1955,6 +1961,7 @@ func (d *DB) AsyncFlush() (<-chan struct{}, error) {
 func (d *DB) Metrics() *Metrics {
 	metrics := &Metrics{}
 	walStats := d.mu.log.manager.Stats()
+	completedCleanupStats := d.cleanupManager.CompletedStats()
 
 	d.mu.Lock()
 	vers := d.mu.versions.currentVersion()
@@ -2015,6 +2022,24 @@ func (d *DB) Metrics() *Metrics {
 		}
 	}
 	metrics.private.optionsFileSize = d.optionsFileSize
+	// The obsolete table metrics have a subtle calculation:
+	//
+	// (A) The vs.metrics.Table.[Local.]ObsoleteSize fields reflect the set of
+	// files currently sitting in vs.obsoleteTables but not yet enqueued to the
+	// cleanup manager.
+	//
+	// (B) The d.mu.fileDeletions.queuedStats field holds the set of files that
+	// have been queued for deletion by the cleanup manager.
+	//
+	// (C) The cleanup manager also maintains cumulative stats for the set of
+	// files that have been deleted.
+	//
+	// The value of currently pending obsolete files is (A) + (B) - (C).
+	pendingObsoleteFileStats := d.mu.fileDeletions.queuedStats
+	pendingObsoleteFileStats.Sub(completedCleanupStats)
+	metrics.Table.Local.ObsoleteSize += pendingObsoleteFileStats.local.size
+	metrics.Table.ObsoleteCount += int64(pendingObsoleteFileStats.total.count)
+	metrics.Table.ObsoleteSize += pendingObsoleteFileStats.total.size
 
 	// TODO(jackson): Consider making these metrics optional.
 	metrics.Keys.RangeKeySetsCount = *rangeKeySetsAnnotator.MultiLevelAnnotation(vers.RangeKeyLevels[:])

internal/invariants/off.go

@@ -42,3 +42,17 @@ func (*Value[V]) Get() V {
 
 // Store stores the value.
 func (*Value[V]) Store(v V) {}
+
+// SafeSub returns a - b. If a < b, it panics in invariant builds and returns 0
+// in non-invariant builds.
+func SafeSub[T Integer](a, b T) T {
+	if a < b {
+		return 0
+	}
+	return a - b
+}
+
+// Integer is a constraint that permits any integer type.
+type Integer interface {
+	~int | ~int8 | ~int16 | ~int32 | ~int64 | ~uint | ~uint8 | ~uint16 | ~uint32 | ~uint64 | ~uintptr
+}

internal/invariants/on.go

@@ -7,6 +7,8 @@
 
 package invariants
 
+import "fmt"
+
 // Enabled is true if we were built with the "invariants" or "race" build tags.
 const Enabled = true
 
@@ -58,3 +60,17 @@ func (v *Value[V]) Get() V {
 func (v *Value[V]) Store(inner V) {
 	v.v = inner
 }
+
+// SafeSub returns a - b. If a < b, it panics in invariant builds and returns 0
+// in non-invariant builds.
+func SafeSub[T Integer](a, b T) T {
+	if a < b {
+		panic(fmt.Sprintf("underflow: %d - %d", a, b))
+	}
+	return a - b
+}
+
+// Integer is a constraint that permits any integer type.
+type Integer interface {
+	~int | ~int8 | ~int16 | ~int32 | ~int64 | ~uint | ~uint8 | ~uint16 | ~uint32 | ~uint64 | ~uintptr
+}

obsolete_files.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/cockroachdb/errors/oserror"
 	"github.com/cockroachdb/pebble/v2/internal/base"
+	"github.com/cockroachdb/pebble/v2/internal/invariants"
 	"github.com/cockroachdb/pebble/v2/objstorage"
 	"github.com/cockroachdb/pebble/v2/vfs"
 	"github.com/cockroachdb/pebble/v2/wal"
@@ -30,10 +31,9 @@ type DeleteCleaner = base.DeleteCleaner
 type ArchiveCleaner = base.ArchiveCleaner
 
 type cleanupManager struct {
-	opts            *Options
-	objProvider     objstorage.Provider
-	onTableDeleteFn func(fileSize uint64, isLocal bool)
-	deletePacer     *deletionPacer
+	opts        *Options
+	objProvider objstorage.Provider
+	deletePacer *deletionPacer
 
 	// jobsCh is used as the cleanup job queue.
 	jobsCh chan *cleanupJob
@@ -44,6 +44,7 @@ type cleanupManager struct {
 		sync.Mutex
 		// totalJobs is the total number of enqueued jobs (completed or in progress).
 		totalJobs              int
+		completedStats         obsoleteTableStats
 		completedJobs          int
 		completedJobsCond      sync.Cond
 		jobsQueueWarningIssued bool
@@ -73,22 +74,19 @@ type obsoleteFile struct {
 type cleanupJob struct {
 	jobID         JobID
 	obsoleteFiles []obsoleteFile
+	tableStats    obsoleteTableStats
 }
 
 // openCleanupManager creates a cleanupManager and starts its background goroutine.
 // The cleanupManager must be Close()d.
 func openCleanupManager(
-	opts *Options,
-	objProvider objstorage.Provider,
-	onTableDeleteFn func(fileSize uint64, isLocal bool),
-	getDeletePacerInfo func() deletionPacerInfo,
+	opts *Options, objProvider objstorage.Provider, getDeletePacerInfo func() deletionPacerInfo,
 ) *cleanupManager {
 	cm := &cleanupManager{
-		opts:            opts,
-		objProvider:     objProvider,
-		onTableDeleteFn: onTableDeleteFn,
-		deletePacer:     newDeletionPacer(time.Now(), int64(opts.TargetByteDeletionRate), getDeletePacerInfo),
-		jobsCh:          make(chan *cleanupJob, jobsQueueDepth),
+		opts:        opts,
+		objProvider: objProvider,
+		deletePacer: newDeletionPacer(time.Now(), int64(opts.TargetByteDeletionRate), getDeletePacerInfo),
+		jobsCh:      make(chan *cleanupJob, jobsQueueDepth),
 	}
 	cm.mu.completedJobsCond.L = &cm.mu.Mutex
 	cm.waitGroup.Add(1)
@@ -102,6 +100,14 @@ func openCleanupManager(
 	return cm
 }
 
+// CompletedStats returns the stats summarizing tables deleted. The returned
+// stats increase monotonically over the lifetime of the DB.
+func (cm *cleanupManager) CompletedStats() obsoleteTableStats {
+	cm.mu.Lock()
+	defer cm.mu.Unlock()
+	return cm.mu.completedStats
+}
+
 // Close stops the background goroutine, waiting until all queued jobs are completed.
 // Delete pacing is disabled for the remaining jobs.
 func (cm *cleanupManager) Close() {
@@ -110,10 +116,13 @@ func (cm *cleanupManager) Close() {
 }
 
 // EnqueueJob adds a cleanup job to the manager's queue.
-func (cm *cleanupManager) EnqueueJob(jobID JobID, obsoleteFiles []obsoleteFile) {
+func (cm *cleanupManager) EnqueueJob(
+	jobID JobID, obsoleteFiles []obsoleteFile, tableStats obsoleteTableStats,
+) {
 	job := &cleanupJob{
 		jobID:         jobID,
 		obsoleteFiles: obsoleteFiles,
+		tableStats:    tableStats,
 	}
 
 	// Report deleted bytes to the pacer, which can use this data to potentially
@@ -165,7 +174,6 @@ func (cm *cleanupManager) mainLoop() {
 			switch of.fileType {
 			case fileTypeTable:
 				cm.maybePace(&tb, of.fileType, of.nonLogFile.fileNum, of.nonLogFile.fileSize)
-				cm.onTableDeleteFn(of.nonLogFile.fileSize, of.nonLogFile.isLocal)
 				cm.deleteObsoleteObject(fileTypeTable, job.jobID, of.nonLogFile.fileNum)
 			case fileTypeLog:
 				cm.deleteObsoleteFile(of.logFile.FS, fileTypeLog, job.jobID, of.logFile.Path,
@@ -177,6 +185,7 @@ func (cm *cleanupManager) mainLoop() {
 			}
 		}
 		cm.mu.Lock()
+		cm.mu.completedStats.Add(job.tableStats)
 		cm.mu.completedJobs++
 		cm.mu.completedJobsCond.Broadcast()
 		cm.maybeLogLocked()
@@ -322,17 +331,6 @@ func (d *DB) getDeletionPacerInfo() deletionPacerInfo {
 	return pacerInfo
 }
 
-// onObsoleteTableDelete is called to update metrics when an sstable is deleted.
-func (d *DB) onObsoleteTableDelete(fileSize uint64, isLocal bool) {
-	d.mu.Lock()
-	d.mu.versions.metrics.Table.ObsoleteCount--
-	d.mu.versions.metrics.Table.ObsoleteSize -= fileSize
-	if isLocal {
-		d.mu.versions.metrics.Table.Local.ObsoleteSize -= fileSize
-	}
-	d.mu.Unlock()
-}
-
 // scanObsoleteFiles scans the filesystem for files that are no longer needed
 // and adds those to the internal lists of obsolete files. Note that the files
 // are not actually deleted by this method. A subsequent call to
@@ -439,7 +437,7 @@ func (d *DB) scanObsoleteFiles(list []string, flushableIngests []*ingestedFlusha
 //
 // d.mu must be held when calling this method.
 func (d *DB) disableFileDeletions() {
-	d.mu.disableFileDeletions++
+	d.mu.fileDeletions.disableCount++
 	d.mu.Unlock()
 	defer d.mu.Lock()
 	d.cleanupManager.Wait()
@@ -450,11 +448,11 @@ func (d *DB) disableFileDeletions() {
 //
 // d.mu must be held when calling this method.
 func (d *DB) enableFileDeletions() {
-	if d.mu.disableFileDeletions <= 0 {
+	if d.mu.fileDeletions.disableCount <= 0 {
 		panic("pebble: file deletion disablement invariant violated")
 	}
-	d.mu.disableFileDeletions--
-	if d.mu.disableFileDeletions > 0 {
+	d.mu.fileDeletions.disableCount--
+	if d.mu.fileDeletions.disableCount > 0 {
 		return
 	}
 	d.deleteObsoleteFiles(d.newJobIDLocked())
@@ -469,7 +467,7 @@ type fileInfo = base.FileInfo
 // Does nothing if file deletions are disabled (see disableFileDeletions). A
 // cleanup job will be scheduled when file deletions are re-enabled.
 func (d *DB) deleteObsoleteFiles(jobID JobID) {
-	if d.mu.disableFileDeletions > 0 {
+	if d.mu.fileDeletions.disableCount > 0 {
 		return
 	}
 	_, noRecycle := d.opts.Cleaner.(base.NeedsFileContents)
@@ -507,6 +505,14 @@ func (d *DB) deleteObsoleteFiles(jobID JobID) {
 	obsoleteOptions := d.mu.versions.obsoleteOptions
 	d.mu.versions.obsoleteOptions = nil
 
+	// Compute the stats for the tables being queued for deletion and add them
+	// to the running total. These stats will be used during DB.Metrics() to
+	// calculate the count and size of pending obsolete tables by diffing these
+	// stats and the stats reported by the cleanup manager.
+	tableStats := calculateObsoleteTableStats(obsoleteTables)
+	d.mu.fileDeletions.queuedStats.Add(tableStats)
+	d.mu.versions.updateObsoleteTableMetricsLocked()
+
 	// Release d.mu while preparing the cleanup job and possibly waiting.
 	// Note the unusual order: Unlock and then Lock.
 	d.mu.Unlock()
@@ -560,7 +566,7 @@ func (d *DB) deleteObsoleteFiles(jobID JobID) {
 		}
 	}
 	if len(filesToDelete) > 0 {
-		d.cleanupManager.EnqueueJob(jobID, filesToDelete)
+		d.cleanupManager.EnqueueJob(jobID, filesToDelete, tableStats)
 	}
 	if d.opts.private.testingAlwaysWaitForCleanup {
 		d.cleanupManager.Wait()
@@ -579,6 +585,49 @@ func (d *DB) maybeScheduleObsoleteTableDeletionLocked() {
 	}
 }
 
+func calculateObsoleteTableStats(objects []tableInfo) obsoleteTableStats {
+	var stats obsoleteTableStats
+	for _, o := range objects {
+		if o.isLocal {
+			stats.local.count++
+			stats.local.size += o.FileSize
+		}
+		stats.total.count++
+		stats.total.size += o.FileSize
+	}
+	return stats
+}
+
+type obsoleteTableStats struct {
+	local countAndSize
+	total countAndSize
+}
+
+func (s *obsoleteTableStats) Add(other obsoleteTableStats) {
+	s.local.Add(other.local)
+	s.total.Add(other.total)
+}
+
+func (s *obsoleteTableStats) Sub(other obsoleteTableStats) {
+	s.local.Sub(other.local)
+	s.total.Sub(other.total)
+}
+
+type countAndSize struct {
+	count uint64
+	size  uint64
+}
+
+func (c *countAndSize) Add(other countAndSize) {
+	c.count += other.count
+	c.size += other.size
+}
+
+func (c *countAndSize) Sub(other countAndSize) {
+	c.count = invariants.SafeSub(c.count, other.count)
+	c.size = invariants.SafeSub(c.size, other.size)
+}
+
 func merge(a, b []fileInfo) []fileInfo {
 	if len(b) == 0 {
 		return a

open.go

@@ -397,7 +397,7 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 
 	d.mu.log.manager = walManager
 
-	d.cleanupManager = openCleanupManager(opts, d.objProvider, d.onObsoleteTableDelete, d.getDeletionPacerInfo)
+	d.cleanupManager = openCleanupManager(opts, d.objProvider, d.getDeletionPacerInfo)
 
 	if manifestExists && !opts.DisableConsistencyCheck {
 		curVersion := d.mu.versions.currentVersion()