awa-pg 0.5.7

Postgres-native background job queue — Python SDK with async/sync workers, transactional enqueue, progress tracking, and web UI

Awa

Postgres-native background job queue for Rust and Python.

Awa (Māori: river) provides durable, transactional job enqueueing with typed handlers in both Rust and Python. All queue state lives in Postgres — no Redis, no RabbitMQ. The Rust runtime handles polling, heartbeating, crash recovery, and dispatch. Python workers run on that same runtime via PyO3, getting Rust-grade reliability with Python-native ergonomics.

Features

• Postgres-only — one dependency you already have.
• Transactional enqueue — insert jobs inside your business transaction. Commit = visible. Rollback = gone.
• Cancel by unique key — cancel scheduled jobs by their insert-time components (kind + args) without storing job IDs.
• Rust and Python workers — same queues, identical semantics, mixed deployments.
• Crash recovery — heartbeat + hard deadline rescue. Stale jobs recovered automatically.
• Web UI — dashboard, job inspector, queue management, cron controls.
• Structured progress — handlers report percent, message, and checkpoint metadata; persisted across retries.
• Periodic/cron jobs — leader-elected scheduler with timezone support and atomic enqueue.
• Storage transition prep — explicit status/prepare/abort surfaces for future storage-engine upgrades without changing the current canonical runtime.
• Webhook callbacks — park jobs for external completion with optional CEL expression filtering.
• Sequential callbacks — wait_for_callback() suspends a handler mid-execution; resume_external() wakes it with a payload. Enables multi-step orchestration within a single handler.
• HTTP Worker — feature-gated Worker that dispatches jobs to serverless functions (Lambda, Cloud Run) via HTTP with HMAC-blake3 callback auth.
• LISTEN/NOTIFY wakeup — sub-10ms pickup latency.
• Production alerting metrics — queue depth, lag, and wait-duration histogram via OpenTelemetry.
• OpenTelemetry — 20+ built-in metrics (counters, histograms, gauges) for Prometheus/Grafana.
• Hot/cold storage — runnable work in a hot table, deferred work in a cold table.
• Rate limiting — per-queue token bucket.
• Weighted concurrency — global worker pool with per-queue guarantees.
• Operator descriptors — code-declared queue and job-kind names/descriptions with stale/drift visibility in the UI.

Local benchmarks show ~5.6k jobs/sec sustained throughput (Rust workers, with OTel metrics enabled), ~3.1k jobs/sec (Python workers), and sub-10ms p50 pickup latency. Enqueue throughput reaches ~30k/s single-producer, ~100k/s multi-producer. See benchmarking notes for methodology and caveats.

Core concurrency invariants (no duplicate processing after rescue, stale completions rejected, shutdown drain ordering) are checked with TLA+ models covering single and multi-instance deployments.

Getting Started

The quickest way to reason about Awa is:

• your app inserts a durable job row inside Postgres, often in the same transaction as business data
• workers claim runnable rows, heartbeat while they are executing, and safely rescue stale work after crashes
• the job row itself is the source of truth for state, progress, callbacks, and retry history
• operators inspect and control that state through the CLI or the built-in UI

If you keep that model in mind, the APIs make more sense: enqueue work, run workers, then inspect the resulting row rather than guessing what happened in memory.

# 1. Install
pip install awa-pg awa-cli     # Python
# or: cargo add awa             # Rust

# 2. Start Postgres and run migrations
awa --database-url $DATABASE_URL migrate

# 3. Write a worker and start processing (see examples below)

# 4. Monitor
awa --database-url $DATABASE_URL serve   # → http://127.0.0.1:3000
awa --database-url $DATABASE_URL storage status
awa --database-url $DATABASE_URL job dump 123
awa --database-url $DATABASE_URL job dump-run 123

Language-specific guides:

• Rust getting started
• Python getting started

Python Example

import awa
import asyncio
from dataclasses import dataclass

@dataclass
class SendEmail:
    to: str
    subject: str

async def main():
    client = awa.AsyncClient("postgres://localhost/mydb")
    await client.migrate()

    @client.task(SendEmail, queue="email")
    async def handle_email(job):
        print(f"Sending to {job.args.to}: {job.args.subject}")

    await client.insert(
        SendEmail(to="[e-mail]", subject="Welcome"),
        queue="email",
    )

    client.start([("email", 2)])
    await asyncio.sleep(1)
    await client.shutdown()

asyncio.run(main())

Progress tracking — checkpoint and resume on retry:

@client.task(BatchImport, queue="etl")
async def handle_import(job):
    last_id = (job.progress or {}).get("metadata", {}).get("last_id", 0)
    for item in fetch_items(after=last_id):
        process(item)
        job.set_progress(50, "halfway")
        job.update_metadata({"last_id": item.id})
    await job.flush_progress()

Transactional enqueue — atomic with your business logic:

async with await client.transaction() as tx:
    await tx.execute("INSERT INTO orders (id) VALUES ($1)", order_id)
    await tx.insert(SendEmail(to="[e-mail]", subject="Order confirmed"))

Sync API for Django/Flask — use awa.Client for sync frameworks; all methods are plain (no suffix):

client = awa.Client("postgres://localhost/mydb")
client.migrate()
job = client.insert(SendEmail(to="[e-mail]", subject="Hello"))

Sequential callbacks — suspend a handler, wait for an external system, then resume:

@client.task(ProcessPayment, queue="payments")
async def handle_payment(job):
    token = await job.register_callback(timeout_seconds=3600)
    send_to_payment_gateway(token.id, job.args.amount)
    result = await job.wait_for_callback(token)
    # result contains the payload from resume_external()
    await record_payment(job.args.order_id, result)

The external system calls await client.resume_external(callback_id, {"status": "paid"}) to wake the handler.

