Redis Streams의 Pending 메시지를 안전하게 회수하는 방법

처음에는 XREADGROUP으로 메시지를 읽었습니다. 처리가 끝나면 XACK하면 된다고 생각했습니다. 정상 흐름만 보면 맞는 설명입니다.

운영에서는 다른 질문이 더 중요했습니다. Worker가 ACK 전에 종료되면 누가 메시지를 다시 처리해야 할까요? 전달된 메시지는 ACK 전까지 목록에 남습니다. 이 글은 MongoDB·Redis Streams·PostgreSQL 파이프라인 가운데 Pending 복구 경로를 좁혀 설명합니다.

이 목록을 Pending Entries List, 즉 PEL이라고 부릅니다. 메시지는 사라지지 않지만 기존 consumer가 소유한 상태로 남습니다. 새 메시지만 읽는 Worker는 이 record를 다시 보지 못합니다.

프로세스가 재시작될 때 consumer 이름이 바뀌면 오래된 Pending이 영원히 남을 수도 있습니다. Pending을 너무 빨리 회수할 수도 있습니다. 그러면 정상 처리 중인 메시지를 다른 Worker가 동시에 실행합니다.

회수 기능 자체보다 idle 기준, 멱등성, ACK 복구, 영구 실패 분리가 더 중요했습니다.

Pending은 실패 큐가 아니라 전달 상태다

Consumer Group은 각 메시지가 어떤 consumer에게 몇 번 전달됐고 마지막 전달 이후 얼마나 지났는지 관리합니다. Worker가 메시지를 읽으면 PEL에 들어가고 ACK되면 빠집니다.

XADD -> 새 메시지
  │
  ▼
XREADGROUP -> consumer-a에 전달, PEL 등록
  │
  ├─ 처리 성공 + XACK -> PEL 제거
  │
  └─ 프로세스 종료 -> consumer-a PEL에 계속 남음
                         │
                         ▼
                    XAUTOCLAIM
                         │
                         ▼
                    consumer-b가 재처리

Pending 수가 0이 아니라는 사실만으로 장애라고 판단할 수는 없습니다. 현재 처리 중인 메시지도 Pending입니다. 세 가지 신호가 중요합니다.

Pending 시간, delivery count, consumer 생존 여부입니다. 운영 화면에서는 unread lag와 Pending을 함께 봤습니다. 아직 한 번도 읽히지 않은 메시지는 lag에 들어가고, 이미 읽혔지만 끝나지 않은 메시지는 Pending에 들어갑니다.

전체 backlog는 두 값을 합쳐야 이해할 수 있습니다.

새 메시지보다 먼저 보는 오래된 Pending

Worker polling loop는 일정 주기마다 claim 대상이 있는지 먼저 확인하고, 그다음 새 메시지를 읽습니다. 매 반복에서 전체 Pending을 조회하면 Redis 부하가 커질 수 있어 claim scan interval을 별도로 두었습니다.

while (running) {
    val claimed = claimPendingIfDue(consumerName)
    if (claimed.isNotEmpty()) {
        process(claimed, Source.PENDING)
    }

    val fresh = readNewRecords(
        consumer = consumerName,
        count = 50,
        block = Duration.ofMillis(500),
    )
    process(fresh, Source.NEW)
}

Pending을 먼저 보는 이유는 새 메시지가 계속 들어오는 환경에서 오래된 실패가 굶지 않게 하기 위해서입니다. 다만 claim 실패가 새 메시지 처리까지 막게 하지는 않았습니다. Redis의 claim 명령이 일시적으로 실패하면 다음 claim 시각을 지연하고 새 record 소비는 계속합니다.

이 선택에는 trade-off가 있습니다. 오래된 Pending 복구를 최우선으로 하면 특정 poison message가 처리량을 독점할 수 있습니다. 그래서 한 번에 claim하는 수를 제한하고 새 메시지와 복구 메시지 모두 진행할 수 있게 했습니다.

XAUTOCLAIM의 min-idle-time이 중복 실행의 경계다

XAUTOCLAIM은 지정한 idle 시간보다 오래된 Pending record의 소유자를 새 consumer로 바꿉니다. idle 기준이 짧으면 정상 처리 중인 record를 빼앗을 수 있고, 너무 길면 장애 복구가 늦어집니다.

