배치 작업을 재실행 가능하게 만드는 체크포인트

배치 작업을 재실행 가능하게 만드는 체크포인트

한눈에 보기

체크포인트는 단순히 lastId를 파일에 저장하는 기능이 아니다. 어디까지의 처리 결과가 확정되었는가를 나타내는 내구성 있는 계약이다. 안정적인 정렬 키로 청크를 읽고, 처리 결과와 체크포인트를 같은 트랜잭션으로 커밋하며, 외부 부작용에는 별도의 멱등성 키를 사용해야 실패 직전 청크를 다시 실행해도 안전하다.

목차

처음부터 다시 실행하는 비용

백만 건의 레코드를 읽어 검색 인덱스를 다시 만드는 작업이 있다고 하자.

const records = await repository.findAll();

for (const record of records) {
  await searchIndex.upsert(toDocument(record));
}

90만 건을 처리한 뒤 외부 검색 서비스의 일시 장애로 실패하면 다음 실행은 다시 1번부터 시작한다.

1차 실행: 1 ~ 900,000 처리 후 실패
2차 실행: 1 ~ 1,000,000 다시 처리
총 처리: 1,900,000건

작업이 단순 DB 갱신이면 시간이 더 드는 정도일 수 있다. 외부 API 호출 비용이 있거나 사용자 알림을 보내면 중복 부작용이 된다. 반대로 실패 지점부터 정확히 이어가려고 메모리의 반복문 인덱스만 저장하면 프로세스 재시작 때 사라진다.

체크포인트는 마지막으로 확정된 위치를 내구성 있게 저장한다.

checkpoint = record_id 900000
재시작 → record_id > 900000부터 조회

하지만 이 한 줄에는 어려운 질문이 숨어 있다.

체크포인트의 의미

“마지막으로 읽은 행”이 아니라 “이 위치까지의 결과가 재실행 없이도 유효하다고 확정되었다”는 뜻이어야 한다.

체크포인트가 보장해야 하는 것

안전한 체크포인트는 적어도 다음 정보를 가진다.

type BatchCheckpoint = Readonly<{
  jobName: string;
  logicalRunId: string;
  partitionKey: string;
  cursor: {
    createdAt: string;
    id: string;
  } | null;
  snapshotBoundary: {
    maxCreatedAt: string;
    maxId: string;
  };
  processorVersion: number;
  processedCount: number;
  updatedAt: string;
}>;
필드 의미
logicalRunId 어느 논리 실행의 상태인가
partitionKey 테넌트·샤드 등 독립 처리 범위
cursor 마지막으로 확정한 정렬 위치
snapshotBoundary 이번 실행에 포함할 상한
processorVersion 어떤 처리 규칙으로 만든 상태인가
processedCount 진전과 검증을 위한 누적 수
updatedAt 정체된 작업 감지

모든 작업에 이 필드가 필요한 것은 아니다. 핵심은 체크포인트를 숫자 하나가 아니라 실행 범위와 처리 의미를 복원할 수 있는 상태로 보는 것이다.

처리 흐름은 다음과 같다.

flowchart LR
    A[체크포인트 읽기] --> B[다음 청크 조회]
    B --> C[청크 처리]
    C --> D{처리 성공?}
    D -->|아니오| E[트랜잭션 롤백]
    E --> A
    D -->|예| F[결과 + 체크포인트 커밋]
    F --> G{상한 도달?}
    G -->|아니오| B
    G -->|예| H[실행 완료]

안정적인 순서와 커서를 선택한다

체크포인트 기반 배치는 순서가 결정적이어야 한다. SQL에서 ORDER BY가 없으면 반환 순서는 보장되지 않는다.

-- 순서를 보장하지 않는 쿼리
SELECT id, created_at, payload
FROM source_records
LIMIT 500;

기본 키가 단조 증가하고 작업 범위에 적합하다면 가장 단순하다.

SELECT id, created_at, payload
FROM source_records
WHERE id > $1
  AND id <= $2
ORDER BY id ASC
LIMIT 500;

$1은 마지막 체크포인트, $2는 실행 시작 시 확정한 상한이다.

정렬 키가 고유하지 않을 때

created_at만 사용하면 같은 시각의 행이 많을 수 있다.

