Queue

Note

Not part of core. Install it separately:

composer require kinetis/queue

A backend-agnostic background job queue: push a job from application code, a separate kinetis queue:work worker process pops and runs it. This package carries only the Kinetis\Queue\QueueInterface contract, the worker, and the CLI commands — application code never sees which backend is running underneath, and every backend lives in its own separate package: Redis (Queue (Redis)), SQL/MySQL/Postgres (Queue (SQL)), Amazon SQS (Queue (SQS)), and RabbitMQ (Queue (RabbitMQ)).

push() and pop() never block the worker while waiting on the backend, so a push can run alongside other work through Concurrency’s concurrently() instead of stalling the request until it completes.

Writing a job

A job is a plain class implementing Kinetis\Queue\Job, constructed the same way as any other DTO in Kinetis:

use Kinetis\Queue\Job;

final readonly class SendWelcomeEmail implements Job
{
    public function __construct(
        public string $email,
        public string $name,
    ) {}

    public function handle(Mailer $mailer): void
    {
        $mailer->send($this->email, "Welcome, {$this->name}!");
    }
}

handle()’s own parameters are resolved through the container at run time, the same way a controller method’s parameters are — not the constructor. A job’s constructor only ever holds the data it needs to survive being written to the queue and read back later by a worker process, which may not be the same process (or even the same machine) that pushed it.

Pushing and processing

use Kinetis\Queue\QueueInterface;

