В асинхронном JavaScript встроенные примитивы конкурентности (Promise.all, Promise.race, Promise.any) не обладают свойством структурной целостности:
Promise.all([p1, p2, p3]) запрос p2 падает с ошибкой через 10 мс, Promise.all немедленно отклоняется (reject). Однако задачи p1 и p3 продолжают выполняться в фоне до своего завершения, продолжая расходовать сокеты, память и процессорное время.Promise.race: Когда один из промисов разрешается первым, проигравшие промисы не прерываются. Их обработчики продолжают исполняться в Event Loop браузера или Node.js.AbortController, что в глубоких деревьях вызовов приводит к ошибкам забытых слушателей.Экосистема Effect-TS решает задачу конкурентности через концепцию структурной конкурентности (Structured Concurrency).
Фундаментом этой модели служат:
В структурной конкурентности Effect вычисления ветвятся в виде строгого дерева виртуальных потоков (Fiber Tree):
+-------------------------------------------------------------------------+
| РОДИТЕЛЬСКОЕ ВОЛОКНО (PARENT FIBER) |
| Effect.all([task1, task2, task3], { concurrency: "unbounded" }) |
+-------------------------------------------------------------------------+
| | |
| fork | fork | fork
v v v
+---------------+ +---------------+ +---------------+
| Дочернее #1 | | Дочернее #2 | | Дочернее #3 |
| (Сетевой I/O) | | (ФАТАЛЬНЫЙ | | (Тяжелый CPU |
| | | СБОЙ: 15ms) | | расчет) |
+---------------+ +---------------+ +---------------+
| | |
| АВТО-ПРЕРЫВАНИЕ | ТРИГГЕР СБОЯ | АВТО-ПРЕРЫВАНИЕ
v (Interrupt) v v (Interrupt)
+-----------------------------------------------------------------------------+
| РАСПРОСТРАНЕНИЕ СИГНАЛА ПРЕРЫВАНИЯ: |
| 1. Сбой Волокна #2 немедленно уведомляет родительский поток. |
| 2. Родитель посылает Fiber.interrupt волокнам #1 и #3. |
| 3. Волокна #1 и #3 немедленно останавливаются, выполняя свои финализаторы |
| Scope (закрывают сокеты, сбрасывают буферы). Никаких фоновых утечек! |
+-----------------------------------------------------------------------------+Выделить ошибки сети (502, 503, 504, ECONNRESET), отделив их от бизнес-сбоев (400, 401, 404).
Сконфигурировать Schedule.exponential с джиттером и максимальным числом попыток (3-5).
Обернуть операцию в Effect.timeout для предотвращения вечных подвисаний сокетов.
При параллельных вызовах задать строгий лимит { concurrency: N } исходя из емкости внешнего пула.
Проверить сценарии обрыва сети и убедиться, что отмененные волокна закрывают ресурсы.
Что произойдет с параллельными операциями при падении одной из них в Effect.all(effects, { concurrency: 5 }), в отличие от нативного Promise.all?
Schedule.exponential.concurrency: 4), где сбой одного критического сервиса немедленно отменяет остальные запросы.Effect.race: первый ответивший регион побеждает, второй сокет немедленно прерывается.Effect.fork, join и interrupt#Любой эффект может быть запущен асинхронно в отдельном волокне через метод Effect.fork:
import { Effect, Fiber, Duration } from 'effect';
const backgroundTask = Effect.gen(function* () {
yield* Effect.logInfo('Фоновое волокно запущено...');
// Имитация длительной операции
yield* Effect.sleep(Duration.seconds(10));
yield* Effect.logInfo('Фоновое волокно завершено.');
return 100;
});
export const mainProgram = Effect.gen(function* () {
// 1. Ответвление задачи в отдельное волокно (Green Thread)
const fiber = yield* Effect.fork(backgroundTask);
// 2. Основной поток работает параллельно
yield* Effect.sleep(Duration.seconds(1));
yield* Effect.logInfo('Основной поток решает прервать фоновую задачу!');
// 3. Безопасное принудительное прерывание волокна
yield* Fiber.interrupt(fiber);
yield* Effect.logInfo('Фоновое волокно успешно остановлено без утечек.');
});Schedule#Модуль Schedule предоставляет математический аппарат для моделирования любых интервалов повторов:
import { Effect, Schedule, Duration, Data } from 'effect';
export class TransientNetworkError extends Data.TaggedError('TransientNetworkError')<{
readonly statusCode: number;
}> {}
export class FatalAuthError extends Data.TaggedError('FatalAuthError')<{
readonly reason: string;
}> {}
// Нестабильный запрос к удаленному шлюзу
let attempt = 0;
const unstableRequest = Effect.gen(function* () {
attempt++;
yield* Effect.logInfo(`Попытка сетевого запроса #${attempt}`);
if (attempt < 3) {
return yield* Effect.fail(new TransientNetworkError({ statusCode: 503 }));
}
return { data: 'Успешный ответ шлюза' };
});
// Конфигурация расписания повторов:
// Начальная пауза 100мс -> экспоненциальный рост до 2с -> максимум 5 попыток -> добавление джиттера
const retrySchedule = pipe(
Schedule.exponential(Duration.millis(100)),
Schedule.jittered, // Защита от Thundering Herd: рандомизация дельты времени (+- 20%)
Schedule.compose(Schedule.recurs(5))
);
// Применение политики повторов ТОЛЬКО для временных ошибок сети
export const resilientCall = unstableRequest.pipe(
Effect.retry({
schedule: retrySchedule,
while: (error) => error._tag === 'TransientNetworkError',
})
);Effect.race)#В Effect комбинатор timeout автоматически прерывает исполняемое волокно при истечении лимита времени:
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. Гонка вычислений: быстрое вычисление побеждает, медленное гарантированно прерывается
export const hedgedRequest = Effect.race(fastPrimaryService, backupService);
// 2. Жесткий таймаут выполнения
export const timedOperation = hedgedRequest.pipe(
// Если операция длится дольше 350мс, она прерывается с ошибкой NoSuchElementException
Effect.timeout(Duration.millis(350)),
// Обработка таймаута
Effect.orElse(() => Effect.succeed('FALLBACK_STALE_CACHE'))
);При необходимости обработать массив из 10 000 элементов без переполнения сокетов базы данных используется опция concurrency:
import { Effect } from 'effect';
const itemIds = Array.from({ length: 1000 }, (_, i) => `item_${i}`);
const processItem = (id: string) =>
Effect.gen(function* () {
yield* Effect.logInfo(`Обработка ${id}`);
yield* Effect.sleep(Duration.millis(50));
return `result_${id}`;
});
// Обработка 1000 элементов пачками строго по 5 параллельных потоков
export const throttledBatch = Effect.all(
itemIds.map((id) => processItem(id)),
{ concurrency: 5 } // Лимит виртуальных волокон
);Запуск виртуального потока с регистрацией таймера таймаута в контексте рантайма.
Сервер вернул 503. Schedule вычисляет экспоненциальную паузу со случайным джиттером.
Волокно засыпает на вычисленную дельту и выполняет повторный вход в ту же операцию.
Если ответ получен до истечения таймаута — возврат результата. Иначе — принудительное закрытие сокета.
Ниже представлен production-модуль отказоустойчивого вызова внешних сервисов с автоматическими повторами, таймаутом и изоляцией сбоев:
import { Effect, Schedule, Duration, Data } 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. Политика повторных попыток с экспоненциальным джиттером
const retryPolicy = pipe(
Schedule.exponential(Duration.millis(config.initialBackoffMs)),
Schedule.jittered,
Schedule.compose(Schedule.recurs(config.maxRetries))
);
return pipe(
action,
// Применяем политику повторов к временным сбоям
Effect.retry(retryPolicy),
// Ограничиваем суммарное время исполнения таймаутом
Effect.timeout(Duration.millis(config.timeoutMs)),
// Преобразуем таймаут в доменную типизированную ошибку
Effect.mapError((err) => {
if (err instanceof Error && err.name === 'TimeoutException') {
return new TimeoutExceededError({
serviceName: config.serviceName,
timeoutMs: config.timeoutMs,
});
}
return err;
}),
// Перехватываем фатальные дефекты и оборачиваем в контролируемый тип
Effect.catchAllDefect((defect) =>
Effect.fail(
new GatewayUnavailableError({
serviceName: config.serviceName,
cause: defect,
})
)
)
);
}Зачем при проектировании политики повторов (Retry Policy) для высоконагруженных внешних API обязательно использовать джиттер (Jitter)?
Смешивание Promise.all с Effect разрушает структурную конкурентность. Если один из промисов внутри Promise.all упадет, остальные промисы продолжат бесконтрольно висеть в памяти Node.js.
Решение: всегда использовать Effect.all(effects, { concurrency: ... }).
Слепое применение Effect.retry(Schedule.recurs(5)) к запросам аутентификации (401 Unauthorized) или валидации данных (400 Bad Request) лишь перегружает сервер бессмысленным трафиком.
Решение: всегда ограничивать условие повторов предикатом while: (err) => isTransient(err).
Волокна Effect являются кооперативными. Если запустить в волокне тяжелый синхронный цикл вычислений (while(true) {}), он заблокирует весь поток JavaScript, не давая другим волокнам даже обработать сигнал прерывания.
Решение: периодически отдавать квант времени планировщику через вызов yield* Effect.yieldNow().