Driver-based queue: Redis and memory drivers, priority, unique jobs, backoff, leases and crash recovery

[techmahedy]

Summary

The queue used to be hardwired to two database models. This PR puts a QueueDriver contract between the queue and its storage, ships database, redis and memory drivers on it, and adds the features that a driver-based design makes possible: job priority, unique jobs, retry backoff, leases with automatic crash recovery, multiple connections, bulk dispatch and queue statistics.

Everything new is additive. Existing jobs, #[Queueable], chains and queue:run keep working. The few behavior changes are listed under Breaking changes.

Why

• Only the database was supported, and pop() was not atomic: two workers could claim the same job.
• A job held by a worker that died stayed reserved forever.
• A failed database call was swallowed and reported as "queue is empty".
• There was no way to run urgent jobs first, avoid duplicate jobs, or wait longer after each retry.

What changed

Architecture

• Contracts/QueueDriver defines the guarantees every backend must give: atomic claim, priority then oldest first, leases, fencing, unique keys, plus the failed-job store.
• Drivers/DatabaseDriver, RedisDriver (Lua scripts, needs predis/predis), and MemoryDriver (for tests).
• QueueManager now resolves connections from runtime/config/queue.php (config/queue.php in the package, merged by the launcher). Without a config file it uses the database connection, as before. Queue::extend() registers custom drivers.
• New value objects: Support\Envelope, Support\ReservedJob, Support\FailedJobRecord.
• QueueLauncher registers the manager with a factory (the container cannot autowire its optional constructor arguments) and aliases queue.worker to it. QueueManager::class is also ghost-loadable.

Features

• Connections: onConnection(), $connection, #[Queueable(onConnection:)], Queue::connection('redis'), and --connection on queue:run, queue:failed, queue:retry, queue:flush, queue:monitor.
• Priority: withPriority(), $priority, #[Queueable(priority:)], range -100 to 100. queue:run --queue=high,default,low drains queues in order.
• Unique jobs: override uniqueId(). A duplicate of a waiting or running job is refused (push() returns null). Enforced by the backend, so it holds across workers.
• Retry backoff: $backoff / #[Queueable(backoff:)], an int or a list indexed by attempt (last value repeats). Falls back to $retryAfter.
• Leases and crash recovery: a claimed job is invisible until its lease (lease, default 90s) expires, then another worker takes it with attempts kept. A job whose attempts are already used up is failed with MaxAttemptsExceededException instead of running again.
• Fencing: delete, release, extend and fail only succeed for the current claim (the attempts value is the token), so a worker that lost its lease cannot remove or fail the job another worker now owns.
• Automatic lease renewal for jobs that run in the timeout child process, every 20s.
• Bulk dispatch: Queue::pushMany(). Stats: Queue::stats() (ready, delayed, reserved) and a new queue:monitor table.
• Queue::retryFailed(), Queue::extendLease(), QueueWorker::runNextJob(), setConnection(), setLeaseRenewInterval().

Database schema

The create migration now includes priority, lease_expires_at and unique_key (unique index). There is intentionally no upgrade migration: 4.1.0 is released as a separate version. Existing installs that already created queue_jobs need those three columns added; the docs include a ready-made migration.

Breaking changes

Before	Now
Queue::pop() returned the QueueJob model	Returns ReservedJob ($queue, $payload, $attempts). No model methods; use Queue::delete/release/markAsFailed.
pop(), delete(), release(), markAsFailed() swallowed backend errors (null / false)	They throw. The worker logs and continues.
push() / dispatch() always returned a job id	Return null when a unique job is refused.
queue:monitor showed Pending / Processing	Shows Ready / Delayed / Processing.
A job that finished but could not be removed was treated as a failure	The worker logs it and skips follow-up work (next chain job, callback); the run that finally removes the job does that, so a chain is never advanced twice.

QueueJob and FailedJob models are still shipped for querying the tables, but the queue no longer uses them. Jobs that only implement JobInterface keep working (priority(), connection(), uniqueId() and backoff() are optional).

Bugs fixed on the way

• Two workers could claim the same database job. Claiming is now an atomic compare-and-swap.
• Jobs reserved by a worker that died stayed reserved forever.
• InteractsWithModelSerialization serialized static properties into the payload and wrote them back over the live value on unserialize.
• unserializeJob() emitted PHP warnings on a corrupt payload and accepted payloads that were not jobs; it now throws QueueException.
• Database driver on real engines (found only by running MySQL and PostgreSQL):
  • delayed is a reserved word in MySQL, which broke the stats query.
  • Integers are bound as integers (SQLite orders any integer before any string, so a live lease looked expired).
  • Payloads that are invalid UTF-8, or contain NUL on PostgreSQL, are stored base64-encoded behind a base64: marker; ordinary payloads stay plain text.
  • Exception text is sanitized (NUL becomes \0, invalid bytes become ?), so recording a failure can never fail.
• queue:flush / queue:retry take --id=; the docs showed a positional id.

Testing

503 tests, PHPStan level 8 clean.

• Driver contract (tests/Contract/QueueDriverContract): one suite run unchanged against Memory, SQLite, Redis, MySQL and PostgreSQL. Time is injected, so nothing sleeps.
• Real processes: ConcurrentClaimTest races 6 worker processes over 600 jobs on SQLite, Redis, MySQL and PostgreSQL and asserts each job is claimed exactly once; a second case kills a worker with SIGKILL and asserts the job returns after its lease. LeaseRenewalTest runs a slow job through the real timeout fork with a rival worker trying to steal it, plus a negative control with renewal off.
• Mutation check: with the compare-and-swap guard disabled, the race test failed (625 claims for 600 jobs).
• Manager, worker (retry, backoff, crash loops, lost leases, chains, unique jobs), commands, and serialization regression tests.
• The existing 77 tests still pass.
• CI: tests.yml now starts MySQL 8, PostgreSQL 16 and Redis 7 and sets QUEUE_TEST_REQUIRE_BACKENDS=1, so a backend that is down fails the run instead of silently skipping its tests. Without that variable the suite still runs anywhere, skipping what it cannot reach.
• The MySQL and PostgreSQL suites DROP and recreate the queue tables, so they refuse to run unless the database or schema name contains "test" or "scratch".

Also verified by hand in a real application: dispatch to Redis and drain with queue:run --connection=redis (priority order respected), and each queue command against the memory connection.