diff --git a/pkg/wal/reader.go b/pkg/wal/reader.go index 186c8ced..b5ac1700 100644 --- a/pkg/wal/reader.go +++ b/pkg/wal/reader.go @@ -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) { @@ -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() @@ -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 @@ -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 } diff --git a/pkg/wal/reader_test.go b/pkg/wal/reader_test.go index b5cb1fcb..403e6da4 100644 --- a/pkg/wal/reader_test.go +++ b/pkg/wal/reader_test.go @@ -6,6 +6,7 @@ package wal import ( + "encoding/binary" "fmt" "io" "os" @@ -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()) + }) + + 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) { diff --git a/pkg/wal/util.go b/pkg/wal/util.go index 31d694a9..8a37ca88 100644 --- a/pkg/wal/util.go +++ b/pkg/wal/util.go @@ -6,6 +6,7 @@ package wal import ( + "encoding/binary" "errors" "fmt" "io" @@ -16,6 +17,8 @@ import ( "strings" "github.com/hyperledger/SmartBFT/pkg/api" + protos "github.com/hyperledger/SmartBFT/smartbftprotos" + "google.golang.org/protobuf/proto" ) var padTable [][]byte @@ -157,13 +160,40 @@ func parseWalFileName(fileName string) (index uint64, err error) { return index, nil } +// copyFile copies the source file to the target, readable and writable by the owner only, +// just like the WAL files themselves (walFilePermPrivateRW). Permissions of an already +// existing target are tightened before anything is written to it, since the open mode +// applies only when the file is created. func copyFile(source, target string) error { - input, err := os.ReadFile(source) + input, err := os.Open(source) if err != nil { return err } + defer input.Close() - return os.WriteFile(target, input, 0o644) + output, err := os.OpenFile(target, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, walFilePermPrivateRW) + if err != nil { + return err + } + + if err = output.Chmod(walFilePermPrivateRW); err != nil { + _ = output.Close() + return err + } + + if _, err = io.Copy(output, input); err != nil { + _ = output.Close() + return err + } + + // The copy is the only evidence left once scanRepairFile truncates or + // deletes the original, so flush it to disk before returning. + if err = output.Sync(); err != nil { + _ = output.Close() + return err + } + + return output.Close() } func truncateCloseFile(f *os.File, offset int64) error { @@ -235,6 +265,10 @@ func scanVerify1(logger api.Logger, r *LogRecordReader, i int, crc uint32) (num // scanRepairFile scans the file to the last good record and truncates after it. If even the CRC-Anchor cannot be // read, the file is deleted. +// +// Only corruption of the last record is repaired. Anything else is refused with +// ErrNotRepairable, since truncating at a corrupt record in the middle of the file would +// drop records that were written before it, possibly ones that were already committed. func scanRepairFile(logger api.Logger, lastFile string) error { logger.Debugf("Trying to repair file: %s", lastFile) @@ -266,6 +300,9 @@ func scanRepairFile(logger api.Logger, lastFile string) error { if readErr != nil { logger.Debugf("Read error: %s", readErr) + // offset is the end of the last good record, hence the start of the corrupt one. + corruptRecordStart := offset + if closeErr := r.Close(); closeErr != nil { return closeErr } @@ -276,6 +313,26 @@ func scanRepairFile(logger api.Logger, lastFile string) error { return nil // no need to repair } + torn, err := isTornTail(logger, lastFile, corruptRecordStart) + if err != nil { + return err + } + + if !torn { + logger.Errorf( + "Write-Ahead-Log will not repair file: %s; the corrupt record at offset %d is followed by further records, truncating it would drop them", + lastFile, + corruptRecordStart, + ) + + return fmt.Errorf( + "%w: corrupt record at offset %d of file: %s", + ErrNotRepairable, + corruptRecordStart, + lastFile, + ) + } + break } // keep the end offset of a good record @@ -283,6 +340,8 @@ func scanRepairFile(logger api.Logger, lastFile string) error { offset, err = r.logFile.Seek(0, io.SeekCurrent) if err != nil { + _ = r.Close() + return err } } @@ -304,3 +363,238 @@ func scanRepairFile(logger api.Logger, lastFile string) error { return nil } + +// isTornTail reports whether the record starting at recordStart is the last one in the +// file, i.e. whether nothing was written after it. +// +// Repair truncates the file after the last good record, which is safe only for a torn +// write of the last record. If real records follow the corrupt one, they were written +// before the crash and may already be committed, so truncating would silently restart the +// node from a state in the past. +// +// The corrupt record cannot be parsed, so its length is unknown and the offset of the next +// record cannot be computed. When, however, only its payload or CRC is corrupt, the header +// still holds the true length, and the next record starts exactly after it - that position +// is probed first, and the first record there is verified against the corrupt header's CRC. +// A record that merely parses at that exact position is also treated as evidence that +// records follow. Otherwise we resynchronize in a single pass: every offset at which a +// record could start is probed once, and a remainder is accepted only if no readable link +// is found from any of them. Requiring a readable record at *some* offset keeps this +// conservative in the safe direction - a false positive refuses the repair, never performs +// a lossy one. +func isTornTail(logger api.Logger, fileName string, recordStart int64) (bool, error) { + f, err := os.Open(fileName) + if err != nil { + return false, err + } + defer f.Close() + + fileInfo, err := f.Stat() + if err != nil { + return false, err + } + + fileSize := fileInfo.Size() + if recordStart >= fileSize { + return true, nil + } + + // The corrupt record cannot be parsed. Read the rest of the file once and + // resynchronize over the bytes in memory: probing every candidate offset + // with its own ReadAt would issue one syscall per offset. + tail := make([]byte, fileSize-recordStart) + if _, err := f.ReadAt(tail, recordStart); err != nil { + return false, err + } + + var corruptCRC uint32 + if len(tail) >= recordHeaderSize { + corruptCRC = uint32((binary.LittleEndian.Uint64(tail) & recordCRCMask) >> 32) + } + + if next, ok := nextRecordOffset(tail); ok { + // The corrupt header still holds the CRC that the next record chains + // from, so a single following record can be verified. A record that + // merely parses at the exact next offset is also enough: refusing the + // repair is safe, truncating is not. + if readableRecordsFollowFrom(tail, next, corruptCRC, true) || readableRecordAt(tail, next) { + logger.Debugf( + "Corrupt record at offset %d of file: %s is followed by readable records from offset %d", + recordStart, + fileName, + recordStart+next, + ) + + return false, nil + } + } + + if readableLinkPresent(tail, corruptCRC) { + logger.Debugf( + "Corrupt record at offset %d of file: %s is followed by readable records", + recordStart, + fileName, + ) + + return false, nil + } + + return true, nil +} + +// readableLinkPresent reports whether data holds two consecutive readable +// records with a verified CRC link. The link is verified either against seedCRC, +// the CRC of the corrupt record before data, or between consecutive records. +// Records are interpreted as LogRecordReader.Read interprets them: an anchor +// reseeds the chain from its header CRC instead of being verified. Every aligned +// offset is parsed once, so the scan does not restart the chain per candidate. +func readableLinkPresent(data []byte, seedCRC uint32) bool { + for pos := int64(recordHeaderSize); pos+int64(recordHeaderSize) <= int64(len(data)); pos += int64(recordHeaderSize) { + current, ok := parseRecordAt(data, pos) + if !ok { + continue + } + + if current.recordType != protos.LogRecord_CRC_ANCHOR && verifyCRC(seedCRC, current.crc, current.payload) { + return true + } + + next, ok := parseRecordAt(data, current.next) + if !ok { + continue + } + + if next.recordType != protos.LogRecord_CRC_ANCHOR { + if verifyCRC(current.crc, next.crc, next.payload) { + return true + } + + continue + } + + afterAnchor, ok := parseRecordAt(data, next.next) + if !ok || afterAnchor.recordType == protos.LogRecord_CRC_ANCHOR { + continue + } + + if verifyCRC(next.crc, afterAnchor.crc, afterAnchor.payload) { + return true + } + } + + return false +} + +// nextRecordOffset returns the offset, relative to the start of data, at which +// the record following the one at the start of data begins, based on the length +// in its header. It reports false when the header does not hold a length that +// leaves room for another record, i.e. when the header is corrupt too. +func nextRecordOffset(data []byte) (int64, bool) { + if len(data) < recordHeaderSize { + return 0, false + } + + length := int64(binary.LittleEndian.Uint64(data) & recordLengthMask) + payloadLength := length + int64(getPadSize(int(length))) + + next := int64(recordHeaderSize) + payloadLength + if payloadLength == 0 || next+int64(recordHeaderSize) > int64(len(data)) { + return 0, false + } + + return next, true +} + +type recordView struct { + recordType protos.LogRecord_Type + payload []byte + crc uint32 + next int64 +} + +// parseRecordAt parses the record starting at pos. It reports false when the +// record is torn, does not unmarshal, or has an unknown type. +func parseRecordAt(data []byte, pos int64) (recordView, bool) { + var view recordView + + if pos+int64(recordHeaderSize) > int64(len(data)) { + return view, false + } + + header := binary.LittleEndian.Uint64(data[pos:]) + length := int64(header & recordLengthMask) + view.crc = uint32((header & recordCRCMask) >> 32) + + payloadLength := length + int64(getPadSize(int(length))) + if payloadLength == 0 || pos+int64(recordHeaderSize)+payloadLength > int64(len(data)) { + return view, false + } + + view.payload = data[pos+int64(recordHeaderSize) : pos+int64(recordHeaderSize)+payloadLength] + + record := &protos.LogRecord{} + if err := proto.Unmarshal(view.payload[:length], record); err != nil { + return view, false + } + + switch record.Type { + case protos.LogRecord_ENTRY, protos.LogRecord_CONTROL, protos.LogRecord_CRC_ANCHOR: + view.recordType = record.Type + default: + return view, false + } + + view.next = pos + int64(recordHeaderSize) + payloadLength + + return view, true +} + +// readableRecordAt reports whether a structurally valid record starts at offset. +func readableRecordAt(data []byte, offset int64) bool { + _, ok := parseRecordAt(data, offset) + + return ok +} + +// readableRecordsFollow reports whether data holds readable records from offset. +// The CRC of the first record is not checked, since it depends on the corrupt +// record. One verified CRC link between consecutive records is enough; a lone +// record that merely decodes proves nothing. +func readableRecordsFollow(data []byte, offset int64) bool { + return readableRecordsFollowFrom(data, offset, 0, false) +} + +// readableRecordsFollowFrom reports whether data holds readable records from +// offset. When haveSeed is true, seedCRC is the CRC of the record before offset +// and the first record's CRC is verified against it. A failed link does not end +// the search: the current record becomes the new seed, so a later verified link +// still counts. +// +// Records are interpreted as LogRecordReader.Read interprets them: an anchor +// (CRC_ANCHOR) reseeds the chain from its header CRC instead of being verified, +// since its header CRC is the CRC of the previous record, not of its payload. +func readableRecordsFollowFrom(data []byte, offset int64, seedCRC uint32, haveSeed bool) bool { + prevCRC := seedCRC + + for pos := offset; ; { + record, ok := parseRecordAt(data, pos) + if !ok { + return false + } + + switch record.recordType { + case protos.LogRecord_CRC_ANCHOR: + prevCRC = record.crc + haveSeed = true + case protos.LogRecord_ENTRY, protos.LogRecord_CONTROL: + if haveSeed && verifyCRC(prevCRC, record.crc, record.payload) { + return true + } + + prevCRC = record.crc + haveSeed = true + } + + pos = record.next + } +} diff --git a/pkg/wal/util_test.go b/pkg/wal/util_test.go index 86f07b11..4537b172 100644 --- a/pkg/wal/util_test.go +++ b/pkg/wal/util_test.go @@ -6,7 +6,9 @@ package wal import ( + "encoding/binary" "fmt" + "hash/crc32" "io" "os" "path/filepath" @@ -16,6 +18,7 @@ import ( "github.com/hyperledger/SmartBFT/smartbftprotos" "github.com/stretchr/testify/assert" "go.uber.org/zap" + "google.golang.org/protobuf/proto" ) func TestWALUtil(t *testing.T) { @@ -185,6 +188,53 @@ func TestWALUtil(t *testing.T) { assert.NoError(t, err) }) + t.Run("scan-repair-corrupt-record-in-the-middle", func(t *testing.T) { + testDir, err := os.MkdirTemp("", "unittest") + assert.NoErrorf(t, err, "generate temporary test dir") + defer os.RemoveAll(testDir) + + wal, err := Create(logger, testDir, &Options{FileSizeBytes: 1024 * 1024}) + assert.NoError(t, err) + for range 20 { + assert.NoError(t, wal.Append(make([]byte, 1024), false)) + } + assert.NoError(t, wal.Close()) + + names, err := dirReadWalNames(testDir) + assert.NoError(t, err) + assert.Len(t, names, 1) + lastFile := filepath.Join(testDir, names[0]) + + // flip a CRC bit of the first record, leaving the records written after it intact + r, err := NewLogRecordReader(logger, lastFile) + assert.NoError(t, err) + _, err = r.Read() // the CRC-anchor + assert.NoError(t, err) + offset, err := r.logFile.Seek(0, io.SeekCurrent) + assert.NoError(t, err) + assert.NoError(t, r.Close()) + + f, err := os.OpenFile(lastFile, os.O_RDWR, walFilePermPrivateRW) + assert.NoError(t, err) + crcByte := make([]byte, 1) + _, err = f.ReadAt(crcByte, offset+5) // upper 32 bits of the header hold the CRC + assert.NoError(t, err) + crcByte[0] ^= 0x01 + _, err = f.WriteAt(crcByte, offset+5) + assert.NoError(t, err) + assert.NoError(t, f.Close()) + + sizeBefore, err := os.Stat(lastFile) + assert.NoError(t, err) + + // the node cannot be repaired: the records after the corrupt one are lost otherwise + assert.ErrorIs(t, Repair(logger, testDir), ErrNotRepairable) + + sizeAfter, err := os.Stat(lastFile) + assert.NoError(t, err) + assert.Equal(t, sizeBefore.Size(), sizeAfter.Size(), "the file must be left untouched") + }) + t.Run("scan-repair-tail", func(t *testing.T) { testDir, err := os.MkdirTemp("", "unittest") assert.NoErrorf(t, err, "generate temporary test dir") @@ -219,6 +269,45 @@ func TestWALUtil(t *testing.T) { assert.NoError(t, err) }) + t.Run("repair makes a private copy of the last file", func(t *testing.T) { + testDir, err := os.MkdirTemp("", "unittest") + assert.NoErrorf(t, err, "generate temporary test dir") + defer os.RemoveAll(testDir) + + make8LogFiles(t, logger, testDir) + + lastFile := filepath.Join(testDir, fmt.Sprintf(walFileTemplate, 8)) + // pre-existing world readable copy must be tightened + lastFileCopy := lastFile + ".copy" + assert.NoError(t, os.WriteFile(lastFileCopy, []byte("junk"), 0o644)) + + // corrupt the last file, so that Repair has something to do + f, err := os.OpenFile(lastFile, os.O_RDWR, walFilePermPrivateRW) + assert.NoError(t, err) + _, err = f.Seek(0, io.SeekStart) + assert.NoError(t, err) + _, err = f.Write(make([]byte, 64)) + assert.NoError(t, err) + assert.NoError(t, f.Close()) + + damaged, err := os.ReadFile(lastFile) + assert.NoError(t, err) + + assert.NoError(t, Repair(logger, testDir)) + + fi, err := os.Stat(lastFileCopy) + assert.NoError(t, err) + assert.Equal(t, os.FileMode(walFilePermPrivateRW), fi.Mode().Perm()) + + copied, err := os.ReadFile(lastFileCopy) + assert.NoError(t, err) + assert.Equal(t, damaged, copied) + + names, err := dirReadWalNames(testDir) + assert.NoError(t, err) + assert.NoError(t, scanVerifyFiles(logger, testDir, names)) + }) + t.Run("scan-repair-bad-anchor", func(t *testing.T) { testDir, err := os.MkdirTemp("", "unittest") assert.NoErrorf(t, err, "generate temporary test dir") @@ -255,6 +344,182 @@ func TestWALUtil(t *testing.T) { err = scanVerifyFiles(logger, testDir, names) assert.NoError(t, err) }) + + t.Run("readable records follow - crc-anchor in the chain", func(t *testing.T) { + // encodeRecord writes a record exactly as the WAL writes it: an 8 byte + // header of length and CRC, then the payload padded to a multiple of 8. + encodeRecord := func(record *smartbftprotos.LogRecord, crc uint32) []byte { + buff, err := proto.Marshal(record) + assert.NoError(t, err) + + length := len(buff) + _, pad := getPadBytes(length) + header := make([]byte, recordHeaderSize) + binary.LittleEndian.PutUint64(header, uint64(length)|uint64(crc)<<32) + + return append(append(header, buff...), pad...) + } + + payloadOf := func(record *smartbftprotos.LogRecord) []byte { + buff, err := proto.Marshal(record) + assert.NoError(t, err) + _, pad := getPadBytes(len(buff)) + + return append(buff, pad...) + } + + follows := func(data []byte, expected bool) { + assert.Equal(t, expected, readableRecordsFollow(data, 0)) + } + + first := &smartbftprotos.LogRecord{Type: smartbftprotos.LogRecord_ENTRY, Data: []byte("first")} + anchor := &smartbftprotos.LogRecord{Type: smartbftprotos.LogRecord_CRC_ANCHOR} + second := &smartbftprotos.LogRecord{Type: smartbftprotos.LogRecord_ENTRY, Data: []byte("second")} + + // The first record sits at the probe offset: its CRC depends on the + // corrupt record and is never checked, it only seeds the chain. An + // anchor follows; its header CRC is the CRC of the previous record, not + // of its payload, so it reseeds the chain, and the record after it must + // chain from that seed - exactly as LogRecordReader.Read accepts it. + anchorCRC := uint32(0xCAFEBABE) + secondCRC := crc32.Update(anchorCRC, crcTable, payloadOf(second)) + + chain := encodeRecord(first, 0xDEADBEEF) + chain = append(chain, encodeRecord(anchor, anchorCRC)...) + chain = append(chain, encodeRecord(second, secondCRC)...) + follows(chain, true) + assert.True(t, readableLinkPresent(chain, 0)) + + // a lone anchor proves nothing: no CRC link was verified + lone := encodeRecord(first, 0xDEADBEEF) + lone = append(lone, encodeRecord(anchor, anchorCRC)...) + follows(lone, false) + assert.False(t, readableLinkPresent(lone, 0)) + + // a record chained to the anchor with a broken CRC does not verify + broken := encodeRecord(first, 0xDEADBEEF) + broken = append(broken, encodeRecord(anchor, anchorCRC)...) + broken = append(broken, encodeRecord(second, secondCRC^0xFFFFFFFF)...) + follows(broken, false) + assert.False(t, readableLinkPresent(broken, 0)) + }) +} + +// TestRepairMidCorruptPlusTornTail is the reproduction from the review of PR #694: +// a corrupt record in the middle of the file, with only the tail torn after it, +// must not be repaired - truncating would drop records written before the crash. +func TestRepairMidCorruptPlusTornTail(t *testing.T) { + basicLog, err := zap.NewDevelopment() + assert.NoError(t, err) + + logger := basicLog.Sugar() + dir := t.TempDir() + + w, err := Create(logger, dir, &Options{FileSizeBytes: 1024 * 1024}) + assert.NoError(t, err) + for range 20 { + assert.NoError(t, w.Append(make([]byte, 1024), false)) + } + assert.NoError(t, w.Close()) + + names, err := dirReadWalNames(dir) + assert.NoError(t, err) + last := filepath.Join(dir, names[0]) + + r, err := NewLogRecordReader(logger, last) + assert.NoError(t, err) + _, err = r.Read() // R0 + assert.NoError(t, err) + off, err := r.logFile.Seek(0, io.SeekCurrent) + assert.NoError(t, err) + assert.NoError(t, r.Close()) + + f, err := os.OpenFile(last, os.O_RDWR, walFilePermPrivateRW) + assert.NoError(t, err) + b := make([]byte, 1) + _, err = f.ReadAt(b, off+5) + assert.NoError(t, err) + b[0] ^= 1 + _, err = f.WriteAt(b, off+5) // corrupt R1's CRC + assert.NoError(t, err) + fi, err := f.Stat() + assert.NoError(t, err) + assert.NoError(t, f.Truncate(fi.Size()-100)) // tear R19 + fi, err = f.Stat() + assert.NoError(t, err) + damagedSize := fi.Size() + assert.NoError(t, f.Close()) + + assert.ErrorIs(t, Repair(logger, dir), ErrNotRepairable) + + // the corrupt record is in the middle of the file: nothing may be truncated + fi, err = os.Stat(last) + assert.NoError(t, err) + assert.Equal(t, damagedSize, fi.Size()) +} + +// TestRepairMidCorruptOneFollowingRecord covers a corrupt record followed by a +// single readable record. Truncating at the corrupt record would drop it. +func TestRepairMidCorruptOneFollowingRecord(t *testing.T) { + basicLog, err := zap.NewDevelopment() + assert.NoError(t, err) + + logger := basicLog.Sugar() + + for _, testCase := range []struct { + name string + offset int64 + }{ + {name: "corrupt payload", offset: int64(recordHeaderSize)}, + {name: "corrupt header CRC", offset: 5}, + {name: "corrupt header length", offset: 2}, + } { + t.Run(testCase.name, func(t *testing.T) { + dir := t.TempDir() + + w, err := Create(logger, dir, &Options{FileSizeBytes: 1024 * 1024}) + assert.NoError(t, err) + for range 5 { + assert.NoError(t, w.Append(make([]byte, 1024), false)) + } + assert.NoError(t, w.Close()) + + names, err := dirReadWalNames(dir) + assert.NoError(t, err) + last := filepath.Join(dir, names[0]) + + // R3 is the corrupt record; only R4 follows it. + r, err := NewLogRecordReader(logger, last) + assert.NoError(t, err) + for range 3 { + _, err = r.Read() + assert.NoError(t, err) + } + recordStart, err := r.logFile.Seek(0, io.SeekCurrent) + assert.NoError(t, err) + assert.NoError(t, r.Close()) + + f, err := os.OpenFile(last, os.O_RDWR, walFilePermPrivateRW) + assert.NoError(t, err) + b := make([]byte, 1) + _, err = f.ReadAt(b, recordStart+testCase.offset) + assert.NoError(t, err) + b[0] ^= 1 + _, err = f.WriteAt(b, recordStart+testCase.offset) + assert.NoError(t, err) + assert.NoError(t, f.Close()) + + fi, err := os.Stat(last) + assert.NoError(t, err) + sizeBefore := fi.Size() + + assert.ErrorIs(t, Repair(logger, dir), ErrNotRepairable) + + fi, err = os.Stat(last) + assert.NoError(t, err) + assert.Equal(t, sizeBefore, fi.Size(), "the file must be left untouched") + }) + } } func arrayToSet(array []string) map[string]bool { diff --git a/pkg/wal/writeaheadlog.go b/pkg/wal/writeaheadlog.go index fdb3f344..7e4ee586 100644 --- a/pkg/wal/writeaheadlog.go +++ b/pkg/wal/writeaheadlog.go @@ -40,13 +40,17 @@ const ( ) var ( - ErrCRC = errors.New("wal: crc verification failed") - ErrWALUnmarshalPayload = errors.New("wal: failed to unmarshal payload") - ErrWriteOnly = errors.New("wal: in WRITE mode") - ErrReadOnly = errors.New("wal: in READ mode") + ErrCRC = errors.New("wal: crc verification failed") + ErrWALUnmarshalPayload = errors.New("wal: failed to unmarshal payload") + ErrUnexpectedRecordType = errors.New("wal: unexpected LogRecord_Type") + ErrRecordTooBig = errors.New("wal: record too big") + ErrWriteOnly = errors.New("wal: in WRITE mode") + ErrReadOnly = errors.New("wal: in READ mode") ErrWALAlreadyExists = errors.New("wal: is already exists") + ErrNotRepairable = errors.New("wal: corruption is not in the last record, not repairable") + crcTable = crc32.MakeTable(crc32.Castagnoli) ) @@ -278,12 +282,14 @@ func Open(logger api.Logger, dirPath string, options *Options) (*WriteAheadLogFi // Repair tries to repair the last file of a WAL, in case an Open() comes back with a io.ErrUnexpectedEOF, // which indicates the possibility to fix the WAL. // -// After a crash, the last log file in the WAL may be left in a state where an attempt tp reopen will result in -// the last Read() returning an error. This is because of several reasons: -// - we use pre-allocated / recycled files, the files have a "garbage" tail -// - the last write may have been torn +// After a crash, the last log file in the WAL may be left in a state where an attempt to reopen will result in +// the last Read() returning an error. This is because of two reasons: +// - the last write may have been torn, which is the only expected source of a corrupt tail // - the failure might have occurred when the log file was being prepared (no anchor) // +// Files are created with O_CREATE and truncated to their used size on close, so a file holds no stale bytes +// beyond the last record, and a torn last write - not a "garbage" tail - is what repair has to deal with. +// // The Repair() tries to repair the last file of the wal by truncating after the last good record. // Before doing so it copies the bad file to a side location for later analysis by operators. // @@ -443,7 +449,7 @@ func (w *WriteAheadLogFile) append(record *protos.LogRecord) error { recordLength := len(payloadBuff) if (uint64(recordLength) & recordCRCMask) != 0 { - return fmt.Errorf("wal: record too big, length does not fit in uint32: %d", recordLength) + return fmt.Errorf("%w, length does not fit in uint32: %d", ErrRecordTooBig, recordLength) } padSize, padBytes := getPadBytes(recordLength) @@ -574,7 +580,9 @@ FileLoop: if index == lastIndex && (errors.Is(readErr, io.ErrUnexpectedEOF) || errors.Is(readErr, ErrCRC) || - errors.Is(readErr, ErrWALUnmarshalPayload)) { + errors.Is(readErr, ErrWALUnmarshalPayload) || + errors.Is(readErr, ErrUnexpectedRecordType) || + errors.Is(readErr, ErrRecordTooBig)) { w.logger.Warnf( "Received an error in the last file, this can possibly be repaired; file: %s; error: %s", r.fileName, @@ -775,7 +783,25 @@ func InitializeAndReadAll( writeAheadLog, err = Open(logger, walDir, options) if err != nil { - return nil, nil, fmt.Errorf("Cannot open Write-Ahead-Log: %w", err) + // A torn CRC-Anchor in the last file surfaces here, on Open, not on + // ReadAll. It is the same repairable condition - the failure happened + // while the last file was being prepared - so repair it before giving up. + if !errors.Is(err, io.ErrUnexpectedEOF) { + return nil, nil, fmt.Errorf("Cannot open Write-Ahead-Log: %w", err) + } + + logger.Infof("Received io.ErrUnexpectedEOF on open, trying to repair Write-Ahead-Log at dir: %s", walDir) + + if err = Repair(logger, walDir); err != nil { + return nil, nil, fmt.Errorf("Cannot repair Write-Ahead-Log: %w", err) + } + + logger.Infof("Opening Write-Ahead-Log, after repair") + + writeAheadLog, err = Open(logger, walDir, options) + if err != nil { + return nil, nil, fmt.Errorf("Cannot open Write-Ahead-Log, after repair: %w", err) + } } initialState, err = writeAheadLog.ReadAll() diff --git a/pkg/wal/writeaheadlog_test.go b/pkg/wal/writeaheadlog_test.go index c3aaa5ab..766af2a0 100644 --- a/pkg/wal/writeaheadlog_test.go +++ b/pkg/wal/writeaheadlog_test.go @@ -7,6 +7,7 @@ package wal import ( "bytes" + "encoding/binary" "fmt" "io" "os" @@ -17,6 +18,7 @@ import ( "github.com/hyperledger/SmartBFT/smartbftprotos" "github.com/stretchr/testify/assert" "go.uber.org/zap" + "google.golang.org/protobuf/proto" ) func TestWriteAheadLogFile_Create(t *testing.T) { @@ -676,6 +678,60 @@ func TestWriteAheadLogFile_Repair(t *testing.T) { assertTestRepair(t, logger, dirPath, NumRec-1) }) + + t.Run("unknown record type at the tail", func(t *testing.T) { + dirPath := filepath.Join(testDir, "unknown-type") + err := os.MkdirAll(dirPath, walDirPermPrivateRWX) + assert.NoError(t, err) + + wal, err := Create(logger, dirPath, &Options{FileSizeBytes: 10 * 1024, BufferSizeBytes: 2048}) + assert.NoError(t, err) + assert.NotNil(t, wal) + if wal == nil { + return + } + + const NumBytes = 1024 + const NumRec = 5 + data1 := make([]byte, NumBytes) + for m := range NumRec { + for n := range NumBytes { + data1[n] = byte(m) + } + err = wal.Append(data1, false) + assert.NoError(t, err) + } + + err = wal.Close() + assert.NoError(t, err) + + // append a well-formed record whose type the reader does not know + names, err := dirReadWalNames(dirPath) + assert.NoError(t, err) + lastFile := filepath.Join(dirPath, names[len(names)-1]) + + rec, err := proto.Marshal(&smartbftprotos.LogRecord{Type: smartbftprotos.LogRecord_Type(7), Data: []byte{1, 2, 3}}) + assert.NoError(t, err) + _, pad := getPadBytes(len(rec)) + header := make([]byte, recordHeaderSize) + binary.LittleEndian.PutUint64(header, uint64(len(rec))) + + f, err := os.OpenFile(lastFile, os.O_RDWR, walFilePermPrivateRW) + assert.NoError(t, err) + _, err = f.Seek(0, io.SeekEnd) + assert.NoError(t, err) + _, err = f.Write(append(append(header, rec...), pad...)) + 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) + }) } func TestWriteAheadLogFile_InitializeAndReadAll(t *testing.T) { @@ -780,6 +836,57 @@ func TestWriteAheadLogFile_InitializeAndReadAll(t *testing.T) { assert.NoError(t, err) } +func TestWriteAheadLogFile_InitializeAndReadAll_TornLastAnchor(t *testing.T) { + testDir, err := os.MkdirTemp("", t.Name()) + assert.NoErrorf(t, err, "generate temporary test dir") + + defer os.RemoveAll(testDir) + + basicLog, err := zap.NewDevelopment() + assert.NoError(t, err) + + logger := basicLog.Sugar() + + options := &Options{FileSizeBytes: 10 * 1024, BufferSizeBytes: 2048} + + wal, _, err := InitializeAndReadAll(logger, testDir, options) + assert.NoError(t, err) + + const NumRec = 5 + for m := range NumRec { + err = wal.Append([]byte{byte(m), 1, 2, 3}, false) + assert.NoError(t, err) + } + assert.NoError(t, wal.Close()) + + // Simulate a crash while the next file was being prepared: the new last file + // exists but its CRC-Anchor was never written. + names, err := dirReadWalNames(testDir) + assert.NoError(t, err) + lastIndex, err := parseWalFileName(names[len(names)-1]) + assert.NoError(t, err) + tornFile := filepath.Join(testDir, fmt.Sprintf(walFileTemplate, lastIndex+1)) + assert.NoError(t, os.WriteFile(tornFile, nil, walFilePermPrivateRW)) + + // Open surfaces the repairable error, instead of reading the state. + _, err = Open(logger, testDir, options) + assert.ErrorIs(t, err, io.ErrUnexpectedEOF) + + wal, entries, err := InitializeAndReadAll(logger, testDir, options) + assert.NoError(t, err) + assert.NotNil(t, wal) + assert.Equal(t, NumRec, len(entries)) + for i, e := range entries { + assert.Equal(t, byte(i), e[0]) + } + assert.NoError(t, wal.Close()) + + // the torn file was removed by the repair (a copy is kept for operators), + // and ReadAll then created a fresh file for writing. + _, err = os.Stat(tornFile + ".copy") + assert.NoError(t, err, "the repair must keep a copy of the torn file") +} + func assertTestRepair(t *testing.T, logger api.Logger, dirPath string, numRec int) { logger.Infof(">>> Open #1") wal, err := Open(logger, dirPath, &Options{FileSizeBytes: 10 * 1024, BufferSizeBytes: 2048})