defer payload attribute computation (#14644)

* defer payload attribute computation

* fire payload event on skipped slots

* changelog

* fix test and missing version attr

* fix lint

* deepsource

* mv head block lookup for missed slots to streamer

---------

Co-authored-by: Kasey Kirkham <[email protected]>

CHANGELOG.md

@@ -55,6 +55,7 @@ The format is based on Keep a Changelog, and this project adheres to Semantic Ve
 - Return early for blob reconstructor during capella fork
 - Updated block endpoint from V1 to V2
 - Rename instances of "deposit receipts" to "deposit requests".
+- non-blocking payload attribute event handling in beacon api [pr](https://github.com/prysmaticlabs/prysm/pull/14644)
 
 ### Deprecated
 

beacon-chain/blockchain/execution_engine.go

@@ -6,8 +6,11 @@ import (
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/pkg/errors"
+	"github.com/prysmaticlabs/prysm/v5/async/event"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/cache"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/blocks"
+	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/feed"
+	statefeed "github.com/prysmaticlabs/prysm/v5/beacon-chain/core/feed/state"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/helpers"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/time"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/transition"
@@ -69,6 +72,7 @@ func (s *Service) notifyForkchoiceUpdate(ctx context.Context, arg *fcuConfig) (*
 	if arg.attributes == nil {
 		arg.attributes = payloadattribute.EmptyWithVersion(headBlk.Version())
 	}
+	go firePayloadAttributesEvent(ctx, s.cfg.StateNotifier.StateFeed(), arg)
 	payloadID, lastValidHash, err := s.cfg.ExecutionEngineCaller.ForkchoiceUpdated(ctx, fcs, arg.attributes)
 	if err != nil {
 		switch {
@@ -167,6 +171,38 @@ func (s *Service) notifyForkchoiceUpdate(ctx context.Context, arg *fcuConfig) (*
 	return payloadID, nil
 }
 
+func firePayloadAttributesEvent(ctx context.Context, f event.SubscriberSender, cfg *fcuConfig) {
+	pidx, err := helpers.BeaconProposerIndex(ctx, cfg.headState)
+	if err != nil {
+		log.WithError(err).
+			WithField("head_root", cfg.headRoot[:]).
+			Error("Could not get proposer index for PayloadAttributes event")
+		return
+	}
+	evd := payloadattribute.EventData{
+		ProposerIndex:   pidx,
+		ProposalSlot:    cfg.headState.Slot(),
+		ParentBlockRoot: cfg.headRoot[:],
+		Attributer:      cfg.attributes,
+		HeadRoot:        cfg.headRoot,
+		HeadState:       cfg.headState,
+		HeadBlock:       cfg.headBlock,
+	}
+	if cfg.headBlock != nil && !cfg.headBlock.IsNil() {
+		headPayload, err := cfg.headBlock.Block().Body().Execution()
+		if err != nil {
+			log.WithError(err).Error("Could not get execution payload for head block")
+			return
+		}
+		evd.ParentBlockHash = headPayload.BlockHash()
+		evd.ParentBlockNumber = headPayload.BlockNumber()
+	}
+	f.Send(&feed.Event{
+		Type: statefeed.PayloadAttributes,
+		Data: evd,
+	})
+}
+
 // getPayloadHash returns the payload hash given the block root.
 // if the block is before bellatrix fork epoch, it returns the zero hash.
 func (s *Service) getPayloadHash(ctx context.Context, root []byte) ([32]byte, error) {

beacon-chain/blockchain/process_block.go

@@ -7,8 +7,6 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/blocks"
-	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/feed"
-	statefeed "github.com/prysmaticlabs/prysm/v5/beacon-chain/core/feed/state"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/helpers"
 	coreTime "github.com/prysmaticlabs/prysm/v5/beacon-chain/core/time"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/transition"
@@ -620,9 +618,6 @@ func (s *Service) lateBlockTasks(ctx context.Context) {
 	if !s.inRegularSync() {
 		return
 	}
-	s.cfg.StateNotifier.StateFeed().Send(&feed.Event{
-		Type: statefeed.MissedSlot,
-	})
 	s.headLock.RLock()
 	headRoot := s.headRoot()
 	headState := s.headState(ctx)
@@ -650,6 +645,13 @@ func (s *Service) lateBlockTasks(ctx context.Context) {
 	attribute := s.getPayloadAttribute(ctx, headState, s.CurrentSlot()+1, headRoot[:])
 	// return early if we are not proposing next slot
 	if attribute.IsEmpty() {
+		fcuArgs := &fcuConfig{
+			headState:  headState,
+			headRoot:   headRoot,
+			headBlock:  nil,
+			attributes: attribute,
+		}
+		go firePayloadAttributesEvent(ctx, s.cfg.StateNotifier.StateFeed(), fcuArgs)
 		return
 	}
 

beacon-chain/core/feed/state/events.go

@@ -31,6 +31,8 @@ const (
 	LightClientFinalityUpdate
 	// LightClientOptimisticUpdate event
 	LightClientOptimisticUpdate
+	// PayloadAttributes events are fired upon a missed slot or new head.
+	PayloadAttributes
 )
 
 // BlockProcessedData is the data sent with BlockProcessed events.

beacon-chain/rpc/eth/events/BUILD.bazel

@@ -19,11 +19,12 @@ go_library(
         "//beacon-chain/core/feed/state:go_default_library",
         "//beacon-chain/core/helpers:go_default_library",
         "//beacon-chain/core/time:go_default_library",
-        "//beacon-chain/core/transition:go_default_library",
         "//config/params:go_default_library",
+        "//consensus-types/payload-attribute:go_default_library",
         "//consensus-types/primitives:go_default_library",
         "//monitoring/tracing/trace:go_default_library",
         "//network/httputil:go_default_library",
+        "//proto/engine/v1:go_default_library",
         "//proto/eth/v1:go_default_library",
         "//proto/eth/v2:go_default_library",
         "//proto/prysm/v1alpha1:go_default_library",
@@ -52,6 +53,7 @@ go_test(
         "//config/fieldparams:go_default_library",
         "//consensus-types/blocks:go_default_library",
         "//consensus-types/interfaces:go_default_library",
+        "//consensus-types/payload-attribute:go_default_library",
         "//consensus-types/primitives:go_default_library",
         "//proto/eth/v1:go_default_library",
         "//proto/prysm/v1alpha1:go_default_library",