Flush pending writes before closing the connection on caught exceptions

Motivation:

When an exception is caught on a connection, VertxHandler#exceptionCaught
dispatches the throwable to the user handlers and then closes the channel
with chctx.close(). A response written from the handler -- for instance a
400 written from request().exceptionHandler() when HttpContentDecompressor
throws a DecoderException on a malformed compressed body -- is enqueued in
the outbound message queue but not flushed, because VertxConnection#write
skips the flush while a read is in progress. chctx.close() then causes Netty
to discard the unflushed entries from the ChannelOutboundBuffer, so the
response never reaches the client: the client observes a connection reset
instead of the intended HTTP response.

Changes:

Route the exception-driven close through the channel so it traverses the
pipeline tail and invokes VertxHandler#close -> VertxConnection#writeClose,
which flushes pending writes (an empty buffer with forceFlush=true) and
closes the channel only once the flush has completed. This is the same
flush-then-close path already used by graceful shutdown, idle close and the
regular close(); the exception path simply was not using it. The bare
chctx.close() is kept as a fallback for the case where no connection has been
set yet (handlerAdded has not run), since VertxHandler#close dereferences it.

Add a regression test, Http1xTest#testRequestDecompressionInvalidBodyDeliversErrorResponse,
that sends a malformed gzip body to a server with decompression enabled and
asserts the client receives the response written from the request exception
handler.

HTTP/2 is unaffected: its codec replaces VertxHandler in the pipeline, so
connection-level exceptions do not go through this path.

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

Author: asalajan

vertx-core/src/main/java/io/vertx/core/net/impl/VertxHandler.java

@@ -151,7 +151,15 @@ public void exceptionCaught(ChannelHandlerContext chctx, final Throwable t) {
       close = true;
     }
     if (close) {
-      chctx.close();
+      if (connection != null) {
+        // Route the close through the pipeline tail so VertxHandler#close -> VertxConnection#writeClose
+        // flushes any pending writes (e.g. a response written from the user's exception handler) before
+        // closing the channel. A bare chctx.close() would let Netty's ChannelOutboundBuffer discard the
+        // unflushed entries.
+        chctx.channel().close();
+      } else {
+        chctx.close();
+      }
     }
   }
 

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

@@ -2875,6 +2875,50 @@ 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);