venturi

Controlled flow from backlog to worker.

venturi is a durable, PostgreSQL-backed job queue for Rust, built to be shared across projects rather than reimplemented per codebase.

A venturi is the narrowed section of a pipe that turns built-up pressure into controlled, measurable flow. This library does the same for work: jobs accumulate safely in your database and are released to workers at a rate you control, with the durability and transactional guarantees Postgres already gives you.

Features

• Durable, at-least-once delivery on PostgreSQL, claimed with FOR UPDATE SKIP LOCKED so many workers contend without blocking.
• Typed tasks. A job is one serializable struct; the same struct is the payload, the dedup identity, and the unit your handler receives.
• Four outcomes from a run: complete, cooperative pause (checkpoint and resume), retryable failure, or permanent failure.
• Fibonacci backoff with deterministic, RNG-free jitter; a per-task or worker-level give-up policy.
• Deduplication with a candidacy key and a full merge decision over the existing job's payload, carry, run count, and journal.
• Reliability: per-claim leases with automatic stale-claim recovery, and cooperative graceful shutdown that drains then releases.
• Scheduling: three priority tiers with weighted-slot anti-starvation, per-kind concurrency caps, delayed/scheduled jobs, and LISTEN/NOTIFY wakeups with a polling fallback.
• Operations: an append-only per-execution journal, a history query, bulk cleanup, a live stats snapshot, tracing events, and optional metrics.

Installation

[dependencies]
venturi = "0.1"
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }

The default postgres feature enables the PostgreSQL adapter. The optional metrics feature emits through the vendor-neutral metrics facade. The optional rustls feature adds the connect_rustls TLS constructor.

Quick start

Define a task, implement its handler, then enqueue from a producer and process with a worker.

use serde::{Deserialize, Serialize};
use std::sync::Arc;
use tokio_util::sync::CancellationToken;
use venturi::postgres::PostgresStore;
use venturi::store::Store;
use venturi::{Context, Handler, Outcome, Queue, Task, TaskError, Worker};

#[derive(Serialize, Deserialize)]
struct SendEmail { to: String, subject: String }

// The producer side: identity and enqueue-time policy.
impl Task for SendEmail {
    const KIND: &'static str = "send_email";
    type Carry = ();
}

// The worker side: the execution logic against shared state `App`.
#[derive(Clone)]
struct App;

impl Handler<App> for SendEmail {
    async fn handle(&self, _ctx: &mut Context<()>, _app: &App) -> Result<Outcome, TaskError> {
        // `?` on any error retries with backoff; `TaskError::permanent(..)` gives up.
        println!("sending {:?} to {}", self.subject, self.to);
        Ok(Outcome::completed())
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let dsn = "host=localhost user=postgres password=postgres dbname=postgres";
    let store = Arc::new(PostgresStore::connect(dsn, "venturi")?);
    store.migrate().await?;

    // Produce.
    let queue = Queue::new(store.clone());
    queue.enqueue(SendEmail { to: "a@example.com".into(), subject: "Hi".into() }).await?;

    // Consume.
    let worker = Worker::builder(App, store).register::<SendEmail>().build();
    let shutdown = CancellationToken::new();
    worker.run(shutdown).await; // returns when `shutdown` is cancelled
    Ok(())
}

Runnable versions live in examples/: cargo run --example producer and cargo run --example worker (set DATABASE_URL).

Documentation

• A full walkthrough from first steps to advanced usage: docs/guide.md.
• API docs: cargo doc --open.

Development

The project uses just:

just ci                # fmt-check, clippy, type-check, fast tests
just integration-test  # database-backed tests (requires Docker)

Database-backed tests run against an ephemeral PostgreSQL container and are marked #[ignore], so the fast just test stays quick.

License

Licensed under the MIT License.