En sistemas frontend de alta demanda y microservicios de Node.js modernos, la gestión de operaciones asíncronas ha superado con creces las cadenas elementales de Promise.all(). En arquitecturas distribuidas de producción, los ingenieros deben resolver desafíos críticos de concurrencia:
El estándar de la plataforma web AbortController / AbortSignal se ha consolidado como el protocolo universal para la cancelación cooperativa de tareas. Combinado con las interfaces AsyncIterable<T> y el patrón Communicating Sequential Processes (CSP / canales estilo Go), TypeScript permite construir tuberías de concurrencia matemáticamente rigurosas y con seguridad de tipos, eliminando el riesgo de excepciones no capturadas o condiciones de carrera.
+-------------------------------------------------------------------------+
| ARQUITECTURA DE TUBERÍA CONCURRENTE |
+-------------------------------------------------------------------------+
| [Producer (SSE / WebSocket / Stream LLM)] |
| |--> Produce flujo continuo: AsyncIterable<ServerPacket> |
+-------------------------------------------------------------------------+
|
v (Backpressure Queue / Channel<T>)
+-------------------------------------------------------------------------+
| [Canal con buffer tipado: Channel<ServerPacket>] |
| |-- push(packet: T): Promise<void> (Se bloquea si el buffer se llena)|
| |-- pop(signal?: AbortSignal): Promise<T> (Se bloquea si está vacío) |
| `-- close(): void |
+-------------------------------------------------------------------------+
|
v (Worker Pool: Concurrency Limit = 3)
+-------------------------------------------------------------------------+
| [Grupo paralelo de consumidores (Consumers / Workers)] |
| |-- Worker 1: Procesa Paquete A (AbortSignal.timeout(5000)) |
| |-- Worker 2: Procesa Paquete B (Interrumpido por abort()) |
| `-- Worker 3: Procesa Paquete C (Finalización exitosa) |
+-------------------------------------------------------------------------+La concurrencia en TypeScript se fundamenta en abstracciones rigurosas de cancelación y paso de mensajes:
Composición de señales de cancelación:
[Señal de expiración: AbortSignal.timeout(5000)] ---\
+--> [AbortSignal.any([...])] --> fetch(url, { signal })
[Señal de botón en UI: userCancelCtrl.signal] -----/
¡La petición se interrumpe si transcurren 5 segundos O si el usuario pulsa «Cancelar»!AsyncIterable<string> con aborto inmediato del pipeline ante la orden «Detener respuesta».concurrencyLimit = 4) para evitar el agotamiento de sockets del sistema operativo.AbortSignal.timeout() y derivación automática a nodos de respaldo (Failover).AbortSignal.any() y .timeout()#En TypeScript moderno (Node.js 20+ y navegadores actuales), combinar tiempos de espera y cancelaciones de usuario es directo gracias a los métodos estáticos de AbortSignal:
export interface FetchWithTimeoutOptions extends RequestInit {
readonly timeoutMs?: number;
readonly externalSignal?: AbortSignal;
}
/**
* Ejecuta una petición HTTP resiliente con cancelación compuesta:
* por tiempo límite O por señal externa explícita.
*/
export async function fetchWithResilience<TResponse>(
url: string,
options: FetchWithTimeoutOptions = {}
): Promise<TResponse> {
const { timeoutMs = 8000, externalSignal, ...fetchOptions } = options;
// 1. Instanciamos la señal de tiempo de espera nativa
const timeoutSignal = AbortSignal.timeout(timeoutMs);
// 2. Combinamos el temporizador con la señal externa mediante AbortSignal.any
const combinedSignal = externalSignal
? AbortSignal.any([timeoutSignal, externalSignal])
: timeoutSignal;
try {
const response = await fetch(url, {
...fetchOptions,
signal: combinedSignal,
});
if (!response.ok) {
throw new Error(`[HttpError] HTTP ${response.status}: ${response.statusText}`);
}
return (await response.json()) as TResponse;
} catch (error: unknown) {
if (combinedSignal.aborted) {
// Discriminamos el origen de la interrupción: timeout vs acción del usuario
if (timeoutSignal.aborted) {
throw new Error(`[TimeoutError] La petición a ${url} excedió el límite de ${timeoutMs}ms`);
}
throw new Error(`[AbortError] La petición a ${url} fue cancelada por la entidad emisora`);
}
throw error instanceof Error ? error : new Error(String(error));
}
}Construcción de un canal para la comunicación entre procesos concurrentes independientes:
interface Resolver<T> {
resolve: (value: T) => void;
reject: (reason?: unknown) => void;
}
/**
* Canal asíncrono con seguridad de tipos basado en Communicating Sequential Processes (CSP).
*/
export class AsyncChannel<T> implements AsyncIterable<T> {
private readonly buffer: T[] = [];
private readonly receivers: Resolver<T>[] = [];
private isClosed = false;
constructor(private readonly maxCapacity: number = Infinity) {}
/**
* Escribe un elemento en el canal.
*/
public push(item: T): boolean {
if (this.isClosed) {
throw new Error('[ChannelError] No es posible escribir en un canal cerrado');
}
// Si existen consumidores esperando, entregamos inmediatamente al primero
if (this.receivers.length > 0) {
const receiver = this.receivers.shift()!;
receiver.resolve(item);
return true;
}
// Control de saturación del búfer
if (this.buffer.length >= this.maxCapacity) {
// Búfer saturado: descartamos el elemento más antiguo (Drop-oldest backpressure)
this.buffer.shift();
}
this.buffer.push(item);
return true;
}
/**
* Lectura asíncrona de un elemento del canal con soporte de cancelación.
*/
public async pop(signal?: AbortSignal): Promise<T | null> {
if (this.buffer.length > 0) {
return this.buffer.shift()!;
}
if (this.isClosed) {
return null;
}
return new Promise<T | null>((resolve, reject) => {
const resolver: Resolver<T> = {
resolve: (val) => {
cleanUp();
resolve(val);
},
reject: (err) => {
cleanUp();
reject(err);
},
};
const onAbort = () => {
const index = this.receivers.indexOf(resolver);
if (index > -1) {
this.receivers.splice(index, 1);
}
cleanUp();
reject(new Error('[AbortError] Espera en el canal cancelada por la señal'));
};
const cleanUp = () => {
signal?.removeEventListener('abort', onAbort);
};
if (signal?.aborted) {
onAbort();
return;
}
signal?.addEventListener('abort', onAbort);
this.receivers.push(resolver);
});
}
public close(): void {
this.isClosed = true;
// Liberamos a todos los receptores en espera notificando el cierre
while (this.receivers.length > 0) {
const receiver = this.receivers.shift()!;
receiver.resolve(null as unknown as T);
}
}
/**
* Implementación del protocolo AsyncIterable para consumo mediante for await (... of ...)
*/
public async *[Symbol.asyncIterator](): AsyncIterator<T> {
while (true) {
const value = await this.pop();
if (value === null && this.isClosed) {
break;
}
if (value !== null) {
yield value;
}
}
}
}export type AsyncTask<T> = (signal: AbortSignal) => Promise<T>;
/**
* Limita el volumen de operaciones asíncronas en ejecución paralela simultánea.
*/
export class ConcurrencyPool {
private activeCount = 0;
private readonly queue: Array<() => void> = [];
constructor(public readonly concurrencyLimit: number) {
if (concurrencyLimit < 1) {
throw new RangeError('El límite de concurrencia debe ser >= 1');
}
}
/**
* Encola y ejecuta una tarea respetando el cupo de concurrencia y la señal de aborto.
*/
public run<T>(task: AsyncTask<T>, signal?: AbortSignal): Promise<T> {
return new Promise<T>((resolve, reject) => {
const execute = async () => {
if (signal?.aborted) {
this.next();
reject(new Error('[AbortError] Tarea abortada antes de comenzar su ejecución'));
return;
}
this.activeCount++;
try {
// Vinculamos un controlador local a la señal externa entrante
const ctrl = new AbortController();
const onAbort = () => ctrl.abort();
signal?.addEventListener('abort', onAbort);
const result = await task(ctrl.signal);
signal?.removeEventListener('abort', onAbort);
resolve(result);
} catch (err: unknown) {
reject(err);
} finally {
this.activeCount--;
this.next();
}
};
if (this.activeCount < this.concurrencyLimit) {
void execute();
} else {
this.queue.push(execute);
}
});
}
private next(): void {
if (this.queue.length > 0 && this.activeCount < this.concurrencyLimit) {
const nextTask = this.queue.shift();
nextTask?.();
}
}
}Las tareas ingresan al grupo. Si activeCount < concurrencyLimit se ejecutan de inmediato; en caso contrario aguardan en cola.
El temporizador y la señal de usuario se combinan vía AbortSignal.any(), propagando la cancelación en toda la cadena.
La tarea se invoca transmitiendo la señal. Al activarse un abort, el descriptor del socket o stream HTTP se cierra de inmediato.
Dentro del bloque finally se decrementa activeCount y el despachador ejecuta la siguiente tarea pendiente en espera.
export interface SseEvent<TData> {
readonly id?: string;
readonly event: string;
readonly data: TData;
readonly timestamp: number;
}
/**
* Lee un flujo Server-Sent Events como un generador asíncrono estrictamente tipado.
*/
export async function* streamSseEvents<TData>(
url: string,
signal?: AbortSignal
): AsyncIterable<SseEvent<TData>> {
const response = await fetch(url, {
headers: {
Accept: 'text/event-stream',
'Cache-Control': 'no-cache',
},
signal,
});
if (!response.ok || !response.body) {
throw new Error(`[SseError] No fue posible establecer conexión: HTTP ${response.status}`);
}
const reader = response.body.getReader();
const decoder = new TextDecoder('utf-8');
let buffer = '';
try {
while (true) {
if (signal?.aborted) {
break;
}
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
const lines = buffer.split('\n\n');
buffer = lines.pop() ?? '';
for (const block of lines) {
if (!block.trim()) continue;
let eventType = 'message';
let eventDataRaw = '';
let eventId: string | undefined;
for (const line of block.split('\n')) {
if (line.startsWith('event:')) {
eventType = line.replace('event:', '').trim();
} else if (line.startsWith('data:')) {
eventDataRaw += line.replace('data:', '').trim();
} else if (line.startsWith('id:')) {
eventId = line.replace('id:', '').trim();
}
}
if (eventDataRaw) {
const parsedData = JSON.parse(eventDataRaw) as TData;
yield {
id: eventId,
event: eventType,
data: parsedData,
timestamp: Date.now(),
};
}
}
}
} finally {
reader.releaseLock();
}
}Recibir signal?: AbortSignal en una rutina asíncrona sin agregar un escucha mediante addEventListener('abort', ...) o sin propagarlo a las API nativas inutiliza la señal por completo: la función continuará ejecutándose hasta el final ignorando la cancelación.
// ❌ ANTIPATRÓN: Se recibe el parámetro signal pero se ignora por completo
async function sleepBad(ms: number, signal?: AbortSignal) {
return new Promise((resolve) => setTimeout(resolve, ms)); // ¡Señal ignorada!
}
// ✅ CORRECTO: Suscripción al aborto y descarte del temporizador
async function sleepGood(ms: number, signal?: AbortSignal) {
return new Promise((resolve, reject) => {
if (signal?.aborted) return reject(new Error('Aborted'));
const id = setTimeout(resolve, ms);
signal?.addEventListener('abort', () => {
clearTimeout(id);
reject(new Error('Aborted'));
}, { once: true });
});
}Registrar escuchas en instancias de AbortSignal con ciclo de vida prolongado sin eliminarlos con removeEventListener ni utilizar la opción { once: true } retiene cierres léxicos y objetos en memoria a lo largo de miles de operaciones.
// ❌ ANTIPATRÓN: Fuga de memoria por acumulación de callbacks
signal.addEventListener('abort', onAbort); // ¡FUGA DE MEMORIA!
// ✅ CORRECTO: Utilizar la opción once: true o limpieza explícita
signal.addEventListener('abort', onAbort, { once: true });Cuando una petición fetch nativa es interrumpida, arroja una excepción con error.name === 'AbortError'. Si la arquitectura no clasifica este evento como un flujo previsto, las plataformas de telemetría se saturarán de alertas falsas.
// ✅ CORRECTO: Discriminación entre cancelación ordinaria y caída de red
try {
await fetch(url, { signal });
} catch (err: unknown) {
if (err instanceof Error && err.name === 'AbortError') {
// Interrupción deliberada: ignorar o registrar en nivel debug
return;
}
console.error('Error de red genuino:', err);
}Si un canal asíncrono o generador no evalúa sus criterios de finalización ni invoca a close(), el bucle for await que lo consume quedará bloqueado indefinidamente a la espera de nuevos paquetes.
// ✅ CORRECTO: Invocación obligatoria de channel.close() dentro de bloques finallyAnalice los límites de tasa de las API externas y asigne un valor acotado a concurrencyLimit (típicamente de 4 a 8).
Implemente AbortSignal.timeout() contra cuelgues de red y vincúlelo a la señal de interfaz mediante AbortSignal.any().
Instancie AsyncChannel<T> con un búfer dimensionado y aplique contrapresión (Backpressure) por descarte o suspensión.
Procese los mensajes a través de ConcurrencyPool, garantizando la transmisión de la señal de aborto a cada tarea.
Verifique en bloques finally la invocación de reader.releaseLock(), channel.close() y la eliminación de suscripciones.