fix(telegram): restart isolated polling on getUpdates conflict and surface it in status

obviyus · obviyus · commit d4819948f37d · 2026-06-12T10:29:56.000+05:30
diff --git a/extensions/telegram/src/polling-session.test.ts b/extensions/telegram/src/polling-session.test.ts
@@ -104,6 +104,7 @@ type WorkerPollSuccessListener = (message: {
 type WorkerPollErrorListener = (message: {
   type: "poll-error";
   message: string;
+  errorCode?: number;
   finishedAt: number;
 }) => void;
 type WorkerMessageListener = (message: TelegramIngressWorkerMessage) => void;
@@ -2356,6 +2357,89 @@ describe("TelegramPollingSession", () => {
     }
   });
 
+  it("restarts isolated ingress on a getUpdates conflict instead of crashing the account", async () => {
+    const abort = new AbortController();
+    const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-"));
+    const log = vi.fn();
+    const setStatus = vi.fn();
+    // 409 conflicts are not "recoverable network errors"; the conflict branch
+    // must restart the cycle before that classifier is consulted.
+    isRecoverableTelegramNetworkErrorMock.mockReturnValue(false);
+    const deleteWebhook = vi.fn(async () => true);
+    createTelegramBotMock.mockImplementation(() => ({
+      api: {
+        deleteWebhook,
+        config: { use: vi.fn() },
+      },
+      init: vi.fn(async () => undefined),
+      handleUpdate: vi.fn(async () => undefined),
+      stop: vi.fn(async () => undefined),
+    }));
+    const transport1 = makeTelegramTransport();
+    const transport2 = makeTelegramTransport();
+    const createTelegramTransport = vi
+      .fn<() => ReturnType<typeof makeTelegramTransport>>()
+      .mockReturnValueOnce(transport2);
+
+    let workerCycle = 0;
+    let listener: WorkerPollErrorListener | undefined;
+    const createWorker = vi.fn(() => ({
+      onMessage: vi.fn((next: WorkerPollErrorListener) => {
+        listener = next;
+        return () => undefined;
+      }),
+      stop: vi.fn(async () => undefined),
+      task: vi.fn(async () => {
+        workerCycle += 1;
+        if (workerCycle === 1) {
+          listener?.({
+            type: "poll-error",
+            message: "Conflict: terminated by other getUpdates request",
+            errorCode: 409,
+            finishedAt: Date.now(),
+          });
+          throw new Error("Telegram ingress worker exited with code 1");
+        }
+        abort.abort();
+      }),
+    }));
+
+    try {
+      const session = createPollingSession({
+        abortSignal: abort.signal,
+        log,
+        setStatus,
+        telegramTransport: transport1,
+        createTelegramTransport,
+        isolatedIngress: {
+          enabled: true,
+          spoolDir: tempDir,
+          createWorker,
+          drainIntervalMs: 100,
+        },
+      });
+
+      await session.runUntilAbort();
+
+      expect(createWorker).toHaveBeenCalledTimes(2);
+      // The conflict resets webhook cleanup so the next cycle re-runs deleteWebhook.
+      expect(deleteWebhook).toHaveBeenCalledTimes(2);
+      // The conflict marks the transport dirty so the next cycle gets a fresh socket.
+      expect(createTelegramTransport).toHaveBeenCalledTimes(1);
+      expectLogIncludes(log, "Another OpenClaw gateway, script, or Telegram poller");
+      expect(
+        statusPatches(setStatus).some(
+          (patch) =>
+            patch.connected === false &&
+            String(patch.lastError).includes("Another OpenClaw gateway"),
+        ),
+      ).toBe(true);
+    } finally {
+      abort.abort();
+      await fs.rm(tempDir, { recursive: true, force: true });
+    }
+  });
+
   it("keeps active spooled lanes blocked across account restarts", async () => {
     vi.useFakeTimers({ shouldAdvanceTime: true });
     const firstAbort = new AbortController();
@@ -3533,9 +3617,11 @@ describe("TelegramPollingSession", () => {
     const watchdogHarness = installPollingStallWatchdogHarness();
 
     const log = vi.fn();
+    const setStatus = vi.fn();
     const session = createPollingSession({
       abortSignal: abort.signal,
       log,
+      setStatus,
     });
 
     try {
@@ -3567,6 +3653,14 @@ describe("TelegramPollingSession", () => {
       abort.abort();
       resolveFirstTask();
       await runPromise;
+
+      // The stall must reach channel status, not just the gateway log.
+      expect(
+        statusPatches(setStatus).some(
+          (patch) =>
+            patch.connected === false && String(patch.lastError).includes("Polling stall detected"),
+        ),
+      ).toBe(true);
     } finally {
       watchdogHarness.restore();
     }
@@ -3614,6 +3708,7 @@ describe("TelegramPollingSession", () => {
   it("logs an actionable duplicate-poller hint for getUpdates conflicts", async () => {
     const abort = new AbortController();
     const log = vi.fn();
+    const setStatus = vi.fn();
     const conflictError = Object.assign(
       new Error("Conflict: terminated by other getUpdates request"),
       {
@@ -3628,11 +3723,19 @@ describe("TelegramPollingSession", () => {
     const session = createPollingSession({
       abortSignal: abort.signal,
       log,
+      setStatus,
     });
 
     await session.runUntilAbort();
 
     expectLogIncludes(log, "Another OpenClaw gateway, script, or Telegram poller");
+    // The hint must reach channel status, not just the gateway log.
+    expect(
+      statusPatches(setStatus).some(
+        (patch) =>
+          patch.connected === false && String(patch.lastError).includes("Another OpenClaw gateway"),
+      ),
+    ).toBe(true);
   });
 
   it("logs polling cycle start after a transport rebuild", async () => {
diff --git a/extensions/telegram/src/polling-session.ts b/extensions/telegram/src/polling-session.ts
@@ -110,6 +110,11 @@ function resolveTelegramRestartDelayMs(
   return { delayMs, stopTimeoutSuffix };
 }
 
+// Surfaced in logs and channel status when getUpdates returns 409; the only
+// user-fixable causes are a second poller on the same token or a stale webhook.
+const TELEGRAM_GET_UPDATES_CONFLICT_HINT =
+  " Another OpenClaw gateway, script, or Telegram poller may be using this bot token; stop the duplicate poller or switch this account to webhook mode.";
+
 const DEFAULT_POLL_STALL_THRESHOLD_MS = 120_000;
 const MIN_POLL_STALL_THRESHOLD_MS = 30_000;
 const MAX_POLL_STALL_THRESHOLD_MS = 600_000;
@@ -813,10 +818,12 @@ export class TelegramPollingSession {
       offset: number | null;
       outcome: string;
       error?: string;
+      errorCode: number | null;
     } = {
       startedAt: null,
       offset: null,
       outcome: "not-started",
+      errorCode: null,
     };
     const liveness = new TelegramPollingLivenessTracker();
     let consecutiveDrainFailures = 0;
@@ -853,6 +860,7 @@ export class TelegramPollingSession {
         pollState.offset = message.offset;
         pollState.outcome = "started";
         delete pollState.error;
+        pollState.errorCode = null;
         return;
       }
       if (message.type === "poll-success") {
@@ -871,6 +879,7 @@ export class TelegramPollingSession {
         liveness.noteGetUpdatesFinished();
         pollState.outcome = "error";
         pollState.error = message.message;
+        pollState.errorCode = message.errorCode ?? null;
         return;
       }
       if (message.type === "update") {
@@ -1002,14 +1011,24 @@ export class TelegramPollingSession {
         if (this.opts.abortSignal?.aborted) {
           return "exit";
         }
-        if (
+        // The worker only issues getUpdates, so a 409 is always a duplicate
+        // poller (or stale webhook) conflict. Mirror the classic polling
+        // cycle: re-clear the webhook, rotate the transport (#69787), and
+        // restart with backoff instead of crashing the whole account.
+        const isConflict = pollState.errorCode === 409;
+        if (isConflict) {
+          this.#webhookCleared = false;
+          this.#transportState.markDirty();
+        } else if (
           pollState.error &&
           !isRecoverableTelegramNetworkError(new Error(pollState.error), { context: "polling" })
         ) {
           this.#status.notePollingError(pollState.error);
           throw new Error(pollState.error, { cause: err });
         }
-        const message = formatErrorMessage(err);
+        const message = isConflict
+          ? `Telegram getUpdates conflict: ${pollState.error}.${TELEGRAM_GET_UPDATES_CONFLICT_HINT}`
+          : formatErrorMessage(err);
         this.opts.log(`[telegram][diag] isolated polling ingress failed: ${message}`);
         this.#status.notePollingError(message);
         clearForceCycleTimer();
@@ -1162,6 +1181,7 @@ export class TelegramPollingSession {
         this.#transportState.markDirty();
         stalledRestart = true;
         this.opts.log(`[telegram] ${stall.message}`);
+        this.#status.notePollingError(stall.message);
         requestStopForRestart();
       }
     }, POLL_WATCHDOG_INTERVAL_MS);
@@ -1210,12 +1230,16 @@ export class TelegramPollingSession {
       }
       const reason = isConflict ? "getUpdates conflict" : "network error";
       const errMsg = formatErrorMessage(err);
-      const conflictHint = isConflict
-        ? " Another OpenClaw gateway, script, or Telegram poller may be using this bot token; stop the duplicate poller or switch this account to webhook mode."
-        : "";
+      const conflictHint = isConflict ? TELEGRAM_GET_UPDATES_CONFLICT_HINT : "";
       this.opts.log(
         `[telegram][diag] polling cycle error reason=${reason} ${liveness.formatDiagnosticFields("lastGetUpdatesError")} err=${errMsg}${conflictHint}`,
       );
+      // Conflicts carry a user-fixable diagnosis, so surface them in channel
+      // status. Recoverable network blips stay log-only; the stall watchdog
+      // owns status for extended outages (see detectStall above).
+      if (isConflict) {
+        this.#status.notePollingError(`Telegram ${reason}: ${errMsg}.${conflictHint}`);
+      }
       clearForceCycleTimer();
       const shouldRestart = await this.#waitBeforeRestart(
         (delay) => `Telegram ${reason}: ${errMsg};${conflictHint} retrying in ${delay}.`,
