Transactional Outbox에서 여러 워커가 이벤트를 안전하게 선점하는 방법

Transactional Outbox를 구현하면 애플리케이션의 데이터 변경과 이벤트 저장을 하나의 DB 트랜잭션으로 묶을 수 있다.

하지만 Outbox 테이블에 이벤트를 저장했다고 모든 문제가 끝나는 것은 아니다.

Worker A ─┐
          ├─ 같은 PENDING 이벤트 조회
Worker B ─┘

여러 워커가 같은 이벤트를 조회하면 동일한 메시지를 Kafka에 중복 발행할 수 있다.

따라서 Outbox Relay에는 다음 설계가 필요하다.

1. 여러 워커가 서로 다른 이벤트를 선점
2. 선점한 워커가 죽으면 다른 워커가 재처리
3. 발행 실패 시 일정 시간 뒤 재시도
4. 중복 발행되더라도 소비자는 한 번만 처리
5. 동일 aggregate 이벤트의 순서 보장

Outbox 테이블

예제 테이블은 다음과 같이 구성할 수 있다.

CREATE TABLE outbox_events (
    id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
    event_id CHAR(36) NOT NULL,
    aggregate_type VARCHAR(100) NOT NULL,
    aggregate_id VARCHAR(100) NOT NULL,
    event_type VARCHAR(100) NOT NULL,
    payload JSON NOT NULL,

    status VARCHAR(20) NOT NULL DEFAULT 'PENDING',
    retry_count INT NOT NULL DEFAULT 0,
    next_attempt_at DATETIME NOT NULL,

    claim_token CHAR(36) NULL,
    processing_until DATETIME NULL,

    created_at DATETIME NOT NULL,
    published_at DATETIME NULL,

    UNIQUE KEY uk_outbox_event_id (event_id),
    INDEX idx_outbox_polling (
        status,
        next_attempt_at,
        id
    )
);

각 컬럼의 역할은 다음과 같다.

status
    PENDING     발행 대기
    PROCESSING  워커가 선점
    PUBLISHED   발행 완료
    DEAD        최대 재시도 횟수 초과

next_attempt_at
    다음 발행 시도 시간

processing_until
    현재 워커의 선점 만료 시간

claim_token
    어떤 워커가 선점했는지 확인하는 소유권 토큰

조건부 UPDATE만으로도 중복 선점은 막을 수 있다

가장 간단한 방식은 다음과 같은 조건부 UPDATE다.

UPDATE outbox_events
SET status = 'PROCESSING',
    claim_token = :claimToken,
    processing_until = DATE_ADD(NOW(), INTERVAL 1 MINUTE)
WHERE id = :eventId
  AND status = 'PENDING';

affected rows가 1이면 선점에 성공한 것이다.

0이면 다른 워커가 먼저 선점했으므로 해당 이벤트를 포기하고 다음 이벤트를 처리하면 된다.

이 방식은 안전하지만 워커가 많아지면 비효율적일 수 있다.

예를 들어 워커 20대가 모두 다음 쿼리를 실행하면:

SELECT *
FROM outbox_events
WHERE status = 'PENDING'
ORDER BY id
LIMIT 100;

모든 워커가 앞쪽의 동일한 이벤트를 조회할 수 있다.

결과적으로 실제 선점에는 한 워커만 성공하지만, 나머지 워커의 SELECT와 UPDATE는 낭비된다.

워커가 많다면 SKIP LOCKED로 나누어 가져간다

Outbox 테이블을 작업 큐처럼 polling한다면 다음 방식이 더 잘 맞을 수 있다.

BEGIN;

SELECT id
FROM outbox_events
WHERE (
        status = 'PENDING'
        AND next_attempt_at <= NOW()
      )
   OR (
        status = 'PROCESSING'
        AND processing_until < NOW()
      )
ORDER BY id
LIMIT 100
FOR UPDATE SKIP LOCKED;

UPDATE outbox_events
SET status = 'PROCESSING',
    claim_token = :claimToken,
    processing_until = DATE_ADD(NOW(), INTERVAL 1 MINUTE)
WHERE id IN (...);

COMMIT;

FOR UPDATE SKIP LOCKED는 다른 워커가 이미 잠근 row를 기다리지 않고 건너뛴다.

Worker A → 1~100번 선점
Worker B → 잠긴 row를 건너뛰고 101~200번 선점
Worker C → 201~300번 선점

