Help us shape the future of Go! Take the 2026 Go Developer Survey

crdbpgx

package module
v1.1.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 17, 2026 License: Apache-2.0 Imports: 17 Imported by: 0

README

crdbpgx

Go Reference CI Status License

A pgx-compatible driver for CockroachDB. crdbpgx supports tracking of CockroachDB gateway nodes health and automatic rebalancing of connections for failing nodes or nodes draining during upgrades. The driver also transparently handles retryable errors.

Inspired by AuthZed's Maximizing CockroachDB Performance: Our Journey to 1 Million QPS and their implementation authzed/crdbpool. Further enhanced with configurable context propagation and improved load-balancing. Distributed under the same license.

[!IMPORTANT] crdbpgx is not a drop-in replacement for pgx. It is a CockroachDB-focused wrapper around pgxpool that adds node health tracking, connection rebalancing, retry handling, and configurable context propagation. Those behaviors intentionally change how connections are acquired, retried, and managed so applications can better tolerate CockroachDB node failures, draining nodes during upgrades, and gateway imbalance.

Example

package main

import (
  "context"
  "fmt"
  "os"
  "os/signal"
  "time"

  "github.com/jackc/pgx/v5/pgxpool"
  "github.com/sumup/crdbpgx"
)

func main() {
  // your application context
  ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, os.Kill)
  defer cancel()

  cfg, err := pgxpool.ParseConfig(os.Getenv("DSN"))
  if err != nil {
    panic(fmt.Errorf("parse dsn: %w", err))
  }

  pool, err := crdbpgx.NewPool(
    ctx, // context used only while the pool is created
    cfg,
    crdbpgx.WithContextPropagationBehaviour(crdbpgx.ContextPropagationOnAcquire),
    crdbpgx.WithBalancer(crdbpgx.NewNodeConnectionBalancer(
      crdbpgx.WithBalancingInterval(30*time.Second),
      crdbpgx.WithBalancingTimeout(10*time.Second),
    )),
  )
  if err != nil {
    panic(fmt.Errorf("create read pool: %w", err))
  }
  defer pool.Close()

  // use your pool
}

Typed Queries and Transactions

Pool embeds a concrete Runner, so typed queries and commands are available as pool.Query, pool.QueryRow, and pool.Exec on Go 1.27+:

users, err := pool.Query(ctx, pgx.RowToStructByName[User], "SELECT id, name FROM users")

err = pool.InTx(ctx, pgx.TxOptions{}, func(tx crdbpgx.Runner) error {
  user, err := tx.QueryRow(ctx, pgx.RowToStructByName[User],
    "SELECT id, name FROM users WHERE id = $1", userID)
  if err != nil {
    return err
  }
  _, err = tx.Exec(ctx, "UPDATE users SET name = $1 WHERE id = $2", name, user.ID)
  return err
})

Store pool.Runner in repository code to use the same concrete type inside and outside transactions:

type Repository struct {
  db crdbpgx.Runner
}

repo := Repository{db: pool.Runner}

Runner.InTx supplies a transaction-bound Runner automatically. Calling it on that runner creates a savepoint. Only the outer pool retries the entire transaction. Callbacks can execute again during retries and should avoid external side effects. Do not retain a transaction runner after its callback.

Both query methods close their rows. QueryRow returns the first row, ignores additional rows, and returns pgx.ErrNoRows for an empty result. On error, Query returns nil, QueryRow returns the zero value, and Exec returns an empty command tag. Typed transaction operations honor the pool's configured context propagation behavior.

Executor is the underlying callback interface for adapters and test doubles. NewRunner(executor) wraps these implementations. Existing callback methods remain available. InTxFunc still supplies a Tx for callers that need the raw pgx.Tx methods.

This is a breaking API change: the former Runner interface is now Executor, Runner is concrete, and InTx callbacks receive func(Runner) error.

Components

There are three main components of crdbpgx:

Pool

The database connection pool is a thin wrapper around pgxpool adjusted to CockroachDB that transparently retries retryable errors. Optional features, such as the node connection balancer, are configured with NewPool options.

Node Health Tracker

Tracks health of nodes in the cluster. This is done in two ways - when executing queries and in a periodical polling loop. In queries, nodes are marked as unhealthy if they fail with a resettable error (i.e. errors that are not a result of a database operation but rather of the underlying connection). The polling mechanism then periodically pings a random node in the cluster and if a node comes back to live or a new node is added their health is tracked as healthy.

After a topology change, the built-in health tracker temporarily uses its topology-change polling interval for 5 minutes. By default this interval is 5 seconds. This helps discover newly added or rejoined nodes faster during rolling upgrades and scale-up events when configured with a lower interval. The accelerated interval and window can be configured with WithTopologyChangePollInterval.

The health tracker is owned by the pool as they are coupled closely together but users of this package are free to provide their own implementation of the node health tracker.

Node Connection Balancer

Node connection balancer periodically acquires idle connections from the pool and prunes connections that make the distribution across healthy nodes uneven. Configure it with WithBalancer(crdbpgx.NewNodeConnectionBalancer(...)) when creating a pool. The pool starts the balancer loop with pool-owned lifecycle context and stops it on Close().

