Documentation

Redis extends AbstractTaskAdapter
in package

Redis adapter class

Tags
category

Pop

author

Nick Sagona, III nick@popphp.org

copyright

Copyright (c) 2009-2026 Nick Sagona, III

license

https://www.popphp.org/license New BSD License

version
3.0.0

Table of Contents

Constants

TASK_CLAIM_TTL  = 90
How long, in seconds, a claim blocks a *same-window* re-claim.

Properties

$indexSetsChecked  : bool
Whether this instance has already reconciled the task and dead-letter index sets against any keys written before those sets existed. See ensureIndexSets().
$leaseSeconds  : int
Reservation lease length, in seconds
$prefix  : string
Queue prefix
$priority  : string
Queue priority
$redis  : Redis|null
Redis object

Methods

__construct()  : mixed
Constructor
bury()  : Redis
Move a job to the dead-letter store
claimTaskRun()  : bool
Atomically claim a task's current due-window. Redis has no single native command for "compare stored value, swap if different-window- or-expired", so this uses a small Lua eval() script - the same approach atomicReclaimIfStillExpired() already uses for job-lease reclaim. The key's value format is "<window>:<expiresAtUnixTimestamp>"; a native Redis EXPIRE is also set on a successful claim so garbage collection happens even if removeTask() is somehow skipped, on top of the explicit del() removeTask() performs.
clear()  : Redis
Clear pending and reserved jobs (not tasks or dead-letter jobs)
clearDead()  : Redis
Clear all dead jobs
clearTasks()  : Redis
Clear all scheduled task
count()  : int
Count of pending + reserved jobs
countDead()  : int
Count of dead jobs
create()  : Redis
Create Redis adapter
delete()  : Redis
Permanently remove a job
deleteDeadJob()  : Redis
Permanently delete a dead job
getAllTasks()  : array<string|int, mixed>
Get every scheduled task, keyed by task ID.
getDeadJob()  : mixed
Get a dead job
getDeadJobs()  : array<string|int, mixed>
Get dead jobs
getPrefix()  : string
Get prefix
getPriority()  : string
getRedis()  : Redis|null
Get Redis object
getTask()  : Task|null
Get scheduled task
getTaskCount()  : int
Get scheduled tasks count
getTasks()  : array<string|int, mixed>
Get scheduled tasks
hasDeadJobs()  : bool
Check if adapter has dead jobs
hasJobs()  : bool
Check if adapter has pending or reserved jobs
hasTasks()  : bool
Has scheduled tasks
isFifo()  : bool
isFilo()  : bool
isLifo()  : bool
isLilo()  : bool
push()  : Redis
Push job on to queue
redis()  : Redis|null
Get Redis object (alias)
release()  : Redis
Put a job back to pending, honoring its backoff schedule unless an explicit delay is given
removeTask()  : Redis
Remove scheduled task
reserve()  : AbstractJob|null
Atomically claim the next eligible job. Reclaims any reserved job whose lease has expired first, then scans the pending list in queue order and skips any job that isn't yet available, atomically claiming the first eligible one via a checked lRem() - if lRem() reports it removed nothing, another worker already claimed this same entry first, so this moves on to the next candidate instead of assuming success.
retryDeadJob()  : Redis
Retry a dead job by pushing it back on to the queue
schedule()  : Redis
Push job on to queue
setPriority()  : AbstractAdapter
updateTask()  : Redis
Update scheduled task
atomicReclaimIfStillExpired()  : int
Atomically remove a reserved-set member if, and only if, its *current* score (re-checked server-side at the moment this runs, not a stale snapshot from an earlier zRangeByScore() read) is still <= $now.
deadSetKey()  : string
Key of the set indexing dead-letter job IDs
ensureIndexSets()  : void
Populate the task and dead-letter index sets from any keys written before those sets existed, once per adapter instance.
reclaimExpiredLeases()  : void
Move any reserved job whose lease has expired back to the pending list, so a crashed worker's claim self-heals instead of being stuck forever.
removeFromReserved()  : string|null
Find a job matching a job ID in the reserved sorted set and remove it
taskSetKey()  : string
Key of the set indexing scheduled task IDs

Constants

TASK_CLAIM_TTL

How long, in seconds, a claim blocks a *same-window* re-claim.

protected mixed TASK_CLAIM_TTL = 90

