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

queue

package
v0.30.0 Latest Latest
Warning

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

 Go to latest Published: Oct 19, 2026 License: MIT Imports: 21 Imported by: 0

Documentation

Overview

Package queue runs typed background jobs. A job is a struct with a Kind; its handler is a plain function registered explicitly: 

type SendInvoice struct{ OrderID int64 }
func (SendInvoice) Kind() string { return "send_invoice" }

q.Register(queue.Handle(func(ctx context.Context, j SendInvoice) error { ... },
	queue.Retries(5), queue.Backoff(queue.Exponential(time.Second)), queue.Timeout(time.Minute)))

err := q.Dispatch(ctx, SendInvoice{OrderID: 1}, queue.OnQueue("billing"), queue.Delay(5*time.Minute))

A handler whose failure retrying cannot fix wraps ErrPermanent to skip its remaining retries; AttemptFrom tells it which attempt it is on and whether that is the last one.

Payloads are JSON. Drivers store messages: Sync runs jobs inline, Memory keeps them in process, the database driver (and Redis) persist them. Queue.Work runs a worker (vukem queue:work); Fake records dispatched jobs in tests. Apps get a queue from app.Queue(), configured by QUEUE_CONNECTION.

See https://github.com/AlexGalhardo/vukem/blob/main/docs/subsystems.md#queue.

Example 
package main

import (
	"context"
	"fmt"
	"log/slog"
	"time"

	"github.com/AlexGalhardo/vukem/queue"
)

type SendInvoice struct{ OrderID int64 }

func (SendInvoice) Kind() string { return "send_invoice" }

func main() {
	ctx := context.Background()

	// Sync runs jobs inline; apps use the database or Redis driver and a
	// worker (vukem queue:work).
	q := queue.New(queue.Sync{}, slog.New(slog.DiscardHandler))

	err := q.Register(queue.Handle(func(_ context.Context, j SendInvoice) error {
		fmt.Println("sending invoice for order", j.OrderID)
		return nil
	}, queue.Retries(5), queue.Backoff(queue.Exponential(time.Second))))
	if err != nil {
		fmt.Println(err)
		return
	}

	err = q.Dispatch(ctx, SendInvoice{OrderID: 1}, queue.OnQueue("billing"))
	fmt.Println(err)
}

Output:
sending invoice for order 1
<nil>

Index

Examples

Constants

View Source 
const (
	DefaultTable       = "jobs"
	DefaultFailedTable = "failed_jobs"
)

DefaultTable and DefaultFailedTable are the tables the queue owns when DatabaseOptions leaves them empty. They are listed with every other table vukem owns in docs/database.md#tables-the-framework-owns; set QUEUE_TABLE and QUEUE_FAILED_TABLE to move them out of the way of an app's own tables.

View Source 
const MaxFreeInterruptions = 3

MaxFreeInterruptions is how many shutdowns in a row may interrupt a job without using one of its attempts (see WorkOptions.StopTimeout).

View Source 
const MigrationID = "20261018000000_vukem_create_queue_tables"

MigrationID is the id of the migration Migrations returns; it names the framework's own migration in vukem migrate:status.

Variables

View Source 
var ErrDuplicateKind = errors.New("queue: job kind already registered")

ErrDuplicateKind is returned by Register for a kind registered twice.

View Source 
var ErrInlineDelay = errors.New(
	"queue: the sync driver cannot delay jobs (use QUEUE_CONNECTION=memory or database, or queue.Fake in tests)",
)

ErrInlineDelay is returned when a delayed job is dispatched to an Inline driver (Sync): running it now would run it early, so it is refused.

View Source 
var ErrNoTx = errors.New("queue: driver does not support DispatchTx")

ErrNoTx is returned by DispatchTx when ex is neither a *database.DB nor a *database.Tx and the driver cannot enqueue inside a transaction.

View Source 
var ErrPermanent = errors.New("queue: permanent failure, will not be retried")

ErrPermanent marks a failure that retrying cannot fix (a corrupt payload, a deleted row, a rejected input). A handler wraps it in its own error: 

