JoobQ::RedisStore
Inherits JoobQ::Store < Reference < Object
Constants
Each process claims into its own processing list, joobq:processing:<queue>:<owner>, and publishes a liveness
heartbeat in the joobq:processing:owners sorted set (score = last beat, ms). Recovery returns a processing
list to its queue only when its owner's heartbeat has lapsed, so a live process's in-flight jobs are never
moved out from under it. joobq:processing:<queue> with no owner suffix is the shared list older (0.5)
workers claimed into; it is migrated once no live owner remains. The heartbeat is independent of the optional
WorkerRegistry, so recovery is correct whether or not a process publishes the richer worker snapshot.
Not under the joobq:processing: prefix, so the processing-list SCANs never match it, and a queue
literally named "owners" does not collide with it.
Source membership is the claim: competing schedulers cannot promote the same member twice. Write the destination first: Lua does not roll back on errors. JSON is parsed/re-encoded in Crystal, preserving Int64 values (Lua cjson uses doubles).
Drains a processing list to the head of its queue, oldest claim first, in one atomic script, and returns the count. Each claim pushed the job onto the head of its processing list (BLMOVE/LMOVE ... LEFT LEFT, and the old BRPOPLPUSH too), so moving head to head preserves claim order: the recovered jobs run next, before the jobs still waiting. Moves up to ARGV[1] jobs from the head of a (dead owner's) processing list to the head of its queue, and returns how many it moved. Bounded so a huge leftover list does not block single-threaded Redis in one script; the caller loops until a short batch signals the list is empty. Nothing else claims from a dead owner's list, so order is preserved across batches.
Takes the job off whichever of this process's processing lists holds it (its own owner list, or the legacy shared list a seeded or pre-0.6 job sits in) and pushes it to the tail of the queue. KEYS: self list, legacy list, queue.
REQUEUE to the head of the queue, for many jobs in one script. KEYS are (self list, legacy list, queue) per
entry, ARGV the jobs, each taken off whichever list holds it and LPUSHed in turn (so the last entry ends up at
the head). An entry whose commands fail (a key of the wrong type, say) is skipped inside the script, so one bad
entry neither stops the others nor produces an error reply; a pipeline would raise at the first error reply and
leave the connection with unread replies.
Constructors
The store the environment asks for: REDIS_URL (with REDIS_TLS_VERIFY and REDIS_CA_FILE) when it is set, otherwise
REDIS_HOST, REDIS_PORT and REDIS_PASS. A REDIS_URL that cannot be used raises RedisEndpoint::Invalid rather
than falling back to a plain connection to REDIS_HOST.
Connects to the server a redis:// or rediss:// URL names (see RedisEndpoint): as an ACL user when the
URL has a user name, over TLS for rediss, on the database in its path. verify and ca_file control how
the server certificate is checked.
Class methods
true/1/yes/on and false/0/no/off (any case); nil when value is unset or empty. Anything else
raises, because a misspelt "off" would otherwise be read as "on".
Instance methods
Claims the oldest job, blocking up to BLOCKING_TIMEOUT, on the caller's dedicated connection (not the pool).
Claims into this process's own processing list, exactly like dequeue, so recovery is unchanged. Returns nil
only on an empty queue (the block timed out). A connection error — including the close used to interrupt the
block on stop — is raised, not swallowed, so the dispatcher can tell "empty" from "Redis is down" and back off
on the latter instead of reconnect-storming.
A dedicated, non-pooled connection for a queue dispatcher's blocking BLMOVE. The command timeout is kept well above the block timeout so the block itself never trips it; closing the connection interrupts a blocked call.
Claims the oldest job (see dequeue)
Claims up to batch_size jobs, oldest first (see dequeue_batch)
Removes a finished job from the processing list and counts it as processed
Batch job cleanup for high performance
Optimized method to clear multiple queues in a single pipeline
Claims the oldest job: producers RPUSH to the tail, so taking the head (BLMOVE ... LEFT LEFT, Redis 6.2+) gives
first-in-first-out order. The job moves to the head of the processing list in the same atomic step, so a crash
after the claim leaves it there for move_job_back_to_queue. Blocks up to BLOCKING_TIMEOUT when the queue is empty.
Claims up to batch_size jobs from the head of the queue without blocking, oldest first (the order the replies
come back in). Each LMOVE is atomic on its own, like the single claim; an empty queue answers nil.
Desired worker count shared between processes, with when it was set.
Claims use LMOVE/BLMOVE (Redis 6.2+). On an older server every claim would fail and be logged while the queue grows, so a queue refuses to start instead. Asks the server itself (LMOVE on a key that does not exist moves nothing) rather than parsing a version string, so Redis-compatible servers that implement the command pass.
Simplified batch job lookup - search in all known locations
Optimized method to get job counts for multiple states at once
Get queue metrics for multiple queues using pipelining
Optimized retrying jobs pagination using Lua script
Host, port and pool only: the default inspect would print the password this store was given.
A page of the waiting jobs, the next to run first (workers claim from the head of the list).
Owner ids whose heartbeat is within the TTL. Lapsed entries are pruned as they are read. Liveness is scored by the Redis clock, so recovery does not depend on the worker and reaper clocks agreeing. A process paused (GC, starvation) longer than OWNER_TTL can still be judged dead and have its jobs recovered, so jobs stay idempotent.
Scored by time of death like every other dead-letter writer, so entries sort
and age consistently; expiration_time is kept for the Store interface.
Mark job as completed with statistics
Reclaims the queue's orphaned (dead-owner) processing lists; never touches a live process's list. Queue#start
passes include_self: true to also clear a previous incarnation's leftovers before this process claims. True
when at least one job was moved; false when none, or when Redis failed (logged). Operate this on a live system
only to recover crashed workers: it cannot take a job a live worker still holds.
High-performance move to dead letter queue using pipelined operations
High-performance move to retry queue using pipelined operations
Identifies this process's processing lists and heartbeat. Stable for the life of the process; a crashed process that restarts gets a new pid and so a new id, which leaves its old list orphaned for recovery. Settable so a spec can act as several processes against one Redis.
Keeps the batch-cleanup pipeline on the same owner id, so a spec that poses as another process cleans up the list it claimed into.
Optional capabilities. Reads are safe defaults; writes raise, so a store that cannot pause or resize queues says so instead of silently ignoring it.
Preserve the legacy return contract: all scanned due jobs, including other queues.
Optimized processing jobs list with batch operations
The processing-list keys that belong to queue_name: one per owner (...:<queue>:<owner>) plus the shared
legacy list (...:<queue>) older workers used. A SCAN by prefix also matches another queue whose name starts
with this one, so each key is checked against the two exact shapes.
Get count of jobs currently in processing queue The jobs in processing across every owner of the queue, plus the legacy shared list.
Optimized batch queue sizes to reduce connection overhead
Returns the jobs of crashed or stopped processes to the head of their queue, in claim order, and deletes the
drained list. A processing list is reclaimed only when its owner's heartbeat has lapsed (live_owners), so a
running process's in-flight jobs are never moved. include_self drains this process's own list too, which is
safe at start-up (no worker of this process is claiming yet) but never from the periodic reaper, where this
process is live and working. The shared legacy list (no owner) is migrated only when no owner is live at all,
the first 0.6 process into an otherwise-drained cluster; stop 0.5 workers first (see the CHANGELOG). Returns
how many jobs were moved.
The Redis server clock, in milliseconds, so every process's heartbeat and every reader's TTL cutoff are on one clock: a process is never judged dead because another node's wall clock runs ahead of it.
Nothing to release: a claim is the job's place in the processing list, which completion, retry or dead-lettering removes
Nothing to release (see release_job_claim)
Puts a claimed job back at the tail of its queue (behind the waiting jobs), in one step: it is taken off the processing list and pushed, or (when it is no longer there, e.g. another process reclaimed it) nothing happens. Not counted as processed. Used for a job bounced for missing encryption keys, which should not come straight back.
Returns jobs abandoned at shutdown, given as {queue, job} pairs in claim order, to the head of their queues so they run next, in that order; returns how many were put back. Each job is taken off this process's own processing list (or the legacy shared list) and skipped if it is no longer there. The entries are pushed last to first (each push goes in front of the one before), one script per hundred; if a script itself fails, its entries are tried one by one and a failing one is logged and skipped.
Schedule delayed retry - simplified version of move_to_retry
Compare-and-set for a reviewed operator change. A concurrent operator's state is not overwritten after the preview was taken.
Optimized batch set sizes to reduce connection overhead
Starts this process's liveness heartbeat (idempotent, process-wide). Queue#start calls it before claiming,
so every processing list a 0.6 process owns has a live heartbeat while the process runs.
Stops the heartbeat and drops this process's owner entry, so recovery on another process can reclaim this process's lists at once rather than waiting out the TTL. A crash skips this and the entry lapses after the TTL.
Verify that a job has been properly removed from processing queue
Verify that a job exists in only one location (for debugging/testing) Returns a hash with the job's locations