Integrate builder client into builder service (#10825)

* Integrate builder client into builder service

* Do nothing if pubkey is not found

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

beacon-chain/builder/BUILD.bazel

@@ -3,6 +3,8 @@ load("@prysm//tools/go:def.bzl", "go_library")
 go_library(
     name = "go_default_library",
     srcs = [
+        "error.go",
+        "metric.go",
         "option.go",
         "service.go",
     ],
@@ -10,13 +12,19 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//api/client/builder:go_default_library",
+        "//beacon-chain/blockchain:go_default_library",
+        "//beacon-chain/db:go_default_library",
         "//cmd/beacon-chain/flags:go_default_library",
         "//consensus-types/primitives:go_default_library",
+        "//encoding/bytesutil:go_default_library",
         "//network:go_default_library",
         "//network/authorization:go_default_library",
         "//proto/engine/v1:go_default_library",
         "//proto/prysm/v1alpha1:go_default_library",
         "@com_github_pkg_errors//:go_default_library",
+        "@com_github_prometheus_client_golang//prometheus:go_default_library",
+        "@com_github_prometheus_client_golang//prometheus/promauto:go_default_library",
         "@com_github_urfave_cli_v2//:go_default_library",
+        "@io_opencensus_go//trace:go_default_library",
     ],
 )

beacon-chain/builder/error.go

@@ -0,0 +1,7 @@
+package builder
+
+import "github.com/pkg/errors"
+
+var (
+	ErrNotRunning = errors.New("builder is not running")
+)

beacon-chain/builder/metric.go

@@ -0,0 +1,37 @@
+package builder
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+var (
+	submitBlindedBlockLatency = promauto.NewHistogram(
+		prometheus.HistogramOpts{
+			Name:    "submit_blinded_block_latency_milliseconds",
+			Help:    "Captures RPC latency for submitting blinded block in milliseconds",
+			Buckets: []float64{1, 2, 5, 10, 20, 50, 100, 200, 500, 1000},
+		},
+	)
+	getHeaderLatency = promauto.NewHistogram(
+		prometheus.HistogramOpts{
+			Name:    "get_header_latency_milliseconds",
+			Help:    "Captures RPC latency for get header in milliseconds",
+			Buckets: []float64{1, 2, 5, 10, 20, 50, 100, 200, 500, 1000},
+		},
+	)
+	getStatusLatency = promauto.NewHistogram(
+		prometheus.HistogramOpts{
+			Name:    "get_status_latency_milliseconds",
+			Help:    "Captures RPC latency for get status in milliseconds",
+			Buckets: []float64{1, 2, 5, 10, 20, 50, 100, 200, 500, 1000},
+		},
+	)
+	registerValidatorLatency = promauto.NewHistogram(
+		prometheus.HistogramOpts{
+			Name:    "register_validator_latency_milliseconds",
+			Help:    "Captures RPC latency for register validator in milliseconds",
+			Buckets: []float64{1, 2, 5, 10, 20, 50, 100, 200, 500, 1000},
+		},
+	)
+)

beacon-chain/builder/option.go

@@ -1,6 +1,7 @@
 package builder
 
 import (
+	"github.com/prysmaticlabs/prysm/beacon-chain/db"
 	"github.com/prysmaticlabs/prysm/cmd/beacon-chain/flags"
 	"github.com/prysmaticlabs/prysm/network"
 	"github.com/prysmaticlabs/prysm/network/authorization"
@@ -26,6 +27,14 @@ func WithBuilderEndpoints(endpoint string) Option {
 	}
 }
 