if errors.Is(err, image.ErrFormat) {
	return fmt.Errorf("convert %d: %w: %w", j.ID, err, queue.ErrPermanent)
}

The job skips its remaining retries and moves to the failed jobs with that error, so it stays visible to queue:failed and queue:retry instead of disappearing. With an Inline driver (Sync) there are no retries to skip and the error reaches the caller unchanged.

View Source 
var ErrReservationLost = errors.New("queue: job reservation lost to another worker")

ErrReservationLost is returned by Ack, Release and Fail when the job's reservation expired and another worker reserved it again: the job belongs to that worker now, so the stale result is dropped.

View Source 
var ErrUnknownKind = errors.New("queue: no handler for job kind")

ErrUnknownKind is returned by Dispatch for a kind without a handler.

Functions

func Constant 

func Constant(d time.Duration) func(int) time.Duration

Constant always waits d.

func Dispatched 

func Dispatched[J Job](f *Fake) []J

Dispatched returns the dispatched jobs of type J, decoded, in order.

func Exponential 

func Exponential(base time.Duration) func(int) time.Duration

Exponential backs off base, 2·base, 4·base... capped at one hour.

func Migrations added in v0.29.0 

func Migrations(o DatabaseOptions) []schema.Migration

Migrations returns the migration that creates the queue tables named by o, so an app can manage them with vukem migrate instead of leaving them to the boot-time CreateTables: 

vukem.WithMigrations(slices.Concat(
	stores.Migrations(stores.Options{}),
	migrations.All(),
))

The migration runs the same statements as CreateTables, which stays as the fallback for apps that do not register it. On a database where the tables already exist (the driver created them before the app opted in) the migration records itself and changes nothing, so an upgrade never fails on a conflicting CREATE TABLE; Down drops both tables.

Types

type Attempt added in v0.29.0 

type Attempt struct {
	// Number is the 1-based attempt number (1 on the first run).
	Number int
	// MaxRetries is how many retries the job was dispatched with, so the
	// job gets at most MaxRetries+1 attempts. It is 0 with an Inline
	// driver (Sync), which does not retry.
	MaxRetries int
}

Attempt is what a handler knows about the attempt it is running; read it with AttemptFrom.

func AttemptFrom added in v0.29.0 

func AttemptFrom(ctx context.Context) (Attempt, bool)

AttemptFrom returns the attempt the handler is running on, and whether there is one: it is false outside a queue handler (a handler function called directly, say from a test).

Check the second value before using the first. A zero Attempt is Number 0 and MaxRetries 0, so Last reports 0 > 0, which is false: it reads as "not the last attempt", and a handler that settles the work it owns only on Last would never settle it when called outside a queue.

func (Attempt) Last added in v0.29.0 

func (a Attempt) Last() bool

Last reports whether this attempt is the job's last, so a handler can settle the work it owns (mark a row failed, notify) instead of leaving it mid-flight when its retries run out. A handler that already knows retrying is pointless returns ErrPermanent rather than waiting for the last attempt.

type Clock added in v0.25.0 

type Clock interface {
	SetClock(now func() time.Time)
}

Clock is implemented by drivers whose availability checks read a clock (Memory, Database); app.Queue sets the app clock on them.

type Database 

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

Database stores jobs in SQL tables (SQLite, Postgres or MySQL). Workers reserve jobs atomically (Postgres and MySQL use FOR UPDATE SKIP LOCKED), so any number of worker processes can share the tables. A job whose worker died is reserved again once its timeout plus a grace period has passed: delivery is at least once, so handlers should be idempotent.

func NewDatabase 

func NewDatabase(db *database.DB, o DatabaseOptions) (*Database, error)

NewDatabase returns a database driver; call CreateTables (or a migration with the same schema) before use.

func (*Database) Ack 

func (d *Database) Ack(ctx context.Context, m *Message) error

Ack implements Driver.

func (*Database) CreateTables 

func (d *Database) CreateTables(ctx context.Context) error

