Close intermediate redirect response bodies before following (#1749)

When following redirects, `RequestRewritingClient` discarded each 3xx
response without closing its body, leaking the underlying connection.

Close the body as soon as we know the status is a redirect and won't
be returned, so the too-many-redirects, missing-`Location`,
invalid-URI and downgrade error paths all release the connection too.

This prevents leaking one `BodyHandlers.ofInputStream()` connection
per hop. Now the code closes the body before following the redirect.

pkl-core/src/main/java/org/pkl/core/http/RequestRewritingClient.java

@@ -16,6 +16,7 @@
 package org.pkl.core.http;
 
 import com.google.errorprone.annotations.ThreadSafe;
+import java.io.Closeable;
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
@@ -118,6 +119,9 @@ final class RequestRewritingClient implements HttpClient {
       if (!HttpUtils.isRedirectStatusCode(response.statusCode())) {
         return response;
       }
+      if (response.body() instanceof Closeable closeable) {
+        closeable.close();
+      }
       if (redirectCount >= MAX_HTTP_REDIRECTS) {
         throw new HttpClientException(
             ErrorMessages.create("httpTooManyRedirects", MAX_HTTP_REDIRECTS));

pkl-core/src/test/kotlin/org/pkl/core/http/RequestRewritingClientTest.kt

@@ -15,6 +15,8 @@
  */
 package org.pkl.core.http
 
+import java.io.ByteArrayInputStream
+import java.io.InputStream
 import java.net.URI
 import java.net.http.HttpClient as JdkHttpClient
 import java.net.http.HttpRequest
@@ -25,6 +27,7 @@ import java.util.Locale
 import org.assertj.core.api.Assertions.assertThat
 import org.assertj.core.api.Assertions.assertThatList
 import org.junit.jupiter.api.Test
+import org.pkl.commons.test.FakeHttpResponse
 import org.pkl.core.util.GlobResolver
 import org.pkl.core.util.IoUtils
 
@@ -441,4 +444,38 @@ class RequestRewritingClientTest {
     assertThatList(captured.request.headers().allValues("user-agent"))
       .containsExactly("My User Agent")
   }
+
+  @Test
+  fun `closes intermediate redirect response body before following the redirect`() {
+    val intermediateBody = CloseTrackingInputStream()
+    val delegate =
+      ReplayingClient(
+        FakeHttpResponse<InputStream>().apply {
+          statusCode = 302
+          headers = httpHeaders("Location", "/bar.pkl")
+          body = intermediateBody
+        },
+        FakeHttpResponse<InputStream>(),
+      )
+    val client =
+      RequestRewritingClient("Pkl", Duration.ofSeconds(42), -1, delegate, mapOf(), mapOf())
+    val request = HttpRequest.newBuilder(URI("https://example.com/foo.pkl")).build()
+
+    client.send(request, BodyHandlers.ofInputStream(), NoopChecker)
+
+    assertThat(intermediateBody.closed)
+      .`as`("intermediate 3xx response body must be closed to release its connection")
+      .isTrue
+  }
+
+  /** @ByteArrayInputStream that records whether it was closed. */
+  private class CloseTrackingInputStream : ByteArrayInputStream(ByteArray(0)) {
+    var closed = false
+      private set
+
+    override fun close() {
+      closed = true
+      super.close()
+    }
+  }
 }

pkl-core/src/test/kotlin/org/pkl/core/http/util.kt

@@ -16,6 +16,9 @@
 package org.pkl.core.http
 
 import java.net.URI
+import java.net.http.HttpHeaders
+import java.net.http.HttpRequest
+import java.net.http.HttpResponse
 import java.util.regex.Pattern
 
 data class HttpSettings(
@@ -32,3 +35,21 @@ fun HttpClient.getConfiguredSettings(): HttpSettings {
 object NoopChecker : HttpClient.HttpRequestChecker {
   override fun check(uri: URI) {}
 }
+
+fun httpHeaders(name: String, value: String): HttpHeaders =
+  HttpHeaders.of(mapOf(name to listOf(value))) { _, _ -> true }
+
+class ReplayingClient(vararg responses: HttpResponse<*>) : HttpClient {
+  private val responses = ArrayDeque(responses.toList())
+
+  @Suppress("UNCHECKED_CAST")
+  override fun <T : Any> send(
+    request: HttpRequest,
+    responseBodyHandler: HttpResponse.BodyHandler<T>,
+    httpRequestChecker: HttpClient.HttpRequestChecker,
+  ): HttpResponse<T> {
+    return responses.removeFirst() as HttpResponse<T>
+  }
+
+  override fun close() {}
+}