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 that needs a client installed is a module of its own, because in Go there is no optional dependency: a contract that carried a Redis client would put it in every project's go.sum, its build and its vulnerability surface.
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.
This package is a bridge. It is removed in v1.0.0; import github.com/arandu-io/hesape/queue directly.
The components moved to github.com/arandu-io/hesape, under new names, and this package is now the old names pointing at them. It answers to two hesape packages:
hesape/queue the Worker, its options and the backoff hesape/queue/jobs the Job, its constructor, GrantFor and Authorized
The death date above is what keeps this from being a second way to import one type. Nothing here holds an implementation: the loop that drains a queue, the decision to park a job and the hash of a payload all run in hesape, and what is left here translates.
This bridge is more envelope than alias, and the reason is one difference: in hesape a Job settles itself -- Release, Delete, Fail -- where here the Queue settled it, and a job is a *jobs.Job there where it is a value here. Four names diverged that way:
Job UUID rather than ID, and a pointer with a driver behind it Queue eleven methods rather than eight, and all but one renamed Handler takes *jobs.Job rather than a Job Worker Daemon rather than Run, and WorkerOptions renamed two fields
None of the four could alias, and none of the four may change shape here. A driver implements Queue by these method names and marshals a Job by these field names, both in a module of its own, and a handler a project wrote against this package implements this Handler. A new job is not written here: the job generator emits a handler against hesape/queue/jobs directly, taking a *jobs.Job, so what these names hold up is code that already exists. An alias would compile in this module and break all of it in silence, which is the one failure a build of the framework cannot catch.
What crosses to hesape is the adapter in worker.go, and it is the only thing that has to know both shapes.
Index ¶
Constants ¶
const DefaultQueue = hjobs.DefaultQueue
DefaultQueue is where a job goes when nobody said otherwise.
Variables ¶
var ErrForged = hjobs.ErrForged
ErrForged is returned when a job claims an action or a tenant the Grant pushing it does not carry.
var ErrNoName = hjobs.ErrNoName
ErrNoName is returned when a job has no name to route by.
var ErrNoTenant = hjobs.ErrNoTenant
ErrNoTenant is returned when a Grant carries no tenant.
It is an error rather than a default: a job with no tenant cannot be scoped, and everything the handler touches would read across customers.
The alias is what keeps errors.Is correct across the two modules: a driver that checks this value and a hesape function that returns jobs.ErrNoTenant are looking at one error.
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.
The decision is hesape's, so the two drivers in this collection and any written against hesape/queue directly refuse the same job.
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.
A wrapper and not an alias: a plain function has no alias form in Go.
func GrantFor ¶
GrantFor rebuilds the Grant a job runs under.
The action and the tenant come from the row, so the worker reissues exactly what the push authorized -- not more. A worker that invented its own Grant would be a way to reach the database with permissions nobody granted.
It is one line: hesape mints the Grant, so there is no second definition of what a job is allowed to do.
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.
It stays declared here rather than aliasing hesape/queue.Handler, which takes a *jobs.Job so the handler can release or park its own job. A handler a project already wrote against this package has this signature, and an alias would ask all of them to be rewritten, which is the opposite of what a bridge is for. A new handler is not written against it: the job generator emits one against hesape/queue directly.
type HandlerFunc ¶
HandlerFunc adapts a function to Handler.
Declared here for the same reason Handler is: it names this package's Job, which is a value where hesape's is a pointer with a driver behind it. A function written for one is not a value of the other.
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.
//
// Renamed on the way to hesape: it is jobs.Job.UUID there.
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.
It stays declared here rather than aliasing hesape/queue/jobs.Job, which renamed ID to UUID, added three fields and put a driver behind the job so it can settle itself. Two things outside this module read this shape: a driver writes j.ID into a column or marshals this struct to JSON and back, and a handler a project already wrote against this package reads j.ID. An alias would rename a column and a wire field at once.
The three fields hesape added -- DisplayName, Exceptions and Attributes -- and the settlement bookkeeping behind them have no counterpart here, and nothing on this side ever sets them, so a job that crosses the bridge and comes back is the job that went in.
func New ¶
New builds a job from a Grant and a payload.
This is the only constructor, so every job in the system carries a tenant, an id and the Grant that authorized it -- there is no shape of Job that skipped any of the three.
A wrapper and not an alias: a plain function has no alias form in Go, and the job it answers with has to cross back over the rename.
type Module ¶ added in v0.33.0
type Module struct {
// contains filtered or unexported fields
}
Module brings the queue's schema, and reports on whether anything is draining it.
It registers no routes and runs no worker: the worker is `aru work`, a separate process from the same image. What this module owns is the schema and the answer to "is anything draining this". Register it in bootstrap/app.go next to the modules that push jobs.
Why this one is declared and not aliased ¶
One method stands in the way. hesape/queue.Module answers Routes with a *routing.Router and this one answers it with a *http.Router, so the hesape module does not satisfy the contract the kernel collects. Everything else is the inner module's, down to the lag thresholds and the sentences the error page prints, so there is one description of a stalled queue and not two that can drift apart.
func NewModule ¶ added in v0.33.0
NewModule returns the module for a queue driver.
Name the queues the application actually uses. A queue nobody watches is a queue that can stop draining without anyone noticing, which is the failure this module exists to make visible. Naming none watches the default queue.
The driver is passed through rather than wrapped, because the schema is discovered by asking it: a driver that keeps jobs in the application's database carries the table it reads, and one that keeps them elsewhere carries nothing. Anything standing between the two would have to answer that question for the driver, and would answer it wrong for every driver it did not know about.
func (*Module) Diagnose ¶ added in v0.33.0
Diagnose reports a backlog and a dead letter queue, on the error page.
Next to the failure somebody is already looking at, which is the moment they are most likely to act on it.
func (*Module) Health ¶ added in v0.33.0
Health fails when a queue stops draining.
A stopped worker looks exactly like an idle one, and the age of the oldest waiting job is what tells them apart. Without this, the first sign is a customer asking why they never got the email.
func (*Module) Migrations ¶ added in v0.33.0
Migrations returns the schema the wired driver needs.
A driver that stores jobs outside the application's database returns nothing here, rather than an empty table it will never read.
func (*Module) Name ¶ added in v0.33.0
Name is the module identifier.
The inner module answers, so the name a migration is collected under and the name a failing health check prints are one string and not two.
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.
It stays declared here, with the old eight method names, rather than aliasing hesape/queue.Queue -- which has eleven and renamed all but two: Reserve is Pop, Ack is delegated to the job itself, Fail is split across Release and Fail on the job, Parked is Failed, Pending is PendingSize and Oldest is CreationTimeOfOldestPendingJob, answering with a time rather than a duration.
A driver implements this interface, by these names, from a module this one does not compile: an alias here would compile in the framework and break every such driver in silence, which is the one failure this bridge exists to prevent. queueAdapter, in worker.go, is what carries an implementation of this across to hesape.
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.
It is an envelope over hesape/queue.Worker because the design diverged in three ways at once: the loop is called Daemon there and answers with an exit status a supervisor reads, a handler takes a *jobs.Job so it can settle its own job, and the queue underneath is asked eleven questions rather than eight. The three translations are here and nothing else is.
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.
func (*Worker) Run ¶
Run drains the queue until the context is cancelled.
Renamed on the way to hesape: it is Worker.Daemon there, and it answers with the exit status a process supervisor reads. This contract has no place to put a status -- a cancelled context is the only way it ever ended -- so the status is dropped and the error is passed through.
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.
//
// Renamed on the way to hesape: it is WorkerOptions.Sleep there.
Poll time.Duration
// MaxAttempts is how many failures a job gets before it is parked.
// Default 5.
//
// Renamed on the way to hesape: it is WorkerOptions.MaxTries there.
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.
//
// Building a Collector on every job unconditionally and then throwing it
// away would make production pay for recording every query with its bound
// arguments and its caller frames, with nobody able to read any of it. 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.
It stays declared here rather than aliasing hesape/queue.WorkerOptions, which renamed Poll to Sleep and MaxAttempts to MaxTries and added nine fields this contract never had. The defaults are not applied here any more: every zero means the same thing on the other side, and hesape decides it once.
Source Files