flow

package module
v0.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 10, 2026 License: MIT Imports: 38 Imported by: 0

README

flow

flow is a Go library for event-driven, durable, distributed work execution backed by PostgreSQL.

command -> worker -> result + events
                         |
                         +-> optional sub-commands

Commands are the only durable unit of orchestration. Workers perform typed work, emit immutable execution-scoped events, and stage bounded sub-commands. Exact event gates provide sequencing and joins. PostgreSQL stores the queue, leases, projections, and a gap-free journal for each execution.

Install

go get github.com/goware/flow

Flow uses the application's existing PostgreSQL database. Its six tables use a flow_ prefix and default to the public schema. flow.WithSchema selects another schema.

The v0.1 release line supports Go 1.26 and PostgreSQL 17 and 18. During v0.x, intentional Go API changes may be made with release notes. Published migration files are immutable: upgrades add forward migrations, and applications must run Migrate before starting a newer runtime. Back up durable data before upgrades.

Run migrations explicitly during deployment:

if err := flow.Migrate(ctx, db); err != nil {
	return err
}

flow.New validates the installed schema and starts no goroutines. Register workers, then call Runtime.Run in each process that should execute work.

A command

type emailArgs struct {
	To string `json:"to"`
}

type emailResult struct {
	MessageID string `json:"message_id"`
}

var sendEmail = flow.DefineCommand[emailArgs, emailResult]("mail.send", 1)

func sendEmailWorker(ctx context.Context, work *flow.Work[emailArgs]) (emailResult, error) {
	return emailResult{MessageID: "provider-123"}, nil
}

runtime, err := flow.New(db)
if err != nil {
	return err
}
if err := runtime.Register(flow.Handle(sendEmail, sendEmailWorker)); err != nil {
	return err
}
go runtime.Run(ctx)

exec, err := sendEmail.With(runtime).Execute(ctx, "email/order-42", emailArgs{
	To: "person@example.com",
})

Execute always creates or rediscovers durable asynchronous work; it never calls a worker inline. A stable non-empty execution key is permanently idempotent by default. flow.WithLiveKey() instead deduplicates only while an execution is non-terminal.

Execute returns the Execution snapshot as of durable acceptance; Created reports whether this call created it. GetExecution and AwaitExecution return the same type with the execution's current or final state.

Composing work

A successful worker may atomically emit events and stage sub-commands:

var charged = flow.DefineEvent[chargeResult]("billing.charged")

func chargeWorker(ctx context.Context, work *flow.Work[chargeArgs]) (chargeResult, error) {
	result := chargeResult{Receipt: "receipt-42"}
	if err := flow.Emit(work, charged, "charge/42", result); err != nil {
		return chargeResult{}, err
	}
	flow.Execute(work, "notify/42", notifyCustomer, notifyArgs{Receipt: result.Receipt})
	return result, nil
}

Repeated declarations with the same key and canonical content coalesce. Conflicting declarations poison the complete decision. A WithCommit callback can update application tables in the same fenced transaction as Flow settlement.

Choose command boundaries around independent retry, side effects, isolation, timeouts, queue ownership, or useful parallelism. Keep small deterministic transformations in the worker that owns them instead of turning every business logic microstep into durable work. Several small writes to the same PostgreSQL database can usually share one WithCommit callback. Keep that callback short and database-only: it holds the execution lock until settlement commits and is not an exactly-once boundary for remote calls.

One execution is one serialized semantic aggregate. Keep causally related work together, but use separate executions for independent bulk items or shards instead of treating one execution as a tenant-wide work container. The default 1,000-command ceiling is a safety limit, not a recommended execution size; ordinary executions are usually clearer in the tens or low hundreds. For a very large fan-out, have bounded batch commands declare later batches, and combine large input sets through hierarchical join commands rather than one enormous child declaration or join.

Exact event gates and inputs

A root or sub-command may wait for exact application events:

var approved = flow.DefineEvent[approval]("orders.approved")

flow.Execute(work, "fulfill/42", fulfill, args).
	WaitFor(approved, "approval/42").
	Within(30 * time.Minute).
	Delay(time.Second)

The waiting worker gets the value attached to a declared event:

value, err := flow.GetEventValue(work, approved, "approval/42")

Multiple waits are AND conditions. Matching is exact on event name and key within one execution. Events recorded before command declaration still satisfy the gate. Within starts at command creation and runs independently of Delay. At most 256 waits may be declared for one command; larger joins should use a tree of join commands or stable external references.

