예약 작업과 큐 운영

앞 절까지 즉시 실행 작업의 접수, 처리, 재시도, 중복 방지를 구현했습니다. 마지막 절에서는 특정 시각 이후 실행하는 작업, 반복 작업, Worker 처리량, 이벤트와 메트릭, 안전한 종료를 하나의 운영 흐름으로 묶습니다.

BullMQ 5.16 이상에서는 반복 작업을 Job Scheduler의 upsertJobScheduler()로 등록합니다. 같은 scheduler ID를 다시 등록하면 중복 생성하지 않고 설정을 갱신할 수 있습니다.

지연 작업 예약하기

delay는 지금 큐에 넣되 지정한 밀리초가 지난 뒤 처리 가능하게 만드는 옵션입니다. 정확한 실행 시각을 보장하는 타이머가 아니라 그 시각 이후 Worker가 가져갈 수 있다는 의미입니다. 큐가 밀렸거나 Worker가 꺼져 있으면 더 늦게 시작합니다.

예약 요청 DTO를 추가합니다.

src/reports/dto/schedule-report.dto.ts

import { IsDateString, IsInt, IsString, Max, Min } from 'class-validator';

export class ScheduleReportDto {
  @IsString()
  reportId: string;

  @IsString()
  requestedBy: string;

  @IsInt()
  @Min(1)
  @Max(100_000)
  rows: number;

  @IsDateString()
  runAt: string;
}

ReportsService에 예약 메서드를 추가합니다. BadRequestException은 3절에서 Injectable, NotFoundException과 합친 기존 @nestjs/common import를 그대로 사용합니다.

src/reports/reports.service.ts (enqueueAt 추가)

import { ScheduleReportDto } from './dto/schedule-report.dto';

async enqueueAt(dto: ScheduleReportDto) {
  const runAt = new Date(dto.runAt).getTime();
  const delay = runAt - Date.now();

  if (delay <= 0) {
    throw new BadRequestException('runAt must be in the future');
  }

  const job = await this.reportsQueue.add(
    GENERATE_REPORT_JOB,
    {
      reportId: dto.reportId,
      requestedBy: dto.requestedBy,
      rows: dto.rows,
    },
    {
      delay,
      attempts: 4,
      backoff: { type: 'exponential', delay: 1_000, jitter: 0.5 },
      deduplication: { id: dto.reportId },
      removeOnComplete: 100,
      removeOnFail: 100,
    },
  );

  return {
    jobId: job.id,
    state: await job.getState(),
    runAt: new Date(runAt).toISOString(),
  };
}

Controller에서 별도 경로로 노출합니다.

src/reports/reports.controller.ts (예약 API 추가)

import { ScheduleReportDto } from './dto/schedule-report.dto';

@Post('scheduled')
@HttpCode(HttpStatus.ACCEPTED)
schedule(@Body() dto: ScheduleReportDto) {
  return this.reportsService.enqueueAt(dto);
}

Node.js로 현재 시각보다 1분 뒤의 ISO 문자열을 만들어 요청하고 상태를 확인합니다. 이 명령은 셸의 날짜 문법에 의존하지 않아 Windows와 macOS, Linux에서 같은 방식으로 실행할 수 있습니다.

node -e "const runAt=new Date(Date.now()+60000).toISOString(); fetch('http://localhost:3000/reports/scheduled',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({reportId:'scheduled-demo',requestedBy:'user-42',rows:3000,runAt})}).then(async r=>console.log(r.status,await r.text()))"

실행 가능 시각 전에는 delayed, 이후 Worker가 가져가면 active가 됩니다. API 서버와 Redis 서버의 시계가 크게 어긋나면 예상 시각도 어긋나므로 운영 환경에서는 시간 동기화를 확인합니다.

Job Scheduler로 반복 작업 만들기

매일 오래된 보고서 메타데이터를 정리하는 작업을 추가합니다. 먼저 작업 이름과 데이터 타입을 확장합니다.

src/reports/reports.constants.ts

export const REPORT_QUEUE = 'reports';
export const GENERATE_REPORT_JOB = 'report.generate';
export const CLEANUP_REPORTS_JOB = 'reports.cleanup';

src/reports/report-job.types.ts (정리 작업 추가)

export interface CleanupReportsData {
  keepDays: number;
  allowedHour: number;
  timeZone: string;
}

export interface CleanupReportsResult {
  removed: number;
  skipped: boolean;
  reason?: string;
}

애플리케이션이 부팅될 때 scheduler를 등록하는 provider를 만듭니다. scheduler ID는 배포마다 바뀌지 않는 상수여야 합니다.

