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

queue

package
v0.39.0 Latest Latest
Warning

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

 Go to latest Published: Oct 2, 2026 License: MIT Imports: 16 Imported by: 0

Documentation

Overview

Package queue runs jobs in the background: work a request starts and doesn't wait for, such as sending a mail. A job is kept by a Store, such as a table in the app's database, from the moment it's pushed until it has run, so a mail server that's down, or an app that restarts, doesn't lose it. A job that fails runs again, after a wait that grows each time, until it works or runs out of attempts, and one that runs out is kept, with its error, for someone to look at. 

q := queue.New(queue.Config{Store: store})
welcome := queue.Handle(q, "welcome-mail", func(ctx context.Context, m WelcomeMail) error {
	...
})
app.Go(q.Run) // tug's App: the workers start and stop with the app
...
err := welcome.Push(ctx, WelcomeMail{User: u.ID})

A job runs at least once. Its worker holds it for a while, and if the app is killed before the job is done, it runs again once the hold is up: a handler should be safe to run twice, as a mail sent twice is.

Where the jobs are kept is the Store's business: the auth starter, tug new -auth, keeps them in its database, SQLite, Postgres or MySQL. Package queuetest checks that a Store keeps its promises, and has one in memory for tests.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func Permanent 

func Permanent(err error) error

Permanent marks err as one that trying again won't mend: a handler that returns it fails its job at once, however many attempts it has left. A payload that doesn't fit the handler's type fails its job this way.

Types

type AtOnceStore added in v0.24.0 

type AtOnceStore interface {
	Store

	// KeepsAtOnce does nothing. It says that the Store keeps the promise,
	// which is in its pushes and its claims, for Handle to know.
	KeepsAtOnce()
}

AtOnceStore is a Store that also keeps a kind's jobs from running more than so many at once, for AtOnce. Its pushes keep a job's AtOnce, and Claim doesn't return a job while as many jobs of its Kind as its AtOnce says are held by claims, on any instance: of claims at once, no more than that many get one.

type Carrier added in v0.34.0 

type Carrier interface {
	Carry(ctx context.Context, into map[string]string)
	Restore(ctx context.Context, from map[string]string) context.Context
}

A Carrier is what a job takes from the context it's pushed from, and gives back to the context it runs in: the ID of the request that pushed it, say, or a trace's parent. Carry puts what it takes from ctx in into, by name, and Restore returns ctx with what from has of it. It's the shape of OpenTelemetry's propagators, with a map for their carrier, as propagation.MapCarrier is, so one of them adapts in a few lines.

Config.Carry lists the queue's. package middleware's CarryRequestID is one, of the ID RequestID gives a request.

type CarryStore added in v0.34.0 

type CarryStore interface {
	Store

	// KeepsCarried does nothing. It says that the Store keeps the promise,
	// which is in its pushes, its claims and its listing, for a push to
	// know.
	KeepsCarried()
}

CarryStore is a Store that also keeps what a job carried from the context it was pushed from, Job.Carried, for Config.Carry: its pushes keep it, a unique job's and a latest one's among them, and Claim, and a FailedStore's Failed, give it back, through the job's retries and a run again. A push that pushes nothing, as a unique one whose job waits, or one that moves the job that waits, as a latest one, leaves the job what its own push carried. A job pushed to a Store that isn't one carries nothing.

type Config 

type Config struct {
	// Store keeps the jobs.
	Store Store

	// Workers is how many jobs Run runs at once. Default 4.
	Workers int

	// Poll is how often Run, with nothing to do, asks the Store for jobs
	// that have come due, or that another instance of the app pushed. A
	// job pushed through this Queue wakes it at once. Default a second.
	Poll time.Duration

	// Grace is how long the jobs running when Run is told to stop have to
	// finish. Those still running after it have their contexts canceled,
	// and are put back to run at the next start. Default 10 seconds.
	Grace time.Duration

	// Observe, when it's set, is told how each run of a job went, as Run
	// and Drain run it, once the Store has kept it: its kind and attempt,
	// what the queue made of it, how long its handler ran, and how long it
	// waited past its time. It's for the app's metrics, as a count of the
	// runs by kind and outcome, which a client of its monitor's keeps. It's
	// called in the worker that ran the job, which waits for it, so it's
	// quick, as a counter is.
	Observe func(Ran)

	// Carry is what a job takes from the context it's pushed from, and
	// gives back to the context it runs in: middleware.CarryRequestID, the
	// ID of the request that pushed it, which the queue's log lines of the
	// job say too, or a trace's parent. A job carries it in a Store that
	// keeps it, a CarryStore, and nothing in one that doesn't.
	Carry []Carrier
}