Pass data computed by a parent directly in child arguments. Use exact events for sibling, cross-branch, or external facts, and stage related events and children in the same worker decision when they belong to one atomic change. Large or sensitive documents should remain in application storage; pass stable references through command arguments or event payloads.

Flow has three event paths:

API Use
flow.Emit(work, ...) stage an event in the current execution with the worker decision
event.Emit(ctx, client, id, ...) record an external event in a known execution
event.Deliver(ctx, client, id, ...) deliberately record a detached event in another known execution, including from an active worker

Deliver needs only the target execution ID. With runtime.InTx(tx), it commits or rolls back with the caller's application writes; with a regular runtime client it commits independently. Keep caller-owned transactions short: an execution lock remains held until the caller commits or rolls back. A committed delivery survives source failure and retry, so producers should use stable keys and deterministic payloads. Same-execution worker events should use staged flow.Emit: explicitly delivering to the current execution is detached and may survive a failed attempt. Delivery is targeted ingress, not publish/subscribe, and target workers remain at-least-once.

Fan-out, fan-in, multi-stage joins, branches, and bounded loops are ordinary command composition. Flow intentionally has no separate coordinator/state-machine API, outcome subscriptions, OR/quorum/race gates, or automatic result dataflow.

Examples

Each example contains its complete, self-documenting logic:

  • examples/direct: one background command;
  • examples/fanout: two command-owned fan-out/join phases;
  • examples/monitor: a command gated by an externally published event;
  • examples/agent: a bounded self-composing command loop.

Run one against PostgreSQL:

FLOW_EXAMPLE_DATABASE_URL='postgres://postgres@localhost/postgres?sslmode=disable' \
  go run ./examples/direct

Operations

  • Claims match exact registered command name/version pairs. Unknown work remains durable until a compatible worker appears.
  • Workers are at-least-once at the application boundary; settlement is fenced and durable progression commits once. External effects still need stable idempotency keys.
  • Lease renewal is bounded and skip-locked: one busy settlement cannot block unrelated renewals. A locked row remains uncertain until settlement, a later renewal, or the conservative local-expiry watchdog resolves it.
  • Deadline, wait-expiry, and lease-recovery maintenance drains full progressing pages promptly but remains sequential and bounded; locked/no-op pages fall back to polling.
  • Required command failure enters reduced fail-fast by default. flow.WithFailFast(false) lets remaining work continue.
  • Execution deadlines, retries, queues, concurrency limits, graceful shutdown, polling, notification hints, observers, history, trace, cancellation, and caller-owned transactions are supported.
  • Publishers may use a Runtime without calling Run or registering workers. Worker pools may be deployed independently.
  • Observer delivery and shutdown drain are best-effort. Observers must return promptly and should honor context cancellation; observation loss never changes durable correctness.
  • Flow has no pruning or archival API. Journal, payload, and terminal execution data remain retained until an operator deliberately archives or removes them outside Flow's supported API.

For bounded domain-row decoration, ListLiveWork and ListHistoryByKeys accept at most 200 exact execution keys and return cursor pages of 100 rows by default (maximum 1,000). Ordinary pages are not a cross-page snapshot; use a Repeatable Read or Serializable caller transaction when one coherent snapshot is required. The same rule applies to caller-owned Trace; Flow-owned Trace uses Repeatable Read automatically.

Tests

The Makefile uses a local flow_test database and sets FLOW_TEST_DATABASE_URL explicitly, so PostgreSQL integration tests fail instead of being skipped when the database is unavailable:

make db-reset
make test

db-reset recreates the database and applies Flow's embedded migrations to the public schema. Individual integration tests continue to create and clean up isolated schemas inside that database. make test always enables Go's race detector.

The database connection can be customized with PG_HOST, PG_PORT, PG_USER, PG_DATABASE, and PGPASSWORD, or by setting FLOW_TEST_DATABASE_URL directly. make test-with-reset is available when a clean database and a test run are both wanted.

Documentation

Overview

Observer delivery is bounded and best-effort. Observers must return promptly and should honor context cancellation; a blocked or failed observer never changes durable execution correctness or prevents runtime shutdown.

External callers record execution-scoped events with Event.Emit. Event.Deliver provides deliberately detached ingress to a known execution, including from an active worker; passing Runtime.InTx joins it to caller-owned application writes. Same-execution worker events should normally use staged Emit instead.

Index

Constants

