아웃박스 공통 구조 분리 (1단계)

src/main/kotlin/com/depromeet/piki/common/outbox/Outbox.kt

@@ -0,0 +1,62 @@
+package com.depromeet.piki.common.outbox
+
+import com.depromeet.piki.common.domain.LongBaseEntity
+import jakarta.persistence.Column
+import jakarta.persistence.EnumType
+import jakarta.persistence.Enumerated
+import jakarta.persistence.MappedSuperclass
+
+@MappedSuperclass
+abstract class Outbox : LongBaseEntity() {
+    @Enumerated(EnumType.STRING)
+    @Column(nullable = false, length = 16)
+    var status: OutboxStatus = OutboxStatus.PENDING
+        protected set
+
+    @Column(nullable = false)
+    var recoveryCount: Int = 0
+        protected set
+
+    fun attempt(): OutboxAttempt = OutboxAttempt(getId(), recoveryCount)
+
+    fun isHeldBy(attempt: OutboxAttempt): Boolean = status == OutboxStatus.PROCESSING && recoveryCount == attempt.recoveryCount
+
+    fun isInProgress(): Boolean = status == OutboxStatus.PENDING || status == OutboxStatus.PROCESSING
+
+    fun isFailed(): Boolean = status == OutboxStatus.FAILED
+
+    fun claim() = transit(from = OutboxStatus.PENDING, to = OutboxStatus.PROCESSING)
+
+    fun requeue(attempt: OutboxAttempt) = settle(attempt, OutboxStatus.PENDING)
+
+    fun succeed(attempt: OutboxAttempt) = settle(attempt, OutboxStatus.SUCCEEDED)
+
+    fun fail(attempt: OutboxAttempt) = settle(attempt, OutboxStatus.FAILED)
+
+    fun recover(maxRecoveries: Int): RecoveryOutcome {
+        if (recoveryCount >= maxRecoveries) {
+            transit(from = OutboxStatus.PROCESSING, to = OutboxStatus.FAILED)
+            return RecoveryOutcome.EXHAUSTED
+        }
+        transit(from = OutboxStatus.PROCESSING, to = OutboxStatus.PENDING)
+        recoveryCount++
+        return RecoveryOutcome.REQUEUED
+    }
+
+    // 회수 뒤 다시 집힌 행도 status 는 PROCESSING 이라 시도까지 맞춰 봐야 늦은 워커의 덮어쓰기를 막음
+    private fun settle(
+        attempt: OutboxAttempt,
+        to: OutboxStatus,
+    ) {
+        check(isHeldBy(attempt)) { "${javaClass.simpleName} ${getIdOrNull()} 은 시도 $attempt 가 쥐고 있지 않아 $to 로 갈 수 없다" }
+        status = to
+    }
+
+    private fun transit(
+        from: OutboxStatus,
+        to: OutboxStatus,
+    ) {
+        check(status == from) { "${javaClass.simpleName} ${getIdOrNull()} 은 $status 라 $to 로 갈 수 없다" }
+        status = to
+    }
+}

src/main/kotlin/com/depromeet/piki/item/domain/ParseAttempt.kt → src/main/kotlin/com/depromeet/piki/common/outbox/OutboxAttempt.kt

@@ -1,7 +1,7 @@
-package com.depromeet.piki.item.domain
+package com.depromeet.piki.common.outbox
 
 // 회수는 횟수를 올리므로 회수 전에 집은 워커의 결과는 기록 때 걸러짐
