fix(ai-moderation): eliminate double-queue on failure and spam-retry
Three root causes patched: 1. aiAnalyzer.ts — processBatch apiFailedMessages path: After reverting messages to 'pending', suppress shouldScheduleNext (was true by default) to prevent scheduleConversationAnalysis from firing immediately and racing with the recovery worker that will pick up those same pending messages on its next poll cycle. Also release the conversationProcessing lock immediately after the revert so the cooldown timer (not the full processing-timeout) gates the next attempt. 2. messages.routes.ts — POST /messages/:id/reanalyze: Add a per-message reanalyzeInFlight Set. Concurrent requests for the same ID now return HTTP 409 instead of issuing duplicate UPDATEs and triggering multiple recovery worker activations. Also narrow the SQL predicate to 'WHERE id = AND ai_status != pending' so a click that arrives while the recovery worker already picked the message up is a no-op at the DB level.
This commit is contained in:
@@ -13,6 +13,15 @@ import { messagesService } from "./messages.service.js";
|
||||
|
||||
const logger = createChildLogger("messages.routes");
|
||||
|
||||
/**
|
||||
* Per-message in-flight guard for the reanalyze endpoint.
|
||||
* Prevents concurrent spam-clicks from issuing duplicate UPDATE + recovery
|
||||
* worker triggers for the same message. Released as soon as the DB write
|
||||
* completes (or fails), which is fast enough that false-positive blocking
|
||||
* is not a practical concern.
|
||||
*/
|
||||
const reanalyzeInFlight = new Set<string>();
|
||||
|
||||
export function createMessagesRouter(): Router {
|
||||
const router = express.Router();
|
||||
|
||||
@@ -62,11 +71,26 @@ export function createMessagesRouter(): Router {
|
||||
return;
|
||||
}
|
||||
|
||||
const pool = getPool();
|
||||
await pool.query(
|
||||
`UPDATE messages SET ai_status = 'pending' WHERE id = $1`,
|
||||
[id],
|
||||
);
|
||||
// Idempotency guard: reject concurrent duplicate requests for the same ID.
|
||||
if (reanalyzeInFlight.has(id)) {
|
||||
res.status(409).json({ error: "REANALYZE_IN_PROGRESS", messageId: id });
|
||||
return;
|
||||
}
|
||||
|
||||
reanalyzeInFlight.add(id);
|
||||
try {
|
||||
const pool = getPool();
|
||||
await pool.query(
|
||||
// Only revert to pending if the message is not currently being
|
||||
// processed (pending) already — prevents write amplification when
|
||||
// the recovery worker already picked it up between UI clicks.
|
||||
`UPDATE messages SET ai_status = 'pending'
|
||||
WHERE id = $1 AND ai_status != 'pending'`,
|
||||
[id],
|
||||
);
|
||||
} finally {
|
||||
reanalyzeInFlight.delete(id);
|
||||
}
|
||||
|
||||
logger.debug({ id }, "Message marked for re-analysis");
|
||||
res.status(200).json({ ok: true });
|
||||
|
||||
Reference in New Issue
Block a user