Job Queues With SKIP LOCKED in Postgres From Rust Explained
Build a reliable job queue in a single table: claim rows with FOR UPDATE SKIP LOCKED, mark them done in the same transaction, retry failures with backoff, and know where this stops scaling.
Before reaching for a message broker, consider that you already run a transactional database. For many services, a table of jobs and one SQL clause give a queue with exactly-once claiming, durable storage and visibility with ordinary SQL. The clause is FOR UPDATE SKIP LOCKED. This post builds a worker with it. Versions: Rust 1.90 with edition 2024, SQLx 0.8, PostgreSQL 17, Tokio 1.47.
The table
CREATE TABLE jobs (
id BIGSERIAL PRIMARY KEY,
kind TEXT NOT NULL,
payload JSONB NOT NULL,
status TEXT NOT NULL DEFAULT 'queued', -- queued | done | dead
attempts INT NOT NULL DEFAULT 0,
max_tries INT NOT NULL DEFAULT 5,
run_at TIMESTAMPTZ NOT NULL DEFAULT now(),
last_error TEXT,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX jobs_ready_idx ON jobs (run_at) WHERE status = 'queued';The partial index covers only queued rows, so it stays small as completed jobs accumulate. Delete or archive finished jobs regularly (partitioning by day makes that cheap).
Claiming a job
SELECT id, kind, payload, attempts
FROM jobs
WHERE status = 'queued' AND run_at <= now()
ORDER BY run_at
LIMIT 1
FOR UPDATE SKIP LOCKED;FOR UPDATE locks the selected row for the transaction. SKIP LOCKED tells other workers running the same query to ignore locked rows and take the next one, instead of waiting. Ten workers polling at once each get a different job, with no coordination code.
The worker loop
use sqlx::{PgPool, Row};
use std::time::Duration;
pub async fn work_one(pool: &PgPool) -> Result<bool, sqlx::Error> {
let mut tx = pool.begin().await?;
let Some(row) = sqlx::query(
"SELECT id, kind, payload, attempts, max_tries FROM jobs
WHERE status = 'queued' AND run_at <= now()
ORDER BY run_at LIMIT 1 FOR UPDATE SKIP LOCKED",
)
.fetch_optional(&mut *tx)
.await?
else {
return Ok(false); // queue empty
};
let id: i64 = row.get("id");
let kind: String = row.get("kind");
let payload: serde_json::Value = row.get("payload");
let attempts: i32 = row.get("attempts");
let max_tries: i32 = row.get("max_tries");
match handle(&kind, &payload, &mut tx).await {
Ok(()) => {
sqlx::query("UPDATE jobs SET status = 'done', attempts = attempts + 1 WHERE id = $1")
.bind(id)
.execute(&mut *tx)
.await?;
}
Err(e) => {
let next = attempts + 1;
if next >= max_tries {
sqlx::query("UPDATE jobs SET status = 'dead', attempts = $2, last_error = $3 WHERE id = $1")
.bind(id).bind(next).bind(e.to_string()).execute(&mut *tx).await?;
} else {
// exponential backoff: 2^attempt * 10 seconds
sqlx::query(
"UPDATE jobs SET attempts = $2, last_error = $3,
run_at = now() + make_interval(secs => $4) WHERE id = $1",
)
.bind(id).bind(next).bind(e.to_string()).bind(10.0 * 2f64.powi(next))
.execute(&mut *tx)
.await?;
}
}
}
tx.commit().await?;
Ok(true)
}
async fn handle(
_kind: &str,
_payload: &serde_json::Value,
_tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
Ok(())
}
pub async fn run(pool: PgPool) {
loop {
match work_one(&pool).await {
Ok(true) => continue, // more work: no sleep
Ok(false) => tokio::time::sleep(Duration::from_secs(1)).await,
Err(e) => {
tracing::error!(error = %e, "worker error");
tokio::time::sleep(Duration::from_secs(5)).await;
}
}
}
}If the process crashes during handle, the transaction rolls back, the row's lock is released and another worker claims the job. Database effects performed through tx in the handler commit atomically with the status change, which gives you exactly-once for database work. A job that touches the outside world (an email, an HTTP call) can run twice if the crash comes after the call and before the commit, so make those calls idempotent (idempotency keys).
Waking workers promptly
Polling every second adds up to a second of latency. Add NOTIFY jobs_ready after inserting a job and have workers LISTEN on a dedicated connection, keeping the poll as a safety net (LISTEN and NOTIFY).
Limits
- Open transactions: a job running for minutes pins a transaction. Use the lease pattern for long jobs.
- Table churn: every attempt updates a row; dead tuples need vacuum, and a very hot queue can bloat. Tune autovacuum for the table.
- Throughput: thousands per second on decent hardware is realistic; beyond that, contention on the index grows.
- Fan-out and ordering: no consumer groups, replay or strict ordering across workers.
If you outgrow it, move to a purpose-built queue; the migration is easy when the handler is already a function of a payload (background job queues compared).