final readonly class RegistrationController
{
    public function __construct(private QueueInterface $queue) {}

    #[Post('/register')]
    public function store(#[Body] RegisterRequest $data): array
    {
        // ...
        $this->queue->push(new SendWelcomeEmail($data->email, $data->name));

        return ['status' => 'registered'];
    }
}

vendor/bin/kinetis queue:work

The worker runs one job at a time: pop, resolve handle()’s dependencies through a fresh container scope, invoke it, then pop the next one. It keeps going until a shutdown signal arrives — see “Stopping a worker” below.

Named and prioritized queues

Every push() targets a named queue — default when none is given:

$this->queue->push(new SendWelcomeEmail($data->email, $data->name));
$this->queue->push(new GenerateReport($reportId), queue: 'reports');
$this->queue->push(new SendPasswordReset($data->email), queue: 'high');

A worker watches one or more queue names, in priority order:

vendor/bin/kinetis queue:work --queue=high,default

Priority is expressed by list order, not a numeric per-job score: the worker above drains everything on high before ever checking default. A queue name absent from --queue is invisible to that worker entirely — jobs pushed to reports sit untouched until some worker actually watches reports. Omitting --queue watches default only.

Choosing a backend

QUEUE_CONNECTION has no default — set it explicitly, and install the matching package:

QUEUE_CONNECTION	Package	Configuration
redis	kinetis/queue-redis	Queue (Redis)
sql	kinetis/queue-sql	Queue (SQL)
sqs	kinetis/queue-sqs	Queue (SQS)
rabbitmq	kinetis/queue-rabbitmq	Queue (RabbitMQ)

QUEUE_CONNECTION=redis
REDIS_HOST=127.0.0.1

Picking a QUEUE_CONNECTION value without its package installed fails clearly, naming the package to install — see Kinetis\Queue\Exception\QueueUnavailableException.

Setting QUEUE_CONNECTION is also all the container wiring there is: this package’s bootstrap class (declared via extra.kinetis, see CLI) binds QueueInterface to the selected backend before AppScope::boot() locks bindings, so the controller above constructor-injects it with no bootstrap.php code. An application registering its own QueueInterface binding in bootstrap.php wins over the package’s — see “Multiple backends” below for when you’d want to.

QUEUE_CONNECTION_NAME picks which named connection of that backend this worker uses (see Configuration’s named-connection convention) — ‘default’ when unset:

QUEUE_CONNECTION=redis
QUEUE_CONNECTION_NAME=cache2
REDIS_CACHE2_HOST=127.0.0.1

Scaling out: multiple workers

kinetis queue:work is safe to run as any number of separate, concurrent processes against the same backend — a plain horizontal-scaling lever, not something that needs coordinating by hand. Every backend guarantees a job is handed to exactly one worker, so two workers running at once never both pick up the same job. Workers don’t need to agree on anything beyond which queue names they watch — start as many as you want, on as many machines as you want, pointed at the same backend.

Stopping a worker: deploys and restarts

SIGTERM and SIGINT stop the loop after the job in flight finishes. The worker acks (or releases) that job, returns, and exits 0 — nothing is left half-done, and nothing stays reserved for the backend to reclaim later.

That is exactly the signal Docker, systemd, and Kubernetes already send before they resort to SIGKILL, so a rolling deploy, docker compose restart, or systemctl restart is graceful with no extra command and no coordination between workers. Give the supervisor a grace period at least as long as your slowest job: Docker’s default is 10 seconds (--stop-timeout, or stop_grace_period in Compose), Kubernetes’ terminationGracePeriodSeconds is 30.

Warning

Graceful shutdown needs ext-pcntl, which is a CLI-only extension and is not loaded in the official PHP Docker images — add docker-php-ext-install pcntl to your image. Without it there is no way to observe SIGTERM at all, and a deploy interrupts whatever job is running. queue:work prints a warning at startup when it is missing, rather than leaving you to discover it during a deploy.

Inspecting and clearing a queue

queue:stats reports how many jobs are waiting:

vendor/bin/kinetis queue:stats --queue=high,default

QUEUE    WAITING
high     12
default  3
----------------
total    15

The count covers jobs waiting to be popped, including ones still inside their push() delay. Jobs a worker currently holds are excluded — those are being worked, not waiting. Amazon SQS reports these numbers as estimates rather than exact figures (see Queue (SQS)), which is fine for the question this answers: whether a queue is draining or backing up.

queue:clear discards waiting jobs, and requires --force because there is no dead-letter copy to restore from:

vendor/bin/kinetis queue:clear --queue=default --force

Jobs a worker has already reserved are untouched — they belong to that worker until it finishes with them.

Multiple backends

Different queues can live on different backends — a RedisQueue (from Queue (Redis)) for low-latency jobs, a SqlQueue (from Queue (SQL)) for jobs that should ride along with an existing database’s backups and transactions, an SqsQueue (from Queue (SQS)) for jobs a separate AWS account or service needs to consume, or a RabbitMqQueue (from Queue (RabbitMQ)) for jobs routed through an existing broker. Register each concrete class directly rather than binding QueueInterface to just one of them:

use Kinetis\QueueRedis\RedisQueue;
use Kinetis\QueueSql\SqlQueue;

$app->instance(RedisQueue::class, new RedisQueue($redisClient));
$app->instance(SqlQueue::class, new SqlQueue($db));

Application code that pushes to a specific backend constructor-injects that concrete class instead of the shared interface:

final readonly class RegistrationController
{
    public function __construct(private RedisQueue $fastQueue) {}
}

Each backend gets its own kinetis queue:work process, run with the QUEUE_CONNECTION/backend-specific environment variables and --queue flag matching the queue names pushed to it — one process per backend-and-queue-set combination, same as scaling a single backend out.

Running synchronously in development: SyncQueue

use Kinetis\Queue\SyncQueue;

if ($appEnv->isDevelopment()) {
    $app->instance(QueueInterface::class, new SyncQueue($app));
}

push() runs the job’s handle() immediately, inline, rather than storing it anywhere — no separate kinetis queue:work process needed while developing locally. Every push() call still gets its own fresh scope, exactly like a real worker gives each job — job code that happens to depend on request-scoped state reachable only by accident would fail the same way here as it would against a real backend, rather than silently working in development and breaking once actually queued. Unlike a real worker, a failing job’s exception isn’t caught — it propagates straight to whatever called push(), so the point of running jobs synchronously (seeing the real error immediately) still holds.

Not selectable via QUEUE_CONNECTION — there’s nothing for a worker process to do against a backend that never stores anything, so SyncQueue is constructed directly in application bootstrap code instead. It accepts a queue argument on push() for signature compatibility with the other backends, but ignores it — there’s only ever one “queue” (immediate execution), so a queue name has nothing to select between.

Delayed jobs

$this->queue->push(new SendReminderEmail($userId), delaySeconds: 3600);

A delayed job isn’t visible to pop() until its delay has elapsed. The Redis and SQL backends check this on their own polling cycle rather than firing at the exact moment the delay ends, so a delayed job can run slightly later than its exact target time — typically by a few seconds, not less. SQS’s own delay is native (SendMessage’s own DelaySeconds, no polling-based promotion at all) but capped at 900 seconds (15 minutes) — see Queue (SQS). RabbitMQ’s delay is also broker-driven, with no such cap — see Queue (RabbitMQ).

A failing job

A job whose handle() throws is logged through whatever Psr\Log\LoggerInterface is registered — it doesn’t stop the worker, which moves on to the next job either way. What happens to the job itself depends on maxAttempts:

$this->queue->push(new SendWelcomeEmail($data->email, $data->name), maxAttempts: 3);

maxAttempts is set per job at push() time, and takes priority whenever it’s set. Once a job’s attempt count reaches it, that attempt’s failure is final: the job is removed instead of being retried again.

A job pushed without maxAttempts falls back to kinetis queue:work’s own default instead, QUEUE_MAX_ATTEMPTS in .env — 0 when unset, meaning no retries at all: a job that fails once is given up on immediately.

QUEUE_MAX_ATTEMPTS=5

Raise it to allow retries. There is no configuration, anywhere, that makes a job retry forever — QueueWorker’s own default is likewise 0, not unlimited, so a job with no maxAttempts of its own is only ever retried if something has explicitly set a cap above 1.

{
    "level": "error",
    "message": "Job \"App\\SendWelcomeEmail\" failed permanently after 3 attempt(s): Connection refused",
    "context": {
        "job": {
            "class": "App\\SendWelcomeEmail",
            "queue": "default",
            "attempts": 3,
            "args": {"email": "a@b.com", "name": "Ana"}
        },
        "exception": "..."
    }
}

That log entry — job class, constructor arguments, and the exception — is the only record kept of a job that’s given up on; there’s no dead-letter table or queue to inspect afterward.

A job that still has attempts remaining logs the same way but without args, and without “permanently” in the message, and is released rather than removed. Its payload is still held by the backend at that point, so copying the arguments into the log would add nothing you couldn’t already recover.

Keeping sensitive arguments out of the log

A job routinely carries a token, an email address, or customer data that has no business in a log aggregator. Mark those constructor parameters Kinetis\Queue\Attributes\Sensitive and they are written as [redacted] in the entry above, while everything else is logged as-is:

use Kinetis\Queue\Attributes\Sensitive;
use Kinetis\Queue\Job;

final readonly class SendPasswordReset implements Job
{
    public function __construct(
        public int $userId,
        #[Sensitive]
        public string $email,
        #[Sensitive]
        public string $resetToken,
    ) {}

    public function handle(Mailer $mailer, UrlSigner $signer): void
    {
        $mailer->send($this->email, $signer->resetLink($this->resetToken));
    }
}

"args": {"userId": 4812, "email": "[redacted]", "resetToken": "[redacted]"}

Leaving userId unmarked is the point: the record stays actionable — you can find the account and trigger a new reset — without the address or the token reaching the log. Marking a parameter holding an array or an object redacts that value whole; there is no per-element redaction within one.

This governs what is logged, not what is stored. The real values still travel to the queue backend, because the worker needs them to run the job — so a backend holding sensitive payloads wants the same access control as the database.

Deferring an event listener onto the queue

Kinetis\Queue\QueuedListenerInvoker implements core’s Kinetis\Events\ListenerInvokerInterface — a listener marked Kinetis\Events\ShouldQueue runs as a real queued job instead of inline:

use Kinetis\Events\ListenerInvokerInterface;
use Kinetis\Queue\QueuedListenerInvoker;

$app->instance(ListenerInvokerInterface::class, new QueuedListenerInvoker($queue));

See Events for writing the listener itself.

See also

• Queue (Redis) — the Redis backend, in its own package.

• Queue (SQL) — the MySQL/Postgres backend, in its own package.

• Queue (SQS) — the Amazon SQS backend, in its own package.

• Queue (RabbitMQ) — the RabbitMQ backend, in its own package.

• Persistence — connecting to MySQL, Postgres, and Redis directly.

• Migrations — running the SQL backend’s own required migration.

• Events — writing an event listener that can be deferred onto the queue with ShouldQueue.