From 32ad22c7c25600560120b3ba6c8c3138f4e92a12 Mon Sep 17 00:00:00 2001 From: Fedor Partanskiy Date: Thu, 8 Oct 2026 17:18:10 +0300 Subject: [PATCH] A corrupt or forged record header could claim 0xFFFFFFFF bytes and drive a ~4GB allocation on Open. Reject such records with ErrRecordTooBig, both on read and on append. Repair copies are now 0600, and an existing .copy gets chmod'ed, since WriteFile does not touch its mode. **Repair no longer performs lossy repairs.** `Repair()` now truncates the last file only when the corruption is at the tail (the last record is torn). When a corrupt record is followed by readable records, the repair is refused with `ErrNotRepairable` and the file is left untouched (a `.copy` is still saved). Previously `Repair()` silently truncated the file at the first bad record, which could drop records that were already synced and possibly committed. This changes whether a node can restart on its own: a mid-file corruption is no longer repaired automatically. **Torn-tail detection.** `isTornTail` now decides "torn" using the same rules as `LogRecordReader.Read`: it accepts a chain as soon as one CRC link is verified (instead of requiring a clean chain to EOF), probes the exact offset of the next record first when the bad header is intact, and reseeds the chain on a `CRC_ANCHOR`. This makes the repair correct for the usual crash state - a corrupt record in the middle plus a torn tail. **Record-size bound.** `readHeader` bounds a record against the bytes left in the file (`fileSizeBytes - pos`), so a corrupt header near the end of a 64MB file is rejected before the payload allocation. A header that cannot fit the file at all is reported as `ErrRecordTooBig`; a record that only outgrows the remaining bytes is still `io.ErrUnexpectedEOF` (a torn tail, the repairable error). The redundant check in `readPayload` is removed. Signed-off-by: Fedor Partanskiy --- pkg/wal/reader.go | 37 +++- pkg/wal/reader_test.go | 162 +++++++++++++++++ pkg/wal/util.go | 326 +++++++++++++++++++++++++++++++++- pkg/wal/util_test.go | 269 +++++++++++++++++++++++++++- pkg/wal/writeaheadlog.go | 48 +++-- pkg/wal/writeaheadlog_test.go | 146 +++++++++++++++ 6 files changed, 968 insertions(+), 20 deletions(-) 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 5572d60d..c779bd0e 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" @@ -267,6 +268,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..e9a8887c 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 @@ -142,6 +145,14 @@ func getPadSize(recordLength int) int { return (8 - recordLength%8) % 8 } +// crcAnchorRecordSize returns the on-disk size of the CRC-Anchor record that +// starts every WAL file. +func crcAnchorRecordSize() int64 { + payloadLength := proto.Size(&protos.LogRecord{Type: protos.LogRecord_CRC_ANCHOR}) + + return int64(recordHeaderSize + payloadLength + getPadSize(payloadLength)) +} + func getPadBytes(recordLength int) (int, []byte) { i := getPadSize(recordLength) @@ -157,13 +168,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() + + output, err := os.OpenFile(target, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, walFilePermPrivateRW) if err != nil { return err } - return os.WriteFile(target, input, 0o644) + 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 { @@ -234,7 +272,12 @@ 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. +// read, the file is deleted only when it holds nothing but that anchor; otherwise the repair is +// refused with ErrNotRepairable. +// +// 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) @@ -242,6 +285,23 @@ func scanRepairFile(logger api.Logger, lastFile string) error { if err != nil { logger.Warnf("Write-Ahead-Log could not open the last file, due to error: %s", err) + // A torn CRC-Anchor is only possible while the file is being prepared, + // before the first Append, so the file holds nothing but that anchor. If + // it is larger, records were synced to it and deleting it would lose them. + fileInfo, statErr := os.Stat(lastFile) + if statErr != nil { + return statErr + } + + if fileInfo.Size() > crcAnchorRecordSize() { + logger.Errorf( + "Write-Ahead-Log will not repair file: %s; its CRC-Anchor is corrupt but the file holds records", + lastFile, + ) + + return fmt.Errorf("%w: corrupt CRC-Anchor of file: %s", ErrNotRepairable, lastFile) + } + if err = os.Remove(lastFile); err != nil { return err } @@ -266,6 +326,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 +339,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 +366,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 +389,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 634ddbf0..81aec779 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) { @@ -177,6 +180,51 @@ func TestWALUtil(t *testing.T) { assert.NoError(t, err) }) + t.Run("scan-repair-corrupt-record-in-the-middle", func(t *testing.T) { + testDir := t.TempDir() + + 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 := t.TempDir() @@ -209,6 +257,43 @@ 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 := t.TempDir() + + 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.SeekEnd) + 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, 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 := t.TempDir() @@ -234,15 +319,189 @@ func TestWALUtil(t *testing.T) { err = scanVerifyFiles(logger, testDir, names) assert.Contains(t, err.Error(), "failed reading CRC-Anchor from log file:") - // repair is good + // The file holds records, so deleting it would lose them. err = scanRepairFile(logger, lastFile) - assert.NoError(t, err) + assert.ErrorIs(t, err, ErrNotRepairable) names, err = dirReadWalNames(testDir) assert.NoError(t, err) - assert.Equal(t, len(names), 7) - err = scanVerifyFiles(logger, testDir, names) - assert.NoError(t, err) + assert.Equal(t, 8, len(names)) }) + + 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 a621316e..bcf0596c 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) { @@ -658,6 +660,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) { @@ -759,6 +815,96 @@ func TestWriteAheadLogFile_InitializeAndReadAll(t *testing.T) { assert.NoError(t, err) } +func TestWriteAheadLogFile_InitializeAndReadAll_TornLastAnchor(t *testing.T) { + testDir := t.TempDir() + + 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 TestRepairCorruptAnchorOfNonEmptyLastFile(t *testing.T) { + basicLog, err := zap.NewDevelopment() + assert.NoError(t, err) + + logger := basicLog.Sugar() + dir := t.TempDir() + options := &Options{FileSizeBytes: 64 * 1024} + + wal, err := Create(logger, dir, options) + assert.NoError(t, err) + for range 100 { + assert.NoError(t, wal.Append(make([]byte, 1024), false)) + } + assert.NoError(t, wal.Close()) + + names, err := dirReadWalNames(dir) + assert.NoError(t, err) + assert.Len(t, names, 2) + lastFile := filepath.Join(dir, names[len(names)-1]) + + // Corrupt the CRC-Anchor of a file that holds records. + f, err := os.OpenFile(lastFile, os.O_RDWR, walFilePermPrivateRW) + assert.NoError(t, err) + b := make([]byte, 1) + _, err = f.ReadAt(b, int64(recordHeaderSize)) + assert.NoError(t, err) + b[0] ^= 0xFF + _, err = f.WriteAt(b, int64(recordHeaderSize)) + assert.NoError(t, err) + assert.NoError(t, f.Close()) + + sizeBefore, err := os.Stat(lastFile) + assert.NoError(t, err) + + _, _, err = InitializeAndReadAll(logger, dir, options) + assert.ErrorIs(t, err, ErrNotRepairable) + + fi, err := os.Stat(lastFile) + assert.NoError(t, err) + assert.Equal(t, sizeBefore.Size(), fi.Size(), "the file must be left untouched") +} + 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})