postgres

package module
v0.18.0 Latest Latest
Warning

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

Go to latest
Published: Oct 16, 2026 License: Apache-2.0 Imports: 19 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_events the durable event trail (see EventStore), whose leaf column is bytea and whose primary key is (run_id, seq); 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. With WithoutMigration they create nothing, for a role with no CREATE on the schema, and refuse a schema that is missing (ErrSchemaMissing) or at another version (ErrSchemaVersion); the owner runs Migrate (bide migrate). 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, (run_id, seq) on bide_events, 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 table, with no partitions or inheritance, 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

View Source
var ErrSchemaMissing = errors.New("postgres: the schema is missing")

ErrSchemaMissing is wrapped by the error Open, New or OpenReadOnly return when the store's schema, or a part of it (the pinned schema, a table, the next_seq function, a lease index), does not exist and the store was not allowed to create it: it was opened WithoutMigration, or read only. Run Migrate (bide migrate) as the role that owns the schema, then open the store again. The error also wraps agent.ErrConfig.

View Source
var ErrSchemaVersion = errors.New("postgres: the schema version does not match")

ErrSchemaVersion is wrapped by the error returned when the database's schema version is not the one this version of the package uses. A newer schema needs a newer version of the program. An older one needs a migration step; there is one schema version (1) today, so an older one means the version table was edited, and Migrate refuses it rather than record a version it did not reach. The error also wraps agent.ErrConfig.

Functions

This section is empty.

Types

type EventStore added in v0.16.0

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

EventStore is a Postgres-backed audit.EventStore: the durable copy of a run's event trail (see audit.PersistJournal and audit.LoadEventLog), in the bide_events table (see WithTablePrefix and WithSchema). It shares the Store's pool, so Close on the Store closes it, and it needs no setup beyond Open or New, which create the table.

A leaf is a row with primary key (run_id, seq), and Append is one INSERT, committed before it returns, that inserts the leaf only at the run's next position with ON CONFLICT DO NOTHING on that key. Writers racing for a position, from any connection or node, all run the statement: Postgres lets one insert, makes the others wait for it, and then they find its row, report a no-op for the same bytes and audit.ErrSeqTaken for different ones. A statement that fails with a serialization failure or a deadlock is run again, as the Store's are. A crash leaves the trail as of the last committed Append: a leaf is never partly written, and a re-mirror (audit.PersistJournal) appends what is missing. Leaves are stored as the exact bytes given.

The table is plain: the store never updates or deletes a leaf, so a deployment that wants the trail insert-only can grant its role INSERT and SELECT on bide_events alone. A store opened with OpenReadOnly does not create the table and can read it only if it exists.

func (*EventStore) Append added in v0.16.0

func (e *EventStore) Append(ctx context.Context, runID string, seq int, leaf []byte) error

Append implements audit.EventStore. A seq past the run's next position, or a negative one, is an error; a different leaf at a stored seq fails with an error that wraps audit.ErrSeqTaken.

func (*EventStore) Load added in v0.16.0

func (e *EventStore) Load(ctx context.Context, runID string) ([][]byte, error)

Load implements audit.EventStore. It reads eventsPage leaves per query, so it holds no connection between pages; the trail only grows, so the pages join into one contiguous trail.

type MigrationResult added in v0.17.0

type MigrationResult struct {
	// Schema is the schema the tables are in.
	Schema string
	// Before is the schema version found before the migration; 0 when the schema had none yet (a
	// new database).
	Before int
	// After is the schema version once the migration is done: the version of this package.
	After int
	// Created names what the migration created (with WithDryRun, would create): tables, the
	// next_seq function and lease indexes, schema-qualified. It is empty on a current schema.
	Created []string
	// Statements are the SQL statements the migration ran, in order, as sent to the server. A
	// CREATE TABLE IF NOT EXISTS (and the version row's INSERT ... ON CONFLICT DO NOTHING) is
	// listed even when the table exists: it is sent, and does nothing. The function and each
	// index are listed only when they were missing.
	Statements []string
	// DryRun is set when the migration was rolled back (WithDryRun).
	DryRun bool
}