SELECT id, created_at, payload
FROM source_records
WHERE created_at > $1
ORDER BY created_at ASC
LIMIT 500;

첫 청크의 마지막 시각이 10:00:00이고 같은 시각의 행이 700개라면 500개 뒤 체크포인트를 10:00:00으로 저장한다. 다음 쿼리의 > 조건은 남은 200개를 건너뛴다.

고유한 tie-breaker를 함께 사용한다.

SELECT id, created_at, payload
FROM source_records
WHERE (created_at, id) > ($1, $2)
  AND (created_at, id) <= ($3, $4)
ORDER BY created_at ASC, id ASC
LIMIT 500;

체크포인트도 두 값을 저장한다.

{
  "cursor": {
    "createdAt": "2025-08-08T10:00:00.000Z",
    "id": "record-example-500"
  }
}

정렬 키는 처리 중 바뀌지 않아야 한다

updated_at을 커서로 사용하면서 배치가 원본 행을 갱신하면 방금 처리한 행이 뒤로 이동해 다시 나타날 수 있다.

UPDATE source_records
SET
  normalized_value = $1,
  updated_at = now()
WHERE id = $2;

읽기 기준 updated_at=10:00
처리 후 updated_at=10:05로 이동
뒤 청크에서 같은 행을 다시 읽음

가능하면 불변인 생성 시각과 ID, 또는 전용 순번을 사용한다. 변경 가능한 값을 써야 한다면 실행 시작 시 스냅샷 상한을 고정하고 처리 대상 ID를 별도 테이블에 materialize하는 방법을 고려한다.

처리 결과와 체크포인트를 함께 커밋한다

DB 안에서 끝나는 처리라면 결과 갱신과 체크포인트 이동을 같은 트랜잭션에 넣는다.

CREATE TABLE batch_checkpoints (
  job_name          text NOT NULL,
  logical_run_id    text NOT NULL,
  partition_key     text NOT NULL,
  cursor_created_at timestamptz,
  cursor_id         text,
  boundary_created_at timestamptz NOT NULL,
  boundary_id       text NOT NULL,
  processor_version integer NOT NULL,
  processed_count   bigint NOT NULL DEFAULT 0,
  updated_at        timestamptz NOT NULL,

  PRIMARY KEY (
    job_name,
    logical_run_id,
    partition_key
  )
);

한 청크를 트랜잭션으로 처리한다.

async function processDatabaseChunk(
  checkpoint: BatchCheckpoint,
  rows: readonly SourceRecord[],
): Promise<BatchCheckpoint> {
  if (rows.length === 0) {
    return checkpoint;
  }

  return database.transaction(async (tx) => {
    for (const row of rows) {
      await tx.query(
        `
          INSERT INTO normalized_records (
            source_id,
            normalized_value,
            processor_version
          )
          VALUES ($1, $2, $3)
          ON CONFLICT (source_id)
          DO UPDATE SET
            normalized_value = EXCLUDED.normalized_value,
            processor_version = EXCLUDED.processor_version
        `,
        [
          row.id,
          normalize(row.payload),
          checkpoint.processorVersion,
        ],
      );
    }

    const last = rows.at(-1)!;

    const result = await tx.query<CheckpointRow>(
      `
        UPDATE batch_checkpoints
        SET
          cursor_created_at = $1,
          cursor_id = $2,
          processed_count =
            processed_count + $3,
          updated_at = now()
        WHERE job_name = $4
          AND logical_run_id = $5
          AND partition_key = $6
          AND processor_version = $7
        RETURNING *
      `,
      [
        last.createdAt,
        last.id,
        rows.length,
        checkpoint.jobName,
        checkpoint.logicalRunId,
        checkpoint.partitionKey,
        checkpoint.processorVersion,
      ],
    );

    return mapCheckpoint(result.rows[0]);
  });
}

트랜잭션이 제공하는 결과는 둘 중 하나다.

1. 처리 결과와 체크포인트가 모두 커밋
2. 둘 다 롤백

체크포인트만 먼저 저장하는 위험한 순서를 보자.

await saveCheckpoint(lastRow);
await writeProcessedResults(rows);

첫 줄 뒤 프로세스가 죽으면 다음 실행은 해당 청크를 건너뛰지만 처리 결과는 없다. 데이터 누락이다.

반대로 처리 결과를 먼저 저장한다.

