Send to builder without share queue

Author: dvush

cmd/receiver-proxy/main.go

@@ -212,16 +212,19 @@ func runMain(cCtx *cli.Context) error {
 	maxUserRPS := cCtx.Int(flagMaxUserRPS)
 
 	proxyConfig := &proxy.ReceiverProxyConfig{
-		ReceiverProxyConstantConfig: proxy.ReceiverProxyConstantConfig{Log: log, FlashbotsSignerAddress: flashbotsSignerAddress},
-		BuilderConfigHubEndpoint:    builderConfigHubEndpoint,
-		ArchiveEndpoint:             archiveEndpoint,
-		ArchiveConnections:          connectionsPerPeer,
-		LocalBuilderEndpoint:        builderEndpoint,
-		EthRPC:                      rpcEndpoint,
-		MaxRequestBodySizeBytes:     maxRequestBodySizeBytes,
-		ConnectionsPerPeer:          connectionsPerPeer,
-		MaxUserRPS:                  maxUserRPS,
-		ArchiveWorkerCount:          archiveWorkerCount,
+		ReceiverProxyConstantConfig: proxy.ReceiverProxyConstantConfig{
+			Log:                    log,
+			FlashbotsSignerAddress: flashbotsSignerAddress,
+			LocalBuilderEndpoint:   builderEndpoint,
+		},
+		BuilderConfigHubEndpoint: builderConfigHubEndpoint,
+		ArchiveEndpoint:          archiveEndpoint,
+		ArchiveConnections:       connectionsPerPeer,
+		EthRPC:                   rpcEndpoint,
+		MaxRequestBodySizeBytes:  maxRequestBodySizeBytes,
+		ConnectionsPerPeer:       connectionsPerPeer,
+		MaxUserRPS:               maxUserRPS,
+		ArchiveWorkerCount:       archiveWorkerCount,
 	}
 
 	instance, err := proxy.NewReceiverProxy(*proxyConfig)

proxy/receiver_api.go

@@ -80,7 +80,7 @@ func (prx *ReceiverProxy) UserJSONRPCHandler(maxRequestBodySizeBytes int64) (*rp
 
 // readyHandler calls /readyz on rbuilder
 func (prx *ReceiverProxy) readyHandler(w http.ResponseWriter, r *http.Request) error {
-	resp, err := http.Get(prx.localBuilderEndpoint + "/readyz")
+	resp, err := http.Get(prx.LocalBuilderEndpoint + "/readyz")
 	if err != nil {
 		prx.Log.Warn("Failed to check builder readiness", slog.Any("error", err))
 		http.Error(w, "not ready", http.StatusServiceUnavailable)
@@ -419,10 +419,13 @@ func (prx *ReceiverProxy) HandleParsedRequest(ctx context.Context, parsedRequest
 	incRequestDurationStep(time.Since(startAt), parsedRequest.method, "", "serialize_parsed_request")
 	startAt = time.Now()
 
-	select {
-	case <-ctx.Done():
-		prx.Log.Error("Shared queue is stalling")
-	case prx.shareQueue <- &parsedRequest:
+	// since we send to local builder while handling the request we can skip sharing request
+	if !parsedRequest.systemEndpoint {
+		select {
+		case <-ctx.Done():
+			prx.Log.Error("Shared queue is stalling")
+		case prx.shareQueue <- &parsedRequest:
+		}
 	}
 
 	incRequestDurationStep(time.Since(startAt), parsedRequest.method, "", "share_queue")
@@ -437,5 +440,14 @@ func (prx *ReceiverProxy) HandleParsedRequest(ctx context.Context, parsedRequest
 	}
 
 	incRequestDurationStep(time.Since(startAt), parsedRequest.method, "", "archive_queue")
+	startAt = time.Now()
+	// since we always send to local builder we do it here to avoid queue entirery
+
+	err = prx.localBuilderSender.SendRequest(&parsedRequest)
+	if err != nil {
+		prx.Log.Debug("Failed to send request to a local builder", slog.Any("error", err))
+	}
+
+	incRequestDurationStep(time.Since(startAt), parsedRequest.method, "", "local_builder")
 	return nil
 }

proxy/receiver_proxy.go

@@ -39,9 +39,6 @@ type ReceiverProxy struct {
 
 	OrderflowSigner *signature.Signer
 
-	localBuilder         rpcclient.RPCClient
-	localBuilderEndpoint string
-
 	UserHandler   http.Handler
 	SystemHandler http.Handler
 
@@ -61,13 +58,16 @@ type ReceiverProxy struct {
 	peerUpdaterClose chan struct{}
 
 	userAPIRateLimiter *rate.Limiter
+
+	localBuilderSender LocalBuilderSender
 }
 
 type ReceiverProxyConstantConfig struct {
 	Log *slog.Logger
 	// Name is optional field and it used to distringuish multiple proxies when running in the same process in tests
 	Name                   string
 	FlashbotsSignerAddress common.Address
+	LocalBuilderEndpoint   string
 }
 
 type ReceiverProxyConfig struct {
@@ -76,7 +76,6 @@ type ReceiverProxyConfig struct {
 	BuilderConfigHubEndpoint string
 	ArchiveEndpoint          string
 	ArchiveConnections       int
-	LocalBuilderEndpoint     string
 
 	// EthRPC should support eth_blockNumber API
 	EthRPC string
@@ -94,26 +93,23 @@ func NewReceiverProxy(config ReceiverProxyConfig) (*ReceiverProxy, error) {
 		return nil, err
 	}
 
-	localCl := HTTPClientLocalhost(DefaultLocalhostMaxIdleConn)
-
-	localBuilder := rpcclient.NewClientWithOpts(config.LocalBuilderEndpoint, &rpcclient.RPCClientOpts{
-		HTTPClient: localCl,
-	})
-
 	limit := rate.Limit(config.MaxUserRPS)
 	if config.MaxUserRPS == 0 {
 		limit = rate.Inf
 	}
 	userAPIRateLimiter := rate.NewLimiter(limit, config.MaxUserRPS)
+	localBuilderSender, err := NewLocalBuilderSender(config.Log, config.LocalBuilderEndpoint, config.ConnectionsPerPeer)
+	if err != nil {
+		return nil, err
+	}
 	prx := &ReceiverProxy{
 		ReceiverProxyConstantConfig: config.ReceiverProxyConstantConfig,
 		ConfigHub:                   NewBuilderConfigHub(config.Log, config.BuilderConfigHubEndpoint),
 		OrderflowSigner:             orderflowSigner,
-		localBuilder:                localBuilder,
-		localBuilderEndpoint:        config.LocalBuilderEndpoint,
 		requestUniqueKeysRLU:        expirable.NewLRU[uuid.UUID, struct{}](requestsRLUSize, nil, requestsRLUTTL),
 		replacementNonceRLU:         expirable.NewLRU[replacementNonceKey, int](replacementNonceSize, nil, replacementNonceTTL),
 		userAPIRateLimiter:          userAPIRateLimiter,
+		localBuilderSender:          localBuilderSender,
 	}
 	maxRequestBodySizeBytes := DefaultMaxRequestBodySizeBytes
 	if config.MaxRequestBodySizeBytes != 0 {
@@ -138,13 +134,12 @@ func NewReceiverProxy(config ReceiverProxyConfig) (*ReceiverProxy, error) {
 	prx.updatePeers = updatePeersCh
 
 	queue := ShareQueue{
-		name:                 prx.Name,
-		log:                  prx.Log,
-		queue:                shareQeueuCh,
-		updatePeers:          updatePeersCh,
-		localBuilderEndpoint: config.LocalBuilderEndpoint,
-		signer:               prx.OrderflowSigner,
-		workersPerPeer:       config.ConnectionsPerPeer,
+		name:           prx.Name,
+		log:            prx.Log,
+		queue:          shareQeueuCh,
+		updatePeers:    updatePeersCh,
+		signer:         prx.OrderflowSigner,
+		workersPerPeer: config.ConnectionsPerPeer,
 	}
 	go queue.Run()
 

proxy/receiver_proxy_test.go

@@ -252,11 +252,11 @@ func createProxy(localBuilder, name, certPath, certKeyPath string) *ReceiverProx
 			Log:                    log,
 			Name:                   name,
 			FlashbotsSignerAddress: flashbotsSigner.Address(),
+			LocalBuilderEndpoint:   localBuilder,
 		},
 
 		BuilderConfigHubEndpoint: builderHub.URL,
 		ArchiveEndpoint:          archiveServer.URL,
-		LocalBuilderEndpoint:     localBuilder,
 		EthRPC:                   "eth-rpc-not-set",
 		MaxUserRPS:               10,
 	})

proxy/sharing.go

@@ -5,7 +5,6 @@ import (
 	"errors"
 	"log/slog"
 	"net/http"
-	"os"
 	"time"
 
 	"github.com/flashbots/go-utils/jsonrpc"
@@ -19,20 +18,19 @@ var (
 	ShareWorkerQueueSize = 10000
 	requestTimeout       = time.Second * 10
 
-	errUnknownRequestType = errors.New("unknwon request type for sharing")
+	errUnknownRequestType = errors.New("unknown request type for sharing")
 )
 
 const (
 	bigRequestSize = 50_000
 )
 
 type ShareQueue struct {
-	name                 string
-	log                  *slog.Logger
-	queue                chan *ParsedRequest
-	updatePeers          chan []ConfighubBuilder
-	localBuilderEndpoint string
-	signer               *signature.Signer
+	name        string
+	log         *slog.Logger
+	queue       chan *ParsedRequest
+	updatePeers chan []ConfighubBuilder
+	signer      *signature.Signer
 	// if > 0 share queue will spawn multiple senders per peer
 	workersPerPeer int
 }
@@ -73,23 +71,7 @@ func (sq *ShareQueue) Run() {
 	if sq.workersPerPeer > 0 {
 		workersPerPeer = sq.workersPerPeer
 	}
-	var (
-		localBuilder *shareQueuePeer
-		peers        []shareQueuePeer
-	)
-	if len(sq.localBuilderEndpoint) > 0 {
-		client, err := NewFastHTTPClient(nil, workersPerPeer)
-		if err != nil {
-			sq.log.Error("Failed to intitialize local builder enpoint")
-			os.Exit(1)
-		}
-		builderPeer := newShareQueuePeer("local-builder", client, ConfighubBuilder{}, sq.localBuilderEndpoint)
-		localBuilder = &builderPeer
-		for worker := range workersPerPeer {
-			go sq.proxyRequests(localBuilder, worker)
-		}
-		defer localBuilder.Close()
-	}
+	var peers []shareQueuePeer
 	for {
 		select {
 		case req, more := <-sq.queue:
@@ -98,9 +80,6 @@ func (sq *ShareQueue) Run() {
 				sq.log.Info("Share queue closing, queue channel closed")
 				return
 			}
-			if localBuilder != nil {
-				localBuilder.SendRequest(sq.log, req)
-			}
 			if !req.systemEndpoint {
 				for _, peer := range peers {
 					peer.SendRequest(sq.log, req)
@@ -166,6 +145,83 @@ func (sq *ShareQueue) Run() {
 	}
 }
 
+type LocalBuilderSender struct {
+	logger   *slog.Logger
+	client   *fasthttp.Client
+	endpoint string
+}
+
+func NewLocalBuilderSender(logger *slog.Logger, endpoint string, maxOpenConnections int) (LocalBuilderSender, error) {
+	logger = logger.With(slog.String("peer", "local-builder"))
+
+	client, err := NewFastHTTPClient(nil, maxOpenConnections)
+	if err != nil {
+		return LocalBuilderSender{}, err
+	}
+
+	return LocalBuilderSender{
+		logger, client, endpoint,
+	}, nil
+}
+
+func (s *LocalBuilderSender) SendRequest(req *ParsedRequest) error {
+	request := fasthttp.AcquireRequest()
+	request.SetRequestURI(s.endpoint)
+	request.Header.SetMethod(http.MethodPost)
+	request.Header.SetContentTypeBytes([]byte("application/json"))
+	defer fasthttp.ReleaseRequest(request)
+
+	return sendShareRequest(s.logger, req, request, s.client, "local-builder")
+}
+
+func sendShareRequest(logger *slog.Logger, req *ParsedRequest, request *fasthttp.Request, client *fasthttp.Client, peerName string) error {
+	if req.serializedJSONRPCRequest == nil {
+		logger.Debug("Skip sharing request that is not serialized properly")
+		return nil
+	}
+
+	timeInQueue := time.Since(req.receivedAt)
+
+	request.Header.Set(signature.HTTPHeader, req.signatureHeader)
+	request.SetBodyRaw(req.serializedJSONRPCRequest)
+
+	resp := fasthttp.AcquireResponse()
+	start := time.Now()
+	err := client.DoTimeout(request, resp, requestTimeout)
+	requestDuration := time.Since(start)
+	timeE2E := timeInQueue + requestDuration
+
+	// in background update metrics and handle response
+	go func() {
+		isBig := req.size >= bigRequestSize
+		timeShareQueuePeerQueueDuration(peerName, timeInQueue, req.method, req.systemEndpoint, isBig)
+		timeShareQueuePeerRPCDuration(peerName, requestDuration.Milliseconds(), isBig)
+		timeShareQueuePeerE2EDuration(peerName, timeE2E, req.method, req.systemEndpoint, isBig)
+
+		logSendErrorLevel := slog.LevelDebug
+		if peerName == "local-builder" {
+			logSendErrorLevel = slog.LevelWarn
+		}
+		if err != nil {
+			logger.Log(context.Background(), logSendErrorLevel, "Error while proxying request", slog.Any("error", err))
+			incShareQueuePeerRPCErrors(peerName)
+		} else {
+			var parsedResp jsonrpc.JSONRPCResponse
+			err = json.Unmarshal(resp.Body(), &parsedResp)
+			if err != nil {
+				logger.Log(context.Background(), logSendErrorLevel, "Error parsing response while proxying", slog.Any("error", err))
+				incShareQueuePeerRPCErrors(peerName)
+			} else if parsedResp.Error != nil {
+				logger.Log(context.Background(), logSendErrorLevel, "Error returned from target while proxying", slog.Any("error", parsedResp.Error))
+				incShareQueuePeerRPCErrors(peerName)
+			}
+		}
+		fasthttp.ReleaseResponse(resp)
+	}()
+
+	return nil
+}
+
 func (sq *ShareQueue) proxyRequests(peer *shareQueuePeer, worker int) {
 	proxiedRequestCount := 0
 	logger := sq.log.With(slog.String("peer", peer.name), slog.String("name", sq.name), slog.Int("worker", worker))
@@ -191,44 +247,10 @@ func (sq *ShareQueue) proxyRequests(peer *shareQueuePeer, worker int) {
 			continue
 		}
 
-		timeInQueue := time.Since(req.receivedAt)
-
-		request.Header.Set(signature.HTTPHeader, req.signatureHeader)
-		request.SetBodyRaw(req.serializedJSONRPCRequest)
-
-		resp := fasthttp.AcquireResponse()
-		start := time.Now()
-		err := peer.client.DoTimeout(request, resp, requestTimeout)
-		requestDuration := time.Since(start)
-		timeE2E := timeInQueue + requestDuration
-
-		// in background update metrics and handle response
-		go func() {
-			isBig := req.size >= bigRequestSize
-			timeShareQueuePeerQueueDuration(peer.name, timeInQueue, req.method, req.systemEndpoint, isBig)
-			timeShareQueuePeerRPCDuration(peer.name, requestDuration.Milliseconds(), isBig)
-			timeShareQueuePeerE2EDuration(peer.name, timeE2E, req.method, req.systemEndpoint, isBig)
-
-			logSendErrorLevel := slog.LevelDebug
-			if peer.name == "local-builder" {
-				logSendErrorLevel = slog.LevelWarn
-			}
-			if err != nil {
-				logger.Log(context.Background(), logSendErrorLevel, "Error while proxying request", slog.Any("error", err))
-				incShareQueuePeerRPCErrors(peer.name)
-			} else {
-				var parsedResp jsonrpc.JSONRPCResponse
-				err = json.Unmarshal(resp.Body(), &parsedResp)
-				if err != nil {
-					logger.Log(context.Background(), logSendErrorLevel, "Error parsing response while proxying", slog.Any("error", err))
-					incShareQueuePeerRPCErrors(peer.name)
-				} else if parsedResp.Error != nil {
-					logger.Log(context.Background(), logSendErrorLevel, "Error returned from target while proxying", slog.Any("error", parsedResp.Error))
-					incShareQueuePeerRPCErrors(peer.name)
-				}
-			}
-			fasthttp.ReleaseResponse(resp)
-		}()
+		err := sendShareRequest(logger, req, request, peer.client, peer.name)
+		if err != nil {
+			logger.Debug("Failed to proxy a request", slog.Any("error", err))
+		}
 
 		proxiedRequestCount += 1
 		logger.Debug("Message proxied")