fix(ai-moderation): bedah delay attachment ~330s -> target <20s
Root cause (trace msg 1541417073245290638): - Race-guard upload-pending balik results:[] diperlakukan sbg SUKSES -> row yatam 'processing' sampai cleanup 300s mengembalikan - Vision gagal 3x utk GIF besar (SSE truncation) tanpa fallback Fix: - Sinyal eksplisit uploadPending dari worker race guard - Classifier murni classifyIndividualWorkerResult(): upload_pending -> requeue pending + reschedule segera (250ms), bukan error palsu; empty-results ok:true kini error transien (bug silent-success mati) - llmVision fallback stream:false sekali saat SSE truncation - Safety-net cleanup stuck processing 300s -> 120s
This commit is contained in:
@@ -100,7 +100,16 @@ type BatchErrorResponse = {
|
||||
rows: MessageRecord[];
|
||||
error: string;
|
||||
};
|
||||
type IndividualOkResponse = { ok: true; results: AnalysisResult[] };
|
||||
type IndividualOkResponse = {
|
||||
ok: true;
|
||||
results: AnalysisResult[];
|
||||
/**
|
||||
* Race-guard signal (2026-08-24): the message's attachment upload is still
|
||||
* in-flight — NO analysis ran. The processor must re-queue the message as
|
||||
* `pending` and re-schedule, never treat this as a completed moderation.
|
||||
*/
|
||||
uploadPending?: boolean;
|
||||
};
|
||||
type IndividualErrorResponse = {
|
||||
ok: false;
|
||||
results: AnalysisResult[];
|
||||
@@ -415,7 +424,7 @@ async function processIndividual(job: {
|
||||
(a) => a.message_id === message.id && a.upload_status === "pending",
|
||||
);
|
||||
if (uploadStillPending) {
|
||||
return { ok: true, results: [] };
|
||||
return { ok: true, results: [], uploadPending: true };
|
||||
}
|
||||
|
||||
try {
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
/**
|
||||
* fallbackResultClassifier.ts
|
||||
*
|
||||
* Pure classifier for the individual-fallback worker response.
|
||||
*
|
||||
* Bug history (2026-08-24): the worker's upload-pending race guard returned
|
||||
* `{ ok: true, results: [] }` (a legacy "no results yet" signal), but the
|
||||
* processor treated ANY `ok:true` as a successful moderation. Empty results
|
||||
* meant nothing was written to the DB — the message stayed stuck in
|
||||
* `ai_status='processing'` with nobody watching it until the 300s cleanup
|
||||
* reverted it. That single gap produced the ~330-400s attachment delay
|
||||
* cluster. Classification now happens in ONE pure function so every outcome
|
||||
* has an explicit, testable owner.
|
||||
*/
|
||||
|
||||
export type WorkerResultKind =
|
||||
| "success"
|
||||
| "upload_pending"
|
||||
| "incomplete"
|
||||
| "error";
|
||||
|
||||
export interface ClassifiableWorkerResult {
|
||||
ok?: boolean;
|
||||
/** Upload-pending marker set by ai-analysis-worker's race guard. */
|
||||
uploadPending?: boolean;
|
||||
results?: Array<{ status?: string; flags?: string[] | string } | undefined>;
|
||||
error?: string;
|
||||
}
|
||||
|
||||
function flagsOf(r: { flags?: string[] | string }): string[] {
|
||||
if (!r.flags) return [];
|
||||
if (Array.isArray(r.flags)) return r.flags;
|
||||
try {
|
||||
const parsed = JSON.parse(r.flags) as unknown;
|
||||
return Array.isArray(parsed) ? (parsed as string[]) : [];
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Classify an individual-fallback worker response:
|
||||
* - "upload_pending": explicit race-guard signal — retry shortly, NOT an error.
|
||||
* - "success": at least one result and none is analysis_incomplete.
|
||||
* - "incomplete": LLM ran but dropped/failed this message after retries
|
||||
* (analysis_incomplete flag) — terminal exhausted path.
|
||||
* - "error": anything else (ok:false, or ok:true with NO explainable
|
||||
* results). The old code silently succeeded here — never again.
|
||||
*/
|
||||
export function classifyIndividualWorkerResult(
|
||||
result: ClassifiableWorkerResult,
|
||||
): WorkerResultKind {
|
||||
if (result.uploadPending === true) return "upload_pending";
|
||||
const results = (result.results ?? []).filter(
|
||||
(r): r is NonNullable<typeof r> => Boolean(r),
|
||||
);
|
||||
if (results.length === 0) return "error";
|
||||
if (result.ok !== true) return "error";
|
||||
for (const r of results) {
|
||||
const flags = flagsOf(r);
|
||||
if (flags.includes("analysis_incomplete")) return "incomplete";
|
||||
if ((r.status ?? "") === "") return "error";
|
||||
}
|
||||
return "success";
|
||||
}
|
||||
@@ -8,6 +8,7 @@ import type {
|
||||
} from "../message-capture/types.js";
|
||||
import { getConversationKey, workerPool } from "./circuitBreaker.js";
|
||||
import { fireAlert } from "./conversationState.js";
|
||||
import { classifyIndividualWorkerResult } from "./fallbackResultClassifier.js";
|
||||
import {
|
||||
broadcastAnalysisCompleted,
|
||||
LAST_ERROR,
|
||||
@@ -78,26 +79,82 @@ async function processIndividualFallback(
|
||||
message,
|
||||
skipNormalAnalysis: false,
|
||||
} as unknown)) as
|
||||
| { ok: true; results: AnalysisResult[] }
|
||||
| { ok: true; results: AnalysisResult[]; uploadPending?: boolean }
|
||||
| { ok: false; results: AnalysisResult[]; error: string };
|
||||
|
||||
// Explicit outcome classification (2026-08-24): the old code treated any
|
||||
// ok:true as a completed moderation, so the upload-pending race guard's
|
||||
// empty results left messages stuck in `processing` until the 300s
|
||||
// cleanup reverted them — the root cause of the ~330s attachment delays.
|
||||
const kind = classifyIndividualWorkerResult(workerResult);
|
||||
|
||||
if (kind === "upload_pending") {
|
||||
// Attachment still uploading — put the row back to `pending` and
|
||||
// re-schedule this conversation immediately. The next scheduler cycle
|
||||
// (~debounce 250ms) re-fetches; once upload_status flips to done the
|
||||
// race guard passes and analysis proceeds. NOT an error: never touches
|
||||
// the circuit breaker counters.
|
||||
const revertedRows = await messageStore
|
||||
.updateMessagesAIAnalysisBulk([
|
||||
{
|
||||
messageId,
|
||||
result: {
|
||||
status: "pending",
|
||||
flags: null,
|
||||
score: null,
|
||||
analysis: null,
|
||||
categories: null,
|
||||
severity: null,
|
||||
confidence: null,
|
||||
recommendedAction: null,
|
||||
analyzedAt: null,
|
||||
error: null,
|
||||
},
|
||||
},
|
||||
])
|
||||
.catch((dbErr: unknown) => {
|
||||
logger.error(
|
||||
{ messageId, error: String(dbErr) },
|
||||
"Failed to revert upload-pending message to pending",
|
||||
);
|
||||
return [] as MessageRecord[];
|
||||
});
|
||||
for (const row of revertedRows) {
|
||||
broadcastAnalysisCompleted(row);
|
||||
}
|
||||
logger.debug(
|
||||
{ messageId, conversationKey },
|
||||
"Individual fallback: attachment upload in-flight — requeued as pending + rescheduled",
|
||||
);
|
||||
setImmediate(() => {
|
||||
import("./batchScheduler.js")
|
||||
.then((m) => m.scheduleConversationAnalysis(conversationKey))
|
||||
.catch(() => {});
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
let analysisResult: { results: AnalysisResult[] } | null = null;
|
||||
|
||||
if (workerResult.ok) {
|
||||
const stillIncomplete = workerResult.results.some((r) =>
|
||||
r.flags.includes("analysis_incomplete"),
|
||||
if (kind === "success") {
|
||||
analysisResult = workerResult;
|
||||
} else if (kind === "incomplete") {
|
||||
exhaustedOnIncomplete = true;
|
||||
analysisResult = null;
|
||||
} else {
|
||||
// "error" — includes ok:true with unexplainable empty results (the old
|
||||
// silent-success bug). Throw so it is treated as a transient failure.
|
||||
throw new Error(
|
||||
(workerResult as { error?: string }).error ??
|
||||
"Individual worker returned no explainable results",
|
||||
);
|
||||
if (stillIncomplete) {
|
||||
exhaustedOnIncomplete = true;
|
||||
analysisResult = null;
|
||||
} else {
|
||||
analysisResult = workerResult;
|
||||
}
|
||||
}
|
||||
|
||||
// No heuristic fallback: an incomplete/errored LLM result stays a
|
||||
// retryable error — the recovery worker picks it up later. Producing a
|
||||
// regex/wordlist verdict here would reintroduce false positives.
|
||||
// (incomplete keeps its exhausted flag so the catch writes the terminal
|
||||
// individual_analysis_exhausted status.)
|
||||
if (!analysisResult) {
|
||||
throw new Error(`LLM analysis failed for message ${messageId}`);
|
||||
}
|
||||
|
||||
@@ -356,10 +356,10 @@ export async function llmVision(
|
||||
promptText: string,
|
||||
imageUrl: { url: string },
|
||||
): Promise<string | null> {
|
||||
const completion = await llmChat({
|
||||
const params = {
|
||||
messages: [
|
||||
{
|
||||
role: "user",
|
||||
role: "user" as const,
|
||||
content: [
|
||||
{ type: "text" as const, text: promptText },
|
||||
{ type: "image_url" as const, image_url: imageUrl },
|
||||
@@ -371,9 +371,26 @@ export async function llmVision(
|
||||
temperature: 0.1,
|
||||
top_p: 0.9,
|
||||
retries: 0,
|
||||
stream: true, // router always streams SSE; non-stream waits for full body and times out
|
||||
timeout: config.AI_LLM_VISION_ANALYSIS_TIMEOUT_MS ?? 60_000,
|
||||
});
|
||||
};
|
||||
|
||||
// Streaming first (the router always streams SSE; a non-stream request
|
||||
// waits for the full body and times out on slow models). Fallback (2026-08-24):
|
||||
// large GIFs/images sometimes get their SSE stream truncated mid-flight by
|
||||
// the upstream ("Stream ended before producing a non-ping SSE event") — all
|
||||
// streaming retries fail identically, so retry ONCE with stream:false where
|
||||
// the router assembles the complete response server-side.
|
||||
let completion;
|
||||
try {
|
||||
completion = await llmChat({ ...params, stream: true });
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
if (/stream ended before producing a non-ping sse/i.test(msg)) {
|
||||
completion = await llmChat({ ...params, stream: false });
|
||||
} else {
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
if (!completion) return null;
|
||||
return completion.choices[0]?.message?.content?.trim() ?? null;
|
||||
|
||||
Reference in New Issue
Block a user