Beacon-api: broadcast blobs in the event of seen block (#13830)

* Beacon-api: broadcast blobs in the event of seen block

* Fix parameters

* Fix test

* Check forkchoice

* Ran go format

* Revert "Ran go format"

This reverts commit 091e77e.

* James feedback

* Radek's feedback

Co-authored-by: Radosław Kapka <rkapka@wp.pl>

* Fix bad tests

---------

Co-authored-by: Radosław Kapka <rkapka@wp.pl>

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

@@ -37,6 +37,7 @@ go_library(
         "//beacon-chain/rpc/eth/helpers:go_default_library",
         "//beacon-chain/rpc/eth/shared:go_default_library",
         "//beacon-chain/rpc/lookup:go_default_library",
+        "//beacon-chain/rpc/prysm/v1alpha1/validator:go_default_library",
         "//beacon-chain/state:go_default_library",
         "//beacon-chain/state/state-native:go_default_library",
         "//beacon-chain/state/stategen:go_default_library",
@@ -121,6 +122,7 @@ go_test(
         "@com_github_gorilla_mux//:go_default_library",
         "@com_github_pkg_errors//:go_default_library",
         "@com_github_prysmaticlabs_go_bitfield//:go_default_library",
+        "@com_github_sirupsen_logrus//hooks/test:go_default_library",
         "@com_github_stretchr_testify//mock:go_default_library",
         "@org_uber_go_mock//gomock:go_default_library",
     ],

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

@@ -21,6 +21,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/db/filters"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/rpc/eth/helpers"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/rpc/eth/shared"
+	"github.com/prysmaticlabs/prysm/v5/beacon-chain/rpc/prysm/v1alpha1/validator"
 	fieldparams "github.com/prysmaticlabs/prysm/v5/config/fieldparams"
 	"github.com/prysmaticlabs/prysm/v5/config/params"
 	consensus_types "github.com/prysmaticlabs/prysm/v5/consensus-types"
@@ -32,6 +33,7 @@ import (
 	eth "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
 	"github.com/prysmaticlabs/prysm/v5/runtime/version"
 	"github.com/prysmaticlabs/prysm/v5/time/slots"
+	"github.com/sirupsen/logrus"
 	"go.opencensus.io/trace"
 )
 
@@ -42,7 +44,8 @@ const (
 )
 
 var (
-	errNilBlock = errors.New("nil block")
+	errNilBlock         = errors.New("nil block")
+	errEquivocatedBlock = errors.New("block is equivocated")
 )
 
 type handled bool
@@ -1254,6 +1257,16 @@ func (s *Server) publishBlockSSZ(ctx context.Context, w http.ResponseWriter, r *
 			},
 		}
 		if err = s.validateBroadcast(ctx, r, genericBlock); err != nil {
+			if errors.Is(err, errEquivocatedBlock) {
+				b, err := blocks.NewSignedBeaconBlock(genericBlock)
+				if err != nil {
+					httputil.HandleError(w, err.Error(), http.StatusBadRequest)
+					return
+				}
+				if err := s.broadcastSeenBlockSidecars(ctx, b, genericBlock.GetDeneb().Blobs, genericBlock.GetDeneb().KzgProofs); err != nil {
+					log.WithError(err).Error("Failed to broadcast blob sidecars")
+				}
+			}
 			httputil.HandleError(w, err.Error(), http.StatusBadRequest)
 			return
 		}
@@ -1383,6 +1396,16 @@ func (s *Server) publishBlock(ctx context.Context, w http.ResponseWriter, r *htt
 		consensusBlock, err = denebBlockContents.ToGeneric()
 		if err == nil {
 			if err = s.validateBroadcast(ctx, r, consensusBlock); err != nil {
+				if errors.Is(err, errEquivocatedBlock) {
+					b, err := blocks.NewSignedBeaconBlock(consensusBlock)
+					if err != nil {
+						httputil.HandleError(w, err.Error(), http.StatusBadRequest)
+						return
+					}
+					if err := s.broadcastSeenBlockSidecars(ctx, b, consensusBlock.GetDeneb().Blobs, consensusBlock.GetDeneb().KzgProofs); err != nil {
+						log.WithError(err).Error("Failed to broadcast blob sidecars")
+					}
+				}
 				httputil.HandleError(w, err.Error(), http.StatusBadRequest)
 				return
 			}
@@ -1547,7 +1570,7 @@ func (s *Server) validateConsensus(ctx context.Context, blk interfaces.ReadOnlyS
 
 func (s *Server) validateEquivocation(blk interfaces.ReadOnlyBeaconBlock) error {
 	if s.ForkchoiceFetcher.HighestReceivedBlockSlot() == blk.Slot() {
-		return fmt.Errorf("block for slot %d already exists in fork choice", blk.Slot())
+		return errors.Wrapf(errEquivocatedBlock, "block for slot %d already exists in fork choice", blk.Slot())
 	}
 	return nil
 }
@@ -2072,3 +2095,37 @@ func (s *Server) GetDepositSnapshot(w http.ResponseWriter, r *http.Request) {
 		},
 	)
 }
+
+// Broadcast blob sidecars even if the block of the same slot has been imported.
+// To ensure safety, we will only broadcast blob sidecars if the header references the same block that was previously seen.
+// Otherwise, a proposer could get slashed through a different blob sidecar header reference.
+func (s *Server) broadcastSeenBlockSidecars(
+	ctx context.Context,
+	b interfaces.SignedBeaconBlock,
+	blobs [][]byte,
+	kzgProofs [][]byte) error {
+	scs, err := validator.BuildBlobSidecars(b, blobs, kzgProofs)
+	if err != nil {
+		return err
+	}
+	for _, sc := range scs {
+		r, err := sc.SignedBlockHeader.Header.HashTreeRoot()
+		if err != nil {
+			log.WithError(err).Error("Failed to hash block header for blob sidecar")
+			continue
+		}
+		if !s.FinalizationFetcher.InForkchoice(r) {
+			log.WithField("root", fmt.Sprintf("%#x", r)).Debug("Block header not in forkchoice, skipping blob sidecar broadcast")
+			continue
+		}
+		if err := s.Broadcaster.BroadcastBlob(ctx, sc.Index, sc); err != nil {
+			log.WithError(err).Error("Failed to broadcast blob sidecar for index ", sc.Index)
+		}
+		log.WithFields(logrus.Fields{
+			"index":         sc.Index,
+			"slot":          sc.SignedBlockHeader.Header.Slot,
+			"kzgCommitment": fmt.Sprintf("%#x", sc.KzgCommitment),
+		}).Info("Broadcasted blob sidecar for already seen block")
+	}
+	return nil
+}

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

@@ -13,6 +13,8 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common/hexutil"
+	mockp2p "github.com/prysmaticlabs/prysm/v5/beacon-chain/p2p/testing"
+	logTest "github.com/sirupsen/logrus/hooks/test"
 	"go.uber.org/mock/gomock"
 
 	"github.com/gorilla/mux"
@@ -2472,7 +2474,9 @@ func TestValidateEquivocation(t *testing.T) {
 		require.NoError(t, err)
 		blk.SetSlot(st.Slot())
 
-		assert.ErrorContains(t, "already exists", server.validateEquivocation(blk.Block()))
+		err = server.validateEquivocation(blk.Block())
+		assert.ErrorContains(t, "already exists", err)
+		require.ErrorIs(t, err, errEquivocatedBlock)
 	})
 }
 
