Fiabilidad de los flujos: rate limits, backpressure y monitorización de eventos

1. Por qué los flujos son especialmente sensibles a la carga

En partes anteriores ya vimos cómo están organizados los eventos MCP, los estados job.progress/job.completed, los async‑jobs y los canales de streaming (SSE/HTTP-stream) para GiftGenius. Ahora es importante ver qué le ocurre a esta arquitectura bajo una carga real.

Mientras tengas un solo usuario que de vez en cuando lanza la búsqueda de un regalo, todo se ve perfecto. Pero en cuanto GiftGenius llega a producción y entran a la vez cientos de solicitudes de «elegir regalos para todos los empleados del evento corporativo», descubres de repente que:

• en el servidor hay cientos de conexiones SSE de larga duración;
• los workers envían job.progress por cualquier nimiedad;
• los logs crecen a gigabytes por día;
• la UI del usuario empieza a atascarse, aunque «el servidor no parece caerse».

Una petición HTTP clásica vive milisegundos o segundos. Un flujo SSE o HTTP‑stream puede vivir minutos e incluso horas. Mantiene la conexión, memoria y descriptores de archivo. Cada evento enviado implica serializar JSON, copiar por la red y trabajo del GC. Si lo miras como «pues solo es otro console.log en el backend», el sistema muy rápido se convierte en un calefactor.

Los MCP‑events tienen otra particularidad: a menudo se generan muchas veces para una misma tarea. Un worker que actualiza el progreso cada 0.1% produce un número impresionante de eventos por job. Al final obtienes «ruido»: una enorme cantidad de mensajes pequeños que:

• cargan la red y la CPU;
• saturan colas y buffers;
• hacen que el debug y el análisis de logs sean dolorosos.

Por eso, tanto a los flujos como a los MCP‑events hay que tratarlos tan en serio como a las consultas a la base de datos o a las llamadas a un modelo: son recursos costosos que requieren normalización, control y monitorización.

Para manejarlo, ten en mente tres grandes temas:

1. Rate limits: limitamos cuánto y con qué frecuencia nos podemos permitir generar y enviar eventos/flujos.
2. Backpressure: reaccionamos cuando el consumidor no alcanza al productor.
3. Monitorización y métricas: medimos qué ocurre y detectamos a tiempo cuándo todo empieza a hervir.

2. Rate limiting de flujos y eventos

Empecemos por lo más evidente: los límites.

Es importante entender que, en escenarios de streaming, el «infractor peligroso» suele ser el servidor, no el cliente. En APIs REST normales limitas la cantidad de peticiones al servidor para que el usuario no se convierta en un DDoS. En el mundo MCP y de flujos es muy fácil montar un DDoS inverso: un worker o el servidor MCP bombardea al cliente con miles de eventos por segundo.

Qué límites hacen falta

Suele pensarse en tres ejes.

En primer lugar, límites por usuario o sesión. No puedes permitir que un usuario abra veinte widgets maestro de GiftGenius en paralelo, cada uno con su propio flujo SSE. Un límite razonable es varios flujos activos por sesión y un límite al número de jobs en estado running para un usuario o inquilino.

En segundo lugar, límites por job. Aquí nos interesa la frecuencia de eventos. Basta con enviar job.progress no más de una vez cada N milisegundos o solo cuando haya un cambio apreciable, por ejemplo cada 5% de progreso. No hace falta mandar un mensaje por cada producto procesado del catálogo. También tiene sentido limitar el tamaño del payload: un evento de progreso no debería llevar megabytes de texto.

En tercer lugar, límites por IP u organización. Esto ya es protección contra abusos, cuando alguien lanza un script que hace spam de tareas, o cuando tu App se vuelve inesperadamente popular. Aquí entran en juego mecanismos conocidos de API gateways y proxies.

Implementación sencilla de un límite de frecuencia de eventos

