Kotlin Flow와 분산락: 락은 걸었지만 실행은 보호되지 못한 이슈 대응

1. 서론: 락은 있었지만 실행을 보호하지 못했다

현재 재직중인 회사에서, 고객(오픈마켓 셀러)분들이 오픈마켓으로 부터 들어온 주문들을 우리쪽 플랫폼에서 보여주는 주문 관리 기능을 고도화 (이하 주문 동기화) 하는 작업을 수행했다. 이 작업을 진행하면서 주문 동기화 쪽에서 DB row lock wait가 반복적으로 관측되는 이슈를 발견하였다. 주문 동기화는 크게 두 가지 경로로만 들어온다. 사용자가 서비스를 사용하는 중에 API 요청을 통해 비동기로 이어지는 동기화 경로가 있었고, 별도로 active user를 대상으로 주기적으로 도는 배치 동기화 경로도 있다.

처음에는 "owner 단위 분산락이 있으니 같은 사용자의 주문 동기화가 겹치지는 않겠지"라고 생각했다. 실제 코드에도 lock은 있었다. 그런데 조사하면서 보니 문제는 락의 존재 여부가 아니었다. 락은 걸고 있었지만, 실제 주문을 수집하고 저장하는 실행 구간을 감싸지 못하고 있어서 분산락이 유효하게 동작하지 않는 이슈가 발견되었다 .

이로 인해, 배치의 주문동기화와 API의 주문동기화가 동시에 일어나서 특정 주문의 lock wait가 발생하는 이슈가 모니터링 슬랙 채널에서 지속적으로 확인되었다.

이 글은 주문 동기화 lock wait 전체에 대한 원인 분석 글은 아니다. 그 문제는 consumer 병렬 처리, batch transaction 범위, market별 저장량, DB row 접근 순서가 함께 얽힌 복합적인 문제였다. 여기서는 그 조사 과정에서 발견한 한 가지 구체적인 버그, 즉 Kotlin Flow의 실행 시점과 분산락 scope가 어긋났던 문제를 다룬다.

핵심은 단순하다.

분산락은 "lock block 안에 어떤 코드를 적었는가"가 아니라, "실제 부작용이 일어나는 시간이 lock block 안인가"로 봐야 한다.

Kotlin Flow처럼 실행을 뒤로 미루는 abstraction을 쓰면 이 차이가 꽤 쉽게 숨어버린다.

1.1. 주문 동기화에는 두 경로가 있었다

주문 동기화는 크게 두 경로로 들어왔다.

첫 번째는 사용자가 서비스를 사용하는 중에 요청하는 API 기반 동기화다. 요청이 들어오면 owner 단위로 동기화 task가 만들어지고, market별 메시지가 consumer로 전달된다. consumer는 market별로 주문을 수집하고 저장한다.

두 번째는 배치 동기화다. active user를 대상으로 주기적으로 주문을 다시 가져온다. 이 배치도 owner별로 마켓 주문을 수집하고 DB에 저장한다.

둘 다 결국 같은 주문 저장 영역을 건드린다. 그래서 같은 owner의 주문 동기화가 동시에 실행되면, 중복 저장이나 긴 row lock wait로 이어질 수 있다. 이를 막기 위해 owner 단위 분산락이 있었다.

여기까지만 보면 구조는 그럴듯하다.

의도한 구조 (owner 단위 lock)

╔════════════ LOCK (owner) ════════════╗   ← owner 단위 분산락
║                                      ║
║    1) active sync check              ║   ← 진행 중인 동기화가 있나?
║    2) collect orders                 ║   ← 외부 마켓 API
║    3) save orders                    ║   ← DB write
║                                      ║
╚══════════════════════════════════════╝

=> 이 세 단계가 같은 lock 안에서 실행되는 것이 의도였다
Plain Text
복사

문제는 실제 배치 코드가 이 순서로 "실행"되고 있지 않았다는 점이다.

