build staking view from blockdao if indexer height behind

Author: envestcc

action/protocol/staking/protocol.go

@@ -74,17 +74,25 @@ 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
-		config                   Configuration
-		candBucketsIndexer       *CandidatesBucketsIndexer
-		contractStakingIndexer   ContractStakingIndexerWithBucketType
-		contractStakingIndexerV2 ContractStakingIndexer
-		contractStakingIndexerV3 ContractStakingIndexer
-		voteReviser              *VoteReviser
-		patch                    *PatchStore
-		helperCtx                HelperCtx
+		addr                         address.Address
+		config                       Configuration
+		candBucketsIndexer           *CandidatesBucketsIndexer
+		contractStakingIndexer       ContractStakingIndexerWithBucketType
+		contractStakingIndexerV2     ContractStakingIndexer
+		contractStakingIndexerV3     ContractStakingIndexer
+		voteReviser                  *VoteReviser
+		patch                        *PatchStore
+		helperCtx                    HelperCtx
+		contractStakingViewBuilder   ContractStakeViewBuilder
+		contractStakingViewV2Builder ContractStakeViewBuilder
+		contractStakingViewV3Builder ContractStakeViewBuilder
+		blockStore                   BlockStore
 	}
 
 	// Configuration is the staking protocol configuration.
@@ -118,6 +126,12 @@ func WithContractStakingIndexerV3(indexer ContractStakingIndexer) Option {
 	}
 }
 
