dag

package
v0.0.5 Latest Latest
Warning

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

 Go to latest Published: Jul 26, 2026 License: MIT Imports: 14 Imported by: 0

README

dag (package)

Import: github.com/kausys/azync/dag

User guide: ../dag.md · GoDoc: package docs via go doc / pkg.go.dev.

Role

Static durable DAG runtime: graph is data; task handlers are at-least-once functions. Not workflow-as-code (see workflow).

Source layout

File / area Responsibility
dag.go, open.go New / Open
define.go, client.go Graph definition, Run / Signal
worker.go, register.go Task handlers, scheduling
manager.go Admin (Get/Tasks/Retry/Compensate/Cancel)
result.go, context.go, options.go ResultOf, ctx accessors, retention

Driver surface

Requires driver.DAGStore (+ Core job store). Tables: azync_dags, azync_dag_deps; jobs with dag_id.

Public surface (summary)

  • Define / Task / Sleep / WaitSignal / Compensate / OnFailure / …
  • Client.Run, Client.Signal
  • Register, ResultOf[T], NotReady
  • WithRetention, WithIdempotencyKey
  • Manager

Boundaries

  • No import of workflow (or vice versa).
  • Task rows exempt from completed-job vacuum until DAG vacuum (retention).
  • Compensations and dead-task policy are graph-level, not replay history.

Tests

go test ./dag/... · DAG conformance in driver/drivertest.

Documentation

Overview

Package dag provides durable static DAGs over an azync Core: a task graph declared up front, executed by ordinary job machinery, with durable timers, signals, task results, compensation and a per-DAG failure policy. There is no workflow-as-code or replay: the DAG is data, and each task handler is a plain function that runs at-least-once. For Go-function workflows with deterministic replay, see package workflow.

Model

A DAG is declared with Define and the builder methods: 

def := dag.Define("user-onboarding",
	dag.OnFailure(dag.Suspend)).
	Task("create", CreateAccount{Email: email},
		dag.Compensate(DeleteAccount{Email: email})).
	Sleep("cooldown", 24*time.Hour, dag.After("create")).
	WaitSignal("approved", dag.After("cooldown")).
	Task("activate", ActivateAccount{Email: email}, dag.After("approved"))

Client.Run validates the definition (unique keys, existing dependencies, no cycles, no reserved prefixes) and inserts the whole graph atomically. Tasks without dependencies are immediately runnable; the rest start blocked and the scheduler promotes each one when everything it declared with After has succeeded. Fan-out and fan-in are just edges: several tasks can share a dependency, and one task can wait on several.

Primitives

Results: a handler registered with Register returns (R, error); the R value is persisted atomically with the task's completion and any downstream task reads it with ResultOf[R](ctx, key). Tasks without output return None.

Timers: Sleep parks the DAG branch for a duration, durably — no worker holds anything while it waits. A signal named after the sleep's key wakes it early.

Signals: WaitSignal parks the branch until Client.Signal(id, name, payload) delivers; the payload becomes the task's result. Signal returns an error wrapping ErrNoSignalMatched when nothing was waiting.

Polling-wait: a handler that finds its external condition not yet met returns NotReady(d); the task re-checks after d without consuming its retry budget — indefinitely, until it succeeds or fails with a real error.

Compensation: a task may declare Compensate(args). When the DAG compensates, one "comp:<key>" task per succeeded task that declared one runs in reverse completion order (a saga).

Failure policy

Each DAG declares at Define time how a dead task (aborted or out of retries) is handled. Cancel — the default — cancels the remaining tasks, runs the compensation chain and settles the DAG failed. Suspend parks the DAG for a manual decision through the Manager: Retry (reset dead tasks with a fresh budget and resume), Compensate or Cancel. A dead task whose dependents all declared IgnoreDeadDeps does not trigger the policy — the tolerant branch keeps running — but a DAG that finishes with any dead task settles failed, never succeeded.

