Implementing Priority Queues with Redis Sorted Sets

This is the per-message priority mechanism referenced in Priority Queues & Job Fairness, under Queue Fundamentals & Architecture: one queue, an ordering score, and an atomic pop that survives worker crashes.

Problem Statement

Your job queue is a Redis list, so ordering is strictly FIFO. Product now needs urgent jobs to jump ahead of a 50,000-job import backlog, and adding a second list only gives you three coarse classes with a starvation problem. You want a single queue where each message carries a numeric priority, the highest-priority job is always served first, jobs of equal priority still run oldest-first, and a worker that dies mid-job does not lose the message.

Prerequisites

  • Redis 6.2+ (ZPOPMIN exists from 5.0; EXPIRE options used here need 6.2). Cluster mode requires all keys in one hash slot — use {jobs} hash tags.
  • A serialisation format for the job envelope — JSON here, though the payload sizing advice in message size limits and serialization applies.
  • A worker loop you control, since this replaces BLPOP with a scripted pop.

Step 1 — Design the Score

A sorted set orders by a single float, so the score must encode priority first and arrival time second. Reserving the integer part for priority and the fractional part for a normalised timestamp gives strict priority with FIFO tie-breaking inside each class.

import time

EPOCH = 1_750_000_000        # fixed reference point; keeps the fraction small
SPAN  = 10 ** 10             # ~317 years of seconds → fraction stays < 1

def score(priority: int, now: float | None = None) -> float:
    """Lower score = served earlier. priority 0 is most urgent.
    Composite: priority.<normalised arrival time>"""
    now = now if now is not None else time.time()
    return priority + ((now - EPOCH) / SPAN)

Float64 gives about 15–16 significant digits, so a priority in the single digits leaves roughly 14 digits for the fraction — sub-millisecond resolution for centuries. Do not be tempted to multiply the timestamp into the integer part: that reintroduces FIFO-across-priorities the moment the clock advances past your multiplier.

Composite score layout in the sorted set Five jobs are shown with their composite scores. The integer part of each score is the priority class and the fractional part is the normalised arrival time. Sorted ascending, all priority-zero jobs come first in arrival order, then priority-one jobs, then priority-two, demonstrating strict priority with first-in-first-out tie-breaking. score = priority . arrival-time integer part → priority fraction → normalised arrival ZPOPMIN takes the smallest 0.4412 2fa_code urgent, arrived first → served first 0.4419 password_reset same class, later arrival → second 1.4401 invoice_pdf normal — waits even though it arrived earliest 2.4405 nightly_digest bulk — drains when the classes above are empty Strict ordering: give bulk a floor share or it starves under sustained urgent load.

Step 2 — Enqueue with the Envelope in a Hash

Keep the sorted set small: store only job IDs there, with the payload in a hash. A sorted set whose members are 40KB JSON blobs makes every ZADD and ZRANGE expensive and inflates the memory that ordering costs you.

import json
import uuid
import redis

r = redis.Redis(decode_responses=True)
QUEUE   = "{jobs}:pending"      # sorted set: id → score
PAYLOAD = "{jobs}:payload"      # hash: id → JSON envelope

def enqueue(task: str, args: dict, priority: int = 1) -> str:
    job_id = uuid.uuid4().hex
    envelope = json.dumps({"id": job_id, "task": task, "args": args,
                           "priority": priority, "enqueued_at": time.time()})
    pipe = r.pipeline()
    pipe.hset(PAYLOAD, job_id, envelope)
    pipe.zadd(QUEUE, {job_id: score(priority)})
    pipe.execute()              # payload written before the id becomes visible
    return job_id

Order matters inside the pipeline: write the payload first. If the ZADD landed first and the process died, a worker could pop an ID whose payload does not exist yet.

Step 3 — Pop Atomically into a Processing Set

ZPOPMIN alone is at-most-once: the job leaves the queue and vanishes if the worker crashes before finishing. Move it to a processing sorted set scored by a lease deadline in the same script, and the pattern becomes at-least-once, exactly like a broker's visibility timeout.

