dagtask

Small library for running tasks with dependencies. Build a DAG from string keys, call Execute — ready tasks start in parallel, the rest wait.

Built for runtime config apply: config is already in the DB, a diff turns into load/unload operations, and they need to run in the right order without taking down the process on the first failure.

Install

go get github.com/aredoff/dagtask

Usage

import "github.com/aredoff/dagtask"

d := dagtask.New(
    dagtask.Task{Key: "save", DependsOn: []string{"fetch"}, Run: save},
    dagtask.Task{Key: "fetch", Run: fetch},
)

if err := d.Validate(); err != nil {
    // cycle, unknown dependency, duplicate key
    return err
}

return d.Execute(ctx)

DependsOn lists keys that must finish successfully first. If fetch fails, save never runs.

Call Validate() before apply when you want to reject a bad graph without touching runtime. Execute validates too, but by then the DAG is already marked as executed.

Errors and cancellation

• first error from Run stops the whole apply (errgroup + context cancel)
• dependents get TaskSkipped and never call Run
• panic inside a task is caught and returned as PanicError
• a cycle in the graph returns ErrCycle; no tasks run

Tasks should watch ctx.Done() themselves — the library cancels context but won't abort an in-flight HTTP call for you.

Events

For progress outside (bus, logs, metrics):

d.Execute(ctx, dagtask.WithEventHandler(func(e dagtask.Event) {
    // e.Type: started / succeeded / failed / skipped
    // e.Key, e.Err, e.Duration
}))

The handler runs synchronously in the task goroutine. If you publish to a bus, keep it fast or hand off to a channel.

Concurrency limit

d.Execute(ctx, dagtask.WithConcurrency(8))

Limits concurrent Run executions, not goroutine count. Tasks can wait on dependencies without holding a slot.

0 means no limit.

Examples

Clone the repo and run:

git clone https://github.com/aredoff/dagtask.git
cd dagtask

go run ./example/basic      # 2 and 3 in parallel, then 1
go run ./example/pipeline   # fetch → transform → save
go run ./example/diamond    # A → B,C → D
go run ./example/failfast   # error cuts downstream
go run ./example/cycle      # Validate catches a cycle
go run ./example/apply      # module loads with deps and events

Error values

ErrCycle	dependency cycle
ErrDuplicateKey	two tasks with the same key
ErrUnknownDependency	DependsOn references a missing key
ErrAlreadyExecuted	second Execute on the same DAG
ErrEmptyKey / ErrNilRun	invalid task at build time

Out of scope

• generating tasks from a config diff
• rollback / compensation
• passing data between tasks (state lives outside, in a registry)
• retries or persisted progress

One DAG, one apply. New config snapshot → new New(...).

Tests

go test ./...