fix(language-detection): 최초 감지 캐시 경합을 직렬화한다

src/main/kotlin/kr/co/vividnext/sodalive/content/LanguageDetectionCacheService.kt

@@ -25,18 +25,22 @@ class LanguageDetectionCacheService(
         )
         if (cached != null) return cached.detectedLanguage
 
-        val detectedLanguage = detector()?.takeIf { it.isNotBlank() } ?: return null
+        val detectedLanguage = detector()?.takeIf { it.isNotBlank() }?.lowercase() ?: return null
-        languageDetectionResultRepository.save(
+        languageDetectionResultRepository.insertIfAbsent(
-            LanguageDetectionResult(
             sourceHash = sourceHash,
             sourceTextSample = normalizedQuery.take(MAX_SAMPLE_LENGTH),
-                detectedLanguage = detectedLanguage.lowercase(),
+            detectedLanguage = detectedLanguage,
             provider = provider,
             confidence = null,
             normalizationVersion = SourceTextNormalizer.NORMALIZATION_VERSION
         )
+        return requireNotNull(
+            languageDetectionResultRepository.findDetectedLanguageForUpdate(
+                sourceHash = sourceHash,
+                provider = provider,
+                normalizationVersion = SourceTextNormalizer.NORMALIZATION_VERSION
+            )
         )
-        return detectedLanguage.lowercase()
     }
 
     companion object {

src/main/kotlin/kr/co/vividnext/sodalive/content/LanguageDetectionResultRepository.kt

@@ -1,6 +1,9 @@
 package kr.co.vividnext.sodalive.content
 
 import org.springframework.data.jpa.repository.JpaRepository
+import org.springframework.data.jpa.repository.Modifying
+import org.springframework.data.jpa.repository.Query
+import org.springframework.data.repository.query.Param
 
 interface LanguageDetectionResultRepository : JpaRepository<LanguageDetectionResult, Long> {
     fun findBySourceHashAndProviderAndNormalizationVersion(
@@ -8,4 +11,55 @@ interface LanguageDetectionResultRepository : JpaRepository<LanguageDetectionRes
         provider: String,
         normalizationVersion: String
     ): LanguageDetectionResult?
+
+    @Modifying
+    @Query(
+        value = """
+            INSERT INTO language_detection_result (
+                source_hash,
+                source_text_sample,
+                detected_language,
+                provider,
+                confidence,
+                normalization_version,
+                created_at,
+                updated_at
+            ) VALUES (
+                :sourceHash,
+                :sourceTextSample,
+                :detectedLanguage,
+                :provider,
+                :confidence,
+                :normalizationVersion,
+                CURRENT_TIMESTAMP,
+                CURRENT_TIMESTAMP
+            ) ON DUPLICATE KEY UPDATE id = id
+        """,
+        nativeQuery = true
+    )
+    fun insertIfAbsent(
+        @Param("sourceHash") sourceHash: String,
+        @Param("sourceTextSample") sourceTextSample: String,
+        @Param("detectedLanguage") detectedLanguage: String,
+        @Param("provider") provider: String,
+        @Param("confidence") confidence: Double?,
+        @Param("normalizationVersion") normalizationVersion: String
+    ): Int
+
+    @Query(
+        value = """
+            SELECT detected_language
+            FROM language_detection_result
+            WHERE source_hash = :sourceHash
+              AND provider = :provider
+              AND normalization_version = :normalizationVersion
+            FOR UPDATE
+        """,
+        nativeQuery = true
+    )
+    fun findDetectedLanguageForUpdate(
+        @Param("sourceHash") sourceHash: String,
+        @Param("provider") provider: String,
+        @Param("normalizationVersion") normalizationVersion: String
+    ): String?
 }

src/test/kotlin/kr/co/vividnext/sodalive/content/CreatorCommunityLanguageDetectionCacheConcurrencyTest.kt

@@ -0,0 +1,336 @@
+package kr.co.vividnext.sodalive.content
+
+import kr.co.vividnext.sodalive.configs.QueryDslConfig
+import kr.co.vividnext.sodalive.content.theme.AudioContentThemeQueryRepository
+import kr.co.vividnext.sodalive.explorer.profile.creatorCommunity.CreatorCommunity
+import kr.co.vividnext.sodalive.explorer.profile.creatorCommunity.CreatorCommunityRepository
+import kr.co.vividnext.sodalive.i18n.translation.LanguageTranslationTargetType
+import kr.co.vividnext.sodalive.i18n.translation.ResourceTranslationJobScheduler
+import kr.co.vividnext.sodalive.i18n.translation.SourceTextNormalizer
+import kr.co.vividnext.sodalive.i18n.translation.TranslationJobRepository
+import kr.co.vividnext.sodalive.i18n.translation.TranslationJobScheduler
+import kr.co.vividnext.sodalive.i18n.translation.TranslationJobStatus
+import kr.co.vividnext.sodalive.i18n.translation.TranslationReadModelMaterializer
+import kr.co.vividnext.sodalive.i18n.translation.TranslationSourceExtractor
+import kr.co.vividnext.sodalive.member.Member
+import kr.co.vividnext.sodalive.member.MemberRepository
+import kr.co.vividnext.sodalive.member.MemberRole
+import kr.co.vividnext.sodalive.v2.creator.channel.community.translation.application.CreatorCommunityTranslationService
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.DisplayName
+import org.junit.jupiter.api.Test
+import org.springframework.aop.support.AopUtils
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.boot.test.autoconfigure.jdbc.AutoConfigureTestDatabase
+import org.springframework.boot.test.autoconfigure.orm.jpa.DataJpaTest
+import org.springframework.context.annotation.Import
+import org.springframework.http.HttpHeaders
+import org.springframework.http.HttpMethod
+import org.springframework.http.HttpStatus
+import org.springframework.http.MediaType
+import org.springframework.http.client.ClientHttpRequest
+import org.springframework.http.client.ClientHttpRequestFactory
+import org.springframework.http.client.ClientHttpResponse
+import org.springframework.test.util.AopTestUtils
+import org.springframework.test.util.ReflectionTestUtils
+import org.springframework.transaction.PlatformTransactionManager
+import org.springframework.transaction.annotation.Propagation
+import org.springframework.transaction.annotation.Transactional
+import org.springframework.transaction.support.TransactionTemplate
+import org.springframework.web.client.RestTemplate
+import java.io.ByteArrayInputStream
+import java.io.ByteArrayOutputStream
+import java.io.OutputStream
+import java.net.URI
+import java.util.UUID
+import java.util.concurrent.Callable
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.ExecutorCompletionService
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicInteger
+
+@DataJpaTest(
+    properties = [
+        "spring.cache.type=none",
+        "spring.datasource.url=" +
+            "jdbc:h2:mem:creator-community-language-detection-cache-concurrency;" +
+            "MODE=MySQL;NON_KEYWORDS=VALUE;DB_CLOSE_ON_EXIT=FALSE",
+        "cloud.naver.papago-client-id=test-client-id",
+        "cloud.naver.papago-client-secret=test-client-secret"
+    ]
+)
+@AutoConfigureTestDatabase(replace = AutoConfigureTestDatabase.Replace.NONE)
+@Import(
+    QueryDslConfig::class,
+    AudioContentThemeQueryRepository::class,
+    LanguageDetectionCacheService::class,
+    LanguageDetectListener::class,
+    CreatorCommunityTranslationService::class,
+    TranslationSourceExtractor::class,
+    TranslationReadModelMaterializer::class,
+    ResourceTranslationJobScheduler::class,
+    TranslationJobScheduler::class
+)
+@Transactional(propagation = Propagation.NOT_SUPPORTED)
+class CreatorCommunityLanguageDetectionCacheConcurrencyTest @Autowired constructor(
+    private val creatorCommunityRepository: CreatorCommunityRepository,
+    private val memberRepository: MemberRepository,
+    private val languageDetectionResultRepository: LanguageDetectionResultRepository,
+    private val translationJobRepository: TranslationJobRepository,
+    private val languageDetectionCacheService: LanguageDetectionCacheService,
+    private val languageDetectListener: LanguageDetectListener,
+    transactionManager: PlatformTransactionManager
+) {
+    private val requiresNewTransactionTemplate = TransactionTemplate(transactionManager).apply {
+        propagationBehavior = Propagation.REQUIRES_NEW.value()
+    }
+    private lateinit var papagoRequests: ControlledPapagoRequestFactory
+
+    @BeforeEach
+    fun setUp() {
+        papagoRequests = ControlledPapagoRequestFactory()
+        val listener: LanguageDetectListener = AopTestUtils.getTargetObject(languageDetectListener)
+        ReflectionTestUtils.setField(listener, "restTemplate", RestTemplate(papagoRequests))
+    }
+
+    @Test
+    @DisplayName("최초 감지 캐시 저장 경합에서도 두 요청 언어의 번역 작업을 보존한다")
+    fun shouldPreserveBothTargetsWhenInitialDetectionCacheWritesRace() {
+        val content = "initial detection cache race ${UUID.randomUUID()}"
+        val contentRevision = 7L
+        val postId = persistPost(content, contentRevision)
+        val executor = Executors.newFixedThreadPool(2)
+        val completions = ExecutorCompletionService<Unit>(executor)
+
+        try {
+            assertTrue(AopUtils.isAopProxy(languageDetectionCacheService))
+            completions.submit(detectionTask(postId, content, contentRevision, "ja"))
+            completions.submit(detectionTask(postId, content, contentRevision, "en"))
+
+            assertTrue(
+                papagoRequests.firstProviderEntry.await(5, TimeUnit.SECONDS),
+                "first cache miss did not reach the provider boundary"
+            )
+            assertTrue(
+                papagoRequests.secondProviderEntry.await(5, TimeUnit.SECONDS),
+                "second cache miss did not reach the provider boundary"
+            )
+
+            papagoRequests.firstProviderRelease.countDown()
+            val firstCompletion = completions.poll(5, TimeUnit.SECONDS)
+                ?: throw AssertionError("first detection transaction did not complete")
+            firstCompletion.get(5, TimeUnit.SECONDS)
+
+            papagoRequests.secondProviderRelease.countDown()
+            val secondCompletion = completions.poll(5, TimeUnit.SECONDS)
+                ?: throw AssertionError("second detection transaction did not complete")
+            secondCompletion.get(5, TimeUnit.SECONDS)
+
+            assertCacheRow(content)
+            assertPostAndJobs(postId, content, contentRevision)
+        } finally {
+            papagoRequests.releaseAll()
+            executor.shutdownNow()
+            assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS))
+        }
+    }
+
+    @Test
+    @DisplayName("서로 다른 게시물의 동일 본문 최초 감지 경합에서도 각각의 전체 번역 작업을 보존한다")
+    fun shouldPreserveAllNonSourceTargetsForDifferentPostsWithSameContentDuringInitialDetectionRace() {
+        val content = "shared initial detection cache race ${UUID.randomUUID()}"
+        val contentRevision = 7L
+        val firstPostId = persistPost(content, contentRevision)
+        val secondPostId = persistPost(content, contentRevision)
+        val executor = Executors.newFixedThreadPool(2)
+        val completions = ExecutorCompletionService<Unit>(executor)
+
+        try {
+            assertTrue(AopUtils.isAopProxy(languageDetectionCacheService))
+            completions.submit(detectionTask(firstPostId, content, contentRevision, null))
+            completions.submit(detectionTask(secondPostId, content, contentRevision, null))
+
+            assertTrue(
+                papagoRequests.firstProviderEntry.await(5, TimeUnit.SECONDS),
+                "first cache miss did not reach the provider boundary"
+            )
+            assertTrue(
+                papagoRequests.secondProviderEntry.await(5, TimeUnit.SECONDS),
+                "second cache miss did not reach the provider boundary"
+            )
+
+            papagoRequests.firstProviderRelease.countDown()
+            val firstCompletion = completions.poll(5, TimeUnit.SECONDS)
+                ?: throw AssertionError("first detection transaction did not complete")
+            firstCompletion.get(5, TimeUnit.SECONDS)
+
+            papagoRequests.secondProviderRelease.countDown()
+            val secondCompletion = completions.poll(5, TimeUnit.SECONDS)
+                ?: throw AssertionError("second detection transaction did not complete")
+            secondCompletion.get(5, TimeUnit.SECONDS)
+
+            assertCacheRow(content)
+            assertPostAndJobs(firstPostId, content, contentRevision)
+            assertPostAndJobs(secondPostId, content, contentRevision)
+        } finally {
+            papagoRequests.releaseAll()
+            executor.shutdownNow()
+            assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS))
+        }
+    }
+
+    private fun detectionTask(
+        postId: Long,
+        content: String,
+        contentRevision: Long,
+        targetLanguage: String?
+    ): Callable<Unit> {
+        return Callable {
+            val listener: LanguageDetectListener = AopTestUtils.getTargetObject(languageDetectListener)
+            requiresNewTransactionTemplate.executeWithoutResult {
+                listener.detectLanguage(
+                    LanguageDetectEvent(
+                        id = postId,
+                        query = content,
+                        targetType = LanguageDetectTargetType.CREATOR_COMMUNITY,
+                        sourceRevision = contentRevision,
+                        targetLanguage = targetLanguage
+                    )
+                )
+            }
+        }
+    }
+
+    private fun persistPost(content: String, contentRevision: Long): Long {
+        return requiresNewTransactionTemplate.execute {
+            val member = memberRepository.save(
+                Member(
+                    password = "password",
+                    nickname = "cache-race-${UUID.randomUUID()}",
+                    role = MemberRole.CREATOR
+                )
+            )
+            creatorCommunityRepository.save(
+                CreatorCommunity(
+                    content = content,
+                    price = 0,
+                    isCommentAvailable = true,
+                    isAdult = false,
+                    languageCode = null,
+                    contentRevision = contentRevision
+                ).apply {
+                    this.member = member
+                }
+            ).id!!
+        }!!
+    }
+
+    private fun assertCacheRow(content: String) {
+        val rows = requiresNewTransactionTemplate.execute {
+            languageDetectionResultRepository.findAll().filter {
+                it.sourceHash == SourceTextNormalizer.hash(content) &&
+                    it.provider == LanguageDetectionCacheService.DEFAULT_PROVIDER &&
+                    it.normalizationVersion == SourceTextNormalizer.NORMALIZATION_VERSION
+            }
+        }.orEmpty()
+
+        assertEquals(1, rows.size)
+        assertEquals("ko", rows.single().detectedLanguage)
+    }
+
+    private fun assertPostAndJobs(postId: Long, content: String, contentRevision: Long) {
+        val post = requiresNewTransactionTemplate.execute {
+            creatorCommunityRepository.findById(postId).orElseThrow()
+        }!!
+        val jobs = requiresNewTransactionTemplate.execute {
+            translationJobRepository.findAll().filter {
+                it.resourceType == LanguageTranslationTargetType.CREATOR_COMMUNITY &&
+                    it.resourceId == postId &&
+                    it.fieldKey == "content"
+            }
+        }.orEmpty()
+
+        assertEquals(content, post.content)
+        assertEquals(contentRevision, post.contentRevision)
+        assertEquals("ko", post.languageCode)
+        assertEquals(2, jobs.size)
+        assertEquals(setOf("ja", "en"), jobs.map { it.targetLanguage }.toSet())
+        assertTrue(
+            jobs.all { job ->
+                job.sourceHash == SourceTextNormalizer.hash(content) &&
+                    job.sourceLanguage == "ko"
+            }
+        )
+        assertTrue(jobs.all { it.status == TranslationJobStatus.PENDING })
+    }
+
+    private class ControlledPapagoRequestFactory : ClientHttpRequestFactory {
+        private val requestOrder = AtomicInteger()
+        val firstProviderEntry = CountDownLatch(1)
+        val secondProviderEntry = CountDownLatch(1)
+        val firstProviderRelease = CountDownLatch(1)
+        val secondProviderRelease = CountDownLatch(1)
+
+        override fun createRequest(uri: URI, httpMethod: HttpMethod): ClientHttpRequest {
+            return object : ClientHttpRequest {
+                private val headers = HttpHeaders()
+                private val body = ByteArrayOutputStream()
+
+                override fun getMethodValue(): String = httpMethod.name
+
+                override fun getURI(): URI = uri
+
+                override fun getHeaders(): HttpHeaders = headers
+
+                override fun getBody(): OutputStream = body
+
+                override fun execute(): ClientHttpResponse {
+                    val languageCode = when (requestOrder.incrementAndGet()) {
+                        1 -> {
+                            firstProviderEntry.countDown()
+                            firstProviderRelease.await()
+                            "ko"
+                        }
+
+                        2 -> {
+                            secondProviderEntry.countDown()
+                            secondProviderRelease.await()
+                            "en"
+                        }
+
+                        else -> error("unexpected provider request")
+                    }
+                    return languageDetectionResponse(languageCode)
+                }
+            }
+        }
+
+        fun releaseAll() {
+            firstProviderRelease.countDown()
+            secondProviderRelease.countDown()
+        }
+
+        private fun languageDetectionResponse(languageCode: String): ClientHttpResponse {
+            return object : ClientHttpResponse {
+                private val headers = HttpHeaders().apply {
+                    contentType = MediaType.APPLICATION_JSON
+                }
+
+                override fun getStatusCode(): HttpStatus = HttpStatus.OK
+
+                override fun getRawStatusCode(): Int = HttpStatus.OK.value()
+
+                override fun getStatusText(): String = "OK"
+
+                override fun getHeaders(): HttpHeaders = headers
+
+                override fun getBody() = ByteArrayInputStream("{\"langCode\":\"$languageCode\"}".toByteArray())
+
+                override fun close() = Unit
+            }
+        }
+    }
+}