await writeProcessedResults(rows);
await saveCheckpoint(lastRow);

두 번째 줄 전에 죽으면 결과는 있지만 체크포인트가 뒤에 있어 같은 청크를 다시 처리한다. 결과 쓰기가 멱등하다면 중복 처리 비용만 발생한다. 멱등하지 않다면 중복 부작용이 생긴다.

원자적으로 묶을 수 없다면 at-least-once를 선택한다

“처리 안 했는데 했다고 표시”하는 누락보다 “처리했을 수도 있어 다시 처리”하는 편이 보통 복구 가능하다. 체크포인트는 결과가 확정된 뒤 이동하고, 재실행에 안전한 처리 로직을 만든다.

트랜잭션 밖의 외부 부작용은 따로 다룬다

외부 검색 API, 이메일, 결제 API는 로컬 DB 트랜잭션에 참여하지 않는다.

await database.transaction(async (tx) => {
  await searchApi.index(document);
  await tx.updateCheckpoint(cursor);
});

이 코드는 외부 API 성공 뒤 DB 트랜잭션이 롤백될 수 있다. 재시작하면 같은 문서를 다시 전송한다.

sequenceDiagram
    participant W as Worker
    participant API as External API
    participant DB as Database

    W->>API: 문서 인덱싱
    API-->>W: 성공
    W->>DB: 체크포인트 저장
    DB--xW: commit 실패
    Note over W: 재시작 후 같은 행 다시 처리
    W->>API: 같은 문서 다시 인덱싱

멱등한 Upsert

외부 시스템이 문서 ID 기준 Upsert를 제공하면 안정적인 원본 ID를 사용한다.

await searchApi.upsertDocument({
  documentId: `source:${row.id}`,
  version: row.version,
  body: toSearchDocument(row),
});

같은 문서를 다시 보내도 결과가 덮어쓰기라면 안전하다. 오래된 재시도가 새 문서를 덮지 않도록 버전 조건을 지원하는지도 확인한다.

Idempotency Key

await externalApi.performAction({
  idempotencyKey:
    `${logicalRunId}:${row.id}:${processorVersion}`,
  payload: buildPayload(row),
});

키의 유효 기간이 배치 재시도 가능 기간보다 짧으면 나중 재실행에서 중복이 생길 수 있다. 외부 API의 키 보존 정책을 확인한다.

Outbox

DB 결과와 “외부로 보낼 의도”를 같은 트랜잭션에 저장하고 별도 Publisher가 전달한다.

BEGIN;

INSERT INTO normalized_records (
  source_id,
  normalized_value
)
VALUES ($1, $2)
ON CONFLICT (source_id)
DO UPDATE SET
  normalized_value = EXCLUDED.normalized_value;

INSERT INTO outbox_events (
  event_id,
  event_type,
  aggregate_id,
  payload,
  created_at
)
VALUES (
  $3,
  'SearchDocumentUpsertRequested',
  $1,
  $4,
  now()
)
ON CONFLICT (event_id)
DO NOTHING;

UPDATE batch_checkpoints
SET
  cursor_created_at = $5,
  cursor_id = $1,
  updated_at = now()
WHERE job_name = $6
  AND logical_run_id = $7;

COMMIT;

Publisher는 Outbox 이벤트를 중복 전달할 수 있으므로 소비자도 이벤트 ID를 기준으로 멱등하게 처리한다. 자세한 패턴은 Outbox Pattern으로 DB와 이벤트 발행 일치시키기에 정리했다.

실패 직전 청크는 다시 실행될 수 있다

체크포인트를 청크마다 저장하면 실패한 청크 안의 일부 행은 이미 처리되었을 수 있다.

청크: 501 ~ 1000
501 ~ 780 처리 성공
781 처리 중 실패
체크포인트는 500
재시작 시 501부터 다시 처리

이 동작은 버그가 아니라 일반적인 at-least-once 처리 결과다. 다음 중 하나가 필요하다.

행별 처리를 멱등하게 만들기

INSERT INTO processed_records (
  source_id,
  processor_version,
  result
)
VALUES ($1, $2, $3)
ON CONFLICT (
  source_id,
  processor_version
)
DO NOTHING;

처리 장부를 확인하기

SELECT source_id
FROM processed_records
WHERE processor_version = $1
  AND source_id = ANY($2);