Config is how a Queue runs its jobs. Every field but Store has a default.

type FailedStore added in v0.10.0 

type FailedStore interface {
	Store

	// Failed returns the jobs that failed for good, with their Error and
	// FailedAt, the latest first.
	Failed(ctx context.Context) ([]*Job, error)

	// RunAgain puts the failed job with the ID back to run, due at at, with
	// no attempts, and says whether there was one: a job that isn't there,
	// or hasn't failed, is left alone.
	RunAgain(ctx context.Context, id string, at time.Time) (bool, error)
}

FailedStore is a Store that also lists the jobs that failed for good, and puts them back to run, for a command of the app's to look at them with.

type HoldBackStore added in v0.24.0 

type HoldBackStore interface {
	Store

	// HoldBack puts j, which its claim holds, back as it was before the
	// claim: due at its RunAt, with the attempt the claim added taken back,
	// and with its Key again, unless another job of its Kind has the key
	// now. And Claim returns no job of j's Kind before until, on any
	// instance. Like Done, Retry and Fail, it leaves the job alone once a
	// later claim has taken it.
	HoldBack(ctx context.Context, j *Job, until time.Time) error
}

HoldBackStore is a Store that also puts a claimed job back as its claim found it, and holds its kind back, for Rate: a job its kind's limit refused hasn't run, and no job of the kind should start until the limit lets one.

type Job 

type Job struct {
	// ID is the Store's name for the job, which Push sets.
	ID string

	// Kind is the name of the handler that runs it, as given to Handle.
	Kind string

	// Payload is the value it was pushed with, as JSON.
	Payload []byte

	// Key is what makes a job of a unique kind the same as another of its
	// kind, which a UniqueStore keeps while the job waits: its payload's
	// hash. It's "" for the jobs of other kinds, and for a claimed job,
	// whose key the claim let go.
	Key string

	// RunAt is when it's due: when it was pushed, or for later, and after
	// an attempt that failed, when it's to run again.
	RunAt time.Time

	// Attempts is how many times it has been claimed, the latest claim
	// included. It tells one claim of a job from the next.
	Attempts int

	// Error is what its latest attempt failed with, for Retry and Fail to
	// keep, and for someone to read.
	Error string

	// OneAtATime keeps the job from running beside another of its Kind and
	// Key, as a OneAtATime kind's jobs are kept: a OneAtATimeStore keeps
	// their key for it as long as it keeps the job, where Key goes at the
	// claim.
	OneAtATime bool

	// AtOnce is the most jobs of its Kind that may run at once, an AtOnce
	// kind's, or 0 for no limit: an AtOnceStore keeps it with the job, and
	// Claim passes the job over while that many of its Kind are held.
	AtOnce int

	// FailedAt is when the job failed for good, which Fail keeps and a
	// FailedStore lists; zero for a job that hasn't.
	FailedAt time.Time

	// Carried is what the context the job was pushed from carried, by
	// name, as Config.Carry's Carriers took it: the ID of the request that
	// pushed it, say. A CarryStore keeps it beside the job, and gives it
	// back with the job's claims, and with its listing as failed. nil for
	// nothing carried.
	Carried map[string]string
}

A Job is one piece of work, as a Store keeps it.

type Kind 

type Kind[T any] struct {
	// contains filtered or unexported fields
}