src/test/kotlin/kr/co/vividnext/sodalive/content/LanguageDetectionCacheServiceTest.kt

@@ -2,6 +2,7 @@ package kr.co.vividnext.sodalive.content
 
 import kr.co.vividnext.sodalive.i18n.translation.SourceTextNormalizer
 import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.DisplayName
 import org.junit.jupiter.api.Test
 import org.mockito.Mockito
 
@@ -37,6 +38,67 @@ class LanguageDetectionCacheServiceTest {
 
         assertEquals("en", detected)
         assertEquals(0, providerCalls)
-        Mockito.verify(repository, Mockito.never()).save(Mockito.any(LanguageDetectionResult::class.java))
+        Mockito.verify(repository).findBySourceHashAndProviderAndNormalizationVersion(
+            sourceHash = sourceHash,
+            provider = "papago",
+            normalizationVersion = SourceTextNormalizer.NORMALIZATION_VERSION
+        )
+        Mockito.verifyNoMoreInteractions(repository)
+    }
+
+    @Test
+    @DisplayName("최초 캐시 저장 경합에서는 저장된 승자의 감지 언어를 반환한다")
+    fun shouldReturnStoredWinnerAfterInitialCacheInsert() {
+        val repository = Mockito.mock(LanguageDetectionResultRepository::class.java)
+        val service = LanguageDetectionCacheService(repository)
+        val sourceHash = SourceTextNormalizer.hash("Hello world")
+
+        Mockito.`when`(
+            repository.findBySourceHashAndProviderAndNormalizationVersion(
+                sourceHash = sourceHash,
+                provider = "papago",
+                normalizationVersion = SourceTextNormalizer.NORMALIZATION_VERSION
+            )
+        ).thenReturn(null)
+        Mockito.`when`(
+            repository.insertIfAbsent(
+                sourceHash = sourceHash,
+                sourceTextSample = "Hello world",
+                detectedLanguage = "en",
+                provider = "papago",
+                confidence = null,
+                normalizationVersion = SourceTextNormalizer.NORMALIZATION_VERSION
+            )
+        ).thenReturn(0)
+        Mockito.`when`(
+            repository.findDetectedLanguageForUpdate(
+                sourceHash = sourceHash,
+                provider = "papago",
+                normalizationVersion = SourceTextNormalizer.NORMALIZATION_VERSION
+            )
+        ).thenReturn("ko")
+
+        val detected = service.detectWithCache("Hello world") { "en" }
+
+        assertEquals("ko", detected)
+        Mockito.verify(repository).findBySourceHashAndProviderAndNormalizationVersion(
+            sourceHash = sourceHash,
+            provider = "papago",
+            normalizationVersion = SourceTextNormalizer.NORMALIZATION_VERSION
+        )
+        Mockito.verify(repository).insertIfAbsent(
+            sourceHash = sourceHash,
+            sourceTextSample = "Hello world",
+            detectedLanguage = "en",
+            provider = "papago",
+            confidence = null,
+            normalizationVersion = SourceTextNormalizer.NORMALIZATION_VERSION
+        )
+        Mockito.verify(repository).findDetectedLanguageForUpdate(
+            sourceHash = sourceHash,
+            provider = "papago",
+            normalizationVersion = SourceTextNormalizer.NORMALIZATION_VERSION
+        )
+        Mockito.verifyNoMoreInteractions(repository)
     }
 }