Documentation
¶
Index ¶
- Constants
- Variables
- func Migrate(ctx context.Context, db *sql.DB) error
- type ClaimStrategy
- type ClaimedRun
- type Store
- func (t *Store) AdvanceSkippedTask(ctx context.Context, runID uuid.UUID, seq int16, claimID uuid.UUID) (bool, error)
- func (t *Store) ClaimRunByID(ctx context.Context, id uuid.UUID) (*ClaimedRun, bool, error)
- func (t *Store) ClaimRuns(ctx context.Context, strategy ClaimStrategy, names []string, limit int) ([]*ClaimedRun, error)
- func (t *Store) CompleteRun(ctx context.Context, runID uuid.UUID, claimID uuid.UUID) (bool, error)
- func (t *Store) GetRun(ctx context.Context, id uuid.UUID) (*WorkflowRun, error)
- func (t *Store) GetTasks(ctx context.Context, runID uuid.UUID) ([]*TaskRun, error)
- func (t *Store) Heartbeat(ctx context.Context, runID uuid.UUID, claimID uuid.UUID) (bool, error)
- func (t *Store) InsertRun(ctx context.Context, run *WorkflowRun, tasks []*TaskRun) error
- func (t *Store) MarkTaskRunning(ctx context.Context, runID uuid.UUID, seq int16, claimID uuid.UUID) (bool, error)
- func (t *Store) OldestPending(ctx context.Context) (*time.Time, error)
- func (t *Store) ParkRunAfterAttempt(ctx context.Context, runID uuid.UUID, seq int16, failure TaskFailure, ...) (bool, error)
- func (t *Store) ParkRunWithoutAttempt(ctx context.Context, runID uuid.UUID, seq int16, failure TaskFailure, ...) (bool, error)
- func (t *Store) PersistTaskFailure(ctx context.Context, runID uuid.UUID, seq int16, failure TaskFailure, ...) (bool, error)
- func (t *Store) PersistTaskSuccess(ctx context.Context, runID uuid.UUID, seq int16, output json.RawMessage, ...) (bool, error)
- func (t *Store) ReleaseClaim(ctx context.Context, runID uuid.UUID, seq int16, claimID uuid.UUID) error
- func (t *Store) SchemaVersion(ctx context.Context) (int64, error)
- func (t *Store) SweepStuck(ctx context.Context, maxAge time.Duration) (int64, error)
- func (t *Store) UpsertRun(ctx context.Context, run *WorkflowRun, tasks []*TaskRun) (uuid.UUID, bool, error)
- func (t *Store) WithTx(tx pgx.Tx) *Store
- type TaskFailure
- type TaskRun
- type TaskStatus
- type WorkflowRun
- type WorkflowStatus
Constants ¶
const ( VersionTable = "workflows_goose_db_version" SchemaVersion int64 = 1 )
const ( TableRuns = "workflow_runs" TableTasks = "workflow_tasks" )
Variables ¶
var ErrSchemaOutdated = errors.New("workflow store: schema outdated")
ErrSchemaOutdated reports a database whose store chain is behind SchemaVersion.
Functions ¶
Types ¶
type ClaimStrategy ¶
type ClaimStrategy int
ClaimStrategy selects the order runs are claimed in. The zero value is unset, so cmp.Or can fill in a default.
const (
ClaimStrategyFIFO ClaimStrategy = iota + 1
)
type ClaimedRun ¶
type ClaimedRun struct {
Run *WorkflowRun
Tasks []*TaskRun
ClaimID uuid.UUID
}
ClaimedRun is a run claimed by one executor. Every later write to the run must pass ClaimID, so a write from an executor that lost the claim matches no row.
type Store ¶
Store provides typed access to the workflow_runs and workflow_tasks tables.
func (*Store) AdvanceSkippedTask ¶
func (t *Store) AdvanceSkippedTask(ctx context.Context, runID uuid.UUID, seq int16, claimID uuid.UUID) (bool, error)
AdvanceSkippedTask advances the run past a skipped task, returning false on a stale claim.
func (*Store) ClaimRunByID ¶
ClaimRunByID claims the run if it is due and unclaimed, and returns false when it is already claimed, not visible, or terminal.
func (*Store) ClaimRuns ¶
func (t *Store) ClaimRuns(ctx context.Context, strategy ClaimStrategy, names []string, limit int) ([]*ClaimedRun, error)
ClaimRuns claims up to limit due, unclaimed runs named in names, with their tasks. It opens its own transaction, so call it on the pool-bound Store.
func (*Store) CompleteRun ¶
CompleteRun marks the run succeeded, returning false on a stale claim.
func (*Store) Heartbeat ¶
Heartbeat refreshes claimed_at so SweepStuck skips a live run. On false or an error the caller must abort and release the claim.
func (*Store) MarkTaskRunning ¶
func (t *Store) MarkTaskRunning(ctx context.Context, runID uuid.UUID, seq int16, claimID uuid.UUID) (bool, error)
MarkTaskRunning moves the current task from pending to running, returning false on a stale claim or a task that is not pending. It stays cancellable: never mark running what ctx cannot run.
func (*Store) OldestPending ¶
OldestPending returns the oldest unclaimed run's next_retry_at, or nil when the backlog is empty.
func (*Store) ParkRunAfterAttempt ¶
func (t *Store) ParkRunAfterAttempt(ctx context.Context, runID uuid.UUID, seq int16, failure TaskFailure, claimID uuid.UUID) (bool, error)
ParkRunAfterAttempt marks the run and its task stuck and records the spent attempt. It returns false on a stale claim.
func (*Store) ParkRunWithoutAttempt ¶
func (t *Store) ParkRunWithoutAttempt(ctx context.Context, runID uuid.UUID, seq int16, failure TaskFailure, claimID uuid.UUID) (bool, error)
ParkRunWithoutAttempt parks the run like ParkRunAfterAttempt but records no attempt.
func (*Store) PersistTaskFailure ¶
func (t *Store) PersistTaskFailure(ctx context.Context, runID uuid.UUID, seq int16, failure TaskFailure, backoff time.Duration, release bool, claimID uuid.UUID) (bool, error)
PersistTaskFailure records a failure and backs off next_retry_at. With release it also clears the claim and resets the task to pending. It returns false with no write on a stale claim.
func (*Store) PersistTaskSuccess ¶
func (t *Store) PersistTaskSuccess(ctx context.Context, runID uuid.UUID, seq int16, output json.RawMessage, claimID uuid.UUID) (bool, error)
PersistTaskSuccess records the task's output and advances the run, returning false with no write when the caller has lost the claim.
func (*Store) ReleaseClaim ¶
func (t *Store) ReleaseClaim(ctx context.Context, runID uuid.UUID, seq int16, claimID uuid.UUID) error
ReleaseClaim hands the run back without spending an attempt. The task reset is guarded on status='running' so a release at a task boundary skips it.
func (*Store) SchemaVersion ¶
SchemaVersion returns the highest version the store chain has applied, or 0 when the chain has never run.
func (*Store) SweepStuck ¶
SweepStuck releases claims older than maxAge and returns how many. Only a task still 'running' is bumped and reset: without that join filter a skipped task would rerun and a pending one would take a premature attempt bump.
type TaskFailure ¶
TaskFailure is what one failed attempt records. Message must be PII-free.
type TaskRun ¶
type TaskRun struct {
RunID uuid.UUID `db:"run_id"`
// Seq is the task's ordinal within the run, not a retry counter.
Seq int16 `db:"seq"`
Handler string `db:"handler"`
Output json.RawMessage `db:"output"`
Status TaskStatus `db:"status"`
AttemptCount int `db:"attempt_count"`
LastError *string `db:"last_error"`
LastErrorCode *string `db:"last_error_code"`
Attempts json.RawMessage `db:"attempts"`
NextRetryAt time.Time `db:"next_retry_at"`
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
}
TaskRun is one snapshotted task row of a WorkflowRun.
type TaskStatus ¶
type TaskStatus int8
TaskStatus is a workflow task's persisted lifecycle state. The numeric values are wire format: append, never renumber.
const ( TaskStatusPending TaskStatus = 1 TaskStatusRunning TaskStatus = 2 TaskStatusSucceeded TaskStatus = 3 TaskStatusStuck TaskStatus = 4 TaskStatusSkipped TaskStatus = 5 )
func (TaskStatus) String ¶
func (s TaskStatus) String() string
String returns the status's durable lowercase name.
type WorkflowRun ¶
type WorkflowRun struct {
ID uuid.UUID `db:"id"`
Name string `db:"name"`
IdempotencyKey *string `db:"idempotency_key"`
Input json.RawMessage `db:"input"`
CallContext json.RawMessage `db:"call_context"`
Status WorkflowStatus `db:"status"`
CurrentTask int16 `db:"current_task"`
ClaimedAt *time.Time `db:"claimed_at"`
ClaimID *uuid.UUID `db:"claim_id"`
NextRetryAt time.Time `db:"next_retry_at"`
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
CompletedAt *time.Time `db:"completed_at"`
}
WorkflowRun is one row of a workflow run.
type WorkflowStatus ¶
type WorkflowStatus int8
WorkflowStatus is a workflow run's persisted lifecycle state. The numeric values are wire format: append, never renumber.
const ( WorkflowStatusRunning WorkflowStatus = 1 WorkflowStatusSucceeded WorkflowStatus = 2 WorkflowStatusStuck WorkflowStatus = 3 WorkflowStatusCancelled WorkflowStatus = 4 )
func (WorkflowStatus) String ¶
func (s WorkflowStatus) String() string
String returns the status's durable lowercase name.
Source Files