Why a queue
Some work shouldn't happen while someone is waiting for a page: sending emails, delivering webhooks, resizing uploaded images, checking that a payment went through, calling a slow API. The page records what needs doing and returns straight away, and something in the background does the work.
In a PHP application the things that need doing are usually rows in a table already. An email that hasn't been sent yet. An upload that hasn't been resized. A payment that hasn't been checked. So the question is how to work through those rows reliably, and with more than one worker when there's a lot to do.
The setup
Let's use emails. The table records each email, whether it's been sent, and how many times sending has been tried:
CREATE TABLE email (
id BIGINT NOT NULL AUTO_INCREMENT,
recipient VARCHAR(250) NOT NULL,
subject VARCHAR(250) NOT NULL,
body TEXT NOT NULL,
status ENUM('waiting', 'sent', 'failed') NOT NULL DEFAULT 'waiting',
attempts INT NOT NULL DEFAULT 0,
retry_at BIGINT NULL,
PRIMARY KEY (id),
INDEX (status, retry_at)
);A page that wants to send an email just inserts a row, and carries on.
The simple way
The obvious worker is a script that cron runs every minute. It picks up the emails that are waiting, sends them, and marks each one sent:
$emails = $db->query("
SELECT * FROM email WHERE status = 'waiting' ORDER BY id LIMIT 100
")->fetchAll(PDO::FETCH_OBJ);
$markSent = $db->prepare("UPDATE email SET status = 'sent' WHERE id = ?");
foreach ($emails as $email) {
sendEmail($email);
$markSent->execute([$email->id]);
}What goes wrong
As long as there's only ever one of these running, it works. But:
- If a batch takes longer than a minute, cron starts the next run while the first is still going. Both select the same waiting rows, and people get two copies of the same email.
- For the same reason you can't run more workers to get through a backlog faster.
- If one email fails every time, it's tried again every minute forever, at the front of the queue.
- If the script dies half way through, nothing tells you which emails in the batch went out.
The usual fix is to claim rows before working on them. You add some columns to the table (who has claimed the row, when the claim runs out, how many tries it's had), or a separate jobs table with a status of pending, running or done. A worker claims a batch with an UPDATE, then reads back the rows it claimed:
UPDATE email SET claimed_by = :worker, claim_expires = :expires
WHERE claimed_by IS NULL AND status = 'waiting'
LIMIT 10;
SELECT * FROM email WHERE claimed_by = :worker;That stops the duplicates, but now every table that's a queue carries bookkeeping columns and indexes that have nothing to do with what the table is for. A worker that crashes leaves its rows claimed, so claims have to expire and something has to notice and release them. And claiming and releasing are extra writes for every item. The database already knows how to do all of this.
In Synthesizing we had a queue of chunks of music to generate, and a first version of a generic queueing system. It worked, but it needed extra columns adding to every table that took part in a queue. The method here improves on that by letting MySQL do the locking, so the table doesn't change at all.
Let MySQL do the locking
MySQL 8 has SELECT ... FOR UPDATE SKIP LOCKED. FOR UPDATE locks the rows a query returns until the end of the transaction. SKIP LOCKED says that if a row is already locked by someone else, don't wait for it: leave it out and carry on to the next. So two workers running the same query at the same moment get different rows.
$db->beginTransaction();
$id = $db->query("
SELECT id FROM email
WHERE status = 'waiting'
ORDER BY id
LIMIT 1
FOR UPDATE SKIP LOCKED
")->fetchColumn();
if ($id !== false) {
$q = $db->prepare("SELECT * FROM email WHERE id = ?");
$q->execute([$id]);
sendEmail($q->fetch(PDO::FETCH_OBJ));
$db->prepare("UPDATE email SET status = 'sent' WHERE id = ?")
->execute([$id]);
}
$db->commit();While the transaction is open, the row is locked and every other worker's query skips over it. Sending marks it sent, so once the transaction commits the row doesn't match the query any more: it has left the queue.
If sending throws, we roll back instead. The status stays at waiting, the lock is released, and the row is back in the queue for the next worker. If the worker dies part way through (killed, out of memory, the server restarting), its connection to MySQL drops, MySQL rolls the transaction back, and again the row is simply back in the queue. There are no claims to expire and nothing to clean up.
The queue is the query
Notice what we didn't need: a queue table, or any columns for the queue. A row that matches the query is, by definition, in the queue. Processing it changes something about the row that takes it out of the query, and that's how it leaves.
So any table can be a queue without being changed at all, as long as there's a query that picks out the rows that need work. A table can be several queues at once, with a different query for each. ORDER BY sets the priority, and a condition on a time column turns it into a schedule (AND send_after <= :now).
The one rule is that processing an item must change the row so that it no longer matches the query. If it doesn't, the row comes straight back round.
A small library
Each queue is then just two functions. One locks the next items and returns their ids, using FOR UPDATE SKIP LOCKED; it takes a limit so the same query can be used for counting. The other processes one item:
function lockNextEmails(PDO $db, int $limit): array {
$q = $db->prepare("
SELECT id FROM email
WHERE status = 'waiting'
AND (retry_at IS NULL OR retry_at <= :now)
ORDER BY id
LIMIT :limit
FOR UPDATE SKIP LOCKED
");
$q->bindValue(':now', time(), PDO::PARAM_INT);
$q->bindValue(':limit', $limit, PDO::PARAM_INT);
$q->execute();
return array_map('intval', $q->fetchAll(PDO::FETCH_COLUMN));
}
function processEmail(PDO $db, int $id): void {
$q = $db->prepare("SELECT * FROM email WHERE id = ?");
$q->execute([$id]);
$email = $q->fetch(PDO::FETCH_OBJ);
sendEmail($email);
$db->prepare("UPDATE email SET status = 'sent' WHERE id = ?")
->execute([$id]);
}Everything else is the same for every queue. Processing one item is a transaction around those two calls, and afterwards a note of how it went:
function processNext(PDO $db, string $name, callable $lockNext, callable $process): bool {
$db->beginTransaction();
ob_start();
$id = null;
try {
$ids = $lockNext($db, 1);
if (!$ids) {
$db->commit();
ob_end_clean();
return false; // nothing waiting
}
$id = $ids[0];
$process($db, $id);
$db->commit();
recordResult($db, $name, $id, true, ob_get_clean());
return true;
} catch (Throwable $e) {
$db->rollBack(); // the item goes back into the queue
$output = ob_get_clean();
if ($id !== null) {
recordResult($db, $name, $id, false, $e->getMessage() . "\n\n" . $output);
}
throw $e;
}
}The record goes in a table of its own, one row for every item processed, whether it worked, and anything the process function printed, which output buffering catches. It's written after the commit or the rollback, never inside the transaction, so the record of a failure isn't rolled back along with the failure:
CREATE TABLE queue_result (
id BIGINT NOT NULL AUTO_INCREMENT,
queue VARCHAR(250) NOT NULL,
row_id BIGINT NOT NULL,
success TINYINT NOT NULL,
output TEXT NULL,
processed_at BIGINT NOT NULL,
PRIMARY KEY (id),
INDEX (queue, processed_at)
);function recordResult(PDO $db, string $queue, int $id, bool $success, string $output): void {
$db->prepare("
INSERT INTO queue_result (queue, row_id, success, output, processed_at)
VALUES (?, ?, ?, ?, ?)
")->execute([$queue, $id, (int)$success, $output, time()]);
}That's the table to look at when someone asks whether their email went, and why not. I also count successes, errors and empty visits for each queue with fast log counters, which is enough to draw a graph of each queue without adding up the results table.
Counting what's waiting uses the same query with a bigger limit, in a transaction that commits straight away so the locks are let go at once:
function countWaiting(PDO $db, callable $lockNext): int {
$db->beginTransaction();
$ids = $lockNext($db, 100);
$db->commit();
return count($ids);
}It stops at 100 so it stays quick on a big backlog, and it only counts rows nobody is working on right now. That's enough for a dashboard, and it's what tells the driver, below, whether to keep going.
Running the queues
Something has to call processNext, over and over, for as long as there's work. One way is a PHP script that loops forever. I don't do that. Instead, PHP processes one item per web request, in the same PHP the site runs in, with the same configuration, deployed along with it. There are no long-running PHP processes to supervise, nothing builds up from one item to the next, and every item starts from a clean request.
Each queue is registered with its two functions and three limits:
$queues = [
'email' => [
'lockNext' => 'lockNextEmails',
'process' => 'processEmail',
'limits' => [
'maxPerSecond' => 10, // start at most ten a second
'maxInFlight' => 20, // with at most twenty running at once
'idleRecheckSeconds' => 5, // when it's empty, look again every five seconds
],
],
];One endpoint serves two requests. A GET lists every queue with its limits and how much is waiting. A POST processes one item from the named queue, and answers with how many are left:
<?php
require 'queues.php';
if (!hash_equals(QUEUE_TOKEN, $_SERVER['HTTP_X_QUEUE_TOKEN'] ?? '')) {
http_response_code(401);
exit;
}
header('Content-Type: application/json');
if ($_SERVER['REQUEST_METHOD'] === 'GET') {
$list = [];
foreach ($queues as $name => $queue) {
$list[] = ['name' => $name, 'waiting' => countWaiting($db, $queue['lockNext'])]
+ $queue['limits'];
}
echo json_encode($list);
exit;
}
$name = json_decode(file_get_contents('php://input'))->queue ?? '';
$queue = $queues[$name] ?? null;
if (!$queue) {
http_response_code(404);
exit;
}
try {
processNext($db, $name, $queue['lockNext'], $queue['process']);
} catch (Throwable $e) {
http_response_code(500); // already recorded, and the item is back in the queue
exit;
}
echo json_encode(['remaining' => countWaiting($db, $queue['lockNext'])]);The endpoint is on the open web, so it checks a shared secret in a header, compared with hash_equals so the comparison takes the same time however much of the token is right.
The driver
PHP only runs when it's asked, so the last piece is an outside timer that keeps asking. It's a small, separate service that I call the driver. It isn't a worker: it does no work itself and never touches the database. It only makes requests to the endpoint, at the right rate, for as long as there's work.
Each queue in the driver is either draining or idle. While it's draining, the driver starts process requests no faster than maxPerSecond, with no more than maxInFlight running at once, and each answer says how many are left. When nothing is left and every request has finished, the queue goes idle. After idleRecheckSeconds it tries a single process request, and if that finds more work, it starts draining again.
const SITE = 'https://example.com/queue.php';
const TOKEN = process.env.QUEUE_TOKEN;
async function call(method, body) {
const response = await fetch(SITE, {
method,
headers: { 'X-Queue-Token': TOKEN, 'Content-Type': 'application/json' },
body: body && JSON.stringify(body),
});
if (!response.ok) throw new Error(`HTTP ${response.status}`);
return response.json();
}
async function processOne(name) {
try {
return await call('POST', { queue: name });
} catch (e) {
console.error(`${name}: ${e.message}`);
return null;
}
}
function startQueue(queue) {
const spacing = 1000 / queue.maxPerSecond;
let draining = false;
let remaining = 1; // assume there's work until the first answer says otherwise
let inFlight = 0;
let lastStart = 0;
let timer = null;
function drain() {
draining = true;
pump();
}
function idle() {
draining = false;
setTimeout(probe, queue.idleRecheckSeconds * 1000);
}
async function probe() {
const result = await processOne(queue.name);
if (result && result.remaining > 0) {
remaining = result.remaining;
drain();
} else {
idle();
}
}
function pump() {
if (!draining) return;
if (remaining <= 0) {
if (inFlight === 0) idle(); // empty, and everything has finished
return;
}
if (inFlight >= queue.maxInFlight) return; // the next one to finish calls pump again
const wait = lastStart + spacing - Date.now();
if (wait > 0) { // too soon: come back when it's time
if (!timer) timer = setTimeout(() => { timer = null; pump(); }, wait);
return;
}
lastStart = Date.now();
inFlight++;
processOne(queue.name).then((result) => {
inFlight--;
remaining = result ? result.remaining : 0; // an error stops draining until the next re-check
pump();
});
pump();
}
drain();
}The list of queues only changes when the site is deployed, so the driver asks for it every thirty seconds or so, and starts any queue it hasn't seen before:
const started = new Set();
async function pollList() {
try {
for (const queue of await call('GET')) {
if (!started.has(queue.name)) {
started.add(queue.name);
startQueue(queue);
}
}
} catch (e) {
console.error(`list: ${e.message}`); // the site is down or deploying: try again later
}
setTimeout(pollList, 30 * 1000);
}
pollList();A few things follow from this:
- New work waits at most
idleRecheckSeconds. Each queue re-checks on its own, so a new email doesn't wait for the list, or for cron's next minute. - An error stops the draining. If the site starts failing, the driver doesn't hammer it with requests: it drops back to one probe every few seconds until things recover.
- The driver can be restarted at any time. It holds no state that matters. Anything it was in the middle of is either committed or rolled back on the PHP side, and the first probe picks up where it left off.
- A second driver can't do any harm. The locks still give every request a different row, so two drivers never process the same item. They'd just double the rate.
- The limits are set by the site, not the driver. They come from the list, next to the code they're for, so changing one is a normal deploy.
The driver can run anywhere that can reach the site. I run it in a small container next to the web server, so it starts and stops with everything else.
Failures and retries
Throwing an exception puts the item straight back in the queue, which is right for a passing problem like a timeout. It's wrong for an email that will never send: it would come straight back round, again and again, and with ORDER BY id it would stay at the front. With the driver it's worse than that: an error stops the draining, so one bad item at the front would hold up the whole queue.
For failures that might not go away, the process function records the failure on the row instead of throwing, and that takes the row out of the query for a while, or for good. The queue still doesn't need anything of its own: how many times an email has been tried and when to try it again are facts about the email.
function processEmail(PDO $db, int $id): void {
$q = $db->prepare("SELECT * FROM email WHERE id = ?");
$q->execute([$id]);
$email = $q->fetch(PDO::FETCH_OBJ);
try {
sendEmail($email);
$db->prepare("UPDATE email SET status = 'sent' WHERE id = ?")
->execute([$id]);
} catch (MailError $e) {
$attempts = $email->attempts + 1;
if ($attempts >= 5) {
// give up: 'failed' is out of the queue for good
$db->prepare("UPDATE email SET status = 'failed', attempts = ?
WHERE id = ?")
->execute([$attempts, $id]);
} else {
// try again later: a minute, then two, then three...
$db->prepare("UPDATE email SET attempts = ?, retry_at = ? WHERE id = ?")
->execute([$attempts, time() + 60 * $attempts, $id]);
}
}
}Because the update happens inside the same transaction, it commits along with everything else, and the row leaves the query until its retry_at comes round.
Things to know
- The transaction stays open while an item is processed, holding a database connection and the row's lock for the length of one request. A few seconds is fine. Processing an item has to fit within PHP's time limit and the web server's. For work that takes minutes, like rendering video on a GPU, it's better for a worker to claim the item and report its progress, which is a different design.
- Every request in flight is a PHP process and a database connection that the site's visitors can't use. Keep
maxInFlightwell below the number of PHP workers the server has, added up across every queue that might be draining at once. maxPerSecondis also how you stay within someone else's limits, like the number of emails a mail provider will take a second.- Sending an email can't be rolled back. If a worker dies after the email has gone but before the commit, the row goes back into the queue and the email is sent again. Like nearly every queue, this one delivers at least once, so where it matters, make processing safe to repeat.
- Keep the queue's query on an index (here
status, retry_at), so finding the next row is quick and doesn't touch rows it doesn't need. SKIP LOCKEDneeds MySQL 8.0 or MariaDB 10.6 or later, and InnoDB tables.
That's all there is to it: a query that says what's waiting, a function that does the work and changes the row, and MySQL's own locks sharing it all out.