@@ -418,7 +418,7 @@ type TestTelegramUpdate = {
418418update_id: number;
419419message: {
420420text: string;
421-chat: { id: number; type: "supergroup" };
421+chat: { id: number; type: "private" | "supergroup" };
422422message_thread_id?: number;
423423is_topic_message?: boolean;
424424};
@@ -436,6 +436,16 @@ function topicUpdate(updateId: number, threadId: number, text: string): TestTele
436436};
437437}
438438439+function directUpdate(updateId: number, chatId: number, text: string): TestTelegramUpdate {
440+return {
441+update_id: updateId,
442+message: {
443+ text,
444+chat: { id: chatId, type: "private" },
445+},
446+};
447+}
448+439449async function waitForAbortSignal(signal: AbortSignal): Promise<void> {
440450if (signal.aborted) {
441451return;
@@ -1795,6 +1805,93 @@ describe("TelegramPollingSession", () => {
17951805});
17961806});
179718071808+for (const scenario of [
1809+{
1810+name: "topic",
1811+conflict: topicUpdate(42, 10, "retryable session init conflict"),
1812+blocked: topicUpdate(43, 10, "same topic must wait behind retry backoff"),
1813+other: topicUpdate(44, 11, "other topic can continue"),
1814+conflictEvent: "topic10:conflict",
1815+blockedEvent: "topic10:overtook",
1816+otherEvent: "topic11",
1817+error: "reply session initialization conflicted for agent:main:telegram:group:-100:topic:10",
1818+},
1819+{
1820+name: "direct message",
1821+conflict: directUpdate(42, 100, "retryable session init conflict"),
1822+blocked: directUpdate(43, 100, "same DM must wait behind retry backoff"),
1823+other: directUpdate(44, 101, "other DM can continue"),
1824+conflictEvent: "dm100:conflict",
1825+blockedEvent: "dm100:overtook",
1826+otherEvent: "dm101",
1827+error: "reply session initialization conflicted for agent:main:telegram:direct:100",
1828+},
1829+]) {
1830+it(`backs off retryable reply session init conflicts for ${scenario.name} lanes`, async () => {
1831+vi.useFakeTimers({ shouldAdvanceTime: true });
1832+try {
1833+await withTempSpool(async (tempDir) => {
1834+const abort = new AbortController();
1835+const log = vi.fn();
1836+let attempts = 0;
1837+const events: string[] = [];
1838+await writeSpooledTestUpdates(tempDir, [
1839+scenario.conflict,
1840+scenario.blocked,
1841+scenario.other,
1842+]);
1843+1844+const { runPromise, stopWorker } = startIsolatedIngressSession({
1845+ abort,
1846+spoolDir: tempDir,
1847+ log,
1848+drainIntervalMs: 100,
1849+handleUpdate: async (update) => {
1850+if (update.update_id === scenario.conflict.update_id) {
1851+attempts += 1;
1852+events.push(`${scenario.conflictEvent}:${attempts}`);
1853+throw new Error(scenario.error);
1854+}
1855+if (update.update_id === scenario.blocked.update_id) {
1856+events.push(scenario.blockedEvent);
1857+return;
1858+}
1859+if (update.update_id === scenario.other.update_id) {
1860+events.push(scenario.otherEvent);
1861+}
1862+},
1863+});
1864+1865+await vi.waitFor(() => expect(attempts).toBe(1));
1866+await vi.advanceTimersByTimeAsync(1_000);
1867+expect(attempts).toBe(1);
1868+await vi.waitFor(() =>
1869+expect(events).toEqual([`${scenario.conflictEvent}:1`, scenario.otherEvent]),
1870+);
1871+expect(await pendingUpdateIds(tempDir, "all")).toEqual([
1872+scenario.conflict.update_id,
1873+scenario.blocked.update_id,
1874+]);
1875+expect(await failedUpdateIds(tempDir)).toEqual([]);
1876+1877+await vi.advanceTimersByTimeAsync(4_500);
1878+await vi.waitFor(() => expect(attempts).toBe(2));
1879+expect(events).not.toContain(scenario.blockedEvent);
1880+expectLogIncludes(
1881+log,
1882+`spooled update ${scenario.conflict.update_id} failed; keeping for retry`,
1883+);
1884+1885+abort.abort();
1886+stopWorker();
1887+await runPromise;
1888+});
1889+} finally {
1890+vi.useRealTimers();
1891+}
1892+});
1893+}
1894+17981895it("dead-letters wrapped missing harness failures", async () => {
17991896await withTempSpool(async (tempDir) => {
18001897const abort = new AbortController();
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。