CreateTables creates the jobs and failed jobs tables if missing. It is the fallback for apps that do not register Migrations: the statements are the same, so running it after the migration changes nothing.

func (*Database) Fail 

func (d *Database) Fail(ctx context.Context, m *Message) error

Fail implements Driver: the job moves to the failed table.

func (*Database) Failed 

func (d *Database) Failed(ctx context.Context) ([]FailedJob, error)

Failed implements FailedStore, newest first.

func (*Database) Forget 

func (d *Database) Forget(ctx context.Context, ids ...string) (int, error)

Forget implements FailedStore: it deletes failed jobs (all when ids is empty).

func (*Database) Pending 

func (d *Database) Pending(ctx context.Context) (int, error)

Pending returns the number of jobs waiting or running.

func (*Database) Pop 

func (d *Database) Pop(ctx context.Context, queues []string) (*Message, error)

Pop implements Driver.

func (*Database) Push 

func (d *Database) Push(ctx context.Context, m *Message) error

Push implements Driver.

func (*Database) PushTx 

func (d *Database) PushTx(ctx context.Context, ex database.Executor, m *Message) error

PushTx inserts m with ex, so it is enqueued only if ex's transaction commits (Queue.DispatchTx).

func (*Database) Release 

func (d *Database) Release(ctx context.Context, m *Message) error

Release implements Driver.

func (*Database) Retry 

func (d *Database) Retry(ctx context.Context, ids ...string) (int, error)

Retry implements FailedStore: the jobs (all when ids is empty) go back to the queue with their attempts reset. The batch shares one seq, so retried jobs of the same millisecond pop in id order.

func (*Database) SetClock added in v0.25.0 

func (d *Database) SetClock(now func() time.Time)

SetClock replaces DatabaseOptions.Now; it implements Clock.

type DatabaseOptions 

type DatabaseOptions struct {
	// Table holds pending jobs (default "jobs").
	Table string
	// FailedTable holds failed jobs (default "failed_jobs").
	FailedTable string
	// Grace is added to a job's timeout before an abandoned reservation is
	// taken over (default 1 minute).
	Grace time.Duration
	// Now is the clock that decides when delayed and abandoned jobs are
	// due (default time.Now); app.Queue passes the app clock.
	Now func() time.Time
}

DatabaseOptions name the tables.

type Driver 

type Driver interface {
	// Push stores m; it becomes available at m.AvailableAt.
	Push(ctx context.Context, m *Message) error
	// Pop reserves the next available message on one of queues (in
	// order), or returns nil when there is none.
	Pop(ctx context.Context, queues []string) (*Message, error)
	// Ack deletes a processed message.
	Ack(ctx context.Context, m *Message) error
	// Release makes m available again at m.AvailableAt (a retry).
	Release(ctx context.Context, m *Message) error
	// Fail moves m to the failed jobs.
	Fail(ctx context.Context, m *Message) error
}

Driver stores messages and hands them to workers.

type FailedJob 

type FailedJob struct {
	Message

	FailedAt time.Time `json:"failed_at"`
}

FailedJob is a job that ran out of attempts.

type FailedStore 

type FailedStore interface {
	// Failed lists failed jobs, newest first.
	Failed(ctx context.Context) ([]FailedJob, error)
	// Retry re-queues failed jobs (all when ids is empty).
	Retry(ctx context.Context, ids ...string) (int, error)
	// Forget deletes failed jobs (all when ids is empty).
	Forget(ctx context.Context, ids ...string) (int, error)
}

FailedStore is implemented by drivers that keep failed jobs (Database).

type Fake 

type Fake struct {
	*Memory
	// contains filtered or unexported fields
}

Fake records dispatched jobs for tests; it stores them like Memory, so a test can also run them with Queue.Work. 

fake := queue.NewFake()
app := vukemtest.New(t, bootstrap.New, vukemtest.Options{AppOptions: []vukem.Option{vukem.WithQueueDriver(fake)}})
... exercise the app ...
invoices := queue.Dispatched[jobs.SendInvoice](fake)
Example 
package main

