Genesis sync from existing snapshots (#3087)

Author: AskAlexSharov

cmd/hack/hack.go

@@ -2589,58 +2589,102 @@ func recsplitWholeChain(chaindata string) error {
 
 	log.Info("Last body number", "last", last)
 	for i := uint64(*block); i < last; i += blocksPerFile {
-		fileName := snapshotsync.FileName(i, i+blocksPerFile, snapshotsync.Transactions)
-		segmentFile := path.Join(snapshotDir, fileName) + ".seg"
+		fileName := snapshotsync.FileName(i, i+blocksPerFile, snapshotsync.Bodies)
+
 		log.Info("Creating", "file", fileName+".seg")
 		db := mdbx.MustOpen(chaindata)
-		firstTxID, err := snapshotsync.DumpTxs(db, "", i, int(blocksPerFile))
-		if err != nil {
+		if err := snapshotsync.DumpBodies(db, "", i, int(blocksPerFile)); err != nil {
 			panic(err)
 		}
 		db.Close()
+		segmentFile := path.Join(snapshotDir, fileName) + ".seg"
 		if err := compress1(chaindata, fileName, segmentFile); err != nil {
 			panic(err)
 		}
-		if err := snapshotsync.TransactionsHashIdx(*chainID, firstTxID, segmentFile); err != nil {
-			panic(err)
-		}
+		//if err := snapshotsync.BodiesIdx(segmentFile, i); err != nil {
+		//	panic(err)
+		//}
 		_ = os.Remove(fileName + ".dat")
 
 		fileName = snapshotsync.FileName(i, i+blocksPerFile, snapshotsync.Headers)
-		segmentFile = path.Join(snapshotDir, fileName) + ".seg"
 		log.Info("Creating", "file", fileName+".seg")
 		db = mdbx.MustOpen(chaindata)
 		if err := snapshotsync.DumpHeaders(db, "", i, int(blocksPerFile)); err != nil {
 			panic(err)
 		}
 		db.Close()
+		segmentFile = path.Join(snapshotDir, fileName) + ".seg"
 		if err := compress1(chaindata, fileName, segmentFile); err != nil {
 			panic(err)
 		}
-
-		if err := snapshotsync.HeadersHashIdx(segmentFile, i); err != nil {
-			panic(err)
-		}
+		//if err := snapshotsync.HeadersHashIdx(segmentFile, i); err != nil {
+		//	panic(err)
+		//}
 		_ = os.Remove(fileName + ".dat")
 
-		fileName = snapshotsync.FileName(i, i+blocksPerFile, snapshotsync.Bodies)
-		segmentFile = path.Join(snapshotDir, fileName) + ".seg"
+		fileName = snapshotsync.FileName(i, i+blocksPerFile, snapshotsync.Transactions)
 		log.Info("Creating", "file", fileName+".seg")
 		db = mdbx.MustOpen(chaindata)
-		if err := snapshotsync.DumpBodies(db, "", i, int(blocksPerFile)); err != nil {
+		firstTxID, err := snapshotsync.DumpTxs(db, "", i, int(blocksPerFile))
+		if err != nil {
 			panic(err)
 		}
 		db.Close()
+		segmentFile = path.Join(snapshotDir, fileName) + ".seg"
 		if err := compress1(chaindata, fileName, segmentFile); err != nil {
 			panic(err)
 		}
-		if err := snapshotsync.BodiesIdx(segmentFile, i); err != nil {
-			panic(err)
-		}
+		_ = firstTxID
+		//if err := snapshotsync.TransactionsHashIdx(*chainID, firstTxID, segmentFile); err != nil {
+		//	panic(err)
+		//}
 		_ = os.Remove(fileName + ".dat")
 
 		//nolint
-		break // TODO: remove me - useful for tests
+		//break // TODO: remove me - useful for tests
+	}
+	return nil
+}
+
+func checkBlockSnapshot(chaindata string) error {
+	database := mdbx.MustOpen(chaindata)
+	defer database.Close()
+	dataDir := path.Dir(chaindata)
+	chainConfig := tool.ChainConfigFromDB(database)
+	chainID, _ := uint256.FromBig(chainConfig.ChainID)
+	_ = chainID
+
+	snapshots := snapshotsync.NewAllSnapshots(path.Join(dataDir, "snapshots"), params.KnownSnapshots(chainConfig.ChainName))
+	snapshots.ReopenSegments()
+	snapshots.ReopenIndices()
+	//if err := snapshots.BuildIndices(context.Background(), *chainID); err != nil {
+	//	panic(err)
+	//}
+
+	snBlockReader := snapshotsync.NewBlockReaderWithSnapshots(snapshots)
+	tx, err := database.BeginRo(context.Background())
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+
+	for i := uint64(0); i < snapshots.BlocksAvailable(); i++ {
+		hash, err := rawdb.ReadCanonicalHash(tx, i)
+		if err != nil {
+			return err
+		}
+		blockFromDB := rawdb.ReadBlock(tx, hash, i)
+		blockFromSnapshot, _, err := snBlockReader.BlockWithSenders(context.Background(), tx, hash, i)
+		if err != nil {
+			return err
+		}
+
+		if blockFromSnapshot.Hash() != blockFromDB.Hash() {
+			panic(i)
+		}
+		if i%1_000 == 0 {
+			log.Info(fmt.Sprintf("Block Num: %dK", i/1_000))
+		}
 	}
 	return nil
 }
@@ -3928,6 +3972,8 @@ func main() {
 		err = compress1(*chaindata, *name, *name)
 	case "recsplitWholeChain":
 		err = recsplitWholeChain(*chaindata)
+	case "checkBlockSnapshot":
+		err = checkBlockSnapshot(*chaindata)
 	case "decompress":
 		err = decompress(*name)
 	case "genstate":

cmd/integration/commands/snapshot_check.go

@@ -459,7 +459,7 @@ func stageBodies(db kv.RwDB, ctx context.Context) error {
 			}
 
 			u := sync.NewUnwindState(stages.Senders, s.BlockNumber-unwind, s.BlockNumber)
-			if err := stagedsync.UnwindBodiesStage(u, tx, stagedsync.StageBodiesCfg(db, nil, nil, nil, nil, 0, *chainConfig, 0), ctx); err != nil {
+			if err := stagedsync.UnwindBodiesStage(u, tx, stagedsync.StageBodiesCfg(db, nil, nil, nil, nil, 0, *chainConfig, 0, allSnapshots(chainConfig), getBlockReader(chainConfig)), ctx); err != nil {
 				return err
 			}
 
@@ -538,7 +538,7 @@ func stageSenders(db kv.RwDB, ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	cfg := stagedsync.StageSendersCfg(db, chainConfig, tmpdir, pm)
+	cfg := stagedsync.StageSendersCfg(db, chainConfig, tmpdir, pm, allSnapshots(chainConfig))
 	if unwind > 0 {
 		u := sync.NewUnwindState(stages.Senders, s.BlockNumber-unwind, s.BlockNumber)
 		err = stagedsync.UnwindSendersStage(u, tx, cfg, ctx)
@@ -558,6 +558,7 @@ func stageSenders(db kv.RwDB, ctx context.Context) error {
 func stageExec(db kv.RwDB, ctx context.Context) error {
 	pm, engine, chainConfig, vmConfig, sync, _, _ := newSync(ctx, db, nil)
 	must(sync.SetCurrentStage(stages.Execution))
+	tmpdir := path.Join(datadir, etl.TmpDirName)
 
 	if reset {
 		genesis, _ := byChain()
@@ -585,7 +586,7 @@ func stageExec(db kv.RwDB, ctx context.Context) error {
 		pm.TxIndex = prune.Distance(s.BlockNumber - pruneTo)
 	}
 
-	cfg := stagedsync.StageExecuteBlocksCfg(db, pm, batchSize, nil, chainConfig, engine, vmConfig, nil, false, tmpDBPath, getBlockReader())
+	cfg := stagedsync.StageExecuteBlocksCfg(db, pm, batchSize, nil, chainConfig, engine, vmConfig, nil, false, tmpdir, getBlockReader(chainConfig))
 	if unwind > 0 {
 		u := sync.NewUnwindState(stages.Execution, s.BlockNumber-unwind, s.BlockNumber)
 		err := stagedsync.UnwindExecutionStage(u, s, nil, ctx, cfg, false)
@@ -615,7 +616,7 @@ func stageExec(db kv.RwDB, ctx context.Context) error {
 }
 
 func stageTrie(db kv.RwDB, ctx context.Context) error {
-	pm, _, _, _, sync, _, _ := newSync(ctx, db, nil)
+	pm, _, chainConfig, _, sync, _, _ := newSync(ctx, db, nil)
 	must(sync.SetCurrentStage(stages.IntermediateHashes))
 	tmpdir := path.Join(datadir, etl.TmpDirName)
 
@@ -644,7 +645,7 @@ func stageTrie(db kv.RwDB, ctx context.Context) error {
 
 	log.Info("StageExec", "progress", execStage.BlockNumber)
 	log.Info("StageTrie", "progress", s.BlockNumber)
-	cfg := stagedsync.StageTrieCfg(db, true, true, tmpdir, getBlockReader())
+	cfg := stagedsync.StageTrieCfg(db, true, true, tmpdir, getBlockReader(chainConfig))
 	if unwind > 0 {
 		u := sync.NewUnwindState(stages.IntermediateHashes, s.BlockNumber-unwind, s.BlockNumber)
 		if err := stagedsync.UnwindIntermediateHashesStage(u, s, tx, cfg, ctx); err != nil {
@@ -911,7 +912,7 @@ func stageHistory(db kv.RwDB, ctx context.Context) error {
 func stageTxLookup(db kv.RwDB, ctx context.Context) error {
 	tmpdir := path.Join(datadir, etl.TmpDirName)
 
-	pm, _, _, _, sync, _, _ := newSync(ctx, db, nil)
+	pm, _, chainConfig, _, sync, _, _ := newSync(ctx, db, nil)
 	must(sync.SetCurrentStage(stages.TxLookup))
 
 	tx, err := db.BeginRw(ctx)
@@ -936,7 +937,7 @@ func stageTxLookup(db kv.RwDB, ctx context.Context) error {
 	}
 	log.Info("Stage", "name", s.ID, "progress", s.BlockNumber)
 
-	cfg := stagedsync.StageTxLookupCfg(db, pm, tmpdir)
+	cfg := stagedsync.StageTxLookupCfg(db, pm, tmpdir, allSnapshots(chainConfig))
 	if unwind > 0 {
 		u := sync.NewUnwindState(stages.TxLookup, s.BlockNumber-unwind, s.BlockNumber)
 		err = stagedsync.UnwindTxLookup(u, s, tx, cfg, ctx)
@@ -1022,21 +1023,28 @@ func byChain() (*core.Genesis, *params.ChainConfig) {
 var openSnapshotOnce sync.Once
 var _allSnapshotsSingleton *snapshotsync.AllSnapshots
 
-func allSnapshots() *snapshotsync.AllSnapshots {
+func allSnapshots(cc *params.ChainConfig) *snapshotsync.AllSnapshots {
 	openSnapshotOnce.Do(func() {
-		snapshotCfg := ethconfig.Snapshot{}
 		if enableSnapshot {
-			snapshotCfg.Enabled = true
-			snapshotCfg.Dir = path.Join(datadir, "snapshots")
-			_allSnapshotsSingleton = snapshotsync.MustOpenAll(snapshotCfg.Dir)
+			snapshotCfg := ethconfig.Snapshot{
+				Enabled: true,
+				Dir:     path.Join(datadir, "snapshots"),
+			}
+			_allSnapshotsSingleton = snapshotsync.NewAllSnapshots(snapshotCfg.Dir, params.KnownSnapshots(cc.ChainName))
+			if err := _allSnapshotsSingleton.ReopenSegments(); err != nil {
+				panic(err)
+			}
+			if err := _allSnapshotsSingleton.ReopenIndices(); err != nil {
+				panic(err)
+			}
 		}
 	})
 	return _allSnapshotsSingleton
 }
 
-func getBlockReader() (blockReader interfaces.FullBlockReader) {
+func getBlockReader(cc *params.ChainConfig) (blockReader interfaces.FullBlockReader) {
 	blockReader = snapshotsync.NewBlockReader()
-	if sn := allSnapshots(); sn != nil {
+	if sn := allSnapshots(cc); sn != nil {
 		blockReader = snapshotsync.NewBlockReaderWithSnapshots(sn)
 	}
 	return blockReader
@@ -1110,7 +1118,7 @@ func newSync(ctx context.Context, db kv.RwDB, miningConfig *params.MiningConfig)
 			stagedsync.StageMiningCreateBlockCfg(db, miner, *chainConfig, engine, nil, nil, tmpdir),
 			stagedsync.StageMiningExecCfg(db, miner, events, *chainConfig, engine, &vm.Config{}, tmpdir),
 			stagedsync.StageHashStateCfg(db, tmpdir),
-			stagedsync.StageTrieCfg(db, false, true, tmpdir, getBlockReader()),
+			stagedsync.StageTrieCfg(db, false, true, tmpdir, getBlockReader(chainConfig)),
 			stagedsync.StageMiningFinishCfg(db, *chainConfig, engine, miner, ctx.Done()),
 		),
 		stagedsync.MiningUnwindOrder,

cmd/integration/commands/stages.go

@@ -180,7 +180,7 @@ func syncBySmallSteps(db kv.RwDB, miningConfig params.MiningConfig, ctx context.
 	stateStages.DisableStages(stages.Headers, stages.BlockHashes, stages.Bodies, stages.Senders,
 		stages.Finish)
 
-	execCfg := stagedsync.StageExecuteBlocksCfg(db, pm, batchSize, changeSetHook, chainConfig, engine, vmConfig, nil, false, tmpDir, getBlockReader())
+	execCfg := stagedsync.StageExecuteBlocksCfg(db, pm, batchSize, changeSetHook, chainConfig, engine, vmConfig, nil, false, tmpDir, getBlockReader(chainConfig))
 
 	execUntilFunc := func(execToBlock uint64) func(firstCycle bool, badBlockUnwind bool, stageState *stagedsync.StageState, unwinder stagedsync.Unwinder, tx kv.RwTx) error {
 		return func(firstCycle bool, badBlockUnwind bool, s *stagedsync.StageState, unwinder stagedsync.Unwinder, tx kv.RwTx) error {
@@ -404,7 +404,7 @@ func checkMinedBlock(b1, b2 *types.Block, chainConfig *params.ChainConfig) {
 }
 
 func loopIh(db kv.RwDB, ctx context.Context, unwind uint64) error {
-	_, _, _, _, sync, _, _ := newSync(ctx, db, nil)
+	_, _, chainConfig, _, sync, _, _ := newSync(ctx, db, nil)
 	tmpdir := path.Join(datadir, etl.TmpDirName)
 	tx, err := db.BeginRw(ctx)
 	if err != nil {
@@ -425,7 +425,7 @@ func loopIh(db kv.RwDB, ctx context.Context, unwind uint64) error {
 	}
 	_ = sync.SetCurrentStage(stages.IntermediateHashes)
 	u = &stagedsync.UnwindState{ID: stages.IntermediateHashes, UnwindPoint: to}
-	if err = stagedsync.UnwindIntermediateHashesStage(u, stage(sync, tx, nil, stages.IntermediateHashes), tx, stagedsync.StageTrieCfg(db, true, true, tmpdir, getBlockReader()), ctx); err != nil {
+	if err = stagedsync.UnwindIntermediateHashesStage(u, stage(sync, tx, nil, stages.IntermediateHashes), tx, stagedsync.StageTrieCfg(db, true, true, tmpdir, getBlockReader(chainConfig)), ctx); err != nil {
 		return err
 	}
 	must(tx.Commit())
@@ -470,6 +470,7 @@ func loopIh(db kv.RwDB, ctx context.Context, unwind uint64) error {
 
 func loopExec(db kv.RwDB, ctx context.Context, unwind uint64) error {
 	pm, engine, chainConfig, vmConfig, sync, _, _ := newSync(ctx, db, nil)
+	tmpdir := path.Join(datadir, etl.TmpDirName)
 
 	tx, err := db.BeginRw(ctx)
 	if err != nil {
@@ -489,7 +490,7 @@ func loopExec(db kv.RwDB, ctx context.Context, unwind uint64) error {
 	from := progress(tx, stages.Execution)
 	to := from + unwind
 
-	cfg := stagedsync.StageExecuteBlocksCfg(db, pm, batchSize, nil, chainConfig, engine, vmConfig, nil, false, tmpDBPath, getBlockReader())
+	cfg := stagedsync.StageExecuteBlocksCfg(db, pm, batchSize, nil, chainConfig, engine, vmConfig, nil, false, tmpdir, getBlockReader(chainConfig))
 
 	// set block limit of execute stage
 	sync.MockExecFunc(stages.Execution, func(firstCycle bool, badBlockUnwind bool, stageState *stagedsync.StageState, unwinder stagedsync.Unwinder, tx kv.RwTx) error {

cmd/integration/commands/state_stages.go

@@ -64,11 +64,11 @@ The daemon should respond with something like:
 INFO [date-time] HTTP endpoint opened url=localhost:8545...
 ```
 
-When RPC daemon runs remotely, by default it maintains a state cache, which is updated every time when Erigon
-imports a new block. When state cache is reasonably warm, it allows such remote RPC daemon to execute queries
-related to `latest` block (i.e. to current state) with comparable performance to a local RPC daemon
-(around 2x slower vs 10x slower without state cache). Since there can be multiple such RPC daemons per one
-Erigon node, it may scale well for some workloads that are heavy on the current state queries.
+When RPC daemon runs remotely, by default it maintains a state cache, which is updated every time when Erigon imports a
+new block. When state cache is reasonably warm, it allows such remote RPC daemon to execute queries related to `latest`
+block (i.e. to current state) with comparable performance to a local RPC daemon
+(around 2x slower vs 10x slower without state cache). Since there can be multiple such RPC daemons per one Erigon node,
+it may scale well for some workloads that are heavy on the current state queries.
 
 ### Healthcheck
 

cmd/rpcdaemon/README.md

@@ -23,9 +23,11 @@ import (
 	"github.com/ledgerwatch/erigon/cmd/rpcdaemon/services"
 	"github.com/ledgerwatch/erigon/cmd/utils"
 	"github.com/ledgerwatch/erigon/common/paths"
+	"github.com/ledgerwatch/erigon/core/rawdb"
 	"github.com/ledgerwatch/erigon/eth/ethconfig"
 	"github.com/ledgerwatch/erigon/internal/debug"
 	"github.com/ledgerwatch/erigon/node"
+	"github.com/ledgerwatch/erigon/params"
 	"github.com/ledgerwatch/erigon/rpc"
 	"github.com/ledgerwatch/erigon/turbo/snapshotsync"
 	"github.com/ledgerwatch/log/v3"
@@ -246,7 +248,25 @@ func RemoteServices(ctx context.Context, cfg Flags, logger log.Logger, rootCance
 
 	if cfg.SingleNodeMode {
 		if cfg.Snapshot.Enabled {
-			allSnapshots, err := snapshotsync.OpenAll(cfg.Snapshot.Dir)
+			var cc *params.ChainConfig
+			if err := db.View(context.Background(), func(tx kv.Tx) error {
+				genesisBlock, err := rawdb.ReadBlockByNumber(tx, 0)
+				if err != nil {
+					return err
+				}
+				cc, err = rawdb.ReadChainConfig(tx, genesisBlock.Hash())
+				if err != nil {
+					return err
+				}
+				return nil
+			}); err != nil {
+				return nil, nil, nil, nil, nil, nil, err
+			}
+			if cc == nil {
+				return nil, nil, nil, nil, nil, nil, fmt.Errorf("chain config not found in db. Need start erigon at least once on this db")
+			}
+
+			allSnapshots := snapshotsync.NewAllSnapshots(cfg.Snapshot.Dir, params.KnownSnapshots(cc.ChainName))
 			if err != nil {
 				return nil, nil, nil, nil, nil, nil, err
 			}

cmd/rpcdaemon/cli/config.go

@@ -12,8 +12,12 @@ type BlockReader interface {
 	BlockWithSenders(ctx context.Context, tx kv.Tx, hash common.Hash, blockHeight uint64) (block *types.Block, senders []common.Address, err error)
 }
 
+type HeaderReader interface {
+	Header(ctx context.Context, tx kv.Getter, hash common.Hash, blockHeight uint64) (*types.Header, error)
+	HeaderByNumber(ctx context.Context, tx kv.Getter, blockHeight uint64) (*types.Header, error)
+}
+
 type FullBlockReader interface {
 	BlockReader
-	Header(ctx context.Context, tx kv.Tx, hash common.Hash, blockHeight uint64) (*types.Header, error)
-	HeaderByNumber(ctx context.Context, tx kv.Tx, blockHeight uint64) (*types.Header, error)
+	HeaderReader
 }