Implementing a Dead-Letter Queue for Failed Embeddings

This page shows how to build a durable dead-letter queue (DLQ) for an embedding ingestion pipeline so that a chunk whose embedding call fails permanently is captured with its full payload instead of silently vanishing. It scopes the problem narrowly: how to model the DLQ table, retry transient failures with capped exponential backoff before diverting to the DLQ on exhaustion, and replay dead-lettered rows idempotently so re-processing never duplicates a stored vector.

Up: Dead-Letter & Retry Handling

An embedding pipeline fails in two distinct regimes, and conflating them corrupts your data. Transient failures — a 429 rate-limit, a 503, a socket timeout — should be retried in place; permanent failures — a 400 on a malformed chunk, a context-length overflow, a persistently poisoned payload — should be parked, not retried forever. A DLQ is the boundary between those regimes: it holds the complete original unit of work (the chunk text plus every piece of metadata needed to reconstruct the job) so a fix-and-replay is a database operation rather than an archaeology project. The retry loop that feeds it reuses the same capped-backoff discipline described in implementing exponential backoff for embedding API calls, and the replay job leans on the idempotent vector upserts with ON CONFLICT contract so a row can be replayed any number of times without duplicating a vector.

Prerequisites #

• PostgreSQL 15+ with the pgvector extension 0.5+ installed (CREATE EXTENSION vector;).
• A target embeddings table keyed on (doc_id, chunk_index) with a UNIQUE or PRIMARY KEY constraint on that pair — the upsert conflict target the replay job depends on.
• psycopg 3.1+ (used here) or asyncpg 0.28+, plus an embedding client that raises typed exceptions you can classify as transient vs permanent.
• A retry policy decided in advance: max attempts, base delay, cap, and jitter — see the dead-letter and retry handling overview for how these interact with your queue depth.
• The payload contract fixed: the DLQ must store enough to fully reconstruct the job (doc_id, chunk_index, model, raw text, and any tenant/metadata keys) as a single jsonb document.

Step-by-step procedure #

1. Create the dead-letter table #

Store the entire original job as a jsonb payload so a replay never needs to reconstruct context from elsewhere. Track the error taxonomy (error_class), the human-readable last_error, the attempts already spent, and both a first_failed_at and last_failed_at so you can measure how long a row has been stuck.

CREATE TABLE IF NOT EXISTS embedding_dlq (
    id              bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    payload         jsonb       NOT NULL,   -- full chunk text + metadata to rebuild the job
    error_class     text        NOT NULL,   -- e.g. 'RateLimit', 'ContextLength', 'ServerError'
    last_error      text        NOT NULL,   -- truncated exception message
    attempts        int         NOT NULL DEFAULT 0,
    first_failed_at timestamptz NOT NULL DEFAULT now(),
    last_failed_at  timestamptz NOT NULL DEFAULT now(),
    resolved_at     timestamptz             -- set when a replay finally succeeds
);

-- fast DLQ-depth and replay-candidate scans
CREATE INDEX IF NOT EXISTS embedding_dlq_open_idx
    ON embedding_dlq (error_class, first_failed_at)
    WHERE resolved_at IS NULL;

-- one open DLQ row per logical job; a second failure updates the same row
CREATE UNIQUE INDEX IF NOT EXISTS embedding_dlq_job_key
    ON embedding_dlq ((payload->>'doc_id'), ((payload->>'chunk_index')::int))
    WHERE resolved_at IS NULL;

The partial UNIQUE index on the extracted doc_id/chunk_index keeps at most one open DLQ entry per logical job, so a chunk that fails, replays, and fails again bumps attempts on the existing row instead of accumulating duplicates.

2. Retry transient failures, then divert to the DLQ #

Wrap the embedding call in a bounded retry loop. Classify each exception: transient classes sleep for min(cap, base * 2**attempt) plus full jitter and retry; permanent classes break immediately. When the loop is exhausted (or the error is permanent), write the whole job into embedding_dlq.

import json, random, time
import psycopg

TRANSIENT = {"RateLimit", "ServerError", "Timeout"}

def classify(exc: Exception) -> str:
    status = getattr(exc, "status_code", None)
    if status == 429:            return "RateLimit"
    if status in (500, 502, 503, 504): return "ServerError"
    if status == 400 and "context length" in str(exc).lower():
        return "ContextLength"
    if isinstance(exc, TimeoutError): return "Timeout"
    return "Permanent"

def process_job(conn, job: dict, embed, *, max_attempts=6, base=0.5, cap=30.0):
    last_exc, error_class = None, "Permanent"
    for attempt in range(max_attempts):
        try:
            vector = embed(job["text"])          # may raise
            upsert_vector(conn, job, vector)      # step 4 contract
            return "ok"
        except Exception as exc:                  # noqa: BLE001
            last_exc, error_class = exc, classify(exc)
            if error_class not in TRANSIENT:
                break                             # permanent: do not retry
            delay = min(cap, base * (2 ** attempt))
            time.sleep(random.uniform(0, delay))  # full jitter
    dead_letter(conn, job, error_class, last_exc, attempts=attempt + 1)
    return "dead-lettered"

def dead_letter(conn, job, error_class, exc, *, attempts):
    conn.execute(
        """
        INSERT INTO embedding_dlq (payload, error_class, last_error, attempts)
        VALUES (%s, %s, %s, %s)
        ON CONFLICT ((payload->>'doc_id'), ((payload->>'chunk_index')::int))
                    WHERE resolved_at IS NULL
        DO UPDATE SET attempts       = embedding_dlq.attempts + EXCLUDED.attempts,
                      last_error      = EXCLUDED.last_error,
                      error_class     = EXCLUDED.error_class,
                      last_failed_at  = now()
        """,
        (json.dumps(job), error_class, str(exc)[:500], attempts),
    )
    conn.commit()

