db: refactor obsolete, zombie file tracking

Add a zombieObjects type for tracking the lifecycle of zombie objects. Today
this is used for zombie sstable objects. With the introduction of blob files
(#112) we'll also need to track the lifecycle of zombie blob file objects.

Additionally, refactor the lifecycle so that a file is considered either zombie
or obsolete but not both. And avoid scanning the list of zombie objects on
every call to Metrics, instead incrementally maintaining stats on the
zombieObjects type.

Author: jbowens

compaction.go

@@ -2406,7 +2406,7 @@ func (d *DB) cleanupVersionEdit(ve *versionEdit) {
 		// Add this file to zombie tables as well, as the versionSet
 		// asserts on whether every obsolete file was at one point
 		// marked zombie.
-		d.mu.versions.zombieTables[of.DiskFileNum] = objectInfo{
+		d.mu.versions.zombieTables.Add(objectInfo{
 			fileInfo: fileInfo{
 				FileNum:  of.DiskFileNum,
 				FileSize: of.Size,
@@ -2417,7 +2417,7 @@ func (d *DB) cleanupVersionEdit(ve *versionEdit) {
 			// disaggregated storage; if this becomes the norm, we should do
 			// an objprovider lookup here.
 			isLocal: true,
-		}
+		})
 	}
 	d.mu.versions.addObsoleteLocked(obsoleteFiles)
 }
@@ -3019,13 +3019,7 @@ func (d *DB) runCompaction(
 			// Add this file to zombie tables as well, as the versionSet
 			// asserts on whether every obsolete file was at one point
 			// marked zombie.
-			d.mu.versions.zombieTables[backing.DiskFileNum] = objectInfo{
-				fileInfo: fileInfo{
-					FileNum:  backing.DiskFileNum,
-					FileSize: backing.Size,
-				},
-				isLocal: true,
-			}
+			d.mu.versions.zombieTables.AddMetadata(&result.Tables[i].ObjMeta, backing.Size)
 		}
 		d.mu.versions.addObsoleteLocked(obsoleteFiles)
 		d.mu.Unlock()

db.go

@@ -1644,6 +1644,7 @@ func (d *DB) Close() error {
 	if err := d.closed.Load(); err != nil {
 		panic(err)
 	}
+	d.compactionSchedulers.Wait()
 	// Compactions can be asynchronously started by the CompactionScheduler
 	// calling d.Schedule. When this Unregister returns, we know that the
 	// CompactionScheduler will never again call a method on the DB. Note that
@@ -1753,7 +1754,6 @@ func (d *DB) Close() error {
 	}
 
 	d.mu.Unlock()
-	d.compactionSchedulers.Wait()
 
 	// Wait for all cleaning jobs to finish.
 	d.cleanupManager.Close()
@@ -1771,7 +1771,7 @@ func (d *DB) Close() error {
 
 	// As a sanity check, ensure that there are no zombie tables. A non-zero count
 	// hints at a reference count leak.
-	if ztbls := len(d.mu.versions.zombieTables); ztbls > 0 {
+	if ztbls := d.mu.versions.zombieTables.Count(); ztbls > 0 {
 		err = firstError(err, errors.Errorf("non-zero zombie file count: %d", ztbls))
 	}
 
@@ -2041,13 +2041,9 @@ func (d *DB) Metrics() *Metrics {
 			metrics.Levels[level].Score = score
 		}
 	}
-	metrics.Table.ZombieCount = int64(len(d.mu.versions.zombieTables))
-	for _, info := range d.mu.versions.zombieTables {
-		metrics.Table.ZombieSize += info.FileSize
-		if info.isLocal {
-			metrics.Table.Local.ZombieSize += info.FileSize
-		}
-	}
+	metrics.Table.ZombieCount = int64(d.mu.versions.zombieTables.Count())
+	metrics.Table.ZombieSize = d.mu.versions.zombieTables.TotalSize()
+	metrics.Table.Local.ZombieSize = d.mu.versions.zombieTables.LocalSize()
 	metrics.private.optionsFileSize = d.optionsFileSize
 
 	// TODO(jackson): Consider making these metrics optional.