13-job-queues.md

Chapter 13 — Job Queues & Task Workers

Part III — Recipes (Leveraging Redis)

← Chapter 12 — Search & Analytics | Table of Contents | Chapter 14 — Real-World Patterns →

---

Background jobs—email sends, image processing, webhook delivery—need a queue that buffers work, survives restarts, and scales horizontally. Redis offers everything from a five-line List queue to production-grade Streams with consumer groups, plus Sorted Sets for delayed scheduling.

This chapter progresses from simple to robust: list queues, reliable Streams workers, delayed jobs, idempotency, and dead-letter handling.

---

Recipe 13.1 — Simple List Queue

Problem

You need a minimal work queue: producers push tasks; workers pull and process. Throughput is moderate; occasional loss on worker crash is acceptable for idempotent tasks.

Solution

Use a List as a FIFO queue: LPUSH to enqueue, BRPOP to blocking dequeue.

redis-cli:

127.0.0.1:6379> LPUSH queue:emails '{"to":"user@example.com","template":"welcome"}'
(integer) 1
127.0.0.1:6379> BRPOP queue:emails 30
1) "queue:emails"
2) "{\"to\":\"user@example.com\",\"template\":\"welcome\"}"

BRPOP blocks up to 30 seconds waiting for an item.

Python:

import json
import redis

r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)

def enqueue(queue: str, job: dict):
    r.lpush(queue, json.dumps(job))

def worker(queue: str):
    while True:
        result = r.brpop(queue, timeout=5)
        if result is None:
            continue
        _, payload = result
        job = json.loads(payload)
        process(job)

Node.js:

async function enqueue(queue, job) {
  await redis.lpush(queue, JSON.stringify(job));
}

async function worker(queue) {
  while (true) {
    const result = await redis.brpop(queue, 5);
    if (!result) continue;
    const [, payload] = result;
    await process(JSON.parse(payload));
  }
}

Go:

func Worker(ctx context.Context, rdb *redis.Client, queue string) {
    for {
        result, err := rdb.BRPop(ctx, 5*time.Second, queue).Result()
        if err == redis.Nil {
            continue
        }
        if err != nil {
            return
        }
        process(ctx, result[1])
    }
}

Reliable-ish variant — RPOPLPUSH to processing list:

127.0.0.1:6379> LPUSH queue:jobs task-1
127.0.0.1:6379> RPOPLPUSH queue:jobs queue:jobs:processing
"task-1"
# After success:
127.0.0.1:6379> LREM queue:jobs:processing 1 task-1

On crash, orphaned items remain in queue:jobs:processing for recovery.

Priority queues with multiple lists:

127.0.0.1:6379> LPUSH queue:jobs:high '{"id":1,"urgent":true}'
127.0.0.1:6379> LPUSH queue:jobs:low '{"id":2,"urgent":false}'

Worker checks high before low:

def dequeue_priority(r, timeout=5):
    result = r.brpop(["queue:jobs:high", "queue:jobs:low"], timeout=timeout)
    if result:
        return json.loads(result[1])
    return None

Node.js — reliable RPOPLPUSH:

async function reliableDequeue() {
  const task = await redis.brpoplpush('queue:jobs', 'queue:jobs:processing', 5);
  if (!task) return null;
  try {
    await process(JSON.parse(task));
    await redis.lrem('queue:jobs:processing', 1, task);
  } catch (err) {
    throw err; // recovery worker handles processing list
  }
}

Discussion

List queues are simple and fast. Weaknesses:

• No built-in ACK; worker crash after BRPOP loses the job unless you use RPOPLPUSH.
• No consumer groups—one job goes to one worker, but no partition assignment.
• No replay or visibility into pending work beyond a processing list.

Graduate to Streams (Recipe 13.2) when jobs must not disappear.

See Also

• Chapter 4 — Lists
• Recipe 13.2 — Streams consumer groups

---

Recipe 13.2 — Streams-Based Reliable Queue with Consumer Groups

Problem

Workers crash mid-job. You need at-least-once delivery, multiple parallel consumers sharing load, and visibility into in-flight messages.

Solution

