En JavaScript asíncrono, las primitivas nativas de concurrencia (Promise.all, Promise.race, Promise.any) carecen de garantías de integridad estructural:
Promise.all([p1, p2, p3]), si la petición p2 es rechazada tras 10 ms, Promise.all falla de inmediato. Sin embargo, p1 y p3 continúan ejecutándose indefinidamente en segundo plano, consumiendo sockets, memoria y ciclos de CPU.Promise.race: Cuando una promesa se resuelve primero, las promesas perdedoras no se cancelan. Sus callbacks continúan procesándose en el Event Loop de Node.js o el navegador.AbortController, lo que en árboles profundos de llamadas provoca fugas de listeners.El ecosistema Effect-TS aborda la ejecución concurrente mediante el paradigma de concurrencia estructurada (Structured Concurrency).
Los cimientos de este modelo son:
Bajo la concurrencia estructurada de Effect, la ejecución se ramifica en un árbol estricto de hilos virtuales (Fiber Tree):
+-------------------------------------------------------------------------+
| FIBRA PADRE (PARENT FIBER) |
| Effect.all([task1, task2, task3], { concurrency: "unbounded" }) |
+-------------------------------------------------------------------------+
| | |
| fork | fork | fork
v v v
+---------------+ +---------------+ +---------------+
| Fibra hija #1 | | Fibra hija #2 | | Fibra hija #3 |
| (I/O de red) | | (FALLO FATAL: | | (Cómputo CPU |
| | | falla a 15ms)| | intensivo) |
+---------------+ +---------------+ +---------------+
| | |
| AUTO-INTERRUPCIÓN | DISPARO DE FALLO | AUTO-INTERRUPCIÓN
v (Interrupt) v v (Interrupt)
+-----------------------------------------------------------------------------+
| PROPAGACIÓN DE LA SEÑAL DE INTERRUPCIÓN: |
| 1. El fallo de la Fibra #2 notifica de inmediato al hilo padre. |
| 2. El padre envía señales Fiber.interrupt a las fibras #1 y #3. |
| 3. Las fibras #1 y #3 se detienen inmediatamente y ejecutan sus |
| finalizadores Scope (cierran sockets, liberan buffers). ¡Cero fugas! |
+-----------------------------------------------------------------------------+Schedule.exponential.concurrency: 4), donde el fallo de un servicio crítico aborta inmediatamente las peticiones restantes.Effect.race: la primera respuesta válida gana y la conexión perdedora se cancela al instante.Effect.fork, join e interrupt#Cualquier efecto puede bifurcarse de forma asíncrona en una fibra dedicada mediante Effect.fork:
import { Effect, Fiber, Duration } from 'effect';
const backgroundTask = Effect.gen(function* () {
yield* Effect.logInfo('Fibra en segundo plano iniciada...');
// Simular tarea prolongada
yield* Effect.sleep(Duration.seconds(10));
yield* Effect.logInfo('Fibra en segundo plano finalizada.');
return 100;
});
export const mainProgram = Effect.gen(function* () {
// 1. Bifurcar tarea en una fibra independiente (Green Thread)
const fiber = yield* Effect.fork(backgroundTask);
// 2. El hilo principal continúa en paralelo
yield* Effect.sleep(Duration.seconds(1));
yield* Effect.logInfo('¡El hilo principal cancela la tarea en segundo plano!');
// 3. Interrupción forzada y segura de la fibra
yield* Fiber.interrupt(fiber);
yield* Effect.logInfo('Fibra en segundo plano detenida limpiamente sin fugas.');
});Schedule#El módulo Schedule aporta una base matemática para definir cualquier cadencia de reintento:
import { Effect, Schedule, Duration, Data, pipe } from 'effect';
export class TransientNetworkError extends Data.TaggedError('TransientNetworkError')<{
readonly statusCode: number;
}> {}
export class FatalAuthError extends Data.TaggedError('FatalAuthError')<{
readonly reason: string;
}> {}
// Llamada de red inestable
let attempt = 0;
const unstableRequest = Effect.gen(function* () {
attempt++;
yield* Effect.logInfo(`Intento de petición de red #${attempt}`);
if (attempt < 3) {
return yield* Effect.fail(new TransientNetworkError({ statusCode: 503 }));
}
return { data: 'Respuesta satisfactoria del gateway' };
});
// Configuración de política de reintentos:
// Pausa inicial de 100ms -> crecimiento exponencial -> máximo 5 reintentos -> jitter aleatorio
const retrySchedule = pipe(
Schedule.exponential(Duration.millis(100)),
Schedule.jittered, // Mitiga Thundering Herd mediante aleatorización (+/- 20%)
Schedule.compose(Schedule.recurs(5))
);
// Aplicar reintentos ESTRICTAMENTE a fallos transitorios de red
export const resilientCall = unstableRequest.pipe(
Effect.retry({
schedule: retrySchedule,
while: (error) => error._tag === 'TransientNetworkError',
})
);Effect.race)#En Effect, el combinador timeout interrumpe automáticamente la fibra activa si se supera el plazo límite:
import { Effect, Duration } from 'effect';
const fastPrimaryService = Effect.sleep(Duration.millis(200)).pipe(
Effect.as('PRIMARY_REGION_DATA')
);
const backupService = Effect.sleep(Duration.millis(500)).pipe(
Effect.as('BACKUP_REGION_DATA')
);
// 1. Competición: la respuesta más veloz gana, la perdedora se cancela de forma garantizada
export const hedgedRequest = Effect.race(fastPrimaryService, backupService);
// 2. Timeout estricto de ejecución
export const timedOperation = hedgedRequest.pipe(
// Si la operación tarda más de 350ms, se cancela arrojando NoSuchElementException
Effect.timeout(Duration.millis(350)),
// Manejo de contingencia
Effect.orElse(() => Effect.succeed('FALLBACK_STALE_CACHE'))
);Para procesar un volumen masivo de 1000 elementos sin agotar sockets de bases de datos, se utiliza la opción concurrency:
import { Effect, Duration } from 'effect';
const itemIds = Array.from({ length: 1000 }, (_, i) => `item_${i}`);
const processItem = (id: string) =>
Effect.gen(function* () {
yield* Effect.logInfo(`Procesando ${id}`);
yield* Effect.sleep(Duration.millis(50));
return `result_${id}`;
});
// Procesamiento en lotes de 1000 elementos limitado a 5 fibras simultáneas
export const throttledBatch = Effect.all(
itemIds.map((id) => processItem(id)),
{ concurrency: 5 } // Límite de fibras virtuales
);Lanzamiento del hilo virtual y registro de temporizadores de timeout en el runtime.
El servicio devuelve 503. Schedule calcula el retardo exponencial con jitter aleatorio.
La fibra duerme el intervalo calculado y reingresa con seguridad a la operación.
Si responde a tiempo, retorna el valor. De lo contrario, interrumpe la fibra y cierra sockets.
A continuación se muestra un módulo para llamadas a servicios externos con reintentos aleatorizados, plazos máximos y contención de fallos:
import { Effect, Schedule, Duration, Data, pipe } from 'effect';
export class TimeoutExceededError extends Data.TaggedError('TimeoutExceededError')<{
readonly serviceName: string;
readonly timeoutMs: number;
}> {}
export class GatewayUnavailableError extends Data.TaggedError('GatewayUnavailableError')<{
readonly serviceName: string;
readonly cause: unknown;
}> {}
export interface GatewayConfig {
readonly serviceName: string;
readonly timeoutMs: number;
readonly maxRetries: number;
readonly initialBackoffMs: number;
}
export function executeWithResilience<A, E>(
action: Effect.Effect<A, E, never>,
config: GatewayConfig
): Effect.Effect<A, E | TimeoutExceededError | GatewayUnavailableError, never> {
// 1. Política de reintentos con jitter exponencial
const retryPolicy = pipe(
Schedule.exponential(Duration.millis(config.initialBackoffMs)),
Schedule.jittered,
Schedule.compose(Schedule.recurs(config.maxRetries))
);
return pipe(
action,
// Aplicar reintentos a fallos transitorios
Effect.retry(retryPolicy),
// Acotar duración total con timeout
Effect.timeout(Duration.millis(config.timeoutMs)),
// Convertir timeout en error de dominio tipado
Effect.mapError((err) => {
if (err instanceof Error && err.name === 'TimeoutException') {
return new TimeoutExceededError({
serviceName: config.serviceName,
timeoutMs: config.timeoutMs,
});
}
return err;
}),
// Capturar defectos no controlados y aislarlos
Effect.catchAllDefect((defect) =>
Effect.fail(
new GatewayUnavailableError({
serviceName: config.serviceName,
cause: defect,
})
)
)
);
}Combinar Promise.all dentro de Effect desintegra la concurrencia estructurada. Si una promesa falla, las demás permanecen huérfanas en la memoria de Node.js.
Solución: utilizar siempre Effect.all(effects, { concurrency: ... }).
Aplicar Effect.retry(Schedule.recurs(5)) sobre fallos de autenticación (401) o validación (400) satura los servidores con tráfico inútil.
Solución: acotar los reintentos mediante el predicado: while: (err) => isTransient(err).
Las fibras de Effect son cooperativas. Un bucle síncrono while(true) {} monopoliza el hilo de JavaScript, impidiendo que otras fibras procesen señales de cancelación.
Solución: ceder periódicamente el control al planificador con yield* Effect.yieldNow().
Diferenciar errores temporales de red (502, 503, 504, ECONNRESET) de errores permanentes (400, 401, 404).
Configurar Schedule.exponential con jitter y un número tope de reintentos (3-5).
Envolver operaciones con Effect.timeout para evitar retenciones indefinidas de sockets.
Fijar { concurrency: N } en llamadas paralelas respetando los límites de los pools receptores.
Validar que la pérdida simulada de conexión interrumpa las fibras y libere los recursos en Scope.