flow

package module

Go to main page

Versions in this module

v0

v0.4.2

Aug 13, 2026

v0.4.1

Aug 12, 2026

Changes in this version

+ func NoRetry(err error) error

type CommandOption

+ func WithRecoveryLease(lease time.Duration) CommandOption

v0.4.0

Aug 12, 2026

Changes in this version

+ func GetQueueStats(ctx context.Context, c Client, queues ...string) (map[string]QueueStats, error)

+ type ActiveCommand struct

+ AttemptOrdinal int

+ CommandCreatedAt time.Time

+ CommandID CommandID

+ CommandKey string

+ CommandName string

+ DefinitionName string

+ KeyScope KeyScope

+ LeaseExpiresAt *time.Time

+ LeaseOwner string

+ NextRunAt time.Time

+ Queue string

+ QueueState QueueState

+ RunID RunID

+ RunKey string

+ RunStatus RunStatus

+ type ActiveCommandFilter struct

+ Cursor string

+ Keys []string

+ PageSize int

+ type ActiveCommandPage struct

+ Commands []ActiveCommand

+ NextCursor string

+ func ListActiveCommands(ctx context.Context, c Client, filter ActiveCommandFilter) (ActiveCommandPage, error)

type Command

+ func (cmd Command[A, R]) GetCurrentRun(ctx context.Context, c Client, key string) (Run, bool, error)

+ func (cmd Command[A, R]) GetResult(ctx context.Context, c Client, id RunID, key string) (R, bool, error)

+ type EnqueueResult struct

+ Created bool

+ RunID RunID

type KeyedHistoryPage

+ func ListHistoryByRunKeys(ctx context.Context, c Client, filter KeyedHistoryFilter) (KeyedHistoryPage, error)

+ type PruneResult struct

+ Commands int64

+ JournalEntries int64

+ Runs int64

+ func PruneTerminalRuns(ctx context.Context, runtime *Runtime, finishedBefore time.Time, limit int) (PruneResult, error)

+ type QueueStats struct

+ Delayed int64

+ OldestReadyFor time.Duration

+ Queue string

+ Ready int64

+ Running int64

type ReplaceRunResult

+ RunID RunID

type Run

+ RootCommandName string

+ RootCommandVersion int

+ RunKey string

type RunFilter

+ RootCommandName string

+ type StagedCommand struct

+ func (node *StagedCommand) Delay(duration time.Duration) *StagedCommand

+ func (node *StagedCommand) Key() string

+ func (node *StagedCommand) WaitFor(event EventRef, key string) *StagedCommand

+ func (node *StagedCommand) Within(duration time.Duration) *StagedCommand

type TraceCommand

+ Status CommandStatus

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]

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL