fix(telegram): drain outbound queue after polling reconnect

steipete · steipete · commit 3b7f3dbd6b6e · 2026-05-15T17:53:57.000+01:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -12,6 +12,7 @@ Docs: https://docs.openclaw.ai
 ### Fixes
 
 - Agents/media: preserve message-tool-only delivery for generated music and video completion handoffs, so group/channel completions do not finish without posting the generated attachment.
+- Telegram: drain queued outbound deliveries after polling reconnect confirms fresh `getUpdates` activity, so stale-socket and network recovery do not leave failed replies stranded. Fixes #50040. Refs #82175. Thanks @dmitriiforpost-commits and @shellyrocklobster.
 - LINE: acknowledge signed webhook events before agent processing so slow model replies do not cause LINE `request_timeout` delivery failures. Fixes #65375. Thanks @myericho.
 - TTS: preserve channel-derived voice-note delivery for `/tts audio` replies even when the provider output is not natively voice-compatible. (#82174) Thanks @xuruiray.
 - Codex/Lossless: keep Codex explicit compaction on native app-server threads while allowing Lossless through the context-engine slot; `openclaw doctor --fix` now migrates legacy `compaction.provider: "lossless-claw"` config to `plugins.slots.contextEngine`.
diff --git a/extensions/telegram/src/polling-session.test.ts b/extensions/telegram/src/polling-session.test.ts
@@ -9,6 +9,7 @@ const createTelegramBotMock = vi.hoisted(() => vi.fn());
 const isRecoverableTelegramNetworkErrorMock = vi.hoisted(() => vi.fn(() => true));
 const computeBackoffMock = vi.hoisted(() => vi.fn(() => 0));
 const sleepWithAbortMock = vi.hoisted(() => vi.fn(async () => undefined));
+const drainPendingDeliveriesMock = vi.hoisted(() => vi.fn(async (_opts: unknown) => undefined));
 
 vi.mock("@grammyjs/runner", () => ({
   run: runMock,
@@ -22,6 +23,10 @@ vi.mock("./network-errors.js", () => ({
   isRecoverableTelegramNetworkError: isRecoverableTelegramNetworkErrorMock,
 }));
 
+vi.mock("openclaw/plugin-sdk/delivery-queue-runtime", () => ({
+  drainPendingDeliveries: drainPendingDeliveriesMock,
+}));
+
 vi.mock("./api-logging.js", () => ({
   withTelegramApiErrorLogging: async ({ fn }: { fn: () => Promise<unknown> }) => await fn(),
 }));
@@ -54,6 +59,18 @@ type TelegramApiMiddleware = (
   method: string,
   payload: unknown,
 ) => Promise<unknown>;
+type DrainPendingDeliveriesCall = {
+  drainKey: string;
+  logLabel: string;
+  selectEntry: (
+    entry: {
+      channel: string;
+      accountId?: string;
+      lastError?: string;
+    },
+    now: number,
+  ) => { match: boolean; bypassBackoff: boolean };
+};
 type AsyncVoidFn = () => Promise<void>;
 type MockCallSource = { mock: { calls: Array<Array<unknown>> } };
 
@@ -164,6 +181,14 @@ function expectTelegramBotTransportSequence(firstTransport: unknown, secondTrans
   expect(createTelegramBotMock.mock.calls.at(1)?.[0]?.telegramTransport).toBe(secondTransport);
 }
 
+function expectDrainPendingDeliveriesCall(index = 0): DrainPendingDeliveriesCall {
+  const call = drainPendingDeliveriesMock.mock.calls[index]?.[0];
+  if (!call || typeof call !== "object") {
+    throw new Error(`Expected drainPendingDeliveries call ${index}`);
+  }
+  return call as DrainPendingDeliveriesCall;
+}
+
 function makeTelegramTransport() {
   return {
     fetch: globalThis.fetch,
@@ -292,6 +317,7 @@ describe("TelegramPollingSession", () => {
     isRecoverableTelegramNetworkErrorMock.mockReset().mockReturnValue(true);
     computeBackoffMock.mockReset().mockReturnValue(0);
     sleepWithAbortMock.mockReset().mockResolvedValue(undefined);
+    drainPendingDeliveriesMock.mockReset().mockResolvedValue(undefined);
   });
 
   it("uses backoff helpers for recoverable polling retries", async () => {
@@ -454,6 +480,58 @@ describe("TelegramPollingSession", () => {
     }
   });
 
+  it("drains Telegram delivery queue after isolated ingress reports poll success", async () => {
+    const abort = new AbortController();
+    const init = vi.fn(async () => undefined);
+    const bot = {
+      api: {
+        deleteWebhook: vi.fn(async () => true),
+        config: { use: vi.fn() },
+      },
+      init,
+      handleUpdate: vi.fn(async () => undefined),
+      stop: vi.fn(async () => undefined),
+    };
+    createTelegramBotMock.mockReturnValueOnce(bot);
+    let onMessage:
+      | ((message: { type: "poll-success"; finishedAt: number; count: number }) => void)
+      | undefined;
+    let stopWorker: (() => void) | undefined;
+    const workerDone = new Promise<void>((resolve) => {
+      stopWorker = resolve;
+    });
+    const createWorker = vi.fn(() => ({
+      onMessage: vi.fn((handler) => {
+        onMessage = handler;
+        return () => undefined;
+      }),
+      stop: vi.fn(async () => {
+        stopWorker?.();
+      }),
+      task: vi.fn(async () => {
+        await workerDone;
+      }),
+    }));
+
+    const session = createPollingSession({
+      abortSignal: abort.signal,
+      isolatedIngress: {
+        enabled: true,
+        createWorker,
+        drainIntervalMs: 10,
+      },
+    });
+
+    const runPromise = session.runUntilAbort();
+    await vi.waitFor(() => expect(init).toHaveBeenCalledTimes(1));
+    onMessage?.({ type: "poll-success", finishedAt: Date.now(), count: 0 });
+
+    await vi.waitFor(() => expect(drainPendingDeliveriesMock).toHaveBeenCalledTimes(1));
+
+    abort.abort();
+    await runPromise;
+  });
+
   it("lets isolated ingress drain interleave different Telegram topic lanes", async () => {
     const abort = new AbortController();
     const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-"));
@@ -1175,6 +1253,87 @@ describe("TelegramPollingSession", () => {
     });
   });
 
+  it("drains Telegram delivery queue after getUpdates confirms polling reconnect", async () => {
+    const abort = new AbortController();
+    const botStop = vi.fn(async () => undefined);
+    const runnerStop = vi.fn(async () => undefined);
+    const getApiMiddleware = mockBotCapturingApiMiddleware(botStop);
+    const resolveFirstTask = mockLongRunningPollingCycle(runnerStop);
+
+    const session = createPollingSession({
+      abortSignal: abort.signal,
+    });
+
+    const runPromise = session.runUntilAbort();
+    const apiMiddleware = await waitForApiMiddleware(getApiMiddleware);
+    await apiMiddleware(
+      vi.fn(async () => []),
+      "getUpdates",
+      { offset: 1 },
+    );
+
+    await vi.waitFor(() => expect(drainPendingDeliveriesMock).toHaveBeenCalledTimes(1));
+    const drain = expectDrainPendingDeliveriesCall();
+    expect(drain.drainKey).toBe("telegram:default");
+    expect(drain.logLabel).toBe("Telegram reconnect drain");
+    expect(drain.selectEntry({ channel: "telegram" }, Date.now())).toEqual({
+      match: true,
+      bypassBackoff: false,
+    });
+    expect(
+      drain.selectEntry(
+        {
+          channel: "telegram",
+          accountId: "default",
+          lastError: "Network request for 'sendMessage' failed!",
+        },
+        Date.now(),
+      ),
+    ).toEqual({
+      match: true,
+      bypassBackoff: true,
+    });
+    expect(drain.selectEntry({ channel: "telegram", accountId: "alerts" }, Date.now()).match).toBe(
+      false,
+    );
+    expect(drain.selectEntry({ channel: "whatsapp" }, Date.now()).match).toBe(false);
+
+    abort.abort();
+    resolveFirstTask();
+    await runPromise;
+  });
+
+  it("drains Telegram delivery queue after each getUpdates success", async () => {
+    const abort = new AbortController();
+    const botStop = vi.fn(async () => undefined);
+    const runnerStop = vi.fn(async () => undefined);
+    const getApiMiddleware = mockBotCapturingApiMiddleware(botStop);
+    const resolveFirstTask = mockLongRunningPollingCycle(runnerStop);
+
+    const session = createPollingSession({
+      abortSignal: abort.signal,
+    });
+
+    const runPromise = session.runUntilAbort();
+    const apiMiddleware = await waitForApiMiddleware(getApiMiddleware);
+    await apiMiddleware(
+      vi.fn(async () => []),
+      "getUpdates",
+      { offset: 1 },
+    );
+    await apiMiddleware(
+      vi.fn(async () => []),
+      "getUpdates",
+      { offset: 2 },
+    );
+
+    await vi.waitFor(() => expect(drainPendingDeliveriesMock).toHaveBeenCalledTimes(2));
+
+    abort.abort();
+    resolveFirstTask();
+    await runPromise;
+  });
+
   it("keeps polling marked connected across recoverable restart cycles", async () => {
     const abort = new AbortController();
     const recoverableError = new Error("recoverable polling error");
diff --git a/extensions/telegram/src/polling-session.ts b/extensions/telegram/src/polling-session.ts
@@ -1,6 +1,7 @@
 import { type RunOptions, run } from "@grammyjs/runner";
 import type { ChannelAccountSnapshot } from "openclaw/plugin-sdk/channel-contract";
 import type { TelegramNetworkConfig } from "openclaw/plugin-sdk/config-contracts";
+import { drainPendingDeliveries } from "openclaw/plugin-sdk/delivery-queue-runtime";
 import {
   computeBackoff,
   formatDurationPrecise,
@@ -44,6 +45,20 @@ const TELEGRAM_POLLING_CLIENT_TIMEOUT_FLOOR_SECONDS = Math.ceil(
   TELEGRAM_GET_UPDATES_REQUEST_TIMEOUT_MS / 1000,
 );
 
+function normalizeTelegramAccountId(accountId?: string | null): string {
+  return accountId?.trim() || "default";
+}
+
+function shouldBypassReconnectBackoff(lastError?: string): boolean {
+  if (!lastError) {
+    return false;
+  }
+  return isRecoverableTelegramNetworkError(new Error(lastError), {
+    context: "send",
+    allowMessageMatch: true,
+  });
+}
+
 type TelegramBot = ReturnType<typeof createTelegramBot>;
 
 const waitForGracefulStop = async (stop: () => Promise<void>) => {
@@ -114,6 +129,7 @@ export class TelegramPollingSession {
   #transportState: TelegramPollingTransportState;
   #status: ReturnType<typeof createTelegramPollingStatusPublisher>;
   #stallThresholdMs: number;
+  #deliveryDrainInFlight = false;
 
   constructor(private readonly opts: TelegramPollingSessionOpts) {
     this.#transportState = new TelegramPollingTransportState({
@@ -202,6 +218,35 @@ export class TelegramPollingSession {
     );
   }
 
+  #drainPendingDeliveriesAfterReconnect() {
+    if (this.#deliveryDrainInFlight) {
+      return;
+    }
+    this.#deliveryDrainInFlight = true;
+    const accountId = normalizeTelegramAccountId(this.opts.accountId);
+    void drainPendingDeliveries({
+      drainKey: `telegram:${accountId}`,
+      logLabel: "Telegram reconnect drain",
+      cfg: this.opts.config,
+      log: {
+        info: (message) => this.opts.log(`[telegram][diag] ${message}`),
+        warn: (message) => this.opts.log(`[telegram] ${message}`),
+        error: (message) => this.opts.log(`[telegram] ${message}`),
+      },
+      selectEntry: (entry) => ({
+        match:
+          entry.channel === "telegram" && normalizeTelegramAccountId(entry.accountId) === accountId,
+        bypassBackoff: shouldBypassReconnectBackoff(entry.lastError),
+      }),
+    })
+      .catch((err) => {
+        this.opts.log(`[telegram] reconnect delivery drain failed: ${formatErrorMessage(err)}`);
+      })
+      .finally(() => {
+        this.#deliveryDrainInFlight = false;
+      });
+  }
+
   async #createPollingBot(): Promise<TelegramBot | undefined> {
     const fetchAbortController = new AbortController();
     this.#activeFetchAbort = fetchAbortController;
@@ -359,6 +404,7 @@ export class TelegramPollingSession {
       }
       if (message.type === "poll-success") {
         this.#status.notePollSuccess(message.finishedAt);
+        this.#drainPendingDeliveriesAfterReconnect();
         pollState.outcome = `ok:${message.count}`;
         return;
       }
@@ -428,7 +474,10 @@ export class TelegramPollingSession {
 
   async #runPollingCycle(bot: TelegramBot): Promise<"continue" | "exit"> {
     const liveness = new TelegramPollingLivenessTracker({
-      onPollSuccess: (finishedAt) => this.#status.notePollSuccess(finishedAt),
+      onPollSuccess: (finishedAt) => {
+        this.#status.notePollSuccess(finishedAt);
+        this.#drainPendingDeliveriesAfterReconnect();
+      },
     });
     bot.api.config.use(async (prev, method, payload, signal) => {
       if (method !== "getUpdates") {
