Queue System

Durable background work — connections, jobs, workers, middleware, batches, chains, and failed-job handling, with the dispatch defect and the operational limits spelled out.

Queue System (`sillo.work.queue`)

What Problem Does This Solve?

When a user hits your API, they expect a response in milliseconds — not seconds. But many operations take seconds: sending emails, generating reports, resizing images, calling slow third-party APIs. If your handler waits for all of these, your user waits too. Worse, if the handler crashes mid-operation, the work is lost.

The queue system solves this by decoupling work from the request. Instead of doing the work inside the handler, you describe what needs to be done (a Job), push that description onto a Connection, and respond to the user immediately. A separate Worker process pulls jobs and executes them — with automatic retry, timeout, and failure logging.

This pattern is called “deferred execution” or “background processing.” Every major web framework has a version of it: Laravel Queues, Django-Q, Celery, Sidekiq, Bull. sillo.work.queue is Sillo’s take — designed to be deeply integrated with the framework’s DI system, app lifecycle, and typing conventions.

System Architecture

The Dispatch Path (Handler → Queue)

The Worker Path (Queue → Execution)

The handler never waits for the work to complete. It builds a Job object, converts it to a portable JSON string via the PayloadSerializer, pushes it onto a Connection, and responds immediately. Total handler time: milliseconds.

The Worker Path (Queue → Execution)

The worker loops forever: pop, decode, instantiate, execute through middleware, ack. If the job fails, retry with backoff. If all retries are exhausted, log to the failed job repository.

Why Separate Processes?

You can run the worker in the same process as the HTTP server (using a SyncConnection), but for production you run workers in separate processes — or separate machines — connected by Redis. This gives you:

Isolation — a crashed worker doesn’t take down the HTTP server
Scaling — add more worker processes to handle more jobs
Resilience — if a worker process dies, jobs stay safely in Redis

Connections

A Connection is a named backend that stores serialized job payloads between dispatch and execution. Every connection must implement five operations in the QueueConnection abstract class.

The Five-Operation Contract

Method	Args	Returns	Called By
`push`	`queue_name, payload, delay`	`str` (job ID)	Handler
`pop`	`queue_name, timeout`	`(str, str) \| None`	Worker
`size`	`queue_name`	`int`	Monitoring
`ack`	`queue_name, job_id`	`None`	Worker
`fail`	`queue_name, job_id, payload, exception`	`None`	Worker

SyncConnection — In-Process, Non-Persistent

Uses an asyncio.PriorityQueue internally. Delayed jobs are held in a time-sorted list and released when their delay expires. All state is lost on process restart. Best for development and single-process deployments.

from sillo.work.queue import SyncConnection

conn = SyncConnection()

# Immediate:
job_id = await conn.push("emails", '{"to":"user@ex.com"}')

# Delayed 30 seconds:
await conn.push("emails", '{"to":"admin@ex.com"}', delay=30)

# Dequeue (blocks up to 5s):
result = await conn.pop("emails", timeout=5)
if result:
    popped_id, payload = result

# Monitoring:
pending = await conn.size("emails")
await conn.clear("emails")

RedisConnection — Persistent, Cross-Process

Stores jobs in Redis. Delayed jobs use sorted sets (scored by wake time). Active jobs use lists. Workers block on BRPOP rather than polling. Jobs survive process restarts and can be consumed by workers on different machines.

from sillo.work.queue import RedisConnection

conn = RedisConnection(
    "redis://localhost:6379",
    prefix="myapp:queue:",    # all keys namespaced under this
)

await conn.push("critical", '{"priority":"high"}')
result = await conn.pop("critical", timeout=30)

How Redis keys are structured:

Purpose	Key
Active jobs	`myapp:queue:emails` (Redis list)
Delayed jobs	`myapp:queue:emails:delayed` (sorted set)

ConnectionManager — The Broker

Registers named connections and provides access by name:

from sillo.work.queue import ConnectionManager, SyncConnection, RedisConnection

mgr = ConnectionManager()
mgr.add("default", SyncConnection())
mgr.add("redis", RedisConnection("redis://localhost:6379"))

conn = mgr.connection("default")
conn = mgr.connection("redis")
# mgr.connection("unknown") → KeyError

Jobs

A Job is a class that encapsulates one unit of work.

