Dead-Letter and Retry Handling for Embedding Pipelines

Every embedding pipeline that touches a remote model API will fail partway through a corpus, and the difference between a resilient system and a lossy one is entirely in how those failures are classified and parked. A naive try/except that retries forever turns a single poison chunk — one document that is too long, wrongly encoded, or produces a dimension the target column rejects — into an infinite loop that starves the whole worker pool. The opposite mistake, a bare except: pass, drops the chunk silently and surfaces weeks later as an unexplained recall regression. The correct design separates transient failures (HTTP 429, 5xx, socket timeouts) that deserve bounded retries with backoff from permanent failures (malformed input, dimension mismatch, a validation error) that must go straight to a dead-letter sink, and it preserves enough of the original payload that any parked item can be replayed idempotently without ever creating a duplicate vector. This page covers the failure taxonomy, two dead-letter sink architectures, a runnable retry-and-park loop, and the replay tooling that closes the loop within Embedding Ingestion Pipeline Engineering.

Up: Embedding Ingestion Pipeline Engineering

Architectural Divergence & Trade-offs #

The first design decision is where a failed chunk goes to rest. Two sinks dominate, and they differ most in whether the dead-letter write shares a transaction with your progress bookkeeping.

In-database DLQ table. The parked item is written to a Postgres table in the same transaction that advances the ingestion manifest — marking the source chunk as failed and inserting the DLQ row commit or roll back together. This gives you exactly-once-ish semantics against the database: there is no window where the manifest says “failed” but the DLQ row is missing, or vice versa. Replay is a plain SELECT ... WHERE next_retry_at <= now(), so the tooling is SQL you already know, and the parked payload sits next to your vectors under the same backup and point-in-time-recovery regime. The cost is that the DLQ shares write capacity and vacuum pressure with the hot ingestion path, and cross-service consumers cannot subscribe to it without polling.

External broker DLQ. The failed item is published to a Redis Stream, an SQS dead-letter queue, or a dedicated Kafka topic. Brokers give you mature replay and inspection tooling, fan-out to multiple consumers, and physical isolation from the primary database so a DLQ backlog never competes with query traffic. The price is a distributed-transaction problem: the embedding write lands in Postgres while the DLQ publish lands in the broker, and without an outbox pattern a crash between the two can drop the failure record or double-publish it. You also take on a second system to operate, monitor, and secure.

For a single-database embedding pipeline the in-database table is the pragmatic default: it collapses the two-phase-commit problem into one transaction and needs no new infrastructure. Reach for a broker only when failures must fan out to independent consumers or when the DLQ volume would meaningfully load the primary database. The rest of this page implements the in-database design, because its transactional guarantee is what makes idempotent replay provable. The idempotency keys it relies on — a stable (doc_id, chunk_index) identity — come from Metadata Mapping & Schema Design, and the backoff timing on the retry side is developed in implementing exponential backoff for embedding API calls.

Parameter Space & Diagnostic Workflow #

Retry and dead-letter behavior is governed by a small set of constants whose interaction determines whether the pipeline self-heals or thrashes. The central tension is between max_retries (how long you fight a transient error before parking) and max_delay (how long a single retry may sleep). Too many retries with a long cap means a genuinely dead endpoint holds a worker for minutes; too few means a brief provider blip parks recoverable work that then needs a manual replay.

The diagnostic workflow starts by reading the DLQ itself, because its shape tells you what is failing and whether the retry logic is working. Group by error_class to separate a provider outage (thousands of transient rows appearing at once) from a data problem (a steady trickle of permanent rows):

SELECT error_class,
       count(*)                                   AS parked,
       min(first_failed_at)                       AS oldest,
       max(last_failed_at)                        AS newest,
       round(avg(attempts), 1)                    AS avg_attempts
FROM embedding_dlq
GROUP BY error_class
ORDER BY parked DESC;

A band of rows with avg_attempts equal to max_retries is exhausted transient work — the endpoint stayed down longer than your retry budget, and these are safe to replay once it recovers. Rows with avg_attempts = 1 are permanent failures that never should have retried; if their error_class looks transient, your classifier is mislabeling and you are wasting retries. Rows whose first_failed_at is hours old and still present mean the replay job is not draining — check that it is scheduled and that next_retry_at is actually in the past for the backlog:

