Skip to content

Drop held input of a body cut off by the close #2386

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mkurz wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Drop held input of a body cut off by the close #2386

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 6 additions & 3 deletions client/src/main/java/org/asynchttpclient/ResponseBodyControl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
* <p>
* 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.
Expand Down
25 changes: 20 additions & 5 deletions client/src/main/java/org/asynchttpclient/netty/handler/Http1ContentDecompressor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
* <p>
* {@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
Expand Down Expand Up @@ -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);
}
Expand Down Expand Up @@ -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);
Expand Down
62 changes: 62 additions & 0 deletions client/src/test/java/org/asynchttpclient/Http1DecompressionSuspensionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand All @@ -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;
Expand All @@ -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;
Expand All @@ -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
Expand Down Expand Up @@ -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<IOException> filtered = new CopyOnWriteArrayList<>();
IOExceptionFilter filter = new IOExceptionFilter() {
@Override
public <T> FilterContext<T> filter(FilterContext<T> 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<Void> 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<Channel> clientChannel = new AtomicReference<>();
Expand Down Expand Up @@ -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);
Expand Down
79 changes: 78 additions & 1 deletion ...c/test/java/org/asynchttpclient/netty/handler/Http1ContentDecompressorSuspensionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -165,6 +166,7 @@ private static final class Body {
int maxPart;
boolean last;
String trailer;
DecoderResult lastResult;

Body drain(EmbeddedChannel channel) {
Object msg;
Expand All @@ -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);
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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<Object> 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);
Expand Down
Loading