Forward validator registrations without decoding

server/register_validator.go

@@ -0,0 +1,100 @@
+package server
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"net/http"
+	"net/url"
+
+	"github.com/flashbots/mev-boost/server/params"
+	"github.com/flashbots/mev-boost/server/types"
+	"github.com/sirupsen/logrus"
+)
+
+func (m *BoostService) registerValidator(log *logrus.Entry, regBytes []byte, header http.Header) error {
+	respErrCh := make(chan error, len(m.relays))
+
+	// Forward request to each relay
+	for _, relay := range m.relays {
+		go func(relay types.RelayEntry) {
+			// Get the URL for this relay
+			requestURL := relay.GetURI(params.PathRegisterValidator)
+			log := log.WithField("url", requestURL)
+
+			// Build the new request
+			req, err := http.NewRequestWithContext(context.Background(), http.MethodPost, requestURL, bytes.NewReader(regBytes))
+			if err != nil {
+				log.WithError(err).Warn("error creating new request")
+				respErrCh <- err
+				return
+			}
+
+			// Extend the request header with our values
+			for key, values := range header {
+				req.Header[key] = values
+			}
+
+			// Send the request
+			resp, err := m.httpClientRegVal.Do(req)
+			if err != nil {
+				log.WithError(err).Warn("error calling registerValidator on relay")
+				respErrCh <- err
+				return
+			}
+			resp.Body.Close()
+
+			// Check if response is successful
+			if resp.StatusCode == http.StatusOK {
+				respErrCh <- nil
+			} else {
+				respErrCh <- fmt.Errorf("%w: %d", errHTTPErrorResponse, resp.StatusCode)
+			}
+		}(relay)
+	}
+
+	// Return OK if any relay responds OK
+	for range m.relays {
+		respErr := <-respErrCh
+		if respErr == nil {
+			// Goroutines are independent, so if there are a lot of configured
+			// relays and the first one responds OK, this will continue to send
+			// validator registrations to the other relays.
+			return nil
+		}
+	}
+
+	// None of the relays responded OK
+	return errNoSuccessfulRelayResponse
+}
+
+func (m *BoostService) sendValidatorRegistrationsToRelayMonitors(log *logrus.Entry, regBytes []byte, header http.Header) {
+	// Forward request to each relay monitor
+	for _, relayMonitor := range m.relayMonitors {
+		go func(relayMonitor *url.URL) {
+			// Get the URL for this relay monitor
+			requestURL := types.GetURI(relayMonitor, params.PathRegisterValidator)
+			log := log.WithField("url", requestURL)
+
+			// Build the new request
+			req, err := http.NewRequestWithContext(context.Background(), http.MethodPost, requestURL, bytes.NewReader(regBytes))
+			if err != nil {
+				log.WithError(err).Warn("error creating new request")
+				return
+			}
+
+			// Extend the request header with our values
+			for key, values := range header {
+				req.Header[key] = values
+			}
+
+			// Send the request
+			resp, err := m.httpClientRegVal.Do(req)
+			if err != nil {
+				log.WithError(err).Warn("error calling registerValidator on relay monitor")
+				return
+			}
+			resp.Body.Close()
+		}(relayMonitor)
+	}
+}

server/service.go

@@ -16,7 +16,6 @@ import (
 	"time"
 
 	builderApi "github.com/attestantio/go-builder-client/api"
-	builderApiV1 "github.com/attestantio/go-builder-client/api/v1"
 	eth2ApiV1Bellatrix "github.com/attestantio/go-eth2-client/api/v1/bellatrix"
 	eth2ApiV1Capella "github.com/attestantio/go-eth2-client/api/v1/capella"
 	eth2ApiV1Deneb "github.com/attestantio/go-eth2-client/api/v1/deneb"
@@ -209,22 +208,6 @@ func (m *BoostService) startBidCacheCleanupTask() {
 	}
 }
 