2. 본론: cold Flow와 분산락의 실행 경계

2.1. 분산락이 있는데 왜 겹쳤을까

배치 쪽 코드는 대략 이런 모양이었다.

val flow = withOrderSyncLock(ownerId) {
    orderProcessor.syncOrderByRangeAsFlow(
        ownerId = ownerId,
        startDate = startDate,
        endDate = endDate,
        markets = markets,
    )
}

flow.collect { orders ->
    orders.forEach { order ->
        orderRepository.save(order)
    }
}
Kotlin
복사

겉으로 보면 withOrderSyncLock 안에서 주문 동기화 Flow를 만들고 있으니, 주문 동기화가 lock 안에서 실행되는 것처럼 보인다. 하지만 여기서 중요한 것은 syncOrderByRangeAsFlow(...)가 실제 주문 수집 결과가 아니라 Flow를 반환한다는 점이다.

Flow는 기본적으로 cold stream이다. 즉, Flow를 만드는 시점과 Flow 내부 로직이 실행되는 시점이 다르다. 위 코드에서 lock 안에서 실행된 것은 주문 수집이 아니라 "나중에 주문을 수집할 수 있는 Flow 객체 생성"이었다.

실제 외부 마켓 API 호출과 DB 저장은 나중에 writer에서 collect할 때 실행됐다. 그 시점에는 이미 owner lock이 풀린 뒤였다.

문제를 더 단순하게 쓰면 이렇게 된다.

processor
╔════════════ LOCK ════════════╗   ← lock 획득
║                              ║
║    create Flow object        ║   ← Flow "객체"만 생성 (부작용 없음)
║                              ║
╚══════════════════════════════╝   ← lock 해제
                │
                ▼   락이 이미 풀린 뒤
writer  (no lock)
┌──────────────────────────────┐
│    collect Flow              │
│    fetch orders              │   ← 외부 마켓 API 호출
│    save orders               │   ← DB write
└──────────────────────────────┘

=> LOCK 박스가 감싼 것은 "실행"이 아니라 Flow "객체" 하나뿐이었다
Plain Text
복사

분산락은 있었다. 하지만 락이 보호한 것은 실제 작업이 아니라 실제 작업을 표현하는 lazy object였다.

2.2. 잠깐 짚고가기: cold Flow는 언제 실행될까

Kotlin Flow를 쓸 때 가장 자주 놓치는 지점이 이 부분이다.

fun orders(): Flow<Order> = flow {
    println("collect 시점에 실행")
    emit(fetchOrder())
}

val flow = orders()
Kotlin
복사

위 코드에서 orders()를 호출해도 println은 찍히지 않는다. fetchOrder()도 호출되지 않는다. 단지 Flow<Order> 객체가 만들어질 뿐이다.

실행은 terminal operator가 호출될 때 시작된다.

flow.collect { order ->
    save(order)
}
Kotlin
복사

이제서야 flow { ... } block이 실행된다. 그래서 다음 코드는 생각보다 위험하다.

val flow = lockManager.withLock("order:sync:$ownerId") {
    flow {
        val orders = externalClient.fetchOrders(ownerId)
        emit(orders)
    }
}

flow.collect { orders ->
    orderRepository.saveAll(orders)
}
Kotlin
복사

얼핏 보면 외부 API 호출이 lock 안에 있는 것 같다. 하지만 실제로 lock 안에서 일어난 일은 flow { ... } 객체 생성뿐이다.

실행 시점을 펼쳐보면 이렇다.

cold Flow 가 실제로 실행되는 시점

╔══════════ withLock ══════════╗   ← 락 보유
║                              ║
║    create Flow object        ║   ← 여기서 "객체"만 생성된다
║                              ║
╚══════════════════════════════╝   ← 락 해제 (여기까지가 lock 구간)
                │
                ▼   terminal operator(collect) 호출 시점
