fix(core): 재발급 클라이언트에 전용 Dispatcher 를 줘 콜드 스타트의 동시 401 교착을 없앤다 (#2160)

core/network/src/main/kotlin/com/afternote/core/network/di/NetworkModule.kt

@@ -16,6 +16,7 @@ import dagger.hilt.android.qualifiers.ApplicationContext
 import dagger.hilt.components.SingletonComponent
 import kotlinx.serialization.json.Json
 import okhttp3.Call
+import okhttp3.Dispatcher
 import okhttp3.Interceptor
 import okhttp3.MediaType.Companion.toMediaType
 import okhttp3.OkHttpClient
@@ -122,7 +123,8 @@ object NetworkModule { // 이 모듈은 오브젝트 클래스 선언해서 딱
 
     /**
      * 모든 클라이언트의 공통 뿌리. 파생은 [OkHttpClient.newBuilder] 로 — 설정값만이 아니라
-     * ConnectionPool · Dispatcher · 스레드풀을 실제로 공유한다.
+     * ConnectionPool · Dispatcher · 스레드풀을 실제로 공유한다. 예외는 재발급 클라이언트의
+     * Dispatcher 하나다([provideRefreshOkHttpClient]).
      * 인터셉터는 여기 두지 않는다: 로깅은 각 클라이언트가 마지막에 달아야 최종 요청·응답을 관찰하는데,
      * base 에 두면 파생 시 맨 앞으로 밀린다.
      */
@@ -131,8 +133,22 @@ object NetworkModule { // 이 모듈은 오브젝트 클래스 선언해서 딱
     @Named("BaseClient")
     fun provideBaseOkHttpClient(): OkHttpClient = OkHttpClient.Builder().withApiTimeouts().build()
 
-    // 리이슈를 할 때 일반용 OkhttpClient만 사용하면 액세스 토큰이 계속 헤더에 포함되어 401을 받는 행위가 무한 반복
-    // 이를 해결하기 위해 401을 받았을 때는 토큰을 헤더에 포함하지 않고 요청을 보내는 버전
+    /**
+     * 재발급 전용. 메인 클라이언트로 재발급하면 만료된 액세스 토큰이 계속 헤더에 붙어 401 이 무한 반복되므로,
+     * 토큰을 붙이는 인터셉터 없이 요청을 보낸다.
+     *
+     * 커넥션 풀은 base 와 공유하지만 Dispatcher 는 따로 둔다 (#2160).
+     *
+     * 재발급을 기다리는 쪽이 메인 클라이언트의 디스패처 스레드다. [TokenAuthenticator] 와
+     * [AuthInterceptor] 는 요청을 실행하던 그 스레드에서 재발급 락을 잡고 끝날 때까지 막혀 있다.
+     * 재발급 호출이 같은 Dispatcher 로 들어가면, 같은 호스트의 401 이 호스트당 실행 칸(기본 5개)을
+     * 다 쥔 채 재발급을 기다리고 재발급은 칸이 비기를 기다리며 대기열에 머문다. 대기열의 호출에는
+     * `callTimeout` 도 걸리지 않아 스스로 풀리지 않는다. 상한을 올리는 것은 교착이 나는 동시 요청 수만
+     * 뒤로 미룬다.
+     *
+     * base 의 실행기를 넘겨 칸 계산만 가르지 않는 이유: 그러면 독립성이 실행기가 무제한 캐시 풀이라는
+     * 전제에 기댄다. 새 [Dispatcher] 는 그 전제 없이 끊긴다.
+     */
     @Provides
     @Singleton
     @Named("RefreshClient")
@@ -142,6 +158,7 @@ object NetworkModule { // 이 모듈은 오브젝트 클래스 선언해서 딱
     ): OkHttpClient =
         baseClient
             .newBuilder()
+            .dispatcher(Dispatcher())
             .addInterceptor(loggingInterceptor)
             .build()
 

core/network/src/test/kotlin/com/afternote/core/network/di/ConcurrentUnauthorizedReissueTest.kt

@@ -0,0 +1,335 @@
+package com.afternote.core.network.di
+
+import com.afternote.core.domain.testing.FakeAuthRepository
+import com.afternote.core.model.TokenBundle
+import com.afternote.core.network.FakeErrorReporter
+import com.afternote.core.network.calladapter.ApiErrorCallAdapterFactory
+import com.afternote.core.network.dto.ReissueRequestDto
+import com.afternote.core.network.interceptor.AuthInterceptor
+import com.afternote.core.network.interceptor.TokenAuthenticator
+import com.afternote.core.network.service.TokenApiService
+import com.afternote.core.network.token.AccessTokenExpiryTracker
+import com.afternote.core.network.token.TokenReissuer
+import okhttp3.Call
+import okhttp3.Callback
+import okhttp3.MediaType.Companion.toMediaType
+import okhttp3.OkHttpClient
+import okhttp3.Request
+import okhttp3.RequestBody.Companion.toRequestBody
+import okhttp3.Response
+import okhttp3.logging.HttpLoggingInterceptor
+import org.junit.After
+import org.junit.Assert.assertEquals
+import org.junit.Assert.fail
+import org.junit.Test
+import retrofit2.create
+import java.io.BufferedInputStream
+import java.io.ByteArrayOutputStream
+import java.io.IOException
+import java.io.InputStream
+import java.net.InetAddress
+import java.net.ServerSocket
+import java.net.Socket
+import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicInteger
+
+/**
+ * 액세스 토큰이 만료된 채 콜드 스타트하면 같은 호스트로 동시에 나간 인증 요청이 전부 401 을 받는다.
+ * 이때 재발급까지 끝나고 모든 요청이 새 토큰으로 성공해야 한다.
+ *
+ * 조립은 운영과 같다. [NetworkModule] 의 base·재발급·메인 클라이언트와 [ServiceModule] 의 재발급
+ * Retrofit 을 그대로 쓰고, 기본 주소만 로컬 서버로 바꾼다. 요청 5건도 콜드 스타트 스레드 덤프에
+ * 찍힌 경로 그대로다. 주간 리포트는 운영처럼 느린 경로용 파생 클라이언트를 탄다.
+ *
+ * 서버는 만료 토큰 요청이 [ColdStartServer.expiredBarrier] 수만큼 모일 때까지 401 을 쥐고 있는다.
+ * 첫 401 이 재발급을 부르기 전에 요청 전부가 디스패처 실행 칸을 차지하게 해, 운 좋게 먼저 끝난
+ * 요청이 칸을 비워 주는 경합을 없애기 위해서다.
+ */
+class ConcurrentUnauthorizedReissueTest {
+    private val closeables = mutableListOf<AutoCloseable>()
+
+    @After
+    fun tearDown() {
+        closeables.asReversed().forEach { runCatching { it.close() } }
+    }
+
+    @Test
+    fun `같은 호스트 동시 401 이 5건이면 재발급이 끝나고 모든 요청이 성공한다`() {
+        assertColdStartCompletes(paths = COLD_START_PATHS)
+    }
+
+    @Test
+    fun `같은 호스트 동시 401 이 4건이면 재발급이 끝나고 모든 요청이 성공한다`() {
+        assertColdStartCompletes(paths = COLD_START_PATHS.take(4))
+    }
+
+    private fun assertColdStartCompletes(paths: List<ColdStartRequest>) {
+        val server = ColdStartServer(expiredBarrier = paths.size).also(closeables::add)
+        val wiring = ColdStartWiring(server.baseUrl).also(closeables::add)
+
+        val finished = CountDownLatch(paths.size)
+        val statuses = ConcurrentLinkedQueue<String>()
+        paths.forEach { request ->
+            wiring.callFactory.newCall(request.toOkHttp(server.baseUrl)).enqueue(
+                object : Callback {
+                    override fun onResponse(
+                        call: Call,
+                        response: Response,
+                    ) {
+                        response.use { statuses += "${request.path}=${it.code}" }
+                        finished.countDown()
+                    }
+
+                    override fun onFailure(
+                        call: Call,
+                        e: IOException,
+                    ) {
+                        statuses += "${request.path}=${e::class.simpleName}"
+                        finished.countDown()
+                    }
+                },
+            )
+        }
+
+        if (!finished.await(COMPLETION_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
+            val dispatcher = wiring.baseClient.dispatcher
+            val stalled =
+                "요청 ${paths.size}건 중 ${paths.size - finished.count}건만 ${COMPLETION_TIMEOUT_SECONDS}초 안에 끝남. " +
+                    "재발급 서버 도달 ${server.reissueHits.get()}회, " +
+                    "디스패처 실행 ${dispatcher.runningCallsCount()}건·대기 ${dispatcher.queuedCallsCount()}건, " +
+                    "maxRequestsPerHost=${dispatcher.maxRequestsPerHost}"
+            // 멈춘 OkHttp 스레드는 데몬이 아니다. 상한을 올려 대기 중인 재발급을 내보내야 테스트 JVM 이
+            // 스레드를 남기지 않는다. 풀리는지 여부 자체가 원인이 호스트당 상한이라는 증거다.
+            dispatcher.maxRequestsPerHost = UNBLOCK_MAX_REQUESTS_PER_HOST
+            val releasedAfterRaise = finished.await(COMPLETION_TIMEOUT_SECONDS, TimeUnit.SECONDS)
+            fail(
+                "$stalled. 상한을 $UNBLOCK_MAX_REQUESTS_PER_HOST 로 올린 뒤 풀림=$releasedAfterRaise, " +
+                    "재발급 서버 도달 ${server.reissueHits.get()}회, 응답=$statuses",
+            )
+        }
+
+        assertEquals("재발급은 한 번만 서버에 도달해야 한다", 1, server.reissueHits.get())
+        assertEquals(
+            paths.map { "${it.path}=200" }.sorted(),
+            statuses.toList().sorted(),
+        )
+    }
+
+    /** 운영 [NetworkModule]·[ServiceModule] 조립을 그대로 따르되 기본 주소만 로컬 서버로 바꾼 묶음. */
+    private class ColdStartWiring(
+        baseUrl: String,
+    ) : AutoCloseable {
+        private val json = NetworkModule.provideJson()
+        private val callAdapterFactory = ApiErrorCallAdapterFactory(json)
+        private val loggingInterceptor = HttpLoggingInterceptor()
+        private val tracker = AccessTokenExpiryTracker { 0L }
+        private val reporter = FakeErrorReporter()
+
+        val baseClient: OkHttpClient = NetworkModule.provideBaseOkHttpClient()
+        private val refreshClient: OkHttpClient = NetworkModule.provideRefreshOkHttpClient(baseClient, loggingInterceptor)
+
+        private val tokenApiService: TokenApiService =
+            ServiceModule
+                .provideRefreshRetrofit(
+                    refreshClient = refreshClient,
+                    json = json,
+                    apiErrorCallAdapterFactory = callAdapterFactory,
+                ).newBuilder()
+                .baseUrl(baseUrl)
+                .build()
+                .create<TokenApiService>()
+
+        /** `AuthRepositoryImpl.rotateToken` 과 같은 순서: 저장 refresh 로 재발급을 치고 새 토큰을 저장한다. */
+        private val repository =
+            FakeAuthRepository(
+                loggedIn = true,
+                accessToken = EXPIRED_ACCESS_TOKEN,
+                refreshToken = STORED_REFRESH_TOKEN,
+                onRotateToken = {
+                    runCatching {
+                        val data = checkNotNull(tokenApiService.reissue(ReissueRequestDto(checkNotNull(refreshToken))).data)
+                        accessToken = data.accessToken
+                        refreshToken = data.refreshToken
+                        TokenBundle(data.accessToken, data.refreshToken, data.expiresIn)
+                    }
+                },
+            )
+
+        private val reissuer = TokenReissuer({ repository }, tracker, reporter)
+
+        private val mainClient: OkHttpClient =
+            NetworkModule.provideMainOkHttpClient(
+                baseClient = baseClient,
+                debugInterceptors = emptySet(),
+                featureInterceptors = emptySet(),
+                loggingInterceptor = loggingInterceptor,
+                authInterceptor = AuthInterceptor({ repository }, tracker, reissuer),
+                tokenAuthenticator = TokenAuthenticator({ repository }, reissuer, reporter),
+            )
+
+        /** 운영 Retrofit 의 호출 팩토리 — 주간 리포트만 느린 경로용 파생 클라이언트로 가른다. */
+        val callFactory: Call.Factory =
+            NetworkModule
+                .provideRetrofit(mainClient, json, callAdapterFactory)
+                .newBuilder()
+                .baseUrl(baseUrl)
+                .build()
+                .callFactory()
+
+        override fun close() {
+            // 재발급 클라이언트는 디스패처를 따로 쓴다(#2160) — 둘 다 닫아야 스레드가 남지 않는다.
+            listOf(baseClient.dispatcher, refreshClient.dispatcher).distinct().forEach { dispatcher ->
+                dispatcher.cancelAll()
+                dispatcher.executorService.shutdown()
+            }
+            baseClient.connectionPool.evictAll()
+        }
+    }
+
+    private data class ColdStartRequest(
+        val method: String,
+        val path: String,
+        val jsonBody: String? = null,
+    ) {
+        fun toOkHttp(baseUrl: String): Request =
+            Request
+                .Builder()
+                .url(baseUrl + path)
+                .method(method, jsonBody?.toRequestBody(JSON_MEDIA_TYPE))
+                .build()
+    }
+
+    /**
+     * 만료 토큰에는 401, 재발급에는 새 토큰, 새 토큰에는 200 으로 답하는 최소 HTTP/1.1 서버.
+     * 새 테스트 의존성 없이 소켓 경계를 실제로 지나게 하려고 `java.net` 만 쓴다.
+     */
+    private class ColdStartServer(
+        expiredBarrier: Int,
+    ) : AutoCloseable {
+        private val serverSocket = ServerSocket(0, BACKLOG, InetAddress.getLoopbackAddress())
+        private val sockets = ConcurrentLinkedQueue<Socket>()
+        private val expiredArrivals = CountDownLatch(expiredBarrier)
+        val reissueHits = AtomicInteger()
+        val baseUrl = "http://${serverSocket.inetAddress.hostAddress}:${serverSocket.localPort}/api/v1/"
+
+        init {
+            daemon("cold-start-server-accept") {
+                while (!serverSocket.isClosed) {
+                    val socket = runCatching { serverSocket.accept() }.getOrNull() ?: break
+                    sockets += socket
+                    daemon("cold-start-server-conn") { serve(socket) }
+                }
+            }
+        }
+
+        private fun serve(socket: Socket) {
+            socket.use {
+                val input = BufferedInputStream(it.getInputStream())
+                val output = it.getOutputStream()
+                while (true) {
+                    val requestLine = input.readHttpLine() ?: return
+                    if (requestLine.isEmpty()) continue
+                    val headers = generateSequence { input.readHttpLine()?.takeIf(String::isNotEmpty) }.toList()
+                    val contentLength =
+                        headers
+                            .firstOrNull { header -> header.startsWith("Content-Length:", ignoreCase = true) }
+                            ?.substringAfter(':')
+                            ?.trim()
+                            ?.toInt() ?: 0
+                    repeat(contentLength) { input.read() }
+                    val authorization =
+                        headers
+                            .firstOrNull { header -> header.startsWith("Authorization:", ignoreCase = true) }
+                            ?.substringAfter(':')
+                            ?.trim()
+                    val (status, body) = respond(requestLine, authorization)
+                    val bytes = body.toByteArray()
+                    output.write(
+                        (
+                            "HTTP/1.1 $status ${if (status == 200) "OK" else "Unauthorized"}\r\n" +
+                                "Content-Type: application/json\r\n" +
+                                "Content-Length: ${bytes.size}\r\n\r\n"
+                        ).toByteArray(),
+                    )
+                    output.write(bytes)
+                    output.flush()
+                }
+            }
+        }
+
+        private fun respond(
+            requestLine: String,
+            authorization: String?,
+        ): Pair<Int, String> =
+            when {
+                requestLine.startsWith("POST /api/v1/auth/reissue ") -> {
+                    reissueHits.incrementAndGet()
+                    200 to REISSUE_RESPONSE
+                }
+
+                authorization == "Bearer $FRESH_ACCESS_TOKEN" -> {
+                    200 to OK_RESPONSE
+                }
+
+                else -> {
+                    expiredArrivals.countDown()
+                    expiredArrivals.await(BARRIER_TIMEOUT_SECONDS, TimeUnit.SECONDS)
+                    401 to EXPIRED_RESPONSE
+                }
+            }
+
+        override fun close() {
+            serverSocket.close()
+            sockets.forEach { runCatching { it.close() } }
+        }
+    }
+
+    private companion object {
+        const val EXPIRED_ACCESS_TOKEN = "expired-access"
+        const val FRESH_ACCESS_TOKEN = "fresh-access"
+        const val STORED_REFRESH_TOKEN = "stored-refresh"
+        const val COMPLETION_TIMEOUT_SECONDS = 5L
+        const val BARRIER_TIMEOUT_SECONDS = 5L
+        const val UNBLOCK_MAX_REQUESTS_PER_HOST = 64
+        const val BACKLOG = 50
+
+        val JSON_MEDIA_TYPE = "application/json".toMediaType()
+
+        /** 콜드 스타트 스레드 덤프(2026-09-23)에서 멈춰 있던 요청 5건. */
+        val COLD_START_PATHS =
+            listOf(
+                ColdStartRequest("GET", "users/me"),
+                ColdStartRequest("GET", "users/receivers"),
+                ColdStartRequest("GET", "daily-questions/today"),
+                ColdStartRequest("PUT", "users/push-tokens", """{"token":"fid","platform":"ANDROID"}"""),
+                ColdStartRequest("GET", "mind-record"),
+            )
+
+        const val OK_RESPONSE = """{"status":200,"code":200,"message":"ok","data":null}"""
+        const val EXPIRED_RESPONSE = """{"status":401,"code":1106,"message":"expired"}"""
+        const val REISSUE_RESPONSE =
+            """{"status":200,"code":200,"message":"ok",""" +
+                """"data":{"accessToken":"$FRESH_ACCESS_TOKEN","refreshToken":"fresh-refresh","expiresIn":3600}}"""
+
+        fun daemon(
+            name: String,
+            block: () -> Unit,
+        ) {
+            Thread(block, name).apply { isDaemon = true }.start()
+        }
+
+        /** CRLF 로 끝나는 한 줄을 읽는다. 스트림이 닫혔으면 null. */
+        fun InputStream.readHttpLine(): String? {
+            val line = ByteArrayOutputStream()
+            while (true) {
+                when (val byte = read()) {
+                    -1 -> return if (line.size() == 0) null else line.toString(Charsets.US_ASCII.name())
+                    '\n'.code -> return line.toString(Charsets.US_ASCII.name()).trimEnd('\r')
+                    else -> line.write(byte)
+                }
+            }
+        }
+    }
+}