xpg

package module
v0.4.0 Latest Latest
Warning

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

 Go to latest Published: Aug 27, 2026 License: MIT Imports: 15 Imported by: 0

README

Postgres toolkit for Go

Lightweight PostgreSQL wrapper for Go, built on top of pgx.

Go Reference Test Coverage

xpg builds on the pgx client with a compact API for common PostgreSQL infrastructure patterns. It adds support for pool lifecycle, transactions and savepoints, PostgreSQL error classification, advisory locks, primary/replica routing, application-level sharding, and observability.

The library uses pgx types and query model directly while keeping its core behavior and reducing boilerplate around connection management, routing, and common production workflows.

Features

  • Pool Lifecycle Management: Thin pool management on top of pgx with direct access to the underlying PostgreSQL client.
  • Transactions and Savepoints: Managed transactions, savepoints, and helpers for common multi-step transactional workflows.
  • Error Classification: Classification of PostgreSQL constraint, transaction, cancellation, connection, and other common database errors.
  • Advisory Locking: Transaction-level advisory locks for coordinating concurrent database operations.
  • Primary/Replica Routing: Logical cluster topologies with explicit read policies, replica selection, primary fallback, and read-only transactions across PostgreSQL nodes.
  • Application-Level Sharding: Rendezvous, range, time-based, and custom routing with colocation checks, key grouping, and bounded parallel operations across shards.
  • Observability: Structured logging, tracing, pool statistics, and optional OpenTelemetry metrics.

Installation

This repository contains the core xpg module. The core module is released from the repository root:

go get github.com/mkbeh/xpg

Optional integrations are released independently under extra:

go get github.com/mkbeh/xpg/extra/otelxpg

Usage

Open an xpg pool and execute a PostgreSQL query:

// urlExample := "postgres://username:password@localhost:5432/database_name"
pool, err := xpg.Open(
	context.Background(),
	os.Getenv("DATABASE_URL"),
	xpg.WithName("example-pool"),
)
if err != nil {
	log.Fatalf("failed to open pool: %v", err)
}
defer pool.Close()

var message string
err = pool.QueryRow(context.Background(), "SELECT 'hello from xpg'").Scan(&message)
if err != nil {
	log.Fatalf("query failed: %v", err)
}

fmt.Println(message) // Outputs: hello from xpg

xpg provides managed transactions using the native pgx transaction API. Returning nil commits the transaction; returning an error rolls it back.

err := pool.InTx(ctx, pgx.TxOptions{}, func(ctx context.Context, tx pgx.Tx) error {
	_, err := tx.Exec(ctx, "UPDATE users SET active = true WHERE id = $1", userID)
	return err
})

Savepoints can isolate optional work without aborting the outer transaction.

Transaction-level PostgreSQL advisory locks can coordinate concurrent work across application instances using the same database. The lock is held for the lifetime of the transaction and released automatically on commit or rollback.

err := pool.InTx(ctx, pgx.TxOptions{}, func(ctx context.Context, tx pgx.Tx) error {
	if err := xpg.AdvisoryXactLock(ctx, tx, lockID); err != nil {
		return err
	}

	_, err := tx.Exec(ctx, "UPDATE jobs SET status = 'running' WHERE id = $1", jobID)
	return err
})

For error handling, xpg provides semantic helpers for classifying PostgreSQL failures and inspecting SQLSTATE codes without manual string matching.

_, err := pool.Exec(ctx, "INSERT INTO users (id, email) VALUES ($1, $2)", userID, email)

switch {
case xpg.IsUniqueViolation(err):
    // Handle duplicate data.
case xpg.IsRetryableTransaction(err):
    // Retry the transaction when the operation is safe to replay.
case err != nil:
    return err
}

The underlying SQLSTATE code is available through xpg.SQLState(err). Helpers cover constraint violations, serialization failures, deadlocks, lock errors, query cancellation, and connection failures.

Clustering

The topology/cluster package groups primary and replica pools into a logical cluster with explicit read routing.

wallets, err := cluster.New(cluster.Config{
	ID:       "wallets",
	Primary:  primary,
	Replicas: []*xpg.Pool{replicaA, replicaB},
})
if err != nil {
	panic(err)
}
defer wallets.Close()

// Route writes explicitly to the primary.
primaryPool := wallets.Primary()

_, err = primaryPool.Exec(ctx, "UPDATE wallets SET frozen = true WHERE id = $1", walletID)
if err != nil {
	panic(err)
}

// Route reads according to the selected policy.
readPool, err := wallets.ReadPool(ctx, cluster.ReadReplicaPreferred)
if err != nil {
	panic(err)
}

var frozen bool
err = readPool.QueryRow(ctx, "SELECT frozen FROM wallets WHERE id = $1", walletID).Scan(&frozen)
if err != nil {
	panic(err)
}
Cluster Transactions

Cluster transactions combine explicit primary/replica routing with the native pgx transaction API.

Primary Transactions

InPrimaryTx runs the transaction on the cluster primary and is intended for atomic multi-step writes.

err := wallets.InPrimaryTx(
    ctx,
    pgx.TxOptions{},
    func(ctx context.Context, tx pgx.Tx) error {
        const debitQuery = `
            UPDATE wallets
            SET balance = balance - $1
            WHERE id = $2 AND balance >= $1
        `

        result, err := tx.Exec(ctx, debitQuery, amount, fromID)
        if err != nil {
            return err
        }

        if result.RowsAffected() == 0 {
            return errors.New("insufficient funds or wallet not found")
        }

        const creditQuery = `
            UPDATE wallets
            SET balance = balance + $1
            WHERE id = $2
        `

        _, err = tx.Exec(ctx, creditQuery, amount, toID)

        return err
    },
)

The transaction commits on nil and rolls back on error.

Read Transactions

InReadTx routes the transaction according to the selected read policy and enforces PostgreSQL read-only mode. Use it for read workloads that can run on replicas.

var (
    totalWallets int64
    totalBalance int64
)

err := wallets.InReadTx(
    ctx,
    cluster.ReadReplicaPreferred,
    cluster.ReadTxOptions{
        IsoLevel: pgx.RepeatableRead,
    },
    func(ctx context.Context, tx pgx.Tx) error {
        const query = `
            SELECT count(*), coalesce(sum(balance), 0)
            FROM wallets
            WHERE created_at > $1
        `

        return tx.QueryRow(ctx, query, since).Scan(
            &totalWallets,
            &totalBalance,
        )
    },
)
Read Routing Policies

Read policies control how reads and read-only transactions are routed across the cluster:

Policy Primary Fallback Behavior
ReadPrimary — Always routes reads to the primary.
ReadReplicaRequired No Requires a replica and returns ErrNoReplica when none can be selected.
ReadReplicaPreferred Yes Prefers a replica and falls back to the primary only when no replica can be selected.

Replica selection uses round-robin by default and can be customized by implementing ReplicaSelector.

Sharding

The topology/shard package provides application-level sharding with explicit key routing across an immutable shard topology. Routing strategies live under topology/shard/resolver.

topology, err := shard.NewTopology(clusterA, clusterB)
if err != nil {
    panic(err)
}
defer topology.Close()

// Route user IDs using rendezvous hashing.
userResolver, err := resolver.NewRendezvous(topology, "users", resolver.Uint64KeyEncoder())
if err != nil {
    panic(err)
}

// Resolve the target shard.
targetShard, err := userResolver.Resolve(userID)
if err != nil {
    panic(err)
}

// Write to the shard primary.
primaryPool := targetShard.Primary()

_, err = primaryPool.Exec(ctx, "UPDATE users SET active = true WHERE id = $1", userID)
if err != nil {
    panic(err)
}

// Read from the same shard using the selected read policy.
readPool, err := targetShard.ReadPool(ctx, cluster.ReadReplicaPreferred)
if err != nil {
    panic(err)
}

var active bool
err = readPool.QueryRow(ctx, "SELECT active FROM users WHERE id = $1", userID).Scan(&active)
if err != nil {
    panic(err)
}

