Explicación del ticket T7 — Cola de trabajos en Postgres¶

Qué es esto, en una frase¶

Es el motor que permite ejecutar cosas fuera del ciclo de un request HTTP (enviar correos, procesar eventos de Stripe, correr reconciliaciones), usando una tabla más de la misma base de datos como cola —en vez de Redis y una librería de colas aparte— con SELECT ... FOR UPDATE SKIP LOCKED como mecanismo de reparto de trabajo entre workers.

De paso, este trabajo también cerró un pendiente que había quedado abierto de la ronda de revisión anterior de T4: mover la generación del id de todas las tablas de Postgres hacia Python, para cumplir al pie de la letra lo que dice el Anexo de T0-contratos.md. Se explica más abajo, en su propia sección, porque toca archivos de T4 y no solo los nuevos de T7.

Cómo se relacionan las piezas¶

app/db/models/jobs.py        -> el modelo SQLAlchemy de la tabla `jobs`
alembic/versions/..._create_jobs.py -> la migración que crea esa tabla
app/services/jobs.py         -> enqueue(), la única puerta de entrada a la cola
app/workers/handlers.py      -> registro de handlers por job_type
app/workers/main.py          -> el bucle del worker: toma, procesa, reintenta
app/api/v1/jobs.py           -> GET /api/v1/jobs/metricas

El flujo real: algo en la aplicación (un endpoint, otro job) llama a enqueue() dentro de su propia transacción de base de datos y recibe un id de vuelta. Esa fila queda en la tabla jobs con estado PENDIENTE. El proceso worker (un contenedor aparte, ver docker-compose.yml, mismo código que la API pero corriendo python -m app.workers.main) sondea esa tabla cada 2 segundos, toma lotes de filas con FOR UPDATE SKIP LOCKED, busca en app/workers/handlers.py qué función sabe procesar ese job_type, la ejecuta, y según el resultado marca el trabajo como completado, lo reintenta más tarde, o lo manda a la cola de fallidos.

El fix retroactivo: UUIDv7 generado en Python, no en Postgres¶

Por qué se cambió¶

Al leer docs/T0-contratos.md completo para sacar la firma exacta de enqueue() (Contrato 4), se encontró una frase en el Anexo que contradice lo que se había implementado en la ronda de "consideraciones" anterior:

"UUID como llave primaria en todas las tablas, generado en la aplicación (...) si no [Postgres trae uuidv7() nativo], se genera desde Python con una librería."

Lo que se había construido en la ronda anterior era justo lo contrario: una función uuid_generate_v7() escrita en PL/pgSQL, corriendo del lado de Postgres vía server_default. La decisión fue seguir el contrato al pie de la letra: generación en Python.

Qué cambió, archivo por archivo¶

app/db/mixins.py: UUIDPKMixin.id pasó de server_default=text("uuid_generate_v7()") a default=uuid7 (de la librería uuid6). La diferencia entre default= y server_default= es justamente de dónde sale el valor: con default=, SQLAlchemy llama a uuid7() en Python antes de armar el INSERT; con server_default=, el valor lo decide Postgres al recibir la sentencia. Consecuencia práctica documentada en el propio archivo: una fila insertada con SQL crudo por fuera de la aplicación ya no recibe un id automático — hay que pasarlo a mano.
pyproject.toml: se agregó la dependencia uuid6 (Postgres 15, la versión que usa este proyecto, no trae uuidv7() nativo; eso llega recién en la 18).
Las 9 migraciones de T4 (fde70d722569, 896c18f5e21f, b91f43a59017, fc9f38e3e34b, f3042831ab77, 0fe28aab4bd2, f783ec8d0d11): se les quitó el server_default=sa.text('uuid_generate_v7()') de la columna id en las 10 tablas que la tenían. La columna quedó como sa.Column('id', sa.UUID(), nullable=False) a secas — ya no necesita ningún default de Postgres, porque el valor llega desde Python en cada INSERT.
839ec2b4026d_enable_pgcrypto_extension.py: esta migración originalmente habilitaba la extensión pgcrypto y definía la función uuid_generate_v7(). Como ninguna tabla depende ya de esa función, se convirtió en un no-op documentado (se explica en su propio docstring por qué existe y por qué ya no hace nada), en vez de borrarla; borrarla rompería la cadena de revisiones de Alembic en cualquier entorno donde ya se hubiera aplicado.

Qué se verificó de este fix¶

Se corrió el ciclo completo alembic upgrade head → downgrade base → upgrade head de nuevo contra un Postgres real, y además:

