db: rework logAndApply

This commit replaces `logAndApply` with an `UpdateVersionLocked`
function that takes a closure and performs the log "locking" around
that closure. This makes it impossible for a caller to mess up the
locking.

Author: RaduBerinde

compaction.go

@@ -1100,10 +1100,10 @@ func (d *DB) clearCompactingState(c *compaction, rollback bool) {
 		// It is a bit peculiar that we are fiddling with th current version state
 		// in a separate critical section from when this version was installed.
 		// But this fiddling is necessary if the compaction failed. When the
-		// compaction succeeded, we've already done this in logAndApply, so this
-		// seems redundant. Anyway, we clear the pickedCompactionCache since we
+		// compaction succeeded, we've already done this in UpdateVersionLocked, so
+		// this seems redundant. Anyway, we clear the pickedCompactionCache since we
 		// may be able to pick a better compaction (though when this compaction
-		// succeeded we've also cleared the cache in logAndApply).
+		// succeeded we've also cleared the cache in UpdateVersionLocked).
 		defer d.mu.versions.logUnlockAndInvalidatePickedCompactionCache()
 		d.mu.versions.l0Organizer.InitCompactingFileInfo(l0InProgress)
 	}()
@@ -1467,44 +1467,38 @@ func (d *DB) flush1() (bytesFlushed uint64, err error) {
 	d.addInProgressCompaction(c)
 
 	jobID := d.newJobIDLocked()
-	d.opts.EventListener.FlushBegin(FlushInfo{
+	info := FlushInfo{
 		JobID:      int(jobID),
 		Input:      inputs,
 		InputBytes: inputBytes,
 		Ingest:     ingest,
-	})
+	}
+	d.opts.EventListener.FlushBegin(info)
+
 	startTime := d.timeNow()
 
 	var ve *manifest.VersionEdit
 	var stats compact.Stats
 	// To determine the target level of the files in the ingestedFlushable, we
-	// need to acquire the logLock, and not release it for that duration. Since,
-	// we need to acquire the logLock below to perform the logAndApply step
-	// anyway, we create the VersionEdit for ingestedFlushable outside of
-	// runCompaction. For all other flush cases, we construct the VersionEdit
-	// inside runCompaction.
+	// need to acquire the logLock, and not release it for that duration. Since
+	// UpdateVersionLocked acquires it anyway, we create the VersionEdit for
+	// ingestedFlushable outside runCompaction. For all other flush cases, we
+	// construct the VersionEdit inside runCompaction.
+	var compactionErr error
 	if c.kind != compactionKindIngestedFlushable {
-		ve, stats, err = d.runCompaction(jobID, c)
+		ve, stats, compactionErr = d.runCompaction(jobID, c)
 	}
 
-	// Acquire logLock. This will be released either on an error, by way of
-	// logUnlock, or through a call to logAndApply if there is no error.
-	d.mu.versions.logLock()
-
-	if c.kind == compactionKindIngestedFlushable {
-		ve, err = d.runIngestFlush(c)
-	}
+	err = d.mu.versions.UpdateVersionLocked(func() (versionUpdate, error) {
+		err := compactionErr
+		if c.kind == compactionKindIngestedFlushable {
+			ve, err = d.runIngestFlush(c)
+		}
+		info.Duration = d.timeNow().Sub(startTime)
+		if err != nil {
+			return versionUpdate{}, err
+		}
 
-	info := FlushInfo{
-		JobID:      int(jobID),
-		Input:      inputs,
-		InputBytes: inputBytes,
-		Duration:   d.timeNow().Sub(startTime),
-		Done:       true,
-		Ingest:     ingest,
-		Err:        err,
-	}
-	if err == nil {
 		validateVersionEdit(ve, d.opts.Comparer.ValidateKey, d.opts.Comparer.FormatKey, d.opts.Logger)
 		for i := range ve.NewTables {
 			e := &ve.NewTables[i]
@@ -1515,9 +1509,6 @@ func (d *DB) flush1() (bytesFlushed uint64, err error) {
 				info.IngestLevels = append(info.IngestLevels, e.Level)
 			}
 		}
-		if len(ve.NewTables) == 0 {
-			info.Err = errEmptyTable
-		}
 
 		// The flush succeeded or it produced an empty sstable. In either case we
 		// want to bump the minimum unflushed log number to the log number of the
@@ -1550,7 +1541,6 @@ func (d *DB) flush1() (bytesFlushed uint64, err error) {
 				// write a file overlapping with the excise span.
 				if ingestFlushable.exciseSpan.OverlapsInternalKeyRange(d.cmp, c2.smallest, c2.largest) {
 					c2.cancel.Store(true)
-					continue
 				}
 			}
 
@@ -1570,17 +1560,13 @@ func (d *DB) flush1() (bytesFlushed uint64, err error) {
 				}
 			}
 		}
-		err = d.mu.versions.logAndApply(jobID, ve, &c.metrics, false, /* forceRotation */
-			func() []compactionInfo { return d.getInProgressCompactionInfoLocked(c) })
-		if err != nil {
-			info.Err = err
-		}
-	} else {
-		// We won't be performing the logAndApply step because of the error, so
-		// logUnlock. We don't need to invalidate the pickedCompactionCache since
-		// the flush failed and so the latest version has not changed.
-		d.mu.versions.logUnlock()
-	}
+		return versionUpdate{
+			VE:                      ve,
+			JobID:                   jobID,
+			Metrics:                 c.metrics,
+			InProgressCompactionsFn: func() []compactionInfo { return d.getInProgressCompactionInfoLocked(c) },
+		}, nil
+	})
 
 	// If err != nil, then the flush will be retried, and we will recalculate
 	// these metrics.
