From 6289389f6683304af8c4c1e7e0e310991c370738 Mon Sep 17 00:00:00 2001 From: Bryce Anderson Date: Mon, 5 Oct 2026 13:53:13 -0600 Subject: [PATCH] http-netty: reset cancelled HTTP/2 streams whose request body is still writing #### Motivation An HTTP/2 client request cancelled after its response finished but before its request body did was never reset. A gRPC client does this without any user cancel when a server rejects a client-streaming or bidirectional call with a Trailers-Only response while the request stream is still open. The server kept waiting for the rest of the body, and the stream held a concurrent-stream slot on both peers until the server reset it or the connection closed. Enough of these made the client open extra connections. The client's close-on-cancel skipped HTTP/2 because it once ran on the parent connection, where closing failed every stream (#1307), but it now runs on each stream. #### Modifications - The client's close-on-cancel now applies to HTTP/2 too, where it closes only the cancelled stream. #### Result Cancelling an HTTP/2 request whose request body is still streaming resets its stream. Behavior change: an HTTP/2 server now receives RST_STREAM(CANCEL) for these streams, including after it has sent its full response. A server that counts such resets against a rapid-reset limit, as the ServiceTalk server does, can reach that limit sooner, for example when it rejects many streaming gRPC calls on one connection. If that happens, raise the server's limit on received resets. --- .../grpc/netty/TrailersOnlyErrorTest.java | 57 ++- .../AbstractStreamingHttpConnection.java | 21 +- .../PartiallyWrittenRequestCancelTest.java | 469 ++++++++++++++++++ .../netty/TimeoutHttpRequesterFilterTest.java | 38 +- 4 files changed, 572 insertions(+), 13 deletions(-) create mode 100644 servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/PartiallyWrittenRequestCancelTest.java diff --git a/servicetalk-grpc-netty/src/test/java/io/servicetalk/grpc/netty/TrailersOnlyErrorTest.java b/servicetalk-grpc-netty/src/test/java/io/servicetalk/grpc/netty/TrailersOnlyErrorTest.java index 58d94bb2fd..88270bfcfd 100644 --- a/servicetalk-grpc-netty/src/test/java/io/servicetalk/grpc/netty/TrailersOnlyErrorTest.java +++ b/servicetalk-grpc-netty/src/test/java/io/servicetalk/grpc/netty/TrailersOnlyErrorTest.java @@ -1,5 +1,5 @@ /* - * Copyright © 2021 Apple Inc. and the ServiceTalk project authors + * Copyright © 2021-2026 Apple Inc. and the ServiceTalk project authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -15,6 +15,7 @@ */ package io.servicetalk.grpc.netty; +import io.servicetalk.client.api.TransportObserverConnectionFactoryFilter; import io.servicetalk.concurrent.api.Single; import io.servicetalk.grpc.api.GrpcClientBuilder; import io.servicetalk.grpc.api.GrpcServerBuilder; @@ -38,8 +39,13 @@ import io.servicetalk.http.api.StreamingHttpResponseFactory; import io.servicetalk.http.api.StreamingHttpServiceFilter; import io.servicetalk.http.utils.BeforeFinallyHttpOperator; +import io.servicetalk.transport.api.ConnectionInfo; +import io.servicetalk.transport.api.ConnectionObserver; +import io.servicetalk.transport.api.ConnectionObserver.MultiplexedObserver; +import io.servicetalk.transport.api.ConnectionObserver.StreamObserver; import io.servicetalk.transport.api.HostAndPort; import io.servicetalk.transport.api.ServerContext; +import io.servicetalk.transport.api.TransportObserver; import io.servicetalk.transport.netty.internal.ExecutionContextExtension; import io.grpc.examples.helloworld.Greeter; @@ -60,6 +66,7 @@ import static io.servicetalk.concurrent.api.Publisher.from; import static io.servicetalk.concurrent.api.Publisher.never; import static io.servicetalk.concurrent.internal.DeliberateException.DELIBERATE_EXCEPTION; +import static io.servicetalk.concurrent.internal.TestTimeoutConstants.DEFAULT_TIMEOUT_SECONDS; import static io.servicetalk.grpc.api.GrpcHeaderNames.GRPC_STATUS; import static io.servicetalk.grpc.api.GrpcStatusCode.UNIMPLEMENTED; import static io.servicetalk.grpc.api.GrpcStatusCode.UNKNOWN; @@ -67,12 +74,14 @@ import static io.servicetalk.transport.netty.internal.AddressUtils.localAddress; import static io.servicetalk.transport.netty.internal.AddressUtils.serverHostAndPort; import static java.util.Collections.singleton; +import static java.util.concurrent.TimeUnit.SECONDS; import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.instanceOf; import static org.hamcrest.Matchers.is; import static org.hamcrest.Matchers.notNullValue; 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 static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -144,6 +153,25 @@ void testServiceThrows() throws Exception { responseLatch.await(); // Make sure all responses complete } + @Test + void serviceThrowsWithOpenRequestStreamResetsStream() throws Exception { + final CountDownLatch streamClosed = new CountDownLatch(1); + final TesterService service = mockTesterService(); + setupServiceThrows(service); + + try (ServerContext serverContext = GrpcServers.forAddress(localAddress(0)) + .initializeHttp(TrailersOnlyErrorTest::applyCtx) + .listenAndAwait(new Tester.ServiceFactory(service)); + TesterClient client = GrpcClients.forAddress(serverHostAndPort(serverContext)) + .initializeHttp(builder -> applyCtx(builder).appendConnectionFactoryFilter( + new TransportObserverConnectionFactoryFilter<>(onStreamClosed(streamClosed)))) + .build(new Tester.ClientFactory())) { + // The request stream never completes, so only the client can end the stream after the Trailers-Only error. + verifyException(client.testBiDiStream(never()).toFuture(), UNKNOWN); + assertTrue(streamClosed.await(DEFAULT_TIMEOUT_SECONDS, SECONDS)); + } + } + @Test void testServiceThrowsBlockingClient() throws Exception { final BlockingQueue asyncErrors = new LinkedBlockingDeque<>(); @@ -260,6 +288,33 @@ private static void verifyException(final Executable exchange, final GrpcStatusC assertThat(e.status().code(), is(expectedCode)); } + private static TransportObserver onStreamClosed(final CountDownLatch latch) { + final StreamObserver streamObserver = new StreamObserver() { + @Override + public void streamClosed(final Throwable error) { + latch.countDown(); + } + + @Override + public void streamClosed() { + latch.countDown(); + } + }; + final MultiplexedObserver multiplexedObserver = new MultiplexedObserver() { + @Override + public StreamObserver onNewStream() { + return streamObserver; + } + }; + final ConnectionObserver connectionObserver = new ConnectionObserver() { + @Override + public MultiplexedObserver multiplexedConnectionEstablished(final ConnectionInfo info) { + return multiplexedObserver; + } + }; + return (localAddress, remoteAddress) -> connectionObserver; + } + private static TesterService mockTesterService() { TesterService service = mock(TesterService.class); when(service.closeAsync()).thenReturn(completed()); diff --git a/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/AbstractStreamingHttpConnection.java b/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/AbstractStreamingHttpConnection.java index 35d8e18869..a980fe356d 100644 --- a/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/AbstractStreamingHttpConnection.java +++ b/servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/AbstractStreamingHttpConnection.java @@ -1,5 +1,5 @@ /* - * Copyright © 2018-2019, 2021-2022 Apple Inc. and the ServiceTalk project authors + * Copyright © 2018-2026 Apple Inc. and the ServiceTalk project authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -158,16 +158,15 @@ public void cancel() { // seen by the transport before cancel. We have no way of knowing at this layer if this indeed // happen. Therefore, we close the connection manually to guarantee closure. // - // For H2 and above, connection are multiplexed and use virtual streams for each - // request-response exchange. At the time users own a Cancellable, the stream already owns - // OnStreamClosedRunnable in H2ClientParentConnectionContext. It will update the concurrency - // controller state if cancellation results in stream closure instead of completion. - if (connectionContext().protocol().major() < 2) { - LOGGER.debug("{} {} request was cancelled before receiving the full response, " + - "closing this {} connection to stop receiving more data", - connectionContext, requestMetaData, connectionContext.protocol()); - closeAsync().subscribe(); - } + // For H2 and above, this connection is a single stream, so closing it resets only this + // exchange. The transport resets a stream whose response is unread, but not one whose response + // completed while its request body was still being written: that stream would stay open and + // hold a concurrent-stream slot on both peers. OnStreamClosedRunnable in + // H2ClientParentConnectionContext updates the concurrency controller when the stream closes. + LOGGER.debug("{} {} request was cancelled before it completed, closing this {} {}", + connectionContext, requestMetaData, connectionContext.protocol(), + connectionContext.protocol().major() < 2 ? "connection" : "stream"); + closeAsync().subscribe(); } }) .firstAndTail(this::newSplicedResponse); diff --git a/servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/PartiallyWrittenRequestCancelTest.java b/servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/PartiallyWrittenRequestCancelTest.java new file mode 100644 index 0000000000..7e8262465c --- /dev/null +++ b/servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/PartiallyWrittenRequestCancelTest.java @@ -0,0 +1,469 @@ +/* + * Copyright © 2026 Apple Inc. and the ServiceTalk project authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.servicetalk.http.netty; + +import io.servicetalk.buffer.api.Buffer; +import io.servicetalk.client.api.TransportObserverConnectionFactoryFilter; +import io.servicetalk.concurrent.BlockingIterator; +import io.servicetalk.concurrent.api.Publisher; +import io.servicetalk.concurrent.api.Single; +import io.servicetalk.http.api.BlockingStreamingHttpClient; +import io.servicetalk.http.api.BlockingStreamingHttpResponse; +import io.servicetalk.http.api.FilterableStreamingHttpConnection; +import io.servicetalk.http.api.Http2Exception; +import io.servicetalk.http.api.Http2SettingsBuilder; +import io.servicetalk.http.api.HttpExecutionStrategy; +import io.servicetalk.http.api.ReservedStreamingHttpConnection; +import io.servicetalk.http.api.SingleAddressHttpClientBuilder; +import io.servicetalk.http.api.StreamingHttpClient; +import io.servicetalk.http.api.StreamingHttpConnectionFilter; +import io.servicetalk.http.api.StreamingHttpRequest; +import io.servicetalk.http.api.StreamingHttpResponse; +import io.servicetalk.transport.api.ConnectionInfo; +import io.servicetalk.transport.api.ConnectionObserver; +import io.servicetalk.transport.api.ConnectionObserver.DataObserver; +import io.servicetalk.transport.api.ConnectionObserver.MultiplexedObserver; +import io.servicetalk.transport.api.ConnectionObserver.ReadObserver; +import io.servicetalk.transport.api.ConnectionObserver.StreamObserver; +import io.servicetalk.transport.api.HostAndPort; +import io.servicetalk.transport.api.ServerContext; +import io.servicetalk.transport.api.TransportObserver; + +import org.junit.jupiter.api.Named; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; +import org.junit.jupiter.params.provider.ValueSource; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.net.SocketAddress; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Future; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Stream; +import javax.annotation.Nullable; + +import static io.servicetalk.concurrent.api.Completable.never; +import static io.servicetalk.concurrent.api.Single.succeeded; +import static io.servicetalk.concurrent.internal.TestTimeoutConstants.DEFAULT_TIMEOUT_SECONDS; +import static io.servicetalk.http.api.Http2ErrorCode.CANCEL; +import static io.servicetalk.http.api.HttpExecutionStrategies.defaultStrategy; +import static io.servicetalk.http.api.HttpExecutionStrategies.offloadAll; +import static io.servicetalk.http.api.HttpExecutionStrategies.offloadNone; +import static io.servicetalk.http.api.HttpResponseStatus.OK; +import static io.servicetalk.http.netty.HttpProtocolConfigs.h1; +import static io.servicetalk.http.netty.HttpProtocolConfigs.h2; +import static io.servicetalk.http.netty.HttpProtocolConfigs.h2Default; +import static io.servicetalk.transport.netty.internal.AddressUtils.localAddress; +import static io.servicetalk.transport.netty.internal.AddressUtils.serverHostAndPort; +import static java.net.InetAddress.getLoopbackAddress; +import static java.nio.charset.StandardCharsets.US_ASCII; +import static java.util.Collections.singletonList; +import static java.util.Objects.requireNonNull; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.instanceOf; +import static org.hamcrest.Matchers.is; +import static org.hamcrest.Matchers.not; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Named.named; + +class PartiallyWrittenRequestCancelTest { + + private static final String PARTIAL_BODY = "partial"; + private static final int MAX_CONCURRENT_STREAMS = 2; + + @ParameterizedTest(name = "{displayName} [{index}] {0}") + @MethodSource("strategies") + void cancelAfterEarlyH2ResponseWhileBodyIsWritingResetsStream(HttpExecutionStrategy strategy) throws Exception { + CountDownLatch responseRead = new CountDownLatch(1); + CountDownLatch streamClosed = new CountDownLatch(1); + CompletableFuture serverRequestBodyError = new CompletableFuture<>(); + try (ServerContext server = HttpServers.forAddress(localAddress(0)).protocols(h2Default()) + .listenStreamingAndAwait((ctx, request, factory) -> { + request.messageBody().ignoreElements().whenOnError(serverRequestBodyError::complete).subscribe(); + return succeeded(factory.ok()); + }); + StreamingHttpClient client = HttpClients.forSingleAddress(serverHostAndPort(server)) + .protocols(h2Default()) + .executionStrategy(strategy) + .appendConnectionFactoryFilter(new TransportObserverConnectionFactoryFilter<>( + observer(() -> { }, responseRead::countDown, streamClosed::countDown))) + .buildStreaming()) { + StreamingHttpResponse response = client.request(client.post("/") + .payloadBody(neverEndingBody(client))).toFuture().get(); + // The transport has read the whole response, so only the write side can reset the stream. + assertTrue(responseRead.await(DEFAULT_TIMEOUT_SECONDS, SECONDS)); + + assertTrue(response.messageBody().ignoreElements().toFuture().cancel(true)); + assertTrue(streamClosed.await(DEFAULT_TIMEOUT_SECONDS, SECONDS)); + Throwable cause = serverRequestBodyError.get(); + assertThat(cause, instanceOf(Http2Exception.class)); + assertThat(((Http2Exception) cause).errorCode(), is(CANCEL)); + } + } + + @ParameterizedTest(name = "{displayName} [{index}] requestReplaced={0}") + @ValueSource(booleans = {false, true}) + void cancelAfterEarlyH2ResponseWhileBodyIsWritingFreesStreamSlot(boolean requestReplaced) throws Exception { + Semaphore responsesRead = new Semaphore(0); + AtomicInteger connections = new AtomicInteger(); + try (ServerContext server = HttpServers.forAddress(localAddress(0)) + .protocols(h2().initialSettings(new Http2SettingsBuilder() + .maxConcurrentStreams(MAX_CONCURRENT_STREAMS).build()).build()) + .listenStreamingAndAwait((ctx, request, factory) -> succeeded(factory.ok()))) { + SingleAddressHttpClientBuilder builder = + HttpClients.forSingleAddress(serverHostAndPort(server)) + .protocols(h2Default()) + .appendConnectionFactoryFilter(new TransportObserverConnectionFactoryFilter<>( + observer(connections::incrementAndGet, responsesRead::release, () -> { }))); + if (requestReplaced) { + // A new request lacks the original's context, so the client frees its stream slot on cancel, not on + // stream close, and a stream left open counts only against Netty's limit. + builder.appendConnectionFilter(ReplaceRequestFilter::new); + } + try (StreamingHttpClient client = builder.buildStreaming()) { + for (int i = 0; i < MAX_CONCURRENT_STREAMS; i++) { + StreamingHttpResponse response = client.request(client.post("/") + .payloadBody(neverEndingBody(client))).toFuture().get(); + assertTrue(responsesRead.tryAcquire(DEFAULT_TIMEOUT_SECONDS, SECONDS)); + assertTrue(response.messageBody().ignoreElements().toFuture().cancel(true)); + } + + StreamingHttpResponse response = client.request(client.get("/")).toFuture().get(); + assertThat(response.status(), is(OK)); + response.messageBody().ignoreElements().toFuture().get(); + assertThat(connections.get(), is(1)); + } + } + } + + @Test + void cancelAfterEarlyH2ResponseOnReservedConnectionResetsOnlyTheStream() throws Exception { + Semaphore responsesRead = new Semaphore(0); + CompletableFuture serverRequestBodyError = new CompletableFuture<>(); + try (ServerContext server = HttpServers.forAddress(localAddress(0)).protocols(h2Default()) + .listenStreamingAndAwait((ctx, request, factory) -> { + request.messageBody().ignoreElements().whenOnError(serverRequestBodyError::complete).subscribe(); + return succeeded(factory.ok()); + }); + StreamingHttpClient client = HttpClients.forSingleAddress(serverHostAndPort(server)) + .protocols(h2Default()) + .appendConnectionFactoryFilter(new TransportObserverConnectionFactoryFilter<>( + observer(() -> { }, responsesRead::release, () -> { }))) + .buildStreaming()) { + ReservedStreamingHttpConnection connection = + client.reserveConnection(client.get("/")).toFuture().get(); + try { + StreamingHttpResponse response = connection.request(connection.post("/") + .payloadBody(neverEndingBody(client))).toFuture().get(); + assertTrue(responsesRead.tryAcquire(DEFAULT_TIMEOUT_SECONDS, SECONDS)); + + assertTrue(response.messageBody().ignoreElements().toFuture().cancel(true)); + Throwable cause = serverRequestBodyError.get(); + assertThat(cause, instanceOf(Http2Exception.class)); + assertThat(((Http2Exception) cause).errorCode(), is(CANCEL)); + + StreamingHttpResponse next = connection.request(connection.get("/")).toFuture().get(); + assertThat(next.status(), is(OK)); + next.messageBody().ignoreElements().toFuture().get(); + } finally { + connection.releaseAsync().toFuture().get(); + } + } + } + + @Test + void closingDrainedH2ResponseIteratorKeepsConnection() throws Exception { + AtomicInteger connections = new AtomicInteger(); + try (ServerContext server = HttpServers.forAddress(localAddress(0)).protocols(h2Default()) + .listenStreamingAndAwait((ctx, request, factory) -> request.messageBody().ignoreElements() + .concat(succeeded(factory.ok().payloadBody(Publisher.from( + ctx.executionContext().bufferAllocator().fromAscii(PARTIAL_BODY)))))); + BlockingStreamingHttpClient client = HttpClients.forSingleAddress(serverHostAndPort(server)) + .protocols(h2Default()) + .appendConnectionFactoryFilter(new TransportObserverConnectionFactoryFilter<>( + observer(connections::incrementAndGet, () -> { }, () -> { }))) + .buildBlockingStreaming()) { + for (int i = 0; i < 2; i++) { + BlockingStreamingHttpResponse response = client.request(client.post("/").payloadBody( + singletonList(client.executionContext().bufferAllocator().fromAscii(PARTIAL_BODY)))); + // Closing the iterator always cancels, here after the exchange has already completed. + try (BlockingIterator iterator = response.payloadBody().iterator()) { + while (iterator.hasNext()) { + iterator.next(); + } + } + } + assertThat(connections.get(), is(1)); + } + } + + @Test + void cancelWhileH2ResponseIsStreamingResetsOnlyTheStream() throws Exception { + CountDownLatch serverResponseCancelled = new CountDownLatch(1); + AtomicInteger connections = new AtomicInteger(); + try (ServerContext server = HttpServers.forAddress(localAddress(0)).protocols(h2Default()) + .listenStreamingAndAwait((ctx, request, factory) -> succeeded("/streaming".equals(request.path()) ? + factory.ok().payloadBody(Publisher.from( + ctx.executionContext().bufferAllocator().fromAscii(PARTIAL_BODY)) + .concat(never()).whenCancel(serverResponseCancelled::countDown)) : + factory.ok())); + StreamingHttpClient client = HttpClients.forSingleAddress(serverHostAndPort(server)) + .protocols(h2Default()) + .appendConnectionFactoryFilter(new TransportObserverConnectionFactoryFilter<>( + observer(connections::incrementAndGet, () -> { }, () -> { }))) + .buildStreaming()) { + StreamingHttpResponse response = client.request(client.get("/streaming")).toFuture().get(); + // Cancels after the first chunk. + response.payloadBody().takeAtMost(1).ignoreElements().toFuture().get(); + assertTrue(serverResponseCancelled.await(DEFAULT_TIMEOUT_SECONDS, SECONDS)); + + StreamingHttpResponse next = client.request(client.get("/")).toFuture().get(); + assertThat(next.status(), is(OK)); + next.messageBody().ignoreElements().toFuture().get(); + assertThat(connections.get(), is(1)); + } + } + + @Test + void cancelMidH1BodyDoesNotLetQueuedRequestEndTruncatedBody() throws Exception { + try (RawServer server = new RawServer(); + StreamingHttpClient client = HttpClients.forSingleAddress(serverHostAndPort(server.address())) + .protocols(h1().maxPipelinedRequests(3).build()) + // Subscribes on the caller thread, so C is queued behind B's write before B is cancelled. + .executionStrategy(offloadNone()) + .buildStreaming()) { + ReservedStreamingHttpConnection connection = + client.reserveConnection(client.get("/reserve")).toFuture().get(); + try { + connection.request(connection.get("/a")).toFuture(); + server.awaitReceived("GET /a"); + Future b = connection.request(connection.post("/b") + .payloadBody(neverEndingBody(client))).toFuture(); + server.awaitReceived(PARTIAL_BODY); + connection.request(connection.get("/c")).toFuture(); + + assertTrue(b.cancel(true)); + // Encoding C's headers ends B's chunked body, so the server would take the truncated B as complete. + String received = server.awaitClosedOrReceived("GET /c"); + assertThat(received, not(containsString("GET /c"))); + assertThat(server.isClosed(), is(true)); + } finally { + connection.closeAsync().toFuture().get(); + } + } + } + + @Test + void cancelAfterEarlyH1ResponseWhileBodyIsWritingClosesConnection() throws Exception { + CountDownLatch responseRead = new CountDownLatch(1); + try (RawServer server = new RawServer(); + StreamingHttpClient client = HttpClients.forSingleAddress(serverHostAndPort(server.address())) + .appendConnectionFactoryFilter(new TransportObserverConnectionFactoryFilter<>( + observer(() -> { }, responseRead::countDown, () -> { }))) + .buildStreaming()) { + ReservedStreamingHttpConnection connection = + client.reserveConnection(client.get("/reserve")).toFuture().get(); + try { + Future responseFuture = connection.request(connection.post("/b") + .payloadBody(neverEndingBody(client))).toFuture(); + server.awaitReceived(PARTIAL_BODY); + server.write("HTTP/1.1 200 OK\r\ncontent-length: 0\r\n\r\n"); + StreamingHttpResponse response = responseFuture.get(); + // The transport has read the whole response, so it ignores the cancel below. + assertTrue(responseRead.await(DEFAULT_TIMEOUT_SECONDS, SECONDS)); + + assertTrue(response.messageBody().ignoreElements().toFuture().cancel(true)); + connection.onClose().toFuture().get(); + } finally { + connection.closeAsync().toFuture().get(); + } + } + } + + private static final class ReplaceRequestFilter extends StreamingHttpConnectionFilter { + ReplaceRequestFilter(FilterableStreamingHttpConnection delegate) { + super(delegate); + } + + @Override + public Single request(StreamingHttpRequest request) { + StreamingHttpRequest replacement = delegate().newRequest(request.method(), request.requestTarget()) + .payloadBody(request.payloadBody()); + replacement.headers().add(request.headers()); + return delegate().request(replacement); + } + } + + private static Stream> strategies() { + return Stream.of(named("offloadNone", offloadNone()), named("defaultStrategy", defaultStrategy()), + named("offloadAll", offloadAll())); + } + + private static Publisher neverEndingBody(StreamingHttpClient client) { + return Publisher.from(client.executionContext().bufferAllocator().fromAscii(PARTIAL_BODY)).concat(never()); + } + + private static TransportObserver observer(Runnable onConnection, Runnable onReadComplete, + Runnable onStreamClosed) { + ReadObserver readObserver = new ReadObserver() { + @Override + public void readComplete() { + onReadComplete.run(); + } + }; + DataObserver dataObserver = new DataObserver() { + @Override + public ReadObserver onNewRead() { + return readObserver; + } + }; + StreamObserver streamObserver = new StreamObserver() { + @Override + public DataObserver streamEstablished() { + return dataObserver; + } + + @Override + public void streamClosed(Throwable error) { + onStreamClosed.run(); + } + + @Override + public void streamClosed() { + onStreamClosed.run(); + } + }; + MultiplexedObserver multiplexedObserver = new MultiplexedObserver() { + @Override + public StreamObserver onNewStream() { + return streamObserver; + } + }; + ConnectionObserver connectionObserver = new ConnectionObserver() { + @Override + public DataObserver connectionEstablished(ConnectionInfo info) { + onConnection.run(); + return dataObserver; + } + + @Override + public MultiplexedObserver multiplexedConnectionEstablished(ConnectionInfo info) { + onConnection.run(); + return multiplexedObserver; + } + }; + return (localAddress, remoteAddress) -> connectionObserver; + } + + /** + * Accepts one connection, records every byte it receives, and writes only what the test tells it to. + */ + private static final class RawServer implements AutoCloseable { + private final ServerSocket serverSocket; + private final Thread acceptor; + private final StringBuilder received = new StringBuilder(); + @Nullable + private Socket socket; + private boolean closed; + + RawServer() throws IOException { + serverSocket = new ServerSocket(0, 50, getLoopbackAddress()); + acceptor = new Thread(this::readConnection, "raw-server"); + acceptor.start(); + } + + SocketAddress address() { + return serverSocket.getLocalSocketAddress(); + } + + private void readConnection() { + try (Socket accepted = serverSocket.accept()) { + synchronized (this) { + socket = accepted; + } + InputStream in = accepted.getInputStream(); + byte[] buffer = new byte[1024]; + int read; + while ((read = in.read(buffer)) >= 0) { + synchronized (this) { + received.append(new String(buffer, 0, read, US_ASCII)); + notifyAll(); + } + } + } catch (IOException closedOrReset) { + // Either way the connection is over. + } finally { + synchronized (this) { + closed = true; + notifyAll(); + } + } + } + + synchronized void write(String data) throws IOException { + // Called only once the server has received data, so the connection has been accepted. + OutputStream out = requireNonNull(socket).getOutputStream(); + out.write(data.getBytes(US_ASCII)); + out.flush(); + } + + synchronized void awaitReceived(String data) throws Exception { + awaitReceived(data, false); + } + + synchronized String awaitClosedOrReceived(String data) throws Exception { + awaitReceived(data, true); + return received.toString(); + } + + synchronized boolean isClosed() { + return closed; + } + + private void awaitReceived(String data, boolean orClosed) throws Exception { + final long deadline = System.nanoTime() + SECONDS.toNanos(DEFAULT_TIMEOUT_SECONDS); + while (received.indexOf(data) < 0 && !(orClosed && closed)) { + final long remainingMillis = (deadline - System.nanoTime()) / 1_000_000; + if (remainingMillis <= 0) { + throw new TimeoutException("Did not receive " + data + "; received so far: " + received); + } + wait(remainingMillis); + } + } + + @Override + public void close() throws Exception { + serverSocket.close(); + synchronized (this) { + if (socket != null) { + socket.close(); + } + } + acceptor.join(); + } + } +} diff --git a/servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/TimeoutHttpRequesterFilterTest.java b/servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/TimeoutHttpRequesterFilterTest.java index 3c17adcfd5..eaa4bc0a9c 100644 --- a/servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/TimeoutHttpRequesterFilterTest.java +++ b/servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/TimeoutHttpRequesterFilterTest.java @@ -1,5 +1,5 @@ /* - * Copyright © 2022 Apple Inc. and the ServiceTalk project authors + * Copyright © 2022-2026 Apple Inc. and the ServiceTalk project authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,6 +17,7 @@ import io.servicetalk.concurrent.api.Single; import io.servicetalk.concurrent.api.TerminalSignalConsumer; +import io.servicetalk.http.api.ReservedStreamingHttpConnection; import io.servicetalk.http.api.StreamingHttpClient; import io.servicetalk.http.api.StreamingHttpConnection; import io.servicetalk.http.api.StreamingHttpConnectionFilter; @@ -31,10 +32,15 @@ import java.time.Duration; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import static io.servicetalk.http.api.HttpResponseStatus.OK; import static io.servicetalk.http.netty.AbstractNettyHttpServerTest.ExecutorSupplier.CACHED; import static io.servicetalk.http.netty.AbstractNettyHttpServerTest.ExecutorSupplier.CACHED_SERVER; +import static io.servicetalk.http.netty.HttpProtocol.HTTP_2; +import static io.servicetalk.http.netty.TestServiceStreaming.SVC_ECHO; import static io.servicetalk.http.netty.TestServiceStreaming.SVC_NEVER; import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.instanceOf; @@ -94,6 +100,36 @@ public void cancel() { } } + @Test + void h2TimeoutResetsStreamWithoutClosingConnection() throws Exception { + final AtomicInteger connections = new AtomicInteger(); + final AtomicBoolean connectionClosing = new AtomicBoolean(); + protocol(HTTP_2.config); + connectionFilterFactory(appendConnectionFilter(c -> { + connections.incrementAndGet(); + c.connectionContext().onClosing().subscribe(() -> connectionClosing.set(true)); + return new StreamingHttpConnectionFilter(c) { }; + }, + new TimeoutHttpRequesterFilter(Duration.ofMillis(100)))); + setUp(CACHED, CACHED_SERVER); + // Release the reserved connection so the client reuses it instead of opening another. + ((ReservedStreamingHttpConnection) streamingHttpConnection()).releaseAsync().toFuture().get(); + final int connectionsBefore = connections.get(); + + StreamingHttpClient client = streamingHttpClient(); + for (int i = 0; i < 15; i++) { + String msg = "Unexpected assertion on iteration #" + i; + ExecutionException ee = assertThrows(ExecutionException.class, + () -> client.request(client.get(SVC_NEVER)).toFuture().get(), msg); + assertThat(msg, ee.getCause(), is(instanceOf(TimeoutException.class))); + StreamingHttpResponse response = client.request(client.get(SVC_ECHO)).toFuture().get(); + assertThat(msg, response.status(), is(OK)); + response.messageBody().ignoreElements().toFuture().get(); + } + assertThat(connectionClosing.get(), is(false)); + assertThat(connections.get(), is(connectionsBefore)); + } + private static StreamingHttpConnectionFilterFactory appendConnectionFilter( final StreamingHttpConnectionFilterFactory first, final StreamingHttpConnectionFilterFactory next) {