Repository: cockroachdb/pebble

Issue:
db: Allow to rename `WALDir`
There might be a situation, when a directory with WAL will be lost or moved to another mount point.
Currently, `open` fails with error `WALDir changed from previous options`.

**Suggestion:**
Allow to skip this check with additional option `SkipCheckCompatibility` or introduce new public method `pebble.RenameWALDir`

Jira issue: PEBBLE-1214

Discussion:
@annikiforov: How about this new function:
```
func RenameWALDir(dirname, walDir string) error {
	var opts *Options
	opts = opts.Clone()
	opts.EnsureDefaults()

	ls, err := opts.FS.List(dirname)
	if err != nil {
		return err
	}
	var previousOptionsFileNum base.DiskFileNum
	var previousOptionsFilename string
	for _, filename := range ls {
		ft, fn, ok := base.ParseFilename(opts.FS, filename)
		if !ok {
			continue
		}
		if ft == base.FileTypeOptions {
			previousOptionsFileNum = fn
			previousOptionsFilename = filename
			break
		}
	}
	path := opts.FS.PathJoin(dirname, previousOptionsFilename)

	previousOptions, err := readOptionsFile(opts, path)
	opts.Parse(previousOptions, nil)
	opts.WALDir = walDir

	serializedOpts := []byte(opts.String())
	tmpPath := base.MakeFilepath(opts.FS, dirname, base.FileTypeTemp, previousOptionsFileNum)
	optionsFile, err := opts.FS.Create(tmpPath, vfs.WriteCategoryUnspecified)
	if err != nil {
		return err
	}
	if _, err := optionsFile.Write(serializedOpts); err != nil {
		return errors.CombineErrors(err, optionsFile.Close())
	}
	if err := optionsFile.Sync(); err != nil {
		return errors.CombineErrors(err, optionsFile.Close())
	}
	if err := optionsFile.Close(); err != nil {
		return err
	}
	if err := opts.FS.Rename(tmpPath, previousOptionsFilename); err != nil {
		return err
	}
	dataDir, err := opts.FS.OpenDir(dirname)
	if err != nil {
		return err
	}
	err = dataDir.Sync()
	if err != nil {
		return err
	}
	return dataDir.Close()
}

```@annikiforov: I've opened the PR #5423, just for test purposes. Sorry, unable to find info how to contribute to `pebble`.@RaduBerinde: You are allowed to change the WAL dir as long as you put the previous dir in `WALRecoveryDirs`. Is this not sufficient?@annikiforov: > You are allowed to change the WAL dir as long as you put the previous dir in `WALRecoveryDirs`. Is this not sufficient?

Unfortunately no. `WALRecoveryDirs` required to have such directories on disk, and they should be not empty. I'm trying to address the case when `WALDir` lost with the disk, and it is needed to start from with the new and empty `WALDir`.@RaduBerinde: I thought it's ok for recovery dirs to be empty. ~In any case, that should cover the "rename" feature no?~

As far as WALs being lost, do you want to open the store even though data may be lost in an inconsistent fashion?  That would definitely be something we never want to happen by accident (misconfiguration)@annikiforov: Yes, I see that opening DB without a WAL may lead to data loss. But when WAL is on a SSD and SST is on an HDD and SSD broken, renaming the WALDir  - is a last chance to read something from SST.
Maybe that can be cover somehow by some "unsafe_" option?
Here is a test to illustrate the situation:
```
func TestPebble(t *testing.T) {
	dir := t.TempDir()
	walDir0 := t.TempDir()
	walDir1 := t.TempDir()

	require.NoError(t, os.MkdirAll(dir, os.ModePerm|0o110))
	require.NoError(t, os.MkdirAll(walDir0, os.ModePerm|0o110))
	require.NoError(t, os.MkdirAll(walDir1, os.ModePerm|0o110))

	opts := &pebble.Options{
		ReadOnly: false,
		WALDir:   walDir0,
	}
	db, err := pebble.Open(dir, opts)
	require.NoError(t, err)
	require.NoError(t, db.Set([]byte("test"), []byte("test"), nil))
	require.NoError(t, db.Close())

	walRecDirs := []wal.Dir{
		{
			FS:      vfs.Default,
			Dirname: walDir0,
		},
	}
	// Simulate data loss on SSD
	require.NoError(t, os.RemoveAll(walDir0))
	opts = &pebble.Options{
		ReadOnly:        true,
		WALDir:          walDir1,
		WALRecoveryDirs: walRecDirs,
	}

	db, err = pebble.Open(dir, opts)
	require.NoError(t, err)
}
```@RaduBerinde: Ok but if you recreate the directory and leave it empty, it works:
```diff
        require.NoError(t, os.RemoveAll(walDir0))
+       require.NoError(t, os.MkdirAll(walDir0, os.ModePerm|0o110))
        opts = &pebble.Options{
```
Is this not sufficient?@annikiforov: In this case, it is necessary to set `WALRecoveryDirs` always, because we will face an error:
```
directory "/tmp/TestPebble1529425533/002" may contain relevant WALs but is not in WALRecoveryDirs
        	            	  WALDir changed from previous options
        	            	  o.WALDir: "/tmp/TestPebble1529425533/003"
        	            	  o.WALRecoveryDirs: 0
```
```diff
	// Simulate data loss on SSD
	require.NoError(t, os.RemoveAll(walDir0))
+	require.NoError(t, os.MkdirAll(walDir0, os.ModePerm|0o110))
	opts = &pebble.Options{
		ReadOnly:        true,
		WALDir:          walDir1,
		WALRecoveryDirs: walRecDirs,
	}

	db, err = pebble.Open(dir, opts)
	require.NoError(t, err)
+	require.NoError(t, db.Close())
+
+	opts = &pebble.Options{
+		ReadOnly: true,
+		WALDir:   walDir1,
+	}
+
+	db, err = pebble.Open(dir, opts)
+	require.NoError(t, err)
```
In case of enclosure with hundred of HDDs it may be a bit unuseful, enclosure may be connected to the different host without the secret knowledge about of previous settings.@RaduBerinde: I see. Yes we need to rethink some of this. For now we can add an `Unsafe.AllowMissingWALDirs` option. What branch/release are you using?@annikiforov: > What branch/release are you using?

v2.1.0, with v2 path.
@RaduBerinde: Tagged v2.1.1 which includes the new flag.

