Documentation

Database extends AbstractTaskAdapter
in package

Database 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

RESERVE_BATCH_SIZE  = 50
How many candidate rows reserve() pulls back per scan.
TASK_CLAIM_TTL  = 90
How long, in seconds, a claim blocks a *same-window* re-claim.

Properties

$db  : AbstractAdapter|null
Database adapter
$leaseSeconds  : int
Reservation lease length, in seconds
$priority  : string
Queue priority
$table  : string|null
Database table

Methods

__construct()  : mixed
Constructor
bury()  : Database
Move a job to the dead-letter store
claimTaskRun()  : bool
Atomically claim a task's current due-window. See buildTaskClaimEligibleWhere() for the eligibility rule.
clear()  : Database
Clear pending and reserved jobs (not tasks or dead-letter jobs)
clearDead()  : Database
Clear all dead-letter jobs
clearTasks()  : Database
Clear all scheduled task
count()  : int
Count of pending + reserved jobs
countDead()  : int
Count of dead-letter jobs
create()  : Database
Create database adapter
createTable()  : Database
Create the database table
db()  : AbstractAdapter|null
Get database adapter (alias)
delete()  : Database
Permanently remove a job
deleteDeadJob()  : Database
Permanently remove a dead-letter job
getAllTasks()  : array<string|int, mixed>
Get every scheduled task, keyed by task ID.
getDb()  : AbstractAdapter|null
Get database adapter
getDeadJob()  : mixed
Get a dead-letter job
getDeadJobs()  : array<string|int, mixed>
Get dead-letter jobs
getPriority()  : string
getTable()  : string|null
Get database table
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-letter jobs
hasJobs()  : bool
Check if adapter has jobs
hasTasks()  : bool
Has scheduled tasks
isFifo()  : bool
isFilo()  : bool
isLifo()  : bool
isLilo()  : bool
push()  : Database
Push job on to queue
release()  : Database
Put a job back to pending, honoring its backoff schedule unless an explicit delay is given
removeTask()  : Database
Remove scheduled task
reserve()  : AbstractJob|null
Atomically claim the next eligible job. Scans pending/expired-lease rows in queue order, skipping any job that isn't yet available (delayed or backed off), and atomically claims the first eligible one via a conditional UPDATE that writes a random claim token into reserved_by alongside status/reserved_until - all in the same statement.
retryDeadJob()  : Database
Move a dead-letter job back to pending
schedule()  : Database
Schedule job with queue
setPriority()  : AbstractAdapter
updateTask()  : Database
Update scheduled task
buildEligibleWhere()  : Where
Build the "pending, or claimed with an expired lease" predicate shared by reserve()'s scan and its per-row claiming UPDATE: type = 'job' AND (status = 1 OR (status = 0 AND reserved_until <= $now)). Built via pop-db's nested PredicateSet API (Where::andNest()/orNest()) rather than a raw SQL string, because pop-db's where()/andWhere() string parser only understands simple "column operator value" expressions - handing it a compound, parenthesized boolean expression silently misparses it instead of raising an error.
buildTaskClaimEligibleWhere()  : Where
Build the eligibility predicate for claiming a task's current due-window: eligible if no claim exists yet, the existing claim is for a *different* window (a new tick is always claimable immediately, regardless of the old claim's expiry), or the existing claim is for the *same* window but has expired. The window is encoded as a prefix of reserved_by itself ("<window>:<token>"), so this needs no new column - reserved_until still means "this claim expires at", reserved_by's prefix now also answers "for which window". "reserved_by NOT LIKE '<window>:%'" is plain, portable SQL LIKE - no dialect-specific string functions - and the window value is purely numeric so it can never contain a LIKE wildcard.
claimedBy()  : string|null
Read back the reserved_by token currently stored for a row. Used by reserve() to prove, via the real resulting row state rather than driver-reported execute metadata, whether its own claiming UPDATE was the one that actually won the row.
claimedByTaskId()  : string|null
Read back the reserved_by token currently stored for a task row.
createIndexes()  : void
Add the indexes the hot queries need to a table this adapter just created.
ensureReservedByColumn()  : void
Add the reserved_by column to a table created by an earlier version of this adapter, if it isn't there already. Same real column-existence check as ensureReservedUntilColumn() (see that method's docblock for why exception-based detection is unsafe here). reserved_by holds the random claim token reserve() writes and re-reads to prove its own claiming UPDATE actually won a given row - see reserve()'s docblock for why that replaced an affected-row count.
ensureReservedUntilColumn()  : void
Add the reserved_until column to a table created by an earlier version of this adapter, if it isn't there already. Gated on a real column-existence check (Pop\Db\Gateway\Table::getTableInfo(), backed by PRAGMA table_info on SQLite / information_schema.columns on Postgres+SQL Server / SHOW COLUMNS elsewhere - genuinely portable across every backend this adapter supports), NOT on catching the exception a duplicate-column ALTER is expected to throw: this repo's SQLite adapter's query() only calls throwError() when the driver reports a non-zero error code, and a duplicate-column ALTER TABLE ... ADD COLUMN against SQLite returns false with error code 0 - it fails silently (surfacing only as a PHP warning), so a try/catch around it never fires. That previously let the backfill below re-run on every single construction against an already-migrated table, zeroing out reserved_until (and the lease it represents) for every currently in-flight job, table-wide, on ordinary worker startup - exactly the double-execution failure this task exists to prevent. Gating on the real column check means the ALTER and backfill run exactly once, ever, per table, and a genuine migration failure (permissions, locked table) now throws normally instead of failing silently.
getEndIndex()  : int
Get queue end index

