Idempotência e outbox em pipelines de IA: tirando o modelo do caminho da requisição

Chamar um modelo dentro do POST /tickets funciona até o dia em que não funciona.

O provedor fica lento e a latência do endpoint sobe junto. Um deploy do provedor derruba a avaliação e você perde a classificação de todos os tickets daquela janela — sem chance de recuperar, porque a informação nunca foi persistida como trabalho pendente. E quando alguém adiciona retry para melhorar a taxa de sucesso, a mesma avaliação passa a ser cobrada duas ou três vezes.

O padrão que resolve os três problemas de uma vez é antigo e conhecido: outbox. A novidade não é o padrão — é que, com LLM, a idempotência deixa de ser uma questão de elegância e vira controle de custo direto.

Em um pipeline CRUD, processar a mesma mensagem duas vezes gera uma linha duplicada que você corrige. Em um pipeline de IA, gera uma segunda fatura.

TL;DR

• Persista o dado principal e um registro de trabalho pendente na mesma transação.
• A chave de idempotência é derivada do conteúdo, não do timestamp nem de um UUID aleatório.
• Claim atômico com FOR UPDATE SKIP LOCKED evita que duas réplicas avaliem o mesmo item.
• Antes de chamar o modelo, verifique se já existe resultado para aquela chave. Isso é o que corta custo duplicado.
• Erro transitório volta para a fila com backoff; erro determinístico vai direto para dead letter.
• Guarde a versão do modelo, do prompt e da policy junto do resultado.
• Reprocessar precisa ser uma operação de rotina, não um script de emergência.

O problema, com números

Compare os dois desenhos no mesmo cenário: 200 mil tickets por mês, avaliação a US$ 0,011, provedor com 99,5% de disponibilidade e uma janela de 40 minutos de indisponibilidade no mês.

A linha que mais importa é a terceira. Retry sem idempotência não é só um risco de consistência — é um multiplicador de fatura.

Passo 1: uma transação, dois registros

A regra central do outbox: o dado principal e a intenção de trabalho nascem juntos ou não nascem.