Solution patch:
diff --git a/blob_rewrite.go b/blob_rewrite.go
index 07a3b792..8c1f8191 100644
--- a/blob_rewrite.go
+++ b/blob_rewrite.go
@@ -16,6 +16,7 @@ import (
 
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/deletepacer"
 	"github.com/cockroachdb/pebble/internal/manifest"
 	"github.com/cockroachdb/pebble/internal/problemspans"
 	"github.com/cockroachdb/pebble/objstorage"
@@ -249,17 +250,17 @@ func (c *blobFileRewriteCompaction) Execute(jobID JobID, d *DB) error {
 	// Ensure we clean up the blob file we created on failure.
 	if err != nil {
 		if objMeta.DiskFileNum != 0 {
-			d.mu.versions.obsoleteBlobs = mergeObsoleteFiles(d.mu.versions.obsoleteBlobs, []obsoleteFile{
+			d.mu.versions.obsoleteBlobs = mergeObsoleteFiles(d.mu.versions.obsoleteBlobs, []deletepacer.ObsoleteFile{
 				{
-					fileType: base.FileTypeBlob,
-					fs:       d.opts.FS,
-					path:     d.objProvider.Path(objMeta),
-					fileNum:  objMeta.DiskFileNum,
+					FileType: base.FileTypeBlob,
+					FS:       d.opts.FS,
+					Path:     d.objProvider.Path(objMeta),
+					FileNum:  objMeta.DiskFileNum,
 					// We don't know the size of the output blob file--it may have
 					// been half-written. We use the input blob file size as an
 					// approximation for deletion pacing.
-					fileSize: c.input.Physical.Size,
-					isLocal:  true,
+					FileSize: c.input.Physical.Size,
+					IsLocal:  true,
 				},
 			})
 		}
diff --git a/compaction.go b/compaction.go
index 04dc6f05..e53ac622 100644
--- a/compaction.go
+++ b/compaction.go
@@ -34,6 +34,7 @@ import (
 	"github.com/cockroachdb/pebble/sstable/blob"
 	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/sstable/block/blockkind"
+	"github.com/cockroachdb/pebble/valsep"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/cockroachdb/redact"
 )
@@ -45,7 +46,6 @@ var errEmptyTable = errors.New("pebble: empty table")
 var ErrCancelledCompaction = errors.New("pebble: compaction cancelled by a concurrent operation, will retry compaction")
 
 var flushLabels = pprof.Labels("pebble", "flush", "output-level", "L0")
-var gcLabels = pprof.Labels("pebble", "gc")
 
 // expandedCompactionByteSizeLimit is the maximum number of bytes in all
 // compacted files. We avoid expanding the lower level file set of a compaction
@@ -250,7 +250,7 @@ type tableCompaction struct {
 	// b) rewrite blob files: The compaction will write eligible values to new
 	// blob files. This consumes more write bandwidth because all values are
 	// rewritten. However it restores locality.
-	getValueSeparation func(JobID, *tableCompaction, ValueStoragePolicy) compact.ValueSeparation
+	getValueSeparation func(JobID, *tableCompaction, ValueStoragePolicy) valsep.ValueSeparation
 
 	// startLevel is the level that is being compacted. Inputs from startLevel
 	// and outputLevel will be merged to produce a set of outputLevel files.
@@ -542,7 +542,7 @@ func (c *tableCompaction) makeInfo(jobID JobID) CompactionInfo {
 	return info
 }
 
-type getValueSeparation func(JobID, *tableCompaction, ValueStoragePolicy) compact.ValueSeparation
+type getValueSeparation func(JobID, *tableCompaction, ValueStoragePolicy) valsep.ValueSeparation
 
 // newCompaction constructs a compaction from the provided picked compaction.
 //
@@ -3461,11 +3461,11 @@ func (d *DB) compactAndWrite(
 		vSep := valueSeparation
 		switch spanPolicy.ValueStoragePolicy {
 		case ValueStorageLowReadLatency:
-			vSep = compact.NeverSeparateValues{}
+			vSep = valsep.NeverSeparateValues{}
 		case ValueStorageLatencyTolerant:
 			// This span of keyspace is more tolerant of latency, so set a more
 			// aggressive value separation policy for this output.
-			vSep.SetNextOutputConfig(compact.ValueSeparationOutputConfig{
+			vSep.SetNextOutputConfig(valsep.ValueSeparationOutputConfig{
 				MinimumSize: latencyTolerantMinimumSize,
 			})
 		}
diff --git a/value_separation.go b/compaction_value_separation.go
similarity index 94%
rename from value_separation.go
rename to compaction_value_separation.go
index a5028a6d..6c726884 100644
--- a/value_separation.go
+++ b/compaction_value_separation.go
@@ -10,12 +10,12 @@ import (
 
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
-	"github.com/cockroachdb/pebble/internal/compact"
 	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/manifest"
 	"github.com/cockroachdb/pebble/objstorage"
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/sstable/blob"
+	"github.com/cockroachdb/pebble/valsep"
 	"github.com/cockroachdb/redact"
 )
 
@@ -24,8 +24,8 @@ import (
 // ValueStorageLatencyTolerant.
 const latencyTolerantMinimumSize = 10
 
-var neverSeparateValues getValueSeparation = func(JobID, *tableCompaction, ValueStoragePolicy) compact.ValueSeparation {
-	return compact.NeverSeparateValues{}
+var neverSeparateValues getValueSeparation = func(JobID, *tableCompaction, ValueStoragePolicy) valsep.ValueSeparation {
+	return valsep.NeverSeparateValues{}
 }
 
 // determineCompactionValueSeparation determines whether a compaction should
@@ -35,14 +35,14 @@ var neverSeparateValues getValueSeparation = func(JobID, *tableCompaction, Value
 // It assumes that the compaction will write tables at d.TableFormat() or above.
 func (d *DB) determineCompactionValueSeparation(
 	jobID JobID, c *tableCompaction, valueStorage ValueStoragePolicy,
-) compact.ValueSeparation {
+) valsep.ValueSeparation {
 	if d.FormatMajorVersion() < FormatValueSeparation ||
 		d.opts.Experimental.ValueSeparationPolicy == nil {
-		return compact.NeverSeparateValues{}
+		return valsep.NeverSeparateValues{}
 	}
 	policy := d.opts.Experimental.ValueSeparationPolicy()
 	if !policy.Enabled {
-		return compact.NeverSeparateValues{}
+		return valsep.NeverSeparateValues{}
 	}
 
 	// We're allowed to write blob references. Determine whether we should carry
@@ -50,7 +50,7 @@ func (d *DB) determineCompactionValueSeparation(
 	minSize := uint64(policy.MinimumSize)
 	switch valueStorage {
 	case ValueStorageLowReadLatency:
-		return compact.NeverSeparateValues{}
+		return valsep.NeverSeparateValues{}
 	case ValueStorageLatencyTolerant:
 		minSize = latencyTolerantMinimumSize
 	default:
@@ -249,10 +249,10 @@ type writeNewBlobFiles struct {
 }
 
 // Assert that *writeNewBlobFiles implements the compact.ValueSeparation interface.
-var _ compact.ValueSeparation = (*writeNewBlobFiles)(nil)
+var _ valsep.ValueSeparation = (*writeNewBlobFiles)(nil)
 
 // SetNextOutputConfig implements the ValueSeparation interface.
-func (vs *writeNewBlobFiles) SetNextOutputConfig(config compact.ValueSeparationOutputConfig) {
+func (vs *writeNewBlobFiles) SetNextOutputConfig(config valsep.ValueSeparationOutputConfig) {
 	vs.minimumSize = config.MinimumSize
 }
 
@@ -372,13 +372,13 @@ func (vs *writeNewBlobFiles) Add(
 
 // FinishOutput closes the current blob file (if any). It returns the stats
 // and metadata of the now completed blob file.
-func (vs *writeNewBlobFiles) FinishOutput() (compact.ValueSeparationMetadata, error) {
+func (vs *writeNewBlobFiles) FinishOutput() (valsep.ValueSeparationMetadata, error) {
 	if vs.writer == nil {
-		return compact.ValueSeparationMetadata{}, nil
+		return valsep.ValueSeparationMetadata{}, nil
 	}
 	stats, err := vs.writer.Close()
 	if err != nil {
-		return compact.ValueSeparationMetadata{}, err
+		return valsep.ValueSeparationMetadata{}, err
 	}
 	vs.writer = nil
 	meta := &manifest.PhysicalBlobFile{
@@ -391,16 +391,18 @@ func (vs *writeNewBlobFiles) FinishOutput() (compact.ValueSeparationMetadata, er
 	// Reset the minimum size for the next output.
 	vs.minimumSize = vs.globalMinimumSize
 
-	return compact.ValueSeparationMetadata{
+	return valsep.ValueSeparationMetadata{
 		BlobReferences: manifest.BlobReferences{
 			manifest.MakeBlobReference(base.BlobFileID(vs.objMeta.DiskFileNum),
 				stats.UncompressedValueBytes, stats.UncompressedValueBytes, meta),
 		},
 		BlobReferenceSize:  stats.UncompressedValueBytes,
 		BlobReferenceDepth: 1,
-		BlobFileStats:      stats,
-		BlobFileObject:     vs.objMeta,
-		BlobFileMetadata:   meta,
+		NewBlobFiles: []valsep.NewBlobFileInfo{{
+			FileStats:    stats,
+			FileObject:   vs.objMeta,
+			FileMetadata: meta,
+		}},
 	}, nil
 }
 
@@ -442,10 +444,10 @@ type pendingReference struct {
 
 // Assert that *preserveBlobReferences implements the compact.ValueSeparation
 // interface.
-var _ compact.ValueSeparation = (*preserveBlobReferences)(nil)
+var _ valsep.ValueSeparation = (*preserveBlobReferences)(nil)
 
 // SetNextOutputConfig implements the ValueSeparation interface.
-func (vs *preserveBlobReferences) SetNextOutputConfig(config compact.ValueSeparationOutputConfig) {}
+func (vs *preserveBlobReferences) SetNextOutputConfig(config valsep.ValueSeparationOutputConfig) {}
 
 // Kind implements the ValueSeparation interface.
 func (vs *preserveBlobReferences) Kind() sstable.ValueSeparationKind {
@@ -535,7 +537,7 @@ func (vs *preserveBlobReferences) Add(
 }
 
 // FinishOutput implements compact.ValueSeparation.
-func (vs *preserveBlobReferences) FinishOutput() (compact.ValueSeparationMetadata, error) {
+func (vs *preserveBlobReferences) FinishOutput() (valsep.ValueSeparationMetadata, error) {
 	if invariants.Enabled {
 		// Assert that the incrementally-maintained totalValueSize matches the
 		// sum of all the reference value sizes.
@@ -544,7 +546,7 @@ func (vs *preserveBlobReferences) FinishOutput() (compact.ValueSeparationMetadat
 			totalValueSize += ref.valueSize
 		}
 		if totalValueSize != vs.totalValueSize {
-			return compact.ValueSeparationMetadata{},
+			return valsep.ValueSeparationMetadata{},
 				errors.AssertionFailedf("totalValueSize mismatch: %d != %d", totalValueSize, vs.totalValueSize)
 		}
 	}
@@ -554,7 +556,7 @@ func (vs *preserveBlobReferences) FinishOutput() (compact.ValueSeparationMetadat
 		ref := vs.currReferences[i]
 		phys, ok := vs.inputBlobPhysicalFiles[ref.blobFileID]
 		if !ok {
-			return compact.ValueSeparationMetadata{},
+			return valsep.ValueSeparationMetadata{},
 				errors.AssertionFailedf("pebble: blob file %s not found among input sstables", ref.blobFileID)
 		}
 		references[i] = manifest.MakeBlobReference(ref.blobFileID, ref.valueSize, ref.valueSize, phys)
@@ -562,7 +564,7 @@ func (vs *preserveBlobReferences) FinishOutput() (compact.ValueSeparationMetadat
 	referenceSize := vs.totalValueSize
 	vs.currReferences = vs.currReferences[:0]
 	vs.totalValueSize = 0
-	return compact.ValueSeparationMetadata{
+	return valsep.ValueSeparationMetadata{
 		BlobReferences:    references,
 		BlobReferenceSize: referenceSize,
 		// The outputBlobReferenceDepth is computed from the input sstables,
diff --git a/db.go b/db.go
index 8abde97c..411b7a01 100644
--- a/db.go
+++ b/db.go
@@ -21,6 +21,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/arenaskl"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/cache"
+	"github.com/cockroachdb/pebble/internal/deletepacer"
 	"github.com/cockroachdb/pebble/internal/inflight"
 	"github.com/cockroachdb/pebble/internal/invalidating"
 	"github.com/cockroachdb/pebble/internal/invariants"
@@ -36,7 +37,6 @@ import (
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/sstable/blob"
 	"github.com/cockroachdb/pebble/sstable/block"
-	"github.com/cockroachdb/pebble/vfs"
 	"github.com/cockroachdb/pebble/vfs/atomicfs"
 	"github.com/cockroachdb/pebble/wal"
 	"github.com/cockroachdb/tokenbucket"
@@ -293,8 +293,7 @@ type DB struct {
 	// objProvider is used to access and manage SSTs.
 	objProvider objstorage.Provider
 
-	dataDirLock *base.DirLock
-	dataDir     vfs.File
+	dirs *resolvedDirs
 
 	fileCache            *fileCacheHandle
 	newIters             tableNewIters
@@ -312,7 +311,7 @@ type DB struct {
 	closed   *atomic.Value
 	closedCh chan struct{}
 
-	cleanupManager *cleanupManager
+	deletePacer *deletepacer.DeletePacer
 
 	compactionScheduler CompactionScheduler
 
@@ -487,14 +486,14 @@ type DB struct {
 			// count acts as a reference count to prohibit file cleaning. See
 			// DB.{disable,enable}FileDeletions().
 			disableCount int
-			// queuedStats holds cumulative stats for files that have been
-			// queued for deletion by the cleanup manager. These stats are
-			// monotonically increasing for the *DB's lifetime.
-			queuedStats obsoleteObjectStats
 		}
 
 		snapshots struct {
-			// The list of active snapshots.
+			// The list of active snapshots, including EFOS snapshots that have not
+			// transitioned. An EFOS snapshot that transitions to FOS is removed
+			// from this list atomically with the transition (using DB.mu). A closed
+			// snapshot (including an EFOS snapshot) may still be in this list
+			// transiently, since the removal is not atomic with the close.
 			snapshotList
 
 			// The cumulative count and size of snapshot-pinned keys written to
@@ -559,7 +558,7 @@ var _ Writer = (*DB)(nil)
 
 // TestOnlyWaitForCleaning MUST only be used in tests.
 func (d *DB) TestOnlyWaitForCleaning() {
-	d.cleanupManager.Wait()
+	d.deletePacer.WaitForTesting()
 }
 
 // Set sets the value for the given key. It overwrites any previous value
@@ -1547,14 +1546,11 @@ func (d *DB) Close() error {
 		panic("pebble: log-writer should be nil in read-only mode")
 	}
 	err = firstError(err, d.mu.log.manager.Close())
-	err = firstError(err, d.dataDirLock.Close())
 
 	// Note that versionSet.close() only closes the MANIFEST. The versions list
 	// is still valid for the checks below.
 	err = firstError(err, d.mu.versions.close())
 
-	err = firstError(err, d.dataDir.Close())
-
 	d.readState.val.unrefLocked()
 
 	current := d.mu.versions.currentVersion()
@@ -1595,7 +1591,7 @@ func (d *DB) Close() error {
 	d.mu.Unlock()
 
 	// Wait for all cleaning jobs to finish.
-	d.cleanupManager.Close()
+	d.deletePacer.Close()
 
 	d.mu.Lock()
 	// Sanity check compaction metrics.
@@ -1627,6 +1623,7 @@ func (d *DB) Close() error {
 	if v := d.mu.snapshots.count(); v > 0 {
 		err = firstError(err, errors.Errorf("leaked snapshots: %d open snapshots on DB %p", v, d))
 	}
+	err = firstError(err, d.dirs.Close())
 
 	if d.iterTracker != nil {
 		d.iterTracker.Close()
@@ -1883,7 +1880,7 @@ func (d *DB) AsyncFlush() (<-chan struct{}, error) {
 func (d *DB) Metrics() *Metrics {
 	metrics := &Metrics{}
 	walStats := d.mu.log.manager.Stats()
-	completedObsoleteFileStats := d.cleanupManager.CompletedStats()
+	deletePacerMetrics := d.deletePacer.Metrics()
 
 	d.mu.Lock()
 	vers := d.mu.versions.currentVersion()
@@ -1947,29 +1944,19 @@ func (d *DB) Metrics() *Metrics {
 	metrics.Table.ZombieSize = d.mu.versions.zombieTables.TotalSize()
 	metrics.Table.Local.ZombieCount, metrics.Table.Local.ZombieSize = d.mu.versions.zombieTables.LocalStats()
 
-	// The obsolete blob/table metrics have a subtle calculation:
-	//
-	// (A) The vs.metrics.{Table,BlobFiles}.[Local.]{ObsoleteCount,ObsoleteSize}
-	// fields reflect the set of files currently sitting in
-	// vs.obsolete{Tables,Blobs} 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(completedObsoleteFileStats)
-	metrics.Table.Local.ObsoleteCount += pendingObsoleteFileStats.tablesLocal.count
-	metrics.Table.Local.ObsoleteSize += pendingObsoleteFileStats.tablesLocal.size
-	metrics.Table.ObsoleteCount += int64(pendingObsoleteFileStats.tablesAll.count)
-	metrics.Table.ObsoleteSize += pendingObsoleteFileStats.tablesAll.size
-	metrics.BlobFiles.Local.ObsoleteCount += pendingObsoleteFileStats.blobFilesLocal.count
-	metrics.BlobFiles.Local.ObsoleteSize += pendingObsoleteFileStats.blobFilesLocal.size
-	metrics.BlobFiles.ObsoleteCount += pendingObsoleteFileStats.blobFilesAll.count
-	metrics.BlobFiles.ObsoleteSize += pendingObsoleteFileStats.blobFilesAll.size
+	// The obsolete blob/table metrics have a subtle calculation: the initial
+	// metrics we copied from vs.metrics reflect (in
+	// {Table,BlobFiles}.[Local.]{ObsoleteCount,ObsoleteSize}) the set of files
+	// currently sitting in vs.obsolete{Tables,Blobs} but not yet enqueued to the
+	// delete pacer. To complete the metrics, we add the currently enqueued files.
+	metrics.Table.Local.ObsoleteCount += deletePacerMetrics.InQueue.Tables.Local.Count
+	metrics.Table.Local.ObsoleteSize += deletePacerMetrics.InQueue.Tables.Local.Bytes
+	metrics.Table.ObsoleteCount += int64(deletePacerMetrics.InQueue.Tables.All.Count)
+	metrics.Table.ObsoleteSize += deletePacerMetrics.InQueue.Tables.All.Bytes
+	metrics.BlobFiles.Local.ObsoleteCount += deletePacerMetrics.InQueue.BlobFiles.Local.Count
+	metrics.BlobFiles.Local.ObsoleteSize += deletePacerMetrics.InQueue.BlobFiles.Local.Bytes
+	metrics.BlobFiles.ObsoleteCount += deletePacerMetrics.InQueue.BlobFiles.All.Count
+	metrics.BlobFiles.ObsoleteSize += deletePacerMetrics.InQueue.BlobFiles.All.Bytes
 	metrics.private.optionsFileSize = d.optionsFileSize
 
 	d.mu.versions.logLock()
@@ -2022,6 +2009,8 @@ func (d *DB) Metrics() *Metrics {
 	metrics.TableIters = d.fileCache.IterCount()
 	metrics.CategoryStats = d.fileCache.SSTStatsCollector().GetStats()
 
+	metrics.DeletePacer = deletePacerMetrics
+
 	metrics.SecondaryCacheMetrics = d.objProvider.Metrics()
 
 	metrics.Uptime = d.opts.private.timeNow().Sub(d.openedAt)
diff --git a/ingest.go b/ingest.go
index 703bb5be..1eb36076 100644
--- a/ingest.go
+++ b/ingest.go
@@ -2216,8 +2216,16 @@ func (d *DB) ingestApply(
 	if exciseSpan.Valid() {
 		for s := d.mu.snapshots.root.next; s != &d.mu.snapshots.root; s = s.next {
 			// Skip non-EFOS snapshots, and also skip any EFOS that were created
-			// *after* the excise.
-			if s.efos == nil || base.Visible(exciseSeqNum, s.efos.seqNum, base.SeqNumMax) {
+			// *after* the excise, or are already closed.
+			//
+			// NB: this code can race with EventuallyFileOnlySnapshot.Close, which
+			// closes the channel that isClosed() reads, without holding DB.mu. This
+			// is harmless: if this code observes !isClosed, that means the EFOS was
+			// not closed until after the excise has already updated the LSM state.
+			// It is then fair to confirm that the EFOS did not overlap with the
+			// excise.
+			if s.efos == nil || base.Visible(exciseSeqNum, s.efos.seqNum, base.SeqNumMax) ||
+				s.efos.isClosed() {
 				continue
 			}
 			efos := s.efos
@@ -2227,7 +2235,13 @@ func (d *DB) ingestApply(
 			// snapshot.
 			for i := range efos.protectedRanges {
 				if efos.protectedRanges[i].OverlapsKeyRange(d.cmp, exciseSpan) {
-					panic("unexpected excise of an EventuallyFileOnlySnapshot's bounds")
+					panic(errors.AssertionFailedf(
+						"unexpected excise of an EventuallyFileOnlySnapshot's bounds: "+
+							"excise [%s, %s) seqnum: %s, efos [%s, %s) seqnum: %s transitioned: %t closed: %t visible: %s",
+						d.opts.Comparer.FormatKey(exciseSpan.Start), d.opts.Comparer.FormatKey(exciseSpan.End),
+						exciseSeqNum, d.opts.Comparer.FormatKey(efos.protectedRanges[i].Start),
+						d.opts.Comparer.FormatKey(efos.protectedRanges[i].End), efos.seqNum,
+						efos.hasTransitioned(), efos.isClosed(), d.mu.versions.visibleSeqNum.Load()))
 				}
 			}
 		}
diff --git a/internal/base/directory_lock.go b/internal/base/directory_lock.go
index 431a8539..e9476803 100644
--- a/internal/base/directory_lock.go
+++ b/internal/base/directory_lock.go
@@ -5,8 +5,10 @@
 package base
 
 import (
+	"fmt"
 	"io"
 	"os"
+	"strings"
 	"sync/atomic"
 
 	"github.com/cockroachdb/errors"
@@ -14,9 +16,37 @@ import (
 	"github.com/cockroachdb/pebble/vfs"
 )
 
-// AcquireOrValidateDirectoryLock attempts to acquire a lock on the
-// provided directory, or validates a pre-acquired DirLock.
-func AcquireOrValidateDirectoryLock(
+// DirLockSet is a set of directory locks that have been acquired.
+type DirLockSet struct {
+	acquired []*DirLock
+}
+
+func (s *DirLockSet) String() string {
+	var buf strings.Builder
+	buf.WriteString("{")
+	for i, l := range s.acquired {
+		if i > 0 {
+			buf.WriteString(", ")
+		}
+		fmt.Fprint(&buf, l.dirname)
+	}
+	buf.WriteString("}")
+	return buf.String()
+}
+
+// Close releases all of the directory locks in the set.
+func (s *DirLockSet) Close() error {
+	var err error
+	for _, l := range s.acquired {
+		err = errors.CombineErrors(err, l.Close())
+	}
+	s.acquired = nil
+	return err
+}
+
+// AcquireOrValidate attempts to acquire a lock on the provided directory, or
+// validates a pre-acquired DirLock.
+func (s *DirLockSet) AcquireOrValidate(
 	preAcquiredLock *DirLock, dirname string, fs vfs.FS,
 ) (*DirLock, error) {
 	// If a pre-acquired lock is provided, check that it matches the directory
@@ -25,11 +55,20 @@ func AcquireOrValidateDirectoryLock(
 		if err := preAcquiredLock.pathMatches(dirname); err != nil {
 			return preAcquiredLock, err
 		}
-		return preAcquiredLock, preAcquiredLock.refForOpen()
+		if err := preAcquiredLock.refForOpen(); err != nil {
+			return preAcquiredLock, err
+		}
+		s.acquired = append(s.acquired, preAcquiredLock)
+		return preAcquiredLock, nil
 	}
 
 	// Otherwise, acquire the lock for the directory.
-	return LockDirectory(dirname, fs)
+	l, err := LockDirectory(dirname, fs)
+	if err != nil {
+		return nil, err
+	}
+	s.acquired = append(s.acquired, l)
+	return l, nil
 }
 
 // LockDirectory acquires the directory lock in the named directory, preventing
diff --git a/internal/compact/run.go b/internal/compact/run.go
index 0012717d..d61b6f3b 100644
--- a/internal/compact/run.go
+++ b/internal/compact/run.go
@@ -15,6 +15,7 @@ import (
 	"github.com/cockroachdb/pebble/objstorage"
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/sstable/blob"
+	"github.com/cockroachdb/pebble/valsep"
 )
 
 // Result stores the result of a compaction - more specifically, the "data" part
@@ -109,56 +110,6 @@ type RunnerConfig struct {
 	GrantHandle base.CompactionGrantHandle
 }
 
-// ValueSeparationOutputConfig is used to configure value separation for an
-// individual compaction output.
-type ValueSeparationOutputConfig struct {
-	// MinimumSize is the minimum size of a value that will be separated into a
-	// blob file.
-	MinimumSize int
-}
-
-// ValueSeparation defines an interface for writing some values to separate blob
-// files.
-type ValueSeparation interface {
-	// SetNextOutputConfig is called when a compaction is starting a new output
-	// sstable. It can be used to configure value separation specifically for
-	// the next compaction output.
-	SetNextOutputConfig(config ValueSeparationOutputConfig)
-	// Kind returns the kind of value separation strategy being used.
-	Kind() sstable.ValueSeparationKind
-	// MinimumSize returns the minimum size a value must be in order to be
-	// separated into a blob file.
-	MinimumSize() int
-	// EstimatedFileSize returns an estimate of the disk space consumed by the
-	// current, pending blob file if it were closed now. If no blob file has
-	// been created, it returns 0.
-	EstimatedFileSize() uint64
-	// EstimatedReferenceSize returns an estimate of the disk space consumed by
-	// the current output sstable's blob references so far.
-	EstimatedReferenceSize() uint64
-	// Add adds the provided key-value pair to the provided sstable writer,
-	// possibly separating the value into a blob file.
-	Add(tw sstable.RawWriter, kv *base.InternalKV, forceObsolete bool, isLikelyMVCCGarbage bool) error
-	// FinishOutput is called when a compaction is finishing an output sstable.
-	// It returns the table's blob references, which will be added to the
-	// table's TableMetadata, and stats and metadata describing a newly
-	// constructed blob file if any.
-	FinishOutput() (ValueSeparationMetadata, error)
-}
-
-// ValueSeparationMetadata describes metadata about a table's blob references,
-// and optionally a newly constructed blob file.
-type ValueSeparationMetadata struct {
-	BlobReferences     manifest.BlobReferences
-	BlobReferenceSize  uint64
-	BlobReferenceDepth manifest.BlobReferenceDepth
-
-	// The below fields are only populated if a new blob file was created.
-	BlobFileStats    blob.FileWriterStats
-	BlobFileObject   objstorage.ObjectMetadata
-	BlobFileMetadata *manifest.PhysicalBlobFile
-}
-
 // Runner is a helper for running the "data" part of a compaction (where we use
 // the compaction iterator to write output tables).
 //
@@ -233,7 +184,7 @@ func (r *Runner) WriteTable(
 	objMeta objstorage.ObjectMetadata,
 	tw sstable.RawWriter,
 	limitKey []byte,
-	valueSeparation ValueSeparation,
+	valueSeparation valsep.ValueSeparation,
 ) {
 	if r.err != nil {
 		panic("error already encountered")
@@ -251,16 +202,18 @@ func (r *Runner) WriteTable(
 	// Inform the value separation policy that the table is finished.
 	valSepMeta, valSepErr := valueSeparation.FinishOutput()
 	if valSepErr != nil {
-		r.err = errors.CombineErrors(r.err, valSepErr)
+		err = errors.CombineErrors(err, valSepErr)
 	} else {
 		r.tables[len(r.tables)-1].BlobReferences = valSepMeta.BlobReferences
 		r.tables[len(r.tables)-1].BlobReferenceDepth = valSepMeta.BlobReferenceDepth
-		if valSepMeta.BlobFileObject.DiskFileNum != 0 {
-			r.blobs = append(r.blobs, OutputBlob{
-				Stats:    valSepMeta.BlobFileStats,
-				ObjMeta:  valSepMeta.BlobFileObject,
-				Metadata: valSepMeta.BlobFileMetadata,
-			})
+		if len(valSepMeta.NewBlobFiles) != 0 {
+			for _, bf := range valSepMeta.NewBlobFiles {
+				r.blobs = append(r.blobs, OutputBlob{
+					Stats:    bf.FileStats,
+					ObjMeta:  bf.FileObject,
+					Metadata: bf.FileMetadata,
+				})
+			}
 		}
 	}
 
@@ -280,13 +233,16 @@ func (r *Runner) WriteTable(
 		return
 	}
 	r.tables[len(r.tables)-1].WriterMeta = *writerMeta
-	r.stats.CumulativeWrittenSize += writerMeta.Size + valSepMeta.BlobFileStats.FileLen
+	r.stats.CumulativeWrittenSize += writerMeta.Size
 	r.stats.CumulativeBlobReferenceSize += valSepMeta.BlobReferenceSize
-	r.stats.CumulativeBlobFileSize += valSepMeta.BlobFileStats.FileLen
+	for _, bf := range valSepMeta.NewBlobFiles {
+		r.stats.CumulativeWrittenSize += bf.FileStats.FileLen
+		r.stats.CumulativeBlobFileSize += bf.FileStats.FileLen
+	}
 }
 
 func (r *Runner) writeKeysToTable(
-	tw sstable.RawWriter, limitKey []byte, valueSeparation ValueSeparation,
+	tw sstable.RawWriter, limitKey []byte, valueSeparation valsep.ValueSeparation,
 ) (splitKey []byte, _ error) {
 	const updateGrantHandleEveryNKeys = 128
 	firstKey := r.FirstKey()
@@ -495,44 +451,3 @@ func spanStartOrNil(s *keyspan.Span) []byte {
 	}
 	return s.Start
 }
-
-// NeverSeparateValues is a ValueSeparation implementation that never separates
-// values into external blob files. It is the default value if no
-// ValueSeparation implementation is explicitly provided.
-type NeverSeparateValues struct{}
-
-// Assert that NeverSeparateValues implements the ValueSeparation interface.
-var _ ValueSeparation = NeverSeparateValues{}
-
-// SetNextOutputConfig implements the ValueSeparation interface.
-func (NeverSeparateValues) SetNextOutputConfig(config ValueSeparationOutputConfig) {}
-
-// Kind implements the ValueSeparation interface.
-func (NeverSeparateValues) Kind() sstable.ValueSeparationKind {
-	return sstable.ValueSeparationNone
-}
-
-// MinimumSize implements the ValueSeparation interface.
-func (v NeverSeparateValues) MinimumSize() int { return 0 }
-
-// EstimatedFileSize implements the ValueSeparation interface.
-func (NeverSeparateValues) EstimatedFileSize() uint64 { return 0 }
-
-// EstimatedReferenceSize implements the ValueSeparation interface.
-func (NeverSeparateValues) EstimatedReferenceSize() uint64 { return 0 }
-
-// Add implements the ValueSeparation interface.
-func (NeverSeparateValues) Add(
-	tw sstable.RawWriter, kv *base.InternalKV, forceObsolete bool, _ bool,
-) error {
-	v, _, err := kv.Value(nil)
-	if err != nil {
-		return err
-	}
-	return tw.Add(kv.K, v, forceObsolete)
-}
-
-// FinishOutput implements the ValueSeparation interface.
-func (NeverSeparateValues) FinishOutput() (ValueSeparationMetadata, error) {
-	return ValueSeparationMetadata{}, nil
-}
diff --git a/internal/deletepacer/delete_pacer.go b/internal/deletepacer/delete_pacer.go
new file mode 100644
index 00000000..478faab5
--- /dev/null
+++ b/internal/deletepacer/delete_pacer.go
@@ -0,0 +1,283 @@
+// Copyright 2025 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package deletepacer
+
+import (
+	"context"
+	"runtime/pprof"
+	"sync"
+	"time"
+
+	"github.com/cockroachdb/crlib/crtime"
+	"github.com/cockroachdb/crlib/fifo"
+	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/invariants"
+	"github.com/cockroachdb/pebble/metrics"
+)
+
+// DiskFreeSpaceFn returns the current amount of free space on the disk in
+// bytes. This is used to determine whether the deletion rate needs to be
+// increased to prevent running out of disk space.
+type DiskFreeSpaceFn func() uint64
+
+// DeleteFn is called to perform the actual deletion of an obsolete file. It
+// should not panic. Any errors must be handled internally. The delete pacer
+// does not retry on deletion failures.
+type DeleteFn func(of ObsoleteFile, jobID int)
+
+// DeletePacer rate-limits deletions of obsolete files to prevent disk
+// performance degradation. On some SSDs, deleting large amounts of data too
+// quickly can negatively impact overall disk performance due to internal garbage
+// collection overhead.
+//
+// The delete pacer maintains a queue of obsolete files and processes them at a
+// controlled rate. The deletion rate adapts dynamically based on:
+//   - A configured baseline rate (minimum deletion throughput)
+//   - Recent deletion patterns (smoothing out bursts over a 5-minute window)
+//   - Queue backlog (accelerating when the queue grows too large)
+//   - Available disk space (accelerating when running low on space)
+//
+// This ensures files are deleted at a steady pace that matches the workload
+// while preventing sudden bursts that could overload the disk, particularly
+// after large compactions or when closing iterators.
+//
+// DeletePacer is safe for concurrent use. It must be created with Open() and
+// closed with Close().
+type DeletePacer struct {
+	opts            Options
+	diskFreeSpaceFn DiskFreeSpaceFn
+	deleteFn        DeleteFn
+	logger          base.Logger
+
+	mu struct {
+		sync.Mutex
+
+		queue             fifo.Queue[queueEntry]
+		queuedPacingBytes uint64
+		// queuedHistory keeps track of pacing bytes added to the queue within the
+		// last 5 minutes.
+		queuedHistory history
+		metrics       Metrics
+		// deletedCond is signaled when we delete a file; used by WaitForTesting().
+		deletedCond sync.Cond
+		closed      bool
+	}
+	// notifyCh is used to wake up the background goroutine. Used when new files
+	// are enqueued, and on Close().
+	notifyCh chan struct{}
+	// waitGroup is used in Close() to wait for the background goroutine to exit.
+	waitGroup sync.WaitGroup
+}
+
+// RecentRateWindow is the timeframe over which we smooth out deletion bursts.
+// The delete pacer uses this window to calculate the recent deletion rate and
+// spread out large bursts over time.
+//
+// Files that have been in the queue for longer than this window are considered
+// backlog, triggering accelerated deletion rates.
+const RecentRateWindow = 5 * time.Minute
+
+// maxQueueSize is a safety valve to prevent unbounded queue growth. If the
+// queue exceeds this size, pacing is temporarily disabled to drain the queue
+// quickly. This prevents memory exhaustion and ensures deletions don't fall too
+// far behind.
+//
+// When this limit is hit, an error is logged (at most once per minute).
+const maxQueueSize = 1000
+
+// Open creates a DeletePacer and starts its background goroutine.
+// The DeletePacer must be Close()d.
+func Open(
+	opts Options, logger base.Logger, diskFreeSpaceFn DiskFreeSpaceFn, deleteFn DeleteFn,
+) *DeletePacer {
+	opts.EnsureDefaults()
+	dp := &DeletePacer{
+		opts:            opts,
+		logger:          logger,
+		diskFreeSpaceFn: diskFreeSpaceFn,
+		deleteFn:        deleteFn,
+		notifyCh:        make(chan struct{}, 1),
+	}
+	dp.mu.queue = fifo.MakeQueue(&queueBackingPool)
+	dp.mu.queuedHistory.Init(crtime.NowMono(), RecentRateWindow)
+	dp.mu.deletedCond.L = &dp.mu.Mutex
+	dp.waitGroup.Add(1)
+	go func() {
+		pprof.Do(context.Background(), pprof.Labels("pebble", "gc"), func(context.Context) {
+			dp.mainLoop()
+		})
+	}()
+	return dp
+}
+
+var queueBackingPool = fifo.MakeQueueBackingPool[queueEntry]()
+
+type queueEntry struct {
+	ObsoleteFile
+	JobID int
+}
+
+// Close stops the background goroutine, waiting until all queued jobs are completed.
+// Delete pacing is disabled for the remaining jobs.
+//
+// EnqueueJob() must not be called after Close().
+func (dp *DeletePacer) Close() {
+	dp.mu.Lock()
+	dp.mu.closed = true
+	dp.mu.Unlock()
+	select {
+	case dp.notifyCh <- struct{}{}:
+	default:
+	}
+	dp.waitGroup.Wait()
+}
+
+// mainLoop is the background goroutine that processes the delete queue.
+//
+// We keep track of a pacing rate in bytes/sec. When we delete a file, we add its
+// size to a "debt" counter. The debt is then paid off at the pacing rate. If the
+// debt is greater than zero, we wait until it is paid off before deleting the
+// next file.
+//
+// The pacing rate is recalculated every time we are about to delete a file or
+// when a new file gets enqueued. The pacing rate is based on:
+//   - the baseline rate configured by the user;
+//   - the recent rate of incoming deletions (over the last 5 minutes);
+//   - the backlog (deletions that have been in the queue for more than 5 minutes);
+//   - whether we are running low on free space.
+func (dp *DeletePacer) mainLoop() {
+	defer dp.waitGroup.Done()
+
+	timer := time.NewTimer(time.Duration(0))
+	defer timer.Stop()
+
+	rateCalc := makeRateCalculator(&dp.opts, dp.diskFreeSpaceFn, crtime.NowMono())
+
+	var lastMaxQueueLog crtime.Mono
+	dp.mu.Lock()
+	defer dp.mu.Unlock()
+	for {
+		if dp.mu.closed && dp.mu.queue.Len() == 0 {
+			return
+		}
+		now := crtime.NowMono()
+		disablePacing := dp.mu.closed
+		if dp.mu.queue.Len() > maxQueueSize {
+			// The queue is getting out of hand; disable pacing.
+			disablePacing = true
+			if lastMaxQueueLog == 0 || now.Sub(lastMaxQueueLog) > time.Minute {
+				lastMaxQueueLog = now
+				dp.logger.Errorf("excessive delete pacer queue size %d; pacing temporarily disabled", dp.mu.queue.Len())
+			}
+		}
+
+		rateCalc.Update(now, dp.mu.queuedHistory.Sum(now), dp.mu.queuedPacingBytes, disablePacing)
+
+		// Processing priority:
+		//   1. Exit if closed and queue empty;
+		//   2. Wait for pacing debt to clear;
+		//   3. Otherwise, delete next file.
+		switch {
+		case dp.mu.queue.Len() == 0:
+			// Nothing to do.
+			dp.mu.Unlock()
+			<-dp.notifyCh
+			dp.mu.Lock()
+
+		case rateCalc.InDebt():
+			// We have files in the queue but we must wait.
+			dp.mu.Unlock()
+			waitTime := rateCalc.DebtWaitTime()
+			// Don't wait more than 10 seconds; we want a chance to recalculate the
+			// rate (and check if we're running low on free space).
+			waitTime = min(waitTime, 10*time.Second)
+			timer.Reset(waitTime)
+			select {
+			case <-timer.C:
+			case <-dp.notifyCh:
+				timer.Stop()
+			}
+			dp.mu.Lock()
+
+		default:
+			// Delete a file.
+			file := *dp.mu.queue.PeekFront()
+			dp.mu.queue.PopFront()
+			if b := file.pacingBytes(); b != 0 {
+				dp.mu.queuedPacingBytes = invariants.SafeSub(dp.mu.queuedPacingBytes, b)
+				rateCalc.AddDebt(b)
+			}
+			func() {
+				dp.mu.Unlock()
+				defer dp.mu.Lock()
+				dp.deleteFn(file.ObsoleteFile, file.JobID)
+			}()
+			dp.mu.metrics.InQueue.Dec(file.FileType, file.FileSize, file.IsLocal)
+			dp.mu.metrics.Deleted.Inc(file.FileType, file.FileSize, file.IsLocal)
+			dp.mu.deletedCond.Broadcast()
+		}
+	}
+}
+
+// Enqueue adds the given files to the delete queue. Enqueue never blocks.
+func (dp *DeletePacer) Enqueue(jobID int, files ...ObsoleteFile) {
+	dp.mu.Lock()
+	defer dp.mu.Unlock()
+	if dp.mu.closed {
+		if invariants.Enabled {
+			panic("Enqueue called after Close")
+		}
+		return
+	}
+	now := crtime.NowMono()
+	for _, file := range files {
+		if b := file.pacingBytes(); b > 0 {
+			dp.mu.queuedPacingBytes += b
+			dp.mu.queuedHistory.Add(now, b)
+		}
+		dp.mu.metrics.InQueue.Inc(file.FileType, file.FileSize, file.IsLocal)
+		dp.mu.queue.PushBack(queueEntry{
+			ObsoleteFile: file,
+			JobID:        jobID,
+		})
+	}
+	select {
+	case dp.notifyCh <- struct{}{}:
+	default:
+	}
+}
+
+// Metrics returns the current metrics.
+func (dp *DeletePacer) Metrics() Metrics {
+	dp.mu.Lock()
+	defer dp.mu.Unlock()
+	return dp.mu.metrics
+}
+
+// Metrics tracks statistics about files in the delete pacer queue and files
+// that have been deleted. This provides visibility into deletion throughput and
+// queue depth.
+type Metrics struct {
+	// InQueue contains the count and total size of files currently waiting in the
+	// delete queue, broken down by file type (tables vs blob files) and locality
+	// (all vs local only).
+	InQueue metrics.FileCountsAndSizes
+
+	// Deleted contains the count and total size of files that have been deleted
+	// since the DeletePacer was started, broken down by file type and locality.
+	Deleted metrics.FileCountsAndSizes
+}
+
+// WaitForTesting waits until the deletion of all files that were already
+// queued. Does not wait for jobs that are enqueued during the call.
+func (dp *DeletePacer) WaitForTesting() {
+	dp.mu.Lock()
+	defer dp.mu.Unlock()
+
+	n := dp.mu.metrics.Deleted.Totals().Count + dp.mu.metrics.InQueue.Totals().Count
+	for dp.mu.metrics.Deleted.Totals().Count < n {
+		dp.mu.deletedCond.Wait()
+	}
+}
diff --git a/internal/deletepacer/history.go b/internal/deletepacer/history.go
new file mode 100644
index 00000000..47a9353c
--- /dev/null
+++ b/internal/deletepacer/history.go
@@ -0,0 +1,71 @@
+// Copyright 2025 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package deletepacer
+
+import (
+	"time"
+
+	"github.com/cockroachdb/crlib/crtime"
+	"github.com/cockroachdb/pebble/internal/invariants"
+)
+
+// history is a helper used to keep track of the recent history of a set of
+// data points (in our case deleted bytes), at limited granularity.
+// Specifically, we split the desired timeframe into 100 "epochs" and all times
+// are effectively rounded down to the nearest epoch boundary.
+type history struct {
+	epochDuration time.Duration
+	startTime     crtime.Mono
+	// currEpoch is the epoch of the most recent operation.
+	currEpoch int64
+	// val contains the recent epoch values.
+	// val[currEpoch % historyEpochs] is the current epoch.
+	// val[(currEpoch + 1) % historyEpochs] is the oldest epoch.
+	val [historyEpochs]uint64
+	// sum is always equal to the sum of values in val.
+	sum uint64
+}
+
+const historyEpochs = 100
+
+// Init the history helper to keep track of data over the given number of
+// seconds.
+func (h *history) Init(now crtime.Mono, timeframe time.Duration) {
+	*h = history{
+		epochDuration: timeframe / time.Duration(historyEpochs),
+		startTime:     now,
+		currEpoch:     0,
+		sum:           0,
+	}
+}
+
+// Add adds a value for the current time.
+func (h *history) Add(now crtime.Mono, val uint64) {
+	h.advance(now)
+	h.val[h.currEpoch%historyEpochs] += val
+	h.sum += val
+}
+
+// Sum returns the sum of recent values. The result is approximate in that the
+// cut-off time is within 1% of the exact one.
+func (h *history) Sum(now crtime.Mono) uint64 {
+	h.advance(now)
+	return h.sum
+}
+
+func (h *history) epoch(t crtime.Mono) int64 {
+	return int64(t.Sub(h.startTime) / h.epochDuration)
+}
+
+// advance advances the time to the given time.
+func (h *history) advance(now crtime.Mono) {
+	epoch := h.epoch(now)
+	for h.currEpoch < epoch {
+		h.currEpoch++
+		// Forget the data for the oldest epoch.
+		h.sum = invariants.SafeSub(h.sum, h.val[h.currEpoch%historyEpochs])
+		h.val[h.currEpoch%historyEpochs] = 0
+	}
+}
diff --git a/internal/deletepacer/obsolete_file.go b/internal/deletepacer/obsolete_file.go
new file mode 100644
index 00000000..89d5fde7
--- /dev/null
+++ b/internal/deletepacer/obsolete_file.go
@@ -0,0 +1,30 @@
+// Copyright 2025 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package deletepacer
+
+import (
+	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/vfs"
+)
+
+// ObsoleteFile describes a file that is to be deleted.
+type ObsoleteFile struct {
+	FileType base.FileType
+	FS       vfs.FS
+	Path     string
+	FileNum  base.DiskFileNum
+	FileSize uint64 // approx for log files
+	IsLocal  bool
+}
+
+// pacingBytes returns the size of the file, or 0 if deleting the file does not
+// require pacing.
+func (of ObsoleteFile) pacingBytes() uint64 {
+	// We only need to pace local objects--sstables and blob files.
+	if of.IsLocal && (of.FileType == base.FileTypeTable || of.FileType == base.FileTypeBlob) {
+		return of.FileSize
+	}
+	return 0
+}
diff --git a/internal/deletepacer/options.go b/internal/deletepacer/options.go
new file mode 100644
index 00000000..cbdb81b1
--- /dev/null
+++ b/internal/deletepacer/options.go
@@ -0,0 +1,69 @@
+// Copyright 2025 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package deletepacer
+
+import "time"
+
+// Options contains configuration options for the delete pacer.
+type Options struct {
+	// BaselineRate returns the baseline deletion rate in bytes/sec. This is the
+	// minimum rate at which files are deleted, even when deletions are rare.
+	//
+	// The actual deletion rate may be higher based on recent deletion patterns,
+	// backlog, or low disk space.
+	//
+	// A returned value of 0 disables deletion pacing (this is also the default).
+	BaselineRate func() uint64
+
+	// BacklogTimeframe is the target duration for clearing the backlog queue.
+	// Deletions that have been queued for longer than RecentRateWindow (5
+	// minutes) are considered backlog. When backlog is detected, the deletion
+	// rate is increased to clear it within this timeframe.
+	//
+	// This prevents the delete queue from growing unbounded when deletions arrive
+	// faster than they can be processed. Shorter timeframes result in more
+	// aggressive deletion rates.
+	//
+	// The default value is 5 minutes.
+	BacklogTimeframe time.Duration
+
+	// FreeSpaceThresholdBytes is the minimum amount of free disk space to
+	// maintain. If free space falls below this threshold, the deletion rate is
+	// increased to reclaim enough space to return to the threshold within
+	// FreeSpaceTimeframe.
+	//
+	// The default value is 16GB.
+	FreeSpaceThresholdBytes uint64
+
+	// FreeSpaceTimeframe is the target duration for reclaiming disk space when
+	// free space falls below FreeSpaceThresholdBytes. The deletion rate is
+	// increased to eliminate the deficit within this timeframe.
+	//
+	// Shorter timeframes result in more aggressive deletion rates when disk space
+	// is low.
+	//
+	// The default value is 10 seconds.
+	FreeSpaceTimeframe time.Duration
+}
+
+// EnsureDefaults ensures that the default values for all options are set if a
+// valid value was not already specified.
+func (o *Options) EnsureDefaults() {
+	if o.BaselineRate == nil {
+		o.BaselineRate = func() uint64 { return 0 }
+	}
+
+	if o.BacklogTimeframe == 0 {
+		o.BacklogTimeframe = 5 * time.Minute
+	}
+
+	if o.FreeSpaceThresholdBytes == 0 {
+		o.FreeSpaceThresholdBytes = 16 << 30 // 16 GB
+	}
+
+	if o.FreeSpaceTimeframe == 0 {
+		o.FreeSpaceTimeframe = 10 * time.Second
+	}
+}
diff --git a/internal/deletepacer/rate_calc.go b/internal/deletepacer/rate_calc.go
new file mode 100644
index 00000000..51569849
--- /dev/null
+++ b/internal/deletepacer/rate_calc.go
@@ -0,0 +1,175 @@
+// Copyright 2025 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package deletepacer
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/cockroachdb/crlib/crtime"
+)
+
+// rateCalculator computes the target deletion throughput (bytes/sec) for a
+// paced deleter. It maintains a baseline rate and adds a corrective "backlog
+// component" using a constant-horizon drain model:
+//
+//   - Feed-forward: assume the next short interval will see new deletions at the
+//     same average rate as the recent window (RecentRateWindow).
+//
+//   - Backlog drain: if a queue backlog is observed (or disk free space is
+//     below a threshold), add just enough extra rate to eliminate that deficit
+//     within a configured timeframe, and *hold that extra rate* until cleared.
+//
+// The calculator also tracks "debt": bytes from already-initiated deletions that
+// we must wait out to respect the current pacing rate. Debt decays at the
+// current rate over time, and callers can query how long to pause before
+// starting the next deletion.
+type rateCalculator struct {
+	opts            *Options
+	diskFreeSpaceFn DiskFreeSpaceFn
+
+	lastUpdate crtime.Mono
+
+	// currentRate is the target deletion throughput in bytes/sec currently in
+	// effect. When pacing is disabled, this is 0.
+	currentRate float64
+
+	// debtBytes is the number of bytes of already-started deletions we must
+	// "wait out" to respect currentRate.
+	debtBytes float64
+
+	// backlogRate is the extra throughput (bytes/sec) needed to eliminate
+	// observed backlog within the configured horizon. Once raised, it does not
+	// decrease until the backlog is gone (constant-horizon drain).
+	backlogRate float64
+
+	// freeSpaceRate is the extra throughput (bytes/sec) needed to eliminate
+	// observed free-space deficit within the configured horizon. Once raised, it
+	// does not decrease until the free space is above the threshold.
+	// (constant-horizon drain).
+	freeSpaceRate float64
+}
+
+const debug = false
+
+func makeRateCalculator(
+	opts *Options, diskFreeSpaceFn DiskFreeSpaceFn, now crtime.Mono,
+) rateCalculator {
+	return rateCalculator{
+		opts:            opts,
+		diskFreeSpaceFn: diskFreeSpaceFn,
+		lastUpdate:      now,
+	}
+}
+
+// Update recalculates the current deletion rate and updates the debt based on
+// elapsed time. This should be called before each file deletion and when new
+// files are enqueued.
+//
+// The deletion rate is determined by:
+//  1. Starting with the baseline rate from the config
+//  2. Taking the max of baseline and recent deletion rate (feed-forward estimate)
+//  3. Adding extra throughput if there's a backlog or low disk space
+//
+// The backlog and free-space rates use a "constant-horizon drain" model: once
+// raised, they don't decrease until the condition is cleared. This prevents
+// oscillation and ensures consistent progress.
+//
+// The recentPacingBytes argument is the number of bytes that were enqueued for
+// deletion in the last RecentRateWindow (regardless if those files were deleted
+// already or not); queuedPacingBytes is the total bytes currently waiting in
+// the queue.
+func (rc *rateCalculator) Update(
+	now crtime.Mono, recentPacingBytes uint64, queuedPacingBytes uint64, disablePacing bool,
+) {
+	deltaT := now.Sub(rc.lastUpdate)
+	rc.lastUpdate = now
+	baselineRate := rc.opts.BaselineRate()
+	if baselineRate == 0 || disablePacing {
+		// Pacing is disabled.
+		rc.currentRate = 0
+		rc.debtBytes = 0
+		rc.backlogRate = 0
+		rc.freeSpaceRate = 0
+		return
+	}
+	rc.debtBytes = max(0, rc.debtBytes-rc.currentRate*deltaT.Seconds())
+
+	// We assume that we will get incoming deletions at the same average rate as
+	// over the last 5 minutes. This is effectively a feed-forward estimate.
+	rc.currentRate = float64(baselineRate)
+	rc.currentRate = max(rc.currentRate, float64(recentPacingBytes)/RecentRateWindow.Seconds())
+
+	// Everything that stayed in the queue for more than 5 minutes is considered
+	// backlog. Once we observe backlog, we increase the rate so that the backlog
+	// is wiped out in the backlog time frame.
+	if queuedPacingBytes > recentPacingBytes {
+		if debug {
+			fmt.Printf("backlog: queued %d  recent: %d\n", queuedPacingBytes, recentPacingBytes)
+		}
+		backlog := float64(queuedPacingBytes - recentPacingBytes)
+		// Constant-horizon drain: don't let the rate go down until the backlog is
+		// cleared. If backlog increases further, we increase the rate.
+		rc.backlogRate = max(rc.backlogRate, backlog/rc.opts.BacklogTimeframe.Seconds())
+	} else {
+		rc.backlogRate = 0
+	}
+
+	if freeBytes := rc.diskFreeSpaceFn(); freeBytes < rc.opts.FreeSpaceThresholdBytes {
+		// Constant-horizon drain: don't let the rate go down until the deficit is
+		// cleared.
+		deficit := float64(rc.opts.FreeSpaceThresholdBytes - freeBytes)
+		rc.freeSpaceRate = max(rc.freeSpaceRate, deficit/rc.opts.FreeSpaceTimeframe.Seconds())
+	} else {
+		rc.freeSpaceRate = 0
+	}
+
+	// We take the maximum of the extra rates (deleting a file helps both).
+	rc.currentRate += max(rc.backlogRate, rc.freeSpaceRate)
+	if debug {
+		fmt.Printf("%s current-rate=%v  backlog-rate=%v  free-space-rate=%v  debt-bytes=%v\n",
+			time.Now().Format(time.Stamp), rc.currentRate, rc.backlogRate, rc.freeSpaceRate, rc.debtBytes)
+	}
+}
+
+// AddDebt adds to the deletion debt when starting to delete a file. The debt
+// represents bytes of files currently being deleted that we must "wait out" to
+// maintain the target deletion rate.
+//
+// Debt decays over time at the current rate. Before starting the next deletion,
+// the pacer waits until the debt is paid off (or low enough).
+//
+// The debt is capped at 1GB as a safety measure to prevent excessively long
+// pauses when deleting very large files.
+func (rc *rateCalculator) AddDebt(bytes uint64) {
+	if rc.currentRate == 0 {
+		// Pacing is disabled.
+		return
+	}
+	// We cap the maximum debt as a fail-safe for extreme cases of very large
+	// files. We don't want to pause deletions for minutes in such a case.
+	const maxDebt = 1 << 30 // 1GB
+	rc.debtBytes += float64(bytes)
+	rc.debtBytes = min(rc.debtBytes, maxDebt)
+}
+
+// InDebt returns whether we are currently in debt (i.e. need to pause).
+func (rc *rateCalculator) InDebt() bool {
+	return rc.debtBytes > 0
+}
+
+// DebtWaitTime returns how long we need to wait before starting the next file
+// deletion to maintain the target deletion rate. The wait time is calculated as
+// the current debt divided by the current deletion rate.
+//
+// This method panics if called when there is no debt (InDebt() returns false).
+func (rc *rateCalculator) DebtWaitTime() time.Duration {
+	if rc.debtBytes == 0 {
+		panic("no debt")
+	}
+	// We add 1ns as a way to round up. We must also make sure we never return 0,
+	// as that will be problematic with synctest (which has exact sleeps).
+	return time.Duration(1 + rc.debtBytes/rc.currentRate*float64(time.Second))
+}
diff --git a/internal/manifest/btree.go b/internal/manifest/btree.go
index 9e0d41a9..b015fe04 100644
--- a/internal/manifest/btree.go
+++ b/internal/manifest/btree.go
@@ -736,13 +736,15 @@ func (t *btree[M]) Clone() btree[M] {
 // Delete removes the provided table from the tree, unrefing the table's files
 // if it's found. If any files are unreferenced to zero, they're added to the
 // provided obsoleteFiles.
-func (t *btree[M]) Delete(item M, of ObsoleteFilesSet) {
+func (t *btree[M]) Delete(item M, of ObsoleteFilesSet) (found bool) {
 	if t.root == nil || t.root.count == 0 {
-		return
+		return false
 	}
-	if out, found := mut(&t.root).Remove(t.bcmp, item); found {
+	out, found := mut(&t.root).Remove(t.bcmp, item)
+	if found {
 		out.Unref(of)
 	}
+	// Note that even if the item was not found, we could have rebalanced a node.
 	if invariants.Enabled {
 		t.root.verifyInvariants()
 	}
@@ -755,6 +757,7 @@ func (t *btree[M]) Delete(item M, of ObsoleteFilesSet) {
 		}
 		old.decRef(false /* contentsToo */, assertNoObsoleteFiles{})
 	}
+	return found
 }
 
 // Insert adds the given item to the tree. If a item in the tree already
diff --git a/internal/manifest/level_metadata.go b/internal/manifest/level_metadata.go
index 1c4b6c74..9e9bf2d6 100644
--- a/internal/manifest/level_metadata.go
+++ b/internal/manifest/level_metadata.go
@@ -12,6 +12,7 @@ import (
 
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/invariants"
+	"github.com/cockroachdb/pebble/metrics"
 )
 
 // LevelMetadata contains metadata for all of the files within
@@ -20,23 +21,19 @@ type LevelMetadata struct {
 	level          int
 	totalTableSize uint64
 	totalRefSize   uint64
-	// NumVirtual is the number of virtual sstables in the level.
-	NumVirtual uint64
-	// VirtualTableSize is the size of the virtual sstables in the level.
-	VirtualTableSize uint64
-	tree             btree[*TableMetadata]
+	virtualTables  metrics.CountAndSize
+	tree           btree[*TableMetadata]
 }
 
 // clone makes a copy of the level metadata, implicitly increasing the ref
 // count of every file contained within lm.
 func (lm *LevelMetadata) clone() LevelMetadata {
 	return LevelMetadata{
-		level:            lm.level,
-		totalTableSize:   lm.totalTableSize,
-		totalRefSize:     lm.totalRefSize,
-		NumVirtual:       lm.NumVirtual,
-		VirtualTableSize: lm.VirtualTableSize,
-		tree:             lm.tree.Clone(),
+		level:          lm.level,
+		totalTableSize: lm.totalTableSize,
+		totalRefSize:   lm.totalRefSize,
+		virtualTables:  lm.virtualTables,
+		tree:           lm.tree.Clone(),
 	}
 }
 
@@ -57,8 +54,7 @@ func MakeLevelMetadata(cmp Compare, level int, files []*TableMetadata) LevelMeta
 		lm.totalTableSize += f.Size
 		lm.totalRefSize += f.EstimatedReferenceSize()
 		if f.Virtual {
-			lm.NumVirtual++
-			lm.VirtualTableSize += f.Size
+			lm.virtualTables.Inc(f.Size)
 		}
 	}
 	return lm
@@ -91,20 +87,21 @@ func (lm *LevelMetadata) insert(f *TableMetadata) error {
 	lm.totalTableSize += f.Size
 	lm.totalRefSize += f.EstimatedReferenceSize()
 	if f.Virtual {
-		lm.NumVirtual++
-		lm.VirtualTableSize += f.Size
+		lm.virtualTables.Inc(f.Size)
 	}
 	return nil
 }
 
-func (lm *LevelMetadata) remove(f *TableMetadata) {
-	lm.totalTableSize -= f.Size
-	lm.totalRefSize -= f.EstimatedReferenceSize()
+func (lm *LevelMetadata) remove(f *TableMetadata) (found bool) {
+	if !lm.tree.Delete(f, assertNoObsoleteFiles{}) {
+		return false
+	}
+	lm.totalTableSize = invariants.SafeSub(lm.totalTableSize, f.Size)
+	lm.totalRefSize = invariants.SafeSub(lm.totalRefSize, f.EstimatedReferenceSize())
 	if f.Virtual {
-		lm.NumVirtual--
-		lm.VirtualTableSize -= f.Size
+		lm.virtualTables.Dec(f.Size)
 	}
-	lm.tree.Delete(f, assertNoObsoleteFiles{})
+	return true
 }
 
 // Empty indicates whether there are any files in the level.
@@ -137,6 +134,12 @@ func (lm *LevelMetadata) EstimatedReferenceSize() uint64 {
 	return lm.totalRefSize
 }
 
+// VirtualTables returns the count and total size of the virtual tables in the
+// level.
+func (lm *LevelMetadata) VirtualTables() metrics.CountAndSize {
+	return lm.virtualTables
+}
+
 // Iter constructs a LevelIterator over the entire level.
 func (lm *LevelMetadata) Iter() LevelIterator {
 	return LevelIterator{iter: tableMetadataIter(&lm.tree)}
diff --git a/internal/manifest/version.go b/internal/manifest/version.go
index 7f2a3d13..e6f5b9fa 100644
--- a/internal/manifest/version.go
+++ b/internal/manifest/version.go
@@ -102,6 +102,10 @@ func NewVersionForTesting(
 		}
 		for _, f := range files[l] {
 			v.Levels[l].totalTableSize += f.Size
+			v.Levels[l].totalRefSize += f.EstimatedReferenceSize()
+			if f.Virtual {
+				v.Levels[l].virtualTables.Inc(f.Size)
+			}
 		}
 	}
 	l0Organizer.ResetForTesting(v)
diff --git a/internal/manifest/version_edit.go b/internal/manifest/version_edit.go
index 98c11684..1b7fd41a 100644
--- a/internal/manifest/version_edit.go
+++ b/internal/manifest/version_edit.go
@@ -1358,7 +1358,11 @@ func (b *BulkVersionEdit) Apply(curr *Version, readCompactionRate int64) (*Versi
 			// counts.  However, because we cloned the previous level's B-Tree,
 			// this should never result in a file's reference count dropping to
 			// zero. The remove call will panic if this happens.
-			v.Levels[level].remove(f)
+			if !v.Levels[level].remove(f) {
+				if invariants.Enabled {
+					panic(errors.AssertionFailedf("deleted table %f not in level", f))
+				}
+			}
 			v.RangeKeyLevels[level].remove(f)
 			v.MarkedForCompaction.Delete(f, level)
 		}
diff --git a/metamorphic/options.go b/metamorphic/options.go
index ae1f6ab2..7b4ade7b 100644
--- a/metamorphic/options.go
+++ b/metamorphic/options.go
@@ -687,10 +687,10 @@ func RandomOptions(rng *rand.Rand, kf KeyFormat, cfg RandomOptionsCfg) *TestOpti
 	opts.FormatMajorVersion = minimumFormatMajorVersion
 	n := int(newestFormatMajorVersionToTest - opts.FormatMajorVersion)
 	opts.FormatMajorVersion += pebble.FormatMajorVersion(rng.IntN(n + 1))
-	opts.Experimental.L0CompactionConcurrency = 1 + rng.IntN(4) // 1-4
-	opts.Experimental.LevelMultiplier = 5 << rng.IntN(7)        // 5 - 320
-	targetByteDeletionRate := 1 << uint(20+rng.IntN(10))        // 1MB - 1GB
-	opts.TargetByteDeletionRate = func() int { return targetByteDeletionRate }
+	opts.Experimental.L0CompactionConcurrency = 1 + rng.IntN(4)  // 1-4
+	opts.Experimental.LevelMultiplier = 5 << rng.IntN(7)         // 5 - 320
+	targetByteDeletionRate := uint64(1) << uint(20+rng.IntN(10)) // 1MB - 1GB
+	opts.DeletionPacing.BaselineRate = func() uint64 { return targetByteDeletionRate }
 	opts.Experimental.ValidateOnIngest = rng.IntN(2) != 0
 	opts.L0CompactionThreshold = 1 + rng.IntN(100)     // 1 - 100
 	opts.L0CompactionFileThreshold = 1 << rng.IntN(11) // 1 - 1024
diff --git a/metrics.go b/metrics.go
index 917cbc73..6940f9a7 100644
--- a/metrics.go
+++ b/metrics.go
@@ -18,6 +18,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/cache"
 	"github.com/cockroachdb/pebble/internal/compression"
+	"github.com/cockroachdb/pebble/internal/deletepacer"
 	"github.com/cockroachdb/pebble/internal/manifest"
 	"github.com/cockroachdb/pebble/internal/manual"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider/sharedcache"
@@ -481,6 +482,10 @@ type Metrics struct {
 		Failover wal.FailoverStats
 	}
 
+	// DeletePacer are metrics from the delete pacer, which manages obsolete file
+	// deletion. These can be relevant if free disk space is unexplainably low.
+	DeletePacer deletepacer.Metrics
+
 	LogWriter struct {
 		FsyncLatency prometheus.Histogram
 		record.LogWriterMetrics
@@ -845,6 +850,14 @@ var (
 		table.Div(),
 		table.String("blob files", 13, table.AlignRight, func(i compressionInfo) string { return i.blobFiles }),
 	)
+	deletePacerTableHeader = `DELETE PACER`
+	deletePacerTable       = table.Define[deletePacerInfo](
+		table.String("", 14, table.AlignRight, func(i deletePacerInfo) string { return i.label }),
+		table.Div(),
+		table.String("in queue", 12, table.AlignCenter, func(i deletePacerInfo) string { return i.inQueue }),
+		table.Div(),
+		table.String("deleted", 12, table.AlignCenter, func(i deletePacerInfo) string { return i.deleted }),
+	)
 )
 
 type commitPipelineInfo struct {
@@ -962,6 +975,12 @@ type compressionInfo struct {
 	blobFiles string
 }
 
+type deletePacerInfo struct {
+	label   string
+	inQueue string
+	deleted string
+}
+
 func makeCompressionInfo(algorithm string, table, blob CompressionStatsForSetting) compressionInfo {
 	i := compressionInfo{algorithm: algorithm}
 	if table.CompressedBytes > 0 {
@@ -1147,8 +1166,30 @@ func (m *Metrics) String() string {
 	compressionContents = slices.DeleteFunc(compressionContents, func(i compressionInfo) bool {
 		return i.tables == "" && i.blobFiles == ""
 	})
-	compressionTable.Render(cur, table.RenderOptions{}, compressionContents...)
+	cur = compressionTable.Render(cur, table.RenderOptions{}, compressionContents...)
+
+	cur = cur.NewlineReturn()
+	cur.WriteString(deletePacerTableHeader)
+	deletePacerContents := []deletePacerInfo{
+		{
+			label:   "tables",
+			inQueue: m.DeletePacer.InQueue.Tables.String(),
+			deleted: m.DeletePacer.Deleted.Tables.String(),
+		},
+		{
+			label:   "blob files",
+			inQueue: m.DeletePacer.InQueue.BlobFiles.String(),
+			deleted: m.DeletePacer.Deleted.BlobFiles.String(),
+		},
+		{
+			label:   "other files",
+			inQueue: m.DeletePacer.InQueue.Other.String(),
+			deleted: m.DeletePacer.Deleted.Other.String(),
+		},
+	}
+	cur = deletePacerTable.Render(cur, table.RenderOptions{}, deletePacerContents...)
 
+	_ = cur
 	return wb.String()
 }
 
@@ -1210,6 +1251,8 @@ func (m *Metrics) StringForTests() string {
 		mCopy.BlockCache.Recent[i].HitsAndMisses = cache.HitsAndMisses{}
 		mCopy.BlockCache.Recent[i].Since = 0
 	}
+	// Clear the delete pacer stats as they can vary based on timing.
+	mCopy.DeletePacer = deletepacer.Metrics{}
 	return redact.StringWithoutMarkers(&mCopy)
 }
 
diff --git a/metrics/count_and_size.go b/metrics/count_and_size.go
new file mode 100644
index 00000000..ac495bd2
--- /dev/null
+++ b/metrics/count_and_size.go
@@ -0,0 +1,244 @@
+// Copyright 2025 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package metrics
+
+import (
+	"github.com/cockroachdb/crlib/crhumanize"
+	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/invariants"
+	"github.com/cockroachdb/redact"
+)
+
+// CountAndSize tracks the count and total size of a set of items.
+type CountAndSize struct {
+	// Count is the number of files.
+	Count uint64
+
+	// Bytes is the total size of all files.
+	Bytes uint64
+}
+
+// Inc increases the count and size for a single item.
+func (cs *CountAndSize) Inc(fileSize uint64) {
+	cs.Count++
+	cs.Bytes += fileSize
+}
+
+// Dec decreases the count and size for a single item.
+func (cs *CountAndSize) Dec(fileSize uint64) {
+	cs.Count = invariants.SafeSub(cs.Count, 1)
+	cs.Bytes = invariants.SafeSub(cs.Bytes, fileSize)
+}
+
+// Accumulate increases the counts and sizes by the given amounts.
+func (cs *CountAndSize) Accumulate(other CountAndSize) {
+	cs.Count += other.Count
+	cs.Bytes += other.Bytes
+}
+
+// Deduct decreases the counts and sizes by the given amounts.
+func (cs *CountAndSize) Deduct(other CountAndSize) {
+	cs.Count = invariants.SafeSub(cs.Count, other.Count)
+	cs.Bytes = invariants.SafeSub(cs.Bytes, other.Bytes)
+}
+
+func (cs CountAndSize) String() string {
+	return redact.StringWithoutMarkers(cs)
+}
+
+// SafeFormat implements redact.SafeFormatter.
+func (cs CountAndSize) SafeFormat(w redact.SafePrinter, verb rune) {
+	w.Printf("%s (%s)", crhumanize.Count(cs.Count, crhumanize.Compact), crhumanize.Bytes(cs.Bytes, crhumanize.Compact, crhumanize.OmitI))
+}
+
+// TableCountsAndSizes contains counts and sizes for tables, broken down by
+// locality.
+type TableCountsAndSizes struct {
+	// All contains counts for all tables (local and remote).
+	All CountAndSize
+	// Local contains counts for local tables only.
+	Local CountAndSize
+}
+
+// Inc increases the count and size for a single table.
+func (cs *TableCountsAndSizes) Inc(tableSize uint64, isLocal bool) {
+	cs.All.Inc(tableSize)
+	if isLocal {
+		cs.Local.Inc(tableSize)
+	}
+}
+
+// Dec decreases the count and size for a single table.
+func (cs *TableCountsAndSizes) Dec(tableSize uint64, isLocal bool) {
+	cs.All.Dec(tableSize)
+	if isLocal {
+		cs.Local.Dec(tableSize)
+	}
+}
+
+// Accumulate increases the counts and sizes by the given amounts.
+func (cs *TableCountsAndSizes) Accumulate(other TableCountsAndSizes) {
+	cs.All.Accumulate(other.All)
+	cs.Local.Accumulate(other.Local)
+}
+
+// Deduct decreases the counts and sizes by the given amounts.
+func (cs *TableCountsAndSizes) Deduct(other TableCountsAndSizes) {
+	cs.All.Deduct(other.All)
+	cs.Local.Deduct(other.Local)
+}
+
+func (cs TableCountsAndSizes) String() string {
+	return redact.StringWithoutMarkers(cs)
+}
+
+// SafeFormat implements redact.SafeFormatter.
+func (cs TableCountsAndSizes) SafeFormat(w redact.SafePrinter, verb rune) {
+	cs.All.SafeFormat(w, verb)
+	if cs.All != cs.Local {
+		w.Printf(" [local: ")
+		cs.Local.SafeFormat(w, verb)
+		w.Printf("]")
+	}
+}
+
+// BlobFileCountsAndSizes contains counts and sizes for blob files, broken down
+// by locality.
+type BlobFileCountsAndSizes struct {
+	// All contains counts for all blob files (local and remote).
+	All CountAndSize
+	// Local contains counts for local blob files only.
+	Local CountAndSize
+}
+
+// Inc increases the count and size for a single blob file.
+func (cs *BlobFileCountsAndSizes) Inc(fileSize uint64, isLocal bool) {
+	cs.All.Inc(fileSize)
+	if isLocal {
+		cs.Local.Inc(fileSize)
+	}
+}
+
+// Dec decreases the count and size for a single blob file.
+func (cs *BlobFileCountsAndSizes) Dec(fileSize uint64, isLocal bool) {
+	cs.All.Dec(fileSize)
+	if isLocal {
+		cs.Local.Dec(fileSize)
+	}
+}
+
+// Accumulate increases the counts and sizes by the given amounts.
+func (cs *BlobFileCountsAndSizes) Accumulate(other BlobFileCountsAndSizes) {
+	cs.All.Accumulate(other.All)
+	cs.Local.Accumulate(other.Local)
+}
+
+// Deduct decreases the counts and sizes by the given amounts.
+func (cs *BlobFileCountsAndSizes) Deduct(other BlobFileCountsAndSizes) {
+	cs.All.Deduct(other.All)
+	cs.Local.Deduct(other.Local)
+}
+
+func (cs BlobFileCountsAndSizes) String() string {
+	return redact.StringWithoutMarkers(cs)
+}
+
+// SafeFormat implements redact.SafeFormatter.
+func (cs BlobFileCountsAndSizes) SafeFormat(w redact.SafePrinter, verb rune) {
+	cs.All.SafeFormat(w, verb)
+	if cs.All != cs.Local {
+		w.Printf(" [local: %s]", cs.Local)
+	}
+}
+
+// FileCountsAndSizes contains counts and sizes for all file types.
+type FileCountsAndSizes struct {
+	// Tables contains counts and sizes for tables.
+	Tables TableCountsAndSizes
+
+	// BlobFiles contains counts and sizes for blob files.
+	BlobFiles BlobFileCountsAndSizes
+
+	// Other contains counts and sizes for other file types (log, manifest, etc).
+	// These are not separated by locality.
+	Other CountAndSize
+}
+
+// TotalCount returns the total
+func (cs *FileCountsAndSizes) Totals() CountAndSize {
+	res := cs.Tables.All
+	res.Accumulate(cs.BlobFiles.All)
+	res.Accumulate(cs.Other)
+	return res
+}
+
+// Inc increases the relevant count and size for a single file.
+func (cs *FileCountsAndSizes) Inc(fileType base.FileType, fileSize uint64, isLocal bool) {
+	switch fileType {
+	case base.FileTypeTable:
+		cs.Tables.Inc(fileSize, isLocal)
+	case base.FileTypeBlob:
+		cs.BlobFiles.Inc(fileSize, isLocal)
+	default:
+		cs.Other.Inc(fileSize)
+	}
+}
+
+// Dec decreases the relevant count and size for a single file.
+func (cs *FileCountsAndSizes) Dec(fileType base.FileType, fileSize uint64, isLocal bool) {
+	switch fileType {
+	case base.FileTypeTable:
+		cs.Tables.Dec(fileSize, isLocal)
+	case base.FileTypeBlob:
+		cs.BlobFiles.Dec(fileSize, isLocal)
+	default:
+		cs.Other.Dec(fileSize)
+	}
+}
+
+// Accumulate increases the counts and sizes by the given amounts.
+func (cs *FileCountsAndSizes) Accumulate(other FileCountsAndSizes) {
+	cs.Tables.Accumulate(other.Tables)
+	cs.BlobFiles.Accumulate(other.BlobFiles)
+	cs.Other.Accumulate(other.Other)
+}
+
+// Deduct decreases the counts and sizes by the given amounts.
+func (cs *FileCountsAndSizes) Deduct(other FileCountsAndSizes) {
+	cs.Tables.Deduct(other.Tables)
+	cs.BlobFiles.Deduct(other.BlobFiles)
+	cs.Other.Deduct(other.Other)
+}
+
+func (cs FileCountsAndSizes) String() string {
+	return redact.StringWithoutMarkers(cs)
+}
+
+// SafeFormat implements redact.SafeFormatter.
+func (cs FileCountsAndSizes) SafeFormat(w redact.SafePrinter, verb rune) {
+	if cs == (FileCountsAndSizes{}) {
+		w.Printf("no files")
+		return
+	}
+	first := true
+	if cs.Tables != (TableCountsAndSizes{}) {
+		first = false
+		w.Printf("tables: %s", cs.Tables)
+	}
+	if cs.BlobFiles != (BlobFileCountsAndSizes{}) {
+		if !first {
+			w.Printf("; ")
+			first = false
+		}
+		w.Printf("blob files: %s", cs.BlobFiles)
+	}
+	if cs.Other != (CountAndSize{}) {
+		if !first {
+			w.Printf("; ")
+			first = false
+		}
+		w.Printf("other: %s", cs.Other)
+	}
+}
diff --git a/metrics/doc.go b/metrics/doc.go
new file mode 100644
index 00000000..5cee002e
--- /dev/null
+++ b/metrics/doc.go
@@ -0,0 +1,6 @@
+// Copyright 2025 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+// Package metrics defines types and helpers used for metrics.
+package metrics
diff --git a/obsolete_files.go b/obsolete_files.go
index ec5b688a..74fb5694 100644
--- a/obsolete_files.go
+++ b/obsolete_files.go
@@ -6,21 +6,16 @@ package pebble
 
 import (
 	"cmp"
-	"context"
-	"runtime/pprof"
 	"slices"
-	"sync"
-	"time"
 
-	"github.com/cockroachdb/crlib/crtime"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/errors/oserror"
 	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/deletepacer"
 	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/objstorage"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/cockroachdb/pebble/wal"
-	"github.com/cockroachdb/tokenbucket"
 )
 
 // Cleaner exports the base.Cleaner type.
@@ -32,331 +27,82 @@ type DeleteCleaner = base.DeleteCleaner
 // ArchiveCleaner exports the base.ArchiveCleaner type.
 type ArchiveCleaner = base.ArchiveCleaner
 
-type cleanupManager struct {
-	opts        *Options
-	objProvider objstorage.Provider
-	deletePacer *deletionPacer
-
-	// jobsCh is used as the cleanup job queue.
-	jobsCh chan *cleanupJob
-	// waitGroup is used to wait for the background goroutine to exit.
-	waitGroup sync.WaitGroup
-	// stopCh is closed when the pacer is disabled after closing the cleanup manager.
-	stopCh chan struct{}
-
-	mu struct {
-		sync.Mutex
-		// totalJobs is the total number of enqueued jobs (completed or in progress).
-		totalJobs              int
-		completedStats         obsoleteObjectStats
-		completedJobs          int
-		completedJobsCond      sync.Cond
-		jobsQueueWarningIssued bool
-	}
-}
-
-// CompletedStats returns the stats summarizing objects deleted. The returned
-// stats increase monotonically over the lifetime of the DB.
-func (m *cleanupManager) CompletedStats() obsoleteObjectStats {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	return m.mu.completedStats
-}
-
-// 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 {
-	fileType base.FileType
-	fs       vfs.FS
-	path     string
-	fileNum  base.DiskFileNum
-	fileSize uint64 // approx for log files
-	isLocal  bool
-}
-
-func (of *obsoleteFile) needsPacing() bool {
-	// We only need to pace local objects--sstables and blob files.
-	return of.isLocal && (of.fileType == base.FileTypeTable || of.fileType == base.FileTypeBlob)
-}
-
-type cleanupJob struct {
-	jobID         JobID
-	obsoleteFiles []obsoleteFile
-	stats         obsoleteObjectStats
-}
-
-// openCleanupManager creates a cleanupManager and starts its background goroutine.
-// The cleanupManager must be Close()d.
-func openCleanupManager(
-	opts *Options, objProvider objstorage.Provider, getDeletePacerInfo func() deletionPacerInfo,
-) *cleanupManager {
-	cm := &cleanupManager{
-		opts:        opts,
-		objProvider: objProvider,
-		deletePacer: newDeletionPacer(
-			crtime.NowMono(),
-			opts.FreeSpaceThresholdBytes,
-			opts.TargetByteDeletionRate,
-			opts.FreeSpaceTimeframe,
-			opts.ObsoleteBytesMaxRatio,
-			opts.ObsoleteBytesTimeframe,
-			getDeletePacerInfo,
-		),
-		jobsCh: make(chan *cleanupJob, jobsQueueDepth),
-		stopCh: make(chan struct{}),
-	}
-	cm.mu.completedJobsCond.L = &cm.mu.Mutex
-	cm.waitGroup.Add(1)
-
-	go func() {
-		pprof.Do(context.Background(), gcLabels, func(context.Context) {
-			cm.mainLoop()
-		})
-	}()
-
-	return cm
-}
-
-// Close stops the background goroutine, waiting until all queued jobs are completed.
-// Delete pacing is disabled for the remaining jobs.
-func (cm *cleanupManager) Close() {
-	close(cm.jobsCh)
-	close(cm.stopCh)
-	cm.waitGroup.Wait()
-}
-
-// EnqueueJob adds a cleanup job to the manager's queue.
-func (cm *cleanupManager) EnqueueJob(
-	jobID JobID, obsoleteFiles []obsoleteFile, stats obsoleteObjectStats,
-) {
-	job := &cleanupJob{
-		jobID:         jobID,
-		obsoleteFiles: obsoleteFiles,
-		stats:         stats,
-	}
-
-	// Report deleted bytes to the pacer, which can use this data to potentially
-	// increase the deletion rate to keep up. We want to do this at enqueue time
-	// rather than when we get to the job, otherwise the reported bytes will be
-	// subject to the throttling rate which defeats the purpose.
-	var pacingBytes uint64
-	for _, of := range obsoleteFiles {
-		if of.needsPacing() {
-			pacingBytes += of.fileSize
-		}
-	}
-	if pacingBytes > 0 {
-		cm.deletePacer.ReportDeletion(crtime.NowMono(), pacingBytes)
-	}
-
-	cm.mu.Lock()
-	cm.mu.totalJobs++
-	cm.maybeLogLocked()
-	cm.mu.Unlock()
-
-	cm.jobsCh <- job
-}
-
-// Wait until the completion of all jobs that were already queued.
-//
-// Does not wait for jobs that are enqueued during the call.
-//
-// Note that DB.mu should not be held while calling this method; the background
-// goroutine needs to acquire DB.mu to update deleted table metrics.
-func (cm *cleanupManager) Wait() {
-	cm.mu.Lock()
-	defer cm.mu.Unlock()
-	n := cm.mu.totalJobs
-	for cm.mu.completedJobs < n {
-		cm.mu.completedJobsCond.Wait()
-	}
-}
-
-// mainLoop runs the manager's background goroutine.
-func (cm *cleanupManager) mainLoop() {
-	defer cm.waitGroup.Done()
-
-	paceTimer := time.NewTimer(time.Duration(0))
-	defer paceTimer.Stop()
-
-	var tb tokenbucket.TokenBucket
-	// Use a token bucket with 1 token / second refill rate and 1 token burst.
-	tb.Init(1.0, 1.0)
-	for job := range cm.jobsCh {
-		cm.deleteObsoleteFilesInJob(job, &tb, paceTimer)
-		cm.mu.Lock()
-		cm.mu.completedJobs++
-		cm.mu.completedStats.Add(job.stats)
-		cm.mu.completedJobsCond.Broadcast()
-		cm.maybeLogLocked()
-		cm.mu.Unlock()
-	}
+func openDeletePacer(
+	opts *Options, objProvider objstorage.Provider, diskFreeSpaceFn deletepacer.DiskFreeSpaceFn,
+) *deletepacer.DeletePacer {
+	return deletepacer.Open(
+		opts.DeletionPacing,
+		opts.Logger,
+		diskFreeSpaceFn,
+		func(of deletepacer.ObsoleteFile, jobID int) {
+			deleteObsoleteFile(opts.Cleaner, objProvider, opts.EventListener, of, jobID)
+		},
+	)
 }
 
-// deleteObsoleteFilesInJob deletes all obsolete files in the given job. If tb
-// and paceTimer are provided, files that need pacing will be throttled
-// according to the deletion rate. If tb is nil, files are deleted immediately
-// without pacing (used when the cleanup manager is being closed).
-func (cm *cleanupManager) deleteObsoleteFilesInJob(
-	job *cleanupJob, tb *tokenbucket.TokenBucket, paceTimer *time.Timer,
+// deleteObsoleteFile deletes a file or object, once the delete pacer decided it is time.
+func deleteObsoleteFile(
+	cleaner Cleaner,
+	objProvider objstorage.Provider,
+	eventListener *EventListener,
+	of deletepacer.ObsoleteFile,
+	jobID int,
 ) {
-	for _, of := range job.obsoleteFiles {
-		switch of.fileType {
-		case base.FileTypeTable, base.FileTypeBlob:
-			if tb != nil {
-				cm.maybePace(tb, &of, paceTimer)
-			}
-			cm.deleteObsoleteObject(of.fileType, job.jobID, of.fileNum)
-		default:
-			cm.deleteObsoleteFile(of.fs, of.fileType, job.jobID, of.path, of.fileNum)
+	path := of.Path
+	var err error
+	if of.FileType == base.FileTypeTable || of.FileType == base.FileTypeBlob {
+		var meta objstorage.ObjectMetadata
+		meta, err = objProvider.Lookup(of.FileType, of.FileNum)
+		if err != nil {
+			path = "<nil>"
+		} else {
+			path = objProvider.Path(meta)
+			err = objProvider.Remove(of.FileType, of.FileNum)
 		}
-	}
-}
-
-// maybePace sleeps before deleting an object if appropriate. It is always
-// called from the background goroutine.
-func (cm *cleanupManager) maybePace(
-	tb *tokenbucket.TokenBucket, of *obsoleteFile, paceTimer *time.Timer,
-) {
-	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
-		// tokens. We don't want that if the pacer decided throttling should be
-		// disabled.
-		return
-	}
-	// Wait for tokens. We use a token bucket instead of sleeping outright because
-	// the token bucket accumulates up to one second of unused tokens.
-	for {
-		ok, d := tb.TryToFulfill(tokenbucket.Tokens(tokens))
-		if ok {
-			break
+		if objProvider.IsNotExistError(err) {
+			return
 		}
-		paceTimer.Reset(d)
-		select {
-		case <-paceTimer.C:
-		case <-cm.stopCh:
-			// The cleanup manager is being closed. Delete without pacing.
+	} else {
+		// TODO(peter): need to handle this error, probably by re-adding the
+		// file that couldn't be deleted to one of the obsolete slices map.
+		err := cleaner.Clean(of.FS, of.FileType, path)
+		if oserror.IsNotExist(err) {
 			return
 		}
 	}
-}
 
-// deleteObsoleteFile deletes a (non-object) file that is no longer needed.
-func (cm *cleanupManager) deleteObsoleteFile(
-	fs vfs.FS, fileType base.FileType, jobID JobID, path string, fileNum base.DiskFileNum,
-) {
-	// TODO(peter): need to handle this error, probably by re-adding the
-	// file that couldn't be deleted to one of the obsolete slices map.
-	err := cm.opts.Cleaner.Clean(fs, fileType, path)
-	if oserror.IsNotExist(err) {
-		return
-	}
-
-	switch fileType {
-	case base.FileTypeLog:
-		cm.opts.EventListener.WALDeleted(WALDeleteInfo{
-			JobID:   int(jobID),
+	switch of.FileType {
+	case base.FileTypeTable:
+		eventListener.TableDeleted(TableDeleteInfo{
+			JobID:   jobID,
 			Path:    path,
-			FileNum: fileNum,
+			FileNum: of.FileNum,
 			Err:     err,
 		})
-	case base.FileTypeManifest:
-		cm.opts.EventListener.ManifestDeleted(ManifestDeleteInfo{
-			JobID:   int(jobID),
+	case base.FileTypeBlob:
+		eventListener.BlobFileDeleted(BlobFileDeleteInfo{
+			JobID:   jobID,
 			Path:    path,
-			FileNum: fileNum,
+			FileNum: of.FileNum,
 			Err:     err,
 		})
-	case base.FileTypeTable, base.FileTypeBlob:
-		panic("invalid deletion of object file")
-	}
-}
-
-func (cm *cleanupManager) deleteObsoleteObject(
-	fileType base.FileType, jobID JobID, fileNum base.DiskFileNum,
-) {
-	if fileType != base.FileTypeTable && fileType != base.FileTypeBlob {
-		panic("not an object")
-	}
-
-	var path string
-	meta, err := cm.objProvider.Lookup(fileType, fileNum)
-	if err != nil {
-		path = "<nil>"
-	} else {
-		path = cm.objProvider.Path(meta)
-		err = cm.objProvider.Remove(fileType, fileNum)
-	}
-	if cm.objProvider.IsNotExistError(err) {
-		return
-	}
-
-	switch fileType {
-	case base.FileTypeTable:
-		cm.opts.EventListener.TableDeleted(TableDeleteInfo{
-			JobID:   int(jobID),
+	case base.FileTypeLog:
+		eventListener.WALDeleted(WALDeleteInfo{
+			JobID:   jobID,
 			Path:    path,
-			FileNum: fileNum,
+			FileNum: of.FileNum,
 			Err:     err,
 		})
-	case base.FileTypeBlob:
-		cm.opts.EventListener.BlobFileDeleted(BlobFileDeleteInfo{
-			JobID:   int(jobID),
+	case base.FileTypeManifest:
+		eventListener.ManifestDeleted(ManifestDeleteInfo{
+			JobID:   jobID,
 			Path:    path,
-			FileNum: fileNum,
+			FileNum: of.FileNum,
 			Err:     err,
 		})
 	}
 }
 
-// maybeLogLocked issues a log if the job queue gets 75% full and issues a log
-// when the job queue gets back to less than 10% full.
-//
-// Must be called with cm.mu locked.
-func (cm *cleanupManager) maybeLogLocked() {
-
-	jobsInQueue := cm.mu.totalJobs - cm.mu.completedJobs
-
-	if !cm.mu.jobsQueueWarningIssued && jobsInQueue > jobsQueueHighThreshold {
-		cm.mu.jobsQueueWarningIssued = true
-		cm.opts.Logger.Infof("cleanup falling behind; job queue has over %d jobs", jobsQueueHighThreshold)
-	}
-
-	if cm.mu.jobsQueueWarningIssued && jobsInQueue < jobsQueueLowThreshold {
-		cm.mu.jobsQueueWarningIssued = false
-		cm.opts.Logger.Infof("cleanup back to normal; job queue has under %d jobs", jobsQueueLowThreshold)
-	}
-}
-
-func (d *DB) getDeletionPacerInfo() deletionPacerInfo {
-	var pacerInfo deletionPacerInfo
-	// Call GetDiskUsage after every file deletion. This may seem inefficient,
-	// but in practice this was observed to take constant time, regardless of
-	// volume size used, at least on linux with ext4 and zfs. All invocations
-	// take 10 microseconds or less.
-	pacerInfo.freeBytes = d.calculateDiskAvailableBytes()
-	d.mu.Lock()
-	m := &d.mu.versions.metrics
-	pacerInfo.obsoleteBytes = m.Table.ObsoleteSize + m.BlobFiles.ObsoleteSize
-	total := m.Total()
-	d.mu.Unlock()
-	pacerInfo.liveBytes = uint64(total.AggregateSize())
-	return pacerInfo
-}
-
 // scanObsoleteFiles compares the provided directory listing to the set of
 // known, in-use files to find files no longer needed and adds those to the
 // internal lists of obsolete files. Note that the files are not actually
@@ -383,26 +129,26 @@ func (d *DB) scanObsoleteFiles(list []string, flushableIngests []*ingestedFlusha
 
 	manifestFileNum := d.mu.versions.manifestFileNum
 
-	var obsoleteTables []obsoleteFile
-	var obsoleteBlobs []obsoleteFile
-	var obsoleteOptions []obsoleteFile
-	var obsoleteManifests []obsoleteFile
+	var obsoleteTables []deletepacer.ObsoleteFile
+	var obsoleteBlobs []deletepacer.ObsoleteFile
+	var obsoleteOptions []deletepacer.ObsoleteFile
+	var obsoleteManifests []deletepacer.ObsoleteFile
 
 	for _, filename := range list {
 		fileType, diskFileNum, ok := base.ParseFilename(d.opts.FS, filename)
 		if !ok {
 			continue
 		}
-		makeObsoleteFile := func() obsoleteFile {
-			of := obsoleteFile{
-				fileType: fileType,
-				fs:       d.opts.FS,
-				path:     d.opts.FS.PathJoin(d.dirname, filename),
-				fileNum:  diskFileNum,
-				isLocal:  true,
+		makeObsoleteFile := func() deletepacer.ObsoleteFile {
+			of := deletepacer.ObsoleteFile{
+				FileType: fileType,
+				FS:       d.opts.FS,
+				Path:     d.opts.FS.PathJoin(d.dirname, filename),
+				FileNum:  diskFileNum,
+				IsLocal:  true,
 			}
 			if stat, err := d.opts.FS.Stat(filename); err == nil {
-				of.fileSize = uint64(stat.Size())
+				of.FileSize = uint64(stat.Size())
 			}
 			return of
 		}
@@ -433,15 +179,15 @@ func (d *DB) scanObsoleteFiles(list []string, flushableIngests []*ingestedFlusha
 			// Ignore object types we don't know about.
 			continue
 		}
-		of := obsoleteFile{
-			fileType: obj.FileType,
-			fs:       d.opts.FS,
-			path:     base.MakeFilepath(d.opts.FS, d.dirname, obj.FileType, obj.DiskFileNum),
-			fileNum:  obj.DiskFileNum,
-			isLocal:  true,
+		of := deletepacer.ObsoleteFile{
+			FileType: obj.FileType,
+			FS:       d.opts.FS,
+			Path:     base.MakeFilepath(d.opts.FS, d.dirname, obj.FileType, obj.DiskFileNum),
+			FileNum:  obj.DiskFileNum,
+			IsLocal:  true,
 		}
 		if size, err := d.objProvider.Size(obj); err == nil {
-			of.fileSize = uint64(size)
+			of.FileSize = uint64(size)
 		}
 		if obj.FileType == base.FileTypeTable {
 			obsoleteTables = append(obsoleteTables, of)
@@ -469,7 +215,7 @@ func (d *DB) disableFileDeletions() {
 	d.mu.fileDeletions.disableCount++
 	d.mu.Unlock()
 	defer d.mu.Lock()
-	d.cleanupManager.Wait()
+	d.deletePacer.WaitForTesting()
 }
 
 // enableFileDeletions enables previously disabled file deletions. A cleanup job
@@ -529,7 +275,7 @@ func (d *DB) deleteObsoleteFiles(jobID JobID) {
 		}
 	}
 
-	var obsoleteManifests []obsoleteFile
+	var obsoleteManifests []deletepacer.ObsoleteFile
 	manifestsToDelete := len(d.mu.versions.obsoleteManifests) - d.opts.NumPrevManifest
 	if manifestsToDelete > 0 {
 		obsoleteManifests = d.mu.versions.obsoleteManifests[:manifestsToDelete]
@@ -542,14 +288,8 @@ func (d *DB) deleteObsoleteFiles(jobID JobID) {
 	obsoleteOptions := d.mu.versions.obsoleteOptions
 	d.mu.versions.obsoleteOptions = nil
 
-	// Compute the stats for the files 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 files by diffing these
-	// stats and the stats reported by the cleanup manager.
-	var objectStats obsoleteObjectStats
-	objectStats.tablesAll, objectStats.tablesLocal = calculateObsoleteObjectStats(obsoleteTables)
-	objectStats.blobFilesAll, objectStats.blobFilesLocal = calculateObsoleteObjectStats(obsoleteBlobs)
-	d.mu.fileDeletions.queuedStats.Add(objectStats)
+	// Update the obsolete object metrics in d.mu.versions.metrics. These metrics
+	// will be combined with the metrics from the delete pacer in DB.Metrics().
 	d.mu.versions.updateObsoleteObjectMetricsLocked()
 
 	// Release d.mu while preparing the cleanup job and possibly waiting.
@@ -558,32 +298,32 @@ func (d *DB) deleteObsoleteFiles(jobID JobID) {
 	defer d.mu.Lock()
 
 	n := len(obsoleteLogs) + len(obsoleteTables) + len(obsoleteBlobs) + len(obsoleteManifests) + len(obsoleteOptions)
-	filesToDelete := make([]obsoleteFile, 0, n)
+	filesToDelete := make([]deletepacer.ObsoleteFile, 0, n)
 	filesToDelete = append(filesToDelete, obsoleteManifests...)
 	filesToDelete = append(filesToDelete, obsoleteOptions...)
 	filesToDelete = append(filesToDelete, obsoleteTables...)
 	filesToDelete = append(filesToDelete, obsoleteBlobs...)
 	for _, f := range obsoleteLogs {
-		filesToDelete = append(filesToDelete, obsoleteFile{
-			fileType: base.FileTypeLog,
-			fs:       f.FS,
-			path:     f.Path,
-			fileNum:  base.DiskFileNum(f.NumWAL),
-			fileSize: f.ApproxFileSize,
-			isLocal:  true,
+		filesToDelete = append(filesToDelete, deletepacer.ObsoleteFile{
+			FileType: base.FileTypeLog,
+			FS:       f.FS,
+			Path:     f.Path,
+			FileNum:  base.DiskFileNum(f.NumWAL),
+			FileSize: f.ApproxFileSize,
+			IsLocal:  true,
 		})
 	}
 	for _, f := range obsoleteTables {
-		d.fileCache.Evict(f.fileNum, base.FileTypeTable)
+		d.fileCache.Evict(f.FileNum, base.FileTypeTable)
 	}
 	for _, f := range obsoleteBlobs {
-		d.fileCache.Evict(f.fileNum, base.FileTypeBlob)
+		d.fileCache.Evict(f.FileNum, base.FileTypeBlob)
 	}
 	if len(filesToDelete) > 0 {
-		d.cleanupManager.EnqueueJob(jobID, filesToDelete, objectStats)
+		d.deletePacer.Enqueue(int(jobID), filesToDelete...)
 	}
 	if d.opts.private.testingAlwaysWaitForCleanup {
-		d.cleanupManager.Wait()
+		d.deletePacer.WaitForTesting()
 	}
 }
 
@@ -595,20 +335,20 @@ func (d *DB) maybeScheduleObsoleteObjectDeletion() {
 	}
 }
 
-func mergeObsoleteFiles(a, b []obsoleteFile) []obsoleteFile {
+func mergeObsoleteFiles(a, b []deletepacer.ObsoleteFile) []deletepacer.ObsoleteFile {
 	if len(b) == 0 {
 		return a
 	}
 
 	a = append(a, b...)
 	slices.SortFunc(a, cmpObsoleteFileNumbers)
-	return slices.CompactFunc(a, func(a, b obsoleteFile) bool {
-		return a.fileNum == b.fileNum
+	return slices.CompactFunc(a, func(a, b deletepacer.ObsoleteFile) bool {
+		return a.FileNum == b.FileNum
 	})
 }
 
-func cmpObsoleteFileNumbers(a, b obsoleteFile) int {
-	return cmp.Compare(a.fileNum, b.fileNum)
+func cmpObsoleteFileNumbers(a, b deletepacer.ObsoleteFile) int {
+	return cmp.Compare(a.FileNum, b.FileNum)
 }
 
 // objectInfo describes an object in object storage (either a sstable or a blob
@@ -618,63 +358,17 @@ type objectInfo struct {
 	isLocal bool
 }
 
-func (o objectInfo) asObsoleteFile(fs vfs.FS, fileType base.FileType, dirname string) obsoleteFile {
-	return obsoleteFile{
-		fileType: fileType,
-		fs:       fs,
-		path:     base.MakeFilepath(fs, dirname, fileType, o.FileNum),
-		fileNum:  o.FileNum,
-		fileSize: o.FileSize,
-		isLocal:  o.isLocal,
-	}
-}
-
-func calculateObsoleteObjectStats(files []obsoleteFile) (total, local countAndSize) {
-	for _, of := range files {
-		if of.isLocal {
-			local.count++
-			local.size += of.fileSize
-		}
-		total.count++
-		total.size += of.fileSize
+func (o objectInfo) asObsoleteFile(
+	fs vfs.FS, fileType base.FileType, dirname string,
+) deletepacer.ObsoleteFile {
+	return deletepacer.ObsoleteFile{
+		FileType: fileType,
+		FS:       fs,
+		Path:     base.MakeFilepath(fs, dirname, fileType, o.FileNum),
+		FileNum:  o.FileNum,
+		FileSize: o.FileSize,
+		IsLocal:  o.isLocal,
 	}
-	return total, local
-}
-
-type obsoleteObjectStats struct {
-	tablesLocal    countAndSize
-	tablesAll      countAndSize
-	blobFilesLocal countAndSize
-	blobFilesAll   countAndSize
-}
-
-func (s *obsoleteObjectStats) Add(other obsoleteObjectStats) {
-	s.tablesLocal.Add(other.tablesLocal)
-	s.tablesAll.Add(other.tablesAll)
-	s.blobFilesLocal.Add(other.blobFilesLocal)
-	s.blobFilesAll.Add(other.blobFilesAll)
-}
-
-func (s *obsoleteObjectStats) Sub(other obsoleteObjectStats) {
-	s.tablesLocal.Sub(other.tablesLocal)
-	s.tablesAll.Sub(other.tablesAll)
-	s.blobFilesLocal.Sub(other.blobFilesLocal)
-	s.blobFilesAll.Sub(other.blobFilesAll)
-}
-
-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 makeZombieObjects() zombieObjects {
diff --git a/open.go b/open.go
index e284e14b..53c4763e 100644
--- a/open.go
+++ b/open.go
@@ -103,18 +103,13 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 	}
 
 	// Open the database and WAL directories first.
-	walDirname, secondaryWalDirName, dataDir, err := prepareAndOpenDirs(dirname, opts)
-	if err != nil {
-		return nil, errors.Wrapf(err, "error opening database at %q", dirname)
-	}
-	defer maybeCleanUp(dataDir.Close)
-
-	// Lock the database directory.
-	fileLock, err := base.AcquireOrValidateDirectoryLock(opts.Lock, dirname, opts.FS)
+	dirs, err := prepareOpenAndLockDirs(dirname, opts)
 	if err != nil {
+		err = errors.Wrapf(err, "error opening database at %q", dirname)
+		err = errors.CombineErrors(err, dirs.Close())
 		return nil, err
 	}
-	defer maybeCleanUp(fileLock.Close)
+	defer maybeCleanUp(dirs.Close)
 
 	rs, err := recoverState(opts, dirname)
 	if err != nil {
@@ -171,8 +166,7 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 		split:               opts.Comparer.Split,
 		abbreviatedKey:      opts.Comparer.AbbreviatedKey,
 		largeBatchThreshold: (opts.MemTableSize - uint64(memTableEmptySize)) / 2,
-		dataDirLock:         fileLock,
-		dataDir:             dataDir,
+		dirs:                dirs,
 		objProvider:         rs.objProvider,
 		closed:              new(atomic.Value),
 		closedCh:            make(chan struct{}),
@@ -213,8 +207,8 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 					t.arenaBuf = manual.Buf{}
 				}
 			}
-			if d.cleanupManager != nil {
-				d.cleanupManager.Close()
+			if d.deletePacer != nil {
+				d.deletePacer.Close()
 			}
 			if d.mu.versions.manifestFile != nil {
 				_ = d.mu.versions.manifestFile.Close()
@@ -300,8 +294,8 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 	})
 
 	walOpts := wal.Options{
-		Primary:              wal.Dir{FS: opts.FS, Dirname: walDirname},
-		Secondary:            wal.Dir{},
+		Primary:              dirs.WALPrimary,
+		Secondary:            dirs.WALSecondary,
 		MinUnflushedWALNum:   wal.NumWAL(d.mu.versions.minUnflushedLogNum),
 		MaxNumRecyclableLogs: opts.MemTableStopWritesThreshold + 1,
 		NoSyncOnClose:        opts.NoSyncOnClose,
@@ -314,66 +308,20 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 		EventListener:        walEventListenerAdaptor{l: opts.EventListener},
 		WriteWALSyncOffsets:  func() bool { return d.FormatMajorVersion() >= FormatWALSyncChunks },
 	}
-	// Ensure we release the WAL directory locks if we fail to open the
-	// database. If we fail before initializing the WAL manager, this defer is
-	// responsible for releasing the locks. If we fail after initializing the
-	// WAL manager, closing the WAL manager will release the locks.
-	//
-	// TODO(jackson): Open's cleanup error handling logic is convoluted; can we
-	// simplify it?
-	defer maybeCleanUp(func() (err error) {
-		if d.mu.log.manager == nil {
-			if walOpts.Primary.Lock != nil {
-				err = errors.CombineErrors(err, walOpts.Primary.Lock.Close())
-			}
-			if walOpts.Secondary.Lock != nil {
-				err = errors.CombineErrors(err, walOpts.Secondary.Lock.Close())
-			}
-			return err
-		}
-		return nil
-	})
-
-	// Lock the dedicated WAL directory, if configured.
-	if walDirname != dirname {
-		walOpts.Primary.Lock, err = base.AcquireOrValidateDirectoryLock(opts.WALDirLock, walDirname, opts.FS)
-		if err != nil {
-			return nil, err
-		}
-	}
 	if !opts.ReadOnly && opts.WALFailover != nil {
-		walOpts.Secondary = opts.WALFailover.Secondary
-		// Lock the secondary WAL directory, if distinct from the data directory
-		// and primary WAL directory.
-		if secondaryWalDirName != dirname && secondaryWalDirName != walDirname {
-			walOpts.Secondary.Lock, err = base.AcquireOrValidateDirectoryLock(
-				opts.WALFailover.Secondary.Lock, secondaryWalDirName, opts.WALFailover.Secondary.FS)
-			if err != nil {
-				return nil, err
-			}
-		}
-		walOpts.Secondary.Dirname = secondaryWalDirName
 		walOpts.FailoverOptions = opts.WALFailover.FailoverOptions
 		walOpts.FailoverWriteAndSyncLatency = prometheus.NewHistogram(prometheus.HistogramOpts{
 			Buckets: FsyncLatencyBuckets,
 		})
 	}
 	walDirs := walOpts.Dirs()
-	walRecoveryLocks := make([]*base.DirLock, len(opts.WALRecoveryDirs))
-	defer func() {
-		// We only need the recovery WALs during Open, so we can release
-		// the locks after the WALs have been scanned.
-		for _, l := range walRecoveryLocks {
-			if l != nil {
-				_ = l.Close()
-			}
-		}
-	}()
-	for i, dir := range opts.WALRecoveryDirs {
+	var recoveryDirLocks base.DirLockSet
+	defer func() { _ = recoveryDirLocks.Close() }()
+	for _, dir := range opts.WALRecoveryDirs {
 		dir.Dirname = resolveStorePath(dirname, dir.Dirname)
 		if dir.Dirname != dirname {
 			// Acquire a lock on the WAL recovery directory.
-			walRecoveryLocks[i], err = base.AcquireOrValidateDirectoryLock(dir.Lock, dir.Dirname, dir.FS)
+			_, err = recoveryDirLocks.AcquireOrValidate(dir.Lock, dir.Dirname, dir.FS)
 			if err != nil {
 				return nil, errors.Wrapf(err, "error acquiring lock on WAL recovery directory %q", dir.Dirname)
 			}
@@ -430,7 +378,12 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 	defer maybeCleanUp(walManager.Close)
 	d.mu.log.manager = walManager
 
-	d.cleanupManager = openCleanupManager(opts, d.objProvider, d.getDeletionPacerInfo)
+	// The delete pacer will call calculateDiskAvailableBytes (and thus
+	// GetDiskUsage) O(number of file deletions) times. This may seem inefficient,
+	// but in practice this was observed to take constant time, regardless of
+	// volume size used, at least on linux with ext4 and zfs. All invocations take
+	// 10 microseconds or less.
+	d.deletePacer = openDeletePacer(d.opts, d.objProvider, d.calculateDiskAvailableBytes)
 
 	fileCacheSize := FileCacheSize(opts.MaxOpenFiles)
 	if opts.FileCache == nil {
@@ -531,7 +484,7 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 		if err := opts.FS.Rename(tmpPath, optionsPath); err != nil {
 			return nil, err
 		}
-		if err := d.dataDir.Sync(); err != nil {
+		if err := d.dirs.DataDir.Sync(); err != nil {
 			return nil, err
 		}
 
@@ -621,62 +574,105 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 	return d, nil
 }
 
-// prepareAndOpenDirs opens the directories for the store (and creates them if
-// necessary).
+// resolvedDirs is a set of resolved directory paths and locks.
+type resolvedDirs struct {
+	DirLocks     base.DirLockSet
+	DataDir      vfs.File
+	WALPrimary   wal.Dir
+	WALSecondary wal.Dir
+}
+
+// Close closes the data directory and the directory locks.
+func (d *resolvedDirs) Close() error {
+	var err error
+	if d.DataDir != nil {
+		err = errors.CombineErrors(err, d.DataDir.Close())
+	}
+	err = errors.CombineErrors(err, d.DirLocks.Close())
+	*d = resolvedDirs{}
+	return err
+}
+
+// prepareOpenAndLockDirs resolves the various directory paths indicated within
+// Options (substituting {store_path} relative paths as necessary), creates the
+// directories if they don't exist, and acquires directory locks as necessary.
 //
-// Returns an error if ReadOnly is set and the directories don't exist.
-func prepareAndOpenDirs(
-	dirname string, opts *Options,
-) (walDirname string, secondaryWalDirName string, dataDir vfs.File, err error) {
-	walDirname = dirname
+// Returns an error if ReadOnly is set and the directories don't exist. Always
+// returns a non-nil resolvedDirs that may be closed to release all resources
+// acquired before any error was encountered.
+func prepareOpenAndLockDirs(dirname string, opts *Options) (dirs *resolvedDirs, err error) {
+	dirs = &resolvedDirs{}
+	dirs.WALPrimary.Dirname = dirname
+	dirs.WALPrimary.FS = opts.FS
 	if opts.WALDir != "" {
-		walDirname = resolveStorePath(dirname, opts.WALDir)
+		dirs.WALPrimary.Dirname = resolveStorePath(dirname, opts.WALDir)
 	}
 	if opts.WALFailover != nil {
-		secondaryWalDirName = resolveStorePath(dirname, opts.WALFailover.Secondary.Dirname)
+		dirs.WALSecondary.Dirname = resolveStorePath(dirname, opts.WALFailover.Secondary.Dirname)
+		dirs.WALSecondary.FS = opts.WALFailover.Secondary.FS
 	}
 
 	// Create directories if needed.
 	if !opts.ReadOnly {
 		f, err := mkdirAllAndSyncParents(opts.FS, dirname)
 		if err != nil {
-			return "", "", nil, err
+			return dirs, err
 		}
 		f.Close()
-		if walDirname != dirname {
-			f, err := mkdirAllAndSyncParents(opts.FS, walDirname)
+		if dirs.WALPrimary.Dirname != dirname {
+			f, err := mkdirAllAndSyncParents(opts.FS, dirs.WALPrimary.Dirname)
 			if err != nil {
-				return "", "", nil, err
+				return dirs, err
 			}
 			f.Close()
 		}
 		if opts.WALFailover != nil {
-			f, err := mkdirAllAndSyncParents(opts.WALFailover.Secondary.FS, secondaryWalDirName)
+			f, err := mkdirAllAndSyncParents(opts.WALFailover.Secondary.FS, dirs.WALSecondary.Dirname)
 			if err != nil {
-				return "", "", nil, err
+				return dirs, err
 			}
 			f.Close()
 		}
 	}
 
-	dataDir, err = opts.FS.OpenDir(dirname)
+	dirs.DataDir, err = opts.FS.OpenDir(dirname)
 	if err != nil {
 		if opts.ReadOnly && oserror.IsNotExist(err) {
-			return "", "", nil, errors.Errorf("pebble: database %q does not exist", dirname)
+			return dirs, errors.Errorf("pebble: database %q does not exist", dirname)
 		}
-		return "", "", nil, err
+		return dirs, err
 	}
-	if opts.ReadOnly && walDirname != dirname {
+	if opts.ReadOnly && dirs.WALPrimary.Dirname != dirname {
 		// Check that the wal dir exists.
-		walDir, err := opts.FS.OpenDir(walDirname)
+		walDir, err := opts.FS.OpenDir(dirs.WALPrimary.Dirname)
 		if err != nil {
-			dataDir.Close()
-			return "", "", nil, err
+			return dirs, err
 		}
 		walDir.Close()
 	}
 
-	return walDirname, secondaryWalDirName, dataDir, nil
+	// Lock the database directory.
+	_, err = dirs.DirLocks.AcquireOrValidate(opts.Lock, dirname, opts.FS)
+	if err != nil {
+		return dirs, err
+	}
+	// Lock the dedicated WAL directory, if configured.
+	if dirs.WALPrimary.Dirname != dirname {
+		dirs.WALPrimary.Lock, err = dirs.DirLocks.AcquireOrValidate(opts.WALDirLock, dirs.WALPrimary.Dirname, opts.FS)
+		if err != nil {
+			return dirs, err
+		}
+	}
+	// Lock the secondary WAL directory, if distinct from the data directory
+	// and primary WAL directory.
+	if opts.WALFailover != nil && dirs.WALSecondary.Dirname != dirname && dirs.WALSecondary.Dirname != dirs.WALPrimary.Dirname {
+		dirs.WALSecondary.Lock, err = dirs.DirLocks.AcquireOrValidate(
+			opts.WALFailover.Secondary.Lock, dirs.WALSecondary.Dirname, dirs.WALSecondary.FS)
+		if err != nil {
+			return dirs, err
+		}
+	}
+	return dirs, nil
 }
 
 // GetVersion returns the engine version string from the latest options
diff --git a/options.go b/options.go
index 14a33ac9..06b1d5a3 100644
--- a/options.go
+++ b/options.go
@@ -21,6 +21,7 @@ import (
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/cache"
+	"github.com/cockroachdb/pebble/internal/deletepacer"
 	"github.com/cockroachdb/pebble/internal/humanize"
 	"github.com/cockroachdb/pebble/internal/keyspan"
 	"github.com/cockroachdb/pebble/internal/manifest"
@@ -1127,7 +1128,8 @@ type Options struct {
 	// the current database state.
 	//
 	// If a previous WAL configuration may have stored WALs elsewhere but there
-	// is not a corresponding entry in WALRecoveryDirs, Open will error.
+	// is not a corresponding entry in WALRecoveryDirs, Open will error (unless
+	// Unsafe.AllowMissingWALDirs is true).
 	WALRecoveryDirs []wal.Dir
 
 	// WALMinSyncInterval is the minimum duration between syncs of the WAL. If
@@ -1142,41 +1144,11 @@ type Options struct {
 	// changing options dynamically?
 	WALMinSyncInterval func() time.Duration
 
-	// The controls below manage deletion pacing, which slows down
-	// deletions when compactions finish or when readers close and
-	// obsolete files must be cleaned up. Rapid deletion of many
-	// files simultaneously can increase disk latency on certain
-	// SSDs, and this functionality helps protect against that.
-
-	// TargetByteDeletionRate is the rate (in bytes per second) at which sstable file
-	// deletions are limited to (under normal circumstances).
-	//
-	// This value is only a best-effort target; the effective rate can be
-	// higher if deletions are falling behind or disk space is running low.
-	//
-	// A returned value of 0 disables deletion pacing (this is also the default).
-	TargetByteDeletionRate func() int
-
-	// FreeSpaceThresholdBytes specifies the minimum amount of free disk space that Pebble
-	// attempts to maintain. If free disk space drops below this threshold, deletions
-	// are accelerated above TargetByteDeletionRate until the threshold is restored.
-	// Default is 16GB.
-	FreeSpaceThresholdBytes uint64
-
-	// FreeSpaceTimeframe sets the duration (in seconds) within which Pebble attempts
-	// to restore the free disk space back to FreeSpaceThreshold. A lower value means
-	// more aggressive deletions. Default is 10s.
-	FreeSpaceTimeframe time.Duration
-
-	// ObsoleteBytesMaxRatio specifies the maximum allowed ratio of obsolete files to
-	// live files. If this ratio is exceeded, Pebble speeds up deletions above the
-	// TargetByteDeletionRate until the ratio is restored. Default is 0.20.
-	ObsoleteBytesMaxRatio float64
-
-	// ObsoleteBytesTimeframe sets the duration (in seconds) within which Pebble aims
-	// to restore the obsolete-to-live bytes ratio below ObsoleteBytesMaxRatio. A lower
-	// value means more aggressive deletions. Default is 300s.
-	ObsoleteBytesTimeframe time.Duration
+	// DeletionPacing manage deletion pacing, which slows down deletions when
+	// compactions finish or when readers close and obsolete files must be cleaned
+	// up. Rapid deletion of many files simultaneously can increase disk latency
+	// on certain SSDs, and this functionality helps protect against that.
+	DeletionPacing DeletionPacingOptions
 
 	// EnableSQLRowSpillMetrics specifies whether the Pebble instance will only be used
 	// to temporarily persist data spilled to disk for row-oriented SQL query execution.
@@ -1188,6 +1160,19 @@ type Options struct {
 	// preemptively reduce internal fragmentation when loaded into the block cache.
 	AllocatorSizeClasses []int
 
+	// Unsafe contains options that must be used very carefully and in exceptional
+	// circumstances.
+	Unsafe struct {
+		// AllowMissingWALDirs, if set to true, allows opening a DB when the WAL or
+		// WAL secondary directory was changed and the previous directory is not in
+		// WALRecoveryDirs. This can be used to move WALs without having to keep the
+		// previous directory in the options forever.
+		//
+		// CAUTION: Enabling this option will lead to data loss if the missing
+		// directory contained any WAL files that were not flushed to sstables.
+		AllowMissingWALDirs bool
+	}
+
 	// private options are only used by internal tests or are used internally
 	// for facilitating upgrade paths of unconfigurable functionality.
 	private struct {
@@ -1232,6 +1217,10 @@ type Options struct {
 	}
 }
 
+// DeletionPacingOptions contains configuration options contorlling the pacing
+// of deletion of obsolete files.
+type DeletionPacingOptions = deletepacer.Options
+
 // ValueSeparationPolicy controls the policy for separating values into
 // external blob files.
 type ValueSeparationPolicy struct {
@@ -1516,26 +1505,7 @@ func (o *Options) EnsureDefaults() {
 	if o.Cleaner == nil {
 		o.Cleaner = DeleteCleaner{}
 	}
-
-	if o.TargetByteDeletionRate == nil {
-		o.TargetByteDeletionRate = func() int { return 0 }
-	}
-
-	if o.FreeSpaceThresholdBytes == 0 {
-		o.FreeSpaceThresholdBytes = 16 << 30 // 16 GB
-	}
-
-	if o.FreeSpaceTimeframe == 0 {
-		o.FreeSpaceTimeframe = 10 * time.Second
-	}
-
-	if o.ObsoleteBytesMaxRatio == 0 {
-		o.ObsoleteBytesMaxRatio = 0.20
-	}
-
-	if o.ObsoleteBytesTimeframe == 0 {
-		o.ObsoleteBytesTimeframe = 300 * time.Second
-	}
+	o.DeletionPacing.EnsureDefaults()
 
 	if o.Experimental.DisableIngestAsFlushable == nil {
 		o.Experimental.DisableIngestAsFlushable = func() bool { return false }
@@ -1810,11 +1780,10 @@ func (o *Options) String() string {
 	fmt.Fprintf(&buf, "  max_open_files=%d\n", o.MaxOpenFiles)
 	fmt.Fprintf(&buf, "  mem_table_size=%d\n", o.MemTableSize)
 	fmt.Fprintf(&buf, "  mem_table_stop_writes_threshold=%d\n", o.MemTableStopWritesThreshold)
-	fmt.Fprintf(&buf, "  min_deletion_rate=%d\n", o.TargetByteDeletionRate())
-	fmt.Fprintf(&buf, "  free_space_threshold_bytes=%d\n", o.FreeSpaceThresholdBytes)
-	fmt.Fprintf(&buf, "  free_space_timeframe=%s\n", o.FreeSpaceTimeframe.String())
-	fmt.Fprintf(&buf, "  obsolete_bytes_max_ratio=%f\n", o.ObsoleteBytesMaxRatio)
-	fmt.Fprintf(&buf, "  obsolete_bytes_timeframe=%s\n", o.ObsoleteBytesTimeframe.String())
+	fmt.Fprintf(&buf, "  min_deletion_rate=%d\n", o.DeletionPacing.BaselineRate())
+	fmt.Fprintf(&buf, "  free_space_threshold_bytes=%d\n", o.DeletionPacing.FreeSpaceThresholdBytes)
+	fmt.Fprintf(&buf, "  free_space_timeframe=%s\n", o.DeletionPacing.FreeSpaceTimeframe.String())
+	fmt.Fprintf(&buf, "  obsolete_bytes_timeframe=%s\n", o.DeletionPacing.BacklogTimeframe.String())
 	fmt.Fprintf(&buf, "  merger=%s\n", o.Merger.Name)
 	if o.Experimental.MultiLevelCompactionHeuristic != nil {
 		fmt.Fprintf(&buf, "  multilevel_compaction_heuristic=%s\n", o.Experimental.MultiLevelCompactionHeuristic().String())
@@ -2181,19 +2150,19 @@ func (o *Options) Parse(s string, hooks *ParseHooks) error {
 				// Do nothing; option existed in older versions of pebble, and
 				// may be meaningful again eventually.
 			case "min_deletion_rate":
-				var rate int
-				rate, err = strconv.Atoi(value)
+				var rate uint64
+				rate, err = strconv.ParseUint(value, 10, 64)
 				if err == nil {
-					o.TargetByteDeletionRate = func() int { return rate }
+					o.DeletionPacing.BaselineRate = func() uint64 { return rate }
 				}
 			case "free_space_threshold_bytes":
-				o.FreeSpaceThresholdBytes, err = strconv.ParseUint(value, 10, 64)
+				o.DeletionPacing.FreeSpaceThresholdBytes, err = strconv.ParseUint(value, 10, 64)
 			case "free_space_timeframe":
-				o.FreeSpaceTimeframe, err = time.ParseDuration(value)
+				o.DeletionPacing.FreeSpaceTimeframe, err = time.ParseDuration(value)
 			case "obsolete_bytes_max_ratio":
-				o.ObsoleteBytesMaxRatio, err = strconv.ParseFloat(value, 64)
+				// No longer used.
 			case "obsolete_bytes_timeframe":
-				o.ObsoleteBytesTimeframe, err = time.ParseDuration(value)
+				o.DeletionPacing.BacklogTimeframe, err = time.ParseDuration(value)
 			case "min_flush_rate":
 				// Do nothing; option existed in older versions of pebble, and
 				// may be meaningful again eventually.
@@ -2527,6 +2496,11 @@ func (o *Options) checkWALDir(storeDir, walDir, errContext string) error {
 		}
 	}
 
+	if o.Unsafe.AllowMissingWALDirs {
+		o.Logger.Infof("directory %q may contain relevant WALs but is not in WALRecoveryDirs (AllowMissingWALDirs enabled)", walDir)
+		return nil
+	}
+
 	var buf bytes.Buffer
 	fmt.Fprintf(&buf, "\n  %s\n", errContext)
 	fmt.Fprintf(&buf, "  o.WALDir: %q\n", o.WALDir)
@@ -2537,6 +2511,7 @@ func (o *Options) checkWALDir(storeDir, walDir, errContext string) error {
 	for _, d := range o.WALRecoveryDirs {
 		fmt.Fprintf(&buf, "\n    %q", d.Dirname)
 	}
+
 	return ErrMissingWALRecoveryDir{Dir: walPath, ExtraInfo: buf.String()}
 }
 
diff --git a/pacer.go b/pacer.go
deleted file mode 100644
index 54ba97b4..00000000
--- a/pacer.go
+++ /dev/null
@@ -1,200 +0,0 @@
-// Copyright 2019 The LevelDB-Go and Pebble Authors. All rights reserved. Use
-// of this source code is governed by a BSD-style license that can be found in
-// the LICENSE file.
-
-package pebble
-
-import (
-	"sync"
-	"time"
-
-	"github.com/cockroachdb/crlib/crtime"
-)
-
-// deletionPacerInfo contains any info from the db necessary to make deletion
-// pacing decisions (to limit background IO usage so that it does not contend
-// with foreground traffic).
-type deletionPacerInfo struct {
-	freeBytes     uint64
-	obsoleteBytes uint64
-	liveBytes     uint64
-}
-
-// deletionPacer rate limits deletions of obsolete files. This is necessary to
-// prevent overloading the disk with too many deletions too quickly after a
-// large compaction, or an iterator close. On some SSDs, disk performance can be
-// negatively impacted if too many blocks are deleted very quickly, so this
-// mechanism helps mitigate that.
-type deletionPacer struct {
-	// If there are less than freeSpaceThreshold bytes of free space on
-	// disk, increase the pace of deletions such that we delete enough bytes to
-	// get back to the threshold within the freeSpaceTimeframe.
-	freeSpaceThreshold uint64
-	freeSpaceTimeframe time.Duration
-
-	// If the ratio of obsolete bytes to live bytes is greater than
-	// obsoleteBytesMaxRatio, increase the pace of deletions such that we delete
-	// enough bytes to get back to the threshold within the obsoleteBytesTimeframe.
-	obsoleteBytesMaxRatio  float64
-	obsoleteBytesTimeframe time.Duration
-
-	mu struct {
-		sync.Mutex
-
-		// history keeps rack of recent deletion history; it used to increase the
-		// deletion rate to match the pace of deletions.
-		history history
-	}
-
-	targetByteDeletionRate func() int
-
-	getInfo func() deletionPacerInfo
-}
-
-const deletePacerHistory = 5 * time.Minute
-
-// newDeletionPacer instantiates a new deletionPacer for use when deleting
-// obsolete files.
-//
-// targetByteDeletionRate is the rate (in bytes/sec) at which we want to
-// normally limit deletes (when we are not falling behind or running out of
-// space). A value of 0.0 disables pacing.
-func newDeletionPacer(
-	now crtime.Mono,
-	freeSpaceThreshold uint64,
-	targetByteDeletionRate func() int,
-	freeSpaceTimeframe time.Duration,
-	obsoleteBytesMaxRatio float64,
-	obsoleteBytesTimeframe time.Duration,
-	getInfo func() deletionPacerInfo,
-) *deletionPacer {
-	d := &deletionPacer{
-		freeSpaceThreshold: freeSpaceThreshold,
-		freeSpaceTimeframe: freeSpaceTimeframe,
-
-		obsoleteBytesMaxRatio:  obsoleteBytesMaxRatio,
-		obsoleteBytesTimeframe: obsoleteBytesTimeframe,
-
-		targetByteDeletionRate: targetByteDeletionRate,
-		getInfo:                getInfo,
-	}
-
-	d.mu.history.Init(now, deletePacerHistory)
-	return d
-}
-
-// ReportDeletion is used to report a deletion to the pacer. The pacer uses it
-// to keep track of the recent rate of deletions and potentially increase the
-// deletion rate accordingly.
-//
-// ReportDeletion is thread-safe.
-func (p *deletionPacer) ReportDeletion(now crtime.Mono, bytesToDelete uint64) {
-	p.mu.Lock()
-	defer p.mu.Unlock()
-	p.mu.history.Add(now, int64(bytesToDelete))
-}
-
-// PacingDelay returns the recommended pacing wait time (in seconds) for
-// deleting the given number of bytes.
-//
-// PacingDelay is thread-safe.
-func (p *deletionPacer) PacingDelay(now crtime.Mono, bytesToDelete uint64) (waitSeconds float64) {
-	targetByteDeletionRate := p.targetByteDeletionRate()
-	if targetByteDeletionRate == 0 {
-		// Pacing disabled.
-		return 0.0
-	}
-
-	baseRate := float64(targetByteDeletionRate)
-	// If recent deletion rate is more than our target, use that so that we don't
-	// fall behind.
-	historicRate := func() float64 {
-		p.mu.Lock()
-		defer p.mu.Unlock()
-		return float64(p.mu.history.Sum(now)) / deletePacerHistory.Seconds()
-	}()
-	if historicRate > baseRate {
-		baseRate = historicRate
-	}
-
-	// Apply heuristics to increase the deletion rate.
-	var extraRate float64
-	info := p.getInfo()
-	if info.freeBytes <= p.freeSpaceThreshold {
-		// Increase the rate so that we can free up enough bytes within the timeframe.
-		extraRate = float64(p.freeSpaceThreshold-info.freeBytes) / p.freeSpaceTimeframe.Seconds()
-	}
-	if info.liveBytes == 0 {
-		// We don't know the obsolete bytes ratio. Disable pacing altogether.
-		return 0.0
-	}
-	obsoleteBytesRatio := float64(info.obsoleteBytes) / float64(info.liveBytes)
-	if obsoleteBytesRatio >= p.obsoleteBytesMaxRatio {
-		// Increase the rate so that we can free up enough bytes within the timeframe.
-		r := (obsoleteBytesRatio - p.obsoleteBytesMaxRatio) * float64(info.liveBytes) / p.obsoleteBytesTimeframe.Seconds()
-		if extraRate < r {
-			extraRate = r
-		}
-	}
-
-	return float64(bytesToDelete) / (baseRate + extraRate)
-}
-
-// history is a helper used to keep track of the recent history of a set of
-// data points (in our case deleted bytes), at limited granularity.
-// Specifically, we split the desired timeframe into 100 "epochs" and all times
-// are effectively rounded down to the nearest epoch boundary.
-type history struct {
-	epochDuration time.Duration
-	startTime     crtime.Mono
-	// currEpoch is the epoch of the most recent operation.
-	currEpoch int64
-	// val contains the recent epoch values.
-	// val[currEpoch % historyEpochs] is the current epoch.
-	// val[(currEpoch + 1) % historyEpochs] is the oldest epoch.
-	val [historyEpochs]int64
-	// sum is always equal to the sum of values in val.
-	sum int64
-}
-
-const historyEpochs = 100
-
-// Init the history helper to keep track of data over the given number of
-// seconds.
-func (h *history) Init(now crtime.Mono, timeframe time.Duration) {
-	*h = history{
-		epochDuration: timeframe / time.Duration(historyEpochs),
-		startTime:     now,
-		currEpoch:     0,
-		sum:           0,
-	}
-}
-
-// Add adds a value for the current time.
-func (h *history) Add(now crtime.Mono, val int64) {
-	h.advance(now)
-	h.val[h.currEpoch%historyEpochs] += val
-	h.sum += val
-}
-
-// Sum returns the sum of recent values. The result is approximate in that the
-// cut-off time is within 1% of the exact one.
-func (h *history) Sum(now crtime.Mono) int64 {
-	h.advance(now)
-	return h.sum
-}
-
-func (h *history) epoch(t crtime.Mono) int64 {
-	return int64(t.Sub(h.startTime) / h.epochDuration)
-}
-
-// advance advances the time to the given time.
-func (h *history) advance(now crtime.Mono) {
-	epoch := h.epoch(now)
-	for h.currEpoch < epoch {
-		h.currEpoch++
-		// Forget the data for the oldest epoch.
-		h.sum -= h.val[h.currEpoch%historyEpochs]
-		h.val[h.currEpoch%historyEpochs] = 0
-	}
-}
diff --git a/snapshot.go b/snapshot.go
index 4dab187e..01d7638e 100644
--- a/snapshot.go
+++ b/snapshot.go
@@ -377,6 +377,15 @@ func (es *EventuallyFileOnlySnapshot) hasTransitioned() bool {
 	return es.mu.vers != nil
 }
 
+func (es *EventuallyFileOnlySnapshot) isClosed() bool {
+	select {
+	case <-es.closed:
+		return true
+	default:
+		return false
+	}
+}
+
 // waitForFlush waits for a flush on any memtables that need to be flushed
 // before this EFOS can transition to a file-only snapshot. If this EFOS is
 // waiting on a flush of the mutable memtable, it forces a rotation within
diff --git a/valsep/value_separation.go b/valsep/value_separation.go
new file mode 100644
index 00000000..b62a3ea8
--- /dev/null
+++ b/valsep/value_separation.go
@@ -0,0 +1,111 @@
+// Copyright 2025 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package valsep
+
+import (
+	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/manifest"
+	"github.com/cockroachdb/pebble/objstorage"
+	"github.com/cockroachdb/pebble/sstable"
+	"github.com/cockroachdb/pebble/sstable/blob"
+)
+
+// ValueSeparationOutputConfig is used to configure value separation for an
+// individual compaction output.
+type ValueSeparationOutputConfig struct {
+	// MinimumSize is the minimum size of a value that will be separated into a
+	// blob file.
+	MinimumSize int
+}
+
+// ValueSeparation defines an interface for writing some values to separate blob
+// files.
+type ValueSeparation interface {
+	// SetNextOutputConfig is called when a compaction is starting a new output
+	// sstable. It can be used to configure value separation specifically for
+	// the next compaction output.
+	SetNextOutputConfig(config ValueSeparationOutputConfig)
+	// Kind returns the kind of value separation strategy being used.
+	Kind() sstable.ValueSeparationKind
+	// MinimumSize returns the minimum size a value must be in order to be
+	// separated into a blob file.
+	MinimumSize() int
+	// EstimatedFileSize returns an estimate of the disk space consumed by the
+	// current, pending blob file if it were closed now. If no blob file has
+	// been created, it returns 0.
+	EstimatedFileSize() uint64
+	// EstimatedReferenceSize returns an estimate of the disk space consumed by
+	// the current output sstable's blob references so far.
+	EstimatedReferenceSize() uint64
+	// Add adds the provided key-value pair to the provided sstable writer,
+	// possibly separating the value into a blob file.
+	Add(tw sstable.RawWriter, kv *base.InternalKV, forceObsolete bool, isLikelyMVCCGarbage bool) error
+	// FinishOutput is called when a compaction is finishing an output sstable.
+	// It returns the table's blob references, which will be added to the
+	// table's TableMetadata, and stats and metadata describing a newly
+	// constructed blob file if any.
+	FinishOutput() (ValueSeparationMetadata, error)
+}
+
+// NewBlobFileInfo describes a newly written blob file
+// created by value separation. This is returned by
+// ValueSeparation.FinishOutput.
+type NewBlobFileInfo struct {
+	FileStats    blob.FileWriterStats
+	FileObject   objstorage.ObjectMetadata
+	FileMetadata *manifest.PhysicalBlobFile
+}
+
+// ValueSeparationMetadata describes metadata about a table's blob references,
+// and optionally a newly constructed blob file.
+type ValueSeparationMetadata struct {
+	BlobReferences     manifest.BlobReferences
+	BlobReferenceSize  uint64
+	BlobReferenceDepth manifest.BlobReferenceDepth
+
+	// The below fields are only populated if a new blob file was created.
+	NewBlobFiles []NewBlobFileInfo
+}
+
+// NeverSeparateValues is a ValueSeparation implementation that never separates
+// values into external blob files. It is the default value if no
+// ValueSeparation implementation is explicitly provided.
+type NeverSeparateValues struct{}
+
+// Assert that NeverSeparateValues implements the ValueSeparation interface.
+var _ ValueSeparation = NeverSeparateValues{}
+
+// SetNextOutputConfig implements the ValueSeparation interface.
+func (NeverSeparateValues) SetNextOutputConfig(config ValueSeparationOutputConfig) {}
+
+// Kind implements the ValueSeparation interface.
+func (NeverSeparateValues) Kind() sstable.ValueSeparationKind {
+	return sstable.ValueSeparationNone
+}
+
+// MinimumSize implements the ValueSeparation interface.
+func (v NeverSeparateValues) MinimumSize() int { return 0 }
+
+// EstimatedFileSize implements the ValueSeparation interface.
+func (NeverSeparateValues) EstimatedFileSize() uint64 { return 0 }
+
+// EstimatedReferenceSize implements the ValueSeparation interface.
+func (NeverSeparateValues) EstimatedReferenceSize() uint64 { return 0 }
+
+// Add implements the ValueSeparation interface.
+func (NeverSeparateValues) Add(
+	tw sstable.RawWriter, kv *base.InternalKV, forceObsolete bool, _ bool,
+) error {
+	v, _, err := kv.Value(nil)
+	if err != nil {
+		return err
+	}
+	return tw.Add(kv.K, v, forceObsolete)
+}
+
+// FinishOutput implements the ValueSeparation interface.
+func (NeverSeparateValues) FinishOutput() (ValueSeparationMetadata, error) {
+	return ValueSeparationMetadata{}, nil
+}
diff --git a/version_set.go b/version_set.go
index daeaa58d..de28332a 100644
--- a/version_set.go
+++ b/version_set.go
@@ -14,6 +14,7 @@ import (
 	"github.com/cockroachdb/crlib/crtime"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/deletepacer"
 	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/manifest"
 	"github.com/cockroachdb/pebble/objstorage"
@@ -67,16 +68,17 @@ type versionSet struct {
 	curCompactionConcurrency atomic.Int32
 
 	// Not all metrics are kept here. See DB.Metrics().
+	// TODO(radu): replace with only the metrics that are actually maintained.
 	metrics Metrics
 
 	// A pointer to versionSet.addObsoleteLocked. Avoids allocating a new closure
 	// on the creation of every version.
 	obsoleteFn func(manifest.ObsoleteFiles)
 	// obsolete{Tables,Blobs,Manifests,Options} are sorted by file number ascending.
-	obsoleteTables    []obsoleteFile
-	obsoleteBlobs     []obsoleteFile
-	obsoleteManifests []obsoleteFile
-	obsoleteOptions   []obsoleteFile
+	obsoleteTables    []deletepacer.ObsoleteFile
+	obsoleteBlobs     []deletepacer.ObsoleteFile
+	obsoleteManifests []deletepacer.ObsoleteFile
+	obsoleteOptions   []deletepacer.ObsoleteFile
 
 	// Zombie tables which have been removed from the current version but are
 	// still referenced by an inuse iterator.
@@ -591,13 +593,13 @@ func (vs *versionSet) UpdateVersionLocked(
 	}
 	if newManifestFileNum != 0 {
 		if vs.manifestFileNum != 0 {
-			vs.obsoleteManifests = append(vs.obsoleteManifests, obsoleteFile{
-				fileType: base.FileTypeManifest,
-				fs:       vs.fs,
-				path:     base.MakeFilepath(vs.fs, vs.dirname, base.FileTypeManifest, vs.manifestFileNum),
-				fileNum:  vs.manifestFileNum,
-				fileSize: prevManifestFileSize,
-				isLocal:  true,
+			vs.obsoleteManifests = append(vs.obsoleteManifests, deletepacer.ObsoleteFile{
+				FileType: base.FileTypeManifest,
+				FS:       vs.fs,
+				Path:     base.MakeFilepath(vs.fs, vs.dirname, base.FileTypeManifest, vs.manifestFileNum),
+				FileNum:  vs.manifestFileNum,
+				FileSize: prevManifestFileSize,
+				IsLocal:  true,
 			})
 		}
 		vs.manifestFileNum = newManifestFileNum
@@ -1019,14 +1021,14 @@ func (vs *versionSet) addObsoleteLocked(obsolete manifest.ObsoleteFiles) {
 	// Note that the zombie objects transition from zombie *to* obsolete, and
 	// will no longer be considered zombie.
 
-	newlyObsoleteTables := make([]obsoleteFile, len(obsolete.TableBackings))
+	newlyObsoleteTables := make([]deletepacer.ObsoleteFile, len(obsolete.TableBackings))
 	for i, bs := range obsolete.TableBackings {
 		newlyObsoleteTables[i] = vs.zombieTables.Extract(bs.DiskFileNum).
 			asObsoleteFile(vs.fs, base.FileTypeTable, vs.dirname)
 	}
 	vs.obsoleteTables = mergeObsoleteFiles(vs.obsoleteTables, newlyObsoleteTables)
 
-	newlyObsoleteBlobFiles := make([]obsoleteFile, len(obsolete.BlobFiles))
+	newlyObsoleteBlobFiles := make([]deletepacer.ObsoleteFile, len(obsolete.BlobFiles))
 	for i, bf := range obsolete.BlobFiles {
 		newlyObsoleteBlobFiles[i] = vs.zombieBlobs.Extract(bf.FileNum).
 			asObsoleteFile(vs.fs, base.FileTypeBlob, vs.dirname)
@@ -1044,19 +1046,17 @@ func (vs *versionSet) addObsolete(obsolete manifest.ObsoleteFiles) {
 }
 
 func (vs *versionSet) updateObsoleteObjectMetricsLocked() {
-	// TODO(jackson): Ideally we would update vs.fileDeletions.queuedStats to
-	// include the files on vs.obsolete{Tables,Blobs}, but there's subtlety in
-	// deduplicating the files before computing the stats. It might also be
-	// possible to refactor to remove the vs.obsolete{Tables,Blobs} intermediary
-	// step. Revisit this.
+	// TODO(jackson, radu): Investigate if we still need the
+	// vs.obsolete{Tables,Blobs} intermediary step or we can directly enqueue all
+	// deletions.
 	vs.metrics.Table.ObsoleteCount = int64(len(vs.obsoleteTables))
 	vs.metrics.Table.ObsoleteSize = 0
 	vs.metrics.Table.Local.ObsoleteSize = 0
 	vs.metrics.Table.Local.ObsoleteCount = 0
 	for _, fi := range vs.obsoleteTables {
-		vs.metrics.Table.ObsoleteSize += fi.fileSize
-		if fi.isLocal {
-			vs.metrics.Table.Local.ObsoleteSize += fi.fileSize
+		vs.metrics.Table.ObsoleteSize += fi.FileSize
+		if fi.IsLocal {
+			vs.metrics.Table.Local.ObsoleteSize += fi.FileSize
 			vs.metrics.Table.Local.ObsoleteCount++
 		}
 	}
@@ -1065,9 +1065,9 @@ func (vs *versionSet) updateObsoleteObjectMetricsLocked() {
 	vs.metrics.BlobFiles.Local.ObsoleteSize = 0
 	vs.metrics.BlobFiles.Local.ObsoleteCount = 0
 	for _, fi := range vs.obsoleteBlobs {
-		vs.metrics.BlobFiles.ObsoleteSize += fi.fileSize
-		if fi.isLocal {
-			vs.metrics.BlobFiles.Local.ObsoleteSize += fi.fileSize
+		vs.metrics.BlobFiles.ObsoleteSize += fi.FileSize
+		if fi.IsLocal {
+			vs.metrics.BlobFiles.Local.ObsoleteSize += fi.FileSize
 			vs.metrics.BlobFiles.Local.ObsoleteCount++
 		}
 	}
@@ -1082,8 +1082,8 @@ func setBasicLevelMetrics(m *Metrics, newVersion *manifest.Version) {
 		l := &m.Levels[i]
 		l.TablesCount = int64(newVersion.Levels[i].Len())
 		l.TablesSize = int64(newVersion.Levels[i].TableSize())
-		l.VirtualTablesCount = newVersion.Levels[i].NumVirtual
-		l.VirtualTablesSize = newVersion.Levels[i].VirtualTableSize
+		l.VirtualTablesCount = newVersion.Levels[i].VirtualTables().Count
+		l.VirtualTablesSize = newVersion.Levels[i].VirtualTables().Bytes
 		l.EstimatedReferencesSize = newVersion.Levels[i].EstimatedReferenceSize()
 		l.Sublevels = 0
 		if l.TablesCount > 0 {
diff --git a/vfs/logging_fs.go b/vfs/logging_fs.go
index b341fdb2..f85e5bc9 100644
--- a/vfs/logging_fs.go
+++ b/vfs/logging_fs.go
@@ -118,6 +118,10 @@ func (fs loggingFS) RemoveAll(name string) error {
 	return err
 }
 
+func (fs loggingFS) Unwrap() FS {
+	return fs.FS
+}
+
 type loggingFile struct {
 	File
 	name  string
diff --git a/wal/failover_manager.go b/wal/failover_manager.go
index f0df5288..36c1eb97 100644
--- a/wal/failover_manager.go
+++ b/wal/failover_manager.go
@@ -746,12 +746,6 @@ func (wm *failoverManager) Close() error {
 	for _, f := range wm.dirHandles {
 		err = firstError(err, f.Close())
 	}
-	if wm.opts.Primary.Lock != nil {
-		err = firstError(err, wm.opts.Primary.Lock.Close())
-	}
-	if wm.opts.Secondary.Lock != nil {
-		err = firstError(err, wm.opts.Secondary.Lock.Close())
-	}
 	return err
 }
 
diff --git a/wal/standalone_manager.go b/wal/standalone_manager.go
index bceea009..7396550d 100644
--- a/wal/standalone_manager.go
+++ b/wal/standalone_manager.go
@@ -251,9 +251,6 @@ func (m *StandaloneManager) Close() error {
 		_, err = m.w.Close()
 	}
 	err = firstError(err, m.walDir.Close())
-	if m.o.Primary.Lock != nil {
-		err = firstError(err, m.o.Primary.Lock.Close())
-	}
 	return err
 }
 



Test patch:
diff --git a/blob_rewrite_test.go b/blob_rewrite_test.go
index 823c272d..3e3b455d 100644
--- a/blob_rewrite_test.go
+++ b/blob_rewrite_test.go
@@ -21,7 +21,6 @@ import (
 	"github.com/cockroachdb/pebble/internal/blobtest"
 	"github.com/cockroachdb/pebble/internal/bytealloc"
 	"github.com/cockroachdb/pebble/internal/cache"
-	"github.com/cockroachdb/pebble/internal/compact"
 	"github.com/cockroachdb/pebble/internal/manifest"
 	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/internal/testutils"
@@ -30,6 +29,7 @@ import (
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/sstable/blob"
 	"github.com/cockroachdb/pebble/sstable/block"
+	"github.com/cockroachdb/pebble/valsep"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/stretchr/testify/require"
 )
@@ -37,7 +37,7 @@ import (
 func TestBlobRewrite(t *testing.T) {
 	var (
 		bv  blobtest.Values
-		vs  compact.ValueSeparation
+		vs  valsep.ValueSeparation
 		tw  sstable.RawWriter
 		fn  base.DiskFileNum
 		buf bytes.Buffer
@@ -131,11 +131,13 @@ func TestBlobRewrite(t *testing.T) {
 
 				meta, err := vs.FinishOutput()
 				require.NoError(t, err)
-				if meta.BlobFileObject.DiskFileNum == 0 {
+				if len(meta.NewBlobFiles) == 0 {
 					fmt.Fprintln(&buf, "no blob file created")
 				} else {
-					fmt.Fprintf(&buf, "Blob file created: %s\n", meta.BlobFileMetadata)
-					fmt.Fprintln(&buf, meta.BlobFileStats)
+					for _, bf := range meta.NewBlobFiles {
+						fmt.Fprintf(&buf, "Blob file created: %s\n", bf.FileMetadata)
+						fmt.Fprintln(&buf, bf.FileStats)
+					}
 				}
 				if len(meta.BlobReferences) == 0 {
 					fmt.Fprintln(&buf, "blobrefs:[]")
diff --git a/compaction_test.go b/compaction_test.go
index e62a2757..fb73b22e 100644
--- a/compaction_test.go
+++ b/compaction_test.go
@@ -3099,7 +3099,7 @@ func TestCompaction_UpdateVersionFails(t *testing.T) {
 }
 
 // TestSharedObjectDeletePacing tests that we don't throttle shared object
-// deletes (see the TargetBytesDeletionRate option).
+// deletes (see the DeletePacer.BaselineRate option).
 func TestSharedObjectDeletePacing(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 
@@ -3109,7 +3109,7 @@ func TestSharedObjectDeletePacing(t *testing.T) {
 		"": remote.NewInMem(),
 	})
 	opts.Experimental.CreateOnShared = remote.CreateOnSharedAll
-	opts.TargetByteDeletionRate = func() int { return 1 }
+	opts.DeletionPacing.BaselineRate = func() uint64 { return 1 }
 	opts.Logger = testutils.Logger{T: t}
 
 	d, err := Open("", &opts)
diff --git a/value_separation_test.go b/compaction_value_separation_test.go
similarity index 93%
rename from value_separation_test.go
rename to compaction_value_separation_test.go
index e6bf4d98..f6ef2aa0 100644
--- a/value_separation_test.go
+++ b/compaction_value_separation_test.go
@@ -18,13 +18,13 @@ import (
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/blobtest"
-	"github.com/cockroachdb/pebble/internal/compact"
 	"github.com/cockroachdb/pebble/internal/manifest"
 	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/objstorage"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/sstable/blob"
+	"github.com/cockroachdb/pebble/valsep"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/stretchr/testify/require"
 )
@@ -32,7 +32,7 @@ import (
 func TestValueSeparationPolicy(t *testing.T) {
 	var (
 		bv  blobtest.Values
-		vs  compact.ValueSeparation
+		vs  valsep.ValueSeparation
 		tw  sstable.RawWriter
 		fn  base.DiskFileNum
 		buf bytes.Buffer
@@ -76,7 +76,7 @@ func TestValueSeparationPolicy(t *testing.T) {
 				bv = blobtest.Values{}
 				switch x := d.CmdArgs[0].String(); x {
 				case "never-separate-values":
-					vs = compact.NeverSeparateValues{}
+					vs = valsep.NeverSeparateValues{}
 				case "preserve-blob-references":
 					pbr := &preserveBlobReferences{}
 					lines := crstrings.Lines(d.Input)
@@ -143,11 +143,13 @@ func TestValueSeparationPolicy(t *testing.T) {
 
 				meta, err := vs.FinishOutput()
 				require.NoError(t, err)
-				if meta.BlobFileObject.DiskFileNum == 0 {
+				if len(meta.NewBlobFiles) == 0 {
 					fmt.Fprintln(&buf, "no blob file created")
 				} else {
-					fmt.Fprintf(&buf, "Blob file created: %s\n", meta.BlobFileMetadata)
-					fmt.Fprintln(&buf, meta.BlobFileStats)
+					for _, bf := range meta.NewBlobFiles {
+						fmt.Fprintf(&buf, "Blob file created: %s\n", bf.FileMetadata)
+						fmt.Fprintln(&buf, bf.FileStats)
+					}
 				}
 				if len(meta.BlobReferences) == 0 {
 					fmt.Fprintln(&buf, "blobrefs:[]")
@@ -209,10 +211,10 @@ type defineDBValueSeparator struct {
 }
 
 // Assert that *defineDBValueSeparator implements the compact.ValueSeparation interface.
-var _ compact.ValueSeparation = (*defineDBValueSeparator)(nil)
+var _ valsep.ValueSeparation = (*defineDBValueSeparator)(nil)
 
 // SetNextOutputConfig implements the compact.ValueSeparation interface.
-func (vs *defineDBValueSeparator) SetNextOutputConfig(config compact.ValueSeparationOutputConfig) {}
+func (vs *defineDBValueSeparator) SetNextOutputConfig(config valsep.ValueSeparationOutputConfig) {}
 
 // Kind implements the ValueSeparation interface.
 func (vs *defineDBValueSeparator) Kind() sstable.ValueSeparationKind {
@@ -277,10 +279,10 @@ func (vs *defineDBValueSeparator) Add(
 }
 
 // FinishOutput implements compact.ValueSeparation.
-func (d *defineDBValueSeparator) FinishOutput() (compact.ValueSeparationMetadata, error) {
+func (d *defineDBValueSeparator) FinishOutput() (valsep.ValueSeparationMetadata, error) {
 	m, err := d.pbr.FinishOutput()
 	if err != nil {
-		return compact.ValueSeparationMetadata{}, err
+		return valsep.ValueSeparationMetadata{}, err
 	}
 	// TODO(jackson): Support setting a specific depth from the datadriven test.
 	m.BlobReferenceDepth = manifest.BlobReferenceDepth(len(m.BlobReferences))
diff --git a/data_test.go b/data_test.go
index 0f6a8456..fc8db472 100644
--- a/data_test.go
+++ b/data_test.go
@@ -26,7 +26,6 @@ import (
 	"github.com/cockroachdb/pebble/bloom"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/blobtest"
-	"github.com/cockroachdb/pebble/internal/compact"
 	"github.com/cockroachdb/pebble/internal/humanize"
 	"github.com/cockroachdb/pebble/internal/keyspan"
 	"github.com/cockroachdb/pebble/internal/manifest"
@@ -39,6 +38,7 @@ import (
 	"github.com/cockroachdb/pebble/objstorage/remote"
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/sstable/block/blockkind"
+	"github.com/cockroachdb/pebble/valsep"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/cockroachdb/pebble/vfs/errorfs"
 	"github.com/cockroachdb/pebble/wal"
@@ -939,7 +939,7 @@ func runDBDefineCmd(td *datadriven.TestData, opts *Options) (*DB, error) {
 		if err != nil {
 			return err
 		}
-		c.getValueSeparation = func(JobID, *tableCompaction, ValueStoragePolicy) compact.ValueSeparation {
+		c.getValueSeparation = func(JobID, *tableCompaction, ValueStoragePolicy) valsep.ValueSeparation {
 			return valueSeparator
 		}
 		// NB: define allows the test to exactly specify which keys go
diff --git a/flushable_test.go b/flushable_test.go
index abfc9105..672a05be 100644
--- a/flushable_test.go
+++ b/flushable_test.go
@@ -100,7 +100,7 @@ func TestIngestedSSTFlushableAPI(t *testing.T) {
 		// point before we update the MANIFEST (via UpdateVersionLocked), otherwise
 		// a crash can have the tables referenced in the MANIFEST, but not present
 		// in the directory.
-		if err := d.dataDir.Sync(); err != nil {
+		if err := d.dirs.DataDir.Sync(); err != nil {
 			t.Fatal(err)
 		}
 
diff --git a/internal/deletepacer/delete_pacer_test.go b/internal/deletepacer/delete_pacer_test.go
new file mode 100644
index 00000000..4a1d9686
--- /dev/null
+++ b/internal/deletepacer/delete_pacer_test.go
@@ -0,0 +1,276 @@
+// Copyright 2025 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+//go:build go1.25
+
+package deletepacer
+
+import (
+	"fmt"
+	"slices"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"testing/synctest"
+	"time"
+
+	"github.com/cockroachdb/crlib/crhumanize"
+	"github.com/cockroachdb/crlib/crstrings"
+	"github.com/cockroachdb/datadriven"
+	"github.com/cockroachdb/datadriven/diagram"
+	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/testutils"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDataDriven(t *testing.T) {
+	for _, path := range []string{"backlog", "basic", "free-space"} {
+		t.Run(path, func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				datadriven.RunTest(t, "testdata/"+path, func(t *testing.T, td *datadriven.TestData) string {
+					switch td.Cmd {
+					case "run":
+						ts := &testState{
+							startTime:  time.Now(),
+							deleteSema: make(chan struct{}, 1),
+						}
+						ts.deleteSema <- struct{}{}
+						ts.baselineRate.Store(128 * MB)
+						ts.diskFreeSpace.Store(100 * GB)
+						opts := Options{
+							FreeSpaceTimeframe: 10 * time.Second,
+							BacklogTimeframe:   50 * time.Minute,
+							BaselineRate:       func() uint64 { return ts.baselineRate.Load() },
+						}
+						if arg, ok := td.Arg("free-space-threshold"); ok {
+							threshold, err := crhumanize.ParseBytes[uint64](arg.Vals[0])
+							require.NoErrorf(t, err, "%s: free-space-threshold %q", td.Pos, arg)
+							opts.FreeSpaceThresholdBytes = threshold
+						}
+
+						diskFreeSpaceFn := func() uint64 { return ts.diskFreeSpace.Load() }
+
+						deleteFn := func(of ObsoleteFile, jobID int) {
+							<-ts.deleteSema
+							ts.deleteSema <- struct{}{}
+							ts.dels = append(ts.dels, logEntry{
+								timestamp: time.Now(),
+								fileSize:  of.FileSize,
+							})
+						}
+
+						ts.dp = Open(opts, testutils.Logger{T: t}, diskFreeSpaceFn, deleteFn)
+						defer ts.dp.Close()
+						executeTest(t, td, ts)
+						return fmt.Sprintf("Enqueue:\n%sDelete:\n%s",
+							plot(ts.startTime, ts.adds, "•"),
+							plot(ts.startTime, ts.dels, "•"),
+						)
+
+					default:
+						td.Fatalf(t, "unknown command %q", td.Cmd)
+						return ""
+					}
+				})
+			})
+		})
+	}
+}
+
+type testState struct {
+	dp            *DeletePacer
+	baselineRate  atomic.Uint64
+	diskFreeSpace atomic.Uint64
+	adds, dels    []logEntry
+	startTime     time.Time
+	deleteSema    chan struct{}
+}
+
+func executeTest(t *testing.T, td *datadriven.TestData, ts *testState) {
+	repeat := 1
+	td.MaybeScanArgs(t, "repeat", &repeat)
+	for _, l := range crstrings.Lines(strings.Repeat(td.Input+"\n", repeat)) {
+		if l == "" {
+			continue
+		}
+		words := strings.Split(l, " ")
+		switch words[0] {
+		case "del":
+			fileSize, err := crhumanize.ParseBytes[uint64](words[1])
+			require.NoErrorf(t, err, "%s: %q", td.Pos, l)
+			ts.dp.Enqueue(0, ObsoleteFile{
+				FileType: base.FileTypeTable,
+				FileSize: fileSize,
+				IsLocal:  true,
+			})
+			ts.adds = append(ts.adds, logEntry{
+				timestamp: time.Now(),
+				fileSize:  fileSize,
+			})
+
+		case "sleep":
+			d, err := time.ParseDuration(words[1])
+			require.NoErrorf(t, err, "%s: %q", td.Pos, l)
+			time.Sleep(d)
+
+		case "block-deletes":
+			<-ts.deleteSema
+
+		case "unblock-deletes":
+			ts.deleteSema <- struct{}{}
+
+		case "baseline-rate":
+			rate, err := crhumanize.ParseBytesPerSec[uint64](words[1])
+			require.NoErrorf(t, err, "%s: %q", td.Pos, l)
+			ts.baselineRate.Store(rate)
+
+		case "free-space":
+			freeSpace, err := crhumanize.ParseBytes[uint64](words[1])
+			require.NoErrorf(t, err, "%s: %q", td.Pos, l)
+			ts.diskFreeSpace.Store(freeSpace)
+		}
+		synctest.Wait()
+	}
+}
+
+type logEntry struct {
+	timestamp time.Time
+	fileSize  uint64
+}
+
+func plot(startTime time.Time, log []logEntry, point string) string {
+	const width = 100
+	const height = 8
+	now := time.Now()
+	deltaT := now.Sub(startTime)
+	tStep := 1 + deltaT/width
+	var vals [width]uint64
+	for _, l := range log {
+		vals[l.timestamp.Sub(startTime)/tStep] += l.fileSize
+	}
+	maxVal := slices.Max(vals[:])
+
+	var wb diagram.Whiteboard
+	for i := range width {
+		wb.Write(height+1, i, "-")
+	}
+	for i := range height + 1 {
+		wb.Write(i, -1, "|")
+	}
+	wb.Write(height+1, -1, "+")
+	for i := range width {
+		if vals[i] != 0 {
+			// The maxVal/2 term is to round instead of flooring.
+			y := int((vals[i]*uint64(height) + maxVal/2) / maxVal)
+			wb.Write(height-y, i, point)
+		}
+	}
+	maxValStr := string(crhumanize.Bytes(maxVal, crhumanize.Compact))
+	wb.Write(0, -len(maxValStr)-2, maxValStr)
+	wb.Write(height, -4, "0B")
+	timeStr := deltaT.String()
+	wb.Write(height+2, 0, "0s")
+	wb.Write(height+2, width-len(timeStr), timeStr)
+	return wb.String()
+}
+
+// TestCloseWithPacing verifies that DeletePacer.Close() disables pacing so it
+// completes in a timely manner.
+func TestCloseWithPacing(t *testing.T) {
+	opts := Options{
+		FreeSpaceThresholdBytes: 1,
+		BaselineRate:            func() uint64 { return 1024 }, // 1 KB/s - slow pacing
+	}
+	opts.EnsureDefaults()
+
+	diskFreeSpaceFn := func() uint64 { return 10 * GB }
+	deleteFn := func(of ObsoleteFile, jobID int) {}
+
+	dp := Open(opts, testutils.Logger{T: t}, diskFreeSpaceFn, deleteFn)
+
+	// Create obsolete files that would normally take a long time to delete.
+	// At 1 KB/s, 100 files of 10 KB each would take 1000 seconds.
+	largeFiles := make([]ObsoleteFile, 100)
+	for i := range largeFiles {
+		largeFiles[i] = ObsoleteFile{
+			FileType: base.FileTypeTable,
+			FileSize: 10 * 1024,
+			IsLocal:  true,
+		}
+	}
+
+	dp.Enqueue(1, largeFiles...)
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		dp.Close()
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(30 * time.Second):
+		t.Fatalf("timed out waiting for cleanupManager.Close() to return")
+	}
+}
+
+// TestFallingBehind verifies that we disable pacing when the queue gets too big.
+func TestFallingBehind(t *testing.T) {
+	var rate atomic.Uint64
+	rate.Store(10 * MB) // 10MB/s
+
+	opts := Options{
+		BaselineRate: func() uint64 { return rate.Load() }, // 10 MB/s
+	}
+	opts.EnsureDefaults()
+
+	diskFreeSpaceFn := func() uint64 { return 100 * GB }
+	deleteFn := func(of ObsoleteFile, jobID int) {}
+
+	dp := Open(opts, testutils.Logger{T: t}, diskFreeSpaceFn, deleteFn)
+	defer dp.Close()
+
+	x := 0
+	addJob := func(fileSize int) {
+		x++
+		dp.Enqueue(1, ObsoleteFile{
+			FileType: base.FileTypeTable,
+			FileSize: uint64(fileSize),
+			IsLocal:  true,
+		})
+	}
+
+	const initial = maxQueueSize / 2
+	for range initial {
+		addJob(1 * MB)
+	}
+	queueSize := func() int {
+		dp.mu.Lock()
+		defer dp.mu.Unlock()
+		return dp.mu.queue.Len()
+	}
+	// 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, queueSize(), initial/2)
+
+	// Add enough jobs to exceed the threshold. We add small jobs so that the
+	// historic rate doesn't grow significantly.
+	for range maxQueueSize {
+		addJob(1)
+	}
+
+	const sleepTime = 10 * time.Millisecond
+	const timeout = 10 * time.Second
+	for i := 0; ; i++ {
+		time.Sleep(10 * time.Millisecond)
+		if queueSize() <= maxQueueSize {
+			break
+		}
+		if i == int(timeout/sleepTime) {
+			t.Fatalf("jobs channel length never dropped below threshold (%d vs %d)", queueSize(), maxQueueSize)
+		}
+	}
+}
diff --git a/internal/deletepacer/rate_calc_test.go b/internal/deletepacer/rate_calc_test.go
new file mode 100644
index 00000000..f6bc2085
--- /dev/null
+++ b/internal/deletepacer/rate_calc_test.go
@@ -0,0 +1,190 @@
+// Copyright 2025 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package deletepacer
+
+import (
+	"fmt"
+	"math"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/cockroachdb/crlib/crhumanize"
+	"github.com/cockroachdb/crlib/crtime"
+	"github.com/cockroachdb/datadriven"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRateCalculatorDataDriven(t *testing.T) {
+	datadriven.RunTest(t, "testdata/rate-calc", func(t *testing.T, td *datadriven.TestData) string {
+		switch td.Cmd {
+		case "simulate":
+			return runSimulation(t, td)
+		default:
+			td.Fatalf(t, "unknown command %q", td.Cmd)
+			return ""
+		}
+	})
+}
+
+const MB = 1 << 20
+const GB = 1 << 30
+
+func runSimulation(t *testing.T, td *datadriven.TestData) string {
+	// Parse configuration from arguments
+	type testConfig struct {
+		baselineRate            uint64
+		freeSpaceThresholdBytes uint64
+		freeSpaceTimeframe      time.Duration
+		backlogTimeframe        time.Duration
+		diskFreeSpace           uint64
+	}
+	cfg := &testConfig{
+		baselineRate:            128 * MB,
+		freeSpaceThresholdBytes: 10 * GB,
+		freeSpaceTimeframe:      10 * time.Second,
+		backlogTimeframe:        5 * time.Minute,
+		diskFreeSpace:           100 * GB,
+	}
+
+	for _, arg := range td.CmdArgs {
+		switch arg.Key {
+		case "baseline-rate":
+			rate, err := crhumanize.ParseBytesPerSec[uint64](arg.Vals[0])
+			require.NoError(t, err)
+			cfg.baselineRate = rate
+		case "free-space-threshold":
+			threshold, err := crhumanize.ParseBytes[uint64](arg.Vals[0])
+			require.NoError(t, err)
+			cfg.freeSpaceThresholdBytes = threshold
+		case "free-space-timeframe":
+			timeframe, err := time.ParseDuration(arg.Vals[0])
+			require.NoError(t, err)
+			cfg.freeSpaceTimeframe = timeframe
+		case "backlog-timeframe":
+			timeframe, err := time.ParseDuration(arg.Vals[0])
+			require.NoError(t, err)
+			cfg.backlogTimeframe = timeframe
+		case "disk-free-space":
+			freeSpace, err := crhumanize.ParseBytes[uint64](arg.Vals[0])
+			require.NoError(t, err)
+			cfg.diskFreeSpace = freeSpace
+		default:
+			td.Fatalf(t, "unknown argument %q", arg.Key)
+		}
+	}
+
+	pacerCfg := &Options{
+		FreeSpaceThresholdBytes: cfg.freeSpaceThresholdBytes,
+		FreeSpaceTimeframe:      cfg.freeSpaceTimeframe,
+		BacklogTimeframe:        cfg.backlogTimeframe,
+		BaselineRate:            func() uint64 { return cfg.baselineRate },
+	}
+
+	diskFreeSpaceFn := func() uint64 { return cfg.diskFreeSpace }
+	rc := makeRateCalculator(pacerCfg, diskFreeSpaceFn, crtime.Mono(0))
+
+	var output strings.Builder
+
+	var currentTime crtime.Mono
+	for _, line := range strings.Split(strings.TrimSpace(td.Input), "\n") {
+		if line == "" || strings.HasPrefix(line, "#") {
+			continue
+		}
+
+		parts := strings.Fields(line)
+		switch parts[0] {
+		case "set-baseline-rate":
+			if len(parts) < 2 {
+				td.Fatalf(t, "invalid set-baseline-rate line: %q (expected: set-baseline-rate rate)", line)
+			}
+			rate, err := crhumanize.ParseBytesPerSec[uint64](parts[1])
+			require.NoError(t, err, "failed to parse baseline rate %q", parts[1])
+			cfg.baselineRate = rate
+			continue
+
+		case "set-free-space":
+			if len(parts) < 2 {
+				td.Fatalf(t, "invalid set-free-space line: %q (expected: set-free-space bytes)", line)
+			}
+			freeSpace, err := crhumanize.ParseBytes[uint64](parts[1])
+			require.NoError(t, err, "failed to parse free space %q", parts[1])
+			cfg.diskFreeSpace = freeSpace
+			continue
+
+		case "add-debt":
+			if len(parts) < 2 {
+				td.Fatalf(t, "invalid add-debt line: %q (expected: add-debt bytes)", line)
+			}
+			debtBytes, err := crhumanize.ParseBytes[uint64](parts[1])
+			require.NoError(t, err, "failed to parse debt bytes %q", parts[1])
+			rc.AddDebt(debtBytes)
+			continue
+		}
+
+		// Parse format: "15s recent=1MiB queued=10MiB [disable-pacing]"
+		duration, err := time.ParseDuration(parts[0])
+		require.NoError(t, err, "failed to parse duration %q", parts[0])
+
+		// Parse key=value pairs for recent and queued
+		var recentBytes, queuedBytes uint64
+		disablePacing := false
+
+		for i := 1; i < len(parts); i++ {
+			part := parts[i]
+			// Handle disable-pacing flag
+			if part == "disable-pacing" {
+				disablePacing = true
+				continue
+			}
+
+			// Parse key=value
+			kv := strings.SplitN(part, "=", 2)
+			if len(kv) != 2 {
+				td.Fatalf(t, "invalid key=value pair %q in line %q", part, line)
+			}
+
+			switch kv[0] {
+			case "recent":
+				recentBytes, err = crhumanize.ParseBytes[uint64](kv[1])
+				require.NoError(t, err, "failed to parse recent bytes %q", kv[1])
+			case "queued":
+				queuedBytes, err = crhumanize.ParseBytes[uint64](kv[1])
+				require.NoError(t, err, "failed to parse queued bytes %q", kv[1])
+			default:
+				td.Fatalf(t, "unknown key %q in line %q", kv[0], line)
+			}
+		}
+
+		currentTime = crtime.Mono(duration)
+		rc.Update(currentTime, recentBytes, queuedBytes, disablePacing)
+
+		// Report current rate, debt and wait time.
+		currentRate := int(math.Round(rc.currentRate))
+		backlogRate := int(math.Round(rc.backlogRate))
+		freeSpaceRate := int(math.Round(rc.freeSpaceRate))
+
+		rateStr := string(crhumanize.BytesPerSec(currentRate, crhumanize.Compact))
+		if backlogRate > 0 || freeSpaceRate > 0 {
+			var extraStrs []string
+			if backlogRate > 0 {
+				extraStrs = append(extraStrs, "backlog="+string(crhumanize.BytesPerSec(backlogRate, crhumanize.Compact)))
+			}
+			if freeSpaceRate > 0 {
+				extraStrs = append(extraStrs, "free-space="+string(crhumanize.BytesPerSec(freeSpaceRate, crhumanize.Compact)))
+			}
+			rateStr += " (" + strings.Join(extraStrs, " ") + ")"
+		}
+
+		if rc.debtBytes == 0 {
+			fmt.Fprintf(&output, "%s: rate=%s debt=0B\n", time.Duration(currentTime), rateStr)
+		} else {
+			waitTime := rc.DebtWaitTime().Round(time.Second)
+			fmt.Fprintf(&output, "%s: rate=%s debt=%s (wait %v)\n", time.Duration(currentTime), rateStr, crhumanize.Bytes(uint64(rc.debtBytes), crhumanize.Compact), waitTime)
+		}
+	}
+
+	return output.String()
+}
diff --git a/internal/deletepacer/testdata/backlog b/internal/deletepacer/testdata/backlog
new file mode 100644
index 00000000..6769ecee
--- /dev/null
+++ b/internal/deletepacer/testdata/backlog
@@ -0,0 +1,103 @@
+# This test artificially puts the pacer in a condition where there are not many
+# recent deletions, but there is a large backlog. Once we unblock deletes (about
+# halfway through the test), we should see the backlog worked down within 5
+# minutes.
+run
+baseline-rate 1MiB/s
+block-deletes
+del 1MiB
+sleep 10s
+del 10GiB
+sleep 10s
+del 10GiB
+sleep 10s
+del 10GiB
+sleep 10s
+del 10GiB
+sleep 10s
+del 10GiB
+sleep 10s
+del 10GiB
+sleep 5m
+unblock-deletes
+sleep 4m
+----
+Enqueue:
+10GiB | • •• • ••
+      |
+      |
+      |
+      |
+      |
+      |
+      |
+   0B |•
+      +----------------------------------------------------------------------------------------------------
+       0s                                                                                             10m0s
+Delete:
+10GiB |                                                           •       •       •       •       •       •
+      |
+      |
+      |
+      |
+      |
+      |
+      |
+   0B |
+      +----------------------------------------------------------------------------------------------------
+       0s                                                                                             10m0s
+
+
+# Repeat the same test but add a free-space condition that temporarily increases
+# the rate during backlog drain. We want to see more frequent deletions during
+# that period, and then a return to the previous frequency.
+run free-space-threshold=100GiB
+baseline-rate 1MiB/s
+block-deletes
+del 1MiB
+sleep 10s
+del 10GiB
+sleep 10s
+del 10GiB
+sleep 10s
+del 10GiB
+sleep 10s
+del 10GiB
+sleep 10s
+del 10GiB
+sleep 10s
+del 10GiB
+sleep 5m
+unblock-deletes
+sleep 30s
+free-space 99GiB
+del 1GiB
+sleep 30s
+free-space 100GiB
+del 1GiB
+sleep 3m
+----
+Enqueue:
+10GiB | • •• • ••
+      |
+      |
+      |
+      |
+      |
+      |
+      |                                                                •    •
+   0B |•
+      +----------------------------------------------------------------------------------------------------
+       0s                                                                                             10m0s
+Delete:
+10GiB |                                                           •     • ••  •     •
+      |
+      |
+      |
+      |
+      |
+      |
+      |
+   0B |
+      +----------------------------------------------------------------------------------------------------
+       0s                                                                                             10m0s
diff --git a/internal/deletepacer/testdata/basic b/internal/deletepacer/testdata/basic
new file mode 100644
index 00000000..20e3cd40
--- /dev/null
+++ b/internal/deletepacer/testdata/basic
@@ -0,0 +1,74 @@
+# Simulate a 4 file compaction every 30 seconds.
+run repeat=20
+baseline-rate 1MiB/s
+del 256MB
+del 256MB
+del 256MB
+del 256MB
+sleep 30s
+----
+Enqueue:
+977MiB |•   •    •    •    •    •    •    •    •    •    •    •    •    •    •    •    •    •    •    •
+       |
+       |
+       |
+       |
+       |
+       |
+       |
+    0B |
+       +----------------------------------------------------------------------------------------------------
+        0s                                                                                             10m0s
+Delete:
+244MiB |•       •    •  •  •  • •  • • •• • •• •• ••• •••• •••• •••• •••• •••• •••• •••• •••• •••• •••• •••
+       |
+       |
+       |
+       |
+       |
+       |
+       |
+    0B |
+       +----------------------------------------------------------------------------------------------------
+        0s                                                                                             10m0s
+
+run repeat=20
+baseline-rate 1MiB/s
+del 256MB
+del 256MB
+del 256MB
+del 256MB
+sleep 15s
+del 128MB
+del 128MB
+del 128MB
+del 128MB
+del 128MB
+del 128MB
+del 128MB
+sleep 15s
+----
+Enqueue:
+977MiB |•   •    •    •    •    •    •    •    •    •    •    •    •    •    •    •    •    •    •    •
+       |  •    •    •    •    •    •    •    •    •    •    •    •    •    •    •    •    •    •    •    •
+       |
+       |
+       |
+       |
+       |
+       |
+    0B |
+       +----------------------------------------------------------------------------------------------------
+        0s                                                                                             10m0s
+Delete:
+488MiB |                                       •    •    •    •    •    •    •   • •  • •  • •  • •  • •  •
+       |
+       |                                    •    • •  •••  •••  •••  •••  •••   •    •    •    •    •    •
+       |
+       |•     •   • •  •     ••• •• •••••••• ••   •  •    •    •    •    •    ••  • •  • •  • •  • •  • •  •
+       |
+       |                •••••      •            •
+       |
+    0B |
+       +----------------------------------------------------------------------------------------------------
+        0s                                                                                             10m0s
diff --git a/internal/deletepacer/testdata/free-space b/internal/deletepacer/testdata/free-space
new file mode 100644
index 00000000..73d19cc4
--- /dev/null
+++ b/internal/deletepacer/testdata/free-space
@@ -0,0 +1,64 @@
+run baseline-rate=1MiB/s free-space-threshold=10GiB
+free-space 20GiB
+del 256MB
+del 256MB
+del 256MB
+del 256MB
+sleep 30s
+del 256MB
+del 256MB
+del 256MB
+del 256MB
+sleep 30s
+del 256MB
+del 256MB
+del 256MB
+del 256MB
+sleep 30s
+free-space 5GiB
+del 256MB
+del 256MB
+del 256MB
+del 256MB
+sleep 30s
+del 256MB
+del 256MB
+del 256MB
+del 256MB
+sleep 30s
+del 256MB
+del 256MB
+del 256MB
+del 256MB
+sleep 30s
+free-space 10GiB
+del 256MB
+del 256MB
+del 256MB
+del 256MB
+sleep 30s
+----
+Enqueue:
+977MiB |•             •             •             •              •             •             •
+       |
+       |
+       |
+       |
+       |
+       |
+       |
+    0B |
+       +----------------------------------------------------------------------------------------------------
+        0s                                                                                             3m30s
+Delete:
+977MiB |                                                         •             •
+       |
+       |                                           •
+       |
+       |•
+       |
+       | ••           ••••          ••••          •                                          ••••
+       |
+    0B |
+       +----------------------------------------------------------------------------------------------------
+        0s                                                                                             3m30s
diff --git a/internal/deletepacer/testdata/rate-calc b/internal/deletepacer/testdata/rate-calc
new file mode 100644
index 00000000..69570e70
--- /dev/null
+++ b/internal/deletepacer/testdata/rate-calc
@@ -0,0 +1,124 @@
+simulate baseline-rate=1MiB/s
+0s recent=0B queued=0B
+1s recent=5MiB queued=5MiB
+2s recent=10MiB queued=10MiB
+3s recent=15MiB queued=15MiB
+add-debt 5MiB
+4s recent=10MiB queued=10MiB
+5s recent=10MiB queued=10MiB
+8s recent=10MiB queued=10MiB
+----
+0s: rate=1MiB/s debt=0B
+1s: rate=1MiB/s debt=0B
+2s: rate=1MiB/s debt=0B
+3s: rate=1MiB/s debt=0B
+4s: rate=1MiB/s debt=4MiB (wait 4s)
+5s: rate=1MiB/s debt=3MiB (wait 3s)
+8s: rate=1MiB/s debt=0B
+
+# Test backlog drain. Even as the backlog starts getting cleared, we should
+# maintain the 100MiB/s rate until it is fully cleared.
+simulate baseline-rate=10MiB/s backlog-timeframe=1m
+0s recent=10MiB queued=6010MiB
+add-debt 1000MiB
+4s recent=10MiB queued=5010MiB
+add-debt 1000MiB
+9s recent=10MiB queued=4010MiB
+add-debt 1000MiB
+15s recent=10MiB queued=3010MiB
+add-debt 3000MiB
+20s recent=10MiB queued=10MiB
+----
+0s: rate=110MiB/s (backlog=100MiB/s) debt=0B
+4s: rate=110MiB/s (backlog=100MiB/s) debt=560MiB (wait 5s)
+9s: rate=110MiB/s (backlog=100MiB/s) debt=474MiB (wait 4s)
+15s: rate=110MiB/s (backlog=100MiB/s) debt=364MiB (wait 3s)
+20s: rate=10MiB/s debt=474MiB (wait 47s)
+
+# If the backlog goes up, the backlog rate increases (and stays there).
+simulate baseline-rate=10MiB/s backlog-timeframe=1m
+0s recent=10MiB queued=6010MiB
+add-debt 1000MiB
+4s recent=10MiB queued=8010MiB
+add-debt 1000MiB
+9s recent=10MiB queued=2010MiB
+----
+0s: rate=110MiB/s (backlog=100MiB/s) debt=0B
+4s: rate=143MiB/s (backlog=133MiB/s) debt=560MiB (wait 4s)
+9s: rate=143MiB/s (backlog=133MiB/s) debt=307MiB (wait 2s)
+
+# Once we are running low on disk space, we calculate a free space rate that we
+# use until the free space goes back above the threshold.
+simulate baseline-rate=1MiB/s free-space-threshold=50GiB disk-free-space=40GB
+0s recent=0B queued=0B
+set-baseline-rate 1MiB/s
+set-free-space 40GiB
+2s recent=1MiB queued=1MiB
+3s recent=2MiB queued=3MiB
+add-debt 30GiB
+set-free-space 45GiB
+5s recent=3MiB queued=3MiB
+set-free-space 50GiB
+8s recent=3MiB queued=3MiB
+set-free-space 49GiB
+10s recent=3MiB queued=3MiB
+----
+0s: rate=1.3GiB/s (free-space=1.3GiB/s) debt=0B
+2s: rate=1.3GiB/s (free-space=1.3GiB/s) debt=0B
+3s: rate=1.3GiB/s (backlog=3.4KiB/s free-space=1.3GiB/s) debt=0B
+5s: rate=1.3GiB/s (free-space=1.3GiB/s) debt=0B
+8s: rate=1MiB/s debt=0B
+10s: rate=103MiB/s (free-space=102MiB/s) debt=0B
+
+simulate baseline-rate=1MiB/s
+0s recent=0B queued=0B
+add-debt 5MiB
+1s recent=0B queued=0B
+2s recent=0B queued=0B
+----
+0s: rate=1MiB/s debt=0B
+1s: rate=1MiB/s debt=4MiB (wait 4s)
+2s: rate=1MiB/s debt=3MiB (wait 3s)
+
+# Test disable pacing.
+simulate baseline-rate=10MiB/s
+0s recent=0B queued=0B
+1s recent=5MiB queued=5MiB disable-pacing
+2s recent=10MiB queued=15MiB
+----
+0s: rate=10MiB/s debt=0B
+1s: rate=0B/s debt=0B
+2s: rate=10MiB/s (backlog=17KiB/s) debt=0B
+
+# Test changing baseline rate.
+simulate baseline-rate=1MiB/s
+0s recent=0B queued=0B
+1s recent=1MiB queued=1MiB
+set-baseline-rate 5MiB/s
+3s recent=1MiB queued=1MiB
+4s recent=1MiB queued=1MiB
+set-baseline-rate 2MiB/s
+5s recent=1MiB queued=1MiB
+----
+0s: rate=1MiB/s debt=0B
+1s: rate=1MiB/s debt=0B
+3s: rate=5MiB/s debt=0B
+4s: rate=5MiB/s debt=0B
+5s: rate=2MiB/s debt=0B
+
+# Test changing free space.
+simulate baseline-rate=1MiB/s free-space-threshold=50GiB
+set-free-space 60GiB
+0s recent=0B queued=0B
+1s recent=1MiB queued=1MiB
+set-free-space 40GiB
+2s recent=1MiB queued=1MiB
+3s recent=1MiB queued=1MiB
+set-free-space 60GiB
+4s recent=1MiB queued=1MiB
+----
+0s: rate=1MiB/s debt=0B
+1s: rate=1MiB/s debt=0B
+2s: rate=1GiB/s (free-space=1GiB/s) debt=0B
+3s: rate=1GiB/s (free-space=1GiB/s) debt=0B
+4s: rate=1MiB/s debt=0B
diff --git a/metamorphic/options_test.go b/metamorphic/options_test.go
index a44f6560..55819c6c 100644
--- a/metamorphic/options_test.go
+++ b/metamorphic/options_test.go
@@ -75,7 +75,7 @@ func TestOptionsRoundtrip(t *testing.T) {
 		"EventListener:",
 		"CompactionConcurrencyRange:",
 		"MaxConcurrentDownloads:",
-		"TargetByteDeletionRate:",
+		"DeletionPacing.BaselineRate:",
 		"Experimental.CompactionGarbageFractionForMaxConcurrency:",
 		"Experimental.DisableIngestAsFlushable:",
 		"Experimental.EnableColumnarBlocks:",
@@ -119,7 +119,7 @@ func TestOptionsRoundtrip(t *testing.T) {
 		expectEqualFn(t, o.Opts.Experimental.CompactionGarbageFractionForMaxConcurrency, parsed.Opts.Experimental.CompactionGarbageFractionForMaxConcurrency)
 		expectEqualFn(t, o.Opts.Experimental.TombstoneDenseCompactionThreshold, parsed.Opts.Experimental.TombstoneDenseCompactionThreshold)
 		expectEqualFn(t, o.Opts.Experimental.ValueSeparationPolicy, parsed.Opts.Experimental.ValueSeparationPolicy)
-		expectEqualFn(t, o.Opts.TargetByteDeletionRate, parsed.Opts.TargetByteDeletionRate)
+		expectEqualFn(t, o.Opts.DeletionPacing.BaselineRate, parsed.Opts.DeletionPacing.BaselineRate)
 
 		expBaseline, expUpper := o.Opts.CompactionConcurrencyRange()
 		parsedBaseline, parsedUpper := parsed.Opts.CompactionConcurrencyRange()
diff --git a/metrics/count_and_size_test.go b/metrics/count_and_size_test.go
new file mode 100644
index 00000000..66fc3321
--- /dev/null
+++ b/metrics/count_and_size_test.go
@@ -0,0 +1,219 @@
+// Copyright 2025 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package metrics
+
+import (
+	"testing"
+
+	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/stretchr/testify/require"
+)
+
+func expect(t *testing.T, cs CountAndSize, expCount uint64, expBytes uint64) {
+	t.Helper()
+	require.Equal(t, expCount, cs.Count)
+	require.Equal(t, expBytes, cs.Bytes)
+}
+
+func TestTableCountsAndSizes(t *testing.T) {
+	var cs TableCountsAndSizes
+
+	// Add local table
+	cs.Inc(1000, true)
+	expect(t, cs.All, 1, 1000)
+	expect(t, cs.Local, 1, 1000)
+
+	// Add remote table
+	cs.Inc(2000, false)
+	expect(t, cs.All, 2, 3000)
+	expect(t, cs.Local, 1, 1000)
+
+	// Subtract local table
+	cs.Dec(500, true)
+	expect(t, cs.All, 1, 2500)
+	expect(t, cs.Local, 0, 500)
+
+	// Subtract remote table
+	cs.Dec(1000, false)
+	expect(t, cs.All, 0, 1500)
+	expect(t, cs.Local, 0, 500)
+
+	// Test Accumulate method
+	cs2 := TableCountsAndSizes{
+		All:   CountAndSize{Count: 5, Bytes: 5000},
+		Local: CountAndSize{Count: 3, Bytes: 3000},
+	}
+	cs.Accumulate(cs2)
+	expect(t, cs.All, 5, 6500)
+	expect(t, cs.Local, 3, 3500)
+
+	// Test Deduct method
+	cs.Deduct(cs2)
+	expect(t, cs.All, 0, 1500)
+	expect(t, cs.Local, 0, 500)
+}
+
+func TestBlobFileCountsAndSizes(t *testing.T) {
+	var cs BlobFileCountsAndSizes
+
+	// Add local blob
+	cs.Inc(5000, true)
+	expect(t, cs.All, 1, 5000)
+	expect(t, cs.Local, 1, 5000)
+
+	// Add remote blob
+	cs.Inc(3000, false)
+	expect(t, cs.All, 2, 8000)
+	expect(t, cs.Local, 1, 5000)
+
+	// Subtract local blob
+	cs.Dec(2000, true)
+	expect(t, cs.All, 1, 6000)
+	expect(t, cs.Local, 0, 3000)
+
+	// Subtract remote blob
+	cs.Dec(3000, false)
+	expect(t, cs.All, 0, 3000)
+	expect(t, cs.Local, 0, 3000)
+
+	// Test Accumulate method
+	cs2 := BlobFileCountsAndSizes{
+		All:   CountAndSize{Count: 4, Bytes: 8000},
+		Local: CountAndSize{Count: 2, Bytes: 4000},
+	}
+	cs.Accumulate(cs2)
+	expect(t, cs.All, 4, 11000)
+	expect(t, cs.Local, 2, 7000)
+
+	// Test Deduct method
+	cs.Deduct(cs2)
+	expect(t, cs.All, 0, 3000)
+	expect(t, cs.Local, 0, 3000)
+}
+
+func TestFileCountsAndSizes(t *testing.T) {
+	cs1 := FileCountsAndSizes{}
+	cs1.Tables.All = CountAndSize{Count: 5, Bytes: 1000}
+	cs1.Tables.Local = CountAndSize{Count: 3, Bytes: 600}
+	cs1.BlobFiles.All = CountAndSize{Count: 2, Bytes: 500}
+	cs1.BlobFiles.Local = CountAndSize{Count: 1, Bytes: 250}
+
+	cs2 := FileCountsAndSizes{}
+	cs2.Tables.All = CountAndSize{Count: 3, Bytes: 400}
+	cs2.Tables.Local = CountAndSize{Count: 2, Bytes: 300}
+	cs2.BlobFiles.All = CountAndSize{Count: 1, Bytes: 200}
+	cs2.BlobFiles.Local = CountAndSize{Count: 1, Bytes: 200}
+
+	sum := cs1
+	sum.Accumulate(cs2)
+
+	x := sum
+	x.Deduct(cs2)
+	require.Equal(t, cs1, x)
+
+	expect(t, sum.Tables.All, 8, 1400)
+	expect(t, sum.Tables.Local, 5, 900)
+	expect(t, sum.BlobFiles.All, 3, 700)
+	expect(t, sum.BlobFiles.Local, 2, 450)
+
+	sum.Inc(base.FileTypeTable, 10000, true /* isLocal */)
+	sum.Dec(base.FileTypeBlob, 250, false)
+
+	expect(t, sum.Tables.All, 9, 11400)
+	expect(t, sum.Tables.Local, 6, 10900)
+	expect(t, sum.BlobFiles.All, 2, 450)
+	expect(t, sum.BlobFiles.Local, 2, 450)
+}
+
+func TestFileCountsAndSizes_Inc_OtherTypes(t *testing.T) {
+	var cs FileCountsAndSizes
+
+	// Add log file
+	cs.Inc(base.FileTypeLog, 1000, true)
+	expect(t, cs.Other, 1, 1000)
+	expect(t, cs.Tables.All, 0, 0)
+	expect(t, cs.BlobFiles.All, 0, 0)
+
+	// Add manifest file
+	cs.Inc(base.FileTypeManifest, 2000, false)
+	expect(t, cs.Other, 2, 3000)
+
+	// Add options file
+	cs.Inc(base.FileTypeOptions, 100, true)
+	expect(t, cs.Other, 3, 3100)
+
+	// Subtract log file
+	cs.Dec(base.FileTypeLog, 500, false)
+	expect(t, cs.Other, 2, 2600)
+}
+
+func TestFileCountsAndSizes_Mixed(t *testing.T) {
+	var cs FileCountsAndSizes
+
+	// Add various files
+	cs.Inc(base.FileTypeTable, 1000, true)    // local table
+	cs.Inc(base.FileTypeTable, 2000, false)   // remote table
+	cs.Inc(base.FileTypeBlob, 3000, true)     // local blob
+	cs.Inc(base.FileTypeBlob, 4000, false)    // remote blob
+	cs.Inc(base.FileTypeLog, 500, true)       // log file
+	cs.Inc(base.FileTypeManifest, 200, false) // manifest file
+
+	expect(t, cs.Tables.All, 2, 3000)
+	expect(t, cs.Tables.Local, 1, 1000)
+	expect(t, cs.BlobFiles.All, 2, 7000)
+	expect(t, cs.BlobFiles.Local, 1, 3000)
+	expect(t, cs.Other, 2, 700)
+
+	// Subtract some files
+	cs.Dec(base.FileTypeTable, 1000, true)
+	cs.Dec(base.FileTypeBlob, 3000, true)
+	cs.Dec(base.FileTypeLog, 500, true)
+
+	expect(t, cs.Tables.All, 1, 2000)
+	expect(t, cs.Tables.Local, 0, 0)
+	expect(t, cs.BlobFiles.All, 1, 4000)
+	expect(t, cs.BlobFiles.Local, 0, 0)
+	expect(t, cs.Other, 1, 200)
+}
+
+func TestFileCountsAndSizes_AccumulateDeduct(t *testing.T) {
+	cs1 := FileCountsAndSizes{
+		Tables: TableCountsAndSizes{
+			All:   CountAndSize{Count: 10, Bytes: 2000},
+			Local: CountAndSize{Count: 6, Bytes: 1200},
+		},
+		BlobFiles: BlobFileCountsAndSizes{
+			All:   CountAndSize{Count: 5, Bytes: 1000},
+			Local: CountAndSize{Count: 3, Bytes: 600},
+		},
+		Other: CountAndSize{Count: 2, Bytes: 300},
+	}
+
+	cs2 := FileCountsAndSizes{
+		Tables: TableCountsAndSizes{
+			All:   CountAndSize{Count: 3, Bytes: 500},
+			Local: CountAndSize{Count: 2, Bytes: 400},
+		},
+		BlobFiles: BlobFileCountsAndSizes{
+			All:   CountAndSize{Count: 2, Bytes: 300},
+			Local: CountAndSize{Count: 1, Bytes: 200},
+		},
+		Other: CountAndSize{Count: 1, Bytes: 100},
+	}
+
+	// Test Accumulate
+	sum := cs1
+	sum.Accumulate(cs2)
+
+	expect(t, sum.Tables.All, 13, 2500)
+	expect(t, sum.Tables.Local, 8, 1600)
+	expect(t, sum.BlobFiles.All, 7, 1300)
+	expect(t, sum.BlobFiles.Local, 4, 800)
+	expect(t, sum.Other, 3, 400)
+
+	// Test Deduct
+	sum.Deduct(cs2)
+	require.Equal(t, cs1, sum)
+}
diff --git a/metrics_test.go b/metrics_test.go
index de4e7f2f..864724ba 100644
--- a/metrics_test.go
+++ b/metrics_test.go
@@ -21,10 +21,12 @@ import (
 	"github.com/cockroachdb/datadriven"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/cache"
+	"github.com/cockroachdb/pebble/internal/deletepacer"
 	"github.com/cockroachdb/pebble/internal/humanize"
 	"github.com/cockroachdb/pebble/internal/manifest"
 	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/internal/testutils"
+	"github.com/cockroachdb/pebble/metrics"
 	"github.com/cockroachdb/pebble/objstorage/remote"
 	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/vfs"
@@ -199,6 +201,18 @@ func exampleMetrics() Metrics {
 	m.WAL.BytesIn = 25
 	m.WAL.BytesWritten = 26
 
+	cs := func(n uint64) metrics.CountAndSize { return metrics.CountAndSize{Count: n, Bytes: n << 20} }
+	m.DeletePacer.InQueue.Tables.All = cs(100)
+	m.DeletePacer.InQueue.Tables.Local = cs(50)
+	m.DeletePacer.InQueue.BlobFiles.All = cs(200)
+	m.DeletePacer.InQueue.BlobFiles.Local = cs(100)
+	m.DeletePacer.InQueue.Other = cs(10)
+	m.DeletePacer.Deleted.Tables.All = cs(1000)
+	m.DeletePacer.Deleted.Tables.Local = cs(500)
+	m.DeletePacer.Deleted.BlobFiles.All = cs(2000)
+	m.DeletePacer.Deleted.BlobFiles.Local = cs(1000)
+	m.DeletePacer.Deleted.Other = cs(100)
+
 	for i := range m.manualMemory {
 		m.manualMemory[i].InUseBytes = uint64((i + 1) * 1024)
 	}
@@ -409,7 +423,7 @@ func TestMetrics(t *testing.T) {
 
 			// The deletion of obsolete files happens asynchronously when an iterator
 			// is closed. Wait for the obsolete tables to be deleted.
-			d.cleanupManager.Wait()
+			d.deletePacer.WaitForTesting()
 			return ""
 
 		case "iter-new":
@@ -452,6 +466,10 @@ func TestMetrics(t *testing.T) {
 					m.CategoryStats[0].CategoryStats = block.CategoryStats{}
 				}
 			}
+			// Some subset of cases show non-determinism in cache hits/misses.
+			if td.HasArg("zero-delete-pacer") {
+				m.DeletePacer = deletepacer.Metrics{}
+			}
 			var buf strings.Builder
 			fmt.Fprintf(&buf, "%s\n", m.StringForTests())
 			if len(m.CategoryStats) > 0 {
diff --git a/obsolete_files_test.go b/obsolete_files_test.go
index c1cb624c..7ae0d320 100644
--- a/obsolete_files_test.go
+++ b/obsolete_files_test.go
@@ -9,14 +9,11 @@ import (
 	"fmt"
 	"sort"
 	"strings"
-	"sync/atomic"
 	"testing"
-	"time"
 
 	"github.com/cockroachdb/datadriven"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/testutils"
-	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/stretchr/testify/require"
 )
@@ -142,131 +139,3 @@ func TestCleaner(t *testing.T) {
 		}
 	})
 }
-
-func TestCleanupManagerCloseWithPacing(t *testing.T) {
-	mem := vfs.NewMem()
-	opts := &Options{
-		FS:                      mem,
-		FreeSpaceThresholdBytes: 1,
-		TargetByteDeletionRate:  func() int { return 1024 }, // 1 KB/s - slow pacing
-	}
-	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 << 30,
-			liveBytes: 10 << 30,
-		}
-	}
-
-	cm := openCleanupManager(opts, objProvider, getDeletePacerInfo)
-
-	// Create obsolete files that would normally take a long time to delete.
-	// At 1 KB/s, 100 files of 10 KB each would take 1000 seconds.
-	largeFiles := make([]obsoleteFile, 100)
-	for i := range largeFiles {
-		largeFiles[i] = obsoleteFile{
-			fileType: base.FileTypeTable,
-			fs:       mem,
-			path:     fmt.Sprintf("test%02d.sst", i+1),
-			fileNum:  base.DiskFileNum(i + 1),
-			fileSize: 10 << 10,
-			isLocal:  true,
-		}
-	}
-
-	cm.EnqueueJob(1, largeFiles, obsoleteObjectStats{})
-
-	done := make(chan struct{})
-	go func() {
-		defer close(done)
-		cm.Close()
-	}()
-
-	select {
-	case <-done:
-	case <-time.After(30 * time.Second):
-		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()
-}
diff --git a/open_test.go b/open_test.go
index 811591b9..9d420f82 100644
--- a/open_test.go
+++ b/open_test.go
@@ -242,6 +242,12 @@ func TestOpen_WALFailover(t *testing.T) {
 				switch cmdArg.Key {
 				case "path":
 					o.FS, dataDir = extractFSAndPath(cmdArg)
+				case "wal-dir":
+					fs, dir := extractFSAndPath(cmdArg)
+					if fs != o.FS {
+						td.Fatalf(t, "WAL fs differs from data fs")
+					}
+					o.WALDir = dir
 				case "secondary":
 					fs, dir := extractFSAndPath(cmdArg)
 					o.WALFailover = &WALFailoverOptions{
@@ -250,6 +256,8 @@ func TestOpen_WALFailover(t *testing.T) {
 				case "wal-recovery-dir":
 					fs, dir := extractFSAndPath(cmdArg)
 					o.WALRecoveryDirs = append(o.WALRecoveryDirs, wal.Dir{FS: fs, Dirname: dir})
+				case "allow-missing-wal-dirs":
+					o.Unsafe.AllowMissingWALDirs = true
 				default:
 					return fmt.Sprintf("unrecognized cmdArg %q", cmdArg.Key)
 				}
@@ -501,13 +509,15 @@ func TestOpenAlreadyLocked(t *testing.T) {
 	// Run tests for different filesystems.
 	t.Run("memfs", func(t *testing.T) {
 		for _, tc := range testCases {
-			mem := vfs.NewMem()
-			var tmpDirs [4]string
-			for i := range tmpDirs {
-				tmpDirs[i] = mem.PathJoin("dir", fmt.Sprintf("%d", i))
-				require.NoError(t, mem.MkdirAll(tmpDirs[i], 0755), "Failed to create temp dir %s", tmpDirs[i])
-			}
-			runTest(t, tmpDirs, tc.setupLocks, mem)
+			t.Run(tc.name, func(t *testing.T) {
+				tmpDirs := [4]string{"dir/0", "dir/1", "dir/2", "dir/3"}
+				mem := vfs.NewMem()
+				for i := range tmpDirs {
+					require.NoError(t, mem.MkdirAll(tmpDirs[i], 0755),
+						"Failed to create temp dir %s", tmpDirs[i])
+				}
+				runTest(t, tmpDirs, tc.setupLocks, mem)
+			})
 		}
 	})
 
diff --git a/options_test.go b/options_test.go
index 052abbcf..496fac05 100644
--- a/options_test.go
+++ b/options_test.go
@@ -124,7 +124,6 @@ func TestDefaultOptionsString(t *testing.T) {
   min_deletion_rate=0
   free_space_threshold_bytes=17179869184
   free_space_timeframe=10s
-  obsolete_bytes_max_ratio=0.200000
   obsolete_bytes_timeframe=5m0s
   merger=pebble.concatenate
   multilevel_compaction_heuristic=wamp(0.00, false)
@@ -398,7 +397,7 @@ func TestOptionsParse(t *testing.T) {
 			opts.FlushDelayDeleteRange = 10 * time.Second
 			opts.FlushDelayRangeKey = 11 * time.Second
 			opts.Experimental.LevelMultiplier = 5
-			opts.TargetByteDeletionRate = func() int { return 200 }
+			opts.DeletionPacing.BaselineRate = func() uint64 { return 200 }
 			opts.WALFailover = &WALFailoverOptions{
 				Secondary: wal.Dir{Dirname: "wal_secondary", FS: vfs.Default},
 			}
diff --git a/pacer_test.go b/pacer_test.go
deleted file mode 100644
index 14f00a3b..00000000
--- a/pacer_test.go
+++ /dev/null
@@ -1,243 +0,0 @@
-// Copyright 2019 The LevelDB-Go and Pebble Authors. All rights reserved. Use
-// of this source code is governed by a BSD-style license that can be found in
-// the LICENSE file.
-
-package pebble
-
-import (
-	"cmp"
-	"fmt"
-	"math"
-	"math/rand/v2"
-	"slices"
-	"sync/atomic"
-	"testing"
-	"time"
-
-	"github.com/cockroachdb/crlib/crtime"
-	"github.com/stretchr/testify/require"
-)
-
-const MB = 1 << 20
-const GB = 1 << 30
-
-func TestDeletionPacer(t *testing.T) {
-	testCases := []struct {
-		freeBytes     uint64
-		obsoleteBytes uint64
-		liveBytes     uint64
-		// history of deletion reporting; first value in the pair is the time,
-		// second value is the deleted bytes. The time of pacing is the same as the
-		// last time in the history.
-		history [][2]int64
-		// expected pacing rate in MB/s.
-		expected float64
-	}{
-		{
-			freeBytes:     160 * GB,
-			obsoleteBytes: 1 * MB,
-			liveBytes:     160 * MB,
-			expected:      100.0,
-		},
-		// As freeBytes is 2GB below the free space threshold, rate should be
-		// increased by 204.8MB/s.
-		{
-			freeBytes:     14 * GB,
-			obsoleteBytes: 1 * MB,
-			liveBytes:     100 * MB,
-			expected:      304.8,
-		},
-		// As freeBytes is 10GB below the free space threshold, rate should be
-		// increased by 1GB/s.
-		{
-			freeBytes:     6 * GB,
-			obsoleteBytes: 1 * MB,
-			liveBytes:     100 * MB,
-			expected:      1124.0,
-		},
-		// obsoleteBytesRatio is 50%. We need to delete 30GB within 5 minutes.
-		{
-			freeBytes:     500 * GB,
-			obsoleteBytes: 50 * GB,
-			liveBytes:     100 * GB,
-			expected:      202.4,
-		},
-		// When obsolete ratio unknown, there should be no throttling.
-		{
-			freeBytes:     500 * GB,
-			obsoleteBytes: 0,
-			liveBytes:     0,
-			expected:      math.Inf(1),
-		},
-		// History shows 200MB/sec deletions on average over last 5 minutes.
-		{
-			freeBytes:     160 * GB,
-			obsoleteBytes: 1 * MB,
-			liveBytes:     160 * MB,
-			history:       [][2]int64{{0, 5 * 60 * 200 * MB}},
-			expected:      200.0,
-		},
-		// History shows 200MB/sec deletions on average over last 5 minutes and
-		// freeBytes is 10GB below the threshold.
-		{
-			freeBytes:     6 * GB,
-			obsoleteBytes: 1 * MB,
-			liveBytes:     160 * MB,
-			history:       [][2]int64{{0, 5 * 60 * 200 * MB}},
-			expected:      1224.0,
-		},
-		// History shows 200MB/sec deletions on average over last 5 minutes and
-		// obsoleteBytesRatio is 50%.
-		{
-			freeBytes:     500 * GB,
-			obsoleteBytes: 50 * GB,
-			liveBytes:     100 * GB,
-			history:       [][2]int64{{0, 5 * 60 * 200 * MB}},
-			expected:      302.4,
-		},
-		// History shows 1000MB/sec deletions on average over last 5 minutes.
-		{
-			freeBytes:     160 * GB,
-			obsoleteBytes: 1 * MB,
-			liveBytes:     160 * MB,
-			history:       [][2]int64{{0, 60 * 1000 * MB}, {3 * 60, 60 * 4 * 1000 * MB}, {4 * 60, 0}},
-			expected:      1000.0,
-		},
-		// First entry in history is too old, it should be discarded.
-		{
-			freeBytes:     160 * GB,
-			obsoleteBytes: 1 * MB,
-			liveBytes:     160 * MB,
-			history:       [][2]int64{{0, 10 * 60 * 10000 * MB}, {3 * 60, 4 * 60 * 200 * MB}, {7 * 60, 1 * 60 * 200 * MB}},
-			expected:      200.0,
-		},
-	}
-	for tcIdx, tc := range testCases {
-		t.Run(fmt.Sprintf("%d", tcIdx), func(t *testing.T) {
-			getInfo := func() deletionPacerInfo {
-				return deletionPacerInfo{
-					freeBytes:     tc.freeBytes,
-					liveBytes:     tc.liveBytes,
-					obsoleteBytes: tc.obsoleteBytes,
-				}
-			}
-			start := crtime.NowMono()
-			last := start
-			var opts Options
-			opts.EnsureDefaults()
-			pacer := newDeletionPacer(
-				start,
-				opts.FreeSpaceThresholdBytes,
-				func() int { return 100 * MB },
-				opts.FreeSpaceTimeframe,
-				opts.ObsoleteBytesMaxRatio,
-				opts.ObsoleteBytesTimeframe,
-				getInfo,
-			)
-			for _, h := range tc.history {
-				last = start + crtime.Mono(time.Second*time.Duration(h[0]))
-				pacer.ReportDeletion(last, uint64(h[1]))
-			}
-			result := 1.0 / pacer.PacingDelay(last, 1*MB)
-			require.InDelta(t, tc.expected, result, 1e-7)
-		})
-	}
-}
-
-func TestDeletionPacerCfgChange(t *testing.T) {
-	getInfo := func() deletionPacerInfo {
-		return deletionPacerInfo{
-			freeBytes:     100 * GB,
-			liveBytes:     10 * GB,
-			obsoleteBytes: 10 * MB,
-		}
-	}
-
-	var targetRate atomic.Int32
-	targetRate.Store(100 * MB)
-
-	start := crtime.NowMono()
-	var opts Options
-	opts.EnsureDefaults()
-	pacer := newDeletionPacer(
-		start,
-		opts.FreeSpaceThresholdBytes,
-		func() int { return int(targetRate.Load()) },
-		opts.FreeSpaceTimeframe,
-		opts.ObsoleteBytesMaxRatio,
-		opts.ObsoleteBytesTimeframe,
-		getInfo,
-	)
-	require.InDelta(t, 1.0/100, pacer.PacingDelay(start, 1*MB), 1e-4)
-	targetRate.Store(200 * MB)
-	require.InDelta(t, 1.0/200, pacer.PacingDelay(start, 1*MB), 1e-4)
-	targetRate.Store(0)
-	require.Equal(t, 0.0, pacer.PacingDelay(start, 1*MB), 1e-4)
-}
-
-// TestDeletionPacerHistory tests the history helper by crosschecking Sum()
-// against a naive implementation.
-func TestDeletionPacerHistory(t *testing.T) {
-	type event struct {
-		time crtime.Mono
-		// If report is 0, this event is a Sum(). Otherwise it is an Add().
-		report int64
-	}
-	numEvents := 1 + rand.IntN(200)
-	timeframe := time.Duration(1+rand.IntN(60*100)) * time.Second
-	events := make([]event, numEvents)
-	startTime := crtime.NowMono()
-	for i := range events {
-		events[i].time = startTime + crtime.Mono(rand.Int64N(int64(timeframe)))
-		if rand.IntN(3) == 0 {
-			events[i].report = 0
-		} else {
-			events[i].report = int64(rand.IntN(100000))
-		}
-	}
-	slices.SortFunc(events, func(a, b event) int { return cmp.Compare(a.time, b.time) })
-
-	var h history
-	h.Init(startTime, timeframe)
-
-	// partialSums[i] := SUM_j<i events[j].report
-	partialSums := make([]int64, len(events)+1)
-	for i := range events {
-		partialSums[i+1] = partialSums[i] + events[i].report
-	}
-
-	for i, e := range events {
-		if e.report != 0 {
-			h.Add(e.time, e.report)
-			continue
-		}
-
-		result := h.Sum(e.time)
-
-		// getIdx returns the largest event index <= i that is before the cutoff
-		// time.
-		getIdx := func(cutoff crtime.Mono) int {
-			for j := i; j >= 0; j-- {
-				if events[j].time < cutoff {
-					return j
-				}
-			}
-			return -1
-		}
-
-		// Sum all report values in the last timeframe, and see if recent events
-		// (allowing 1% error in the cutoff time) match the result.
-		a := getIdx(e.time - crtime.Mono(timeframe*(historyEpochs+1)/historyEpochs))
-		b := getIdx(e.time - crtime.Mono(timeframe*(historyEpochs-1)/historyEpochs))
-		found := false
-		for j := a; j <= b; j++ {
-			if partialSums[i+1]-partialSums[j+1] == result {
-				found = true
-				break
-			}
-		}
-		if !found {
-			t.Fatalf("incorrect Sum() result %d; %v", result, events[a+1:i+1])
-		}
-	}
-}
diff --git a/replay/testdata/replay b/replay/testdata/replay
index 7003f5cd..5a78b3f4 100644
--- a/replay/testdata/replay
+++ b/replay/testdata/replay
@@ -11,7 +11,7 @@ tree
      508      000007.sst
        0      LOCK
      133      MANIFEST-000001
-    2769      OPTIONS-000002
+    2733      OPTIONS-000002
        0      marker.format-version.000001.013
        0      marker.manifest.000001.MANIFEST-000001
             simple/
@@ -21,7 +21,7 @@ tree
       11        000004.log
      480        000005.sst
       85        MANIFEST-000001
-    2769        OPTIONS-000002
+    2733        OPTIONS-000002
        0        marker.format-version.000001.013
        0        marker.manifest.000001.MANIFEST-000001
 
@@ -59,7 +59,6 @@ cat build/OPTIONS-000002
   min_deletion_rate=0
   free_space_threshold_bytes=17179869184
   free_space_timeframe=10s
-  obsolete_bytes_max_ratio=0.200000
   obsolete_bytes_timeframe=5m0s
   merger=pebble.concatenate
   multilevel_compaction_heuristic=wamp(0.00, false)
diff --git a/replay/testdata/replay_ingest b/replay/testdata/replay_ingest
index 6ef2e344..9dbac546 100644
--- a/replay/testdata/replay_ingest
+++ b/replay/testdata/replay_ingest
@@ -12,7 +12,7 @@ tree
        0      LOCK
      172      MANIFEST-000001
      209      MANIFEST-000008
-    2770      OPTIONS-000002
+    2734      OPTIONS-000002
        0      marker.format-version.000012.025
        0      marker.manifest.000002.MANIFEST-000008
             simple_ingest/
@@ -25,7 +25,7 @@ tree
      678        000004.sst
      661        000005.sst
      172        MANIFEST-000001
-    2770        OPTIONS-000002
+    2734        OPTIONS-000002
        0        marker.format-version.000001.025
        0        marker.manifest.000001.MANIFEST-000001
 
@@ -63,7 +63,6 @@ cat build/OPTIONS-000002
   min_deletion_rate=0
   free_space_threshold_bytes=17179869184
   free_space_timeframe=10s
-  obsolete_bytes_max_ratio=0.200000
   obsolete_bytes_timeframe=5m0s
   merger=pebble.concatenate
   multilevel_compaction_heuristic=wamp(0.00, false)
diff --git a/replay/testdata/replay_paced b/replay/testdata/replay_paced
index 8d6392f7..04d22328 100644
--- a/replay/testdata/replay_paced
+++ b/replay/testdata/replay_paced
@@ -14,7 +14,7 @@ tree
        0      LOCK
      133      MANIFEST-000001
      205      MANIFEST-000010
-    2769      OPTIONS-000002
+    2733      OPTIONS-000002
        0      marker.format-version.000001.013
        0      marker.manifest.000002.MANIFEST-000010
             high_read_amp/
@@ -26,7 +26,7 @@ tree
       11        000008.log
      454        000009.sst
      157        MANIFEST-000010
-    2769        OPTIONS-000002
+    2733        OPTIONS-000002
        0        marker.format-version.000001.013
        0        marker.manifest.000001.MANIFEST-000010
 
diff --git a/replay/testdata/replay_val_sep b/replay/testdata/replay_val_sep
index c852d26d..d2954815 100644
--- a/replay/testdata/replay_val_sep
+++ b/replay/testdata/replay_val_sep
@@ -17,7 +17,7 @@ tree
        0      LOCK
      152      MANIFEST-000010
      250      MANIFEST-000013
-    2947      OPTIONS-000002
+    2911      OPTIONS-000002
        0      marker.format-version.000011.024
        0      marker.manifest.000003.MANIFEST-000013
             simple_val_sep/
@@ -32,7 +32,7 @@ tree
       11        000011.log
      707        000012.sst
      187        MANIFEST-000013
-    2947        OPTIONS-000002
+    2911        OPTIONS-000002
        0        marker.format-version.000001.024
        0        marker.manifest.000001.MANIFEST-000013
 
@@ -73,7 +73,6 @@ cat build/OPTIONS-000002
   min_deletion_rate=0
   free_space_threshold_bytes=17179869184
   free_space_timeframe=10s
-  obsolete_bytes_max_ratio=0.200000
   obsolete_bytes_timeframe=5m0s
   merger=pebble.concatenate
   multilevel_compaction_heuristic=wamp(0.00, false)
diff --git a/testdata/cleaner b/testdata/cleaner
index 7eb60c15..30088409 100644
--- a/testdata/cleaner
+++ b/testdata/cleaner
@@ -15,6 +15,7 @@ open-dir: db_wal
 close: db_wal
 open-dir: db
 lock: db/LOCK
+lock: db_wal/LOCK
 open-dir: db
 open-dir: db
 open-dir: db
@@ -24,7 +25,6 @@ create: db/marker.manifest.000001.MANIFEST-000001
 sync: db/marker.manifest.000001.MANIFEST-000001
 close: db/marker.manifest.000001.MANIFEST-000001
 sync: db
-lock: db_wal/LOCK
 open-dir: db_wal
 create: db/temporary.000002.dbtmp
 sync: db/temporary.000002.dbtmp
@@ -148,6 +148,7 @@ open-dir: db1_wal
 close: db1_wal
 open-dir: db1
 lock: db1/LOCK
+lock: db1_wal/LOCK
 open-dir: db1
 open-dir: db1
 open-dir: db1
@@ -157,7 +158,6 @@ create: db1/marker.manifest.000001.MANIFEST-000001
 sync: db1/marker.manifest.000001.MANIFEST-000001
 close: db1/marker.manifest.000001.MANIFEST-000001
 sync: db1
-lock: db1_wal/LOCK
 open-dir: db1_wal
 create: db1/temporary.000002.dbtmp
 sync: db1/temporary.000002.dbtmp
@@ -242,12 +242,12 @@ open-dir: db1_wal
 close: db1_wal
 open-dir: db1
 lock: db1/LOCK
+lock: db1_wal/LOCK
 open-dir: db1
 open-dir: db1
 open-dir: db1
 open: db1/MANIFEST-000001
 close: db1/MANIFEST-000001
-lock: db1_wal/LOCK
 remove: db1_wal/000003.log
 open-dir: db1_wal
 open: db1/OPTIONS-000002
diff --git a/testdata/compaction/l0_to_lbase_compaction b/testdata/compaction/l0_to_lbase_compaction
index defe2c8a..1abd7992 100644
--- a/testdata/compaction/l0_to_lbase_compaction
+++ b/testdata/compaction/l0_to_lbase_compaction
@@ -98,5 +98,11 @@ COMPRESSION
     algorithm |        tables |    blob files
 --------------+---------------+--------------
          none |         234KB |
+
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
 ----
 ----
diff --git a/testdata/compaction/value_separation b/testdata/compaction/value_separation
index 640c0851..5bfdf14f 100644
--- a/testdata/compaction/value_separation
+++ b/testdata/compaction/value_separation
@@ -218,6 +218,12 @@ COMPRESSION
 --------------+----------------+--------------
          none |            56B |           64B
        snappy | 134B (CR=1.15) |
+
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
 ----
 ----
 
@@ -521,6 +527,12 @@ COMPRESSION
 --------------+----------------+--------------
          none |            61B |          149B
        snappy | 130B (CR=1.31) |
+
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
 ----
 ----
 
diff --git a/testdata/compaction/virtual_rewrite b/testdata/compaction/virtual_rewrite
index 28196314..2b122824 100644
--- a/testdata/compaction/virtual_rewrite
+++ b/testdata/compaction/virtual_rewrite
@@ -202,6 +202,12 @@ COMPRESSION
 --------------+----------------+--------------
          none |           116B |
        snappy | 296B (CR=1.23) |
+
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
 ----
 ----
 
diff --git a/testdata/event_listener b/testdata/event_listener
index c91a4587..f5c598da 100644
--- a/testdata/event_listener
+++ b/testdata/event_listener
@@ -14,6 +14,7 @@ open-dir: wal
 close: wal
 open-dir: db
 lock: db/LOCK
+lock: wal/LOCK
 open-dir: db
 open-dir: db
 open-dir: db
@@ -24,7 +25,6 @@ sync: db/marker.manifest.000001.MANIFEST-000001
 close: db/marker.manifest.000001.MANIFEST-000001
 sync: db
 [JOB 1] MANIFEST created 000001
-lock: wal/LOCK
 open-dir: wal
 create: db/temporary.000002.dbtmp
 sync: db/temporary.000002.dbtmp
@@ -344,6 +344,12 @@ COMPRESSION
 --------------+---------------+--------------
          none |          260B |
        snappy | 73B (CR=1.15) |
+
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
 ----
 ----
 
@@ -487,6 +493,12 @@ COMPRESSION
 --------------+----------------+--------------
          none |           520B |
        snappy | 146B (CR=1.15) |
+
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
 ----
 ----
 
diff --git a/testdata/ingest b/testdata/ingest
index 637c2faf..ebc69763 100644
--- a/testdata/ingest
+++ b/testdata/ingest
@@ -93,6 +93,12 @@ COMPRESSION
     algorithm |        tables |    blob files
 --------------+---------------+--------------
       unknown |          463B |
+
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
 ----
 ----
 
diff --git a/testdata/metrics b/testdata/metrics
index 492b6ba5..04f68a3c 100644
--- a/testdata/metrics
+++ b/testdata/metrics
@@ -85,6 +85,12 @@ COMPRESSION
         minlz |    1GB (CR=3) |   10GB (CR=3)
          zstd |   10GB (CR=5) |  100GB (CR=5)
       unknown |         500MB |          50MB
+
+DELETE PACER   |             in queue             |            deleted
+---------------+----------------------------------+------------------------------
+        tables |  100 (100MB) [local: 50 (50MB)]  | 1K (1GB) [local: 500 (500MB)]
+    blob files | 200 (200MB) [local: 100 (100MB)] |  2K (2GB) [local: 1K (1GB)]
+   other files |            10 (10MB)             |          100 (100MB)
 ----
 ----
 
@@ -175,6 +181,12 @@ COMPRESSION
          none |           36B |
        snappy | 76B (CR=1.14) |
 
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
+
 Iter category stats:
    pebble-compaction, non-latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
                    a, non-latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
@@ -184,7 +196,7 @@ Iter category stats:
 
 disk-usage
 ----
-3,709B
+3,673B
 
 batch
 set b 2
@@ -277,6 +289,12 @@ COMPRESSION
 --------------+---------------+--------------
          none |          230B |
 
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
+
 Iter category stats:
    pebble-compaction, non-latency: {BlockBytes:738 BlockBytesInCache:112 BlockReadDuration:20ms}
                    a, non-latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
@@ -287,7 +305,7 @@ Iter category stats:
 
 disk-usage
 ----
-6,002B
+5,966B
 
 # Closing iter a will release one of the zombie memtables.
 
@@ -362,6 +380,12 @@ COMPRESSION
 --------------+---------------+--------------
          none |          230B |
 
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
+
 Iter category stats:
    pebble-compaction, non-latency: {BlockBytes:738 BlockBytesInCache:112 BlockReadDuration:20ms}
                    a, non-latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
@@ -444,6 +468,12 @@ COMPRESSION
 --------------+---------------+--------------
          none |          230B |
 
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
+
 Iter category stats:
    pebble-compaction, non-latency: {BlockBytes:738 BlockBytesInCache:112 BlockReadDuration:20ms}
                    a, non-latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
@@ -454,7 +484,7 @@ Iter category stats:
 
 disk-usage
 ----
-5,295B
+5,259B
 
 # Closing iter b will release the last zombie sstable and the last zombie memtable.
 
@@ -529,6 +559,12 @@ COMPRESSION
 --------------+---------------+--------------
          none |          230B |
 
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
+
 Iter category stats:
    pebble-compaction, non-latency: {BlockBytes:738 BlockBytesInCache:112 BlockReadDuration:20ms}
                    a, non-latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
@@ -539,7 +575,7 @@ Iter category stats:
 
 disk-usage
 ----
-4,588B
+4,552B
 
 additional-metrics
 ----
@@ -654,6 +690,12 @@ COMPRESSION
          none |           622B |          308B
        snappy | 630B (CR=1.27) |
 
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
+
 Iter category stats:
    pebble-compaction, non-latency: {BlockBytes:738 BlockBytesInCache:112 BlockReadDuration:20ms}
                    a, non-latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
@@ -764,6 +806,12 @@ COMPRESSION
          none |          622B |          308B
        snappy | 615B (CR=1.3) |
 
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
+
 Iter category stats:
    pebble-compaction, non-latency: {BlockBytes:5575 BlockBytesInCache:112 BlockReadDuration:160ms}
                    a, non-latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
@@ -925,6 +973,12 @@ COMPRESSION
          none |            1KB |          308B
        snappy | 843B (CR=1.26) |
 
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
+
 Iter category stats:
    pebble-compaction, non-latency: {BlockBytes:5575 BlockBytesInCache:112 BlockReadDuration:160ms}
                    a, non-latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
@@ -1048,6 +1102,12 @@ COMPRESSION
          none |           1.3KB |          308B
        snappy | 1.3KB (CR=1.21) |
 
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
+
 Iter category stats:
    pebble-compaction, non-latency: {BlockBytes:5575 BlockBytesInCache:112 BlockReadDuration:160ms}
                    a, non-latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
@@ -1114,7 +1174,7 @@ virtual-size
 0
 0B
 
-metrics zero-cache-hits-misses
+metrics zero-cache-hits-misses zero-delete-pacer
 ----
 ----
 LSM                             |    vtables   |   value sep   |        |   ingested   |    amp
@@ -1180,6 +1240,12 @@ COMPRESSION
          none |           1.3KB |          308B
        snappy | 1.2KB (CR=1.22) |
 
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
+
 Iter category stats:
    pebble-compaction, non-latency: {BlockBytes:5575 BlockBytesInCache:112 BlockReadDuration:160ms}
                    a, non-latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
@@ -1290,7 +1356,7 @@ virtual-size
 0
 0B
 
-metrics zero-cache-hits-misses
+metrics zero-cache-hits-misses zero-delete-pacer
 ----
 ----
 LSM                             |    vtables   |   value sep   |        |   ingested   |    amp
@@ -1356,6 +1422,12 @@ COMPRESSION
          none |         1.5KB |          308B
        snappy | 682B (CR=1.3) |
 
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
+
 Iter category stats:
    pebble-compaction, non-latency: {BlockBytes:11679 BlockBytesInCache:457 BlockReadDuration:330ms}
                    a, non-latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
@@ -1447,6 +1519,12 @@ COMPRESSION
          none |           108B |
        snappy | 228B (CR=1.14) |
 
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
+
 Iter category stats:
    pebble-compaction, non-latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
 ----
@@ -1528,6 +1606,12 @@ COMPRESSION
 --------------+---------------+--------------
          none |          345B |
 
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
+
 Iter category stats:
    pebble-compaction, non-latency: {BlockBytes:1878 BlockBytesInCache:0 BlockReadDuration:60ms}
 ----
@@ -1625,6 +1709,12 @@ COMPRESSION
 --------------+---------------+--------------
          none |          460B |
 
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
+
 Iter category stats:
    pebble-compaction, non-latency: {BlockBytes:1878 BlockBytesInCache:0 BlockReadDuration:60ms}
 ----
@@ -1715,6 +1805,12 @@ COMPRESSION
          none |          496B |
        snappy | 76B (CR=1.14) |
 
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
+
 Iter category stats:
    pebble-compaction, non-latency: {BlockBytes:1878 BlockBytesInCache:0 BlockReadDuration:60ms}
 ----
@@ -1725,7 +1821,7 @@ Iter category stats:
 init reopen
 ----
 
-metrics zero-cache-hits-misses
+metrics zero-cache-hits-misses zero-delete-pacer
 ----
 ----
 LSM                             |    vtables   |   value sep   |        |   ingested   |    amp
@@ -1791,6 +1887,12 @@ COMPRESSION
          none |          496B |
        snappy | 76B (CR=1.14) |
 
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
+
 Iter category stats:
    pebble-compaction, non-latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
 ----
@@ -1804,7 +1906,7 @@ L6:
   000010:[c#0,SET-c#0,SET]
 
 # All tables are local after compaction.
-metrics zero-cache-hits-misses
+metrics zero-cache-hits-misses zero-delete-pacer
 ----
 ----
 LSM                             |    vtables   |   value sep   |        |   ingested   |    amp
@@ -1867,6 +1969,12 @@ COMPRESSION
 --------------+---------------+--------------
          none |          345B |
 
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
+
 Iter category stats:
    pebble-compaction, non-latency: {BlockBytes:342 BlockBytesInCache:0 BlockReadDuration:30ms}
 ----
@@ -1987,6 +2095,12 @@ COMPRESSION
          none |           266B |
        snappy | 221B (CR=1.19) |
 
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
+
 Iter category stats:
    pebble-compaction, non-latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
 ----
@@ -2066,6 +2180,12 @@ COMPRESSION
          none |           266B |
        snappy | 221B (CR=1.19) |
 
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
+
 Iter category stats:
    pebble-compaction, non-latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
 ----
diff --git a/testdata/open_wal_failover b/testdata/open_wal_failover
index 0c1bf9c9..9f7a0393 100644
--- a/testdata/open_wal_failover
+++ b/testdata/open_wal_failover
@@ -69,3 +69,52 @@ directory "secondary-wals" may contain relevant WALs but is not in WALRecoveryDi
 open path=(a,data) wal-recovery-dir=(b,secondary-wals)
 ----
 ok
+
+open path=(c,data) wal-dir=(c,wal1)
+----
+ok
+
+# We cannot change the WAL directory without providing the previous one as a recovery dir.
+open path=(c,data) wal-dir=(c,wal2)
+----
+directory "wal1" may contain relevant WALs but is not in WALRecoveryDirs
+  WALDir changed from previous options
+  o.WALDir: "wal2"
+  o.WALRecoveryDirs: 0
+
+open path=(c,data) wal-dir=(c,wal2) wal-recovery-dir=(c,wal1)
+----
+ok
+
+# Once we opened the database once (and thus replayed any WALs in the previous
+# WAL dir), we can open without a recovery dir.
+open path=(c,data) wal-dir=(c,wal2)
+----
+ok
+
+open path=(c,data) wal-dir=(c,wal3)
+----
+directory "wal2" may contain relevant WALs but is not in WALRecoveryDirs
+  WALDir changed from previous options
+  o.WALDir: "wal3"
+  o.WALRecoveryDirs: 0
+
+# Test the Unsafe.AllowMissingWALDirs option.
+open path=(c,data) wal-dir=(c,wal3) allow-missing-wal-dirs
+----
+ok
+
+open path=(c,data) wal-dir=(c,wal3) secondary=(d,secondary)
+----
+ok
+
+open path=(c,data) wal-dir=(c,wal3)
+----
+directory "secondary" may contain relevant WALs but is not in WALRecoveryDirs
+  WALFailover.Secondary changed from previous options
+  o.WALDir: "wal3"
+  o.WALRecoveryDirs: 0
+
+open path=(c,data) wal-dir=(c,wal3) allow-missing-wal-dirs
+----
+ok
diff --git a/tool/testdata/db_lsm b/tool/testdata/db_lsm
index 5f92381d..718bb36c 100644
--- a/tool/testdata/db_lsm
+++ b/tool/testdata/db_lsm
@@ -70,6 +70,12 @@ COMPRESSION
     algorithm |        tables |    blob files
 --------------+---------------+--------------
       unknown |          709B |
+
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
 ----
 ----
 
@@ -136,6 +142,12 @@ COMPRESSION
     algorithm |        tables |    blob files
 --------------+---------------+--------------
       unknown |          709B |
+
+DELETE PACER   |   in queue   |   deleted
+---------------+--------------+-------------
+        tables |    0 (0B)    |    0 (0B)
+    blob files |    0 (0B)    |    0 (0B)
+   other files |    0 (0B)    |    0 (0B)
 LSM viewer: https://raduberinde.github.io/lsmview/decode.html#eJyE0EFLw0AQBeC7v2J4uU5lN42W7lHsrTe9SSgTOi2hm13NRqGV_HdJCaUWMXvaxzfMwPuG1y_1Ce5t_G6CNAqHtbk3YHRSeR1ZKvVwKMBI9UnhFmbJSI14r6nbHPQIZxhe2v0lW8ZWO6nPJ2CGVziqpM1swc-rNc1oF2Nm5_yyeh0XO1qY5dMQ9CN8NsmRzWlGdjj8HuvQpf820JWN8ZTZ_KI3wyFSK2GvtB1qKPuy59sm7HUPf3g-4fMJLyb8YcIff3vJOOjx3HclLRi7GFH2dz8BAAD__2dulBM=
 ----
 ----
diff --git a/version_set_test.go b/version_set_test.go
index b6b88211..163959d5 100644
--- a/version_set_test.go
+++ b/version_set_test.go
@@ -23,6 +23,7 @@ import (
 	"github.com/cockroachdb/crlib/crstrings"
 	"github.com/cockroachdb/datadriven"
 	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/deletepacer"
 	"github.com/cockroachdb/pebble/internal/manifest"
 	"github.com/cockroachdb/pebble/internal/strparse"
 	"github.com/cockroachdb/pebble/internal/testkeys"
@@ -233,7 +234,7 @@ func TestVersionSet(t *testing.T) {
 			}
 		}
 		buf.WriteString(vs.latest.virtualBackings.String())
-		printObjectBreakdown := func(kind string, zombies zombieObjects, obsolete []obsoleteFile) {
+		printObjectBreakdown := func(kind string, zombies zombieObjects, obsolete []deletepacer.ObsoleteFile) {
 			if zombies.Count() == 0 {
 				buf.WriteString(fmt.Sprintf("no zombie %s\n", kind))
 			} else {
@@ -250,7 +251,7 @@ func TestVersionSet(t *testing.T) {
 			} else {
 				buf.WriteString(fmt.Sprintf("obsolete %s:", kind))
 				for _, fi := range obsolete {
-					fmt.Fprintf(&buf, " %s", fi.fileNum)
+					fmt.Fprintf(&buf, " %s", fi.FileNum)
 				}
 				buf.WriteString("\n")
 			}