kinetis / queue-sql

v1.3.2 2026-09-13 11:01 UTC

Requires

• php: ^8.4
• kinetis/framework: ^1.8.1
• kinetis/persistence: ^1.2.2
• kinetis/queue: ^1.3.1

Requires (Dev)

• infection/infection: ^0.35.0
• phpstan/phpstan: ^2.2.8
• phpunit/phpunit: ^13.3.3
• vimeo/psalm: ^6.17

MIT

• Alon Noy

Published on Packagist.org on 2026-09-13 11:15 UTC

README

kinetis/queue-sql
A SQL-backed (MySQL/Postgres) queue implementation for kinetis/queue's QueueInterface

Part of Kinetis, a non-blocking PHP framework for API-first applications, developed in the kinetis-dev/kinetis monorepo.

Adds MySQL/Postgres as a queue backend, riding an existing database instead of a separate service. push()/pop()/ack()/release()/fail() work exactly like any other backend — only your configuration changes. pop() relies on SELECT ... FOR UPDATE SKIP LOCKED to guarantee two workers never receive the same job — MySQL 8.0+ or MariaDB 10.6+.

use Kinetis\Config\Config;
use Kinetis\QueueSql\SqlQueueFactory;

$queue = SqlQueueFactory::fromConfig($config);

$queue->push(new SendWelcomeEmail($email, $name), queue: 'default');

The queue needs a table

Two ready-to-copy migration stubs, one per dialect:

vendor/kinetis/queue-sql/resources/migrations/create_kinetis_queue_jobs_table.mysql.php.stub
vendor/kinetis/queue-sql/resources/migrations/create_kinetis_queue_jobs_table.pgsql.php.stub

Copy whichever matches your database into your own migrations/ directory with a timestamp prefix, then run vendor/bin/kinetis migrate.

SqlQueue declares Kinetis\Queue\ClearableQueueInterface. Clearing deletes every row on the queue whose reserved_at is null, and reports how many the DELETE removed. That is narrower than what size() counts: an expired reservation — one older than QUEUE_VISIBILITY_TIMEOUT_SECONDS — counts as waiting and pop() may reclaim it, but clear() still leaves it alone — the worker holding it may simply be slow, and still has a settlement to make.

Every reservation and every timeout reclaim writes a fresh random reserved_token, and ack()/release()/fail() match on the row id and that token. A settlement arriving after another worker reclaimed the row therefore writes nothing and raises Kinetis\Queue\Exception\StaleJobHandleException, which queue:work reports as a lost delivery instead of settling somebody else's. Still keep the visibility timeout comfortably longer than your slowest job: fencing keeps a late settlement from doing damage, it does not stop the job from running twice.

Configuration

QUEUE_CONNECTION=sql
DB_CONNECTION=mysql
DB_HOST=127.0.0.1
DB_NAME=app
DB_USER=app
DB_PASSWORD=secret

DB_* are the exact keys kinetis/persistence already reads. The one key this package introduces itself:

Key	Default	Purpose
QUEUE_VISIBILITY_TIMEOUT_SECONDS	300	Seconds before a crashed worker's reserved job becomes poppable again. Must be a positive integer.

Both are scoped by QUEUE_CONNECTION_NAME the same way every other backend's keys are. kinetis/queue's own keys (QUEUE_CONNECTION, QUEUE_MAX_ATTEMPTS, ...) are documented in that package; full reference: kinetis.dev/docs/config.html.

Installation

composer require kinetis/queue-sql

Requires PHP 8.4+, kinetis/framework, kinetis/queue, and kinetis/persistence. Full documentation: kinetis.dev/docs/queue-sql.html.

License

MIT — see LICENSE.