From fc1c3690c33853627ebceef5b72fa3514da19a90 Mon Sep 17 00:00:00 2001 From: Matthias Kurz Date: Sat, 7 Nov 2026 19:02:07 +0100 Subject: [PATCH] Drop held input of a body cut off by the close When the connection closed while a suspended response held compressed input back, Http1ContentDecompressor decompressed all of it at once and handed it on. For a body that the close cut off, that work is wasted: the response cannot complete, and the request fails or is retried right after. The input held back is now dropped unless the end of the body has arrived, either held with the input or queued behind it. A body that has ended is handed on at the close as before: a held last chunk, the end of a body delimited by the connection, and a last chunk that failed to parse. A dropped body is not decompressed first, so it no longer fails on a corrupt or oversized remainder. The request is handled like any other whose connection closed before the end of its body: it is retried or fails with a RemotelyClosedException, and the IOExceptionFilters see a ChannelClosedException. Whether an orderly close by the server (a TCP FIN) is noticed while reads are suspended depends on the transport: native transports such as epoll notice it, NIO only once the input held back has been used up. A close on the client's side, for example when the client is closed, reaches the decompressor on every transport. Claude Code on behalf of Matthias Kurz Co-Authored-By: Claude Opus 5.5 --- .../asynchttpclient/ResponseBodyControl.java | 9 ++- .../handler/Http1ContentDecompressor.java | 25 ++++-- .../Http1DecompressionSuspensionTest.java | 62 +++++++++++++++ ...ttp1ContentDecompressorSuspensionTest.java | 79 ++++++++++++++++++- 4 files changed, 166 insertions(+), 9 deletions(-) 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);