diff --git a/changes.md b/changes.md index b11842b2dd..ea200e87ce 100644 --- a/changes.md +++ b/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 diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientCall.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientCall.java index 3cf4cf1de4..5f47c3e176 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientCall.java +++ b/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,11 +1006,15 @@ 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(); @@ -995,7 +1022,6 @@ void onResponseComplete(io.netty.handler.codec.http.HttpHeaders trailers) { 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; } diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientHandler.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientHandler.java index 7f79b04614..185f5f6b40 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientHandler.java +++ b/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 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; diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientHelper.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientHelper.java index 05b351570d..1e0cefd571 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientHelper.java +++ b/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(); diff --git a/org.restlet/src/test/java/org/restlet/engine/connector/ClientLateFailureTestCase.java b/org.restlet/src/test/java/org/restlet/engine/connector/ClientLateFailureTestCase.java new file mode 100644 index 0000000000..84f97cb04d --- /dev/null +++ b/org.restlet/src/test/java/org/restlet/engine/connector/ClientLateFailureTestCase.java @@ -0,0 +1,185 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * 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 + *

+ * 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(); + } +}