When used with the built-in health tracker, the balancer also reacts to topology changes: it rebalances immediately, then temporarily lowers its interval from 30 seconds to 5 seconds for 5 minutes. The accelerated interval and window can be configured with WithTopologyChangeBalancingInterval.

If you need manual lifecycle control, create a balancer with NewNodeConnectionBalancer, call Run(ctx, pool) in your own goroutine for the periodic loop, or call Rebalance(ctx, pool) to perform a single balancing operation.

Instrumentation

Instrumentation is optional. Choose the adapters you need:

Module Provides
crdbotel OpenTelemetry metrics, query spans, and connection failover events
crdbprom Native Prometheus metrics

Register instrumentation with WithHooks when creating a pool:

pool, err := crdbpgx.NewPool(ctx, cfg,
  crdbpgx.WithName("orders"),
  crdbpgx.WithHooks(crdbotel.NewMeter(), crdbotel.NewTracer()),
)
if err != nil {
  return err
}
defer pool.Close()

Metrics and tracing can be enabled independently or combined with other observers. For example, WithHooks(prometheusMeter, crdbotel.NewTracer()) combines native Prometheus metrics with OpenTelemetry query spans and failover events. Query hooks are automatically composed with any existing cfg.ConnConfig.Tracer.

crdbotel.NewTracer(crdbotel.WithTracerProvider(provider)) selects a provider for query spans. Without an explicit provider, the global OpenTelemetry tracer provider is used.

crdbotel.NewMeter() automatically registers pool statistics through WithHooks and unregisters them on pool.Close(). Use a unique WithName for each pool. Statistics include idle and acquired connections (db.client.connection.count, with idle and used states), the pool limit (db.client.connection.max), and crdbpgx.pool.* metrics for connections being constructed, successful/canceled acquisitions, acquisitions that waited, new connections, and connections destroyed due to lifetime or idle limits. All carry db.client.connection.pool.name and any WithMeterAttributes attributes. Acquisition duration and wait duration are cumulative seconds for successful acquisitions. pgx does not expose the number of pending acquisitions or separate timeouts from other cancellations, so those metrics are not emitted.

Pool hooks also cover the built-in balancer and automatically created health tracker. For a caller-created or shared tracker, configure instrumentation with WithHealthTrackerHooks and pass the tracker to each pool with WithHealthTracker:

tracker, err := crdbpgx.NewNodeHealthTracker(cfg.ConnConfig,
  crdbpgx.WithHealthTrackerHooks(meter),
)
Custom Hooks

Embed crdbpgx.HooksDefaults by value and implement the callbacks you need:

type retryLogger struct {
  crdbpgx.HooksDefaults
}

var _ crdbpgx.Hook = retryLogger{}

func (retryLogger) OnOperationEnd(ctx context.Context, data crdbpgx.OperationEndData) {
  if data.Retries > 0 {
    slog.DebugContext(ctx, "database operation retried",
      "pool", data.PoolName, "retries", data.Retries)
  }
}

// Pass this option when creating a pool.
option := crdbpgx.WithHooks(meter, retryLogger{})

Hook defines all lifecycle callbacks; HooksDefaults supplies no-op implementations, including for callbacks added in future versions. Repeated WithHooks and WithHealthTrackerHooks calls append observers. Hooks run synchronously in registration order and may run concurrently across operations. They must return promptly and support concurrent use.

Pool hooks also implement pgx's TraceQueryStart and TraceQueryEnd. These callbacks observe each Exec, Query, and QueryRow attempt, including queries on directly acquired connections. TraceQueryStart must return a context, which is passed to subsequent hooks and pgx. TraceQueryEnd runs when pgx completes the query or closes its rows. OnOperationEnd reports the full retry-managed operation.

Maintainers

Documentation

Overview

Package crdbpgx provides a CockroachDB-aware wrapper around github.com/jackc/pgx/v5/pgxpool.

It is intended for applications that connect to CockroachDB with pgx and want pool-level behavior that is specific to CockroachDB clusters: automatic retry handling for retryable transaction errors, connection reset and reacquisition for node-level failures, node health tracking, and optional connection rebalancing across CockroachDB gateway nodes.

The package is inspired by the mechanisms described in AuthZed's article "Maximizing CockroachDB Performance: Our Journey to 1 Million QPS": https://authzed.com/blog/maximizing-cockroachdb-performance

Pool lifecycle

Create a Pool with NewPool from a pgxpool.Config. The context passed to NewPool is used only while the underlying pgx pool is created. Runtime background work started by this package, such as the default health tracker and the optional node connection balancer, is owned by the Pool and is stopped by Pool.Close.

Always close the pool when it is no longer needed: 

defer pool.Close()

Pool.Close starts shutdown once and waits for internal goroutines and the underlying pgx pool to finish closing.

Context propagation

Database operation contexts can affect connection health in pgx: canceling a context while a query is active can leave the connection unusable. crdbpgx exposes WithContextPropagationBehaviour to control how caller contexts are propagated to connection acquisition and database operations.

The default, ContextPropagationDetach, detaches both acquisition and database operations from caller cancellation. ContextPropagationOnAcquire keeps caller cancellation for acquiring a connection, but detaches the actual database operation. ContextPropagationPassthrough passes caller contexts directly to pgx.