-data class ParseAttempt(
-    val itemParseOutboxId: Long,
+data class OutboxAttempt(
+    val outboxId: Long,
     val recoveryCount: Int,
 )

src/main/kotlin/com/depromeet/piki/common/outbox/OutboxJpaRepository.kt

@@ -0,0 +1,47 @@
+package com.depromeet.piki.common.outbox
+
+import jakarta.persistence.LockModeType
+import jakarta.persistence.QueryHint
+import org.springframework.data.domain.Limit
+import org.springframework.data.jpa.repository.JpaRepository
+import org.springframework.data.jpa.repository.Lock
+import org.springframework.data.jpa.repository.Query
+import org.springframework.data.jpa.repository.QueryHints
+import org.springframework.data.repository.NoRepositoryBean
+import org.springframework.data.repository.query.Param
+import java.time.LocalDateTime
+
+// Hibernate LockOptions.SKIP_LOCKED. 동시에 집는 디스패처가 서로를 기다리지 않게 한다(#770 교착).
+private const val SKIP_LOCKED = "-2"
+
+@NoRepositoryBean
+interface OutboxJpaRepository<T : Outbox> : JpaRepository<T, Long> {
+    @Lock(LockModeType.PESSIMISTIC_WRITE)
+    @QueryHints(QueryHint(name = "jakarta.persistence.lock.timeout", value = SKIP_LOCKED))
+    @Query(
+        "select o from #{#entityName} o where o.status = com.depromeet.piki.common.outbox.OutboxStatus.PENDING " +
+            "and o.deletedAt is null order by o.createdAt asc, o.id asc",
+    )
+    fun findPendingForUpdate(limit: Limit): List<T>
+
+    // 집을 때 updated_at 이 갱신되므로 집은 시각으로 읽음. 처리 중에 행을 고치면 그만큼 회수가 늦어짐
+    @Lock(LockModeType.PESSIMISTIC_WRITE)
+    @QueryHints(QueryHint(name = "jakarta.persistence.lock.timeout", value = SKIP_LOCKED))
+    @Query(
+        "select o from #{#entityName} o where o.status = com.depromeet.piki.common.outbox.OutboxStatus.PROCESSING " +
+            "and o.updatedAt < :claimedBefore and o.deletedAt is null order by o.updatedAt asc, o.id asc",
+    )
+    fun findStuckForUpdate(
+        @Param("claimedBefore") claimedBefore: LocalDateTime,
+        limit: Limit,
+    ): List<T>
+
+    @Lock(LockModeType.PESSIMISTIC_WRITE)
+    @Query("select o from #{#entityName} o where o.id = :id and o.deletedAt is null")
+    fun findByIdForUpdate(
+        @Param("id") id: Long,
+    ): T?
+}
+
+fun <T : Outbox> OutboxJpaRepository<T>.findHeldForUpdate(attempt: OutboxAttempt): T? =
+    findByIdForUpdate(attempt.outboxId)?.takeIf { it.isHeldBy(attempt) }

src/main/kotlin/com/depromeet/piki/common/outbox/OutboxPolicy.kt

@@ -0,0 +1,18 @@
+package com.depromeet.piki.common.outbox
+
+import java.time.Duration
+
+data class OutboxPolicy(
+    val stuckAfter: Duration,
+    val maxRecoveries: Int,
+    val processingLimit: Duration,
+) {
+    init {
+        require(stuckAfter.isPositive) { "회수 기준($stuckAfter)은 양수여야 한다" }
+        require(maxRecoveries >= 0) { "회수 상한($maxRecoveries)은 0 이상이어야 한다" }
+        // 처리 중에는 행을 잠그지 않아 처리 상한이 회수 기준보다 길면 살아 있는 워커의 요청을 회수함
+        require(processingLimit < stuckAfter) {
+            "처리 상한(${processingLimit.toMillis()}ms)은 회수 기준(${stuckAfter.toMillis()}ms)보다 짧아야 한다"
+        }
+    }
+}

src/main/kotlin/com/depromeet/piki/common/outbox/OutboxQueue.kt

@@ -0,0 +1,56 @@
+package com.depromeet.piki.common.outbox
+
+import org.springframework.data.domain.Limit
+import org.springframework.transaction.PlatformTransactionManager
+import org.springframework.transaction.TransactionDefinition
+import org.springframework.transaction.support.TransactionSynchronizationManager
+import org.springframework.transaction.support.TransactionTemplate
+import java.time.LocalDateTime
+
+// 집기·회수는 final 이라 하위 클래스가 RC 를 빼지 못함. 대신 하위 클래스에 프록시를 거는 어노테이션을 붙이면 필드 없는 프록시에서 돌아 NPE
+abstract class OutboxQueue<T : Outbox>(
+    private val jpaRepository: OutboxJpaRepository<T>,
+    transactionManager: PlatformTransactionManager,
+    val policy: OutboxPolicy,
+) {
+    // RR 이면 잠금 읽기가 status 인덱스의 gap 까지 잠가 새 PENDING insert 가 막힘(실측 #1261)
+    private val readCommitted =
+        TransactionTemplate(transactionManager).apply {
+            isolationLevel = TransactionDefinition.ISOLATION_READ_COMMITTED
+        }
+
+    // 회수 트랜잭션 안에서 돈다
+    protected abstract fun onExhausted(outbox: T)
+
+    fun claim(limit: Int): List<OutboxAttempt> =
+        inReadCommitted {
+            jpaRepository.findPendingForUpdate(Limit.of(limit)).map { outbox ->
+                outbox.claim()
+                outbox.attempt()
+            }
+        }
+
+    fun requeue(attempt: OutboxAttempt) =
+        inReadCommitted {
+            jpaRepository.findHeldForUpdate(attempt)?.requeue(attempt)
+            Unit
+        }
+
+    fun recoverStuck(limit: Int): List<OutboxRecovery> =
+        inReadCommitted {
+            jpaRepository.findStuckForUpdate(LocalDateTime.now().minus(policy.stuckAfter), Limit.of(limit)).map { outbox ->
+                val outcome = outbox.recover(policy.maxRecoveries)
+                when (outcome) {
+                    RecoveryOutcome.REQUEUED -> Unit
+                    RecoveryOutcome.EXHAUSTED -> onExhausted(outbox)
+                }
+                OutboxRecovery(outbox.getId(), outcome)
+            }
+        }
+
+    // 바깥 트랜잭션에 합류하면 Spring 이 isolation 설정을 무시함
+    private fun <R : Any> inReadCommitted(block: () -> R): R {
+        check(!TransactionSynchronizationManager.isActualTransactionActive()) { "아웃박스 집기·되돌리기·회수는 바깥 트랜잭션 안에서 부를 수 없다" }
+        return readCommitted.execute { block() }
+    }
+}

src/main/kotlin/com/depromeet/piki/common/outbox/OutboxRecovery.kt

@@ -0,0 +1,6 @@
+package com.depromeet.piki.common.outbox
+
+data class OutboxRecovery(
+    val outboxId: Long,
+    val outcome: RecoveryOutcome,
+)

src/main/kotlin/com/depromeet/piki/common/outbox/OutboxRepository.kt

@@ -0,0 +1,8 @@
+package com.depromeet.piki.common.outbox
+
+// 집기·회수 조회는 RC 를 거치도록 OutboxQueue 만 쓰게 하고 여기 열지 않음
+interface OutboxRepository<T : Outbox> {
+    fun save(outbox: T): T
+
+    fun findHeldForUpdate(attempt: OutboxAttempt): T?
+}

src/main/kotlin/com/depromeet/piki/common/outbox/OutboxRepositoryAdapter.kt

@@ -0,0 +1,10 @@
+package com.depromeet.piki.common.outbox
+
+// 구현체의 @Repository 예외 변환 프록시(CGLIB)가 가로채야 하므로 final 로 막지 않음
+abstract class OutboxRepositoryAdapter<T : Outbox>(
+    private val jpaRepository: OutboxJpaRepository<T>,
+) : OutboxRepository<T> {
+    override fun save(outbox: T): T = jpaRepository.save(outbox)
+
+    override fun findHeldForUpdate(attempt: OutboxAttempt): T? = jpaRepository.findHeldForUpdate(attempt)
+}

src/main/kotlin/com/depromeet/piki/common/outbox/OutboxStatus.kt

@@ -0,0 +1,10 @@
+package com.depromeet.piki.common.outbox
+
+enum class OutboxStatus(
+    val description: String,
+) {
+    PENDING("처리 전"),
+    PROCESSING("처리 중"),
+    SUCCEEDED("처리 성공"),
+    FAILED("처리 실패"),
+}

src/main/kotlin/com/depromeet/piki/item/domain/RecoveryOutcome.kt → src/main/kotlin/com/depromeet/piki/common/outbox/RecoveryOutcome.kt

@@ -1,8 +1,8 @@
-package com.depromeet.piki.item.domain
+package com.depromeet.piki.common.outbox
 
 enum class RecoveryOutcome(
     val description: String,
 ) {
-    REQUEUED("대기로 되돌려 다시 파싱"),
+    REQUEUED("대기로 되돌려 다시 처리"),
     EXHAUSTED("회수 상한에 닿아 실패로 닫음"),
 }

src/main/kotlin/com/depromeet/piki/item/domain/ItemParseOutbox.kt

@@ -1,6 +1,6 @@
 package com.depromeet.piki.item.domain
 
-import com.depromeet.piki.common.domain.LongBaseEntity
+import com.depromeet.piki.common.outbox.Outbox
 import com.depromeet.piki.product.domain.ParseInputType
 import com.depromeet.piki.product.domain.ProductLink
 import jakarta.persistence.Column
@@ -35,60 +35,14 @@ class ItemParseOutbox(
     @Column(length = ProductLink.MAX_LENGTH)
     val normalizedInput: NormalizedInput?,
 
-) : LongBaseEntity() {
-    @Enumerated(EnumType.STRING)
-    @Column(nullable = false, length = 16)
-    var status: ItemParseOutboxStatus = ItemParseOutboxStatus.PENDING
-        protected set
-
-    @Column(nullable = false)
-    var recoveryCount: Int = 0
-        protected set
-
+) : Outbox() {
     val slot: ItemSlot get() = ItemSlot(slotType, slotId)
 
     val input: ParseInput get() = ParseInput.of(inputType, inputValue)
 
     fun isRefresh(): Boolean = purpose == ParsePurpose.REFRESH
 
-    fun attempt(): ParseAttempt = ParseAttempt(getId(), recoveryCount)
-
-    fun isHeldBy(attempt: ParseAttempt): Boolean = status == ItemParseOutboxStatus.PROCESSING && recoveryCount == attempt.recoveryCount
-
-    fun isInProgress(): Boolean = status == ItemParseOutboxStatus.PENDING || status == ItemParseOutboxStatus.PROCESSING
-
-    fun isFailed(): Boolean = status == ItemParseOutboxStatus.FAILED
-
-    fun claim() = transit(from = ItemParseOutboxStatus.PENDING, to = ItemParseOutboxStatus.PROCESSING)
-
-    fun requeue() = transit(from = ItemParseOutboxStatus.PROCESSING, to = ItemParseOutboxStatus.PENDING)
-
-    fun succeed() = transit(from = ItemParseOutboxStatus.PROCESSING, to = ItemParseOutboxStatus.SUCCEEDED)
-
-    fun fail() = transit(from = ItemParseOutboxStatus.PROCESSING, to = ItemParseOutboxStatus.FAILED)
-
-    // 매번 멈추는 요청이 영원히 다시 파싱되지 않게 회수 횟수에 상한을 둠
-    fun recover(): RecoveryOutcome {
-        if (recoveryCount >= MAX_RECOVERIES) {
-            fail()
-            return RecoveryOutcome.EXHAUSTED
-        }
-        transit(from = ItemParseOutboxStatus.PROCESSING, to = ItemParseOutboxStatus.PENDING)
-        recoveryCount++
-        return RecoveryOutcome.REQUEUED
-    }
-
-    private fun transit(
-        from: ItemParseOutboxStatus,
-        to: ItemParseOutboxStatus,
-    ) {
-        check(status == from) { "item parse outbox ${getIdOrNull()} 은 $status 라 $to 로 갈 수 없다" }
-        status = to
-    }
-
     companion object {
-        private const val MAX_RECOVERIES = 1
-
         fun of(
             slot: ItemSlot,
             purpose: ParsePurpose,

src/main/kotlin/com/depromeet/piki/item/repository/ItemParseOutboxJpaRepository.kt

@@ -1,49 +1,14 @@
 package com.depromeet.piki.item.repository
 
+import com.depromeet.piki.common.outbox.OutboxJpaRepository
 import com.depromeet.piki.item.domain.ItemParseOutbox
 import com.depromeet.piki.item.domain.ItemSlotType
-import jakarta.persistence.LockModeType
-import jakarta.persistence.QueryHint
-import org.springframework.data.domain.Limit
-import org.springframework.data.jpa.repository.JpaRepository
-import org.springframework.data.jpa.repository.Lock
 import org.springframework.data.jpa.repository.Query
-import org.springframework.data.jpa.repository.QueryHints
 import org.springframework.data.repository.query.Param
-import java.time.LocalDateTime
-
-// Hibernate LockOptions.SKIP_LOCKED. 동시에 집는 디스패처가 서로를 기다리지 않게 한다(#770 교착).
-private const val SKIP_LOCKED = "-2"
-
-interface ItemParseOutboxJpaRepository : JpaRepository<ItemParseOutbox, Long> {
-    @Lock(LockModeType.PESSIMISTIC_WRITE)
-    @QueryHints(QueryHint(name = "jakarta.persistence.lock.timeout", value = SKIP_LOCKED))
-    @Query(
-        "select o from ItemParseOutbox o where o.status = com.depromeet.piki.item.domain.ItemParseOutboxStatus.PENDING " +
-            "and o.deletedAt is null order by o.createdAt asc, o.id asc",
-    )
-    fun findPendingForUpdate(limit: Limit): List<ItemParseOutbox>
-
-    // 집을 때 updated_at 이 갱신되고 처리 중에는 바뀌지 않으므로 집은 시각으로 읽음
-    @Lock(LockModeType.PESSIMISTIC_WRITE)
-    @QueryHints(QueryHint(name = "jakarta.persistence.lock.timeout", value = SKIP_LOCKED))
-    @Query(
-        "select o from ItemParseOutbox o where o.status = com.depromeet.piki.item.domain.ItemParseOutboxStatus.PROCESSING " +
-            "and o.updatedAt < :claimedBefore and o.deletedAt is null order by o.updatedAt asc, o.id asc",
-    )
-    fun findStuckForUpdate(
-        @Param("claimedBefore") claimedBefore: LocalDateTime,
-        limit: Limit,
-    ): List<ItemParseOutbox>
 
+interface ItemParseOutboxJpaRepository : OutboxJpaRepository<ItemParseOutbox> {
     fun findByIdAndDeletedAtIsNull(id: Long): ItemParseOutbox?
 
-    @Lock(LockModeType.PESSIMISTIC_WRITE)
-    @Query("select o from ItemParseOutbox o where o.id = :id and o.deletedAt is null")
-    fun findByIdForUpdate(
-        @Param("id") id: Long,
-    ): ItemParseOutbox?
-
     @Query(
         "select o from ItemParseOutbox o where o.slotType = :type and o.slotId in :ids and o.deletedAt is null " +
             "and o.id = (select max(o2.id) from ItemParseOutbox o2 where o2.slotType = o.slotType " +