Управление конкуренцией: комбинаторы, ограниченный fan-out и отмена

Четыре комбинатора и семантика, которую путают

Прежде чем тянуться за Promise.all, спроси себя: если одна из задач упадёт, я хочу, чтобы упал весь батч? Ответ определяет, какой комбинатор правильный — и ошибка здесь либо скрывает частичные сбои, либо выбрасывает результаты, которые уже получены.

Четыре статических комбинатора Promise выглядят взаимозаменяемыми, пока сбой не вскроет, насколько по-разному они оседают. Выбери не тот — и ты либо упадёшь на частичном сбое, который мог бы стерпеть, либо тихо сохранишь результаты работы, которую уже бросил.

const settle = (ms, ok = true, v) =>
  new Promise((res, rej) => setTimeout(() => (ok ? res(v) : rej(new Error(v))), ms));

// Promise.all — резолвится в упорядоченный массив ИЛИ реджектится на ПЕРВОМ реджекте.
await Promise.all([settle(10, true, "a"), settle(20, true, "b")]);    // ["a","b"]
await Promise.all([settle(10, false, "x"), settle(30, true, "b")]);   // throws Error("x") at 10ms

// Promise.allSettled — НИКОГДА не реджектится. Один дескриптор на вход, по порядку.
await Promise.allSettled([settle(10, true, "a"), settle(20, false, "x")]);
// [{ status: "fulfilled", value: "a" }, { status: "rejected", reason: Error("x") }]

// Promise.race — оседает ПЕРВЫМ осевшим, победа ИЛИ поражение.
await Promise.race([settle(50, true, "slow"), settle(10, false, "fast-fail")]); // throws at 10ms

// Promise.any — первое ВЫПОЛНЕНИЕ; если ВСЕ реджектнутся — бросает AggregateError(.errors[])
await Promise.any([settle(10, false, "x"), settle(20, true, "b")]);   // "b"
await Promise.any([settle(10, false, "x"), settle(20, false, "y")]);  // AggregateError

Механизм, который тебя кусает, умещается в одно предложение: реджект Promise.all не отменяет остальные промисы. Промис — это не задача, которую можно убить; это одностороннее уведомление о том, что некая уже идущая работа завершилась. Когда Promise.all видит первый реджект, он немедленно реджектит свой собственный промис, но остальные fetch’и, запросы и таймеры продолжают бежать до конца в фоне — ты просто перестал слушать. Так что Promise.all над 50 вызовами БД, где вызов №3 реджектится на 10 мс, всё равно держит все 50 соединений открытыми, пока не вернётся самый медленный из оставшихся 47. Используй all, когда частичный сбой недопустим (нужна вся пачка или ничего), и allSettled, когда частичный сбой ожидаем и нужно изучить каждый исход.

Ловушка неограниченного fan-out

Вот строка, из-за которой поднимают людей по тревоге, и та, что выглядит как фикс, но им не является:

// АНТИПАТТЕРН A — неограниченный fan-out: 48 000 запросов одновременно
await Promise.all(ids.map(id => fetchFromDb(id)));
// .map синхронно запускает 48 000 thunk'ов → 48 000 pending-промисов → 48 000
// запросов на соединение давят пул из 20. Пул иссякает, очередь переполняется,
// ETIMEDOUT — и вы устроили DoS собственной БД. Память тоже скачет:
// 48 000 in-flight объектов запросов + их буферы.

// АНТИПАТТЕРН B — полностью последовательный: корректный, но черепаший
for (const id of ids) await fetchFromDb(id);
// По одному. 48 000 × 40 мс = 1920 с ≈ 32 минуты. Вы обменяли крэш на задачу,
// которая не укладывается в окно. Пул теперь простаивает на 95%.

Обе неправильны, и неправильны в противоположных направлениях: A просит бесконечный параллелизм, B не просит никакого. Сеньорский ответ — ограниченная конкуренция: держи ровно N операций в полёте в любой момент, где N подобрано под пропускную способность нижестоящего звена (пул на 20 соединений хочет N ≈ 10–16, оставляя запас другим вызывающим). Пропускная способность тогда примерно N / latency: при вызовах по 40 мс N=1 финиширует за ~32 мин, N=16 — за 48000 × 40ms / 16 ≈ 120с, а N=∞ «финиширует» падением базы. Весь навык — выбирать N осознанно, а не давать .map выбрать ∞ за тебя.

Ограниченный пул воркеров с нуля

Библиотека для этого не нужна, а написав его однажды, цементируешь механизм. Паттерн: запусти N «воркеров», каждый из которых тянет следующую задачу из общего курсора и крутится, пока очередь не опустеет. Задачи — это thunk’и, () => Promise, а не уже запущенные промисы, потому что вся суть — управлять тем, когда каждый из них стартует.

async function runPool(tasks, limit) {
  const results = new Array(tasks.length);
  let next = 0; // shared cursor; ++ is atomic between awaits (single-threaded)

  async function worker() {
    while (next < tasks.length) {
      const i = next++;            // claim an index
      results[i] = await tasks[i]();// run one, then loop for the next
    }
  }

  // Запускаем ровно `limit` воркеров; никогда больше `limit` задач не в полёте.
  const workers = Array.from({ length: Math.min(limit, tasks.length) }, worker);
  await Promise.all(workers);      // all workers drain the queue
  return results;                  // results stay aligned with input order
}

// Использование: не более 16 вызовов БД в полёте в любой момент, сколько бы ни было ids.
const thunks = ids.map(id => () => fetchFromDb(id));
const rows = await runPool(thunks, 16);

Две сеньорские детали. Первая: в таком виде это небезопасно по части allSettled — если какой-то tasks[i]() реджектится, await внутри worker бросает, этот воркер умирает, и внешний Promise.all(workers) реджектится (а остальные воркеры продолжают вычерпывать очередь в фоне — то же правило неотменяемости, что и выше). Если частичный сбой допустим, оберни вызов: results[i] = await tasks[i]().then(v => ({ ok: true, v }), e => ({ ok: false, e })). Вторая: в проде бери проверенную в бою версию — p-limit (оборачивай каждый вызов: limit(() => fetchFromDb(id))) или Promise.map-с-конкуренцией из bluebird/p-map. Они добавляют тот же лимит плюс удобства (per-call AbortSignal, агрегацию ошибок), но движок — ровно тот цикл выше.

Отмена, которая реально отменяет: AbortController

Ограниченный пул не даёт тебе запустить слишком много. AbortController позволяет остановить работу, уже идущую в полёте — то, чего Promise.race не может. Promise.race([work, timeout]) оседает, когда побеждает таймаут, но work продолжает бежать, продолжает держать своё соединение и может позже реджектнуться, когда уже никто не слушает (необработанный реджект). Это утечка проигравшего. AbortController вместо этого сигналит нижележащей операции реально прерваться.

// ПРОТЕКАЮЩИЙ таймаут: race оседает на 2 с, но fetch идёт до своего дефолта 30+ с,
// удерживает сокет, и если позже реджектится → unhandledRejection.
const timeout = new Promise((_, rej) => setTimeout(() => rej(new Error("timeout")), 2000));
await Promise.race([fetch(url), timeout]); // fetch НЕ отменяется

// НАСТОЯЩАЯ отмена: сигнал рвёт сокет на 2 с.
const res = await fetch(url, { signal: AbortSignal.timeout(2000) }); // бросает TimeoutError; сокет закрыт

// Ручное управление + объединение сигналов. AbortSignal.any прерывает, когда ЛЮБОЙ вход делает это.
const ac = new AbortController();
const signal = AbortSignal.any([ac.signal, AbortSignal.timeout(5000)]);
signal.addEventListener("abort", () => cleanup(), { once: true });
const data = await fetch(url, { signal }); // прерывается на ac.abort() ИЛИ через 5 с
// ...позже, например при отключении клиента: ac.abort(new Error("client gone"));

Числа важны для выживания под нагрузкой. У fetch/undici в Node по умолчанию нет общего таймаута — застрявший upstream может удерживать запрос десятки секунд (а дефолты уровня соединения в районе 5–10 с, а не лимит на запрос), и с протекающим race эти сокеты копятся, пока ты не исчерпаешь пул — это снова давка на соединения, но с исходящей стороны. AbortSignal.timeout(2000) ограничивает цену одного медленного upstream’а 2 секундами и немедленно освобождает сокет, так что бюджет в 2 с вместо зависания на 30+ с — это 15-кратное снижение worst-case-занятости на вызов. Протяни один и тот же сигнал в каждую await-операцию запроса — fetch, чтения fs, задачи твоего пула — чтобы одна отмена раскрутила всё дерево.