Defining a Job

from sillo.work.queue import Job

class SendWelcomeEmail(Job):
    queue = "emails"
    tries = 3
    timeout = 30
    backoff = 10
    delete_when_completed = True
    middleware = []

    def __init__(self, user_id: str, template: str = "welcome"):
        self.user_id = user_id
        self.template = template

    async def handle(self):
        user = await User.get(id=self.user_id)
        html = render_template(self.template, user=user)
        await mail_service.send(user.email, "Welcome!", html)

    async def failed(self, exception):
        await alert(f"Welcome email permanently failed for {self.user_id}: {exception}")

Why Classes, Not Functions?

A class carries state (constructor arguments) and metadata (class attributes) together. When a worker deserializes a job from JSON, it can reconstruct it completely: import the class, call the constructor with the stored data, then call handle(). Functions can’t be serialized portably.

Class Attributes — Complete Reference

Attribute	Type	Default	Description
`queue`	`str`	`"default"`	Which ConnectionManager name to dispatch to
`tries`	`int`	`1`	Total execution attempts. `1` = no retry
`timeout`	`float\|None`	`30.0`	Seconds before cancellation. `None` = no timeout
`backoff`	`int`	`0`	Seconds before first retry. Doubles each attempt
`delete_when_completed`	`bool`	`True`	Remove from queue after success
`middleware`	`list`	`[]`	Middleware instances, applied in list order

Dispatching

Push through the connection, which is an ordinary coroutine:

import json


async def enqueue(job_class, *args, delay: int = 0, **kwargs) -> str:
    connection = app.state["queue_connection"]
    payload = json.dumps(
        {"job": job_class.__name__, "args": args, "kwargs": kwargs}, default=str
    )
    return await connection.push(job_class.queue, payload, delay=delay)


await enqueue(SendWelcomeEmail, "user-42")                    # immediate
await enqueue(SendWelcomeEmail, "user-42", template="vip")    # with kwargs
await enqueue(SendWelcomeEmail, "user-42", delay=3600)        # in an hour
await enqueue(SendWelcomeEmail, "user-1")                     # queue from the class

For a one-off queue override, pass the name to push() rather than calling on_queue() — that classmethod mutates the class globally and affects every later dispatch in the process.

To run a job inline, bypassing the queue entirely — useful in tests — construct it and await handle() directly:

await SendWelcomeEmail("user-42").handle()

Real-World: Order Processing Pipeline

class ValidateOrder(Job):
    queue = "orders"; tries = 2; timeout = 30
    def __init__(self, order_id): self.order_id = order_id
    async def handle(self):
        order = await Order.get(id=self.order_id)
        if not order.items: raise ValueError("Empty order")
        order.status = "validated"; await order.save()
        await enqueue(ProcessPayment, order.id)  # chain to next step

class ProcessPayment(Job):
    queue = "payments"; tries = 3; timeout = 60
    def __init__(self, order_id): self.order_id = order_id
    async def handle(self):
        order = await Order.get(id=self.order_id)
        charge = await gateway.charge(order.total, order.currency,
            idempotency_key=f"order-{order.id}")
        order.payment_id = charge.id; order.status = "paid"
        await order.save()
        await enqueue(FulfillOrder, order.id)

class FulfillOrder(Job):
    queue = "fulfillment"; tries = 5; timeout = 300
    def __init__(self, order_id): self.order_id = order_id
    async def handle(self):
        order = await Order.get(id=self.order_id)
        label = await shipping.create_label(order)
        order.status = "fulfilled"; order.tracking = label.tracking
        await order.save()

@app.post("/orders", request_model=CreateOrderForm)
async def create_order(request, response):
    order = await Order.create(...)
    await enqueue(ValidateOrder, order.id)
    return response.json({"order_id": order.id, "status": "pending"}, status_code=202)

Why three jobs instead of one? Each step can be retried independently. If the payment gateway is temporarily down, only ProcessPayment fails and retries — validated orders aren’t affected. If fulfillment takes 5 minutes, it has its own timeout. This is “separation of concerns at the job level.”

Payload Serializer

Why It Exists

Jobs are code. Code can’t be sent over a network. The PayloadSerializer converts a Job into a JSON string (for the queue) and back (for the worker). It encodes:

