Add Blob Gossip (#12413)

beacon-chain/sync/pending_blocks_queue.go

@@ -19,6 +19,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v4/encoding/bytesutil"
 	"github.com/prysmaticlabs/prysm/v4/encoding/ssz/equality"
 	"github.com/prysmaticlabs/prysm/v4/monitoring/tracing"
+	eth "github.com/prysmaticlabs/prysm/v4/proto/prysm/v1alpha1"
 	"github.com/prysmaticlabs/prysm/v4/time/slots"
 	"github.com/sirupsen/logrus"
 	"github.com/trailofbits/go-mutexasserts"
@@ -427,6 +428,21 @@ func (s *Service) pendingBlocksInCache(slot primitives.Slot) []interfaces.ReadOn
 	return blks
 }
 
+// This returns signed blob sidecar given input key from slotToPendingBlobs.
+func (s *Service) pendingBlobsInCache(slot primitives.Slot) []*eth.SignedBlobSidecar {
+	k := slotToCacheKey(slot)
+	value, ok := s.slotToPendingBlobs.Get(k)
+	if !ok {
+		return []*eth.SignedBlobSidecar{}
+	}
+	bs, ok := value.([]*eth.SignedBlobSidecar)
+	if !ok {
+		log.Debug("pendingBlobsInCache: value is not of type []*eth.SignedBlobSidecar")
+		return []*eth.SignedBlobSidecar{}
+	}
+	return bs
+}
+
 // This adds input signed beacon block to slotToPendingBlocks cache.
 func (s *Service) addPendingBlockToCache(b interfaces.ReadOnlySignedBeaconBlock) error {
 	if err := blocks.BeaconBlockIsNil(b); err != nil {
@@ -445,6 +461,23 @@ func (s *Service) addPendingBlockToCache(b interfaces.ReadOnlySignedBeaconBlock)
 	return nil
 }
 
+// This adds blob to slotToPendingBlobs cache.
+func (s *Service) addPendingBlobToCache(b *eth.SignedBlobSidecar) error {
+	blobs := s.pendingBlobsInCache(b.Message.Slot)
+
+	// If we already have seen the index. Ignore it.
+	for _, blob := range blobs {
+		if blob.Message.Index == b.Message.Index {
+			return nil
+		}
+	}
+
+	blobs = append(blobs, b)
+	k := slotToCacheKey(b.Message.Slot)
+	s.slotToPendingBlobs.Set(k, blobs, pendingBlockExpTime)
+	return nil
+}
+
 // This converts input string to slot.
 func cacheKeyToSlot(s string) primitives.Slot {
 	b := []byte(s)