Repository: cockroachdb/pebble
Pull request #2309 (2023-02-06 15:46:18)

crl-release-22.2: vfs: handle concurrent directory Syncs in disk-health checking 
22.2 backport of #2298.

----

**db: add Options.WithFSDefaults**

Add a facility for easily layering in the default VFS middleware—currently the
disk-health checking FS. Options.EnsureDefaults by default uses the disk-health
checking FS, but most of our tests explicitly set a VFS, and in particular a
*vfs.MemFS. These tests have always run without the disk-health checking
filesystem layer. Use the new WithFSDefaults method across many Pebble unit
tests and the Pebble metamorphic tests.

This is sufficient to surface the concurrent `Sync` operations observed in
https://github.com/cockroachdb/cockroach/issues/96422 and https://github.com/cockroachdb/cockroach/issues/96414. See https://github.com/cockroachdb/pebble/pull/2282 for
context on where this panic is originating.

**vfs: handle concurrent directory Syncs in disk-health checking**

The file-level disk-health checker requires that a file not be used
concurrently, because it only supports timing a single in-flight operation at a
time. Pebble did not adhere to this contract for the data directory, which it
synced concurrently. This had the potential to leave a data directory Sync
untimed if an in-flight Syncs' timestamp was overwritten by the completion of
another Sync.

In https://github.com/cockroachdb/pebble/pull/2282 we began checking for serialized writes in `invariants` builds. This
revealed these concurrent syncs in CockroachDB test failures under `-race`:
https://github.com/cockroachdb/cockroach/issues/96414 and https://github.com/cockroachdb/cockroach/issues/96422.

cockroach-teamcity: 

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

jbowens: TFTR!

