Documentation
¶
Overview ¶
Package postgres provides a Postgres-backed Store that implements the four interfaces the workflow engine and its worker need:
- github.com/deepnoodle-ai/workflow/experimental/worker.QueueStore — the run queue, claim, heartbeat, reaper, and terminal state.
- github.com/deepnoodle-ai/workflow.Checkpointer — via Store.NewCheckpointer, lease-fenced checkpoint persistence per claimed run.
- github.com/deepnoodle-ai/workflow.StepProgressStore — step progress observability.
- github.com/deepnoodle-ai/workflow.ActivityLogger — activity operation log.
The Store keeps one table for runs (workflow_runs), one for step progress (workflow_step_progress), and one for activity history (workflow_activity_log). Schema migrations are applied idempotently by Store.Migrate.
All writes to workflow_runs that belong to a running execution fence on (claimed_by, attempt) so that a worker that has lost its lease cannot corrupt a newer attempt's state. Fencing failures surface as github.com/deepnoodle-ai/workflow/experimental/worker.ErrLeaseLost.
Index ¶
- Constants
- Variables
- type Option
- type Run
- type RunCursor
- type RunFilter
- type Store
- func (s *Store) AppendEvent(ctx context.Context, event *worker.Event) error
- func (s *Store) Balance(ctx context.Context, orgID string) (int, error)
- func (s *Store) ClaimQueued(ctx context.Context, workerID string) (*worker.Claim, error)
- func (s *Store) CleanupEvents(ctx context.Context, olderThan time.Time) (int, error)
- func (s *Store) Complete(ctx context.Context, claim *worker.Claim, outcome worker.Outcome) error
- func (s *Store) CountRuns(ctx context.Context, orgID string, filter runquery.RunFilter) (int, error)
- func (s *Store) DeadLetterStale(ctx context.Context, staleBefore time.Time, maxAttempts int, ...) ([]worker.DeadLetteredRun, error)
- func (s *Store) Debit(ctx context.Context, orgID, runID, workflowType string, amount int) error
- func (s *Store) DeleteRun(ctx context.Context, orgID, id string) error
- func (s *Store) Enqueue(ctx context.Context, run worker.NewRun) error
- func (s *Store) EnqueueTx(ctx context.Context, tx pgx.Tx, run worker.NewRun) error
- func (s *Store) EnqueueWebhook(ctx context.Context, delivery *worker.WebhookDelivery) error
- func (s *Store) GetActivityHistory(ctx context.Context, executionID string) ([]*workflow.ActivityLogEntry, error)
- func (s *Store) GetRun(ctx context.Context, orgID, id string) (*runquery.Run, error)
- func (s *Store) GetStepProgress(ctx context.Context, executionID string) ([]workflow.StepProgress, error)
- func (s *Store) HasRefund(ctx context.Context, orgID, runID string) (bool, error)
- func (s *Store) Heartbeat(ctx context.Context, claim *worker.Claim) error
- func (s *Store) IncrementTriggerAttempts(ctx context.Context, id string, errMsg string) error
- func (s *Store) IncrementWebhookAttempts(ctx context.Context, id string, lastError string) error
- func (s *Store) InsertTriggers(ctx context.Context, triggers []worker.Trigger) error
- func (s *Store) ListEvents(ctx context.Context, runID string, afterSeq int64) ([]*worker.Event, error)
- func (s *Store) ListPendingTriggers(ctx context.Context, limit int) ([]worker.Trigger, error)
- func (s *Store) ListPendingWebhooks(ctx context.Context, limit int) ([]*worker.WebhookDelivery, error)
- func (s *Store) ListRefundPending(ctx context.Context, limit int) ([]worker.FailedRun, error)
- func (s *Store) ListRuns(ctx context.Context, orgID string, filter runquery.RunFilter) ([]*runquery.Run, *runquery.RunCursor, error)
- func (s *Store) LogActivity(ctx context.Context, entry *workflow.ActivityLogEntry) error
- func (s *Store) MarkTriggerCompleted(ctx context.Context, id string, childRunID string) error
- func (s *Store) MarkTriggerFailed(ctx context.Context, id string, errMsg string) error
- func (s *Store) MarkTriggerProcessing(ctx context.Context, id string) error
- func (s *Store) MarkWebhookDelivered(ctx context.Context, id string) error
- func (s *Store) MarkWebhookFailed(ctx context.Context, id string, errMsg string) error
- func (s *Store) MarkWebhookProcessing(ctx context.Context, id string) error
- func (s *Store) Migrate(ctx context.Context) error
- func (s *Store) NewActivityLogger(_ *worker.Claim) workflow.ActivityLogger
- func (s *Store) NewCheckpointer(claim *worker.Claim) workflow.Checkpointer
- func (s *Store) NewStepProgressStore(_ *worker.Claim) workflow.StepProgressStore
- func (s *Store) Pool() *pgxpool.Pool
- func (s *Store) ReclaimStale(ctx context.Context, staleBefore time.Time, maxAttempts int, ...) (int, error)
- func (s *Store) Refund(ctx context.Context, orgID, runID, workflowType string, amount int) error
- func (s *Store) Schema() string
- func (s *Store) UpdateRunSpec(ctx context.Context, claim *worker.Claim, spec []byte) error
- func (s *Store) UpdateStepProgress(ctx context.Context, executionID string, p workflow.StepProgress) error
Constants ¶
const DefaultSchema = "public"
DefaultSchema is the Postgres schema used when WithSchema is not supplied. Matches the behavior of the default search_path on a fresh Postgres install.
Variables ¶
var ErrCannotDeleteRunning = runquery.ErrCannotDeleteRunning
ErrCannotDeleteRunning is an alias for runquery.ErrCannotDeleteRunning.
var ErrRunNotFound = runquery.ErrRunNotFound
ErrRunNotFound is an alias for runquery.ErrRunNotFound so existing callers comparing against postgres.ErrRunNotFound keep working. New code should use runquery.ErrRunNotFound directly.
Functions ¶
This section is empty.
Types ¶
type Option ¶
type Option func(*Store)
Option configures a Store.
func WithLogger ¶
WithLogger attaches a structured logger. Defaults to a discard logger.
func WithSchema ¶
WithSchema selects the Postgres schema (namespace) that will hold the store's tables. Defaults to "public". The schema name is validated as a simple SQL identifier (letters, digits, underscore, starting with a letter or underscore) to rule out injection, and then used verbatim as a quoted identifier in every query.
Migrate will run `CREATE SCHEMA IF NOT EXISTS` on the selected schema before creating tables, so the schema does not need to exist in advance.
type Run ¶
Run aliases runquery.Run so callers can write postgres.Run during the transition. New code should use runquery.Run.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store is a Postgres-backed implementation of the worker QueueStore and the workflow engine's persistence interfaces. Construct with New and call Migrate once on startup to ensure the schema is in place.
func New ¶
New constructs a Store bound to the given pgx pool. The pool's lifecycle is owned by the caller. Panics if pool is nil.
func (*Store) AppendEvent ¶
AppendEvent implements worker.EventStore.
func (*Store) ClaimQueued ¶
ClaimQueued implements worker.QueueStore using SELECT ... FOR UPDATE SKIP LOCKED to atomically claim the oldest queued run.
func (*Store) CleanupEvents ¶
CleanupEvents implements worker.EventStore.
func (*Store) CountRuns ¶
func (s *Store) CountRuns(ctx context.Context, orgID string, filter runquery.RunFilter) (int, error)
CountRuns returns the total number of rows matching filter. The cursor field on filter is ignored: counts are over the entire filtered set, not a single page.
func (*Store) DeadLetterStale ¶
func (s *Store) DeadLetterStale(ctx context.Context, staleBefore time.Time, maxAttempts int, excludeIDs []string) ([]worker.DeadLetteredRun, error)
DeadLetterStale implements worker.QueueStore. Returns the metadata for each dead-lettered run so the worker can refund credits inline.
func (*Store) DeleteRun ¶
DeleteRun removes a run row by ID. Running runs cannot be deleted; the caller must cancel or wait for the run first.
Implemented as a single DELETE ... RETURNING status so the check and the delete happen atomically. If RETURNING yields no row, we fall back to a cheap existence probe to distinguish "running" from "not found."
func (*Store) Enqueue ¶
Enqueue implements worker.QueueStore. The insert runs in its own connection. When the insert must be atomic with writes to adjacent tables (credit ledger, idempotency keys, audit records, …), use EnqueueTx inside a caller-owned transaction instead.
func (*Store) EnqueueTx ¶
EnqueueTx inserts a queued run inside a caller-provided pgx transaction. The caller owns the tx lifecycle (Begin, Commit, Rollback). Use this when the run insert must be atomic with writes to tables outside the store's schema — e.g., debiting a credit ledger and creating the run in one commit.
The tx must be against the same database as the Store's pool; the library does not verify this.
func (*Store) EnqueueWebhook ¶
EnqueueWebhook implements worker.WebhookStore.
func (*Store) GetActivityHistory ¶
func (s *Store) GetActivityHistory(ctx context.Context, executionID string) ([]*workflow.ActivityLogEntry, error)
GetActivityHistory implements workflow.ActivityLogger.
func (*Store) GetRun ¶
GetRun returns a single run by ID, scoped to orgID. An empty orgID matches rows with NULL org_id (single-tenant). Returns runquery.ErrRunNotFound when no matching row exists.
func (*Store) GetStepProgress ¶
func (s *Store) GetStepProgress(ctx context.Context, executionID string) ([]workflow.StepProgress, error)
GetStepProgress returns every step progress row recorded for an execution, ordered by started_at (NULLS LAST) then step_name. One row per (step_name, branch_id). Returns an empty slice if no rows exist. Use this on the read side to render per-step status for a run whose identity came back from runquery.Store.GetRun, which intentionally does not carry step progress.
func (*Store) Heartbeat ¶
Heartbeat implements worker.QueueStore with (claimed_by, attempt) fencing. Rows with a status other than running, or a mismatched lease, produce ErrLeaseLost.
func (*Store) IncrementTriggerAttempts ¶
IncrementTriggerAttempts implements worker.TriggerStore.
func (*Store) IncrementWebhookAttempts ¶
IncrementWebhookAttempts implements worker.WebhookStore.
func (*Store) InsertTriggers ¶
InsertTriggers implements worker.TriggerStore.
func (*Store) ListEvents ¶
func (s *Store) ListEvents(ctx context.Context, runID string, afterSeq int64) ([]*worker.Event, error)
ListEvents implements worker.EventStore.
func (*Store) ListPendingTriggers ¶
ListPendingTriggers implements worker.TriggerStore.
func (*Store) ListPendingWebhooks ¶
func (s *Store) ListPendingWebhooks(ctx context.Context, limit int) ([]*worker.WebhookDelivery, error)
ListPendingWebhooks implements worker.WebhookStore.
func (*Store) ListRefundPending ¶
ListRefundPending implements worker.QueueStore by joining workflow_runs against the credit ledger: runs in StatusFailed with a matching debit but no matching refund.
func (*Store) ListRuns ¶
func (s *Store) ListRuns(ctx context.Context, orgID string, filter runquery.RunFilter) ([]*runquery.Run, *runquery.RunCursor, error)
ListRuns returns runs matching filter, ordered newest-first with keyset pagination. Returns the rows and a cursor to pass back on the next call; the cursor is nil when no more rows exist.
orgID == "" lists runs with NULL org_id (single-tenant). Pass a real org ID for scoped B2B listings.
func (*Store) LogActivity ¶
LogActivity implements workflow.ActivityLogger.
func (*Store) MarkTriggerCompleted ¶
MarkTriggerCompleted implements worker.TriggerStore.
func (*Store) MarkTriggerFailed ¶
MarkTriggerFailed implements worker.TriggerStore.
func (*Store) MarkTriggerProcessing ¶
MarkTriggerProcessing implements worker.TriggerStore. Uses a compare-and-swap on status to prevent multiple workers from processing the same trigger concurrently.
func (*Store) MarkWebhookDelivered ¶
MarkWebhookDelivered implements worker.WebhookStore.
func (*Store) MarkWebhookFailed ¶
MarkWebhookFailed implements worker.WebhookStore.
func (*Store) MarkWebhookProcessing ¶
MarkWebhookProcessing implements worker.WebhookStore with a compare-and-swap to prevent duplicate delivery.
func (*Store) Migrate ¶
Migrate applies the schema to the database. Idempotent: safe to call on every startup.
func (*Store) NewActivityLogger ¶
func (s *Store) NewActivityLogger(_ *worker.Claim) workflow.ActivityLogger
NewActivityLogger returns a workflow.ActivityLogger backed by this Store for the given claim. Activity log rows are append-only and not lease-fenced.
func (*Store) NewCheckpointer ¶
func (s *Store) NewCheckpointer(claim *worker.Claim) workflow.Checkpointer
NewCheckpointer returns a lease-fenced workflow.Checkpointer for the given claim. Writes fence on (claimed_by, attempt); a fencing failure returns worker.ErrLeaseLost from the SaveCheckpoint call.
Reads (LoadCheckpoint) are unfenced: a fresh attempt must be able to resume regardless of which worker originally wrote the snapshot.
func (*Store) NewStepProgressStore ¶
func (s *Store) NewStepProgressStore(_ *worker.Claim) workflow.StepProgressStore
NewStepProgressStore returns a workflow.StepProgressStore backed by this Store for the given claim. The current implementation ignores the claim (progress rows are not lease-fenced) but the signature matches HandlerStores so consumers can wire it directly into a worker.
func (*Store) Pool ¶
Pool returns the underlying pgxpool.Pool for queries the high-level API does not cover. Consumers are responsible for not breaking the store's invariants (lease fencing, status transitions, etc.).
func (*Store) ReclaimStale ¶
func (s *Store) ReclaimStale(ctx context.Context, staleBefore time.Time, maxAttempts int, excludeIDs []string) (int, error)
ReclaimStale implements worker.QueueStore.
func (*Store) UpdateRunSpec ¶
UpdateRunSpec replaces the spec on a running claim. It fences on (claim_id, worker_id, attempt) and status = running, and returns ErrLeaseLost if the fence fails — matching Heartbeat and Complete.
Use this during long-running activities that mutate the run spec incrementally (e.g., a KB-apply loop persisting progress between steps) and need the update durable without waiting for the next checkpoint. The caller retains responsibility for producing a valid spec; the store does not inspect it.
func (*Store) UpdateStepProgress ¶
func (s *Store) UpdateStepProgress(ctx context.Context, executionID string, p workflow.StepProgress) error
UpdateStepProgress implements workflow.StepProgressStore by upserting into workflow_step_progress. Keyed on (execution_id, step_name, branch_id) — a step running on two branches produces two rows.
Source Files