이미 처리된 행을 건너뛴다. 조회 비용과 장부 보관 비용이 추가된다.

각 행 뒤 체크포인트 이동

중복 범위는 줄지만 DB 쓰기 횟수가 늘고 처리량이 낮아진다. 결과와 체크포인트를 매 행 트랜잭션으로 커밋한다.

선택은 부작용 비용과 처리량의 균형이다.

전략 중복 재처리 범위 DB 커밋 수 적합한 작업
실행 끝에 한 번 전체 실행 최소 빠르고 완전 멱등
청크마다 최대 청크 크기 중간 일반적인 대량 작업
행마다 최대 1행 최대 비싼 비멱등 부작용
처리 장부 이미 처리된 행 건너뜀 높음 감사·정확성이 중요

원본 데이터가 실행 중 바뀌는 경우

배치가 실행되는 동안 새 행이 추가되고 기존 행이 수정될 수 있다.

실행 시작 시 상한을 고정한다

현재 최대 커서를 읽어 이번 실행 범위를 확정한다.

SELECT created_at, id
FROM source_records
ORDER BY created_at DESC, id DESC
LIMIT 1;

이 값을 boundary로 저장하고 다음 청크는 상한 이하만 읽는다.

SELECT id, created_at, payload
FROM source_records
WHERE (created_at, id) > ($1, $2)
  AND (created_at, id) <= ($3, $4)
ORDER BY created_at ASC, id ASC
LIMIT $5;

실행 중 추가된 더 큰 커서의 행은 다음 실행에서 처리한다. 작업의 끝이 계속 밀려 영원히 완료되지 않는 문제도 막는다.

상한만으로 수정 일관성이 생기지는 않는다

상한을 고정해도 아직 읽지 않은 행의 payload가 바뀔 수 있다. 기본 Read Committed에서는 각 SQL 문이 시작할 때의 커밋 상태를 보므로 뒤 청크는 나중 변경을 볼 수 있다.

필요한 일관성 수준에 따라 선택한다.

수시간의 Repeatable Read 트랜잭션은 오래된 행 버전 정리를 방해하고 장애 복구 범위를 키울 수 있다. 일관성 요구를 먼저 확인하고 무조건 사용하지 않는다.

삭제된 행

아직 처리하지 않은 행이 삭제되면 단순 원본 스캔에서는 존재 자체를 알 수 없다. 검색 인덱스 동기화처럼 삭제 전파가 필요하면 soft delete, tombstone 이벤트, CDC 로그가 필요하다.

{
  "eventType": "RecordDeleted",
  "recordId": "record-example-42",
  "sequence": 1042
}

현재 테이블 스냅샷만 읽는 배치는 “없는 행”을 변경으로 감지하지 못한다.

체크포인트에 버전과 실행 범위를 포함한다

처리 코드가 바뀌었는데 과거 커서부터 이어서 실행하면 결과가 섞일 수 있다.

1 ~ 500,000: processor v1
500,001 ~ 1,000,000: processor v2

결과가 버전 독립적이지 않다면 체크포인트에 처리기 버전을 둔다.

ALTER TABLE batch_checkpoints
ADD COLUMN processor_version integer NOT NULL;

재시작 시 실행 코드가 기대한 버전과 비교한다.

function assertCompatibleCheckpoint(
  checkpoint: BatchCheckpoint,
  expectedVersion: number,
): void {
  if (
    checkpoint.processorVersion !== expectedVersion
  ) {
    throw new Error(
      "checkpoint processor version is incompatible",
    );
  }
}

버전이 바뀌었을 때 정책은 작업마다 다르다.

restart: 새 logicalRunId로 처음부터 재처리
migrate: 체크포인트 상태를 새 버전으로 변환
resume: 결과가 하위 호환될 때만 이어서 처리
dual-run: 새 결과 공간에 병렬 생성 후 교체

실행 범위도 저장한다.

{
  "filter": {
    "tenantId": "tenant-example",
    "createdBefore": "2025-08-08T00:00:00Z"
  },
  "boundary": {
    "createdAt": "2025-08-07T23:59:59Z",
    "id": "record-example-max"
  },
  "processorVersion": 3
}

재시작 명령의 필터가 달라졌는데 기존 체크포인트를 재사용하면 누락이나 중복이 생긴다. 필터 전체 또는 정규화된 해시를 함께 저장해 호환성을 확인한다.

