Skip to content

transport-netty-internal: don't leave cancelled, partially written client requests open #3669

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

Draft
bryce-anderson wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

transport-netty-internal: don't leave cancelled, partially written client requests open #3669

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
196 changes: 196 additions & 0 deletions ...http-netty/src/test/java/io/servicetalk/http/netty/PartiallyWrittenRequestCancelTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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<Throwable> serverRequestBodyError = new CompletableFuture<>();
try (ServerContext server = HttpServers.forAddress(localAddress(0)).protocols(h2Default())

@idelpivnitskiy idelpivnitskiy Oct 2, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would be great to test HTTP/1.1 scenarios as well. IIUC, before your fix the server can read the next request as payload of the previous one

.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<HostAndPort, InetSocketAddress> 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<StreamingHttpResponse> 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<Buffer> 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;
}
}
31 changes: 29 additions & 2 deletions ...internal/src/main/java/io/servicetalk/transport/netty/internal/WriteStreamSubscriber.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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.
Expand Down Expand Up @@ -86,6 +86,7 @@ final class WriteStreamSubscriber implements PublisherSource.Subscriber<Object>,
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<WriteStreamSubscriber, Subscription> subscriptionUpdater =
Expand Down Expand Up @@ -138,7 +139,10 @@ public void onSubscribe(Subscription s) {
s.cancel();
return;
}
subscriber.onSubscribe(concurrentSubscription);
subscriber.onSubscribe(isClient ? () -> {

@idelpivnitskiy idelpivnitskiy Oct 2, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Optional: wdyt if instead of allocating a captured lambda here we implement Cancellable interface and reuse current test-only cancel() method?

final class WriteStreamSubscriber implements PublisherSource.Subscriber<Object>, ChannelOutboundListener,
                                             Cancellable {
    ...
    subscriber.onSubscribe(isClient ? this : concurrentSubscription);
    ...

    @Override
    public void cancel() {
        Subscription oldVal = subscriptionUpdater.getAndSet(this, CANCELLED);
        if (oldVal == null || oldVal == CANCELLED) {
            return;
        }
        if (eventLoop.inEventLoop()) {
            cancel0(oldVal);
        } else {
            eventLoop.execute(() -> cancel0(oldVal));
        }
    }

    private void cancel0(Subscription oldVal) {
        if (isClient) {
            promise.sourceCancelled();
        }
        oldVal.cancel();
    }

It may also help to fix "cancel twice" problem bcz of subscriptionUpdater and skips work after a finished source.

sourceCancelled();

@idelpivnitskiy idelpivnitskiy Oct 2, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When this runs on the event loop, we can cancel twice: if an HTTP/2 write is stalled by flow control and the response is done, closing the stream fails that write. setFailure0 then cancels the subscription and calls onError from inside cancel(), and the lambda cancels it again. As a result writeCancelled() fires twice, which breaks the FlushStrategy rule of at most one call, and observers see ClosedChannelException for what was a user cancel.

concurrentSubscription.cancel();
} : concurrentSubscription);
if (eventLoop.inEventLoop()) {
initialRequestN(concurrentSubscription);
} else {
Expand Down Expand Up @@ -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);

@idelpivnitskiy idelpivnitskiy Oct 2, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When the cancel comes from another thread, sourceCancelled is queued, but DefaultNettyConnection has already pointed channelOutboundListener back at the connection. If the last chunk or trailers write was already queued, its OutboundDataEndEvent goes to the connection's no-op listener instead of this subscriber. sourceCancelled then treats a fully written request as partial and closes outbound. On HTTP/1.1 that half-closes a keep-alive connection that should have been reusable. On HTTP/2 a stream that already completed gets an unnecessary RST.

}
}

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
Expand Down Expand Up @@ -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);

@idelpivnitskiy idelpivnitskiy Oct 2, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If we consider a recent case of HTTP/1.1 with pipelining, then it doesn't wait for reading responses of earlier pipelined requests that were fully written, like NettyPipelinedConnection.readWithTurn waits for its turn. Remote peers that do not support half-closed connections will close entire TCP channel affecting other responses.

}
}

void sourceTerminated(@Nullable Throwable cause, boolean markCancelled) {
assert eventLoop.inEventLoop();
if (isAnySet(state, SUBSCRIBER_OR_SOURCE_TERMINATED)) {
Expand Down
54 changes: 53 additions & 1 deletion ...rnal/src/test/java/io/servicetalk/transport/netty/internal/WriteStreamSubscriberTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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.
Expand All @@ -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;
Expand Down Expand Up @@ -367,6 +368,57 @@ void clientTerminateSource() {
verifyNoMoreInteractions(subscription, closeHandler, completableSubscriber);
}

@Test
void clientCancelAfterPartialWriteClosesOutbound() {

@idelpivnitskiy idelpivnitskiy Oct 2, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Consider testing some of these scenarios in WriteStreamSubscriberOutOfEventloopTest to provide coverage for both paths

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> 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());
Expand Down
Loading