Constants

RESERVE_BATCH_SIZE

How many candidate rows reserve() pulls back per scan.

protected mixed RESERVE_BATCH_SIZE = 50

reserve() returns a single job, but it has to look past any candidate it can't claim (delayed, backed off, corrupt, or won by another worker), so it needs more than one row in hand - and, when a whole batch turns out to be unclaimable, the freedom to page further back. Without a bound it read the entire eligible set into PHP to hand back one job, which made the cost of reserving grow with the depth of the queue: every worker paid for every queued job, on every job it ran.

Sized to clear the ordinary case in one round trip - a run of delayed jobs at the head of the queue is the usual reason for a skip, and it is rarely dozens long - while staying small enough that the payloads fetched alongside them stay cheap.

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

$leaseSeconds

Reservation lease length, in seconds

protected int $leaseSeconds = 60

$table

Database table

protected string|null $table = null

Methods

__construct()

Constructor

public __construct(AbstractAdapter $db[, string $table = 'pop_queue' ][, string|null $priority = null ][, int $leaseSeconds = 60 ]) : mixed

Instantiate the database adapter object

Parameters
$db : AbstractAdapter
$table : string = 'pop_queue'
$priority : string|null = null
$leaseSeconds : int = 60

claimTaskRun()

Atomically claim a task's current due-window. See buildTaskClaimEligibleWhere() for the eligibility rule.

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() : Database
Return values
Database

count()

Count of pending + reserved jobs

public count() : int
Return values
int

countDead()

Count of dead-letter jobs

public countDead() : int
Return values
int

create()

Create database adapter

public static create(AbstractAdapter $db[, string $table = 'pop_queue' ][, string|null $priority = null ][, int $leaseSeconds = 60 ]) : Database
Parameters
$db : AbstractAdapter
$table : string = 'pop_queue'
$priority : string|null = null
$leaseSeconds : int = 60
Return values
Database

createTable()

Create the database table

public createTable(string $table) : Database
Parameters
$table : string
Return values
Database

deleteDeadJob()

Permanently remove a dead-letter job

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

getAllTasks()

Get every scheduled task, keyed by task ID.

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

One query for the whole set, instead of the inherited "list the IDs, then SELECT each payload by ID" - which is a query per scheduled task, run on every Queue::run() and so on every tick of a worker's schedule loop. The payloads are decoded exactly as getTask() decodes its one, corrupt entries omitted rather than returned.

Return values
array<string|int, mixed> —

