db: move replayWAL into recovery.go

Author: jbowens

open.go

@@ -19,8 +19,6 @@ import (
 	"github.com/cockroachdb/crlib/crtime"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/errors/oserror"
-	"github.com/cockroachdb/pebble/batchrepr"
-	"github.com/cockroachdb/pebble/internal/arenaskl"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/cache"
 	"github.com/cockroachdb/pebble/internal/inflight"
@@ -30,7 +28,6 @@ import (
 	"github.com/cockroachdb/pebble/internal/manual"
 	"github.com/cockroachdb/pebble/objstorage"
 	"github.com/cockroachdb/pebble/objstorage/remote"
-	"github.com/cockroachdb/pebble/record"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/cockroachdb/pebble/wal"
 	"github.com/prometheus/client_golang/prometheus"
@@ -753,231 +750,6 @@ func (d *DB) replayIngestedFlushable(
 	return d.newIngestedFlushableEntry(meta, seqNum, logNum, exciseSpan)
 }
 
-// replayWAL replays the edits in the specified WAL. If the DB is in read
-// only mode, then the WALs are replayed into memtables and not flushed. If
-// the DB is not in read only mode, then the contents of the WAL are
-// guaranteed to be flushed when a flush is scheduled after this method is run.
-// Note that this flushing is very important for guaranteeing durability:
-// the application may have had a number of pending
-// fsyncs to the WAL before the process crashed, and those fsyncs may not have
-// happened but the corresponding data may now be readable from the WAL (while
-// sitting in write-back caches in the kernel or the storage device). By
-// reading the WAL (including the non-fsynced data) and then flushing all
-// these changes (flush does fsyncs), we are able to guarantee that the
-// initial state of the DB is durable.
-//
-// This method mutates d.mu.mem.queue and possibly d.mu.mem.mutable and replays
-// WALs into the flushable queue. Flushing of the queue is expected to be handled
-// by callers. A list of flushable ingests (but not memtables) replayed is returned.
-//
-// 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) replayWAL(
-	jobID JobID, ll wal.LogicalLog, strictWALTail bool,
-) (flushableIngests []*ingestedFlushable, maxSeqNum base.SeqNum, err error) {
-	rr := ll.OpenForRead()
-	defer func() { _ = rr.Close() }()
-	var (
-		b               Batch
-		buf             bytes.Buffer
-		mem             *memTable
-		entry           *flushableEntry
-		offset          wal.Offset
-		lastFlushOffset int64
-		keysReplayed    int64 // number of keys replayed
-		batchesReplayed int64 // number of batches replayed
-	)
-
-	// TODO(jackson): This function is interspersed with panics, in addition to
-	// corruption error propagation. Audit them to ensure we're truly only
-	// panicking where the error points to Pebble bug and not user or
-	// hardware-induced corruption.
-
-	// "Flushes" (ie. closes off) the current memtable, if not nil.
-	flushMem := func() {
-		if mem == nil {
-			return
-		}
-		mem.writerUnref()
-		if d.mu.mem.mutable == mem {
-			d.mu.mem.mutable = nil
-		}
-		entry.flushForced = !d.opts.ReadOnly
-		var logSize uint64
-		mergedOffset := offset.Physical + offset.PreviousFilesBytes
-		if mergedOffset >= lastFlushOffset {
-			logSize = uint64(mergedOffset - lastFlushOffset)
-		}
-		// Else, this was the initial memtable in the read-only case which must have
-		// been empty, but we need to flush it since we don't want to add to it later.
-		lastFlushOffset = mergedOffset
-		entry.logSize = logSize
-		mem, entry = nil, nil
-	}
-
-	mem = d.mu.mem.mutable
-	if mem != nil {
-		entry = d.mu.mem.queue[len(d.mu.mem.queue)-1]
-		if !d.opts.ReadOnly {
-			flushMem()
-		}
-	}
-
-	// Creates a new memtable if there is no current memtable.
-	ensureMem := func(seqNum base.SeqNum) {
-		if mem != nil {
-			return
-		}
-		mem, entry = d.newMemTable(base.DiskFileNum(ll.Num), seqNum, 0 /* minSize */)
-		d.mu.mem.mutable = mem
-		d.mu.mem.queue = append(d.mu.mem.queue, entry)
-	}
-
-	defer func() {
-		if err != nil {
-			err = errors.WithDetailf(err, "replaying wal %d, offset %s", ll.Num, offset)
-		}
-	}()
-
-	for {
-		var r io.Reader
-		var err error
-		r, offset, err = rr.NextRecord()
-		if err == nil {
-			_, err = io.Copy(&buf, r)
-		}
-		if err != nil {
-			// It is common to encounter a zeroed or invalid chunk due to WAL
-			// preallocation and WAL recycling. However zeroed or invalid chunks
-			// can also be a consequence of corruption / disk rot. When the log
-			// reader encounters one of these cases, it attempts to disambiguate
-			// by reading ahead looking for a future record. If a future chunk
-			// indicates the chunk at the original offset should've been valid, it
-			// surfaces record.ErrInvalidChunk or record.ErrZeroedChunk. These
-			// errors are always indicative of corruption and data loss.
-			//
-			// Otherwise, the reader surfaces record.ErrUnexpectedEOF indicating
-			// that the WAL terminated uncleanly and ambiguously. If the WAL is
-			// the most recent logical WAL, the caller passes in
-			// (strictWALTail=false), indicating we should tolerate the unclean
-			// ending. If the WAL is an older WAL, the caller passes in
-			// (strictWALTail=true), indicating that the WAL should have been
-			// closed cleanly, and we should interpret the
-			// `record.ErrUnexpectedEOF` as corruption and stop recovery.
-			if errors.Is(err, io.EOF) {
-				break
-			} else if errors.Is(err, record.ErrUnexpectedEOF) && !strictWALTail {
-				break
-			} else if (errors.Is(err, record.ErrUnexpectedEOF) && strictWALTail) ||
-				errors.Is(err, record.ErrInvalidChunk) || errors.Is(err, record.ErrZeroedChunk) {
-				// If a read-ahead returns record.ErrInvalidChunk or
-				// record.ErrZeroedChunk, then there's definitively corruption.
-				//
-				// If strictWALTail=true, then record.ErrUnexpectedEOF should
-				// also be considered corruption because the strictWALTail
-				// indicates we expect a clean end to the WAL.
-				//
-				// Other I/O related errors should not be marked with corruption
-				// and simply returned.
-				err = errors.Mark(err, ErrCorruption)
-			}
-
-			return nil, 0, errors.Wrap(err, "pebble: error when replaying WAL")
-		}
-
-		if buf.Len() < batchrepr.HeaderLen {
-			return nil, 0, base.CorruptionErrorf("pebble: corrupt wal %s (offset %s)",
-				errors.Safe(base.DiskFileNum(ll.Num)), offset)
-		}
-
-		if d.opts.ErrorIfNotPristine {
-			return nil, 0, errors.WithDetailf(ErrDBNotPristine, "location: %q", d.dirname)
-		}
-
-		// Specify Batch.db so that Batch.SetRepr will compute Batch.memTableSize
-		// which is used below.
-		b = Batch{}
-		b.db = d
-		if err := b.SetRepr(buf.Bytes()); err != nil {
-			return nil, 0, err
-		}
-		seqNum := b.SeqNum()
-		maxSeqNum = seqNum + base.SeqNum(b.Count())
-		keysReplayed += int64(b.Count())
-		batchesReplayed++
-		{
-			br := b.Reader()
-			if kind, _, _, ok, err := br.Next(); err != nil {
-				return nil, 0, err
-			} else if ok && (kind == InternalKeyKindIngestSST || kind == InternalKeyKindExcise) {
-				// We're in the flushable ingests (+ possibly excises) case.
-				//
-				// Ingests require an up-to-date view of the LSM to determine the target
-				// level of ingested sstables, and to accurately compute excises. Instead of
-				// doing an ingest in this function, we just enqueue a flushable ingest
-				// in the flushables queue and run a regular flush.
-				flushMem()
-				// mem is nil here.
-				entry, err = d.replayIngestedFlushable(&b, base.DiskFileNum(ll.Num))
-				if err != nil {
-					return nil, 0, err
-				}
-				fi := entry.flushable.(*ingestedFlushable)
-				flushableIngests = append(flushableIngests, fi)
-				d.mu.mem.queue = append(d.mu.mem.queue, entry)
-				// A flushable ingest is always followed by a WAL rotation.
-				break
-			}
-		}
-
-		if b.memTableSize >= uint64(d.largeBatchThreshold) {
-			flushMem()
-			// Make a copy of the data slice since it is currently owned by buf and will
-			// be reused in the next iteration.
-			b.data = slices.Clone(b.data)
-			b.flushable, err = newFlushableBatch(&b, d.opts.Comparer)
-			if err != nil {
-				return nil, 0, err
-			}
-			entry := d.newFlushableEntry(b.flushable, base.DiskFileNum(ll.Num), b.SeqNum())
-			// Disable memory accounting by adding a reader ref that will never be
-			// removed.
-			entry.readerRefs.Add(1)
-			d.mu.mem.queue = append(d.mu.mem.queue, entry)
-		} else {
-			ensureMem(seqNum)
-			if err = mem.prepare(&b); err != nil && err != arenaskl.ErrArenaFull {
-				return nil, 0, err
-			}
-			// We loop since DB.newMemTable() slowly grows the size of allocated memtables, so the
-			// batch may not initially fit, but will eventually fit (since it is smaller than
-			// largeBatchThreshold).
-			for err == arenaskl.ErrArenaFull {
-				flushMem()
-				ensureMem(seqNum)
-				err = mem.prepare(&b)
-				if err != nil && err != arenaskl.ErrArenaFull {
-					return nil, 0, err
-				}
-			}
-			if err = mem.apply(&b, seqNum); err != nil {
-				return nil, 0, err
-			}
-			mem.writerUnref()
-		}
-		buf.Reset()
-	}
-
-	d.opts.Logger.Infof("[JOB %d] WAL %s stopped reading at offset: %s; replayed %d keys in %d batches",
-		jobID, ll.String(), offset, keysReplayed, batchesReplayed)
-	if !d.opts.ReadOnly {
-		flushMem()
-	}
-
-	// mem is nil here, if !ReadOnly.
-	return flushableIngests, maxSeqNum, err
-}
-
 func readOptionsFile(opts *Options, path string) (string, error) {
 	f, err := opts.FS.Open(path)
 	if err != nil {