tests/queue.integration.test.ts

import { afterAll, beforeAll, describe, expect, it } from 'vitest';
import { randomUUID } from 'node:crypto';
import { Queue, QueueEvents, Worker } from 'bullmq';
import IORedis from 'ioredis';

// 独立队列前缀确保不会读取、清空或执行开发环境的真实任务。
describe.skipIf(!process.env.TEST_REDIS_URL)(
  'BullMQ 真实 Redis 调度与重试',
  () => {
    const prefix = `test-${randomUUID()}`;
    let connection: IORedis;
    let queue: Queue;
    let events: QueueEvents;
    let worker: Worker;
    const calls = new Map<string, number>();
    beforeAll(async () => {
      connection = new IORedis(
        process.env.TEST_REDIS_URL ?? 'redis://127.0.0.1:6380',
        { maxRetriesPerRequest: null },
      );
      queue = new Queue('test-tasks', { connection, prefix });
      events = new QueueEvents('test-tasks', { connection, prefix });
      worker = new Worker(
        'test-tasks',
        async (job) => {
          const count = (calls.get(job.id!) ?? 0) + 1;
          calls.set(job.id!, count);
          await job.updateProgress(50);
          if (job.data.failAlways || (count < 3 && job.data.failTwice))
            throw new Error('测试临时失败');
          await job.updateProgress(100);
          return '已完成';
        },
        { connection, prefix },
      );

      await events.waitUntilReady();
      await worker.waitUntilReady();
    });
    afterAll(async () => {
      await worker.close();
      await events.close();
      await queue.obliterate({ force: true });
      await queue.close();
      connection.disconnect();
    });
    it('真实队列自动退避重试两次后成功，并保留进度', async () => {
      const job = await queue.add(
        'retry',
        { failTwice: true },
        { attempts: 3, backoff: { type: 'exponential', delay: 30 } },
      );
      expect(await job.waitUntilFinished(events, 5000)).toBe('已完成');
      const finished = await queue.getJob(job.id!);
      expect(finished?.attemptsMade).toBe(3);
      expect(finished?.progress).toBe(100);
    });
    it('达到三次上限停止执行，重试失败任务可重新执行', async () => {
      const job = await queue.add(
        'fail',
        { failAlways: true },
        { attempts: 3, backoff: { type: 'exponential', delay: 30 } },
      );
      await expect(job.waitUntilFinished(events, 5000)).rejects.toThrow(
        '测试临时失败',
      );
      expect((await queue.getJob(job.id!))?.attemptsMade).toBe(3);
      await job.updateData({});
      await job.retry();
      expect(await job.waitUntilFinished(events, 5000)).toBe('已完成');
    });
    it('延迟任务在到期后执行，同 jobId 重复投递只执行一次', async () => {
      const id = randomUUID();
      const start = Date.now();
      const job = await queue.add('delay', {}, { jobId: id, delay: 300 });
      await queue.add('delay', {}, { jobId: id, delay: 300 });
      expect(await job.getState()).toBe('delayed');
      await job.waitUntilFinished(events, 5000);
      expect(Date.now() - start).toBeGreaterThanOrEqual(250);
      expect(calls.get(id)).toBe(1);
    });
  },
);