From b8ab0ed7ed30e71c8fb9409d16a4394b2c3a7838 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Tue, 20 Oct 2026 19:07:29 -0400 Subject: [PATCH] fix: stabilize Netty connector tests failing on 3.0 CI Address the failures detected on PR #1553 and on the 3.0 branch: - ChannelOutputStream: writes from a call thread only update the channel writability once the event loop processes them, so a fast producer could queue a whole response body and never notice that the stream was reset (flaky Http2CancellationTestCase). Bound the bytes in flight to the channel high water mark and fail on the first failed write. - NettyServerHandler: end the request body at once for HTTP/1.1 requests without a body. The call could otherwise complete before the empty last content was delivered and close a connection announced as kept alive (flaky RedirectTestCase.stopsAfterMaxRedirects). - NettyServerCall: split writeHead to fix the SonarCloud cognitive complexity issue (java:S3776), without behavior change. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../engine/netty/ChannelOutputStream.java | 86 ++++++++++++++++++- .../restlet/engine/netty/NettyServerCall.java | 77 +++++++++-------- .../engine/netty/NettyServerHandler.java | 11 +++ 3 files changed, 136 insertions(+), 38 deletions(-) diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/ChannelOutputStream.java b/org.restlet/src/main/java/org/restlet/engine/netty/ChannelOutputStream.java index 3945ee057a..b7e79884b3 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/ChannelOutputStream.java +++ b/org.restlet/src/main/java/org/restlet/engine/netty/ChannelOutputStream.java @@ -12,7 +12,12 @@ import io.netty.channel.Channel; import io.netty.channel.ChannelFuture; import java.io.IOException; +import java.io.InterruptedIOException; import java.io.OutputStream; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.ReentrantLock; import java.util.function.Function; /** @@ -42,6 +47,16 @@ final class ChannelOutputStream extends OutputStream { private boolean closed; + /** Bytes written to the channel whose write has not completed yet. */ + private final AtomicLong inFlightBytes = new AtomicLong(); + + private final ReentrantLock inFlightLock = new ReentrantLock(); + + private final Condition inFlightChanged = inFlightLock.newCondition(); + + /** The cause of the first failed write, if any. */ + private volatile Throwable writeFailure; + /** * Constructor. * @@ -75,8 +90,8 @@ public void write(int b) throws IOException { @Override public void write(byte[] b, int off, int len) throws IOException { - ensureOpen(); while (len > 0) { + ensureOpen(); ensureBuffer(); int n = Math.min(len, buffer.writableBytes()); buffer.writeBytes(b, off, n); @@ -131,19 +146,88 @@ private void ensureOpen() throws IOException { if (closed) { throw new IOException("Stream closed"); } + checkWriteFailure(); if (!channel.isActive()) { throw new IOException("Connection closed by peer"); } } + private void checkWriteFailure() throws IOException { + Throwable cause = writeFailure; + if (cause != null) { + throw new IOException("Unable to write to the connection", cause); + } + } + private void send(boolean flush) throws IOException { Writability.await(channel, writeTimeoutMs); + awaitInFlightBytes(); ByteBuf chunk = buffer; buffer = null; + int size = chunk.readableBytes(); + inFlightBytes.addAndGet(size); Object msg = contentFactory.apply(chunk); ChannelFuture future = flush ? channel.writeAndFlush(msg) : channel.write(msg); + future.addListener(f -> onWriteDone(size, f.cause())); if (future.isDone() && !future.isSuccess()) { throw new IOException("Unable to write to the connection", future.cause()); } } + + /** Called when a write completes, possibly on the event loop. */ + private void onWriteDone(int size, Throwable cause) { + inFlightLock.lock(); + try { + inFlightBytes.addAndGet(-size); + if (cause != null && writeFailure == null) { + writeFailure = cause; + } + inFlightChanged.signalAll(); + } finally { + inFlightLock.unlock(); + } + } + + /** + * Waits until the bytes written but not yet sent fall below the channel high water mark. Writes + * made outside the event loop only update the channel writability once the event loop has + * processed them, so a fast producer could otherwise queue unbounded data, and miss the failure + * of writes to a connection closed in the meantime. + */ + private void awaitInFlightBytes() throws IOException { + int limit = Math.max(bufferSize, channel.config().getWriteBufferHighWaterMark()); + if (inFlightBytes.get() <= limit || channel.eventLoop().inEventLoop()) { + checkWriteFailure(); + return; + } + + channel.flush(); + long remaining = + (writeTimeoutMs > 0) + ? TimeUnit.MILLISECONDS.toNanos(writeTimeoutMs) + : Long.MAX_VALUE; + inFlightLock.lock(); + try { + while (inFlightBytes.get() > limit && writeFailure == null) { + if (!channel.isActive()) { + throw new IOException("Connection closed by peer"); + } + if (remaining <= 0) { + throw new InterruptedIOException("Write timed out"); + } + // Wake up periodically in case the channel closes without failing a write. + long slice = Math.min(remaining, TimeUnit.MILLISECONDS.toNanos(100)); + long left = inFlightChanged.awaitNanos(slice); + if (writeTimeoutMs > 0) { + remaining -= (slice - Math.max(0, left)); + } + } + } catch (InterruptedException _) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException("Interrupted while writing"); + } finally { + inFlightLock.unlock(); + } + checkWriteFailure(); + } } diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java index 43d7e14c3a..96793418b0 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java +++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java @@ -608,6 +608,27 @@ private void writeHead(boolean emptyBody) { responseHeadWritten = true; int code = statusCode; + HttpResponse head = newResponseHead(code); + boolean bodyAllowed = + code != 204 && code != 304 && code >= 200 && !"HEAD".equals(getMethod()); + boolean hasTrailers = bodyAllowed && !responseTrailers.isEmpty(); + // Streams can carry trailers added after the head was written. + responseTrailersEnabled = hasTrailers || (kind.isStream() && bodyAllowed); + + if (kind.isStream()) { + // Connection-specific headers are forbidden in HTTP/2 and HTTP/3. + head.headers().remove(HttpHeaderNames.CONNECTION); + head.headers().remove(HttpHeaderNames.TRANSFER_ENCODING); + head.headers().remove("Keep-Alive"); + } else { + setHttp1Framing(head, emptyBody, hasTrailers, bodyAllowed); + } + + channel.write(head); + } + + /** Creates the response head with the status and the response headers. */ + private HttpResponse newResponseHead(int code) { String reason = reasonPhrase; HttpResponseStatus status = (reason != null && !reason.isEmpty()) @@ -624,45 +645,27 @@ private void writeHead(boolean emptyBody) { if (altSvc != null && !head.headers().contains("Alt-Svc")) { head.headers().set(kind.isStream() ? "alt-svc" : "Alt-Svc", altSvc); } - - boolean hasTrailers = !responseTrailers.isEmpty(); - boolean hasLength = head.headers().contains(HttpHeaderNames.CONTENT_LENGTH); - boolean noBodyStatus = code == 204 || code == 304 || code < 200; - - if (noBodyStatus || "HEAD".equals(getMethod())) { - hasTrailers = false; - } - responseTrailersEnabled = - hasTrailers || (kind.isStream() && !noBodyStatus && !"HEAD".equals(getMethod())); - - if (kind.isStream()) { - // Connection-specific headers are forbidden in HTTP/2 and HTTP/3. - head.headers().remove(HttpHeaderNames.CONNECTION); - head.headers().remove(HttpHeaderNames.TRANSFER_ENCODING); - head.headers().remove("Keep-Alive"); - } else { - if (emptyBody) { - if (hasTrailers) { - head.headers().remove(HttpHeaderNames.CONTENT_LENGTH); - HttpUtil.setTransferEncodingChunked(head, true); - } else if (!hasLength && !noBodyStatus && !"HEAD".equals(getMethod())) { - head.headers().set(HttpHeaderNames.CONTENT_LENGTH, 0); - } - } else if (hasTrailers) { - head.headers().remove(HttpHeaderNames.CONTENT_LENGTH); - HttpUtil.setTransferEncodingChunked(head, true); - } else if (!hasLength) { - head.headers().set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED); - } - if (hasTrailers) { - setTrailerNames(head.headers(), responseTrailers); - } - if (!HttpUtil.isKeepAlive(request)) { - head.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE); + return head; + } + + /** Sets the HTTP/1.1 message framing and connection headers of the response head. */ + private void setHttp1Framing( + HttpResponse head, boolean emptyBody, boolean hasTrailers, boolean bodyAllowed) { + HttpHeaders headers = head.headers(); + if (hasTrailers) { + headers.remove(HttpHeaderNames.CONTENT_LENGTH); + HttpUtil.setTransferEncodingChunked(head, true); + setTrailerNames(headers, responseTrailers); + } else if (!headers.contains(HttpHeaderNames.CONTENT_LENGTH)) { + if (!emptyBody) { + headers.set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED); + } else if (bodyAllowed) { + headers.set(HttpHeaderNames.CONTENT_LENGTH, 0); } } - - channel.write(head); + if (!HttpUtil.isKeepAlive(request)) { + headers.set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE); + } } /** Completes the exchange: ends the response and frees the connection for the next request. */ diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java index d6c65a2310..e6c7223b69 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java +++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java @@ -115,6 +115,11 @@ private void onRequest(ChannelHandlerContext ctx, HttpRequest request) { ChannelInputStream body = new ChannelInputStream(ctx.channel(), helper.getConnectorIdleTimeout()); currentBody = body; + if (kind == StreamKind.HTTP1 && !hasBody(request)) { + // End the body now: the call could otherwise complete before the last (empty) content + // is delivered, and close a connection that the response announced as kept alive. + body.end(); + } NettyServerCall call = new NettyServerCall(helper, ctx.channel(), kind, request, body, confidential); @@ -190,6 +195,12 @@ private void onContent(ChannelHandlerContext ctx, HttpContent content) { // Otherwise the body stream requests the next read once the consumer catches up. } + /** Indicates whether an HTTP/1.1 request has a body, according to its framing headers. */ + private static boolean hasBody(HttpRequest request) { + return HttpUtil.isTransferEncodingChunked(request) + || HttpUtil.getContentLength(request, 0L) != 0; + } + private void runCall(NettyServerCall call) { try { helper.handle(call);