fix(http): close connections of non-pooled clients after each request

Port of Danny-Dasilva/CycleTLS#426. With enable_connection_reuse=false,
getOrCreateClient builds a fresh client outside the pool and Do() drops
it without closing what it opened - dialTLS caches the live conn into
rt.cachedConnections and nothing ever frees it. On HTTP/2 (the common
ALPN outcome) the ClientConn read/write loops pin the socket open, so
every request leaked a socket + 2 goroutines until EMFILE. HTTP/1 was
already covered by DisableKeepAlives; h2 has no such knob.

- CycleTLS.Do(): close all connections on the roundTripper when the
  client came from outside the pool. Safe: nothing shares it and the
  body is fully read by then. Covers every FFI entry point.
- dispatcherAsync: non-pooled clients now close all conns instead of
  keeping the just-served hostPort conn cached on a dropped client.
- roundTripper.CloseIdleConnections: also drain idle conns inside
  cached transports (graceful h2 GOAWAY) and drop transport entries
  that have no cached conn (http-scheme path was never cleaned).

Regression test drives 8 reuse-off requests against a local h2 server
and asserts goroutine/fd counts stay flat; fails on the unfixed code
(+24 goroutines, +16 fds).

Author: smeinecke

golang/conn_leak_test.go

@@ -0,0 +1,172 @@
+package main
+
+import (
+	"net"
+	nhttp "net/http"
+	"net/http/httptest"
+	"os"
+	"runtime"
+	"testing"
+	"time"
+
+	http "github.com/Danny-Dasilva/fhttp"
+)
+
+// countingTransport records CloseIdleConnections calls so tests can verify
+// that roundTripper.CloseIdleConnections propagates into cached transports.
+type countingTransport struct {
+	calls int
+}
+
+func (c *countingTransport) RoundTrip(*http.Request) (*http.Response, error) {
+	return nil, nil
+}
+
+func (c *countingTransport) CloseIdleConnections() { c.calls++ }
+
+// TestCloseIdleConnectionsDrainsTransports asserts that CloseIdleConnections
+// also asks each cached transport to drop its idle connections - not just the
+// raw conns in cachedConnections. Without propagation an HTTP/2 ClientConn's
+// read/write loops would keep running on the dropped conn.
+func TestCloseIdleConnectionsDrainsTransports(t *testing.T) {
+	rt := newRoundTripper(Browser{}).(*roundTripper)
+
+	serverConn, clientConn := net.Pipe()
+	defer serverConn.Close()
+	rt.cachedConnections["example.test:443"] = clientConn
+
+	ct := &countingTransport{}
+	rt.cachedTransports["example.test:443"] = ct
+
+	rt.CloseIdleConnections()
+
+	if ct.calls == 0 {
+		t.Errorf("expected CloseIdleConnections to be called on the cached transport")
+	}
+	if len(rt.cachedConnections) != 0 || len(rt.cachedTransports) != 0 {
+		t.Errorf("expected both caches empty; got %d conns, %d transports",
+			len(rt.cachedConnections), len(rt.cachedTransports))
+	}
+	if _, err := clientConn.Write([]byte("x")); err == nil {
+		t.Errorf("expected cached conn to be closed, write succeeded")
+	}
+}
+
+// TestCloseIdleConnectionsKeepsSelectedAddr pins the keep-one-address path:
+// only non-matching conns/transports are drained.
+func TestCloseIdleConnectionsKeepsSelectedAddr(t *testing.T) {
+	rt := newRoundTripper(Browser{}).(*roundTripper)
+
+	s1, c1 := net.Pipe()
+	s2, c2 := net.Pipe()
+	defer s1.Close()
+	defer s2.Close()
+
+	rt.cachedConnections["keep.test:443"] = c1
+	rt.cachedConnections["drop.test:443"] = c2
+	keepTr := &countingTransport{}
+	dropTr := &countingTransport{}
+	rt.cachedTransports["keep.test:443"] = keepTr
+	rt.cachedTransports["drop.test:443"] = dropTr
+
+	rt.CloseIdleConnections("keep.test:443")
+
+	if keepTr.calls != 0 {
+		t.Errorf("kept transport should not be drained")
+	}
+	if dropTr.calls == 0 {
+		t.Errorf("dropped transport should be drained")
+	}
+	if rt.cachedConnections["keep.test:443"] == nil || rt.cachedTransports["keep.test:443"] == nil {
+		t.Errorf("kept addr lost its conn/transport")
+	}
+	if _, ok := rt.cachedConnections["drop.test:443"]; ok {
+		t.Errorf("dropped conn still cached")
+	}
+	if _, ok := rt.cachedTransports["drop.test:443"]; ok {
+		t.Errorf("dropped transport still cached")
+	}
+}
+
+// countOpenFDs returns the number of open file descriptors for this process.
+// Linux-only (/proc/self/fd); callers should skip elsewhere.
+func countOpenFDs(t *testing.T) int {
+	t.Helper()
+	entries, err := os.ReadDir("/proc/self/fd")
+	if err != nil {
+		t.Fatalf("cannot read /proc/self/fd: %v", err)
+	}
+	return len(entries)
+}
+
+// TestDoClosesNonPooledClientConnections is the end-to-end regression for
+// upstream CycleTLS PR #426: with enable_connection_reuse=false the client
+// is built outside the pool and dropped after a single request - nothing
+// used to close its connections. Against an HTTP/2 server every request
+// leaked a socket plus the h2 read/write loop goroutines (EMFILE over time).
+// HTTP/1.x was spared because DisableKeepAlives makes the inner transport
+// close the conn itself; h2 has no such knob, so Do() must close them.
+func TestDoClosesNonPooledClientConnections(t *testing.T) {
+	if runtime.GOOS != "linux" {
+		t.Skip("fd counting requires /proc/self/fd")
+	}
+
+	srv := httptest.NewUnstartedServer(nhttp.HandlerFunc(func(w nhttp.ResponseWriter, r *nhttp.Request) {
+		w.WriteHeader(200)
+		_, _ = w.Write([]byte("ok"))
+	}))
+	srv.EnableHTTP2 = true
+	srv.StartTLS()
+	defer srv.Close()
+
+	client := CycleTLS{}
+	opts := Options{
+		EnableConnectionReuse: false,
+		InsecureSkipVerify:    true,
+		Timeout:               10,
+		UserAgent:             "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
+	}
+
+	do := func() {
+		resp, err := client.Do(srv.URL, opts, "GET")
+		if err != nil {
+			t.Fatalf("request failed: %v", err)
+		}
+		// Do() reports request-level failures inside the Response rather
+		// than via err - require a real 200 so the test can't pass on a
+		// conn that never came up.
+		if resp.Status != 200 {
+			t.Fatalf("expected 200, got status %d body %q", resp.Status, resp.Body)
+		}
+	}
+
+	// Warmup: one request so steady-state goroutines (timers, GC helpers)
+	// are spun up before we take the baseline.
+	do()
+	runtime.GC()
+	time.Sleep(200 * time.Millisecond)
+	baseG := runtime.NumGoroutine()
+	baseFD := countOpenFDs(t)
+
+	const n = 8
+	for i := 0; i < n; i++ {
+		do()
+	}
+
+	// Give closed conns a moment to release fds and let h2 loops exit.
+	time.Sleep(500 * time.Millisecond)
+	runtime.GC()
+	time.Sleep(200 * time.Millisecond)
+
+	gotG := runtime.NumGoroutine() - baseG
+	gotFD := countOpenFDs(t) - baseFD
+
+	// Each leaked h2 conn costs ~2 goroutines and 1 fd. Allow slack for
+	// unrelated runtime churn but fail if the per-request leak is present.
+	if gotG >= n {
+		t.Errorf("goroutine count grew by %d after %d reuse-off requests - non-pooled client conns not closed", gotG, n)
+	}
+	if gotFD >= n/2 {
+		t.Errorf("fd count grew by %d after %d reuse-off requests - non-pooled client conns not closed", gotFD, n)
+	}
+}

