Postgres World

import postgresworld "github.com/eriicafes/go-workflow/worlds/postgres"

The Postgres World stores workflow state in PostgreSQL and is safe to use from many processes at once: API servers, workers, or both in one binary.

Usage

world, err := postgresworld.Open(ctx, postgresworld.Config{
	BaseURL: os.Getenv("WORKFLOW_BASE_URL"), // https://workflows.example.com/v1/workflow
	URL:     os.Getenv("DATABASE_URL"),
})
if err != nil {
	log.Fatal(err)
}
defer world.Close()

runners := worlds.Runners(workflow.Use(flows.GenerateReport))
go func() {
	if err := postgresworld.Worker(ctx, world, runners); err != nil && !errors.Is(err, context.Canceled) {
		log.Printf("workflow worker stopped: %v", err)
	}
}()
log.Fatal(http.ListenAndServe(":8081", postgresworld.HTTP(world)))

Open creates and owns a pgxpool. New(ctx, pool, config) uses an existing pool and leaves it open on Close. Stop Worker loops by cancelling their context before calling Close.

Configuration

Field	Default	Meaning
BaseURL	required	Public URL of HTTP(world), including /v1/workflow.
URL	required by Open	Connection string.
PollInterval	250 ms	Longest sleep of workers and waiters when no notification arrives.
Concurrency	10	Deliveries one Worker call executes at once.
LeaseDuration	1 min	How long a claimed delivery is hidden from other workers. Workers renew it while executing, so it only bounds how quickly a crashed worker's delivery is retried.
Routes	none	The queue of each routed workflow. See routing.
Retention, Retentions	keep forever	How long finished runs are kept, by default and per workflow. See retention.

Entry points

• Worker(ctx, world, runners, options...): the queue worker. It blocks until ctx is cancelled, waits for in-flight deliveries, and returns ctx.Err(). WorkerOptions.Queues limits the queues it claims, and WorkerOptions.Concurrency overrides Config.Concurrency.
• HTTP(world): RPC and webhook endpoints under /v1/workflow.
• HTTPPrefix(world, prefix): the same endpoints under prefix.
• RPC(world) and Webhook(world): each endpoint alone.

When the application and worker share a process, pass the World to workflow.NewClient. When they are separate, serve HTTP(world) from the worker and use workflow.NewHTTPClient in the application, or open the same database from both.

Schema and migrations

Open and New migrate the schema. Migrations run in one transaction under pg_advisory_xact_lock, so processes starting together migrate once, and the applied versions are recorded in go_workflow_schema_version. A database with a newer version than the library knows is rejected.

Tables are created in the connection's search_path. Set search_path in the connection string or pool config to use a dedicated schema.

Table	Contents
go_workflow_runs	Run view plus event_count. The next slot is event_count + 1.
go_workflow_events	Event log keyed by (run_id, slot).
go_workflow_steps	Step view keyed by (run_id, step_id).
go_workflow_hooks	Hook view, including webhook settings and retention.
go_workflow_hook_tokens	Token ownership. The primary key makes a token single-owner.
go_workflow_queue	Pending deliveries with not_before, attempts, and lease_until.
go_workflow_streams	Stream state: closed, writer session, sequence, chunk_count.
go_workflow_stream_chunks	Stream chunks keyed by (run_id, name, idx).

Payloads are bytea, attributes are jsonb, and timestamps are timestamptz at microsecond precision. Times returned by the World are UTC.

Consistency

Every event append is one transaction. It locks the run row with SELECT ... FOR UPDATE, checks the transition rules, inserts the events at the next slots, and updates the views. Writers to one run are serialized across processes, slots are contiguous, and a batch is all-or-nothing.

EventsCreate(hook_created) locks the token row. A forced claim over a live owner also locks the owner's run, taking run locks in run ID order to avoid deadlocks. Exactly one concurrent claimant wins a free token. Transactions are retried on serialization failures and deadlocks.

Routing workflows to queues

world, err := postgresworld.Open(ctx, postgresworld.Config{
	// BaseURL, URL...
	Routes: worlds.Routes(worlds.Route[workflows.Backfill]("backfill")),
})

go postgresworld.Worker(ctx, world, runners, postgresworld.WorkerOptions{Queues: []string{postgresworld.DefaultQueue}})
go postgresworld.Worker(ctx, world, runners, postgresworld.WorkerOptions{Queues: []string{"backfill"}, Concurrency: 2})

Every delivery of a routed workflow's runs is stored on its queue, and other workflows' on DefaultQueue. A worker without Queues claims every queue.

Worker model

Queue inserts a row and sends NOTIFY go_workflow_queue in the same transaction. An IdempotencyKey deduplicates against a pending row of the same run. Claiming the row releases the key, so a message queued during execution is never dropped.

Worker claims due rows with FOR UPDATE SKIP LOCKED, sets lease_until, increments attempts, and executes each:

• success deletes the row.
• failure keeps it and sets not_before to now plus a capped exponential backoff (1 s, 2 s, 4 s, … up to 5 minutes), and logs the error.
• shutdown releases the lease so another worker picks the row up immediately.

Retention

With Config.Retention set, each Worker deletes expired runs in the background: at start and then every minute, in batches of 100, until no expired run is left. One statement per batch picks runs that completed, failed or were cancelled longer ago than their workflow's retention (Config.Retentions, else Retention) and deletes their events, steps, hooks and the tokens those hooks own, queued deliveries, streams and chunks, and finally the runs, so a batch is all or nothing. Its FOR UPDATE SKIP LOCKED lets every worker sharing the database run it without waiting on each other. A run with a hook whose token retention has not ended is kept until it ends, and runs that are still pending or running never expire.

LISTEN/NOTIFY

Each World opens one dedicated connection on first use and listens on go_workflow_queue, go_workflow_runs (sent when a run turns terminal), and go_workflow_streams. Workers, RunsWaitForTerminalStatus, and StreamsGetChunks wake on these notifications and also re-check every PollInterval, so a lost notification only adds latency.

Connection poolers

Poolers in transaction mode, such as PgBouncer, do not support LISTEN. Point the World at Postgres directly or use session pooling.

Scaling

Run Worker loops across as many processes as needed. SKIP LOCKED gives each row to one worker at a time. Size Concurrency to the work your runners do, and the pool to roughly Concurrency plus your API traffic. Each World uses one extra connection for LISTEN.

Security

• Give the database role only the permissions it needs on the go_workflow_ tables (plus CREATE on the schema for migrations, or migrate with a separate role), require TLS, and keep the database reachable only from your application and worker network.
• HTTP(world) performs no authentication or per-run authorization. Prefer a private route from the application to the worker, or wrap RPC(world) in your own auth middleware.
• The webhook route is authorized only by its token.

Tests

docker run -p 5432:5432 -e POSTGRES_PASSWORD=postgres postgres
go test -race ./worlds/postgres/...

The tests use DATABASE_URL, defaulting to postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable. Each database test creates and drops its own schema.