RustQueue

An async job processing engine built in Rust. Submit jobs via HTTP, a concurrent worker picks them up, processes them, retries on failure, and streams live status updates to any connected client over Server-Sent Events.

Built to understand how job queues like BullMQ and Sidekiq actually work under the hood -- not just use them as a black box.

Stack: Axum, Tokio, sqlx, PostgreSQL, lettre (AWS SES)

Status: Active development. Core is working and stable. Open to contributions -- see CONTRIBUTING.md.

---

What it does

• Accepts jobs via a REST API (currently supports email job type)
• Persists jobs to PostgreSQL with full status tracking
• Claims jobs safely using SELECT FOR UPDATE SKIP LOCKED -- no double-processing, even with multiple concurrent workers
• Processes up to 10 jobs concurrently via a Semaphore-capped worker pool
• Retries failed jobs with exponential backoff (1s, 5s, 25s...) up to max_attempts
• Recovers jobs orphaned by crashed workers via a stale-job reaper (runs every 5 minutes)
• Wakes the worker instantly on new jobs via Postgres LISTEN/NOTIFY -- no polling delay
• Streams live status updates to connected clients via Server-Sent Events

---

Architecture

HTTP clients
    |
    |  POST /jobs      GET /jobs      GET /jobs/:id      GET /events
    v
  Axum router  ---- Arc<AppState> (PgPool + broadcast::Sender) ----+
                                                                    |
              +-----------------------------------------------------+
              |  Worker loop (tokio::spawn)                         |
              |    wakes on LISTEN/NOTIFY or 2s fallback            |
              |    SELECT FOR UPDATE SKIP LOCKED                     |
              |    Semaphore-capped concurrency (max 10)             |
              |    executor dispatch via HandlerRegistry             |
              |    exponential backoff retries                       |
              |                                                      |
              |  Stale-job reaper (tokio::spawn)                    |
              |    runs every 5 minutes                             |
              |    reclaims jobs stuck in "processing" > 5 min      |
              +-----------------------------------------------------+
              |
         broadcast::Sender<String>  ->  SSE /events stream
              |
         PostgreSQL  (jobs table)

---

Jobs table schema

column	type	notes
id	UUID	primary key, auto-generated
job_type	VARCHAR	"email" -- extensible via HandlerRegistry
payload	JSONB	job-specific data
status	VARCHAR	pending -> processing -> done / failed
attempts	INT	how many times tried (including successes)
max_attempts	INT	default 3
last_error	TEXT	nullable, last failure message
scheduled_at	TIMESTAMPTZ	run not before this time
processing_started_at	TIMESTAMPTZ	set on claim, cleared on exit -- used by stale-job reaper
created_at	TIMESTAMPTZ
updated_at	TIMESTAMPTZ

---

API

method	route	purpose
POST	/jobs	Submit a new job
GET	/jobs	List jobs (supports ?limit=N and ?status=pending/done/failed/processing)
GET	/jobs/:id	Get a specific job by ID
GET	/events	SSE stream of live job status updates

Submit a job

POST /jobs
Content-Type: application/json

{
  "job_type": "email",
  "payload": {
    "to": "someone@example.com"
  }
}

Filter jobs

GET /jobs?status=failed&limit=10

---

Getting started

Prerequisites

• Rust (stable)
• PostgreSQL
• An AWS SES account (with at least one verified email address)
• sqlx-cli: cargo install sqlx-cli --no-default-features --features postgres

Setup

git clone https://github.com/Ekojoecovenant/rustqueue.git
cd rustqueue

cp .env.example .env
# fill in your credentials

sqlx migrate run

cargo run

The server starts on the port defined in your .env (default 3000).

Environment variables

DATABASE_URL=postgresql://user:password@localhost:5432/rustqueue
PORT=3000
SMTP_USERNAME=your_ses_smtp_username
SMTP_PASSWORD=your_ses_smtp_password
SMTP_HOST=email-smtp.eu-north-1.amazonaws.com
SMTP_PORT=587
FROM_EMAIL=you@yourdomain.com

Note: AWS SES sandbox accounts can only send to verified email addresses. You will need to verify the to address in your SES console, or request production access.

---

Adding a new job type

1. Create src/executor/your_handler.rs implementing the JobHandler trait:

   pub struct YourHandler { /* config fields */ }
   
   #[async_trait]
   impl JobHandler for YourHandler {
       async fn execute(&self, payload: &Value) -> Result<()> {
           // your logic here
           Ok(())
       }
   }

2. Register it in HandlerRegistry::new() in src/executor/mod.rs:

   handlers.insert("your_type".to_string(), Arc::new(YourHandler::new(...)?));

3. Submit jobs with "job_type": "your_type" and a matching payload.

The worker loop, retry logic, SSE updates, and stale-job recovery all work automatically for any registered job type.

---

Project structure

rustqueue/
+-- src/
|   +-- main.rs              # startup, router, shared state wiring
|   +-- config.rs            # env config
|   +-- db.rs                # sqlx pool setup
|   +-- models/
|   |   +-- job.rs           # Job struct, CreateJobRequest
|   +-- routes/
|   |   +-- jobs.rs          # POST /jobs, GET /jobs, GET /jobs/:id
|   |   +-- events.rs        # GET /events (SSE)
|   +-- worker/
|   |   +-- mod.rs           # worker loop, stale-job reaper
|   +-- executor/
|       +-- mod.rs           # JobHandler trait, HandlerRegistry
|       +-- email.rs         # email job handler (AWS SES via lettre)
+-- migrations/              # sqlx migration files
+-- .env.example

---

Roadmap

• Telegram job handler (second executor, proving extensibility)
• Scheduled jobs via API (run_at field on job submission)
• Typed JobStatus enum replacing raw strings
• Bulk email sends (one job, multiple recipients)
• Cursor-based pagination on GET /jobs
• Graceful shutdown (let in-flight jobs finish before exit)
• Optional SMTP config (fail at dispatch time, not startup)
• Explore Cloudflare Email Service as alternative executor

---

Contributing

See CONTRIBUTING.md.

---

Author

Ekojoe Covenant Lemom -- Backend Engineer https://ekojoe.name.ng | https://github.com/Ekojoecovenant

---

License

MIT