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 <kasey@users.noreply.github.com>

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",

beacon-chain/rpc/eth/events/events.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"strconv"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common/hexutil"
@@ -18,11 +19,12 @@ import (
 	statefeed "github.com/prysmaticlabs/prysm/v5/beacon-chain/core/feed/state"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/helpers"
 	chaintime "github.com/prysmaticlabs/prysm/v5/beacon-chain/core/time"
-	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/transition"
 	"github.com/prysmaticlabs/prysm/v5/config/params"
+	payloadattribute "github.com/prysmaticlabs/prysm/v5/consensus-types/payload-attribute"
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/primitives"
 	"github.com/prysmaticlabs/prysm/v5/monitoring/tracing/trace"
 	"github.com/prysmaticlabs/prysm/v5/network/httputil"
+	engine "github.com/prysmaticlabs/prysm/v5/proto/engine/v1"
 	ethpb "github.com/prysmaticlabs/prysm/v5/proto/eth/v1"
 	ethpbv2 "github.com/prysmaticlabs/prysm/v5/proto/eth/v2"
 	eth "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
@@ -31,6 +33,7 @@ import (
 )
 
 const DefaultEventFeedDepth = 1000
+const payloadAttributeTimeout = 2 * time.Second
 
 const (
 	InvalidTopic = "__invalid__"
@@ -89,12 +92,12 @@ var opsFeedEventTopics = map[feed.EventType]string{
 
 var stateFeedEventTopics = map[feed.EventType]string{
 	statefeed.NewHead:                     HeadTopic,
-	statefeed.MissedSlot:                  PayloadAttributesTopic,
 	statefeed.FinalizedCheckpoint:         FinalizedCheckpointTopic,
 	statefeed.LightClientFinalityUpdate:   LightClientFinalityUpdateTopic,
 	statefeed.LightClientOptimisticUpdate: LightClientOptimisticUpdateTopic,
 	statefeed.Reorg:                       ChainReorgTopic,
 	statefeed.BlockProcessed:              BlockTopic,
+	statefeed.PayloadAttributes:           PayloadAttributesTopic,
 }
 
 var topicsForStateFeed = topicsForFeed(stateFeedEventTopics)
@@ -418,10 +421,9 @@ func topicForEvent(event *feed.Event) string {
 		return ChainReorgTopic
 	case *statefeed.BlockProcessedData:
 		return BlockTopic
+	case payloadattribute.EventData:
+		return PayloadAttributesTopic
 	default:
-		if event.Type == statefeed.MissedSlot {
-			return PayloadAttributesTopic
-		}
 		return InvalidTopic
 	}
 }
@@ -431,31 +433,17 @@ func (s *Server) lazyReaderForEvent(ctx context.Context, event *feed.Event, topi
 	if !topics.requested(eventName) {
 		return nil, errNotRequested
 	}
-	if eventName == PayloadAttributesTopic {
-		return s.currentPayloadAttributes(ctx)
-	}
 	if event == nil || event.Data == nil {
 		return nil, errors.New("event or event data is nil")
 	}
 	switch v := event.Data.(type) {
+	case payloadattribute.EventData:
+		return s.payloadAttributesReader(ctx, v)
 	case *ethpb.EventHead:
 		// The head event is a special case because, if the client requested the payload attributes topic,
 		// we send two event messages in reaction; the head event and the payload attributes.
-		headReader := func() io.Reader {
-			return jsonMarshalReader(eventName, structs.HeadEventFromV1(v))
-		}
-		// Don't do the expensive attr lookup unless the client requested it.
-		if !topics.requested(PayloadAttributesTopic) {
-			return headReader, nil
-		}
-		// Since payload attributes could change before the outbox is written, we need to do a blocking operation to
-		// get the current payload attributes right here.
-		attrReader, err := s.currentPayloadAttributes(ctx)
-		if err != nil {
-			return nil, errors.Wrap(err, "could not get payload attributes for head event")
-		}
 		return func() io.Reader {
-			return io.MultiReader(headReader(), attrReader())
+			return jsonMarshalReader(eventName, structs.HeadEventFromV1(v))
 		}, nil
 	case *operation.AggregatedAttReceivedData:
 		return func() io.Reader {
@@ -556,115 +544,202 @@ func (s *Server) lazyReaderForEvent(ctx context.Context, event *feed.Event, topi
 	}
 }
 
-// This event stream is intended to be used by builders and relays.
-// Parent fields are based on state at N_{current_slot}, while the rest of fields are based on state of N_{current_slot + 1}
-func (s *Server) currentPayloadAttributes(ctx context.Context) (lazyReader, error) {
-	headRoot, err := s.HeadFetcher.HeadRoot(ctx)
-	if err != nil {
-		return nil, errors.Wrap(err, "could not get head root")
-	}
-	st, err := s.HeadFetcher.HeadState(ctx)
-	if err != nil {
-		return nil, errors.Wrap(err, "could not get head state")
-	}
-	// advance the head state
-	headState, err := transition.ProcessSlotsIfPossible(ctx, st, s.ChainInfoFetcher.CurrentSlot()+1)
-	if err != nil {
-		return nil, errors.Wrap(err, "could not advance head state")
+var errUnsupportedPayloadAttribute = errors.New("cannot compute payload attributes pre-Bellatrix")
+
+func (s *Server) computePayloadAttributes(ctx context.Context, ev payloadattribute.EventData) (payloadattribute.Attributer, error) {
+	v := ev.HeadState.Version()
+	if v < version.Bellatrix {
+		return nil, errors.Wrapf(errUnsupportedPayloadAttribute, "%s is not supported", version.String(v))
 	}
 
-	headBlock, err := s.HeadFetcher.HeadBlock(ctx)
-	if err != nil {
-		return nil, errors.Wrap(err, "could not get head block")
-	}
-
-	headPayload, err := headBlock.Block().Body().Execution()
-	if err != nil {
-		return nil, errors.Wrap(err, "could not get execution payload")
-	}
-
-	t, err := slots.ToTime(headState.GenesisTime(), headState.Slot())
+	t, err := slots.ToTime(ev.HeadState.GenesisTime(), ev.HeadState.Slot())
 	if err != nil {
 		return nil, errors.Wrap(err, "could not get head state slot time")
 	}
-
-	prevRando, err := helpers.RandaoMix(headState, chaintime.CurrentEpoch(headState))
+	timestamp := uint64(t.Unix())
+	prevRando, err := helpers.RandaoMix(ev.HeadState, chaintime.CurrentEpoch(ev.HeadState))
 	if err != nil {
 		return nil, errors.Wrap(err, "could not get head state randao mix")
 	}
-
-	proposerIndex, err := helpers.BeaconProposerIndex(ctx, headState)
+	proposerIndex, err := helpers.BeaconProposerIndex(ctx, ev.HeadState)
 	if err != nil {
 		return nil, errors.Wrap(err, "could not get head state proposer index")
 	}
-	feeRecipient := params.BeaconConfig().DefaultFeeRecipient.Bytes()
+	feeRecpt := params.BeaconConfig().DefaultFeeRecipient.Bytes()
 	tValidator, exists := s.TrackedValidatorsCache.Validator(proposerIndex)
 	if exists {
-		feeRecipient = tValidator.FeeRecipient[:]
-	}
-	var attributes interface{}
-	switch headState.Version() {
-	case version.Bellatrix:
-		attributes = &structs.PayloadAttributesV1{
-			Timestamp:             fmt.Sprintf("%d", t.Unix()),
-			PrevRandao:            hexutil.Encode(prevRando),
-			SuggestedFeeRecipient: hexutil.Encode(feeRecipient),
-		}
-	case version.Capella:
-		withdrawals, _, err := headState.ExpectedWithdrawals()
-		if err != nil {
-			return nil, errors.Wrap(err, "could not get head state expected withdrawals")
-		}
-		attributes = &structs.PayloadAttributesV2{
-			Timestamp:             fmt.Sprintf("%d", t.Unix()),
-			PrevRandao:            hexutil.Encode(prevRando),
-			SuggestedFeeRecipient: hexutil.Encode(feeRecipient),
-			Withdrawals:           structs.WithdrawalsFromConsensus(withdrawals),
-		}
-	case version.Deneb, version.Electra:
-		withdrawals, _, err := headState.ExpectedWithdrawals()
-		if err != nil {
-			return nil, errors.Wrap(err, "could not get head state expected withdrawals")
-		}
-		parentRoot, err := headBlock.Block().HashTreeRoot()
-		if err != nil {
-			return nil, errors.Wrap(err, "could not get head block root")
-		}
-		attributes = &structs.PayloadAttributesV3{
-			Timestamp:             fmt.Sprintf("%d", t.Unix()),
-			PrevRandao:            hexutil.Encode(prevRando),
-			SuggestedFeeRecipient: hexutil.Encode(feeRecipient),
-			Withdrawals:           structs.WithdrawalsFromConsensus(withdrawals),
-			ParentBeaconBlockRoot: hexutil.Encode(parentRoot[:]),
-		}
-	default:
-		return nil, errors.Wrapf(err, "Payload version %s is not supported", version.String(headState.Version()))
+		feeRecpt = tValidator.FeeRecipient[:]
 	}
 
-	attributesBytes, err := json.Marshal(attributes)
-	if err != nil {
-		return nil, errors.Wrap(err, "errors marshaling payload attributes to json")
-	}
-	eventData := structs.PayloadAttributesEventData{
-		ProposerIndex:     fmt.Sprintf("%d", proposerIndex),
-		ProposalSlot:      fmt.Sprintf("%d", headState.Slot()),
-		ParentBlockNumber: fmt.Sprintf("%d", headPayload.BlockNumber()),
-		ParentBlockRoot:   hexutil.Encode(headRoot),
-		ParentBlockHash:   hexutil.Encode(headPayload.BlockHash()),
-		PayloadAttributes: attributesBytes,
-	}
-	eventDataBytes, err := json.Marshal(eventData)
-	if err != nil {
-		return nil, errors.Wrap(err, "errors marshaling payload attributes event data to json")
-	}
-	return func() io.Reader {
-		return jsonMarshalReader(PayloadAttributesTopic, &structs.PayloadAttributesEvent{
-			Version: version.String(headState.Version()),
-			Data:    eventDataBytes,
+	if v == version.Bellatrix {
+		return payloadattribute.New(&engine.PayloadAttributes{
+			Timestamp:             timestamp,
+			PrevRandao:            prevRando,
+			SuggestedFeeRecipient: feeRecpt,
 		})
+	}
+
+	w, _, err := ev.HeadState.ExpectedWithdrawals()
+	if err != nil {
+		return nil, errors.Wrap(err, "could not get withdrawals from head state")
+	}
+	if v == version.Capella {
+		return payloadattribute.New(&engine.PayloadAttributesV2{
+			Timestamp:             timestamp,
+			PrevRandao:            prevRando,
+			SuggestedFeeRecipient: feeRecpt,
+			Withdrawals:           w,
+		})
+	}
+
+	pr, err := ev.HeadBlock.Block().HashTreeRoot()
+	if err != nil {
+		return nil, errors.Wrap(err, "could not compute head block root")
+	}
+	return payloadattribute.New(&engine.PayloadAttributesV3{
+		Timestamp:             timestamp,
+		PrevRandao:            prevRando,
+		SuggestedFeeRecipient: feeRecpt,
+		Withdrawals:           w,
+		ParentBeaconBlockRoot: pr[:],
+	})
+}
+
+type asyncPayloadAttrData struct {
+	data    json.RawMessage
+	version string
+	err     error
+}
+
+func (s *Server) fillEventData(ctx context.Context, ev payloadattribute.EventData) (payloadattribute.EventData, error) {
+	if ev.HeadBlock == nil || ev.HeadBlock.IsNil() {
+		hb, err := s.HeadFetcher.HeadBlock(ctx)
+		if err != nil {
+			return ev, errors.Wrap(err, "Could not look up head block")
+		}
+		root, err := hb.Block().HashTreeRoot()
+		if err != nil {
+			return ev, errors.Wrap(err, "Could not compute head block root")
+		}
+		if ev.HeadRoot != root {
+			return ev, errors.Wrap(err, "head root changed before payload attribute event handler execution")
+		}
+		ev.HeadBlock = hb
+		payload, err := hb.Block().Body().Execution()
+		if err != nil {
+			return ev, errors.Wrap(err, "Could not get execution payload for head block")
+		}
+		ev.ParentBlockHash = payload.BlockHash()
+		ev.ParentBlockNumber = payload.BlockNumber()
+	}
+
+	attr := ev.Attributer
+	if attr == nil || attr.IsEmpty() {
+		attr, err := s.computePayloadAttributes(ctx, ev)
+		if err != nil {
+			return ev, errors.Wrap(err, "Could not compute payload attributes")
+		}
+		ev.Attributer = attr
+	}
+	return ev, nil
+}
+
+// This event stream is intended to be used by builders and relays.
+// Parent fields are based on state at N_{current_slot}, while the rest of fields are based on state of N_{current_slot + 1}
+func (s *Server) payloadAttributesReader(ctx context.Context, ev payloadattribute.EventData) (lazyReader, error) {
+	ctx, cancel := context.WithTimeout(ctx, payloadAttributeTimeout)
+	edc := make(chan asyncPayloadAttrData)
+	go func() {
+		d := asyncPayloadAttrData{
+			version: version.String(ev.HeadState.Version()),
+		}
+
+		defer func() {
+			edc <- d
+		}()
+		ev, err := s.fillEventData(ctx, ev)
+		if err != nil {
+			d.err = errors.Wrap(err, "Could not fill event data")
+			return
+		}
+		attributesBytes, err := marshalAttributes(ev.Attributer)
+		if err != nil {
+			d.err = errors.Wrap(err, "errors marshaling payload attributes to json")
+			return
+		}
+		d.data, d.err = json.Marshal(structs.PayloadAttributesEventData{
+			ProposerIndex:     strconv.FormatUint(uint64(ev.ProposerIndex), 10),
+			ProposalSlot:      strconv.FormatUint(uint64(ev.ProposalSlot), 10),
+			ParentBlockNumber: strconv.FormatUint(ev.ParentBlockNumber, 10),
+			ParentBlockRoot:   hexutil.Encode(ev.ParentBlockRoot),
+			ParentBlockHash:   hexutil.Encode(ev.ParentBlockHash),
+			PayloadAttributes: attributesBytes,
+		})
+		if d.err != nil {
+			d.err = errors.Wrap(d.err, "errors marshaling payload attributes event data to json")
+		}
+	}()
+	return func() io.Reader {
+		defer cancel()
+		select {
+		case <-ctx.Done():
+			log.WithError(ctx.Err()).Warn("Context canceled while waiting for payload attributes event data")
+			return nil
+		case ed := <-edc:
+			if ed.err != nil {
+				log.WithError(ed.err).Warn("Error while marshaling payload attributes event data")
+				return nil
+			}
+			return jsonMarshalReader(PayloadAttributesTopic, &structs.PayloadAttributesEvent{
+				Version: ed.version,
+				Data:    ed.data,
+			})
+		}
 	}, nil
 }
 
+func marshalAttributes(attr payloadattribute.Attributer) ([]byte, error) {
+	v := attr.Version()
+	if v < version.Bellatrix {
+		return nil, errors.Wrapf(errUnsupportedPayloadAttribute, "Payload version %s is not supported", version.String(v))
+	}
+
+	timestamp := strconv.FormatUint(attr.Timestamp(), 10)
+	prevRandao := hexutil.Encode(attr.PrevRandao())
+	feeRecpt := hexutil.Encode(attr.SuggestedFeeRecipient())
+	if v == version.Bellatrix {
+		return json.Marshal(&structs.PayloadAttributesV1{
+			Timestamp:             timestamp,
+			PrevRandao:            prevRandao,
+			SuggestedFeeRecipient: feeRecpt,
+		})
+	}
+	w, err := attr.Withdrawals()
+	if err != nil {
+		return nil, errors.Wrap(err, "could not get withdrawals from payload attributes event")
+	}
+	withdrawals := structs.WithdrawalsFromConsensus(w)
+	if v == version.Capella {
+		return json.Marshal(&structs.PayloadAttributesV2{
+			Timestamp:             timestamp,
+			PrevRandao:            prevRandao,
+			SuggestedFeeRecipient: feeRecpt,
+			Withdrawals:           withdrawals,
+		})
+	}
+	parentRoot, err := attr.ParentBeaconBlockRoot()
+	if err != nil {
+		return nil, errors.Wrap(err, "could not get parent beacon block root from payload attributes event")
+	}
+	return json.Marshal(&structs.PayloadAttributesV3{
+		Timestamp:             timestamp,
+		PrevRandao:            prevRandao,
+		SuggestedFeeRecipient: feeRecpt,
+		Withdrawals:           withdrawals,
+		ParentBeaconBlockRoot: hexutil.Encode(parentRoot),
+	})
+}
+
 func newStreamingResponseController(rw http.ResponseWriter, timeout time.Duration) *streamingResponseWriterController {
 	rc := http.NewResponseController(rw)
 	return &streamingResponseWriterController{

beacon-chain/rpc/eth/events/events_test.go

@@ -21,6 +21,7 @@ import (
 	fieldparams "github.com/prysmaticlabs/prysm/v5/config/fieldparams"
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/blocks"
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/interfaces"
+	payloadattribute "github.com/prysmaticlabs/prysm/v5/consensus-types/payload-attribute"
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/primitives"
 	ethpb "github.com/prysmaticlabs/prysm/v5/proto/eth/v1"
 	eth "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
@@ -489,7 +490,21 @@ func TestStreamEvents_OperationsEvents(t *testing.T) {
 				require.NoError(t, err)
 				request := topics.testHttpRequest(testSync.ctx, t)
 				w := NewStreamingResponseWriterRecorder(testSync.ctx)
-				events := []*feed.Event{&feed.Event{Type: statefeed.MissedSlot}}
+				events := []*feed.Event{
+					&feed.Event{
+						Type: statefeed.PayloadAttributes,
+						Data: payloadattribute.EventData{
+							ProposerIndex:     0,
+							ProposalSlot:      0,
+							ParentBlockNumber: 0,
+							ParentBlockRoot:   make([]byte, 32),
+							ParentBlockHash:   make([]byte, 32),
+							HeadState:         st,
+							HeadBlock:         b,
+							HeadRoot:          [fieldparams.RootLength]byte{},
+						},
+					},
+				}
 
 				go func() {
 					s.StreamEvents(w, request)

consensus-types/payload-attribute/BUILD.bazel

@@ -10,8 +10,12 @@ go_library(
     importpath = "github.com/prysmaticlabs/prysm/v5/consensus-types/payload-attribute",
     visibility = ["//visibility:public"],
     deps = [
+        "//beacon-chain/state:go_default_library",
+        "//config/fieldparams:go_default_library",
         "//consensus-types:go_default_library",
         "//consensus-types/blocks:go_default_library",
+        "//consensus-types/interfaces:go_default_library",
+        "//consensus-types/primitives:go_default_library",
         "//proto/engine/v1:go_default_library",
         "//runtime/version:go_default_library",
         "@com_github_pkg_errors//:go_default_library",

consensus-types/payload-attribute/getters.go

@@ -38,6 +38,16 @@ func (a *data) Withdrawals() ([]*enginev1.Withdrawal, error) {
 	return a.withdrawals, nil
 }
 
+func (a *data) ParentBeaconBlockRoot() ([]byte, error) {
+	if len(a.parentBeaconBlockRoot) == 0 {
+		return nil, errNoParentRoot
+	}
+	if a.version < version.Deneb {
+		return nil, consensus_types.ErrNotSupported("ParentBeaconBlockRoot", a.version)
+	}
+	return a.parentBeaconBlockRoot, nil
+}
+
 // PbV1 returns the payload attribute in version 1.
 func (a *data) PbV1() (*enginev1.PayloadAttributes, error) {
 	if a == nil {
@@ -97,6 +107,9 @@ func (a *data) PbV3() (*enginev1.PayloadAttributesV3, error) {
 
 // IsEmpty returns whether the given payload attribute is empty
 func (a *data) IsEmpty() bool {
+	if a == nil {
+		return true
+	}
 	if len(a.PrevRandao()) != 0 {
 		return false
 	}

consensus-types/payload-attribute/interface.go

@@ -10,6 +10,7 @@ type Attributer interface {
 	Timestamp() uint64
 	SuggestedFeeRecipient() []byte
 	Withdrawals() ([]*enginev1.Withdrawal, error)
+	ParentBeaconBlockRoot() ([]byte, error)
 	PbV1() (*enginev1.PayloadAttributes, error)
 	PbV2() (*enginev1.PayloadAttributesV2, error)
 	PbV3() (*enginev1.PayloadAttributesV3, error)

consensus-types/payload-attribute/types.go

@@ -2,7 +2,11 @@ package payloadattribute
 
 import (
 	"github.com/pkg/errors"
+	"github.com/prysmaticlabs/prysm/v5/beacon-chain/state"
+	field_params "github.com/prysmaticlabs/prysm/v5/config/fieldparams"
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/blocks"
+	"github.com/prysmaticlabs/prysm/v5/consensus-types/interfaces"
+	"github.com/prysmaticlabs/prysm/v5/consensus-types/primitives"
 	enginev1 "github.com/prysmaticlabs/prysm/v5/proto/engine/v1"
 	"github.com/prysmaticlabs/prysm/v5/runtime/version"
 )
@@ -23,6 +27,7 @@ type data struct {
 var (
 	errNilPayloadAttribute         = errors.New("received nil payload attribute")
 	errUnsupportedPayloadAttribute = errors.New("unsupported payload attribute")
+	errNoParentRoot                = errors.New("parent root is empty")
 )
 
 // New returns a new payload attribute with the given input object.
@@ -89,3 +94,16 @@ func initPayloadAttributeFromV3(a *enginev1.PayloadAttributesV3) (Attributer, er
 		parentBeaconBlockRoot: a.ParentBeaconBlockRoot,
 	}, nil
 }
+
+// EventData holds the values for a PayloadAttributes event.
+type EventData struct {
+	ProposerIndex     primitives.ValidatorIndex
+	ProposalSlot      primitives.Slot
+	ParentBlockNumber uint64
+	ParentBlockRoot   []byte
+	ParentBlockHash   []byte
+	Attributer        Attributer
+	HeadState         state.BeaconState
+	HeadBlock         interfaces.ReadOnlySignedBeaconBlock
+	HeadRoot          [field_params.RootLength]byte
+}