Job queues

Many workers taking jobs from one collection: each job claimed by one of them under a lease, then acked, extended or failed — every step a statement.

One statement claims. A set with order and limit writes the rows a get with the same clauses would answer, and returning hands them back as written:

set jobs {owner: $1, run_at: now() + 30000, attempts: attempts + 1}
  where run_at <= now() order run_at limit 10 returning *

The ten oldest ready jobs are found and written under the one writer's lock, so two workers never take the same one — what PostgreSQL's FOR UPDATE SKIP LOCKED is for, with no locks to skip. A job's run_at is the time it may next run: when it was enqueued, or later for a delayed one, and while a worker holds it, the end of its lease. So a worker that dies holds nothing for long: its lease lapses, the job is ready again, and the next claim takes it. See Writes that pick their rows for the rules.

This is Redis's LPUSH and BRPOP, with a lease so that a job outlives the worker that took it. The other jobs an app adds Redis for — locks, rate limits, caches, streams, pub/sub — are recipes of Instead of Redis.

Every FenecQL block on this page is run, statement by statement, by a test (crates/fenec-core/tests/queues_docs.rs), and each -- → x is the answer it checks: a write's count, or the first value of the first row it answered. A line -- 20 s later moves the test's clock. The costs are make queue-bench on an M1 Air, over a million jobs.

The queue

A collection a queue. run_at takes @sorted, so a claim walks the ready jobs from the oldest and stops at its page.

create collection jobs (payload json, run_at timestamp @sorted, owner text, attempts int, error text)
put jobs {payload: {"to": "ada@example.com"}, run_at: now(), attempts: 0}
put jobs [{payload: {"to": "bo@example.com"}, run_at: now(), attempts: 0}, {payload: {"to": "cy@example.com"}, run_at: now(), attempts: 0}]
put jobs {payload: {"to": "di@example.com"}, run_at: now() + 3600000, attempts: 0}     -- → 1: delayed an hour

• A delayed job is one whose run_at is ahead: no claim takes it before then.
• Jobs enqueued at the same time come out in the order they were written, their ids'.
• Several queues are several collections. Kept in one, a queue field goes without @hash: the walk tests it on each job it passes, where through a hash index a claim would read the whole queue to put it in order.

Claim

Each worker claims a batch, the oldest ready jobs, each its own for 30 seconds:

set jobs {owner: "w1", run_at: now() + 30000, attempts: attempts + 1} where run_at <= now() order run_at limit 2 returning id, payload.to   -- → 1: jobs 1 and 2
set jobs {owner: "w2", run_at: now() + 30000, attempts: attempts + 1} where run_at <= now() order run_at limit 2 returning id, payload.to   -- → 3: the one left ready
set jobs {owner: "w3", run_at: now() + 30000, attempts: attempts + 1} where run_at <= now() order run_at limit 2 returning id   -- → no row: none ready

• None is taken twice. The claim finds and writes its jobs under the write lock. Sixteen threads claiming batches of ten from 6 000 jobs, and eight HTTP clients from 2 000, never handed a job to two of them, and every job was acked (crates/fenec-core/tests/claims.rs, crates/fenec-http/tests/claims.rs).
• returning * answers every field; a list answers those, paths into a json field among them. Without returning the answer is the count.
• limit 1 require 1 is one job or a refusal (412): require counts the rows the claim answers.
• An empty queue answers no rows at once — or, sent over HTTP with Fenec-Wait, holds the claim at the server until a job comes (Waiting for a job).

Waiting for a job

A worker whose claim finds nothing need not sleep and claim again. Sent with Fenec-Wait, milliseconds, the claim is held at the server while it would take nothing, and answered the moment it takes a job — one enqueued, a delayed one come due, a lease lapsed — or at the end of the wait with no row: Redis's BLPOP, SQS's long poll.

curl -X POST http://127.0.0.1:8080/query -H 'Fenec-Wait: 30000' \
  -d '{"query": "set jobs {owner: $1, run_at: now() + 30000, attempts: attempts + 1} where run_at <= now() order run_at limit 10 returning *", "params": ["w1:1842"]}'

