Versions in this module Expand all Collapse all v0 v0.3.3 Aug 12, 2026 Changes in this version type Command + func (c Command[A, R]) Queue() string v0.3.2 Aug 12, 2026 Changes in this version + func GetResult[A, R any](ctx context.Context, c Client, id RunID, key string, cmd Command[A, R]) (R, bool, error) v0.3.1 Aug 12, 2026 v0.3.0 Aug 12, 2026 Changes in this version + const RunStatusCancelled + const RunStatusExpired + const RunStatusFailed + const RunStatusFailing + const RunStatusRunning + const RunStatusSucceeded + func CancelRun(ctx context.Context, c Client, id RunID, reason string) error type Command + func (cmd Command[A, R]) Enqueue(ctx context.Context, client Client, key string, args A, opts ...RunOption) (Run, error) + func (cmd Command[A, R]) ReplaceCurrentRun(ctx context.Context, client Client, expected RunID, key string, args A, ...) (ReplaceRunResult, error) type CommandInfo + RunID RunID + RunKey string type HistoryEntry + RunID RunID type HistoryKind + const HistoryRunFailing + const HistoryRunStarted type KeyedHistoryEntry + RunKey string type LiveWork + RunID RunID + RunKey string + RunStatus RunStatus type Node + func Enqueue[W, A, R any](work *Work[W], key string, cmd Command[A, R], args A) *Node type Observation + RunID RunID type ObservationKind + const ObservationRun type Option + func WithMaxCommandsPerRun(max int) Option + type ReplaceRunResult struct + Replaced bool + Run Run + type Run struct + CommandCount int + Created bool + CreatedAt time.Time + DeadlineAt *time.Time + FailFast bool + Failure *Failure + FinishedAt *time.Time + ID RunID + Key string + MaxCommands int + Metadata json.RawMessage + OpenCommands int + RootCommandID CommandID + Status RunStatus + StatusAt time.Time + Type string + UpdatedAt time.Time + Version int + func AwaitRun(ctx context.Context, c Client, id RunID) (Run, error) + func GetCurrentRun(ctx context.Context, c Client, typ, key string) (Run, bool, error) + func GetRun(ctx context.Context, c Client, id RunID) (Run, error) + type RunFilter struct + CreatedAfter *time.Time + CreatedBefore *time.Time + Cursor string + KeyPrefix string + Metadata map[string]string + PageSize int + Statuses []RunStatus + Type string + type RunID string + type RunOption interface + func WithRunDeadline(deadline time.Duration) RunOption + func WithoutRunDeadline() RunOption + type RunPage struct + NextCursor string + Runs []Run + func ListRuns(ctx context.Context, c Client, filter RunFilter) (RunPage, error) + type RunStatus string + type RunTrace struct + Commands []TraceCommand + Events []TraceEvent + History []HistoryEntry + Run Run + type TransactionClient struct + func (c *TransactionClient) BeginApplicationWrites() error v0.2.0 Aug 10, 2026 v0.1.0 Aug 10, 2026 Changes in this version + const CommandStatusCancelled + const CommandStatusExpired + const CommandStatusFailed + const CommandStatusPending + const CommandStatusReady + const CommandStatusRetryWait + const CommandStatusRunning + const CommandStatusSucceeded + const DefaultReadPageSize + const ExecutionStatusCancelled + const ExecutionStatusExpired + const ExecutionStatusFailed + const ExecutionStatusFailing + const ExecutionStatusRunning + const ExecutionStatusSucceeded + const KeyScopeLive + const KeyScopePermanent + const MaxReadKeys + const MaxReadPageSize + const QueueStateReady + const QueueStateRetryWait + const QueueStateRunning + const StatusCancelled + const StatusExpired + const StatusFailed + const StatusSucceeded + const TerminalStatusCancelled + const TerminalStatusExpired + const TerminalStatusFailed + const TerminalStatusSucceeded + var ErrClosed = flowerr.ErrClosed + var ErrConflict = flowerr.ErrConflict + var ErrInvalid = flowerr.ErrInvalid + var ErrInvalidState = flowerr.ErrInvalidState + var ErrLeaseLost = flowerr.ErrLeaseLost + var ErrNotFound = flowerr.ErrNotFound + var ErrPayloadTooLarge = flowerr.ErrPayloadTooLarge + var ErrSchema = flowerr.ErrSchema + var ErrTerminal = flowerr.ErrTerminal + func CancelCommand(ctx context.Context, c Client, id CommandID, reason string) error + func CancelExecution(ctx context.Context, c Client, id ExecutionID, reason string) error + func Emit[W, T any](work *Work[W], event Event[T], key string, payload T) error + func GetEventValue[W, T any](work *Work[W], event Event[T], key string) (T, error) + func Migrate(ctx context.Context, db *pgkit.DB, opts ...MigrateOption) error + func MigrationFS(opts ...MigrateOption) (fs.FS, error) + func Permanent(err error) error + func ResultOf[A, R any](trace ExecutionTrace, key string, cmd Command[A, R]) (R, error) + func RetryAfter(delay time.Duration, err error) error + func WithSchema(schema string) schemaOption + type AttemptID string + type Client interface + type Command struct + func DefineCommand[A, R any](name string, version int, opts ...CommandOption) Command[A, R] + func (c Command[A, R]) Name() string + func (c Command[A, R]) Version() int + func (c Command[A, R]) With(client Client) Command[A, R] + func (cmd Command[A, R]) Execute(ctx context.Context, key string, args A, opts ...ExecutionOption) (Execution, error) + type CommandFailure = Failure + type CommandID string + type CommandInfo struct + Attempt int + AttemptStartedAt time.Time + BudgetStartedAt time.Time + CommandID CommandID + CommandKey string + CreatedAt time.Time + ExecutionID ExecutionID + Name string + Version int + type CommandOption interface + func WithQueue(queue string) CommandOption + func WithRetry(policy RetryPolicy) CommandOption + func WithTimeout(timeout time.Duration) CommandOption + type CommandStatus string + type Commit struct + Args A + Info CommandInfo + Result R + type Error struct + Category error + ID string + Op string + Reason string + Resource string + func (e *Error) Error() string + func (e *Error) Unwrap() error + type Event struct + func DefineEvent[T any](name string) Event[T] + func (e Event[T]) Name() string + func (event Event[T]) Deliver(ctx context.Context, client Client, target ExecutionID, key string, payload T) error + func (event Event[T]) Emit(ctx context.Context, c Client, id ExecutionID, key string, payload T) error + type EventID string + type EventRef interface + type Execution struct + CommandCount int + Created bool + CreatedAt time.Time + DeadlineAt *time.Time + FailFast bool + Failure *Failure + FinishedAt *time.Time + ID ExecutionID + Key string + MaxCommands int + Metadata json.RawMessage + OpenCommands int + RootCommandID CommandID + Status ExecutionStatus + StatusAt time.Time + Type string + UpdatedAt time.Time + Version int + func AwaitExecution(ctx context.Context, c Client, id ExecutionID) (Execution, error) + func GetExecution(ctx context.Context, c Client, id ExecutionID) (Execution, error) + func LookupLiveExecution(ctx context.Context, c Client, typ, key string) (Execution, bool, error) + type ExecutionFilter struct + CreatedAfter *time.Time + CreatedBefore *time.Time + Cursor string + KeyPrefix string + Metadata map[string]string + PageSize int + Statuses []ExecutionStatus + Type string + type ExecutionID string + type ExecutionOption interface + func WaitFor(event EventRef, key string) ExecutionOption + func WithExecutionDeadline(deadline time.Duration) ExecutionOption + func WithFailFast(enabled bool) ExecutionOption + func WithLiveKey() ExecutionOption + func WithMetadata(metadata map[string]string) ExecutionOption + func WithStartDelay(delay time.Duration) ExecutionOption + func Within(duration time.Duration) ExecutionOption + func WithoutExecutionDeadline() ExecutionOption + type ExecutionPage struct + Executions []Execution + NextCursor string + func ListExecutions(ctx context.Context, c Client, filter ExecutionFilter) (ExecutionPage, error) + type ExecutionStatus string + type ExecutionTrace struct + Commands []TraceCommand + Events []TraceEvent + Execution Execution + History []HistoryEntry + func Trace(ctx context.Context, c Client, id ExecutionID, opts ...TraceOption) (ExecutionTrace, error) + type Failure = failure.Value + type HistoryEntry struct + AttemptID AttemptID + Body json.RawMessage + BodyHash string + CausationPosition *JournalPosition + CommandID CommandID + EntryID JournalEntryID + EventClass string + EventID EventID + EventKey string + EventName string + EventNamespace string + ExecutionID ExecutionID + Kind HistoryKind + Position JournalPosition + RecordedAt time.Time + TerminalStatus TerminalStatus + func History(ctx context.Context, c Client, id ExecutionID, opts ...HistoryOption) ([]HistoryEntry, error) + type HistoryKind string + const HistoryAttemptConcluded + const HistoryAttemptStarted + const HistoryCommandCreated + const HistoryEventRecorded + const HistoryExecutionFailing + const HistoryExecutionStarted + type HistoryOption interface + func HistoryAfter(position JournalPosition) HistoryOption + func HistoryLimit(limit int) HistoryOption + type JournalEntryID string + type JournalPosition uint64 + type KeyScope string + type KeyedHistoryEntry struct + DefinitionName string + ExecutionKey string + KeyScope KeyScope + type KeyedHistoryFilter struct + Cursor string + Keys []string + PageSize int + type KeyedHistoryPage struct + Entries []KeyedHistoryEntry + NextCursor string + func ListHistoryByKeys(ctx context.Context, c Client, filter KeyedHistoryFilter) (KeyedHistoryPage, error) + type LiveWork struct + AttemptOrdinal int + CommandCreatedAt time.Time + CommandID CommandID + CommandKey string + CommandName string + DefinitionName string + ExecutionID ExecutionID + ExecutionKey string + ExecutionStatus ExecutionStatus + KeyScope KeyScope + LeaseExpiresAt *time.Time + LeaseOwner string + NextRunAt time.Time + Queue string + QueueState QueueState + type LiveWorkFilter struct + Cursor string + Keys []string + PageSize int + type LiveWorkPage struct + NextCursor string + Work []LiveWork + func ListLiveWork(ctx context.Context, c Client, filter LiveWorkFilter) (LiveWorkPage, error) + type MigrateOption interface + type Node struct + func Execute[W, A, R any](work *Work[W], key string, cmd Command[A, R], args A) *Node + func (node *Node) Delay(duration time.Duration) *Node + func (node *Node) Key() string + func (node *Node) Optional() *Node + func (node *Node) WaitFor(event EventRef, key string) *Node + func (node *Node) Within(duration time.Duration) *Node + type None = struct + type NopObserver struct + func (NopObserver) Observe(context.Context, Observation) + type Observation struct + CommandID CommandID + CommandKey string + Count int64 + Duration time.Duration + ExecutionID ExecutionID + Kind ObservationKind + Name string + OccurredAt time.Time + Operation string + Outcome string + Queue string + Version int + Worker string + type ObservationKind string + const ObservationAttempt + const ObservationClaim + const ObservationCommand + const ObservationEvent + const ObservationExecution + const ObservationLease + const ObservationRuntime + const ObservationWait + type Observer interface + Observe func(context.Context, Observation) + type Option interface + func WithMaxCommandsPerExecution(max int) Option + func WithNotifications(enabled bool) Option + func WithObserver(observer Observer) Option + func WithPollInterval(interval time.Duration) Option + func WithQueueConcurrency(queue string, concurrency int) Option + func WithShutdownGrace(grace time.Duration) Option + func WithWorkerConcurrency(concurrency int) Option + type QueueDepth struct + Delayed int64 + OldestReadyFor time.Duration + Queue string + Ready int64 + Running int64 + func GetQueueDepth(ctx context.Context, c Client, queue string) (QueueDepth, error) + type QueueState string + type Registration interface + func Handle[A, R any](cmd Command[A, R], worker func(context.Context, *Work[A]) (R, error), ...) Registration + type RetryPolicy = retrypolicy.PublicPolicy + func Attempts(max int) RetryPolicy + func RetryFor(maxElapsed time.Duration) RetryPolicy + type Runtime struct + func New(db *pgkit.DB, opts ...Option) (*Runtime, error) + func (r *Runtime) InTx(tx pgx.Tx) Client + func (r *Runtime) Register(definitions ...Registration) error + func (r *Runtime) Run(ctx context.Context) error + func (r *Runtime) Stop(ctx context.Context) error + type SchemaStatus struct + AppliedAt time.Time + Compatible bool + CurrentVersion int + MinReaderVersion int + MinWriterVersion int + Schema string + func CheckSchema(ctx context.Context, db *pgkit.DB, opts ...MigrateOption) (SchemaStatus, error) + type TerminalStatus string + type TraceAttempt struct + Attempt int + Classification string + ConsumedAttempts int + ConsumedBudget bool + Failure *Failure + FinishedAt *time.Time + ID AttemptID + NextAttemptAt *time.Time + StartedAt time.Time + Worker string + type TraceCommand struct + Args json.RawMessage + AttemptOrdinal int + Attempts []TraceAttempt + BudgetStartedAt *time.Time + ConsumedAttempts int + CreatedAt time.Time + CreatedPosition JournalPosition + DeliveryState QueueState + Failure *Failure + FinishedAt *time.Time + ID CommandID + InitialDelay time.Duration + Key string + LastError *Failure + LeaseExpiresAt *time.Time + LeaseOwner string + LeaseStartedAt *time.Time + Name string + NextAttemptAt *time.Time + ParentCommandID CommandID + Queue string + Required bool + Result json.RawMessage + State CommandStatus + StatusAt time.Time + TerminalPosition *JournalPosition + UnsatisfiedWaits int + UpdatedAt time.Time + Version int + WaitDeadlineAt *time.Time + WaitStartedAt *time.Time + Waits []TraceEventWait + Within time.Duration + type TraceEvent struct + Body json.RawMessage + CausationPosition *JournalPosition + Class string + CommandID CommandID + ID EventID + Key string + Name string + Namespace string + Position JournalPosition + RecordedAt time.Time + TerminalStatus TerminalStatus + type TraceEventWait struct + Key string + Name string + SatisfiedPosition *JournalPosition + type TraceOption interface + type Tx interface + Exec func(context.Context, string, ...any) (pgconn.CommandTag, error) + Query func(context.Context, string, ...any) (pgx.Rows, error) + QueryRow func(context.Context, string, ...any) pgx.Row + type Work struct + Args A + func (w *Work[A]) Info() CommandInfo + type WorkerOption interface + func WithCommit[A, R any](fn func(context.Context, Tx, Commit[A, R]) error) WorkerOption[A, R]
Go to main page