Replaying Dead-Letter Geofence Triggers Safely

A dead-letter topic is a holding pen, not a graveyard — most of what lands there is a trigger that failed because a downstream was briefly down, a schema was mid-migration, or a dependency timed out, and once the underlying cause is fixed those triggers should be delivered. The hazard is that a naive drain — republish every DLQ record back onto the main stream — double-fires every trigger that had partially executed before failing, double-charges a surge fare, and re-sends an ENTER alert a rider already saw. This page sits under dead-letter topics and poison-message handling and the broader event routing and backpressure architecture, and it answers one operational question: how do you drain the DLQ back into production without the replay itself becoming a duplicate-generation machine or a second outage. The answer rests on three properties — idempotent replay keyed on a deterministic idempotency key, a rate-limited bounded drain, and re-quarantine of anything still poison.

Concept and specification #

Safe replay is the composition of three invariants. First, idempotency: every trigger carries a deterministic idempotency key (see idempotent trigger emission semantics), and the replay path checks that key against a dedup store before re-emitting, so a trigger whose side effect already succeeded is dropped rather than re-fired. Second, rate limiting: the drain runs at a fraction of main-stream capacity so replaying a large backlog cannot itself saturate the downstream that just recovered. Third, re-quarantine: a record that fails again during replay is not retried forever — it returns to the DLQ (or a dlq-parked topic) with its attempt count intact.

The duplicate-suppression guarantee is a set-membership test. Let $K$ be the set of idempotency keys the downstream has already durably processed. For a DLQ record with key $k$, replay emits the trigger only if $k\notin K$, then adds $k$ to $K$. Because the key is deterministic over (device_id, fence_id, transition, event_time_bucket), a replayed trigger and its original produce the same $k$, so the membership test collapses the two to one delivery. The cost of the test is a single dedup-store lookup, $O(1)$, against a Redis SETNX or an in-process TTL LRU.

Re-quarantine has its own budget. A DLQ drain is not a guarantee that every record replays successfully — a fraction were quarantined for a structural reason that the fix did not address, and those must not loop. The replay worker carries a replay_attempt counter in the record’s own envelope (distinct from the original delivery-attempt count), and a record that fails replay increments it and returns to a parked topic once it crosses a small ceiling. Budgeting for this, a healthy post-fix drain should replay >95% of records successfully and park <5%; a parked fraction above that means the root-cause fix was incomplete and the drain should be halted before it churns the DLQ. This keeps the drain monotone: every pass strictly reduces the backlog, and no record cycles indefinitely between the DLQ and the replay worker.

The drain rate is bounded by a token bucket. For a downstream that sustains $R$ triggers/sec at health, the replay budget is $\rho R$ with $\rho \in [0.1,0.3]$ so live traffic keeps the majority of capacity:

$$r_{\text{replay}}=\rho \cdot R,t_{\text{drain}}=\frac{N_{\text{dlq}}}{r_{\text{replay}}}$$

so draining a 90,000-record backlog at $\rho =0.2$ against a 25k/sec downstream ($r_{\text{replay}}=5,000$/sec) takes ~18 seconds of steady drain — fast enough to recover, slow enough to leave 80% of capacity for live triggers.

Step-by-step implementation #

Prerequisites: Python 3.11+, aiokafka>=0.10, redis>=5.0 (async client), a DLQ topic whose records carry the enriched envelope (origin_*, reason, payload_b64) written by the consumer, and triggers carrying an idempotency_key. The replay worker is intentionally a separate process from the main consumer so its rate limit and failures are isolated.

1. Gate every replay on the idempotency key. Check-and-set against the dedup store; if the key is already present, the trigger’s side effect already happened and the DLQ record is dropped.

import redis.asyncio as redis

DEDUP_TTL_S: int = 7 * 24 * 3600  # >= DLQ retention so no old dup slips through

async def already_delivered(r: redis.Redis, key: str) -> bool:
    # SETNX returns True only for a first-ever key; a False means the trigger
    # was already processed, so replay must NOT re-fire it.
    was_new: bool = await r.set(name=f"idem:{key}", value=b"1",
                                nx=True, ex=DEDUP_TTL_S)
    return not was_new

Gotcha: set the key after the downstream side effect is durable, not before. Setting it first, then crashing before delivery, marks an undelivered trigger as delivered and silently drops it on replay.

2. Bound the drain with a token bucket. Refill at rate tokens/sec; each replayed record costs one token. This is the leaky-bucket discipline covered in token-bucket vs leaky-bucket for telemetry shedding.

import asyncio
import time
from dataclasses import dataclass

@dataclass(slots=True)
class TokenBucket:
    rate: float                 # tokens per second = replay budget
    capacity: float             # max burst
    _tokens: float = 0.0
    _last: float = 0.0

    async def acquire(self) -> None:
        while True:
            now = time.monotonic()
            self._tokens = min(self.capacity,
                               self._tokens + (now - self._last) * self.rate)
            self._last = now
            if self._tokens >= 1.0:
                self._tokens -= 1.0
                return
            await asyncio.sleep((1.0 - self._tokens) / self.rate)  # wait exact deficit

3. Drain in bounded batches, dedup, re-emit, re-quarantine. The worker reads a small batch, replays only novel keys, and returns any record that fails again to the DLQ with its attempt count preserved.

import json
from aiokafka import AIOKafkaConsumer, AIOKafkaProducer