• A job enqueued reaches a waiting worker in 0.38 ms at the median and 0.76 at p99, where a worker polling every 10 ms took 0.83 and 10.1 and every 100 ms 15.5 and 85.8 — and polling every 10 ms sent ten times the claims (costs).
• Time wakes it too. No write announces a delayed job's run_at or a lease's end, so the server works the moment out from the claim's filter: run_at <= now() changes its answer only as the time passes a job's run_at, and the next one is a step of the @sorted index past now. The claim is held until that moment and runs then, never before.
• One job, many workers: it goes to the worker that has waited longest, and the others stay held. Claims that would take the same jobs — the same filter, scope and parameters in it, whatever they write — wait in line, and the line looks once whether a job is there; the first runs, and having taken a job hands the turn to the next, since more may be ready. A write to the queue costs a line one look under the read lock, not a claim a worker.
• Claims of their own, one job: it goes to the claim whose own order puts it first, and on a tie to the one that has waited longest. A courier's claim is a line of its own -- its point is in its filter, its token's subject in its rules -- so every courier's claim looks when an order is placed, all of them by one held request under one read lock, each among the orders placed since it last looked; each that finds a row bids for it, and once every claim that could beat the best bid has looked, the row is the best's, its claim alone is woken, and the rest wait behind its run (Who gets the job).
• At most 30 seconds, and the answer is then what the claim answers on an empty queue: [], or a require's 412. A worker sends the next.
• A set or a del, alone or in a /batch beside reads: a put writes every time, and a request holding one is refused (400), as is a filter that reads the time other than as a field compared with now() give or take a number — bucket(now(), 60000) — since nothing says when its answer changes.

Who gets the job

A job every waiting claim would take goes to one claim, by the claim's own order:

• A claim ordering by a field -- order run_at, a queue's -- ranks a job as every other claim does, so the tie goes to the claim that has waited longest: first come, first served across claims as within one line, as Redis serves BLPOP's clients. A burst of jobs goes to the waiting claims in the order they came, the oldest job to the claim held longest, each claim looking for the next once the run before it is done.
• A claim ordering by its own point -- a courier's order distance(pickup, $1) limit 1 (The nearest job) -- ranks a job by how far it is from the courier, so of the couriers waiting the nearest takes it, the others looking again once its run is done, each for the job nearest it then. That needs every waiting claim's look, which costs what was written rather than what each claim would read: one order at a time to couriers waiting in a city 10 km across, the nearest took every one, run for once, 0.18 ms from the put at 10 couriers, 0.31-0.32 ms at 100 and 1.7 ms at 1 000, the server's CPU 0.2, 0.4-0.5 and 3.1-3.2 ms an order, against a raced 0.2-0.4 ms to a courier a median of 3.3-5 km further, two runs an order (make queue-bench's held nearest, in turns).
• No header turns it on. The claim's order says what "first" means, and a tie keeps the order claims came in.

Ack, extend

A worker acks a job it holds by deleting it, and keeps one longer by moving its lease on. Each names the owner it was claimed by and requires its row, so a worker whose lease lapsed, and whose job another worker claimed meanwhile, finds out:

del jobs where id = 1 and owner = "w1" require 1                       -- → 1: done
-- 20 s later
set jobs {run_at: now() + 30000} where id = 2 and owner = "w1" require 1   -- → 1: 30 s more
-- 15 s later
set jobs {owner: "w3", run_at: now() + 30000, attempts: attempts + 1} where run_at <= now() order run_at limit 10 returning id, attempts   -- → 3: w2's lease lapsed
del jobs where id = 3 and owner = "w2" require 1                       -- → refused: 412, w3 holds it now
del jobs where id = 3 and owner = "w3" require 1                       -- → 1

• At least once. A job whose worker died is taken again once its lease lapses, so a job can run twice — once by a worker that was only slow. Work that must not happen twice keeps a check of its own: a @unique key on what it writes, or an Idempotency-Key on the request it sends.
• A token a claim rather than the worker's name as the owner — "w1:1842" — also tells apart two claims of one job by the same worker.
• To keep a job done rather than delete it, give it no ready time: set jobs {run_at: null, owner: null} where id = $1 and owner = $2 require 1. No claim takes a job without one.

