README
¶
arandu-io/queue
The job queue over Arandu's own application database.
About
queue is the default job queue for Arandu: a table in the application's own
database. That is the point, not a limitation — a job pushed inside a
transaction is committed by that same transaction, so there is never a job for
a row that was rolled back, or a row without its job. The contract —Job,
Handler, Queue, the worker — lives in the framework; this package is one
implementation of it, and the one that needs nothing installed.
Moving into hesape
This repository is the previous address. Its content now lives inside
arandu-io/hesape, at hesape/queue, with the RESP driver under
hesape/queue/connectors/redis as its own module. See
ADR-0048 for the reasoning: hesape
collects the framework's components in one place, and a developer looking for
the queue driver finds it inside that collection instead of needing to know it
lives in a repository of its own.
This module is not being deleted. It stays published, and the Go module proxy
keeps serving every version already tagged. New work happens in hesape; this
address is for whatever already depends on it.
What it delivers
One dependency: modernc.org/sqlite. No driver, no client — the jobs table
sits in the database the application already has, which is what makes the
transactional guarantee below possible in the first place.
Push requires a security.Grant, checked against the job before it is
written, and the rest of the surface follows the queue as it actually runs:
Reserve, Ack, Fail, Parked, Retry, Pending, Oldest. Exponential
backoff and a dead-letter queue come from Fail and Parked; nothing about
retrying or parking a job needs SQL written by hand.
The central guarantee: a job is committed by the same transaction as the
row it refers to. This is what a driver over a separate store cannot offer —
the outbox guarantee applied to work instead of to events. For volume beyond
what a table handles comfortably, the RESP driver in hesape/queue/connectors/redis
is the same contract over a different store, one line different in wiring.
794 lines of production code, 850 of test.
Installation
go get github.com/arandu-io/hesape/queue
go get github.com/arandu-io/hesape/queue/connectors/redis # optional, for volume beyond a table
This module's own path, github.com/arandu-io/queue, still resolves for
anything already pinned to it.
Related repositories
arandu-io/hesape— where this adapter lives now.arandu-io/framework— the core this package implements:Job,Handler,Queue.arandu-io/database,arandu-io/kv,arandu-io/storage— the other adapters making the same move.
Learning Arandu
The API reference is generated from the doc comments and lives on pkg.go.dev. Every exported symbol carries one, and that is deliberate: it is the documentation that cannot drift from the code, because it sits in the same file.
The CLI documents itself. aru help lists every command, and each one explains
what it writes and what to do with it. aru doctor explains what it found and
what breaks, not which rule was violated.
A guide and a website do not exist yet, and that is a decision rather than a gap: a guide written against an API that still moves is work done twice, and the second time is worse — there is wrong documentation published. The site is the next phase, and it will be an Arandu application.
Contributing
See CONTRIBUTING.md. Before opening a pull request, the three commands at the top of that file have to pass, and CI runs exactly them.
Security Vulnerabilities
Please review our security policy on how to report a vulnerability. Never open a public issue for one.
License
Open-sourced software licensed under the MIT license.
Documentation
¶
Overview ¶
Package queue is the job queue over the application's database.
The contract lives in the core, in framework/jobs: Job, Handler, Queue and the Worker. This package is one implementation of it, and it is the default one for a reason -- it needs nothing installed. The jobs table sits in the database the application already has, and a job is committed by the same transaction as the row it is about.
That last part is what a Redis queue cannot offer. Pushing a job inside data.Transaction means the job exists if and only if the write did, which is the outbox guarantee applied to work instead of to events.
For volume beyond what a table handles comfortably, github.com/arandu-io/queue/kv is the same contract over RESP. Same Worker, same handlers, one line different in main.
Index ¶
- type CreateJobsTable
- type Module
- type Store
- func (s *Store) Ack(ctx context.Context, j jobs.Job) error
- func (s *Store) Fail(ctx context.Context, j jobs.Job, cause error, retryAt time.Time, park bool) error
- func (s *Store) Oldest(ctx context.Context, queue string) (time.Duration, error)
- func (s *Store) Parked(ctx context.Context, limit int) ([]jobs.Job, error)
- func (s *Store) Pending(ctx context.Context, queue string) (int, error)
- func (s *Store) Push(ctx context.Context, g security.Grant, j jobs.Job) error
- func (s *Store) Reserve(ctx context.Context, queue string, n int, lease time.Duration) ([]jobs.Job, error)
- func (s *Store) Retry(ctx context.Context, id string) error
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type CreateJobsTable ¶ added in v0.4.0
type CreateJobsTable struct{ migrations.BaseMigration }
CreateJobsTable is the jobs table this driver reads and writes.
func (CreateJobsTable) Down ¶ added in v0.4.0
func (CreateJobsTable) Down(ctx context.Context, conn migrations.Connection) error
Down drops the table, which takes its indexes with it.
func (CreateJobsTable) GetName ¶ added in v0.4.0
func (CreateJobsTable) GetName() string
GetName is the migration's identity, and carries its order.
func (CreateJobsTable) Up ¶ added in v0.4.0
func (CreateJobsTable) Up(ctx context.Context, conn migrations.Connection) error
Up creates the table and the two indexes the reserve and diagnosis queries need.
One statement per call: the connection sends what it is given, and a pair separated by a semicolon only reaches a server that was asked for multi-statement support.
type Module ¶
type Module struct {
// contains filtered or unexported fields
}
Module brings the jobs table, and reports on the queue.
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".
func NewModule ¶
NewModule returns the module.
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.
func (*Module) Diagnose ¶
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) Migrations ¶
func (*Module) Migrations() []migrations.Migration
Migrations returns the jobs table.
The module is asked rather than registering itself: an application on a different driver has no jobs table, and a package-level init would create one for every project that imports this one.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store is the queue backed by the application's database.
func (*Store) Ack ¶
Ack removes a finished job.
Deleted rather than marked done. A jobs table that keeps every job ever run is a table that needs its own cleanup job, and the history that matters -- what ran, how long it took, what it queried -- is on the console.
func (*Store) Fail ¶
func (s *Store) Fail(ctx context.Context, j jobs.Job, cause error, retryAt time.Time, park bool) error
Fail records the failure and schedules the retry, or parks the job.
func (*Store) Oldest ¶
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. It feeds the health check.
func (*Store) Push ¶
Push adds a job.
Inside data.Transaction it joins it, which is the property this driver exists for: the job is committed by the same transaction as the row it describes, so it cannot refer to a write that rolled back.
func (*Store) Reserve ¶
func (s *Store) Reserve(ctx context.Context, queue string, n int, lease time.Duration) ([]jobs.Job, error)
Reserve takes jobs off the queue and hides them for the lease.
Two statements rather than one, and the reason is portability: the tight form is UPDATE ... RETURNING with a FOR UPDATE SKIP LOCKED subquery, which Postgres has and SQLite does not. Selecting the candidates and then claiming each one by its id -- with reserved_until in the WHERE -- is correct on every engine, because the claim itself is the compare-and-set.
The cost is that two workers can pick the same candidate and one of them loses the claim. It gets nothing back, which is exactly right.
Source Files