Tâches en arrière-plan

Vue d’ensemble

Un job est une ligne dans rebase.jobs. Il est réservé par exactement un worker, réessayé avec un délai croissant si son gestionnaire lève une exception, et conservé dans la table lorsqu’il abandonne finalement afin que quelqu’un puisse l’examiner.

Il n’y a rien à installer et rien à exécuter aux côtés de Postgres. Un job mis en file d’attente au sein d’une transaction qui subit un rollback n’a jamais été mis en file d’attente.

Utilisez-le pour les tâches qui ne doivent pas être perdues et ne doivent pas se dérouler au cours d’une requête : envoi d’e-mails, appel à un service tiers, génération de fichiers, réconciliation avec un système externe.

	Exécution	Survit à un redémarrage
Cron	Selon un calendrier	Oui — le calendrier est dans le code
Jobs	Une fois, dès qu’un worker est libre	Oui — le job est une ligne
Un setTimeout dans un callback	Une fois, dans ce processus	Non

Activation

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

Désactivé par défaut : un worker continue de chercher du travail tant que le processus vit, ce qui n’est pas un comportement par défaut souhaité. Il nécessite un pilote capable d’exécuter du SQL — sur un pilote qui ne le peut pas (MongoDB), la file d’attente est indisponible et vous en êtes informé au démarrage plutôt qu’à la première mise en file d’attente.

Depuis 0.25 Un worker inactif n’interroge pas la base toutes les deux secondes. Chaque mise en file d’attente envoie un NOTIFY sur le canal rebase_jobs lorsqu’elle est validée, et un worker qui détient un LISTEN dessus réserve aussitôt — un job mis en file d’attente par un autre processus (le rôle api, une autre réplique) démarre en quelques millisecondes, et un job différé démarre à son échéance. En arrière-plan, il regarde de toute façon toutes les 30 secondes. Un job que ce processus met lui-même en file d’attente, une nouvelle tentative qu’il a planifiée et un job qu’il a récupéré le réveillent sans notification.

Le LISTEN a besoin d’une connexion à part, trouvée comme le fait le temps réel : DATABASE_DIRECT_URL si elle est définie, sinon DATABASE_URL. Derrière un pooler en mode transaction tel que PgBouncer, définissez DATABASE_DIRECT_URL : un pooler ne conserve pas un LISTEN, et les jobs d’autres processus attendraient alors la vérification de 30 secondes. Lorsqu’il n’existe aucune connexion de ce type (la base de développement en processus), un worker inactif ralentit à la place : il regarde après pollIntervalMs (2 secondes), puis deux fois plus longtemps après chaque vérification qui n’a rien trouvé, jusqu’à 30 secondes, et repart de zéro dès qu’il trouve du travail. Un worker inactif effectue environ 2 900 réservations par jour au lieu de 43 200.

Mise en file d’attente

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

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

Options

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 fusionne un double-clic, une requête réessayée et deux instances réagissant au même événement en un seul job. Sa portée est limitée aux tâches non terminées, de sorte que la clé redevient réutilisable une fois le job terminé — sans quoi « le récapitulatif nocturne pour l’utilisateur 7 » ne pourrait être envoyé qu’une seule fois dans l’absolu. Une mise en file d’attente en doublon renvoie null au lieu de lever une exception : la tâche demandée est bien en file d’attente, ce qui correspond au résultat souhaité.

Échecs

Un gestionnaire échoue en levant une exception. Il n’y a pas de return false — un booléen serait ignoré silencieusement par chaque gestionnaire ayant oublié d’en renvoyer un, et l’échec doit être le comportement par défaut.

• Tentatives restantes → retour à pending, avec run_at repoussé par le backoff (1s, 5s, 25s … plafonné à une heure ; surchargeable avec backoff).
• Plus de tentatives → failed, et la ligne reste. Une file d’attente qui supprime silencieusement ce qu’elle n’a pas pu traiter est indiscernable d’une file sans aucune tâche à accomplir.

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

Les lignes en échec sont conservées 30 jours ; celles ayant réussi, 3 jours. Chaque worker supprime les lignes expirées à son démarrage, puis toutes les heures, par lots, en dehors du chemin des requêtes.

Que se passe-t-il lorsqu’un worker s’arrête brutalement

Un processus interrompu en plein traitement ne peut pas libérer sa réservation ; seul un timeout permettra donc de débloquer la ligne. Pendant l’exécution d’un gestionnaire, son worker renouvelle la réservation à chaque tiers de visibilityTimeoutMs (5 minutes par défaut). Une réservation restée aussi longtemps sans être renouvelée est récupérée — retour à pending s’il reste des tentatives au job, sinon envoyée en rebut (dead-letter) avec une erreur explicitant ce qui s’est produit.

Ainsi, un gestionnaire lent sur un worker actif conserve son job, quelle que soit la durée du traitement, et le timeout ne détermine que le délai au bout duquel les jobs d’un worker mort reviennent. Une tentative qui a perdu sa réservation ne peut pas non plus écraser le résultat du job : son succès ou son échec est journalisé puis ignoré, et c’est la tentative qui détient la réservation qui tranche.

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 gestionnaire qui ne se termine jamais

L’envers du décor : un gestionnaire qui attend un fetch qui ne répond jamais conserve son emplacement de worker, et sa réservation, tant que le processus vit — avec cinq d’entre eux, le worker n’exécute plus rien d’autre, alors que /health reste vert. Depuis 0.25 timeoutMs borne une tentative. À l’échéance, la tentative échoue — retentée tant qu’il reste des tentatives, comme n’importe quelle exception — son emplacement est libéré, et le signal du gestionnaire est interrompu. Transmettez le signal, pour que le travail s’arrête lui aussi ; un gestionnaire qui l’ignore continue de s’exécuter en arrière-plan sans que rien n’enregistre son résultat.

S’il est omis, il vaut 15 minutes (900_000), et le worker l’indique dans une ligne de journal au démarrage. Une tâche qui dure légitimement plus longtemps définit sa propre valeur, et timeoutMs: 0 supprime la limite.

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 });
        }
    }
}

Les gestionnaires de file et les abonnements à un topic reçoivent le même signal dans leur contexte.

Plusieurs instances

Sûr par conception. Les workers effectuent leur réservation avec SELECT … FOR UPDATE SKIP LOCKED, de sorte que chaque job est attribué à exactement l’un d’entre eux et que les autres passent à la ligne suivante plutôt que d’attendre derrière lui. Nul besoin d’élire un leader.

Une instance ne réserve que les tâches pour lesquelles elle a un handler. Lors d’un déploiement progressif (rolling deploy), une instance exécutant une ancienne version du code laisse de côté les jobs dont seule la nouvelle version implémente la tâche : ils restent pending, sans qu’aucune tentative ne soit consommée, jusqu’à ce qu’une instance mise à jour les réserve. Une tâche qu’aucune instance n’implémente reste elle aussi pending, plutôt que d’être placée en dead-letter.

Webhooks durables

Par défaut, WebhookDispatcher met en file d’attente ses envois en mémoire, ce qui signifie qu’un crash ou un déploiement entre la modification et l’envoi entraîne la perte de l’événement. Confiez-lui la file d’attente et chaque envoi devient une ligne :

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));

Seul l’identifiant du webhook est stocké dans le job, jamais le webhook lui-même — son secret de signature se trouverait sinon en clair dans rebase.jobs pendant toute la durée de rétention de la ligne, et un webhook modifié entre la mise en file d’attente et l’envoi doit être transmis avec sa configuration actuelle. Passer ctx en second argument fait que chaque tentative envoie son propre X-Webhook-Attempt ; toutes les tentatives d’un même envoi envoient le même X-Webhook-Delivery, ce qui permet à un destinataire de dédoublonner les nouvelles tentatives.

Un envoi que le dispatcher refuse d’effectuer (une destination loopback, link-local, privée ou non http(s)), ou auquel le destinataire répond par une redirection, échoue de la même manière à chaque fois. Il est placé en dead-letter après une seule tentative, avec la raison dans last_error. Tout autre échec est retenté comme pour n’importe quel job.

Arrêt

shutdown() empêche le worker de réserver de nouveaux jobs et attend la fin de ceux en cours de traitement, évitant ainsi qu’un déploiement n’exécute deux fois la fin d’un lot. L’attente est plafonnée aux deux tiers du timeout d’arrêt (10 secondes sur les 15 par défaut), afin qu’un handler qui ne se termine jamais ne puisse pas bloquer l’arrêt. Un job encore en cours d’exécution après ce délai conserve sa réservation et est récupéré par le timeout de visibilité.

Prochaines étapes

• Cron Jobs — planifier des tâches régulières
• Webhooks — notifier d’autres systèmes lors d’une modification