diff --git a/pg/cluster.go b/pg/cluster.go index f8163a5..bc7dc79 100644 --- a/pg/cluster.go +++ b/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 diff --git a/pg/cluster_test.go b/pg/cluster_test.go index 0ee7e1f..50e6bd3 100644 --- a/pg/cluster_test.go +++ b/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 { diff --git a/pg/dsn.go b/pg/dsn.go index 28d5fd8..bbb30ac 100644 --- a/pg/dsn.go +++ b/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, "" } diff --git a/pg/dsn_test.go b/pg/dsn_test.go index 28c68ee..47e4491 100644 --- a/pg/dsn_test.go +++ b/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) + } }) } }