Repository: petermattis/pebble (go)

Issue:
db: sprinkle directory fsyncs where necessary
Directories need fsyncs just like files. Whenever we manipulate filenames (file creation, or renaming) and expect that operation to survive a crash, we need to fsync the parent directory.

Pull request description:
Sync the DB dir whenever we manipulate directory entries
Whenever an sstable is created (due to flush or compaction) or a WAL or
MANIFEST is created, we need to sync the parent directory in order to
ensure that the directory update is persisted to disk.

Set O_CLOEXEC and all of the files created and opened by Pebble, to
match RocksDB behavior.

Fixes #125

Patch:
diff --git a/compaction.go b/compaction.go
index e3f8806..30f8be8 100644
--- a/compaction.go
+++ b/compaction.go
@@ -478,8 +478,7 @@ func (d *DB) flush1() error {
 		})
 	}
 
-	meta, err := d.writeLevel0Table(d.opts.VFS, iter,
-		true /* allowRangeTombstoneElision */)
+	meta, err := d.writeLevel0Table(iter, true /* allowRangeTombstoneElision */)
 
 	if d.opts.EventListener != nil && d.opts.EventListener.FlushEnd != nil {
 		info := db.FlushInfo{
@@ -507,7 +506,7 @@ func (d *DB) flush1() error {
 		}
 	}
 
-	err = d.mu.versions.logAndApply(ve)
+	err = d.mu.versions.logAndApply(ve, d.dir)
 	for i := range ve.newFiles {
 		f := &ve.newFiles[i]
 		if _, ok := d.mu.compact.pendingOutputs[f.meta.fileNum]; !ok {
@@ -538,7 +537,7 @@ func (d *DB) flush1() error {
 // d.mu must be held when calling this, but the mutex may be dropped and
 // re-acquired during the course of this method.
 func (d *DB) writeLevel0Table(
-	fs vfs.FS, iiter internalIterator, allowRangeTombstoneElision bool,
+	iiter internalIterator, allowRangeTombstoneElision bool,
 ) (meta fileMetadata, err error) {
 	meta.fileNum = d.mu.versions.nextFileNum()
 	filename := dbFilename(d.dirname, fileTypeTable, meta.fileNum)
@@ -609,12 +608,12 @@ func (d *DB) writeLevel0Table(
 			err = firstError(err, tw.Close())
 		}
 		if err != nil {
-			fs.Remove(filename)
+			d.opts.FS.Remove(filename)
 			meta = fileMetadata{}
 		}
 	}()
 
-	file, err = fs.Create(filename)
+	file, err = d.opts.FS.Create(filename)
 	if err != nil {
 		return fileMetadata{}, err
 	}
@@ -652,6 +651,10 @@ func (d *DB) writeLevel0Table(
 		return fileMetadata{}, errEmptyTable
 	}
 
+	if err := d.dir.Sync(); err != nil {
+		return fileMetadata{}, err
+	}
+
 	writerMeta, err := tw.Metadata()
 	if err != nil {
 		return fileMetadata{}, err
@@ -762,7 +765,7 @@ func (d *DB) compact1() (err error) {
 	if err != nil {
 		return err
 	}
-	err = d.mu.versions.logAndApply(ve)
+	err = d.mu.versions.logAndApply(ve, d.dir)
 	for _, fileNum := range pendingOutputs {
 		if _, ok := d.mu.compact.pendingOutputs[fileNum]; !ok {
 			panic("pebble: expected pending output not present")
@@ -837,7 +840,7 @@ func (d *DB) compactDiskTables(c *compaction) (ve *versionEdit, pendingOutputs [
 		}
 		if retErr != nil {
 			for _, filename := range filenames {
-				d.opts.VFS.Remove(filename)
+				d.opts.FS.Remove(filename)
 			}
 		}
 	}()
@@ -854,7 +857,7 @@ func (d *DB) compactDiskTables(c *compaction) (ve *versionEdit, pendingOutputs [
 		d.mu.Unlock()
 
 		filename := dbFilename(d.dirname, fileTypeTable, fileNum)
-		file, err := d.opts.VFS.Create(filename)
+		file, err := d.opts.FS.Create(filename)
 		if err != nil {
 			return err
 		}
@@ -957,7 +960,7 @@ func (d *DB) compactDiskTables(c *compaction) (ve *versionEdit, pendingOutputs [
 	}
 
 	if err := finishOutput(db.InternalKey{}); err != nil {
-		return nil, pendingOutputs, nil
+		return nil, pendingOutputs, err
 	}
 
 	for i := range c.inputs {
@@ -968,6 +971,10 @@ func (d *DB) compactDiskTables(c *compaction) (ve *versionEdit, pendingOutputs [
 			}] = true
 		}
 	}
+
+	if err := d.dir.Sync(); err != nil {
+		return nil, pendingOutputs, err
+	}
 	return ve, pendingOutputs, nil
 }
 
@@ -984,7 +991,7 @@ func (d *DB) scanObsoleteFiles() {
 	d.mu.Unlock()
 	defer d.mu.Lock()
 
-	fs := d.opts.VFS
+	fs := d.opts.FS
 	list, err := fs.List(d.dirname)
 	if err != nil {
 		// Ignore any filesystem errors.
@@ -1115,7 +1122,7 @@ func (d *DB) deleteObsoleteFiles(jobID int) {
 			}
 
 			path := dbFilename(d.dirname, f.fileType, fileNum)
-			err := d.opts.VFS.Remove(path)
+			err := d.opts.FS.Remove(path)
 
 			if err != os.ErrNotExist && d.opts.EventListener != nil {
 				switch f.fileType {
diff --git a/db.go b/db.go
index df8a3c7..61ead4d 100644
--- a/db.go
+++ b/db.go
@@ -130,6 +130,7 @@ type Writer interface {
 //	})
 type DB struct {
 	dirname        string
+	dir            vfs.File
 	opts           *db.Options
 	cmp            db.Compare
 	equal          db.Equal
@@ -599,6 +600,8 @@ func (d *DB) Close() error {
 	d.commit.Close()
 	d.mu.closed = true
 
+	err = firstError(err, d.dir.Close())
+
 	if err == nil {
 		d.readState.val.unrefLocked()
 
@@ -791,11 +794,17 @@ func (d *DB) makeRoomForWrite(b *Batch) error {
 			recycleLogNumber := d.logRecycler.peek()
 			if recycleLogNumber > 0 {
 				recycleLogName := dbFilename(d.dirname, fileTypeLog, recycleLogNumber)
-				err = d.opts.VFS.Rename(recycleLogName, newLogName)
+				err = d.opts.FS.Rename(recycleLogName, newLogName)
+			}
+
+			if err == nil {
+				newLogFile, err = d.opts.FS.Create(newLogName)
 			}
 
 			if err == nil {
-				newLogFile, err = d.opts.VFS.Create(newLogName)
+				// TODO(peter): RocksDB delays sync of the parent directory until the
+				// first time the log is synced. Is that worthwhile?
+				err = d.dir.Sync()
 			}
 
 			if err == nil {
diff --git a/db/options.go b/db/options.go
index 9a52f20..398d66d 100644
--- a/db/options.go
+++ b/db/options.go
@@ -218,6 +218,11 @@ type Options struct {
 	// flushes, compactions, and table deletion.
 	EventListener *EventListener
 
+	// FS provides the interface for persistent file storage.
+	//
+	// The default value uses the underlying operating system's file system.
+	FS vfs.FS
+
 	// The number of files necessary to trigger an L0 compaction.
 	L0CompactionThreshold int
 
@@ -276,12 +281,11 @@ type Options struct {
 	// TableFormatRocksDBv2 which creates RocksDB compatible sstables. Use
 	// TableFormatLevelDB to create LevelDB compatible sstable which can be used
 	// by a wider range of tools and libraries.
-	TableFormat TableFormat
-
-	// VFS provides the interface for persistent file storage.
 	//
-	// The default value uses the underlying operating system's file system.
-	VFS vfs.FS
+	// TODO(peter): TableFormatLevelDB does not support all of the functionality
+	// of TableFormatRocksDBv2. We should ensure it is only used when writing an
+	// sstable directly, and not used when opening a database.
+	TableFormat TableFormat
 }
 
 // EnsureDefaults ensures that the default values for all options are set if a
@@ -342,8 +346,8 @@ func (o *Options) EnsureDefaults() *Options {
 	if o.Merger == nil {
 		o.Merger = DefaultMerger
 	}
-	if o.VFS == nil {
-		o.VFS = vfs.Default
+	if o.FS == nil {
+		o.FS = vfs.Default
 	}
 	return o
 }
diff --git a/ingest.go b/ingest.go
index b75dcfe..4ae0521 100644
--- a/ingest.go
+++ b/ingest.go
@@ -14,12 +14,12 @@ import (
 )
 
 func ingestLoad1(opts *db.Options, path string, fileNum uint64) (*fileMetadata, error) {
-	stat, err := opts.VFS.Stat(path)
+	stat, err := opts.FS.Stat(path)
 	if err != nil {
 		return nil, err
 	}
 
-	f, err := opts.VFS.Open(path)
+	f, err := opts.FS.Open(path)
 	if err != nil {
 		return nil, err
 	}
@@ -119,9 +119,9 @@ func ingestLink(
 ) error {
 	for i := range paths {
 		target := dbFilename(dirname, fileTypeTable, meta[i].fileNum)
-		err := opts.VFS.Link(paths[i], target)
+		err := opts.FS.Link(paths[i], target)
 		if err != nil {
-			if err2 := ingestCleanup(opts.VFS, dirname, meta[:i]); err2 != nil {
+			if err2 := ingestCleanup(opts.FS, dirname, meta[:i]); err2 != nil {
 				opts.Logger.Infof("ingest cleanup failed: %v", err2)
 			}
 			return err
@@ -285,6 +285,13 @@ func (d *DB) Ingest(paths []string) error {
 	if err := ingestLink(d.opts, d.dirname, paths, meta); err != nil {
 		return err
 	}
+	// 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.
+	if err := d.dir.Sync(); err != nil {
+		return err
+	}
 
 	var mem flushable
 	prepare := func() {
@@ -339,7 +346,7 @@ func (d *DB) Ingest(paths []string) error {
 	d.commit.AllocateSeqNum(prepare, apply)
 
 	if err != nil {
-		if err2 := ingestCleanup(d.opts.VFS, d.dirname, meta); err2 != nil {
+		if err2 := ingestCleanup(d.opts.FS, d.dirname, meta); err2 != nil {
 			d.opts.Logger.Infof("ingest cleanup failed: %v", err2)
 		}
 	}
@@ -382,7 +389,7 @@ func (d *DB) ingestApply(meta []*fileMetadata) (*versionEdit, error) {
 		ve.newFiles[i].level = ingestTargetLevel(d.cmp, current, m)
 		ve.newFiles[i].meta = *m
 	}
-	if err := d.mu.versions.logAndApply(ve); err != nil {
+	if err := d.mu.versions.logAndApply(ve, d.dir); err != nil {
 		return nil, err
 	}
 	d.updateReadStateLocked()
diff --git a/open.go b/open.go
index add7583..252e5e2 100644
--- a/open.go
+++ b/open.go
@@ -26,13 +26,13 @@ func createDB(dirname string, opts *db.Options) (retErr error) {
 		nextFileNumber: manifestFileNum + 1,
 	}
 	manifestFilename := dbFilename(dirname, fileTypeManifest, manifestFileNum)
-	f, err := opts.VFS.Create(manifestFilename)
+	f, err := opts.FS.Create(manifestFilename)
 	if err != nil {
 		return fmt.Errorf("pebble: could not create %q: %v", manifestFilename, err)
 	}
 	defer func() {
 		if retErr != nil {
-			opts.VFS.Remove(manifestFilename)
+			opts.FS.Remove(manifestFilename)
 		}
 	}()
 	defer f.Close()
@@ -50,7 +50,7 @@ func createDB(dirname string, opts *db.Options) (retErr error) {
 	if err != nil {
 		return err
 	}
-	return setCurrentFile(dirname, opts.VFS, manifestFileNum)
+	return setCurrentFile(dirname, opts.FS, manifestFileNum)
 }
 
 // Open opens a LevelDB whose files live in the given directory.
@@ -72,7 +72,7 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 	if tableCacheSize < minTableCacheSize {
 		tableCacheSize = minTableCacheSize
 	}
-	d.tableCache.init(dirname, opts.VFS, d.opts, tableCacheSize)
+	d.tableCache.init(dirname, opts.FS, d.opts, tableCacheSize)
 	d.newIters = d.tableCache.newIters
 	d.commit = newCommitPipeline(commitEnv{
 		logSeqNum:     &d.mu.versions.logSeqNum,
@@ -95,11 +95,15 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 	defer d.mu.Unlock()
 
 	// Lock the database directory.
-	err := opts.VFS.MkdirAll(dirname, 0755)
+	err := opts.FS.MkdirAll(dirname, 0755)
 	if err != nil {
 		return nil, err
 	}
-	fileLock, err := opts.VFS.Lock(dbFilename(dirname, fileTypeLock, 0))
+	d.dir, err = opts.FS.OpenDir(dirname)
+	if err != nil {
+		return nil, err
+	}
+	fileLock, err := opts.FS.Lock(dbFilename(dirname, fileTypeLock, 0))
 	if err != nil {
 		return nil, err
 	}
@@ -109,11 +113,14 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 		}
 	}()
 
-	if _, err := opts.VFS.Stat(dbFilename(dirname, fileTypeCurrent, 0)); os.IsNotExist(err) {
+	if _, err := opts.FS.Stat(dbFilename(dirname, fileTypeCurrent, 0)); os.IsNotExist(err) {
 		// Create the DB if it did not already exist.
 		if err := createDB(dirname, opts); err != nil {
 			return nil, err
 		}
+		if err := d.dir.Sync(); err != nil {
+			return nil, err
+		}
 	} else if err != nil {
 		return nil, fmt.Errorf("pebble: database %q: %v", dirname, err)
 	} else if opts.ErrorIfDBExists {
@@ -126,7 +133,7 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 		return nil, err
 	}
 
-	ls, err := opts.VFS.List(dirname)
+	ls, err := opts.FS.List(dirname)
 	if err != nil {
 		return nil, err
 	}
@@ -158,7 +165,7 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 	})
 	var ve versionEdit
 	for _, lf := range logFiles {
-		maxSeqNum, err := d.replayWAL(&ve, opts.VFS, filepath.Join(dirname, lf.name), lf.num)
+		maxSeqNum, err := d.replayWAL(&ve, opts.FS, filepath.Join(dirname, lf.name), lf.num)
 		if err != nil {
 			return nil, err
 		}
@@ -172,10 +179,13 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 	// Create an empty .log file.
 	ve.logNumber = d.mu.versions.nextFileNum()
 	d.mu.log.queue = append(d.mu.log.queue, ve.logNumber)
-	logFile, err := opts.VFS.Create(dbFilename(dirname, fileTypeLog, ve.logNumber))
+	logFile, err := opts.FS.Create(dbFilename(dirname, fileTypeLog, ve.logNumber))
 	if err != nil {
 		return nil, err
 	}
+	if err := d.dir.Sync(); err != nil {
+		return nil, err
+	}
 	logFile = vfs.NewSyncingFile(logFile, vfs.SyncingFileOptions{
 		BytesPerSync:    d.opts.BytesPerSync,
 		PreallocateSize: d.walPreallocateSize(),
@@ -183,14 +193,14 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 	d.mu.log.LogWriter = record.NewLogWriter(logFile, ve.logNumber)
 
 	// Write a new manifest to disk.
-	if err := d.mu.versions.logAndApply(&ve); err != nil {
+	if err := d.mu.versions.logAndApply(&ve, d.dir); err != nil {
 		return nil, err
 	}
 	d.updateReadStateLocked()
 
 	// Write the current options to disk.
 	d.optionsFileNum = d.mu.versions.nextFileNum()
-	optionsFile, err := opts.VFS.Create(dbFilename(dirname, fileTypeOptions, d.optionsFileNum))
+	optionsFile, err := opts.FS.Create(dbFilename(dirname, fileTypeOptions, d.optionsFileNum))
 	if err != nil {
 		return nil, err
 	}
@@ -198,6 +208,9 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 		return nil, err
 	}
 	optionsFile.Close()
+	if err := d.dir.Sync(); err != nil {
+		return nil, err
+	}
 
 	jobID := d.mu.nextJobID
 	d.mu.nextJobID++
@@ -284,7 +297,7 @@ func (d *DB) replayWAL(
 	}
 
 	if mem != nil && !mem.empty() {
-		meta, err := d.writeLevel0Table(fs, mem.newIter(nil),
+		meta, err := d.writeLevel0Table(mem.newIter(nil),
 			true /* allowRangeTombstoneElision */)
 		if err != nil {
 			return 0, err
@@ -300,7 +313,7 @@ func (d *DB) replayWAL(
 }
 
 func checkOptions(opts *db.Options, path string) error {
-	f, err := opts.VFS.Open(path)
+	f, err := opts.FS.Open(path)
 	if err != nil {
 		return err
 	}
diff --git a/version_set.go b/version_set.go
index 7c77d80..a0df50a 100644
--- a/version_set.go
+++ b/version_set.go
@@ -59,7 +59,7 @@ func (vs *versionSet) load(dirname string, opts *db.Options, mu *sync.Mutex) err
 	vs.versions.mu = mu
 	vs.writerCond.L = mu
 	vs.opts = opts
-	vs.fs = opts.VFS
+	vs.fs = opts.FS
 	vs.cmp = opts.Comparer.Compare
 	vs.cmpName = opts.Comparer.Name
 	vs.versions.init()
@@ -157,7 +157,7 @@ func (vs *versionSet) load(dirname string, opts *db.Options, mu *sync.Mutex) err
 // to the current version, and installs the new version. DB.mu must be held
 // when calling this method and will be released temporarily while performing
 // file I/O.
-func (vs *versionSet) logAndApply(ve *versionEdit) error {
+func (vs *versionSet) logAndApply(ve *versionEdit, dir vfs.File) error {
 	// Wait for any existing writing to the manifest to complete, then mark the
 	// manifest as busy.
 	for vs.writing {
@@ -223,6 +223,9 @@ func (vs *versionSet) logAndApply(ve *versionEdit) error {
 			if err := setCurrentFile(vs.dirname, vs.fs, newManifestFileNumber); err != nil {
 				return err
 			}
+			if err := dir.Sync(); err != nil {
+				return err
+			}
 		}
 		picker = newCompactionPicker(newVersion, vs.opts)
 		return nil
diff --git a/vfs/mem_fs.go b/vfs/mem_fs.go
index 203451b..7ddc89b 100644
--- a/vfs/mem_fs.go
+++ b/vfs/mem_fs.go
@@ -27,8 +27,8 @@ func (nopCloser) Close() error {
 // NewMem returns a new memory-backed FS implementation.
 func NewMem() FS {
 	return &memFS{
-		root: &node{
-			children: make(map[string]*node),
+		root: &memNode{
+			children: make(map[string]*memNode),
 			isDir:    true,
 		},
 	}
@@ -37,7 +37,7 @@ func NewMem() FS {
 // memFS implements FS.
 type memFS struct {
 	mu   sync.Mutex
-	root *node
+	root *memNode
 }
 
 func (y *memFS) String() string {
@@ -65,7 +65,7 @@ func (y *memFS) String() string {
 //   - "/", "y", false
 //   - "/y/", "z", false
 //   - "/y/z/", "", true
-func (y *memFS) walk(fullname string, f func(dir *node, frag string, final bool) error) error {
+func (y *memFS) walk(fullname string, f func(dir *memNode, frag string, final bool) error) error {
 	y.mu.Lock()
 	defer y.mu.Unlock()
 
@@ -106,15 +106,15 @@ func (y *memFS) walk(fullname string, f func(dir *node, frag string, final bool)
 }
 
 func (y *memFS) Create(fullname string) (File, error) {
-	var ret *file
-	err := y.walk(fullname, func(dir *node, frag string, final bool) error {
+	var ret *memFile
+	err := y.walk(fullname, func(dir *memNode, frag string, final bool) error {
 		if final {
 			if frag == "" {
 				return errors.New("pebble/vfs: empty file name")
 			}
-			n := &node{name: frag}
+			n := &memNode{name: frag}
 			dir.children[frag] = n
-			ret = &file{
+			ret = &memFile{
 				n:     n,
 				write: true,
 			}
@@ -128,8 +128,8 @@ func (y *memFS) Create(fullname string) (File, error) {
 }
 
 func (y *memFS) Link(oldname, newname string) error {
-	var n *node
-	err := y.walk(oldname, func(dir *node, frag string, final bool) error {
+	var n *memNode
+	err := y.walk(oldname, func(dir *memNode, frag string, final bool) error {
 		if final {
 			if frag == "" {
 				return errors.New("pebble/vfs: empty file name")
@@ -144,7 +144,7 @@ func (y *memFS) Link(oldname, newname string) error {
 	if n == nil {
 		return errors.New("pebble/vfs: no such file or directory")
 	}
-	return y.walk(newname, func(dir *node, frag string, final bool) error {
+	return y.walk(newname, func(dir *memNode, frag string, final bool) error {
 		if final {
 			if frag == "" {
 				return errors.New("pebble/vfs: empty file name")
@@ -155,15 +155,21 @@ func (y *memFS) Link(oldname, newname string) error {
 	})
 }
 
-func (y *memFS) Open(fullname string) (File, error) {
-	var ret *file
-	err := y.walk(fullname, func(dir *node, frag string, final bool) error {
+func (y *memFS) open(fullname string, allowEmptyName bool) (File, error) {
+	var ret *memFile
+	err := y.walk(fullname, func(dir *memNode, frag string, final bool) error {
 		if final {
 			if frag == "" {
-				return errors.New("pebble/vfs: empty file name")
+				if !allowEmptyName {
+					return errors.New("pebble/vfs: empty file name")
+				}
+				ret = &memFile{
+					n: dir,
+				}
+				return nil
 			}
 			if n := dir.children[frag]; n != nil {
-				ret = &file{
+				ret = &memFile{
 					n:    n,
 					read: true,
 				}
@@ -184,8 +190,16 @@ func (y *memFS) Open(fullname string) (File, error) {
 	return ret, nil
 }
 
+func (y *memFS) Open(fullname string) (File, error) {
+	return y.open(fullname, false /* allowEmptyName */)
+}
+
+func (y *memFS) OpenDir(fullname string) (File, error) {
+	return y.open(fullname, true /* allowEmptyName */)
+}
+
 func (y *memFS) Remove(fullname string) error {
-	return y.walk(fullname, func(dir *node, frag string, final bool) error {
+	return y.walk(fullname, func(dir *memNode, frag string, final bool) error {
 		if final {
 			if frag == "" {
 				return errors.New("pebble/vfs: empty file name")
@@ -201,8 +215,8 @@ func (y *memFS) Remove(fullname string) error {
 }
 
 func (y *memFS) Rename(oldname, newname string) error {
-	var n *node
-	err := y.walk(oldname, func(dir *node, frag string, final bool) error {
+	var n *memNode
+	err := y.walk(oldname, func(dir *memNode, frag string, final bool) error {
 		if final {
 			if frag == "" {
 				return errors.New("pebble/vfs: empty file name")
@@ -218,7 +232,7 @@ func (y *memFS) Rename(oldname, newname string) error {
 	if n == nil {
 		return errors.New("pebble/vfs: no such file or directory")
 	}
-	return y.walk(newname, func(dir *node, frag string, final bool) error {
+	return y.walk(newname, func(dir *memNode, frag string, final bool) error {
 		if final {
 			if frag == "" {
 				return errors.New("pebble/vfs: empty file name")
@@ -230,7 +244,7 @@ func (y *memFS) Rename(oldname, newname string) error {
 }
 
 func (y *memFS) MkdirAll(dirname string, perm os.FileMode) error {
-	return y.walk(dirname, func(dir *node, frag string, final bool) error {
+	return y.walk(dirname, func(dir *memNode, frag string, final bool) error {
 		if frag == "" {
 			if final {
 				return nil
@@ -239,9 +253,9 @@ func (y *memFS) MkdirAll(dirname string, perm os.FileMode) error {
 		}
 		child := dir.children[frag]
 		if child == nil {
-			dir.children[frag] = &node{
+			dir.children[frag] = &memNode{
 				name:     frag,
-				children: make(map[string]*node),
+				children: make(map[string]*memNode),
 				isDir:    true,
 			}
 			return nil
@@ -264,7 +278,7 @@ func (y *memFS) List(dirname string) ([]string, error) {
 		dirname += sep
 	}
 	var ret []string
-	err := y.walk(dirname, func(dir *node, frag string, final bool) error {
+	err := y.walk(dirname, func(dir *memNode, frag string, final bool) error {
 		if final {
 			if frag != "" {
 				panic("unreachable")
@@ -291,43 +305,43 @@ func (y *memFS) Stat(name string) (os.FileInfo, error) {
 	return f.Stat()
 }
 
-// node holds a file's data or a directory's children, and implements os.FileInfo.
-type node struct {
+// memNode holds a file's data or a directory's children, and implements os.FileInfo.
+type memNode struct {
 	name     string
 	data     []byte
 	modTime  time.Time
-	children map[string]*node
+	children map[string]*memNode
 	isDir    bool
 }
 
-func (f *node) IsDir() bool {
+func (f *memNode) IsDir() bool {
 	return f.isDir
 }
 
-func (f *node) ModTime() time.Time {
+func (f *memNode) ModTime() time.Time {
 	return f.modTime
 }
 
-func (f *node) Mode() os.FileMode {
+func (f *memNode) Mode() os.FileMode {
 	if f.isDir {
 		return os.ModeDir | 0755
 	}
 	return 0755
 }
 
-func (f *node) Name() string {
+func (f *memNode) Name() string {
 	return f.name
 }
 
-func (f *node) Size() int64 {
+func (f *memNode) Size() int64 {
 	return int64(len(f.data))
 }
 
-func (f *node) Sys() interface{} {
+func (f *memNode) Sys() interface{} {
 	return nil
 }
 
-func (f *node) dump(w *bytes.Buffer, level int) {
+func (f *memNode) dump(w *bytes.Buffer, level int) {
 	if f.isDir {
 		w.WriteString("          ")
 	} else {
@@ -353,18 +367,18 @@ func (f *node) dump(w *bytes.Buffer, level int) {
 	}
 }
 
-// file is a reader or writer of a node's data, and implements File.
-type file struct {
-	n           *node
+// memFile is a reader or writer of a node's data, and implements File.
+type memFile struct {
+	n           *memNode
 	rpos        int
 	read, write bool
 }
 
-func (f *file) Close() error {
+func (f *memFile) Close() error {
 	return nil
 }
 
-func (f *file) Read(p []byte) (int, error) {
+func (f *memFile) Read(p []byte) (int, error) {
 	if !f.read {
 		return 0, errors.New("pebble/vfs: file was not opened for reading")
 	}
@@ -379,7 +393,7 @@ func (f *file) Read(p []byte) (int, error) {
 	return n, nil
 }
 
-func (f *file) ReadAt(p []byte, off int64) (int, error) {
+func (f *memFile) ReadAt(p []byte, off int64) (int, error) {
 	if !f.read {
 		return 0, errors.New("pebble/vfs: file was not opened for reading")
 	}
@@ -392,7 +406,7 @@ func (f *file) ReadAt(p []byte, off int64) (int, error) {
 	return copy(p, f.n.data[off:]), nil
 }
 
-func (f *file) Write(p []byte) (int, error) {
+func (f *memFile) Write(p []byte) (int, error) {
 	if !f.write {
 		return 0, errors.New("pebble/vfs: file was not created for writing")
 	}
@@ -404,10 +418,10 @@ func (f *file) Write(p []byte) (int, error) {
 	return len(p), nil
 }
 
-func (f *file) Stat() (os.FileInfo, error) {
+func (f *memFile) Stat() (os.FileInfo, error) {
 	return f.n, nil
 }
 
-func (f *file) Sync() error {
+func (f *memFile) Sync() error {
 	return nil
 }
diff --git a/vfs/vfs.go b/vfs/vfs.go
index e1b744e..831b811 100644
--- a/vfs/vfs.go
+++ b/vfs/vfs.go
@@ -7,6 +7,7 @@ package vfs
 import (
 	"io"
 	"os"
+	"syscall"
 )
 
 // File is a readable, writable sequence of bytes.
@@ -37,6 +38,9 @@ type FS interface {
 	// Open opens the named file for reading.
 	Open(name string) (File, error)
 
+	// OpenDir opens the named directory for syncing.
+	OpenDir(name string) (File, error)
+
 	// Remove removes the named file or directory.
 	Remove(name string) error
 
@@ -84,7 +88,7 @@ var Default FS = defaultFS{}
 type defaultFS struct{}
 
 func (defaultFS) Create(name string) (File, error) {
-	return os.OpenFile(name, os.O_RDWR|os.O_CREATE, 0666)
+	return os.OpenFile(name, os.O_RDWR|os.O_CREATE|syscall.O_CLOEXEC, 0666)
 }
 
 func (defaultFS) Link(oldname, newname string) error {
@@ -92,7 +96,11 @@ func (defaultFS) Link(oldname, newname string) error {
 }
 
 func (defaultFS) Open(name string) (File, error) {
-	return os.Open(name)
+	return os.OpenFile(name, os.O_RDONLY|syscall.O_CLOEXEC, 0)
+}
+
+func (defaultFS) OpenDir(name string) (File, error) {
+	return os.OpenFile(name, syscall.O_CLOEXEC, 0)
 }
 
 func (defaultFS) Remove(name string) error {


Test patch:
diff --git a/batch_test.go b/batch_test.go
index 09fe9a3..9d179dc 100644
--- a/batch_test.go
+++ b/batch_test.go
@@ -115,7 +115,7 @@ func TestBatchGet(t *testing.T) {
 	for _, method := range []string{"build", "apply"} {
 		t.Run(method, func(t *testing.T) {
 			d, err := Open("", &db.Options{
-				VFS: vfs.NewMem(),
+				FS: vfs.NewMem(),
 			})
 			if err != nil {
 				t.Fatalf("Open: %v", err)
diff --git a/compaction_test.go b/compaction_test.go
index 3c9858a..1059fe0 100644
--- a/compaction_test.go
+++ b/compaction_test.go
@@ -581,7 +581,7 @@ func TestCompaction(t *testing.T) {
 
 	mem := vfs.NewMem()
 	d, err := Open("", &db.Options{
-		VFS:          mem,
+		FS:           mem,
 		MemTableSize: memTableSize,
 	})
 	if err != nil {
@@ -694,7 +694,7 @@ func TestManualCompaction(t *testing.T) {
 	}
 
 	d, err := Open("", &db.Options{
-		VFS: mem,
+		FS: mem,
 	})
 	if err != nil {
 		t.Fatal(err)
diff --git a/data_test.go b/data_test.go
index c2500f3..afe7aa0 100644
--- a/data_test.go
+++ b/data_test.go
@@ -192,7 +192,7 @@ func runDBDefineCmd(td *datadriven.TestData) (*DB, error) {
 	}
 
 	opts := db.Options{
-		VFS: vfs.NewMem(),
+		FS: vfs.NewMem(),
 	}
 	var snapshots []uint64
 	for _, arg := range td.CmdArgs {
@@ -247,8 +247,7 @@ func runDBDefineCmd(td *datadriven.TestData) (*DB, error) {
 		if rangeDelIter := mem.newRangeDelIter(nil); rangeDelIter != nil {
 			iter = newMergingIter(d.cmp, iter, rangeDelIter)
 		}
-		meta, err := d.writeLevel0Table(d.opts.VFS, iter,
-			false /* allowRangeTombstoneElision */)
+		meta, err := d.writeLevel0Table(iter, false /* allowRangeTombstoneElision */)
 		if err != nil {
 			return nil
 		}
@@ -304,7 +303,7 @@ func runDBDefineCmd(td *datadriven.TestData) (*DB, error) {
 	}
 
 	if len(ve.newFiles) > 0 {
-		if err := d.mu.versions.logAndApply(ve); err != nil {
+		if err := d.mu.versions.logAndApply(ve, d.dir); err != nil {
 			return nil, err
 		}
 		d.updateReadStateLocked()
diff --git a/db_test.go b/db_test.go
index 23b6e3b..aa2fa57 100644
--- a/db_test.go
+++ b/db_test.go
@@ -197,7 +197,7 @@ func TestBasicReads(t *testing.T) {
 			continue
 		}
 		d, err := Open("", &db.Options{
-			VFS: fs,
+			FS: fs,
 		})
 		if err != nil {
 			t.Errorf("%s: Open failed: %v", tc.dirname, err)
@@ -224,7 +224,7 @@ func TestBasicReads(t *testing.T) {
 
 func TestBasicWrites(t *testing.T) {
 	d, err := Open("", &db.Options{
-		VFS: vfs.NewMem(),
+		FS: vfs.NewMem(),
 	})
 	if err != nil {
 		t.Fatal(err)
@@ -363,7 +363,7 @@ func TestBasicWrites(t *testing.T) {
 
 func TestRandomWrites(t *testing.T) {
 	d, err := Open("", &db.Options{
-		VFS:          vfs.NewMem(),
+		FS:           vfs.NewMem(),
 		MemTableSize: 8 * 1024,
 	})
 	if err != nil {
@@ -419,7 +419,7 @@ func TestRandomWrites(t *testing.T) {
 
 func TestLargeBatch(t *testing.T) {
 	d, err := Open("", &db.Options{
-		VFS:                         vfs.NewMem(),
+		FS:                          vfs.NewMem(),
 		MemTableSize:                1400,
 		MemTableStopWritesThreshold: 100,
 	})
@@ -470,7 +470,7 @@ func TestLargeBatch(t *testing.T) {
 
 func TestGetMerge(t *testing.T) {
 	d, err := Open("", &db.Options{
-		VFS: vfs.NewMem(),
+		FS: vfs.NewMem(),
 	})
 	if err != nil {
 		t.Fatal(err)
@@ -512,7 +512,7 @@ func TestIterLeak(t *testing.T) {
 			for _, flush := range []bool{true, false} {
 				t.Run(fmt.Sprintf("flush=%t", flush), func(t *testing.T) {
 					d, err := Open("", &db.Options{
-						VFS: vfs.NewMem(),
+						FS: vfs.NewMem(),
 					})
 					if err != nil {
 						t.Fatal(err)
@@ -554,7 +554,7 @@ func TestCacheEvict(t *testing.T) {
 	cache := cache.New(10 << 20)
 	d, err := Open("", &db.Options{
 		Cache: cache,
-		VFS:   vfs.NewMem(),
+		FS:    vfs.NewMem(),
 	})
 	if err != nil {
 		t.Fatal(err)
@@ -602,7 +602,7 @@ func TestCacheEvict(t *testing.T) {
 
 func TestFlushEmpty(t *testing.T) {
 	d, err := Open("", &db.Options{
-		VFS: vfs.NewMem(),
+		FS: vfs.NewMem(),
 	})
 	if err != nil {
 		t.Fatal(err)
@@ -619,7 +619,7 @@ func TestFlushEmpty(t *testing.T) {
 func TestRollManifest(t *testing.T) {
 	d, err := Open("", &db.Options{
 		MaxManifestFileSize: 1,
-		VFS:                 vfs.NewMem(),
+		FS:                  vfs.NewMem(),
 	})
 	if err != nil {
 		t.Fatal(err)
@@ -632,7 +632,7 @@ func TestRollManifest(t *testing.T) {
 	}
 
 	current := func() string {
-		f, err := d.opts.VFS.Open(dbFilename(d.dirname, fileTypeCurrent, 0))
+		f, err := d.opts.FS.Open(dbFilename(d.dirname, fileTypeCurrent, 0))
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -669,7 +669,7 @@ func TestRollManifest(t *testing.T) {
 		}
 	}
 
-	files, err := d.opts.VFS.List("")
+	files, err := d.opts.FS.List("")
 	if err != nil {
 		t.Fatal(err)
 	}
diff --git a/event_listener_test.go b/event_listener_test.go
index 4fb56be..13c8368 100644
--- a/event_listener_test.go
+++ b/event_listener_test.go
@@ -7,10 +7,13 @@ package pebble
 import (
 	"bytes"
 	"fmt"
+	"io"
 	"sync"
 	"testing"
 
 	"github.com/petermattis/pebble/db"
+	"github.com/petermattis/pebble/internal/datadriven"
+	"github.com/petermattis/pebble/sstable"
 	"github.com/petermattis/pebble/vfs"
 )
 
@@ -19,6 +22,12 @@ type syncedBuffer struct {
 	buf bytes.Buffer
 }
 
+func (b *syncedBuffer) Reset() {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.buf.Reset()
+}
+
 func (b *syncedBuffer) Write(p []byte) (n int, err error) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
@@ -31,78 +40,150 @@ func (b *syncedBuffer) String() string {
 	return b.buf.String()
 }
 
-func TestEventListener(t *testing.T) {
-	var buf syncedBuffer
+type loggingFS struct {
+	vfs.FS
+	w io.Writer
+}
 
-	d, err := Open("", &db.Options{
-		VFS: vfs.NewMem(),
-		EventListener: &db.EventListener{
-			CompactionBegin: func(info db.CompactionInfo) {
-				fmt.Fprintf(&buf, "#%d: compaction begin: L%d -> L%d\n", info.JobID,
-					info.Input.Level, info.Input.Level+1)
-			},
-			CompactionEnd: func(info db.CompactionInfo) {
-				fmt.Fprintf(&buf, "#%d: compaction end: L%d -> L%d\n", info.JobID,
-					info.Input.Level, info.Input.Level+1)
-			},
-			FlushBegin: func(info db.FlushInfo) {
-				fmt.Fprintf(&buf, "#%d: flush begin\n", info.JobID)
-			},
-			FlushEnd: func(info db.FlushInfo) {
-				fmt.Fprintf(&buf, "#%d: flush end: %d\n", info.JobID, info.Output.FileNum)
-			},
-			TableDeleted: func(info db.TableDeleteInfo) {
-				fmt.Fprintf(&buf, "#%d: table deleted: %d\n", info.JobID, info.FileNum)
-			},
-			TableIngested: func(info db.TableIngestInfo) {
-				fmt.Fprintf(&buf, "#%d: table ingested\n", info.JobID)
-			},
-			WALCreated: func(info db.WALCreateInfo) {
-				fmt.Fprintf(&buf, "#%d: WAL created: %d recycled=%d\n",
-					info.JobID, info.FileNum, info.RecycledFileNum)
-			},
-			WALDeleted: func(info db.WALDeleteInfo) {
-				fmt.Fprintf(&buf, "#%d: WAL deleted: %d\n", info.JobID, info.FileNum)
-			},
-		},
-	})
+func (fs loggingFS) Create(name string) (vfs.File, error) {
+	fmt.Fprintf(fs.w, "create: %s\n", name)
+	f, err := fs.FS.Create(name)
 	if err != nil {
-		t.Fatal(err)
+		return nil, err
 	}
+	return loggingFile{f, name, fs.w}, nil
+}
 
-	if err := d.Set([]byte("a"), nil, nil); err != nil {
-		t.Fatal(err)
-	}
-	if err := d.Flush(); err != nil {
-		t.Fatal(err)
-	}
-	if err := d.Compact([]byte("a"), []byte("b")); err != nil {
-		t.Fatal(err)
-	}
-	if err := d.Delete([]byte("a"), nil); err != nil {
-		t.Fatal(err)
+func (fs loggingFS) Link(oldname, newname string) error {
+	fmt.Fprintf(fs.w, "link: %s -> %s\n", oldname, newname)
+	return fs.FS.Link(oldname, newname)
+}
+
+func (fs loggingFS) OpenDir(name string) (vfs.File, error) {
+	fmt.Fprintf(fs.w, "open-dir: %s\n", name)
+	f, err := fs.FS.OpenDir(name)
+	if err != nil {
+		return nil, err
 	}
-	if err := d.Compact([]byte("a"), []byte("b")); err != nil {
-		t.Fatal(err)
+	return loggingFile{f, name, fs.w}, nil
+}
+
+func (fs loggingFS) Rename(oldname, newname string) error {
+	fmt.Fprintf(fs.w, "rename: %s -> %s\n", oldname, newname)
+	return fs.FS.Rename(oldname, newname)
+}
+
+type loggingFile struct {
+	vfs.File
+	name string
+	w    io.Writer
+}
+
+func (f loggingFile) Sync() error {
+	fmt.Fprintf(f.w, "sync: %s\n", f.name)
+	return f.File.Sync()
+}
+
+func newLoggingEventListener(w io.Writer) *db.EventListener {
+	return &db.EventListener{
+		CompactionBegin: func(info db.CompactionInfo) {
+			fmt.Fprintf(w, "#%d: compaction begin: L%d -> L%d\n", info.JobID,
+				info.Input.Level, info.Input.Level+1)
+		},
+		CompactionEnd: func(info db.CompactionInfo) {
+			fmt.Fprintf(w, "#%d: compaction end: L%d -> L%d\n", info.JobID,
+				info.Input.Level, info.Input.Level+1)
+		},
+		FlushBegin: func(info db.FlushInfo) {
+			fmt.Fprintf(w, "#%d: flush begin\n", info.JobID)
+		},
+		FlushEnd: func(info db.FlushInfo) {
+			fmt.Fprintf(w, "#%d: flush end: %d\n", info.JobID, info.Output.FileNum)
+		},
+		TableDeleted: func(info db.TableDeleteInfo) {
+			fmt.Fprintf(w, "#%d: table deleted: %d\n", info.JobID, info.FileNum)
+		},
+		TableIngested: func(info db.TableIngestInfo) {
+			fmt.Fprintf(w, "#%d: table ingested\n", info.JobID)
+		},
+		WALCreated: func(info db.WALCreateInfo) {
+			fmt.Fprintf(w, "#%d: WAL created: %d recycled=%d\n",
+				info.JobID, info.FileNum, info.RecycledFileNum)
+		},
+		WALDeleted: func(info db.WALDeleteInfo) {
+			fmt.Fprintf(w, "#%d: WAL deleted: %d\n", info.JobID, info.FileNum)
+		},
 	}
-	if err := d.Close(); err != nil {
+}
+
+// Verify event listener actions, as well as expected filesystem operations.
+func TestEventListener(t *testing.T) {
+	var d *DB
+	var buf syncedBuffer
+	mem := vfs.NewMem()
+	err := mem.MkdirAll("ext", 0755)
+	if err != nil {
 		t.Fatal(err)
 	}
 
-	expected := `#2: WAL created: 5 recycled=0
-#3: flush begin
-#3: flush end: 6
-#4: compaction begin: L0 -> L1
-#4: compaction end: L0 -> L1
-#5: WAL created: 7 recycled=2
-#6: flush begin
-#6: flush end: 8
-#7: compaction begin: L0 -> L1
-#7: compaction end: L0 -> L1
-#7: table deleted: 6
-#7: table deleted: 8
-`
-	if v := buf.String(); expected != v {
-		t.Fatalf("expected\n%s\nbut found\n%s", expected, v)
-	}
+	datadriven.RunTest(t, "testdata/event_listener", func(td *datadriven.TestData) string {
+		switch td.Cmd {
+		case "open":
+			buf.Reset()
+			var err error
+			d, err = Open("db", &db.Options{
+				FS:                  loggingFS{mem, &buf},
+				EventListener:       newLoggingEventListener(&buf),
+				MaxManifestFileSize: 1,
+			})
+			if err != nil {
+				return err.Error()
+			}
+			return buf.String()
+
+		case "flush":
+			buf.Reset()
+			if err := d.Set([]byte("a"), nil, nil); err != nil {
+				return err.Error()
+			}
+			if err := d.Flush(); err != nil {
+				return err.Error()
+			}
+			return buf.String()
+
+		case "compact":
+			buf.Reset()
+			if err := d.Set([]byte("a"), nil, nil); err != nil {
+				return err.Error()
+			}
+			if err := d.Compact([]byte("a"), []byte("b")); err != nil {
+				return err.Error()
+			}
+			return buf.String()
+
+		case "ingest":
+			buf.Reset()
+			f, err := mem.Create("ext/0")
+			if err != nil {
+				return err.Error()
+			}
+			w := sstable.NewWriter(f, nil, db.LevelOptions{})
+			if err := w.Add(db.MakeInternalKey([]byte("a"), 0, db.InternalKeyKindSet), nil); err != nil {
+				return err.Error()
+			}
+			if err := w.Close(); err != nil {
+				return err.Error()
+			}
+			if err := d.Ingest([]string{"ext/0"}); err != nil {
+				return err.Error()
+			}
+			if err := mem.Remove("ext/0"); err != nil {
+				return err.Error()
+			}
+			return buf.String()
+
+		default:
+			return fmt.Sprintf("unknown command: %s", td.Cmd)
+		}
+	})
 }
diff --git a/flush_test.go b/flush_test.go
index 034799e..5bfe2a2 100644
--- a/flush_test.go
+++ b/flush_test.go
@@ -16,7 +16,7 @@ import (
 
 func TestManualFlush(t *testing.T) {
 	d, err := Open("", &db.Options{
-		VFS: vfs.NewMem(),
+		FS: vfs.NewMem(),
 	})
 	if err != nil {
 		t.Fatal(err)
@@ -70,7 +70,7 @@ func TestManualFlush(t *testing.T) {
 
 		case "reset":
 			d, err = Open("", &db.Options{
-				VFS: vfs.NewMem(),
+				FS: vfs.NewMem(),
 			})
 			if err != nil {
 				return err.Error()
diff --git a/ingest_test.go b/ingest_test.go
index a787333..8dd44d9 100644
--- a/ingest_test.go
+++ b/ingest_test.go
@@ -49,7 +49,7 @@ func TestIngestLoad(t *testing.T) {
 
 			opts := &db.Options{
 				Comparer: db.DefaultComparer,
-				VFS:      mem,
+				FS:       mem,
 			}
 			meta, err := ingestLoad(opts, []string{"ext"}, []uint64{1})
 			if err != nil {
@@ -129,7 +129,7 @@ func TestIngestLoadRand(t *testing.T) {
 
 	opts := &db.Options{
 		Comparer: db.DefaultComparer,
-		VFS:      mem,
+		FS:       mem,
 	}
 	meta, err := ingestLoad(opts, paths, pending)
 	if err != nil {
@@ -143,7 +143,7 @@ func TestIngestLoadRand(t *testing.T) {
 func TestIngestLoadNonExistent(t *testing.T) {
 	opts := &db.Options{
 		Comparer: db.DefaultComparer,
-		VFS:      vfs.NewMem(),
+		FS:       vfs.NewMem(),
 	}
 	if _, err := ingestLoad(opts, []string{"non-existent"}, []uint64{1}); err == nil {
 		t.Fatalf("expected error, but found success")
@@ -160,7 +160,7 @@ func TestIngestLoadEmpty(t *testing.T) {
 
 	opts := &db.Options{
 		Comparer: db.DefaultComparer,
-		VFS:      mem,
+		FS:       mem,
 	}
 	if _, err := ingestLoad(opts, []string{"empty"}, []uint64{1}); err == nil {
 		t.Fatalf("expected error, but found success")
@@ -245,7 +245,7 @@ func TestIngestLink(t *testing.T) {
 	for i := 0; i <= count; i++ {
 		t.Run("", func(t *testing.T) {
 			mem := vfs.NewMem()
-			opts := &db.Options{VFS: mem}
+			opts := &db.Options{FS: mem}
 			opts.EnsureDefaults()
 			if err := mem.MkdirAll(dir, 0755); err != nil {
 				t.Fatal(err)
@@ -473,7 +473,7 @@ func TestIngest(t *testing.T) {
 	}
 
 	d, err := Open("", &db.Options{
-		VFS:                   mem,
+		FS:                    mem,
 		L0CompactionThreshold: 100,
 	})
 	if err != nil {
diff --git a/log_recycler_test.go b/log_recycler_test.go
index fd28bb8..90af95e 100644
--- a/log_recycler_test.go
+++ b/log_recycler_test.go
@@ -76,7 +76,7 @@ func TestLogRecycler(t *testing.T) {
 
 func TestRecycleLogs(t *testing.T) {
 	d, err := Open("", &db.Options{
-		VFS: vfs.NewMem(),
+		FS: vfs.NewMem(),
 	})
 	if err != nil {
 		t.Fatal(err)
diff --git a/open_test.go b/open_test.go
index 82e1587..dab17b4 100644
--- a/open_test.go
+++ b/open_test.go
@@ -23,7 +23,7 @@ func TestErrorIfDBExists(t *testing.T) {
 	for _, b := range [...]bool{false, true} {
 		mem := vfs.NewMem()
 		d0, err := Open("", &db.Options{
-			VFS: mem,
+			FS: mem,
 		})
 		if err != nil {
 			t.Errorf("b=%v: d0 Open: %v", b, err)
@@ -35,7 +35,7 @@ func TestErrorIfDBExists(t *testing.T) {
 		}
 
 		d1, err := Open("", &db.Options{
-			VFS:             mem,
+			FS:              mem,
 			ErrorIfDBExists: b,
 		})
 		if d1 != nil {
@@ -52,7 +52,7 @@ func TestNewDBFilenames(t *testing.T) {
 	fooBar := filepath.Join("foo", "bar")
 	mem := vfs.NewMem()
 	d, err := Open(fooBar, &db.Options{
-		VFS: mem,
+		FS: mem,
 	})
 	if err != nil {
 		t.Fatalf("Open: %v", err)
@@ -79,7 +79,7 @@ func TestNewDBFilenames(t *testing.T) {
 
 func testOpenCloseOpenClose(t *testing.T, fs vfs.FS, root string) {
 	opts := &db.Options{
-		VFS: fs,
+		FS: fs,
 	}
 
 	for _, startFromEmpty := range []bool{false, true} {
@@ -144,7 +144,7 @@ func testOpenCloseOpenClose(t *testing.T, fs vfs.FS, root string) {
 			}
 
 			{
-				got, err := opts.VFS.List(dirname)
+				got, err := opts.FS.List(dirname)
 				if err != nil {
 					t.Fatalf("List: %v", err)
 				}
@@ -189,7 +189,7 @@ func TestOpenCloseOpenClose(t *testing.T) {
 
 func TestOpenOptionsCheck(t *testing.T) {
 	mem := vfs.NewMem()
-	opts := &db.Options{VFS: mem}
+	opts := &db.Options{FS: mem}
 
 	d, err := Open("", opts)
 	if err != nil {
@@ -201,14 +201,14 @@ func TestOpenOptionsCheck(t *testing.T) {
 
 	opts = &db.Options{
 		Comparer: &db.Comparer{Name: "foo"},
-		VFS:      mem,
+		FS:       mem,
 	}
 	_, err = Open("", opts)
 	require.Regexp(t, `comparer name from file.*!=.*`, err)
 
 	opts = &db.Options{
 		Merger: &db.Merger{Name: "bar"},
-		VFS:    mem,
+		FS:     mem,
 	}
 	_, err = Open("", opts)
 	require.Regexp(t, `merger name from file.*!=.*`, err)
diff --git a/range_del_test.go b/range_del_test.go
index bc983c1..aecaa73 100644
--- a/range_del_test.go
+++ b/range_del_test.go
@@ -115,7 +115,7 @@ func TestRangeDel(t *testing.T) {
 func TestRangeDelCompactionTruncation(t *testing.T) {
 	// Use a small target file size so that there is a single key per sstable.
 	d, err := Open("", &db.Options{
-		VFS: vfs.NewMem(),
+		FS: vfs.NewMem(),
 		Levels: []db.LevelOptions{
 			{TargetFileSize: 100},
 			{TargetFileSize: 100},
@@ -249,7 +249,7 @@ func BenchmarkRangeDelIterate(b *testing.B) {
 					mem := vfs.NewMem()
 					d, err := Open("", &db.Options{
 						Cache: cache.New(128 << 20), // 128 MB
-						VFS:   mem,
+						FS:    mem,
 					})
 					if err != nil {
 						b.Fatal(err)
diff --git a/read_state_test.go b/read_state_test.go
index 0dd9480..92c5f36 100644
--- a/read_state_test.go
+++ b/read_state_test.go
@@ -16,7 +16,7 @@ import (
 
 func BenchmarkReadState(b *testing.B) {
 	d, err := Open("", &db.Options{
-		VFS: vfs.NewMem(),
+		FS: vfs.NewMem(),
 	})
 	if err != nil {
 		b.Fatal(err)
diff --git a/snapshot_test.go b/snapshot_test.go
index 8b5f012..c5d9444 100644
--- a/snapshot_test.go
+++ b/snapshot_test.go
@@ -49,7 +49,7 @@ func TestSnapshot(t *testing.T) {
 		case "define":
 			var err error
 			d, err = Open("", &db.Options{
-				VFS: vfs.NewMem(),
+				FS: vfs.NewMem(),
 			})
 			if err != nil {
 				return err.Error()
diff --git a/testdata/event_listener b/testdata/event_listener
new file mode 100644
index 0000000..d893382
--- /dev/null
+++ b/testdata/event_listener
@@ -0,0 +1,82 @@
+open
+----
+open-dir: db
+create: db/MANIFEST-000001
+create: db/CURRENT.000001.dbtmp
+sync: db/CURRENT.000001.dbtmp
+rename: db/CURRENT.000001.dbtmp -> db/CURRENT
+sync: db
+create: db/000002.log
+sync: db
+create: db/MANIFEST-000003
+sync: db/MANIFEST-000003
+create: db/CURRENT.000003.dbtmp
+sync: db/CURRENT.000003.dbtmp
+rename: db/CURRENT.000003.dbtmp -> db/CURRENT
+sync: db
+create: db/OPTIONS-000004
+sync: db
+
+flush
+----
+sync: db/000002.log
+create: db/000005.log
+sync: db
+sync: db/000002.log
+#2: WAL created: 5 recycled=0
+#3: flush begin
+create: db/000006.sst
+sync: db/000006.sst
+sync: db
+#3: flush end: 6
+create: db/MANIFEST-000007
+sync: db/MANIFEST-000007
+create: db/CURRENT.000007.dbtmp
+sync: db/CURRENT.000007.dbtmp
+rename: db/CURRENT.000007.dbtmp -> db/CURRENT
+sync: db
+
+compact
+----
+sync: db/000005.log
+rename: db/000002.log -> db/000008.log
+create: db/000008.log
+sync: db
+sync: db/000005.log
+#4: WAL created: 8 recycled=2
+#5: flush begin
+create: db/000009.sst
+sync: db/000009.sst
+sync: db
+#5: flush end: 9
+create: db/MANIFEST-000010
+sync: db/MANIFEST-000010
+create: db/CURRENT.000010.dbtmp
+sync: db/CURRENT.000010.dbtmp
+rename: db/CURRENT.000010.dbtmp -> db/CURRENT
+sync: db
+#6: compaction begin: L0 -> L1
+create: db/000011.sst
+sync: db/000011.sst
+sync: db
+#6: compaction end: L0 -> L1
+create: db/MANIFEST-000012
+sync: db/MANIFEST-000012
+create: db/CURRENT.000012.dbtmp
+sync: db/CURRENT.000012.dbtmp
+rename: db/CURRENT.000012.dbtmp -> db/CURRENT
+sync: db
+#6: table deleted: 6
+#6: table deleted: 9
+
+ingest
+----
+link: ext/0 -> db/000013.sst
+sync: db
+create: db/MANIFEST-000014
+sync: db/MANIFEST-000014
+create: db/CURRENT.000014.dbtmp
+sync: db/CURRENT.000014.dbtmp
+rename: db/CURRENT.000014.dbtmp -> db/CURRENT
+sync: db
+#7: table ingested