fix(telegram): release stuck ingress claims on timeout instead of dead-lettering

lzyyzznl · lzyyzznl · commit a7316b0555da · 2026-06-22T04:58:10.000+08:00
- Reduce ISOLATED_INGRESS_BACKLOG_STALL_MS from 25 min to 5 min so stuck handlers are detected sooner - Replace failTelegramSpooledUpdateClaim with releaseTelegramSpooledUpdateClaim in recoverTimedOutSpooledHandler so the update is requeued (attempts++) and retried rather than permanently dead-lettered - Add options.lastError to releaseTelegramSpooledUpdateClaim so the timeout reason is recorded in the queue row for diagnostics Previously a hanging handler held its claim indefinitely (>25 min before timeout detection), blocked all subsequent same-lane updates, and when finally detected dead-lettered the update permanently — requiring manual SQLite intervention to restore Telegram inbound processing. Fixes #95350
diff --git a/extensions/telegram/src/polling-session.test.ts b/extensions/telegram/src/polling-session.test.ts
@@ -1881,12 +1881,10 @@ describe("TelegramPollingSession", () => {
       const secondRunPromise = secondSession.runUntilAbort();
 
       await vi.advanceTimersByTimeAsync(1_000);
-      await vi.waitFor(async () => expect(await failedUpdateIds(tempDir)).toEqual([42]));
+      await vi.waitFor(async () => expect(await pendingUpdateIds(tempDir, "all")).toEqual([42]));
       await vi.waitFor(() =>
         expect(
-          log.mock.calls.some(([line]) =>
-            String(line).includes("timed out spooled update 42 did not stop"),
-          ),
+          log.mock.calls.some(([line]) => String(line).includes("timed out spooled update 42")),
         ).toBe(true),
       );
       expect(handleUpdate).toHaveBeenCalledTimes(1);
@@ -2820,7 +2818,7 @@ describe("TelegramPollingSession", () => {
     }
   });
 
-  it("fails a timed-out spooled handler and restarts before draining later same-lane updates", async () => {
+  it("releases a timed-out spooled handler for retry and restarts before draining later same-lane updates", async () => {
     vi.useFakeTimers({ shouldAdvanceTime: true });
     const abort = new AbortController();
     const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-"));
@@ -2881,14 +2879,16 @@ describe("TelegramPollingSession", () => {
 
       await vi.advanceTimersByTimeAsync(1_000);
       await vi.waitFor(() => expect(worker.createWorker).toHaveBeenCalledTimes(2));
-      await vi.waitFor(() => expect(events).toEqual(["first:42", "second:43"]));
+      // After timeout the claim is released back to pending, so the second bot
+      // retries update 42 instead of skipping to update 43.
+      await vi.waitFor(() => expect(events).toEqual(["first:42", "second:42"]));
       await runPromise;
 
       expect(createTelegramBotMock).toHaveBeenCalledTimes(2);
       expect(firstBot.stop).toHaveBeenCalledTimes(1);
       expect(secondBot.stop).toHaveBeenCalledTimes(1);
-      expect(await pendingUpdateIds(tempDir, "all")).toEqual([]);
-      expect(await failedUpdateIds(tempDir)).toEqual([42]);
+      expect(await pendingUpdateIds(tempDir, "all")).toEqual([43]);
+      expect(await failedUpdateIds(tempDir)).toEqual([]);
       expectLogIncludes(log, "spool handler timed out behind update 42");
     } finally {
       abort.abort();
@@ -2949,8 +2949,8 @@ describe("TelegramPollingSession", () => {
 
       expect(worker.createWorker).toHaveBeenCalledTimes(1);
       expect(events).toEqual(["first:42"]);
-      expect(await failedUpdateIds(tempDir)).toEqual([42]);
-      expect(await pendingUpdateIds(tempDir, "all")).toEqual([43]);
+      expect(await failedUpdateIds(tempDir)).toEqual([]);
+      expect(await pendingUpdateIds(tempDir, "all")).toEqual([42, 43]);
 
       releaseFirstTurn?.();
       abort.abort();
@@ -3103,11 +3103,13 @@ describe("TelegramPollingSession", () => {
 
       await vi.advanceTimersByTimeAsync(500);
       expect(events).toEqual(["first:42"]);
-      expect(await pendingUpdateIds(tempDir, "all")).toEqual([43]);
+      // The timed-out update 42 is released back to pending, so it remains
+      // in the queue alongside update 43.
+      expect(await pendingUpdateIds(tempDir, "all")).toEqual([42, 43]);
 
       releaseFirstWorker?.();
       await vi.waitFor(() => expect(createWorker).toHaveBeenCalledTimes(2));
-      await vi.waitFor(() => expect(events).toEqual(["first:42", "second:43"]));
+      await vi.waitFor(() => expect(events).toEqual(["first:42", "second:42"]));
       await runPromise;
     } finally {
       abort.abort();
@@ -3118,7 +3120,7 @@ describe("TelegramPollingSession", () => {
     }
   });
 
-  it("keeps a timed-out lane guarded when its failed state cannot be written", async () => {
+  it("keeps a timed-out lane guarded when its release for retry cannot be written", async () => {
     vi.useFakeTimers({ shouldAdvanceTime: true });
     const abort = new AbortController();
     const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-"));
@@ -3130,8 +3132,8 @@ describe("TelegramPollingSession", () => {
       releaseRegularTurn = resolve;
     });
     const spoolModule = await import("./telegram-ingress-spool.js");
-    const failSpy = vi
-      .spyOn(spoolModule, "failTelegramSpooledUpdateClaim")
+    const releaseSpy = vi
+      .spyOn(spoolModule, "releaseTelegramSpooledUpdateClaim")
       .mockRejectedValueOnce(new Error("disk full"));
     createTelegramBotMock.mockReturnValueOnce({
       api: {
@@ -3184,12 +3186,13 @@ describe("TelegramPollingSession", () => {
       const runPromise = session.runUntilAbort();
       await vi.waitFor(() => expect(events).toEqual(["handled:42"]));
       await vi.advanceTimersByTimeAsync(150);
-      await vi.waitFor(() => expectLogIncludes(log, "could not be marked failed: disk full"));
+      await vi.waitFor(() => expectLogIncludes(log, "could not be released for retry: disk full"));
 
       await vi.advanceTimersByTimeAsync(500);
       expect(createWorker).toHaveBeenCalledTimes(1);
       expect(events).toEqual(["handled:42"]);
       expect(await failedUpdateIds(tempDir)).toEqual([]);
+      // Release threw, so the claim stays claimed (not released back to pending).
       expect(await pendingUpdateIds(tempDir, "all")).toEqual([43]);
       expect(
         (await listTelegramSpooledUpdateClaims({ spoolDir: tempDir })).map(
@@ -3213,7 +3216,7 @@ describe("TelegramPollingSession", () => {
       await vi.advanceTimersByTimeAsync(20_000);
       await runPromise;
     } finally {
-      failSpy.mockRestore();
+      releaseSpy.mockRestore();
       releaseRegularTurn?.();
       abort.abort();
       stopWorker?.();
@@ -3323,9 +3326,11 @@ describe("TelegramPollingSession", () => {
       // lone active handler on the same lane, hangs the same way, and is now also
       // recovered on timeout rather than stranded with no backlog behind it (#84158).
       // Each recovery restarts ingress, so the worker is created once more per
-      // recovered handler (initial + two restarts).
-      await vi.waitFor(async () => expect(await failedUpdateIds(tempDir)).toEqual([42, 43]));
-      expect(createWorker).toHaveBeenCalledTimes(3);
+      // recovered handler (initial + two restarts). The timed-out updates are
+      // released back to pending (not failed) so they can be retried. After
+      // restart the drain re-claims update 42, leaving only 43 pending.
+      await vi.waitFor(async () => expect(await pendingUpdateIds(tempDir, "all")).toEqual([43]));
+      expect(createWorker.mock.calls.length).toBeGreaterThanOrEqual(3);
 
       abort.abort();
       await vi.advanceTimersByTimeAsync(20_000);
diff --git a/extensions/telegram/src/polling-session.ts b/extensions/telegram/src/polling-session.ts
@@ -126,7 +126,7 @@ const MIN_POLL_STALL_THRESHOLD_MS = 30_000;
 const MAX_POLL_STALL_THRESHOLD_MS = 600_000;
 const POLL_WATCHDOG_INTERVAL_MS = 30_000;
 const POLL_STOP_GRACE_MS = 15_000;
-const ISOLATED_INGRESS_BACKLOG_STALL_MS = 25 * 60_000;
+const ISOLATED_INGRESS_BACKLOG_STALL_MS = 5 * 60_000;
 const TELEGRAM_SPOOLED_HANDLER_ABORT_GRACE_MS = 5_000;
 const TELEGRAM_SPOOLED_HANDLER_TIMEOUT_ENV = "OPENCLAW_TELEGRAM_SPOOLED_HANDLER_TIMEOUT_MS";
 const TELEGRAM_SPOOLED_DRAIN_START_LIMIT = 100;
@@ -886,24 +886,19 @@ export class TelegramPollingSession {
     }
     const age = formatDurationPrecise(timedOutHandler.ageMs);
     activeHandler.timedOutAt = Date.now();
-    const message = `Telegram isolated polling spool handler timed out behind update ${handler.updateId} on lane ${handler.laneKey} after ${age}; marking the update failed, aborting active reply work, and restarting isolated ingress so later updates can drain.`;
+    const message = `Telegram isolated polling spool handler timed out behind update ${handler.updateId} on lane ${handler.laneKey} after ${age}; releasing the claim for retry, aborting active reply work, and restarting isolated ingress so later updates can drain.`;
     activeHandler.timeoutMessage = message;
     try {
-      const failed = await failTelegramSpooledUpdateClaim({
-        update: handler.update,
-        reason: "handler-timeout",
-        message,
+      // Release the claim back to pending so the drain loop can retry it,
+      // instead of dead-lettering with a permanent failure. The attempts
+      // counter on the queue record is incremented on each release, so
+      // repeated timeouts are observable via SQLite diagnostics.
+      await releaseTelegramSpooledUpdateClaim(handler.update, {
+        lastError: message,
       });
-      if (!failed) {
-        this.opts.log(
-          `[telegram][diag] timed out spooled update ${handler.updateId} no longer had a processing marker to fail.`,
-        );
-        this.#status.notePollingError(message);
-        return { handlerKey: handler.handlerKey, restart: false };
-      }
     } catch (err) {
       this.opts.log(
-        `[telegram][diag] timed out spooled update ${handler.updateId} could not be marked failed: ${formatErrorMessage(err)}`,
+        `[telegram][diag] timed out spooled update ${handler.updateId} could not be released for retry: ${formatErrorMessage(err)}`,
       );
       this.#status.notePollingError(message);
       return { handlerKey: handler.handlerKey, restart: false };
diff --git a/extensions/telegram/src/telegram-ingress-spool.ts b/extensions/telegram/src/telegram-ingress-spool.ts
@@ -267,9 +267,11 @@ export async function claimTelegramSpooledUpdate(
 
 export async function releaseTelegramSpooledUpdateClaim(
   update: ClaimedTelegramSpooledUpdate,
+  options?: { lastError?: string },
 ): Promise<void> {
   await createTelegramIngressQueue(path.dirname(update.pendingPath)).release(
     queueMutationTarget(update),
+    { lastError: options?.lastError },
   );
 }
 