행 단위와 청크 단위 체크포인트의 선택

체크포인트 주기를 작게 할수록 재처리 범위는 줄지만 저장 비용이 늘어난다.

간단한 비용 모델을 세울 수 있다.

평균 행 처리 시간 = 20ms
청크 크기 = 500
청크 처리 시간 ≈ 10초
실패 시 최대 재처리 = 500행, 약 10초

외부 API 비용이 행당 높다면 500행 재실행이 비쌀 수 있다. 반대로 단순 계산이면 10초 재처리는 checkpoint DB 쓰기를 행마다 하는 것보다 싸다.

청크 크기 선택에는 다음을 함께 본다.

동적인 청크 크기를 사용할 수도 있다.

function adjustChunkSize(
  current: number,
  durationMs: number,
  targetDurationMs: number,
): number {
  if (durationMs > targetDurationMs * 1.5) {
    return Math.max(50, Math.floor(current / 2));
  }

  if (durationMs < targetDurationMs * 0.5) {
    return Math.min(2000, current * 2);
  }

  return current;
}

다만 청크 크기가 변해도 커서는 마지막 실제 행을 기준으로 해야 한다. “페이지 번호 × 크기”를 체크포인트로 쓰면 크기 변경과 중간 데이터 변화에 취약하다.

재시작 가능한 Worker 구조

전체 흐름을 예시 코드로 조립해 보자. 코드는 특정 프로젝트에서 가져온 것이 아니라 패턴을 설명하기 위한 가상 구현이다.

type RunOptions = Readonly<{
  logicalRunId: string;
  partitionKey: string;
  processorVersion: number;
  initialChunkSize: number;
}>;

class ResumableBatchWorker {
  constructor(
    private readonly checkpointStore:
      CheckpointStore,
    private readonly sourceRepository:
      SourceRepository,
    private readonly processor:
      ChunkProcessor,
    private readonly logger: Logger,
  ) {}

  async run(options: RunOptions): Promise<void> {
    let checkpoint =
      await this.checkpointStore.loadOrCreate(
        options,
        () =>
          this.sourceRepository.captureBoundary(
            options.partitionKey,
          ),
      );

    assertCompatibleCheckpoint(
      checkpoint,
      options.processorVersion,
    );

    let chunkSize = options.initialChunkSize;

    while (true) {
      const startedAt = performance.now();

      const rows =
        await this.sourceRepository.loadNext({
          partitionKey: options.partitionKey,
          after: checkpoint.cursor,
          through: checkpoint.snapshotBoundary,
          limit: chunkSize,
        });

      if (rows.length === 0) {
        await this.checkpointStore.markCompleted(
          checkpoint,
        );
        return;
      }

      checkpoint =
        await this.processor.processAndCheckpoint({
          checkpoint,
          rows,
        });

      const durationMs =
        performance.now() - startedAt;

      this.logger.info(
        {
          logicalRunId: options.logicalRunId,
          processedCount:
            checkpoint.processedCount,
          chunkSize: rows.length,
          durationMs,
          cursor: checkpoint.cursor,
        },
        "batch chunk committed",
      );

      chunkSize = adjustChunkSize(
        chunkSize,
        durationMs,
        5_000,
      );
    }
  }
}

중단 신호를 청크 경계에서 처리한다

class ShutdownSignal {
  #requested = false;

  request(): void {
    this.#requested = true;
  }

  isRequested(): boolean {
    return this.#requested;
  }
}

while (true) {
  if (shutdownSignal.isRequested()) {
    logger.info(
      { cursor: checkpoint.cursor },
      "batch stopped at committed checkpoint",
    );
    return;
  }

  const rows = await loadNextChunk(checkpoint);
  checkpoint = await processAndCommit(rows);
}

청크 중간에 강제로 체크포인트를 앞으로 이동하지 않는다. 현재 트랜잭션을 끝내거나 롤백한 뒤 마지막 확정 위치에서 종료한다.

작업 한 청크가 종료 grace period보다 오래 걸리면 청크를 줄이거나 DB 쿼리와 외부 호출에 취소 신호를 전달해야 한다.

실패 복구와 수동 재실행

체크포인트가 있어도 어떤 실행을 자동으로 이어갈지 정책이 필요하다.

