Jobs are widely regarded as the most complex and hardest to master part of the Laravel framework. Recently I have been rereading Laravel queues in action by Mohamed Said and I started wondering not only how jobs work, but how exactly are they implemented. After some time of reading Laravel code, blogs and writing some mock code I ended up writing a simple experiment called Hobby.

Defining terms

There are several terms that we need to define before deep diving into writing code.

1. queue: place where hobbies go to await their destiny
2. queue driver: hobbies are async processes that wait to be executed for a long time, so there needs to be some way of saving their data, like input parameters, queue and similar. Driver is a store that saves that data. In Laravel it can be Redis, database, SQS (and some other). In our experiment we will use Redis
3. hobbyist: is a continuous process that checks if there are hobbies to be executed -- and if there are -- executes them.

Let's start coding

As said before, hobbyist is a continuous process -- basically a never-ending while loop -- that checks if there are hobbies in the queue and then processes them. So it should look something like this:

<?php

declare(strict_types=1);

require_once __DIR__ . '/vendor/autoload.php';

while (true) {
    // process jobs
}

We want to have a possibility of running the hobbyist for a specific queue, so it should have an optional parameter that has a default value of default.

$queue = $argv[1] ?? 'default';

Loop can be interrupted using SIGINT and SIGTERM signals. However, simply terminating the process is dangerous, since that could lead to losing data, inconsistent data, and similar. Fortunately, PHP allows us to define what we want to do when our process gets those signals, using the pcntl_signal(int $signal, $handler, bool $restart_syscalls = true) method. It takes signal type as a first parameter and as a second parameter we can pass the handler that defines what should happen when a signal is received. So, if we move our hobbyist logic to a class, we can do something like this:

<?php

declare(strict_types=1);

namespace src;

final class Hobbyist
{
    private bool $running = true;

    public function __construct(private readonly Client $redis)
    {
        pcntl_signal(SIGTERM, fn() => $this->running = false);
        pcntl_signal(SIGINT,  fn() => $this->running = false);
    }