Redis Streams with consumer groups provide ACK-based delivery and a Pending Entries List (PEL) for recovery.

Setup (redis-cli):

127.0.0.1:6379> XADD stream:jobs * type send_email payload '{"to":"a@b.com"}'
"1716892800000-0"

127.0.0.1:6379> XGROUP CREATE stream:jobs workers $ MKSTREAM
OK

127.0.0.1:6379> XREADGROUP GROUP workers consumer-a COUNT 1 BLOCK 5000 STREAMS stream:jobs >
1) 1) "stream:jobs"
   2) 1) 1) "1716892800000-0"
         2) 1) "type"
            2) "send_email"
            ...

127.0.0.1:6379> XACK stream:jobs workers 1716892800000-0
(integer) 1

The > ID reads only new messages never delivered to this group.

Python — full worker loop:

STREAM = "stream:jobs"
GROUP = "workers"
CONSUMER = "worker-1"

def ensure_group(r):
    try:
        r.xgroup_create(STREAM, GROUP, id="0", mkstream=True)
    except redis.ResponseError as e:
        if "BUSYGROUP" not in str(e):
            raise

def worker_loop(r):
    ensure_group(r)
    while True:
        messages = r.xreadgroup(
            GROUP, CONSUMER,
            {STREAM: ">"},
            count=5, block=5000,
        )
        for stream, entries in messages:
            for msg_id, fields in entries:
                try:
                    handle_job(fields)
                    r.xack(stream, GROUP, msg_id)
                except Exception:
                    # Leave unacked — reclaimed later via XCLAIM
                    log.exception("Job failed: %s", msg_id)

Recover stale pending messages:

def reclaim_stale(r, min_idle_ms=60000):
    pending = r.xpending_range(STREAM, GROUP, "-", "+", 100)
    for entry in pending:
        if entry["time_since_delivered"] >= min_idle_ms:
            claimed = r.xclaim(
                STREAM, GROUP, CONSUMER,
                min_idle_time=min_idle_ms,
                message_ids=[entry["message_id"]],
            )
            for msg_id, fields in claimed:
                handle_job(fields)
                r.xack(STREAM, GROUP, msg_id)

Node.js:

await redis.xgroup('CREATE', 'stream:jobs', 'workers', '0', 'MKSTREAM').catch(() => {});

const msgs = await redis.xreadgroup(
  'GROUP', 'workers', 'consumer-a',
  'COUNT', '5', 'BLOCK', '5000',
  'STREAMS', 'stream:jobs', '>'
);

for (const [stream, entries] of msgs || []) {
  for (const [id, fields] of entries) {
    await handleJob(fields);
    await redis.xack(stream, 'workers', id);
  }
}

Go:

_, err := rdb.XGroupCreateMkStream(ctx, "stream:jobs", "workers", "0").Result()

streams, _ := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
    Group:    "workers",
    Consumer: "consumer-a",
    Streams:  []string{"stream:jobs", ">"},
    Count:    5,
    Block:    5 * time.Second,
}).Result()

for _, s := range streams {
    for _, msg := range s.Messages {
        handleJob(msg.Values)
        rdb.XAck(ctx, "stream:jobs", "workers", msg.ID)
    }
}

Trim stream to bound memory:

127.0.0.1:6379> XADD stream:jobs MAXLEN ~ 100000 * type noop payload {}

Discussion

Consumer groups give you:

• Load balancing — each message to one consumer in the group.
• At-least-once — unacked messages stay in PEL.
• Horizontal scale — add consumers with unique names.

Design for idempotent handlers (Recipe 13.4)—at-least-once implies duplicates after reclaim.

Monitor: XINFO GROUPS, XPENDING, PEL length alerts.

See Also

• Chapter 9 — Streams
• Recipe 13.5 — dead letter queue
• Recipe 13.4 — idempotency

---

Recipe 13.3 — Delayed Jobs with Sorted Sets

Problem

Jobs must run later: send reminder in 24 hours, retry in 5 minutes, schedule report at midnight. Neither Lists nor Streams natively delay by wall-clock time.

Solution