MigrationResult is what a migration did, or with WithDryRun would have done.

func Migrate added in v0.17.0

func Migrate(ctx context.Context, dsn string, opts ...Option) (MigrationResult, error)

Migrate creates or checks the store's schema at dsn (a pgx DSN, as Open takes) as the role in the DSN, which needs the privileges of the schema's owner (CREATE on the schema), and reports the schema version before and after. It is the migration Open runs, on its own, so that the running service can open the store WithoutMigration under a role that has no DDL rights.

It is idempotent: on a current schema it creates nothing and changes nothing. It never alters or drops an existing table or replaces a function, and refuses (ErrSchemaVersion) a schema newer than this version's. opts are Open's (WithSchema, WithTablePrefix) and WithDryRun; WithoutMigration is an agent.ErrConfig error. After the migration it checks the tables and the function as Open does. Run it with the owner role, pin the schema with WithSchema, and see the Postgres store guide for the roles.

func MigrateDB added in v0.17.0

func MigrateDB(ctx context.Context, db *sql.DB, opts ...Option) (MigrationResult, error)

MigrateDB is Migrate over db, which the caller opened (with the pgx driver) and closes, as New is to Open.

type Option added in v0.9.0

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

Option configures Open and New.

func WithDryRun added in v0.17.0

func WithDryRun() Option

WithDryRun makes Migrate and MigrateDB run the migration in a transaction and roll it back: the result lists the statements and the versions as if it had committed, the role is shown to hold the privileges the migration needs, and nothing is left behind. Open, New and OpenReadOnly refuse it (agent.ErrConfig).

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", prefix+"events" 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.

func WithoutMigration added in v0.17.0

func WithoutMigration() Option

WithoutMigration makes Open and New verify the schema instead of creating it. The store runs no CREATE and no INSERT into the version table, so its role needs no CREATE on the schema: the least-privilege deployment runs Migrate (bide migrate) once as the owner role, and the running service uses this option. Opening still checks everything the migration would have created and refuses what is not there:

  • a missing schema, table (steps, leases, schema version, events), next_seq function or lease index fails with an error that wraps ErrSchemaMissing;
  • a schema version other than this package's fails with an error that wraps ErrSchemaVersion;
  • a table that differs from the migration's (a column of another type, nullable or missing, a nondeterministic collation, an extra NOT NULL column with no default, a trigger, rule or row-level security on it), one without the uniqueness the store depends on, a lease index that is not the migration's plain btree on the leases table, and a next_seq function that is not the one this version creates, fail as they do for Open (agent.ErrConfig).

Both errors also wrap agent.ErrConfig. The role needs USAGE on the schema, SELECT on the version table (and the steps, leases and events tables), the privileges its operations use (see the Postgres store guide) and EXECUTE on the next_seq function. Migrate refuses this option.

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, or with WithoutMigration verifies it (see Open).

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, creating it when it does not; with WithoutMigration it only verifies it, for a role with no DDL rights (see Migrate). Close closes the connection pool.

func OpenReadOnly added in v0.13.0

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

OpenReadOnly connects to the Postgres journal at dsn (a pgx DSN, as Open takes) for reading only. Every connection is opened with default_transaction_read_only on, so the server refuses any write on it, and the store neither migrates nor creates anything: it reads the schema version and checks the tables as Open does after migrating. Get, Load and Runs (agent.Lister) read as they do on a Store from Open; Insert and the agent.Leaser methods fail with the server's read-only error (wrapping agent.ErrStorage for Insert). opts are Open's (WithSchema, WithTablePrefix).

It refuses a database with no journal tables, and one whose schema version is newer than this version's (ErrConfig). Close closes the pool. It is for tools that inspect a journal, such as bide view, never for driving a run.

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) EventStore added in v0.16.0

func (s *Store) EventStore() *EventStore

EventStore returns the audit.EventStore over s's database. It is cheap; the handle holds no state of its own.

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) 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) 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