Snails

Module: ActiveJob

Defined in:
runtime/active_job/base.rb,
runtime/active_job/base.rbs,
runtime/active_job/database_queue.rbs,
runtime/active_job/launcher.rbs,
runtime/active_job/process_queue.rbs,
runtime/active_job/scheduler.rbs,
runtime/active_job/worker.rbs,
runtime/active_job/database_queue.rb,
runtime/active_job/launcher.rb,
runtime/active_job/process_queue.rb,
runtime/active_job/scheduler.rb,
runtime/active_job/worker.rb

Overview

The job stack's drain loop: poll the queue adapter, run one drain, and repeat. ActiveJob::Launcher.start spawns one of these per worker thread (prefork children included); the periodic half lives in active_job/scheduler.

A recurring beat only enqueues the job (ActiveJob.recurring); this loop executes it like any other queued job.

Defined Under Namespace

Modules: Launcher, Scheduler, Worker Classes: Base, DatabaseQueue, ProcessQueue

Constant Summary collapse

EXECUTORS =

The executor registry: one entry per registration — the array's length is the whole "an executor exists in this process" flag executor_registered reads.

[]
QUEUE_ADAPTER =

The adapter slot: the selected adapter class, empty for the default (ProcessQueue). The app selects one at boot, before the server starts, with queue_adapter= — any class carrying the adapter interface (the file header).

[]
RECURRING =

The recurring registrations, one Hash per entry (the PERIODIC shape) — recurring appends here and the wired beat carries the entry, so the registry is what boot-time introspection and the contract tests read.

[]

Class Method Summary collapse

Class Method Details

.enqueue(job_class, payload = nil) ⇒ void

the hand-off entry: the job class is the argument (its job_name keys the entry) and the work derives from the class — one call shape, perform(payload), so a payload-less job still declares the parameter and ignores it.

The proc is what keeps the process-local queue expressible on a target with one element type per container — every entry is () -> void whatever it closed over, so jobs of any arity share one Array. The selected adapter takes it from here: ProcessQueue keeps the proc; a store-backed adapter stores the name and payload and ignores the proc; the drain resolves them itself (Object.const_get). With an executor registered the work is queued and this returns immediately; with none it runs at the call site. A bounded adapter may refuse the hand-off when its queue is full (the process queue drops the new job and reports locally) — a request path must not assume every hand-off became queued work.

Examples:

class NotifyJob < ActiveJob::Base
  def self.perform(data)
    id = data["id"]
    ...
  end
end

ActiveJob.enqueue(NotifyJob, { "id" => 42 })

Parameters:

  • job_class (Class) —

    the job class — job_name + perform(data)

  • payload (Object, nil) (defaults to: nil) —

    the serializable arguments for a serializing adapter — a JSON value, or nil for a payload-less job



63
 
# File 'runtime/active_job/base.rb', line 63

def self.enqueue: (Class job_class, ?untyped? payload) -> void

.executor_registered ⇒ Boolean

whether an executor exists in this process: enqueue queues only while this answers true and runs the job at the call site otherwise, because a job nobody drains would be silently lost.

Returns:

  • (Boolean) —

    true when an executor is registered



140
 
# File 'runtime/active_job/base.rb', line 140

def self.executor_registered: () -> bool

.queue_adapter ⇒ Class

the adapter class every hand-off and drain goes through: the selected one, or ProcessQueue when none was selected.

Returns:

  • (Class) —

    the selected adapter class



109
 
# File 'runtime/active_job/base.rb', line 109

def self.queue_adapter: () -> Class

.queue_adapter=(adapter) ⇒ void

select a queue adapter. Any class is taken: the interface is the contract (see the file header), and the setter's job is the swap, not a registry — an allowlist would turn every new adapter into a core change. A typo'd constant fails at compile/load; a class missing the interface fails loudly at its first dispatch.

Parameters:

  • adapter (Class) —

    an adapter class — ProcessQueue or an app's own



125
 
# File 'runtime/active_job/base.rb', line 125

def self.queue_adapter=: (Class adapter) -> void

.recurring(job_class, interval) ⇒ void

one call declares a job that runs once per interval seconds. The class carries the job contract (Base) — job_name (the key) and a class-side perform — and the registration wires the beat (the scheduler's periodic registry; each fire enqueues once per interval — the interval's start (scheduled_at_for) goes to enqueue_once, where a second enqueue for the same (name, scheduled_at) must be a no-op, so several server processes firing the same beat still run the job once). There is no handler catalog: a cross-process drain resolves the stored name itself — Object.const_get — so the registration's class is what must be compiled in. The beat's run proc takes no argument (a process-local adapter runs it as the job). Call it after the adapter is selected.

Examples:

ActiveJob.recurring(CleanupJob, 300)

Parameters:

  • job_class (Class) —

    the job class — job_name + class-side perform

  • interval (Integer) —

    seconds between runs



175
 
# File 'runtime/active_job/base.rb', line 175

def self.recurring: (Class job_class, Integer interval) -> void

.register_executor ⇒ void

declare that an executor exists in this process; enqueue queues only after this. Snails::Main.start calls it once at boot, so the flip is exit-only: there is no unregister.



149
 
# File 'runtime/active_job/base.rb', line 149

def self.register_executor: () -> void

.schedule(job_class, at, payload = nil) ⇒ void

hand off a delayed one-off run: the same class-oriented hand-off as enqueue, but the work waits until at (Time) instead of running at the next drain. The adapter stores the run time (enqueue_at), and a drain runs the job once the clock passes it. With no executor registered this raises: a schedule needs one, and running it at the call site would ignore the delay — the same silent-loss rule enqueue follows, kept loud here.

at carries its own contract: the Time slot refuses a non-Time argument at the build or the call boundary (the ext layer reopens Time — ext/time.rb), so no in-body guard stands here.

Examples:

ActiveJob.schedule(NotifyJob, Time.now + 900, { "id" => 42 })

Parameters:

  • job_class (Class) —

    the job class — job_name + perform(data)

  • at (Time) —

    the run time

  • payload (Object, nil) (defaults to: nil) —

    the serializable arguments for a serializing adapter — a JSON value, or nil for a payload-less job

Raises:

  • (RuntimeError) —

    no executor is registered — a schedule needs one



91
 
# File 'runtime/active_job/base.rb', line 91

def self.schedule: (Class job_class, Time at, ?untyped? payload) -> void

.scheduled_at_for(interval, now = Time.now.utc) ⇒ Time

the identity a recurring fire enqueues with: the interval's start (now floored to the interval) as a Time, so every process firing inside the same interval computes the same value and the adapter's unique key drops the repeats.

Parameters:

  • interval (Integer) —

    seconds per interval

  • now (Time) (defaults to: Time.now.utc) —

    the clock to floor

Returns:

  • (Time) —

    the interval's start



192
 
# File 'runtime/active_job/base.rb', line 192

def self.scheduled_at_for: (Integer interval, ?untyped now) -> Time