Durable job queue

Deep dive

How it works

Parts:

• log — append-only record of ENQ, ACK, RETRY, DEAD; fsync'd in batches
• job table — every live job with attempts and run_at
• ready heap — jobs ordered by run_at
• lease table and deadline heap — in-memory lease token per claimed job, never logged
• dead-letter queue — parking place for jobs past max_attempts

Operations: enqueue, claim, ack, fail or lease expiry, recovery (steps below).

State per job: ready(run_at), leased(token, deadline), done, dead.

1. Enqueue: append ENQ(id, run_at, payload) to the log, fsync (batched), then reply to the producer.
2. Claim: pop the earliest job with run_at <= now, give it a fresh lease token and deadline now + visibility. Not logged.
3. Ack(id, token): if the token matches the current lease, append ACK(id) and drop the job. A stale token (lease expired and re-granted) is rejected: the token is a fencing token.
4. Fail / lease expiry: attempts += 1; if attempts >= max, append DEAD(id) and move it to the dead-letter queue; else append RETRY(id, now + backoff(attempts)). Counting expiries as attempts is what stops a job that crashes its worker from looping forever.
5. Recovery: replay the log, truncate a torn tail, and mark every live job ready. Leases held at crash time are gone, so those jobs run again.

producer ─ENQ─► [log | fsync] ─► ready heap ─claim─► worker (lease token 7, deadline t+30s)
                                     ▲                    │ ack(7) ─► ACK ─► gone
                 RETRY(run_at+2^k·b) ┤◄── fail / expired ─┘
                                     └── attempts ≥ max ─► DEAD ─► dead-letter queue

In a relational database the same machine is one table plus SELECT ... FOR UPDATE SKIP LOCKED:

UPDATE jobs SET state = 'leased', attempts = attempts + 1, lease_until = now() + interval '30 s'
WHERE id IN (SELECT id FROM jobs WHERE state = 'ready' AND run_at <= now()
             ORDER BY run_at LIMIT 10 FOR UPDATE SKIP LOCKED)
RETURNING id, payload;

Code

A single-node queue over an append-only log file, std only. now is passed in (milliseconds) so tests are deterministic.

use std::{cmp::Reverse, collections::{BinaryHeap, HashMap}};
use std::fs::{File, OpenOptions};
use std::io::{self, Read, Write};

const ENQ: u8 = 1; const ACK: u8 = 2; const RETRY: u8 = 3; const DEAD: u8 = 4;
const HDR: usize = 25; // tag u8 | id u64 | run_at u64 | attempts u32 | len u32, then payload

struct Job { payload: Vec<u8>, attempts: u32, run_at: u64 }

pub struct Queue {
    log: File,
    buf: Vec<u8>,                                     // records not yet fsync'd
    jobs: HashMap<u64, Job>,                          // every live job
    ready: BinaryHeap<Reverse<(u64, u64)>>,           // (run_at, id); stale entries skipped
    leased: HashMap<u64, u64>,                        // id -> lease token
    deadlines: BinaryHeap<Reverse<(u64, u64, u64)>>,  // (deadline, id, token)
    next_id: u64, next_token: u64, max_attempts: u32, base_backoff: u64,
    pub dead: Vec<(u64, Vec<u8>)>,
}

impl Queue {
    pub fn open(path: &str, max_attempts: u32, base_backoff: u64) -> io::Result<Queue> {
        let mut log = OpenOptions::new().create(true).read(true).append(true).open(path)?;
        let (mut data, mut off) = (Vec::new(), 0); log.read_to_end(&mut data)?;
        let mut q = Queue { log: log.try_clone()?, buf: Vec::new(), jobs: HashMap::new(),
            ready: BinaryHeap::new(), leased: HashMap::new(), deadlines: BinaryHeap::new(),
            next_id: 1, next_token: 1, max_attempts, base_backoff, dead: Vec::new() };
        let (u64at, u32at) = (|d: &[u8], o: usize| u64::from_le_bytes(d[o..o + 8].try_into().unwrap()),
                              |d: &[u8], o: usize| u32::from_le_bytes(d[o..o + 4].try_into().unwrap()));
        while off + HDR <= data.len() {
            let (tag, id, run_at) = (data[off], u64at(&data, off + 1), u64at(&data, off + 9));
            let (attempts, len) = (u32at(&data, off + 17), u32at(&data, off + 21) as usize);
            if off + HDR + len > data.len() { break; } // torn tail from a crash mid-write
            let payload = data[off + HDR..off + HDR + len].to_vec();
            match tag {
                ENQ => { q.jobs.insert(id, Job { payload, attempts, run_at }); q.next_id = q.next_id.max(id + 1); }
                RETRY => if let Some(j) = q.jobs.get_mut(&id) { j.run_at = run_at; j.attempts = attempts; },
                ACK => { q.jobs.remove(&id); }
                DEAD => if let Some(j) = q.jobs.remove(&id) { q.dead.push((id, j.payload)); },
                _ => break,
            }
            off += HDR + len;
        }
        log.set_len(off as u64)?; // cut the torn tail so new appends start on a record boundary
        for (id, j) in &q.jobs { q.ready.push(Reverse((j.run_at, *id))); } // leases die with the process
        Ok(q)
    }

