fix: increase max active requests and adjust debounce timing for conversation analysis
This commit is contained in:
@@ -32,7 +32,7 @@ const conversationErrorCooldown = new Map<string, number>();
|
|||||||
|
|
||||||
let activeRequests = 0;
|
let activeRequests = 0;
|
||||||
let lastError: string | null = null;
|
let lastError: string | null = null;
|
||||||
const MAX_ACTIVE_REQUESTS = 1;
|
const MAX_ACTIVE_REQUESTS = 2;
|
||||||
const DEBOUNCE_MS = 1500;
|
const DEBOUNCE_MS = 1500;
|
||||||
const RECOVERY_INTERVAL_MS = 15000;
|
const RECOVERY_INTERVAL_MS = 15000;
|
||||||
const ERROR_COOLDOWN_MS = 30000;
|
const ERROR_COOLDOWN_MS = 30000;
|
||||||
@@ -186,6 +186,10 @@ function scheduleConversationAnalysis(conversationKey: string): void {
|
|||||||
clearTimeout(existingTimer);
|
clearTimeout(existingTimer);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// If we have available slots, process immediately with shorter debounce
|
||||||
|
const debounceTime =
|
||||||
|
activeRequests < MAX_ACTIVE_REQUESTS ? Math.min(DEBOUNCE_MS, 500) : DEBOUNCE_MS;
|
||||||
|
|
||||||
// Set new debounced timer
|
// Set new debounced timer
|
||||||
const timer = setTimeout(async () => {
|
const timer = setTimeout(async () => {
|
||||||
conversationDebounceTimers.delete(conversationKey);
|
conversationDebounceTimers.delete(conversationKey);
|
||||||
@@ -205,7 +209,7 @@ function scheduleConversationAnalysis(conversationKey: string): void {
|
|||||||
if (messages.length > 0) {
|
if (messages.length > 0) {
|
||||||
await processBatch(conversationKey, messages);
|
await processBatch(conversationKey, messages);
|
||||||
}
|
}
|
||||||
}, DEBOUNCE_MS);
|
}, debounceTime);
|
||||||
|
|
||||||
conversationDebounceTimers.set(conversationKey, timer);
|
conversationDebounceTimers.set(conversationKey, timer);
|
||||||
}
|
}
|
||||||
@@ -271,6 +275,11 @@ export function startPendingAIAnalysisWorker(): void {
|
|||||||
const conversationKeys = await getPendingConversationKeys(100);
|
const conversationKeys = await getPendingConversationKeys(100);
|
||||||
|
|
||||||
for (const key of conversationKeys) {
|
for (const key of conversationKeys) {
|
||||||
|
// Stop if we've reached max active requests
|
||||||
|
if (activeRequests >= MAX_ACTIVE_REQUESTS) {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
// Skip if already scheduled
|
// Skip if already scheduled
|
||||||
if (conversationDebounceTimers.has(key)) {
|
if (conversationDebounceTimers.has(key)) {
|
||||||
continue;
|
continue;
|
||||||
|
|||||||
@@ -5,7 +5,10 @@ import {
|
|||||||
getAnalysisQueueStatus,
|
getAnalysisQueueStatus,
|
||||||
queueMessageAnalysis,
|
queueMessageAnalysis,
|
||||||
} from "../moderation/aiAnalyzer";
|
} from "../moderation/aiAnalyzer";
|
||||||
import { getMessageById } from "../moderation/messageStore";
|
import {
|
||||||
|
getMessageById,
|
||||||
|
updateMessageAIAnalysis,
|
||||||
|
} from "../moderation/messageStore";
|
||||||
|
|
||||||
export function createAnalysisRoutes(): Router {
|
export function createAnalysisRoutes(): Router {
|
||||||
const router = express.Router();
|
const router = express.Router();
|
||||||
@@ -35,6 +38,17 @@ export function createAnalysisRoutes(): Router {
|
|||||||
throw new AppError("Message not found", "MESSAGE_NOT_FOUND", 404);
|
throw new AppError("Message not found", "MESSAGE_NOT_FOUND", 404);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Reset analysis status to pending so it gets picked up by the analyzer
|
||||||
|
await updateMessageAIAnalysis(id, {
|
||||||
|
status: "pending",
|
||||||
|
flags: null,
|
||||||
|
score: null,
|
||||||
|
raw: null,
|
||||||
|
analysis: null,
|
||||||
|
analyzedAt: null,
|
||||||
|
error: null,
|
||||||
|
});
|
||||||
|
|
||||||
// Queue for analysis
|
// Queue for analysis
|
||||||
await queueMessageAnalysis(id);
|
await queueMessageAnalysis(id);
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user