From 26aef8c48bc40d2ea456c2cfc3eca126dca33888 Mon Sep 17 00:00:00 2001 From: Alix-007 Date: Fri, 10 Jul 2026 21:31:16 +0800 Subject: [PATCH] fix(telegram): recover when media downloads stall (#103020) * fix(telegram): add request timeout to media file downloads * fix(telegram): bound inbound media startup Co-authored-by: llagy009 <0668001470@xydigit.com> Co-authored-by: Alix-007 * docs(changelog): note Telegram media recovery * fix(net): normalize preflight abort errors * test(media): accept nullable request signals * fix(telegram): abort webhook media on shutdown * fix(telegram): own webhook media cancellation * fix(telegram): retry aborted media groups * fix(telegram): cancel isolated media work * fix(telegram): stop draining ended polling cycles * fix(telegram): separate cycle media cancellation * fix(telegram): preserve live polling media fallbacks * chore: move release note to PR body --------- Co-authored-by: llagy009 <0668001470@xydigit.com> Co-authored-by: Peter Steinberger --- .../telegram/src/bot-handlers.runtime.ts | 36 ++- ...te-telegram-bot.channel-post-media.test.ts | 124 ++++++++- extensions/telegram/src/bot.test.ts | 161 ++++++++++- extensions/telegram/src/bot.types.ts | 2 + .../bot/delivery.resolve-media-retry.test.ts | 46 ++++ .../src/bot/delivery.resolve-media.ts | 18 +- .../telegram/src/polling-session.test.ts | 255 ++++++++++++++++++ extensions/telegram/src/polling-session.ts | 79 ++++-- .../telegram/src/telegram-ingress-spool.ts | 9 + extensions/telegram/src/webhook.test.ts | 35 +++ extensions/telegram/src/webhook.ts | 7 + src/channels/message/ingress-queue.test.ts | 12 + src/channels/message/ingress-queue.ts | 12 +- src/infra/net/fetch-guard.ssrf.test.ts | 52 ++++ src/infra/net/fetch-guard.ts | 79 ++++-- src/media/fetch.test.ts | 250 ++++++++++++++++- src/media/fetch.ts | 28 +- 17 files changed, 1145 insertions(+), 60 deletions(-) diff --git a/extensions/telegram/src/bot-handlers.runtime.ts b/extensions/telegram/src/bot-handlers.runtime.ts index 49251584a69e..9ba232a4f1b4 100644 --- a/extensions/telegram/src/bot-handlers.runtime.ts +++ b/extensions/telegram/src/bot-handlers.runtime.ts @@ -233,9 +233,13 @@ export const registerTelegramHandlers = ({ token: opts.token, transport: telegramTransport, }); + const mediaAbortSignal = + opts.mediaAbortSignal && opts.fetchAbortSignal + ? AbortSignal.any([opts.mediaAbortSignal, opts.fetchAbortSignal]) + : (opts.mediaAbortSignal ?? opts.fetchAbortSignal); const mediaRuntimeWithAbort = { ...mediaRuntimeOptions, - abortSignal: opts.fetchAbortSignal, + abortSignal: mediaAbortSignal, }; const DEFAULT_TEXT_FRAGMENT_MAX_GAP_MS = 1500; const TELEGRAM_TEXT_FRAGMENT_START_THRESHOLD_CHARS = 4000; @@ -670,6 +674,7 @@ export const registerTelegramHandlers = ({ ...spooledReplayOptions(spooledReplayParticipants), }, dispatchDedupeKeys: last.dispatchDedupeKeys, + durableMediaReplay: spooledReplayParticipants.length > 0, }); settleSpooledReplayParticipants(spooledReplayParticipants, result); return; @@ -719,6 +724,7 @@ export const registerTelegramHandlers = ({ dispatchDedupeKeys: mergeDispatchDedupeKeys( ...entries.map((entry) => entry.dispatchDedupeKeys), ), + durableMediaReplay: spooledReplayParticipants.length > 0, }); settleSpooledReplayParticipants(spooledReplayParticipants, result); } catch (err) { @@ -1080,6 +1086,14 @@ export const registerTelegramHandlers = ({ ...mediaRuntimeWithAbort, }); } catch (mediaErr) { + // Only durable ingress can replay an aborted album. Classic polling keeps + // its best-effort partial delivery so Telegram does not acknowledge a drop. + if ( + mediaRuntimeWithAbort.abortSignal?.aborted && + entry.spooledReplayParticipants.length > 0 + ) { + throw mediaErr; + } if (!isRecoverableMediaGroupError(mediaErr)) { throw mediaErr; } @@ -1138,6 +1152,7 @@ export const registerTelegramHandlers = ({ ...spooledReplayOptions(entry.spooledReplayParticipants), }, dispatchDedupeKeys: entry.dispatchDedupeKeys, + durableMediaReplay: entry.spooledReplayParticipants.length > 0, }); settleSpooledReplayParticipants(entry.spooledReplayParticipants, result); } catch (err) { @@ -1197,6 +1212,7 @@ export const registerTelegramHandlers = ({ ...spooledReplayOptions(entry.spooledReplayParticipants), }, dispatchDedupeKeys: entry.dispatchDedupeKeys, + durableMediaReplay: entry.spooledReplayParticipants.length > 0, }); settleSpooledReplayParticipants(entry.spooledReplayParticipants, result); } catch (err) { @@ -1424,6 +1440,7 @@ export const registerTelegramHandlers = ({ ctx: TelegramContext, chain: TelegramCachedMessageNode[], shouldHydrateMedia: (node: TelegramCachedMessageNode, index: number) => Promise, + durableMediaReplay: boolean, ): Promise<{ replyMedia: TelegramMediaRef[]; replyChain: TelegramReplyChainEntry[] }> => { const replyMedia: TelegramMediaRef[] = []; const replyChain: TelegramReplyChainEntry[] = []; @@ -1440,7 +1457,7 @@ export const registerTelegramHandlers = ({ ctx: { message: node.sourceMessage, me: ctx.me, - getFile: async () => await bot.api.getFile(replyFileId), + getFile: async (signal) => await bot.api.getFile(replyFileId, signal), }, maxBytes: mediaMaxBytes, ...mediaRuntimeWithAbort, @@ -1453,6 +1470,12 @@ export const registerTelegramHandlers = ({ } : undefined; } catch (err) { + // Only durable ingress can replay a reply-media abort. Live polling must + // preserve the current text instead of acknowledging it without dispatch. + if (mediaRuntimeWithAbort.abortSignal?.aborted && durableMediaReplay) { + recordTelegramMessageProcessingResult({ kind: "failed-retryable", error: err }); + throw err; + } logger.warn( { chatId: ctx.message.chat.id, error: String(err) }, "reply media fetch failed", @@ -1475,11 +1498,15 @@ export const registerTelegramHandlers = ({ storeAllowFrom: string[]; options?: TelegramMessageContextOptions; dispatchDedupeKeys?: string[]; + durableMediaReplay?: boolean; }): Promise => { let dispatchDedupeCommitted = false; let dispatchDedupeRollbackAttempted = false; - const spooledReplay = - params.options?.spooledReplay === true || isTelegramSpooledReplayUpdate(params.ctx.update); + // Callback-submit retries also set options.spooledReplay without durable ingress. + // Media aborts retry only when the update frame or a buffered participant owns replay. + const durableMediaReplay = + params.durableMediaReplay === true || isTelegramSpooledReplayUpdate(params.ctx.update); + const spooledReplay = params.options?.spooledReplay === true || durableMediaReplay; const forgetCommittedDispatchDedupeKeys = async () => { dispatchDedupeRollbackAttempted = true; await forgetDispatchDedupeKeys(params.dispatchDedupeKeys ?? []); @@ -1547,6 +1574,7 @@ export const registerTelegramHandlers = ({ params.ctx, replyChainNodes, shouldHydrateReplyMedia, + durableMediaReplay, ); const promptContextMediaByMessageId = new Map(); const currentMessageId = diff --git a/extensions/telegram/src/bot.create-telegram-bot.channel-post-media.test.ts b/extensions/telegram/src/bot.create-telegram-bot.channel-post-media.test.ts index 30e150e83fa5..56d694f4b829 100644 --- a/extensions/telegram/src/bot.create-telegram-bot.channel-post-media.test.ts +++ b/extensions/telegram/src/bot.create-telegram-bot.channel-post-media.test.ts @@ -67,8 +67,11 @@ const { } = harness; const { createTelegramBotCore: createTelegramBotBase, setTelegramBotRuntimeForTest } = await import("./bot-core.js"); -const { runWithTelegramUpdateProcessingFrame, withTelegramSpooledReplayUpdate } = - await import("./bot-processing-outcome.js"); +const { + runWithTelegramSpooledReplayUpdate, + runWithTelegramUpdateProcessingFrame, + withTelegramSpooledReplayUpdate, +} = await import("./bot-processing-outcome.js"); const { MediaFetchError } = await import("./telegram-media.runtime.js"); let createTelegramBot: ( @@ -870,6 +873,123 @@ describe("createTelegramBot channel_post media", () => { } }); + it("durably retries every spooled album update when shutdown aborts a download", async () => { + setOpenChannelPostConfig(); + sendMessageSpy.mockClear(); + replySpy.mockClear(); + + const shutdown = new AbortController(); + saveRemoteMedia.mockImplementationOnce(async () => { + shutdown.abort(); + throw Object.assign(new Error("aborted"), { name: "AbortError" }); + }); + + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + try { + createTelegramBot({ + token: "tok", + testTimings: TELEGRAM_TEST_TIMINGS, + fetchAbortSignal: shutdown.signal, + }); + const handler = getOnHandler("channel_post") as ( + ctx: Record, + ) => Promise; + const firstUpdate = { update_id: 98079 }; + const secondUpdate = { update_id: 98080 }; + const first = await runWithTelegramSpooledReplayUpdate(firstUpdate, () => + handler({ + ...createChannelPostContext({ + messageId: 98079, + caption: "shutdown album", + date: 1736380800, + mediaGroupId: "shutdown-album-1", + photoFileId: "p1", + }), + update: firstUpdate, + }), + ); + const second = await runWithTelegramSpooledReplayUpdate(secondUpdate, () => + handler({ + ...createChannelPostContext({ + messageId: 98080, + date: 1736380801, + mediaGroupId: "shutdown-album-1", + photoFileId: "p2", + }), + update: secondUpdate, + }), + ); + + expect(first.deferredWork).toBeDefined(); + expect(second.deferredWork).toBeDefined(); + if (!first.deferredWork || !second.deferredWork) { + throw new Error("Expected both album updates to register durable replay work"); + } + await flushChannelPostMediaGroup(setTimeoutSpy); + + const [firstResult, secondResult] = await Promise.all([ + first.deferredWork.task, + second.deferredWork.task, + ]); + expect(firstResult).toEqual({ + kind: "failed-retryable", + error: expect.any(MediaFetchError), + }); + expect(secondResult).toEqual({ + kind: "failed-retryable", + error: expect.any(MediaFetchError), + }); + expect(sendMessageSpy).not.toHaveBeenCalled(); + expect(replySpy).not.toHaveBeenCalled(); + } finally { + setTimeoutSpy.mockRestore(); + } + }); + + it("keeps live album delivery when classic polling aborts a download", async () => { + setOpenChannelPostConfig(); + sendMessageSpy.mockClear(); + replySpy.mockClear(); + + const shutdown = new AbortController(); + saveRemoteMedia + .mockImplementationOnce(async () => ({ + path: "/tmp/classic-restart-first.jpg", + contentType: "image/jpeg", + })) + .mockImplementationOnce(async () => { + shutdown.abort(); + throw Object.assign(new Error("aborted"), { name: "AbortError" }); + }); + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + try { + createTelegramBot({ + token: "tok", + testTimings: TELEGRAM_TEST_TIMINGS, + fetchAbortSignal: shutdown.signal, + }); + const handler = getOnHandler("channel_post") as ( + ctx: Record, + ) => Promise; + await queueChannelPostAlbum(handler, { + caption: "classic restart album", + mediaGroupId: "classic-restart-album-1", + firstMessageId: 98081, + secondMessageId: 98082, + }); + await flushChannelPostMediaGroup(setTimeoutSpy); + await waitForMockCalls(replySpy, 1); + + expect(replySpy).toHaveBeenCalledTimes(1); + expect(replyPayload()).toMatchObject({ + Body: expect.stringContaining("classic restart album"), + MediaPaths: ["/tmp/classic-restart-first.jpg"], + }); + } finally { + setTimeoutSpy.mockRestore(); + } + }); + it("drops the media group when a non-recoverable media error occurs", async () => { replySpy.mockReset(); setOpenChannelPostConfig(); diff --git a/extensions/telegram/src/bot.test.ts b/extensions/telegram/src/bot.test.ts index 396b26b6d079..b087db2542fb 100644 --- a/extensions/telegram/src/bot.test.ts +++ b/extensions/telegram/src/bot.test.ts @@ -50,6 +50,8 @@ const { wasSentByBot, } = await import("./bot.create-telegram-bot.test-harness.js"); const { recordOutboundMessageForPromptContext } = await import("./outbound-message-context.js"); +const { runWithTelegramUpdateProcessingFrame, withTelegramSpooledReplayUpdate } = + await import("./bot-processing-outcome.js"); let createTelegramBotBase: typeof import("./bot-core.js").createTelegramBotCore; let setTelegramBotRuntimeForTest: typeof import("./bot-core.js").setTelegramBotRuntimeForTest; @@ -2989,6 +2991,9 @@ describe("createTelegramBot", () => { onSpy.mockClear(); replySpy.mockClear(); getFileSpy.mockClear(); + const botShutdown = new AbortController(); + const mediaAbort = new AbortController(); + let replyGetFileSignal: AbortSignal | undefined; loadWebMedia.mockResolvedValueOnce({ path: "/tmp/reply-photo.png", contentType: "image/png" }); const mediaFetch = vi.fn( @@ -3003,6 +3008,8 @@ describe("createTelegramBot", () => { try { createTelegramBot({ token: "tok", + fetchAbortSignal: botShutdown.signal, + mediaAbortSignal: mediaAbort.signal, telegramTransport: { fetch: mediaFetch as typeof fetch, sourceFetch: mediaFetch as typeof fetch, @@ -3025,7 +3032,15 @@ describe("createTelegramBot", () => { me: { username: "openclaw_bot" }, getFile: async () => ({}), }); + replyGetFileSignal = mockArg( + getFileSpy as unknown as MockCallSource, + 0, + 1, + "reply getFile signal", + ) as AbortSignal; + expect(replyGetFileSignal.aborted).toBe(false); } finally { + mediaAbort.abort(); ssrfMock.mockRestore(); } @@ -3041,11 +3056,147 @@ describe("createTelegramBot", () => { ReplyToBody?: string; }; expect(payload.ReplyToBody).toBe(""); - expect(getFileSpy).toHaveBeenCalledWith("reply-photo-1"); + expect(getFileSpy).toHaveBeenCalledWith("reply-photo-1", expect.any(AbortSignal)); + expect(replyGetFileSignal?.aborted).toBe(true); + expect(botShutdown.signal.aborted).toBe(false); + botShutdown.abort(); expect(loadWebMedia).not.toHaveBeenCalled(); expect(mediaFetch).toHaveBeenCalledTimes(1); }); + it("dispatches the current text when best-effort reply media times out", async () => { + onSpy.mockClear(); + replySpy.mockClear(); + getFileSpy.mockClear(); + const timeout = Object.assign(new Error("media response headers timed out"), { + name: "TimeoutError", + }); + const mediaFetch = vi.fn(async () => { + throw timeout; + }); + const ssrfMock = mockPinnedHostnameResolution(); + + try { + createTelegramBot({ + token: "tok", + telegramTransport: { + fetch: mediaFetch as typeof fetch, + sourceFetch: mediaFetch as typeof fetch, + close: async () => {}, + }, + }); + const handler = getOnHandler("message") as (ctx: Record) => Promise; + + await handler({ + message: { + chat: { id: 7, type: "private" }, + text: "continue without the old image", + date: 1736380800, + reply_to_message: { + message_id: 9001, + photo: [{ file_id: "reply-photo-1" }], + from: { first_name: "Ada" }, + }, + }, + me: { username: "openclaw_bot" }, + getFile: async () => ({}), + }); + } finally { + ssrfMock.mockRestore(); + } + + expect(mediaFetch).toHaveBeenCalledTimes(1); + expect(getFileSpy).toHaveBeenCalledWith("reply-photo-1", expect.any(AbortSignal)); + expect(replySpy).toHaveBeenCalledTimes(1); + const payload = mockMsgContextArg(replySpy as unknown as MockCallSource, 0, 0, "replySpy call"); + expect(payload.Body).toContain("continue without the old image"); + }); + + it("dispatches the current text when classic polling aborts reply media", async () => { + onSpy.mockClear(); + replySpy.mockClear(); + getFileSpy.mockClear(); + const botShutdown = new AbortController(); + getFileSpy.mockImplementationOnce(async () => { + botShutdown.abort(); + throw Object.assign(new Error("aborted"), { name: "AbortError" }); + }); + + createTelegramBot({ token: "tok", fetchAbortSignal: botShutdown.signal }); + const handler = getOnHandler("message") as (ctx: Record) => Promise; + + const { result } = await runWithTelegramUpdateProcessingFrame(() => + handler({ + message: { + chat: { id: 7, type: "private" }, + text: "continue after polling restart", + date: 1736380800, + reply_to_message: { + message_id: 9001, + photo: [{ file_id: "reply-photo-1" }], + from: { first_name: "Ada" }, + }, + }, + me: { username: "openclaw_bot" }, + getFile: async () => ({}), + }), + ); + + expect(result).toEqual({ kind: "completed" }); + expect(getFileSpy).toHaveBeenCalledWith("reply-photo-1", expect.any(AbortSignal)); + expect(replySpy).toHaveBeenCalledTimes(1); + const payload = mockMsgContextArg(replySpy as unknown as MockCallSource, 0, 0, "replySpy call"); + expect(payload.Body).toContain("continue after polling restart"); + }); + + it("durably retries a spooled reply when shutdown aborts reply media", async () => { + onSpy.mockClear(); + replySpy.mockClear(); + getFileSpy.mockClear(); + const botShutdown = new AbortController(); + const mediaAbort = new AbortController(); + getFileSpy.mockImplementationOnce(async () => { + botShutdown.abort(); + throw new Error("Bad Request: file is too big"); + }); + + createTelegramBot({ + token: "tok", + fetchAbortSignal: botShutdown.signal, + mediaAbortSignal: mediaAbort.signal, + }); + const handler = getOnHandler("message") as (ctx: Record) => Promise; + const update = { + update_id: 98081, + message: { + chat: { id: 7, type: "private" }, + text: "keep the old image", + date: 1736380800, + reply_to_message: { + message_id: 9001, + photo: [{ file_id: "reply-photo-1" }], + from: { first_name: "Ada" }, + }, + }, + }; + + const { result } = await runWithTelegramUpdateProcessingFrame(() => + withTelegramSpooledReplayUpdate(update, () => + handler({ + update, + message: update.message, + me: { username: "openclaw_bot" }, + getFile: async () => ({}), + }), + ), + ); + + expect(result).toEqual({ kind: "failed-retryable", error: expect.any(Error) }); + expect(getFileSpy).toHaveBeenCalledWith("reply-photo-1", expect.any(AbortSignal)); + expect(replySpy).not.toHaveBeenCalled(); + expect(mediaAbort.signal.aborted).toBe(false); + }); + it("hydrates reply chains from cached Telegram messages", async () => { onSpy.mockClear(); replySpy.mockClear(); @@ -3164,7 +3315,7 @@ describe("createTelegramBot", () => { expect(messagesById.get("9000")?.media_path).toMatch(/^media:\/\/inbound\//); expect(messagesById.get("9000")?.media_path).not.toBe(payload.ReplyChain?.[1]?.mediaPath); expect(messagesById.get("9000")?.media_ref).toBeUndefined(); - expect(getFileSpy).toHaveBeenCalledWith("root-photo-1"); + expect(getFileSpy).toHaveBeenCalledWith("root-photo-1", expect.any(AbortSignal)); expect(mediaFetch).toHaveBeenCalledTimes(1); }); @@ -3322,7 +3473,7 @@ describe("createTelegramBot", () => { is_reply_target: true, }); if (expectHydrated) { - expect(getFileSpy).toHaveBeenCalledWith("generated-photo-1"); + expect(getFileSpy).toHaveBeenCalledWith("generated-photo-1", expect.any(AbortSignal)); expect(mediaFetch).toHaveBeenCalledTimes(1); } else { expect(getFileSpy).not.toHaveBeenCalled(); @@ -3582,7 +3733,7 @@ describe("createTelegramBot", () => { if (expectHydrated) { expect(replyMessage?.media_path).toMatch(/^media:\/\/inbound\//); expect(replyMessage?.media_ref).toBeUndefined(); - expect(getFileSpy).toHaveBeenCalledWith("allowed-photo-1"); + expect(getFileSpy).toHaveBeenCalledWith("allowed-photo-1", expect.any(AbortSignal)); expect(mediaFetch).toHaveBeenCalledTimes(1); } else { expect(replyMessage?.media_path).toBeUndefined(); @@ -3729,7 +3880,7 @@ describe("createTelegramBot", () => { await flushTimer?.(); await replyDelivered; - expect(getFileSpy).toHaveBeenCalledWith("reply-photo-1"); + expect(getFileSpy).toHaveBeenCalledWith("reply-photo-1", expect.any(AbortSignal)); expect(mediaFetch).toHaveBeenCalled(); } finally { setTimeoutSpy.mockRestore(); diff --git a/extensions/telegram/src/bot.types.ts b/extensions/telegram/src/bot.types.ts index 355fea6fccf0..67f02899c2ef 100644 --- a/extensions/telegram/src/bot.types.ts +++ b/extensions/telegram/src/bot.types.ts @@ -20,6 +20,8 @@ export type TelegramBotOptions = { botInfo?: TelegramBotInfo; /** Signal to abort in-flight Telegram API fetch requests (e.g. getUpdates) on shutdown. */ fetchAbortSignal?: AbortSignal; + /** Signal to abort inbound media resolution without cancelling adopted-turn Bot API calls. */ + mediaAbortSignal?: AbortSignal; /** Minimum grammY client timeout when timeoutSeconds is configured on long-polling bots. */ minimumClientTimeoutSeconds?: number; updateOffset?: { diff --git a/extensions/telegram/src/bot/delivery.resolve-media-retry.test.ts b/extensions/telegram/src/bot/delivery.resolve-media-retry.test.ts index 28ad0688d9c0..2c089103d1f4 100644 --- a/extensions/telegram/src/bot/delivery.resolve-media-retry.test.ts +++ b/extensions/telegram/src/bot/delivery.resolve-media-retry.test.ts @@ -594,8 +594,11 @@ describe("resolveMedia getFile retry", () => { fetchImpl: callerFetch, dispatcherAttempts, trustExplicitProxyDns: true, + responseHeaderTimeoutMs: 120_000, readIdleTimeoutMs: 30_000, }); + expect(params.timeoutMs).toBeUndefined(); + expect(params.retry).toBeUndefined(); expect(typeof params.shouldRetryFetchError).toBe("function"); expectFetchSsrfPolicyFields({ allowRfc2544BenchmarkRange: true, @@ -625,6 +628,49 @@ describe("resolveMedia getFile retry", () => { expectReadRemoteMediaBufferFields({ fetchImpl: callerFetch }); }); + it.each([ + { mediaField: "document" as const, filePath: "documents/file_42.pdf" }, + { mediaField: "sticker" as const, filePath: "stickers/file_0.webp" }, + ])("keeps the session abort signal attached to $mediaField downloads", async (scenario) => { + const shutdown = new AbortController(); + const getFile = vi.fn().mockResolvedValue({ file_path: scenario.filePath }); + readRemoteMediaBuffer.mockResolvedValueOnce({ + buffer: Buffer.from("media"), + contentType: scenario.mediaField === "sticker" ? "image/webp" : "application/pdf", + fileName: scenario.filePath.split("/").at(-1), + }); + saveMediaBuffer.mockResolvedValueOnce({ + path: `/tmp/${scenario.filePath.split("/").at(-1)}`, + contentType: scenario.mediaField === "sticker" ? "image/webp" : "application/pdf", + }); + + await resolveMediaWithDefaults(makeCtx(scenario.mediaField, getFile), { + abortSignal: shutdown.signal, + }); + + expect(requireReadRemoteMediaBufferParams()).toMatchObject({ + requestInit: { signal: shutdown.signal }, + responseHeaderTimeoutMs: 120_000, + readIdleTimeoutMs: 30_000, + }); + }); + + it("omits nested download retries so callers own failure handling", async () => { + const timeout = Object.assign(new Error("request timed out"), { name: "TimeoutError" }); + const fetchError = Object.assign(new Error("failed to fetch media", { cause: timeout }), { + name: "MediaFetchError", + code: "fetch_failed", + }); + const getFile = vi.fn().mockResolvedValue({ file_path: "documents/first.pdf" }); + readRemoteMediaBuffer.mockRejectedValueOnce(fetchError); + + await expect(resolveMediaWithDefaults(makeCtx("document", getFile))).rejects.toBe(fetchError); + + expect(readRemoteMediaBuffer).toHaveBeenCalledTimes(1); + expect(saveRemoteMedia).toHaveBeenCalledTimes(1); + expect(requireReadRemoteMediaBufferParams().retry).toBeUndefined(); + }); + it("allows an explicit Telegram apiRoot host without broadening the default SSRF allowlist", async () => { const getFile = vi.fn().mockResolvedValue({ file_path: "documents/file_42.pdf" }); readRemoteMediaBuffer.mockResolvedValueOnce({ diff --git a/extensions/telegram/src/bot/delivery.resolve-media.ts b/extensions/telegram/src/bot/delivery.resolve-media.ts index 1af89640b62c..574e08fe8f92 100644 --- a/extensions/telegram/src/bot/delivery.resolve-media.ts +++ b/extensions/telegram/src/bot/delivery.resolve-media.ts @@ -184,6 +184,8 @@ function resolveRequiredTelegramTransport(transport?: TelegramTransport): Telegr /** Default idle timeout for Telegram media downloads (30 seconds). */ const TELEGRAM_DOWNLOAD_IDLE_TIMEOUT_MS = 30_000; +/** Maximum wait for Telegram media response headers (120 seconds). */ +const TELEGRAM_DOWNLOAD_RESPONSE_HEADER_TIMEOUT_MS = 120_000; function usesTrustedTelegramExplicitProxy(transport: TelegramTransport): boolean { return ( @@ -274,6 +276,7 @@ async function downloadAndSaveTelegramFile(params: { apiRoot?: string; trustedLocalFileRoots?: readonly string[]; dangerouslyAllowPrivateNetwork?: boolean; + abortSignal?: AbortSignal; }) { const trustedLocalFile = resolveTrustedLocalTelegramRoot( params.filePath, @@ -345,17 +348,12 @@ async function downloadAndSaveTelegramFile(params: { dispatcherAttempts: transport.dispatcherAttempts, trustExplicitProxyDns: usesTrustedTelegramExplicitProxy(transport), shouldRetryFetchError: shouldRetryTelegramTransportFallback, - retry: { - attempts: 3, - minDelayMs: 1000, - maxDelayMs: 4000, - jitter: 0.2, - label: "telegram:media-download", - onRetry: ({ attempt, maxAttempts }) => - logVerbose(`telegram: media download retry ${attempt}/${maxAttempts}`), - }, + // The update spool and best-effort album/reply callers own failure handling. + // Nested retries would multiply this header deadline before those owners act. + ...(params.abortSignal ? { requestInit: { signal: params.abortSignal } } : {}), filePathHint: params.filePath, maxBytes: params.maxBytes, + responseHeaderTimeoutMs: TELEGRAM_DOWNLOAD_RESPONSE_HEADER_TIMEOUT_MS, readIdleTimeoutMs: TELEGRAM_DOWNLOAD_IDLE_TIMEOUT_MS, ssrfPolicy: buildTelegramMediaSsrfPolicy(params.apiRoot, params.dangerouslyAllowPrivateNetwork), fallbackContentType: params.mimeType, @@ -409,6 +407,7 @@ async function resolveStickerMedia(params: { apiRoot: params.apiRoot, trustedLocalFileRoots: params.trustedLocalFileRoots, dangerouslyAllowPrivateNetwork: params.dangerouslyAllowPrivateNetwork, + abortSignal, }); // Check sticker cache for existing description @@ -516,6 +515,7 @@ export async function resolveMedia(params: { apiRoot, trustedLocalFileRoots, dangerouslyAllowPrivateNetwork, + abortSignal, }); const placeholder = saved.contentType?.startsWith("audio/") ? "" diff --git a/extensions/telegram/src/polling-session.test.ts b/extensions/telegram/src/polling-session.test.ts index 74561fe310c0..8c8ab8108b8e 100644 --- a/extensions/telegram/src/polling-session.test.ts +++ b/extensions/telegram/src/polling-session.test.ts @@ -3510,6 +3510,34 @@ describe("TelegramPollingSession", () => { const runPromise = session.runUntilAbort(); await vi.waitFor(() => expect(createWorker).toHaveBeenCalledTimes(2)); + const firstFetchSignal = mockObjectArg( + createTelegramBotMock, + "first createTelegramBot", + 0, + ).fetchAbortSignal; + const firstMediaSignal = mockObjectArg( + createTelegramBotMock, + "first createTelegramBot", + 0, + ).mediaAbortSignal; + const secondFetchSignal = mockObjectArg( + createTelegramBotMock, + "second createTelegramBot", + 1, + ).fetchAbortSignal; + const secondMediaSignal = mockObjectArg( + createTelegramBotMock, + "second createTelegramBot", + 1, + ).mediaAbortSignal; + expect(firstFetchSignal).toBeInstanceOf(AbortSignal); + expect(firstMediaSignal).toBeInstanceOf(AbortSignal); + expect(secondFetchSignal).toBeInstanceOf(AbortSignal); + expect(secondMediaSignal).toBeInstanceOf(AbortSignal); + expect((firstFetchSignal as AbortSignal).aborted).toBe(false); + expect((firstMediaSignal as AbortSignal).aborted).toBe(true); + expect((secondFetchSignal as AbortSignal).aborted).toBe(false); + expect((secondMediaSignal as AbortSignal).aborted).toBe(false); expectLogIncludes(log, "isolated polling ingress failed: worker crashed"); expect( statusPatches(setStatus).some( @@ -3520,12 +3548,239 @@ describe("TelegramPollingSession", () => { abort.abort(); await vi.advanceTimersByTimeAsync(20_000); await runPromise; + expect((firstFetchSignal as AbortSignal).aborted).toBe(true); + expect((secondFetchSignal as AbortSignal).aborted).toBe(true); + expect((secondMediaSignal as AbortSignal).aborted).toBe(true); } finally { vi.useRealTimers(); await fs.rm(tempDir, { recursive: true, force: true }); } }); + it("waits for a fresh bot before draining updates after an isolated worker crash", async () => { + vi.useFakeTimers({ shouldAdvanceTime: true }); + const abort = new AbortController(); + const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-")); + let releaseBackoff: (() => void) | undefined; + const backoff = new Promise((resolve) => { + releaseBackoff = resolve; + }); + sleepWithAbortMock.mockImplementationOnce(async () => { + await backoff; + return undefined; + }); + + let firstMediaSignal: AbortSignal | undefined; + let rejectFirstWorker: ((err: Error) => void) | undefined; + const firstWorkerDone = new Promise((_resolve, reject) => { + rejectFirstWorker = reject; + }); + const firstHandleUpdate = vi.fn(async () => { + rejectFirstWorker?.(new Error("worker crashed")); + if (!firstMediaSignal) { + throw new Error("Expected the first polling cycle signal"); + } + await waitForAbortSignal(firstMediaSignal); + throw Object.assign(new Error("aborted"), { name: "AbortError" }); + }); + const secondHandleUpdate = vi.fn(async () => undefined); + const createBot = (handleUpdate: (update: { update_id?: number }) => Promise) => ({ + api: { + deleteWebhook: vi.fn(async () => true), + config: { use: vi.fn() }, + }, + init: vi.fn(async () => undefined), + handleUpdate, + stop: vi.fn(async () => undefined), + }); + createTelegramBotMock + .mockImplementationOnce((opts: { mediaAbortSignal?: AbortSignal }) => { + firstMediaSignal = opts.mediaAbortSignal; + return createBot(firstHandleUpdate); + }) + .mockReturnValueOnce(createBot(secondHandleUpdate)); + + let workerIndex = 0; + let stopSecondWorker: (() => void) | undefined; + const secondWorkerDone = new Promise((resolve) => { + stopSecondWorker = resolve; + }); + const createWorker = vi.fn(() => { + workerIndex += 1; + if (workerIndex === 1) { + return { + onMessage: vi.fn(() => () => undefined), + stop: vi.fn(async () => undefined), + task: vi.fn(async () => await firstWorkerDone), + }; + } + return { + onMessage: vi.fn(() => () => undefined), + stop: vi.fn(async () => { + stopSecondWorker?.(); + }), + task: vi.fn(async () => await secondWorkerDone), + }; + }); + + try { + const session = createPollingSession({ + abortSignal: abort.signal, + isolatedIngress: { + enabled: true, + spoolDir: tempDir, + createWorker, + drainIntervalMs: 10, + }, + }); + const runPromise = session.runUntilAbort(); + await vi.waitFor(() => expect(createWorker).toHaveBeenCalledTimes(1)); + + await writeSpooledTestUpdates(tempDir, [ + topicUpdate(42, 10, "crash the old bot"), + topicUpdate(43, 11, "wait for the fresh bot"), + ]); + await vi.advanceTimersByTimeAsync(50); + await vi.waitFor(() => expect(firstHandleUpdate).toHaveBeenCalledTimes(1)); + await vi.waitFor(() => expect(sleepWithAbortMock).toHaveBeenCalledTimes(1)); + await vi.advanceTimersByTimeAsync(5_000); + expect(firstHandleUpdate).toHaveBeenCalledTimes(1); + await vi.waitFor(async () => { + const unstarted = ( + await listTelegramSpooledUpdates({ spoolDir: tempDir, limit: "all" }) + ).find((update) => update.updateId === 43); + expect(unstarted).toMatchObject({ attempts: 0 }); + expect(unstarted?.lastAttemptAt).toBeUndefined(); + expect(unstarted?.lastError).toBeUndefined(); + }); + + releaseBackoff?.(); + await vi.advanceTimersByTimeAsync(2_000); + await vi.waitFor(() => expect(secondHandleUpdate).toHaveBeenCalledTimes(2)); + abort.abort(); + await vi.advanceTimersByTimeAsync(20_000); + await runPromise; + + expect(createWorker).toHaveBeenCalledTimes(2); + expect(firstHandleUpdate).toHaveBeenCalledTimes(1); + expect(await pendingUpdateIds(tempDir, "all")).toEqual([]); + } finally { + releaseBackoff?.(); + abort.abort(); + stopSecondWorker?.(); + vi.useRealTimers(); + await fs.rm(tempDir, { recursive: true, force: true }); + } + }); + + it("keeps adopted-turn Bot API delivery alive when an isolated worker crashes", async () => { + vi.useFakeTimers({ shouldAdvanceTime: true }); + const abort = new AbortController(); + const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-")); + let releaseBackoff: (() => void) | undefined; + const backoff = new Promise((resolve) => { + releaseBackoff = resolve; + }); + sleepWithAbortMock.mockImplementationOnce(async () => { + await backoff; + return undefined; + }); + + let adopted: (() => void) | undefined; + const adoptedTurn = new Promise((resolve) => { + adopted = resolve; + }); + let deliveryResolve: (() => void) | undefined; + let deliveryReject: ((err: unknown) => void) | undefined; + const delivery = new Promise((resolve, reject) => { + deliveryResolve = resolve; + deliveryReject = reject; + }); + let firstFetchSignal: AbortSignal | undefined; + let firstMediaSignal: AbortSignal | undefined; + const sendMessage = vi.fn(async () => { + if (firstFetchSignal?.aborted) { + throw new Error("adopted-turn Bot API client was aborted"); + } + }); + createTelegramBotMock.mockImplementationOnce( + (opts: { fetchAbortSignal?: AbortSignal; mediaAbortSignal?: AbortSignal }) => { + firstFetchSignal = opts.fetchAbortSignal; + firstMediaSignal = opts.mediaAbortSignal; + const api = { + deleteWebhook: vi.fn(async () => true), + sendMessage, + config: { use: vi.fn() }, + }; + return { + api, + init: vi.fn(async () => undefined), + handleUpdate: vi.fn(async (update: { update_id?: number }) => { + const participant = createTelegramSpooledReplayDeferredParticipant( + `test-adopted-delivery:${update.update_id}`, + ); + if (!participant || !firstMediaSignal) { + throw new Error("Expected a spooled participant and media signal"); + } + participant.settle({ kind: "completed" }); + adopted?.(); + void (async () => { + try { + await waitForAbortSignal(firstMediaSignal); + // Streaming, edits, and native-quote fallbacks retain this adopted bot client. + await api.sendMessage(); + deliveryResolve?.(); + } catch (err) { + deliveryReject?.(err); + } + })(); + }), + stop: vi.fn(async () => undefined), + }; + }, + ); + + const createWorker = vi.fn(() => ({ + onMessage: vi.fn(() => () => undefined), + stop: vi.fn(async () => undefined), + task: vi.fn(async () => { + await adoptedTurn; + throw new Error("worker crashed after turn adoption"); + }), + })); + + try { + await writeSpooledTestUpdates(tempDir, [topicUpdate(42, 10, "finish after worker crash")]); + const session = createPollingSession({ + abortSignal: abort.signal, + isolatedIngress: { + enabled: true, + spoolDir: tempDir, + createWorker, + drainIntervalMs: 10, + }, + }); + const runPromise = session.runUntilAbort(); + + await delivery; + expect(sendMessage).toHaveBeenCalledTimes(1); + expect(firstMediaSignal?.aborted).toBe(true); + expect(firstFetchSignal?.aborted).toBe(false); + await vi.waitFor(async () => expect(await pendingUpdateIds(tempDir, "all")).toEqual([])); + + abort.abort(); + releaseBackoff?.(); + await vi.advanceTimersByTimeAsync(20_000); + await runPromise; + expect(firstFetchSignal?.aborted).toBe(true); + } finally { + releaseBackoff?.(); + abort.abort(); + vi.useRealTimers(); + await fs.rm(tempDir, { recursive: true, force: true }); + } + }); + it("treats isolated ingress worker rejection after abort as clean shutdown", async () => { vi.useFakeTimers({ shouldAdvanceTime: true }); const abort = new AbortController(); diff --git a/extensions/telegram/src/polling-session.ts b/extensions/telegram/src/polling-session.ts index c756b8e0fbb0..71c3dc437a4e 100644 --- a/extensions/telegram/src/polling-session.ts +++ b/extensions/telegram/src/polling-session.ts @@ -37,6 +37,7 @@ import { TELEGRAM_SPOOLED_RETRY_MAX_ATTEMPTS, } from "./spooled-update-retry-policy.js"; import { + abandonTelegramSpooledUpdateClaim, claimNextTelegramSpooledUpdate, completeTelegramSpooledUpdate, failTelegramSpooledUpdateClaim, @@ -328,7 +329,7 @@ export class TelegramPollingSession { #webhookCleared = false; #forceRestarted = false; #activeRunner: ReturnType | undefined; - #activeFetchAbort: AbortController | undefined; + #activeCycleAbort: AbortController | undefined; #spooledUpdateHandlerKeys = new Set(); #deferredSpooledUpdateClaimKeys = new Set(); #transportState: TelegramPollingTransportState; @@ -372,7 +373,7 @@ export class TelegramPollingSession { } abortActiveFetch() { - this.#activeFetchAbort?.abort(); + this.#activeCycleAbort?.abort(); } async runUntilAbort(): Promise { @@ -493,8 +494,16 @@ export class TelegramPollingSession { } async #createPollingBot(): Promise { - const fetchAbortController = new AbortController(); - this.#activeFetchAbort = fetchAbortController; + const cycleAbortController = new AbortController(); + this.#activeCycleAbort = cycleAbortController; + const cycleAbortSignal = this.opts.abortSignal + ? AbortSignal.any([this.opts.abortSignal, cycleAbortController.signal]) + : cycleAbortController.signal; + // Isolated turns can outlive their polling worker after adoption. Keep their + // Bot API client session-owned while media remains cycle-owned and retryable. + const botApiAbortSignal = this.opts.isolatedIngress?.enabled + ? this.opts.abortSignal + : cycleAbortSignal; const telegramTransport = this.#transportState.acquireForNextCycle(); const persistedLastUpdateId = this.opts.getLastUpdateId(); const lastUpdateId = this.opts.isolatedIngress?.enabled ? null : persistedLastUpdateId; @@ -511,15 +520,16 @@ export class TelegramPollingSession { config: this.opts.config, accountId: this.opts.accountId, botInfo: this.opts.botInfo, - fetchAbortSignal: fetchAbortController.signal, + ...(botApiAbortSignal ? { fetchAbortSignal: botApiAbortSignal } : {}), + mediaAbortSignal: cycleAbortSignal, minimumClientTimeoutSeconds: TELEGRAM_POLLING_CLIENT_TIMEOUT_FLOOR_SECONDS, ...(updateOffset ? { updateOffset } : {}), telegramTransport, }); } catch (err) { await this.#waitBeforeRetryOnRecoverableSetupError(err, "Telegram setup network error"); - if (this.#activeFetchAbort === fetchAbortController) { - this.#activeFetchAbort = undefined; + if (this.#activeCycleAbort === cycleAbortController) { + this.#activeCycleAbort = undefined; } return undefined; } @@ -900,6 +910,7 @@ export class TelegramPollingSession { async #drainSpooledUpdates(params: { bot: TelegramBot; isDrainHealthy: () => boolean; + shouldStop: () => boolean; spoolDir: string; }): Promise { const activeLaneKeys = this.#activeSpooledUpdateLaneKeysForSpool(params.spoolDir); @@ -948,7 +959,7 @@ export class TelegramPollingSession { ]); let started = 0; while (started < TELEGRAM_SPOOLED_DRAIN_START_LIMIT) { - if (this.opts.abortSignal?.aborted) { + if (params.shouldStop() || this.opts.abortSignal?.aborted) { break; } const claimedUpdate = await this.#claimNextSpooledUpdate({ @@ -959,13 +970,21 @@ export class TelegramPollingSession { if (!claimedUpdate) { break; } + if (params.shouldStop() || this.opts.abortSignal?.aborted) { + try { + await abandonTelegramSpooledUpdateClaim(claimedUpdate); + } catch (err) { + this.opts.log( + `[telegram][diag] spooled update ${claimedUpdate.updateId} could not be requeued after its polling cycle ended: ${formatErrorMessage(err)}`, + ); + } + break; + } const laneKey = this.#spooledUpdateLaneKey(claimedUpdate); const handlerKey = buildSpooledUpdateHandlerKey({ spoolDir: params.spoolDir, laneKey }); if (activeSpooledUpdateHandlersByLane.has(handlerKey)) { blockedByLane.add(handlerKey); - await releaseTelegramSpooledUpdateClaim(claimedUpdate, { - lastError: "active Telegram spool handler already owns this lane", - }); + await abandonTelegramSpooledUpdateClaim(claimedUpdate); blockedLaneKeys.add(laneKey); continue; } @@ -1141,9 +1160,17 @@ export class TelegramPollingSession { if (!ingress?.enabled) { return this.#runPollingCycle(bot); } + const cycleAbortController = this.#activeCycleAbort; + const abortMedia = () => { + cycleAbortController?.abort(); + }; try { await bot.init(); } catch (err) { + abortMedia(); + if (this.#activeCycleAbort === cycleAbortController) { + this.#activeCycleAbort = undefined; + } const shouldRetry = await this.#waitBeforeRetryOnRecoverableSetupError( err, "Telegram bot init failed", @@ -1196,6 +1223,11 @@ export class TelegramPollingSession { const stalledBacklogKeys = new Set(); let requestImmediateDrain: () => void = () => undefined; let drainRequested = false; + let cycleEnding = false; + const endCycle = () => { + cycleEnding = true; + abortMedia(); + }; const unsubscribe = worker.onMessage((message) => { const ackSpooledUpdate = ( requestId: string, @@ -1268,6 +1300,7 @@ export class TelegramPollingSession { } }); const stopOnAbort = () => { + endCycle(); void stopWorker(); }; this.opts.abortSignal?.addEventListener("abort", stopOnAbort, { once: true }); @@ -1295,6 +1328,7 @@ export class TelegramPollingSession { return; } restartRequested = true; + endCycle(); void stopWorker(); if (!forceCycleTimer) { forceCycleTimer = setTimeout(() => { @@ -1310,7 +1344,7 @@ export class TelegramPollingSession { } }; const drainOnce = async () => { - if (restartRequested || this.opts.abortSignal?.aborted) { + if (cycleEnding || restartRequested || this.opts.abortSignal?.aborted) { return; } if (drainActive) { @@ -1324,6 +1358,7 @@ export class TelegramPollingSession { const drain = await this.#drainSpooledUpdates({ bot, isDrainHealthy, + shouldStop: () => cycleEnding, spoolDir, }); consecutiveDrainFailures = 0; @@ -1373,7 +1408,12 @@ export class TelegramPollingSession { drainHealth.lastCompletedAt = Date.now(); } drainActive = false; - if (drainRequested && !restartRequested && !this.opts.abortSignal?.aborted) { + if ( + drainRequested && + !cycleEnding && + !restartRequested && + !this.opts.abortSignal?.aborted + ) { drainRequested = false; // Handler finalizers clear active lane guards in microtasks; redrain // after them so newly unblocked same-lane rows can claim immediately. @@ -1410,10 +1450,12 @@ export class TelegramPollingSession { try { await Promise.race([worker.task(), forceCyclePromise]); clearForceCycleTimer(); + endCycle(); } catch (err) { if (this.opts.abortSignal?.aborted) { return "exit"; } + endCycle(); // The worker only issues getUpdates, so a 409 is always a duplicate // poller (or stale webhook) conflict. Mirror the classic polling // cycle: re-clear the webhook, rotate the transport (#69787), and @@ -1469,12 +1511,17 @@ export class TelegramPollingSession { clearForceCycleTimer(); unsubscribe(); this.opts.abortSignal?.removeEventListener("abort", stopOnAbort); + // End media work before waiting for durable handlers so every interrupted claim can retry. + endCycle(); await stopWorker(); if (!restartRequested) { await drainOnce(); await waitForGracefulStop(() => this.#waitForSpooledUpdateHandlers()); } await waitForGracefulStop(stopBot); + if (this.#activeCycleAbort === cycleAbortController) { + this.#activeCycleAbort = undefined; + } } } @@ -1508,7 +1555,7 @@ export class TelegramPollingSession { const runner = run(bot, this.opts.runnerOptions); this.opts.log(`[telegram][diag] polling cycle started ${liveness.formatDiagnosticFields()}`); this.#activeRunner = runner; - const fetchAbortController = this.#activeFetchAbort; + const fetchAbortController = this.#activeCycleAbort; const abortFetch = () => { fetchAbortController?.abort(); }; @@ -1657,8 +1704,8 @@ export class TelegramPollingSession { await waitForGracefulStop(stopRunner); await waitForGracefulStop(stopBot); this.#activeRunner = undefined; - if (this.#activeFetchAbort === fetchAbortController) { - this.#activeFetchAbort = undefined; + if (this.#activeCycleAbort === fetchAbortController) { + this.#activeCycleAbort = undefined; } } } diff --git a/extensions/telegram/src/telegram-ingress-spool.ts b/extensions/telegram/src/telegram-ingress-spool.ts index 5f17f272c59c..8c24387f8804 100644 --- a/extensions/telegram/src/telegram-ingress-spool.ts +++ b/extensions/telegram/src/telegram-ingress-spool.ts @@ -341,6 +341,15 @@ export async function releaseTelegramSpooledUpdateClaim( ); } +export async function abandonTelegramSpooledUpdateClaim( + update: ClaimedTelegramSpooledUpdate, +): Promise { + await createTelegramIngressQueue(path.dirname(update.pendingPath)).release( + queueMutationTarget(update), + { recordAttempt: false }, + ); +} + export async function refreshTelegramSpooledUpdateClaim( update: ClaimedTelegramSpooledUpdate, options?: { refreshedAt?: number }, diff --git a/extensions/telegram/src/webhook.test.ts b/extensions/telegram/src/webhook.test.ts index b03f58cc9e34..e8ef620d1fd8 100644 --- a/extensions/telegram/src/webhook.test.ts +++ b/extensions/telegram/src/webhook.test.ts @@ -575,6 +575,41 @@ describe("startTelegramWebhook", () => { ); }); + it("aborts bot media fetches when the webhook stops", async () => { + const callerAbort = new AbortController(); + const started = await startTelegramWebhook({ + token: TELEGRAM_TOKEN, + secret: TELEGRAM_SECRET, + port: 0, + abortSignal: callerAbort.signal, + path: TELEGRAM_WEBHOOK_PATH, + spoolDir: requireWebhookSpoolDir(), + }); + + try { + const botParams = requireRecord( + requireMockCall(createTelegramBotSpy, 0, "createTelegramBot")[0], + "createTelegramBot params", + ); + const fetchAbortSignal = botParams.fetchAbortSignal; + expect(fetchAbortSignal).toBeInstanceOf(AbortSignal); + if (!(fetchAbortSignal instanceof AbortSignal)) { + throw new Error("expected bot fetch abort signal"); + } + const aborted = new Promise((resolve) => { + fetchAbortSignal.addEventListener("abort", () => resolve(), { once: true }); + }); + + await started.stop(); + + await expect(aborted).resolves.toBeUndefined(); + expect(callerAbort.signal.aborted).toBe(false); + } finally { + await started.stop(); + callerAbort.abort(); + } + }); + it("keeps local listener alive and retries when setWebhook has a recoverable startup failure", async () => { const runtimeLog = vi.fn(); const runtimeError = vi.fn(); diff --git a/extensions/telegram/src/webhook.ts b/extensions/telegram/src/webhook.ts index c5bc6fe7b758..ea6885db417a 100644 --- a/extensions/telegram/src/webhook.ts +++ b/extensions/telegram/src/webhook.ts @@ -700,10 +700,15 @@ export async function startTelegramWebhook(opts: { closeTransportPromise ??= telegramTransport.close(); return closeTransportPromise; }; + const botAbortController = new AbortController(); + const botFetchAbortSignal = opts.abortSignal + ? AbortSignal.any([opts.abortSignal, botAbortController.signal]) + : botAbortController.signal; const bot = createTelegramBot({ token: opts.token, runtime, proxyFetch: opts.fetch, + fetchAbortSignal: botFetchAbortSignal, config: opts.config, accountId: opts.accountId, telegramTransport, @@ -716,6 +721,7 @@ export async function startTelegramWebhook(opts: { retryPolicy: webhookRegistrationRetryPolicy, }); } catch (err) { + botAbortController.abort(); await bot.stop(); await closeTransportOnce(); throw err; @@ -976,6 +982,7 @@ export async function startTelegramWebhook(opts: { if (shutDown) { return; } + botAbortController.abort(); shutDown = true; if (drainTimer) { clearInterval(drainTimer); diff --git a/src/channels/message/ingress-queue.test.ts b/src/channels/message/ingress-queue.test.ts index 689fbb8b5364..afa4809bd059 100644 --- a/src/channels/message/ingress-queue.test.ts +++ b/src/channels/message/ingress-queue.test.ts @@ -248,6 +248,18 @@ describe("channel ingress queue", () => { lastAttemptAt: 20, lastError: "retry", }); + + const reclaimed = await queue.claim("b", { ownerId: "replacement" }); + if (!reclaimed) { + throw new Error("Expected the released ingress event to be claimable"); + } + expect(await queue.release(reclaimed, { recordAttempt: false, releasedAt: 30 })).toBe(true); + expect((await queue.listPending()).find((record) => record.id === "b")).toMatchObject({ + attempts: 1, + lastAttemptAt: 20, + lastError: "retry", + updatedAt: 30, + }); }); }); diff --git a/src/channels/message/ingress-queue.ts b/src/channels/message/ingress-queue.ts index 4368c8a44d22..2f2d8b94fe97 100644 --- a/src/channels/message/ingress-queue.ts +++ b/src/channels/message/ingress-queue.ts @@ -156,7 +156,7 @@ export type ChannelIngressQueue; release( idOrClaim: string | ChannelIngressQueueClaimRef, - options?: { lastError?: string; releasedAt?: number }, + options?: { lastError?: string; releasedAt?: number; recordAttempt?: boolean }, ): Promise; fail( idOrClaim: string | ChannelIngressQueueClaimRef, @@ -769,8 +769,14 @@ export function createChannelIngressQueue< claim_token: null, claim_owner: null, claimed_at: null, - attempts: eb("attempts", "+", 1), - last_attempt_at: releasedAt, + // A claim can lose its owner before processing starts. Returning it + // must not consume retry budget or erase the previous real failure. + ...(releaseOptions?.recordAttempt === false + ? {} + : { + attempts: eb("attempts", "+", 1), + last_attempt_at: releasedAt, + }), ...(releaseOptions?.lastError === undefined ? {} : { last_error: releaseOptions.lastError }), diff --git a/src/infra/net/fetch-guard.ssrf.test.ts b/src/infra/net/fetch-guard.ssrf.test.ts index bb78381522c4..ead965459c47 100644 --- a/src/infra/net/fetch-guard.ssrf.test.ts +++ b/src/infra/net/fetch-guard.ssrf.test.ts @@ -208,6 +208,16 @@ describe("fetchWithSsrFGuard hardening", () => { const createPublicLookup = (): LookupFn => vi.fn(async () => [{ address: "93.184.216.34", family: 4 }]) as unknown as LookupFn; + const createStalledLookup = () => { + let release: (() => void) | undefined; + const lookupFn = vi.fn( + async () => + await new Promise>((resolve) => { + release = () => resolve([{ address: "93.184.216.34", family: 4 }]); + }), + ) as unknown as LookupFn; + return { lookupFn, release: () => release?.() }; + }; const createLoopbackLookup = (): LookupFn => vi.fn(async () => [{ address: "127.0.0.1", family: 4 }]) as unknown as LookupFn; const createIpv6LoopbackLookup = (): LookupFn => @@ -1902,6 +1912,48 @@ describe("fetchWithSsrFGuard hardening", () => { expect(outcome).toBe("TimeoutError"); }); + it("aborts a stalled DNS preflight from the caller signal without dispatching", async () => { + const stalledLookup = createStalledLookup(); + const fetchImpl = vi.fn(async () => okResponse()); + const controller = new AbortController(); + const abortError = Object.assign(new Error("gateway shutdown"), { name: "AbortError" }); + const fetchPromise = fetchWithSsrFGuard({ + url: "https://public.example/resource", + fetchImpl, + lookupFn: stalledLookup.lookupFn, + signal: controller.signal, + }).catch((error: unknown) => error); + + await vi.waitFor(() => expect(stalledLookup.lookupFn).toHaveBeenCalledOnce()); + controller.abort(abortError); + const outcome = await raceWithTimeoutResult(fetchPromise, 250, new Error("hung")); + + expect(outcome).toBe(abortError); + expect(fetchImpl).not.toHaveBeenCalled(); + stalledLookup.release(); + await Promise.resolve(); + expect(fetchImpl).not.toHaveBeenCalled(); + }); + + it("applies timeoutMs while DNS preflight is stalled", async () => { + const stalledLookup = createStalledLookup(); + const fetchImpl = vi.fn(async () => okResponse()); + const fetchPromise = fetchWithSsrFGuard({ + url: "https://public.example/resource", + fetchImpl, + lookupFn: stalledLookup.lookupFn, + timeoutMs: 1, + }).catch((error: unknown) => error); + + const outcome = await raceWithTimeoutResult(fetchPromise, 250, new Error("hung")); + + expect(outcome).toMatchObject({ name: "TimeoutError" }); + expect(fetchImpl).not.toHaveBeenCalled(); + stalledLookup.release(); + await Promise.resolve(); + expect(fetchImpl).not.toHaveBeenCalled(); + }); + it("inherits the configured global stream timeout for guarded direct dispatchers", async () => { try { ensureGlobalUndiciStreamTimeouts({ timeoutMs: 1_900_000 }); diff --git a/src/infra/net/fetch-guard.ts b/src/infra/net/fetch-guard.ts index f5fb72aa7338..d0a42a735240 100644 --- a/src/infra/net/fetch-guard.ts +++ b/src/infra/net/fetch-guard.ts @@ -3,6 +3,8 @@ import type { Dispatcher } from "undici"; import { logWarn } from "../../logger.js"; import { buildTimeoutAbortSignal } from "../../utils/fetch-timeout.js"; +import { createAbortError } from "../abort-signal.js"; +import { toErrorObject } from "../errors.js"; import { normalizeHeadersInitForFetch, normalizeRequestInitHeadersForFetch, @@ -121,6 +123,44 @@ type GuardedFetchPresetOptions = Omit< const DEFAULT_MAX_REDIRECTS = 3; const OPENCLAW_DEBUG_PROXY_ENABLED = "OPENCLAW_DEBUG_PROXY_ENABLED"; +async function runAbortablePreflight(run: () => Promise, signal?: AbortSignal): Promise { + if (!signal) { + return await run(); + } + if (signal.aborted) { + throw signal.reason ?? createAbortError("Guarded fetch aborted during network preflight"); + } + return await new Promise((resolve, reject) => { + let settled = false; + const settle = (complete: () => void) => { + if (settled) { + return; + } + settled = true; + signal.removeEventListener("abort", onAbort); + complete(); + }; + const onAbort = () => + settle(() => + reject( + toErrorObject( + signal.reason ?? createAbortError("Guarded fetch aborted during network preflight"), + "Guarded fetch aborted during network preflight", + ), + ), + ); + signal.addEventListener("abort", onAbort, { once: true }); + if (signal.aborted) { + onAbort(); + return; + } + void run().then( + (value) => settle(() => resolve(value)), + (error: unknown) => settle(() => reject(toErrorObject(error, "Network preflight failed"))), + ); + }); +} + function getRedirectVisitKey(url: string, init: RequestInit | undefined): string { return `${init?.method?.toUpperCase() ?? "GET"} ${url}`; } @@ -216,6 +256,7 @@ async function assertExplicitProxyAllowed( dispatcherPolicy: PinnedDispatcherPolicy | undefined, lookupFn: LookupFn | undefined, policy: SsrFPolicy | undefined, + signal: AbortSignal | undefined, ): Promise { // Explicit proxies are operator-configured, but the proxy host still needs // basic URL and private-network validation before target validation proceeds. @@ -242,10 +283,14 @@ async function assertExplicitProxyAllowed( ...(dispatcherPolicy.allowPrivateProxy === true ? { allowPrivateNetwork: true } : {}), } : undefined; - await resolvePinnedHostnameWithPolicy(parsedProxyUrl.hostname, { - lookupFn, - policy: proxyPolicy, - }); + await runAbortablePreflight( + async () => + await resolvePinnedHostnameWithPolicy(parsedProxyUrl.hostname, { + lookupFn, + policy: proxyPolicy, + }), + signal, + ); } function isRedirectStatus(status: number): boolean { @@ -493,6 +538,15 @@ async function fetchWithSsrFGuardInternal( // Resolve inside the redirect loop so exact-origin trust never carries across origins. const policyForUrl = resolveSsrFPolicyForUrl(parsedUrl, params.policy); const dispatcherPolicy = params.resolveDispatcherPolicy?.(parsedUrl) ?? params.dispatcherPolicy; + const resolvePinnedHostname = async () => + await runAbortablePreflight( + async () => + await resolvePinnedHostnameWithPolicy(parsedUrl.hostname, { + lookupFn: params.lookupFn, + policy: policyForUrl, + }), + signal, + ); try { const usesTrustedExplicitProxyMode = mode === GUARDED_FETCH_MODE.TRUSTED_EXPLICIT_PROXY && @@ -502,7 +556,7 @@ async function fetchWithSsrFGuardInternal( dispatcherPolicy, usesTrustedExplicitProxyMode ? false : params.pinDns, ); - await assertExplicitProxyAllowed(dispatcherPolicy, params.lookupFn, params.policy); + await assertExplicitProxyAllowed(dispatcherPolicy, params.lookupFn, params.policy, signal); const isStrictManagedProxyActive = mode === GUARDED_FETCH_MODE.STRICT && isManagedProxyActive(); const shouldCheckManagedProxyBypass = @@ -538,10 +592,7 @@ async function fetchWithSsrFGuardInternal( dispatcher = createHttp1EnvHttpProxyAgent(undefined, timeoutMs); } else if (canUseManagedProxy) { if (shouldCheckManagedProxyBypass) { - const pinned = await resolvePinnedHostnameWithPolicy(parsedUrl.hostname, { - lookupFn: params.lookupFn, - policy: policyForUrl, - }); + const pinned = await resolvePinnedHostname(); dispatcher = shouldUseConfiguredLocalOriginManagedProxyBypass({ url: parsedUrl, managedProxyBypass: params.managedProxyBypass, @@ -562,16 +613,10 @@ async function fetchWithSsrFGuardInternal( // real fetches continue through pinned DNS below. assertHostnameAllowedWithPolicy(parsedUrl.hostname, policyForUrl); } else if (params.pinDns === false) { - await resolvePinnedHostnameWithPolicy(parsedUrl.hostname, { - lookupFn: params.lookupFn, - policy: policyForUrl, - }); + await resolvePinnedHostname(); dispatcher = createPolicyDispatcherWithoutPinnedDns(dispatcherPolicy, timeoutMs); } else { - const pinned = await resolvePinnedHostnameWithPolicy(parsedUrl.hostname, { - lookupFn: params.lookupFn, - policy: policyForUrl, - }); + const pinned = await resolvePinnedHostname(); dispatcher = createPinnedDispatcher(pinned, dispatcherPolicy, policyForUrl, timeoutMs); } diff --git a/src/media/fetch.test.ts b/src/media/fetch.test.ts index 299b09c30170..958cb7aca636 100644 --- a/src/media/fetch.test.ts +++ b/src/media/fetch.test.ts @@ -69,6 +69,12 @@ function makeLookupFn(): LookupFn { return vi.fn(async () => ({ address: "149.154.167.220", family: 4 })) as unknown as LookupFn; } +function abortReasonError(signal?: AbortSignal | null): Error { + return signal?.reason instanceof Error + ? signal.reason + : new Error("request aborted", { cause: signal?.reason }); +} + function requireFetchGuardRequest(): unknown { const [call] = fetchWithSsrFGuardMock.mock.calls; if (!call) { @@ -233,6 +239,7 @@ describe("readRemoteMediaBuffer", () => { url: string; fetchImpl?: (input: RequestInfo | URL, init?: RequestInit) => Promise; init?: RequestInit; + signal?: AbortSignal; }; if (params.url.startsWith("http://127.0.0.1/")) { throw new Error("Blocked hostname or private/internal/special-use IP address"); @@ -242,7 +249,10 @@ describe("readRemoteMediaBuffer", () => { throw new Error("fetch is not available"); } return { - response: await fetcher(params.url, params.init), + response: await fetcher(params.url, { + ...params.init, + ...(params.signal ? { signal: params.signal } : {}), + }), finalUrl: params.url, release: async () => {}, }; @@ -392,6 +402,155 @@ describe("readRemoteMediaBuffer", () => { }); }); + it("aborts when response headers exceed their deadline", async () => { + vi.useFakeTimers(); + try { + const fetchImpl = vi.fn( + async (_input: RequestInfo | URL, init?: RequestInit) => + await new Promise((_resolve, reject) => { + const signal = init?.signal; + const rejectForAbort = () => reject(abortReasonError(signal)); + if (signal?.aborted) { + rejectForAbort(); + return; + } + signal?.addEventListener("abort", rejectForAbort, { once: true }); + }), + ); + + const result = readRemoteMediaBuffer({ + url: "https://example.com/file.bin", + fetchImpl, + lookupFn: makeLookupFn(), + maxBytes: 1024, + responseHeaderTimeoutMs: 20, + }).catch((error: unknown) => error); + + await vi.advanceTimersByTimeAsync(25); + + await expect(result).resolves.toMatchObject({ + name: "MediaFetchError", + code: "fetch_failed", + cause: { name: "TimeoutError" }, + }); + } finally { + vi.useRealTimers(); + } + }); + + it("clears the response-header deadline while a healthy body keeps progressing", async () => { + vi.useFakeTimers(); + try { + const fetchImpl = vi.fn(async (_input: RequestInfo | URL, init?: RequestInit) => { + const signal = init?.signal; + return new Response( + new ReadableStream({ + start(controller) { + const failForAbort = () => controller.error(signal?.reason); + if (signal?.aborted) { + failForAbort(); + return; + } + signal?.addEventListener("abort", failForAbort, { once: true }); + setTimeout(() => controller.enqueue(new Uint8Array([1])), 25); + setTimeout(() => controller.enqueue(new Uint8Array([2])), 50); + setTimeout(() => { + signal?.removeEventListener("abort", failForAbort); + controller.close(); + }, 75); + }, + }), + { status: 200 }, + ); + }); + + const result = readRemoteMediaBuffer({ + url: "https://example.com/file.bin", + fetchImpl, + lookupFn: makeLookupFn(), + maxBytes: 1024, + responseHeaderTimeoutMs: 10, + readIdleTimeoutMs: 30, + }); + + await vi.advanceTimersByTimeAsync(80); + + await expect(result).resolves.toMatchObject({ buffer: Buffer.from([1, 2]) }); + } finally { + vi.useRealTimers(); + } + }); + + it("propagates a parent abort while waiting for response headers", async () => { + const parent = new AbortController(); + const fetchImpl = vi.fn( + async (_input: RequestInfo | URL, init?: RequestInit) => + await new Promise((_resolve, reject) => { + const signal = init?.signal; + const rejectForAbort = () => reject(abortReasonError(signal)); + if (signal?.aborted) { + rejectForAbort(); + return; + } + signal?.addEventListener("abort", rejectForAbort, { once: true }); + }), + ); + const result = readRemoteMediaBuffer({ + url: "https://example.com/file.bin", + fetchImpl, + requestInit: { signal: parent.signal }, + lookupFn: makeLookupFn(), + maxBytes: 1024, + responseHeaderTimeoutMs: 60_000, + }).catch((error: unknown) => error); + + parent.abort(); + + await expect(result).resolves.toMatchObject({ + name: "MediaFetchError", + code: "fetch_failed", + cause: { name: "AbortError" }, + }); + }); + + it("keeps the parent abort active while reading the response body", async () => { + const parent = new AbortController(); + const fetchImpl = vi.fn(async (_input: RequestInfo | URL, init?: RequestInit) => { + const signal = init?.signal; + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array([1])); + const failForAbort = () => controller.error(signal?.reason); + if (signal?.aborted) { + failForAbort(); + return; + } + signal?.addEventListener("abort", failForAbort, { once: true }); + }, + }), + { status: 200 }, + ); + }); + const result = readRemoteMediaBuffer({ + url: "https://example.com/file.bin", + fetchImpl, + requestInit: { signal: parent.signal }, + lookupFn: makeLookupFn(), + maxBytes: 1024, + responseHeaderTimeoutMs: 60_000, + }).catch((error: unknown) => error); + + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(1)); + parent.abort(); + + await expect(result).resolves.toMatchObject({ + name: "MediaFetchError", + code: "fetch_failed", + cause: { name: "AbortError" }, + }); + }); + it("retries transient fetch failures when retry is enabled", async () => { const transientError = Object.assign(new Error("socket hang up"), { code: "ECONNRESET" }); const fetchImpl = vi @@ -614,10 +773,12 @@ describe("readRemoteMediaBuffer", () => { it("passes request timeout through the guarded fetch path", async () => { const fetchImpl = vi.fn(async () => new Response("ok", { status: 200 })); + const parent = new AbortController(); await readRemoteMediaBuffer({ url: "https://example.com/file.bin", fetchImpl, + requestInit: { signal: parent.signal }, lookupFn: makeLookupFn(), maxBytes: 1024, timeoutMs: 1234, @@ -627,6 +788,7 @@ describe("readRemoteMediaBuffer", () => { expect(requireFetchGuardRequest()).toMatchObject({ url: "https://example.com/file.bin", timeoutMs: 1234, + signal: parent.signal, }); }); @@ -656,6 +818,92 @@ describe("readRemoteMediaBuffer", () => { await expect(fs.readFile(saved.path)).resolves.toStrictEqual(Buffer.from([1, 2, 3, 4])); }); + it("keeps saving a healthy streaming body after the response-header deadline", async () => { + vi.useFakeTimers(); + try { + const fetchImpl = vi.fn(async (_input: RequestInfo | URL, init?: RequestInit) => { + const signal = init?.signal; + return new Response( + new ReadableStream({ + start(controller) { + const failForAbort = () => controller.error(signal?.reason); + if (signal?.aborted) { + failForAbort(); + return; + } + signal?.addEventListener("abort", failForAbort, { once: true }); + setTimeout(() => controller.enqueue(new Uint8Array([1])), 25); + setTimeout(() => controller.enqueue(new Uint8Array([2])), 50); + setTimeout(() => { + signal?.removeEventListener("abort", failForAbort); + controller.close(); + }, 75); + }, + }), + { status: 200, headers: { "content-type": "application/octet-stream" } }, + ); + }); + const result = saveRemoteMedia({ + url: "https://example.com/download", + fetchImpl, + lookupFn: makeLookupFn(), + maxBytes: 8, + responseHeaderTimeoutMs: 10, + readIdleTimeoutMs: 30, + }); + + await vi.advanceTimersByTimeAsync(80); + + const saved = await result; + await expect(fs.readFile(saved.path)).resolves.toStrictEqual(Buffer.from([1, 2])); + } finally { + vi.useRealTimers(); + } + }); + + it("keeps the parent abort active while saving the response body", async () => { + const parent = new AbortController(); + let bodyStarted!: () => void; + const bodyReady = new Promise((resolve) => { + bodyStarted = resolve; + }); + const fetchImpl = vi.fn(async (_input: RequestInfo | URL, init?: RequestInit) => { + const signal = init?.signal; + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array([1])); + const failForAbort = () => controller.error(signal?.reason); + if (signal?.aborted) { + failForAbort(); + return; + } + signal?.addEventListener("abort", failForAbort, { once: true }); + bodyStarted(); + }, + }), + { status: 200 }, + ); + }); + const result = saveRemoteMedia({ + url: "https://example.com/download", + fetchImpl, + requestInit: { signal: parent.signal }, + lookupFn: makeLookupFn(), + maxBytes: 8, + responseHeaderTimeoutMs: 60_000, + }).catch((error: unknown) => error); + + await bodyReady; + parent.abort(); + + await expect(result).resolves.toMatchObject({ + name: "MediaFetchError", + code: "fetch_failed", + cause: { name: "AbortError" }, + }); + }); + it("clamps oversized saved-response idle timeout timers", async () => { const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); try { diff --git a/src/media/fetch.ts b/src/media/fetch.ts index d4659622d932..59bd47c94653 100644 --- a/src/media/fetch.ts +++ b/src/media/fetch.ts @@ -3,7 +3,7 @@ import { MAX_DOCUMENT_BYTES } from "@openclaw/media-core/constants"; import { parseMediaContentLength } from "@openclaw/media-core/content-length"; import { basenameFromAnyPath, extnameFromAnyPath } from "@openclaw/media-core/file-name"; import { detectMime, extensionForMime } from "@openclaw/media-core/mime"; -import { isAbortError } from "../infra/abort-signal.js"; +import { isAbortError, mergeAbortSignals } from "../infra/abort-signal.js"; import { formatErrorMessage } from "../infra/errors.js"; import { readChunkWithIdleTimeout, @@ -19,6 +19,7 @@ import type { LookupFn, PinnedDispatcherPolicy, SsrFPolicy } from "../infra/net/ import { retryAsync, type RetryOptions } from "../infra/retry.js"; import { isTransientNetworkError } from "../infra/unhandled-rejections.js"; import { redactSensitiveText } from "../logging/redact.js"; +import { buildTimeoutAbortSignal } from "../utils/fetch-timeout.js"; import { saveMediaBuffer, saveMediaStream, type SavedMedia } from "./store.js"; /** Default remote media fetch cap shared by buffer reads and store writes. */ @@ -75,8 +76,10 @@ type FetchMediaOptions = { filePathHint?: string; maxBytes?: number; maxRedirects?: number; - /** Abort the guarded fetch request if it has not completed by this deadline (ms). */ + /** Abort the complete guarded fetch and body operation after this deadline (ms). */ timeoutMs?: number; + /** Abort if final response headers have not arrived by this deadline (ms). */ + responseHeaderTimeoutMs?: number; /** Abort if the response body stops yielding data for this long (ms). */ readIdleTimeoutMs?: number; ssrfPolicy?: SsrFPolicy; @@ -189,6 +192,7 @@ async function fetchGuardedMediaResponse( requestInit, maxRedirects, timeoutMs, + responseHeaderTimeoutMs, ssrfPolicy, lookupFn, dispatcherPolicy, @@ -203,6 +207,15 @@ async function fetchGuardedMediaResponse( dispatcherAttempts && dispatcherAttempts.length > 0 ? dispatcherAttempts : [{ dispatcherPolicy, lookupFn }]; + const responseHeaderDeadline = buildTimeoutAbortSignal({ + timeoutMs: responseHeaderTimeoutMs, + operation: "media response headers", + url, + }); + const requestSignal = mergeAbortSignals([ + requestInit?.signal ?? undefined, + responseHeaderDeadline.signal, + ]); const runGuardedFetch = async (attempt: FetchDispatcherAttempt) => await fetchWithSsrFGuard( (trustExplicitProxyDns && attempt.dispatcherPolicy?.mode === "explicit-proxy" @@ -213,6 +226,7 @@ async function fetchGuardedMediaResponse( init: requestInit, maxRedirects, ...(timeoutMs !== undefined ? { timeoutMs } : {}), + ...(requestSignal.signal ? { signal: requestSignal.signal } : {}), policy: ssrfPolicy, lookupFn: attempt.lookupFn ?? lookupFn, dispatcherPolicy: attempt.dispatcherPolicy, @@ -253,13 +267,21 @@ async function fetchGuardedMediaResponse( attemptErrors.push(err); } } + // Clear only the header timer. The merged parent signal stays attached until + // release so shutdown can still interrupt a response body read. + responseHeaderDeadline.cleanup(); return { response: result.response, finalUrl: result.finalUrl, - release: result.release, + release: async () => { + requestSignal.dispose(); + await result.release(); + }, sourceUrl, }; } catch (err) { + responseHeaderDeadline.cleanup(); + requestSignal.dispose(); throw new MediaFetchError( "fetch_failed", `Failed to fetch media from ${sourceUrl}: ${formatErrorMessage(err)}`,