Resolvers support multiple placement strategies, while shard utilities provide colocation checks, strict grouping, tolerant partitioning, and bounded parallel operations across shards.

Routing Strategies

Resolvers bind a data-placement strategy to an immutable shard topology. Every resolver exposes the same routing contract, allowing application code to resolve keys independently of the selected strategy.

Resolver Best Suited For Routing Model
RendezvousResolver Keys without natural ranges Deterministic Highest Random Weight (HRW) hashing within a namespace.
RangeResolver Ordered numeric or string keys Bounded, non-overlapping half-open intervals [Start, End).
TimeRangeResolver Time-series or partitioned event data Bounded chronological intervals normalized to UTC.
CustomResolver Domain-specific placement rules Application-defined mapping from a key to shard.ID. 
// Rendezvous hashing distributes arbitrary keys deterministically across the topology.
usersByHash, _ := resolver.NewRendezvous(topology, "users", resolver.Uint64KeyEncoder())

// Ordered ranges provide explicit control over the keyspace.
usersByRange, _ := resolver.NewRange(topology, []resolver.Range[uint64]{
    {Start: 0,   End: 100, ShardID: "shard-a"},
    {Start: 100, End: 200, ShardID: "shard-b"},
})

// Time ranges route records through bounded chronological intervals.
t0, _ := time.Parse(time.RFC3339, "2026-01-01T00:00:00Z")
t1 := t0.AddDate(0, 1, 0)
t2 := t0.AddDate(0, 2, 0)

eventsByTime, _ := resolver.NewTimeRange(topology, []resolver.TimeRange{
    {Start: t0, End: t1, ShardID: "shard-a"},
    {Start: t1, End: t2, ShardID: "shard-b"},
})

// Custom routing keeps domain-specific placement rules in application code.
tenantsByRegion, _ := resolver.NewCustom(topology, func(region string) (shard.ID, error) {
    switch region {
    case "eu":
        return "shard-a", nil
    case "us":
        return "shard-b", nil
    default:
        return "", shard.ErrNoShard
    }
})

Regardless of the selected strategy, routing uses the same Resolve contract:

targetShard, err := usersByHash.Resolve(userID)
if err != nil {
    panic(err)
}

log.Printf("resolved shard: %s", targetShard.ID())

Range and time-range resolvers may contain intentional gaps in the configured keyspace; keys that do not match any range return ErrNoShard. Custom resolvers can return the same error when a domain key has no valid destination.

[!IMPORTANT] For rendezvous routing, the namespace, key encoding, and stable shard IDs are part of the placement contract.

Multi-Key Routing

For complex batch operations, xpg provides routing primitives to analyze, group, and partition multi-key workloads across a shard topology.

// Add range-based routing over the same shard topology.
rangeResolver, _ := resolver.NewRange(topology, []resolver.Range[uint64]{
    {Start: 0,   End: 100, ShardID: "shard-a"},
    {Start: 100, End: 200, ShardID: "shard-b"},
})
Strict Colocation Checks

Use SameShard to guarantee that a set of keys resolves to the same shard before executing a shard-local transaction or another operation that must remain colocated.

// Verify that all keys resolve to the same shard.
targetShard, err := shard.SameShard(rangeResolver, 42, 43)
if err != nil {
    panic(err)
}

// Use the resolved shard for a shard-local operation.
log.Printf("resolved shard: %s", targetShard.ID())
Strict Batch Grouping

GroupByShard groups a slice of keys by destination shard. It uses strict routing semantics and fails if any key cannot be resolved.

keys := []uint64{42, 142, 43, 143}

groups, err := shard.GroupByShard(rangeResolver, keys)
if err != nil {
    panic(err) // Fails if any key cannot be resolved.
}

for _, group := range groups {
    // Execute one shard-local batch update for each resolved group.
    _ = group.Shard.InPrimaryTx(ctx, pgx.TxOptions{}, func(ctx context.Context, tx pgx.Tx) error {
        const query = `
            UPDATE users
            SET active = true
            WHERE id = ANY($1)
        `

        _, err := tx.Exec(ctx, query, group.Keys)

        return err
    })
}
Tolerant Partitioning