import (
	"context"
	"fmt"
	"log/slog"

	"github.com/AlexGalhardo/vukem/queue"
)

type SendInvoice struct{ OrderID int64 }

func (SendInvoice) Kind() string { return "send_invoice" }

func main() {
	ctx := context.Background()
	fake := queue.NewFake()
	q := queue.New(fake, slog.New(slog.DiscardHandler))

	// Dispatch only accepts registered kinds; the Fake stores the job
	// instead of running it.
	_ = q.Register(queue.Handle(func(context.Context, SendInvoice) error { return nil }))
	_ = q.Dispatch(ctx, SendInvoice{OrderID: 7})

	for _, j := range queue.Dispatched[SendInvoice](fake) {
		fmt.Println("dispatched order", j.OrderID)
	}
}

Output:
dispatched order 7

func NewFake 

func NewFake() *Fake

NewFake returns an empty fake driver.

func (*Fake) Messages 

func (f *Fake) Messages() []Message

Messages returns every dispatched message in order.

func (*Fake) Push 

func (f *Fake) Push(ctx context.Context, m *Message) error

Push implements Driver and records m.

type Inline 

type Inline interface {
	Inline() bool
}

Inline is implemented by drivers that run jobs during Dispatch (Sync).

type Job 

type Job interface {
	Kind() string
}

Job is a unit of background work. Kind names its handler and must be stable across deploys (it is stored with the payload).

type JobEvent added in v0.19.0 

type JobEvent struct {
	ID    string
	Kind  string
	Queue string
	// Status is the step reported.
	Status JobStatus
	// Attempt is the 1-based attempt number (0 on JobDispatched).
	Attempt int
	// MaxRetries is how many retries the job was dispatched with, so
	// Attempt of MaxRetries+1 is its last one. It is 0 with an Inline
	// driver (Sync), which does not retry.
	MaxRetries int
	// Final reports, on JobFailed, that the job was not retried: it moved
	// to the failed jobs, either because the handler returned
	// ErrPermanent or because its retries ran out. It is false on the
	// other statuses, and false for an attempt a shutdown gave back (see
	// MaxFreeInterruptions), which Attempt and MaxRetries alone cannot
	// tell apart from an exhausted one.
	Final    bool
	Duration time.Duration
	Err      error
}

JobEvent describes one step of a job for an Observer. It carries no payload: payloads may hold user data.

type JobStatus added in v0.19.0 

type JobStatus string

JobStatus is the step a JobEvent reports. 

const (
	// JobDispatched reports that the job was stored (or is about to run inline).
	JobDispatched JobStatus = "dispatched"
	// JobProcessed reports that an attempt succeeded.
	JobProcessed JobStatus = "processed"
	// JobFailed reports that an attempt failed (it may still be retried).
	JobFailed JobStatus = "failed"
)

Job steps reported to an Observer.

type Memory 

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

Memory keeps messages in process: jobs are lost on restart. Use it for tests and single-process apps that can afford that.

func NewMemory 

func NewMemory() *Memory

NewMemory returns an empty in-memory driver.

func (*Memory) Ack 

func (d *Memory) Ack(_ context.Context, m *Message) error

Ack implements Driver.

func (*Memory) Fail 

func (d *Memory) Fail(_ context.Context, m *Message) error

Fail implements Driver.

func (*Memory) Failed 

func (d *Memory) Failed() []Message

Failed returns the failed messages.

func (*Memory) Pending 

func (d *Memory) Pending() int

Pending returns the number of waiting (including delayed) messages.

func (*Memory) Pop 

func (d *Memory) Pop(_ context.Context, queues []string) (*Message, error)

Pop implements Driver: the earliest available message of the first queue in queues that has one.

func (*Memory) Push 

func (d *Memory) Push(_ context.Context, m *Message) error

Push implements Driver.

func (*Memory) Release 

func (d *Memory) Release(_ context.Context, m *Message) error

Release implements Driver.

func (*Memory) SetClock added in v0.25.0 

