MongoDB·Redis Streams·PostgreSQL 유실 방지 설계

MongoDB에 원본을 저장하고 PostgreSQL에 조회용 데이터를 만드는 파이프라인을 운영했습니다. 수집 API는 빠르게 응답해야 했고, 조회 API는 복잡한 필터와 통계를 제공해야 했습니다. 두 요구를 한 데이터베이스에 맡기기보다 MongoDB를 원본 저장소로, PostgreSQL을 조회 저장소로 분리했습니다.

처음 구조만 보면 간단합니다. 수집 요청을 받은 서버가 MongoDB에 문서를 저장하고 Redis Streams에 식별자를 발행합니다. Worker가 메시지를 읽어 원본을 변환한 뒤 PostgreSQL에 적재하면 됩니다.

그러나 서로 다른 세 시스템에는 하나의 트랜잭션을 걸 수 없습니다. MongoDB 저장 직후 프로세스가 죽을 수 있고, Redis 발행은 성공했지만 Worker가 ACK 전에 종료될 수도 있습니다. PostgreSQL 반영은 끝났는데 MongoDB의 완료 표시만 실패하는 경우도 생깁니다.

결국 이 문제는 데이터를 옮기는 방법보다 각 단계의 성공을 무엇으로 정의할 것인가에 관한 문제였습니다. 저장, 전달, 멱등 처리, 완료 확인, 사후 복구를 따로 고치면 한 지점의 실패가 다른 지점의 유실로 이어졌습니다. 그래서 각 단계에 남는 상태를 기준으로 전체 흐름을 다시 설계했습니다.

Source of Truth를 하나로 정한 이유

두 저장소를 모두 원본이라고 부르면 장애가 발생했을 때 어느 쪽을 기준으로 복구해야 하는지 결정할 수 없습니다. 이 구조에서는 MongoDB 문서를 Source of Truth로 정했습니다. 수집 API의 성공 기준은 MongoDB 저장이며 PostgreSQL은 읽기 성능을 위한 파생 모델입니다.

원본 문서에는 처리 완료 여부를 나타내는 syncedAt을 두었습니다. 문서가 생성될 때는 비어 있고, Worker가 PostgreSQL 반영을 끝낸 뒤에만 값이 채워집니다.

data class EventRecord(
    val id: ObjectId,
    val occurredAt: Instant,
    val sourceKey: String,
    val payload: Document,
    val createdAt: Instant = Instant.now(),
    val syncedAt: Instant? = null,
)

syncedAt == null은 반드시 실패를 뜻하지 않습니다. 정상 처리 중인 문서도 잠시 이 상태에 머뭅니다. 대신 일정 시간보다 오래 비어 있다면 파이프라인 어느 지점에서 처리가 멈췄다는 관찰 가능한 신호가 됩니다.

이 필드 덕분에 애플리케이션 로그가 유실돼도 원본 저장소만 조회해 미완료 범위를 찾을 수 있습니다. MongoDB 저장과 Redis 발행 사이에는 여전히 원자성의 틈이 있습니다. 발행 실패를 수집 요청 실패로 돌리면 클라이언트가 같은 데이터를 재전송하면서 중복 원본을 만들 수 있습니다.

반대로 발행 실패를 무시하면 조회 저장소에 데이터가 나타나지 않습니다. 여기서는 원본 저장을 우선 보장하고, 발행 실패는 뒤에서 설명할 Reconciliation Job이 복구하도록 했습니다.

원본 대신 참조만 담은 Stream 메시지

MongoDB 문서 전체를 Redis 메시지에 복제하면 전달 계층이 또 하나의 데이터 저장소가 됩니다. 원본 스키마가 커질수록 Stream 메모리 사용량이 증가하고, 재처리 시 Redis의 오래된 payload와 MongoDB의 현재 상태가 다를 수도 있습니다. 그래서 메시지에는 원본을 다시 읽는 데 필요한 식별자와 라우팅 정보만 담았습니다.

Stream 필드도 payload 하나로 고정해 producer와 consumer의 계약을 작게 유지했습니다.

data class EventMessage(
    val eventId: String,
    val occurredAt: Instant,
    val sourceKey: String,
    val deviceKey: String,
)

val record = MapRecord.create(
    "event-stream",
    mapOf("payload" to objectMapper.writeValueAsString(message)),
)
redisTemplate.opsForStream<String, String>().add(record)

Worker는 메시지를 받으면 eventId로 MongoDB 원본을 다시 조회합니다. 원본이 아직 보이지 않거나 일시적으로 조회에 실패하면 ACK하지 않습니다. Redis Streams의 Pending Entries List에 남겨 다음 처리 기회를 만듭니다.

