Queues and workers¶

naf/queue separates job submission from execution. An HTTP handler enqueues data and returns; a CLI worker constructs the job and calls execute(). Installing Queue also installs CLI. The default file driver needs writable local storage and a running worker.

flowchart TD
    H["HTTP handler or command"] -- "queue()->push()" --> Q[("Queue storage")]
    Q -- reserve --> W["queue:consume worker"]
    W -- "execute()" --> J["Your job"]
    W -- "attempts exhausted" --> D[("Deadletter")]
    D -- "queue:retry-failed" --> Q

Required packages

In an existing NAF application:

composer require naf/queue

Also installs these transitive dependencies: naf/cli.

A job¶

Start from Your first application, install Queue, and create this file. The example writes a confirmation to worker output without sending mail or modifying accounts.

app/Jobs/RecordSignup.php

<?php

declare(strict_types=1);

namespace App\Jobs;

use Naf\CLI\Core\Output;
use Naf\Queue\Core\QueueJobInterface;

final class RecordSignup implements QueueJobInterface
{
    public function __construct(private array $payload) {}

    public function execute(Output $output): void
    {
        $id = $this->payload['user_id'] ?? null;
        if (!is_int($id) || $id < 1) {
            throw new \InvalidArgumentException('A positive user_id is required.');
        }

        $output->writeLine('Recorded signup for user ' . $id);
    }
}

Jobs implement QueueJobInterface. The worker builds the job with make(RecordSignup::class, $payload): a constructor parameter typed array receives the whole payload (why), and other services in the constructor are injected as usual. Keep the payload serializable and pass identifiers or values rather than entity objects. Throw an exception to report failure. Drivers with deadletter support retain failed work for inspection/retry.

Queueing it¶

Create this CLI script to submit the example job:

bin/enqueue-demo.php

<?php

declare(strict_types=1);

require dirname(__DIR__) . '/bootstrap.php';

use App\Jobs\RecordSignup;
use function Naf\Queue\queue;

queue()->push(RecordSignup::class, ['user_id' => 42]);
echo "Job queued.\n";

Run from the project root:

composer dump-autoload
php bin/enqueue-demo.php
vendor/bin/naf queue:consume --once

Expect Job queued. and then Recorded signup for user 42. The second command exits after one job. Enqueueing alone does not execute work.

Channels¶

Channels separate workloads. Add this fragment to a handler or application service after loading the application:

use App\Jobs\RecordSignup;
use function Naf\Queue\queue;

queue('accounts')->push(RecordSignup::class, ['user_id' => 42]);

Run workers for the channels they should consume:

vendor/bin/naf queue:consume --channel=accounts
vendor/bin/naf queue:consume --channels=default,accounts,thumbnails

The unqualified queue() uses the default channel. A job in another channel waits until a worker consumes that channel. Channel support depends on the selected driver.

Running the worker¶

vendor/bin/naf queue:consume

Option	Effect
`--once`	Consume at most one available job and exit
`--max-jobs=N`	Exit after N processed jobs
`--max-runtime=N`	Exit after N seconds
`--channel=name`	Consume one channel
`--channels=a,b,c`	Consume several channels
`--verbose`, `-v`	Print additional worker output

With --once, a failed job or missing job class produces a nonzero exit status. Use this in CI and operational checks. Supervise persistent workers and bound their lifetime to limit accumulated memory and stale state. Restart workers after deployments.

[program:naf-queue]
directory=/var/www/my-app
command=php /var/www/my-app/vendor/bin/naf queue:consume --channels=default,accounts --max-jobs=500
autostart=true
autorestart=true

Replace the application path and run the supervisor with access to the same configuration and storage as the HTTP application. See Deployment.

One-off background attempts¶

pushAndRun($class, $payload = []) enqueues a job and launches a detached vendor/bin/naf queue:consume --once process. It always launches the unqualified worker; it does not pass the queue's selected channel. A named-channel job can therefore remain queued, and the process can consume an earlier default-channel job instead. Its output is discarded and the caller does not wait for completion. Use supervised channel-aware workers for dependable execution and visible failures.

