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

요청을 보내기 전에 이 장의 HTTP 경로 기준을 확인하세요. 앞 장의 api prefix와 URI 버전 1을 유지한다면 본문의 /reports 대신 /api/v1/reports로 요청하고, 응답에 담는 statusUrl도 /api/v1/reports/jobs/:id 형식으로 만듭니다.

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

이 글에서는 앞 글의 보고서 작업에 자동 재시도, 지수 백오프(exponential backoff), 범위가 분명한 중복 제거, 멱등한 결과 저장, 수동 복구를 차례로 추가합니다. 아래에서 쓰는 BullMQ 옵션과 Job.retry()의 서명은 BullMQ v6.5.3 공식 소스를 기준으로 확인했습니다.

재시도할 실패와 바로 끝낼 실패

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

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

BullMQ Worker가 일반 Error를 던지면 남은 attempts가 있을 때 작업을 다시 실행합니다. 반면 UnrecoverableError를 던지면 남은 실행 기회와 상관없이 작업을 바로 최종 failed 상태로 보냅니다.

실패를 실험하기 위해 앞 글의 작업 데이터에 선택 속성을 하나 추가합니다. tenantId와 requestedBy는 계속 서버가 확인한 인증 주체에서 가져옵니다.

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

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

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

export interface ReportPrincipal {
  tenantId: string;
  userId: string;
}

DTO에는 장애를 일부러 일으키는 실험용 필드만 추가합니다. 운영 API에서는 이 필드를 외부에 노출하지 않고 테스트 환경에서만 켭니다.

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

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

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

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

failUntilAttempt는 @IsOptional()이므로 보내지 않으면 검사를 건너뛰고, 보낼 때는 0부터 5까지의 정수만 허용합니다. 뒤의 Worker는 이 값만큼의 실행을 일부러 실패시킵니다. 예를 들어 2를 보내면 처음 두 번의 실행이 실패하고 세 번째 실행에서 보고서를 만들며, 뒤에서 정할 attempts: 4보다 크거나 같은 값을 보내면 모든 실행이 실패합니다.

같은 업무를 나타내는 키 만들기

reportId 하나만 중복 제거 키로 쓰면, tenant가 다르거나 결과를 바꾸는 입력이 다른 요청까지 하나로 합쳐질 수 있습니다. 이 예제는 작업 종류와 버전, 인증된 tenant와 사용자, 보고서 ID, 행 수를 모두 직렬화한 뒤 해시합니다. 실제 업무에서는 결과를 바꾸는 모든 입력을 포함하되, 비밀 값 자체는 payload나 키에 넣지 않습니다.

import { createHash } from 'node:crypto';
import type { GenerateReportData } from './report-job.types';

export function reportOperationKey(
  data: Pick<
    GenerateReportData,
    'tenantId' | 'requestedBy' | 'reportId' | 'rows'
  >,
) {
  return createHash('sha256')
    .update(
      JSON.stringify([
        'report.generate.v1',
        data.tenantId,
        data.requestedBy,
        data.reportId,
        data.rows,
      ]),
    )
    .digest('hex');
}

해시는 같은 업무의 범위를 정하는 수단일 뿐, 권한 검사를 대신하지 않습니다.

Producer의 재시도 옵션

앞 글의 ReportsService.enqueue()에서 인증 주체로 작업 데이터를 만들고, 옵션을 다음과 같이 확장합니다. 앞 글과 비교하면 작업 데이터를 변수 data로 먼저 만들어 failUntilAttempt까지 담고, 같은 데이터로 operationKey를 계산합니다. 옵션에서는 attempts를 4로 늘리고 jitter와 deduplication을 더했으며, 각 값의 뜻은 코드 아래에서 설명합니다. 발췌에 없는 import로는 import { reportOperationKey } from './report-operation-key';를 service 파일 위쪽에 추가합니다.

const data: GenerateReportData = {
  reportId: dto.reportId,
  tenantId: principal.tenantId,
  requestedBy: principal.userId,
  rows: dto.rows,
  failUntilAttempt: dto.failUntilAttempt,
};

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