View Source 
const (
	// MaxReadKeys bounds every by-keys batch read before duplicate removal.
	MaxReadKeys = store.MaxReadKeys

	// DefaultReadPageSize is used when a by-key read filter has PageSize zero.
	DefaultReadPageSize = 100
	// MaxReadPageSize is the largest public by-key read page.
	MaxReadPageSize = 1000
)
View Source 
const (
	ExecutionStatusRunning   ExecutionStatus = "running"
	ExecutionStatusFailing   ExecutionStatus = "failing"
	ExecutionStatusSucceeded ExecutionStatus = "succeeded"
	ExecutionStatusFailed    ExecutionStatus = "failed"
	ExecutionStatusCancelled ExecutionStatus = "cancelled"
	ExecutionStatusExpired   ExecutionStatus = "expired"

	CommandStatusPending   CommandStatus = "pending"
	CommandStatusReady     CommandStatus = "ready"
	CommandStatusRunning   CommandStatus = "running"
	CommandStatusRetryWait CommandStatus = "retry_wait"
	CommandStatusSucceeded CommandStatus = "succeeded"
	CommandStatusFailed    CommandStatus = "failed"
	CommandStatusCancelled CommandStatus = "cancelled"
	CommandStatusExpired   CommandStatus = "expired"

	QueueStateReady     QueueState = "ready"
	QueueStateRetryWait QueueState = "retry_wait"
	QueueStateRunning   QueueState = "running"

	KeyScopePermanent KeyScope = "permanent"
	KeyScopeLive      KeyScope = "live"

	TerminalStatusSucceeded TerminalStatus = "succeeded"
	TerminalStatusFailed    TerminalStatus = "failed"
	TerminalStatusCancelled TerminalStatus = "cancelled"
	TerminalStatusExpired   TerminalStatus = "expired"

	// Short names remain source-compatible aliases for terminal command states.
	StatusSucceeded = CommandStatusSucceeded
	StatusFailed    = CommandStatusFailed
	StatusCancelled = CommandStatusCancelled
	StatusExpired   = CommandStatusExpired
)

Variables

View Source 
var (
	ErrNotFound        = flowerr.ErrNotFound
	ErrConflict        = flowerr.ErrConflict
	ErrInvalid         = flowerr.ErrInvalid
	ErrInvalidState    = flowerr.ErrInvalidState
	ErrTerminal        = flowerr.ErrTerminal
	ErrLeaseLost       = flowerr.ErrLeaseLost
	ErrPayloadTooLarge = flowerr.ErrPayloadTooLarge
	ErrClosed          = flowerr.ErrClosed
	ErrSchema          = flowerr.ErrSchema
)

Functions

func CancelCommand 

func CancelCommand(ctx context.Context, c Client, id CommandID, reason string) error

func CancelExecution 

func CancelExecution(ctx context.Context, c Client, id ExecutionID, reason string) error

func Emit 

func Emit[W, T any](work *Work[W], event Event[T], key string, payload T) error

Emit stages an application event in a worker decision. It performs no database work and becomes durable only when the enclosing decision settles successfully.

func GetEventValue 

func GetEventValue[W, T any](work *Work[W], event Event[T], key string) (T, error)

GetEventValue returns the typed value attached to an exact event gate declared by the current command. The value is already in memory when the worker starts; this function does not wait or query the database.

func Migrate 

func Migrate(ctx context.Context, db *pgkit.DB, opts ...MigrateOption) error

Migrate applies every unapplied embedded Flow migration in its own execution-serialized transaction and verifies all previously recorded checksums before writing.

func MigrationFS 

func MigrationFS(opts ...MigrateOption) (fs.FS, error)

MigrationFS returns schema-rendered SQL files for an external transactional migration runner. Each file records the same checksum and compatibility row as Migrate, so CheckSchema accepts either application path.

func Permanent 

func Permanent(err error) error

Permanent classifies an application error as terminal for the current command delivery.

func ResultOf 

func ResultOf[A, R any](trace ExecutionTrace, key string, cmd Command[A, R]) (R, error)

func RetryAfter 

func RetryAfter(delay time.Duration, err error) error

RetryAfter classifies an error as retryable after a requested delay. The command's immutable retry bounds still apply.

func WithSchema 

func WithSchema(schema string) schemaOption

WithSchema places Flow's fixed flow_ tables in a validated PostgreSQL schema. The default is public. Runtime options adopt the same value in the runtime phase; table names always retain their flow_ prefix.

Types

type AttemptID 

type AttemptID string

type Client 

type Client interface {
	// contains filtered or unexported methods
}

Client is a sealed capability implemented by Runtime and transaction-scoped clients. Applications pass it or bind it with Definition.With; they cannot construct another implementation that bypasses Flow's durable operations.