Consideremos un worker de GiftGenius que en segundo plano selecciona regalos para una lista larga de destinatarios y envía periódicamente el progreso mediante una notificación MCP event/progress. Queremos que los eventos se envíen como máximo una vez cada 500 milisegundos y solo cuando el porcentaje cambie al menos 5 puntos.

Pseudocódigo de TS para el worker:

// supongamos que hay 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;

  // enviamos solo si han pasado >= 500 ms O ha aumentado >= un 5%
  if (percentDelta >= 5 || timeDelta >= 500) {
    mcpClient.sendNotification("event/progress", {
      jobId,
      percent,
      message,
    });
    lastSentPercent = percent;
    lastSentAt = now;
  }
}

Este enfoque se llama throttling: «aclaremos» el flujo de eventos en función del tiempo y del cambio de valor.

Si divides en etapas («Etapa 1 de 3», «Etapa 2 de 3»), la lógica es aún más simple: enviar eventos solo cuando cambie la etapa.

Límite al número de flujos abiertos simultáneamente

En el lado del servidor MCP probablemente tengas un manejador HTTP de 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 función canOpenMoreStreams puede comprobar el número actual de conexiones abiertas para el usuario y compararlo con un umbral (por ejemplo, no más de tres flujos en paralelo). Si se supera el límite, devolvemos 429 y en las instrucciones para GPT explicamos al modelo que, en tal situación, mejor no iniciar otro asistente largo, sino sugerir al usuario que «ya hay una búsqueda activa, esperemos a que termine».

En sistemas pequeños, comprobaciones de este tipo pueden implementarse en memoria del proceso. En una infraestructura más seria, esto se lleva a un MCP gateway o a un servicio de rate limiting separado.

3. Backpressure: qué hacer cuando el consumidor no alcanza

Los rate limits restringen cuánto queremos producir eventos. Pero incluso con límites prudentes puede darse la situación en la que el consumidor «se ahogue»: el usuario tiene mala conexión móvil, la pestaña del navegador se congeló, o ChatGPT está muy cargado en ese momento.

El backpressure es la reacción del sistema cuando el consumidor no llega. En lugar de acumular datos infinitamente y terminar cayendo con un OOM, conscientemente:

• reducimos la velocidad;
• agregamos eventos;
• descartamos los menos importantes.

Dónde aparece la presión

Un escenario típico para GiftGenius puede ser así. El worker escribe eventos en una cola (por ejemplo, Redis Streams o simplemente una tabla en la BD), el servidor MCP los lee y los envía por un canal SSE. Si el cliente es lento (3G, portátil antiguo, muchas otras pestañas), el buffer TCP empieza a llenarse, el proceso de Node no consigue vaciar por completo la cola y acaba acumulando eventos en memoria. Después ves lo de siempre:

FATAL ERROR: Ineffective mark-compacts near heap limit

Backpressure a nivel de red (TCP) ya tienes, pero no conoce tus entidades de dominio. Simplemente dice: «Oye, frena, el buffer está lleno». Nuestra tarea es interpretarlo al nivel de eventos MCP.

Bufferización con límite y descarte de eventos

Para progreso y estados tenemos una ventaja: no todos los eventos valen lo mismo. Al usuario le importa el último porcentaje actualizado, no el histórico de todos los intermedios «51%, 52%, 53%, 54%». Esto significa que podemos descartar parte de los eventos y enviar solo el último.

Supongamos que tenemos una capa que recibe eventos de progreso de los workers y los coloca en un buffer para cada 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);

  // limitamos el tamaño del buffer
  if (buffer.length > MAX_BUFFER) {
    // dejamos solo los últimos varios eventos
    progressBuffers.set(event.jobId, buffer.slice(-MAX_BUFFER));
  } else {
    progressBuffers.set(event.jobId, buffer);
  }
}

Un temporizador independiente, por ejemplo cada 500 ms, mira el buffer y envía solo el último evento, ignorando los demás:

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, []); // limpiamos
  }
}, 500);

