Snails

Class: ActiveJob::DatabaseQueue

Inherits:
Object
  • Object
show all
Defined in:
runtime/active_job/database_queue.rbs,
runtime/active_job/database_queue.rb

Constant Summary collapse

ORPHANED_AFTER_SECONDS =

A row running longer than this belongs to a dead process — the next drain returns it to pending (at-least-once: the row's attempts counter has already recorded the lost run).

600
BATCH_SIZE =

The most rows one drain runs — the rest waits for the next drain, so one slow queue never holds the worker loop.

10
DRAIN_INTERVAL_SECONDS =

The seconds between idle drains: a drain that fills its batch keeps the loop coming straight back on its 50ms poll, so this interval gates only the idle check — the database sees one drain per interval instead of one per poll while the queue is empty.

1
LAST_DRAIN_AT =

The clock of the last drain — epoch seconds, shared state by design: one adapter object serves the process's worker loop.

[ 0 ]
LAST_DRAIN_WAS_FULL =

True when the last drain ran BATCH_SIZE rows — a full drain means more due rows may be waiting, so the loop comes straight back instead of waiting out the clock. A shorter drain clears it, so an idle queue still checks once per second and no more.

[ false ]
DRAIN_FLAGS_LOCK =

Guards the two flags above. WORKER_CONCURRENCY runs several worker threads in one process, and they read/write these together; the lock covers the flags only, never a drain's job time.

Mutex.new

Class Method Summary collapse

Class Method Details

.drain ⇒ Integer

release orphaned claims, then run up to BATCH_SIZE due rows oldest first, and answer how many ran. An unresolvable name or a raising body is recorded on its row and the drain moves on — one bad job never costs the queue the rest. A worker thread calls this under its own database lease, and several worker threads may call it at once: the claim is a single atomic UPDATE, so the rows are split between them and never duplicated (a claim that lost the race answers nil and the drain ends). The drain flags are mutex-guarded; a full batch sets LAST_DRAIN_WAS_FULL so the next drain is due at once, a shorter drain — even an empty one — clears it and the idle interval takes over. With N worker threads, N rows run at once (one per worker thread) and a drain processes up to BATCH_SIZE rows per worker thread; a row abandoned by a dead process returns to pending after ORPHANED_AFTER_SECONDS.

Returns:

  • (Integer) —

    the number of rows this drain ran



138
 
# File 'runtime/active_job/database_queue.rb', line 138

def self.drain: () -> Integer

.drain_due? ⇒ Boolean

whether a drain is due: the interval passed since the last drain, OR the last drain ran a full batch. A full batch means more due rows may be waiting, so a worker thread (its 50ms poll) comes straight back and works through the backlog; a shorter drain clears the flag and the idle interval takes over. What runs inside a drain is decided by the rows, not by this gate. The read is mutex-guarded: concurrent worker threads share the flags.

Returns:

  • (Boolean) —

    true when a drain should run



117
 
# File 'runtime/active_job/database_queue.rb', line 117

def self.drain_due?: () -> bool

.enqueue(job_name, work, payload) ⇒ void

hand off a job due now: write one pending row and return. The body does not travel — the row carries the name and the payload, and a drain (this process or another) resolves them when the row is claimed.

Parameters:

  • job_name (String) —

    the job's name — the stored constant path

  • work (Proc) —

    the job, closed over its arguments (unused here — the row is the job)

  • payload (Object, nil) —

    the serializable arguments, stored as JSON text (nil for a payload-less job)



60
 
# File 'runtime/active_job/database_queue.rb', line 60

def self.enqueue: (String job_name, ^() -> void work, untyped? payload) -> void

.enqueue_at(job_name, work, payload, scheduled_at) ⇒ void

hand off a delayed one-off: the row waits in pending until a drain's clock passes scheduled_at. A one-off is NOT deduplicated: every call writes its own row, so the same name and run time with different payloads are different jobs.

Parameters:

  • job_name (String) —

    the job's name — the stored constant path

  • work (Proc) —

    the job, closed over its arguments (unused here)

  • payload (Object, nil) —

    the serializable arguments (JSON text)

  • scheduled_at (Time) —

    the run time — the Time slot is the contract, so a non-Time never reaches the row (the .rbs pins it)



76
 
# File 'runtime/active_job/database_queue.rb', line 76

def self.enqueue_at: (String job_name, ^() -> void work, untyped? payload, Time scheduled_at) -> void

.enqueue_once(name, work, payload, scheduled_at) ⇒ Boolean

the recurring half: one row per (name, interval start), so several processes firing the same beat still queue the occurrence once. The ledger insert and the queue row are one transaction — a ledger collision (another process won) answers false with no row written, so the ledger can never keep an occurrence whose row was lost. A one-off enqueue_at never comes through here.

Parameters:

  • name (String) —

    the job's name — the stored constant path

  • work (Proc) —

    the job, closed over its arguments (unused here)

  • payload (Object, nil) —

    the serializable arguments (unused — a recurring beat carries none)

  • scheduled_at (Time) —

    the interval's start, as the beat computed it (ActiveJob.scheduled_at_for)

Returns:

  • (Boolean) —

    true when this call queued the occurrence



95
 
# File 'runtime/active_job/database_queue.rb', line 95

def self.enqueue_once: (String name, ^() -> void work, untyped? payload, Time scheduled_at) -> bool