Use a Sorted Set where score = Unix timestamp when the job becomes runnable. A scheduler polls (or blocks on nearest deadline) and moves due jobs to the main queue.

redis-cli:

127.0.0.1:6379> ZADD queue:delayed 1716979200 '{"id":"job-42","type":"reminder"}'
(integer) 1

# When now >= score, fetch due jobs:
127.0.0.1:6379> ZRANGEBYSCORE queue:delayed 0 1716979200 LIMIT 0 10
1) "{\"id\":\"job-42\",\"type\":\"reminder\"}"

127.0.0.1:6379> ZREM queue:delayed "{\"id\":\"job-42\",\"type\":\"reminder\"}"
127.0.0.1:6379> XADD stream:jobs * payload '{"id":"job-42","type":"reminder"}'

Python — scheduler:

import time
import json

DELAYED = "queue:delayed"
STREAM = "stream:jobs"

def schedule_delayed(r, job: dict, run_at: float):
    r.zadd(DELAYED, {json.dumps(job): run_at})

def promote_due_jobs(r, batch_size: int = 100):
    now = time.time()
    due = r.zrangebyscore(DELAYED, 0, now, start=0, num=batch_size)
    if not due:
        return 0
    pipe = r.pipeline()
    for payload in due:
        pipe.zrem(DELAYED, payload)
        pipe.xadd(STREAM, {"payload": payload})
    pipe.execute()
    return len(due)

def scheduler_loop(r, interval: float = 1.0):
    while True:
        promote_due_jobs(r)
        time.sleep(interval)

Atomic promote with Lua (avoid duplicate dispatch):

-- promote.lua: ZRANGEBYSCORE + ZREM + XADD atomically per job
local jobs = redis.call('ZRANGEBYSCORE', KEYS[1], 0, ARGV[1], 'LIMIT', 0, ARGV[2])
for _, job in ipairs(jobs) do
  redis.call('ZREM', KEYS[1], job)
  redis.call('XADD', KEYS[2], '*', 'payload', job)
end
return #jobs

Node.js — schedule:

const runAt = Date.now() / 1000 + 3600; // 1 hour from now
await redis.zadd('queue:delayed', runAt, JSON.stringify({ id: 'job-42' }));

Go:

runAt := float64(time.Now().Add(time.Hour).Unix())
rdb.ZAdd(ctx, "queue:delayed", redis.Z{Score: runAt, Member: payload})

Discussion

• Poll interval trades latency vs CPU. Sub-second delays need tighter loops or keyspace notifications (fragile).
• For many delayed jobs, shard ZSETs by time bucket (queue:delayed:20260528).
• Redisson, Sidekiq (via Ruby), Celery (Redis backend), and BullMQ (Node) implement this pattern; understand the primitive before adopting frameworks.

Combine with Recipe 13.2: delayed ZSET → promote → Stream → workers.

See Also

• Chapter 7 — Sorted Sets
• Recipe 13.2 — Stream workers

---

Recipe 13.4 — Idempotency Keys

Problem

At-least-once delivery means the same job may run twice. Payment captures, email sends, and inventory deductions must not double-apply.

Solution

Each job carries an idempotency key. Before processing, atomically claim the key with SET NX EX. Duplicate deliveries skip work.

redis-cli:

127.0.0.1:6379> SET idempotent:job:pay-9001-uuid "processing" NX EX 86400
OK
# Second attempt:
127.0.0.1:6379> SET idempotent:job:pay-9001-uuid "processing" NX EX 86400
(nil)

Python:

def process_idempotent(r, idempotency_key: str, ttl: int, handler):
    claim_key = f"idempotent:job:{idempotency_key}"
    if not r.set(claim_key, "processing", nx=True, ex=ttl):
        return {"status": "duplicate", "skipped": True}
    try:
        result = handler()
        r.set(claim_key, "completed", ex=ttl)
        return {"status": "ok", "result": result}
    except Exception:
        r.delete(claim_key)  # allow retry on failure
        raise

Store result for replay (optional):

