BullMQ v5로 LLM API 쿼터 초과를 방어하는 작업 큐 — 동시성 제한·우선순위·지수 백오프를 실제 코드로

import { Queue, Worker, Job } from 'bullmq';
import { Redis } from 'ioredis';
 
// ioredis 기본값은 명령 실패 시 자동 재시도를 수행하는데,
// BullMQ의 블로킹 명령과 충돌해 예기치 않은 타임아웃이 발생합니다.
// BullMQ와 함께 쓸 때는 반드시 null로 설정해야 합니다.
const connection = new Redis({ maxRetriesPerRequest: null });
 
const llmQueue = new Queue('llm-requests', {
  connection,
  defaultJobOptions: {
    attempts: 6,
    backoff: {
      type: 'exponential',
      delay: 2000, // 2s → 4s → 8s → 16s → 32s (5회 재시도)
    },
    removeOnComplete: { count: 500 },
    removeOnFail: { count: 200 },
  },
});
 
// 클러스터 전체에서 동시에 최대 5개 Job만 처리
await llmQueue.setGlobalConcurrency(5);

// processor 함수는 Worker 생성 전에 선언되며, worker 변수는 아래에서 할당됩니다.
let worker: Worker;
 
const processor = async (job: Job<{ prompt: string; model: string }>) => {
  try {
    return await callLLMApi(job.data);
  } catch (err: any) {
    if (err.status === 429) {
      const retryAfterRaw = err.headers?.['retry-after'] ?? '30';
      const retryAfter = parseInt(retryAfterRaw, 10);
      // Retry-After 헤더는 HTTP-date 형식으로도 올 수 있으며,
      // 그 경우 parseInt가 NaN을 반환합니다.
      const waitMs = Number.isNaN(retryAfter) ? 30_000 : retryAfter * 1000;
      await worker.rateLimit(waitMs);
      throw Worker.RateLimitError(); // processor 내부에서만 동작
    }
    throw err;
  }
};
 
worker = new Worker('llm-requests', processor, {
  connection,
  concurrency: 3,
  settings: {
    backoffStrategy: (attemptsMade: number) => {
      const base = Math.pow(2, attemptsMade) * 1000;
      const jitter = Math.random() * 1000; // 0~1초 랜덤 지터
      return base + jitter;
    },
  },
});
 
await llmQueue.add('generate', data, {
  attempts: 6,
  backoff: { type: 'custom' },
});

import { Queue, Worker, QueueEvents, Job } from 'bullmq';
import OpenAI from 'openai';
import { Redis } from 'ioredis';
 
const connection = new Redis({ maxRetriesPerRequest: null });
const openai = new OpenAI();
 
const llmQueue = new Queue('llm-requests', {
  connection,
  defaultJobOptions: {
    attempts: 6,
    backoff: { type: 'exponential', delay: 2000 },
    removeOnComplete: { count: 500 },
    removeOnFail: { count: 200 },
  },
});
 
await llmQueue.setGlobalConcurrency(5);
 
let worker: Worker;
 
const processor = async (job: Job<{ prompt: string; model: string }>) => {
  const { prompt, model } = job.data;
  try {
    const response = await openai.chat.completions.create({
      model,
      messages: [{ role: 'user', content: prompt }],
    });
    return response.choices[0].message.content;
  } catch (err: any) {
    if (err.status === 429) {
      const retryAfterRaw = err.headers?.['retry-after'] ?? '30';
      const retryAfter = parseInt(retryAfterRaw, 10);
      // Retry-After가 HTTP-date 형식일 경우 parseInt는 NaN을 반환합니다.
      const waitMs = Number.isNaN(retryAfter) ? 30_000 : retryAfter * 1000;
      await worker.rateLimit(waitMs);
      throw Worker.RateLimitError();
    }
    throw err;
  }
};
 
worker = new Worker('llm-requests', processor, {
  connection,
  concurrency: 3,
  limiter: {
    max: 60,          // 분당 최대 60 요청
    duration: 60_000,
  },
  lockDuration: 90_000,
});

// 프리미엄 사용자: 즉각 처리
await llmQueue.add('generate', { userId, prompt }, {
  priority: 1,
  attempts: 6,
  backoff: { type: 'exponential', delay: 1000 },
});
 
// 무료 사용자 실시간 요청
await llmQueue.add('generate', { userId, prompt }, {
  priority: 10,
  attempts: 6,
  backoff: { type: 'exponential', delay: 2000 },
});
 
// 야간 배치 작업
await llmQueue.add('batch-generate', { userId, prompt }, {
  priority: 100,
  attempts: 3,
  backoff: { type: 'exponential', delay: 3000 },
});

const worker = new Worker('llm-requests', processor, {
  connection,
  concurrency: 3,
  lockDuration: 90_000,    // 모델 최대 응답 시간 + 여유 시간
  maxStalledCount: 2,      // 2회 stalled 발생 시 강제 실패로 전환
  stalledInterval: 30_000, // 30초마다 stalled 체크
});
 
worker.on('stalled', (jobId: string) => {
  console.warn(`Job ${jobId} stalled — 재시도 대기열로 복귀`);
});

const priorityQueue = new Queue('llm-priority-queue', { connection });
const dlqQueue = new Queue('llm-dlq', { connection });
 
const priorityWorker = new Worker('llm-priority-queue', processor, {
  connection,
  concurrency: 3,
  limiter: { max: 40, duration: 60_000 },
  lockDuration: 90_000,
  maxStalledCount: 2,
});
 
const batchWorker = new Worker('llm-batch-queue', processor, {
  connection,
  concurrency: 1,
  limiter: { max: 10, duration: 60_000 },
  lockDuration: 120_000,
  maxStalledCount: 2,
});
 
// 각 큐마다 동일한 패턴으로 DLQ 핸들러를 연결합니다
const priorityQueueEvents = new QueueEvents('llm-priority-queue', { connection });
 
priorityQueueEvents.on('failed', async ({ jobId, failedReason }) => {
  const job = await Job.fromId(priorityQueue, jobId);
  if (job && job.attemptsMade >= (job.opts.attempts ?? 1)) {
    await dlqQueue.add('failed-job', {
      originalData: job.data,
      reason: failedReason,
      source: 'llm-priority-queue',
    });
  }
  alertSlack(`Job ${jobId} failed: ${failedReason}`);
});
 
priorityQueueEvents.on('stalled', ({ jobId }) => {
  alertSlack(`Job ${jobId} stalled in llm-priority-queue — 확인 필요`);
});

// 잘못된 예 — 아무런 효과 없음
worker.on('failed', async (job, err: any) => {
  if (err.status === 429) {
    throw Worker.RateLimitError(); // 여기선 동작하지 않음
  }
});
 
// 올바른 예 — processor 내부에서 처리
const processor = async (job: Job<{ prompt: string; model: string }>) => {
  try {
    return await callLLMApi(job.data);
  } catch (err: any) {
    if (err.status === 429) {
      await worker.rateLimit(30_000);
      throw Worker.RateLimitError();
    }
    throw err;
  }
};