collect 호출
┌──────────────────────────────┐
│    run flow { } block        │   ← 이제서야 block 실행
│    call external API         │
│    save orders               │
└──────────────────────────────┘

=> 락은 "실행 계획(Flow 객체)"만 감쌌고 "실행"은 감싸지 못했다
Plain Text
복사

락이 실행 계획을 감싸고 있을 뿐, 실행 자체를 감싸고 있지 않다. 이 차이가 이번 문제의 핵심이었다.

2.3. lock scope와 execution scope가 달랐다

배치 구조에서는 processor와 writer가 나뉘어 있었다. processor는 owner별로 처리할 값을 만들고, writer는 chunk 단위로 실제 저장을 수행한다.

기존 구조에서 processor는 owner lock을 잡고 Flow를 만들었다. writer는 나중에 그 Flow를 collect하면서 주문을 가져오고 저장했다.

── lock scope ──    (processor 가 쥐는 구간)
╔════════════ LOCK ════════════╗
║    create Flow object        ║
╚══════════════════════════════╝   ← 여기서 락이 풀린다
                ╎
                ▼   (락이 없는 시간)
── execution scope ──    (writer 가 실제 실행, 락 없음)
┌──────────────────────────────┐
│    collect Flow              │
│    fetch orders              │   ← 외부 마켓 API
│    save orders               │   ← DB write
└──────────────────────────────┘

=> lock scope 와 execution scope 가 어긋나 있다
Plain Text
복사

즉, lock scope와 execution scope가 달랐다.

이 문제는 코드 리뷰에서 눈으로 보기도 애매하다. withOrderSyncLock { syncOrderBy...AsFlow(...) }라는 표현만 보면 "락 안에서 동기화하네"라고 읽히기 쉽다. 하지만 AsFlow라는 이름이 말하듯, 이 함수는 결과를 지금 실행해서 돌려주는 함수가 아니라 나중에 실행될 stream을 돌려주는 함수였다.

분산락을 검증할 때는 lock block 안의 코드 모양보다 다음 질문이 더 중요하다.

•

외부 API 호출은 언제 실행되는가?

•

DB save는 언제 실행되는가?

•

그 시점에도 같은 owner lock이 유지되는가?

•

실패하거나 skip되는 경우에도 lock 경계가 의도대로 유지되는가?

이번 경우에는 마지막 두 질문의 답이 "아니오"였다.

2.4. lock을 실제 작업 범위로 옮기기

수정 방향은 processor가 Flow만 넘기지 않도록 만드는 것이었다. writer가 lock key를 만들 수 있어야 실제 실행 구간을 lock으로 감쌀 수 있다.

그래서 owner와 lazy Flow를 함께 담은 작업 단위를 만들었다.

data class OrderSyncFlowWork(
    val ownerId: Long,
    val flow: Flow<List<Order>>,
)
Kotlin
복사

processor는 이제 주문 수집을 시작하지 않는다. "이 owner에 대해 나중에 lock 안에서 실행할 작업"만 만든다.

OrderSyncFlowWork(
    ownerId = ownerId,
    flow = flow {
        emitAll(
            orderProcessor.syncOrderByRangeAsFlow(
                ownerId = ownerId,
                startDate = startDate,
                endDate = endDate,
                markets = markets,
            )
        )
    },
)
Kotlin
복사

실제 실행은 writer가 담당한다. writer는 work item을 받아 owner lock을 잡고, 그 안에서 active sync check, Flow collection, 주문 저장, sync log 저장까지 수행한다.

lockManager.withLock("order:sync:$ownerId") {
    if (activeChecker.existsActiveSync(ownerId)) {
        return@withLock
    }

    work.flow.collect { orders ->
        orders.forEach { order ->
            orderRepository.save(order)
        }
    }

    orderSyncLogRepository.save(OrderSyncLog(ownerId))
}
Kotlin
복사

이제 실행 경계는 이렇게 바뀐다.

