Allow validators registration batching on Builder API /eth/v1/builder/validators (#13178)

* builder `NewClient`: Simplify + fix some typos.

* Validator client: Implement `validator-registration-batch-size` option

* Address Potuz comments

* Address Potuz's comments

---------

Co-authored-by: Radosław Kapka <rkapka@wp.pl>
Co-authored-by: james-prysm <90280386+james-prysm@users.noreply.github.com>
Co-authored-by: prylabs-bulldozer[bot] <58059840+prylabs-bulldozer[bot]@users.noreply.github.com>

api/client/builder/BUILD.bazel

@@ -20,8 +20,6 @@ go_library(
         "//encoding/bytesutil:go_default_library",
         "//math:go_default_library",
         "//monitoring/tracing:go_default_library",
-        "//network:go_default_library",
-        "//network/authorization:go_default_library",
         "//proto/engine/v1:go_default_library",
         "//proto/prysm/v1alpha1:go_default_library",
         "//runtime/version:go_default_library",

api/client/builder/client.go

@@ -19,8 +19,6 @@ import (
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
 	"github.com/prysmaticlabs/prysm/v4/encoding/bytesutil"
 	"github.com/prysmaticlabs/prysm/v4/monitoring/tracing"
-	"github.com/prysmaticlabs/prysm/v4/network"
-	"github.com/prysmaticlabs/prysm/v4/network/authorization"
 	v1 "github.com/prysmaticlabs/prysm/v4/proto/engine/v1"
 	ethpb "github.com/prysmaticlabs/prysm/v4/proto/prysm/v1alpha1"
 	"github.com/prysmaticlabs/prysm/v4/runtime/version"
@@ -104,8 +102,7 @@ type Client struct {
 // `host` is the base host + port used to construct request urls. This value can be
 // a URL string, or NewClient will assume an http endpoint if just `host:port` is used.
 func NewClient(host string, opts ...ClientOpt) (*Client, error) {
-	endpoint := covertEndPoint(host)
-	u, err := urlForHost(endpoint.Url)
+	u, err := urlForHost(host)
 	if err != nil {
 		return nil, err
 	}
@@ -121,8 +118,7 @@ func NewClient(host string, opts ...ClientOpt) (*Client, error) {
 
 func urlForHost(h string) (*url.URL, error) {
 	// try to parse as url (being permissive)
-	u, err := url.Parse(h)
-	if err == nil && u.Host != "" {
+	if u, err := url.Parse(h); err == nil && u.Host != "" {
 		return u, nil
 	}
 	// try to parse as host:port
@@ -140,7 +136,7 @@ func (c *Client) NodeURL() string {
 
 type reqOption func(*http.Request)
 
-// do is a generic, opinionated request function to reduce boilerplate amongst the methods in this package api/client/builder/types.go.
+// do is a generic, opinionated request function to reduce boilerplate amongst the methods in this package api/client/builder.
 func (c *Client) do(ctx context.Context, method string, path string, body io.Reader, opts ...reqOption) (res []byte, err error) {
 	ctx, span := trace.StartSpan(ctx, "builder.client.do")
 	defer func() {
@@ -457,12 +453,3 @@ func non200Err(response *http.Response) error {
 		return errors.Wrap(ErrNotOK, fmt.Sprintf("unsupported error code: %d", response.StatusCode))
 	}
 }
-
-func covertEndPoint(ep string) network.Endpoint {
-	return network.Endpoint{
-		Url: ep,
-		Auth: network.AuthorizationData{ // Auth is not used for builder.
-			Method: authorization.None,
-			Value:  "",
-		}}
-}

api/client/builder/types.go

@@ -117,7 +117,7 @@ type VersionResponse struct {
 	Version string `json:"version"`
 }
 
-// ExecHeaderResponse is a JSON representation of  the builder API header response for Bellatrix.
+// ExecHeaderResponse is a JSON representation of the builder API header response for Bellatrix.
 type ExecHeaderResponse struct {
 	Version string `json:"version"`
 	Data    struct {

cmd/validator/flags/flags.go

@@ -346,7 +346,7 @@ var (
 		Name: "suggested-fee-recipient",
 		Usage: "Sets ALL validators' mapping to a suggested eth address to receive gas fees when proposing a block." +
 			" note that this is only a suggestion when integrating with a Builder API, which may choose to specify a different fee recipient as payment for the blocks it builds." +
-			" For additional setting overrides use the --" + ProposerSettingsFlag.Name + " or --" + ProposerSettingsURLFlag.Name + " Flags. ",
+			" For additional setting overrides use the --" + ProposerSettingsFlag.Name + " or --" + ProposerSettingsURLFlag.Name + " flags. ",
 		Value: params.BeaconConfig().EthBurnAddressHex,
 	}
 
@@ -364,6 +364,13 @@ var (
 		Usage: "Sets gas limit for the builder to use for constructing a payload for all the validators",
 		Value: fmt.Sprint(params.BeaconConfig().DefaultBuilderGasLimit),
 	}
+
+	// ValidatorRegistrationBatchSizeFlag sets the maximum size for one batch of validator registrations. Use a non-positive value to disable batching.
+	ValidatorRegistrationBatchSizeFlag = &cli.IntFlag{
+		Name:  "validator-registration-batch-size",
+		Usage: "Sets the maximum size for one batch of validator registrations. Use a non-positive value to disable batching.",
+		Value: 0,
+	}
 )
 
 // DefaultValidatorDir returns OS-specific default validator directory.

cmd/validator/main.go

@@ -79,6 +79,7 @@ var appFlags = []cli.Flag{
 	flags.ProposerSettingsFlag,
 	flags.EnableBuilderFlag,
 	flags.BuilderGasLimitFlag,
+	flags.ValidatorRegistrationBatchSizeFlag,
 	////////////////////
 	cmd.DisableMonitoringFlag,
 	cmd.MonitoringHostFlag,

cmd/validator/usage.go

@@ -113,6 +113,7 @@ var appHelpFlagGroups = []flagGroup{
 			flags.SuggestedFeeRecipientFlag,
 			flags.EnableBuilderFlag,
 			flags.BuilderGasLimitFlag,
+			flags.ValidatorRegistrationBatchSizeFlag,
 		},
 	},
 	{

encoding/bytesutil/bytes.go

@@ -136,7 +136,7 @@ func PadTo(b []byte, size int) []byte {
 }
 
 // ReverseByteOrder Switch the endianness of a byte slice by reversing its order.
-// this function does not modify the actual input bytes.
+// This function does not modify the actual input bytes.
 func ReverseByteOrder(input []byte) []byte {
 	b := make([]byte, len(input))
 	copy(b, input)

validator/client/registration.go

@@ -16,11 +16,13 @@ import (
 	"go.opencensus.io/trace"
 )
 
-// SubmitValidatorRegistrations signs validator registration objects and submits it to the beacon node.
+// SubmitValidatorRegistrations signs validator registration objects and submits it to the beacon node by batch of validatorRegsBatchSize size maximum.
+// If at least one error occurs during a registration call to the beacon node, the last error is returned.
 func SubmitValidatorRegistrations(
 	ctx context.Context,
 	validatorClient iface.ValidatorClient,
 	signedRegs []*ethpb.SignedValidatorRegistrationV1,
+	validatorRegsBatchSize int,
 ) error {
 	ctx, span := trace.StartSpan(ctx, "validator.SubmitValidatorRegistrations")
 	defer span.End()
@@ -29,17 +31,33 @@ func SubmitValidatorRegistrations(
 		return nil
 	}
 
-	if _, err := validatorClient.SubmitValidatorRegistrations(ctx, &ethpb.SignedValidatorRegistrationsV1{
-		Messages: signedRegs,
-	}); err != nil {
-		if strings.Contains(err.Error(), builder.ErrNoBuilder.Error()) {
-			log.Warnln("Beacon node does not utilize a custom builder via the --http-mev-relay flag. Validator registration skipped.")
-			return nil
+	chunks := chunkSignedValidatorRegistrationV1(signedRegs, validatorRegsBatchSize)
+	var lastErr error
+
+	for _, chunk := range chunks {
+		innerSignerRegs := ethpb.SignedValidatorRegistrationsV1{
+			Messages: chunk,
+		}
+
+		if _, err := validatorClient.SubmitValidatorRegistrations(ctx, &innerSignerRegs); err != nil {
+			lastErr = errors.Wrap(err, "could not submit signed registrations to beacon node")
+
+			if strings.Contains(err.Error(), builder.ErrNoBuilder.Error()) {
+				log.Warnln("Beacon node does not utilize a custom builder via the --http-mev-relay flag. Validator registration skipped.")
+
+				// We stop early the loop here, since if the builder endpoint is not configured for this chunk, it is useless to check the following chunks
+				break
+			}
 		}
-		return errors.Wrap(err, "could not submit signed registrations to beacon node")
 	}
-	log.Infoln("Submitted builder validator registration settings for custom builders")
-	return nil
+
+	if lastErr == nil {
+		log.Infoln("Submitted builder validator registration settings for custom builders")
+	} else {
+		log.WithError(lastErr).Warn("Could not submit all signed registrations to beacon node")
+	}
+
+	return lastErr
 }
 
 // Sings validator registration obj with the proposer domain and private key.
@@ -100,3 +118,35 @@ func isValidatorRegistrationSame(cachedVR *ethpb.ValidatorRegistrationV1, newVR
 	}
 	return isSame
 }
+
+// chunkSignedValidatorRegistrationV1 chunks regs into chunks of size chunkSize (the last chunk may be smaller). If chunkSize is non-positive, returns only one chunk.
+func chunkSignedValidatorRegistrationV1(regs []*ethpb.SignedValidatorRegistrationV1, chunkSize int) [][]*ethpb.SignedValidatorRegistrationV1 {
+	if chunkSize <= 0 {
+		chunkSize = len(regs)
+	}
+
+	regsCount := len(regs)
+
+	chunksCount := (regsCount + chunkSize - 1) / chunkSize
+	lastChunkSize := regsCount % chunkSize
+
+	if lastChunkSize == 0 {
+		lastChunkSize = chunkSize
+	}
+
+	chunks := make([][]*ethpb.SignedValidatorRegistrationV1, chunksCount)
+
+	for i := 0; i < chunksCount-1; i++ {
+		chunks[i] = make([]*ethpb.SignedValidatorRegistrationV1, chunkSize)
+	}
+
+	chunks[chunksCount-1] = make([]*ethpb.SignedValidatorRegistrationV1, lastChunkSize)
+
+	for i, reg := range regs {
+		chunkIndex := i / chunkSize
+		chunkOffset := i % chunkSize
+		chunks[chunkIndex][chunkOffset] = reg
+	}
+
+	return chunks
+}

validator/client/registration_test.go

@@ -21,27 +21,76 @@ func TestSubmitValidatorRegistrations(t *testing.T) {
 	defer finish()
 
 	ctx := context.Background()
-	require.NoError(t, nil, SubmitValidatorRegistrations(ctx, m.validatorClient, []*ethpb.SignedValidatorRegistrationV1{}))
+	validatorRegsBatchSize := 2
+	require.NoError(t, nil, SubmitValidatorRegistrations(ctx, m.validatorClient, []*ethpb.SignedValidatorRegistrationV1{}, validatorRegsBatchSize))
 
-	reg := &ethpb.ValidatorRegistrationV1{
-		FeeRecipient: bytesutil.PadTo([]byte("fee"), 20),
-		GasLimit:     123456,
-		Timestamp:    uint64(time.Now().Unix()),
-		Pubkey:       validatorKey.PublicKey().Marshal(),
+	regs := [...]*ethpb.ValidatorRegistrationV1{
+		{
+			FeeRecipient: bytesutil.PadTo([]byte("fee"), 20),
+			GasLimit:     123,
+			Timestamp:    uint64(time.Now().Unix()),
+			Pubkey:       validatorKey.PublicKey().Marshal(),
+		},
+		{
+			FeeRecipient: bytesutil.PadTo([]byte("fee"), 20),
+			GasLimit:     456,
+			Timestamp:    uint64(time.Now().Unix()),
+			Pubkey:       validatorKey.PublicKey().Marshal(),
+		},
+		{
+			FeeRecipient: bytesutil.PadTo([]byte("fee"), 20),
+			GasLimit:     789,
+			Timestamp:    uint64(time.Now().Unix()),
+			Pubkey:       validatorKey.PublicKey().Marshal(),
+		},
 	}
 
-	m.validatorClient.EXPECT().
-		SubmitValidatorRegistrations(gomock.Any(), &ethpb.SignedValidatorRegistrationsV1{
-			Messages: []*ethpb.SignedValidatorRegistrationV1{
-				{Message: reg,
-					Signature: params.BeaconConfig().ZeroHash[:]},
+	gomock.InOrder(
+		m.validatorClient.EXPECT().
+			SubmitValidatorRegistrations(gomock.Any(), &ethpb.SignedValidatorRegistrationsV1{
+				Messages: []*ethpb.SignedValidatorRegistrationV1{
+					{
+						Message:   regs[0],
+						Signature: params.BeaconConfig().ZeroHash[:],
+					},
+					{
+						Message:   regs[1],
+						Signature: params.BeaconConfig().ZeroHash[:],
+					},
+				},
+			}).
+			Return(nil, nil),
+
+		m.validatorClient.EXPECT().
+			SubmitValidatorRegistrations(gomock.Any(), &ethpb.SignedValidatorRegistrationsV1{
+				Messages: []*ethpb.SignedValidatorRegistrationV1{
+					{
+						Message:   regs[2],
+						Signature: params.BeaconConfig().ZeroHash[:],
+					},
+				},
+			}).
+			Return(nil, nil),
+	)
+
+	require.NoError(t, nil, SubmitValidatorRegistrations(
+		ctx, m.validatorClient,
+		[]*ethpb.SignedValidatorRegistrationV1{
+			{
+				Message:   regs[0],
+				Signature: params.BeaconConfig().ZeroHash[:],
 			},
-		}).
-		Return(nil, nil)
-	require.NoError(t, nil, SubmitValidatorRegistrations(ctx, m.validatorClient, []*ethpb.SignedValidatorRegistrationV1{
-		{Message: reg,
-			Signature: params.BeaconConfig().ZeroHash[:]},
-	}))
+			{
+				Message:   regs[1],
+				Signature: params.BeaconConfig().ZeroHash[:],
+			},
+			{
+				Message:   regs[2],
+				Signature: params.BeaconConfig().ZeroHash[:],
+			},
+		},
+		validatorRegsBatchSize,
+	))
 }
 
 func TestSubmitValidatorRegistration_CantSign(t *testing.T) {
@@ -49,6 +98,7 @@ func TestSubmitValidatorRegistration_CantSign(t *testing.T) {
 	defer finish()
 
 	ctx := context.Background()
+	validatorRegsBatchSize := 500
 	reg := &ethpb.ValidatorRegistrationV1{
 		FeeRecipient: bytesutil.PadTo([]byte("fee"), 20),
 		GasLimit:     123456,
@@ -67,7 +117,7 @@ func TestSubmitValidatorRegistration_CantSign(t *testing.T) {
 	require.ErrorContains(t, "could not sign", SubmitValidatorRegistrations(ctx, m.validatorClient, []*ethpb.SignedValidatorRegistrationV1{
 		{Message: reg,
 			Signature: params.BeaconConfig().ZeroHash[:]},
-	}))
+	}, validatorRegsBatchSize))
 }
 
 func Test_signValidatorRegistration(t *testing.T) {
@@ -225,3 +275,113 @@ func TestValidator_SignValidatorRegistrationRequest(t *testing.T) {
 		})
 	}
 }
+
+func TestChunkSignedValidatorRegistrationV1(t *testing.T) {
+	tests := map[string]struct {
+		regs      []*ethpb.SignedValidatorRegistrationV1
+		chunkSize int
+		expected  [][]*ethpb.SignedValidatorRegistrationV1
+	}{
+		"All buckets are full": {
+			regs: []*ethpb.SignedValidatorRegistrationV1{
+				{Signature: []byte("1")},
+				{Signature: []byte("2")},
+				{Signature: []byte("3")},
+				{Signature: []byte("4")},
+				{Signature: []byte("5")},
+				{Signature: []byte("6")},
+			},
+			chunkSize: 3,
+			expected: [][]*ethpb.SignedValidatorRegistrationV1{
+				{
+					{Signature: []byte("1")},
+					{Signature: []byte("2")},
+					{Signature: []byte("3")},
+				},
+				{
+					{Signature: []byte("4")},
+					{Signature: []byte("5")},
+					{Signature: []byte("6")},
+				},
+			},
+		},
+		"Last bucket is not full": {
+			regs: []*ethpb.SignedValidatorRegistrationV1{
+				{Signature: []byte("1")},
+				{Signature: []byte("2")},
+				{Signature: []byte("3")},
+				{Signature: []byte("4")},
+				{Signature: []byte("5")},
+				{Signature: []byte("6")},
+				{Signature: []byte("7")},
+			},
+			chunkSize: 3,
+			expected: [][]*ethpb.SignedValidatorRegistrationV1{
+				{
+					{Signature: []byte("1")},
+					{Signature: []byte("2")},
+					{Signature: []byte("3")},
+				},
+				{
+					{Signature: []byte("4")},
+					{Signature: []byte("5")},
+					{Signature: []byte("6")},
+				},
+				{
+					{Signature: []byte("7")},
+				},
+			},
+		},
+		"Not enough items": {
+			regs: []*ethpb.SignedValidatorRegistrationV1{
+				{Signature: []byte("1")},
+				{Signature: []byte("2")},
+				{Signature: []byte("3")},
+			},
+			chunkSize: 42,
+			expected: [][]*ethpb.SignedValidatorRegistrationV1{
+				{
+					{Signature: []byte("1")},
+					{Signature: []byte("2")},
+					{Signature: []byte("3")},
+				},
+			},
+		},
+		"Null chunk size": {
+			regs: []*ethpb.SignedValidatorRegistrationV1{
+				{Signature: []byte("1")},
+				{Signature: []byte("2")},
+				{Signature: []byte("3")},
+			},
+			chunkSize: 0,
+			expected: [][]*ethpb.SignedValidatorRegistrationV1{
+				{
+					{Signature: []byte("1")},
+					{Signature: []byte("2")},
+					{Signature: []byte("3")},
+				},
+			},
+		},
+		"Negative chunk size": {
+			regs: []*ethpb.SignedValidatorRegistrationV1{
+				{Signature: []byte("1")},
+				{Signature: []byte("2")},
+				{Signature: []byte("3")},
+			},
+			chunkSize: -1,
+			expected: [][]*ethpb.SignedValidatorRegistrationV1{
+				{
+					{Signature: []byte("1")},
+					{Signature: []byte("2")},
+					{Signature: []byte("3")},
+				},
+			},
+		},
+	}
+
+	for name, test := range tests {
+		t.Run(name, func(t *testing.T) {
+			require.DeepEqual(t, test.expected, chunkSignedValidatorRegistrationV1(test.regs, test.chunkSize))
+		})
+	}
+}

validator/client/service.go

@@ -74,6 +74,7 @@ type ValidatorService struct {
 	graffiti              []byte
 	Web3SignerConfig      *remoteweb3signer.SetupConfig
 	proposerSettings      *validatorserviceconfig.ProposerSettings
+	validatorRegBatchSize int
 }
 
 // Config for the validator service.
@@ -99,6 +100,7 @@ type Config struct {
 	ProposerSettings           *validatorserviceconfig.ProposerSettings
 	BeaconApiEndpoint          string
 	BeaconApiTimeout           time.Duration
+	ValidatorRegBatchSize      int
 }
 
 // NewValidatorService creates a new validator service for the service
@@ -127,6 +129,7 @@ func NewValidatorService(ctx context.Context, cfg *Config) (*ValidatorService, e
 		graffitiStruct:        cfg.GraffitiStruct,
 		Web3SignerConfig:      cfg.Web3SignerConfig,
 		proposerSettings:      cfg.ProposerSettings,
+		validatorRegBatchSize: cfg.ValidatorRegBatchSize,
 	}
 
 	dialOpts := ConstructDialOptions(
@@ -220,6 +223,7 @@ func (v *ValidatorService) Start() {
 		proposerSettings:               v.proposerSettings,
 		walletInitializedChannel:       make(chan *wallet.Wallet, 1),
 		prysmBeaconClient:              prysmBeaconClient,
+		validatorRegBatchSize:          v.validatorRegBatchSize,
 	}
 
 	// To resolve a race condition at startup due to the interface

validator/client/validator.go

@@ -105,6 +105,7 @@ type validator struct {
 	proposerSettings                   *validatorserviceconfig.ProposerSettings
 	walletInitializedChannel           chan *wallet.Wallet
 	prysmBeaconClient                  iface.PrysmBeaconChainClient
+	validatorRegBatchSize              int
 }
 
 type validatorStatus struct {
@@ -1043,7 +1044,7 @@ func (v *validator) PushProposerSettings(ctx context.Context, km keymanager.IKey
 	if err != nil {
 		return err
 	}
-	if err := SubmitValidatorRegistrations(ctx, v.validatorClient, signedRegReqs); err != nil {
+	if err := SubmitValidatorRegistrations(ctx, v.validatorClient, signedRegReqs, v.validatorRegBatchSize); err != nil {
 		return errors.Wrap(ErrBuilderValidatorRegistration, err.Error())
 	}
 

validator/node/node.go

@@ -439,6 +439,7 @@ func (c *ValidatorClient) registerValidatorService(cliCtx *cli.Context) error {
 		ProposerSettings:           bpc,
 		BeaconApiTimeout:           time.Second * 30,
 		BeaconApiEndpoint:          c.cliCtx.String(flags.BeaconRESTApiProviderFlag.Name),
+		ValidatorRegBatchSize:      c.cliCtx.Int(flags.ValidatorRegistrationBatchSizeFlag.Name),
 	})
 	if err != nil {
 		return errors.Wrap(err, "could not initialize validator service")