fix: bound forwards by the configured timeout and close proxy response body

proxy.go

@@ -39,8 +39,9 @@ type BuilderResponse struct {
 
 // ProxyEntry is an entry consisting of a URL and a proxy
 type ProxyEntry struct {
-	URL   *url.URL
-	Proxy *httputil.ReverseProxy
+	URL     *url.URL
+	Proxy   *httputil.ReverseProxy
+	Timeout time.Duration
 }
 
 // BeaconEntry consists of a URL from a beacon client and latest timestamp recorded
@@ -210,24 +211,23 @@ func (p *ProxyService) callBuilders(req *http.Request, requestJSON JSONRPCReques
 		go func(entry *ProxyEntry) {
 			defer wg.Done()
 			url := entry.URL
-			proxy := entry.Proxy
-			resp, err := SendProxyRequest(req, proxy, bodyBytes)
+			resp, cancel, err := SendProxyRequest(req, entry, bodyBytes)
 			if err != nil {
 				log.WithError(err).WithField("url", url.String()).Error("error sending request to builder")
 				return
 			}
+			defer cancel()
+			defer resp.Body.Close()
 
-			reader := resp.Body
-			responseBytes, err := io.ReadAll(reader)
+			responseBytes, err := io.ReadAll(resp.Body)
 			if err != nil {
 				p.log.WithError(err).Error("failed to read response body")
 				return
 			}
-			defer resp.Body.Close()
 
 			var uncompressedResponseBytes []byte
 			if !resp.Uncompressed && resp.Header.Get("Content-Encoding") == "gzip" {
-				reader, err = gzip.NewReader(io.NopCloser(bytes.NewBuffer(responseBytes)))
+				reader, err := gzip.NewReader(io.NopCloser(bytes.NewBuffer(responseBytes)))
 				if err != nil {
 					p.log.WithError(err).Error("failed to decompress response body")
 					return
@@ -282,11 +282,14 @@ func (p *ProxyService) callProxies(req *http.Request, bodyBytes []byte) {
 	// call other proxies to forward requests from other beacon nodes
 	for _, entry := range p.proxyEntries {
 		go func(entry *ProxyEntry) {
-			_, err := SendProxyRequest(req, entry.Proxy, bodyBytes)
+			resp, cancel, err := SendProxyRequest(req, entry, bodyBytes)
 			if err != nil {
 				log.WithError(err).WithField("url", entry.URL.String()).Error("error sending request to proxy")
 				return
 			}
+			defer cancel()
+			_, _ = io.Copy(io.Discard, resp.Body)
+			resp.Body.Close()
 		}(entry)
 	}
 }
@@ -422,6 +425,7 @@ func buildProxyEntry(proxyURL *url.URL, timeout time.Duration) ProxyEntry {
 		IdleConnTimeout:       90 * time.Second,
 		TLSHandshakeTimeout:   10 * time.Second,
 		ExpectContinueTimeout: 1 * time.Second,
+		ResponseHeaderTimeout: timeout,
 	}
-	return ProxyEntry{Proxy: proxy, URL: proxyURL}
+	return ProxyEntry{Proxy: proxy, URL: proxyURL, Timeout: timeout}
 }

proxy_test.go

@@ -263,6 +263,32 @@ func TestBuilders(t *testing.T) {
 	})
 }
 
+func TestRequestTimeouts(t *testing.T) {
+	t.Run("slow builder response should time out and fall back to other builder", func(t *testing.T) {
+		backend := newTestBackend(t, 2, 0, 100*time.Millisecond, 100*time.Millisecond)
+
+		backend.builders[0].ResponseDelay = 500 * time.Millisecond
+		backend.builders[0].Response = []byte(mockNewPayloadResponseValid)
+		backend.builders[1].Response = []byte(mockNewPayloadResponseSyncing)
+
+		start := time.Now()
+		rr := backend.request(t, []byte(mockNewPayloadRequest), from)
+		require.Less(t, time.Since(start), 400*time.Millisecond, "request should not wait for the slow builder")
+		require.Equal(t, http.StatusOK, rr.Code, rr.Body.String())
+		// fallback: response comes from the non-delayed secondary builder
+		require.Equal(t, mockNewPayloadResponseSyncing, rr.Body.String())
+	})
+
+	t.Run("all builders slow should return bad gateway", func(t *testing.T) {
+		backend := newTestBackend(t, 1, 0, 100*time.Millisecond, 100*time.Millisecond)
+
+		backend.builders[0].ResponseDelay = 500 * time.Millisecond
+
+		rr := backend.request(t, []byte(mockNewPayloadRequest), from)
+		require.Equal(t, http.StatusBadGateway, rr.Code)
+	})
+}
+
 func TestUpdateBestBeaconNode(t *testing.T) {
 	var data JSONRPCRequest
 	json.Unmarshal([]byte(mockForkchoiceRequestWithPayloadAttributesV1), &data)

utils.go

@@ -10,23 +10,36 @@ import (
 	"strings"
 )
 
-func BuildProxyRequest(req *http.Request, proxy *httputil.ReverseProxy, bodyBytes []byte) *http.Request {
-	// Copy and redirect request to EL endpoint
-	proxyReq := req.Clone(context.Background())
+func BuildProxyRequest(ctx context.Context, req *http.Request, proxy *httputil.ReverseProxy, bodyBytes []byte) *http.Request {
+	// Copy and redirect request to EL endpoint. The context is detached from the
+	// inbound request on purpose (client cancellation must not abort the forward)
+	// but bounded by the configured timeout.
+	proxyReq := req.Clone(ctx)
 	appendHostToXForwardHeader(proxyReq.Header, req.URL.Host)
 	proxyReq.Body = io.NopCloser(bytes.NewBuffer(bodyBytes))
 
 	proxy.Director(proxyReq)
 	return proxyReq
 }
-func SendProxyRequest(req *http.Request, proxy *httputil.ReverseProxy, bodyBytes []byte) (*http.Response, error) {
-	proxyReq := BuildProxyRequest(req, proxy, bodyBytes)
-	resp, err := proxy.Transport.RoundTrip(proxyReq)
+
+// SendProxyRequest forwards the request to the entry's backend, bounded by the
+// entry's timeout. On success the caller must close resp.Body and then call
+// cancel once done reading it.
+func SendProxyRequest(req *http.Request, entry *ProxyEntry, bodyBytes []byte) (*http.Response, context.CancelFunc, error) {
+	ctx := context.Background()
+	cancel := context.CancelFunc(func() {})
+	if entry.Timeout > 0 {
+		ctx, cancel = context.WithTimeout(ctx, entry.Timeout)
+	}
+
+	proxyReq := BuildProxyRequest(ctx, req, entry.Proxy, bodyBytes)
+	resp, err := entry.Proxy.Transport.RoundTrip(proxyReq)
 	if err != nil {
-		return nil, err
+		cancel()
+		return nil, nil, err
 	}
 
-	return resp, nil
+	return resp, cancel, nil
 }
 
 func copyHeader(dst, src http.Header) {