dag

package
v0.0.8 Latest Latest
Warning

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

 Go to latest Published: Jul 31, 2026 License: MIT Imports: 15 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/Stats/Definitions/TaskCounts/TaskAttempts/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 — inspection (Tasks returns each task's DependsOn, so the slice is the graph), Stats / Definitions / TaskCounts for listings and the definition navigator, TaskAttempts for the failure trail, plus the operator verbs

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
	// StatePaused marks a DAG an operator froze with Pause; Retry resumes it.
	StatePaused = driver.DAGPaused
	// 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
	// TaskSkipped is the terminal state of a task that deliberately did no
	// work (the handler returned Skip): it satisfied its dependents like a
	// success, carries no result and never compensates.
	TaskSkipped = driver.StateSkipped
	// TaskPaused marks a task held out of the ready set by an operator
	// (Manager.Pause); Manager.Retry releases it.
	TaskPaused = driver.StatePaused
)

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

Variables

View Source 
var ErrTaskSkipped = errors.New("dag: task was skipped")

ErrTaskSkipped reports that ResultOf targeted a task that settled as skipped: it deliberately did no work, so there is no result to read — and silently handing back a zero value would hide exactly the distinction the skipped state exists to make. Test with errors.Is and branch.

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 — unless the task declares a Deadline, which bounds the wait: past it, the next NotReady dead-letters the task and the workflow's failure policy reacts. Use it when the task is waiting on an external condition (e.g. a verification still pending on a provider), and pair it with Deadline when waiting forever is itself a failure.

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 settled before its dependents run). For a WaitSignal task the result is the signal payload.

The failure modes are distinguishable: a context without a resolver (built by NewContext instead of a worker), a key with no settled outcome (absent from the workflow or not settled yet), a dependency that was deliberately skipped — ErrTaskSkipped, testable with errors.Is, never a silent zero value — 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 Skip added in v0.0.6 

func Skip(reason string) error

Skip settles the task as skipped: terminal, deliberately-no-work. Use it when the handler finds nothing to do (the resource is already in the target state) so ops can distinguish "ran and worked" from "ran and had nothing to do". A skipped task satisfies its dependents like a succeeded one, carries no result — ResultOf on it returns ErrTaskSkipped — and never compensates.

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 AttemptError added in v0.0.7 

type AttemptError = driver.AttemptError

AttemptError is one recorded failure in a task's retry history.

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, opts ...SignalOption) error

Signal delivers a named signal with payload (marshaled to JSON) to one live 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. The signal is durable: when nothing is waiting yet — the task still blocked behind dependencies, or the scheduler's promotion racing this call — it is buffered and delivered by the scheduler once the task becomes deliverable, never lost. It returns an error only on a marshal failure or for a missing or terminal workflow (test with IsNotFound).

func (*Client) SignalByKey added in v0.0.6 

func (c *Client) SignalByKey(ctx context.Context, name, idempotencyKey, signalName string, payload any, opts ...SignalOption) error

SignalByKey resolves the live workflow holding (name, idempotencyKey) and delivers the signal to it, in one call — the shape a webhook handler wants: it knows the provider's business key (the same string passed to WithIdempotencyKey at Run), never the run UUID, and needs no bookkeeping table mapping one to the other. At most one live run holds a key (the dedupe barrier); terminal runs free it, so a webhook arriving after the run settled gets a not-found error (test with IsNotFound) — late, not wrong. The resolve and the delivery are two calls: a run settling between them also surfaces as not-found from the delivery, the same right answer.

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 DefinitionInfo added in v0.0.7 

type DefinitionInfo struct {
	Name  string
	Stats Stats
}

DefinitionInfo is one DAG definition the backend still holds runs for, with those runs counted by state. (Definition itself is the graph builder in define.go; this is the admin projection of a name that has been run.)

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) Definitions added in v0.0.7 

func (m *Manager) Definitions(ctx context.Context) ([]DefinitionInfo, error)

Definitions returns every definition the backend holds runs for, sorted by name, each with its runs counted by state.

It is the definition navigator's one read, and it answers a question nothing else does: Filter selects BY name, but no other call enumerates the names, so a caller without this has to learn them from whichever page of List it happens to be showing — which silently hides every definition whose runs are off the current page.

Same read as Stats (the counts are grouped by name and summed there), so a screen showing both a navigator and a state-tab bar can serve them from one call rather than two.

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) Pause added in v0.0.6 

func (m *Manager) Pause(ctx context.Context, id uuid.UUID, reason string) error

Pause freezes a RUNNING DAG without burning anything: its pending and scheduled tasks are held out of the ready set, blocked/waiting tasks stay as they are (nothing promotes them while paused), incoming signals buffer for delivery after resume, and time spent paused never consumes a task's snooze Deadline budget. The one resume verb is Retry. Use it to freeze in-flight workflows during a provider outage instead of letting them fail their way into suspended. An active (leased) task keeps its lease and settles on its own. It returns a not-found error (see IsNotFound) for a missing DAG or one in any state but running.

func (*Manager) Retry 

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

Retry resumes a non-terminal DAG after failures or an operator Pause: dead tasks are reset to pending with a fresh budget, paused tasks return to the ready set (both with a fresh snooze Deadline budget), and a suspended or paused 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) RunNow added in v0.0.6 

func (m *Manager) RunNow(ctx context.Context, id uuid.UUID, taskKey string) error