return {
  jobId: job.id,
  accepted: true,
  statusUrl: '/reports/jobs/' + job.id,
};

attempts: 4는 최초 실행을 포함해 최대 네 번 실행할 수 있다는 뜻입니다.

delay: 1_000인 지수 백오프의 최대 대기 시간은 첫 실패 뒤 1초, 두 번째 실패 뒤 2초, 세 번째 실패 뒤 4초입니다. BullMQ 소스는 2^(attemptsMade - 1) × delay로 최대치를 계산하고, jitter: 0.5가 있으면 그 최대치의 50%에서 100% 사이에서 실제 대기를 무작위로 고릅니다. 이렇게 하면 여러 작업이 같은 순간에 한꺼번에 다시 몰리는 현상을 줄일 수 있습니다.

재시도 간격은 외부 시스템이 복구되는 데 걸리는 시간과 업무 마감 시간을 기준으로 정합니다. 너무 짧으면 장애 중인 시스템을 다시 압박하고, 너무 길면 사용자가 복구를 기다리는 시간이 늘어납니다.

위 그림은 Worker의 실행 결과가 세 갈래로 나뉘는 모습을 보여 줍니다. 정상 반환만 completed가 되고, UnrecoverableError로 분류한 영구 실패는 곧바로 최종 failed로 갑니다. 일시적인 Error는 최초 실행을 포함한 네 번의 실행 기회가 남았는지 확인해, 남았다면 최대 1·2·4초의 backoff 뒤 다시 실행하고 모두 썼다면 failed로 합류합니다. 그림 아래쪽 단계 설명은 실행 기회마다 실제 대기 시간이 jitter 때문에 어느 범위에 들어가는지 정리한 것입니다.

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

위 설정은 TTL이 없는 BullMQ의 Simple 모드 중복 제거(deduplication)입니다. 같은 deduplication.id를 가진 작업이 아직 completed나 failed가 아니라면, 뒤에 들어온 추가 요청은 새 작업으로 저장되지 않고 deduplicated 이벤트만 발생합니다. 이 기능이 하는 일은 큐 안에서 아직 끝나지 않은 같은 작업이 다시 등록되는 것을 막는 데까지입니다.

• 원래 작업이 completed나 failed가 되면 Simple 모드의 키도 사라지므로, 같은 업무 요청을 다시 추가할 수 있습니다.
• Worker가 lock을 갱신하지 못해 stalled가 되면 작업은 waiting으로 돌아가 다시 실행될 수 있습니다.
• 파일을 저장한 직후 Worker가 종료되면, BullMQ가 완료를 기록하지 못한 채 같은 부수 효과를 다시 수행할 수 있습니다.
• removeOnComplete와 removeOnFail로 작업이 지워지면 큐 기록은 상태 조회용으로도, 영구 중복 방지 원장으로도 쓸 수 없습니다.

따라서 큐의 중복 제거는 정확히 한 번(exactly-once) 실행이나 결과의 한 번 저장을 보장하지 않습니다. 최종 부수 효과 자체를 업무 키로 멱등하게 만들어야 합니다.

실습에서는 동작을 눈으로 확인하기 위해 메모리 저장소를 사용합니다.

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

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

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

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

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

메모리 Map은 프로세스를 다시 시작하거나 Worker를 여러 대 띄우면 공유되지 않습니다. 한 프로세스 안에서 개념을 보여 주는 예제일 뿐이며, 운영 환경에서는 데이터베이스의 고유 제약과 원자적 upsert, 객체 저장소의 조건부 생성, 외부 API의 idempotency key로 바꿉니다.

Worker는 같은 operation key로 기존 결과를 먼저 조회하고, 정상 반환할 결과도 같은 키로 저장합니다.

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 { reportOperationKey } from './report-operation-key';
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,
      typeof GENERATE_REPORT_JOB
    >,
  ): 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 operationKey = reportOperationKey(job.data);
    const existing = this.artifacts.find(operationKey);
    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(operationKey, {
      operationKey,
      reportId: job.data.reportId,
      objectKey:
        'reports/' + job.data.tenantId + '/' + operationKey + '.csv',
      generatedRows: job.data.rows,
      reused: false,
    });
  }
}

