Help us shape the future of Go! Take the 2026 Go Developer Survey

quacker

package module
v1.4.0 Latest Latest
Warning

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

 Go to latest Published: Oct 8, 2026 License: MIT Imports: 18 Imported by: 0

README

quacker 🦆

An embeddable task & workflow orchestration engine for Go — a single-process alternative to Hatchet. No server, no Postgres, no workers to deploy: state lives in SQLite (in-memory by default), tasks are plain Go functions, and execution state is readable at any moment without pausing execution.

q, _ := quacker.Open() // ephemeral in-memory state

charge := quacker.NewTask("orders.charge", func(ctx context.Context, o Order) (Receipt, error) {
    return chargeCard(ctx, o)
})

h, _ := quacker.Enqueue(ctx, q, charge, Order{ID: "o-1"})
receipt, _ := h.Result(ctx)

// Any time, from anywhere in your process:
snap, _ := q.Execution(ctx, h.RunID()) // status, attempts, step DAG, timings
events, stop := q.Subscribe(h.RunID()) // live transition stream

Why quacker

Hatchet quacker
Deployment Server + Postgres (+ Docker) One go get, embedded in your binary
State Postgres SQLite (default, pure Go), Postgres, or MySQL
Introspection Web UI / API over HTTP In-process snapshots & streams, zero network
Scope Distributed fleet One process, all cores

Use quacker when your app is the worker: background jobs, pipelines, agent orchestration, fan-out/fan-in — anything where "spin up a queueing platform" is the wrong amount of infrastructure.

Features

  • Workflows & groups — NewWorkflow for a step DAG; NewGroupWorkflow
    • NewGroup(name, …).After(group) gives each stage a single group-level dependency (drawn as box-to-box edges) without repeating it on every step
  • Typed tasks — NewTask[I, O] with JSON payloads (or a pluggable codec)
  • Retries — attempt counts, exponential/constant backoff with jitter
  • Timeouts — per-attempt, via context
  • Queues — named queues with independent concurrency limits and priorities
  • Worker labels — WithLabels on a task and WithWorkerLabels on an engine route work to the workers that can run it
  • Per-key concurrency — WithKey serializes (or caps at N) runs sharing a key, enforced by counting RUNNING rows at claim time
  • Rate limiting — WithRate(queue, n, per) caps starts per sliding window; the window is persisted, so it survives File-mode restarts
  • Middleware — quacker.Use / Wrap decorators around task execution for timing, tracing, error taxonomies, and custom metrics
  • Plugins — compile-time WithPlugin lifecycle hooks (enqueue, step, run finished, emit) that can observe or veto in-process
  • Pluggable codec — WithCodec replaces JSON for user payloads (task input/output, dependencies, events, durable values)
  • Bring-your-own logs — task logs flow to a sink you choose (engine slog by default); opt into SQLite persistence with WithLogStorage(true)
  • Retention / purge — q.Purge and WithRetention bound a long-lived process's growth in every storage mode
  • Metrics callback — WithMetricsFunc pushes periodic state snapshots
  • DAG workflows — steps with dependencies, upstream outputs via DepOutput
  • DAG visualization — q.DAGJSON / q.DAGSVG render a run's step graph with the current state of every node
  • Child runs — quacker.EnqueueChild from inside a task records lineage; Execution exposes a run's children
  • Batch enqueue — quacker.EnqueueBatch inserts many runs in one write transaction
  • Cron & delayed runs — cron specs ("@daily", "0 9 * * 1-5"), sub-second "@every 250ms", and quacker.WithDelay; crons persist and re-arm on restart
  • In-process events — q.Emit(name, payload) fans out to tasks bound with quacker.On(name, task), each receiving the payload as input
  • Durable execution — quacker.SleepDurable and quacker.WaitFor suspend a run without holding a worker slot and survive restarts; quacker.RunOnce keeps side effects from repeating when a task replays
  • Cancellation — instant, propagated to running task contexts
  • Non-blocking introspection — Execution, Runs, Metrics, DAGJSON, Logs, Subscribe — never pause or slow the workers
  • Debug log stream — q.DebugLogs() yields engine activity (claims, retries, suspensions, completions) in-process, without serving HTTP
  • OpenTelemetry tracing — WithTracerProvider emits spans for enqueue, step execution (linked to enqueue; survives restarts), and emit
  • Graceful shutdown — drains in-flight work, marks stragglers, optional recovery on restart (File storage)

Storage modes

quacker.Open(quacker.WithStorage(quacker.Memory()))    // pure :memory: SQLite
quacker.Open(quacker.WithStorage(quacker.Ephemeral())) // default; temp file, WAL, deleted on Close
quacker.Open(quacker.WithStorage(quacker.File("state.db").RecoverRunningOnBoot(true)))

import _ "github.com/kartikbazzad/quacker/postgres" // register the driver
quacker.Open(quacker.WithStorage(quacker.Postgres("postgres://user:pass@host/db?sslmode=disable")))

import _ "github.com/kartikbazzad/quacker/mysql" // register the driver
quacker.Open(quacker.WithStorage(quacker.Driver("mysql", "user:pass@tcp(host:3306)/db")))
  • Memory — nothing touches the filesystem. Note: SQLite :memory: databases cannot use WAL, so reads can briefly wait on an in-flight write (absorbed by busy_timeout).
  • Ephemeral (default) — WAL-backed temp file with synchronous=OFF, deleted on Close. Same zero-durability semantics as memory, but readers are truly lock-free versus the write path.
  • File — survives restarts. Runs interrupted by a previous process are re-queued on boot (or marked failed with RecoverRunningOnBoot(false)). Opening a File path takes an exclusive kernel advisory lock (flock/LockFileEx) on a <path>.quacker.lock sidecar file, so a second engine on the same path fails fast — and the lock is released automatically when the process dies, even on SIGKILL. On platforms without kernel advisory locks (!unix && !windows), only same-process exclusion holds.
  • Postgres — durable, networked, and multi-instance. Requires importing the driver package (import _ "github.com/kartikbazzad/quacker/postgres"), which keeps pgx out of builds that don't use it; without it, Open fails with a clear "backend not registered" error. Several engines may share one database: each gets a worker id, claims carry a lease extended by a heartbeat, and a leaderless reaper re-queues a crashed node's expired leases. Claim batches and per-run completion decisions are serialized with advisory/row locks, and crons fire once per occurrence via a compare-and-swap. Migrations too are serialized, so nodes booting at once can't race DDL.
  • Driver("mysql", dsn) — durable, networked, multi-instance MySQL / MariaDB, same guarantees as Postgres. Requires import _ "github.com/kartikbazzad/quacker/mysql". New SQL databases are a driver: implement driver.Backend and register it; see Storage drivers. Storage.WithDB(*sql.DB) reuses a pool you already own (also PostgresWithDB).
import _ "github.com/kartikbazzad/quacker/postgres"

q, _ := quacker.Open(
    quacker.WithStorage(quacker.Postgres(dsn)),
    quacker.WithWorkerID("worker-a"),  // unique per running engine (default: generated)
    quacker.WithLeaseTTL(30*time.Second),
)

WAL modes (Ephemeral, File) run a passive wal_checkpoint every 60s (WithCheckpointInterval) and a truncating checkpoint on Close, so a clean shutdown leaves the -wal file empty (usually deleted).

Running several engines (Postgres). Each engine has a unique worker id and leases every step it claims; a heartbeat extends the leases of its in-flight steps and a reaper re-queues expired ones, so a crashed node's work is picked up without a human. Graceful Close interrupts only that worker's steps. Step labels route work to capable workers, and crons fire once per occurrence across the fleet. All of it is opt-in via the storage choice — SQLite stays single-process and unchanged.

How non-blocking introspection works: the store keeps a single writer connection and a separate read-only pool. In WAL mode readers never block the writer, so Execution()/Metrics() can run in the middle of a thousand concurrent task transitions without adding contention. A push-based event bus (Subscribe) covers hot-path observation without touching SQLite at all; snapshots remain the authoritative view.

Workflows (DAGs)

ship := quacker.NewTask("ship", func(ctx context.Context, o Order) (Shipment, error) {
    rec, err := quacker.DepOutput[Receipt](ctx, "charge") // upstream output
    ...
})

wf := quacker.NewWorkflow[Order]("fulfill",
    quacker.Step("charge", charge),
    quacker.Step("ship", ship, "charge"),  // after charge
    quacker.Step("notify", notify, "ship"), // after ship — and therefore charge
)

h, _ := quacker.EnqueueWorkflow[Shipment](ctx, q, wf, order) // Result = last step's output

Steps start the moment their dependencies succeed. If a step exhausts its retries, the run fails and remaining steps are cancelled. Task logs written inside steps go to your configured log sink (see Task logs).

Dependencies are direct requirements. Ordering is transitive — a step named as a dep of ship is already ordered before anything ship precedes — so you only list "charge" on notify if you want the edge explicit or robust to ship's definition changing. There is one hard reason to list it: DepOutput sees only declared deps, so a step must name a dep to read its output (see ship reading charge above).

Groups

An ELT-style pipeline is often stages of parallel steps, each stage gated on the previous one. NewGroupWorkflow expresses that directly: a group declares its dependency once with After, and the group's member steps only declare their intra-group deps. Group deps expand to member steps at enqueue, so nothing extra runs and there is no gate step.

extract := quacker.NewGroup[Order]("extract",
    quacker.Step("customers", extractCustomers),
    quacker.Step("orders", extractOrders),
)
transform := quacker.NewGroup[Order]("transform",
    quacker.Step("dim_customers", transformCustomers),
    quacker.Step("facts_orders", transformOrders, "dim_customers"), // same group
).After("extract")   // the whole group waits for extract
load := quacker.NewGroup[Order]("load",
    quacker.Step("load_all", loadTask),
).After("transform")

wf := quacker.NewGroupWorkflow[Order]("elt", extract, transform, load)

DAG/DAGSVG draw a box per group and a group-to-group edge for each After, plus intra-group step edges — not the expanded cross-group step mesh. Execution.Groups carries each group's name, deps, and members. The same semantics apply to child workflows in a run tree: a spawner step is contracted into the group its child runs form, and DAGTree links groups to the groups they depend on.

Limit parallel steps

Use Concurrency(n) to cap running members of a group for each workflow run independently. For example, declare 100 steps but run at most five at once:

steps := make([]quacker.StepDef[Order], 0, 100)
for i := range 100 {
    steps = append(steps, quacker.Step(fmt.Sprintf("partition_%03d", i), processPartition))
}
process := quacker.NewGroup[Order]("process", steps...).
    After("prepare").
    Concurrency(5)

wf := quacker.NewGroupWorkflow[Order]("partitioned", prepare, process)

Here prepare is an earlier group, and processPartition is a task taking Order; each step receives the same workflow input. Inside the task, use sc, ok := quacker.StepFromContext(ctx) and read sc.Step when ok is true to identify its partition.