type Command 

type Command[A, R any] struct {
	// contains filtered or unexported fields
}

func DefineCommand 

func DefineCommand[A, R any](name string, version int, opts ...CommandOption) Command[A, R]

func (Command[A, R]) Execute 

func (cmd Command[A, R]) Execute(ctx context.Context, key string, args A, opts ...ExecutionOption) (Execution, error)

func (Command[A, R]) Name 

func (c Command[A, R]) Name() string

func (Command[A, R]) Version 

func (c Command[A, R]) Version() int

func (Command[A, R]) With 

func (c Command[A, R]) With(client Client) Command[A, R]

type CommandFailure 

type CommandFailure = Failure

type CommandID 

type CommandID string

type CommandInfo 

type CommandInfo struct {
	ExecutionID ExecutionID
	CommandID   CommandID
	CommandKey  string
	Name        string
	Version     int

	CreatedAt        time.Time
	BudgetStartedAt  time.Time
	Attempt          int
	AttemptStartedAt time.Time
}

type CommandOption 

type CommandOption interface {
	// contains filtered or unexported methods
}

func WithQueue 

func WithQueue(queue string) CommandOption

func WithRetry 

func WithRetry(policy RetryPolicy) CommandOption

func WithTimeout 

func WithTimeout(timeout time.Duration) CommandOption

type CommandStatus 

type CommandStatus string

type Commit 

type Commit[A, R any] struct {
	Args   A
	Result R
	Info   CommandInfo
}

type Error 

type Error struct {
	Category error
	Op       string
	Resource string
	ID       string
	Reason   string
}

Error adds safe structured context to a sentinel category. Its fields must contain identifiers and bounded reasons only, never payloads, SQL, secrets, or lease tokens.

func (*Error) Error 

func (e *Error) Error() string

func (*Error) Unwrap 

func (e *Error) Unwrap() error

type Event 

type Event[T any] struct {
	// contains filtered or unexported fields
}

func DefineEvent 

func DefineEvent[T any](name string) Event[T]

func (Event[T]) Deliver 

func (event Event[T]) Deliver(ctx context.Context, client Client, target ExecutionID, key string, payload T) error

Deliver records an event in a known execution, including from inside an active worker attempt. Delivery is detached from that attempt: pass a Runtime.InTx client to join a caller-owned transaction. Once committed, the event is not retracted if the source attempt fails or retries; equivalent repeats retain ordinary event idempotency. Use Emit(work, ...) for same-execution events that must settle atomically with the worker decision.

func (Event[T]) Emit 

func (event Event[T]) Emit(ctx context.Context, c Client, id ExecutionID, key string, payload T) error

func (Event[T]) Name 

func (e Event[T]) Name() string

type EventID 

type EventID string

type EventRef 

type EventRef interface {
	// contains filtered or unexported methods
}

type Execution 

type Execution struct {
	ID            ExecutionID
	Type          string
	Version       int
	Key           string
	RootCommandID CommandID
	Status        ExecutionStatus
	FailFast      bool
	MaxCommands   int
	CommandCount  int
	OpenCommands  int
	DeadlineAt    *time.Time
	Failure       *Failure
	CreatedAt     time.Time
	UpdatedAt     time.Time
	StatusAt      time.Time
	FinishedAt    *time.Time
	Metadata      json.RawMessage
	Created       bool
}

Execution is a durable execution state snapshot. Execute returns the snapshot as of durable acceptance; GetExecution, AwaitExecution, and other inspection reads return the current or final state. Created reports whether the producing Execute call created the execution; it is false for an idempotent rediscovery and always false on inspection reads.

func AwaitExecution 

func AwaitExecution(ctx context.Context, c Client, id ExecutionID) (Execution, error)

AwaitExecution polls the durable execution row until it reaches a terminal state or ctx ends. It consumes no command worker, connection while waiting, or durable lease.

func GetExecution 

func GetExecution(ctx context.Context, c Client, id ExecutionID) (Execution, error)

func LookupLiveExecution 

func LookupLiveExecution(ctx context.Context, c Client, typ, key string) (Execution, bool, error)

LookupLiveExecution finds the one non-terminal execution currently holding a live-scoped key for the definition, if any. Live keys admit many settled executions per key over time but at most one live holder; this is the lookup that matches that invariant. found=false means no live holder — settled executions with the key may still exist.

type ExecutionFilter 