percent: 100은 Worker가 기록한 진행 값일 뿐 완료가 아닙니다. 그 뒤의 조건부 저장이 성공하고 process()가 정상 반환해야 BullMQ가 completed와 returnvalue를 기록합니다.

기능 모듈의 provider에도 저장소를 추가합니다. module 파일 위쪽에는 import { ReportArtifactsService } from './report-artifacts.service';를 더합니다.

providers: [
  ReportsService,
  ReportsProcessor,
  ReportArtifactsService,
],

ReportsProcessor가 생성자로 ReportArtifactsService를 주입받으므로, 이 등록이 빠지면 Nest가 Worker를 만들 때 의존성을 찾지 못해 애플리케이션이 시작하지 않습니다.

운영 저장소의 원자성

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

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

위 그림의 윗줄은 업무 키가 쓰이는 세 곳이 서로 다른 것을 보장한다는 점을 보여 줍니다. HTTP 쪽은 tenant·사용자·입력으로 업무의 범위를 정하고, 큐의 Simple 중복 제거는 끝나지 않은 작업의 추가 등록만 막으며, 결과 저장소의 원자적 저장이 재실행되어도 같은 결과를 남기게 합니다. 아랫줄은 실패한 작업에 따로 적용하는 수동 복구 절차로, 권한·경쟁·결과를 확인한 뒤 외부 감사 기록을 먼저 남기고 마지막에 Job.retry()를 호출하는 순서입니다.

자동 재시도 확인하기

처음 두 번은 실패하고 세 번째에 성공하는 작업을 제출합니다. requestedBy와 tenantId는 access token으로 확인한 주체에서 채워집니다.

curl -i -X POST http://localhost:3000/reports \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <access-token>" \
  -d '{"reportId":"retry-demo","rows":1000,"failUntilAttempt":2}'

응답의 jobId나 statusUrl로 상태를 반복해서 조회합니다. 자동 증가하는 ID를 추측해 /jobs/2처럼 고정해 두지 않습니다.

JOB_ID='POST 응답의 jobId'
curl -H "Authorization: Bearer <access-token>" \
  "http://localhost:3000/reports/jobs/$JOB_ID"

조회 시점에 따라 delayed, waiting, active 가운데 일부를 보지 못하고 지나칠 수 있으며, 최종 성공은 state === 'completed'와 result로 확인합니다. 다음 동작도 함께 확인해 봅니다.

1. 같은 인증 주체가 같은 reportId와 rows를 첫 작업이 끝나기 전에 다시 제출하면, Simple 중복 제거가 뒤의 추가 요청을 막습니다.
2. rows처럼 결과를 바꾸는 입력이 달라지면 operation key도 달라지므로 별도 작업이 됩니다.
3. 같은 결과가 이미 저장된 뒤 새 작업을 실행하면, 멱등 저장소가 기존 결과를 돌려주고 reused가 true가 됩니다.
4. failUntilAttempt를 5로 보내면 네 번의 실행 기회를 모두 쓴 뒤 failed가 됩니다.

실패 작업을 수동으로 복구하기

자동 재시도를 모두 쓴 작업은 원인을 고친 뒤 권한이 있는 운영자가 다시 실행할 수 있어야 합니다. 수동 복구는 다음 순서를 지킵니다.

1. 인증과 재시도 권한, 작업의 소유 범위를 확인합니다. 이 예제는 tenantId와 requestedBy가 모두 일치해야 하며, 권한이 없으면 존재 여부를 숨기고 404로 처리합니다.
2. 같은 operation key의 결과나 새로 진행 중인 작업이 없는지 확인하고, 운영 저장소의 lock이나 원자적 상태 전이로 복구 경쟁을 막습니다.
3. payload의 비밀이 아닌 식별자, failedReason, stacktrace, attemptsMade, attemptsStarted, 실행자와 이유를 외부 감사 저장소에 먼저 기록합니다.
4. 장애 원인이 해결된 뒤에만 Job.retry()를 호출하고, 바뀐 상태를 상태 API로 다시 확인합니다.

