quacker

package module

Go to main page

Versions in this module

v1

v1.4.0

Oct 8, 2026

Changes in this version

type Group

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

type RunGroup

+ MaxConcurrency int

v1.3.1

Sep 21, 2026

v1.3.0

Sep 21, 2026

v1.2.0

Sep 20, 2026

Changes in this version

type DAGGroup

+ Deps []string

+ Parent string

+ Steps []string

type Execution

+ Groups []RunGroup

+ type Group struct

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

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

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

+ type RunGroup struct

+ Deps []string

+ Name string

+ Steps []string

type Workflow

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

v1.1.1

Sep 20, 2026

Changes in this version

+ const Version

v1.1.0

Sep 19, 2026

Changes in this version

type DAG

+ Groups []DAGGroup

+ type DAGGroup struct

+ Label string

+ Name string

+ Status Status

type DAGNode

+ Group string

type Quacker

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

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

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

type RunSummary

+ ParentStep string

type Task

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

v1.0.0

Sep 19, 2026

Changes in this version

+ const ConcurrencyCancelInProgress

+ const ConcurrencyCancelNewest

+ const ConcurrencyCancelQueuedExceptNewest

+ const ConcurrencyCancelQueuedExceptOldest

+ const ConcurrencyHold

+ const UniqueError

+ const UniqueReplace

+ const UniqueReuse

+ var ErrClosed = engine.ErrClosed

+ var ErrDuplicateJob = store.ErrDuplicateJob

+ var ErrExternalTxUnsupported = engine.ErrExternalTxUnsupported

+ var ErrJournalMisaligned = engine.ErrJournalMisaligned

+ var ErrNonTerminalPurge = store.ErrNonTerminalPurge

+ var ErrNotDeadLetter = store.ErrNotDeadLetter

+ var ErrNotFound = store.ErrNotFound

+ var ErrRunCancelled = engine.ErrRunCancelled

+ var ErrRunGone = engine.ErrRunGone

+ var ErrRunInterrupted = engine.ErrRunInterrupted

+ var ErrRunRunning = store.ErrRunRunning

+ var ErrRunTerminal = store.ErrRunTerminal

+ var ErrWaitTimeout = engine.ErrWaitTimeout

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

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

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

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

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

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

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

+ func RunIDFromContext(ctx context.Context) string

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

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

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

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

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

+ type Backoff = engine.Backoff

+ func Constant(d time.Duration) Backoff

+ func Exponential(base time.Duration) Backoff

+ type Codec = engine.Codec

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

+ type ConcurrencyStrategy = store.ConcurrencyStrategy

+ type CronInfo struct

+ Name string

+ Next time.Time

+ Spec string

+ Task string

+ type DAG struct

+ Edges []DAGEdge

+ Kind Kind

+ Nodes []DAGNode

+ Queue string

+ RunID string

+ Status Status

+ Workflow string

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

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

+ type DAGEdge struct

+ From string

+ To string

+ type DAGNode struct

+ Attempts int

+ Deps []string

+ Error string

+ MaxAttempts int

+ Name string

+ Status Status

+ Task string

+ type DeadLetterFilter struct

+ Limit int

+ Offset int

+ Queue string

+ Workflow string

+ type DebugRecord = engine.DebugRecord

+ type EmitInfo = engine.EmitInfo

+ type EnqueueInfo = engine.EnqueueInfo

+ type EnqueueOption func(*enqueueConfig)

+ func WithDelay(d time.Duration) EnqueueOption

+ func WithPriority(p int64) EnqueueOption

+ func WithRunAt(t time.Time) EnqueueOption

+ func WithUniqueKey(key string) EnqueueOption

+ type Event struct

+ At time.Time

+ Error string

+ From Status

+ RunID string

+ Step string

+ To Status

+ type EventRecord struct

+ At time.Time

+ Name string

+ Payload json.RawMessage

+ type Execution struct

+ Attempts int

+ Children []RunSummary

+ CompletedAt time.Time

+ CreatedAt time.Time

+ Error string

+ Input json.RawMessage

+ Key string

+ Kind Kind

+ MaxAttempts int

+ Output json.RawMessage

+ Priority int64

+ Queue string

+ RunAt time.Time

+ RunID string

+ StartedAt time.Time

+ Status Status

+ Steps []StepState

+ TraceParent string

+ Workflow string

+ type Handler = engine.Handler

+ type Hooks = engine.Hooks

+ type JSONCodec = engine.JSONCodec

+ type Kind string

+ const KindTask

+ const KindWorkflow

+ type LogEntry struct

+ At time.Time

+ Level string

+ Message string

