db: add fail-safe to the delete pacer

When we reach 3/4 of the capacity of the deletion jobs channel, we
disable pacing. This should act as a catch-all if the delete pacing
logic fails to keep up.

Informs #5424

Author: RaduBerinde

obsolete_files.go

@@ -65,6 +65,8 @@ func (m *cleanupManager) CompletedStats() obsoleteObjectStats {
 
 // We can queue this many jobs before we have to block EnqueueJob.
 const jobsQueueDepth = 1000
+const jobsQueueHighThreshold = jobsQueueDepth * 3 / 4
+const jobsQueueLowThreshold = jobsQueueDepth / 10
 
 // obsoleteFile holds information about a file that needs to be deleted soon.
 type obsoleteFile struct {
@@ -223,7 +225,11 @@ func (cm *cleanupManager) maybePace(
 	if !of.needsPacing() {
 		return
 	}
-
+	if len(cm.jobsCh) >= jobsQueueHighThreshold {
+		// If there are many jobs queued up, disable pacing. In this state, we
+		// execute deletion jobs at the same rate as new jobs get queued.
+		return
+	}
 	tokens := cm.deletePacer.PacingDelay(crtime.NowMono(), of.fileSize)
 	if tokens == 0.0 {
 		// The token bucket might be in debt; it could make us wait even for 0
@@ -321,19 +327,17 @@ func (cm *cleanupManager) deleteObsoleteObject(
 //
 // Must be called with cm.mu locked.
 func (cm *cleanupManager) maybeLogLocked() {
-	const highThreshold = jobsQueueDepth * 3 / 4
-	const lowThreshold = jobsQueueDepth / 10
 
 	jobsInQueue := cm.mu.totalJobs - cm.mu.completedJobs
 
-	if !cm.mu.jobsQueueWarningIssued && jobsInQueue > highThreshold {
+	if !cm.mu.jobsQueueWarningIssued && jobsInQueue > jobsQueueHighThreshold {
 		cm.mu.jobsQueueWarningIssued = true
-		cm.opts.Logger.Infof("cleanup falling behind; job queue has over %d jobs", highThreshold)
+		cm.opts.Logger.Infof("cleanup falling behind; job queue has over %d jobs", jobsQueueHighThreshold)
 	}
 
-	if cm.mu.jobsQueueWarningIssued && jobsInQueue < lowThreshold {
+	if cm.mu.jobsQueueWarningIssued && jobsInQueue < jobsQueueLowThreshold {
 		cm.mu.jobsQueueWarningIssued = false
-		cm.opts.Logger.Infof("cleanup back to normal; job queue has under %d jobs", lowThreshold)
+		cm.opts.Logger.Infof("cleanup back to normal; job queue has under %d jobs", jobsQueueLowThreshold)
 	}
 }
 

obsolete_files_test.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"sort"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -195,3 +196,77 @@ func TestCleanupManagerCloseWithPacing(t *testing.T) {
 		t.Fatalf("timed out waiting for cleanupManager.Close() to return")
 	}
 }
+
+// TestCleanupManagerFallingBehind verifies that we disable pacing when the jobs
+// channel reaches the high threshold.
+func TestCleanupManagerFallingBehind(t *testing.T) {
+	mem := vfs.NewMem()
+	var rate atomic.Int32
+	rate.Store(10 * MB) // 10MB/s
+	opts := &Options{
+		FS:                      mem,
+		FreeSpaceThresholdBytes: 1,
+		TargetByteDeletionRate:  func() int { return int(rate.Load()) }, // 10 MB/s
+	}
+	opts.EnsureDefaults()
+
+	objProvider, err := objstorageprovider.Open(objstorageprovider.Settings{
+		FS:        mem,
+		FSDirName: "/",
+	})
+	require.NoError(t, err)
+	defer objProvider.Close()
+
+	getDeletePacerInfo := func() deletionPacerInfo {
+		return deletionPacerInfo{
+			freeBytes: 10 * GB,
+			liveBytes: 10 * GB,
+		}
+	}
+
+	cm := openCleanupManager(opts, objProvider, getDeletePacerInfo)
+
+	x := 0
+	addJob := func(fileSize int) {
+		x++
+		cm.EnqueueJob(1, []obsoleteFile{{
+			fileType: base.FileTypeTable,
+			fs:       mem,
+			path:     fmt.Sprintf("test%02d.sst", x),
+			fileNum:  base.DiskFileNum(x),
+			fileSize: uint64(fileSize),
+			isLocal:  true,
+		}}, obsoleteObjectStats{})
+	}
+
+	for range jobsQueueLowThreshold {
+		addJob(1 * MB)
+	}
+	// At 1MB, each job will take 100ms each. Note that the rate increase based on
+	// history won't make much difference, since the enqueued size is averaged
+	// over 5 minutes.
+	time.Sleep(50 * time.Millisecond)
+	require.Greater(t, len(cm.jobsCh), jobsQueueLowThreshold/2)
+	t.Logf("%d", len(cm.jobsCh))
+
+	// Add enough jobs to exceed the high threshold. We add small jobs so that the
+	// historic rate doesn't grow significantly.
+	require.Greater(t, jobsQueueDepth, jobsQueueHighThreshold+jobsQueueLowThreshold)
+	t.Logf("B")
+	for range jobsQueueHighThreshold {
+		addJob(1)
+	}
+
+	for i := 0; ; i++ {
+		time.Sleep(10 * time.Millisecond)
+		if len(cm.jobsCh) <= jobsQueueHighThreshold {
+			break
+		}
+		if i == 1000 {
+			t.Fatalf("jobs channel length never dropped below high threshold (%d vs %d)", len(cm.jobsCh), jobsQueueHighThreshold)
+		}
+	}
+	// Set a high rate so the rest of the jobs finish quickly.
+	rate.Store(1 * GB)
+	cm.Close()
+}