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/compaction.go b/compaction.go
index 0166758d..103493d7 100644
--- a/compaction.go
+++ b/compaction.go
@@ -2125,10 +2125,9 @@ type deleteCompactionHint struct {
 	// The type of key span that generated this hint (point key, range key, or
 	// both).
 	hintType deleteCompactionHintType
-	// start and end are user keys specifying a key range [start, end) of
-	// deleted keys.
-	start []byte
-	end   []byte
+	// bounds specifies the bounds of the keyspace deleted by the hint's
+	// tombstones.
+	bounds base.UserKeyBounds
 	// The level of the file containing the range tombstone(s) when the hint
 	// was created. Only lower levels need to be searched for files that may
 	// be deleted.
@@ -2153,8 +2152,8 @@ type deleteCompactionHint struct {
 
 func (h deleteCompactionHint) String() string {
 	return fmt.Sprintf(
-		"L%d.%s %s-%s seqnums(tombstone=%d-%d, file-smallest=%d, type=%s)",
-		h.tombstoneLevel, h.tombstoneFile.FileNum, h.start, h.end,
+		"L%d.%s %s seqnums(tombstone=%d-%d, file-smallest=%d, type=%s)",
+		h.tombstoneLevel, h.tombstoneFile.FileNum, h.bounds,
 		h.tombstoneSmallestSeqNum, h.tombstoneLargestSeqNum, h.fileSmallestSeqNum,
 		h.hintType,
 	)
@@ -2200,7 +2199,8 @@ func (h *deleteCompactionHint) canDelete(cmp Compare, m *fileMetadata, snapshots
 	}
 
 	// The file's keys must be completely contained within the hint range.
-	return cmp(h.start, m.Smallest.UserKey) <= 0 && cmp(m.Largest.UserKey, h.end) < 0
+	tableBounds := m.UserKeyBounds()
+	return h.bounds.ContainsBounds(cmp, &tableBounds)
 }
 
 func (d *DB) maybeUpdateDeleteCompactionHints(c *compaction) {
@@ -2226,7 +2226,8 @@ func (d *DB) maybeUpdateDeleteCompactionHints(c *compaction) {
 		// If the compaction's key space is disjoint from the hint's key
 		// space, the zeroing of sequence numbers won't affect the hint. Keep
 		// the hint.
-		keysDisjoint := d.cmp(h.end, c.smallest.UserKey) < 0 || d.cmp(h.start, c.largest.UserKey) > 0
+		compactionBounds := base.UserKeyBoundsFromInternal(c.smallest, c.largest)
+		keysDisjoint := !h.bounds.Overlaps(d.cmp, &compactionBounds)
 		if keysDisjoint {
 			updatedHints = append(updatedHints, h)
 			continue
@@ -2311,7 +2312,7 @@ func checkDeleteCompactionHints(
 		// The hint h will be resolved and dropped, regardless of whether
 		// there are any tables that can be deleted.
 		for l := h.tombstoneLevel + 1; l < numLevels; l++ {
-			overlaps := v.Overlaps(l, base.UserKeyBoundsEndExclusive(h.start, h.end))
+			overlaps := v.Overlaps(l, h.bounds)
 			iter := overlaps.Iter()
 			for m := iter.First(); m != nil; m = iter.Next() {
 				if m.IsCompacting() || !h.canDelete(cmp, m, snapshots) || files[m] {
diff --git a/internal/manifest/version_edit.go b/internal/manifest/version_edit.go
index 2af5ef29..868aee05 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
@@ -471,7 +469,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
@@ -727,7 +725,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
 	}
@@ -740,7 +738,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
 }
@@ -756,8 +754,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
 		}
diff --git a/table_stats.go b/table_stats.go
index ea47bd06..e4f29e8d 100644
--- a/table_stats.go
+++ b/table_stats.go
@@ -7,6 +7,7 @@ package pebble
 import (
 	"fmt"
 	"math"
+	"slices"
 
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
@@ -447,18 +448,16 @@ func (d *DB) loadTableRangeDelStats(
 		if hintSeqNum == math.MaxUint64 {
 			continue
 		}
+		bounds := base.UserKeyBoundsEndExclusive(slices.Clone(start), slices.Clone(end))
 		hint := deleteCompactionHint{
 			hintType:                hintType,
-			start:                   make([]byte, len(start)),
-			end:                     make([]byte, len(end)),
+			bounds:                  bounds,
 			tombstoneFile:           meta,
 			tombstoneLevel:          level,
 			tombstoneLargestSeqNum:  s.LargestSeqNum(),
 			tombstoneSmallestSeqNum: s.SmallestSeqNum(),
 			fileSmallestSeqNum:      hintSeqNum,
 		}
-		copy(hint.start, start)
-		copy(hint.end, end)
 		compactionHints = append(compactionHints, hint)
 	}
 	if err != nil {



Test patch:
diff --git a/compaction_test.go b/compaction_test.go
index 953a54d7..55cad089 100644
--- a/compaction_test.go
+++ b/compaction_test.go
@@ -1825,6 +1825,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 {
@@ -1839,6 +1850,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()
@@ -1872,8 +1889,7 @@ func TestCompactionDeleteOnlyHints(t *testing.T) {
 
 					h := deleteCompactionHint{
 						hintType:                hintType,
-						start:                   start,
-						end:                     end,
+						bounds:                  base.UserKeyBoundsEndExclusive(start, end),
 						fileSmallestSeqNum:      parseUint64(parts[4]),
 						tombstoneLevel:          tombstoneLevel,
 						tombstoneFile:           tombstoneFile,
@@ -1992,6 +2008,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 d0520793..71d05afb 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 84e0cfa1..001a00a6 100644
--- a/testdata/compaction_delete_only_hints
+++ b/testdata/compaction_delete_only_hints
@@ -49,12 +49,12 @@ L4:
 
 get-hints
 ----
-L0.000004 b-r seqnums(tombstone=200-230, file-smallest=90, type=point-key-only)
+L0.000004 [b, r) seqnums(tombstone=200-230, file-smallest=90, type=point-key-only)
 
 maybe-compact
 ----
 Deletion hints:
-  L0.000004 b-r seqnums(tombstone=200-230, file-smallest=90, type=point-key-only)
+  L0.000004 [b, r) seqnums(tombstone=200-230, file-smallest=90, type=point-key-only)
 Compactions:
   (none)
 
@@ -81,7 +81,7 @@ L4:
 
 get-hints
 ----
-L0.000004 b-r seqnums(tombstone=200-230, file-smallest=90, type=point-key-only)
+L0.000004 [b, r) seqnums(tombstone=200-230, file-smallest=90, type=point-key-only)
 
 maybe-compact
 ----
@@ -119,8 +119,8 @@ L4:
 
 get-hints
 ----
-L0.000004 a-k seqnums(tombstone=300-300, file-smallest=110, type=point-key-only)
-L1.000005 b-r seqnums(tombstone=200-230, file-smallest=90, type=point-key-only)
+L0.000004 [a, k) seqnums(tombstone=300-300, file-smallest=110, type=point-key-only)
+L1.000005 [b, r) seqnums(tombstone=200-230, file-smallest=90, type=point-key-only)
 
 maybe-compact
 ----
@@ -152,7 +152,7 @@ L4:
 
 get-hints
 ----
-L0.000004 b-r seqnums(tombstone=200-230, file-smallest=90, type=point-key-only)
+L0.000004 [b, r) seqnums(tombstone=200-230, file-smallest=90, type=point-key-only)
 
 compact a-z
 ----
@@ -194,7 +194,7 @@ L4:
 
 get-hints
 ----
-L1.000004 b-r seqnums(tombstone=200-230, file-smallest=90, type=point-key-only)
+L1.000004 [b, r) seqnums(tombstone=200-230, file-smallest=90, type=point-key-only)
 
 # Tables 000005 and 000006 can be deleted as their largest sequence numbers fall
 # below the smallest sequence number of the range del. Table 000007 falls
@@ -240,13 +240,13 @@ get-hints
 force-set-hints
 L0.000001 a-z 0 5-27 point_key_only
 ----
-L0.000001 a-z seqnums(tombstone=5-27, file-smallest=0, type=point-key-only)
+L0.000001 [a, z) seqnums(tombstone=5-27, file-smallest=0, type=point-key-only)
 
 # Hints on the table are unchanged, as the new sstable is at L6, and hints are
 # not generated on tables at this level.
 get-hints
 ----
-L0.000001 a-z seqnums(tombstone=5-27, file-smallest=0, type=point-key-only)
+L0.000001 [a, z) seqnums(tombstone=5-27, file-smallest=0, type=point-key-only)
 
 # Closing snapshot 10 triggers an elision-only compaction in L6 rather than a
 # deletion-only compaction, as the earliest snapshot that remains open is 25,
@@ -405,9 +405,9 @@ L6:
 
 get-hints
 ----
-L0.000013 a-i seqnums(tombstone=19-19, file-smallest=12, type=point-key-only)
-L0.000013 i-r seqnums(tombstone=19-19, file-smallest=13, type=point-and-range-key)
-L0.000013 r-z seqnums(tombstone=19-19, file-smallest=17, type=range-key-only)
+L0.000013 [a, i) seqnums(tombstone=19-19, file-smallest=12, type=point-key-only)
+L0.000013 [i, r) seqnums(tombstone=19-19, file-smallest=13, type=point-and-range-key)
+L0.000013 [r, z) seqnums(tombstone=19-19, file-smallest=17, type=range-key-only)
 
 maybe-compact
 ----
@@ -415,3 +415,62 @@ Deletion hints:
   (none)
 Compactions:
   [JOB 100] compacted(delete-only) L6 [000006 000007 000008 000009 000011] (3.5KB) Score=0.00 -> L6 [] (0B), in 1.0s (2.0s total), output rate 0B/s
+
+# 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] (871B) Score=0.00 -> L6 [] (0B), in 1.0s (2.0s total), output rate 0B/s
+
+maybe-compact
+----
+Deletion hints:
+  (none)
+Compactions:
+  (none)
diff --git a/version_set_test.go b/version_set_test.go
index f38f9678..aad26acd 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.NewSource(int64(seed)))
+
+	// 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 crashed atomic.Bool
+	var memFS *vfs.MemFS
+	makeFS := func(iter uint64) vfs.FS {
+		memFS = vfs.NewStrictMem()
+		return errorfs.Wrap(memFS, errorfs.InjectorFunc(func(op errorfs.Op) error {
+			if crashed.Load() || 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 {
+				memFS.SetIgnoreSyncs(true)
+				crashed.Store(true)
+			}
+			return nil
+		}))
+	}
+
+	opts := &Options{Logger: testLogger{t: t}}
+	lel := MakeLoggingEventListener(opts.Logger)
+	opts.EventListener = &lel
+
+	k := append(append([]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() {
+			crashed.Store(false)
+
+			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; !crashed.Load(); j++ {
+					keyIndex++
+					require.NoError(t, d.Set(newKey(keyIndex), []byte("value"), Sync))
+					if j%10 == 0 {
+						_, err := d.AsyncFlush()
+						require.NoError(t, err)
+					}
+				}
+			}()
+			memFS.SetIgnoreSyncs(false)
+			memFS.ResetToSyncedState()
+
+			opts.FS = memFS
+			d, err = Open("foo", opts)
+			require.NoError(t, err)
+			require.NoError(t, d.Close())
+		}()
+	}
+}