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

engine

package
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: 24 Imported by: 0

Documentation

Overview

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.

Index

Constants

View Source 
const DefaultQueueConcurrency = 10

DefaultQueueConcurrency is the concurrency assigned to queues that were never configured via SetQueue.

Variables

View Source 
var (
	// ErrRunCancelled is returned to callers waiting on a cancelled run.
	ErrRunCancelled = errors.New("quacker: run cancelled")
	// ErrRunInterrupted is returned when a run was dropped by shutdown.
	ErrRunInterrupted = errors.New("quacker: run interrupted by shutdown")
	// ErrClosed is returned by enqueue calls after Close.
	ErrClosed = errors.New("quacker: engine is closed")
	// ErrUnknownTask is returned by RegisterCron for an unregistered task.
	ErrUnknownTask = errors.New("quacker: unknown task")
	// ErrExternalTxUnsupported is returned by EnqueueOnTx for storage that
	// cannot share a caller-owned transaction (SQLite).
	ErrExternalTxUnsupported = errors.New("quacker: storage does not support enqueue on a caller transaction (Postgres/MySQL only)")
	// ErrRunGone is returned when an ephemeral run was deleted before its
	// result could be observed (e.g. a reused run executed by another engine).
	ErrRunGone = errors.New("quacker: ephemeral run already finished and was discarded")
)
View Source 
var ErrJournalMisaligned = errors.New("quacker: durable journal misaligned (task code changed across a suspend?)")

ErrJournalMisaligned is returned when a durable task's replayed calls do not match its persisted journal — the usual cause is changing task code (reordering, adding, or removing durable calls) while runs are suspended. It is a hard failure rather than silent wrong-typed data.

View Source 
var ErrWaitTimeout = errors.New("quacker: wait timed out")

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

Functions

func DepOutput 

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

DepOutput decodes the recorded output of a succeeded dependency step into out using the engine's codec. It returns an error outside a task function, for unknown dependencies, or on a decode failure.

func RunIDFromContext 

func RunIDFromContext(ctx context.Context) string

RunIDFromContext returns the current run ID (empty outside a task).

func RunOnce 

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

RunOnce runs fn at most once per step journal slot, memoizing its result across suspensions and retries. It never suspends. On fn error the entry is left undone, so a retry re-invokes fn — exactly-once on success, at-least-once on failure. key is validated on replay.

func SleepDurable 

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

SleepDurable suspends the running step for at least d without consuming an attempt or holding a queue or per-key slot. On resume the task is re-invoked from the top and this call returns nil immediately.

Code before a durable await re-executes on resume; wrap side effects in RunOnce. Do not recover() around a durable helper.

func TaskLogger 

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

TaskLogger returns a logger that persists log lines to the run, visible via Logs(runID). Lines are buffered and written in batches; when the buffer is full lines are dropped rather than blocking the task.

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, then decodes its payload into T. A timeout > 0 bounds the wait and yields ErrWaitTimeout; timeout <= 0 waits indefinitely. Only events emitted after the wait registers count (subscription semantics).

func WithLogSink 

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

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

Types

type Backoff 

type Backoff struct {
	// Base is the delay before the first retry. Default 500ms.
	Base time.Duration
	// Factor is the exponential growth multiplier. Default 2.
	Factor float64
	// Max caps the delay. Default 30s.
	Max time.Duration
	// Jitter in [0,1] randomizes the delay upward by up to that fraction to
	// avoid retry thundering herds. Zero means no jitter; the 0.1 default is
	// applied at the task layer (quacker.Exponential/defaultBackoff).
	Jitter float64
}

Backoff configures retry delays. Zero fields fall back to the defaults documented per field.

func (Backoff) Delay 

func (b Backoff) Delay(attempt int, now time.Time) time.Duration

Delay returns the wait before retry number attempt (1-based).

type Codec 

type Codec interface {
	Name() string
	Marshal(v any) ([]byte, error)
	Unmarshal(data []byte, v any) error
}

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 regardless of the codec, because the SQL gates read them.

func CodecFromContext 

func CodecFromContext(ctx context.Context) Codec

CodecFromContext returns the codec of the engine executing the task, or the default JSON codec outside a task (used by task bodies and DepOutput).

type CronInfo 

type CronInfo struct {
	Name string
	Spec string
	Task string
	Next time.Time
}

CronInfo describes a registered cron trigger.