-- claim.lua — KEYS[1]=pending KEYS[2]=processing KEYS[3]=payload
-- ARGV[1]=now ARGV[2]=lease_seconds ARGV[3]=worker_id
local popped = redis.call('ZPOPMIN', KEYS[1], 1)
if #popped == 0 then return nil end
local job_id = popped[1]
redis.call('ZADD', KEYS[2], tonumber(ARGV[1]) + tonumber(ARGV[2]), job_id)
redis.call('HSET', KEYS[3] .. ':owner', job_id, ARGV[3])
return redis.call('HGET', KEYS[3], job_id)
CLAIM = r.register_script(open("claim.lua").read())

def claim(worker_id: str, lease: int = 60):
    raw = CLAIM(keys=[QUEUE, "{jobs}:processing", PAYLOAD],
                args=[time.time(), lease, worker_id])
    return json.loads(raw) if raw else None

def ack(job_id: str) -> None:
    pipe = r.pipeline()
    pipe.zrem("{jobs}:processing", job_id)
    pipe.hdel(PAYLOAD, job_id)
    pipe.hdel(PAYLOAD + ":owner", job_id)
    pipe.execute()

Step 4 — Reclaim Expired Leases

A crashed worker leaves its job in the processing set with a deadline in the past. A sweeper returns those to the pending set, incrementing an attempt counter so poison messages eventually reach the dead-letter queue rather than looping forever.

MAX_ATTEMPTS = 5

def reclaim(now: float | None = None) -> int:
    now = now or time.time()
    expired = r.zrangebyscore("{jobs}:processing", "-inf", now, start=0, num=100)
    for job_id in expired:
        raw = r.hget(PAYLOAD, job_id)
        if raw is None:                      # acked concurrently — nothing to do
            r.zrem("{jobs}:processing", job_id)
            continue
        env = json.loads(raw)
        env["attempts"] = env.get("attempts", 0) + 1
        pipe = r.pipeline()
        pipe.zrem("{jobs}:processing", job_id)
        if env["attempts"] >= MAX_ATTEMPTS:
            pipe.lpush("{jobs}:dlq", json.dumps(env))
            pipe.hdel(PAYLOAD, job_id)
        else:
            pipe.hset(PAYLOAD, job_id, json.dumps(env))
            # Re-enqueue at the SAME priority, with a fresh arrival time so it
            # queues behind peers rather than jumping the whole class.
            pipe.zadd(QUEUE, {job_id: score(env["priority"])})
        pipe.execute()
    return len(expired)
Claim, acknowledge, reclaim, dead-letter A job in the pending sorted set is claimed atomically into a processing set with a lease deadline. On acknowledgement it is deleted. If the lease expires the sweeper returns it to pending with an incremented attempt count, and once attempts reach the maximum it is pushed to the dead-letter list. Reliable pop without losing the priority order pending (zset) ordered by score claim.lua processing (zset) score = lease deadline ack deleted lease expired → reclaim, attempts + 1 attempts = 5 dead-letter list Run the sweeper on every worker; it is idempotent and cheap at 100 ids per pass.

Step 5 — Give Bulk Work a Floor

Strict score ordering starves the lowest class under sustained urgent load. The cheapest correction is an occasional deliberate pop from the tail: on a configurable fraction of iterations, claim the oldest bulk job instead of the highest-priority one.

import random

BULK_FLOOR = 0.1     # 10% of claims are taken from the bulk end

def claim_with_floor(worker_id: str, lease: int = 60):
    if random.random() < BULK_FLOOR:
        # Highest score = lowest priority, oldest within it.
        tail = r.zrange(QUEUE, -1, -1)
        if tail:
            return claim_specific(tail[0], worker_id, lease)
    return claim(worker_id, lease)

