[13.x] Add queue concurrency driver

[mathiasgrimm]

Summary

This PR adds a queue driver to the Concurrency component. It is the same blocking Concurrency::run() API, but the closures run on your queue workers instead of local processes, and the results come back to the caller:

$results = Concurrency::driver('queue')->run([
    'thumbnail' => fn () => ProcessImage::thumbnail($path),
    'preview' => fn () => ProcessImage::preview($path),
    'optimized' => fn () => ProcessImage::optimize($path),
]);

Same contract as the other drivers: it blocks, keys and order are preserved, each task returns its value, and when a task throws, the exception is rebuilt and rethrown in the caller.

You can also pick the connection, queue or result store at runtime:

Concurrency::driver('queue')
    ->onConnection('redis')
    ->onQueue('images')
    ->run([...]);

It only needs what every Laravel app already has, a queue and a cache. No new tables, no migrations, no new services.

Why

The process and fork drivers can only use the machine they run on, so the calling instance has to be big enough for the work it spawns. With the queue driver a small (RAM/CPU) web instance does not need to resize images or build exports by spawning PHP processes on itself. The web tier stays small and the heavy work runs on workers sized for it, in parallel, so the response time is close to the slowest task instead of the sum of all of them.

This combines very well with Laravel Cloud managed queues. Flex workers scale to zero while the queue is idle and wake in under a second when jobs arrive, and they are billed per second only while running. So you get real cost savings (no oversized web instances, no always-on workers) together with a performance win: a burst of tasks is absorbed by workers that exist only for the seconds they are needed.

It also makes it possible to keep some endpoints synchronous where today you would build an async flow (submit, poll a status endpoint, wait for a webhook) just because the work is too heavy for the web node. With sub-second worker wake up, the endpoint can dispatch, block for the result and return it in the same response. The wait is bounded (60 seconds by default), so work the client should not wait for still belongs in a normal job.

Demo

There is a demo application at https://github.com/mathiasgrimm/concurrency-demo with a browser UI for every scenario: the parallel run, the sync contrast, exception propagation, timeouts, defer(), and a benchmark of the same three tasks on the sync, process, and queue drivers. Each demo shows the route code and the JSON response side by side, and the readme covers running it locally and deploying it to Laravel Cloud with managed queues.

Notes

• Each task is one queued job; the result (or the exception) travels back through a shared cache store, reusing the process driver's result envelope. Task failures also stay visible in failed_jobs / Horizon.
• No breaking changes: the default driver is still process and the existing concurrency.driver config keeps working through a fallback. The new concurrency.drivers map also allows named instances like Concurrency::driver('reports').
• Job batches are not used because they cannot carry return values back to a blocking caller.
• Do not call run() from a worker that consumes the same queue, it can starve until the timeout. Use a dedicated queue or spare capacity.
• Covered by unit and integration tests, including a real database queue worker round trip; all existing Concurrency tests pass unchanged.

Docs PR to follow if there is interest. Happy to adjust naming or the config shape.

---

Failover support and safer retries (added after the first review)

These later commits make the driver behave correctly when the queue connection is a failover connection, and when the cache uses a custom store. That is more than the title promises, so I am happy to split it into a separate PR if you would rather review it on its own.

Some background. A failover connection tries each of its connections in turn and treats any error from a connection as that connection being down. The sync connection runs the job right away, inside that call. So if a task threw an error while running on a sync fallback, the failover connection thought the connection was down and ran the task again on the next one. With the primary connection down and a task that throws, this is what happened before the change:

connections	task ran	the caller got
[redis, sync]	once	the wrong exception type
[redis, sync, sync]	twice	the task ran again on the second sync
[redis, sync, database]	once, plus a leftover job on the database queue	the task's exception
[redis, sync, dead]	once	the connection error, not the task's own exception
a connection that lists itself	crash	the process ran out of memory

What changed:

• When a job runs on the sync connection and the task throws, the job now reports the error and returns instead of throwing. The failover connection no longer mistakes the task's failure for the connection being down.
• A job checks whether its result is already stored and returns early if it is, so the same task is never run twice.
• After a run finishes, its cancel flag is kept in the cache for a short while, so a job delivered again after the caller already has its answer skips itself instead of running.
• Whether the tasks run in the current request is decided by looking at every connection in a failover list. A list that includes a connection whose jobs would never run (null, deferred, background), an empty list, and a list that points back to itself are all rejected before any job is sent. Failover cache stores are checked the same way.
• defer() now sends its own job class, which follows the same sync rule and is otherwise a normal queued closure.

Separately, QueueDriver::wait() was calling many() on the cache repository, but that method is only on the concrete cache class, not on the Repository contract the method is typed against. A custom cache repository that implements only the contract made the driver fail. It now uses getMultiple(), which is on the contract.

Things to know

• Two existing tests change on purpose. The cancel flag now stays in the cache after a run (that is the flag a redelivered job checks), and defer() sends InvokeDeferredClosure rather than CallQueuedClosure directly (it still is a CallQueuedClosure).
• Code listening to the RetrievingManyKeys cache event sees the keys the way getMultiple() reports them, not the way many() did.
• A task that fails on the sync connection through defer() is now reported rather than marked as a failed job. On a real queue the worker still records the failure.
• When a failover connection falls through to sync, the tasks run one after another in the current request instead of in parallel. That is what failover is for, and the driver cannot tell in advance which connection will take the job.
• Failover decides where a job is sent, not where it runs later, so a task that ends up on database still needs a worker running on database.
• With a failover cache store, a result written to the backup store while the main store is down cannot be read once the main store comes back, so the run times out. Point the driver at a single shared store to avoid this.
• The skip checks are best effort, not a guarantee. A cache that can return stale reads, or two workers picking up the same job at once, can still run a task twice. As with any queue, a task that must run once should be safe to run twice.

Every check has a test that fails when the check is removed.

🤖 Generated with Claude Code