postgres

package module
v0.10.0 Latest Latest
Warning

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

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

Documentation

Overview

Package postgres is a Postgres-backed agent.Store for high-availability durable agents: shared Postgres storage lets any node resume any run, so a dead worker's in-flight agents pick up elsewhere. It implements agent.Store, agent.Lister and agent.Leaser.

The tables are named with a prefix, "bide_" by default (see WithTablePrefix): bide_steps holds the journal (run_id, seq, name, data), where data is bytea, each entry's bytes verbatim; bide_leases holds the leases; bide_schema_version the schema version; and the function bide_next_seq_v1 gives an insert its position. Open and New create the tables and the function if they do not exist and refuse a database whose schema version is newer than this version knows. They never alter an existing table or replace a function, so they refuse a table that lacks a uniqueness the store's statements depend on (unique (run_id, seq) and (run_id, name) on bide_steps, and run_id on bide_leases) and a next_seq function with another definition. Run IDs are compared and ordered by their bytes (collation "C"), so a Lister's order does not depend on the database's locale.

One statement per write

Every write the store makes after Open (an insert, AcquireLease, RenewLease, ReleaseLease) is a single statement sent on its own, which Postgres runs as a transaction of its own and commits before it replies. No transaction spans two round trips, so a client that stops between two of its round trips (a SIGSTOP, a suspended VM, a long GC pause, a partition) holds no row or advisory lock while it is stopped. Another node therefore takes an expired lease one TTL after the last renewal that committed, or records the run's next step, however long the stall lasts. Reads are single SELECT statements too, and nothing holds a connection between two pages.

Each statement is atomic by itself: AcquireLease is an INSERT ... ON CONFLICT DO UPDATE whose WHERE grants the lease only when it is free, expired or already the holder's; RenewLease and ReleaseLease are an UPDATE and a DELETE conditioned on the holder (and, for a renewal, on the lease being live); and Insert is an INSERT whose position comes from next_seq.

Inserts into one run

next_seq takes the run's transaction-level advisory lock and then reads MAX(seq)+1. The lock is held by the insert's own transaction, the one statement, and released when it commits, so a stalled client holds it no longer than any statement's row locks. Inserts into one run queue on it; the function is VOLATILE, so its read takes a snapshot after the lock is granted and, at read committed, sees every insert that held the lock before. Each insert therefore takes its position in its first attempt, and positions follow commit order. Should two inserts still read the same MAX (at repeatable read or serializable, where the function reads the transaction's snapshot, or beside a writer that does not take the lock), they collide on the (run_id, seq) key: the second waits for the first to commit, fails, and is run again. The lock's key is the one earlier versions took in a statement of their own, so nodes of both versions queue on the same lock during a rolling upgrade.

The key is hashtextextended(run_id, 0), which names neither the table prefix nor the schema, and govern/postgreslog takes the same key for an entity. So stores with different prefixes, stores in different schemas of one database, and a governed entity named like a run share one lock. That costs throughput, not correctness: the unique (run_id, seq) index keeps each store's positions distinct whatever the lock does. The sharing stays, since the upgrade depends on the key.

At repeatable read or serializable, a queued insert's snapshot was taken before it was granted the lock, so under sustained contention on one run most inserts collide once or more and retry with a backoff (see Retries and deadlines): positions stay correct, but throughput on that run falls well below read committed's, and a caller without a deadline may wait long. A deployment that writes one run from many goroutines at once is better served at read committed.

Schema and trust

Every name in every statement the store sends is qualified: its tables and next_seq function with the store's schema, and every function, type, collation and operator with pg_catalog (operators written OPERATOR(pg_catalog.<op>)). No statement, Open's and the migration's included, looks a name up through the search path. Postgres still resolves some things without a name, none of them through the search path: ORDER BY, DISTINCT and the ON CONFLICT arbiter use the type's default operator class and the table's indexes, and casts use pg_cast. The statement check (sqlcheck.go) holds every statement to this.

