db: scan and filter WALs during recovery

Move the scanning of WAL directories' files into recoverState. Semantically,
this is a part of recovery.

Author: jbowens

open.go

@@ -102,18 +102,7 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 		}
 	}
 
-	// Open the database and WAL directories first.
-	dirs, err := prepareOpenAndLockDirs(dirname, opts)
-	if err != nil {
-		err = errors.Wrapf(err, "error opening database at %q", dirname)
-		err = errors.CombineErrors(err, dirs.Close())
-		return nil, err
-	}
-	// Locks in RecoveryDirLocks can be closed as soon as we've finished opening
-	// the database.
-	defer func() { _ = dirs.RecoveryDirLocks.Close() }()
-	defer maybeCleanUp(dirs.Close)
-
+	// Recover the current database state.
 	rs, err := recoverState(opts, dirname)
 	if err != nil {
 		return nil, err
@@ -149,7 +138,7 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 	}
 
 	if !opts.ReadOnly {
-		if err := rs.RemoveObsolete(); err != nil {
+		if err := rs.RemoveObsolete(opts); err != nil {
 			return nil, err
 		}
 	}
@@ -169,7 +158,7 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 		split:               opts.Comparer.Split,
 		abbreviatedKey:      opts.Comparer.AbbreviatedKey,
 		largeBatchThreshold: (opts.MemTableSize - uint64(memTableEmptySize)) / 2,
-		dirs:                dirs,
+		dirs:                rs.dirs,
 		objProvider:         rs.objProvider,
 		closed:              new(atomic.Value),
 		closedCh:            make(chan struct{}),
@@ -297,8 +286,8 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 	})
 
 	walOpts := wal.Options{
-		Primary:              dirs.WALPrimary,
-		Secondary:            dirs.WALSecondary,
+		Primary:              rs.dirs.WALPrimary,
+		Secondary:            rs.dirs.WALSecondary,
 		MinUnflushedWALNum:   wal.NumWAL(d.mu.versions.minUnflushedLogNum),
 		MaxNumRecyclableLogs: opts.MemTableStopWritesThreshold + 1,
 		NoSyncOnClose:        opts.NoSyncOnClose,
@@ -323,45 +312,7 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 		return nil, err
 	}
 
-	// Remove obsolete WAL files now (as opposed to relying on asynchronous
-	// cleanup) to prevent crash loops due to no disk space (ENOSPC).
-	var retainedWALs wal.Logs
-	for _, w := range wals {
-		// Any WALs with file numbers ≥ minUnflushedLogNum must be replayed to
-		// recover the state.
-		if base.DiskFileNum(w.Num) >= d.mu.versions.minUnflushedLogNum {
-			retainedWALs = append(retainedWALs, w)
-			continue
-		}
-		// Skip removal of obsolete WALs in read-only mode.
-		if opts.ReadOnly {
-			continue
-		}
-		// Remove obsolete WALs, logging each removal.
-		for i := range w.NumSegments() {
-			fs, path := w.SegmentLocation(i)
-			if err := fs.Remove(path); err != nil {
-				// It's not a big deal if we can't delete the file now.
-				// We'll try to remove it later in the cleanup process.
-				d.opts.EventListener.WALDeleted(WALDeleteInfo{
-					JobID:   0,
-					Path:    path,
-					FileNum: base.DiskFileNum(w.Num),
-					Err:     err,
-				})
-				retainedWALs = append(retainedWALs, w)
-			} else {
-				d.opts.EventListener.WALDeleted(WALDeleteInfo{
-					JobID:   0,
-					Path:    path,
-					FileNum: base.DiskFileNum(w.Num),
-					Err:     nil,
-				})
-			}
-		}
-	}
-
-	walManager, err := wal.Init(walOpts, retainedWALs)
+	walManager, err := wal.Init(walOpts, rs.walsReplay)
 	if err != nil {
 		return nil, err
 	}
@@ -536,6 +487,12 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 	d.maybeScheduleFlush()
 	d.maybeScheduleCompaction()
 
