Fiabilité des flux : rate‑limits, backpressure et monitoring des événements

1. Pourquoi les flux sont particulièrement sensibles à la charge

Dans les parties précédentes, nous avons déjà décrit la structure des événements MCP, les statuts job.progress/job.completed, les jobs asynchrones et les canaux de streaming (SSE/HTTP-stream) pour GiftGenius. Il est maintenant important de voir comment cette architecture se comporte sous une charge réelle.

Tant que vous n’avez qu’un seul utilisateur qui lance de temps en temps une recherche de cadeau, tout semble parfait. Mais dès que GiftGenius arrive en production et reçoit simultanément des centaines de requêtes « trouver des cadeaux pour tous les employés d’un événement d’entreprise », vous découvrez soudain que :

• le serveur maintient des centaines de connexions SSE longue durée ;
• les workers envoient joyeusement des job.progress à tout bout de champ ;
• les logs grossissent de plusieurs gigaoctets par jour ;
• l’UI côté utilisateur commence à saccader, alors que « le serveur ne tombe pas ».

Une requête HTTP classique vit des millisecondes ou des secondes. Un flux SSE ou HTTP‑stream peut vivre des minutes voire des heures. Il retient une connexion, de la mémoire, des descripteurs de fichiers. Chaque événement envoyé implique une sérialisation JSON, une copie sur le réseau, du travail pour le GC. Si vous considérez cela comme « encore un simple console.log côté back‑end », votre système se transformera très vite en radiateur.

Les événements MCP ont une autre particularité : ils sont souvent générés de multiples fois pour la même tâche. Un worker qui met à jour le progrès tous les 0,1 % produit un nombre impressionnant d’événements par job. Au final, vous obtenez du « bruit » : une énorme quantité de petits messages qui :

• chargent le réseau et le CPU ;
• engorgent les files et les buffers ;
• rendent le debug et l’analyse des logs pénibles.

C’est pourquoi les flux et les événements MCP doivent être traités aussi sérieusement que des requêtes à la base de données ou des appels à un modèle : ce sont des ressources coûteuses qui exigent normalisation, contrôle et monitoring.

Pour gérer cela, gardons en tête trois grands thèmes :

1. Rate‑limits — limiter combien et à quelle fréquence nous pouvons nous permettre de générer et d’envoyer des événements/flux.
2. Backpressure — réagir lorsque le consommateur n’arrive pas à suivre le producteur.
3. Monitoring et métriques — mesurer ce qui se passe et détecter à temps quand tout commence à bouillir.

2. Rate‑limiting des flux et des événements

Commençons par le plus évident — les limites.

Il est important de comprendre que, dans les scénarios de streaming, le « perturbateur dangereux » n’est souvent pas le client mais le serveur. Dans les API REST classiques, vous limitez le nombre de requêtes vers le serveur pour éviter qu’un utilisateur ne se transforme en DDoS. Dans le monde MCP et des flux, il est très facile de provoquer un DDoS inversé : un worker ou un serveur MCP bombarde le client de milliers d’événements par seconde.

Quels types de limites sont nécessaires

On y pense généralement sur trois axes.

Premièrement, des limites par utilisateur ou session. On ne peut pas autoriser un utilisateur à ouvrir vingt widgets « master » GiftGenius en parallèle, chacun avec son propre flux SSE. Une limite raisonnable — quelques flux actifs par session et une restriction sur le nombre de jobs avec le statut running pour un même utilisateur ou locataire.

Deuxièmement, des limites par job. Ici, nous nous intéressons à la fréquence des événements. Il suffit d’envoyer job.progress au plus une fois toutes les N millisecondes ou seulement en cas de changement significatif, par exemple tous les 5 points de progrès. Inutile d’envoyer un message pour chaque produit traité du catalogue. Il est également pertinent de limiter la taille du payload : un événement de progrès ne devrait pas transporter des mégaoctets de texte.