Shared by every concrete adapter's claimTaskRun() implementation. Not configurable - it has no relationship to any task's cron recurrence interval (the explicit window value each implementation compares against is what makes that safe). It does need to outlast the longest window a claim must survive: a claim is never refreshed or released while its task runs, and a coarse (non-sub-minute) task is due across its entire ~60-second window (evaluate() stays true for the whole minute, not just at :00), so a second worker can legitimately re-evaluate the same coarse task's window many seconds after the first worker claimed it. 90 seconds covers a full 60-second coarse window plus slack, not just one claim-then-execute round trip.

Properties

$indexSetsChecked

Whether this instance has already reconciled the task and dead-letter index sets against any keys written before those sets existed. See ensureIndexSets().

protected bool $indexSetsChecked = false

$leaseSeconds

Reservation lease length, in seconds

protected int $leaseSeconds = 60

$prefix

Queue prefix

protected string $prefix = 'pop-queue'

$redis

Redis object

protected Redis|null $redis = null

Methods

__construct()

Constructor

public __construct([string $host = 'localhost' ][, int|string $port = 6379 ][, string $prefix = 'pop-queue' ][, string|null $priority = null ][, int $leaseSeconds = 60 ][, string|null $password = null ][, array<string|int, mixed>|null $context = null ]) : mixed

Instantiate the redis adapter

Parameters
$host : string = 'localhost'
$port : int|string = 6379
$prefix : string = 'pop-queue'
$priority : string|null = null
$leaseSeconds : int = 60
$password : string|null = null

Optional password for AUTH

$context : array<string|int, mixed>|null = null

Optional stream context (e.g. ['stream' => ['ssl' => [...]]] for TLS)

Tags
throws
Exception|RedisException

bury()

Move a job to the dead-letter store

public bury(AbstractJob $job[, string|null $reason = null ]) : Redis
Parameters
$job : AbstractJob
$reason : string|null = null
Return values
Redis

claimTaskRun()

Atomically claim a task's current due-window. Redis has no single native command for "compare stored value, swap if different-window- or-expired", so this uses a small Lua eval() script - the same approach atomicReclaimIfStillExpired() already uses for job-lease reclaim. The key's value format is "<window>:<expiresAtUnixTimestamp>"; a native Redis EXPIRE is also set on a successful claim so garbage collection happens even if removeTask() is somehow skipped, on top of the explicit del() removeTask() performs.

public claimTaskRun(string $taskId, string $window) : bool
Parameters
$taskId : string
$window : string
Return values
bool

clear()

Clear pending and reserved jobs (not tasks or dead-letter jobs)

public clear() : Redis
Return values
Redis

clearDead()

Clear all dead jobs

public clearDead() : Redis
Return values
Redis

clearTasks()

Clear all scheduled task

public clearTasks() : Redis
Return values
Redis

count()

Count of pending + reserved jobs

public count() : int
Return values
int

countDead()

Count of dead jobs

public countDead() : int
Return values
int

create()

Create Redis adapter

public static create([string $host = 'localhost' ][, int|string $port = 6379 ][, string $prefix = 'pop-queue' ][, string|null $priority = null ][, int $leaseSeconds = 60 ][, string|null $password = null ][, array<string|int, mixed>|null $context = null ]) : Redis
Parameters
$host : string = 'localhost'
$port : int|string = 6379
$prefix : string = 'pop-queue'
$priority : string|null = null
$leaseSeconds : int = 60
$password : string|null = null
$context : array<string|int, mixed>|null = null
Tags
throws
Exception|RedisException
Return values
Redis

deleteDeadJob()

Permanently delete a dead job

public deleteDeadJob(string $jobId) : Redis
Parameters
$jobId : string
Return values
Redis

getAllTasks()

Get every scheduled task, keyed by task ID.

public getAllTasks() : array<string|int, mixed>

One SMEMBERS plus one MGET, instead of the inherited "list the IDs, then GET each payload" - which is a round trip per scheduled task, paid on every Queue::run() and so on every tick of a worker's schedule loop.

Return values
array<string|int, mixed> —

taskId => Task

getDeadJob()

Get a dead job

public getDeadJob(string $jobId[, bool $unserialize = true ]) : mixed
Parameters
$jobId : string
$unserialize : bool = true

getDeadJobs()

Get dead jobs

public getDeadJobs([bool $unserialize = true ]) : array<string|int, mixed>
Parameters
$unserialize : bool = true
Return values
array<string|int, mixed>

