feat: 대기열 완성도 보강 - 전체 대기 인원 노출, 통합/컨슈머 테스트, 처리량 정책 일원화

8주차 체크리스트 보완:
- /queue/enter, /queue/position 응답에 totalWaiting(전체 대기 인원) 추가
- 대기열 Redis 통합 테스트 신설: 동시 진입 50명 무유실/고유 순번,
  동시 중복 진입 차단, ZPOPMIN 동시 dequeue 무중복, 토큰 TTL 실만료,
  배치 크기 초과 시 처리량 제어
- 배치 크기 산정 근거 문서화 (docs/design/06-waiting-queue.md)

9주차 검증 보완:
- commerce-streamer 첫 테스트: 컨슈머 단위 11개(가중치 라우팅, 멱등 스킵,
  부분 실패 ack) + Redis 통합 8개(가중치 계산, TTL 2일, carry-over 10%)
- 수제 Avro 스키마 대신 정본 avsc(AvroSchemaProvider) 기반 레코드 사용

엄격 리뷰 반영 (10건):
- 처리량 산정 입력값을 발제 예시(풀 50)에서 실측 설정(jpa.yml 풀 40)으로 정정.
  175 TPS/18명이 140 TPS/14명이 되고, QueueThroughput 정책 객체 하나에서
  BATCH_SIZE와 예상 대기시간을 파생시켜 3중 하드코딩 제거
- 예상 대기시간 내림을 올림으로 변경 (141~279위가 전부 1초로 과소 안내되던 문제)
- fixedDelay를 fixedRate로 변경 (설계 산식의 고정 주기 전제와 일치)
- position/totalWaiting 비원자 스냅샷 모순을 읽기 순서 + 클램프로 차단
- 복제 지연 시 rank null 폴백을 1번째가 아닌 맨 뒤 추정으로 수정
- 테스트 프로파일 전역에서 스케줄러 비활성 (공유 Redis 컨테이너를
  캐시된 다른 컨텍스트의 스케줄러가 드레인하는 flaky 차단)
- TTL 테스트 고정 sleep을 폴링 대기로 교체, carry-over에 날짜 파라미터
  오버로드 추가 (자정 경계 제거)
- FakeWaitingQueueRepository를 ZSET 의미론(동점 score 허용, member 사전순
  tie-break)에 맞게 재작성 (같은 밀리초 진입 유저가 덮어써지던 잠복 결함)

참고: ProductApiE2ETest '상품 목록을 조회할 수 있다' 실패는 본 변경 이전
HEAD에서도 재현되는 기존 문제로 이 커밋 범위가 아님.

Author: hyungkishin

apps/commerce-api/src/main/kotlin/com/loopers/application/queue/EnterQueueUseCase.kt

@@ -1,6 +1,7 @@
 package com.loopers.application.queue
 
 import com.loopers.domain.queue.EntryTokenRepository
+import com.loopers.domain.queue.QueueThroughput
 import com.loopers.domain.queue.WaitingQueueRepository
 import org.springframework.stereotype.Component
 
