Promise.all, allSettled, race hay any: ngữ nghĩa, xử lý lỗi, timeout và giới hạn concurrency

Trang dashboard tài khoản load sáu widget: profile, đơn hàng, hóa đơn, gợi ý sản phẩm, thông báo và điểm thưởng. Có người gom cả sáu lời gọi vô Promise.all vì nó nhanh hơn await từng cái một. Rồi một buổi chiều service gợi ý sản phẩm trục trặc, và request dashboard nào cũng trả về 500, trong khi năm service còn lại vẫn khỏe re. Cũng tuần đó, một job chạy ban đêm làm Promise.all(customers.map(sendEmail)) cho 8.000 khách hàng, mở cả ngàn socket cùng lúc và bị nhà cung cấp email rate limit chỉ sau vài giây.

Hai sự cố đó chung một gốc: coi mấy hàm gộp promise như nhau, kiểu "chạy song song cho nhanh" là xong. Không phải vậy. Promise.all, Promise.allSettled, Promise.race và Promise.any khác nhau ở chỗ khi nào thì settle, xử lý lỗi ra sao, và trả về cái gì. Không cái nào giới hạn concurrency, và cũng không cái nào hủy (cancel) được việc gì. Chọn đúng hàm, rồi tự bổ sung mấy mảnh còn thiếu.

Bốn hàm gộp nhìn lướt qua

Cả bốn đều nhận một iterable gồm promise (hoặc giá trị thường) và trả về một promise duy nhất. Khác nhau ở luật settle:

Có hai sự thật đúng với cả bốn, và gây ra phần lớn bug ngoài đời:

1. Công việc đã bắt đầu trước khi gọi hàm gộp. Gọi fetch(url) là request chạy liền. Promise.all chỉ đứng ngoài quan sát mấy promise đang chạy sẵn. Đó là lý do Promise.all(customers.map(sendEmail)) bắn 8.000 request một lượt: chính cái map đã khởi động hết rồi.
2. Không có gì bị hủy hết. Khi Promise.all reject sớm, hay Promise.race chọn ra người thắng, mấy tác vụ còn lại vẫn chạy tới cùng. Kết quả của tụi nó chỉ bị bỏ qua thôi. Muốn hủy thiệt thì phải thêm AbortController.

Chọn hàm nào cho đúng

Promise.all: được ăn cả, fail là dừng liền

Xài Promise.all khi kết quả nào cũng bắt buộc phải có, và chỉ một lỗi là cả thao tác mất ý nghĩa. Muốn dựng hóa đơn thì cần khách hàng, các dòng sản phẩm và thuế suất; thiếu một cái là không có hóa đơn.

async function buildInvoice(orderId: string) {
  const [customer, items, taxRates] = await Promise.all([
    customers.forOrder(orderId),
    orderItems.list(orderId),
    tax.ratesFor(orderId),
  ]);

  return renderInvoice(customer, items, taxRates);
}

Mảng kết quả theo thứ tự input, không phải thứ tự hoàn thành, và destructuring giữ type chính xác trong TypeScript. Promise reject với lý do của lần reject đầu tiên ngay khi nó xảy ra, rất tốt cho latency: câu trả lời đã là "lỗi" rồi thì khỏi phải chờ mấy anh em chậm chạp.

Mặt trái: mấy promise kia vẫn đang chạy, và nếu sau đó tụi nó reject thì Promise.all đã gắn handler sẵn bên trong, nên mình không bị cảnh báo unhandled rejection, nhưng cũng chẳng nghe được gì về tụi nó. Nếu một tác vụ anh em có side effect (ghi dữ liệu, thanh toán), side effect đó vẫn xảy ra.

Promise.allSettled: có một phần kết quả cũng được

Dashboard là ví dụ kinh điển cho allSettled: mỗi widget độc lập, widget nào hư thì hiện "tạm thời không khả dụng", chứ không kéo sập cả trang.

type Widget = 'profile' | 'orders' | 'recommendations';

async function loadDashboard(userId: string) {
  const loaders: Record<Widget, () => Promise<unknown>> = {
    profile: () => api.profile(userId),
    orders: () => api.orders(userId),
    recommendations: () => api.recommendations(userId),
  };

  const names = Object.keys(loaders) as Widget[];
  const results = await Promise.allSettled(names.map((name) => loaders[name]()));

  return Object.fromEntries(
    results.map((result, i) => {
      if (result.status === 'fulfilled') {
        return [names[i], { ok: true, data: result.value }];
      }

      logger.warn('widget failed', { widget: names[i], reason: String(result.reason) });

      return [names[i], { ok: false, data: null }];
    }),
  );
}

