diff --git a/cmd/geth/chaincmd.go b/cmd/geth/chaincmd.go index d333c17559..8d439e43d4 100644 --- a/cmd/geth/chaincmd.go +++ b/cmd/geth/chaincmd.go @@ -319,13 +319,13 @@ func importChain(ctx *cli.Context) error { var importErr error if ctx.Args().Len() == 1 { - if err := utils.ImportChain(chain, ctx.Args().First()); err != nil { + if err := blockchainio.ImportChain(chain, ctx.Args().First()); err != nil { importErr = err log.Error("Import error", "err", err) } } else { for _, arg := range ctx.Args().Slice() { - if err := utils.ImportChain(chain, arg); err != nil { + if err := blockchainio.ImportChain(chain, arg); err != nil { importErr = err log.Error("Import error", "file", arg, "err", err) } @@ -377,7 +377,7 @@ func exportChain(ctx *cli.Context) error { var err error fp := ctx.Args().First() if ctx.Args().Len() < 3 { - err = utils.ExportChain(chain, fp) + err = blockchainio.ExportChain(chain, fp) } else { // This can be improved to allow for numbers larger than 9223372036854775807 first, ferr := strconv.ParseInt(ctx.Args().Get(1), 10, 64) @@ -391,7 +391,7 @@ func exportChain(ctx *cli.Context) error { if head := chain.CurrentSnapBlock(); uint64(last) > head.Number.Uint64() { utils.Fatalf("Export error: block number %d larger than head block %d\n", uint64(last), head.Number.Uint64()) } - err = utils.ExportAppendChain(chain, fp, uint64(first), uint64(last)) + err = blockchainio.ExportAppendChain(chain, fp, uint64(first), uint64(last)) } if err != nil { utils.Fatalf("Export error: %v\n", err) @@ -449,7 +449,7 @@ func importHistory(ctx *cli.Context) error { network = networks[0] } - if err := utils.ImportHistory(chain, db, dir, network); err != nil { + if err := blockchainio.ImportHistory(chain, db, dir, network); err != nil { return err } fmt.Printf("Import done in %v\n", time.Since(start)) @@ -483,7 +483,7 @@ func exportHistory(ctx *cli.Context) error { if head := chain.CurrentSnapBlock(); uint64(last) > head.Number.Uint64() { utils.Fatalf("Export error: block number %d larger than head block %d\n", uint64(last), head.Number.Uint64()) } - err := utils.ExportHistory(chain, dir, uint64(first), uint64(last), uint64(era.MaxEra1Size)) + err := blockchainio.ExportHistory(chain, dir, uint64(first), uint64(last), uint64(era.MaxEra1Size)) if err != nil { utils.Fatalf("Export error: %v\n", err) } @@ -507,7 +507,7 @@ func importPreimages(ctx *cli.Context) error { defer db.Close() start := time.Now() - if err := utils.ImportPreimages(db, ctx.Args().First()); err != nil { + if err := blockchainio.ImportPreimages(db, ctx.Args().First()); err != nil { utils.Fatalf("Import error: %v\n", err) } fmt.Printf("Import done in %v\n", time.Since(start)) diff --git a/cmd/geth/dbcmd.go b/cmd/geth/dbcmd.go index 1d885bd58d..3d7555380f 100644 --- a/cmd/geth/dbcmd.go +++ b/cmd/geth/dbcmd.go @@ -630,7 +630,7 @@ func importLDBdata(ctx *cli.Context) error { }() db := utils.MakeChainDatabase(ctx, stack, false) defer db.Close() - return utils.ImportLDBData(db, fName, int64(start), stop) + return blockchainio.ImportLDBData(db, fName, int64(start), stop) } type preimageIterator struct { @@ -727,7 +727,7 @@ func exportChaindata(ctx *cli.Context) error { }() db := utils.MakeChainDatabase(ctx, stack, true) defer db.Close() - return utils.ExportChaindata(ctx.Args().Get(1), kind, exporter(db), stop) + return blockchainio.ExportChaindata(ctx.Args().Get(1), kind, exporter(db), stop) } func showMetaData(ctx *cli.Context) error { diff --git a/cmd/geth/snapshot.go b/cmd/geth/snapshot.go index 4284005a02..3632140186 100644 --- a/cmd/geth/snapshot.go +++ b/cmd/geth/snapshot.go @@ -655,7 +655,7 @@ func snapshotExportPreimages(ctx *cli.Context) error { if err != nil { return err } - return utils.ExportSnapshotPreimages(chaindb, snaptree, ctx.Args().First(), root) + return blockchainio.ExportSnapshotPreimages(chaindb, snaptree, ctx.Args().First(), root) } // checkAccount iterates the snap data layers, and looks up the given account diff --git a/cmd/utils/blockchainio.go b/cmd/utils/blockchainio.go new file mode 100644 index 0000000000..675fdf2922 --- /dev/null +++ b/cmd/utils/blockchainio.go @@ -0,0 +1,765 @@ +// Copyright 2014 The go-ethereum Authors +// This file is part of go-ethereum. +// +// go-ethereum is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// go-ethereum is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. +// +// You should have received a copy of the GNU General Public License +// along with go-ethereum. If not, see . + +// Package blockchainio contains internal helper functions for go-ethereum commands related to importing and exporting blockchain. +package blockchainio + +import ( + "bufio" + "bytes" + "compress/gzip" + "crypto/sha256" + "errors" + "fmt" + "io" + "os" + "path" + "strings" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core" + "github.com/ethereum/go-ethereum/core/rawdb" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/rlp" +) + +const ( + importBatchSize = 2500 +) + +func ImportChain(chain *core.BlockChain, fn string) error { + // Watch for Ctrl-C while the import is running. + // If a signal is received, the import will stop at the next batch. + interrupt := make(chan os.Signal, 1) + stop := make(chan struct{}) + signal.Notify(interrupt, syscall.SIGINT, syscall.SIGTERM) + defer signal.Stop(interrupt) + defer close(interrupt) + go func() { + if _, ok := <-interrupt; ok { + log.Info("Interrupted during import, stopping at next batch") + } + close(stop) + }() + checkInterrupt := func() bool { + select { + case <-stop: + return true + default: + return false + } + } + + log.Info("Importing blockchain", "file", fn) + + // Open the file handle and potentially unwrap the gzip stream + fh, err := os.Open(fn) + if err != nil { + return err + } + defer fh.Close() + + var reader io.Reader = fh + if strings.HasSuffix(fn, ".gz") { + if reader, err = gzip.NewReader(reader); err != nil { + return err + } + } + stream := rlp.NewStream(reader, 0) + + // Run actual the import. + blocks := make(types.Blocks, importBatchSize) + n := 0 + for batch := 0; ; batch++ { + // Load a batch of RLP blocks. + if checkInterrupt() { + return errors.New("interrupted") + } + i := 0 + for ; i < importBatchSize; i++ { + var b types.Block + if err := stream.Decode(&b); err == io.EOF { + break + } else if err != nil { + return fmt.Errorf("at block %d: %v", n, err) + } + // don't import first block + if b.NumberU64() == 0 { + i-- + continue + } + blocks[i] = &b + n++ + } + if i == 0 { + break + } + // Import the batch. + if checkInterrupt() { + return errors.New("interrupted") + } + missing := missingBlocks(chain, blocks[:i]) + if len(missing) == 0 { + log.Info("Skipping batch as all blocks present", "batch", batch, "first", blocks[0].Hash(), "last", blocks[i-1].Hash()) + continue + } + if failindex, err := chain.InsertChain(missing); err != nil { + var failnumber uint64 + if failindex > 0 && failindex < len(missing) { + failnumber = missing[failindex].NumberU64() + } else { + failnumber = missing[0].NumberU64() + } + return fmt.Errorf("invalid block %d: %v", failnumber, err) + } + } + return nil +} + +func readList(filename string) ([]string, error) { + b, err := os.ReadFile(filename) + if err != nil { + return nil, err + } + return strings.Split(string(b), "\n"), nil +} + +// ImportHistory imports Era1 files containing historical block information, +// starting from genesis. +func ImportHistory(chain *core.BlockChain, db ethdb.Database, dir string, network string) error { + if chain.CurrentSnapBlock().Number.BitLen() != 0 { + return fmt.Errorf("history import only supported when starting from genesis") + } + entries, err := era.ReadDir(dir, network) + if err != nil { + return fmt.Errorf("error reading %s: %w", dir, err) + } + checksums, err := readList(path.Join(dir, "checksums.txt")) + if err != nil { + return fmt.Errorf("unable to read checksums.txt: %w", err) + } + if len(checksums) != len(entries) { + return fmt.Errorf("expected equal number of checksums and entries, have: %d checksums, %d entries", len(checksums), len(entries)) + } + var ( + start = time.Now() + reported = time.Now() + imported = 0 + forker = core.NewForkChoice(chain, nil) + h = sha256.New() + buf = bytes.NewBuffer(nil) + ) + for i, filename := range entries { + err := func() error { + f, err := os.Open(path.Join(dir, filename)) + if err != nil { + return fmt.Errorf("unable to open era: %w", err) + } + defer f.Close() + + // Validate checksum. + if _, err := io.Copy(h, f); err != nil { + return fmt.Errorf("unable to recalculate checksum: %w", err) + } + if have, want := common.BytesToHash(h.Sum(buf.Bytes()[:])).Hex(), checksums[i]; have != want { + return fmt.Errorf("checksum mismatch: have %s, want %s", have, want) + } + h.Reset() + buf.Reset() + + // Import all block data from Era1. + e, err := era.From(f) + if err != nil { + return fmt.Errorf("error opening era: %w", err) + } + it, err := era.NewIterator(e) + if err != nil { + return fmt.Errorf("error making era reader: %w", err) + } + for it.Next() { + block, err := it.Block() + if err != nil { + return fmt.Errorf("error reading block %d: %w", it.Number(), err) + } + if block.Number().BitLen() == 0 { + continue // skip genesis + } + receipts, err := it.Receipts() + if err != nil { + return fmt.Errorf("error reading receipts %d: %w", it.Number(), err) + } + if status, err := chain.HeaderChain().InsertHeaderChain([]*types.Header{block.Header()}, start, forker); err != nil { + return fmt.Errorf("error inserting header %d: %w", it.Number(), err) + } else if status != core.CanonStatTy { + return fmt.Errorf("error inserting header %d, not canon: %v", it.Number(), status) + } + if _, err := chain.InsertReceiptChain([]*types.Block{block}, []types.Receipts{receipts}, 2^64-1); err != nil { + return fmt.Errorf("error inserting body %d: %w", it.Number(), err) + } + imported += 1 + + // Give the user some feedback that something is happening. + if time.Since(reported) >= 8*time.Second { + log.Info("Importing Era files", "head", it.Number(), "imported", imported, "elapsed", common.PrettyDuration(time.Since(start))) + imported = 0 + reported = time.Now() + } + } + return nil + }() + if err != nil { + return err + } + } + + return nil +} + +func missingBlocks(chain *core.BlockChain, blocks []*types.Block) []*types.Block { + head := chain.CurrentBlock() + for i, block := range blocks { + // If we're behind the chain head, only check block, state is available at head + if head.Number.Uint64() > block.NumberU64() { + if !chain.HasBlock(block.Hash(), block.NumberU64()) { + return blocks[i:] + } + continue + } + // If we're above the chain head, state availability is a must + if !chain.HasBlockAndState(block.Hash(), block.NumberU64()) { + return blocks[i:] + } + } + return nil +} + +// ExportChain exports a blockchain into the specified file, truncating any data +// already present in the file. +func ExportChain(blockchain *core.BlockChain, fn string) error { + log.Info("Exporting blockchain", "file", fn) + + // Open the file handle and potentially wrap with a gzip stream + fh, err := os.OpenFile(fn, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, os.ModePerm) + if err != nil { + return err + } + defer fh.Close() + + var writer io.Writer = fh + if strings.HasSuffix(fn, ".gz") { + writer = gzip.NewWriter(writer) + defer writer.(*gzip.Writer).Close() + } + // Iterate over the blocks and export them + if err := blockchain.Export(writer); err != nil { + return err + } + log.Info("Exported blockchain", "file", fn) + + return nil +} + +// ExportAppendChain exports a blockchain into the specified file, appending to +// the file if data already exists in it. +func ExportAppendChain(blockchain *core.BlockChain, fn string, first uint64, last uint64) error { + log.Info("Exporting blockchain", "file", fn) + + // Open the file handle and potentially wrap with a gzip stream + fh, err := os.OpenFile(fn, os.O_CREATE|os.O_APPEND|os.O_WRONLY, os.ModePerm) + if err != nil { + return err + } + defer fh.Close() + + var writer io.Writer = fh + if strings.HasSuffix(fn, ".gz") { + writer = gzip.NewWriter(writer) + defer writer.(*gzip.Writer).Close() + } + // Iterate over the blocks and export them + if err := blockchain.ExportN(writer, first, last); err != nil { + return err + } + log.Info("Exported blockchain to", "file", fn) + return nil +} + +// ExportHistory exports blockchain history into the specified directory, +// following the Era format. +func ExportHistory(bc *core.BlockChain, dir string, first, last, step uint64) error { + log.Info("Exporting blockchain history", "dir", dir) + if head := bc.CurrentBlock().Number.Uint64(); head < last { + log.Warn("Last block beyond head, setting last = head", "head", head, "last", last) + last = head + } + network := "unknown" + if name, ok := params.NetworkNames[bc.Config().ChainID.String()]; ok { + network = name + } + if err := os.MkdirAll(dir, os.ModePerm); err != nil { + return fmt.Errorf("error creating output directory: %w", err) + } + var ( + start = time.Now() + reported = time.Now() + h = sha256.New() + buf = bytes.NewBuffer(nil) + checksums []string + ) + for i := first; i <= last; i += step { + err := func() error { + filename := path.Join(dir, era.Filename(network, int(i/step), common.Hash{})) + f, err := os.Create(filename) + if err != nil { + return fmt.Errorf("could not create era file: %w", err) + } + defer f.Close() + + w := era.NewBuilder(f) + for j := uint64(0); j < step && j <= last-i; j++ { + var ( + n = i + j + block = bc.GetBlockByNumber(n) + ) + if block == nil { + return fmt.Errorf("export failed on #%d: not found", n) + } + receipts := bc.GetReceiptsByHash(block.Hash()) + if receipts == nil { + return fmt.Errorf("export failed on #%d: receipts not found", n) + } + td := bc.GetTd(block.Hash(), block.NumberU64()) + if td == nil { + return fmt.Errorf("export failed on #%d: total difficulty not found", n) + } + if err := w.Add(block, receipts, td); err != nil { + return err + } + } + root, err := w.Finalize() + if err != nil { + return fmt.Errorf("export failed to finalize %d: %w", step/i, err) + } + // Set correct filename with root. + os.Rename(filename, path.Join(dir, era.Filename(network, int(i/step), root))) + + // Compute checksum of entire Era1. + if _, err := f.Seek(0, io.SeekStart); err != nil { + return err + } + if _, err := io.Copy(h, f); err != nil { + return fmt.Errorf("unable to calculate checksum: %w", err) + } + checksums = append(checksums, common.BytesToHash(h.Sum(buf.Bytes()[:])).Hex()) + h.Reset() + buf.Reset() + return nil + }() + if err != nil { + return err + } + if time.Since(reported) >= 8*time.Second { + log.Info("Exporting blocks", "exported", i, "elapsed", common.PrettyDuration(time.Since(start))) + reported = time.Now() + } + } + + os.WriteFile(path.Join(dir, "checksums.txt"), []byte(strings.Join(checksums, "\n")), os.ModePerm) + + log.Info("Exported blockchain to", "dir", dir) + + return nil +} + +// ImportPreimages imports a batch of exported hash preimages into the database. +// It's a part of the deprecated functionality, should be removed in the future. +func ImportPreimages(db ethdb.Database, fn string) error { + log.Info("Importing preimages", "file", fn) + + // Open the file handle and potentially unwrap the gzip stream + fh, err := os.Open(fn) + if err != nil { + return err + } + defer fh.Close() + + var reader io.Reader = bufio.NewReader(fh) + if strings.HasSuffix(fn, ".gz") { + if reader, err = gzip.NewReader(reader); err != nil { + return err + } + } + stream := rlp.NewStream(reader, 0) + + // Import the preimages in batches to prevent disk thrashing + preimages := make(map[common.Hash][]byte) + + for { + // Read the next entry and ensure it's not junk + var blob []byte + + if err := stream.Decode(&blob); err != nil { + if err == io.EOF { + break + } + return err + } + // Accumulate the preimages and flush when enough ws gathered + preimages[crypto.Keccak256Hash(blob)] = common.CopyBytes(blob) + if len(preimages) > 1024 { + rawdb.WritePreimages(db, preimages) + preimages = make(map[common.Hash][]byte) + } + } + // Flush the last batch preimage data + if len(preimages) > 0 { + rawdb.WritePreimages(db, preimages) + } + return nil +} + +// ExportPreimages exports all known hash preimages into the specified file, +// truncating any data already present in the file. +// It's a part of the deprecated functionality, should be removed in the future. +func ExportPreimages(db ethdb.Database, fn string) error { + log.Info("Exporting preimages", "file", fn) + + // Open the file handle and potentially wrap with a gzip stream + fh, err := os.OpenFile(fn, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, os.ModePerm) + if err != nil { + return err + } + defer fh.Close() + + var writer io.Writer = fh + if strings.HasSuffix(fn, ".gz") { + writer = gzip.NewWriter(writer) + defer writer.(*gzip.Writer).Close() + } + // Iterate over the preimages and export them + it := db.NewIterator([]byte("secure-key-"), nil) + defer it.Release() + + for it.Next() { + if err := rlp.Encode(writer, it.Value()); err != nil { + return err + } + } + log.Info("Exported preimages", "file", fn) + return nil +} + +// ExportSnapshotPreimages exports the preimages corresponding to the enumeration of +// the snapshot for a given root. +func ExportSnapshotPreimages(chaindb ethdb.Database, snaptree *snapshot.Tree, fn string, root common.Hash) error { + log.Info("Exporting preimages", "file", fn) + + fh, err := os.OpenFile(fn, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, os.ModePerm) + if err != nil { + return err + } + defer fh.Close() + + // Enable gzip compressing if file name has gz suffix. + var writer io.Writer = fh + if strings.HasSuffix(fn, ".gz") { + gz := gzip.NewWriter(writer) + defer gz.Close() + writer = gz + } + buf := bufio.NewWriter(writer) + defer buf.Flush() + writer = buf + + type hashAndPreimageSize struct { + Hash common.Hash + Size int + } + hashCh := make(chan hashAndPreimageSize) + + var ( + start = time.Now() + logged = time.Now() + preimages int + ) + go func() { + defer close(hashCh) + accIt, err := snaptree.AccountIterator(root, common.Hash{}) + if err != nil { + log.Error("Failed to create account iterator", "error", err) + return + } + defer accIt.Release() + + for accIt.Next() { + acc, err := types.FullAccount(accIt.Account()) + if err != nil { + log.Error("Failed to get full account", "error", err) + return + } + preimages += 1 + hashCh <- hashAndPreimageSize{Hash: accIt.Hash(), Size: common.AddressLength} + + if acc.Root != (common.Hash{}) && acc.Root != types.EmptyRootHash { + stIt, err := snaptree.StorageIterator(root, accIt.Hash(), common.Hash{}) + if err != nil { + log.Error("Failed to create storage iterator", "error", err) + return + } + for stIt.Next() { + preimages += 1 + hashCh <- hashAndPreimageSize{Hash: stIt.Hash(), Size: common.HashLength} + + if time.Since(logged) > time.Second*8 { + logged = time.Now() + log.Info("Exporting preimages", "count", preimages, "elapsed", common.PrettyDuration(time.Since(start))) + } + } + stIt.Release() + } + if time.Since(logged) > time.Second*8 { + logged = time.Now() + log.Info("Exporting preimages", "count", preimages, "elapsed", common.PrettyDuration(time.Since(start))) + } + } + }() + + for item := range hashCh { + preimage := rawdb.ReadPreimage(chaindb, item.Hash) + if len(preimage) == 0 { + return fmt.Errorf("missing preimage for %v", item.Hash) + } + if len(preimage) != item.Size { + return fmt.Errorf("invalid preimage size, have %d", len(preimage)) + } + rlpenc, err := rlp.EncodeToBytes(preimage) + if err != nil { + return fmt.Errorf("error encoding preimage: %w", err) + } + if _, err := writer.Write(rlpenc); err != nil { + return fmt.Errorf("failed to write preimage: %w", err) + } + } + log.Info("Exported preimages", "count", preimages, "elapsed", common.PrettyDuration(time.Since(start)), "file", fn) + return nil +} + +// exportHeader is used in the export/import flow. When we do an export, +// the first element we output is the exportHeader. +// Whenever a backwards-incompatible change is made, the Version header +// should be bumped. +// If the importer sees a higher version, it should reject the import. +type exportHeader struct { + Magic string // Always set to 'gethdbdump' for disambiguation + Version uint64 + Kind string + UnixTime uint64 +} + +const exportMagic = "gethdbdump" +const ( + OpBatchAdd = 0 + OpBatchDel = 1 +) + +// ImportLDBData imports a batch of snapshot data into the database +func ImportLDBData(db ethdb.Database, f string, startIndex int64, interrupt chan struct{}) error { + log.Info("Importing leveldb data", "file", f) + + // Open the file handle and potentially unwrap the gzip stream + fh, err := os.Open(f) + if err != nil { + return err + } + defer fh.Close() + + var reader io.Reader = bufio.NewReader(fh) + if strings.HasSuffix(f, ".gz") { + if reader, err = gzip.NewReader(reader); err != nil { + return err + } + } + stream := rlp.NewStream(reader, 0) + + // Read the header + var header exportHeader + if err := stream.Decode(&header); err != nil { + return fmt.Errorf("could not decode header: %v", err) + } + if header.Magic != exportMagic { + return errors.New("incompatible data, wrong magic") + } + if header.Version != 0 { + return fmt.Errorf("incompatible version %d, (support only 0)", header.Version) + } + log.Info("Importing data", "file", f, "type", header.Kind, "data age", + common.PrettyDuration(time.Since(time.Unix(int64(header.UnixTime), 0)))) + + // Import the snapshot in batches to prevent disk thrashing + var ( + count int64 + start = time.Now() + logged = time.Now() + batch = db.NewBatch() + ) + for { + // Read the next entry + var ( + op byte + key, val []byte + ) + if err := stream.Decode(&op); err != nil { + if err == io.EOF { + break + } + return err + } + if err := stream.Decode(&key); err != nil { + return err + } + if err := stream.Decode(&val); err != nil { + return err + } + if count < startIndex { + count++ + continue + } + switch op { + case OpBatchDel: + batch.Delete(key) + case OpBatchAdd: + batch.Put(key, val) + default: + return fmt.Errorf("unknown op %d", op) + } + if batch.ValueSize() > ethdb.IdealBatchSize { + if err := batch.Write(); err != nil { + return err + } + batch.Reset() + } + // Check interruption emitted by ctrl+c + if count%1000 == 0 { + select { + case <-interrupt: + if err := batch.Write(); err != nil { + return err + } + log.Info("External data import interrupted", "file", f, "count", count, "elapsed", common.PrettyDuration(time.Since(start))) + return nil + default: + } + } + if count%1000 == 0 && time.Since(logged) > 8*time.Second { + log.Info("Importing external data", "file", f, "count", count, "elapsed", common.PrettyDuration(time.Since(start))) + logged = time.Now() + } + count += 1 + } + // Flush the last batch snapshot data + if batch.ValueSize() > 0 { + if err := batch.Write(); err != nil { + return err + } + } + log.Info("Imported chain data", "file", f, "count", count, + "elapsed", common.PrettyDuration(time.Since(start))) + return nil +} + +// ChainDataIterator is an interface wraps all necessary functions to iterate +// the exporting chain data. +type ChainDataIterator interface { + // Next returns the key-value pair for next exporting entry in the iterator. + // When the end is reached, it will return (0, nil, nil, false). + Next() (byte, []byte, []byte, bool) + + // Release releases associated resources. Release should always succeed and can + // be called multiple times without causing error. + Release() +} + +// ExportChaindata exports the given data type (truncating any data already present) +// in the file. If the suffix is 'gz', gzip compression is used. +func ExportChaindata(fn string, kind string, iter ChainDataIterator, interrupt chan struct{}) error { + log.Info("Exporting chain data", "file", fn, "kind", kind) + defer iter.Release() + + // Open the file handle and potentially wrap with a gzip stream + fh, err := os.OpenFile(fn, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, os.ModePerm) + if err != nil { + return err + } + defer fh.Close() + + var writer io.Writer = fh + if strings.HasSuffix(fn, ".gz") { + writer = gzip.NewWriter(writer) + defer writer.(*gzip.Writer).Close() + } + // Write the header + if err := rlp.Encode(writer, &exportHeader{ + Magic: exportMagic, + Version: 0, + Kind: kind, + UnixTime: uint64(time.Now().Unix()), + }); err != nil { + return err + } + // Extract data from source iterator and dump them out to file + var ( + count int64 + start = time.Now() + logged = time.Now() + ) + for { + op, key, val, ok := iter.Next() + if !ok { + break + } + if err := rlp.Encode(writer, op); err != nil { + return err + } + if err := rlp.Encode(writer, key); err != nil { + return err + } + if err := rlp.Encode(writer, val); err != nil { + return err + } + if count%1000 == 0 { + // Check interruption emitted by ctrl+c + select { + case <-interrupt: + log.Info("Chain data exporting interrupted", "file", fn, + "kind", kind, "count", count, "elapsed", common.PrettyDuration(time.Since(start))) + return nil + default: + } + if time.Since(logged) > 8*time.Second { + log.Info("Exporting chain data", "file", fn, "kind", kind, + "count", count, "elapsed", common.PrettyDuration(time.Since(start))) + logged = time.Now() + } + } + count++ + } + log.Info("Exported chain data", "file", fn, "kind", kind, "count", count, + "elapsed", common.PrettyDuration(time.Since(start))) + return nil +} diff --git a/cmd/utils/cmd.go b/cmd/utils/cmd.go index 4b57164665..fca716d4b8 100644 --- a/cmd/utils/cmd.go +++ b/cmd/utils/cmd.go @@ -143,724 +143,3 @@ func monitorFreeDiskSpace(sigc chan os.Signal, path string, freeDiskSpaceCritica time.Sleep(30 * time.Second) } } - -func ImportChain(chain *core.BlockChain, fn string) error { - // Watch for Ctrl-C while the import is running. - // If a signal is received, the import will stop at the next batch. - interrupt := make(chan os.Signal, 1) - stop := make(chan struct{}) - signal.Notify(interrupt, syscall.SIGINT, syscall.SIGTERM) - defer signal.Stop(interrupt) - defer close(interrupt) - go func() { - if _, ok := <-interrupt; ok { - log.Info("Interrupted during import, stopping at next batch") - } - close(stop) - }() - checkInterrupt := func() bool { - select { - case <-stop: - return true - default: - return false - } - } - - log.Info("Importing blockchain", "file", fn) - - // Open the file handle and potentially unwrap the gzip stream - fh, err := os.Open(fn) - if err != nil { - return err - } - defer fh.Close() - - var reader io.Reader = fh - if strings.HasSuffix(fn, ".gz") { - if reader, err = gzip.NewReader(reader); err != nil { - return err - } - } - stream := rlp.NewStream(reader, 0) - - // Run actual the import. - blocks := make(types.Blocks, importBatchSize) - n := 0 - for batch := 0; ; batch++ { - // Load a batch of RLP blocks. - if checkInterrupt() { - return errors.New("interrupted") - } - i := 0 - for ; i < importBatchSize; i++ { - var b types.Block - if err := stream.Decode(&b); err == io.EOF { - break - } else if err != nil { - return fmt.Errorf("at block %d: %v", n, err) - } - // don't import first block - if b.NumberU64() == 0 { - i-- - continue - } - blocks[i] = &b - n++ - } - if i == 0 { - break - } - // Import the batch. - if checkInterrupt() { - return errors.New("interrupted") - } - missing := missingBlocks(chain, blocks[:i]) - if len(missing) == 0 { - log.Info("Skipping batch as all blocks present", "batch", batch, "first", blocks[0].Hash(), "last", blocks[i-1].Hash()) - continue - } - if failindex, err := chain.InsertChain(missing); err != nil { - var failnumber uint64 - if failindex > 0 && failindex < len(missing) { - failnumber = missing[failindex].NumberU64() - } else { - failnumber = missing[0].NumberU64() - } - return fmt.Errorf("invalid block %d: %v", failnumber, err) - } - } - return nil -} - -func readList(filename string) ([]string, error) { - b, err := os.ReadFile(filename) - if err != nil { - return nil, err - } - return strings.Split(string(b), "\n"), nil -} - -// ImportHistory imports Era1 files containing historical block information, -// starting from genesis. -func ImportHistory(chain *core.BlockChain, db ethdb.Database, dir string, network string) error { - if chain.CurrentSnapBlock().Number.BitLen() != 0 { - return fmt.Errorf("history import only supported when starting from genesis") - } - entries, err := era.ReadDir(dir, network) - if err != nil { - return fmt.Errorf("error reading %s: %w", dir, err) - } - checksums, err := readList(path.Join(dir, "checksums.txt")) - if err != nil { - return fmt.Errorf("unable to read checksums.txt: %w", err) - } - if len(checksums) != len(entries) { - return fmt.Errorf("expected equal number of checksums and entries, have: %d checksums, %d entries", len(checksums), len(entries)) - } - var ( - start = time.Now() - reported = time.Now() - imported = 0 - forker = core.NewForkChoice(chain, nil) - h = sha256.New() - buf = bytes.NewBuffer(nil) - ) - for i, filename := range entries { - err := func() error { - f, err := os.Open(path.Join(dir, filename)) - if err != nil { - return fmt.Errorf("unable to open era: %w", err) - } - defer f.Close() - - // Validate checksum. - if _, err := io.Copy(h, f); err != nil { - return fmt.Errorf("unable to recalculate checksum: %w", err) - } - if have, want := common.BytesToHash(h.Sum(buf.Bytes()[:])).Hex(), checksums[i]; have != want { - return fmt.Errorf("checksum mismatch: have %s, want %s", have, want) - } - h.Reset() - buf.Reset() - - // Import all block data from Era1. - e, err := era.From(f) - if err != nil { - return fmt.Errorf("error opening era: %w", err) - } - it, err := era.NewIterator(e) - if err != nil { - return fmt.Errorf("error making era reader: %w", err) - } - for it.Next() { - block, err := it.Block() - if err != nil { - return fmt.Errorf("error reading block %d: %w", it.Number(), err) - } - if block.Number().BitLen() == 0 { - continue // skip genesis - } - receipts, err := it.Receipts() - if err != nil { - return fmt.Errorf("error reading receipts %d: %w", it.Number(), err) - } - if status, err := chain.HeaderChain().InsertHeaderChain([]*types.Header{block.Header()}, start, forker); err != nil { - return fmt.Errorf("error inserting header %d: %w", it.Number(), err) - } else if status != core.CanonStatTy { - return fmt.Errorf("error inserting header %d, not canon: %v", it.Number(), status) - } - if _, err := chain.InsertReceiptChain([]*types.Block{block}, []types.Receipts{receipts}, 2^64-1); err != nil { - return fmt.Errorf("error inserting body %d: %w", it.Number(), err) - } - imported += 1 - - // Give the user some feedback that something is happening. - if time.Since(reported) >= 8*time.Second { - log.Info("Importing Era files", "head", it.Number(), "imported", imported, "elapsed", common.PrettyDuration(time.Since(start))) - imported = 0 - reported = time.Now() - } - } - return nil - }() - if err != nil { - return err - } - } - - return nil -} - -func missingBlocks(chain *core.BlockChain, blocks []*types.Block) []*types.Block { - head := chain.CurrentBlock() - for i, block := range blocks { - // If we're behind the chain head, only check block, state is available at head - if head.Number.Uint64() > block.NumberU64() { - if !chain.HasBlock(block.Hash(), block.NumberU64()) { - return blocks[i:] - } - continue - } - // If we're above the chain head, state availability is a must - if !chain.HasBlockAndState(block.Hash(), block.NumberU64()) { - return blocks[i:] - } - } - return nil -} - -// ExportChain exports a blockchain into the specified file, truncating any data -// already present in the file. -func ExportChain(blockchain *core.BlockChain, fn string) error { - log.Info("Exporting blockchain", "file", fn) - - // Open the file handle and potentially wrap with a gzip stream - fh, err := os.OpenFile(fn, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, os.ModePerm) - if err != nil { - return err - } - defer fh.Close() - - var writer io.Writer = fh - if strings.HasSuffix(fn, ".gz") { - writer = gzip.NewWriter(writer) - defer writer.(*gzip.Writer).Close() - } - // Iterate over the blocks and export them - if err := blockchain.Export(writer); err != nil { - return err - } - log.Info("Exported blockchain", "file", fn) - - return nil -} - -// ExportAppendChain exports a blockchain into the specified file, appending to -// the file if data already exists in it. -func ExportAppendChain(blockchain *core.BlockChain, fn string, first uint64, last uint64) error { - log.Info("Exporting blockchain", "file", fn) - - // Open the file handle and potentially wrap with a gzip stream - fh, err := os.OpenFile(fn, os.O_CREATE|os.O_APPEND|os.O_WRONLY, os.ModePerm) - if err != nil { - return err - } - defer fh.Close() - - var writer io.Writer = fh - if strings.HasSuffix(fn, ".gz") { - writer = gzip.NewWriter(writer) - defer writer.(*gzip.Writer).Close() - } - // Iterate over the blocks and export them - if err := blockchain.ExportN(writer, first, last); err != nil { - return err - } - log.Info("Exported blockchain to", "file", fn) - return nil -} - -// ExportHistory exports blockchain history into the specified directory, -// following the Era format. -func ExportHistory(bc *core.BlockChain, dir string, first, last, step uint64) error { - log.Info("Exporting blockchain history", "dir", dir) - if head := bc.CurrentBlock().Number.Uint64(); head < last { - log.Warn("Last block beyond head, setting last = head", "head", head, "last", last) - last = head - } - network := "unknown" - if name, ok := params.NetworkNames[bc.Config().ChainID.String()]; ok { - network = name - } - if err := os.MkdirAll(dir, os.ModePerm); err != nil { - return fmt.Errorf("error creating output directory: %w", err) - } - var ( - start = time.Now() - reported = time.Now() - h = sha256.New() - buf = bytes.NewBuffer(nil) - checksums []string - ) - for i := first; i <= last; i += step { - err := func() error { - filename := path.Join(dir, era.Filename(network, int(i/step), common.Hash{})) - f, err := os.Create(filename) - if err != nil { - return fmt.Errorf("could not create era file: %w", err) - } - defer f.Close() - - w := era.NewBuilder(f) - for j := uint64(0); j < step && j <= last-i; j++ { - var ( - n = i + j - block = bc.GetBlockByNumber(n) - ) - if block == nil { - return fmt.Errorf("export failed on #%d: not found", n) - } - receipts := bc.GetReceiptsByHash(block.Hash()) - if receipts == nil { - return fmt.Errorf("export failed on #%d: receipts not found", n) - } - td := bc.GetTd(block.Hash(), block.NumberU64()) - if td == nil { - return fmt.Errorf("export failed on #%d: total difficulty not found", n) - } - if err := w.Add(block, receipts, td); err != nil { - return err - } - } - root, err := w.Finalize() - if err != nil { - return fmt.Errorf("export failed to finalize %d: %w", step/i, err) - } - // Set correct filename with root. - os.Rename(filename, path.Join(dir, era.Filename(network, int(i/step), root))) - - // Compute checksum of entire Era1. - if _, err := f.Seek(0, io.SeekStart); err != nil { - return err - } - if _, err := io.Copy(h, f); err != nil { - return fmt.Errorf("unable to calculate checksum: %w", err) - } - checksums = append(checksums, common.BytesToHash(h.Sum(buf.Bytes()[:])).Hex()) - h.Reset() - buf.Reset() - return nil - }() - if err != nil { - return err - } - if time.Since(reported) >= 8*time.Second { - log.Info("Exporting blocks", "exported", i, "elapsed", common.PrettyDuration(time.Since(start))) - reported = time.Now() - } - } - - os.WriteFile(path.Join(dir, "checksums.txt"), []byte(strings.Join(checksums, "\n")), os.ModePerm) - - log.Info("Exported blockchain to", "dir", dir) - - return nil -} - -// ImportPreimages imports a batch of exported hash preimages into the database. -// It's a part of the deprecated functionality, should be removed in the future. -func ImportPreimages(db ethdb.Database, fn string) error { - log.Info("Importing preimages", "file", fn) - - // Open the file handle and potentially unwrap the gzip stream - fh, err := os.Open(fn) - if err != nil { - return err - } - defer fh.Close() - - var reader io.Reader = bufio.NewReader(fh) - if strings.HasSuffix(fn, ".gz") { - if reader, err = gzip.NewReader(reader); err != nil { - return err - } - } - stream := rlp.NewStream(reader, 0) - - // Import the preimages in batches to prevent disk thrashing - preimages := make(map[common.Hash][]byte) - - for { - // Read the next entry and ensure it's not junk - var blob []byte - - if err := stream.Decode(&blob); err != nil { - if err == io.EOF { - break - } - return err - } - // Accumulate the preimages and flush when enough ws gathered - preimages[crypto.Keccak256Hash(blob)] = common.CopyBytes(blob) - if len(preimages) > 1024 { - rawdb.WritePreimages(db, preimages) - preimages = make(map[common.Hash][]byte) - } - } - // Flush the last batch preimage data - if len(preimages) > 0 { - rawdb.WritePreimages(db, preimages) - } - return nil -} - -// ExportPreimages exports all known hash preimages into the specified file, -// truncating any data already present in the file. -// It's a part of the deprecated functionality, should be removed in the future. -func ExportPreimages(db ethdb.Database, fn string) error { - log.Info("Exporting preimages", "file", fn) - - // Open the file handle and potentially wrap with a gzip stream - fh, err := os.OpenFile(fn, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, os.ModePerm) - if err != nil { - return err - } - defer fh.Close() - - var writer io.Writer = fh - if strings.HasSuffix(fn, ".gz") { - writer = gzip.NewWriter(writer) - defer writer.(*gzip.Writer).Close() - } - // Iterate over the preimages and export them - it := db.NewIterator([]byte("secure-key-"), nil) - defer it.Release() - - for it.Next() { - if err := rlp.Encode(writer, it.Value()); err != nil { - return err - } - } - log.Info("Exported preimages", "file", fn) - return nil -} - -// ExportSnapshotPreimages exports the preimages corresponding to the enumeration of -// the snapshot for a given root. -func ExportSnapshotPreimages(chaindb ethdb.Database, snaptree *snapshot.Tree, fn string, root common.Hash) error { - log.Info("Exporting preimages", "file", fn) - - fh, err := os.OpenFile(fn, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, os.ModePerm) - if err != nil { - return err - } - defer fh.Close() - - // Enable gzip compressing if file name has gz suffix. - var writer io.Writer = fh - if strings.HasSuffix(fn, ".gz") { - gz := gzip.NewWriter(writer) - defer gz.Close() - writer = gz - } - buf := bufio.NewWriter(writer) - defer buf.Flush() - writer = buf - - type hashAndPreimageSize struct { - Hash common.Hash - Size int - } - hashCh := make(chan hashAndPreimageSize) - - var ( - start = time.Now() - logged = time.Now() - preimages int - ) - go func() { - defer close(hashCh) - accIt, err := snaptree.AccountIterator(root, common.Hash{}) - if err != nil { - log.Error("Failed to create account iterator", "error", err) - return - } - defer accIt.Release() - - for accIt.Next() { - acc, err := types.FullAccount(accIt.Account()) - if err != nil { - log.Error("Failed to get full account", "error", err) - return - } - preimages += 1 - hashCh <- hashAndPreimageSize{Hash: accIt.Hash(), Size: common.AddressLength} - - if acc.Root != (common.Hash{}) && acc.Root != types.EmptyRootHash { - stIt, err := snaptree.StorageIterator(root, accIt.Hash(), common.Hash{}) - if err != nil { - log.Error("Failed to create storage iterator", "error", err) - return - } - for stIt.Next() { - preimages += 1 - hashCh <- hashAndPreimageSize{Hash: stIt.Hash(), Size: common.HashLength} - - if time.Since(logged) > time.Second*8 { - logged = time.Now() - log.Info("Exporting preimages", "count", preimages, "elapsed", common.PrettyDuration(time.Since(start))) - } - } - stIt.Release() - } - if time.Since(logged) > time.Second*8 { - logged = time.Now() - log.Info("Exporting preimages", "count", preimages, "elapsed", common.PrettyDuration(time.Since(start))) - } - } - }() - - for item := range hashCh { - preimage := rawdb.ReadPreimage(chaindb, item.Hash) - if len(preimage) == 0 { - return fmt.Errorf("missing preimage for %v", item.Hash) - } - if len(preimage) != item.Size { - return fmt.Errorf("invalid preimage size, have %d", len(preimage)) - } - rlpenc, err := rlp.EncodeToBytes(preimage) - if err != nil { - return fmt.Errorf("error encoding preimage: %w", err) - } - if _, err := writer.Write(rlpenc); err != nil { - return fmt.Errorf("failed to write preimage: %w", err) - } - } - log.Info("Exported preimages", "count", preimages, "elapsed", common.PrettyDuration(time.Since(start)), "file", fn) - return nil -} - -// exportHeader is used in the export/import flow. When we do an export, -// the first element we output is the exportHeader. -// Whenever a backwards-incompatible change is made, the Version header -// should be bumped. -// If the importer sees a higher version, it should reject the import. -type exportHeader struct { - Magic string // Always set to 'gethdbdump' for disambiguation - Version uint64 - Kind string - UnixTime uint64 -} - -const exportMagic = "gethdbdump" -const ( - OpBatchAdd = 0 - OpBatchDel = 1 -) - -// ImportLDBData imports a batch of snapshot data into the database -func ImportLDBData(db ethdb.Database, f string, startIndex int64, interrupt chan struct{}) error { - log.Info("Importing leveldb data", "file", f) - - // Open the file handle and potentially unwrap the gzip stream - fh, err := os.Open(f) - if err != nil { - return err - } - defer fh.Close() - - var reader io.Reader = bufio.NewReader(fh) - if strings.HasSuffix(f, ".gz") { - if reader, err = gzip.NewReader(reader); err != nil { - return err - } - } - stream := rlp.NewStream(reader, 0) - - // Read the header - var header exportHeader - if err := stream.Decode(&header); err != nil { - return fmt.Errorf("could not decode header: %v", err) - } - if header.Magic != exportMagic { - return errors.New("incompatible data, wrong magic") - } - if header.Version != 0 { - return fmt.Errorf("incompatible version %d, (support only 0)", header.Version) - } - log.Info("Importing data", "file", f, "type", header.Kind, "data age", - common.PrettyDuration(time.Since(time.Unix(int64(header.UnixTime), 0)))) - - // Import the snapshot in batches to prevent disk thrashing - var ( - count int64 - start = time.Now() - logged = time.Now() - batch = db.NewBatch() - ) - for { - // Read the next entry - var ( - op byte - key, val []byte - ) - if err := stream.Decode(&op); err != nil { - if err == io.EOF { - break - } - return err - } - if err := stream.Decode(&key); err != nil { - return err - } - if err := stream.Decode(&val); err != nil { - return err - } - if count < startIndex { - count++ - continue - } - switch op { - case OpBatchDel: - batch.Delete(key) - case OpBatchAdd: - batch.Put(key, val) - default: - return fmt.Errorf("unknown op %d", op) - } - if batch.ValueSize() > ethdb.IdealBatchSize { - if err := batch.Write(); err != nil { - return err - } - batch.Reset() - } - // Check interruption emitted by ctrl+c - if count%1000 == 0 { - select { - case <-interrupt: - if err := batch.Write(); err != nil { - return err - } - log.Info("External data import interrupted", "file", f, "count", count, "elapsed", common.PrettyDuration(time.Since(start))) - return nil - default: - } - } - if count%1000 == 0 && time.Since(logged) > 8*time.Second { - log.Info("Importing external data", "file", f, "count", count, "elapsed", common.PrettyDuration(time.Since(start))) - logged = time.Now() - } - count += 1 - } - // Flush the last batch snapshot data - if batch.ValueSize() > 0 { - if err := batch.Write(); err != nil { - return err - } - } - log.Info("Imported chain data", "file", f, "count", count, - "elapsed", common.PrettyDuration(time.Since(start))) - return nil -} - -// ChainDataIterator is an interface wraps all necessary functions to iterate -// the exporting chain data. -type ChainDataIterator interface { - // Next returns the key-value pair for next exporting entry in the iterator. - // When the end is reached, it will return (0, nil, nil, false). - Next() (byte, []byte, []byte, bool) - - // Release releases associated resources. Release should always succeed and can - // be called multiple times without causing error. - Release() -} - -// ExportChaindata exports the given data type (truncating any data already present) -// in the file. If the suffix is 'gz', gzip compression is used. -func ExportChaindata(fn string, kind string, iter ChainDataIterator, interrupt chan struct{}) error { - log.Info("Exporting chain data", "file", fn, "kind", kind) - defer iter.Release() - - // Open the file handle and potentially wrap with a gzip stream - fh, err := os.OpenFile(fn, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, os.ModePerm) - if err != nil { - return err - } - defer fh.Close() - - var writer io.Writer = fh - if strings.HasSuffix(fn, ".gz") { - writer = gzip.NewWriter(writer) - defer writer.(*gzip.Writer).Close() - } - // Write the header - if err := rlp.Encode(writer, &exportHeader{ - Magic: exportMagic, - Version: 0, - Kind: kind, - UnixTime: uint64(time.Now().Unix()), - }); err != nil { - return err - } - // Extract data from source iterator and dump them out to file - var ( - count int64 - start = time.Now() - logged = time.Now() - ) - for { - op, key, val, ok := iter.Next() - if !ok { - break - } - if err := rlp.Encode(writer, op); err != nil { - return err - } - if err := rlp.Encode(writer, key); err != nil { - return err - } - if err := rlp.Encode(writer, val); err != nil { - return err - } - if count%1000 == 0 { - // Check interruption emitted by ctrl+c - select { - case <-interrupt: - log.Info("Chain data exporting interrupted", "file", fn, - "kind", kind, "count", count, "elapsed", common.PrettyDuration(time.Since(start))) - return nil - default: - } - if time.Since(logged) > 8*time.Second { - log.Info("Exporting chain data", "file", fn, "kind", kind, - "count", count, "elapsed", common.PrettyDuration(time.Since(start))) - logged = time.Now() - } - } - count++ - } - log.Info("Exported chain data", "file", fn, "kind", kind, "count", count, - "elapsed", common.PrettyDuration(time.Since(start))) - return nil -}