Class RateLimiter

packages/filesystem/limiter.ts:10–83

 * 控制并发操作数量，防止过多并发请求
 */
export class RateLimiter {
  private queue: Array<() => void> = [];

  private running = 0;

  private maxConcurrent: number;

  constructor(maxConcurrent = 5) {
    this.maxConcurrent = maxConcurrent;
  }

  /**
   * 执行限速操作
   * @param fn 要执行的操作函数
   * @param op 操作类型，用于在遇到 429 时判断是否允许自动重试。默认值 "unknown" 不在白名单内，会被视为不可重试
   * @returns 操作结果
   */
  async execute<T>(fn: () => Promise<T>, op = "unknown"): Promise<T> {
    // 如果当前运行的操作数已达到上限，则等待
    while (this.running >= this.maxConcurrent) {
      await new Promise<void>((resolve) => {
        this.queue.push(resolve);
      });
    }

    this.running++;
    try {
      return await this.executeWithRetry(fn, op);
    } finally {
      this.running--;
      // 执行完成后，从队列中取出下一个等待的操作
      const next = this.queue.shift();
      if (next) {
        next();
      }
    }
  }

  /**
   * 执行操作并处理 429 错误重试
   * @param fn 要执行的操作函数
   * @param op 操作类型，用于判定该操作在遇到 429 时是否进入指数退避重试
   * @returns 操作结果
   */
  private async executeWithRetry<T>(fn: () => Promise<T>, op: string): Promise<T> {
    // 最多重试 10 次
    for (let i = 0; i <= 10; i++) {
      try {
        return await fn();
      } catch (error) {
        // 检查错误字符串中是否包含 429
        const errorStr = String(error).toLowerCase();
        if (this.shouldRetry429(op, ` ${errorStr} `) && i < 10) {
          // 遇到 429 错误且未达到重试上限，采用指数退避策略延迟后继续重试
          const delay = Math.min(2000 * Math.pow(2, i), 60000);
          await new Promise((resolve) => setTimeout(resolve, delay));
          // 继续下一次循环重试
          continue;