mirror of
https://github.com/openclaw/openclaw.git
synced 2026-10-03 09:39:25 +00:00
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 <hi@obviy.us>
This commit is contained in:
Oleg Tsvetkov
GitHub
parent
f30b1e9d97
commit
cb1b3be8fb
4 changed files with 351 additions and 3 deletions
|
|
@ -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<typeof resolveTelegramTransport>,
|
||||
): Promise<unknown> {
|
||||
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. */
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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";
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<void>;
|
||||
};
|
||||
|
||||
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<RefusingProxy> {
|
||||
const connectTargets: string[] = [];
|
||||
const sockets = new Set<Socket>();
|
||||
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<void>((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<void>((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:<redacted>",
|
||||
})}`,
|
||||
);
|
||||
} finally {
|
||||
closeOpenClawAgentDatabasesForTest();
|
||||
closeOpenClawStateDatabaseForTest();
|
||||
}
|
||||
});
|
||||
} finally {
|
||||
await proxy.close();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue