Promise 与任务取消

概念定义

异步工作需要一个可组合的关断开关：用户放弃的请求、被首个成功者取代的扇出、超期的截止时间。取消令牌就是这个开关；操作通过检查信号并确定性展开来配合。

工作机制

控制器升起信号，操作侧观察它：被等待的原语以“取消专用错误”拒绝、循环在迭代边界检查、清理放进 finally。组合规则是子令牌随父令牌取消、link 合并多信号、截止时间是定时器驱动的令牌。

使用场景

一切可能活得比用处更久的操作：HTTP 请求、数据库查询、批任务、扇出作用域。SDK 签名应暴露取消参数，让调用方贯穿每一层传递令牌。

常见误区

丢弃引用不等于取消——游离 promise 照跑到底照耗资源。超时包装器若只与定时器竞速而不传信号，底层请求在超时后仍在暗跑。取消是控制流不是失败：把它当异常塞进通用错误处理，会对用户已放弃的工作发起幽灵重试。库必须在取消路径释放锁、连接与部分写入，并用测试覆盖；长命控制器上的取消监听泄漏则是慢性内存漏。

代码示例

interface RetryableError extends Error {
  retryable?: boolean;
}

class PermanentError extends Error {
  retryable = false;
}

function classify(err: unknown): RetryableError {
  if (err instanceof PermanentError) return err;

  const message = err instanceof Error ? err.message : String(err);
  const transient =
    /timeout|econnreset|socket hang up|503|429/i.test(message);

  return Object.assign(
    err instanceof Error ? err : new Error(message),
    { retryable: transient },
  );
}

async function withRetry<T>(
  fn: () => Promise<T>,
  { attempts = 3, baseMs = 200 } = {},
): Promise<T> {
  let lastError: RetryableError;

  for (let i = 1; i <= attempts; i++) {
    try {
      return await fn();
    } catch (raw) {
      lastError = classify(raw);

      if (!lastError.retryable || i === attempts) {
        throw lastError;
      }
      await new Promise((r) => setTimeout(r, baseMs * 2 ** (i - 1)));
    }
  }
  throw lastError!;
}

const result = await withRetry(async () => {
  if (Math.random() < 0.7) {
    const err = new Error('upstream 503 unavailable');
    throw err;
  }
  return 'settled';
});
console.log('outcome:', result);

export function mapPool(items, mapper, concurrency = 4) {
  if (!Array.isArray(items)) throw new TypeError("items must be an array");
  if (typeof mapper !== "function") throw new TypeError("mapper must be a function");
  if (!Number.isInteger(concurrency) || concurrency < 1) {
    throw new RangeError("concurrency must be a positive integer");
  }

  const results = new Array(items.length);
  let nextIndex = 0;

  async function worker() {
    while (nextIndex < items.length) {
      const index = nextIndex;
      nextIndex += 1;
      results[index] = await mapper(items[index], index);
    }
  }

  const workerCount = Math.min(concurrency, items.length);
  const workers = Array.from({ length: workerCount }, () => worker());
  return Promise.all(workers).then(() => results);
}