From a0f8f451cd2fec2c03bae2c08a9ffcaaaf9eb7de Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Thu, 16 Jul 2026 01:44:24 -0700 Subject: [PATCH] refactor(discord): reuse shared probe body deadline (#108768) --- extensions/discord/src/probe.intents.test.ts | 73 ++++++++++++-------- extensions/discord/src/probe.ts | 62 ++++++----------- 2 files changed, 65 insertions(+), 70 deletions(-) diff --git a/extensions/discord/src/probe.intents.test.ts b/extensions/discord/src/probe.intents.test.ts index 241c9999e2e9..ce12ab9b8c91 100644 --- a/extensions/discord/src/probe.intents.test.ts +++ b/extensions/discord/src/probe.intents.test.ts @@ -31,19 +31,27 @@ function oversizedDiscordProbeJsonResponse(onCancel: () => void): Response { return response; } -function abortableTricklingDiscordProbeJsonResponse( +function trackedTricklingDiscordProbeJsonResponse( signal: AbortSignal | null | undefined, - onAbort: () => void, + onTerminate: () => void, ): Response { let interval: ReturnType | undefined; + let terminated = false; + const terminate = () => { + if (terminated) { + return; + } + terminated = true; + if (interval) { + clearInterval(interval); + } + onTerminate(); + }; return new Response( new ReadableStream({ start(controller) { const abort = () => { - if (interval) { - clearInterval(interval); - } - onAbort(); + terminate(); controller.error(signal?.reason ?? new Error("request aborted")); }; if (signal?.aborted) { @@ -55,24 +63,30 @@ function abortableTricklingDiscordProbeJsonResponse( signal?.addEventListener("abort", abort, { once: true }); }, cancel() { - if (interval) { - clearInterval(interval); - } + terminate(); }, }), { headers: { "content-type": "application/json" }, status: 200 }, ); } -function abortableStalledDiscordJsonResponse( +function trackedStalledDiscordJsonResponse( signal: AbortSignal | null | undefined, - onAbort: () => void, + onTerminate: () => void, ): Response { + let terminated = false; + const terminate = () => { + if (terminated) { + return; + } + terminated = true; + onTerminate(); + }; return new Response( new ReadableStream({ start(controller) { const abort = () => { - onAbort(); + terminate(); controller.error(signal?.reason ?? new Error("request aborted")); }; if (signal?.aborted) { @@ -81,6 +95,9 @@ function abortableStalledDiscordJsonResponse( } signal?.addEventListener("abort", abort, { once: true }); }, + cancel() { + terminate(); + }, }), { headers: { "content-type": "application/json" }, status: 200 }, ); @@ -199,10 +216,10 @@ describe("resolveDiscordPrivilegedIntentsFromFlags", () => { it("times out and cancels stalled getMe probe JSON response bodies", async () => { vi.useFakeTimers(); try { - let abortCount = 0; + let terminationCount = 0; const fetcher = withFetchPreconnect(async (_input, init) => - abortableStalledDiscordJsonResponse(init?.signal, () => { - abortCount += 1; + trackedStalledDiscordJsonResponse(init?.signal, () => { + terminationCount += 1; }), ); @@ -215,7 +232,7 @@ describe("resolveDiscordPrivilegedIntentsFromFlags", () => { await vi.advanceTimersByTimeAsync(0); await vi.advanceTimersByTimeAsync(50); await assertion; - expect(abortCount).toBe(1); + expect(terminationCount).toBe(1); } finally { vi.useRealTimers(); } @@ -224,13 +241,13 @@ describe("resolveDiscordPrivilegedIntentsFromFlags", () => { it("uses one total deadline across getMe headers and a trickling JSON body", async () => { vi.useFakeTimers(); try { - let abortCount = 0; + let terminationCount = 0; const fetcher = withFetchPreconnect(async (_input, init) => { await new Promise((resolve) => { setTimeout(resolve, 30); }); - return abortableTricklingDiscordProbeJsonResponse(init?.signal, () => { - abortCount += 1; + return trackedTricklingDiscordProbeJsonResponse(init?.signal, () => { + terminationCount += 1; }); }); @@ -242,7 +259,7 @@ describe("resolveDiscordPrivilegedIntentsFromFlags", () => { await vi.advanceTimersByTimeAsync(50); await assertion; - expect(abortCount).toBe(1); + expect(terminationCount).toBe(1); } finally { vi.useRealTimers(); } @@ -251,15 +268,15 @@ describe("resolveDiscordPrivilegedIntentsFromFlags", () => { it("bounds stalled application-summary response bodies during probes", async () => { vi.useFakeTimers(); try { - let abortCount = 0; + let terminationCount = 0; const fetcher = withFetchPreconnect(async (input, init) => { const url = typeof input === "string" ? input : input instanceof URL ? input.toString() : input.url; if (url.endsWith("/users/@me")) { return jsonResponse({ id: "bot-1", username: "openclaw" }); } - return abortableStalledDiscordJsonResponse(init?.signal, () => { - abortCount += 1; + return trackedStalledDiscordJsonResponse(init?.signal, () => { + terminationCount += 1; }); }); @@ -280,7 +297,7 @@ describe("resolveDiscordPrivilegedIntentsFromFlags", () => { ok: true, bot: { id: "bot-1", username: "openclaw" }, }); - expect(abortCount).toBe(1); + expect(terminationCount).toBe(1); } finally { vi.useRealTimers(); } @@ -289,10 +306,10 @@ describe("resolveDiscordPrivilegedIntentsFromFlags", () => { it("bounds stalled application-id response bodies", async () => { vi.useFakeTimers(); try { - let abortCount = 0; + let terminationCount = 0; const fetcher = withFetchPreconnect(async (_input, init) => - abortableStalledDiscordJsonResponse(init?.signal, () => { - abortCount += 1; + trackedStalledDiscordJsonResponse(init?.signal, () => { + terminationCount += 1; }), ); @@ -301,7 +318,7 @@ describe("resolveDiscordPrivilegedIntentsFromFlags", () => { await vi.advanceTimersByTimeAsync(50); await expect(lookup).resolves.toBeUndefined(); - expect(abortCount).toBe(1); + expect(terminationCount).toBe(1); } finally { vi.useRealTimers(); } diff --git a/extensions/discord/src/probe.ts b/extensions/discord/src/probe.ts index b942def5cdd5..b43019651f1f 100644 --- a/extensions/discord/src/probe.ts +++ b/extensions/discord/src/probe.ts @@ -1,7 +1,6 @@ // Discord plugin module implements probe behavior. import type { BaseProbeResult } from "openclaw/plugin-sdk/channel-contract"; import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; -import { buildTimeoutAbortSignal } from "openclaw/plugin-sdk/extension-shared"; import { resolveFetch } from "openclaw/plugin-sdk/fetch-runtime"; import { readResponseWithLimit } from "openclaw/plugin-sdk/response-limit-runtime"; import { fetchWithTimeout } from "openclaw/plugin-sdk/text-utility-runtime"; @@ -117,11 +116,15 @@ function getResolvedFetch(fetcher: typeof fetch): typeof fetch { async function readDiscordProbeGetMeJson( response: Response, timeoutMs: number, + deadlineMs: number, ): Promise<{ id?: string; username?: string }> { const bytes = await readResponseWithLimit(response, DISCORD_PROBE_JSON_MAX_BYTES, { chunkTimeoutMs: timeoutMs, onIdleTimeout: ({ chunkTimeoutMs }) => new Error(`${DISCORD_PROBE_GET_ME_LABEL}: JSON response stalled after ${chunkTimeoutMs}ms`), + timeoutMs: Math.max(1, deadlineMs - Date.now()), + onTimeout: () => + new Error(`${DISCORD_PROBE_GET_ME_LABEL}: JSON response timed out after ${timeoutMs}ms`), onOverflow: ({ maxBytes }) => new Error(`${DISCORD_PROBE_GET_ME_LABEL}: JSON response exceeds ${maxBytes} bytes`), }); @@ -157,49 +160,24 @@ export async function probeDiscord( let res: Response | undefined; try { const getMeUrl = `${DISCORD_API_BASE}/users/@me`; - const getMeTimeout = buildTimeoutAbortSignal({ + const getMeDeadlineMs = Date.now() + timeoutMs; + res = await fetchWithTimeout( + getMeUrl, + { headers: { Authorization: `Bot ${normalized}` } }, timeoutMs, - operation: DISCORD_PROBE_GET_ME_LABEL, - url: getMeUrl, - }); - try { - res = await fetchWithTimeout( - getMeUrl, - { - headers: { Authorization: `Bot ${normalized}` }, - signal: getMeTimeout.signal, - }, - timeoutMs, - getResolvedFetch(fetcher), - ); - if (!res.ok) { - result.status = res.status; - result.error = `getMe failed (${res.status})`; - return { ...result, elapsedMs: Date.now() - started }; - } - let json: { id?: string; username?: string }; - try { - json = await readDiscordProbeGetMeJson(res, timeoutMs); - } catch (error) { - if (getMeTimeout.signal?.aborted) { - const message = `${DISCORD_PROBE_GET_ME_LABEL}: JSON response timed out after ${timeoutMs}ms`; - if (error instanceof Error) { - error.message = message; - throw error; - } - throw new Error(message, { cause: error }); - } - throw error; - } - result.ok = true; - result.bot = { - id: json.id ?? null, - username: json.username ?? null, - }; - } finally { - // The timeout must outlive header receipt so the same caller budget covers body reads. - getMeTimeout.cleanup(); + getResolvedFetch(fetcher), + ); + if (!res.ok) { + result.status = res.status; + result.error = `getMe failed (${res.status})`; + return { ...result, elapsedMs: Date.now() - started }; } + const json = await readDiscordProbeGetMeJson(res, timeoutMs, getMeDeadlineMs); + result.ok = true; + result.bot = { + id: json.id ?? null, + username: json.username ?? null, + }; if (includeApplication) { // Application metadata is optional. Keep its deadline inside the outer status budget so a // stalled secondary response cannot discard the already-resolved bot identity.