Implement SendDataColumnSidecarsByRangeRequest and SendDataColumnSidecarsByRootRequest. (#15430)

* Implement `SendDataColumnSidecarsByRangeRequest` and `SendDataColumnSidecarsByRootRequest`.

* Fix James's comment.

* Fix James's comment.

* Fix James' comment.

* Fix marshaller.

beacon-chain/p2p/types/types.go

@@ -213,7 +213,7 @@ func (s BlobSidecarsByRootReq) Len() int {
 // ====================================
 // DataColumnsByRootIdentifiers section
 // ====================================
-var _ ssz.Marshaler = (*DataColumnsByRootIdentifiers)(nil)
+var _ ssz.Marshaler = DataColumnsByRootIdentifiers{}
 var _ ssz.Unmarshaler = (*DataColumnsByRootIdentifiers)(nil)
 
 // DataColumnsByRootIdentifiers is used to specify a list of data column targets (root+index) in a DataColumnSidecarsByRoot RPC request.
@@ -275,33 +275,33 @@ func (d *DataColumnsByRootIdentifiers) UnmarshalSSZ(buf []byte) error {
 	return nil
 }
 
-func (d *DataColumnsByRootIdentifiers) MarshalSSZ() ([]byte, error) {
+func (d DataColumnsByRootIdentifiers) MarshalSSZ() ([]byte, error) {
 	var err error
-	count := len(*d)
+	count := len(d)
 	maxSize := params.BeaconConfig().MaxRequestBlocksDeneb
 	if uint64(count) > maxSize {
 		return nil, errors.Errorf("data column identifiers list exceeds max size: %d > %d", count, maxSize)
 	}
 
-	if len(*d) == 0 {
+	if len(d) == 0 {
 		return []byte{}, nil
 	}
 	sizes := make([]uint32, count)
 	valTotal := uint32(0)
-	for i, elem := range *d {
+	for i, elem := range d {
 		if elem == nil {
 			return nil, errors.New("nil item in DataColumnsByRootIdentifiers list")
 		}
 		sizes[i] = uint32(elem.SizeSSZ())
 		valTotal += sizes[i]
 	}
-	offSize := uint32(4 * len(*d))
+	offSize := uint32(4 * len(d))
 	out := make([]byte, offSize, offSize+valTotal)
 	for i := range sizes {
 		binary.LittleEndian.PutUint32(out[i*4:i*4+4], offSize)
 		offSize += sizes[i]
 	}
-	for _, elem := range *d {
+	for _, elem := range d {
 		out, err = elem.MarshalSSZTo(out)
 		if err != nil {
 			return nil, err
@@ -312,7 +312,7 @@ func (d *DataColumnsByRootIdentifiers) MarshalSSZ() ([]byte, error) {
 }
 
 // MarshalSSZTo implements ssz.Marshaler. It appends the serialized DataColumnSidecarsByRootReq value to the provided byte slice.
-func (d *DataColumnsByRootIdentifiers) MarshalSSZTo(dst []byte) ([]byte, error) {
+func (d DataColumnsByRootIdentifiers) MarshalSSZTo(dst []byte) ([]byte, error) {
 	obj, err := d.MarshalSSZ()
 	if err != nil {
 		return nil, err
@@ -321,11 +321,11 @@ func (d *DataColumnsByRootIdentifiers) MarshalSSZTo(dst []byte) ([]byte, error)
 }
 
 // SizeSSZ implements ssz.Marshaler. It returns the size of the serialized representation.
-func (d *DataColumnsByRootIdentifiers) SizeSSZ() int {
+func (d DataColumnsByRootIdentifiers) SizeSSZ() int {
 	size := 0
-	for i := 0; i < len(*d); i++ {
+	for i := 0; i < len(d); i++ {
 		size += 4
-		size += (*d)[i].SizeSSZ()
+		size += (d)[i].SizeSSZ()
 	}
 	return size
 }

beacon-chain/sync/rpc_send_request.go

@@ -4,12 +4,14 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"slices"
 
 	"github.com/OffchainLabs/prysm/v6/beacon-chain/blockchain"
 	"github.com/OffchainLabs/prysm/v6/beacon-chain/p2p"
 	"github.com/OffchainLabs/prysm/v6/beacon-chain/p2p/encoder"
 	p2ptypes "github.com/OffchainLabs/prysm/v6/beacon-chain/p2p/types"
 	"github.com/OffchainLabs/prysm/v6/beacon-chain/verification"
+	fieldparams "github.com/OffchainLabs/prysm/v6/config/fieldparams"
 	"github.com/OffchainLabs/prysm/v6/config/params"
 	"github.com/OffchainLabs/prysm/v6/consensus-types/blocks"
 	"github.com/OffchainLabs/prysm/v6/consensus-types/interfaces"
@@ -30,16 +32,18 @@ var errBlobUnmarshal = errors.New("Could not unmarshal chunk-encoded blob")
 // Any error from the following declaration block should result in peer downscoring.
 var (
 	// ErrInvalidFetchedData is used to signal that an error occurred which should result in peer downscoring.
-	ErrInvalidFetchedData             = errors.New("invalid data returned from peer")
-	errBlobIndexOutOfBounds           = errors.Wrap(verification.ErrBlobInvalid, "blob index out of range")
-	errMaxRequestBlobSidecarsExceeded = errors.Wrap(verification.ErrBlobInvalid, "peer exceeded req blob chunk tx limit")
-	errChunkResponseSlotNotAsc        = errors.Wrap(verification.ErrBlobInvalid, "blob slot not higher than previous block root")
-	errChunkResponseIndexNotAsc       = errors.Wrap(verification.ErrBlobInvalid, "blob indices for a block must start at 0 and increase by 1")
-	errUnrequested                    = errors.Wrap(verification.ErrBlobInvalid, "received BlobSidecar in response that was not requested")
-	errBlobResponseOutOfBounds        = errors.Wrap(verification.ErrBlobInvalid, "received BlobSidecar with slot outside BlobSidecarsByRangeRequest bounds")
-	errChunkResponseBlockMismatch     = errors.Wrap(verification.ErrBlobInvalid, "blob block details do not match")
-	errChunkResponseParentMismatch    = errors.Wrap(verification.ErrBlobInvalid, "parent root for response element doesn't match previous element root")
-	errDataColumnChunkedReadFailure   = errors.New("failed to read stream of chunk-encoded data columns")
+	ErrInvalidFetchedData                    = errors.New("invalid data returned from peer")
+	errBlobIndexOutOfBounds                  = errors.Wrap(verification.ErrBlobInvalid, "blob index out of range")
+	errMaxRequestBlobSidecarsExceeded        = errors.Wrap(verification.ErrBlobInvalid, "peer exceeded req blob chunk tx limit")
+	errChunkResponseSlotNotAsc               = errors.Wrap(verification.ErrBlobInvalid, "blob slot not higher than previous block root")
+	errChunkResponseIndexNotAsc              = errors.Wrap(verification.ErrBlobInvalid, "blob indices for a block must start at 0 and increase by 1")
+	errUnrequested                           = errors.Wrap(verification.ErrBlobInvalid, "received BlobSidecar in response that was not requested")
+	errBlobResponseOutOfBounds               = errors.Wrap(verification.ErrBlobInvalid, "received BlobSidecar with slot outside BlobSidecarsByRangeRequest bounds")
+	errChunkResponseBlockMismatch            = errors.Wrap(verification.ErrBlobInvalid, "blob block details do not match")
+	errChunkResponseParentMismatch           = errors.Wrap(verification.ErrBlobInvalid, "parent root for response element doesn't match previous element root")
+	errDataColumnChunkedReadFailure          = errors.New("failed to read stream of chunk-encoded data columns")
+	errMaxRequestDataColumnSidecarsExceeded  = errors.New("count of requested data column sidecars exceeds MAX_REQUEST_DATA_COLUMN_SIDECARS")
+	errMaxResponseDataColumnSidecarsExceeded = errors.New("peer returned more data column sidecars than requested")
 )
 
 // ------
@@ -397,6 +401,245 @@ func readChunkedBlobSidecar(stream network.Stream, encoding encoder.NetworkEncod
 // Data column sidecars
 // --------------------
 
+// SendDataColumnSidecarsByRangeRequest sends a request for data column sidecars by range
+// and returns the fetched data column sidecars.
+func SendDataColumnSidecarsByRangeRequest(
+	ctx context.Context,
+	tor blockchain.TemporalOracle,
+	p2pApi p2p.P2P,
+	pid peer.ID,
+	ctxMap ContextByteVersions,
+	request *ethpb.DataColumnSidecarsByRangeRequest,
+) ([]blocks.RODataColumn, error) {
+	// Return early if nothing to request.
+	if request == nil || request.Count == 0 || len(request.Columns) == 0 {
+		return nil, nil
+	}
+
+	beaconConfig := params.BeaconConfig()
+	numberOfColumns := beaconConfig.NumberOfColumns
+	maxRequestDataColumnSidecars := params.BeaconConfig().MaxRequestDataColumnSidecars
+
+	// Check if we do not request too many sidecars.
+	columnsCount := uint64(len(request.Columns))
+	totalCount := request.Count * columnsCount
+	if totalCount > maxRequestDataColumnSidecars {
+		return nil, errors.Wrapf(errMaxRequestDataColumnSidecarsExceeded, "requestedCount=%d, allowedCount=%d", totalCount, maxRequestDataColumnSidecars)
+	}
+
+	// Build the topic.
+	currentSlot := tor.CurrentSlot()
+	currentEpoch := slots.ToEpoch(currentSlot)
+	topic, err := p2p.TopicFromMessage(p2p.DataColumnSidecarsByRangeName, currentEpoch)
+	if err != nil {
+		return nil, errors.Wrap(err, "topic from message")
+	}
+
+	// Build the logs.
+	var columnsLog interface{} = "all"
+	if columnsCount < numberOfColumns {
+		columns := request.Columns
+		slices.Sort(columns)
+		columnsLog = columns
+	}
+
+	log := log.WithFields(logrus.Fields{
+		"peer":       pid,
+		"topic":      topic,
+		"startSlot":  request.StartSlot,
+		"count":      request.Count,
+		"columns":    columnsLog,
+		"totalCount": totalCount,
+	})
+
+	// Send the request.
+	stream, err := p2pApi.Send(ctx, request, topic, pid)
+	if err != nil {
+		return nil, errors.Wrap(err, "p2p send")
+	}
+	defer closeStream(stream, log)
+
+	// Read the data column sidecars from the stream.
+	roDataColumns := make([]blocks.RODataColumn, 0, totalCount)
+	for range totalCount {
+		// Avoid reading extra chunks if the context is done.
+		if err := ctx.Err(); err != nil {
+			return nil, err
+		}
+
+		validatorSlotWithinBounds, err := isSidecarSlotWithinBounds(request)
+		if err != nil {
+			return nil, errors.Wrap(err, "is sidecar slot within bounds")
+		}
+
+		roDataColumn, err := readChunkedDataColumnSidecar(
+			stream, p2pApi, ctxMap,
+			validatorSlotWithinBounds,
+			isSidecarIndexRequested(request),
+		)
+		if errors.Is(err, io.EOF) {
+			return roDataColumns, nil
+		}
+		if err != nil {
+			return nil, errors.Wrap(err, "read chunked data column sidecar")
+		}
+
+		if roDataColumn == nil {
+			return nil, errors.New("nil data column sidecar, should never happen")
+		}
+
+		roDataColumns = append(roDataColumns, *roDataColumn)
+	}
+
+	// All requested sidecars were delivered by the peer. Expecting EOF.
+	if _, err := readChunkedDataColumnSidecar(stream, p2pApi, ctxMap); !errors.Is(err, io.EOF) {
+		return nil, errors.Wrapf(errMaxResponseDataColumnSidecarsExceeded, "requestedCount=%d", totalCount)
+	}
+
+	return roDataColumns, nil
+}
+
+// isSidecarSlotWithinBounds verifies that the slot of the data column sidecar is within the bounds of the request.
+func isSidecarSlotWithinBounds(request *ethpb.DataColumnSidecarsByRangeRequest) (DataColumnResponseValidation, error) {
+	// endSlot is exclusive (while request.StartSlot is inclusive).
+	endSlot, err := request.StartSlot.SafeAdd(request.Count)
+	if err != nil {
+		return nil, errors.Wrap(err, "calculate end slot")
+	}
+
+	validator := func(sidecar blocks.RODataColumn) error {
+		slot := sidecar.Slot()
+
+		if !(request.StartSlot <= slot && slot < endSlot) {
+			return errors.Errorf("data column sidecar slot %d out of range [%d, %d[", slot, request.StartSlot, endSlot)
+		}
+
+		return nil
+	}
+
+	return validator, nil
+}
+
+// isSidecarIndexRequested verifies that the index of the data column sidecar is found in the requested indices.
+func isSidecarIndexRequested(request *ethpb.DataColumnSidecarsByRangeRequest) DataColumnResponseValidation {
+	requestedIndices := make(map[uint64]bool)
+	for _, col := range request.Columns {
+		requestedIndices[col] = true
+	}
+
+	return func(sidecar blocks.RODataColumn) error {
+		columnIndex := sidecar.Index
+		if !requestedIndices[columnIndex] {
+			return errors.Errorf("data column sidecar index %d not found in requested indices", columnIndex)
+		}
+
+		return nil
+	}
+}
+
+// SendDataColumnSidecarsByRootRequest sends a request for data column sidecars by root
+// and returns the fetched data column sidecars.
+func SendDataColumnSidecarsByRootRequest(
+	ctx context.Context,
+	tor blockchain.TemporalOracle,
+	p2pApi p2p.P2P,
+	pid peer.ID,
+	ctxMap ContextByteVersions,
+	request p2ptypes.DataColumnsByRootIdentifiers,
+) ([]blocks.RODataColumn, error) {
+	// Return early if the request is nil.
+	if request == nil {
+		return nil, nil
+	}
+
+	// Compute how many sidecars are requested.
+	count := uint64(0)
+	for _, identifier := range request {
+		count += uint64(len(identifier.Columns))
+	}
+
+	// Return early if nothing to request.
+	if count == 0 {
+		return nil, nil
+	}
+
+	// Verify that the request count is within the maximum allowed.
+	maxRequestDataColumnSidecars := params.BeaconConfig().MaxRequestDataColumnSidecars
+	if count > maxRequestDataColumnSidecars {
+		return nil, errors.Wrapf(errMaxRequestDataColumnSidecarsExceeded, "current: %d, max: %d", count, maxRequestDataColumnSidecars)
+	}
+
+	// Get the topic for the request.
+	topic, err := p2p.TopicFromMessage(p2p.DataColumnSidecarsByRootName, slots.ToEpoch(tor.CurrentSlot()))
+	if err != nil {
+		return nil, errors.Wrap(err, "topic from message")
+	}
+
+	// Send the request to the peer.
+	stream, err := p2pApi.Send(ctx, request, topic, pid)
+	if err != nil {
+		return nil, errors.Wrap(err, "p2p api send")
+	}
+	defer closeStream(stream, log)
+
+	// Read the data column sidecars from the stream.
+	roDataColumns := make([]blocks.RODataColumn, 0, count)
+
+	// Read the data column sidecars from the stream.
+	for range count {
+		roDataColumn, err := readChunkedDataColumnSidecar(stream, p2pApi, ctxMap, isSidecarIndexRootRequested(request))
+		if errors.Is(err, io.EOF) {
+			return roDataColumns, nil
+		}
+		if err != nil {
+			return nil, errors.Wrap(err, "read chunked data column sidecar")
+		}
+
+		if roDataColumn == nil {
+			return nil, errors.Wrap(err, "nil data column sidecar, should never happen")
+		}
+
+		roDataColumns = append(roDataColumns, *roDataColumn)
+	}
+
+	// All requested sidecars were delivered by the peer. Expecting EOF.
+	if _, err := readChunkedDataColumnSidecar(stream, p2pApi, ctxMap); !errors.Is(err, io.EOF) {
+		return nil, errors.Wrapf(errMaxResponseDataColumnSidecarsExceeded, "requestedCount=%d", count)
+	}
+
+	return roDataColumns, nil
+}
+
+func isSidecarIndexRootRequested(request p2ptypes.DataColumnsByRootIdentifiers) DataColumnResponseValidation {
+	columnsIndexFromRoot := make(map[[fieldparams.RootLength]byte]map[uint64]bool)
+
+	for _, sidecar := range request {
+		blockRoot := bytesutil.ToBytes32(sidecar.BlockRoot)
+		if columnsIndexFromRoot[blockRoot] == nil {
+			columnsIndexFromRoot[blockRoot] = make(map[uint64]bool)
+		}
+
+		for _, column := range sidecar.Columns {
+			columnsIndexFromRoot[blockRoot][column] = true
+		}
+	}
+
+	return func(sidecar blocks.RODataColumn) error {
+		root, index := sidecar.BlockRoot(), sidecar.Index
+		indices, ok := columnsIndexFromRoot[root]
+
+		if !ok {
+			return errors.Errorf("root #%x returned by peer but not requested", root)
+		}
+
+		if !indices[index] {
+			return errors.Errorf("index %d for root #%x returned by peer but not requested", index, root)
+		}
+
+		return nil
+	}
+}
+
 // DataColumnResponseValidation represents a function that can validate aspects of a single unmarshaled data column sidecar
 // that was received from a peer in response to an rpc request.
 type DataColumnResponseValidation func(column blocks.RODataColumn) error

beacon-chain/sync/rpc_send_request_test.go

@@ -11,7 +11,9 @@ import (
 
 	"github.com/OffchainLabs/prysm/v6/beacon-chain/p2p"
 	p2ptest "github.com/OffchainLabs/prysm/v6/beacon-chain/p2p/testing"
+	"github.com/OffchainLabs/prysm/v6/beacon-chain/p2p/types"
 	p2pTypes "github.com/OffchainLabs/prysm/v6/beacon-chain/p2p/types"
+	p2ptypes "github.com/OffchainLabs/prysm/v6/beacon-chain/p2p/types"
 	"github.com/OffchainLabs/prysm/v6/beacon-chain/startup"
 	"github.com/OffchainLabs/prysm/v6/beacon-chain/verification"
 	fieldparams "github.com/OffchainLabs/prysm/v6/config/fieldparams"
@@ -885,6 +887,544 @@ func TestErrInvalidFetchedDataDistinction(t *testing.T) {
 	require.Equal(t, false, errors.Is(ErrInvalidFetchedData, verification.ErrBlobInvalid))
 }
 
+func TestSendDataColumnSidecarsByRangeRequest(t *testing.T) {
+	nilTestCases := []struct {
+		name    string
+		request *ethpb.DataColumnSidecarsByRangeRequest
+	}{
+		{
+			name:    "nil request",
+			request: nil,
+		},
+		{
+			name:    "count is 0",
+			request: &ethpb.DataColumnSidecarsByRangeRequest{},
+		},
+		{
+			name:    "columns is nil",
+			request: &ethpb.DataColumnSidecarsByRangeRequest{Count: 1},
+		},
+	}
+
+	for _, tc := range nilTestCases {
+		t.Run(tc.name, func(t *testing.T) {
+			actual, err := SendDataColumnSidecarsByRangeRequest(t.Context(), nil, nil, "aRandomPID", nil, tc.request)
+			require.NoError(t, err)
+			require.IsNil(t, actual)
+		})
+	}
+
+	t.Run("too many columns in request", func(t *testing.T) {
+		params.SetupTestConfigCleanup(t)
+		beaconConfig := params.BeaconConfig()
+		beaconConfig.MaxRequestDataColumnSidecars = 0
+		params.OverrideBeaconConfig(beaconConfig)
+
+		request := &ethpb.DataColumnSidecarsByRangeRequest{Count: 1, Columns: []uint64{1, 2, 3}}
+		_, err := SendDataColumnSidecarsByRangeRequest(t.Context(), nil, nil, "aRandomPID", nil, request)
+		require.ErrorContains(t, errMaxRequestDataColumnSidecarsExceeded.Error(), err)
+	})
+
+	type slotIndex struct {
+		Slot  primitives.Slot
+		Index uint64
+	}
+
+	createSidecar := func(slotIndex slotIndex) *ethpb.DataColumnSidecar {
+		const count = 4
+		kzgCommitmentsInclusionProof := make([][]byte, 0, count)
+		for range count {
+			kzgCommitmentsInclusionProof = append(kzgCommitmentsInclusionProof, make([]byte, 32))
+		}
+
+		return &ethpb.DataColumnSidecar{
+			Index: slotIndex.Index,
+			SignedBlockHeader: &ethpb.SignedBeaconBlockHeader{
+				Header: &ethpb.BeaconBlockHeader{
+					Slot:       slotIndex.Slot,
+					ParentRoot: make([]byte, fieldparams.RootLength),
+					StateRoot:  make([]byte, fieldparams.RootLength),
+					BodyRoot:   make([]byte, fieldparams.RootLength),
+				},
+				Signature: make([]byte, fieldparams.BLSSignatureLength),
+			},
+			KzgCommitmentsInclusionProof: kzgCommitmentsInclusionProof,
+		}
+	}
+
+	testCases := []struct {
+		name          string
+		slotIndices   []slotIndex
+		expectedError error
+	}{
+		{
+			name: "too many responses",
+			slotIndices: []slotIndex{
+				{Slot: 0, Index: 1},
+				{Slot: 0, Index: 2},
+				{Slot: 0, Index: 3},
+				{Slot: 1, Index: 1},
+				{Slot: 1, Index: 2},
+				{Slot: 1, Index: 3},
+				{Slot: 0, Index: 3}, // Duplicate
+			},
+			expectedError: errMaxResponseDataColumnSidecarsExceeded,
+		},
+		{
+			name: "perfect match",
+			slotIndices: []slotIndex{
+				{Slot: 0, Index: 1},
+				{Slot: 0, Index: 2},
+				{Slot: 0, Index: 3},
+				{Slot: 1, Index: 1},
+				{Slot: 1, Index: 2},
+				{Slot: 1, Index: 3},
+			},
+		},
+		{
+			name: "few responses than maximum possible",
+			slotIndices: []slotIndex{
+				{Slot: 0, Index: 1},
+				{Slot: 0, Index: 2},
+				{Slot: 0, Index: 3},
+				{Slot: 1, Index: 1},
+				{Slot: 1, Index: 2},
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			protocol := fmt.Sprintf("%s/ssz_snappy", p2p.RPCDataColumnSidecarsByRangeTopicV1)
+			clock := startup.NewClock(time.Now(), [fieldparams.RootLength]byte{})
+
+			p1, p2 := p2ptest.NewTestP2P(t), p2ptest.NewTestP2P(t)
+			p1.Connect(p2)
+
+			expected := make([]*ethpb.DataColumnSidecar, 0, len(tc.slotIndices))
+			for _, slotIndex := range tc.slotIndices {
+				sidecar := createSidecar(slotIndex)
+				expected = append(expected, sidecar)
+			}
+
+			requestSent := &ethpb.DataColumnSidecarsByRangeRequest{
+				StartSlot: 0,
+				Count:     2,
+				Columns:   []uint64{1, 3, 2},
+			}
+
+			var wg sync.WaitGroup
+			wg.Add(1)
+
+			p2.SetStreamHandler(protocol, func(stream network.Stream) {
+				wg.Done()
+
+				requestReceived := new(ethpb.DataColumnSidecarsByRangeRequest)
+				err := p2.Encoding().DecodeWithMaxLength(stream, requestReceived)
+				assert.NoError(t, err)
+				assert.DeepSSZEqual(t, requestSent, requestReceived)
+
+				for _, sidecar := range expected {
+					err := WriteDataColumnSidecarChunk(stream, clock, p2.Encoding(), sidecar)
+					assert.NoError(t, err)
+				}
+
+				err = stream.CloseWrite()
+				assert.NoError(t, err)
+			})
+
+			ctx := t.Context()
+			ctxMap := ContextByteVersions{[4]byte{245, 165, 253, 66}: version.Fulu}
+
+			actual, err := SendDataColumnSidecarsByRangeRequest(ctx, clock, p1, p2.PeerID(), ctxMap, requestSent)
+			if tc.expectedError != nil {
+				require.ErrorContains(t, tc.expectedError.Error(), err)
+				if util.WaitTimeout(&wg, time.Second) {
+					t.Fatal("Did not receive stream within 1 sec")
+				}
+
+				return
+			}
+
+			require.Equal(t, len(expected), len(actual))
+			for i := range expected {
+				require.DeepSSZEqual(t, expected[i], actual[i].DataColumnSidecar)
+			}
+		})
+	}
+}
+
+func TestIsSidecarSlotWithinBounds(t *testing.T) {
+	request := &ethpb.DataColumnSidecarsByRangeRequest{
+		StartSlot: 10,
+		Count:     10,
+	}
+
+	validator, err := isSidecarSlotWithinBounds(request)
+	require.NoError(t, err)
+
+	testCases := []struct {
+		name            string
+		slot            primitives.Slot
+		isErrorExpected bool
+	}{
+		{
+			name:            "too soon",
+			slot:            9,
+			isErrorExpected: true,
+		},
+		{
+			name:            "too late",
+			slot:            20,
+			isErrorExpected: true,
+		},
+		{
+			name:            "within bounds",
+			slot:            15,
+			isErrorExpected: false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			const count = 4
+			kzgCommitmentsInclusionProof := make([][]byte, 0, count)
+			for range count {
+				kzgCommitmentsInclusionProof = append(kzgCommitmentsInclusionProof, make([]byte, 32))
+			}
+
+			sidecarPb := &ethpb.DataColumnSidecar{
+				SignedBlockHeader: &ethpb.SignedBeaconBlockHeader{
+					Header: &ethpb.BeaconBlockHeader{
+						Slot:       tc.slot,
+						ParentRoot: make([]byte, fieldparams.RootLength),
+						StateRoot:  make([]byte, fieldparams.RootLength),
+						BodyRoot:   make([]byte, fieldparams.RootLength),
+					},
+					Signature: make([]byte, fieldparams.BLSSignatureLength),
+				},
+				KzgCommitmentsInclusionProof: kzgCommitmentsInclusionProof,
+			}
+
+			sidecar, err := blocks.NewRODataColumn(sidecarPb)
+			require.NoError(t, err)
+
+			err = validator(sidecar)
+			if tc.isErrorExpected {
+				require.NotNil(t, err)
+				return
+			}
+
+			require.NoError(t, err)
+		})
+	}
+}
+
+func TestIsSidecarIndexRequested(t *testing.T) {
+	request := &ethpb.DataColumnSidecarsByRangeRequest{
+		Columns: []uint64{2, 9, 4},
+	}
+
+	validator := isSidecarIndexRequested(request)
+
+	testCases := []struct {
+		name            string
+		index           uint64
+		isErrorExpected bool
+	}{
+		{
+			name:            "not requested",
+			index:           1,
+			isErrorExpected: true,
+		},
+		{
+			name:            "requested",
+			index:           9,
+			isErrorExpected: false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			const count = 4
+			kzgCommitmentsInclusionProof := make([][]byte, 0, count)
+			for range count {
+				kzgCommitmentsInclusionProof = append(kzgCommitmentsInclusionProof, make([]byte, 32))
+			}
+
+			sidecarPb := &ethpb.DataColumnSidecar{
+				SignedBlockHeader: &ethpb.SignedBeaconBlockHeader{
+					Header: &ethpb.BeaconBlockHeader{
+						Slot:       0,
+						ParentRoot: make([]byte, fieldparams.RootLength),
+						StateRoot:  make([]byte, fieldparams.RootLength),
+						BodyRoot:   make([]byte, fieldparams.RootLength),
+					},
+					Signature: make([]byte, fieldparams.BLSSignatureLength),
+				},
+				KzgCommitmentsInclusionProof: kzgCommitmentsInclusionProof,
+				Index:                        tc.index,
+			}
+
+			sidecar, err := blocks.NewRODataColumn(sidecarPb)
+			require.NoError(t, err)
+
+			err = validator(sidecar)
+			if tc.isErrorExpected {
+				require.NotNil(t, err)
+				return
+			}
+
+			require.NoError(t, err)
+		})
+	}
+}
+
+func TestSendDataColumnSidecarsByRootRequest(t *testing.T) {
+	nilTestCases := []struct {
+		name    string
+		request p2ptypes.DataColumnsByRootIdentifiers
+	}{
+		{
+			name:    "nil request",
+			request: nil,
+		},
+		{
+			name:    "count is 0",
+			request: p2ptypes.DataColumnsByRootIdentifiers{{}, {}},
+		},
+	}
+
+	for _, tc := range nilTestCases {
+		t.Run(tc.name, func(t *testing.T) {
+			actual, err := SendDataColumnSidecarsByRootRequest(t.Context(), nil, nil, "aRandomPID", nil, tc.request)
+			require.NoError(t, err)
+			require.IsNil(t, actual)
+		})
+	}
+
+	t.Run("too many columns in request", func(t *testing.T) {
+		params.SetupTestConfigCleanup(t)
+		beaconConfig := params.BeaconConfig()
+		beaconConfig.MaxRequestDataColumnSidecars = 4
+		params.OverrideBeaconConfig(beaconConfig)
+
+		request := p2ptypes.DataColumnsByRootIdentifiers{
+			{Columns: []uint64{1, 2, 3}},
+			{Columns: []uint64{4, 5, 6}},
+		}
+
+		_, err := SendDataColumnSidecarsByRootRequest(t.Context(), nil, nil, "aRandomPID", nil, request)
+		require.ErrorContains(t, errMaxRequestDataColumnSidecarsExceeded.Error(), err)
+	})
+
+	type slotIndex struct {
+		Slot  primitives.Slot
+		Index uint64
+	}
+
+	createSidecar := func(rootIndex slotIndex) blocks.RODataColumn {
+		const count = 4
+		kzgCommitmentsInclusionProof := make([][]byte, 0, count)
+		for range count {
+			kzgCommitmentsInclusionProof = append(kzgCommitmentsInclusionProof, make([]byte, 32))
+		}
+
+		sidecarPb := &ethpb.DataColumnSidecar{
+			Index: rootIndex.Index,
+			SignedBlockHeader: &ethpb.SignedBeaconBlockHeader{
+				Header: &ethpb.BeaconBlockHeader{
+					ParentRoot: make([]byte, fieldparams.RootLength),
+					StateRoot:  make([]byte, fieldparams.RootLength),
+					BodyRoot:   make([]byte, fieldparams.RootLength),
+				},
+				Signature: make([]byte, fieldparams.BLSSignatureLength),
+			},
+			KzgCommitmentsInclusionProof: kzgCommitmentsInclusionProof,
+		}
+
+		roSidecar, err := blocks.NewRODataColumn(sidecarPb)
+		require.NoError(t, err)
+
+		return roSidecar
+	}
+
+	testCases := []struct {
+		name          string
+		slotIndices   []slotIndex
+		expectedError error
+	}{
+		{
+			name: "too many responses",
+			slotIndices: []slotIndex{
+				{Slot: 1, Index: 1},
+				{Slot: 1, Index: 2},
+				{Slot: 1, Index: 3},
+				{Slot: 2, Index: 1},
+				{Slot: 2, Index: 2},
+				{Slot: 2, Index: 3},
+				{Slot: 1, Index: 3}, // Duplicate
+			},
+			expectedError: errMaxResponseDataColumnSidecarsExceeded,
+		},
+		{
+			name: "perfect match",
+			slotIndices: []slotIndex{
+				{Slot: 1, Index: 1},
+				{Slot: 1, Index: 2},
+				{Slot: 1, Index: 3},
+				{Slot: 2, Index: 1},
+				{Slot: 2, Index: 2},
+				{Slot: 2, Index: 3},
+			},
+		},
+		{
+			name: "few responses than maximum possible",
+			slotIndices: []slotIndex{
+				{Slot: 1, Index: 1},
+				{Slot: 1, Index: 2},
+				{Slot: 1, Index: 3},
+				{Slot: 2, Index: 1},
+				{Slot: 2, Index: 2},
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			protocol := fmt.Sprintf("%s/ssz_snappy", p2p.RPCDataColumnSidecarsByRootTopicV1)
+			clock := startup.NewClock(time.Now(), [fieldparams.RootLength]byte{})
+
+			p1, p2 := p2ptest.NewTestP2P(t), p2ptest.NewTestP2P(t)
+			p1.Connect(p2)
+
+			expected := make([]blocks.RODataColumn, 0, len(tc.slotIndices))
+			for _, slotIndex := range tc.slotIndices {
+				roSidecar := createSidecar(slotIndex)
+				expected = append(expected, roSidecar)
+			}
+
+			blockRoot1, blockRoot2 := expected[0].BlockRoot(), expected[3].BlockRoot()
+
+			sentRequest := p2ptypes.DataColumnsByRootIdentifiers{
+				{BlockRoot: blockRoot1[:], Columns: []uint64{1, 2, 3}},
+				{BlockRoot: blockRoot2[:], Columns: []uint64{1, 2, 3}},
+			}
+
+			var wg sync.WaitGroup
+			wg.Add(1)
+
+			p2.SetStreamHandler(protocol, func(stream network.Stream) {
+				wg.Done()
+
+				requestReceived := new(p2ptypes.DataColumnsByRootIdentifiers)
+				err := p2.Encoding().DecodeWithMaxLength(stream, requestReceived)
+				assert.NoError(t, err)
+
+				require.Equal(t, len(sentRequest), len(*requestReceived))
+				for i := range sentRequest {
+					require.DeepSSZEqual(t, (sentRequest)[i], (*requestReceived)[i])
+				}
+
+				for _, sidecar := range expected {
+					err := WriteDataColumnSidecarChunk(stream, clock, p2.Encoding(), sidecar.DataColumnSidecar)
+					assert.NoError(t, err)
+				}
+
+				err = stream.CloseWrite()
+				assert.NoError(t, err)
+			})
+
+			ctx := t.Context()
+			ctxMap := ContextByteVersions{[4]byte{245, 165, 253, 66}: version.Fulu}
+
+			actual, err := SendDataColumnSidecarsByRootRequest(ctx, clock, p1, p2.PeerID(), ctxMap, sentRequest)
+			if tc.expectedError != nil {
+				require.ErrorContains(t, tc.expectedError.Error(), err)
+				if util.WaitTimeout(&wg, time.Second) {
+					t.Fatal("Did not receive stream within 1 sec")
+				}
+
+				return
+			}
+
+			require.Equal(t, len(expected), len(actual))
+			for i := range expected {
+				require.DeepSSZEqual(t, expected[i], actual[i])
+			}
+		})
+	}
+}
+
+func TestIsSidecarIndexRootRequested(t *testing.T) {
+	testCases := []struct {
+		name            string
+		root            [fieldparams.RootLength]byte
+		index           uint64
+		isErrorExpected bool
+	}{
+		{
+			name:            "non requested root",
+			root:            [fieldparams.RootLength]byte{2},
+			isErrorExpected: true,
+		},
+		{
+			name:            "non requested index",
+			root:            [fieldparams.RootLength]byte{1},
+			index:           3,
+			isErrorExpected: true,
+		},
+		{
+			name:            "nominal",
+			root:            [fieldparams.RootLength]byte{1},
+			index:           2,
+			isErrorExpected: false,
+		},
+	}
+
+	request := types.DataColumnsByRootIdentifiers{
+		{BlockRoot: []byte{1}, Columns: []uint64{1, 2}},
+	}
+
+	validator := isSidecarIndexRootRequested(request)
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			const count = 4
+			kzgCommitmentsInclusionProof := make([][]byte, 0, count)
+			for range count {
+				kzgCommitmentsInclusionProof = append(kzgCommitmentsInclusionProof, make([]byte, 32))
+			}
+
+			sidecarPb := &ethpb.DataColumnSidecar{
+				SignedBlockHeader: &ethpb.SignedBeaconBlockHeader{
+					Header: &ethpb.BeaconBlockHeader{
+						ParentRoot: make([]byte, fieldparams.RootLength),
+						StateRoot:  make([]byte, fieldparams.RootLength),
+						BodyRoot:   make([]byte, fieldparams.RootLength),
+					},
+					Signature: make([]byte, fieldparams.BLSSignatureLength),
+				},
+				KzgCommitmentsInclusionProof: kzgCommitmentsInclusionProof,
+				Index:                        tc.index,
+			}
+
+			// There is a discrepancy between `tc.root` and the real root,
+			// but we don't care about it here.
+			sidecar, err := blocks.NewRODataColumnWithRoot(sidecarPb, tc.root)
+			require.NoError(t, err)
+
+			err = validator(sidecar)
+			if tc.isErrorExpected {
+				require.NotNil(t, err)
+				return
+			}
+
+			require.NoError(t, err)
+		})
+	}
+}
+
 func TestReadChunkedDataColumnSidecar(t *testing.T) {
 	t.Run("non nil status code", func(t *testing.T) {
 		const reason = "a dummy reason"

changelog/manu-peerdas-send-data-column-requests.md

@@ -0,0 +1,3 @@
+### Added 
+- Implement `SendDataColumnSidecarsByRangeRequest`.
+- Implement `SendDataColumnSidecarsByRootRequest`.