Kind is a kind of job, as Handle made it: the name its jobs are kept by, and the handler that runs them. Its Push puts one on the queue.

func Handle 

func Handle[T any](q *Queue, name string, fn func(ctx context.Context, v T) error, opts ...Option) *Kind[T]

Handle gives q the handler for the jobs of kind name, which runs fn with the value each was pushed with, and returns the Kind to push them with. The value is kept as JSON, so T is a type encoding/json turns into JSON and back: IDs, say, rather than the user they name, who may have changed by the time the job runs. The name is what the Store keeps to find the handler by, so it stays the same from one version of the app to the next: jobs one pushes, the next may run.

Handle panics when q has a handler for name already, or once q runs, or when q's Store isn't what the kind's options need: a UniqueStore for a Unique kind, and the extra its UniqueOptions need, an AtOnceStore for AtOnce, and a HoldBackStore for Rate. It panics for an OnFail that doesn't take T.

func (*Kind[T]) In added in v0.10.0 

func (k *Kind[T]) In(s Store) *Kind[T]

In is the kind, pushing its jobs to s rather than to the queue's Store: one the app makes from a transaction of its own, as the auth starter's jobs.in does, so that a job is kept with what else the transaction writes, as it commits, or not at all, as it rolls back. The queue's workers can't see such a job until the commit, so its Push doesn't wake them: the queue's Wake, after the commit, does, or else the next Poll finds the job.

In panics when s isn't what the kind needs of a Store, as Handle does of the queue's: a UniqueStore for a Unique kind, say.

func (*Kind[T]) Push 

func (k *Kind[T]) Push(ctx context.Context, v T) error

Push puts a job on the queue, to run as soon as a worker is free.

func (*Kind[T]) PushAt 

func (k *Kind[T]) PushAt(ctx context.Context, at time.Time, v T) error

PushAt puts a job on the queue, to run at at, or as soon after as a worker is free. For a Unique kind, it does nothing while a job with the same value waits, which keeps its own time, or with Latest, gives that job its own time.

func (*Kind[T]) Schedule added in v0.4.0 

func (k *Kind[T]) Schedule(s Schedule, v T)

Schedule runs a job of this kind on its own, with v, at each time s names: queue.Cron("0 2 * * *") at 2:00 each day, say. Each run is pushed ahead, as a job due at its time, once the run before it has come round, so a run that comes due while the app is down runs as it starts again: once, however many runs it missed. A new schedule's first run is its next time, not one that has passed. Every instance of the app that runs the queue pushes the runs, and the Store sees that each is pushed once: Schedule needs a ScheduleStore.

Schedule panics when q's Store isn't a ScheduleStore, when the kind has a schedule already, or once q runs.

type LatestStore added in v0.10.0 

type LatestStore interface {
	UniqueStore

	// PushLatest pushes j, which has a Key, or, when a job of its Kind and
	// Key waits, gives that job j's RunAt in place of its own, and sets
	// j.ID to the job that waits, either way. Of several pushes of one key
	// at once, one job waits, with one of their times.
	PushLatest(ctx context.Context, j *Job) error
}

LatestStore is a UniqueStore that also moves a job that waits to the time of a push of its key, for Latest.

type Limiter added in v0.24.0 

type Limiter interface {
	Try(ctx context.Context, key string) (wait time.Duration, err error)
}

A Limiter says whether one more may go now, counting it, or how long to wait: auth.Throttle's Try is one, which counts in its Store, so every instance of the app counts together.

type OneAtATimeStore added in v0.10.0 

type OneAtATimeStore interface {
	UniqueStore

	// KeepsOneAtATime does nothing. It says that the Store keeps the
	// promise, which is in its pushes and its claims, for Handle to know.
	KeepsOneAtATime()
}

OneAtATimeStore is a UniqueStore that also keeps a unique kind's jobs of one key from running at once, for OneAtATime. Its pushes keep a job's OneAtATime, and with it its Key, for as long as it keeps the job, where the Key a claim lets go goes: Claim doesn't return such a job while a claim holds another of its Kind and key that has it too.

