From 49fe31e207a7fecb770fc7955e5764d8629e4315 Mon Sep 17 00:00:00 2001 From: Alexandru Salajan Date: Mon, 8 Jun 2026 14:18:15 +0300 Subject: [PATCH 1/6] 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 --- .../io/vertx/core/net/impl/VertxHandler.java | 10 ++++- .../java/io/vertx/tests/http/Http1xTest.java | 44 +++++++++++++++++++ 2 files changed, 53 insertions(+), 1 deletion(-) diff --git a/vertx-core/src/main/java/io/vertx/core/net/impl/VertxHandler.java b/vertx-core/src/main/java/io/vertx/core/net/impl/VertxHandler.java index d5c2cdeb644..a431f44e15e 100644 --- a/vertx-core/src/main/java/io/vertx/core/net/impl/VertxHandler.java +++ b/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(); + } } } diff --git a/vertx-core/src/test/java/io/vertx/tests/http/Http1xTest.java b/vertx-core/src/test/java/io/vertx/tests/http/Http1xTest.java index 03d5f135c11..f370b655f0d 100644 --- a/vertx-core/src/test/java/io/vertx/tests/http/Http1xTest.java +++ b/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 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); From 4c5a389422fbbcf914105aecb3c63de535ef1cd9 Mon Sep 17 00:00:00 2001 From: Alexandru Salajan Date: Wed, 10 Jun 2026 17:50:08 +0300 Subject: [PATCH 2/6] Add a VertxConnectionTest reproducing the flush before close on exception Covers the fix at the connection layer: a write queued (unflushed) when a caught exception triggers the close must reach the channel before it closes, otherwise Netty's ChannelOutboundBuffer discards the unflushed entry. The test fails without the VertxHandler.exceptionCaught change and passes with it. Signed-off-by: Alexandru Salajan --- .../vertx/tests/net/VertxConnectionTest.java | 23 +++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/vertx-core/src/test/java/io/vertx/tests/net/VertxConnectionTest.java b/vertx-core/src/test/java/io/vertx/tests/net/VertxConnectionTest.java index 0a0372192e7..83066484cdb 100644 --- a/vertx-core/src/test/java/io/vertx/tests/net/VertxConnectionTest.java +++ b/vertx-core/src/test/java/io/vertx/tests/net/VertxConnectionTest.java @@ -922,4 +922,27 @@ public void testResumeWhenReadInProgress() { outbound = ch.readOutbound(); assertSame(expected, outbound); } + + @Test + public void testFlushPendingWriteOnExceptionClose() { + EmbeddedChannel ch = new EmbeddedChannel(); + ChannelPipeline pipeline = ch.pipeline(); + pipeline.addLast(VertxHandler.create(chctx -> new TestConnection(chctx))); + TestConnection connection = (TestConnection) pipeline.get(VertxHandler.class).getConnection(); + connection.exceptionHandler(err -> {}); + // Queue a write without flushing it: this is the state produced when a response is written + // while a read is in progress (e.g. an inbound codec failing mid-decode, which is how a + // request-body decompression failure surfaces). The message sits in the channel outbound + // buffer as an unflushed entry. + Object response = new Object(); + connection.unsafeWrite(response, false); + assertNull(ch.readOutbound()); + // The failure surfaces as a caught exception that asks the connection to close. The close + // must flush the pending write before closing the channel, otherwise Netty's + // ChannelOutboundBuffer discards the unflushed entry and the peer never sees the response. + pipeline.fireExceptionCaught(new RuntimeException()); + ch.runPendingTasks(); + assertSame(response, ch.readOutbound()); + assertFalse(ch.isOpen()); + } } From 8b68f18c1499c5328e39415696eeb10407910d04 Mon Sep 17 00:00:00 2001 From: Alexandru Salajan Date: Wed, 24 Jun 2026 11:05:55 +0300 Subject: [PATCH 3/6] 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 --- .../java/io/vertx/tests/http/Http1xTest.java | 44 ------------------- .../test/java/io/vertx/tests/net/NetTest.java | 27 ++++++++++++ 2 files changed, 27 insertions(+), 44 deletions(-) diff --git a/vertx-core/src/test/java/io/vertx/tests/http/Http1xTest.java b/vertx-core/src/test/java/io/vertx/tests/http/Http1xTest.java index f370b655f0d..03d5f135c11 100644 --- a/vertx-core/src/test/java/io/vertx/tests/http/Http1xTest.java +++ b/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 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); diff --git a/vertx-core/src/test/java/io/vertx/tests/net/NetTest.java b/vertx-core/src/test/java/io/vertx/tests/net/NetTest.java index 04262de4635..f8ce61c28b7 100755 --- a/vertx-core/src/test/java/io/vertx/tests/net/NetTest.java +++ b/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 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)); + } } From c992749629ac8b5970c55a82b7c74b92f1ce1fde Mon Sep 17 00:00:00 2001 From: Alexandru Salajan Date: Wed, 24 Jun 2026 19:04:21 +0300 Subject: [PATCH 4/6] Always close through the channel in VertxHandler.exceptionCaught A regular connection close already closes at the channel level, so the exception-driven close should too; there is no good reason to close from the channel handler context. Drop the connection != null / chctx.close() fallback and always use chctx.channel().close(). Signed-off-by: Alexandru Salajan --- .../java/io/vertx/core/net/impl/VertxHandler.java | 13 ++++--------- 1 file changed, 4 insertions(+), 9 deletions(-) diff --git a/vertx-core/src/main/java/io/vertx/core/net/impl/VertxHandler.java b/vertx-core/src/main/java/io/vertx/core/net/impl/VertxHandler.java index a431f44e15e..468468be010 100644 --- a/vertx-core/src/main/java/io/vertx/core/net/impl/VertxHandler.java +++ b/vertx-core/src/main/java/io/vertx/core/net/impl/VertxHandler.java @@ -151,15 +151,10 @@ public void exceptionCaught(ChannelHandlerContext chctx, final Throwable t) { close = true; } if (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(); - } + // Close through the channel so it traverses the pipeline tail into VertxHandler#close -> + // VertxConnection#writeClose, flushing pending writes before the channel closes -- the same + // path used by a regular connection close. + chctx.channel().close(); } } From 9d2f09a47c3e51a06647f02aa3e23c8d4cdf731b Mon Sep 17 00:00:00 2001 From: Alexandru Salajan Date: Thu, 25 Jun 2026 10:43:08 +0300 Subject: [PATCH 5/6] Remove comment Signed-off-by: Alexandru Salajan --- .../src/main/java/io/vertx/core/net/impl/VertxHandler.java | 3 --- 1 file changed, 3 deletions(-) diff --git a/vertx-core/src/main/java/io/vertx/core/net/impl/VertxHandler.java b/vertx-core/src/main/java/io/vertx/core/net/impl/VertxHandler.java index 468468be010..3e614fea7e8 100644 --- a/vertx-core/src/main/java/io/vertx/core/net/impl/VertxHandler.java +++ b/vertx-core/src/main/java/io/vertx/core/net/impl/VertxHandler.java @@ -151,9 +151,6 @@ public void exceptionCaught(ChannelHandlerContext chctx, final Throwable t) { close = true; } if (close) { - // Close through the channel so it traverses the pipeline tail into VertxHandler#close -> - // VertxConnection#writeClose, flushing pending writes before the channel closes -- the same - // path used by a regular connection close. chctx.channel().close(); } } From 1f474fe505e46e9d47bd7df9a16e276427d86acf Mon Sep 17 00:00:00 2001 From: Alexandru Salajan Date: Tue, 30 Jun 2026 17:37:12 +0300 Subject: [PATCH 6/6] Skip the WebSocket closing handshake timeout on exception When an exception aborts the connection there is no point waiting for the peer's close frame, so set the closing timeout to 0 in handleException. Signed-off-by: Alexandru Salajan --- .../core/http/impl/websocket/WebSocketConnectionImpl.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/vertx-core/src/main/java/io/vertx/core/http/impl/websocket/WebSocketConnectionImpl.java b/vertx-core/src/main/java/io/vertx/core/http/impl/websocket/WebSocketConnectionImpl.java index 903e0804df5..6b99b8793e6 100644 --- a/vertx-core/src/main/java/io/vertx/core/http/impl/websocket/WebSocketConnectionImpl.java +++ b/vertx-core/src/main/java/io/vertx/core/http/impl/websocket/WebSocketConnectionImpl.java @@ -36,7 +36,7 @@ */ public final class WebSocketConnectionImpl extends VertxConnection { - private final long closingTimeoutMS; + private long closingTimeoutMS; private ScheduledFuture closingTimeout; private final boolean server; private final WebSocketMetrics webSocketMetrics; @@ -137,6 +137,7 @@ private CloseWebSocketFrame closeFrame(short statusCode, String reason) { @Override public boolean handleException(Throwable t) { + closingTimeoutMS = 0L; WebSocketImplBase ws = webSocket; if (ws != null) { ws.context().execute(t, ws::handleException);