Fail, back off, dead-letter

A job that failed goes back with a backoff that grows with its attempts, and after the fifth it has no ready time — a dead letter, which no claim takes and a read finds:

set jobs {owner: null, error: "smtp timeout", run_at: case when attempts >= 5 then null else now() + 1000 * attempts * attempts end} where id = 2 and owner = "w1" require 1   -- → 1: again in 1 s
-- 2 s later
set jobs {owner: "w1", run_at: now() + 30000, attempts: attempts + 1} where run_at <= now() order run_at limit 1 returning id, attempts   -- → 2: its second attempt
set jobs {attempts: 5} where id = 2                         -- three failures on
set jobs {owner: null, error: "smtp timeout", run_at: case when attempts >= 5 then null else now() + 1000 * attempts * attempts end} where id = 2 and owner = "w1" require 1   -- → 1: the fifth: dead
get jobs select id, error where run_at is null              -- → 2: the dead letters
set jobs {run_at: now(), attempts: 0, error: null} where id = 2 and run_at is null   -- → 1: sent again by hand

• The backoff here is the square of the attempts in seconds — 1, 4, 9, 16 — worked out over the row. One that doubles, or has jitter, is the worker's to send: it read attempts in its claim.
• A worker that dies on a job never fails it, and its lease lapses instead: attempts counts claims, so a job that comes back to a worker with attempts past five is failed by it, not run again — SQS's receive count.

Priorities

With a priority, a claim takes the highest first, and among jobs of one priority those written first:

create collection tasks (payload json, run_at timestamp @sorted, priority int @sorted, owner text, attempts int)
put tasks [{payload: {"n": 1}, run_at: now(), priority: 0, attempts: 0}, {payload: {"n": 2}, run_at: now(), priority: 9, attempts: 0}]
set tasks {owner: "w1", run_at: now() + 30000, attempts: attempts + 1} where run_at <= now() order priority desc limit 1 returning payload.n   -- → 2: the higher priority

• The walk is the priority's index, testing each job's ready time as it passes: jobs of a high priority that are leased or delayed are passed over at every claim, so a queue holding many of them is a collection of its own. While the ready jobs number a few thousand at most, the claim reads them by run_at and puts them in order instead.

The nearest job

Where a job has a place and a worker one of its own — a pickup and a courier, a ride and a driver — the claim takes the nearest ready job, in the same one statement: a key of order may be an expression, and the distance from a point is walked to the page by a @geo index (The nearest job):

create collection rides (pickup geo @geo, run_at timestamp @sorted, owner text, attempts int)
put rides [{pickup: [29.0100, 41.0400], run_at: now(), attempts: 0}, {pickup: [29.0300, 41.0500], run_at: now(), attempts: 0}]
set rides {owner: "w1", run_at: now() + 30000, attempts: attempts + 1} where run_at <= now() order distance(pickup, [29.0290, 41.0490]) limit 1 returning id   -- → 2: the nearer
set rides {owner: "w2", run_at: now() + 30000, attempts: attempts + 1} where run_at <= now() order distance(pickup, [29.0290, 41.0490]) limit 1 returning id   -- → 1: the one left

• Any expression orders the claim — order priority * 10 - attempts desc, order case when vip then 0 else 1 end, run_at — every ready job's key worked out and the page put in order. Only the distance from a point over @geo and an @sorted field are walked to the page.
• The nearest worker to each job, where they wait. A claim takes the job nearest the worker that asks; claims held at the server for a job (Fenec-Wait) hand a job placed to the waiting worker nearest it, each claim's own distance its rank (Who gets the job).

A pop

Where a job may be lost but never run twice — Redis's LPOP — a del takes and answers it at once:

del tasks where run_at <= now() order priority desc limit 10 returning id, payload   -- → 1: taken and gone

A worker

The loop, in JavaScript over HTTP: claim a batch — held at the server until there is one — run each job, and ack it or fail it.

import { connect, expr, inc, raw } from '@fenecdb/web/client';