type Option 

type Option func(*handler)

Option changes how a kind of job runs, for Handle.

func AtOnce added in v0.24.0 

func AtOnce(n int) Option

AtOnce keeps the kind's jobs from running more than n at once, on all the instances of the app together: a hundred photos to resize take n workers, and leave the rest to the other kinds. A job waits while n of its kind are held, and runs once one is done, has failed or gone back, or has lost its worker and its hold has run out. AtOnce needs a Store that counts them, an AtOnceStore.

func Attempts 

func Attempts(n int) Option

Attempts is how many times a job may run before it fails for good, the first time included. Default 10: with the default Backoff, the last runs about four hours after the first.

func Backoff 

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

Backoff is how long a job waits to run again after its attempt-th attempt failed, the first being 1. Default attempt⁴ seconds, with a tenth more at most at random: 1s, 16s, 81s, and on, a day at most.

func OnFail added in v0.24.0 

func OnFail[T any](fn func(ctx context.Context, v T, err error) error) Option

OnFail has fn run when a job of the kind fails for good, after its last attempt or a Permanent error, with the value it was pushed with and the error its last attempt returned: to mark what it was about, as a post whose photo never came, or to tell someone. It runs once, not after each attempt, once the Store has kept the job as failed, on the instance that ran it, with a context done after the kind's Timeout. Its error, or its panic, goes to the log. fn takes the kind's value: Handle panics for another type.

func Rate added in v0.24.0 

func Rate(l Limiter) Option

Rate keeps the kind's jobs from starting faster than l lets them: a newsletter's mails go no faster than the mail provider takes them, with an auth.Throttle of so many a second. Each job, as it's claimed, tries l by its kind's name, and one l refuses goes back as it was, its attempts unchanged, as it hasn't run, while no job of the kind is claimed until l's wait is over. A limiter that fails, as when its store is down, holds the job back too, for a while, rather than let it past the limit. Rate needs a Store that holds a kind back, a HoldBackStore.

func Timeout 

func Timeout(d time.Duration) Option

Timeout is how long a job may run: its context is done after it, so a handler that heeds its context fails the attempt. Default a minute.

func Unique added in v0.7.0 

func Unique(opts ...UniqueOption) Option

Unique makes the kind's jobs unique by their value: while a job of the kind waits, pushing another with the same value does nothing, and the one that waits runs, at its own time. Once a worker has it, a push is pushed, and runs too, maybe beside it on another worker, unless OneAtATime says otherwise: the running job may have read what the push is about. What varies from one push of the same work to the next, such as when it was asked for, stays out of the value. Unique needs a Store that keeps the keys, a UniqueStore, and its options their own extras.

type Outcome added in v0.29.0 

type Outcome int

Outcome is what the queue made of a run of a job. 

const (
	// Done is a run that worked: the job is done.
	Done Outcome = iota + 1

	// Retried is a run that failed, or was stopped with the queue: the
	// job is put back, to run again, after its backoff.
	Retried

	// Failed is a run that failed for good, at its last attempt or by a
	// Permanent error: the job is kept as failed, and runs no more.
	Failed

	// HeldBack is a run its kind's Rate held back before it started, or
	// the queue as it stopped, which isn't an attempt: the job waits to
	// run as it would have.
	HeldBack
)

func (Outcome) String added in v0.29.0 

func (o Outcome) String() string

String is the outcome as a metric's label has it: "done", "retried", "failed" or "held back".

type Queue 

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

Queue pushes jobs to its Store, and runs them.

func New 

func New(cfg Config) *Queue

New returns a Queue on cfg.Store. It panics without one.

func (*Queue) Drain 

func (q *Queue) Drain(ctx context.Context) error

Drain runs the jobs that are due, one at a time in the calling goroutine, until none is, and returns the first error from the Store, or ctx's once it's done. A job that fails is put back or kept as failed, as Run does, and runs again in the same Drain if it's due again before the others are done. It's for tests, with a Store such as queuetest.Memory, and for programs that run what's due and exit.