BullMQ v6.5.3의 Job.retry()는 작업을 다시 대기열로 옮기면서 failedReason, finishedOn, processedOn, returnvalue를 지웁니다. 그래서 실패 당시의 기록은 호출하기 전에 남겨 두어야 합니다.

재시도 이유를 검증할 DTO를 추가합니다.

import { IsString, MinLength } from 'class-validator';

export class RetryReportJobDto {
  @IsString()
  @MinLength(10)
  reason: string;
}

아래 recoveryAudit.append()는 데이터베이스나 변경할 수 없는 감사 저장소에 기록을 덧붙이는 애플리케이션 서비스이고, artifacts는 앞에서 만든 결과 저장소라고 가정합니다. 기존 Nest import에는 BadRequestException, ConflictException, NotFoundException을 합칩니다. 두 객체는 ReportsService가 생성자로 주입받아야 하므로, 기존 reportsQueue 매개변수 뒤에 private readonly artifacts: ReportArtifactsService와 private readonly recoveryAudit: ReportRecoveryAuditService를 추가하고, service 파일 위쪽에 import { ReportArtifactsService } from './report-artifacts.service';를 더합니다. ReportArtifactsService는 앞에서 providers에 등록했습니다. 반면 ReportRecoveryAuditService는 append() 메서드를 가진 감사 서비스라고 가정한 이름이며 이 글에서 구현하지 않으므로, 직접 만들어 import하고 같은 모듈의 providers에 등록해야 컴파일과 주입이 됩니다.

예제의 조회 조건은 판단 순서를 보여 주기 위한 것입니다. 운영 환경에서는 producer와 수동 재시도가 함께 참여하는 operation key 단위의 lock이나 원자적 복구 상태 전이 안에서, 기존 결과 확인, 진행 중인 작업 확인, 감사 기록, 재시도를 차례로 수행해야 합니다.

async retry(
  jobId: string,
  reason: string,
  principal: ReportPrincipal,
) {
  const job = await this.reportsQueue.getJob(jobId);
  if (
    !job ||
    job.data.tenantId !== principal.tenantId ||
    job.data.requestedBy !== principal.userId
  ) {
    throw new NotFoundException('report job not found');
  }

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

  const operationKey = reportOperationKey(job.data);
  const liveJobId =
    await this.reportsQueue.getDeduplicationJobId(operationKey);

  if (liveJobId && liveJobId !== job.id) {
    throw new ConflictException(
      'another report job is already in progress',
    );
  }

  if (this.artifacts.find(operationKey)) {
    throw new ConflictException('report result already exists');
  }

  await this.recoveryAudit.append({
    operationKey,
    jobId: job.id,
    tenantId: job.data.tenantId,
    requestedBy: job.data.requestedBy,
    reportId: job.data.reportId,
    rows: job.data.rows,
    failedReason: job.failedReason,
    stacktrace: job.stacktrace ?? [],
    attemptsMade: job.attemptsMade,
    attemptsStarted: job.attemptsStarted,
    retriedBy: principal.userId,
    reason,
    recordedAt: new Date().toISOString(),
  });

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

  return {
    jobId: job.id,
    accepted: true,
    statusUrl: '/reports/jobs/' + job.id,
  };
}

resetAttemptsMade: true는 자동 재시도 판단에 쓰는 attemptsMade를 0으로 되돌려, 작업에 attempts: 4의 실행 기회를 처음부터 다시 줍니다. resetAttemptsStarted: true는 작업이 active로 옮겨진 횟수인 attemptsStarted를 따로 0으로 되돌립니다. 두 카운터를 되돌리는 일은 단순한 정리가 아니라, 감사 기록과 실행 기회를 바꾸는 운영 결정입니다.

모두 쓴 attemptsMade를 되돌리지 않으면, 다시 실행했을 때 오류가 나는 순간 추가 자동 재시도 없이 최종 실패합니다. 반대로 원인을 고치지 않은 채 되돌리면 같은 장애를 최대 네 번 더 겪게 됩니다.