The fully-qualified class name ("mymodule.SendWelcomeEmail")
Constructor keyword arguments ({"user_id": "42", ...})
Metadata: max_tries, timeout, delay, priority, queue

Why JSON, Not Pickle?

Pickle is Python-specific, unsafe (arbitrary code execution on deserialization), and fragile across Python versions. JSON is portable, safe, and human-readable. The trade-off: constructor arguments must be simple types (strings, numbers, dicts, lists). Complex objects should be looked up by ID inside handle().

from sillo.work.queue import PayloadSerializer

serializer = PayloadSerializer()

# Encode:
payload = serializer.serialize(
    "mymodule.SendWelcomeEmail",
    {"user_id": "42", "template": "welcome"},
    max_tries=3, timeout=30, queue="emails",
)

# Decode:
data = serializer.deserialize(payload)
# → {"job_class": "mymodule.SendWelcomeEmail", "data": {...}, "max_tries": 3, ...}

Workers

QueueWorker

from sillo.work.queue import QueueWorker, WorkerOptions, PayloadSerializer, MemoryFailedRepository

worker = QueueWorker(
    mgr,              # ConnectionManager — where to find queues
    PayloadSerializer(),
    MemoryFailedRepository(),
    options=WorkerOptions(
        concurrency=4,
        queues=["critical", "default", "emails"],
        timeout=60.0,
        sleep=3.0,
        max_jobs=1000,
        backoff=2.0,
    ),
)

await worker.run()
worker.pause(); worker.resume(); worker.stop()

WorkerOptions — Every Parameter

Parameter	Default	Meaning
`concurrency`	`4`	Parallel asyncio tasks pulling jobs
`queues`	`["default"]`	Queue names in priority order — index 0 checked first
`timeout`	`60.0`	Default per-job deadline. Job’s `timeout` attr overrides
`sleep`	`3.0`	Wait when ALL queues empty before re-checking
`max_jobs`	`0`	Exit after N jobs (0 = unlimited). Use with process supervisor
`max_exec_time`	`0`	Exit after N seconds (0 = unlimited)
`backoff`	`0.0`	Default base retry delay. Job’s `backoff` overrides
`memory_limit`	`128`	Exit if RSS exceeds N MB. Mitigates slow leaks

WorkerPool

Manages multiple QueueWorker instances:

from sillo.work.queue import WorkerPool

email_worker = QueueWorker(mgr, serializer, repo,
    options=WorkerOptions(queues=["emails"], concurrency=4))

report_worker = QueueWorker(mgr, serializer, repo,
    options=WorkerOptions(queues=["reports"], concurrency=2, timeout=300))

pool = WorkerPool().add(email_worker).add(report_worker)
await pool.start()
await pool.shutdown()

Deep Dive: Worker Tuning

Concurrency vs. CPU cores: Concurrency is the number of asyncio tasks pulling jobs simultaneously, not processes. Four tasks can all be waiting on I/O (network, database) at the same time. Set it higher than your CPU core count if jobs are I/O-bound:

4 CPU cores, mostly I/O (email, HTTP calls): concurrency=12–16
4 CPU cores, mixed (some compute): concurrency=6–8
4 CPU cores, CPU-heavy: concurrency=4–6

Higher concurrency means more memory (each task has its own stack), so monitor RSS.

Queue priority ordering: Queues are checked left-to-right in a tight loop:

WorkerOptions(queues=["critical", "default", "reports"])

This means:

Pop from critical if not empty
Else pop from default if not empty
Else pop from reports if not empty
If all empty, sleep for sleep seconds

A thousand jobs in reports do not block critical jobs — the worker checks critical first on every poll. But if critical is constantly full, reports will starve. Mitigate by running separate workers:

# process 1: fast, low-latency work
python -m myapp.worker --queues critical,default --concurrency 8

# process 2: long, low-priority work
python -m myapp.worker --queues reports --concurrency 2 --max_exec_time 3600

Timeout strategy: A job timing out is not a soft failure; it is a hard crash:

asyncio.TimeoutError("Timeout waiting for job to finish")

The job then retries (if tries > 1). Set timeouts below external API timeouts so you control the retry:

class CallPaymentAPI(Job):
    timeout = 20  # API's own timeout is 30s; we fail first and retry
    tries = 5
    backoff = 10