@@ -9,15 +10,6 @@ class EnterQueueUseCase(
     private val waitingQueueRepository: WaitingQueueRepository,
     private val entryTokenRepository: EntryTokenRepository,
 ) {
-    companion object {
-        /**
-         * 유저당 예상 처리 시간(초).
-         * DB 커넥션 풀 50, 평균 처리 200ms → 175 TPS → 유저당 약 1/175초 ≈ 0.006초
-         * 대기열 앞 유저 1명당 약 0.006초 대기로 산정하되, 스케줄러 100ms 주기 고려하여 반올림.
-         */
-        private const val SECONDS_PER_USER = 0.006
-    }
-
     fun enter(userId: Long): QueueEntryResult {
         val existingToken = entryTokenRepository.findToken(userId)
         if (existingToken != null) {
@@ -27,9 +19,16 @@ class EnterQueueUseCase(
         val score = System.currentTimeMillis().toDouble()
         waitingQueueRepository.enqueue(userId, score)
 
-        val position = (waitingQueueRepository.getPosition(userId) ?: 0L) + 1
-        val estimatedWaitSeconds = (position * SECONDS_PER_USER).toLong().coerceAtLeast(1)
+        // 복제 지연으로 rank가 아직 안 보이면 방금 진입한 유저이므로 맨 뒤로 추정한다.
+        // (rank 폴백 0은 "1번째"라는 거짓 안내가 된다)
+        val queueSize = waitingQueueRepository.getQueueSize()
+        val rank = waitingQueueRepository.getPosition(userId)
+        val position = (rank ?: (queueSize - 1).coerceAtLeast(0)) + 1
+
+        // size와 rank는 별도 읽기라 그 사이에 스케줄러가 dequeue하면 모순될 수 있다.
+        // position > totalWaiting인 응답만은 클램프로 막는다.
+        val totalWaiting = queueSize.coerceAtLeast(position)
 
-        return QueueEntryResult.queued(position, estimatedWaitSeconds)
+        return QueueEntryResult.queued(position, QueueThroughput.estimateWaitSeconds(position), totalWaiting)
     }
 }

apps/commerce-api/src/main/kotlin/com/loopers/application/queue/GetQueuePositionUseCase.kt

@@ -1,6 +1,7 @@
 package com.loopers.application.queue
 
 import com.loopers.domain.queue.EntryTokenRepository
+import com.loopers.domain.queue.QueueThroughput
 import com.loopers.domain.queue.WaitingQueueRepository
 import org.springframework.stereotype.Component
 
@@ -9,22 +10,22 @@ class GetQueuePositionUseCase(
     private val waitingQueueRepository: WaitingQueueRepository,
     private val entryTokenRepository: EntryTokenRepository,
 ) {
-    companion object {
-        private const val SECONDS_PER_USER = 0.006
-    }
-
     fun getPosition(userId: Long): QueuePositionResult {
         val token = entryTokenRepository.findToken(userId)
         if (token != null) {
             return QueuePositionResult.authorized(token)
         }
 
+        // size를 rank보다 먼저 읽는다: 두 읽기 사이에 스케줄러가 dequeue해도
+        // 먼저 읽은 size가 더 크므로 position > totalWaiting 모순이 생기지 않는다.
+        // 반대 방향(사이에 enqueue 증가)은 클램프로 막는다.
+        val queueSize = waitingQueueRepository.getQueueSize()
         val rank = waitingQueueRepository.getPosition(userId)
-            ?: return QueuePositionResult.notInQueue()
+            ?: return QueuePositionResult.notInQueue(queueSize)
 
         val position = rank + 1
-        val estimatedWaitSeconds = (position * SECONDS_PER_USER).toLong().coerceAtLeast(1)
+        val totalWaiting = queueSize.coerceAtLeast(position)
 
-        return QueuePositionResult.waiting(position, estimatedWaitSeconds)
+        return QueuePositionResult.waiting(position, QueueThroughput.estimateWaitSeconds(position), totalWaiting)
     }
 }

apps/commerce-api/src/main/kotlin/com/loopers/application/queue/QueueEntryResult.kt

@@ -5,14 +5,15 @@ data class QueueEntryResult(
     val position: Long?,
     val estimatedWaitSeconds: Long?,
     val token: String?,
+    val totalWaiting: Long?,
 ) {
     enum class QueueStatus { QUEUED, ALREADY_AUTHORIZED }
 
     companion object {
-        fun queued(position: Long, estimatedWaitSeconds: Long) =
-            QueueEntryResult(QueueStatus.QUEUED, position, estimatedWaitSeconds, null)
+        fun queued(position: Long, estimatedWaitSeconds: Long, totalWaiting: Long) =
+            QueueEntryResult(QueueStatus.QUEUED, position, estimatedWaitSeconds, null, totalWaiting)
 
         fun alreadyAuthorized(token: String) =
-            QueueEntryResult(QueueStatus.ALREADY_AUTHORIZED, null, null, token)
+            QueueEntryResult(QueueStatus.ALREADY_AUTHORIZED, null, null, token, null)
     }
 }

apps/commerce-api/src/main/kotlin/com/loopers/application/queue/QueueEntryScheduler.kt

@@ -1,12 +1,17 @@
 package com.loopers.application.queue
 
 import com.loopers.domain.queue.EntryTokenRepository
+import com.loopers.domain.queue.QueueThroughput
 import com.loopers.domain.queue.WaitingQueueRepository
 import org.slf4j.LoggerFactory
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
 import org.springframework.scheduling.annotation.Scheduled
 import org.springframework.stereotype.Component
 import java.util.UUID
 
+// 테스트 프로파일에서는 백그라운드 폴링이 공유 Redis의 대기열을 오염시키므로 끈다.
+// (apps/commerce-api/src/test/resources/application-test.yml)
+@ConditionalOnProperty(name = ["queue.scheduler.enabled"], havingValue = "true", matchIfMissing = true)
 @Component
 class QueueEntryScheduler(
     private val waitingQueueRepository: WaitingQueueRepository,
@@ -15,22 +20,15 @@ class QueueEntryScheduler(
     private val log = LoggerFactory.getLogger(javaClass)
 
     companion object {
-        /**
-         * 배치 크기 산정 근거:
-         * - DB 커넥션 풀: 50
-         * - 주문 1건 평균 처리 시간: 200ms
-         * - 이론적 최대 TPS: 50 / 0.2 = 250 TPS
-         * - 안전 마진 70%: 175 TPS
-         * - 스케줄러 주기: 100ms
-         * - 배치 크기: 175 * 0.1 ≈ 18명
-         */
-        const val BATCH_SIZE = 18L
         const val TOKEN_TTL_SECONDS = 300L
     }
 
-    @Scheduled(fixedDelay = 100)
+    // fixedDelay가 아니라 fixedRate인 이유: 설계 처리량 140 TPS는 100ms "주기"를 전제로
+    // 산정했다 (QueueThroughput). fixedDelay는 실행 시간이 주기에 가산되어 처리량이
+    // 항상 설계값 밑으로 떨어진다.
+    @Scheduled(fixedRate = QueueThroughput.SCHEDULER_INTERVAL_MS)
     fun processQueue() {
-        val userIds = waitingQueueRepository.dequeueTopN(BATCH_SIZE)
+        val userIds = waitingQueueRepository.dequeueTopN(QueueThroughput.BATCH_SIZE)
         if (userIds.isEmpty()) return
 
         for (userId in userIds) {

apps/commerce-api/src/main/kotlin/com/loopers/application/queue/QueuePositionResult.kt

@@ -5,17 +5,18 @@ data class QueuePositionResult(
     val position: Long?,
     val estimatedWaitSeconds: Long?,
     val token: String?,
+    val totalWaiting: Long?,
 ) {
     enum class PositionStatus { WAITING, AUTHORIZED, NOT_IN_QUEUE }
 
     companion object {
-        fun waiting(position: Long, estimatedWaitSeconds: Long) =
-            QueuePositionResult(PositionStatus.WAITING, position, estimatedWaitSeconds, null)
+        fun waiting(position: Long, estimatedWaitSeconds: Long, totalWaiting: Long) =
+            QueuePositionResult(PositionStatus.WAITING, position, estimatedWaitSeconds, null, totalWaiting)
 
         fun authorized(token: String) =
-            QueuePositionResult(PositionStatus.AUTHORIZED, 0, 0, token)
+            QueuePositionResult(PositionStatus.AUTHORIZED, 0, 0, token, null)
 
-        fun notInQueue() =
-            QueuePositionResult(PositionStatus.NOT_IN_QUEUE, null, null, null)
+        fun notInQueue(totalWaiting: Long) =
+            QueuePositionResult(PositionStatus.NOT_IN_QUEUE, null, null, null, totalWaiting)
     }
 }

apps/commerce-api/src/main/kotlin/com/loopers/domain/queue/QueueThroughput.kt

@@ -0,0 +1,29 @@
+package com.loopers.domain.queue
+
+import kotlin.math.ceil
+
+/**
+ * 대기열 처리량 정책. 산정 근거: docs/design/06-waiting-queue.md
+ *
+ * - DB 커넥션 풀 40 (modules/jpa/jpa.yml maximum-pool-size)
+ * - 주문 1건 평균 처리 시간 200ms
+ * - 이론적 최대 TPS: 40 / 0.2 = 200
+ * - 안전 마진 70%: 설계 처리량 140 TPS
+ *
+ * 스케줄러 배치 크기와 예상 대기 시간이 모두 이 값에서 파생된다.
+ * 커넥션 풀이나 처리 시간이 바뀌면 이 파일만 수정한다.
+ */
+object QueueThroughput {
+    const val DESIGN_TPS = 140.0
+    const val SCHEDULER_INTERVAL_MS = 100L
+
+    val BATCH_SIZE: Long = (DESIGN_TPS * SCHEDULER_INTERVAL_MS / 1000).toLong()
+
+    /**
+     * 순번 기준 예상 대기 시간(초). 스케줄러가 100ms 단위로 이산 처리하므로
+     * 내림이 아니라 올림으로 계산해 과소 안내를 막는다.
+     */
+    fun estimateWaitSeconds(position: Long): Long {
+        return ceil(position / DESIGN_TPS).toLong().coerceAtLeast(1)
+    }
+}

apps/commerce-api/src/main/kotlin/com/loopers/interfaces/api/v1/queue/QueueEnterResponse.kt

@@ -7,13 +7,15 @@ data class QueueEnterResponse(
     val position: Long?,
     val estimatedWaitSeconds: Long?,
     val token: String?,
+    val totalWaiting: Long?,
 ) {
     companion object {
         fun from(result: QueueEntryResult) = QueueEnterResponse(
             status = result.status.name,
             position = result.position,
             estimatedWaitSeconds = result.estimatedWaitSeconds,
             token = result.token,
+            totalWaiting = result.totalWaiting,
         )
     }
 }

apps/commerce-api/src/main/kotlin/com/loopers/interfaces/api/v1/queue/QueuePositionResponse.kt

@@ -7,13 +7,15 @@ data class QueuePositionResponse(
     val position: Long?,
     val estimatedWaitSeconds: Long?,
     val token: String?,
+    val totalWaiting: Long?,
 ) {
     companion object {
         fun from(result: QueuePositionResult) = QueuePositionResponse(
             status = result.status.name,
             position = result.position,
             estimatedWaitSeconds = result.estimatedWaitSeconds,
             token = result.token,
+            totalWaiting = result.totalWaiting,
         )
     }
 }

apps/commerce-api/src/test/kotlin/com/loopers/application/queue/EnterQueueUseCaseTest.kt

@@ -26,6 +26,7 @@ class EnterQueueUseCaseTest {
         assertThat(result.status).isEqualTo(QueueEntryResult.QueueStatus.QUEUED)
         assertThat(result.position).isEqualTo(1)
         assertThat(result.estimatedWaitSeconds).isNotNull()
+        assertThat(result.totalWaiting).isEqualTo(1)
     }
 
     @Test
@@ -55,6 +56,15 @@ class EnterQueueUseCaseTest {
         assertThat(result.position).isEqualTo(2)
     }
 
+    @Test
+    fun `진입 결과에 전체 대기 인원이 포함되어야 한다`() {
+        enterQueueUseCase.enter(1L)
+        enterQueueUseCase.enter(2L)
+        val result = enterQueueUseCase.enter(3L)
+
+        assertThat(result.totalWaiting).isEqualTo(3)
+    }
+
     companion object {
         private const val USER_ID = 1L
     }

apps/commerce-api/src/test/kotlin/com/loopers/application/queue/GetQueuePositionUseCaseTest.kt

@@ -27,6 +27,7 @@ class GetQueuePositionUseCaseTest {
 
         assertThat(result.status).isEqualTo(QueuePositionResult.PositionStatus.WAITING)
         assertThat(result.position).isEqualTo(1)
+        assertThat(result.totalWaiting).isEqualTo(1)
     }
 
     @Test
@@ -46,6 +47,28 @@ class GetQueuePositionUseCaseTest {
         assertThat(result.status).isEqualTo(QueuePositionResult.PositionStatus.NOT_IN_QUEUE)
     }
 
+    @Test
+    fun `대기 중 조회 시 전체 대기 인원이 포함되어야 한다`() {
+        waitingQueueRepository.enqueue(1L, 1.0)
+        waitingQueueRepository.enqueue(2L, 2.0)
+        waitingQueueRepository.enqueue(3L, 3.0)
+
+        val result = getQueuePositionUseCase.getPosition(2L)
+
+        assertThat(result.totalWaiting).isEqualTo(3)
+    }
+
+    @Test
+    fun `대기열에 없는 유저도 전체 대기 인원은 확인할 수 있어야 한다`() {
+        waitingQueueRepository.enqueue(1L, 1.0)
+        waitingQueueRepository.enqueue(2L, 2.0)
+
+        val result = getQueuePositionUseCase.getPosition(999L)
+
+        assertThat(result.status).isEqualTo(QueuePositionResult.PositionStatus.NOT_IN_QUEUE)
+        assertThat(result.totalWaiting).isEqualTo(2)
+    }
+
     companion object {
         private const val USER_ID = 1L
     }