db: pull timeNow into Options.private

Pull the timeNow func that permits mocking time into the private section of the
Options struct so that callers may mock time within Open itself.

Author: jbowens

blob_rewrite.go

@@ -61,7 +61,7 @@ func (c *pickedBlobFileCompaction) ConstructCompaction(
 	// when it completes.
 	c.vers.Ref()
 	return &blobFileRewriteCompaction{
-		beganAt:           d.timeNow(),
+		beganAt:           d.opts.private.timeNow(),
 		grantHandle:       grantHandle,
 		version:           c.vers,
 		input:             c.file,
@@ -188,12 +188,12 @@ func (c *blobFileRewriteCompaction) Execute(jobID JobID, d *DB) error {
 		},
 	}
 	d.opts.EventListener.BlobFileRewriteBegin(info)
-	startTime := d.timeNow()
+	startTime := d.opts.private.timeNow()
 
 	// Run the blob file rewrite.
 	objMeta, ve, err := d.runBlobFileRewriteLocked(ctx, jobID, c)
 
-	info.Duration = d.timeNow().Sub(startTime)
+	info.Duration = d.opts.private.timeNow().Sub(startTime)
 
 	// Update the version with the remapped blob file.
 	if err == nil {
@@ -266,7 +266,7 @@ func (c *blobFileRewriteCompaction) Execute(jobID JobID, d *DB) error {
 	}
 
 	// Notify the event listener that the compaction has ended.
-	now := d.timeNow()
+	now := d.opts.private.timeNow()
 	info.TotalDuration = now.Sub(c.beganAt)
 	info.Done = true
 	info.Err = err
@@ -348,7 +348,7 @@ func (d *DB) runBlobFileRewriteLocked(
 		FileNum:      objMeta.DiskFileNum,
 		Size:         stats.FileLen,
 		ValueSize:    stats.UncompressedValueBytes,
-		CreationTime: uint64(d.timeNow().Unix()),
+		CreationTime: uint64(d.opts.private.timeNow().Unix()),
 	}
 	physical.PopulateProperties(&stats.Properties)
 

compaction.go

@@ -1431,7 +1431,7 @@ func (d *DB) maybeScheduleDelayedFlush(tbl *memTable, dur time.Duration) {
 	if mem == nil || mem.flushForced {
 		return
 	}
-	deadline := d.timeNow().Add(dur)
+	deadline := d.opts.private.timeNow().Add(dur)
 	if !mem.delayedFlushForcedAt.IsZero() && deadline.After(mem.delayedFlushForcedAt) {
 		// Already scheduled to flush sooner than within `dur`.
 		return
@@ -1720,7 +1720,7 @@ func (d *DB) flush1() (bytesFlushed uint64, err error) {
 		d.mu.versions.latest.l0Organizer,
 		d.mu.versions.picker.getBaseLevel(),
 		d.mu.mem.queue[:n],
-		d.timeNow(),
+		d.opts.private.timeNow(),
 		d.shouldCreateShared(0),
 		d.determineCompactionValueSeparation,
 	)
@@ -1738,7 +1738,7 @@ func (d *DB) flush1() (bytesFlushed uint64, err error) {
 	}
 	d.opts.EventListener.FlushBegin(info)
 
-	startTime := d.timeNow()
+	startTime := d.opts.private.timeNow()
 
 	var ve *manifest.VersionEdit
 	var stats compact.Stats
@@ -1758,7 +1758,7 @@ func (d *DB) flush1() (bytesFlushed uint64, err error) {
 		if c.kind == compactionKindIngestedFlushable {
 			ve, err = d.runIngestFlush(c)
 		}
-		info.Duration = d.timeNow().Sub(startTime)
+		info.Duration = d.opts.private.timeNow().Sub(startTime)
 		if err != nil {
 			return versionUpdate{}, err
 		}
@@ -1885,7 +1885,7 @@ func (d *DB) flush1() (bytesFlushed uint64, err error) {
 		info.Err = errEmptyTable
 	}
 	info.Done = true
-	info.TotalDuration = d.timeNow().Sub(startTime)
+	info.TotalDuration = d.opts.private.timeNow().Sub(startTime)
 	d.opts.EventListener.FlushEnd(info)
 
 	// The order of these operations matters here for ease of testing.
@@ -2280,7 +2280,7 @@ func (d *DB) tryScheduleDeleteOnlyCompaction() bool {
 	d.mu.compact.deletionHints = unresolvedHints
 
 	if len(inputs) > 0 {
-		c := newDeleteOnlyCompaction(d.opts, v, inputs, d.timeNow(), resolvedHints, exciseEnabled)
+		c := newDeleteOnlyCompaction(d.opts, v, inputs, d.opts.private.timeNow(), resolvedHints, exciseEnabled)
 		d.mu.compact.compactingCount++
 		d.mu.compact.compactProcesses++
 		c.AddInProgressLocked(d)
@@ -2635,7 +2635,7 @@ func (d *DB) compact(c compaction, errChannel chan error) {
 			// must be atomic with the above removal of c from
 			// d.mu.compact.InProgress to ensure Metrics.Compact.Duration does not
 			// miss or double count a completing compaction's duration.
-			d.mu.compact.duration += d.timeNow().Sub(c.BeganAt())
+			d.mu.compact.duration += d.opts.private.timeNow().Sub(c.BeganAt())
 		}()
 		// Done must not be called while holding any lock that needs to be
 		// acquired by Schedule. Also, it must be called after new Version has
@@ -2737,12 +2737,12 @@ func (d *DB) cleanupVersionEdit(ve *manifest.VersionEdit) {
 func (d *DB) compact1(jobID JobID, c *tableCompaction) (err error) {
 	info := c.makeInfo(jobID)
 	d.opts.EventListener.CompactionBegin(info)
-	startTime := d.timeNow()
+	startTime := d.opts.private.timeNow()
 
 	ve, stats, outputBlobs, err := d.runCompaction(jobID, c)
 
 	info.Annotations = append(info.Annotations, c.annotations...)
-	info.Duration = d.timeNow().Sub(startTime)
+	info.Duration = d.opts.private.timeNow().Sub(startTime)
 	if err == nil {
 		validateVersionEdit(ve, d.opts.Comparer.ValidateKey, d.opts.Comparer.FormatKey, d.opts.Logger)
 		_, err = d.mu.versions.UpdateVersionLocked(func() (versionUpdate, error) {
@@ -2799,7 +2799,7 @@ func (d *DB) compact1(jobID JobID, c *tableCompaction) (err error) {
 	d.mu.versions.incrementCompactions(c.kind, c.extraLevels, c.metrics.bytesWritten.Load(), err)
 	d.mu.versions.incrementCompactionBytes(-c.metrics.bytesWritten.Load())
 
-	info.TotalDuration = d.timeNow().Sub(c.metrics.beganAt)
+	info.TotalDuration = d.opts.private.timeNow().Sub(c.metrics.beganAt)
 	d.opts.EventListener.CompactionEnd(info)
 
 	// Update the read state before deleting obsolete files because the

compaction_picker.go

@@ -245,7 +245,7 @@ func (pc *pickedTableCompaction) ConstructCompaction(
 	return newCompaction(
 		pc,
 		d.opts,
-		d.timeNow(),
+		d.opts.private.timeNow(),
 		d.ObjProvider(),
 		grantHandle,
 		d.shouldCreateShared(pc.outputLevel.level),

compaction_test.go

@@ -2890,6 +2890,11 @@ func TestMarkedForCompaction(t *testing.T) {
 				EventListener:               &eventListener,
 				Logger:                      testutils.Logger{T: t},
 			}
+			t := time.Now()
+			opts.private.timeNow = func() time.Time {
+				t = t.Add(time.Second)
+				return t
+			}
 			opts.Experimental.CompactionScheduler = func() CompactionScheduler {
 				return NewConcurrencyLimitSchedulerWithNoPeriodicGrantingForTest()
 			}
@@ -2899,11 +2904,7 @@ func TestMarkedForCompaction(t *testing.T) {
 			}
 			d.mu.Lock()
 			defer d.mu.Unlock()
-			t := time.Now()
-			d.timeNow = func() time.Time {
-				t = t.Add(time.Second)
-				return t
-			}
+
 			s := d.mu.versions.currentVersion().DebugString()
 			return s
 

db.go

@@ -547,8 +547,6 @@ type DB struct {
 	// quickly.
 	problemSpans problemspans.ByLevel
 
-	// Normally equal to time.Now() but may be overridden in tests.
-	timeNow func() time.Time
 	// the time at database Open; may be used to compute metrics like effective
 	// compaction concurrency
 	openedAt time.Time
@@ -1901,7 +1899,7 @@ func (d *DB) Metrics() *Metrics {
 	metrics.Compact.Duration = d.mu.compact.duration
 	for c := range d.mu.compact.inProgress {
 		if !c.IsFlush() {
-			metrics.Compact.Duration += d.timeNow().Sub(c.BeganAt())
+			metrics.Compact.Duration += d.opts.private.timeNow().Sub(c.BeganAt())
 		}
 	}
 	metrics.Compact.NumProblemSpans = d.problemSpans.Len()
@@ -2026,7 +2024,7 @@ func (d *DB) Metrics() *Metrics {
 
 	metrics.SecondaryCacheMetrics = d.objProvider.Metrics()
 
-	metrics.Uptime = d.timeNow().Sub(d.openedAt)
+	metrics.Uptime = d.opts.private.timeNow().Sub(d.openedAt)
 
 	metrics.manualMemory = manual.GetMetrics()
 

download.go

@@ -67,12 +67,12 @@ func (d *DB) Download(ctx context.Context, spans []DownloadSpan) error {
 		JobID: int(d.newJobID()),
 		Spans: spans,
 	}
-	startTime := d.timeNow()
+	startTime := d.opts.private.timeNow()
 	d.opts.EventListener.DownloadBegin(info)
 
 	for info.RestartCount = 0; ; info.RestartCount++ {
 		tasks := d.createDownloadTasks(spans)
-		info.Duration = d.timeNow().Sub(startTime)
+		info.Duration = d.opts.private.timeNow().Sub(startTime)
 		if len(tasks) == 0 {
 			// We are done.
 			info.Done = true
@@ -96,7 +96,7 @@ func (d *DB) Download(ctx context.Context, spans []DownloadSpan) error {
 
 		if err != nil {
 			info.Err = err
-			info.Duration = d.timeNow().Sub(startTime)
+			info.Duration = d.opts.private.timeNow().Sub(startTime)
 			d.opts.EventListener.DownloadEnd(info)
 			return err
 		}
@@ -446,7 +446,8 @@ func (d *DB) tryLaunchDownloadForFile(
 
 	download.numLaunchedDownloads++
 	doneCh = make(chan error, 1)
-	c := newCompaction(pc, d.opts, d.timeNow(), d.objProvider, noopGrantHandle{}, d.shouldCreateShared(pc.outputLevel.level), d.determineCompactionValueSeparation)
+	c := newCompaction(pc, d.opts, d.opts.private.timeNow(), d.objProvider, noopGrantHandle{},
+		d.shouldCreateShared(pc.outputLevel.level), d.determineCompactionValueSeparation)
 	c.isDownload = true
 	d.mu.compact.downloadingCount++
 	c.AddInProgressLocked(d)

event_listener_test.go

@@ -77,16 +77,16 @@ func TestEventListener(t *testing.T) {
 				L0CompactionThreshold: 10,
 				WALDir:                "wal",
 			}
+			t := time.Now()
+			opts.private.timeNow = func() time.Time {
+				t = t.Add(time.Second)
+				return t
+			}
 			var err error
 			d, err = Open("db", opts)
 			if err != nil {
 				return err.Error()
 			}
-			t := time.Now()
-			d.timeNow = func() time.Time {
-				t = t.Add(time.Second)
-				return t
-			}
 			d.opts.private.testingAlwaysWaitForCleanup = true
 			return memLog.String()
 

open.go

@@ -15,7 +15,6 @@ import (
 	"slices"
 	"sync"
 	"sync/atomic"
-	"time"
 
 	"github.com/cockroachdb/crlib/crtime"
 	"github.com/cockroachdb/errors"
@@ -253,17 +252,15 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 	d.mu.versions.logSeqNum.Store(base.SeqNumStart)
 	d.mu.formatVers.vers.Store(uint64(formatVersion))
 	d.mu.formatVers.marker = rs.fmvMarker
-
-	d.timeNow = time.Now
-	d.openedAt = d.timeNow()
+	d.openedAt = d.opts.private.timeNow()
 
 	d.mu.Lock()
 	defer d.mu.Unlock()
 
 	jobID := d.newJobIDLocked()
 
 	blobRewriteHeuristic := manifest.BlobRewriteHeuristic{
-		CurrentTime: d.timeNow,
+		CurrentTime: d.opts.private.timeNow,
 		MinimumAge:  opts.Experimental.ValueSeparationPolicy().RewriteMinimumAge,
 	}
 

options.go

@@ -1214,6 +1214,10 @@ type Options struct {
 		// before calling DB.ingestApply.
 		testingBeforeIngestApplyFunc func()
 
+		// timeNow returns the current time. It defaults to time.Now. It's
+		// configurable here so that tests can mock the current time.
+		timeNow func() time.Time
+
 		// fsCloser holds a closer that should be invoked after a DB using these
 		// Options is closed. This is used to automatically stop the
 		// long-running goroutine associated with the disk-health-checking FS.
@@ -1683,6 +1687,9 @@ func (o *Options) EnsureDefaults() {
 	if o.Experimental.VirtualTableRewriteUnreferencedFraction == nil {
 		o.Experimental.VirtualTableRewriteUnreferencedFraction = func() float64 { return defaultVirtualTableUnreferencedFraction }
 	}
+	if o.private.timeNow == nil {
+		o.private.timeNow = time.Now
+	}
 	// TODO(jackson): Enable value separation by default once we have confidence
 	// in a default policy.
 

testdata/event_listener

@@ -147,7 +147,7 @@ close: db/marker.manifest.000002.MANIFEST-000006
 remove: db/marker.manifest.000001.MANIFEST-000001
 sync: db
 [JOB 3] MANIFEST created 000006
-[JOB 3] flushed 1 memtable (100B) to L0 [000005] (657B), in 1.0s (2.0s total), output rate 657B/s
+[JOB 3] flushed 1 memtable (100B) to L0 [000005] (657B), in 1.0s (3.0s total), output rate 657B/s
 
 compact
 ----
@@ -172,7 +172,7 @@ close: db/marker.manifest.000003.MANIFEST-000009
 remove: db/marker.manifest.000002.MANIFEST-000006
 sync: db
 [JOB 5] MANIFEST created 000009
-[JOB 5] flushed 1 memtable (100B) to L0 [000008] (657B), in 1.0s (2.0s total), output rate 657B/s
+[JOB 5] flushed 1 memtable (100B) to L0 [000008] (657B), in 1.0s (3.0s total), output rate 657B/s
 remove: db/MANIFEST-000001
 [JOB 5] MANIFEST deleted 000001
 [JOB 6] compacting(default) L0 [000005 000008] (1.3KB) Score=0.00 + L6 [] (0B) Score=0.00; OverlappingRatio: Single 0.00, Multi 0.00
@@ -206,7 +206,7 @@ close: db/marker.manifest.000004.MANIFEST-000011
 remove: db/marker.manifest.000003.MANIFEST-000009
 sync: db
 [JOB 6] MANIFEST created 000011
-[JOB 6] compacted(default) L0 [000005 000008] (1.3KB) Score=0.00 + L6 [] (0B) Score=0.00 -> L6 [000010] (652B), in 1.0s (3.0s total), output rate 652B/s
+[JOB 6] compacted(default) L0 [000005 000008] (1.3KB) Score=0.00 + L6 [] (0B) Score=0.00 -> L6 [000010] (652B), in 1.0s (4.0s total), output rate 652B/s
 close: db/000005.sst
 close: db/000008.sst
 remove: db/MANIFEST-000006
@@ -242,7 +242,7 @@ close: db/marker.manifest.000005.MANIFEST-000014
 remove: db/marker.manifest.000004.MANIFEST-000011
 sync: db
 [JOB 8] MANIFEST created 000014
-[JOB 8] flushed 1 memtable (100B) to L0 [000013] (657B), in 1.0s (2.0s total), output rate 657B/s
+[JOB 8] flushed 1 memtable (100B) to L0 [000013] (657B), in 1.0s (3.0s total), output rate 657B/s
 
 enable-file-deletions
 ----
@@ -401,7 +401,7 @@ sync-data: db/000022.sst
 close: db/000022.sst
 sync: db
 sync: db/MANIFEST-000016
-[JOB 15] flushed 1 memtable (100B) to L0 [000022] (657B), in 1.0s (2.0s total), output rate 657B/s
+[JOB 15] flushed 1 memtable (100B) to L0 [000022] (657B), in 1.0s (3.0s total), output rate 657B/s
 [JOB 16] flushing 2 ingested tables
 create: db/MANIFEST-000023
 close: db/MANIFEST-000016
@@ -412,7 +412,7 @@ close: db/marker.manifest.000007.MANIFEST-000023
 remove: db/marker.manifest.000006.MANIFEST-000016
 sync: db
 [JOB 16] MANIFEST created 000023
-[JOB 16] flushed 2 ingested flushables L0:000017 (652B) + L6:000018 (652B) in 1.0s (2.0s total), output rate 1.3KB/s
+[JOB 16] flushed 2 ingested flushables L0:000017 (652B) + L6:000018 (652B) in 1.0s (3.0s total), output rate 1.3KB/s
 remove: db/MANIFEST-000014
 [JOB 16] MANIFEST deleted 000014
 [JOB 17] flushing 1 memtable (100B) to L0