class

ActiveJob::SolidQueue::Adapter

Inherits ActiveJob::Adapter < Reference < Object

Constructors

connect(url : String = ENV.fetch("DATABASE_URL", "postgres:///"), stale_claim_seconds : Int32 = 3600) : Adapter
Source
new(db : DB::Database, stale_claim_seconds : Int32 = 3600)
Source

Instance methods

claim(queues : Array(String), limit : Int32) : Array(Hash(String, ArgValue))

Claims up to limit due jobs from queues (priority DESC — higher number runs first, verified against solid_queue 1.2.4 — then FIFO by id, FOR UPDATE SKIP LOCKED). Returns job data hashes ready for Base.deserialize, carrying provider_job_id (the row id).

Source
claim(queue : String, limit : Int32) : Array(Hash(String, ArgValue))
Source
claimed_count
Source
clear
Source
db
enqueue(record : EnqueuedJob) : Nil
Source
enqueue_recurring(task_key : String, run_at : Time, record : EnqueuedJob) : Nil

Enqueue for recurring tasks: records task_key + run_at, dedups on (task_key, run_at) — the analog of RecurringExecution + unique index. The job itself is ready immediately (scheduled_at = now()): like solid_queue, the dispatcher enqueues AT the occurrence time, run_at records which occurrence this is.

Source
ensure_schema
Source
finished(data : Hash(String, ArgValue)) : Nil

Marks a performed job finished (the analog of ClaimedExecution#finished — the row stays for history).

Source
json_to_arg_value(v : JSON::Any) : ArgValue
Source
last_run_at(task_key : String) : Time | Nil
Source
pending_count(queue : String | Nil = nil) : Int32
Source
perform_one(data : Hash(String, ArgValue)) : Nil
Source
release(data : Hash(String, ArgValue)) : Nil

Releases a claimed job without finishing it (perform_now raised an unhandled error): the row becomes claimable again.

Source
release_stale_claims

Reclaims stale claims of dead workers (crash recovery).

Source
rows

Raw row rendering in the oracle's row() format (spec mirror).

Source
schedule_recurring(tasks : Array(RecurringTask), now : Time = Time.utc) : Nil

Scheduler tick: enqueue every task's next occurrence (dedup makes it idempotent). Call periodically from a dispatcher fiber.

Source
scheduled_count
Source
scheduled_first(n : Int32) : Array(Time)

First n scheduled (unfinished, future) run times, earliest first — the scheduler ordering (scheduled_at ASC, priority ASC, id ASC).

Source
serialize_payload(record : EnqueuedJob) : String

Stored payload = the ActiveJob serialize hash (job_class .. timezone) minus volatile identity (job_id/enqueued_at/scheduled_at) — the same payload solid_queue keeps in its arguments column.

Source
stale_claim_seconds
work(queues : Array(String) = ["default"], batch : Int32 = 10, poll_interval : Time::Span = 0.1.seconds, stop : Channel(Nil) | Nil = nil) : Nil

Runs the worker loop in the current fiber until stop is closed.

Source