    public function run(): void
    {
        while ($this->running) {
            pcntl_signal_dispatch();

            $this->process();
        }
    }

Here, handlers will simply toggle $running flag when signals are received, ensuring that the hobby is finished and the hobbyist is gracefully terminated. The process() method should check the queue and process any hobby that is found there, but before that we need to dispatch them.

Dispatching hobbies

Dispatching hobbies essentially means placing them in the queue. Before that is done, we need to serialize and save some hobby-related data, like input parameters, queue that the hobby is dispatched to, delay and similar. We will use three attributes for customizing hobbies in this experiment: Delay, MaxAttempts and OnQueue.

<?php

declare(strict_types=1);

namespace src\Attributes;

use Attribute;

#[Attribute(Attribute::TARGET_CLASS)]
final readonly class Delay
{
    public function __construct(
        public int $seconds = 0,
    ) {}
}

// ...

#[Attribute(Attribute::TARGET_CLASS)]
final readonly class MaxAttempts
{
    public function __construct(
        public int $tries = 3,
    ) {}
}

// ...

#[Attribute(Attribute::TARGET_CLASS)]
final readonly class OnQueue
{
    public function __construct(
        public string $name = 'default',
    ) {}
}

Delay is given when we don't want the hobby to be executed immediately. It is useful when we want to make sure that all prerequisites are completed before the hobby is dispatched. MaxAttempts defines how many times a hobby can be executed with failure before we definitely mark it as failed. OnQueue makes it possible for a hobby to be dispatched on different queues, providing flexibility in their management.

With all of that in mind, the Dispatcher class could look something like this:

<?php

declare(strict_types=1);

namespace src;

use Predis\Client;
use ReflectionClass;
use src\Attributes\Delay;
use src\Attributes\OnQueue;
use src\Contracts\Hobby;

final readonly class Dispatcher
{
    public function __construct(private Client $redis)
    {}

    /**
     * @throws \ReflectionException
     */
    public function dispatch(Hobby $hobby): void
    {
        $queue = $this->resolveQueue($hobby);
        $delay = $this->resolveDelay($hobby);
        $payload = json_encode([
            'class' => $hobby::class,
            'args' => $this->extractArgs($hobby),
            'attempts' => 0,
            'queue' => $queue,
        ]);

        if ($delay > 0) {
            /**
             * adding a hobby in a sorted fashion, with a score
             * this score is the timestamp when the hobby should be promoted
             * to the main queue and be available for processing
             */
            $this->redis->zadd('queue:delayed', [strval($payload) => time() + $delay]);
        } else {
            $this->redis->rpush("queue:{$queue}", (array) $payload);
        }
    }

    private function resolveDelay(object $hobby): int
    {
        /** resolve delay using Reflection API */
    }

    private function resolveQueue(object $hobby): string
    {
        /** resolve queue using Reflection API */
    }

    private function extractArgs(object $hobby): array
    {
        /** extract arguments using Reflection API */
    }
}

Now that we have a dispatcher, we can implement the process() method from Hobbyist. Its task is pretty simple, it needs to deserialize the payload, execute the hobby and in a case of failure it either retries it -- if the MaxAttempts is set -- or adds it to the queue:failed list in Redis. In reality, failed hobbies would be permanently saved in a database, for easier debugging.

try {
    $hobby->handle();
    $this->output("✓ {$class} succeeded (attempt {$attempts}/{$maxAttempts})");
} catch (\Throwable $e) {
    if ($attempts < $maxAttempts) {
        /**
         * push the hobby to the right (end) of the queue
         * this means that queue is FIFO
         */
        $this->redis->rpush("queue:{$queue}", (array) json_encode([
            'class' => $class,
            'args' => $args,
            'attempts' => $attempts,
            'queue' => $queue,
        ]));
        $this->output("↺ {$class} failed, retrying (attempt {$attempts}/{$maxAttempts}): {$e->getMessage()}");
    } else {
        $this->redis->rpush('queue:failed', (array) json_encode([
            'class' => $class,
            'args' => $args,
            'attempts' => $attempts,
            'queue' => $queue,
            'error' => $e->getMessage(),
        ]));
        $this->output("✗ {$class} failed permanently after {$attempts} attempts: {$e->getMessage()}");
    }
}

Promoting delayed hobbies

But wait -- we mentioned that delayed hobbies are stored in a sorted set, so how do they ever reach the main queue? Before each loop iteration checks for work, the hobbyist promotes any delayed hobbies whose time has come:

private function promoteDelayedJobs(): void
{
    /**
     * get all the hobbies that are due for promotion (score <= current timestamp)
     */
    $payloads = $this->redis->zrangebyscore('queue:delayed', 0, time());

    foreach ($payloads as $payload) {
        $targetQueue = json_decode($payload, associative: true)['queue'];
        /**
         * push to the main queue first, then remove from the delayed set
         * this order ensures the hobby is never lost if something crashes in between
         */
        $this->redis->rpush("queue:{$targetQueue}", (array) $payload);
        $this->redis->zrem('queue:delayed', $payload);
    }
}

It queries ZRANGEBYSCORE for all hobbies whose score (the Unix timestamp we set during dispatch) is less than or equal to the current time -- meaning they are ready. For each one, it pushes the payload to the main queue and then removes it from the delayed set. The order matters: push first, remove second. If the process crashes between the two operations, the hobby ends up in both places, but that is safer than losing it entirely.

Example of hobbies

Lastly, we will create some simple hobbies. They will implement a simple Hobby interface that only has a handle() method.

interface Hobby
{
    public function handle(): void;
}

// ...

#[OnQueue('test')]
#[MaxAttempts(3)]
#[Delay(10)]
readonly class DelayedMessageHobby implements Hobby
{
    public function __construct(
        private string $message,
    ) {}

    public function handle(): void
    {
        $line = sprintf("[%s] (delayed) %s\n", date('Y-m-d H:i:s'), $this->message);

        file_put_contents(__DIR__ . '/../storage/logs.txt', $line, FILE_APPEND);
    }
}

// ...

#[OnQueue('default')]
#[MaxAttempts(3)]
readonly class FailingHobby implements Hobby
{
    public function __construct() {}

    public function handle(): void
    {
        throw new \RuntimeException("This hobby always fails.");
    }
}

// ...

#[OnQueue('default')]
#[MaxAttempts(3)]
readonly class LogMessageHobby implements Hobby
{
    public function __construct(
        private string $message,
    ) {}

    public function handle(): void
    {
        $path = __DIR__ . '/../storage/logs.txt';

        if (! is_dir(dirname($path))) {
            mkdir(dirname($path), recursive: true);
        }

        $line = sprintf("[%s] %s\n", date('Y-m-d H:i:s'), $this->message);

        if (file_put_contents($path, $line, FILE_APPEND) === false) {
            throw new \RuntimeException("Failed to write to {$path}");
        }
    }
}

We will dispatch them using a helper file dispatch.php that simply calls the Dispatcher class.

Testing

In one terminal we will run php hobbyist.php and in the other php hobbyist.php test. The first one will process all hobbies from the default queue and the other one all hobbies from the test queue (in our case it is DelayedMessageHobby).

In the third terminal we will call php dispatch.php.

php hobbyist.php
[10:49:03] ✓ hobbies\LogMessageHobby succeeded (attempt 1/3)
[10:49:03] ↺ hobbies\FailingHobby failed, retrying (attempt 1/3): This hobby always fails.
[10:49:03] ↺ hobbies\FailingHobby failed, retrying (attempt 2/3): This hobby always fails.
[10:49:03] ✗ hobbies\FailingHobby failed permanently after 3 attempts: This hobby always fails.

...

php hobbyist.php test
[10:49:13] ✓ hobbies\DelayedMessageHobby succeeded (attempt 1/3)

So as we can see LogMessageHobby was executed successfully, FailingHobby is marked as failed after failing 3 times, and DelayedMessageHobby was executed after being delayed for 10 seconds.

Using redis-cli we can see what the queue:failed key looks like:

lrange queue:failed 0 -1
1) "{\"class\":\"hobbies\\\\FailingHobby\",\"args\":[],\"attempts\":3,\"queue\":\"default\",\"error\":\"This hobby always fails.\"}"

This info -- and especially the error message -- is vital for successful debugging.

We can also see what a hobby looks like while it is in the queue. If we dispatch LogMessageHobby to the inspect-hobby queue it would look like this:

lrange queue:inspect-hobby 0 -1
1) "{\"class\":\"hobbies\\\\LogMessageHobby\",\"args\":[\"Hello from the queue!\"],\"attempts\":0,\"queue\":\"inspect-hobby\"}"

Conclusion

Learning by creating is always the best way to learn something and this experiment was an interesting way to better understand the mechanics behind Laravel jobs. Code can be found on my GitHub profile.