CREATE TABLE ticket (
    id           uuid PRIMARY KEY,
    subject      text NOT NULL,
    message      text NOT NULL,
    created_at   timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE evaluation_outbox (
    id               bigserial PRIMARY KEY,
    ticket_id        uuid NOT NULL REFERENCES ticket(id),
    idempotency_key  text NOT NULL UNIQUE,
    status           text NOT NULL DEFAULT 'pending',
    attempts         int  NOT NULL DEFAULT 0,
    available_at     timestamptz NOT NULL DEFAULT now(),
    locked_by        text,
    locked_until     timestamptz,
    last_error       text,
    created_at       timestamptz NOT NULL DEFAULT now()
);

-- índice que sustenta o claim: só o que está pronto para rodar
CREATE INDEX idx_outbox_claimable
    ON evaluation_outbox (available_at)
    WHERE status = 'pending';

CREATE TABLE evaluation_result (
    idempotency_key  text PRIMARY KEY,
    ticket_id        uuid NOT NULL REFERENCES ticket(id),
    is_urgent        double precision,
    department       text,
    urgency_score    double precision,
    model_version    text NOT NULL,
    prompt_version   text NOT NULL,
    policy_version   text NOT NULL,
    computed_at      timestamptz NOT NULL DEFAULT now()
);

Repare que evaluation_result tem a chave de idempotência como chave primária. Essa é a estrutura que torna a checagem "já avaliei isso?" uma consulta de índice único, e não uma varredura.

A escrita:

def create_ticket(conn, subject: str, message: str) -> Ticket:
    ticket_id = uuid4()
    key = idempotency_key(subject, message)

    with conn.transaction():                       # atômico: os dois ou nenhum
        conn.execute(
            "INSERT INTO ticket (id, subject, message) VALUES (%s, %s, %s)",
            (ticket_id, subject, message),
        )
        conn.execute(
            """INSERT INTO evaluation_outbox (ticket_id, idempotency_key)
               VALUES (%s, %s)
               ON CONFLICT (idempotency_key) DO NOTHING""",
            (ticket_id, key),
        )
    return Ticket(id=ticket_id, subject=subject, message=message, evaluation=None)

O ON CONFLICT DO NOTHING já resolve o caso do usuário que clicou duas vezes: dois tickets, um único trabalho de avaliação.

Passo 2: a chave vem do conteúdo

Aqui está a decisão que mais gente erra.

Se a chave de idempotência for um UUID aleatório ou incluir timestamp, ela é única por definição — e portanto inútil. Ela precisa ser determinística sobre aquilo que de fato determina o resultado.

import hashlib
import json

PROMPT_VERSION = "triage-v3"
MODEL_VERSION = "gpt-6-luna@medium"

def idempotency_key(subject: str, message: str) -> str:
    """Determinística sobre tudo que muda o resultado. Nada além disso."""
    payload = {
        "subject": subject.strip(),
        "message": message.strip(),
        "prompt": PROMPT_VERSION,
        "model": MODEL_VERSION,
    }
    blob = json.dumps(payload, sort_keys=True, ensure_ascii=False)
    return hashlib.sha256(blob.encode()).hexdigest()[:32]

O que entra e o que fica de fora tem consequência direta:

Deixar ticket_id de fora é a escolha que produz o ganho mais visível: em bases com muita duplicata — formulários reenviados, integrações que reentregam —, uma fatia relevante do volume nunca chega ao modelo.

E incluir a versão do prompt no hash dá algo de graça: mudar o prompt invalida o cache automaticamente. A versão nova gera chaves novas, e o reprocessamento acontece sem nenhuma migração.

Passo 3: claim atômico, sem duas réplicas no mesmo item

Com múltiplos workers, o erro clássico é SELECT ... WHERE status='pending' seguido de UPDATE. Entre os dois comandos, outro worker leu a mesma linha.

FOR UPDATE SKIP LOCKED resolve isso em uma única ida ao banco:

CLAIM_SQL = """
UPDATE evaluation_outbox
SET status = 'processing',
    locked_by = %(worker)s,
    locked_until = now() + interval '5 minutes',
    attempts = attempts + 1
WHERE id IN (
    SELECT id FROM evaluation_outbox
    WHERE status = 'pending'
      AND available_at <= now()
    ORDER BY available_at
    FOR UPDATE SKIP LOCKED
    LIMIT %(batch)s
)
RETURNING id, ticket_id, idempotency_key, attempts;
"""

def claim(conn, worker_id: str, batch: int = 10) -> list[OutboxRow]:
    rows = conn.execute(CLAIM_SQL, {"worker": worker_id, "batch": batch}).fetchall()
    return [OutboxRow(*r) for r in rows]

SKIP LOCKED faz o worker pular linhas já travadas em vez de esperar por elas. Dez réplicas rodando o mesmo comando pegam dez lotes disjuntos, sem coordenação externa.

O locked_until cobre o worker que morre no meio do trabalho. Um varredor devolve o item à fila quando o lock expira:

UPDATE evaluation_outbox
SET status = 'pending', locked_by = NULL, locked_until = NULL
WHERE status = 'processing' AND locked_until < now();

Esse varredor é o que garante entrega pelo menos uma vez. Quem garante que "pelo menos uma vez" não vira "cobrado duas vezes" é o passo seguinte.

Passo 4: checar antes de chamar — a linha que economiza dinheiro

def process(conn, row: OutboxRow, client, deadline_s: float = 30.0) -> None:
    # 1. o resultado já existe para esta chave? então não chame o modelo.
    cached = conn.execute(
        "SELECT * FROM evaluation_result WHERE idempotency_key = %s",
        (row.idempotency_key,),
    ).fetchone()

    if cached is not None:
        metrics.increment("outbox.dedup_hit")
        finish(conn, row.id, status="done")
        return

    # 2. só agora custa dinheiro
    try:
        raw = client.evaluate(
            state=load_state(conn, row.ticket_id),
            questions=TRIAGE_QUESTIONS,
            timeout=deadline_s,
        )
    except TransientError as error:
        reschedule(conn, row, error)              # volta para a fila
        return
    except PermanentError as error:
        dead_letter(conn, row, error)             # não adianta repetir
        return

    evaluation = validate(raw)
    if evaluation is None:
        dead_letter(conn, row, "contract_violation")
        return

    # 3. persistir resultado e fechar o item na mesma transação
    with conn.transaction():
        conn.execute(
            """INSERT INTO evaluation_result
               (idempotency_key, ticket_id, is_urgent, department, urgency_score,
                model_version, prompt_version, policy_version)
               VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
               ON CONFLICT (idempotency_key) DO NOTHING""",
            (row.idempotency_key, row.ticket_id, evaluation.is_urgent,
             evaluation.department, evaluation.urgency_score,
             MODEL_VERSION, PROMPT_VERSION, POLICY_VERSION),
        )
        finish(conn, row.id, status="done")

O bloco 1 é a diferença entre um pipeline que reprocessa de graça e um que reprocessa pagando. Ele transforma "pelo menos uma vez" na entrega em "exatamente uma vez" no efeito cobrável, que é a propriedade que realmente interessa.

A métrica outbox.dedup_hit merece painel próprio. Ela mostra quanto dinheiro a idempotência está economizando, e costuma surpreender.

Passo 5: transitório volta, determinístico morre

Nem toda falha merece nova tentativa. Misturar as duas categorias é o que enche a fila de trabalho inútil.

TRANSIENT = (TimeoutError, RateLimited, ProviderUnavailable, ConnectionError)
MAX_ATTEMPTS = 5

def reschedule(conn, row: OutboxRow, error: Exception) -> None:
    if row.attempts >= MAX_ATTEMPTS:
        dead_letter(conn, row, f"max attempts: {error}")
        return

    # backoff exponencial com jitter determinístico por chave
    base = min(2 ** row.attempts, 300)
    jitter = int(row.idempotency_key[:4], 16) % 30       # espalha o rebanho
    delay = base + jitter

    conn.execute(
        """UPDATE evaluation_outbox
           SET status = 'pending',
               available_at = now() + make_interval(secs => %s),
               locked_by = NULL, locked_until = NULL, last_error = %s
           WHERE id = %s""",
        (delay, str(error)[:500], row.id),
    )

def dead_letter(conn, row: OutboxRow, reason: str) -> None:
    conn.execute(
        "UPDATE evaluation_outbox SET status = 'dead', last_error = %s WHERE id = %s",
        (reason[:500], row.id),
    )
    metrics.increment("outbox.dead_letter", tags={"reason": classify(reason)})

O jitter derivado da própria chave, em vez de random, mantém o comportamento reprodutível — útil quando você precisa explicar por que cem itens voltaram exatamente no mesmo segundo.

E a fila dead não é um cemitério. Ela é uma lista de bugs esperando triagem. Um alerta simples basta:

SELECT classify(last_error) AS motivo, count(*), max(created_at)
FROM evaluation_outbox
WHERE status = 'dead' AND created_at > now() - interval '1 day'
GROUP BY 1 ORDER BY 2 DESC;

Passo 6: reprocessar como rotina

Melhorou o prompt. Trocou o modelo. Ajustou os critérios. Agora você quer reavaliar os últimos trinta dias.

Com chave derivada do conteúdo e da versão, isso é uma inserção:

def reprocess(conn, since: datetime, new_prompt_version: str) -> int:
    """Reenfileira avaliações sob uma versão nova de prompt."""
    rows = conn.execute(
        "SELECT id, subject, message FROM ticket WHERE created_at >= %s", (since,)
    ).fetchall()

    inserted = 0
    for ticket_id, subject, message in rows:
        key = idempotency_key_with(subject, message, prompt=new_prompt_version)
        result = conn.execute(
            """INSERT INTO evaluation_outbox (ticket_id, idempotency_key, available_at)
               VALUES (%s, %s, now() + (random() * interval '30 minutes'))
               ON CONFLICT (idempotency_key) DO NOTHING
               RETURNING id""",
            (ticket_id, key),
        ).fetchone()
        inserted += 1 if result else 0
    return inserted

Dois detalhes deliberados: o available_at espalhado em trinta minutos evita que 200 mil itens batam no provedor ao mesmo tempo, e o ON CONFLICT garante que rodar o script duas vezes não duplica nada.

Como a versão antiga continua em evaluation_result, você pode comparar as duas antes de promover — que é exatamente o modo shadow aplicado a dados históricos.

O que observar

A idade do item mais antigo é a métrica mais honesta. Uma fila de 50 mil itens processando rápido é saudável; uma fila de 200 itens parada há seis horas não é.

Quando não usar outbox

Sejamos justos com a complexidade que isso adiciona. Outbox não vale a pena quando:

• o usuário precisa da avaliação na mesma resposta (autocomplete, sugestão em tempo real);
• o volume é baixo e a avaliação é barata o suficiente para o retry não importar;
• não existe banco transacional no caminho — sem a transação, o padrão perde a garantia principal.

Para tudo que é enriquecimento assíncrono de um dado que já foi persistido, ele é quase sempre a escolha certa.

Checklist

• Dado principal e registro de outbox nascem na mesma transação.
• A chave de idempotência deriva de conteúdo, prompt, modelo e esforço.
• ticket_id, timestamp e aleatoriedade estão fora da chave.
• O claim usa FOR UPDATE SKIP LOCKED.
• Existe varredor que devolve itens com lock expirado.
• O worker consulta o resultado antes de chamar o modelo.
• Erro transitório e determinístico seguem caminhos diferentes.
• A dead letter tem alerta e dono.
• O reprocessamento é um comando documentado, não um script improvisado.
• Resultado guarda versão de modelo, prompt e policy.

Conclusão

Outbox não é sobre elegância arquitetural. É sobre três propriedades concretas: o endpoint principal para de depender da latência do modelo, o trabalho perdido em um incidente vira trabalho pendente, e a mesma avaliação para de ser cobrada duas vezes.

A terceira é a que muda o cálculo em relação a pipelines tradicionais. Em CRUD, reprocessar é um incômodo. Com LLM, é uma segunda fatura — e a única defesa é uma chave derivada do conteúdo, consultada antes da chamada.

Se você guarda essa chave, o resto do padrão se monta sozinho: retry fica seguro, reprocessamento vira rotina, e melhorar o prompt deixa de ser uma decisão irreversível.

Leituras relacionadas

• Degradação graciosa em produtos com IA
• Custo por tarefa: a métrica que substitui preço por token
• JEV na prática: decisões tipadas em FastAPI

Os exemplos usam PostgreSQL 16 e psycopg 3. FOR UPDATE SKIP LOCKED existe desde o Postgres 9.5; em outros bancos, verifique o equivalente antes de copiar o padrão.