diff --git a/valdi/src/java/com/snap/valdi/network/DefaultHTTPRequestManager.kt b/valdi/src/java/com/snap/valdi/network/DefaultHTTPRequestManager.kt index e0eefca17..942ad34d7 100644 --- a/valdi/src/java/com/snap/valdi/network/DefaultHTTPRequestManager.kt +++ b/valdi/src/java/com/snap/valdi/network/DefaultHTTPRequestManager.kt @@ -1,17 +1,64 @@ package com.snap.valdi.network import android.content.Context -import com.snap.valdi.utils.ExecutorsUtil +import androidx.annotation.VisibleForTesting +import java.io.IOException import java.io.InputStream import java.net.HttpURLConnection import java.net.URL -import java.util.concurrent.Executors +import java.net.URLConnection +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.LinkedBlockingQueue +import java.util.concurrent.ThreadPoolExecutor +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger import com.snapchat.client.valdi.* import com.snapchat.client.valdi_core.* -class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() { +class DefaultHTTPRequestManager @VisibleForTesting constructor( + context: Context, + private val keepAliveMs: Long = DEFAULT_KEEP_ALIVE_MS, + private val openConnection: (URL) -> URLConnection, +): HTTPRequestManager() { - private class RequestTask(val url: URL, val method: String, val body: ByteArray?, val headers: Map, completion: HTTPRequestManagerCompletion): HTTPRequestTask(completion), Runnable { + constructor(context: Context): this(context, openConnection = { it.openConnection() }) + + private class RequestTask(val url: URL, val method: String, val body: ByteArray?, val headers: Map, val openConnection: (URL) -> URLConnection, completion: HTTPRequestManagerCompletion): HTTPRequestTask(completion), Runnable { + + private var connection: HttpURLConnection? = null + private var cancelled = false + + override fun cancel() { + super.cancel() + + val connectionToClose = synchronized(this) { + cancelled = true + connection + } + + // Closing from another thread makes the worker's blocked read throw, which is + // how the thread gets reclaimed. Do it outside the lock so a slow close cannot stall + // the task binding its connection. Swallow like the worker's own teardown does: this + // races that teardown, and native is the only caller left to hand a throw to. + try { + connectionToClose?.disconnect() + } catch (exc: Exception) {} + } + + private fun isCancelled(): Boolean = synchronized(this) { cancelled } + + /** + * Hands the connection to [cancel] so it can be torn down. Returns false when the request + * was already cancelled, in which case the caller must not go on to perform it. + */ + private fun bindConnection(urlConnection: HttpURLConnection): Boolean = synchronized(this) { + if (cancelled) { + false + } else { + connection = urlConnection + true + } + } private fun doPerformRequestWithURLConnection(urlConnection: HttpURLConnection): HTTPResponse { urlConnection.instanceFollowRedirects = true @@ -25,6 +72,14 @@ class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() { urlConnection.doInput = true + // Nothing above this point touches the network. Writing the body connects, and so + // does reading responseCode when there is no body, and disconnect() cannot tear + // down a connection that has not connected yet -- so this is the last point a + // cancel that arrived during setup can be honoured. + if (isCancelled()) { + throw IOException("Request was cancelled") + } + if (body != null) { urlConnection.doOutput = true urlConnection.outputStream.write(body) @@ -53,9 +108,14 @@ class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() { } private fun performRequest(): HTTPResponse { - val urlConnection = url.openConnection() + val urlConnection = openConnection(url) if (urlConnection is HttpURLConnection) { + if (!bindConnection(urlConnection)) { + urlConnection.disconnect() + throw IOException("Request was cancelled") + } + return doPerformRequestWithURLConnection(urlConnection) } else { urlConnection.doInput = true @@ -66,17 +126,26 @@ class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() { } override fun run() { + // Cancelled while queued: opening the connection at all would be wasted work. + if (isCancelled()) { + return + } + try { val response = performRequest() notifySuccess(response) - } catch (error: Exception) { + } catch (error: Throwable) { + // Once cancelled the completion is already gone, so a teardown IOException lands + // here and goes nowhere -- matching how iOS swallows NSURLErrorCancelled. + // Catches Throwable so that reading an oversized body into memory fails the + // request rather than leaving the completion unsettled. notifyFailure("HTTP Request failed: ${error.message}") } } companion object { - fun from(request: HTTPRequest, completion: HTTPRequestManagerCompletion): RequestTask { + fun from(request: HTTPRequest, openConnection: (URL) -> URLConnection, completion: HTTPRequestManagerCompletion): RequestTask { val url = URL(request.url) val method = request.method val body = request.body @@ -93,34 +162,93 @@ class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() { } } - return RequestTask(url, method, body, headers, completion) + return RequestTask(url, method, body, headers, openConnection, completion) + } + } + } + + private val threadCount = AtomicInteger(0) + + // One pool per host, matching NSURLSession.httpMaximumConnectionsPerHost, so a stalled host + // cannot starve requests to other hosts. + private val executors = ConcurrentHashMap() + + private fun newHostPool(): ThreadPoolExecutor = + ThreadPoolExecutor( + MAX_CONCURRENT_REQUESTS_PER_HOST, + MAX_CONCURRENT_REQUESTS_PER_HOST, + keepAliveMs, + TimeUnit.MILLISECONDS, + LinkedBlockingQueue(), + ) { r -> + Thread(r).apply { + name = "Valdi Network Thread ${threadCount.incrementAndGet()}" + priority = Thread.NORM_PRIORITY } + }.apply { + // An unbounded queue never rejects, so a pool only ever grows to its core size and + // maximumPoolSize is inert. Concurrency has to come from the core size, and this + // restores the idle reaping that the previous core size of zero provided. + allowCoreThreadTimeOut(true) + } + + // Picking the pool and submitting to it as two steps would let a sweep shut it down in between, + // so compute() does both under the bin lock for the host. + private fun submitToHostPool(task: RequestTask) { + executors.compute(task.url.host.orEmpty().lowercase()) { _, pool -> + (pool ?: newHostPool()).also { it.execute(task) } } } - private var executors = ExecutorsUtil.newSingleThreadCachedExecutor { r -> - Thread(r).apply { - name = "Valdi Network Thread" - priority = Thread.NORM_PRIORITY + /** + * A pool with no threads has had none for a full keep-alive window, and [ThreadPoolExecutor] + * adds a worker back whenever the queue is non-empty, so no threads also means nothing queued. + * + * Runs after a request has been submitted rather than on a timer: [HTTPRequestManager] has no + * dispose hook, so there would be nowhere to stop a scheduler thread. Sweeping after the submit + * rather than before it keeps the pool this request just used out of the sweep. + */ + private fun evictDormantHostPools() { + for (host in executors.keys) { + executors.computeIfPresent(host) { _, pool -> + if (pool.poolSize == 0) { + pool.shutdown() + null + } else { + pool + } + } } } + @VisibleForTesting + fun hostPools(): Set = executors.keys.toSet() + override fun performRequest(request: HTTPRequest, completion: HTTPRequestManagerCompletion): Cancelable { - val task: RequestTask try { - task = RequestTask.from(request, completion) - } catch (exception: Exception) { - completion.onFail("Failed to build request: ${exception.message}") + val task = RequestTask.from(request, openConnection, completion) + + submitToHostPool(task) + evictDormantHostPools() + + return task + } catch (throwable: Throwable) { + // Anything escaping to JNI leaves the completion unsettled, which hangs the native + // request instead of failing it. Catches Throwable because starting a pool thread + // fails with OutOfMemoryError, not an Exception. + completion.onFail("Failed to perform request: ${throwable.message}") return object: Cancelable() { override fun cancel() { } } } + } - executors.submit(task) + companion object { + const val MAX_CONCURRENT_REQUESTS_PER_HOST = 4 - return task + private const val DEFAULT_KEEP_ALIVE_MS = 60_000L } } diff --git a/valdi/test/java/network/DefaultHTTPRequestManagerTest.kt b/valdi/test/java/network/DefaultHTTPRequestManagerTest.kt new file mode 100644 index 000000000..60ea4252b --- /dev/null +++ b/valdi/test/java/network/DefaultHTTPRequestManagerTest.kt @@ -0,0 +1,593 @@ +package com.snap.valdi.network + +import android.content.Context +import androidx.test.core.app.ApplicationProvider +import com.snapchat.client.valdi_core.Cancelable +import com.snapchat.client.valdi_core.HTTPRequest +import com.snapchat.client.valdi_core.HTTPRequestManagerCompletion +import com.snapchat.client.valdi_core.HTTPResponse +import org.junit.After +import org.junit.Assert.assertFalse +import org.junit.Assert.assertTrue +import org.junit.Assert.fail +import org.junit.Before +import org.junit.Test +import org.junit.runner.RunWith +import org.robolectric.RobolectricTestRunner +import org.robolectric.annotation.Config +import java.io.ByteArrayInputStream +import java.io.ByteArrayOutputStream +import java.io.Closeable +import java.io.InputStream +import java.io.OutputStream +import java.net.HttpURLConnection +import java.net.InetAddress +import java.net.ServerSocket +import java.net.Socket +import java.net.URL +import java.util.Collections +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger + +private const val OK_RESPONSE = "HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok" + +private const val SHORT_KEEP_ALIVE_MS = 50L + +private const val DORMANT_HOSTS = 12 + +private fun await(timeoutMs: Long, condition: () -> Boolean): Boolean { + val deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(timeoutMs) + while (System.nanoTime() < deadline) { + if (condition()) { + return true + } + Thread.sleep(25) + } + return condition() +} + +/** + * A loopback HTTP server with two modes. While holding, it accepts a connection, consumes the + * request head and then leaves the socket open indefinitely, so the client stays blocked in read. + * That is what lets a test observe whether a cancelled request actually hangs up. + */ +private class FakeServer(respondImmediately: Boolean) : Closeable { + + private val serverSocket = ServerSocket(0, 128, InetAddress.getLoopbackAddress()) + private val accepted = AtomicInteger(0) + private val clients = Collections.synchronizedList(mutableListOf()) + private val requestLines = Collections.synchronizedList(mutableListOf()) + private val hungUp = AtomicInteger(0) + + @Volatile private var respond = respondImmediately + @Volatile private var running = true + @Volatile private var releasing = false + + init { + Thread({ acceptLoop() }, "fake-server-accept").apply { isDaemon = true }.start() + } + + fun url(path: String): String = + "http://${serverSocket.inetAddress.hostAddress}:${serverSocket.localPort}$path" + + /** + * The same socket, reached under a different hostname. The connection pool is keyed on host, + * so this is how a test addresses one server as if it were two. + */ + fun urlForHost(host: String, path: String): String = + "http://$host:${serverSocket.localPort}$path" + + fun acceptedCount(): Int = accepted.get() + + fun sawRequestFor(path: String): Boolean = + synchronized(requestLines) { requestLines.toList() }.any { it.contains(" $path ") } + + fun awaitAccepted(count: Int, timeoutMs: Long): Boolean = await(timeoutMs) { accepted.get() >= count } + + /** + * Waits until the server has read [count] request heads. Stronger than [awaitAccepted]: the + * kernel completes the TCP handshake off the listen backlog before the client has written + * anything, so an accepted connection does not yet mean the client is awaiting a response. + */ + fun awaitRequestsRead(count: Int, timeoutMs: Long): Boolean = + await(timeoutMs) { synchronized(requestLines) { requestLines.size } >= count } + + fun hungUpCount(): Int = hungUp.get() + + fun awaitHungUp(count: Int, timeoutMs: Long): Boolean = await(timeoutMs) { hungUp.get() >= count } + + fun startResponding() { + respond = true + } + + /** Close every held connection so any blocked client unwinds. */ + fun releaseAll() { + releasing = true + synchronized(clients) { clients.toList() }.forEach { runCatching { it.close() } } + } + + override fun close() { + running = false + releaseAll() + runCatching { serverSocket.close() } + } + + private fun acceptLoop() { + while (running) { + val socket = try { + serverSocket.accept() + } catch (exception: Exception) { + return + } + clients.add(socket) + accepted.incrementAndGet() + Thread({ serve(socket) }, "fake-server-connection").apply { isDaemon = true }.start() + } + } + + private fun serve(socket: Socket) { + try { + val input = socket.getInputStream() + readRequestLine(input)?.let { requestLines.add(it) } + + if (respond) { + socket.getOutputStream().apply { + write(OK_RESPONSE.toByteArray()) + flush() + } + socket.close() + return + } + + // Hold the connection open. read() returning -1, or throwing a reset, is the server + // seeing the client hang up, which is what a real cancellation looks like from here. + while (input.read() != -1) { + // drain + } + if (!releasing) { + hungUp.incrementAndGet() + } + } catch (exception: Exception) { + if (!releasing) { + hungUp.incrementAndGet() + } + } + } + + private fun readRequestLine(input: InputStream): String? { + val head = StringBuilder() + var newlines = 0 + while (newlines < 2) { + val byte = input.read() + if (byte == -1) { + return null + } + head.append(byte.toChar()) + when (byte) { + '\n'.code -> newlines++ + '\r'.code -> Unit + else -> newlines = 0 + } + } + return head.lineSequence().firstOrNull() + } +} + +/** + * Parks the worker inside the header loop: past the point where the task has bound its connection, + * but before anything has touched the network. That is the window a cancel has to be caught in, + * because [HttpURLConnection.disconnect] cannot tear down a connection that has not connected yet. + */ +private class ParkingConnection( + url: URL, + private val failDisconnect: Boolean = false, +) : HttpURLConnection(url) { + + val reachedHeaders = CountDownLatch(1) + val release = CountDownLatch(1) + + private val disconnects = AtomicInteger(0) + private val sentBody = ByteArrayOutputStream() + + @Volatile var openedOutputStream = false + private set + + /** cancel() and the worker's finally block both disconnect, so two means the worker unwound. */ + fun awaitUnwound(timeoutMs: Long): Boolean = await(timeoutMs) { disconnects.get() >= 2 } + + override fun setRequestProperty(key: String, value: String) { + reachedHeaders.countDown() + release.await() + super.setRequestProperty(key, value) + } + + override fun getOutputStream(): OutputStream { + openedOutputStream = true + return sentBody + } + + override fun getInputStream(): InputStream = ByteArrayInputStream(ByteArray(0)) + + override fun getResponseCode(): Int = 200 + + override fun getHeaderFields(): Map> = emptyMap() + + override fun connect() = Unit + + override fun disconnect() { + disconnects.incrementAndGet() + if (failDisconnect) { + throw IllegalStateException("disconnect failed") + } + } + + override fun usingProxy(): Boolean = false +} + +/** Settles at once, leaving its pool free to go dormant as soon as the keep-alive elapses. */ +private class ImmediateConnection(url: URL) : HttpURLConnection(url) { + + override fun getInputStream(): InputStream = ByteArrayInputStream(ByteArray(0)) + + override fun getResponseCode(): Int = 200 + + override fun getHeaderFields(): Map> = emptyMap() + + override fun connect() = Unit + + override fun disconnect() = Unit + + override fun usingProxy(): Boolean = false +} + +private class RecordingCompletion : HTTPRequestManagerCompletion() { + + private val latch = CountDownLatch(1) + + @Volatile var response: HTTPResponse? = null + private set + + @Volatile var error: String? = null + private set + + override fun onComplete(response: HTTPResponse) { + this.response = response + latch.countDown() + } + + override fun onFail(error: String) { + this.error = error + latch.countDown() + } + + fun awaitSettled(timeoutMs: Long): Boolean = latch.await(timeoutMs, TimeUnit.MILLISECONDS) + + fun hasSettled(): Boolean = latch.count == 0L + + fun describeOutcome(): String = response?.let { "onComplete(${it.statusCode})" } ?: "onFail($error)" +} + +@RunWith(RobolectricTestRunner::class) +@Config(sdk = [28], manifest = Config.NONE) +class DefaultHTTPRequestManagerTest { + + private val servers = mutableListOf() + private lateinit var manager: DefaultHTTPRequestManager + + @Before + fun setUp() { + manager = DefaultHTTPRequestManager(ApplicationProvider.getApplicationContext()) + } + + @After + fun tearDown() { + servers.forEach { it.close() } + } + + @Test(timeout = 30_000) + fun cancelDisconnectsInFlightRequest() { + val server = holdingServer() + val cancelable = perform(server.url("/hold")) + + assertTrue("the server never read the request", server.awaitRequestsRead(1, 5_000)) + + cancelable.cancel() + + assertTrue( + "cancel() left the connection open: the server never saw the client hang up", + server.awaitHungUp(1, 2_000), + ) + } + + @Test(timeout = 30_000) + fun cancelReturnsCapacityToThePool() { + val holding = holdingServer() + val inFlight = (0 until DefaultHTTPRequestManager.MAX_CONCURRENT_REQUESTS_PER_HOST).map { + perform(holding.url("/hold")) + } + + // Every worker has to be awaiting a response before cancelling. A connection that has not + // started its call yet cannot be torn down, because openConnection() does not connect and + // disconnect() is a no-op until it has. + assertTrue( + "the pool never filled", + holding.awaitRequestsRead(DefaultHTTPRequestManager.MAX_CONCURRENT_REQUESTS_PER_HOST, 5_000), + ) + + inFlight.forEach { it.cancel() } + + val expected = DefaultHTTPRequestManager.MAX_CONCURRENT_REQUESTS_PER_HOST + val allHungUp = holding.awaitHungUp(expected, 5_000) + assertTrue( + "cancel() tore down only ${holding.hungUpCount()} of $expected connections", + allHungUp, + ) + + val responding = respondingServer() + val completion = RecordingCompletion() + manager.performRequest(getRequest(responding.url("/probe")), completion) + val probeSettled = completion.awaitSettled(5_000) + + assertTrue( + "every connection was torn down but the pool never took a new request", + probeSettled, + ) + } + + @Test(timeout = 30_000) + fun cancelWhileQueuedNeitherConnectsNorCompletes() { + val server = holdingServer() + + // Saturate the pool so the next request has to queue. Waiting for every worker to be + // awaiting a response, rather than just for connections to be accepted, is what makes the + // next request certain to still be queued when it is cancelled. + repeat(DefaultHTTPRequestManager.MAX_CONCURRENT_REQUESTS_PER_HOST) { perform(server.url("/hold")) } + assertTrue( + "the pool never filled", + server.awaitRequestsRead(DefaultHTTPRequestManager.MAX_CONCURRENT_REQUESTS_PER_HOST, 5_000), + ) + + val queued = RecordingCompletion() + manager.performRequest(getRequest(server.url("/queued")), queued).cancel() + + // Drain the queue: unblock the workers and stop holding new connections. + server.startResponding() + server.releaseAll() + + // The queue is FIFO, so a probe submitted after the cancelled request cannot reach a + // worker before it. A settled probe means the cancelled request has had its turn. + val probe = RecordingCompletion() + manager.performRequest(getRequest(respondingServer().url("/probe")), probe) + assertTrue("the pool never drained", probe.awaitSettled(5_000)) + + assertFalse( + "a request cancelled while queued still opened a connection", + server.sawRequestFor("/queued"), + ) + assertFalse( + "a request cancelled while queued reported ${queued.describeOutcome()}", + queued.hasSettled(), + ) + } + + @Test(timeout = 30_000) + fun concurrentRequestsAreNotSerialised() { + val server = holdingServer() + val expected = DefaultHTTPRequestManager.MAX_CONCURRENT_REQUESTS_PER_HOST + + repeat(expected) { perform(server.url("/hold")) } + + val reachedAll = server.awaitAccepted(expected, 5_000) + + assertTrue( + "expected $expected connections in flight, saw ${server.acceptedCount()}", + reachedAll, + ) + } + + @Test(timeout = 30_000) + fun aStalledHostDoesNotBlockOtherHosts() { + val perHost = DefaultHTTPRequestManager.MAX_CONCURRENT_REQUESTS_PER_HOST + val stalled = holdingServer() + + repeat(perHost) { perform(stalled.url("/hold")) } + + // Every worker for that host has to be blocked in its read before the probe is sent. + // Otherwise a shared pool could still have spare capacity, and the probe would settle + // whether or not the pools are split by host. + assertTrue("the pool never filled", stalled.awaitRequestsRead(perHost, 5_000)) + + // The stalled server is addressed as 127.0.0.1, so reaching this one as localhost makes it + // a different host, and therefore a different pool. + val probe = RecordingCompletion() + manager.performRequest(getRequest(respondingServer().urlForHost("localhost", "/probe")), probe) + + assertTrue( + "a request to another host queued behind a stalled host", + probe.awaitSettled(5_000), + ) + } + + @Test(timeout = 30_000) + fun cancelSuppressesCompletion() { + val server = holdingServer() + val completion = RecordingCompletion() + val cancelable = manager.performRequest(getRequest(server.url("/hold")), completion) + + assertTrue("the server never read the request", server.awaitRequestsRead(1, 5_000)) + + cancelable.cancel() + + val settled = completion.awaitSettled(1_500) + + assertFalse("a cancelled request reported ${completion.describeOutcome()}", settled) + } + + @Test(timeout = 30_000) + fun cancelledPostNeverWritesItsBody() { + val url = "http://example.invalid/post" + val connection = ParkingConnection(URL(url)) + val postManager = DefaultHTTPRequestManager( + ApplicationProvider.getApplicationContext(), + ) { connection } + + val request = HTTPRequest( + url, + "POST", + hashMapOf("Content-Type" to "application/json"), + """{"cancelled":true}""".toByteArray(), + 0, + ) + + val cancelable = postManager.performRequest(request, RecordingCompletion()) + + assertTrue( + "the worker never reached header setup", + connection.reachedHeaders.await(5, TimeUnit.SECONDS), + ) + + cancelable.cancel() + connection.release.countDown() + + assertTrue("the worker never unwound", connection.awaitUnwound(5_000)) + assertFalse( + "a POST cancelled before it connected still wrote its body", + connection.openedOutputStream, + ) + } + + @Test(timeout = 30_000) + fun cancelSurvivesAThrowingDisconnect() { + val url = "http://example.invalid/throwing" + val connection = ParkingConnection(URL(url), failDisconnect = true) + val throwingManager = DefaultHTTPRequestManager( + ApplicationProvider.getApplicationContext(), + ) { connection } + + val request = HTTPRequest(url, "GET", hashMapOf("Accept" to "*/*"), null, 0) + val cancelable = throwingManager.performRequest(request, RecordingCompletion()) + + assertTrue( + "the worker never reached header setup", + connection.reachedHeaders.await(5, TimeUnit.SECONDS), + ) + + // cancel() is reached from native through the djinni Cancelable bridge, so anything it + // throws escapes into JNI rather than into a caller that can handle it. + try { + cancelable.cancel() + } catch (exception: Exception) { + connection.release.countDown() + fail("cancel() let $exception escape to its caller") + } + + connection.release.countDown() + + assertTrue("the worker never unwound", connection.awaitUnwound(5_000)) + } + + @Test(timeout = 30_000) + fun dormantHostPoolsAreEvicted() { + val evictingManager = evictingManager() + + repeat(DORMANT_HOSTS) { settle(evictingManager, "http://host-$it.invalid/probe") } + + // Only a request sweeps, and only once the keep-alive has left the pools threadless, so + // this has to keep making them until that window has passed. + val swept = await(5_000) { + settle(evictingManager, "http://sweeper.invalid/probe") + evictingManager.hostPools().none { it.startsWith("host-") } + } + + assertTrue("dormant pools were never evicted: ${evictingManager.hostPools()}", swept) + } + + @Test(timeout = 30_000) + fun anActiveHostPoolIsNotEvicted() { + val parked = ParkingConnection(URL("http://parked.invalid/hold")) + val evictingManager = DefaultHTTPRequestManager( + ApplicationProvider.getApplicationContext(), + SHORT_KEEP_ALIVE_MS, + ) { url -> if (url.host == "parked.invalid") parked else ImmediateConnection(url) } + + val completion = RecordingCompletion() + val request = HTTPRequest("http://parked.invalid/hold", "GET", hashMapOf("Accept" to "*/*"), null, 0) + evictingManager.performRequest(request, completion) + + assertTrue( + "the worker never reached header setup", + parked.reachedHeaders.await(5, TimeUnit.SECONDS), + ) + + repeat(DORMANT_HOSTS) { settle(evictingManager, "http://host-$it.invalid/probe") } + + val swept = await(5_000) { + settle(evictingManager, "http://sweeper.invalid/probe") + evictingManager.hostPools().none { it.startsWith("host-") } + } + + assertTrue("dormant pools were never evicted: ${evictingManager.hostPools()}", swept) + + assertTrue( + "the sweep evicted a pool with a request still in flight", + evictingManager.hostPools().contains("parked.invalid"), + ) + + parked.release.countDown() + + assertTrue( + "the request on the surviving pool never settled", + completion.awaitSettled(5_000), + ) + } + + @Test(timeout = 30_000) + fun aRequestToAnEvictedHostStillSucceeds() { + val evictingManager = evictingManager() + val evicted = "http://evicted.invalid/probe" + + settle(evictingManager, evicted) + repeat(DORMANT_HOSTS) { settle(evictingManager, "http://host-$it.invalid/probe") } + + val evictedThisHost = await(5_000) { + settle(evictingManager, "http://sweeper.invalid/probe") + !evictingManager.hostPools().contains("evicted.invalid") + } + + assertTrue("the dormant pool for the host was never evicted", evictedThisHost) + + // Eviction shuts the pool down, so reusing it here would reject the task and hang the + // completion rather than fail outright. + val completion = RecordingCompletion() + evictingManager.performRequest(getRequest(evicted), completion) + + assertTrue( + "a request to an evicted host reported ${completion.describeOutcome()}", + completion.awaitSettled(5_000), + ) + } + + /** Reaps pools in milliseconds rather than the production minute, so a sweep is observable. */ + private fun evictingManager(): DefaultHTTPRequestManager = + DefaultHTTPRequestManager( + ApplicationProvider.getApplicationContext(), + SHORT_KEEP_ALIVE_MS, + ) { ImmediateConnection(it) } + + private fun settle(manager: DefaultHTTPRequestManager, url: String) { + val completion = RecordingCompletion() + manager.performRequest(getRequest(url), completion) + assertTrue("the request to $url never settled", completion.awaitSettled(5_000)) + } + + private fun holdingServer(): FakeServer = FakeServer(respondImmediately = false).also { servers.add(it) } + + private fun respondingServer(): FakeServer = FakeServer(respondImmediately = true).also { servers.add(it) } + + private fun perform(url: String): Cancelable = + manager.performRequest(getRequest(url), RecordingCompletion()) + + private fun getRequest(url: String): HTTPRequest = + HTTPRequest(url, "GET", HashMap(), null, 0) +}