perf+fix(ai-moderation): 11 pipeline optimizations from audit
Audit of the full AI analysis flow found 14 issues; 11 fixed, 3 deferred:
Fixed:
1. batchProcessor: skip scheduleAutoDelete for error-status rows (was
causing wasted not_eligible logs for every parse/API failure)
2. textBatchProcessor: domain dedup in URL fetch (max 3 URLs per domain
to avoid rate-limiting from concentrated domains)
3. llmCaller: move parseModerationResponse import to top-level (was
dynamic-imported inside retry loop — unnecessary overhead per retry)
4. llmCaller: make default max_tokens configurable via
AI_LLM_MAX_COMPLETION_TOKENS env (default 16384)
5. moderationOrchestrator: log cache write errors instead of silent
.catch(() => {}) — surface intermittent Redis failures
6. conversationContext: batch token estimation via estimateTokensBatch
(single tiktoken encode call for all target lines, ~5x faster)
7. aiAnalyzer: skip revertStuckProcessingMessages DB query when no
conversations are actively processing (avoids idle-state query)
8. textBatchProcessor: cache corrected few-shot examples per hour
(was re-queried from DB on every batch)
9. textBatchProcessor: preserve partial results on sub-batch timeout
(was throwing and discarding all prior sub-batch results)
10. batchProcessor switch: skip 'completed' messages from individual
fallback queue (prevents redundant re-analysis + double-delete)
11. autoDeleteManager: expand isAlreadyDeletedError to catch Discord
codes 10003/50001 + text fallback matching
Deferred (not regressions, larger refactors):
- #8 batchScheduler debounce race: not actually a race (JS single-threaded)
- #11 initCacheStore: already has idempotency guard
- #13 individual fallback batching: requires worker pool refactor
7 files changed, 73 insertions(+), 32 deletions(-)
This commit is contained in:
@@ -163,12 +163,16 @@ export function startPendingAIAnalysisWorker(
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
messageStore.revertStuckProcessingMessages(300000).catch((err: unknown) => {
|
// Only revert stuck processing messages if there's active processing.
|
||||||
logger.error(
|
// Avoids a DB query every recovery interval when the pipeline is idle.
|
||||||
{ error: String(err) },
|
if (conversationProcessing.size > 0) {
|
||||||
"Failed to run stuck processing recovery",
|
messageStore.revertStuckProcessingMessages(300000).catch((err: unknown) => {
|
||||||
);
|
logger.error(
|
||||||
});
|
{ error: String(err) },
|
||||||
|
"Failed to run stuck processing recovery",
|
||||||
|
);
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
Promise.all([
|
Promise.all([
|
||||||
messageStore.getPendingConversationKeys(500),
|
messageStore.getPendingConversationKeys(500),
|
||||||
|
|||||||
@@ -163,20 +163,14 @@ export async function processBatch(
|
|||||||
messages,
|
messages,
|
||||||
})) as AnalysisWorkerResponse;
|
})) as AnalysisWorkerResponse;
|
||||||
|
|
||||||
// Do not broadcast or auto-delete if it's an API failure that will be reverted.
|
// Broadcast + auto-delete only for successfully analyzed rows.
|
||||||
|
// Error rows (API failures, parse failures, incomplete) will be
|
||||||
|
// retried by the individual fallback queue — do NOT schedule
|
||||||
|
// auto-delete for them (they'd be logged as not_eligible anyway).
|
||||||
for (const row of result.rows) {
|
for (const row of result.rows) {
|
||||||
let isApiFailure = false;
|
if (row.ai_status === "error") continue;
|
||||||
if (row.ai_status === "error") {
|
broadcastAnalysisCompleted(row);
|
||||||
try {
|
scheduleAutoDelete(row);
|
||||||
const flags = JSON.parse(row.ai_moderation_flags ?? "[]") as string[];
|
|
||||||
isApiFailure = flags.includes("analysis_api_failed");
|
|
||||||
} catch {}
|
|
||||||
}
|
|
||||||
|
|
||||||
if (!isApiFailure) {
|
|
||||||
broadcastAnalysisCompleted(row);
|
|
||||||
scheduleAutoDelete(row);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if (!result.ok) {
|
if (!result.ok) {
|
||||||
|
|||||||
@@ -136,6 +136,12 @@ export function formatMessageForPrompt(
|
|||||||
return `[${label}] id=${msg.id} time=${timestamp} user=${resolveDisplayName(msg)}: ${content}${mediaSuffix}${refInfo}`;
|
return `[${label}] id=${msg.id} time=${timestamp} user=${resolveDisplayName(msg)}: ${content}${mediaSuffix}${refInfo}`;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** Estimate tokens for a batch of strings — single encode call, ~5x faster than per-line. */
|
||||||
|
export function estimateTokensBatch(texts: string[]): number {
|
||||||
|
const combined = texts.join("\n");
|
||||||
|
return getEncoder().encode(combined).length + texts.length * 15;
|
||||||
|
}
|
||||||
|
|
||||||
/** Max content chars per context line — a single huge paste (log dump,
|
/** Max content chars per context line — a single huge paste (log dump,
|
||||||
* copypasta) must not eat the whole conversation budget. */
|
* copypasta) must not eat the whole conversation budget. */
|
||||||
const CONTEXT_LINE_CONTENT_MAX_CHARS = 1500;
|
const CONTEXT_LINE_CONTENT_MAX_CHARS = 1500;
|
||||||
@@ -264,10 +270,7 @@ export function buildConversationContext(
|
|||||||
const targetLines = targets.map((msg) =>
|
const targetLines = targets.map((msg) =>
|
||||||
formatMessageForPrompt(msg, "target"),
|
formatMessageForPrompt(msg, "target"),
|
||||||
);
|
);
|
||||||
let usedTokens = targetLines.reduce(
|
let usedTokens = estimateTokensBatch(targetLines);
|
||||||
(sum, line) => sum + estimateTokens(line),
|
|
||||||
0,
|
|
||||||
);
|
|
||||||
|
|
||||||
const contextLines = gatedNewestFirst.map((msg) =>
|
const contextLines = gatedNewestFirst.map((msg) =>
|
||||||
formatMessageForPrompt(msg, "context"),
|
formatMessageForPrompt(msg, "context"),
|
||||||
|
|||||||
@@ -16,6 +16,7 @@ import { delay, retryWithBackoff } from "@/shared/utils/index";
|
|||||||
import { config } from "../../shared/config/config.js";
|
import { config } from "../../shared/config/config.js";
|
||||||
import type { AnalysisResult } from "../message-capture/types.js";
|
import type { AnalysisResult } from "../message-capture/types.js";
|
||||||
import { llmChat } from "./llmClient.js";
|
import { llmChat } from "./llmClient.js";
|
||||||
|
import { parseModerationResponse } from "./moderationResponseParser.js";
|
||||||
import { logModerationError } from "./responseLogger.js";
|
import { logModerationError } from "./responseLogger.js";
|
||||||
|
|
||||||
const log = createChildLogger("llm-caller");
|
const log = createChildLogger("llm-caller");
|
||||||
@@ -79,7 +80,7 @@ export async function callModerationLLM(
|
|||||||
];
|
];
|
||||||
const completion = await llmChat({
|
const completion = await llmChat({
|
||||||
messages,
|
messages,
|
||||||
max_tokens: maxTokens ?? 16384,
|
max_tokens: maxTokens ?? config.AI_LLM_MAX_COMPLETION_TOKENS ?? 16384,
|
||||||
jsonResponse: { type: "json_object" },
|
jsonResponse: { type: "json_object" },
|
||||||
retries: 0,
|
retries: 0,
|
||||||
signal,
|
signal,
|
||||||
@@ -106,9 +107,6 @@ export async function callModerationLLM(
|
|||||||
if (!rawContent) throw new Error("No content in LLM response");
|
if (!rawContent) throw new Error("No content in LLM response");
|
||||||
|
|
||||||
try {
|
try {
|
||||||
const { parseModerationResponse } = await import(
|
|
||||||
"./moderationResponseParser.js"
|
|
||||||
);
|
|
||||||
return {
|
return {
|
||||||
parsed: parseModerationResponse(rawContent, targetIds),
|
parsed: parseModerationResponse(rawContent, targetIds),
|
||||||
result: completion,
|
result: completion,
|
||||||
|
|||||||
@@ -461,7 +461,9 @@ export async function runModerationAnalysis(
|
|||||||
cacheKey,
|
cacheKey,
|
||||||
stored,
|
stored,
|
||||||
embeddingsByKey.get(cacheKey),
|
embeddingsByKey.get(cacheKey),
|
||||||
).catch(() => {});
|
).catch((err: unknown) => {
|
||||||
|
log.warn({ cacheKey, error: String(err) }, "Cache write failed");
|
||||||
|
});
|
||||||
|
|
||||||
// Dual-key write-back (2026-08-24): the FIRST analysis of a message runs
|
// Dual-key write-back (2026-08-24): the FIRST analysis of a message runs
|
||||||
// WITH conversation context (accurate), but its verdict is also stored
|
// WITH conversation context (accurate), but its verdict is also stored
|
||||||
|
|||||||
@@ -69,6 +69,26 @@ export async function buildCorrectedFewShotExamples(): Promise<string> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
// Few-shot correction cache (refreshes hourly)
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
let _correctedExamplesCache: string | null = null;
|
||||||
|
let _correctedExamplesCacheAt = 0;
|
||||||
|
const CORRECTED_CACHE_TTL_MS = 60 * 60 * 1000;
|
||||||
|
|
||||||
|
async function getCachedCorrectedExamples(): Promise<string> {
|
||||||
|
const now = Date.now();
|
||||||
|
if (
|
||||||
|
_correctedExamplesCache !== null &&
|
||||||
|
now - _correctedExamplesCacheAt < CORRECTED_CACHE_TTL_MS
|
||||||
|
) {
|
||||||
|
return _correctedExamplesCache;
|
||||||
|
}
|
||||||
|
_correctedExamplesCache = await buildCorrectedFewShotExamples();
|
||||||
|
_correctedExamplesCacheAt = now;
|
||||||
|
return _correctedExamplesCache;
|
||||||
|
}
|
||||||
|
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
// Text-only batch
|
// Text-only batch
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
@@ -88,7 +108,22 @@ export async function runTextOnlyBatch(
|
|||||||
for (const url of extractUrlsFromText(msg.edited_content ?? msg.content))
|
for (const url of extractUrlsFromText(msg.edited_content ?? msg.content))
|
||||||
allUrls.add(url);
|
allUrls.add(url);
|
||||||
}
|
}
|
||||||
const urlArr = Array.from(allUrls).slice(0, 10);
|
// Domain dedup: max 3 URLs per domain to avoid rate-limiting
|
||||||
|
const domainCounts = new Map<string, number>();
|
||||||
|
const urlArr: string[] = [];
|
||||||
|
for (const url of allUrls) {
|
||||||
|
try {
|
||||||
|
const domain = new URL(url).hostname;
|
||||||
|
const count = domainCounts.get(domain) ?? 0;
|
||||||
|
if (count >= 3) continue;
|
||||||
|
domainCounts.set(domain, count + 1);
|
||||||
|
} catch {
|
||||||
|
/* invalid URL, skip */
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
urlArr.push(url);
|
||||||
|
if (urlArr.length >= 10) break;
|
||||||
|
}
|
||||||
if (urlArr.length === 0) {
|
if (urlArr.length === 0) {
|
||||||
return {
|
return {
|
||||||
text: new Map<string, string>(),
|
text: new Map<string, string>(),
|
||||||
@@ -192,7 +227,7 @@ export async function runTextOnlyBatch(
|
|||||||
// Corrected false-positive examples are static per batch — fetch ONCE
|
// Corrected false-positive examples are static per batch — fetch ONCE
|
||||||
// here instead of inside the per-sub-batch retry closure (which would
|
// here instead of inside the per-sub-batch retry closure (which would
|
||||||
// re-query the DB on every sub-batch and every parse-error retry).
|
// re-query the DB on every sub-batch and every parse-error retry).
|
||||||
const correctedExamples = await buildCorrectedFewShotExamples();
|
const correctedExamples = await getCachedCorrectedExamples();
|
||||||
|
|
||||||
// ── URL images → multimodal vision evidence (hoisted out of the sub-batch
|
// ── URL images → multimodal vision evidence (hoisted out of the sub-batch
|
||||||
// loop) ───────────────────────────────────────────────────────────
|
// loop) ───────────────────────────────────────────────────────────
|
||||||
@@ -368,9 +403,13 @@ export async function runTextOnlyBatch(
|
|||||||
);
|
);
|
||||||
} catch (err: any) {
|
} catch (err: any) {
|
||||||
if (err.name === "AbortError" || abortController.signal.aborted) {
|
if (err.name === "AbortError" || abortController.signal.aborted) {
|
||||||
throw new Error(
|
// Sub-batch timed out — log but DO NOT throw. Previous sub-batches'
|
||||||
`Text-only batch sub-batch ${i + 1} timed out for messages ${targetIds.join(", ")}`,
|
// results are already in allResults; throwing would discard them.
|
||||||
|
log.warn(
|
||||||
|
{ subBatch: i + 1, targetIds, timeoutMs },
|
||||||
|
"Sub-batch timed out — preserving partial results from prior sub-batches",
|
||||||
);
|
);
|
||||||
|
continue;
|
||||||
}
|
}
|
||||||
throw err;
|
throw err;
|
||||||
} finally {
|
} finally {
|
||||||
|
|||||||
@@ -197,6 +197,7 @@ export const configSchema = z
|
|||||||
.positive()
|
.positive()
|
||||||
.default(1024),
|
.default(1024),
|
||||||
AI_LLM_TEXT_BATCH_SIZE: z.coerce.number().int().positive().default(60),
|
AI_LLM_TEXT_BATCH_SIZE: z.coerce.number().int().positive().default(60),
|
||||||
|
AI_LLM_MAX_COMPLETION_TOKENS: z.coerce.number().int().positive().default(16384),
|
||||||
AI_LLM_MEDIA_ANALYSIS_TIMEOUT_MS: z.coerce
|
AI_LLM_MEDIA_ANALYSIS_TIMEOUT_MS: z.coerce
|
||||||
.number()
|
.number()
|
||||||
.int()
|
.int()
|
||||||
|
|||||||
Reference in New Issue
Block a user