controller에서는 인증 guard에 더해 재시도 전용 권한 guard를 적용합니다. 아래의 ReportRetryGuard는 운영자의 재시도 권한을 확인하는 guard라고 가정하며, 이 글에서 구현하지 않습니다. UseGuards와 RetryReportJobDto는 기존 controller import에 합칩니다.

@UseGuards(ReportRetryGuard)
@Post('jobs/:jobId/retry')
@HttpCode(HttpStatus.ACCEPTED)
retry(
  @Param('jobId') jobId: string,
  @Body() dto: RetryReportJobDto,
  @Req() request: AuthenticatedRequest,
) {
  return this.reportsService.retry(
    jobId,
    dto.reason,
    request.user,
  );
}

Job.retry()가 반환된 직후 Worker가 작업을 가져갈 수 있으므로 응답은 waiting을 약속하지 않습니다. 202 Accepted와 같은 jobId, statusUrl만 반환하고 실제 상태는 다시 조회합니다.

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

removeOnFail: true로 모든 실패 작업을 끝나자마자 지우면 getJob()과 수동 재시도를 할 수 없고, 원인과 payload도 조사할 수 없습니다.

개수나 기간으로 지정한 자동 제거는 BullMQ 문서의 설명대로 지연 방식(lazy)으로 동작합니다. 다른 새 작업이 completed나 failed가 될 때에야 제거가 일어나므로, removeOnComplete: 100과 removeOnFail: 100은 정확한 보존 시간이 아니라 Redis 사용량을 제한하는 정책입니다.

지워진 작업은 상태·결과·오류를 더 이상 조회하거나 재시도할 수 없습니다. 오래 남겨야 할 감사 기록과 복구용 기록은 보존 기간을 정한 별도 저장소로 내보냅니다.

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

정리

• 일시적 실패는 일반 Error로 던져 attempts와 backoff에 맡기고, 영구적 실패는 UnrecoverableError로 던져 바로 끝냅니다.
• exponential backoff의 최대 대기는 2^(attemptsMade - 1) × delay이고, jitter는 그 최대치 아래쪽 일부 범위에서 실제 대기를 무작위로 고릅니다.
• Simple 모드의 중복 제거는 끝나지 않은 같은 작업의 추가 등록만 막을 뿐, 정확히 한 번 실행을 보장하지 않습니다.
• 부수 효과의 중복은 업무 키를 쓰는 원자적 저장으로 막고, 메모리 Map은 한 프로세스 안의 실습용으로만 씁니다.
• 수동 재시도 전에는 권한과 경쟁을 확인하고 감사 기록을 먼저 남깁니다. Job.retry()가 실패 정보를 지우기 때문입니다.

연습 문제

1. 몇 번째 실행이 성공할까

기존 결과가 없는 같은 프로세스에서 attempts: 4, failUntilAttempt: 3으로 실행합니다. 일반 오류만 발생하고 stalled 재실행은 없다고 가정합니다. 몇 번째 실행이 성공하며, 그 전에 대기하는 구간은 몇 번이고 각각 어느 범위일까요? 완료 뒤 같은 업무를 다시 등록할 수 있다면 새 파일을 반드시 만들어야 할까요?

2. 원인을 고치지 않고 재시도하면

failUntilAttempt: 5로 제출한 작업이 네 번 모두 실패해 failed가 되었습니다. 운영자가 원인을 확인하지 않고 이 글의 retry()를 호출했습니다. failUntilAttempt 값은 그대로입니다. 작업은 최종적으로 어떻게 되고, 처음 제출한 뒤부터 Worker는 모두 몇 번 실행될까요?

3. 지원하지 않는 작업 이름

누군가 같은 큐에 이름이 report.export인 작업을 attempts: 4로 넣었습니다. 이 글의 Worker는 이 작업을 어떻게 처리하고, 상태 조회에서 state와 attemptsMade는 어떤 값이 될까요? Worker가 UnrecoverableError 대신 일반 Error를 던졌다면 무엇이 달라질까요?

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