Example: create a pool 

cfg, err := pgxpool.ParseConfig(os.Getenv("DATABASE_URL"))
if err != nil {
	return fmt.Errorf("parse database URL: %w", err)
}

pool, err := crdbpgx.NewPool(
	context.Background(),
	cfg,
	crdbpgx.WithName("orders"),
	crdbpgx.WithContextPropagationBehaviour(crdbpgx.ContextPropagationOnAcquire),
	crdbpgx.WithBalancer(crdbpgx.NewNodeConnectionBalancer(
		crdbpgx.WithBalancingInterval(30*time.Second),
		crdbpgx.WithBalancingTimeout(10*time.Second),
	)),
)
if err != nil {
	return fmt.Errorf("create pool: %w", err)
}
defer pool.Close()

Example: run a transaction with automatic retries 

err := pool.InTx(ctx, pgx.TxOptions{}, func(db crdbpgx.Runner) error {
	return db.ExecFunc(ctx, func(_ context.Context, _ pgconn.CommandTag, err error) error {
		return err
	}, "UPDATE accounts SET balance = balance - $1 WHERE id = $2", amount, accountID)
})
if err != nil {
	return fmt.Errorf("transfer funds: %w", err)
}

Index

Examples

Constants

View Source 
const (
	// DefaultMaxRetries gives the number of transaction retries for a single transaction.
	// This is the same value as its used in the official cockroach-go library.
	// https://github.com/cockroachdb/cockroach-go/blob/master/crdb/tx.go#L114
	DefaultMaxRetries = 50

	// DefaultConnectRate limits how often the pool opens new database connections.
	DefaultConnectRate = 100 * time.Millisecond

	// DefaultContextPropagationBehaviour is the default behavior for context propagation.
	DefaultContextPropagationBehaviour = ContextPropagationDetach
)
View Source 
const (
	// CrdbRetryErrCode is the PostgreSQL error code for transaction retry errors in CockroachDB.
	// https://www.cockroachlabs.com/docs/stable/common-errors.html#restart-transaction
	CrdbRetryErrCode = "40001"

	// CrdbAmbiguousErrorCode is the PostgreSQL error code for ambiguous result errors in CockroachDB.
	// https://www.cockroachlabs.com/docs/stable/common-errors.html#result-is-ambiguous
	CrdbAmbiguousErrorCode = "40003"

	// CrdbServerNotAcceptingClients is the PostgreSQL error code when a CockroachDB node is shutting down.
	// https://www.cockroachlabs.com/docs/stable/node-shutdown.html#connection-retry-loop
	CrdbServerNotAcceptingClients = "57P01"

	// CrdbUnknownSQLState is returned when the SQL state cannot be determined.
	CrdbUnknownSQLState = "XXUUU"

	// CrdbClockSkewMessage is the error message when CockroachDB nodes have large clock skew.
	CrdbClockSkewMessage = "cannot specify timestamp in the future"
)

Variables

View Source 
var CtxDisableRetries = ctxDisableRetries{}

CtxDisableRetries is a context key used to disable automatic retries.

Functions

func IsResettableError 

func IsResettableError(ctx context.Context, err error) bool

IsResettableError returns true if the error indicates that the connection should be reset and a retry attempted on a different node.

Resettable errors typically indicate node failures, network issues, or ambiguous transaction results. The connection is no longer usable and should be replaced.

func IsRetryableError 

func IsRetryableError(ctx context.Context, err error) bool

IsRetryableError returns true if the error is a CockroachDB retryable error.

Retryable errors are serialization conflicts that can be resolved by retrying the transaction. The connection remains valid and can be reused.

func NewNodeHealthTracker 

func NewNodeHealthTracker(cfg *pgx.ConnConfig, opts ...NodeHealthTrackerOption) (_ *nodeHealthTracker, err error)

NewNodeHealthTracker creates a node health tracker. Call Poll in a goroutine to discover healthy nodes until its context is canceled. When sharing the tracker through WithHealthTracker, the caller owns its polling lifecycle.

func RetriesDisabled 

func RetriesDisabled(ctx context.Context) bool

RetriesDisabled returns true if automatic retries are disabled in the context.

func WithDisableRetries 

func WithDisableRetries(ctx context.Context) context.Context

WithDisableRetries returns a context that disables automatic retry logic for database operations. This is useful when you need predictable behavior or want to handle retries manually.

Types

type ContextPropagationBehaviour 

type ContextPropagationBehaviour int

ContextPropagationBehaviour controls how contexts are propagated from callers to database operations. This affects context cancellation behavior and is important for preventing unwanted connection closures during operations. 

const (
	// ContextPropagationPassthrough propagates the caller's context directly to
	// database operations without modification. This is the standard behavior but
	// may cause connection closures if contexts are canceled during operations.
	ContextPropagationPassthrough ContextPropagationBehaviour = iota

	// ContextPropagationOnAcquire propagates the caller's context when acquiring
	// a connection from the pool, but detaches it for the actual database operation.
	// This prevents context cancellation from affecting ongoing operations while
	// still respecting cancellation during connection acquisition.
	ContextPropagationOnAcquire

	// ContextPropagationDetach removes caller cancellation and deadlines from
	// both connection acquisition and database operations. Context values,
	// including tracing information, are preserved.
	ContextPropagationDetach
)

