README

A queue driver for Laravel's Concurrency component. The same blocking Concurrency::run(), but your closures run on queue workers.

Note

This is an independent, community package. It is not an official or first-party Laravel package, and is not affiliated with, endorsed by, or sponsored by Laravel or Laravel Cloud. "Laravel" is a trademark of its respective owner.

Laravel's Concurrency component runs closures in parallel and hands you back their results. Its process and fork drivers can only use the machine they are called on, so the calling instance has to be big enough for everything it spawns: a small web node that fans out three image transforms is suddenly running four PHP processes.

Laravel Queue Concurrency adds a queue driver with the same contract. The closures are dispatched as queued jobs, your workers execute them, and the results travel back to the blocked caller through a shared cache store:

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

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. It needs nothing your app does not already have: a queue and a cache. No new tables, no migrations, no new services.

It pairs particularly well with scale-to-zero workers, such as Laravel Cloud managed queues. Workers that wake in under a second and bill per second mean a burst of tasks is absorbed by compute that exists only for the seconds it is needed. It also makes it possible to keep an endpoint synchronous where today you would build a submit-poll-webhook flow just because the work is too heavy for the web node.

Features

The same contract as every other driver. It blocks, keys and order are preserved, each task returns its value, and a task that throws has its exception rebuilt and rethrown in the caller.
Runs on your existing queue and cache. No tables, no migrations, no extra services.
Runtime targeting. Pick the connection, queue or result store per call with onConnection(), onQueue() and store().
Bounded waits. A configurable timeout, after which the caller gets a TaskTimedOutException naming exactly how many results arrived.
Cooperative cancellation. A timed out run writes a cancellation flag, so a worker that picks the jobs up afterwards refuses to run them.
Failures stay visible. A failing task still lands in failed_jobs and Horizon, while the caller receives the original exception type.

Requirements

PHP 8.2+
Laravel 12 or 13

Installation

composer require mathiasgrimm/laravel-queue-concurrency

The driver registers itself. Nothing else is required if your defaults already point at a real queue and a shared cache:

QUEUE_CONNECTION=redis
CACHE_STORE=redis

To make it the application-wide default for a plain Concurrency::run():

CONCURRENCY_DRIVER=queue

Publish the config only if you want to change something in it:

php artisan vendor:publish --tag=queue-concurrency-config

Usage

Resolve the driver by name and call it exactly like any other:

use Illuminate\Support\Facades\Concurrency;

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

// ['thumbnail' => ..., 'preview' => ...]

A per-call timeout, as seconds or a CarbonInterval:

Concurrency::driver('queue')->run([...], timeout: 15);

Targeting at runtime. Each of these returns a new driver instance, so the one held by the manager is never mutated:

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

Fire and forget, which returns immediately and dispatches after the response is sent. No results are collected:

Concurrency::driver('queue')->defer([
    fn () => Report::rebuild(),
    fn () => Cache::forget('dashboard'),
]);

Handling a timeout:

use MathiasGrimm\QueueConcurrency\TaskTimedOutException;

try {
    $results = Concurrency::driver('queue')->run([...], timeout: 10);
} catch (TaskTimedOutException $e) {
    // $e->received, $e->total, $e->seconds, $e->connection, $e->queue, $e->store
}

Configuration

config/queue-concurrency.php:

Option	Env	Default	Description
`connection`	`CONCURRENCY_QUEUE_CONNECTION`	`queue.default`	The queue connection tasks are dispatched to.
`queue`	`CONCURRENCY_QUEUE`	the connection's own default	The queue tasks are pushed onto.
`store`	`CONCURRENCY_CACHE_STORE`	`cache.default`	The cache store results travel back through.
`timeout`	`CONCURRENCY_TIMEOUT`	`60`	Seconds the caller blocks before throwing.
`ttl`		`timeout + 60`	Seconds a result envelope is kept. Never lower than the timeout plus 60 seconds of grace.
`poll`		`100`	Milliseconds between checks for results. The final wait is clamped to the remaining budget.

Named instances

Give a workload its own settings and resolve it by name:

// config/queue-concurrency.php
'instances' => [
    'reports' => ['queue' => 'reports', 'timeout' => 120],
],

Concurrency::driver('reports')->run([...]);

The concurrency.drivers.<name> shape proposed in the framework pull request is read too, so a config written for the merged version keeps working:

// config/concurrency.php
'drivers' => [
    'reports' => ['driver' => 'queue', 'queue' => 'reports', 'timeout' => 120],
],

Two rules the concurrency manager's own routing imposes, both enforced with a clear exception rather than a silently wrong queue:

Define an instance in one place, with at least one option. Laravel's manager also reads a legacy concurrency.driver.<name> array. When it finds one it hands the driver that entry and nothing else, so options for the same name kept under queue-concurrency.instances or concurrency.drivers could never reach it. The package refuses that split. An entry that declares only 'driver' => 'queue' is refused too, because it is indistinguishable from the default instance. Keep every option for the instance in the entry the manager reads, or remove that entry.
Instances are wired up the first time the concurrency manager is used. Config files and AppServiceProvider::boot() are both early enough. An instance added to config after something has already called Concurrency::driver() is not picked up until the next request.
Do not name an instance process, sync or fork. Custom creators win over the manager's built in drivers, so an instance called process would quietly turn the framework's default driver, and every plain Concurrency::run(), queue backed. Those names are refused.

Things to know