PartitionByShard provides a relaxed alternative to strict grouping. Routable keys are grouped by shard, while keys that do not resolve to any shard are collected separately.

keys := []uint64{42, 142, 250, 43, 143} // 250 falls outside the configured ranges.

partition, err := shard.PartitionByShard(rangeResolver, keys)
if err != nil {
    panic(err) // Resolver errors other than ErrNoShard still abort the operation.
}

// Process all routable groups.
for _, group := range partition.Groups {
    log.Printf("process shard=%s user_ids=%v", group.Shard.ID(), group.Keys)
}

// Handle unresolved keys separately.
if len(partition.Unresolved) != 0 {
    log.Printf("unresolved keys: %v", partition.Unresolved)
}
Parallel Fan-Out Operations

ForEachShard executes an operation across the entire topology with bounded concurrency. maxConcurrency controls how many shard callbacks may run at the same time; setting it to 1 makes execution sequential.

timeoutCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
defer cancel()

const maxConcurrency = 4

expiredBefore := time.Now()

results, err := topology.ForEachShard(
    timeoutCtx,
    maxConcurrency,
    func(ctx context.Context, s shard.Shard) error {
        primary := s.Primary()
        if primary == nil {
            return cluster.ErrNoPrimary
        }

        const query = `
            DELETE FROM sessions
            WHERE expired_at < $1
        `

        _, err := primary.Exec(ctx, query, expiredBefore)

        return err
    },
)
if err != nil {
    log.Printf("fan-out completed with errors: %v", err)
}

// Inspect individual shard failures when detailed handling is required.
for _, result := range results {
    if result.Err != nil {
        log.Printf("shard=%s failed: %v", result.ShardID, result.Err)
    }
}

Results preserve topology registration order and retain individual shard failures, while the returned error aggregates callback and context cancellation errors.

Examples

See the examples directory for runnable examples covering the main xpg usage patterns.

License

This project is licensed under the MIT License.

Documentation

Overview

Package xpg provides PostgreSQL infrastructure utilities built on pgx.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func AdvisoryXactLock 

func AdvisoryXactLock(ctx context.Context, tx pgx.Tx, key int64) error

AdvisoryXactLock acquires an exclusive transaction-level advisory lock.

The call blocks until the lock is acquired or ctx is canceled. PostgreSQL releases the lock automatically when tx is committed or rolled back.

func InSavepoint 

func InSavepoint(
	ctx context.Context,
	tx pgx.Tx,
	fn func(context.Context, pgx.Tx) error,
) error

InSavepoint executes fn within a PostgreSQL savepoint.

If fn returns nil, the savepoint is released; otherwise it is rolled back. If fn panics, rollback is attempted before the panic is propagated. The callback must not call Commit or Rollback; InSavepoint owns savepoint finalization.

The callback receives ctx unchanged and should observe its cancellation.

func IsCheckViolation 

func IsCheckViolation(err error) bool

IsCheckViolation reports whether err is a PostgreSQL check_violation.

func IsConnectionError 

func IsConnectionError(err error) bool

IsConnectionError reports whether err represents a PostgreSQL connection failure known to pgx or the Go networking stack.

func IsDeadlock 

func IsDeadlock(err error) bool

IsDeadlock reports whether err is a PostgreSQL deadlock_detected error.

func IsForeignKeyViolation 

func IsForeignKeyViolation(err error) bool

IsForeignKeyViolation reports whether err is a PostgreSQL foreign_key_violation.

func IsLockNotAvailable 

func IsLockNotAvailable(err error) bool

IsLockNotAvailable reports whether err is a PostgreSQL lock_not_available error.

func IsNoRows 

func IsNoRows(err error) bool

IsNoRows reports whether err indicates that a query returned no rows.

func IsNotNullViolation 

func IsNotNullViolation(err error) bool

IsNotNullViolation reports whether err is a PostgreSQL not_null_violation.

func IsQueryCanceled 

func IsQueryCanceled(err error) bool

IsQueryCanceled reports whether PostgreSQL canceled the query.

