@@ -967,6 +967,7 @@ describe("TelegramPollingSession", () => {
967967expect(handleUpdate).toHaveBeenCalledWith({ update_id: 42, message: { text: "hello" } }),
968968);
969969await vi.waitFor(async () => expect(await pendingUpdateIds(tempDir, "all")).toEqual([]));
970+abort.abort();
970971stopWorker?.();
971972await runPromise;
972973} finally {
@@ -979,17 +980,43 @@ describe("TelegramPollingSession", () => {
979980it("drains worker-spooled updates that arrive during an active drain", async () => {
980981const abort = new AbortController();
981982const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-"));
983+let releaseFirstClaim: (() => void) | undefined;
984+let firstClaimStarted: (() => void) | undefined;
985+const firstClaimGate = new Promise<void>((resolve) => {
986+releaseFirstClaim = resolve;
987+});
988+const firstClaimStartedPromise = new Promise<void>((resolve) => {
989+firstClaimStarted = resolve;
990+});
991+let blockedFirstClaim = false;
992+setTelegramRuntime({
993+state: {
994+resolveStateDir: () => tempDir,
995+openChannelIngressQueue: (
996+options?: Omit<Parameters<typeof createChannelIngressQueue>[0], "channelId">,
997+) => {
998+const queue = createChannelIngressQueue({ ...options, channelId: "telegram" });
999+return {
1000+ ...queue,
1001+claim: async (...args: Parameters<typeof queue.claim>) => {
1002+if (args[0] === "0000000000000001" && !blockedFirstClaim) {
1003+blockedFirstClaim = true;
1004+firstClaimStarted?.();
1005+await firstClaimGate;
1006+}
1007+return queue.claim(...args);
1008+},
1009+};
1010+},
1011+},
1012+} as TelegramRuntime);
98210139831014await writeTelegramSpooledUpdate({
9841015spoolDir: tempDir,
9851016update: { update_id: 1, message: { text: "pre-seeded" } },
9861017});
9871018988-const handleUpdate = vi.fn(async (update) => {
989-if (update.update_id === 1) {
990-await new Promise<void>((resolve) => { setTimeout(resolve, 300); });
991-}
992-});
1019+const handleUpdate = vi.fn(async () => undefined);
99310209941021const bot = {
9951022api: {
@@ -1033,13 +1060,8 @@ describe("TelegramPollingSession", () => {
10331060});
1034106110351062const runPromise = session.runUntilAbort();
1036-1037-await vi.waitFor(() =>
1038-expect(handleUpdate).toHaveBeenCalledWith({
1039-update_id: 1,
1040-message: { text: "pre-seeded" },
1041-}),
1042-);
1063+await vi.waitFor(() => expect(onMessage).toBeDefined());
1064+await firstClaimStartedPromise;
1043106510441066onMessage?.({
10451067type: "update",
@@ -1052,17 +1074,27 @@ describe("TelegramPollingSession", () => {
10521074expect(ackSpooledUpdate).toHaveBeenCalledWith("write-2", { ok: true, updateId: 2 }),
10531075);
10541076onMessage?.({ type: "spooled", updateId: 2, queued: 1 });
1077+releaseFirstClaim?.();
1078+releaseFirstClaim = undefined;
105510791080+await vi.waitFor(() =>
1081+expect(handleUpdate).toHaveBeenCalledWith({
1082+update_id: 1,
1083+message: { text: "pre-seeded" },
1084+}),
1085+);
10561086await vi.waitFor(() =>
10571087expect(handleUpdate).toHaveBeenCalledWith({
10581088update_id: 2,
10591089message: { text: "during-drain" },
10601090}),
10611091);
10621092await vi.waitFor(async () => expect(await pendingUpdateIds(tempDir, "all")).toEqual([]));
1093+abort.abort();
10631094stopWorker?.();
10641095await runPromise;
10651096} finally {
1097+releaseFirstClaim?.();
10661098abort.abort();
10671099stopWorker?.();
10681100await fs.rm(tempDir, { recursive: true, force: true });
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。