processor
┌──────────────────────────────────┐
│    create OrderSyncFlowWork      │   ← ownerId + lazy Flow (아직 실행 X)
└──────────────────────────────────┘
                  │
                  ▼   work 를 writer 로 전달
writer
╔══════════════ LOCK ══════════════╗   ← lock 획득
║                                  ║
║    active sync check             ║
║    collect Flow                  ║
║    fetch orders                  ║   ← 외부 마켓 API 호출
║    save orders                   ║   ← DB write
║    save sync log                 ║
║                                  ║
╚══════════════════════════════════╝   ← lock 해제
                  │
                  ▼
┌──────────────────────────────────┐
│    publish events                │   ← 락 밖, 의도된 후처리
└──────────────────────────────────┘

=> 실제 부작용(API 호출 / DB 저장)이 전부 LOCK 박스 안에서 실행된다
Plain Text
복사

이벤트 발행은 lock 밖으로 뺐다. 주문 저장 결과를 바탕으로 알림/트래킹 이벤트를 발행하는 후처리이고, owner lock을 오래 잡고 있을 이유가 적었기 때문이다. lock 안에 꼭 있어야 하는 것은 같은 owner의 중복 수집/저장을 막기 위한 핵심 구간이었다.

수정하면서 몇 가지 다른 선택지도 검토했다.

선택지	판단
Flow 내부에서 lock을 잡기	collection은 lock 안으로 들어오지만, downstream writer의 save는 lock 밖에 남는다.
processor에서 eager하게 collect해서 list로 넘기기	저장은 여전히 writer에서 lock 밖에 남고, owner별 주문을 메모리에 크게 올릴 수 있다.
각 배치 writer에 lock/save 로직을 중복 구현하기	두 배치의 lock 정책이 갈라질 수 있고 테스트 비용이 늘어난다.
ownerId + flow work unit을 writer로 넘기기	collect와 save를 같은 owner lock 안에 둘 수 있다. 이 방향을 선택했다.

핵심은 writer가 flow만 받으면 안 된다는 점이었다. writer가 owner를 알아야 lock key를 만들 수 있고, 실제 실행 구간을 감쌀 수 있다.

2.5. 두 번째 문제: No transaction in context

lock scope를 바로잡고 나니 다른 문제가 드러났다.

배치를 로컬에서 실행했을 때 No transaction in context 예외가 반복됐다. 처음에는 lock 수정과 직접 관련이 없어 보일 수 있다. 하지만 실행 경계를 따라가면 연결이 보인다.

수정 전에는 processor에서 동기식 lock wrapper(withOrderSyncLock)가 쓰였고, 실제 Flow 실행은 writer의 기존 흐름에서 일어났다. 이때 lock 안에 있던 것은 Flow 객체를 반환하는 syncOrderByRangeAsFlow(...) 호출뿐이었다. 일반 함수라 동기 블록 안에서 호출해도 문제가 없었고, 그래서 동기식 wrapper로 충분했다.

수정 후에는 writer가 owner lock 안에서 실제 Flow를 collect하게 됐다. 그런데 Flow.collect는 suspend 함수다. collect를 lock 블록 안으로 옮기는 순간 그 블록은 suspend 블록이 되어야 했고, 자연히 락도 동기식 wrapper가 아니라 suspend 블록을 받는 withLock이어야 했다. 동기 락을 그대로 유지하려면 블록 안에서 runBlocking으로 collect를 감싸야 하는데, 그러면 외부 마켓 API 호출이 끝날 때까지 스레드를 통째로 붙잡게 되어 Flow를 쓰는 의미가 사라진다.

그렇게 도입한 suspending withLock은 내부에서 IO dispatcher로 전환했다. 그리고 바로 이 스레드 전환이 다음 문제로 이어졌다.

문제는 Exposed transaction context가 ThreadLocal 기반이라는 점이었다.

