Fix Blobs By Range RPC Handler (#14910)

* Add tests for TestSendBlobsByRangeRequest. Currently not working with sequential blob validation.

* Copy Root First

* Allow Test For Maximum Amount of Blobs

* Fails with the Same error

* Fix Last Test Assertion

* Add in Fix

* Changelog

---------

Co-authored-by: Preston Van Loon <preston@pvl.dev>

beacon-chain/sync/rpc_send_request_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v5/testing/assert"
 	"github.com/prysmaticlabs/prysm/v5/testing/require"
 	"github.com/prysmaticlabs/prysm/v5/testing/util"
+	"github.com/prysmaticlabs/prysm/v5/time/slots"
 )
 
 func TestSendRequest_SendBeaconBlocksByRangeRequest(t *testing.T) {
@@ -691,3 +692,122 @@ func TestSeqBlobValid(t *testing.T) {
 		})
 	}
 }
+
+func TestSendBlobsByRangeRequest(t *testing.T) {
+	topic := fmt.Sprintf("%s/ssz_snappy", p2p.RPCBlobSidecarsByRangeTopicV1)
+	ctx := context.Background()
+
+	t.Run("single blob - Deneb", func(t *testing.T) {
+		// Setup genesis such that we are currently in deneb.
+		s := uint64(slots.UnsafeEpochStart(params.BeaconConfig().DenebForkEpoch)) * params.BeaconConfig().SecondsPerSlot
+		clock := startup.NewClock(time.Now().Add(-time.Second*time.Duration(s)), [32]byte{})
+		ctxByte, err := ContextByteVersionsForValRoot(clock.GenesisValidatorsRoot())
+		require.NoError(t, err)
+		// Setup peers
+		p1 := p2ptest.NewTestP2P(t)
+		p2 := p2ptest.NewTestP2P(t)
+		p1.Connect(p2)
+		// Set current slot to a deneb slot.
+		slot := slots.UnsafeEpochStart(params.BeaconConfig().DenebForkEpoch + 1)
+		// Create a simple handler that will return a valid response.
+		p2.SetStreamHandler(topic, func(stream network.Stream) {
+			defer func() {
+				assert.NoError(t, stream.Close())
+			}()
+
+			req := &ethpb.BlobSidecarsByRangeRequest{}
+			assert.NoError(t, p2.Encoding().DecodeWithMaxLength(stream, req))
+			assert.Equal(t, slot, req.StartSlot)
+			assert.Equal(t, uint64(1), req.Count)
+
+			// Create a sequential set of blobs with the appropriate header information.
+			var prevRoot [32]byte
+			for i := req.StartSlot; i < req.StartSlot+primitives.Slot(req.Count); i++ {
+				b := util.HydrateBlobSidecar(&ethpb.BlobSidecar{})
+				b.SignedBlockHeader.Header.Slot = i
+				b.SignedBlockHeader.Header.ParentRoot = prevRoot[:]
+				ro, err := blocks.NewROBlob(b)
+				require.NoError(t, err)
+				vro := blocks.NewVerifiedROBlob(ro)
+				prevRoot = vro.BlockRoot()
+				assert.NoError(t, WriteBlobSidecarChunk(stream, clock, p2.Encoding(), vro))
+			}
+		})
+		req := &ethpb.BlobSidecarsByRangeRequest{
+			StartSlot: slot,
+			Count:     1,
+		}
+
+		blobs, err := SendBlobsByRangeRequest(ctx, clock, p1, p2.PeerID(), ctxByte, req)
+		assert.NoError(t, err)
+		assert.Equal(t, 1, len(blobs))
+	})
+
+	t.Run("Deneb - Electra epoch boundary crossing", func(t *testing.T) {
+		cfg := params.BeaconConfig()
+		cfg.ElectraForkEpoch = cfg.DenebForkEpoch + 1
+		undo, err := params.SetActiveWithUndo(cfg)
+		require.NoError(t, err)
+		defer func() {
+			require.NoError(t, undo())
+		}()
+		// Setup genesis such that we are currently in deneb.
+		s := uint64(slots.UnsafeEpochStart(params.BeaconConfig().DenebForkEpoch)) * params.BeaconConfig().SecondsPerSlot
+		clock := startup.NewClock(time.Now().Add(-time.Second*time.Duration(s)), [32]byte{})
+		ctxByte, err := ContextByteVersionsForValRoot(clock.GenesisValidatorsRoot())
+		require.NoError(t, err)
+		// Setup peers
+		p1 := p2ptest.NewTestP2P(t)
+		p2 := p2ptest.NewTestP2P(t)
+		p1.Connect(p2)
+		// Set current slot to the first slot of the last deneb epoch.
+		slot := slots.UnsafeEpochStart(params.BeaconConfig().DenebForkEpoch)
+		// Create a simple handler that will return a valid response.
+		p2.SetStreamHandler(topic, func(stream network.Stream) {
+			defer func() {
+				assert.NoError(t, stream.Close())
+			}()
+
+			req := &ethpb.BlobSidecarsByRangeRequest{}
+			assert.NoError(t, p2.Encoding().DecodeWithMaxLength(stream, req))
+			assert.Equal(t, slot, req.StartSlot)
+			assert.Equal(t, uint64(params.BeaconConfig().SlotsPerEpoch)*3, req.Count)
+
+			// Create a sequential set of blobs with the appropriate header information.
+			var prevRoot [32]byte
+			for i := req.StartSlot; i < req.StartSlot+primitives.Slot(req.Count); i++ {
+				maxBlobsForSlot := cfg.MaxBlobsPerBlock(i)
+				parentRoot := prevRoot
+				header := util.HydrateSignedBeaconHeader(&ethpb.SignedBeaconBlockHeader{})
+				header.Header.Slot = i
+				header.Header.ParentRoot = parentRoot[:]
+				bRoot, err := header.Header.HashTreeRoot()
+				require.NoError(t, err)
+				prevRoot = bRoot
+				// Send the maximum possible blobs per slot.
+				for j := 0; j < maxBlobsForSlot; j++ {
+					b := util.HydrateBlobSidecar(&ethpb.BlobSidecar{})
+					b.SignedBlockHeader = header
+					b.Index = uint64(j)
+					ro, err := blocks.NewROBlob(b)
+					require.NoError(t, err)
+					vro := blocks.NewVerifiedROBlob(ro)
+					assert.NoError(t, WriteBlobSidecarChunk(stream, clock, p2.Encoding(), vro))
+				}
+			}
+		})
+		req := &ethpb.BlobSidecarsByRangeRequest{
+			StartSlot: slot,
+			Count:     uint64(params.BeaconConfig().SlotsPerEpoch) * 3,
+		}
+		maxDenebBlobs := cfg.MaxBlobsPerBlockAtEpoch(cfg.DenebForkEpoch)
+		maxElectraBlobs := cfg.MaxBlobsPerBlockAtEpoch(cfg.ElectraForkEpoch)
+		totalDenebBlobs := primitives.Slot(maxDenebBlobs) * params.BeaconConfig().SlotsPerEpoch
+		totalElectraBlobs := primitives.Slot(maxElectraBlobs) * 2 * params.BeaconConfig().SlotsPerEpoch
+		totalExpectedBlobs := totalDenebBlobs + totalElectraBlobs
+
+		blobs, err := SendBlobsByRangeRequest(ctx, clock, p1, p2.PeerID(), ctxByte, req)
+		assert.NoError(t, err)
+		assert.Equal(t, int(totalExpectedBlobs), len(blobs))
+	})
+}