All checks were successful
CI / test (push) Successful in 51s
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
179 lines
6.1 KiB
Markdown
179 lines
6.1 KiB
Markdown
# dbx
|
|
|
|
PostgreSQL cluster library for Go built on pgx/v5. Master/replica routing, automatic retries with exponential backoff, round-robin load balancing, background health checking, panic-safe transactions, and context-based Querier injection.
|
|
|
|
```
|
|
go get git.codelab.vc/pkg/dbx
|
|
```
|
|
|
|
## Quick start
|
|
|
|
```go
|
|
cluster, err := dbx.NewCluster(ctx, dbx.Config{
|
|
Master: dbx.NodeConfig{
|
|
Name: "master",
|
|
DSN: "postgres://user:pass@master:5432/mydb",
|
|
Pool: dbx.PoolConfig{MaxConns: 20, MinConns: 5},
|
|
},
|
|
Replicas: []dbx.NodeConfig{
|
|
{Name: "replica-1", DSN: "postgres://user:pass@replica1:5432/mydb"},
|
|
{Name: "replica-2", DSN: "postgres://user:pass@replica2:5432/mydb"},
|
|
},
|
|
})
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
defer cluster.Close()
|
|
|
|
// Write → master
|
|
cluster.Exec(ctx, "INSERT INTO users (name) VALUES ($1)", "alice")
|
|
|
|
// Read → replica with automatic fallback to master
|
|
rows, err := cluster.ReadQuery(ctx, "SELECT * FROM users WHERE active = $1", true)
|
|
|
|
// Transaction → master, panic-safe
|
|
cluster.RunTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
|
|
tx.Exec(ctx, "UPDATE accounts SET balance = balance - $1 WHERE id = $2", 100, fromID)
|
|
tx.Exec(ctx, "UPDATE accounts SET balance = balance + $1 WHERE id = $2", 100, toID)
|
|
return nil
|
|
})
|
|
```
|
|
|
|
## Components
|
|
|
|
| Component | What it does |
|
|
|-----------|-------------|
|
|
| `Cluster` | Entry point. Connects to master + replicas, routes queries, manages lifecycle. |
|
|
| `Node` | Wraps `pgxpool.Pool` with health state and a human-readable name. |
|
|
| `Balancer` | Interface for replica selection. Built-in: `RoundRobinBalancer`. |
|
|
| `retrier` | Exponential backoff with jitter, node fallback, custom error classifiers. |
|
|
| `healthChecker` | Background goroutine that pings all nodes on an interval. |
|
|
| `Querier` injection | `InjectQuerier` / `ExtractQuerier` — context-based Querier for service layers. |
|
|
| `MetricsHook` | Optional callbacks: query start/end, retry, node up/down, replica fallback. |
|
|
|
|
## Routing
|
|
|
|
The library uses explicit method-based routing (no SQL parsing):
|
|
|
|
```
|
|
┌──────────────┐
|
|
│ Cluster │
|
|
└──────┬───────┘
|
|
│
|
|
┌───────────────┴───────────────┐
|
|
│ │
|
|
Write ops Read ops
|
|
Exec, Query, QueryRow ReadQuery, ReadQueryRow
|
|
Begin, BeginTx, RunTx
|
|
CopyFrom, SendBatch
|
|
│ │
|
|
▼ ▼
|
|
┌──────────┐ ┌────────────────────────┐
|
|
│ Master │ │ Balancer → Replicas │
|
|
└──────────┘ │ fallback → Master │
|
|
└────────────────────────┘
|
|
```
|
|
|
|
Direct node access: `cluster.Master()` and `cluster.Replica()` return `DB`.
|
|
|
|
## Multi-replica setup
|
|
|
|
```go
|
|
cluster, _ := dbx.NewCluster(ctx, dbx.Config{
|
|
Master: dbx.NodeConfig{
|
|
Name: "master",
|
|
DSN: "postgres://master:5432/mydb",
|
|
Pool: dbx.PoolConfig{MaxConns: 20, MinConns: 5},
|
|
},
|
|
Replicas: []dbx.NodeConfig{
|
|
{Name: "replica-1", DSN: "postgres://replica1:5432/mydb"},
|
|
{Name: "replica-2", DSN: "postgres://replica2:5432/mydb"},
|
|
{Name: "replica-3", DSN: "postgres://replica3:5432/mydb"},
|
|
},
|
|
Retry: dbx.RetryConfig{
|
|
MaxAttempts: 5,
|
|
BaseDelay: 100 * time.Millisecond,
|
|
MaxDelay: 2 * time.Second,
|
|
},
|
|
HealthCheck: dbx.HealthCheckConfig{
|
|
Interval: 3 * time.Second,
|
|
Timeout: 1 * time.Second,
|
|
},
|
|
})
|
|
defer cluster.Close()
|
|
```
|
|
|
|
## Transactions
|
|
|
|
`RunTx` is panic-safe — if the callback panics, the transaction is rolled back and the panic is re-raised:
|
|
|
|
```go
|
|
err := cluster.RunTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
|
|
_, err := tx.Exec(ctx, "UPDATE accounts SET balance = balance - $1 WHERE id = $2", amount, fromID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = tx.Exec(ctx, "UPDATE accounts SET balance = balance + $1 WHERE id = $2", amount, toID)
|
|
return err
|
|
})
|
|
```
|
|
|
|
For custom isolation levels use `RunTxOptions`:
|
|
|
|
```go
|
|
cluster.RunTxOptions(ctx, pgx.TxOptions{
|
|
IsoLevel: pgx.Serializable,
|
|
}, fn)
|
|
```
|
|
|
|
## Context-based Querier injection
|
|
|
|
Pass the Querier through context so service layers work both inside and outside transactions:
|
|
|
|
```go
|
|
// Repository
|
|
func CreateUser(ctx context.Context, db dbx.Querier, name string) error {
|
|
q := dbx.ExtractQuerier(ctx, db)
|
|
_, err := q.Exec(ctx, "INSERT INTO users (name) VALUES ($1)", name)
|
|
return err
|
|
}
|
|
|
|
// Outside transaction — uses cluster directly
|
|
CreateUser(ctx, cluster, "alice")
|
|
|
|
// Inside transaction — uses tx
|
|
cluster.RunTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
|
|
ctx = dbx.InjectQuerier(ctx, tx)
|
|
return CreateUser(ctx, cluster, "alice") // will use tx from context
|
|
})
|
|
```
|
|
|
|
## Error classification
|
|
|
|
```go
|
|
dbx.IsRetryable(err) // connection errors, serialization failures, deadlocks, too_many_connections
|
|
dbx.IsConnectionError(err) // PG class 08 + common connection error strings
|
|
dbx.IsConstraintViolation(err) // PG class 23 (unique, FK, check violations)
|
|
dbx.PgErrorCode(err) // extract raw PG error code
|
|
```
|
|
|
|
Sentinel errors: `ErrNoHealthyNode`, `ErrClusterClosed`, `ErrRetryExhausted`.
|
|
|
|
## dbxtest helpers
|
|
|
|
The `dbxtest` package provides test helpers:
|
|
|
|
```go
|
|
func TestMyRepo(t *testing.T) {
|
|
cluster := dbxtest.NewTestCluster(t, dbx.WithLogger(&dbxtest.TestLogger{T: t}))
|
|
// cluster is auto-closed when test finishes
|
|
// skips test if DB is not reachable
|
|
}
|
|
```
|
|
|
|
Set `DBX_TEST_DSN` env var to override the default DSN (`postgres://postgres:postgres@localhost:5432/dbx_test?sslmode=disable`).
|
|
|
|
## Requirements
|
|
|
|
Go 1.24+, [pgx/v5](https://github.com/jackc/pgx).
|