Documentation
¶
Overview ¶
Package pgstore is an opt-in, Postgres-backed supplemental store for river-workflow. It implements riverworkflow.Store and adds durable results (beyond River's job retention), an indexed run-job lookup for the engine's wake path, listing/search, and push-based completion waiting (Await), all in a single table the library owns alongside River's — without modifying River.
It reuses the same *pgxpool.Pool the application already gives River, and places its table in the same schema River uses. Wire it in by constructing a Store, migrating it, passing it to riverworkflow.EngineOpts.Store, and (for Await) starting its listener:
store := pgstore.New(pool, &pgstore.Opts{Schema: schema})
if err := store.Migrate(ctx); err != nil { ... }
if err := store.Start(ctx); err != nil { ... } // for Await
defer store.Stop()
engine := riverworkflow.NewEngine[pgx.Tx](reg, &riverworkflow.EngineOpts{Store: store})
See docs/design/supplemental-store.md for the design and its trade-offs.
Operational notes: the store's tables (river_workflow, river_workflow_command, river_workflow_signal) are owned by this package and are not touched by River's job cleaner, so finished workflows' rows persist until you delete them — call Store.PruneFinished periodically to apply your own retention. A failed workflow's error message is stored verbatim and returned by Get/List/Result/Await, so sanitize errors that may carry sensitive data before returning them from workflow or activity functions. See docs/caveats.md ("Caveats specific to the pgstore store").
Index ¶
- Variables
- func Result[O any](ctx context.Context, s *Store, id string) (O, bool, error)
- type ListFilter
- type Opts
- type Record
- type Store
- func (s *Store) Await(ctx context.Context, id string) (riverworkflow.Status, error)
- func (s *Store) Checkpoint(ctx context.Context, id string, gen int, ...) error
- func (s *Store) ClearWake(ctx context.Context, id string) error
- func (s *Store) Get(ctx context.Context, id string) (*Record, error)
- func (s *Store) List(ctx context.Context, filter ListFilter) ([]*Record, error)
- func (s *Store) LoadHistory(ctx context.Context, id string, gen int) ([]riverworkflow.CommandRecord, []riverworkflow.SignalRecord, error)
- func (s *Store) LoadState(ctx context.Context, id string) (riverworkflow.WorkflowState, error)
- func (s *Store) MarkWake(ctx context.Context, id string) error
- func (s *Store) Migrate(ctx context.Context) error
- func (s *Store) PendingWakes(ctx context.Context, limit int) ([]riverworkflow.Wake, error)
- func (s *Store) PruneFinished(ctx context.Context, olderThan time.Time) (int64, error)
- func (s *Store) PruneGenerationsBefore(ctx context.Context, id string, gen int) error
- func (s *Store) PruneWorkflow(ctx context.Context, id string) error
- func (s *Store) RecordCancelled(ctx context.Context, id string) error
- func (s *Store) RecordCompleted(ctx context.Context, id string, result json.RawMessage) error
- func (s *Store) RecordFailed(ctx context.Context, id, failure string) error
- func (s *Store) RecordStarted(ctx context.Context, id, kind string, gen int, runJobID int64) error
- func (s *Store) RunJobID(ctx context.Context, id string) (int64, bool, error)
- func (s *Store) Start(ctx context.Context) error
- func (s *Store) Stop()
- func (s *Store) WakeEvents() <-chan string
Constants ¶
This section is empty.
Variables ¶
var ErrNotFound = errors.New("pgstore: workflow not found")
ErrNotFound is returned when no workflow with the given id exists in the store.
Functions ¶
func Result ¶
Result returns a finished workflow's output from the store, decoded into O.
It mirrors riverworkflow.Result, but reads the store's durable table instead of the run job — so it keeps working after River has reaped the job. The boolean is true only when the workflow completed successfully. A workflow that failed or was cancelled returns an error wrapping riverworkflow.ErrWorkflowFailed; one still running returns (zero, false, nil); an unknown one returns ErrNotFound.
Types ¶
type ListFilter ¶
type ListFilter struct {
Kind string // exact workflow kind
Status riverworkflow.Status // exact status
Limit int // max rows (default 100)
}
ListFilter narrows a List query. Zero-valued fields are not applied.
type Opts ¶
type Opts struct {
// Schema is the Postgres schema the workflow table lives in. Empty means the
// connection's search_path (typically "public"). It should match the schema
// River is configured with (river.Config.Schema).
Schema string
}
Opts configure a Store.
type Record ¶
type Record struct {
ID string
Kind string
Status riverworkflow.Status
Gen int
RunJobID *int64
Result json.RawMessage
Error string
CreatedAt time.Time
UpdatedAt time.Time
FinishedAt *time.Time
}
Record is a workflow's row in the store.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store is a Postgres-backed riverworkflow.Store. It is safe for concurrent use. Construct it with New.
func New ¶
New returns a Store over the given pool. opts may be nil. The pool is borrowed, not owned: the caller remains responsible for closing it.
func (*Store) Await ¶
Await blocks until the workflow reaches a terminal status and returns it, or until ctx is done. It returns ErrNotFound if the workflow isn't in the store.
With the listener running (see Start) it is push-based: N concurrent Awaits share the one listener connection and cost only a channel each — no per-caller polling. Without Start it falls back to polling.
func (*Store) Checkpoint ¶
func (s *Store) Checkpoint(ctx context.Context, id string, gen int, commands []riverworkflow.CommandRecord, signals []riverworkflow.SignalRecord) error
Checkpoint durably records newly-terminal commands and delivered signals in a single batch. Idempotent: a command/signal already present is left unchanged (its terminal data never changes). Part of riverworkflow.HistoryStore.
func (*Store) ClearWake ¶
ClearWake clears the workflow's wake-pending flag. Part of riverworkflow.WakeStore.
func (*Store) List ¶
List returns workflow records matching the filter, most-recently-updated first.
func (*Store) LoadHistory ¶
func (s *Store) LoadHistory(ctx context.Context, id string, gen int) ([]riverworkflow.CommandRecord, []riverworkflow.SignalRecord, error)
LoadHistory returns the durably-checkpointed commands for one generation and the full durable signal inbox for the workflow. Part of riverworkflow.HistoryStore.
func (*Store) LoadState ¶
func (s *Store) LoadState(ctx context.Context, id string) (riverworkflow.WorkflowState, error)
LoadState returns the store's authoritative record of a workflow.
func (*Store) MarkWake ¶
MarkWake sets the workflow's wake-pending flag and, if it was still running, emits a cross-process wake notification. Part of riverworkflow.WakeStore.
func (*Store) Migrate ¶
Migrate creates the workflow table and its index if they don't already exist. It is idempotent and safe to run on every startup.
func (*Store) PendingWakes ¶
PendingWakes returns up to limit running workflows whose wake-pending flag is set. Part of riverworkflow.WakeStore.
func (*Store) PruneFinished ¶
PruneFinished deletes the store rows of workflows that finished before olderThan, returning how many workflow summary rows were removed. It is the store's retention control: River's job cleaner never touches these tables, so a finished workflow's summary row (and its durable result) persists until pruned. Call it periodically (e.g. from a ticker) with a cutoff past which you no longer need results. Pruned results are unrecoverable.
It also removes any residual command/signal history for those workflows, which is normally already cleared on completion but is deleted here defensively. Running workflows (finished_at IS NULL) are never affected.
func (*Store) PruneGenerationsBefore ¶
PruneGenerationsBefore deletes checkpointed commands for generations older than gen. Part of riverworkflow.HistoryStore.
func (*Store) PruneWorkflow ¶
PruneWorkflow deletes all checkpointed history (commands and signals) for a finished workflow. Its summary row and result are kept. Part of riverworkflow.HistoryStore.
func (*Store) RecordCancelled ¶
RecordCancelled marks the workflow cancelled. A no-op once finished.
func (*Store) RecordCompleted ¶
RecordCompleted marks the workflow completed with its result.
func (*Store) RecordFailed ¶
RecordFailed marks the workflow failed with a message.
func (*Store) RecordStarted ¶
RecordStarted upserts the workflow as running. It never resurrects a finished workflow (guarded by finished_at IS NULL) and never regresses to an older generation (guarded by EXCLUDED.gen >= w.gen), so a retried older generation can't overwrite the live one's run job.
func (*Store) Start ¶
Start begins the completion listener and its backstop sweep, enabling push-based Await. It holds one dedicated pool connection for LISTEN. Call it after Migrate; call Stop to release. Safe to call once; a second call is a no-op. Await works without Start (it falls back to polling), but Start makes it push-based and cheap at scale.
func (*Store) Stop ¶
func (s *Store) Stop()
Stop halts the listener and sweep and waits for them to exit. Safe to call more than once. It does not close the pool (which the store borrows).
func (*Store) WakeEvents ¶
WakeEvents returns the channel of cross-process wake notifications. Part of riverworkflow.WakeStore.
Source Files