If a job consistently times out, your concurrency is too high. The worker checks memory_limit and exits the process — triggering a restart — but waiting for a restart is slow. Lower concurrency first.

Worker Lifecycle and Signals

A worker runs until one of five things happens:

Exit Reason	Trigger	Graceful?
Manual stop	`worker.stop()`	Yes — waits for current job
max_jobs reached	N jobs processed	Yes — finishes current job
max_exec_time reached	N seconds elapsed	Yes — finishes current job
memory_limit exceeded	RSS exceeds threshold	No — exits immediately
SIGTERM (orchestrator)	`kill -TERM <pid>`	Yes, if handler installed

In production, install a SIGTERM handler:

import signal

worker = QueueWorker(...)

def handle_sigterm(signum, frame):
    worker.stop()

signal.signal(signal.SIGTERM, handle_sigterm)
await worker.run()

Without it, orchestrators (Kubernetes, systemd) wait for the grace period (usually 30 seconds) and then SIGKILL, losing the current job. With the handler, a job mid-execution is returned to the queue for another worker to retry.

Debugging and Observability

What a worker logs:

[INFO] QueueWorker starting: 4 concurrent, queues=[critical, default]
[DEBUG] Polling critical (1 jobs), default (0 jobs)
[DEBUG] Got job-1234 from critical: SendEmail
[DEBUG] Running SendEmail with {'user_id': '42'}
[ERROR] SendEmail failed: ConnectionError: Network error. Retries: 1/3
[INFO] SendEmail succeeded in 1.234s
[INFO] Processed 100 jobs, success_rate=98%

Add custom logging inside jobs:

import logging

logger = logging.getLogger(__name__)

class SendEmail(Job):
    async def handle(self):
        logger.info(f"Sending to {self.user_id}")
        try:
            await mail.send(self.user_id)
            logger.info(f"Email sent to {self.user_id}")
        except Exception as e:
            logger.error(f"Failed: {e}", exc_info=True)
            raise

Monitor these metrics:

Metric	What it means	Alert on
`queue.depth`	Jobs waiting	> expected baseline
`job.age_seconds`	Oldest job’s wait time	> 5 minutes
`job.duration_ms`	Execution time (p50, p95, p99)	p99 > timeout
`job.retry_rate`	Failures as % of throughput	> 5%
`worker.memory_mb`	RSS of worker process	Trending up (memory leak)
`worker.exit_count`	Restarts	> 1/hour (something wrong)

Expose these via a metrics endpoint:

@app.get("/admin/queue-metrics")
async def metrics(request, response):
    conn = request.app.state["queue_connection"]
    return response.json({
        "critical_depth": await conn.size("critical"),
        "default_depth": await conn.size("default"),
        "reports_depth": await conn.size("reports"),
    })

Middleware

Middleware wraps every execution attempt. Each middleware is a class with __call__(self, handler) returning a new async handler.

Built-in

from sillo.work.queue import QRetryMiddleware, QTimeoutMiddleware, QRateLimitMiddleware

class MyJob(Job):
    middleware = [
        QRetryMiddleware(max_attempts=10, base_delay=5.0, max_delay=300),
        QTimeoutMiddleware(seconds=60),
        QRateLimitMiddleware(max_jobs=5, per_seconds=60),
    ]

Custom Middleware

Each middleware is a callable that receives the next handler and returns a new async callable:

import time

class TimingMiddleware:
    def __init__(self, metrics_registry):
        self.registry = metrics_registry

    def __call__(self, handler):
        async def wrapper():
            start = time.monotonic()
            try:
                result = await handler()
                self.registry.counter("job.success").inc()
                return result
            except Exception:
                self.registry.counter("job.failure").inc()
                raise
            finally:
                elapsed = (time.monotonic() - start) * 1000
                self.registry.histogram("job.duration_ms").observe(elapsed)
        return wrapper


class SendEmail(Job):
    middleware = [TimingMiddleware(metrics_registry)]

Middleware runs per attempt, so it sees retries:

import logging

logger = logging.getLogger(__name__)

class LoggingMiddleware:
    def __call__(self, handler):
        async def wrapper():
            logger.info(f"Starting job")
            try:
                result = await handler()
                logger.info(f"Succeeded")
                return result
            except Exception as e:
                logger.error(f"Failed: {e}")
                raise
        return wrapper

