Merge pull request 'Add read/write splitting to the pg module' (#2) from benvin/pg-rw-split into main

Reviewed-on: #2

README.md

@@ -11,7 +11,7 @@ imports it inherits them.
 
 | Import | What it does |
 | --- | --- |
-| `git.unkin.net/unkin/golib/pg` | Postgres: DSN from the environment, pgxpool construction, and the estate's migration runner. |
+| `git.unkin.net/unkin/golib/pg` | Postgres: DSN from the environment, pgxpool construction, read/write splitting across a primary and a replica, and the estate's migration runner. |
 | `git.unkin.net/unkin/golib/pg/pgtest` | A throwaway Postgres container for a consumer's own `_test.go` files. Test-only. |
 
 ### pg
@@ -65,6 +65,111 @@ rather than on the first query. `pg.NewMigrated` does that and then migrates.
 `pg.LockKey(name)` exposes the derivation, so a service migrating off a
 hardcoded key can assert the two agree before switching over.
 
+#### Read/write splitting
+
+`pg.Cluster` wraps a primary pool and an optional read-replica pool:
+
+```go
+primaryDSN, replicaDSN, err := pg.ClusterDSNsFromEnv("ENCAPI_")
+if err != nil {
+    return err
+}
+db, err := pg.NewCluster(ctx, pg.ClusterConfig{
+    PrimaryDSN: primaryDSN,
+    ReplicaDSN: replicaDSN,
+    Logger:     log,
+})
+if err != nil {
+    return err
+}
+defer db.Close()
+
+if err := db.Migrate(ctx, migrations.FS, pg.MigrateOptions{LockName: "encapi-migrations"}); err != nil {
+    return err
+}
+
+rows, err := db.Read().Query(ctx, "SELECT id, name FROM nodes")   // replica when healthy
+tag, err := db.Write().Exec(ctx, "UPDATE nodes SET ...")          // always the primary
+row := db.Primary().QueryRow(ctx, "SELECT ... WHERE id = $1", id) // read-your-writes
+```
+
+| Call | Goes to |
+| --- | --- |
+| `Write()` | The primary, always. |
+| `Read()` | The replica when one is configured and healthy; the primary otherwise. |
+| `Primary()` | The primary. Same pool as `Write()`, named for reads that must not be stale. |
+
+**Routing is explicit; nothing inspects SQL.** Deciding from the statement text
+gets it wrong in both directions: a CTE with an `INSERT` in it reads as a
+`SELECT`, and `SELECT ... FOR UPDATE` takes row locks a replica cannot grant.
+The caller knows which it wants, so the caller picks.
+
+**Replica lag is real.** A replica serves a slightly stale snapshot, so a read
+that has to observe a write this process just made goes to `Primary()`. The
+usual shape is a handler that writes and then re-reads what it wrote, or a
+redirect straight into a `GET` of the row just created — both need the primary.
+Everything else (list endpoints, dashboards, reports, background aggregation)
+can take the replica.
+
+**A missing replica is not an error.** With `ReplicaDSN` empty, `Read()` returns
+the primary and the `Cluster` is an ordinary single-pool handle, so a service
+can use `Cluster` unconditionally and let the deployment decide whether reads
+split. A `ReplicaDSN` equal to `PrimaryDSN` does the same rather than opening a
+second pool to the same place.
+
+**An unhealthy replica falls back.** The `Cluster` keeps a small circuit over
+the replica:
+
+- A replica that fails its startup ping does not fail startup; reads begin on
+  the primary and move over once it answers a probe.
+- `db.ReportReplicaError(err)` — pass the error from a query run on the pool
+  `Read()` handed out — trips the circuit, and reads move to the primary.
+- While tripped, `Read()` issues no probes until the backoff window expires; the
+  next `Read()` after that pays for one probe that either promotes the replica
+  or doubles the window. The window runs from `ReplicaRetryMin` to
+  `ReplicaRetryMax` (5s to 2m by default) and resets on recovery. A healthy
+  replica is never probed at all, so the split costs nothing on the read path.
+- Reporting is advisory. A caller that never reports still routes correctly; it
+  just does not react to a replica that dies mid-flight.
+
+**Migrations always target the primary.** `Cluster.Migrate` runs through
+`Write()`. A replica is physically read-only, and a schema change has to
+originate on the primary to reach the replica at all.
+
+`pg.ClusterDSNsFromEnv(prefix)` returns both connection strings. The primary
+follows `DSNFromEnv` exactly. The replica resolves, highest first:
+
+1. `<PREFIX>DATABASE_RO_URL` — used verbatim.
+2. `DATABASE_RO_URL` — likewise.
+3. `<PREFIX>DB_RO_HOST`, or bare `DB_RO_HOST` — the primary's port, user,
+   password, database and sslmode with that host substituted.
+4. Nothing set — an empty replica DSN, so the `Cluster` runs single-pool.
+
+Nothing is derived. The read-only host is never rewritten out of the primary's,
+because a wrong guess silently sends reads somewhere unintended. Setting only
+`DB_RO_HOST` while the primary comes from a whole `DATABASE_URL` is an error,
+not a guess: the fields to substitute into are not known.
+
+CloudNativePG publishes exactly the two endpoints this expects — `<cluster>-rw`
+routes to the primary and `<cluster>-ro` to the replicas — so a Deployment names
+both:
+
+```yaml
+env:
+  - name: ENCAPI_DBHOST
+    value: encapi-db-rw
+  - name: ENCAPI_DB_RO_HOST
+    value: encapi-db-ro
+  - name: ENCAPI_DBNAME
+    value: encapi
+  - name: ENCAPI_DBUSER
+    valueFrom: { secretKeyRef: { name: encapi-db-app, key: username } }
+  - name: ENCAPI_DBPASS
+    valueFrom: { secretKeyRef: { name: encapi-db-app, key: password } }
+```
+
+Dropping `ENCAPI_DB_RO_HOST` turns the split off without a code change.
+
 ### pgtest
 
 `pgtest` starts `postgres:17-alpine` via testcontainers. Import it only from

pg/cluster.go

@@ -0,0 +1,257 @@
+package pg
+
+import (
+	"context"
+	"fmt"
+	"io/fs"
+	"log/slog"
+	"sync"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// Replica health defaults. The retry window is deliberately coarse: a replica
+// that just fell over is not coming back within a request, and probing it more
+// often only moves the failure onto the read path.
+const (
+	defaultReplicaRetryMin = 5 * time.Second
+	defaultReplicaRetryMax = 2 * time.Minute
+	replicaProbeTimeout    = 5 * time.Second
+)
+
+// ClusterConfig configures a Cluster.
+type ClusterConfig struct {
+	// PrimaryDSN is the read-write endpoint. Required.
+	PrimaryDSN string
+
+	// ReplicaDSN is the read-only endpoint. Empty puts the Cluster in
+	// single-pool mode, where Read returns the primary. A ReplicaDSN equal to
+	// PrimaryDSN does the same rather than opening a second pool to the same
+	// place.
+	ReplicaDSN string
+
+	// Logger receives replica health transitions. Nil discards them.
+	Logger *slog.Logger
+
+	// ReplicaRetryMin and ReplicaRetryMax bound the backoff window between
+	// probes of an unhealthy replica; the window doubles from Min up to Max.
+	// Zero means the defaults, 5s and 2m.
+	ReplicaRetryMin time.Duration
+	ReplicaRetryMax time.Duration
+}
+
+// Cluster routes queries across a primary and an optional read replica.
+//
+// Routing is explicit: the caller picks Read or Write per query. Nothing
+// inspects SQL to decide, because statement inspection gets it wrong in both
+// directions — a CTE with an INSERT in it reads as a SELECT, and
+// SELECT ... FOR UPDATE takes row locks a replica cannot grant.
+//
+// A Cluster with no replica is a working single-pool Cluster, so a service can
+// use it unconditionally and a deployment decides whether reads are split.
+type Cluster struct {
+	primary *pgxpool.Pool
+	// replica is nil in single-pool mode.
+	replica *pgxpool.Pool
+	health  *replicaCircuit
+}
+
+// NewCluster opens the primary and, when configured, the replica.
+//
+// The replica pool is built first. pgxpool connects lazily, so that dials
+// nothing and costs nothing, but it does parse the replica DSN: a mistyped one
+// fails before the primary is opened, leaving no pool behind to clean up.
+//
+// A primary that cannot be reached is fatal: the service has nowhere to write.
+// A replica that cannot be reached is not — the Cluster starts with reads on
+// the primary and promotes the replica once it answers a probe, which is the
+// same degradation a replica failing later gets.
+//
+// The caller owns the Cluster and must Close it.
+func NewCluster(ctx context.Context, cfg ClusterConfig) (*Cluster, error) {
+	replica, err := openReplica(ctx, cfg)
+	if err != nil {
+		return nil, err
+	}
+	primary, err := New(ctx, cfg.PrimaryDSN, cfg.Logger)
+	if err != nil {
+		if replica != nil {
+			replica.Close()
+		}
+		return nil, err
+	}
+	return newCluster(ctx, primary, replica, cfg), nil
+}
+
+// openReplica builds the replica pool, or nil when the Cluster is to run
+// single-pool: no replica configured, or one pointed at the primary, which is
+// answered with the pool already open rather than a second one to the same
+// place.
+func openReplica(ctx context.Context, cfg ClusterConfig) (*pgxpool.Pool, error) {
+	if cfg.ReplicaDSN == "" || cfg.ReplicaDSN == cfg.PrimaryDSN {
+		return nil, nil
+	}
+	replica, err := pgxpool.New(ctx, cfg.ReplicaDSN)
+	if err != nil {
+		return nil, fmt.Errorf("connect postgres replica: %w", err)
+	}
+	return replica, nil
+}
+
+// newCluster joins an already-open primary to an already-open replica, nil for
+// single-pool mode, and probes the replica once to set its starting health.
+func newCluster(ctx context.Context, primary, replica *pgxpool.Pool, cfg ClusterConfig) *Cluster {
+	log := logger(cfg.Logger)
+	c := &Cluster{primary: primary}
+	if replica == nil {
+		log.Debug("postgres cluster in single-pool mode, reads go to the primary")
+		return c
+	}
+
+	c.replica = replica
+	c.health = newReplicaCircuit(replica, log, cfg.ReplicaRetryMin, cfg.ReplicaRetryMax)
+	if err := replica.Ping(ctx); err != nil {
+		c.health.trip(err)
+	} else {
+		log.Debug("postgres replica pool ready", "host", replica.Config().ConnConfig.Host)
+	}
+	return c
+}
+
+// Write returns the primary pool. Every statement that changes data, takes row
+// locks, or must be read back immediately goes here.
+func (c *Cluster) Write() *pgxpool.Pool { return c.primary }
+
+// Read returns the pool to run a read-only query on: the replica when one is
+// configured and healthy, the primary otherwise.
+//
+// Reads served by a replica see a slightly stale snapshot. Use Primary for a
+// read that must observe a write this process just made.
+func (c *Cluster) Read() *pgxpool.Pool {
+	if c.replica == nil || !c.health.use() {
+		return c.primary
+	}
+	return c.replica
+}
+
+// Primary returns the primary pool, whatever the replica's state. It is the
+// same pool as Write, named for the read-your-writes case: a read that must not
+// be served stale asks for the primary explicitly.
+func (c *Cluster) Primary() *pgxpool.Pool { return c.primary }
+
+// ReportReplicaError tells the Cluster a query on the replica failed, so reads
+// move to the primary until the replica answers a probe again. Pass the error
+// from a query run on the pool Read returned; a nil error, or a Cluster with no
+// replica, is a no-op.
+//
+// It is advisory. A caller that does not report anything still gets correct
+// routing, just no reaction to a replica that dies mid-flight.
+func (c *Cluster) ReportReplicaError(err error) {
+	if err == nil || c.replica == nil {
+		return
+	}
+	c.health.trip(err)
+}
+
+// Migrate brings the schema up to date on the primary. Migrations never run
+// against a replica: a replica is physically read-only, and a schema change has
+// to originate on the primary to reach it at all.
+func (c *Cluster) Migrate(ctx context.Context, fsys fs.FS, opts MigrateOptions) error {
+	return Migrate(ctx, c.Write(), fsys, opts)
+}
+
+// Close closes both pools.
+func (c *Cluster) Close() {
+	if c.replica != nil {
+		c.replica.Close()
+	}
+	c.primary.Close()
+}
+
+// pinger is the slice of *pgxpool.Pool the health circuit drives, so the
+// circuit is testable without a database.
+type pinger interface {
+	Ping(ctx context.Context) error
+}
+
+// replicaCircuit tracks whether the replica is worth sending reads to.
+//
+// While healthy it costs nothing: Read consults a boolean and no probe is
+// issued. A reported error or a failed startup ping trips it, and reads go to
+// the primary until the backoff window expires, at which point the next Read
+// pays for one probe that either promotes the replica or extends the window.
+type replicaCircuit struct {
+	probe pinger
+	log   *slog.Logger
+	now   func() time.Time
+	min   time.Duration
+	max   time.Duration
+
+	mu      sync.Mutex
+	tripped bool
+	retryAt time.Time
+	backoff time.Duration
+}
+
+func newReplicaCircuit(probe pinger, log *slog.Logger, retryMin, retryMax time.Duration) *replicaCircuit {
+	if retryMin <= 0 {
+		retryMin = defaultReplicaRetryMin
+	}
+	if retryMax < retryMin {
+		retryMax = defaultReplicaRetryMax
+	}
+	if retryMax < retryMin {
+		retryMax = retryMin
+	}
+	return &replicaCircuit{probe: probe, log: log, now: time.Now, min: retryMin, max: retryMax}
+}
+
+// use reports whether the replica may serve the next read, probing it when the
+// backoff window has expired. The probe runs under the mutex so a burst of
+// concurrent reads issues one probe between them rather than one each.
+func (c *replicaCircuit) use() bool {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if !c.tripped {
+		return true
+	}
+	if c.now().Before(c.retryAt) {
+		return false
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), replicaProbeTimeout)
+	defer cancel()
+	if err := c.probe.Ping(ctx); err != nil {
+		c.log.Debug("postgres replica still unhealthy", "err", err)
+		c.backOffLocked()
+		return false
+	}
+	c.log.Info("postgres replica healthy again, reads return to it")
+	c.tripped = false
+	c.backoff = 0
+	return true
+}
+
+// trip marks the replica unhealthy. An already-tripped circuit keeps the window
+// it has, so a flood of reported errors cannot push the retry out indefinitely.
+func (c *replicaCircuit) trip(err error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.tripped {
+		return
+	}
+	c.log.Warn("postgres replica unhealthy, reads move to the primary", "err", err)
+	c.backOffLocked()
+}
+
+func (c *replicaCircuit) backOffLocked() {
+	c.tripped = true
+	switch c.backoff {
+	case 0:
+		c.backoff = c.min
+	default:
+		c.backoff = min(c.backoff*2, c.max)
+	}
+	c.retryAt = c.now().Add(c.backoff)
+}

pg/cluster_test.go

@@ -0,0 +1,331 @@
+package pg
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+	"testing/fstest"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// A second unreachable DSN, distinct from deadDSN, so a test can tell the two
+// pools of a Cluster apart by pointer.
+const deadReplicaDSN = "postgres://u:p@127.0.0.1:2/d?sslmode=disable&connect_timeout=2"
+
+// fakePinger answers probes from a scripted sequence of results, so recovery
+// can be driven without a database.
+type fakePinger struct {
+	results []error
+	calls   int
+}
+
+func (p *fakePinger) Ping(context.Context) error {
+	p.calls++
+	if len(p.results) == 0 {
+		return nil
+	}
+	i := min(p.calls-1, len(p.results)-1)
+	return p.results[i]
+}
+
+// fakeClock is a manually advanced clock for the backoff window.
+type fakeClock struct{ t time.Time }
+
+func (c *fakeClock) now() time.Time          { return c.t }
+func (c *fakeClock) advance(d time.Duration) { c.t = c.t.Add(d) }
+
+// lazyPool opens a pool without connecting: pgxpool is lazy, so this needs no
+// server and lets a test build a Cluster with two distinguishable pools.
+func lazyPool(t *testing.T, dsn string) *pgxpool.Pool {
+	t.Helper()
+	pool, err := pgxpool.New(context.Background(), dsn)
+	if err != nil {
+		t.Fatalf("pgxpool.New(%q): %v", dsn, err)
+	}
+	t.Cleanup(pool.Close)
+	return pool
+}
+
+// testCluster builds a Cluster with a replica and a circuit under a fake clock.
+func testCluster(t *testing.T, probe pinger) (*Cluster, *fakeClock) {
+	t.Helper()
+	c := &Cluster{
+		primary: lazyPool(t, deadDSN),
+		replica: lazyPool(t, deadReplicaDSN),
+	}
+	clock := &fakeClock{t: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)}
+	c.health = newReplicaCircuit(probe, testLogger(), time.Second, 4*time.Second)
+	c.health.now = clock.now
+	return c, clock
+}
+
+// The routing table: what Read and Write return in each replica state.
+func TestCluster_Routing(t *testing.T) {
+	single := &Cluster{primary: lazyPool(t, deadDSN)}
+	if single.Read() != single.primary {
+		t.Error("with no replica, Read must return the primary")
+	}
+	if single.Write() != single.primary || single.Primary() != single.primary {
+		t.Error("Write and Primary must return the primary")
+	}
+
+	c, clock := testCluster(t, &fakePinger{})
+	if c.Read() != c.replica {
+		t.Fatal("a healthy replica must serve reads")
+	}
+	if c.Write() != c.primary {
+		t.Fatal("writes must always go to the primary")
+	}
+	if c.Primary() != c.primary {
+		t.Fatal("Primary must return the primary even with a healthy replica")
+	}
+
+	c.ReportReplicaError(errors.New("replica exploded"))
+	if c.Read() != c.primary {
+		t.Fatal("an unhealthy replica must not serve reads")
+	}
+
+	// Inside the backoff window the decision stands without a probe.
+	clock.advance(500 * time.Millisecond)
+	if c.Read() != c.primary {
+		t.Fatal("reads must stay on the primary inside the backoff window")
+	}
+
+	// Once the window expires the next read probes, and a healthy probe
+	// promotes the replica again.
+	clock.advance(time.Second)
+	if c.Read() != c.replica {
+		t.Fatal("a recovered replica must serve reads again")
+	}
+}
+
+// A nil error and a Cluster without a replica must both be no-ops rather than a
+// nil-pointer dereference on the circuit.
+func TestCluster_ReportReplicaErrorIgnoresNoOps(t *testing.T) {
+	single := &Cluster{primary: lazyPool(t, deadDSN)}
+	single.ReportReplicaError(errors.New("no replica to blame"))
+
+	c, _ := testCluster(t, &fakePinger{})
+	c.ReportReplicaError(nil)
+	if c.Read() != c.replica {
+		t.Fatal("a nil error must not trip the circuit")
+	}
+}
+
+func TestReplicaCircuit_BackoffDoublesToTheCeiling(t *testing.T) {
+	probe := &fakePinger{results: []error{errors.New("down")}}
+	clock := &fakeClock{t: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)}
+	c := newReplicaCircuit(probe, testLogger(), time.Second, 4*time.Second)
+	c.now = clock.now
+
+	c.trip(errors.New("down"))
+	// 1s, then 2s, then 4s, then held at the 4s ceiling.
+	for _, want := range []time.Duration{time.Second, 2 * time.Second, 4 * time.Second, 4 * time.Second} {
+		if c.backoff != want {
+			t.Fatalf("backoff = %s, want %s", c.backoff, want)
+		}
+		// Just short of the window: no probe, still tripped.
+		before := probe.calls
+		clock.advance(want - time.Millisecond)
+		if c.use() {
+			t.Fatal("the circuit reopened before its window expired")
+		}
+		if probe.calls != before {
+			t.Fatal("the circuit probed inside its window")
+		}
+		clock.advance(time.Millisecond)
+		if c.use() {
+			t.Fatal("a failing probe must leave the circuit closed")
+		}
+		if probe.calls != before+1 {
+			t.Fatalf("probe calls = %d, want %d", probe.calls, before+1)
+		}
+	}
+}
+
+// Reported errors while already tripped must not push the retry out: a busy
+// service reporting on every read would otherwise never probe again.
+func TestReplicaCircuit_RepeatedTripsKeepTheWindow(t *testing.T) {
+	clock := &fakeClock{t: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)}
+	c := newReplicaCircuit(&fakePinger{}, testLogger(), time.Second, 4*time.Second)
+	c.now = clock.now
+
+	c.trip(errors.New("first"))
+	retryAt := c.retryAt
+	clock.advance(900 * time.Millisecond)
+	c.trip(errors.New("second"))
+	if !c.retryAt.Equal(retryAt) {
+		t.Fatalf("retryAt moved from %s to %s on a repeated trip", retryAt, c.retryAt)
+	}
+}
+
+// After a recovery the backoff restarts at the minimum rather than resuming
+// where the previous outage left off.
+func TestReplicaCircuit_RecoveryResetsTheBackoff(t *testing.T) {
+	probe := &fakePinger{results: []error{errors.New("down"), nil}}
+	clock := &fakeClock{t: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)}
+	c := newReplicaCircuit(probe, testLogger(), time.Second, 4*time.Second)
+	c.now = clock.now
+
+	c.trip(errors.New("down"))
+	clock.advance(time.Second)
+	if c.use() {
+		t.Fatal("the first probe fails, so the circuit stays closed")
+	}
+	clock.advance(2 * time.Second)
+	if !c.use() {
+		t.Fatal("the second probe succeeds, so the circuit must reopen")
+	}
+	if c.backoff != 0 {
+		t.Fatalf("backoff = %s after recovery, want 0", c.backoff)
+	}
+
+	c.trip(errors.New("down again"))
+	if c.backoff != time.Second {
+		t.Fatalf("backoff = %s on the next outage, want the 1s minimum", c.backoff)
+	}
+}
+
+func TestNewReplicaCircuit_Defaults(t *testing.T) {
+	c := newReplicaCircuit(&fakePinger{}, testLogger(), 0, 0)
+	if c.min != defaultReplicaRetryMin || c.max != defaultReplicaRetryMax {
+		t.Fatalf("defaults = %s/%s, want %s/%s", c.min, c.max, defaultReplicaRetryMin, defaultReplicaRetryMax)
+	}
+	// A minimum above the default ceiling must not produce max < min.
+	c = newReplicaCircuit(&fakePinger{}, testLogger(), time.Hour, 0)
+	if c.max < c.min {
+		t.Fatalf("max %s is below min %s", c.max, c.min)
+	}
+}
+
+func TestNewCluster_SinglePoolModes(t *testing.T) {
+	tests := []struct {
+		name string
+		cfg  ClusterConfig
+	}{
+		{"no replica configured", ClusterConfig{PrimaryDSN: deadDSN}},
+		{"replica DSN equal to the primary", ClusterConfig{PrimaryDSN: deadDSN, ReplicaDSN: deadDSN}},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			replica, err := openReplica(shortCtx(t), tc.cfg)
+			if err != nil {
+				t.Fatalf("openReplica: %v", err)
+			}
+			if replica != nil {
+				t.Fatal("no replica pool may be opened in single-pool mode")
+			}
+			c := newCluster(shortCtx(t), lazyPool(t, deadDSN), replica, tc.cfg)
+			if c.replica != nil {
+				t.Fatal("expected single-pool mode")
+			}
+			if c.Read() != c.Write() {
+				t.Fatal("in single-pool mode Read and Write must be the same pool")
+			}
+		})
+	}
+}
+
+// A replica that will not answer must degrade to primary reads, not fail
+// startup: the service can still serve everything, just without the split.
+func TestNewCluster_UnreachableReplicaStartsDegraded(t *testing.T) {
+	cfg := ClusterConfig{PrimaryDSN: deadDSN, ReplicaDSN: deadReplicaDSN, Logger: testLogger()}
+	replica, err := openReplica(shortCtx(t), cfg)
+	if err != nil {
+		t.Fatalf("openReplica: %v", err)
+	}
+	c := newCluster(shortCtx(t), lazyPool(t, deadDSN), replica, cfg)
+	t.Cleanup(func() { c.replica.Close() })
+	if c.replica == nil {
+		t.Fatal("the replica pool must still be opened")
+	}
+	if !c.health.tripped {
+		t.Fatal("a replica that failed its startup ping must start tripped")
+	}
+	if c.Read() != c.primary {
+		t.Fatal("reads must start on the primary while the replica is down")
+	}
+}
+
+// An unparseable replica DSN must be reported without the primary ever being
+// opened, so the failure path has no pool to leak. The primary here is
+// unreachable too: had it been opened first, its ping would have failed and
+// masked the replica error.
+func TestNewCluster_RejectsAnUnparseableReplicaDSNBeforeOpeningThePrimary(t *testing.T) {
+	c, err := NewCluster(shortCtx(t), ClusterConfig{
+		PrimaryDSN: deadDSN,
+		ReplicaDSN: "://not a dsn",
+	})
+	if err == nil {
+		c.Close()
+		t.Fatal("expected an error for an unparseable replica DSN")
+	}
+	if !strings.Contains(err.Error(), "connect postgres replica") {
+		t.Fatalf("error %q does not identify the failing step", err)
+	}
+	if strings.Contains(err.Error(), "ping postgres") {
+		t.Fatalf("the primary was opened before the replica DSN was parsed: %v", err)
+	}
+}
+
+// The mirror case: the replica DSN parses, so its pool exists when the primary
+// turns out to be unreachable, and NewCluster must close it on the way out.
+func TestNewCluster_PropagatesPrimaryFailure(t *testing.T) {
+	c, err := NewCluster(shortCtx(t), ClusterConfig{PrimaryDSN: deadDSN, ReplicaDSN: deadReplicaDSN})
+	if err == nil {
+		c.Close()
+		t.Fatal("expected NewCluster to fail against an unreachable primary")
+	}
+	if !strings.Contains(err.Error(), "ping postgres") {
+		t.Fatalf("error %q does not identify the failing step", err)
+	}
+}
+
+// Migrations must reach the primary even when the replica is the pool Read
+// would hand out. A closed replica pool fails Acquire with a distinctive error,
+// so routing the run to it would be visible here.
+func TestCluster_MigrateTargetsThePrimary(t *testing.T) {
+	c, _ := testCluster(t, &fakePinger{})
+	if c.Read() != c.replica {
+		t.Fatal("the replica must be healthy for this test to mean anything")
+	}
+	c.replica.Close()
+
+	err := c.Migrate(shortCtx(t), testFS(), MigrateOptions{LockName: testLockName})
+	if err == nil {
+		t.Fatal("expected the migration to fail against an unreachable primary")
+	}
+	if strings.Contains(err.Error(), "closed pool") {
+		t.Fatalf("the migration ran against the replica: %v", err)
+	}
+	if !strings.Contains(err.Error(), "acquire migration connection") {
+		t.Fatalf("error %q does not identify the failing step", err)
+	}
+}
+
+func TestCluster_MigrateRequiresALockName(t *testing.T) {
+	c := &Cluster{primary: lazyPool(t, deadDSN)}
+	if err := c.Migrate(shortCtx(t), fstest.MapFS{}, MigrateOptions{}); err == nil {
+		t.Fatal("expected Migrate to reject an empty LockName")
+	}
+}
+
+func TestCluster_CloseClosesBothPools(t *testing.T) {
+	c := &Cluster{
+		primary: lazyPool(t, deadDSN),
+		replica: lazyPool(t, deadReplicaDSN),
+	}
+	c.health = newReplicaCircuit(c.replica, testLogger(), 0, 0)
+	c.Close()
+
+	// Acquiring from a closed pool fails immediately; a live one would dial.
+	for name, pool := range map[string]*pgxpool.Pool{"primary": c.primary, "replica": c.replica} {
+		if _, err := pool.Acquire(context.Background()); err == nil || !strings.Contains(err.Error(), "closed pool") {
+			t.Errorf("%s pool was not closed: %v", name, err)
+		}
+	}
+	// Close is registered again by lazyPool's cleanup; pgxpool tolerates it.
+}

