BlobSidecarsByRoot (#12420)

* BlobSidecarsByRoot RPC handler

* BlobSidecarsByRange rpc handler (#12499)

Co-authored-by: Kasey Kirkham <kasey@users.noreply.github.com>

---------

Co-authored-by: Kasey Kirkham <kasey@users.noreply.github.com>

beacon-chain/p2p/peers/peers_test.go

@@ -22,6 +22,8 @@ func TestMain(m *testing.M) {
 	flags.Init(&flags.GlobalFlags{
 		BlockBatchLimit:            64,
 		BlockBatchLimitBurstFactor: 10,
+		BlobBatchLimit:             8,
+		BlobBatchLimitBurstFactor:  2,
 	})
 	defer func() {
 		flags.Init(resetFlags)

beacon-chain/p2p/rpc_topic_mappings.go

@@ -37,6 +37,12 @@ const PingMessageName = "/ping"
 // MetadataMessageName specifies the name for the metadata message topic.
 const MetadataMessageName = "/metadata"
 
+// BlobSidecarsByRangeName is the name for the BlobSidecarsByRange v1 message topic.
+const BlobSidecarsByRangeName = "/blob_sidecars_by_range"
+
+// BlobSidecarsByRootName is the name for the BlobSidecarsByRoot v1 message topic.
+const BlobSidecarsByRootName = "/blob_sidecars_by_root"
+
 const (
 	// V1 RPC Topics
 	// RPCStatusTopicV1 defines the v1 topic for the status rpc method.
@@ -52,6 +58,14 @@ const (
 	// RPCMetaDataTopicV1 defines the v1 topic for the metadata rpc method.
 	RPCMetaDataTopicV1 = protocolPrefix + MetadataMessageName + SchemaVersionV1
 
+	// RPCBlobSidecarsByRangeTopicV1 is a topic for requesting blob sidecars
+	// in the slot range [start_slot, start_slot + count), leading up to the current head block as selected by fork choice.
+	// Protocol ID: /eth2/beacon_chain/req/blob_sidecars_by_range/1/ - New in deneb.
+	RPCBlobSidecarsByRangeTopicV1 = protocolPrefix + BlobSidecarsByRangeName + SchemaVersionV1
+	// RPCBlobSidecarsByRootTopicV1 is a topic for requesting blob sidecars by their block root. New in deneb.
+	// /eth2/beacon_chain/req/blob_sidecars_by_root/1/
+	RPCBlobSidecarsByRootTopicV1 = protocolPrefix + BlobSidecarsByRootName + SchemaVersionV1
+
 	// V2 RPC Topics
 	// RPCBlocksByRangeTopicV2 defines v2 the topic for the blocks by range rpc method.
 	RPCBlocksByRangeTopicV2 = protocolPrefix + BeaconBlocksByRangeMessageName + SchemaVersionV2
@@ -83,6 +97,10 @@ var RPCTopicMappings = map[string]interface{}{
 	// RPC Metadata Message
 	RPCMetaDataTopicV1: new(interface{}),
 	RPCMetaDataTopicV2: new(interface{}),
+	// BlobSidecarsByRange v1 Message
+	RPCBlobSidecarsByRangeTopicV1: new(pb.BlobSidecarsByRangeRequest),
+	// BlobSidecarsByRoot v1 Message
+	RPCBlobSidecarsByRootTopicV1: new(p2ptypes.BlobSidecarsByRootReq),
 }
 
 // Maps all registered protocol prefixes.
@@ -99,6 +117,8 @@ var messageMapping = map[string]bool{
 	BeaconBlocksByRootsMessageName: true,
 	PingMessageName:                true,
 	MetadataMessageName:            true,
+	BlobSidecarsByRangeName:        true,
+	BlobSidecarsByRootName:         true,
 }
 
 // Maps all the RPC messages which are to updated in altair.
@@ -113,6 +133,19 @@ var versionMapping = map[string]bool{
 	SchemaVersionV2: true,
 }
 
+// OmitContextBytesV1 keeps track of which RPC methods do not write context bytes in their v1 incarnations.
+// Phase0 did not have the notion of context bytes, which prefix wire-encoded values with a [4]byte identifier
+// to convey the schema for the receiver to use. These RPCs had a version bump to V2 when the context byte encoding
+// was introduced. For other RPC methods, context bytes are always required.
+var OmitContextBytesV1 = map[string]bool{
+	StatusMessageName:              true,
+	GoodbyeMessageName:             true,
+	BeaconBlocksByRangeMessageName: true,
+	BeaconBlocksByRootsMessageName: true,
+	PingMessageName:                true,
+	MetadataMessageName:            true,
+}
+
 // VerifyTopicMapping verifies that the topic and its accompanying
 // message type is correct.
 func VerifyTopicMapping(topic string, msg interface{}) error {

beacon-chain/p2p/types/BUILD.bazel

@@ -41,7 +41,9 @@ go_test(
         "//config/params:go_default_library",
         "//consensus-types/primitives:go_default_library",
         "//encoding/bytesutil:go_default_library",
+        "//proto/prysm/v1alpha1:go_default_library",
         "//testing/assert:go_default_library",
         "//testing/require:go_default_library",
+        "@com_github_prysmaticlabs_fastssz//:go_default_library",
     ],
 )

beacon-chain/p2p/types/rpc_errors.go

@@ -12,4 +12,6 @@ var (
 	ErrRateLimited            = errors.New("rate limited")
 	ErrIODeadline             = errors.New("i/o deadline exceeded")
 	ErrInvalidRequest         = errors.New("invalid range, step or count")
+	ErrBlobLTMinRequest       = errors.New("blob slot < minimum_request_epoch")
+	ErrMaxBlobReqExceeded     = errors.New("requested more than MAX_REQUEST_BLOB_SIDECARS")
 )

beacon-chain/p2p/types/types.go

@@ -4,9 +4,13 @@
 package types
 
 import (
+	"bytes"
+	"sort"
+
 	"github.com/pkg/errors"
 	ssz "github.com/prysmaticlabs/fastssz"
 	"github.com/prysmaticlabs/prysm/v4/config/params"
+	eth "github.com/prysmaticlabs/prysm/v4/proto/prysm/v1alpha1"
 )
 
 const rootLength = 32
@@ -64,7 +68,7 @@ func (r *BeaconBlockByRootsReq) UnmarshalSSZ(buf []byte) error {
 	bufLen := len(buf)
 	maxLength := int(params.BeaconNetworkConfig().MaxRequestBlocks * rootLength)
 	if bufLen > maxLength {
-		return errors.Errorf("expected buffer with length of upto %d but received length %d", maxLength, bufLen)
+		return errors.Errorf("expected buffer with length of up to %d but received length %d", maxLength, bufLen)
 	}
 	if bufLen%rootLength != 0 {
 		return ssz.ErrIncorrectByteSize
@@ -120,3 +124,90 @@ func (m *ErrorMessage) UnmarshalSSZ(buf []byte) error {
 	*m = errMsg
 	return nil
 }
+
+// BlobSidecarsByRootReq is used to specify a list of blob targets (root+index) in a BlobSidecarsByRoot RPC request.
+type BlobSidecarsByRootReq []*eth.BlobIdentifier
+
+// BlobIdentifier is a fixed size value, so we can compute its fixed size at start time (see init below)
+var blobIdSize int
+
+// SizeSSZ returns the size of the serialized representation.
+func (b *BlobSidecarsByRootReq) SizeSSZ() int {
+	return len(*b) * blobIdSize
+}
+
+// MarshalSSZTo appends the serialized BlobSidecarsByRootReq value to the provided byte slice.
+func (b *BlobSidecarsByRootReq) MarshalSSZTo(dst []byte) ([]byte, error) {
+	// A List without an enclosing container is marshaled exactly like a vector, no length offset required.
+	marshalledObj, err := b.MarshalSSZ()
+	if err != nil {
+		return nil, err
+	}
+	return append(dst, marshalledObj...), nil
+}
+
+// MarshalSSZ serializes the BlobSidecarsByRootReq value to a byte slice.
+func (b *BlobSidecarsByRootReq) MarshalSSZ() ([]byte, error) {
+	buf := make([]byte, len(*b)*blobIdSize)
+	for i, id := range *b {
+		by, err := id.MarshalSSZ()
+		if err != nil {
+			return nil, err
+		}
+		copy(buf[i*blobIdSize:(i+1)*blobIdSize], by)
+	}
+	return buf, nil
+}
+
+// UnmarshalSSZ unmarshals the provided bytes buffer into the
+// BlobSidecarsByRootReq value.
+func (b *BlobSidecarsByRootReq) UnmarshalSSZ(buf []byte) error {
+	bufLen := len(buf)
+	maxLength := int(params.BeaconNetworkConfig().MaxRequestBlobSidecars) * blobIdSize
+	if bufLen > maxLength {
+		return errors.Errorf("expected buffer with length of up to %d but received length %d", maxLength, bufLen)
+	}
+	if bufLen%blobIdSize != 0 {
+		return errors.Wrapf(ssz.ErrIncorrectByteSize, "size=%d", bufLen)
+	}
+	count := bufLen / blobIdSize
+	*b = make([]*eth.BlobIdentifier, count)
+	for i := 0; i < count; i++ {
+		id := &eth.BlobIdentifier{}
+		err := id.UnmarshalSSZ(buf[i*blobIdSize : (i+1)*blobIdSize])
+		if err != nil {
+			return err
+		}
+		(*b)[i] = id
+	}
+	return nil
+}
+
+var _ sort.Interface = BlobSidecarsByRootReq{}
+
+// Less reports whether the element with index i must sort before the element with index j.
+// BlobIdentifier will be sorted in lexicographic order by root, with Blob Index as tiebreaker for a given root.
+func (s BlobSidecarsByRootReq) Less(i, j int) bool {
+	rootCmp := bytes.Compare(s[i].BlockRoot, s[j].BlockRoot)
+	if rootCmp != 0 {
+		// They aren't equal; return true if i < j, false if i > j.
+		return rootCmp < 0
+	}
+	// They are equal; blob index is the tie breaker.
+	return s[i].Index < s[j].Index
+}
+
+// Swap swaps the elements with indexes i and j.
+func (s BlobSidecarsByRootReq) Swap(i, j int) {
+	s[i], s[j] = s[j], s[i]
+}
+
+// Len is the number of elements in the collection.
+func (s BlobSidecarsByRootReq) Len() int {
+	return len(s)
+}
+
+func init() {
+	sizer := &eth.BlobIdentifier{}
+	blobIdSize = sizer.SizeSSZ()
+}

beacon-chain/p2p/types/types_test.go

@@ -4,12 +4,82 @@ import (
 	"encoding/hex"
 	"testing"
 
+	ssz "github.com/prysmaticlabs/fastssz"
 	"github.com/prysmaticlabs/prysm/v4/config/params"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
+	"github.com/prysmaticlabs/prysm/v4/encoding/bytesutil"
+	eth "github.com/prysmaticlabs/prysm/v4/proto/prysm/v1alpha1"
 	"github.com/prysmaticlabs/prysm/v4/testing/assert"
 	"github.com/prysmaticlabs/prysm/v4/testing/require"
 )
 
+func generateBlobIdentifiers(n int) []*eth.BlobIdentifier {
+	r := make([]*eth.BlobIdentifier, n)
+	for i := 0; i < n; i++ {
+		r[i] = &eth.BlobIdentifier{
+			BlockRoot: bytesutil.PadTo([]byte{byte(i)}, 32),
+			Index:     0,
+		}
+	}
+	return r
+}
+
+func TestBlobSidecarsByRootReq_MarshalSSZ(t *testing.T) {
+	cases := []struct {
+		name         string
+		ids          []*eth.BlobIdentifier
+		marshalErr   error
+		unmarshalErr error
+		unmarshalMod func([]byte) []byte
+	}{
+		{
+			name: "empty list",
+		},
+		{
+			name: "single item list",
+			ids:  generateBlobIdentifiers(1),
+		},
+		{
+			name: "10 item list",
+			ids:  generateBlobIdentifiers(10),
+		},
+		{
+			name: "wonky unmarshal size",
+			ids:  generateBlobIdentifiers(10),
+			unmarshalMod: func(in []byte) []byte {
+				in = append(in, byte(0))
+				return in
+			},
+			unmarshalErr: ssz.ErrIncorrectByteSize,
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			r := BlobSidecarsByRootReq(c.ids)
+			by, err := r.MarshalSSZ()
+			if c.marshalErr != nil {
+				require.ErrorIs(t, err, c.marshalErr)
+				return
+			}
+			require.NoError(t, err)
+			if c.unmarshalMod != nil {
+				by = c.unmarshalMod(by)
+			}
+			got := &BlobSidecarsByRootReq{}
+			err = got.UnmarshalSSZ(by)
+			if c.unmarshalErr != nil {
+				require.ErrorIs(t, err, c.unmarshalErr)
+				return
+			}
+			require.NoError(t, err)
+			for i, gid := range *got {
+				require.DeepEqual(t, c.ids[i], gid)
+			}
+		})
+	}
+}
+
 func TestBeaconBlockByRootsReq_Limit(t *testing.T) {
 	fixedRoots := make([][32]byte, 0)
 	for i := uint64(0); i < params.BeaconNetworkConfig().MaxRequestBlocks+100; i++ {
@@ -25,7 +95,7 @@ func TestBeaconBlockByRootsReq_Limit(t *testing.T) {
 		buf = append(buf, rt[:]...)
 	}
 	req2 := BeaconBlockByRootsReq(nil)
-	require.ErrorContains(t, "expected buffer with length of upto", req2.UnmarshalSSZ(buf))
+	require.ErrorContains(t, "expected buffer with length of up to", req2.UnmarshalSSZ(buf))
 }
 
 func TestErrorResponse_Limit(t *testing.T) {

beacon-chain/sync/BUILD.bazel

@@ -22,6 +22,8 @@ go_library(
         "rpc.go",
         "rpc_beacon_blocks_by_range.go",
         "rpc_beacon_blocks_by_root.go",
+        "rpc_blob_sidecars_by_range.go",
+        "rpc_blob_sidecars_by_root.go",
         "rpc_chunked_response.go",
         "rpc_goodbye.go",
         "rpc_metadata.go",
@@ -89,6 +91,7 @@ go_library(
         "//cache/lru:go_default_library",
         "//cmd/beacon-chain/flags:go_default_library",
         "//config/features:go_default_library",
+        "//config/fieldparams:go_default_library",
         "//config/params:go_default_library",
         "//consensus-types/blocks:go_default_library",
         "//consensus-types/interfaces:go_default_library",
@@ -136,6 +139,7 @@ go_test(
     size = "small",
     srcs = [
         "batch_verifier_test.go",
+        "blobs_test.go",
         "block_batcher_test.go",
         "broadcast_bls_changes_test.go",
         "context_test.go",
@@ -147,6 +151,8 @@ go_test(
         "rate_limiter_test.go",
         "rpc_beacon_blocks_by_range_test.go",
         "rpc_beacon_blocks_by_root_test.go",
+        "rpc_blob_sidecars_by_range_test.go",
+        "rpc_blob_sidecars_by_root_test.go",
         "rpc_chunked_response_test.go",
         "rpc_goodbye_test.go",
         "rpc_handler_test.go",