type Executor added in v1.1.0 

type Executor interface {
	ExecFunc(ctx context.Context, tagFunc func(context.Context, pgconn.CommandTag, error) error, sql string, arguments ...any) error
	QueryFunc(ctx context.Context, rowsFunc func(context.Context, pgx.Rows) error, sql string, optionsAndArgs ...any) error
	QueryRowFunc(ctx context.Context, rowFunc func(context.Context, pgx.Row) error, sql string, optionsAndArgs ...any) error
	// InTxFunc ignores [pgx.TxOptions] when invoked as a nested (pseudo) transaction.
	InTxFunc(context.Context, pgx.TxOptions, func(Tx) error) error
}

Executor is the callback execution interface implemented by both Pool and Tx.

Use Runner for typed queries and transaction callbacks. Executor is the underlying execution seam for adapters and test implementations.

type HealthUpdateData added in v1.1.0 

type HealthUpdateData struct {
	// NodeID identifies the node whose status was updated.
	NodeID uint32
	// Healthy is the node's effective health after applying the failure threshold.
	Healthy bool
	// HealthyNodes counts healthy nodes in the tracker after this update.
	HealthyNodes int
}

HealthUpdateData is a snapshot of health after the tracker's failure threshold has been applied. Healthy is the effective status, not the requested status. Health belongs to the tracker, which may be shared by multiple pools.

type Hook added in v1.1.0 

type Hook interface {
	// QueryTracer observes individual Exec, Query, and QueryRow attempts through
	// pgx, including operations on directly acquired connections. Query start
	// contexts flow through subsequent hooks and into query end callbacks.
	pgx.QueryTracer

	// OnPoolCreated runs once after the underlying pool is initialized, before
	// NewPool returns. The returned cleanup, if non-nil, runs once on Close,
	// before the underlying pool is closed. It must stop any registered observers.
	// Neither callback may close the pool.
	OnPoolCreated(*Pool) func()

	// OnOperationEnd observes retry-managed operations after their connections
	// have been released, including when initial acquisition fails.
	OnOperationEnd(context.Context, OperationEndData)

	// OnNodeSwitchStart observes attempts to acquire a replacement connection.
	OnNodeSwitchStart(context.Context, NodeSwitchStartData)

	// OnNodeSwitchEnd observes replacement acquisition results, including failures.
	OnNodeSwitchEnd(context.Context, NodeSwitchEndData)

	// OnRebalanceEnd observes built-in balancing runs after idle connections have
	// been released.
	OnRebalanceEnd(context.Context, RebalanceEndData)

	// OnHealthUpdate observes each SetStatus call on an instrumented built-in
	// tracker. Concurrent updates may be delivered out of order.
	OnHealthUpdate(HealthUpdateData)
}

Hook observes pool activity. Implementations should embed HooksDefaults by value and override the callbacks they need. Embedding also supplies defaults for callbacks added in future versions. Nil entries are ignored.

Hooks run synchronously in registration order, outside internal locks, and may run concurrently. They must be safe for concurrent use, return promptly, and must not close the pool. Contexts retain caller values when cancellation is detached by the pool's context propagation policy. Hooks may also implement pgx.BatchTracer to observe SendBatch; these callbacks are discovered automatically and composed with existing tracers.

type HooksDefaults added in v1.1.0 

type HooksDefaults struct{}

HooksDefaults implements Hook with no-op callbacks. Embed it by value in custom hook implementations and override only the callbacks needed. Its zero value is ready to use and safe for concurrent use.

func (HooksDefaults) OnHealthUpdate added in v1.1.0 

func (HooksDefaults) OnHealthUpdate(HealthUpdateData)

OnHealthUpdate implements Hook as a no-op.

func (HooksDefaults) OnNodeSwitchEnd added in v1.1.0 

func (HooksDefaults) OnNodeSwitchEnd(context.Context, NodeSwitchEndData)

OnNodeSwitchEnd implements Hook as a no-op.

func (HooksDefaults) OnNodeSwitchStart added in v1.1.0 

func (HooksDefaults) OnNodeSwitchStart(context.Context, NodeSwitchStartData)

OnNodeSwitchStart implements Hook as a no-op.

func (HooksDefaults) OnOperationEnd added in v1.1.0 

func (HooksDefaults) OnOperationEnd(context.Context, OperationEndData)

OnOperationEnd implements Hook as a no-op.

func (HooksDefaults) OnPoolCreated added in v1.1.1 

func (HooksDefaults) OnPoolCreated(*Pool) func()

OnPoolCreated implements Hook without registering any observers.

func (HooksDefaults) OnRebalanceEnd added in v1.1.0 

func (HooksDefaults) OnRebalanceEnd(context.Context, RebalanceEndData)

OnRebalanceEnd implements Hook as a no-op.

func (HooksDefaults) TraceQueryEnd added in v1.1.0 

func (HooksDefaults) TraceQueryEnd(context.Context, *pgx.Conn, pgx.TraceQueryEndData)

