client: release the connection after any response

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)
 }
 

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)
+			}
+		})
+	}
+}