Job Queues on the Database with SKIP LOCKED

Scenario

Example app: an invoicing system stores email jobs in a jobs table. Three worker processes poll every second, take the oldest pending job, send it, and mark it done. Volume: ~20,000 invoices on the first of each month.

What you see in production

Customers receive two or three copies of the same invoice email. Logs show different workers picking the same job_id within milliseconds. Adding FOR UPDATE stops the duplicates but the workers now crawl: two of three are always waiting.

The naive code

while True:
    with Session(engine) as s, s.begin():
        job = s.execute(select(Job).where(Job.status == "pending")
                        .order_by(Job.id).limit(1)).scalar_one_or_none()   # every worker gets the same row
        if job:
            send_email(job.payload)
            job.status = "done"
    time.sleep(1)

$job = DB::table('jobs')->where('status', 'pending')->orderBy('id')->first();   // no lock
Mail::send(...);
DB::table('jobs')->where('id', $job->id)->update(['status' => 'done']);

Reproduce it

Three workers poll at the same moment. Compare a plain read, FOR UPDATE (waits) and FOR UPDATE SKIP LOCKED (skips rows other workers hold):

def poll(mode, workers=("w1", "w2", "w3"), jobs=(1, 2, 3, 4, 5)):
    locked, result = {}, {}
    for w in workers:                                   # all poll in the same instant
        if mode == "plain":
            result[w] = "job 1"                         # no lock: everyone sees job 1 as pending
        elif mode == "for update":
            if 1 in locked:
                result[w] = f"waits for job 1 ({locked[1]} holds it)"
            else:
                locked[1] = w
                result[w] = "job 1"
        else:                                           # FOR UPDATE SKIP LOCKED
            job = next(j for j in jobs if j not in locked)
            locked[job] = w
            result[w] = f"job {job}"
    return result

for mode in ("plain", "for update", "skip locked"):
    shown = ", ".join(f"{w}->{r}" for w, r in poll(mode).items())
    print(f"{mode:12} {shown}")

plain        w1->job 1, w2->job 1, w3->job 1
for update   w1->job 1, w2->waits for job 1 (w1 holds it), w3->waits for job 1 (w1 holds it)
skip locked  w1->job 1, w2->job 2, w3->job 3

Root cause

A plain SELECT doesn't lock, so every worker sees the same oldest job. FOR UPDATE fixes the duplicates but forces the other workers to wait for the same row. FOR UPDATE SKIP LOCKED tells the database: give me the first rows nobody else has locked — each worker immediately gets a different job.

The fix

-- claim one job (in a short transaction), then process it
UPDATE jobs SET status = 'running', locked_by = $1, locked_at = now()
WHERE id = (
  SELECT id FROM jobs
  WHERE status = 'pending' AND run_at <= now()
  ORDER BY id
  FOR UPDATE SKIP LOCKED
  LIMIT 1
)
RETURNING id, payload;            -- MySQL: SELECT … FOR UPDATE SKIP LOCKED, then UPDATE by id

with Session(engine) as s, s.begin():
    job = s.execute(select(Job).where(Job.status == "pending", Job.run_at <= func.now())
                    .order_by(Job.id).limit(1).with_for_update(skip_locked=True)).scalar_one_or_none()
    if job:
        job.status, job.locked_at = "running", func.now()
if job:
    send_email(job.payload)                       # outside the claim transaction
    mark_done(job.id)

@Transactional
public Optional<Job> claim(String worker) {
    String sql = "UPDATE jobs SET status = 'running', locked_by = ?, locked_at = now() "
        + "WHERE id = (SELECT id FROM jobs WHERE status = 'pending' ORDER BY id "
        + "FOR UPDATE SKIP LOCKED LIMIT 1) RETURNING id, payload";
    return jdbc.query(sql, JOB_MAPPER, worker).stream().findFirst();
}

$job = DB::transaction(function () {
    $job = DB::table('jobs')->where('status', 'pending')->orderBy('id')
        ->lock('for update skip locked')->first();
    if ($job) DB::table('jobs')->where('id', $job->id)->update(['status' => 'running', 'locked_at' => now()]);
    return $job;
});

const [job] = await prisma.$queryRaw<Job[]>`
  UPDATE jobs SET status = 'running', locked_at = now()
  WHERE id = (SELECT id FROM jobs WHERE status = 'pending' ORDER BY id FOR UPDATE SKIP LOCKED LIMIT 1)
  RETURNING id, payload`;

Claiming in a short transaction and processing outside it keeps locks brief. A sweeper resets jobs stuck in running past a lease (worker crashed), and the email send uses the job id as an idempotency key so a re-run is harmless. Production-grade libraries built on this pattern include Solid Queue for Rails (source).

Trade-offs

• Great for work queues, where any worker may take any item and order is approximate.
• Wrong for seat maps and inventory: a customer asking for seat F7 must wait for (or be refused) F7, not silently get "nothing available" because F7 was momentarily locked.
• Strict ordering is impossible with several workers; if order matters per customer, partition by key.
• Very high throughput (tens of thousands of jobs/s) with heavy update churn creates table bloat — a dedicated broker may fit better (see Queues & Background Processing).

Detect it in production

• Duplicate side effects keyed by job id (log "sending invoice for job 123" and count repeats).
• Queue metrics: pending count, age of the oldest pending job, jobs stuck in running beyond the lease.
• Index check: (status, run_at, id) or a partial index WHERE status = 'pending' keeps the claim query fast.

Key points

• Plain reads let every worker take the same job; FOR UPDATE serializes the workers.
• FOR UPDATE SKIP LOCKED gives each worker a different unlocked row — ideal for job queues.
• Claim briefly, process outside the transaction, reclaim expired leases, make processing idempotent.
• Don't use SKIP LOCKED where the user wants a specific row.