Client-side context cancellation remains available through errors.Is with context.Canceled or context.DeadlineExceeded.

func IsRetryableTransaction 

func IsRetryableTransaction(err error) bool

IsRetryableTransaction reports whether PostgreSQL aborted the transaction because of a serialization failure or a deadlock.

The entire transaction callback must still be safe to replay. Connection failures are deliberately not classified as transaction-retryable.

func IsSerializationFailure 

func IsSerializationFailure(err error) bool

IsSerializationFailure reports whether err is a PostgreSQL serialization_failure.

func IsUniqueViolation 

func IsUniqueViolation(err error) bool

IsUniqueViolation reports whether err is a PostgreSQL unique_violation.

func SQLState 

func SQLState(err error) string

SQLState returns the PostgreSQL SQLSTATE code carried by err. It returns an empty string when the error tree does not contain a pgconn.PgError.

func TryAdvisoryXactLock 

func TryAdvisoryXactLock(ctx context.Context, tx pgx.Tx, key int64) (bool, error)

TryAdvisoryXactLock attempts to acquire an exclusive transaction-level advisory lock without waiting.

PostgreSQL releases an acquired lock automatically when tx is committed or rolled back.

Types

type Metrics 

type Metrics interface {
	Register(pool *Pool) (MetricsRegistration, error)
}

Metrics registers metrics for a Pool.

Implementations must be safe to reuse across multiple pools. Register is called after the underlying pgxpool.Pool has been created.

type MetricsRegistration 

type MetricsRegistration interface {
	Close()
}

MetricsRegistration represents a metrics registration for one Pool.

Close is called once before the underlying pgxpool.Pool is closed.

type Option 

type Option interface {
	// contains filtered or unexported methods
}

Option configures a Pool.

The interface is sealed so options can only be created by this package.

func WithLabel 

func WithLabel(key, value string) Option

WithLabel adds or replaces one pool label.

func WithLabels 

func WithLabels(labels map[string]string) Option

WithLabels merges labels into the pool metadata.

Labels are defensively copied. When the same key is configured more than once, the last value wins.

func WithLogger 

func WithLogger(logger tracelog.Logger, level tracelog.LogLevel) Option

WithLogger attaches a pgx-compatible logger to the pool.

Logging uses pgx tracelog and participates in the same tracing pipeline as tracers configured through xpg. pgx tracelog may include SQL text and query arguments in log records; applications are responsible for choosing an appropriate level and handling sensitive values.

func WithMetrics 

func WithMetrics(metrics Metrics) Option

WithMetrics attaches one metrics implementation to the pool.

Metrics are registered when the pool is created and unregistered automatically when the Pool is closed.

func WithName 

func WithName(name string) Option

WithName assigns a stable logical name to the pool.

Name is metadata for diagnostics and observability. It does not change the PostgreSQL application_name runtime parameter.

func WithTracer 

func WithTracer(tracer pgx.QueryTracer) Option

WithTracer attaches one pgx query tracer to the pool.

The option may be specified multiple times. Configured loggers and tracers are combined through pgx multitracer. When xpg logging or tracing options are configured, the resulting tracing pipeline replaces any tracer already configured on the pgx connection config.

func WithTracers 

func WithTracers(tracers ...pgx.QueryTracer) Option

WithTracers attaches multiple pgx query tracers to the pool.

Configured loggers and tracers are invoked in configuration order and combined through pgx multitracer. When xpg logging or tracing options are configured, the resulting tracing pipeline replaces any tracer already configured on the pgx connection config.

type Pool 

type Pool struct {
	// contains filtered or unexported fields
}

Pool is a concurrency-safe PostgreSQL connection pool backed by pgxpool.

func New 

func New(ctx context.Context, config *pgxpool.Config, options ...Option) (*Pool, error)

New creates a Pool from config.

Config must have been created by pgxpool.ParseConfig. New passes a defensive copy to pgxpool, so subsequent changes to config do not affect the Pool.

As with pgxpool.Config.Copy, the referenced tls.Config remains shared and must not be modified after it has been used to create connections.

func Open 

