Server WebSocket: a frame pipelined with the upgrade request is dropped and its buffer leaks (regression from 4.5) #1

Activity

1. Sanne-ai-bot commented on Oct 30, 2026

   

   Sanne-ai-bot

   on Oct 30, 2026

   Last edited by Sanne-ai-bot

   Author

   More actions

   Review of eclipse-vertx#6476: one more case to handle

   I verified the fix in eclipse-vertx#6476. It works:

   • The three new tests pass, and all three fail on master's code (timeout waiting for the echo).
   • With the reproducers from this issue on a build of the PR, the pipelined frame arrives in every mode, including webSocketHandshakeHandler + accept(). PipelinedFrameLeaks reports 0 leaks, against 27 on 5.2.2.
   • No regressions: Http1xTest, WebSocketTest and the HTTP/1 server suites pass, 800 tests and 0 failures.
   • incus-spawn's MITM proxy test for this case passes against the PR build.

   What is still missing: a frame pipelined with a failed upgrade is held until the connection closes

   The frame does not reach Http1ServerConnection.handleMessage straight away.
   It is decoded while VertxConnection.read is already inside handleMessage for the request's LastHttpContent, so it waits in VertxConnection's pending queue.
   When the 101 write fails synchronously, the handshaker's listener calls webSocketUpgradeFailed() before that frame leaves the queue.
   releasePendingWebSocketFrames() then finds nothing to release, and the frame is buffered after the cleanup.
   It is released only by handleClosed(), so it stays allocated for as long as the connection stays open.

   The success path does not have this problem: once webSocketUpgraded() has run, a late frame is forwarded to the WebSocket handler.

   A test that fails the 101 write shows it.
   On the PR as it stands, the frame's refcount is still 1 after the upgrade has failed, and drops to 0 only after the connection closes.

   The fix

   Remember that the upgrade failed, and release late frames straight away:

   --- a/vertx-core/src/main/java/io/vertx/core/http/impl/http1/Http1ServerConnection.java
   +++ b/vertx-core/src/main/java/io/vertx/core/http/impl/http1/Http1ServerConnection.java
   @@ -98,6 +98,7 @@ public class Http1ServerConnection extends Http1Connection implements HttpServer
      // WebSocket upgrade: frames decoded before the WebSocket handler replaces this connection handler
      private ArrayDeque<WebSocketFrame> pendingWebSocketFrames;
      private VertxHandler<?> webSocketHandler;
   +  private boolean webSocketUpgradeFailed;
    
      public final HttpServerMetrics httpMetrics;
      private final TransportMetrics<?> transportMetrics;
   @@ -266,6 +267,11 @@ public class Http1ServerConnection extends Http1Connection implements HttpServer
       * connection. Such frame is buffered until the upgrade completes and then forwarded to the WebSocket handler.
       */
      private void onWebSocketFrame(WebSocketFrame frame) {
   +    if (webSocketUpgradeFailed) {
   +      // The upgrade failed before this frame left the inbound queue
   +      ReferenceCountUtil.release(frame);
   +      return;
   +    }
        VertxHandler<?> handler = webSocketHandler;
        if (handler == null) {
          // Upgrade in progress
   @@ -317,6 +323,7 @@ public class Http1ServerConnection extends Http1Connection implements HttpServer
       */
      public void webSocketUpgradeFailed() {
        assert chctx.executor().inEventLoop();
   +    webSocketUpgradeFailed = true;
        releasePendingWebSocketFrames();
      }
    

   The test (in WebSocketTest, beside the PR's three) fails the 101 write in an outbound handler at the head of the pipeline.
   The handshaker writes the response from the encoder's context, so the handler matches the encoded bytes.
   It checks the frame's refcount once the current read has been processed:

     @Test
     public void testFramePipelinedWithUpgradeRequestFailedUpgrade() throws Exception {
       server = vertx.createHttpServer(new HttpServerOptions().setPort(DEFAULT_HTTP_PORT));
       server.webSocketHandler(ws -> fail("upgrade should fail"));
       List<io.netty.buffer.ByteBuf> frames = Collections.synchronizedList(new ArrayList<>());
       java.util.concurrent.CompletableFuture<Integer> refCntAfterFailure = new java.util.concurrent.CompletableFuture<>();
       server.connectionHandler(conn -> {
         ChannelHandlerContext chctx = ((Http1ServerConnection) conn).channelHandlerContext();
         chctx.pipeline().addBefore("handler", "tracker", new ChannelInboundHandlerAdapter() {
           @Override
           public void channelRead(ChannelHandlerContext ctx, Object msg) {
             if (msg instanceof io.netty.handler.codec.http.websocketx.WebSocketFrame) {
               frames.add(((io.netty.handler.codec.http.websocketx.WebSocketFrame) msg).content());
             }
             ctx.fireChannelRead(msg);
           }
         });
         // Fail the 101 write
         chctx.pipeline().addFirst("fail101", new ChannelOutboundHandlerAdapter() {
           @Override
           public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
             if (msg instanceof io.netty.buffer.ByteBuf
               && ((io.netty.buffer.ByteBuf) msg).toString(StandardCharsets.ISO_8859_1).startsWith("HTTP/1.1 101")) {
               io.netty.util.ReferenceCountUtil.release(msg);
               promise.setFailure(new java.io.IOException("simulated 101 write failure"));
               // Check the frame once the current read has been processed
               ctx.executor().execute(() -> refCntAfterFailure.complete(frames.isEmpty() ? -1 : frames.get(0).refCnt()));
               return;
             }
             ctx.write(msg, promise);
           }
         });
       });
       server.listen().await();
       byte[] mask = { 1, 2, 3, 4 };
       byte[] payload = "hi".getBytes(StandardCharsets.UTF_8);
       Buffer request = Buffer.buffer("GET / HTTP/1.1\r\nHost: localhost\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n" +
         "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\nSec-WebSocket-Version: 13\r\n\r\n")
         .appendByte((byte) 0x81).appendByte((byte) (0x80 | payload.length)).appendBytes(mask);
       for (int i = 0; i < payload.length; i++) {
         request.appendByte((byte) (payload[i] ^ mask[i % 4]));
       }
       NetSocket so = vertx.createNetClient().connect(DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST).await();
       so.write(request);
       // The frame is released once the upgrade fails, not only when the connection closes
       assertEquals(0, (int) refCntAfterFailure.get(10, TimeUnit.SECONDS));
       assertEquals(1, frames.size());
       so.close();
     }

   Without the flag it fails with expected:<0> but was:<1>.
   With it, the four pipelined-frame tests and the whole WebSocketTest pass (218 tests, 1 skipped as on master).

   Status of the commit

   Pushed as a4c5fa2c8 ("Release WebSocket frames pipelined with a failed upgrade", signed off) to Sanne-ai-bot/vert.x:ws-pipelined-frame-leak, on top of 00e6eff30.
   An earlier attempt failed with GitHub 500 errors on every new git object write; a retry later succeeded unchanged.

   Smaller points a reviewer may raise (optional)

   • onWebSocketFrame calls channelReadComplete once per buffered frame. Calling it once after draining in webSocketUpgraded() would be tidier.
   • Nothing caps the buffer. It only lasts until the 101 is flushed, so it is bounded in practice, but a reviewer may ask.
   • A client cannot legitimately compress a frame before the server has accepted permessage-deflate. Such a frame used to be dropped; it now reaches the application before the inflater is installed. Rejecting it may be preferable. This case is not tested.
2. Sanne-ai-bot

   added a commit that references this issue on Oct 30, 2026

   Deliver and release WebSocket frames pipelined with the server upgrad…

   ...

   c4dcf50