+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 {
@@ -196,6 +210,15 @@ func NewProtocol(
 	for _, opt := range opts {
 		opt(p)
 	}
+	if p.contractStakingIndexer != nil {
+		p.contractStakingViewBuilder = NewContractStakeViewBuilder(p.contractStakingIndexer, p.blockStore)
+	}
+	if p.contractStakingIndexerV2 != nil {
+		p.contractStakingViewV2Builder = NewContractStakeViewBuilder(p.contractStakingIndexerV2, p.blockStore)
+	}
+	if p.contractStakingIndexerV3 != nil {
+		p.contractStakingViewV3Builder = NewContractStakeViewBuilder(p.contractStakingIndexerV3, p.blockStore)
+	}
 	return p, nil
 }
 
@@ -232,22 +255,22 @@ func (p *Protocol) Start(ctx context.Context, sr protocol.StateReader) (protocol
 	}
 
 	c.contractsStake = &contractStakeView{}
-	if p.contractStakingIndexer != nil {
-		view, err := p.contractStakingIndexer.StartView(ctx)
+	if p.contractStakingViewBuilder != nil {
+		view, err := p.contractStakingViewBuilder.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)
+	if p.contractStakingViewV2Builder != nil {
+		view, err := p.contractStakingViewV2Builder.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)
+	if p.contractStakingViewV3Builder != nil {
+		view, err := p.contractStakingViewV3Builder.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(0), nil).AnyTimes()
 
 	v, err := p.Start(ctx, sm)
 	require.NoError(err)

action/protocol/staking/stakeview_builder.go

@@ -0,0 +1,68 @@
+package staking
+
+import (
+	"context"
+
+	"github.com/pkg/errors"
+
+	"github.com/iotexproject/iotex-core/v2/action"
+	"github.com/iotexproject/iotex-core/v2/blockchain/block"
+)
+
+type (
+	BlockStore interface {
+		GetBlockByHeight(uint64) (*block.Block, error)
+		GetReceipts(uint64) ([]*action.Receipt, 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
+	} else 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++ {
+		blk, err := b.blockdao.GetBlockByHeight(h)
+		if err != nil {
+			return nil, errors.Wrapf(err, "failed to get block at height %d", h)
+		}
+		if blk.Receipts == nil {
+			receipts, err := b.blockdao.GetReceipts(h)
+			if err != nil {
+				return nil, errors.Wrapf(err, "failed to get receipts at height %d", h)
+			}
+			blk.Receipts = receipts
+		}
+		if err = view.BuildWithBlock(ctx, blk); 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

@@ -14,6 +14,7 @@ import (
 
 	"github.com/iotexproject/iotex-core/v2/action"
 	"github.com/iotexproject/iotex-core/v2/action/protocol"
+	"github.com/iotexproject/iotex-core/v2/blockchain/block"
 )
 
 type (
@@ -24,6 +25,7 @@ type (
 		Handle(ctx context.Context, receipt *action.Receipt) error
 		Commit()
 		BucketsByCandidate(ownerAddr address.Address) ([]*VoteBucket, error)
+		BuildWithBlock(ctx context.Context, blk *block.Block) error
 	}
 	// ViewData is the data that need to be stored in protocol's view
 	ViewData struct {

blockindex/contractstaking/indexer.go

@@ -204,26 +204,32 @@ 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 := handleBlock(ctx, blk, &s.config, s.cache)
+	if err != nil {
+		return errors.Wrapf(err, "failed to put block %d", blk.Height())
+	}
+	return s.commit(handler, blk.Height())
+}
+
+func handleBlock(ctx context.Context, blk *block.Block, cfg *Config, cache *contractStakingCache) (*contractStakingEventHandler, error) {
 	// new event handler for this block
-	handler := newContractStakingEventHandler(s.cache)
+	handler := newContractStakingEventHandler(cache)
 
 	// handle events of block
 	for _, receipt := range blk.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
+				return handler, err
 			}
 		}
 	}
-
-	// commit the result
-	return s.commit(handler, blk.Height())
+	return handler, nil
 }
 
 func (s *Indexer) commit(handler *contractStakingEventHandler, height uint64) error {

blockindex/contractstaking/stakeview.go

@@ -6,10 +6,12 @@ import (
 
 	"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-core/v2/blockchain/block"
 )
 
 type stakeView struct {
@@ -95,3 +97,25 @@ func (s *stakeView) Commit() {
 		s.dirty = nil
 	}
 }
+
+func (s *stakeView) BuildWithBlock(ctx context.Context, blk *block.Block) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	expectHeight := s.clean.Height() + 1
+	if expectHeight < s.helper.config.ContractDeployHeight {
+		expectHeight = s.helper.config.ContractDeployHeight
+	}
+	if blk.Height() < expectHeight {
+		return nil
+	}
+	if blk.Height() > expectHeight {
+		return errors.Errorf("invalid block height %d, expect %d", blk.Height(), expectHeight)
+	}
+
+	handler, err := handleBlock(ctx, blk, &s.helper.config, s.clean)
+	if err != nil {
+		return err
+	}
+	_, delta := handler.Result()
+	return s.clean.Merge(delta, blk.Height())
+}

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,

systemcontractindex/stakingindex/stakeview.go

@@ -5,10 +5,13 @@ import (
 	"sync"
 
 	"github.com/iotexproject/iotex-address/address"
+	"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"
+	"github.com/iotexproject/iotex-core/v2/blockchain/genesis"
 )
 
 type stakeView struct {
@@ -63,3 +66,33 @@ func (s *stakeView) Handle(ctx context.Context, receipt *action.Receipt) error {
 }
 
 func (s *stakeView) Commit() {}
+
+func (s *stakeView) BuildWithBlock(ctx context.Context, blk *block.Block) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if blk.Height() != s.height+1 {
+		return errors.Errorf("block height %d does not match stake view height %d", blk.Height(), s.height+1)
+	}
+	g, ok := genesis.ExtractGenesisContext(ctx)
+	if !ok {
+		return errors.New("failed to extract genesis context")
+	}
+	blkCtx := protocol.BlockCtx{
+		BlockHeight:    blk.Height(),
+		BlockTimeStamp: blk.Timestamp(),
+		GasLimit:       g.BlockGasLimitByHeight(blk.Height()),
+		Producer:       blk.PublicKey().Address(),
+		BaseFee:        blk.BaseFee(),
+		ExcessBlobGas:  blk.BlobGasUsed(),
+	}
+	ctx = protocol.WithBlockCtx(ctx, blkCtx)
+	muted := s.helper.muteHeight > 0 && blk.Height() >= s.helper.muteHeight
+	handler := newEventHandler(s.helper.bucketNS, s.cache, blkCtx, s.helper.timestamped, muted)
+	for _, receipt := range blk.Receipts {
+		if err := s.helper.handleReceipt(ctx, handler, receipt); err != nil {
+			return errors.Wrapf(err, "failed to handle receipt at height %d", blk.Height())
+		}
+	}
+	s.height = blk.Height()
+	return nil
+}