A DAG moves through running -> succeeded | failed | cancelled, with suspended (parked for an operator) and compensating (saga in flight) alongside. Terminal DAGs are removed by the vacuum after the configured retention (WithRetention, default 30 days; 0 retains forever). A succeeded task's row and result live for as long as its DAG does, regardless of the completed-job retention (azync.WithCompletedRetention, which is why this package has no variant of it): task jobs are exempt from that sweep and are only ever removed as part of their own DAG's vacuum, so a task parked behind a long Sleep or WaitSignal never loses the result ResultOf and CompleteDAGs depend on.

Execution

Compose a Runtime over a shared Core with New, or standalone with Open; the driver must implement the DAG capability (driver.DAGStore). Register handlers with Register / RegisterKind before Worker.Start. Handlers receive the decoded task arguments; task metadata travels on ctx (ID, TaskKey, Attempt, ...). Execution is at-least-once — idempotency of external effects belongs to the handler — and every scheduler operation is set-based and idempotent, so any number of worker instances can run concurrently without leader election.

Run combined with WithIdempotencyKey is also the fan-in barrier across DAGs: any number of concurrent Run calls with the same (name, key) yield exactly one live execution (see Client.Run).

Index

Constants

View Source 
const (
	// StateRunning marks a DAG whose tasks are executing.
	StateRunning = driver.DAGRunning
	// StateSuspended marks a DAG parked for a manual decision (Retry,
	// Compensate or Cancel).
	StateSuspended = driver.DAGSuspended
	// StateCompensating marks a DAG whose compensation chain is executing.
	StateCompensating = driver.DAGCompensating
	// StateSucceeded is the terminal state of a DAG whose tasks all succeeded.
	StateSucceeded = driver.DAGSucceeded
	// StateFailed is the terminal state of a failed DAG (after its
	// compensations, if any, finished).
	StateFailed = driver.DAGFailed
	// StateCancelled is the terminal state of an operator-cancelled DAG.
	StateCancelled = driver.DAGCancelled
)

DAG lifecycle states, re-exported from the driver contract.

View Source 
const (
	// TaskPending marks a task ready to be leased.
	TaskPending = driver.StatePending
	// TaskScheduled marks a task with a future run_at: a started timer, a retry
	// backoff or a NotReady re-check.
	TaskScheduled = driver.StateScheduled
	// TaskActive marks a task currently leased by a worker.
	TaskActive = driver.StateActive
	// TaskBlocked marks a task whose dependencies are not all satisfied yet.
	TaskBlocked = driver.StateBlocked
	// TaskWaiting marks a WaitSignal task parked until its signal arrives.
	TaskWaiting = driver.StateWaiting
	// TaskSucceeded is the terminal state of a completed task.
	TaskSucceeded = driver.StateSucceeded
	// TaskDead marks a task that aborted or exhausted its retry budget.
	TaskDead = driver.StateDead
	// TaskCancelled is the terminal state of a task cancelled by the failure
	// policy or an operator verb.
	TaskCancelled = driver.StateCancelled
)

Task lifecycle states, re-exported from the driver contract.

Variables

View Source 
var ErrNoSignalMatched = errors.New("dag: signal matched no waiting task")

ErrNoSignalMatched reports that Client.Signal found nothing to deliver to: no waiting signal task and no pending timer carries the given name on the target dag. Test with errors.Is.

Functions

func Abort 

func Abort(err error) error

Abort sends the task straight to the dead letter — the error is permanent. The workflow's failure policy reacts on the next scheduler pass.

func Attempt 

func Attempt(ctx context.Context) int

Attempt is the 1-based execution attempt; the first run is attempt 1. A NotReady re-check does not advance it. Zero outside a task.

func ID 

func ID(ctx context.Context) uuid.UUID

ID is the id of the DAG execution the task belongs to. uuid.Nil outside a task.

func IsNotFound 

func IsNotFound(err error) bool

IsNotFound reports whether err is the driver's not-found / wrong-state error, returned by Manager verbs whose target dag was absent or in an unexpected state.

func IsRetry 

func IsRetry(ctx context.Context) bool

IsRetry reports whether this is a re-execution (Attempt > 1). False outside a task.

func MaxAttempts 

func MaxAttempts(ctx context.Context) int

MaxAttempts is the resolved retry budget for the task. Zero outside a task.

func Metadata 

func Metadata(ctx context.Context) map[string]string

