diff --git a/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java b/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java index 5886eda34..0b4cf3a2d 100644 --- a/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java +++ b/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java @@ -52,9 +52,12 @@ public interface ResponseBodyControl { * waits for demand only while the system property {@code io.netty.compression.defaultMaxForwardBytes} is positive, * as it is by default. Once the body has been decompressed in full, the final empty part, trailing headers and * completion can still arrive while reads are suspended. If the connection closes while compressed input is held - * back, the rest of it is decompressed and delivered at once, bounded only by - * {@link AsyncHttpClientConfig#getMaxDecompressedResponseSize()}. A transfer coding, such as - * {@code Transfer-Encoding: gzip}, is decompressed as soon as it is read. + * back, the rest of a body that has ended is decompressed and delivered at once, bounded only by + * {@link AsyncHttpClientConfig#getMaxDecompressedResponseSize()}. The rest of a body that the close cut off is + * dropped, and the request fails as for any connection closed before the end of the body. Native transports such + * as epoll notice an orderly close by the server (a TCP FIN) even while reads are suspended; NIO notices it only + * once the input held back has been used up. A transfer coding, such as {@code Transfer-Encoding: gzip}, is + * decompressed as soon as it is read. *

* While reads are suspended, the read timeout is paused but the request timeout remains active. If the request * timeout is disabled, failing to resume or cancel the response can retain its transport resources indefinitely. diff --git a/client/src/main/java/org/asynchttpclient/netty/handler/Http1ContentDecompressor.java b/client/src/main/java/org/asynchttpclient/netty/handler/Http1ContentDecompressor.java index a0efe5760..664cbf449 100644 --- a/client/src/main/java/org/asynchttpclient/netty/handler/Http1ContentDecompressor.java +++ b/client/src/main/java/org/asynchttpclient/netty/handler/Http1ContentDecompressor.java @@ -88,8 +88,9 @@ * request, such as the one {@code resume()} issues, decompresses that input before it is passed on to the transport. * That is why this handler also sees outbound events; it passes all others through. Only parts with content wait for * a resume: once the body has been decompressed in full, its end, with any trailers, is handed on even to a suspended - * response, as for an unencoded body. If the connection closes while input is held back, the rest of it is - * decompressed and handed on at once, bounded only by {@code maxDecompressedBytes}. + * response, as for an unencoded body. If the connection closes while input is held back and the body has ended, the + * rest of it is decompressed and handed on at once, bounded only by {@code maxDecompressedBytes}. If the close cut the + * body off, the response fails anyway, and the input held back is dropped without being decompressed. *

* {@code br}, {@code zstd} and {@code snappy} are decompressed the same way, one part at a time, by * {@link PullDecompression} with Netty's pull-based {@code Decompressor}. Netty added that API in 4.2.17; with an @@ -293,11 +294,11 @@ public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { - // The pipeline is about to be torn down, so what is held back is handed on now, even to a suspended - // response. + // The pipeline is about to be torn down, so what is held back of a body that has ended is handed on now, even + // to a suspended response. A body that the close cut off fails anyway, so what is held back of it is dropped. inputClosed = true; try { - if (paused && !draining) { + if (paused && !draining && bodyEnded()) { readPending = false; resume(ctx); } @@ -392,6 +393,20 @@ private void decodeQueued(ChannelHandlerContext ctx, HttpObject msg) throws Exce } } + // Whether the end of the paused body has arrived, with the input that is held or behind it. HttpObjectDecoder hands + // on the end of a body delimited by the connection before the connection's close. + private boolean bodyEnded() { + if (heldLast != null) { + return true; + } + for (HttpObject msg : queued) { + if (msg instanceof LastHttpContent) { + return true; + } + } + return false; + } + private void releaseQueued() { for (HttpObject msg; (msg = queued.poll()) != null; ) { ReferenceCountUtil.release(msg); diff --git a/client/src/test/java/org/asynchttpclient/Http1DecompressionSuspensionTest.java b/client/src/test/java/org/asynchttpclient/Http1DecompressionSuspensionTest.java index 3d0bfbb25..3dc5fa51c 100644 --- a/client/src/test/java/org/asynchttpclient/Http1DecompressionSuspensionTest.java +++ b/client/src/test/java/org/asynchttpclient/Http1DecompressionSuspensionTest.java @@ -24,6 +24,7 @@ import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.channel.epoll.Epoll; import io.netty.channel.group.ChannelGroup; import io.netty.channel.group.DefaultChannelGroup; import io.netty.channel.nio.NioEventLoopGroup; @@ -45,6 +46,8 @@ import io.netty.util.HashedWheelTimer; import io.netty.util.concurrent.GlobalEventExecutor; import org.asynchttpclient.channel.ChannelPool; +import org.asynchttpclient.exception.ChannelClosedException; +import org.asynchttpclient.exception.RemotelyClosedException; import org.asynchttpclient.filter.FilterContext; import org.asynchttpclient.filter.IOExceptionFilter; import org.asynchttpclient.filter.ResponseFilter; @@ -53,7 +56,11 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.EnabledOnOs; +import org.junit.jupiter.api.condition.OS; import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import java.io.ByteArrayOutputStream; import java.io.IOException; @@ -65,6 +72,7 @@ import java.util.Random; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ExecutionException; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -82,9 +90,12 @@ import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assumptions.assumeTrue; /** * A suspended HTTP/1.1 response must not keep inflating the compressed bytes it has already read. A highly @@ -268,6 +279,43 @@ public void bodyEndedByConnectionCloseIsDeliveredIntactAcrossSuspensions() throw } } + /** + * Native transports read the close of a connection even while reads are suspended. What is held back of a body + * that the close cuts off is dropped, so the request fails as a closed connection, without the rest of the body + * and without the size limit that inflating it would have hit first. + */ + @ParameterizedTest + @ValueSource(strings = {"/zeros-cut", "/zeros-cut-chunked"}) + @EnabledOnOs(OS.LINUX) + public void aSuspendedBodyThatTheConnectionCutsOffFailsWithoutTheRestOfIt(String path) throws Exception { + assumeTrue(Epoll.isAvailable(), "epoll is not available"); + List filtered = new CopyOnWriteArrayList<>(); + IOExceptionFilter filter = new IOExceptionFilter() { + @Override + public FilterContext filter(FilterContext ctx) { + filtered.add(ctx.getIOException()); + return ctx; + } + }; + try (AsyncHttpClient client = asyncHttpClient(config() + .setUseNativeTransport(true) + .setUseOnlyEpollNativeTransport(true) + .setMaxRequestRetry(0) + .setMaxDecompressedResponseSize(1024 * 1024) + .addIOExceptionFilter(filter) + .setRequestTimeout(Duration.ofSeconds(60)))) { + SuspendingHandler handler = new SuspendingHandler(); + ListenableFuture response = client.prepareGet(url(path)).execute(handler); + + ExecutionException failure = assertThrows(ExecutionException.class, () -> response.get(10, SECONDS)); + assertInstanceOf(RemotelyClosedException.class, failure.getCause()); + assertEquals(1, handler.partCount.get(), "nothing held back may be delivered after the close"); + assertFalse(handler.corrupt.get()); + assertEquals(1, filtered.size()); + assertInstanceOf(ChannelClosedException.class, filtered.get(0)); + } + } + @Test public void cancellationWhileCompressedInputIsBufferedClosesTheConnection() throws Exception { AtomicReference clientChannel = new AtomicReference<>(); @@ -612,6 +660,20 @@ protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) ctx.writeAndFlush(last); break; } + case "/zeros-cut": + case "/zeros-cut-chunked": { + HttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK); + response.headers().set(HttpHeaderNames.CONTENT_ENCODING, HttpHeaderValues.GZIP); + if ("/zeros-cut".equals(request.uri())) { + HttpUtil.setContentLength(response, ZEROS_GZIP.length); + } else { + HttpUtil.setTransferEncodingChunked(response, true); + } + ctx.write(response); + ctx.writeAndFlush(new DefaultHttpContent(Unpooled.wrappedBuffer(ZEROS_GZIP, 0, ZEROS_GZIP.length / 2))) + .addListener(ChannelFutureListener.CLOSE); + break; + } case "/text-close": { HttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK); response.headers().set(HttpHeaderNames.CONTENT_ENCODING, HttpHeaderValues.GZIP); diff --git a/client/src/test/java/org/asynchttpclient/netty/handler/Http1ContentDecompressorSuspensionTest.java b/client/src/test/java/org/asynchttpclient/netty/handler/Http1ContentDecompressorSuspensionTest.java index 34c8c3f90..bc1e0379e 100644 --- a/client/src/test/java/org/asynchttpclient/netty/handler/Http1ContentDecompressorSuspensionTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/handler/Http1ContentDecompressorSuspensionTest.java @@ -23,6 +23,7 @@ import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.DecoderResult; import io.netty.handler.codec.compression.Brotli; import io.netty.handler.codec.compression.DecompressionException; import io.netty.handler.codec.compression.SnappyFrameEncoder; @@ -165,6 +166,7 @@ private static final class Body { int maxPart; boolean last; String trailer; + DecoderResult lastResult; Body drain(EmbeddedChannel channel) { Object msg; @@ -180,6 +182,7 @@ Body drain(EmbeddedChannel channel) { last = msg instanceof LastHttpContent; if (last) { trailer = ((LastHttpContent) msg).trailingHeaders().get("x-trailer"); + lastResult = ((LastHttpContent) msg).decoderResult(); } } ReferenceCountUtil.release(msg); @@ -230,6 +233,15 @@ private static final class SuspendedExchange { SuspendedExchange(Http1ContentDecompressor decompressor, String contentEncoding, byte[] body, int fragmentSize, boolean suspendBeforeBody, boolean autoRead, Input kind) { + this(decompressor, contentEncoding, body, fragmentSize, suspendBeforeBody, autoRead, kind, true); + } + + /** + * Without {@code ended}, every fragment is an {@link HttpContent}, so the body has not ended, as one that the + * connection cuts off. + */ + SuspendedExchange(Http1ContentDecompressor decompressor, String contentEncoding, byte[] body, + int fragmentSize, boolean suspendBeforeBody, boolean autoRead, Input kind, boolean ended) { suspendedBeforeBody = suspendBeforeBody; channel = new EmbeddedChannel(); channel.config().setAutoRead(autoRead); @@ -261,7 +273,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) { ByteBuf fragment = input(body, offset, length, kind); input.add(fragment); offset += length; - if (offset < body.length) { + if (offset < body.length || !ended) { channel.writeInbound(new DefaultHttpContent(fragment)); } else { LastHttpContent last = new DefaultLastHttpContent(fragment); @@ -454,6 +466,71 @@ public void channelInactive(ChannelHandlerContext ctx) { exchange.channel.finishAndReleaseAll(); } + /** + * A body that the connection cuts off fails anyway, so what is held back of it is dropped at the close instead of + * decompressed, even if decompressing it would have failed the body first. + */ + @ParameterizedTest + @ValueSource(strings = {"gzip", "deflate", "br", "zstd", "snappy"}) + public void inputHeldBackOfABodyThatTheCloseCutsOffIsDropped(String contentEncoding) { + // Decompressing what is held back would exceed the limit. + SuspendedExchange exchange = new SuspendedExchange(new Http1ContentDecompressor(false, 256 * 1024), + contentEncoding, compress(contentEncoding, new byte[1024 * 1024]), 4096, false, true, Input.HEAP, + false); + List events = new ArrayList<>(); + exchange.channel.pipeline().addLast(new ChannelInboundHandlerAdapter() { + @Override + public void channelInactive(ChannelHandlerContext ctx) { + events.add("inactive"); + } + }); + Body body = new Body().drain(exchange.channel); + assertEquals(1, body.parts); + + exchange.channel.pipeline().fireChannelInactive(); + body.drain(exchange.channel); + exchange.channel.checkException(); + assertEquals(1, body.parts, contentEncoding + ": nothing held back may be handed on"); + assertFalse(body.last); + assertEquals(List.of("inactive"), events); + exchange.assertInputReleased(); + assertFalse(exchange.channel.finishAndReleaseAll()); + } + + /** + * Once the end of the body has arrived, what is held back is handed on at the close. That includes the end of a body + * delimited by the connection, which arrives just before the close, and the failed end of a body whose last chunk + * could not be parsed, which still fails the exchange with the parser's error. + */ + @ParameterizedTest + @ValueSource(strings = {"delimited", "failed"}) + public void inputHeldBackOfABodyThatHasEndedIsHandedOnAtTheClose(String end) { + byte[] payload = concat(TEXT, new byte[400_000], TEXT); + for (String contentEncoding : List.of("gzip", "zstd")) { + SuspendedExchange exchange = new SuspendedExchange(new Http1ContentDecompressor(false, 0), + contentEncoding, compress(contentEncoding, payload), 4096, false, true, Input.HEAP, false); + Body body = new Body().drain(exchange.channel); + assertEquals(1, body.parts); + + LastHttpContent last = new DefaultLastHttpContent(); + if ("failed".equals(end)) { + last.setDecoderResult(DecoderResult.failure(new IllegalArgumentException("bad chunk"))); + } + exchange.channel.writeInbound(last); + body.drain(exchange.channel); + assertEquals(1, body.parts, contentEncoding + ": the end must wait behind the input held back"); + + exchange.channel.pipeline().fireChannelInactive(); + body.drain(exchange.channel); + exchange.channel.checkException(); + assertTrue(body.last, contentEncoding); + assertArrayEquals(payload, body.bytes.toByteArray(), contentEncoding); + assertEquals("failed".equals(end), body.lastResult.isFailure(), contentEncoding); + exchange.assertInputReleased(); + assertFalse(exchange.channel.finishAndReleaseAll()); + } + } + @Test public void removalReleasesHeldInput() { SuspendedExchange exchange = new SuspendedExchange(0);