Troisièmement, des limites par IP ou organisation. C’est une protection contre les abus lorsqu’on lance un script qui spamme des tâches ou quand votre App devient subitement populaire. Ici, on fait intervenir des mécanismes connus d’API‑gateways et de proxies.

Implémentation simple d’un limiteur de fréquence d’événements

Considérons un worker GiftGenius qui, en arrière‑plan, cherche des cadeaux pour une longue liste de destinataires et envoie périodiquement le progrès via une notification MCP event/progress. Nous voulons que les événements soient envoyés au plus une fois toutes les 500 ms et uniquement en cas de variation d’au moins 5 points.

Pseudo‑code TS pour le worker :

// supposons qu’il existe un mcpClient.sendNotification(...)
let lastSentPercent = 0;
let lastSentAt = 0;

function reportProgress(jobId: string, percent: number, message: string) {
  const now = Date.now();
  const percentDelta = percent - lastSentPercent;
  const timeDelta = now - lastSentAt;

  // envoyer uniquement si >= 500 ms se sont écoulées OU si la hausse est >= 5 %
  if (percentDelta >= 5 || timeDelta >= 500) {
    mcpClient.sendNotification("event/progress", {
      jobId,
      percent,
      message,
    });
    lastSentPercent = percent;
    lastSentAt = now;
  }
}

Cette approche s’appelle le throttling : on « éclaircit » le flux d’événements dans le temps et selon l’ampleur du changement.

Si vous découpez en étapes (« Étape 1 sur 3 », « Étape 2 sur 3 »), la logique est encore plus simple : n’envoyer des événements que lors d’un changement d’étape.

Limite du nombre de flux ouverts simultanément

Côté serveur MCP, vous avez probablement un handler HTTP pour SSE :

// app/api/events/[userId]/route.ts (Next.js 16 App Router)
export async function GET(
  req: Request,
  { params }: { params: { userId: string } },
) {
  const userId = params.userId;

  if (!canOpenMoreStreams(userId)) {
    return new Response("Too many streams", { status: 429 });
  }

  const stream = new ReadableStream({
    start(controller) {
      registerSseClient(userId, controller);
    },
    cancel() {
      unregisterSseClient(userId);
    },
  });

  return new Response(stream, {
    headers: { "Content-Type": "text/event-stream" },
  });
}

La fonction canOpenMoreStreams peut vérifier le nombre de connexions ouvertes pour l’utilisateur et le comparer à un seuil (par exemple, pas plus de trois flux parallèles). Si la limite est dépassée, on renvoie 429 et, dans les instructions GPT, on explique au modèle que, dans cette situation, il vaut mieux ne pas lancer un autre « master » long, mais suggérer à l’utilisateur qu’« une recherche est déjà en cours, attendons qu’elle se termine ».

Dans de petits systèmes, de tels contrôles peuvent être implémentés en mémoire de processus. Dans une infrastructure plus sérieuse, cela migre vers un MCP‑gateway ou un service de rate‑limit dédié.

3. Backpressure : que faire quand le consommateur n’arrive pas à suivre

Les rate‑limits restreignent combien d’événements nous voulons produire. Mais même avec des limites prudentes, il est possible que le consommateur « s’étouffe » : l’utilisateur a un mauvais Internet mobile, l’onglet du navigateur gèle, ChatGPT est fortement chargé à ce moment‑là.

Le backpressure est la réaction du système au fait que le consommateur ne suit pas. Au lieu d’accumuler indéfiniment des données et de finir par tomber avec un OOM, nous décidons sciemment de :

• ralentir ;
• agréger les événements ;
• écarter les moins importants.

Où la pression s’exerce

Un scénario typique pour GiftGenius peut ressembler à ceci. Le worker écrit les événements dans une file (par exemple, Redis Streams ou simplement une table BD), le serveur MCP les lit et les pousse dans le canal SSE. Si le client est lent (3G, vieux laptop, plein d’autres onglets), le buffer TCP commence à se remplir, le processus Node n’arrive pas à vider entièrement la file et finit par accumuler des événements en mémoire. Ensuite, vous voyez le classique :

FATAL ERROR: Ineffective mark-compacts near heap limit

Le backpressure au niveau réseau (TCP) est déjà là, mais il ne connaît rien de vos entités métier. Il dit juste : « Hé, ralentis, le buffer est plein ». Notre tâche est d’interpréter cela au niveau des événements MCP.

Mise en mémoire tampon bornée et suppression d’événements

Pour le progrès et les statuts, nous avons un avantage : tous les événements n’ont pas la même valeur. L’utilisateur a besoin du dernier pourcentage pertinent, pas de l’historique des « 51 %, 52 %, 53 %, 54 % ». Cela signifie que nous pouvons sans crainte jeter une partie des événements et n’envoyer que le dernier.

Supposons que nous ayons une couche qui reçoit les événements de progrès des workers et les place dans un buffer pour chaque jobId :

type ProgressEvent = { jobId: string; percent: number; message: string };

const progressBuffers = new Map<string, ProgressEvent[]>();
const MAX_BUFFER = 10;

function bufferProgress(event: ProgressEvent) {
  const buffer = progressBuffers.get(event.jobId) ?? [];
  buffer.push(event);

  // limiter la taille du buffer
  if (buffer.length > MAX_BUFFER) {
    // ne garder que les derniers événements
    progressBuffers.set(event.jobId, buffer.slice(-MAX_BUFFER));
  } else {
    progressBuffers.set(event.jobId, buffer);
  }
}

Un timer séparé, par exemple toutes les 500 ms, regarde le buffer et n’envoie que le dernier événement, en ignorant les autres :

setInterval(() => {
  for (const [jobId, buffer] of progressBuffers.entries()) {
    if (!buffer.length) continue;

    const last = buffer[buffer.length - 1];
    sendProgressToClient(last); // SSE/MCP notification

    progressBuffers.set(jobId, []); // on vide
  }
}, 500);

C’est un exemple de tactique de conflation : regrouper plusieurs mises à jour en une seule, à jour. Pour le progrès — c’est un pattern en or.

Pour des événements de type « log » ou partial_result, la stratégie peut être différente. La perte d’événements y est souvent inacceptable : le texte des logs est important, et un chunk JSON perdu peut casser la structure des données. Dans ces cas, vous pouvez :

• agréger les messages (coller plusieurs lignes de logs en un seul paquet) ;
• ou envoyer un signal de contrôle au worker « ralentis la génération de logs ».

Dans les systèmes asynchrones, la deuxième option est plus complexe, mais il vaut la peine au moins d’y réfléchir.

Limiter la profondeur des files

Le backpressure ne se limite pas au buffer d’événements juste avant l’envoi. Il faut regarder toutes les files du système :

• la file des tâches en attente d’un worker ;
• la file des événements entre le worker et le serveur MCP ;
• les buffers à l’intérieur des bibliothèques de streaming côté serveur.

Pour chaque file, il est important de fixer une limite raisonnable de profondeur. Si une file déborde, soit vous répondez aux clients « le système est surchargé, réessayez plus tard », soit vous abandonnez les jobs moins importants, soit vous basculez certains scénarios en « mode hors ligne » (par exemple, vous générez un rapport et envoyez son lien plus tard).

Une approche intéressante — prioriser les types d’événements. En surcharge, vous pouvez commencer à n’envoyer que job.completed et job.failed, et diminuer la priorité de job.progress, voire le désactiver complètement.

4. Monitoring des flux et des événements

Sans mesures, toute cette belle histoire de rate‑limits et de backpressure vire à la magie. Il faut voir quand le nombre de flux devient suspect, quand les événements arrivent avec du lag, et quand les clients se déconnectent en masse.

Les flux se comportent différemment des requêtes HTTP classiques : leur durée peut se compter en minutes et heures, donc les métriques « requêtes par seconde » et « latence moyenne » ne donnent pas toute l’image.

