From a37436ca08c988a5381935f6e0723cf89a049f5b Mon Sep 17 00:00:00 2001 From: MoeMahhouk Date: Wed, 19 Aug 2026 14:22:32 +0000 Subject: [PATCH] feat: add -mirror-mode for async forwarding with empty ACK When sync-proxy sits behind an nginx mirror the caller discards the response body, so waiting on the backend EL only ties up the caller. With -mirror-mode (env MIRROR_MODE) the request is parsed and filtered as before, forwarded to builders in a detached bounded goroutine, and acknowledged immediately with an empty 200. Default behavior unchanged. --- main.go | 7 +++++++ mock_builder.go | 9 +++++++++ proxy.go | 24 ++++++++++++++++++++++++ proxy_test.go | 47 +++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 87 insertions(+) diff --git a/main.go b/main.go index 0499602..c69e2b8 100644 --- a/main.go +++ b/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, } diff --git a/mock_builder.go b/mock_builder.go index 6e9c975..8f8198c 100644 --- a/mock_builder.go +++ b/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 +} diff --git a/proxy.go b/proxy.go index 75c528d..fe411d1 100644 --- a/proxy.go +++ b/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 diff --git a/proxy_test.go b/proxy_test.go index 27c416d..d5e1b7f 100644 --- a/proxy_test.go +++ b/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)