diff --git a/proxy.go b/proxy.go index a7e7d83..75c528d 100644 --- a/proxy.go +++ b/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} } diff --git a/proxy_test.go b/proxy_test.go index 70216af..27c416d 100644 --- a/proxy_test.go +++ b/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) diff --git a/utils.go b/utils.go index 2ce138e..ff81c88 100644 --- a/utils.go +++ b/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) {