type ExecutionFilter struct {
	Type          string
	KeyPrefix     string
	Statuses      []ExecutionStatus
	CreatedAfter  *time.Time
	CreatedBefore *time.Time
	Metadata      map[string]string
	PageSize      int
	Cursor        string
}

ExecutionFilter is the bounded, indexed filter supported by ListExecutions. CreatedBefore is exclusive and CreatedAfter is inclusive.

type ExecutionID 

type ExecutionID string

type ExecutionOption 

type ExecutionOption interface {
	// contains filtered or unexported methods
}

ExecutionOption is a sealed command execution option.

func WaitFor 

func WaitFor(event EventRef, key string) ExecutionOption

WaitFor gates a root command on one exact application event inside the execution it creates. Multiple waits are AND conditions. Worker decisions use the matching Node.WaitFor method.

func WithExecutionDeadline 

func WithExecutionDeadline(deadline time.Duration) ExecutionOption

func WithFailFast 

func WithFailFast(enabled bool) ExecutionOption

func WithLiveKey 

func WithLiveKey() ExecutionOption

WithLiveKey scopes the execution key to live executions: while a running or failing execution holds the key, Execute rediscovers it without comparing start identity — a silent, queue-style dedupe no-op — and once that execution reaches a terminal status the key is released for a new start. Live-keyed starts therefore give at-most-one live execution per key, not at-most-one execution ever. Requires a non-empty key.

func WithMetadata 

func WithMetadata(metadata map[string]string) ExecutionOption

func WithStartDelay 

func WithStartDelay(delay time.Duration) ExecutionOption

WithStartDelay schedules an execution's root command to become deliverable after the delay instead of immediately, mirroring Delay for sub-commands.

func Within 

func Within(duration time.Duration) ExecutionOption

Within bounds how long a direct root command waits for its exact events. It is valid only when the same start declares at least one WaitFor option.

func WithoutExecutionDeadline 

func WithoutExecutionDeadline() ExecutionOption

type ExecutionPage 

type ExecutionPage struct {
	Executions []Execution
	NextCursor string
}

func ListExecutions 

func ListExecutions(ctx context.Context, c Client, filter ExecutionFilter) (ExecutionPage, error)

type ExecutionStatus 

type ExecutionStatus string

type ExecutionTrace 

type ExecutionTrace struct {
	Execution Execution
	Commands  []TraceCommand
	Events    []TraceEvent
	History   []HistoryEntry
}

func Trace 

func Trace(ctx context.Context, c Client, id ExecutionID, opts ...TraceOption) (ExecutionTrace, error)

Trace reconstructs one execution and overlays its current operational projections. A Runtime client gets one Flow-owned Repeatable Read snapshot. A transaction-scoped client inherits the caller's isolation; callers needing a coherent multi-statement snapshot must use Repeatable Read or Serializable.

type Failure 

type Failure = failure.Value

type HistoryEntry 

type HistoryEntry struct {
	ExecutionID       ExecutionID
	Position          JournalPosition
	EntryID           JournalEntryID
	Kind              HistoryKind
	RecordedAt        time.Time
	CausationPosition *JournalPosition
	CommandID         CommandID
	AttemptID         AttemptID
	EventID           EventID
	EventNamespace    string
	EventName         string
	EventKey          string
	EventClass        string
	TerminalStatus    TerminalStatus
	Body              json.RawMessage
	BodyHash          string
}

func History 

func History(ctx context.Context, c Client, id ExecutionID, opts ...HistoryOption) ([]HistoryEntry, error)

type HistoryKind 

type HistoryKind string
const (
	HistoryExecutionStarted HistoryKind = "execution_started"
	HistoryExecutionFailing HistoryKind = "execution_failing"
	HistoryCommandCreated   HistoryKind = "command_created"
	HistoryAttemptStarted   HistoryKind = "attempt_started"
	HistoryAttemptConcluded HistoryKind = "attempt_concluded"
	HistoryEventRecorded    HistoryKind = "event_recorded"
)

type HistoryOption 

type HistoryOption interface {
	// contains filtered or unexported methods
}

func HistoryAfter 

func HistoryAfter(position JournalPosition) HistoryOption

func HistoryLimit 

func HistoryLimit(limit int) HistoryOption

type JournalEntryID 

type JournalEntryID string

type JournalPosition 

type JournalPosition uint64

type KeyScope 

type KeyScope string

type KeyedHistoryEntry 

type KeyedHistoryEntry struct {
	DefinitionName string
	ExecutionKey   string
	KeyScope       KeyScope
	HistoryEntry
}

