Construire une file de tâches simple avec Redis

#redis#backend#nodejs

Envoyer un e-mail de bienvenue, redimensionner une image téléversée, générer un rapport PDF : rien de tout cela ne devrait se faire dans la requête HTTP qui l'a déclenché. Une file de tâches permet de déléguer le travail et de répondre immédiatement à l'utilisateur, et Redis dispose juste de ce qu'il faut de primitives intégrées pour en construire une raisonnablement solide, sans recourir à un broker de messages plus lourd.

Pourquoi ne pas simplement faire le travail en ligne

La version naïve effectue le travail lent directement dans le gestionnaire de requête :

app.post('/signup', async (req, res) => {
  const user = await createUser(req.body);
  await sendWelcomeEmail(user); // blocks the response on an SMTP round trip
  res.json(user);
});

Si le fournisseur d'e-mails est lent ou brièvement indisponible, chaque inscription se bloque ou échoue, alors que la création de l'utilisateur a réussi. Une file découple « l'utilisateur a été créé » de « l'e-mail a été envoyé » : la requête rend la main vite, et un processus worker distinct gère l'envoi, avec nouvelles tentatives, de façon indépendante.

La primitive de base : un retrait bloquant dans une liste

Les listes Redis prennent en charge LPUSH (ajouter en tête) et BRPOP (retirer par la fin en bloquant, en attendant si la liste est vide). Cette paire forme une file :

import { createClient } from 'redis';

const redis = createClient();
await redis.connect();

// Producer: enqueue a job
async function enqueue(queueName, job) {
  await redis.lPush(queueName, JSON.stringify(job));
}

// Consumer: block until a job is available, then process it
async function dequeue(queueName) {
  const result = await redis.brPop(queueName, 0); // 0 = block indefinitely
  return JSON.parse(result.element);
}

Utilisation :

await enqueue('emails', { type: 'welcome', userId: user.id });

// worker.js — runs as a separate process
while (true) {
  const job = await dequeue('emails');
  await sendWelcomeEmail(job.userId);
}

Cela fonctionne, mais il y a une faille sérieuse : si le worker plante après que BRPOP a retiré la tâche de la liste mais avant qu'il ait fini de la traiter, cette tâche est perdue. BRPOP est destructif dès l'instant où il retourne : rien n'indique que la tâche a un jour été prise en charge.

La rendre fiable : le schéma de file fiable

La documentation de Redis décrit un schéma pour cela : déplacer atomiquement la tâche de la file principale vers une liste « en cours de traitement » propre à chaque worker, de sorte qu'elle ne se trouve jamais dans un état où elle n'existe nulle part.

async function dequeueReliable(queueName, processingKey) {
  const raw = await redis.brPopLPush(queueName, processingKey, 0);
  return raw ? JSON.parse(raw) : null;
}

async function ack(processingKey, rawJob) {
  await redis.lRem(processingKey, 1, rawJob);
}

while (true) {
  const raw = await redis.brPopLPush('emails', 'emails:processing', 0);
  try {
    const job = JSON.parse(raw);
    await sendWelcomeEmail(job.userId);
    await redis.lRem('emails:processing', 1, raw);
  } catch (err) {
    console.error('job failed, leaving in processing list for recovery', err);
  }
}

Si le worker meurt en pleine tâche, la tâche reste dans emails:processing au lieu de disparaître. Un processus de récupération distinct (exécuté périodiquement, ou au démarrage d'un worker) peut parcourir cette liste à la recherche de tâches plus anciennes qu'un certain seuil et les remettre dans la file principale pour une nouvelle tentative.

Ajouter des nouvelles tentatives avec backoff

Une tâche qui échoue ne doit pas simplement revenir aussitôt et échouer de nouveau en boucle serrée. Suivez un compteur de tentatives dans le contenu de la tâche et utilisez une remise en file différée :

async function handleJob(raw, processingKey) {
  const job = JSON.parse(raw);
  try {
    await sendWelcomeEmail(job.userId);
    await redis.lRem(processingKey, 1, raw);
  } catch (err) {
    await redis.lRem(processingKey, 1, raw);
    job.attempts = (job.attempts ?? 0) + 1;

    if (job.attempts >= 5) {
      await redis.lPush('emails:dead-letter', JSON.stringify(job));
      return;
    }

    const delayMs = 1000 * 2 ** job.attempts;
    setTimeout(() => enqueue('emails', job), delayMs);
  }
}

Notez la file des lettres mortes (dead-letter queue) : après suffisamment d'échecs, cessez de réessayer automatiquement et déplacez la tâche à un endroit où un humain peut l'examiner. Sans cela, une tâche définitivement cassée (données invalides, service en aval réellement disparu) est réessayée indéfiniment et noie le vrai travail dans les logs.

Tâches planifiées ou différées avec un ensemble trié

LPUSH/BRPOP ne gère que « exécuter dès que possible ». Pour « exécuter à un moment précis du futur » (un e-mail de rappel, une vérification de renouvellement d'abonnement), un ensemble trié convient mieux, avec l'horodatage prévu comme score :

async function scheduleAt(queueName, job, runAtMs) {
  await redis.zAdd(`${queueName}:scheduled`, {
    score: runAtMs,
    value: JSON.stringify(job),
  });
}

Un poller déplace périodiquement les tâches échues vers la file ordinaire :

async function promoteDueJobs(queueName) {
  const now = Date.now();
  const due = await redis.zRangeByScore(`${queueName}:scheduled`, 0, now);

  for (const raw of due) {
    await redis.multi()
      .zRem(`${queueName}:scheduled`, raw)
      .lPush(queueName, raw)
      .exec();
  }
}

setInterval(() => promoteDueJobs('emails'), 1000);

La transaction multi() est importante ici : elle garantit qu'une tâche n'est jamais retirée de l'ensemble planifié sans atterrir aussi dans la file active, même si le processus est interrompu entre les deux opérations.

Quand arrêter de tout construire soi-même

Vous obtenez ainsi une file réellement utile, mais les systèmes de tâches en production ont aussi besoin de choses qui demandent un vrai effort pour être bien faites : priorités des tâches, limites de concurrence par file, tableau de bord d'inspection des tâches bloquées, garanties de livraison quasi « exactement une fois » lors d'un basculement Redis. À ce stade, recourez à une bibliothèque maintenue : BullMQ est le choix standard de l'écosystème Node et implémente tous les schémas ci-dessus (et davantage) au-dessus de Redis. Construire d'abord soi-même une version minimale vaut la peine une fois : c'est le moyen le plus rapide de comprendre ce dont une file de tâches est réellement responsable, pour évaluer si les valeurs par défaut d'une bibliothèque conviennent à vos modes de défaillance au lieu de leur faire aveuglément confiance.