Se confirmó con un INSERT real (vía SQLAlchemy ORM, no SQL crudo) que el id que llega a la fila es un UUID versión 7 genuino (nibble de versión en 7, nibble de variante en el rango RFC 4122).
Se confirmó con una consulta directa a pg_proc y pg_extension que, tras el upgrade head, la función uuid_generate_v7() y la extensión pgcrypto no existen en la base — la generación es 100% del lado de Python, no queda ningún remanente server-side.

La tabla `jobs`¶

Columna	Para qué
`id`	PK, UUIDv7 generado en Python (igual que el resto de las tablas, tras el fix de arriba).
`job_type`	`"dominio.accion"` en minúscula (ej. `email.enviar`, `stripe.procesar_evento`). Es texto libre, sin `CHECK`: la lista de nombres válidos vive en el registro de handlers del worker, no en la base — así, agregar un `job_type` nuevo no exige una migración.
`payload`	JSONB. Guarda identificadores (`{"comision_id": "018f..."}`), nunca objetos serializados completos — cuando el worker lo procese, puede ser minutos después de un reintento, y debe leer el estado actual, no una foto vieja.
`estado`	`PENDIENTE` / `EN_PROCESO` / `COMPLETADO` / `AGOTADO`, con `CHECK`.
`intentos` / `max_attempts`	Cuántas veces se intentó y el tope antes de darse por vencido. Un `CHECK (intentos <= max_attempts)` a nivel de base es la última línea de defensa.
`run_at`	Desde cuándo puede tomarse el job. Por default es "ahora" (`server_default=now()`); un valor futuro es lo que habilita trabajos programados.
`locked_at` / `locked_by`	Se llenan cuando un worker toma el job (arrendamiento). Un job `EN_PROCESO` con `locked_at` de hace más de 5 minutos se considera huérfano.
`ultimo_error`	El mensaje del fallo más reciente, para depurar sin ir a buscar logs. Se sobreescribe en cada intento, no es un historial.
`idempotency_key`	Obligatoria solo para los `job_type` que mueven dinero (ver más abajo). Junto con `job_type` forma un `UNIQUE` — es la base de la deduplicación de `enqueue()`.
`created_at` / `updated_at`	Estándar del resto del esquema.

Dos índices compuestos acompañan la tabla: (estado, run_at) para que el sondeo del worker ("dame lo PENDIENTE cuyo run_at ya llegó") no sea un recorrido completo de la tabla a medida que crece, y (estado, locked_at) para que el barrido de huérfanos sea igual de barato.

`enqueue()` — Contrato 4 de T0¶

La firma implementada es la que fija el contrato, con un agregado: un primer parámetro db: Session.

def enqueue(
    db: Session,
    job_type: str,
    payload: dict,
    *,
    run_at: datetime | None = None,
    idempotency_key: str | None = None,
    max_attempts: int = 5,
) -> uuid.UUID:
    ...

Ese db no estaba en el snippet del ticket (probablemente omitido por brevedad), pero es imprescindible para cumplir la regla más importante del contrato: enqueue() tiene que correr dentro de la misma transacción que la escritura de dominio que lo origina. Por eso la función nunca hace su propio commit() — quien la llama decide cuándo confirmar, y si esa transacción se revierte, el job se revierte con ella.

Las cinco reglas del Contrato 4 y cómo quedaron resueltas:

Misma transacción que la escritura de dominio. Se logra recibiendo la sesión del caller en vez de abrir una propia.
Deduplicación con ON CONFLICT DO NOTHING, nunca con una excepción. Se implementó con pg_insert(Job).on_conflict_do_nothing(constraint=...) .returning(Job.id). Si no se devuelve fila (hubo conflicto), se hace un SELECT de respaldo para traer el id del job que ya existía. Nunca se deja que un IntegrityError suba, porque eso abortaría la transacción completa, incluida la escritura de dominio que envuelve al enqueue().
La llave de idempotencia protege el encolado, no la ejecución. Cada handler es responsable de ser idempotente por su cuenta (esto queda como requisito para T8 y el motor de comisiones, no algo que enqueue() pueda garantizar).
idempotency_key obligatoria en los trabajos de dinero. Se verifica en tiempo de ejecución: si job_type está en la lista (comision.calcular, comision.corregir, pago_aliado.registrar, stripe.procesar_evento) y no viene idempotency_key, se lanza ValueError — no es una convención en prosa, es una comprobación real.
El payload guarda identificadores, no objetos. Queda como convención documentada en el docstring del modelo y de enqueue(); no hay forma de forzarlo desde el código sin saber la forma de cada dominio, así que es responsabilidad de quien encola.