golang/index.go

@@ -728,11 +728,18 @@ func dispatcherAsync(res fullRequest, chanWrite *safeChannelWriter) {
 	}
 
 	// Don't close connections when finished - they'll be reused for the same host
-	// Instead, tell the roundtripper to keep this connection but close others
+	// Instead, tell the roundtripper to keep this connection but close others.
+	// A client built outside the pool (connection reuse disabled) is not
+	// shared with any other request, so close all of its connections instead
+	// of keeping this one - otherwise each request strands a socket.
 	defer func() {
 		// Use type assertion to access the roundTripper
 		if transport, ok := res.client.Transport.(*roundTripper); ok {
-			transport.CloseIdleConnections(hostPort)
+			if res.options.Options.EnableConnectionReuse {
+				transport.CloseIdleConnections(hostPort)
+			} else {
+				transport.CloseIdleConnections()
+			}
 		}
 	}()
 
@@ -1772,6 +1779,18 @@ func (client CycleTLS) Do(URL string, options Options, Method string) (Response,
 		return Response{}, err
 	}
 
+	if !enableConnectionReuse {
+		// The client was built outside the pool, so no other request shares
+		// it and Do() reads the body to completion before returning. Close
+		// all of its connections on the way out - otherwise every request
+		// strands a socket (and, on HTTP/2, the conn's read/write loops).
+		defer func() {
+			if transport, ok := httpClient.Transport.(*roundTripper); ok {
+				transport.CloseIdleConnections()
+			}
+		}()
+	}
+
 	// Create request using fhttp
 	var bodyReader io.Reader
 	if len(options.BodyBytes) > 0 {

golang/roundtripper.go

@@ -658,32 +658,33 @@ func (rt *roundTripper) CloseIdleConnections(selectedAddr ...string) {
 	rt.Lock()
 	defer rt.Unlock()
 
-	// If we have a specific address to keep, only close other connections
-	if len(selectedAddr) > 0 && selectedAddr[0] != "" {
-		addr := selectedAddr[0]
-		// Keep the connection for the provided address, close others
-		for connAddr, conn := range rt.cachedConnections {
-			if connAddr != addr {
-				if conn != nil {
-					// Ignore errors from closing - connection might already be closed
-					_ = conn.Close()
-				}
-				delete(rt.cachedConnections, connAddr)
-				// Also clean up the transport cache for this address
-				delete(rt.cachedTransports, connAddr)
-			}
+	keep := ""
+	if len(selectedAddr) > 0 {
+		keep = selectedAddr[0]
+	}
+
+	// Let each cached transport drain its own idle connections first - on
+	// HTTP/2 this performs a graceful GOAWAY instead of just cutting the
+	// underlying socket.
+	for trAddr, tr := range rt.cachedTransports {
+		if trAddr == keep {
+			continue
 		}
-	} else {
-		// No address specified, close all connections (original behavior)
-		for addr, conn := range rt.cachedConnections {
-			if conn != nil {
-				// Ignore errors from closing - connection might already be closed
-				_ = conn.Close()
-			}
-			delete(rt.cachedConnections, addr)
-			// Also clean up the transport cache for this address
-			delete(rt.cachedTransports, addr)
+		if idleCloser, ok := tr.(interface{ CloseIdleConnections() }); ok {
+			idleCloser.CloseIdleConnections()
+		}
+		delete(rt.cachedTransports, trAddr)
+	}
+
+	for connAddr, conn := range rt.cachedConnections {
+		if connAddr == keep {
+			continue
+		}
+		if conn != nil {
+			// Ignore errors from closing - connection might already be closed
+			_ = conn.Close()
 		}
+		delete(rt.cachedConnections, connAddr)
 	}
 }