From ca11c4902da68751fc7f27e50ffeaa99e5f4eeb6 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 31 Jul 2026 05:21:42 -0700 Subject: [PATCH] fix(discord): retry safe webhook and reaction requests (#116821) Co-authored-by: Peter Steinberger --- extensions/discord/src/retry.test.ts | 26 ++++- extensions/discord/src/retry.ts | 40 ++++++- extensions/discord/src/send.reactions.ts | 34 ++++-- .../send.sends-basic-channel-messages.test.ts | 70 ++++++++++++ .../discord/src/send.webhook.proxy.test.ts | 106 +++++++++++++++++- .../discord/src/send.webhook.timeout.test.ts | 15 +++ extensions/discord/src/send.webhook.ts | 41 ++++--- 7 files changed, 296 insertions(+), 36 deletions(-) diff --git a/extensions/discord/src/retry.test.ts b/extensions/discord/src/retry.test.ts index 1663068a05b0..13a7afad76d4 100644 --- a/extensions/discord/src/retry.test.ts +++ b/extensions/discord/src/retry.test.ts @@ -1,10 +1,14 @@ // Discord tests cover retry plugin behavior. -import { describe, expect, it, vi } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { RateLimitError } from "./internal/discord.js"; import { createDiscordRetryRunner } from "./retry.js"; const ZERO_DELAY_RETRY = { attempts: 2, minDelayMs: 0, maxDelayMs: 0, jitter: 0 }; +afterEach(() => { + vi.useRealTimers(); +}); + function createRateLimitError(retryAfter = 0): RateLimitError { const response = new Response(null, { status: 429, @@ -138,6 +142,26 @@ describe("createDiscordRetryRunner create safety", () => { }); describe("createDiscordRetryRunner", () => { + it("cancels retry backoff immediately when the request deadline aborts", async () => { + vi.useFakeTimers(); + const controller = new AbortController(); + const timeout = Object.assign(new Error("request timed out"), { name: "TimeoutError" }); + const fn = vi.fn().mockRejectedValue(new TypeError("fetch failed")); + const runner = createDiscordRetryRunner({ + retry: { attempts: 2, minDelayMs: 60_000, maxDelayMs: 60_000, jitter: 0 }, + signal: controller.signal, + }); + const rejection = expect(runner(fn, "request")).rejects.toBe(timeout); + + await vi.advanceTimersByTimeAsync(0); + expect(vi.getTimerCount()).toBe(1); + controller.abort(timeout); + + await rejection; + expect(fn).toHaveBeenCalledOnce(); + expect(vi.getTimerCount()).toBe(0); + }); + it("retries transient transport errors", async () => { const fn = vi.fn().mockRejectedValueOnce(new TypeError("fetch failed")).mockResolvedValue("ok"); const runner = createDiscordRetryRunner({ retry: ZERO_DELAY_RETRY }); diff --git a/extensions/discord/src/retry.ts b/extensions/discord/src/retry.ts index 9b74eebabc7e..74a3352af652 100644 --- a/extensions/discord/src/retry.ts +++ b/extensions/discord/src/retry.ts @@ -10,8 +10,10 @@ import { classifyTransientNetworkErrorCode, createChannelApiRetryRunner, resolveRetryConfig, + retryAsync, type RetryConfig, } from "openclaw/plugin-sdk/retry-runtime"; +import { sleepWithAbort } from "openclaw/plugin-sdk/runtime-env"; import { RateLimitError } from "./internal/discord.js"; const DISCORD_RETRY_DEFAULTS = { @@ -111,6 +113,7 @@ export function createDiscordRetryRunner(params: { retry?: RetryConfig; verbose?: boolean; isGatewayDisconnected?: () => boolean; + signal?: AbortSignal; }): DiscordRetryRunner { const retryConfig = resolveRetryConfig(DISCORD_RETRY_DEFAULTS, params.retry); // Extend only the per-request runner. A delivery may contain several REST @@ -124,6 +127,11 @@ export function createDiscordRetryRunner(params: { const isRetryable = resolveDiscordRetryPredicate(options?.safety ?? "idempotent"); let observedGatewayDisconnect = false; const runRequest = async () => { + if (params.signal?.aborted) { + throw params.signal.reason instanceof Error + ? params.signal.reason + : new Error("Discord request aborted"); + } observedGatewayDisconnect ||= params.isGatewayDisconnected?.() === true; try { return await fn(); @@ -132,14 +140,36 @@ export function createDiscordRetryRunner(params: { throw err; } }; + const shouldRetry = (err: unknown, attempt: number) => + isRetryable(err) && + (attempt < retryConfig.attempts || + (observedGatewayDisconnect && isRetryableDiscordGatewayTransportError(err))); + const retryAfterMs = (err: unknown) => + err instanceof RateLimitError ? err.retryAfter * 1000 : undefined; + const signal = params.signal; + if (signal) { + return retryAsync(runRequest, { + ...retryConfig, + attempts, + label, + shouldRetry, + retryAfterMs, + sleep: async (delayMs) => { + try { + await sleepWithAbort(delayMs, signal); + } catch (error) { + // Preserve the owner's timeout error and clear the pending retry timer + // when a webhook deadline expires in the middle of Discord backoff. + throw signal.aborted && signal.reason instanceof Error ? signal.reason : error; + } + }, + }); + } const runWithRetry = createChannelApiRetryRunner({ retry: { ...retryConfig, attempts }, - shouldRetry: (err, attempt) => - isRetryable(err) && - (attempt < retryConfig.attempts || - (observedGatewayDisconnect && isRetryableDiscordGatewayTransportError(err))), + shouldRetry, strictShouldRetry: true, - retryAfterMs: (err) => (err instanceof RateLimitError ? err.retryAfter * 1000 : undefined), + retryAfterMs, verbose: params.verbose, }); return runWithRetry(runRequest, label); diff --git a/extensions/discord/src/send.reactions.ts b/extensions/discord/src/send.reactions.ts index 4c2465d8715a..96e13a42b773 100644 --- a/extensions/discord/src/send.reactions.ts +++ b/extensions/discord/src/send.reactions.ts @@ -58,11 +58,14 @@ export async function removeReactionDiscord( emoji: string, opts: DiscordReactOpts, ) { - const { rest } = isDiscordReactionRuntimeContext(opts) + const { rest, request } = isDiscordReactionRuntimeContext(opts) ? createDiscordReactionRuntimeClient(opts) : resolveDiscordReactionClient(opts); const encoded = normalizeReactionEmoji(emoji); - await deleteOwnMessageReaction(rest, channelId, messageId, encoded); + await request( + () => deleteOwnMessageReaction(rest, channelId, messageId, encoded), + "reaction-remove", + ); return { ok: true }; } @@ -71,10 +74,13 @@ export async function removeOwnReactionsDiscord( messageId: string, opts: DiscordReactOpts, ): Promise<{ ok: true; removed: string[] }> { - const { rest } = isDiscordReactionRuntimeContext(opts) + const { rest, request } = isDiscordReactionRuntimeContext(opts) ? createDiscordReactionRuntimeClient(opts) : resolveDiscordReactionClient(opts); - const message = (await getChannelMessage(rest, channelId, messageId)) as { + const message = (await request( + () => getChannelMessage(rest, channelId, messageId), + "reaction-list", + )) as { reactions?: Array<{ emoji: { id?: string | null; name?: string | null } }>; }; const identifiers = new Set(); @@ -92,7 +98,11 @@ export async function removeOwnReactionsDiscord( // failure and falsely report every identifier as removed. await Promise.all( removed.map((identifier) => - deleteOwnMessageReaction(rest, channelId, messageId, normalizeReactionEmoji(identifier)), + request( + () => + deleteOwnMessageReaction(rest, channelId, messageId, normalizeReactionEmoji(identifier)), + "reaction-remove", + ), ), ); return { ok: true, removed }; @@ -103,10 +113,13 @@ export async function fetchReactionsDiscord( messageId: string, opts: DiscordReactOpts & { limit?: number }, ): Promise { - const { rest } = isDiscordReactionRuntimeContext(opts) + const { rest, request } = isDiscordReactionRuntimeContext(opts) ? createDiscordReactionRuntimeClient(opts) : resolveDiscordReactionClient(opts); - const message = (await getChannelMessage(rest, channelId, messageId)) as { + const message = (await request( + () => getChannelMessage(rest, channelId, messageId), + "reaction-list", + )) as { reactions?: Array<{ count: number; emoji: { id?: string | null; name?: string | null }; @@ -128,9 +141,10 @@ export async function fetchReactionsDiscord( continue; } const encoded = encodeURIComponent(identifier); - const users = await listMessageReactionUsers(rest, channelId, messageId, encoded, { - limit, - }); + const users = await request( + () => listMessageReactionUsers(rest, channelId, messageId, encoded, { limit }), + "reaction-users", + ); summaries.push({ emoji: { id: reaction.emoji.id ?? null, diff --git a/extensions/discord/src/send.sends-basic-channel-messages.test.ts b/extensions/discord/src/send.sends-basic-channel-messages.test.ts index 7f9c62e1ccc8..8cb9a017bf19 100644 --- a/extensions/discord/src/send.sends-basic-channel-messages.test.ts +++ b/extensions/discord/src/send.sends-basic-channel-messages.test.ts @@ -1029,6 +1029,23 @@ describe("removeReactionDiscord", () => { Routes.channelMessageOwnReaction("chan1", "msg1", "%E2%9C%85"), ); }); + + it("retries transient failures while removing an idempotent reaction", async () => { + const { rest, deleteMock } = makeDiscordRest(); + deleteMock + .mockRejectedValueOnce(Object.assign(new Error("bad gateway"), { status: 502 })) + .mockResolvedValueOnce(undefined); + + await expect( + removeReactionDiscord("chan1", "msg1", "✅", { + rest, + token: "t", + cfg: DISCORD_TEST_CFG, + retry: { attempts: 2, minDelayMs: 0, maxDelayMs: 0, jitter: 0 }, + }), + ).resolves.toEqual({ ok: true }); + expect(deleteMock).toHaveBeenCalledTimes(2); + }); }); describe("removeOwnReactionsDiscord", () => { @@ -1058,6 +1075,27 @@ describe("removeOwnReactionsDiscord", () => { ); }); + it("retries transient failures while listing and clearing owned reactions", async () => { + const { rest, getMock, deleteMock } = makeDiscordRest(); + getMock + .mockRejectedValueOnce(Object.assign(new Error("service unavailable"), { status: 503 })) + .mockResolvedValueOnce({ reactions: [{ emoji: { name: "✅", id: null } }] }); + deleteMock + .mockRejectedValueOnce(Object.assign(new Error("bad gateway"), { status: 502 })) + .mockResolvedValueOnce(undefined); + + await expect( + removeOwnReactionsDiscord("chan1", "msg1", { + rest, + token: "t", + cfg: DISCORD_TEST_CFG, + retry: { attempts: 2, minDelayMs: 0, maxDelayMs: 0, jitter: 0 }, + }), + ).resolves.toEqual({ ok: true, removed: ["✅"] }); + expect(getMock).toHaveBeenCalledTimes(2); + expect(deleteMock).toHaveBeenCalledTimes(2); + }); + it("surfaces a failed deletion instead of reporting false success", async () => { const { rest, getMock, deleteMock } = makeDiscordRest(); getMock.mockResolvedValue({ @@ -1111,6 +1149,38 @@ describe("fetchReactionsDiscord", () => { }, ]); }); + + it.each([ + { operation: "message lookup", firstFailure: true, status: 503 }, + { operation: "reaction-user lookup", firstFailure: false, status: 502 }, + ])("retries a transient $operation failure", async ({ firstFailure, status }) => { + const { rest, getMock } = makeDiscordRest(); + const transientError = Object.assign(new Error("Discord temporarily unavailable"), { status }); + const message = { reactions: [{ count: 1, emoji: { name: "✅", id: null } }] }; + const users = [{ id: "u1", username: "alpha" }]; + if (firstFailure) { + getMock.mockRejectedValueOnce(transientError).mockResolvedValueOnce(message); + } else { + getMock.mockResolvedValueOnce(message).mockRejectedValueOnce(transientError); + } + getMock.mockResolvedValueOnce(users); + + await expect( + fetchReactionsDiscord("chan1", "msg1", { + rest, + token: "t", + cfg: DISCORD_TEST_CFG, + retry: { attempts: 2, minDelayMs: 0, maxDelayMs: 0, jitter: 0 }, + }), + ).resolves.toEqual([ + { + emoji: { id: null, name: "✅", raw: "✅" }, + count: 1, + users: [{ id: "u1", username: "alpha", tag: "alpha" }], + }, + ]); + expect(getMock).toHaveBeenCalledTimes(3); + }); }); describe("fetchChannelPermissionsDiscord", () => { diff --git a/extensions/discord/src/send.webhook.proxy.test.ts b/extensions/discord/src/send.webhook.proxy.test.ts index e954e95f3f7d..495d172dbc77 100644 --- a/extensions/discord/src/send.webhook.proxy.test.ts +++ b/extensions/discord/src/send.webhook.proxy.test.ts @@ -46,6 +46,7 @@ describe("sendWebhookMessageDiscord proxy support", () => { afterEach(() => { vi.unstubAllEnvs(); + vi.useRealTimers(); }); it("falls back to global fetch when the Discord proxy URL is invalid", async () => { @@ -267,13 +268,108 @@ describe("sendWebhookMessageDiscord proxy support", () => { globalFetchMock.mockRestore(); }); - it("throws typed rate limit errors for webhook 429 responses", async () => { - const globalFetchMock = vi.spyOn(globalThis, "fetch").mockResolvedValue( - new Response(JSON.stringify({ message: "Slow down", retry_after: 0.25, global: false }), { - status: 429, - }), + it("retries rate-limited webhook sends after the Discord retry-after delay", async () => { + vi.useFakeTimers(); + const globalFetchMock = vi + .spyOn(globalThis, "fetch") + .mockResolvedValueOnce( + new Response(JSON.stringify({ message: "Slow down", retry_after: 0.75, global: false }), { + status: 429, + }), + ) + .mockResolvedValueOnce( + new Response(JSON.stringify({ id: "msg-retried", channel_id: "thread-1" }), { + status: 200, + headers: { "content-type": "application/json" }, + }), + ); + + const sent = sendWebhookMessageDiscord("hello", { + cfg: { channels: { discord: { token: "Bot test-token" } } } as OpenClawConfig, + accountId: "default", + webhookId: "123", + webhookToken: "abc", + threadId: "thread-1", + wait: true, + }); + const outcome = sent.then( + (value) => ({ ok: true as const, value }), + (error: unknown) => ({ ok: false as const, error }), ); + await vi.advanceTimersByTimeAsync(749); + expect(globalFetchMock).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(1_000); + + await expect(outcome).resolves.toMatchObject({ + ok: true, + value: { messageId: "msg-retried", channelId: "thread-1" }, + }); + expect(globalFetchMock).toHaveBeenCalledTimes(2); + }); + + it("retries proven pre-connect webhook failures", async () => { + vi.useFakeTimers(); + const globalFetchMock = vi + .spyOn(globalThis, "fetch") + .mockRejectedValueOnce(Object.assign(new Error("connect refused"), { code: "ECONNREFUSED" })) + .mockResolvedValueOnce( + new Response(JSON.stringify({ id: "msg-connected", channel_id: "thread-1" }), { + status: 200, + headers: { "content-type": "application/json" }, + }), + ); + + const sent = sendWebhookMessageDiscord("hello", { + cfg: { channels: { discord: { token: "Bot test-token" } } } as OpenClawConfig, + accountId: "default", + webhookId: "123", + webhookToken: "abc", + wait: true, + }); + const outcome = sent.then( + (value) => ({ ok: true as const, value }), + (error: unknown) => ({ ok: false as const, error }), + ); + await vi.advanceTimersByTimeAsync(1_000); + + await expect(outcome).resolves.toMatchObject({ + ok: true, + value: { messageId: "msg-connected" }, + }); + expect(globalFetchMock).toHaveBeenCalledTimes(2); + }); + + it("never retries ambiguous webhook server failures that could duplicate a message", async () => { + const globalFetchMock = vi + .spyOn(globalThis, "fetch") + .mockResolvedValueOnce(new Response("bad gateway", { status: 502 })) + .mockResolvedValueOnce( + new Response(JSON.stringify({ id: "unexpected-duplicate" }), { status: 200 }), + ); + + await expect( + sendWebhookMessageDiscord("hello", { + cfg: { channels: { discord: { token: "Bot test-token" } } } as OpenClawConfig, + accountId: "default", + webhookId: "123", + webhookToken: "abc", + wait: true, + }), + ).rejects.toMatchObject({ status: 502 }); + expect(globalFetchMock).toHaveBeenCalledTimes(1); + }); + + it("throws typed rate limit errors for webhook 429 responses", async () => { + const globalFetchMock = vi.spyOn(globalThis, "fetch").mockImplementation(async () => { + return new Response( + JSON.stringify({ message: "Slow down", retry_after: 0.25, global: false }), + { + status: 429, + }, + ); + }); + const cfg = { channels: { discord: { diff --git a/extensions/discord/src/send.webhook.timeout.test.ts b/extensions/discord/src/send.webhook.timeout.test.ts index 8043a1c1ba22..238f32d453d0 100644 --- a/extensions/discord/src/send.webhook.timeout.test.ts +++ b/extensions/discord/src/send.webhook.timeout.test.ts @@ -84,6 +84,21 @@ describe("sendWebhookMessageDiscord timeout", () => { await expectWebhookTimeout(sendWebhookMessageDiscord("hello", opts)); }); + it("aborts rate-limit backoff at the request deadline without leaving a retry timer", async () => { + vi.useFakeTimers(); + const fetchMock = vi.fn(async () => { + return new Response(JSON.stringify({ message: "Slow down", retry_after: 60 }), { + status: 429, + }); + }); + vi.stubGlobal("fetch", fetchMock); + + await expectWebhookTimeout(sendWebhookMessageDiscord("hello", opts)); + + expect(fetchMock).toHaveBeenCalledOnce(); + expect(vi.getTimerCount()).toBe(0); + }); + it("cleans up the deadline after a normal response", async () => { vi.useFakeTimers(); vi.stubGlobal( diff --git a/extensions/discord/src/send.webhook.ts b/extensions/discord/src/send.webhook.ts index 8d0e3e1aed53..204866dfe9a8 100644 --- a/extensions/discord/src/send.webhook.ts +++ b/extensions/discord/src/send.webhook.ts @@ -19,6 +19,7 @@ import { } from "./internal/rest-errors.js"; import { rewriteDiscordKnownMentions } from "./mentions.js"; import { DISCORD_REST_TIMEOUT_MS } from "./proxy-request-client.js"; +import { createDiscordRetryRunner } from "./retry.js"; import { createDiscordSendResult } from "./send.receipt.js"; import type { DiscordSendResult } from "./send.types.js"; @@ -142,23 +143,33 @@ export async function sendWebhookMessageDiscord( timeoutMs: DISCORD_WEBHOOK_TIMEOUT_MS, operation: "discord.webhook.send", }); + const request = createDiscordRetryRunner({ signal: deadline.signal }); try { - const response = await (proxyFetch ?? fetch)(url, { - method: "POST", - headers: { - "content-type": "application/json", + const response = await request( + async () => { + const attemptResponse = await (proxyFetch ?? fetch)(url, { + method: "POST", + headers: { + "content-type": "application/json", + }, + body: JSON.stringify({ + content: rewrittenText, + username: normalizeOptionalString(opts.username), + avatar_url: normalizeOptionalString(opts.avatarUrl), + ...(messageReference ? { message_reference: messageReference } : {}), + }), + signal: deadline.signal, + }); + if (!attemptResponse.ok) { + await throwWebhookResponseError(attemptResponse, deadline.signal); + } + return attemptResponse; }, - body: JSON.stringify({ - content: rewrittenText, - username: normalizeOptionalString(opts.username), - avatar_url: normalizeOptionalString(opts.avatarUrl), - ...(messageReference ? { message_reference: messageReference } : {}), - }), - signal: deadline.signal, - }); - if (!response.ok) { - await throwWebhookResponseError(response, deadline.signal); - } + "webhook", + // Webhooks cannot enforce a Discord nonce, so replay only explicit 429s + // and proven pre-connect failures; an ambiguous 5xx could duplicate delivery. + { safety: "non-idempotent-create" }, + ); const payload: { id?: string;