processing ConsensusSidecars in blob-processors

beacond/cmd/defaults.go

@@ -54,7 +54,7 @@ func DefaultComponents() []any {
 		components.ProvideBlsSigner,
 		components.ProvideBlobProcessor[
 			*AvailabilityStore, *BeaconBlockBody, *BeaconBlockHeader,
-			*BlobSidecar, *BlobSidecars, *Logger,
+			*ConsensusSidecars, *BlobSidecar, *BlobSidecars, *Logger,
 		],
 		components.ProvideBlobProofVerifier,
 		components.ProvideBlobVerifier[

beacond/cmd/types.go

@@ -74,6 +74,7 @@ type (
 		*AvailabilityStore,
 		*BeaconBlockBody,
 		*BeaconBlockHeader,
+		*ConsensusSidecars,
 		*BlobSidecar,
 		*BlobSidecars,
 	]

mod/da/pkg/blob/processor.go

@@ -36,6 +36,7 @@ type Processor[
 	],
 	BeaconBlockBodyT any,
 	BeaconBlockHeaderT BeaconBlockHeader,
+	ConsensusSidecarsT ConsensusSidecars[BlobSidecarsT, BeaconBlockHeaderT],
 	BlobSidecarT Sidecar[BeaconBlockHeaderT],
 	BlobSidecarsT Sidecars[BlobSidecarT],
 ] struct {
@@ -59,6 +60,7 @@ func NewProcessor[
 	],
 	BeaconBlockBodyT any,
 	BeaconBlockHeaderT BeaconBlockHeader,
+	ConsensusSidecarsT ConsensusSidecars[BlobSidecarsT, BeaconBlockHeaderT],
 	BlobSidecarT Sidecar[BeaconBlockHeaderT],
 	BlobSidecarsT Sidecars[BlobSidecarT],
 ](
@@ -69,11 +71,11 @@ func NewProcessor[
 	telemetrySink TelemetrySink,
 ) *Processor[
 	AvailabilityStoreT, BeaconBlockBodyT, BeaconBlockHeaderT,
-	BlobSidecarT, BlobSidecarsT,
+	ConsensusSidecarsT, BlobSidecarT, BlobSidecarsT,
 ] {
 	return &Processor[
 		AvailabilityStoreT, BeaconBlockBodyT, BeaconBlockHeaderT,
-		BlobSidecarT, BlobSidecarsT,
+		ConsensusSidecarsT, BlobSidecarT, BlobSidecarsT,
 	]{
 		logger:            logger,
 		chainSpec:         chainSpec,
@@ -84,10 +86,15 @@ func NewProcessor[
 }
 
 // VerifySidecars verifies the blobs and ensures they match the local state.
-func (sp *Processor[AvailabilityStoreT, _, _, _, BlobSidecarsT]) VerifySidecars(
-	sidecars BlobSidecarsT,
+func (sp *Processor[
+	AvailabilityStoreT, _, _, ConsensusSidecarsT, _, _,
+]) VerifySidecars(
+	cs ConsensusSidecarsT,
 ) error {
-	startTime := time.Now()
+	var (
+		startTime = time.Now()
+		sidecars  = cs.GetSidecars()
+	)
 	defer sp.metrics.measureVerifySidecarsDuration(
 		startTime, math.U64(sidecars.Len()),
 	)
@@ -109,7 +116,7 @@ func (sp *Processor[AvailabilityStoreT, _, _, _, BlobSidecarsT]) VerifySidecars(
 
 // slot :=  processes the blobs and ensures they match the local state.
 func (sp *Processor[
-	AvailabilityStoreT, _, _, _, BlobSidecarsT,
+	AvailabilityStoreT, _, _, _, _, BlobSidecarsT,
 ]) ProcessSidecars(
 	avs AvailabilityStoreT,
 	sidecars BlobSidecarsT,

mod/da/pkg/blob/types.go

@@ -66,6 +66,11 @@ type BlobVerifier[BlobSidecarsT any] interface {
 	VerifySidecars(sidecars BlobSidecarsT, kzgOffset uint64) error
 }
 
+type ConsensusSidecars[BlobSidecarsT any, BeaconBlockHeaderT any] interface {
+	GetSidecars() BlobSidecarsT
+	GetHeader() BeaconBlockHeaderT
+}
+
 type Sidecar[BeaconBlockHeaderT any] interface {
 	GetBeaconBlockHeader() BeaconBlockHeaderT
 	GetBlob() eip4844.Blob

mod/da/pkg/da/service.go

@@ -41,7 +41,7 @@ type Service[
 	avs AvailabilityStoreT
 	bp  BlobProcessor[
 		AvailabilityStoreT,
-		BlobSidecarsT,
+		ConsensusSidecarsT, BlobSidecarsT,
 	]
 	dispatcher asynctypes.EventDispatcher
 	logger     log.Logger
@@ -60,7 +60,8 @@ func NewService[
 ](
 	avs AvailabilityStoreT,
 	bp BlobProcessor[
-		AvailabilityStoreT, BlobSidecarsT,
+		AvailabilityStoreT,
+		ConsensusSidecarsT, BlobSidecarsT,
 	],
 	dispatcher asynctypes.EventDispatcher,
 	logger log.Logger,
@@ -191,15 +192,15 @@ func (s *Service[_, ConsensusSidecarsT, _, _]) verifySidecars(
 	cs ConsensusSidecarsT,
 ) error {
 	sidecars := cs.GetSidecars()
-	// If there are no blobs to verify, return early.
 	if sidecars.IsNil() || sidecars.Len() == 0 {
+		// nothing to verify
 		return nil
 	}
 
 	s.logger.Info("Received incoming blob sidecars")
 
 	// Verify the blobs and ensure they match the local state.
-	if err := s.bp.VerifySidecars(sidecars); err != nil {
+	if err := s.bp.VerifySidecars(cs); err != nil {
 		s.logger.Error(
 			"rejecting incoming blob sidecars",
 			"reason", err,

mod/da/pkg/da/types.go

@@ -21,17 +21,18 @@
 package da
 
 // BlobProcessor is the interface for the blobs processor.
-type BlobProcessor[AvailabilityStoreT any, BlobSidecarsT any] interface {
+type BlobProcessor[
+	AvailabilityStoreT,
+	ConsensusSidecarsT, BlobSidecarsT any,
+] interface {
 	// ProcessSidecars processes the blobs and ensures they match the local
 	// state.
 	ProcessSidecars(
 		avs AvailabilityStoreT,
 		sidecars BlobSidecarsT,
 	) error
 	// VerifySidecars verifies the blobs and ensures they match the local state.
-	VerifySidecars(
-		sidecars BlobSidecarsT,
-	) error
+	VerifySidecars(sidecars ConsensusSidecarsT) error
 }
 
 type ConsensusSidecars[BlobSidecarsT any, BeaconBlockHeaderT any] interface {

mod/node-core/pkg/components/blobs.go

@@ -96,19 +96,21 @@ func ProvideBlobProcessor[
 	AvailabilityStoreT AvailabilityStore[BeaconBlockBodyT, BlobSidecarsT],
 	BeaconBlockBodyT any,
 	BeaconBlockHeaderT BeaconBlockHeader[BeaconBlockHeaderT],
+	ConsensusSidecarsT ConsensusSidecars[BlobSidecarsT, BeaconBlockHeaderT],
 	BlobSidecarT BlobSidecar[BeaconBlockHeaderT],
 	BlobSidecarsT BlobSidecars[BlobSidecarsT, BlobSidecarT],
 	LoggerT log.AdvancedLogger[LoggerT],
 ](
 	in BlobProcessorIn[BlobSidecarsT, LoggerT],
 ) *dablob.Processor[
 	AvailabilityStoreT, BeaconBlockBodyT, BeaconBlockHeaderT,
-	BlobSidecarT, BlobSidecarsT,
+	ConsensusSidecarsT, BlobSidecarT, BlobSidecarsT,
 ] {
 	return dablob.NewProcessor[
 		AvailabilityStoreT,
 		BeaconBlockBodyT,
 		BeaconBlockHeaderT,
+		ConsensusSidecarsT,
 		BlobSidecarT,
 		BlobSidecarsT,
 	](
@@ -123,15 +125,15 @@ func ProvideBlobProcessor[
 // DAServiceIn is the input for the BlobService.
 type DAServiceIn[
 	AvailabilityStoreT any,
-	BeaconBlockBodyT any,
+	ConsensusSidecarsT any,
 	BlobSidecarsT any,
 	LoggerT any,
 ] struct {
 	depinject.In
 
 	AvailabilityStore AvailabilityStoreT
 	BlobProcessor     BlobProcessor[
-		AvailabilityStoreT, BeaconBlockBodyT, BlobSidecarsT,
+		AvailabilityStoreT, ConsensusSidecarsT, BlobSidecarsT,
 	]
 	Dispatcher Dispatcher
 	Logger     LoggerT
@@ -149,7 +151,7 @@ func ProvideDAService[
 	LoggerT log.AdvancedLogger[LoggerT],
 ](
 	in DAServiceIn[
-		AvailabilityStoreT, BeaconBlockBodyT, BlobSidecarsT, LoggerT,
+		AvailabilityStoreT, ConsensusSidecarsT, BlobSidecarsT, LoggerT,
 	],
 ) *da.Service[
 	AvailabilityStoreT,

mod/node-core/pkg/components/interfaces.go

@@ -227,7 +227,7 @@ type (
 	// BlobProcessor is the interface for the blobs processor.
 	BlobProcessor[
 		AvailabilityStoreT any,
-		BeaconBlockBodyT any,
+		ConsensusSidecarsT any,
 		BlobSidecarsT any,
 	] interface {
 		// ProcessSidecars processes the blobs and ensures they match the local
@@ -239,7 +239,7 @@ type (
 		// VerifySidecars verifies the blobs and ensures they match the local
 		// state.
 		VerifySidecars(
-			sidecars BlobSidecarsT,
+			sidecars ConsensusSidecarsT,
 		) error
 	}