TraceQueryEnd implements pgx.QueryTracer as a no-op.

func (HooksDefaults) TraceQueryStart added in v1.1.0 

func (HooksDefaults) TraceQueryStart(ctx context.Context, _ *pgx.Conn, _ pgx.TraceQueryStartData) context.Context

TraceQueryStart implements pgx.QueryTracer by returning ctx unchanged.

type Jitter 

type Jitter interface {
	// Jitter consumes an interval from a ticker and returns a jittered duration.
	Jitter(time.Duration) time.Duration
}

Jitter applies randomness to durations to prevent thundering herd problems.

type JitteredTicker 

type JitteredTicker struct {
	// Jitter determines the delay between ticks. It must not be nil.
	Jitter
	// C receives ticks and is closed when the ticker's loop stops.
	C <-chan time.Time
	// contains filtered or unexported fields
}

JitteredTicker is a ticker with an additional jitter applied to the interval. Similar to time.Ticker but with randomized intervals.

func NewJitteredTicker 

func NewJitteredTicker(d time.Duration, j Jitter) *JitteredTicker

NewJitteredTicker creates a new jittered ticker that ticks at intervals around the specified duration, modified by the provided jitter.

func (*JitteredTicker) Stop 

func (t *JitteredTicker) Stop()

Stop requests that the ticker stop and close C. It must be called only once.

type MaxRetryError 

type MaxRetryError struct {
	// MaxRetries is the configured retry limit. Zero means retries were disabled.
	MaxRetries uint
	// LastErr is the error from the final attempt.
	LastErr error
}

MaxRetryError is returned when the maximum number of retries has been exceeded.

func (*MaxRetryError) Error 

func (e *MaxRetryError) Error() string

Error describes the exhausted retry limit and the last error, if present.

func (*MaxRetryError) GRPCStatus 

func (e *MaxRetryError) GRPCStatus() *status.Status

GRPCStatus returns the last error's gRPC status, or nil if it has none.

func (*MaxRetryError) Unwrap 

func (e *MaxRetryError) Unwrap() error

Unwrap returns LastErr.

type NodeBalancer 

type NodeBalancer interface {
	// Run starts the periodic balancing loop and blocks until ctx is canceled.
	Run(ctx context.Context, pool *Pool)
	// Rebalance performs one balancing operation.
	Rebalance(ctx context.Context, pool *Pool)
}

NodeBalancer balances pooled connections across CockroachDB nodes.

func NewNodeConnectionBalancer 

func NewNodeConnectionBalancer(opts ...NodeConnectionBalancerOption) NodeBalancer

NewNodeConnectionBalancer creates a connection balancer that distributes connections across CockroachDB nodes.

The balancer ensures an even distribution of connections across healthy nodes, pruning excess connections from over-utilized nodes and marking unhealthy node connections for garbage collection.

Pass the returned balancer to WithBalancer to run it with a Pool, or call NodeBalancer.Run yourself if you need to manage its lifecycle directly.

If no interval is specified via WithBalancingInterval, the default is 30 seconds.

type NodeConnectionBalancerOption 

type NodeConnectionBalancerOption func(*nodeConnectionBalancer)

NodeConnectionBalancerOption is a configuration option for the connection balancer.

func WithBalancingInterval 

func WithBalancingInterval(interval time.Duration) NodeConnectionBalancerOption

WithBalancingInterval sets how often to perform connection rebalancing. Default: 30 seconds.

func WithBalancingTimeout 

func WithBalancingTimeout(timeout time.Duration) NodeConnectionBalancerOption

WithBalancingTimeout sets the timeout for an individual balancing operation. Default: 10 seconds.

func WithTopologyChangeBalancingInterval 

func WithTopologyChangeBalancingInterval(interval, window time.Duration) NodeConnectionBalancerOption

WithTopologyChangeBalancingInterval configures a temporary faster balancing interval used after the tracker detects a topology change.

The faster interval is active for window. Defaults: 5 seconds for 5 minutes.

type NodeHealthTracker 

type NodeHealthTracker interface {
	// SetStatus records a health observation. Implementations may apply a failure
	// threshold before marking a node unhealthy.
	SetStatus(nodeID uint32, healthy bool)
	// IsHealthy reports whether the node is currently considered healthy.
	IsHealthy(nodeID uint32) bool
	// NumHealthy returns the number of nodes currently considered healthy.
	NumHealthy() int
}

NodeHealthTracker tracks the health of nodes in a CockroachDB cluster. Implementations must be safe for concurrent use by pools and balancers.

type NodeHealthTrackerOption 

type NodeHealthTrackerOption func(*nodeHealthTrackerConfig)

NodeHealthTrackerOption configures a tracker created by NewNodeHealthTracker.

func WithHealthTrackerHooks added in v1.1.0 

func WithHealthTrackerHooks(hooks ...Hook) NodeHealthTrackerOption

WithHealthTrackerHooks configures instrumentation on a caller-owned health tracker, including one shared by multiple pools. Hooks are appended in registration order. Nil entries are ignored. Pool hooks do not replace them.

func WithPollInterval 

func WithPollInterval(d time.Duration) NodeHealthTrackerOption

WithPollInterval configures the interval at which the node health tracker attempts to connect to a random node in the cluster and assert its health.

