🐘 Talos-Queue

Durable, Distributed Job Queue for PostgreSQL & Supabase (Zero Redis, Zero Celery)

Durable, concurrent background jobs directly in your existing PostgreSQL database.
Atomic row leases, non-blocking parallel worker claims via SKIP LOCKED (zero lock contention), automatic crash recovery, idempotency deduplication, and exponential backoff.

Why Not Redis? • Lifecycle & State Machine • Installation • Quickstart • SQL Migration

---

⚡ Why Talos-Queue? (Postgres as a Queue Done Right)

In modern architectures, adding Redis, RabbitMQ, or Celery introduces extra infrastructure, dual-write consistency issues, and operational headache.

Feature	Redis / BullMQ	Celery + RabbitMQ	pg-lease-queue
Infrastructure Overhead	⚠️ Extra server/service	❌ Two extra servers	✅ Zero (Uses your existing Postgres/Supabase)
ACID Consistency	❌ Dual-write desync	❌ Separate state	✅ Atomic within same Postgres transaction
Worker Crash Recovery	⚠️ Needs ACK timeout	⚠️ Complex visibility timeout	✅ Deterministic Leases & Heartbeats
Duplicate Prevention	Custom Lua scripts	Manual locking	✅ Partial Unique Index on (kind, dedupe_key)
Row Level Security (RLS)	❌ None	❌ None	✅ Fully locked to Service Role

---

🏗️ State Machine & Guarantees

stateDiagram-v2
    [*] --> PENDING: enqueue(kind, dedupe_key, payload)
    PENDING --> CLAIMED: claim() via FOR UPDATE SKIP LOCKED
    
    state CLAIMED {
        [*] --> RUNNING
        RUNNING --> HEARTBEAT: heartbeat() every 30s
        HEARTBEAT --> RUNNING: Lease extended
        RUNNING --> STAGE_UPDATE: set_stage('transcoding')
    }
    
    CLAIMED --> FINISHED: finish(result)
    CLAIMED --> RETRYING: Temporary Exception (attempts under max_attempts)
    RETRYING --> PENDING: Exponential backoff delay
    CLAIMED --> FAILED: PermanentError OR attempts >= max_attempts
    
    CLAIMED --> PENDING: Worker crashes / Lease expires
    FINISHED --> [*]
    FAILED --> [*]

Loading

🛡️ Concurrency Model & Delivery Semantics

1. Row-Level Parallel Claims (FOR UPDATE SKIP LOCKED): Multiple workers claim jobs concurrently using PostgreSQL row locks. By using SKIP LOCKED, workers bypass already locked rows immediately without blocking each other or deadlocking.
2. At-Least-Once Delivery & Crash Recovery: Jobs are guaranteed at-least-once execution. If a worker process crashes, drops network connection, or runs out of memory, its lease expires and the job is automatically reclaimed.
3. Transactional Boundary (Zero Dual-Write): Enqueueing can participate in the same PostgreSQL transaction as your business entities (BEGIN ... COMMIT). If the business write fails or rolls back, the job is never queued.
4. Deterministic Lease Renewal: Long-running tasks (transcoding, scraping, AI) dynamically renew their lease via background heartbeats. If a worker fails to heartbeat before the lease window expires, its late output is discarded safely.

---

📦 Installation

# 1. Apply the migration to your database (Supabase SQL Editor or psql)
psql -d mydb -f sql/001_job_queue.sql

# 2. Install the client library
pip install git+https://github.com/Kaio-Comk/Talos-Queue.git

---

🚀 Quickstart

Works out-of-the-box with supabase-py or standard PostgreSQL connection pools:

import asyncio
from supabase import create_client
from pgleasequeue import LeaseQueue, PermanentError, run_worker

# 1. Initialize queue with Service Role key
supabase = create_client("https://xyz.supabase.co", "SERVICE_ROLE_KEY")
queue = LeaseQueue(supabase)

# 2. Producer: Enqueue idempotent job
queue.enqueue(
    kind="transcode",
    dedupe_key="video:892",
    payload={"video_id": 892, "target_format": "aac"},
    priority=100
)

# 3. Worker implementation
async def process_job(job):
    print(f"Processing job {job.id}: {job.payload}")
    
    # Non-retryable error
    if not job.payload.get("video_id"):
        raise PermanentError("Invalid payload: missing video_id")
        
    # Long running work (heartbeat runs automatically in background)
    queue.set_stage(job.id, worker_id="worker-1", stage="encoding")
    await asyncio.sleep(5)
    
    return {"status": "success", "file_url": "https://storage.com/892.mp4"}

# 4. Start concurrent worker loop
async def main():
    await run_worker(queue, process_job, worker_id="worker-1", kind="transcode")

asyncio.run(main())

---

🧪 Testing

pip install -e ".[dev]"
pytest -v

---

📄 License

Distributed under the MIT License. See LICENSE for details.

_{Built with ❤️ by Kaio Lima. If Talos-Queue simplified your background jobs, please star ⭐ the project!}