El bucle del worker¶

El diseño está partido en dos fases, y es la parte más importante de entender para saber por qué cumple el criterio de "matar el worker a mitad no pierde el trabajo":

Fase 1 — tomar un lote (_tomar_lote()). En una única transacción corta: SELECT id FROM jobs WHERE estado='PENDIENTE' AND run_at <= now() ORDER BY run_at LIMIT 10 FOR UPDATE SKIP LOCKED, seguido de un UPDATE que marca esas filas como EN_PROCESO con locked_at/locked_by, y commit(). SKIP LOCKED es lo que permite que varios workers corriendo en paralelo tomen lotes disjuntos sin bloquearse esperando el lock del otro — cada uno se queda con las filas que el otro no había tomado ya.

Fase 2 — procesar cada job (_procesar()). Cada job del lote se procesa en su propia transacción, separada de la que tomó el lote. Si el handler tiene éxito, el job pasa a COMPLETADO. Si lanza una excepción, se cuenta como fallo (ver abajo). Esta separación es deliberada: si el proceso muriera durante la fase 1, no habría ninguna transacción a medio terminar que retenga nada; y si muriera durante la fase 2 (procesando un job puntual), ese job específico queda EN_PROCESO sin que ninguna otra fila del lote se vea afectada.

Reclamo de huérfanos (_reclamar_huerfanos()). Corre al principio de cada ciclo del worker, antes de tomar un lote nuevo. Cualquier job EN_PROCESO con locked_at de más de 5 minutos se considera abandonado (el worker que lo tomó murió, se reinició, lo que sea) y vuelve a PENDIENTE, con locked_at/locked_by limpios. Es este mecanismo el que garantiza que ningún trabajo quede tomado "para siempre": si el worker que lo tenía no vuelve, cualquier otro worker vivo (o el mismo, si se reinicia) lo recupera pasados esos 5 minutos.

Reintentos con backoff exponencial. Cuando un handler falla, se incrementa intentos y, si todavía no se llegó a max_attempts, el job vuelve a PENDIENTE con un run_at futuro según esta tabla (la misma que fija el Contrato 4):

Intento que acaba de fallar	Espera hasta el próximo
1	10 segundos
2	40 segundos
3	3 minutos
4	12 minutos
5	50 minutos

Si un job en particular tiene max_attempts mayor a 5, se repite la última espera (50 minutos) en los intentos siguientes, en vez de fallar por quedarse sin valores en la tabla.

Cola de fallidos (dead letter). Cuando intentos llega a max_attempts, el job pasa a AGOTADO y deja de tomarse — el reproceso desde ahí es manual (la pantalla que menciona el Contrato 4 para reencolar con un run_id nuevo es de otro ticket, fuera de T7). Si el job_type agotado es uno de los que mueve dinero, se emite además un log.critical bien visible a modo de alerta — no hay todavía un canal de alertas real (Slack, PagerDuty, etc.), así que esto queda como un placeholder explícito, documentado en el código, para reemplazar cuando exista esa infraestructura.

Handlers. app/workers/handlers.py es un registro: cada job_type se asocia a una función (db, payload) -> None con el decorador @registrar("..."). Si se encola un job_type sin handler registrado, el worker lo trata como un error de configuración (falla, reintenta, y eventualmente se agota) en vez de marcarlo COMPLETADO en silencio — lo segundo sería peor, porque nadie notaría que el trabajo nunca hizo nada de verdad. Hoy solo existe un handler de humo (_smoke.echo), que no es negocio real: sirve para poder probar el motor completo sin depender de que T8 o las tareas de notificaciones ya existan. El propio ticket de T7 lo anticipa ("Bloquea: T8 (opcional), notificaciones") — el motor ya está completo; falta que cada dominio registre su handler ahí.

El endpoint de métricas¶

GET /api/v1/jobs/metricas devuelve:

{
  "pendientes": 3,
  "fallidos": 1,
  "antiguedad_mas_viejo_segundos": 42.7
}

pendientes: cuántos jobs están en PENDIENTE ahora mismo (incluye los programados a futuro que todavía no llega su run_at).
fallidos: cuántos están en AGOTADO — la cola de fallidos.
antiguedad_mas_viejo_segundos: hace cuánto se creó (por created_at) el job PENDIENTE más antiguo. Es el indicador clásico de si el worker está al día o se está quedando atrás; es null cuando no hay ningún PENDIENTE.

Queda un TODO explícito en el código: cuando exista T6 (roles y capacidades), este endpoint necesita requires("jobs:admin") — hoy no tiene autorización propia porque esa pieza todavía no existe.

Qué se verificó¶

Todo se probó contra un Postgres real, no solo leyendo el código:

Ciclo de migraciones. alembic upgrade head desde una base vacía (incluida la tabla jobs nueva) y alembic downgrade base de vuelta, sin errores ni residuos. alembic check confirma que el modelo Job y la migración coinciden exactamente (cero diferencias detectadas por autogenerate).
enqueue(), las 5 reglas, una por una: un encolado simple sin idempotency_key; un job de dinero sin llave (lanza ValueError); un job de dinero con llave (se encola); el mismo job_type + idempotency_key encolado dos veces (la segunda vez devuelve el mismo id, sin ninguna excepción); un run_at futuro explícito (se respeta, no lo pisa el default de la columna); y un enqueue() seguido de rollback() del caller (el job desaparece con el rollback, confirmando que vive en la misma transacción).
FOR UPDATE SKIP LOCKED bajo concurrencia real: se abrió una conexión que tomó y bloqueó una fila a propósito (sin confirmarla todavía) y se confirmó que una segunda conexión, al tomar su lote, la saltea y toma solamente las filas libres.
Reintentos y backoff: un handler que siempre falla, con max_attempts=3, mostró la progresión exacta 10s → 40s → AGOTADO en el tercer fallo, con ultimo_error guardado en cada paso.
Alerta de dinero agotado: un job_type de dinero con max_attempts=1 que falla, se agota inmediatamente y dispara el log.critical esperado.
Huérfanos: un job puesto a mano en EN_PROCESO con un locked_at de hace 10 minutos volvió a PENDIENTE (con locked_at/locked_by limpios) tras correr el barrido de huérfanos, y se pudo volver a tomar y procesar con normalidad.
El criterio de aceptación completo, con procesos reales (no solo funciones internas): se encoló un job con un handler que tarda unos segundos en correr; se arrancó el worker como proceso aparte (python -m app.workers.main); una vez que el worker lo tomó (EN_PROCESO), se lo mató con kill -9 a mitad del procesamiento; el job quedó EN_PROCESO, sin perderse ni marcarse de ninguna forma; se simuló el paso de los 5 minutos de arrendamiento; se arrancó un segundo worker (representando a ECS reemplazando la tarea muerta), que reclamó el huérfano en su primer ciclo y lo completó con éxito.
El endpoint de métricas: probado contra la aplicación FastAPI real (no mockeada), con jobs en distintos estados y edades, confirmando que los tres números devueltos son correctos.

¿Está terminado el ticket T7?¶

Sí, según su propio criterio de aceptación ("se encola un job de prueba y el worker lo procesa; matar el worker a mitad no pierde el trabajo") — verificado contra un Postgres real y con el worker corriendo como proceso de verdad, no solo simulado en el mismo proceso de prueba. Repasando el "qué se hace" del ticket, punto por punto:

Tabla jobs con tipo, payload, estado, intentos, run_at, locked_at, último error — sí.
Función enqueue() con la firma acordada en T0 — sí, con las 5 reglas del Contrato 4 probadas una por una.
Bucle del worker con SELECT ... FOR UPDATE SKIP LOCKED — sí, probado bajo concurrencia real (dos conexiones).
Reintentos con backoff exponencial y tope de intentos — sí, con la progresión exacta que fija el contrato (10s, 40s, 3min, 12min, 50min).
Cola de fallidos (dead letter) para lo que agota reintentos — sí, estado AGOTADO, con alerta (placeholder de log) para los job_types de dinero.
Soporte de trabajos programados (run_at futuro) — sí.
El servicio worker de ECS corre la misma imagen con otro comando — esto ya existía desde T1 (docker-compose.yml, Dockerfile); T7 le dio contenido real al bucle que antes era un placeholder.
Métrica/endpoint con pendientes, fallidos y antigüedad del más viejo — sí, GET /api/v1/jobs/metricas.

Lo que queda deliberadamente fuera del alcance de T7, y que el propio ticket marca como bloqueado por él (no al revés): los handlers reales de negocio. Hoy solo existe el handler de humo _smoke.echo. Cuando lleguen T8 (Stripe, comisiones) y las tareas de notificaciones (email, despachos), cada uno registra su propio handler en app/workers/handlers.py — el motor que los va a ejecutar ya está completo y probado.