Documentation

File extends AbstractTaskAdapter
in package

File 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

$endIndex  : int|null
Highest job index this adapter instance knows to be taken, or null when it hasn't looked yet. Purely a starting hint for push() - never a source of truth. See getEndIndex() for why being wrong in either direction is safe.
$folder  : string|null
Folder
$leaseSeconds  : int
Reservation lease length, in seconds
$priority  : string
Queue priority

Methods

__construct()  : mixed
Constructor
bury()  : File
Move a job to the dead-letter store
claimTaskRun()  : bool
Atomically claim a task's current due-window via a small sidecar file, deliberately separate from the task's own serialized definition file (task-<taskId>) so a claim attempt never touches or re-serializes the closure-bearing Task object. Content format is "<window>:<expiresAtUnixTimestamp>". flock() provides real cross-process mutual exclusion for the read-decide-write.
clear()  : File
Clear pending and reserved jobs (not tasks or dead-letter jobs)
clearDead()  : File
Clear all dead-letter jobs
clearTasks()  : File
Clear all scheduled task
count()  : int
Count of pending + reserved jobs
countDead()  : int
Count of dead-letter jobs
create()  : File
Create file adapter
delete()  : File
Permanently remove a job
deleteDeadJob()  : File
Permanently remove a dead-letter job
folder()  : string|null
Get folder (alias)
getAllTasks()  : array<string|int, mixed>
Get every scheduled task, keyed by task ID.
getDeadJob()  : mixed
Get a dead-letter job
getDeadJobs()  : array<string|int, mixed>
Get dead-letter jobs
getFiles()  : array<string|int, mixed>
Get files from folder
getFolder()  : string|null
Get folder
getFolders()  : array<string|int, mixed>
Get folders
getPriority()  : string
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()  : File
Push job on to queue
release()  : File
Put a job back to pending, honoring its backoff schedule unless an explicit delay is given
removeTask()  : File
Remove scheduled task
reserve()  : AbstractJob|null
Atomically claim the next eligible job. Reclaims any reserved job whose lease has expired first, then scans pending jobs in FIFO/FILO order, skipping any that aren't yet available, and atomically claims the first eligible one via rename() - if the rename fails, another worker won the race and this moves on to the next candidate.
retryDeadJob()  : File
Move a dead-letter job back to pending
schedule()  : File
Schedule job with queue
setPriority()  : AbstractAdapter
updateTask()  : File
Update scheduled task
claimPendingDir()  : bool
Atomically claim a pending job directory by moving it into reserved/.
findReservedIndexForJob()  : int|null
Find the reserved-job directory index holding a given job, if any
getDeadJobIds()  : array<string|int, mixed>
Get the dead-letter job IDs
getEndIndex()  : int
Get queue end index across both pending and reserved jobs (indices must stay unique across both so a reclaimed reserved job can never collide with a newly-pushed one)
getLeaseUntil()  : int|null
Read a reserved job's lease expiry, if any
isLeaseExpired()  : bool
Determine whether a reserved job's lease has expired. The directory's own mtime is the primary signal - reserve() freshens it at claim time (rename() itself doesn't update it), so a fresh claim's mtime alone is sufficient to prove it isn't expired, regardless of what stale lease content happens to still be sitting in the directory (a job that was previously release()d or reclaimed carries its old, already-expired lease file back into pending/ with it - nothing unlinks it, so the next claim's directory starts out holding a stale expired lease next to a brand new mtime). The lease file is only consulted to confirm expiry once the mtime already looks stale, never to override a fresh mtime.
pendingPath()  : string
Get the pending-jobs subdirectory path
readDirectory()  : array<string|int, mixed>
List the entries of a directory, keeping either the subdirectories or the plain files.
readVerifiedPayload()  : string|false
Read a payload file and verify its signature, or false if it can't be read
reclaimExpiredLeases()  : void
Move any reserved job whose lease has expired back to pending, so a crashed worker's claim self-heals instead of being stuck forever.
removeJobDir()  : void
Empty and remove a job directory.
reservedPath()  : string
Get the reserved-jobs subdirectory path
taskClaimPath()  : string
Get the claim-marker file path for a task

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

$endIndex

Highest job index this adapter instance knows to be taken, or null when it hasn't looked yet. Purely a starting hint for push() - never a source of truth. See getEndIndex() for why being wrong in either direction is safe.

protected int|null $endIndex = null

