[pull] develop from OffchainLabs:develop

beacon-chain/blockchain/execution_engine_test.go

@@ -1053,40 +1053,3 @@ func TestKZGCommitmentToVersionedHashes(t *testing.T) {
 	require.Equal(t, vhs[0].String(), vh0)
 	require.Equal(t, vhs[1].String(), vh1)
 }
-
-func TestComputePayloadAttribute(t *testing.T) {
-	service, tr := minimalTestService(t, WithPayloadIDCache(cache.NewPayloadIDCache()))
-	ctx := tr.ctx
-
-	st, _ := util.DeterministicGenesisStateBellatrix(t, 1)
-
-	service.cfg.TrackedValidatorsCache.Set(cache.TrackedValidator{Active: true, Index: 0})
-	// Cache hit, advance state, no fee recipient
-	slot := primitives.Slot(1)
-	service.cfg.PayloadIDCache.Set(slot, [32]byte{}, [8]byte{})
-	blk := util.NewBeaconBlockBellatrix()
-	signed, err := consensusblocks.NewSignedBeaconBlock(blk)
-	require.NoError(t, err)
-	roblock, err := consensusblocks.NewROBlockWithRoot(signed, [32]byte{'a'})
-	require.NoError(t, err)
-	cfg := &postBlockProcessConfig{
-		ctx:     ctx,
-		roblock: roblock,
-	}
-	fcu := &fcuConfig{
-		headState:     st,
-		proposingSlot: slot,
-		headRoot:      [32]byte{},
-	}
-	require.NoError(t, service.computePayloadAttributes(cfg, fcu))
-	require.Equal(t, false, fcu.attributes.IsEmpty())
-	require.Equal(t, params.BeaconConfig().EthBurnAddressHex, common.BytesToAddress(fcu.attributes.SuggestedFeeRecipient()).String())
-
-	// Cache hit, advance state, has fee recipient
-	suggestedAddr := common.HexToAddress("123")
-	service.cfg.TrackedValidatorsCache.Set(cache.TrackedValidator{Active: true, FeeRecipient: primitives.ExecutionAddress(suggestedAddr), Index: 0})
-	service.cfg.PayloadIDCache.Set(slot, [32]byte{}, [8]byte{})
-	require.NoError(t, service.computePayloadAttributes(cfg, fcu))
-	require.Equal(t, false, fcu.attributes.IsEmpty())
-	require.Equal(t, suggestedAddr, common.BytesToAddress(fcu.attributes.SuggestedFeeRecipient()))
-}

beacon-chain/blockchain/forkchoice_update_execution.go

