Trabajos en segundo plano

Descripción general

Un trabajo es una fila en rebase.jobs. Es reclamado exactamente por un worker, reintentado con un retraso creciente si su handler lanza un error, y se mantiene en la tabla cuando finalmente se rinde para que alguien pueda revisarlo.

No hay nada que instalar ni nada que ejecutar junto a Postgres. Un trabajo encolado dentro de una transacción que hace rollback nunca fue encolado.

Úselo para trabajo que no debe perderse y no debe ocurrir dentro de una solicitud: enviar correos, llamar a servicios de terceros, generar un archivo, conciliar con un sistema externo.

	Se ejecuta	Sobrevive a un reinicio
Cron	Según una programación	Sí — la programación está en el código
Trabajos	Una vez, tan pronto como un worker esté libre	Sí — el trabajo es una fila
Un setTimeout en un callback	Una vez, en este proceso	No

Habilitación

await initializeRebaseBackend({
    jobs: {
        enabled: true,
        tasks: {
            "send-welcome": async ({ payload }) => {
                await sendEmail((payload as { email: string }).email);
            }
        }
    }
});

Desactivado a menos que lo solicite: un worker sigue buscando trabajo mientras el proceso viva, lo cual no es un valor predeterminado que cualquiera elegiría. Necesita un controlador capaz de ejecutar SQL; en uno que no puede (MongoDB), la cola no está disponible y se le informa al arrancar en lugar de al realizar el primer encolado.

Desde 0.25 Un worker inactivo no sondea cada dos segundos. Cada encolado envía un NOTIFY en el canal rebase_jobs cuando se confirma, y un worker que mantiene un LISTEN sobre él reclama de inmediato: un trabajo encolado por otro proceso (el rol api, otra réplica) empieza en milisegundos, y uno diferido cuando vence. Aun así, revisa cada 30 segundos de todos modos. Un trabajo que encola este proceso, un reintento que programó y un trabajo que recuperó lo despiertan sin notificación.

El LISTEN necesita una conexión propia, que se busca como la busca el tiempo real: DATABASE_DIRECT_URL si está definida, si no DATABASE_URL. Detrás de un pooler en modo transacción como PgBouncer, define DATABASE_DIRECT_URL: un pooler no mantiene un LISTEN, y los trabajos de otros procesos esperarían entonces a la revisión de 30 segundos. Donde no hay ninguna conexión así (la base de datos de desarrollo en proceso), un worker inactivo espacia sus revisiones: mira tras pollIntervalMs (2 segundos), luego el doble de tiempo tras cada revisión que no encontró nada, hasta 30 segundos, y vuelve a empezar cada vez que encuentra trabajo. Un worker inactivo hace unos 2.900 reclamos al día en lugar de 43.200.

Encolado

const { jobQueue } = await initializeRebaseBackend({ jobs: { enabled: true, tasks } });

await jobQueue?.enqueue("send-welcome", { email: "ada@example.com" });

Opciones

await jobQueue?.enqueue("send-digest", { userId: "u7" }, {
    delayMs: 60_000,               // not before a minute from now
    maxAttempts: 5,                // default 3
    idempotencyKey: "digest:u7"    // at most one *unfinished* job with this key
});

idempotencyKey unifica un doble clic, una solicitud reintentada y dos instancias que reaccionan al mismo evento en un solo trabajo. Está delimitado al trabajo no finalizado, por lo que la clave vuelve a ser reutilizable una vez que el trabajo se completa; de lo contrario, “el resumen nocturno para el usuario 7” solo se podría enviar exactamente una vez en la historia. Un encolado duplicado se resuelve como null en lugar de lanzar un error: el trabajo solicitado ya está en cola, que es el resultado deseado.

Fallos

Un handler falla al lanzar una excepción (throw). No existe un return false: un booleano sería ignorado silenciosamente por cualquier handler que olvidara retornar uno, y el fallo debe ser lo que se obtenga por defecto.

• Intentos restantes → vuelve a pending, con run_at pospuesto por el retroceso (1s, 5s, 25s … con un límite de una hora; modifíquelo con backoff).
• Sin intentos restantes → failed, y la fila permanece. Una cola que descarta silenciosamente lo que no pudo entregar es indistinguible de una que no tiene nada que hacer.

SELECT task, attempts, last_error, updated_at
FROM rebase.jobs WHERE status = 'failed'
ORDER BY updated_at DESC;

Las filas fallidas se conservan 30 días; las exitosas, 3. Cada worker elimina las filas caducadas al arrancar y después cada hora, por lotes y fuera de la ruta de las solicitudes.

Qué sucede cuando un worker muere

Un proceso terminado a mitad de un trabajo no puede liberar su bloqueo, por lo que solo un tiempo de espera (timeout) liberará la fila. Mientras un handler se ejecuta, su worker renueva el bloqueo cada tercio de visibilityTimeoutMs (por defecto 5 minutos). Un bloqueo que pasa ese tiempo sin renovarse se vuelve a reclamar: vuelve a pending si al trabajo le quedan intentos; de lo contrario, se envía a la cola de fallidos (dead-letter) con un error que explica lo sucedido.