Ten percent is usually enough to keep a bulk queue draining without materially moving urgent p99 latency. Measure both before tuning it, using the oldest-message-age signal described in Priority Queues & Job Fairness.

Step 6 — Know What This Costs Compared to a List

A sorted set is not free. ZADD and ZPOPMIN are O(log N) against a list's O(1), and the skip-list representation carries more per-entry overhead than a quicklist. The practical numbers on a single Redis instance are worth knowing before you convert a hot path.

Operation List (LPUSH/BLPOP) Sorted set (ZADD/ZPOPMIN)
Complexity O(1) O(log N)
Approx. throughput at 1M entries ~180k ops/s ~120k ops/s
Memory per entry (id only, 32 chars) ~64 B ~140 B
Blocking pop BLPOP BZPOPMIN (not scriptable)
Ordering arrival only any score you compute

At a million queued jobs the memory difference is roughly 75MB — irrelevant on a 16GB instance, significant on a 512MB one, which is the trade the sizing guidance in in-memory vs persistent queue storage is really about. The throughput difference only becomes a bottleneck above roughly 100k enqueues per second, at which point the queue itself is no longer your constraint. For nearly every workload, paying 30% throughput for real priority ordering is the right trade; where it is not, keep three separate lists and poll them with weights instead.

A deliberate bulk floor inside a strict-priority queue Ten consecutive worker claims are shown. Nine take the lowest-scored job, which is the highest priority. One claim, chosen at random with ten percent probability, deliberately takes the highest-scored job at the tail of the sorted set, which is the oldest bulk work. This bounds bulk wait time without materially changing urgent latency. Ten claims: nine from the head, one from the tail bulk blue = ZPOPMIN, the highest-priority waiting job orange = the deliberate tail claim that keeps bulk draining A 10% floor costs urgent work about 10% of throughput and turns unbounded bulk wait into bounded wait.

Verification

  1. Ordering is correct. Enqueue priority 2, then 0, then 1, then another 0. Claims must return the two priority-0 jobs in enqueue order, then priority 1, then 2.
  2. The pop is atomic. Run 50 concurrent workers against 10,000 jobs and assert that the set of processed IDs has no duplicates and no gaps: redis-cli ZCARD {jobs}:pending must reach 0 with exactly 10,000 acks.
  3. Crashes are recovered. kill -9 a worker holding a job, then confirm the sweeper returns it to pending within one lease period with attempts = 1.
  4. Bulk keeps moving. Under a continuous priority-0 load, the oldest bulk job's age must plateau rather than grow without bound.
# Live view of the queue by class
redis-cli ZRANGE '{jobs}:pending' 0 -1 WITHSCORES | paste - - | head
redis-cli ZCOUNT '{jobs}:pending' 0 0.999999   # urgent backlog
redis-cli ZCOUNT '{jobs}:processing' -inf "$(date +%s)"   # expired leases

Gotchas & Edge Cases

Cluster mode needs hash tags. A Lua script touching pending, processing, and payload requires all three in one slot. The {jobs} prefix used above pins them; without it the script fails with CROSSSLOT.

Float precision is finite. Encoding a millisecond timestamp into the fraction without normalising it (dividing by a large constant) overruns float64's significant digits and collapses ordering inside a class. Keep the fraction below 1 and derived from a fixed epoch.

The processing set is unbounded if nothing sweeps. If every worker relies on some other worker to run reclaim, a fleet-wide restart leaves jobs leased forever. Run the sweeper in-process on every worker.

Lease length must exceed the slowest job. A lease shorter than the job's runtime causes the sweeper to re-enqueue work that is still running — duplicate execution with no error anywhere. Either lease generously or extend the deadline periodically from the worker, the same heartbeat pattern used for long-running workers.

ZPOPMIN has no blocking form with scripts. BZPOPMIN blocks but cannot be wrapped in Lua. Either poll with a short sleep (10–50ms) or use BZPOPMIN and accept a small race window on the processing-set insert, reconciled by the sweeper.

Related