Skip to content

Commit 13ecca5

Browse files
VACIncobviyus
VACInc
authored and
obviyus
committed
fix(telegram): back off session init spool retries
1 parent c68484a commit 13ecca5

3 files changed

Lines changed: 135 additions & 2 deletions

File tree

‎extensions/telegram/src/polling-session.test.ts‎

Lines changed: 98 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -418,7 +418,7 @@ type TestTelegramUpdate = {
418418
update_id: number;
419419
message: {
420420
text: string;
421-
chat: { id: number; type: "supergroup" };
421+
chat: { id: number; type: "private" | "supergroup" };
422422
message_thread_id?: number;
423423
is_topic_message?: boolean;
424424
};
@@ -436,6 +436,16 @@ function topicUpdate(updateId: number, threadId: number, text: string): TestTele
436436
};
437437
}
438438

439+
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+
439449
async function waitForAbortSignal(signal: AbortSignal): Promise<void> {
440450
if (signal.aborted) {
441451
return;
@@ -1795,6 +1805,93 @@ describe("TelegramPollingSession", () => {
17951805
});
17961806
});
17971807

1808+
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+
17981895
it("dead-letters wrapped missing harness failures", async () => {
17991896
await withTempSpool(async (tempDir) => {
18001897
const abort = new AbortController();

‎extensions/telegram/src/polling-session.ts‎

Lines changed: 29 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -131,11 +131,14 @@ const TELEGRAM_SPOOLED_HANDLER_ABORT_GRACE_MS = 5_000;
131131
const TELEGRAM_SPOOLED_HANDLER_TIMEOUT_ENV = "OPENCLAW_TELEGRAM_SPOOLED_HANDLER_TIMEOUT_MS";
132132
const TELEGRAM_SPOOLED_DRAIN_START_LIMIT = 100;
133133
const TELEGRAM_SPOOLED_DRAIN_SCAN_LIMIT = TELEGRAM_SPOOLED_DRAIN_START_LIMIT * 10;
134+
const TELEGRAM_SPOOLED_SESSION_INIT_CONFLICT_RETRY_BASE_MS = 5_000;
135+
const TELEGRAM_SPOOLED_SESSION_INIT_CONFLICT_RETRY_MAX_MS = 60_000;
134136
const TELEGRAM_POLLING_CLIENT_TIMEOUT_FLOOR_SECONDS = Math.ceil(
135137
TELEGRAM_GET_UPDATES_REQUEST_TIMEOUT_MS / 1000,
136138
);
137139
const MISSING_AGENT_HARNESS_ERROR_NAME = "MissingAgentHarnessError";
138140
const MISSING_AGENT_HARNESS_MESSAGE_RE = /Requested agent harness "[^"]+" is not registered\./u;
141+
const REPLY_SESSION_INIT_CONFLICT_MESSAGE_RE = /reply session initialization conflicted for \S+/u;
139142

140143
function normalizeTelegramAccountId(accountId?: string | null): string {
141144
return accountId?.trim() || "default";
@@ -169,6 +172,24 @@ function resolveNonRetryableSpooledUpdateFailure(
169172
return null;
170173
}
171174

175+
function resolveSpooledUpdateRetryDelayMs(update: TelegramSpooledUpdate, now = Date.now()): number {
176+
const attempts = update.attempts ?? 0;
177+
if (
178+
!update.lastError ||
179+
!REPLY_SESSION_INIT_CONFLICT_MESSAGE_RE.test(update.lastError) ||
180+
update.lastAttemptAt === undefined ||
181+
attempts <= 0
182+
) {
183+
return 0;
184+
}
185+
const exponent = Math.min(attempts - 1, 8);
186+
const delayMs = Math.min(
187+
TELEGRAM_SPOOLED_SESSION_INIT_CONFLICT_RETRY_MAX_MS,
188+
TELEGRAM_SPOOLED_SESSION_INIT_CONFLICT_RETRY_BASE_MS * 2 ** exponent,
189+
);
190+
return Math.max(0, update.lastAttemptAt + delayMs - now);
191+
}
192+
172193
type TelegramBot = ReturnType<typeof createTelegramBot>;
173194

174195
const waitForGracefulStop = async (stop: () => Promise<void>) => {
@@ -777,7 +798,9 @@ export class TelegramPollingSession {
777798
}
778799
}
779800
try {
780-
await releaseTelegramSpooledUpdateClaim(params.update);
801+
await releaseTelegramSpooledUpdateClaim(params.update, {
802+
lastError: formatErrorMessage(params.err),
803+
});
781804
} catch (releaseErr) {
782805
this.opts.log(
783806
`[telegram][diag] spooled update ${params.update.updateId} failed and could not be requeued: ${formatErrorMessage(releaseErr)}`,
@@ -865,6 +888,10 @@ export class TelegramPollingSession {
865888
if (this.opts.abortSignal?.aborted) {
866889
break;
867890
}
891+
if (resolveSpooledUpdateRetryDelayMs(update) > 0) {
892+
claimedLaneKeys.add(laneKey);
893+
continue;
894+
}
868895
const handlerKey = buildSpooledUpdateHandlerKey({ spoolDir: params.spoolDir, laneKey });
869896
if (activeSpooledUpdateHandlersByLane.has(handlerKey)) {
870897
blockedByLane.add(handlerKey);
@@ -1533,6 +1560,7 @@ export const testing = {
15331560
createTelegramRestartBackoffState,
15341561
resetTelegramRestartBackoffState,
15351562
resolveTelegramRestartDelayMs,
1563+
resolveSpooledUpdateRetryDelayMs,
15361564
resolveSpooledUpdateHandlerAbortGraceMs: (valueMs: unknown): number =>
15371565
resolvePositiveTimerTimeoutMs(valueMs, TELEGRAM_SPOOLED_HANDLER_ABORT_GRACE_MS),
15381566
};

‎extensions/telegram/src/telegram-ingress-spool.ts‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,9 @@ export type TelegramSpooledUpdate = {
3838
path: string;
3939
update: unknown;
4040
receivedAt: number;
41+
attempts?: number;
42+
lastAttemptAt?: number;
43+
lastError?: string;
4144
claim?: TelegramSpooledUpdateClaimOwner;
4245
};
4346

@@ -166,6 +169,9 @@ function parseQueueRecord(
166169
path: pendingPath(spoolDir, payload.updateId),
167170
update: payload.update,
168171
receivedAt: payload.receivedAt,
172+
attempts: record.attempts,
173+
...(record.lastAttemptAt === undefined ? {} : { lastAttemptAt: record.lastAttemptAt }),
174+
...(record.lastError === undefined ? {} : { lastError: record.lastError }),
169175
};
170176
}
171177

@@ -267,9 +273,11 @@ export async function claimTelegramSpooledUpdate(
267273

268274
export async function releaseTelegramSpooledUpdateClaim(
269275
update: ClaimedTelegramSpooledUpdate,
276+
options?: { lastError?: string; releasedAt?: number },
270277
): Promise<void> {
271278
await createTelegramIngressQueue(path.dirname(update.pendingPath)).release(
272279
queueMutationTarget(update),
280+
options,
273281
);
274282
}
275283

0 commit comments

Comments
 (0)