diff --git a/client/fetcher.go b/client/fetcher.go index 774c62ef8..3034027df 100644 --- a/client/fetcher.go +++ b/client/fetcher.go @@ -77,6 +77,14 @@ func (h HTTPFetcher) fetch(ctx context.Context, p string) ([]byte, error) { if err != nil { return nil, fmt.Errorf("get(%q): %v", u.String(), err) } + // Close the body on every path, having read it to the end, so that the + // connection can be reused. + defer func() { + _, _ = io.Copy(io.Discard, r.Body) + if err := r.Body.Close(); err != nil { + slog.ErrorContext(ctx, "resp.Body.Close", slog.Any("error", err)) + } + }() switch r.StatusCode { case http.StatusOK: // All good, continue below @@ -86,12 +94,6 @@ func (h HTTPFetcher) fetch(ctx context.Context, p string) ([]byte, error) { default: return nil, fmt.Errorf("get(%q): %v", u.String(), r.StatusCode) } - - defer func() { - if err := r.Body.Close(); err != nil { - slog.ErrorContext(ctx, "resp.Body.Close", slog.Any("error", err)) - } - }() return io.ReadAll(r.Body) } diff --git a/client/fetcher_test.go b/client/fetcher_test.go index b9ffb8741..abd51b476 100644 --- a/client/fetcher_test.go +++ b/client/fetcher_test.go @@ -3,6 +3,12 @@ package client import ( "context" "errors" + "fmt" + "net" + "net/http" + "net/http/httptest" + "net/url" + "sync/atomic" "testing" ) @@ -31,3 +37,40 @@ func TestFileFetcherContextCancellation(t *testing.T) { t.Errorf("ReadEntryBundle: got error %v, want %v", err, context.Canceled) } } + +// TestHTTPFetcherReleasesConnections checks that HTTPFetcher releases the +// connection after every response, so the next request can reuse it. A +// response body that is never closed keeps its connection, and the transport's +// goroutines, for as long as the request's context lives. +func TestHTTPFetcherReleasesConnections(t *testing.T) { + for _, status := range []int{http.StatusOK, http.StatusNotFound, http.StatusInternalServerError} { + t.Run(http.StatusText(status), func(t *testing.T) { + var conns atomic.Int64 + srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(status) + _, _ = fmt.Fprint(w, "body") + })) + srv.Config.ConnState = func(_ net.Conn, s http.ConnState) { + if s == http.StateNew { + conns.Add(1) + } + } + srv.Start() + defer srv.Close() + u, err := url.Parse(srv.URL + "/") + if err != nil { + t.Fatal(err) + } + f, err := NewHTTPFetcher(u, srv.Client()) + if err != nil { + t.Fatal(err) + } + for range 10 { + _, _ = f.ReadCheckpoint(t.Context()) + } + if n := conns.Load(); n != 1 { + t.Errorf("10 sequential requests answered %d used %d connections, want 1", status, n) + } + }) + } +}