fix: avoid NPE resetting metrics for a not-yet-begun pipelined HTTP/1 request on connection close

vertx-core/src/main/java/io/vertx/core/http/impl/http1/Http1ServerConnection.java

@@ -517,8 +517,11 @@ protected void handleClosed() {
     if (requestInProgress != null && requestInProgress != responseInProgress && !requestInProgress.isEnded()) {
       // The request body was not fully received before the connection closed,
       // report a request reset so metrics like active requests are not leaked
-      if (httpMetrics != null) {
-        httpMetrics.requestReset(requestInProgress.metric());
+      Object metric = requestInProgress.metric();
+      // metric is null when the request is a pipelined request still waiting for the
+      // response in progress to complete: it was never begun and has nothing to reset
+      if (httpMetrics != null && metric != null) {
+        httpMetrics.requestReset(metric);
       }
     }
     super.handleClosed();

vertx-core/src/test/java/io/vertx/tests/metrics/Http1xMetricsLeakTest.java

@@ -19,7 +19,14 @@
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.io.BufferedReader;
+import java.io.ByteArrayOutputStream;
+import java.io.InputStreamReader;
+import java.io.PrintStream;
 import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
 /**
@@ -59,14 +66,61 @@ public void testActiveRequestResetWhenConnectionClosesBeforeRequestBody() throws
         "\r\n"
       ).getBytes());
       socket.getOutputStream().flush();
-      // Read the early 401 response then close without sending the declared body byte
-      while (socket.getInputStream().read() >= 0) {
-        // consume
+      // Read the early 401 response headers (there is no response body to read) then
+      // close the socket without sending the declared request body byte, instead of
+      // waiting for a server-initiated EOF that never comes for a keep-alive connection
+      BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream(), StandardCharsets.US_ASCII));
+      String statusLine = reader.readLine();
+      Assert.assertNotNull(statusLine);
+      Assert.assertTrue(statusLine, statusLine.contains("401"));
+      String line;
+      while ((line = reader.readLine()) != null && !line.isEmpty()) {
+        // consume headers
       }
     }
 
     // The request body was never completed, so requestReset must have been
     // reported and the metric must be marked as failed (i.e. removed/reset)
     assertWaitUntil(() -> metricRef.get() != null && metricRef.get().failed.get());
   }
-}
+
+  @Test
+  public void testPipelinedRequestNotYetBegunOnClose() throws Exception {
+    // A pipelined request whose headers were parsed while the previous response was still
+    // in progress never gets its metric started (it is deferred until it becomes the
+    // response in progress). Closing the connection at that point must not attempt to
+    // reset a metric that was never begun.
+    CountDownLatch firstRequestReceived = new CountDownLatch(1);
+    server.requestHandler(req -> firstRequestReceived.countDown());
+    startServer(testAddress);
+
+    ByteArrayOutputStream capturedErr = new ByteArrayOutputStream();
+    PrintStream originalErr = System.err;
+    System.setErr(new PrintStream(capturedErr, true, StandardCharsets.UTF_8));
+    try (Socket socket = new Socket("127.0.0.1", testAddress.port())) {
+      socket.setSoTimeout(2000);
+      // First request is left unanswered by the handler, keeping its response in progress.
+      // Second (pipelined) request is only sent as headers, so it never begins.
+      socket.getOutputStream().write((
+        "GET / HTTP/1.1\r\n" +
+        "Host: 127.0.0.1\r\n" +
+        "Connection: keep-alive\r\n" +
+        "\r\n" +
+        "POST /pipelined HTTP/1.1\r\n" +
+        "Host: 127.0.0.1\r\n" +
+        "Content-Length: 5\r\n" +
+        "Connection: keep-alive\r\n" +
+        "\r\n"
+      ).getBytes());
+      socket.getOutputStream().flush();
+      Assert.assertTrue(firstRequestReceived.await(2, TimeUnit.SECONDS));
+      // Give the server time to parse the pipelined request headers before closing
+      Thread.sleep(300);
+    } finally {
+      Thread.sleep(300);
+      System.setErr(originalErr);
+    }
+
+    Assert.assertFalse(capturedErr.toString(StandardCharsets.UTF_8).contains("NullPointerException"));
+  }
+}