const db = connect('http://127.0.0.1:8080', { token: 'a worker token' });
const jobs = db.from('jobs');
const me = `w-${crypto.randomUUID()}`;

async function send(to: unknown): Promise<void> {
  // the work
}

for (;;) {
  const owner = `${me}:${Date.now()}`;
  const claimed = await jobs
    .where(raw('run_at <= now()'))
    .order('run_at')
    .limit(10)
    .update(
      { owner, run_at: expr('now() + ?', 30000), attempts: inc(1) },
      { returning: true, wait: 30000 },
    );
  for (const job of claimed) {
    try {
      await send((job.payload as { to: string }).to);
      await jobs.where('id', job.id).where('owner', owner).delete({ require: 1 });
    } catch {
      const backoff = 1000 * 2 ** Number(job.attempts);
      await jobs.where('id', job.id).where('owner', owner).update(
        { owner: null, run_at: expr('now() + ?', backoff) },
        { require: 1 },
      ).catch(() => {});
    }
  }
}

• Every builder writes the claim — Python's .update(patch, returning=True), Go's Update(ctx, patch, fenecdb.Returning("*")) and its Result.Rows, .NET's UpdateAsync(patch, returning: ["*"]), and Swift's, Kotlin's and Dart's updateReturning — held to the same text.
• And waits for a job the client's way: JavaScript's { wait: 30000 } beside { signal }, Python's wait=30 (seconds), Go's db.Wait(30 * time.Second) under the request's context, .NET's db.WithWait(TimeSpan.FromSeconds(30)) under its cancellation token, and Swift's, Kotlin's and Dart's withWait. Each client waits for the answer as much longer than its own timeout, and gives the request up when it is cancelled.
• Over HTTP the claim is a POST /query; sent under an Idempotency-Key, a retry after a timeout is answered the jobs the first try took, and takes none.
• A worker's token claims through its rules: a policy line jobs update(owner, run_at, attempts) where queue = $jwt.queue for worker lets it take its queue's jobs alone and change nothing else of them, and what it answers is held to its read rules as well, as it writes them: a worker that reads only where owner = $jwt.sub is answered the jobs it claims. A transition says what a claim may make of a job — where run_at <= now() and new.owner = $jwt.sub and new.attempts = attempts + 1, so no claim is made for another worker or wipes the count a dead letter keeps (Scoped access).
• Not on a sync replica. A replica would pick its own rows and answer them, and the server its own: a claim there is refused, and goes to the server itself.

What it costs

Over a million jobs, make queue-bench:

• The three statements a client sends without returning — read the ready page, set the jobs only where they are still ready, read back which it won — lose a race whenever two workers read the same page: 43% of the jobs read in process, and 77% over HTTP, had gone to another worker by the time it wrote, and were read again. Half the jobs a second in process, a third over HTTP.
• Sixteen claims wait for each other, as every write does: a claim's p99 among sixteen threads is the lock's queue, where one worker's is 25.8 us.
• The claim reads its page and no more: the ready jobs are a range of the index, the time worked out once before the plan. A queue holding a million leased or delayed jobs and three ready claims the three, where a walk testing each job's time would pass every leased one.

Compared

• PostgreSQL: UPDATE jobs SET ... WHERE id IN (SELECT id FROM jobs WHERE run_at <= now() ORDER BY run_at LIMIT 10 FOR UPDATE SKIP LOCKED) RETURNING *. Its row locks let claims run side by side; here the one writer runs them one after another, each a few microseconds, and none has rows to skip.
• SQS: a claim is ReceiveMessage, the lease its visibility timeout, an extension ChangeMessageVisibility, attempts its receive count and a dead letter its redrive. A delayed job is DelaySeconds, with no fifteen-minute ceiling.
• BullMQ: delayed, prioritised, retried with a backoff and kept as failed, as here; its jobs live in Redis's memory, these in the file, durable by the server's --sync policy, replicated with it, and read with any query.

What this does not cover

• Exactly once. A lease that lapses under a slow worker hands its job to another; the work's own check makes the second run harmless.
• Claims side by side. They are writes, so each waits for the one before: tens of thousands a second, not millions.