Move the flush-before-close-on-exception regression test to NetTest

The fix is at the transport layer (VertxHandler/VertxConnection) and is not HTTP specific, so exercise it over plain TCP rather than HTTP. Drop Http1xTest#testRequestDecompressionInvalidBodyDeliversErrorResponse and add NetTest#testFlushPendingWriteOnExceptionClose, which parks an unflushed write on a NetServer connection, triggers a caught exception that closes it, and asserts the client still receives the bytes before the close.

Signed-off-by: Alexandru Salajan <alexandru17@gmail.com>

Author: asalajan

vertx-core/src/test/java/io/vertx/tests/http/Http1xTest.java

@@ -2875,50 +2875,6 @@ public void testContentDecompression() throws Exception {
       .await();
   }
 
-  @Test
-  // Regression test: when an inbound request body fails to decompress, the application's
-  // exception handler must still be able to deliver an error response to the client. Before
-  // the VertxHandler#exceptionCaught flush-before-close fix, the response written from the
-  // exception handler was discarded and the client only saw a connection reset.
-  public void testRequestDecompressionInvalidBodyDeliversErrorResponse() throws Exception {
-    server = vertx.createHttpServer(new HttpServerOptions().setDecompressionSupported(true));
-    // A valid gzip header (the first 10 bytes) followed by a corrupted deflate payload, so that
-    // Netty's HttpContentDecompressor accepts the stream as gzip and then fails while decoding it.
-    byte[] corrupted = TestUtils.compressGzip(TestUtils.randomAlphaString(256));
-    for (int i = 10; i < corrupted.length; i++) {
-      corrupted[i] = (byte) ~corrupted[i];
-    }
-    server.requestHandler(req -> {
-      req.exceptionHandler(err -> {
-        if (!req.response().ended()) {
-          req.response().setStatusCode(400).end("decode-failed");
-        }
-      });
-      // Should not be reached for a malformed body.
-      req.bodyHandler(body -> req.response().end());
-    });
-    startServer(testAddress);
-    NetClient netClient = vertx.createNetClient();
-    CompletableFuture<String> result = new CompletableFuture<>();
-    netClient.connect(testAddress).onComplete(TestUtils.onSuccess(so -> {
-      Buffer respBuff = Buffer.buffer();
-      so.handler(respBuff::appendBuffer);
-      so.closeHandler(v -> result.complete(respBuff.toString()));
-      so.exceptionHandler(result::completeExceptionally);
-      Buffer request = Buffer.buffer()
-        .appendString("PUT /somepath HTTP/1.1\r\n")
-        .appendString("Host: localhost\r\n")
-        .appendString("Content-Encoding: gzip\r\n")
-        .appendString("Content-Length: " + corrupted.length + "\r\n")
-        .appendString("\r\n")
-        .appendBytes(corrupted);
-      so.write(request);
-    }));
-    String resp = result.get(20, TimeUnit.SECONDS);
-    assertTrue("Expected the response to start with \"HTTP/1.1 400\" but got: [" + resp + "]", resp.startsWith("HTTP/1.1 400"));
-    assertTrue("Expected the error body in the response but got: [" + resp + "]", resp.contains("decode-failed"));
-  }
-
   @Test
   public void testResetClientRequestNotYetSent(Checkpoint checkpoint) throws Exception {
     testResetClientRequestNotYetSent(checkpoint, false, false);

vertx-core/src/test/java/io/vertx/tests/net/NetTest.java

@@ -40,6 +40,7 @@
 import io.vertx.core.json.JsonObject;
 import io.vertx.core.net.*;
 import io.vertx.core.net.impl.HAProxyMessageCompletionHandler;
+import io.vertx.core.net.impl.VertxConnection;
 import io.vertx.core.net.impl.VertxHandler;
 import io.vertx.core.net.impl.tcp.CleanableNetClient;
 import io.vertx.core.internal.net.NetServerInternal;
@@ -4641,4 +4642,30 @@ public void testCloseServerUnwritableSocket(Checkpoint checkpoint1, Checkpoint c
     TestUtils.assertWaitUntil(closing::get);
     TestUtils.assertWaitUntil(closed::get);
   }
+
+  @Test
+  public void testFlushPendingWriteOnExceptionClose() throws Exception {
+    server.connectHandler(so -> {
+      so.exceptionHandler(err -> {});
+      so.handler(data -> {
+        VertxConnection conn = (VertxConnection) so;
+        // A response queued on the connection but not yet flushed (the state produced when a
+        // response is written while a read is in progress), then a pipeline failure that tears
+        // the connection down.
+        conn.unsafeWrite(Unpooled.copiedBuffer("pending-response", StandardCharsets.UTF_8), false);
+        conn.channelHandlerContext().pipeline().fireExceptionCaught(new RuntimeException("boom"));
+      });
+    });
+    startServer();
+    Buffer received = Buffer.buffer();
+    CompletableFuture<String> result = new CompletableFuture<>();
+    NetSocket socket = client.connect(testAddress).await();
+    socket.handler(received::appendBuffer);
+    socket.closeHandler(v -> result.complete(received.toString()));
+    socket.exceptionHandler(result::completeExceptionally);
+    socket.write("ping").await();
+    // With the fix the pending response is flushed before the connection closes; without it the
+    // client only sees the close (Netty discards the unflushed entry).
+    assertEquals("pending-response", result.get(20, TimeUnit.SECONDS));
+  }
 }