재시도, 중복 방지와 실패 복구

작업 큐를 운영하면 외부 저장소의 순간 장애, 네트워크 단절, 잘못된 입력처럼 성격이 다른 실패를 만나게 됩니다. 모든 실패를 같은 방식으로 재시도하면 복구 가능성이 없는 작업이 자원을 계속 차지하고, 중복 실행이 보고서를 여러 번 저장할 수 있습니다.

이번 절에서는 앞 절의 보고서 작업에 자동 재시도, 지수 백오프, 중복 제거, 멱등한 결과 저장, 수동 복구를 차례로 추가합니다.

재시도할 실패와 즉시 끝낼 실패

먼저 실패를 두 종류로 나눕니다.

일시적 실패: 외부 저장소의 503 응답, 잠깐의 연결 끊김처럼 시간이 지나면 성공할 가능성이 있습니다.
영구적 실패: 존재하지 않는 보고서 종류, 허용 범위를 벗어난 입력처럼 같은 데이터로 다시 실행해도 실패합니다.

BullMQ는 Worker가 Error를 던지면 attempts 횟수 안에서 작업을 다시 실행합니다. UnrecoverableError를 던지면 남은 시도 횟수와 관계없이 즉시 failed 상태로 보냅니다.

실패 실험을 위해 작업 데이터에 선택 속성을 추가합니다.

src/reports/report-job.types.ts

export interface GenerateReportData {
  reportId: string;
  requestedBy: string;
  rows: number;
  failUntilAttempt?: number;
}

export interface GenerateReportResult {
  reportId: string;
  objectKey: string;
  generatedRows: number;
  reused: boolean;
}

export interface ReportProgress {
  step: 'load' | 'render' | 'upload';
  percent: number;
}

DTO에도 실험 필드를 추가합니다. 운영 API에서는 이런 필드를 외부에 노출하지 말고 장애 주입 환경에서만 사용합니다.

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

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

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

  @IsString()
  requestedBy: string;

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

  @IsOptional()
  @IsInt()
  @Min(0)
  @Max(5)
  failUntilAttempt?: number;
}

Producer의 재시도 옵션

ReportsService.enqueue()에서 작업 옵션을 확장합니다.

src/reports/reports.service.ts (enqueue 일부)

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

attempts: 4는 최초 실행을 포함해 최대 네 번 시도한다는 뜻입니다. 지수 백오프는 실패가 반복될수록 대기 시간을 늘리고, jitter는 여러 작업이 같은 순간에 다시 몰리는 현상을 줄입니다.

재시도 간격을 너무 짧게 잡으면 장애 중인 외부 시스템을 더 압박합니다. 너무 길게 잡으면 사용자가 복구를 오래 기다립니다. 외부 시스템의 복구 시간과 업무 마감 시간을 기준으로 정합니다.

중복 제거와 멱등성은 다른 문제다

deduplication.id가 같은 작업이 아직 완료되거나 실패하지 않았다면 BullMQ는 뒤의 추가 요청을 무시합니다. 이 기능은 짧은 시간에 같은 보고서 요청이 여러 번 들어와 큐가 불어나는 것을 막습니다.

하지만 큐의 중복 제거만으로 결과가 한 번만 저장된다고 보장할 수는 없습니다. Worker가 파일을 저장한 직후 완료 상태를 기록하기 전에 종료되면 같은 작업이 다시 실행될 수 있습니다. 따라서 부수 효과도 멱등하게 만들어야 합니다.

실습에서는 동작을 눈으로 확인하기 위해 메모리 저장소를 사용합니다. 운영 환경에서는 이 역할을 데이터베이스의 고유 키, 객체 저장소의 조건부 쓰기, 또는 원자적 upsert로 바꿔야 합니다.

src/reports/report-artifacts.service.ts

import { Injectable } from '@nestjs/common';
import { GenerateReportResult } from './report-job.types';

@Injectable()
export class ReportArtifactsService {
  private readonly artifacts = new Map<string, GenerateReportResult>();

  find(reportId: string) {
    return this.artifacts.get(reportId);
  }

  saveOnce(result: GenerateReportResult) {
    const existing = this.artifacts.get(result.reportId);
    if (existing) return existing;

    this.artifacts.set(result.reportId, result);
    return result;
  }
}

메모리 Map은 프로세스를 재시작하거나 Worker를 여러 대 띄우면 공유되지 않습니다. 여기서는 결과 키 확인 → 없을 때만 저장이라는 코드 경계를 보여주기 위한 구현입니다.

Worker에 저장소를 주입하고 실패 분기를 추가합니다.

src/reports/reports.processor.ts

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

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

@Processor(REPORT_QUEUE)
export class ReportsProcessor extends WorkerHost {
  constructor(private readonly artifacts: ReportArtifactsService) {
    super();
  }