func WithTopologyChangePollInterval 

func WithTopologyChangePollInterval(interval, window time.Duration) NodeHealthTrackerOption

WithTopologyChangePollInterval configures a temporary faster polling interval used after the tracker detects a topology change.

The faster interval is active for window. Defaults: 5 seconds for 5 minutes.

type NodeSwitchEndData added in v1.1.0 

type NodeSwitchEndData struct {
	// PoolName is the name returned by [Pool.Name].
	PoolName string
	// FromNodeID is the node being avoided, or zero if its identity is unknown.
	FromNodeID uint32
	// ToNodeID is the replacement node, or zero if acquisition failed or its
	// identity is unknown.
	ToNodeID uint32
	// Reused reports that acquisition allowed the original node because at most
	// one healthy node was known. The replacement may still be a different node.
	Reused bool
	// Err is the connection acquisition error, or nil on success.
	Err error
}

NodeSwitchEndData describes the result of acquiring a replacement connection.

type NodeSwitchStartData added in v1.1.0 

type NodeSwitchStartData struct {
	// PoolName is the name returned by [Pool.Name].
	PoolName string
	// FromNodeID is the node being avoided, or zero if its identity is unknown.
	FromNodeID uint32
}

NodeSwitchStartData describes an attempt to acquire a replacement connection.

type OperationEndData added in v1.1.0 

type OperationEndData struct {
	// PoolName is the name returned by [Pool.Name].
	PoolName string
	// Attempts counts calls to the operation, excluding connection acquisition.
	// It is zero if the initial connection acquisition failed.
	Attempts int
	// Retries counts attempts after the first.
	Retries int
	// Resets counts connections discarded following resettable errors or closure.
	Resets int
	// RetryLimitReached reports whether the operation exhausted its retry limit.
	RetryLimitReached bool
	// Err is the final operation or connection acquisition error, or nil on success.
	Err error
}

OperationEndData describes the result of a retry-managed operation. CopyFrom, SendBatch, and operations on directly acquired connections are excluded.

type Option 

type Option func(*poolCfg)

Option is a configuration option for a Pool.

func WithBalancer 

func WithBalancer(b NodeBalancer) Option

WithBalancer configures the node connection balancer used by the Pool.

The pool starts the balancer after the underlying pgx pool is created and stops it when Pool.Close is called. If no balancer is configured, connection rebalancing is disabled.

func WithConnectRate 

func WithConnectRate(rate time.Duration) Option

WithConnectRate configures the rate limit for establishing new connections to the database. This helps prevent connection storms. The default is DefaultConnectRate.

func WithContextPropagationBehaviour 

func WithContextPropagationBehaviour(behavior ContextPropagationBehaviour) Option

WithContextPropagationBehaviour configures how contexts are propagated from callers to database operations. See ContextPropagationBehaviour for available options. The default is DefaultContextPropagationBehaviour.

func WithHealthTracker 

func WithHealthTracker(h NodeHealthTracker) Option

WithHealthTracker configures the health tracker used by the Pool.

You might want to initialize and pass the health tracker to the pool yourself if you use multiple pools for the same database inside one application in which case the pools can share the health tracker.

If no custom health tracker is provided a default per-pool one is used.

func WithHooks added in v1.1.0 

func WithHooks(hooks ...Hook) Option

WithHooks configures optional instrumentation for the pool, its built-in balancer, and its automatically created health tracker. Hooks are appended in registration order, including across multiple WithHooks calls. Nil entries are ignored. A custom health tracker keeps its own instrumentation; configure a shared built-in tracker with WithHealthTrackerHooks. Query callbacks run in registration order, followed by any existing pgx tracer.

func WithLogger 

func WithLogger(l *slog.Logger) Option

WithLogger configures the logger for the Pool. A nil logger uses slog.Default, which is also the default.

func WithMaxRetries 

func WithMaxRetries(retries int) Option

WithMaxRetries configures the maximum number of retries per database operation when encountering retryable errors. The default is DefaultMaxRetries.

func WithName 

func WithName(name string) Option

WithName configures the pool name used for observability (metrics and tracing). The default if not specified is "crdbpgx".

type Pool 

type Pool struct {
	// Runner provides typed queries and transactions backed by this pool.
	Runner
	// contains filtered or unexported fields
}

Pool is a CockroachDB-aware connection pool that wraps pgxpool.Pool with automatic retry logic, node health tracking, and optional connection rebalancing capabilities.

Pool provides the same interface as pgxpool.Pool but adds resilience features specifically designed for CockroachDB clusters, including automatic handling of retryable and resettable errors, and optional connection management across nodes.

func NewPool 

func NewPool(
	ctx context.Context,
	pgxConfig *pgxpool.Config,
	opts ...Option,
) (_ *Pool, err error)

NewPool creates a new CockroachDB-aware connection pool.

The pool wraps pgxpool.Pool and adds CockroachDB-specific features including automatic retry logic, node health tracking, and optional connection rebalancing.

NewPool copies pgxConfig before applying options. ctx controls initialization; the pool stops its background work when Pool.Close is called. Use WithHealthTracker to share a caller-owned health tracker across pools. NewPool returns an error if the configuration is invalid or pool creation fails.