SELECT count(*) FILTER (WHERE next_retry_at <= now())  AS ready_to_replay,
       count(*) FILTER (WHERE next_retry_at > now())    AS still_backing_off,
       min(next_retry_at)                               AS next_due
FROM embedding_dlq;

Step-by-Step Implementation #

The following builds the in-database dead-letter path end to end: a classifier, the DLQ schema, a capped-backoff retry loop that parks on exhaustion, and an idempotent replay job. It assumes the async producer–consumer structure from async processing with Python asyncio is calling into it from each consumer’s exception handler.

Step 1 — Classify the failure. Classification is the whole game: it decides whether an error consumes a retry attempt or parks immediately. Key off the HTTP status and exception type, never off a substring of the message.

import httpx

TRANSIENT_STATUS = {408, 425, 429, 500, 502, 503, 504}

def classify(exc) -> str:
    """Return 'transient' (retry) or a permanent error_class (park now)."""
    if isinstance(exc, (httpx.TimeoutException, httpx.ConnectError)):
        return "transient"
    if isinstance(exc, httpx.HTTPStatusError):
        code = exc.response.status_code
        return "transient" if code in TRANSIENT_STATUS else f"http_{code}"
    if isinstance(exc, ValueError):          # e.g. wrong dimension, bad encoding
        return "dimension_mismatch"
    return "permanent_unknown"

Step 2 — Create the DLQ table. The payload column is jsonb and stores the entire original chunk plus its metadata, so a parked item carries everything needed to re-embed it — nothing is lost when the source stream has moved on. The (doc_id, chunk_index) unique constraint means a chunk that fails twice updates its DLQ row instead of duplicating it.

CREATE TABLE embedding_dlq (
    id              bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    doc_id          text        NOT NULL,
    chunk_index     int         NOT NULL,
    payload         jsonb       NOT NULL,          -- original chunk + metadata
    error_class     text        NOT NULL,
    last_error      text,
    attempts        int         NOT NULL DEFAULT 0,
    first_failed_at timestamptz NOT NULL DEFAULT now(),
    last_failed_at  timestamptz NOT NULL DEFAULT now(),
    next_retry_at   timestamptz NOT NULL DEFAULT now(),
    UNIQUE (doc_id, chunk_index)
);

CREATE INDEX embedding_dlq_ready_idx
    ON embedding_dlq (next_retry_at)
    WHERE error_class = 'transient';

Step 3 — Retry with capped exponential backoff. In-process, transient errors are retried a bounded number of times with full jitter before the item is handed to the park routine. The 429 path honors an explicit Retry-After header when present rather than guessing.

import asyncio, random

BASE_DELAY, BACKOFF_FACTOR, MAX_DELAY, MAX_RETRIES = 1.0, 2.0, 45.0, 6

async def embed_with_retry(session, item):
    for attempt in range(MAX_RETRIES):
        try:
            return await embed(session, item["text"])
        except Exception as exc:
            klass = classify(exc)
            if klass != "transient":
                await park(item, klass, str(exc), attempts=attempt + 1)
                return None                       # permanent: stop immediately
            delay = min(BASE_DELAY * BACKOFF_FACTOR ** attempt, MAX_DELAY)
            await asyncio.sleep(random.uniform(0, delay))   # full jitter
    await park(item, "transient", "max_retries exhausted", attempts=MAX_RETRIES)
    return None

Step 4 — Park transactionally. The park routine writes the DLQ row and advances the manifest in one transaction, so bookkeeping and the dead-letter record can never disagree. ON CONFLICT bumps attempts and pushes next_retry_at forward with the same backoff curve, keeping a repeatedly failing chunk from being replayed in a tight loop.