Así, un handler lento en un worker vivo conserva su trabajo, tarde lo que tarde, y el tiempo de espera solo decide cuánto tardan en volver los trabajos de un worker muerto. Un intento que perdió su bloqueo tampoco puede sobrescribir el resultado del trabajo: su éxito o fallo se registra en el log y se descarta, y decide el intento que tenga el bloqueo.

jobs: {
    enabled: true,
    concurrency: 5,              // jobs at once, per instance
    pollIntervalMs: 2_000,       // the first wait after a look that found nothing; it doubles up to 30 s
    visibilityTimeoutMs: 300_000 // how long an unrenewed claim holds
}

Un handler que nunca termina

La otra cara: un handler que espera un fetch que nunca responde retiene su ranura de worker, y su reclamo, mientras el proceso viva — cinco de ellos y el worker no ejecuta nada más, mientras /health sigue en verde. Desde 0.25 timeoutMs acota un intento. Al llegar al límite, el intento falla — se reintenta mientras queden intentos, como cualquier excepción —, su ranura se libera y el signal del handler se aborta. Pásalo a lo que corresponda, para que el trabajo también se detenga; un handler que lo ignora sigue ejecutándose en segundo plano sin que nada registre su resultado.

Si se omite, son 15 minutos (900_000), y el worker lo indica en una línea de log al arrancar. Un job que legítimamente dura más fija su propio valor, y timeoutMs: 0 elimina el límite.

jobs: {
    enabled: true,
    timeoutMs: 120_000,          // one attempt; 15 minutes when left out, 0 for no limit
    tasks: {
        "sync-crm": async ({ payload, signal }) => {
            await fetch(crmUrl, { method: "POST", body: JSON.stringify(payload), signal });
        }
    }
}

Los handlers de cola y las suscripciones a temas reciben el mismo signal en su contexto.

Varias instancias

Seguro por diseño. Los workers reclaman con SELECT … FOR UPDATE SKIP LOCKED, por lo que cada trabajo va a exactamente uno de ellos y los demás pasan a la siguiente fila en lugar de hacer cola detrás de ella. No es necesario elegir ningún líder.

Una instancia solo reclama las tareas para las que tiene un handler. Durante un despliegue progresivo (rolling deploy), una instancia que ejecuta código más antiguo deja de lado los trabajos cuya tarea solo implementa el código nuevo: permanecen en pending, sin gastar ningún intento, hasta que un par actualizado los reclama. Una tarea que ninguna instancia implementa también permanece en pending, en lugar de ir a la cola de mensajes fallidos (dead-letter).

Webhooks duraderos

WebhookDispatcher encola sus entregas en memoria por defecto, lo que significa que una caída o un despliegue entre el cambio y la entrega descarta el evento. Proporciónele la cola y cada entrega se convertirá en una fila:

import { WebhookDispatcher, WEBHOOK_DELIVERY_TASK } from "@rebasepro/server";

const { jobQueue } = await initializeRebaseBackend({ jobs: { enabled: true } });

const dispatcher = new WebhookDispatcher({ jobQueue });
dispatcher.setWebhooks(myWebhooks);

jobQueue?.register(WEBHOOK_DELIVERY_TASK, ctx => dispatcher.deliverQueuedJob(ctx.payload as never, ctx));

Solo el id del webhook se almacena en el trabajo, nunca el webhook en sí; de lo contrario, su secreto de firma permanecería en texto plano en rebase.jobs durante todo el tiempo que la retención conserve la fila, y un webhook editado entre el encolado y la entrega debe enviarse con su estado actual. Pasar ctx como segundo argumento hace que cada intento envíe su propio X-Webhook-Attempt; todos los intentos de una misma entrega envían el mismo X-Webhook-Delivery, de modo que un receptor puede deduplicar los reintentos.

Una entrega que el dispatcher se niega a enviar (un destino loopback, link-local, privado o que no sea http(s)), o una que el receptor responde con una redirección, falla de la misma manera cada vez. Pasa a dead-letter tras un solo intento, con el motivo en last_error. Cualquier otro fallo se reintenta como en cualquier trabajo.

Apagado

shutdown() detiene al worker para que no reclame nuevos trabajos y espera a los que están en curso, de modo que un despliegue no ejecute dos veces el final de un lote. La espera está limitada a dos tercios del tiempo de espera de apagado (10 segundos de los 15 predeterminados), para que un handler que nunca termina no pueda retener el apagado. Un trabajo que siga ejecutándose después conserva su reclamo y se recupera mediante el tiempo de espera de visibilidad.

Próximos pasos

• Trabajos Cron — trabajo según una programación
• Webhooks — notificar a otros sistemas ante un cambio