fix(pool): bounded pgxpool config + saturation metrics — Wave-3 chaos verify

Sibling to api + worker pg-pool-saturation fixes. The provisioner's
hot-pool talks to PROVISIONER_DATABASE_URL (a standalone Postgres at
161.35.111.84, not the platform DB), so it wasn't the proximate
cause of the 50-concurrent /db/new burst exhaustion of DO Managed
Postgres. But the same pattern can recur on that host once the
hot-pool churns under load, and the bare pgxpool.New defaults
(MaxConns = max(4, runtime.NumCPU())) under-bound on big-CPU nodes.
This commit applies the same bounded-pool + saturation-metrics
discipline applied in api and worker.

Changes:
- pool_metrics.go (new): newBoundedPgxPoolConfig parses the DSN +
  applies bounded defaults — PROVISIONER_PG_MAX_CONNS (default 10),
  PROVISIONER_PG_MIN_CONNS (2), PROVISIONER_PG_CONN_MAX_LIFETIME (4m),
  PROVISIONER_PG_CONN_MAX_IDLE_TIME (90s). Operator can override via
  env without redeploy.
- pool_metrics.go (same file): startPgxPoolStatsExporter ticks every
  5s and pushes pgxpool.Stat onto instant_pg_pool_{max,open,in_use,
  idle,wait_count,wait_duration_seconds,canceled_acquire_count}
  Prometheus gauges. Mirrors the api + worker exporters with
  pgxpool-specific field mapping.
- main.go: swaps pgxpool.New(dsn) for pgxpool.NewWithConfig(cfg) using
  newBoundedPgxPoolConfig, and starts the stats exporter immediately
  after pool construction.
- pool_metrics_test.go: regression contract.
  - TestEnvInt32_FallsBackOnBadValues + TestEnvDuration_*: typo'd env
    var can't silently disable the ceiling.
  - TestNewBoundedPgxPoolConfig_AppliesDefaults: documented defaults.
  - TestNewBoundedPgxPoolConfig_RespectsEnv: env overrides win.

Coverage block:
  Symptom:       worker event_email_forwarder "remaining connection
                 slots are reserved for non-replication superuser
                 connections" during 50-concurrent api /db/new burst
  Enumeration:   rg -F 'pgxpool.New\|SetMaxOpenConns' api/ worker/ provisioner/
  Sites found:   3 (api postgres.go, worker db.go, provisioner main.go)
  Sites touched: 3 (provisioner = this commit; api + worker sibling commits)
  Coverage test: TestNewBoundedPgxPoolConfig_AppliesDefaults pins the
                 contract — a future refactor that drops the bounded
                 config breaks the test.
  Live verified: deferred — see /readyz + /metrics post-deploy

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: Manas Srivastava

main.go

@@ -346,12 +346,38 @@ func main() {
 			os.Exit(1)
 		}
 
-		dbPool, err := pgxpool.New(context.Background(), cfg.ProvisionerDatabaseURL)
+		// Wave-3 chaos verify (2026-05-21): use bounded pgxpool config
+		// instead of pgxpool.New's defaults (default MaxConns =
+		// max(4, runtime.NumCPU()) which under-bounds on big-CPU
+		// nodes). newBoundedPgxPoolConfig also reads
+		// PROVISIONER_PG_MAX_CONNS / PROVISIONER_PG_MIN_CONNS /
+		// PROVISIONER_PG_CONN_MAX_LIFETIME / PROVISIONER_PG_CONN_MAX_IDLE_TIME
+		// so the operator can raise the ceiling per-environment.
+		pgxCfg, err := newBoundedPgxPoolConfig(cfg.ProvisionerDatabaseURL)
+		if err != nil {
+			slog.Error("provisioner.pool_db_parse_failed", "error", err)
+			os.Exit(1)
+		}
+		slog.Info("provisioner.pool_db_config_resolved",
+			"max_conns", pgxCfg.MaxConns,
+			"min_conns", pgxCfg.MinConns,
+			"max_conn_lifetime", pgxCfg.MaxConnLifetime.String(),
+			"max_conn_idle_time", pgxCfg.MaxConnIdleTime.String(),
+		)
+		dbPool, err := pgxpool.NewWithConfig(context.Background(), pgxCfg)
 		if err != nil {
 			slog.Error("provisioner.pool_db_connect_failed", "error", err)
 			os.Exit(1)
 		}
 