XAUTOCLAIM event-stream event-workers worker-2 300000 0-0 COUNT 50

위 예시는 마지막 전달 후 5분이 지난 record를 최대 50개 회수합니다. 5분이라는 값은 임의로 정할 숫자가 아닙니다. 정상 batch 처리 시간의 p99, 데이터베이스 lock 대기, 외부 의존성 timeout을 측정하고 그보다 충분히 길게 잡아야 합니다.

처리 시간이 유동적이고 몇 분 이상 걸리는 작업이라면 단순 idle 기준만으로는 부족합니다. 작업 heartbeat 또는 별도 lease가 필요할 수 있습니다. 반면 이 파이프라인처럼 한 batch의 정상 처리 시간이 짧고 상한이 명확하다면 min-idle-time이 실용적인 복구 기준이 됩니다.

claim cursor도 관리해야 합니다. 한 번의 호출로 전체 PEL을 스캔하지 않고 반환된 next start ID를 다음 호출에 사용합니다. 끝까지 돌면 0-0으로 돌아가 다음 scan을 시작합니다.

Claim 이후 다시 확인하는 PEL 상태

네트워크 오류가 끼어들면 클라이언트가 받은 claim 결과와 Redis의 실제 PEL 상태가 다를 수 있습니다. 이미 다른 경로에서 ACK된 record가 claim 응답에 남아 있거나 delivery count 조회가 실패할 수도 있습니다. 회수한 record는 XPENDING으로 현재 delivery count를 확인한 뒤 처리 대상으로 삼았습니다.

PEL에 더 이상 존재하지 않으면 업무 처리를 시작하지 않았습니다.

val claimed = nativeOperations.autoClaim(
    consumerName = consumerName,
    minIdle = pendingIdle,
    startId = autoClaimCursor,
    count = claimCount,
)

val deliveries = nativeOperations.pendingDeliveryCounts(
    claimed.records.map { it.id.value },
)

val valid = claimed.records.filter { deliveries.containsKey(it.id.value) }
process(valid)

이 검증이 없으면 이미 완료된 메시지의 오래된 응답을 다시 처리할 수 있습니다. 물론 업무 처리가 멱등해야 한다는 원칙은 그대로지만, 불필요한 PostgreSQL 쓰기와 원본 조회를 줄일 수 있습니다.

재처리는 반드시 멱등해야 한다

Pending 회수는 중복 전달을 의도적으로 만드는 기능입니다. Worker가 PostgreSQL 커밋 후 ACK 직전에 죽었다면 회수된 record는 이미 반영된 데이터를 다시 처리합니다. 부모와 자식 테이블에는 업무 키 기반 고유 제약을 두고 insert-if-absent를 사용했습니다.

MongoDB 원본의 syncedAt이 이미 채워져 있다면 PostgreSQL processing을 건너뛰고 ACK만 수행합니다.

when {
    document == null -> keepPendingForRetry(record)
    document.syncedAt != null -> ackOnly += record.id
    document.syncFailure != null -> handleTerminalFailure(record, document)
    else -> processingTargets += ProcessingTarget(record, document)
}

이 빠른 경로는 ACK 실패를 복구하는 데 특히 중요합니다. 업무 처리와 syncedAt 갱신은 성공했지만 XACK만 실패한 경우, 다음 delivery에서 전체 변환을 반복하지 않고 원본 상태를 확인해 ACK를 다시 시도합니다.

재전달 시 원본 상태	처리 방식	이유
syncedAt != null	처리 생략, ACK 재시도	이미 최종 상태 확인됨
syncedAt == null, 정상 원본	멱등 처리 재실행	이전 실행 위치를 확정할 수 없음
원본 없음, delivery 적음	ACK하지 않고 재시도	저장 가시성 또는 일시 오류 가능
원본 없음, 기준 초과	Dead Letter	영구 누락 가능성
syncFailure 존재	실패 정책에 따라 DLQ/ACK	정상 backlog와 분리

XACK 실패도 부분 성공을 고려했다

여러 record를 한 번에 ACK하면 Redis는 실제로 ACK된 개수를 반환합니다. 요청한 ID가 10개인데 반환값이 8이라면 전체 성공으로 기록할 수 없습니다. 이미 ACK된 8개를 다시 ACK하는 것은 문제가 없지만 남은 ID가 무엇인지 확인해야 합니다.