상태

pending → running → succeeded
             │
             ├→ failed → retrying
             └→ paused → running

자동 재시도 가능 오류

수동 검토가 필요한 오류

수동 재실행 명령은 기본적으로 기존 논리 실행을 이어갈지, 새 실행으로 처음부터 할지를 명시한다.

type ResumeCommand =
  | {
      mode: "resume";
      logicalRunId: string;
    }
  | {
      mode: "restart";
      previousRunId: string;
      newLogicalRunId: string;
      processorVersion: number;
    };

체크포인트를 직접 숫자로 수정하는 운영 절차는 피한다. 부득이하다면 변경 전후 값, 작업자, 사유를 감사 로그에 남기고 처리 결과와 일치하는지 검증한다.

체크포인트를 앞으로 옮기는 것은 데이터 삭제와 비슷하다

아직 처리되지 않은 범위를 영구적으로 건너뛸 수 있다. 뒤로 이동하는 것은 중복을 만들지만 멱등 처리로 회복할 여지가 있다. 수동 조정은 이 비대칭을 이해하고 수행해야 한다.

테스트와 관측

정상 완료 테스트보다 “어느 줄에서 죽어도 복구되는가”를 검증하는 것이 중요하다.

장애 주입 테스트

it("reprocesses only the uncommitted chunk", async () => {
  const processor = createProcessor({
    failAfterRows: 250,
  });

  await expect(
    worker.run(testRunOptions),
  ).rejects.toThrow();

  const checkpoint =
    await checkpointStore.load(testRunId);

  expect(checkpoint.processedCount).toBe(0);

  processor.disableFailure();
  await worker.run(testRunOptions);

  expect(await resultStore.count()).toBe(1000);
  expect(await resultStore.hasDuplicates()).toBe(false);
});

첫 청크 전체가 하나의 트랜잭션이면 250행 처리 뒤 실패해도 체크포인트와 결과가 모두 롤백되어야 한다.

커밋 직후 종료

1. 결과와 체크포인트 COMMIT 성공
2. 성공 로그 전에 프로세스 강제 종료
3. 재시작
4. 다음 커서부터 시작하는지 확인

로그가 없다는 이유로 DB 커밋까지 실패했다고 판단하면 안 된다. DB 상태가 진실의 원천이다.

동시 변경 테스트

체크포인트 불변식

function assertCheckpointInvariant(
  previous: BatchCheckpoint,
  next: BatchCheckpoint,
): void {
  if (
    compareCursor(next.cursor, previous.cursor) < 0
  ) {
    throw new Error(
      "checkpoint cursor must not move backwards",
    );
  }

  if (
    compareCursor(
      next.cursor,
      next.snapshotBoundary,
    ) > 0
  ) {
    throw new Error(
      "checkpoint must not exceed boundary",
    );
  }

  if (
    next.processedCount <
    previous.processedCount
  ) {
    throw new Error(
      "processed count must be monotonic",
    );
  }
}

수동 재처리처럼 뒤로 이동해야 하는 작업은 기존 체크포인트를 수정하지 않고 새 논리 실행을 만든다.

메트릭

progress =
  processedCount /
  estimatedTotalCount

총 행 수 계산이 비싸거나 원본이 변하면 정확한 퍼센트보다 마지막 커서, 처리 속도, checkpoint age를 보여주는 편이 낫다.

운영 체크리스트

마무리

체크포인트는 긴 배치를 빠르게 재개하기 위한 성능 최적화이면서, 처리 결과의 확정 범위를 나타내는 일관성 장치다.

안정적인 복합 커서와 실행 상한을 사용하고, DB 처리 결과와 체크포인트를 같은 트랜잭션으로 커밋해야 한다. 외부 시스템과 원자적으로 묶을 수 없을 때는 마지막 청크가 다시 실행될 것을 전제로 Upsert, Idempotency Key, Outbox를 적용한다.

체크포인트에 실행 ID, 파티션, 처리기 버전, 범위를 함께 저장하면 배포와 재시작 뒤에도 같은 의미로 작업을 이어갈 수 있다. 결국 좋은 체크포인트 설계는 “어디서 다시 시작할까”보다 “어디까지의 결과를 믿을 수 있는가”에 답한다.

참고 자료

관련 노트