Scheduling Escalations with Redis Sorted Sets

A dispute that has been sitting in a queue past its SLA window needs to escalate to the next reviewer tier without a human polling a dashboard. A Redis sorted set (ZSET) scored by the escalation’s due-time epoch turns that requirement into a priority timeline: ZADD schedules the event, a poller drains whatever has crossed the score threshold, and a short-lived lock keeps two workers from firing the same escalation twice. Redis is a reasonable place to run this rather than a database-backed cron sweep because the ZSET already sorts by score server-side — a poller never scans a full table to find the single most-overdue item, it just asks Redis for the lowest score. This page implements that scheduler end to end and slots underneath SLA-Gated Escalation Rules as the mechanism that actually fires the escalations the rule engine defines. It assumes the item already missed the window enforced by Tracking Dispute Resolution SLAs in Python and needs to land in a higher Manual Review Queue tier. The mechanics below generalize past dispute SLAs to any workflow where “fire this event no earlier than time T, exactly once” is the requirement.

Prerequisites

Step 1 — Model the escalation and enqueue it with ZADD

The ZSET only needs to hold a member identifier and a score (the deadline as a UTC epoch float); the full payload — including the Decimal materiality used to weight priority — lives in a companion hash keyed by the same identifier. Keeping the payload out of the ZSET member string means the sorted set stays cheap to scan and re-score, and it means a promotion (Step 4) can rewrite the score without touching or re-serializing the payload.

python
from __future__ import annotations

import json
import time
from datetime import datetime, timezone
from decimal import Decimal
from uuid import UUID

import redis

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

PENDING_ZSET = "escalations:pending"
PAYLOAD_HASH_PREFIX = "escalation:payload:"

def enqueue_escalation(
    dispute_id: UUID,
    tier: str,
    deadline_utc: datetime,
    materiality_usd: Decimal,
    source_hash: str,
) -> None:
    if deadline_utc.tzinfo is None:
        raise ValueError("deadline_utc must be timezone-aware UTC")
    due_epoch = deadline_utc.astimezone(timezone.utc).timestamp()
    payload = {
        "dispute_id": str(dispute_id),
        "tier": tier,
        "materiality_usd": str(materiality_usd.quantize(Decimal("0.01"))),
        "source_hash": source_hash,
        "enqueued_at": datetime.now(timezone.utc).isoformat(),
    }
    pipe = r.pipeline(transaction=True)
    pipe.hset(f"{PAYLOAD_HASH_PREFIX}{dispute_id}", mapping=payload)
    pipe.zadd(PENDING_ZSET, {str(dispute_id): due_epoch})
    pipe.execute()

Scoring by epoch seconds turns the ZSET into a min-heap ordered by urgency for free — ZRANGE ... WITHSCORES 0 0 always returns the single most-overdue member without a separate priority index. The pipe.execute() call matters: writing the payload hash and the ZSET score in one MULTI/EXEC transaction means a poller can never observe a scored member whose payload hash hasn’t landed yet, which is exactly the scenario that produces a STUCK_MEMBER later.

Step 2 — Poll due members with ZRANGEBYSCORE and ZPOPMIN

ZRANGEBYSCORE gives a non-destructive read of everything at or below now, which is useful when a downstream step can fail and you want the member to remain claimable. ZPOPMIN is the atomic alternative: it always pops the single lowest-scored member regardless of whether that score is actually due, so it must be paired with a score check and a push-back if the earliest item hasn’t matured yet.

python
def fetch_due(batch_size: int) -> list[tuple[str, float]]:
    now = time.time()
    return r.zrangebyscore(
        PENDING_ZSET, min=0, max=now, start=0, num=batch_size, withscores=True
    )

def pop_if_due() -> tuple[str, float] | None:
    popped = r.zpopmin(PENDING_ZSET, count=1)
    if not popped:
        return None
    member, score = popped[0]
    if score > time.time():
        r.zadd(PENDING_ZSET, {member: score})  # not due yet, push back
        return None
    return member, score

Production pollers favor ZRANGEBYSCORE for batch draining — it lets a single loop iteration claim, lock, and process up to batch_size members in one pass instead of round-tripping through Redis per item. ZPOPMIN remains useful for a single-consumer worker that wants strict lock-step ordering and is willing to pay the push-back round trip whenever it races ahead of the clock; most fleets run several pollers in parallel and prefer the read-then-lock-then-remove flow described in Steps 3 and 4 because it degrades gracefully under contention instead of thrashing pushes and pops.