If a job retries twice, this middleware logs the start and failure three times.

Middleware Stacking and Order

Middleware is applied in list order, and each wraps the next. The outer middleware sees both the inner middleware and the handler:

class Middleware1:
    def __call__(self, handler):
        async def wrapper():
            print("M1 start")
            result = await handler()
            print("M1 end")
            return result
        return wrapper

class Middleware2:
    def __call__(self, handler):
        async def wrapper():
            print("M2 start")
            result = await handler()
            print("M2 end")
            return result
        return wrapper

class MyJob(Job):
    middleware = [Middleware1(), Middleware2()]
    async def handle(self):
        print("job")

# Output on success:
# M1 start
# M2 start
# job
# M2 end
# M1 end

Put timing and metrics on the outside (outermost), and retries and rate-limiting on the inside (innermost):

class MyJob(Job):
    middleware = [
        TimingMiddleware(),      # outer: times everything
        LoggingMiddleware(),      # logs all attempts
        QRetryMiddleware(...),    # inner: controls retries
    ]

Common Middleware Patterns

Request tracing:

class TracingMiddleware:
    def __init__(self, tracer):
        self.tracer = tracer

    def __call__(self, handler):
        async def wrapper():
            with self.tracer.span("job") as span:
                span.set_attribute("job_id", self.job_id)
                return await handler()
        return wrapper

Timeout with graceful degradation:

class GracefulTimeoutMiddleware:
    def __init__(self, timeout_seconds):
        self.timeout = timeout_seconds

    def __call__(self, handler):
        async def wrapper():
            try:
                return await asyncio.wait_for(handler(), timeout=self.timeout)
            except asyncio.TimeoutError:
                logger.warning(f"Job timed out; marking as partial success")
                return {"status": "partial", "reason": "timeout"}
        return wrapper

Transaction wrapper:

from sillo.record import transaction

class TransactionMiddleware:
    def __call__(self, handler):
        async def wrapper():
            async with transaction():
                return await handler()
        return wrapper

Failed Jobs

When a job exhausts its retries, it is not lost — it is logged to a failed-job repository. This is your safety net.

The Retry Lifecycle

[Attempt 1] SendEmail fails → backoff 10s
[Attempt 2] SendEmail fails → backoff 20s (backoff doubled)
[Attempt 3] SendEmail fails → backoff 40s
[Attempt 4] Retries exhausted → logged to repository

A job with tries=4 and backoff=10 will retry at T+10s, T+30s, T+70s (if all fail). That’s roughly two minutes from first failure to dead-letter. Set tries to match the failure’s severity:

Job	tries	backoff	Why
`SendEmail`	3–5	10–30	Transient network, DNS, timeouts
`ChargeCard`	2–3	30–60	Payment gateway down (rarer, retry slower)
`DeleteOldLogs`	1	0	Logging never retries; idempotent anyway

MemoryFailedRepository

from sillo.work.queue import MemoryFailedRepository

repo = MemoryFailedRepository()

# Logged automatically by the worker on exhausted retries
# await repo.log("emails", "job-123", "SendEmail", payload, traceback)

# Inspect failures
for fj in await repo.all(limit=20):
    print(f"{fj.id}: {fj.job_class} failed: {fj.exception[:200]}")

# Manually retry after a fix ships
failed_job = await repo.get("job-123")
if failed_job:
    await connection.push(failed_job.queue, failed_job.payload)
    await repo.forget(failed_job.id)

# Or forget entirely (careful!)
await repo.forget("job-123")
await repo.flush()  # Clear all

Custom FailedRepository

Implement the abstract class to store failures durably:

from sillo.work.queue import FailedRepository

class DatabaseFailedRepository(FailedRepository):
    async def log(self, queue_name, job_id, job_class, payload, exception):
        await FailedJob.create(
            queue_name=queue_name,
            job_id=job_id,
            job_class=job_class,
            payload=payload,
            exception=str(exception),
            traceback=traceback.format_exc(),
            failed_at=datetime.now(),
        )

    async def get(self, job_id):
        row = await FailedJob.get_or_none(job_id=job_id)
        if row:
            return FailedJobRecord(
                id=row.job_id,
                queue=row.queue_name,
                job_class=row.job_class,
                payload=row.payload,
                exception=row.exception,
            )

    async def all(self, limit=100):
        rows = await FailedJob.all().limit(limit)
        return [... convert to FailedJobRecord ...]

    async def forget(self, job_id):
        await FailedJob.delete_by_id(job_id)

    async def flush(self):
        await FailedJob.all().delete()

