moves publish blobs to private api and adds tests

api/server/structs/conversions_blob.go

@@ -54,4 +54,4 @@ func (sc *Sidecar) ToConsensus() (*eth.BlobSidecar, error) {
 		}
 
 		return bsc, nil
-}
+}

api/server/structs/endpoints_blob.go

@@ -19,5 +19,5 @@ type BlobSidecars struct {
 
 type PublishBlobsRequest struct {
 	BlobSidecars *BlobSidecars `json:"blob_sidecars"`
-	BeaconRoot   string        `json:"beacon_root"`
-}
+	BlockRoot    string        `json:"block_root"`
+}

beacon-chain/rpc/endpoints.go

@@ -465,7 +465,6 @@ func (s *Service) beaconEndpoints(
 		Broadcaster:                   s.cfg.Broadcaster,
 		BlockReceiver:                 s.cfg.BlockReceiver,
 		StateGenService:               s.cfg.StateGen,
-		BlobReceiver:                  s.cfg.BlobReceiver,
 		Stater:                        stater,
 		Blocker:                       blocker,
 		OptimisticModeFetcher:         s.cfg.OptimisticModeFetcher,
@@ -558,16 +557,6 @@ func (s *Service) beaconEndpoints(
 			handler: server.PublishBlockV2,
 			methods: []string{http.MethodPost},
 		},
-		{
-			template: "/eth/v2/beacon/blobs",
-			name:     namespace + ".PublishBlobs",
-			middleware: []mux.MiddlewareFunc{
-				middleware.ContentTypeHandler([]string{api.JsonMediaType, api.OctetStreamMediaType}),
-				middleware.AcceptHeaderHandler([]string{api.JsonMediaType}),
-			},
-			handler: server.PublishBlobs,
-			methods: []string{http.MethodPost},
-		},
 		{
 			template: "/eth/v2/beacon/blinded_blocks",
 			name:     namespace + ".PublishBlindedBlockV2",
@@ -953,6 +942,8 @@ func (s *Service) prysmBeaconEndpoints(
 		ChainInfoFetcher:      s.cfg.ChainInfoFetcher,
 		FinalizationFetcher:   s.cfg.FinalizationFetcher,
 		CoreService:           coreService,
+		Broadcaster: 		   s.cfg.Broadcaster,
+		BlobReceiver: 		   s.cfg.BlobReceiver,
 	}
 
 	const namespace = "prysm.beacon"
@@ -1003,6 +994,16 @@ func (s *Service) prysmBeaconEndpoints(
 			handler: server.GetChainHead,
 			methods: []string{http.MethodGet},
 		},
+		{
+			template: "/prysm/v1/beacon/blobs",
+			name:     namespace + ".PublishBlobs",
+			middleware: []middleware.Middleware{
+				middleware.ContentTypeHandler([]string{api.JsonMediaType}),
+				middleware.AcceptHeaderHandler([]string{api.JsonMediaType}),
+			},
+			handler: server.PublishBlobs,
+			methods: []string{http.MethodPost},
+		},
 	}
 }
 

beacon-chain/rpc/endpoints_test.go

