feat: Transaction이 종료(커밋, 롤백) 되면 transaction retry watch에서 제거

README.md

@@ -40,16 +40,16 @@ class Application {
 
 #### Properties
 
-| key                     | example   | description                                                                                                                                        |
-|-------------------------|-----------|----------------------------------------------------------------------------------------------------------------------------------------------------|
-| **netx.mode**           | redis     | 트랜잭션 관리에 사용할 메시지 큐 구현체의 mode 입니다.                                                                                                                  |
-| **netx.host**           | localhost | 트랜잭션 관리에 사용할 메시지 큐 의 host url 입니다. (ex. redis host)                                                                                                |
-| **netx.port**           | 6379      | 트랜잭션 관리에 사용할 메시지 큐의 port 입니다.                                                                                                                      |
-| **netx.group**          | pay-group | 분산 노드의 그룹입니다. 트랜잭션 이벤트는 같은 그룹내 하나의 노드로만 전송됩니다.                                                                                                     |
-| **netx.node-id**        | 1         | id 생성에 사용될 식별자입니다. 모든 서버는 반드시 다른 id를 할당받아야 하며, 1~256 만큼의 id를 설정할 수 있습니다. _`중복된 id 생성을 방지하기위해 twitter snowflake 알고리즘으로 id를 생성합니다.`_                 |
-| **netx.node-name**      | pay-1     | _`netx.group`_ 에 참여할 서버의 이름입니다. 같은 그룹내에 중복된 이름이 존재하면 안됩니다.                                                                                         |
-| **netx.recovery-milli** | 60000     | _`netx.recovery-milli`_ 마다 _`netx.orphan-milli`_ 동안 처리 되지 않는 트랜잭션을 찾아 재실행합니다. 기본값은 60000(60초) 입니다.                                                 |
-| **netx.orphan-milli**   | 10000     | 트랜잭션이 PENDING 상태가 되었지만 orphan-milli가 지나도 ACK 상태가 되지 않는경우 다른 노드에게 처리를 위임합니다. 기본값은 10000(10초) 입니다.                 |
+| key                     | example | description                                                                                                               |
+|-------------------------|---------|---------------------------------------------------------------------------------------------------------------------------|
+| **netx.mode**           | redis   | 트랜잭션 관리에 사용할 메시지 큐 구현체의 mode 입니다.                                                                                         |
+| **netx.host**           | localhost | 트랜잭션 관리에 사용할 메시지 큐 의 host url 입니다. (ex. redis host)                                                                       |
+| **netx.port**           | 6379    | 트랜잭션 관리에 사용할 메시지 큐의 port 입니다.                                                                                             |
+| **netx.group**          | pay-group | 분산 노드의 그룹입니다. 트랜잭션 이벤트는 같은 그룹내 하나의 노드로만 전송됩니다.                                                                            |
+| **netx.node-id**        | 1       | id 생성에 사용될 식별자입니다. 모든 서버는 반드시 다른 id를 할당받아야 하며, 1~256 만큼의 id를 설정할 수 있습니다. _`중복된 id 생성을 방지하기위해 twitter snowflake 알고리즘으로 id를 생성합니다.`_ |
+| **netx.node-name**      | pay-1   | _`netx.group`_ 에 참여할 서버의 이름입니다. 같은 그룹내에 중복된 이름이 존재하면 안됩니다.                                                                |
+| **netx.recovery-milli** | 60000   | _`netx.recovery-milli`_ 마다 _`netx.orphan-milli`_ 동안 처리 되지 않는 트랜잭션을 찾아 재실행합니다. 기본값은 60000(60초) 입니다.                        |
+| **netx.orphan-milli**   | 10000   | 트랜잭션이 PENDING 상태가 되었지만 orphan-milli가 지나도 ACK 상태가 되지 않는경우 다른 노드에게 처리를 위임합니다. 기본값은 10000(10초) 입니다.                          |
 
 ### Usage example
 

src/main/kotlin/org/rooftop/netx/redis/ByteArrayRedisSerializer.kt

@@ -1,4 +1,4 @@
-package org.rooftop.pay.infra.transaction
+package org.rooftop.netx.redis
 
 import org.springframework.data.redis.serializer.RedisSerializer
 

src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionDispatcher.kt

@@ -38,14 +38,6 @@ class RedisStreamTransactionDispatcher(
                     StreamOffset.create(transactionId, ReadOffset.from(">"))
                 ).publishOn(Schedulers.parallel())
                     .map { Transaction.parseFrom(it.value["data"]?.toByteArray()) to it.id.value }
-                    .flatMap { (transaction, messageId) ->
-                        when (transaction.state) {
-                            TransactionState.TRANSACTION_STATE_ROLLBACK ->
-                                findOwnTransaction(transaction).map { it to messageId }
-
-                            else -> Mono.just(transaction to messageId)
-                        }
-                    }
             }
     }
 

