postgres.go 1.3 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556
  1. // Package postgres wraps pgx for the few queries the services need.
  2. // M0 doesn't use Postgres; M1 needs it for routing + delivery.
  3. //
  4. // We use the pgxpool connection pool so concurrent queries don't
  5. // step on each other.
  6. package postgres
  7. import (
  8. "context"
  9. "fmt"
  10. "time"
  11. "github.com/jackc/pgx/v5/pgxpool"
  12. )
  13. // Pool is a thin alias for *pgxpool.Pool so the call sites don't
  14. // import pgx directly.
  15. type Pool = pgxpool.Pool
  16. // Connect dials Postgres with retries. The first migration run will
  17. // hit this before the DB is up, so we wait.
  18. func Connect(ctx context.Context, dsn string) (*Pool, error) {
  19. cfg, err := pgxpool.ParseConfig(dsn)
  20. if err != nil {
  21. return nil, fmt.Errorf("parse dsn: %w", err)
  22. }
  23. cfg.MaxConns = 20
  24. cfg.MinConns = 2
  25. cfg.MaxConnIdleTime = 5 * time.Minute
  26. var pool *Pool
  27. deadline, hasDeadline := ctx.Deadline()
  28. if !hasDeadline {
  29. deadline = time.Now().Add(30 * time.Second)
  30. }
  31. for {
  32. pool, err = pgxpool.NewWithConfig(ctx, cfg)
  33. if err == nil {
  34. pingCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
  35. err = pool.Ping(pingCtx)
  36. cancel()
  37. if err == nil {
  38. return pool, nil
  39. }
  40. pool.Close()
  41. }
  42. if time.Now().After(deadline) {
  43. return nil, fmt.Errorf("postgres connect timeout: %w", err)
  44. }
  45. select {
  46. case <-ctx.Done():
  47. return nil, ctx.Err()
  48. case <-time.After(1 * time.Second):
  49. }
  50. }
  51. }