Este es un ejemplo de táctica de conflation: unir varias actualizaciones en una sola actual. Para progreso — un patrón de oro.

Para eventos del tipo «log» o partial_result la estrategia puede ser distinta. Ahí la pérdida de eventos suele ser inaceptable: el texto de logs es importante, y un JSON chunk perdido puede romper la estructura de datos. En esos casos puedes:

• agregar el mensaje (unir varias líneas de logs en un solo paquete);
• o enviar una señal de control al worker para «frenar la generación de logs».

En sistemas asíncronos el segundo enfoque es más complejo, pero al menos conviene considerarlo.

Límite de profundidad de colas

El backpressure no se limita al buffer de eventos justo antes del envío. Hay que observar todas las colas del sistema:

• la cola de tareas en espera de un worker;
• la cola de eventos entre el worker y el servidor MCP;
• los buffers dentro de las bibliotecas de streaming en el lado del servidor.

Para cada cola es importante fijar un límite razonable de profundidad. Si la cola se desborda, o bien empiezas a responder a los clientes «el sistema está saturado, inténtelo más tarde», o descartas los jobs menos importantes, o pasas parte de los escenarios a «modo offline» (por ejemplo, generas un informe y envías un enlace más tarde).

Otro truco interesante es priorizar tipos de eventos. Bajo sobrecarga puedes empezar a enviar solo job.completed y job.failed, y bajar de prioridad o desactivar por completo job.progress.

4. Monitorización de flujos y eventos

Sin mediciones, toda esta maravilla de rate limits y backpressure se convierte en brujería. Hay que ver que hay sospechosamente muchos flujos, que los eventos llegan con lag y que los clientes se caen en masa.

Los flujos se comportan distinto que las peticiones HTTP normales: su duración puede contarse en minutos y horas, por lo que las métricas clásicas de «peticiones por segundo» y «latencia media» no dan el cuadro completo.

Métricas clave

Para flujos SSE o HTTP/stream es útil seguir varios grupos de indicadores.

1. Métricas de conexiones. ¿Cuántos flujos SSE activos hay ahora? ¿Cuánto vive de media una conexión? ¿Qué porcentaje de flujos termina con error o timeout? Un pico brusco de conexiones activas habla de una potencial tormenta de tráfico o fuga de recursos (los clientes no cierran conexiones). Una caída brusca — de cortes masivos (por ejemplo, problemas de red o un bug crítico en el servidor).
2. Métricas de eventos. ¿Cuántos eventos envías por segundo en todos los flujos (EPS — events per second, básicamente número de eventos por segundo)? ¿Cuál es el tamaño medio del evento? ¿Cuántos errores de deserialización o validación del payload observas? Si de repente ves crecer el tamaño de los eventos, quizá alguien empezó a enviar en job.progress el texto completo de un informe en lugar de una cadena corta.
3. Métricas de jobs. Distribución por estados (pending, running, completed, failed, canceled), tiempo medio de ejecución por tipo de tarea, porcentaje de jobs que entran en retry o en dead‑letter. Esto ayuda a entender que los problemas no solo están en el nivel de red, sino también en los workers: un API externo se ha vuelto más lento, han aparecido errores masivos.
4. Métricas de backpressure e indicadores del sistema. En sistemas de streaming se suele observar la profundidad de buffers y colas entre componentes, así como el porcentaje de tiempo en que el flujo está bloqueado esperando a que el consumidor libere espacio. Si tus colas están casi siempre llenas hasta arriba, es una señal clara de que el sistema está al límite. También es importante seguir indicadores del sistema: CPU y memoria en los servidores que se ocupan del streaming, y errores/timeouts a nivel de red. A veces el cuello de botella es precisamente el ancho de banda entre el servidor MCP y ChatGPT.

En conjunto, estos cuatro grupos te dan respuesta a tres preguntas: cuántos flujos viven ahora, cuántos datos mueves, cómo se comportan los jobs y dónde exactamente empieza a ahogarse el sistema.

