Queues and Jobs

In the Authentication and Authorization lesson the gate skipped the action and the response returned at once. When the gate lets the request through, the action runs, and inside the action there may be a job that delays the response: producing the thumbnail of an image, preparing a report. This lesson asks: should this job make the response wait, and if not, who runs it and when?

First, naming. In this lesson the model is in the same process: there is no separate server, separate process or network. The “queue” is an array, and the worker is the name of the process() call that takes jobs from this array and runs them. The call is made after the response of the request has returned. In a real system the worker would be a separate process; here the distinction is only whether the response waits for the job. A similar question appeared in the JVM, Concurrency and Performance course (M08/K04), but there the search was for who carries out the step: which thread ran it was measured. The question here is different: whether the response waits for the job.

AS2. The queue is an array, the worker is a synchronous loop. Timeouts, concurrent workers, locking and the durability of job order are not modeled; no queue driver name is used. Each attempt is tracked by a counter and sleep is never called: the retry interval is the attempt number, not real time. An exception thrown inside a job is caught in the worker and never reaches the response.

The Measurement Core

The same measurement is used in every lesson of the course. ran.php records whether an expected part ran in this request; the file is given verbatim in every lesson, and it has no output block, since it is only saved to disk. Ledger keeps three lists: the parts placed on the request’s path, the parts called within the request, and the parts left outside the request. classify() checks them in order. The decision order is classify() itself: ran, deferred, skipped, not wired. The developer names the part; what is measured are the parts the developer expects to run in this request.

<?php
// ran.php — did an expected part run in this request?
declare(strict_types=1);

final class Ledger {
    public static array $path = [];        // parts placed on the request's path
    public static array $ran = [];         // parts called within the request
    public static array $deferred = [];    // parts left outside the request
    public static function newRequest(): void { self::$path = self::$ran = self::$deferred = []; }
    public static function onPath(string $part): void { self::$path[] = $part; }
    public static function ran(string $part): void { self::$ran[] = $part; }
    public static function defer(string $part): void { self::$deferred[] = $part; }
}

function classify(string $part): string {
    return match (true) {
        in_array($part, Ledger::$ran, true) => 'ran',
        in_array($part, Ledger::$deferred, true) => 'deferred',
        in_array($part, Ledger::$path, true) => 'skipped',
        default => 'not wired',
    };
}

This lesson uses all three calls of the core. The thumbnail job is placed on the path (Ledger::onPath()), marked as left outside the request with Ledger::defer() when it is sent to the queue, and marked as called with Ledger::ran() on the sync driver.

The Job Class, the Driver and the Worker

jobs.php is the whole slice. BaseJob is the contract of a job: a name(), a run() and an attempt property that counts how many times the job has been tried. ThumbnailJob takes a single post id, and when run() is called it adds the id to an array; the id being written to the array is the proof that the job ran.

The driver decides how a job is sent. The Driver interface has a single method, send(). SyncDriver does not put the job on a queue; it marks it with Ledger::ran() and runs it at once. MemoryDriver marks it with Ledger::defer(), adds the job to an array and returns the response; the job waits there. process() runs the waiting jobs: it pulls each job from the array, increments the attempt, calls run(), and if it catches an exception it puts the job back at the end of the array when attempts remain, or moves it to the failed list when none remain.

PostsController::save() does not know the driver. It talks only to the Driver interface: it creates a job object, calls send() and returns 201. The controller’s code is the same on both drivers; what changes is what the driver does with the job.

<?php
// jobs.php — the job class, two drivers and the worker
declare(strict_types=1);
require_once 'ran.php';

abstract class BaseJob {
    public int $attempt = 0;
    abstract public function name(): string;
    abstract public function run(): void;
}

final class ThumbnailJob extends BaseJob {
    public static array $produced = [];
    public function __construct(private int $postId) {}
    public function name(): string { return 'thumbnail'; }
    public function run(): void { self::$produced[] = $this->postId; }
}

interface Driver {
    public function send(BaseJob $job): void;
}

final class SyncDriver implements Driver {
    public function send(BaseJob $job): void {
        Ledger::ran($job->name());
        $job->run();
    }
}

final class MemoryDriver implements Driver {
    private array $pending = [];
    public array $failed = [];
    public function __construct(private int $maxAttempts = 3) {}
    public function send(BaseJob $job): void {
        Ledger::defer($job->name());
        $this->pending[] = $job;
    }
    public function pending(): int { return count($this->pending); }
    public function process(): array {         // worker: runs independently of the request
        $log = [];
        while ($job = array_shift($this->pending)) {
            $job->attempt++;
            try {
                $job->run();
                $log[] = "{$job->name()} attempt {$job->attempt}: ran";
            } catch (Throwable $error) {
                $log[] = "{$job->name()} attempt {$job->attempt}: " . get_class($error);
                if ($job->attempt < $this->maxAttempts) $this->pending[] = $job;
                else $this->failed[] = $job;
            }
        }
        return $log;
    }
}

final class PostsController {
    public function __construct(private Driver $driver) {}
    public function save(int $postId): array {
        Ledger::newRequest();
        Ledger::onPath('save');
        Ledger::onPath('thumbnail');
        Ledger::ran('save');
        $this->driver->send(new ThumbnailJob($postId));
        return [201, 'saved'];
    }
}

A Job Pushed onto the Queue

The first measurement asks about two parts: save (the controller’s own step) and thumbnail (the job pushed onto the queue). The driver is the queue. After the response returns, process() is called and the worker runs. The contrasting measurement runs the same controller with the sync driver.