@@ -3630,3 +3634,27 @@ func TestGetDepositSnapshot(t *testing.T) {
 		assert.Equal(t, finalized, len(resp.Finalized))
 	})
 }
+
+func TestServer_broadcastBlobSidecars(t *testing.T) {
+	hook := logTest.NewGlobal()
+	blockToPropose := util.NewBeaconBlockContentsDeneb()
+	blockToPropose.Blobs = [][]byte{{0x01}, {0x02}, {0x03}}
+	blockToPropose.KzgProofs = [][]byte{{0x01}, {0x02}, {0x03}}
+	blockToPropose.Block.Block.Body.BlobKzgCommitments = [][]byte{bytesutil.PadTo([]byte("kc"), 48), bytesutil.PadTo([]byte("kc1"), 48), bytesutil.PadTo([]byte("kc2"), 48)}
+	d := &eth.GenericSignedBeaconBlock_Deneb{Deneb: blockToPropose}
+	b := &eth.GenericSignedBeaconBlock{Block: d}
+
+	server := &Server{
+		Broadcaster:         &mockp2p.MockBroadcaster{},
+		FinalizationFetcher: &chainMock.ChainService{NotFinalized: true},
+	}
+
+	blk, err := blocks.NewSignedBeaconBlock(b.Block)
+	require.NoError(t, err)
+	require.NoError(t, server.broadcastSeenBlockSidecars(context.Background(), blk, b.GetDeneb().Blobs, b.GetDeneb().KzgProofs))
+	require.LogsDoNotContain(t, hook, "Broadcasted blob sidecar for already seen block")
+
+	server.FinalizationFetcher = &chainMock.ChainService{NotFinalized: false}
+	require.NoError(t, server.broadcastSeenBlockSidecars(context.Background(), blk, b.GetDeneb().Blobs, b.GetDeneb().KzgProofs))
+	require.LogsContain(t, hook, "Broadcasted blob sidecar for already seen block")
+}