func (d *Memory) SetClock(now func() time.Time)

SetClock sets the clock Pop compares availability with (default time.Now); it implements Clock.

type Message 

type Message struct {
	ID          string          `json:"id"`
	Kind        string          `json:"kind"`
	Queue       string          `json:"queue"`
	Payload     json.RawMessage `json:"payload"`
	Attempts    int             `json:"attempts"`
	MaxRetries  int             `json:"max_retries"`
	Timeout     time.Duration   `json:"timeout"`
	AvailableAt time.Time       `json:"available_at"`
	CreatedAt   time.Time       `json:"created_at"`
	// LastError is set when the job failed.
	LastError string `json:"last_error,omitempty"`
}

Message is a stored job.

type Observer added in v0.19.0 

type Observer interface {
	ObserveJob(ctx context.Context, e JobEvent)
}

Observer is notified when jobs are dispatched and attempted. It runs on the caller's goroutine, so it must be fast and safe for concurrent use.

type Option 

type Option func(*options)

Option configures a handler (defaults) or a single dispatch.

func Backoff 

func Backoff(fn func(attempt int) time.Duration) Option

Backoff sets the delay before retry attempt n (1-based).

func Delay 

func Delay(d time.Duration) Option

Delay makes the job available after d.

func OnQueue 

func OnQueue(name string) Option

OnQueue selects the queue (default "default").

func Retries 

func Retries(n int) Option

Retries sets how many times a failed job is retried (default 3).

func Timeout 

func Timeout(d time.Duration) Option

Timeout bounds one attempt (default 1 minute).

type Queue 

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

Queue dispatches jobs to a driver and runs them in workers.

func New 

func New(driver Driver, log *slog.Logger) *Queue

New returns a queue on driver. log may be nil.

func (*Queue) AddObserver added in v0.25.0 

func (q *Queue) AddObserver(obs Observer)

AddObserver adds obs after the queue's current observers (devtools next to a tracing observer). Call it before the queue is used.

func (*Queue) Dispatch 

func (q *Queue) Dispatch(ctx context.Context, j Job, opts ...Option) error

Dispatch enqueues j. Options override the handler's defaults for this dispatch. With an Inline driver the job runs now and its error is returned; Inline drivers cannot delay jobs, so Delay returns ErrInlineDelay there.

func (*Queue) DispatchKind 

func (q *Queue) DispatchKind(ctx context.Context, kind string, payload any, opts ...Option) error

DispatchKind enqueues payload (encoded as JSON) for the handler of kind.

func (*Queue) DispatchKindTx added in v0.25.0 

func (q *Queue) DispatchKindTx(ctx context.Context, ex database.Executor, kind string, payload any,
	opts ...Option,
) error

DispatchKindTx is DispatchTx for a payload sent to the handler of kind (framework jobs such as queued mail, see DispatchKind).

func (*Queue) DispatchTx 

func (q *Queue) DispatchTx(ctx context.Context, ex database.Executor, j Job, opts ...Option) error

DispatchTx enqueues j with the transaction of ex: the job exists only if the transaction commits, so it never runs for rolled-back work and is never lost for committed work.

Drivers with transactional storage (Database) insert the job inside the transaction. Other drivers (Sync, Memory, Fake) dispatch it right after the commit (database.Tx.AfterCommit): Sync runs it then, and since the caller already returned, a handler error is logged instead of returned. Errors known up front (unknown kind, ErrInlineDelay) are returned inside the transaction. With a *database.DB (no open transaction) the job is dispatched now.

func (*Queue) Driver 

func (q *Queue) Driver() Driver

Driver returns the queue's driver.

func (*Queue) Message 

func (q *Queue) Message(j Job, opts ...Option) (*Message, error)

Message builds the stored message for j (drivers with their own transactional push use it).

func (*Queue) Process 

func (q *Queue) Process(ctx context.Context, m *Message) error

Process runs one reserved message and acks, releases (retry with backoff) or fails it.

func (*Queue) Register 

func (q *Queue) Register(regs ...Registration) error