src/reports/report-scheduler.service.ts

import { InjectQueue } from '@nestjs/bullmq';
import { Injectable, OnApplicationBootstrap } from '@nestjs/common';
import { Queue } from 'bullmq';
import {
  CLEANUP_REPORTS_JOB,
  REPORT_QUEUE,
} from './reports.constants';

@Injectable()
export class ReportSchedulerService implements OnApplicationBootstrap {
  constructor(
    @InjectQueue(REPORT_QUEUE)
    private readonly reportsQueue: Queue,
  ) {}

  async onApplicationBootstrap() {
    await this.reportsQueue.upsertJobScheduler(
      'daily-report-cleanup',
      {
        pattern: '0 0 3 * * *',
        tz: 'Asia/Seoul',
      },
      {
        name: CLEANUP_REPORTS_JOB,
        data: {
          keepDays: 30,
          allowedHour: 3,
          timeZone: 'Asia/Seoul',
        },
        opts: {
          attempts: 3,
          backoff: { type: 'exponential', delay: 5_000 },
          removeOnComplete: 30,
          removeOnFail: 30,
        },
      },
    );
  }
}

0 0 3 * * *는 tz: 'Asia/Seoul'을 기준으로 매일 오전 3시를 뜻하는 6필드 cron 패턴입니다. 호스트 운영체제의 로컬 시간대에 해석을 맡기지 않습니다.

BullMQ 5.19 이상에서는 새 scheduler ID를 처음 등록하면 cron 시각과 관계없이 첫 작업이 즉시 생성됩니다. 같은 ID를 다시 upsert할 때는 즉시 작업이 반복 생성되지 않지만, ID를 바꾸어 배포하면 다시 최초 등록으로 간주됩니다.

따라서 오전 3시 정리만 허용하는 이 예제는 Worker가 payload의 allowedHour와 timeZone을 검사해 최초 즉시 작업을 안전한 no-op으로 끝냅니다. 즉시 실행해도 안전한 업무라면 guard를 생략할 수 있지만, 위험한 정리 작업은 등록 시점에 의존하지 않고 실행 경계에서 한 번 더 막습니다.

Job Scheduler는 마지막 작업이 active로 이동할 때 다음 작업을 만듭니다. Worker 처리량이 부족하면 설정한 간격보다 실제 생성 주기가 느려질 수 있으므로 정확한 초 단위 타이머로 해석하지 않습니다.

등록된 반복 규칙은 scheduler ID로 조회하고 제거합니다.

src/reports/report-scheduler.service.ts (관리 메서드)

getSchedulers() {
  return this.reportsQueue.getJobSchedulers(0, 20, true);
}

removeDailyCleanup() {
  return this.reportsQueue.removeJobScheduler('daily-report-cleanup');
}

Job Scheduler가 만드는 작업은 내부적으로 특별한 ID를 사용하므로 job template에 임의 jobId를 지정하지 않습니다. 업무 중복 방지가 필요하면 Worker의 멱등성 키를 사용합니다.

@nestjs/schedule의 @Cron()은 한 프로세스 안에서 콜백을 실행할 때 간단합니다. 하지만 모든 API 복제본에서 동시에 실행될 수 있고 실행 상태가 Redis에 남지 않습니다. 재시도와 상태 보관이 필요한 업무 작업은 Job Scheduler로 큐에 넣는 편이 안전합니다.

Worker에서 작업 이름 분기

BullMQ의 WorkerHost는 큐마다 process() 하나를 구현합니다. 작업 이름이 여러 개라면 switch로 명시적으로 분기합니다. 아래 코드는 앞 절의 보고서 생성 로직과 이번 절의 정리 작업, 동시성, Worker 이벤트를 합친 최종 형태입니다.

src/reports/reports.processor.ts

import { OnWorkerEvent, Processor, WorkerHost } from '@nestjs/bullmq';
import { Logger } from '@nestjs/common';
import { Job, UnrecoverableError } from 'bullmq';
import { ReportArtifactsService } from './report-artifacts.service';
import {
  CleanupReportsData,
  CleanupReportsResult,
  GenerateReportData,
  GenerateReportResult,
} from './report-job.types';
import {
  CLEANUP_REPORTS_JOB,
  GENERATE_REPORT_JOB,
  REPORT_QUEUE,
} from './reports.constants';

const wait = (milliseconds: number) =>
  new Promise((resolve) => setTimeout(resolve, milliseconds));

