diff --git a/cmd/geth/chaincmd.go b/cmd/geth/chaincmd.go index 98ed348d8c..0702ad15a5 100644 --- a/cmd/geth/chaincmd.go +++ b/cmd/geth/chaincmd.go @@ -538,10 +538,9 @@ func importHistory(ctx *cli.Context) error { default: return fmt.Errorf("unknown --era.format %q (expected 'era1' or 'erae')", format) } - if err := utils.ImportHistory(chain, dir, network, from); err != nil { + if err := utils.ImportHistory(chain, db, dir, network, from); err != nil { return err } - fmt.Printf("Import done in %v\n", time.Since(start)) return nil } diff --git a/cmd/utils/cmd.go b/cmd/utils/cmd.go index e490f613b3..0771e4218e 100644 --- a/cmd/utils/cmd.go +++ b/cmd/utils/cmd.go @@ -249,13 +249,10 @@ func readList(filename string) ([]string, error) { return strings.Split(string(b), "\n"), nil } -// ImportHistory imports Era1 files containing historical block information, +// ImportHistory imports Era1/Erae files containing historical block information, // starting from genesis. The assumption is held that the provided chain -// segment in Era1 file should all be canonical and verified. -func ImportHistory(chain *core.BlockChain, dir string, network string, from func(f era.ReadAtSeekCloser) (era.Era, error)) error { - if chain.CurrentSnapBlock().Number.BitLen() != 0 { - return errors.New("history import only supported when starting from genesis") - } +// segment in Era1/Erae file should all be canonical and verified. +func ImportHistory(chain *core.BlockChain, db ethdb.Database, dir string, network string, from func(f era.ReadAtSeekCloser) (era.Era, error)) error { entries, err := era.ReadDir(dir, network) if err != nil { return fmt.Errorf("error reading %s: %w", dir, err) @@ -269,6 +266,13 @@ func ImportHistory(chain *core.BlockChain, dir string, network string, from func len(checksums), len(entries)) } + // Determine resume point from last successfully imported block. + var resumeBlock uint64 + if tail := rawdb.ReadEraImportTail(db); tail != nil { + resumeBlock = *tail + log.Info("Resuming era import", "lastBlock", resumeBlock) + } + var ( start = time.Now() reported = time.Now() @@ -281,13 +285,33 @@ func ImportHistory(chain *core.BlockChain, dir string, network string, from func err := func() error { path := filepath.Join(dir, file) - // Validate against checksum file in directory. f, err := os.Open(path) if err != nil { return fmt.Errorf("open %s: %w", path, err) } defer f.Close() + // Peek at era block range to check if we can skip entirely. + e, err := from(f) + if err != nil { + f.Close() + return fmt.Errorf("error opening era: %w", err) + } + eraStart := e.Start() + eraEnd := eraStart + e.Count() - 1 + e.Close() + + // Skip era files fully behind resume point. + if resumeBlock > 0 && eraEnd <= resumeBlock { + log.Debug("Skipping already imported Era file", "file", file, "eraEnd", eraEnd, "resumeBlock", resumeBlock) + return nil + } + + // reopen for checksum + import. + f, err = os.Open(path) + if err != nil { + return fmt.Errorf("open %s: %w", path, err) + } if _, err := io.Copy(h, f); err != nil { return fmt.Errorf("checksum %s: %w", path, err) } @@ -297,8 +321,12 @@ func ImportHistory(chain *core.BlockChain, dir string, network string, from func if got != checksums[i] { return fmt.Errorf("%s checksum mismatch: have %s want %s", file, got, checksums[i]) } - // Import all block data from Era1. - e, err := from(f) + + // Seek back for import. + if _, err := f.Seek(0, io.SeekStart); err != nil { + return fmt.Errorf("seek %s: %w", path, err) + } + e, err = from(f) if err != nil { return fmt.Errorf("error opening era: %w", err) } @@ -321,10 +349,14 @@ func ImportHistory(chain *core.BlockChain, dir string, network string, from func return fmt.Errorf("error inserting blocks %d-%d: %w", blocks[0].NumberU64(), blocks[len(blocks)-1].NumberU64(), err) } + // Persist tail after each successful batch. + lastBlock := blocks[len(blocks)-1].NumberU64() + rawdb.WriteEraImportTail(db, lastBlock) + resumeBlock = lastBlock + imported += len(blocks) if time.Since(reported) >= 8*time.Second { - head := blocks[len(blocks)-1].NumberU64() - log.Info("Importing Era files", "head", head, "imported", imported, + log.Info("Importing Era files", "head", lastBlock, "imported", imported, "elapsed", common.PrettyDuration(time.Since(start))) imported = 0 reported = time.Now() @@ -334,6 +366,7 @@ func ImportHistory(chain *core.BlockChain, dir string, network string, from func return nil } ) + for it.Next() { block, err := it.Block() if err != nil { @@ -342,6 +375,10 @@ func ImportHistory(chain *core.BlockChain, dir string, network string, from func if block.Number().BitLen() == 0 { continue // skip genesis } + // Skip blocks already imported (mid-era resume). + if resumeBlock > 0 && block.Number().Uint64() <= resumeBlock { + continue + } receipts, err := it.Receipts() if err != nil { return fmt.Errorf("error reading receipts %d: %w", it.Number(), err) diff --git a/cmd/utils/history_test.go b/cmd/utils/history_test.go index 6631946129..4b94117587 100644 --- a/cmd/utils/history_test.go +++ b/cmd/utils/history_test.go @@ -182,7 +182,7 @@ func TestHistoryImportAndExport(t *testing.T) { if err != nil { t.Fatalf("unable to initialize chain: %v", err) } - if err := ImportHistory(imported, dir, "mainnet", tt.from); err != nil { + if err := ImportHistory(imported, db2, dir, "mainnet", tt.from); err != nil { t.Fatalf("failed to import chain: %v", err) } if have, want := imported.CurrentHeader(), chain.CurrentHeader(); have.Hash() != want.Hash() { diff --git a/core/rawdb/accessors_indexes.go b/core/rawdb/accessors_indexes.go index 8c8c3ec9bb..c2426291db 100644 --- a/core/rawdb/accessors_indexes.go +++ b/core/rawdb/accessors_indexes.go @@ -32,6 +32,21 @@ import ( "github.com/ethereum/go-ethereum/rlp" ) +func ReadEraImportTail(db ethdb.KeyValueReader) *uint64 { + data, _ := db.Get(eraImportTailKey) + if len(data) == 0 { + return nil + } + tail := binary.BigEndian.Uint64(data) + return &tail +} + +func WriteEraImportTail(db ethdb.KeyValueWriter, blockNumber uint64) { + buf := make([]byte, 8) + binary.BigEndian.PutUint64(buf, blockNumber) + db.Put(eraImportTailKey, buf) +} + // DecodeTxLookupEntry decodes the supplied tx lookup data. func DecodeTxLookupEntry(data []byte, db ethdb.Reader) *uint64 { // Database v6 tx lookup just stores the block number diff --git a/core/rawdb/schema.go b/core/rawdb/schema.go index 54c76143b4..3a5f669b6c 100644 --- a/core/rawdb/schema.go +++ b/core/rawdb/schema.go @@ -87,6 +87,8 @@ var ( // txIndexTailKey tracks the oldest block whose transactions have been indexed. txIndexTailKey = []byte("TransactionIndexTail") + eraImportTailKey = []byte("eraImportTail") // eraImportTailKey -> last fully imported block + // fastTxLookupLimitKey tracks the transaction lookup limit during fast sync. // This flag is deprecated, it's kept to avoid reporting errors when inspect // database.