Register adds handlers; a kind may be registered once.

func (*Queue) SetClock added in v0.25.0 

func (q *Queue) SetClock(now func() time.Time)

SetClock sets the clock that stamps dispatched jobs and schedules retries (default time.Now). Delays only follow it when the driver compares availability with the same clock (Memory.SetClock, DatabaseOptions.Now); app.Queue wires the app clock (WithClock) into both.

func (*Queue) SetObserver added in v0.19.0 

func (q *Queue) SetObserver(obs Observer)

SetObserver replaces the queue's observers with obs (development tooling such as devtools; nil, or a typed nil pointer, removes them). Call it before the queue is used; AddObserver keeps the existing ones.

func (*Queue) Work 

func (q *Queue) Work(ctx context.Context, o WorkOptions) error

Work processes jobs until ctx is cancelled. On cancellation it stops reserving new jobs and waits for running ones (graceful drain): they keep a context that is not cancelled, bounded by their timeout and by StopTimeout.

type Registration 

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

Registration binds a handler to its job kind; create it with Handle.

func Handle 

func Handle[J Job](fn func(ctx context.Context, j J) error, opts ...Option) Registration

Handle registers fn for jobs of type J with default options.

func HandleKind 

func HandleKind[T any](kind string, fn func(ctx context.Context, v T) error, opts ...Option) Registration

HandleKind registers fn for jobs named kind whose payload decodes into T (framework jobs such as queued mail and listeners; DispatchKind sends them).

func (Registration) Kind 

func (r Registration) Kind() string

Kind returns the job kind of the registration.

type Sync 

type Sync struct{}

Sync runs jobs inline during Dispatch (development, tests, simple apps). Retries do not apply (the handler's error goes to the caller) and delayed jobs are refused with ErrInlineDelay; DispatchTx runs the job after the transaction commits.

func (Sync) Ack 

func (Sync) Ack(context.Context, *Message) error

Ack implements Driver.

func (Sync) Fail 

func (Sync) Fail(context.Context, *Message) error

Fail implements Driver.

func (Sync) Inline 

func (Sync) Inline() bool

Inline implements Inline.

func (Sync) Pop 

func (Sync) Pop(context.Context, []string) (*Message, error)

Pop implements Driver.

func (Sync) Push 

func (Sync) Push(context.Context, *Message) error

Push implements Driver (never called for Sync).

func (Sync) Release 

func (Sync) Release(context.Context, *Message) error

Release implements Driver.

type TxPusher 

type TxPusher interface {
	PushTx(ctx context.Context, ex database.Executor, m *Message) error
}

TxPusher is implemented by drivers that can enqueue inside a database transaction (Database).

type WorkOptions 

type WorkOptions struct {
	// Queues are polled in priority order (default ["default"]).
	Queues []string
	// Concurrency is the number of parallel workers (default 1).
	Concurrency int
	// Sleep is the wait when no job is available (default 1s).
	Sleep time.Duration
	// MaxJobs stops after processing this many jobs (0 = no limit), e.g.
	// for "queue:work --once".
	MaxJobs int
	// StopWhenEmpty returns when no job is available. A reserve error then
	// ends Work with that error (without it, the error is logged and the
	// worker keeps polling).
	StopWhenEmpty bool
	// StopTimeout bounds the drain once ctx is cancelled: jobs still
	// running after it have their context cancelled. A job that then fails
	// with that cancellation (context.Canceled or DeadlineExceeded) goes
	// back to the queue without using an attempt, MaxFreeInterruptions
	// times in a row; later interruptions use attempts, and other errors
	// are retried or failed as usual. Jobs run again from the start, so
	// they must be idempotent. 0 waits for them (each is still bounded by
	// its own timeout). Jobs must honor ctx; one that ignores it still
	// holds Work.
	StopTimeout time.Duration
}

WorkOptions configure Work.

Source Files

View all Source files

Directories

Path Synopsis
Package queuetest is the conformance suite for queue drivers:
 Package queuetest is the conformance suite for queue drivers:

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.