Repository: cockroachdb/pebble

Issue:
manifest: version edit decode misinterprets ErrUnexpectedEOF
During version edit decoding, we read from an `io.Reader` backed by the `record` package. If a record in the MANIFEST is spread across multiple chunks, this reader handles reading new chunks as the caller pulls from the reader. It's possible for a record that is split across multiple chunks to be only partially persisted if it was in the process of being written when the process exited. When this happens, the underlying reader will surface an `io.ErrUnexpectedEOF`.

During version edit decoding, when we expect to read _n_ bytes from a record and find fewer than _n_ bytes in the payload, this gets surfaced from `io.ReadFull` as an `io.ErrUnexpectedEOF`. This case should be considered corruption, because the payload is improper.

Currently, because these scenarios both use `io.ErrUnexpectedEOF` these cases are indistinguishable. If the `io.ErrUnexpectedEOF` was surfaced by the underlying record reader, we want to treat this as a logical EOF, not an error.


Jira issue: PEBBLE-404

Discussion:

Solution patch:
diff --git a/internal/manifest/version_edit.go b/internal/manifest/version_edit.go
index 28b2799b..334d40fe 100644
--- a/internal/manifest/version_edit.go
+++ b/internal/manifest/version_edit.go
@@ -24,8 +24,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
@@ -473,7 +471,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
@@ -731,7 +729,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
 	}
@@ -744,7 +742,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
 }
@@ -760,8 +758,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
 	}
