forked from EduCraft/curriculum-project-hub
feat: add retry with exponential backoff for file send/download
- withRetry helper: configurable maxAttempts, baseDelayMs, shouldRetry - sendFile: upload and message send wrapped with 3 retries, 1s/2s/4s backoff - downloadMessageFile: wrapped with 3 retries - Add unit tests (6) for retry behavior and sendFile retry integration
This commit is contained in:
+44
-11
@@ -27,6 +27,33 @@ export interface OutboundPayload {
|
||||
readonly content: string;
|
||||
}
|
||||
|
||||
export async function withRetry<T>(
|
||||
fn: () => Promise<T>,
|
||||
options: {
|
||||
maxAttempts?: number;
|
||||
baseDelayMs?: number;
|
||||
shouldRetry?: (error: unknown) => boolean;
|
||||
} = {},
|
||||
): Promise<T> {
|
||||
const maxAttempts = Math.max(1, options.maxAttempts ?? 3);
|
||||
const baseDelayMs = options.baseDelayMs ?? 1000;
|
||||
const shouldRetry = options.shouldRetry ?? (() => true);
|
||||
|
||||
for (let attempt = 1; attempt <= maxAttempts; attempt += 1) {
|
||||
try {
|
||||
return await fn();
|
||||
} catch (error) {
|
||||
if (attempt === maxAttempts || !shouldRetry(error)) {
|
||||
throw error;
|
||||
}
|
||||
const delay = baseDelayMs * 2 ** (attempt - 1);
|
||||
await new Promise((resolve) => setTimeout(resolve, delay));
|
||||
}
|
||||
}
|
||||
|
||||
throw new Error("withRetry reached an unreachable state");
|
||||
}
|
||||
|
||||
/** Raw `im.message.receive_v1` event payload. */
|
||||
export interface MessageReceiveEvent {
|
||||
/** Feishu's SDK dispatcher flattens v2 headers onto the top-level payload. */
|
||||
@@ -340,12 +367,14 @@ export async function downloadMessageFile(
|
||||
): Promise<void> {
|
||||
await mkdir(dirname(savePath), { recursive: true });
|
||||
try {
|
||||
const res = await rt.client.request({
|
||||
method: "GET",
|
||||
url: `/open-apis/im/v1/messages/${messageId}/resources/${fileKey}`,
|
||||
params: { type: "file" },
|
||||
responseType: "arraybuffer",
|
||||
}) as { data: Buffer };
|
||||
const res = await withRetry(async () => {
|
||||
return rt.client.request({
|
||||
method: "GET",
|
||||
url: `/open-apis/im/v1/messages/${messageId}/resources/${fileKey}`,
|
||||
params: { type: "file" },
|
||||
responseType: "arraybuffer",
|
||||
}) as Promise<{ data: Buffer }>;
|
||||
});
|
||||
await writeFile(savePath, res.data);
|
||||
} catch (e) {
|
||||
rt.logger.warn({ messageId, fileKey, err: e instanceof Error ? e.message : String(e) }, "downloadMessageFile failed");
|
||||
@@ -364,16 +393,20 @@ export async function sendFile(rt: FeishuRuntime, chatId: string, filePath: stri
|
||||
const buf = await readFile(filePath);
|
||||
const ext = fileName.split(".").pop() ?? "";
|
||||
const fileType = ext === "pdf" ? "pdf" : ext === "doc" ? "doc" : ext === "xls" ? "xls" : ext === "ppt" ? "ppt" : "stream";
|
||||
const uploadRes = await client.im.v1.file.create({ data: { file_type: fileType, file_name: fileName, file: buf } });
|
||||
const uploadRes = await withRetry(async () =>
|
||||
client.im.v1.file.create({ data: { file_type: fileType, file_name: fileName, file: buf } }),
|
||||
);
|
||||
const fileKey = fileKeyFromResponse(uploadRes);
|
||||
if (fileKey === undefined) {
|
||||
rt.logger.warn({ chatId, filePath }, "sendFile failed: upload response missing file_key");
|
||||
return null;
|
||||
}
|
||||
const msgRes = await client.im.v1.message.create({
|
||||
params: { receive_id_type: "chat_id" },
|
||||
data: { receive_id: chatId, msg_type: "file", content: JSON.stringify({ file_key: fileKey }) },
|
||||
});
|
||||
const msgRes = await withRetry(async () =>
|
||||
client.im.v1.message.create({
|
||||
params: { receive_id_type: "chat_id" },
|
||||
data: { receive_id: chatId, msg_type: "file", content: JSON.stringify({ file_key: fileKey }) },
|
||||
}),
|
||||
);
|
||||
const messageId = messageIdFromResponse(msgRes);
|
||||
if (messageId === null) {
|
||||
rt.logger.warn({ chatId, filePath }, "sendFile failed: message response missing message_id");
|
||||
|
||||
@@ -0,0 +1,163 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi, type Mock } from "vitest";
|
||||
import { mkdtemp, rm, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { sendFile, withRetry, type FeishuRuntime } from "../../src/feishu/client.js";
|
||||
|
||||
describe("withRetry", () => {
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers();
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it("succeeds on first attempt without delaying", async () => {
|
||||
const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout");
|
||||
const fn = vi.fn(async () => "ok");
|
||||
|
||||
await expect(withRetry(fn)).resolves.toBe("ok");
|
||||
|
||||
expect(fn).toHaveBeenCalledTimes(1);
|
||||
expect(setTimeoutSpy).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("retries on failure then succeeds", async () => {
|
||||
const fn = vi.fn()
|
||||
.mockRejectedValueOnce(new Error("temporary failure"))
|
||||
.mockResolvedValueOnce("ok");
|
||||
|
||||
const result = withRetry(fn);
|
||||
await Promise.resolve();
|
||||
|
||||
expect(fn).toHaveBeenCalledTimes(1);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
|
||||
await expect(result).resolves.toBe("ok");
|
||||
expect(fn).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("exhausts all attempts and rethrows", async () => {
|
||||
const error = new Error("still failing");
|
||||
const fn = vi.fn(async () => {
|
||||
throw error;
|
||||
});
|
||||
|
||||
const result = withRetry(fn);
|
||||
const assertion = expect(result).rejects.toBe(error);
|
||||
await Promise.resolve();
|
||||
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
await vi.advanceTimersByTimeAsync(2000);
|
||||
|
||||
await assertion;
|
||||
expect(fn).toHaveBeenCalledTimes(3);
|
||||
});
|
||||
|
||||
it("respects shouldRetry predicate", async () => {
|
||||
const error = new Error("validation failure");
|
||||
const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout");
|
||||
const shouldRetry = vi.fn(() => false);
|
||||
const fn = vi.fn(async () => {
|
||||
throw error;
|
||||
});
|
||||
|
||||
await expect(withRetry(fn, { shouldRetry })).rejects.toBe(error);
|
||||
|
||||
expect(fn).toHaveBeenCalledTimes(1);
|
||||
expect(shouldRetry).toHaveBeenCalledWith(error);
|
||||
expect(setTimeoutSpy).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("uses exponential backoff delays", async () => {
|
||||
const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout");
|
||||
const fn = vi.fn()
|
||||
.mockRejectedValueOnce(new Error("first failure"))
|
||||
.mockRejectedValueOnce(new Error("second failure"))
|
||||
.mockResolvedValueOnce("ok");
|
||||
|
||||
const result = withRetry(fn);
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
await vi.advanceTimersByTimeAsync(2000);
|
||||
|
||||
await expect(result).resolves.toBe("ok");
|
||||
expect(setTimeoutSpy.mock.calls.map(([, delay]) => delay)).toEqual([1000, 2000]);
|
||||
});
|
||||
});
|
||||
|
||||
describe("sendFile retry", () => {
|
||||
afterEach(() => {
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it("retries upload on transient failure", async () => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "hub-feishu-retry-"));
|
||||
try {
|
||||
const file = join(dir, "student.pdf");
|
||||
await writeFile(file, "pdf bytes");
|
||||
|
||||
let failFirstUpload: (error: Error) => void = () => {
|
||||
throw new Error("upload rejection was not initialized");
|
||||
};
|
||||
const firstUpload = new Promise<never>((_resolve, reject) => {
|
||||
failFirstUpload = reject;
|
||||
});
|
||||
const fileCreate = vi.fn()
|
||||
.mockReturnValueOnce(firstUpload)
|
||||
.mockResolvedValueOnce({ data: { file_key: "file-key-1" } });
|
||||
const messageCreate = vi.fn(async () => ({ data: { message_id: "message-1" } }));
|
||||
const rt = mockRuntime({ fileCreate, messageCreate });
|
||||
|
||||
const result = sendFile(rt, "chat-1", file, "student.pdf");
|
||||
await waitForCalls(fileCreate, 1);
|
||||
vi.useFakeTimers();
|
||||
failFirstUpload(new Error("temporary upload failure"));
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
|
||||
await expect(result).resolves.toBe("message-1");
|
||||
expect(fileCreate).toHaveBeenCalledTimes(2);
|
||||
expect(messageCreate).toHaveBeenCalledTimes(1);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
async function waitForCalls(mock: Mock, expectedCalls: number): Promise<void> {
|
||||
for (let attempt = 0; attempt < 50; attempt += 1) {
|
||||
if (mock.mock.calls.length >= expectedCalls) return;
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
}
|
||||
throw new Error(`Timed out waiting for mock to be called ${expectedCalls} time(s)`);
|
||||
}
|
||||
|
||||
function mockRuntime(options: {
|
||||
readonly fileCreate: (payload: unknown) => Promise<unknown>;
|
||||
readonly messageCreate: (payload: unknown) => Promise<unknown>;
|
||||
}): FeishuRuntime {
|
||||
return {
|
||||
client: {
|
||||
im: {
|
||||
v1: {
|
||||
file: { create: options.fileCreate },
|
||||
message: { create: options.messageCreate },
|
||||
},
|
||||
},
|
||||
} as unknown as FeishuRuntime["client"],
|
||||
logger: {
|
||||
info() {},
|
||||
warn() {},
|
||||
error() {},
|
||||
debug() {},
|
||||
fatal() {},
|
||||
child() { return this; },
|
||||
level: "silent",
|
||||
} as unknown as FeishuRuntime["logger"],
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user