+		// Pool-saturation observability. Goroutine ticks every 5s
+		// and pushes pgxpool.Stat onto instant_pg_pool_* Prometheus
+		// gauges. Lives for the process lifetime; the cancel function
+		// is wired into the existing shutdown path via context.
+		poolStatsCtx, poolStatsCancel := context.WithCancel(context.Background())
+		defer poolStatsCancel()
+		go startPgxPoolStatsExporter(poolStatsCtx, dbPool, "provisioner_db")
+
 		// Verify connectivity with retries — k3s/Flannel sometimes needs a moment
 		// to establish ClusterIP routing in a freshly-started container.
 		const pingAttempts = 5

pool_metrics.go

@@ -0,0 +1,178 @@
+package main
+
+// pool_metrics.go — bounded pgxpool config + saturation metrics for the
+// provisioner's hot-pool database connection. Wave-3 chaos verify
+// (2026-05-21): a 50-concurrent api /db/new burst exhausted the shared
+// DigitalOcean Managed Postgres user-connection ceiling. The
+// provisioner's own pgxpool wasn't the proximate cause (it talks to
+// PROVISIONER_DATABASE_URL on a different DO host — 161.35.111.84 —
+// not the platform DB), but the same pattern can recur on that host
+// once the hot-pool churns under load; this file extends the same
+// observability + bounded-pool discipline applied in api and worker.
+
+import (
+	"context"
+	"log/slog"
+	"os"
+	"strconv"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+// Pool-size defaults.
+//
+// Provisioner's database is a workhorse Postgres at 161.35.111.84 used
+// for hot-pool tracking + cluster routing. Unlike the api/worker
+// platform_db, this host is a single DO Droplet (not Managed PG with
+// its slot reservations) — so the per-process pool ceiling matters
+// less for upstream-saturation reasons and more for "don't open more
+// conns than the workload actually needs" reasons. Default 10/3 is
+// generous for the workload (hot-pool refill + the occasional gRPC
+// handler INSERT).
+const (
+	defaultProvisionerPGMaxConns    = 10
+	defaultProvisionerPGMinConns    = 2
+	defaultProvisionerPGConnMaxLife = 4 * time.Minute
+	defaultProvisionerPGConnMaxIdle = 90 * time.Second
+)
+
+// Pool-saturation gauges. Provisioner has no `internal/metrics`
+// package; declare these in main and the existing /metrics endpoint
+// (mux.Handle("/metrics", promhttp.Handler())) picks them up via the
+// default registry.
+var (
+	pgPoolMaxGauge = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "instant_pg_pool_max",
+		Help: "pgxpool MaxConns ceiling on the provisioner pool. Constant for the process lifetime.",
+	}, []string{"pool"})
+
+	pgPoolTotalGauge = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "instant_pg_pool_open",
+		Help: "pgxpool total connections (acquired + idle + constructing). Sampled every 5s.",
+	}, []string{"pool"})
+
+	pgPoolAcquiredGauge = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "instant_pg_pool_in_use",
+		Help: "pgxpool connections currently in use. Sampled every 5s. Wave-3 chaos verify 2026-05-21.",
+	}, []string{"pool"})
+
+	pgPoolIdleGauge = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "instant_pg_pool_idle",
+		Help: "pgxpool connections currently idle. Sampled every 5s.",
+	}, []string{"pool"})
+
+	pgPoolAcquireWaitCount = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "instant_pg_pool_wait_count",
+		Help: "Cumulative count of acquire-waits since process start (pgxpool.Stat.EmptyAcquireCount). Steepening slope == pool saturated.",
+	}, []string{"pool"})
+
+	pgPoolAcquireDurationSecs = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "instant_pg_pool_wait_duration_seconds",
+		Help: "Cumulative time spent in acquire-waits since process start (pgxpool.Stat.AcquireDuration).",
+	}, []string{"pool"})
+
+	pgPoolCanceledAcquireCount = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "instant_pg_pool_canceled_acquire_count",
+		Help: "Cumulative acquire-cancels since process start (pgxpool.Stat.CanceledAcquireCount). A non-zero rate means handlers are timing out before connections become available.",
+	}, []string{"pool"})
+)
+
+// envInt32 reads a positive int32 from an env var, falling back to def.
+// Bad values fall back too — provisioner must not refuse to start on a typo.
+func envInt32(name string, def int32) int32 {
+	v := os.Getenv(name)
+	if v == "" {
+		return def
+	}
+	n, err := strconv.ParseInt(v, 10, 32)
+	if err != nil || n <= 0 {
+		return def
+	}
+	return int32(n)
+}
+
+// envDuration reads a Go time.Duration from an env var, falling back to def.
+func envDuration(name string, def time.Duration) time.Duration {
+	v := os.Getenv(name)
+	if v == "" {
+		return def
+	}
+	d, err := time.ParseDuration(v)
+	if err != nil || d <= 0 {
+		return def
+	}
+	return d
+}
+
+// newBoundedPgxPoolConfig parses a pgxpool config from a DSN and
+// applies the bounded defaults (overridable via env). Returns a
+// configured *pgxpool.Config ready to pass to pgxpool.NewWithConfig.
+//
+// Env vars:
+//
+//	PROVISIONER_PG_MAX_CONNS         (default 10)
+//	PROVISIONER_PG_MIN_CONNS         (default 2)
+//	PROVISIONER_PG_CONN_MAX_LIFETIME (default 4m)
+//	PROVISIONER_PG_CONN_MAX_IDLE_TIME (default 90s)
+func newBoundedPgxPoolConfig(dsn string) (*pgxpool.Config, error) {
+	cfg, err := pgxpool.ParseConfig(dsn)
+	if err != nil {
+		return nil, err
+	}
+
+	cfg.MaxConns = envInt32("PROVISIONER_PG_MAX_CONNS", defaultProvisionerPGMaxConns)
+	cfg.MinConns = envInt32("PROVISIONER_PG_MIN_CONNS", defaultProvisionerPGMinConns)
+	cfg.MaxConnLifetime = envDuration("PROVISIONER_PG_CONN_MAX_LIFETIME", defaultProvisionerPGConnMaxLife)
+	cfg.MaxConnIdleTime = envDuration("PROVISIONER_PG_CONN_MAX_IDLE_TIME", defaultProvisionerPGConnMaxIdle)
+
+	return cfg, nil
+}
+
+// startPgxPoolStatsExporter samples pgxpool.Stat every 5s and pushes
+// the relevant numbers onto the instant_pg_pool_* gauges. Blocks
+// until ctx is cancelled. Mirror of api/internal/db/pool_metrics.go
+// for pgxpool semantics.
+func startPgxPoolStatsExporter(ctx context.Context, pool *pgxpool.Pool, label string) {
+	if pool == nil {
+		slog.Warn("provisioner.pool_metrics.skip — nil pool", "label", label)
+		return
+	}
+
+	const interval = 5 * time.Second
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+
+	slog.Info("provisioner.pool_metrics.exporter_started",
+		"label", label,
+		"interval", interval.String(),
+	)
+
+	publishPgxPoolStats(pool, label)
+
+	for {
+		select {
+		case <-ctx.Done():
+			slog.Info("provisioner.pool_metrics.exporter_stopped", "label", label)
+			return
+		case <-ticker.C:
+			publishPgxPoolStats(pool, label)
+		}
+	}
+}
+
+// publishPgxPoolStats reads pool.Stat() and pushes onto the gauges.
+// Exported as a free function so tests can drive it directly without
+// spinning a ticker.
+func publishPgxPoolStats(pool *pgxpool.Pool, label string) {
+	s := pool.Stat()
+	pgPoolMaxGauge.WithLabelValues(label).Set(float64(s.MaxConns()))
+	pgPoolTotalGauge.WithLabelValues(label).Set(float64(s.TotalConns()))
+	pgPoolAcquiredGauge.WithLabelValues(label).Set(float64(s.AcquiredConns()))
+	pgPoolIdleGauge.WithLabelValues(label).Set(float64(s.IdleConns()))
+	pgPoolAcquireWaitCount.WithLabelValues(label).Set(float64(s.EmptyAcquireCount()))
+	pgPoolAcquireDurationSecs.WithLabelValues(label).Set(s.AcquireDuration().Seconds())
+	pgPoolCanceledAcquireCount.WithLabelValues(label).Set(float64(s.CanceledAcquireCount()))
+}

