flow

package module
v0.3.3 Latest Latest
Warning

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

 Go to latest Published: Aug 12, 2026 License: MIT Imports: 39 Imported by: 0

README

flow

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

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

Commands are the only durable unit of orchestration. Workers perform typed work, emit immutable run-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 run.

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 current v0.x 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)

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

Command.Name, Command.Version, and Command.Queue inspect the immutable definition without accessing the database. Queue returns the configured delivery lane, or Flow's normalized "default" lane when WithQueue was not specified.

Work[A] is the attempt-local scope for one claimed command. It is not the whole Run and it is not the immutable Command[A, R] definition. Each worker invocation receives a fresh Work containing typed arguments, run/command/attempt identity, materialized event inputs, and the private decision state used by Enqueue, Emit, and GetEventValue. It is valid only for that worker call and must not be retained or used concurrently.

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

Enqueue returns the Run snapshot as of durable acceptance; Created reports whether this call created it. GetRun and AwaitRun return the same type with the run's current or final state.

Read one successful command result by its stable key without loading the full run trace:

value, found, err := flow.GetResult(ctx, runtime, run.ID, "finalize", finalizeOrder)

found=false means no successful result is currently available. Use Trace when the complete command graph, attempts, events, or journal is needed.

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.Enqueue(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 run lock until settlement commits and is not an exactly-once boundary for remote calls.

One run is one serialized semantic aggregate. Keep causally related work together, but use separate runs for independent bulk items or shards instead of treating one run as a tenant-wide work container. The default 1,000-command ceiling is a safety limit, not a recommended run size; ordinary runs 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.Enqueue(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, found, err := flow.GetEventValue(work, approved, "approval/42")
if err != nil {
	return result{}, err
}
if !found {
	return result{}, errors.New("required approval is absent")
}

Multiple waits are AND conditions. Matching is exact on event name and key within one run. 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 two event paths:

API Use
flow.Emit(work, ...) stage an event in the current run with the worker decision
event.Deliver(ctx, client, runID, ...) immediately record a detached event in a known run, including from an active worker

Deliver needs the exact target run ID. With a transaction client it commits or rolls back with the caller's application writes; with a regular runtime client it commits independently. A committed delivery survives source failure and retry, so producers should use stable event keys and deterministic payloads. Same-run worker events should use staged flow.Emit: explicitly delivering to the current run is detached and may survive a failed attempt. Delivery is targeted ingress, not publish/subscribe, and target workers remain at-least-once.

Typed event definitions name stable fact kinds. Put entity and generation identity in one deterministic event-key helper used by WaitFor, Deliver, and GetEventValue. If an external publisher knows only a domain key, it may call GetCurrentRun, then Deliver to the returned ID. The run can settle between those operations, so ErrTerminal is an expected race to handle explicitly.

Positive durable durations may be fractional: Flow rounds them upward once to the next whole millisecond before fingerprinting or persistence. Zero and negative values retain each option's validation rules.

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.
  • examples/pipeline: multiple queues, atomic worker events, an external transaction, an all-of join, generation-fenced keys, and dynamic work.

Run one against PostgreSQL:

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

Operations

Caller-owned transactions

Create exactly one transaction client for each pgx.Tx, do all Flow writes first, mark the application phase, and then touch application rows:

tx, err := db.Conn.Begin(ctx)
if err != nil {
	return err
}
defer tx.Rollback(ctx)

flowTx := runtime.InTx(tx) // once for this transaction; do not use concurrently
if err := approved.Deliver(ctx, flowTx, runID, eventKey, value); err != nil {
	return err
}
if err := flowTx.BeginApplicationWrites(); err != nil {
	return err
}
if _, err := tx.Exec(ctx, applicationSQL); err != nil {
	return err
}
return tx.Commit(ctx)

TransactionClient does not commit or roll back the transaction and must not outlive it. Repeating runtime.InTx(tx) creates an independent lock-order guard and is invalid usage. After BeginApplicationWrites, every Flow write or run-locking operation through that client fails before issuing SQL. Keep the transaction short because locked run rows remain locked until caller commit.

For a live-key root, Command.ReplaceCurrentRun atomically cancels an exact expected generation and creates its successor. If a retry finds a different, declaration-equivalent current generation, it rediscovers that committed successor; a different declaration conflicts. An exact expected generation is always replaced, even when its declaration equals the requested successor.

  • 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.
  • Run 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 run 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 run 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

Package flow provides event-driven, durable, distributed work on PostgreSQL.

Core model

Flow has a small set of foundational concepts:

  • Command is an immutable, typed definition of work. Its argument and result types are part of the Go API, while its stable name and version are part of durable identity.
  • A worker, registered with Handle, implements one command definition. Each invocation receives a fresh Work: the attempt-local scope for one claimed command, containing typed arguments, durable identity, event inputs, and the decision being built. Work is neither the whole Run nor the immutable Command definition.
  • Event is an immutable, typed definition of a durable fact. An event name describes the fact kind; its key carries domain and generation identity.
  • Run is one durable command graph and its consistency boundary. It owns the root command, staged descendants, exact event inputs, attempts, and ordered journal.
  • Runtime is both a Client for durable operations and, when passed to Runtime.Run, a processor for locally registered workers.

The usual shape is: 

Command definition --Enqueue--> Run
Runtime.Run -----------claim----> attempt-local Work
Work ----------------Enqueue----> staged child command
Work -----------------Emit------> staged application event
Event + WaitFor ----------------> runnable command

Root run starts are durable and asynchronous: Enqueue always enqueues rather than calling a worker inline. Inside a worker, Enqueue and Emit build one typed decision in memory. That decision, the worker result, and an optional short same-database WithCommit callback settle atomically after the attempt fence is rechecked.

Exact event gates provide durable sequencing and all-of joins without consuming a worker or database connection while waiting. Matching is scoped to one run and uses the tuple (event name, event key). Values for the current command's declared gates are materialized before invocation and read from memory with GetEventValue.

Definitions, clients, and processing

Definitions are immutable values and registration is runtime-local; Flow keeps no process-global registry. A Client is a sealed durable-operation capability implemented by Runtime and its transaction-scoped client. New validates the explicitly migrated schema and starts no goroutines, so a Runtime that is not passed to Run remains a lightweight client for API and publisher processes.

Runtime.Run processes compatible commands with bounded concurrency, renewable leases, settlement fencing, and anonymous takeover across replicas. Renewal calls are internally time-bounded and skip rows held by another Flow transaction so one settlement cannot delay unrelated attempts. A process-local watchdog conservatively cancels attempts whose last known lease window expires; PostgreSQL attempt ID and lease-token fencing remains the durable ownership authority.

Handler invocation is at-least-once. Durable PostgreSQL progression is fenced so that only the current attempt can settle. Application handlers should therefore use stable idempotency keys for remote effects rather than interpreting fenced settlement as exactly-once handler invocation.

Run identity and history

A stable non-empty run key is permanently idempotent by default. WithLiveKey instead gives at most one non-terminal run for a command definition and key; after that run becomes terminal, a new generation may start with the same key. GetCurrentRun resolves the current non-terminal generation when an external caller knows the domain key but not its exact RunID. Terminal generations remain durable history.

Flow retains journal, payload, and terminal data indefinitely and exposes no pruning API. Inspection, history, and trace APIs read durable state without invoking application code. GetResult reads one typed successful command result by run and command key directly from its projection, without replaying the run journal.

Choosing command boundaries

A command should mark an independent retry, side-effect, isolation, queue, timeout, external-wait, or parallelism boundary—not every deterministic business-logic step. Keep causally related commands in one run, but use separate runs for independent bulk items because one run is a serialized semantic aggregate.

Large fan-outs should be chunked into bounded command batches and large all-of inputs reduced through hierarchical joins. Parent-produced data belongs directly in child arguments; large or sensitive values should stay in application storage behind stable references.

Transactions, events, and operations

WithCommit is intended for short same-database writes and must not contain remote calls. Caller-owned transactions should also be short because a run lock remains held until the caller commits or rolls back. Create exactly one TransactionClient with Runtime.InTx for each caller transaction, perform Flow writes first, call TransactionClient.BeginApplicationWrites, and then perform application row locks/writes. The client is non-concurrent, does not own the transaction, and must not outlive it.

External callers record run-scoped events with Event.Deliver, which provides deliberately detached ingress to a known run, including from an active worker; passing Runtime.InTx joins it to caller-owned application writes. Same-run worker events should normally use staged Emit so they commit atomically with the worker result. External code that knows a domain key rather than an exact run ID may compose GetCurrentRun with Event.Deliver, handling the ordinary race in which the selected run settles before delivery. Event definitions should name stable fact kinds; deterministic keys should carry entity and generation identity.

Command.ReplaceCurrentRun atomically cancels an exact expected live-key generation and creates a distinct successor. Retries can rediscover a declaration-equivalent successor only after the current run ID differs from the expected predecessor.

Positive fractional public durations are rounded upward to a whole millisecond before durable fingerprints or rows are produced. Stored and decoded durations remain strictly exact milliseconds.

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

The current v0.x line supports Go 1.26 and PostgreSQL 17 and 18. Published migrations are immutable and upgrades are forward-only. During v0.x, intentional Go API changes may be described in release notes.

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 (
	RunStatusRunning   RunStatus = "running"
	RunStatusFailing   RunStatus = "failing"
	RunStatusSucceeded RunStatus = "succeeded"
	RunStatusFailed    RunStatus = "failed"
	RunStatusCancelled RunStatus = "cancelled"
	RunStatusExpired   RunStatus = "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 CancelRun added in v0.3.0 

func CancelRun(ctx context.Context, c Client, id RunID, 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, bool, error)

GetEventValue returns the typed value attached to an exact event gate materialized for the current command. The value is already in memory when the worker starts; this function does not wait or query the database. found=false reports ordinary absence without poisoning the worker decision.

func GetResult added in v0.3.2 

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

GetResult reads the typed result projection for one command key without loading or replaying the run trace. found=false means that the run exists but the command has no successful result yet, including when the command is absent, pending, running, or terminal without success. A stored command with a different name or version returns ErrConflict.

func Migrate 

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

Migrate applies every unapplied embedded Flow migration in its own run-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 RunTrace, 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 directly to durable operations; 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]) Enqueue added in v0.3.0 

func (cmd Command[A, R]) Enqueue(ctx context.Context, client Client, key string, args A, opts ...RunOption) (Run, error)

func (Command[A, R]) Name 

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

func (Command[A, R]) Queue added in v0.3.3 

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

Queue returns the command's normalized delivery queue. It returns an empty string for a zero or invalid command definition.

func (Command[A, R]) ReplaceCurrentRun added in v0.3.0 

func (cmd Command[A, R]) ReplaceCurrentRun(
	ctx context.Context,
	client Client,
	expected RunID,
	key string,
	args A,
	reason string,
	opts ...RunOption,
) (ReplaceRunResult, error)

ReplaceCurrentRun atomically cancels expected and creates a distinct live-key successor. If expected is stale, an equivalent already-committed successor is returned with Replaced=false; a different current declaration conflicts and no state is changed.

func (Command[A, R]) Version 

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

type CommandFailure 

type CommandFailure = Failure

type CommandID 

type CommandID string

type CommandInfo 

type CommandInfo struct {
	RunID      RunID
	RunKey     string
	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 RunID, key string, payload T) error

Deliver records an event in a known run, 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-run events that must settle atomically with the worker decision.

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 Failure 

type Failure = failure.Value

type HistoryEntry 

type HistoryEntry struct {
	RunID             RunID
	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 RunID, opts ...HistoryOption) ([]HistoryEntry, error)

type HistoryKind 

type HistoryKind string
const (
	HistoryRunStarted       HistoryKind = "execution_started"
	HistoryRunFailing       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
	RunKey         string
	KeyScope       KeyScope
	HistoryEntry
}

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

type KeyedHistoryFilter 

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

KeyedHistoryFilter selects bounded retained history for exact run 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 run that ever held one of filter.Keys. Rows are ordered by key, definition, run creation, run ID, and journal position. Journal order is preserved within each run. Transaction-scoped clients use the caller's transaction and observe their uncommitted writes.

type LiveWork 

type LiveWork struct {
	RunID            RunID
	DefinitionName   string
	RunKey           string
	KeyScope         KeyScope
	RunStatus        RunStatus
	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 run, carrying the run'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 run 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 runs whose run key is in filter.Keys. Rows are ordered by key, definition, run creation, run 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 Enqueue added in v0.3.0 

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

Enqueue 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
	RunID      RunID
	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 (
	ObservationRun     ObservationKind = "run"
	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 WithMaxCommandsPerRun added in v0.3.0 

func WithMaxCommandsPerRun(max int) Option

WithMaxCommandsPerRun sets the command ceiling copied into each newly created run. 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 ReplaceRunResult added in v0.3.0 

type ReplaceRunResult struct {
	Run      Run
	Replaced bool
}

ReplaceRunResult reports the current run after an atomic live-key replacement attempt. Replaced is true only for the call that cancelled the expected predecessor and created Run.

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 Run added in v0.3.0 

type Run struct {
	ID            RunID
	Type          string
	Version       int
	Key           string
	RootCommandID CommandID
	Status        RunStatus
	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
}

Run is a durable run state snapshot. Enqueue returns the snapshot as of durable acceptance; GetRun, AwaitRun, and other inspection reads return the current or final state. Created reports whether the producing Enqueue call created the run; it is false for an idempotent rediscovery and always false on inspection reads.

func AwaitRun added in v0.3.0 

func AwaitRun(ctx context.Context, c Client, id RunID) (Run, error)

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

func GetCurrentRun added in v0.3.0 

func GetCurrentRun(ctx context.Context, c Client, typ, key string) (Run, bool, error)

GetCurrentRun finds the one non-terminal run currently holding a live-scoped key for the definition, if any. Live keys admit many settled runs 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 runs with the key may still exist.

func GetRun added in v0.3.0 

func GetRun(ctx context.Context, c Client, id RunID) (Run, error)

type RunFilter added in v0.3.0 

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

RunFilter is the bounded, indexed filter supported by ListRuns. CreatedBefore is exclusive and CreatedAfter is inclusive.

type RunID added in v0.3.0 

type RunID string

type RunOption added in v0.3.0 

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

RunOption is a sealed command run option.

func WaitFor 

func WaitFor(event EventRef, key string) RunOption

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

func WithFailFast 

func WithFailFast(enabled bool) RunOption

func WithLiveKey 

func WithLiveKey() RunOption

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

func WithMetadata 

func WithMetadata(metadata map[string]string) RunOption

func WithRunDeadline added in v0.3.0 

func WithRunDeadline(deadline time.Duration) RunOption

func WithStartDelay 

func WithStartDelay(delay time.Duration) RunOption

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

func Within 

func Within(duration time.Duration) RunOption

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 WithoutRunDeadline added in v0.3.0 

func WithoutRunDeadline() RunOption

type RunPage added in v0.3.0 

type RunPage struct {
	Runs       []Run
	NextCursor string
}

func ListRuns added in v0.3.0 

func ListRuns(ctx context.Context, c Client, filter RunFilter) (RunPage, error)

type RunStatus added in v0.3.0 

type RunStatus string

type RunTrace added in v0.3.0 

type RunTrace struct {
	Run      Run
	Commands []TraceCommand
	Events   []TraceEvent
	History  []HistoryEntry
}

func Trace 

func Trace(ctx context.Context, c Client, id RunID, opts ...TraceOption) (RunTrace, error)

Trace reconstructs one run 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 Runtime 

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

Runtime is a configured PostgreSQL-backed Flow client. New starts no goroutines; run 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) *TransactionClient

InTx returns a transaction-scoped client. Call it once at the transaction boundary and thread the returned value through every Flow operation in that transaction; repeated calls for the same pgx.Tx create independent order guards and are invalid usage.

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 TransactionClient added in v0.3.0 

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

TransactionClient joins Flow operations to one caller-owned PostgreSQL transaction. Create it exactly once per pgx.Tx, use Flow operations before application row locks/writes, and do not use it concurrently or after the transaction ends. Flow never commits or rolls back the transaction.

func (*TransactionClient) BeginApplicationWrites added in v0.3.0 

func (c *TransactionClient) BeginApplicationWrites() error

BeginApplicationWrites marks the irreversible boundary after which this client rejects every Flow write or run-locking operation before issuing SQL. It does not execute SQL or prove that the caller has not already taken application locks.

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
}

Work is the attempt-local command scope passed to a worker. It represents one invocation of one claimed command, not the whole Run and not the immutable Command definition. Args contains the command's typed arguments; Info returns its durable run, command, and attempt identity. The private scope backs Enqueue, Emit, and GetEventValue for the decision being built by this invocation.

A fresh Work is created for every command attempt. It is valid only during the worker call and must not be retained or used concurrently.

func (*Work[A]) Info 

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

Info returns immutable identity and timing information for the claimed command and its current attempt.

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
pipeline command
Command pipeline demonstrates when to split work into durable commands: independent queues, retryable side effects, an external wait, an all-of join, and a dynamic successor.
 Command pipeline demonstrates when to split work into durable commands: independent queues, retryable side effects, an external wait, an all-of join, and a dynamic successor.
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.
uuid
Package uuid generates Flow's durable identifiers and re-exports the identifier type so identifier-producing packages depend on one policy.
 Package uuid generates Flow's durable identifiers and re-exports the identifier type so identifier-producing packages depend on one policy.
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.