db: fix leak for compaction bytes-in-progress metric

We were counting flushes for the bytes-in-progress metric (which is
not the intention of the metric). When replaying the WAL, we perform a
flush but do not clean up the metric, resulting in an one-time leak.
The result is that we always see the same non-zero value instead of
zero.

The fix is to not count at all during flushes. As a test, we run an
invariants-only check on the metrics on close; we could potentially
check other things here as well. This new assertion fails without this
fix.

Fixes #2548

compaction.go

@@ -2071,7 +2071,6 @@ func (d *DB) flush1() (bytesFlushed uint64, err error) {
 	d.clearCompactingState(c, err != nil)
 	delete(d.mu.compact.inProgress, c)
 	d.mu.versions.incrementCompactions(c.kind, c.extraLevels)
-	d.mu.versions.incrementCompactionBytes(-c.bytesWritten)
 
 	var flushed flushableList
 	if err == nil {
@@ -2850,10 +2849,12 @@ func (d *DB) runCompaction(
 			Path:    d.objProvider.Path(objMeta),
 			FileNum: fileNum,
 		})
-		writable = &compactionWritable{
-			Writable: writable,
-			versions: d.mu.versions,
-			written:  &c.bytesWritten,
+		if c.kind != compactionKindFlush {
+			writable = &compactionWritable{
+				Writable: writable,
+				versions: d.mu.versions,
+				written:  &c.bytesWritten,
+			}
 		}
 		createdFiles = append(createdFiles, fileNum.DiskFileNum())
 		cacheOpts := private.SSTableCacheOpts(d.cacheID, fileNum.DiskFileNum()).(sstable.WriterOption)

db.go

@@ -1523,6 +1523,16 @@ func (d *DB) Close() error {
 	d.mu.Unlock()
 	d.deleters.Wait()
 	d.compactionSchedulers.Wait()
+
+	// Sanity check metrics.
+	if invariants.Enabled {
+		m := d.Metrics()
+		if m.Compact.NumInProgress > 0 || m.Compact.InProgressBytes > 0 {
+			d.mu.Lock()
+			panic(fmt.Sprintf("invalid metrics on close:\n%s", m))
+		}
+	}
+
 	d.mu.Lock()
 
 	// As a sanity check, ensure that there are no zombie tables. A non-zero count