Example 
package main

import (
	"context"
	"log"
	"os"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"

	"github.com/sumup/crdbpgx"
)

func main() {
	cfg, err := pgxpool.ParseConfig(os.Getenv("DATABASE_URL"))
	if err != nil {
		log.Fatal(err)
	}

	pool, err := crdbpgx.NewPool(
		context.Background(),
		cfg,
		crdbpgx.WithName("orders"),
		crdbpgx.WithContextPropagationBehaviour(crdbpgx.ContextPropagationOnAcquire),
		crdbpgx.WithBalancer(crdbpgx.NewNodeConnectionBalancer(
			crdbpgx.WithBalancingInterval(30*time.Second),
			crdbpgx.WithBalancingTimeout(10*time.Second),
		)),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer pool.Close()
}

func (*Pool) Acquire 

func (p *Pool) Acquire(ctx context.Context) (*pgxpool.Conn, error)

Acquire returns a connection from the pool. The connection must be released back to the pool by calling pgxpool.Conn.Release on the returned connection.

func (*Pool) AcquireAllIdle 

func (p *Pool) AcquireAllIdle(ctx context.Context) []*pgxpool.Conn

AcquireAllIdle atomically acquires all currently idle connections from the pool. This is primarily used by the node connection balancer.

func (*Pool) AcquireFunc 

func (p *Pool) AcquireFunc(ctx context.Context, f func(*pgxpool.Conn) error) error

AcquireFunc acquires a connection, calls the provided function with it, and releases the connection when the function returns.

func (*Pool) Close 

func (p *Pool) Close()

Close closes all connections in the pool and prevents new connections from being acquired.

func (*Pool) Config 

func (p *Pool) Config() *pgxpool.Config

Config returns a copy of the pool's configuration.

func (*Pool) CopyFrom 

func (p *Pool) CopyFrom(ctx context.Context, tableName pgx.Identifier, columnNames []string, rowSrc pgx.CopyFromSource) (int64, error)

CopyFrom efficiently copies data from rowSrc to the specified table using the PostgreSQL COPY protocol. Note that this operation does not have automatic retry logic.

func (*Pool) ExecFunc 

func (p *Pool) ExecFunc(ctx context.Context, tagFunc func(ctx context.Context, tag pgconn.CommandTag, err error) error, sql string, arguments ...any) error

ExecFunc executes a SQL command and calls tagFunc with the command result. The operation includes automatic retry logic for retryable and resettable errors.

func (*Pool) GCMark 

func (p *Pool) GCMark(conn *pgx.Conn)

GCMark marks a connection to be destroyed on the next pool operation (acquire/release). This is used internally for connection rebalancing and when recovering from node failures.

func (*Pool) InTxFunc 

func (p *Pool) InTxFunc(ctx context.Context, txOptions pgx.TxOptions, txFunc func(Tx) error) error

InTxFunc executes a callback with the underlying transaction. It retries the entire callback on retryable errors and resets connections on resettable errors. Use Runner.InTx for a callback with typed query methods.

func (*Pool) MaxConns 

func (p *Pool) MaxConns() int

MaxConns returns the maximum number of connections in the underlying pool.

func (*Pool) MinConns 

func (p *Pool) MinConns() int

MinConns returns the minimum number of connections in the underlying pool.

func (*Pool) Name 

func (p *Pool) Name() string

Name returns the configured name of the pool, used for observability purposes.

func (*Pool) NodeID 

func (p *Pool) NodeID(conn *pgx.Conn) uint32

NodeID returns the CockroachDB node ID for the given connection. Returns 0 if the connection is not tracked.

func (*Pool) Ping 

func (p *Pool) Ping(ctx context.Context) error

Ping verifies that a connection to the database is still alive.

func (*Pool) QueryFunc 

func (p *Pool) QueryFunc(ctx context.Context, rowsFunc func(ctx context.Context, rows pgx.Rows) error, sql string, optionsAndArgs ...any) error

QueryFunc executes a SQL query and calls rowsFunc with the resulting rows. The operation includes automatic retry logic for retryable and resettable errors. rowsFunc must close the rows when done.

func (*Pool) QueryRowFunc 

func (p *Pool) QueryRowFunc(ctx context.Context, rowFunc func(ctx context.Context, row pgx.Row) error, sql string, optionsAndArgs ...any) error

QueryRowFunc executes a SQL query that returns at most one row and calls rowFunc with the result. The operation includes automatic retry logic for retryable and resettable errors.

func (*Pool) SendBatch 

func (p *Pool) SendBatch(ctx context.Context, b *pgx.Batch) pgx.BatchResults

SendBatch sends a batch of queries to the database for execution. Note that batch operations do not have automatic retry logic.

func (*Pool) Stat 

func (p *Pool) Stat() *pgxpool.Stat

Stat returns statistics about the connection pool.

type RebalanceEndData added in v1.1.0 

type RebalanceEndData struct {
	// PoolName is the name returned by [Pool.Name].
	PoolName string
	// Acquired counts idle connections acquired for this balancing run.
	Acquired int
	// Pruned counts healthy connections marked for pruning to balance the pool.
	// It excludes connections marked because their nodes are unhealthy.
	Pruned int
}

RebalanceEndData describes one completed built-in balancing run.

type ResettableError 

type ResettableError struct {
	// Err is the underlying error that requires replacing the connection.
	Err error
}

ResettableError wraps an error that requires connection reset and retry on a different node.

func (*ResettableError) Error 

func (e *ResettableError) Error() string

Error describes the resettable error and its underlying cause, if present.

func (*ResettableError) GRPCStatus 

func (e *ResettableError) GRPCStatus() *status.Status

GRPCStatus returns an Unavailable status describing the resettable error.

func (*ResettableError) Unwrap 

func (e *ResettableError) Unwrap() error

Unwrap returns Err.

type RetryableError 

type RetryableError struct {
	// Err is the underlying error that permits retrying on the same connection.
	Err error
}

RetryableError wraps an error that can be retried on the same connection.

func (*RetryableError) Error 

func (e *RetryableError) Error() string

Error describes the retryable error and its underlying cause, if present.

func (*RetryableError) Unwrap 

func (e *RetryableError) Unwrap() error

Unwrap returns Err.

type Runner 

type Runner struct {
	Executor
}

Runner executes typed queries and transactions through an Executor. Pool embeds Runner, so its methods can be called directly on the pool. A transaction callback receives another Runner bound to that transaction.

func NewRunner added in v1.1.0 

func NewRunner(db Executor) Runner

NewRunner binds typed operations to db. It does not own or close db. Pools initialize their embedded Runner automatically.

func (Runner) Exec added in v1.1.0 

func (r Runner) Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)