Patch:
diff --git a/options.go b/options.go
index 4e3d5775..a06a2341 100644
--- a/options.go
+++ b/options.go
@@ -224,7 +224,7 @@ func (o *IterOptions) getLogger() Logger {
 // Specifically, when configured with a RangeKeyMasking.Suffix _s_, and there
 // exists a range key with suffix _r_ covering a point key with suffix _p_, and
 //
-//     _s_ ≤ _r_ < _p_
+//	_s_ ≤ _r_ < _p_
 //
 // then the point key is elided.
 //
@@ -927,15 +927,8 @@ func (o *Options) EnsureDefaults() *Options {
 	if o.FormatMajorVersion == FormatDefault {
 		o.FormatMajorVersion = FormatMostCompatible
 	}
-
 	if o.FS == nil {
-		o.FS, o.private.fsCloser = vfs.WithDiskHealthChecks(vfs.Default, 5*time.Second,
-			func(name string, duration time.Duration) {
-				o.EventListener.DiskSlow(DiskSlowInfo{
-					Path:     name,
-					Duration: duration,
-				})
-			})
+		o.WithFSDefaults()
 	}
 	if o.FlushSplitBytes <= 0 {
 		o.FlushSplitBytes = 2 * o.Levels[0].TargetFileSize
@@ -960,6 +953,22 @@ func (o *Options) EnsureDefaults() *Options {
 	return o
 }
 
+// WithFSDefaults configures the Options to wrap the configured filesystem with
+// the default virtual file system middleware, like disk-health checking.
+func (o *Options) WithFSDefaults() *Options {
+	if o.FS == nil {
+		o.FS = vfs.Default
+	}
+	o.FS, o.private.fsCloser = vfs.WithDiskHealthChecks(o.FS, 5*time.Second,
+		func(name string, duration time.Duration) {
+			o.EventListener.DiskSlow(DiskSlowInfo{
+				Path:     name,
+				Duration: duration,
+			})
+		})
+	return o
+}
+
 func (o *Options) equal() Equal {
 	if o.Comparer.Equal == nil {
 		return bytes.Equal
diff --git a/vfs/disk_health.go b/vfs/disk_health.go
index 916f5a0b..bce0377a 100644
--- a/vfs/disk_health.go
+++ b/vfs/disk_health.go
@@ -175,6 +175,26 @@ func (d *diskHealthCheckingFile) timeDiskOp(op func()) {
 	op()
 }
 
+// diskHealthCheckingDir implements disk-health checking for directories. Unlike
+// other files, we allow directories to receive concurrent write operations
+// (Syncs are the only write operations supported by a directory.) Since the
+// diskHealthCheckingFile's timeDiskOp can only track a single in-flight
+// operation at a time, we time the operation using the filesystem-level
+// timeFilesystemOp function instead.
+type diskHealthCheckingDir struct {
+	File
+	name string
+	fs   *diskHealthCheckingFS
+}
+
+// Sync implements the io.Syncer interface.
+func (d *diskHealthCheckingDir) Sync() (err error) {
+	d.fs.timeFilesystemOp(d.name, func() {
+		err = d.File.Sync()
+	})
+	return err
+}
+
 // diskHealthCheckingFS adds disk-health checking facilities to a VFS.
 // It times disk write operations in two ways:
 //
@@ -459,11 +479,11 @@ func (d *diskHealthCheckingFS) OpenDir(name string) (File, error) {
 	}
 	// Directories opened with OpenDir must be opened with health checking,
 	// because they may be explicitly synced.
-	checkingFile := newDiskHealthCheckingFile(f, d.diskSlowThreshold, func(duration time.Duration) {
-		d.onSlowDisk(name, duration)
-	})
-	checkingFile.startTicker()
-	return checkingFile, nil
+	return &diskHealthCheckingDir{
+		File: f,
+		name: name,
+		fs:   d,
+	}, nil
 }
 
 // PathBase implements the FS interface.


Test patch:
diff --git a/cleaner_test.go b/cleaner_test.go
index c953fc0f..29db4411 100644
--- a/cleaner_test.go
+++ b/cleaner_test.go
@@ -25,11 +25,11 @@ func TestArchiveCleaner(t *testing.T) {
 
 	var buf syncedBuffer
 	mem := vfs.NewMem()
-	opts := &Options{
+	opts := (&Options{
 		Cleaner: ArchiveCleaner{},
 		FS:      loggingFS{mem, &buf},
 		WALDir:  "wal",
-	}
+	}).WithFSDefaults()
 
 	datadriven.RunTest(t, "testdata/cleaner", func(td *datadriven.TestData) string {
 		switch td.Cmd {
diff --git a/format_major_version_test.go b/format_major_version_test.go
index 634ac591..8dbc5276 100644
--- a/format_major_version_test.go
+++ b/format_major_version_test.go
@@ -32,7 +32,7 @@ func TestFormatMajorVersion_MigrationDefined(t *testing.T) {
 
 func TestRatchetFormat(t *testing.T) {
 	fs := vfs.NewMem()
-	d, err := Open("", &Options{FS: fs})
+	d, err := Open("", (&Options{FS: fs}).WithFSDefaults())
 	require.NoError(t, err)
 	require.NoError(t, d.Set([]byte("foo"), []byte("bar"), Sync))
 	require.Equal(t, FormatMostCompatible, d.FormatMajorVersion())
@@ -60,7 +60,7 @@ func TestRatchetFormat(t *testing.T) {
 
 	// If we Open the database again, leaving the default format, the
 	// database should Open using the persisted FormatNewest.
-	d, err = Open("", &Options{FS: fs})
+	d, err = Open("", (&Options{FS: fs}).WithFSDefaults())
 	require.NoError(t, err)
 	require.Equal(t, FormatNewest, d.FormatMajorVersion())
 	require.NoError(t, d.Close())
@@ -71,10 +71,10 @@ func TestRatchetFormat(t *testing.T) {
 	require.NoError(t, m.Move("999999"))
 	require.NoError(t, m.Close())
 
-	_, err = Open("", &Options{
+	_, err = Open("", (&Options{
 		FS:                 fs,
 		FormatMajorVersion: FormatVersioned,
-	})
+	}).WithFSDefaults())
 	require.Error(t, err)
 	require.EqualError(t, err, `pebble: database "" written in format major version 999999`)
 }
@@ -105,10 +105,10 @@ func TestFormatMajorVersions(t *testing.T) {
 	for vers := FormatMostCompatible; vers <= FormatNewest; vers++ {
 		t.Run(fmt.Sprintf("vers=%03d", vers), func(t *testing.T) {
 			fs := vfs.NewStrictMem()
-			opts := &Options{
+			opts := (&Options{
 				FS:                 fs,
 				FormatMajorVersion: vers,
-			}
+			}).WithFSDefaults()
 
 			// Create a database at this format major version and perform
 			// some very basic operations.
@@ -251,7 +251,7 @@ func TestSplitUserKeyMigration(t *testing.T) {
 					}
 					buf.Reset()
 				}
-				opts = &Options{
+				opts = (&Options{
 					FormatMajorVersion: FormatBlockPropertyCollector,
 					EventListener: EventListener{
 						CompactionEnd: func(info CompactionInfo) {
@@ -263,7 +263,7 @@ func TestSplitUserKeyMigration(t *testing.T) {
 						},
 					},
 					DisableAutomaticCompactions: true,
-				}
+				}).WithFSDefaults()
 				var err error
 				if d, err = runDBDefineCmd(td, opts); err != nil {
 					return err.Error()
@@ -359,10 +359,10 @@ func TestPebblev1Migration(t *testing.T) {
 						return fmt.Sprintf("unknown argument: %s", cmd)
 					}
 				}
-				opts := &Options{
+				opts := (&Options{
 					FS:                 vfs.NewMem(),
 					FormatMajorVersion: FormatMajorVersion(version),
-				}
+				}).WithFSDefaults()
 				d, err = Open("", opts)
 				if err != nil {
 					return err.Error()
@@ -480,13 +480,13 @@ func TestPebblev1MigrationRace(t *testing.T) {
 	defer cache.Unref()
 	tableCache := NewTableCache(cache, 1, 5)
 	defer tableCache.Unref()
-	d, err := Open("", &Options{
+	d, err := Open("", (&Options{
 		Cache:              cache,
 		FS:                 vfs.NewMem(),
 		FormatMajorVersion: FormatMajorVersion(FormatPrePebblev1Marked - 1),
 		TableCache:         tableCache,
 		Levels:             []LevelOptions{{TargetFileSize: 1}},
-	})
+	}).WithFSDefaults())
 	require.NoError(t, err)
 	defer d.Close()
 
@@ -525,7 +525,7 @@ func TestPebblev1MigrationRace(t *testing.T) {
 // Regression test for #2044, where multiple concurrent compactions can lead
 // to an indefinite wait on the compaction goroutine in compactMarkedFilesLocked.
 func TestPebblev1MigrationConcurrencyRace(t *testing.T) {
-	opts := &Options{
+	opts := (&Options{
 		Comparer:           testkeys.Comparer,
 		FS:                 vfs.NewMem(),
 		FormatMajorVersion: FormatSplitUserKeysMarked,
@@ -533,7 +533,7 @@ func TestPebblev1MigrationConcurrencyRace(t *testing.T) {
 		MaxConcurrentCompactions: func() int {
 			return 4
 		},
-	}
+	}).WithFSDefaults()
 	func() {
 		d, err := Open("", opts)
 		require.NoError(t, err)
diff --git a/internal/metamorphic/meta_test.go b/internal/metamorphic/meta_test.go
index 18be696c..a704297d 100644
--- a/internal/metamorphic/meta_test.go
+++ b/internal/metamorphic/meta_test.go
@@ -166,6 +166,8 @@ func testMetaRun(t *testing.T, runDir string, seed uint64, historyPath string) {
 			opts.FS = vfs.NewMem()
 		}
 	}
+	opts.WithFSDefaults()
+
 	threads := testOpts.threads
 	if *maxThreads < threads {
 		threads = *maxThreads