Merge 134be59d23 into 12eabbd76d

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
 }

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)

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() {

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

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.