fix(telegram): age-gate retry-limit spool tombstones (#98776)

obviyus · obviyus · commit 55d31beeef00 · 2026-07-01T15:32:00.000-07:00
diff --git a/extensions/telegram/src/polling-session.test.ts b/extensions/telegram/src/polling-session.test.ts
@@ -477,9 +477,10 @@ async function waitForTestReplyFenceAbort(params: { key: string; laneKey: string
 async function writeSpooledTestUpdates(
   spoolDir: string,
   updates: readonly TestTelegramUpdate[],
+  options?: { now?: number },
 ): Promise<void> {
   for (const update of updates) {
-    await writeTelegramSpooledUpdate({ spoolDir, update });
+    await writeTelegramSpooledUpdate({ spoolDir, update, now: options?.now });
   }
 }
 
@@ -829,6 +830,47 @@ describe("TelegramPollingSession", () => {
         1_999,
       ),
     ).toBe(0);
+    expect(
+      pollingSessionTesting.resolveSpooledUpdateRetryDelayMs(
+        {
+          updateId: 44,
+          path: "/tmp/44.json",
+          update: { update_id: 44 },
+          receivedAt: 0,
+          attempts: pollingSessionTesting.spooledRetryMaxAttempts,
+          lastAttemptAt: 1_000,
+          lastError: "state store outage",
+        },
+        1_999,
+      ),
+    ).toBeGreaterThan(0);
+  });
+
+  it("keeps generic retryable failures pending until they are old enough to dead-letter", () => {
+    const update = {
+      updateId: 42,
+      path: "/tmp/42.json",
+      update: { update_id: 42 },
+      receivedAt: 1_000,
+      attempts: pollingSessionTesting.spooledRetryMaxAttempts - 1,
+      lastAttemptAt: 2_000,
+      lastError: "state store outage",
+    };
+
+    expect(
+      pollingSessionTesting.shouldDeadLetterRetryableSpooledUpdate(
+        update,
+        pollingSessionTesting.spooledRetryMaxAttempts,
+        1_000 + pollingSessionTesting.spooledRetryDeadLetterMinAgeMs - 1,
+      ),
+    ).toBe(false);
+    expect(
+      pollingSessionTesting.shouldDeadLetterRetryableSpooledUpdate(
+        update,
+        pollingSessionTesting.spooledRetryMaxAttempts,
+        1_000 + pollingSessionTesting.spooledRetryDeadLetterMinAgeMs,
+      ),
+    ).toBe(true);
   });
 
   it("does not call getUpdates for offset confirmation (avoiding 409 conflicts)", async () => {
@@ -2093,10 +2135,13 @@ describe("TelegramPollingSession", () => {
         const log = vi.fn();
         const events: string[] = [];
         let poisonAttempts = 0;
-        await writeSpooledTestUpdates(tempDir, [
-          topicUpdate(42, 10, "poison"),
-          topicUpdate(44, 10, "after poison"),
-        ]);
+        await writeSpooledTestUpdates(
+          tempDir,
+          [topicUpdate(42, 10, "poison"), topicUpdate(44, 10, "after poison")],
+          {
+            now: Date.now() - pollingSessionTesting.spooledRetryDeadLetterMinAgeMs,
+          },
+        );
 
         const { runPromise, stopWorker } = startIsolatedIngressSession({
           abort,
diff --git a/extensions/telegram/src/polling-session.ts b/extensions/telegram/src/polling-session.ts
@@ -138,6 +138,7 @@ const TELEGRAM_SPOOLED_CLAIM_HEALTH_GRACE_MS = 2 * TELEGRAM_SPOOLED_CLAIM_REFRES
 const TELEGRAM_SPOOLED_RETRY_MAX_ATTEMPTS = 8;
 const TELEGRAM_SPOOLED_RETRY_BASE_MS = 1_000;
 const TELEGRAM_SPOOLED_RETRY_MAX_MS = 3 * 60_000;
+const TELEGRAM_SPOOLED_RETRY_DEAD_LETTER_MIN_AGE_MS = 24 * 60 * 60 * 1000;
 const TELEGRAM_POLLING_CLIENT_TIMEOUT_FLOOR_SECONDS = Math.ceil(
   TELEGRAM_GET_UPDATES_REQUEST_TIMEOUT_MS / 1000,
 );
@@ -178,12 +179,7 @@ function resolveNonRetryableSpooledUpdateFailure(
 
 function resolveSpooledUpdateRetryDelayMs(update: TelegramSpooledUpdate, now = Date.now()): number {
   const attempts = update.attempts ?? 0;
-  if (
-    !update.lastError ||
-    update.lastAttemptAt === undefined ||
-    attempts <= 0 ||
-    attempts >= TELEGRAM_SPOOLED_RETRY_MAX_ATTEMPTS
-  ) {
+  if (!update.lastError || update.lastAttemptAt === undefined || attempts <= 0) {
     return 0;
   }
   const exponent = Math.min(attempts - 1, 8);
@@ -198,6 +194,17 @@ function resolveSpooledUpdateAttemptNumber(update: TelegramSpooledUpdate): numbe
   return (update.attempts ?? 0) + 1;
 }
 
+function shouldDeadLetterRetryableSpooledUpdate(
+  update: TelegramSpooledUpdate,
+  attempt: number,
+  now = Date.now(),
+): boolean {
+  return (
+    attempt >= TELEGRAM_SPOOLED_RETRY_MAX_ATTEMPTS &&
+    now - update.receivedAt >= TELEGRAM_SPOOLED_RETRY_DEAD_LETTER_MIN_AGE_MS
+  );
+}
+
 type TelegramBot = ReturnType<typeof createTelegramBot>;
 
 const waitForGracefulStop = async (stop: () => Promise<void>) => {
@@ -843,7 +850,7 @@ export class TelegramPollingSession {
       }
     }
     const attempt = resolveSpooledUpdateAttemptNumber(params.update);
-    if (attempt >= TELEGRAM_SPOOLED_RETRY_MAX_ATTEMPTS) {
+    if (shouldDeadLetterRetryableSpooledUpdate(params.update, attempt)) {
       const message = formatErrorMessage(params.err);
       try {
         const failed = await failTelegramSpooledUpdateClaim({
@@ -857,8 +864,8 @@ export class TelegramPollingSession {
           );
           return;
         }
-        // Retryable poison updates must eventually become tombstones so the
-        // lowest update id stops blocking every later turn in the same lane.
+        // Retryable poison updates must eventually become tombstones, but not
+        // during ordinary transient provider or state-store outages.
         this.opts.log(
           `[telegram][warn] spooled update ${params.update.updateId} on lane ${laneKey} reached retry limit after ${attempt} attempts; dead-lettered: ${message}`,
         );
@@ -1718,7 +1725,9 @@ export const testing = {
   resetTelegramRestartBackoffState,
   resolveTelegramRestartDelayMs,
   resolveSpooledUpdateRetryDelayMs,
+  shouldDeadLetterRetryableSpooledUpdate,
   spooledRetryMaxAttempts: TELEGRAM_SPOOLED_RETRY_MAX_ATTEMPTS,
+  spooledRetryDeadLetterMinAgeMs: TELEGRAM_SPOOLED_RETRY_DEAD_LETTER_MIN_AGE_MS,
   isolatedIngressBacklogStallMs: ISOLATED_INGRESS_BACKLOG_STALL_MS,
   spooledClaimRefreshIntervalMs: TELEGRAM_SPOOLED_CLAIM_REFRESH_INTERVAL_MS,
   resolveSpooledUpdateHandlerAbortGraceMs: (valueMs: unknown): number =>