-func (m *BoostService) sendValidatorRegistrationsToRelayMonitors(payload []builderApiV1.SignedValidatorRegistration) {
-	log := m.log.WithField("method", "sendValidatorRegistrationsToRelayMonitors").WithField("numRegistrations", len(payload))
-	for _, relayMonitor := range m.relayMonitors {
-		go func(relayMonitor *url.URL) {
-			url := types.GetURI(relayMonitor, params.PathRegisterValidator)
-			log = log.WithField("url", url)
-			_, err := SendHTTPRequest(context.Background(), m.httpClientRegVal, http.MethodPost, url, "", nil, payload, nil)
-			if err != nil {
-				log.WithError(err).Warn("error calling registerValidator on relay monitor")
-				return
-			}
-			log.Debug("sent validator registrations to relay monitor")
-		}(relayMonitor)
-	}
-}
-
 func (m *BoostService) handleRoot(w http.ResponseWriter, _ *http.Request) {
 	m.respondOK(w, nilResponse)
 }
@@ -240,54 +223,42 @@ func (m *BoostService) handleStatus(w http.ResponseWriter, _ *http.Request) {
 	}
 }
 
-// handleRegisterValidator returns StatusOK if at least one relay returns StatusOK, else StatusBadGateway
+// handleRegisterValidator returns StatusOK if at least one relay returns StatusOK, else StatusBadGateway.
+// This forwards the message from the node to relays with minimal overhead. The registrations will maintain their
+// original encoding (SSZ or JSON) from the node.
 func (m *BoostService) handleRegisterValidator(w http.ResponseWriter, req *http.Request) {
 	log := m.log.WithField("method", "registerValidator")
-	log.Debug("registerValidator")
-
-	payload := []builderApiV1.SignedValidatorRegistration{}
-	if err := DecodeJSON(req.Body, &payload); err != nil {
-		m.respondError(w, http.StatusBadRequest, err.Error())
-		return
-	}
+	log.Debug("handling request")
 
+	// Get the user agent
 	ua := UserAgent(req.Header.Get("User-Agent"))
-	log = log.WithFields(logrus.Fields{
-		"numRegistrations": len(payload),
-		"ua":               ua,
-	})
+	log = log.WithFields(logrus.Fields{"ua": ua})
 
-	// Add request headers
-	headers := map[string]string{
-		HeaderStartTimeUnixMS: fmt.Sprintf("%d", time.Now().UTC().UnixMilli()),
-	}
+	// Additional header fields
+	header := req.Header
+	header.Set("User-Agent", wrapUserAgent(ua))
+	header.Set(HeaderStartTimeUnixMS, fmt.Sprintf("%d", time.Now().UTC().UnixMilli()))
 
-	relayRespCh := make(chan error, len(m.relays))
-
-	for _, relay := range m.relays {
-		go func(relay types.RelayEntry) {
-			url := relay.GetURI(params.PathRegisterValidator)
-			log := log.WithField("url", url)
-
-			_, err := SendHTTPRequest(context.Background(), m.httpClientRegVal, http.MethodPost, url, ua, headers, payload, nil)
-			if err != nil {
-				log.WithError(err).Warn("error calling registerValidator on relay")
-			}
-			relayRespCh <- err
-		}(relay)
+	// Read the validator registrations
+	regBytes, err := io.ReadAll(req.Body)
+	if err != nil {
+		m.respondError(w, http.StatusInternalServerError, err.Error())
+		return
 	}
+	req.Body.Close()
 
-	go m.sendValidatorRegistrationsToRelayMonitors(payload)
+	// Send the registrations to relay monitors, if configured
+	go m.sendValidatorRegistrationsToRelayMonitors(log, regBytes, header)
 
-	for i := 0; i < len(m.relays); i++ {
-		respErr := <-relayRespCh
-		if respErr == nil {
-			m.respondOK(w, nilResponse)
-			return
-		}
+	// Send the registrations to each relay
+	err = m.registerValidator(log, regBytes, header)
+	if err == nil {
+		// One of the relays responded OK
+		m.respondOK(w, nilResponse)
+	} else {
+		// None of the relays responded OK
+		m.respondError(w, http.StatusBadGateway, err.Error())
 	}
-
-	m.respondError(w, http.StatusBadGateway, errNoSuccessfulRelayResponse.Error())
 }
 
 // handleGetHeader requests bids from the relays

server/utils.go

@@ -268,3 +268,7 @@ func getPayloadResponseIsEmpty(payload *builderApi.VersionedSubmitBlindedBlockRe
 	}
 	return false
 }
+
+func wrapUserAgent(ua UserAgent) string {
+	return strings.TrimSpace(fmt.Sprintf("mev-boost/%s %s", config.Version, ua))
+}