@@ -903,6 +903,172 @@ describe("TelegramPollingSession", () => {
903903}
904904});
905905906+it("drains worker-spooled updates without waiting for the next drain interval", async () => {
907+const abort = new AbortController();
908+const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-"));
909+const handleUpdate = vi.fn(async () => {
910+abort.abort();
911+});
912+const bot = {
913+api: {
914+deleteWebhook: vi.fn(async () => true),
915+config: { use: vi.fn() },
916+},
917+init: vi.fn(async () => undefined),
918+ handleUpdate,
919+stop: vi.fn(async () => undefined),
920+};
921+createTelegramBotMock.mockReturnValueOnce(bot);
922+let onMessage: WorkerMessageListener | undefined;
923+let stopWorker: (() => void) | undefined;
924+const workerDone = new Promise<void>((resolve) => {
925+stopWorker = resolve;
926+});
927+const ackSpooledUpdate = vi.fn();
928+const createWorker = vi.fn(() => ({
929+onMessage: vi.fn((listener: WorkerMessageListener) => {
930+onMessage = listener;
931+return () => undefined;
932+}),
933+ ackSpooledUpdate,
934+stop: vi.fn(async () => {
935+stopWorker?.();
936+}),
937+task: vi.fn(async () => {
938+await workerDone;
939+}),
940+}));
941+942+try {
943+const session = createPollingSession({
944+abortSignal: abort.signal,
945+isolatedIngress: {
946+enabled: true,
947+spoolDir: tempDir,
948+ createWorker,
949+drainIntervalMs: 60_000,
950+},
951+});
952+953+const runPromise = session.runUntilAbort();
954+await vi.waitFor(() => expect(onMessage).toBeDefined());
955+onMessage?.({
956+type: "update",
957+requestId: "write-1",
958+update: { update_id: 42, message: { text: "hello" } },
959+queued: 1,
960+});
961+962+await vi.waitFor(() =>
963+expect(ackSpooledUpdate).toHaveBeenCalledWith("write-1", { ok: true, updateId: 42 }),
964+);
965+onMessage?.({ type: "spooled", updateId: 42, queued: 1 });
966+await vi.waitFor(() =>
967+expect(handleUpdate).toHaveBeenCalledWith({ update_id: 42, message: { text: "hello" } }),
968+);
969+await vi.waitFor(async () => expect(await pendingUpdateIds(tempDir, "all")).toEqual([]));
970+stopWorker?.();
971+await runPromise;
972+} finally {
973+abort.abort();
974+stopWorker?.();
975+await fs.rm(tempDir, { recursive: true, force: true });
976+}
977+});
978+979+it("drains worker-spooled updates that arrive during an active drain", async () => {
980+const abort = new AbortController();
981+const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-"));
982+983+await writeTelegramSpooledUpdate({
984+spoolDir: tempDir,
985+update: { update_id: 1, message: { text: "pre-seeded" } },
986+});
987+988+const handleUpdate = vi.fn(async (update) => {
989+if (update.update_id === 1) {
990+await new Promise<void>((resolve) => { setTimeout(resolve, 300); });
991+}
992+});
993+994+const bot = {
995+api: {
996+deleteWebhook: vi.fn(async () => true),
997+config: { use: vi.fn() },
998+},
999+init: vi.fn(async () => undefined),
1000+ handleUpdate,
1001+stop: vi.fn(async () => undefined),
1002+};
1003+createTelegramBotMock.mockReturnValueOnce(bot);
1004+let onMessage: WorkerMessageListener | undefined;
1005+let stopWorker: (() => void) | undefined;
1006+const workerDone = new Promise<void>((resolve) => {
1007+stopWorker = resolve;
1008+});
1009+const ackSpooledUpdate = vi.fn();
1010+const createWorker = vi.fn(() => ({
1011+onMessage: vi.fn((listener: WorkerMessageListener) => {
1012+onMessage = listener;
1013+return () => undefined;
1014+}),
1015+ ackSpooledUpdate,
1016+stop: vi.fn(async () => {
1017+stopWorker?.();
1018+}),
1019+task: vi.fn(async () => {
1020+await workerDone;
1021+}),
1022+}));
1023+1024+try {
1025+const session = createPollingSession({
1026+abortSignal: abort.signal,
1027+isolatedIngress: {
1028+enabled: true,
1029+spoolDir: tempDir,
1030+ createWorker,
1031+drainIntervalMs: 60_000,
1032+},
1033+});
1034+1035+const runPromise = session.runUntilAbort();
1036+1037+await vi.waitFor(() =>
1038+expect(handleUpdate).toHaveBeenCalledWith({
1039+update_id: 1,
1040+message: { text: "pre-seeded" },
1041+}),
1042+);
1043+1044+onMessage?.({
1045+type: "update",
1046+requestId: "write-2",
1047+update: { update_id: 2, message: { text: "during-drain" } },
1048+queued: 1,
1049+});
1050+1051+await vi.waitFor(() =>
1052+expect(ackSpooledUpdate).toHaveBeenCalledWith("write-2", { ok: true, updateId: 2 }),
1053+);
1054+onMessage?.({ type: "spooled", updateId: 2, queued: 1 });
1055+1056+await vi.waitFor(() =>
1057+expect(handleUpdate).toHaveBeenCalledWith({
1058+update_id: 2,
1059+message: { text: "during-drain" },
1060+}),
1061+);
1062+await vi.waitFor(async () => expect(await pendingUpdateIds(tempDir, "all")).toEqual([]));
1063+stopWorker?.();
1064+await runPromise;
1065+} finally {
1066+abort.abort();
1067+stopWorker?.();
1068+await fs.rm(tempDir, { recursive: true, force: true });
1069+}
1070+});
1071+9061072it("drains existing isolated ingress spool entries below the persisted offset", async () => {
9071073const abort = new AbortController();
9081074const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-"));
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。