Pass it to the worker:

repo = DatabaseFailedRepository()
worker = QueueWorker(mgr, serializer, repo, options=...)

Handling Failures at Scale

As jobs fail, they accumulate in the repository. Without a policy, that store grows unbounded and becomes a support problem.

Decide in advance:

Who gets paged? If failures exceed a threshold (50 in 1 hour), alert on-call.
How do failed jobs get retried? After a fix ships, re-enqueue them from the repository.
Retention? Keep entries for 7 days; auto-delete older ones.

class RetryFailedJobs(Job):
    queue = "admin"
    timeout = 300

    async def handle(self):
        repo = get_repository()
        now = datetime.now()

        # Retry recent failures (< 1 hour old)
        for fj in await repo.all(limit=1000):
            age = (now - fj.failed_at).total_seconds()
            if age < 3600:
                # Re-enqueue
                await connection.push(fj.queue, fj.payload)
                await repo.forget(fj.id)
                logger.info(f"Retried {fj.id}")

        # Delete old failures (> 7 days)
        stale = await FailedJob.filter(failed_at__lt=now - timedelta(days=7))
        for row in stale:
            await row.delete()
            logger.info(f"Purged {row.id}")

Run this via the scheduler daily:

@scheduler.cron("0 3 * * *")  # 3 AM daily
async def retry_failed_jobs():
    await RetryFailedJobs.dispatch()

Batching & Chaining

Batch tracks a set of job ids and fires a callback when all of them have reported in. JobChain runs a list of async callables one after another.

from sillo.work.queue import Batch

batch = Batch(
    "import-users",
    on_complete=lambda b: notify(f"{b.completed_count}/{b.total} imported"),
    allow_failures=True,
    timeout=600,
)

for user in users:
    batch.add(await enqueue(ImportUser, user.id))    # add() takes the job id

await batch.wait(timeout=600)
print(batch.completed_count, batch.failed_count, batch.is_done)

from sillo.work.queue import JobChain

chain = JobChain()
chain.then(lambda: ValidateFile("data.csv").handle())
chain.then(lambda: TransformFile("data.csv").handle())
results = await chain.run()

then() takes a zero-argument async callable, not a dispatch result — chain.then(SomeJob.dispatch(...)) would evaluate the dispatch immediately and append its return value, which is not callable. Wrap each step in a lambda or a partial.

run() executes the steps sequentially in the current process and returns their results. It is not a queue construct: nothing is persisted, a failure aborts the remaining steps by propagating, and a restart mid-chain loses everything after the current step. For a durable pipeline, have each job enqueue the next — see Jobs.

Custom Backend

Implement QueueConnection for any storage:

from sillo.work.queue import QueueConnection

class PostgresBackend(QueueConnection):
    async def push(self, queue_name, payload, *, delay=0):
        await db.execute(
            "INSERT INTO jobs (queue, payload, available_at) "
            "VALUES ($1, $2, NOW() + interval '$3 seconds')",
            queue_name, payload, str(delay),
        )
        return str(uuid4())

    async def pop(self, queue_name, *, timeout=0):
        row = await db.fetchrow(
            "DELETE FROM jobs WHERE queue = $1 AND available_at <= NOW() "
            "ORDER BY available_at FOR UPDATE SKIP LOCKED LIMIT 1 "
            "RETURNING id, payload",
            queue_name,
        )
        return (row["id"], row["payload"]) if row else None

    async def size(self, queue_name):
        return await db.fetchval("SELECT COUNT(*) FROM jobs WHERE queue = $1", queue_name)

    async def clear(self, queue_name):
        await db.execute("DELETE FROM jobs WHERE queue = $1", queue_name)

    async def ack(self, queue_name, job_id): pass  # deleted on pop

    async def fail(self, queue_name, job_id, payload, exception):
        await db.execute("INSERT INTO failed_jobs (...) VALUES (...)")

