MythEclipse/GMW
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(discord-gateway): use Discord CDN for image analysis, uploader archive-only

Browse Source 
- mediaDownloader: flip URL candidate order so discord_url is tried
  before uploaded_url (uploaded_url is archive-only fallback)
- ai-analysis-worker: remove upload-pending race guard that blocked
  analysis until Tele upload completed; analysis now runs immediately
  on the Discord CDN URL
- batchProcessor: remove upload-pending defer/poll-backoff logic
- individualFallbackProcessor: remove upload_pending requeue loop
- batchOutcomeClassifier/fallbackResultClassifier: drop upload_pending
  classification (no longer needed)
- tests: update batchOutcomeClassifier + fallbackResultClassifier tests
  to reflect removed upload_pending signal
This commit is contained in:
asepharyana
2026-08-28 22:07:37 +07:00
parent ffbe9959ab
commit 9c9cd8917e
8 changed files with 56 additions and 331 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/discord-gateway/src/modules/ai-moderation/ai-analysis-worker.ts
+6 -57
View File
@@ -93,12 +93,6 @@ type BatchOkResponse = {
ok: true;
conversationKey: string;
rows: MessageRecord[];
/**
* Race-guard signal (2026-08-25): target ids whose attachment upload is
* still in-flight — NO analysis ran for them. The processor must defer
* these (requeue + poll), never fan them out as failures.
*/
uploadPendingIds?: string[];
};
type BatchErrorResponse = {
ok: false;
@@ -109,12 +103,6 @@ type BatchErrorResponse = {
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;
@@ -309,40 +297,9 @@ async function processBatch(job: {
...contextIds,
]);
// Attachment-upload race guard: a message whose attachment is still being
// uploaded (upload_status='pending') must not be analyzed yet. Its
// uploaded_url is not ready, and falling back to the Discord CDN link often
// 404s (expired/purged) — which used to silently produce a text-only
// verdict ("lampiran yang gagal terbaca"). Leave those targets pending; the
// next worker cycle picks them up after the upload lands.
const pendingUploadTargetIds = new Set(
(attachments ?? [])
.filter((a) => a.upload_status === "pending")
.map((a) => a.message_id),
);
const readyMessages =
pendingUploadTargetIds.size === 0
? messages
: messages.filter((m) => !pendingUploadTargetIds.has(m.id));
if (readyMessages.length === 0) {
// Explicit signal (2026-08-25): every target is still upload-pending.
// Returning bare {ok:true, rows:[]} made the processor classify all of
// them "incomplete" and fan out to the individual queue — a hot ~300ms
// requeue loop for the whole upload duration.
return {
ok: true,
conversationKey,
rows: [],
uploadPendingIds: messages.map((m) => m.id),
};
}
// The orchestrator handles text/media split + caching + parallel paths
// internally, so a 20-message batch = 1 text LLM call (+1 media call
// when media is present), not N per-message calls.
const analysisStart = Date.now();
const moderationResult = await runModerationAnalysis({
targets: readyMessages,
targets: messages,
contextBlock,
attachments,
});
@@ -351,7 +308,7 @@ async function processBatch(job: {
const results = moderationResult.results.map((r) =>
normalizeResult(
r as unknown as AnalysisResult,
readyMessages.find((m) => m.id === r.messageId),
messages.find((m) => m.id === r.messageId),
),
);
@@ -380,10 +337,9 @@ async function processBatch(job: {
logger.info(
{
total: readyMessages.length,
total: messages.length,
saved: allRows.length,
conversationKey,
skippedPendingUpload: messages.length - readyMessages.length,
},
"LLM batch analysis complete",
);
@@ -431,16 +387,9 @@ async function processIndividual(job: {
...contextIds,
]);
// Same attachment-upload race guard as the batch path: while the upload is
// still in-flight the uploaded_url is not ready and the Discord CDN fallback
// often 404s — analyzing now would silently produce a text-only verdict.
// Return no results so the message stays pending for the next cycle.
const uploadStillPending = (attachments ?? []).some(
(a) => a.message_id === message.id && a.upload_status === "pending",
);
if (uploadStillPending) {
return { ok: true, results: [], uploadPending: true };
}
// Analysis uses the Discord CDN URL directly (archive-only uploader).
// No upload-pending race guard: analysis proceeds regardless of upload
// status, since discord_url is available immediately at capture time.
try {
const moderationResult = await runModerationAnalysis({
services/discord-gateway/src/modules/ai-moderation/batchOutcomeClassifier.ts
+4 -34
View File
@@ -1,15 +1,7 @@
/**
* batchOutcomeClassifier.ts
*
* Pure partitioner of the batch worker response (2026-08-25).
*
* Bug history: the batch race guard returned `{ok:true, rows:[]}` when every
* target's attachment upload was still in-flight. The processor classified all
* of them as "incomplete" and fanned out to the individual queue, where the
* guard there requeued + rescheduled at the 250ms debounce — a hot ~300ms loop
* for the entire upload duration (~10 cycles in 3s in prod logs). Root fix:
* the worker now reports `uploadPendingIds` explicitly and this pure function
* partitions the outcome so upload-pending targets NEVER enter the fanout.
* Pure partitioner of the batch worker response.
*/
export interface BatchRowLike {
@@ -21,15 +13,12 @@ export interface BatchRowLike {
export interface BatchWorkerResponseLike {
ok?: boolean;
rows?: BatchRowLike[];
/** Explicit race-guard signal from the worker (2026-08-25). */
uploadPendingIds?: string[];
error?: string;
}
/** One target's per-message disposition after a batch attempt. */
export type BatchTargetKind =
| "completed"
| "upload_pending"
| "incomplete"
| "parse_failed"
| "api_failed";
@@ -40,7 +29,7 @@ function flagsOf(row: { ai_moderation_flags?: string | null }): string[] {
const parsed = JSON.parse(row.ai_moderation_flags) as unknown;
return Array.isArray(parsed) ? (parsed as string[]) : [];
} catch {
return [] as string[];
return [];
}
}
@@ -48,14 +37,13 @@ function flagsOf(row: { ai_moderation_flags?: string | null }): string[] {
* Partition the input message ids into per-message dispositions for one batch
* worker response. Pure: no DB/Piscina/logger — unit-testable directly.
*
* Priority per id: explicit uploadPendingIds → completed row → flag-based
* failure kinds → unexplained missing (treated like incomplete).
* Priority per id: completed row → flag-based failure kinds → unexplained
* missing (treated like incomplete).
*/
export function partitionBatchOutcome(
messages: ReadonlyArray<{ id: string }>,
response: BatchWorkerResponseLike,
): Map<string, BatchTargetKind> {
const pendingSet = new Set(response.uploadPendingIds ?? []);
const rowsById = new Map(
(response.rows ?? [])
.filter((r): r is BatchRowLike & { id: string } => Boolean(r?.id))
@@ -64,10 +52,6 @@ export function partitionBatchOutcome(
const out = new Map<string, BatchTargetKind>();
for (const msg of messages) {
if (pendingSet.has(msg.id)) {
out.set(msg.id, "upload_pending");
continue;
}
const row = rowsById.get(msg.id);
if (!row) {
// Unexplained drop: LLM silently omitted it. Same retryable bucket as
@@ -92,17 +76,3 @@ export function partitionBatchOutcome(
}
return out;
}
/**
* Linear backoff ramp for consecutive upload-pending polls:
* poll N (1-based) waits min(base × N, cap). Keeps latency low for fast
* uploads while bounding total polling cost for long uploads.
*/
export function computeUploadPollDelayMs(
consecutivePolls: number,
baseMs: number,
capMs: number,
): number {
const n = Math.max(1, Math.floor(consecutivePolls));
return Math.min(Math.round(baseMs * n), Math.round(capMs));
}
services/discord-gateway/src/modules/ai-moderation/batchProcessor.ts
+6 -94
View File
@@ -4,10 +4,7 @@ import { isAgeRestrictedMetadata } from "../message-capture/messageMetadata.js";
import { messageStore } from "../message-capture/messageStore.js";
import type { MessageRecord } from "../message-capture/types.js";
import { pickBatchWithinBudget as pickBatchWithinBudgetPure } from "./batchBudget.js";
import {
computeUploadPollDelayMs,
partitionBatchOutcome,
} from "./batchOutcomeClassifier.js";
import { partitionBatchOutcome } from "./batchOutcomeClassifier.js";
import { workerPool } from "./circuitBreaker.js";
import { estimateTokens } from "./conversationContext.js";
import {
@@ -25,20 +22,11 @@ import {
const logger = createChildLogger("batch-processor");
/**
* Consecutive upload-pending poll counter per conversation (2026-08-25).
* Drives the linear backoff ramp while attachments are still uploading;
* cleared as soon as a batch comes back with no upload-pending targets.
*/
const conversationUploadPolls = new Map<string, number>();
export interface AnalysisWorkerResponse {
ok: boolean;
conversationKey: string;
rows: MessageRecord[];
error?: string;
/** Explicit upload-in-flight signal from the batch race guard (2026-08-25). */
uploadPendingIds?: string[];
}
// ---------------------------------------------------------------------------
@@ -154,8 +142,6 @@ export async function processBatch(
activeRequests++;
let shouldScheduleNext = false;
/** Set when upload-pending targets defer the next cycle by this many ms. */
let deferredUploadRescheduleMs: number | null = null;
try {
const result = (await workerPool.run({
type: "batch",
@@ -208,29 +194,22 @@ export async function processBatch(
return;
}
// Batch succeeded -- partition per-message outcome explicitly (2026-08-25).
// upload_pending targets are DEFERRED (never fanned out): the old code
// treated them as incomplete -> individual queue -> requeue+250ms
// reschedule -> hot ~300ms loop for the whole upload duration.
// Batch succeeded -- partition per-message outcome (2026-08-25).
const outcomeById = partitionBatchOutcome(messages, result);
const messagesForIndividualQueue: MessageRecord[] = [];
const apiFailedMessages: MessageRecord[] = [];
const uploadPendingMessages: MessageRecord[] = [];
for (const msg of messages) {
switch (outcomeById.get(msg.id)) {
case "upload_pending":
uploadPendingMessages.push(msg);
case "completed":
// Successfully analyzed — already broadcast + auto-delete scheduled
// above. Do NOT re-enqueue for individual fallback.
break;
case "api_failed":
// Preserve the dedicated api-failure semantics below: revert +
// conversation cooldown instead of an immediate individual retry.
apiFailedMessages.push(msg);
break;
case "completed":
// Successfully analyzed — already broadcast + auto-delete scheduled
// above. Do NOT re-enqueue for individual fallback.
break;
default:
// incomplete / parse_failed / unexplained drops stay retryable via
// the individual fallback queue (same semantics as before).
@@ -239,59 +218,6 @@ export async function processBatch(
}
}
if (uploadPendingMessages.length > 0) {
const polls = (conversationUploadPolls.get(conversationKey) ?? 0) + 1;
conversationUploadPolls.set(conversationKey, polls);
const delayMs = computeUploadPollDelayMs(
polls,
config.AI_ANALYSIS_UPLOAD_POLL_MS,
config.AI_ANALYSIS_MAX_UPLOAD_POLL_MS,
);
logger.debug(
{
conversationKey,
count: uploadPendingMessages.length,
ids: uploadPendingMessages.map((m) => m.id),
pollAttempt: polls,
delayMs,
},
"Attachment upload in-flight for batch targets — deferring with poll backoff",
);
// Put the rows back to `pending` so the scheduler owns them again.
await messageStore
.updateMessagesAIAnalysisBulk(
uploadPendingMessages.map((msg) => ({
messageId: msg.id,
result: {
status: "pending",
flags: null,
score: null,
analysis: null,
categories: null,
severity: null,
confidence: null,
recommendedAction: null,
analyzedAt: null,
error: null,
},
})),
)
.catch((err: unknown) => {
logger.error(
{ error: String(err), ids: uploadPendingMessages.map((m) => m.id) },
"Failed to revert upload-pending batch targets to pending",
);
return [] as MessageRecord[];
});
// Poll backoff instead of the 250ms debounce: the finally-block
// schedules the next cycle after this delay instead of immediately.
deferredUploadRescheduleMs = delayMs;
} else {
conversationUploadPolls.delete(conversationKey);
}
if (messagesForIndividualQueue.length > 0) {
logger.warn(
{
@@ -368,11 +294,7 @@ export async function processBatch(
resetConversationBatchFailures(conversationKey);
conversationErrorCooldown.delete(conversationKey);
}
// Upload-pending defer owns the next-cycle timing; don't let the default
// immediate schedule override it.
if (deferredUploadRescheduleMs === null) {
shouldScheduleNext = true;
}
} catch (error) {
recordConversationBatchFailure(conversationKey);
@@ -409,17 +331,7 @@ export async function processBatch(
if (conversationProcessing.get(conversationKey) === processingStartedAt) {
conversationProcessing.delete(conversationKey);
}
if (deferredUploadRescheduleMs !== null) {
// Upload still in-flight: re-schedule after the backoff delay instead of
// immediately (the old path hot-looped at ~250-300ms per cycle).
const delayMs = deferredUploadRescheduleMs;
setTimeout(() => {
// Dynamic import to avoid circular dependency at module scope
import("./batchScheduler.js").then((m) =>
m.scheduleConversationAnalysis(conversationKey),
);
}, delayMs).unref();
} else if (shouldScheduleNext) {
if (shouldScheduleNext) {
setImmediate(() => {
// Dynamic import to avoid circular dependency at module scope
import("./batchScheduler.js").then((m) =>
services/discord-gateway/src/modules/ai-moderation/fallbackResultClassifier.ts
+1 -9
View File
@@ -13,16 +13,10 @@
* has an explicit, testable owner.
*/
export type WorkerResultKind =
| "success"
| "upload_pending"
| "incomplete"
| "error";
export type WorkerResultKind = "success" | "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;
}
@@ -40,7 +34,6 @@ function flagsOf(r: { flags?: string[] | string }): string[] {
/**
* 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.
@@ -50,7 +43,6 @@ function flagsOf(r: { flags?: string[] | string }): string[] {
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),
);
services/discord-gateway/src/modules/ai-moderation/individualFallbackProcessor.ts
+7 -61
View File
@@ -79,75 +79,21 @@ async function processIndividualFallback(
message,
skipNormalAnalysis: false,
} as unknown)) as
| { ok: true; results: AnalysisResult[]; uploadPending?: boolean }
| { ok: true; results: AnalysisResult[] }
| { 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.
// Explicit outcome classification: the old code treated any
// ok:true as a completed moderation, so empty or unexplainable
// results left messages stuck in `processing` until the 300s
// cleanup reverted them.
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 (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",
);
exhaustedOnIncomplete = kind === "incomplete";
analysisResult = null;
}
// No heuristic fallback: an incomplete/errored LLM result stays a
services/discord-gateway/src/modules/ai-moderation/mediaDownloader.ts
+6 -7
View File
@@ -296,14 +296,13 @@ export async function downloadAndExtractFrame(
imageMap: Map<string, MessageImagePart[]>,
): Promise<void> {
const log = createChildLogger("mediaAnalysis");
// Prefer the upload proxy (uploaded_url); the Discord CDN link can expire
// or be purged (404), and a non-OK response used to silently drop the image
// from vision analysis (no log, empty image map → text-only verdict). Try
// each candidate URL in order and surface failures.
// Analysis now uses the Discord CDN URL directly (uploaded_url is archive-only).
// Try discord_url first; fall back to uploaded_url (Tele proxy) if the CDN
// link returns a non-OK response (expired/purged).
const urlCandidates = [
att.uploaded_url,
att.discord_url && att.discord_url !== att.uploaded_url
? att.discord_url
att.discord_url,
att.uploaded_url && att.uploaded_url !== att.discord_url
? att.uploaded_url
: null,
].filter((u): u is string => Boolean(u));
if (urlCandidates.length === 0) return;
services/discord-gateway/tests/batchOutcomeClassifier.test.ts
+21 -53
View File
@@ -1,53 +1,35 @@
// ═══════════════════════════════════════════════════════════════════════════
// partitionBatchOutcome — upload-pending defer vs fanout (2026-08-25)
// ═══════════════════════════════════════════════════════════════════════════
// Bug history: the batch worker's race guard returned {ok:true, rows:[]} while
// attachments were still uploading; every target was classified "incomplete",
// fanned out to the individual queue, requeued there, rescheduled at 250ms —
// a hot ~300ms loop for the whole upload duration (~10 cycles in 3s in prod).
// partitionBatchOutcome — per-message disposition (2026-08-28)
//
// Upload-pending race guard removed: analysis no longer depends on the Tele
// uploader. The worker always runs analysis on the Discord CDN URL directly,
// so there are no upload-pending targets to defer.
import { describe, expect, it } from "vitest";
import {
computeUploadPollDelayMs,
partitionBatchOutcome,
} from "../src/modules/ai-moderation/batchOutcomeClassifier.js";
import { partitionBatchOutcome } from "../src/modules/ai-moderation/batchOutcomeClassifier.js";
const msgs = (...ids: string[]) => ids.map((id) => ({ id }));
describe("partitionBatchOutcome", () => {
it("marks ALL targets upload_pending when the full-batch guard fires", () => {
it("classifies a fully-analyzed batch as completed", () => {
const out = partitionBatchOutcome(msgs("a", "b"), {
ok: true,
rows: [],
uploadPendingIds: ["a", "b"],
});
expect(out.get("a")).toBe("upload_pending");
expect(out.get("b")).toBe("upload_pending");
});
it("never classifies an explicit upload_pending id as incomplete", () => {
// The regression this file exists for: uploadPendingIds must win over the
// missing-row heuristic.
const out = partitionBatchOutcome(msgs("a"), {
ok: true,
rows: [],
uploadPendingIds: ["a"],
});
expect(out.get("a")).not.toBe("incomplete");
expect(out.get("a")).toBe("upload_pending");
});
it("partitions a mixed batch: completed + upload-pending + missing", () => {
const out = partitionBatchOutcome(msgs("ok1", "up1", "gone1"), {
ok: true,
rows: [
{ id: "ok1", ai_status: "clean" },
// up1 has NO row but IS in uploadPendingIds -> deferred, not failed
{ id: "a", ai_status: "clean" },
{ id: "b", ai_status: "flagged" },
],
uploadPendingIds: ["up1"],
});
expect(out.get("ok1")).toBe("completed");
expect(out.get("up1")).toBe("upload_pending");
expect(out.get("gone1")).toBe("incomplete"); // unexplained drop stays retryable
expect(out.get("a")).toBe("completed");
expect(out.get("b")).toBe("completed");
});
it("treats an unexplained missing row as incomplete (retryable)", () => {
const out = partitionBatchOutcome(msgs("a", "gone1"), {
ok: true,
rows: [{ id: "a", ai_status: "clean" }],
// gone1 is missing → incomplete
});
expect(out.get("a")).toBe("completed");
expect(out.get("gone1")).toBe("incomplete");
});
it("routes flag-based failures to their buckets", () => {
@@ -94,17 +76,3 @@ describe("partitionBatchOutcome", () => {
expect(out.get("r")).toBe("completed");
});
});
describe("computeUploadPollDelayMs", () => {
it("ramps linearly and respects the cap", () => {
expect(computeUploadPollDelayMs(1, 1500, 8000)).toBe(1500);
expect(computeUploadPollDelayMs(2, 1500, 8000)).toBe(3000);
expect(computeUploadPollDelayMs(3, 1500, 8000)).toBe(4500);
expect(computeUploadPollDelayMs(9, 1500, 8000)).toBe(8000); // capped
});
it("is safe on degenerate input", () => {
expect(computeUploadPollDelayMs(0, 1500, 8000)).toBe(1500);
expect(computeUploadPollDelayMs(-5, 1000, 4000)).toBe(1000);
});
});
services/discord-gateway/tests/fallbackResultClassifier.test.ts
+4 -15
View File
@@ -1,23 +1,12 @@
// ═══════════════════════════════════════════════════════════════════════════
// classifyIndividualWorkerResult — upload-pending vs success vs incomplete vs error
// ═══════════════════════════════════════════════════════════════════════════
// Bug (2026-08-24): the worker's upload-pending race guard returned
// {ok:true, results:[]}; the processor treated it as a completed moderation,
// leaving messages stuck in `processing` until the 300s cleanup reverted them.
// classifyIndividualWorkerResult — success vs incomplete vs error
//
// Upload-pending race guard removed (2026-08-28): analysis no longer
// depends on the Tele uploader, so there is no upload_pending signal.
import { describe, expect, it } from "vitest";
import { classifyIndividualWorkerResult } from "../src/modules/ai-moderation/fallbackResultClassifier.js";
describe("classifyIndividualWorkerResult", () => {
it("classifies the upload-pending race guard signal FIRST", () => {
expect(
classifyIndividualWorkerResult({
ok: true,
results: [],
uploadPending: true,
}),
).toBe("upload_pending");
});
it("classifies a normal verdict as success", () => {
expect(
classifyIndividualWorkerResult({