pg/dsn.go

@@ -32,6 +32,10 @@ var (
 	fieldPass = dsnField{"DBPASS", "PGPASSWORD"}
 	fieldName = dsnField{"DBNAME", "PGDATABASE"}
 	fieldSSL  = dsnField{"DBSSL", "PGSSLMODE"}
+
+	// The replica host has no libpq counterpart; the second name is the bare
+	// variable a prefixed lookup falls back to.
+	fieldROHost = dsnField{"DB_RO_HOST", "DB_RO_HOST"}
 )
 
 // DSNFromEnv builds a libpq/pgx connection string from the environment.
@@ -58,35 +62,103 @@ var (
 // User and database name have no default: DSNFromEnv reports an error naming
 // the variables it looked at rather than connecting somewhere unintended.
 func DSNFromEnv(prefix string) (string, error) {
-	if v := os.Getenv(prefix + "DATABASE_URL"); v != "" {
+	if v := urlFromEnv(prefix, "DATABASE_URL"); v != "" {
 		return v, nil
 	}
-	if v := os.Getenv("DATABASE_URL"); v != "" {
+	p, err := connPartsFromEnv(prefix)
-		return v, nil
+	if err != nil {
+		return "", err
+	}
+	return p.dsn(), nil
 }
 
-	host := lookup(prefix, fieldHost, defaultHost)
+// ClusterDSNsFromEnv resolves the primary and replica connection strings for a
+// Cluster. The primary follows DSNFromEnv exactly; the replica resolves,
+// highest first:
+//
+//  1. <PREFIX>DATABASE_RO_URL — used verbatim.
+//  2. DATABASE_RO_URL — likewise.
+//  3. <PREFIX>DB_RO_HOST, or bare DB_RO_HOST — the primary's port, user,
+//     password, database and sslmode with that host substituted, which is the
+//     CNPG shape where the <cluster>-rw and <cluster>-ro services differ only
+//     in hostname.
+//  4. Nothing set: an empty replica DSN, so the Cluster runs single-pool.
+//
+// Nothing is derived. A deployment that wants split reads names the read-only
+// endpoint; the replica host is never rewritten out of the primary's, because a
+// wrong guess silently sends reads somewhere unintended.
+//
+// A host-only replica variable set alongside a primary given as a whole URL is
+// an error rather than a guess: the fields to substitute into are not known.
+func ClusterDSNsFromEnv(prefix string) (primary, replica string, err error) {
+	replica = urlFromEnv(prefix, "DATABASE_RO_URL")
+	roHost, roHostVar := lookupNamed(prefix, fieldROHost, "")
+
+	if u := urlFromEnv(prefix, "DATABASE_URL"); u != "" {
+		if replica == "" && roHost != "" {
+			return "", "", fmt.Errorf("%s is set but the primary comes from a connection URL: set %sDATABASE_RO_URL instead",
+				roHostVar, prefix)
+		}
+		return u, replica, nil
+	}
+
+	p, err := connPartsFromEnv(prefix)
+	if err != nil {
+		return "", "", err
+	}
+	primary = p.dsn()
+	if replica != "" || roHost == "" {
+		return primary, replica, nil
+	}
+	p.host = roHost
+	return primary, p.dsn(), nil
+}
+
+// urlFromEnv reads a whole-connection-string variable: prefixed, then bare.
+func urlFromEnv(prefix, name string) string {
+	if v := os.Getenv(prefix + name); v != "" {
+		return v
+	}
+	return os.Getenv(name)
+}
+
+// connParts is one resolved set of connection fields.
+type connParts struct {
+	host string
+	port int
+	user string
+	pass string
+	name string
+	ssl  string
+}
+
+func (p connParts) dsn() string { return DSN(p.host, p.port, p.user, p.pass, p.name, p.ssl) }
+
+// connPartsFromEnv resolves levels 3 to 5 of DSNFromEnv's precedence.
+func connPartsFromEnv(prefix string) (connParts, error) {
 	portStr := lookup(prefix, fieldPort, strconv.Itoa(defaultPort))
-	user := lookup(prefix, fieldUser, "")
-	pass := lookup(prefix, fieldPass, "")
-	name := lookup(prefix, fieldName, "")
-	ssl := lookup(prefix, fieldSSL, defaultSSLMode)
-
 	port, err := strconv.Atoi(portStr)
 	if err != nil {
-		return "", fmt.Errorf("invalid %s%s: %w", prefix, fieldPort.suffix, err)
+		return connParts{}, fmt.Errorf("invalid %s%s: %w", prefix, fieldPort.suffix, err)
 	}
 	if port < 1 || port > 65535 {
-		return "", fmt.Errorf("invalid %s%s: port %d out of range", prefix, fieldPort.suffix, port)
+		return connParts{}, fmt.Errorf("invalid %s%s: port %d out of range", prefix, fieldPort.suffix, port)
 	}
-	if user == "" {
+	p := connParts{
-		return "", fmt.Errorf("no database user: set %s%s or %s", prefix, fieldUser.suffix, fieldUser.libpq)
+		host: lookup(prefix, fieldHost, defaultHost),
+		port: port,
+		user: lookup(prefix, fieldUser, ""),
+		pass: lookup(prefix, fieldPass, ""),
+		name: lookup(prefix, fieldName, ""),
+		ssl:  lookup(prefix, fieldSSL, defaultSSLMode),
 	}
-	if name == "" {
+	if p.user == "" {
-		return "", fmt.Errorf("no database name: set %s%s or %s", prefix, fieldName.suffix, fieldName.libpq)
+		return connParts{}, fmt.Errorf("no database user: set %s%s or %s", prefix, fieldUser.suffix, fieldUser.libpq)
 	}
-
+	if p.name == "" {
-	return DSN(host, port, user, pass, name, ssl), nil
+		return connParts{}, fmt.Errorf("no database name: set %s%s or %s", prefix, fieldName.suffix, fieldName.libpq)
+	}
+	return p, nil
 }
 
 // DSN renders the estate's standard connection string. User and password are
@@ -115,11 +187,19 @@ func hostPort(host string, port int) string {
 
 // lookup resolves one field: prefixed variable, then libpq variable, then def.
 func lookup(prefix string, f dsnField, def string) string {
-	if v := os.Getenv(prefix + f.suffix); v != "" {
+	v, _ := lookupNamed(prefix, f, def)
 	return v
 }
+
+// lookupNamed resolves one field and reports the variable the value came from,
+// empty when the default was used. An error message names the variable the
+// operator actually set rather than the prefixed form they may never have used.
+func lookupNamed(prefix string, f dsnField, def string) (value, name string) {
+	if v := os.Getenv(prefix + f.suffix); v != "" {
+		return v, prefix + f.suffix
+	}
 	if v := os.Getenv(f.libpq); v != "" {
-		return v
+		return v, f.libpq
 	}
-	return def
+	return def, ""
 }

pg/dsn_test.go

@@ -14,6 +14,7 @@ var dsnVars = []string{
 	"PGHOST", "PGPORT", "PGUSER", "PGPASSWORD", "PGDATABASE", "PGSSLMODE",
 	"APP_DATABASE_URL", "APP_DBHOST", "APP_DBPORT", "APP_DBUSER", "APP_DBPASS",
 	"APP_DBNAME", "APP_DBSSL",
+	"DATABASE_RO_URL", "DB_RO_HOST", "APP_DATABASE_RO_URL", "APP_DB_RO_HOST",
 }
 
 // setEnv clears every variable DSNFromEnv consults, then sets the given ones.
@@ -227,6 +228,186 @@ func TestDSNFromEnv_ErrorNamesPrefixedVar(t *testing.T) {
 	}
 }
 
+func TestClusterDSNsFromEnv(t *testing.T) {
+	// The primary every field-based case resolves to, spelled once.
+	const wantPrimary = "postgres://app:pw@db-rw:5432/appdb?sslmode=require"
+	fieldEnv := map[string]string{
+		"APP_DBHOST": "db-rw", "APP_DBUSER": "app", "APP_DBPASS": "pw",
+		"APP_DBNAME": "appdb", "APP_DBSSL": "require",
+	}
+	withFields := func(extra map[string]string) map[string]string {
+		env := map[string]string{}
+		for k, v := range fieldEnv {
+			env[k] = v
+		}
+		for k, v := range extra {
+			env[k] = v
+		}
+		return env
+	}
+
+	tests := []struct {
+		name        string
+		prefix      string
+		env         map[string]string
+		wantPrimary string
+		wantReplica string
+	}{
+		{
+			name:        "no replica variables leaves the replica empty",
+			prefix:      "APP_",
+			env:         withFields(nil),
+			wantPrimary: wantPrimary,
+			wantReplica: "",
+		},
+		{
+			name:        "prefixed DB_RO_HOST substitutes only the host",
+			prefix:      "APP_",
+			env:         withFields(map[string]string{"APP_DB_RO_HOST": "db-ro"}),
+			wantPrimary: wantPrimary,
+			wantReplica: "postgres://app:pw@db-ro:5432/appdb?sslmode=require",
+		},
+		{
+			name:        "bare DB_RO_HOST fills in for a prefixed lookup",
+			prefix:      "APP_",
+			env:         withFields(map[string]string{"DB_RO_HOST": "db-ro"}),
+			wantPrimary: wantPrimary,
+			wantReplica: "postgres://app:pw@db-ro:5432/appdb?sslmode=require",
+		},
+		{
+			name:   "prefixed DB_RO_HOST wins over the bare one",
+			prefix: "APP_",
+			env: withFields(map[string]string{
+				"APP_DB_RO_HOST": "db-ro", "DB_RO_HOST": "ignored",
+			}),
+			wantPrimary: wantPrimary,
+			wantReplica: "postgres://app:pw@db-ro:5432/appdb?sslmode=require",
+		},
+		{
+			name:   "DATABASE_RO_URL passes through verbatim and wins over the host",
+			prefix: "APP_",
+			env: withFields(map[string]string{
+				"APP_DATABASE_RO_URL": "postgres://ro@ro-host/db?application_name=reader",
+				"APP_DB_RO_HOST":      "ignored",
+			}),
+			wantPrimary: wantPrimary,
+			wantReplica: "postgres://ro@ro-host/db?application_name=reader",
+		},
+		{
+			name:   "prefixed DATABASE_RO_URL wins over the bare one",
+			prefix: "APP_",
+			env: withFields(map[string]string{
+				"APP_DATABASE_RO_URL": "postgres://app-ro@app-ro-host/db",
+				"DATABASE_RO_URL":     "postgres://bare-ro@bare-ro-host/db",
+			}),
+			wantPrimary: wantPrimary,
+			wantReplica: "postgres://app-ro@app-ro-host/db",
+		},
+		{
+			name:   "both endpoints as whole URLs",
+			prefix: "",
+			env: map[string]string{
+				"DATABASE_URL":    "postgres://u:p@rw/db",
+				"DATABASE_RO_URL": "postgres://u:p@ro/db",
+			},
+			wantPrimary: "postgres://u:p@rw/db",
+			wantReplica: "postgres://u:p@ro/db",
+		},
+		{
+			name:   "an unprefixed deployment reads the bare names",
+			prefix: "",
+			env: map[string]string{
+				"DBHOST": "cnpg-rw", "DBUSER": "u", "DBNAME": "d",
+				"DB_RO_HOST": "cnpg-ro",
+			},
+			wantPrimary: "postgres://u:@cnpg-rw:5432/d?sslmode=disable",
+			wantReplica: "postgres://u:@cnpg-ro:5432/d?sslmode=disable",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			setEnv(t, tc.env)
+			primary, replica, err := ClusterDSNsFromEnv(tc.prefix)
+			if err != nil {
+				t.Fatalf("ClusterDSNsFromEnv: %v", err)
+			}
+			if primary != tc.wantPrimary {
+				t.Errorf("primary = %q, want %q", primary, tc.wantPrimary)
+			}
+			if replica != tc.wantReplica {
+				t.Errorf("replica = %q, want %q", replica, tc.wantReplica)
+			}
+		})
+	}
+}
+
+func TestClusterDSNsFromEnv_Errors(t *testing.T) {
+	tests := []struct {
+		name    string
+		prefix  string
+		env     map[string]string
+		wantSub string
+		// notSub, when set, must not appear in the error.
+		notSub string
+	}{
+		{
+			name:    "a broken primary is reported",
+			prefix:  "",
+			env:     map[string]string{"DBUSER": "u"},
+			wantSub: "set DBNAME or PGDATABASE",
+		},
+		{
+			// The fields to substitute the read-only host into are unknown, so
+			// the alternative to an error is guessing at the connection.
+			name:   "a host-only replica alongside a URL primary is ambiguous",
+			prefix: "APP_",
+			env: map[string]string{
+				"APP_DATABASE_URL": "postgres://u:p@rw/db",
+				"APP_DB_RO_HOST":   "db-ro",
+			},
+			wantSub: "APP_DB_RO_HOST is set but the primary comes from a connection URL: set APP_DATABASE_RO_URL instead",
+		},
+		{
+			// The same ambiguity, but the host came from the bare variable the
+			// prefixed lookup falls back to. The error must name the variable
+			// the operator actually set: naming APP_DB_RO_HOST sends them
+			// looking for something that is not in their environment.
+			name:   "the bare replica host variable is named, not the prefixed one",
+			prefix: "APP_",
+			env: map[string]string{
+				"APP_DATABASE_URL": "postgres://u:p@rw/db",
+				"DB_RO_HOST":       "db-ro",
+			},
+			wantSub: "DB_RO_HOST is set but",
+			notSub:  "APP_DB_RO_HOST",
+		},
+		{
+			name:   "a broken port is reported when the replica is resolved",
+			prefix: "",
+			env: map[string]string{
+				"DBUSER": "u", "DBNAME": "d", "DBPORT": "70000", "DB_RO_HOST": "ro",
+			},
+			wantSub: "out of range",
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			setEnv(t, tc.env)
+			primary, replica, err := ClusterDSNsFromEnv(tc.prefix)
+			if err == nil {
+				t.Fatalf("expected an error, got %q / %q", primary, replica)
+			}
+			if !strings.Contains(err.Error(), tc.wantSub) {
+				t.Fatalf("error %q does not mention %q", err, tc.wantSub)
+			}
+			if tc.notSub != "" && strings.Contains(err.Error(), tc.notSub) {
+				t.Fatalf("error %q names %q, which is not the variable that was set", err, tc.notSub)
+			}
+		})
+	}
+}
+
 func TestDSN_EmptyPasswordMatchesLegacyFormat(t *testing.T) {
 	// The Sprintf builders rendered an unset password as an empty string
 	// between the colon and the "@"; keep that shape so DSNs do not churn.

pg/integration_test.go

@@ -2,6 +2,7 @@ package pg_test
 
 import (
 	"context"
+	"errors"
 	"net/url"
 	"strings"
 	"testing"
@@ -167,6 +168,58 @@ func TestDSNFromEnv_ConnectsToRealPostgres(t *testing.T) {
 	pool.Close()
 }
 
+// A Cluster against a real server. One container plays both roles — what is
+// under test is the routing, not Postgres' own replication — so the read pool
+// is the same server reached through a second, distinct DSN.
+func TestCluster_AgainstRealPostgres(t *testing.T) {
+	ctx := testCtx(t)
+	dsn := pgtest.MustStartPostgres(ctx, t)
+
+	c, err := pg.NewCluster(ctx, pg.ClusterConfig{
+		PrimaryDSN: dsn,
+		ReplicaDSN: dsn + "&application_name=reader",
+		// Short enough that the recovery probe lands inside the test.
+		ReplicaRetryMin: 10 * time.Millisecond,
+		ReplicaRetryMax: 10 * time.Millisecond,
+	})
+	if err != nil {
+		t.Fatalf("NewCluster: %v", err)
+	}
+	t.Cleanup(c.Close)
+
+	if c.Read() == c.Write() {
+		t.Fatal("a configured, healthy replica must be a distinct pool")
+	}
+	if err := c.Migrate(ctx, migrations, pg.MigrateOptions{LockName: "golib-pg-cluster-integration"}); err != nil {
+		t.Fatalf("Migrate: %v", err)
+	}
+	if _, err := c.Write().Exec(ctx, "INSERT INTO widgets (name) VALUES ($1)", "sprocket"); err != nil {
+		t.Fatalf("insert on the primary: %v", err)
+	}
+
+	var n int
+	if err := c.Read().QueryRow(ctx, "SELECT count(*) FROM widgets").Scan(&n); err != nil {
+		t.Fatalf("read: %v", err)
+	}
+	if err := c.Primary().QueryRow(ctx, "SELECT count(*) FROM widgets").Scan(&n); err != nil {
+		t.Fatalf("read-your-writes: %v", err)
+	}
+	if n != 1 {
+		t.Fatalf("widgets has %d rows, want 1", n)
+	}
+
+	// A reported replica failure moves reads to the primary; a replica that
+	// still answers earns them back on the next probe.
+	c.ReportReplicaError(errors.New("simulated replica failure"))
+	if c.Read() != c.Write() {
+		t.Fatal("a reported replica failure must move reads to the primary")
+	}
+	time.Sleep(20 * time.Millisecond)
+	if c.Read() == c.Write() {
+		t.Fatal("a replica that answers its probe must get reads back")
+	}
+}
+
 // dsnParts is the container DSN split back into the fields DSNFromEnv reads.
 type dsnParts struct{ host, port, user, pass, name string }
 

pg/pg.go

@@ -1,6 +1,7 @@
 // Package pg holds the estate's shared Postgres plumbing: environment-driven
-// DSN construction, pgxpool construction, and the migration runner every
+// DSN construction, pgxpool construction, read/write splitting across a primary
-// service uses to bring its own schema up to date at startup.
+// and a read replica, and the migration runner every service uses to bring its
+// own schema up to date at startup.
 package pg
 
 import (