Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 36 additions & 1 deletion pkg/wal/reader.go
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,10 @@ type LogRecordReader struct {
logger api.Logger
logFile *os.File
crc uint32
// fileSizeBytes is the size of the log file. A record never spans beyond
// the file it is in, so it bounds the length declared in a record header
// and prevents a corrupted or forged header from driving a huge allocation.
fileSizeBytes int64
}

func NewLogRecordReader(logger api.Logger, fileName string) (*LogRecordReader, error) {
Expand All @@ -42,6 +46,15 @@ func NewLogRecordReader(logger api.Logger, fileName string) (*LogRecordReader, e
return nil, err
}

fileInfo, err := r.logFile.Stat()
if err != nil {
_ = r.Close()

return nil, err
}

r.fileSizeBytes = fileInfo.Size()

_, err = r.logFile.Seek(0, io.SeekStart)
if err != nil {
_ = r.Close()
Expand Down Expand Up @@ -136,7 +149,7 @@ func (r *LogRecordReader) Read() (*protos.LogRecord, error) {
case protos.LogRecord_CRC_ANCHOR:
r.crc = crc
default:
return nil, fmt.Errorf("wal: unexpected LogRecord_Type: %v", record.Type)
return nil, fmt.Errorf("%w: %v", ErrUnexpectedRecordType, record.Type)
}

return record, nil
Expand All @@ -158,6 +171,28 @@ func (r *LogRecordReader) readHeader() (length, crc uint32, err error) {
length = uint32(header & recordLengthMask)
crc = uint32((header & recordCRCMask) >> 32)

// A record, together with its header, must fit in the remaining part of the
// file. Otherwise a corrupted header, especially near the end of the file,
// makes us allocate up to the whole file size before the read fails.
pos, err := r.logFile.Seek(0, io.SeekCurrent)
if err != nil {
return 0, 0, err
}

recSize := int64(length) + int64(getPadSize(int(length)))

if remaining := r.fileSizeBytes - pos; recSize > remaining {
r.logger.Warnf("Record length %d in file: %s exceeds the remaining file size %d", length, r.fileName, remaining)

// A header claiming more than the whole file can hold is corrupt. A
// record that only outgrows what is left is a torn tail.
if recSize > r.fileSizeBytes-int64(recordHeaderSize) {
return 0, 0, fmt.Errorf("%w: length=%d, file size=%d, file: %s", ErrRecordTooBig, length, r.fileSizeBytes, r.fileName)
}

return 0, 0, io.ErrUnexpectedEOF
}

return length, crc, nil
}

Expand Down
162 changes: 162 additions & 0 deletions pkg/wal/reader_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
package wal

import (
"encoding/binary"
"fmt"
"io"
"os"
Expand Down Expand Up @@ -270,6 +271,167 @@ func TestLogRecordReader(t *testing.T) {
assert.Contains(t, err.Error(), "wal: failed to unmarshal payload")
assert.Nil(t, record)
})

t.Run("record length in header is bounded", func(t *testing.T) {
fileName := filepath.Join(testDir, "test7.wal")
sourceName := filepath.Join(testDir, fmt.Sprintf(walFileTemplate, 1))
assert.NoError(t, copyFile(sourceName, fileName))

r, err := NewLogRecordReader(logger, fileName)
assert.NoError(t, err)
assertReadRecord(t, r, rec1, crc1)
assertReadRecord(t, r, rec2, crc2)
offset, err := r.logFile.Seek(0, io.SeekCurrent)
assert.NoError(t, err)
err = r.Close()
assert.NoError(t, err)

f, err := os.OpenFile(fileName, os.O_RDWR, walFilePermPrivateRW)
assert.NoError(t, err)
_, err = f.Seek(offset, io.SeekStart)
assert.NoError(t, err)
// a header claiming a record of 0xFFFFFFFF bytes, i.e. ~4GB
header := make([]byte, recordHeaderSize)
binary.LittleEndian.PutUint64(header, 0xFFFFFFFF)
_, err = f.Write(header)
assert.NoError(t, err)
err = f.Close()
assert.NoError(t, err)

r, err = NewLogRecordReader(logger, fileName)
assert.NoError(t, err)
assertReadRecord(t, r, rec1, crc1)
assertReadRecord(t, r, rec2, crc2)

record, err := r.Read()
assert.ErrorIs(t, err, ErrRecordTooBig)
assert.Nil(t, record)
assert.NoError(t, r.Close())
})

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Worth a case in TestWriteAheadLogFile_Repair as well — assertTestRepair only does assert.Error on the first ReadAll, so it can't tell a repairable error from a fatal one. This fails on the current branch and passes with the ReadAll change:

t.Run("oversized record header", func(t *testing.T) {
	dirPath := filepath.Join(testDir, "oversized")
	assert.NoError(t, os.MkdirAll(dirPath, walDirPermPrivateRWX))

	wal, err := Create(logger, dirPath, &Options{FileSizeBytes: 10 * 1024, BufferSizeBytes: 2048})
	assert.NoError(t, err)

	const NumBytes = 1024
	const NumRec = 102
	data1 := make([]byte, NumBytes)
	for m := range NumRec {
		for n := range NumBytes {
			data1[n] = byte(m)
		}
		assert.NoError(t, wal.Append(data1, false))
	}
	assert.NoError(t, wal.Close())

	// append a header claiming a ~4GB record to the last file
	names, err := dirReadWalNames(dirPath)
	assert.NoError(t, err)
	lastFile := filepath.Join(dirPath, names[len(names)-1])
	f, err := os.OpenFile(lastFile, os.O_RDWR, walFilePermPrivateRW)
	assert.NoError(t, err)
	_, err = f.Seek(0, io.SeekEnd)
	assert.NoError(t, err)
	header := make([]byte, recordHeaderSize)
	binary.LittleEndian.PutUint64(header, 0xFFFFFFFF)
	_, err = f.Write(header)
	assert.NoError(t, err)
	assert.NoError(t, f.Close())

	wal, err = Open(logger, dirPath, &Options{FileSizeBytes: 10 * 1024, BufferSizeBytes: 2048})
	assert.NoError(t, err)
	_, err = wal.ReadAll()
	assert.ErrorIs(t, err, io.ErrUnexpectedEOF)
	assert.NoError(t, wal.Close())

	assertTestRepair(t, logger, dirPath, NumRec)
})

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

fix

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good comment


t.Run("oversized record header", func(t *testing.T) {
dirPath := filepath.Join(testDir, "oversized")
assert.NoError(t, os.MkdirAll(dirPath, walDirPermPrivateRWX))

wal, err := Create(logger, dirPath, &Options{FileSizeBytes: 10 * 1024, BufferSizeBytes: 2048})
assert.NoError(t, err)

const NumBytes = 1024
const NumRec = 102
data1 := make([]byte, NumBytes)
for m := range NumRec {
for n := range NumBytes {
data1[n] = byte(m)
}
assert.NoError(t, wal.Append(data1, false))
}
assert.NoError(t, wal.Close())

// append a header claiming a ~4GB record to the last file
names, err := dirReadWalNames(dirPath)
assert.NoError(t, err)
lastFile := filepath.Join(dirPath, names[len(names)-1])
f, err := os.OpenFile(lastFile, os.O_RDWR, walFilePermPrivateRW)
assert.NoError(t, err)
_, err = f.Seek(0, io.SeekEnd)
assert.NoError(t, err)
header := make([]byte, recordHeaderSize)
binary.LittleEndian.PutUint64(header, 0xFFFFFFFF)
_, err = f.Write(header)
assert.NoError(t, err)
assert.NoError(t, f.Close())

wal, err = Open(logger, dirPath, &Options{FileSizeBytes: 10 * 1024, BufferSizeBytes: 2048})
assert.NoError(t, err)
_, err = wal.ReadAll()
assert.ErrorIs(t, err, io.ErrUnexpectedEOF)
assert.NoError(t, wal.Close())

assertTestRepair(t, logger, dirPath, NumRec)
})

t.Run("truncated payload after an intact header", func(t *testing.T) {
dirPath := filepath.Join(testDir, "truncated-payload")
assert.NoError(t, os.MkdirAll(dirPath, walDirPermPrivateRWX))

wal, err := Create(logger, dirPath, &Options{FileSizeBytes: 10 * 1024, BufferSizeBytes: 2048})
assert.NoError(t, err)

const NumBytes = 1024
const NumRec = 5
data1 := make([]byte, NumBytes)
for m := range NumRec {
for n := range NumBytes {
data1[n] = byte(m)
}
assert.NoError(t, wal.Append(data1, false))
}
assert.NoError(t, wal.Close())

// Append a header that declares a payload, without writing the payload.
// The header is intact and claims fewer bytes than the whole file, so the
// reader must report a torn tail, not a clean end of file.
names, err := dirReadWalNames(dirPath)
assert.NoError(t, err)
lastFile := filepath.Join(dirPath, names[len(names)-1])
f, err := os.OpenFile(lastFile, os.O_RDWR, walFilePermPrivateRW)
assert.NoError(t, err)
_, err = f.Seek(0, io.SeekEnd)
assert.NoError(t, err)
header := make([]byte, recordHeaderSize)
binary.LittleEndian.PutUint64(header, 100)
_, err = f.Write(header)
assert.NoError(t, err)
assert.NoError(t, f.Close())

wal, err = Open(logger, dirPath, &Options{FileSizeBytes: 10 * 1024, BufferSizeBytes: 2048})
assert.NoError(t, err)
_, err = wal.ReadAll()
assert.ErrorIs(t, err, io.ErrUnexpectedEOF)
assert.NoError(t, wal.Close())

assertTestRepair(t, logger, dirPath, NumRec)
})

t.Run("record bound follows the file size, not the default", func(t *testing.T) {
fileName := filepath.Join(testDir, "test8.wal")
sourceName := filepath.Join(testDir, fmt.Sprintf(walFileTemplate, 1))
assert.NoError(t, copyFile(sourceName, fileName))

fileInfo, err := os.Stat(fileName)
assert.NoError(t, err)

r, err := NewLogRecordReader(logger, fileName)
assert.NoError(t, err)
assert.Equal(t, fileInfo.Size(), r.fileSizeBytes)

assertReadRecord(t, r, rec1, crc1)
assertReadRecord(t, r, rec2, crc2)
offset, err := r.logFile.Seek(0, io.SeekCurrent)
assert.NoError(t, err)
assert.NoError(t, r.Close())

// A header claiming more than the file holds, as after a torn write.
f, err := os.OpenFile(fileName, os.O_RDWR, walFilePermPrivateRW)
assert.NoError(t, err)
_, err = f.Seek(offset, io.SeekStart)
assert.NoError(t, err)
header := make([]byte, recordHeaderSize)
binary.LittleEndian.PutUint64(header, uint64(fileInfo.Size()))
_, err = f.Write(header)
assert.NoError(t, err)
assert.NoError(t, f.Close())

r, err = NewLogRecordReader(logger, fileName)
assert.NoError(t, err)
assertReadRecord(t, r, rec1, crc1)
assertReadRecord(t, r, rec2, crc2)

record, err := r.Read()
assert.ErrorIs(t, err, ErrRecordTooBig)
assert.Nil(t, record)
assert.NoError(t, r.Close())
})
}

func assertReadRecord(t *testing.T, r *LogRecordReader, expRec *smartbftprotos.LogRecord, expCRC uint32) {
Expand Down
Loading
Loading