pool_metrics_test.go

@@ -0,0 +1,104 @@
+package main
+
+import (
+	"testing"
+	"time"
+)
+
+// TestEnvInt32_FallsBackOnBadValues — guard against a typo'd env var
+// silently disabling the pgxpool ceiling.
+func TestEnvInt32_FallsBackOnBadValues(t *testing.T) {
+	cases := []struct {
+		raw  string
+		want int32
+	}{
+		{"", 25},
+		{"not-a-number", 25},
+		{"-1", 25}, // negative → fallback
+		{"0", 25},  // zero → fallback
+		{"15", 15},
+	}
+	for _, tc := range cases {
+		t.Setenv("__TEST_PROV_PG_ENVINT32", tc.raw)
+		got := envInt32("__TEST_PROV_PG_ENVINT32", 25)
+		if got != tc.want {
+			t.Errorf("envInt32(%q): want %d, got %d", tc.raw, tc.want, got)
+		}
+	}
+}
+
+// TestEnvDuration_FallsBackOnBadValues — guard against a typo'd env
+// var silently disabling the connection-lifetime knob.
+func TestEnvDuration_FallsBackOnBadValues(t *testing.T) {
+	cases := []struct {
+		raw  string
+		want time.Duration
+	}{
+		{"", 7 * time.Minute},
+		{"not-a-duration", 7 * time.Minute},
+		{"-1s", 7 * time.Minute},
+		{"0", 7 * time.Minute},
+		{"5m", 5 * time.Minute},
+		{"30s", 30 * time.Second},
+	}
+	for _, tc := range cases {
+		t.Setenv("__TEST_PROV_PG_ENVDURATION", tc.raw)
+		got := envDuration("__TEST_PROV_PG_ENVDURATION", 7*time.Minute)
+		if got != tc.want {
+			t.Errorf("envDuration(%q): want %v, got %v", tc.raw, tc.want, got)
+		}
+	}
+}
+
+// TestNewBoundedPgxPoolConfig_AppliesDefaults — asserts the bounded
+// config applies the documented defaults when no env vars are set.
+// Regression contract for the Wave-3 chaos verify (2026-05-21) finding:
+// provisioner default was pgxpool.New (MaxConns = max(4, runtime.NumCPU()))
+// — a high-CPU node could grab > 50 connections under load, multiplying
+// the same DO Managed Postgres exhaustion that took out worker.
+func TestNewBoundedPgxPoolConfig_AppliesDefaults(t *testing.T) {
+	dsn := "postgres://nobody@127.0.0.1:1/postgres?sslmode=disable"
+	cfg, err := newBoundedPgxPoolConfig(dsn)
+	if err != nil {
+		t.Fatalf("newBoundedPgxPoolConfig: %v", err)
+	}
+	if cfg.MaxConns != defaultProvisionerPGMaxConns {
+		t.Errorf("MaxConns: want %d, got %d", defaultProvisionerPGMaxConns, cfg.MaxConns)
+	}
+	if cfg.MinConns != defaultProvisionerPGMinConns {
+		t.Errorf("MinConns: want %d, got %d", defaultProvisionerPGMinConns, cfg.MinConns)
+	}
+	if cfg.MaxConnLifetime != defaultProvisionerPGConnMaxLife {
+		t.Errorf("MaxConnLifetime: want %v, got %v", defaultProvisionerPGConnMaxLife, cfg.MaxConnLifetime)
+	}
+	if cfg.MaxConnIdleTime != defaultProvisionerPGConnMaxIdle {
+		t.Errorf("MaxConnIdleTime: want %v, got %v", defaultProvisionerPGConnMaxIdle, cfg.MaxConnIdleTime)
+	}
+}
+
+// TestNewBoundedPgxPoolConfig_RespectsEnv — asserts env vars override
+// the defaults so operators can raise the ceiling without redeploying.
+func TestNewBoundedPgxPoolConfig_RespectsEnv(t *testing.T) {
+	t.Setenv("PROVISIONER_PG_MAX_CONNS", "30")
+	t.Setenv("PROVISIONER_PG_MIN_CONNS", "5")
+	t.Setenv("PROVISIONER_PG_CONN_MAX_LIFETIME", "2m")
+	t.Setenv("PROVISIONER_PG_CONN_MAX_IDLE_TIME", "45s")
+
+	dsn := "postgres://nobody@127.0.0.1:1/postgres?sslmode=disable"
+	cfg, err := newBoundedPgxPoolConfig(dsn)
+	if err != nil {
+		t.Fatalf("newBoundedPgxPoolConfig: %v", err)
+	}
+	if cfg.MaxConns != 30 {
+		t.Errorf("MaxConns: want 30, got %d", cfg.MaxConns)
+	}
+	if cfg.MinConns != 5 {
+		t.Errorf("MinConns: want 5, got %d", cfg.MinConns)
+	}
+	if cfg.MaxConnLifetime != 2*time.Minute {
+		t.Errorf("MaxConnLifetime: want 2m, got %v", cfg.MaxConnLifetime)
+	}
+	if cfg.MaxConnIdleTime != 45*time.Second {
+		t.Errorf("MaxConnIdleTime: want 45s, got %v", cfg.MaxConnIdleTime)
+	}
+}