postgres

package module
v0.2.1 Latest Latest
Warning

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

 Go to latest Published: Aug 9, 2026 License: Apache-2.0 Imports: 29 Imported by: 0

Documentation

Overview

Package postgres implements Warren's persistence, outbox and inbox ports against Postgres.

It provides a UnitOfWork whose Do commits aggregate state and the outbox rows for the events those aggregates raised in ONE transaction, resolves that transaction onto repositories through the context, ships a durable outbox store with LISTEN/NOTIFY wake-up, an advisory-lock leader elector, a durable inbox dedupe store, and a ping health check.

Repositories are plain SQL over a small driver-free query handle. There is no ORM, by design, and no pgx type in any signature here except Raw's.

It never runs a migration

Schema is a deploy step: see Schema and Migrate. Migrating from a lifecycle hook races every replica of a rolling deploy, applies DDL the still-serving old replicas were not written against, and turns one bad file into a simultaneous crash-loop. There is no option to enable it.

Index

Constants

View Source 
const ModuleName = "warren/persistence/postgres"

ModuleName is the name of the module Module returns — the scope name that appears in boot diagnostics.

Variables

View Source 
var ErrNoRows = pgx.ErrNoRows

ErrNoRows is returned by Row.Scan when the query matched nothing. A repository maps it to a semantic error rather than leaking it upward: 

if errors.Is(err, postgres.ErrNoRows) {
    return nil, errors.NotFound("user", id)
}
View Source 
var Schema fs.FS = mustSub(schemaFS, "schema")

Schema is the DDL for the tables this package owns — warren_outbox and warren_inbox — as numbered SQL files in goose's file format, so a project already running goose, atlas or dbmate applies them with one line.

Warren never applies them for you. See Migrate.

Functions

func Migrate 

func Migrate(ctx context.Context, dsn string, fsys fs.FS) error

Migrate applies every unapplied file in fsys, in name order, each in its own transaction, under a Postgres advisory lock so concurrent runners serialise. Applied versions are recorded in warren_schema_migrations.

It is called by your deploy job — and by `warren migrate` once the CLI grows that command. It is NEVER called from a lifecycle hook, and there is no option to make it one. Under a rolling deploy, migrating at boot races N replicas: N−1 block and get killed for missing their readiness deadline, the winner applies DDL that the still serving old replicas were not written against, and one bad file puts every replica into a crash loop at the same moment. Schema is a deploy step.

There are no down migrations. A migration that must be undone is a migration you write forwards, because that is the one you can test.

func Module 

func Module(opts ...Option) warren.Module

Module returns the warren.Module providing the connection pool, the UnitOfWork, the query-handle resolver repositories inject, and a ping health check.

The constructor PARSES: a malformed DSN or pool setting fails at wiring, before any hook runs, because pgxpool.ParseConfig does no I/O. The lifecycle hook CONNECTS: the pool is created and pinged in OnStart under ConnectTimeout, so an unreachable database rolls the boot back instead of leaving a live pool and open sockets behind a boot that failed later for an unrelated reason.

Because everything else here depends on the pool, its hook is appended first and reverse-order teardown closes it last — after the outbox relay's final flush.

It provides, into the graph: 

*UnitOfWork             the concrete unit of work, for OnCommit
persistence.UnitOfWork  the port handlers take
DB                      the resolver a repository injects

func RequireTx 

func RequireTx(ctx context.Context, op string) error

RequireTx returns a diagnostic unless a Postgres transaction is in scope on ctx. It is the aggregate-free half of the rule: Store.Append and any write that carries no root call it directly.

A repository Save or Delete does NOT call it. Those go through persistence.Write, which makes this same check AND enlists the aggregate — the two were separate statements until a field test deleted the second one and the row committed while the event evaporated.

Reads outside UnitOfWork.Do are fine and go to the pool — the persistence contract suite depends on it. Writes are not: the row would autocommit while persistence.Track stayed a no-op, so the aggregate's events would sit pending on an object about to go out of scope, and vanish. Silently, with no error anywhere. That is the exact loss the outbox exists to prevent, so it is refused rather than permitted quietly.