val requested = recordIds.distinct()
val acknowledged = streamOps.acknowledge(group, *requested.toTypedArray())

if (acknowledged != requested.size.toLong()) {
    val stillPending = nativeOperations.pendingDeliveryCounts(
        requested.map(RecordId::getValue),
    )
    require(stillPending.isEmpty()) {
        "XACK incomplete: pending=${stillPending.keys}"
    }
}

XACK 응답 전에 연결이 끊기면 서버에서 실행됐는지 알 수 없습니다. 이 경우 같은 ID로 한 번 더 ACK하고 PEL을 확인합니다. 업무 성공과 ACK 성공을 같은 로그로 뭉치지 않고 단계별로 기록해야 어디에서 반복되고 있는지 알 수 있습니다.

재시도로 해결되지 않는 메시지의 분리

잘못된 JSON, 유효하지 않은 식별자, 구조적으로 필요한 필드가 없는 원본은 반복해도 성공하지 않습니다. 이런 record를 계속 claim하면 delivery count만 커지고 정상 Pending을 가립니다. 영구 실패 기준을 카테고리별로 정하고 Dead Letter Stream에 원본 payload, category, 민감 정보를 제거한 reason, delivery count를 기록했습니다.

원본 record ACK와 DLQ 기록이 둘 중 하나만 성공하면 다시 모호한 상태가 생깁니다. 그래서 Redis Lua로 XADD와 XACK을 한 번에 실행하는 방법을 사용했습니다.

redis.call('XADD', KEYS[2], '*',
    'payload', ARGV[2],
    'category', ARGV[3],
    'reason', ARGV[4])

local acknowledged = redis.call('XACK', KEYS[1], ARGV[1], ARGV[5])
if acknowledged == 0 then
    return redis.error_reply('source record was not pending')
end
return acknowledged

DLQ가 생겼다고 실패 처리가 끝난 것은 아닙니다. category별 건수, 최근 항목, 원본의 현재 syncFailure 상태를 운영 화면에서 확인할 수 있어야 합니다. 수정 가능한 데이터라면 원본을 고친 뒤 명시적으로 재큐잉하는 절차도 필요합니다.

모니터링에서 보아야 할 숫자

Redis Streams 자체의 메트릭과 업무 저장소의 상태를 함께 보지 않으면 처리 완료를 오해하기 쉽습니다.

• unread lag: 아직 어떤 consumer에게도 전달되지 않은 수.
• Pending total: 전달됐지만 ACK되지 않은 수.
• oldest pending idle: 가장 오래 멈춘 record의 시간.
• delivery count 분포: 반복 실패 또는 consumer churn 신호.
• consumer별 Pending과 idle: 죽은 consumer의 소유 메시지 탐지.
• claim 성공·실패·소요 시간.
• XACK 요청 수와 실제 ACK 수의 차이.
• MongoDB의 오래된 syncedAt == null 문서 수.
• Dead Letter category별 증가율.

Pending이 계속 0인 것이 항상 좋은 것도 아닙니다. Worker가 메시지를 읽자마자 ACK하는 잘못된 구현도 Pending은 0입니다. 원본의 완료 상태와 PostgreSQL 반영 수를 함께 대조해야 합니다.

Pending 복구가 완료되는 기준

Redis Streams는 메시지를 자동으로 정확히 한 번 처리해 주지 않습니다. 대신 PEL, delivery count, claim, ACK라는 상태를 제공해 애플리케이션이 복구 정책을 만들 수 있게 합니다. 안전한 Pending 회수에는 네 가지가 함께 필요했습니다.

정상 처리 시간보다 긴 idle 기준, 중복 실행을 흡수하는 멱등 저장, 최종 상태를 확인한 뒤의 ACK, 반복해도 성공하지 않는 메시지의 Dead Letter 분리입니다. 이 조건이 없으면 XAUTOCLAIM은 유실을 막는 장치가 아니라 중복과 부하를 늘리는 장치가 됩니다. 메시지 시스템의 안정성은 정상적으로 한 번 처리되는 경로가 아니라 ACK 전 어느 줄에서 프로세스가 죽어도 최종 상태로 수렴하는지로 판단해야 합니다.