Close the backend request when the client leaves before the WebSocket upgrade completes

src/main/java/io/vertx/httpproxy/impl/ReverseProxy.java

@@ -176,6 +176,15 @@ public Future<ProxyResponse> sendRequest() {
         if (isWebSocket) {
           HttpServerRequest proxiedRequest = request().proxiedRequest();
           return resolveOrigin(this).compose(request -> {
+            // The client and backend sockets are only joined once the backend answers 101: a client that goes away
+            // before that would leave the request to the backend open. The handler is set before the check, so a client
+            // closing in between is caught by one or the other.
+            HttpServerResponse clientResponse = proxiedRequest.response();
+            clientResponse.exceptionHandler(throwable -> request.reset(0L, throwable));
+            if (clientResponse.closed()) {
+              request.reset();
+              return Future.failedFuture(new HttpClosedException("Client connection closed"));
+            }
             request.setMethod(request().getMethod());
             request.setURI(request().getURI());
             for (Map.Entry<String, String> header : request().headers()) {
@@ -232,9 +241,9 @@ public Future<Void> sendResponse() {
                 responseSocket.closeHandler(v -> proxyResponseSocket.close());
                 proxyResponseSocket.closeHandler(v -> responseSocket.close());
               } else {
-                // Find reproducer
-                System.err.println("Handle this case");
-                ar3.cause().printStackTrace();
+                // The client went away while the backend answered 101
+                log.trace("Could not upgrade the client connection", ar3.cause());
+                proxiedResponse.netSocket().close();
               }
             });
           } else {

src/test/java/io/vertx/tests/WebSocketTest.java

@@ -11,6 +11,7 @@
 package io.vertx.tests;
 
 import io.vertx.core.Future;
+import io.vertx.core.Promise;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.http.*;
 import io.vertx.core.net.SocketAddress;
@@ -260,4 +261,49 @@ public void testWebSocketExtensionsNegotiatedBetweenClientAndBackend(TestContext
       ws.writeTextMessage("hello");
     }));
   }
+
+  private static final Buffer UPGRADE = Buffer.buffer("GET /ws HTTP/1.1\r\n" +
+    "Host: localhost:8080\r\n" +
+    "Connection: Upgrade\r\n" +
+    "Upgrade: websocket\r\n" +
+    "Sec-WebSocket-Version: 13\r\n" +
+    "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n" +
+    "\r\n");
+
+  @Test
+  public void testClientCloseDuringUpgradeClosesBackendConnection(TestContext ctx) {
+    Async backendClosed = ctx.async();
+    Promise<Void> upgradeReceived = Promise.promise();
+    SocketAddress backend = startNetBackend(ctx, 8081, so -> {
+      // Never answers the upgrade
+      so.handler(buff -> upgradeReceived.tryComplete());
+      so.closeHandler(v -> backendClosed.complete());
+    });
+    startProxy(backend);
+    vertx.createNetClient().connect(8080, "localhost").onComplete(ctx.asyncAssertSuccess(so -> {
+      upgradeReceived.future().onComplete(ctx.asyncAssertSuccess(v -> so.close()));
+      so.write(UPGRADE);
+    }));
+  }
+
+  @Test
+  public void testClientCloseBeforeOriginResolvedSendsNoUpgrade(TestContext ctx) {
+    Async backendClosed = ctx.async();
+    Promise<Void> resolving = Promise.promise();
+    SocketAddress backend = startNetBackend(ctx, 8081, so -> {
+      so.handler(buff -> ctx.fail("Upgrade sent for a client that has gone"));
+      so.closeHandler(v -> backendClosed.complete());
+    });
+    startProxy(proxy -> proxy.origin(OriginRequestProvider.selector(proxyContext -> {
+      Promise<SocketAddress> origin = Promise.promise();
+      // Resolved only once the client has gone
+      proxyContext.request().proxiedRequest().connection().closeHandler(v -> origin.complete(backend));
+      resolving.complete();
+      return origin.future();
+    })));
+    vertx.createNetClient().connect(8080, "localhost").onComplete(ctx.asyncAssertSuccess(so -> {
+      resolving.future().onComplete(ctx.asyncAssertSuccess(v -> so.close()));
+      so.write(UPGRADE);
+    }));
+  }
 }