Step 3 — Guard against double firing with a SETNX lock

Two poller processes can both read the same due member in the window between ZRANGEBYSCORE and the eventual ZREM. A short-lived lock, acquired with SET ... NX EX, makes the escalation step idempotent: only the worker that wins the lock is allowed to notify the next tier and remove the member from the ZSET.

python
LOCK_PREFIX = "escalation:lock:"

def try_acquire_lock(dispute_id: str, lock_ttl_s: int, worker_id: str) -> bool:
    return bool(
        r.set(f"{LOCK_PREFIX}{dispute_id}", worker_id, nx=True, ex=lock_ttl_s)
    )

def release_lock(dispute_id: str, worker_id: str) -> None:
    # Only release if we still hold it — avoids releasing a lock another
    # worker acquired after ours expired mid-processing.
    lock_key = f"{LOCK_PREFIX}{dispute_id}"
    if r.get(lock_key) == worker_id:
        r.delete(lock_key)

lock_ttl_s must comfortably exceed the worst-case processing time for one escalation notification, or the lock expires mid-flight and a second worker acquires it while the first is still delivering — the classic LOCK_CONTENTION failure covered below. SET NX EX is preferable to a separate SETNX plus EXPIRE pair because it is a single atomic command; splitting the two leaves a window where a process can crash after acquiring the lock but before setting its expiry, producing a lock that never releases.

Step 4 — Promote the tier and emit the audit record

A successful escalation does three things atomically from the caller’s perspective: it notifies the next tier’s review queue, re-scores the member into the next SLA window (or removes it entirely if the ladder is exhausted), and writes the audit triple.

python
import logging

log = logging.getLogger("sla.escalation")

TIER_LADDER = ["standard", "senior", "compliance"]

def promote_and_log(
    dispute_id: str, worker_id: str, lock_ttl_s: int, next_sla_seconds: int
) -> str:
    if not try_acquire_lock(dispute_id, lock_ttl_s, worker_id):
        log.warning("escalation.lock_contended", extra={"trace_id": dispute_id})
        return "LOCK_CONTENTION"
    try:
        payload = r.hgetall(f"{PAYLOAD_HASH_PREFIX}{dispute_id}")
        if not payload:
            return "STUCK_MEMBER"
        current_tier = payload["tier"]
        idx = TIER_LADDER.index(current_tier)
        if idx + 1 >= len(TIER_LADDER):
            decision = "TERMINAL_ESCALATION"
            r.zrem(PENDING_ZSET, dispute_id)
        else:
            next_tier = TIER_LADDER[idx + 1]
            next_due = time.time() + next_sla_seconds
            r.hset(f"{PAYLOAD_HASH_PREFIX}{dispute_id}", "tier", next_tier)
            r.zadd(PENDING_ZSET, {dispute_id: next_due})
            decision = f"PROMOTED_TO_{next_tier.upper()}"
        log.info(
            "escalation.promoted",
            extra={
                "trace_id": dispute_id,
                "source_hash": payload["source_hash"],
                "match_decision": decision,
            },
        )
        return decision
    finally:
        release_lock(dispute_id, worker_id)

The lock wraps the read-modify-write of both the payload hash and the ZSET so a concurrent poller sees either the fully-promoted state or nothing at all, never a half-applied tier change. This mirrors the append-only discipline used for dispute state in Tracking Dispute Resolution SLAs in Python — every transition is logged, never silently overwritten.

Configuration boundary table

Parameter Default Valid range Notes
poll_interval_s 5 1–60 Base cadence for ZRANGEBYSCORE sweeps
batch_size 50 1–500 Max members claimed per poll iteration
lock_ttl_s 30 5–300 Must exceed worst-case delivery latency
tier_ladder ["standard", "senior", "compliance"] ordered list Terminal tier stops re-enqueueing
jitter 0.2 0.0–0.5 Fractional randomization on poll_interval_s to avoid thundering-herd polls