MAIN_TOPIC = "geofence.triggers"
DLQ_TOPIC = "geofence.triggers.dlq"
RE_CEILING = 3

async def replay_worker(
    consumer: AIOKafkaConsumer,     # subscribed to DLQ_TOPIC, manual commit
    producer: AIOKafkaProducer,
    r: redis.Redis,
    bucket: TokenBucket,
) -> dict[str, int]:
    stats = {"replayed": 0, "suppressed": 0, "requarantined": 0}
    async for record in consumer:
        envelope = json.loads(record.value)
        payload = envelope["payload_b64"].encode("latin1")
        trigger = json.loads(payload)
        key: str = trigger["idempotency_key"]

        if await already_delivered(r, key):
            stats["suppressed"] += 1                 # duplicate collapsed, no emit
            await consumer.commit()
            continue

        await bucket.acquire()                       # rate-limited drain
        try:
            await producer.send_and_wait(MAIN_TOPIC, value=payload, key=record.key)
            stats["replayed"] += 1
        except Exception:                            # still failing -> re-quarantine
            attempt = int(envelope.get("replay_attempt", 0)) + 1
            if attempt < RE_CEILING:
                envelope["replay_attempt"] = attempt
                await producer.send_and_wait(DLQ_TOPIC, value=json.dumps(envelope).encode())
            stats["requarantined"] += 1              # parked, not looped forever
        await consumer.commit()                      # commit after durable outcome
    return stats

Gotcha: commit the DLQ consumer offset only after the replay outcome (emit or re-quarantine) is durable. Commit-first drops any record whose replay crashes mid-flight.

4. Preserve per-device ordering where it matters. Replay can reorder a device’s transitions relative to live traffic. Because the downstream is idempotent and transition-aware, an out-of-order replayed ENTER is deduped against the live state rather than double-firing — but if strict ordering is required, key the replay producer by device_id so all of one device’s replayed triggers land on the same partition in DLQ order.

Benchmark and verification #

The metric that matters is the duplicate-trigger rate the downstream observes after a drain — the fraction of replayed triggers that fire a side effect a second time. Measured over a 90k-record DLQ backlog replayed against a recovered 25k/sec downstream, comparing a naive republish loop with no dedup against the idempotent, rate-limited worker above:

The naive loop drains faster in wall-clock terms but re-saturates the downstream (2,900 fresh 5xx) and re-fires 4.1% of triggers — the ones whose side effect had succeeded before the acknowledgement was lost. The idempotent worker suppresses those against the dedup store (0% duplicates), paces the drain so live P99 barely moves (+3ms), and correctly parks 611 genuinely poison records instead of looping them. A minimal verification harness confirms the suppression:

import time

async def verify_no_double_fire(r: redis.Redis, keys: list[str]) -> dict[str, float]:
    # Replay each key twice; a correct dedup store admits it exactly once.
    t0 = time.perf_counter()
    fired = 0
    for key in keys * 2:                             # each key presented twice
        if not await already_delivered(r, key):
            fired += 1
    return {"unique_fires": fired, "expected": len(set(keys)),
            "us_per_check": (time.perf_counter() - t0) / (len(keys) * 2) * 1e6}

unique_fires must equal len(set(keys)) — every trigger fires exactly once across two presentations — and us_per_check should stay under ~200µs against a local Redis, confirming the dedup lookup is not the drain bottleneck.

Failure modes and edge cases #

• Idempotency key set before the side effect is durable. The classic bug: marking a key delivered, then crashing before the downstream write commits, drops the trigger permanently on replay. Always set the key after the acknowledgement, accepting that a crash in the window replays once (at-least-once), which the dedup then collapses on the next attempt.
• Dedup TTL shorter than DLQ retention. If a trigger sits in the DLQ for 5 days but the dedup key expires after 1 day, a replay past day 1 sees a clean key and re-fires. Size the TTL at or above the DLQ retention window.
• Still-poison record looping. A record that fails replay for a structural reason (schema drift never fixed) must re-quarantine with an incremented replay_attempt, not re-enter the same drain — the RE_CEILING cap parks it after a bounded number of replay attempts. Without the cap, a genuinely poison trigger cycles the DLQ forever.
• Replay re-saturating the recovered downstream. Draining a large backlog at full rate can knock over the dependency that just recovered. The token bucket at $\rho \le 0.3$ is the guard; if the downstream 5xx rate rises during drain, lower $\rho$ dynamically rather than pushing harder.
• Reordering across replay and live streams. A replayed ENTER can arrive after a live EXIT for the same device. The downstream must resolve transitions against event time, not arrival order — see deduplicating ENTER/EXIT events with idempotency keys for the event-time bucketing that makes this safe.
• Empty or malformed envelope. A DLQ record whose payload_b64 is truncated cannot be replayed; route it to a dlq-parked topic for manual inspection rather than crashing the drain worker on a json.loads error.

Related #

• Dead-Letter Topics & Poison-Message Handling for Trigger Streams — parent overview of quarantine, retry ceilings, and DLQ topic design.
• Idempotent Trigger Emission Semantics for Geofencing — the deterministic keys the replay dedup relies on.
• Deduplicating ENTER/EXIT Events with Idempotency Keys — event-time bucketing that keeps reordered replays safe.
• Token-Bucket vs Leaky-Bucket for Telemetry Shedding — the rate-limiting primitive that paces the drain.