Fila de tarefas

Ponto único para pedir downloads, raspagens e execuções de crawlers. Quem pede recebe um id na hora (HTTP 202) e acompanha o job até o fim; a execução acontece num subprocesso isolado, com timeout, cancelamento, novas tentativas e artefatos guardados por job.

cliente ──POST /jobs──▶ API ──▶ PostgreSQL, schema jobs
                                   ▲        │ proxima()  (advisory lock + SKIP LOCKED)
                                   │        ▼
            finalizar()/heartbeat  │   executor (core/jobs/runner.py, 1 por máquina,
                                   │        │     N máquinas por fila)
                                   │        │ python -m core.jobs.worker <id>
                                   └────────┴──▶ tarefa registrada (core/jobs/tarefas/)
                                                    └─▶ JOBS_DIR/execucoes/<id>/ (disco do executor)

Banco de dados

Tudo fica no schema jobs do PostgreSQL, isolado do resto do banco:

As regras ficam no banco, não só no código: CHECK de status, nome de tarefa, faixa de prioridade e tentativas, caminho de artefato sem .., e o índice único parcial que garante uma chave ativa por vez.

Conexão: JOBS_DATABASE_URL (postgresql://user:senha@host:5432/banco) ou, na falta dela, o banco principal (DATABASE_URL), o mesmo do resto do repositório (ver core/database/config.py). Sem nenhuma das duas a API sobe, mas o executor fica desligado (aviso no log).

Migrations

Alembic com ambiente próprio em core/jobs/migrations/. A tabela de versão mora em jobs.alembic_version, então não interfere em nada fora do schema.

uv run python -m core.jobs migrar                       # upgrade head
uv run alembic -c core/jobs/alembic.ini current
uv run alembic -c core/jobs/alembic.ini upgrade head --sql   # só gera o SQL, para revisão
uv run alembic -c core/jobs/alembic.ini downgrade -1
uv run alembic -c core/jobs/alembic.ini revision -m "descricao"   # nova migration

Migrations em SQL explícito (op.execute), sem autogenerate: não há modelos SQLAlchemy da fila, e o SQL é o que se revisa.

Subir

uv run python -m core.jobs migrar

# API + executor no mesmo processo (default)
uv run uvicorn api.main:app --port 8000

# ou o executor à parte (ex.: no poly-lab, onde ficam o token A3 e o X11)
JOBS_RUNNER_NA_API=0 uv run uvicorn api.main:app --port 8000
JOBS_TAREFAS='inss.*,govbr.*' uv run python -m core.jobs
uv run python -m core.jobs --tarefas    # catálogo em JSON

Vários executores

Qualquer número de máquinas pode consumir a mesma fila. Cada uma escolhe o que aceita com JOBS_TAREFAS (padrões com *; vazio = tudo). Na mesma máquina só um processo executa (flock em JOBS_DIR/runner.lock).

Cada job rodando tem executor (host:pid) e heartbeat_em, renovado a cada 5 s. Se um executor morrer, o lease vence em JOBS_LEASE_S e qualquer outro executor dá o job como interrompido (e ele volta à fila se houver tentativas). Um executor que reinicia recolhe na hora o que o processo antigo do mesmo host deixou rodando.

Artefatos no MinIO

Com o MinIO configurado (MINIO_ENDPOINT, MINIO_ACCESS_KEY, MINIO_SECRET_KEY), o executor envia cada artefato ao fim do job para s3://<JOBS_BUCKET>/<JOBS_BUCKET_PREFIXO>/<job>/<caminho> (default s3://crawlers-jobs/jobs/<job>/...; o bucket é criado se não existir) e grava o uri em jobs.artefatos. A API serve o arquivo do disco local quando ele está lá e, senão, faz streaming do MinIO — de qualquer máquina.

Sem MinIO (ou com JOBS_BUCKET= vazio), os artefatos ficam só no disco de quem executou; nas outras máquinas o download responde 409. Falha no envio não derruba o job: o artefato fica sem uri e o erro vai para o log.

API

Todas as rotas exigem API key (Authorization: Bearer crw_…) ou a sessão do painel; cada usuário vê só os próprios jobs. Ver docs/contas-e-cofre.md. Uma tarefa que usa credencial do cofre recebe credencial_id no pedido.

curl -X POST localhost:8000/jobs -H "Authorization: $TOKEN" -H 'content-type: application/json' -d '{
  "tarefa": "download.http",
  "parametros": {"url": "https://exemplo.jus.br/relatorio.pdf"},
  "prioridade": 5,
  "chave_idempotencia": "relatorio-2026-10",
  "origem": "n8n"
}'

• prioridade (-10..10): maior roda antes; empate por ordem de chegada.
• chave_idempotencia: se já houver job ativo com a chave, ele é devolvido ("duplicado": true) em vez de criar outro.
• max_tentativas: sobrescreve o default da tarefa.

Estados

na_fila → rodando → um destes:

O traceback fica em erro.txt e a saída do processo em execucao.log, ambos só no servidor — a API não os expõe, porque podem conter URLs com sessão.

Tarefas disponíveis

Recurso é um nome de exclusão mútua: jobs com o mesmo recurso nunca rodam juntos, mesmo com JOBS_CONCORRENCIA > 1. Tarefas que gastam tentativa de PIN têm max_tentativas=1: repetir sozinho pode bloquear o token.

Adicionar uma tarefa

# core/jobs/tarefas/tjxx.py
from pydantic import BaseModel, ConfigDict
from core.jobs.registry import tarefa

class Params(BaseModel):
    model_config = ConfigDict(extra="forbid")   # parâmetro desconhecido = 422
    processo: str

@tarefa("tjxx.extrair_processo", parametros=Params, timeout_s=600, max_tentativas=2)
async def extrair(ctx, params: Params) -> dict:
    """Primeira linha da docstring vira a descrição no catálogo."""
    ctx.etapa("Abrindo processo")             # aparece em GET /jobs/{id}
    (ctx.pasta / "processo.json").write_text("...")  # vira artefato
    if ctx.cancelado():
        return {"status": "falha", "detalhe": "cancelado"}
    return {"status": "sucesso", "arquivo": "processo.json"}

Depois, importe o módulo em core/jobs/tarefas/__init__.py. Imports pesados (Playwright, crawlers) ficam dentro da função, para a API listar o catálogo sem carregá-los. Segredos nunca são parâmetros: a tarefa lê do ambiente.

Variáveis

Relação com as filas existentes

processo.fila_downloads (usada pelo extractor) é outra coisa: fila de documentos amarrada a processo.processos, particionada por tribunal. Não foi alterada; um crawler de processo pode virar tarefa daqui e continuar gravando lá.

core/precatorios (painel /precatorios/monitor/ui) e core/health têm filas próprias em SQLite local. Elas seguem funcionando como estão; migrar as coletas LOA para tarefas precatorios.* desta fila é o próximo passo natural.