fix: keep the status of synchronous responses on late client failures

changes.md

@@ -218,6 +218,11 @@ Changes log
         - In the "HTTP2" transport mode, the HTTP client uses HTTP/1.1 for cleartext requests
           sent through a forward proxy, instead of HTTP/2 with prior knowledge that HTTP proxies
           generally refuse. Issue #1571.
+        - The HTTP client no longer replaces the status of a synchronous response with a
+          communication error (1001) when the connection fails after the response head, for
+          example when the client stops before the end of the body: the failure reaches the
+          entity stream only, and is ignored after the head of a response announcing an empty
+          body. The failures caused by stopping the client report "Client stopped". Issue #1601.
         - The `Max-Age` and `Path` attributes of a cookie setting are no longer quoted. Issue #1334.
         - `Restlet.handle()` no longer fails with a `NullPointerException` when the Restlet has no context and cannot be started.
         - `CookieAuthenticator` no longer lets the next restlet overwrite the redirection sent after

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

@@ -163,6 +163,12 @@ class NettyClientCall {
     /** Indicates whether the response was fully received. */
     private boolean responseComplete;
 
+    /**
+     * Indicates whether the end of the response body is known: received, or announced by a zero
+     * "Content-Length" header.
+     */
+    private boolean bodyEnded;
+
     /** Indicates whether the connection can carry another exchange. */
     private boolean reusable = true;
 
@@ -373,9 +379,10 @@ private void dispatchResponseFailureCallback(Throwable cause) {
         if (request == null || currentResponse == null) {
             return;
         }
-        currentResponse.setStatus(Status.CONNECTOR_ERROR_COMMUNICATION, cause);
         Uniform callback = request.getOnResponse();
         if (callback == null) {
+            // The status was returned to a synchronous caller and is kept: the failure reaches the
+            // response entity stream only.
             return;
         }
         if (responseCallbackLock.isHeldByCurrentThread()) {
@@ -384,8 +391,10 @@ private void dispatchResponseFailureCallback(Throwable cause) {
         }
         responseCallbackLock.lock();
         try {
+            // Otherwise, the initial callback, not dispatched yet, reports the failure.
             if (responseCallbackDispatched && !responseFailureCallbackDispatched) {
                 responseFailureCallbackDispatched = true;
+                currentResponse.setStatus(Status.CONNECTOR_ERROR_COMMUNICATION, cause);
                 invokeResponseCallback(callback, request, currentResponse);
             }
         } finally {
@@ -708,7 +717,7 @@ private Representation getResponseEntity(Status status) {
             InputStream stream = getResponseEntityStream();
             // Multiplexed streams may remain idle after their response headers arrive.
             if (!kind.isStream()) {
-                stream = nonEmpty(stream);
+                stream = nonEmpty(stream, size == 0);
             }
             if (stream != null) {
                 result = new InputRepresentation(stream, null);
@@ -719,8 +728,16 @@ private Representation getResponseEntity(Status status) {
         return HeaderUtils.extractEntityHeaders(responseHeaders, result);
     }
 
-    /** Returns a stream equivalent to the given one, or null if it is null or empty. */
-    private static InputStream nonEmpty(InputStream stream) {
+    /**
+     * Returns a stream equivalent to the given one, or null if it is null or empty. A stream
+     * failing before its first byte is returned as is, so that reading it reports the truncated
+     * response, unless the response announced an empty body.
+     *
+     * @param stream The response entity stream.
+     * @param emptyAnnounced True if the response announced an empty body.
+     * @return An equivalent stream, or null.
+     */
+    private static InputStream nonEmpty(InputStream stream, boolean emptyAnnounced) {
         if (stream == null) {
             return null;
         }
@@ -735,7 +752,8 @@ private static InputStream nonEmpty(InputStream stream) {
                 return result;
             }
         } catch (IOException _) {
-            // End of the response entity stream.
+            // Each later read of the stream reports the failure again.
+            return emptyAnnounced ? null : stream;
         }
         return null;
     }
@@ -942,10 +960,15 @@ void onResponseHead(HttpResponse response) {
                                 + "."
                                 + response.protocolVersion().minorVersion();
         this.body = new ChannelInputStream(channel, helper.getIdleTimeout());
+        // A failure after the head of a response announcing an empty body loses nothing.
+        boolean empty =
+                HttpUtil.getContentLength(response, -1L) == 0
+                        && !HttpUtil.isTransferEncodingChunked(response);
 
         lock.lock();
         try {
             head = true;
+            bodyEnded = empty;
             headReceived.signalAll();
         } finally {
             lock.unlock();
@@ -983,19 +1006,22 @@ void onResponseComplete(io.netty.handler.codec.http.HttpHeaders trailers) {
         lock.lock();
         try {
             responseComplete = true;
+            bodyEnded = true;
             emptyEntity = pendingEmptyEntity;
             pendingEmptyEntity = null;
         } finally {
             lock.unlock();
         }
+        // Release the connection first, so that a synchronous caller resuming at the end of the
+        // body finds it back in the pool for its next request.
+        maybeRelease();
         ChannelInputStream stream = this.body;
         if (stream != null) {
             stream.end();
         }
         if (emptyEntity != null) {
             emptyEntity.release();
         }
-        maybeRelease();
     }
 
     /**
@@ -1019,7 +1045,7 @@ void onFailure(Throwable cause) {
             if (!head) {
                 failure = cause;
                 headReceived.signalAll();
-            } else if (responseFailure == null) {
+            } else if (responseFailure == null && !bodyEnded) {
                 responseFailure = cause;
                 notifyResponseFailure = true;
             }

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

@@ -8,6 +8,7 @@
  */
 package org.restlet.engine.netty;
 
+import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.handler.codec.DecoderResultProvider;
@@ -16,6 +17,7 @@
 import io.netty.handler.codec.http.HttpResponseStatus;
 import io.netty.handler.codec.http.HttpStatusClass;
 import io.netty.handler.codec.http.LastHttpContent;
+import io.netty.util.AttributeKey;
 import io.netty.util.ReferenceCountUtil;
 import java.io.IOException;
 
@@ -28,6 +30,10 @@
  */
 class NettyClientHandler extends ChannelInboundHandlerAdapter {
 
+    /** Marks the connections closed because the client connector stops. */
+    static final AttributeKey<Boolean> CLIENT_STOPPED =
+            AttributeKey.valueOf(NettyClientHandler.class, "clientStopped");
+
     /** The call in progress, if any. */
     private volatile NettyClientCall call;
 
@@ -121,11 +127,25 @@ private boolean onContent(NettyClientCall current, HttpContent content) {
     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
         NettyClientCall current = this.call;
         if (current != null) {
-            current.onFailure(new IOException("Connection closed by peer"));
+            current.onFailure(
+                    new IOException(
+                            isClientStopped(ctx.channel())
+                                    ? "Client stopped"
+                                    : "Connection closed by peer"));
         }
         super.channelInactive(ctx);
     }
 
+    /** Indicates whether a channel, or its parent connection, was closed by the client's stop. */
+    private static boolean isClientStopped(Channel channel) {
+        for (Channel c = channel; c != null; c = c.parent()) {
+            if (Boolean.TRUE.equals(c.attr(CLIENT_STOPPED).get())) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
         NettyClientCall current = this.call;

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

@@ -1054,6 +1054,7 @@ public synchronized void stop() throws Exception {
         EventLoopGroup g = group;
         group = null;
         if (connections != null) {
+            connections.forEach(c -> c.attr(NettyClientHandler.CLIENT_STOPPED).set(Boolean.TRUE));
             connections.close().awaitUninterruptibly(5, TimeUnit.SECONDS);
         }
         pools.clear();

org.restlet/src/test/java/org/restlet/engine/connector/ClientLateFailureTestCase.java

@@ -0,0 +1,185 @@
+/**
+ * Copyright 2005-2026 Qlik
+ *<p>
+ * The content of this file is subject to the terms of the Apache 2.0 open
+ * source license available at https://www.opensource.org/licenses/apache-2.0
+ *<p>
+ * Restlet is a registered trademark of QlikTech International AB.
+ */
+package org.restlet.engine.connector;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.restlet.Client;
+import org.restlet.Context;
+import org.restlet.Request;
+import org.restlet.Response;
+import org.restlet.data.Method;
+import org.restlet.data.Protocol;
+import org.restlet.data.Status;
+import org.restlet.engine.Engine;
+
+/**
+ * Tests the failures of the client connector after the response head was returned to a synchronous
+ * caller (issue #1601).
+ */
+class ClientLateFailureTestCase {
+
+    private ServerSocket serverSocket;
+
+    private final ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
+
+    private Client client;
+
+    @BeforeEach
+    void setUp() throws Exception {
+        Engine.clearThreadLocalVariables();
+        Engine.register(false);
+        Engine.getInstance().getRegisteredClients().add(new HttpClientHelper(null));
+        serverSocket = new ServerSocket(0, 50, InetAddress.getLoopbackAddress());
+        client = new Client(new Context(), Protocol.HTTP);
+        client.start();
+    }
+
+    @AfterEach
+    void tearDown() throws Exception {
+        client.stop();
+        serverSocket.close();
+        executor.shutdownNow();
+        Engine.clearThreadLocalVariables();
+    }
+
+    @Test
+    void keepsTheStatusWhenTheClientStopsBeforeTheEndOfTheResponse() throws Exception {
+        CountDownLatch clientStopped = new CountDownLatch(1);
+        executor.execute(
+                () -> {
+                    try (Socket socket = serverSocket.accept()) {
+                        readRequestHead(socket.getInputStream());
+                        OutputStream out = socket.getOutputStream();
+                        write(
+                                out,
+                                "HTTP/1.1 403 Forbidden\r\n"
+                                        + "Content-Type: text/plain\r\n"
+                                        + "Transfer-Encoding: chunked\r\n\r\n"
+                                        + "9\r\nForbidden\r\n");
+                        // The end of the response, sent once the client is stopped.
+                        clientStopped.await(5, TimeUnit.SECONDS);
+                        write(out, "0\r\n\r\n");
+                    } catch (IOException | InterruptedException _) {
+                        // Connection closed by the client.
+                    }
+                });
+
+        Response response = client.handle(new Request(Method.GET, uri("/")));
+        assertEquals(Status.CLIENT_ERROR_FORBIDDEN, response.getStatus());
+
+        client.stop();
+        clientStopped.countDown();
+
+        // The failure reaches the response entity stream...
+        IOException failure =
+                assertThrows(IOException.class, response.getEntity().getStream()::readAllBytes);
+
+        // ...but the status returned to the caller is kept.
+        long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(500);
+        while (System.nanoTime() < deadline) {
+            assertEquals(Status.CLIENT_ERROR_FORBIDDEN, response.getStatus());
+            Thread.sleep(10);
+        }
+        assertEquals("Client stopped", failure.getMessage());
+    }
+
+    @Test
+    void reportsABodyTruncatedBeforeItsFirstByte() throws Exception {
+        executor.execute(
+                () -> {
+                    try (Socket socket = serverSocket.accept()) {
+                        readRequestHead(socket.getInputStream());
+                        write(
+                                socket.getOutputStream(),
+                                "HTTP/1.1 200 OK\r\n"
+                                        + "Content-Type: text/plain\r\n"
+                                        + "Content-Length: 10\r\n\r\n");
+                    } catch (IOException _) {
+                        // Connection closed by the client.
+                    }
+                });
+
+        Response response = client.handle(new Request(Method.GET, uri("/")));
+
+        // The status is kept, and reading the entity reports the truncated body.
+        assertEquals(Status.SUCCESS_OK, response.getStatus());
+        assertNotNull(response.getEntity());
+        assertThrows(IOException.class, response.getEntity().getStream()::readAllBytes);
+    }
+
+    @Test
+    void completesResponsesWithAnEmptyBodyWithTheirHead() throws Exception {
+        AtomicInteger connections = new AtomicInteger();
+        executor.execute(
+                () -> {
+                    try (Socket socket = serverSocket.accept()) {
+                        connections.incrementAndGet();
+                        for (int i = 0; i < 2; i++) {
+                            readRequestHead(socket.getInputStream());
+                            write(
+                                    socket.getOutputStream(),
+                                    "HTTP/1.1 403 Forbidden\r\nContent-Length: 0\r\n\r\n");
+                        }
+                        socket.getInputStream().read();
+                    } catch (IOException _) {
+                        // Connection closed by the client.
+                    }
+                });
+
+        for (int i = 0; i < 2; i++) {
+            Response response = client.handle(new Request(Method.GET, uri("/" + i)));
+            assertEquals(Status.CLIENT_ERROR_FORBIDDEN, response.getStatus());
+            assertTrue(
+                    (response.getEntity() == null) || response.getEntity().isEmpty(),
+                    String.valueOf(response.getEntity()));
+        }
+
+        // The connection is reused for the second request.
+        assertEquals(1, connections.get());
+    }
+
+    private String uri(String path) {
+        return "http://localhost:" + serverSocket.getLocalPort() + path;
+    }
+
+    private static void readRequestHead(InputStream in) throws IOException {
+        int matched = 0;
+        while (matched < 4) {
+            int b = in.read();
+            if (b < 0) {
+                throw new IOException("Connection closed");
+            }
+            matched = (b == "\r\n\r\n".charAt(matched)) ? matched + 1 : ((b == '\r') ? 1 : 0);
+        }
+    }
+
+    private static void write(OutputStream out, String text) throws IOException {
+        out.write(text.getBytes(StandardCharsets.US_ASCII));
+        out.flush();
+    }
+}