Types

type CacheMode 

type CacheMode string

CacheMode is how the driver caches prepared statements. 

const (
	// StatementCachePrepare is the default: named prepared statements. It is
	// the fastest, and it is incompatible with a pgbouncer in transaction
	// pooling mode, which does not keep a session for the statement to live in.
	StatementCachePrepare CacheMode = "prepare"
	// StatementCacheDescribe uses unnamed statements: slower, and what a
	// transaction-pooling pgbouncer requires.
	StatementCacheDescribe CacheMode = "describe"
)

type DB 

type DB func(ctx context.Context) Queryer

DB resolves the handle a repository uses for one call: the transaction ctx carries if UnitOfWork.Do put one there, and the pool otherwise.

It is the only piece of framework machinery in a repository, and it does one thing. Hold it in a field; never store the context: 

type UserRepository struct{ db postgres.DB }

row := r.db(ctx).QueryRow(ctx, `SELECT … WHERE id = $1`, id)

Both handles are boxed once — once per transaction, and once per process for the pool — so this allocates nothing per call.

type InboxOption 

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

InboxOption configures the inbox store.

func InboxRetention 

func InboxRetention(d time.Duration) InboxOption

InboxRetention is how long a seen key is remembered. The default is 24h: it must exceed the longest redelivery window the broker can produce, or a late duplicate is processed twice.

type LockOption 

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

LockOption configures the advisory-lock elector.

func LockKey 

func LockKey(name string) LockOption

LockKey names the lock. The default is "warren/outbox". Two services sharing one database AND one key will contend for a single leadership, and one of them will never drain — give each its own.

func RetryInterval 

func RetryInterval(d time.Duration) LockOption

RetryInterval is how often a follower retries acquisition. The default is 5s: the window between a leader dying and the outbox draining again.

type Option 

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

Option configures the Postgres module.

func ApplicationName 

func ApplicationName(name string) Option

ApplicationName is what this service calls itself to Postgres. It defaults to the name of the running binary.

It is not decoration. An operator looking at pg_stat_activity or pg_locks on a shared database sees one row per connection, and without this every row is anonymous: which service holds the outbox's advisory lock, which one is running the query that will not finish, which one to restart. A multi-replica outbox has exactly one leader by design, and finding it started with a blank column.

A DSN that already sets application_name WINS — an explicit choice in the connection string is not overridden by a default derived from argv.

func Configure 

func Configure(fn func(*pgxpool.Config) error) Option

Configure runs fn against the parsed pool configuration — after pgxpool.ParseConfig and BEFORE the pool is created, which is the only moment ConnConfig.Tracer, a custom type registration, or a dial function can still be set. An error from fn fails the boot as a WIRING failure, because ParseConfig does no I/O and nothing has been dialled yet.

It is the second place a pgx type appears in this package's exported surface (AGENT.md invariant 3), and it exists because query spans cannot be wired any other way: the tracer seam is a pgx interface, and warren/observability may not import this module (invariant 4). So database instrumentation is one explicit line in main rather than something this package can do for you: 

postgres.Configure(func(c *pgxpool.Config) error {
    c.ConnConfig.Tracer = otelpgx.NewTracer()
    return nil
})

func ConnectTimeout 

func ConnectTimeout(d time.Duration) Option

ConnectTimeout bounds one connection attempt, including the ping in OnStart. The default is 5s: a boot that hangs on an unreachable database is worse than a boot that fails.

func DSN 

func DSN(dsn string) Option

DSN sets the Postgres connection string. Required: omitting it fails the boot. The password never appears in a log line, an error, or a diagnostic — see redact.

func HealthTimeout 

func HealthTimeout(d time.Duration) Option

HealthTimeout bounds one readiness ping. The default is 2s. The check gates readiness: a service whose database is unreachable should leave the load balancer's rotation.

