Deneb: blob sidecar events (#12928)

* adding in deneb blob event triggers

* fixing linting

* kasey's feedback

---------

Co-authored-by: prylabs-bulldozer[bot] <58059840+prylabs-bulldozer[bot]@users.noreply.github.com>

beacon-chain/blockchain/execution_engine.go

@@ -405,8 +405,13 @@ func kzgCommitmentsToVersionedHashes(body interfaces.ReadOnlyBeaconBlockBody) ([
 
 	versionedHashes := make([]common.Hash, len(commitments))
 	for i, commitment := range commitments {
-		versionedHashes[i] = sha256.Sum256(commitment)
-		versionedHashes[i][0] = blobCommitmentVersionKZG
+		versionedHashes[i] = ConvertKzgCommitmentToVersionedHash(commitment)
 	}
 	return versionedHashes, nil
 }
+
+func ConvertKzgCommitmentToVersionedHash(commitment []byte) common.Hash {
+	versionedHash := sha256.Sum256(commitment)
+	versionedHash[0] = blobCommitmentVersionKZG
+	return versionedHash
+}

beacon-chain/core/feed/operation/events.go

@@ -22,6 +22,9 @@ const (
 
 	// BLSToExecutionChangeReceived is sent after a BLS to execution change object has been received from gossip or rpc.
 	BLSToExecutionChangeReceived
+
+	// BlobSidecarReceived is sent after a blob sidecar is received from gossip or rpc.
+	BlobSidecarReceived = 6
 )
 
 // UnAggregatedAttReceivedData is the data sent with UnaggregatedAttReceived events.
@@ -52,3 +55,8 @@ type SyncCommitteeContributionReceivedData struct {
 type BLSToExecutionChangeReceivedData struct {
 	Change *ethpb.SignedBLSToExecutionChange
 }
+
+// BlobSidecarReceivedData is the data sent with BlobSidecarReceived events.
+type BlobSidecarReceivedData struct {
+	Blob *ethpb.SignedBlobSidecar
+}

beacon-chain/rpc/eth/events/BUILD.bazel

@@ -16,6 +16,7 @@ go_library(
         "//beacon-chain/core/helpers:go_default_library",
         "//beacon-chain/core/time:go_default_library",
         "//beacon-chain/core/transition:go_default_library",
+        "//encoding/bytesutil:go_default_library",
         "//proto/engine/v1:go_default_library",
         "//proto/eth/service:go_default_library",
         "//proto/eth/v1:go_default_library",
@@ -38,6 +39,7 @@ go_test(
     embed = [":go_default_library"],
     deps = [
         "//async/event:go_default_library",
+        "//beacon-chain/blockchain:go_default_library",
         "//beacon-chain/blockchain/testing:go_default_library",
         "//beacon-chain/core/blocks:go_default_library",
         "//beacon-chain/core/feed:go_default_library",
@@ -47,6 +49,7 @@ go_test(
         "//beacon-chain/core/time:go_default_library",
         "//config/fieldparams:go_default_library",
         "//consensus-types/blocks:go_default_library",
+        "//encoding/bytesutil:go_default_library",
         "//proto/engine/v1:go_default_library",
         "//proto/eth/v1:go_default_library",
         "//proto/migration:go_default_library",

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

@@ -5,12 +5,14 @@ import (
 
 	gwpb "github.com/grpc-ecosystem/grpc-gateway/v2/proto/gateway"
 	"github.com/pkg/errors"
+	"github.com/prysmaticlabs/prysm/v4/beacon-chain/blockchain"
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/core/feed"
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/core/feed/operation"
 	statefeed "github.com/prysmaticlabs/prysm/v4/beacon-chain/core/feed/state"
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/core/helpers"
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/core/time"
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/core/transition"
+	"github.com/prysmaticlabs/prysm/v4/encoding/bytesutil"
 	enginev1 "github.com/prysmaticlabs/prysm/v4/proto/engine/v1"
 	ethpbservice "github.com/prysmaticlabs/prysm/v4/proto/eth/service"
 	ethpb "github.com/prysmaticlabs/prysm/v4/proto/eth/v1"
@@ -43,6 +45,8 @@ const (
 	BLSToExecutionChangeTopic = "bls_to_execution_change"
 	// PayloadAttributesTopic represents a new payload attributes for execution payload building event topic.
 	PayloadAttributesTopic = "payload_attributes"
+	// BlobSidecarTopic represents a new blob sidecar event topic
+	BlobSidecarTopic = "blob_sidecar"
 )
 
 var casesHandled = map[string]bool{
@@ -55,6 +59,7 @@ var casesHandled = map[string]bool{
 	SyncCommitteeContributionTopic: true,
 	BLSToExecutionChangeTopic:      true,
 	PayloadAttributesTopic:         true,
+	BlobSidecarTopic:               true,
 }
 
 // StreamEvents allows requesting all events from a set of topics defined in the Ethereum consensus API standard.
@@ -161,7 +166,26 @@ func handleBlockOperationEvents(
 		}
 		v2Change := migration.V1Alpha1SignedBLSToExecChangeToV2(changeData.Change)
 		return streamData(stream, BLSToExecutionChangeTopic, v2Change)
-
+	case operation.BlobSidecarReceived:
+		if _, ok := requestedTopics[BlobSidecarTopic]; !ok {
+			return nil
+		}
+		blobData, ok := event.Data.(*operation.BlobSidecarReceivedData)
+		if !ok {
+			return nil
+		}
+		if blobData == nil || blobData.Blob == nil {
+			return nil
+		}
+		versionedHash := blockchain.ConvertKzgCommitmentToVersionedHash(blobData.Blob.Message.KzgCommitment)
+		blobEvent := &ethpb.EventBlobSidecar{
+			BlockRoot:     bytesutil.SafeCopyBytes(blobData.Blob.Message.BlockRoot),
+			Index:         blobData.Blob.Message.Index,
+			Slot:          blobData.Blob.Message.Slot,
+			VersionedHash: bytesutil.SafeCopyBytes(versionedHash.Bytes()),
+			KzgCommitment: bytesutil.SafeCopyBytes(blobData.Blob.Message.KzgCommitment),
+		}
+		return streamData(stream, BlobSidecarTopic, blobEvent)
 	default:
 		return nil
 	}

beacon-chain/rpc/eth/events/events_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/grpc-ecosystem/grpc-gateway/v2/proto/gateway"
 	"github.com/prysmaticlabs/go-bitfield"
 	"github.com/prysmaticlabs/prysm/v4/async/event"
+	"github.com/prysmaticlabs/prysm/v4/beacon-chain/blockchain"
 	mockChain "github.com/prysmaticlabs/prysm/v4/beacon-chain/blockchain/testing"
 	b "github.com/prysmaticlabs/prysm/v4/beacon-chain/core/blocks"
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/core/feed"
@@ -19,6 +20,7 @@ import (
 	prysmtime "github.com/prysmaticlabs/prysm/v4/beacon-chain/core/time"
 	fieldparams "github.com/prysmaticlabs/prysm/v4/config/fieldparams"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/blocks"
+	"github.com/prysmaticlabs/prysm/v4/encoding/bytesutil"
 	enginev1 "github.com/prysmaticlabs/prysm/v4/proto/engine/v1"
 	ethpb "github.com/prysmaticlabs/prysm/v4/proto/eth/v1"
 	"github.com/prysmaticlabs/prysm/v4/proto/migration"
@@ -233,6 +235,52 @@ func TestStreamEvents_OperationsEvents(t *testing.T) {
 			feed: srv.OperationNotifier.OperationFeed(),
 		})
 	})
+	t.Run(BlobSidecarTopic, func(t *testing.T) {
+		ctx := context.Background()
+		srv, ctrl, mockStream := setupServer(ctx, t)
+		defer ctrl.Finish()
+		commitment, err := hexutil.Decode("0x1b66ac1fb663c9bc59509846d6ec05345bd908eda73e670af888da41af171505cc411d61252fb6cb3fa0017b679f8000")
+		require.NoError(t, err)
+		wantedBlobV1alpha1 := &eth.SignedBlobSidecar{
+			Message: &eth.BlobSidecar{
+				BlockRoot:     make([]byte, fieldparams.RootLength),
+				Index:         1,
+				Slot:          3,
+				KzgCommitment: commitment,
+			},
+			Signature: make([]byte, 96),
+		}
+		versionedHash := blockchain.ConvertKzgCommitmentToVersionedHash(commitment)
+		blobEvent := &ethpb.EventBlobSidecar{
+			BlockRoot:     bytesutil.SafeCopyBytes(wantedBlobV1alpha1.Message.BlockRoot),
+			Index:         wantedBlobV1alpha1.Message.Index,
+			Slot:          wantedBlobV1alpha1.Message.Slot,
+			VersionedHash: bytesutil.SafeCopyBytes(versionedHash.Bytes()),
+			KzgCommitment: bytesutil.SafeCopyBytes(wantedBlobV1alpha1.Message.KzgCommitment),
+		}
+		genericResponse, err := anypb.New(blobEvent)
+		require.NoError(t, err)
+
+		wantedMessage := &gateway.EventSource{
+			Event: BlobSidecarTopic,
+			Data:  genericResponse,
+		}
+
+		assertFeedSendAndReceive(ctx, &assertFeedArgs{
+			t:             t,
+			srv:           srv,
+			topics:        []string{BlobSidecarTopic},
+			stream:        mockStream,
+			shouldReceive: wantedMessage,
+			itemToSend: &feed.Event{
+				Type: operation.BlobSidecarReceived,
+				Data: &operation.BlobSidecarReceivedData{
+					Blob: wantedBlobV1alpha1,
+				},
+			},
+			feed: srv.OperationNotifier.OperationFeed(),
+		})
+	})
 }
 
 func TestStreamEvents_StateEvents(t *testing.T) {

beacon-chain/sync/subscriber_blob_sidecar.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/prysmaticlabs/prysm/v4/beacon-chain/core/feed"
+	opfeed "github.com/prysmaticlabs/prysm/v4/beacon-chain/core/feed/operation"
 	eth "github.com/prysmaticlabs/prysm/v4/proto/prysm/v1alpha1"
 	"google.golang.org/protobuf/proto"
 )
@@ -22,5 +24,11 @@ func (s *Service) blobSubscriber(ctx context.Context, msg proto.Message) error {
 
 	s.cfg.chain.SendNewBlobEvent([32]byte(b.Message.BlockRoot), b.Message.Index)
 
+	s.cfg.operationNotifier.OperationFeed().Send(&feed.Event{
+		Type: opfeed.BlobSidecarReceived,
+		Data: &opfeed.BlobSidecarReceivedData{
+			Blob: b,
+		},
+	})
 	return nil
 }