Module: ActiveJob::Scheduler

Defined in:: runtime/active_job/scheduler.rbs,
runtime/active_job/scheduler.rb

Constant Summary collapse

POLL_INTERVAL = How often the scheduler loop looks for due buckets, in seconds. A cooperative poll rather than a wakeup because the callback registry is a plain Array owned by this process.

0.05

PERIODIC = the periodic callbacks the scheduler runs, one entry per registration. Process-local like the job queue: every server process runs its own scheduler, so a callback that must not double-run across processes takes a lease of its own. A Hash per entry rather than an array triple: an element write (entry["last_bucket"] = …) is a shape the compiler handles; an index write (entry[1] = …) is not.

[]

Class Method Summary collapse

.register_periodic(seconds, work) ⇒ void
add a beat the scheduler runs once per seconds-second interval, keyed to the interval's CALENDAR bucket: the scheduler fires when the current bucket (epoch floored to the interval — the same grid ActiveJob.scheduled_at_for stamps a recurring scheduled_at with) differs from the last bucket this entry fired.
.run ⇒ void
the scheduler loop: poll, fire the due callbacks, and repeat.
.run_periodic(now = Time.now.to_i) ⇒ void
fire every callback whose calendar bucket has advanced since its last fire, and record that bucket as fired.

Class Method Details

.register_periodic(seconds, work) ⇒ `void`

add a beat the scheduler runs once per seconds-second interval, keyed to the interval's CALENDAR bucket: the scheduler fires when the current bucket (epoch floored to the interval — the same grid ActiveJob.scheduled_at_for stamps a recurring scheduled_at with) differs from the last bucket this entry fired. A fresh entry carries no bucket, so its first run fires the current one — a process that restarts more often than its interval still covers every bucket it was alive for, instead of waiting out a full interval from each boot (an uptime-anchored first fire starves retention-class beats under frequent deploys). Buckets that passed while no process was alive are not backfilled: only the current bucket is covered. An app declares periodic work through ActiveJob.recurring, which registers the beat here. The callback runs on the scheduler's thread with its own Sqlx::Database lease, so it may touch models. The proc argument (not a block) mirrors ActiveJob.enqueue: the RBS cannot type a captured &block. A non-positive interval is rejected here — the bucket arithmetic below would divide by zero, and registration is the point that can still name the bad value.

Parameters:

seconds (Integer) —
seconds per bucket
work (Proc) —
the zero-argument callback the scheduler invokes

Raises:

(ArgumentError) —
when the interval is not a positive Integer

43	# File 'runtime/active_job/scheduler.rb', line 43 def self.register_periodic: (Integer seconds, ^() -> void work) -> void

.run ⇒ `void`

the scheduler loop: poll, fire the due callbacks, and repeat. Launcher.start gives it its own thread; it never drains the queue. A beat only enqueues its job (ActiveJob.recurring); Worker.run executes the queued job. A failure in the run is reported by fire_due_callbacks and the next run still fires; a control or fatal exception stops the loop and is reported here.

91	# File 'runtime/active_job/scheduler.rb', line 91 def self.run: () -> void

.run_periodic(now = Time.now.to_i) ⇒ `void`

fire every callback whose calendar bucket has advanced since its last fire, and record that bucket as fired. Recording happens before the call, so a raising callback still counts as its bucket's attempt — the next fire is the next bucket, not a retry loop inside this one. The scheduler calls this each poll; a callback that raises becomes a scheduler.callback_failed record and does not stop the others (the job drain's rule) or the loop. Control and fatal exceptions propagate to the scheduler loop. now is injectable so the contract is testable without sleeping.