diff --git a/extensions/google/speech-provider.ts b/extensions/google/speech-provider.ts index 482444945fb6..c24191bdfe56 100644 --- a/extensions/google/speech-provider.ts +++ b/extensions/google/speech-provider.ts @@ -7,6 +7,7 @@ import { sanitizeConfiguredModelProviderRequest, } from "openclaw/plugin-sdk/provider-http"; import type { OpenClawConfig } from "openclaw/plugin-sdk/provider-onboard"; +import { retryAsync } from "openclaw/plugin-sdk/retry-runtime"; import { normalizeResolvedSecretInputString } from "openclaw/plugin-sdk/secret-input"; import type { SpeechDirectiveTokenParseContext, @@ -495,18 +496,11 @@ async function synthesizeGoogleTtsPcm(params: { speakerName?: string; timeoutMs: number; }): Promise { - let lastError: unknown; - for (let attempt = 0; attempt < 2; attempt += 1) { - try { - return await synthesizeGoogleTtsPcmOnce(params); - } catch (err) { - lastError = err; - if (!isGoogleTtsRetryableError(err) || attempt > 0) { - throw err; - } - } - } - throw lastError instanceof Error ? lastError : new Error(String(lastError)); + return await retryAsync(() => synthesizeGoogleTtsPcmOnce(params), { + attempts: 2, + minDelayMs: 0, + shouldRetry: isGoogleTtsRetryableError, + }); } type GoogleTtsSynthesisRequest = Pick< diff --git a/extensions/msteams/src/messenger.ts b/extensions/msteams/src/messenger.ts index 51e5b3577281..bd3dc1f8662f 100644 --- a/extensions/msteams/src/messenger.ts +++ b/extensions/msteams/src/messenger.ts @@ -9,8 +9,9 @@ import { resolveSendableOutboundReplyParts, type ReplyPayload, } from "openclaw/plugin-sdk/reply-payload"; +import { retryAsync } from "openclaw/plugin-sdk/retry-runtime"; +import { sleepWithAbort } from "openclaw/plugin-sdk/runtime-env"; import { normalizeOptionalLowercaseString } from "openclaw/plugin-sdk/string-coerce-runtime"; -import { sleep } from "openclaw/plugin-sdk/text-utility-runtime"; import { loadWebMedia } from "openclaw/plugin-sdk/web-media"; import type { MarkdownTableMode, MSTeamsReplyStyle, OpenClawConfig } from "../runtime-api.js"; import { AI_GENERATED_ENTITY } from "./ai-entity.js"; @@ -211,10 +212,6 @@ function computeRetryDelayMs( return clampMs(exponential, opts.maxDelayMs); } -function shouldRetry(classification: ReturnType): boolean { - return classification.kind === "replay-safe"; -} - export function renderReplyPayloadsToMessages( replies: ReplyPayload[], options: MSTeamsReplyRenderOptions, @@ -428,34 +425,25 @@ export async function sendMSTeamsMessages(params: { return await sendOnce(); } - for (const attempt of Array.from( - { length: retryOptions.maxAttempts }, - (_, index) => index + 1, - )) { - try { - return await sendOnce(); - } catch (err) { - const classification = classifyMSTeamsSendError(err); - const canRetry = attempt < retryOptions.maxAttempts && shouldRetry(classification); - if (!canRetry) { - throw err; - } - - const delayMs = computeRetryDelayMs(attempt, classification, retryOptions); - const nextAttempt = attempt + 1; + return await retryAsync(sendOnce, { + attempts: retryOptions.maxAttempts, + minDelayMs: 0, + maxDelayMs: retryOptions.maxDelayMs, + shouldRetry: (err) => classifyMSTeamsSendError(err).kind === "replay-safe", + delayMs: ({ attempt, err }) => + computeRetryDelayMs(attempt, classifyMSTeamsSendError(err), retryOptions), + onRetry: ({ attempt, err, delayMs }) => { params.onRetry?.({ messageIndex: meta.messageIndex, messageCount: meta.messageCount, - nextAttempt, + nextAttempt: attempt + 1, maxAttempts: retryOptions.maxAttempts, delayMs, - classification, + classification: classifyMSTeamsSendError(err), }); - - await sleep(delayMs); - } - } - throw new Error("unreachable Teams send retry loop exit"); + }, + sleep: (delayMs) => sleepWithAbort(delayMs), + }); }; let providerDispatchStarted = false; diff --git a/extensions/nostr/src/nostr-cursor.ts b/extensions/nostr/src/nostr-cursor.ts index 467b340c5c38..b1c0815459ce 100644 --- a/extensions/nostr/src/nostr-cursor.ts +++ b/extensions/nostr/src/nostr-cursor.ts @@ -1,4 +1,6 @@ import type { Event } from "nostr-tools"; +import { retryAsync } from "openclaw/plugin-sdk/retry-runtime"; +import { sleepWithAbort } from "openclaw/plugin-sdk/runtime-env"; const CURSOR_WRITE_RETRY_MS = [0, 100, 300] as const; const CURSOR_RECOVERY_RETRY_MS = 1_000; @@ -72,21 +74,16 @@ export function createNostrCursorStateWriter(options: { let recoveryFlush: Promise | undefined; const writeWithRetry = async (cursor: number): Promise => { - let lastError: unknown; - for (const delayMs of CURSOR_WRITE_RETRY_MS) { - if (delayMs > 0) { - await new Promise((resolve) => { - setTimeout(resolve, delayMs); - }); - } - try { - await options.write(cursor); - return; - } catch (error) { - lastError = error; - } + try { + await retryAsync(() => options.write(cursor), { + attempts: CURSOR_WRITE_RETRY_MS.length, + minDelayMs: 0, + delayMs: ({ attempt }) => CURSOR_WRITE_RETRY_MS[attempt] ?? 0, + sleep: (delayMs) => sleepWithAbort(delayMs), + }); + } catch (error) { + throw new Error("Nostr cursor state write failed.", { cause: error }); } - throw new Error("Nostr cursor state write failed.", { cause: lastError }); }; const enqueueWrite = (cursor: number): Promise => { diff --git a/extensions/nostr/src/nostr-ingress.ts b/extensions/nostr/src/nostr-ingress.ts index d1bc93437ffe..76297d5cc739 100644 --- a/extensions/nostr/src/nostr-ingress.ts +++ b/extensions/nostr/src/nostr-ingress.ts @@ -8,6 +8,8 @@ import { type ChannelIngressQueue, } from "openclaw/plugin-sdk/channel-outbound"; import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; +import { retryAsync } from "openclaw/plugin-sdk/retry-runtime"; +import { sleepWithAbort } from "openclaw/plugin-sdk/runtime-env"; import { inspectNostrIngressEvent, isNostrIngressRecord, @@ -246,28 +248,29 @@ export function createNostrIngress(options: { ); } - let lastError: unknown; - for (const delayMs of NOSTR_INGRESS_APPEND_RETRY_MS) { - if (delayMs > 0) { - await new Promise((resolve) => { - setTimeout(resolve, delayMs); - }); - } - try { - const result = await getQueue().enqueue(prepared.facts.eventId, prepared.payload, { - receivedAt: prepared.receivedAt, - laneKey: prepared.facts.laneKey, - }); - options.afterDurableAppend(prepared.event); - monitor.requestDrain(); - return result.kind === "accepted" ? "accepted" : "duplicate"; - } catch (error) { - lastError = error; - } + try { + return await retryAsync( + async () => { + const result = await getQueue().enqueue(prepared.facts.eventId, prepared.payload, { + receivedAt: prepared.receivedAt, + laneKey: prepared.facts.laneKey, + }); + options.afterDurableAppend(prepared.event); + monitor.requestDrain(); + return result.kind === "accepted" ? "accepted" : "duplicate"; + }, + { + attempts: NOSTR_INGRESS_APPEND_RETRY_MS.length, + minDelayMs: 0, + delayMs: ({ attempt }) => NOSTR_INGRESS_APPEND_RETRY_MS[attempt] ?? 0, + sleep: (delayMs) => sleepWithAbort(delayMs), + }, + ); + } catch (error) { + throw new Error(`Nostr durable admission failed: ${formatErrorMessage(error)}`, { + cause: error, + }); } - throw new Error(`Nostr durable admission failed: ${formatErrorMessage(lastError)}`, { - cause: lastError, - }); }; return { diff --git a/extensions/openai/realtime-quicksilver-sideband.ts b/extensions/openai/realtime-quicksilver-sideband.ts index a199ce59dc90..9fdb53ad5fa7 100644 --- a/extensions/openai/realtime-quicksilver-sideband.ts +++ b/extensions/openai/realtime-quicksilver-sideband.ts @@ -1,3 +1,4 @@ +import { sleepWithAbort } from "openclaw/plugin-sdk/runtime-env"; import type { ClientOptions, RawData } from "ws"; import { openAIQuicksilverAuthHeaders, @@ -128,38 +129,6 @@ function waitForSocketOpen(params: { }); } -function waitForRetryDelay(ms: number, signal: AbortSignal): Promise { - return new Promise((resolve, reject) => { - // Typed as Error so the rejection reason is provably an Error at the call site; - // onAbort normalizes a non-Error AbortSignal reason before it reaches here. - const finish = (error?: Error) => { - clearTimeout(timer); - signal.removeEventListener("abort", onAbort); - if (error) { - reject(error); - } else { - resolve(); - } - }; - const onAbort = () => { - const reason = signal.reason; - finish( - reason instanceof Error - ? reason - : new Error(reason === undefined ? "GPT-Live session stopped" : String(reason), { - cause: reason, - }), - ); - }; - const timer = setTimeout(() => finish(), ms); - timer.unref?.(); - signal.addEventListener("abort", onAbort, { once: true }); - if (signal.aborted) { - onAbort(); - } - }); -} - export async function connectOpenAIQuicksilverSideband(params: { auth: OpenAIQuicksilverAuth; createSocket: OpenAIQuicksilverSocketFactory; @@ -226,7 +195,16 @@ export async function connectOpenAIQuicksilverSideband(params: { throw params.signal.reason; } if (attempt + 1 < SIDEBAND_CONNECT_ATTEMPTS) { - await waitForRetryDelay(SIDEBAND_RETRY_BASE_MS * 2 ** attempt, params.signal); + await sleepWithAbort(SIDEBAND_RETRY_BASE_MS * 2 ** attempt, params.signal, { + ref: false, + }).catch(() => { + const reason = params.signal.reason; + throw reason instanceof Error + ? reason + : new Error(reason === undefined ? "GPT-Live session stopped" : String(reason), { + cause: reason, + }); + }); } } } diff --git a/extensions/slack/src/client-delivery.ts b/extensions/slack/src/client-delivery.ts index cff078b655c0..17aa8c64c38c 100644 --- a/extensions/slack/src/client-delivery.ts +++ b/extensions/slack/src/client-delivery.ts @@ -9,7 +9,8 @@ import { import { buildTimeoutAbortSignal } from "openclaw/plugin-sdk/extension-shared"; import { withTrustedEnvProxyGuardedFetchMode } from "openclaw/plugin-sdk/fetch-runtime"; import { extensionForMime } from "openclaw/plugin-sdk/media-mime"; -import { logVerbose } from "openclaw/plugin-sdk/runtime-env"; +import { retryAsync } from "openclaw/plugin-sdk/retry-runtime"; +import { logVerbose, sleepWithAbort } from "openclaw/plugin-sdk/runtime-env"; import { fetchWithSsrFGuard, type SsrFPolicy } from "openclaw/plugin-sdk/ssrf-runtime"; import { formatSlackError } from "./errors.js"; import { @@ -78,12 +79,6 @@ function hasSlackDnsRequestSignal(err: unknown): boolean { return false; } -function delaySlackDnsRetry(attempt: number): Promise { - return new Promise((resolve) => { - setTimeout(resolve, SLACK_DNS_RETRY_BASE_DELAY_MS * Math.max(1, attempt)); - }); -} - function resolveSlackUploadTimeoutLogUrl(url: string): string | undefined { // Slack puts the upload capability in the URL path. Timeout diagnostics may // name the origin, but must not retain that capability-bearing path. @@ -197,20 +192,18 @@ export async function withSlackDnsRequestRetry( operation: string, fn: () => Promise, ): Promise { - for (const attempt of Array.from({ length: SLACK_DNS_RETRY_ATTEMPTS + 1 }, (_, index) => index)) { - try { - return await fn(); - } catch (err) { - if (attempt >= SLACK_DNS_RETRY_ATTEMPTS || !hasSlackDnsRequestSignal(err)) { - throw err; - } + return await retryAsync(fn, { + attempts: SLACK_DNS_RETRY_ATTEMPTS + 1, + minDelayMs: 0, + shouldRetry: hasSlackDnsRequestSignal, + delayMs: ({ attempt }) => SLACK_DNS_RETRY_BASE_DELAY_MS * Math.max(1, attempt), + onRetry: ({ attempt }) => { logVerbose( - `slack send: retrying ${operation} after transient DNS request error (${attempt + 1}/${SLACK_DNS_RETRY_ATTEMPTS})`, + `slack send: retrying ${operation} after transient DNS request error (${attempt}/${SLACK_DNS_RETRY_ATTEMPTS})`, ); - await delaySlackDnsRetry(attempt + 1); - } - } - throw new Error("unreachable Slack DNS retry loop exit"); + }, + sleep: (delayMs) => sleepWithAbort(delayMs), + }); } export function requireSlackPostMessageTimestamp( diff --git a/extensions/synology-chat/src/client.ts b/extensions/synology-chat/src/client.ts index 893f14ead0a1..cf29b125d1e1 100644 --- a/extensions/synology-chat/src/client.ts +++ b/extensions/synology-chat/src/client.ts @@ -9,8 +9,8 @@ import { collectErrorGraphCandidates, extractErrorCode } from "openclaw/plugin-s import { safeParseJsonWithSchema, safeParseWithSchema } from "openclaw/plugin-sdk/extension-shared"; import { parseStrictNonNegativeInteger } from "openclaw/plugin-sdk/number-runtime"; import { readByteStreamWithLimit } from "openclaw/plugin-sdk/response-limit-runtime"; -import { classifyTransientNetworkErrorCode } from "openclaw/plugin-sdk/retry-runtime"; -import { sleep } from "openclaw/plugin-sdk/runtime-env"; +import { classifyTransientNetworkErrorCode, retryAsync } from "openclaw/plugin-sdk/retry-runtime"; +import { sleep, sleepWithAbort } from "openclaw/plugin-sdk/runtime-env"; import { formatErrorMessage } from "openclaw/plugin-sdk/ssrf-runtime"; import { normalizeLowercaseStringOrEmpty } from "openclaw/plugin-sdk/string-coerce-runtime"; import { chunkTextForOutbound } from "openclaw/plugin-sdk/text-chunking"; @@ -149,24 +149,26 @@ export async function sendMessage( // Synology Chat API requires numeric user_ids to specify the recipient. const body = buildWebhookBody({ text: chunk }, userId); // Retry only proven pre-connect failures; ambiguous webhook replays can duplicate messages. - for (let attempt = 0; attempt < 3; attempt++) { - await waitForSendSlot(); - await onPlatformSendDispatch?.(); - try { - const result = await doPost(incomingUrl, body, allowInsecureSsl); - if (result === "accepted") { - break; - } - return false; - } catch (error) { - if (!isProvenPreConnectFailure(error)) { - return false; - } - } - if (attempt === 2) { - return false; - } - await sleep(300 * 2 ** attempt); + await waitForSendSlot(); + await onPlatformSendDispatch?.(); + let result: SynologyHostedFileSendResult["status"]; + try { + result = await retryAsync(() => doPost(incomingUrl, body, allowInsecureSsl), { + attempts: 3, + minDelayMs: 0, + shouldRetry: isProvenPreConnectFailure, + delayMs: ({ attempt }) => 300 * 2 ** (attempt - 1), + sleep: async (delayMs) => { + await sleepWithAbort(delayMs); + await waitForSendSlot(); + await onPlatformSendDispatch?.(); + }, + }); + } catch { + return false; + } + if (result !== "accepted") { + return false; } } return true; diff --git a/extensions/tlon/src/monitor/index.ts b/extensions/tlon/src/monitor/index.ts index 731e17655276..40c234fa389f 100644 --- a/extensions/tlon/src/monitor/index.ts +++ b/extensions/tlon/src/monitor/index.ts @@ -13,6 +13,7 @@ import { } from "openclaw/plugin-sdk/channel-outbound"; import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; import type { GetReplyOptions, ReplyPayload } from "openclaw/plugin-sdk/reply-runtime"; +import { retryAsync } from "openclaw/plugin-sdk/retry-runtime"; import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime"; import { sleepWithAbort } from "openclaw/plugin-sdk/runtime-env"; import { @@ -108,29 +109,34 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise { - for (const attempt of Array.from( - { length: Math.max(1, maxAttempts) }, - (_, index) => index + 1, - )) { - if (opts.abortSignal?.aborted) { - throw new Error("Aborted while waiting to authenticate"); - } - try { - runtime.log?.(`[tlon] Attempting authentication to ${accountUrl}...`); - return await authenticate(accountUrl, accountCode, { ssrfPolicy }); - } catch (error: unknown) { - runtime.error?.( - `[tlon] Failed to authenticate (attempt ${attempt}): ${formatErrorMessage(error)}`, - ); - if (attempt >= maxAttempts) { + let authAttempt = 0; + return await retryAsync( + async () => { + authAttempt += 1; + if (opts.abortSignal?.aborted) { + throw new Error("Aborted while waiting to authenticate"); + } + try { + runtime.log?.(`[tlon] Attempting authentication to ${accountUrl}...`); + return await authenticate(accountUrl, accountCode, { ssrfPolicy }); + } catch (error: unknown) { + runtime.error?.( + `[tlon] Failed to authenticate (attempt ${authAttempt}): ${formatErrorMessage(error)}`, + ); throw error; } - const delay = Math.min(30000, 1000 * 2 ** (attempt - 1)); - runtime.log?.(`[tlon] Retrying authentication in ${delay}ms...`); - await sleepWithAbort(delay, opts.abortSignal); - } - } - throw new Error("unreachable Tlon authentication retry loop exit"); + }, + { + attempts: Math.max(1, maxAttempts), + minDelayMs: 0, + shouldRetry: () => !opts.abortSignal?.aborted, + delayMs: ({ attempt }) => Math.min(30_000, 1_000 * 2 ** (attempt - 1)), + onRetry: ({ delayMs }) => { + runtime.log?.(`[tlon] Retrying authentication in ${delayMs}ms...`); + }, + sleep: (delayMs) => sleepWithAbort(delayMs, opts.abortSignal), + }, + ); } let api: UrbitSSEClient | null = null; diff --git a/extensions/voice-call/src/providers/twilio.ts b/extensions/voice-call/src/providers/twilio.ts index 145fcd15cf91..8910a7438f97 100644 --- a/extensions/voice-call/src/providers/twilio.ts +++ b/extensions/voice-call/src/providers/twilio.ts @@ -1,6 +1,6 @@ // Voice Call plugin module implements twilio behavior. import crypto from "node:crypto"; -import { setTimeout as sleep } from "node:timers/promises"; +import { retryAsync } from "openclaw/plugin-sdk/retry-runtime"; import { sleepWithAbort } from "openclaw/plugin-sdk/runtime-env"; import { safeEqualSecret } from "openclaw/plugin-sdk/security-runtime"; import { normalizeOptionalString } from "openclaw/plugin-sdk/string-coerce-runtime"; @@ -246,21 +246,18 @@ export class TwilioProvider implements VoiceCallProvider { twiml: string, operation: string, ): Promise { - for (const retryDelayMs of TWILIO_CALL_UPDATE_RETRY_DELAYS_MS) { - try { - await this.apiRequest(`/Calls/${providerCallId}.json`, { Twiml: twiml }); - return; - } catch (err) { - if (!isTwilioCallNotInProgressError(err)) { - throw err; - } + await retryAsync(() => this.apiRequest(`/Calls/${providerCallId}.json`, { Twiml: twiml }), { + attempts: TWILIO_CALL_UPDATE_RETRY_DELAYS_MS.length + 1, + minDelayMs: 0, + shouldRetry: isTwilioCallNotInProgressError, + delayMs: ({ attempt }) => TWILIO_CALL_UPDATE_RETRY_DELAYS_MS[attempt - 1] ?? 0, + onRetry: ({ delayMs }) => { console.warn( - `[voice-call] Twilio ${operation} update hit call state race (21220); retrying in ${retryDelayMs}ms`, + `[voice-call] Twilio ${operation} update hit call state race (21220); retrying in ${delayMs}ms`, ); - await sleep(retryDelayMs); - } - } - await this.apiRequest(`/Calls/${providerCallId}.json`, { Twiml: twiml }); + }, + sleep: (delayMs) => sleepWithAbort(delayMs), + }); } /**