$folder

Folder

protected string|null $folder = null

$leaseSeconds

Reservation lease length, in seconds

protected int $leaseSeconds = 60

Methods

__construct()

Constructor

public __construct(string $folder[, string|null $priority = null ][, int $leaseSeconds = 60 ]) : mixed

Instantiate the file object

Parameters
$folder : string
$priority : string|null = null
$leaseSeconds : int = 60
Tags
throws
Exception

bury()

Move a job to the dead-letter store

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

claimTaskRun()

Atomically claim a task's current due-window via a small sidecar file, deliberately separate from the task's own serialized definition file (task-<taskId>) so a claim attempt never touches or re-serializes the closure-bearing Task object. Content format is "<window>:<expiresAtUnixTimestamp>". flock() provides real cross-process mutual exclusion for the read-decide-write.

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

clearDead()

Clear all dead-letter jobs

public clearDead() : File
Return values
File

clearTasks()

Clear all scheduled task

public clearTasks() : File
Return values
File

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 file adapter

public static create(string $folder[, string|null $priority = null ][, int $leaseSeconds = 60 ]) : File
Parameters
$folder : string
$priority : string|null = null
$leaseSeconds : int = 60
Tags
throws
Exception
Return values
File

deleteDeadJob()

Permanently remove a dead-letter job

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

folder()

Get folder (alias)

public folder() : string|null
Return values
string|null

getAllTasks()

Get every scheduled task, keyed by task ID.

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

Concrete, not abstract, so an adapter only overrides it if its storage can genuinely do better than one fetch per task - which the Database and Redis adapters both can, and do. This fallback is the loop it replaces, kept so that adding the method to TaskAdapterInterface doesn't oblige every adapter to reimplement it.

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>

getFiles()

Get files from folder

public getFiles(string $folder) : array<string|int, mixed>
Parameters
$folder : string
Return values
array<string|int, mixed>

getFolder()

Get folder

public getFolder() : string|null
Return values
string|null

getFolders()

Get folders

public getFolders(string $folder) : array<string|int, mixed>
Parameters
$folder : string
Return values
array<string|int, mixed>

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

removeTask()

Remove scheduled task

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

reserve()

Atomically claim the next eligible job. Reclaims any reserved job whose lease has expired first, then scans pending jobs in FIFO/FILO order, skipping any that aren't yet available, and atomically claims the first eligible one via rename() - if the rename fails, another worker won the race and this moves on to the next candidate.

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

retryDeadJob()

Move a dead-letter job back to pending

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

schedule()

Schedule job with queue

public schedule(Task $task) : File
Parameters
$task : Task
Return values
File

updateTask()

Update scheduled task

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

claimPendingDir()

Atomically claim a pending job directory by moving it into reserved/.

protected claimPendingDir(string $pendingDir, string $reservedDir) : bool

rename() is the whole claim: exactly one worker's rename off a given source directory can succeed. Isolated into its own method (rather than inlined in reserve()) purely so a test can deterministically simulate a concurrent reclaim landing in the narrow window between this rename winning and reserve()'s follow-up touch()/lease write - see the guard in reserve() for what that window can otherwise damage.

Parameters
$pendingDir : string
$reservedDir : string
Return values
bool

findReservedIndexForJob()

Find the reserved-job directory index holding a given job, if any

protected findReservedIndexForJob(AbstractJob $job) : int|null

Called by release(), delete() and bury() - so once per job a worker finishes, whatever the outcome. It used to answer by reading and unserializing every reserved payload in turn, which meant reconstructing whole job objects (closures included) purely to read one string off each.

reserve() now writes the claimed job's ID into a 'job-id' file beside the payload, so the common path compares a short string read against a string, and the payload is only unserialized for directories written before this file existed - or by a reserve() that died between its rename() and its sidecar write. Keeping that fallback is what makes the sidecar a pure optimization: its absence costs speed, never correctness.

Parameters
$job : AbstractJob
Return values
int|null

getDeadJobIds()

Get the dead-letter job IDs

protected getDeadJobIds() : array<string|int, mixed>
Return values
array<string|int, mixed>

getEndIndex()

Get queue end index across both pending and reserved jobs (indices must stay unique across both so a reclaimed reserved job can never collide with a newly-pushed one)

protected getEndIndex() : int