type DebugRecord 

type DebugRecord struct {
	At      time.Time         `json:"at"`
	Level   string            `json:"level"`
	Message string            `json:"message"`
	Attrs   map[string]string `json:"attrs,omitempty"`
}

DebugRecord is one engine debug log line, streamed to consumers via DebugLogs rather than served over HTTP. Attrs is flattened to strings.

type EmitInfo 

type EmitInfo struct{ Event string }

EmitInfo describes an event about to be emitted.

type Engine 

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

Engine executes runs against a store.

func New 

func New(o Options) (*Engine, error)

New returns an engine. Call Start to begin scheduling.

func (*Engine) Cancel 

func (e *Engine) Cancel(runID string) error

Cancel cancels a run. Queued/blocked steps are CANCELLED immediately and running steps have their contexts cancelled. Waiting callers receive ErrRunCancelled. Cancelling an unknown or already-finished run is a no-op.

func (*Engine) Close 

func (e *Engine) Close(ctx context.Context) error

Close stops the engine: no new work is claimed, in-flight executions drain until ctx expires, leftovers are marked INTERRUPTED, and pending waiters are released with ErrRunInterrupted.

func (*Engine) Codec 

func (e *Engine) Codec() Codec

Codec returns the engine's payload codec.

func (*Engine) Crons 

func (e *Engine) Crons() []CronInfo

Crons lists registered cron triggers with their next fire time. Crons loaded from the store but not yet armed (their task is not registered in this process) are included with their persisted next time.

func (*Engine) DeadLetters 

func (e *Engine) DeadLetters(ctx context.Context, workflow, queue string, limit, offset int) ([]*store.Run, error)

DeadLetters lists dead-lettered runs, newest first.

func (*Engine) DebugLogger 

func (e *Engine) DebugLogger() *slog.Logger

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

func (*Engine) DebugLogs 

func (e *Engine) DebugLogs() <-chan DebugRecord

DebugLogs returns the engine's debug stream. It is bounded and drops records when the consumer falls behind; it closes on Close, ending a range. Only one consumer should read it.

func (*Engine) DismissDeadLetter 

func (e *Engine) DismissDeadLetter(ctx context.Context, runID string) error

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

func (*Engine) Emit 

func (e *Engine) Emit(ctx context.Context, name string, payload json.RawMessage) (n int, rerr error)

Emit records an event and dispatches it: it persists the event and wakes every armed durable WaitFor on that event atomically, then enqueues one run per armed On binding. The enqueue half is best-effort in-process: a crash after the atomic wake but before the enqueues loses those run bindings (unbound events and event waits still deliver). It returns how many On binding runs were enqueued.

func (*Engine) Enqueue 

func (e *Engine) Enqueue(ctx context.Context, req *EnqueueRequest) (*Waiter, error)

Enqueue validates req, persists the run and its steps, and returns a waiter for the run. The first wake-up signal is sent to the scheduler. The caller's ctx governs the insert: a cancelled ctx means no run.

func (*Engine) EnqueueBatch 

func (e *Engine) EnqueueBatch(ctx context.Context, reqs []*EnqueueRequest) ([]*Waiter, error)

EnqueueBatch enqueues several runs in one write transaction and returns one waiter per request, in order. It is all-or-nothing: a validation or insert failure rolls the whole batch back. The scheduler is woken once.

func (*Engine) EnqueueBatchOnTx 

func (e *Engine) EnqueueBatchOnTx(ctx context.Context, tx *sql.Tx, reqs []*EnqueueRequest) ([]string, error)

EnqueueBatchOnTx is EnqueueOnTx for several requests, inserted on the same caller transaction. It returns the new run ids in order.

func (*Engine) EnqueueOnTx 

func (e *Engine) EnqueueOnTx(ctx context.Context, tx *sql.Tx, req *EnqueueRequest) (string, error)

EnqueueOnTx inserts one run on a caller-owned transaction and returns its id. It registers no waiter and never commits: the caller owns the transaction, so the insert is atomic with the caller's business writes and the run can execute on any engine after commit. Postgres/MySQL only.

func (*Engine) ForgetTask 

func (e *Engine) ForgetTask(name string)

ForgetTask drops a task registration (used by tests to exercise the unregistered-task park path).

func (*Engine) PauseQueue 

func (e *Engine) PauseQueue(ctx context.Context, name string) error