Metadata returns the workflow's string-valued annotations (definition WithMeta plus run WithRunMeta). Nil outside a task.

func NewContext 

func NewContext(parent context.Context, info TaskInfo) context.Context

NewContext returns a copy of parent carrying info, so a handler can be exercised in isolation in a test without a running worker: build a TaskInfo, attach it, and the accessors below read from it exactly as they do in production. A context built this way carries no result resolver, so ResultOf returns a clear error on it.

func NotReady 

func NotReady(d time.Duration) error

NotReady parks the task for d and re-checks then, WITHOUT consuming the retry budget: the polling-wait primitive. Unlike Retry it is not a failure — no attempt is recorded, the attempt counter is handed back, and the task re-polls indefinitely until it succeeds or returns a different error. Use it when the task is waiting on an external condition with no deadline of its own (e.g. a verification still pending on a provider).

func Register 

func Register[T TaskArgs, R any](w *Worker, fn func(ctx context.Context, task T) (R, error), opts ...RegisterOption) error

Register binds the handler for T's kind on the worker: sugar over RegisterKind that decodes the payload into T, hands the handler the pure domain value, and persists the returned R as the task's durable result (readable downstream through ResultOf). T and R are inferred from the handler signature. For a task without a result use R = None, which persists nothing. Task metadata travels on ctx (ID, TaskKey, Attempt, ...). It fails on duplicate kinds and after Start — registration happens in the composition root, before the worker runs.

func RegisterKind 

func RegisterKind(w *Worker, kind string, handler func(ctx context.Context, payload json.RawMessage) (json.RawMessage, error), opts ...RegisterOption) error

RegisterKind binds a raw handler for an explicit kind string — the seam for dynamic kinds, where the handler receives the undecoded JSON payload and returns the raw result to persist (nil for none). Task metadata travels on ctx (ID, TaskKey, Attempt, ...) and dependency outputs are read with ResultOf. Same rules as Register: it fails on a kind with the reserved "$" prefix, on duplicate kinds (typed or raw) and after Start.

func Reportable 

func Reportable(err error) error

Reportable retries like Retry but flags the error for loud reporting when retries are exhausted.

func ResultOf 

func ResultOf[T any](ctx context.Context, key string) (T, error)

ResultOf returns the persisted result of the workflow task key, decoded into T — the way a task reads the output of a dependency it declared with After (a dependency is guaranteed succeeded before its dependents run). For a WaitSignal task the result is the signal payload.

The three failure modes are distinguishable: a context without a resolver (built by NewContext instead of a worker), a key with no persisted result (absent from the workflow or not succeeded), and a result that does not decode into T. A task that succeeded without producing a result (a Sleep, or a handler returning None) yields T's zero value with a nil error.

func Retry 

func Retry(err error) error

Retry reschedules with exponential backoff (also the default for plain errors).

func RetryAfter 

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

RetryAfter reschedules with a fixed delay — rate limits, resource warm-up.

func TaskKey 

func TaskKey(ctx context.Context) string

TaskKey is the task's key within its workflow DAG ("comp:<key>" for a compensation task). Empty outside a task.

Types

type Client 

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

Client creates and signals dags.

func (*Client) Run 

func (c *Client) Run(ctx context.Context, def *Definition, opts ...RunOption) (RunResult, error)

Run validates def and durably inserts the whole workflow — header, tasks and dependency edges — in one atomic operation. Dependency-free tasks are immediately runnable; the rest start blocked and are promoted by the scheduler as their dependencies succeed.

Run is safe to call from inside a task handler of another workflow: combined with WithIdempotencyKey it is the barrier pattern for fan-in across dags. When N dags must collectively start one downstream workflow (say, the last task of each upstream flow checks "are all siblings done?" and fires the next stage), every one of them simply calls Run with the same key: exactly one insert wins and the others get the winner's id with Deduplicated=true. The at-least-once re-execution of the calling task is absorbed by the same key — no distributed lock needed.

func (*Client) Signal 

func (c *Client) Signal(ctx context.Context, id uuid.UUID, name string, payload any) error

