From 85fa8af4bbd3fcdab5ee2f1525c39a24d9adacbf Mon Sep 17 00:00:00 2001 From: xingzhou Date: Sat, 11 Jul 2026 10:27:44 +0800 Subject: [PATCH] fix(telegram): close evicted client-options cache transports after in-flight sends (#101783) Long-running Telegram senders leaked HTTP transports: the client-options cache in extensions/telegram/src/send.ts created an owned undici dispatcher per account/network entry but evicted entries without closing it. Cache entries now own their transport; eviction retires the entry and closes the transport once the operation-level lease releases, so cache pressure never aborts an in-flight, retrying, or still-preparing send. Includes a real-socket e2e regression against a local Bot API server. Co-authored-by: zhangguiping-xydt Co-authored-by: Ayaan Zaidi --- extensions/telegram/src/send.proxy.test.ts | 322 ++++++++++++++++++ extensions/telegram/src/send.test.ts | 88 ++++- .../src/send.transport-close-proof.test.ts | 139 ++++++++ extensions/telegram/src/send.ts | 303 ++++++++++++++-- 4 files changed, 819 insertions(+), 33 deletions(-) create mode 100644 extensions/telegram/src/send.transport-close-proof.test.ts diff --git a/extensions/telegram/src/send.proxy.test.ts b/extensions/telegram/src/send.proxy.test.ts index 0999a2a23d76..a6a8957ec4db 100644 --- a/extensions/telegram/src/send.proxy.test.ts +++ b/extensions/telegram/src/send.proxy.test.ts @@ -14,7 +14,9 @@ const { botApi, botCtorSpy } = vi.hoisted(() => ({ }; return { config: { use: vi.fn() }, + getChat: vi.fn(), sendMessage, + sendDocument: vi.fn(), setMessageReaction: vi.fn(), deleteMessage: vi.fn(), raw: { @@ -45,6 +47,14 @@ const { resolveTelegramTransport } = vi.hoisted(() => ({ resolveTelegramTransport: vi.fn(), })); +const { loadWebMedia } = vi.hoisted(() => ({ + loadWebMedia: vi.fn(), +})); + +const { maybePersistResolvedTelegramTarget } = vi.hoisted(() => ({ + maybePersistResolvedTelegramTarget: vi.fn(), +})); + const resolveTelegramApiBase = vi.hoisted( () => (apiRoot?: string) => apiRoot?.trim()?.replace(/\/+$/, "") || "https://api.telegram.org", ); @@ -68,6 +78,18 @@ vi.mock("./fetch.js", () => ({ resolveTelegramApiBase, })); +vi.mock("./send.runtime.js", async () => { + const actual = await vi.importActual("./send.runtime.js"); + return { + ...actual, + loadWebMedia, + }; +}); + +vi.mock("./target-writeback.js", () => ({ + maybePersistResolvedTelegramTarget, +})); + vi.mock("grammy", () => ({ API_CONSTANTS: { DEFAULT_UPDATE_TYPES: ["message"], @@ -134,9 +156,14 @@ describe("telegram proxy client", () => { beforeEach(() => { resetTelegramClientOptionsCacheForTests(); vi.unstubAllEnvs(); + botApi.getChat.mockResolvedValue({ id: "123" }); botApi.sendMessage.mockResolvedValue({ message_id: 1, chat: { id: "123" } }); + botApi.sendDocument.mockResolvedValue({ message_id: 1, chat: { id: "123" } }); botApi.setMessageReaction.mockResolvedValue(undefined); botApi.deleteMessage.mockResolvedValue(true); + loadWebMedia.mockReset(); + maybePersistResolvedTelegramTarget.mockReset(); + maybePersistResolvedTelegramTarget.mockResolvedValue(undefined); botApi.config.use.mockClear(); botCtorSpy.mockClear(); loadConfig.mockReturnValue(TELEGRAM_PROXY_CFG); @@ -169,6 +196,301 @@ describe("telegram proxy client", () => { expect(botCtorSpy).toHaveBeenNthCalledWith(2, "tok", firstOptions); }); + it("closes the evicted Telegram transport when the client options cache exceeds its limit", async () => { + vi.stubEnv("VITEST", ""); + vi.stubEnv("NODE_ENV", "production"); + const closeSpies: Array> = []; + makeProxyFetch.mockImplementation(() => vi.fn() as unknown as typeof fetch); + resolveTelegramTransport.mockImplementation(() => { + const fetchImpl = vi.fn() as unknown as typeof fetch; + const close = vi.fn(async () => undefined); + closeSpies.push(close); + return { + fetch: fetchImpl, + sourceFetch: fetchImpl, + close, + }; + }); + const cfg = { + channels: { + telegram: { + accounts: Object.fromEntries( + Array.from({ length: 65 }, (_unused, index) => [ + `acct-${index}`, + { proxy: `http://proxy-${index}.test:8080` }, + ]), + ), + }, + }, + }; + + for (let index = 0; index < 65; index += 1) { + await sendMessageTelegram("123", `message ${index}`, { + cfg, + token: "tok", + accountId: `acct-${index}`, + }); + } + + expect(resolveTelegramTransport).toHaveBeenCalledTimes(65); + expect(closeSpies[0]).toHaveBeenCalledTimes(1); + expect(closeSpies.slice(1).every((close) => close.mock.calls.length === 0)).toBe(true); + }); + + it.each([ + { + name: "an active send finishes", + setupFirstSend: () => { + let resolveFirstSend: (value: { + message_id: number; + chat: { id: string }; + }) => void = () => {}; + botApi.sendMessage.mockImplementationOnce( + () => + new Promise((resolve) => { + resolveFirstSend = resolve; + }), + ); + return { + waitUntilActive: async () => { + await vi.waitFor(() => expect(botApi.sendMessage).toHaveBeenCalledTimes(1)); + }, + finish: async () => { + resolveFirstSend({ message_id: 1, chat: { id: "123" } }); + }, + }; + }, + }, + { + name: "a retryable send is waiting between attempts", + setupFirstSend: () => { + vi.useFakeTimers(); + botApi.sendMessage + .mockRejectedValueOnce({ + error_code: 429, + description: "Too Many Requests: retry after 1", + parameters: { retry_after: 1 }, + }) + .mockResolvedValueOnce({ message_id: 1, chat: { id: "123" } }); + return { + waitUntilActive: async () => { + await vi.waitFor(() => expect(botApi.sendMessage).toHaveBeenCalledTimes(1)); + }, + finish: async () => { + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(() => expect(botApi.sendMessage).toHaveBeenCalledTimes(66)); + vi.useRealTimers(); + }, + }; + }, + }, + ])("defers closing an evicted Telegram transport until $name", async ({ setupFirstSend }) => { + vi.stubEnv("VITEST", ""); + vi.stubEnv("NODE_ENV", "production"); + const closeSpies: Array> = []; + makeProxyFetch.mockImplementation(() => vi.fn() as unknown as typeof fetch); + resolveTelegramTransport.mockImplementation(() => { + const fetchImpl = vi.fn() as unknown as typeof fetch; + const close = vi.fn(async () => undefined); + closeSpies.push(close); + return { + fetch: fetchImpl, + sourceFetch: fetchImpl, + close, + }; + }); + const cfg = { + channels: { + telegram: { + accounts: Object.fromEntries( + Array.from({ length: 65 }, (_unused, index) => [ + `acct-${index}`, + { proxy: `http://proxy-${index}.test:8080` }, + ]), + ), + }, + }, + }; + botApi.sendMessage.mockClear(); + const firstSendControl = setupFirstSend(); + + const firstSend = sendMessageTelegram("123", "first", { + cfg, + token: "tok", + accountId: "acct-0", + retry: { attempts: 2, minDelayMs: 1000, maxDelayMs: 1000, jitter: 0 }, + }); + await firstSendControl.waitUntilActive(); + + await Promise.all( + Array.from({ length: 64 }, (_unused, index) => + sendMessageTelegram("123", `message ${index + 1}`, { + cfg, + token: "tok", + accountId: `acct-${index + 1}`, + }), + ), + ); + + expect(resolveTelegramTransport).toHaveBeenCalledTimes(65); + expect(closeSpies[0]).not.toHaveBeenCalled(); + + await firstSendControl.finish(); + await firstSend; + + expect(closeSpies[0]).toHaveBeenCalledTimes(1); + vi.useRealTimers(); + }); + + it("defers closing an evicted Telegram transport while media loads before the first API request", async () => { + vi.stubEnv("VITEST", ""); + vi.stubEnv("NODE_ENV", "production"); + const closeSpies: Array> = []; + makeProxyFetch.mockImplementation(() => vi.fn() as unknown as typeof fetch); + resolveTelegramTransport.mockImplementation(() => { + const fetchImpl = vi.fn() as unknown as typeof fetch; + const close = vi.fn(async () => undefined); + closeSpies.push(close); + return { + fetch: fetchImpl, + sourceFetch: fetchImpl, + close, + }; + }); + const cfg = { + channels: { + telegram: { + accounts: Object.fromEntries( + Array.from({ length: 65 }, (_unused, index) => [ + `acct-${index}`, + { proxy: `http://proxy-${index}.test:8080` }, + ]), + ), + }, + }, + }; + let resolveMedia: (value: { + buffer: Buffer; + contentType: string; + fileName: string; + }) => void = () => {}; + loadWebMedia.mockImplementationOnce( + () => + new Promise((resolve) => { + resolveMedia = resolve; + }), + ); + + const firstSend = sendMessageTelegram("123", "caption", { + cfg, + token: "tok", + accountId: "acct-0", + mediaUrl: "file:///tmp/example.bin", + forceDocument: true, + }); + await vi.waitFor(() => expect(loadWebMedia).toHaveBeenCalledTimes(1)); + + await Promise.all( + Array.from({ length: 64 }, (_unused, index) => + sendMessageTelegram("123", `message ${index + 1}`, { + cfg, + token: "tok", + accountId: `acct-${index + 1}`, + }), + ), + ); + + expect(resolveTelegramTransport).toHaveBeenCalledTimes(65); + expect(closeSpies[0]).not.toHaveBeenCalled(); + + resolveMedia({ + buffer: Buffer.from("file"), + contentType: "application/octet-stream", + fileName: "example.bin", + }); + await firstSend; + + expect(botApi.sendDocument).toHaveBeenCalledTimes(1); + expect(closeSpies[0]).toHaveBeenCalledTimes(1); + }); + + it("defers closing an evicted Telegram transport while reactions persist a resolved target before the first action request", async () => { + vi.stubEnv("VITEST", ""); + vi.stubEnv("NODE_ENV", "production"); + const closeSpies: Array> = []; + makeProxyFetch.mockImplementation(() => vi.fn() as unknown as typeof fetch); + resolveTelegramTransport.mockImplementation(() => { + const fetchImpl = vi.fn() as unknown as typeof fetch; + const close = vi.fn(async () => undefined); + closeSpies.push(close); + return { + fetch: fetchImpl, + sourceFetch: fetchImpl, + close, + }; + }); + const cfg = { + channels: { + telegram: { + accounts: Object.fromEntries( + Array.from({ length: 65 }, (_unused, index) => [ + `acct-${index}`, + { proxy: `http://proxy-${index}.test:8080` }, + ]), + ), + }, + }, + }; + let resolvePersist: () => void = () => {}; + maybePersistResolvedTelegramTarget.mockImplementationOnce( + () => + new Promise((resolve) => { + resolvePersist = resolve; + }), + ); + + const firstReaction = reactMessageTelegram("@target", "456", "✅", { + cfg, + token: "tok", + accountId: "acct-0", + }); + await vi.waitFor(() => expect(maybePersistResolvedTelegramTarget).toHaveBeenCalledTimes(1)); + + await Promise.all( + Array.from({ length: 64 }, (_unused, index) => + reactMessageTelegram("123", "456", "✅", { + cfg, + token: "tok", + accountId: `acct-${index + 1}`, + }), + ), + ); + + expect(resolveTelegramTransport).toHaveBeenCalledTimes(65); + expect(closeSpies[0]).not.toHaveBeenCalled(); + + resolvePersist(); + await firstReaction; + + expect(botApi.setMessageReaction).toHaveBeenCalledTimes(65); + expect(closeSpies[0]).toHaveBeenCalledTimes(1); + }); + + it("does not allocate cached client transport when a Telegram API override is provided", async () => { + vi.stubEnv("VITEST", ""); + vi.stubEnv("NODE_ENV", "production"); + + await reactMessageTelegram("123", "456", "✅", { + cfg: TELEGRAM_PROXY_CFG, + token: "tok", + accountId: "foo", + api: botApi as unknown as Parameters[3]["api"], + }); + + expect(makeProxyFetch).not.toHaveBeenCalled(); + expect(resolveTelegramTransport).not.toHaveBeenCalled(); + }); + it.each([ { name: "sendMessage", diff --git a/extensions/telegram/src/send.test.ts b/extensions/telegram/src/send.test.ts index 6d6c4ad47471..057073eef4b6 100644 --- a/extensions/telegram/src/send.test.ts +++ b/extensions/telegram/src/send.test.ts @@ -54,6 +54,7 @@ const { createForumTopicTelegram, deleteMessageTelegram, editForumTopicTelegram, + editMessageReplyMarkupTelegram, editMessageTelegram, pinMessageTelegram, reactMessageTelegram, @@ -183,6 +184,69 @@ function installTelegramStateRuntimeForTest(): void { } as TelegramRuntime); } +describe("Telegram send Promise contract", () => { + const contextFailureCalls: ReadonlyArray Promise]> = [ + [ + "sendMessageTelegram", + () => sendMessageTelegramImported("123", "hello", { cfg: TELEGRAM_TEST_CFG }), + ], + ["sendTypingTelegram", () => sendTypingTelegram("123", { cfg: TELEGRAM_TEST_CFG })], + [ + "reactMessageTelegram", + () => reactMessageTelegram("123", 1, "👍", { cfg: TELEGRAM_TEST_CFG }), + ], + ["deleteMessageTelegram", () => deleteMessageTelegram("123", 1, { cfg: TELEGRAM_TEST_CFG })], + ["pinMessageTelegram", () => pinMessageTelegram("123", 1, { cfg: TELEGRAM_TEST_CFG })], + [ + "unpinMessageTelegram", + () => unpinMessageTelegram("123", undefined, { cfg: TELEGRAM_TEST_CFG }), + ], + [ + "editForumTopicTelegram", + () => editForumTopicTelegram("123", 1, { cfg: TELEGRAM_TEST_CFG, name: "topic" }), + ], + [ + "editMessageReplyMarkupTelegram", + () => editMessageReplyMarkupTelegram("123", 1, [], { cfg: TELEGRAM_TEST_CFG }), + ], + [ + "editMessageTelegram", + () => editMessageTelegram("123", 1, "hello", { cfg: TELEGRAM_TEST_CFG }), + ], + [ + "sendStickerTelegram", + () => sendStickerTelegram("123", "file-id", { cfg: TELEGRAM_TEST_CFG }), + ], + [ + "sendPollTelegram", + () => + sendPollTelegram( + "123", + { question: "Question?", options: ["A", "B"] }, + { cfg: TELEGRAM_TEST_CFG }, + ), + ], + [ + "createForumTopicTelegram", + () => createForumTopicTelegram("123", "topic", { cfg: TELEGRAM_TEST_CFG }), + ], + ]; + + it.each(contextFailureCalls)( + "%s reports context failures as Promise rejections", + async (_name, invoke) => { + let operation: Promise | undefined; + expect(() => { + operation = invoke(); + }).not.toThrow(); + if (!operation) { + throw new Error("expected Telegram operation promise"); + } + await expect(operation).rejects.toThrow(/Telegram bot token missing/i); + }, + ); +}); + async function expectChatNotFoundWithChatId( action: Promise, expectedChatId: string, @@ -784,20 +848,25 @@ describe("sendMessageTelegram", () => { }); }); - it("rejects empty topic edits", async () => { + it("rejects empty topic edits before creating a Telegram client", async () => { + botCtorSpy.mockClear(); + await expect( editForumTopicTelegram("-1001234567890", 271, { cfg: TELEGRAM_TEST_CFG, + token: "tok", accountId: "default", }), ).rejects.toThrow("Telegram forum topic update requires a name or iconCustomEmojiId"); await expect( editForumTopicTelegram("-1001234567890", 271, { cfg: TELEGRAM_TEST_CFG, + token: "tok", accountId: "default", iconCustomEmojiId: " ", }), ).rejects.toThrow("Telegram forum topic icon custom emoji ID is required"); + expect(botCtorSpy).not.toHaveBeenCalled(); }); it("applies timeoutSeconds config precedence", async () => { @@ -3724,12 +3793,15 @@ describe("sendStickerTelegram", () => { }); } - it("throws error when fileId is blank", async () => { + it("rejects a blank fileId before creating a Telegram client", async () => { + botCtorSpy.mockClear(); + for (const fileId of ["", " "]) { await expect( sendStickerTelegram("123", fileId, { cfg: TELEGRAM_TEST_CFG, token: "tok" }), ).rejects.toThrow(/file_id is required/i); } + expect(botCtorSpy).not.toHaveBeenCalled(); }); it("fails sticker sends instead of retrying without message_thread_id", async () => { @@ -4507,4 +4579,16 @@ describe("createForumTopicTelegram", () => { expect(result).toEqual(testCase.expectedResult); }); } + + it("rejects an invalid topic name before creating a Telegram client", async () => { + botCtorSpy.mockClear(); + + await expect( + createForumTopicTelegram("-1001234567890", " ", { + cfg: TELEGRAM_TEST_CFG, + token: "tok", + }), + ).rejects.toThrow("Forum topic name is required"); + expect(botCtorSpy).not.toHaveBeenCalled(); + }); }); diff --git a/extensions/telegram/src/send.transport-close-proof.test.ts b/extensions/telegram/src/send.transport-close-proof.test.ts new file mode 100644 index 000000000000..e515b26fee6a --- /dev/null +++ b/extensions/telegram/src/send.transport-close-proof.test.ts @@ -0,0 +1,139 @@ +// E2E proof for the transport cache-eviction lifecycle: no module mocks — real +// grammY Bot, real undici agents, production-mode cache, against a local HTTP +// server standing in for the Telegram Bot API. Observes actual TCP sockets. +import { createServer, type Server } from "node:http"; +import type { AddressInfo } from "node:net"; +import type { Socket } from "node:net"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; + +let sendMessageTelegram: typeof import("./send.js").sendMessageTelegram; +let resetTelegramClientOptionsCacheForTests: typeof import("./send.js").resetTelegramClientOptionsCacheForTests; + +describe("telegram transport cache eviction over real sockets", () => { + let server: Server; + let apiRoot: string; + const liveSockets = new Set(); + const sockets = { opened: 0, closed: 0 }; + let sendMessageCalls = 0; + let slowMode = false; + let slowRequestReceived: () => void = () => {}; + + beforeAll(async () => { + server = createServer((req, res) => { + let body = ""; + req.on("data", (chunk: Buffer) => { + body += chunk; + }); + req.on("end", () => { + const url = req.url ?? ""; + const respond = (result: unknown) => { + res.setHeader("content-type", "application/json"); + res.end(JSON.stringify({ ok: true, result })); + }; + if (url.includes("/sendMessage")) { + sendMessageCalls += 1; + if (slowMode) { + slowRequestReceived(); + setTimeout(() => respond({ message_id: sendMessageCalls, chat: { id: 123 } }), 800); + return; + } + respond({ message_id: sendMessageCalls, chat: { id: 123 } }); + return; + } + if (url.includes("/getChat")) { + respond({ id: 123, type: "private" }); + return; + } + respond(true); + }); + }); + server.on("connection", (socket) => { + sockets.opened += 1; + liveSockets.add(socket); + socket.on("close", () => { + sockets.closed += 1; + liveSockets.delete(socket); + }); + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + apiRoot = `http://127.0.0.1:${(server.address() as AddressInfo).port}`; + ({ sendMessageTelegram, resetTelegramClientOptionsCacheForTests } = await import("./send.js")); + }); + + afterAll(async () => { + resetTelegramClientOptionsCacheForTests(); + vi.unstubAllEnvs(); + for (const socket of liveSockets) { + socket.destroy(); + } + await new Promise((resolve) => server.close(() => resolve())); + }); + + it("closes evicted transports, deferring close for an in-flight send", async () => { + // The cache is disabled under test env; force the production path. + vi.stubEnv("VITEST", ""); + vi.stubEnv("NODE_ENV", "production"); + resetTelegramClientOptionsCacheForTests(); + + const ACCOUNTS = 70; + const cfg = { + channels: { + telegram: { + accounts: Object.fromEntries( + Array.from({ length: ACCOUNTS }, (_, i) => [ + `acct-${i}`, + { botToken: `10${i}:e2e-token-${i}`, apiRoot }, + ]), + ), + }, + }, + }; + + // Fill the cache to its 64-entry cap: one real agent + socket per account. + for (let i = 0; i < 64; i += 1) { + const result = await sendMessageTelegram("123", `hello ${i}`, { + cfg, + accountId: `acct-${i}`, + }); + expect(result.messageId).toBeTruthy(); + } + expect(sockets.opened).toBe(64); + // Keep-alive is 30s; nothing may have closed yet. + expect(sockets.closed).toBe(0); + + // Put acct-0 (the oldest cache entry) mid-flight, then evict it. + slowMode = true; + const inFlight = new Promise((resolve) => { + slowRequestReceived = resolve; + }); + const slowSend = sendMessageTelegram("123", "slow", { cfg, accountId: "acct-0" }); + await inFlight; + slowMode = false; + + // New cache key -> evicts acct-0 while its send holds the lease. + const evictor = await sendMessageTelegram("123", "evictor", { cfg, accountId: "acct-64" }); + expect(evictor.messageId).toBeTruthy(); + // Deferred close: the evicted transport must NOT be closed mid-request. + expect(sockets.closed).toBe(0); + + const slow = await slowSend; + expect(slow.messageId).toBeTruthy(); + // Lease released -> the retired acct-0 transport closes its real socket. + await vi.waitFor(() => expect(sockets.closed).toBeGreaterThanOrEqual(1), { timeout: 3000 }); + + // Five more evictions against idle entries (acct-1..acct-5) close immediately. + for (let i = 65; i < ACCOUNTS; i += 1) { + const result = await sendMessageTelegram("123", `hello ${i}`, { + cfg, + accountId: `acct-${i}`, + }); + expect(result.messageId).toBeTruthy(); + } + await vi.waitFor(() => expect(sockets.closed).toBe(6), { timeout: 3000 }); + + // All sends succeeded; retained cache entries keep their sockets open. + expect(sendMessageCalls).toBe(ACCOUNTS + 1); + expect(sockets.opened).toBe(ACCOUNTS); + expect(liveSockets.size).toBe(ACCOUNTS - 6); + }); +}); diff --git a/extensions/telegram/src/send.ts b/extensions/telegram/src/send.ts index 8bcfe8722a00..9cc00b2f88a3 100644 --- a/extensions/telegram/src/send.ts +++ b/extensions/telegram/src/send.ts @@ -26,7 +26,7 @@ import { buildTypingThreadParams } from "./bot/helpers.js"; import type { TelegramInlineButtons } from "./button-types.js"; import { splitTelegramCaption } from "./caption.js"; import { asTelegramClientFetch, createTelegramClientFetch } from "./client-fetch.js"; -import { resolveTelegramTransport } from "./fetch.js"; +import { resolveTelegramTransport, type TelegramTransport } from "./fetch.js"; import { renderTelegramHtmlText, splitTelegramHtmlChunks, @@ -306,7 +306,21 @@ const MESSAGE_DELETE_NOOP_RE = const CHAT_NOT_FOUND_RE = /400: Bad Request: chat not found/i; const sendLogger = createSubsystemLogger("telegram/send"); const diagLogger = createSubsystemLogger("telegram/diagnostic"); -const telegramClientOptionsCache = new Map(); +type CachedTelegramClientOptions = { + activeLeases: number; + clientOptions: ApiClientOptions | undefined; + closeStarted: boolean; + retired: boolean; + transport: TelegramTransport; +}; +type TelegramClientOptionsLease = { + release: () => void; +}; +type ResolvedTelegramClientOptions = { + clientOptions: ApiClientOptions | undefined; + lease?: () => TelegramClientOptionsLease; +}; +const telegramClientOptionsCache = new Map(); const MAX_TELEGRAM_CLIENT_OPTIONS_CACHE_SIZE = 64; export function resetTelegramClientOptionsCacheForTests(): void { @@ -346,23 +360,66 @@ function buildTelegramClientOptionsCacheKey(params: { return `${params.account.accountId}::${proxyKey}::${autoSelectFamilyKey}::${dnsResultOrderKey}::${apiRootKey}::${timeoutSecondsKey}`; } +function closeCachedTelegramClientOptions(entry: CachedTelegramClientOptions): void { + // Eviction may retire a cache entry while a send still holds a lease; defer + // transport.close until the last op-level lease releases so mid-request sockets stay open. + entry.retired = true; + if (entry.activeLeases > 0 || entry.closeStarted) { + return; + } + entry.closeStarted = true; + void entry.transport.close().catch((err: unknown) => { + diagLogger.warn( + `telegram client options cache transport close failed: ${redactSensitiveText( + formatUncaughtError(err), + )}`, + ); + }); +} + +function leaseCachedTelegramClientOptions( + entry: CachedTelegramClientOptions, +): TelegramClientOptionsLease { + entry.activeLeases += 1; + let released = false; + return { + release: () => { + if (released) { + return; + } + released = true; + entry.activeLeases = Math.max(0, entry.activeLeases - 1); + if (entry.retired) { + closeCachedTelegramClientOptions(entry); + } + }, + }; +} + function setCachedTelegramClientOptions( cacheKey: string, - clientOptions: ApiClientOptions | undefined, -): ApiClientOptions | undefined { - telegramClientOptionsCache.set(cacheKey, clientOptions); + entry: CachedTelegramClientOptions, +): ResolvedTelegramClientOptions { + telegramClientOptionsCache.set(cacheKey, entry); if (telegramClientOptionsCache.size > MAX_TELEGRAM_CLIENT_OPTIONS_CACHE_SIZE) { const oldestKey = telegramClientOptionsCache.keys().next().value; if (oldestKey !== undefined) { + const evictedEntry = telegramClientOptionsCache.get(oldestKey); telegramClientOptionsCache.delete(oldestKey); + if (evictedEntry) { + closeCachedTelegramClientOptions(evictedEntry); + } } } - return clientOptions; + return { + clientOptions: entry.clientOptions, + lease: () => leaseCachedTelegramClientOptions(entry), + }; } function resolveTelegramClientOptions( account: ResolvedTelegramAccount, -): ApiClientOptions | undefined { +): ResolvedTelegramClientOptions { const timeoutSeconds = typeof account.config.timeoutSeconds === "number" && Number.isFinite(account.config.timeoutSeconds) @@ -377,7 +434,13 @@ function resolveTelegramClientOptions( }) : null; if (cacheKey && telegramClientOptionsCache.has(cacheKey)) { - return telegramClientOptionsCache.get(cacheKey); + const entry = telegramClientOptionsCache.get(cacheKey); + if (entry) { + return { + clientOptions: entry.clientOptions, + lease: () => leaseCachedTelegramClientOptions(entry), + }; + } } const proxyUrl = normalizeOptionalString(account.config.proxy); @@ -401,9 +464,15 @@ function resolveTelegramClientOptions( } : undefined; if (cacheKey) { - return setCachedTelegramClientOptions(cacheKey, clientOptions); + return setCachedTelegramClientOptions(cacheKey, { + activeLeases: 0, + clientOptions, + closeStarted: false, + retired: false, + transport, + }); } - return clientOptions; + return { clientOptions }; } function resolveToken(explicit: string | undefined, params: { accountId: string; token: string }) { @@ -564,6 +633,7 @@ type TelegramApiContext = { cfg: OpenClawConfig; account: ResolvedTelegramAccount; api: TelegramApi; + clientOptionsLease?: TelegramClientOptionsLease | undefined; }; function resolveTelegramApiContext(opts: { @@ -578,16 +648,32 @@ function resolveTelegramApiContext(opts: { accountId: opts.accountId, }); const token = resolveToken(opts.token, account); - const client = resolveTelegramClientOptions(account); let api: TelegramApi; + let clientOptionsLease: TelegramClientOptionsLease | undefined; if (opts.api) { api = opts.api as TelegramApi; } else { - const bot = new Bot(token, client ? { client } : undefined); + const client = resolveTelegramClientOptions(account); + // One op-level lease covers the full send/action (including pre-request work + // and retries) so eviction cannot close the transport mid-operation. + clientOptionsLease = client.lease?.(); + const bot = new Bot(token, client.clientOptions ? { client: client.clientOptions } : undefined); bot.api.config.use(getOrCreateAccountThrottler(token)); api = bot.api; } - return { cfg, account, api }; + return { + cfg, + account, + api, + ...(clientOptionsLease ? { clientOptionsLease } : {}), + }; +} + +function withTelegramApiContextLease( + context: TelegramApiContext, + operation: Promise, +): Promise { + return operation.finally(() => context.clientOptionsLease?.release()); } type TelegramRequestWithDiag = ( @@ -704,7 +790,20 @@ export async function sendMessageTelegram( text: string, opts: TelegramSendOpts, ): Promise { - const { cfg, account, api } = resolveTelegramApiContext(opts); + const context = resolveTelegramApiContext(opts); + return withTelegramApiContextLease( + context, + sendMessageTelegramWithContext(to, text, opts, context), + ); +} + +async function sendMessageTelegramWithContext( + to: string, + text: string, + opts: TelegramSendOpts, + apiContext: TelegramApiContext, +): Promise { + const { cfg, account, api } = apiContext; const botUserId = resolveTelegramBotUserIdFromToken(opts.token || account.token); const target = parseTelegramTarget(to); const chatId = await resolveAndPersistChatId({ @@ -1469,7 +1568,16 @@ export async function sendTypingTelegram( to: string, opts: TelegramTypingOpts, ): Promise<{ ok: true }> { - const { cfg, account, api } = resolveTelegramApiContext(opts); + const context = resolveTelegramApiContext(opts); + return withTelegramApiContextLease(context, sendTypingTelegramWithContext(to, opts, context)); +} + +async function sendTypingTelegramWithContext( + to: string, + opts: TelegramTypingOpts, + context: TelegramApiContext, +): Promise<{ ok: true }> { + const { cfg, account, api } = context; const target = parseTelegramTarget(to); const chatId = await resolveAndPersistChatId({ cfg, @@ -1504,7 +1612,21 @@ export async function reactMessageTelegram( emoji: string, opts: TelegramReactionOpts, ): Promise<{ ok: true } | { ok: false; warning: string }> { - const { cfg, account, api } = resolveTelegramApiContext(opts); + const context = resolveTelegramApiContext(opts); + return withTelegramApiContextLease( + context, + reactMessageTelegramWithContext(chatIdInput, messageIdInput, emoji, opts, context), + ); +} + +async function reactMessageTelegramWithContext( + chatIdInput: string | number, + messageIdInput: string | number, + emoji: string, + opts: TelegramReactionOpts, + context: TelegramApiContext, +): Promise<{ ok: true } | { ok: false; warning: string }> { + const { cfg, account, api } = context; const rawTarget = String(chatIdInput); const chatId = await resolveAndPersistChatId({ cfg, @@ -1561,7 +1683,20 @@ export async function deleteMessageTelegram( messageIdInput: string | number, opts: TelegramDeleteOpts, ): Promise<{ ok: true } | { ok: false; warning: string }> { - const { cfg, account, api } = resolveTelegramApiContext(opts); + const context = resolveTelegramApiContext(opts); + return withTelegramApiContextLease( + context, + deleteMessageTelegramWithContext(chatIdInput, messageIdInput, opts, context), + ); +} + +async function deleteMessageTelegramWithContext( + chatIdInput: string | number, + messageIdInput: string | number, + opts: TelegramDeleteOpts, + context: TelegramApiContext, +): Promise<{ ok: true } | { ok: false; warning: string }> { + const { cfg, account, api } = context; const rawTarget = String(chatIdInput); const chatId = await resolveAndPersistChatId({ cfg, @@ -1603,7 +1738,20 @@ export async function pinMessageTelegram( messageIdInput: string | number, opts: TelegramDeleteOpts, ): Promise<{ ok: true; messageId: string; chatId: string }> { - const { cfg, account, api } = resolveTelegramApiContext(opts); + const context = resolveTelegramApiContext(opts); + return withTelegramApiContextLease( + context, + pinMessageTelegramWithContext(chatIdInput, messageIdInput, opts, context), + ); +} + +async function pinMessageTelegramWithContext( + chatIdInput: string | number, + messageIdInput: string | number, + opts: TelegramDeleteOpts, + context: TelegramApiContext, +): Promise<{ ok: true; messageId: string; chatId: string }> { + const { cfg, account, api } = context; const rawTarget = String(chatIdInput); const chatId = await resolveAndPersistChatId({ cfg, @@ -1636,7 +1784,20 @@ export async function unpinMessageTelegram( messageIdInput: string | number | undefined, opts: TelegramDeleteOpts, ): Promise<{ ok: true; chatId: string; messageId?: string }> { - const { cfg, account, api } = resolveTelegramApiContext(opts); + const context = resolveTelegramApiContext(opts); + return withTelegramApiContextLease( + context, + unpinMessageTelegramWithContext(chatIdInput, messageIdInput, opts, context), + ); +} + +async function unpinMessageTelegramWithContext( + chatIdInput: string | number, + messageIdInput: string | number | undefined, + opts: TelegramDeleteOpts, + context: TelegramApiContext, +): Promise<{ ok: true; chatId: string; messageId?: string }> { + const { cfg, account, api } = context; const rawTarget = String(chatIdInput); const chatId = await resolveAndPersistChatId({ cfg, @@ -1697,7 +1858,28 @@ export async function editForumTopicTelegram( throw new Error("Telegram forum topic update requires a name or iconCustomEmojiId"); } - const { cfg, account, api } = resolveTelegramApiContext(opts); + const context = resolveTelegramApiContext(opts); + return withTelegramApiContextLease( + context, + editForumTopicTelegramWithContext(chatIdInput, messageThreadIdInput, opts, context), + ); +} + +async function editForumTopicTelegramWithContext( + chatIdInput: string | number, + messageThreadIdInput: string | number, + opts: TelegramEditForumTopicOpts, + context: TelegramApiContext, +): Promise<{ + ok: true; + chatId: string; + messageThreadId: number; + name?: string; + iconCustomEmojiId?: string; +}> { + const trimmedName = opts.name?.trim(); + const trimmedIconCustomEmojiId = opts.iconCustomEmojiId?.trim(); + const { cfg, account, api } = context; const rawTarget = String(chatIdInput); const target = parseTelegramTarget(rawTarget); const chatId = await resolveAndPersistChatId({ @@ -1788,10 +1970,21 @@ export async function editMessageReplyMarkupTelegram( buttons: TelegramInlineButtons, opts: TelegramEditReplyMarkupOpts, ): Promise<{ ok: true; messageId: string; chatId: string }> { - const { cfg, account, api } = resolveTelegramApiContext({ - ...opts, - cfg: opts.cfg, - }); + const context = resolveTelegramApiContext(opts); + return withTelegramApiContextLease( + context, + editMessageReplyMarkupTelegramWithContext(chatIdInput, messageIdInput, buttons, opts, context), + ); +} + +async function editMessageReplyMarkupTelegramWithContext( + chatIdInput: string | number, + messageIdInput: string | number, + buttons: TelegramInlineButtons, + opts: TelegramEditReplyMarkupOpts, + context: TelegramApiContext, +): Promise<{ ok: true; messageId: string; chatId: string }> { + const { cfg, account, api } = context; const rawTarget = String(chatIdInput); const chatId = await resolveAndPersistChatId({ cfg, @@ -1832,10 +2025,21 @@ export async function editMessageTelegram( text: string, opts: TelegramEditOpts, ): Promise<{ ok: true; messageId: string; chatId: string }> { - const { cfg, account, api } = resolveTelegramApiContext({ - ...opts, - cfg: opts.cfg, - }); + const context = resolveTelegramApiContext(opts); + return withTelegramApiContextLease( + context, + editMessageTelegramWithContext(chatIdInput, messageIdInput, text, opts, context), + ); +} + +async function editMessageTelegramWithContext( + chatIdInput: string | number, + messageIdInput: string | number, + text: string, + opts: TelegramEditOpts, + context: TelegramApiContext, +): Promise<{ ok: true; messageId: string; chatId: string }> { + const { cfg, account, api } = context; const rawTarget = String(chatIdInput); const chatId = await resolveAndPersistChatId({ cfg, @@ -2063,7 +2267,20 @@ export async function sendStickerTelegram( throw new Error("Telegram sticker file_id is required"); } - const { cfg, account, api } = resolveTelegramApiContext(opts); + const context = resolveTelegramApiContext(opts); + return withTelegramApiContextLease( + context, + sendStickerTelegramWithContext(to, fileId, opts, context), + ); +} + +async function sendStickerTelegramWithContext( + to: string, + fileId: string, + opts: TelegramStickerOpts, + context: TelegramApiContext, +): Promise { + const { cfg, account, api } = context; const target = parseTelegramTarget(to); const chatId = await resolveAndPersistChatId({ cfg, @@ -2145,7 +2362,17 @@ export async function sendPollTelegram( poll: PollInput, opts: TelegramPollOpts, ): Promise<{ messageId: string; chatId: string; pollId?: string }> { - const { cfg, account, api } = resolveTelegramApiContext(opts); + const context = resolveTelegramApiContext(opts); + return withTelegramApiContextLease(context, sendPollTelegramWithContext(to, poll, opts, context)); +} + +async function sendPollTelegramWithContext( + to: string, + poll: PollInput, + opts: TelegramPollOpts, + context: TelegramApiContext, +): Promise<{ messageId: string; chatId: string; pollId?: string }> { + const { cfg, account, api } = context; const target = parseTelegramTarget(to); const chatId = await resolveAndPersistChatId({ cfg, @@ -2267,7 +2494,21 @@ export async function createForumTopicTelegram( throw new Error("Forum topic name must be 128 characters or fewer"); } - const { cfg, account, api } = resolveTelegramApiContext(opts); + const context = resolveTelegramApiContext(opts); + return withTelegramApiContextLease( + context, + createForumTopicTelegramWithContext(chatId, name, opts, context), + ); +} + +async function createForumTopicTelegramWithContext( + chatId: string, + name: string, + opts: TelegramCreateForumTopicOpts, + context: TelegramApiContext, +): Promise { + const trimmedName = name.trim(); + const { cfg, account, api } = context; // Accept topic-qualified targets (e.g. telegram:group::topic:) // but createForumTopic must always target the base supergroup chat id. const target = parseTelegramTarget(chatId);