class Divider::Job
A job: a class whose enqueue inserts a row into divider_jobs, which
a worker later runs. The class body sets its
options with statements that run at compile time:
class Cleanup < Divider::Job
queue :maintenance # default :default
max_attempts 5 # default 20
timeout 2.minutes # default none
unique period: 1.hour # at most one live Cleanup an hour
end
def schedule(repo: Rotor::Repo) -> Void
Cleanup.enqueue(repo) # now
Cleanup.enqueue(repo, after: 10.minutes) # in ten minutes, by the database's clock
Cleanup.enqueue(repo, at: Time.utc(2026, 12, 1), priority: 1)
end
enqueue runs in the repository's transaction when it is bound to
one (repo.transaction { |tx| ... Cleanup.enqueue(tx) }): the job
exists, and a worker hears of it, only once that commits.
/home/runner/work/divider/divider/src/job.qz:23
Class methods
def self.job_queue -> String
The queue (generated by queue).
def self.job_max_attempts -> Int
How many attempts before the job is discarded (generated by
max_attempts).
def self.job_timeout_ms -> Int?
The time an attempt may take, in milliseconds, or nil for no limit
(generated by timeout).
def self.job_unique -> Divider::Unique?
The uniqueness of the job's rows, or nil (generated by unique).
static def self.queue(name: Symbol) -> Void
The job's queue: a name of letters, digits and _ (at compile
time: an error at the statement otherwise).
static def self.max_attempts(n: Int) -> Void
How many attempts the job gets before it is discarded: 1 or more.
static def self.timeout(limit: Duration) -> Void
The time an attempt may take: longer, the attempt is stopped and counts as failed. At least a millisecond.
static def self.job_args[A](type: Type[A]) -> Void
The job's arguments: the fields of the data type type, declared
in the class (data Args(user_id: Int, template: String = "welcome"),
then job_args Args), JSON-encoded into the row. Generates
enqueue(repo:, <the fields>, at:, after:, priority:), keywords only: a field with a default is optional;enqueue_args(repo:, args:, at:, after:, priority:), with an args value.
A field's class is checked here: Int, Integer, Float,
String, Bool, Symbol, Time, Bytes, Array, Hash, Set,
another data type of such fields, or a nullable or union one
(which Type#fields shows as Object). Only the outer class is seen
(quarz-lang/quarz#401): a nullable or union field and the elements of
a collection are not checked here (Pen? or Array[Pen] passes, and
is found when the worker decodes the job); the keywords are typed by
those classes only, without type arguments or nullability, and a
collection's keyword is typed Object (a keyword's class carries no
type arguments), while enqueue_args checks its value fully against
the type. The worker decodes the stored arguments (JSON.decode) and
calls perform(args, job); arguments that do not decode discard the
job with Divider::ArgsError, which names where. A field cannot be
named repo, at, after or priority, and job_args comes once.
def self.job_args_type -> Type[Object]?
The class of the job's arguments, or nil for a job without (generated
by job_args).
static def self.unique(*, by: Array[Symbol] = …, period: Duration? = …) -> Void
At most one live job (scheduled, available, running, or failed with
a retry pending) of the class with the same values of the arguments
by within each period (from the epoch, by the database's clock),
or ever without one: enqueue gives the existing job's Enqueued
instead of inserting. Under REPEATABLE READ or SERIALIZABLE, an
enqueue whose key a transaction committed since the snapshot holds
raises PostgreSQL's serialization failure instead: retry the
transaction, as for any such conflict.
by names fields of the job's job_args that have no default (a field
left to its default is absent from the row's JSON), so unique
comes after job_args.
def self.enqueue(repo: Rotor::Repo, *, at: Time? = …, after: Duration? = …, priority: Int = …) -> Divider::Enqueued
Inserts the job, due now, at at, or after from the database's
now (not both), with priority (0, the first, to 9; 5 by default).
For a job without arguments: one with job_args takes them as keywords
(enqueue(repo: repo, user_id: 42)), and calling this form on it
(which enqueue(repo: repo) without its arguments also resolves to)
raises ArgumentError.
Methods
def perform(job: Divider::Context) -> Void
Runs the job: what a job without job_args defines, with what the worker
tells of the attempt (job.repo for its queries). Returning ends the
job completed; Divider::Discard discards it, Divider::Snooze
puts it off, any other exception fails the attempt.