이 방식은 메시지만으로 처리를 끝낼 수 없다는 비용이 있습니다. Worker가 MongoDB에 접근해야 하고, 원본 보존 기간도 파이프라인의 최대 복구 시간보다 길어야 합니다. 대신 원본이 하나뿐이므로 재처리할 때도 같은 데이터를 기준으로 변환할 수 있습니다.

ACK는 읽었을 때가 아니라 최종 상태를 확인한 뒤 보냈다

Redis Streams Consumer Group에서 메시지를 읽었다는 사실과 업무 처리가 끝났다는 사실은 다릅니다. 메시지를 꺼내자마자 ACK하면 이후 PostgreSQL 쓰기 실패를 Redis가 다시 알려줄 방법이 없습니다. Worker의 처리 순서는 다음과 같습니다.

Stream record를 읽고 JSON과 식별자를 검증합니다.
MongoDB에서 원본 문서를 조회합니다.
PostgreSQL에 이벤트, 세그먼트, 결과 항목 같은 조회 모델을 적재합니다.
통계 버킷처럼 같은 이벤트에서 파생되는 데이터를 갱신합니다.
MongoDB 문서의 syncedAt을 원자적으로 갱신합니다.
syncedAt이 실제로 채워졌는지 다시 확인합니다.
마지막으로 Redis record를 ACK합니다.

val outcomes = processingService.processBatch(targets)
val succeededIds = outcomes.successfulEventIds()

mongoRepository.markSynced(succeededIds)
val verified = mongoRepository.findSyncedIds(succeededIds)

val ackable = records
    .filter { it.eventId in verified }
    .map { it.recordId }

if (ackable.isNotEmpty()) {
    streamOperations.acknowledge(group, *ackable.toTypedArray())
}

중간에 실패하면 record는 Pending으로 남습니다. 이 선택은 중복 처리 가능성을 받아들이는 대신 유실을 막습니다. 따라서 PostgreSQL 쓰기는 반드시 멱등해야 합니다.

At-least-once에서는 중복을 오류가 아닌 정상 경로로 봐야 한다

Worker가 PostgreSQL 커밋 직후 종료되고 ACK하지 못하면 같은 record가 다시 처리됩니다. 이 상황은 예외가 아니라 at-least-once 전달에서 정상적으로 발생할 수 있는 흐름입니다. 부모 이벤트에는 업무 식별자를 고유 키로 두고 중복 삽입을 무시했습니다.

세그먼트와 결과 항목 같은 자식도 이벤트 ID와 인덱스 조합을 고유 키로 만들었습니다. 반면 통계에서 count = count + 1을 그대로 재실행하면 중복 처리 때마다 숫자가 늘어납니다.

따라서 처리한 이벤트를 기록하거나 같은 입력에 대해 한 번만 반영되는 조건이 필요했습니다.

INSERT INTO event_record (event_id, occurred_at, source_key, has_issue)
VALUES (:eventId, :occurredAt, :sourceKey, :hasIssue)
ON CONFLICT (event_id, occurred_at) DO NOTHING;

INSERT INTO event_finding
    (event_id, occurred_at, segment_index, finding_index, name)
VALUES
    (:eventId, :occurredAt, :segmentIndex, :findingIndex, :name)
ON CONFLICT DO NOTHING;

업데이트가 섞인 모델은 더 주의해야 합니다. 같은 이벤트를 두 번 적용해도 결과가 같아지는 absolute value upsert가 가능한지, 아니면 별도 processed-event 테이블이 필요한지 데이터별로 판단해야 합니다. 하나의 processBatch 트랜잭션 안에서 관련 행과 통계를 함께 처리하면 PostgreSQL 내부의 부분 성공은 줄일 수 있지만 MongoDB와 Redis까지 묶을 수는 없습니다.

실패 지점마다 달라지는 복구 경로

파이프라인은 한 종류의 재시도로 모든 실패를 해결하지 않았습니다. 실패가 발생한 위치에 따라 책임을 나눴습니다.

실패 지점	남아 있는 증거	복구 방식
MongoDB 저장 실패	원본 없음	수집 요청 실패 반환
MongoDB 저장 후 Redis 발행 실패	`syncedAt == null` 원본	Reconciliation Job이 재발행
Worker 처리 중 종료	Redis Pending record	idle 시간이 지난 record claim
PostgreSQL 반영 후 Mongo 갱신 실패	멱등한 PostgreSQL 행, `syncedAt == null`	같은 record 재처리
ACK 실패	처리 완료 원본과 Pending record	다시 읽어 `syncedAt` 확인 후 ACK
형식이 영구적으로 잘못된 원본	반복 실패 record	실패 표식과 Dead Letter 분리

