Repository: cockroachdb/pebble
Pull request #529 (2020-02-11 01:11:34)

internal/cache: explicitly manage Cache lifetime
Add `Cache.{Ref,Unref}()` to explicitly manage the lifetime of a
`Cache`.

petermattis: 

<!-- Reviewable:start -->
This change is [<img src="https://reviewable.io/review_button.png" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/cockroachdb/pebble/529)
<!-- Reviewable:end -->

petermattis: Ok, `AutoReclaim` has been excised. I need to do another cleanup pass on this PR to see if I can remove some of the redundancy in the test changes. But the general shape of this PR is good now.

Patch:
diff --git a/.travis.yml b/.travis.yml
index 04d9fb88..fe770966 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -30,6 +30,11 @@ matrix:
       go: 1.13.x
       os: windows
       script: go test ./...
+    - name: "go1.13.x-freebsd"
+      go: 1.13.x
+      os: linux
+      # NB: "env: GOOS=freebsd" does not have the desired effect.
+      script: GOOS=freebsd go build -v ./...
 
 notifications:
   email:
diff --git a/cmd/pebble/db.go b/cmd/pebble/db.go
index 2ed8fa43..821433b7 100644
--- a/cmd/pebble/db.go
+++ b/cmd/pebble/db.go
@@ -47,8 +47,10 @@ type pebbleDB struct {
 }
 
 func newPebbleDB(dir string) DB {
+	cache := pebble.NewCache(cacheSize)
+	defer cache.Unref()
 	opts := &pebble.Options{
-		Cache:                       pebble.NewCache(cacheSize),
+		Cache:                       cache,
 		Comparer:                    mvccComparer,
 		DisableWAL:                  disableWAL,
 		MemTableSize:                64 << 20,
diff --git a/db.go b/db.go
index 5936574b..ae4e44d0 100644
--- a/db.go
+++ b/db.go
@@ -808,6 +808,9 @@ func (d *DB) Close() error {
 		panic(ErrClosed)
 	}
 	atomic.StoreInt32(&d.closed, 1)
+
+	defer d.opts.Cache.Unref()
+
 	for d.mu.compact.compactingCount > 0 || d.mu.compact.flushing {
 		d.mu.compact.cond.Wait()
 	}
diff --git a/internal/cache/clockpro.go b/internal/cache/clockpro.go
index 1f95c6b6..8c06b45a 100644
--- a/internal/cache/clockpro.go
+++ b/internal/cache/clockpro.go
@@ -22,6 +22,7 @@ import (
 	"os"
 	"runtime"
 	"runtime/debug"
+	"strings"
 	"sync"
 	"sync/atomic"
 
@@ -218,6 +219,7 @@ func (c *shard) Set(id, fileNum, offset uint64, value *Value) Handle {
 		// cache entry was a test page
 		c.sizeTest -= e.size
 		c.metaDel(e)
+		c.metaCheck(e)
 
 		c.coldTarget += e.size
 		if c.coldTarget > c.targetSize() {
@@ -278,6 +280,20 @@ func (c *shard) EvictFile(id, fileNum uint64) {
 	}
 }
 
+func (c *shard) Free() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	// NB: we use metaDel rather than metaEvict in order to avoid the expensive
+	// metaCheck call when the "invariants" build tag is specified.
+	for c.handHot != nil {
+		c.metaDel(c.handHot)
+	}
+
+	c.blocks.free()
+	c.files.free()
+}
+
 func (c *shard) Reserve(n int) {
 	c.mu.Lock()
 	c.reservedSize += int64(n)
@@ -381,8 +397,6 @@ func (c *shard) metaDel(e *entry) {
 	} else {
 		c.files.Put(fkey, next)
 	}
-
-	c.metaCheck(e)
 }
 
 // Check that the specified entry is not referenced by the cache.
@@ -425,6 +439,7 @@ func (c *shard) metaEvict(e *entry) {
 		c.sizeTest -= e.size
 	}
 	c.metaDel(e)
+	c.metaCheck(e)
 	e.free()
 }
 
@@ -498,6 +513,7 @@ func (c *shard) runHandTest() {
 			c.coldTarget = 0
 		}
 		c.metaDel(e)
+		c.metaCheck(e)
 		e.free()
 	}
 
@@ -576,36 +592,39 @@ type Metrics struct {
 // used in combination by specifying `-tags invariants,tracing`. Note that
 // "tracing" produces a significant slowdown, while "invariants" does not.
 type Cache struct {
+	refs    int64
 	maxSize int64
 	idAlloc uint64
 	shards  []shard
+
+	// Traces recorded by Cache.trace. Used for debugging.
+	tr struct {
+		sync.Mutex
+		msgs []string
+	}
 }
 
 // New creates a new cache of the specified size. Memory for the cache is
-// allocated on demand, not during initialization.
+// allocated on demand, not during initialization. The cache is created with a
+// reference count of 1. Each DB it is associated with adds a reference, so the
+// creator of the cache should usually release their reference after the DB is
+// created.
+//
+//   c := cache.New(...)
+//   d, err := pebble.Open(pebble.Options{Cache: c})
+//   c.Unref()
 func New(size int64) *Cache {
 	return newShards(size, 2*runtime.NumCPU())
 }
 
-func clearCache(obj interface{}) {
-	c := obj.(*Cache)
-	for i := range c.shards {
-		s := &c.shards[i]
-		s.mu.Lock()
-		s.maxSize = 0
-		s.evict()
-		s.blocks.free()
-		s.files.free()
-		s.mu.Unlock()
-	}
-}
-
 func newShards(size int64, shards int) *Cache {
 	c := &Cache{
+		refs:    1,
 		maxSize: size,
 		idAlloc: 1,
 		shards:  make([]shard, shards),
 	}
+	c.trace("alloc", c.refs)
 	for i := range c.shards {
 		c.shards[i] = shard{
 			maxSize:    size / int64(len(c.shards)),
@@ -615,14 +634,28 @@ func newShards(size int64, shards int) *Cache {
 		c.shards[i].blocks.init(16)
 		c.shards[i].files.init(16)
 	}
-	// TODO(peter): This finalizer is used to clear the cache when the Cache
-	// itself is GC'd. Investigate making this explicit, and then changing the
-	// finalizer to only be installed when invariants.Enabled is true and to only
-	// check that all of the manual memory has been freed.
-	runtime.SetFinalizer(c, clearCache)
+	runtime.SetFinalizer(c, func(obj interface{}) {
+		c := obj.(*Cache)
+		if v := atomic.LoadInt64(&c.refs); v != 0 {
+			c.tr.Lock()
+			fmt.Fprintf(os.Stderr, "pebble: cache (%p) has non-zero reference count: %d\n%s",
+				c, v, strings.Join(c.tr.msgs, "\n"))
+			c.tr.Unlock()
+			os.Exit(1)
+		}
+	})
 	return c
 }
 
+func (c *Cache) trace(msg string, refs int64) {
+	if invariants.Enabled {
+		s := fmt.Sprintf("%s: refs=%d\n%s", msg, refs, debug.Stack())
+		c.tr.Lock()
+		c.tr.msgs = append(c.tr.msgs, s)
+		c.tr.Unlock()
+	}
+}
+
 func (c *Cache) getShard(id, fileNum, offset uint64) *shard {
 	if id == 0 {
 		panic("pebble: 0 cache ID is invalid")
@@ -652,6 +685,30 @@ func (c *Cache) getShard(id, fileNum, offset uint64) *shard {
 	return &c.shards[h%uint64(len(c.shards))]
 }
 
+// Ref adds a reference to the cache. The cache only remains valid as long a
+// reference is maintained to it.
+func (c *Cache) Ref() {
+	v := atomic.AddInt64(&c.refs, 1)
+	if v <= 1 {
+		panic(fmt.Sprintf("pebble: inconsistent reference count: %d", v))
+	}
+	c.trace("ref", v)
+}
+
+// Unref releases a reference on the cache.
+func (c *Cache) Unref() {
+	v := atomic.AddInt64(&c.refs, -1)
+	c.trace("unref", v)
+	switch {
+	case v < 0:
+		panic(fmt.Sprintf("pebble: inconsistent reference count: %d", v))
+	case v == 0:
+		for i := range c.shards {
+			c.shards[i].Free()
+		}
+	}
+}
+
 // Get retrieves the cache value for the specified file and offset, returning
 // nil if no value is present.
 func (c *Cache) Get(id, fileNum, offset uint64) Handle {
diff --git a/internal/cache/robin_hood.go b/internal/cache/robin_hood.go
index 7ad32fc2..b69844cf 100644
--- a/internal/cache/robin_hood.go
+++ b/internal/cache/robin_hood.go
@@ -134,15 +134,16 @@ func maxDistForSize(size uint32) uint32 {
 func newRobinHoodMap(initialCapacity int) *robinHoodMap {
 	m := &robinHoodMap{}
 	m.init(initialCapacity)
-	runtime.SetFinalizer(m, clearRobinHoodMap)
+	runtime.SetFinalizer(m, func(obj interface{}) {
+		m := obj.(*robinHoodMap)
+		if m.entries.ptr != nil {
+			fmt.Fprintf(os.Stderr, "%p: robin-hood map not freed\n", m)
+			os.Exit(1)
+		}
+	})
 	return m
 }
 
-func clearRobinHoodMap(obj interface{}) {
-	m := obj.(*robinHoodMap)
-	m.free()
-}
-
 func (m *robinHoodMap) init(initialCapacity int) {
 	if initialCapacity < 1 {
 		initialCapacity = 1
diff --git a/internal/cache/value_invariants.go b/internal/cache/value_invariants.go
index eca839f5..bb30b840 100644
--- a/internal/cache/value_invariants.go
+++ b/internal/cache/value_invariants.go
@@ -13,15 +13,6 @@ import (
 	"sync/atomic"
 )
 
-func checkValue(obj interface{}) {
-	v := obj.(*Value)
-	if v.buf != nil {
-		fmt.Fprintf(os.Stderr, "%p: cache value was not freed: refs=%d\n%s",
-			v.buf, atomic.LoadInt32(&v.refs), v.traces())
-		os.Exit(1)
-	}
-}
-
 // newManualValue creates a Value with a manually managed buffer of size n.
 //
 // This definition of newManualValue is used when either the "invariants" or
@@ -35,7 +26,14 @@ func newManualValue(n int) *Value {
 	b := allocNew(n)
 	v := &Value{buf: b, refs: 1}
 	v.trace("alloc")
-	runtime.SetFinalizer(v, checkValue)
+	runtime.SetFinalizer(v, func(obj interface{}) {
+		v := obj.(*Value)
+		if v.buf != nil {
+			fmt.Fprintf(os.Stderr, "%p: cache value was not freed: refs=%d\n%s",
+				v, atomic.LoadInt32(&v.refs), v.traces())
+			os.Exit(1)
+		}
+	})
 	return v
 }
 
diff --git a/open.go b/open.go
index cd799e3c..ad7ab9a9 100644
--- a/open.go
+++ b/open.go
@@ -16,6 +16,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/rate"
 	"github.com/cockroachdb/pebble/internal/record"
 	"github.com/cockroachdb/pebble/vfs"
@@ -24,7 +25,7 @@ import (
 const initialMemTableSize = 256 << 10 // 256 KB
 
 // Open opens a LevelDB whose files live in the given directory.
-func Open(dirname string, opts *Options) (*DB, error) {
+func Open(dirname string, opts *Options) (db *DB, _ error) {
 	// Make a copy of the options so that we don't mutate the passed in options.
 	opts = opts.Clone()
 	opts = opts.EnsureDefaults()
@@ -32,6 +33,12 @@ func Open(dirname string, opts *Options) (*DB, error) {
 		return nil, err
 	}
 
+	if opts.Cache == nil {
+		opts.Cache = cache.New(cacheDefaultSize)
+	} else {
+		opts.Cache.Ref()
+	}
+
 	d := &DB{
 		cacheID:             opts.Cache.NewID(),
 		dirname:             dirname,
@@ -45,6 +52,16 @@ func Open(dirname string, opts *Options) (*DB, error) {
 		largeBatchThreshold: (opts.MemTableSize - int(memTableEmptySize)) / 2,
 		logRecycler:         logRecycler{limit: opts.MemTableStopWritesThreshold + 1},
 	}
+	defer func() {
+		if db == nil {
+			// We're failing to return the DB for some reason. Release our references
+			// to the Cache. Note that both the DB, and tableCache have a reference
+			// and we need to release both.
+			opts.Cache.Unref()
+			opts.Cache.Unref()
+		}
+	}()
+
 	if d.equal == nil {
 		d.equal = bytes.Equal
 	}
diff --git a/options.go b/options.go
index 6519ef80..f1a35ffd 100644
--- a/options.go
+++ b/options.go
@@ -18,6 +18,10 @@ import (
 	"github.com/cockroachdb/pebble/vfs"
 )
 
+const (
+	cacheDefaultSize = 8 << 20 // 8 MB
+)
+
 // Compression exports the base.Compression type.
 type Compression = sstable.Compression
 
@@ -404,9 +408,6 @@ func (o *Options) EnsureDefaults() *Options {
 	if o.BytesPerSync <= 0 {
 		o.BytesPerSync = 512 << 10 // 512 KB
 	}
-	if o.Cache == nil {
-		o.Cache = cache.New(8 << 20) // 8 MB
-	}
 	if o.Cleaner == nil {
 		o.Cleaner = DeleteCleaner{}
 	}
@@ -522,12 +523,17 @@ func filterPolicyName(p FilterPolicy) string {
 func (o *Options) String() string {
 	var buf bytes.Buffer
 
+	cacheSize := int64(cacheDefaultSize)
+	if o.Cache != nil {
+		cacheSize = o.Cache.MaxSize()
+	}
+
 	fmt.Fprintf(&buf, "[Version]\n")
 	fmt.Fprintf(&buf, "  pebble_version=0.1\n")
 	fmt.Fprintf(&buf, "\n")
 	fmt.Fprintf(&buf, "[Options]\n")
 	fmt.Fprintf(&buf, "  bytes_per_sync=%d\n", o.BytesPerSync)
-	fmt.Fprintf(&buf, "  cache_size=%d\n", o.Cache.MaxSize())
+	fmt.Fprintf(&buf, "  cache_size=%d\n", cacheSize)
 	fmt.Fprintf(&buf, "  cleaner=%s\n", o.Cleaner)
 	fmt.Fprintf(&buf, "  comparer=%s\n", o.Comparer.Name)
 	fmt.Fprintf(&buf, "  disable_wal=%t\n", o.DisableWAL)
diff --git a/sstable/options.go b/sstable/options.go
index f9be26f3..a7d79827 100644
--- a/sstable/options.go
+++ b/sstable/options.go
@@ -107,9 +107,6 @@ type ReaderOptions struct {
 }
 
 func (o ReaderOptions) ensureDefaults() ReaderOptions {
-	if o.Cache == nil {
-		o.Cache = cache.New(0)
-	}
 	if o.Comparer == nil {
 		o.Comparer = base.DefaultComparer
 	}
diff --git a/sstable/reader.go b/sstable/reader.go
index 7a0baf2c..5286ba3f 100644
--- a/sstable/reader.go
+++ b/sstable/reader.go
@@ -1028,6 +1028,8 @@ type Reader struct {
 
 // Close implements DB.Close, as documented in the pebble package.
 func (r *Reader) Close() error {
+	r.opts.Cache.Unref()
+
 	if r.err != nil {
 		if r.file != nil {
 			r.file.Close()
@@ -1569,6 +1571,12 @@ func NewReader(f vfs.File, o ReaderOptions, extraOpts ...ReaderOption) (*Reader,
 		file: f,
 		opts: o,
 	}
+	if r.opts.Cache == nil {
+		r.opts.Cache = cache.New(0)
+	} else {
+		r.opts.Cache.Ref()
+	}
+
 	if f == nil {
 		r.err = errors.New("pebble/table: nil file")
 		return nil, r.Close()
diff --git a/table_cache.go b/table_cache.go
index 8e3e5a81..da216a53 100644
--- a/table_cache.go
+++ b/table_cache.go
@@ -30,6 +30,7 @@ var emptyIter = &errorIter{err: nil}
 var tableCacheLabels = pprof.Labels("pebble", "table-cache")
 
 type tableCache struct {
+	cache         *Cache
 	shards        []tableCacheShard
 	filterMetrics FilterMetrics
 }
@@ -37,6 +38,9 @@ type tableCache struct {
 func (c *tableCache) init(
 	cacheID uint64, dirname string, fs vfs.FS, opts *Options, size, hitBuffer int,
 ) {
+	c.cache = opts.Cache
+	c.cache.Ref()
+
 	c.shards = make([]tableCacheShard, runtime.NumCPU())
 	for i := range c.shards {
 		c.shards[i].init(cacheID, dirname, fs, opts, size/len(c.shards), hitBuffer)
@@ -91,13 +95,12 @@ func (c *tableCache) iterCount() int64 {
 }
 
 func (c *tableCache) Close() error {
+	var err error
 	for i := range c.shards {
-		err := c.shards[i].Close()
-		if err != nil {
-			return err
-		}
+		err = firstError(err, c.shards[i].Close())
 	}
-	return nil
+	c.cache.Unref()
+	return err
 }
 
 type tableCacheShard struct {
@@ -169,6 +172,7 @@ func (c *tableCacheShard) newIters(
 		!opts.TableFilter(n.reader.Properties.UserProperties) {
 		// Return the empty iterator. This iterator has no mutable state, so
 		// using a singleton is fine.
+		c.unrefNode(n)
 		return emptyIter, nil, nil
 	}
 	var iter sstable.Iterator
@@ -187,6 +191,7 @@ func (c *tableCacheShard) newIters(
 	// Check for errors during iterator creation.
 	if iter.Error() != nil {
 		// `Close()` returns the same error as `Error()`.
+		c.unrefNode(n)
 		return nil, nil, iter.Close()
 	}
 
@@ -195,6 +200,7 @@ func (c *tableCacheShard) newIters(
 	rangeDelIter, err := n.reader.NewRangeDelIter()
 	if err != nil {
 		iter.Close()
+		c.unrefNode(n)
 		return nil, nil, err
 	}
 	if rangeDelIter != nil {
@@ -393,16 +399,20 @@ func (c *tableCacheShard) Close() error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
+	// Check for leaked iterators. Note that we'll still perform cleanup below in
+	// the case that there are leaked iterators.
+	var err error
 	if v := atomic.LoadInt32(&c.iterCount); v > 0 {
 		if !invariants.RaceEnabled {
-			return fmt.Errorf("leaked iterators: %d", v)
-		}
-		var buf bytes.Buffer
-		fmt.Fprintf(&buf, "leaked iterators: %d\n", v)
-		for _, stack := range c.mu.iters {
-			fmt.Fprintf(&buf, "%s\n", stack)
+			err = fmt.Errorf("leaked iterators: %d", v)
+		} else {
+			var buf bytes.Buffer
+			fmt.Fprintf(&buf, "leaked iterators: %d\n", v)
+			for _, stack := range c.mu.iters {
+				fmt.Fprintf(&buf, "%s\n", stack)
+			}
+			err = errors.New(buf.String())
 		}
-		return errors.New(buf.String())
 	}
 
 	for n := c.mu.lru.next; n != &c.mu.lru; n = n.next {
@@ -420,7 +430,7 @@ func (c *tableCacheShard) Close() error {
 	c.mu.lru.prev = nil
 
 	c.releasing.Wait()
-	return nil
+	return err
 }
 
 type tableCacheNode struct {
diff --git a/tool/tool.go b/tool/tool.go
index 15653f12..3d6ad5f5 100644
--- a/tool/tool.go
+++ b/tool/tool.go
@@ -5,6 +5,8 @@
 package tool
 
 import (
+	"runtime"
+
 	"github.com/cockroachdb/pebble"
 	"github.com/cockroachdb/pebble/bloom"
 	"github.com/cockroachdb/pebble/internal/base"
@@ -42,9 +44,11 @@ type T struct {
 
 // New creates a new introspection tool.
 func New() *T {
+	cache := pebble.NewCache(128 << 20 /* 128 MB */)
+
 	t := &T{
 		opts: pebble.Options{
-			Cache:    pebble.NewCache(128 << 20 /* 128 MB */),
+			Cache:    cache,
 			Filters:  make(map[string]FilterPolicy),
 			FS:       vfs.Default,
 			ReadOnly: true,
@@ -71,6 +75,10 @@ func New() *T {
 		t.sstable.Root,
 		t.wal.Root,
 	}
+
+	runtime.SetFinalizer(t, func(obj interface{}) {
+		cache.Unref()
+	})
 	return t
 }
 


Test patch:
diff --git a/checkpoint_test.go b/checkpoint_test.go
index b14a6cce..cc51dde4 100644
--- a/checkpoint_test.go
+++ b/checkpoint_test.go
@@ -12,10 +12,17 @@ import (
 
 	"github.com/cockroachdb/pebble/internal/datadriven"
 	"github.com/cockroachdb/pebble/vfs"
+	"github.com/stretchr/testify/require"
 )
 
 func TestCheckpoint(t *testing.T) {
 	dbs := make(map[string]*DB)
+	defer func() {
+		for _, db := range dbs {
+			require.NoError(t, db.Close())
+		}
+	}()
+
 	var buf syncedBuffer
 	mem := vfs.NewMem()
 	opts := &Options{
diff --git a/cleaner_test.go b/cleaner_test.go
index b41782b9..dcfca57c 100644
--- a/cleaner_test.go
+++ b/cleaner_test.go
@@ -12,10 +12,17 @@ import (
 
 	"github.com/cockroachdb/pebble/internal/datadriven"
 	"github.com/cockroachdb/pebble/vfs"
+	"github.com/stretchr/testify/require"
 )
 
 func TestArchiveCleaner(t *testing.T) {
 	dbs := make(map[string]*DB)
+	defer func() {
+		for _, db := range dbs {
+			require.NoError(t, db.Close())
+		}
+	}()
+
 	var buf syncedBuffer
 	mem := vfs.NewMem()
 	opts := &Options{
diff --git a/compaction_test.go b/compaction_test.go
index 7e3715f6..2fcfa4e6 100644
--- a/compaction_test.go
+++ b/compaction_test.go
@@ -9,7 +9,6 @@ import (
 	"fmt"
 	"math"
 	"regexp"
-	"runtime"
 	"sort"
 	"strconv"
 	"strings"
@@ -21,6 +20,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/manifest"
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/vfs"
+	"github.com/stretchr/testify/require"
 )
 
 type compactionPickerForTesting struct {
@@ -773,8 +773,15 @@ func TestCompaction(t *testing.T) {
 func TestManualCompaction(t *testing.T) {
 	var mem vfs.FS
 	var d *DB
+	defer func() {
+		require.NoError(t, d.Close())
+	}()
 
 	reset := func() {
+		if d != nil {
+			require.NoError(t, d.Close())
+		}
+
 		mem = vfs.NewMem()
 		err := mem.MkdirAll("ext", 0755)
 		if err != nil {
@@ -819,6 +826,12 @@ func TestManualCompaction(t *testing.T) {
 			return runLSMCmd(td, d)
 
 		case "define":
+			if d != nil {
+				if err := d.Close(); err != nil {
+					return err.Error()
+				}
+			}
+
 			var err error
 			if d, err = runDBDefineCmd(td, nil /* options */); err != nil {
 				return err.Error()
@@ -1341,6 +1354,11 @@ func TestCompactionInuseKeyRanges(t *testing.T) {
 
 func TestCompactionAllowZeroSeqNum(t *testing.T) {
 	var d *DB
+	defer func() {
+		if d != nil {
+			require.NoError(t, d.Close())
+		}
+	}()
 
 	metaRE := regexp.MustCompile(`^L([0-9]+):([^-]+)-(.+)$`)
 	parseMeta := func(s string) (level int, meta fileMetadata) {
@@ -1363,6 +1381,12 @@ func TestCompactionAllowZeroSeqNum(t *testing.T) {
 		func(td *datadriven.TestData) string {
 			switch td.Cmd {
 			case "define":
+				if d != nil {
+					if err := d.Close(); err != nil {
+						return err.Error()
+					}
+				}
+
 				var err error
 				if d, err = runDBDefineCmd(td, nil /* options */); err != nil {
 					return err.Error()
@@ -1525,15 +1549,17 @@ func TestFlushInvariant(t *testing.T) {
 		t.Run(fmt.Sprintf("disableWAL=%t", disableWAL), func(t *testing.T) {
 			for i := 0; i < 2; i++ {
 				t.Run("", func(t *testing.T) {
-					errCh := make(chan error)
+					errCh := make(chan error, 1)
 					defer close(errCh)
 					d, err := Open("", &Options{
 						DisableWAL: disableWAL,
 						FS:         vfs.NewMem(),
 						EventListener: EventListener{
 							BackgroundError: func(err error) {
-								errCh <- err
-								runtime.Goexit() // ensure we don't try to reschedule the flush
+								select {
+								case errCh <- err:
+								default:
+								}
 							},
 						},
 						DebugCheck: true,
@@ -1574,6 +1600,8 @@ func TestFlushInvariant(t *testing.T) {
 							t.Fatalf("expected error but found success")
 						}
 					}
+
+					require.NoError(t, d.Close())
 				})
 			}
 		})
diff --git a/db_test.go b/db_test.go
index 8ca894ed..1bf9b9d5 100644
--- a/db_test.go
+++ b/db_test.go
@@ -17,7 +17,6 @@ import (
 	"time"
 
 	"github.com/cockroachdb/pebble/internal/base"
-	"github.com/cockroachdb/pebble/internal/cache"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/exp/rand"
@@ -705,8 +704,11 @@ func TestIterLeak(t *testing.T) {
 }
 
 func TestMemTableReservation(t *testing.T) {
+	cache := NewCache(128 << 10 /* 128 KB */)
+	defer cache.Unref()
+
 	opts := &Options{
-		Cache:        cache.New(128 << 10 /* 128 KB */),
+		Cache:        cache,
 		MemTableSize: initialMemTableSize,
 		FS:           vfs.NewMem(),
 	}
@@ -786,6 +788,8 @@ func TestMemTableReservationLeak(t *testing.T) {
 
 func TestCacheEvict(t *testing.T) {
 	cache := NewCache(10 << 20)
+	defer cache.Unref()
+
 	d, err := Open("", &Options{
 		Cache: cache,
 		FS:    vfs.NewMem(),
@@ -1026,6 +1030,8 @@ func TestDBApplyBatchNilDB(t *testing.T) {
 	if b1.memTableSize != b2.memTableSize {
 		t.Fatalf("expected memTableSize %d, but found %d", b1.memTableSize, b2.memTableSize)
 	}
+
+	require.NoError(t, d.Close())
 }
 
 func TestDBApplyBatchMismatch(t *testing.T) {
@@ -1052,6 +1058,9 @@ func TestDBApplyBatchMismatch(t *testing.T) {
 	if err == nil || !strings.Contains(err.Error(), "pebble: batch db mismatch:") {
 		t.Fatalf("expected error, but found %v", err)
 	}
+
+	require.NoError(t, srcDB.Close())
+	require.NoError(t, applyDB.Close())
 }
 
 func TestCloseCleanerRace(t *testing.T) {
diff --git a/error_test.go b/error_test.go
index bd68cab8..0d44c2a1 100644
--- a/error_test.go
+++ b/error_test.go
@@ -15,6 +15,7 @@ import (
 
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/vfs"
+	"github.com/stretchr/testify/require"
 )
 
 type panicLogger struct{}
@@ -335,34 +336,23 @@ func TestRequireReadError(t *testing.T) {
 			FS:     fs,
 			Logger: panicLogger{},
 		})
-		if err != nil {
-			t.Fatalf("%v", err)
-		}
+		require.NoError(t, err)
+		defer func() {
+			if d != nil {
+				require.NoError(t, d.Close())
+			}
+		}()
 
 		key1 := []byte("a1")
 		key2 := []byte("a2")
 		value := []byte("b")
-		if err := d.Set(key1, value, nil); err != nil {
-			t.Fatalf("%v", err)
-		}
-		if err := d.Set(key2, value, nil); err != nil {
-			t.Fatalf("%v", err)
-		}
-		if err := d.Flush(); err != nil {
-			t.Fatalf("%v", err)
-		}
-		if err := d.Compact(key1, key2); err != nil {
-			t.Fatalf("%v", err)
-		}
-		if err := d.DeleteRange(key1, key2, nil); err != nil {
-			t.Fatalf("%v", err)
-		}
-		if err := d.Set(key1, value, nil); err != nil {
-			t.Fatalf("%v", err)
-		}
-		if err := d.Flush(); err != nil {
-			t.Fatalf("%v", err)
-		}
+		require.NoError(t, d.Set(key1, value, nil))
+		require.NoError(t, d.Set(key2, value, nil))
+		require.NoError(t, d.Flush())
+		require.NoError(t, d.Compact(key1, key2))
+		require.NoError(t, d.DeleteRange(key1, key2, nil))
+		require.NoError(t, d.Set(key1, value, nil))
+		require.NoError(t, d.Flush())
 		expectLSM(`
 0:
   000007:[a1#3,SET-a2#72057594037927935,RANGEDEL]
@@ -390,6 +380,7 @@ func TestRequireReadError(t *testing.T) {
 		if err := d.Close(); err != nil {
 			return err
 		}
+		d = nil
 		// Reaching here implies all read operations succeeded. This
 		// should only happen when we reached a large enough index at
 		// which `errorFS` did not return any error.
@@ -428,31 +419,22 @@ func TestCorruptReadError(t *testing.T) {
 		if err != nil {
 			t.Fatalf("%v", err)
 		}
+		defer func() {
+			if d != nil {
+				require.NoError(t, d.Close())
+			}
+		}()
 
 		key1 := []byte("a1")
 		key2 := []byte("a2")
 		value := []byte("b")
-		if err := d.Set(key1, value, nil); err != nil {
-			t.Fatalf("%v", err)
-		}
-		if err := d.Set(key2, value, nil); err != nil {
-			t.Fatalf("%v", err)
-		}
-		if err := d.Flush(); err != nil {
-			t.Fatalf("%v", err)
-		}
-		if err := d.Compact(key1, key2); err != nil {
-			t.Fatalf("%v", err)
-		}
-		if err := d.DeleteRange(key1, key2, nil); err != nil {
-			t.Fatalf("%v", err)
-		}
-		if err := d.Set(key1, value, nil); err != nil {
-			t.Fatalf("%v", err)
-		}
-		if err := d.Flush(); err != nil {
-			t.Fatalf("%v", err)
-		}
+		require.NoError(t, d.Set(key1, value, nil))
+		require.NoError(t, d.Set(key2, value, nil))
+		require.NoError(t, d.Flush())
+		require.NoError(t, d.Compact(key1, key2))
+		require.NoError(t, d.DeleteRange(key1, key2, nil))
+		require.NoError(t, d.Set(key1, value, nil))
+		require.NoError(t, d.Flush())
 		expectLSM(`
 0:
   000007:[a1#3,SET-a2#72057594037927935,RANGEDEL]
@@ -480,6 +462,7 @@ func TestCorruptReadError(t *testing.T) {
 		if err := d.Close(); err != nil {
 			return err
 		}
+		d = nil
 		// Reaching here implies all read operations succeeded. This
 		// should only happen when we reached a large enough index at
 		// which `corruptFS` did not inject any corruption.
diff --git a/flush_test.go b/flush_test.go
index e6bd427b..78e9c9b6 100644
--- a/flush_test.go
+++ b/flush_test.go
@@ -12,15 +12,17 @@ import (
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/datadriven"
 	"github.com/cockroachdb/pebble/vfs"
+	"github.com/stretchr/testify/require"
 )
 
 func TestManualFlush(t *testing.T) {
 	d, err := Open("", &Options{
 		FS: vfs.NewMem(),
 	})
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, d.Close())
+	}()
 
 	datadriven.RunTest(t, "testdata/manual_flush", func(td *datadriven.TestData) string {
 		switch td.Cmd {
@@ -69,6 +71,9 @@ func TestManualFlush(t *testing.T) {
 			return s
 
 		case "reset":
+			if err := d.Close(); err != nil {
+				return err.Error()
+			}
 			d, err = Open("", &Options{
 				FS: vfs.NewMem(),
 			})
diff --git a/ingest_test.go b/ingest_test.go
index f64a68df..46a3fbf5 100644
--- a/ingest_test.go
+++ b/ingest_test.go
@@ -21,6 +21,7 @@ import (
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/kr/pretty"
+	"github.com/stretchr/testify/require"
 	"golang.org/x/exp/rand"
 )
 
@@ -426,6 +427,13 @@ func TestIngestMemtableOverlaps(t *testing.T) {
 
 func TestIngestTargetLevel(t *testing.T) {
 	var d *DB
+	defer func() {
+		if d != nil {
+			// Ignore errors because this test defines fake in-progress transactions
+			// that prohibit clean shutdown.
+			_ = d.Close()
+		}
+	}()
 
 	parseMeta := func(s string) fileMetadata {
 		parts := strings.Split(s, "-")
@@ -441,6 +449,12 @@ func TestIngestTargetLevel(t *testing.T) {
 	datadriven.RunTest(t, "testdata/ingest_target_level", func(td *datadriven.TestData) string {
 		switch td.Cmd {
 		case "define":
+			if d != nil {
+				// Ignore errors because this test defines fake in-progress
+				// transactions that prohibit clean shutdown.
+				_ = d.Close()
+			}
+
 			var err error
 			if d, err = runDBDefineCmd(td, nil); err != nil {
 				return err.Error()
@@ -471,7 +485,8 @@ func TestIngestTargetLevel(t *testing.T) {
 			var buf bytes.Buffer
 			for _, target := range strings.Split(td.Input, "\n") {
 				meta := parseMeta(target)
-				level, err := ingestTargetLevel(d.newIters, IterOptions{logger: d.opts.Logger}, d.cmp, d.mu.versions.currentVersion(), 1, d.mu.compact.inProgress, &meta)
+				level, err := ingestTargetLevel(d.newIters, IterOptions{logger: d.opts.Logger},
+					d.cmp, d.mu.versions.currentVersion(), 1, d.mu.compact.inProgress, &meta)
 				if err != nil {
 					return err.Error()
 				}
@@ -488,8 +503,15 @@ func TestIngestTargetLevel(t *testing.T) {
 func TestIngest(t *testing.T) {
 	var mem vfs.FS
 	var d *DB
+	defer func() {
+		require.NoError(t, d.Close())
+	}()
 
 	reset := func() {
+		if d != nil {
+			require.NoError(t, d.Close())
+		}
+
 		mem = vfs.NewMem()
 		err := mem.MkdirAll("ext", 0755)
 		if err != nil {
@@ -603,6 +625,8 @@ func TestIngestCompact(t *testing.T) {
 			t.Fatal(err)
 		}
 	}
+
+	require.NoError(t, d.Close())
 }
 
 func TestConcurrentIngest(t *testing.T) {
@@ -662,6 +686,8 @@ func TestConcurrentIngest(t *testing.T) {
 			t.Fatal(err)
 		}
 	}
+
+	require.NoError(t, d.Close())
 }
 
 func TestConcurrentIngestCompact(t *testing.T) {
@@ -786,6 +812,8 @@ func TestConcurrentIngestCompact(t *testing.T) {
 				// if there is insufficient synchronization between ingestion and
 				// compaction.
 			}
+
+			require.NoError(t, d.Close())
 		})
 	}
 }
@@ -839,6 +867,8 @@ func TestIngestFlushQueuedMemTable(t *testing.T) {
 	}
 
 	ingest("a")
+
+	require.NoError(t, d.Close())
 }
 
 func TestIngestFlushQueuedLargeBatch(t *testing.T) {
@@ -888,4 +918,6 @@ func TestIngestFlushQueuedLargeBatch(t *testing.T) {
 	}
 
 	ingest("a")
+
+	require.NoError(t, d.Close())
 }
diff --git a/internal/cache/clockpro_test.go b/internal/cache/clockpro_test.go
index bf6bc430..c3b310fd 100644
--- a/internal/cache/clockpro_test.go
+++ b/internal/cache/clockpro_test.go
@@ -22,6 +22,8 @@ func TestCache(t *testing.T) {
 	}
 
 	cache := newShards(200, 1)
+	defer cache.Unref()
+
 	scanner := bufio.NewScanner(f)
 	line := 1
 
@@ -67,6 +69,8 @@ func testAutoValue(cache *Cache, s string, repeat int) *Value {
 
 func TestWeakHandle(t *testing.T) {
 	cache := newShards(5, 1)
+	defer cache.Unref()
+
 	cache.Set(1, 1, 0, testAutoValue(cache, "a", 5)).Release()
 	h := cache.Set(1, 0, 0, testAutoValue(cache, "b", 5))
 	if v := h.Get(); string(v) != "bbbbb" {
@@ -85,6 +89,8 @@ func TestWeakHandle(t *testing.T) {
 
 func TestCacheDelete(t *testing.T) {
 	cache := newShards(100, 1)
+	defer cache.Unref()
+
 	cache.Set(1, 0, 0, testManualValue(cache, "a", 5)).Release()
 	cache.Set(1, 1, 0, testManualValue(cache, "a", 5)).Release()
 	cache.Set(1, 2, 0, testManualValue(cache, "a", 5)).Release()
@@ -114,6 +120,8 @@ func TestCacheDelete(t *testing.T) {
 
 func TestEvictFile(t *testing.T) {
 	cache := newShards(100, 1)
+	defer cache.Unref()
+
 	cache.Set(1, 0, 0, testManualValue(cache, "a", 5)).Release()
 	cache.Set(1, 1, 0, testManualValue(cache, "a", 5)).Release()
 	cache.Set(1, 2, 0, testManualValue(cache, "a", 5)).Release()
@@ -140,12 +148,16 @@ func TestEvictAll(t *testing.T) {
 	// Verify that it is okay to evict all of the data from a cache. Previously
 	// this would trigger a nil-pointer dereference.
 	cache := newShards(100, 1)
+	defer cache.Unref()
+
 	cache.Set(1, 0, 0, testManualValue(cache, "a", 101)).Release()
 	cache.Set(1, 1, 0, testManualValue(cache, "a", 101)).Release()
 }
 
 func TestMultipleDBs(t *testing.T) {
 	cache := newShards(100, 1)
+	defer cache.Unref()
+
 	cache.Set(1, 0, 0, testManualValue(cache, "a", 5)).Release()
 	cache.Set(2, 0, 0, testManualValue(cache, "b", 5)).Release()
 	if expected, size := int64(10), cache.Size(); expected != size {
@@ -162,16 +174,22 @@ func TestMultipleDBs(t *testing.T) {
 	h = cache.Get(2, 0, 0)
 	if v := h.Get(); string(v) != "bbbbb" {
 		t.Fatalf("expected bbbbb, but found %s", v)
+	} else {
+		h.Release()
 	}
 }
 
 func TestZeroSize(t *testing.T) {
 	cache := newShards(0, 1)
+	defer cache.Unref()
+
 	cache.Set(1, 0, 0, testManualValue(cache, "a", 5)).Release()
 }
 
 func TestReserve(t *testing.T) {
 	cache := newShards(4, 2)
+	defer cache.Unref()
+
 	cache.Set(1, 0, 0, testManualValue(cache, "a", 1)).Release()
 	cache.Set(2, 0, 0, testManualValue(cache, "a", 1)).Release()
 	require.EqualValues(t, 2, cache.Size())
@@ -191,6 +209,8 @@ func TestReserve(t *testing.T) {
 
 func TestReserveDoubleRelease(t *testing.T) {
 	cache := newShards(100, 1)
+	defer cache.Unref()
+
 	r := cache.Reserve(10)
 	r()
 
diff --git a/internal/cache/robin_hood_test.go b/internal/cache/robin_hood_test.go
index af952530..1437c56f 100644
--- a/internal/cache/robin_hood_test.go
+++ b/internal/cache/robin_hood_test.go
@@ -17,6 +17,8 @@ import (
 func TestRobinHoodMap(t *testing.T) {
 	rng := rand.New(rand.NewSource(uint64(time.Now().UnixNano())))
 	rhMap := newRobinHoodMap(0)
+	defer rhMap.free()
+
 	goMap := make(map[key]*entry)
 
 	randomKey := func() key {
diff --git a/internal/metamorphic/meta_test.go b/internal/metamorphic/meta_test.go
index 86185179..b0463a6d 100644
--- a/internal/metamorphic/meta_test.go
+++ b/internal/metamorphic/meta_test.go
@@ -187,6 +187,10 @@ func TestMeta(t *testing.T) {
 	// written to <run-dir>/OPTIONS and a child process is created to actually
 	// execute the test.
 	runOptions := func(t *testing.T, opts *testOptions) {
+		if opts.opts.Cache != nil {
+			defer opts.opts.Cache.Unref()
+		}
+
 		runDir := filepath.Join(metaDir, path.Base(t.Name()))
 		if err := os.MkdirAll(runDir, 0755); err != nil {
 			t.Fatal(err)
diff --git a/internal/metamorphic/test.go b/internal/metamorphic/test.go
index d813cf79..05b756fb 100644
--- a/internal/metamorphic/test.go
+++ b/internal/metamorphic/test.go
@@ -51,6 +51,8 @@ func (t *test) init(h *history, dir string, testOpts *testOptions) error {
 	t.opts.EventListener = pebble.MakeLoggingEventListener(t.opts.Logger)
 	t.opts.DebugCheck = true
 
+	defer t.opts.Cache.Unref()
+
 	// If an error occurs and we were using an in-memory FS, attempt to clone to
 	// on-disk in order to allow post-mortem debugging. Note that always using
 	// the on-disk FS isn't desirable because there is a large performance
@@ -150,6 +152,7 @@ func (t *test) restartDB() error {
 	if !t.testOpts.strictFS {
 		return nil
 	}
+	t.opts.Cache.Ref()
 	fs := t.opts.FS.(*vfs.MemFS)
 	fs.SetIgnoreSyncs(true)
 	if err := t.db.Close(); err != nil {
@@ -159,6 +162,7 @@ func (t *test) restartDB() error {
 	fs.SetIgnoreSyncs(false)
 	var err error
 	t.db, err = pebble.Open(t.dir, t.opts)
+	t.opts.Cache.Unref()
 	return err
 }
 
diff --git a/iterator_test.go b/iterator_test.go
index 0814121a..47c45a31 100644
--- a/iterator_test.go
+++ b/iterator_test.go
@@ -16,6 +16,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/datadriven"
 	"github.com/cockroachdb/pebble/vfs"
+	"github.com/stretchr/testify/require"
 	"golang.org/x/exp/rand"
 )
 
@@ -410,10 +411,21 @@ func (c *minSeqNumPropertyCollector) Name() string {
 
 func TestIteratorTableFilter(t *testing.T) {
 	var d *DB
+	defer func() {
+		if d != nil {
+			require.NoError(t, d.Close())
+		}
+	}()
 
 	datadriven.RunTest(t, "testdata/iterator_table_filter", func(td *datadriven.TestData) string {
 		switch td.Cmd {
 		case "define":
+			if d != nil {
+				if err := d.Close(); err != nil {
+					return err.Error()
+				}
+			}
+
 			opts := &Options{}
 			opts.TablePropertyCollectors = append(opts.TablePropertyCollectors,
 				func() TablePropertyCollector {
@@ -478,8 +490,15 @@ func TestIteratorTableFilter(t *testing.T) {
 func TestIteratorNextPrev(t *testing.T) {
 	var mem vfs.FS
 	var d *DB
+	defer func() {
+		require.NoError(t, d.Close())
+	}()
 
 	reset := func() {
+		if d != nil {
+			require.NoError(t, d.Close())
+		}
+
 		mem = vfs.NewMem()
 		err := mem.MkdirAll("ext", 0755)
 		if err != nil {
diff --git a/level_checker_test.go b/level_checker_test.go
index a71318fa..f5d32187 100644
--- a/level_checker_test.go
+++ b/level_checker_test.go
@@ -45,8 +45,15 @@ func TestCheckLevelsCornerCases(t *testing.T) {
 	memFS := vfs.NewMem()
 	cmp := DefaultComparer.Compare
 	var levels [][]fileMetadata
+
 	// Indexed by fileNum
 	var readers []*sstable.Reader
+	defer func() {
+		for _, r := range readers {
+			r.Close()
+		}
+	}()
+
 	var fileNum uint64
 	newIters :=
 		func(meta *fileMetadata, opts *IterOptions, bytesIterated *uint64) (internalIterator, internalIterator, error) {
diff --git a/level_iter_test.go b/level_iter_test.go
index 5211a2cf..f2da80ef 100644
--- a/level_iter_test.go
+++ b/level_iter_test.go
@@ -154,6 +154,9 @@ func (lt *levelIterTest) newIters(
 
 func (lt *levelIterTest) runClear(d *datadriven.TestData) string {
 	lt.mem = vfs.NewMem()
+	for _, r := range lt.readers {
+		r.Close()
+	}
 	lt.readers = nil
 	lt.files = nil
 	return ""
@@ -234,6 +237,8 @@ func (lt *levelIterTest) runBuild(d *datadriven.TestData) string {
 
 func TestLevelIterBoundaries(t *testing.T) {
 	lt := newLevelIterTest()
+	defer lt.runClear(nil)
+
 	datadriven.RunTest(t, "testdata/level_iter_boundaries", func(d *datadriven.TestData) string {
 		switch d.Cmd {
 		case "clear":
@@ -301,6 +306,8 @@ func (i *levelIterTestIter) SeekLT(key []byte) (*InternalKey, []byte) {
 
 func TestLevelIterSeek(t *testing.T) {
 	lt := newLevelIterTest()
+	defer lt.runClear(nil)
+
 	datadriven.RunTest(t, "testdata/level_iter_seek", func(d *datadriven.TestData) string {
 		switch d.Cmd {
 		case "clear":
@@ -362,6 +369,8 @@ func buildLevelIterTables(
 	}
 
 	opts := sstable.ReaderOptions{Cache: NewCache(128 << 20)}
+	defer opts.Cache.Unref()
+
 	readers := make([]*sstable.Reader, len(files))
 	for i := range files {
 		f, err := mem.Open(fmt.Sprintf("bench%d", i))
diff --git a/merging_iter_test.go b/merging_iter_test.go
index a4dd21d1..375c35ba 100644
--- a/merging_iter_test.go
+++ b/merging_iter_test.go
@@ -128,8 +128,15 @@ func TestMergingIterCornerCases(t *testing.T) {
 	memFS := vfs.NewMem()
 	cmp := DefaultComparer.Compare
 	var v version
+
 	// Indexed by fileNum.
 	var readers []*sstable.Reader
+	defer func() {
+		for _, r := range readers {
+			r.Close()
+		}
+	}()
+
 	var fileNum uint64
 	newIters :=
 		func(meta *fileMetadata, opts *IterOptions, bytesIterated *uint64) (internalIterator, internalIterator, error) {
@@ -140,6 +147,7 @@ func TestMergingIterCornerCases(t *testing.T) {
 			}
 			return r.NewIter(opts.GetLowerBound(), opts.GetUpperBound()), rangeDelIter, nil
 		}
+
 	datadriven.RunTest(t, "testdata/merging_iter", func(d *datadriven.TestData) string {
 		switch d.Cmd {
 		case "define":
@@ -290,6 +298,8 @@ func buildMergingIterTables(
 	}
 
 	opts := sstable.ReaderOptions{Cache: NewCache(128 << 20)}
+	defer opts.Cache.Unref()
+
 	readers := make([]*sstable.Reader, len(files))
 	for i := range files {
 		f, err := mem.Open(fmt.Sprintf("bench%d", i))
diff --git a/metrics_test.go b/metrics_test.go
index 8ae43a10..e4beb3e6 100644
--- a/metrics_test.go
+++ b/metrics_test.go
@@ -11,6 +11,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/datadriven"
 	"github.com/cockroachdb/pebble/vfs"
+	"github.com/stretchr/testify/require"
 )
 
 func TestMetricsFormat(t *testing.T) {
@@ -88,10 +89,17 @@ func TestMetrics(t *testing.T) {
 	d, err := Open("", &Options{
 		FS: vfs.NewMem(),
 	})
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, d.Close())
+	}()
+
 	iters := make(map[string]*Iterator)
+	defer func() {
+		for _, i := range iters {
+			require.NoError(t, i.Close())
+		}
+	}()
 
 	datadriven.RunTest(t, "testdata/metrics", func(td *datadriven.TestData) string {
 		switch td.Cmd {
diff --git a/open_test.go b/open_test.go
index 47eed41b..30f2e098 100644
--- a/open_test.go
+++ b/open_test.go
@@ -173,6 +173,7 @@ func testOpenCloseOpenClose(t *testing.T, fs vfs.FS, root string) {
 							startFromEmpty, length, err)
 						continue
 					}
+					got = append([]byte(nil), got...)
 				}
 				err = d1.Close()
 				if err != nil {
diff --git a/options_test.go b/options_test.go
index ad306289..24b30b20 100644
--- a/options_test.go
+++ b/options_test.go
@@ -190,6 +190,9 @@ func TestOptionsParse(t *testing.T) {
 			if str != parsedStr {
 				t.Fatalf("expected\n%s\nbut found\n%s", str, parsedStr)
 			}
+			if parsedOptions.Cache != nil {
+				parsedOptions.Cache.Unref()
+			}
 		})
 	}
 }
diff --git a/range_del_test.go b/range_del_test.go
index a7cd0854..4842376e 100644
--- a/range_del_test.go
+++ b/range_del_test.go
@@ -15,14 +15,26 @@ import (
 	"github.com/cockroachdb/pebble/internal/datadriven"
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/vfs"
+	"github.com/stretchr/testify/require"
 )
 
 func TestRangeDel(t *testing.T) {
 	var d *DB
+	defer func() {
+		if d != nil {
+			require.NoError(t, d.Close())
+		}
+	}()
 
 	datadriven.RunTest(t, "testdata/range_del", func(td *datadriven.TestData) string {
 		switch td.Cmd {
 		case "define":
+			if d != nil {
+				if err := d.Close(); err != nil {
+					return err.Error()
+				}
+			}
+
 			var err error
 			if d, err = runDBDefineCmd(td, nil /* options */); err != nil {
 				return err.Error()
@@ -437,8 +449,11 @@ func BenchmarkRangeDelIterate(b *testing.B) {
 			for _, deleted := range []int{entries, entries - 1} {
 				b.Run(fmt.Sprintf("deleted=%d", deleted), func(b *testing.B) {
 					mem := vfs.NewMem()
+					cache := NewCache(128 << 20) // 128 MB
+					defer cache.Unref()
+
 					d, err := Open("", &Options{
-						Cache:      NewCache(128 << 20), // 128 MB
+						Cache:      cache,
 						FS:         mem,
 						DebugCheck: true,
 					})
diff --git a/snapshot_test.go b/snapshot_test.go
index f4b0e7bd..be527710 100644
--- a/snapshot_test.go
+++ b/snapshot_test.go
@@ -44,9 +44,23 @@ func TestSnapshot(t *testing.T) {
 	var d *DB
 	var snapshots map[string]*Snapshot
 
+	close := func() {
+		for _, s := range snapshots {
+			require.NoError(t, s.Close())
+		}
+		snapshots = nil
+		if d != nil {
+			require.NoError(t, d.Close())
+			d = nil
+		}
+	}
+	defer close()
+
 	datadriven.RunTest(t, "testdata/snapshot", func(td *datadriven.TestData) string {
 		switch td.Cmd {
 		case "define":
+			close()
+
 			var err error
 			d, err = Open("", &Options{
 				FS: vfs.NewMem(),
@@ -190,4 +204,6 @@ func TestSnapshotClosed(t *testing.T) {
 	require.EqualValues(t, ErrClosed, catch(func() { _ = snap.Close() }))
 	require.EqualValues(t, ErrClosed, catch(func() { _, _ = snap.Get(nil) }))
 	require.EqualValues(t, ErrClosed, catch(func() { snap.NewIter(nil) }))
+
+	require.NoError(t, d.Close())
 }
diff --git a/sstable/properties_test.go b/sstable/properties_test.go
index efc264d1..aa8c6ad9 100644
--- a/sstable/properties_test.go
+++ b/sstable/properties_test.go
@@ -79,11 +79,11 @@ func TestPropertiesLoad(t *testing.T) {
 		if err != nil {
 			t.Fatal(err)
 		}
-		defer f.Close()
 		r, err := NewReader(f, ReaderOptions{})
 		if err != nil {
 			t.Fatal(err)
 		}
+		defer r.Close()
 
 		if diff := pretty.Diff(expected, r.Properties); diff != nil {
 			t.Fatalf("%s", strings.Join(diff, "\n"))
diff --git a/sstable/reader_test.go b/sstable/reader_test.go
index 99cfdee2..16690146 100644
--- a/sstable/reader_test.go
+++ b/sstable/reader_test.go
@@ -228,6 +228,11 @@ func runTestReader(t *testing.T, o WriterOptions, dir string, r *Reader) {
 				return fmt.Sprintf("unknown command: %s", d.Cmd)
 			}
 		})
+
+		if r != nil {
+			r.Close()
+			r = nil
+		}
 	})
 }
 
@@ -435,8 +440,10 @@ func buildTestTable(
 	if err != nil {
 		t.Fatal(err)
 	}
+	c := cache.New(128 << 20)
+	defer c.Unref()
 	r, err := NewReader(f1, ReaderOptions{
-		Cache: cache.New(128 << 20),
+		Cache: c,
 	})
 	if err != nil {
 		t.Fatal(err)
@@ -476,8 +483,10 @@ func buildBenchmarkTable(b *testing.B, blockSize, restartInterval int) (*Reader,
 	if err != nil {
 		b.Fatal(err)
 	}
+	c := cache.New(128 << 20)
+	defer c.Unref()
 	r, err := NewReader(f1, ReaderOptions{
-		Cache: cache.New(128 << 20),
+		Cache: c,
 	})
 	if err != nil {
 		b.Fatal(err)
diff --git a/sstable/table_test.go b/sstable/table_test.go
index c0fa57e0..d6c0716e 100644
--- a/sstable/table_test.go
+++ b/sstable/table_test.go
@@ -492,9 +492,7 @@ func TestBloomFilterFalsePositiveRate(t *testing.T) {
 			c.Name(): c,
 		},
 	})
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
 
 	const n = 10000
 	// key is a buffer that will be re-used for n Get calls, each with a
@@ -528,6 +526,8 @@ func TestBloomFilterFalsePositiveRate(t *testing.T) {
 	if got := float64(100*c.falsePositives) / n; got < 0.2 || 5 < got {
 		t.Errorf("false positive rate: got %v%%, want approximately 1%%", got)
 	}
+
+	require.NoError(t, r.Close())
 }
 
 type countingFilterPolicy struct {
@@ -659,10 +659,7 @@ func TestReaderGlobalSeqNum(t *testing.T) {
 		t.Fatal(err)
 	}
 	r, err := NewReader(f, ReaderOptions{})
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer r.Close()
+	require.NoError(t, err)
 
 	const globalSeqNum = 42
 	r.Properties.GlobalSeqNum = globalSeqNum
@@ -673,9 +670,8 @@ func TestReaderGlobalSeqNum(t *testing.T) {
 			t.Fatalf("expected %d, but found %d", globalSeqNum, i.Key().SeqNum())
 		}
 	}
-	if err := i.Close(); err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, i.Close())
+	require.NoError(t, r.Close())
 }
 
 func TestMetaIndexEntriesSorted(t *testing.T) {
@@ -686,9 +682,7 @@ func TestMetaIndexEntriesSorted(t *testing.T) {
 	}
 
 	r, err := NewReader(f, ReaderOptions{})
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
 
 	b, err := r.readBlock(r.metaIndexBH, nil /* transform */, false /* weak */)
 	if err != nil {
@@ -708,9 +702,8 @@ func TestMetaIndexEntriesSorted(t *testing.T) {
 		t.Fatalf("metaindex block out of order: %v", keys)
 	}
 
-	if err := i.Close(); err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, i.Close())
+	require.NoError(t, r.Close())
 }
 
 func TestFooterRoundTrip(t *testing.T) {
diff --git a/table_cache_test.go b/table_cache_test.go
index 7ed5b9aa..db47c8e5 100644
--- a/table_cache_test.go
+++ b/table_cache_test.go
@@ -162,8 +162,12 @@ func newTableCache() (*tableCache, *tableCacheTestFS, error) {
 	fs.closeCounts = map[string]int{}
 	fs.mu.Unlock()
 
-	opts := &Options{}
+	opts := &Options{
+		Cache: NewCache(8 << 20), // 8 MB
+	}
 	opts.EnsureDefaults()
+	defer opts.Cache.Unref()
+
 	c := &tableCache{}
 	c.init(opts.Cache.NewID(), "", fs, opts, tableCacheTestCacheSize, tableCacheTestHitBufferSize)
 	return c, fs, nil
diff --git a/version_set_test.go b/version_set_test.go
index 0951cb21..ecc1c113 100644
--- a/version_set_test.go
+++ b/version_set_test.go
@@ -48,4 +48,5 @@ func TestVersionSetCheckpoint(t *testing.T) {
 	}
 	checkValue("a", "b")
 	checkValue("c", "d")
+	require.NoError(t, d.Close())
 }