Which schema is the store's is the one thing the search path can still decide, and only when the schema is not pinned. With WithSchema, the recommended deployment, the store uses the schema given and the search path plays no part. Without it, every Open discovers the schema: the first schema on the search path holding a relation named like the steps table, or the first schema on the path (current_schema()), where the migration creates the tables, when none does; it refuses a first relation of that name that is not an ordinary or partitioned table, and logs a warning. Discovery runs again at every Open, so a role that can create a schema earlier on the path can redirect a restarting node to a store of its own: any role with CREATE on the database can create the "$user" schema the default search path puts first. Discovery is safe only when every schema on the search path is trusted. Within one process, after Open, no role can redirect the store, pinned or not, as it could in every earlier version.

The store trusts the owner of its schema and every role that can create objects in it, as it trusts the owner of its tables: such a role could replace a table, and so it could replace the next_seq function. Within that boundary, the store refuses what it can check. The next_seq function must be owned by the owner of the steps table, run with search_path = pg_catalog, pg_temp and have exactly this version's body, whose names are all qualified; Open refuses it otherwise. Unpinned, the store also trusts every role that can create a schema earlier on its search path (see above).

Any role that can connect can take a run's advisory lock key itself (pg_advisory_lock(hashtextextended(run_id, 0))) and hold it, and every insert into that run then waits until it lets go. That was so before next_seq too, since earlier versions took the same key; it stalls a run's writes, and corrupts nothing.

The next_seq function's body names its schema. After ALTER SCHEMA ... RENAME, every insert fails and Open refuses the function: drop it (DROP FUNCTION <schema>.bide_next_seq_v1(text), with the prefix) and Open again, and the migration creates it under the new name.

Isolation

The statements behave the same whatever the deployment's default_transaction_isolation (set on the server, database, role or DSN). At read committed, a statement that meets a row another transaction changed after the statement began waits for that transaction and then acts on the row's latest version. At repeatable read or serializable, Postgres fails the statement with a serialization failure (40001) instead, and at serializable it may fail a read that conflicts with concurrent serializable writes the same way. A statement that fails that way has changed nothing, so the store runs it again, with a new snapshot; once the conflicting transaction has committed, that is the outcome read committed reaches in one attempt. The schema migration in Open is the one transaction of several statements; it sets read committed itself (see txOptions).

Retries and deadlines

A statement is run again after a serialization failure or a deadlock, and an insert also after losing its position to another insert. A retry does not always follow another transaction's commit: at serializable, Postgres may fail a statement for a conflict with a transaction that has not committed yet, and that transaction may itself abort. Under heavy contention on one run (many writers inserting into it at once) an insert may lose its position several times. So the store spaces the attempts with a capped, jittered exponential backoff (from 1ms up to 100ms, a random fraction of it each time, so contending writers do not retry in lockstep), and keeps retrying until the statement succeeds, fails for another reason, or ctx is done: a bound on the attempts would turn contention into errors. A caller that needs a bound on the time a call may take sets a deadline on ctx; when ctx ends, the call returns an error that wraps both ctx's error and the last attempt's.

The migration also sets idle_in_transaction_session_timeout for its own transaction, so a node stopped inside it cannot hold the migration lock. The store sets no session-wide timeout: no other statement it sends is ever inside a transaction, so the setting would guard nothing of the store's, and on a pool passed to New it would change the caller's own transactions. A deployment that wants the bound for its own code sets it on the role or the database.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Option added in v0.9.0

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

Option configures Open and New.

func WithSchema added in v0.9.0

func WithSchema(name string) Option

WithSchema pins the schema the store's tables and next_seq function are in, instead of discovering it through the search path at every Open. It is the recommended deployment: with it, the search path plays no part in which schema the store uses, and no role can redirect a restarting node by creating a schema earlier on the path. The name is used as given (quoted), so "App" and "app" are different schemas, and Postgres truncates it, like any identifier, to 63 bytes. The schema must exist: Open fails with ErrConfig otherwise, and never creates it. An empty name, and a system schema (information_schema, or a name starting with pg_), is an ErrConfig error.

func WithTablePrefix added in v0.9.0

func WithTablePrefix(prefix string) Option

WithTablePrefix names the tables with prefix instead of "bide_": prefix+"steps", prefix+"leases" and prefix+"schema_version". It must be a lower-case SQL identifier (letters, digits and underscores, not starting with a digit), or Open and New return an ErrConfig error.

type Store

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

Store is a Postgres-backed agent.Store.

