manifest,record: disambiguate unexpected EOFs

Disambiguate an unexpected EOF in the record envelope (eg, missing chunks) from
an unexpected EOF within the body of the record. An unexpected EOF of the
envelope may occur if the process crashes while appending a version edit to the
manifest file. An unexpected EOF within the record indicates corruption.

Previously, some version edit code paths re-mapped an unexpected EOF while
parsing the record envelope into a corruption error. In this case, recovery
failed improperly when it should've ignored the incomplete version edit.

In the other direction, some code paths would return io.ErrUnexpectedEOF while
decoding the version edit structure within a valid envelope record. In this
case, the caller interpreted the ErrUnexpectedEOF as a sudden end to the record
envelope, and recovery succeeded improperly when it should've aborted with a
corruption error.

Author: jbowens

internal/manifest/version_edit.go

@@ -25,8 +25,6 @@ import (
 // TODO(peter): describe the MANIFEST file format, independently of the C++
 // project.
 
-var errCorruptManifest = base.CorruptionErrorf("pebble: corrupt manifest")
-
 type byteReader interface {
 	io.ByteReader
 	io.Reader
@@ -551,7 +549,7 @@ func (v *VersionEdit) Decode(r io.Reader) error {
 			return base.CorruptionErrorf("column families are not supported")
 
 		default:
-			return errCorruptManifest
+			return base.CorruptionErrorf("MANIFEST: unknown tag: %d", tag)
 		}
 	}
 	return nil
@@ -847,7 +845,7 @@ func (d versionEditDecoder) readBytes() ([]byte, error) {
 	_, err = io.ReadFull(d, s)
 	if err != nil {
 		if err == io.ErrUnexpectedEOF {
-			return nil, errCorruptManifest
+			return nil, base.CorruptionErrorf("pebble: corrupt manifest: failed to read %d bytes", n)
 		}
 		return nil, err
 	}
@@ -860,7 +858,7 @@ func (d versionEditDecoder) readLevel() (int, error) {
 		return 0, err
 	}
 	if u >= NumLevels {
-		return 0, errCorruptManifest
+		return 0, base.CorruptionErrorf("pebble: corrupt manifest: level %d >= %d", u, NumLevels)
 	}
 	return int(u), nil
 }
@@ -876,8 +874,8 @@ func (d versionEditDecoder) readFileNum() (base.FileNum, error) {
 func (d versionEditDecoder) readUvarint() (uint64, error) {
 	u, err := binary.ReadUvarint(d)
 	if err != nil {
-		if err == io.EOF {
-			return 0, errCorruptManifest
+		if err == io.EOF || err == io.ErrUnexpectedEOF {
+			return 0, base.CorruptionErrorf("pebble: corrupt manifest: failed to read uvarint")
 		}
 		return 0, err
 	}

open.go

@@ -950,20 +950,29 @@ func (d *DB) replayWAL(
 			// surfaces record.ErrInvalidChunk or record.ErrZeroedChunk. These
 			// errors are always indicative of corruption and data loss.
 			//
-			// Otherwise, the reader surfaces io.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 `io.ErrUnexpectedEOF` as corruption and stop recovery.
+			// 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, io.ErrUnexpectedEOF) && !strictWALTail {
+			} else if errors.Is(err, record.ErrUnexpectedEOF) && !strictWALTail {
 				break
-			} else if errors.Is(err, record.ErrInvalidChunk) || errors.Is(err, record.ErrZeroedChunk) {
-				// If a read-ahead returns one of these errors, they should be marked with corruption.
-				// Other I/O related errors should not be marked with corruption and simply returned.
+			} 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)
 			}
 

record/record.go

@@ -215,13 +215,28 @@ var (
 	// header, length, or checksum. This usually occurs when a log is recycled,
 	// but can also occur due to corruption.
 	ErrInvalidChunk = errors.New("pebble/record: invalid chunk")
+
+	// ErrUnexpectedEOF is returned if a log file ends unexpectedly. It
+	// indicates the unexpected end of the log file or an in-progress record
+	// envelope itself. ErrUnexpectedEOF may be returned by Reader when it
+	// encounters an invalid chunk but observes no evidence that the invalid
+	// chunk is caused by corruption (i.e., no future chunk indicates that
+	// offset should be valid and durably synced.)
+	//
+	// This error is defined separately from io.ErrUnexpectedEOF to disambiguate
+	// this case from from the case of an unexpected end of the record's payload
+	// while decoding at a higher-level (eg, version edit decoding). If a
+	// higher-level decoding routine returns record.ErrUnexpectedEOF, it
+	// unambiguously indicates that the log file itself ended unexpectedly. The
+	// record.Reader will never return io.ErrUnexpectedEOF, just record.ErrUnexpectedEOF.
+	ErrUnexpectedEOF = errors.New("pebble/record: unexpected EOF")
 )
 
 // IsInvalidRecord returns true if the error matches one of the error types
 // returned for invalid records. These are treated in a way similar to io.EOF
 // in recovery code.
 func IsInvalidRecord(err error) bool {
-	return err == ErrZeroedChunk || err == ErrInvalidChunk || err == io.ErrUnexpectedEOF
+	return err == ErrZeroedChunk || err == ErrInvalidChunk || err == ErrUnexpectedEOF
 }
 
 // Reader reads records from an underlying io.Reader.
@@ -269,7 +284,7 @@ func NewReader(r io.Reader, logNum base.DiskFileNum) *Reader {
 		logNum:   uint32(logNum),
 		blockNum: -1,
 		// invalidOffset is initialized as MaxUint64 so that reading ahead
-		// with the old chunk wire formats results in io.ErrUnexpectedEOF.
+		// with the old chunk wire formats results in ErrUnexpectedEOF.
 		invalidOffset: math.MaxUint64,
 	}
 }
@@ -396,7 +411,7 @@ func (r *Reader) nextChunk(wantFirst bool) error {
 		if err != nil && err != io.ErrUnexpectedEOF {
 			if err == io.EOF && !wantFirst {
 				r.invalidOffset = uint64(r.blockNum)*blockSize + uint64(r.begin)
-				return io.ErrUnexpectedEOF
+				return ErrUnexpectedEOF
 			}
 			return err
 		}
@@ -454,20 +469,19 @@ func (r *Reader) readAheadForCorruption() error {
 		}
 
 		if errors.Is(err, io.EOF) {
-			// io.ErrUnexpectedEOF is returned instead of
-			// io.EOF because io library functions clear
-			// an error when it is io.EOF. io.ErrUnexpectedEOF
-			// is returned so that the error is not cleared
-			// when the io library makes calls to Reader.Read().
+			// ErrUnexpectedEOF is returned instead of io.EOF because io library
+			// functions clear an error when it is io.EOF. ErrUnexpectedEOF is
+			// returned so that the error is not cleared when the io library
+			// makes calls to Reader.Read().
 			//
-			// Since no sync offset was found to indicate that the
-			// invalid chunk should have been valid, the chunk represents
-			// an abrupt, unclean termination of the logical log. This
-			// abrupt end of file represented by io.ErrUnexpectedEOF.
+			// Since no sync offset was found to indicate that the invalid chunk
+			// should have been valid, the chunk represents an abrupt, unclean
+			// termination of the logical log. This abrupt end of file
+			// represented by ErrUnexpectedEOF.
 			if r.loggerForTesting != nil {
-				r.loggerForTesting.logf("\tEncountered io.EOF; returning io.ErrUnexpectedEOF since no sync offset found.\n")
+				r.loggerForTesting.logf("\tEncountered io.EOF; returning ErrUnexpectedEOF since no sync offset found.\n")
 			}
-			return io.ErrUnexpectedEOF
+			return ErrUnexpectedEOF
 		}
 		// The last block of a log can be less than 32KiB, which is
 		// the length of r.buf. Thus, we should still parse the data in

record/record_test.go

@@ -439,7 +439,7 @@ func TestCorruptBlock(t *testing.T) {
 	if err == nil {
 		t.Fatal("Expected a checksum mismatch error, got nil")
 	}
-	if err != io.ErrUnexpectedEOF {
+	if !errors.Is(err, ErrUnexpectedEOF) {
 		t.Fatalf("Unexpected error returned: %v", err)
 	}
 }
@@ -584,7 +584,7 @@ func TestSeekRecordEndOfFile(t *testing.T) {
 	if err == nil {
 		t.Fatalf("Seek past the end of a file didn't cause an error")
 	}
-	if err != io.ErrUnexpectedEOF {
+	if err != ErrUnexpectedEOF {
 		t.Fatalf("Seeking past EOF raised unexpected error: %v", err)
 	}
 }
@@ -673,8 +673,8 @@ func TestInvalidLogNum(t *testing.T) {
 
 	{
 		r := NewReader(bytes.NewReader(buf.Bytes()), 2)
-		if _, err := r.Next(); err != io.ErrUnexpectedEOF {
-			t.Fatalf("expected %s, but found %s\n", io.ErrUnexpectedEOF, err)
+		if _, err := r.Next(); !errors.Is(err, ErrUnexpectedEOF) {
+			t.Fatalf("expected %s, but found %s\n", ErrUnexpectedEOF, err)
 		}
 	}
 }
@@ -759,15 +759,15 @@ func TestRecycleLog(t *testing.T) {
 			rr, err := r.Next()
 			if err != nil {
 				// If we limited output then an EOF, zeroed, or invalid chunk is expected.
-				if limitedBuf.limit < 0 && (err == io.EOF || err == io.ErrUnexpectedEOF || err == ErrZeroedChunk || err == ErrInvalidChunk) {
+				if limitedBuf.limit < 0 && (err == io.EOF || IsInvalidRecord(err)) {
 					break
 				}
 				t.Fatalf("%d/%d: %v", i, j, err)
 			}
 			x, err := io.ReadAll(rr)
 			if err != nil {
 				// If we limited output then an EOF, zeroed, or invalid chunk is expected.
-				if limitedBuf.limit < 0 && (err == io.EOF || err == io.ErrUnexpectedEOF || err == ErrZeroedChunk || err == ErrInvalidChunk) {
+				if limitedBuf.limit < 0 && (err == io.EOF || IsInvalidRecord(err)) {
 					break
 				}
 				t.Fatalf("%d/%d: %v", i, j, err)
@@ -776,7 +776,7 @@ func TestRecycleLog(t *testing.T) {
 				t.Fatalf("%d/%d: expected record %d, but found %d", i, j, sizes[j], len(x))
 			}
 		}
-		if _, err := r.Next(); err != io.EOF && err != io.ErrUnexpectedEOF && err != ErrZeroedChunk && err != ErrInvalidChunk {
+		if _, err := r.Next(); err != io.EOF && err != ErrUnexpectedEOF && err != ErrZeroedChunk && err != ErrInvalidChunk {
 			t.Fatalf("%d: expected EOF, but found %v", i, err)
 		}
 	}
@@ -797,7 +797,7 @@ func TestTruncatedLog(t *testing.T) {
 	rr, err := r.Next()
 	require.NoError(t, err)
 	_, err = io.ReadAll(rr)
-	require.EqualValues(t, err, io.ErrUnexpectedEOF)
+	require.EqualValues(t, err, ErrUnexpectedEOF)
 }
 
 func TestRecycleLogWithPartialBlock(t *testing.T) {
@@ -901,7 +901,7 @@ func TestRecycleLogWithPartialRecord(t *testing.T) {
 	require.NoError(t, err)
 
 	_, err = io.ReadAll(rr)
-	require.Equal(t, err, io.ErrUnexpectedEOF)
+	require.True(t, errors.Is(err, ErrUnexpectedEOF))
 }
 
 type readerLogger struct {