@@ -31,7 +31,6 @@ func Test_endpoints(t *testing.T) {
 		"/eth/v2/beacon/blinded_blocks":                              {http.MethodPost},
 		"/eth/v1/beacon/blocks":                                      {http.MethodPost},
 		"/eth/v2/beacon/blocks":                                      {http.MethodPost},
-		"/eth/v2/beacon/blobs":                                       {http.MethodPost},
 		"/eth/v1/beacon/blocks/{block_id}":                           {http.MethodGet},
 		"/eth/v2/beacon/blocks/{block_id}":                           {http.MethodGet},
 		"/eth/v1/beacon/blocks/{block_id}/root":                      {http.MethodGet},
@@ -116,6 +115,7 @@ func Test_endpoints(t *testing.T) {
 		"/eth/v1/beacon/states/{state_id}/validator_count":   {http.MethodGet},
 		"/prysm/v1/beacon/states/{state_id}/validator_count": {http.MethodGet},
 		"/prysm/v1/beacon/chain_head":                        {http.MethodGet},
+		"/prysm/v1/beacon/blobs":                             {http.MethodPost},
 	}
 
 	prysmNodeRoutes := map[string][]string{

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

@@ -958,54 +958,6 @@ func (s *Server) publishBlock(ctx context.Context, w http.ResponseWriter, r *htt
 	httputil.HandleError(w, "Body does not represent a valid block type", http.StatusBadRequest)
 }
 
-func (s *Server) PublishBlobs(w http.ResponseWriter, r *http.Request) {
-	ctx, span := trace.StartSpan(r.Context(), "beacon.PublishBlobs")
-	defer span.End()
-	if shared.IsSyncing(r.Context(), w, s.SyncChecker, s.HeadFetcher, s.TimeFetcher, s.OptimisticModeFetcher) {
-		return
-	}
-
-	var req structs.PublishBlobsRequest
-	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
-		httputil.HandleError(w, "Could not decode JSON request body", http.StatusBadRequest)
-		return
-	}
-
-	root, err := bytesutil.DecodeHexWithLength(req.BeaconRoot, 32)
-	if err != nil {
-		httputil.HandleError(w, "Could not decode beacon root", http.StatusBadRequest)
-		return
-	}
-
-	var beaconRoot [32]byte
-	copy(beaconRoot[:], root)
-
-	for _, blobSidecar := range req.BlobSidecars.Sidecars {
-
-		sc, err := blobSidecar.ToConsensus()
-		if err != nil {
-			http.Error(w, "Could not decode blob sidecar", http.StatusBadRequest)
-			return
-		}
-
-		readOnlySc, err := blocks.NewROBlobWithRoot(sc, beaconRoot)
-		if err != nil {
-			http.Error(w, "Could not create read-only blob", http.StatusInternalServerError)
-			return
-		}
-
-		verifiedBlob := blocks.NewVerifiedROBlob(readOnlySc)
-		if err := s.BlobReceiver.ReceiveBlob(ctx, verifiedBlob); err != nil {
-			http.Error(w, "Could not receive blob", http.StatusInternalServerError)
-			return
-		}
-
-		if err := s.Broadcaster.BroadcastBlob(ctx, sc.Index, sc); err != nil {
-			log.WithError(err).Error("Failed to broadcast blob")
-		}
-	}
-}
-
 func (s *Server) proposeBlock(ctx context.Context, w http.ResponseWriter, blk *eth.GenericSignedBeaconBlock) {
 	_, err := s.V1Alpha1ValidatorServer.ProposeBeaconBlock(ctx, blk)
 	if err != nil {

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

@@ -3874,16 +3874,3 @@ func TestServer_broadcastBlobSidecars(t *testing.T) {
 	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")
 }
-
-func TestPublishBlobs(t *testing.T) {
-	server := &Server{
-		BlobReceiver:  &chainMock.ChainService{},
-		Broadcaster:   &mockp2p.MockBroadcaster{},
-		SyncChecker:   &mockSync.Sync{IsSyncing: false},
-	}
-	request := httptest.NewRequest(http.MethodPost, "http://foo.example", bytes.NewReader([]byte(rpctesting.PublishBlobsRequest)))
-	writer := httptest.NewRecorder()
-	writer.Body = &bytes.Buffer{}
-	server.PublishBlobs(writer, request)
-	assert.Equal(t, http.StatusOK, writer.Code)
-}

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

@@ -48,5 +48,4 @@ type Server struct {
 	BLSChangesPool                blstoexec.PoolManager
 	ForkchoiceFetcher             blockchain.ForkchoiceFetcher
 	CoreService                   *core.Service
-	BlobReceiver                  blockchain.BlobReceiver
 }

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

@@ -15,7 +15,9 @@ import (
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/rpc/core"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/rpc/eth/shared"
 	"github.com/prysmaticlabs/prysm/v5/config/params"
+	"github.com/prysmaticlabs/prysm/v5/consensus-types/blocks"
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/primitives"
+	"github.com/prysmaticlabs/prysm/v5/encoding/bytesutil"
 	"github.com/prysmaticlabs/prysm/v5/network/httputil"
 	ethpb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
 	"github.com/prysmaticlabs/prysm/v5/time/slots"
@@ -183,3 +185,48 @@ func (s *Server) GetChainHead(w http.ResponseWriter, r *http.Request) {
 	}
 	httputil.WriteJson(w, response)
 }
+
+func (s *Server) PublishBlobs(w http.ResponseWriter, r *http.Request) {
+	ctx, span := trace.StartSpan(r.Context(), "beacon.PublishBlobs")
+	defer span.End()
+	if shared.IsSyncing(r.Context(), w, s.SyncChecker, s.HeadFetcher, s.TimeFetcher, s.OptimisticModeFetcher) {
+		return
+	}
+
+	var req structs.PublishBlobsRequest
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		httputil.HandleError(w, "Could not decode JSON request body", http.StatusBadRequest)
+		return
+	}
+
+	root, err := bytesutil.DecodeHexWithLength(req.BlockRoot, 32)
+	if err != nil {
+		httputil.HandleError(w, "Could not decode block root", http.StatusBadRequest)
+		return
+	}
+
+	for _, blobSidecar := range req.BlobSidecars.Sidecars {
+		sc, err := blobSidecar.ToConsensus()
+		if err != nil {
+			httputil.HandleError(w, "Could not decode blob sidecar", http.StatusBadRequest)
+			return
+		}
+
+		readOnlySc, err := blocks.NewROBlobWithRoot(sc, bytesutil.ToBytes32(root))
+		if err != nil {
+			httputil.HandleError(w, "Could not create read-only blob", http.StatusInternalServerError)
+			return
+		}
+
+		verifiedBlob := blocks.NewVerifiedROBlob(readOnlySc)
+		if err := s.BlobReceiver.ReceiveBlob(ctx, verifiedBlob); err != nil {
+			httputil.HandleError(w, "Could not receive blob", http.StatusInternalServerError)
+			return
+		}
+
+		if err := s.Broadcaster.BroadcastBlob(ctx, sc.Index, sc); err != nil {
+			httputil.HandleError(w, "Failed to broadcast blob", http.StatusInternalServerError)
+			return
+		}
+	}
+}

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

@@ -15,6 +15,8 @@ import (
 	"github.com/prysmaticlabs/go-bitfield"
 	"github.com/prysmaticlabs/prysm/v5/api/server/structs"
 	chainMock "github.com/prysmaticlabs/prysm/v5/beacon-chain/blockchain/testing"
+	mockp2p "github.com/prysmaticlabs/prysm/v5/beacon-chain/p2p/testing"
+	mockSync "github.com/prysmaticlabs/prysm/v5/beacon-chain/sync/initial-sync/testing"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/helpers"
 	dbTest "github.com/prysmaticlabs/prysm/v5/beacon-chain/db/testing"
 	doublylinkedtree "github.com/prysmaticlabs/prysm/v5/beacon-chain/forkchoice/doubly-linked-tree"
@@ -28,6 +30,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/primitives"
 	"github.com/prysmaticlabs/prysm/v5/encoding/bytesutil"
 	ethpb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
+	rpctesting "github.com/prysmaticlabs/prysm/v5/beacon-chain/rpc/prysm/testing"
 	"github.com/prysmaticlabs/prysm/v5/testing/assert"
 	"github.com/prysmaticlabs/prysm/v5/testing/require"
 	"github.com/prysmaticlabs/prysm/v5/testing/util"
@@ -870,3 +873,74 @@ func TestServer_GetChainHead(t *testing.T) {
 	assert.DeepEqual(t, hexutil.Encode(fRoot[:]), ch.FinalizedBlockRoot, "Unexpected FinalizedBlockRoot")
 	assert.Equal(t, false, ch.OptimisticStatus)
 }
+
+func TestPublishBlobs_InvalidJson(t *testing.T) {
+	server := &Server{
+		BlobReceiver:  &chainMock.ChainService{},
+		Broadcaster:   &mockp2p.MockBroadcaster{},
+		SyncChecker:   &mockSync.Sync{IsSyncing: false},
+	}
+
+	request := httptest.NewRequest(http.MethodPost, "http://foo.example", bytes.NewReader([]byte(rpctesting.InvalidJson)))
+	writer := httptest.NewRecorder()
+	writer.Body = &bytes.Buffer{}
+	server.PublishBlobs(writer, request)
+	assert.Equal(t, http.StatusBadRequest, writer.Code)
+	assert.StringContains(t, "Could not decode JSON request body", writer.Body.String())
+
+	assert.Equal(t, len(server.BlobReceiver.(*chainMock.ChainService).Blobs), 0)
+	assert.Equal(t, server.Broadcaster.(*mockp2p.MockBroadcaster).BroadcastCalled.Load(), false)
+}
+
+func TestPublishBlobs_MissingBlob(t *testing.T) {
+	server := &Server{
+		BlobReceiver:  &chainMock.ChainService{},
+		Broadcaster:   &mockp2p.MockBroadcaster{},
+		SyncChecker:   &mockSync.Sync{IsSyncing: false},
+	}
+
+	request := httptest.NewRequest(http.MethodPost, "http://foo.example", bytes.NewReader([]byte(rpctesting.PublishBlobsRequestMissingBlob)))
+	writer := httptest.NewRecorder()
+	writer.Body = &bytes.Buffer{}
+	server.PublishBlobs(writer, request)
+	assert.Equal(t, http.StatusBadRequest, writer.Code)
+	assert.StringContains(t, "Could not decode blob sidecar", writer.Body.String())
+
+	assert.Equal(t, len(server.BlobReceiver.(*chainMock.ChainService).Blobs), 0)
+	assert.Equal(t, server.Broadcaster.(*mockp2p.MockBroadcaster).BroadcastCalled.Load(), false)
+}
+
+func TestPublishBlobs_BadBlockRoot(t *testing.T) {
+	server := &Server{
+		BlobReceiver:  &chainMock.ChainService{},
+		Broadcaster:   &mockp2p.MockBroadcaster{},
+		SyncChecker:   &mockSync.Sync{IsSyncing: false},
+	}
+
+	request := httptest.NewRequest(http.MethodPost, "http://foo.example", bytes.NewReader([]byte(rpctesting.PublishBlobsRequestBadBlockRoot)))
+	writer := httptest.NewRecorder()
+	writer.Body = &bytes.Buffer{}
+	server.PublishBlobs(writer, request)
+	assert.Equal(t, http.StatusBadRequest, writer.Code)
+	assert.StringContains(t, "Could not decode block root", writer.Body.String())
+
+	assert.Equal(t, len(server.BlobReceiver.(*chainMock.ChainService).Blobs), 0)
+	assert.Equal(t, server.Broadcaster.(*mockp2p.MockBroadcaster).BroadcastCalled.Load(), false)
+}
+
+func TestPublishBlobs(t *testing.T) {
+	server := &Server{
+		BlobReceiver:  &chainMock.ChainService{},
+		Broadcaster:   &mockp2p.MockBroadcaster{},
+		SyncChecker:   &mockSync.Sync{IsSyncing: false},
+	}
+
+	request := httptest.NewRequest(http.MethodPost, "http://foo.example", bytes.NewReader([]byte(rpctesting.PublishBlobsRequest)))
+	writer := httptest.NewRecorder()
+	writer.Body = &bytes.Buffer{}
+	server.PublishBlobs(writer, request)
+	assert.Equal(t, http.StatusOK, writer.Code)
+
+	assert.Equal(t, len(server.BlobReceiver.(*chainMock.ChainService).Blobs), 1)
+	assert.Equal(t, server.Broadcaster.(*mockp2p.MockBroadcaster).BroadcastCalled.Load(), true)
+}

beacon-chain/rpc/prysm/beacon/server.go

@@ -3,6 +3,7 @@ package beacon
 import (
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/blockchain"
 	beacondb "github.com/prysmaticlabs/prysm/v5/beacon-chain/db"
+	"github.com/prysmaticlabs/prysm/v5/beacon-chain/p2p"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/rpc/core"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/rpc/lookup"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/state/stategen"
@@ -20,4 +21,6 @@ type Server struct {
 	ChainInfoFetcher      blockchain.ChainInfoFetcher
 	FinalizationFetcher   blockchain.FinalizationFetcher
 	CoreService           *core.Service
+	Broadcaster           p2p.Broadcaster
+	BlobReceiver          blockchain.BlobReceiver
 }

beacon-chain/rpc/prysm/testing/BUILD.bazel

@@ -0,0 +1,11 @@
+load("@prysm//tools/go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    testonly = True,
+    srcs = [
+        "json.go",
+    ],
+    importpath = "github.com/prysmaticlabs/prysm/v5/beacon-chain/rpc/prysm/testing",
+    visibility = ["//visibility:public"],
+)