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 SQLite. 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 ¶
Types ¶
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
}
Config is how a Queue runs its jobs. Every field but Store has a default.
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
}
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 for a Unique kind when q's Store isn't a UniqueStore.
func (*Kind[T]) PushAt ¶
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.
func (*Kind[T]) Schedule ¶ added in v0.4.0
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 Option ¶
type Option func(*handler)
Option changes how a kind of job runs, for Handle.
func Attempts ¶
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 ¶
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 Timeout ¶
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() 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: 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.
type Queue ¶
type Queue struct {
// contains filtered or unexported fields
}
Queue pushes jobs to its Store, and runs them.
func (*Queue) Drain ¶
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. 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 ¶
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.
type Schedule ¶ added in v0.4.0
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
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
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
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, 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 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
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. |