feat: optimize epoch chain sync with specialized staged sync pipeline

Author: GheisMohammadi

api/service/synchronize/stagedstreamsync/default_stages.go

@@ -189,3 +189,29 @@ func DefaultStages(ctx context.Context,
 		},
 	}
 }
+
+func EpochStages(ctx context.Context,
+	seCfg StageEpochCfg,
+	finishCfg StageFinishCfg,
+) []*Stage {
+
+	handlerStageEpochSync := NewStageEpoch(seCfg)
+	handlerStageFinish := NewStageFinish(finishCfg)
+
+	return []*Stage{
+		{
+			ID:                 SyncEpoch,
+			Description:        "Sync only Last Block of Epoch",
+			Handler:            handlerStageEpochSync,
+			RangeMode:          OnlyShortRange,
+			ChainExecutionMode: OnlyEpochChain,
+		},
+		{
+			ID:                 Finish,
+			Description:        "Finalize Changes",
+			Handler:            handlerStageFinish,
+			RangeMode:          LongRangeAndShortRange,
+			ChainExecutionMode: AllChains,
+		},
+	}
+}

api/service/synchronize/stagedstreamsync/downloader.go

@@ -53,6 +53,7 @@ func NewDownloader(host p2p.Host,
 	setNodeSyncStatus func(bool)) *Downloader {
 
 	config.fixValues()
+	isEpochChain := !isBeaconNode && bc.ShardID() == shard.BeaconChainShardID
 
 	protoCfg := protocolConfig(host, bc, nodeConfig, isBeaconNode, config)
 	sp := streamSyncProtocol.NewProtocol(*protoCfg)
@@ -80,7 +81,13 @@ func NewDownloader(host p2p.Host,
 	ctx, cancel := context.WithCancel(context.Background())
 
 	// create an instance of staged sync for the downloader
-	stagedSyncInstance, err := CreateStagedSync(ctx, bc, nodeConfig, consensus, dbDir, sp, config, isBeaconNode, logger, setNodeSyncStatus)
+	var stagedSyncInstance *StagedStreamSync
+	var err error
+	if isEpochChain {
+		stagedSyncInstance, err = CreateStagedEpochSync(ctx, bc, nodeConfig, consensus, dbDir, sp, config, isBeaconNode, logger, setNodeSyncStatus)
+	} else {
+		stagedSyncInstance, err = CreateStagedSync(ctx, bc, nodeConfig, consensus, dbDir, sp, config, isBeaconNode, logger, setNodeSyncStatus)
+	}
 	if err != nil {
 		cancel()
 		return nil

api/service/synchronize/stagedstreamsync/syncing.go

@@ -151,7 +151,6 @@ func CreateStagedSync(ctx context.Context,
 		Str("SyncMode", config.SyncMode.String()).
 		Bool("serverOnly", config.ServerOnly).
 		Int("minStreams", config.MinStreams).
-		Str("dbDir", dbDir).
 		Msg(WrapStagedSyncMsg("staged stream sync created successfully"))
 
 	return New(
@@ -172,6 +171,94 @@ func CreateStagedSync(ctx context.Context,
 	), nil
 }
 
+// CreateStagedEpochSync creates an instance of staged sync for epoch chain
+func CreateStagedEpochSync(ctx context.Context,
+	bc core.BlockChain,
+	nodeConfig *nodeconfig.ConfigType,
+	consensus *consensus.Consensus,
+	dbDir string,
+	protocol syncProtocol,
+	config Config,
+	isBeaconNode bool,
+	logger zerolog.Logger,
+	setNodeSyncStatus func(bool),
+) (*StagedStreamSync, error) {
+
+	logger.Info().
+		Uint32("shard", bc.ShardID()).
+		Bool("isBeaconNode", isBeaconNode).
+		Bool("memdb", config.UseMemDB).
+		Str("dbDir", dbDir).
+		Bool("serverOnly", config.ServerOnly).
+		Int("minStreams", config.MinStreams).
+		Msg(WrapStagedSyncMsg("creating staged epoch sync"))
+
+	isExplorer := nodeConfig.Role() == nodeconfig.ExplorerNode
+	isValidator := nodeConfig.Role() == nodeconfig.Validator
+	isBeaconShard := true
+	isEpochChain := true
+	isBeaconValidator := false
+	joinConsensus := false
+
+	var mainDB kv.RwDB
+	if config.UseMemDB {
+		mdbPath := getEpochDbPath(dbDir)
+		logger.Info().
+			Str("path", mdbPath).
+			Msg(WrapStagedSyncMsg("creating epoch main db in memory"))
+		mainDB = mdbx.NewMDBX(log.New()).InMem(mdbPath).MustOpen()
+	} else {
+		mdbPath := getEpochDbPath(dbDir)
+		logger.Info().
+			Str("path", mdbPath).
+			Msg(WrapStagedSyncMsg("creating epoch main db in disk"))
+		mainDB = mdbx.NewMDBX(log.New()).Path(mdbPath).MustOpen()
+	}
+
+	stageSyncEpochCfg := NewStageEpochCfg(bc, nil, logger)
+	stageFinishCfg := NewStageFinishCfg(nil, logger)
+
+	// init stages order based on sync mode
+	initStagesOrder(config.SyncMode)
+
+	epochStages := EpochStages(ctx,
+		stageSyncEpochCfg,
+		stageFinishCfg,
+	)
+
+	logger.Info().
+		Uint32("shard", bc.ShardID()).
+		Bool("isEpochChain", isEpochChain).
+		Bool("isExplorer", isExplorer).
+		Bool("isValidator", isValidator).
+		Bool("isBeaconShard", isBeaconShard).
+		Bool("isBeaconValidator", isBeaconValidator).
+		Bool("joinConsensus", joinConsensus).
+		Bool("memdb", config.UseMemDB).
+		Str("dbDir", dbDir).
+		Str("SyncMode", config.SyncMode.String()).
+		Bool("serverOnly", config.ServerOnly).
+		Int("minStreams", config.MinStreams).
+		Msg(WrapStagedSyncMsg("staged stream epoch sync created successfully"))
+
+	return New(
+		bc,
+		consensus,
+		mainDB,
+		epochStages,
+		protocol,
+		isEpochChain,
+		isBeaconShard,
+		isBeaconValidator,
+		isExplorer,
+		isValidator,
+		joinConsensus,
+		config,
+		logger,
+		setNodeSyncStatus,
+	), nil
+}
+
 // initDB inits the sync loop main database and create buckets
 func initDB(ctx context.Context, mainDB kv.RwDB, dbs []kv.RwDB) error {
 
@@ -236,6 +323,11 @@ func getBlockDbPath(shardID uint32, beacon bool, workerID int, dbDir string) str
 	}
 }
 
+// getEpochDbPath returns the path of the cache database which stores epoch blocks
+func getEpochDbPath(dbDir string) string {
+	return filepath.Join(dbDir, "cache/epoch_db_main")
+}
+
 func (s *StagedStreamSync) Debug(source string, msg interface{}) {
 	// only log the msg in debug mode
 	if !s.config.DebugMode {