The cache store must be shared between the caller and the workers. That is how results get home. array, null, session, octane and apc are rejected outright for asynchronous connections, with an error saying so. Use redis, memcached or database.
The queue connection must actually be consumed. null, deferred and background connections are rejected, because their jobs would never run while the caller waits.
Do not call run() from a worker consuming the same queue. It can starve until the timeout. Give the tasks a dedicated queue, or spare capacity.
The wait is bounded, and that is the point. Work the client should not wait for still belongs in an ordinary queued job.
Task closures are serialized. Keep them small and avoid capturing large objects. Two traps in particular: do not define a task inside an arrow function, which captures its enclosing scope by value and can blow the stack during serialization, and put each task on its own source line, because closures with the same signature on one line cannot be told apart and the later ones get the first one's body.
Failures are reported twice, deliberately. The caller gets the original exception rethrown, and the worker still records a failed job, so nothing disappears from failed_jobs or Horizon.
Tasks never see uncommitted data. The jobs are dispatched immediately, ignoring a connection's after_commit setting, because the caller blocks on them and a job held until commit would never run before the timeout. The consequence: run() inside DB::transaction() hands the workers a database that does not yet contain the rows you just wrote. Commit first, then fan out. defer() is the opposite: its deferred job does honour after_commit.
The sync connection is supported and runs inline. It is useful for tests and local work, but the tasks run one after another, so there is no parallelism to gain.
Failover chains are supported, with one honest caveat. A failover connection such as ['redis', 'sync'] works: while redis is up the tasks run on workers, and if the chain falls through to sync they run right there in the request, one after another and no longer bounded by the timeout. That is what failover is for, and the driver cannot warn you it happened. A task that fell to a real queue such as database needs a worker on that connection too, since failover is push-only. Chains containing a connection that would never run the tasks (null, deferred, background), chains that refer back to themselves, and failover cache stores whose fallback is not shared are all refused up front. A task's failure on a synchronous link is reported and enveloped rather than rethrown, so it never reads as a dead link and never runs again on the next one; defer() uses the same rule.
A failover cache store can hide a result during an outage. If the main store is down and a result is written to a backup store, that result cannot be read once the main store comes back, and the run times out. Point the driver at a single shared store rather than a failover store to avoid it.
A finished run leaves its cancellation flag behind for the result lifetime, so a job redelivered after the caller was answered refuses to run instead of running the task a second time. Exactly-once still needs idempotent tasks: two workers racing the same redelivered job can both get past that check.

Relationship to laravel/framework#61273

This package is the code from laravel/framework#61273, packaged so it can be used before that pull request is reviewed and merged. The result envelope and both exception classes are carried over unchanged apart from their namespace. The driver and the queued job started that way and are now a little ahead of the pull request, in changes that are proposed upstream: they handle failover chains correctly (a task failing on a synchronous link is not a dead link, a chain is validated link by link, a finished run leaves a cancel flag so a redelivered job skips itself, a job whose result already exists does not run again), defer() dispatches the package's own job for the same reason, and results are read through the cache contract's getMultiple() rather than the concrete repository's many(). Everything else the pull request's test suite pins is preserved and covered here.

No framework patch is needed. ConcurrencyManager extends MultipleInstanceManager, which already accepts custom driver creators, so the service provider registers the driver through extend() on a stock Laravel.

The config keys, the environment variable names and the driver name are all identical to the pull request, so migrating once it lands is a config move and an import change:

- use MathiasGrimm\QueueConcurrency\TaskTimedOutException;
+ use Illuminate\Concurrency\TaskTimedOutException;

The package steps aside on its own: if Illuminate\Concurrency\QueueDriver ever exists, the service provider registers nothing and the first-party driver wins for Concurrency::driver('queue'). Named instances declared under queue-concurrency.instances are the exception: nothing else knows about them, so they stop resolving at that point. Move them to concurrency.drivers (which the merged framework reads) before upgrading, then remove the package.

Testing

composer test
composer analyse
composer lint

There is also a throwaway application under workbench/ for driving the driver by hand against genuinely separate worker processes, which no in-process test can do. Build it once:

vendor/bin/testbench workbench:build
# SQLite serialises writers, so let the workers share the queue table.
php -r 'file_exists($f = "workbench/database/database.sqlite") && (new PDO("sqlite:$f"))->exec("PRAGMA journal_mode=WAL");'

testbench serve and testbench queue:work boot separate processes that read their own .env rather than testbench.yaml, so pass the settings through the environment. In one shell:

export DB_CONNECTION=sqlite QUEUE_CONNECTION=database CACHE_STORE=file
export DB_DATABASE="$PWD/workbench/database/database.sqlite"
vendor/bin/testbench serve

And a few workers in another, so there is something to parallelise across:

export DB_CONNECTION=sqlite QUEUE_CONNECTION=database CACHE_STORE=file
export DB_DATABASE="$PWD/workbench/database/database.sqlite"
for i in 1 2 3; do vendor/bin/testbench queue:work --queue=default --tries=1 & done

Then open http://127.0.0.1:8000/ for the list of demo endpoints. /demo-benchmark runs the same three two second tasks on all three drivers:

    sync: 6.01s   one pid, one task after another
 process: 2.17s   three local PHP processes
   queue: 2.54s   three queue worker pids

Contributing

Pull requests are welcome. Please keep composer test, composer analyse and composer lint green.

Security

If you discover a security issue, please email mathiasgrimm@gmail.com rather than using the issue tracker.

Credits

License

MIT. See LICENSE.md.

mathiasgrimm / laravel-queue-concurrency

Package info

Security

Statistics

README

Features

Requirements

Installation

Usage

Configuration

Named instances

Things to know

Relationship to laravel/framework#61273

Testing

Contributing

Security

Credits

License