Spring Batch chunk transaction이 thread A에 묶여 있다고 해보자. 그런데 suspending lock 내부에서 withContext(IO)로 thread B로 넘어가면, thread B에는 기존 ThreadLocal transaction이 없다.

thread A  (Spring Batch worker)
┌──────────────────────────────────────┐
│    chunk transaction starts          │
│    tx lives in ThreadLocal           │
│    withLock { withContext(IO) }      │   ← 여기서 IO 스레드로 전환
└──────────────────────────────────────┘
                    ╎
                    ▼   thread 전환 (A -> B)
thread B  (IO dispatcher 가 올라탄 스레드)
┌──────────────────────────────────────┐
│    tx NOT in ThreadLocal             │   ← 트랜잭션이 없다
│    repository.read()                 │
│      -> No transaction in context    │   ← 예외 발생
└──────────────────────────────────────┘

=> dispatcher 가 바뀌면 ThreadLocal transaction 이 사라진다
Plain Text
복사

그 상태에서 lock 안의 Flow가 외부 마켓 주문을 수집하기 위해 authorization 정보를 읽거나, 기존 주문을 조회하는 Exposed repository를 호출했다. 일부 repository method는 자기 안에서 transaction { ... }을 열지 않고 ambient transaction이 있다고 가정하고 있었다.

그 가정이 dispatcher 전환 뒤에 깨진 것이다.

여기서 중요한 해석이 하나 있다. 이 문제는 "lock fix 때문에 완전히 새로운 문제가 생겼다"기보다, lock boundary를 실제 실행 위치로 옮기자 기존에 숨어 있던 ThreadLocal transaction 가정이 드러난 것에 가깝다.

실제 실행 구간을 lock 안으로 가져왔더니, 그 실행 구간 안에 있던 repository read들이 더 이상 우연히 기존 thread context에 기대지 못하게 된 것이다.

2.6. transaction fix 선택지 비교

이 문제를 고치는 방법도 몇 가지가 있었다.

2.6.1. Option A. lock 안 전체를 하나의 suspended transaction으로 감싸기

가장 단순하게 생각하면 owner lock 안의 전체 작업을 하나의 newSuspendedTransaction으로 감쌀 수 있다.

lockManager.withLock(key) {
    newSuspendedTransaction {
        work.flow.collect { orders ->
            orders.forEach { orderRepository.save(order) }
        }
    }
}
Kotlin
복사

하지만 이 방향은 선택하지 않았다. 주문 수집 과정에는 외부 마켓 API 호출이 포함된다. 외부 HTTP 호출은 수 초에서 수십 초까지 걸릴 수 있고, retry나 rate limit이 섞이면 더 길어질 수도 있다. 특히 ESM (지마켓, 옥션)이 제일 문제가 되었는데, ESM은 주문 조회 API를 유저 별로 5초에 1번만 호출하도록 rate limit이 구성되어 있었다.

그 시간 동안 DB transaction과 connection을 계속 잡고 있는 것은 위험하다. Hikari pool 고갈, long-running transaction, 외부 API 실패와 DB rollback의 결합 같은 문제가 생길 수 있다.

문제를 해결하려다 더 큰 transaction scope를 만드는 셈이다.

2.6.2. Option B-1. 필요한 read repository method를 작은 transaction으로 감싸기

최종 선택은 이 방향이었다.

lock 안에서 호출되는 Exposed read method 중 자체 transaction이 없던 것들을 각각 작은 transaction { ... }으로 감쌌다.

fun findOneByUserId(userId: Long): Authorization? = transaction {
    AuthorizationDao.find { AuthorizationTable.userId eq userId }
        .firstOrNull()
        ?.toEntity()
}
Kotlin
복사

이렇게 하면 dispatcher가 바뀌어도 해당 method는 자기 실행에 필요한 transaction을 직접 연다. 그리고 transaction 범위는 repository method 안의 짧은 DB read에만 한정된다. 외부 마켓 API 호출 시간 동안 DB connection을 붙잡지 않는다.