Qué registrar en logs

Los logs son el segundo pilar de la observabilidad. Es importante registrar eventos y conexiones de forma que luego pueda reconstruirse la historia de un job concreto.

Normalmente en los logs para cada evento y flujo se añade:

• jobId y/o eventId;
• userId y sessionId (si hay multiinquilinato);
• tipo de evento (progress, completed, failed, resource.updated);
• tipo de canal (SSE o HTTP/stream);
• timestamp de envío y, si es posible, timestamp de nacimiento del evento en el worker.

Así puedes calcular el lag: la diferencia entre el momento en que el worker generó el evento y el momento en que salió por el socket. El crecimiento de este tiempo de lag es un buen indicador de problemas con backpressure.

Hay que ser cuidadoso para que los logs no se conviertan en fuente de sobrecarga. Para eventos de alta frecuencia como job.progress no siempre es razonable registrar cada evento; se puede activar sampling — registrar cada N‑ésimo evento en lugar de todos — o agregar estadísticas.

En código puede verse como un helper sencillo:

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

En un proyecto real lo envolverás con una librería de structured logging, pero la idea es la misma: máximo contexto útil en cada entrada.

5. Alertas y políticas de degradación

Cuando ya tienes métricas y logs, el siguiente paso es configurar alertas y pensar cómo debe «degradar» el sistema cuando está mal. La idea es que es mejor funcionar peor con honestidad que caer de golpe.

Ejemplos de alertas

Para GiftGenius tiene sentido vigilar varias situaciones típicas.

Primero, cantidad anómala de flujos activos. Si normalmente tienes decenas de conexiones SSE activas y de repente pasan a ser miles, conviene saber qué pasa. Puede que te hayas vuelto popular, o puede que haya un bug y las conexiones no se cierren.

Segundo, retraso entre la finalización real del job y la recepción de job.completed en el cliente. Si ese retraso supera un umbral (digamos, 5–10 segundos), significa que en algún lugar entre el worker y el cliente se están acumulando eventos o las conexiones patinan.

Tercero, alta proporción de job.failed o job.canceled respecto a las exitosas. La causa puede estar en el worker (API externo roto, bug nuevo) o en mayor sensibilidad del usuario a los retrasos (empiezan a cancelar tareas con más frecuencia).

Por último, mayor nivel de errores de conexión y rupturas del stream: si crece el número de disconnects no estándar, quizá haya problemas de red o del lado del cliente, y conviene pensar en escenarios de fallback.

Patrones de degradación

Cuando el sistema está sobrecargado, puedes activar un «modo de ahorro de recursos». Esto es mejor que responder 500 a todo.

El patrón más común es la frecuencia adaptativa de eventos. Si ves que el event‑rate (número de eventos por segundo) se ha disparado diez veces por encima de lo normal y el lag empieza a crecer en las colas, reduce la frecuencia de eventos de progreso. Si antes era cada 1%, pásalo a cada 10%. Si antes era cada 500 ms, pásalo a una vez cada 2–3 segundos. El usuario puede vivir perfectamente sin un progreso superpreciso, pero con una UI completamente colgada — no tanto.

Para eventos menos importantes — por ejemplo, resource.updated en la actualización en background del feed de productos — se puede desactivar temporalmente el envío mientras el sistema esté bajo carga.

Otro recurso es pasar parte de los escenarios de flujos a polling periódico. Si los canales SSE se caen, el servidor MCP puede enviar al widget un evento del sistema como system.overloaded, y el widget cambiar a la estrategia de «cada N segundos consulto el endpoint REST sobre el estado del job».

6. Pequeño fragmento práctico para GiftGenius

Para unir todo, imaginemos que ya tenemos:

• un MCP‑tool startGiftSearch que crea un job y devuelve jobId;
• un worker que realiza la búsqueda y envía event/progress y event/completed;
• un endpoint SSE /api/events/[userId] al que se conecta el widget en Next.js.

