Skip to content
Open
Show file tree
Hide file tree
Changes from 15 commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
9d14356
outlier-detection: exclude client and hedging cancellations from call…
AgraVator Jul 22, 2026
7abb485
Update @since tag to 1.84.0 and add tracer cancellation in FailingCli…
AgraVator Jul 22, 2026
f9cc87f
core, inprocess, binder: synchronize client cancellation tracer notif…
AgraVator Jul 23, 2026
120e35c
test: add unit tests for race condition and transport tracer cancella…
AgraVator Jul 24, 2026
f9dd0fb
test: add inprocess client stream cancellation unit tests
AgraVator Jul 24, 2026
fe581ae
Revert "test: add inprocess client stream cancellation unit tests"
AgraVator Jul 24, 2026
5e558a5
test: fix checkstyle line length violations in InProcessTransportTest
AgraVator Jul 24, 2026
634be81
Revert "test: fix checkstyle line length violations in InProcessTrans…
AgraVator Jul 24, 2026
20682d9
test: fix checkstyle violations and missing server.start in InProcess…
AgraVator Jul 27, 2026
bf90eb5
util: exclude client cancellations and deadline exceeded from outlier…
AgraVator Aug 3, 2026
6d3cf19
core: trigger clientCancelled in closeListener based on stopDelivery …
AgraVator Aug 3, 2026
9331806
util: restore incrementCallCount(boolean success) method signature
AgraVator Aug 3, 2026
5f3af3d
test: add explicit stopDelivery unit tests in AbstractClientStreamTest
AgraVator Aug 3, 2026
be50f7b
test: add comprehensive stopDelivery and server vs client cancellatio…
AgraVator Aug 3, 2026
c6e2f4a
test: remove redundant directAssertions test case
AgraVator Aug 3, 2026
f27ce70
core, binder: notify clientCancelled directly in transportReportStatu…
AgraVator Aug 5, 2026
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: 9 additions & 0 deletions api/src/main/java/io/grpc/ClientStreamTracer.java
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,15 @@ public void inboundTrailers(Metadata trailers) {
public void addOptionalLabel(String key, String value) {
}

/**
* The stream was cancelled from the client side before a normal response was received.
*
* @param status the cancellation status
* @since 1.84.0
*/
public void cancelled(Status status) {
}

/**
* Factory class for {@link ClientStreamTracer}.
*/
Expand Down
3 changes: 3 additions & 0 deletions binder/src/main/java/io/grpc/binder/internal/Inbound.java
Original file line number Diff line number Diff line change
Expand Up @@ -268,6 +268,9 @@ private final void deliverInternal() {

@GuardedBy("this")
final void closeOnCancel(Status status) {
if (!isClosed() && statsTraceContext != null) {
Comment thread
AgraVator marked this conversation as resolved.
statsTraceContext.clientCancelled(status);
}
closeAbnormal(Status.CANCELLED, status, false);
}

Expand Down
16 changes: 13 additions & 3 deletions core/src/main/java/io/grpc/internal/AbstractClientStream.java
Original file line number Diff line number Diff line change
Expand Up @@ -197,6 +197,9 @@ public final void halfClose() {
@Override
public final void cancel(Status reason) {
Preconditions.checkArgument(!reason.isOk(), "Should not cancel with OK status");
if (cancelled || transportState().isListenerClosed()) {
Comment thread
AgraVator marked this conversation as resolved.
Outdated
return;
}
cancelled = true;
abstractClientStreamSink().cancel(reason);
}
Expand Down Expand Up @@ -251,6 +254,10 @@ protected TransportState(
}
}

protected final boolean isListenerClosed() {
Comment thread
AgraVator marked this conversation as resolved.
Outdated
return listenerClosed;
}

private void setFullStreamDecompression(boolean fullStreamDecompression) {
this.fullStreamDecompression = fullStreamDecompression;
}
Expand Down Expand Up @@ -435,13 +442,13 @@ public final void transportReportStatus(

if (deframerClosed) {
deframerClosedTask = null;
closeListener(status, rpcProgress, trailers);
closeListener(status, rpcProgress, trailers, stopDelivery);
} else {
deframerClosedTask =
new Runnable() {
@Override
public void run() {
closeListener(status, rpcProgress, trailers);
closeListener(status, rpcProgress, trailers, stopDelivery);
}
};
closeDeframer(stopDelivery);
Expand All @@ -454,9 +461,12 @@ public void run() {
* @throws IllegalStateException if the call has not yet been started.
*/
private void closeListener(
Status status, RpcProgress rpcProgress, Metadata trailers) {
Status status, RpcProgress rpcProgress, Metadata trailers, boolean stopDelivery) {
Comment thread
AgraVator marked this conversation as resolved.
Outdated
if (!listenerClosed) {
listenerClosed = true;
if (stopDelivery) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Using stopDelivery == true as the only indicator of a client-initiated cancellation is flawed because its meaning is context-dependent:

  1. Client Cancels (App / Hedging): Passes stopDelivery = true.
  2. Client Timeouts (DEADLINE_EXCEEDED): Passes stopDelivery = true.
  3. Transport Forceful Shutdown (UNAVAILABLE): Passes stopDelivery = true.
  4. Abrupt Network Drop (UNAVAILABLE): Passes stopDelivery = false.
  5. Server Sends Reset (CANCEL or INTERNAL): Passes stopDelivery = false.

(3) is a problem. In Netty transport for example, NettyClientHandler.forcefulClose explicitly sets stopDelivery = true (passing Status.UNAVAILABLE or similar).

Change this to:

         // We must ensure the status code actually reflects a client-initiated action!
         if (stopDelivery && (status.getCode() == Status.Code.CANCELLED || 
                             status.getCode() == Status.Code.DEADLINE_EXCEEDED)) {
          statsTraceCtx.clientCancelled(status);
          }

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.

No, we definitely shouldn't be doing any logic here based on the status code.

What is wrong with "Transport Forceful Shutdown'? That is defined as cancelling all streams, so it should behave the same as call.cancel().

I can believe we may need to change some cases to stop using stopDelivery=true; I bet there are/were some cases where it previously didn't matter what the value was. I think we can rename stopDelivery to cancelled, as that's the only time we should be discarding data that we've received.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I wanted to count transport forceful shutdown in response to network drop as failure because it is not client app initiated although it might be client network stack initiated. Envoy does count connection drops as failures (LocalOriginConnectFailed) without any HTTP status code received from the remote peer.

@AgraVator AgraVator Aug 5, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Renamed stopDelivery to cancelled and moved statsTraceCtx.clientCancelled(status) directly into transportReportStatus without inspecting status codes, keeping core status-agnostic.

However, we should consider the following cases in addition to the one Kannan pointed out:

  1. Inconsistent Server RST_STREAM(CANCEL) Handling
    On inbound RST_STREAM:

    • OkHttp (OkHttpClientTransport.java#L1484-L1491) computes stopDelivery conditionally based on status code:
      boolean stopDelivery = (status.getCode() == Code.CANCELLED || status.getCode() == Code.DEADLINE_EXCEEDED);
      finishStream(streamId, status, progress, stopDelivery, null, null);
      On RST_STREAM(CANCEL), OkHttp truncates deframer delivery immediately (stopDelivery = true), discarding completed messages in the buffer.
    • Netty (NettyClientHandler.java#L475) hardcodes stopDelivery = false, gracefully delivering completed messages already received in full over the wire.
  2. Cronet Transport Tear-down
    During mobile network handoffs or engine GOAWAYs, Cronet (CronetClientTransport.java#L142) hardcodes stopDelivery = true, treating all interrupted streams as client cancellations instead of transport failures.


Proposal

Should we decouple these two responsibilities in AbstractClientStream.TransportState by introducing a separate boolean parameter (e.g., isClientCancelled) alongside stopDelivery?

public final void transportReportStatus(
    Status status, RpcProgress rpcProgress, boolean stopDelivery, boolean isClientCancelled, Metadata trailers)

This allows transport call sites to specify stopDelivery = true (truncating deframer delivery on teardown) while setting isClientCancelled = false (properly recording connection/transport failures in Outlier Detection).

Or Alternatively, we can go and change the stopDelivery values for these cases (although it will result in behaviour changes)

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

You can ignore the case I mentioned NettyClientHandler.forcefulClose - it is only called during channel shutdown by the application.

  1. The GOAWAY case also happens in NettyClientHandler.createStream.

  2. Connection closes while stream is buffered: stopDelivery passed as true in NettyClientHandler.createStreamTraced

These should be counted as failures.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Noted.
For the cases pointed out above do you have a preference ?

statsTraceCtx.clientCancelled(status);
}
statsTraceCtx.streamClosed(status);
if (getTransportTracer() != null) {
getTransportTracer().reportStreamClosed(status.isOk());
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,11 @@ public void addOptionalLabel(String key, String value) {
delegate().addOptionalLabel(key, value);
}

@Override
public void cancelled(Status status) {
delegate().cancelled(status);
}

@Override
public void streamClosed(Status status) {
delegate().streamClosed(status);
Expand Down
13 changes: 13 additions & 0 deletions core/src/main/java/io/grpc/internal/StatsTraceContext.java
Original file line number Diff line number Diff line change
Expand Up @@ -167,6 +167,19 @@ public void serverCallMethodResolved(MethodDescriptor<?, ?> method) {
}
}

/**
* See {@link ClientStreamTracer#cancelled}. For client-side only.
*
* <p>Called from abstract stream implementations.
*/
public void clientCancelled(Status status) {
for (StreamTracer tracer : tracers) {
if (tracer instanceof ClientStreamTracer) {
((ClientStreamTracer) tracer).cancelled(status);
}
}
}

/**
* See {@link StreamTracer#streamClosed}. This may be called multiple times, and only the first
* value will be taken.
Expand Down
134 changes: 134 additions & 0 deletions core/src/test/java/io/grpc/internal/AbstractClientStreamTest.java
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@

import io.grpc.Attributes;
import io.grpc.CallOptions;
import io.grpc.ClientStreamTracer;
import io.grpc.Codec;
import io.grpc.Deadline;
import io.grpc.Grpc;
Expand Down Expand Up @@ -155,6 +156,139 @@ public void cancel(Status errorStatus) {
verify(mockListener).closed(any(Status.class), same(PROCESSED), any(Metadata.class));
}

@Test
public void cancel_notifiesStatsTraceContext() {
ClientStreamTracer mockTracer = mock(ClientStreamTracer.class);
StatsTraceContext customStatsTraceCtx = new StatsTraceContext(new StreamTracer[] {mockTracer});
final BaseTransportState state = new BaseTransportState(customStatsTraceCtx, transportTracer);
AbstractClientStream stream = new BaseAbstractClientStream(allocator, state, new BaseSink() {
@Override
public void cancel(Status errorStatus) {
state.transportReportStatus(errorStatus, true, new Metadata());
}
}, customStatsTraceCtx, transportTracer);
stream.start(mockListener);

Status cancelStatus = Status.CANCELLED.withDescription("Cancelled by test");
stream.cancel(cancelStatus);

verify(mockTracer).cancelled(cancelStatus);
}

@Test
public void transportReportStatus_okFirst_lateCancellationDoesNotNotifyTracerCancelled() {
ClientStreamTracer mockTracer = mock(ClientStreamTracer.class);
StatsTraceContext customStatsTraceCtx = new StatsTraceContext(new StreamTracer[] {mockTracer});
final BaseTransportState state = new BaseTransportState(customStatsTraceCtx, transportTracer);
AbstractClientStream stream = new BaseAbstractClientStream(allocator, state, new BaseSink() {
@Override
public void cancel(Status errorStatus) {
state.transportReportStatus(errorStatus, true, new Metadata());
}
}, customStatsTraceCtx, transportTracer);
stream.start(mockListener);

// Report Status.OK first
state.transportReportStatus(Status.OK, false, new Metadata());

// Subsequent late cancellation
verify(mockTracer, never()).cancelled(any(Status.class));
}

@Test
public void transportReportStatus_stopDeliveryFalse_doesNotNotifyTracerCancelled() {
ClientStreamTracer mockTracer = mock(ClientStreamTracer.class);
StatsTraceContext customStatsTraceCtx = new StatsTraceContext(new StreamTracer[] {mockTracer});
final BaseTransportState state = new BaseTransportState(customStatsTraceCtx, transportTracer);
AbstractClientStream stream = new BaseAbstractClientStream(allocator, state, new BaseSink() {},
customStatsTraceCtx, transportTracer);
stream.start(mockListener);

// Server-initiated CANCELLED (stopDelivery = false)
state.transportReportStatus(Status.CANCELLED, false, new Metadata());

verify(mockTracer, never()).cancelled(any(Status.class));
verify(mockTracer).streamClosed(Status.CANCELLED);
}

@Test
public void transportReportStatus_stopDeliveryTrue_notifiesTracerCancelled() {
ClientStreamTracer mockTracer = mock(ClientStreamTracer.class);
StatsTraceContext customStatsTraceCtx = new StatsTraceContext(new StreamTracer[] {mockTracer});
final BaseTransportState state = new BaseTransportState(customStatsTraceCtx, transportTracer);
AbstractClientStream stream = new BaseAbstractClientStream(allocator, state, new BaseSink() {},
customStatsTraceCtx, transportTracer);
stream.start(mockListener);

// Client/Transport-initiated cancellation (stopDelivery = true)
Status cancelStatus = Status.CANCELLED.withDescription("Client cancelled");
state.transportReportStatus(cancelStatus, true, new Metadata());

verify(mockTracer).cancelled(cancelStatus);
verify(mockTracer).streamClosed(cancelStatus);
}

@Test
public void transportReportStatus_stopDeliveryFalse_deadlineExceeded_noTracerCancelled() {
ClientStreamTracer mockTracer = mock(ClientStreamTracer.class);
StatsTraceContext customStatsTraceCtx = new StatsTraceContext(new StreamTracer[] {mockTracer});
final BaseTransportState state = new BaseTransportState(customStatsTraceCtx, transportTracer);
AbstractClientStream stream = new BaseAbstractClientStream(
allocator, state, new BaseSink() {}, customStatsTraceCtx, transportTracer);
stream.start(mockListener);

// Server-initiated DEADLINE_EXCEEDED (stopDelivery = false)
Status status = Status.DEADLINE_EXCEEDED.withDescription("Server deadline exceeded");
state.transportReportStatus(status, false, new Metadata());

verify(mockTracer, never()).cancelled(any(Status.class));
verify(mockTracer).streamClosed(status);
}

@Test
public void transportReportStatus_stopDeliveryTrue_deadlineExceeded_notifiesTracerCancelled() {
ClientStreamTracer mockTracer = mock(ClientStreamTracer.class);
StatsTraceContext customStatsTraceCtx = new StatsTraceContext(new StreamTracer[] {mockTracer});
final BaseTransportState state = new BaseTransportState(customStatsTraceCtx, transportTracer);
AbstractClientStream stream = new BaseAbstractClientStream(
allocator, state, new BaseSink() {}, customStatsTraceCtx, transportTracer);
stream.start(mockListener);

// Client/Transport-initiated deadline exceeded (stopDelivery = true)
Status status = Status.DEADLINE_EXCEEDED.withDescription("Client deadline exceeded");
state.transportReportStatus(status, true, new Metadata());

verify(mockTracer).cancelled(status);
verify(mockTracer).streamClosed(status);
}

@Test
public void closeListener_deferredDeframerClose_stopDeliveryFalse_delaysCloseListener() {
ClientStreamTracer mockTracer = mock(ClientStreamTracer.class);
StatsTraceContext customStatsTraceCtx = new StatsTraceContext(new StreamTracer[] {mockTracer});
BaseTransportState state = new BaseTransportState(customStatsTraceCtx, transportTracer);
AbstractClientStream stream = new BaseAbstractClientStream(
allocator, state, new BaseSink() {}, customStatsTraceCtx, transportTracer);
stream.start(mockListener);

// Send partial message into deframer
byte[] data = new byte[] {0, 0, 0, 0, 2, 1}; // 2-byte frame, only 1 byte delivered
state.deframe(ReadableBuffers.wrap(data));

Status statusFalse = Status.CANCELLED.withDescription("deferred stopDelivery false");
state.transportReportStatus(statusFalse, false, new Metadata());

// Listener is not closed yet because deframer is mid-frame and waiting for complete frame
verify(mockTracer, never()).cancelled(any(Status.class));
verify(mockTracer, never()).streamClosed(any(Status.class));

// Request message and provide remaining byte of frame to complete deframer processing
stream.request(1);
state.deframe(ReadableBuffers.wrap(new byte[] {2}));
verify(mockTracer, never()).cancelled(any(Status.class));
verify(mockTracer).streamClosed(any(Status.class));
}

@Test
public void startFailsOnNullListener() {
AbstractClientStream stream =
Expand Down
49 changes: 49 additions & 0 deletions core/src/test/java/io/grpc/internal/StatsTraceContextTest.java
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
/*
* Copyright 2026 The gRPC 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.grpc.internal;

import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;

import io.grpc.ClientStreamTracer;
import io.grpc.ServerStreamTracer;
import io.grpc.Status;
import io.grpc.StreamTracer;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;

/** Unit tests for {@link StatsTraceContext}. */
@RunWith(JUnit4.class)
public class StatsTraceContextTest {

@Test
public void clientCancelled_notifiesClientStreamTracers() {
ClientStreamTracer clientTracer = mock(ClientStreamTracer.class);
ServerStreamTracer serverTracer = mock(ServerStreamTracer.class);

StatsTraceContext statsTraceCtx = new StatsTraceContext(
new StreamTracer[] {clientTracer, serverTracer});

Status cancelledStatus = Status.CANCELLED.withDescription("Client cancelled");
statsTraceCtx.clientCancelled(cancelledStatus);

verify(clientTracer).cancelled(cancelledStatus);
verifyNoInteractions(serverTracer);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -846,6 +846,7 @@ public void cancel(Status reason) {
if (!internalCancel(serverStatus, serverStatus)) {
return;
}
statsTraceCtx.clientCancelled(reason);
serverStream.clientCancelled(reason);
streamClosed();
}
Expand Down
Loading
Loading