Jobs that failed¶

The worker defaults to three attempts with a five-second retry delay. Configure queue:max_attempts and queue:retry_delay in app/config.php to change these values. With the file driver, a failed attempt is requeued until the limit, then moved to deadletter storage. --once exits nonzero after one failed attempt even if another attempt remains. Use --verbose to see failure messages in worker output; exhausted failures are also logged.

Inspect the error and correct its cause before retrying exhausted jobs:

vendor/bin/naf queue:retry-failed
vendor/bin/naf queue:retry-failed --keep

--keep retains the original deadletter entry. The command has no channel option and calls retryFailed() on the unqualified queue driver. With the default FileDriver, that retries the default channel only. Channel-specific retries require the driver's retryFailedFrom() API; do not assume an unsupported --channel option selects them.

A driver without deadletter support returns an error from this command. Deadletter inspection and retention are operational responsibilities.

Where the jobs live¶

The default file driver stores jobs in storage/queue/ and failed jobs in storage/queue/deadletter/, with channel-specific subdirectories. Preserve these locations between releases and grant write access to producers and workers. The default bootstrap passes these paths explicitly to FileDriver; bind a different Queue/FileDriver to customize them rather than assuming configuration alone changes that binding.

Local files are not automatically shared across machines. Queue also ships SQLite and PDO drivers. Custom drivers implement QueueDriverInterface; channel, deadletter and lease capabilities use additional interfaces.

Custom file locations¶

For example, keep pending and failed jobs in separate application directories. Add this fragment to root bootstrap.php, after Composer autoloading and before app()->run():

use Naf\Queue\Core\Queue;
use Naf\Queue\Drivers\FileDriver;
use function Naf\app;

$container = app()->container();
$container->set(Queue::class, static fn() => new Queue(new FileDriver(
    BASE_PATH . '/storage/jobs/pending',
    BASE_PATH . '/storage/jobs/failed',
)));

Create both directories with mkdir -p storage/jobs/pending storage/jobs/failed and make them writable by the producer and worker. The driver adds channel subdirectories itself. Repeat the enqueue and consume commands above: jobs now pass through storage/jobs/pending/; exhausted failures go to storage/jobs/failed/. Run every producer, ticker, worker and retry command with this same binding. Register it before any service retains a Queue instance; replacing a binding cannot change an object already held by another service. If you include app/schedule.php from the Scheduling guide, put this binding before that include: registering a schedule resolves the scheduler, which retains its Queue instance.

The database driver, for more than one machine¶

Use PDODriver for a queue shared through a database. This fragment belongs in an explicit installation/migration step, with a configured PDO connection:

use Naf\Queue\Drivers\PDODriver;

(new PDODriver($pdo, 300))->install();

Then add the lazy binding to application bootstrap.php, before consumers resolve Queue. The second constructor argument is the lease duration in seconds:

use Naf\Queue\Core\Queue;
use Naf\Queue\Drivers\PDODriver;
use function Naf\app;

$container = app()->container();
$container->set(Queue::class, static fn() => new Queue(
    new PDODriver($container->get(PDO::class), 300),
));

With Database 0.2.4+, the configured connection is bound under PDO::class. Alternatively, bind your own connection. Do not create queue tables during every request.

A reserved job remains leased until acknowledgement. A crashed worker's lease expires and makes the job available again. Fencing rejects acknowledgements from an obsolete reservation. Supported database locking prevents two active claims on the same unexpired reservation; lease expiry can still cause concurrent execution if the first worker continues running.

Claim outside an application transaction. Enqueueing can participate in a caller transaction, but a job must finish within its lease or renew it. Delivery is at least once: make side effects tolerate a repeated job. The worker acknowledges after success; custom consumers must acknowledge their own reservations.