def process_with_cached_result(r, key: str, ttl: int, handler):
    result_key = f"idempotent:result:{key}"
    cached = r.get(result_key)
    if cached:
        return json.loads(cached)
    if not r.set(f"idempotent:lock:{key}", "1", nx=True, ex=300):
        # Wait for in-flight duplicate
        for _ in range(50):
            time.sleep(0.1)
            cached = r.get(result_key)
            if cached:
                return json.loads(cached)
        raise ConcurrentDuplicateError(key)
    try:
        result = handler()
        r.setex(result_key, ttl, json.dumps(result))
        return result
    finally:
        r.delete(f"idempotent:lock:{key}")

Node.js:

async function processIdempotent(key, ttlSec, handler) {
  const claimKey = `idempotent:job:${key}`;
  const claimed = await redis.set(claimKey, 'processing', 'NX', 'EX', ttlSec);
  if (!claimed) return { skipped: true };
  return handler();
}

Go:

ok, err := rdb.SetNX(ctx, "idempotent:job:"+key, "processing", 24*time.Hour).Result()
if !ok {
    return ErrDuplicate
}

Discussion

• Key = business identifier: payment intent ID, webhook delivery ID, (order_id, event_type).
• TTL must exceed max retry window; 24–72 hours is common.
• On success, keep key (completed) to block replays. On transient failure, delete to allow retry.
• Idempotency complements—not replaces—database unique constraints.

See Also

• Chapter 22 — Transactions
• Recipe 13.2 — at-least-once Streams

---

Recipe 13.5 — Dead Letter Handling

Problem

A job fails repeatedly—bad payload, downstream API gone, logic bug. Infinite retry blocks the queue and alerts flood. You need a dead letter queue (DLQ) for manual inspection and reprocessing.

Solution

Track retry count in the job or PEL reclaim cycles. After N failures, move to a DLQ Stream or List and ACK the original to clear PEL.

redis-cli — dead letter stream:

127.0.0.1:6379> XADD stream:jobs:dlq * original_id 1716892800000-0 reason "max_retries" payload '{"type":"broken"}'

Python — retry with DLQ:

MAX_RETRIES = 5
RETRY_DELAY = 60  # seconds

def handle_with_retry(r, stream, group, consumer, msg_id, fields):
    retry_key = f"job:retries:{msg_id}"
    retries = int(r.get(retry_key) or 0)

    try:
        process(fields)
        r.xack(stream, group, msg_id)
        r.delete(retry_key)
    except TransientError:
        retries += 1
        r.setex(retry_key, 86400, retries)
        if retries >= MAX_RETRIES:
            move_to_dlq(r, msg_id, fields, "max_retries")
            r.xack(stream, group, msg_id)
            r.delete(retry_key)
        else:
            schedule_delayed(r, {
                "stream": stream, "group": group,
                "msg_id": msg_id, "fields": fields,
            }, time.time() + RETRY_DELAY)
            r.xack(stream, group, msg_id)
    except PermanentError as e:
        move_to_dlq(r, msg_id, fields, str(e))
        r.xack(stream, group, msg_id)

def move_to_dlq(r, original_id, fields, reason):
    r.xadd("stream:jobs:dlq", {
        "original_id": original_id,
        "reason": reason,
        "payload": json.dumps(fields),
        "failed_at": str(int(time.time())),
    })

Node.js — DLQ inspection:

const dlqEntries = await redis.xrange('stream:jobs:dlq', '-', '+', 'COUNT', 100);

Go:

rdb.XAdd(ctx, &redis.XAddArgs{
    Stream: "stream:jobs:dlq",
    Values: map[string]interface{}{
        "original_id": msgID,
        "reason":      reason,
        "payload":     payload,
    },
})

Reprocess from DLQ:

127.0.0.1:6379> XRANGE stream:jobs:dlq - + COUNT 1
127.0.0.1:6379> XADD stream:jobs * payload '...' reprocessed_from 1716892800000-0
127.0.0.1:6379> XDEL stream:jobs:dlq 1716892900000-0

Discussion

• Separate transient (retry with backoff) from permanent (DLQ immediately) errors.
• Alert on DLQ growth rate; build admin UI to replay or discard.
• Include original message ID, stack trace hash, and failure reason in DLQ entries.
• Consider XCLAIM delivery count (XPENDING shows times delivered) as retry signal instead of manual counters.