Añadamos una capa simple de protección contra la «tormenta de eventos» y una monitorización mínima.

Limitación del progreso por pasos y tiempo

En el worker añadimos throttling y conflation, como comentamos arriba. Ahora los eventos se envían como máximo una vez cada medio segundo y cuando el cambio es de al menos 5%.

Conteo de flujos activos

En el endpoint SSE guardamos un contador por usuario:

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

  // aquí guardas el controller en alguna estructura,
  // para escribir después eventos en este flujo
}

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

El servidor puede además enviar métricas sobre activeStreams.size a Prometheus/Grafana o cualquier otro sistema de monitorización.

Métrica más simple de event‑rate

Para empezar, al menos cuenta cuántos eventos envías:

let eventsSentLastMinute = 0;

function sendProgressToClient(ev: ProgressEvent) {
  // ... serialización y escritura en el flujo SSE
  eventsSentLastMinute++;
}

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

Con el tiempo esto se puede sustituir por contadores y alertas «de verdad», pero como punto de partida — ya está bien.

Si juntamos todo lo anterior — límites, backpressure, métricas/alertas y un UX de fallback adecuado — tu GiftGenius deja de ser «una demo para demos» y aguanta tormentas reales de tráfico. En los siguientes módulos, donde hablaremos de gateways, arquitectura de producción y observabilidad completa, estos patrones volverán a ser útiles.

7. Errores típicos al trabajar con flujos, rate limits y monitorización

Error n.º 1: ausencia de límites al número de flujos y a la frecuencia de eventos.
Los desarrolladores añadieron SSE «para que quedara bonito», los workers envían honestamente el progreso por cada objeto procesado y todo parece funcionar en la demo. Pero en el primer pico de usuarios reales el servidor empieza a gastar la mayor parte de los recursos en serializar y enviar miles de eventos minúsculos, y la UI en ChatGPT se convierte en un pase de diapositivas.

Error n.º 2: intentar bufferizar «todo y de golpe» sin límites.
En el código aparece un array sin límite de «eventos no enviados» que crece hasta que el cliente se recupere. Spoiler: no se recupera, el servidor muere antes. Cualquier buffer debe tener un máximo estricto y la lógica de manejo de desbordes — ser explícita.

Error n.º 3: tratar todos los tipos de eventos por igual.
El progreso se puede agregar y descartar (el último porcentaje es más importante que el historial del movimiento). Con los logs y los resultados parciales no se puede hacer eso: perder un chunk puede significar datos dañados. Cuando diseñes el sistema, agrupa de antemano los eventos por importancia y planea para cada grupo una estrategia bajo sobrecarga.

Error n.º 4: falta de observabilidad.
Sin métricas de flujos activos, sin conteo de event‑rate, y en los logs solo «algo salió mal». En esa situación te enteras de los problemas solo por los comentarios de usuarios y la gráfica de carga de CPU. Configurar al menos métricas básicas y logs por jobId y eventId no es un lujo, es una necesidad.

Error n.º 5: UX rígido que no contempla la degradación.
El widget y las instrucciones para GPT parten de que el flujo siempre está disponible, el progreso se actualiza «en tiempo real» y los partial‑results llegan según el guion. Ante los primeros problemas de red el usuario ve una barra de progreso «congelada» y ninguna explicación. Mucho mejor prever un fallback honesto: «Ahora hay problemas con la actualización en vivo; seguiré con la selección y te avisaré cuando termine» — y pasar a actualizaciones más espaciadas o a polling.

Error n.º 6: confiar en que «nuestros usuarios no crearán muchas tareas simultáneas».
La práctica muestra que, si no limitas el número de jobs y flujos en paralelo, alguien abrirá cinco pestañas, lanzará en cada una la selección de regalos «al máximo» y se irá a tomar café. La idea de «a ver si hay suerte» en producción casi siempre acaba con un curso acelerado de monitorización al son de alertas.