getPrefix()

Get prefix

public getPrefix() : string
Return values
string

getRedis()

Get Redis object

public getRedis() : Redis|null
Return values
Redis|null

getTask()

Get scheduled task

public getTask(string $taskId) : Task|null
Parameters
$taskId : string
Return values
Task|null

getTaskCount()

Get scheduled tasks count

public getTaskCount() : int
Return values
int

getTasks()

Get scheduled tasks

public getTasks() : array<string|int, mixed>
Return values
array<string|int, mixed>

hasDeadJobs()

Check if adapter has dead jobs

public hasDeadJobs() : bool
Return values
bool

hasJobs()

Check if adapter has pending or reserved jobs

public hasJobs() : bool
Return values
bool

hasTasks()

Has scheduled tasks

public hasTasks() : bool
Return values
bool

redis()

Get Redis object (alias)

public redis() : Redis|null
Return values
Redis|null

release()

Put a job back to pending, honoring its backoff schedule unless an explicit delay is given

public release(AbstractJob $job[, int|null $delay = null ]) : Redis
Parameters
$job : AbstractJob
$delay : int|null = null
Return values
Redis

removeTask()

Remove scheduled task

public removeTask(string $taskId) : Redis
Parameters
$taskId : string
Return values
Redis

reserve()

Atomically claim the next eligible job. Reclaims any reserved job whose lease has expired first, then scans the pending list in queue order and skips any job that isn't yet available, atomically claiming the first eligible one via a checked lRem() - if lRem() reports it removed nothing, another worker already claimed this same entry first, so this moves on to the next candidate instead of assuming success.

public reserve() : AbstractJob|null
Return values
AbstractJob|null

retryDeadJob()

Retry a dead job by pushing it back on to the queue

public retryDeadJob(string $jobId) : Redis
Parameters
$jobId : string
Return values
Redis

updateTask()

Update scheduled task

public updateTask(Task $task) : Redis
Parameters
$task : Task
Return values
Redis

atomicReclaimIfStillExpired()

Atomically remove a reserved-set member if, and only if, its *current* score (re-checked server-side at the moment this runs, not a stale snapshot from an earlier zRangeByScore() read) is still <= $now.

protected atomicReclaimIfStillExpired(string $value, int $now) : int

This closes an ABA race: a worker's earlier "this looks expired" read can go stale if another worker reclaims and freshly re-claims the same entry before the first worker acts on it. Because a reclaim never re-serializes the job, the serialized value alone can't tell "the same expired claim" apart from "a fresh claim that happens to match" - a plain zRem($key, $value) would remove the fresh claim too, since it matches by value only and ignores the current score. Redis executes Lua scripts atomically, so this re-check-and-remove can't be interleaved by another command.

Parameters
$value : string
$now : int
Return values
int —

1 if removed, 0 if the entry is missing or no longer expired

deadSetKey()

Key of the set indexing dead-letter job IDs

protected deadSetKey() : string
Return values
string

ensureIndexSets()

Populate the task and dead-letter index sets from any keys written before those sets existed, once per adapter instance.

protected ensureIndexSets() : void

Those two collections used to be enumerated with KEYS, which Redis evaluates against its entire keyspace while blocking every other client on the server - and this adapter reached for it constantly, including to answer questions as small as hasTasks(). Maintaining the membership in a set instead turns all of it into SMEMBERS/SCARD/SISMEMBER against one key.

Which leaves the keys an older version already wrote and never indexed. A marker key records that the reconciliation has happened, so KEYS runs at most once per Redis database, ever, rather than never running and quietly orphaning every task and dead job that predates the upgrade. Two processes racing here is harmless: SADD is idempotent, so the worst case is the same work done twice.

reclaimExpiredLeases()

Move any reserved job whose lease has expired back to the pending list, so a crashed worker's claim self-heals instead of being stuck forever.

protected reclaimExpiredLeases() : void

removeFromReserved()

Find a job matching a job ID in the reserved sorted set and remove it

protected removeFromReserved(string $jobId) : string|null
Parameters
$jobId : string
Return values
string|null —

the removed member's serialized value, if found

taskSetKey()

Key of the set indexing scheduled task IDs

protected taskSetKey() : string
Return values
string 

        

        
    




    

    

                                

                

                                
    

        On this page

        
    


                

                            

            

    

        

            
Search results