<?php
// queue.php — the same job with two drivers
require_once 'jobs.php';

function measure(string $name, Driver $driver): void {
    ThumbnailJob::$produced = [];
    $response = (new PostsController($driver))->save(7);
    printf("%s -> %d %s\n", $name, $response[0], $response[1]);
    foreach (['save', 'thumbnail'] as $part) printf("  %-11s %s\n", $part, classify($part));
    echo '  on path: ', in_array('thumbnail', Ledger::$path, true) ? 'yes' : 'no', "\n";
    echo '  produced at response time: ', count(ThumbnailJob::$produced), "\n";
}

$queue = new MemoryDriver();
measure('queue', $queue);
echo '  pending jobs: ', $queue->pending(), "\n";
foreach ($queue->process() as $line) echo '  worker: ', $line, "\n";
echo '  produced after worker: ', count(ThumbnailJob::$produced), "\n";

measure('sync', new SyncDriver());

queue -> 201 saved
  save        ran
  thumbnail   deferred
  on path: yes
  produced at response time: 0
  pending jobs: 1
  worker: thumbnail attempt 1: ran
  produced after worker: 1
sync -> 201 saved
  save        ran
  thumbnail   ran
  on path: yes
  produced at response time: 1

This lesson’s observation is in the first block. The response returned 201, but thumbnail was deferred: it had been placed on the path, it was not called within the request, and it was left outside the request. The “produced at response time” count is 0, so the thumbnail does not exist yet; the pending job count is 1. When the worker ran through process(), the job ran on the first attempt and the produced count became 1. The client that received the response did not wait for the thumbnail.

The decision order is useful here. Because thumbnail is on the path, the on path line says yes; that line alone does not say the job ran. classify() first looks at the ran list and does not find it there; then it looks at the deferred list and finds it. Being “on the path” and “running” are separate observations, and a deferred part sits exactly between them.

The Contrast: The Sync Driver

The second block runs the same controller with SyncDriver. Same job, same send() call; this time thumbnail ran and the count produced at response time is 1. The client receives its response after the job has finished. The code is the same, and the only thing that changed is the driver. This shows that the queue does not make the job smaller; it only moves the moment the job runs outside the request.

To put the distinction in one sentence: in M08/K04 we looked for who carries out the step; here what matters is not who carries it out but whether the response waits for it. On the sync driver the response waits, on the queue it does not.

The Bounding Measurement

Deferring the job through the queue also has a cost: because the job runs after the response, an error does not reach the client and stays in the worker. This is why retries and the failed jobs list are needed. The measurement below builds two jobs. The first throws an exception only on its first attempt; the second throws on its first five attempts. The attempt allowance is 3.

<?php
// retry.php — a job that fails on its first attempt
require_once 'jobs.php';

final class FlakyJob extends BaseJob {
    public function __construct(private int $failingAttempts) {}
    public function name(): string { return 'flaky'; }
    public function run(): void {
        if ($this->attempt <= $this->failingAttempts) throw new RuntimeException('temporary error');
    }
}

foreach ([[1, 3], [5, 3]] as [$failing, $allowance]) {
    Ledger::newRequest();
    $queue = new MemoryDriver($allowance);
    $queue->send(new FlakyJob($failing));
    echo "failing attempts $failing, allowance $allowance\n";
    echo '  in request: ', classify('flaky'), "\n";
    foreach ($queue->process() as $line) echo '  worker: ', $line, "\n";
    echo '  failed jobs: ', count($queue->failed), "\n";
}

failing attempts 1, allowance 3
  in request: deferred
  worker: flaky attempt 1: RuntimeException
  worker: flaky attempt 2: ran
  failed jobs: 0
failing attempts 5, allowance 3
  in request: deferred
  worker: flaky attempt 1: RuntimeException
  worker: flaky attempt 2: RuntimeException
  worker: flaky attempt 3: RuntimeException
  failed jobs: 1

Both jobs were deferred in the request: the response never saw the error in either case. The first job threw RuntimeException on its first attempt, the worker caught it, put the job back at the end of the array because its allowance was not used up, and on the second attempt the job ran. The failed jobs list stayed empty. The second job failed on all three attempts. Because no allowance remained after the third attempt, it was not put back and fell into the failed list: there is 1 job in the list. The fifth attempt was never made.

The numbers say this: retrying is for temporary errors and does not try forever if the error persists; there is a limited allowance, and when it runs out the job falls into a visible place, the failed jobs list. If that list is not empty, someone needs to look at it. A response returning 201 says only that the post was saved; it does not show that the thumbnail job succeeded, because that job has only been pushed onto the queue.

Summary

• This lesson’s outcome is deferred: the controller pushed the thumbnail job onto the queue, the response returned 201, the job was not called at response time, and it ran in the worker.
• The model is in the same process; the worker is the name of the process() call and runs after the response.
• Its contrast: on the sync driver the same job ran and the response waited for it; the controller code did not change, only the driver did.
• In the bounding measurement, the job that failed on its first attempt ran on the second attempt; the job that failed on all three attempts fell into the failed jobs list.
• Having been pushed onto the queue is not the same as having succeeded: 201 says the post was saved, and it does not say what became of the job in the queue.

Next Step

The thumbnail job went onto the queue, but the controller knew it by name: new ThumbnailJob(...). When a new side effect is needed, for example updating a search index when a post is saved, a new line has to be added to the controller. The Events and Listeners lesson loosens this tie: the controller dispatches an event, and a listener handles it. But what happens if the event name does not match the name the listener is registered under?