Key design:

pop() uses DELETE ... RETURNING + FOR UPDATE SKIP LOCKED to atomically claim a job — no two workers can get the same one.
ack() is a no-op because the job was deleted on pop. If the worker crashes before ack, the job is lost — but the DELETE was committed.
For “at-least-once” delivery, move the job to a “processing” list on pop and delete on ack.

Operating a queue

The API is half the story; the other half is what you need in place before a queue is something you can rely on at three in the morning.

Run workers as their own processes

Workers should not share a process with your web server. They have different scaling characteristics — web processes scale with request rate, workers with backlog depth — and different failure modes. A worker that OOMs on a large job should not take down request serving.

# web
uvicorn myapp.app:app --workers 4

# workers, scaled independently
python -m myapp.worker --queue billing --concurrency 4
python -m myapp.worker --queue default,reports --concurrency 8

Give every queue a purpose

One queue for everything means a thousand slow report jobs delay every password-reset email behind them. Split by latency expectation rather than by feature:

Queue	Contains	Workers
`critical`	Password resets, payment captures	Many, always warm
`default`	Ordinary user-triggered work	Scaled to backlog
`reports`	Long, heavy, tolerant of delay	Few, separate machines

A job in the wrong queue is the most common cause of “the queue is backed up” being both true and irrelevant.

Monitor depth, age, and failure rate

Depth alone is a poor signal — a queue with ten thousand jobs that drains in a minute is healthy, and one with three jobs that have been stuck for an hour is not.

@app.get("/admin/queues")
async def queue_health(request, response):
    connection = request.app.state["queue_connection"]
    return response.json({
        name: await connection.size(name)
        for name in ("critical", "default", "reports")
    })

Alert on oldest job age rather than count, and on failure rate as a proportion of throughput. Both catch problems that depth misses.

Have a dead-letter policy

Jobs that exhaust their retries land in the failed-job repository. That store is only useful if somebody looks at it. Decide in advance who is paged when it grows, how a failed job gets retried after a fix ships, and how long entries are kept.

repo = MemoryFailedRepository()

for failed_job in await repo.all():
    if failed_job.job_class == "ChargeCard":
        await connection.push(failed_job.queue, failed_job.payload)
        await repo.forget(failed_job.id)

A failed-job store nobody reads is a silent data-loss mechanism with extra steps.

Deploy without losing work

Workers must finish or release the job they are holding before exiting. On SIGTERM, stop accepting new jobs, wait for the current one up to a bound, and let anything unfinished return to the queue for another worker to pick up. That bound must be shorter than your orchestrator’s grace period, and your longest job’s timeout should be shorter still — a job that takes ten minutes cannot be drained inside a thirty-second window, and will be killed and retried on every single deploy.

import signal
import asyncio

async def main():
    worker = QueueWorker(mgr, serializer, repo, options=...)

    def handle_sigterm(signum, frame):
        # Stop accepting new jobs
        worker.stop()
        # Worker finishes current job and exits gracefully

    signal.signal(signal.SIGTERM, handle_sigterm)

    try:
        await worker.run()
    except KeyboardInterrupt:
        pass  # also safe; just exits

Orchestrator configuration:

spec:
  terminationGracePeriodSeconds: 120  # time to drain
  containers:
  - name: worker
    image: myapp:latest
    command: ["python", "-m", "myapp.worker"]
    lifecycle:
      preStop:
        exec:
          command: ["sh", "-c", "sleep 5"]  # extra buffer for SIGTERM handling

[Service]
ExecStart=/path/to/venv/bin/python -m myapp.worker
TimeoutStopSec=120
KillMode=mixed

Verify the drain works:

# Start worker, let it process a job
python -m myapp.worker &

# Stop it gracefully
kill -TERM $!

# Watch logs — should see "Waiting for current job" and graceful exit

If the process is killed (SIGKILL) before it finishes, the job is returned to the queue and retried by another worker. That’s expected — but if it happens on every deploy, your timeouts are wrong. Profile your jobs:

# Monitor job duration in production
psql mydb -c "SELECT job_class, AVG(duration_seconds) FROM job_logs GROUP BY job_class;"

Capacity Planning

Estimate queue size over time:

dispatch_rate = 1000 jobs/min = 16.7 jobs/sec
job_duration = 5 sec (p50)
workers = 4
concurrency = 8
throughput = 4 workers × 8 concurrent = 32 jobs/sec

Queue growth = 16.7 dispatched - 32 processed = -15.3 jobs/sec
→ Queue drains. This is healthy.

If throughput < dispatch rate, the queue backs up. Add workers:

Queue size = 10,000 jobs
dispatch_rate = 16.7 jobs/sec
current throughput = 16.7 jobs/sec
→ Queue stays at 10,000 for 10,000 / 16.7 = 600 seconds (10 min)

If throughput were 8 jobs/sec:
→ Queue grows by 8.7 jobs/sec → 10 min to reach 16,000 jobs

Scaling strategy:

Monitor queue depth and age
Alert if oldest job > 5 minutes
Horizontally add workers (new processes/machines)
Do NOT raise concurrency without testing — it can hide memory leaks

Connection Resilience

Redis failover:

If your Redis node dies, jobs in transit are lost (they’re on the network). Jobs at rest survive on a replica if you have one. For critical work:

from redis.sentinel import Sentinel

sentinel = Sentinel([("sentinel1", 26379), ("sentinel2", 26379)])
conn = RedisConnection(
    redis_client=sentinel.master_for("mymaster", socket_timeout=0.1)
)

Database failover (if using DatabaseFailedRepository):

Same pattern — use your database’s built-in replication and failover. The failed-job repository is read-only during normal operation (only during manual retry), so queries can be routed to a read replica.

Monitoring and Alerting

Dashboard queries:

-- Queue depth by type (1-min granularity)
SELECT
  queue_name,
  COUNT(*) as depth,
  NOW() as ts
FROM queue_jobs
GROUP BY queue_name;

-- Failed jobs (past 24 hours)
SELECT job_class, COUNT(*) as count FROM failed_jobs
WHERE failed_at > NOW() - INTERVAL '24 hours'
GROUP BY job_class ORDER BY count DESC;

-- Retry rate (jobs that failed and were retried)
SELECT
  SUM(CASE WHEN attempts > 1 THEN 1 ELSE 0 END) as retried,
  COUNT(*) as total,
  ROUND(100.0 * SUM(CASE WHEN attempts > 1 THEN 1 ELSE 0 END) / COUNT(*), 2) as retry_pct
FROM completed_jobs
WHERE completed_at > NOW() - INTERVAL '1 hour';

Alerts to set:

Alert	Condition	Action
Queue backed up	`oldest_job_age > 5 min`	Page on-call; check worker health
High failure rate	`failure_rate > 10%`	Check logs for errors; may be upstream service down
Worker down	`no jobs processed in 10 min`	Check if process crashed; restart if needed
Memory leak	`worker RSS trending up`	Restart worker; investigate in dev

Queue System

Queue System (sillo.work.queue)

What Problem Does This Solve?

System Architecture

The Dispatch Path (Handler → Queue)

The Worker Path (Queue → Execution)

The Worker Path (Queue → Execution)

Why Separate Processes?

Connections

The Five-Operation Contract

SyncConnection — In-Process, Non-Persistent

RedisConnection — Persistent, Cross-Process

ConnectionManager — The Broker

Jobs

Defining a Job

Why Classes, Not Functions?

Class Attributes — Complete Reference

Dispatching

Real-World: Order Processing Pipeline

Payload Serializer

Why It Exists

Why JSON, Not Pickle?

Workers

QueueWorker

WorkerOptions — Every Parameter

WorkerPool

Deep Dive: Worker Tuning

Worker Lifecycle and Signals

Debugging and Observability

Middleware

Built-in

Custom Middleware

Middleware Stacking and Order

Common Middleware Patterns

Failed Jobs

The Retry Lifecycle

MemoryFailedRepository

Custom FailedRepository

Handling Failures at Scale

Batching & Chaining

Custom Backend

Operating a queue

Run workers as their own processes

Give every queue a purpose

Monitor depth, age, and failure rate

Have a dead-letter policy

Deploy without losing work

Capacity Planning

Connection Resilience

Monitoring and Alerting

Start Here

Core Concepts

Validation

ORM (Record)

Deep Dive

Helpers

Authentication & Authorization

Security

Queue System (`sillo.work.queue`)