  async process(
    job: Job<GenerateReportData, GenerateReportResult, string>,
  ): Promise<GenerateReportResult> {
    if (job.name !== GENERATE_REPORT_JOB) {
      throw new UnrecoverableError(`unsupported report job: ${job.name}`);
    }

    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,
    });
  }
}

기능 모듈의 provider에도 저장소를 추가합니다.

src/reports/reports.module.ts (providers 일부)

providers: [
  ReportsService,
  ReportsProcessor,
  ReportArtifactsService,
],

운영 저장소의 원자성

실제 저장소에서는 find()와 save()를 서로 떨어진 두 쿼리로 구현하면 두 Worker가 동시에 통과할 수 있습니다. 다음 중 하나로 원자성을 확보합니다.

reportId에 고유 인덱스를 두고 insert 충돌 시 기존 결과를 조회합니다.
데이터베이스의 INSERT ... ON CONFLICT 또는 트랜잭션을 사용합니다.
객체 저장소에서 같은 key에 대한 조건부 생성 기능을 사용합니다.
외부 API가 idempotency key를 지원한다면 reportId를 함께 전달합니다.

자동 재시도 관찰하기

첫 두 번은 실패하고 세 번째에 성공하는 작업을 제출합니다.

curl -X POST http://localhost:3000/reports \
  -H "Content-Type: application/json" \
  -d '{"reportId":"retry-demo","requestedBy":"user-42","rows":1000,"failUntilAttempt":2}'

상태를 반복 조회하면 delayed와 active를 거쳐 completed가 되고 attemptsMade가 증가합니다.

curl http://localhost:3000/reports/jobs/2

다음도 확인합니다.

같은 reportId를 첫 작업이 끝나기 전에 다시 제출하면 큐의 중복 제거가 작동합니다.
같은 보고서가 이미 저장된 뒤 다시 실행되면 결과의 reused가 true가 됩니다.
failUntilAttempt를 5로 보내면 최대 네 번을 모두 사용한 뒤 failed가 됩니다.

실패 작업을 수동 복구하기

자동 재시도를 모두 사용한 작업은 원인을 고친 뒤 운영자가 다시 실행할 수 있어야 합니다. 상태를 확인하고 failed 작업만 재시도하는 메서드를 추가합니다.

reports.service.ts에 이미 있는 Nest import에 BadRequestException을 합치고 메서드를 추가합니다. 같은 모듈에서 import 문을 하나 더 만들지 않습니다.

src/reports/reports.service.ts (상단 import 교체와 retry 추가)

import {
  BadRequestException,
  Injectable,
  NotFoundException,
} from '@nestjs/common';

async retry(jobId: string) {
  const job = await this.reportsQueue.getJob(jobId);
  if (!job) {
    throw new NotFoundException(`report job ${jobId} not found`);
  }

  const state = await job.getState();
  if (state !== 'failed') {
    throw new BadRequestException(`only failed jobs can be retried: ${state}`);
  }

  await job.retry('failed', {
    resetAttemptsMade: true,
    resetAttemptsStarted: true,
  });

  return { jobId: job.id, state: await job.getState() };
}

Controller에 복구 엔드포인트를 연결합니다.

src/reports/reports.controller.ts (retry 추가)

@Post('jobs/:jobId/retry')
@HttpCode(HttpStatus.ACCEPTED)
retry(@Param('jobId') jobId: string) {
  return this.reportsService.retry(jobId);
}

수동 재시도는 실패 원인이 해결됐는지 확인한 뒤 실행합니다. 잘못된 데이터를 그대로 다시 넣는 버튼이 되어서는 안 됩니다. 누가, 언제, 어떤 이유로 재시도했는지도 감사 로그에 남깁니다.

실패 기록을 너무 빨리 지우지 않는다

removeOnFail: true로 모든 실패를 즉시 삭제하면 원인과 payload를 조사할 수 없습니다. 반대로 무제한 보관하면 Redis 메모리가 계속 증가합니다. 개수 또는 기간 기준으로 일정량을 남기고, 장기 감사가 필요하면 별도 저장소로 내보냅니다.

정책	목적	잘못 설정했을 때
`attempts`	일시 장애 복구	영구 실패를 반복 실행
`backoff`	재시도 부하 분산	장애 중인 시스템을 재압박
`deduplication.id`	대기 중 중복 작업 억제	서로 다른 요청을 같은 ID로 묶음
업무 고유 키	부수 효과 중복 방지	같은 파일·결제·메일이 여러 번 생성
실패 보관	조사와 수동 복구	너무 적으면 증거 손실, 너무 많으면 메모리 증가

복구 가능한 시스템은 단순히 많이 재시도하는 시스템이 아닙니다. 실패의 성격을 구분하고, 다시 실행해도 결과가 망가지지 않게 만들며, 자동 복구가 끝난 뒤의 운영 절차까지 갖춘 시스템입니다.

다음 절에서는 지연 작업과 Job Scheduler를 추가하고, Worker 동시성·이벤트·메트릭·종료 절차를 운영 관점에서 연결합니다.