func MaxConnIdleTime 

func MaxConnIdleTime(d time.Duration) Option

MaxConnIdleTime closes a connection idle for d. The default is 30m.

func MaxConnLifetime 

func MaxConnLifetime(d time.Duration) Option

MaxConnLifetime retires a connection after d regardless of health, which is how a pool behind a failing-over primary or a rotating credential eventually moves. The default is 1h.

func MaxConns 

func MaxConns(n int32) Option

MaxConns caps the pool. The default is 10. It is the concurrency limit of every handler that touches the database, so size it against the request budget rather than against the server's max_connections.

func MinConns 

func MinConns(n int32) Option

MinConns keeps n connections warm, so the first request after an idle period does not pay a handshake. The default is 0.

func Raw 

func Raw(fn func(context.Context, *pgxpool.Pool) error) Option

Raw runs fn against the raw pgx pool during OnStart, after the pool is live and the ping succeeded, and before any dependent hook starts. An error from fn fails the boot.

It is an explicit opt-out: run a warm-up query, inspect the pool's stats, reach a pgx API this package does not model. It CANNOT install a query tracer — ConnConfig.Tracer is read when the pool is constructed, which has already happened by the time this runs. Use Configure for that.

func StatementCacheMode 

func StatementCacheMode(m CacheMode) Option

StatementCacheMode selects how prepared statements are cached. Use StatementCacheDescribe behind a pgbouncer in transaction pooling mode.

func StatementTimeout 

func StatementTimeout(d time.Duration) Option

StatementTimeout bounds a single statement, on both sides of the wire. The default is 30 seconds; 0 disables it.

It is set as the connection's server-side statement_timeout AND applied as a context deadline by the query wrapper, and it needs both. The server-side half is what cancels a genuinely slow query. The client-side half is what survives a server that cannot answer at all: a paused, partitioned or swapping Postgres enforces no timeout of its own, because enforcing one is work it has to do — so without a deadline here a request goroutine waits for ever, and the pool empties behind it.

A caller's own deadline always wins; this only bounds a context that has none. A statement that exceeds it is UNAVAILABLE, which is what warren.md §3.3 promises for a database that cannot serve and what makes app.Retrying re-run the transaction.

func WithAdvisoryLock 

func WithAdvisoryLock(opts ...LockOption) Option

WithAdvisoryLock enables an outbox.Elector backed by a Postgres session-level advisory lock, so exactly one replica drains the outbox.

It holds one connection outside the pool's rotation for as long as it leads, and cancels the context it handed the relay the moment that connection dies. A lock lost is leadership lost, immediately — the alternative is two replicas publishing the same rows. 

postgres.Module(postgres.DSN(url), postgres.WithOutbox(), postgres.WithAdvisoryLock())

func WithInbox 

func WithInbox(opts ...InboxOption) Option

WithInbox enables a durable inbox.Store over the warren_inbox table.

What durability buys is REACH, not atomicity. The seen set survives a restart, a rolling deploy and a partition rebalance, and every replica consults the same one — so a redelivery landing on a different process, or after this one restarted, is still suppressed. The memory store forgets all of that at exit, which is the whole difference. Re-marking refreshes the window.

The row is NOT written in the handler's transaction, and nothing here makes it so. This comment used to claim it was, and a field test measured the two rows landing at different xmins. broker.Deduplicate is a consumer-ring stage: it calls MarkSeen after the handler has returned, by which time app.Transactional has already committed and the transaction is off the context. Handler success and mark-seen are two commits, and a crash between them redelivers the message. That is at-least-once — warren.md §5.6's guarantee for every store, this one included — and handlers must be idempotent regardless.

Its OnStart verifies the table exists and fails the boot naming the fix. It does not create it: see Migrate.

func WithOutbox 

func WithOutbox(opts ...OutboxOption) Option