taskId => Task

getDeadJob()

Get a dead-letter job

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

getDeadJobs()

Get dead-letter jobs

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

getTable()

Get database table

public getTable() : string|null
Return values
string|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-letter jobs

public hasDeadJobs() : bool
Return values
bool

hasJobs()

Check if adapter has jobs

public hasJobs() : bool
Return values
bool

hasTasks()

Has scheduled tasks

public hasTasks() : bool
Return values
bool

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 ]) : Database
Parameters
$job : AbstractJob
$delay : int|null = null
Return values
Database

removeTask()

Remove scheduled task

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

reserve()

Atomically claim the next eligible job. Scans pending/expired-lease rows in queue order, skipping any job that isn't yet available (delayed or backed off), and atomically claims the first eligible one via a conditional UPDATE that writes a random claim token into reserved_by alongside status/reserved_until - all in the same statement.

public reserve() : AbstractJob|null

The scan runs in batches of RESERVE_BATCH_SIZE rather than over the whole eligible set at once, paging forward only when an entire batch turns out to be unclaimable. Paging is inherently approximate - rows claimed by other workers between batches shift the offset, so a candidate can slip past - but the scan was already racy by construction (that is what the read-back below exists to handle), and anything missed is simply picked up by the next reserve() call rather than lost.

Success is proven by re-reading reserved_by immediately after the UPDATE and comparing it to the token this call generated, not by inspecting the UPDATE's driver-reported affected-row count. getNumberOfAffectedRows() is not portable enough for this: pop-db's Pgsql adapter reads it off the prepare result rather than the execute result, so pg_affected_rows() on a PREPARE always reports 0

  • meaning a real, successful claim on Postgres would still read back as "0 affected rows" and reserve() would treat every genuine win as a lost race, lease every row it touches, and never return a job. Reading back the actual resulting row state instead works identically across every backend, because it isn't asking the driver to describe what it did - it's asking the database what's really there now.

If reserved_by doesn't come back as this call's token (someone else's token is there, or the row's state changed/vanished), this worker lost the race between the scan and its UPDATE, so this moves on to the next candidate instead of assuming success.

Return values
AbstractJob|null

retryDeadJob()

Move a dead-letter job back to pending

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

buildEligibleWhere()

Build the "pending, or claimed with an expired lease" predicate shared by reserve()'s scan and its per-row claiming UPDATE: type = 'job' AND (status = 1 OR (status = 0 AND reserved_until <= $now)). Built via pop-db's nested PredicateSet API (Where::andNest()/orNest()) rather than a raw SQL string, because pop-db's where()/andWhere() string parser only understands simple "column operator value" expressions - handing it a compound, parenthesized boolean expression silently misparses it instead of raising an error.

protected buildEligibleWhere(AbstractSql $sql, int $now) : Where
Parameters
$sql : AbstractSql
$now : int
Return values
Where

buildTaskClaimEligibleWhere()

Build the eligibility predicate for claiming a task's current due-window: eligible if no claim exists yet, the existing claim is for a *different* window (a new tick is always claimable immediately, regardless of the old claim's expiry), or the existing claim is for the *same* window but has expired. The window is encoded as a prefix of reserved_by itself ("<window>:<token>"), so this needs no new column - reserved_until still means "this claim expires at", reserved_by's prefix now also answers "for which window". "reserved_by NOT LIKE '<window>:%'" is plain, portable SQL LIKE - no dialect-specific string functions - and the window value is purely numeric so it can never contain a LIKE wildcard.

protected buildTaskClaimEligibleWhere(AbstractSql $sql, string $taskId, string $window, int $now) : Where
Parameters
$sql : AbstractSql
$taskId : string
$window : string
$now : int
Return values
Where

claimedBy()

Read back the reserved_by token currently stored for a row. Used by reserve() to prove, via the real resulting row state rather than driver-reported execute metadata, whether its own claiming UPDATE was the one that actually won the row.

protected claimedBy(int $id) : string|null
Parameters
$id : int
Return values
string|null

claimedByTaskId()

