Payload ID caching (#10481)

beacon-chain/blockchain/BUILD.bazel

@@ -48,6 +48,7 @@ go_library(
         "//beacon-chain/core/transition:go_default_library",
         "//beacon-chain/db:go_default_library",
         "//beacon-chain/db/filters:go_default_library",
+        "//beacon-chain/db/kv:go_default_library",
         "//beacon-chain/forkchoice:go_default_library",
         "//beacon-chain/forkchoice/doubly-linked-tree:go_default_library",
         "//beacon-chain/forkchoice/protoarray:go_default_library",

beacon-chain/blockchain/chain_info_norace_test.go

@@ -19,10 +19,11 @@ func TestHeadSlot_DataRace(t *testing.T) {
 	}
 	b, err := wrapper.WrappedSignedBeaconBlock(util.NewBeaconBlock())
 	require.NoError(t, err)
+	st, _ := util.DeterministicGenesisState(t, 1)
 	wait := make(chan struct{})
 	go func() {
 		defer close(wait)
-		require.NoError(t, s.saveHead(context.Background(), [32]byte{}, b))
+		require.NoError(t, s.saveHead(context.Background(), [32]byte{}, b, st))
 	}()
 	s.HeadSlot()
 	<-wait
@@ -37,9 +38,11 @@ func TestHeadRoot_DataRace(t *testing.T) {
 	b, err := wrapper.WrappedSignedBeaconBlock(util.NewBeaconBlock())
 	require.NoError(t, err)
 	wait := make(chan struct{})
+	st, _ := util.DeterministicGenesisState(t, 1)
 	go func() {
 		defer close(wait)
-		require.NoError(t, s.saveHead(context.Background(), [32]byte{}, b))
+		require.NoError(t, s.saveHead(context.Background(), [32]byte{}, b, st))
+
 	}()
 	_, err = s.HeadRoot(context.Background())
 	require.NoError(t, err)
@@ -57,9 +60,11 @@ func TestHeadBlock_DataRace(t *testing.T) {
 	b, err := wrapper.WrappedSignedBeaconBlock(util.NewBeaconBlock())
 	require.NoError(t, err)
 	wait := make(chan struct{})
+	st, _ := util.DeterministicGenesisState(t, 1)
 	go func() {
 		defer close(wait)
-		require.NoError(t, s.saveHead(context.Background(), [32]byte{}, b))
+		require.NoError(t, s.saveHead(context.Background(), [32]byte{}, b, st))
+
 	}()
 	_, err = s.HeadBlock(context.Background())
 	require.NoError(t, err)
@@ -74,9 +79,11 @@ func TestHeadState_DataRace(t *testing.T) {
 	b, err := wrapper.WrappedSignedBeaconBlock(util.NewBeaconBlock())
 	require.NoError(t, err)
 	wait := make(chan struct{})
+	st, _ := util.DeterministicGenesisState(t, 1)
 	go func() {
 		defer close(wait)
-		require.NoError(t, s.saveHead(context.Background(), [32]byte{}, b))
+		require.NoError(t, s.saveHead(context.Background(), [32]byte{}, b, st))
+
 	}()
 	_, err = s.HeadState(context.Background())
 	require.NoError(t, err)

beacon-chain/blockchain/head.go

@@ -44,7 +44,11 @@ func (s *Service) UpdateAndSaveHeadWithBalances(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	return s.saveHead(ctx, headRoot, headBlock)
+	headState, err := s.cfg.StateGen.StateByRoot(ctx, headRoot)
+	if err != nil {
+		return errors.Wrap(err, "could not retrieve head state in DB")
+	}
+	return s.saveHead(ctx, headRoot, headBlock, headState)
 }
 
 // This defines the current chain service's view of head.
@@ -101,7 +105,7 @@ func (s *Service) updateHead(ctx context.Context, balances []uint64) ([32]byte,
 
 // This saves head info to the local service cache, it also saves the
 // new head root to the DB.
-func (s *Service) saveHead(ctx context.Context, headRoot [32]byte, headBlock block.SignedBeaconBlock) error {
+func (s *Service) saveHead(ctx context.Context, headRoot [32]byte, headBlock block.SignedBeaconBlock, headState state.BeaconState) error {
 	ctx, span := trace.StartSpan(ctx, "blockChain.saveHead")
 	defer span.End()
 
@@ -116,22 +120,16 @@ func (s *Service) saveHead(ctx context.Context, headRoot [32]byte, headBlock blo
 	if err := helpers.BeaconBlockIsNil(headBlock); err != nil {
 		return err
 	}
+	if headState == nil || headState.IsNil() {
+		return errors.New("cannot save nil head state")
+	}
 
 	// If the head state is not available, just return nil.
 	// There's nothing to cache
 	if !s.cfg.BeaconDB.HasStateSummary(ctx, headRoot) {
 		return nil
 	}
 
-	// Get the new head state from cached state or DB.
-	newHeadState, err := s.cfg.StateGen.StateByRoot(ctx, headRoot)
-	if err != nil {
-		return errors.Wrap(err, "could not retrieve head state in DB")
-	}
-	if newHeadState == nil || newHeadState.IsNil() {
-		return errors.New("cannot save nil head state")
-	}
-
 	// A chain re-org occurred, so we fire an event notifying the rest of the services.
 	headSlot := s.HeadSlot()
 	newHeadSlot := headBlock.Block().Slot()
@@ -170,7 +168,7 @@ func (s *Service) saveHead(ctx context.Context, headRoot [32]byte, headBlock blo
 	}
 
 	// Cache the new head info.
-	s.setHead(headRoot, headBlock, newHeadState)
+	s.setHead(headRoot, headBlock, headState)
 
 	// Save the new head root to DB.
 	if err := s.cfg.BeaconDB.SaveHeadBlockRoot(ctx, headRoot); err != nil {
@@ -180,7 +178,7 @@ func (s *Service) saveHead(ctx context.Context, headRoot [32]byte, headBlock blo
 	// Forward an event capturing a new chain head over a common event feed
 	// done in a goroutine to avoid blocking the critical runtime main routine.
 	go func() {
-		if err := s.notifyNewHeadEvent(ctx, newHeadSlot, newHeadState, newStateRoot, headRoot[:]); err != nil {
+		if err := s.notifyNewHeadEvent(ctx, newHeadSlot, headState, newStateRoot, headRoot[:]); err != nil {
 			log.WithError(err).Error("Could not notify event feed of new chain head")
 		}
 	}()

beacon-chain/blockchain/head_test.go

@@ -31,7 +31,8 @@ func TestSaveHead_Same(t *testing.T) {
 	service.head = &head{slot: 0, root: r}
 	b, err := wrapper.WrappedSignedBeaconBlock(util.NewBeaconBlock())
 	require.NoError(t, err)
-	require.NoError(t, service.saveHead(context.Background(), r, b))
+	st, _ := util.DeterministicGenesisState(t, 1)
+	require.NoError(t, service.saveHead(context.Background(), r, b, st))
 	assert.Equal(t, types.Slot(0), service.headSlot(), "Head did not stay the same")
 	assert.Equal(t, r, service.headRoot(), "Head did not stay the same")
 }
@@ -69,7 +70,7 @@ func TestSaveHead_Different(t *testing.T) {
 	require.NoError(t, headState.SetSlot(1))
 	require.NoError(t, service.cfg.BeaconDB.SaveStateSummary(context.Background(), &ethpb.StateSummary{Slot: 1, Root: newRoot[:]}))
 	require.NoError(t, service.cfg.BeaconDB.SaveState(context.Background(), headState, newRoot))
-	require.NoError(t, service.saveHead(context.Background(), newRoot, wsb))
+	require.NoError(t, service.saveHead(context.Background(), newRoot, wsb, headState))
 
 	assert.Equal(t, types.Slot(1), service.HeadSlot(), "Head did not change")
 
@@ -115,7 +116,7 @@ func TestSaveHead_Different_Reorg(t *testing.T) {
 	require.NoError(t, headState.SetSlot(1))
 	require.NoError(t, service.cfg.BeaconDB.SaveStateSummary(context.Background(), &ethpb.StateSummary{Slot: 1, Root: newRoot[:]}))
 	require.NoError(t, service.cfg.BeaconDB.SaveState(context.Background(), headState, newRoot))
-	require.NoError(t, service.saveHead(context.Background(), newRoot, wsb))
+	require.NoError(t, service.saveHead(context.Background(), newRoot, wsb, headState))
 
 	assert.Equal(t, types.Slot(1), service.HeadSlot(), "Head did not change")
 
@@ -159,7 +160,8 @@ func TestUpdateHead_MissingJustifiedRoot(t *testing.T) {
 	service.store.SetBestJustifiedCheckpt(&ethpb.Checkpoint{})
 	headRoot, err := service.updateHead(context.Background(), []uint64{})
 	require.NoError(t, err)
-	require.NoError(t, service.saveHead(context.Background(), headRoot, wsb))
+	st, _ := util.DeterministicGenesisState(t, 1)
+	require.NoError(t, service.saveHead(context.Background(), headRoot, wsb, st))
 }
 
 func Test_notifyNewHeadEvent(t *testing.T) {

beacon-chain/blockchain/optimistic_sync.go

@@ -5,22 +5,29 @@ import (
 	"fmt"
 
 	"github.com/pkg/errors"
+	types "github.com/prysmaticlabs/eth2-types"
 	"github.com/prysmaticlabs/prysm/beacon-chain/core/blocks"
 	"github.com/prysmaticlabs/prysm/beacon-chain/core/helpers"
+	"github.com/prysmaticlabs/prysm/beacon-chain/core/time"
+	"github.com/prysmaticlabs/prysm/beacon-chain/core/transition"
+	"github.com/prysmaticlabs/prysm/beacon-chain/db/kv"
 	"github.com/prysmaticlabs/prysm/beacon-chain/powchain"
+	"github.com/prysmaticlabs/prysm/beacon-chain/state"
+	fieldparams "github.com/prysmaticlabs/prysm/config/fieldparams"
 	"github.com/prysmaticlabs/prysm/config/params"
 	"github.com/prysmaticlabs/prysm/encoding/bytesutil"
 	enginev1 "github.com/prysmaticlabs/prysm/proto/engine/v1"
 	ethpb "github.com/prysmaticlabs/prysm/proto/prysm/v1alpha1"
 	"github.com/prysmaticlabs/prysm/proto/prysm/v1alpha1/block"
+	"github.com/prysmaticlabs/prysm/time/slots"
 	"github.com/sirupsen/logrus"
 	"go.opencensus.io/trace"
 )
 
 // notifyForkchoiceUpdate signals execution engine the fork choice updates. Execution engine should:
 // 1. Re-organizes the execution payload chain and corresponding state to make head_block_hash the head.
 // 2. Applies finality to the execution state: it irreversibly persists the chain of all execution payloads and corresponding state, up to and including finalized_block_hash.
-func (s *Service) notifyForkchoiceUpdate(ctx context.Context, headBlk block.BeaconBlock, headRoot [32]byte, finalizedRoot [32]byte) (*enginev1.PayloadIDBytes, error) {
+func (s *Service) notifyForkchoiceUpdate(ctx context.Context, headState state.BeaconState, headBlk block.BeaconBlock, headRoot [32]byte, finalizedRoot [32]byte) (*enginev1.PayloadIDBytes, error) {
 	ctx, span := trace.StartSpan(ctx, "blockChain.notifyForkchoiceUpdate")
 	defer span.End()
 
@@ -66,8 +73,13 @@ func (s *Service) notifyForkchoiceUpdate(ctx context.Context, headBlk block.Beac
 		FinalizedBlockHash: finalizedHash,
 	}
 
-	// payload attribute is only required when requesting payload, here we are just updating fork choice, so it is nil.
-	payloadID, _, err := s.cfg.ExecutionEngineCaller.ForkchoiceUpdated(ctx, fcs, nil /*payload attribute*/)
+	nextSlot := s.CurrentSlot() + 1 // Cache payload ID for next slot proposer.
+	hasAttr, attr, proposerId, err := s.getPayloadAttribute(ctx, headState, nextSlot)
+	if err != nil {
+		return nil, errors.Wrap(err, "could not get payload attribute")
+	}
+
+	payloadID, _, err := s.cfg.ExecutionEngineCaller.ForkchoiceUpdated(ctx, fcs, attr)
 	if err != nil {
 		switch err {
 		case powchain.ErrAcceptedSyncingPayloadStatus:
@@ -84,6 +96,11 @@ func (s *Service) notifyForkchoiceUpdate(ctx context.Context, headBlk block.Beac
 	if err := s.cfg.ForkChoiceStore.SetOptimisticToValid(ctx, headRoot); err != nil {
 		return nil, errors.Wrap(err, "could not set block to valid")
 	}
+	if hasAttr { // If the forkchoice update call has an attribute, update the proposer payload ID cache.
+		var pId [8]byte
+		copy(pId[:], payloadID[:])
+		s.cfg.ProposerSlotIndexCache.SetProposerAndPayloadIDs(nextSlot, proposerId, pId)
+	}
 	return payloadID, nil
 }
 
@@ -188,6 +205,55 @@ func (s *Service) optimisticCandidateBlock(ctx context.Context, blk block.Beacon
 	return parentIsExecutionBlock, nil
 }
 
+// getPayloadAttributes returns the payload attributes for the given state and slot.
+// The attribute is required to initiate a payload build process in the context of an `engine_forkchoiceUpdated` call.
+func (s *Service) getPayloadAttribute(ctx context.Context, st state.BeaconState, slot types.Slot) (bool, *enginev1.PayloadAttributes, types.ValidatorIndex, error) {
+	proposerID, _, ok := s.cfg.ProposerSlotIndexCache.GetProposerPayloadIDs(slot)
+	if !ok { // There's no need to build attribute if there is no proposer for slot.
+		return false, nil, 0, nil
+	}
+
+	// Get previous randao.
+	st = st.Copy()
+	st, err := transition.ProcessSlotsIfPossible(ctx, st, slot)
+	if err != nil {
+		return false, nil, 0, err
+	}
+	prevRando, err := helpers.RandaoMix(st, time.CurrentEpoch(st))
+	if err != nil {
+		return false, nil, 0, nil
+	}
+
+	// Get fee recipient.
+	feeRecipient := params.BeaconConfig().DefaultFeeRecipient
+	recipient, err := s.cfg.BeaconDB.FeeRecipientByValidatorID(ctx, proposerID)
+	switch {
+	case errors.Is(err, kv.ErrNotFoundFeeRecipient):
+		if feeRecipient.String() == fieldparams.EthBurnAddressHex {
+			logrus.WithFields(logrus.Fields{
+				"validatorIndex": proposerID,
+				"burnAddress":    fieldparams.EthBurnAddressHex,
+			}).Error("Fee recipient not set. Using burn address")
+		}
+	case err != nil:
+		return false, nil, 0, errors.Wrap(err, "could not get fee recipient in db")
+	default:
+		feeRecipient = recipient
+	}
+
+	// Get timestamp.
+	t, err := slots.ToTime(uint64(s.genesisTime.Unix()), slot)
+	if err != nil {
+		return false, nil, 0, err
+	}
+	attr := &enginev1.PayloadAttributes{
+		Timestamp:             uint64(t.Unix()),
+		PrevRandao:            prevRando,
+		SuggestedFeeRecipient: feeRecipient.Bytes(),
+	}
+	return true, attr, proposerID, nil
+}
+
 // removeInvalidBlockAndState removes the invalid block and its corresponding state from the cache and DB.
 func (s *Service) removeInvalidBlockAndState(ctx context.Context, blkRoots [][32]byte) error {
 	for _, root := range blkRoots {

beacon-chain/blockchain/optimistic_sync_test.go

@@ -5,6 +5,9 @@ import (
 	"testing"
 	"time"
 
+	"github.com/ethereum/go-ethereum/common"
+	types "github.com/prysmaticlabs/eth2-types"
+	"github.com/prysmaticlabs/prysm/beacon-chain/cache"
 	"github.com/prysmaticlabs/prysm/beacon-chain/core/blocks"
 	testDB "github.com/prysmaticlabs/prysm/beacon-chain/db/testing"
 	"github.com/prysmaticlabs/prysm/beacon-chain/forkchoice/protoarray"
@@ -24,6 +27,7 @@ import (
 	"github.com/prysmaticlabs/prysm/testing/require"
 	"github.com/prysmaticlabs/prysm/testing/util"
 	"github.com/prysmaticlabs/prysm/time/slots"
+	logTest "github.com/sirupsen/logrus/hooks/test"
 )
 
 func Test_NotifyForkchoiceUpdate(t *testing.T) {
@@ -44,8 +48,13 @@ func Test_NotifyForkchoiceUpdate(t *testing.T) {
 		WithDatabase(beaconDB),
 		WithStateGen(stategen.New(beaconDB)),
 		WithForkChoiceStore(fcs),
+		WithProposerIdsCache(cache.NewProposerPayloadIDsCache()),
 	}
 	service, err := NewService(ctx, opts...)
+	st, _ := util.DeterministicGenesisState(t, 1)
+	service.head = &head{
+		state: st,
+	}
 	require.NoError(t, err)
 	require.NoError(t, fcs.InsertOptimisticBlock(ctx, 0, [32]byte{}, [32]byte{}, params.BeaconConfig().ZeroHash, 0, 0))
 
@@ -157,7 +166,8 @@ func Test_NotifyForkchoiceUpdate(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			service.cfg.ExecutionEngineCaller = &mockPOW.EngineClient{ErrForkchoiceUpdated: tt.newForkchoiceErr}
-			_, err := service.notifyForkchoiceUpdate(ctx, tt.blk, service.headRoot(), tt.finalizedRoot)
+			st, _ := util.DeterministicGenesisState(t, 1)
+			_, err := service.notifyForkchoiceUpdate(ctx, st, tt.blk, service.headRoot(), tt.finalizedRoot)
 			if tt.errString != "" {
 				require.ErrorContains(t, tt.errString, err)
 			} else {
@@ -588,6 +598,47 @@ func Test_IsOptimisticShallowExecutionParent(t *testing.T) {
 	require.Equal(t, true, candidate)
 }
 
+func Test_GetPayloadAttribute(t *testing.T) {
+	ctx := context.Background()
+	beaconDB := testDB.SetupDB(t)
+	opts := []Option{
+		WithDatabase(beaconDB),
+		WithStateGen(stategen.New(beaconDB)),
+		WithProposerIdsCache(cache.NewProposerPayloadIDsCache()),
+	}
+
+	// Cache miss
+	service, err := NewService(ctx, opts...)
+	require.NoError(t, err)
+	hasPayload, _, vId, err := service.getPayloadAttribute(ctx, nil, 0)
+	require.NoError(t, err)
+	require.Equal(t, false, hasPayload)
+	require.Equal(t, types.ValidatorIndex(0), vId)
+
+	// Cache hit, advance state, no fee recipient
+	suggestedVid := types.ValidatorIndex(1)
+	slot := types.Slot(1)
+	service.cfg.ProposerSlotIndexCache.SetProposerAndPayloadIDs(slot, suggestedVid, [8]byte{})
+	st, _ := util.DeterministicGenesisState(t, 1)
+	hook := logTest.NewGlobal()
+	hasPayload, attr, vId, err := service.getPayloadAttribute(ctx, st, slot)
+	require.NoError(t, err)
+	require.Equal(t, true, hasPayload)
+	require.Equal(t, suggestedVid, vId)
+	require.Equal(t, fieldparams.EthBurnAddressHex, common.BytesToAddress(attr.SuggestedFeeRecipient).String())
+	require.LogsContain(t, hook, "Fee recipient not set. Using burn address")
+
+	// Cache hit, advance state, has fee recipient
+	suggestedAddr := common.HexToAddress("123")
+	require.NoError(t, service.cfg.BeaconDB.SaveFeeRecipientsByValidatorIDs(ctx, []types.ValidatorIndex{suggestedVid}, []common.Address{suggestedAddr}))
+	service.cfg.ProposerSlotIndexCache.SetProposerAndPayloadIDs(slot, suggestedVid, [8]byte{})
+	hasPayload, attr, vId, err = service.getPayloadAttribute(ctx, st, slot)
+	require.NoError(t, err)
+	require.Equal(t, true, hasPayload)
+	require.Equal(t, suggestedVid, vId)
+	require.Equal(t, suggestedAddr, common.BytesToAddress(attr.SuggestedFeeRecipient))
+}
+
 func Test_UpdateLastValidatedCheckpoint(t *testing.T) {
 	params.SetupTestConfigCleanup(t)
 	params.OverrideBeaconConfig(params.MainnetConfig())

beacon-chain/blockchain/options.go

@@ -2,6 +2,7 @@ package blockchain
 
 import (
 	"github.com/prysmaticlabs/prysm/async/event"
+	"github.com/prysmaticlabs/prysm/beacon-chain/cache"
 	"github.com/prysmaticlabs/prysm/beacon-chain/cache/depositcache"
 	statefeed "github.com/prysmaticlabs/prysm/beacon-chain/core/feed/state"
 	"github.com/prysmaticlabs/prysm/beacon-chain/db"
@@ -66,6 +67,14 @@ func WithDepositCache(c *depositcache.DepositCache) Option {
 	}
 }
 
+// WithProposerIdsCache for proposer id cache.
+func WithProposerIdsCache(c *cache.ProposerPayloadIDsCache) Option {
+	return func(s *Service) error {
+		s.cfg.ProposerSlotIndexCache = c
+		return nil
+	}
+}
+
 // WithAttestationPool for attestation lifecycle after chain inclusion.
 func WithAttestationPool(p attestations.Pool) Option {
 	return func(s *Service) error {