@@ -80,6 +80,11 @@ type WorkerPollSuccessListener = (message: {
8080count: number;
8181finishedAt: number;
8282}) => void;
83+type WorkerPollErrorListener = (message: {
84+type: "poll-error";
85+message: string;
86+finishedAt: number;
87+}) => void;
8388type AsyncVoidFn = () => Promise<void>;
8489type MockCallSource = { mock: { calls: Array<Array<unknown>> } };
8590@@ -1480,6 +1485,192 @@ describe("TelegramPollingSession", () => {
14801485}
14811486});
148214871488+it("restarts isolated ingress when the worker task rejects before shutdown", async () => {
1489+vi.useFakeTimers({ shouldAdvanceTime: true });
1490+const abort = new AbortController();
1491+const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-"));
1492+const log = vi.fn();
1493+const setStatus = vi.fn();
1494+createTelegramBotMock.mockImplementation(() => ({
1495+api: {
1496+deleteWebhook: vi.fn(async () => true),
1497+config: { use: vi.fn() },
1498+},
1499+init: vi.fn(async () => undefined),
1500+handleUpdate: vi.fn(async () => undefined),
1501+stop: vi.fn(async () => undefined),
1502+}));
1503+1504+let workerTaskCalls = 0;
1505+const createWorker = vi.fn(() => {
1506+let stopWorker: (() => void) | undefined;
1507+const workerDone = new Promise<void>((resolve) => {
1508+stopWorker = resolve;
1509+});
1510+return {
1511+onMessage: vi.fn(() => () => undefined),
1512+stop: vi.fn(async () => {
1513+stopWorker?.();
1514+}),
1515+task: vi.fn(async () => {
1516+workerTaskCalls += 1;
1517+if (workerTaskCalls === 1) {
1518+throw new Error("worker crashed");
1519+}
1520+await workerDone;
1521+}),
1522+};
1523+});
1524+1525+try {
1526+const session = createPollingSession({
1527+abortSignal: abort.signal,
1528+ log,
1529+ setStatus,
1530+isolatedIngress: {
1531+enabled: true,
1532+spoolDir: tempDir,
1533+ createWorker,
1534+drainIntervalMs: 100,
1535+},
1536+});
1537+1538+const runPromise = session.runUntilAbort();
1539+await vi.waitFor(() => expect(createWorker).toHaveBeenCalledTimes(2));
1540+expectLogIncludes(log, "isolated polling ingress failed: worker crashed");
1541+expect(
1542+statusPatches(setStatus).some(
1543+(patch) => patch.connected === false && patch.lastError === "worker crashed",
1544+),
1545+).toBe(true);
1546+1547+abort.abort();
1548+await vi.advanceTimersByTimeAsync(20_000);
1549+await runPromise;
1550+} finally {
1551+vi.useRealTimers();
1552+await fs.rm(tempDir, { recursive: true, force: true });
1553+}
1554+});
1555+1556+it("treats isolated ingress worker rejection after abort as clean shutdown", async () => {
1557+vi.useFakeTimers({ shouldAdvanceTime: true });
1558+const abort = new AbortController();
1559+const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-"));
1560+const log = vi.fn();
1561+createTelegramBotMock.mockImplementation(() => ({
1562+api: {
1563+deleteWebhook: vi.fn(async () => true),
1564+config: { use: vi.fn() },
1565+},
1566+init: vi.fn(async () => undefined),
1567+handleUpdate: vi.fn(async () => undefined),
1568+stop: vi.fn(async () => undefined),
1569+}));
1570+1571+let rejectWorker: ((err: Error) => void) | undefined;
1572+const workerDone = new Promise<void>((_resolve, reject) => {
1573+rejectWorker = reject;
1574+});
1575+const createWorker = vi.fn(() => ({
1576+onMessage: vi.fn(() => () => undefined),
1577+stop: vi.fn(async () => {
1578+rejectWorker?.(new Error("worker exited with code 1"));
1579+}),
1580+task: vi.fn(async () => {
1581+await workerDone;
1582+}),
1583+}));
1584+1585+try {
1586+const session = createPollingSession({
1587+abortSignal: abort.signal,
1588+ log,
1589+isolatedIngress: {
1590+enabled: true,
1591+spoolDir: tempDir,
1592+ createWorker,
1593+drainIntervalMs: 100,
1594+},
1595+});
1596+1597+const runPromise = session.runUntilAbort();
1598+await vi.waitFor(() => expect(createWorker).toHaveBeenCalledTimes(1));
1599+abort.abort();
1600+await vi.advanceTimersByTimeAsync(20_000);
1601+await runPromise;
1602+1603+expect(createWorker).toHaveBeenCalledTimes(1);
1604+expectLogExcludes(log, "isolated polling ingress failed");
1605+} finally {
1606+vi.useRealTimers();
1607+await fs.rm(tempDir, { recursive: true, force: true });
1608+}
1609+});
1610+1611+it("propagates fatal isolated ingress polling errors", async () => {
1612+vi.useFakeTimers({ shouldAdvanceTime: true });
1613+const abort = new AbortController();
1614+const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-"));
1615+const log = vi.fn();
1616+const setStatus = vi.fn();
1617+isRecoverableTelegramNetworkErrorMock.mockReturnValue(false);
1618+createTelegramBotMock.mockImplementation(() => ({
1619+api: {
1620+deleteWebhook: vi.fn(async () => true),
1621+config: { use: vi.fn() },
1622+},
1623+init: vi.fn(async () => undefined),
1624+handleUpdate: vi.fn(async () => undefined),
1625+stop: vi.fn(async () => undefined),
1626+}));
1627+1628+let listener: WorkerPollErrorListener | undefined;
1629+const createWorker = vi.fn(() => ({
1630+onMessage: vi.fn((next: WorkerPollErrorListener) => {
1631+listener = next;
1632+return () => undefined;
1633+}),
1634+stop: vi.fn(async () => undefined),
1635+task: vi.fn(async () => {
1636+listener?.({
1637+type: "poll-error",
1638+message: "Unauthorized",
1639+finishedAt: Date.now(),
1640+});
1641+throw new Error("Telegram ingress worker exited with code 1");
1642+}),
1643+}));
1644+1645+try {
1646+const session = createPollingSession({
1647+abortSignal: abort.signal,
1648+ log,
1649+ setStatus,
1650+isolatedIngress: {
1651+enabled: true,
1652+spoolDir: tempDir,
1653+ createWorker,
1654+drainIntervalMs: 100,
1655+},
1656+});
1657+1658+await expect(session.runUntilAbort()).rejects.toThrow("Unauthorized");
1659+1660+expect(createWorker).toHaveBeenCalledTimes(1);
1661+expectLogExcludes(log, "isolated polling ingress failed");
1662+expect(
1663+statusPatches(setStatus).some(
1664+(patch) => patch.connected === false && patch.lastError === "Unauthorized",
1665+),
1666+).toBe(true);
1667+} finally {
1668+abort.abort();
1669+vi.useRealTimers();
1670+await fs.rm(tempDir, { recursive: true, force: true });
1671+}
1672+});
1673+14831674it("keeps active spooled lanes blocked across account restarts", async () => {
14841675vi.useFakeTimers({ shouldAdvanceTime: true });
14851676const firstAbort = new AbortController();
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。