Signed execution payload header for sync (#14363)

* Signed execution payload header for sync

* Use RO state

* SignedExecutionPayloadHeader by hash and root

* Fix execution headers cache

beacon-chain/blockchain/testing/mock.go

@@ -708,3 +708,8 @@ func (c *ChainService) ReceiveBlob(_ context.Context, b blocks.VerifiedROBlob) e
 func (c *ChainService) TargetRootForEpoch(_ [32]byte, _ primitives.Epoch) ([32]byte, error) {
 	return c.TargetRoot, nil
 }
+
+// HashInForkchoice mocks the same method in the chain service
+func (c *ChainService) HashInForkchoice([32]byte) bool {
+	return false
+}

beacon-chain/cache/BUILD.bazel

@@ -21,6 +21,7 @@ go_library(
         "proposer_indices_disabled.go",  # keep
         "proposer_indices_type.go",
         "registration.go",
+        "signed_execution_header.go",
         "skip_slot_cache.go",
         "subnet_ids.go",
         "sync_committee.go",
@@ -49,6 +50,7 @@ go_library(
         "//encoding/bytesutil:go_default_library",
         "//math:go_default_library",
         "//monitoring/tracing/trace:go_default_library",
+        "//proto/engine/v1:go_default_library",
         "//proto/prysm/v1alpha1:go_default_library",
         "//runtime/version:go_default_library",
         "@com_github_ethereum_go_ethereum//common:go_default_library",
@@ -77,6 +79,7 @@ go_test(
         "private_access_test.go",
         "proposer_indices_test.go",
         "registration_test.go",
+        "signed_execution_header_test.go",
         "skip_slot_cache_test.go",
         "subnet_ids_test.go",
         "sync_committee_head_state_test.go",
@@ -93,6 +96,7 @@ go_test(
         "//consensus-types/primitives:go_default_library",
         "//crypto/bls:go_default_library",
         "//encoding/bytesutil:go_default_library",
+        "//proto/engine/v1:go_default_library",
         "//proto/prysm/v1alpha1:go_default_library",
         "//testing/assert:go_default_library",
         "//testing/require:go_default_library",

beacon-chain/cache/signed_execution_header.go

@@ -0,0 +1,76 @@
+package cache
+
+import (
+	"bytes"
+	"sync"
+
+	"github.com/prysmaticlabs/prysm/v5/consensus-types/primitives"
+	enginev1 "github.com/prysmaticlabs/prysm/v5/proto/engine/v1"
+)
+
+// ExecutionPayloadHeaders is used  by the sync service to store signed execution payload headers after they pass validation,
+// and filter out subsequent headers with lower value.
+// The signed header from this cache could be used by the proposer when proposing the next slot.
+type ExecutionPayloadHeaders struct {
+	headers map[primitives.Slot][]*enginev1.SignedExecutionPayloadHeader
+	sync.RWMutex
+}
+
+func NewExecutionPayloadHeaders() *ExecutionPayloadHeaders {
+	return &ExecutionPayloadHeaders{
+		headers: make(map[primitives.Slot][]*enginev1.SignedExecutionPayloadHeader),
+	}
+}
+
+// SaveSignedExecutionPayloadHeader saves the signed execution payload header to the cache.
+// The cache stores headers for up to two slots. If the input slot is higher than the lowest slot
+// currently in the cache, the lowest slot is removed to make space for the new header.
+// Only the highest value header for a given parent block hash will be stored.
+// This function assumes caller already checks header's slot is current or next slot, it doesn't account for slot validation.
+func (c *ExecutionPayloadHeaders) SaveSignedExecutionPayloadHeader(header *enginev1.SignedExecutionPayloadHeader) {
+	c.Lock()
+	defer c.Unlock()
+
+	for s := range c.headers {
+		if s+1 < header.Message.Slot {
+			delete(c.headers, s)
+		}
+	}
+
+	// Add or update the header in the map
+	if _, ok := c.headers[header.Message.Slot]; !ok {
+		c.headers[header.Message.Slot] = []*enginev1.SignedExecutionPayloadHeader{header}
+	} else {
+		found := false
+		for i, h := range c.headers[header.Message.Slot] {
+			if bytes.Equal(h.Message.ParentBlockHash, header.Message.ParentBlockHash) && bytes.Equal(h.Message.ParentBlockRoot, header.Message.ParentBlockRoot) {
+				if header.Message.Value > h.Message.Value {
+					c.headers[header.Message.Slot][i] = header
+				}
+				found = true
+				break
+			}
+		}
+		if !found {
+			c.headers[header.Message.Slot] = append(c.headers[header.Message.Slot], header)
+		}
+	}
+}
+
+// SignedExecutionPayloadHeader returns the signed payload header for the given slot and parent block hash and block root.
+// Returns nil if the header is not found.
+// This should be used when the caller wants the header to match parent block hash and parent block root such as proposer choosing a header to propose.
+func (c *ExecutionPayloadHeaders) SignedExecutionPayloadHeader(slot primitives.Slot, parentBlockHash []byte, parentBlockRoot []byte) *enginev1.SignedExecutionPayloadHeader {
+	c.RLock()
+	defer c.RUnlock()
+
+	if headers, ok := c.headers[slot]; ok {
+		for _, header := range headers {
+			if bytes.Equal(header.Message.ParentBlockHash, parentBlockHash) && bytes.Equal(header.Message.ParentBlockRoot, parentBlockRoot) {
+				return header
+			}
+		}
+	}
+
+	return nil
+}

beacon-chain/cache/signed_execution_header_test.go

@@ -0,0 +1,243 @@
+package cache
+
+import (
+	"testing"
+
+	enginev1 "github.com/prysmaticlabs/prysm/v5/proto/engine/v1"
+	"github.com/prysmaticlabs/prysm/v5/testing/require"
+)
+
+func Test_SaveSignedExecutionPayloadHeader(t *testing.T) {
+	t.Run("First header should be added to cache", func(t *testing.T) {
+		c := NewExecutionPayloadHeaders()
+		header := &enginev1.SignedExecutionPayloadHeader{
+			Message: &enginev1.ExecutionPayloadHeaderEPBS{
+				Slot:            1,
+				ParentBlockHash: []byte("parent1"),
+				Value:           100,
+			},
+		}
+		c.SaveSignedExecutionPayloadHeader(header)
+		require.Equal(t, 1, len(c.headers))
+		require.Equal(t, header, c.headers[1][0])
+	})
+
+	t.Run("Second header with higher slot should be added, and both slots should be in cache", func(t *testing.T) {
+		c := NewExecutionPayloadHeaders()
+		header1 := &enginev1.SignedExecutionPayloadHeader{
+			Message: &enginev1.ExecutionPayloadHeaderEPBS{
+				Slot:            1,
+				ParentBlockHash: []byte("parent1"),
+				Value:           100,
+			},
+		}
+		header2 := &enginev1.SignedExecutionPayloadHeader{
+			Message: &enginev1.ExecutionPayloadHeaderEPBS{
+				Slot:            2,
+				ParentBlockHash: []byte("parent2"),
+				Value:           100,
+			},
+		}
+		c.SaveSignedExecutionPayloadHeader(header1)
+		c.SaveSignedExecutionPayloadHeader(header2)
+		require.Equal(t, 2, len(c.headers))
+		require.Equal(t, header1, c.headers[1][0])
+		require.Equal(t, header2, c.headers[2][0])
+	})
+
+	t.Run("Third header with higher slot should replace the oldest slot", func(t *testing.T) {
+		c := NewExecutionPayloadHeaders()
+		header1 := &enginev1.SignedExecutionPayloadHeader{
+			Message: &enginev1.ExecutionPayloadHeaderEPBS{
+				Slot:            1,
+				ParentBlockHash: []byte("parent1"),
+				Value:           100,
+			},
+		}
+		header2 := &enginev1.SignedExecutionPayloadHeader{
+			Message: &enginev1.ExecutionPayloadHeaderEPBS{
+				Slot:            2,
+				ParentBlockHash: []byte("parent2"),
+				Value:           100,
+			},
+		}
+		header3 := &enginev1.SignedExecutionPayloadHeader{
+			Message: &enginev1.ExecutionPayloadHeaderEPBS{
+				Slot:            3,
+				ParentBlockHash: []byte("parent3"),
+				Value:           100,
+			},
+		}
+		c.SaveSignedExecutionPayloadHeader(header1)
+		c.SaveSignedExecutionPayloadHeader(header2)
+		c.SaveSignedExecutionPayloadHeader(header3)
+		require.Equal(t, 2, len(c.headers))
+		require.Equal(t, header2, c.headers[2][0])
+		require.Equal(t, header3, c.headers[3][0])
+	})
+
+	t.Run("Header with same slot but higher value should replace the existing one", func(t *testing.T) {
+		c := NewExecutionPayloadHeaders()
+		header1 := &enginev1.SignedExecutionPayloadHeader{
+			Message: &enginev1.ExecutionPayloadHeaderEPBS{
+				Slot:            2,
+				ParentBlockHash: []byte("parent2"),
+				Value:           100,
+			},
+		}
+		header2 := &enginev1.SignedExecutionPayloadHeader{
+			Message: &enginev1.ExecutionPayloadHeaderEPBS{
+				Slot:            2,
+				ParentBlockHash: []byte("parent2"),
+				Value:           200,
+			},
+		}
+		c.SaveSignedExecutionPayloadHeader(header1)
+		c.SaveSignedExecutionPayloadHeader(header2)
+		require.Equal(t, 1, len(c.headers[2]))
+		require.Equal(t, header2, c.headers[2][0])
+	})
+
+	t.Run("Header with different parent block hash should be appended to the same slot", func(t *testing.T) {
+		c := NewExecutionPayloadHeaders()
+		header1 := &enginev1.SignedExecutionPayloadHeader{
+			Message: &enginev1.ExecutionPayloadHeaderEPBS{
+				Slot:            2,
+				ParentBlockHash: []byte("parent1"),
+				Value:           100,
+			},
+		}
+		header2 := &enginev1.SignedExecutionPayloadHeader{
+			Message: &enginev1.ExecutionPayloadHeaderEPBS{
+				Slot:            2,
+				ParentBlockHash: []byte("parent2"),
+				Value:           200,
+			},
+		}
+		c.SaveSignedExecutionPayloadHeader(header1)
+		c.SaveSignedExecutionPayloadHeader(header2)
+		require.Equal(t, 2, len(c.headers[2]))
+		require.Equal(t, header1, c.headers[2][0])
+		require.Equal(t, header2, c.headers[2][1])
+	})
+}
+
+func TestSignedExecutionPayloadHeader(t *testing.T) {
+	t.Run("Return header when slot and parentBlockHash match", func(t *testing.T) {
+		c := NewExecutionPayloadHeaders()
+		header := &enginev1.SignedExecutionPayloadHeader{
+			Message: &enginev1.ExecutionPayloadHeaderEPBS{
+				Slot:            1,
+				ParentBlockHash: []byte("parent1"),
+				ParentBlockRoot: []byte("root1"),
+				Value:           100,
+			},
+		}
+		c.SaveSignedExecutionPayloadHeader(header)
+		result := c.SignedExecutionPayloadHeader(1, []byte("parent1"), []byte("root1"))
+		require.NotNil(t, result)
+		require.Equal(t, header, result)
+	})
+
+	t.Run("Return nil when no matching slot and parentBlockHash", func(t *testing.T) {
+		c := NewExecutionPayloadHeaders()
+		header := &enginev1.SignedExecutionPayloadHeader{
+			Message: &enginev1.ExecutionPayloadHeaderEPBS{
+				Slot:            1,
+				ParentBlockHash: []byte("parent1"),
+				ParentBlockRoot: []byte("root1"),
+				Value:           100,
+			},
+		}
+		c.SaveSignedExecutionPayloadHeader(header)
+		result := c.SignedExecutionPayloadHeader(2, []byte("parent2"), []byte("root1"))
+		require.IsNil(t, result)
+	})
+
+	t.Run("Return nil when no matching slot and parentBlockRoot", func(t *testing.T) {
+		c := NewExecutionPayloadHeaders()
+		header := &enginev1.SignedExecutionPayloadHeader{
+			Message: &enginev1.ExecutionPayloadHeaderEPBS{
+				Slot:            1,
+				ParentBlockHash: []byte("parent1"),
+				ParentBlockRoot: []byte("root1"),
+				Value:           100,
+			},
+		}
+		c.SaveSignedExecutionPayloadHeader(header)
+		result := c.SignedExecutionPayloadHeader(2, []byte("parent1"), []byte("root2"))
+		require.IsNil(t, result)
+	})
+
+	t.Run("Return header when there are two slots in the cache and a match is found", func(t *testing.T) {
+		c := NewExecutionPayloadHeaders()
+		header1 := &enginev1.SignedExecutionPayloadHeader{
+			Message: &enginev1.ExecutionPayloadHeaderEPBS{
+				Slot:            1,
+				ParentBlockHash: []byte("parent1"),
+				Value:           100,
+			},
+		}
+		header2 := &enginev1.SignedExecutionPayloadHeader{
+			Message: &enginev1.ExecutionPayloadHeaderEPBS{
+				Slot:            2,
+				ParentBlockHash: []byte("parent2"),
+				Value:           200,
+			},
+		}
+		c.SaveSignedExecutionPayloadHeader(header1)
+		c.SaveSignedExecutionPayloadHeader(header2)
+
+		// Check for the first header
+		result1 := c.SignedExecutionPayloadHeader(1, []byte("parent1"), []byte{})
+		require.NotNil(t, result1)
+		require.Equal(t, header1, result1)
+
+		// Check for the second header
+		result2 := c.SignedExecutionPayloadHeader(2, []byte("parent2"), []byte{})
+		require.NotNil(t, result2)
+		require.Equal(t, header2, result2)
+	})
+
+	t.Run("Return nil when slot is evicted from cache", func(t *testing.T) {
+		c := NewExecutionPayloadHeaders()
+		header1 := &enginev1.SignedExecutionPayloadHeader{
+			Message: &enginev1.ExecutionPayloadHeaderEPBS{
+				Slot:            1,
+				ParentBlockHash: []byte("parent1"),
+				Value:           100,
+			},
+		}
+		header2 := &enginev1.SignedExecutionPayloadHeader{
+			Message: &enginev1.ExecutionPayloadHeaderEPBS{
+				Slot:            2,
+				ParentBlockHash: []byte("parent2"),
+				Value:           200,
+			},
+		}
+		header3 := &enginev1.SignedExecutionPayloadHeader{
+			Message: &enginev1.ExecutionPayloadHeaderEPBS{
+				Slot:            3,
+				ParentBlockHash: []byte("parent3"),
+				Value:           300,
+			},
+		}
+		c.SaveSignedExecutionPayloadHeader(header1)
+		c.SaveSignedExecutionPayloadHeader(header2)
+		c.SaveSignedExecutionPayloadHeader(header3)
+
+		// The first slot should be evicted, so result should be nil
+		result := c.SignedExecutionPayloadHeader(1, []byte("parent1"), []byte{})
+		require.IsNil(t, result)
+
+		// The second slot should still be present
+		result = c.SignedExecutionPayloadHeader(2, []byte("parent2"), []byte{})
+		require.NotNil(t, result)
+		require.Equal(t, header2, result)
+
+		// The third slot should be present
+		result = c.SignedExecutionPayloadHeader(3, []byte("parent3"), []byte{})
+		require.NotNil(t, result)
+		require.Equal(t, header3, result)
+	})
+}

beacon-chain/node/node.go

@@ -103,6 +103,7 @@ type BeaconNode struct {
 	trackedValidatorsCache  *cache.TrackedValidatorsCache
 	payloadAttestationCache *cache.PayloadAttestationCache
 	payloadEnvelopeCache    *sync.Map
+	executionHeaderCache    *cache.ExecutionPayloadHeaders
 	payloadIDCache          *cache.PayloadIDCache
 	stateFeed               *event.Feed
 	blockFeed               *event.Feed
@@ -155,6 +156,7 @@ func New(cliCtx *cli.Context, cancel context.CancelFunc, opts ...Option) (*Beaco
 		trackedValidatorsCache:  cache.NewTrackedValidatorsCache(),
 		payloadAttestationCache: &cache.PayloadAttestationCache{},
 		payloadEnvelopeCache:    &sync.Map{},
+		executionHeaderCache:    cache.NewExecutionPayloadHeaders(),
 		payloadIDCache:          cache.NewPayloadIDCache(),
 		slasherBlockHeadersFeed: new(event.Feed),
 		slasherAttestationsFeed: new(event.Feed),
@@ -846,6 +848,7 @@ func (b *BeaconNode) registerSyncService(initialSyncComplete chan struct{}, bFil
 		regularsync.WithSlasherAttestationsFeed(b.slasherAttestationsFeed),
 		regularsync.WithSlasherBlockHeadersFeed(b.slasherBlockHeadersFeed),
 		regularsync.WithPayloadAttestationCache(b.payloadAttestationCache),
+		regularsync.WithExecutionPayloadHeaderCache(b.executionHeaderCache),
 		regularsync.WithPayloadEnvelopeCache(b.payloadEnvelopeCache),
 		regularsync.WithPayloadReconstructor(web3Service),
 		regularsync.WithClockWaiter(b.clockWaiter),