From 85d02e6bc2fc6efdef603c239abd28292a931fb2 Mon Sep 17 00:00:00 2001 From: tx3stn <14163530+tx3stn@users.noreply.github.com> Date: Sat, 8 Aug 2026 16:40:55 +0100 Subject: [PATCH 1/4] fix: align android http connection concurrency with ios and fix no-op cancelation --- .../network/DefaultHTTPRequestManager.kt | 83 ++++- .../network/DefaultHTTPRequestManagerTest.kt | 336 ++++++++++++++++++ 2 files changed, 415 insertions(+), 4 deletions(-) create mode 100644 valdi/test/java/network/DefaultHTTPRequestManagerTest.kt diff --git a/valdi/src/java/com/snap/valdi/network/DefaultHTTPRequestManager.kt b/valdi/src/java/com/snap/valdi/network/DefaultHTTPRequestManager.kt index e0eefca1..d63894ce 100644 --- a/valdi/src/java/com/snap/valdi/network/DefaultHTTPRequestManager.kt +++ b/valdi/src/java/com/snap/valdi/network/DefaultHTTPRequestManager.kt @@ -1,11 +1,15 @@ package com.snap.valdi.network import android.content.Context -import com.snap.valdi.utils.ExecutorsUtil +import java.io.IOException import java.io.InputStream import java.net.HttpURLConnection import java.net.URL -import java.util.concurrent.Executors +import java.util.concurrent.ExecutorService +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.* @@ -13,6 +17,38 @@ class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() { private class RequestTask(val url: URL, val method: String, val body: ByteArray?, val headers: Map, 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 is what 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. + connectionToClose?.disconnect() + } + + 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 @@ -31,6 +67,13 @@ class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() { urlConnection.outputStream.close() } + // Reading responseCode is what opens the socket, and disconnect() cannot tear down + // a connection that has not connected yet. Checking here keeps a cancel that + // arrived during setup from starting a request nobody is waiting for. + if (isCancelled()) { + throw IOException("Request was cancelled") + } + val responseCode = urlConnection.responseCode val responseHeaders = hashMapOf() @@ -56,6 +99,11 @@ class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() { val urlConnection = url.openConnection() if (urlConnection is HttpURLConnection) { + if (!bindConnection(urlConnection)) { + urlConnection.disconnect() + throw IOException("Request was cancelled") + } + return doPerformRequestWithURLConnection(urlConnection) } else { urlConnection.doInput = true @@ -66,10 +114,17 @@ 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) { + // Once cancelled the completion is already gone, so a teardown IOException lands + // here and goes nowhere -- matching how iOS swallows NSURLErrorCancelled. notifyFailure("HTTP Request failed: ${error.message}") } } @@ -98,11 +153,24 @@ class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() { } } - private var executors = ExecutorsUtil.newSingleThreadCachedExecutor { r -> + private val threadCount = AtomicInteger(0) + + private var executors: ExecutorService = ThreadPoolExecutor( + MAX_CONCURRENT_REQUESTS, + MAX_CONCURRENT_REQUESTS, + 60L, + TimeUnit.SECONDS, + LinkedBlockingQueue(), + ) { r -> Thread(r).apply { - name = "Valdi Network Thread" + 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) } override fun performRequest(request: HTTPRequest, completion: HTTPRequestManagerCompletion): Cancelable { @@ -123,4 +191,11 @@ class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() { return task } + companion object { + // Matches NSURLSession.HTTPMaximumConnectionsPerHost, which the iOS and macOS default + // request managers inherit. That limit is per host and this one is total, so it is the + // conservative reading of the same number. + const val MAX_CONCURRENT_REQUESTS = 4 + } + } diff --git a/valdi/test/java/network/DefaultHTTPRequestManagerTest.kt b/valdi/test/java/network/DefaultHTTPRequestManagerTest.kt new file mode 100644 index 00000000..a1b9be34 --- /dev/null +++ b/valdi/test/java/network/DefaultHTTPRequestManagerTest.kt @@ -0,0 +1,336 @@ +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.Before +import org.junit.Test +import org.junit.runner.RunWith +import org.robolectric.RobolectricTestRunner +import org.robolectric.annotation.Config +import java.io.Closeable +import java.io.InputStream +import java.net.InetAddress +import java.net.ServerSocket +import java.net.Socket +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 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" + + 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 } + + /** Block until the accept count has stopped moving. */ + fun settleAcceptedCount(settleMs: Long) { + var previous = -1 + while (accepted.get() != previous) { + previous = accepted.get() + Thread.sleep(settleMs) + } + } + + 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() + } +} + +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 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() { + // Keep-alive would let a later request reuse an earlier socket, which breaks the + // accepted-connection counting these tests rely on. + System.setProperty("http.keepAlive", "false") + 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).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, 5_000), + ) + + inFlight.forEach { it.cancel() } + + val expected = DefaultHTTPRequestManager.MAX_CONCURRENT_REQUESTS + 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 cancelWhileQueuedNeverOpensAConnection() { + val server = holdingServer() + + // Saturate the pool so the next request has to queue. + repeat(DefaultHTTPRequestManager.MAX_CONCURRENT_REQUESTS) { perform(server.url("/hold")) } + assertTrue(server.awaitAccepted(1, 5_000)) + + perform(server.url("/queued")).cancel() + + // Drain the queue: unblock the workers and stop holding new connections. + server.startResponding() + server.releaseAll() + server.settleAcceptedCount(500) + + assertFalse( + "a request cancelled while queued still opened a connection", + server.sawRequestFor("/queued"), + ) + } + + @Test(timeout = 30_000) + fun concurrentRequestsAreNotSerialised() { + val server = holdingServer() + val expected = DefaultHTTPRequestManager.MAX_CONCURRENT_REQUESTS + + 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 cancelSuppressesCompletion() { + val server = holdingServer() + val completion = RecordingCompletion() + val cancelable = manager.performRequest(getRequest(server.url("/hold")), completion) + + assertTrue(server.awaitAccepted(1, 5_000)) + + cancelable.cancel() + + val settled = completion.awaitSettled(1_500) + + assertFalse("a cancelled request reported ${completion.describeOutcome()}", settled) + } + + @Test(timeout = 30_000) + fun cancelBeforeConnectionOpens() { + val server = holdingServer() + val completion = RecordingCompletion() + + manager.performRequest(getRequest(server.url("/hold")), completion).cancel() + + val settled = completion.awaitSettled(1_500) + + assertFalse("a request cancelled before it ran reported ${completion.describeOutcome()}", settled) + } + + 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) +} From b8001885bfa8f1710eec7c045f3d47352042f577 Mon Sep 17 00:00:00 2001 From: tx3stn <14163530+tx3stn@users.noreply.github.com> Date: Sat, 8 Aug 2026 16:57:41 +0100 Subject: [PATCH 2/4] fix: check for cancelation before writing body to properly support POST/PATCH cancelation --- .../network/DefaultHTTPRequestManager.kt | 36 ++++---- .../network/DefaultHTTPRequestManagerTest.kt | 83 +++++++++++++++++++ 2 files changed, 102 insertions(+), 17 deletions(-) diff --git a/valdi/src/java/com/snap/valdi/network/DefaultHTTPRequestManager.kt b/valdi/src/java/com/snap/valdi/network/DefaultHTTPRequestManager.kt index d63894ce..63805a94 100644 --- a/valdi/src/java/com/snap/valdi/network/DefaultHTTPRequestManager.kt +++ b/valdi/src/java/com/snap/valdi/network/DefaultHTTPRequestManager.kt @@ -5,6 +5,7 @@ import java.io.IOException import java.io.InputStream import java.net.HttpURLConnection import java.net.URL +import java.net.URLConnection import java.util.concurrent.ExecutorService import java.util.concurrent.LinkedBlockingQueue import java.util.concurrent.ThreadPoolExecutor @@ -13,9 +14,12 @@ import java.util.concurrent.atomic.AtomicInteger import com.snapchat.client.valdi.* import com.snapchat.client.valdi_core.* -class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() { +class DefaultHTTPRequestManager( + context: Context, + private val openConnection: (URL) -> URLConnection = { it.openConnection() }, +): HTTPRequestManager() { - private class RequestTask(val url: URL, val method: String, val body: ByteArray?, val headers: Map, completion: HTTPRequestManagerCompletion): HTTPRequestTask(completion), Runnable { + 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 @@ -28,7 +32,7 @@ class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() { connection } - // Closing from another thread is what makes the worker's blocked read throw, which is + // 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. connectionToClose?.disconnect() @@ -61,19 +65,20 @@ 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) urlConnection.outputStream.close() } - // Reading responseCode is what opens the socket, and disconnect() cannot tear down - // a connection that has not connected yet. Checking here keeps a cancel that - // arrived during setup from starting a request nobody is waiting for. - if (isCancelled()) { - throw IOException("Request was cancelled") - } - val responseCode = urlConnection.responseCode val responseHeaders = hashMapOf() @@ -96,7 +101,7 @@ class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() { } private fun performRequest(): HTTPResponse { - val urlConnection = url.openConnection() + val urlConnection = openConnection(url) if (urlConnection is HttpURLConnection) { if (!bindConnection(urlConnection)) { @@ -131,7 +136,7 @@ class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() { 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 @@ -148,7 +153,7 @@ class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() { } } - return RequestTask(url, method, body, headers, completion) + return RequestTask(url, method, body, headers, openConnection, completion) } } } @@ -176,7 +181,7 @@ class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() { override fun performRequest(request: HTTPRequest, completion: HTTPRequestManagerCompletion): Cancelable { val task: RequestTask try { - task = RequestTask.from(request, completion) + task = RequestTask.from(request, openConnection, completion) } catch (exception: Exception) { completion.onFail("Failed to build request: ${exception.message}") @@ -192,9 +197,6 @@ class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() { } companion object { - // Matches NSURLSession.HTTPMaximumConnectionsPerHost, which the iOS and macOS default - // request managers inherit. That limit is per host and this one is total, so it is the - // conservative reading of the same number. const val MAX_CONCURRENT_REQUESTS = 4 } diff --git a/valdi/test/java/network/DefaultHTTPRequestManagerTest.kt b/valdi/test/java/network/DefaultHTTPRequestManagerTest.kt index a1b9be34..17424fad 100644 --- a/valdi/test/java/network/DefaultHTTPRequestManagerTest.kt +++ b/valdi/test/java/network/DefaultHTTPRequestManagerTest.kt @@ -14,11 +14,16 @@ 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 @@ -166,6 +171,51 @@ private class FakeServer(respondImmediately: Boolean) : Closeable { } } +/** + * 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) : 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() + } + + override fun usingProxy(): Boolean = false +} + private class RecordingCompletion : HTTPRequestManagerCompletion() { private val latch = CountDownLatch(1) @@ -324,6 +374,39 @@ class DefaultHTTPRequestManagerTest { assertFalse("a request cancelled before it ran 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, + ) + } + private fun holdingServer(): FakeServer = FakeServer(respondImmediately = false).also { servers.add(it) } private fun respondingServer(): FakeServer = FakeServer(respondImmediately = true).also { servers.add(it) } From d5e298a2991bd3d8747d6c2eda91e126e1254ff1 Mon Sep 17 00:00:00 2001 From: tx3stn <14163530+tx3stn@users.noreply.github.com> Date: Sat, 8 Aug 2026 17:13:35 +0100 Subject: [PATCH 3/4] fix: cancel should guard on disconnect --- .../network/DefaultHTTPRequestManager.kt | 7 +++- .../network/DefaultHTTPRequestManagerTest.kt | 39 ++++++++++++++++++- 2 files changed, 43 insertions(+), 3 deletions(-) diff --git a/valdi/src/java/com/snap/valdi/network/DefaultHTTPRequestManager.kt b/valdi/src/java/com/snap/valdi/network/DefaultHTTPRequestManager.kt index 63805a94..8c25c5b5 100644 --- a/valdi/src/java/com/snap/valdi/network/DefaultHTTPRequestManager.kt +++ b/valdi/src/java/com/snap/valdi/network/DefaultHTTPRequestManager.kt @@ -34,8 +34,11 @@ class DefaultHTTPRequestManager( // 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. - connectionToClose?.disconnect() + // 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 } diff --git a/valdi/test/java/network/DefaultHTTPRequestManagerTest.kt b/valdi/test/java/network/DefaultHTTPRequestManagerTest.kt index 17424fad..6a3dbf63 100644 --- a/valdi/test/java/network/DefaultHTTPRequestManagerTest.kt +++ b/valdi/test/java/network/DefaultHTTPRequestManagerTest.kt @@ -9,6 +9,7 @@ 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 @@ -176,7 +177,10 @@ private class FakeServer(respondImmediately: Boolean) : Closeable { * 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) : HttpURLConnection(url) { +private class ParkingConnection( + url: URL, + private val failDisconnect: Boolean = false, +) : HttpURLConnection(url) { val reachedHeaders = CountDownLatch(1) val release = CountDownLatch(1) @@ -211,6 +215,9 @@ private class ParkingConnection(url: URL) : HttpURLConnection(url) { override fun disconnect() { disconnects.incrementAndGet() + if (failDisconnect) { + throw IllegalStateException("disconnect failed") + } } override fun usingProxy(): Boolean = false @@ -407,6 +414,36 @@ class DefaultHTTPRequestManagerTest { ) } + @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)) + } + private fun holdingServer(): FakeServer = FakeServer(respondImmediately = false).also { servers.add(it) } private fun respondingServer(): FakeServer = FakeServer(respondImmediately = true).also { servers.add(it) } From 0bfcae7a5dfdf271fae1f948930c9cbd859676c9 Mon Sep 17 00:00:00 2001 From: tx3stn <14163530+tx3stn@users.noreply.github.com> Date: Sat, 8 Aug 2026 17:38:27 +0100 Subject: [PATCH 4/4] test: harden the unit tests --- .../network/DefaultHTTPRequestManager.kt | 2 +- .../network/DefaultHTTPRequestManagerTest.kt | 53 ++++++++----------- 2 files changed, 24 insertions(+), 31 deletions(-) diff --git a/valdi/src/java/com/snap/valdi/network/DefaultHTTPRequestManager.kt b/valdi/src/java/com/snap/valdi/network/DefaultHTTPRequestManager.kt index 8c25c5b5..c4d09c60 100644 --- a/valdi/src/java/com/snap/valdi/network/DefaultHTTPRequestManager.kt +++ b/valdi/src/java/com/snap/valdi/network/DefaultHTTPRequestManager.kt @@ -163,7 +163,7 @@ class DefaultHTTPRequestManager( private val threadCount = AtomicInteger(0) - private var executors: ExecutorService = ThreadPoolExecutor( + private val executors: ExecutorService = ThreadPoolExecutor( MAX_CONCURRENT_REQUESTS, MAX_CONCURRENT_REQUESTS, 60L, diff --git a/valdi/test/java/network/DefaultHTTPRequestManagerTest.kt b/valdi/test/java/network/DefaultHTTPRequestManagerTest.kt index 6a3dbf63..d05993d0 100644 --- a/valdi/test/java/network/DefaultHTTPRequestManagerTest.kt +++ b/valdi/test/java/network/DefaultHTTPRequestManagerTest.kt @@ -86,15 +86,6 @@ private class FakeServer(respondImmediately: Boolean) : Closeable { fun awaitHungUp(count: Int, timeoutMs: Long): Boolean = await(timeoutMs) { hungUp.get() >= count } - /** Block until the accept count has stopped moving. */ - fun settleAcceptedCount(settleMs: Long) { - var previous = -1 - while (accepted.get() != previous) { - previous = accepted.get() - Thread.sleep(settleMs) - } - } - fun startResponding() { respond = true } @@ -245,6 +236,8 @@ private class RecordingCompletion : HTTPRequestManagerCompletion() { 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)" } @@ -257,9 +250,6 @@ class DefaultHTTPRequestManagerTest { @Before fun setUp() { - // Keep-alive would let a later request reuse an earlier socket, which breaks the - // accepted-connection counting these tests rely on. - System.setProperty("http.keepAlive", "false") manager = DefaultHTTPRequestManager(ApplicationProvider.getApplicationContext()) } @@ -319,24 +309,39 @@ class DefaultHTTPRequestManagerTest { } @Test(timeout = 30_000) - fun cancelWhileQueuedNeverOpensAConnection() { + fun cancelWhileQueuedNeitherConnectsNorCompletes() { val server = holdingServer() - // Saturate the pool so the next request has to queue. + // 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) { perform(server.url("/hold")) } - assertTrue(server.awaitAccepted(1, 5_000)) + assertTrue( + "the pool never filled", + server.awaitRequestsRead(DefaultHTTPRequestManager.MAX_CONCURRENT_REQUESTS, 5_000), + ) - perform(server.url("/queued")).cancel() + 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() - server.settleAcceptedCount(500) + + // 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) @@ -360,7 +365,7 @@ class DefaultHTTPRequestManagerTest { val completion = RecordingCompletion() val cancelable = manager.performRequest(getRequest(server.url("/hold")), completion) - assertTrue(server.awaitAccepted(1, 5_000)) + assertTrue("the server never read the request", server.awaitRequestsRead(1, 5_000)) cancelable.cancel() @@ -369,18 +374,6 @@ class DefaultHTTPRequestManagerTest { assertFalse("a cancelled request reported ${completion.describeOutcome()}", settled) } - @Test(timeout = 30_000) - fun cancelBeforeConnectionOpens() { - val server = holdingServer() - val completion = RecordingCompletion() - - manager.performRequest(getRequest(server.url("/hold")), completion).cancel() - - val settled = completion.awaitSettled(1_500) - - assertFalse("a request cancelled before it ran reported ${completion.describeOutcome()}", settled) - } - @Test(timeout = 30_000) fun cancelledPostNeverWritesItsBody() { val url = "http://example.invalid/post"