fix: send CONNECT first when recovering a HTTPS request (#2077)

# Issue description

AHC has retry mechanism enabled with up to 5 attempts by default. But
the initial CONNECT is omitted when recovering the HTTPS requests with
IO exceptions. This MR fixes this issue and guarantees the proper
workflow in retries.

It's related to #2071 and fixes a different failing case.

# How the issue is fixed

* For any new connections, make sure there is an initial CONNECT for
WebSocket/HTTPS request.
* For the condition check that a CONNECT has been sent, make sure the
connection the current future attaches is reusable/active.

# Unit test

IOException has various reasons but in the unit test, we emulate it by
closing the connection after receiving the CONNECT request. The internal
recovery process will retry another 4 times, and through an IOException
eventually.

Signed-off-by: Jason Joo <[email protected]>

Author: jasonjoo2010

client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java

@@ -97,6 +97,13 @@ public NettyRequestSender(AsyncHttpClientConfig config, ChannelManager channelMa
         requestFactory = new NettyRequestFactory(config);
     }
 
+    // needConnect returns true if the request is secure/websocket and a HTTP proxy is set
+    private boolean needConnect(final Request request, final ProxyServer proxyServer) {
+        return proxyServer != null
+                && proxyServer.getProxyType().isHttp()
+                && (request.getUri().isSecured() || request.getUri().isWebSocket());
+    }
+
     public <T> ListenableFuture<T> sendRequest(final Request request, final AsyncHandler<T> asyncHandler, NettyResponseFuture<T> future) {
         if (isClosed()) {
             throw new IllegalStateException("Closed");
@@ -106,9 +113,7 @@ public <T> ListenableFuture<T> sendRequest(final Request request, final AsyncHan
         ProxyServer proxyServer = getProxyServer(config, request);
 
         // WebSockets use connect tunneling to work with proxies
-        if (proxyServer != null && proxyServer.getProxyType().isHttp() &&
-                (request.getUri().isSecured() || request.getUri().isWebSocket()) &&
-                !isConnectAlreadyDone(request, future)) {
+        if (needConnect(request, proxyServer) && !isConnectAlreadyDone(request, future)) {
             // Proxy with HTTPS or WebSocket: CONNECT for sure
             if (future != null && future.isConnectAllowed()) {
                 // Perform CONNECT
@@ -125,6 +130,8 @@ public <T> ListenableFuture<T> sendRequest(final Request request, final AsyncHan
 
     private static boolean isConnectAlreadyDone(Request request, NettyResponseFuture<?> future) {
         return future != null
+                // If the channel can't be reused or closed, a CONNECT is still required
+                && future.isReuseChannel() && Channels.isChannelActive(future.channel())
                 && future.getNettyRequest() != null
                 && future.getNettyRequest().getHttpRequest().method() == HttpMethod.CONNECT
                 && !request.getMethod().equals(CONNECT);
@@ -137,11 +144,19 @@ private static boolean isConnectAlreadyDone(Request request, NettyResponseFuture
      */
     private <T> ListenableFuture<T> sendRequestWithCertainForceConnect(Request request, AsyncHandler<T> asyncHandler, NettyResponseFuture<T> future,
                                                                        ProxyServer proxyServer, boolean performConnectRequest) {
-        NettyResponseFuture<T> newFuture = newNettyRequestAndResponseFuture(request, asyncHandler, future, proxyServer, performConnectRequest);
         Channel channel = getOpenChannel(future, request, proxyServer, asyncHandler);
-        return Channels.isChannelActive(channel)
-                ? sendRequestWithOpenChannel(newFuture, asyncHandler, channel)
-                : sendRequestWithNewChannel(request, proxyServer, newFuture, asyncHandler);
+        if (Channels.isChannelActive(channel)) {
+            NettyResponseFuture<T> newFuture = newNettyRequestAndResponseFuture(request, asyncHandler, future,
+                    proxyServer, performConnectRequest);
+            return sendRequestWithOpenChannel(newFuture, asyncHandler, channel);
+        } else {
+            // A new channel is not expected when performConnectRequest is false. We need to
+            // revisit the condition of sending
+            // the CONNECT request to the new channel.
+            NettyResponseFuture<T> newFuture = newNettyRequestAndResponseFuture(request, asyncHandler, future,
+                    proxyServer, needConnect(request, proxyServer));
+            return sendRequestWithNewChannel(request, proxyServer, newFuture, asyncHandler);
+        }
     }
 
     /**

client/src/test/java/org/asynchttpclient/proxy/HttpsProxyTest.java

@@ -13,6 +13,7 @@
 package org.asynchttpclient.proxy;
 
 import io.github.artsok.RepeatedIfExceptionsTest;
+import io.netty.handler.codec.http.DefaultHttpHeaders;
 import jakarta.servlet.ServletException;
 import jakarta.servlet.http.HttpServletRequest;
 import jakarta.servlet.http.HttpServletResponse;
@@ -43,8 +44,10 @@
 import static org.asynchttpclient.test.TestUtils.addHttpConnector;
 import static org.asynchttpclient.test.TestUtils.addHttpsConnector;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrowsExactly;
 
 import java.io.IOException;
+import java.util.concurrent.ExecutionException;
 
 /**
  * Proxy usage tests.
@@ -156,7 +159,7 @@ public void testPooledConnectionsWithProxy() throws Exception {
     public void testFailedConnectWithProxy() throws Exception {
         try (AsyncHttpClient asyncHttpClient = asyncHttpClient(config().setFollowRedirect(true).setUseInsecureTrustManager(true).setKeepAlive(true))) {
         	Builder proxyServer = proxyServer("localhost", port1);
-        	proxyServer.setCustomHeaders(r -> r.getHeaders().add(ProxyHandler.HEADER_FORBIDDEN, "1"));
+        	proxyServer.setCustomHeaders(r -> new DefaultHttpHeaders().set(ProxyHandler.HEADER_FORBIDDEN, "1"));
             RequestBuilder rb = get(getTargetUrl2()).setProxyServer(proxyServer);
 
             Response response1 = asyncHttpClient.executeRequest(rb.build()).get();
@@ -170,16 +173,39 @@ public void testFailedConnectWithProxy() throws Exception {
         }
     }
 
+    @RepeatedIfExceptionsTest(repeats = 5)
+    public void testClosedConnectionWithProxy() throws Exception {
+        try (AsyncHttpClient asyncHttpClient = asyncHttpClient(
+                config().setFollowRedirect(true).setUseInsecureTrustManager(true).setKeepAlive(true))) {
+            Builder proxyServer = proxyServer("localhost", port1);
+            proxyServer.setCustomHeaders(r -> new DefaultHttpHeaders().set(ProxyHandler.HEADER_FORBIDDEN, "2"));
+            RequestBuilder rb = get(getTargetUrl2()).setProxyServer(proxyServer);
+
+            assertThrowsExactly(ExecutionException.class, () -> asyncHttpClient.executeRequest(rb.build()).get());
+            assertThrowsExactly(ExecutionException.class, () -> asyncHttpClient.executeRequest(rb.build()).get());
+            assertThrowsExactly(ExecutionException.class, () -> asyncHttpClient.executeRequest(rb.build()).get());
+        }
+    }
+
     public static class ProxyHandler extends ConnectHandler {
     	final static String HEADER_FORBIDDEN = "X-REJECT-REQUEST";
 
         @Override
         public void handle(String s, Request r, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException {
             if (HttpConstants.Methods.CONNECT.equalsIgnoreCase(request.getMethod())) {
-                if (request.getHeader(HEADER_FORBIDDEN) != null) {
+                String headerValue = request.getHeader(HEADER_FORBIDDEN);
+                if (headerValue == null) {
+                    headerValue = "";
+                }
+                switch (headerValue) {
+                case "1":
                     response.setStatus(HttpServletResponse.SC_FORBIDDEN);
                     r.setHandled(true);
                     return;
+                case "2":
+                    r.getHttpChannel().getConnection().close();
+                    r.setHandled(true);
+                    return;
                 }
             }
             super.handle(s, r, request, response);