Libraries: BullMQ calls this "failed" set; AWS SQS dead-letter queues inspired the pattern.

See Also

• Recipe 13.2 — PEL and XCLAIM
• Recipe 13.3 — delayed retry
• Chapter 24 — Troubleshooting

---

Chapter Summary

Recipe	Structure	Reliability	When to use
13.1	List + BRPOP	Low	Prototypes, idempotent fire-and-forget
13.2	Streams + groups	At-least-once	Production workers
13.3	ZSET delayed	Scheduler-dependent	Scheduled / retry delays
13.4	SET NX idempotency	Exactly-once effect	Side-effectful jobs
13.5	DLQ Stream	Operational safety	Poison messages

The progression mirrors most teams' journey: List queue proves the idea; Streams with idempotency and DLQ carry production load.

---

Recipe 13.6 — Observability and Backpressure

Problem

Queues look fine until they're not: PEL grows silently, delayed ZSET never drains, workers fall behind during traffic spikes.

Solution

Instrument every queue layer with metrics and health checks.

redis-cli — queue depth inspection:

127.0.0.1:6379> XLEN stream:jobs
(integer) 4521

127.0.0.1:6379> XPENDING stream:jobs workers
1) (integer) 127
2) "1716892800000-0"
3) "1716892900000-0"
4) 1) 1) "consumer-a"
      2) "98"
   2) 1) "consumer-b"
      2) "29"

127.0.0.1:6379> ZCARD queue:delayed
(integer) 890

Python — health endpoint data:

def queue_stats(r) -> dict:
    groups = r.xinfo_groups("stream:jobs")
    pending = sum(g["pending"] for g in groups)
    return {
        "stream_length": r.xlen("stream:jobs"),
        "pending": pending,
        "delayed": r.zcard("queue:delayed"),
        "dlq_length": r.xlen("stream:jobs:dlq"),
    }

Backpressure — slow producers when queue is full:

MAX_STREAM_LEN = 100_000

def enqueue_with_backpressure(r, job: dict) -> bool:
    length = r.xlen("stream:jobs")
    if length >= MAX_STREAM_LEN:
        return False  # reject or spill to overflow storage
    r.xadd("stream:jobs", {"payload": json.dumps(job)})
    return True

Node.js — alert on PEL age:

const pending = await redis.xpending('stream:jobs', 'workers', '-', '+', 10);
for (const entry of pending) {
  if (entry.idle > 300000) { // 5 minutes
    console.warn(`Stale job ${entry.id} idle ${entry.idle}ms`);
  }
}

Go — consumer lag gauge:

info, _ := rdb.XInfoGroups(ctx, "stream:jobs").Result()
for _, g := range info {
    metrics.Gauge("redis_stream_pending", float64(g.Pending), "group", g.Name)
}

Discussion

Define SLOs: max PEL age, max stream length, max DLQ entries per hour. Scale consumers horizontally when XLEN grows linearly under steady enqueue rate. Adding consumers stops helping when the bottleneck is downstream (SMTP, S3)—detect with per-job processing timers.

Use XTRIM or MAXLEN ~ on the main stream only after jobs are ACKed; trimming unprocessed entries loses work.

Graceful shutdown: stop accepting new jobs, drain PEL with XREADGROUP on ID 0 (pending first), then exit.

See Also

• Chapter 24 — Troubleshooting
• Recipe 13.2 — consumer group mechanics

---

Choosing Your Queue Tier

Requirement	List queue	Streams	External (SQS, NATS)
Setup time	Minutes	Hours	Days (infra)
At-least-once	Manual	Built-in	Built-in
Delayed jobs	Awkward	+ ZSET	Native (SQS)
Cross-region	Replication lag	Replication lag	First-class
Ops burden	Low	Medium	Varies

Stay on Redis queues while team and traffic fit one Redis cluster. Move to dedicated brokers when you need multi-region fan-out, very long retention, or compliance features Redis does not provide.

---

← Chapter 12 — Search & Analytics | Table of Contents | Chapter 14 — Real-World Patterns →