allSettled không bao giờ reject, nghĩa là mình phải tự đi coi mấy cái lỗi. Lỗi hay gặp là gọi allSettled, map qua value, rồi lặng lẽ render undefined cho mấy entry bị fail. Luôn rẽ nhánh theo status, và log hoặc đếm reason, không thì mất sạch khả năng quan sát lỗi. Trong TypeScript, narrow bằng result.status === 'fulfilled' sẽ cho ra PromiseFulfilledResult<T>, nên result.value có type đàng hoàng.

Promise.race: timeout, và cái timer bị bỏ quên

race settle theo input nào settle trước, fulfil hay reject cũng vậy. Công dụng thực tế chính của nó là bọc timeout quanh một thao tác vốn không hỗ trợ timeout:

export function withTimeout<T>(promise: Promise<T>, ms: number, label = 'operation'): Promise<T> {
  let timer: ReturnType<typeof setTimeout> | undefined;

  const timeout = new Promise<never>((_, reject) => {
    timer = setTimeout(() => reject(new Error(`${label} timed out after ${ms}ms`)), ms);
  });

  return Promise.race([promise, timeout]).finally(() => clearTimeout(timer));
}

Có hai chỗ dễ làm sai. Thứ nhất là clearTimeout trong finally: thiếu nó thì mỗi lần gọi thành công lại để lại một timer còn sống, làm process Node không chịu tắt lúc shutdown, còn trên server chạy lâu thì thành ra cả ngàn timer treo. Thứ hai, timeout chỉ giúp mình khỏi phải chờ. Thao tác bên dưới vẫn đang chạy. Với fetch, truyền AbortSignal.timeout(ms) vô signal để request bị hủy thiệt; còn pattern race thì để dành cho mấy thứ không abort được, như lời gọi SDK của bên thứ ba.

Nhớ thêm là trong race, reject cũng tính là thắng. Thao tác fail sau 5 ms thì race reject liền với lỗi đó, thường cũng là điều mình muốn.

Promise.any: ai thành công trước thì thắng

any bỏ qua mọi lần reject cho tới khi tất cả input đều reject. Nó hợp với bài toán dự phòng: hỏi ba mirror hay ba replica, cái nào trả lời thành công trước thì lấy.

async function fetchFromFastestMirror(path: string): Promise<unknown> {
  const mirrors = ['https://eu.cdn.example.com', 'https://us.cdn.example.com', 'https://ap.cdn.example.com'];

  try {
    return await Promise.any(
      mirrors.map(async (base) => {
        const res = await fetch(`${base}${path}`);
        if (!res.ok) {
          throw new Error(`${base} answered ${res.status}`);
        }
        return res.json();
      }),
    );
  } catch (error) {
    if (error instanceof AggregateError) {
      console.error('all mirrors failed', error.errors.map(String));
    }
    throw error;
  }
}

Để ý dòng if (!res.ok) throw. fetch chỉ reject khi lỗi mạng, không reject khi gặp HTTP 500, nên thiếu check đó thì một mirror trả trang lỗi thật nhanh sẽ "thắng". AggregateError.errors chứa từng lý do lỗi theo thứ tự input, đúng thứ mình cần để ghi log. Xài any ở đây làm tải lên mirror tăng gấp ba, nên chỉ dành cho mấy lượt đọc nhỏ và idempotent.

Giới hạn concurrency: thứ mà không hàm nào làm giùm

Với 8.000 email, không hàm gộp nào cứu được, vì công việc đã chạy hết bên trong map. Cách sửa là kiểm soát lúc nào mỗi tác vụ được bắt đầu: một pool nhỏ gồm vài worker, mỗi worker tự lấy item kế tiếp từ một index dùng chung.

export async function mapWithConcurrency<T, R>(
  items: readonly T[],
  limit: number,
  fn: (item: T, index: number) => Promise<R>,
): Promise<R[]> {
  if (!Number.isInteger(limit) || limit < 1) {
    throw new RangeError('limit must be a positive integer');
  }

  const results = new Array<R>(items.length);
  let next = 0;
  let failed = false;

  async function worker(): Promise<void> {
    while (!failed && next < items.length) {
      const index = next++;
      try {
        results[index] = await fn(items[index], index);
      } catch (error) {
        failed = true;
        throw error;
      }
    }
  }

  const workers = Array.from({ length: Math.min(limit, items.length) }, () => worker());
  await Promise.all(workers);

  return results;
}

Vì JavaScript chạy đơn luồng nên const index = next++ là atomic, khỏi cần lock. Kết quả được ghi vô results[index], vậy nên thứ tự output khớp với input bất kể task nào xong trước. Cờ failed cho mình hành vi fail-fast giống Promise.all, và còn chặn pool bắt đầu thêm item mới sau lỗi đầu tiên, chuyện này quan trọng khi mỗi item là một email hay một lần trừ tiền thẻ. Nếu muốn gom hết kết quả kể cả lỗi, catch ngay trong worker rồi lưu { ok: false, error }.

