MongoDB에 원본을 저장한 뒤 Redis Streams로 이벤트를 발행하고, Worker가 PostgreSQL 조회 모델을 만드는 구조를 운영했습니다. 전체 데이터 파이프라인의 처리 순서는 정상 흐름만 보면 단순하지만 MongoDB 저장과 Redis 발행은 하나의 트랜잭션이 아니었습니다. 저장 직후 프로세스가 종료되거나 Redis가 잠시 응답하지 않으면 원본은 존재하지만 이벤트는 없는 상태가 생깁니다.
이 틈을 애플리케이션 재시도만으로 막으려 하면 또 다른 문제가 생깁니다. 수집 요청을 실패로 반환하면 클라이언트가 전체 요청을 다시 보내 중복 원본을 만들 수 있습니다. 발행을 무한 재시도하면 HTTP 연결과 수집 처리량이 Redis 상태에 종속됩니다.
분산 트랜잭션을 도입하는 것도 시스템 복잡성과 운영 비용에 비해 과했습니다. 그래서 불일치가 잠시 생길 수 있음을 인정하고, 저장소의 상태로 이를 찾아 평소 처리 경로에 다시 넣는 Reconciliation Job을 만들었습니다. 핵심은 누락 데이터를 직접 PostgreSQL에 쓰는 별도 로직이 아니라 미완료 원본을 찾아 원래 이벤트를 다시 발행하는 것이었습니다.
불일치를 질의할 수 있는 상태가 먼저 필요했다
로그에서 Redis 발행 실패를 찾는 방식은 충분하지 않았습니다. 로그는 보존 기간이 있고, 프로세스가 저장 직후 죽으면 실패 로그조차 남지 않습니다. 복구 작업은 애플리케이션의 기억이 아니라 데이터의 현재 상태를 기준으로 해야 했습니다.
MongoDB 원본에 syncedAt을 두었습니다. Worker가 PostgreSQL 반영과 최종 확인을 마치면 이 필드를 갱신합니다. 오래전에 생성됐지만 syncedAt이 비어 있는 문서는 파이프라인에서 끝까지 처리되지 않은 후보가 됩니다.
{
_id: ObjectId("..."),
createdAt: ISODate("2026-08-17T00:00:00Z"),
occurredAt: ISODate("2026-08-16T23:59:58Z"),
sourceKey: "source-a",
deviceKey: "device-1",
syncedAt: null,
lastRequeuedAt: null,
syncFailure: null
}
모든 syncedAt == null 문서가 누락은 아닙니다. 방금 저장돼 Redis나 Worker에서 정상 처리 중일 수 있습니다. 생성 직후의 문서를 Job이 다시 발행하면 중복 처리량만 늘어납니다.
그래서 최소 age를 두고 일정 시간보다 오래된 문서만 대상으로 삼았습니다. syncFailure는 재시도로 해결되지 않는 원본 오류를 표시합니다. 필수 필드가 없거나 구조가 영구적으로 잘못된 문서까지 Job이 계속 발행하면 DLQ와 Pending만 증가합니다.
정상 재처리 후보와 운영자 조치가 필요한 실패를 검색 조건에서 분리했습니다.
Job은 직접 복구하지 않고 같은 Stream으로 되돌려 보냈다
Reconciliation Job 안에서 MongoDB 문서를 다시 변환하고 PostgreSQL 여러 테이블에 쓰는 것은 피했습니다. 그렇게 하면 Worker와 Job에 두 개의 업무 처리 구현이 생깁니다. 한쪽에 새 필드나 통계 로직을 추가할 때 다른 쪽을 놓치면 복구 결과가 정상 처리 결과와 달라집니다.
Job의 책임은 미완료 원본을 찾아 EventMessage로 변환하고 Redis Streams에 재발행하는 것까지입니다.
val payload = objectMapper.writeValueAsString(document.toMessage())
redisTemplate.opsForStream<String, String>().add(
MapRecord.create(
"event-stream",
mapOf("payload" to payload),
),
)
재발행된 메시지는 새 메시지와 같은 Consumer Group으로 들어갑니다. Worker는 MongoDB 원본을 다시 읽고 고유 키 기반 insert와 upsert를 수행합니다. 이미 PostgreSQL 일부 단계가 끝났더라도 멱등 처리로 최종 상태에 수렴합니다.
이 구조는 복구 경로가 평소 경로를 재사용한다는 장점이 있습니다. 대신 Worker의 멱등성이 약하면 Reconciliation이 중복 데이터나 통계 증가를 만듭니다. Job을 만들기 전에 정상 consumer가 같은 메시지를 여러 번 받아도 안전한지 확인해야 했습니다.
Reconciliation의 안전 지연과 처리 상한
Job은 오래된 미완료 문서를 생성 시각 순서로 최대 1000건만 읽습니다. 전체 backlog를 한 실행에서 없애려고 하면 MongoDB scan과 Redis 발행이 온라인 트래픽을 방해할 수 있습니다.
val now = Instant.now()
val minimumAge = now.minus(Duration.ofMinutes(5))
val requeueThreshold = now.minus(Duration.ofMinutes(30))
val query = Query(
Criteria.where("syncedAt").isNull
.and("syncFailure").isNull
.and("createdAt").lt(minimumAge)
.andOperator(
Criteria().orOperator(
Criteria.where("lastRequeuedAt").exists(false),
Criteria.where("lastRequeuedAt").lt(requeueThreshold),
),
),
).with(Sort.by(Sort.Direction.ASC, "createdAt"))
.limit(1000)
가장 오래된 문서부터 처리하면 backlog의 최대 나이를 줄일 수 있습니다. 상한에 도달해도 Job을 실패로 보지 않고 다음 외부 스케줄에서 이어갑니다. 중요한 것은 한 번에 모두 처리하는 속도보다 온라인 서비스에 영향을 주지 않으면서 결국 0으로 수렴하는 것입니다.
이 쿼리가 전체 collection scan이 되지 않도록 검색 조건과 정렬에 맞는 인덱스를 만들었습니다. syncedAt, syncFailure, createdAt, lastRequeuedAt의 선택도와 실제 실행 계획을 운영과 유사한 데이터 분포에서 확인해야 합니다.
기존 처리 경로를 재사용하는 복구
스케줄러 중복 실행, 수동 재실행, 긴 Job 시간 때문에 두 인스턴스가 같은 문서를 조회할 수 있습니다. 조회 결과를 메모리에 가져온 뒤 바로 XADD하면 동일 문서가 여러 번 재큐잉됩니다. Worker가 멱등하더라도 불필요한 부하와 delivery count를 만듭니다.
각 문서를 발행하기 전에 lastRequeuedAt을 조건부 갱신해 claim했습니다. 조회할 때 사용한 조건이 여전히 맞고 다른 실행이 최근에 claim하지 않은 경우에만 modified count가 1이 됩니다.
val attemptedAt = Instant.now()
val claimed = mongoTemplate.updateFirst(
Query(
Criteria.where("_id").`is`(document.id)
.and("syncedAt").isNull
.and("syncFailure").isNull
.andOperator(requeueEligible(requeueThreshold)),
),
Update().set("lastRequeuedAt", attemptedAt),
"event",
)
if (claimed.modifiedCount != 1L) return@forEach
publish(document.toMessage())
이것은 긴 분산 락이 아닙니다. MongoDB의 단일 문서 조건부 UPDATE를 이용한 짧은 lease에 가깝습니다. 다른 인스턴스는 최근 lastRequeuedAt이 있는 문서를 건너뜁니다.
일정 시간이 지나도 syncedAt이 채워지지 않으면 다음 실행에서 다시 대상이 됩니다.
Job 전체를 분산 락으로 묶으면 구현은 단순하지만 한 인스턴스의 지연이 복구 전체를 멈춥니다. 반대로 중복 발행을 모두 허용하면 코드는 짧아도 backlog와 Worker 부하가 커집니다. 이 구조에서는 각 문서를 짧게 claim해 병렬성과 부분 진행을 확보하고, 발행 실패 때만 소유권을 확인해 claim을 되돌리는 편이 맞았습니다.
신규 시스템이라면 MongoDB 원본 저장과 같은 트랜잭션에 outbox 문서를 만들어 이벤트 기록까지 원자화하는 방식도 고려할 수 있습니다. 다만 별도 dispatcher와 schema를 운영해야 하므로, 이미 원본에 완료 상태가 있고 Job이 batch로 실행되는 현재 구조에는 문서별 claim이 더 작고 명확한 해법이었습니다.
Redis 발행 실패 뒤의 claim 복구
claim 업데이트는 성공했지만 XADD가 실패할 수 있습니다. lastRequeuedAt을 그대로 두면 다음 실행은 재큐잉 간격이 끝날 때까지 문서를 건너뜁니다. 빠른 복구를 위해 자신이 설정한 정확한 시각과 일치할 때만 값을 되돌렸습니다.
fun rollbackClaim(id: ObjectId, attemptedAt: Instant) {
mongoTemplate.updateFirst(
Query(
Criteria.where("_id").`is`(id)
.and("lastRequeuedAt").`is`(attemptedAt),
),
Update().unset("lastRequeuedAt"),
"event",
)
}
ID만 조건으로 unset하면 더 늦게 성공한 다른 실행의 claim까지 지울 수 있습니다. 소유 토큰처럼 attemptedAt을 비교함으로써 자신이 만든 상태만 되돌립니다. rollback 자체도 실패할 수 있습니다.
이 경우 문서는 영구 유실되는 것이 아니라 재큐잉 간격 이후 다시 후보가 됩니다. 오류 로그에는 원본 ID를 남기되 전체 payload나 민감 필드는 기록하지 않았습니다.
Job 성공과 문서 성공을 구분했다
1000개 중 일부만 발행에 실패했는데 Job을 성공으로 끝내면 스케줄러와 알림 시스템이 문제를 놓칠 수 있습니다. 반대로 한 문서 실패에서 즉시 중단하면 뒤의 정상 문서가 복구되지 않습니다. 각 문서를 독립적으로 시도해 성공과 실패를 세고, batch 끝에서 실패가 하나라도 있으면 Job을 실패 상태로 종료했습니다.
이렇게 하면 가능한 문서는 최대한 복구하면서 외부 운영 시스템에는 문제가 드러납니다.
var success = 0
var failed = 0
documents.forEach { document ->
runCatching { claimAndPublish(document) }
.onSuccess { success++ }
.onFailure { failed++ }
}
check(failed == 0) {
"reconciliation failed for $failed document(s)"
}
Spring Batch의 execution 상태와 별도로 다음 지표를 남겼습니다.
- 스캔된 문서 수, claim 성공 수, 경쟁으로 건너뛴 수.
- Redis 발행 성공·실패 수.
- 가장 오래된 미완료 문서의 age.
- 동일 문서의 requeue count 또는
lastRequeuedAt. - Reconciliation으로 복구된 뒤
syncedAt이 채워지는 데 걸린 시간. syncFailurecategory별 보류 문서 수.
완료 확인을 담당하는 Worker
Job이 XADD에 성공했다고 PostgreSQL 동기화가 끝난 것은 아닙니다. Job이 Redis 메시지를 다시 조회하며 완료를 기다리면 Worker와 결합되고 실행 시간이 길어집니다. Reconciliation의 완료는 재큐잉 성공까지로 제한했습니다.
최종 처리 완료는 원래 Worker가 PostgreSQL 쓰기와 MongoDB syncedAt 갱신 후 ACK하는 흐름에서 결정됩니다. 다음 Job 실행에서 syncedAt이 채워진 문서는 자연스럽게 검색 대상에서 빠집니다. 시스템의 각 구성 요소가 하나의 상태 전이를 소유하도록 한 것입니다.
오래된 lastRequeuedAt 문서가 반복해서 나타난다면 단순 발행 문제가 아니라 Worker 처리 실패일 가능성이 높습니다. 이때 delivery count, DLQ, PostgreSQL 오류와 함께 진단해야 합니다. Reconciliation의 재발행 횟수만 늘리는 것은 해결이 아닙니다.
Dual Write의 빈틈을 상태로 드러내고 메우기
서로 다른 저장소 사이에서 Dual Write의 틈을 완전히 없애기 어렵다면, 그 틈을 발견하고 복구할 수 있는 상태를 설계해야 합니다. 여기서는 MongoDB의 syncedAt이 불일치의 증거가 됐고 Reconciliation Job은 오래된 미완료 문서를 원래 Stream으로 되돌려 보냈습니다. 안전 지연은 정상 처리와 복구의 경합을 줄였고, 배치 상한은 온라인 부하를 통제했습니다.
문서별 조건부 claim과 소유자 확인 rollback은 중복 Job 실행에서도 같은 원본이 과도하게 발행되는 것을 막았습니다. 무엇보다 Job이 별도 PostgreSQL 쓰기 로직을 갖지 않고 평소 Worker 경로를 재사용해 정상 처리와 복구 처리의 결과가 갈라지지 않았습니다. Reconciliation은 장애가 났을 때 한 번 실행하는 임시 스크립트가 아닙니다.
Dual Write를 선택했다면 누락 탐지와 재발행, 최종 완료 확인까지가 원래 파이프라인의 일부입니다.
답글 남기기