feat: add -mirror-mode for async forwarding with empty ACK

main.go

@@ -19,6 +19,7 @@ var (
 	defaultLogJSON    = os.Getenv("LOG_JSON") != ""
 	defaultListenAddr = getEnv("PROXY_LISTEN_ADDR", "localhost:25590")
 	defaultTimeoutMs  = getEnvInt("BUILDER_TIMEOUT_MS", 2000) // timeout for all the requests to the builders
+	defaultMirrorMode = os.Getenv("MIRROR_MODE") != ""
 
 	// Flags
 	logJSON          = flag.Bool("json", defaultLogJSON, "log in JSON format instead of text")
@@ -28,6 +29,7 @@ var (
 	builderTimeoutMs = flag.Int("request-timeout", defaultTimeoutMs, "timeout for requests to a builder [ms]")
 	proxyURLs        = flag.String("proxies", "", "proxy urls - other proxies to forward BN requests to (scheme://host)")
 	proxyTimeoutMs   = flag.Int("proxy-request-timeout", defaultTimeoutMs, "timeout for redundant beacon node requests to another proxy [ms]")
+	mirrorMode       = flag.Bool("mirror-mode", defaultMirrorMode, "acknowledge requests immediately with an empty 200 and forward to builders asynchronously (for use behind an nginx mirror)")
 )
 
 var log = logrus.WithField("module", "sync-proxy")
@@ -69,12 +71,17 @@ func main() {
 	proxyTimeout := time.Duration(*proxyTimeoutMs) * time.Millisecond
 
 	// Create a new proxy service.
+	if *mirrorMode {
+		log.Info("mirror mode enabled: acknowledging requests with an empty 200 and forwarding asynchronously")
+	}
+
 	opts := ProxyServiceOpts{
 		ListenAddr:     *listenAddr,
 		Builders:       builders,
 		BuilderTimeout: builderTimeout,
 		Proxies:        proxies,
 		ProxyTimeout:   proxyTimeout,
+		MirrorMode:     *mirrorMode,
 		Log:            log,
 	}
 

mock_builder.go

@@ -26,6 +26,7 @@ type mockServer struct {
 	// Used to count each engine made to the service, either if it fails or not, for each method
 	mu           sync.Mutex
 	requestCount map[string]int
+	lastHeaders  http.Header
 
 	// Responses placeholders that can be overridden
 	Response []byte
@@ -78,6 +79,7 @@ func (m *mockServer) newTestMiddleware(next http.Handler) http.Handler {
 			err = json.Unmarshal(bodyBytes, &req)
 			require.NoError(m.t, err)
 			m.requestCount[req.Method]++
+			m.lastHeaders = r.Header.Clone()
 
 			r.Body = io.NopCloser(bytes.NewBuffer(bodyBytes))
 
@@ -99,3 +101,10 @@ func (m *mockServer) GetRequestCount(method string) int {
 	defer m.mu.Unlock()
 	return m.requestCount[method]
 }
+
+// GetLastHeaders returns the headers of the most recent request
+func (m *mockServer) GetLastHeaders() http.Header {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.lastHeaders
+}

proxy.go

@@ -57,6 +57,7 @@ type ProxyServiceOpts struct {
 	BuilderTimeout time.Duration
 	Proxies        []*url.URL
 	ProxyTimeout   time.Duration
+	MirrorMode     bool
 	Log            *logrus.Entry
 }
 
@@ -67,6 +68,7 @@ type ProxyService struct {
 	builderEntries  []*ProxyEntry
 	proxyEntries    []*ProxyEntry
 	bestBeaconEntry *BeaconEntry
+	mirrorMode      bool
 
 	log *logrus.Entry
 	mu  sync.Mutex
@@ -94,6 +96,7 @@ func NewProxyService(opts ProxyServiceOpts) (*ProxyService, error) {
 		listenAddr:     opts.ListenAddr,
 		builderEntries: builderEntries,
 		proxyEntries:   proxyEntries,
+		mirrorMode:     opts.MirrorMode,
 		log:            opts.Log,
 	}, nil
 }
@@ -184,6 +187,13 @@ func (p *ProxyService) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 		return
 	}
 
+	if p.mirrorMode {
+		p.mirrorToBuilders(req, requestJSON, bodyBytes)
+		p.callProxies(req, bodyBytes)
+		w.WriteHeader(http.StatusOK)
+		return
+	}
+
 	builderResponse, err := p.callBuilders(req, requestJSON, bodyBytes)
 	p.callProxies(req, bodyBytes)
 
@@ -197,6 +207,20 @@ func (p *ProxyService) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	io.Copy(w, io.NopCloser(bytes.NewBuffer(builderResponse.Body)))
 }
 
+// mirrorToBuilders forwards the request to the builders without waiting for
+// the responses. The inbound request is cloned before the handler returns,
+// since *http.Request must not be used after ServeHTTP completes. Responses
+// are still awaited inside callBuilders (bounded by the builder timeout) for
+// logging and divergence detection; only the reply to the caller changes.
+func (p *ProxyService) mirrorToBuilders(req *http.Request, requestJSON JSONRPCRequest, bodyBytes []byte) {
+	detachedReq := req.Clone(context.Background())
+	go func() {
+		if _, err := p.callBuilders(detachedReq, requestJSON, bodyBytes); err != nil {
+			p.log.WithError(err).WithField("method", requestJSON.Method).Error("mirror-mode forward failed")
+		}
+	}()
+}
+
 func (p *ProxyService) callBuilders(req *http.Request, requestJSON JSONRPCRequest, bodyBytes []byte) (BuilderResponse, error) {
 	numSuccessRequestsToBuilder := 0
 	var mu sync.Mutex

proxy_test.go

@@ -289,6 +289,53 @@ func TestRequestTimeouts(t *testing.T) {
 	})
 }
 
+func TestMirrorMode(t *testing.T) {
+	t.Run("mirror mode should ack immediately with empty body and still forward", func(t *testing.T) {
+		backend := newTestBackend(t, 1, 0, time.Second, time.Second)
+		backend.proxyService.mirrorMode = true
+
+		backend.builders[0].ResponseDelay = 300 * time.Millisecond
+
+		req, err := http.NewRequest(http.MethodPost, "/", bytes.NewReader([]byte(mockNewPayloadRequest)))
+		require.NoError(t, err)
+		req.RemoteAddr = from
+		req.Header.Set("Authorization", "Bearer test-token")
+		rr := httptest.NewRecorder()
+
+		start := time.Now()
+		backend.proxyService.ServeHTTP(rr, req)
+		require.Less(t, time.Since(start), 200*time.Millisecond, "ack must not wait for the builder")
+		require.Equal(t, http.StatusOK, rr.Code)
+		require.Equal(t, "", rr.Body.String())
+
+		// the forward still happens (async), with the original Authorization header
+		require.Eventually(t, func() bool {
+			return backend.builders[0].GetRequestCount(newPayloadPath) == 1
+		}, 2*time.Second, 10*time.Millisecond)
+		require.Equal(t, "Bearer test-token", backend.builders[0].GetLastHeaders().Get("Authorization"))
+	})
+
+	t.Run("mirror mode off should still return builder response", func(t *testing.T) {
+		backend := newTestBackend(t, 1, 0, time.Second, time.Second)
+
+		rr := backend.request(t, []byte(mockNewPayloadRequest), from)
+		require.Equal(t, http.StatusOK, rr.Code)
+		require.Equal(t, mockNewPayloadResponseValid, rr.Body.String())
+	})
+
+	t.Run("mirror mode should still filter non-engine requests", func(t *testing.T) {
+		backend := newTestBackend(t, 1, 0, time.Second, time.Second)
+		backend.proxyService.mirrorMode = true
+
+		rr := backend.request(t, []byte(mockEthChainIDRequest), from)
+		require.Equal(t, http.StatusOK, rr.Code)
+		require.Equal(t, "", rr.Body.String())
+		// give any (incorrect) forward a chance to land, then assert none did
+		time.Sleep(100 * time.Millisecond)
+		require.Equal(t, 0, backend.builders[0].GetRequestCount("eth_chainId"))
+	})
+}
+
 func TestUpdateBestBeaconNode(t *testing.T) {
 	var data JSONRPCRequest
 	json.Unmarshal([]byte(mockForkchoiceRequestWithPayloadAttributesV1), &data)