Exec executes a SQL command and returns its command tag. A pool retries retryable errors. Transaction errors propagate to the outer transaction retry loop. On error Exec returns an empty command tag.

func (Runner) InTx added in v1.1.0 

func (r Runner) InTx(ctx context.Context, opts pgx.TxOptions, fn func(Runner) error) error

InTx runs fn with a Runner bound to a transaction and commits on success. On retryable errors a pool retries the whole transaction, including fn. When already inside a transaction, InTx creates a savepoint and propagates errors to the outer retry loop. Nested transactions ignore opts and inherit the ambient isolation level. fn must not retain the transaction Runner.

func (Runner) Query added in v1.1.0 

func (r Runner) Query[T any](ctx context.Context, rowTo pgx.RowToFunc[T], sql string, args ...any) ([]T, error)

Query executes a query and collects its rows with rowTo. It closes the rows before returning and returns nil on error. A pool retries the entire query, including collection, on retryable errors. Within a transaction, errors propagate to the enclosing transaction's retry loop. rowTo may be called again on retry and should avoid external side effects.

Example

The same helper accepts a pool or a transaction through Runner. 

package main

import (
	"context"

	"github.com/jackc/pgx/v5"

	"github.com/sumup/crdbpgx"
)

func main() {
	listIDs := func(ctx context.Context, db crdbpgx.Runner) ([]int64, error) {
		return db.Query(ctx, pgx.RowTo[int64],
			"SELECT id FROM users WHERE active = @active", pgx.NamedArgs{"active": true})
	}
	_ = listIDs
}

func (Runner) QueryRow added in v1.1.0 

func (r Runner) QueryRow[T any](ctx context.Context, rowTo pgx.RowToFunc[T], sql string, args ...any) (T, error)

QueryRow executes a query and collects its first row with rowTo. It closes the rows before returning, ignores additional rows, and returns pgx.ErrNoRows if the query returns no rows. On error it returns the zero value of T. It has the same retry behavior and rowTo requirements as Query.

Example 
package main

import (
	"context"

	"github.com/jackc/pgx/v5"

	"github.com/sumup/crdbpgx"
)

func main() {
	type User struct {
		ID   int64  `db:"id"`
		Name string `db:"name"`
	}
	getUser := func(ctx context.Context, db crdbpgx.Runner, id int64) (User, error) {
		return db.QueryRow(ctx, pgx.RowToStructByName[User],
			"SELECT id, name FROM users WHERE id = $1", id)
	}
	_ = getUser
}

type Tx 

type Tx interface {
	pgx.Tx
	Executor
}

Tx represents a CockroachDB database transaction.

Tx can be used in place of pgx.Tx and provides additional methods that are also defined on Pool such that users can use the same set of functions both outside and inside a transaction (ExecFunc, QueryFunc, QueryRowFunc, InTxFunc).

type UniformJitter 

type UniformJitter struct {
	// Source supplies randomness. A nil Source uses the default random source.
	// A non-nil Source must not be shared concurrently without synchronization.
	Source *rand.Rand
	// Min is the inclusive lower bound of the jittered duration.
	Min time.Duration
}

UniformJitter implements uniform jitter distribution.

The jittered duration is uniformly distributed between Min and the provided duration. This helps distribute load more evenly and prevents synchronized retries across clients.

func (UniformJitter) Jitter 

func (u UniformJitter) Jitter(d time.Duration) time.Duration

Jitter returns a duration uniformly distributed over the range [Min, d). The zero-value is ready to use with the default random source. Panics if d <= Min.

Source Files

View all Source files

Directories

Path Synopsis
crdbotel module
crdbprom module

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.