func Open(ctx context.Context, connString string, options ...Option) (*Pool, error)

Open parses a PostgreSQL connection string and creates a Pool.

func (*Pool) Close 

func (p *Pool) Close()

Close closes the pool and waits for acquired connections to be returned. Close is safe to call multiple times.

func (*Pool) CopyFrom 

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

CopyFrom copies rows into the specified table.

func (*Pool) Exec 

func (p *Pool) Exec(ctx context.Context, sql string, arguments ...any) (pgconn.CommandTag, error)

Exec executes SQL against the pool.

func (*Pool) InTx 

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

InTx executes fn in a transaction configured by txOptions.

If fn returns nil, the transaction is committed; otherwise it is rolled back. If fn panics, rollback is attempted before the panic is propagated. The callback must not call Commit or Rollback; InTx owns transaction finalization.

The callback receives ctx unchanged. Context cancellation does not automatically finalize the transaction while fn is running; fn should observe ctx and return promptly.

func (*Pool) Labels 

func (p *Pool) Labels() map[string]string

Labels returns a copy of the pool labels.

func (*Pool) Name 

func (p *Pool) Name() string

Name returns the logical pool name.

If WithName is not configured, the name is derived from the connection host, port, and database.

func (*Pool) Ping 

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

Ping verifies connectivity to PostgreSQL.

func (*Pool) Query 

func (p *Pool) Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)

Query executes SQL and returns the resulting rows.

func (*Pool) QueryRow 

func (p *Pool) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row

QueryRow executes SQL that is expected to return at most one row.

func (*Pool) Raw 

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

Raw returns the underlying pgxpool.Pool.

The returned pool is owned by Pool and must not be closed directly.

func (*Pool) SendBatch 

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

SendBatch sends a batch of queries through the pool.

func (*Pool) Stats 

func (p *Pool) Stats() PoolStats

Stats returns a detached snapshot of the current pool statistics.

type PoolStats 

type PoolStats struct {

	// AcquiredConns is the number of connections currently checked out from the
	// pool.
	AcquiredConns int32

	// ConstructingConns is the number of connections currently being created.
	ConstructingConns int32

	// IdleConns is the number of currently idle connections.
	IdleConns int32

	// MaxConns is the maximum number of connections allowed by the pool.
	MaxConns int32

	// TotalConns is the number of acquired, idle, and constructing connections.
	TotalConns int32

	// AcquireCount is the cumulative number of successful connection acquires.
	AcquireCount int64

	// AcquireDuration is the cumulative duration of successful connection
	// acquires.
	AcquireDuration time.Duration

	// CanceledAcquireCount is the cumulative number of connection acquires
	// canceled by context cancellation.
	CanceledAcquireCount int64

	// EmptyAcquireCount is the cumulative number of successful acquires that
	// waited because the pool was empty.
	EmptyAcquireCount int64

	// EmptyAcquireWaitTime is the cumulative time spent waiting on successful
	// acquires while the pool was empty.
	EmptyAcquireWaitTime time.Duration

	// NewConnsCount is the cumulative number of connections created by the pool.
	NewConnsCount int64

	// MaxIdleDestroyCount is the cumulative number of connections closed because
	// they exceeded MaxConnIdleTime.
	MaxIdleDestroyCount int64

	// MaxLifetimeDestroyCount is the cumulative number of connections closed
	// because they exceeded MaxConnLifetime.
	MaxLifetimeDestroyCount int64
}

PoolStats is a detached point-in-time snapshot of connection pool statistics.

Counter fields are cumulative for the lifetime of the pool.

Source Files

View all Source files

Directories

Path Synopsis
extra
otelxpg module
slogxpg module
topology
cluster
Package cluster provides primary/replica routing for PostgreSQL connection pools.
 Package cluster provides primary/replica routing for PostgreSQL connection pools.
shard
Package shard provides application-level routing across PostgreSQL clusters.
 Package shard provides application-level routing across PostgreSQL clusters.
shard/resolver
Package resolver provides routing strategies for shard.Topology.
 Package resolver provides routing strategies for shard.Topology.

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.