README
¶
workflow (package)
Import: github.com/kausys/azync/workflow
User guide: ../workflow.md.
Role
Workflow-as-code runtime: deterministic replay of registered Go functions over append-only history. Effects only via leased Operations.
Source layout
| Path | Responsibility |
|---|---|
workflow.go, open.go |
New / Open, composition |
client.go, manager.go |
Start/Signal; admin + ResolveUncertain |
worker.go, register.go |
Workflow + Operation registration, run loop |
operation.go |
Leased Operation execution, heartbeats, settlement |
future.go, selector.go, sleep.go, signal.go |
Futures / Select / timers / signals |
context.go, options.go, retention.go |
Replay clock, knobs, vacuum ticker |
kernel/ |
Pure in-memory history cursor + replay (no I/O) |
workflow/kernel
Internal engine: append-only event log, command cursor, replay/park decisions. No driver, no network, no SQL.
- Runtime (
workflow) loads history fromWorkflowStore, feedskernel, persists new events, schedules jobs. - Apps should import
workflow, notkernel, unless building tools/tests against the pure engine. - Name is kernel (core of the WAC runtime), not a separate product.
Driver surface
Requires driver.WorkflowStore (+ Core). Migrations: 00003_workflows.sql, 00004_operation_uncertain.sql. Jobs with run_id / source=workflow.
Public surface (summary)
RegisterWorkflow/RegisterOperationClient.Start,Client.SignalExecuteOperation,Sleep,WaitSignal,SelectManager.Get/Cancel/ResolveUncertainWithRetention,WithBusinessIdempotencyKey, worker modes, lease/timeouts
Boundaries
- No import of
dag(sibling on Core only). - Non-determinism in workflow code is a bug; I/O belongs in Operations.
- Ambiguous Operation settlement → job
uncertain+ executionsuspendeduntil admin resolve. - Terminal executions vacuumed by retention; completed workflow jobs exempt from Core completed-job vacuum while
run_idis set.
Tests
go test ./workflow/... · ./workflow/kernel/... · WAC conformance in driver/drivertest.
Documentation
¶
Overview ¶
Package workflow is the workflow-as-code (WAC) runtime: workflows and Operations defined as ordinary Go functions, executed by deterministic replay over an append-only history. It is a sibling of package dag (the static-DAG runtime) over the same azync.Core; the two never import each other (see the repository's boundary test).
User guide: workflow.md at the repository root.
- Client.Start creates a workflow execution, deduplicated by (name, BusinessIdempotencyKey) against live executions.
- Workflow and Operation functions register by (name, version) with RegisterWorkflow and RegisterOperation.
- Worker replays a workflow function against its history on every workflow-task job. ExecuteOperation schedules a leased Operation job (source=workflow, kind "$op:name@version") and parks until the Operation executor records OperationCompleted / OperationFailed and wakes a fresh workflow-task. Sleep and WaitSignal return Futures; Select parks when none are ready.
- Operation handlers may return ErrUncertain (or hit StartToClose timeout) to enter StateUncertain and suspend the run; Manager.ResolveUncertain applies complete / fail / retry.
- Client.Signal appends the signal to the inbox (deduped by MessageID), records SignalReceived in history, and wakes a workflow-task job.
Terminal executions are vacuumed after WithRetention (default 30 days; 0 = forever). Side effects are at-least-once — make Operations idempotent via ExecutionKey / business keys. Operation handlers receive lease heartbeats (ExtendLease every LeaseTTL/2) while they run.
Index ¶
- Variables
- func ExecutionKey(ctx context.Context) string
- func IsNotFound(err error) bool
- func IsUncertain(err error) bool
- func RegisterOperation[TIn, TOut any](w *Worker, name, version string, ...)
- func RegisterWorkflow[TIn, TOut any](w *Worker, name, version string, fn func(ctx Context, in TIn) (TOut, error))
- func Select(ctx Context, futures ...Future) (int, error)
- type Client
- type Context
- type Future
- type Manager
- func (m *Manager) Cancel(ctx context.Context, id uuid.UUID) error
- func (m *Manager) Get(ctx context.Context, id uuid.UUID) (View, error)
- func (m *Manager) ResolveUncertain(ctx context.Context, operationJobID uuid.UUID, decision string, ...) error
- func (m *Manager) Resume(ctx context.Context, id uuid.UUID) error
- func (m *Manager) Suspend(ctx context.Context, id uuid.UUID, reason string) error
- type Option
- func WithConcurrency(n int) Option
- func WithCoreOptions(opts ...azync.Option) Option
- func WithDefaultMaxRetries(n int) Option
- func WithFetchPollInterval(d time.Duration) Option
- func WithLeaseTTL(d time.Duration) Option
- func WithOperationRetryDelay(d time.Duration) Option
- func WithOperationTimeout(d time.Duration) Option
- func WithRetention(d time.Duration) Option
- func WithShutdownDrain(d time.Duration) Option
- func WithWorkerMode(mode WorkerMode) Option
- type Runtime
- type SignalOption
- type StartOption
- type StartResult
- type View
- type Worker
- type WorkerMode
Constants ¶
This section is empty.
Variables ¶
var ErrSelectNoFutures = errors.New("workflow: Select requires at least one future")
ErrSelectNoFutures is returned by Select when called with no futures.
var ErrUncertain = errors.New("workflow: operation uncertain")
ErrUncertain is returned by an Operation handler when the outcome of a mutation cannot be proven. The executor marks the Operation uncertain and suspends the workflow (docs/workflow-v1-spec.md §8).
var ErrUnknownOperation = errors.New("workflow: unknown operation (name, version)")
ErrUnknownOperation is returned when replay reaches an ExecuteOperation call for a (name, version) pair no RegisterOperation call bound on this worker.
var ErrUnknownWorkflow = errors.New("workflow: unknown workflow (name, version)")
ErrUnknownWorkflow is returned when a workflow-task job names a (name, version) pair no RegisterWorkflow call bound on this worker.
Functions ¶
func ExecutionKey ¶
ExecutionKey returns the automatic per-attempt Operation identity from an Operation handler's context, or "" outside an Operation.
func IsNotFound ¶
IsNotFound reports whether err is the driver's not-found error, returned by Manager verbs and Client calls whose target execution was absent or in an unexpected state.
func IsUncertain ¶
IsUncertain reports whether err is (or wraps) ErrUncertain.
func RegisterOperation ¶
func RegisterOperation[TIn, TOut any](w *Worker, name, version string, fn func(ctx context.Context, in TIn) (TOut, error))
RegisterOperation binds a typed Operation handler on w under (name, version). An Operation is the only allowed external I/O in workflow-as-code (see docs/workflow-v1-spec.md §4); its handler receives a plain context.Context, never the deterministic workflow.Context.
func RegisterWorkflow ¶
func RegisterWorkflow[TIn, TOut any](w *Worker, name, version string, fn func(ctx Context, in TIn) (TOut, error))
RegisterWorkflow binds a typed workflow function on w under (name, version). Workers refuse workflow-task jobs whose execution names a pair nothing registered (see ErrUnknownWorkflow). Call every RegisterWorkflow / RegisterOperation before Worker.Start.
func Select ¶
Select deterministically resolves the first ready future in futures, returning its index. Readiness is checked in registration (argument) order, so replaying the same history always picks the same branch regardless of any real-world race between the futures becoming ready.
When none are ready, Select parks the workflow task (see doc.go): the worker schedules a follow-up workflow-task job for the earliest wakeAt among any Sleep-derived futures in the set, and otherwise (a pure WaitSignal wait) leaves waking it to Client.Signal. Select is the only primitive that parks — Sleep and WaitSignal alone never do.
Types ¶
type Client ¶
type Client struct {
// contains filtered or unexported fields
}
Client starts workflow executions and delivers signals to them.
func (*Client) Signal ¶
func (c *Client) Signal(ctx context.Context, workflowID uuid.UUID, name string, payload any, opts ...SignalOption) (bool, error)
Signal delivers a named signal (with payload, marshaled to JSON) to one workflow execution. SignalWorkflow atomically records the delivery in the inbox (deduped by MessageID), appends it to history as SignalReceived, and — unless the execution is already terminal — enqueues a fresh workflow-task job to wake a parked WaitSignal / Select, all in one call (see driver.WorkflowStore.SignalWorkflow): a newly delivered signal is never left recorded with no live task able to act on it. Signals may arrive before any WaitSignal call ever runs ("early" signals) — the workflow's next replay pass finds them in history regardless of arrival order.
func (*Client) Start ¶
func (c *Client) Start(ctx context.Context, name, version string, input any, opts ...StartOption) (StartResult, error)
Start creates a new execution of the workflow registered as (name, version). StartWorkflow durably records its WorkflowStarted history event and enqueues its first workflow-task job atomically with the execution header itself (see driver.WorkflowStore.StartWorkflow) — a caller never observes a newly inserted execution with no history or task, so there is no crash window between "started" and "schedulable" to recover from. input is marshaled to JSON and handed to the registered workflow function.
type Context ¶
type Context interface {
context.Context
// WorkflowID returns the stable identity of this execution.
WorkflowID() uuid.UUID
// RunID returns the current processing epoch (MVP: equals WorkflowID;
// see docs/workflow-v1-spec.md §1).
RunID() uuid.UUID
// Now returns the backend time captured once at the start of this replay
// pass. Stable across every primitive call within the same pass.
Now() time.Time
}
Context is the deterministic execution context a workflow function receives: the standard context.Context surface (cancellation, deadlines) plus workflow identity and the backend clock captured once for this replay pass. Workflow code must read time through Now, never time.Now — determinism requires every replay of the same history to see the same clock reading at the same decision point.
type Future ¶
type Future struct {
// contains filtered or unexported fields
}
Future is the result of a workflow primitive (ExecuteOperation, Sleep, WaitSignal): either already resolved from history, or not yet — in which case only Select can park the workflow task until it resolves. Sleep and WaitSignal never block by themselves; a Future they return may be unready.
func ExecuteOperation ¶
ExecuteOperation schedules a leased Operation (or replays its recorded outcome). While the Operation is in flight it parks the workflow task; the Operation executor appends the terminal history event and wakes a fresh workflow-task job. Side effects belong only in RegisterOperation handlers (docs/workflow-v1-spec.md §4, §7–8).
func Sleep ¶
Sleep starts (or, on replay, recovers) a durable timer for d against the backend clock and returns its Future immediately — ready when the timer already fired, unready otherwise. Sleep never blocks by itself: route the result through Select (even alone, e.g. Select(ctx, Sleep(ctx, d))) to actually park the workflow task until it fires.
func WaitSignal ¶
WaitSignal looks for the named signal in history and returns its Future immediately — ready with the signal's payload when one has already arrived (including "early", before this call ever ran; see docs/workflow-v1-spec.md §9), unready otherwise. WaitSignal never blocks by itself: route the result through Select to actually park the workflow task until a matching signal arrives.
func (Future) Err ¶
Err returns the future's error, if it resolved to one. Calling it on an unready future returns nil.
type Manager ¶
type Manager struct {
// contains filtered or unexported fields
}
Manager administers workflow executions outside the deterministic replay path: read, cancel and resolve an uncertain Operation.
func (*Manager) Cancel ¶
Cancel administratively cancels a non-terminal execution (see docs/workflow-v1-spec.md §15: no programmable cleanup hooks or automatic compensation in the MVP). It returns a not-found error when the execution is absent or already terminal.
func (*Manager) Get ¶
Get returns one execution's current header, or a not-found error (see IsNotFound) when it does not exist.
func (*Manager) ResolveUncertain ¶
func (m *Manager) ResolveUncertain(ctx context.Context, operationJobID uuid.UUID, decision string, result json.RawMessage) error
ResolveUncertain applies an audited decision to an Operation in the uncertain state (docs/workflow-v1-spec.md §8):
- "complete" — record OperationCompleted with result and wake replay
- "fail" — record OperationFailed and wake replay
- "retry" — re-queue the Operation job without a history outcome
func (*Manager) Resume ¶ added in v0.0.6
Resume returns a suspended execution to running AND re-enqueues its workflow-task immediately — the second half is load-bearing: the worker consumed (acked) any task that arrived during the suspension, so flipping the state alone would strand the execution until the stalled-workflow reconciler eventually rescued it. This is the documented recovery path for a replay-error suspension: fix the code (new version), deploy, Resume. It returns a not-found error for a missing execution or one in any state but suspended.
func (*Manager) Suspend ¶ added in v0.0.6
Suspend parks a non-terminal execution: no workflow-task replays until Resume (the worker consumes any task that arrives while suspended). The reason lands in the execution's failure reason, alertable via Get. The operator-freeze verb — say, while the code fix for a determinism violation deploys, or a downstream provider is out. It returns a not-found error for a missing or terminal execution.
type Option ¶
type Option func(*config) error
Option configures a workflow Runtime. Options compose; later options win.
func WithConcurrency ¶ added in v0.0.4
WithConcurrency sets how many workflow-task/Operation-task passes this worker runs concurrently (default 1). Each pass still leases and processes one job at a time (ProcessNext's Limit is always 1); this controls how many such passes run in parallel across goroutines. Must be positive.
func WithCoreOptions ¶
WithCoreOptions forwards options to the Core that Open builds internally (schema, logger, notify channel, shared defaults...). Valid only with Open; New rejects it because the Core is already constructed.
func WithDefaultMaxRetries ¶
WithDefaultMaxRetries overrides the retry budget applied to workflow-task and Operation jobs. Must be positive.
func WithFetchPollInterval ¶
WithFetchPollInterval overrides the worker's idle polling period. Must be positive.
func WithLeaseTTL ¶
WithLeaseTTL overrides how long the worker holds a workflow-task job's lease. Must be positive.
func WithOperationRetryDelay ¶
WithOperationRetryDelay sets the backoff used when an Operation fails with attempts remaining (retry_wait). Must be positive.
func WithOperationTimeout ¶
WithOperationTimeout sets StartToClose for Operation handlers. Zero disables the timeout (not recommended). Must be non-negative.
func WithRetention ¶
WithRetention overrides how long terminal workflow executions (succeeded, failed or cancelled) are kept before the vacuum removes them together with their history, signals, timers and jobs (default 30 days). A negative value is rejected; zero means retain forever.
func WithShutdownDrain ¶ added in v0.0.4
WithShutdownDrain overrides how long Start waits for in-flight workflow-task/Operation passes on shutdown before cancelling their context (default 25s, matching the core default). A pass past that budget is cancelled but still given a final bounded grace period to finish settling before Start gives up on it and returns anyway (see internal/engine's identical drain shape). Must be positive.
func WithWorkerMode ¶
func WithWorkerMode(mode WorkerMode) Option
WithWorkerMode selects combined / workflow-only / operation-only dequeue (docs/workflow-v1-spec.md §12). Default is combined.
type Runtime ¶
type Runtime struct {
// contains filtered or unexported fields
}
Runtime is the workflow-as-code system over one azync Core: the Client, the Worker and the Manager, all operating the workflow job source only (Source SourceWorkflow). It requires a driver with the driver.WorkflowStore capability; New and Open fail with a clear error otherwise. Runtime is a sibling of dag.Runtime; see docs/workflow-v1-spec.md "Coexistence".
func New ¶
New composes a workflow-as-code runtime over a shared Core. It fails when the Core's driver does not implement driver.WorkflowStore.
func Open ¶
Open builds a standalone workflow-as-code runtime that owns a private Core opened from dsn (pass Core options through WithCoreOptions). Close closes the owned Core. Open never migrates; call Migrate before using a fresh schema.
func (*Runtime) Close ¶
Close releases the runtime's resources: the private Core when the runtime was built with Open, nothing when it composes over a shared Core. When the runtime owns its Core, Close first waits (bounded by ctx) for a running Worker to finish draining, so the store is not closed out from under in-flight settlements; on timeout it logs a warning and closes anyway rather than hanging indefinitely.
type SignalOption ¶
type SignalOption func(*signalOptions)
SignalOption customizes one Signal call.
func WithMessageID ¶
func WithMessageID(id string) SignalOption
WithMessageID deduplicates the signal by (WorkflowID, name, MessageID): a repeated MessageID delivers nothing (Client.Signal returns delivered=false, nil error). Without it every Signal call is a distinct message (see docs/workflow-v1-spec.md §9).
type StartOption ¶
type StartOption func(*startOptions)
StartOption customizes one Start call.
func WithBusinessIdempotencyKey ¶
func WithBusinessIdempotencyKey(key string) StartOption
WithBusinessIdempotencyKey deduplicates the start within name: while a workflow with the same (name, key) is live (running or suspended), Start inserts nothing and returns the live execution's WorkflowID with Deduplicated=true. A terminal execution never reuses its WorkflowID and frees the key for a fresh Start (see docs/workflow-v1-spec.md §2).
func WithStartMeta ¶
func WithStartMeta(key, value string) StartOption
WithStartMeta attaches one string-valued annotation to the execution header (repeatable).
func WithTaskQueue ¶
func WithTaskQueue(name string) StartOption
WithTaskQueue routes this execution's workflow-task jobs through a named task queue instead of "default" (see docs/workflow-v1-spec.md §12). The MVP worker does not yet filter by task queue; it is recorded on the execution header for a later version to act on.
func WithWorkflowID ¶
func WithWorkflowID(id uuid.UUID) StartOption
WithWorkflowID overrides the automatically assigned WorkflowID. The caller is responsible for its uniqueness.
type StartResult ¶
type StartResult struct {
// WorkflowID identifies the execution: the new one, or — when
// Deduplicated is true — the live execution that already held the
// business idempotency key.
WorkflowID uuid.UUID
// Deduplicated is true when a BusinessIdempotencyKey matched a live
// execution and nothing was inserted.
Deduplicated bool
}
StartResult reports the outcome of Start.
type Worker ¶
type Worker struct {
// contains filtered or unexported fields
}
Worker replays registered workflow functions against their durable history and executes registered Operations, driven by workflow-task jobs (Source SourceWorkflow). Register every workflow and Operation with RegisterWorkflow / RegisterOperation before Start.
func (*Worker) ProcessNext ¶
ProcessNext dequeues and processes at most one due workflow-task or Operation-task job, returning processed=false when none was ready. It is the synchronous building block Start loops on, and is directly useful in tests that want deterministic, one-step control over replay.
It first promotes any due scheduled job (a durable Sleep or Operation retry_wait) to pending: unlike dag.Worker and queue.Worker, this worker has no separate background maintenance loop over driver.Store.PromoteDue.
func (*Worker) Start ¶
Start runs the worker until ctx is cancelled: WithConcurrency parallel polling passes (each leasing and processing at most one job at a time, exactly like ProcessNext), a vacuum loop that removes terminal executions past WithRetention, and a reconciler loop that re-enqueues a workflow-task for any running execution ListStalledWorkflows finds with no live task (defense-in-depth; see its doc comment — StartWorkflow/SignalWorkflow already schedule their task atomically with their own effect).
On cancellation, in-flight passes drain for up to WithShutdownDrain before their context is cancelled, then a final bounded grace period before Start gives up on them and returns anyway — the same shape as internal/engine's drain, and for the same reason: a handler that ignores cancellation must not hang shutdown forever. Start returns nil after a graceful shutdown, drained or not.
func (*Worker) Wait ¶ added in v0.0.4
Wait blocks until a Start call has returned, or ctx ends first, whichever comes first. If Start was never called, Wait returns immediately (there is nothing to wait for). Close uses Wait to avoid closing a shared store out from under an in-flight drain; callers coordinating their own shutdown (stop ctx, then Wait, then release other resources) should do the same.
type WorkerMode ¶
type WorkerMode string
WorkerMode selects which job kinds a Worker dequeues (spec §12).
const ( WorkerModeCombined WorkerMode = "combined" WorkerModeWorkflowOnly WorkerMode = "workflow-only" WorkerModeOperationOnly WorkerMode = "operation-only" )
Source Files
Directories