Queues & Scheduled Tasks

A queue holds work to be done later. A job runs once and leaves the queue; a task is a job with a schedule, which stays and runs again each time it comes due. Both are stored through an adapter — Redis, a database table, files on disk, memory or SQS — and both are executed by a worker, whether that is a cron entry firing once a minute or a long-running daemon.

composer require popphp/pop-queue

Kettle drives the running side — queue:config, queue:work and queue:scheduler are covered in Kettle. This page is what goes into a queue and how it is defined.

Jobs#

A job wraps something callable. Job::create() takes it plus any parameters, and addJob() pushes it to the adapter:

use Pop\Queue\Adapter\File;
use Pop\Queue\Process\Job;
use Pop\Queue\Queue;

$queue = new Queue('pop-queue', new File(__DIR__ . '/queue'));

$job = Job::create(function() {
    echo 'This is a job' . PHP_EOL;
});

$queue->addJob($job);

Nothing has run yet — a worker does that, covered under Workers and daemon mode. $queue->work() runs one job directly, which is enough to try a queue out.

A job needing the application object receives it as the first parameter, ahead of the ones you passed:

use Pop\Queue\Process\Job;
use Pop\Queue\Queue;

$queue = Queue::fake();

$queue->addJob(Job::create(function($app, $to) {
    echo 'Sending to ' . $to . ' via ' . get_class($app) . PHP_EOL;
}, ['nick@test.com']));

$queue->work($application);

It only arrives if something supplies it — work()'s argument here, or the Worker constructor. Without one the callable is invoked with your parameters alone, so a job written to expect it fails with Too few arguments to function ..., 1 passed and exactly 2 expected rather than receiving null.

Every job carries a 40-character id from getJobId() and reports its own state through hasNotRun(), isRunning(), isComplete() and hasFailed().

Retries, Backoff and Timeouts#

Four settings decide what happens when the work does not go smoothly. setMaxAttempts() caps the retries and defaults to 0, meaning retry forever. setBackoff() delays each retry, taking one number of seconds or a per-attempt schedule like [10, 30, 60]. delay() holds a job back before its first run, taking seconds or an absolute time, and setTimeout() interrupts a job that overruns.

use Pop\Queue\Process\Job;

$job = Job::create('App\Service\DigestService->handle', ['nick@test.com']);

$job->setMaxAttempts(3)
    ->setBackoff([10, 30, 60])
    ->setTimeout(30)
    ->delay('2026-12-01 09:00:00');

A job that throws is released for another attempt until its attempts run out, at which point it is buried rather than dropped — moved to the adapter's dead-letter store with the exception message recorded against it.

$queue->adapter()->countDead();     // how many are buried
$queue->adapter()->getDeadJobs();   // the job objects, keyed by id
$queue->adapter()->retryDeadJob($id);
$queue->adapter()->deleteDeadJob($id);

getFailedMessages() on a recovered job holds one message per failed attempt, the last being the reason it was buried. ./kettle queue:jobs prints exactly that — see Kettle.

setTimeout() on a callable job is a soft timeout enforced with a pcntl alarm, so it needs ext-pcntl. On a shell-command job it goes through Symfony\Process, which needs no extension.

Class-Based Jobs and Application Commands#

A closure suits a one-liner. Anything larger belongs in a class, given as a class-string in the same place the closure went:

use Pop\Queue\Process\Job;

// Construct the class and call the method
$job1 = Job::create('App\Service\DigestService->handle', ['nick@test.com']);

// Call a static method
$job2 = Job::create('App\Service\DigestService::handleStatic', ['nick@test.com']);

// Construct the class with the parameters; the object becomes the job's results
$job3 = Job::create('App\Service\DigestService', ['nick@test.com']);

The application is prepended exactly as it is for a closure, so write the method to take it first. What the method returns becomes the job's results:

<?php

namespace App\Service;

class DigestService
{

    public function handle($application, string $to): string
    {
        return 'sent to ' . $to;
    }

}

Two things make this the right default for work of any size. The class needs no route and no registration — being autoloadable in the worker process is enough. And only the class name and the parameters are serialized, so the stored payload stays small: queue identifiers, not objects.

An application command is the other shape. Job::command() takes a command exactly as you would type it, and the application's own router resolves it:

