Repository: cockroachdb/pebble
Pull request #2310 (2023-02-06 15:50:24)

crl-release-22.1: vfs: handle concurrent directory Syncs in disk-health checking
22.1 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/2310)
<!-- Reviewable:end -->

jbowens: CI revealed a race in a `vfs.Options` initialization
jbowens: TFTR!

Patch:
diff --git a/options.go b/options.go
index cb6b39a0..cf277805 100644
--- a/options.go
+++ b/options.go
@@ -203,7 +203,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.
 //
@@ -848,13 +848,7 @@ func (o *Options) EnsureDefaults() *Options {
 	}
 
 	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
@@ -879,6 +873,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/internal/metamorphic/meta_test.go b/internal/metamorphic/meta_test.go
index 3da3d657..b15a899a 100644
--- a/internal/metamorphic/meta_test.go
+++ b/internal/metamorphic/meta_test.go
@@ -171,6 +171,7 @@ func testMetaRun(t *testing.T, runDir string, seed uint64, historyPath string) {
 			opts.FS = vfs.NewMem()
 		}
 	}
+	opts.WithFSDefaults()
 
 	dir := opts.FS.PathJoin(runDir, "data")
 	// Set up the initial database state if configured to start from a non-empty
@@ -238,7 +239,7 @@ func readHistory(t *testing.T, historyPath string) []string {
 // subsequent invocation will overwrite that output. A test can be re-run by
 // using the `--run-dir` flag. For example:
 //
-//   go test -v -run TestMeta --run-dir _meta/standard-017
+//	go test -v -run TestMeta --run-dir _meta/standard-017
 //
 // This will reuse the existing operations present in _meta/ops, rather than
 // generating a new set.
@@ -247,7 +248,7 @@ func readHistory(t *testing.T, historyPath string) []string {
 // pseudorandom number generator seed. If a failure occurs, the seed is
 // printed, and the full suite of tests may be re-run using the `--seed` flag:
 //
-//   go test -v -run TestMeta --seed 1594395154492165000
+//	go test -v -run TestMeta --seed 1594395154492165000
 //
 // This will generate a new `_meta/<test>` directory, with the same operations
 // and options. This must be run on the same commit SHA as the original