@@ -12,6 +12,7 @@ import (
 	payloadattribute "github.com/OffchainLabs/prysm/v7/consensus-types/payload-attribute"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/primitives"
 	"github.com/OffchainLabs/prysm/v7/monitoring/tracing/trace"
+	"github.com/OffchainLabs/prysm/v7/runtime/version"
 	"github.com/OffchainLabs/prysm/v7/time/slots"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
@@ -53,58 +54,53 @@ type fcuConfig struct {
 }
 
 // sendFCU handles the logic to notify the engine of a forckhoice update
-// for the first time when processing an incoming block during regular sync. It
-// always updates the shuffling caches and handles epoch transitions when the
-// incoming block is late, preparing payload attributes in this case while it
-// only sends a message with empty attributes for early blocks.
-func (s *Service) sendFCU(cfg *postBlockProcessConfig, fcuArgs *fcuConfig) error {
-	if !s.isNewHead(cfg.headRoot) {
-		return nil
+// when processing an incoming block during regular sync. It
+// always updates the shuffling caches and handles epoch transitions .
+func (s *Service) sendFCU(cfg *postBlockProcessConfig, fcuArgs *fcuConfig) {
+	if cfg.postState.Version() < version.Fulu {
+		// update the caches to compute the right proposer index
+		// this function is called under a forkchoice lock which we need to release.
+		s.ForkChoicer().Unlock()
+		s.updateCachesPostBlockProcessing(cfg)
+		s.ForkChoicer().Lock()
 	}
-	if fcuArgs.attributes != nil && !fcuArgs.attributes.IsEmpty() && s.shouldOverrideFCU(cfg.headRoot, s.CurrentSlot()+1) {
-		return nil
+	if err := s.getFCUArgs(cfg, fcuArgs); err != nil {
+		log.WithError(err).Error("Could not get forkchoice update argument")
+		return
 	}
-	return s.forkchoiceUpdateWithExecution(cfg.ctx, fcuArgs)
-}
-
-// sendFCUWithAttributes computes the payload attributes and sends an FCU message
-// to the engine if needed
-func (s *Service) sendFCUWithAttributes(cfg *postBlockProcessConfig, fcuArgs *fcuConfig) {
-	slotCtx, cancel := context.WithTimeout(context.Background(), slotDeadline)
-	defer cancel()
-	cfg.ctx = slotCtx
-	s.cfg.ForkChoiceStore.RLock()
-	defer s.cfg.ForkChoiceStore.RUnlock()
-	if err := s.computePayloadAttributes(cfg, fcuArgs); err != nil {
-		log.WithError(err).Error("Could not compute payload attributes")
+	// If head has not been updated and attributes are nil, we can skip the FCU.
+	if !s.isNewHead(cfg.headRoot) && (fcuArgs.attributes == nil || fcuArgs.attributes.IsEmpty()) {
 		return
 	}
-	if fcuArgs.attributes.IsEmpty() {
+	// If we are proposing and we aim to reorg the block, we have already sent FCU with attributes on lateBlockTasks
+	if fcuArgs.attributes != nil && !fcuArgs.attributes.IsEmpty() && s.shouldOverrideFCU(cfg.headRoot, s.CurrentSlot()+1) {
 		return
 	}
-	if _, err := s.notifyForkchoiceUpdate(cfg.ctx, fcuArgs); err != nil {
-		log.WithError(err).Error("Could not update forkchoice with payload attributes for proposal")
+	if s.inRegularSync() {
+		go s.forkchoiceUpdateWithExecution(cfg.ctx, fcuArgs)
+	}
+
+	if s.isNewHead(fcuArgs.headRoot) {
+		if err := s.saveHead(cfg.ctx, fcuArgs.headRoot, fcuArgs.headBlock, fcuArgs.headState); err != nil {
+			log.WithError(err).Error("Could not save head")
+		}
+		s.pruneAttsFromPool(s.ctx, fcuArgs.headState, fcuArgs.headBlock)
 	}
 }
 
-// fockchoiceUpdateWithExecution is a wrapper around notifyForkchoiceUpdate. It decides whether a new call to FCU should be made.
-func (s *Service) forkchoiceUpdateWithExecution(ctx context.Context, args *fcuConfig) error {
+// fockchoiceUpdateWithExecution is a wrapper around notifyForkchoiceUpdate. It gets a forkchoice lock and calls the engine.
+// The caller of this function should NOT have a lock in forkchoice store.
+func (s *Service) forkchoiceUpdateWithExecution(ctx context.Context, args *fcuConfig) {
 	_, span := trace.StartSpan(ctx, "beacon-chain.blockchain.forkchoiceUpdateWithExecution")
 	defer span.End()
 	// Note: Use the service context here to avoid the parent context being ended during a forkchoice update.
 	ctx = trace.NewContext(s.ctx, span)
+	s.ForkChoicer().Lock()
+	defer s.ForkChoicer().Unlock()
 	_, err := s.notifyForkchoiceUpdate(ctx, args)
 	if err != nil {
-		return errors.Wrap(err, "could not notify forkchoice update")
+		log.WithError(err).Error("Could not notify forkchoice update")
 	}
-
-	if err := s.saveHead(ctx, args.headRoot, args.headBlock, args.headState); err != nil {
-		log.WithError(err).Error("Could not save head")
-	}
-
-	// Only need to prune attestations from pool if the head has changed.
-	s.pruneAttsFromPool(s.ctx, args.headState, args.headBlock)
-	return nil
 }
 
 // shouldOverrideFCU checks whether the incoming block is still subject to being

beacon-chain/blockchain/forkchoice_update_execution_test.go

@@ -97,7 +97,7 @@ func TestService_forkchoiceUpdateWithExecution_exceptionalCases(t *testing.T) {
 		headBlock:     wsb,
 		proposingSlot: service.CurrentSlot() + 1,
 	}
-	require.NoError(t, service.forkchoiceUpdateWithExecution(ctx, args))
+	service.forkchoiceUpdateWithExecution(ctx, args)
 
 	payloadID, has := service.cfg.PayloadIDCache.PayloadID(2, [32]byte{2})
 	require.Equal(t, true, has)
@@ -151,7 +151,7 @@ func TestService_forkchoiceUpdateWithExecution_SameHeadRootNewProposer(t *testin
 		headRoot:      r,
 		proposingSlot: service.CurrentSlot() + 1,
 	}
-	require.NoError(t, service.forkchoiceUpdateWithExecution(ctx, args))
+	service.forkchoiceUpdateWithExecution(ctx, args)
 }
 
 func TestShouldOverrideFCU(t *testing.T) {

beacon-chain/blockchain/process_block.go

@@ -66,9 +66,6 @@ func (s *Service) postBlockProcess(cfg *postBlockProcessConfig) error {
 	startTime := time.Now()
 	fcuArgs := &fcuConfig{}
 
-	if s.inRegularSync() {
-		defer s.handleSecondFCUCall(cfg, fcuArgs)
-	}
 	if features.Get().EnableLightClient && slots.ToEpoch(s.CurrentSlot()) >= params.BeaconConfig().AltairForkEpoch {
 		defer s.processLightClientUpdates(cfg)
 	}
@@ -105,14 +102,12 @@ func (s *Service) postBlockProcess(cfg *postBlockProcessConfig) error {
 		s.logNonCanonicalBlockReceived(cfg.roblock.Root(), cfg.headRoot)
 		return nil
 	}
-	if err := s.getFCUArgs(cfg, fcuArgs); err != nil {
-		log.WithError(err).Error("Could not get forkchoice update argument")
-		return nil
-	}
-	if err := s.sendFCU(cfg, fcuArgs); err != nil {
-		return errors.Wrap(err, "could not send FCU to engine")
-	}
+	s.sendFCU(cfg, fcuArgs)
 
+	// Pre-Fulu the caches are updated when computing the payload attributes
+	if cfg.postState.Version() >= version.Fulu {
+		go s.updateCachesPostBlockProcessing(cfg)
+	}
 	return nil
 }
 
@@ -322,6 +317,7 @@ func (s *Service) areSidecarsAvailable(ctx context.Context, avs das.Availability
 	return nil
 }
 
+// the caller of this function must not hold a lock in forkchoice store.
 func (s *Service) updateEpochBoundaryCaches(ctx context.Context, st state.BeaconState) error {
 	e := coreTime.CurrentEpoch(st)
 	if err := helpers.UpdateCommitteeCache(ctx, st, e); err != nil {
@@ -351,7 +347,9 @@ func (s *Service) updateEpochBoundaryCaches(ctx context.Context, st state.Beacon
 	if e > 0 {
 		e = e - 1
 	}
+	s.ForkChoicer().RLock()
 	target, err := s.cfg.ForkChoiceStore.TargetRootForEpoch(r, e)
+	s.ForkChoicer().RUnlock()
 	if err != nil {
 		log.WithError(err).Error("Could not update proposer index state-root map")
 		return nil
@@ -364,7 +362,7 @@ func (s *Service) updateEpochBoundaryCaches(ctx context.Context, st state.Beacon
 }
 
 // Epoch boundary tasks: it copies the headState and updates the epoch boundary
-// caches.
+// caches. The caller of this function must not hold a lock in forkchoice store.
 func (s *Service) handleEpochBoundary(ctx context.Context, slot primitives.Slot, headState state.BeaconState, blockRoot []byte) error {
 	ctx, span := trace.StartSpan(ctx, "blockChain.handleEpochBoundary")
 	defer span.End()
@@ -904,8 +902,6 @@ func (s *Service) lateBlockTasks(ctx context.Context) {
 	if currentSlot == s.HeadSlot() {
 		return
 	}
-	s.cfg.ForkChoiceStore.RLock()
-	defer s.cfg.ForkChoiceStore.RUnlock()
 	// return early if we are in init sync
 	if !s.inRegularSync() {
 		return
@@ -918,14 +914,30 @@ func (s *Service) lateBlockTasks(ctx context.Context) {
 	if lastState == nil {
 		lastRoot, lastState = headRoot[:], headState
 	}
-	// Copy all the field tries in our cached state in the event of late
-	// blocks.
-	lastState.CopyAllTries()
-	if err := transition.UpdateNextSlotCache(ctx, lastRoot, lastState); err != nil {
-		log.WithError(err).Debug("Could not update next slot state cache")
-	}
-	if err := s.handleEpochBoundary(ctx, currentSlot, headState, headRoot[:]); err != nil {
-		log.WithError(err).Error("Could not update epoch boundary caches")
+	// Before Fulu we need to process the next slot to find out if we are proposing.
+	if lastState.Version() < version.Fulu {
+		// Copy all the field tries in our cached state in the event of late
+		// blocks.
+		lastState.CopyAllTries()
+		if err := transition.UpdateNextSlotCache(ctx, lastRoot, lastState); err != nil {
+			log.WithError(err).Debug("Could not update next slot state cache")
+		}
+		if err := s.handleEpochBoundary(ctx, currentSlot, headState, headRoot[:]); err != nil {
+			log.WithError(err).Error("Could not update epoch boundary caches")
+		}
+	} else {
+		// After Fulu, we can update the caches asynchronously after sending FCU to the engine
+		defer func() {
+			go func() {
+				lastState.CopyAllTries()
+				if err := transition.UpdateNextSlotCache(ctx, lastRoot, lastState); err != nil {
+					log.WithError(err).Debug("Could not update next slot state cache")
+				}
+				if err := s.handleEpochBoundary(ctx, currentSlot, headState, headRoot[:]); err != nil {
+					log.WithError(err).Error("Could not update epoch boundary caches")
+				}
+			}()
+		}()
 	}
 	// return early if we already started building a block for the current
 	// head root
@@ -955,6 +967,8 @@ func (s *Service) lateBlockTasks(ctx context.Context) {
 		headBlock:  headBlock,
 		attributes: attribute,
 	}
+	s.cfg.ForkChoiceStore.Lock()
+	defer s.cfg.ForkChoiceStore.Unlock()
 	_, err = s.notifyForkchoiceUpdate(ctx, fcuArgs)
 	if err != nil {
 		log.WithError(err).Debug("could not perform late block tasks: failed to update forkchoice with engine")

beacon-chain/blockchain/process_block_helpers.go

@@ -42,14 +42,8 @@ func (s *Service) getFCUArgs(cfg *postBlockProcessConfig, fcuArgs *fcuConfig) er
 	if err := s.getFCUArgsEarlyBlock(cfg, fcuArgs); err != nil {
 		return err
 	}
-	if !s.inRegularSync() {
-		return nil
-	}
-	slot := cfg.roblock.Block().Slot()
-	if slots.WithinVotingWindow(s.genesisTime, slot) {
-		return nil
-	}
-	return s.computePayloadAttributes(cfg, fcuArgs)
+	fcuArgs.attributes = s.getPayloadAttribute(cfg.ctx, fcuArgs.headState, fcuArgs.proposingSlot, cfg.headRoot[:])
+	return nil
 }
 
 func (s *Service) getFCUArgsEarlyBlock(cfg *postBlockProcessConfig, fcuArgs *fcuConfig) error {
@@ -173,26 +167,19 @@ func (s *Service) processLightClientUpdates(cfg *postBlockProcessConfig) {
 
 // updateCachesPostBlockProcessing updates the next slot cache and handles the epoch
 // boundary in order to compute the right proposer indices after processing
-// state transition. This function is called on late blocks while still locked,
-// before sending FCU to the engine.
-func (s *Service) updateCachesPostBlockProcessing(cfg *postBlockProcessConfig) error {
+// state transition. The caller of this function must not hold a lock in forkchoice store.
+func (s *Service) updateCachesPostBlockProcessing(cfg *postBlockProcessConfig) {
 	slot := cfg.postState.Slot()
 	root := cfg.roblock.Root()
 	if err := transition.UpdateNextSlotCache(cfg.ctx, root[:], cfg.postState); err != nil {
-		return errors.Wrap(err, "could not update next slot state cache")
+		log.WithError(err).Error("Could not update next slot state cache")
+		return
 	}
 	if !slots.IsEpochEnd(slot) {
-		return nil
+		return
 	}
-	return s.handleEpochBoundary(cfg.ctx, slot, cfg.postState, root[:])
-}
-
-// handleSecondFCUCall handles a second call to FCU when syncing a new block.
-// This is useful when proposing in the next block and we want to defer the
-// computation of the next slot shuffling.
-func (s *Service) handleSecondFCUCall(cfg *postBlockProcessConfig, fcuArgs *fcuConfig) {
-	if (fcuArgs.attributes == nil || fcuArgs.attributes.IsEmpty()) && cfg.headRoot == cfg.roblock.Root() {
-		go s.sendFCUWithAttributes(cfg, fcuArgs)
+	if err := s.handleEpochBoundary(cfg.ctx, slot, cfg.postState, root[:]); err != nil {
+		log.WithError(err).Error("Could not handle epoch boundary")
 	}
 }
 
@@ -202,20 +189,6 @@ func reportProcessingTime(startTime time.Time) {
 	onBlockProcessingTime.Observe(float64(time.Since(startTime).Milliseconds()))
 }
 
-// computePayloadAttributes modifies the passed FCU arguments to
-// contain the right payload attributes with the tracked proposer. It gets
-// called on blocks that arrive after the attestation voting window, or in a
-// background routine after syncing early blocks.
-func (s *Service) computePayloadAttributes(cfg *postBlockProcessConfig, fcuArgs *fcuConfig) error {
-	if cfg.roblock.Root() == cfg.headRoot {
-		if err := s.updateCachesPostBlockProcessing(cfg); err != nil {
-			return err
-		}
-	}
-	fcuArgs.attributes = s.getPayloadAttribute(cfg.ctx, fcuArgs.headState, fcuArgs.proposingSlot, cfg.headRoot[:])
-	return nil
-}
-
 // getBlockPreState returns the pre state of an incoming block. It uses the parent root of the block
 // to retrieve the state in DB. It verifies the pre state's validity and the incoming block
 // is in the correct time window.