3. Persist a successful embedding idempotently #

The success path and the replay path share one upsert so both are duplicate-free. Conflict on (doc_id, chunk_index) and overwrite, never append.

def upsert_vector(conn, job: dict, vector) -> None:
    conn.execute(
        """
        INSERT INTO doc_chunks (doc_id, chunk_index, model, embedding)
        VALUES (%(doc_id)s, %(chunk_index)s, %(model)s, %(embedding)s)
        ON CONFLICT (doc_id, chunk_index)
        DO UPDATE SET embedding = EXCLUDED.embedding,
                      model     = EXCLUDED.model
        """,
        {**job, "embedding": vector},
    )

4. Replay dead-lettered rows idempotently #

The replay job claims a batch of open rows with FOR UPDATE SKIP LOCKED (so parallel replayers never fight), re-runs the same process_job path, and on success stamps resolved_at inside the same transaction as the upsert. Because the upsert conflicts on the natural key, replaying a row that a prior run already succeeded on is a harmless no-op.

def replay_batch(conn, embed, *, limit=100) -> dict:
    rows = conn.execute(
        """
        SELECT id, payload FROM embedding_dlq
        WHERE resolved_at IS NULL
        ORDER BY first_failed_at
        LIMIT %s
        FOR UPDATE SKIP LOCKED
        """,
        (limit,),
    ).fetchall()

    stats = {"replayed": 0, "still_failing": 0}
    for dlq_id, payload in rows:
        try:
            vector = embed(payload["text"])
            upsert_vector(conn, payload, vector)              # idempotent
            conn.execute(
                "UPDATE embedding_dlq SET resolved_at = now() WHERE id = %s",
                (dlq_id,),
            )
            stats["replayed"] += 1
        except Exception as exc:                              # noqa: BLE001
            conn.execute(
                """UPDATE embedding_dlq
                   SET attempts = attempts + 1, last_error = %s,
                       error_class = %s, last_failed_at = now()
                   WHERE id = %s""",
                (str(exc)[:500], classify(exc), dlq_id),
            )
            stats["still_failing"] += 1
        conn.commit()                                         # per-row durability
    return stats

Parameter reference #

Verification #

Watch DLQ depth and age by cause, and confirm a replay actually drained rows and produced vectors.

-- open DLQ depth and oldest stuck row, per failure class
SELECT error_class,
       count(*)                                    AS open_rows,
       max(attempts)                               AS worst_attempts,
       now() - min(first_failed_at)                AS oldest_age
FROM embedding_dlq
WHERE resolved_at IS NULL
GROUP BY error_class
ORDER BY open_rows DESC;

-- replay health: rows resolved in the last hour, and any that resolved
-- but left no vector behind (a broken idempotency contract)
SELECT count(*) FILTER (WHERE d.resolved_at > now() - interval '1 hour') AS replayed_1h,
       count(*) FILTER (WHERE c.doc_id IS NULL)                          AS resolved_without_vector
FROM embedding_dlq d
LEFT JOIN doc_chunks c
       ON c.doc_id = d.payload->>'doc_id'
      AND c.chunk_index = (d.payload->>'chunk_index')::int
WHERE d.resolved_at IS NOT NULL;

A healthy pipeline shows resolved_without_vector = 0: every resolved DLQ row corresponds to a persisted vector.

Troubleshooting #

• A poison message is retried forever. A malformed chunk keeps failing 400 but was misclassified as transient, so every replay bumps attempts and never resolves. Diagnose with the depth query — look for rows where worst_attempts far exceeds max_attempts — and fix by adding the cause to the permanent set (or a max_replay_attempts guard that flips such rows to a quarantined state instead of re-queuing them).
• Replays duplicate vectors. The replay upsert is missing or targeting the wrong constraint, so re-processing appends rows. Confirm the embeddings table has a UNIQUE/PRIMARY KEY on (doc_id, chunk_index) and that both the success path and replay path go through the same ON CONFLICT upsert from idempotent vector upserts with ON CONFLICT.
• DLQ rows exist but the payload cannot rebuild the job. The insert stored an opaque ID instead of the full chunk, and the source is gone by replay time. Fix the contract to serialize text, model, and metadata into payload at divert time; backfill is impossible once the upstream buffer has rolled over.
• Two replay workers process the same row. The SELECT is missing FOR UPDATE SKIP LOCKED, so concurrent replayers both grab a row and one write is wasted. Add the row lock; the upsert makes the duplicate work harmless but the lock avoids it entirely.
• DLQ depth climbs but the retry loop looks fine. The backoff cap is too low for a sustained 429 storm, so jobs exhaust max_attempts faster than the rate limit recovers. Raise cap and add full jitter as covered in implementing exponential backoff for embedding API calls, and consider a circuit breaker that pauses the feed rather than filling the DLQ.

Related #

• Dead-Letter & Retry Handling — the retry-and-recovery stage this procedure belongs to
• Implementing exponential backoff for embedding API calls — the capped-backoff loop that feeds the DLQ
• Idempotent vector upserts with ON CONFLICT — the upsert contract that makes replay duplicate-free
• Building a resilient Python embedding pipeline with Celery — wiring retries and a DLQ into a task queue
• Up: Dead-Letter & Retry Handling