func New added in v0.9.0

func New(ctx context.Context, db *sql.DB, opts ...Option) (*Store, error)

New returns a Store over db, which the caller opened (with the pgx driver) and closes, and ensures the schema exists.

func Open

func Open(ctx context.Context, dsn string, opts ...Option) (*Store, error)

Open connects to Postgres via a pgx DSN (e.g. "postgres://user:pass@host:5432/db") and ensures the schema exists. Close closes the connection pool.

func (*Store) AcquireLease

func (s *Store) AcquireLease(ctx context.Context, runID, holder string, ttl time.Duration) (bool, error)

AcquireLease implements agent.Leaser: claim runID for holder until now()+ttl. The upsert grants the lease when the run is unleased, already held by holder (renewal), or the current lease has expired, and grants nothing when another holder's lease is still live. Expiry uses the database clock (now()) so all nodes compare against one clock, not their own.

It is one statement, committed before Postgres replies, so a holder that stops after it has sent it holds no lock on the lease's row (see the package documentation).

func (*Store) Close

func (s *Store) Close() error

Close closes the connection pool Open opened. A Store made with New leaves its db open.

func (*Store) Do deprecated

func (s *Store) Do(ctx context.Context, runID, name string, fn func(context.Context) (agent.Record, error)) (agent.Record, error)

Do runs a memoized step through s's Journal (see agent.Journal.Do).

Deprecated: transitional; removed by the 1.0 rewrite. Use a Journal over the store.

func (*Store) Get added in v0.9.0

func (s *Store) Get(ctx context.Context, runID, name string) (agent.Entry, bool, error)

Get implements agent.Store.

func (*Store) History deprecated

func (s *Store) History(ctx context.Context, runID string) ([]agent.Record, error)

History reads a run back through s's Journal (see agent.Journal.History).

Deprecated: transitional; removed by the 1.0 rewrite. Use a Journal over the store.

func (*Store) Insert added in v0.9.0

func (s *Store) Insert(ctx context.Context, runID, name string, data []byte) (agent.Entry, bool, error)

Insert implements agent.Store. It is one INSERT statement, committed before Postgres replies, that takes the next position (MAX(seq)+1) in its own snapshot. An insert computes position n+1 only when the entry at n is committed and visible to it, so positions follow commit order: a reader never sees an entry before one with a lower Seq. Two inserts that read the same MAX collide on UNIQUE (run_id, seq); the later one waits for the first to commit, fails with 23505, and runs again with a snapshot that sees it.

func (*Store) Journal deprecated added in v0.9.0

func (s *Store) Journal() *agent.Journal

Journal returns the Journal over s that its Do and History shims delegate to.

Deprecated: transitional; removed by the 1.0 rewrite. Use agent.NewJournal(s).

func (*Store) Load added in v0.9.0

func (s *Store) Load(ctx context.Context, runID string, after int64) iter.Seq2[agent.Entry, error]

Load implements agent.Store. It reads pageSize entries per query and yields them with no connection held, so the caller may write to the store inside its loop.

func (*Store) ReapLeases added in v0.10.0

func (s *Store) ReapLeases(ctx context.Context, ended []string) (int, error)

ReapLeases implements agent.Leaser: one DELETE of the lapsed leases whose run is finished (holds an entry named in ended), not in the steps table, or a session's or a sub-agent's (its ID contains '>'), which checks the expiry itself, so a lease taken or renewed meanwhile is kept.

func (*Store) ReleaseLease

func (s *Store) ReleaseLease(ctx context.Context, runID, holder string) error

ReleaseLease implements agent.Leaser: relinquish runID if held by holder (a no-op otherwise). It is one statement, like AcquireLease.

func (*Store) RenewLease

func (s *Store) RenewLease(ctx context.Context, runID, holder string, ttl time.Duration) (bool, error)

RenewLease implements agent.Leaser: extend holder's still-live lease on runID. Returns false if holder no longer holds it (expired or taken over). It is one statement, like AcquireLease.

func (*Store) Runs

func (s *Store) Runs(ctx context.Context, f agent.RunFilter) iter.Seq2[string, error]

Runs implements agent.Lister. The filter is evaluated in the query, runsPage run IDs at a time, with no connection held between pages.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL