record: add a bitflip check to WAL corruption

Checks if there was a bitflip that occurred when a chunk corruption
is found. Similar to #4406.

Author: EdwardX29

internal/bitflip/bitflip.go

@@ -0,0 +1,31 @@
+package bitflip
+
+// CheckSliceForBitFlip flips bits in data to see if it matches the expected checksum.
+// Returns the index and bit if successful.
+func CheckSliceForBitFlip(
+	data []byte, computeChecksum func([]byte) uint32, expectedChecksum uint32,
+) (found bool, indexFound int, bitFound int) {
+	// TODO(edward) This checking process likely can be made faster.
+	iterationLimit := 40 * (1 << 10) // 40KB
+	for i := 0; i < min(len(data), iterationLimit); i++ {
+		foundFlip, bit := checkByteForFlip(data, i, computeChecksum, expectedChecksum)
+		if foundFlip {
+			return true, i, bit
+		}
+	}
+	return false, 0, 0
+}
+
+func checkByteForFlip(
+	data []byte, i int, computeChecksum func([]byte) uint32, expectedChecksum uint32,
+) (found bool, bit int) {
+	for bit := 0; bit < 8; bit++ {
+		data[i] ^= (1 << bit)
+		var computedChecksum = computeChecksum(data)
+		data[i] ^= (1 << bit)
+		if computedChecksum == expectedChecksum {
+			return true, bit
+		}
+	}
+	return false, 0
+}

open_test.go

@@ -16,6 +16,7 @@ import (
 	"os"
 	"path/filepath"
 	"reflect"
+	"regexp"
 	"runtime"
 	"runtime/debug"
 	"slices"
@@ -1789,3 +1790,68 @@ func TestWALCorruption(t *testing.T) {
 	})
 	require.True(t, errors.Is(err, ErrCorruption))
 }
+
+func TestWALCorruptionBitFlip(t *testing.T) {
+	fs := vfs.NewMem()
+	d, err := Open("", testingRandomized(t, &Options{
+		FS:                 fs,
+		FormatMajorVersion: FormatWALSyncChunks,
+	}))
+	require.NoError(t, err)
+	require.NoError(t, d.Flush())
+
+	fourKiBValue := bytes.Repeat([]byte{'a'}, 4096)
+	for i := 1; i <= 32; i++ {
+		require.NoError(t, d.Set([]byte(fmt.Sprintf("key-%d", i)), fourKiBValue, Sync))
+	}
+	require.NoError(t, d.Close())
+
+	// We should have two WALs.
+	logs, err := fs.List("")
+	require.NoError(t, err)
+	logs = slices.DeleteFunc(logs, func(s string) bool { return filepath.Ext(s) != ".log" })
+	sort.Slice(logs, func(i, j int) bool {
+		return logs[i] < logs[j]
+	})
+	lastLog := logs[len(logs)-1]
+
+	// Corrupt the WAL by flipping one byte, 100 bytes from the end
+	// of the file.
+	f, err := fs.OpenReadWrite(lastLog, vfs.WriteCategoryUnspecified)
+	require.NoError(t, err)
+
+	buf := []byte{0}
+	_, err = f.ReadAt(buf, 100)
+	require.NoError(t, err)
+
+	bitToFlip := byte(1 << rand.IntN(8))
+	buf[0] ^= bitToFlip
+	_, err = f.WriteAt(buf, 100)
+	require.NoError(t, err)
+	require.NoError(t, f.Close())
+	t.Logf("zeroed one byte in %s at offset %d\n", lastLog, 100)
+
+	// Re-opening the database should detect and report the corruption and bit flip.
+	_, err = Open("", &Options{
+		FS:                 fs,
+		FormatMajorVersion: FormatWALSyncChunks,
+	})
+	require.True(t, errors.Is(err, ErrCorruption))
+
+	checkBitFlipErr := func(err error, t *testing.T) bool {
+		if err != nil {
+			details := errors.GetAllSafeDetails(err)
+			re := regexp.MustCompile(`bit flip found.+byte index \d+\. got: 0x[0-9A-Fa-f]{1,2}\. want: 0x[0-9A-Fa-f]{1,2}\.`)
+			for _, d := range details {
+				for _, s := range d.SafeDetails {
+					if re.MatchString(s) {
+						return true
+					}
+				}
+			}
+			require.Fail(t, "expected at least one detail to match bit flip found pattern", err)
+		}
+		return false
+	}
+	checkBitFlipErr(err, t)
+}

record/record.go

@@ -112,6 +112,7 @@ import (
 
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/bitflip"
 	"github.com/cockroachdb/pebble/internal/crc"
 )
 
@@ -359,9 +360,19 @@ func (r *Reader) nextChunk(wantFirst bool) error {
 				r.invalidOffset = uint64(r.blockNum)*blockSize + uint64(r.begin)
 				return ErrInvalidChunk
 			}
-			if checksum != crc.New(r.buf[r.begin-headerSize+6:r.end]).Value() {
+			data := r.buf[r.begin-headerSize+6 : r.end]
+			if checksum != crc.New(data).Value() {
+				computeChecksum := func(data []byte) uint32 { return crc.New(data).Value() }
+				// Check if there was a bit flip.
+				found, indexFound, bitFound := bitflip.CheckSliceForBitFlip(data, computeChecksum, checksum)
+				err := ErrInvalidChunk
+				if found {
+					err = errors.WithSafeDetails(err, ". bit flip found: block num %d. wal offset %d. byte index %d. got: 0x%x. want: 0x%x.",
+						errors.Safe(r.blockNum), errors.Safe(r.invalidOffset), errors.Safe(indexFound), errors.Safe(data[indexFound]), errors.Safe(data[indexFound]^(1<<bitFound)))
+				}
+
 				r.invalidOffset = uint64(r.blockNum)*blockSize + uint64(r.begin)
-				return ErrInvalidChunk
+				return err
 			}
 			if wantFirst {
 				if chunkPosition != fullChunkPosition && chunkPosition != firstChunkPosition {

sstable/block/block.go

@@ -17,6 +17,7 @@ import (
 	"github.com/cockroachdb/crlib/fifo"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/bitflip"
 	"github.com/cockroachdb/pebble/internal/cache"
 	"github.com/cockroachdb/pebble/internal/crc"
 	"github.com/cockroachdb/pebble/internal/invariants"
@@ -174,53 +175,30 @@ func ValidateChecksum(checksumType ChecksumType, b []byte, bh Handle) error {
 	if expectedChecksum != computedChecksum {
 		// Check if the checksum was due to a singular bit flip and report it.
 		data := slices.Clone(b[:bh.Length+1])
-		found, indexFound, bitFound := checkSliceForBitFlip(data, checksumType, expectedChecksum)
+		var checksumFunction func([]byte) uint32
+		switch checksumType {
+		case ChecksumTypeCRC32c:
+			checksumFunction = func(data []byte) uint32 {
+				return crc.New(data).Value()
+			}
+		case ChecksumTypeXXHash64:
+			checksumFunction = func(data []byte) uint32 {
+				return uint32(xxhash.Sum64(data))
+			}
+		}
+		found, indexFound, bitFound := bitflip.CheckSliceForBitFlip(data, checksumFunction, expectedChecksum)
 		err := base.CorruptionErrorf("block %d/%d: %s checksum mismatch %x != %x",
 			errors.Safe(bh.Offset), errors.Safe(bh.Length), checksumType,
 			expectedChecksum, computedChecksum)
 		if found {
-			err = errors.WithSafeDetails(err, ". bit flip found: byte index %d. got: %x. want: %x.",
-				indexFound, data[indexFound], data[indexFound]^(1<<bitFound))
+			err = errors.WithSafeDetails(err, ". bit flip found: byte index %d. got: 0x%x. want: 0x%x.",
+				errors.Safe(indexFound), errors.Safe(data[indexFound]), errors.Safe(data[indexFound]^(1<<bitFound)))
 		}
 		return err
 	}
 	return nil
 }
 
-func checkSliceForBitFlip(
-	data []byte, checksumType ChecksumType, expectedChecksum uint32,
-) (found bool, indexFound int, bitFound int) {
-	// TODO(edward) This checking process likely can be made faster.
-	iterationLimit := 40 * (1 << 10) // 40KB
-	for i := 0; i < min(len(data), iterationLimit); i++ {
-		foundFlip, bit := checkByteForFlip(data, i, checksumType, expectedChecksum)
-		if foundFlip {
-			return true, i, bit
-		}
-	}
-	return false, 0, 0
-}
-
-func checkByteForFlip(
-	data []byte, i int, checksumType ChecksumType, expectedChecksum uint32,
-) (found bool, bit int) {
-	for bit := 0; bit < 8; bit++ {
-		data[i] ^= (1 << bit)
-		var computedChecksum uint32
-		switch checksumType {
-		case ChecksumTypeCRC32c:
-			computedChecksum = crc.New(data).Value()
-		case ChecksumTypeXXHash64:
-			computedChecksum = uint32(xxhash.Sum64(data))
-		}
-		data[i] ^= (1 << bit)
-		if computedChecksum == expectedChecksum {
-			return true, bit
-		}
-	}
-	return false, 0
-}
-
 // Metadata is an in-memory buffer that stores metadata for a block. It is
 // allocated together with the buffer storing the block and is initialized once
 // when the block is read from disk.

sstable/reader_test.go

@@ -1512,15 +1512,15 @@ func TestReaderChecksumErrors(t *testing.T) {
 									checkBitFlipErr := func(err error) bool {
 										if err != nil {
 											details := errors.GetAllSafeDetails(err)
-											re := regexp.MustCompile(`bit flip found`)
+											re := regexp.MustCompile(`bit flip found.+byte index \d+\. got: 0x[0-9A-Fa-f]{1,2}\. want: 0x[0-9A-Fa-f]{1,2}\.`)
 											for _, d := range details {
 												for _, s := range d.SafeDetails {
 													if re.MatchString(s) {
 														return true
 													}
 												}
 											}
-											require.Fail(t, "expected at least one detail to match bit flip found", err)
+											require.Fail(t, "expected at least one detail to match bit flip pattern", err)
 										}
 										return false
 									}