Snails

Class: ActiveJob::ProcessQueue

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

Constant Summary collapse

MAX_PENDING =

The most jobs the queue holds. A producer can outrun the drain — a slow log receiver makes LogDeliveryJob entries accumulate faster than the worker executes them — and an unbounded Array would grow the process to OOM. The bound is a safety valve, not a throughput target: when it is reached the new job is dropped and reported locally once per overflow episode. 1,000 entries stays far above the batches a healthy 100-record flush produces in one flush interval.

1_000
PENDING =

The queue: one entry per job — a Hash carrying the zero-argument proc and its due time (due_at, epoch seconds; enqueue writes 0, enqueue_at the scheduled run time). Requests enqueue from their own threads and worker threads run the due jobs, so both ends run under a lock — and drain runs the due jobs OUTSIDE the lock, so a slow job never holds the queue against requests still enqueueing. One drain takes the whole due set, so several worker threads do not split a backlog; they can split jobs enqueued while another drain is running. Throughput parallelism lives in the DB adapter, which claims one row at a time.

[]
NEXT_ID =

Process-local monotonic job id: the process queue has no stored row, so the id only has to be unique within this process's lifetime. It resets on restart and is not a cross-process correlation key.

[ 0 ]
QUEUE_LOCK = 
Mutex.new
OVERFLOW_REPORTED =

Whether this overflow episode already wrote its warning. The warning is local-only (report_local) so it cannot enqueue a log delivery job and feed the backlog it reports. The flag re-arms after a drain leaves the queue at or below half the bound.

[ false ]

Class Method Summary collapse

Class Method Details

.drain ⇒ Integer

run every due job, in queue order, and answer how many ran. A job scheduled for a later time stays queued (due_at gates it); a job that enqueues another is reached by the same drain. The worker loop calls it (inside the loop's Sqlx::Database lease); the contract tests call it directly. A unit failure — anything the boundary does not re-raise — is one lost job, never a lost drain: job.failed records it and the queue moves on (an unhandled job error must not take the other jobs, or the process, with it). Control and fatal exceptions propagate. Each run sets the job values Context carries, so the records name the job.

Returns:

  • (Integer) —

    the number of jobs that ran



104
 
# File 'runtime/active_job/process_queue.rb', line 104

def self.drain: () -> Integer

.drain_due? ⇒ Boolean

whether a drain should run now: at least one entry's time has passed. A queue that holds only future jobs answers false until the clock reaches them.

Returns:

  • (Boolean) —

    true when any queued job is due



203
 
# File 'runtime/active_job/process_queue.rb', line 203

def self.drain_due?: () -> bool

.enqueue(job_name, work, payload) ⇒ void

put a job on the queue, due now. The name and payload are part of the adapter interface — enqueue passes all three to whichever adapter is selected — but this adapter keeps only the name (for the run's record); the proc IS the job, closed over its arguments. A full queue refuses the new job instead of growing without bound; the refusal is reported locally once per overflow episode.

Parameters:

  • job_name (String) —

    the job's class name, kept for the run's record

  • work (Proc) —

    the job, closed over its arguments

  • payload (Object, nil) —

    the serializable arguments (unused here)



57
 
# File 'runtime/active_job/process_queue.rb', line 57

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

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

put a job on the queue with a run time: the drain leaves it unrun until the clock (Time.now) passes scheduled_at. A store-backed adapter stores the same value in its row instead; the time is the whole difference from enqueue here. The Time slot refuses a non-Time argument at the build or the call boundary (ext/time.rb), so no in-body guard stands here. The same bound and refusal rule as enqueue applies: a scheduled job waiting for its time counts against the queue.

Parameters:

  • job_name (String) —

    the job's class name, kept for the run's record

  • work (Proc) —

    the job, closed over its arguments

  • payload (Object, nil) —

    the serializable arguments (unused here)

  • scheduled_at (Time) —

    the run time



74
 
# File 'runtime/active_job/process_queue.rb', line 74

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 of the adapter interface: this adapter is process-local, so the run's scheduled time has no cross-process work to do — the call runs the enqueue path. A store-backed adapter keys its row by (name, scheduled_at) under a unique constraint instead.

It answers whether THIS call enqueued the job: the process-local queue has no cross-process duplicate to drop, so it answers true unless the queue is full; a store-backed adapter answers false for an existing (name, scheduled_at) row.

Returns:

  • (Boolean) —

    true when this call enqueued the job



89
 
# File 'runtime/active_job/process_queue.rb', line 89

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

.pending_count ⇒ Integer

how many jobs are waiting (a scheduled job counts as waiting until its time passes). The contract tests read it; the worker loop's due check goes through drain_due?.

Returns:

  • (Integer) —

    the number of queued jobs



190
 
# File 'runtime/active_job/process_queue.rb', line 190

def self.pending_count: () -> Integer