This is a rolling limit: as soon as one task frees its slot, the next eligible task can start, without waiting for all five to finish. Queue and task key limits also apply, so they may reduce concurrency further. The limit covers all of the group's queues and workers sharing the database, survives restarts, and is reported as Execution.Groups[].MaxConcurrency (max_concurrency in JSON).

The default and Concurrency(0) add no group limit; negative values fail at enqueue. Concurrency(1) serializes execution without guaranteeing order. Retry backoff and durable suspension release slots; retries and resumes must reacquire them. NewGroupWorkflow snapshots the setting, so later changes to the group affect only newly constructed workflows.

Each run has its own budget. Preventing overlap on a shared resource requires a limit of 1 covering every task that accesses that resource; use a shared WithKeyLimit when that protection must span groups or workflow runs. This option covers declared member steps, not child runs spawned inside them.

DAG visualization

Inspect a run's step graph — with the current state of every node — as JSON or as a self-contained SVG (no Graphviz, no browser needed):

js, _ := q.DAGJSON(ctx, runID) // nodes, edges, statuses, attempts, errors
svg, _ := q.DAGSVG(ctx, runID) // status-coloured boxes in dependency order
os.WriteFile("dag.svg", svg, 0o644)

d, _ := q.DAG(ctx, runID) // the structured model, if you prefer

Nodes are laid out in dependency levels (left to right) and coloured by status. Here is a workflow mid-failure — charge succeeded, ship failed, notify was cancelled:

example DAG

DAGJSON marshals to JSON for persistence or an HTTP endpoint; DAGSVG returns a standalone document you can write to a file or serve inline.

Concurrency & rate control

// At most one run per customer at a time, across every queue:
charge := quacker.NewTask("orders.charge", fn,
    quacker.WithKey(func(o Order) string { return o.CustomerID }),
    // quacker.WithKeyConcurrency(3), // default 1 = strict per-key serialization
)

// At most 50 starts per minute on the "emails" queue — a sliding window
// counted over persisted claim times, so it survives File-mode restarts:
q, _ := quacker.Open(
    quacker.WithRate("emails", 50, time.Minute),
)
q.SetRateLimit("emails", 100, time.Minute) // adjustable at runtime

A due step whose key is saturated — or whose queue's window is full — stays QUEUED until a slot opens; nothing is rejected or dropped. Keys are visible in Execution snapshots (Key on the run and each step).

Worker labels

// This task must run on a worker that advertises "gpu":
train := quacker.NewTask("train-model", fn, quacker.WithLabels("gpu"))

// An engine claims it only if its worker labels are a superset:
q, _ := quacker.Open(quacker.WithWorkerLabels("gpu", "linux"))

Routing is a subset check: an engine with extra labels can still run a task that needs fewer. A task with no labels runs on any engine; an engine with no labels claims only unlabeled tasks. Runs whose labels no engine advertises simply stay QUEUED (visible in Execution, with Labels on the step) until a matching worker is opened. Labels are per task and compared at claim time — no separate routing table to configure.

Middleware

// Engine-wide: the first registered is the outermost.
q.Use(func(next quacker.Handler) quacker.Handler {
    return func(ctx context.Context, in json.RawMessage) (json.RawMessage, error) {
        start := time.Now()
        out, err := next(ctx, in)
        observe(quacker.RunIDFromContext(ctx), time.Since(start), err)
        return out, err
    }
})

// Or per task:
task := quacker.NewTask("charge", fn, quacker.Wrap(myTracingMW))

Middleware runs once per attempt, so timing and retry observability are accurate. Order is global → per-task → body; a middleware may short-circuit by returning without calling next. Panics are recovered exactly like a task panic. StepFromContext exposes RunID, Step, Task, and Attempt.

Plugins

Plugins are compile-time — a Go type passed to Open(WithPlugin(p)), not a dynamically loaded .so or an out-of-process server. A plugin returns a Hooks struct and sets only the callbacks it needs:

type auditor struct{}

func (auditor) Name() string { return "audit" }
func (auditor) Hooks() quacker.Hooks {
    return quacker.Hooks{
        // Runs before the task body; return an error to veto (step FAILED, no retries).
        BeforeStep: func(ctx context.Context, s quacker.StepInfo) (context.Context, error) {
            if blocked(s) {
                return ctx, fmt.Errorf("rejected %s", s.RunID)
            }
            return ctx, nil
        },
        OnRunFinished: func(ctx context.Context, r quacker.RunInfo) { record(r) },
    }
}

q, _ := quacker.Open(quacker.WithPlugin(auditor{}))

Before* callbacks run in registration order and may veto; After* callbacks run in reverse and are observe-only (their panics are recovered, never changing the outcome). Hooks run per attempt, for workflows and recovered runs. They are trusted, in-process code and must be safe for concurrent use. For wrapping the task body use middleware; for a post-hoc transition stream use Subscribe.

Payload codec

Payloads are JSON by default. WithCodec swaps in your own encoder for user payloads only — task input and output, dependency outputs, event payloads, and durable RunOnce/WaitFor values:

type msgpackCodec struct{}

func (msgpackCodec) Name() string                          { return "msgpack" }
func (msgpackCodec) Marshal(v any) ([]byte, error)         { return msgpack.Marshal(v) }
func (msgpackCodec) Unmarshal(b []byte, v any) error       { return msgpack.Unmarshal(b, v) }

q, _ := quacker.Open(quacker.WithCodec(msgpackCodec{}))

Schema structures — steps.depends_on and steps.labels — are always JSON, because the SQL claim gates read them, so a custom codec never affects scheduling, dependencies, or labels. It is engine-wide, and the same codec must be used across restarts for a given database (durable journal values and dependency outputs are decoded by the engine that encoded them). See design notes §33 for the scope.

Task logs

Task logs are not stored by the engine by default. TaskLogger(ctx) writes each line to a sink you choose: the engine logger by default, or whatever you pass to WithTaskLogSink (and/or redirect per step with WithLogSink inside middleware). This keeps a long-lived daemon from growing logs it never reads.

q, _ := quacker.Open(
    // Send lines anywhere — file, ring buffer, external collector:
    quacker.WithTaskLogSink(func(e quacker.LogEntry) { myStore.Add(e) }),
)

// Or opt back into built-in SQLite persistence + q.Logs:
q, _ := quacker.Open(quacker.WithLogStorage(true))

Sinks run synchronously in the task's goroutine, so keep them fast (buffer or hand off if the destination can block). Built-in storage writes through the same batched path as before (~50ms batches, dropped on overflow rather than slowing your task) and is read back with q.Logs.

task := quacker.NewTask("import", func(ctx context.Context, in Input) (Output, error) {
    log := quacker.TaskLogger(ctx)
    log.Info("starting batch", "size", len(in.Rows))
    ...
})

Retention & purge

res, err := q.Purge(ctx, quacker.PurgeOptions{
    OlderThan: 24 * time.Hour,        // terminal runs finished before this
    // Statuses: []quacker.Status{quacker.StatusSucceeded}, // default: all terminal
    // KeepLogs: true,                // default false: delete logs with the run
})
// res.Runs, res.Steps, res.Logs

// Or let the engine do it on a schedule:
q, _ := quacker.Open(quacker.WithRetention(quacker.RetentionPolicy{
    OlderThan: 24 * time.Hour,
    Interval:  time.Hour,
}))