KeyedHistoryEntry is a history entry carrying its execution's identity.

type KeyedHistoryFilter 

type KeyedHistoryFilter struct {
	Keys     []string
	PageSize int
	Cursor   string
}

KeyedHistoryFilter selects bounded retained history for exact execution keys. Cursor values are opaque and may be reused only with the same Keys filter.

type KeyedHistoryPage 

type KeyedHistoryPage struct {
	Entries    []KeyedHistoryEntry
	NextCursor string
}

KeyedHistoryPage contains one bounded page and an opaque cursor for the next page. NextCursor is empty when no later row was observed.

func ListHistoryByKeys 

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

ListHistoryByKeys returns one bounded retained-history page for every execution that ever held one of filter.Keys. Rows are ordered by key, definition, execution creation, execution ID, and journal position. Journal order is preserved within each execution. Transaction-scoped clients use the caller's transaction and observe their uncommitted writes.

type LiveWork 

type LiveWork struct {
	ExecutionID      ExecutionID
	DefinitionName   string
	ExecutionKey     string
	KeyScope         KeyScope
	ExecutionStatus  ExecutionStatus
	CommandID        CommandID
	CommandKey       string
	CommandName      string
	Queue            string
	QueueState       QueueState
	NextRunAt        time.Time
	LeaseOwner       string
	LeaseExpiresAt   *time.Time
	AttemptOrdinal   int
	CommandCreatedAt time.Time
}

LiveWork is one queued (or leased) command of a non-terminal execution, carrying the execution's identity. It is the batch, key-addressed read for consumers that decorate their own domain rows with dispatch state, without touching Flow's tables.

type LiveWorkFilter 

type LiveWorkFilter struct {
	Keys     []string
	PageSize int
	Cursor   string
}

LiveWorkFilter selects bounded queued work for exact execution keys. Cursor values are opaque and may be reused only with the same Keys filter.

type LiveWorkPage 

type LiveWorkPage struct {
	Work       []LiveWork
	NextCursor string
}

LiveWorkPage contains one bounded page and an opaque cursor for the next page. NextCursor is empty when no later row was observed.

func ListLiveWork 

func ListLiveWork(ctx context.Context, c Client, filter LiveWorkFilter) (LiveWorkPage, error)

ListLiveWork returns one bounded page of queued commands for non-terminal executions whose execution key is in filter.Keys. Rows are ordered by key, definition, execution creation, execution ID, and command ID. An ordinary client does not provide a cross-page snapshot; a transaction-scoped client uses the caller's transaction and observes its uncommitted writes.

type MigrateOption 

type MigrateOption interface {
	// contains filtered or unexported methods
}

MigrateOption is a sealed migration and schema-inspection option.

type Node 

type Node struct {
	// contains filtered or unexported fields
}

Node is an ephemeral builder for a command staged by a worker decision. It is valid only for the duration of that decision.

func Execute 

func Execute[W, A, R any](work *Work[W], key string, cmd Command[A, R], args A) *Node

Execute requests a command from a worker. It never invokes the worker inline; the command is staged in the enclosing durable decision.

func (*Node) Delay 

func (node *Node) Delay(duration time.Duration) *Node

func (*Node) Key 

func (node *Node) Key() string

func (*Node) Optional 

func (node *Node) Optional() *Node

func (*Node) WaitFor 

func (node *Node) WaitFor(event EventRef, key string) *Node

func (*Node) Within 

func (node *Node) Within(duration time.Duration) *Node

type None 

type None = struct{}

type NopObserver 

type NopObserver struct{}

func (NopObserver) Observe 

func (NopObserver) Observe(context.Context, Observation)

type Observation 

type Observation struct {
	Kind        ObservationKind
	Operation   string
	Outcome     string
	ExecutionID ExecutionID
	CommandID   CommandID
	CommandKey  string
	Name        string
	Version     int
	Queue       string
	Worker      string
	Count       int64
	Duration    time.Duration
	OccurredAt  time.Time
}

Observation contains only bounded operational metadata. It intentionally has no payload, result, raw SQL, connection, or lease token field.

type ObservationKind 

type ObservationKind string
const (
	ObservationExecution ObservationKind = "execution"
	ObservationCommand   ObservationKind = "command"
	ObservationEvent     ObservationKind = "event"
	ObservationAttempt   ObservationKind = "attempt"
	ObservationClaim     ObservationKind = "claim"
	ObservationLease     ObservationKind = "lease"
	ObservationWait      ObservationKind = "wait"
	ObservationRuntime   ObservationKind = "runtime"
)