이 선택은 가장 이상적인 장기 구조라기보다, 이번 회귀를 작은 변경으로 정확히 막는 hotfix에 가까웠다.

2.6.3. Option B-2. read replica transaction까지 통일하기

다른 방향으로는 이참에 read repository들을 transaction(readDatabase) 형태로 통일하는 방법도 있었다.

하지만 이건 이번 문제의 해결 조건이 아니었다. No transaction in context를 막기 위해 필요한 것은 transaction context가 존재하는 것이지, read replica 라우팅을 이번 변경에 함께 넣는 것이 아니었다.

read replica 라우팅까지 통일하려면 repository 생성자, DI 설정, 테스트 fixture까지 변경 표면이 커진다. 회귀를 막기 위한 hotfix와는 성격이 달랐다. 그래서 후속 cleanup으로 분리하는 편이 낫다고 판단했다.

2.6.4. Option C. authorization/read 호출을 lock 밖으로 빼기

또 다른 방법은 lock 안에서 문제가 되는 read 호출 자체를 밖으로 옮기는 것이다.

하지만 이건 단순한 위치 이동이 아니다. owner lock이 어떤 일관성을 보호해야 하는지 다시 설계해야 한다. 자격증명 조회, active sync check, 외부 주문 수집, 저장 중 어디까지를 같은 critical section으로 볼 것인지 결정해야 한다.

이번 hotfix의 범위를 넘어서는 선택이었다.

정리하면 최종 선택은 B-1이었다.

옵션	내용	판단
A	lock 안 전체를 newSuspendedTransaction으로 감싼다	외부 HTTP 호출 동안 DB transaction/connection을 잡게 되어 위험하다.
B-1	영향받은 read method만 작은 transaction {}으로 감싼다	채택. 변경 표면이 작고 transaction이 짧다.
B-2	read replica transaction까지 통일한다	이번 회귀 해결과 직교하고 변경 표면이 크다.
C	read 호출을 lock 밖으로 옮긴다	lock scope 의도 자체를 다시 봐야 하므로 범위를 넘는다.

3. 결론: lock은 실행 범위로 검증해야 한다

이번 문제를 겪으면서 다시 확인한 것은 분산락의 검증 기준이다. 분산락을 썼는지보다 중요한 것은 그 락이 실제 부작용이 일어나는 시간을 감싸고 있는지다.

특히 Flow, Sequence, coroutine, lazy supplier처럼 실행을 뒤로 미루는 abstraction이 들어오면 코드의 작성 위치와 실행 위치가 쉽게 분리된다. lock block 안에 코드가 보인다고 해서 그 코드의 부작용도 lock 안에서 실행된다고 단정하면 안 된다.

비슷한 코드를 볼 때는 다음 질문을 체크해볼 만하다.

•

lock 안에서 실제 작업이 실행되는가, 아니면 실행 계획만 만들어지는가?

•

terminal operation은 어디에서 호출되는가?

•

외부 API 호출과 DB save가 같은 critical section 안에 있는가?

•

writer나 downstream collector에서 중요한 side effect가 실행되고 있지 않은가?

•

coroutine dispatcher 전환 뒤에도 ThreadLocal 기반 context가 유지된다고 가정하고 있지 않은가?

•

hotfix에서 해결하려는 회귀와 별도 cleanup을 섞고 있지 않은가?

이번 수정은 코드 한 줄을 lock 안으로 옮긴 작업이라기보다, "실행 경계"를 다시 그린 작업이었다.

락은 선언이 아니라 시간의 문제다. 어느 시점에 획득되고, 어느 시점에 해제되며, 그 사이에 어떤 부작용이 실제로 일어나는지 봐야 한다. Kotlin Flow를 쓰는 코드에서는 이 질문이 특히 중요해진다.