+	// Locks in RecoveryDirLocks can be closed as soon as we've finished opening
+	// the database.
+	if err = rs.dirs.RecoveryDirLocks.Close(); err != nil {
+		return nil, err
+	}
+
 	// Note: this is a no-op if invariants are disabled or race is enabled.
 	//
 	// Setting a finalizer on *DB causes *DB to never be reclaimed and the

recovery.go

@@ -17,6 +17,7 @@ import (
 	"github.com/cockroachdb/pebble/record"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/cockroachdb/pebble/vfs/atomicfs"
+	"github.com/cockroachdb/pebble/wal"
 )
 
 // recoverState reads the named database directory and recovers the set of files
@@ -34,9 +35,16 @@ func recoverState(opts *Options, dirname string) (s *recoveredState, err error)
 }
 
 func (rs *recoveredState) init(opts *Options, dirname string) error {
+	dirs, err := prepareOpenAndLockDirs(dirname, opts)
+	if err != nil {
+		err = errors.Wrapf(err, "error opening database at %q", dirname)
+		err = errors.CombineErrors(err, dirs.Close())
+		return err
+	}
+	rs.dirs = dirs
+
 	// List the directory contents. This also happens to include WAL log files,
 	// if they are in the same dir.
-	var err error
 	if rs.ls, err = opts.FS.List(dirname); err != nil {
 		return errors.Wrapf(err, "pebble: database %q", dirname)
 	}
@@ -100,13 +108,27 @@ func (rs *recoveredState) init(opts *Options, dirname string) error {
 			rs.obsoleteTempFilenames = append(rs.obsoleteTempFilenames, filename)
 		}
 	}
+
+	// Find all the WAL files across the various WAL directories.
+	wals, err := wal.Scan(rs.dirs.WALDirs()...)
+	if err != nil {
+		return err
+	}
+	for _, w := range wals {
+		if rs.recoveredVersion == nil || base.DiskFileNum(w.Num) >= rs.recoveredVersion.minUnflushedLogNum {
+			rs.walsReplay = append(rs.walsReplay, w)
+		} else {
+			rs.walsObsolete = append(rs.walsObsolete, w)
+		}
+	}
 	return nil
 }
 
 // recoveredState encapsulates state recovered from reading the database
 // directory.
 type recoveredState struct {
 	dirname                 string
+	dirs                    *resolvedDirs
 	fmv                     FormatMajorVersion
 	fmvMarker               *atomicfs.Marker
 	fs                      vfs.FS
@@ -118,10 +140,12 @@ type recoveredState struct {
 	objProvider             objstorage.Provider
 	previousOptionsFilename string
 	recoveredVersion        *recoveredVersion
+	walsObsolete            wal.Logs
+	walsReplay              wal.Logs
 }
 
 // RemoveObsolete removes obsolete files uncovered during recovery.
-func (rs *recoveredState) RemoveObsolete() error {
+func (rs *recoveredState) RemoveObsolete(opts *Options) error {
 	var err error
 	// Atomic markers may leave behind obsolete files if there's a crash
 	// mid-update.
@@ -137,6 +161,20 @@ func (rs *recoveredState) RemoveObsolete() error {
 	for _, filename := range rs.obsoleteTempFilenames {
 		err = errors.CombineErrors(err, rs.fs.Remove(rs.fs.PathJoin(rs.dirname, filename)))
 	}
+	// Remove any WAL files that are already obsolete. Pebble keeps some old WAL
+	// files around for recycling.
+	for _, w := range rs.walsObsolete {
+		for i := range w.NumSegments() {
+			fs, path := w.SegmentLocation(i)
+			rmErr := fs.Remove(path)
+			opts.EventListener.WALDeleted(WALDeleteInfo{
+				JobID:   0,
+				Path:    path,
+				FileNum: base.DiskFileNum(w.Num),
+				Err:     rmErr,
+			})
+		}
+	}
 	return err
 }
 
@@ -153,6 +191,9 @@ func (rs *recoveredState) Close() error {
 	if rs.objProvider != nil {
 		err = errors.CombineErrors(err, rs.objProvider.Close())
 	}
+	if rs.dirs != nil {
+		err = errors.CombineErrors(err, rs.dirs.Close())
+	}
 	return err
 }