diff --git a/record/record.go b/record/record.go
index 7b6d5258..db4e915c 100644
--- a/record/record.go
+++ b/record/record.go
@@ -148,13 +148,26 @@ var (
 	// header, length, or checksum. This usually occurs when a log is recycled,
 	// but can also occur due to corruption.
 	ErrInvalidChunk = base.CorruptionErrorf("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.
+	//
+	// This error is defined separately from io.ErrUnexpectedEOF to disambiguate
+	// this case 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.
@@ -283,7 +296,7 @@ func (r *Reader) nextChunk(wantFirst bool) error {
 		n, err := io.ReadFull(r.r, r.buf[:])
 		if err != nil && err != io.ErrUnexpectedEOF {
 			if err == io.EOF && !wantFirst {
-				return io.ErrUnexpectedEOF
+				return ErrUnexpectedEOF
 			}
 			return err
 		}



Test patch:
diff --git a/compaction_test.go b/compaction_test.go
index 89b82606..15597ac3 100644
--- a/compaction_test.go
+++ b/compaction_test.go
@@ -1384,6 +1384,17 @@ func TestCompactionDeleteOnlyHints(t *testing.T) {
 	datadriven.RunTest(t, "testdata/compaction_delete_only_hints",
 		func(t *testing.T, td *datadriven.TestData) string {
 			switch td.Cmd {
+			case "batch":
+				b := d.NewBatch()
+				err := runBatchDefineCmd(td, b)
+				if err != nil {
+					return err.Error()
+				}
+				if err = b.Commit(Sync); err != nil {
+					return err.Error()
+				}
+				return ""
+
 			case "define":
 				opts, err = reset()
 				if err != nil {
@@ -1398,6 +1409,12 @@ func TestCompactionDeleteOnlyHints(t *testing.T) {
 				d.mu.Unlock()
 				return s
 
+			case "flush":
+				if err := d.Flush(); err != nil {
+					return err.Error()
+				}
+				return runLSMCmd(td, d)
+
 			case "force-set-hints":
 				d.mu.Lock()
 				defer d.mu.Unlock()
@@ -1548,6 +1565,13 @@ func TestCompactionDeleteOnlyHints(t *testing.T) {
 				}
 				return ""
 
+			case "snapshot":
+				// NB: It's okay that we're letting the snapshot out of scope
+				// without closing it; the close snapshot command will pull the
+				// relevant seqnum off the DB to close it.
+				s := d.NewSnapshot()
+				return fmt.Sprintf("snapshot seqnum: %d", s.seqNum)
+
 			case "ingest":
 				if err = runBuildCmd(td, d, d.opts.FS); err != nil {
 					return err.Error()
diff --git a/record/record_test.go b/record/record_test.go
index 885d221f..a8b0880d 100644
--- a/record/record_test.go
+++ b/record/record_test.go
@@ -733,7 +733,7 @@ func TestSeekRecord(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)
 	}
 	r.recover() // Verify recovery works.
@@ -912,7 +912,7 @@ 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 == ErrZeroedChunk || err == ErrInvalidChunk) {
+				if limitedBuf.limit < 0 && (err == io.EOF || IsInvalidRecord(err)) {
 					break
 				}
 				t.Fatalf("%d/%d: %v", i, j, err)
@@ -920,7 +920,7 @@ func TestRecycleLog(t *testing.T) {
 			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 == ErrZeroedChunk || err == ErrInvalidChunk) {
+				if limitedBuf.limit < 0 && (err == io.EOF || IsInvalidRecord(err)) {
 					break
 				}
 				t.Fatalf("%d/%d: %v", i, j, err)
@@ -929,7 +929,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 != 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)
 		}
 	}
@@ -948,7 +948,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) {
diff --git a/testdata/compaction_delete_only_hints b/testdata/compaction_delete_only_hints
index 52668e56..96f7322f 100644
--- a/testdata/compaction_delete_only_hints
+++ b/testdata/compaction_delete_only_hints
@@ -702,3 +702,56 @@ o: (o, .)
 get-hints
 ----
 (none)
+
+
+# Reset the database.
+
+reset
+----
+
+# Simulate an ingestion of a CockroachDB snapshot sstable with a range delete and a
+# range key delete covering the keyspace.
+
+ingest ext
+set c c
+set d d
+set e e
+set f f
+set g g
+del-range c h
+range-key-del c h
+----
+OK
+
+describe-lsm
+----
+L6:
+  000004:[c#10,RANGEKEYDEL-h#inf,RANGEDEL]
+
+snapshot
+----
+snapshot seqnum: 11
+
+# Simulate a replica removal of the previously ingested snapshot through a range
+# delete and range key delete.
+
+batch
+del-range c h
+range-key-del c h
+----
+
+flush
+----
+L0.0:
+  000006:[c#12,RANGEKEYDEL-h#inf,RANGEDEL]
+L6:
+  000004:[c#10,RANGEKEYDEL-h#inf,RANGEDEL]
+
+get-hints
+----
+L0.000006 c-h seqnums(tombstone=11-12, file-smallest=10, type=point-and-range-key)
+
+close-snapshot
+11
+----
+[JOB 100] compacted(delete-only) L6 [000004] (987B) Score=0.00 -> L6 [] (0B), in 1.0s (2.0s total), output rate 0B/s
diff --git a/version_set_test.go b/version_set_test.go
index 271279fa..4397e322 100644
--- a/version_set_test.go
+++ b/version_set_test.go
@@ -5,6 +5,7 @@
 package pebble
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"io"
@@ -13,6 +14,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 
 	"github.com/cockroachdb/datadriven"
@@ -24,6 +26,7 @@ import (
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/cockroachdb/pebble/vfs/atomicfs"
+	"github.com/cockroachdb/pebble/vfs/errorfs"
 	"github.com/stretchr/testify/require"
 )
 
@@ -317,3 +320,78 @@ func TestVersionSetSeqNums(t *testing.T) {
 	// logSeqNum is always one greater than the last assigned sequence number.
 	require.Equal(t, d.mu.versions.logSeqNum.Load(), lastSeqNum+1)
 }
+
+// TestCrashDuringManifestWrite_LargeKeys tests a crash mid-manifest write. It
+// uses randomly-sized keys with a very high max in order to test version edits
+// of variable sizes. Large version edits may be broken into multiple 'chunks'
+// across multiple 32KiB blocks within the record package's encoding. There have
+// previously been issues specifically decoding these multi-block version edits.
+func TestCrashDuringManifestWrite_LargeKeys(t *testing.T) {
+	seed := rand.Uint64()
+	t.Logf("seed: %d", seed)
+	rng := rand.New(rand.NewPCG(seed, 0))
+
+	// crashClone is nil until a clone of the memFS is constructed, where the
+	// clone will lose 50% of the unsynced data. Each iteration constructs one
+	// clone at a random time, and the DB keeps setting values until the clone
+	// is created. Then a new DB is opened with the cloned memFS.
+	var crashClone atomic.Pointer[vfs.MemFS]
+	makeFS := func(iter uint64) vfs.FS {
+		memFS := vfs.NewCrashableMem()
+		return errorfs.Wrap(memFS, errorfs.InjectorFunc(func(op errorfs.Op) error {
+			if crashClone.Load() != nil || op.Kind != errorfs.OpFileWrite {
+				return nil
+			}
+			typ, _, ok := base.ParseFilename(memFS, memFS.PathBase(op.Path))
+			if !ok || typ != base.FileTypeManifest {
+				return nil
+			}
+			if rng.IntN(5) == 0 {
+				crashClone.Store(memFS.CrashClone(vfs.CrashCloneCfg{
+					UnsyncedDataPercent: 50,
+					RNG:                 rand.New(rand.NewPCG(seed, iter)),
+				}))
+			}
+			return nil
+		}))
+	}
+
+	opts := &Options{Logger: testLogger{t: t}}
+	lel := MakeLoggingEventListener(opts.Logger)
+	opts.EventListener = &lel
+
+	k := slices.Concat([]byte("averyl"), bytes.Repeat([]byte{'o'}, rng.IntN(100000)), []byte("ngkey"))
+	baseLen := len(k)
+	newKey := func(i int) []byte {
+		return append(k[:baseLen], fmt.Sprintf("%10d", i)...)
+	}
+
+	const numIterations = 10
+	var keyIndex int
+	for i := 0; i < numIterations; i++ {
+		func() {
+			crashClone.Store(nil)
+
+			opts.FS = makeFS(uint64(i))
+			d, err := Open("foo", opts)
+			require.NoError(t, err)
+			func() {
+				defer func() { require.NoError(t, d.Close()) }()
+				for j := 0; crashClone.Load() == nil; j++ {
+					keyIndex++
+					require.NoError(t, d.Set(newKey(keyIndex), []byte("value"), Sync))
+					if j%10 == 0 {
+						_, err := d.AsyncFlush()
+						require.NoError(t, err)
+					}
+				}
+				require.NotNil(t, crashClone.Load())
+			}()
+
+			opts.FS = crashClone.Load()
+			d, err = Open("foo", opts)
+			require.NoError(t, err)
+			require.NoError(t, d.Close())
+		}()
+	}
+}