Signal delivers a named signal with payload (marshaled to JSON) to one workflow: a waiting WaitSignal task of that name completes with the payload as its result, and a pending Sleep timer of that name wakes early. When nothing on the workflow was waiting for the name, it returns an error wrapping ErrNoSignalMatched — the workflow may have moved on, or not reached the wait yet; callers deciding to retry can test with errors.Is.

type DefineOption 

type DefineOption func(*Definition)

DefineOption customizes a Definition.

func OnFailure 

func OnFailure(policy FailurePolicy) DefineOption

OnFailure declares the workflow's failure policy (default Cancel).

func WithMeta 

func WithMeta(key, value string) DefineOption

WithMeta attaches one string-valued annotation to the workflow (repeatable). Meta is stamped onto the workflow header and onto every task job, and is readable from handlers through Metadata. Run-time entries added with WithRunMeta override definition entries on key conflicts.

type Definition 

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

Definition is a declared workflow: a name, a failure policy and a static DAG of tasks built with Task, Sleep and WaitSignal. Build one with Define; run it with Client.Run, which validates the DAG and inserts it atomically. A Definition is a template: it is not mutated by Run and may be reused across runs.

func Define 

func Define(name string, opts ...DefineOption) *Definition

Define starts a workflow definition. Add tasks with Task, Sleep and WaitSignal; validation happens in Client.Run, not here.

func (*Definition) Sleep 

func (d *Definition) Sleep(key string, dur time.Duration, opts ...TaskOption) *Definition

Sleep declares a durable timer task: once its dependencies are satisfied it waits dur (resolved against the backend clock) and then succeeds, without running any handler. A signal named after the task's key wakes it early (Client.Signal(id, key, ...)). It returns the Definition for chaining.

func (*Definition) Task 

func (d *Definition) Task(key string, args TaskArgs, opts ...TaskOption) *Definition

Task declares one handler-backed task: key identifies it within the DAG and args carries its kind and payload (the handler registered for args.Kind() executes it). It returns the Definition for chaining.

func (*Definition) WaitSignal 

func (d *Definition) WaitSignal(key string, opts ...TaskOption) *Definition

WaitSignal declares a wait-for-signal task: once its dependencies are satisfied it parks until Client.Signal(id, key, payload) completes it, with the payload persisted as its result (readable downstream through ResultOf). The signal name is the task's key. It returns the Definition for chaining.

type FailurePolicy 

type FailurePolicy = driver.OnFailurePolicy

FailurePolicy is a workflow's declared reaction to a dead task (a task that aborted or exhausted its retry budget). 

const (
	// Cancel is the default failure policy: cancel the remaining tasks, run the
	// compensations of the succeeded tasks that declared one (in reverse
	// completion order), and settle the workflow failed.
	Cancel FailurePolicy = driver.OnFailureCancel
	// Suspend parks the workflow as suspended, leaving its tasks untouched, so
	// an operator decides through the Manager between Retry, Compensate and
	// Cancel.
	Suspend FailurePolicy = driver.OnFailureSuspend
)

type Filter 

type Filter = driver.DAGFilter

Filter selects DAGs for List. A zero field means "no bound".

type Manager 

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

Manager is the DAG administration surface: inspection, retry, compensation and cancellation. Pure library — no auth, no HTTP; embed it behind your own ops endpoints. It operates the dag source only.

func (*Manager) Cancel 

func (m *Manager) Cancel(ctx context.Context, id uuid.UUID) error

Cancel cancels a non-terminal DAG without compensating (compensation is its own verb): remaining tasks are cancelled and the DAG becomes cancelled. On a compensating DAG the in-flight compensation settles first and the DAG then lands on cancelled. It returns a not-found error for a missing or already terminal DAG.

func (*Manager) Compensate 

func (m *Manager) Compensate(ctx context.Context, id uuid.UUID) error

Compensate manually triggers compensation on a running or suspended DAG, exactly like the Cancel failure policy: remaining tasks are cancelled, the compensation chain of the succeeded tasks that declared one is inserted in reverse completion order, and the DAG moves to compensating (or straight to failed when there is nothing to compensate). It returns a not-found error for a missing DAG or one in any other state.

func (*Manager) Get 

