From 113e37261d1fb5c1ae99e562fe34424d539ad9ff Mon Sep 17 00:00:00 2001 From: "stainless-app[bot]" <142633134+stainless-app[bot]@users.noreply.github.com> Date: Wed, 29 Jan 2025 23:47:08 +0000 Subject: [PATCH] fix(client): don't leak responses when retrying (#103) --- .../anthropic/core/http/RetryingHttpClient.kt | 36 +++++++------ .../core/http/RetryingHttpClientTest.kt | 52 ++++++++++++++++++- 2 files changed, 71 insertions(+), 17 deletions(-) diff --git a/anthropic-java-core/src/main/kotlin/com/anthropic/core/http/RetryingHttpClient.kt b/anthropic-java-core/src/main/kotlin/com/anthropic/core/http/RetryingHttpClient.kt index 2e441e7d..ae3fb89f 100644 --- a/anthropic-java-core/src/main/kotlin/com/anthropic/core/http/RetryingHttpClient.kt +++ b/anthropic-java-core/src/main/kotlin/com/anthropic/core/http/RetryingHttpClient.kt @@ -57,15 +57,17 @@ private constructor( } response - } catch (t: Throwable) { - if (++retries > maxRetries || !shouldRetry(t)) { - throw t + } catch (throwable: Throwable) { + if (++retries > maxRetries || !shouldRetry(throwable)) { + throw throwable } null } val backoffMillis = getRetryBackoffMillis(retries, response) + // All responses must be closed, so close the failed one before retrying. + response?.close() Thread.sleep(backoffMillis.toMillis()) } } @@ -113,6 +115,8 @@ private constructor( } val backoffMillis = getRetryBackoffMillis(retries, response) + // All responses must be closed, so close the failed one before retrying. + response?.close() return sleepAsync(backoffMillis.toMillis()).thenCompose { executeWithRetries(requestWithRetryCount, requestOptions) } @@ -223,23 +227,23 @@ private constructor( return Duration.ofNanos((TimeUnit.SECONDS.toNanos(1) * backoffSeconds * jitter).toLong()) } - private fun sleepAsync(millis: Long): CompletableFuture { - val future = CompletableFuture() - TIMER.schedule( - object : TimerTask() { - override fun run() { - future.complete(null) - } - }, - millis - ) - return future - } - companion object { private val TIMER = Timer("RetryingHttpClient", true) + private fun sleepAsync(millis: Long): CompletableFuture { + val future = CompletableFuture() + TIMER.schedule( + object : TimerTask() { + override fun run() { + future.complete(null) + } + }, + millis + ) + return future + } + @JvmStatic fun builder() = Builder() } diff --git a/anthropic-java-core/src/test/kotlin/com/anthropic/core/http/RetryingHttpClientTest.kt b/anthropic-java-core/src/test/kotlin/com/anthropic/core/http/RetryingHttpClientTest.kt index c6f5a396..336e4c28 100644 --- a/anthropic-java-core/src/test/kotlin/com/anthropic/core/http/RetryingHttpClientTest.kt +++ b/anthropic-java-core/src/test/kotlin/com/anthropic/core/http/RetryingHttpClientTest.kt @@ -1,10 +1,13 @@ package com.anthropic.core.http import com.anthropic.client.okhttp.OkHttpClient +import com.anthropic.core.RequestOptions import com.github.tomakehurst.wiremock.client.WireMock.* import com.github.tomakehurst.wiremock.junit5.WireMockRuntimeInfo import com.github.tomakehurst.wiremock.junit5.WireMockTest import com.github.tomakehurst.wiremock.stubbing.Scenario +import java.io.InputStream +import java.util.concurrent.CompletableFuture import org.assertj.core.api.Assertions.assertThat import org.junit.jupiter.api.BeforeEach import org.junit.jupiter.params.ParameterizedTest @@ -13,11 +16,49 @@ import org.junit.jupiter.params.provider.ValueSource @WireMockTest internal class RetryingHttpClientTest { + private var openResponseCount = 0 private lateinit var httpClient: HttpClient @BeforeEach fun beforeEach(wmRuntimeInfo: WireMockRuntimeInfo) { - httpClient = OkHttpClient.builder().baseUrl(wmRuntimeInfo.httpBaseUrl).build() + val okHttpClient = OkHttpClient.builder().baseUrl(wmRuntimeInfo.httpBaseUrl).build() + httpClient = + object : HttpClient { + override fun execute( + request: HttpRequest, + requestOptions: RequestOptions + ): HttpResponse = trackClose(okHttpClient.execute(request, requestOptions)) + + override fun executeAsync( + request: HttpRequest, + requestOptions: RequestOptions + ): CompletableFuture = + okHttpClient.executeAsync(request, requestOptions).thenApply { trackClose(it) } + + override fun close() = okHttpClient.close() + + private fun trackClose(response: HttpResponse): HttpResponse { + openResponseCount++ + return object : HttpResponse { + private var isClosed = false + + override fun statusCode(): Int = response.statusCode() + + override fun headers(): Headers = response.headers() + + override fun body(): InputStream = response.body() + + override fun close() { + response.close() + if (isClosed) { + return + } + openResponseCount-- + isClosed = true + } + } + } + } resetAllScenarios() } @@ -35,6 +76,7 @@ internal class RetryingHttpClientTest { assertThat(response.statusCode()).isEqualTo(200) verify(1, postRequestedFor(urlPathEqualTo("/something"))) + assertNoResponseLeaks() } @ParameterizedTest @@ -60,6 +102,7 @@ internal class RetryingHttpClientTest { assertThat(response.statusCode()).isEqualTo(200) verify(1, postRequestedFor(urlPathEqualTo("/something"))) + assertNoResponseLeaks() } @ParameterizedTest @@ -116,6 +159,7 @@ internal class RetryingHttpClientTest { postRequestedFor(urlPathEqualTo("/something")) .withHeader("x-stainless-retry-count", equalTo("2")) ) + assertNoResponseLeaks() } @ParameterizedTest @@ -156,6 +200,7 @@ internal class RetryingHttpClientTest { postRequestedFor(urlPathEqualTo("/something")) .withHeader("x-stainless-retry-count", equalTo("42")) ) + assertNoResponseLeaks() } @ParameterizedTest @@ -186,8 +231,13 @@ internal class RetryingHttpClientTest { assertThat(response.statusCode()).isEqualTo(200) verify(2, postRequestedFor(urlPathEqualTo("/something"))) + assertNoResponseLeaks() } private fun HttpClient.execute(request: HttpRequest, async: Boolean): HttpResponse = if (async) executeAsync(request).get() else execute(request) + + // When retrying, all failed responses should be closed. Only the final returned response should + // be open. + private fun assertNoResponseLeaks() = assertThat(openResponseCount).isEqualTo(1) }