Read back the reserved_by token currently stored for a task row.

protected claimedByTaskId(string $taskId) : string|null

Used by claimTaskRun() to prove, via the real resulting row state, whether its own claiming UPDATE actually won - the same approach claimedBy() uses for job claiming (see reserve()'s docblock for why an affected-row count isn't portable enough for this).

Parameters
$taskId : string
Return values
string|null

createIndexes()

Add the indexes the hot queries need to a table this adapter just created.

protected createIndexes(string $table) : void

Every query on the hot paths filters on type and then either orders by index (reserve(), getEndIndex()) or looks a row up by job_id (getTask(), claimedByTaskId(), the dead-letter accessors). Unindexed, all of those are full table scans, and reserve()'s is a scan plus a sort - paid by every worker on every job it runs, against a table whose whole purpose is to accumulate rows.

Issued as separate statements, and separate schema objects, for two reasons that are easy to get wrong:

  • Chaining index() onto the create() above renders valid DDL but does not execute it. The rendered schema becomes several statements separated by semicolons, and the adapters hand the whole string to one driver call - SQLite3::query() runs the first statement and silently discards the rest, so the table would appear and the indexes just wouldn't.
  • Casting a schema object to string consumes it. Rendering one to inspect it (or reusing one across two query() calls) leaves an empty builder behind, which also fails silently.

The index names are given explicitly because deriving them would involve the "index" column, a reserved word on every backend here.

Only ever called for a table this adapter is creating from scratch. An existing table is left alone deliberately: building an index on a live queue table can hold a lock for as long as the table is large, which is a decision for whoever operates the database, not one a library should make on their behalf on first connect.

Parameters
$table : string

ensureReservedByColumn()

Add the reserved_by column to a table created by an earlier version of this adapter, if it isn't there already. Same real column-existence check as ensureReservedUntilColumn() (see that method's docblock for why exception-based detection is unsafe here). reserved_by holds the random claim token reserve() writes and re-reads to prove its own claiming UPDATE actually won a given row - see reserve()'s docblock for why that replaced an affected-row count.

protected ensureReservedByColumn(string $table) : void
Parameters
$table : string

ensureReservedUntilColumn()

Add the reserved_until column to a table created by an earlier version of this adapter, if it isn't there already. Gated on a real column-existence check (Pop\Db\Gateway\Table::getTableInfo(), backed by PRAGMA table_info on SQLite / information_schema.columns on Postgres+SQL Server / SHOW COLUMNS elsewhere - genuinely portable across every backend this adapter supports), NOT on catching the exception a duplicate-column ALTER is expected to throw: this repo's SQLite adapter's query() only calls throwError() when the driver reports a non-zero error code, and a duplicate-column ALTER TABLE ... ADD COLUMN against SQLite returns false with error code 0 - it fails silently (surfacing only as a PHP warning), so a try/catch around it never fires. That previously let the backfill below re-run on every single construction against an already-migrated table, zeroing out reserved_until (and the lease it represents) for every currently in-flight job, table-wide, on ordinary worker startup - exactly the double-execution failure this task exists to prevent. Gating on the real column check means the ALTER and backfill run exactly once, ever, per table, and a genuine migration failure (permissions, locked table) now throws normally instead of failing silently.

protected ensureReservedUntilColumn(string $table) : void

The backfill sets reserved_until = 0 for any row already sitting at status = 0 - a job reserved under the pre-lease Phase 1 contract. Without it, such a row would have reserved_until = NULL forever, and "NULL <= now" evaluates to NULL in SQL - matching neither the "status = 1" nor the "status = 0 AND reserved_until <= now" branch of reserve()'s eligibility check, making the row invisible to reserve() permanently. Backfilling it to 0 makes it immediately eligible for reclaim on the very next reserve() call, which is the correct behavior for a job whose reservation state predates leasing entirely.

Parameters
$table : string

getEndIndex()

Get queue end index

protected getEndIndex() : int
Return values
int 

        

        
    




    

    

                                

                

                                
    

        On this page

        
    


                

                            

            

    

        

            
Search results