-
Notifications
You must be signed in to change notification settings - Fork 332
(Codex) 收紧 LimiterFileSystem 的 429 自动重试范围,避免写操作被盲目重放
#1376
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from 3 commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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); | ||
| }); | ||
| }); |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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> { | ||
| // 如果当前运行的操作数已达到上限,则等待 | ||
|
Comment on lines
21
to
28
|
||
| 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) { | ||
| 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)); | ||
|
|
@@ -68,6 +70,13 @@ export class RateLimiter { | |
| } | ||
| throw new Error("Max retries exceeded"); | ||
|
||
| } | ||
|
|
||
| private shouldRetry429(op: string, errorStr: string): boolean { | ||
| return ( | ||
| ((errorStr.includes("429") && /[^a-z\d]429[^a-z\d]/.test(errorStr)) || errorStr.includes("too many requests")) && | ||
| RETRYABLE_429_OPS.has(op) | ||
| ); | ||
| } | ||
| } | ||
|
|
||
| // 文件系统限速器,防止并发请求过多达到服务器限制 | ||
|
|
@@ -83,47 +92,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"); | ||
| } | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
当前新增的“按操作类型决定 429 是否重试”的策略里,
RETRYABLE_429_OPS覆盖了verify/open/openDir/getDirUrl等操作,但测试只覆盖了list/read会重试以及delete/createDir/write不重试。建议补充至少对verify/open/openDir/getDirUrl的 429 重试测试,以及对create本身(fs.create 抛 429)不重试的测试,避免后续字符串/集合项改动导致策略悄然失效。