beacon-chain/rpc/prysm/v1alpha1/validator/proposer.go

@@ -341,7 +341,7 @@ func (vs *Server) handleUnblindedBlock(block interfaces.SignedBeaconBlock, req *
 	if dbBlockContents == nil {
 		return nil, nil
 	}
-	return buildBlobSidecars(block, dbBlockContents.Blobs, dbBlockContents.KzgProofs)
+	return BuildBlobSidecars(block, dbBlockContents.Blobs, dbBlockContents.KzgProofs)
 }
 
 // broadcastReceiveBlock broadcasts a block and handles its reception.

beacon-chain/rpc/prysm/v1alpha1/validator/proposer_deneb.go

@@ -56,8 +56,8 @@ func (c *blobsBundleCache) prune(minSlot primitives.Slot) {
 	}
 }
 
-// buildBlobSidecars given a block, builds the blob sidecars for the block.
-func buildBlobSidecars(blk interfaces.SignedBeaconBlock, blobs [][]byte, kzgProofs [][]byte) ([]*ethpb.BlobSidecar, error) {
+// BuildBlobSidecars given a block, builds the blob sidecars for the block.
+func BuildBlobSidecars(blk interfaces.SignedBeaconBlock, blobs [][]byte, kzgProofs [][]byte) ([]*ethpb.BlobSidecar, error) {
 	if blk.Version() < version.Deneb {
 		return nil, nil // No blobs before deneb.
 	}

beacon-chain/rpc/prysm/v1alpha1/validator/proposer_deneb_test.go

@@ -51,7 +51,7 @@ func TestServer_buildBlobSidecars(t *testing.T) {
 	require.NoError(t, blk.SetBlobKzgCommitments(kzgCommitments))
 	proof, err := hexutil.Decode("0xb4021b0de10f743893d4f71e1bf830c019e832958efd6795baf2f83b8699a9eccc5dc99015d8d4d8ec370d0cc333c06a")
 	require.NoError(t, err)
-	scs, err := buildBlobSidecars(blk, [][]byte{
+	scs, err := BuildBlobSidecars(blk, [][]byte{
 		make([]byte, fieldparams.BlobLength), make([]byte, fieldparams.BlobLength),
 	}, [][]byte{
 		proof, proof,