type Observer 

type Observer interface {
	// Observe receives best-effort operational metadata. Implementations must
	// return promptly and should stop work when ctx is cancelled. Flow never
	// waits indefinitely for an observer during runtime shutdown.
	Observe(context.Context, Observation)
}

type Option 

type Option interface {
	// contains filtered or unexported methods
}

Option is a sealed runtime configuration option.

func WithMaxCommandsPerExecution 

func WithMaxCommandsPerExecution(max int) Option

WithMaxCommandsPerExecution sets the command ceiling copied into each newly created execution. Zero explicitly disables the ceiling.

func WithNotifications 

func WithNotifications(enabled bool) Option

WithNotifications enables or disables transactional PostgreSQL wake hints. It defaults to enabled. Polling always remains active and is the correctness path, so disabling notifications is suitable for transaction-pooling proxies and deliberately poll-only deployments.

func WithObserver 

func WithObserver(observer Observer) Option

WithObserver installs the optional no-op-by-default operational observer.

func WithPollInterval 

func WithPollInterval(interval time.Duration) Option

WithPollInterval configures the fallback scheduler and maintenance poll.

func WithQueueConcurrency 

func WithQueueConcurrency(queue string, concurrency int) Option

WithQueueConcurrency optionally gives one queue lane a smaller process-local handler limit. The lane still shares the runtime's global worker capacity.

func WithShutdownGrace 

func WithShutdownGrace(grace time.Duration) Option

WithShutdownGrace configures how long Run waits before interrupting handlers.

func WithWorkerConcurrency 

func WithWorkerConcurrency(concurrency int) Option

WithWorkerConcurrency bounds command handlers running in this process.

type QueueDepth 

type QueueDepth struct {
	Queue          string
	Ready          int64
	Delayed        int64
	Running        int64
	OldestReadyFor time.Duration
}

QueueDepth is a point-in-time operational snapshot of one queue lane. Ready commands are deliverable now, Delayed commands wait out a retry backoff or start delay, and Running commands hold an attempt lease. OldestReadyFor is how long the oldest deliverable command has been ready; a growing value with stable Ready means no compatible worker is claiming the lane.

func GetQueueDepth 

func GetQueueDepth(ctx context.Context, c Client, queue string) (QueueDepth, error)

GetQueueDepth reports the lane's current deliverable, scheduled, and leased command counts. It reads operational delivery state, not application events: the counts change as attempts are claimed and settled.

type QueueState 

type QueueState string

type Registration 

type Registration interface {
	// contains filtered or unexported methods
}

func Handle 

func Handle[A, R any](
	cmd Command[A, R],
	worker func(context.Context, *Work[A]) (R, error),
	opts ...WorkerOption[A, R],
) Registration

type RetryPolicy 

type RetryPolicy = retrypolicy.PublicPolicy

RetryPolicy is immutable declarative data. Its fields are sealed so durable behavior can only be constructed through validated builders.

func Attempts 

func Attempts(max int) RetryPolicy

func RetryFor 

func RetryFor(maxElapsed time.Duration) RetryPolicy

type Runtime 

type Runtime struct {
	// contains filtered or unexported fields
}

Runtime is a configured PostgreSQL-backed Flow client. New starts no goroutines; execution operations are usable before background processing is started.

func New 

func New(db *pgkit.DB, opts ...Option) (*Runtime, error)

New validates configuration and schema compatibility without migrating or starting background work.

func (*Runtime) InTx 

func (r *Runtime) InTx(tx pgx.Tx) Client

InTx returns a client whose writes participate in the supplied caller-owned transaction. Flow never commits or rolls back that transaction.

func (*Runtime) Register 

func (r *Runtime) Register(definitions ...Registration) error

func (*Runtime) Run 

func (r *Runtime) Run(ctx context.Context) error

func (*Runtime) Stop 

func (r *Runtime) Stop(ctx context.Context) error

type SchemaStatus 

type SchemaStatus struct {
	Schema           string
	CurrentVersion   int
	MinReaderVersion int
	MinWriterVersion int
	Compatible       bool
	AppliedAt        time.Time
}

SchemaStatus describes the verified migration and compatibility state.

func CheckSchema 

func CheckSchema(ctx context.Context, db *pgkit.DB, opts ...MigrateOption) (SchemaStatus, error)

CheckSchema verifies migration checksums, reader/writer compatibility, and the fixed Flow table inventory without changing the database.

type TerminalStatus 

type TerminalStatus string

type TraceAttempt 