@@ -1610,11 +1596,15 @@ func (d *DB) flush1() (bytesFlushed uint64, err error) {
 			}
 		}
 		d.maybeTransitionSnapshotsToFileOnlyLocked()
-
 	}
 	// Signal FlushEnd after installing the new readState. This helps for unit
 	// tests that use the callback to trigger a read using an iterator with
 	// IterOptions.OnlyReadGuaranteedDurable.
+	info.Err = err
+	if info.Err == nil && len(ve.NewTables) == 0 {
+		info.Err = errEmptyTable
+	}
+	info.Done = true
 	info.TotalDuration = d.timeNow().Sub(startTime)
 	d.opts.EventListener.FlushEnd(info)
 
@@ -2468,9 +2458,7 @@ func (d *DB) compact1(c *compaction, errChannel chan error) (err error) {
 	info.Duration = d.timeNow().Sub(startTime)
 	if err == nil {
 		validateVersionEdit(ve, d.opts.Comparer.ValidateKey, d.opts.Comparer.FormatKey, d.opts.Logger)
-		err = func() error {
-			var err error
-			d.mu.versions.logLock()
+		err = d.mu.versions.UpdateVersionLocked(func() (versionUpdate, error) {
 			// Check if this compaction had a conflicting operation (eg. a d.excise())
 			// that necessitates it restarting from scratch. Note that since we hold
 			// the manifest lock, we don't expect this bool to change its value
@@ -2485,17 +2473,18 @@ func (d *DB) compact1(c *compaction, errChannel chan error) (err error) {
 				// would not have been true). We should delete any tables already
 				// created, as d.runCompaction did not do that.
 				d.cleanupVersionEdit(ve)
-				// logAndApply calls logUnlock. If we didn't call it, we need to call
-				// logUnlock ourselves. We also invalidate the pickedCompactionCache
-				// since this failed compaction may be the highest priority to run
-				// next.
-				d.mu.versions.logUnlockAndInvalidatePickedCompactionCache()
-				return err
+				// Note that UpdateVersionLocked invalidates the pickedCompactionCache
+				// when we return, which is relevant because this failed compaction
+				// may be the highest priority to run next.
+				return versionUpdate{}, err
 			}
-			return d.mu.versions.logAndApply(jobID, ve, &c.metrics, false /* forceRotation */, func() []compactionInfo {
-				return d.getInProgressCompactionInfoLocked(c)
-			})
-		}()
+			return versionUpdate{
+				VE:                      ve,
+				JobID:                   jobID,
+				Metrics:                 c.metrics,
+				InProgressCompactionsFn: func() []compactionInfo { return d.getInProgressCompactionInfoLocked(c) },
+			}, nil
+		})
 	}
 
 	info.Done = true

compaction_picker.go

@@ -695,8 +695,8 @@ type compactionPickerByScore struct {
 	// This means that at some point in the future a compactionPickerByScore
 	// created in the past will have mutually inconsistent state in vers and
 	// l0Organizer. This is not a problem since (a) a new picker is created in
-	// logAndApply when a new version is installed, and (b) only the latest picker
-	// is used for picking compactions. This is ensured by holding
+	// UpdateVersionLocked when a new version is installed, and (b) only the
+	// latest picker is used for picking compactions. This is ensured by holding
 	// versionSet.logLock for both (a) and (b).
 	l0Organizer     *manifest.L0Organizer
 	virtualBackings *manifest.VirtualBackings

compaction_test.go

@@ -2558,7 +2558,7 @@ func (i *createManifestErrorInjector) MaybeError(op errorfs.Op) error {
 		return nil
 	}
 	// This necessitates having a MaxManifestSize of 1, to reliably induce
-	// logAndApply errors.
+	// UpdateVersionLocked errors.
 	if strings.Contains(op.Path, "MANIFEST") && op.Kind == errorfs.OpCreate {
 		return errorfs.ErrInjected
 	}
@@ -2567,11 +2567,11 @@ func (i *createManifestErrorInjector) MaybeError(op errorfs.Op) error {
 
 var _ errorfs.Injector = &createManifestErrorInjector{}
 
-// TestCompaction_LogAndApplyFails exercises a flush or ingest encountering an
-// unrecoverable error during logAndApply.
+// TestCompaction_UpdateVersionFails exercises a flush or ingest encountering an
+// unrecoverable error during UpdateVersionLocked.
 //
 // Regression test for #1669.
-func TestCompaction_LogAndApplyFails(t *testing.T) {
+func TestCompaction_UpdateVersionFails(t *testing.T) {
 	// flushKeys writes the given keys to the DB, flushing the resulting memtable.
 	var key = []byte("foo")
 	flushErrC := make(chan error)
@@ -2654,8 +2654,9 @@ func TestCompaction_LogAndApplyFails(t *testing.T) {
 		err = addFn(db)
 		require.True(t, errors.Is(err, errorfs.ErrInjected))
 
-		// Under normal circumstances, such an error in logAndApply would panic and
-		// cause the DB to terminate here. Assert that we captured the fatal error.
+		// Under normal circumstances, such an error in UpdateVersionLocked would
+		// panic and cause the DB to terminate here. Assert that we captured the
+		// fatal error.
 		require.True(t, errors.Is(logger.err, errorfs.ErrInjected))
 	}
 	for _, tc := range testCases {

data_test.go

@@ -1102,10 +1102,15 @@ func runDBDefineCmdReuseFS(td *datadriven.TestData, opts *Options) (*DB, error)
 		ve.NewBlobFiles = slices.Collect(maps.Values(valueSeparator.metas))
 
 		jobID := d.newJobIDLocked()
-		d.mu.versions.logLock()
-		if err := d.mu.versions.logAndApply(jobID, ve, newFileMetrics(ve.NewTables), false, func() []compactionInfo {
-			return nil
-		}); err != nil {
+		err = d.mu.versions.UpdateVersionLocked(func() (versionUpdate, error) {
+			return versionUpdate{
+				VE:                      ve,
+				JobID:                   jobID,
+				Metrics:                 newFileMetrics(ve.NewTables),
+				InProgressCompactionsFn: func() []compactionInfo { return nil },
+			}, nil
+		})
+		if err != nil {
 			return nil, err
 		}
 		d.updateReadStateLocked(nil)

db.go

@@ -322,7 +322,7 @@ type DB struct {
 	//
 	// Care is taken to avoid holding DB.mu during IO operations. Accomplishing
 	// this sometimes requires releasing DB.mu in a method that was called with
-	// it held. See versionSet.logAndApply() and DB.makeRoomForWrite() for
+	// it held. See versionSet.UpdateVersionLocked() and DB.makeRoomForWrite() for
 	// examples. This is a common pattern, so be careful about expectations that
 	// DB.mu will be held continuously across a set of calls.
 	mu struct {

file_cache_test.go

@@ -393,7 +393,7 @@ func TestVirtualReadsWiring(t *testing.T) {
 	d.checkVirtualBounds(v2)
 
 	// Write the version edit.
-	fileMetrics := func(ve *versionEdit) *levelMetricsDelta {
+	fileMetrics := func(ve *versionEdit) levelMetricsDelta {
 		metrics := newFileMetrics(ve.NewTables)
 		for de, f := range ve.DeletedTables {
 			lm := metrics[de.Level]
@@ -408,11 +408,13 @@ func TestVirtualReadsWiring(t *testing.T) {
 	}
 
 	applyVE := func(ve *versionEdit) error {
-		d.mu.versions.logLock()
-		jobID := d.newJobIDLocked()
-
-		err := d.mu.versions.logAndApply(jobID, ve, fileMetrics(ve), false, func() []compactionInfo {
-			return d.getInProgressCompactionInfoLocked(nil)
+		err := d.mu.versions.UpdateVersionLocked(func() (versionUpdate, error) {
+			return versionUpdate{
+				VE:                      ve,
+				JobID:                   d.newJobIDLocked(),
+				Metrics:                 fileMetrics(ve),
+				InProgressCompactionsFn: func() []compactionInfo { return d.getInProgressCompactionInfoLocked(nil) },
+			}, nil
 		})
 		d.updateReadStateLocked(nil)
 		return err

flushable_test.go

@@ -96,9 +96,9 @@ func TestIngestedSSTFlushableAPI(t *testing.T) {
 		}
 
 		// Fsync the directory we added the tables to. We need to do this at some
-		// point before we update the MANIFEST (via logAndApply), otherwise a crash
-		// can have the tables referenced in the MANIFEST, but not present in the
-		// directory.
+		// 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 {
 			t.Fatal(err)
 		}

format_major_version.go

@@ -566,47 +566,45 @@ func (d *DB) markFilesLocked(findFn findFilesFunc) error {
 
 	// Lock the manifest for a coherent view of the LSM. The database lock has
 	// been re-acquired by the defer within the above anonymous function.
-	d.mu.versions.logLock()
-	vers := d.mu.versions.currentVersion()
-	for l, filesToMark := range files {
-		if len(filesToMark) == 0 {
-			continue
-		}
-		for _, f := range filesToMark {
-			// Ignore files to be marked that have already been compacted or marked.
-			if f.CompactionState == manifest.CompactionStateCompacted ||
-				f.MarkedForCompaction {
+	return d.mu.versions.UpdateVersionLocked(func() (versionUpdate, error) {
+		vers := d.mu.versions.currentVersion()
+		for l, filesToMark := range files {
+			if len(filesToMark) == 0 {
 				continue
 			}
-			// Else, mark the file for compaction in this version.
-			vers.Stats.MarkedForCompaction++
-			f.MarkedForCompaction = true
+			for _, f := range filesToMark {
+				// Ignore files to be marked that have already been compacted or marked.
+				if f.CompactionState == manifest.CompactionStateCompacted ||
+					f.MarkedForCompaction {
+					continue
+				}
+				// Else, mark the file for compaction in this version.
+				vers.Stats.MarkedForCompaction++
+				f.MarkedForCompaction = true
+			}
+			// The compaction picker uses the markedForCompactionAnnotator to
+			// quickly find files marked for compaction, or to quickly determine
+			// that there are no such files marked for compaction within a level.
+			// A b-tree node may be annotated with an annotation recording that
+			// there are no files marked for compaction within the node's subtree,
+			// based on the assumption that it's static.
+			//
+			// Since we're marking files for compaction, these b-tree nodes'
+			// annotations will be out of date. Clear the compaction-picking
+			// annotation, so that it's recomputed the next time the compaction
+			// picker looks for a file marked for compaction.
+			markedForCompactionAnnotator.InvalidateLevelAnnotation(vers.Levels[l])
 		}
-		// The compaction picker uses the markedForCompactionAnnotator to
-		// quickly find files marked for compaction, or to quickly determine
-		// that there are no such files marked for compaction within a level.
-		// A b-tree node may be annotated with an annotation recording that
-		// there are no files marked for compaction within the node's subtree,
-		// based on the assumption that it's static.
-		//
-		// Since we're marking files for compaction, these b-tree nodes'
-		// annotations will be out of date. Clear the compaction-picking
-		// annotation, so that it's recomputed the next time the compaction
-		// picker looks for a file marked for compaction.
-		markedForCompactionAnnotator.InvalidateLevelAnnotation(vers.Levels[l])
-	}
-
-	// The 'marked-for-compaction' bit is persisted in the MANIFEST file
-	// metadata. We've already modified the in-memory table metadata, but the
-	// manifest hasn't been updated. Force rotation to a new MANIFEST file,
-	// which will write every table metadata to the new manifest file and ensure
-	// that the now marked-for-compaction table metadata are persisted as marked.
-	// NB: This call to logAndApply will unlockthe MANIFEST, which we locked up
-	// above before obtaining `vers`.
-	return d.mu.versions.logAndApply(
-		jobID,
-		&manifest.VersionEdit{},
-		nil,  /* metrics */
-		true, /* forceRotation */
-		func() []compactionInfo { return d.getInProgressCompactionInfoLocked(nil) })
+		// The 'marked-for-compaction' bit is persisted in the MANIFEST file
+		// metadata. We've already modified the in-memory table metadata, but the
+		// manifest hasn't been updated. Force rotation to a new MANIFEST file,
+		// which will write every table metadata to the new manifest file and ensure
+		// that the now marked-for-compaction table metadata are persisted as marked.
+		return versionUpdate{
+			VE:                      &manifest.VersionEdit{},
+			JobID:                   jobID,
+			ForceManifestRotation:   true,
+			InProgressCompactionsFn: func() []compactionInfo { return d.getInProgressCompactionInfoLocked(nil) },
+		}, nil
+	})
 }