fix: share one HTTP client per provider and size its connection pool

docs/content/features/opentelemetry.mdoc

@@ -44,6 +44,17 @@ Number of events successfully delivered.
 | `type` | Destination type |
 | `status` | Delivery status (`success`, `failed`) |
 
+### `delivery_connections`
+
+Outbound HTTP connections used for delivery, split by whether the connection came from the idle pool or had to be opened.
+
+A high `reused=false` share means the idle connection pool is not holding connections long enough to be reused — expected at low traffic, but under sustained load it indicates the pool ceiling is binding.
+
+| Dimension | Description |
+|-----------|-------------|
+| `type` | Destination type (`webhook`, `hookdeck`) |
+| `reused` | Whether the connection was reused from the idle pool |
+
 ### `published_events`
 
 Number of events published via the Publish API.

internal/config/destinations.go

@@ -27,6 +27,7 @@ func (c *DestinationsConfig) ToConfig(cfg *Config) destregistrydefault.RegisterD
 		IncludeMillisecondTimestamp: c.IncludeMillisecondTimestamp,
 		Webhook:                     c.Webhook.toConfig(),
 		AWSKinesis:                  c.AWSKinesis.toConfig(),
+		DeliveryMaxConcurrency:      cfg.DeliveryMaxConcurrency,
 	}
 }
 

internal/config/logging.go