    fn record(&mut self, tag: u8, id: u64, run_at: u64, attempts: u32, payload: &[u8]) {
        self.buf.push(tag);
        for f in [&id.to_le_bytes()[..], &run_at.to_le_bytes(), &attempts.to_le_bytes(),
                  &(payload.len() as u32).to_le_bytes(), payload] { self.buf.extend_from_slice(f); }
    }

    /// One write + one fsync for everything since the last commit (group commit).
    /// Reply "enqueued" to producers only after this returns.
    pub fn commit(&mut self) -> io::Result<()> {
        if self.buf.is_empty() { return Ok(()); }
        self.log.write_all(&self.buf)?;
        self.log.sync_data()?;
        self.buf.clear(); Ok(())
    }

    pub fn enqueue(&mut self, payload: &[u8], run_at: u64) -> u64 {
        let id = self.next_id; self.next_id += 1;
        self.record(ENQ, id, run_at, 0, payload);
        self.jobs.insert(id, Job { payload: payload.to_vec(), attempts: 0, run_at });
        self.ready.push(Reverse((run_at, id))); id
    }

    /// Lease the earliest due job for `visibility` ms. Not logged: after a crash it simply runs again.
    pub fn claim(&mut self, now: u64, visibility: u64) -> Option<(u64, u64, Vec<u8>)> {
        self.reap(now);
        while let Some(&Reverse((run_at, id))) = self.ready.peek() {
            if run_at > now { return None; } self.ready.pop();
            let live = self.jobs.get(&id).map_or(false, |j| j.run_at == run_at);
            if !live || self.leased.contains_key(&id) { continue; } // defensive: stale heap entry
            let token = self.next_token; self.next_token += 1;
            self.leased.insert(id, token);
            self.deadlines.push(Reverse((now + visibility, id, token)));
            return Some((id, token, self.jobs[&id].payload.clone()));
        }
        None
    }

    /// Token check = fencing: a worker whose lease expired cannot ack the re-leased job.
    pub fn ack(&mut self, id: u64, token: u64) -> bool {
        if self.leased.get(&id) != Some(&token) { return false; }
        self.leased.remove(&id); self.jobs.remove(&id);
        self.record(ACK, id, 0, 0, &[]); true
    }

    pub fn fail(&mut self, id: u64, token: u64, now: u64) -> bool {
        if self.leased.get(&id) != Some(&token) { return false; }
        self.leased.remove(&id); self.retry_or_dead(id, now); true
    }

    /// Expired leases count as failed attempts, so a job that crashes its worker ends up dead.
    pub fn reap(&mut self, now: u64) {
        while let Some(&Reverse((deadline, id, token))) = self.deadlines.peek() {
            if deadline > now { break; } self.deadlines.pop();
            if self.leased.get(&id) == Some(&token) { self.leased.remove(&id); self.retry_or_dead(id, now); }
        }
    }

    fn retry_or_dead(&mut self, id: u64, now: u64) {
        let j = self.jobs.get_mut(&id).unwrap();
        j.attempts += 1; let attempts = j.attempts;
        if attempts >= self.max_attempts {
            let j = self.jobs.remove(&id).unwrap();
            self.record(DEAD, id, 0, attempts, &[]);
            self.dead.push((id, j.payload));
        } else {
            j.run_at = now + (self.base_backoff << (attempts - 1).min(20)); // add jitter in production
            let run_at = j.run_at;
            self.record(RETRY, id, run_at, attempts, &[]);
            self.ready.push(Reverse((run_at, id)));
        }
    }
}

Complexity and constants

• Enqueue / claim / ack: O(log n) heap work plus a 25-byte header (+ payload) appended to a memory buffer.
• The cost driver is fsync. Measured with this code on an Apple-silicon laptop (Rust's sync_data on macOS issues F_FULLFSYNC): about 4 ms per commit, so 200 single-job commits took 0.8 s, while 10,000 enqueues in one commit took about 5 ms in total. Data-center NVMe with power-loss protection is order of tens of µs to 1 ms per fsync; batching still wins by orders of magnitude.
• Claims cost no I/O, because leases are not logged. Acks need not be fsync'd before replying to the worker: a lost ack only re-runs the job.
• Memory: one map entry and one heap entry per live job, order of 100 bytes plus payload. Keep large payloads in blob storage and enqueue a reference.
• Postgres queues: each claim and ack is an UPDATE or DELETE that leaves a dead row version, so throughput is bounded by WAL writes and vacuum; batch claims and acks to cut per-job cost.

Efficiency

• Cost: enqueue, claim and ack are O(log n) heap work plus a 25-byte header appended to a memory buffer. The cost driver is fsync: measured with the code above on an Apple-silicon laptop (F_FULLFSYNC), about 4 ms per commit. Memory is order of 100 bytes per live job plus payload. Bound: I/O (fsync) latency.
• Levers:
  • group commit — one write and one fsync per batch, orders of magnitude fewer fsyncs (measured above) — any durable queue
  • unlogged leases — claims cost no I/O; a lost lease only re-runs the job — in-memory lease state
  • batch claims and acks (10-100 per RPC or LIMIT) — fewer round trips, fewer dead row versions and WAL writes on Postgres — SQL queues
  • payload in blob storage, reference in the queue — small log records and heap entries — large payloads
  • log compaction when dead bytes exceed live bytes — bounded log and replay time — long-running queues
• Naive → standard → state of the art:
  • fsync per job · about 4 ms each on the laptop above
  • group commit, unlogged leases · about 5 ms per 10,000 enqueues
  • group commit plus batched claims and acks, hierarchical timing wheel for very many timers · O(1) timer insert (see Implementation notes)

Implementation notes

• Commit loop: one thread drains requests for a few hundred µs or until N records, calls commit() once, then answers all waiting producers (group commit).
• Log compaction: the log grows forever. Periodically write only the live jobs to a new file, fsync it, rename it over the old one, fsync the directory. Trigger when dead bytes exceed live bytes.
• Checksums: add a CRC32C per record. Without it a torn or corrupted record in the middle is only detected by luck; the length check above handles only a torn tail.
• Backoff: exponential with full jitter (exponential backoff with jitter); without jitter, a dependency outage releases every retried job at the same instant.
• Visibility timeout: set to the p99 run time with headroom; long jobs extend their lease with heartbeats (SQS ChangeMessageVisibility), keyed by the same token.
• Batch claims: hand out 10-100 jobs per claim RPC or LIMIT to cut round trips; with SKIP LOCKED this also keeps workers off each other's rows.
• Idempotent handlers: an idempotency key (the job id) checked by the side effect's target, or the lease token passed as a fencing token, turns at-least-once into effectively-once.
• Enqueue inside the business transaction: with a SQL queue, insert the job in the same transaction as the data change; otherwise use a transactional outbox.
• Delayed jobs: a heap keyed by run_at is fine to millions; with very many timers, a hierarchical timing wheel gives O(1) insert.

Pitfalls

• Acking on receipt instead of after the work (Celery's default without task_acks_late): a worker crash loses the job.
• Visibility timeout shorter than the job: two workers run it concurrently, and the slow one's ack must be rejected by the token check.
• Expired leases not counted as attempts: a payload that segfaults the worker cycles forever.
• Long-running Postgres transactions block vacuum; dead job rows pile up and claim latency grows until the queue runs away (Brandur's "Failure by MVCC").
• Polling with SELECT ... FOR UPDATE without SKIP LOCKED: all workers queue on the same head row.
• Replying to the producer before fsync: a crash loses jobs the producer believes were accepted.
• Unbounded retry with no dead-letter queue hides bugs and burns capacity.

Variants and alternatives

In production

E-mail, webhook and billing back ends: Sidekiq (Redis; 25 retries then Dead set), Celery (RabbitMQ/Redis brokers), Rails Solid Queue, Faktory. Cloud messaging: Amazon SQS (visibility timeout, redrive to DLQ). Postgres-based queues: Oban and River (SKIP LOCKED), pgmq (Postgres extension with SQS-like visibility timeouts).

Implementations

Sources

• PostgreSQL documentation, SELECT: The Locking Clause (SKIP LOCKED for queue-like tables). docs
• AWS, Amazon SQS visibility timeout, SQS Developer Guide. docs
• Redis documentation, LMOVE, "Pattern: reliable queue". docs
• Sidekiq wiki, Error Handling (retry schedule, Dead set). wiki
• Brandur Leach, Postgres Job Queues & Failure By MVCC, 2015. post
• Brandur Leach, Transactionally Staged Job Drains in Postgres, 2017. post
• Marc Brooker, Exponential Backoff And Jitter, AWS Architecture Blog 2015. post