class Divider::Worker
Runs jobs: for each of its queues, a dispatcher task fetches the jobs
that are due (when enqueue notifies it, and every poll for the
scheduled ones and lost notifications) and runs each in a task of its
own, up to the queue's concurrency; a maintenance task beats for this
node's running jobs, gives back those of nodes that stopped beating, and
prunes old rows.
worker = Divider::Worker.new(ENV["DATABASE_URL"], jobs: [Cleanup, PurgeSessions],
queues: {default: 10, mail: 5})
Signal.trap(:TERM) { worker.shutdown(25.seconds) }
worker.run # until shutdown; or Task.spawn { worker.run } beside an HTTP server
The worker opens its own pool of connections to url (by default the
queues' concurrency plus 2) and one connection that listens for
enqueue's notifications.
A worker fetches only the classes in jobs, and leaves the others in
the table, never touching them: every node running a queue should know
all of that queue's classes. Queues are how work is split between
nodes of different kinds (a mail node, a report node). A row whose
class no deploy knows any more stays available until Divider.cancel
removes it.
An attempt that returns ends the job completed. One that raises
Divider::Discard discards it; Divider::Snooze puts it off without
counting the attempt; Divider::Cancelled, or an exception after
Divider.cancel asked it to stop, cancels it; any other exception (a
timeout too) fails the attempt, recorded in the job's errors, and
the job runs again after attempt ** 4 + 15 seconds plus a random 0 to
30 (as Oban), until its max_attempts, after which it is discarded.
/home/runner/work/divider/divider/src/worker.qz:33
Constants
RECORD_LIMIT: Duration
The longest a row's update after an attempt, or a fetch, may take.
CRON_CHECK: Duration
How often the cron entries are checked.
LISTEN_WAIT: Duration
The longest the listener waits for a notification at once.
LOCK_KEY: Int
The key of the advisory lock one node's maintenance holds.
FETCH: String
GUARD: String
(Each of these: $1 the id, $2 the node, $3 the attempt.)
FINISH_COMPLETED: String
FINISH_DELETE: String
FINISH_RELEASED: String
FINISH_SNOOZED: String
(A cancellation asked meanwhile wins over a snooze, a failure or a
discard: the row ends cancelled.)
FINISH_DISCARDED: String
FINISH_CANCELLED: String
FINISH_FAILED: String
RESCUE: String
Running jobs whose node stopped beating: back to available (the
attempt counted), discarded when that was the last, cancelled when
that was asked.
PRUNE: String
Class methods
def self.new(url: String, *, jobs: Array[Type[Divider::Job]], queues: Hash[Symbol, Int] = …, pool_size: Int? = …, poll: Duration = …, node: String? = …, heartbeat: Duration = …, stale_after: Duration = …, keep_completed: Duration = …, keep_discarded: Duration = …, on_error: shareable (Exception, Divider::Context) -> Void? = …, on_event: shareable (Divider::Event) -> Void? = …, cron: Divider::Cron? = …) -> Divider::Worker
A worker of jobs (the classes it may run) over the database url,
running queues (each queue's name, and how many of its jobs run at
once), and the recurring jobs of cron. node names this process in
the rows it runs (by default
HOSTNAME-pid- and a random suffix): it must differ from every other
running worker's.
Methods
def node -> String
This process's name in the rows it runs.
def run -> Void
Runs the queues until shutdown, then closes the worker's
connections. A worker runs once.
def shutdown(grace: Duration? = …) -> Void
Stops fetching, lets the running jobs finish for grace (nil: as long
as they take), then cancels them: their rows go back to available,
the attempt not counted (or to cancelled, when that was asked). A
second call cancels them at once. run returns then.