“At-least-once delivery” is written in documentation as a guarantee. Read it again as a warning: your handler will run more than once on the same message, and the system is telling you in advance so you cannot claim surprise.

The processing plane in NewsCrawl runs cleaning, LLM extraction and embedding as separate Redis Streams consumer groups. Here is what it took to make at-least-once actually safe.

Consumer groups in one paragraph

XADD appends to a stream. A consumer group tracks a delivery cursor and a Pending Entries List - messages delivered to some consumer but not yet acknowledged. XREADGROUP delivers new messages and moves them into the PEL. XACK removes them. Anything in the PEL is, by definition, work someone claimed and did not finish.

The PEL is the whole reliability story. It is also the thing that quietly fills up when you get this wrong.

Rule 1: idempotency is the handler’s job, not the broker’s

There is no broker configuration that gives you exactly-once. Not in Redis, not in Kafka, not in SQS. The delivery layer can guarantee “at least once”; only the handler can make a redelivery harmless.

The mechanism that works is a deterministic key derived from the content, not from the message id:

def handle(msg_id: str, fields: dict) -> None:
    doc_id = fields["doc_id"]
    stage = "embedding"
    # Content-derived key: the same document always produces the same key,
    # no matter how many times or through which message it arrives.
    key = f"{stage}:{doc_id}:{fields['content_hash']}"

    with db.transaction():
        if db.exists("stage_results", key=key):
            redis.xack(STREAM, GROUP, msg_id)   # already done - ack and move on
            return

        result = embed(fields["text"])
        db.upsert("stage_results", key=key, result=result)

    redis.xack(STREAM, GROUP, msg_id)

Two details that matter more than they look:

The key is content-derived. Keying on the Redis message id fails immediately - a reclaimed message gets a new id, so an id-keyed guard sees fresh work every time.

The XACK is after the commit, not before. Ack first and a crash in between loses the message permanently. Commit first and a crash in between means a redelivery - which the guard at the top absorbs. Choose the failure you can recover from.

Rule 2: retries need to be delayed, and delays need somewhere to live

Redis Streams has no native delayed redelivery. If you handle a failure by not acking, the message sits in the PEL until something reclaims it - which means your retry cadence is your reclaim interval, for every failure, regardless of cause.

I use a sorted set as a delay buffer:

def defer(msg_id: str, fields: dict, attempt: int) -> None:
    delay = min(2 ** attempt * 5, 900)          # 5s → 15min ceiling
    ready_at = time.time() + delay
    redis.zadd(DELAY_SET, {json.dumps({**fields, "attempt": attempt}): ready_at})
    redis.xack(STREAM, GROUP, msg_id)           # release the PEL slot

# a scheduler pass, every second
def promote() -> None:
    now = time.time()
    due = redis.zrangebyscore(DELAY_SET, 0, now, start=0, num=100)
    for payload in due:
        if redis.zrem(DELAY_SET, payload):      # atomic claim - only one promoter wins
            redis.xadd(STREAM, json.loads(payload))

The zrem-returns-1 check is the concurrency guard. Multiple scheduler instances can run this loop; only the one whose ZREM actually removed the member republishes.

And note that deferring acks the original message. A failed message that stays in the PEL is a message that will also be reclaimed, and now the same work exists in two places.

Rule 3: a dead-letter stream, with the context attached

Some messages will never succeed. A malformed document, a URL that 404s permanently, a schema the extractor cannot satisfy. Retrying those forever burns capacity that healthy work needs.

After a bounded number of attempts, the message goes to a dead-letter stream - and the useful part is what travels with it:

redis.xadd(DLQ_STREAM, {
    **fields,
    "failed_at": now_iso(),
    "attempts": attempt,
    "stage": stage,
    "error_type": type(exc).__name__,
    "error": str(exc)[:1000],
    "traceback": tb[:4000],
})

A DLQ without the error context is a graveyard. A DLQ with it is a triage queue - you can group by error_type, notice that four hundred failures share one cause, fix it, and replay the batch back onto the main stream.

Replay is just an XADD back to the source stream. Which works precisely because Rule 1 made the handlers idempotent.

Rule 4: reclaim what the dead left behind

A worker killed mid-message leaves an entry in the PEL that nobody will ever ack. XAUTOCLAIM sweeps them:

def reclaim() -> None:
    cursor = "0-0"
    while True:
        cursor, messages, _ = redis.xautoclaim(
            STREAM, GROUP, MY_CONSUMER,
            min_idle_time=STAGE_VISIBILITY_MS,
            start=cursor, count=50,
        )
        for msg_id, fields in messages:
            delivery_count = pel_delivery_count(msg_id)
            if delivery_count > MAX_DELIVERIES:
                to_dlq(msg_id, fields, reason="poison")   # kills the poison-pill loop
            else:
                handle(msg_id, fields)
        if cursor == "0-0":
            break

min_idle_time must exceed your slowest legitimate processing time, or you will reclaim work from consumers that are merely slow - and then two workers are processing the same message concurrently. When I added LLM extraction, p99 went from two seconds to forty, and my ten-second idle threshold started stealing work from healthy consumers. Visibility timeout is a per-stage property, not a global constant.

The delivery_count check is the poison-pill guard. A message that crashes its handler will be reclaimed, crash again, be reclaimed again - forever, consuming a worker each cycle. Bounding deliveries turns an infinite loop into a DLQ entry.

What I watch

Four signals, and they are the same four for any stream system:

  • PEL depth per group. Rising means consumers are claiming faster than they finish. This leads every other symptom.
  • Oldest PEL entry age. A single stuck message that reclaim is not catching.
  • DLQ arrival rate. A step change here is a deploy that broke something.
  • Stream length vs. consumed offset. The genuine backlog, independent of consumer health.

Queue depth is the metric that tells you about an incident before your users do. It is worth wiring to Prometheus on the first day, not the day after the first outage.

The short version

Redis Streams gives you durable delivery and per-group cursors. It does not give you correctness. Correctness is four things you build on top:

  1. Handlers keyed on content, so redelivery is a no-op.
  2. Delayed retries in a sorted set, so failures back off instead of spinning.
  3. A dead-letter stream carrying enough context to triage and replay.
  4. Reclaim with a per-stage idle threshold and a delivery-count ceiling.

Skip any one and at-least-once becomes at-least-once-and-occasionally-wrong.