Documentation
¶
Overview ¶
Package jobs is the queue contract: work that happens after the response.
The contract lives in the core and the drivers do not, for the same reason as data.Repository: Push takes a security.Grant, and the tenant comes from it. Moving that into an optional package would make the guarantee optional, and an optional guarantee is not one.
A driver is a separate module under github.com/arandu-io/queue, because in Go there is no optional dependency and a core that carried a Redis client would put it in every project's go.sum.
Delivery is at-least-once. A handler that cannot run twice safely is a handler with a bug -- the process can die between doing the work and acknowledging it, and no queue anywhere solves that.
Index ¶
Constants ¶
const DefaultQueue = "default"
DefaultQueue is where a job goes when nobody said otherwise.
Variables ¶
var ErrForged = errors.New("jobs: the job does not match the Grant pushing it")
ErrForged is returned when a job claims an action or a tenant the Grant pushing it does not carry.
var ErrNoName = errors.New("jobs: a job with no name cannot be routed to a handler")
ErrNoName is returned when a job has no name to route by.
var ErrNoTenant = errors.New("jobs: the Grant carries no tenant, and a job without one cannot be scoped")
ErrNoTenant is returned when a Grant carries no tenant.
It is an error rather than a default, and that is RULE 14 with teeth: a job with no tenant cannot be scoped, and everything the handler touches would read across customers.
Functions ¶
func Authorized ¶ added in v0.10.0
Authorized reports whether a job may be pushed under this Grant.
Every driver calls it at the top of Push, and it closes an escalation the contract otherwise allows. New builds a job from the Grant, so what it produces always matches -- but Push takes a Job, and a Job is a struct anybody can fill in:
j := jobs.Job{ID: id, Name: "invoice.send", Action: "invoice.delete", TenantID: other}
queue.Push(ctx, viewGrant, j)
The worker rebuilds the Grant from the row -- GrantFor gives SystemGrant(j.Action, j.TenantID) -- so the handler would run with an action nobody authorized, in a tenant nobody authorized, and every Policy downstream would say yes because the Grant looks legitimate. The queue would be the one way past the authorization the whole framework exists to enforce. Found by audit.
Checked here rather than in each driver, because a driver that forgets is a driver that reopens it.
func ExponentialBackoff ¶
ExponentialBackoff doubles the wait each attempt, capped at an hour.
Capped, because unbounded doubling means the eleventh attempt is next year -- and a job nobody will ever see fail is worse than one that parks.
Types ¶
type Handler ¶
Handler does the work.
The Grant is rebuilt from the job's tenant and action, so a handler reaches repositories the same way a service does. There is no unauthorized path into the database from a worker, which is the whole point of the Grant existing.
type HandlerFunc ¶
HandlerFunc adapts a function to Handler.
type Job ¶
type Job struct {
// ID is the deduplication key. It is stable across retries, which is what
// makes a handler able to recognize work it already did.
ID string
// Queue separates work by urgency: a password reset email and a monthly
// report should not wait behind each other.
Queue string
// Name routes the job to its handler: "invoice.send", "report.monthly".
Name string
// TenantID is who the work belongs to. It comes from the Grant at Push, and
// the worker rebuilds a Grant from it -- a job with no tenant cannot be
// scoped, and everything downstream of it reads across customers.
TenantID string
// Payload is the arguments, as JSON. Keep it to facts and ids: a payload
// that says "look it up" is a payload that reads a row which has already
// changed.
Payload []byte
// AuthorizedBy and Action record the Grant that pushed it, which is the
// audit trail and what the worker reissues the work under.
AuthorizedBy string
Action string
// RunAt is when it becomes eligible. Zero means now.
RunAt time.Time
// Attempts counts the deliveries INCLUDING the current one: a job being
// handled for the first time has Attempts == 1. LastError is why the most
// recent one failed -- stored rather than logged, because the thing anyone
// needs at 3am is "this failed twelve times with this message".
Attempts int
LastError string
}
Job is one unit of work waiting to run.
type Queue ¶
type Queue interface {
// Push adds a job. The tenant comes from the Grant.
Push(ctx context.Context, g security.Grant, j Job) error
// Reserve takes up to n jobs off a queue and hides them for the lease.
// Jobs whose lease expires become visible again -- which is what makes a
// worker crash recoverable and delivery at-least-once.
//
// The returned jobs carry Attempts INCLUDING this delivery: a job handed
// over for the first time has Attempts == 1. A driver that returns the
// count from before the delivery makes the worker park a job one attempt
// early, and with MaxAttempts of 2 it parks on the first failure and never
// retries at all.
Reserve(ctx context.Context, queue string, n int, lease time.Duration) ([]Job, error)
// Ack removes a finished job.
Ack(ctx context.Context, j Job) error
// Fail records a failure and schedules the retry, or parks the job when it
// has had enough attempts.
Fail(ctx context.Context, j Job, cause error, retryAt time.Time, park bool) error
// Parked lists the jobs that gave up, so they can be inspected and retried.
Parked(ctx context.Context, limit int) ([]Job, error)
// Retry puts a parked job back in line with its attempts reset.
Retry(ctx context.Context, id string) error
// Pending is how many jobs are waiting on a queue. It feeds the health
// check: a queue that only grows is a worker that is not running.
Pending(ctx context.Context, queue string) (int, error)
// Oldest is how long the oldest waiting job has been waiting. A stopped
// worker looks exactly like an idle one, and this is what tells them apart.
Oldest(ctx context.Context, queue string) (time.Duration, error)
}
Queue is what a driver implements.
Reserve/Ack/Fail rather than a channel of jobs: the job has to stay in the store, invisible to other workers, until it is acknowledged. A worker that dies mid-job must not lose it, and that is not something a channel can offer.
type Worker ¶
type Worker struct {
// contains filtered or unexported fields
}
Worker runs jobs off a queue.
In the same binary as the application, started by `aru work`, which is the same image with a different argument. Not a second artifact: one image is what keeps the deploy story in doc 17 true.
func (*Worker) Handle ¶
Handle registers the handler for a job name.
Registering twice panics rather than replacing. Two handlers for one name is an import nobody meant to add, and finding out at boot beats finding out from work that silently went to the wrong place.
func (*Worker) HandleFunc ¶
func (w *Worker) HandleFunc(name string, f HandlerFunc) *Worker
HandleFunc registers a function.
type WorkerOptions ¶
type WorkerOptions struct {
// Queue is which queue to drain. Empty means DefaultQueue.
Queue string
// Concurrency is how many jobs run at once. Default 4.
Concurrency int
// Lease is how long a reserved job stays invisible to other workers. It has
// to exceed the longest handler, or a second worker picks up work still in
// progress. Default 5 minutes.
Lease time.Duration
// Poll is how long to wait before asking again when the queue was empty.
// Default 1 second.
Poll time.Duration
// MaxAttempts is how many failures a job gets before it is parked.
// Default 5.
MaxAttempts int
// Recorder receives each finished job, so it shows on /_arandu/debug with
// its queries and its timeline -- exactly like a request.
//
// Nil means no instrumentation, and that is what production looks like: no
// Collector is built, FromContext returns nil, and every Record method is a
// no-op on a nil receiver. Zero cost, not low cost.
//
// It used to build a Collector on every job unconditionally and then throw
// it away -- so production paid for recording every query with its bound
// arguments and its caller frames, and nobody could read any of it. Found by
// audit. Pass kernel.Recorder() to turn it on.
Recorder *observability.Recorder
// Backoff returns how long to wait before attempt n. Default is
// exponential, capped at an hour.
Backoff func(attempt int) time.Duration
}
WorkerOptions configures the loop.
Source Files