- netty: code cleaup + minor fixes

jknack · jknack · commit 141377b49380 · 2026-03-13T14:00:52.000-03:00
diff --git a/modules/jooby-netty/src/main/java/io/jooby/internal/netty/NettyContext.java b/modules/jooby-netty/src/main/java/io/jooby/internal/netty/NettyContext.java
@@ -422,15 +422,18 @@ public Context upgrade(WebSocket.Initializer handler) {
               ? conf.getBytes("websocket.maxSize").intValue()
               : WebSocket.MAX_BUFFER_SIZE;
       String webSocketURL = getProtocol() + "://" + req.headers().get(HttpHeaderNames.HOST) + path;
+
       var config =
           WebSocketDecoderConfig.newBuilder()
               .allowExtensions(true)
               .allowMaskMismatch(false)
               .withUTF8Validator(false)
               .maxFramePayloadLength(maxSize)
               .build();
+
       webSocket = new NettyWebSocket(this);
       handler.init(Context.readOnly(this), webSocket);
+
       var webSocketRequest =
           new DefaultFullHttpRequest(
               HTTP_1_1,
@@ -439,21 +442,48 @@ public Context upgrade(WebSocket.Initializer handler) {
               Unpooled.EMPTY_BUFFER,
               req.headers(),
               EmptyHttpHeaders.INSTANCE);
+
       var codec = ctx.pipeline().get(NettyServerCodec.class);
       codec.webSocketHandshake(ctx);
+
       WebSocketServerHandshakerFactory factory =
           new WebSocketServerHandshakerFactory(webSocketURL, null, config);
       WebSocketServerHandshaker handshaker = factory.newHandshaker(webSocketRequest);
-      handshaker.handshake(ctx.channel(), webSocketRequest);
-      webSocket.fireConnect();
-      long timeout =
-          conf.hasPath("websocket.idleTimeout")
-              ? conf.getDuration("websocket.idleTimeout", MILLISECONDS)
-              : MINUTES.toMillis(5);
-      if (timeout > 0) {
-        IdleStateHandler idle = new IdleStateHandler(timeout, 0, 0, MILLISECONDS);
-        ctx.pipeline().addBefore("handler", "idle", idle);
+
+      if (handshaker == null) {
+        WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
+        return this;
       }
+
+      handshaker
+          .handshake(ctx.channel(), webSocketRequest)
+          .addListener(
+              future -> {
+                if (future.isSuccess()) {
+                  // Only notify the application once the TCP write completes successfully
+                  webSocket.fireConnect();
+
+                  // FIX 3: Safely apply the WebSocket-specific idle timeout
+                  long timeout =
+                      conf.hasPath("websocket.idleTimeout")
+                          ? conf.getDuration("websocket.idleTimeout", MILLISECONDS)
+                          : MINUTES.toMillis(5);
+
+                  if (timeout > 0) {
+                    IdleStateHandler idle = new IdleStateHandler(timeout, 0, 0, MILLISECONDS);
+                    // If the global server timeout is already there, replace it. Otherwise, add it.
+                    if (ctx.pipeline().get(IdleStateHandler.class) != null) {
+                      ctx.pipeline().replace(IdleStateHandler.class, "idle", idle);
+                    } else {
+                      ctx.pipeline().addBefore("handler", "idle", idle);
+                    }
+                  }
+                } else {
+                  // Handshake failed (e.g., client hung up during negotiation)
+                  ctx.fireExceptionCaught(future.cause());
+                }
+              });
+
     } catch (Throwable x) {
       sendError(x);
     }
diff --git a/modules/jooby-netty/src/main/java/io/jooby/internal/netty/NettyGrpcExchange.java b/modules/jooby-netty/src/main/java/io/jooby/internal/netty/NettyGrpcExchange.java
@@ -5,9 +5,12 @@
  */
 package io.jooby.internal.netty;
 
+import java.net.URLEncoder;
 import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
 
 import io.jooby.rpc.grpc.GrpcExchange;
@@ -17,7 +20,6 @@
 import io.netty.handler.codec.http.DefaultHttpContent;
 import io.netty.handler.codec.http.DefaultHttpResponse;
 import io.netty.handler.codec.http.DefaultLastHttpContent;
-import io.netty.handler.codec.http.HttpContent;
 import io.netty.handler.codec.http.HttpHeaderNames;
 import io.netty.handler.codec.http.HttpRequest;
 import io.netty.handler.codec.http.HttpResponse;
@@ -29,7 +31,8 @@ public class NettyGrpcExchange implements GrpcExchange {
 
   private final ChannelHandlerContext ctx;
   private final HttpRequest request;
-  private boolean headersSent = false;
+
+  private final AtomicBoolean headersSent = new AtomicBoolean(false);
 
   public NettyGrpcExchange(ChannelHandlerContext ctx, HttpRequest request) {
     this.ctx = ctx;
@@ -58,12 +61,11 @@ public Map<String, String> getHeaders() {
   }
 
   private void sendHeadersIfNecessary() {
-    if (!headersSent) {
+    if (headersSent.compareAndSet(false, true)) {
       // Send the initial HTTP/2 HEADERS frame (Status 200)
       HttpResponse response = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
       response.headers().set(HttpHeaderNames.CONTENT_TYPE, "application/grpc");
       ctx.write(response);
-      headersSent = true;
     }
   }
 
@@ -72,7 +74,7 @@ public void send(ByteBuffer payload, Consumer<Throwable> callback) {
     sendHeadersIfNecessary();
 
     // Wrap the NIO ByteBuffer in a Netty ByteBuf without copying
-    HttpContent chunk = new DefaultHttpContent(Unpooled.wrappedBuffer(payload));
+    var chunk = new DefaultHttpContent(Unpooled.wrappedBuffer(payload));
 
     // Write and flush, then map Netty's Future to your single-lambda callback
     ctx.writeAndFlush(chunk)
@@ -88,28 +90,41 @@ public void send(ByteBuffer payload, Consumer<Throwable> callback) {
 
   @Override
   public void close(int statusCode, String description) {
-    if (headersSent) {
-      // Trailers-Appended: Send the final HTTP/2 HEADERS frame with END_STREAM flag
+    var encodedDescription = encodeGrpcMessage(description);
+
+    // If headers were already sent, we just need to send the final trailers
+    if (headersSent.get()) {
       LastHttpContent lastContent = new DefaultLastHttpContent();
       lastContent.trailingHeaders().set("grpc-status", String.valueOf(statusCode));
-      if (description != null) {
-        lastContent.trailingHeaders().set("grpc-message", description);
+
+      if (encodedDescription != null) {
+        lastContent.trailingHeaders().set("grpc-message", encodedDescription);
       }
-      // writeAndFlush the LastHttpContent, then close the Netty stream channel
+
       ctx.writeAndFlush(lastContent).addListener(ChannelFutureListener.CLOSE);
+
     } else {
-      // Trailers-Only: No body was sent, so standard headers become the trailers
-      HttpResponse response = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
+      // Trailers-Only fast path: Headers and trailers combined
+      var response = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
       response.headers().set(HttpHeaderNames.CONTENT_TYPE, "application/grpc");
       response.headers().set("grpc-status", String.valueOf(statusCode));
-      if (description != null) {
-        response.headers().set("grpc-message", description);
+
+      if (encodedDescription != null) {
+        response.headers().set("grpc-message", encodedDescription);
       }
-      ctx.write(response);
 
-      // Close out the stream with an empty DATA frame possessing the END_STREAM flag
+      ctx.write(response);
       ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT)
           .addListener(ChannelFutureListener.CLOSE);
     }
   }
+
+  /** gRPC specification requires the grpc-message trailer to be percent-encoded. */
+  private static String encodeGrpcMessage(String description) {
+    if (description == null) {
+      return null;
+    }
+    // URLEncoder uses '+' for spaces, but gRPC strictly expects '%20'
+    return URLEncoder.encode(description, StandardCharsets.UTF_8).replace("+", "%20");
+  }
 }
diff --git a/modules/jooby-netty/src/main/java/io/jooby/internal/netty/NettyGrpcHandler.java b/modules/jooby-netty/src/main/java/io/jooby/internal/netty/NettyGrpcHandler.java
@@ -5,6 +5,9 @@
  */
 package io.jooby.internal.netty;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import io.jooby.rpc.grpc.GrpcProcessor;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
@@ -14,6 +17,8 @@
 
 public class NettyGrpcHandler extends ChannelInboundHandlerAdapter {
 
+  private static final Logger log = LoggerFactory.getLogger(NettyGrpcHandler.class);
+
   private final GrpcProcessor processor;
   private final boolean isHttp2;
 
@@ -39,7 +44,6 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
       if (processor.isGrpcMethod(path)
           && contentType != null
           && contentType.startsWith("application/grpc")) {
-        isGrpc = true;
 
         if (!isHttp2) {
           // gRPC requires HTTP/2. Reject HTTP/1.1 calls immediately.
@@ -53,6 +57,9 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
           return;
         }
 
+        // This prevents leaking state on rejected HTTP/1.1 connections.
+        isGrpc = true;
+
         // We will implement NettyGrpcExchange in the next step
         var exchange = new NettyGrpcExchange(ctx, req);
         var subscriber = processor.process(exchange);
@@ -92,6 +99,7 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
 
   @Override
   public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+    log.debug("gRPC stream exception caught", cause);
     if (isGrpc) {
       ctx.close();
     } else {
diff --git a/modules/jooby-netty/src/main/java/io/jooby/internal/netty/NettyGrpcInputBridge.java b/modules/jooby-netty/src/main/java/io/jooby/internal/netty/NettyGrpcInputBridge.java
@@ -50,20 +50,27 @@ public void cancel() {
     ctx.close(); // Abort the connection
   }
 
-  /** Called by the NettyGrpcHandler when a new chunk arrives from the network. */
   public void onChunk(HttpContent chunk) {
     try {
       ByteBuf content = chunk.content();
       if (content.isReadable()) {
-        // Convert Netty ByteBuf to standard Java ByteBuffer
-        ByteBuffer buffer = content.nioBuffer();
 
-        // Pass to the gRPC deframer
-        subscriber.onNext(buffer);
+        // Copy the bytes. content.nioBuffer() shares memory with the ByteBuf,
+        // which gets released in the finally block of NettyGrpcHandler.
+        // We must copy it to an unmanaged heap buffer to prevent use-after-free corruption.
+        byte[] bytes = new byte[content.readableBytes()];
+        content.readBytes(bytes);
+        ByteBuffer buffer = ByteBuffer.wrap(bytes);
 
+        // If onNext synchronously calls request(1), that request() will see demand transition
+        // from 0 to 1 and safely trigger ctx.read().
         long currentDemand = demand.decrementAndGet();
+
+        // Pass the isolated buffer to the gRPC deframer
+        subscriber.onNext(buffer);
+
         if (currentDemand > 0) {
-          // Still have demand, ask Netty for the next chunk
+          // Still have demand from previous requests, ask Netty for the next chunk
           ctx.read();
         }
       }
diff --git a/modules/jooby-netty/src/main/java/io/jooby/internal/netty/NettyHandler.java b/modules/jooby-netty/src/main/java/io/jooby/internal/netty/NettyHandler.java
@@ -21,6 +21,7 @@
 import io.netty.handler.codec.http.multipart.*;
 import io.netty.handler.codec.http.websocketx.WebSocketFrame;
 import io.netty.handler.timeout.IdleStateEvent;
+import io.netty.util.ReferenceCountUtil;
 
 public class NettyHandler extends ChannelInboundHandlerAdapter {
   private final Logger log = LoggerFactory.getLogger(NettyServer.class);
@@ -88,6 +89,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
           if (req.getClass() == DefaultFullHttpRequest.class) {
             // HTTP2 aggregates all into a full http request.
             if (((DefaultFullHttpRequest) req).content().readableBytes() > maxRequestSize) {
+              release(req);
               router.match(context).execute(context, Route.REQUEST_ENTITY_TOO_LARGE);
               return;
             }
@@ -127,6 +129,8 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
     } else if (isWebSocketFrame(msg)) {
       if (context.webSocket != null) {
         context.webSocket.handleFrame((WebSocketFrame) msg);
+      } else {
+        release(msg);
       }
     }
   }
@@ -189,9 +193,9 @@ public void writeChunks(Object header, Object body, ChannelPromise promise) {
     }
   }
 
-  private void release(HttpContent ref) {
-    if (ref.refCnt() > 0) {
-      ref.release();
+  private void release(Object ref) {
+    if (ReferenceCountUtil.refCnt(ref) > 0) {
+      ReferenceCountUtil.release(ref);
     }
   }
 
diff --git a/modules/jooby-netty/src/main/java/io/jooby/internal/netty/NettyPipeline.java b/modules/jooby-netty/src/main/java/io/jooby/internal/netty/NettyPipeline.java
@@ -112,6 +112,29 @@ private void setupHttp11Upgrade(ChannelPipeline pipeline) {
             protocol -> "h2c".equals(protocol.toString()) ? createH2CUpgradeCodec() : null,
             (int) maxRequestSize));
 
+    pipeline.addLast(
+        "h2upgrade-cleaner",
+        new ChannelInboundHandlerAdapter() {
+          @Override
+          public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+            if (evt instanceof HttpServerUpgradeHandler.UpgradeEvent) {
+              ChannelPipeline p = ctx.pipeline();
+
+              // Strip out HTTP/1.1 handlers that are no longer needed on the parent channel.
+              // The Http2StreamInitializer will attach new ones for the multiplexed child channels.
+              if (p.context("grpc") != null) p.remove("grpc");
+              if (p.context("handler") != null) p.remove("handler");
+              if (p.context("expect-continue") != null) p.remove("expect-continue");
+              if (p.context("compressor") != null) p.remove("compressor");
+              if (p.context("ws-compressor") != null) p.remove("ws-compressor");
+
+              // Remove this cleaner itself so it doesn't linger
+              p.remove(this);
+            }
+            super.userEventTriggered(ctx, evt);
+          }
+        });
+
     addCommonHandlers(pipeline);
 
     // Inject gRPC handler (isHttp2 = false to trigger 426 Upgrade Required)

Original file line number	Diff line number	Diff line change
`@@ -21,6 +21,7 @@`
`21`	`21`	`import io.netty.handler.codec.http.multipart.*;`
`22`	`22`	`import io.netty.handler.codec.http.websocketx.WebSocketFrame;`
`23`	`23`	`import io.netty.handler.timeout.IdleStateEvent;`
	`24`	`+import io.netty.util.ReferenceCountUtil;`
`24`	`25`
`25`	`26`	`public class NettyHandler extends ChannelInboundHandlerAdapter {`
`26`	`27`	`private final Logger log = LoggerFactory.getLogger(NettyServer.class);`
`@@ -88,6 +89,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {`
`88`	`89`	`if (req.getClass() == DefaultFullHttpRequest.class) {`
`89`	`90`	`// HTTP2 aggregates all into a full http request.`
`90`	`91`	`if (((DefaultFullHttpRequest) req).content().readableBytes() > maxRequestSize) {`
	`92`	`+ release(req);`
`91`	`93`	`router.match(context).execute(context, Route.REQUEST_ENTITY_TOO_LARGE);`
`92`	`94`	`return;`
`93`	`95`	`}`
`@@ -127,6 +129,8 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {`
`127`	`129`	`} else if (isWebSocketFrame(msg)) {`
`128`	`130`	`if (context.webSocket != null) {`
`129`	`131`	`context.webSocket.handleFrame((WebSocketFrame) msg);`
	`132`	`+ } else {`
	`133`	`+ release(msg);`
`130`	`134`	`}`
`131`	`135`	`}`
`132`	`136`	`}`
`@@ -189,9 +193,9 @@ public void writeChunks(Object header, Object body, ChannelPromise promise) {`
`189`	`193`	`}`
`190`	`194`	`}`
`191`	`195`
`192`		`- private void release(HttpContent ref) {`
`193`		`- if (ref.refCnt() > 0) {`
`194`		`- ref.release();`
	`196`	`+ private void release(Object ref) {`
	`197`	`+ if (ReferenceCountUtil.refCnt(ref) > 0) {`
	`198`	`+ ReferenceCountUtil.release(ref);`
`195`	`199`	`}`
`196`	`200`	`}`
`197`	`201`