Periodic jobs — leader-elected cron scheduling with timezone support:

client.periodic(
    "daily_report", "0 9 * * *",
    GenerateReport, GenerateReport(format="pdf"),
    timezone="Pacific/Auckland",
)

6-field expressions with seconds precision are also supported: "*/15 * * * * *" fires every 15 seconds.

See examples/python/ for complete runnable scripts tested in CI.

Rust Example

use awa::{Client, QueueConfig, JobArgs, JobResult, JobError, JobContext, Worker};
use serde::{Serialize, Deserialize};

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

struct SendEmailWorker;

#[async_trait::async_trait]
impl Worker for SendEmailWorker {
    fn kind(&self) -> &'static str { "send_email" }

    async fn perform(&self, ctx: &JobContext) -> Result<JobResult, JobError> {
        let args: SendEmail = serde_json::from_value(ctx.job.args.clone())
            .map_err(|e| JobError::terminal(e.to_string()))?;
        send_email(&args.to, &args.subject).await
            .map_err(JobError::retryable)?;
        Ok(JobResult::Completed)
    }
}

// Insert a job (with uniqueness)
awa::insert_with(&pool, &SendEmail { to: "[e-mail]".into(), subject: "Welcome".into() },
    awa::InsertOpts { unique: Some(awa::UniqueOpts { by_args: true, ..Default::default() }), ..Default::default() },
).await?;

// Cancel by unique key (e.g., when the triggering condition is resolved)
awa::admin::cancel_by_unique_key(&pool, "send_email", None, Some(&serde_json::json!({"to": "[e-mail]", "subject": "Welcome"})), None).await?;

// Start workers with a typed lifecycle hook
let client = Client::builder(pool)
    .queue("default", QueueConfig::default())
    .register_worker(SendEmailWorker)
    .on_event::<SendEmail, _, _>(|event| async move {
        if let awa::JobEvent::Exhausted { args, error, .. } = event {
            tracing::error!(to = %args.to, error = %error, "email job exhausted retries");
        }
    })
    .build()?;
client.start().await?;

Installation

Python

pip install awa-pg       # SDK: insert, worker, admin, progress
pip install awa-cli      # CLI: migrations, queue admin, web UI

Rust

[dependencies]
awa = "0.5"

CLI

Available via pip (no Rust toolchain needed) or cargo:

pip install awa-cli
# or: cargo install awa-cli

awa --database-url $DATABASE_URL migrate
awa --database-url $DATABASE_URL serve
awa --database-url $DATABASE_URL queue stats
awa --database-url $DATABASE_URL job list --state failed
awa --database-url $DATABASE_URL job dump 123
awa --database-url $DATABASE_URL job dump-run 123

Architecture

 ┌────────────────┐  ┌────────────────┐
 │ Rust producer  │  │  Python (pip)  │
 └───────┬────────┘  └────────┬───────┘
         └────────┬───────────┘
                  ▼
       ┌────────────────────┐
       │     PostgreSQL     │
       │     jobs_hot       │
       │     scheduled_jobs │
       └─────────┬──────────┘
                 │
       ┌─────────┼─────────┐
       ▼         ▼         ▼
   ┌────────┐┌────────┐┌────────┐
   │ Worker ││ Worker ││ Worker │
   │ (Rust) ││ (PyO3) ││ (PyO3) │
   └────────┘└────────┘└────────┘

All coordination through Postgres. The Rust runtime owns polling, heartbeats, shutdown, and crash recovery for both languages. Mixed Rust and Python workers coexist on the same queues. See architecture overview for full details.

Workspace

Crate	Purpose
awa	Main crate — re-exports awa-model + awa-worker
awa-model	Types, queries, migrations, admin ops
awa-macros	#[derive(JobArgs)] proc macro
awa-worker	Runtime: dispatch, heartbeat, maintenance
awa-ui	Web UI (axum API + embedded React frontend)
awa-cli	CLI binary (migrations, admin, serve)
awa-python	PyO3 extension module (pip install awa-pg)
awa-testing	Test helpers (TestClient)

Documentation

Doc	Description
Rust getting started	From cargo add to a job reaching completed
Python getting started	From pip install to a job reaching completed
Deployment guide	Docker, Kubernetes, pool sizing, graceful shutdown
Migration guide	Fresh installs, upgrades, extracted SQL, rollback strategy
Configuration reference	QueueConfig, ClientBuilder, Python start(), env vars
Security & Postgres roles	Minimum-privilege roles, callback auth, operational guidance
Troubleshooting	Stuck running jobs, leader delays, heartbeat timeouts
Architecture overview	System design, data flow, state machine, crash recovery
Web UI design	API endpoints, pages, component library
Benchmarking notes	Methodology, headline numbers, how to run
Validation test plan	Full test matrix with 100+ test cases
TLA+ correctness models	Formal verification of core invariants
Grafana dashboards	Pre-built Prometheus dashboards for monitoring

Architecture Decision Records (ADRs)

• 001: Postgres-only
• 002: BLAKE3 uniqueness
• 003: Heartbeat + deadline hybrid
• 004: PyO3 async bridge
• 005: Priority aging
• 006: AwaTransaction as narrow SQL surface
• 007: Periodic cron jobs
• 008: COPY batch ingestion
• 009: Python sync support
• 010: Per-queue rate limiting
• 011: Weighted concurrency
• 012: Split hot and deferred job storage
• 013: Durable run leases and guarded finalization
• 014: Structured progress and metadata
• 015: Builder-side post-commit lifecycle hooks
• 016: Sequential callbacks and callback heartbeats
• 017: Python insert-only transaction bridging
• 018: HTTP Worker for serverless job dispatch

License

MIT OR Apache-2.0