func (*Queue) Run 

func (q *Queue) Run(ctx context.Context) error

Run runs jobs as they come due, Workers at a time, until ctx is done. Then it takes no more, gives the ones running the Grace to finish, and returns nil once they have, or have been put back. An error from the Store is logged, and Run goes on after a Poll: a database that's down for a moment shouldn't stop the jobs for good.

Run is meant for tug's App.Go, which starts it with the app and stops it as the app shuts down.

func (*Queue) Wake added in v0.10.0 

func (q *Queue) Wake()

Wake has Run look for jobs that are due at once, rather than at its next Poll, as a push through the queue does: for the jobs a transaction of the app's pushed with Kind.In, once it has committed, as the workers couldn't see them as they were pushed.

type Ran added in v0.29.0 

type Ran struct {
	// Kind is the job's kind, as given to Handle.
	Kind string

	// Attempt is the run's attempt, 1 for the first. A run held back takes
	// none, and is the attempt it would have been.
	Attempt int

	// Outcome is what the queue made of the run.
	Outcome Outcome

	// Err is what the run failed with, for one Retried or Failed.
	Err error

	// Took is how long the job's handler ran: 0 for a run held back,
	// which didn't run it.
	Took time.Duration

	// Waited is how long the job waited past its time, from when it was
	// due to when it was claimed: how far behind its work the queue is.
	Waited time.Duration
}

Ran is how a run of a job went, which Config.Observe is told: for the app's metrics, which a client of its monitor's counts.

type Schedule added in v0.4.0 

type Schedule interface {
	Next(after time.Time) time.Time
}

A Schedule says when a kind of job runs on its own, for Kind.Schedule: Next is its first time after a time, or the zero time when there's none. Every instance of the app works the times out for itself, so Next has to give them all the same answer.

func Cron added in v0.4.0 

func Cron(expr string) Schedule

Cron is a Schedule written as a cron expression, in UTC; CronIn reads one on another zone's clock. It has five fields, the minute, the hour, the day of the month, the month and the day of the week, each a number, a range such as 1-5, a list such as 1,15, or * for every one, and any of those with a step, as */15 for every fifteenth. Months and days can be names, as JAN or MON, and Sunday is 0 or 7. A day restricted in both day fields matches either, as cron has it. @yearly, @monthly, @weekly, @daily and @hourly stand for their expressions. 

queue.Cron("0 2 * * *")       // at 2:00 each day
queue.Cron("*/15 * * * *")    // every 15 minutes
queue.Cron("0 9 * * MON-FRI") // at 9:00 each weekday

Cron panics on an expression that isn't one, or that never comes round, as the 30th of February doesn't: schedules are set as the app starts, which is the time to find out.

func CronIn added in v0.7.0 

func CronIn(zone, expr string) Schedule

CronIn is Cron on the clock of a time zone: zone is its name in the IANA database, as "Europe/Paris", or "Local" for the server's own. Where the clock goes forward or back, as for daylight saving, it does as cron does. A fixed time, one with no * in its minute or hour, runs once whatever the clock does: as the clock goes forward over it, and the first time the clock shows it when the clock goes back. A time with * follows the clock, so every 15 minutes is every 15 minutes, and @hourly every hour, however the clock goes. 

queue.CronIn("Europe/Paris", "30 2 * * *") // 2:30 in Paris, or 3:00 the night 2:00 becomes 3:00

Every instance of the app works the times out for itself, so they need the same zone database: a binary that imports time/tzdata carries its own. CronIn panics on a zone it can't load, as on an expression that isn't one.

func Every added in v0.4.0 

func Every(d time.Duration) Schedule

Every is a Schedule for each multiple of d, counted as time.Truncate counts them, from midnight UTC: Every(time.Hour) runs on the hour, Every(15*time.Minute) at :00, :15, :30 and :45, and Every(24*time.Hour) at midnight UTC. A time on a zone's clock, such as midnight in Paris, is CronIn's. It panics for a d of 0 or less.