PauseQueue stops claims from a queue; running steps finish and new work accumulates QUEUED. Persisted, so every engine sharing the store observes it.

func (*Engine) PauseRun 

func (e *Engine) 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 store.ErrRunTerminal, an unknown run store.ErrNotFound.

func (*Engine) PausedQueues 

func (e *Engine) PausedQueues(ctx context.Context) ([]string, error)

PausedQueues returns the paused queue names.

func (*Engine) RegisterCron 

func (e *Engine) RegisterCron(name, spec string, def *TaskDef, input json.RawMessage) error

RegisterCron registers (or replaces) a cron trigger that enqueues task def with input on the schedule. The trigger is persisted, so with File storage it is re-armed on the next Start as soon as the task is registered.

func (*Engine) RegisterEvent 

func (e *Engine) RegisterEvent(event string, def *TaskDef) error

RegisterEvent binds an event name to a task: every Emit of that event enqueues one run of def with the event payload as input. The binding is persisted, so with File storage it re-arms on the next Start once the task is registered. Re-registering the same (event, task) is idempotent and refreshes the task definition.

func (*Engine) RegisterTask 

func (e *Engine) RegisterTask(def *TaskDef)

RegisterTask makes a task executable by name. Registration is idempotent; it overwrites any previous definition. The def is stored as-is (never mutated) so concurrent enqueues of the same task are race-free. Registering a task also arms any persisted crons and event subscriptions that target it.

func (*Engine) RemoveCron 

func (e *Engine) RemoveCron(name string) error

RemoveCron deletes a cron trigger, whether armed or still pending.

func (*Engine) RemoveEvent 

func (e *Engine) RemoveEvent(event, task string) error

RemoveEvent deletes an event→task binding, armed or pending.

func (*Engine) ResumeQueue 

func (e *Engine) ResumeQueue(ctx context.Context, name string) error

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

func (*Engine) ResumeRun 

func (e *Engine) ResumeRun(ctx context.Context, runID string) error

ResumeRun returns a paused run to QUEUED and wakes the scheduler. A run that is not paused is a no-op; a terminal run returns store.ErrRunTerminal.

func (*Engine) RetryDeadLetter 

func (e *Engine) RetryDeadLetter(ctx context.Context, runID string) error

RetryDeadLetter reopens a dead-lettered run and wakes the scheduler.

func (*Engine) SetQueue 

func (e *Engine) SetQueue(name string, concurrency int)

SetQueue configures a queue's concurrency, creating it if needed.

func (*Engine) SetRateLimit 

func (e *Engine) SetRateLimit(name string, n int64, window time.Duration)

SetRateLimit caps how many steps a queue may start per window (a sliding window counted over claim timestamps, so the cap survives a File-mode restart). n<=0 disables the cap; window<=0 is treated as one second.

func (*Engine) Snooze 

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

Snooze moves a non-running run's start time to until. A run with a RUNNING step returns store.ErrRunRunning; a terminal run store.ErrRunTerminal.

func (*Engine) Start 

func (e *Engine) Start()

Start launches the scheduler, cron, and log-flush loops. Queues that exist in the store (e.g. recovered runs after a restart) are registered with default concurrency unless already configured. Persisted crons and event subscriptions are loaded into their pending maps; each arms when its target task is later registered, so a File-mode app only needs Register (not a re-Cron/re-On) at startup.

func (*Engine) Subscribe 

func (e *Engine) Subscribe(runID string, buffer int) (<-chan bus.Event, func())

Subscribe exposes the engine's transition bus. Cancel closes the channel.

func (*Engine) SupportsExternalTx 

func (e *Engine) SupportsExternalTx() bool

SupportsExternalTx reports whether this engine can enqueue on a caller-owned *sql.Tx (Postgres/MySQL, not SQLite).

func (*Engine) Use 

func (e *Engine) Use(mw ...Middleware)

Use appends engine-wide middleware. It is safe to call before or after Start; each execution snapshots the chain under the engine lock.

func (*Engine) WaitersLen 

func (e *Engine) WaitersLen() int

WaitersLen returns the number of unreleased run waiters.

type EnqueueInfo 

type EnqueueInfo struct {
	Workflow string
	Kind     string
	Queue    string
	Priority int64
}

EnqueueInfo describes a run about to be enqueued.

type EnqueueRequest 

type EnqueueRequest struct {
	Workflow string
	Kind     string // store.KindTask or store.KindWorkflow
	Input    json.RawMessage
	Queue    string
	Priority int64
	RunAt    time.Time // zero = now
	// ParentID links this run to the run that enqueued it ("" for a root
	// run). ParentStep is the step that spawned it, so a child run can be
	// attached to its spawning step in the run tree. Lineage is informational.
	ParentID   string
	ParentStep string
	// UniqueKey, when non-empty, makes the run unique per (workflow, key)
	// among non-terminal runs. Conflict selects the outcome on collision.
	UniqueKey string
	Conflict  store.UniqueConflict
	// Ephemeral deletes the run on terminal and skips boot recovery.
	Ephemeral bool
	// Groups, when non-empty, is the run's group structure as JSON (name,
	// group-level deps, member steps), persisted for DAG rendering. Execution
	// does not read it: group deps are already expanded into Steps.
	Groups []byte
	Steps  []StepReq
}

EnqueueRequest creates a run. For a single task there is exactly one step.

type Handler 

type Handler func(ctx context.Context, input json.RawMessage) (json.RawMessage, error)

Handler is the JSON-level signature of a task body. Middleware wraps Handlers.

type Hooks 

type Hooks struct {
	BeforeEnqueue func(ctx context.Context, e EnqueueInfo) (context.Context, error)
	AfterEnqueue  func(ctx context.Context, e EnqueueInfo, err error)
	BeforeStep    func(ctx context.Context, s StepInfo) (context.Context, error)
	AfterStep     func(ctx context.Context, s StepInfo, err error)
	OnRunFinished func(ctx context.Context, r RunInfo)
	BeforeEmit    func(ctx context.Context, e EmitInfo) (context.Context, error)
	AfterEmit     func(ctx context.Context, e EmitInfo, dispatched int, err error)
}

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

type JSONCodec 

type JSONCodec struct{}

JSONCodec is the default codec; it wraps github.com/goccy/go-json.

func (JSONCodec) Marshal 

func (JSONCodec) Marshal(v any) ([]byte, error)

func (JSONCodec) Name 

func (JSONCodec) Name() string

func (JSONCodec) Unmarshal 

func (JSONCodec) Unmarshal(data []byte, v any) error

type KeyDef 

type KeyDef struct {
	Name  string
	Fn    func(c Codec, input json.RawMessage) string
	Limit int
}

KeyDef is a named concurrency key with a limit.

type MetricsSnapshot 

type MetricsSnapshot struct {
	Runs   map[string]int64
	Queues map[string]store.QueueStats
}

MetricsSnapshot is a point-in-time engine-wide state snapshot handed to the OnMetrics callback: run counts keyed by status and per-queue depths.

type Middleware 

type Middleware func(next Handler) Handler

Middleware decorates task execution. 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 not calling next.

type Options 

type Options struct {
	Store *store.Store
	Bus   *bus.Bus
	Log   *slog.Logger
	// PollInterval is how often the scheduler re-checks for due work while runs
	// are in flight. An idle engine instead sleeps until the next due time and
	// wakes early on an enqueue, retry, resume, or completion.
	PollInterval time.Duration
	// Now overrides the clock (tests).
	Now func() time.Time
	// Middleware is the engine-wide middleware chain, applied outside every
	// task's own Wrappers.
	Middleware []Middleware
	// WorkerLabels is this engine's label set; it claims only steps whose
	// task labels are a subset. Empty claims only unlabeled steps.
	WorkerLabels []string
	// TracerProvider, when set, enables OpenTelemetry spans for enqueue, step
	// execution, and emit. The library depends only on the OTel API; the
	// caller supplies the SDK/provider.
	TracerProvider trace.TracerProvider
	// WorkerID identifies this engine for step leases (multi-instance
	// backends). Empty is replaced with a generated id.
	WorkerID string
	// LeaseTTL is how long a claim's lease lasts before the reaper may
	// requeue it (leases mode only). <=0 defaults to 30s.
	LeaseTTL time.Duration
	// Hooks are plugin lifecycle callbacks, invoked in slice order.
	Hooks []Hooks
	// Codec (de)serializes user payloads; nil uses JSONCodec.
	Codec Codec
	// LogSink is the base task-log destination. When nil and LogStorage is
	// off, task logs go to the engine logger; when nil and LogStorage is on,
	// they persist to SQLite only.
	LogSink func(store.LogEntry)
	// LogStorage enables the built-in SQLite task-log sink (and q.Logs).
	LogStorage bool
	// OnMetrics, when set, is called every MetricsInterval with a snapshot.
	OnMetrics       func(MetricsSnapshot)
	MetricsInterval time.Duration
	// Retentions, when set, each run PurgeRuns on their own interval.
	Retentions []*RetentionPolicy
}

