fix(concurrency): eliminate all identified double-queue and race conditions
Implements full audit from double_queue_audit.md. ## Critical fix(backend): reanalyze-batch — per-scope in-flight guard (messages.routes.ts) Two concurrent admin sessions clicking 'Retry All Errors' simultaneously now get a 409 REANALYZE_BATCH_IN_PROGRESS for the same guildId:channelId scope. Prevents the recovery worker from being triggered twice for the same set of error messages. fix(discord-gateway): messageUpdate embed resolution skip (messageCapture.ts) Discord fires messageUpdate when link previews resolve 1-2s after send even though the message body is unchanged. Compare newContent vs existingContent before resetting ai_status to pending and re-queueing LLM. Eliminates a spurious duplicate analysis that could overwrite a valid result. ## High fix(discord-gateway): scheduleAutoDelete idempotency (aiAnalyzer.ts) Add autoDeleteInFlight Set. Both processBatch and processIndividualFallback call scheduleAutoDelete; without the guard, a message that races through both paths launches two concurrent attemptAutoDeleteFlaggedMessage calls, producing a duplicate moderation-action log entry and a Discord 10008 error. The Set is cleaned up in a .finally() block after each attempt completes. ## Medium fix(frontend): revert optimistic pending state on HTTP failure (useMessages.ts) Capture the prior MessageRecord inside the setMessages functional updater (no extra useCallback deps needed). If reanalyzeMessage() throws, restore the snapshot so the UI reflects the real DB state instead of lying. fix(discord-gateway): exclude individual_analysis_exhausted from recovery queries Both getConversationKeysWithIncompleteAnalysis and getIncompleteMessagesByConversation now add a NOT LIKE guard for individual_analysis_exhausted. Prevents an infinite recovery loop if a bug ever writes both flags to the same row. ## Low fix(discord-gateway): unified timer path in scheduleConversationAnalysis Remove the separate cooldown-path that created a secondary timer calling scheduleConversationAnalysis recursively. Replace with a single clear-and-reset pattern where delayMs = max(cooldownRemainder+500, debounce). Eliminates the edge case where both timers were live simultaneously. ## Misc fix(backend): cast req.params.id to String() to satisfy Express typings Pre-existing tsc error (string | string[]) exposed by our edit. String() is correct; route params are always scalar strings at runtime.
This commit is contained in:
@@ -14,14 +14,21 @@ import { messagesService } from "./messages.service.js";
|
||||
const logger = createChildLogger("messages.routes");
|
||||
|
||||
/**
|
||||
* Per-message in-flight guard for the reanalyze endpoint.
|
||||
* Per-message in-flight guard for the single 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.
|
||||
* worker triggers for the same message.
|
||||
*/
|
||||
const reanalyzeInFlight = new Set<string>();
|
||||
|
||||
/**
|
||||
* Per-scope in-flight guard for the batch reanalyze endpoint.
|
||||
* Scope key = "guildId:channelId" (empty string used for undefined parts).
|
||||
* Two concurrent batch-reanalyze requests for the same scope are rejected
|
||||
* with 409 so the recovery worker is not triggered multiple times for the
|
||||
* same set of error messages.
|
||||
*/
|
||||
const reanalyzeBatchInFlight = new Set<string>();
|
||||
|
||||
export function createMessagesRouter(): Router {
|
||||
const router = express.Router();
|
||||
|
||||
@@ -50,22 +57,40 @@ export function createMessagesRouter(): Router {
|
||||
messageIds?: string[];
|
||||
};
|
||||
|
||||
const count = await messagesService.reanalyzeErrorBatch({
|
||||
guildId,
|
||||
channelId,
|
||||
messageIds,
|
||||
});
|
||||
// Idempotency guard: one concurrent batch-reanalyze per scope.
|
||||
// Prevents two admin sessions clicking simultaneously from each
|
||||
// triggering the recovery worker for the same set of messages.
|
||||
const scopeKey = `${guildId ?? ""}:${channelId ?? ""}`;
|
||||
if (reanalyzeBatchInFlight.has(scopeKey)) {
|
||||
res
|
||||
.status(409)
|
||||
.json({ error: "REANALYZE_BATCH_IN_PROGRESS", scope: scopeKey });
|
||||
return;
|
||||
}
|
||||
|
||||
reanalyzeBatchInFlight.add(scopeKey);
|
||||
let count = 0;
|
||||
try {
|
||||
count = await messagesService.reanalyzeErrorBatch({
|
||||
guildId,
|
||||
channelId,
|
||||
messageIds,
|
||||
});
|
||||
} finally {
|
||||
reanalyzeBatchInFlight.delete(scopeKey);
|
||||
}
|
||||
|
||||
logger.info({ count, guildId, channelId }, "Batch reanalyze completed");
|
||||
res.status(200).json({ ok: true, count });
|
||||
}),
|
||||
);
|
||||
|
||||
|
||||
// POST /api/messages/:id/reanalyze - Mark single message for re-analysis
|
||||
router.post(
|
||||
"/messages/:id/reanalyze",
|
||||
asyncHandler(async (req: Request, res: Response) => {
|
||||
const id = req.params.id;
|
||||
const id = String(req.params.id ?? "");
|
||||
if (!id) {
|
||||
res.status(400).json({ error: "MISSING_ID" });
|
||||
return;
|
||||
|
||||
@@ -61,9 +61,21 @@ function broadcastAnalysisCompleted(row: MessageRecord): void {
|
||||
|
||||
function scheduleAutoDelete(row: MessageRecord): void {
|
||||
if (row.ai_status !== "flagged" && row.ai_status !== "warn") return;
|
||||
|
||||
// Idempotency guard: if a concurrent path (batch + individual fallback) both
|
||||
// produce a result for the same message, only the first call proceeds.
|
||||
if (autoDeleteInFlight.has(row.id)) {
|
||||
logger.debug(
|
||||
{ messageId: row.id },
|
||||
"Auto-delete skipped: already in-flight for this message",
|
||||
);
|
||||
return;
|
||||
}
|
||||
autoDeleteInFlight.add(row.id);
|
||||
|
||||
const run = () => {
|
||||
attemptAutoDeleteFlaggedMessage(moderationClient, row).catch(
|
||||
(error: unknown) => {
|
||||
attemptAutoDeleteFlaggedMessage(moderationClient, row)
|
||||
.catch((error: unknown) => {
|
||||
logger.error(
|
||||
{
|
||||
messageId: row.id,
|
||||
@@ -71,8 +83,10 @@ function scheduleAutoDelete(row: MessageRecord): void {
|
||||
},
|
||||
"Unexpected auto-delete error",
|
||||
);
|
||||
},
|
||||
);
|
||||
})
|
||||
.finally(() => {
|
||||
autoDeleteInFlight.delete(row.id);
|
||||
});
|
||||
};
|
||||
|
||||
if (config.AUTO_DELETE_FLAGGED_DELAY_MS > 0) {
|
||||
@@ -82,6 +96,7 @@ function scheduleAutoDelete(row: MessageRecord): void {
|
||||
setImmediate(run);
|
||||
}
|
||||
|
||||
|
||||
function isAgeRestrictedMessage(message: MessageRecord): boolean {
|
||||
return isAgeRestrictedMetadata(message.metadata);
|
||||
}
|
||||
@@ -148,6 +163,16 @@ const conversationProcessing = new Map<string, number>();
|
||||
/** Cooldown expiry timestamp per conversation key after an error. */
|
||||
const conversationErrorCooldown = new Map<string, number>();
|
||||
|
||||
/**
|
||||
* Per-message in-flight guard for the auto-delete side-effect.
|
||||
* `scheduleAutoDelete` is called from both `processBatch` (on batch success)
|
||||
* and `processIndividualFallback` (on individual success). For a message that
|
||||
* races through both paths, without this guard two concurrent
|
||||
* `attemptAutoDeleteFlaggedMessage` calls would be launched — producing a
|
||||
* duplicate moderation-action log and an unnecessary Discord 10008 error.
|
||||
*/
|
||||
const autoDeleteInFlight = new Set<string>();
|
||||
|
||||
let activeRequests = 0;
|
||||
let lastError: string | null = null;
|
||||
let moderationClient: Client | undefined;
|
||||
@@ -875,6 +900,11 @@ async function processBatch(
|
||||
* .catch() so DB errors don't produce unhandled promise rejections.
|
||||
* FIX #6: Calls pickBatchWithinBudget after fetching messages so token budget
|
||||
* is respected before handing the batch to the LLM.
|
||||
* FIX #7: Unified single-timer path — always clear-and-reset one timer per
|
||||
* conversation key regardless of whether a cooldown is active. The delay is
|
||||
* simply max(cooldownRemainder+500, debounce) so the same timer serves both
|
||||
* the "throttled by error cooldown" and "normal debounce" cases, eliminating
|
||||
* the previous two-path logic that could leave both timers live simultaneously.
|
||||
*/
|
||||
function scheduleConversationAnalysis(conversationKey: string): void {
|
||||
if (isConversationProcessingLocked(conversationKey)) {
|
||||
@@ -884,24 +914,20 @@ function scheduleConversationAnalysis(conversationKey: string): void {
|
||||
const convoCooldown = conversationErrorCooldown.get(conversationKey) ?? 0;
|
||||
const convoErrors = conversationConsecutiveErrors.get(conversationKey) ?? 0;
|
||||
|
||||
if (convoCooldown && Date.now() < convoCooldown) {
|
||||
if (!conversationDebounceTimers.has(conversationKey)) {
|
||||
const remaining = convoCooldown - Date.now();
|
||||
const timer = setTimeout(() => {
|
||||
conversationDebounceTimers.delete(conversationKey);
|
||||
scheduleConversationAnalysis(conversationKey);
|
||||
}, remaining + 500);
|
||||
conversationDebounceTimers.set(conversationKey, timer);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
// Block scheduling if this conversation already hit the per-conversation
|
||||
// error threshold — prevents rapid retry loops on the same broken convo.
|
||||
// Hard-block: circuit breaker threshold reached AND cooldown still active.
|
||||
if (convoErrors >= MAX_CONSECUTIVE_ERRORS && Date.now() < convoCooldown) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Unified delay: honour the cooldown window if active, otherwise use the
|
||||
// normal debounce interval. Always clear-and-reset so only ONE timer is
|
||||
// ever pending per conversation key regardless of call source.
|
||||
const now = Date.now();
|
||||
const delayMs =
|
||||
convoCooldown > now
|
||||
? convoCooldown - now + 500
|
||||
: config.AI_ANALYSIS_DEBOUNCE_MS;
|
||||
|
||||
const existingTimer = conversationDebounceTimers.get(conversationKey);
|
||||
if (existingTimer) {
|
||||
clearTimeout(existingTimer);
|
||||
@@ -956,11 +982,12 @@ function scheduleConversationAnalysis(conversationKey: string): void {
|
||||
"Failed to fetch or dispatch pending messages for scheduled analysis",
|
||||
);
|
||||
});
|
||||
}, config.AI_ANALYSIS_DEBOUNCE_MS);
|
||||
}, delayMs);
|
||||
|
||||
conversationDebounceTimers.set(conversationKey, timer);
|
||||
}
|
||||
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Public API
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
@@ -236,6 +236,21 @@ export function registerMessageCapture(client: Client): void {
|
||||
const existing = await getMessageById(newMessage.id);
|
||||
|
||||
if (existing) {
|
||||
const newContent = getDisplayContent(newMessage as Message);
|
||||
const existingContent = existing.edited_content ?? existing.content;
|
||||
|
||||
// Skip if the displayed text is identical — Discord fires `messageUpdate`
|
||||
// for embed resolution (link previews) which does NOT change the message
|
||||
// body. Re-setting ai_status + re-queuing LLM in that case wastes a call
|
||||
// and risks overwriting a valid completed analysis with a duplicate.
|
||||
if (newContent === existingContent) {
|
||||
logger.debug(
|
||||
{ messageId: newMessage.id },
|
||||
"messageUpdate skipped: content unchanged (embed resolution or no-op)",
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const editedAt = Date.now();
|
||||
await updateMessageAsEdited(
|
||||
newMessage.id,
|
||||
|
||||
@@ -807,15 +807,21 @@ export async function getConversationKeysWithIncompleteAnalysis(
|
||||
try {
|
||||
const database = db();
|
||||
const rows = await database
|
||||
.selectDistinct<Array<{ thread_id: string | null; channel_id: string }>>({
|
||||
thread_id: messagesTable.thread_id,
|
||||
channel_id: messagesTable.channel_id,
|
||||
})
|
||||
.selectDistinct<Array<{ thread_id: string | null; channel_id: string }>>(
|
||||
{
|
||||
thread_id: messagesTable.thread_id,
|
||||
channel_id: messagesTable.channel_id,
|
||||
},
|
||||
)
|
||||
.from(messagesTable)
|
||||
.where(
|
||||
and(
|
||||
eq(messagesTable.ai_status, "error"),
|
||||
sql`${messagesTable.ai_moderation_flags} LIKE ${"%analysis_incomplete%"}`,
|
||||
// Exclude rows that have already been exhausted by the individual
|
||||
// fallback pipeline — prevents an infinite recovery loop if both
|
||||
// flags are ever written to the same row due to a bug.
|
||||
sql`(${messagesTable.ai_moderation_flags} IS NULL OR ${messagesTable.ai_moderation_flags} NOT LIKE ${"%individual_analysis_exhausted%"})`,
|
||||
isNull(messagesTable.deleted_at),
|
||||
),
|
||||
)
|
||||
@@ -861,6 +867,9 @@ export async function getIncompleteMessagesByConversation(
|
||||
),
|
||||
eq(messagesTable.ai_status, "error"),
|
||||
sql`${messagesTable.ai_moderation_flags} LIKE ${"%analysis_incomplete%"}`,
|
||||
// Same guard as getConversationKeysWithIncompleteAnalysis: exclude
|
||||
// rows that are already exhausted to prevent re-entry to recovery.
|
||||
sql`(${messagesTable.ai_moderation_flags} IS NULL OR ${messagesTable.ai_moderation_flags} NOT LIKE ${"%individual_analysis_exhausted%"})`,
|
||||
isNull(messagesTable.deleted_at),
|
||||
),
|
||||
)
|
||||
@@ -880,6 +889,7 @@ export async function getIncompleteMessagesByConversation(
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// Message Reviews CRUD
|
||||
// ====================
|
||||
|
||||
|
||||
@@ -78,8 +78,13 @@ export function useMessages() {
|
||||
}, [cursor, loadingMore]);
|
||||
|
||||
const reanalyze = useCallback(async (id: string): Promise<void> => {
|
||||
setMessages((prev) =>
|
||||
prev.map((message) =>
|
||||
// Capture prior state inside the functional updater so we don't need
|
||||
// `messages` as a useCallback dependency (avoids stale closure churn).
|
||||
let saved: MessageRecord | undefined;
|
||||
|
||||
setMessages((prev) => {
|
||||
saved = prev.find((m) => m.id === id);
|
||||
return prev.map((message) =>
|
||||
message.id === id
|
||||
? {
|
||||
...message,
|
||||
@@ -88,11 +93,24 @@ export function useMessages() {
|
||||
ai_analysis: null,
|
||||
}
|
||||
: message,
|
||||
),
|
||||
);
|
||||
await reanalyzeMessage(id);
|
||||
);
|
||||
});
|
||||
|
||||
try {
|
||||
await reanalyzeMessage(id);
|
||||
} catch (err) {
|
||||
// HTTP failed — revert the optimistic update so the UI stays truthful.
|
||||
if (saved) {
|
||||
const snapshot = saved;
|
||||
setMessages((prev) =>
|
||||
prev.map((message) => (message.id === id ? snapshot : message)),
|
||||
);
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
}, []);
|
||||
|
||||
|
||||
const reanalyzeAllErrors = useCallback(async (): Promise<number> => {
|
||||
// Optimistically mark all error messages as pending
|
||||
setMessages((prev) =>
|
||||
|
||||
Reference in New Issue
Block a user