type TraceAttempt struct {
	ID               AttemptID
	Attempt          int
	StartedAt        time.Time
	FinishedAt       *time.Time
	Worker           string
	Classification   string
	ConsumedBudget   bool
	ConsumedAttempts int
	NextAttemptAt    *time.Time
	Failure          *Failure
}

type TraceCommand 

type TraceCommand struct {
	ID               CommandID
	Key              string
	Name             string
	Version          int
	ParentCommandID  CommandID
	Required         bool
	State            CommandStatus
	Args             json.RawMessage
	Result           json.RawMessage
	Queue            string
	InitialDelay     time.Duration
	BudgetStartedAt  *time.Time
	NextAttemptAt    *time.Time
	Within           time.Duration
	Waits            []TraceEventWait
	CreatedPosition  JournalPosition
	TerminalPosition *JournalPosition
	Failure          *Failure
	LastError        *Failure
	UnsatisfiedWaits int
	AttemptOrdinal   int
	ConsumedAttempts int
	WaitStartedAt    *time.Time
	WaitDeadlineAt   *time.Time
	DeliveryState    QueueState
	LeaseOwner       string
	LeaseStartedAt   *time.Time
	LeaseExpiresAt   *time.Time
	CreatedAt        time.Time
	UpdatedAt        time.Time
	StatusAt         time.Time
	FinishedAt       *time.Time
	Attempts         []TraceAttempt
}

type TraceEvent 

type TraceEvent struct {
	ID                EventID
	Position          JournalPosition
	Namespace         string
	Name              string
	Key               string
	Class             string
	TerminalStatus    TerminalStatus
	CommandID         CommandID
	RecordedAt        time.Time
	CausationPosition *JournalPosition
	Body              json.RawMessage
}

type TraceEventWait 

type TraceEventWait struct {
	Name              string
	Key               string
	SatisfiedPosition *JournalPosition
}

type TraceOption 

type TraceOption interface {
	// contains filtered or unexported methods
}

type Tx 

type Tx interface {
	Exec(context.Context, string, ...any) (pgconn.CommandTag, error)
	Query(context.Context, string, ...any) (pgx.Rows, error)
	QueryRow(context.Context, string, ...any) pgx.Row
}

type Work 

type Work[A any] struct {
	Args A
	// contains filtered or unexported fields
}

func (*Work[A]) Info 

func (w *Work[A]) Info() CommandInfo

type WorkerOption 

type WorkerOption[A, R any] interface {
	// contains filtered or unexported methods
}

func WithCommit 

func WithCommit[A, R any](fn func(context.Context, Tx, Commit[A, R]) error) WorkerOption[A, R]

Source Files

View all Source files

Directories

Path Synopsis
examples
agent command
direct command
fanout command
monitor command
Package flowtest provides database-free helpers backed by Flow's production codecs and deterministic engine primitives.
 Package flowtest provides database-free helpers backed by Flow's production codecs and deterministic engine primitives.
replaytest
Package replaytest provides PostgreSQL-backed assertions for verifying that Flow's journal replay agrees with its live projections.
 Package replaytest provides PostgreSQL-backed assertions for verifying that Flow's journal replay agrees with its live projections.
internal
canonical
Package canonical provides the deterministic JSON representation used by Flow's durable identities.
 Package canonical provides the deterministic JSON representation used by Flow's durable identities.
definition
durable
Package durable contains validation shared by public construction and the PostgreSQL store for values whose durable representation is narrower than their Go representation.
 Package durable contains validation shared by public construction and the PostgreSQL store for values whose durable representation is narrower than their Go representation.
failure
fault
Package fault defines named internal fault points used by integration and crash-recovery tests.
 Package fault defines named internal fault points used by integration and crash-recovery tests.
flowerr
pgschema
replay
Package replay folds retained journal entries into settled orchestration projections.
 Package replay folds retained journal entries into settled orchestration projections.
retry
store
store/journalcodec
Package journalcodec owns version validation for internal journal bodies.
 Package journalcodec owns version validation for internal journal bodies.
testengine
Package testengine is the private bridge between flow's production deterministic recorders and the public database-free flowtest package.
 Package testengine is the private bridge between flow's production deterministic recorders and the public database-free flowtest package.
testpg
Package testpg creates isolated PostgreSQL schemas for integration tests.
 Package testpg creates isolated PostgreSQL schemas for integration tests.
tools
db command
Command db manages the disposable PostgreSQL database used by Flow's tests.
 Command db manages the disposable PostgreSQL database used by Flow's tests.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.