fix(telegram): cool down unhealthy transports

obviyus · obviyus · commit 1e8ca266166b · 2026-05-10T09:51:11.000+05:30
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -711,6 +711,7 @@ Docs: https://docs.openclaw.ai
 - Memory/wiki: preserve representation from both corpora in `corpus=all` searches while backfilling unused result capacity, so memory hits are not starved by numerically higher wiki integer scores. Fixes #77337. Thanks @hclsys.
 - Docker/compose: pin container-side `OPENCLAW_CONFIG_DIR` and `OPENCLAW_WORKSPACE_DIR` on both gateway and CLI services so the host paths written into `.env` by `scripts/docker/setup.sh` (used as Compose bind-mount sources) cannot leak into runtime code via the `env_file` import. Fixes regressions on macOS Docker setups where the first agent reply died with `EACCES: permission denied, mkdir '/Users'` because the host-style workspace path got persisted into `agents.defaults.workspace`. Fixes #77436. Thanks @lonexreb.
 - Telegram: clean up tool-only draft previews after assistant message boundaries so transient `Surfacing...` tool-status bubbles do not linger when no matching final preview arrives. Thanks @BunsDev.
+- Telegram: cool down repeatedly failing Bot API transport fallbacks so long polling stops hammering a blackholed Telegram route. Fixes #77900. Thanks @bryce-d-greybeard.
 - Slack: report `unknown error` instead of `undefined` in socket-mode startup retry logs and label the retry reason explicitly.
 - Telegram: let explicit forum-topic `requireMention` settings override persisted `/activate` and `/deactivate` state, so per-topic mention gates work consistently. Fixes #49864. Thanks @Panniantong.
 - Cron: surface failed isolated-run diagnostics in `cron show`, status, and run history when requested tools are unavailable, so blocked cron runs report the actual tool-policy failure instead of a misleading green result. Fixes #75763. Thanks @RyanSandoval.
diff --git a/extensions/telegram/src/fetch.test.ts b/extensions/telegram/src/fetch.test.ts
@@ -179,6 +179,10 @@ function buildFetchFallbackError(code: string) {
   });
 }
 
