Add in-progress

beacon-chain/cache/BUILD.bazel

@@ -33,6 +33,7 @@ go_library(
         "//tools:__subpackages__",
     ],
     deps = [
+        "//beacon-chain/forkchoice/types:go_default_library",
         "//beacon-chain/state:go_default_library",
         "//cache/lru:go_default_library",
         "//config/fieldparams:go_default_library",
@@ -78,6 +79,7 @@ go_test(
     ],
     embed = [":go_default_library"],
     deps = [
+        "//beacon-chain/forkchoice/types:go_default_library",
         "//beacon-chain/state:go_default_library",
         "//beacon-chain/state/state-native:go_default_library",
         "//config/fieldparams:go_default_library",

beacon-chain/cache/attestation_data.go

@@ -5,16 +5,15 @@ import (
 	"errors"
 	"sync"
 
+	forkchoicetypes "github.com/prysmaticlabs/prysm/v4/beacon-chain/forkchoice/types"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
 )
 
 type AttestationConsensusData struct {
-	Slot        primitives.Slot
-	HeadRoot    []byte
-	TargetRoot  []byte
-	TargetEpoch primitives.Epoch
-	SourceRoot  []byte
-	SourceEpoch primitives.Epoch
+	Slot     primitives.Slot
+	HeadRoot []byte
+	Target   forkchoicetypes.Checkpoint
+	Source   forkchoicetypes.Checkpoint
 }
 
 // AttestationCache stores cached results of AttestationData requests.
@@ -28,24 +27,36 @@ func NewAttestationCache() *AttestationCache {
 	return &AttestationCache{}
 }
 
-// Get retrieves cached attestation data, recording a cache hit or miss.
+// Get retrieves cached attestation data, recording a cache hit or miss. This method is lock free.
 func (c *AttestationCache) Get(ctx context.Context) (*AttestationConsensusData, error) {
-	c.lock.RLock()
-	defer c.lock.RUnlock()
-
 	return c.a, nil
 }
 
-// Put adds a response to the cache.
+// Put adds a response to the cache. This method is lock free.
 func (c *AttestationCache) Put(ctx context.Context, a *AttestationConsensusData) error {
 	if a == nil {
 		return errors.New("attestation cannot be nil")
 	}
+	c.a = a
+	return nil
+}
 
+// Lock locks the cache for writing.
+func (c *AttestationCache) Lock() {
 	c.lock.Lock()
-	defer c.lock.Unlock()
+}
 
-	c.a = a
+// Unlock unlocks the cache for writing.
+func (c *AttestationCache) Unlock() {
+	c.lock.Unlock()
+}
 
-	return nil
+// RLock locks the cache for reading.
+func (c *AttestationCache) RLock() {
+	c.lock.RLock()
+}
+
+// RUnlock unlocks the cache for reading.
+func (c *AttestationCache) RUnlock() {
+	c.lock.RUnlock()
 }

beacon-chain/cache/attestation_data_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/cache"
+	forkchoicetypes "github.com/prysmaticlabs/prysm/v4/beacon-chain/forkchoice/types"
 	"github.com/stretchr/testify/require"
 )
 
@@ -17,12 +18,16 @@ func TestAttestationCache_RoundTrip(t *testing.T) {
 	require.Nil(t, a)
 
 	insert := &cache.AttestationConsensusData{
-		Slot:        1,
-		HeadRoot:    []byte{1},
-		TargetRoot:  []byte{2},
-		TargetEpoch: 3,
-		SourceRoot:  []byte{4},
-		SourceEpoch: 5,
+		Slot:     1,
+		HeadRoot: []byte{1},
+		Target: forkchoicetypes.Checkpoint{
+			Epoch: 2,
+			Root:  [32]byte{3},
+		},
+		Source: forkchoicetypes.Checkpoint{
+			Epoch: 4,
+			Root:  [32]byte{5},
+		},
 	}
 	err = c.Put(ctx, insert)
 	require.NoError(t, err)
@@ -32,12 +37,16 @@ func TestAttestationCache_RoundTrip(t *testing.T) {
 	require.Equal(t, insert, a)
 
 	insert = &cache.AttestationConsensusData{
-		Slot:        6,
-		HeadRoot:    []byte{7},
-		TargetRoot:  []byte{8},
-		TargetEpoch: 9,
-		SourceRoot:  []byte{10},
-		SourceEpoch: 11,
+		Slot:     6,
+		HeadRoot: []byte{7},
+		Target: forkchoicetypes.Checkpoint{
+			Epoch: 8,
+			Root:  [32]byte{9},
+		},
+		Source: forkchoicetypes.Checkpoint{
+			Epoch: 10,
+			Root:  [32]byte{11},
+		},
 	}
 
 	err = c.Put(ctx, insert)

beacon-chain/rpc/core/BUILD.bazel

@@ -20,6 +20,7 @@ go_library(
         "//beacon-chain/core/helpers:go_default_library",
         "//beacon-chain/core/time:go_default_library",
         "//beacon-chain/core/transition:go_default_library",
+        "//beacon-chain/forkchoice/types:go_default_library",
         "//beacon-chain/operations/synccommittee:go_default_library",
         "//beacon-chain/p2p:go_default_library",
         "//beacon-chain/state:go_default_library",

beacon-chain/rpc/core/validator.go

@@ -16,6 +16,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/core/helpers"
 	coreTime "github.com/prysmaticlabs/prysm/v4/beacon-chain/core/time"
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/core/transition"
+	forkchoicetypes "github.com/prysmaticlabs/prysm/v4/beacon-chain/forkchoice/types"
 	beaconState "github.com/prysmaticlabs/prysm/v4/beacon-chain/state"
 	fieldparams "github.com/prysmaticlabs/prysm/v4/config/fieldparams"
 	"github.com/prysmaticlabs/prysm/v4/config/params"
@@ -326,22 +327,51 @@ func (s *Service) GetAttestationData(
 		return nil, &RpcError{Reason: BadRequest, Err: errors.Errorf("invalid request: slot %d is not the current slot %d", req.Slot, s.GenesisTimeFetcher.CurrentSlot())}
 	}
 
+	s.AttestationCache.RLock()
 	res, err := s.AttestationCache.Get(ctx)
 	if err != nil {
 		return nil, &RpcError{Reason: Internal, Err: errors.Errorf("could not retrieve data from attestation cache: %v", err)}
 	}
 	if res != nil && res.Slot == req.Slot {
+		s.AttestationCache.RUnlock()
 		return &ethpb.AttestationData{
 			Slot:            res.Slot,
 			CommitteeIndex:  req.CommitteeIndex,
 			BeaconBlockRoot: res.HeadRoot,
 			Source: &ethpb.Checkpoint{
-				Epoch: res.SourceEpoch,
-				Root:  res.SourceRoot,
+				Epoch: res.Source.Epoch,
+				Root:  res.Source.Root[:],
 			},
 			Target: &ethpb.Checkpoint{
-				Epoch: res.TargetEpoch,
-				Root:  res.TargetRoot,
+				Epoch: res.Target.Epoch,
+				Root:  res.Target.Root[:],
+			},
+		}, nil
+	}
+	s.AttestationCache.RUnlock()
+
+	s.AttestationCache.Lock()
+	defer s.AttestationCache.Unlock()
+
+	// We check the cache again as in the event there are multiple inflight requests for
+	// the same attestation data, the cache might have been filled while we were waiting
+	// to acquire the lock.
+	res, err = s.AttestationCache.Get(ctx)
+	if err != nil {
+		return nil, &RpcError{Reason: Internal, Err: errors.Errorf("could not retrieve data from attestation cache: %v", err)}
+	}
+	if res != nil && res.Slot == req.Slot {
+		return &ethpb.AttestationData{
+			Slot:            res.Slot,
+			CommitteeIndex:  req.CommitteeIndex,
+			BeaconBlockRoot: res.HeadRoot,
+			Source: &ethpb.Checkpoint{
+				Epoch: res.Source.Epoch,
+				Root:  res.Source.Root[:],
+			},
+			Target: &ethpb.Checkpoint{
+				Epoch: res.Target.Epoch,
+				Root:  res.Target.Root[:],
 			},
 		}, nil
 	}
@@ -357,12 +387,16 @@ func (s *Service) GetAttestationData(
 	}
 	justifiedCheckpoint := s.FinalizedFetcher.CurrentJustifiedCheckpt()
 	if err = s.AttestationCache.Put(ctx, &cache.AttestationConsensusData{
-		Slot:        req.Slot,
-		HeadRoot:    headRoot,
-		TargetRoot:  targetRoot[:],
-		TargetEpoch: targetEpoch,
-		SourceRoot:  justifiedCheckpoint.Root,
-		SourceEpoch: justifiedCheckpoint.Epoch,
+		Slot:     req.Slot,
+		HeadRoot: headRoot,
+		Target: forkchoicetypes.Checkpoint{
+			Epoch: targetEpoch,
+			Root:  targetRoot,
+		},
+		Source: forkchoicetypes.Checkpoint{
+			Epoch: justifiedCheckpoint.Epoch,
+			Root:  bytesutil.ToBytes32(justifiedCheckpoint.Root),
+		},
 	}); err != nil {
 		log.WithError(err).Error("Failed to put attestation data into cache")
 	}