WithOutbox enables a durable outbox.Store over the warren_outbox table, and registers outbox.Sink on the UnitOfWork so events drained at commit become rows in the same transaction.

The store implements outbox.Waiter with LISTEN/NOTIFY. The notify is issued INSIDE the business transaction, so Postgres delivers it exactly when the rows become visible — a missed signal costs the relay's poll interval and never loses a record.

Its OnStart VERIFIES the table exists and fails the boot naming the fix if it does not. It does not create it: see Migrate.

It is an option rather than a sibling module because it needs the pool, and a sibling module cannot see another module's providers.

type OutboxOption 

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

OutboxOption configures the outbox store.

func Encoder 

func Encoder(e outbox.Encoder) OutboxOption

Encoder sets how a domain event becomes an outbox record. The default is outbox.JSONEncoder().

func Retention 

func Retention(d time.Duration) OutboxOption

Retention deletes published rows older than d on a sweep, so the table does not grow without bound. The default is 24h; zero keeps rows forever, which is right only when something else archives them.

func Table 

func Table(name string) OutboxOption

Table renames the outbox table. The default is "warren_outbox". Changing it after rows exist is a migration you write.

type Queryer 

type Queryer interface {
	// Query runs a query returning many rows. Close the Rows.
	Query(ctx context.Context, sql string, args ...any) (Rows, error)
	// QueryRow runs a query returning at most one row. Its error surfaces
	// from Scan, and a missing row is ErrNoRows.
	QueryRow(ctx context.Context, sql string, args ...any) Row
	// Exec runs a statement and returns the number of rows it affected.
	Exec(ctx context.Context, sql string, args ...any) (int64, error)
}

Queryer is the driver-free query handle. Its method set is deliberately three methods wide: it is the surface every non-pgx implementation of the same idea would have to provide. CopyFrom and SendBatch are reachable through Raw.

type Row 

type Row interface {
	Scan(dest ...any) error
}

Row is one row, or the error that prevented it. It is byte-identical to pgx.Row, so a pgx row satisfies it with no wrapping and no allocation.

type Rows 

type Rows interface {
	Next() bool
	Scan(dest ...any) error
	Err() error
	Close()
}

Rows is a result set. It is the subset of pgx.Rows a repository needs, so a pgx result satisfies it unwrapped.

type UnitOfWork 

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

UnitOfWork runs a function inside one Postgres transaction and makes the state it wrote and the events its aggregates raised commit together.

It is provided by Module; construct it only in a test.

func (*UnitOfWork) Do 

func (u *UnitOfWork) Do(ctx context.Context, fn func(context.Context) error, opts ...persistence.Option) error

Do begins a transaction, puts it on the context, runs fn, drains the events of every aggregate that enlisted through persistence.Track, runs the commit sinks inside that transaction, and commits. fn returning an error, a sink failing, or a panic rolls everything back; a panic is re-raised.

A nested Do JOINS the transaction in scope: it opens nothing, commits nothing, drains nothing, and returns fn's error — so an inner failure rolls back the outer transaction too. A caller who handles the inner error and returns nil from the outer fn therefore commits state the inner call may have left inconsistent; that is the cost of join semantics, and it is the same on every driver.

Passing an Option to a nested Do is an error, not a silent no-op: Postgres sets isolation and read-only on a transaction's FIRST statement, so a nested Isolation(Serializable) is unimplementable — and ignoring it would mean a handler asking for Serializable and quietly getting Read Committed.

func (*UnitOfWork) OnCommit 

func (u *UnitOfWork) OnCommit(fn func(context.Context, []domain.Event) error)

OnCommit registers a sink for the events drained at commit — how the outbox writer attaches, and how a test asserts on what was published.

Sinks run INSIDE the transaction, on its context, so a sink's own writes join it and a failing sink fails the commit. That atomicity is the entire pattern: rows and events land together or not at all. 

uow.OnCommit(outbox.Sink(store, outbox.JSONEncoder()))

Outbox wires this for you.

Source Files

View all Source files

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.