src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionRemover.kt

@@ -0,0 +1,53 @@
+package org.rooftop.netx.redis
+
+import org.rooftop.netx.api.TransactionCommitEvent
+import org.rooftop.netx.api.TransactionRollbackEvent
+import org.springframework.context.event.EventListener
+import org.springframework.data.domain.Range
+import org.springframework.data.redis.core.ReactiveRedisTemplate
+import reactor.core.scheduler.Schedulers
+import reactor.util.retry.RetrySpec
+import java.time.Duration
+
+class RedisStreamTransactionRemover(
+    private val nodeGroup: String,
+    private val reactiveRedisTemplate: ReactiveRedisTemplate<String, ByteArray>,
+) {
+
+    @EventListener(TransactionCommitEvent::class)
+    fun handleTransactionCommitEvent(event: TransactionCommitEvent) {
+        deleteElastic(event.transactionId)
+    }
+
+    @EventListener(TransactionRollbackEvent::class)
+    fun handleTransactionRollbackEvent(event: TransactionRollbackEvent) {
+        deleteElastic(event.transactionId)
+    }
+
+    private fun deleteElastic(transactionId: String) {
+        reactiveRedisTemplate.opsForStream<String, String>()
+            .pending(transactionId, nodeGroup, Range.closed("-", "+"), Long.MAX_VALUE)
+            .filter {
+                when (it.get().toList().isEmpty()) {
+                    true -> true
+                    false -> error(TRANSACTION_IS_PENDING_STATUS)
+                }
+            }
+            .retryWhen(retryIfTransactionPending)
+            .flatMap {
+                reactiveRedisTemplate.opsForSet()
+                    .remove(nodeGroup, transactionId.toByteArray())
+            }
+            .subscribeOn(Schedulers.parallel())
+            .subscribe()
+    }
+
+    companion object {
+        private const val TRANSACTION_IS_PENDING_STATUS =
+            "Transaction message remains in pending status."
+        private val retryIfTransactionPending =
+            RetrySpec.fixedDelay(Long.MAX_VALUE, Duration.ofMillis(3000))
+                .jitter(1.0)
+                .filter { it.message == TRANSACTION_IS_PENDING_STATUS }
+    }
+}

src/main/kotlin/org/rooftop/netx/redis/RedisTransactionConfigurer.kt

@@ -4,7 +4,6 @@ import org.redisson.Redisson
 import org.redisson.api.RedissonReactiveClient
 import org.redisson.config.Config
 import org.rooftop.netx.api.TransactionManager
-import org.rooftop.pay.infra.transaction.ByteArrayRedisSerializer
 import org.springframework.beans.factory.annotation.Value
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
 import org.springframework.context.ApplicationEventPublisher
@@ -64,6 +63,14 @@ class RedisTransactionConfigurer(
             recoveryMilli = recoveryMilli,
         )
 