따라서 여러 워커가 같은 앞쪽 이벤트만 반복해서 조회하는 현상을 줄일 수 있다.

여기서 중요한 것은 Kafka 발행 전에 DB 트랜잭션을 종료하는 것이다.

잘못된 방식

BEGIN
SELECT ... FOR UPDATE
Kafka 발행
UPDATE PUBLISHED
COMMIT

Kafka 응답을 기다리는 동안 DB row lock을 계속 잡고 있기 때문에 장애가 발생하면 트랜잭션이 길어질 수 있다.

다음처럼 선점 트랜잭션을 짧게 끝내는 편이 안전하다.

BEGIN
이벤트 선점
PROCESSING 변경
COMMIT

Kafka 발행

PUBLISHED 변경

Laravel에서 이벤트 선점하기

use Illuminate\Support\Facades\DB;
use Illuminate\Support\Str;

$claimToken = (string) Str::uuid();

$eventIds = DB::transaction(function () use ($claimToken) {
    $eventIds = DB::table('outbox_events')
        ->where(function ($query) {
            $query->where(function ($query) {
                $query->where('status', 'PENDING')
                    ->where('next_attempt_at', '<=', now());
            })->orWhere(function ($query) {
                $query->where('status', 'PROCESSING')
                    ->where('processing_until', '<', now());
            });
        })
        ->orderBy('id')
        ->limit(100)
        ->lock('FOR UPDATE SKIP LOCKED')
        ->pluck('id')
        ->all();

    if ($eventIds === []) {
        return [];
    }

    DB::table('outbox_events')
        ->whereIn('id', $eventIds)
        ->update([
            'status' => 'PROCESSING',
            'claim_token' => $claimToken,
            'processing_until' => now()->addMinute(),
        ]);

    return $eventIds;
});

이 트랜잭션이 종료된 다음 선점한 이벤트를 조회하여 Kafka에 발행한다.

$events = DB::table('outbox_events')
    ->whereIn('id', $eventIds)
    ->where('claim_token', $claimToken)
    ->get();

foreach ($events as $event) {
    publishToKafka(
        topic: $event->event_type,
        key: $event->aggregate_id,
        payload: $event->payload,
    );

    $updated = DB::table('outbox_events')
        ->where('id', $event->id)
        ->where('status', 'PROCESSING')
        ->where('claim_token', $claimToken)
        ->update([
            'status' => 'PUBLISHED',
            'published_at' => now(),
            'processing_until' => null,
            'claim_token' => null,
        ]);

    if ($updated === 0) {
        // 선점권이 만료되었거나 다른 워커가 재선점한 상태
        // 운영 로그 및 메트릭을 남긴다.
    }
}

claim_token 조건이 필요한 이유는 오래 멈춰 있던 워커가 뒤늦게 살아나는 경우 때문이다.

Worker A 선점
→ A가 오래 멈춤
→ lease 만료
→ Worker B 재선점
→ A가 뒤늦게 처리 완료

claim_token을 확인하지 않으면 A가 B의 처리 상태를 덮어쓸 수 있다.

다만 claim token도 Kafka 중복 발행 자체를 완전히 막아주지는 못한다.

Kafka 발행 성공 후 DB 업데이트가 실패할 수 있다

다음 장애 구간은 완전히 제거하기 어렵다.

1. Kafka 발행 성공
2. 워커 종료
3. PUBLISHED 업데이트 실패
4. lease 만료
5. 다른 워커가 같은 이벤트 재발행

Kafka와 애플리케이션 DB를 하나의 로컬 트랜잭션으로 묶을 수 없기 때문이다.

따라서 Outbox Relay는 일반적으로 다음 전달 방식을 전제로 한다.

발행: at-least-once
소비: idempotent

Kafka producer의 idempotence는 broker 통신 과정의 재시도로 발생하는 중복을 줄여주지만, 애플리케이션이 같은 Outbox 이벤트를 다시 send()하는 상황까지 모두 해결하지는 못한다.

소비자 멱등 처리는 비즈니스 로직과 같은 트랜잭션에서 수행한다

단순히 eventId가 처리됐는지 조회하는 것만으로는 중복 처리를 완전히 막을 수 없다.

예를 들어 소비자가 다음 순서로 이벤트를 처리한다고 해보자.

1. processed_events에서 eventId 조회
2. 처리 기록이 없으면 주문 상태 변경
3. 처리가 끝나면 processed_events에 eventId 저장

 

