fix: keep or announce the HTTP/1.1 connection when the request body is unread

changes.md

@@ -232,6 +232,11 @@ Changes log
         - The gRPC client could reject a call with `RESOURCE_EXHAUSTED` ("gRPC call limit exceeded")
           right after the end of the previous one: the call slot was released after the caller was
           notified. Issue #1622.
+        - The HTTP/1.1 server connector no longer closes a kept-alive connection without notice when
+          the application answers without reading the request body (for example an early 401, 403
+          or 406). The unread body is discarded when it is already received, or when at most
+          256 KiB of it remain in transit; otherwise the response announces `Connection: close`.
+          Clients reusing the connection failed with a "Communication Error" (1001).
     - Misc
         - Fixed JDK HTTP client accessibility in core and WebSocket tests in Eclipse and VS Code,
           without adding a production module dependency.

org.restlet/src/main/java/org/restlet/engine/netty/ChannelInputStream.java

@@ -58,6 +58,12 @@ final class ChannelInputStream extends InputStream {
 
     private boolean readPaused;
 
+    /** The total number of content bytes offered, consumed or not. */
+    private long receivedBytes;
+
+    /** Whether content offered after closing is discarded without pausing channel reads. */
+    private boolean discarding;
+
     /**
      * Constructor.
      *
@@ -79,9 +85,10 @@ public ChannelInputStream(Channel channel, long readTimeoutMs) {
     public boolean offer(ByteBuf buf) {
         lock.lock();
         try {
+            receivedBytes += buf.readableBytes();
             if (closed || endOfStream || !buf.isReadable()) {
                 buf.release();
-                return !closed;
+                return !closed || discarding;
             }
             buffers.addLast(buf);
             bufferedBytes += buf.readableBytes();
@@ -128,17 +135,46 @@ public void fail(Throwable cause) {
     }
 
     /**
-     * Indicates whether the whole content has been received and consumed.
+     * Indicates whether the whole content has been received, consumed or not.
      *
-     * @return True if the whole content has been received and consumed.
+     * @return True if the whole content has been received.
+     */
+    public boolean isFullyReceived() {
+        lock.lock();
+        try {
+            return endOfStream;
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    /**
+     * Returns the number of content bytes received so far, consumed or not.
+     *
+     * @return The number of content bytes received so far.
+     */
+    public long getReceivedBytes() {
+        lock.lock();
+        try {
+            return receivedBytes;
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    /**
+     * Closes the stream and discards the rest of the content: buffered content is released, and
+     * content offered later is released as it arrives, without pausing channel reads. The caller is
+     * responsible for bounding the content still expected.
      */
-    public boolean isFullyConsumed() {
+    public void discard() {
         lock.lock();
         try {
-            return endOfStream && buffers.isEmpty();
+            discarding = true;
         } finally {
             lock.unlock();
         }
+        close();
     }
 
     @Override

org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java

@@ -80,6 +80,12 @@
  */
 class NettyServerCall {
 
+    /**
+     * Maximum number of unread HTTP/1.1 request body bytes still in transit that a response can
+     * discard while keeping the connection alive.
+     */
+    static final long MAX_DISCARDED_BYTES = ChannelInputStream.HIGH_WATER_MARK;
+
     /** The parent helper. */
     private final NettyServerHelper helper;
 
@@ -125,6 +131,9 @@ class NettyServerCall {
     /** Indicates whether the response head was written. */
     private volatile boolean responseHeadWritten;
 
+    /** Indicates whether the HTTP/1.1 response head announced a persistent connection. */
+    private volatile boolean keepAliveAnnounced;
+
     /** Indicates that an empty multiplexed response ended with its initial headers. */
     private boolean responseEndedWithHead;
 
@@ -699,9 +708,33 @@ private void setHttp1Framing(
                 headers.set(HttpHeaderNames.CONTENT_LENGTH, 0);
             }
         }
-        if (!HttpUtil.isKeepAlive(request)) {
+        if (!HttpUtil.isKeepAlive(request) || !isRequestBodyDiscardable()) {
             headers.set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE);
         }
+        keepAliveAnnounced = HttpUtil.isKeepAlive(head);
+    }
+
+    /**
+     * Indicates whether the rest of the HTTP/1.1 request body, if left unread by the application,
+     * can be discarded without closing the connection. It can when it was fully received, or when
+     * its declared length leaves at most {@link #MAX_DISCARDED_BYTES} bytes to receive. A chunked
+     * body still in transit has an unknown length, so the connection must then be closed.
+     *
+     * <p>This is decided when the response head is written, so that a connection which can't be
+     * reused is announced with a "Connection: close" header instead of being closed without notice
+     * after the response.
+     *
+     * @return True if the rest of the request body can be discarded.
+     */
+    private boolean isRequestBodyDiscardable() {
+        if (requestBody.isFullyReceived()) {
+            return true;
+        }
+        if (HttpUtil.isTransferEncodingChunked(request)) {
+            return false;
+        }
+        long length = HttpUtil.getContentLength(request, -1L);
+        return length >= 0 && length - requestBody.getReceivedBytes() <= MAX_DISCARDED_BYTES;
     }
 
     /** Completes the exchange: ends the response and frees the connection for the next request. */
@@ -734,7 +767,9 @@ void complete() {
                 return;
             }
 
-            boolean keepAlive = HttpUtil.isKeepAlive(request) && requestBody.isFullyConsumed();
+            // The response head announced whether the connection persists. If so, the unread
+            // request body is discarded: it is either already received or bounded.
+            boolean keepAlive = !kind.isStream() && keepAliveAnnounced;
             LastHttpContent lastContent = LastHttpContent.EMPTY_LAST_CONTENT;
             if (responseTrailersEnabled) {
                 DefaultLastHttpContent withTrailers =
@@ -743,7 +778,11 @@ void complete() {
                 lastContent = withTrailers;
             }
             ChannelFuture future = trackDelivery(channel.writeAndFlush(lastContent));
-            requestBody.close();
+            if (keepAlive) {
+                requestBody.discard();
+            } else {
+                requestBody.close();
+            }
 
             if (!kind.isStream()) {
                 completeHttp1(future, keepAlive);