+    @Bean
+    @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis")
+    fun redisStreamTransactionDeleter(): RedisStreamTransactionRemover =
+        RedisStreamTransactionRemover(
+            nodeGroup = nodeGroup,
+            reactiveRedisTemplate = reactiveRedisTemplate(),
+        )
+
     @Bean
     @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis")
     fun reactiveRedisTemplate(): ReactiveRedisTemplate<String, ByteArray> {

src/main/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporter.kt

@@ -13,13 +13,14 @@ import reactor.core.scheduler.Schedulers
 import java.util.concurrent.TimeUnit
 
 class RedisTransactionRetrySupporter(
+    recoveryMilli: Long,
     private val nodeGroup: String,
     private val nodeName: String,
     private val reactiveRedisTemplate: ReactiveRedisTemplate<String, ByteArray>,
     private val redissonReactiveClient: RedissonReactiveClient,
     private val transactionDispatcher: AbstractTransactionDispatcher,
     private val orphanMilli: Long,
-    recoveryMilli: Long,
+    private val lockKey: String = "$nodeGroup-key",
 ) : AbstractTransactionRetrySupporter(recoveryMilli) {
 
     override fun watchTransaction(transactionId: String): Mono<String> {
@@ -41,7 +42,7 @@ class RedisTransactionRetrySupporter(
             .pending(transactionId, nodeGroup, Range.closed("-", "+"), Long.MAX_VALUE)
             .filter { it.get().toList().isNotEmpty() }
             .flatMap { pendingMessage ->
-                redissonReactiveClient.getLock("$nodeGroup-key")
+                redissonReactiveClient.getLock(lockKey)
                     .tryLock(0, orphanMilli, TimeUnit.MILLISECONDS)
                     .map { pendingMessage }
             }
@@ -55,12 +56,12 @@ class RedisTransactionRetrySupporter(
             }
             .map { Transaction.parseFrom(it.value["data"]?.toByteArray()) to it.id.toString() }
             .flatMap { transactionWithMessageId ->
-                redissonReactiveClient.getLock("$nodeGroup-key")
+                redissonReactiveClient.getLock(lockKey)
                     .forceUnlock()
                     .flatMapMany { Flux.just(transactionWithMessageId) }
             }
             .doOnError {
-                redissonReactiveClient.getLock("$nodeGroup-key")
+                redissonReactiveClient.getLock(lockKey)
                     .unlock()
                     .subscribeOn(Schedulers.parallel())
                     .subscribe()

src/test/kotlin/org/rooftop/netx/redis/NoAckRedisTransactionConfigurer.kt

@@ -4,7 +4,6 @@ import org.redisson.Redisson
 import org.redisson.api.RedissonReactiveClient
 import org.redisson.config.Config
 import org.rooftop.netx.api.TransactionManager
-import org.rooftop.pay.infra.transaction.ByteArrayRedisSerializer
 import org.springframework.beans.factory.annotation.Value
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
 import org.springframework.boot.test.context.TestConfiguration

src/test/kotlin/org/rooftop/netx/redis/RedisAssertions.kt

@@ -20,4 +20,14 @@ internal class RedisAssertions(
 
         pendingMessageCount shouldBe count
     }
+
+    fun retryTransactionShouldBeNotExists(transactionId: String) {
+        val retryTransaction = reactiveRedisOperations.opsForSet()
+            .members(nodeGroup)
+            .map { String(it) }
+            .any {  it == transactionId  }
+            .block()
+
+        retryTransaction shouldBe false
+    }
 }

src/test/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManagerTest.kt

@@ -129,10 +129,8 @@ internal class RedisStreamTransactionManagerTest(
                 transactionManager.rollback(transactionId, "rollback occured for test").block()
 
                 eventually(5.minutes) {
-                    eventCapture.capturedCount(TransactionRollbackEvent::class)
+                    eventCapture.capturedCount(TransactionRollbackEvent::class) shouldBe 1
                 }
-
-                Thread.sleep(10.minutes.inWholeMilliseconds)
             }
         }
 

src/test/kotlin/org/rooftop/netx/redis/RedisStreamTransactionRemoverTest.kt

@@ -0,0 +1,63 @@
+package org.rooftop.netx.redis
+
+import io.kotest.assertions.nondeterministic.eventually
+import io.kotest.core.annotation.DisplayName
+import io.kotest.core.spec.style.DescribeSpec
+import io.kotest.matchers.shouldBe
+import org.rooftop.netx.api.TransactionCommitEvent
+import org.rooftop.netx.api.TransactionManager
+import org.rooftop.netx.api.TransactionRollbackEvent
+import org.rooftop.netx.autoconfig.AutoConfigureDistributedTransaction
+import org.springframework.test.context.ContextConfiguration
+import org.springframework.test.context.TestPropertySource
+import reactor.core.scheduler.Schedulers
+import kotlin.time.Duration.Companion.seconds
+
+@AutoConfigureDistributedTransaction
+@ContextConfiguration(
+    classes = [
+        RedisContainer::class,
+        RedisAssertions::class,
+        EventCapture::class,
+    ]
+)
+@TestPropertySource("classpath:application.properties")
+@DisplayName("RedisStreamTransactionRemover 클래스의")
+internal class RedisStreamTransactionRemoverTest(
+    private val redisAssertions: RedisAssertions,
+    private val transactionManager: TransactionManager,
+    private val eventCapture: EventCapture,
+) : DescribeSpec({
+    describe("handleTransactionCommitEvent 메소드는") {
+        context("TransactionCommitEvent 가 발행되면,") {
+            val transactionId = transactionManager.start("RedisStreamTransactionRemoverTest")
+                .block()!!
+
+            it("Transaction 을 retry watch 대기열에서 삭제한다.") {
+                transactionManager.commit(transactionId)
+                    .subscribeOn(Schedulers.parallel())
+                    .subscribe()
+
+                eventually(10.seconds) {
+                    eventCapture.capturedCount(TransactionCommitEvent::class) shouldBe 1
+                    redisAssertions.retryTransactionShouldBeNotExists(transactionId)
+                }
+            }
+        }
+
+        context("TransactionRollbackEvent 가 발행되면,") {
+            val transactionId = transactionManager.start("RedisStreamTransactionRemoverTest")
+                .block()!!
+
+            it("Transaction 을 retry watch 대기열에서 삭제한다.") {
+                transactionManager.rollback(transactionId, "rollback occured for test").block()
+
+                eventually(10.seconds) {
+                    eventCapture.capturedCount(TransactionRollbackEvent::class) shouldBe 1
+                    redisAssertions.retryTransactionShouldBeNotExists(transactionId)
+                }
+            }
+        }
+    }
+}
+)