Repository: eclipse-vertx/vert.x

Problem statement:
HTTP server connection keeps processing pending pipelined requests when connection closed
Currently the HTTP server keeps processing pipelined HTTP messages when the connection is closed during the processing of a request and there are pending HTTP request in the server pipeline (i.e the response has not been sent). This is due to a bug that when a message has not been processed during a check, the connection redo the check and it must not do it.

Changes: we only trigger an message processing check from the message check itself when a message has been effectively processed.


Hints:

Patch:
diff --git a/src/main/java/io/vertx/core/http/impl/Http1xServerConnection.java b/src/main/java/io/vertx/core/http/impl/Http1xServerConnection.java
--- a/src/main/java/io/vertx/core/http/impl/Http1xServerConnection.java
+++ b/src/main/java/io/vertx/core/http/impl/Http1xServerConnection.java
@@ -81,7 +81,6 @@ public class Http1xServerConnection extends Http1xConnectionBase implements Http
 
   private static final int CHANNEL_PAUSE_QUEUE_SIZE = 5;
 
-  private final Deque<Object> pending = new ArrayDeque<>(8);
   private final String serverOrigin;
   private final SSLHelper sslHelper;
   final HttpServerOptions options;
@@ -94,12 +93,15 @@ public class Http1xServerConnection extends Http1xConnectionBase implements Http
   private HttpServerRequestImpl currentRequest;
   private HttpServerResponseImpl pendingResponse;
   private ServerWebSocketImpl ws;
-  private boolean channelPaused;
-  private boolean paused;
-  private boolean sentCheck;
   private long bytesRead;
   private long bytesWritten;
+
+  // (pending == true) <=> (paused && pending.size() > 0)
+  private final Deque<Object> pending = new ArrayDeque<>(8);
+  private boolean paused;
+  private boolean sentCheck;
   private boolean queueing;
+  private boolean channelPaused;
 
   public Http1xServerConnection(VertxInternal vertx,
                                 SSLHelper sslHelper,
@@ -130,30 +132,40 @@ synchronized void pause() {
   synchronized void resume() {
     if (paused) {
       paused = false;
-      checkNextTick();
+      if (pending.isEmpty()) {
+        unsetQueueing();
+      } else {
+        checkNextTick();
+      }
     }
   }
 
   synchronized void handleMessage(Object msg) {
-    if (queueing) {
+    if (queueing || !processMessage(msg)) {
       enqueue(msg);
-    } else {
-      if (processMessage(msg)) {
-        checkNextTick();
-      } else {
-        enqueue(msg);
-      }
+    }
+  }
+
+  /**
+   * Set in non queueing mode, i.e (pending.size() == 0 && !paused)
+   */
+  private void unsetQueueing() {
+    queueing = false;
+    if (channelPaused) {
+      // Resume the actual channel
+      channelPaused = false;
+      doResume();
     }
   }
 
   private void enqueue(Object msg) {
-    //We queue requests if paused or a request is in progress to prevent responses being written in the wrong order
+    // We queue requests if paused or a request is in progress to prevent responses being written in the wrong order
     queueing = true;
     pending.add(msg);
     if (pending.size() == CHANNEL_PAUSE_QUEUE_SIZE) {
-      //We pause the channel too, to prevent the queue growing too large, but we don't do this
-      //until the queue reaches a certain size, to avoid pausing it too often
-      super.doPause();
+      // We pause the channel too, to prevent the queue growing too large, but we don't do this
+      // until the queue reaches a certain size, to avoid pausing it too often
+      doPause();
       channelPaused = true;
     }
   }
@@ -475,30 +487,23 @@ private void handleOther(Object msg) {
 
   private void checkNextTick() {
     // Check if there are more pending messages in the queue that can be processed next time around
-    if (!paused && !sentCheck) {
+    if (!paused && queueing && !sentCheck) {
       sentCheck = true;
-      vertx.runOnContext(v -> {
+      context.runOnContext(v -> {
         synchronized (Http1xServerConnection.this) {
           sentCheck = false;
           if (!paused) {
+            // The only place we poll the pending queue, so we are sure that pending.size() > 0
+            // since we got there because queueing was true
             Object msg = pending.poll();
-            if (msg != null) {
-              if (processMessage(msg)) {
-                checkNextTick();
+            if (processMessage(msg)) {
+              if (pending.isEmpty()) {
+                unsetQueueing();
               } else {
-                pending.addFirst(msg);
-              }
-            }
-            if (pending.isEmpty()) {
-              queueing = false;
-              if (channelPaused) {
-                // Resume the actual channel
-                channelPaused = false;
-                Http1xServerConnection.super.doResume();
+                checkNextTick();
               }
             } else {
-              queueing = true;
-              checkNextTick();
+              pending.addFirst(msg);
             }
           }
         }


Test patch:
diff --git a/src/test/java/io/vertx/test/core/Http1xTest.java b/src/test/java/io/vertx/test/core/Http1xTest.java
--- a/src/test/java/io/vertx/test/core/Http1xTest.java
+++ b/src/test/java/io/vertx/test/core/Http1xTest.java
@@ -1233,6 +1233,33 @@ public void testPipeliningLimit() throws Exception {
     await();
   }
 
+  @Repeat(times = 10)
+  @Test
+  public void testCloseServerConnectionWithPendingMessages() throws Exception {
+    int n = 5;
+    server.requestHandler(req -> {
+      vertx.setTimer(100, id -> {
+        req.response().close();
+      });
+    });
+    startServer();
+    client.close();
+    client = vertx.createHttpClient(new HttpClientOptions().setMaxPoolSize(n).setPipelining(true));
+    AtomicBoolean completed = new AtomicBoolean();
+    for (int i = 0;i < n * 2;i++) {
+      client.get(DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST, DEFAULT_TEST_URI, resp -> {
+        fail();
+      }).connectionHandler(conn -> {
+        conn.closeHandler(v -> {
+          if (completed.compareAndSet(false, true)) {
+            testComplete();
+          }
+        });
+      }).end();
+    }
+    await();
+  }
+
   @Test
   public void testPipeliningFailure() throws Exception {
     int n = 5;