Address review findings on the pg read/write split

Name the replica host variable that was actually resolved. lookup falls
back from the prefixed name to the bare one, but the ambiguous-replica
error printed the prefixed form unconditionally, so a deployment setting
APP_DATABASE_URL alongside a bare DB_RO_HOST was told to go and look at
APP_DB_RO_HOST, which is not in its environment. lookupNamed reports the
variable the value came from and the error names that one.

Build the replica pool before the primary in NewCluster. pgxpool
connects lazily, so this dials nothing, but it parses the replica DSN: a
mistyped one now fails before the primary is opened, which removes the
cleanup branch that closed a primary pool nobody could observe and makes
the failure testable through the public constructor. A primary that
cannot be reached closes the lazy replica pool on the way out.

pg/cluster.go

@@ -59,6 +59,10 @@ type Cluster struct {
 
 // 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
@@ -66,31 +70,45 @@ type Cluster struct {
 //
 // 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
 	}
-	c, err := newCluster(ctx, primary, cfg)
-	if err != nil {
-		primary.Close()
-		return nil, err
-	}
-	return c, nil
+	return newCluster(ctx, primary, replica, cfg), nil
 }
 
-// newCluster attaches the replica half to an already-open primary pool.
-func newCluster(ctx context.Context, primary *pgxpool.Pool, cfg ClusterConfig) (*Cluster, error) {
-	log := logger(cfg.Logger)
-	c := &Cluster{primary: primary}
+// 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 {
-		log.Debug("postgres cluster in single-pool mode, reads go to the primary")
-		return c, nil
+		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 {
@@ -98,7 +116,7 @@ func newCluster(ctx context.Context, primary *pgxpool.Pool, cfg ClusterConfig) (
 	} else {
 		log.Debug("postgres replica pool ready", "host", replica.Config().ConnConfig.Host)
 	}
-	return c, nil
+	return c
 }
 
 // Write returns the primary pool. Every statement that changes data, takes row

pg/cluster_test.go

@@ -211,10 +211,14 @@ func TestNewCluster_SinglePoolModes(t *testing.T) {
 	}
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			c, err := newCluster(shortCtx(t), lazyPool(t, deadDSN), tc.cfg)
+			replica, err := openReplica(shortCtx(t), tc.cfg)
 			if err != nil {
-				t.Fatalf("newCluster: %v", err)
+				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")
 			}
@@ -228,14 +232,12 @@ func TestNewCluster_SinglePoolModes(t *testing.T) {
 // 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) {
-	c, err := newCluster(shortCtx(t), lazyPool(t, deadDSN), ClusterConfig{
-		PrimaryDSN: deadDSN,
-		ReplicaDSN: deadReplicaDSN,
-		Logger:     testLogger(),
-	})
+	cfg := ClusterConfig{PrimaryDSN: deadDSN, ReplicaDSN: deadReplicaDSN, Logger: testLogger()}
+	replica, err := openReplica(shortCtx(t), cfg)
 	if err != nil {
-		t.Fatalf("newCluster: %v", err)
+		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")
@@ -248,19 +250,29 @@ func TestNewCluster_UnreachableReplicaStartsDegraded(t *testing.T) {
 	}
 }
 
-func TestNewCluster_RejectsAnUnparseableReplicaDSN(t *testing.T) {
-	_, err := newCluster(shortCtx(t), lazyPool(t, deadDSN), ClusterConfig{
+// 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 {

pg/dsn.go

@@ -92,12 +92,12 @@ func DSNFromEnv(prefix string) (string, error) {
 // 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 := lookup(prefix, fieldROHost, "")
+	roHost, roHostVar := lookupNamed(prefix, fieldROHost, "")
 
 	if u := urlFromEnv(prefix, "DATABASE_URL"); u != "" {
 		if replica == "" && roHost != "" {
-			return "", "", fmt.Errorf("%s%s is set but the primary comes from a connection URL: set %sDATABASE_RO_URL instead",
-				prefix, fieldROHost.suffix, prefix)
+			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
 	}
@@ -187,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 {
+	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
+		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

@@ -348,6 +348,8 @@ func TestClusterDSNsFromEnv_Errors(t *testing.T) {
 		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",
@@ -364,7 +366,21 @@ func TestClusterDSNsFromEnv_Errors(t *testing.T) {
 				"APP_DATABASE_URL": "postgres://u:p@rw/db",
 				"APP_DB_RO_HOST":   "db-ro",
 			},
-			wantSub: "set APP_DATABASE_RO_URL instead",
+			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",
@@ -385,6 +401,9 @@ func TestClusterDSNsFromEnv_Errors(t *testing.T) {
 			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)
+			}
 		})
 	}
 }