[staking] build staking view from blockdao if indexer height behind

action/protocol/staking/protocol.go

@@ -74,6 +74,10 @@ type (
 		ReceiptStatus() uint64
 	}
 
+	ContractStakeViewBuilder interface {
+		Build(ctx context.Context, target uint64) (ContractStakeView, error)
+	}
+
 	// Protocol defines the protocol of handling staking
 	Protocol struct {
 		addr                     address.Address
@@ -85,6 +89,7 @@ type (
 		voteReviser              *VoteReviser
 		patch                    *PatchStore
 		helperCtx                HelperCtx
+		blockStore               BlockStore
 	}
 
 	// Configuration is the staking protocol configuration.
@@ -119,6 +124,13 @@ func WithContractStakingIndexerV3(indexer ContractStakingIndexer) Option {
 	}
 }
 
+// WithBlockStore sets the block store
+func WithBlockStore(bs BlockStore) Option {
+	return func(p *Protocol) {
+		p.blockStore = bs
+	}
+}
+
 // FindProtocol return a registered protocol from registry
 func FindProtocol(registry *protocol.Registry) *Protocol {
 	if registry == nil {
@@ -240,21 +252,21 @@ func (p *Protocol) Start(ctx context.Context, sr protocol.StateReader) (protocol
 
 	c.contractsStake = &contractStakeView{}
 	if p.contractStakingIndexer != nil {
-		view, err := p.contractStakingIndexer.StartView(ctx)
+		view, err := NewContractStakeViewBuilder(p.contractStakingIndexer, p.blockStore).Build(ctx, height)
 		if err != nil {
 			return nil, errors.Wrap(err, "failed to start contract staking indexer")
 		}
 		c.contractsStake.v1 = view
 	}
 	if p.contractStakingIndexerV2 != nil {
-		view, err := p.contractStakingIndexerV2.StartView(ctx)
+		view, err := NewContractStakeViewBuilder(p.contractStakingIndexerV2, p.blockStore).Build(ctx, height)
 		if err != nil {
 			return nil, errors.Wrap(err, "failed to start contract staking indexer v2")
 		}
 		c.contractsStake.v2 = view
 	}
 	if p.contractStakingIndexerV3 != nil {
-		view, err := p.contractStakingIndexerV3.StartView(ctx)
+		view, err := NewContractStakeViewBuilder(p.contractStakingIndexerV3, p.blockStore).Build(ctx, height)
 		if err != nil {
 			return nil, errors.Wrap(err, "failed to start contract staking indexer v3")
 		}

action/protocol/staking/protocol_test.go

@@ -463,6 +463,7 @@ func TestProtocol_ActiveCandidates(t *testing.T) {
 		return blkHeight, nil
 	}).AnyTimes()
 	csIndexer.EXPECT().StartView(gomock.Any()).Return(nil, nil)
+	csIndexer.EXPECT().Height().Return(uint64(blkHeight), nil).AnyTimes()
 
 	v, err := p.Start(ctx, sm)
 	require.NoError(err)

action/protocol/staking/stakeview_builder.go

@@ -0,0 +1,71 @@
+package staking
+
+import (
+	"context"
+
+	"github.com/pkg/errors"
+
+	"github.com/iotexproject/iotex-core/v2/action"
+	"github.com/iotexproject/iotex-core/v2/action/protocol"
+	"github.com/iotexproject/iotex-core/v2/blockchain/block"
+)
+
+type (
+	BlockStore interface {
+		GetReceipts(uint64) ([]*action.Receipt, error)
+		HeaderByHeight(height uint64) (*block.Header, error)
+	}
+
+	contractStakeViewBuilder struct {
+		indexer  ContractStakingIndexer
+		blockdao BlockStore
+	}
+)
+
+func NewContractStakeViewBuilder(
+	indexer ContractStakingIndexer,
+	blockdao BlockStore,
+) *contractStakeViewBuilder {
+	return &contractStakeViewBuilder{
+		indexer:  indexer,
+		blockdao: blockdao,
+	}
+}
+
+func (b *contractStakeViewBuilder) Build(ctx context.Context, height uint64) (ContractStakeView, error) {
+	view, err := b.indexer.StartView(ctx)
+	if err != nil {
+		return nil, err
+	}
+	indexerHeight, err := b.indexer.Height()
+	if err != nil {
+		return nil, err
+	}
+	if indexerHeight == height {
+		return view, nil
+	}
+	if indexerHeight > height {
+		return nil, errors.Errorf("indexer height %d is greater than requested height %d", indexerHeight, height)
+	}
+	if b.blockdao == nil {
+		return nil, errors.Errorf("blockdao is nil, cannot build view for height %d", height)
+	}
+	for h := indexerHeight + 1; h <= height; h++ {
+		receipts, err := b.blockdao.GetReceipts(h)
+		if err != nil {
+			return nil, errors.Wrapf(err, "failed to get receipts at height %d", h)
+		}
+		header, err := b.blockdao.HeaderByHeight(h)
+		if err != nil {
+			return nil, errors.Wrapf(err, "failed to get header at height %d", h)
+		}
+		ctx = protocol.WithBlockCtx(ctx, protocol.BlockCtx{
+			BlockHeight:    h,
+			BlockTimeStamp: header.Timestamp(),
+		})
+		if err = view.AddBlockReceipts(ctx, receipts); err != nil {
+			return nil, errors.Wrapf(err, "failed to build view with block at height %d", h)
+		}
+	}
+	return view, nil
+}

action/protocol/staking/viewdata.go

@@ -30,6 +30,7 @@ type (
 		Handle(ctx context.Context, receipt *action.Receipt) error
 		// BucketsByCandidate returns the buckets by candidate address
 		BucketsByCandidate(ownerAddr address.Address) ([]*VoteBucket, error)
+		AddBlockReceipts(ctx context.Context, receipts []*action.Receipt) error
 	}
 	// ViewData is the data that need to be stored in protocol's view
 	ViewData struct {

blockchain/blockdao/blockdao.go

@@ -129,7 +129,7 @@ func NewBlockDAOWithIndexersAndCache(blkStore BlockStore, indexers []BlockIndexe
 
 // Start starts block DAO and initiates the top height if it doesn't exist
 func (dao *blockDAO) Start(ctx context.Context) error {
-	err := dao.lifecycle.OnStart(ctx)
+	err := dao.lifecycle.OnStartSequentially(ctx)
 	if err != nil {
 		return errors.Wrap(err, "failed to start child services")
 	}

blockchain/blockdao/blockdao_test.go

@@ -83,7 +83,7 @@ func Test_blockDAO_Start(t *testing.T) {
 		p := gomonkey.NewPatches()
 		defer p.Reset()
 
-		p.ApplyMethodReturn(&lifecycle.Lifecycle{}, "OnStart", errors.New(t.Name()))
+		p.ApplyMethodReturn(&lifecycle.Lifecycle{}, "OnStartSequentially", errors.New(t.Name()))
 
 		err := blockdao.Start(context.Background())
 		r.ErrorContains(err, t.Name())
@@ -93,7 +93,7 @@ func Test_blockDAO_Start(t *testing.T) {
 		p := gomonkey.NewPatches()
 		defer p.Reset()
 
-		p.ApplyMethodReturn(&lifecycle.Lifecycle{}, "OnStart", nil)
+		p.ApplyMethodReturn(&lifecycle.Lifecycle{}, "OnStartSequentially", nil)
 		mockblockdao.EXPECT().Height().Return(uint64(0), errors.New(t.Name())).Times(1)
 
 		err := blockdao.Start(context.Background())
@@ -106,7 +106,7 @@ func Test_blockDAO_Start(t *testing.T) {
 
 		expectedHeight := uint64(1)
 
-		p.ApplyMethodReturn(&lifecycle.Lifecycle{}, "OnStart", nil)
+		p.ApplyMethodReturn(&lifecycle.Lifecycle{}, "OnStartSequentially", nil)
 		mockblockdao.EXPECT().Height().Return(expectedHeight, nil).Times(1)
 		p.ApplyPrivateMethod(&blockDAO{}, "checkIndexers", func(*blockDAO, context.Context) error { return nil })
 

blockindex/contractstaking/indexer.go

@@ -16,6 +16,7 @@ import (
 	"github.com/iotexproject/iotex-proto/golang/iotextypes"
 	"github.com/pkg/errors"
 
+	"github.com/iotexproject/iotex-core/v2/action"
 	"github.com/iotexproject/iotex-core/v2/action/protocol"
 	"github.com/iotexproject/iotex-core/v2/action/protocol/staking"
 	"github.com/iotexproject/iotex-core/v2/blockchain/block"
@@ -326,31 +327,38 @@ func (s *Indexer) PutBlock(ctx context.Context, blk *block.Block) error {
 	if blk.Height() > expectHeight {
 		return errors.Errorf("invalid block height %d, expect %d", blk.Height(), expectHeight)
 	}
+	handler, err := handleReceipts(ctx, blk.Height(), blk.Receipts, &s.config, cache)
+	if err != nil {
+		return errors.Wrapf(err, "failed to put block %d", blk.Height())
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	// commit the result
+	if err := s.commit(handler, blk.Height()); err != nil {
+		return errors.Wrapf(err, "failed to commit block %d", blk.Height())
+	}
+	return nil
+}
+
+func handleReceipts(ctx context.Context, height uint64, receipts []*action.Receipt, cfg *Config, cache stakingCache) (*contractStakingEventHandler, error) {
 	// new event handler for this block
 	handler := newContractStakingEventHandler(cache)
 
 	// handle events of block
-	for _, receipt := range blk.Receipts {
+	for _, receipt := range receipts {
 		if receipt.Status != uint64(iotextypes.ReceiptStatus_Success) {
 			continue
 		}
 		for _, log := range receipt.Logs() {
-			if log.Address != s.config.ContractAddress {
+			if log.Address != cfg.ContractAddress {
 				continue
 			}
-			if err := handler.HandleEvent(ctx, blk.Height(), log); err != nil {
-				return err
+			if err := handler.HandleEvent(ctx, height, log); err != nil {
+				return nil, err
 			}
 		}
 	}
-
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	// commit the result
-	if err := s.commit(handler, blk.Height()); err != nil {
-		return errors.Wrapf(err, "failed to commit block %d", blk.Height())
-	}
-	return nil
+	return handler, nil
 }
 
 func (s *Indexer) commit(handler *contractStakingEventHandler, height uint64) error {

blockindex/contractstaking/stakeview.go

@@ -5,10 +5,12 @@ import (
 	"time"
 
 	"github.com/iotexproject/iotex-address/address"
+	"github.com/iotexproject/iotex-proto/golang/iotextypes"
+	"github.com/pkg/errors"
+
 	"github.com/iotexproject/iotex-core/v2/action"
 	"github.com/iotexproject/iotex-core/v2/action/protocol"
 	"github.com/iotexproject/iotex-core/v2/action/protocol/staking"
-	"github.com/iotexproject/iotex-proto/golang/iotextypes"
 )
 
 type stakeView struct {
@@ -88,3 +90,27 @@ func (s *stakeView) Handle(ctx context.Context, receipt *action.Receipt) error {
 func (s *stakeView) Commit() {
 	s.cache = s.cache.Commit()
 }
+
+func (s *stakeView) AddBlockReceipts(ctx context.Context, receipts []*action.Receipt) error {
+	blkCtx := protocol.MustGetBlockCtx(ctx)
+	height := blkCtx.BlockHeight
+	expectHeight := s.height + 1
+	if expectHeight < s.helper.config.ContractDeployHeight {
+		expectHeight = s.helper.config.ContractDeployHeight
+	}
+	if height < expectHeight {
+		return nil
+	}
+	if height > expectHeight {
+		return errors.Errorf("invalid block height %d, expect %d", height, expectHeight)
+	}
+
+	handler, err := handleReceipts(ctx, height, receipts, &s.helper.config, s.cache)
+	if err != nil {
+		return err
+	}
+	_, delta := handler.Result()
+	s.cache = delta.Commit()
+	s.height = height
+	return nil
+}

chainservice/builder.go

@@ -685,6 +685,7 @@ func (builder *Builder) registerStakingProtocol() error {
 	if builder.cs.contractStakingIndexerV3 != nil {
 		opts = append(opts, staking.WithContractStakingIndexerV3(builder.cs.contractStakingIndexerV3))
 	}
+	opts = append(opts, staking.WithBlockStore(builder.cs.blockdao))
 	stakingProtocol, err := staking.NewProtocol(
 		staking.HelperCtx{
 			DepositGas:    rewarding.DepositGas,