use Pop\Queue\Process\Job;

$job = Job::command('greet Nick');

// An argument containing spaces needs the argv-style array form
$job = Job::command(['greet', 'Nick Sagona']);

Its results are the lines the command printed — output is captured, not echoed, so a worker servicing a command job stays quiet. This is what ./kettle queue:work runs your own commands through; see Kettle and Building Console Applications.

Give a queue's route table a '*' catch-all so a command matching no route reports it — otherwise the job completes with no results.

A shell command is the third: Job::exec(), run through Symfony\Process.

use Pop\Queue\Process\Job;

// String form — goes through a shell, so pipes and redirects work
$job1 = Job::exec('echo hello | tr a-z A-Z');   // results: ['HELLO']

// Array form — no shell at all, so metacharacters are inert
$job2 = Job::exec(['echo', 'hello | tr a-z A-Z']);   // results: ['hello | tr a-z A-Z']

That difference is the security boundary: build the command from anything you do not fully control and it has to be the array form. A command exiting non-zero fails the job with The command "..." failed. and the exit code, then travels the normal retry-then-bury path.

Read getResults(), isComplete() and getFailedMessages() off the job work() returns rather than off the one you added — the adapter stores a serialized copy and hands back a fresh object.

Scheduled Tasks#

A task is a job with a schedule. It's added with addTask(), it stays in the queue after it runs, and it's a worker's run() rather than work() that evaluates it:

use Pop\Queue\Process\Task;
use Pop\Queue\Queue;

$queue = Queue::fake();

$task = Task::create(function() {
    echo 'This is a scheduled task' . PHP_EOL;
})->every30Minutes();

$queue->addTask($task);

$queue->run();

The schedule is a cron expression underneath, and the fluent methods write it: every30Minutes() produces */30 * * * *, dailyAt('09:30') produces 30 9 * * *. The set runs from everySecond() through yearly(), with weekdays(), weekends() and a method per day. schedule() takes a cron string directly when none of them fit:

use Pop\Queue\Process\Task;

$task = Task::create(function() {})->schedule('* */2 1,15 1-4 *');

Pop's format extends cron with an optional leading seconds field, so a six-field string schedules below the minute — every15Seconds() is */15 * * * * *. That's the one thing plain cron cannot express, and the reason to run the scheduler as a daemon rather than from a crontab.

Reach for one scheduling method per task. Each of the coarse ones rewrites the whole expression, so dailyAt('09:30')->weekdays() and weekdays()->dailyAt('09:30') give different results.

runUntil() gives a task an expiry, taking a date string or a timestamp; isExpired() reports on it and isValid() folds in the max-attempts check too:

use Pop\Queue\Process\Task;

$task = Task::create(function() {})
    ->every30Minutes()
    ->runUntil('2027-11-30 23:59:59');

The grace period decides how late a worker may be and still count a task as due. It defaults to -1, which disregards seconds entirely, so a scheduler invoked at any point during 09:30 runs a dailyAt('09:30') task. setGracePeriod(10) narrows that to ten seconds.

A task is claimed in shared storage before it executes, so it runs at most once per window however many times a worker is invoked inside it. The claim is best-effort rather than consensus — workers whose clocks disagree by more than a few seconds can both fire — so make a task idempotent when the side effects matter.

A window that passes without a worker being invoked is simply missed; the grace period widens the window rather than building a backlog.

Adapters#

The adapter is where the queue is stored. Five ship with the component, and the queue takes one in its constructor — new Queue('pop-queue', new Redis()), and so on.

Four of the five implement the whole contract: delay() eligibility, setBackoff() on retry, lease-based crash recovery and a dead-letter store. A reserved job is leased for leaseSeconds, 60 by default, so a job whose worker dies becomes reservable again rather than stranded. Set it above your longest job, or a slow one gets handed to a second worker while the first is still on it.

Memory takes leaseSeconds first and priority second; File, Database and Redis all take priority first.

Each fails differently when its backing store is not there, and the differences matter more than the similarities:

• File throws Pop\Queue\Adapter\Exception from its constructor — Error: The folder '/path' does not exist. It does not create the directory.
• Redis throws RedisException: Connection refused from its constructor, not from the first push.
• Database connects through the pop-db adapter you hand it, so a bad connection has already thrown before the queue adapter exists. It creates its own table on first use.
• Memory cannot fail — and cannot persist. Everything is gone when the process ends.
• Sqs constructs happily against a queue URL that does not exist, and fails on the first push() with an Aws\Sqs\Exception\SqsException. Nothing is checked until then.

Memory is the one to test against — the same lifecycle as the persistent three, with no server, extension or disk. Queue::fake() is the shorthand:

use Pop\Queue\Queue;

$queue = Queue::fake();                      // Queue::create('pop-queue', new Memory())
$queue = Queue::fake('test-queue', 'FILO', 30);

SQS differs in three ways. addTask() is unsupported. Burying deletes the message rather than storing it, so use a native SQS redrive policy for dead jobs. And setBackoff() is ignored, since release() re-sends without recomputing a delay; AWS enforces the initial delay() itself, capped at 900 seconds and ignored on a .fifo queue.

Priority and Signed Payloads#

A queue pops jobs in one of two orders. FIFO, the default, takes the oldest job first; FILO takes the newest.

use Pop\Queue\Adapter\Memory;
use Pop\Queue\Queue;

$queue = new Queue('pop-queue', new Memory(), Queue::FILO);

$queue->setPriority(Queue::FIFO);   // or after the fact
$queue->getPriority();              // 'FIFO'
$queue->isFifo();                   // true

Three jobs added as first, second, third and worked three times come back in that order under FIFO and as third second first under FILO. The constants are those four-letter strings and nothing more, so setPriority('FILO') is identical to setPriority(Queue::FILO); isLilo() and isLifo() are aliases. Every adapter takes a priority in its own constructor too, and setting it on the queue delegates there.

Priority orders jobs within one queue. Which queue a worker reaches for first is a separate setting — see Workers and daemon mode.

Every persistent adapter serializes jobs to store them and unserializes them on the way out, so anything able to write to the underlying storage can plant a payload and have a worker execute it. PayloadSigner closes that. Set a key once at bootstrap, before any queue operation:

use Pop\Queue\Process\PayloadSigner;

PayloadSigner::setKey($_ENV['QUEUE_SIGNING_KEY']);

From then on every adapter HMAC-signs what it writes and verifies the signature before unserializing. A payload that does not verify is skipped, never unserialized and never run.

A queue holding a payload the current key cannot verify reports it as pending and work() leaves it there, so rotate keys with APP_PREVIOUS_KEYS set.

Workers and Daemon Mode#

A worker services one or more queues. Adding a queue with a weight says which matters more:

use Pop\Queue\Worker;

$worker = Worker::create();

$worker->addQueue($reports, 1);
$worker->addQueue($emails, 10);

$worker->work();      // tries emails first, falls through to reports
$worker->workAll();   // one job from every queue, highest weight first

getQueues(), workAll() and runAll() iterate in weight order rather than insertion order, so the pair above comes back as emails, reports. Weight defaults to 0, leaving queues in the order added. getQueue(), hasQueue() and getWeight() read the registration back, and the worker is Countable and iterable.

work() handles jobs and run() handles scheduled tasks; workAll() and runAll() do the same across every queue. From a cron entry once a minute, those four are the whole story. For a long-running process, workLoop() and runLoop() call their non-looping counterparts forever:

#!/usr/bin/env php
<?php
require __DIR__ . '/vendor/autoload.php';

$queue  = new Pop\Queue\Queue('pop-queue', new Pop\Queue\Adapter\File(__DIR__ . '/queue'));
$worker = Pop\Queue\Worker::create($queue);

$worker->workLoop(1);

The argument is how many seconds to sleep after a pass that found nothing anywhere. A pass that found work loops again immediately with no sleep at all.

workLoop() and runLoop() are separate — one works jobs, the other runs scheduled tasks — so a daemon deployment runs two OS processes.

stop() ends either loop and isStopped() reads the flag. With ext-pcntl loaded, both loops install SIGTERM and SIGINT handlers that call stop(), so kill and Ctrl-C take the same graceful path — the loop finishes the job it is on and returns. Without pcntl, only a programmatic stop() ends a loop.

Two things change when you move from cron to a daemon, and both bite.