func (m *Manager) Get(ctx context.Context, id uuid.UUID) (*View, error)

Get returns one DAG header or nil when it does not exist.

func (*Manager) List 

func (m *Manager) List(ctx context.Context, filter Filter, page, size int) (Page, error)

List returns one page of dags matching filter, newest first (page is 0-based; size defaults to 50).

func (*Manager) Retry 

func (m *Manager) Retry(ctx context.Context, id uuid.UUID) error

Retry resumes a non-terminal DAG after failures: dead tasks are reset to pending with a fresh budget and a suspended DAG resumes (to running, or back to compensating when a compensation chain exists — original tasks never rerun once compensation started). It returns a not-found error (see IsNotFound) for a missing or terminal DAG.

func (*Manager) Tasks 

func (m *Manager) Tasks(ctx context.Context, id uuid.UUID) ([]TaskView, error)

Tasks returns every task of the DAG (compensation tasks included) ordered by creation time — compensations follow the original tasks; the relative order of tasks inserted together at Run is stable, not the declaration order — or nil when the DAG does not exist (a DAG always has at least one task, so nil unambiguously means absence).

type None 

type None = struct{}

None is the result type of a task that produces no output: a handler returning None persists no result (ResultOf on it yields the zero value).

type Option 

type Option func(*config) error

Option configures a workflow Runtime. Options compose; later options win.

func WithCoreOptions 

func WithCoreOptions(opts ...azync.Option) Option

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 WithDefaultConcurrency 

func WithDefaultConcurrency(n int) Option

WithDefaultConcurrency overrides the per-kind concurrency used when a registration does not set its own. Must be positive.

func WithDefaultMaxRetries 

func WithDefaultMaxRetries(n int) Option

WithDefaultMaxRetries overrides the retry budget applied to tasks declared without an explicit budget (see the MaxRetries task option and the WithMaxRetries register option). Must be positive.

func WithDefaultTaskTimeout 

func WithDefaultTaskTimeout(d time.Duration) Option

WithDefaultTaskTimeout overrides the default per-task wall clock applied to registrations that do not set their own (default 5m). Must be positive; a single kind can still opt out with the WithTaskTimeout(0) register option.

func WithFetchBatchSize 

func WithFetchBatchSize(n int) Option

WithFetchBatchSize overrides how many tasks one dequeue leases. Must be positive.

func WithFetchCooldown 

func WithFetchCooldown(d time.Duration) Option

WithFetchCooldown overrides the pause after a productive fetch. Must be positive.

func WithFetchPollInterval 

func WithFetchPollInterval(d time.Duration) Option

WithFetchPollInterval overrides the idle polling period. Must be positive.

func WithIdleBackoffMax 

func WithIdleBackoffMax(d time.Duration) Option

WithIdleBackoffMax overrides the idle backoff cap of the fetch loops. Must be positive.

func WithLeaseTTL 

func WithLeaseTTL(d time.Duration) Option

WithLeaseTTL overrides the shared lease duration for this runtime. Must be positive.

func WithMaxConcurrency 

func WithMaxConcurrency(n int) Option

WithMaxConcurrency overrides the total concurrent-handler cap across every kind. Must be positive.

func WithMaxReaps 

func WithMaxReaps(n int) Option

WithMaxReaps overrides how many lease expirations a task survives before the reaper kills it. Must be positive.

func WithRetention 

func WithRetention(d time.Duration) Option

WithRetention overrides how long terminal dags (succeeded, failed or cancelled) are kept before the vacuum removes them together with their task jobs and dependency edges (default 30 days). A negative value is rejected; zero means retain forever.

func WithShutdownDrain 

func WithShutdownDrain(d time.Duration) Option

WithShutdownDrain overrides how long Start waits for in-flight tasks on shutdown. Must be positive.

func WithStatsRetention 

func WithStatsRetention(d time.Duration) Option

WithStatsRetention overrides how long daily stat counters are kept. A negative value is rejected; zero means retain forever.

type Page 

type Page struct {
	Items []View
	Page  int
	Size  int
	Total int64
}

Page is one page of dags for the admin list.

type RegisterOption 

type RegisterOption func(*registerOptions)