Add jitter by sampling poll_interval_s * (1 + random.uniform(-jitter, jitter)) before each sleep — without it, every poller in a fleet wakes on the same tick and contends for the same due members. The tier_ladder order matters beyond naming: promote_and_log walks it with list.index, so reordering the ladder in a deploy silently changes what “next tier” means for every dispute already scored against the old ordering. Treat a ladder change as a migration, not a config edit — re-read every pending member’s current tier and re-validate its position before the new ladder goes live.

Verification and testing

Time-travel the ZSET by enqueueing a member with a deadline a few seconds in the past and asserting it is immediately claimable, then re-enqueueing one in the future and asserting it is not.

python
from datetime import timedelta
from uuid import uuid4

def test_due_member_is_claimable(monkeypatch):
    r.delete(PENDING_ZSET)
    dispute_id = uuid4()
    past_deadline = datetime.now(timezone.utc) - timedelta(seconds=5)
    enqueue_escalation(
        dispute_id, "standard", past_deadline, Decimal("1250.00"), "a1b2c3"
    )
    due = fetch_due(batch_size=10)
    assert any(member == str(dispute_id) for member, _ in due)

def test_future_member_is_not_claimable():
    r.delete(PENDING_ZSET)
    dispute_id = uuid4()
    future_deadline = datetime.now(timezone.utc) + timedelta(minutes=10)
    enqueue_escalation(
        dispute_id, "standard", future_deadline, Decimal("980.50"), "d4e5f6"
    )
    due = fetch_due(batch_size=10)
    assert not any(member == str(dispute_id) for member, _ in due)

def test_double_escalation_is_blocked():
    dispute_id, worker_a, worker_b = str(uuid4()), "worker-a", "worker-b"
    r.hset(
        f"{PAYLOAD_HASH_PREFIX}{dispute_id}",
        mapping={"tier": "standard", "source_hash": "abc123"},
    )
    assert try_acquire_lock(dispute_id, lock_ttl_s=30, worker_id=worker_a)
    assert try_acquire_lock(dispute_id, lock_ttl_s=30, worker_id=worker_b) is False

Run the fixture against a disposable Redis database (FLUSHDB in a test-only DB index, never the default), and assert on the returned match_decision string rather than on internal Redis state, so the test still passes if the storage layout changes. For a full drain test, seed a handful of members with deadlines spread across the past and future, run one poller iteration, and assert that ZCARD on escalations:pending decreased by exactly the number of past-due members — a mismatch here is usually the first sign of the ZSET_DRIFT symptom described below, caught in CI before it reaches production.

Troubleshooting

  • DOUBLE_ESCALATION — the same dispute notifies two review queues. Root cause: promote_and_log ran without first acquiring the SETNX lock, usually because a caller bypassed try_acquire_lock for a “fast path.” Fix: route every promotion through the locked function; never call ZREM/ZADD for promotion outside the lock.
  • CLOCK_SKEW — a member is claimed well before or after its wall-clock deadline. Root cause: worker hosts computed due_epoch from local time instead of datetime.now(timezone.utc), or NTP drift diverged between the enqueueing and polling hosts. Fix: always convert with .astimezone(timezone.utc) before .timestamp(), and run NTP sync on every host that calls enqueue_escalation or fetch_due.
  • STUCK_MEMBER — a dispute ID sits in the ZSET with no matching payload hash. Root cause: the payload hash expired (a TTL was mistakenly set on it) or was deleted out of band while the ZSET entry remained. Fix: never TTL the payload hash independently of the ZSET member; delete both together, and have the poller ZREM any member whose HGETALL returns empty.
  • LOCK_CONTENTION — a promotion consistently loses the lock race. Root cause: lock_ttl_s is shorter than actual delivery latency to the next tier’s queue, so the lock expires and a second worker grabs it mid-processing. Fix: raise lock_ttl_s above p99 delivery latency, or split notification and promotion into a retryable two-phase sequence so a lost lock only delays rather than duplicates the escalation.
  • ZSET_DRIFT — ZCARD on escalations:pending grows without bound. Root cause: promote_and_log reached the terminal tier but a caller re-enqueued the member anyway, or an exception between ZREM and lock release left the member orphaned. Fix: guard the terminal branch so TERMINAL_ESCALATION never calls ZADD, and wrap the whole promotion body in try/finally as shown in Step 4.

Part of SLA-Gated Escalation Rules within Exception Routing & Human-in-the-Loop Workflows.