Documentation
¶
Overview ¶
Package postgres is the production HA queue driver. Unlike the SQLite fast-path, which leans on the engine's single-writer guarantee, this driver uses SELECT … FOR UPDATE SKIP LOCKED so N parallel workers (and N filex nodes behind a load balancer) can safely race for the same queue without losing ops or double-claiming them.
The schema is identical to the SQLite version (see db/migrations/postgres/00006_queue.sql) modulo column types: payload is JSONB and timestamps are TIMESTAMPTZ. JSONB is read back via the `::text` cast so the Go scan target stays a string and the rest of the helpers can be shared verbatim with sqlite.
Index ¶
- type Driver
- func (d *Driver) Ack(ctx context.Context, id string) error
- func (d *Driver) Cancel(ctx context.Context, id string) error
- func (d *Driver) Close() error
- func (d *Driver) Dequeue(ctx context.Context, types []string) (queue.Op, error)
- func (d *Driver) Enqueue(ctx context.Context, op queue.Op) (string, error)
- func (d *Driver) Fail(ctx context.Context, id, errMsg string, retry bool) error
- func (d *Driver) Get(ctx context.Context, id string) (queue.Op, error)
- func (d *Driver) Init(_ context.Context, cfg map[string]any) error
- func (d *Driver) List(ctx context.Context, status string, limit, offset int) ([]queue.Op, int64, error)
- func (Driver) Name() string
- func (d *Driver) RecoverOrphans(ctx context.Context, olderThan time.Duration) (int64, error)
- func (d *Driver) Retry(ctx context.Context, id string) error
- func (d *Driver) Stats(ctx context.Context) (queue.Stats, error)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Driver ¶
type Driver struct {
// contains filtered or unexported fields
}
Driver is the Postgres-backed queue.
func (*Driver) Close ¶
Close releases the underlying *sql.DB only when this driver opened it (cfg["dsn"] path). When wired against the application's shared *sql.DB (cfg["db"]), Close is a no-op so we don't tear down the metadata store — same semantics as the sqlite driver.
func (*Driver) Dequeue ¶
Dequeue claims the next eligible op atomically using SELECT … FOR UPDATE SKIP LOCKED. SKIP LOCKED is the canonical Postgres pattern for work-stealing queues: parallel workers each get a different row in one round-trip, with no application-level coordination.
func (*Driver) Fail ¶
Fail records the failure. retry=true requeues with a 30 s delay so a flapping upstream doesn't immediately re-burn an attempt; retry=false terminates the op as failed.
func (*Driver) Init ¶
Init opens the queue's DB. cfg keys:
dsn — Postgres connection string (e.g. postgres://u:p@host/db?sslmode=require).
Required when cfg["db"] is absent.
db — *sql.DB (preferred). When set, dsn is ignored. The bootstrap
path passes the application *sql.DB so the queue piggybacks on
the existing connection pool.
func (*Driver) List ¶
func (d *Driver) List(ctx context.Context, status string, limit, offset int) ([]queue.Op, int64, error)
List returns ops with optional status filter and pagination.
func (*Driver) RecoverOrphans ¶
RecoverOrphans flips long-running rows back to pending. Called on boot to handle ungraceful shutdowns. The interval is parameterised as a floating-point second count via make_interval() — that's safer than string-concatenating the value into the SQL because olderThan is caller-controlled.
Source Files