Set retry configuration on every job under workLoop(). A failed job counts as work found, so the loop keeps going rather than sleeping between attempts.

Run the daemon under a supervisor that restarts it — an exception out of an adapter propagates out of workLoop() and ends the process.

Clearing runs from the worker or the queue: clear() drops completed jobs, clearFailed() empties the dead-letter store and clearTasks() removes scheduled tasks, each with a clearAll*() variant across every registered queue. ./kettle queue:clear is those three behind its flags, and queue:work and queue:scheduler wrap the loops — see Kettle.

Observability#

Two separate things answer "how are my workers doing", and they answer different questions.

Events#

Events tell you about work that ran. A queue fires lifecycle events through Pop\Event\Manager, the same system Pop\Application uses, so a listener has the finished job in hand — duration, throughput and failure rate need no extra machinery:

use Pop\Event\Manager;
use Pop\Queue\Queue;

$events = new Manager();

$events->on('queue.job.post', function($job, $queue) {
    echo 'job ' . $job->getJobId() . ' done in ' . $job->getDuration() . 's' . PHP_EOL;
});

$events->on('queue.job.failed', function($job, $queue, $exception) {
    echo 'job ' . $job->getJobId() . ' failed: ' . $exception->getMessage() . PHP_EOL;
});

$queue = Queue::fake();
$queue->setEvents($events);

Worker fires its own six — worker.work_loop.tick, .idle and .shutdown, and the run_loop equivalents — set the same way with $worker->setEvents(). A queue-level manager wins outright; only when none is set does an Application passed into work() supply its own. getDuration() is whole seconds, so most jobs report 0.

Write queue listeners with positional parameters: Manager::trigger() strips the keys off the params it passes, so an event fired with ['job' => $job, 'queue' => $queue] arrives as function($job, $queue).

The Worker Registry#

The registry tells you a worker is alive. Events cannot report a worker that died, because a dead worker emits nothing. Give a worker a registry and it registers on start, heartbeats each pass, records what it is working on and deregisters on a graceful stop:

use Pop\Queue\Registry\Adapter\Redis as RegistryRedis;
use Pop\Queue\Registry\WorkerRegistry;

$registry = new WorkerRegistry(new RegistryRedis());

$worker->setName('billing-worker-01');
$worker->setRegistry($registry);

$worker->workLoop();

The backend is chosen independently of the queue adapter — Memory, File, Database and Redis all exist under Pop\Queue\Registry\Adapter — so an SQS-backed queue still gets worker visibility. Query it from a status command, a health check or a dashboard:

use Pop\Queue\Registry\Adapter\Redis as RegistryRedis;
use Pop\Queue\Registry\WorkerRegistry;

$registry = new WorkerRegistry(new RegistryRedis());

foreach ($registry->getWorkers() as $record) {
    echo $record->getName() . ' on ' . $record->getHost() . ' (pid ' . $record->getPid() . ') - '
        . $record->getJobsProcessed() . ' done, ' . $record->getJobsFailed() . ' failed' . PHP_EOL;
}

$registry->countWorkers();
$registry->getStaleWorkers();   // the heartbeat has gone quiet
$registry->getStuckWorkers();   // quiet AND holding a job past its own timeout
$registry->prune();             // reap records untouched for an hour

billing-worker-01 on web-01 (pid 329826) - 0 done, 0 failed

PHP is synchronous, so a worker grinding through a long job cannot heartbeat. Hence the two queries: getStaleWorkers() is quiet with no current job, getStuckWorkers() is quiet while holding a job past its own setTimeout(). Alert on the second, and give jobs a timeout so it has a yardstick.

A single work() or workAll() registers on entry and deregisters on exit, so a cron-invoked worker shows in the registry only while it runs.

pop-queue exposes more than this page covers — the full adapter contract for writing your own, and every option on Job, Task and Worker — see the pop-queue README.

See Also#

• Kettle — queue:config, queue:work, queue:scheduler and the rest of the command surface
• Building Console Applications — writing the commands a Job::command() job runs
• Applications & Bootstrap — where PayloadSigner::setKey() and the queue's services belong
• Events — the Pop\Event\Manager the queue's lifecycle events are fired through
• Connecting & Adapters — the pop-db connection the Database adapter takes
• pop-queue README — the full queue API surface