When a business write and its job must succeed together, bind PDODriver to the same PDO instance used for the write and call queue()->push() before committing that transaction. Rolling back removes both changes. The default file driver cannot participate in a database transaction, and using another PDO connection does not make the writes atomic. Do not hold the transaction open while a worker performs network requests or other slow work.

PDODriver also deduplicates pending jobs when their payload includes _job_id (Queue 0.2.4+):

use App\Jobs\RecordSignup;
use function Naf\Queue\queue;

queue()->push(RecordSignup::class, ['user_id' => 42, '_job_id' => 'signup:42']);

This uses the job above and the PDO binding in this section. The key is scoped to the job class; repeating the same class and key keeps the existing job and its original payload. Choose an ID for one logical operation. A successful acknowledgement deletes the job and frees the key, so deduplication is not a permanent record of completed work. A crash after an external effect can still repeat that effect on retry. Use application-level idempotency for saved replies, payments or other effects that must not be duplicated.

FileDriver uses _job_id as a job filename within its channel. Reusing it can replace the pending file and its payload; it does not keep the original job or scope the key to its class. File keys must pass safePath, so the colon-containing example above is for PDODriver. Do not assume the two drivers have the same duplicate-key semantics.

Set queue:heartbeat_file to an application-specific writable path for a worker heartbeat. Monitor its age together with failures and backlog; it proves polling, not successful jobs.

Persist a job for later¶

Queue 0.2.5+ adds pushAfter() for drivers implementing Naf\Queue\Drivers\DelayedQueueDriverInterface. The PDO driver supports it; the file, legacy SQLite and channel drivers do not advertise this optional capability.

After host boot, with the PDO binding above, delay the RecordSignup job defined in this chapter:

use App\Jobs\RecordSignup;
use function Naf\Queue\queue;

queue()->pushAfter(30, RecordSignup::class, [
    'user_id' => 42,
    '_job_id' => 'signup:42',
]);

pushAfter(int $delaySeconds, string $class, array $payload = []): void persists the job immediately and makes it eligible after the non-negative delay. Zero is valid. Negative delays or timestamp overflow throw InvalidArgumentException. An unsupported driver throws LogicException; it does not execute immediately or sleep in the worker.

DelayedQueueDriverInterface extends QueueDriverInterface with enqueueAt(string $class, array $payload, int $availableAt): void. A non-negative Unix timestamp in the past is immediately eligible. The PDO driver uses its existing available_at column, so no schema migration is needed. Ordinary push() remains immediate.

Delayed enqueue participates in the caller's PDO transaction. Rolling back removes the job. Coalescing retains the first pending job and its original due time; another enqueue with the same class and _job_id does not reschedule it. Other ready work can run while it waits. Leases, acknowledgements, retries and deadletter recovery retain their existing behavior. The due time is the earliest reservation time, not a completion deadline.

Use delayed jobs for individual events and Scheduling for recurring checks. The job must recheck current application state and access when it executes. Queue eligibility does not cancel business work that has become obsolete.

Configuration¶

Key	Type	Default	Effect
`queue:max_attempts`	int	`3`	Attempts per job before the file driver moves it to deadletter storage
`queue:retry_delay`	int	`5`	Seconds to wait before retrying a failed job
`queue:heartbeat_file`	string	—	File the worker updates with a timestamp on every polling pass

If the result is different¶

What you see	What to check
Jobs are queued but never run	A worker consumes the job's channel: `queue:consume` reads `default` unless you pass `--channel` or `--channels`
`Job class App\Jobs\… not found.`	Run `composer dump-autoload`; the class name and namespace match the file
`… does not implement QueueJobInterface.`	The job class implements `Naf\Queue\Core\QueueJobInterface`
A job keeps failing, then disappears	It reached `queue:max_attempts` and moved to deadletter storage; inspect it, fix the cause, run `queue:retry-failed`
`Cannot write queue heartbeat.`	The directory of `queue:heartbeat_file` exists and is writable for the worker
Workers on two machines do not share jobs	The file driver is local; use the database driver