+function buildCodeLessFetchFallbackError() {
+  return new TypeError("fetch failed");
+}
+
 const STICKY_IPV4_FALLBACK_NETWORK = {
   network: {
     autoSelectFamily: true,
@@ -955,6 +959,63 @@ describe("resolveTelegramFetch", () => {
     expect(getDispatcherFromUndiciCall(4)).toBe(getDispatcherFromUndiciCall(3));
   });
 
+  it("falls back on code-less fetch failed envelopes", async () => {
+    undiciFetch
+      .mockRejectedValueOnce(buildCodeLessFetchFallbackError())
+      .mockResolvedValueOnce({ ok: true } as Response);
+
+    const resolved = resolveTelegramFetchOrThrow(undefined, {
+      network: {
+        autoSelectFamily: true,
+        dnsResultOrder: "ipv4first",
+      },
+    });
+
+    await resolved("https://api.telegram.org/botx/deleteWebhook");
+
+    expect(undiciFetch).toHaveBeenCalledTimes(2);
+    expect(getDispatcherFromUndiciCall(1)).not.toBe(getDispatcherFromUndiciCall(2));
+  });
+
+  it("cools down a repeatedly failing sticky fallback and probes earlier attempts", async () => {
+    for (let i = 0; i < 7; i += 1) {
+      undiciFetch.mockRejectedValueOnce(buildFetchFallbackError("ENETUNREACH"));
+    }
+    undiciFetch
+      .mockRejectedValueOnce(buildFetchFallbackError("ENETUNREACH"))
+      .mockRejectedValueOnce(buildFetchFallbackError("ENETUNREACH"));
+
+    const resolved = resolveTelegramFetchOrThrow(undefined, {
+      network: {
+        autoSelectFamily: true,
+        dnsResultOrder: "ipv4first",
+      },
+    });
+
+    await expect(resolved("https://api.telegram.org/botx/deleteWebhook")).rejects.toThrow(
+      "fetch failed",
+    );
+    for (let i = 0; i < 4; i += 1) {
+      await expect(resolved("https://api.telegram.org/botx/getUpdates")).rejects.toThrow(
+        "fetch failed",
+      );
+    }
+    await expect(resolved("https://api.telegram.org/botx/getUpdates")).rejects.toThrow(
+      "temporarily unhealthy",
+    );
+
+    expect(undiciFetch).toHaveBeenCalledTimes(9);
+    expect(getDispatcherFromUndiciCall(7)).toBe(getDispatcherFromUndiciCall(3));
+    expect(getDispatcherFromUndiciCall(8)).toBe(getDispatcherFromUndiciCall(1));
+    expect(getDispatcherFromUndiciCall(9)).toBe(getDispatcherFromUndiciCall(2));
+    expect(loggerWarn).toHaveBeenCalledWith(
+      expect.stringContaining("telegram transport attempt marked temporarily unhealthy"),
+    );
+    expect(loggerDebug).toHaveBeenCalledWith(
+      expect.stringContaining("fetch fallback: re-probing primary dispatcher"),
+    );
+  });
+
   it("preserves caller-provided dispatcher across fallback retry", async () => {
     const fetchError = buildFetchFallbackError("EHOSTUNREACH");
     undiciFetch.mockRejectedValueOnce(fetchError).mockResolvedValueOnce({ ok: true } as Response);
diff --git a/extensions/telegram/src/fetch.ts b/extensions/telegram/src/fetch.ts
@@ -42,6 +42,9 @@ const TELEGRAM_DISPATCHER_KEEP_ALIVE_MAX_TIMEOUT_MS = 600_000;
 const TELEGRAM_DISPATCHER_CONNECTIONS_PER_ORIGIN = 10;
 const TELEGRAM_DISPATCHER_PIPELINING = 1;
 const TELEGRAM_STICKY_FALLBACK_PRIMARY_PROBE_SUCCESS_THRESHOLD = 5;
+const TELEGRAM_TRANSPORT_ATTEMPT_FAILURE_THRESHOLD = 5;
+const TELEGRAM_TRANSPORT_ATTEMPT_INITIAL_COOLDOWN_MS = 10_000;
+const TELEGRAM_TRANSPORT_ATTEMPT_MAX_COOLDOWN_MS = 60_000;
 
 type TelegramAgentPoolOptions = {
   allowH2: false;
@@ -80,6 +83,12 @@ type TelegramTransportAttempt = {
   logMessage?: string;
 };
 
+type TelegramTransportAttemptHealth = {
+  consecutiveFailures: number;
+  cooldownMs: number;
+  unhealthyUntilMs: number;
+};
+
 type TelegramDnsResultOrder = "ipv4first" | "verbatim";
 
 type LookupCallback =
@@ -110,22 +119,6 @@ type TelegramTransportFallbackContext = {
   codes: Set<string>;
 };
 
-type TelegramTransportFallbackRule = {
-  name: string;
-  matches: (ctx: TelegramTransportFallbackContext) => boolean;
-};
-
-const TELEGRAM_TRANSPORT_FALLBACK_RULES: readonly TelegramTransportFallbackRule[] = [
-  {
-    name: "fetch-failed-envelope",
-    matches: ({ message }) => message.includes("fetch failed"),
-  },
-  {
-    name: "known-network-code",
-    matches: ({ codes }) => FALLBACK_RETRY_ERROR_CODES.some((code) => codes.has(code)),
-  },
-];
-
 function normalizeDnsResultOrder(value: string | null): TelegramDnsResultOrder | null {
   if (value === "ipv4first" || value === "verbatim") {
     return value;
@@ -446,20 +439,28 @@ function formatErrorCodes(err: unknown): string {
   return codes.length > 0 ? codes.join(",") : "none";
 }
 
+class TelegramTransportAttemptUnhealthyError extends Error {
+  constructor(unhealthyUntilMs: number) {
+    const remainingMs = Math.max(0, unhealthyUntilMs - Date.now());
+    super(`telegram transport attempt temporarily unhealthy; retry after ${remainingMs}ms`);
+    this.name = "TelegramTransportAttemptUnhealthyError";
+  }
+}
+
 function shouldUseTelegramTransportFallback(err: unknown): boolean {
+  if (err instanceof TelegramTransportAttemptUnhealthyError) {
+    return true;
+  }
   const ctx: TelegramTransportFallbackContext = {
     message:
       err && typeof err === "object" && "message" in err
         ? normalizeLowercaseStringOrEmpty(String(err.message))
         : "",
     codes: collectErrorCodes(err),
   };
-  for (const rule of TELEGRAM_TRANSPORT_FALLBACK_RULES) {
-    if (!rule.matches(ctx)) {
-      return false;
-    }
-  }
-  return true;
+  const hasFetchFailedEnvelope = ctx.message.includes("fetch failed");
+  const hasKnownNetworkCode = FALLBACK_RETRY_ERROR_CODES.some((code) => ctx.codes.has(code));
+  return hasKnownNetworkCode || (hasFetchFailedEnvelope && ctx.codes.size === 0);
 }
 
 export function shouldRetryTelegramTransportFallback(err: unknown): boolean {
@@ -643,12 +644,46 @@ export function resolveTelegramTransport(
   let stickyAttemptIndex = 0;
   let stickySuccessCount = 0;
   let primaryProbeDue = false;
+  const attemptHealth = transportAttempts.map<TelegramTransportAttemptHealth>(() => ({
+    consecutiveFailures: 0,
+    cooldownMs: TELEGRAM_TRANSPORT_ATTEMPT_INITIAL_COOLDOWN_MS,
+    unhealthyUntilMs: 0,
+  }));
 
   const resetStickyRecoveryProbe = (): void => {
     stickySuccessCount = 0;
     primaryProbeDue = false;
   };
 
+  const getAttemptCooldownError = (attemptIndex: number): Error | null => {
+    const health = attemptHealth[attemptIndex];
+    if (health.unhealthyUntilMs <= Date.now()) {
+      return null;
+    }
+    return new TelegramTransportAttemptUnhealthyError(health.unhealthyUntilMs);
+  };
+
+  const recordAttemptFailure = (attemptIndex: number, err: unknown): void => {
+    if (!shouldUseTelegramTransportFallback(err)) {
+      return;
+    }
+    const health = attemptHealth[attemptIndex];
+    health.consecutiveFailures += 1;
+    if (health.consecutiveFailures < TELEGRAM_TRANSPORT_ATTEMPT_FAILURE_THRESHOLD) {
+      return;
+    }
+    const cooldownMs = Math.min(
+      TELEGRAM_TRANSPORT_ATTEMPT_MAX_COOLDOWN_MS,
+      Math.max(TELEGRAM_TRANSPORT_ATTEMPT_INITIAL_COOLDOWN_MS, health.cooldownMs),
+    );
+    health.consecutiveFailures = 0;
+    health.cooldownMs = Math.min(TELEGRAM_TRANSPORT_ATTEMPT_MAX_COOLDOWN_MS, cooldownMs * 2);
+    health.unhealthyUntilMs = Date.now() + cooldownMs;
+    log.warn(
+      `telegram transport attempt marked temporarily unhealthy for ${cooldownMs}ms (codes=${formatErrorCodes(err)})`,
+    );
+  };
+
   const promoteStickyAttempt = (nextIndex: number, err: unknown, reason?: string): boolean => {
     if (nextIndex <= stickyAttemptIndex || nextIndex >= transportAttempts.length) {
       return false;
@@ -669,6 +704,11 @@ export function resolveTelegramTransport(
   };
 
   const recordSuccessfulAttempt = (attemptIndex: number): void => {
+    const health = attemptHealth[attemptIndex];
+    health.consecutiveFailures = 0;
+    health.cooldownMs = TELEGRAM_TRANSPORT_ATTEMPT_INITIAL_COOLDOWN_MS;
+    health.unhealthyUntilMs = 0;
+
     if (stickyAttemptIndex === 0) {
       resetStickyRecoveryProbe();
       return;
@@ -700,50 +740,63 @@ export function resolveTelegramTransport(
       (init as RequestInitWithDispatcher | undefined)?.dispatcher,
     );
     const stickyStartIndex = Math.min(stickyAttemptIndex, transportAttempts.length - 1);
-    const primaryProbe = !callerProvidedDispatcher && primaryProbeDue && stickyStartIndex > 0;
+    const stickyCooldownError = callerProvidedDispatcher
+      ? null
+      : getAttemptCooldownError(stickyStartIndex);
+    const primaryProbe =
+      !callerProvidedDispatcher &&
+      stickyStartIndex > 0 &&
+      (primaryProbeDue || stickyCooldownError !== null);
     const startIndex = primaryProbe ? 0 : stickyStartIndex;
     if (primaryProbe) {
       primaryProbeDue = false;
-      log.debug("fetch fallback: re-probing primary dispatcher after sticky fallback successes");
-    }
-    let err: unknown;
-
-    try {
-      const response = await sourceFetch(
-        input,
-        withDispatcherIfMissing(init, transportAttempts[startIndex].createDispatcher()),
+      log.debug(
+        stickyCooldownError
+          ? "fetch fallback: re-probing primary dispatcher while sticky fallback is cooling down"
+          : "fetch fallback: re-probing primary dispatcher after sticky fallback successes",
       );
-      captureHttpExchange({
-        url: resolveRequestUrl(input),
-        method: init?.method ?? "GET",
-        requestHeaders: init?.headers as Headers | Record<string, string> | undefined,
-        requestBody: (init as RequestInit & { body?: BodyInit | null })?.body ?? null,
-        response,
-        flowId: randomUUID(),
-        meta: { subsystem: "telegram-fetch" },
-      });
-      if (!callerProvidedDispatcher) {
-        recordSuccessfulAttempt(startIndex);
-      }
-      return response;
-    } catch (caught) {
-      err = caught;
     }
+    let err: unknown;
 
-    if (!shouldUseTelegramTransportFallback(err)) {
-      throw err;
-    }
     if (callerProvidedDispatcher) {
-      return sourceFetch(input, init ?? {});
+      try {
+        const response = await sourceFetch(input, init);
+        captureHttpExchange({
+          url: resolveRequestUrl(input),
+          method: init?.method ?? "GET",
+          requestHeaders: init?.headers as Headers | Record<string, string> | undefined,
+          requestBody: (init as RequestInit & { body?: BodyInit | null })?.body ?? null,
+          response,
+          flowId: randomUUID(),
+          meta: { subsystem: "telegram-fetch" },
+        });
+        return response;
+      } catch (caught) {
+        if (!shouldUseTelegramTransportFallback(caught)) {
+          throw caught;
+        }
+        return sourceFetch(input, init ?? {});
+      }
     }
 
-    for (let nextIndex = startIndex + 1; nextIndex < transportAttempts.length; nextIndex += 1) {
-      const nextAttempt = transportAttempts[nextIndex];
-      promoteStickyAttempt(nextIndex, err);
+    for (
+      let attemptIndex = startIndex;
+      attemptIndex < transportAttempts.length;
+      attemptIndex += 1
+    ) {
+      const attempt = transportAttempts[attemptIndex];
+      if (attemptIndex > startIndex) {
+        promoteStickyAttempt(attemptIndex, err);
+      }
+      const cooldownError = getAttemptCooldownError(attemptIndex);
+      if (cooldownError) {
+        err = cooldownError;
+        continue;
+      }
       try {
         const response = await sourceFetch(
           input,
-          withDispatcherIfMissing(init, nextAttempt.createDispatcher()),
+          withDispatcherIfMissing(init, attempt.createDispatcher()),
         );
         captureHttpExchange({
           url: resolveRequestUrl(input),
@@ -752,15 +805,19 @@ export function resolveTelegramTransport(
           requestBody: (init as RequestInit & { body?: BodyInit | null })?.body ?? null,
           response,
           flowId: randomUUID(),
-          meta: { subsystem: "telegram-fetch", fallbackAttempt: nextIndex },
+          meta:
+            attemptIndex === startIndex
+              ? { subsystem: "telegram-fetch" }
+              : { subsystem: "telegram-fetch", fallbackAttempt: attemptIndex },
         });
-        recordSuccessfulAttempt(nextIndex);
+        recordSuccessfulAttempt(attemptIndex);
         return response;
       } catch (caught) {
         err = caught;
         if (!shouldUseTelegramTransportFallback(err)) {
           throw err;
         }
+        recordAttemptFailure(attemptIndex, err);
       }
     }
 