async def park(item, error_class, last_error, attempts):
    delay = min(BASE_DELAY * BACKOFF_FACTOR ** attempts, MAX_DELAY)
    async with pool.acquire() as conn:
        async with conn.transaction():            # atomic with bookkeeping
            await conn.execute(
                """
                INSERT INTO embedding_dlq
                    (doc_id, chunk_index, payload, error_class,
                     last_error, attempts, next_retry_at)
                VALUES ($1, $2, $3::jsonb, $4, $5, $6, now() + $7 * interval '1 second')
                ON CONFLICT (doc_id, chunk_index) DO UPDATE SET
                    error_class    = EXCLUDED.error_class,
                    last_error     = EXCLUDED.last_error,
                    attempts       = embedding_dlq.attempts + 1,
                    last_failed_at = now(),
                    next_retry_at  = EXCLUDED.next_retry_at
                """,
                item["doc_id"], item["chunk_index"], json.dumps(item),
                error_class, last_error, attempts, delay,
            )
            await conn.execute(
                "UPDATE ingest_manifest SET status='failed', updated_at=now() "
                "WHERE doc_id=$1 AND chunk_index=$2",
                item["doc_id"], item["chunk_index"])

Step 5 — Replay idempotently. The replay job selects rows whose next_retry_at has passed, re-embeds them, and upserts on (doc_id, chunk_index). Because the target vector write is itself an ON CONFLICT DO UPDATE, a replayed chunk overwrites the same logical row instead of appending a duplicate — replaying twice is a no-op. On success the DLQ row is deleted in the same transaction as the vector write.

async def replay(pool, session, batch_size=200):
    async with pool.acquire() as conn:
        rows = await conn.fetch(
            """SELECT id, payload FROM embedding_dlq
               WHERE next_retry_at <= now() AND error_class = 'transient'
               ORDER BY next_retry_at
               LIMIT $1
               FOR UPDATE SKIP LOCKED""", batch_size)
        for row in rows:
            item = json.loads(row["payload"])
            vec = await embed_with_retry(session, item)
            if vec is None:
                continue                          # re-parked by embed_with_retry
            vec_literal = "[" + ",".join(map(repr, vec)) + "]"
            async with conn.transaction():
                await conn.execute(
                    """INSERT INTO doc_embeddings
                         (doc_id, chunk_index, content_hash, embedding)
                       VALUES ($1,$2,$3,$4)
                       ON CONFLICT (doc_id, chunk_index) DO UPDATE
                         SET embedding = EXCLUDED.embedding,
                             content_hash = EXCLUDED.content_hash,
                             updated_at = now()""",
                    item["doc_id"], item["chunk_index"],
                    item["content_hash"], vec_literal)
                await conn.execute("DELETE FROM embedding_dlq WHERE id=$1", row["id"])

The FOR UPDATE SKIP LOCKED clause lets several replay workers drain the DLQ in parallel without processing the same row twice. For a multi-node evolution of this whole pattern with broker-backed retries, see building a resilient Python embedding pipeline with Celery.

Validation & Recall Testing #

A dead-letter path is worthless if items leak out of it or if replay corrupts the vector table. Validate on three axes: nothing is dropped, replay converges, and replay is idempotent.

Nothing is dropped. Every source chunk must be accounted for as either a stored vector or a DLQ row. A chunk that is in neither is a silent drop — the single most dangerous failure mode. This reconciliation query returns the leaks:

SELECT m.doc_id, m.chunk_index
FROM ingest_manifest m
LEFT JOIN doc_embeddings e USING (doc_id, chunk_index)
LEFT JOIN embedding_dlq   d USING (doc_id, chunk_index)
WHERE e.doc_id IS NULL AND d.doc_id IS NULL;

Replay converges. Track the replay success rate — the fraction of drained rows that land as vectors versus re-park — over successive passes. A healthy DLQ trends toward zero after the upstream fault clears; one that plateaus is full of permanent errors masquerading as transient, or the classifier is wrong.

-- run before and after a replay pass; parked should fall, stored should rise
SELECT (SELECT count(*) FROM embedding_dlq)                       AS parked,
       (SELECT count(*) FROM doc_embeddings)                      AS stored;

Replay is idempotent. The proof is that replaying the same DLQ batch twice produces the same row count and does not move updated_at on the second pass beyond genuinely changed content. Run a replay, snapshot max(updated_at), replay the identical set again, and assert the count is unchanged and no new duplicate (doc_id, chunk_index) pairs exist:

SELECT doc_id, chunk_index, count(*)
FROM doc_embeddings
GROUP BY doc_id, chunk_index
HAVING count(*) > 1;      -- must return zero rows