RegisterOption customizes Register.

func WithConcurrency 

func WithConcurrency(n int) RegisterOption

WithConcurrency caps how many tasks of this kind run at once (default WithDefaultConcurrency).

func WithMaxRetries 

func WithMaxRetries(n int) RegisterOption

WithMaxRetries overrides the retry budget for tasks of this kind (default WithDefaultMaxRetries). The override is resolved durably on a task's first lease unless the task was declared with an explicit MaxRetries task option, which always wins.

func WithTaskTimeout 

func WithTaskTimeout(d time.Duration) RegisterOption

WithTaskTimeout overrides the per-task wall clock for this kind (default WithDefaultTaskTimeout on the runtime; 0 = unlimited).

type RunOption 

type RunOption func(*runOptions)

RunOption customizes one Run.

func WithIdempotencyKey 

func WithIdempotencyKey(key string) RunOption

WithIdempotencyKey deduplicates the run within the definition's name: while a workflow with the same (name, key) is live (running, suspended or compensating), Run inserts nothing and returns the live execution's id with Deduplicated=true. A terminal workflow frees the key, so a finished flow can be re-run with the same key.

func WithRunMeta 

func WithRunMeta(key, value string) RunOption

WithRunMeta attaches one string-valued annotation to this run (repeatable). It merges over the definition's WithMeta entries; on a key conflict the run entry wins. (The names differ because both option sets live in this package.)

type RunResult 

type RunResult struct {
	// ID identifies the workflow: the new one, or — when Deduplicated is true —
	// the live execution that already held the idempotency key.
	ID uuid.UUID
	// Deduplicated is true when an idempotency key matched a live execution and
	// nothing was inserted.
	Deduplicated bool
}

RunResult reports the outcome of a Run.

type Runtime 

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

Runtime is the workflow system over one azync Core: the Client, the Worker and the Manager, all operating the workflow job source only. It requires a driver with the driver.DAGStore capability; New and Open fail with a clear error otherwise.

func New 

func New(core *azync.Core, opts ...Option) (*Runtime, error)

New composes a workflow runtime over a shared Core. Settings start from the Core's defaults and workflow options override them per runtime. It fails when the Core's driver does not implement driver.DAGStore.

func Open 

func Open(dsn string, opts ...Option) (*Runtime, error)

Open builds a standalone workflow 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) Client 

func (r *Runtime) Client() *Client

Client returns the workflow creation and signalling client.

func (*Runtime) Close 

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

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.

func (*Runtime) Manager 

func (r *Runtime) Manager() *Manager

Manager returns the workflow administration client.

func (*Runtime) Migrate 

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

Migrate brings the backend schema up to date (requires a driver.Migrator). Open and New never migrate automatically.

func (*Runtime) Worker 

func (r *Runtime) Worker() *Worker

Worker returns the task execution and scheduling runtime.

type State 

type State = driver.DAGState

State is the persisted lifecycle state of a DAG execution.

type TaskArgs 

type TaskArgs interface {
	Kind() string
}

TaskArgs identifies a task's unit of work by its wire-stable Kind (decoupled from the Go type path), e.g. "kyc.submit_verification" — the same contract as the queue's JobArgs.

type TaskInfo 

type TaskInfo struct {
	DAGID       uuid.UUID
	TaskKey     string
	Kind        string
	Attempt     int // 1-based: first execution is attempt 1
	MaxAttempts int
	EnqueuedAt  time.Time
	Meta        map[string]string
}

TaskInfo is the cross-cutting metadata of one task execution. Handlers receive the decoded arguments as their typed value; everything about the task itself — its workflow, key, kind, attempt and the workflow's annotations — travels on the context and is read through the package accessors (ID, TaskKey, Attempt, ...).

func TaskFromContext 

func TaskFromContext(ctx context.Context) (TaskInfo, bool)

TaskFromContext returns the TaskInfo carried by ctx and whether one was present. Outside a task (a ctx that never passed through a worker) it returns the zero TaskInfo and false.

type TaskOption 

type TaskOption func(*taskDecl)

TaskOption customizes one declared task.

func After 

func After(keys ...string) TaskOption

