From cb1b3be8fb9821adddb30a9983e552ae9989c19f Mon Sep 17 00:00:00 2001 From: Oleg Tsvetkov Date: Thu, 10 Sep 2026 00:01:34 +0300 Subject: [PATCH] fix(telegram): replies are lost when the HTTP proxy refuses the tunnel (#140388) Classify transport-proven proxy tunnel failures through Telegram's existing request-not-started error so queued replies remain safely retryable. Preserve caller cancellation and ambiguous-send protection. Verified real Telegram Test Server recovery across Gateway restart through explicit and environment proxies, discriminating abort/TLS controls, native reply context, and focused regression tests. Co-authored-by: Ayaan Zaidi --- extensions/telegram/src/fetch.test.ts | 90 ++++++++ extensions/telegram/src/fetch.ts | 57 ++++- extensions/telegram/src/network-errors.ts | 4 +- ...bound-permanent-rejection-loopback.test.ts | 203 ++++++++++++++++++ 4 files changed, 351 insertions(+), 3 deletions(-) diff --git a/extensions/telegram/src/fetch.test.ts b/extensions/telegram/src/fetch.test.ts index 7f8d79f5799b..1609196589f1 100644 --- a/extensions/telegram/src/fetch.test.ts +++ b/extensions/telegram/src/fetch.test.ts @@ -1345,4 +1345,94 @@ describe("resolveTelegramFetch", () => { }); }); }); + +describe("resolveTelegramTransport proxy tunnel failures", () => { + function buildProxyTunnelRejection(statusCode: number) { + const tunnelError = Object.assign( + new Error(`Proxy response (${statusCode}) !== 200 when HTTP Tunneling`), + { name: "AbortError", code: "UND_ERR_ABORTED" }, + ); + return Object.assign(new TypeError("fetch failed"), { cause: tunnelError }); + } + + async function captureTransportError( + transport: ReturnType, + ): Promise { + try { + await transport.fetch("https://api.telegram.org/botTOKEN/sendMessage"); + } catch (error) { + return error; + } + throw new Error("expected the Telegram transport fetch to reject"); + } + + it("marks a CONNECT refused by the explicit proxy as request-not-started", async () => { + vi.stubEnv("OPENCLAW_PROXY_URL", "http://127.0.0.1:7788"); + const rejection = buildProxyTunnelRejection(503); + undiciFetch.mockRejectedValue(rejection); + + const transport = resolveTelegramTransport(undefined, { + network: { autoSelectFamily: false, dnsResultOrder: "ipv4first" }, + }); + const caught = await captureTransportError(transport); + + expect(caught).toBeInstanceOf(TelegramRequestNotStartedError); + expect((caught as Error).message).toContain("503"); + expect((caught as Error).cause).toBe(rejection); + expect(isSafeToRetrySendError(caught)).toBe(true); + expect(undiciFetch).toHaveBeenCalledTimes(1); + }); + + it("marks a proxy connection failure while opening the tunnel as request-not-started", async () => { + vi.stubEnv("OPENCLAW_PROXY_URL", "http://127.0.0.1:7788"); + const rejection = Object.assign(new TypeError("fetch failed"), { + cause: Object.assign(new Error("Proxy Connection failed"), { + name: "ProxyConnectionError", + code: "UND_ERR_PRX_CONN", + }), + }); + undiciFetch.mockRejectedValue(rejection); + + const transport = resolveTelegramTransport(undefined, { + network: { autoSelectFamily: false, dnsResultOrder: "ipv4first" }, + }); + const caught = await captureTransportError(transport); + + expect(caught).toBeInstanceOf(TelegramRequestNotStartedError); + expect((caught as Error).cause).toBe(rejection); + expect(isSafeToRetrySendError(caught)).toBe(true); + }); + + it("keeps a generic abort on the explicit proxy dispatcher ambiguous", async () => { + vi.stubEnv("OPENCLAW_PROXY_URL", "http://127.0.0.1:7788"); + const rejection = Object.assign(new TypeError("fetch failed"), { + cause: Object.assign(new Error("Request aborted"), { + name: "AbortError", + code: "UND_ERR_ABORTED", + }), + }); + undiciFetch.mockRejectedValue(rejection); + + const transport = resolveTelegramTransport(undefined, { + network: { autoSelectFamily: false, dnsResultOrder: "ipv4first" }, + }); + const caught = await captureTransportError(transport); + + expect(caught).toBe(rejection); + expect(isSafeToRetrySendError(caught)).toBe(false); + }); + + it("does not claim a tunnel refusal for a direct dispatcher", async () => { + const rejection = buildProxyTunnelRejection(503); + undiciFetch.mockRejectedValue(rejection); + + const transport = resolveTelegramTransport(undefined, { + network: { autoSelectFamily: false, dnsResultOrder: "ipv4first" }, + }); + const caught = await captureTransportError(transport); + + expect(caught).toBe(rejection); + expect(isSafeToRetrySendError(caught)).toBe(false); + }); +}); /* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */ diff --git a/extensions/telegram/src/fetch.ts b/extensions/telegram/src/fetch.ts index 1bdbb829d096..8307dc913233 100644 --- a/extensions/telegram/src/fetch.ts +++ b/extensions/telegram/src/fetch.ts @@ -2,7 +2,11 @@ import { randomUUID } from "node:crypto"; import * as dns from "node:dns"; import type { TelegramNetworkConfig } from "openclaw/plugin-sdk/config-contracts"; -import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; +import { + collectErrorGraphCandidates, + extractErrorCode, + formatErrorMessage, +} from "openclaw/plugin-sdk/error-runtime"; import { expectDefined } from "openclaw/plugin-sdk/expect-runtime"; import { createHttp1EnvHttpProxyAgent, @@ -436,6 +440,45 @@ function shouldUseTelegramTransportFallback(err: unknown): boolean { return hasKnownNetworkCode || (hasFetchFailedEnvelope && ctx.codes.size === 0); } +// undici's ProxyAgent reports a non-200 CONNECT reply as a generic +// RequestAbortedError (`UND_ERR_ABORTED`, the code an aborted in-flight request +// also carries) and a socket failure while the tunnel is still being set up as +// ProxyConnectionError (`UND_ERR_PRX_CONN`). Both happen before the TLS session +// to api.telegram.org exists, so no request bytes reached Telegram. The tunnel +// wording is the only signal that separates a refused CONNECT from a real +// abort, so it is matched verbatim and only for the proxy dispatchers this +// transport builds itself. +const UNDICI_PROXY_TUNNEL_REJECTED_RE = /^Proxy response \((\d{3})\) !== 200 when HTTP Tunneling$/; +const UNDICI_PROXY_CONNECTION_ERROR_CODE = "UND_ERR_PRX_CONN"; + +function describeProxyTunnelFailure( + mode: TelegramDispatcherMode | undefined, + err: unknown, +): string | undefined { + if (mode !== "explicit-proxy" && mode !== "env-proxy") { + return undefined; + } + for (const candidate of collectErrorGraphCandidates(err, (current) => [ + current.cause, + ...(Array.isArray(current.errors) ? current.errors : []), + ])) { + const code = extractErrorCode(candidate); + if (code === UNDICI_PROXY_CONNECTION_ERROR_CODE) { + return "proxy connection failed while opening the tunnel"; + } + if (code !== "UND_ERR_ABORTED" || !candidate || typeof candidate !== "object") { + continue; + } + const message = "message" in candidate ? candidate.message : undefined; + const rejected = + typeof message === "string" ? UNDICI_PROXY_TUNNEL_REJECTED_RE.exec(message) : null; + if (rejected) { + return `proxy answered CONNECT with ${rejected[1]}`; + } + } + return undefined; +} + export function shouldRetryTelegramTransportFallback(err: unknown): boolean { return shouldUseTelegramTransportFallback(err); } @@ -814,6 +857,18 @@ export function resolveTelegramTransport( return response; } catch (caught) { signal?.throwIfAborted(); + const tunnelFailure = describeProxyTunnelFailure( + attempt.exportAttempt.dispatcherPolicy?.mode, + caught, + ); + if (tunnelFailure) { + // This transport built the proxy dispatcher, so a tunnel that never + // opened proves the request did not reach Telegram. + throw new TelegramRequestNotStartedError( + `Telegram proxy tunnel did not open: ${tunnelFailure}`, + { cause: caught }, + ); + } err = caught; if (!shouldUseTelegramTransportFallback(err)) { throw err; diff --git a/extensions/telegram/src/network-errors.ts b/extensions/telegram/src/network-errors.ts index e9511d76dc38..6630552c44eb 100644 --- a/extensions/telegram/src/network-errors.ts +++ b/extensions/telegram/src/network-errors.ts @@ -16,8 +16,8 @@ const TELEGRAM_SUPERGROUP_MIGRATION_DESCRIPTION = "Bad Request: group chat was upgraded to a supergroup chat"; export class TelegramRequestNotStartedError extends Error { - constructor(message = "Telegram request did not start") { - super(message); + constructor(message = "Telegram request did not start", options?: ErrorOptions) { + super(message, options); this.name = "TelegramRequestNotStartedError"; } } diff --git a/test/telegram-outbound-permanent-rejection-loopback.test.ts b/test/telegram-outbound-permanent-rejection-loopback.test.ts index 1afa22d79da7..f0c99595a226 100644 --- a/test/telegram-outbound-permanent-rejection-loopback.test.ts +++ b/test/telegram-outbound-permanent-rejection-loopback.test.ts @@ -213,3 +213,206 @@ describe("Telegram permanent rejection over real Bot API transport", () => { } }); }); + +const PROXY_TUNNEL_DELIVERY_INTENT_ID = "telegram-loopback-proxy-tunnel-refusal"; + +type RefusingProxy = { + url: string; + connectTargets: string[]; + close: () => Promise; +}; + +type QueueRow = { recoveryState: string | null; retryCount: number; status: string }; + +function readQueueRow(stateDir: string): QueueRow | undefined { + const { db } = openOpenClawStateDatabase({ + env: { ...process.env, OPENCLAW_STATE_DIR: stateDir }, + }); + const row = db + // sqlite-allow-raw: The proof reads one exact queue owner after each recovery pass. + .prepare( + "SELECT status, retry_count, recovery_state FROM delivery_queue_entries WHERE queue_name = ? AND id = ?", + ) + .get(OUTBOUND_DELIVERY_QUEUE_NAME, PROXY_TUNNEL_DELIVERY_INTENT_ID) as + | { recovery_state: string | null; retry_count: number; status: string } + | undefined; + return row + ? { recoveryState: row.recovery_state, retryCount: row.retry_count, status: row.status } + : undefined; +} + +// A forward proxy whose CONNECT handler always answers 503, the shape a stalled +// upstream (for example an ssh -D hop behind an HTTP proxy) produces. The TLS +// session to api.telegram.org never starts, so nothing can reach Telegram. +async function startRefusingProxy(): Promise { + const connectTargets: string[] = []; + const sockets = new Set(); + const server: Server = createServer((_request, response) => { + response.writeHead(405, { "content-length": "0" }); + response.end(); + }); + server.on("connect", (request, socket) => { + connectTargets.push(request.url ?? ""); + socket.end( + "HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + ); + }); + server.on("connection", (socket) => { + sockets.add(socket); + socket.on("close", () => sockets.delete(socket)); + }); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolve); + }); + const { port } = server.address() as AddressInfo; + return { + url: `http://127.0.0.1:${port}`, + connectTargets, + close: async () => { + for (const socket of sockets) { + socket.destroy(); + } + await new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }); + }, + }; +} + +describe("Telegram proxy tunnel refusal over the real proxy transport", () => { + afterEach(() => { + closeOpenClawAgentDatabasesForTest(); + closeOpenClawStateDatabaseForTest(); + resetGlobalHookRunner(); + resetPluginRuntimeStateForTest(); + setActivePluginRegistry(createEmptyPluginRegistry()); + }); + + it("keeps a durable send replayable while the proxy refuses the CONNECT tunnel", async () => { + const proxy = await startRefusingProxy(); + try { + const { telegramPlugin } = await import("../extensions/telegram/api.js"); + const cfg = { + channels: { + telegram: { + botToken: "123456:loopback", + proxy: proxy.url, + }, + }, + } satisfies OpenClawConfig; + setActivePluginRegistry( + createTestRegistry([{ pluginId: "telegram", plugin: telegramPlugin, source: "test" }]), + ); + + await withStateDirEnv("openclaw-telegram-proxy-tunnel-loopback-", async ({ stateDir }) => { + try { + const staged = await sendDurableMessageBatch({ + cfg, + channel: "telegram", + to: "123", + accountId: "default", + durability: "required", + deliveryIntentId: PROXY_TUNNEL_DELIVERY_INTENT_ID, + completionRetention: { + idPrefix: "telegram-loopback-", + maxAgeMs: 60_000, + maxEntries: 10, + }, + maxRetries: 10, + payloads: [{ text: "reply that must survive a proxy tunnel refusal" }], + deps: { + telegram: async () => { + throw new PlatformMessageNotDispatchedError( + "staged before transport for recovery proof", + { cause: new Error("loopback proxy not released yet") }, + ); + }, + }, + }); + expect(staged.status).toBe("failed"); + expect(proxy.connectTargets).toHaveLength(0); + expect( + getDeliveryQueueEntryStatus( + OUTBOUND_DELIVERY_QUEUE_NAME, + PROXY_TUNNEL_DELIVERY_INTENT_ID, + stateDir, + ), + ).toBe("pending"); + const stagedRow = readQueueRow(stateDir); + expect(stagedRow?.status).toBe("pending"); + const stagedRetryCount = stagedRow?.retryCount ?? 0; + + const log = { info: vi.fn(), warn: vi.fn(), error: vi.fn() }; + const drain = (logLabel: string) => + drainPendingDeliveries({ + drainKey: "telegram:default", + logLabel, + cfg, + stateDir, + log, + selectEntry: (entry) => ({ + match: entry.channel === "telegram", + bypassBackoff: true, + }), + }); + + await drain("Telegram loopback proxy tunnel recovery"); + + // Every attempt ended at the proxy hop: the in-process retry ran, each + // attempt targeted the Bot API tunnel, and nothing was accepted. + const firstPassConnects = proxy.connectTargets.length; + expect(firstPassConnects).toBeGreaterThanOrEqual(2); + expect(new Set(proxy.connectTargets)).toEqual(new Set(["api.telegram.org:443"])); + const afterFirstPass = readQueueRow(stateDir); + expect(afterFirstPass).toMatchObject({ + retryCount: stagedRetryCount + 1, + status: "pending", + }); + // The typed no-send proof keeps the entry replayable instead of parking + // it as an ambiguous send that recovery must refuse to replay. + expect(afterFirstPass?.recoveryState).not.toBe("unknown_after_send"); + for (const call of log.warn.mock.calls) { + expect(String(call[0])).not.toContain("unknown_after_send"); + } + + closeOpenClawAgentDatabasesForTest(); + closeOpenClawStateDatabaseForTest(); + expect( + getDeliveryQueueEntryStatus( + OUTBOUND_DELIVERY_QUEUE_NAME, + PROXY_TUNNEL_DELIVERY_INTENT_ID, + stateDir, + ), + ).toBe("pending"); + + await drain("Telegram loopback post-restart proxy tunnel recovery"); + expect(proxy.connectTargets.length).toBeGreaterThan(firstPassConnects); + const afterSecondPass = readQueueRow(stateDir); + expect(afterSecondPass).toMatchObject({ + retryCount: stagedRetryCount + 2, + status: "pending", + }); + expect(afterSecondPass?.recoveryState).not.toBe("unknown_after_send"); + + console.log( + `[telegram proxy-tunnel proof] ${JSON.stringify({ + queueStatus: afterSecondPass?.status, + recoveryState: afterSecondPass?.recoveryState, + retryCountAfterTwoRecoveryPasses: afterSecondPass?.retryCount, + connectAttempts: proxy.connectTargets.length, + proxyResponse: 503, + classification: "request not started (proxy tunnel never opened)", + transport: "grammY Bot API HTTPS via CONNECT to 127.0.0.1:", + })}`, + ); + } finally { + closeOpenClawAgentDatabasesForTest(); + closeOpenClawStateDatabaseForTest(); + } + }); + } finally { + await proxy.close(); + } + }); +});