收紧 LimiterFileSystem 的 429 自动重试范围，避免写操作被盲目重放

Author: cyfung1031

packages/filesystem/limiter.test.ts

@@ -0,0 +1,107 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+import type FileSystem from "./filesystem";
+import type { FileInfo, FileReader, FileWriter } from "./filesystem";
+import LimiterFileSystem from "./limiter";
+
+function createFs(): FileSystem {
+  return {
+    verify: vi.fn(async () => {}),
+    open: vi.fn(async () => {
+      const reader: FileReader = {
+        read: vi.fn(async () => "content"),
+      };
+      return reader;
+    }),
+    openDir: vi.fn(async () => createFs()),
+    create: vi.fn(async () => {
+      const writer: FileWriter = {
+        write: vi.fn(async () => {}),
+      };
+      return writer;
+    }),
+    createDir: vi.fn(async () => {}),
+    delete: vi.fn(async () => {}),
+    list: vi.fn(async () => []),
+    getDirUrl: vi.fn(async () => "url"),
+  };
+}
+
+const file: FileInfo = {
+  name: "test.user.js",
+  path: "/test.user.js",
+  size: 1,
+  digest: "digest",
+  createtime: 1,
+  updatetime: 1,
+};
+
+describe("LimiterFileSystem", () => {
+  afterEach(() => {
+    vi.useRealTimers();
+    vi.restoreAllMocks();
+  });
+
+  it("should retry list on 429", async () => {
+    vi.useFakeTimers();
+    const fs = createFs();
+    vi.mocked(fs.list).mockRejectedValueOnce(new Error("429 Too Many Requests")).mockResolvedValueOnce([]);
+    const limiter = new LimiterFileSystem(fs);
+
+    const promise = limiter.list();
+    await vi.runOnlyPendingTimersAsync();
+
+    await expect(promise).resolves.toEqual([]);
+    expect(fs.list).toHaveBeenCalledTimes(2);
+  });
+
+  it("should not retry delete on 429", async () => {
+    const fs = createFs();
+    vi.mocked(fs.delete).mockRejectedValueOnce(new Error("429 Too Many Requests"));
+    const limiter = new LimiterFileSystem(fs);
+
+    await expect(limiter.delete("/test.user.js")).rejects.toThrow("429 Too Many Requests");
+    expect(fs.delete).toHaveBeenCalledTimes(1);
+  });
+
+  it("should not retry createDir on 429", async () => {
+    const fs = createFs();
+    vi.mocked(fs.createDir).mockRejectedValueOnce(new Error("429 Too Many Requests"));
+    const limiter = new LimiterFileSystem(fs);
+
+    await expect(limiter.createDir("/dir")).rejects.toThrow("429 Too Many Requests");
+    expect(fs.createDir).toHaveBeenCalledTimes(1);
+  });
+
+  it("should not retry writer.write on 429", async () => {
+    const fs = createFs();
+    const write = vi.fn(async () => {});
+    vi.mocked(fs.create).mockResolvedValueOnce({
+      write,
+    });
+    write.mockRejectedValueOnce(new Error("429 Too Many Requests"));
+    const limiter = new LimiterFileSystem(fs);
+    const writer = await limiter.create(file.path);
+
+    await expect(writer.write("content")).rejects.toThrow("429 Too Many Requests");
+    expect(write).toHaveBeenCalledTimes(1);
+  });
+
+  it("should retry reader.read on 429", async () => {
+    vi.useFakeTimers();
+    const fs = createFs();
+    const read = vi.fn(async () => "content");
+    vi.mocked(fs.open).mockResolvedValueOnce({
+      read,
+    });
+    read.mockRejectedValueOnce(new Error("429 Too Many Requests"));
+    read.mockResolvedValueOnce("content");
+    const limiter = new LimiterFileSystem(fs);
+    const reader = await limiter.open(file);
+
+    const promise = reader.read("string");
+    await vi.runOnlyPendingTimersAsync();
+
+    await expect(promise).resolves.toBe("content");
+    expect(read).toHaveBeenCalledTimes(2);
+  });
+});

