pg-queue

A PostgreSQL-based job queue, pub/sub, and cache library for Rust. One dependency instead of two — if you already run Postgres, you don't need Redis.

What it does

Most applications that outgrow a single process need three things: a way to queue work for later, a way to broadcast events in real time, and a way to cache expensive lookups. The standard answer is to bolt Redis onto your stack. pg-queue provides all three using features already built into PostgreSQL — no additional infrastructure to deploy, monitor, or pay for.

In plain terms: instead of adding a separate Redis server to handle background jobs and live updates, pg-queue lets your existing Postgres database do it. Fewer moving parts, fewer things to break at 3am.

How it works

SKIP LOCKED lets multiple workers grab jobs concurrently without blocking each other. LISTEN/NOTIFY pushes events to subscribers the instant a row is inserted — no polling. UNLOGGED tables skip write-ahead logging for cache-tier speed (data is not crash-safe, which is fine for a cache).

Setup

Run the migration against your database:

psql -f migrations/setup.sql your_database

Then create as many queues as you need:

SELECT pg_queue_create_queue('emails');
SELECT pg_queue_create_queue('image_processing');

Each call creates a dedicated table (queue_emails, queue_image_processing) with a partial index on pending jobs and a trigger that fires NOTIFY on insert.

Add to your Cargo.toml:

[dependencies]
pg-queue = "0.1"

Usage

Job queue

use pg_queue::{PgQueueManager, QueueName};

let pool = sqlx::PgPool::connect("postgres://localhost/mydb").await?;
let mgr = PgQueueManager::new(pool);
let emails = QueueName::new("emails")?;

// Producer: enqueue a job
mgr.queue.push(&emails, &serde_json::json!({
    "to": "user@example.com",
    "template": "welcome"
})).await?;

// Consumer: claim and process
if let Some(job) = mgr.queue.pop::<serde_json::Value>(&emails).await? {
    // process job.payload ...
    mgr.queue.complete(&emails, job.id).await?;
}

Multiple consumers can call pop concurrently — SKIP LOCKED ensures each job is claimed by exactly one worker.

Pub/Sub

// Publisher
mgr.notify.notify("events.user_signup", r#"{"user_id": 42}"#).await?;

// Subscriber
let mut listener = mgr.create_listener().await?;
listener.listen("events.user_signup").await?;

loop {
    if let Some(msg) = listener.recv_timeout(Duration::from_secs(5)).await? {
        println!("{}: {}", msg.channel, msg.payload);
    }
}

Cache

// Set with 5 minute TTL
mgr.cache.set("user:42:profile", &profile_data, 300).await?;

// Get (returns None if expired or missing)
let cached: Option<Profile> = mgr.cache.get("user:42:profile").await?;

// Get-or-set with fallback
let profile = mgr.cache.get_or_set("user:42:profile", 300, || async {
    fetch_profile_from_db(42).await
}).await?;

Request-Response (RPC)

let workers = QueueName::new("rpc_workers")?;

// Caller: push request and block until response
let result: MyResponse = mgr.request_response
    .push_and_wait(&workers, &my_request, Duration::from_secs(10))
    .await?;

// Worker: process and respond
if let Some(job) = mgr.queue.pop::<RequestWrapper>(&workers).await? {
    let response = handle(job.payload);
    mgr.request_response.store_response(&job.payload.request_id, &response).await?;
    mgr.queue.complete(&workers, job.id).await?;
}

Use cases

• Background job processing — email delivery, image resizing, PDF generation, webhook dispatch. Any work that should happen outside the request cycle.
• Event-driven microservices — publish domain events (order placed, user signed up) and let downstream services react via LISTEN/NOTIFY without polling.
• Caching hot data — store frequently-read, expensive-to-compute results (API responses, aggregations, session data) with automatic TTL expiry.
• Task orchestration — fan-out work to a pool of workers with guaranteed exactly-once delivery via SKIP LOCKED.
• Reducing infrastructure — if your Redis is only doing job queues and caching, pg-queue lets you drop it entirely and run everything on the Postgres you already have.

Tradeoffs

This is the right tool when your job volume fits comfortably within PostgreSQL's throughput (tens of thousands of jobs per second on modern hardware). If you need millions of messages per second or global pub/sub across data centres, dedicated message brokers like Kafka or NATS are better suited.

The cache uses UNLOGGED tables — faster writes, but data is lost on crash. That's the correct tradeoff for a cache (the source of truth lives elsewhere), but don't store anything you can't recompute.

License

MIT OR Apache-2.0