type ScheduleStore added in v0.4.0 

type ScheduleStore interface {
	Store

	// PushScheduled pushes j, the run of the schedule named schedule that's
	// due at j.RunAt, unless a run of it due then or later has been pushed
	// already, and says whether it did. It keeps the run and pushes the job
	// together or not at all: of several instances pushing the same run at
	// once, one does. In a Store that's a UniqueStore too, the run of a
	// unique kind, whose job has a Key, is kept without a job of its own
	// while a job of its Kind and Key waits, which runs for it.
	PushScheduled(ctx context.Context, schedule string, j *Job) (bool, error)
}

ScheduleStore is a Store that also keeps which runs of a schedule have been pushed, for Kind.Schedule. Each instance of the app pushes each run of a schedule, as none knows whether the others have, so one push must win.

type Store 

type Store interface {
	// Push keeps a new job, due at its RunAt, and sets its ID.
	Push(ctx context.Context, j *Job) error

	// Claim returns the job due soonest at now that no claim holds,
	// holding it until until and adding one to its Attempts, or nil when
	// there's none. Two claims never return the same job while the first
	// one's hold lasts.
	Claim(ctx context.Context, now, until time.Time) (*Job, error)

	// Done removes a job that ran.
	Done(ctx context.Context, j *Job) error

	// Retry puts a job back to run again, due at its RunAt, with its Error.
	Retry(ctx context.Context, j *Job) error

	// Fail keeps a job that won't run again, with its Error, and its
	// FailedAt or the time by the Store's own clock, where no claim finds
	// it.
	Fail(ctx context.Context, j *Job) error
}

Store keeps a Queue's jobs, from Push until they're done, or kept as failed: a table in the app's database, say. Its methods are called from several goroutines at once, and with a database from several instances of the app. Package queuetest's TestStore checks that a Store keeps these promises.

A claim holds a job for a while, and a job whose hold runs out is taken to have lost its worker, as when the app was killed, and can be claimed again. So Done, Retry and Fail change a job only while the claim that returned it still has it: a late one, after a later claim took the job, leaves it alone, and returns nil. The job's Attempts tell the claims apart.

type UniqueOption added in v0.10.0 

type UniqueOption func(*handler)

UniqueOption changes how a Unique kind's jobs of one value go, for Unique.

func Latest added in v0.10.0 

func Latest() UniqueOption

Latest has the latest push of a value say when its job runs: a push while the job waits gives it the push's time, earlier or later, where without Latest the first push keeps its own. A reindex pushed for a minute on at each edit runs a minute after the last. A scheduled run doesn't move a job that waits, as a schedule's time isn't a push's. Latest needs a Store that moves the jobs that wait, a LatestStore.

func OneAtATime added in v0.10.0 

func OneAtATime() UniqueOption

OneAtATime keeps the kind's jobs of one value from running at once: one isn't claimed while another of the value runs, and runs once that one is done, or has failed, or has lost its worker and its hold has run out. A push while one runs is still pushed, as for any Unique kind, and waits for it. OneAtATime needs a Store that keeps the jobs' keys while they run, a OneAtATimeStore.

type UniqueStore added in v0.7.0 

type UniqueStore interface {
	Store

	// PushUnique pushes j, which has a Key, unless a job of its Kind and
	// Key waits, and says whether it did: of several pushes of one key at
	// once, one does. A job keeps its key until a claim has it: Claim lets
	// it go, so that a job pushed while the first runs is pushed, and Retry
	// doesn't take it back.
	PushUnique(ctx context.Context, j *Job) (bool, error)
}

UniqueStore is a Store that also keeps the keys of a unique kind's jobs, for Unique: while a job with a key waits, no other of its kind and key is pushed.

Source Files

View all Source files

Directories

Path Synopsis
Package queuetest is for testing with package queue: Memory is a Store for tests, and TestStore checks that a Store keeps the promises the queue depends on.
 Package queuetest is for testing with package queue: Memory is a Store for tests, and TestStore checks that a Store keeps the promises the queue depends on.

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.