diff --git a/services/discord-gateway/src/modules/ai-moderation/ai-analysis-worker.ts b/services/discord-gateway/src/modules/ai-moderation/ai-analysis-worker.ts index c29e61de..0d138797 100644 --- a/services/discord-gateway/src/modules/ai-moderation/ai-analysis-worker.ts +++ b/services/discord-gateway/src/modules/ai-moderation/ai-analysis-worker.ts @@ -93,6 +93,12 @@ 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; @@ -319,7 +325,16 @@ async function processBatch(job: { ? messages : messages.filter((m) => !pendingUploadTargetIds.has(m.id)); if (readyMessages.length === 0) { - return { ok: true, conversationKey, rows: [] }; + // 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 diff --git a/services/discord-gateway/src/modules/ai-moderation/batchOutcomeClassifier.ts b/services/discord-gateway/src/modules/ai-moderation/batchOutcomeClassifier.ts new file mode 100644 index 00000000..10d95b31 --- /dev/null +++ b/services/discord-gateway/src/modules/ai-moderation/batchOutcomeClassifier.ts @@ -0,0 +1,108 @@ +/** + * 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. + */ + +export interface BatchRowLike { + id?: string; + ai_status?: string | null; + ai_moderation_flags?: string | null; +} + +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"; + +function flagsOf(row: { ai_moderation_flags?: string | null }): string[] { + if (!row.ai_moderation_flags) return []; + try { + const parsed = JSON.parse(row.ai_moderation_flags) as unknown; + return Array.isArray(parsed) ? (parsed as string[]) : []; + } catch { + return [] as 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). + */ +export function partitionBatchOutcome( + messages: ReadonlyArray<{ id: string }>, + response: BatchWorkerResponseLike, +): Map { + const pendingSet = new Set(response.uploadPendingIds ?? []); + const rowsById = new Map( + (response.rows ?? []) + .filter((r): r is BatchRowLike & { id: string } => Boolean(r?.id)) + .map((r) => [r.id, r]), + ); + + const out = new Map(); + 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 + // analysis_incomplete — never a silent success. + out.set(msg.id, "incomplete"); + continue; + } + if (row.ai_status !== "error") { + out.set(msg.id, "completed"); + continue; + } + const flags = flagsOf(row); + if (flags.includes("analysis_incomplete")) { + out.set(msg.id, "incomplete"); + } else if (flags.includes("analysis_parse_failed")) { + out.set(msg.id, "parse_failed"); + } else if (flags.includes("analysis_api_failed")) { + out.set(msg.id, "api_failed"); + } else { + out.set(msg.id, "incomplete"); + } + } + 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)); +} diff --git a/services/discord-gateway/src/modules/ai-moderation/batchProcessor.ts b/services/discord-gateway/src/modules/ai-moderation/batchProcessor.ts index a435f0aa..145f750d 100644 --- a/services/discord-gateway/src/modules/ai-moderation/batchProcessor.ts +++ b/services/discord-gateway/src/modules/ai-moderation/batchProcessor.ts @@ -4,6 +4,10 @@ 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 { workerPool } from "./circuitBreaker.js"; import { estimateTokens } from "./conversationContext.js"; import { @@ -21,11 +25,20 @@ 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(); + 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[]; } // --------------------------------------------------------------------------- @@ -141,6 +154,8 @@ 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", @@ -199,37 +214,85 @@ export async function processBatch( return; } - // Batch succeeded -- check for messages the LLM silently dropped or failed - const incompleteMessages: MessageRecord[] = []; - const parseFailedMessages: MessageRecord[] = []; + // 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. + const outcomeById = partitionBatchOutcome(messages, result); + const messagesForIndividualQueue: MessageRecord[] = []; const apiFailedMessages: MessageRecord[] = []; + const uploadPendingMessages: MessageRecord[] = []; for (const msg of messages) { - const row = result.rows.find((r) => r.id === msg.id); - if (!row) { - incompleteMessages.push(msg); - continue; - } - if (row.ai_status === "error") { - let flags: string[] = []; - try { - flags = JSON.parse(row.ai_moderation_flags ?? "[]") as string[]; - } catch {} - - if (flags.includes("analysis_incomplete")) { - incompleteMessages.push(msg); - } else if (flags.includes("analysis_parse_failed")) { - parseFailedMessages.push(msg); - } else if (flags.includes("analysis_api_failed")) { + switch (outcomeById.get(msg.id)) { + case "upload_pending": + uploadPendingMessages.push(msg); + break; + case "api_failed": + // Preserve the dedicated api-failure semantics below: revert + + // conversation cooldown instead of an immediate individual retry. apiFailedMessages.push(msg); - } + break; + default: + // incomplete / parse_failed / unexplained drops stay retryable via + // the individual fallback queue (same semantics as before). + messagesForIndividualQueue.push(msg); + break; } } - const messagesForIndividualQueue = [ - ...incompleteMessages, - ...parseFailedMessages, - ]; + 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( @@ -307,7 +370,11 @@ export async function processBatch( resetConversationBatchFailures(conversationKey); conversationErrorCooldown.delete(conversationKey); } - shouldScheduleNext = true; + // 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); @@ -344,7 +411,17 @@ export async function processBatch( if (conversationProcessing.get(conversationKey) === processingStartedAt) { conversationProcessing.delete(conversationKey); } - if (shouldScheduleNext) { + 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) { setImmediate(() => { // Dynamic import to avoid circular dependency at module scope import("./batchScheduler.js").then((m) => diff --git a/services/discord-gateway/src/shared/config/index.ts b/services/discord-gateway/src/shared/config/index.ts index 6e85fec2..9e5ab7f6 100644 --- a/services/discord-gateway/src/shared/config/index.ts +++ b/services/discord-gateway/src/shared/config/index.ts @@ -254,6 +254,12 @@ export const configSchema = z .positive() .default(10000), AI_ANALYSIS_ERROR_COOLDOWN_MS: z.coerce.number().positive().default(30000), + // Upload-pending batch poll (2026-08-25): when a batch is deferred because + // attachments are still uploading, the processor re-schedules with this + // base delay (linear ramp per consecutive poll, capped) instead of the + // 250ms debounce — the old path hot-looped ~300ms for the whole upload. + AI_ANALYSIS_UPLOAD_POLL_MS: z.coerce.number().positive().default(1500), + AI_ANALYSIS_MAX_UPLOAD_POLL_MS: z.coerce.number().positive().default(8000), // ── AI Analysis Batch ─────────────────────────────────────────────── AI_ANALYSIS_MAX_BATCH_SIZE: z.coerce.number().int().positive().default(200), diff --git a/services/discord-gateway/tests/batchOutcomeClassifier.test.ts b/services/discord-gateway/tests/batchOutcomeClassifier.test.ts new file mode 100644 index 00000000..bfc797b4 --- /dev/null +++ b/services/discord-gateway/tests/batchOutcomeClassifier.test.ts @@ -0,0 +1,110 @@ +// ═══════════════════════════════════════════════════════════════════════════ +// 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). +import { describe, expect, it } from "vitest"; +import { + computeUploadPollDelayMs, + 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", () => { + 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 + ], + 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 + }); + + it("routes flag-based failures to their buckets", () => { + const out = partitionBatchOutcome(msgs("i", "p", "f"), { + ok: true, + rows: [ + { + id: "i", + ai_status: "error", + ai_moderation_flags: JSON.stringify(["analysis_incomplete"]), + }, + { + id: "p", + ai_status: "error", + ai_moderation_flags: JSON.stringify(["analysis_parse_failed"]), + }, + { + id: "f", + ai_status: "error", + ai_moderation_flags: JSON.stringify(["analysis_api_failed"]), + }, + ], + }); + expect(out.get("i")).toBe("incomplete"); + expect(out.get("p")).toBe("parse_failed"); + expect(out.get("f")).toBe("api_failed"); + }); + + it("treats error rows with no known flag as retryable incomplete", () => { + const out = partitionBatchOutcome(msgs("x"), { + ok: true, + rows: [ + { id: "x", ai_status: "error", ai_moderation_flags: '["weird_flag"]' }, + ], + }); + expect(out.get("x")).toBe("incomplete"); + }); + + it("accepts DB-round-trip MessageRecord shape (null ai_status)", () => { + const out = partitionBatchOutcome([{ id: "r", ai_status: null }] as never, { + ok: true, + rows: [{ id: "r", ai_status: null }], + }); + 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); + }); +});