shuzhiren-comfyui/任务队列后端/worker_threads/callback_timeout/callbackTimeout.js

import { parentPort } from 'worker_threads'
import redis from '../../redis/index.js'
import initQueue from '../../redis/initQueue.js'

const logger = {
  info: (message) => {
    const timestamp = new Date().toISOString();
    console.log(`[${timestamp}] [CallbackTimeout] INFO: ${message}`);
  },
  warn: (message) => {
    const timestamp = new Date().toISOString();
    console.warn(`[${timestamp}] [CallbackTimeout] WARN: ${message}`);
  },
  error: (message, error) => {
    const timestamp = new Date().toISOString();
    console.error(`[${timestamp}] [CallbackTimeout] ERROR: ${message}`, error || '');
  },
  debug: (message) => {
    const timestamp = new Date().toISOString();
    console.debug(`[${timestamp}] [CallbackTimeout] DEBUG: ${message}`);
  }
};

async function checkTimeoutTasks() {
  try {
    const pendingTasks = await initQueue.getCallbackPendingTasks();
    const now = Date.now();
    const timeoutTasks = [];
    
    for (const [remoteTaskId, taskInfoStr] of Object.entries(pendingTasks)) {
      try {
        const taskInfo = JSON.parse(taskInfoStr);
        const elapsed = now - taskInfo.createdAt;
        
        if (elapsed > initQueue.CALLBACK_TIMEOUT) {
          logger.warn(`检测到超时任务: remoteTaskId=${remoteTaskId}, taskId=${taskInfo.taskId}, 已等待${Math.round(elapsed/1000)}秒`);
          timeoutTasks.push({
            remoteTaskId,
            taskId: taskInfo.taskId,
            aigc: taskInfo.aigc,
            platform: taskInfo.platform,
            elapsed
          });
        }
      } catch (parseError) {
        logger.error(`解析任务信息失败: ${remoteTaskId}`, parseError);
        await initQueue.removeCallbackPendingTask(remoteTaskId);
      }
    }
    
    return timeoutTasks;
  } catch (error) {
    logger.error('检查超时任务失败:', error);
    return [];
  }
}

async function processTimeoutTasks(timeoutTasks) {
  if (timeoutTasks.length === 0) {
    return;
  }
  
  const multi = redis.multi();
  const taskCountMap = new Map();
  
  for (const task of timeoutTasks) {
    const taskKey = `${initQueue.prefix}:task:${task.taskId}`;
    
    const errorMessage = JSON.stringify({
      error: 'callback_timeout',
      message: `回调超时，等待时间超过${Math.round(initQueue.CALLBACK_TIMEOUT/1000)}秒`,
      elapsed: task.elapsed
    });
    
    multi.hSet(taskKey, 'resultData', errorMessage);
    multi.hSet(taskKey, 'status', 'failed');
    multi.lPush(initQueue.errorList, task.taskId);
    multi.del(`${initQueue.callback}:${task.remoteTaskId}`);
    
    const key = `${task.aigc}:${task.platform}`;
    if (taskCountMap.has(key)) {
      taskCountMap.set(key, taskCountMap.get(key) + 1);
    } else {
      taskCountMap.set(key, 1);
    }
    
    await initQueue.removeCallbackPendingTask(task.remoteTaskId);
  }
  
  await multi.exec();
  
  if (taskCountMap.size > 0) {
    await initQueue.addEQtaskALL(timeoutTasks.length);
    logger.info(`已处理${timeoutTasks.length}个超时任务，已推入错误队列`);
  }
}

async function cleanupStaleMappings() {
  try {
    const keys = await redis.keys(`${initQueue.callback}:*`);
    let cleanedCount = 0;
    
    for (const key of keys) {
      if (key === initQueue.callbackPending) {
        continue;
      }
      
      const remoteTaskId = key.replace(`${initQueue.callback}:`, '');
      const taskId = await redis.get(key);
      
      if (taskId) {
        const taskExists = await redis.exists(`${initQueue.prefix}:task:${taskId}`);
        
        if (!taskExists) {
          await redis.del(key);
          await initQueue.removeCallbackPendingTask(remoteTaskId);
          cleanedCount++;
          logger.debug(`清理孤立映射: ${key} -> ${taskId}`);
        }
      }
    }
    
    if (cleanedCount > 0) {
      logger.info(`清理了${cleanedCount}个孤立的回调映射`);
    }
  } catch (error) {
    logger.error('清理孤立映射失败:', error);
  }
}

async function syncCounters() {
  try {
    const actualQueueLength = await redis.lLen(initQueue.callback);
    const storedCount = await initQueue.getCQtasksALL();
    
    if (actualQueueLength !== storedCount) {
      logger.warn(`检测到计数器不一致: 实际队列长度=${actualQueueLength}, 存储计数=${storedCount}`);
      await redis.json.set(initQueue.initInfoKey, '$.CQtasksALL', actualQueueLength);
      logger.info(`已同步计数器: CQtasksALL=${actualQueueLength}`);
    }
    
    
  } catch (error) {
    logger.error('同步计数器失败:', error);
  }
}

(async () => {
  logger.info('回调超时检测线程启动');
  logger.info(`回调超时时间: ${initQueue.CALLBACK_TIMEOUT/1000}秒`);
  
  let checkCount = 0;
  
  while (true) {
    try {
      checkCount++;
      
      const timeoutTasks = await checkTimeoutTasks();
      
      if (timeoutTasks.length > 0) {
        logger.warn(`发现${timeoutTasks.length}个超时任务`);
        await processTimeoutTasks(timeoutTasks);
      }
      
      if (checkCount % 10 === 0) {
        await cleanupStaleMappings();
        await syncCounters();
        checkCount = 0;
      }
      
      const pendingCount = await initQueue.getCallbackPendingCount();
      if (pendingCount > 0) {
        logger.debug(`当前等待回调的任务数: ${pendingCount}`);
      }
      
      await new Promise(resolve => setTimeout(resolve, 30000));
      
    } catch (error) {
      logger.error('回调超时检测循环出错:', error);
      await new Promise(resolve => setTimeout(resolve, 5000));
    }
  }
})();