Purging is storage-agnostic and bounds growth in every mode, including Memory (which otherwise grows RAM for the process's lifetime) and Ephemeral. Only terminal runs are ever eligible, and never while any of their steps is still RUNNING — so a purge cannot race a step recording its own cancellation. When logs are purged, an orphan sweep removes logs whose run no longer exists.

Cron & events

// Standard specs, descriptors, and sub-second @every:
err := quacker.Cron(q, "nightly", "@every 250ms", report, Report{Day: "yesterday"})

// Event bindings: on emit, each bound task runs with the payload as input.
err = quacker.On(q, "order.placed", receipt)
err = quacker.On(q, "order.placed", notify)
n, err := q.Emit(ctx, "order.placed", OrderPlaced{ID: "o-1", Total: 4200})
events, _ := q.Events(ctx, 100) // newest-first persisted emit audit

With File storage, crons and event bindings persist: on the next Open they are loaded and arm as soon as their task is registered, so the app only needs quacker.Register. A stored fire time still in the future is honored; a missed one is recomputed from now (no catch-up burst).

Child runs

parent := quacker.NewTask("import", func(ctx context.Context, batch Batch) (Summary, error) {
    for _, row := range batch.Rows {
        // Recorded as a child of this run; runs independently.
        if _, err := quacker.EnqueueChild(ctx, q, processRow, row); err != nil {
            return Summary{}, err
        }
    }
    return Summary{Queued: len(batch.Rows)}, nil
})

snap, _ := q.Execution(ctx, parentRunID) // snap.Children lists them

Children are ordinary runs with ParentID set. There is no implicit join — the parent can finish while its children run — and a failed child does not fail the parent. RunFilter{ParentID: ...} lists a run's children, and lineage is informational: purging a parent leaves its children (with a dangling ParentID) intact.

A child enqueued before a durable await re-enqueues on replay (the task re-runs from the top), so wrap it in RunOnce when the task also uses SleepDurable/WaitFor:

childID, err := quacker.RunOnce(ctx, "spawn-children", func() ([]string, error) {
    // enqueue once, return the ids for later
})

Durable execution

task := quacker.NewTask("orders.fulfill", func(ctx context.Context, id string) (string, error) {
    // Runs exactly once, even though the task replays after the sleep below.
    receipt, err := quacker.RunOnce(ctx, "reserve", func() (string, error) {
        return reserve(id) // a side effect you don't want to repeat
    })
    if err != nil {
        return "", err
    }

    if err := quacker.SleepDurable(ctx, 2*time.Hour); err != nil { // no worker slot held
        return "", err
    }

    // Suspend until an event arrives (or time out). Wait is durable and
    // subscription-style: only events emitted after it registers count.
    var payment Payment
    payment, err = quacker.WaitFor[Payment](ctx, "payment.received", 24*time.Hour)
    if err != nil {
        return "", err // quacker.ErrWaitTimeout on timeout
    }
    return ship(receipt, payment), nil
})

q.Emit(ctx, "payment.received", payment) wakes every waiting run with the payload in one transaction; it also still triggers On-bound tasks.

SleepDurable and WaitFor mark the step SUSPENDED: they release the queue and per-key slots and are not charged an attempt. When a sleep's wake time passes — or the awaited event is emitted — the step is claimed again and the task is re-invoked from the top, at which point the await returns immediately. With File storage this survives a restart; the per-attempt timeout covers active execution only, not suspended time. Execution().Steps[i].Status shows SUSPENDED with WaitEvent/ResumeAt. Delivery is durable: Emit records the event and wakes every waiter in one transaction, and a wait that has timed out is never resurrected by a later emit.

Two rules make this safe:

  • Code before a durable await re-executes on resume. Wrap side effects in RunOnce(key, fn) — its result is memoized, so it runs once on success and re-runs on failure (idempotency stays with you on the failure path).
  • Do not recover() around a durable helper, in a task or in middleware — suspension unwinds via an internal panic, and recovering swallows it.

Replay is validated: if a task's durable calls no longer match its persisted journal (e.g. you deployed changed code while runs were suspended), the step fails with quacker.ErrJournalMisaligned rather than resuming with wrong data.

Cancellation & shutdown

q.Cancel(h.RunID())        // queued steps → CANCELLED, running contexts cancelled
// Result returns quacker.ErrRunCancelled

q.Close(ctx)               // stop claiming, drain in-flight (30s default deadline),
// Result returns quacker.ErrRunInterrupted if the deadline expires

Observability

snap, _ := q.Execution(ctx, h.RunID())          // full snapshot (JSON-tagged)
runs, _ := q.Runs(ctx, quacker.RunFilter{Status: quacker.StatusRunning})
m, _ := q.Metrics(ctx)                          // counts by status, queue depths
dag, _ := q.DAGJSON(ctx, h.RunID())             // step graph + current state
logs, _ := q.Logs(ctx, h.RunID(), 100)          // needs WithLogStorage(true)
events, stop := q.Subscribe("")                 // "" = all runs

q.Logs returns nothing unless built-in storage is enabled. For a turnkey push, quacker.WithMetricsFunc(fn) (with WithMetricsInterval) invokes fn(*Metrics) on an interval — a one-liner for Prometheus until OTel lands. All snapshot types marshal to JSON, so exposing them over HTTP is trivial.

Debug logging

The engine streams its own activity — claims, retries, suspensions, run completions, plus anything you log through q.DebugLogger() — so you can observe it without the engine serving HTTP:

go func() {
    for rec := range q.DebugLogs() {
        log.Printf("%s %s %v", rec.Level, rec.Message, rec.Attrs)
    }
}()

q.DebugLogger().Info("checkpoint reached", "batch", 7) // joins the same stream

The stream is bounded and drops records when the consumer falls behind — it never blocks the engine or grows unbounded — and it closes on Close, ending the range. For a lossless-per-subscriber status stream use Subscribe.

Tracing (OpenTelemetry)

q, _ := quacker.Open(quacker.WithTracerProvider(tp)) // your SDK provider

The engine emits three spans:

  • quacker.enqueue per run — parented to your context, so it nests under the HTTP handler / job that enqueued.
  • quacker.step per execution — a new root linked to the enqueue span (the standard async shape: the producer is long gone). It carries run_id, step, task, attempt, and queue, and runs under the task context so your task/middleware can create child spans.
  • quacker.emit per event.

The span active at enqueue is serialized as a W3C traceparent and persisted on the run (visible as Execution().TraceParent), so the step link survives a restart or a different process. The library imports only the OTel API — bring your own SDK/exporter; without WithTracerProvider there is no tracing overhead at all.

Notes & semantics

  • Task names are part of the storage format. Renaming a task orphans its in-flight runs.
  • Re-registering a task name replaces its function. Enqueue, Register, Cron, and On all overwrite the registered def for that name; in-flight runs execute the currently registered def at claim time.
  • Crons and event bindings persist. With File storage they are loaded on Open and arm automatically the moment their target task is registered — so a restarted app only needs quacker.Register, not a re-Cron/re-On. A stored fire time still in the future is kept; a missed one is skipped (no catch-up burst). Unregistered targets simply stay pending.
  • Events are best-effort and in-process. Emit records the event and atomically wakes every durable WaitFor on it; the On-binding fan-out (enqueueing a run per bound task) is best-effort, so a crash between the two can lose those enqueues. Event waits are durable and at-least-once.
  • Tasks should honor ctx — timeouts and cancellation are cooperative.
  • One engine per process assumes a single writer to a given File path — enforced by the .quacker.lock kernel-lock guard: a second engine on the same path fails fast at Open instead of corrupting assumptions.

Not yet

A web UI. Everything else on the old list has shipped: strict ordered per-key concurrency (WithSequence), compile-time lifecycle-hook plugins, a pluggable payload codec, and a public storage-driver contract with MySQL/MariaDB — new SQL databases are a driver you can implement out-of-tree. Multi-process scaling over Postgres/MySQL, worker labels, OTel tracing, durable execution, and DAG visualization (including workflow groups and child-run trees) are all shipped.

Documentation

Examples

runnable programs live in examples/: simple · dag · dagsvg · cron · events · durable · children · introspect · eltgroups

Development

go test ./...          # unit + integration suite
go test -race ./...    # includes an introspection-under-load stress test
go vet ./... && gofmt -l .

Documentation

Overview

Package quacker is an embeddable task and workflow orchestration engine — a single-process alternative to Hatchet. State lives in in-memory SQLite (or a file, if you want durability), tasks are plain Go functions, and execution state is readable at any moment without pausing execution: 

q := quacker.Open()  // ephemeral in-memory state
task := quacker.NewTask("greet", func(ctx context.Context, in Input) (string, error) {
    return "hi " + in.Name, nil
})
h, _ := quacker.Enqueue(ctx, q, task, Input{Name: "world"})
out, _ := h.Result(ctx)
snap, _ := q.Execution(ctx, h.RunID()) // QUEUED/RUNNING/... at any instant
Example

Example is the smallest useful program: define a task, enqueue it, await the result. 

q, _ := Open(WithStorage(Memory()))
defer q.Close(context.Background())

greet := NewTask("greet", func(ctx context.Context, name string) (string, error) {
	return "hello " + name, nil
})

h, _ := Enqueue(context.Background(), q, greet, "world")
out, _ := h.Result(context.Background())
fmt.Println(out)

Output:
hello world

Index

Examples

Constants

View Source 
const (
	// UniqueReuse returns the live run's handle and creates nothing.
	UniqueReuse = store.UniqueReuse
	// UniqueError fails the enqueue with ErrDuplicateJob.
	UniqueError = store.UniqueError
	// UniqueReplace cancels the live run and enqueues the new one.
	UniqueReplace = store.UniqueReplace
)
View Source 
const (
	// ConcurrencyHold queues the run until a key slot frees (default).
	ConcurrencyHold = store.ConcurrencyHold
	// ConcurrencyCancelInProgress cancels running (else oldest) same-key runs
	// to make room for the new one.
	ConcurrencyCancelInProgress = store.ConcurrencyCancelInProgress
	// ConcurrencyCancelNewest cancels the incoming run.
	ConcurrencyCancelNewest = store.ConcurrencyCancelNewest
	// ConcurrencyCancelQueuedExceptNewest keeps the newest keyLimit queued
	// same-key runs and cancels older ones.
	ConcurrencyCancelQueuedExceptNewest = store.ConcurrencyCancelQueuedExceptNewest
	// ConcurrencyCancelQueuedExceptOldest keeps the oldest keyLimit queued
	// same-key runs and cancels newer ones.
	ConcurrencyCancelQueuedExceptOldest = store.ConcurrencyCancelQueuedExceptOldest
)
View Source 
const Version = "v1.4.0"

Version is the released semantic version of this build. Releases are tagged vMAJOR.MINOR.PATCH (see docs/STABILITY.md); fixes bump the patch component.

Variables

View Source 
var (
	// ErrRunCancelled is returned by Result when the run was cancelled.
	ErrRunCancelled = engine.ErrRunCancelled
	// ErrRunInterrupted is returned by Result when the engine shut down
	// before the run finished.
	ErrRunInterrupted = engine.ErrRunInterrupted
	// ErrNotFound is returned by Execution for unknown run IDs.
	ErrNotFound = store.ErrNotFound
	// ErrClosed is returned by enqueue calls after Close has begun.
	ErrClosed = engine.ErrClosed
	// ErrRunRunning is returned by Snooze for a run with a RUNNING step.
	ErrRunRunning = store.ErrRunRunning
	// ErrRunTerminal is returned by Snooze (and PauseRun) for a finished run.
	ErrRunTerminal = store.ErrRunTerminal
	// ErrNotDeadLetter is returned by RetryDeadLetter/DismissDeadLetter for a
	// run that is not dead-lettered.
	ErrNotDeadLetter = store.ErrNotDeadLetter
	// ErrRunGone is returned by Result for an ephemeral run that was deleted
	// before its result could be observed.
	ErrRunGone = engine.ErrRunGone
)

Sentinel errors surfaced by Result and Cancel.

View Source 
var ErrDuplicateJob = store.ErrDuplicateJob

ErrDuplicateJob is returned by an enqueue under UniqueError when a live run already holds the unique key.

View Source 
var ErrExternalTxUnsupported = engine.ErrExternalTxUnsupported

ErrExternalTxUnsupported is returned by EnqueueTx/EnqueueBatchTx on storage that cannot share a caller transaction (SQLite).

View Source 
var ErrJournalMisaligned = engine.ErrJournalMisaligned

ErrJournalMisaligned is returned when a durable task's replayed durable calls do not match its persisted journal — almost always because task code changed while runs were suspended. It fails the step loudly instead of resuming with wrong data.

View Source 
var ErrNonTerminalPurge = store.ErrNonTerminalPurge

ErrNonTerminalPurge is returned when a purge is asked to delete a non-terminal status. Only terminal runs may ever be purged.

View Source 
var ErrWaitTimeout = engine.ErrWaitTimeout

ErrWaitTimeout is returned by WaitFor when its timeout elapses before the awaited event arrives.

Functions

func Cron 

func Cron[I any, O any](q *Quacker, name, spec string, t *Task[I, O], input I) error

Cron registers a trigger that enqueues task with input on the cron spec (standard 5-field syntax plus descriptors like "@hourly" and "@every 5m"). Crons are in-memory: re-register at startup, ideally before Open resumes work.

func DepOutput 

func DepOutput[T any](ctx context.Context, name string) (T, error)

DepOutput decodes the recorded output of a succeeded dependency step. Only usable inside workflow steps that declared the named step as a dependency via Step(name, task, "dependency-name").

func EnqueueBatchTx 

func EnqueueBatchTx[I any, O any](ctx context.Context, tx *sql.Tx, q *Quacker, t *Task[I, O], inputs []I, opts ...EnqueueOption) ([]string, error)

EnqueueBatchTx is EnqueueTx for several inputs on the caller's *sql.Tx. It returns the new run ids in input order.

func EnqueueTx 

func EnqueueTx[I any, O any](ctx context.Context, tx *sql.Tx, q *Quacker, t *Task[I, O], input I, opts ...EnqueueOption) (string, error)

EnqueueTx inserts one run of task on the caller's *sql.Tx and returns its id without a handle. It never commits: the caller's Commit makes the run and any business writes in the same transaction visible atomically, which is the outbox/idempotency pattern. The run may execute on any engine once committed, so wait via Execution rather than a handle. The caller's tx must target the same Postgres/MySQL database as the engine; SQLite returns ErrExternalTxUnsupported.

func Off 

func Off[I any, O any](q *Quacker, event string, t *Task[I, O]) error

Off removes an event→task binding. Removing an unknown binding is a no-op.

func On 

func On[I any, O any](q *Quacker, event string, t *Task[I, O]) error

On binds an event name to a task: every Emit of that event enqueues one run of t with the event payload as the task input. The binding is persisted, so with File storage it re-arms on the next Open once the task is registered (typically via Register). Registering the same (event, task) pair again is idempotent and refreshes the task definition.

func Register 

func Register[I, O any](q *Quacker, t *Task[I, O])

Register makes a task executable without enqueuing a run. Needed when using File storage: register tasks at startup so runs recovered from the previous process can execute. Registration also happens automatically on Enqueue and Cron.

func RunIDFromContext 

func RunIDFromContext(ctx context.Context) string

RunIDFromContext returns the current run's ID inside a task function (empty otherwise).

func RunOnce 

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

RunOnce runs fn at most once per call site across suspensions and retries, memoizing its result. Use it for side effects that must not repeat when a durable task replays after a SleepDurable or WaitFor. fn's error is propagated without being memoized, so a retry re-runs fn (exactly-once on success, at-least-once on failure). It never suspends. Only valid inside a task or workflow step.

func SleepDurable 

func SleepDurable(ctx context.Context, d time.Duration) error

SleepDurable suspends the running step for at least d, durably. It does not consume an attempt or hold a queue or per-key slot while sleeping, and it survives a process restart (File storage): when the wake time passes the step is claimed again and the task is re-invoked from the top, at which point this call returns nil immediately.

Because the task re-runs from the beginning after a suspension, any side effects before a durable await happen again on resume; wrap them in RunOnce to execute exactly once. Do not recover() around a durable helper. Only valid inside a task or workflow step.

Example

ExampleSleepDurable suspends a run without holding a worker slot and resumes it; RunOnce makes the side effect happen once across the replay. 

q, _ := Open(WithStorage(Memory()), WithPollInterval(time.Millisecond))
defer q.Close(context.Background())

task := NewTask("job", func(ctx context.Context, in string) (string, error) {
	reservations, err := RunOnce(ctx, "reserve", func() (int, error) { return 1, nil })
	if err != nil {
		return "", err
	}
	if err := SleepDurable(ctx, time.Millisecond); err != nil {
		return "", err
	}
	return fmt.Sprintf("done %s (%d)", in, reservations), nil
})

h, _ := Enqueue(context.Background(), q, task, "x")
out, _ := h.Result(context.Background())
fmt.Println(out)

Output:
done x (1)

func TaskLogger 

func TaskLogger(ctx context.Context) *slog.Logger

TaskLogger returns a logger whose lines are persisted with the run and readable via Logs. Outside a task function it returns slog's default.

func WaitFor 

func WaitFor[T any](ctx context.Context, event string, timeout time.Duration) (T, error)

WaitFor suspends the running step until an event named event is emitted (via Quacker.Emit), then decodes the event's payload into T. A timeout > 0 bounds the wait and returns ErrWaitTimeout; timeout <= 0 waits indefinitely.

Delivery is durable: when an event is emitted the engine records it and wakes every waiting step in one transaction, so a wait survives a restart. Only events emitted after the wait registers count — an event emitted earlier does not satisfy a later WaitFor. Emitting also triggers any On bindings as usual. Only valid inside a task or workflow step.

func WithLogSink 

func WithLogSink(ctx context.Context, fn func(LogEntry)) context.Context

WithLogSink returns a context whose TaskLogger writes each line to fn instead of the engine's configured sink. Outside a task function it is a no-op; a nil fn is ignored. Middleware uses it to route task logs to its own storage or forwarder.

fn runs synchronously in the task's goroutine, so it must be fast — buffer or hand off if your destination can block.

Types

type Backoff 

type Backoff = engine.Backoff

Backoff configures retry delays. A zero value is not "no delay": empty fields fall back to the library defaults (base 500ms, factor 2, cap 30s, no jitter) — use Exponential or Constant (or set fields) for real policies.

func Constant 

func Constant(d time.Duration) Backoff

Constant returns a fixed retry delay with no jitter.

func Exponential 

func Exponential(base time.Duration) Backoff

Exponential returns a backoff that doubles from base, capped at 30s, with the default 10% upward jitter.

type Codec 

type Codec = engine.Codec

Codec (de)serializes user payloads: task inputs and outputs, dependency outputs, event payloads, and durable journal values. Schema structures (steps.depends_on and steps.labels) are always JSON, because the SQL gates read them, so a custom codec does not affect them.

The same codec must be used across restarts for a given database, since journal values and dependency outputs are decoded by the engine that encoded them.

func NewEncryptedJSONCodec 

func NewEncryptedJSONCodec(key []byte) (Codec, error)

NewEncryptedJSONCodec returns a Codec that encrypts payloads with AES-256-GCM under a 32-byte key; it errors on any other key length.

type ConcurrencyStrategy 

type ConcurrencyStrategy = store.ConcurrencyStrategy

ConcurrencyStrategy selects what an enqueue does when a key is at capacity, set per task with WithKeyStrategy.

type CronInfo 

type CronInfo struct {
	Name string    `json:"name"`
	Spec string    `json:"spec"`
	Task string    `json:"task"`
	Next time.Time `json:"next"`
}

CronInfo describes a registered cron trigger.

type DAG 

type DAG struct {
	RunID    string     `json:"run_id"`
	Workflow string     `json:"workflow"`
	Kind     Kind       `json:"kind"`
	Status   Status     `json:"status"`
	Queue    string     `json:"queue"`
	Nodes    []DAGNode  `json:"nodes"`
	Edges    []DAGEdge  `json:"edges"`
	Groups   []DAGGroup `json:"groups,omitempty"`
}

DAG is a run's step graph with the current state of every node. It is a point-in-time snapshot: reads never pause execution, so a later call sees later states.

func (*DAG) JSON 

func (d *DAG) JSON() ([]byte, error)

JSON renders the DAG (with current state) as indented JSON.

func (*DAG) SVG 

func (d *DAG) SVG() []byte

SVG renders the DAG as a standalone SVG document.

type DAGEdge 

type DAGEdge struct {
	From string `json:"from"`
	To   string `json:"to"`
}

DAGEdge is a dependency edge: From must succeed before To runs.

type DAGGroup added in v1.1.0 

type DAGGroup struct {
	Name   string `json:"name"`  // run id
	Label  string `json:"label"` // human label, e.g. "partition_jobs #3"
	Status Status `json:"status"`
	// Parent is the enclosing group's Name, or "" when this group hangs off
	// the root run. It lets the renderer nest clusters instead of guessing
	// containment from node coordinates.
	Parent string `json:"parent,omitempty"`
	// Deps are group-level dependencies: names of other groups this whole
	// group waits on. They render as box-to-box edges; individual member
	// steps do not repeat them.
	Deps []string `json:"deps,omitempty"`
	// Steps are the group's member step names (a flat grouped workflow). Empty
	// for a run-tree group, whose members are the child run's own nodes.
	Steps []string `json:"steps,omitempty"`
}

DAGGroup is one run's cluster of nodes within a DAGTree — the root run or a child run — drawn as a labelled box behind its nodes. A plain single-run DAG has no groups.

type DAGNode 

type DAGNode struct {
	Name string `json:"name"`
	// Task is the registered task that executes the step (differs from Name
	// for named workflow steps).
	Task        string   `json:"task,omitempty"`
	Status      Status   `json:"status"`
	Deps        []string `json:"deps,omitempty"`
	Attempts    int      `json:"attempts"`
	MaxAttempts int      `json:"max_attempts"`
	Error       string   `json:"error,omitempty"`
	// Group is the run this node belongs to, when the DAG spans several runs
	// (a DAGTree). Empty for a single-run DAG.
	Group string `json:"group,omitempty"`
}

DAGNode is one step of a run's DAG with its current state.

type DeadLetterFilter 

type DeadLetterFilter struct {
	// Workflow and Queue, when non-empty, restrict the result.
	Workflow string
	Queue    string
	// Limit caps the result (default 100); Offset skips that many.
	Limit  int
	Offset int
}

DeadLetterFilter selects dead-lettered runs for DeadLetters.

type DebugRecord 

type DebugRecord = engine.DebugRecord

DebugRecord is one engine debug log line, streamed via DebugLogs.

type EmitInfo 

type EmitInfo = engine.EmitInfo

EmitInfo describes an event about to be emitted.

type EnqueueInfo 

type EnqueueInfo = engine.EnqueueInfo

EnqueueInfo describes a run about to be enqueued.

type EnqueueOption 

type EnqueueOption func(*enqueueConfig)

EnqueueOption configures a single enqueued run.

func WithDelay 

func WithDelay(d time.Duration) EnqueueOption

WithDelay schedules the run to start at least d from now.

func WithPriority 

func WithPriority(p int64) EnqueueOption

WithPriority orders the run ahead of lower-priority work in the same queue. Higher numbers run first.

func WithRunAt 

func WithRunAt(t time.Time) EnqueueOption

WithRunAt schedules the run to start no earlier than t.

func WithUniqueKey 

func WithUniqueKey(key string) EnqueueOption

WithUniqueKey sets the run's unique key explicitly, overriding a task's WithUnique. The key is scoped to the task/workflow name, and "" means the run is not unique.

type Event 

type Event struct {
	RunID string    `json:"run_id"`
	Step  string    `json:"step,omitempty"` // empty for run-level transitions
	From  Status    `json:"from"`
	To    Status    `json:"to"`
	At    time.Time `json:"at"`
	Error string    `json:"error,omitempty"`
}

Event is a live status transition delivered by Subscribe.

type EventRecord 

type EventRecord struct {
	Name    string          `json:"name"`
	Payload json.RawMessage `json:"payload,omitempty"`
	At      time.Time       `json:"at"`
}

EventRecord is one persisted emit, returned by Events.

type Execution 

type Execution struct {
	RunID       string          `json:"run_id"`
	Workflow    string          `json:"workflow"`
	Kind        Kind            `json:"kind"`
	Status      Status          `json:"status"`
	Queue       string          `json:"queue"`
	Priority    int64           `json:"priority"`
	Attempts    int             `json:"attempts"`
	MaxAttempts int             `json:"max_attempts"`
	Input       json.RawMessage `json:"input,omitempty"`
	Output      json.RawMessage `json:"output,omitempty"`
	Error       string          `json:"error,omitempty"`
	// Key mirrors the first step's concurrency key (per-step keys in
	// workflows may differ).
	Key string `json:"key,omitempty"`
	// TraceParent is the W3C traceparent captured at enqueue ("" without
	// tracing), for correlating this run with your traces.
	TraceParent string      `json:"trace_parent,omitempty"`
	RunAt       time.Time   `json:"run_at"`
	CreatedAt   time.Time   `json:"created_at"`
	StartedAt   time.Time   `json:"started_at,omitempty"`
	CompletedAt time.Time   `json:"completed_at,omitempty"`
	Steps       []StepState `json:"steps"`
	// Groups are the run's logical groups (from a grouped workflow), each with
	// its group-level dependencies and member step names. Empty for a plain
	// workflow.
	Groups []RunGroup `json:"groups,omitempty"`
	// Children are runs enqueued from inside this run (via EnqueueChild),
	// oldest first. They run independently; the parent does not wait for them.
	Children []RunSummary `json:"children,omitempty"`
}

Execution is a consistent point-in-time view of a run. Reads never pause task execution.

type Group added in v1.2.0 

type Group[I any] struct {
	// contains filtered or unexported fields
}

Group is a named set of workflow steps with a group-level dependency: the whole group runs after the groups named in After, and member steps only declare their intra-group dependencies. Group deps expand to member steps at enqueue, so nothing extra runs and no gate step is needed.

func NewGroup added in v1.2.0 

func NewGroup[I any](name string, steps ...StepDef[I]) *Group[I]

NewGroup creates a group named name from step declarations. A group can be passed to NewGroupWorkflow and chained with After and Concurrency.

func (*Group[I]) After added in v1.2.0 

func (g *Group[I]) After(groups ...string) *Group[I]

After declares that this whole group runs only after the named groups have completed. It returns the group so calls can be chained.

func (*Group[I]) Concurrency added in v1.4.0 

func (g *Group[I]) Concurrency(n int) *Group[I]

Concurrency limits simultaneously RUNNING member steps per workflow run. A free slot admits the next eligible step immediately, without waiting for a whole batch. Zero (the default) adds no group limit; negative values are rejected at enqueue. One serializes execution but does not guarantee order. Queue and task key limits still apply. Retry backoff and durable suspension release the slot; each retry or resume must acquire it again. NewGroupWorkflow snapshots this setting when constructing the workflow.

func (*Group[I]) Name added in v1.2.0 

func (g *Group[I]) Name() string

Name returns the group's name.

type Handler 

type Handler = engine.Handler

Handler is a task body at the engine's JSON level: it receives the step context and the JSON input and returns the JSON output. Middleware wraps Handlers.

type Hooks 

type Hooks = engine.Hooks

Hooks is the set of engine lifecycle callbacks a plugin may provide; only the non-nil fields are invoked. Before* hooks run in registration order and may return an error to veto; After* hooks are observe-only, run in reverse order, and never change the outcome (their panics are recovered and logged).

A step veto (BeforeStep returning an error) fails the step immediately with no retries. Hooks are invoked concurrently for parallel steps, so a plugin must be safe for concurrent use. In-process plugins are trusted code.

type JSONCodec 

type JSONCodec = engine.JSONCodec

JSONCodec is the default codec (goccy/go-json).

type Kind 

type Kind string

Kind distinguishes single-task runs from workflow runs. 

const (
	KindTask     Kind = store.KindTask
	KindWorkflow Kind = store.KindWorkflow
)

type LogEntry 

type LogEntry struct {
	RunID   string    `json:"run_id"`
	Step    string    `json:"step"`
	At      time.Time `json:"at"`
	Level   string    `json:"level"`
	Message string    `json:"message"`
}

LogEntry is one task log line captured via TaskLogger.

type Metrics 

type Metrics struct {
	Runs   map[Status]int64      `json:"runs"`
	Queues map[string]QueueStats `json:"queues"`
}

Metrics is a global state snapshot.

type MetricsSnapshot 

type MetricsSnapshot = Metrics

MetricsSnapshot is the value handed to the WithMetricsFunc callback; it has the same shape as Metrics.

type Middleware 

type Middleware = engine.Middleware

Middleware decorates task execution. Compose with WithMiddleware / Quacker.Use (engine-wide) or Wrap (per task). The first middleware registered is the outermost: global middlewares run before per-task ones, which run before the task body. A middleware may short-circuit by returning without calling next.

Inside a middleware the step identity is available via StepFromContext, dependency outputs via DepOutput, and task logs can be redirected with WithLogSink: 

q.Use(func(next quacker.Handler) quacker.Handler {
    return func(ctx context.Context, in json.RawMessage) (json.RawMessage, error) {
        started := time.Now()
        ctx = quacker.WithLogSink(ctx, myLogStore)
        out, err := next(ctx, in)
        observe(quacker.RunIDFromContext(ctx), time.Since(started), err)
        return out, err
    }
})

type Option 

type Option func(*config)

Option configures Open.

func WithCheckpointInterval 

func WithCheckpointInterval(d time.Duration) Option

WithCheckpointInterval sets how often WAL-backed storage (Ephemeral, File) runs a passive wal_checkpoint to bound WAL growth. The checkpoint never blocks readers or the writer. <=0 uses the 60s default; values below 10ms are clamped to 10ms. Memory storage has no WAL and never checkpoints.

func WithCodec 

func WithCodec(c Codec) Option

WithCodec sets the payload codec (default JSONCodec). It applies engine-wide and is used for task input/output encoding and decoding, DepOutput, event payloads, and durable RunOnce/WaitFor values.

func WithLeaseTTL 

func WithLeaseTTL(d time.Duration) Option

WithLeaseTTL sets how long a claimed step's lease lasts before a reaper may requeue it (multi-instance leases only). Keep it comfortably longer than a scheduler stall; a running step's lease is extended by a heartbeat at a third of this interval. Default 30s.

func WithLogStorage 

func WithLogStorage(enabled bool) Option

WithLogStorage enables the built-in SQLite task-log store (and q.Logs). It is off by default so a long-lived engine does not grow logs it never reads; when off, lines go to the base sink (engine logger unless WithTaskLogSink is set). When on and no custom sink is set, logs persist to SQLite only. Log lines are written through the same batched path as before.

func WithLogger 

func WithLogger(l *slog.Logger) Option

WithLogger sets the engine logger (default slog.Default()).

func WithMetricsFunc 

func WithMetricsFunc(fn func(*Metrics)) Option

WithMetricsFunc registers a callback invoked with a fresh snapshot every WithMetricsInterval (default 15s). Panics in the callback are recovered and logged. The callback should not block for long; it runs on its own goroutine and never delays the scheduler.

func WithMetricsInterval 

func WithMetricsInterval(d time.Duration) Option

WithMetricsInterval sets how often the metrics callback runs. <=0 uses the 15s default; values below 100ms are clamped to 100ms. Without WithMetricsFunc no loop is started.

func WithMiddleware 

func WithMiddleware(mw ...Middleware) Option

WithMiddleware registers engine-wide middleware, applied outside every task's own Wrap middleware. The first registered is outermost.

func WithPayloadKey 

func WithPayloadKey(key []byte) Option

WithPayloadKey encrypts user payloads at rest with AES-256-GCM. key must be 32 bytes; Open rejects any other length, and rejects combining it with WithCodec.

Every user payload goes through the codec — task input/output, DepOutput, event payloads, and durable RunOnce/WaitFor values — so all of it is encrypted. Schema structures the SQL gates read (steps.depends_on, steps.labels, runs.unique_key, sequence keys) and task logs stay plaintext, and introspection (Execution, Events) returns ciphertext. Use the same key across restarts.

func WithPlugin 

func WithPlugin(p Plugin) Option

WithPlugin registers a plugin's hooks with the engine. Plugins are invoked in registration order (Before* forward, After* reverse). It is the compile-time extension point: implement Plugin in your own package and pass it here — no dynamic loading, no out-of-process server.

Example

ExampleWithPlugin registers a plugin that observes run completion. 

finished := make(chan string, 1)
q, _ := Open(WithStorage(Memory()), WithPlugin(examplePlugin{finished: finished}))
defer q.Close(context.Background())

task := NewTask("greet", func(ctx context.Context, in string) (string, error) {
	return "hi " + in, nil
})
h, _ := Enqueue(context.Background(), q, task, "world")
out, _ := h.Result(context.Background())
fmt.Println(out, <-finished) // OnRunFinished fires around Result

Output:
hi world SUCCEEDED

func WithPollInterval 

func WithPollInterval(d time.Duration) Option

WithPollInterval sets how often the scheduler scans for due work while runs are in flight. When the engine is idle it does not poll at all: it sleeps until the next step's due time, and a wake (an enqueue, retry, resume, or completion) interrupts that sleep. So the interval bounds how often the scheduler re-checks under load, not how late an idle run starts. Default 50ms.

func WithQueue 

func WithQueue(name string, concurrency int) Option

WithQueue sets a queue's maximum concurrency. Queues that are never configured run with a default concurrency of 10.

func WithRate 

func WithRate(queue string, n int, per time.Duration) Option

WithRate caps how many runs a queue may start per window — a sliding window counted over claim timestamps, so the cap survives a File-mode restart instead of allowing a fresh-process burst. The scheduler simply holds due work QUEUED while the window is full; nothing is rejected. n<=0 disables the cap; per<=0 is treated as one second.

func WithRetention 

func WithRetention(p RetentionPolicy) Option

WithRetention enables automatic purging of terminal runs older than the policy's OlderThan, on its own interval. Purging is storage-agnostic and bounds growth in every mode (Memory included). OlderThan must be > 0 for the loop to act; interval defaults to one minute.

Call it more than once to layer policies: a global one (empty Queue) plus per-queue overrides with their own cutoff. Each runs on its own interval and purges only the rows it selects.

func WithStorage 

func WithStorage(s Storage) Option

WithStorage selects the storage backend (default Ephemeral).

func WithTaskLogSink 

func WithTaskLogSink(fn func(LogEntry)) Option

WithTaskLogSink sets the base destination for task log lines, replacing the default engine-logger sink. It composes with WithLogStorage(true), which adds SQLite persistence in addition. A nil fn is ignored.

func WithTracerProvider 

func WithTracerProvider(tp trace.TracerProvider) Option

WithTracerProvider enables OpenTelemetry tracing: a span per enqueue, one per step execution (a new root linked to the enqueue span, the standard async shape), and one per emit. The library imports only the OTel API — pass the provider from your SDK/exporter setup, or omit this to disable tracing entirely (no overhead). When enabled, the W3C traceparent active at enqueue is persisted on the run, so execution links back across restarts.

func WithWorkerID 

func WithWorkerID(id string) Option

WithWorkerID sets this engine's identity for multi-instance step leases. Must be unique per running engine against a shared database; empty generates one. Only meaningful for leases backends (Postgres).

func WithWorkerLabels 

func WithWorkerLabels(labels ...string) Option

WithWorkerLabels declares this engine's worker labels. It claims a task's steps only when their WithLabels set is a subset of these; an engine with no labels claims only unlabeled tasks. Use it to route work to the workers that can run it (e.g. "gpu", "linux").

type Plugin 

type Plugin interface {
	// Name identifies the plugin (for logging/diagnostics).
	Name() string
	// Hooks returns the callbacks to install; leave fields nil to ignore them.
	Hooks() Hooks
}

Plugin contributes lifecycle hooks to an engine.

type PurgeOptions 

type PurgeOptions struct {
	// OlderThan is how long a terminal run must have been finished before it
	// is eligible. Required (> 0). Purging is storage-agnostic: it bounds
	// growth in every mode, including Memory.
	OlderThan time.Duration
	// Statuses restricts the purge to these terminal statuses; empty means
	// all four. A non-terminal status returns ErrNonTerminalPurge.
	Statuses []Status
	// Queue, when non-empty, restricts the purge to runs in that queue.
	Queue string
	// ExcludeDeadLettered keeps dead-lettered runs out of the purge.
	ExcludeDeadLettered bool
	// KeepLogs retains the logs of purged runs (and skips the orphan sweep).
	// The default (false) deletes them with the run.
	KeepLogs bool
	// BatchSize bounds rows deleted per transaction. <= 0 uses 500.
	BatchSize int
}

PurgeOptions selects terminal runs to delete with Quacker.Purge.

type PurgeResult 

type PurgeResult = store.PurgeResult

PurgeResult reports how many rows a purge removed.

type Quacker 

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

Quacker is an embedded orchestration engine. Create with Open; safe for concurrent use.

func Open 

func Open(opts ...Option) (*Quacker, error)

Open starts an engine. State is stored per WithStorage (default Ephemeral: a temp-file WAL database deleted on Close).

func (*Quacker) Cancel 

func (q *Quacker) Cancel(runID string) error

Cancel cancels a run. Queued/blocked steps become CANCELLED immediately; running steps have their contexts cancelled. Waiters receive ErrRunCancelled. Cancelling an unknown or finished run is a no-op.

func (*Quacker) Close 

func (q *Quacker) Close(ctx context.Context) error

Close shuts down gracefully: no new work is claimed, in-flight tasks drain until ctx expires (30s if ctx has no deadline), leftovers are marked INTERRUPTED, and pending Results return ErrRunInterrupted. Ephemeral and Memory storage are deleted; File storage is kept for the next Open.

func (*Quacker) Crons 

func (q *Quacker) Crons() []CronInfo

Crons lists registered cron triggers.

func (*Quacker) DAG 

func (q *Quacker) DAG(ctx context.Context, runID string) (*DAG, error)

DAG returns the step graph of a run with the current state of every node. For a single-task run it is one node with no edges. Returns ErrNotFound for an unknown run. It covers only this run's own steps; use DAGTree to include child runs (spawned with EnqueueChild).

For a grouped workflow (NewGroupWorkflow), member steps carry their group in DAGNode.Group, group-level dependencies become group-to-group edges, and the expanded cross-group step edges are omitted so the graph reads as groups rather than a step mesh.

func (*Quacker) DAGJSON 

func (q *Quacker) DAGJSON(ctx context.Context, runID string) ([]byte, error)

DAGJSON returns a run's DAG, with the current state of every step, as indented JSON — easy to persist or expose over HTTP.

func (*Quacker) DAGSVG 

func (q *Quacker) DAGSVG(ctx context.Context, runID string) ([]byte, error)

DAGSVG returns a run's DAG rendered as a standalone SVG document: steps as status-coloured boxes in dependency order, edges as arrows.

func (*Quacker) DAGTree added in v1.1.0 

func (q *Quacker) DAGTree(ctx context.Context, runID string) (*DAG, error)

DAGTree expands a run's DAG to include its child runs: each child run's steps are added, namespaced by a short child-run id to keep names unique, and attached to the step that spawned them (Recorded as runs.parent_step). It recurses into grandchildren, bounded by depth and node count.

Groups are first-class: a step that spawns child runs is contracted into the group those runs form (the step node disappears; the group box stands in for it), and the group inherits that step's dependencies as group-level edges. A group therefore runs after the groups/nodes it depends on without every member step repeating the dependency. Intra-group task edges are kept.

This is the run-level view: DAG/DAGSVG show only one run's own steps, because a child run is an independent run (EnqueueChild does not join it to the parent). DAGTree is what draws the whole family.

func (*Quacker) DAGTreeJSON added in v1.1.0 

func (q *Quacker) DAGTreeJSON(ctx context.Context, runID string) ([]byte, error)

DAGTreeJSON renders the run tree (parent + child runs, with current state) as indented JSON.

func (*Quacker) DAGTreeSVG added in v1.1.0 

func (q *Quacker) DAGTreeSVG(ctx context.Context, runID string) ([]byte, error)

DAGTreeSVG renders the run tree as a standalone SVG document — the parent DAG with each child run's steps attached to the step that spawned them.

func (*Quacker) DeadLetters 

func (q *Quacker) DeadLetters(ctx context.Context, f DeadLetterFilter) ([]RunSummary, error)

DeadLetters lists runs dead-lettered by a task with WithDeadLetter, newest first.

func (*Quacker) DebugLogger 

func (q *Quacker) DebugLogger() *slog.Logger

DebugLogger returns a logger whose records join the engine's debug stream. Use it to interleave your own diagnostics with the engine's.

func (*Quacker) DebugLogs 

func (q *Quacker) DebugLogs() <-chan DebugRecord

DebugLogs returns a bounded stream of engine debug records — claims, retries, suspensions, run completions, and anything logged through DebugLogger — so the engine can be observed without serving HTTP: 

for rec := range q.DebugLogs() {
    fmt.Println(rec.Level, rec.Message, rec.Attrs)
}

The stream drops records when the consumer falls behind (it never blocks the engine) and closes on Close, ending a range. Only one consumer should read it; use Subscribe for a lossless-per-subscriber transition stream.

func (*Quacker) DismissDeadLetter 

func (q *Quacker) DismissDeadLetter(ctx context.Context, runID string) error

DismissDeadLetter clears a run's dead-letter marker without retrying it.

func (*Quacker) Emit 

func (q *Quacker) Emit(ctx context.Context, event string, payload any) (int, error)

Emit persists an event and dispatches it to every bound task, one run each, returning how many runs were enqueued. payload is marshaled as JSON (a json.RawMessage is used as-is) and becomes each task's input; a payload that does not decode to a bound task's input type fails that run at execution, like any other input. Delivery is best-effort and in-process: unbound events still persist, but a crash between the persist and the enqueues can lose those dispatches.

func (*Quacker) Events 

func (q *Quacker) Events(ctx context.Context, limit int) ([]EventRecord, error)

Events lists up to limit recent emitted events, newest first.

func (*Quacker) Execution 

func (q *Quacker) Execution(ctx context.Context, runID string) (*Execution, error)

Execution returns a snapshot of a run. Safe to call at any moment: it reads through the read-only connection pool and never blocks (or is blocked by) the workers' write path. The run row is read before its steps and step statuses only advance forward, so a terminal run is never shown with non-terminal steps. Returns ErrNotFound for unknown runs.

For workflow runs, Attempts counts total step attempts across the DAG and MaxAttempts mirrors the first step's setting; per-step values are on Steps.

func (*Quacker) Logs 

func (q *Quacker) Logs(ctx context.Context, runID string, limit int) ([]LogEntry, error)

Logs returns up to limit task log lines for a run, oldest first. Lines are emitted with TaskLogger inside task functions and are batched (~50ms), so very recent lines may lag slightly.

func (*Quacker) Metrics 

func (q *Quacker) Metrics(ctx context.Context) (*Metrics, error)

Metrics returns run counts by status and per-queue depths.

func (*Quacker) PauseQueue 

func (q *Quacker) PauseQueue(ctx context.Context, name string) error

PauseQueue stops claims from a queue until ResumeQueue. Running steps finish and new runs accumulate QUEUED. The pause is persisted, so every engine on the database observes it. Pausing an unknown queue is allowed.

func (*Quacker) PauseRun 

func (q *Quacker) PauseRun(ctx context.Context, runID string) error

PauseRun pauses a run: its steps stop being claimed, and a currently-running step is interrupted and re-queued on resume. Idempotent; a terminal run returns ErrRunTerminal and an unknown run ErrNotFound.

func (*Quacker) PausedQueues 

func (q *Quacker) PausedQueues(ctx context.Context) ([]string, error)

PausedQueues returns the paused queue names.

func (*Quacker) Purge 

func (q *Quacker) Purge(ctx context.Context, opts PurgeOptions) (PurgeResult, error)

Purge deletes terminal runs older than opts.OlderThan, along with their steps and (unless KeepLogs) their logs, in batches. It is safe to call while work is running: a run is never eligible while any of its steps is still RUNNING. Use it to bound a long-lived process's growth.

func (*Quacker) RemoveCron 

func (q *Quacker) RemoveCron(name string) error

RemoveCron deletes a cron trigger.

func (*Quacker) ResumeQueue 

func (q *Quacker) ResumeQueue(ctx context.Context, name string) error

ResumeQueue re-enables claims for a queue and wakes the scheduler.

func (*Quacker) ResumeRun 

func (q *Quacker) ResumeRun(ctx context.Context, runID string) error

ResumeRun returns a paused run to QUEUED. A run that is not paused is a no-op; a terminal run returns ErrRunTerminal.

func (*Quacker) RetryDeadLetter 

func (q *Quacker) RetryDeadLetter(ctx context.Context, runID string) error

RetryDeadLetter reopens a dead-lettered run in place and returns it to the queue. It returns ErrNotDeadLetter if the run exists but is not dead-lettered, and ErrNotFound if it does not exist.

func (*Quacker) Runs 

func (q *Quacker) Runs(ctx context.Context, f RunFilter) ([]RunSummary, error)

Runs lists runs matching the filter, newest first.

func (*Quacker) SetQueue 

func (q *Quacker) SetQueue(name string, concurrency int)

SetQueue adjusts a queue's concurrency at runtime.

func (*Quacker) SetRateLimit 

func (q *Quacker) SetRateLimit(name string, n int, window time.Duration)

SetRateLimit adjusts a queue's start-rate cap at runtime — at most n runs started per sliding window. n<=0 disables the cap; window<=0 is one second.

func (*Quacker) Snooze 

func (q *Quacker) Snooze(ctx context.Context, runID string, until time.Time) error

Snooze reschedules a non-running run to start no earlier than until. A run with a RUNNING step returns ErrRunRunning (cancel or pause it first); a terminal run returns ErrRunTerminal.

func (*Quacker) SnoozeFor 

func (q *Quacker) SnoozeFor(ctx context.Context, runID string, d time.Duration) error

SnoozeFor is Snooze(now + d).

func (*Quacker) Subscribe 

func (q *Quacker) Subscribe(runID string) (<-chan Event, func())

Subscribe returns a channel of live status transitions for runID (pass "" to observe all runs) and a cancel function that stops the stream and closes the channel. Events are dropped rather than delivered late if a consumer stalls — use Execution for an authoritative snapshot.

func (*Quacker) Use 

func (q *Quacker) Use(mw ...Middleware)

Use appends engine-wide middleware at runtime; safe before or after work starts. The first registered middleware is the outermost wrapper.

type QueueStats 

type QueueStats struct {
	Queued    int64 `json:"queued"`
	Running   int64 `json:"running"`
	Blocked   int64 `json:"blocked"`
	Suspended int64 `json:"suspended"`
}

QueueStats is a per-queue depth snapshot.

type RetentionPolicy 

type RetentionPolicy struct {
	// OlderThan must be > 0 for the loop to act.
	OlderThan time.Duration
	// Statuses restricts which terminal statuses may be purged; empty means
	// all of them.
	Statuses []Status
	// Queue, when non-empty, restricts the policy to runs in that queue. Use
	// several WithRetention policies (a global one plus per-queue overrides)
	// for different cutoffs per queue.
	Queue string
	// ExcludeDeadLettered keeps dead-lettered runs out of this policy.
	ExcludeDeadLettered bool
	// KeepLogs retains the logs of purged runs.
	KeepLogs bool
	// Interval is how often the purge runs; <= 0 uses one minute (min 1s).
	Interval time.Duration
}

RetentionPolicy configures automatic purging via WithRetention.

type RunError 

type RunError struct {
	RunID   string
	Status  Status
	Message string
}

RunError is returned by RunHandle.Result when a run fails.

func (*RunError) Error 

func (e *RunError) Error() string

type RunFilter 

type RunFilter struct {
	Status   Status
	Queue    string
	Workflow string
	// ParentID limits the result to a run's children.
	ParentID string
	Limit    int
	Offset   int
}

RunFilter selects runs for Runs.

type RunGroup added in v1.2.0 

type RunGroup struct {
	Name string `json:"name"`
	// Deps are the names of groups this group runs after.
	Deps []string `json:"deps,omitempty"`
	// Steps are the member step names.
	Steps []string `json:"steps"`
	// MaxConcurrency caps RUNNING member steps in this run. Zero adds no limit.
	MaxConcurrency int `json:"max_concurrency,omitempty"`
}

RunGroup is a named set of steps that share a group-level dependency. Group deps gate the whole group; member steps do not repeat them.

type RunHandle 

type RunHandle[O any] struct {
	// contains filtered or unexported fields
}

RunHandle tracks an enqueued run.

func Enqueue 

func Enqueue[I any, O any](ctx context.Context, q *Quacker, t *Task[I, O], input I, opts ...EnqueueOption) (*RunHandle[O], error)

Enqueue enqueues one run of task with input.

func EnqueueBatch 

func EnqueueBatch[I any, O any](ctx context.Context, q *Quacker, t *Task[I, O], inputs []I, opts ...EnqueueOption) ([]*RunHandle[O], error)

EnqueueBatch enqueues one run of task per input in a single write transaction, returning handles in input order. It is all-or-nothing: a validation or insert failure creates no runs. Use it to fan out many runs cheaply instead of calling Enqueue in a loop.

func EnqueueChild 

func EnqueueChild[I any, O any](ctx context.Context, q *Quacker, t *Task[I, O], input I, opts ...EnqueueOption) (*RunHandle[O], error)

EnqueueChild enqueues one run of task as a child of the run currently executing (from RunIDFromContext). It is only valid inside a task or workflow step; the child runs independently (no implicit join), and Execution exposes it under Children.

func EnqueueWorkflow 

func EnqueueWorkflow[O any, I any](ctx context.Context, q *Quacker, wf *Workflow[I], input I, opts ...EnqueueOption) (*RunHandle[O], error)

EnqueueWorkflow enqueues a workflow run. All steps receive input; the handle's Result decodes the output of the last declared step, so callers specify O explicitly while I is inferred from the workflow: 

h, err := quacker.EnqueueWorkflow[Shipment](ctx, q, wf, order)

func EnqueueWorkflowChild 

func EnqueueWorkflowChild[O any, I any](ctx context.Context, q *Quacker, wf *Workflow[I], input I, opts ...EnqueueOption) (*RunHandle[O], error)

EnqueueWorkflowChild enqueues a workflow run as a child of the currently executing run. Only valid inside a task or workflow step.

func (*RunHandle[O]) Done 

func (h *RunHandle[O]) Done() <-chan struct{}

Done returns a channel closed when the run reaches a terminal status.

func (*RunHandle[O]) Result 

func (h *RunHandle[O]) Result(ctx context.Context) (O, error)

Result waits for the run to finish and decodes its output. A failed run returns a *RunError; a cancelled run returns ErrRunCancelled.

func (*RunHandle[O]) RunID 

func (h *RunHandle[O]) RunID() string

RunID returns the run's stable identifier.

type RunInfo 

type RunInfo = engine.RunInfo

RunInfo describes a run that reached a terminal status.

type RunSummary 

type RunSummary struct {
	RunID    string `json:"run_id"`
	Workflow string `json:"workflow"`
	Kind     Kind   `json:"kind"`
	Status   Status `json:"status"`
	Queue    string `json:"queue"`
	Priority int64  `json:"priority"`
	Error    string `json:"error,omitempty"`
	ParentID string `json:"parent_id,omitempty"`
	// ParentStep is the step (in ParentID) that spawned this run.
	ParentStep  string    `json:"parent_step,omitempty"`
	CreatedAt   time.Time `json:"created_at"`
	StartedAt   time.Time `json:"started_at,omitempty"`
	CompletedAt time.Time `json:"completed_at,omitempty"`
	// DeadLetteredAt is set when the run was dead-lettered (zero otherwise).
	DeadLetteredAt time.Time `json:"dead_lettered_at,omitempty"`
	// Ephemeral is true for a run that is deleted on terminal.
	Ephemeral bool `json:"ephemeral,omitempty"`
}

RunSummary is a run without payloads, for list views.

type Status 

type Status string

Status is the lifecycle state of a run or step. 

const (
	StatusQueued      Status = store.StatusQueued
	StatusRunning     Status = store.StatusRunning
	StatusBlocked     Status = store.StatusBlocked
	StatusSuspended   Status = store.StatusSuspended
	StatusPaused      Status = store.StatusPaused
	StatusSucceeded   Status = store.StatusSucceeded
	StatusFailed      Status = store.StatusFailed
	StatusCancelled   Status = store.StatusCancelled
	StatusInterrupted Status = store.StatusInterrupted
)

type StepContext 

type StepContext = engine.StepContext

StepContext identifies the currently executing step inside a task function.

func StepFromContext 

func StepFromContext(ctx context.Context) (StepContext, bool)

StepFromContext returns the executing step's identity inside a task function (ok=false otherwise).

type StepDef 

type StepDef[I any] struct {
	// contains filtered or unexported fields
}

StepDef declares one step of a workflow. Create with Step.

func Step 

func Step[I any, O any](name string, t *Task[I, O], deps ...string) StepDef[I]

Step declares a workflow step running task t (which takes the workflow input type I). Optional deps name steps that must succeed first; a step with dependencies stays BLOCKED until all of them succeed.

type StepInfo 

type StepInfo = engine.StepInfo

StepInfo describes a step about to run (or that just ran).

type StepState 

type StepState struct {
	Name string `json:"name"`
	// Task is the registered task that executes the step; it differs from
	// Name for named workflow steps.
	Task   string   `json:"task,omitempty"`
	Status Status   `json:"status"`
	Deps   []string `json:"deps,omitempty"`
	// Key is the step's concurrency key (empty when unkeyed). Steps
	// sharing a key run at most KeyConcurrency at a time.
	Key         string          `json:"key,omitempty"`
	Attempts    int             `json:"attempts"`
	MaxAttempts int             `json:"max_attempts"`
	Timeout     time.Duration   `json:"timeout,omitempty"`
	Input       json.RawMessage `json:"input,omitempty"`
	Output      json.RawMessage `json:"output,omitempty"`
	Error       string          `json:"error,omitempty"`
	RunAt       time.Time       `json:"run_at"`
	CreatedAt   time.Time       `json:"created_at"`
	StartedAt   time.Time       `json:"started_at,omitempty"`
	CompletedAt time.Time       `json:"completed_at,omitempty"`
	// WaitEvent is the event a SUSPENDED step awaits ("" for a sleep or a
	// non-suspended step); ResumeAt is when a sleeping/suspended step becomes
	// claimable again (zero for event-only waits).
	WaitEvent string    `json:"wait_event,omitempty"`
	ResumeAt  time.Time `json:"resume_at,omitempty"`
	// Labels are the worker labels the step requires (empty = any engine).
	Labels []string `json:"labels,omitempty"`
}

StepState is a point-in-time snapshot of one step.

type Storage 

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

Storage selects where run state lives. Construct with Memory, Ephemeral, File, Postgres, or Driver (for a third-party storage driver).

func Driver 

func Driver(name, dsn string) Storage

Driver selects a registered storage driver by name with a connection string. Register drivers from a package init (see the driver package); the built-in names are "sqlite", "postgres", and "mysql". Without the driver's import, Open fails with a clear "driver is not registered" error. 

q, _ := quacker.Open(quacker.WithStorage(quacker.Driver("mysql", dsn)))

func Ephemeral 

func Ephemeral() Storage

Ephemeral keeps state in a temporary WAL-backed database that is deleted on Close. Durability semantics are identical to Memory (nothing survives the process), but readers never block the write path. This is the default.

func File 

func File(path string) Storage

File persists state to path (WAL mode) so runs survive restarts. Runs interrupted by a previous process are handled according to RecoverRunningOnBoot.

func Memory 

func Memory() Storage

Memory keeps all state in a pure in-memory SQLite database. Nothing is written to the filesystem and everything is lost on Close. Note: SQLite :memory: databases cannot use WAL journaling, so under heavy write load introspection reads may briefly wait for an in-flight write transaction.

func Postgres 

func Postgres(dsn string) Storage

Postgres persists state to the database at dsn. It requires importing the driver package for its side effect: 

import _ "github.com/kartikbazzad/quacker/postgres"

Without that import, Open fails with a clear "driver not registered" error.

func PostgresWithDB 

func PostgresWithDB(db *sql.DB) Storage

PostgresWithDB persists state to an existing pgx-backed pool that the caller owns; shorthand for Postgres("").WithDB(db).

func (Storage) RecoverRunningOnBoot 

func (s Storage) RecoverRunningOnBoot(requeue bool) Storage

RecoverRunningOnBoot configures startup recovery for File storage: when true (default) runs left RUNNING/INTERRUPTED by a previous process are re-queued; when false they are marked FAILED.

func (Storage) WithDB 

func (s Storage) WithDB(db *sql.DB) Storage

WithDB hands quacker an existing database/sql pool the caller owns instead of one it dials from a DSN. Supported by single-pool drivers (Postgres via the pgx stdlib driver, MySQL via the go-sql-driver). quacker runs its migrations on the pool but never closes it, so the caller keeps using it after Quacker.Close; pool sizing and lifetime stay the caller's. 

db, _ := sql.Open("pgx", dsn)
q, _ := quacker.Open(quacker.WithStorage(quacker.Postgres("").WithDB(db)))

type Task 

type Task[I, O any] struct {
	// contains filtered or unexported fields
}

Task is a named, typed unit of work. Create with NewTask; the same value is used to enqueue runs, register for restart recovery, and attach crons.

func NewTask 

func NewTask[I, O any](name string, fn func(ctx context.Context, in I) (O, error), opts ...TaskOption) *Task[I, O]

NewTask creates a task named name. The name is the task's identity (see ID) and must be unique per process; it is persisted with run state, so it is part of the storage format — renaming a task orphans its in-flight runs, and registering two different functions under one name silently keeps the last.

func (*Task[I, O]) ID added in v1.1.0 

func (t *Task[I, O]) ID() string

ID returns the task's stable identity — its name, which is also what is persisted with each run and used by Register, Cron, and On. Use it wherever you would otherwise repeat the name as a string (a step name, a workflow dependency), so the name is defined once and cannot be misspelled: 

charge := quacker.NewTask("charge", chargeFn)
wf := quacker.NewWorkflow[Order]("fulfill",
    quacker.Step("charge", charge),
    quacker.Step("ship", ship, charge.ID()), // not the literal "charge"
)

func (*Task[I, O]) Name 

func (t *Task[I, O]) Name() string

Name returns the task's registered name.

type TaskOption 

type TaskOption func(*taskConfig)

TaskOption configures a Task.

func BackoffPolicy 

func BackoffPolicy(b Backoff) TaskOption

Backoff sets the retry delay policy (default: exponential from 500ms, capped at 30s, with 10% jitter).

func Queue 

func Queue(name string) TaskOption

Queue routes the task to a named queue with its own concurrency limit. Default: "default".

func Retries 

func Retries(n int) TaskOption

Retries sets how many times a failed attempt is retried (maxAttempts = retries + 1). Default: no retries.

func Timeout 

func Timeout(d time.Duration) TaskOption

Timeout bounds each attempt. A timed-out attempt is treated like any other failure (retried when retries remain). Default: no timeout.

func WithDeadLetter 

func WithDeadLetter() TaskOption

WithDeadLetter marks a run dead-lettered when it exhausts its retries, so it appears in DeadLetters and can be retried with RetryDeadLetter. Runs of a task without it that fail are ordinary FAILED runs.

func WithEphemeral 

func WithEphemeral() TaskOption

WithEphemeral makes runs of this task ephemeral: persisted while in flight (so claiming, retries, and introspection work) but deleted when they reach a terminal state and discarded on restart rather than recovered. No history remains; a cross-engine Result returns ErrRunGone if the run was deleted before it could be observed.

func WithKey 

func WithKey[I any](fn func(I) string) TaskOption

WithKey extracts a concurrency key from the task input. Steps sharing a key are capped at WithKeyConcurrency (default 1) simultaneous executions across all queues — the scheduler holds a due step QUEUED until its key has a free slot. The cap is enforced by counting RUNNING rows at claim time, so it survives restarts and needs no per-key bookkeeping. A task without WithKey is never key-gated; a key function returning "" leaves that run unkeyed.

func WithKeyConcurrency 

func WithKeyConcurrency(n int) TaskOption

WithKeyConcurrency sets how many runs sharing one WithKey may execute concurrently (default 1 = strict serialization per key). Meaningless without WithKey; values <1 are treated as 1.

func WithKeyLimit 

func WithKeyLimit[I any](name string, fn func(I) string, limit int) TaskOption

WithKeyLimit declares an additional named concurrency key with its own limit. A step is claimed only when every key it declares has a free slot, so declaring several gates on multiple dimensions at once. Keys are scoped by name and value: two tasks using the same name and equal values share one budget, which is how a limit is "shared" across tasks.

func WithKeyStrategy 

func WithKeyStrategy(s ConcurrencyStrategy) TaskOption

WithKeyStrategy selects what happens when a run is enqueued while its key is already at WithKeyConcurrency: ConcurrencyHold (default) queues it, while the cancel strategies cancel running or queued same-key runs (or the incoming run) instead. Meaningless without WithKey.

func WithLabels 

func WithLabels(labels ...string) TaskOption

WithLabels requires worker labels for this task: only an engine opened with WithWorkerLabels covering all of them claims its steps (labels are a subset — an engine with extra labels may still run it). Without a matching engine the runs stay QUEUED. No labels means any engine.

func WithSequence 

func WithSequence[I any](fn func(I) string) TaskOption

WithSequence makes runs of this task execute strictly one-at-a-time in insertion order within a sequence key derived from the input, while different keys run in parallel. An earlier run that is still queued, running, retrying, or suspended holds the line (head-of-line blocking); a key returning "" is unsequenced. Sequences are a stronger guarantee than WithKey's concurrency cap, which does not fix execution order across retries.

func WithUnique 

func WithUnique[I any](fn func(I) string) TaskOption

WithUnique makes runs of this task unique per derived key among non-terminal runs: while a run with the key is QUEUED/RUNNING/BLOCKED/SUSPENDED, another enqueue with the same key resolves by the conflict policy (default UniqueReuse). The key is computed from input at enqueue; returning "" leaves that run not unique. Uniqueness is scoped to the task name.

func WithUniqueConflict 

func WithUniqueConflict(m UniqueConflict) TaskOption

WithUniqueConflict sets what a unique run does when its key is already held: UniqueReuse (default) returns the live run's handle, UniqueError fails with ErrDuplicateJob, UniqueReplace cancels the live run and enqueues the new one.

func Wrap 

func Wrap(mw ...Middleware) TaskOption

Wrap attaches per-task middleware around this task's body. Engine-wide middleware (WithMiddleware / Quacker.Use) still wraps the result, so the order is global → per-task → body.

type UniqueConflict 

type UniqueConflict = store.UniqueConflict

UniqueConflict selects what a unique run does when its key is already held by a live run. Set the default per task with WithUniqueConflict.

type Workflow 

type Workflow[I any] struct {
	// contains filtered or unexported fields
}

Workflow is a DAG of steps sharing one input. Steps run as soon as their dependencies succeed; the run output is the output of the last declared step. If any step exhausts its retries the run fails and remaining steps are cancelled.

func NewGroupWorkflow added in v1.2.0 

func NewGroupWorkflow[I any](name string, groups ...*Group[I]) *Workflow[I]

NewGroupWorkflow creates a workflow from groups. Each group's After dependencies expand to the member steps of those groups, so a group runs only after its dependencies; member steps never repeat the group dep. Group structure is persisted with the run so DAG/DAGSVG draw group boxes and group-to-group edges instead of the expanded step mesh.

func NewWorkflow 

func NewWorkflow[I any](name string, steps ...StepDef[I]) *Workflow[I]

NewWorkflow creates a workflow from step declarations. Step names must be unique and dependencies must form a DAG (cycles are rejected at enqueue).

Example

ExampleNewWorkflow runs a two-step DAG where the second step reads the first step's output. 

q, _ := Open(WithStorage(Memory()))
defer q.Close(context.Background())

charge := NewTask("charge", func(ctx context.Context, in string) (string, error) {
	return "charged:" + in, nil
})
ship := NewTask("ship", func(ctx context.Context, in string) (string, error) {
	rec, err := DepOutput[string](ctx, "charge")
	if err != nil {
		return "", err
	}
	return "shipped(" + rec + ")", nil
})
wf := NewWorkflow[string]("fulfill",
	Step("charge", charge),
	Step("ship", ship, "charge"),
)

h, _ := EnqueueWorkflow[string](context.Background(), q, wf, "o-1")
out, _ := h.Result(context.Background())
fmt.Println(out)

Output:
shipped(charged:o-1)

func (*Workflow[I]) Name 

func (w *Workflow[I]) Name() string

Name returns the workflow's registered name.

Source Files

View all Source files

Directories

Path Synopsis
Package driver defines quacker's storage-driver contract.
 Package driver defines quacker's storage-driver contract.
examples
children command
Child runs: a parent task fans out to per-item child runs, and the lineage is visible through Execution/Runs.
 Child runs: a parent task fans out to per-item child runs, and the lineage is visible through Execution/Runs.
cron command
Cron and delayed triggers: a task that runs on a schedule, plus one-off delayed runs, cancellable while waiting.
 Cron and delayed triggers: a task that runs on a schedule, plus one-off delayed runs, cancellable while waiting.
dag command
A workflow DAG: charge, then ship, then notify — with data flowing from step to step via dependency outputs.
 A workflow DAG: charge, then ship, then notify — with data flowing from step to step via dependency outputs.
dagsvg command
DAG visualization: run a workflow, then emit its current step graph as JSON and as a standalone SVG file.
 DAG visualization: run a workflow, then emit its current step graph as JSON and as a standalone SVG file.
durable command
Durable execution: a task that sleeps and waits for an event without holding a worker slot, resuming where it left off with side effects that run exactly once.
 Durable execution: a task that sleeps and waits for an event without holding a worker slot, resuming where it left off with side effects that run exactly once.
eltdag command
ELT DAG visualization: build an extract -> transform -> stage -> load pipeline (parallel per-shard fan-out, a per-shard series chain, a fan-in, and per-partition child workflows), run it, then save the run's step graph as dag.json and dag.svg.
 ELT DAG visualization: build an extract -> transform -> stage -> load pipeline (parallel per-shard fan-out, a per-shard series chain, a fan-in, and per-partition child workflows), run it, then save the run's step graph as dag.json and dag.svg.
eltgroups command
Grouped ELT DAG: four groups — extract, transform, load, report — where each group fans out into parallel steps, the next group runs only after the previous group (declared once with After), and a step may depend on a sibling in its own group.
 Grouped ELT DAG: four groups — extract, transform, load, report — where each group fans out into parallel steps, the next group runs only after the previous group (declared once with After), and a step may depend on a sibling in its own group.
eltnested command
Prototype: ELT groups as nested (child) workflows.
 Prototype: ELT groups as nested (child) workflows.
events command
In-process events: emit an event and every bound task runs with the event payload as its input.
 In-process events: emit an event and every bound task runs with the event payload as its input.
introspect command
Live introspection: watch a slow workflow's state change in real time while it keeps executing — nothing is paused or slowed down.
 Live introspection: watch a slow workflow's state change in real time while it keeps executing — nothing is paused or slowed down.
simple command
Simplest possible quacker app: define a task, enqueue it, await the result.
 Simplest possible quacker app: define a task, enqueue it, await the result.
internal
bus
Package bus is a tiny in-process pub/sub for run state transitions.
 Package bus is a tiny in-process pub/sub for run state transitions.
engine
Package engine is quacker's execution core: a scheduler that claims due steps from SQLite, per-queue worker pools, retries, DAG progression, cron firing, and graceful shutdown.
 Package engine is quacker's execution core: a scheduler that claims due steps from SQLite, per-queue worker pools, retries, DAG progression, cron firing, and graceful shutdown.
store
Package store owns all persistence for quacker.
 Package store owns all persistence for quacker.
Package mysql registers the MySQL/MariaDB storage driver for quacker.
 Package mysql registers the MySQL/MariaDB storage driver for quacker.
Package postgres registers the PostgreSQL storage backend for quacker.
 Package postgres registers the PostgreSQL storage backend for quacker.

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.