fix(discord): retry safe webhook and reaction requests (#116821)

Co-authored-by: Peter Steinberger <steipete@macos.shared>

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 });

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);

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<string>();
@@ -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<DiscordReactionSummary[]> {
-  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,

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", () => {

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: {

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<typeof fetch>(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(

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;