Distributed Task Queue

A PostgreSQL-backed distributed task queue library in Rust, providing at-least-once delivery with no additional infrastructure dependencies beyond Postgres. The at-least-once guarantee applies to successfully submitted tasks — if submit returns Ok, the task is durably persisted and will be delivered at least once.

• Priority-ordered dequeue — tasks with lower priority numbers are processed first (0 = highest, 9 = lowest)
• Concurrent task claiming — SELECT ... FOR UPDATE SKIP LOCKED ensures multiple consumers never block each other
• Visibility-timeout-based redelivery — crashed consumers' tasks are automatically reclaimed after timeout expiry; workers can call extend_visibility to prevent reclaim of long-running tasks
• Delayed/scheduled tasks — submit tasks with a future scheduled_at for deferred execution
• Exponential backoff retries — configurable per-task retry policy with backoff formula base * 2^retry_count, capped at 1 hour
• Dead letter queue — tasks exceeding max_retries are moved to a separate dead_letter_tasks table
• LISTEN/NOTIFY — optional low-latency wake-up via Postgres notifications (INSERT and NOTIFY are atomic within a single transaction), with automatic reconnection and poll fallback
• Batch enqueue/dequeue — amortize round-trip cost under high throughput
• Prometheus metrics — counters for enqueued/dequeued/completed/failed/retried/dead-lettered, gauge for queue depth, histogram for processing duration. The library provides a prometheus::Registry; the caller is responsible for exposing it (e.g., via an HTTP endpoint)
• Graceful shutdown — in-flight tasks are given a configurable grace period to complete before abort
• Horizontal scaling — multiple producers and consumers across processes

Producer ──enqueue──▶ PostgreSQL ◀──dequeue── Consumer(s)
                        │                        │
                   NOTIFY channel ──────▶ NotificationListener

The NotificationListener is an internal component that maintains a dedicated PostgreSQL connection (separate from the connection pool) for LISTEN/NOTIFY. It collapses incoming notifications into a single wake-up signal for the consumer's dequeue loop and reconnects automatically with exponential backoff if the connection drops. When disabled (notify_pg_url = None), the consumer falls back to poll-only mode.

Single tasks table with partial indexes keeps the dequeue path fast regardless of total row count. A background archival process deletes completed tasks beyond a configurable retention period. Aggressive autovacuum tuning handles the high UPDATE churn.

See design/distributed-task-queue.md for the full design document including decision log, schema, SQL query patterns, edge cases, and known limitations.

• Rust 1.70+
• PostgreSQL 14+ (uses gen_random_uuid())
• Docker (for local development)

docker compose up -d

Run the SQL files in migrations/ against your database in order (002 depends on tables created by 001; 002 is idempotent on fresh installs):

psql postgres://taskqueue:taskqueue@localhost:5432/taskqueue \
  -f migrations/001_init.sql \
  -f migrations/002_add_check_constraints.sql

use distributed_task_queue::*;
use sqlx::postgres::PgPoolOptions;
use std::sync::Arc;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let pool = PgPoolOptions::new()
        .max_connections(10)
        .connect("postgres://taskqueue:taskqueue@localhost:5432/taskqueue")
        .await?;

    let metrics = MetricsCollector::new();
    let store = TaskStore::new(pool.clone(), metrics.clone());

    // --- Producer ---
    let producer = Producer::new(store.clone());
    let task_id = producer
        .submit(TaskSubmission {
            queue_name: "emails".to_string(),
            payload: serde_json::json!({"to": "user@example.com", "subject": "Hello"}),
            priority: 3,
            max_retries: 5,
            visibility_timeout_secs: 60,
            // Defaults: retry_backoff_base_secs = 1, delay_secs = 0
            ..Default::default()
        })
        .await?;
    println!("Submitted task: {task_id}");

    // --- Consumer ---
    let config = ConsumerConfig {
        queue_name: "emails".to_string(),
        worker_id: "worker-1".to_string(),
        concurrency: 10,
        poll_interval_secs: 5,
        notify_pg_url: Some("postgres://taskqueue:taskqueue@localhost:5432/taskqueue".to_string()),
        // Defaults: batch_size = 10, archival_retention_secs = 3600,
        //           shutdown_grace_period_secs = 30
        ..Default::default()
    };
    let consumer = Arc::new(Consumer::new(store, config, metrics)?);

    // Trigger shutdown on Ctrl-C.
    let consumer_handle = consumer.clone();
    tokio::spawn(async move {
        tokio::signal::ctrl_c().await.ok();
        consumer_handle.shutdown(); // synchronous, signals the run loop to stop
    });

    consumer
        .run(|task| async move {
            println!("Processing task {}: {:?}", task.id, task.payload);
            // Your processing logic here.
            Ok(())
        })
        .await?;

    Ok(())
}

# Start Postgres
docker compose up -d

# Run integration + e2e tests
cargo test

# Run performance benchmarks (ignored by default)
cargo test --test perf_tests -- --ignored --nocapture

44 integration tests, 7 end-to-end tests, and 6 performance benchmarks.

The design document was created using the sisyphus-design skill — a structured approach to producing detailed, implementation-ready software designs with explicit decision logs, edge case analysis, and known limitations.

The code was implemented by Claude Code (dev) and reviewed/fixed by Codex (review).

Prompt used to generate the initial implementation:

Implement the distributed task queue system specified in design/distributed-task-queue.md. That file is the complete spec — follow it exactly.

Use Docker to provision PostgreSQL for tests. Include full unit tests (against real PG, not mocks) and end-to-end tests covering concurrency with multiple consumers.

Plan first — identify what can be parallelized, then execute. Use subagents (Agent tool, model opus) to build independent modules simultaneously. Don't ask for approval on the plan, just go.

Prompt used for the review-fix loop:

Run codex review --uncommitted using a subagent (Agent tool, model sonnet). The subagent should parse the output and return only the identified issues (severity + description) — discard all intermediate/diagnostic output.

Then loop: fix all P1/High issues, re-run the review via subagent, and repeat until the review reports no P1 or High severity issues remaining.

The implementation follows the design closely but deviates in several places — some are intentional improvements, some are pragmatic API changes, and one is a missing feature.

MIT