refactor staged stream sync to process fast sync and new block insertion

api/service/stagedstreamsync/adapter.go

@@ -31,6 +31,6 @@ type blockChain interface {
 	engine.ChainReader
 	Engine() engine.Engine
 
-	InsertChain(chain types.Blocks, verifyHeaders bool) (int, error)
+	InsertChain(chain types.Blocks, verifyHeaders bool, blockExecution bool) (int, error)
 	WriteCommitSig(blockNum uint64, lastCommits []byte) error
 }

api/service/stagedstreamsync/const.go

@@ -40,6 +40,16 @@ const (
 	ShortRangeTimeout time.Duration = 1 * time.Minute
 )
 
+// SyncMode represents the synchronization mode of the downloader.
+// It is a uint32 as it is used with atomic operations.
+type SyncMode uint32
+
+const (
+	FullSync SyncMode = iota // Synchronize the entire blockchain history from full blocks
+	FastSync                 // Download all blocks and states
+	SnapSync                 // Download the chain and the state via compact snapshots
+)
+
 type (
 	// Config is the downloader config
 	Config struct {
@@ -47,6 +57,9 @@ type (
 		// TODO: remove this when stream sync is fully up.
 		ServerOnly bool
 
+		// Synchronization mode of the downloader
+		SyncMode SyncMode
+
 		// parameters
 		Network              nodeconfig.NetworkType
 		Concurrency          int // Number of concurrent sync requests

api/service/stagedstreamsync/sig_verify.go

@@ -20,16 +20,16 @@ func (e *sigVerifyErr) Error() string {
 	return fmt.Sprintf("[VerifyHeaderSignature] %v", e.err.Error())
 }
 
-func verifyAndInsertBlocks(bc blockChain, blocks types.Blocks) (int, error) {
+func verifyAndInsertBlocks(bc blockChain, blocks types.Blocks, blockExecution bool) (int, error) {
 	for i, block := range blocks {
-		if err := verifyAndInsertBlock(bc, block, blocks[i+1:]...); err != nil {
+		if err := verifyAndInsertBlock(bc, block, blockExecution, blocks[i+1:]...); err != nil {
 			return i, err
 		}
 	}
 	return len(blocks), nil
 }
 
-func verifyAndInsertBlock(bc blockChain, block *types.Block, nextBlocks ...*types.Block) error {
+func verifyAndInsertBlock(bc blockChain, block *types.Block, blockExecution bool, nextBlocks ...*types.Block) error {
 	var (
 		sigBytes bls.SerializedSignature
 		bitmap   []byte

api/service/stagedstreamsync/stage_short_range.go

@@ -136,7 +136,9 @@ func (sr *StageShortRange) doShortRangeSync(ctx context.Context, s *StageState)
 		sh.streamsFailed(whitelist, "remote nodes cannot provide blocks with target hashes")
 	}
 
-	n, err := verifyAndInsertBlocks(sr.configs.bc, blocks)
+	utils.Logger().Info().Int("num blocks", len(blocks)).Msg("getBlockByHashes result")
+
+	n, err := verifyAndInsertBlocks(sr.configs.bc, blocks, true)
 	numBlocksInsertedShortRangeHistogramVec.With(s.state.promLabels()).Observe(float64(n))
 	if err != nil {
 		utils.Logger().Warn().Err(err).Int("blocks inserted", n).Msg("Insert block failed")

api/service/stagedstreamsync/stage_state.go

@@ -19,12 +19,13 @@ type StageStates struct {
 	configs StageStatesCfg
 }
 type StageStatesCfg struct {
-	bc          core.BlockChain
+	bc             core.BlockChain
-	db          kv.RwDB
+	db             kv.RwDB
-	blockDBs    []kv.RwDB
+	blockDBs       []kv.RwDB
-	concurrency int
+	concurrency    int
-	logger      zerolog.Logger
+	blockExecution bool
-	logProgress bool
+	logger         zerolog.Logger
+	logProgress    bool
 }
 
 func NewStageStates(cfg StageStatesCfg) *StageStates {
@@ -38,16 +39,18 @@ func NewStageStatesCfg(
 	db kv.RwDB,
 	blockDBs []kv.RwDB,
 	concurrency int,
+	blockExecution bool,
 	logger zerolog.Logger,
 	logProgress bool) StageStatesCfg {
 
 	return StageStatesCfg{
-		bc:          bc,
+		bc:             bc,
-		db:          db,
+		db:             db,
-		blockDBs:    blockDBs,
+		blockDBs:       blockDBs,
-		concurrency: concurrency,
+		concurrency:    concurrency,
-		logger:      logger,
+		blockExecution: blockExecution,
-		logProgress: logProgress,
+		logger:         logger,
+		logProgress:    logProgress,
 	}
 }
 
@@ -108,6 +111,8 @@ func (stg *StageStates) Exec(ctx context.Context, firstCycle bool, invalidBlockR
 		fmt.Print("\033[s") // save the cursor position
 	}
 
+	s.state.currentCycle.ReceiptHashes = make(map[uint64]common.Hash)
+
 	for i := currProgress + 1; i <= targetHeight; i++ {
 		blkKey := marshalData(i)
 		loopID, streamID := gbm.GetDownloadDetails(i)
@@ -157,7 +162,7 @@ func (stg *StageStates) Exec(ctx context.Context, firstCycle bool, invalidBlockR
 			return ErrInvalidBlockNumber
 		}
 
-		if err := verifyAndInsertBlock(stg.configs.bc, block); err != nil {
+		if err := verifyAndInsertBlock(stg.configs.bc, block, stg.configs.blockExecution); err != nil {
 			stg.configs.logger.Warn().Err(err).Uint64("cycle target block", targetHeight).
 				Uint64("block number", block.NumberU64()).
 				Msg(WrapStagedSyncMsg("insert blocks failed in long range"))
@@ -169,6 +174,10 @@ func (stg *StageStates) Exec(ctx context.Context, firstCycle bool, invalidBlockR
 			return err
 		}
 
+		// TODO: only for fast sync
+		// add receipt hash for next stage
+		s.state.currentCycle.ReceiptHashes[block.NumberU64()]=block.Header().ReceiptHash()
+
 		if invalidBlockRevert {
 			if s.state.invalidBlock.Number == i {
 				s.state.invalidBlock.resolve()

api/service/stagedstreamsync/staged_stream_sync.go

@@ -102,9 +102,10 @@ type Timing struct {
 }
 
 type SyncCycle struct {
-	Number       uint64
+	Number        uint64
-	TargetHeight uint64
+	TargetHeight  uint64
-	lock         sync.RWMutex
+	ReceiptHashes map[uint64]common.Hash
+	lock          sync.RWMutex
 }
 
 func (s *StagedStreamSync) Len() int                    { return len(s.stages) }

api/service/stagedstreamsync/syncing.go

@@ -81,14 +81,17 @@ func CreateStagedSync(ctx context.Context,
 		return nil, errInitDB
 	}
 
+	fastSync := config.SyncMode == FastSync
+
 	stageHeadsCfg := NewStageHeadersCfg(bc, mainDB)
 	stageShortRangeCfg := NewStageShortRangeCfg(bc, mainDB)
 	stageSyncEpochCfg := NewStageEpochCfg(bc, mainDB)
 
 	stageBodiesCfg := NewStageBodiesCfg(bc, mainDB, dbs, config.Concurrency, protocol, isBeaconNode, config.LogProgress)
-	stageStatesCfg := NewStageStatesCfg(bc, mainDB, dbs, config.Concurrency, logger, config.LogProgress)
+	stageStatesCfg := NewStageStatesCfg(bc, mainDB, dbs, config.Concurrency, !fastSync, logger, config.LogProgress)
 	stageStateSyncCfg := NewStageStateSyncCfg(bc, mainDB, config.Concurrency, protocol, logger, config.LogProgress)
-	lastMileCfg := NewStageLastMileCfg(ctx, bc, mainDB)
+	stageReceiptsCfg := NewStageReceiptsCfg(bc, mainDB, dbs, config.Concurrency, protocol, isBeaconNode, config.LogProgress)
+	lastMileCfg := NewStageLastMileCfg(ctx, bc, mainDB)	
 	stageFinishCfg := NewStageFinishCfg(mainDB)
 
 	stages := DefaultStages(ctx,
@@ -98,6 +101,7 @@ func CreateStagedSync(ctx context.Context,
 		stageBodiesCfg,
 		stageStateSyncCfg,
 		stageStatesCfg,
+		stageReceiptsCfg,
 		lastMileCfg,
 		stageFinishCfg,
 	)