postgres

Documentation

Overview

Package postgres provides a durable, multi-process registry.Registry backed by PostgreSQL. Unlike the in-memory registry, allocations are persisted, so a sequence block is never handed out twice even across process restarts, and uniqueness holds for every node sharing the same database and table.

The registry talks to Postgres through the pgx driver. Construct it with a *pgxpool.Pool (or any pgx connection or transaction that satisfies DB):

pool, err := pgxpool.New(ctx, dsn)
if err != nil {
	log.Fatal(err)
}
defer pool.Close()

reg, err := postgres.New(pool)
if err != nil {
	log.Fatal(err)
}
// Create the table once at startup (or manage it with migrations instead).
if err := reg.EnsureSchema(ctx); err != nil {
	log.Fatal(err)
}

node, err := crid.New(reg)

Each Allocate is a single atomic UPSERT, so concurrent callers - within one process or across many - always receive non-overlapping blocks for a given timestamp.

Index

• Variables
• type DB
• type Option
  • func WithTable(name string) Option
• type Registry
  • func New(db DB, opts ...Option) (*Registry, error)
  • func (r *Registry) Allocate(ctx context.Context, timestamp, blockSize int64) (int64, error)
  • func (r *Registry) EnsureSchema(ctx context.Context) error
  • func (r *Registry) EvictBefore(ctx context.Context, cutoff int64) error
  • func (r *Registry) VerifySchema(ctx context.Context) (bool, error)

Variables

var (
	// ErrNilDB is returned when New is called with a nil DB.
	ErrNilDB = errors.New("db cannot be nil")

	// ErrInvalidTable is returned when New is called with a table name that is not a
	// valid SQL identifier (optionally schema-qualified).
	ErrInvalidTable = errors.New("table must be a valid SQL identifier, optionally schema-qualified")

	// ErrAllocate is returned when the registry fails to reserve a block from Postgres.
	ErrAllocate = errors.New("allocate block failed")

	// ErrEnsureSchema is returned when EnsureSchema fails to create the table.
	ErrEnsureSchema = errors.New("ensure schema failed")

	// ErrVerifySchema is returned when VerifySchema fails to check whether the table exists.
	ErrVerifySchema = errors.New("verify schema failed")

	// ErrEvict is returned when EvictBefore fails to delete rows from Postgres.
	ErrEvict = errors.New("evict failed")
)

Types

type DB

type DB interface {
	QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
	Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
}

DB is the subset of the pgx API the registry depends on. It is satisfied by *pgxpool.Pool, *pgx.Conn, and pgx.Tx, so the registry can run on a connection pool, a single connection, or inside a caller-managed transaction.

type Option

type Option func(*Registry)

Option configures a Registry at creation time.

func WithTable

func WithTable(name string) Option

WithTable sets the table the registry reads and writes. The name must be a valid SQL identifier, optionally schema-qualified (for example "app.crid_allocations"). The default is "crid_allocations".

type Registry

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

Registry is a registry.Registry backed by Postgres. It is safe for concurrent use by multiple goroutines and across multiple processes sharing the same database and table.

func New

func New(db DB, opts ...Option) (*Registry, error)

New returns a Registry that allocates sequence blocks from db, which is typically a *pgxpool.Pool. The returned Registry does not create its table; call EnsureSchema once at startup, or create the table out of band (see EnsureSchema for the schema).

func (*Registry) Allocate

func (r *Registry) Allocate(ctx context.Context, timestamp, blockSize int64) (int64, error)

Allocate reserves blockSize sequence numbers for timestamp and returns the starting value of the reserved block. It runs a single atomic UPSERT, so successive and concurrent calls for the same timestamp - across every process sharing the table - return non-overlapping blocks.

func (*Registry) EnsureSchema

func (r *Registry) EnsureSchema(ctx context.Context) error

EnsureSchema creates the registry's table if it does not already exist. It is a convenience for development and simple deployments. Production schemas are usually managed by migrations.

func (*Registry) EvictBefore

func (r *Registry) EvictBefore(ctx context.Context, cutoff int64) error

EvictBefore deletes all allocation rows for timestamps strictly before cutoff. cutoff is seconds since the Unix epoch, the same unit as the ts column. Rows for timestamps still in use must not be evicted; only call this with a cutoff safely in the past (for example, one hour ago). It is safe to call concurrently with Allocate.

func (*Registry) VerifySchema

func (r *Registry) VerifySchema(ctx context.Context) (bool, error)

VerifySchema reports whether the registry's table already exists. Use it to fail fast at startup when the schema is managed out of band (by migrations) rather than created with EnsureSchema. The table name is resolved against the connection's search_path the same way Allocate resolves it, so a bare name and a schema-qualified name behave consistently across both calls.