After declares dependencies: the task stays blocked until every named task succeeded (repeatable; keys accumulate). A dependency on a missing key, and any dependency cycle, is rejected by Client.Run.

func Compensate 

func Compensate(args TaskArgs) TaskOption

Compensate declares the task's compensation: when the workflow compensates (Cancel policy, Manager.Compensate) and this task had succeeded, a "comp:<key>" task of args.Kind() runs with args as its payload, chained in reverse completion order with the other compensations.

func IgnoreDeadDeps 

func IgnoreDeadDeps() TaskOption

IgnoreDeadDeps lets the task run even when a dependency ended dead or cancelled, treating it as satisfied. When every dependent of a dead task declares it, the failure policy does not fire for that death and the tolerant branch keeps running — but a workflow that completes with any dead task still settles failed, never succeeded. The exemption is never vacuous: a dead task with no dependents always triggers the policy.

func MaxRetries 

func MaxRetries(n int) TaskOption

MaxRetries overrides the retry budget for this task (highest precedence: task option > register option > runtime default).

type TaskState 

type TaskState = driver.JobState

TaskState is the persisted lifecycle state of one task job.

type TaskView 

type TaskView struct {
	// ID is the task job's primary key (usable with attempt-history admin
	// tooling).
	ID uuid.UUID
	// Key is the task's key within the DAG ("comp:<key>" for a compensation).
	Key string
	// Kind is the handler kind, or an internal kind ("$sleep", "$signal").
	Kind        string
	State       TaskState
	Attempt     int
	MaxAttempts int
	// RunAt is when the task becomes (or became) due.
	RunAt time.Time
	// CompletedAt is zero until the task completed (succeeded or cancelled).
	CompletedAt time.Time
	// LastError is the most recent failure message.
	LastError string
	// HasResult reports whether the task persisted a durable result (the
	// payload itself is read by the tasks that depend on it, via ResultOf).
	HasResult bool
}

TaskView is the admin projection of one task job of a DAG. Optional timestamps are zero when absent (IsZero reports absence).

type TxRunnerClient 

type TxRunnerClient[TTx any] struct {
	// contains filtered or unexported fields
}

TxRunnerClient creates dags inside the caller's own backend transaction, so the creation commits atomically with the caller's writes (outbox pattern). Build one with TxRunner.

func TxRunner 

func TxRunner[TTx any](r *Runtime) (*TxRunnerClient[TTx], error)

TxRunner builds the transactional workflow-creation client for the driver's transaction handle type TTx (e.g. pgx.Tx for the pg driver). It fails immediately when the runtime's driver does not support transactional workflow creation for that type.

func (*TxRunnerClient[TTx]) RunTx 

func (c *TxRunnerClient[TTx]) RunTx(ctx context.Context, tx TTx, def *Definition, opts ...RunOption) (RunResult, error)

RunTx performs Run within tx, letting the caller atomically commit application writes and the workflow creation. Same validation, options and dedupe semantics as Run.

type View 

type View = driver.DAGView

View is the admin projection of one DAG header.

type Worker 

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

Worker is the workflow runtime: per-kind fetch loops feeding an executor pool on the shared engine, the engine's maintenance loops (promotion, reaper, vacuums — scoped to the workflow source), and the workflow scheduler loop that drives the DAG machinery. Handlers register via Register / RegisterKind before Start; the internal Sleep and WaitSignal tasks are never registered — the scheduler resolves them without running any handler.

func (*Worker) Ready 

func (w *Worker) Ready() <-chan struct{}

Ready closes after wakeup setup succeeds and the polling loops are running. Polling-only workers become ready immediately after Start.

func (*Worker) Start 

func (w *Worker) Start(ctx context.Context) error

Start runs the worker until ctx is cancelled: the shared engine (fetch, execute, settle, maintenance) plus the workflow scheduler loop. The scheduler is set-based and idempotent, so every worker instance runs it on its own tick without leader election. On cancellation in-flight tasks drain for up to the shutdown drain budget.

func (*Worker) Wait added in v0.0.4 

func (w *Worker) Wait(ctx context.Context) error

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.

Source Files

View all Source files

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.