@@ -3,6 +3,7 @@ package config
 import (
 	"strings"
 
+	"github.com/hookdeck/outpost/internal/destregistry"
 	destregistrydefault "github.com/hookdeck/outpost/internal/destregistry/providers"
 	"github.com/hookdeck/outpost/internal/version"
 	"go.uber.org/zap"
@@ -43,6 +44,10 @@ func (c *Config) LogConfigurationSummary() []zap.Field {
 	// DISABLE_* flags.
 	webhookCfg := c.Destinations.Webhook.toConfig()
 
+	// Delivery connection pool. Not configurable — derived from the delivery
+	// worker pool size — so the resolved values are logged.
+	fanOutPool := destregistry.SizeFanOutPool(c.DeliveryMaxConcurrency)
+
 	fields := []zap.Field{
 		// General
 		zap.String("service", c.Service),
@@ -102,6 +107,8 @@ func (c *Config) LogConfigurationSummary() []zap.Field {
 		// Event Delivery
 		zap.Int("max_destinations_per_tenant", c.MaxDestinationsPerTenant),
 		zap.Int("delivery_timeout_seconds", c.DeliveryTimeoutSeconds),
+		zap.Int("delivery_max_idle_conns", fanOutPool.MaxIdleConns),
+		zap.Int("delivery_max_idle_conns_per_host", fanOutPool.MaxIdleConnsPerHost),
 
 		// Idempotency
 		zap.Int("publish_idempotency_key_ttl", c.PublishIdempotencyKeyTTL),

internal/destregistry/connpool.go

@@ -0,0 +1,97 @@
+package destregistry
+
+// Connection pool sizing for outbound delivery clients.
+//
+// Go's defaults are wrong for this workload in both directions: two idle
+// connections per host means reuse collapses above two concurrent deliveries
+// to a destination, and a 100 connection total means destinations evict each
+// other in any fan-out deployment. The pool is sized from the delivery worker
+// pool (DELIVERY_MAX_CONCURRENCY) as a sane default that scales with the
+// deployment; an explicit config knob can be exposed later if a workload
+// needs one.
+//
+// MaxIdleConns bounds only parked (idle) connections, never active ones —
+// exceeding it causes connection churn, not errors. That makes oversizing
+// cheap (idle FDs) and undersizing merely a reuse-rate loss.
+
+const (
+	// IdleConnsPerConcurrency scales the total idle pool with the delivery
+	// worker count: at ~3s per delivery a worker revisits roughly 32 distinct
+	// destinations within the 90s IdleConnTimeout, and slow destinations are
+	// where reuse matters most.
+	IdleConnsPerConcurrency = 32
+
+	// MinTotalIdleConns floors the total for low-concurrency fanout.
+	// Concurrency bounds simultaneous requests, not distinct hosts touched
+	// over time — a single worker at ~100ms per delivery still cycles through
+	// ~900 destinations per idle window.
+	MinTotalIdleConns = 512
+
+	// MaxTotalIdleConns caps the parked-FD/memory cost where the reuse hit
+	// rate decays. The cap never binds below the concurrency level itself —
+	// see SizeFanOutPool.
+	MaxTotalIdleConns = 4096
+
+	// MinIdleConnsPerHost is the floor for per-host depth, matching Go's
+	// default. DELIVERY_MAX_CONCURRENCY defaults to 1, which would otherwise
+	// size us below stock behavior.
+	MinIdleConnsPerHost = 2
+)
+
+// PoolSizing is the resolved connection pool configuration.
+type PoolSizing struct {
+	// MaxIdleConns bounds breadth — how many distinct destinations can hold a
+	// warm connection at all.
+	MaxIdleConns int
+
+	// MaxIdleConnsPerHost bounds depth — how many warm connections a single
+	// destination keeps.
+	MaxIdleConnsPerHost int
+}
+
+// SizeFanOutPool sizes a pool for a client that talks to arbitrarily many
+// destination hosts (the webhook providers). Depth comes from the delivery
+// worker pool — it caps how many deliveries can be in flight, so it caps how
+// many connections one destination could need. Breadth scales with the same
+// number: total = clamp(32×C, 512, max(4096, C)), the ceiling raised to C so
+// the cap never undersizes the pool below the concurrency level.
+//
+// deliveryMaxConcurrency <= 0 means "unknown"; the floors apply.
+func SizeFanOutPool(deliveryMaxConcurrency int) PoolSizing {
+	total := deliveryMaxConcurrency * IdleConnsPerConcurrency
+	if total < MinTotalIdleConns {
+		total = MinTotalIdleConns
+	}
+	if ceiling := max(MaxTotalIdleConns, deliveryMaxConcurrency); total > ceiling {
+		total = ceiling
+	}
+
+	perHost := deliveryMaxConcurrency
+	if perHost < MinIdleConnsPerHost {
+		perHost = MinIdleConnsPerHost
+	}
+	// Depth can't exceed the total pool — a per-host limit above the total
+	// would silently never be reachable.
+	if perHost > total {
+		perHost = total
+	}
+
+	return PoolSizing{
+		MaxIdleConns:        total,
+		MaxIdleConnsPerHost: perHost,
+	}
+}
+
+// SizeSingleHostPool sizes a pool for a client that talks to one host (the
+// hookdeck provider). It needs depth, not breadth, so the total is the
+// per-host value rather than a fan-out ceiling.
+func SizeSingleHostPool(deliveryMaxConcurrency int) PoolSizing {
+	perHost := deliveryMaxConcurrency
+	if perHost < MinIdleConnsPerHost {
+		perHost = MinIdleConnsPerHost
+	}
+	return PoolSizing{
+		MaxIdleConns:        perHost,
+		MaxIdleConnsPerHost: perHost,
+	}
+}

internal/destregistry/connpool_test.go

@@ -0,0 +1,70 @@
+package destregistry_test
+
+import (
+	"testing"
+
+	"github.com/hookdeck/outpost/internal/destregistry"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSizeFanOutPool_PerHostDerivesFromConcurrency(t *testing.T) {
+	t.Parallel()
+
+	// Per-host depth tracks the delivery worker pool, floored at Go's default
+	// so DELIVERY_MAX_CONCURRENCY=1 never sizes us below stock behavior.
+	assert.Equal(t, destregistry.MinIdleConnsPerHost, destregistry.SizeFanOutPool(0).MaxIdleConnsPerHost,
+		"unknown concurrency should use the floor")
+	assert.Equal(t, destregistry.MinIdleConnsPerHost, destregistry.SizeFanOutPool(1).MaxIdleConnsPerHost,
+		"concurrency below the floor should use the floor")
+	assert.Equal(t, 64, destregistry.SizeFanOutPool(64).MaxIdleConnsPerHost)
+}
+
+func TestSizeFanOutPool_TotalScalesWithConcurrency(t *testing.T) {
+	t.Parallel()
+
+	// total = clamp(32×C, 512, max(4096, C)).
+	for _, tc := range []struct {
+		concurrency int
+		total       int
+	}{
+		{0, destregistry.MinTotalIdleConns},   // unknown → floor
+		{1, destregistry.MinTotalIdleConns},   // 32 → floor
+		{16, destregistry.MinTotalIdleConns},  // 512 → exactly the floor
+		{64, 2048},                            // 32×64, between floor and cap
+		{128, destregistry.MaxTotalIdleConns}, // 4096 → exactly the cap
+		{1000, destregistry.MaxTotalIdleConns},
+	} {
+		assert.Equal(t, tc.total, destregistry.SizeFanOutPool(tc.concurrency).MaxIdleConns,
+			"concurrency %d", tc.concurrency)
+	}
+}
+
+func TestSizeFanOutPool_CapNeverBindsBelowConcurrency(t *testing.T) {
+	t.Parallel()
+
+	// Above the cap, the total rises to the concurrency level itself so the
+	// per-host depth (== concurrency) is always reachable.
+	pool := destregistry.SizeFanOutPool(10_000)
+	assert.Equal(t, 10_000, pool.MaxIdleConns)
+	assert.Equal(t, 10_000, pool.MaxIdleConnsPerHost)
+
+	for _, concurrency := range []int{0, 1, 16, 512, 100_000} {
+		pool := destregistry.SizeFanOutPool(concurrency)
+		assert.LessOrEqual(t, pool.MaxIdleConnsPerHost, pool.MaxIdleConns,
+			"a per-host limit above the total would never be reachable (concurrency %d)", concurrency)
+	}
+}
+
+func TestSizeSingleHostPool_IsDepthOnly(t *testing.T) {
+	t.Parallel()
+
+	// One host: the total is the per-host value. No breadth needed, so this
+	// stays small regardless of the fan-out formula.
+	pool := destregistry.SizeSingleHostPool(32)
+	assert.Equal(t, 32, pool.MaxIdleConnsPerHost)
+	assert.Equal(t, 32, pool.MaxIdleConns)
+
+	floored := destregistry.SizeSingleHostPool(1)
+	assert.Equal(t, destregistry.MinIdleConnsPerHost, floored.MaxIdleConnsPerHost)
+	assert.Equal(t, destregistry.MinIdleConnsPerHost, floored.MaxIdleConns)
+}

internal/destregistry/httpclient.go

@@ -3,6 +3,7 @@ package destregistry
 import (
 	"fmt"
 	"net/http"
+	"net/http/httptrace"
 	"net/url"
 	"time"
 )
@@ -18,6 +19,25 @@ type HTTPClientConfig struct {
 	// underlying transport plus the parsed proxy URL; returns the
 	// RoundTripper to use thereafter.
 	WrapTransport func(*http.Transport, *url.URL) http.RoundTripper
+
+	// Pool sizes the transport's idle connection pool. The zero value leaves
+	// Go's defaults in place (2 idle per host, 100 total), which is only
+	// appropriate for clients outside the delivery path. Use SizeFanOutPool
+	// or SizeSingleHostPool to derive it.
+	Pool PoolSizing
+
+	// OnConnection, if set, is invoked once per request with whether the
+	// underlying connection was reused. This is the signal that the pool
+	// ceiling is binding.
+	OnConnection func(reused bool)
+}
+
+func (c HTTPClientConfig) needsTransport() bool {
+	return c.ProxyURL != nil ||
+		c.UserAgent != nil ||
+		c.OnConnection != nil ||
+		c.Pool.MaxIdleConns > 0 ||
+		c.Pool.MaxIdleConnsPerHost > 0
 }
 
 // NewHTTPClient builds an *http.Client from config. Free function — no
@@ -29,11 +49,17 @@ func NewHTTPClient(config HTTPClientConfig) (*http.Client, error) {
 		client.Timeout = *config.Timeout
 	}
 
-	if config.ProxyURL == nil && config.UserAgent == nil {
+	if !config.needsTransport() {
 		return client, nil
 	}
 
 	transport := http.DefaultTransport.(*http.Transport).Clone()
+	if config.Pool.MaxIdleConns > 0 {
+		transport.MaxIdleConns = config.Pool.MaxIdleConns
+	}
+	if config.Pool.MaxIdleConnsPerHost > 0 {
+		transport.MaxIdleConnsPerHost = config.Pool.MaxIdleConnsPerHost
+	}
 
 	var rt http.RoundTripper = transport
 
@@ -55,10 +81,34 @@ func NewHTTPClient(config HTTPClientConfig) (*http.Client, error) {
 		}
 	}
 
+	if config.OnConnection != nil {
+		rt = &connTraceTransport{onConnection: config.OnConnection, transport: rt}
+	}
+
 	client.Transport = rt
 	return client, nil
 }
 
+// connTraceTransport reports whether each request got a pooled connection or
+// had to open a new one. Outermost wrapper so the trace is installed before
+// any other RoundTripper runs.
+type connTraceTransport struct {
+	onConnection func(reused bool)
+	transport    http.RoundTripper
+}
+
+func (t *connTraceTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	trace := &httptrace.ClientTrace{
+		GotConn: func(info httptrace.GotConnInfo) {
+			t.onConnection(info.Reused)
+		},
+	}
+	// httptrace composes with any trace already on the context rather than
+	// replacing it.
+	req = req.WithContext(httptrace.WithClientTrace(req.Context(), trace))
+	return t.transport.RoundTrip(req)
+}
+
 // userAgentTransport wraps an http.RoundTripper to inject a User-Agent header
 type userAgentTransport struct {
 	userAgent string