HTTP exporter driver leaks one connection, 3 goroutines and ~27 KB heap per push

[imtiazmangerah]

Component: internal/replication/drivers/http/driver.go, Driver.Accept
Versions: main (99c372e3f) and v2.4.12; the file is identical on release/v2.4.

Symptom

A long-running ledger worker with an HTTP exporter grows open sockets, goroutines and heap
linearly with the number of pushes until it is restarted.

Cause

Accept calls c.httpClient.Do(req) and never reads or closes rsp.Body. The keep-alive
connection is never returned to the transport, so every push dials a new connection and the
previous one stays alive with its read/write goroutines and buffers.

Reproduction

Needs only Go (the go.mod toolchain, go1.25) and network access for the clone. No Postgres,
no running ledger. Every command below was run as written against a fresh clone.

git clone --depth 1 https://github.com/formancehq/ledger.git && cd ledger

# 1. static: the analyzer flags the call site
go run github.com/timakin/bodyclose@latest ./internal/replication/drivers/http/
#   internal/replication/drivers/http/driver.go:45:29: response body must be closed

# 2. runtime: 200 pushes against an httptest server
cat > internal/replication/drivers/http/leak_probe_test.go <<'PROBE'
package http

import (
	"context"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/http/httptest"
	"runtime"
	"sync/atomic"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	logging "github.com/formancehq/go-libs/v5/pkg/observe/log"

	ledger "github.com/formancehq/ledger/internal"
	"github.com/formancehq/ledger/internal/replication/drivers"
)

func TestLeakProbe(t *testing.T) {
	var conns atomic.Int32
	srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = io.Copy(io.Discard, r.Body)
		w.WriteHeader(http.StatusAccepted)
		_, _ = w.Write([]byte(`{"accepted":true}`))
	}))
	srv.Config.ConnState = func(_ net.Conn, s http.ConnState) {
		if s == http.StateNew {
			conns.Add(1)
		}
	}
	srv.Start()
	t.Cleanup(srv.Close)

	d, err := NewDriver(Config{URL: srv.URL}, logging.Testing())
	require.NoError(t, err)
	log := drivers.NewLogWithLedger("m", ledger.NewLog(ledger.CreatedTransaction{Transaction: ledger.NewTransaction()}))

	runtime.GC()
	before := runtime.NumGoroutine()
	var ms runtime.MemStats
	runtime.ReadMemStats(&ms)
	heapBefore := ms.HeapInuse

	const pushes = 200
	for i := 0; i < pushes; i++ {
		_, err := d.Accept(context.TODO(), log)
		require.NoError(t, err)
	}
	time.Sleep(200 * time.Millisecond)
	runtime.GC()
	runtime.ReadMemStats(&ms)
	fmt.Printf("pushes=%d server_connections=%d goroutines_before=%d goroutines_after=%d heap_inuse_delta_kb=%d\n",
		pushes, conns.Load(), before, runtime.NumGoroutine(), int64(ms.HeapInuse-heapBefore)/1024)
}
PROBE
go test ./internal/replication/drivers/http/ -run TestLeakProbe -count=1 -v | grep '^pushes='
#   pushes=200 server_connections=200 goroutines_before=3 goroutines_after=603 heap_inuse_delta_kb=5544

# 3. apply the fix (fix.diff, below) and rerun both
git apply fix.diff
go run github.com/timakin/bodyclose@latest ./internal/replication/drivers/http/
#   (no output)
go test ./internal/replication/drivers/http/ -run TestLeakProbe -count=1 -v | grep '^pushes='
#   pushes=200 server_connections=1 goroutines_before=3 goroutines_after=6 heap_inuse_delta_kb=384

200 pushes	main	with fix
server connections opened	200	1
goroutines, before → after	3 → 603	3 → 6
heap in use, delta	+5.5 MB	+0.4 MB

Connection and goroutine counts are exact; the heap delta varies by a few hundred KB between runs.

fix.diff

diff --git a/internal/replication/drivers/http/driver.go b/internal/replication/drivers/http/driver.go
index cf3185286..3717a346b 100644
--- a/internal/replication/drivers/http/driver.go
+++ b/internal/replication/drivers/http/driver.go
@@ -5,8 +5,10 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"net/http"
 	"net/url"
+	"time"
 
 	"github.com/pkg/errors"
 
@@ -16,9 +18,19 @@ import (
 	"github.com/formancehq/ledger/internal/replication/drivers"
 )
 
+const (
+	// maxResponseDrainBytes caps how much of an exporter's response body Accept reads
+	// before closing it; ingest endpoints answer with a few bytes at most.
+	maxResponseDrainBytes = 64 << 10
+	// responseDrainTimeout caps how long Accept waits for those bytes. A stalled body
+	// costs the connection, not the worker.
+	responseDrainTimeout = 5 * time.Second
+)
+
 type Driver struct {
-	config     Config
-	httpClient *http.Client
+	config       Config
+	httpClient   *http.Client
+	drainTimeout time.Duration
 }
 
 func (c *Driver) Stop(_ context.Context) error {
@@ -40,12 +52,25 @@ func (c *Driver) Accept(ctx context.Context, logs ...drivers.LogWithLedger) ([]e
 	if err != nil {
 		return nil, err
 	}
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
 	req = req.WithContext(ctx)
 
 	rsp, err := c.httpClient.Do(req)
 	if err != nil {
 		return nil, err
 	}
+	defer func() {
+		// Drain and close so the underlying connection is released back to the
+		// transport instead of leaking one socket per push. The drain is bounded in
+		// bytes and in time: cancelling the request context interrupts a stalled body
+		// read, and the transport then drops that connection, which is preferable to
+		// blocking the worker on a misbehaving exporter.
+		stop := time.AfterFunc(c.drainTimeout, cancel)
+		defer stop.Stop()
+		_, _ = io.Copy(io.Discard, io.LimitReader(rsp.Body, maxResponseDrainBytes))
+		_ = rsp.Body.Close()
+	}()
 
 	if rsp.StatusCode < 200 || rsp.StatusCode > 299 {
 		return nil, fmt.Errorf("invalid status code, expect something between 200 and 299, got %d", rsp.StatusCode)
@@ -56,8 +81,9 @@ func (c *Driver) Accept(ctx context.Context, logs ...drivers.LogWithLedger) ([]e
 
 func NewDriver(config Config, _ logging.Logger) (*Driver, error) {
 	return &Driver{
-		config:     config,
-		httpClient: http.DefaultClient,
+		config:       config,
+		httpClient:   http.DefaultClient,
+		drainTimeout: responseDrainTimeout,
 	}, nil
 }
 

Fix

Drain and close the body after Do, as in fix.diff. PR with this change and a regression test:
formancehq/ledger#1905. The test, TestHTTPDriverReusesConnections, requires 20 pushes to open one
connection; it fails on main (20) and passes with the fix.

Disclosure

Triaged with Claude Fable 5.1 (Anthropic, model id claude-fable-5-1, via Claude Code). The
finding comes from our production export worker needing periodic restarts. Every number above and
the fix were reproduced by running the listed commands, and the findings, numbers and fix were
reviewed and verified by the author.