Any rows here mean the target lacks its unique constraint and replay is appending duplicate vectors — a defect that silently inflates recall denominators and index size. If duplicates exist, retrieval quality drifts even though every individual vector is correct, so confirm this query is empty before trusting any recall benchmark run against the ingested set.

Failure Modes & Gotchas #

• Silent drops with no DLQ. A try/except that logs and continues but never persists the failed chunk loses it the instant the source stream advances. Without the reconciliation query above you will not notice until a recall regression. Every caught exception must either retry or park — never just log.
• Poison message retried forever. A chunk that always fails (too long for the model, bad encoding) will consume a worker indefinitely if there is no attempt cap. max_retries plus classification is what breaks the loop: permanent classes park at attempt one, transient classes park after the cap.
• Non-idempotent replay creating duplicates. Replaying with a plain INSERT instead of INSERT ... ON CONFLICT appends a second vector for the same chunk every pass. The unique key on (doc_id, chunk_index) and the upsert are both mandatory; verify with the duplicate query above.
• Losing the original payload. Storing only an error message and a doc_id, not the chunk text and metadata, makes an item unreplayable once the source is gone. Persist the full payload as jsonb so the DLQ is self-sufficient.
• Classifier keyed off message strings. Matching on "rate limit" in str(exc) breaks the moment a provider rewords its error. Key off status codes and exception types, which are part of the API contract.
• Backoff without jitter. A provider outage fails thousands of chunks at the same instant; retrying them all after exactly base_delay × factor^n reproduces the thundering herd that caused the outage. Full jitter spreads the retries out.
• DLQ writes competing with the hot path. A large transient backlog writing to the in-database DLQ shares vacuum and WAL budget with live ingestion. Keep the DLQ narrow, index only what replay queries need, and vacuum it on its own schedule.

Monitoring & Alerting Hooks #

The dead-letter queue is a leading indicator: its depth and growth rate predict a data-quality or provider incident before it shows up as a recall drop. Export DLQ depth as a gauge and alert on its first derivative, not just its absolute value — a slowly rising floor of permanent errors is normal, a sudden vertical is an incident.

from prometheus_client import Gauge

DLQ_DEPTH = Gauge("embed_dlq_depth", "Rows parked in the DLQ", ["error_class"])
DLQ_READY = Gauge("embed_dlq_ready", "DLQ rows past next_retry_at")

async def sample_dlq_metrics(pool):
    async with pool.acquire() as conn:
        for r in await conn.fetch(
                "SELECT error_class, count(*) c FROM embedding_dlq "
                "GROUP BY error_class"):
            DLQ_DEPTH.labels(r["error_class"]).set(r["c"])
        ready = await conn.fetchval(
            "SELECT count(*) FROM embedding_dlq WHERE next_retry_at <= now()")
        DLQ_READY.set(ready)

On the database side, alert on DLQ growth and on a stalled replay. A rising ready_to_replay count that never falls means the replay job is dead or wedged:

-- growth over the last 5 minutes; page if this is a sharp positive spike
SELECT count(*) FILTER (WHERE first_failed_at > now() - interval '5 minutes')
         AS new_parked_5m,
       count(*) FILTER (WHERE next_retry_at <= now())
         AS ready_to_replay,
       max(attempts) AS worst_attempts
FROM embedding_dlq;

Wire two alerts: new_parked_5m crossing a threshold (a provider or data incident in progress) and ready_to_replay staying above zero for longer than the replay interval (the drain has stopped). Pair the first with the backoff dashboards from implementing exponential backoff for embedding API calls to distinguish a rate-limit storm, which the retry loop should absorb on its own, from a genuine outage that needs a human.

FAQ #

Related #

• Implementing exponential backoff for embedding API calls — the retry timing that feeds this dead-letter path
• Async Processing with Python asyncio — the consumer loop whose exception handler calls into the park routine
• Building a resilient Python embedding pipeline with Celery — a multi-node evolution with broker-backed retries
• Metadata Mapping & Schema Design — the stable conflict keys that make idempotent replay possible
• Up: Embedding Ingestion Pipeline Engineering