Mấy thư viện như p-limit và p-map làm đúng ý tưởng này với nhiều tùy chọn hơn. Cái chính là hiểu được hình dạng: số promise đang bay là cố định, chứ không phải số batch cố định. Chia nhóm 50 cái rồi Promise.all từng nhóm cũng chạy được, nhưng nhóm nào cũng phải chờ thằng chậm nhất, nên throughput tụt xuống bằng tốc độ của request tệ nhất trong mỗi batch.

Kiểm thử logic gộp promise với Vitest

Code concurrency là chỗ "máy em chạy được mà" hay xuất hiện nhất. Test đúng mấy tính chất quan trọng: concurrency đỉnh, thứ tự, và hành vi khi lỗi. Fake timer giúp test timeout chạy tức thì và ổn định.

import { afterEach, describe, expect, it, vi } from 'vitest';
import { mapWithConcurrency, withTimeout } from './async-utils';

const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));

describe('mapWithConcurrency', () => {
  it('never runs more than limit tasks and keeps input order', async () => {
    let active = 0;
    let peak = 0;

    const out = await mapWithConcurrency([30, 10, 20, 5], 2, async (ms) => {
      active++;
      peak = Math.max(peak, active);
      await sleep(ms);
      active--;
      return ms * 2;
    });

    expect(out).toEqual([60, 20, 40, 10]);
    expect(peak).toBe(2);
  });

  it('rejects with the first error and starts no new work', async () => {
    const started: number[] = [];

    await expect(
      mapWithConcurrency([1, 2, 3, 4], 1, async (n) => {
        started.push(n);
        if (n === 2) throw new Error('boom');
        return n;
      }),
    ).rejects.toThrow('boom');

    expect(started).toEqual([1, 2]);
  });
});

describe('withTimeout', () => {
  afterEach(() => vi.useRealTimers());

  it('rejects when the promise is too slow', async () => {
    vi.useFakeTimers();
    const pending = withTimeout(new Promise(() => {}), 1000, 'report');
    const assertion = expect(pending).rejects.toThrow('report timed out after 1000ms');

    await vi.advanceTimersByTimeAsync(1000);
    await assertion;
  });
});

Gắn assertion rejects trước khi tua timer; không thì lúc reject chưa có handler nào, và Vitest sẽ báo là unhandled.

Những lỗi hay gặp

• Tưởng Promise.all sẽ hủy mấy tác vụ còn lại. Không có đâu. Muốn vậy thì truyền AbortSignal vô từng tác vụ và abort nó trong catch.
• Promise.all trên một mảng không giới hạn. items.map(asyncFn) khởi động tất cả một lượt. Cái gì đụng tới network hay database thì phải có giới hạn concurrency.
• Lơ mấy cái fail của allSettled. Nó không bao giờ reject, nên lỗi biến mất luôn nếu mình không check status và log reason.
• Promise.race([]) treo mãi mãi. Danh sách có khả năng rỗng thì phải chặn trước.
• Timeout để rò timer. Luôn clear timer trong finally.
• Coi lỗi HTTP là thành công. fetch resolve với 4xx và 5xx. Check res.ok trước khi đưa promise cho any hay all.
• await tuần tự trong vòng lặp dù các việc độc lập nhau. for (const id of ids) await load(id) tốn tổng latency của tất cả. Khi cần giữ thứ tự hay giữ nhịp thì ổn, còn không thì đó là kiểu chậm vô tình rất hay gặp.

Khi nào xài cái nào, và khi nào không

• Promise.all: một nhóm nhỏ, cố định các kết quả bắt buộc, lỗi một cái là bỏ cả. Không dành cho danh sách lớn hay danh sách do user kiểm soát mà không có giới hạn concurrency.
• Promise.allSettled: các tác vụ độc lập, thành công một phần cũng chấp nhận được, lỗi thì báo riêng từng cái. Không hợp khi một lỗi cần dừng phần còn lại cho nhanh; nó luôn chờ thằng chậm nhất.
• Promise.race: timeout và mấy tín hiệu kiểu "cái nào settle trước", ví dụ đua công việc với tín hiệu shutdown. Không dùng để chọn nguồn thành công nhanh nhất, vì một cái fail nhanh sẽ thắng.
• Promise.any: đọc dữ liệu dự phòng, idempotent, cái nào thành công cũng được. Không dành cho thao tác ghi, và không hợp khi tải thêm lên backend là vấn đề.
• Bộ giới hạn concurrency: mọi job batch chạy trên nhiều item. Kết hợp với kiểu xử lý lỗi của all hay allSettled tùy chuyện một lỗi có nên dừng cả batch hay không.

Mấy hàm gộp trả lời câu "chờ ra sao?". Giới hạn concurrency và cancel trả lời câu "khởi động bao nhiêu?" và "dừng lại kiểu gì?". Code production thường cần đủ cả ba.