diff --git a/core/rawdb/freezer_batch.go b/core/rawdb/freezer_batch.go index 76c46a070b..8e3c18f9a9 100644 --- a/core/rawdb/freezer_batch.go +++ b/core/rawdb/freezer_batch.go @@ -180,8 +180,8 @@ func (batch *freezerTableBatch) maybeCommit() error { return nil } -// commit writes the batched items to the backing freezerTable. Note both index -// and data file are not fsync'd after the file write, the data could be lost +// commit writes the batched items to the backing freezerTable. Note neither index +// nor data file are fsync'd after the file write, the recent write can be lost // after the power failure. func (batch *freezerTableBatch) commit() error { _, err := batch.t.head.Write(batch.dataBuffer) diff --git a/core/rawdb/freezer_table.go b/core/rawdb/freezer_table.go index 84bfa0727a..e09bb60e9b 100644 --- a/core/rawdb/freezer_table.go +++ b/core/rawdb/freezer_table.go @@ -17,6 +17,7 @@ package rawdb import ( + "bufio" "bytes" "encoding/binary" "errors" @@ -377,9 +378,9 @@ func (t *freezerTable) repair() error { // be in order. This function identifies any corrupted entries and truncates items // occurring after the corruption point. // -// The corruption can be made because of the power failure. As in the Linux kernel, -// the file metadata update and data update are not necessarily performed at the -// same time. Typically, the metadata will be flushed/journaled ahead of the file +// corruption can occur because of the power failure. In the Linux kernel, the +// file metadata update and data update are not necessarily performed at the +// same time. Typically, the metadata will be flushed/journalled ahead of the file // data. Therefore, we make the pessimistic assumption that the file is first // extended with invalid "garbage" data (normally zero bytes) and that afterwards // the correct data replaces the garbage. As all the items in index file are @@ -399,36 +400,29 @@ func (t *freezerTable) repairIndex() error { } size := stat.Size() - var ( - start = time.Now() - batchSize = int64(indexEntrySize * 1024 * 1024) - buffer = make([]byte, batchSize) // pre-allocate for batch reading - prev indexEntry - head indexEntry - readOffset int64 - consumed int64 + // Move the read cursor to the beginning of the file. + _, err = t.index.Seek(0, 0) + if err != nil { + return err + } + fr := bufio.NewReader(t.index) - read = func(offset int64) (indexEntry, error) { - if offset+indexEntrySize > size { - return indexEntry{}, fmt.Errorf("slice bounds out of range, offset: %d, size: %d", offset, size) + var ( + start = time.Now() + buff = make([]byte, indexEntrySize) + prev indexEntry + head indexEntry + + read = func() (indexEntry, error) { + n, err := fr.Read(buff) + if err != nil { + return indexEntry{}, err } - if offset >= readOffset { - n, err := t.index.ReadAt(buffer, readOffset) - if err != nil && !errors.Is(err, io.EOF) { - return indexEntry{}, err - } - expect := batchSize - if size-readOffset < batchSize { - expect = size - readOffset - } - if expect != int64(n) { - return indexEntry{}, fmt.Errorf("failed to read from index, want: %d, got: %d", expect, n) - } - consumed = readOffset - readOffset += int64(n) + if n != indexEntrySize { + return indexEntry{}, fmt.Errorf("failed to read from index, n: %d", n) } var entry indexEntry - entry.unmarshalBinary(buffer[offset-consumed : offset-consumed+indexEntrySize]) + entry.unmarshalBinary(buff) return entry, nil } truncate = func(offset int64) error { @@ -443,7 +437,7 @@ func (t *freezerTable) repairIndex() error { } ) for offset := int64(0); offset < size; offset += indexEntrySize { - entry, err := read(offset) + entry, err := read() if err != nil { return err } @@ -451,9 +445,11 @@ func (t *freezerTable) repairIndex() error { head = entry continue } - // ensure the first non-head index denotes to the earliest file + // Ensure that the first non-head index refers to the earliest file, + // or the next file if the earliest file is not sufficient to + // place the first item. if offset == indexEntrySize { - if entry.filenum != head.filenum { + if entry.filenum != head.filenum && entry.filenum != head.filenum+1 { return truncate(offset) } prev = entry @@ -465,6 +461,13 @@ func (t *freezerTable) repairIndex() error { } prev = entry } + // Move the read cursor to the end of the file. While theoretically, the + // cursor should reach the end by reading all the items in the file, perform + // the seek operation anyway as a precaution. + _, err = t.index.Seek(0, io.SeekEnd) + if err != nil { + return err + } log.Debug("Verified index file", "items", size/indexEntrySize, "elapsed", common.PrettyDuration(time.Since(start))) return nil } diff --git a/core/rawdb/freezer_table_test.go b/core/rawdb/freezer_table_test.go index 91b4943e59..fd6e3cf199 100644 --- a/core/rawdb/freezer_table_test.go +++ b/core/rawdb/freezer_table_test.go @@ -1367,3 +1367,69 @@ func TestRandom(t *testing.T) { t.Fatal(err) } } + +func TestIndexValidation(t *testing.T) { + const ( + items = 30 + dataSize = 10 + ) + garbage := indexEntry{ + filenum: 100, + offset: 200, + } + var cases = []struct { + offset int64 + data []byte + expItems int + }{ + // extend index file with zero bytes at the end + { + offset: (items + 1) * indexEntrySize, + data: make([]byte, indexEntrySize), + expItems: 30, + }, + // write garbage in the first non-head item + { + offset: indexEntrySize, + data: garbage.append(nil), + expItems: 0, + }, + // write garbage in the first non-head item + { + offset: (items/2 + 1) * indexEntrySize, + data: garbage.append(nil), + expItems: items / 2, + }, + } + for _, c := range cases { + fn := fmt.Sprintf("t-%d", rand.Uint64()) + f, err := newTable(os.TempDir(), fn, metrics.NewMeter(), metrics.NewMeter(), metrics.NewGauge(), 100, true, false) + if err != nil { + t.Fatal(err) + } + writeChunks(t, f, items, dataSize) + + // write corrupted data + f.index.WriteAt(c.data, c.offset) + f.Close() + + // reopen the table, corruption should be truncated + f, err = newTable(os.TempDir(), fn, metrics.NewMeter(), metrics.NewMeter(), metrics.NewGauge(), 100, true, false) + if err != nil { + t.Fatal(err) + } + for i := 0; i < c.expItems; i++ { + exp := getChunk(10, i) + got, err := f.Retrieve(uint64(i)) + if err != nil { + t.Fatalf("Failed to read from table, %v", err) + } + if !bytes.Equal(exp, got) { + t.Fatalf("Unexpected item data, want: %v, got: %v", exp, got) + } + } + if f.items.Load() != uint64(c.expItems) { + t.Fatalf("Unexpected item number, want: %d, got: %d", c.expItems, f.items.Load()) + } + } +} diff --git a/triedb/pathdb/nodebuffer.go b/triedb/pathdb/nodebuffer.go index 1f029cba86..a4788ff9ba 100644 --- a/triedb/pathdb/nodebuffer.go +++ b/triedb/pathdb/nodebuffer.go @@ -229,8 +229,10 @@ func (b *nodebuffer) flush(db ethdb.KeyValueStore, freezer ethdb.AncientWriter, ) // Explicitly sync the state freezer, ensuring that all written // data is transferred to disk before updating the key-value store. - if err := freezer.Sync(); err != nil { - return err + if freezer != nil { + if err := freezer.Sync(); err != nil { + return err + } } nodes := writeNodes(batch, b.nodes, clean) rawdb.WritePersistentStateID(batch, id)