Repository: cockroachdb/pebble
Pull request #3901 (2024-08-28 20:17:15)

crl-release-24.1: backport WAL observability fixes
24.1 backport of #3862 and #3864.

----

    tool: fix up `wal dump`, add `wal dump-merged`

    Adjust the wal dump tool to error on unrecognized key kinds, and surface errors
    encountered more prominently in the output.

    Additionally, add a new `wal dump-merged` command that dumps the merged view of
    a WAL that's composed of multiple physical segment files.

    db: fix WAL offset in replayWAL errors

    If an error occurs while replaying the WALs during Open, the error is annotated
    with the log number and the offset within the log. Previously, the offset was
    always zero. This also caused the queued flushables to be created with a zero
    logSize. This didn't matter in practice because the queued flushables would be
    flushed before Open returned.

    This commit adds a PreviousFilesBytes total to the wal.Offset type to describe
    the number of bytes replayed from previous WAL segment files.


Patch:
diff --git a/open.go b/open.go
index dcd06a07..ab0988b4 100644
--- a/open.go
+++ b/open.go
@@ -766,7 +766,7 @@ func (d *DB) replayWAL(
 		buf             bytes.Buffer
 		mem             *memTable
 		entry           *flushableEntry
-		offset          int64 // byte offset in rr
+		offset          wal.Offset
 		lastFlushOffset int64
 		keysReplayed    int64 // number of keys replayed
 		batchesReplayed int64 // number of batches replayed
@@ -792,12 +792,13 @@ func (d *DB) replayWAL(
 			return
 		}
 		var logSize uint64
-		if offset >= lastFlushOffset {
-			logSize = uint64(offset - lastFlushOffset)
+		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 = offset
+		lastFlushOffset = mergedOffset
 		entry.logSize = logSize
 		if !d.opts.ReadOnly {
 			toFlush = append(toFlush, entry)
@@ -836,12 +837,14 @@ func (d *DB) replayWAL(
 	}
 	defer func() {
 		if err != nil {
-			err = errors.WithDetailf(err, "replaying wal %d, offset %d", ll.Num, offset)
+			err = errors.WithDetailf(err, "replaying wal %d, offset %s", ll.Num, offset)
 		}
 	}()
 
 	for {
-		r, offset, err := rr.NextRecord()
+		var r io.Reader
+		var err error
+		r, offset, err = rr.NextRecord()
 		if err == nil {
 			_, err = io.Copy(&buf, r)
 		}
@@ -1048,7 +1051,7 @@ func (d *DB) replayWAL(
 		buf.Reset()
 	}
 
-	d.opts.Logger.Infof("[JOB %d] WAL %s stopped reading at offset: %d; replayed %d keys in %d batches",
+	d.opts.Logger.Infof("[JOB %d] WAL %s stopped reading at offset: %s; replayed %d keys in %d batches",
 		jobID, base.DiskFileNum(ll.Num).String(), offset, keysReplayed, batchesReplayed)
 	flushMem()
 
diff --git a/tool/wal.go b/tool/wal.go
index 911634a2..e8f4da83 100644
--- a/tool/wal.go
+++ b/tool/wal.go
@@ -9,8 +9,11 @@ import (
 	"encoding/binary"
 	"fmt"
 	"io"
+	"os"
 
+	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble"
+	"github.com/cockroachdb/pebble/batchrepr"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/rangekey"
 	"github.com/cockroachdb/pebble/record"
@@ -22,8 +25,9 @@ import (
 // walT implements WAL-level tools, including both configuration state and the
 // commands themselves.
 type walT struct {
-	Root *cobra.Command
-	Dump *cobra.Command
+	Root       *cobra.Command
+	Dump       *cobra.Command
+	DumpMerged *cobra.Command
 
 	opts     *pebble.Options
 	fmtKey   keyFormatter
@@ -56,8 +60,19 @@ Print the contents of the WAL files.
 		Args: cobra.MinimumNArgs(1),
 		Run:  w.runDump,
 	}
+	w.DumpMerged = &cobra.Command{
+		Use:   "dump-merged <wal-files>",
+		Short: "print WAL contents",
+		Long: `
+Print the merged contents of multiple WAL segment files that
+together form a single logical WAL.
+`,
+		Args: cobra.MinimumNArgs(1),
+		Run:  w.runDumpMerged,
+	}
 
 	w.Root.AddCommand(w.Dump)
+	w.Root.AddCommand(w.DumpMerged)
 	w.Root.PersistentFlags().BoolVarP(&w.verbose, "verbose", "v", false, "verbose output")
 
 	w.Dump.Flags().Var(
@@ -67,10 +82,24 @@ Print the contents of the WAL files.
 	return w
 }
 
+type errAndArg struct {
+	err error
+	arg string
+}
+
 func (w *walT) runDump(cmd *cobra.Command, args []string) {
 	stdout, stderr := cmd.OutOrStdout(), cmd.OutOrStderr()
 	w.fmtKey.setForComparer(w.defaultComparer, w.comparers)
 	w.fmtValue.setForComparer(w.defaultComparer, w.comparers)
+	var errs []errAndArg
+	logErr := func(arg string, offset int64, err error) {
+		err = errors.Wrapf(err, "%s: offset %d: ", arg, offset)
+		fmt.Fprintf(stderr, "%s\n", err)
+		errs = append(errs, errAndArg{
+			arg: arg,
+			err: err,
+		})
+	}
 
 	for _, arg := range args {
 		func() {
@@ -107,10 +136,10 @@ func (w *walT) runDump(cmd *cobra.Command, args []string) {
 					// preallocation and WAL recycling. We need to distinguish these
 					// errors from EOF in order to recognize that the record was
 					// truncated, but want to otherwise treat them like EOF.
-					switch err {
-					case record.ErrZeroedChunk:
+					switch {
+					case errors.Is(err, record.ErrZeroedChunk):
 						fmt.Fprintf(stdout, "EOF [%s] (may be due to WAL preallocation)\n", err)
-					case record.ErrInvalidChunk:
+					case errors.Is(err, record.ErrInvalidChunk):
 						fmt.Fprintf(stdout, "EOF [%s] (may be due to WAL recycling)\n", err)
 					default:
 						fmt.Fprintf(stdout, "%s\n", err)
@@ -123,50 +152,148 @@ func (w *walT) runDump(cmd *cobra.Command, args []string) {
 					fmt.Fprintf(stdout, "corrupt batch within log file %q: %v", arg, err)
 					return
 				}
-				fmt.Fprintf(stdout, "%d(%d) seq=%d count=%d\n",
-					offset, len(b.Repr()), b.SeqNum(), b.Count())
-				for r, idx := b.Reader(), 0; ; idx++ {
-					kind, ukey, value, ok, err := r.Next()
-					if !ok {
-						if err != nil {
-							fmt.Fprintf(stdout, "corrupt batch within log file %q: %v", arg, err)
-						}
-						break
-					}
-					fmt.Fprintf(stdout, "    %s(", kind)
-					switch kind {
-					case base.InternalKeyKindDelete:
-						fmt.Fprintf(stdout, "%s", w.fmtKey.fn(ukey))
-					case base.InternalKeyKindSet:
-						fmt.Fprintf(stdout, "%s,%s", w.fmtKey.fn(ukey), w.fmtValue.fn(ukey, value))
-					case base.InternalKeyKindMerge:
-						fmt.Fprintf(stdout, "%s,%s", w.fmtKey.fn(ukey), w.fmtValue.fn(ukey, value))
-					case base.InternalKeyKindLogData:
-						fmt.Fprintf(stdout, "<%d>", len(value))
-					case base.InternalKeyKindIngestSST:
-						fileNum, _ := binary.Uvarint(ukey)
-						fmt.Fprintf(stdout, "%s", base.FileNum(fileNum))
-					case base.InternalKeyKindSingleDelete:
-						fmt.Fprintf(stdout, "%s", w.fmtKey.fn(ukey))
-					case base.InternalKeyKindSetWithDelete:
-						fmt.Fprintf(stdout, "%s", w.fmtKey.fn(ukey))
-					case base.InternalKeyKindRangeDelete:
-						fmt.Fprintf(stdout, "%s,%s", w.fmtKey.fn(ukey), w.fmtKey.fn(value))
-					case base.InternalKeyKindRangeKeySet, base.InternalKeyKindRangeKeyUnset, base.InternalKeyKindRangeKeyDelete:
-						ik := base.MakeInternalKey(ukey, b.SeqNum()+uint64(idx), kind)
-						s, err := rangekey.Decode(ik, value, nil)
-						if err != nil {
-							fmt.Fprintf(stdout, "%s: error decoding %s", w.fmtKey.fn(ukey), err)
-						} else {
-							fmt.Fprintf(stdout, "%s", s.Pretty(w.fmtKey.fn))
-						}
-					case base.InternalKeyKindDeleteSized:
-						v, _ := binary.Uvarint(value)
-						fmt.Fprintf(stdout, "%s,%d", w.fmtKey.fn(ukey), v)
-					}
-					fmt.Fprintf(stdout, ")\n")
-				}
+				fmt.Fprintf(stdout, "%d(%d) seq=%d count=%d, len=%d\n",
+					offset, len(b.Repr()), b.SeqNum(), b.Count(), buf.Len())
+				w.dumpBatch(stdout, &b, b.Reader(), func(err error) {
+					logErr(arg, offset, err)
+				})
 			}
 		}()
 	}
+	if len(errs) > 0 {
+		fmt.Fprintln(stderr, "Errors: ")
+		for _, ea := range errs {
+			fmt.Fprintf(stderr, "%s: %s\n", ea.arg, ea.err)
+		}
+	}
+}
+
+func (w *walT) runDumpMerged(cmd *cobra.Command, args []string) {
+	stdout, stderr := cmd.OutOrStdout(), cmd.OutOrStderr()
+	w.fmtKey.setForComparer(w.defaultComparer, w.comparers)
+	w.fmtValue.setForComparer(w.defaultComparer, w.comparers)
+	var a wal.FileAccumulator
+	for _, arg := range args {
+		isLog, err := a.MaybeAccumulate(w.opts.FS, arg)
+		if !isLog {
+			fmt.Fprintf(stderr, "%q does not parse as a log file\n", arg)
+			os.Exit(1)
+		} else if err != nil {
+			fmt.Fprintf(stderr, "%s: %s\n", arg, err)
+			os.Exit(1)
+		}
+	}
+	logs := a.Finish()
+	var errs []error
+	for _, log := range logs {
+		fmt.Fprintf(stdout, "log file %s contains %d segment files:\n", log.Num, log.NumSegments())
+		errs = append(errs, w.runDumpMergedOne(cmd, log)...)
+	}
+	if len(errs) > 0 {
+		fmt.Fprintln(stderr, "Errors: ")
+		for _, err := range errs {
+			fmt.Fprintf(stderr, "%s\n", err)
+		}
+	}
+}
+
+func (w *walT) runDumpMergedOne(cmd *cobra.Command, ll wal.LogicalLog) []error {
+	stdout, stderr := cmd.OutOrStdout(), cmd.OutOrStderr()
+	var buf bytes.Buffer
+	var errs []error
+	var b pebble.Batch
+
+	logErr := func(offset wal.Offset, err error) {
+		err = errors.Wrapf(err, "offset %s: ", offset)
+		fmt.Fprintln(stderr, err)
+		errs = append(errs, err)
+	}
+
+	rr := ll.OpenForRead()
+	for {
+		buf.Reset()
+		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. We need to distinguish these
+			// errors from EOF in order to recognize that the record was
+			// truncated and to avoid replaying subsequent WALs, but want
+			// to otherwise treat them like EOF.
+			if err == io.EOF {
+				break
+			} else if record.IsInvalidRecord(err) {
+				break
+			}
+			return append(errs, err)
+		}
+		if buf.Len() < batchrepr.HeaderLen {
+			logErr(offset, errors.Newf("%d-byte batch too short", buf.Len()))
+			continue
+		}
+		b = pebble.Batch{}
+		if err := b.SetRepr(buf.Bytes()); err != nil {
+			logErr(offset, errors.Newf("unable to parse batch: %x", buf.Bytes()))
+			continue
+		}
+		fmt.Fprintf(stdout, "%s(%d) seq=%d count=%d, len=%d\n",
+			offset, len(b.Repr()), b.SeqNum(), b.Count(), buf.Len())
+		w.dumpBatch(stdout, &b, b.Reader(), func(err error) {
+			logErr(offset, err)
+		})
+	}
+	return nil
+}
+
+func (w *walT) dumpBatch(
+	stdout io.Writer, b *pebble.Batch, r batchrepr.Reader, logErr func(error),
+) {
+	for idx := 0; ; idx++ {
+		kind, ukey, value, ok, err := r.Next()
+		if !ok {
+			if err != nil {
+				logErr(errors.Newf("unable to decode %d'th key in batch; %s", idx, err))
+			}
+			break
+		}
+		fmt.Fprintf(stdout, "    %s(", kind)
+		switch kind {
+		case base.InternalKeyKindDelete:
+			fmt.Fprintf(stdout, "%s", w.fmtKey.fn(ukey))
+		case base.InternalKeyKindSet:
+			fmt.Fprintf(stdout, "%s,%s", w.fmtKey.fn(ukey), w.fmtValue.fn(ukey, value))
+		case base.InternalKeyKindMerge:
+			fmt.Fprintf(stdout, "%s,%s", w.fmtKey.fn(ukey), w.fmtValue.fn(ukey, value))
+		case base.InternalKeyKindLogData:
+			fmt.Fprintf(stdout, "<%d>", len(value))
+		case base.InternalKeyKindIngestSST:
+			fileNum, _ := binary.Uvarint(ukey)
+			fmt.Fprintf(stdout, "%s", base.FileNum(fileNum))
+		case base.InternalKeyKindSingleDelete:
+			fmt.Fprintf(stdout, "%s", w.fmtKey.fn(ukey))
+		case base.InternalKeyKindSetWithDelete:
+			fmt.Fprintf(stdout, "%s", w.fmtKey.fn(ukey))
+		case base.InternalKeyKindRangeDelete:
+			fmt.Fprintf(stdout, "%s,%s", w.fmtKey.fn(ukey), w.fmtKey.fn(value))
+		case base.InternalKeyKindRangeKeySet, base.InternalKeyKindRangeKeyUnset, base.InternalKeyKindRangeKeyDelete:
+			ik := base.MakeInternalKey(ukey, b.SeqNum()+uint64(idx), kind)
+			s, err := rangekey.Decode(ik, value, nil)
+			if err != nil {
+				logErr(errors.Newf("%s: error decoding %s", w.fmtKey.fn(ukey), err))
+			} else {
+				fmt.Fprintf(stdout, "%s", s.Pretty(w.fmtKey.fn))
+			}
+		case base.InternalKeyKindDeleteSized:
+			v, _ := binary.Uvarint(value)
+			fmt.Fprintf(stdout, "%s,%d", w.fmtKey.fn(ukey), v)
+		default:
+			err := errors.Newf("invalid key kind %d in key at index %d/%d of batch with seqnum %d at offset %d",
+				kind, idx, b.Count(), b.SeqNum())
+			fmt.Fprintf(stdout, "<error: %s>", err)
+			logErr(err)
+		}
+		fmt.Fprintf(stdout, ")\n")
+	}
 }
diff --git a/wal/failover_writer.go b/wal/failover_writer.go
index a8adbce8..6ce13868 100644
--- a/wal/failover_writer.go
+++ b/wal/failover_writer.go
@@ -773,19 +773,23 @@ func (ww *failoverWriter) closeInternal() (logicalOffset int64, err error) {
 	lastRecordIndex := record.PendingSyncIndex{Index: ww.q.getLastIndex()}
 	ww.mu.Lock()
 	defer ww.mu.Unlock()
+	numWriters := ww.mu.nextWriterIndex
 	// Every iteration starts and ends with the mutex held.
 	//
-	// Invariant: ww.mu.nextWriterIndex >= 1.
+	// Invariant: numWriters >= 1.
 	//
 	// We will loop until we have closed the lastWriter (and use
 	// lastWriter.err). We also need to call close on all LogWriters
 	// that will not close themselves, i.e., those that have already been
 	// created and installed in failoverWriter.writers (this set may change
 	// while failoverWriter.Close runs).
-	for !lastWriter.closed {
-		numWriters := ww.mu.nextWriterIndex
+	for !lastWriter.closed || numWriters > lastWriter.index+1 {
 		if numWriters > closeCalledCount {
-			// INVARIANT: numWriters > closeCalledCount.
+			// lastWriter.index may or may not have advanced. If it has advanced, we
+			// need to reinitialize lastWriterState. If it hasn't advanced, and
+			// numWriters > closeCalledCount, we know that we haven't called close
+			// on it, so nothing in lastWriterState needs to be retained. For
+			// simplicity, we overwrite in both cases.
 			lastWriter = lastWriterState{
 				index: numWriters - 1,
 			}
@@ -857,7 +861,34 @@ func (ww *failoverWriter) closeInternal() (logicalOffset int64, err error) {
 		if !lastWriter.closed {
 			// Either waiting for creation of last writer or waiting for the close
 			// to finish, or something else to become the last writer.
+			//
+			// It is possible that what we think of as the last writer (lastWriter)
+			// closes itself while ww.mu is no longer held here, and a new LogWriter
+			// is created too. All the records are synced, but the real last writer
+			// may still be writing some records. Specifically, consider the
+			// following sequence while this wait does not hold the mutex:
+			//
+			// - recordQueue has an entry, with index 50, that does not require a
+			//   sync.
+			// - Last writer created at index 10 and entry 50 is handed to it.
+			// - lastWriter.index is still 9 and it closes itself and signals this
+			//   cond. It has written entry 50 and synced (since close syncs).
+			// - The wait completes.
+			//
+			// Now the writer at index 10 will never be closed and will never sync.
+			// A crash can cause some part of what it writes to be lost. Note that
+			// there is no data loss, but there are some unfortunate consequences:
+			//
+			// - We never closed a file descriptor.
+			// - virtualWALReader.NextRecord can return an error on finding a
+			//   malformed chunk in the last writer (at index 10) instead of
+			//   swallowing the error. This can cause DB.Open to fail.
+			//
+			// To avoid this, we grab the latest value of numWriters on reacquiring
+			// the mutex, and will continue looping until the writer at index 10 is
+			// closed (or writer at index 11 is created).
 			ww.mu.cond.Wait()
+			numWriters = ww.mu.nextWriterIndex
 		}
 	}
 	if ww.mu.writers[lastWriter.index].w != nil {
@@ -867,7 +898,10 @@ func (ww *failoverWriter) closeInternal() (logicalOffset int64, err error) {
 	err = lastWriter.err
 	ww.mu.metrics = lastWriter.metrics
 	ww.mu.closed = true
-	_, _ = ww.q.popAll(err)
+	n, m := ww.q.popAll(err)
+	if err == nil && (n > 0 || m > 0) {
+		panic(errors.AssertionFailedf("no error but recordQueue had %d records and %d syncs", n, m))
+	}
 	return logicalOffset, err
 }
 
diff --git a/wal/reader.go b/wal/reader.go
index 6cec8f80..19422b71 100644
--- a/wal/reader.go
+++ b/wal/reader.go
@@ -360,6 +360,7 @@ func (r *virtualWALReader) nextFile() error {
 	}
 
 	fs, path := r.LogicalLog.SegmentLocation(r.currIndex)
+	r.off.PreviousFilesBytes += r.off.Physical
 	r.off.PhysicalFile = path
 	r.off.Physical = 0
 	var err error
diff --git a/wal/wal.go b/wal/wal.go
index 9aee2afe..8ab7a25e 100644
--- a/wal/wal.go
+++ b/wal/wal.go
@@ -423,10 +423,19 @@ type Offset struct {
 	// Physical indicates the file offset at which a record begins within
 	// the physical file named by PhysicalFile.
 	Physical int64
+	// PreviousFilesBytes is the bytes read from all the previous physical
+	// segment files that have been read up to the current log segment. If WAL
+	// failover is not in use, PreviousFileBytes will always be zero. Otherwise,
+	// it may be non-zero when replaying records from multiple segment files
+	// that make up a single logical WAL.
+	PreviousFilesBytes int64
 }
 
 // String implements fmt.Stringer, returning a string representation of the
 // offset.
 func (o Offset) String() string {
+	if o.PreviousFilesBytes > 0 {
+		return fmt.Sprintf("(%s: %d), %d from previous files", o.PhysicalFile, o.Physical, o.PreviousFilesBytes)
+	}
 	return fmt.Sprintf("(%s: %d)", o.PhysicalFile, o.Physical)
 }


Test patch:
diff --git a/tool/testdata/wal_dump b/tool/testdata/wal_dump
index 2fa4e30f..95725fca 100644
--- a/tool/testdata/wal_dump
+++ b/tool/testdata/wal_dump
@@ -6,15 +6,15 @@ wal dump
 ../testdata/db-stage-2/000002.log
 ----
 000002.log
-0(21) seq=10 count=1
+0(21) seq=10 count=1, len=21
     SET(test formatter: foo,test value formatter: one)
-32(21) seq=11 count=1
+32(21) seq=11 count=1, len=21
     SET(test formatter: bar,test value formatter: two)
-64(23) seq=12 count=1
+64(23) seq=12 count=1, len=23
     SET(test formatter: baz,test value formatter: three)
-98(22) seq=13 count=1
+98(22) seq=13 count=1, len=22
     SET(test formatter: foo,test value formatter: four)
-131(17) seq=14 count=1
+131(17) seq=14 count=1, len=17
     DEL(test formatter: bar)
 EOF
 
@@ -24,15 +24,15 @@ wal dump
 --value=size
 ----
 000002.log
-0(21) seq=10 count=1
+0(21) seq=10 count=1, len=21
     SET(foo,<3>)
-32(21) seq=11 count=1
+32(21) seq=11 count=1, len=21
     SET(bar,<3>)
-64(23) seq=12 count=1
+64(23) seq=12 count=1, len=23
     SET(baz,<5>)
-98(22) seq=13 count=1
+98(22) seq=13 count=1, len=22
     SET(foo,<4>)
-131(17) seq=14 count=1
+131(17) seq=14 count=1, len=17
     DEL(bar)
 EOF
 
@@ -42,11 +42,11 @@ wal dump
 ../testdata/db-stage-4/000005.log
 ----
 000005.log
-0(22) seq=15 count=1
+0(22) seq=15 count=1, len=22
     SET(foo,<4>)
-33(22) seq=16 count=1
+33(22) seq=16 count=1, len=22
     SET(quux,<3>)
-66(17) seq=17 count=1
+66(17) seq=17 count=1, len=17
     DEL(baz)
 EOF
 
@@ -56,11 +56,11 @@ wal dump
 --value=%x
 ----
 000005.log
-0(22) seq=15 count=1
+0(22) seq=15 count=1, len=22
     SET(666f6f,66697665)
-33(22) seq=16 count=1
+33(22) seq=16 count=1, len=22
     SET(71757578,736978)
-66(17) seq=17 count=1
+66(17) seq=17 count=1, len=17
     DEL(62617a)
 EOF
 
@@ -70,11 +70,11 @@ wal dump
 --value=pretty:test-comparer
 ----
 000005.log
-0(22) seq=15 count=1
+0(22) seq=15 count=1, len=22
     SET(foo,test value formatter: five)
-33(22) seq=16 count=1
+33(22) seq=16 count=1, len=22
     SET(quux,test value formatter: six)
-66(17) seq=17 count=1
+66(17) seq=17 count=1, len=17
     DEL(baz)
 EOF
 
@@ -84,11 +84,11 @@ wal dump
 --value=%x
 ----
 000005.log
-0(22) seq=15 count=1
+0(22) seq=15 count=1, len=22
     SET(test formatter: foo,66697665)
-33(22) seq=16 count=1
+33(22) seq=16 count=1, len=22
     SET(test formatter: quux,736978)
-66(17) seq=17 count=1
+66(17) seq=17 count=1, len=17
     DEL(test formatter: baz)
 EOF
 
@@ -98,11 +98,11 @@ wal dump
 --value=quoted
 ----
 000005.log
-0(22) seq=15 count=1
+0(22) seq=15 count=1, len=22
     SET(foo,five)
-33(22) seq=16 count=1
+33(22) seq=16 count=1, len=22
     SET(quux,six)
-66(17) seq=17 count=1
+66(17) seq=17 count=1, len=17
     DEL(baz)
 EOF
 
@@ -110,7 +110,7 @@ wal dump
 ./testdata/mixed/000004.log
 ----
 000004.log
-0(42) seq=39 count=4
+0(42) seq=39 count=4, len=42
     SET(test formatter: a@2,test value formatter: )
     RANGEKEYSET(test formatter: a-test formatter: z:{(#40,RANGEKEYSET,@3)})
     RANGEKEYUNSET(test formatter: a-test formatter: z:{(#41,RANGEKEYUNSET,@4)})
diff --git a/wal/testdata/reader b/wal/testdata/reader
index bf3010fb..23aa7248 100644
--- a/wal/testdata/reader
+++ b/wal/testdata/reader
@@ -52,13 +52,13 @@ r.NextRecord() = (rr, (000001.log: 1035), <nil>)
 r.NextRecord() = (rr, (000001.log: 1076), <nil>)
   io.ReadAll(rr) = ("1500000000000000320000004f091e9a83fdeae0ec55eb233a9b5394cb3c7856... <512000-byte record>", <nil>)
   BatchHeader: [seqNum=21,count=50]
-r.NextRecord() = (rr, (000001-001.log: 0), <nil>)
+r.NextRecord() = (rr, (000001-001.log: 0), 513252 from previous files, <nil>)
   io.ReadAll(rr) = ("16000000000000000200000038d0ccacfb33b57fb3d386cbe2b67a2fbdc82214... <412-byte record>", <nil>)
   BatchHeader: [seqNum=22,count=2]
-r.NextRecord() = (rr, (000001-001.log: 498), <nil>)
+r.NextRecord() = (rr, (000001-001.log: 498), 513252 from previous files, <nil>)
   io.ReadAll(rr) = ("180000000000000001000000ede8f156c48faf84dd55235d19a2df01d13021fc... <100-byte record>", <nil>)
   BatchHeader: [seqNum=24,count=1]
-r.NextRecord() = (rr, (000001-001.log: 609), EOF)
+r.NextRecord() = (rr, (000001-001.log: 609), 513252 from previous files, EOF)
 
 # Test a recycled log file. Recycle 000001.log as 000002.log. This time, do not
 # exit cleanly. This simulates a hard process exit (eg, during a fatal shutdown,
@@ -134,13 +134,13 @@ r.NextRecord() = (rr, (000003.log: 111), <nil>)
 r.NextRecord() = (rr, (000003.log: 272), <nil>)
   io.ReadAll(rr) = ("2a000000000000000100000019458dc5400169e5", <nil>)
   BatchHeader: [seqNum=42,count=1]
-r.NextRecord() = (rr, (000003-001.log: 192), <nil>)
+r.NextRecord() = (rr, (000003-001.log: 192), 303 from previous files, <nil>)
   io.ReadAll(rr) = ("2b00000000000000030000009cbf29476e797bac2db8bfea65bda29ea50ddbe4... <80-byte record>", <nil>)
   BatchHeader: [seqNum=43,count=3]
-r.NextRecord() = (rr, (000003-001.log: 283), <nil>)
+r.NextRecord() = (rr, (000003-001.log: 283), 303 from previous files, <nil>)
   io.ReadAll(rr) = ("2e000000000000000900000027337fa5bd626044dc5d9d08085bf4ce13bc8d00... <2055-byte record>", <nil>)
   BatchHeader: [seqNum=46,count=9]
-r.NextRecord() = (rr, (000003-001.log: 2349), EOF)
+r.NextRecord() = (rr, (000003-001.log: 2349), 303 from previous files, EOF)
 
 # Extend logical log file 000003 with another log file, the result of failing
 # back to the original the device. This time do an "unclean" close.
@@ -166,16 +166,16 @@ r.NextRecord() = (rr, (000003.log: 111), <nil>)
 r.NextRecord() = (rr, (000003.log: 272), <nil>)
   io.ReadAll(rr) = ("2a000000000000000100000019458dc5400169e5", <nil>)
   BatchHeader: [seqNum=42,count=1]
-r.NextRecord() = (rr, (000003-001.log: 192), <nil>)
+r.NextRecord() = (rr, (000003-001.log: 192), 303 from previous files, <nil>)
   io.ReadAll(rr) = ("2b00000000000000030000009cbf29476e797bac2db8bfea65bda29ea50ddbe4... <80-byte record>", <nil>)
   BatchHeader: [seqNum=43,count=3]
-r.NextRecord() = (rr, (000003-001.log: 283), <nil>)
+r.NextRecord() = (rr, (000003-001.log: 283), 303 from previous files, <nil>)
   io.ReadAll(rr) = ("2e000000000000000900000027337fa5bd626044dc5d9d08085bf4ce13bc8d00... <2055-byte record>", <nil>)
   BatchHeader: [seqNum=46,count=9]
-r.NextRecord() = (rr, (000003-002.log: 2157), <nil>)
+r.NextRecord() = (rr, (000003-002.log: 2157), 2652 from previous files, <nil>)
   io.ReadAll(rr) = ("370000000000000002000000ff0710201f4008e679428b4994708a1af8507303... <205-byte record>", <nil>)
   BatchHeader: [seqNum=55,count=2]
-r.NextRecord() = (rr, (000003-002.log: 2373), EOF)
+r.NextRecord() = (rr, (000003-002.log: 2373), 2652 from previous files, EOF)
 
 # Test reading a log file that does not exist.
 
@@ -243,16 +243,16 @@ r.NextRecord() = (rr, (000005.log: 909), <nil>)
 r.NextRecord() = (rr, (000005.log: 3445), <nil>)
   io.ReadAll(rr) = ("0b7401000000000000010000907cd29c9a6deaf239e76e3374f6e9eef047f57f... <2566-byte record>", <nil>)
   BatchHeader: [seqNum=95243,count=256]
-r.NextRecord() = (rr, (000005-001.log: 0), <nil>)
+r.NextRecord() = (rr, (000005-001.log: 0), 6022 from previous files, <nil>)
   io.ReadAll(rr) = ("0b75010000000000020000006cad8a0a1461d1ec53bb834b47c6853e040ae9ce... <44-byte record>", <nil>)
   BatchHeader: [seqNum=95499,count=2]
-r.NextRecord() = (rr, (000005-001.log: 55), <nil>)
+r.NextRecord() = (rr, (000005-001.log: 55), 6022 from previous files, <nil>)
   io.ReadAll(rr) = ("0d7501000000000005000000c78be2f74d28753a03854ed63e6fd0f17113688d... <416-byte record>", <nil>)
   BatchHeader: [seqNum=95501,count=5]
-r.NextRecord() = (rr, (000005-001.log: 482), <nil>)
+r.NextRecord() = (rr, (000005-001.log: 482), 6022 from previous files, <nil>)
   io.ReadAll(rr) = ("12750100000000001d00000096cedf6103af61c008d9f850e63a1dfc7518b9a7... <199-byte record>", <nil>)
   BatchHeader: [seqNum=95506,count=29]
-r.NextRecord() = (rr, (000005-001.log: 692), pebble/record: invalid chunk)
+r.NextRecord() = (rr, (000005-001.log: 692), 6022 from previous files, pebble/record: invalid chunk)
 
 # Read again, this time pretending we found a third segment with the
 # logNameIndex=002. This helps exercise error conditions switching to a new
@@ -272,16 +272,16 @@ r.NextRecord() = (rr, (000005.log: 909), <nil>)
 r.NextRecord() = (rr, (000005.log: 3445), <nil>)
   io.ReadAll(rr) = ("0b7401000000000000010000907cd29c9a6deaf239e76e3374f6e9eef047f57f... <2566-byte record>", <nil>)
   BatchHeader: [seqNum=95243,count=256]
-r.NextRecord() = (rr, (000005-001.log: 0), <nil>)
+r.NextRecord() = (rr, (000005-001.log: 0), 6022 from previous files, <nil>)
   io.ReadAll(rr) = ("0b75010000000000020000006cad8a0a1461d1ec53bb834b47c6853e040ae9ce... <44-byte record>", <nil>)
   BatchHeader: [seqNum=95499,count=2]
-r.NextRecord() = (rr, (000005-001.log: 55), <nil>)
+r.NextRecord() = (rr, (000005-001.log: 55), 6022 from previous files, <nil>)
   io.ReadAll(rr) = ("0d7501000000000005000000c78be2f74d28753a03854ed63e6fd0f17113688d... <416-byte record>", <nil>)
   BatchHeader: [seqNum=95501,count=5]
-r.NextRecord() = (rr, (000005-001.log: 482), <nil>)
+r.NextRecord() = (rr, (000005-001.log: 482), 6022 from previous files, <nil>)
   io.ReadAll(rr) = ("12750100000000001d00000096cedf6103af61c008d9f850e63a1dfc7518b9a7... <199-byte record>", <nil>)
   BatchHeader: [seqNum=95506,count=29]
-r.NextRecord() = (rr, (000005-002.log: 0), opening WAL file segment "000005-002.log": open 000005-002.log: file does not exist)
+r.NextRecord() = (rr, (000005-002.log: 0), 6714 from previous files, opening WAL file segment "000005-002.log": open 000005-002.log: file does not exist)
 
 # Test a scenario where 4 unique batches are split across three physical log
 # files. The first log contains (b0, b1, b2), the second log (b1) and the third
@@ -325,10 +325,10 @@ r.NextRecord() = (rr, (000006.log: 406), <nil>)
 r.NextRecord() = (rr, (000006.log: 94105), <nil>)
   io.ReadAll(rr) = ("1c0200000000000001000000404841433f5369713ee90d8f86c50c5903fa38e9... <180-byte record>", <nil>)
   BatchHeader: [seqNum=540,count=1]
-r.NextRecord() = (rr, (000006-001.log: 93890), <nil>)
+r.NextRecord() = (rr, (000006-001.log: 93890), 94296 from previous files, <nil>)
   io.ReadAll(rr) = ("1d0200000000000005000000b68c7a260135dce1ce5c5498550793d15edfae62... <2055-byte record>", <nil>)
   BatchHeader: [seqNum=541,count=5]
-r.NextRecord() = (rr, (000006-001.log: 95956), EOF)
+r.NextRecord() = (rr, (000006-001.log: 95956), 94296 from previous files, EOF)
 
 # Test corrupting the tail of a batch that's large enough to be split into
 # multiple reads. Regression test for #3865.
@@ -365,10 +365,10 @@ r.NextRecord() = (rr, (000007.log: 55), <nil>)
 r.NextRecord() = (rr, (000007.log: 482), <nil>)
   io.ReadAll(rr) = ("12750100000000001d000000362bba27f0ed6f5433a12bc502873a27c67f256c... <199-byte record>", <nil>)
   BatchHeader: [seqNum=95506,count=29]
-r.NextRecord() = (rr, (000007-001.log: 0), <nil>)
+r.NextRecord() = (rr, (000007-001.log: 0), 692 from previous files, <nil>)
   io.ReadAll(rr) = ("2f75010000000000130000001ddd809cbb45782c44544a15a15dd52fb7b81a74... <45991-byte record>", <nil>)
   BatchHeader: [seqNum=95535,count=19]
-r.NextRecord() = (rr, (000007-001.log: 46013), <nil>)
+r.NextRecord() = (rr, (000007-001.log: 46013), 692 from previous files, <nil>)
   io.ReadAll(rr) = ("427501000000000013000000b30c11cf619ea65167511346cc55bb784a9af26f... <292-byte record>", <nil>)
   BatchHeader: [seqNum=95554,count=19]
-r.NextRecord() = (rr, (000007-001.log: 46316), EOF)
+r.NextRecord() = (rr, (000007-001.log: 46316), 692 from previous files, EOF)