packages/filesystem/limiter.ts

@@ -1,6 +1,8 @@
 import type FileSystem from "./filesystem";
 import type { FileInfo, FileReader, FileWriter } from "./filesystem";
 
+const RETRYABLE_429_OPS = new Set(["verify", "open", "read", "openDir", "list", "getDirUrl"]);
+
 /**
  * 速率限制器
  * 控制并发操作数量，防止过多并发请求
@@ -21,7 +23,7 @@ export class RateLimiter {
    * @param fn 要执行的操作函数
    * @returns 操作结果
    */
-  async execute<T>(fn: () => Promise<T>): Promise<T> {
+  async execute<T>(fn: () => Promise<T>, op = "unknown"): Promise<T> {
     // 如果当前运行的操作数已达到上限，则等待
     while (this.running >= this.maxConcurrent) {
       await new Promise<void>((resolve) => {
@@ -31,7 +33,7 @@ export class RateLimiter {
 
     this.running++;
     try {
-      return await this.executeWithRetry(fn);
+      return await this.executeWithRetry(fn, op);
     } finally {
       this.running--;
       // 执行完成后，从队列中取出下一个等待的操作
@@ -47,15 +49,15 @@ export class RateLimiter {
    * @param fn 要执行的操作函数
    * @returns 操作结果
    */
-  private async executeWithRetry<T>(fn: () => Promise<T>): Promise<T> {
+  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);
-        if (errorStr.includes("429") && i < 10) {
+        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));
@@ -68,6 +70,10 @@ export class RateLimiter {
     }
     throw new Error("Max retries exceeded");
   }
+
+  private shouldRetry429(op: string, errorStr: string): boolean {
+    return errorStr.includes("429") && RETRYABLE_429_OPS.has(op);
+  }
 }
 
 // 文件系统限速器，防止并发请求过多达到服务器限制
@@ -83,47 +89,47 @@ export default class LimiterFileSystem implements FileSystem {
   }
 
   verify(): Promise<void> {
-    return this.limiter.execute(() => this.fs.verify());
+    return this.limiter.execute(() => this.fs.verify(), "verify");
   }
 
   async open(file: FileInfo): Promise<FileReader> {
     return this.limiter.execute(async () => {
       const reader = await this.fs.open(file);
       return {
-        read: (type) => this.limiter.execute(() => reader.read(type)),
+        read: (type) => this.limiter.execute(() => reader.read(type), "read"),
       };
-    });
+    }, "open");
   }
 
   async openDir(path: string): Promise<FileSystem> {
     return this.limiter.execute(async () => {
       const fs = await this.fs.openDir(path);
       return new LimiterFileSystem(fs, this.limiter);
-    });
+    }, "openDir");
   }
 
   async create(path: string): Promise<FileWriter> {
     return this.limiter.execute(async () => {
       const writer = await this.fs.create(path);
       return {
-        write: (content) => this.limiter.execute(() => writer.write(content)),
+        write: (content) => this.limiter.execute(() => writer.write(content), "write"),
       };
-    });
+    }, "create");
   }
 
   createDir(dir: string): Promise<void> {
-    return this.limiter.execute(() => this.fs.createDir(dir));
+    return this.limiter.execute(() => this.fs.createDir(dir), "createDir");
   }
 
   delete(path: string): Promise<void> {
-    return this.limiter.execute(() => this.fs.delete(path));
+    return this.limiter.execute(() => this.fs.delete(path), "delete");
   }
 
   list(): Promise<FileInfo[]> {
-    return this.limiter.execute(() => this.fs.list());
+    return this.limiter.execute(() => this.fs.list(), "list");
   }
 
   getDirUrl(): Promise<string> {
-    return this.limiter.execute(() => this.fs.getDirUrl());
+    return this.limiter.execute(() => this.fs.getDirUrl(), "getDirUrl");
   }
 }