From 0934c445d3c19e4a9ea243e7138da0c81e55737a Mon Sep 17 00:00:00 2001 From: Marcus Castro <7562095+mcaxtr@users.noreply.github.com> Date: Mon, 13 Jul 2026 21:19:32 -0300 Subject: [PATCH] refactor(whatsapp): use retryAsync for outbound delivery retries (#105579) * refactor(whatsapp): reuse shared outbound retry runtime * refactor(whatsapp): narrow outbound retry classification * refactor(whatsapp): remove obsolete sleep export --- .../src/auto-reply/deliver-reply.test.ts | 82 ++++++------- .../whatsapp/src/auto-reply/deliver-reply.ts | 5 +- .../whatsapp/src/outbound-media-contract.ts | 47 -------- .../whatsapp/src/outbound-retry.test.ts | 110 ++++++++++++++++++ extensions/whatsapp/src/outbound-retry.ts | 77 ++++++++++++ extensions/whatsapp/src/text-runtime.ts | 2 +- 6 files changed, 232 insertions(+), 91 deletions(-) create mode 100644 extensions/whatsapp/src/outbound-retry.test.ts create mode 100644 extensions/whatsapp/src/outbound-retry.ts diff --git a/extensions/whatsapp/src/auto-reply/deliver-reply.test.ts b/extensions/whatsapp/src/auto-reply/deliver-reply.test.ts index 8d800b820e96..97fd17de7cd2 100644 --- a/extensions/whatsapp/src/auto-reply/deliver-reply.test.ts +++ b/extensions/whatsapp/src/auto-reply/deliver-reply.test.ts @@ -5,7 +5,6 @@ import { listMessageReceiptPlatformIds, } from "openclaw/plugin-sdk/channel-outbound"; import { logVerbose } from "openclaw/plugin-sdk/runtime-env"; -import { sleep } from "openclaw/plugin-sdk/text-utility-runtime"; import { beforeAll, describe, expect, it, vi } from "vitest"; import { createAcceptedWhatsAppSendResult } from "../inbound/send-result.test-helper.js"; import { createTestWebInboundMessage } from "../inbound/test-message.test-helper.js"; @@ -39,16 +38,6 @@ vi.mock("openclaw/plugin-sdk/runtime-env", async () => { }; }); -vi.mock("openclaw/plugin-sdk/text-utility-runtime", async () => { - const actual = await vi.importActual( - "openclaw/plugin-sdk/text-utility-runtime", - ); - return { - ...actual, - sleep: vi.fn(async () => {}), - }; -}); - vi.mock("../media.js", () => ({ loadWebMedia: vi.fn(), })); @@ -191,6 +180,18 @@ function mockSecondReplySuccess(msg: AdmittedWebInboundMessage) { ).mockResolvedValueOnce(createAcceptedWhatsAppSendResult("text", "reply-retry-2")); } +async function runWithFakeTimers(run: () => Promise): Promise { + vi.useFakeTimers(); + try { + const promise = run(); + await vi.runAllTimersAsync(); + return await promise; + } finally { + vi.clearAllTimers(); + vi.useRealTimers(); + } +} + const replyLogger = { info: vi.fn(), warn: vi.fn(), @@ -406,17 +407,18 @@ describe("deliverWebReply", () => { mockFirstReplyFailure(msg, errorMessage); mockSecondReplySuccess(msg); - await deliverWebReply({ - replyResult: { text: "hi" }, - msg, - maxMediaBytes: 1024 * 1024, - textLimit: 200, - replyLogger, - skipLog: true, - }); + await runWithFakeTimers(() => + deliverWebReply({ + replyResult: { text: "hi" }, + msg, + maxMediaBytes: 1024 * 1024, + textLimit: 200, + replyLogger, + skipLog: true, + }), + ); expect(msg.platform.reply).toHaveBeenCalledTimes(2); - expect(sleep).toHaveBeenCalledWith(500); }, ); @@ -425,23 +427,23 @@ describe("deliverWebReply", () => { mockFirstReplyFailureWithWrappedError(msg, "connection closed"); mockSecondReplySuccess(msg); - await deliverWebReply({ - replyResult: { text: "hi" }, - msg, - maxMediaBytes: 1024 * 1024, - textLimit: 200, - replyLogger, - skipLog: true, - }); + await runWithFakeTimers(() => + deliverWebReply({ + replyResult: { text: "hi" }, + msg, + maxMediaBytes: 1024 * 1024, + textLimit: 200, + replyLogger, + skipLog: true, + }), + ); expect(msg.platform.reply).toHaveBeenCalledTimes(2); - expect(sleep).toHaveBeenCalledWith(500); }); it("does not retry terminal socket operation timeouts", async () => { const msg = makeMsg(); const timeout = new WhatsAppSocketOperationTimeoutError("sendMessage", 60_000); - (sleep as unknown as { mockClear: () => void }).mockClear(); ( msg.platform.reply as unknown as { mockRejectedValueOnce: (v: unknown) => void } ).mockRejectedValueOnce(timeout); @@ -458,7 +460,6 @@ describe("deliverWebReply", () => { ).rejects.toBe(timeout); expect(msg.platform.reply).toHaveBeenCalledTimes(1); - expect(sleep).not.toHaveBeenCalled(); }); it("sends image media with caption and then remaining text", async () => { @@ -586,17 +587,18 @@ describe("deliverWebReply", () => { msg.platform.sendMedia as unknown as { mockResolvedValueOnce: (v: unknown) => void } ).mockResolvedValueOnce(createAcceptedWhatsAppSendResult("media", "media-retry-2")); - await deliverWebReply({ - replyResult: { text: "caption", mediaUrl: "http://example.com/img.jpg" }, - msg, - maxMediaBytes: 1024 * 1024, - textLimit: 200, - replyLogger, - skipLog: true, - }); + await runWithFakeTimers(() => + deliverWebReply({ + replyResult: { text: "caption", mediaUrl: "http://example.com/img.jpg" }, + msg, + maxMediaBytes: 1024 * 1024, + textLimit: 200, + replyLogger, + skipLog: true, + }), + ); expect(msg.platform.sendMedia).toHaveBeenCalledTimes(2); - expect(sleep).toHaveBeenCalledWith(500); }); it("falls back to text-only when the first media send fails", async () => { diff --git a/extensions/whatsapp/src/auto-reply/deliver-reply.ts b/extensions/whatsapp/src/auto-reply/deliver-reply.ts index f9f53ff3833b..804eacd4fc8d 100644 --- a/extensions/whatsapp/src/auto-reply/deliver-reply.ts +++ b/extensions/whatsapp/src/auto-reply/deliver-reply.ts @@ -22,8 +22,8 @@ import { normalizeWhatsAppOutboundPayload, normalizeWhatsAppPayloadTextPreservingIndentation, prepareWhatsAppOutboundMedia, - sendWhatsAppOutboundWithRetry, } from "../outbound-media-contract.js"; +import { sendWhatsAppOutboundWithRetry } from "../outbound-retry.js"; import { buildQuotedMessageOptions, lookupInboundMessageMeta } from "../quoted-message.js"; import { newConnectionId } from "../reconnect.js"; import { formatError } from "../session.js"; @@ -159,11 +159,10 @@ export async function deliverWebReply(params: { }); }; - const sendWithRetry = async (fn: () => Promise, label: string, maxAttempts = 3) => { + const sendWithRetry = async (fn: () => Promise, label: string) => { try { return await sendWhatsAppOutboundWithRetry({ send: fn, - maxAttempts, onRetry: ({ attempt, maxAttempts: retryMaxAttempts, backoffMs, errorText }) => { logVerbose( `Retrying ${label} to ${conversationId} after failure (${attempt}/${retryMaxAttempts - 1}) in ${backoffMs}ms: ${errorText}`, diff --git a/extensions/whatsapp/src/outbound-media-contract.ts b/extensions/whatsapp/src/outbound-media-contract.ts index bba666631ca5..7bfe146b8373 100644 --- a/extensions/whatsapp/src/outbound-media-contract.ts +++ b/extensions/whatsapp/src/outbound-media-contract.ts @@ -7,13 +7,10 @@ import { writeExternalFileWithinRoot } from "openclaw/plugin-sdk/security-runtim import { normalizeUniqueStringEntries } from "openclaw/plugin-sdk/string-coerce-runtime"; import { resolvePreferredOpenClawTmpDir, withTempWorkspace } from "openclaw/plugin-sdk/temp-path"; import { resolveWhatsAppDocumentFileName } from "./document-filename.js"; -import { formatError } from "./session-errors.js"; -import { isWhatsAppSocketOperationTimeoutError } from "./socket-timing.js"; import { sanitizeAssistantVisibleText, sanitizeAssistantVisibleTextWithProfile, stripToolCallXmlTags, - sleep, } from "./text-runtime.js"; type WhatsAppOutboundPayloadLike = { @@ -269,47 +266,3 @@ function deriveWhatsAppDocumentFileName(mediaUrl: string | undefined): string | return fileName || undefined; } } - -function isRetryableWhatsAppOutboundError(error: unknown): boolean { - if (isWhatsAppSocketOperationTimeoutError(error)) { - return false; - } - return /closed|reset|timed\s*out|disconnect/i.test(formatError(error)); -} - -export async function sendWhatsAppOutboundWithRetry(params: { - send: () => Promise; - onRetry?: (params: { - attempt: number; - maxAttempts: number; - backoffMs: number; - error: unknown; - errorText: string; - }) => Promise | void; - maxAttempts?: number; -}): Promise { - const maxAttempts = params.maxAttempts ?? 3; - let lastError: unknown; - for (let attempt = 1; attempt <= maxAttempts; attempt += 1) { - try { - return await params.send(); - } catch (error) { - lastError = error; - const errorText = formatError(error); - const isLastAttempt = attempt === maxAttempts; - if (!isRetryableWhatsAppOutboundError(error) || isLastAttempt) { - throw error; - } - const backoffMs = 500 * attempt; - await params.onRetry?.({ - attempt, - maxAttempts, - backoffMs, - error, - errorText, - }); - await sleep(backoffMs); - } - } - throw lastError; -} diff --git a/extensions/whatsapp/src/outbound-retry.test.ts b/extensions/whatsapp/src/outbound-retry.test.ts new file mode 100644 index 000000000000..40a8870477df --- /dev/null +++ b/extensions/whatsapp/src/outbound-retry.test.ts @@ -0,0 +1,110 @@ +// WhatsApp tests cover outbound retry behavior. +import { describe, expect, it, vi } from "vitest"; +import { sendWhatsAppOutboundWithRetry } from "./outbound-retry.js"; +import { WhatsAppSocketOperationTimeoutError } from "./socket-timing.js"; + +async function runWithFakeTimers(run: () => Promise): Promise { + vi.useFakeTimers(); + try { + const promise = run(); + await vi.runAllTimersAsync(); + return await promise; + } finally { + vi.clearAllTimers(); + vi.useRealTimers(); + } +} + +describe("sendWhatsAppOutboundWithRetry", () => { + it.each([new Error("connection closed"), { code: "ECONNRESET" }])( + "retries a directly retryable error", + async (error) => { + const send = vi + .fn<() => Promise>() + .mockRejectedValueOnce(error) + .mockResolvedValue("ok"); + + await expect(runWithFakeTimers(() => sendWhatsAppOutboundWithRetry({ send }))).resolves.toBe( + "ok", + ); + + expect(send).toHaveBeenCalledTimes(2); + }, + ); + + it.each([ + { name: "a non-retryable direct error", error: new Error("invalid recipient") }, + { + name: "a retryable signal only in the cause", + error: new Error("request failed", { cause: new Error("socket disconnected") }), + }, + ])("does not retry $name", async ({ error }) => { + const send = vi.fn<() => Promise>().mockRejectedValue(error); + const onRetry = vi.fn(); + + const failure = await sendWhatsAppOutboundWithRetry({ send, onRetry }).catch( + (caught: unknown) => caught, + ); + + expect(failure).toBe(error); + expect(send).toHaveBeenCalledOnce(); + expect(onRetry).not.toHaveBeenCalled(); + }); + + it("does not retry a direct unknown-delivery socket timeout", async () => { + const timeout = new WhatsAppSocketOperationTimeoutError("sendMessage", 60_000); + const send = vi.fn<() => Promise>().mockRejectedValue(timeout); + const onRetry = vi.fn(); + + const failure = await sendWhatsAppOutboundWithRetry({ send, onRetry }).catch( + (caught: unknown) => caught, + ); + + expect(failure).toBe(timeout); + expect(send).toHaveBeenCalledOnce(); + expect(onRetry).not.toHaveBeenCalled(); + }); + + it("preserves attempts, delays, callback fields, and terminal error identity", async () => { + const firstError = { + output: { + statusCode: 503, + payload: { + statusCode: 503, + error: "Service Unavailable", + message: "connection closed", + }, + }, + }; + const secondError = new Error("socket reset"); + const terminalError = { code: "ECONNRESET", marker: "terminal" }; + const send = vi + .fn<() => Promise>() + .mockRejectedValueOnce(firstError) + .mockRejectedValueOnce(secondError) + .mockRejectedValueOnce(terminalError); + const onRetry = vi.fn(); + + const failure = await runWithFakeTimers(() => + sendWhatsAppOutboundWithRetry({ send, onRetry }).catch((caught: unknown) => caught), + ); + + expect(failure).toBe(terminalError); + expect(send).toHaveBeenCalledTimes(3); + expect(onRetry).toHaveBeenCalledTimes(2); + expect(onRetry).toHaveBeenNthCalledWith(1, { + attempt: 1, + maxAttempts: 3, + backoffMs: 500, + error: firstError, + errorText: "status=503 Service Unavailable connection closed", + }); + expect(onRetry).toHaveBeenNthCalledWith(2, { + attempt: 2, + maxAttempts: 3, + backoffMs: 1_000, + error: secondError, + errorText: "socket reset", + }); + }); +}); diff --git a/extensions/whatsapp/src/outbound-retry.ts b/extensions/whatsapp/src/outbound-retry.ts new file mode 100644 index 000000000000..3f33bf5bf326 --- /dev/null +++ b/extensions/whatsapp/src/outbound-retry.ts @@ -0,0 +1,77 @@ +// WhatsApp plugin module implements outbound retry behavior. +import { retryAsync } from "openclaw/plugin-sdk/retry-runtime"; +import { formatError } from "./session-errors.js"; +import { isWhatsAppSocketOperationTimeoutError } from "./socket-timing.js"; + +const WHATSAPP_OUTBOUND_MAX_ATTEMPTS = 3; +const WHATSAPP_OUTBOUND_MIN_DELAY_MS = 500; +const WHATSAPP_OUTBOUND_MAX_DELAY_MS = 1_000; +const WHATSAPP_RETRYABLE_OUTBOUND_ERROR_PATTERN = /closed|reset|timed\s*out|disconnect/i; + +class WhatsAppOutboundRetryError extends Error { + constructor(readonly original: unknown) { + super(formatError(original), { cause: original }); + } +} + +function isRetryableWhatsAppOutboundError(error: unknown): boolean { + // Outbound sends surface direct failures; inspecting wrappers or causes can + // replay a non-idempotent send. A direct local timeout may have delivered it. + if (isWhatsAppSocketOperationTimeoutError(error)) { + return false; + } + return WHATSAPP_RETRYABLE_OUTBOUND_ERROR_PATTERN.test(formatError(error)); +} + +type WhatsAppOutboundRetryInfo = { + attempt: number; + maxAttempts: number; + backoffMs: number; + error: unknown; + errorText: string; +}; + +export async function sendWhatsAppOutboundWithRetry(params: { + send: () => Promise; + onRetry?: (info: WhatsAppOutboundRetryInfo) => void; +}): Promise { + try { + return await retryAsync( + async () => { + try { + return await params.send(); + } catch (error) { + // retryAsync normalizes non-Error throws. Keep the original value in + // an Error wrapper so the WhatsApp adapter can restore exact identity. + throw new WhatsAppOutboundRetryError(error); + } + }, + { + attempts: WHATSAPP_OUTBOUND_MAX_ATTEMPTS, + minDelayMs: WHATSAPP_OUTBOUND_MIN_DELAY_MS, + maxDelayMs: WHATSAPP_OUTBOUND_MAX_DELAY_MS, + jitter: 0, + shouldRetry: (error) => + error instanceof WhatsAppOutboundRetryError && + isRetryableWhatsAppOutboundError(error.original), + onRetry: ({ attempt, maxAttempts, delayMs, err }) => { + if (!(err instanceof WhatsAppOutboundRetryError)) { + return; + } + params.onRetry?.({ + attempt, + maxAttempts, + backoffMs: delayMs, + error: err.original, + errorText: formatError(err.original), + }); + }, + }, + ); + } catch (error) { + if (error instanceof WhatsAppOutboundRetryError) { + throw error.original; + } + throw error; + } +} diff --git a/extensions/whatsapp/src/text-runtime.ts b/extensions/whatsapp/src/text-runtime.ts index f03a3d245f41..fa47fcf8198d 100644 --- a/extensions/whatsapp/src/text-runtime.ts +++ b/extensions/whatsapp/src/text-runtime.ts @@ -5,7 +5,7 @@ export { sanitizeAssistantVisibleTextWithProfile, stripToolCallXmlTags, } from "openclaw/plugin-sdk/text-chunking"; -export { normalizeE164, resolveUserPath, sleep } from "openclaw/plugin-sdk/text-utility-runtime"; +export { normalizeE164, resolveUserPath } from "openclaw/plugin-sdk/text-utility-runtime"; export { assertWebChannel, isSelfChatMode,