README
¶
queue (package)
Import: github.com/kausys/azync/queue
User guide: ../queue.md · GoDoc: package docs via go doc / pkg.go.dev.
Role
Producer / worker / manager for durable jobs with source=queue on a shared azync.Core.
Source layout
| File / area | Responsibility |
|---|---|
queue.go, open.go |
New / Open, composition over Core |
producer.go, tx.go |
Enqueue, transactional outbox |
worker.go, register.go |
Consume + typed handlers |
manager.go, cron.go |
Admin API, leader cron |
context.go, options.go |
Job metadata on context, knobs |
Driver surface
Requires Core + job store. Cron needs driver.LeaderElector. No DAG/WorkflowStore.
Public surface (summary)
New/Open→Producer,Worker,ManagerRegister/RegisterCronTxProducer[T]- Enqueue / worker / manager options (see GoDoc)
Boundaries
- Does not own schema migrations (
Core.Migrate). - Does not implement HTTP admin —
Manageris a library API. - Sibling packages (
event,dag,workflow) share Core; they do not importqueue.
Tests
go test ./queue/... · driver conformance via driver/drivertest.
Documentation
¶
Overview ¶
Package queue provides durable background jobs over an azync Core.
Jobs are persisted rows a Producer enqueues and a Worker leases, executes and settles. A job moves through the states pending -> active -> succeeded, with scheduled (a future run_at or a retry backoff), paused (an operator hold) and dead (aborted or out of retries) alongside; succeeded rows are retained as history until the completed-retention vacuum trims them.
Delivery is at-least-once: a worker leases a job for a bounded TTL, renews the lease at half-life while the handler runs, and settles with the lease token as a fencing credential — a worker that lost its lease cannot settle a job now owned by another. Expired leases are reclaimed by the reaper; poison jobs die after too many reaps. Failed handlers reschedule on a deterministic exponential backoff, or per their error's taxonomy (Abort, Retry, RetryAfter, Reportable). A handler panic is recovered and settles as an ordinary failure — retried, then dead-lettered — never crashing the worker process.
Compose a Runtime over a shared Core with New, or standalone with Open. Register handlers with Register / RegisterKind before Worker.Start, and periodic jobs with RegisterCron (leader-elected, deduplicated per occurrence, no backfill). The Manager exposes the admin surface: inspection, retry, archive, pause/resume, purge and vacuums.
Index ¶
- Constants
- func Abort(err error) error
- func Attempt(ctx context.Context) int
- func EnqueuedAt(ctx context.Context) time.Time
- func IsNotFound(err error) bool
- func IsRetry(ctx context.Context) bool
- func JobID(ctx context.Context) uuid.UUID
- func Kind(ctx context.Context) string
- func MaxAttempts(ctx context.Context) int
- func Metadata(ctx context.Context) map[string]string
- func NewContext(parent context.Context, j JobInfo) context.Context
- func Register[T JobArgs](w *Worker, handler func(ctx context.Context, args T) error, ...) error
- func RegisterKind(w *Worker, kind string, ...) error
- func Reportable(err error) error
- func Retry(err error) error
- func RetryAfter(err error, d time.Duration) error
- type AttemptError
- type DailyCount
- type EnqueueOption
- type EnqueueResult
- type JobArgs
- type JobInfo
- type JobListPage
- type JobState
- type JobView
- type Manager
- func (m *Manager) AllStats(ctx context.Context) (QueueStats, error)
- func (m *Manager) Archive(ctx context.Context, id uuid.UUID) error
- func (m *Manager) Delete(ctx context.Context, id uuid.UUID, state JobState) error
- func (m *Manager) Get(ctx context.Context, id uuid.UUID) (*JobView, error)
- func (m *Manager) JobAttempts(ctx context.Context, id uuid.UUID) ([]AttemptError, error)
- func (m *Manager) List(ctx context.Context, queue string, state JobState, page, size int) (JobListPage, error)
- func (m *Manager) ListAllJobs(ctx context.Context, state JobState, page, size int) (JobListPage, error)
- func (m *Manager) ListQueues(ctx context.Context) ([]QueueInfo, error)
- func (m *Manager) NukeAll(ctx context.Context) (NukeReport, error)
- func (m *Manager) Pause(ctx context.Context, id uuid.UUID) error
- func (m *Manager) Purge(ctx context.Context, queue string) (PurgeReport, error)
- func (m *Manager) Resume(ctx context.Context, id uuid.UUID) error
- func (m *Manager) Retry(ctx context.Context, id uuid.UUID) error
- func (m *Manager) RetryAll(ctx context.Context, queue string) (int64, error)
- func (m *Manager) RunNow(ctx context.Context, id uuid.UUID) error
- func (m *Manager) Stats(ctx context.Context, queue string) (QueueStats, error)
- func (m *Manager) VacuumDead(ctx context.Context, queue string, olderThan time.Duration) (int64, error)
- type NukeReport
- type Option
- func WithCompletedRetention(d time.Duration) Option
- func WithCoreOptions(opts ...azync.Option) Option
- func WithCron(enabled bool) Option
- func WithCronTick(d time.Duration) Option
- func WithDeadRetention(d time.Duration) Option
- func WithDefaultConcurrency(n int) Option
- func WithDefaultJobTimeout(d time.Duration) Option
- func WithDefaultMaxRetries(n int) Option
- func WithFetchBatchSize(n int) Option
- func WithFetchCooldown(d time.Duration) Option
- func WithFetchPollInterval(d time.Duration) Option
- func WithIdleBackoffMax(d time.Duration) Option
- func WithLeaseTTL(d time.Duration) Option
- func WithMaxConcurrency(n int) Option
- func WithMaxReaps(n int) Option
- func WithShutdownDrain(d time.Duration) Option
- func WithStatsRetention(d time.Duration) Option
- type Producer
- type PurgeReport
- type QueueInfo
- type QueueStats
- type RegisterOption
- type Runtime
- type TxProducerClient
- type Worker
Constants ¶
const ( StatePending = driver.StatePending StateScheduled = driver.StateScheduled StateActive = driver.StateActive StateDead = driver.StateDead StatePaused = driver.StatePaused StateSucceeded = driver.StateSucceeded )
Job lifecycle states, re-exported from the driver contract.
Variables ¶
This section is empty.
Functions ¶
func Attempt ¶
Attempt is the 1-based execution attempt; the first run is attempt 1. Zero outside a job.
func EnqueuedAt ¶
EnqueuedAt is when the job was durably inserted. Zero outside a job.
func IsNotFound ¶
IsNotFound reports whether err is the queue's not-found/wrong-state error.
func MaxAttempts ¶
MaxAttempts is the resolved retry budget for the job. Zero outside a job.
func Metadata ¶
Metadata returns the string-valued annotations attached at enqueue time. Nil outside a job.
func NewContext ¶
NewContext returns a copy of parent carrying j, so a handler can be exercised in isolation in a test without a running worker: build a JobInfo, attach it, and the accessors below read from it exactly as they do in production.
func Register ¶
func Register[T JobArgs](w *Worker, handler func(ctx context.Context, args T) error, opts ...RegisterOption) error
Register binds the handler for T's kind on the worker: sugar over RegisterKind that decodes the payload into T and hands the handler the pure domain value. Job metadata travels on ctx (JobID, Attempt, IsRetry, ...). It fails on duplicate kinds and after Start — registration happens in the composition root, before the worker runs.
func RegisterKind ¶
func RegisterKind(w *Worker, kind string, handler func(ctx context.Context, payload json.RawMessage) error, opts ...RegisterOption) error
RegisterKind binds a raw handler for an explicit kind string — the seam for dynamic kinds, where the handler receives the undecoded JSON payload and reads its metadata from ctx (JobID, Kind, Attempt, ...). Same rules as Register: it fails on duplicate kinds (typed or raw) and after Start.
func Reportable ¶
Reportable retries like Retry but flags the error for loud reporting when retries are exhausted.
Types ¶
type AttemptError ¶
type AttemptError = driver.AttemptError
AttemptError is one recorded failure in a job's retry history.
type DailyCount ¶
type DailyCount = driver.DailyCount
DailyCount is one day of throughput counters for a kind.
type EnqueueOption ¶
type EnqueueOption func(*enqueueOptions)
EnqueueOption customizes one Enqueue.
func At ¶
func At(t time.Time) EnqueueOption
At schedules the job for an absolute time (wins over Delay).
func Delay ¶
func Delay(d time.Duration) EnqueueOption
Delay schedules the job for now()+d (resolved on the backend clock).
func IdempotencyKey ¶
func IdempotencyKey(k string) EnqueueOption
IdempotencyKey deduplicates while a live job (pending/scheduled/active/ paused) holds the key; completion or death frees it.
func IdempotencyKeyTTL ¶
func IdempotencyKeyTTL(k string, window time.Duration) EnqueueOption
IdempotencyKeyTTL deduplicates within a time window that survives job completion (cron occurrences, webhook deliveries).
func MaxRetries ¶
func MaxRetries(n int) EnqueueOption
MaxRetries overrides the retry budget for this job.
func Meta ¶
func Meta(key, value string) EnqueueOption
Meta attaches one metadata entry (repeatable).
type EnqueueResult ¶
EnqueueResult reports the outcome of an Enqueue.
type JobArgs ¶
type JobArgs interface {
Kind() string
}
JobArgs identifies a unit of work by its wire-stable Kind (decoupled from the Go type path), e.g. "auth.email_otp.send".
type JobInfo ¶
type JobInfo struct {
ID uuid.UUID
Kind string
Attempt int // 1-based: first execution is attempt 1
MaxAttempts int
EnqueuedAt time.Time
Meta map[string]string
}
JobInfo is the cross-cutting metadata of one job execution. Handlers receive the decoded arguments as their typed value; everything about the job itself — its id, kind, attempt and publish-time annotations — travels on the context and is read through the package accessors (JobID, Kind, Attempt, ...).
type JobListPage ¶
JobListPage is one page of jobs for a queue+state.
type JobView ¶
type JobView struct {
ID uuid.UUID
Kind string
State JobState
Attempt int
MaxAttempts int
EnqueuedAt time.Time
RunAt time.Time
LeaseDeadline time.Time
// StartedAt is when the current attempt was leased, so
// CompletedAt.Sub(StartedAt) is execution time — RunAt cannot answer that,
// being the due time a retry backoff or snooze rewrites. Zero before the
// first lease.
StartedAt time.Time
FailedAt time.Time
CompletedAt time.Time
Payload []byte
Meta map[string]string
LastError string
}
JobView is the admin projection of one job. Optional timestamps are zero when absent (IsZero reports absence).
type Manager ¶
type Manager struct {
// contains filtered or unexported fields
}
Manager is the queue administration surface: inspection, retry, archive, pause/resume, purge and vacuum. Pure library — no auth, no HTTP; embed it behind your own ops endpoints. It operates the queue source only.
func (*Manager) AllStats ¶
func (m *Manager) AllStats(ctx context.Context) (QueueStats, error)
AllStats returns system-wide stats: depths summed across every kind plus the cross-queue daily throughput window (zero-filled, oldest first). Queue is empty to signal the aggregate scope.
func (*Manager) Archive ¶
Archive parks a pending/scheduled job in the dead letter without running it.
func (*Manager) Delete ¶
Delete removes one job in the given state (active jobs cannot be deleted — pass their real state; a running job's row is owned by its worker).
func (*Manager) JobAttempts ¶
JobAttempts returns one job's failure history, oldest attempt first.
func (*Manager) List ¶
func (m *Manager) List(ctx context.Context, queue string, state JobState, page, size int) (JobListPage, error)
List returns one page of jobs (page is 0-based).
func (*Manager) ListAllJobs ¶
func (m *Manager) ListAllJobs(ctx context.Context, state JobState, page, size int) (JobListPage, error)
ListAllJobs returns one page of jobs of a state across every queue (each JobView carries its own Kind so the caller can show which queue a job belongs to).
func (*Manager) ListQueues ¶
ListQueues returns every known kind (live jobs or recent stats).
func (*Manager) NukeAll ¶
func (m *Manager) NukeAll(ctx context.Context) (NukeReport, error)
NukeAll wipes every queue job, stat and idempotency key — dev reset only.
func (*Manager) Purge ¶
Purge empties the queue's pending, scheduled and dead jobs. Active jobs are owned by their workers and paused jobs were parked deliberately by an operator — both survive.
func (*Manager) RetryAll ¶
RetryAll re-enqueues every dead job of the queue (single statement). An empty queue targets every kind.
func (*Manager) RunNow ¶ added in v0.0.6
RunNow expedites one scheduled job — a retry backoff or a Snooze — to run immediately: run_at moves to now, the job returns to pending and workers wake. It returns a not-found error (see IsNotFound) when the job is not scheduled.
type Option ¶
type Option func(*config) error
Option configures a queue Runtime. Options compose; later options win.
func WithCompletedRetention ¶
WithCompletedRetention overrides how long succeeded jobs are kept. A negative value is rejected; zero means retain forever.
func WithCoreOptions ¶
WithCoreOptions forwards options to the Core that Open builds internally (schema, logger, notify channel, shared defaults...). Valid only with Open; New rejects it because the Core is already constructed.
func WithCronTick ¶
WithCronTick overrides how often the cron leader checks its schedules (default 30s). Must be positive.
func WithDeadRetention ¶ added in v0.0.4
WithDeadRetention overrides how long dead (exhausted-retry) jobs are kept. A negative value is rejected; zero (the default) means retain forever.
func WithDefaultConcurrency ¶
WithDefaultConcurrency overrides the per-kind concurrency used when a registration does not set its own. Must be positive.
func WithDefaultJobTimeout ¶
WithDefaultJobTimeout overrides the default per-job wall clock applied to registrations that do not set their own (default 5m). Must be positive; a single kind can still opt out with the WithJobTimeout(0) register option.
func WithDefaultMaxRetries ¶
WithDefaultMaxRetries overrides the retry budget applied to jobs enqueued without an explicit budget. Must be positive.
func WithFetchBatchSize ¶
WithFetchBatchSize overrides how many jobs one dequeue leases. Must be positive.
func WithFetchCooldown ¶
WithFetchCooldown overrides the pause after a productive fetch. Must be positive.
func WithFetchPollInterval ¶
WithFetchPollInterval overrides the idle polling period. Must be positive.
func WithIdleBackoffMax ¶
WithIdleBackoffMax overrides the idle backoff cap of the fetch loops. Must be positive.
func WithLeaseTTL ¶
WithLeaseTTL overrides the shared lease duration for this runtime. Must be positive.
func WithMaxConcurrency ¶
WithMaxConcurrency overrides the total concurrent-handler cap. Must be positive.
func WithMaxReaps ¶
WithMaxReaps overrides how many lease expirations a job survives before the reaper kills it. Must be positive.
func WithShutdownDrain ¶
WithShutdownDrain overrides how long Start waits for in-flight jobs on shutdown. Must be positive.
func WithStatsRetention ¶
WithStatsRetention overrides how long daily stat counters are kept. A negative value is rejected; zero means retain forever.
type Producer ¶
type Producer struct {
// contains filtered or unexported fields
}
Producer enqueues jobs.
func (*Producer) Enqueue ¶
func (p *Producer) Enqueue(ctx context.Context, args JobArgs, opts ...EnqueueOption) (EnqueueResult, error)
Enqueue durably inserts a job for args.Kind(). It returns Deduplicated=true when an idempotency key dropped the insert. A producer span wraps the call (SpanKindProducer), and ctx's trace context (if any) travels with the job via Meta so the handler's consumer span becomes its child (see engine.ExtractTraceContext).
type PurgeReport ¶
PurgeReport counts what Purge removed per state, plus the active jobs it deliberately left running.
type QueueInfo ¶
type QueueInfo struct {
Name string
Namespace string // the kind, display-only
Pending int64
Scheduled int64
Active int64
Paused int64
Dead int64
Succeeded int64
}
QueueInfo identifies one queue (= job kind) for the selector, with its instantaneous per-state counters.
type QueueStats ¶
type QueueStats struct {
Queue string
Pending int64
Scheduled int64
Active int64
Dead int64
Paused int64
Succeeded int64
Enqueued int64 // window totals
Processed int64
Failed int64
Reaped int64
WindowDays int
Daily []DailyCount
}
QueueStats is the admin stats payload: instantaneous depths plus the daily throughput window (zero-filled, oldest first) and its totals.
type RegisterOption ¶
type RegisterOption func(*registerOptions)
RegisterOption customizes Register.
func WithConcurrency ¶
func WithConcurrency(n int) RegisterOption
WithConcurrency caps how many jobs of this kind run at once (default WithDefaultConcurrency).
func WithJobTimeout ¶
func WithJobTimeout(d time.Duration) RegisterOption
WithJobTimeout overrides the per-job wall clock for this kind (default WithDefaultJobTimeout on the runtime; 0 = unlimited).
func WithMaxRetries ¶
func WithMaxRetries(n int) RegisterOption
WithMaxRetries overrides the retry budget for jobs of this kind (default WithDefaultMaxRetries). The override is resolved durably on a job's first lease unless the job was enqueued with an explicit MaxRetries.
type Runtime ¶
type Runtime struct {
// contains filtered or unexported fields
}
Runtime is the queue system over one azync Core: the Producer, the Worker and the Manager, all operating the queue job source only.
func New ¶
New composes a queue runtime over a shared Core. Settings start from the Core's defaults and queue options override them per runtime.
func Open ¶
Open builds a standalone queue runtime that owns a private Core opened from dsn (pass Core options through WithCoreOptions). Close closes the owned Core. Open never migrates; call Migrate before using a fresh schema.
func (*Runtime) Close ¶
Close releases the runtime's resources: the private Core when the runtime was built with Open, nothing when it composes over a shared Core. When the runtime owns its Core, Close first waits (bounded by ctx) for a running Worker to finish draining, so the store is not closed out from under in-flight settlements; on timeout it logs a warning and closes anyway rather than hanging indefinitely.
func (*Runtime) Migrate ¶
Migrate brings the backend schema up to date (requires a driver.Migrator). Open and New never migrate automatically.
type TxProducerClient ¶
type TxProducerClient[TTx any] struct { // contains filtered or unexported fields }
TxProducerClient enqueues jobs inside the caller's own backend transaction, so the enqueue commits atomically with the caller's writes (outbox pattern). Build one with TxProducer.
func TxProducer ¶
func TxProducer[TTx any](r *Runtime) (*TxProducerClient[TTx], error)
TxProducer builds the transactional enqueue client for the driver's transaction handle type TTx (e.g. pgx.Tx for the pg driver). It fails immediately when the runtime's driver does not support transactional enqueues for that type.
func (*TxProducerClient[TTx]) EnqueueTx ¶
func (c *TxProducerClient[TTx]) EnqueueTx(ctx context.Context, tx TTx, args JobArgs, opts ...EnqueueOption) (EnqueueResult, error)
EnqueueTx performs Enqueue within tx, letting the caller atomically commit application writes and the enqueue.
type Worker ¶
type Worker struct {
// contains filtered or unexported fields
}
Worker is the job runtime: per-kind fetch loops feeding an executor pool, the maintenance loops (promotion, reaper, vacuums), and the leader-elected cron scheduler. Handlers register via Register before Start.
func (*Worker) Ready ¶
func (w *Worker) Ready() <-chan struct{}
Ready closes after wakeup setup succeeds and the polling loops are running. Polling-only workers become ready immediately after Start.
func (*Worker) RegisterCron ¶
func (w *Worker) RegisterCron(name, spec string, args JobArgs, opts ...EnqueueOption) error
RegisterCron schedules args to be enqueued on the cron spec (standard 5-field or @descriptors). Missed occurrences are not backfilled on first acquisition: the leader starts counting from "now"; a leader that later re-acquires after briefly losing leadership resumes from where it left off instead (see cronLoop), so a failover window does not silently skip a due occurrence.
opts must not set an idempotency key (IdempotencyKey or IdempotencyKeyTTL): cron already dedupes each occurrence with its own key, and a caller override would replace that key, disabling cron's only defense against a duplicate fire during a leadership handover.
func (*Worker) Start ¶
Start runs the worker until ctx is cancelled: the shared engine (fetch, execute, settle, maintenance) plus the cron leader loop when cron schedules are registered and the driver supports leader election. On cancellation in-flight jobs drain for up to the shutdown drain budget.
Start fails immediately, without running anything, if cron schedules are registered but the driver has no leader-election capability: running cron schedules unelected would enqueue every occurrence once per process, not once per cluster. Disable cron explicitly with WithCron(false) to run without it on such a driver.
func (*Worker) Wait ¶ added in v0.0.4
Wait blocks until a Start call has returned, or ctx ends first, whichever comes first. If Start was never called, Wait returns immediately (there is nothing to wait for). Close uses Wait to avoid closing a shared store out from under an in-flight drain; callers coordinating their own shutdown (stop ctx, then Wait, then release other resources) should do the same.
Source Files