@Processor(REPORT_QUEUE, { concurrency: 3 })
export class ReportsProcessor extends WorkerHost {
  private readonly logger = new Logger(ReportsProcessor.name);

  constructor(private readonly artifacts: ReportArtifactsService) {
    super();
  }

  async process(job: Job): Promise<unknown> {
    switch (job.name) {
      case GENERATE_REPORT_JOB:
        return this.generateReport(
          job as Job<GenerateReportData, GenerateReportResult>,
        );
      case CLEANUP_REPORTS_JOB:
        return this.cleanupReports(
          job as Job<CleanupReportsData, CleanupReportsResult>,
        );
      default:
        throw new UnrecoverableError(`unsupported report job: ${job.name}`);
    }
  }

  @OnWorkerEvent('active')
  onActive(job: Job) {
    this.logger.log(`report job active: ${job.id}`);
  }

  @OnWorkerEvent('failed')
  onFailed(job: Job | undefined, error: Error) {
    this.logger.error(`report job failed: ${job?.id ?? 'unknown'} ${error.message}`);
  }

  private async generateReport(
    job: Job<GenerateReportData, GenerateReportResult>,
  ): Promise<GenerateReportResult> {
    if (job.data.rows < 1) {
      throw new UnrecoverableError('rows must be greater than zero');
    }

    const existing = this.artifacts.find(job.data.reportId);
    if (existing) return { ...existing, reused: true };

    if (job.attemptsMade < (job.data.failUntilAttempt ?? 0)) {
      throw new Error(`temporary storage failure on attempt ${job.attemptsMade + 1}`);
    }

    await job.updateProgress({ step: 'load', percent: 20 });
    await wait(400);
    await job.updateProgress({ step: 'render', percent: 70 });
    await wait(400);
    await job.updateProgress({ step: 'upload', percent: 95 });
    await wait(400);
    await job.updateProgress({ step: 'upload', percent: 100 });

    return this.artifacts.saveOnce({
      reportId: job.data.reportId,
      objectKey: `reports/${job.data.reportId}.csv`,
      generatedRows: job.data.rows,
      reused: false,
    });
  }

  private async cleanupReports(
    job: Job<CleanupReportsData, CleanupReportsResult>,
  ): Promise<CleanupReportsResult> {
    const localHour = Number(
      new Intl.DateTimeFormat('en-US', {
        timeZone: job.data.timeZone,
        hour: '2-digit',
        hourCycle: 'h23',
      }).format(new Date()),
    );

    if (localHour !== job.data.allowedHour) {
      return {
        removed: 0,
        skipped: true,
        reason: `outside cleanup hour: ${localHour}`,
      };
    }

    const removed = await this.artifacts.removeOlderThan(job.data.keepDays);
    return { removed, skipped: false };
  }
}

실습의 ReportArtifactsService에는 호출 형태를 확인할 수 있도록 다음 메서드를 추가합니다. 메모리 저장소에는 생성 시각을 저장하지 않았으므로 0을 반환하고, 운영 구현에서는 데이터베이스의 생성 시각을 기준으로 삭제합니다.

src/reports/report-artifacts.service.ts (정리 메서드 추가)

async removeOlderThan(_keepDays: number) {
  return 0;
}

기능 모듈에 ReportSchedulerService도 provider로 등록합니다.

src/reports/reports.module.ts (scheduler 등록 단계)

import { ReportArtifactsService } from './report-artifacts.service';
import { ReportSchedulerService } from './report-scheduler.service';

providers: [
  ReportsService,
  ReportsProcessor,
  ReportArtifactsService,
  ReportSchedulerService,
],

동시성과 처리량 조절

위 Worker처럼 I/O 대기가 많은 작업에 concurrency: 3을 설정하면 한 Worker가 최대 세 작업을 함께 처리할 수 있습니다.

동시성은 무조건 크게 잡지 않습니다.

데이터베이스 connection pool보다 많은 작업을 동시에 실행하면 대기만 늘어납니다.
외부 API의 rate limit을 넘으면 실패와 재시도가 함께 증가합니다.
CPU 집약 작업은 Node.js 이벤트 루프를 막을 수 있으므로 별도 Worker 프로세스나 sandboxed processor를 검토합니다.
여러 인스턴스를 띄우면 총 동시성은 인스턴스 수 × 인스턴스별 concurrency가 됩니다.

이벤트와 큐 상태 관찰하기

@OnWorkerEvent는 위 코드처럼 현재 Worker의 active와 failed 흐름을 관찰할 때 사용합니다. 여러 Worker에서 발생한 큐 전체 이벤트는 QueueEventsListener를 사용합니다.

