From c6af313c334015e21abcd71aeeb88690f0cf993e Mon Sep 17 00:00:00 2001 From: MythEclipse Date: Thu, 28 May 2026 01:09:57 +0700 Subject: [PATCH] fix(moderation): immediately abort retries on 429 Too Many Requests - In llmModerationClient.ts (inner retry), if OpenAI throws a 429 (or 401/403), throw p-retry's AbortError to immediately exit the 3-attempt inner retry loop. - In aiAnalyzer.ts (outer retry), propagate the AbortError from runModerationAnalysis so the 2-attempt outer retry loop also aborts immediately. - This ensures that a burst of 20 concurrent tasks hitting rate limits immediately returns the messages to the DB queue (as 'analysis_incomplete') and rapidly increments the individual circuit breaker, pausing processing and preventing a thundering herd instead of making 12 API calls per stuck message. --- src/moderation/aiAnalyzer.ts | 55 ++++++++---- src/moderation/llmModerationClient.ts | 125 +++++++++++++++----------- 2 files changed, 107 insertions(+), 73 deletions(-) diff --git a/src/moderation/aiAnalyzer.ts b/src/moderation/aiAnalyzer.ts index ef56c2b..9d330c2 100644 --- a/src/moderation/aiAnalyzer.ts +++ b/src/moderation/aiAnalyzer.ts @@ -1,5 +1,6 @@ import { existsSync } from "node:fs"; import { fileURLToPath } from "node:url"; +import { AbortError } from "p-retry"; import { Piscina } from "piscina"; import { config } from "../config.js"; import { createChildLogger } from "../logger.js"; @@ -227,28 +228,44 @@ async function processIndividualFallback( const analysisResult = await retryWithBackoff( async () => { - const result = await runModerationAnalysis({ - targets: [message], - contextText: contextLines.join("\n"), - attachments, - }); + try { + const result = await runModerationAnalysis({ + targets: [message], + contextText: contextLines.join("\n"), + attachments, + }); - // If the LLM still dropped our only target, convert to a retryable - // throw so backoff kicks in. Track this so the catch block can - // distinguish it from a transient network/parse failure. - const stillIncomplete = result.results.some((r) => - r.flags.includes("analysis_incomplete"), - ); - if (stillIncomplete) { - exhaustedOnIncomplete = true; - throw new Error( - `LLM returned no result for single-target message ${messageId} — will retry with backoff`, + // If the LLM still dropped our only target, convert to a retryable + // throw so backoff kicks in. Track this so the catch block can + // distinguish it from a transient network/parse failure. + const stillIncomplete = result.results.some((r) => + r.flags.includes("analysis_incomplete"), ); - } + if (stillIncomplete) { + exhaustedOnIncomplete = true; + throw new Error( + `LLM returned no result for single-target message ${messageId} — will retry with backoff`, + ); + } - // Got a real result — clear the incomplete flag. - exhaustedOnIncomplete = false; - return result; + // Got a real result — clear the incomplete flag. + exhaustedOnIncomplete = false; + + return result; + } catch (err: any) { + // Propagate AbortError so outer retry is immediately cancelled on 429. + if (err instanceof AbortError) { + throw err; + } + if ( + err?.status === 429 || + err?.status === 401 || + err?.status === 403 + ) { + throw new AbortError(err); + } + throw err; + } }, { retries: 2, diff --git a/src/moderation/llmModerationClient.ts b/src/moderation/llmModerationClient.ts index 7d50bf9..d9249a8 100644 --- a/src/moderation/llmModerationClient.ts +++ b/src/moderation/llmModerationClient.ts @@ -1,4 +1,5 @@ import OpenAI from "openai"; +import { AbortError } from "p-retry"; import { z } from "zod"; import { config } from "../config.js"; import { createChildLogger } from "../logger.js"; @@ -34,7 +35,10 @@ const openai = new OpenAI({ // Override headers to bypass Cloudflare WAF Bot Fight Mode const headers = new Headers(init?.headers); - headers.set("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"); + headers.set( + "User-Agent", + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", + ); for (const key of Array.from(headers.keys())) { if (key.toLowerCase().startsWith("x-stainless")) { headers.delete(key); @@ -610,61 +614,74 @@ CRITICAL: "message_id" HARUS berupa STRING (dibungkus tanda kutip ganda). Jangan try { const analysis = await retryWithBackoff( async () => { - const completion = await openai.chat.completions.create({ - model: config.AI_LLM_MODEL, - messages: [ - { - role: "user", - content: buildMessageContent(), - }, - ], - temperature: 0.2, - top_p: 0.95, - max_tokens: 16384, - response_format: { - type: "json_object", - }, - stream: false, - chat_template_kwargs: { enable_thinking: false }, - reasoning_budget: 0, - } as OpenAI.Chat.Completions.ChatCompletionCreateParamsNonStreaming); - - if ( - !completion.choices || - !Array.isArray(completion.choices) || - !completion.choices[0] - ) { - throw new Error("Invalid LLM response structure"); - } - - const content = completion.choices[0].message?.content; - if (!content) { - throw new Error("No content in LLM response"); - } - try { - return { - parsed: parseModerationResponse(content, targetIds), - result: completion, - }; - } catch (parseError) { - lastParseError = - parseError instanceof Error - ? parseError.message - : String(parseError); - lastInvalidContent = content; - log.warn( - { - error: lastParseError, - contentLength: content.length, - contentPreview: content.substring(0, 1000), - fullContent: content, - targetIds, - model: config.AI_LLM_MODEL, + const completion = await openai.chat.completions.create({ + model: config.AI_LLM_MODEL, + messages: [ + { + role: "user", + content: buildMessageContent(), + }, + ], + temperature: 0.2, + top_p: 0.95, + max_tokens: 16384, + response_format: { + type: "json_object", }, - "Failed to parse moderation response from LLM", - ); - throw parseError; + stream: false, + chat_template_kwargs: { enable_thinking: false }, + reasoning_budget: 0, + } as OpenAI.Chat.Completions.ChatCompletionCreateParamsNonStreaming); + + if ( + !completion.choices || + !Array.isArray(completion.choices) || + !completion.choices[0] + ) { + throw new Error("Invalid LLM response structure"); + } + + const content = completion.choices[0].message?.content; + if (!content) { + throw new Error("No content in LLM response"); + } + + try { + return { + parsed: parseModerationResponse(content, targetIds), + result: completion, + }; + } catch (parseError) { + lastParseError = + parseError instanceof Error + ? parseError.message + : String(parseError); + lastInvalidContent = content; + log.warn( + { + error: lastParseError, + contentLength: content.length, + contentPreview: content.substring(0, 1000), + fullContent: content, + targetIds, + model: config.AI_LLM_MODEL, + }, + "Failed to parse moderation response from LLM", + ); + throw parseError; + } + } catch (apiError: any) { + // Immediately abort retries on rate limits or auth errors so the + // message can return to the DB queue instead of bursting retries. + if ( + apiError?.status === 429 || + apiError?.status === 401 || + apiError?.status === 403 + ) { + throw new AbortError(apiError); + } + throw apiError; } }, {