Indicateurs clés

Pour des flux SSE ou HTTP/stream, il est utile de suivre plusieurs groupes d’indicateurs.

1. Métriques de connexions. Combien de flux SSE sont actuellement actifs ? Quelle est la durée de vie moyenne d’une connexion ? Quel pourcentage de flux se termine par une erreur ou un timeout ? Une flambée brutale des connexions actives indique une possible tempête de trafic ou une fuite de ressources (les clients ne ferment pas les connexions). Une chute brutale — une déconnexion massive (par exemple, un problème réseau ou un bug critique côté serveur).
2. Métriques d’événements. Combien d’événements envoyez‑vous par seconde sur l’ensemble des flux (EPS — events per second, en clair nombre d’événements par seconde) ? Quelle est la taille moyenne d’un événement ? Combien d’erreurs de désérialisation ou de validation du payload observez‑vous ? Si vous voyez soudain une hausse de la taille des événements, il est possible que quelqu’un se mette à envoyer dans job.progress le texte complet d’un rapport au lieu d’une courte chaîne.
3. Métriques des jobs. Répartition par statuts (pending, running, completed, failed, canceled), temps moyen d’exécution par type de tâche, pourcentage de jobs qui partent en retry ou en dead‑letter. Cela aide à comprendre que les problèmes ne se limitent pas au réseau mais touchent aussi les workers : une API externe est devenue plus lente, des erreurs massives sont apparues.
4. Métriques de backpressure et indicateurs système. Dans les systèmes de streaming, on observe souvent la profondeur des buffers et des files entre composants, ainsi que le pourcentage de temps pendant lequel le flux est bloqué en attente que le consommateur libère de la place. Si vos files sont presque toujours pleines à ras bord, c’est un signal clair que le système est à la limite. Il est aussi important de surveiller les indicateurs système : CPU et mémoire sur les serveurs qui gèrent le streaming, et les erreurs/timeouts au niveau réseau. Parfois, c’est la bande passante entre le serveur MCP et ChatGPT qui devient le goulet d’étranglement.

Au total, ces quatre groupes répondent à trois questions : combien de flux vivent en ce moment, combien de données vous transférez, comment se comportent les jobs et où exactement le système commence à s’étouffer.

Que journaliser

Les logs sont le deuxième pilier de l’observabilité. Il est important de journaliser les événements et les connexions de manière à pouvoir reconstituer l’historique d’un job spécifique.

On ajoute généralement aux logs, pour chaque événement et chaque flux :

• jobId et/ou eventId ;
• userId et sessionId (en cas de multi‑tenant) ;
• type d’événement (progress, completed, failed, resource.updated) ;
• type de canal (SSE ou HTTP/stream) ;
• timestamp d’envoi et, si possible, timestamp de naissance de l’événement côté worker.

Ainsi, on peut calculer le lag : la différence entre le moment où le worker a généré l’événement et le moment où il est parti dans le socket. La hausse de ce lag est un bon indicateur de problèmes de backpressure.

Il faut rester prudent pour que les logs ne deviennent pas eux‑mêmes une source de surcharge. Pour des événements à haute fréquence comme job.progress, journaliser chaque événement n’est pas toujours judicieux ; on peut activer de l’échantillonnage — consigner chaque N^e événement au lieu de tous — ou agréger des statistiques.

Côté code, cela peut ressembler à un simple helper :

function logEvent(event: {
  type: string;
  jobId: string;
  userId?: string;
  channel: "sse" | "http-stream";
  payload: unknown;
}) {
  console.info({
    ...event,
    timestamp: new Date().toISOString(),
  });
}

Dans un projet réel, vous envelopperez cela dans une bibliothèque de structured logging, mais l’idée reste : un maximum de contexte utile dans chaque entrée.

5. Alertes et politiques de dégradation

Une fois que vous avez des métriques et des logs, l’étape suivante consiste à configurer des alertes et à réfléchir à la façon dont le système doit « se dégrader » quand il va mal. L’idée : mieux vaut fonctionner honnêtement moins bien que tomber d’un coup.

Exemples d’alertes

Pour GiftGenius, il est judicieux de surveiller quelques situations typiques.

Premièrement, un nombre anormal de flux actifs. Si d’ordinaire vous avez des dizaines de connexions SSE actives et que, soudain, vous en avez des milliers, il faut comprendre ce qui se passe. Peut‑être êtes‑vous devenu populaire, ou peut‑être avez‑vous un bug et les connexions ne se ferment pas.

Deuxièmement, le retard entre l’achèvement effectif d’un job et la réception de job.completed côté client. Si ce retard commence à dépasser un seuil (disons 5–10 secondes), cela signifie qu’entre le worker et le client les événements s’accumulent ou que les connexions patinent.

Troisièmement, une part élevée de job.failed ou job.canceled par rapport aux succès. La cause peut venir du worker (API externe cassée, nouveau bug) ou d’une sensibilité accrue des utilisateurs aux retards (ils annulent plus souvent).

Enfin, un taux accru d’erreurs de connexion et de ruptures de flux : si le nombre de disconnects non standard augmente, il peut y avoir des problèmes réseau ou côté client, et il vaut la peine de réfléchir à des scénarios de repli.

Schémas de dégradation

Quand le système est surchargé, on peut activer un « mode économie de ressources ». C’est mieux que de répondre des 500 partout.

Le schéma le plus courant — une fréquence d’événements adaptative. Si vous voyez que l’event‑rate (nombre d’événements par seconde) a été multiplié par dix et que le lag commence à grimper dans les files, réduisez la fréquence des événements de progrès. Elle était à chaque 1 % — passez à chaque 10 %. Elle était toutes les 500 ms — faites une fois toutes les 2–3 secondes. L’utilisateur vivra très bien sans un progrès ultra‑précis, mais beaucoup moins avec une UI complètement figée.

Pour des événements moins importants — par exemple, resource.updated lors de la mise à jour en arrière‑plan d’un flux de produits — on peut désactiver temporairement l’envoi tant que le système est sous charge.

Une autre approche — basculer une partie des scénarios des flux vers un polling périodique. Si les canaux SSE s’effondrent, le serveur MCP peut envoyer au widget un événement système tel que system.overloaded, et le widget basculer vers une stratégie « toutes les N secondes, j’interroge un endpoint REST sur le statut du job ».

6. Petit fragment pratique pour GiftGenius

Pour relier le tout, imaginons que nous ayons déjà :

• un MCP‑tool startGiftSearch qui crée un job et renvoie un jobId ;
• un worker qui exécute la recherche et envoie event/progress et event/completed ;
• un endpoint SSE /api/events/[userId] auquel le widget se connecte dans Next.js.

Ajoutons un simple niveau de protection contre les « tempêtes d’événements » et un monitoring minimal.

Limiter le progrès par pas et par temps

Dans le worker, nous ajoutons le throttling et la conflation, comme évoqué plus haut. Désormais, les événements sont envoyés au plus une fois toutes les demi‑secondes et en cas de variation d’au moins 5 %.

Comptage des flux actifs

Dans l’endpoint SSE, nous conservons un compteur par utilisateur :

const activeStreams = new Map<string, number>();
const STREAM_LIMIT = 3;

function canOpenMoreStreams(userId: string) {
  const current = activeStreams.get(userId) ?? 0;
  return current < STREAM_LIMIT;
}

function registerSseClient(userId: string, controller: ReadableStreamDefaultController) {
  const current = activeStreams.get(userId) ?? 0;
  activeStreams.set(userId, current + 1);

  // ici, vous enregistrez le controller dans une structure
  // pour pouvoir écrire des événements dans ce flux plus tard
}

function unregisterSseClient(userId: string) {
  const current = activeStreams.get(userId) ?? 1;
  activeStreams.set(userId, Math.max(0, current - 1));
}

Le serveur peut en plus exporter des métriques basées sur activeStreams.size vers Prometheus/Grafana ou tout autre système de monitoring.

Métrique event‑rate la plus simple

Pour commencer, on peut au moins compter combien d’événements on envoie :

let eventsSentLastMinute = 0;

function sendProgressToClient(ev: ProgressEvent) {
  // ... sérialisation et écriture dans le flux SSE
  eventsSentLastMinute++;
}

setInterval(() => {
  console.info({
    metric: "events_per_minute",
    value: eventsSentLastMinute,
    timestamp: new Date().toISOString(),
  });
  eventsSentLastMinute = 0;
}, 60_000);

Avec le temps, on pourra remplacer cela par de vrais compteurs et des alertes, mais comme point de départ — c’est déjà pas mal.

Si vous assemblez tout ce qui précède — limites, backpressure, métriques/alertes et UX de repli adéquat — votre GiftGenius cesse d’être « une démo pour démo » et résiste à de vraies tempêtes de trafic. Dans les modules suivants, où nous parlerons de gateway, d’architecture de production et d’observabilité complète, ces patterns nous seront encore utiles.

7. Erreurs courantes avec les flux, les rate‑limits et le monitoring

Erreur n° 1 : absence de limites sur le nombre de flux et la fréquence des événements.
Les développeurs ont ajouté SSE « pour faire joli », les workers envoient fidèlement le progrès à chaque objet traité, et tout semble fonctionner en démo. Mais au premier pic d’utilisateurs réels, le serveur consacre la majorité de ses ressources à la sérialisation et à l’envoi de milliers de minuscules événements, et l’UI dans ChatGPT se transforme en diaporama.

Erreur n° 2 : tenter de mettre « tout en buffer » sans limite.
Le code contient un tableau sans limite des « événements non envoyés », qui grossit jusqu’à ce que le client se rétablisse. Spoiler : il ne se rétablit pas, le serveur meurt avant. Tout buffer doit avoir un maximum strict, et la logique de gestion du débordement doit être explicite.

Erreur n° 3 : traiter tous les types d’événements de la même façon.
Le progrès peut être agrégé et jeté (le dernier pourcentage compte plus que l’historique du mouvement). Avec les logs et les résultats partiels, on ne peut pas faire ça — la perte d’un chunk peut signifier des données corrompues. Lors de la conception du système, regroupez à l’avance les événements par importance et définissez pour chaque groupe une stratégie en cas de surcharge.

Erreur n° 4 : absence d’observabilité.
Aucune métrique sur les flux actifs, aucun suivi de l’event‑rate, et dans les logs — seulement « quelque chose s’est mal passé ». Dans cette situation, vous n’apprenez les problèmes que via les retours des utilisateurs et la courbe de charge CPU. Mettre en place au moins des métriques de base et des logs par jobId et eventId — ce n’est pas un luxe, c’est une nécessité.

Erreur n° 5 : un UX rigide qui n’anticipe pas la dégradation.
Le widget et les instructions GPT partent du principe que le flux est toujours disponible, que le progrès se met à jour « en temps réel », que les partial‑results arrivent strictement selon le scénario. Au premier problème réseau, l’utilisateur voit une barre de progrès « figée » sans aucune explication. Il est bien préférable de prévoir un repli honnête : « En ce moment, il y a des problèmes avec la mise à jour en direct, je continue malgré tout la recherche et je vous informe quand j’ai terminé » — et passer à des mises à jour plus rares ou au polling.

Erreur n° 6 : croire que « nos utilisateurs ne créeront pas beaucoup de tâches simultanées ».
L’expérience montre que si vous ne limitez pas le nombre de jobs et de flux parallèles, quelqu’un ouvrira forcément cinq onglets, lancera dans chacun une recherche de cadeaux « au maximum » et ira prendre un café. L’idée « ça passera » en production se termine presque toujours par une découverte forcée du monitoring sur fond de vacarme d’alertes.