Scans the directories once per adapter instance and then tracks the index forward in memory, because the scan is what made push() quadratic: it walked both directories on every single push, so the cost of pushing the Nth job grew with the number of jobs already queued.

Caching it is safe precisely because push() never trusted this value in the first place - mkdir() is the real allocator, and push()'s retry loop already handles the index being taken. A cached value that is too low (another process pushed since the scan) costs one wasted mkdir() attempt per collision and the loop walks up; a cached value that is too high (jobs were cleared elsewhere) just leaves a gap in the numbering, which nothing depends on - indices only ever need to be unique and ordered, never contiguous.

Return values
int

getLeaseUntil()

Read a reserved job's lease expiry, if any

protected getLeaseUntil(string $reservedDir) : int|null
Parameters
$reservedDir : string
Return values
int|null

isLeaseExpired()

Determine whether a reserved job's lease has expired. The directory's own mtime is the primary signal - reserve() freshens it at claim time (rename() itself doesn't update it), so a fresh claim's mtime alone is sufficient to prove it isn't expired, regardless of what stale lease content happens to still be sitting in the directory (a job that was previously release()d or reclaimed carries its old, already-expired lease file back into pending/ with it - nothing unlinks it, so the next claim's directory starts out holding a stale expired lease next to a brand new mtime). The lease file is only consulted to confirm expiry once the mtime already looks stale, never to override a fresh mtime.

protected isLeaseExpired(string $dir, int $now) : bool
Parameters
$dir : string
$now : int
Return values
bool

pendingPath()

Get the pending-jobs subdirectory path

protected pendingPath() : string
Return values
string

readDirectory()

List the entries of a directory, keeping either the subdirectories or the plain files.

protected readDirectory(string $folder, bool $directories) : array<string|int, mixed>

Both public listers route through here rather than each running their own scandir(). Two things made that pairing expensive on the hot paths - reserve() and count() call it on every invocation, once per pending or reserved job:

  • scandir() returns names only, so deciding what each entry is meant an is_dir() stat syscall per entry. FilesystemIterator carries the type along with the directory read, so isDir() answers from what the OS already handed back.
  • scandir() sorts alphabetically by default, and every caller here either wants numeric order (reserve() re-sorts these index names itself) or no order at all, so that sort was pure waste.

A missing directory reads as empty rather than raising, matching the is_dir() guard both listers carried before.

Parameters
$folder : string
$directories : bool
Return values
array<string|int, mixed>

readVerifiedPayload()

Read a payload file and verify its signature, or false if it can't be read

protected readVerifiedPayload(string $path) : string|false

Every caller already treats a false return as "unusable payload, skip it", so an unreadable file joins the corrupt and tampered ones on that path. The explicit check matters because file_get_contents() signals failure with false rather than '', and under declare(strict_types=1) passing that to PayloadSigner::verify(string) is a TypeError - which would turn an everyday race (another worker claiming and unlinking the same payload between the file_exists() check and the read) into a crashed worker.

Parameters
$path : string
Return values
string|false

reclaimExpiredLeases()

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

protected reclaimExpiredLeases() : void

Reclaimed jobs are eligible on this or a later reserve() call, not necessarily returned by this one.

Deliberately not throttled to one sweep per second, which would otherwise look free: every input to the expiry decision has one-second resolution, so two sweeps within the same second agree whenever the only thing moving is the clock. That is not the only thing that can move. A lease can be expired by writing to it, and a caller that does so and then calls reserve() is entitled to see the reclaim happen on that call rather than on whichever one lands in the next second. The sweep is cheap in any case, because reserved/ only ever holds jobs currently in flight - it is bounded by the number of live workers, not by queue depth.

removeJobDir()

Empty and remove a job directory.

protected removeJobDir(string $dir) : void

Deliberately generic rather than unlinking 'payload' and 'lease' by name: rmdir() fails on a non-empty directory, so every file a job directory can hold has to be accounted for here, and naming them individually means any future addition silently turns delete() into a no-op that leaves the directory behind. Clearing whatever is actually in there cannot drift out of sync that way.

Parameters
$dir : string

reservedPath()

Get the reserved-jobs subdirectory path

protected reservedPath() : string
Return values
string

taskClaimPath()

Get the claim-marker file path for a task

protected taskClaimPath(string $taskId) : string
Parameters
$taskId : string
Return values
string 

        

        
    




    

    

                                

                

                                
    

        On this page

        
    


                

                            

            

    

        

            
Search results