RunNow expedites one scheduled task of the DAG — a NotReady re-check, a retry backoff or a started Sleep timer — to run immediately, addressed by its task key (how an operator thinks, no job UUID needed). It returns a not-found error (see IsNotFound) for a missing DAG or key, or a task in any state but scheduled.

func (*Manager) Stats added in v0.0.7 

func (m *Manager) Stats(ctx context.Context) (Stats, error)

Stats returns how many DAGs sit in each state, across every definition.

func (*Manager) TaskAttempts added in v0.0.7 

func (m *Manager) TaskAttempts(ctx context.Context, id uuid.UUID, taskKey string) ([]AttemptError, error)

TaskAttempts returns one task's failure history, oldest attempt first, addressed by its key within the DAG (how an operator thinks, no job UUID needed). LastError on a TaskView is only the most recent failure; this is the whole trail, which is what distinguishes "flaky, succeeded on retry 3" from "the same error four times".

A task that never failed returns no entries. It returns a not-found error (see IsNotFound) for a missing DAG or key.

func (*Manager) TaskCounts added in v0.0.7 

func (m *Manager) TaskCounts(ctx context.Context, ids []uuid.UUID) (map[uuid.UUID]map[TaskState]int64, error)

TaskCounts returns, per DAG id, how many of its tasks sit in each task state — the batch read behind a listing that shows each run's progress. Doing it per row instead would be one query per listed DAG.

Ids with no tasks (unknown ones included) are absent from the result.

func (*Manager) TaskResult added in v0.0.6 

func (m *Manager) TaskResult(ctx context.Context, id uuid.UUID, taskKey string) (json.RawMessage, error)

TaskResult returns the persisted result of one settled task — the deliberate, single-task read for "what did the provider return here?" during an investigation. It is a separate call, not a TaskView field, on purpose: task results routinely carry sensitive payloads (PII, raw provider responses), so listings never bulk-expose them and the caller can gate this accessor behind its own authorization.

A succeeded task without a result returns (nil, nil); a skipped task returns ErrTaskSkipped (errors.Is); a task that has not settled — or a missing DAG or key — returns a not-found error (IsNotFound).

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).

Each view carries its DependsOn keys, so the returned slice is the whole graph and not just a list: a caller can lay the run out by dependency depth rather than infer an order from timestamps, which would render a parallel fan-out as a chain and never look wrong.

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 DAG 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 DAG system over one azync Core: the Client, the Worker and the Manager, all operating the dag 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 DAG runtime over a shared Core. Settings start from the Core's defaults and DAG 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 DAG 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 DAG 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 DAG 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 SignalOption added in v0.0.6 

type SignalOption func(*signalOptions)

SignalOption customizes one Signal delivery.

func WithMessageID added in v0.0.6 

func WithMessageID(id string) SignalOption

WithMessageID deduplicates the delivery within (workflow, signal name): while an earlier signal with the same id exists on the workflow, a repeat is accepted and dropped without effect. Use the sender's event id for at-least-once webhooks, so a redelivery never double-fires.

type State 

type State = driver.DAGState

State is the persisted lifecycle state of a DAG execution.

type Stats added in v0.0.7 

type Stats struct {
	Running      int64
	Suspended    int64
	Compensating int64
	Paused       int64
	Succeeded    int64
	Failed       int64
	Cancelled    int64
	// Total is every DAG the backend still retains; a vacuumed run is counted
	// nowhere, so this is not a lifetime total.
	Total int64
}

Stats counts DAG executions by state — the one read behind a state-tab bar, which would otherwise be a List call per state.

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
	// DeadlineAt is the task's stamped snooze deadline (see the Deadline task
	// option); zero until the first NotReady of a task that declared one. A
	// polling handler can read it to adapt its re-check cadence as the budget
	// runs out.
	DeadlineAt 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 Deadline added in v0.0.6 

func Deadline(d time.Duration) TaskOption

Deadline bounds the task's NotReady loop: once d has elapsed since the task FIRST reported NotReady (backend clock — the budget measures time spent waiting for the resource, not the workflow's age), the next NotReady escalates the task to dead instead of re-polling, triggering the workflow's OnFailure policy. It does not bound ordinary retries or a task that never snoozes — the retry budget governs those — so time spent failing and backing off never consumes it. Manager.Retry clears the stamped deadline: a retried or unpaused task waits with a fresh budget. Only meaningful on handler tasks that poll with NotReady; d <= 0 is ignored (no deadline).

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
	// DependsOn are the keys this task waits for — the edges declared at Run,
	// plus the compensation-chain links for a "comp:" task. Nil for a root
	// task; nil and empty mean the same thing. Together across every task of
	// the DAG these are the graph, so an admin surface can lay a run out by
	// dependency depth instead of inferring an order from timestamps.
	DependsOn []string
	// EnqueuedAt is when the task row was created.
	EnqueuedAt time.Time
	// RunAt is when the task becomes (or became) due.
	RunAt time.Time
	// StartedAt is when the current attempt was leased, so
	// CompletedAt.Sub(StartedAt) is how long the task actually ran. RunAt
	// cannot answer that — promotion, retry backoff and snooze each rewrite it
	// — and CompletedAt.Sub(EnqueuedAt) folds in the wait, which for a task
	// parked on a poll or a signal dwarfs the execution. Zero before the first
	// lease.
	StartedAt 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 DAG task runtime: per-kind fetch loops feeding an executor pool on the shared engine, the engine's maintenance loops (promotion, reaper, vacuums — scoped to the dag source), and the DAG 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 DAG 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.