Options configures New.

type RetentionPolicy 

type RetentionPolicy struct {
	OlderThan time.Duration
	// Statuses restricts which terminal statuses may be purged; empty means
	// every terminal status.
	Statuses []string
	// Queue, when non-empty, restricts the policy to that queue.
	Queue string
	// ExcludeDeadLettered keeps dead-lettered runs out of the policy.
	ExcludeDeadLettered bool
	// KeepLogs retains the logs of purged runs (and skips the orphan sweep).
	KeepLogs bool
	Interval time.Duration
}

RetentionPolicy configures the automatic purge loop. A zero OlderThan disables purging; a zero Interval uses one minute (clamped to >= 1s).

type RunError 

type RunError struct {
	RunID string
	Msg   string
}

RunError is the error surfaced to waiters when a run fails.

func (*RunError) Error 

func (e *RunError) Error() string

type RunInfo 

type RunInfo struct {
	RunID    string
	Workflow string
	Status   string
	Error    string
}

RunInfo describes a run that reached a terminal status.

type StepContext 

type StepContext struct {
	RunID string
	// Step is the DAG identity of the executing step (its name in a
	// workflow, or the task name for a single-task run).
	Step string
	// Task is the registered task that executes the step. It differs from
	// Step for named workflow steps.
	Task    string
	Attempt int
}

StepContext carries per-execution information into task functions via context.Context.

func StepFromContext 

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

StepFromContext returns the executing step's identity, or ok=false outside a task function.

type StepInfo 

type StepInfo struct {
	RunID   string
	Step    string
	Task    string
	Queue   string
	Attempt int
	Key     string
	Labels  []string
}

StepInfo describes a step about to run (or that just ran), with no internal store types leaked.

type StepReq 

type StepReq struct {
	Name string
	Deps []string
	Def  *TaskDef
	// Group and GroupConcurrency declare an additional per-run group budget.
	Group            string
	GroupConcurrency int
}

StepReq is one step of an enqueue request.

type TaskDef 

type TaskDef struct {
	Name        string
	Queue       string
	MaxAttempts int
	Timeout     time.Duration
	Backoff     Backoff
	Fn          Handler
	// Wrappers are per-task middlewares, applied inside the engine's global
	// middleware chain.
	Wrappers []Middleware
	// Labels are required worker labels: only an engine whose WorkerLabels are
	// a superset claims this task's steps. Empty means any engine.
	Labels []string
	// KeyFn extracts a concurrency key from the step input; steps sharing
	// a key are capped at KeyLimit concurrent executions across all queues.
	// nil means unkeyed.
	KeyFn func(c Codec, input json.RawMessage) string
	// KeyLimit is the max simultaneously-running steps sharing one key;
	// <1 is normalized to 1 at enqueue when a key is present.
	KeyLimit int
	// KeyStrategy selects what an enqueue does when the key is at capacity.
	KeyStrategy store.ConcurrencyStrategy
	// ExtraKeys are additional named concurrency keys, each with its own limit;
	// a step is claimable only when every one has a free slot.
	ExtraKeys []KeyDef
	// DeadLetter marks a run dead-lettered when it exhausts retries, so it can
	// be listed and retried from the dead-letter queue.
	DeadLetter bool
	// SequenceFn extracts a sequence key; runs sharing a key execute strictly
	// one-at-a-time in insertion order. nil means unsequenced.
	SequenceFn func(c Codec, input json.RawMessage) string
	// Ephemeral deletes a run on terminal and skips boot recovery.
	Ephemeral bool
}

TaskDef is the engine-level, non-generic description of a task.

type Waiter 

type Waiter struct {
	RunID string
	// contains filtered or unexported fields
}

Waiter tracks a run created by this process until it reaches a terminal status. Safe for concurrent use.

func (*Waiter) Done 

func (w *Waiter) Done() <-chan struct{}

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

func (*Waiter) Wait 

func (w *Waiter) Wait(ctx context.Context) (status string, output json.RawMessage, err error)

Wait blocks until the run finishes or ctx expires.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.