src/reports/report-queue-events.listener.ts

import {
  OnQueueEvent,
  QueueEventsHost,
  QueueEventsListener,
} from '@nestjs/bullmq';
import { Logger } from '@nestjs/common';
import { REPORT_QUEUE } from './reports.constants';

@QueueEventsListener(REPORT_QUEUE)
export class ReportQueueEventsListener extends QueueEventsHost {
  private readonly logger = new Logger(ReportQueueEventsListener.name);

  @OnQueueEvent('completed')
  onCompleted({ jobId }: { jobId: string }) {
    this.logger.log(`report job completed: ${jobId}`);
  }

  @OnQueueEvent('failed')
  onFailed({ jobId, failedReason }: { jobId: string; failedReason: string }) {
    this.logger.error(`report job failed: ${jobId} ${failedReason}`);
  }

  @OnQueueEvent('deduplicated')
  onDeduplicated({
    jobId,
    deduplicatedJobId,
  }: {
    jobId: string;
    deduplicatedJobId: string;
  }) {
    this.logger.warn(`duplicate ${deduplicatedJobId} kept ${jobId}`);
  }
}

리스너도 기능 모듈의 providers에 등록해야 동작합니다.

src/reports/reports.module.ts (providers 최종)

import { ReportQueueEventsListener } from './report-queue-events.listener';

providers: [
  ReportsService,
  ReportsProcessor,
  ReportArtifactsService,
  ReportSchedulerService,
  ReportQueueEventsListener,
],

큐 길이를 확인하는 운영 API를 추가합니다.

src/reports/reports.service.ts (getQueueHealth 추가)

async getQueueHealth() {
  return this.reportsQueue.getJobCounts(
    'waiting',
    'active',
    'delayed',
    'completed',
    'failed',
  );
}

src/reports/reports.controller.ts (큐 상태 API 추가)

@Get('queue/health')
getQueueHealth() {
  return this.reportsService.getQueueHealth();
}

waiting이 계속 증가하면 Worker 처리량이 유입량보다 낮은 것입니다. failed 증가율, 작업 대기 시간, 처리 시간, stalled 이벤트를 함께 봐야 합니다. completed 누계만으로는 현재 병목을 알 수 없습니다.

안전하게 종료하기

Nest 애플리케이션이 종료 신호를 받아 lifecycle hook을 실행하도록 설정합니다.

src/main.ts (종료 hook 추가)

async function bootstrap() {
  const app = await NestFactory.create(AppModule);
  app.enableShutdownHooks();
  app.useGlobalPipes(new ValidationPipe({ whitelist: true, transform: true }));
  await app.listen(process.env.PORT ?? 3000);
}

Nest의 BullMQ 통합이 관리하는 Worker와 Queue는 애플리케이션 종료 과정에서 연결을 정리합니다. raw BullMQ Worker를 직접 만든 경우에는 종료 시 await worker.close()를 호출해야 합니다. close()는 새 작업을 가져오지 않고 현재 작업이 끝나기를 기다리며 자체 timeout이 없으므로 작업 코드에도 종료 가능한 상한을 둡니다.

강제 종료되면 active 작업은 stalled 감지를 거쳐 다른 Worker가 다시 가져갈 수 있습니다. 이때도 결과가 안전하려면 3절의 멱등성 규칙이 필요합니다. 현재 BullMQ는 stalled 작업 감지를 자체적으로 처리하므로 별도 보조 클래스를 만들지 않습니다.

운영 점검표

관찰 항목	정상 신호	위험 신호
waiting	부하가 줄면 다시 감소	지속적으로 우상향
active	설정한 총 동시성 안에서 변화	오래 같은 작업이 머묾
delayed	예약·백오프 작업 수와 비슷	과거 시각 작업이 계속 남음
failed	원인별 알림과 복구 절차 존재	같은 원인이 빠르게 반복
stalled	드물게 발생하고 복구됨	CPU 정지나 강제 종료로 반복
종료 시간	배포 유예 시간 안에 완료	`close()`가 끝없이 대기

6장에서는 긴 HTTP 요청을 작업 큐로 분리하고, Producer와 Worker의 계약, 상태 조회, 재시도와 멱등성, 지연·반복 실행, 동시성, 관측, 종료까지 연결했습니다.

작업 큐는 마이크로서비스의 전제 조건은 아니지만, 실행 책임을 프로세스 밖의 영속 상태로 옮기는 첫 경험을 제공합니다. 다음 장에서는 이 경계를 서비스 간 통신으로 확장하고 gRPC와 Kafka의 책임을 비교합니다.