2번에서 주문 상태를 변경한 직후, 3번의 처리 기록을 저장하기 전에 서버가 종료될 수 있다.

이 경우 주문 변경은 완료됐지만 processed_events에는 기록이 남지 않는다. 그러면 Kafka가 같은 이벤트를 다시 전달했을 때 소비자는 처음 받은 이벤트로 판단하고 주문 처리 로직을 다시 실행할 수 있다.

따라서 다음 두 작업은 반드시 하나의 DB 트랜잭션으로 처리해야 한다.

.

1. processed_events에 eventId 저장
2. 주문 상태 변경 등 실제 비즈니스 로직 실행

두 작업을 하나의 트랜잭션으로 묶으면 처리 도중 서버가 종료되더라도 둘 다 함께 롤백된다. 반대로 트랜잭션이 정상적으로 완료되면 eventId와 주문 변경 결과가 함께 저장된다.

같은 이벤트가 다시 전달되면 eventId의 UNIQUE 제약조건으로 이미 처리한 이벤트라는 것을 확인하고, 비즈니스 로직을 다시 실행하지 않는다.

 

DB::transaction(function () use ($event) {
    $inserted = DB::table('processed_events')
        ->insertOrIgnore([
            'event_id' => $event->eventId,
            'processed_at' => now(),
        ]);

    if ($inserted === 0) {
        // 이미 처리한 이벤트
        return;
    }

    DB::table('orders')
        ->where('id', $event->orderId)
        ->update([
            'status' => 'PAID',
        ]);
});

두 작업이 같은 트랜잭션에 있으므로 다음 둘 중 하나만 발생한다.

processed_events 저장 + 주문 변경 모두 성공

또는

둘 다 rollback

재시도는 선점 충돌과 발행 실패를 구분한다

선점 실패는 장애가 아니다.

affected rows = 0
→ 다른 워커가 먼저 가져감
→ 즉시 다음 이벤트 처리

반면 timeout이나 일시적인 Kafka 장애는 재시도할 수 있다.

1차 실패 → 1초 후
2차 실패 → 5초 후
3차 실패 → 30초 후
4차 실패 → 2분 후
5차 실패 → DEAD

실제 운영에서는 모든 워커가 동시에 재시도하지 않도록 jitter를 추가하는 것이 좋다.

$delaySeconds = min(
    300,
    (2 ** $retryCount) + random_int(0, 5)
);

잘못된 payload, schema 오류, 인증 오류처럼 재시도로 해결되지 않는 오류도 구분해야 한다.

여러 워커를 사용하면 이벤트 순서가 뒤바뀔 수 있다

중복뿐만 아니라 순서도 고려해야 한다.

OrderCreated
OrderPaid

두 이벤트를 서로 다른 워커가 선점하면 OrderPaid가 먼저 Kafka에 발행될 수 있다.

Kafka message key를 aggregate_id로 지정하면 동일 주문의 이벤트가 같은 partition으로 들어가지만, Outbox Relay가 이미 반대 순서로 발행했다면 Kafka가 순서를 복구해주지는 않는다.

순서가 반드시 필요하다면 다음 중 하나가 추가로 필요하다.

aggregate_id 기준으로 worker shard 분리
aggregate_version을 통한 순서 검증
동일 aggregate 이벤트 직렬 처리
CDC/Debezium 기반 Outbox Relay 사용

정리

Outbox에서 중요한 것은 특정 락 하나를 선택하는 것이 아니다.

여러 워커가 이벤트를 어떻게 나누어 가져가는가
선점한 워커가 죽으면 어떻게 복구하는가
오래된 워커가 상태를 덮어쓰지 않게 하는가
발행 실패를 언제, 어떻게 재시도하는가
중복 이벤트를 소비자가 어떻게 막는가
동일 aggregate의 이벤트 순서를 어떻게 지키는가

워커가 적고 충돌이 드물다면 조건부 UPDATE만으로도 충분하다.

반면 여러 워커가 Outbox를 계속 polling하는 구조라면 짧은 트랜잭션에서 SELECT FOR UPDATE SKIP LOCKED로 batch를 선점하는 방식이 더 효율적일 수 있다.

최종적으로는 다음 조합으로 이해하면 된다.

SKIP LOCKED       → 워커끼리 이벤트를 나눠 가져감
processing_until  → 죽은 워커의 작업을 복구
claim_token       → 오래된 워커의 상태 변경 방지
consumer event_id → 중복 이벤트 처리 방지
aggregate_id      → 이벤트 순서 설계