특히 ACK 실패를 처리 성공으로 간주하면 안 됩니다. 네트워크 오류로 XACK 응답을 받지 못했을 때 서버에서는 이미 ACK됐을 수도 있고 아닐 수도 있습니다. 재시도 전에 Pending 상태를 확인하고, 원본의 syncedAt이 채워져 있다면 업무 처리를 반복하지 않고 ACK만 회복할 수 있습니다.

JSON 파싱 실패나 잘못된 ObjectId처럼 재시도로 해결되지 않는 오류도 있습니다. 이런 record를 계속 Pending에 두면 정상 backlog와 영구 실패를 구분하기 어렵습니다. 일정 기준을 넘긴 영구 오류는 Dead Letter Stream으로 옮기되, 원본 record ACK와 Dead Letter 기록 사이에도 원자성이 필요합니다.

Redis Lua를 사용하면 두 동작을 한 서버 실행으로 묶을 수 있습니다.

Reconciliation은 임시 스크립트가 아니라 파이프라인의 일부다

MongoDB 저장과 Redis 발행을 분산 트랜잭션으로 묶지 않은 대신, 오래된 미완료 문서를 찾는 정식 Batch Job을 두었습니다. 생성된 지 얼마 안 된 문서는 Worker가 정상 처리 중일 수 있으므로 안전 지연 시간을 두고, 한 번에 읽는 건수에도 상한을 걸었습니다.

val query = Query(
    Criteria.where("syncedAt").isNull
        .and("syncFailure").isNull
        .and("createdAt").lt(now.minus(Duration.ofMinutes(5)))
).with(Sort.by("createdAt")).limit(1000)

mongoTemplate.find(query, EventRecord::class.java)
    .forEach { publish(it.toMessage()) }

여러 Job 인스턴스가 같은 문서를 동시에 발행하지 않도록 lastRequeuedAt을 조건부 갱신해 claim했습니다. 발행에 실패하면 자신이 설정한 시각과 일치할 때만 claim을 되돌렸습니다. 처리 결과를 로컬 메모리가 아니라 MongoDB 상태로 표현했기 때문에 Job이 중간에 종료돼도 다음 실행이 이어받을 수 있습니다.

운영에서는 처리량보다 미완료의 나이를 봤다

초당 처리 건수만 보면 파이프라인이 정상처럼 보여도 특정 record가 오래 Pending에 남아 있을 수 있습니다. 그래서 다음 지표를 함께 관찰했습니다.

Stream의 unread lag와 Pending 수.
가장 오래된 Pending record의 idle time.
syncedAt == null 문서 수와 가장 오래된 createdAt.
claim, 처리, ACK 단계별 실패 횟수.
Dead Letter category별 누적 수.
PostgreSQL batch 처리 시간과 충돌 재시도 횟수.

backlog 총량은 unread + pending으로 봐야 합니다. unread만 확인하면 이미 소비됐지만 끝나지 않은 record를 놓칩니다. 반대로 Pending 수만 보면 아직 어떤 consumer에게도 전달되지 않은 새 메시지를 알 수 없습니다.

이 파이프라인의 성공 기준은 모든 단계가 한 번씩 성공하는 것이 아닙니다. 같은 이벤트가 여러 번 도착하고 프로세스가 어느 줄에서 종료돼도 최종적으로 MongoDB 원본과 PostgreSQL 조회 모델이 일치하는 것입니다.

유실을 지연으로 바꾸는 조건

MongoDB, Redis Streams, PostgreSQL을 연결한다고 자동으로 유실 없는 파이프라인이 되지는 않습니다. 유실을 막은 핵심은 특정 제품보다 상태와 순서였습니다. MongoDB를 Source of Truth로 정하고, syncedAt으로 완료 여부를 남기고, 최종 상태를 확인한 뒤 ACK하며, PostgreSQL 쓰기를 멱등하게 만들었습니다.

저장과 발행 사이의 틈은 Reconciliation이 메웠습니다. exactly-once라는 표현으로 복잡성을 감추기보다 at-least-once를 인정하는 편이 설계가 명확했습니다. 중복은 고유 키와 상태 전이로 흡수하고, 실패는 Pending과 원본 상태로 관찰하며, 복구는 평소 처리 경로를 재사용합니다.

이 세 가지가 맞물려야 프로세스 재시작과 네트워크 장애가 데이터 유실이 아니라 지연으로 끝납니다.