+// WithDatabase sets the database for the beacon chain builder service.
+func WithDatabase(database db.HeadAccessDatabase) Option {
+	return func(s *Service) error {
+		s.cfg.beaconDB = database
+		return nil
+	}
+}
+
 func covertEndPoint(ep string) network.Endpoint {
 	return network.Endpoint{
 		Url: ep,

beacon-chain/builder/service.go

@@ -2,26 +2,33 @@ package builder
 
 import (
 	"context"
+	"time"
 
 	"github.com/pkg/errors"
 	"github.com/prysmaticlabs/prysm/api/client/builder"
+	"github.com/prysmaticlabs/prysm/beacon-chain/blockchain"
+	"github.com/prysmaticlabs/prysm/beacon-chain/db"
 	types "github.com/prysmaticlabs/prysm/consensus-types/primitives"
+	"github.com/prysmaticlabs/prysm/encoding/bytesutil"
 	"github.com/prysmaticlabs/prysm/network"
 	v1 "github.com/prysmaticlabs/prysm/proto/engine/v1"
 	ethpb "github.com/prysmaticlabs/prysm/proto/prysm/v1alpha1"
+	"go.opencensus.io/trace"
 )
 
 // BlockBuilder defines the interface for interacting with the block builder
 type BlockBuilder interface {
 	SubmitBlindedBlock(ctx context.Context, block *ethpb.SignedBlindedBeaconBlockBellatrix) (*v1.ExecutionPayload, error)
 	GetHeader(ctx context.Context, slot types.Slot, parentHash [32]byte, pubKey [48]byte) (*ethpb.SignedBuilderBid, error)
-	Status() error
+	Status(ctx context.Context) error
 	RegisterValidator(ctx context.Context, reg *ethpb.SignedValidatorRegistrationV1) error
 }
 
 // config defines a config struct for dependencies into the service.
 type config struct {
 	builderEndpoint network.Endpoint
+	beaconDB        db.HeadAccessDatabase
+	headFetcher     blockchain.HeadFetcher
 }
 
 // Service defines a service that provides a client for interacting with the beacon chain and MEV relay network.
@@ -56,22 +63,59 @@ func (*Service) Stop() error {
 	return nil
 }
 
-// SubmitBlindedBlock is currently a stub.
-func (*Service) SubmitBlindedBlock(context.Context, *ethpb.SignedBlindedBeaconBlockBellatrix) (*v1.ExecutionPayload, error) {
-	return nil, errors.New("not implemented")
+// SubmitBlindedBlock submits a blinded block to the builder relay network.
+func (s *Service) SubmitBlindedBlock(ctx context.Context, b *ethpb.SignedBlindedBeaconBlockBellatrix) (*v1.ExecutionPayload, error) {
+	ctx, span := trace.StartSpan(ctx, "builder.SubmitBlindedBlock")
+	defer span.End()
+	start := time.Now()
+	defer func() {
+		submitBlindedBlockLatency.Observe(float64(time.Since(start).Milliseconds()))
+	}()
+
+	return s.c.SubmitBlindedBlock(ctx, b)
 }
 
-// GetHeader is currently a stub.
-func (*Service) GetHeader(context.Context, types.Slot, [32]byte, [48]byte) (*ethpb.SignedBuilderBid, error) {
-	return nil, errors.New("not implemented")
+// GetHeader retrieves the header for a given slot and parent hash from the builder relay network.
+func (s *Service) GetHeader(ctx context.Context, slot types.Slot, parentHash [32]byte, pubKey [48]byte) (*ethpb.SignedBuilderBid, error) {
+	ctx, span := trace.StartSpan(ctx, "builder.GetHeader")
+	defer span.End()
+	start := time.Now()
+	defer func() {
+		getHeaderLatency.Observe(float64(time.Since(start).Milliseconds()))
+	}()
+
+	return s.c.GetHeader(ctx, slot, parentHash, pubKey)
 }
 
-// Status is currently a stub.
-func (*Service) Status() error {
-	return errors.New("not implemented")
+// Status retrieves the status of the builder relay network.
+func (s *Service) Status(ctx context.Context) error {
+	ctx, span := trace.StartSpan(ctx, "builder.Status")
+	defer span.End()
+	start := time.Now()
+	defer func() {
+		getStatusLatency.Observe(float64(time.Since(start).Milliseconds()))
+	}()
+
+	return s.c.Status(ctx)
 }
 
-// RegisterValidator is currently a stub.
-func (*Service) RegisterValidator(context.Context, *ethpb.SignedValidatorRegistrationV1) error {
-	return errors.New("not implemented")
+// RegisterValidator registers a validator with the builder relay network.
+// It also saves the registration object to the DB.
+func (s *Service) RegisterValidator(ctx context.Context, reg *ethpb.SignedValidatorRegistrationV1) error {
+	ctx, span := trace.StartSpan(ctx, "builder.RegisterValidator")
+	defer span.End()
+	start := time.Now()
+	defer func() {
+		registerValidatorLatency.Observe(float64(time.Since(start).Milliseconds()))
+	}()
+
+	idx, exists := s.cfg.headFetcher.HeadPublicKeyToValidatorIndex(bytesutil.ToBytes48(reg.Message.Pubkey))
+	if !exists {
+		return nil // If the pubkey is not found, it is not a validator. Do nothing.
+	}
+	if err := s.c.RegisterValidator(ctx, reg); err != nil {
+		return errors.Wrap(err, "could not register validator")
+	}
+
+	return s.cfg.beaconDB.SaveRegistrationsByValidatorIDs(ctx, []types.ValidatorIndex{idx}, []*ethpb.ValidatorRegistrationV1{reg.Message})
 }

beacon-chain/db/iface/interface.go

@@ -86,6 +86,7 @@ type NoHeadAccessDatabase interface {
 	RunMigrations(ctx context.Context) error
 	// Fee reicipients operations.
 	SaveFeeRecipientsByValidatorIDs(ctx context.Context, ids []types.ValidatorIndex, addrs []common.Address) error
+	SaveRegistrationsByValidatorIDs(ctx context.Context, ids []types.ValidatorIndex, regs []*ethpb.ValidatorRegistrationV1) error
 
 	CleanUpDirtyStates(ctx context.Context, slotsPerArchivedPoint types.Slot) error
 }