From fe7089c242f12b7a75675ebe9a55d4690e9c88de Mon Sep 17 00:00:00 2001 From: Bryce Anderson Date: Thu, 1 Oct 2026 09:40:22 -0600 Subject: [PATCH] transport-netty-internal: don't leave cancelled, partially written client requests open #### Motivation Cancelling a client write cancelled only the request body's source, so the transport never ended a partially written request. On HTTP/2, a stream cancelled after its response finished but before its request body did was never reset: 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. #### Modifications - A client write cancelled after part of the request was written now closes the outbound side: HTTP/2 resets the stream, and HTTP/1.x shuts down output. Writes that finished or never started, and server writes, are unaffected. #### Result A cancelled client request no longer stays open after it was partially written. Behavior change: an HTTP/2 server now receives RST_STREAM(CANCEL) for a client stream cancelled while its request body is still being written, including when the response has already finished. No action is needed. --- .../PartiallyWrittenRequestCancelTest.java | 196 ++++++++++++++++++ .../netty/internal/WriteStreamSubscriber.java | 31 ++- .../internal/WriteStreamSubscriberTest.java | 54 ++++- 3 files changed, 278 insertions(+), 3 deletions(-) create mode 100644 servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/PartiallyWrittenRequestCancelTest.java 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..184f4b7a3b --- /dev/null +++ b/servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/PartiallyWrittenRequestCancelTest.java @@ -0,0 +1,196 @@ +/* + * 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.api.Publisher; +import io.servicetalk.concurrent.api.Single; +import io.servicetalk.http.api.FilterableStreamingHttpConnection; +import io.servicetalk.http.api.Http2Exception; +import io.servicetalk.http.api.Http2SettingsBuilder; +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.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.net.InetSocketAddress; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Semaphore; +import java.util.concurrent.atomic.AtomicInteger; + +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.HttpResponseStatus.OK; +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.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.junit.jupiter.api.Assertions.assertTrue; + +class PartiallyWrittenRequestCancelTest { + + private static final String PARTIAL_BODY = "partial"; + private static final int MAX_CONCURRENT_STREAMS = 2; + + @Test + void cancelAfterEarlyH2ResponseWhileBodyIsWritingResetsStream() 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()) + .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)); + } + } + } + + 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 Publisher neverEndingBody(StreamingHttpClient client) { + return Publisher.from(client.executionContext().bufferAllocator().fromAscii(PARTIAL_BODY)).concat(never()); + } + + private static TransportObserver observer(Runnable onConnection, Runnable onStreamReadComplete, + Runnable onStreamClosed) { + ReadObserver readObserver = new ReadObserver() { + @Override + public void readComplete() { + onStreamReadComplete.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 MultiplexedObserver multiplexedConnectionEstablished(ConnectionInfo info) { + onConnection.run(); + return multiplexedObserver; + } + }; + return (localAddress, remoteAddress) -> connectionObserver; + } +} diff --git a/servicetalk-transport-netty-internal/src/main/java/io/servicetalk/transport/netty/internal/WriteStreamSubscriber.java b/servicetalk-transport-netty-internal/src/main/java/io/servicetalk/transport/netty/internal/WriteStreamSubscriber.java index a14df7f524..abfbae4b0a 100644 --- a/servicetalk-transport-netty-internal/src/main/java/io/servicetalk/transport/netty/internal/WriteStreamSubscriber.java +++ b/servicetalk-transport-netty-internal/src/main/java/io/servicetalk/transport/netty/internal/WriteStreamSubscriber.java @@ -1,5 +1,5 @@ /* - * Copyright © 2018, 2020 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. @@ -86,6 +86,7 @@ final class WriteStreamSubscriber implements PublisherSource.Subscriber, private static final byte CLOSE_OUTBOUND_ON_SUBSCRIBER_TERMINATION = 1 << 2; private static final byte SUBSCRIBER_TERMINATED = 1 << 3; private static final byte SOURCE_OUTBOUND_CLOSED = 1 << 4; + private static final byte SOURCE_CANCELLED = 1 << 5; private static final byte SUBSCRIBER_OR_SOURCE_TERMINATED = SOURCE_TERMINATED | SUBSCRIBER_TERMINATED; private static final Subscription CANCELLED = newEmptySubscription(); private static final AtomicReferenceFieldUpdater subscriptionUpdater = @@ -138,7 +139,10 @@ public void onSubscribe(Subscription s) { s.cancel(); return; } - subscriber.onSubscribe(concurrentSubscription); + subscriber.onSubscribe(isClient ? () -> { + sourceCancelled(); + concurrentSubscription.cancel(); + } : concurrentSubscription); if (eventLoop.inEventLoop()) { initialRequestN(concurrentSubscription); } else { @@ -295,6 +299,14 @@ private void close0(@Nullable Subscription oldVal, Throwable closedException, bo promise.close(closedException, closeOutboundIfIdle); } + private void sourceCancelled() { + if (eventLoop.inEventLoop()) { + promise.sourceCancelled(); + } else { + eventLoop.execute(promise::sourceCancelled); + } + } + void cancel() { // Visible only for tests. // In order to prevent concurrent access to the subscription, we use the EventLoop. The alternative would be // some additional protection around calling subscription.request and subscription.cancel, but since this method @@ -434,6 +446,21 @@ void outboundClosed() { markCancelled(); } + void sourceCancelled() { + assert eventLoop.inEventLoop(); + if (!isWritable()) { + // Already terminated or closed, or the protocol has seen the end of this write. + return; + } + // Also drops anything the source emits after the cancel. + state = set(state, SOURCE_CANCELLED); + if (written) { + // Part of the request is on the wire and nothing can end it cleanly. Closing outbound lets the peer + // see the truncation, and keeps the next request from being written after it. + closeHandler.closeChannelOutbound(channel); + } + } + void sourceTerminated(@Nullable Throwable cause, boolean markCancelled) { assert eventLoop.inEventLoop(); if (isAnySet(state, SUBSCRIBER_OR_SOURCE_TERMINATED)) { diff --git a/servicetalk-transport-netty-internal/src/test/java/io/servicetalk/transport/netty/internal/WriteStreamSubscriberTest.java b/servicetalk-transport-netty-internal/src/test/java/io/servicetalk/transport/netty/internal/WriteStreamSubscriberTest.java index 07257659db..b4fe67f44f 100644 --- a/servicetalk-transport-netty-internal/src/test/java/io/servicetalk/transport/netty/internal/WriteStreamSubscriberTest.java +++ b/servicetalk-transport-netty-internal/src/test/java/io/servicetalk/transport/netty/internal/WriteStreamSubscriberTest.java @@ -1,5 +1,5 @@ /* - * Copyright © 2018 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. @@ -15,6 +15,7 @@ */ package io.servicetalk.transport.netty.internal; +import io.servicetalk.concurrent.Cancellable; import io.servicetalk.concurrent.PublisherSource.Subscription; import io.servicetalk.concurrent.api.TestSubscription; import io.servicetalk.transport.netty.internal.NoopTransportObserver.NoopWriteObserver; @@ -367,6 +368,57 @@ void clientTerminateSource() { verifyNoMoreInteractions(subscription, closeHandler, completableSubscriber); } + @Test + void clientCancelAfterPartialWriteClosesOutbound() { + setUp(true, false); + writeAndFlush("Hello"); + + writeCancellable().cancel(); + verify(subscription).cancel(); + verify(closeHandler).closeChannelOutbound(channel); + verifyNoMoreInteractions(closeHandler); + } + + @Test + void clientCancelBeforeWriteDropsLaterItems() { + setUp(true, false); + + writeCancellable().cancel(); + verify(subscription).cancel(); + subscriber.onNext("Hello"); + verifyWriteSuccessful(); + verifyNoInteractions(closeHandler); + } + + @Test + void clientCancelAfterSourceCompletesDoesNotClose() { + setUp(true, false); + writeAndFlush("Hello"); + subscriber.onComplete(); + verifyListenerSuccessful(); + + writeCancellable().cancel(); + verifyWriteSuccessful("Hello"); + verifyNoInteractions(closeHandler); + } + + @Test + void serverCancelAfterPartialWriteDoesNotClose() { + setUp(false, false); + writeAndFlush("Hello"); + + writeCancellable().cancel(); + verify(subscription).cancel(); + verifyWriteSuccessful("Hello"); + verifyNoInteractions(closeHandler); + } + + private Cancellable writeCancellable() { + ArgumentCaptor cancellable = forClass(Cancellable.class); + verify(completableSubscriber).onSubscribe(cancellable.capture()); + return cancellable.getValue(); + } + private void failingWriteClosesChannel(Runnable enableWriteFailure) throws InterruptedException { WriteInfo info1 = writeAndFlush("Hello1"); verify(completableSubscriber).onSubscribe(any());