+ RunID string

+ Step string

+ type Metrics struct

+ Queues map[string]QueueStats

+ Runs map[Status]int64

+ type MetricsSnapshot = Metrics

+ type Middleware = engine.Middleware

+ type Option func(*config)

+ func WithCheckpointInterval(d time.Duration) Option

+ func WithCodec(c Codec) Option

+ func WithLeaseTTL(d time.Duration) Option

+ func WithLogStorage(enabled bool) Option

+ func WithLogger(l *slog.Logger) Option

+ func WithMetricsFunc(fn func(*Metrics)) Option

+ func WithMetricsInterval(d time.Duration) Option

+ func WithMiddleware(mw ...Middleware) Option

+ func WithPayloadKey(key []byte) Option

+ func WithPlugin(p Plugin) Option

+ func WithPollInterval(d time.Duration) Option

+ func WithQueue(name string, concurrency int) Option

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

+ func WithRetention(p RetentionPolicy) Option

+ func WithStorage(s Storage) Option

+ func WithTaskLogSink(fn func(LogEntry)) Option

+ func WithTracerProvider(tp trace.TracerProvider) Option

+ func WithWorkerID(id string) Option

+ func WithWorkerLabels(labels ...string) Option

+ type Plugin interface

+ Hooks func() Hooks

+ Name func() string

+ type PurgeOptions struct

+ BatchSize int

+ ExcludeDeadLettered bool

+ KeepLogs bool

+ OlderThan time.Duration

+ Queue string

+ Statuses []Status

+ type PurgeResult = store.PurgeResult

+ type Quacker struct

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

+ type QueueStats struct

+ Blocked int64

+ Queued int64

+ Running int64

+ Suspended int64

+ type RetentionPolicy struct

+ ExcludeDeadLettered bool

+ Interval time.Duration

+ KeepLogs bool

+ OlderThan time.Duration

+ Queue string

+ Statuses []Status

+ type RunError struct

+ Message string

+ RunID string

+ Status Status

+ func (e *RunError) Error() string

+ type RunFilter struct

+ Limit int

+ Offset int

+ ParentID string

+ Queue string

+ Status Status

+ Workflow string

+ type RunHandle struct

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

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

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

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

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

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

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

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

+ type RunInfo = engine.RunInfo

+ type RunSummary struct

+ CompletedAt time.Time

+ CreatedAt time.Time

+ DeadLetteredAt time.Time

+ Ephemeral bool

+ Error string

+ Kind Kind

+ ParentID string

+ Priority int64

+ Queue string

+ RunID string

+ StartedAt time.Time

+ Status Status

+ Workflow string

+ type Status string

+ const StatusBlocked

+ const StatusCancelled

+ const StatusFailed

+ const StatusInterrupted

+ const StatusPaused

+ const StatusQueued

+ const StatusRunning

+ const StatusSucceeded

+ const StatusSuspended

+ type StepContext = engine.StepContext

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

+ type StepDef struct

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

+ type StepInfo = engine.StepInfo

+ type StepState struct

+ Attempts int

+ CompletedAt time.Time

+ CreatedAt time.Time

+ Deps []string

+ Error string

+ Input json.RawMessage

+ Key string

+ Labels []string

+ MaxAttempts int

+ Name string

+ Output json.RawMessage

+ ResumeAt time.Time

+ RunAt time.Time

+ StartedAt time.Time

+ Status Status

+ Task string

+ Timeout time.Duration

+ WaitEvent string

+ type Storage struct

+ func Driver(name, dsn string) Storage

+ func Ephemeral() Storage

+ func File(path string) Storage

+ func Memory() Storage

+ func Postgres(dsn string) Storage

+ func PostgresWithDB(db *sql.DB) Storage

+ func (s Storage) RecoverRunningOnBoot(requeue bool) Storage

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

+ type Task struct

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

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

+ type TaskOption func(*taskConfig)

+ func BackoffPolicy(b Backoff) TaskOption

+ func Queue(name string) TaskOption

+ func Retries(n int) TaskOption

+ func Timeout(d time.Duration) TaskOption

+ func WithDeadLetter() TaskOption

+ func WithEphemeral() TaskOption

+ func WithKeyConcurrency(n int) TaskOption

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

+ func WithKeyStrategy(s ConcurrencyStrategy) TaskOption

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

+ func WithLabels(labels ...string) TaskOption

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

+ func WithUniqueConflict(m UniqueConflict) TaskOption

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

+ func Wrap(mw ...Middleware) TaskOption

+ type UniqueConflict = store.UniqueConflict

+ type Workflow struct

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

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

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL