Backpressure и ограниченная конкурентность

Скрипт миграции читает миллион ID пользователей и «для каждого достань профиль и сделай upsert». Написанный очевидным async-способом — await Promise.all(ids.map(processUser)) — он запускает миллион конкурентных операций в один и тот же миг. База открывает соединения, пока не откажет, память лезет вверх, пока миллион ожидающих промисов и их замыканий копятся, и процесс убивают по OOM раньше, чем закончится первая тысяча. Код был корректен. Конкурентность была неограниченной, а event loop радостно запустит бесконечную работу, которую никогда не сможет закончить.

Цикл тебя не остановит

Дар event loop — запустить тысячи операций дёшево — это и заряженное ружьё. Ничто в Promise.all(items.map(fn)) не ограничивает, сколько fn в полёте; он запускает их все, немедленно. Для десяти элементов это нормально. Для миллиона это самонанесённый отказ в обслуживании: каждая операция в полёте держит память (буферы, замыкания, ожидающий промис), и каждая молотит то, что вызывает. Две разные дисциплины укрощают это: backpressure (дать медленному консьюмеру толкнуть назад быстрого продюсера) и ограниченная конкурентность (ограничить, сколько операций бежит разом). Они лечат одну болезнь — производство быстрее, чем потребление — на разных слоях.

Backpressure: консьюмер толкает назад

Backpressure — это сигнал обратной связи от медленного консьюмера быстрому продюсеру: «перестань слать, я полон». В Node-стримах он встроен. Когда ты writable.write(chunk) и внутренний буфер превышает highWaterMark (по умолчанию 16 КБ для байтовых стримов, 16 объектов в object mode), write() возвращает false — твой сигнал перестать писать, пока стрим не испустит событие drain, означающее, что буфер опустел. Уважь сигнал — и память остаётся ограниченной; проигнорируй — и Node продолжит буферить каждый chunk в памяти, пока процесс не словит OOM.

Причина, почему pipe() и pipeline() — рекомендуемый способ соединять стримы, в том, что они проводят это рукопожатие автоматически — паузя readable, когда writable полон, возобновляя на drain — поэтому файл в 50 ГБ копируется через несколько сотен КБ живого буфера вместо загрузки целиком. Ручные циклы read/write, пропускающие танец false/drain, — классический источник OOM при стриминге.

Ограниченная конкурентность: ограничь fan-out

Стримы решают случай байтового пайплайна. Другой случай — N независимых async-задач — достань эти 10 000 URL, обработай этот миллион строк — где фикс это лимит конкурентности: бежать максимум k разом, и как только каждая заканчивается, запускать следующую. Миграция выше становится безопасной заменой Promise.all(ids.map(fn)) на лимитер (например p-limit(20)), чтобы в полёте было только 20 операций, или паттерном worker-pool/for await, который тянет из списка по мере освобождения ёмкости.

Числа доказывают: последовательный цикл for...of await и ограниченный Promise.all могут отличаться на порядки по реальному времени (один бенчмарк: ~30 с последовательно против ~340 мс конкурентно), но неограниченная конкурентность не обгоняет ограниченную — за пределом того, что downstream способен впитать, лишняя работа в полёте лишь добавляет очередь, память и отказы. Сладкая точка — наибольшее k, которое downstream терпит, а не бесконечность. Это k обычно задаётся собственным лимитом downstream: пул соединений на 20, лимит API, IOPS диска.

Три режима, одно решение

Представь спектр. Последовательный (await в цикле) — по одному за раз: безопасно, медленно, ни на что не давит. Неограниченный (Promise.all по огромному списку) — всё сразу: быстро стартует, катастрофичен на масштабе. Ограниченный (лимит k) — продакшен-ответ: достаточно быстро, предсказуемая память, уважает downstream. Старший рефлекс при виде Promise.all(bigArray.map(...)) мгновенный: что это ограничивает? Если размер массива контролируется атакующим или данными, неограниченный Promise.all — латентный сбой.