From 98fe45ab2b071cb3093b5bd3a37c6500d9415be4 Mon Sep 17 00:00:00 2001 From: jdcormie Date: Sun, 25 Oct 2026 18:15:34 -0700 Subject: [PATCH 1/2] binder: Deliver closed() to streams started after shutdown BinderClientTransport.newStream() registered a new ClientInbound in ongoingCalls right away but the Inbound only learned about its ClientStreamListener later, in ClientStream.start(). If the transport terminated in between (binderDied(), onUnbound(), a peer's SHUTDOWN_TRANSPORT or shutdownNow(), all of which close ongoing calls asynchronously), closeAbnormal() found an UNINITIALIZED Inbound with no listener or StatsTraceContext to report to. start() then attached the listener to an Inbound that was already closed and the RPC hung forever with its tracers never closed. Registering at newStream() was also a leak waiting to happen: gRPC core can abandon a ClientStream it got from newStream() without ever calling start() or cancel() on it (DelayedStream.setStream() silently drops the real stream if the DelayedStream was cancelled in the meantime). Such a stream stayed in ongoingCalls and counted as in-use forever, so a graceful shutdown() of that transport could never terminate. Make newStream() do allocations only and move registration into start(), as OkHttp and Netty do. start() now initializes the Inbound with its listener and then registers the call under the transport lock so, relative to shutdownInternal()'s snapshot of ongoingCalls, there are only two possible outcomes: either shutdown ran first, in which case the call is never registered and start() reports closed(shutdownStatus, MISCARRIED) (mirroring OkHttp's streamReadyToStart() and allowing a transparent retry on a fresh transport), or we registered an Inbound that already has a listener for the eventual close to be delivered to. That MISCARRIED close is delivered under the Inbound lock alone, like every other close. Because a closed stream may still legally be written to and half-closed, the miscarried stream's Outbound is closed first so that nothing can ever go on the wire for a call the server must never see. Two behaviour changes follow, both consistent with OkHttp/Netty: - transportInUse(true) is reported at start() rather than newStream(). - A graceful shutdown() between newStream() and start() no longer waits for the unstarted stream, whose start() then miscarries. Since nothing can reach an Inbound that was never registered, the only way a client Inbound can still be UNINITIALIZED when cancel() runs is if cancel() came before start(). ClientStream#cancel says that's not allowed and gRPC core's paths honour it (ClientCallImpl arms Context and deadline cancellation only after start(), see issue #1343), so cancel() now throws IllegalStateException instead of silently closing a stream nobody is listening to. A cancel() after a miscarried start() remains a legal no-op. unregisterInbound() only decrements the in-use count for an Inbound that was actually registered. --- .../internal/BinderClientTransport.java | 90 ++++- .../grpc/binder/internal/BinderTransport.java | 17 +- .../java/io/grpc/binder/internal/Inbound.java | 41 ++- .../internal/MultiMessageClientStream.java | 13 +- .../io/grpc/binder/internal/Outbound.java | 14 +- .../internal/SingleMessageClientStream.java | 13 +- .../binder/internal/ClientInboundTest.java | 339 +++++++++++++++++- .../RobolectricBinderTransportTest.java | 38 ++ .../binder/internal/TransactionBuilder.java | 32 ++ 9 files changed, 554 insertions(+), 43 deletions(-) diff --git a/binder/src/main/java/io/grpc/binder/internal/BinderClientTransport.java b/binder/src/main/java/io/grpc/binder/internal/BinderClientTransport.java index 58e7d7e2b31..e06a83aa09b 100644 --- a/binder/src/main/java/io/grpc/binder/internal/BinderClientTransport.java +++ b/binder/src/main/java/io/grpc/binder/internal/BinderClientTransport.java @@ -49,6 +49,8 @@ import io.grpc.binder.InboundParcelablePolicy; import io.grpc.binder.SecurityPolicy; import io.grpc.internal.ClientStream; +import io.grpc.internal.ClientStreamListener; +import io.grpc.internal.ClientStreamListener.RpcProgress; import io.grpc.internal.ClientTransportFactory.ClientTransportOptions; import io.grpc.internal.ConnectionClientTransport; import io.grpc.internal.FailingClientStream; @@ -251,6 +253,7 @@ public synchronized ClientStream newStream( tracers); } + // Just allocations here. The new call is registered when it's start()ed, in startStream(). int callId = latestCallId++; if (latestCallId == LAST_CALL_ID) { latestCallId = FIRST_CALL_ID; @@ -260,30 +263,91 @@ public synchronized ClientStream newStream( Inbound.ClientInbound inbound = new Inbound.ClientInbound( this, attributes, callId, GrpcUtil.shouldBeCountedForInUse(callOptions)); - if (ongoingCalls.putIfAbsent(callId, inbound) != null) { + Outbound.ClientOutbound outbound = + new Outbound.ClientOutbound(this, callId, method, headers, statsTraceContext); + if (method.getType().clientSendsOneMessage()) { + return new SingleMessageClientStream(this, inbound, outbound, attributes); + } else { + return new MultiMessageClientStream(this, inbound, outbound, attributes); + } + } + + /** + * Starts a stream allocated by {@link #newStream}: connects {@code inbound} to its Outbound and + * listener and registers it as an ongoing call. + * + *

If this transport has already shut down, the call isn't registered at all and {@code + * listener} is instead told that the stream was closed with {@link RpcProgress#MISCARRIED}, which + * is accurate: nothing has been sent on its behalf yet, so the server never saw it and the RPC + * can be transparently retried on another transport. + * + *

Takes the transport lock, so must not be called with any Inbound or Outbound lock held. + * + * @return true iff the call was registered, in which case the caller may proceed to use the + * stream normally. False iff the stream is now closed and {@code listener} has already been + * told so, in which case the caller must not call it again. + */ + @CheckReturnValue + boolean startStream( + Inbound.ClientInbound inbound, + Outbound.ClientOutbound outbound, + ClientStreamListener listener) { + // Initializing first means that, from the moment it's visible in ongoingCalls, the Inbound has + // a listener to report any close to. + synchronized (inbound) { + inbound.init(outbound, listener); + } + Status failure = registerCall(inbound); + if (failure != null) { + // Nobody else can reach a never-registered Inbound so we can finish up without our own lock, + // like every other close. MISCARRIED promises the server never saw this call, yet a closed + // stream may still be written to and half-closed (the listener's state has no bearing on + // what a caller may do; see ClientCall). So first make sure nothing can ever go on the wire + // for this call, then tell the listener. + synchronized (outbound) { + outbound.close(); + } + synchronized (inbound) { + inbound.closeMiscarried(failure); + } + return false; + } + return true; + } + + /** + * Registers {@code inbound} as an ongoing call, unless we've shut down. + * + * @return null iff {@code inbound} is now registered, otherwise the reason it couldn't be + */ + @Nullable + private synchronized Status registerCall(Inbound.ClientInbound inbound) { + // Registering under our lock means that, relative to shutdownInternal()'s snapshot of + // ongoingCalls, either shutdown ran first (and the stream miscarries) or we're in the snapshot + // (and the stream gets closed like any other). + if (isShutdown()) { + return shutdownStatus; + } + if (ongoingCalls.putIfAbsent(inbound.callId, inbound) != null) { Status failure = Status.INTERNAL.withDescription("Clashing call IDs"); shutdownInternal(failure, true); - return newFailingClientStream(failure, attributes, headers, tracers); + return failure; } - if (inbound.countsForInUse() && numInUseStreams.getAndIncrement() == 0) { clientTransportListener.transportInUse(true); } - Outbound.ClientOutbound outbound = - new Outbound.ClientOutbound(this, callId, method, headers, statsTraceContext); - if (method.getType().clientSendsOneMessage()) { - return new SingleMessageClientStream(inbound, outbound, attributes); - } else { - return new MultiMessageClientStream(inbound, outbound, attributes); - } + return null; } @Override - protected void unregisterInbound(Inbound inbound) { - if (inbound.countsForInUse() && numInUseStreams.decrementAndGet() == 0) { + protected boolean unregisterInbound(Inbound inbound) { + boolean removed = super.unregisterInbound(inbound); + // Only ever registered Inbounds counted towards in-use, and forced termination may have already + // unregistered everything (see notifyTerminated() for the corresponding transportInUse(false)). + if (removed && inbound.countsForInUse() && numInUseStreams.decrementAndGet() == 0) { clientTransportListener.transportInUse(false); } - super.unregisterInbound(inbound); + return removed; } @Override diff --git a/binder/src/main/java/io/grpc/binder/internal/BinderTransport.java b/binder/src/main/java/io/grpc/binder/internal/BinderTransport.java index db06d3052ab..c5ccbff9f06 100644 --- a/binder/src/main/java/io/grpc/binder/internal/BinderTransport.java +++ b/binder/src/main/java/io/grpc/binder/internal/BinderTransport.java @@ -396,11 +396,21 @@ protected synchronized void sendPing(int id) throws StatusException { } } - protected void unregisterInbound(Inbound inbound) { - unregisterCall(inbound.callId); + /** + * Removes {@code inbound} from the set of ongoing calls, if it's there. + * + * @return true iff {@code inbound} was registered (and now isn't) + */ + protected boolean unregisterInbound(Inbound inbound) { + return unregisterCall(inbound.callId); } - final void unregisterCall(int callId) { + /** + * Removes the call identified by {@code callId} from the set of ongoing calls, if it's there. + * + * @return true iff {@code callId} was registered (and now isn't) + */ + final boolean unregisterCall(int callId) { boolean removed = (ongoingCalls.remove(callId) != null); if (removed && ongoingCalls.isEmpty()) { // If inState(SHUTDOWN), we'd need to terminate (as the last call to finish). But we can't @@ -416,6 +426,7 @@ final void unregisterCall(int callId) { } }); } + return removed; } final void sendTransaction(int callId, ParcelHolder parcel) throws StatusException { diff --git a/binder/src/main/java/io/grpc/binder/internal/Inbound.java b/binder/src/main/java/io/grpc/binder/internal/Inbound.java index bef25fb4de2..b4358d33b2f 100644 --- a/binder/src/main/java/io/grpc/binder/internal/Inbound.java +++ b/binder/src/main/java/io/grpc/binder/internal/Inbound.java @@ -154,12 +154,11 @@ private Inbound(T transport, Attributes attributes, int callId) { @GuardedBy("this") final void init(Outbound outbound, L listener) { + checkState(isUninitialized(), "init() in state %s", deliveryState); this.outbound = outbound; this.statsTraceContext = outbound.getStatsTraceContext(); this.listener = listener; - if (!isClosed()) { - onDeliveryState(State.INITIALIZED); - } + onDeliveryState(State.INITIALIZED); } final void unregister() { @@ -184,6 +183,12 @@ protected final boolean isClosed() { return deliveryState == State.CLOSED; } + /** Returns true iff our delivery state is still {@link State#UNINITIALIZED}. */ + @GuardedBy("this") + final boolean isUninitialized() { + return deliveryState == State.UNINITIALIZED; + } + @GuardedBy("this") private final boolean messageAvailable() { return firstMessage != null || nextCompleteMessageEnd > 0; @@ -285,7 +290,7 @@ final void closeAbnormal(Status status) { private final void closeAbnormal( Status outboundStatus, Status internalStatus, boolean isOobFromRemote) { if (!isClosed()) { - boolean wasInitialized = (deliveryState != State.UNINITIALIZED); + boolean wasInitialized = !isUninitialized(); onDeliveryState(State.CLOSED); if (wasInitialized) { statsTraceContext.streamClosed(internalStatus); @@ -616,6 +621,34 @@ protected void deliverSuffix() { protected void deliverCloseAbnormal(Status status) { listener.closed(status, RpcProgress.PROCESSED, new Metadata()); } + + /** + * Handles {@link io.grpc.internal.ClientStream#cancel}, which "may only be called after + * start()". + * + *

Nothing can reach us before we're registered, which happens at start(), so still being + * {@link State#UNINITIALIZED} here can only mean that contract was broken. + */ + @GuardedBy("this") + void cancel(Status status) { + checkState(!isUninitialized(), "cancel() before start()"); + closeOnCancel(status); + } + + /** + * Closes a stream that couldn't be registered because its transport had already shut down. + * + *

Nothing has been sent on the stream's behalf, so this is reported as {@link + * RpcProgress#MISCARRIED}. And since the stream was never registered, there's no out-of-band + * close to send and nothing to unregister(). + */ + @GuardedBy("this") + void closeMiscarried(Status status) { + checkState(!isClosed(), "already closed"); + onDeliveryState(State.CLOSED); + statsTraceContext.streamClosed(status); + listener.closed(status, RpcProgress.MISCARRIED, new Metadata()); + } } // ====================================== diff --git a/binder/src/main/java/io/grpc/binder/internal/MultiMessageClientStream.java b/binder/src/main/java/io/grpc/binder/internal/MultiMessageClientStream.java index 9873adcb447..5b2a14753fe 100644 --- a/binder/src/main/java/io/grpc/binder/internal/MultiMessageClientStream.java +++ b/binder/src/main/java/io/grpc/binder/internal/MultiMessageClientStream.java @@ -46,12 +46,17 @@ */ final class MultiMessageClientStream implements ClientStream { + private final BinderClientTransport transport; private final Inbound.ClientInbound inbound; private final Outbound.ClientOutbound outbound; private final Attributes attributes; MultiMessageClientStream( - Inbound.ClientInbound inbound, Outbound.ClientOutbound outbound, Attributes attributes) { + BinderClientTransport transport, + Inbound.ClientInbound inbound, + Outbound.ClientOutbound outbound, + Attributes attributes) { + this.transport = transport; this.inbound = inbound; this.outbound = outbound; this.attributes = attributes; @@ -59,8 +64,8 @@ final class MultiMessageClientStream implements ClientStream { @Override public void start(ClientStreamListener listener) { - synchronized (inbound) { - inbound.init(outbound, listener); + if (!transport.startStream(inbound, outbound, listener)) { + return; // Already closed as MISCARRIED and 'listener' has been told. } if (outbound.isReady()) { listener.onReady(); @@ -120,7 +125,7 @@ public void halfClose() { @Override public void cancel(Status status) { synchronized (inbound) { - inbound.closeOnCancel(status); + inbound.cancel(status); } } diff --git a/binder/src/main/java/io/grpc/binder/internal/Outbound.java b/binder/src/main/java/io/grpc/binder/internal/Outbound.java index 7db5bf0fbe4..0de448bb986 100644 --- a/binder/src/main/java/io/grpc/binder/internal/Outbound.java +++ b/binder/src/main/java/io/grpc/binder/internal/Outbound.java @@ -211,12 +211,24 @@ final void send() throws StatusException { sendInternal(); } catch (StatusException se) { // Ensure we don't send anything else and rethrow. - onOutboundState(State.CLOSED); + close(); throw se; } } } + /** + * Moves to {@link State#CLOSED} so that nothing (more) is ever sent. + * + *

Anything subsequently handed to this Outbound is accepted but never sent. Besides the send + * failure case, this is for a stream that closed before anything was sent on its behalf, where + * sending anything now would be for a call the server must never hear about. + */ + @GuardedBy("this") + final void close() { + onOutboundState(State.CLOSED); + } + @GuardedBy("this") @SuppressWarnings("fallthrough") protected final void sendInternal() throws StatusException { diff --git a/binder/src/main/java/io/grpc/binder/internal/SingleMessageClientStream.java b/binder/src/main/java/io/grpc/binder/internal/SingleMessageClientStream.java index 8a899d621a2..e6cf08ad518 100644 --- a/binder/src/main/java/io/grpc/binder/internal/SingleMessageClientStream.java +++ b/binder/src/main/java/io/grpc/binder/internal/SingleMessageClientStream.java @@ -47,6 +47,7 @@ */ final class SingleMessageClientStream implements ClientStream { + private final BinderClientTransport transport; private final Inbound.ClientInbound inbound; private final Outbound.ClientOutbound outbound; private final Attributes attributes; @@ -55,7 +56,11 @@ final class SingleMessageClientStream implements ClientStream { @Nullable private Deadline pendingDeadline; SingleMessageClientStream( - Inbound.ClientInbound inbound, Outbound.ClientOutbound outbound, Attributes attributes) { + BinderClientTransport transport, + Inbound.ClientInbound inbound, + Outbound.ClientOutbound outbound, + Attributes attributes) { + this.transport = transport; this.inbound = inbound; this.outbound = outbound; this.attributes = attributes; @@ -63,8 +68,8 @@ final class SingleMessageClientStream implements ClientStream { @Override public void start(ClientStreamListener listener) { - synchronized (inbound) { - inbound.init(outbound, listener); + if (!transport.startStream(inbound, outbound, listener)) { + return; // Already closed as MISCARRIED and 'listener' has been told. } if (outbound.isReady()) { listener.onReady(); @@ -114,7 +119,7 @@ public void halfClose() { @Override public void cancel(Status status) { synchronized (inbound) { - inbound.closeOnCancel(status); + inbound.cancel(status); } } diff --git a/binder/src/test/java/io/grpc/binder/internal/ClientInboundTest.java b/binder/src/test/java/io/grpc/binder/internal/ClientInboundTest.java index db9a9202702..48b89fde0bd 100644 --- a/binder/src/test/java/io/grpc/binder/internal/ClientInboundTest.java +++ b/binder/src/test/java/io/grpc/binder/internal/ClientInboundTest.java @@ -19,10 +19,19 @@ import static com.google.common.truth.Truth.assertThat; import static io.grpc.StatusSubject.assertThat; import static io.grpc.binder.internal.TransactionBuilder.newOutOfBandCloseTxnBuilder; +import static io.grpc.binder.internal.TransactionBuilder.newShutdownTransportTxnBuilder; import static io.grpc.binder.internal.TransactionBuilder.newStreamTxnToClientBuilder; import static io.grpc.binder.internal.TransactionBuilder.utf8; import static io.grpc.binder.internal.TransactionUtils.FLAG_MESSAGE_DATA; +import static org.junit.Assert.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; import static org.robolectric.Shadows.shadowOf; import android.content.ComponentName; @@ -38,14 +47,17 @@ import io.grpc.binder.AndroidComponentAddress; import io.grpc.internal.ClientStream; import io.grpc.internal.ClientStreamListener; +import io.grpc.internal.ClientStreamListener.RpcProgress; import io.grpc.internal.FixedObjectPool; import io.grpc.internal.GrpcUtil; import io.grpc.internal.ManagedClientTransport; import io.grpc.internal.StreamListener.MessageProducer; +import io.grpc.internal.testing.TestClientStreamTracer; import java.net.SocketAddress; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; +import org.mockito.InOrder; import org.robolectric.RobolectricTestRunner; /** @@ -489,25 +501,314 @@ public void clientCancelUnregisters() throws Exception { } @Test - public void abnormalCloseBeforeStreamStartDoesNotThrow() throws Exception { - transport.newStream( + public void newStream_doesNotRegisterOrCountAsInUseUntilStart() throws Exception { + ManagedClientTransport.Listener transportListener = mock(ManagedClientTransport.Listener.class); + BinderClientTransport transport = + new DummyClientTransportBuilder().setTransportListener(transportListener).build(); + ClientStream unstartedStream = newUnstartedStream(transport); + drainExecutors(); + + assertThat(transport.getOngoingCalls()).doesNotContainKey(BinderTransport.FIRST_CALL_ID); + verify(transportListener, never()).transportInUse(anyBoolean()); + + unstartedStream.start(new FakeClientStreamListener()); + drainExecutors(); + + assertThat(transport.getOngoingCalls()).containsKey(BinderTransport.FIRST_CALL_ID); + verify(transportListener).transportInUse(true); + } + + @Test + public void shutdownNowBeforeStart_closesListenerAndTracersUponStart() throws Exception { + TestClientStreamTracer tracer = new TestClientStreamTracer(); + ClientStream unstartedStream = newUnstartedStream(new ClientStreamTracer[] {tracer}); + + transport.shutdownNow(Status.UNAVAILABLE.withDescription("test shutdownNow")); + drainExecutors(); + assertThat(transport.getOngoingCalls()).isEmpty(); + assertThat(tracer.getStatus()).isNull(); + + FakeClientStreamListener unstartedListener = new FakeClientStreamListener(); + unstartedStream.start(unstartedListener); + drainExecutors(); + + assertThat(unstartedListener.isClosed()).isTrue(); + assertThat(unstartedListener.getClosedStatus()).hasCode(Status.Code.UNAVAILABLE); + assertThat(unstartedListener.getClosedStatus().getDescription()).isEqualTo("test shutdownNow"); + assertThat(unstartedListener.getClosedProgress()).isEqualTo(RpcProgress.MISCARRIED); + assertThat(unstartedListener.getClosedTrailers().keys()).isEmpty(); + assertThat(tracer.getStatus()).hasCode(Status.Code.UNAVAILABLE); + assertThat(tracer.getStatus().getDescription()).isEqualTo("test shutdownNow"); + assertThat(tracer.getOutboundHeaders()).isFalse(); + assertThat(transport.getOngoingCalls()).isEmpty(); + } + + @Test + public void binderDiedBeforeStart_closesListenerUponStart() throws Exception { + ClientStream unstartedStream = newUnstartedStream(new ClientStreamTracer[0]); + + transport.binderDied(); + drainExecutors(); + assertThat(transport.getOngoingCalls()).isEmpty(); + + FakeClientStreamListener unstartedListener = new FakeClientStreamListener(); + unstartedStream.start(unstartedListener); + drainExecutors(); + + assertThat(unstartedListener.isClosed()).isTrue(); + assertThat(unstartedListener.getClosedStatus()).hasCode(Status.Code.UNAVAILABLE); + assertThat(unstartedListener.getClosedStatus().getDescription()).contains("binderDied"); + assertThat(unstartedListener.getClosedProgress()).isEqualTo(RpcProgress.MISCARRIED); + } + + @Test + public void peerShutdownBeforeStart_closesListenerUponStart() throws Exception { + ClientStream unstartedStream = newUnstartedStream(new ClientStreamTracer[0]); + + newShutdownTransportTxnBuilder().dispatchTo(transport); + drainExecutors(); + assertThat(transport.getOngoingCalls()).isEmpty(); + + FakeClientStreamListener unstartedListener = new FakeClientStreamListener(); + unstartedStream.start(unstartedListener); + drainExecutors(); + + assertThat(unstartedListener.isClosed()).isTrue(); + assertThat(unstartedListener.getClosedStatus()).hasCode(Status.Code.UNAVAILABLE); + assertThat(unstartedListener.getClosedStatus().getDescription()) + .isEqualTo("transport shutdown by peer"); + assertThat(unstartedListener.getClosedProgress()).isEqualTo(RpcProgress.MISCARRIED); + } + + @Test + public void shutdownBeforeStart_closesListenerUponStart() throws Exception { + ClientStream unstartedStream = newUnstartedStream(new ClientStreamTracer[0]); + + // The stream from setUp() is still ongoing so this graceful shutdown won't terminate anything. + transport.shutdown(Status.UNAVAILABLE.withDescription("test shutdown")); + drainExecutors(); + assertThat(transport.getOngoingCalls()).containsKey(inbound.callId); + + FakeClientStreamListener unstartedListener = new FakeClientStreamListener(); + unstartedStream.start(unstartedListener); + drainExecutors(); + + assertThat(unstartedListener.isClosed()).isTrue(); + assertThat(unstartedListener.getClosedStatus()).hasCode(Status.Code.UNAVAILABLE); + assertThat(unstartedListener.getClosedStatus().getDescription()).isEqualTo("test shutdown"); + assertThat(unstartedListener.getClosedProgress()).isEqualTo(RpcProgress.MISCARRIED); + assertThat(transport.getOngoingCalls()).doesNotContainKey(inbound.callId + 1); + assertThat(transport.getOngoingCalls()).containsKey(inbound.callId); + } + + @Test + public void shutdownBeforeStart_doesNotWaitForUnstartedStream() throws Exception { + ManagedClientTransport.Listener transportListener = mock(ManagedClientTransport.Listener.class); + BinderClientTransport transport = + new DummyClientTransportBuilder().setTransportListener(transportListener).build(); + ClientStream unstartedStream = newUnstartedStream(transport); + + transport.shutdown(Status.UNAVAILABLE.withDescription("test shutdown")); + drainExecutors(); + + // Nothing was in flight as far as the transport is concerned, so it's already gone. + verify(transportListener).transportTerminated(); + verify(transportListener, never()).transportInUse(anyBoolean()); + + FakeClientStreamListener unstartedListener = new FakeClientStreamListener(); + unstartedStream.start(unstartedListener); + drainExecutors(); + + assertThat(unstartedListener.isClosed()).isTrue(); + assertThat(unstartedListener.getClosedStatus()).hasCode(Status.Code.UNAVAILABLE); + assertThat(unstartedListener.getClosedStatus().getDescription()).isEqualTo("test shutdown"); + assertThat(unstartedListener.getClosedProgress()).isEqualTo(RpcProgress.MISCARRIED); + verify(transportListener, never()).transportInUse(anyBoolean()); + } + + @Test + public void shutdownNowWithStartedAndUnstartedStreams_reportsNotInUseOnceBeforeTerminated() + throws Exception { + ManagedClientTransport.Listener transportListener = mock(ManagedClientTransport.Listener.class); + BinderClientTransport transport = + new DummyClientTransportBuilder().setTransportListener(transportListener).build(); + ClientStream startedStream = newUnstartedStream(transport); + FakeClientStreamListener startedListener = new FakeClientStreamListener(); + startedStream.start(startedListener); + ClientStream unstartedStream = newUnstartedStream(transport); + drainExecutors(); + verify(transportListener).transportInUse(true); + + transport.shutdownNow(Status.UNAVAILABLE.withDescription("test shutdownNow")); + drainExecutors(); + FakeClientStreamListener unstartedListener = new FakeClientStreamListener(); + unstartedStream.start(unstartedListener); + drainExecutors(); + + assertThat(startedListener.isClosed()).isTrue(); + assertThat(startedListener.getClosedProgress()).isEqualTo(RpcProgress.PROCESSED); + assertThat(unstartedListener.isClosed()).isTrue(); + assertThat(unstartedListener.getClosedProgress()).isEqualTo(RpcProgress.MISCARRIED); + // Only the started stream ever counted as in use, and the forced termination (not the + // never-registered stream's start()) is what un-counts it, exactly once, before terminating. + InOrder inOrder = inOrder(transportListener); + inOrder.verify(transportListener).transportInUse(true); + inOrder.verify(transportListener).transportInUse(false); + inOrder.verify(transportListener).transportTerminated(); + verify(transportListener, times(1)).transportInUse(false); + } + + @Test + public void shutdownWithStartedAndUnstartedStreams_terminatesWhenStartedStreamCompletes() + throws Exception { + ManagedClientTransport.Listener transportListener = mock(ManagedClientTransport.Listener.class); + BinderClientTransport transport = + new DummyClientTransportBuilder().setTransportListener(transportListener).build(); + ClientStream startedStream = newUnstartedStream(transport); + FakeClientStreamListener startedListener = new FakeClientStreamListener(); + startedStream.start(startedListener); + ClientStream unstartedStream = newUnstartedStream(transport); + drainExecutors(); + + transport.shutdown(Status.UNAVAILABLE.withDescription("test shutdown")); + drainExecutors(); + verify(transportListener).transportShutdown(any(), any()); + verify(transportListener, never()).transportTerminated(); + + // The unstarted stream miscarries but the started one keeps the transport alive. + FakeClientStreamListener unstartedListener = new FakeClientStreamListener(); + unstartedStream.start(unstartedListener); + drainExecutors(); + assertThat(unstartedListener.getClosedProgress()).isEqualTo(RpcProgress.MISCARRIED); + verify(transportListener, never()).transportTerminated(); + + // Until it completes. + Inbound.ClientInbound startedInbound = + (Inbound.ClientInbound) transport.getOngoingCalls().get(BinderTransport.FIRST_CALL_ID); + newStreamTxnToClientBuilder(0) + .withPrefix(new Metadata()) + .withSuffix(Status.OK, new Metadata()) + .dispatchTo(startedInbound); + drainExecutors(); + + assertThat(startedListener.isClosed()).isTrue(); + assertThat(startedListener.getClosedStatus()).isOk(); + InOrder inOrder = inOrder(transportListener); + inOrder.verify(transportListener).transportInUse(true); + inOrder.verify(transportListener).transportInUse(false); + inOrder.verify(transportListener).transportTerminated(); + verify(transportListener, times(1)).transportInUse(false); + } + + @Test + public void unaryStreamStartedAfterShutdownNow_closesListenerOnceAndSendsNothing() + throws Exception { + OneWayBinderProxy outgoingBinder = mock(OneWayBinderProxy.class); + BinderClientTransport transport = + new DummyClientTransportBuilder().setOutgoingBinder(outgoingBinder).build(); + MethodDescriptor unaryMethod = + newDummyMethodDescriptorBuilder().setType(MethodDescriptor.MethodType.UNARY).build(); + ClientStream unaryStream = + transport.newStream( + unaryMethod, new Metadata(), CallOptions.DEFAULT, new ClientStreamTracer[0]); + int callId = BinderTransport.FIRST_CALL_ID; + assertThat(transport.getOngoingCalls()).doesNotContainKey(callId); + verify(outgoingBinder, never()).transact(eq(callId), any()); // Nothing is sent until start(). + + transport.shutdownNow(Status.UNAVAILABLE.withDescription("test shutdownNow")); + drainExecutors(); + // The transport never heard of this call so shutdownNow() has nothing to close on its behalf. + verify(outgoingBinder, never()).transact(eq(callId), any()); + + // The ClientStream API allows all of this after a close (and it's how a unary RPC typically + // drives a stream, without waiting to see whether start() closed it). + FakeClientStreamListener unaryListener = new FakeClientStreamListener(); + unaryStream.start(unaryListener); + unaryStream.request(2); + unaryStream.writeMessage(unaryMethod.getRequestMarshaller().stream("request")); + unaryStream.halfClose(); + drainExecutors(); + + // FakeClientStreamListener throws if closed() is called twice or onReady() follows closed(). + assertThat(unaryListener.isClosed()).isTrue(); + assertThat(unaryListener.getClosedStatus()).hasCode(Status.Code.UNAVAILABLE); + assertThat(unaryListener.getClosedProgress()).isEqualTo(RpcProgress.MISCARRIED); + // None of start()/writeMessage()/halfClose() put anything on the wire for a call the server + // must never see. + verify(outgoingBinder, never()).transact(eq(callId), any()); + } + + @Test + public void cancelBeforeStart_throwsIllegalStateException() throws Exception { + // ClientStream#cancel() "may only be called after start()". gRPC core's normal paths honour + // that (ClientCallImpl arms Context/deadline cancellation only after start(); DelayedStream and + // RetriableStream start() a stream before they'd ever cancel() it) and other transports NPE if + // it happens. Fail loudly rather than inventing a close for a stream nobody is listening to. + OneWayBinderProxy outgoingBinder = mock(OneWayBinderProxy.class); + ManagedClientTransport.Listener transportListener = mock(ManagedClientTransport.Listener.class); + BinderClientTransport transport = + new DummyClientTransportBuilder() + .setOutgoingBinder(outgoingBinder) + .setTransportListener(transportListener) + .build(); + ClientStream unstartedStream = newUnstartedStream(transport); + int callId = BinderTransport.FIRST_CALL_ID; + + assertThrows(IllegalStateException.class, () -> unstartedStream.cancel(Status.CANCELLED)); + drainExecutors(); + + // The rejected cancel() left no trace. + assertThat(transport.getOngoingCalls()).doesNotContainKey(callId); + verify(outgoingBinder, never()).transact(eq(callId), any()); + verify(transportListener, never()).transportInUse(anyBoolean()); + } + + @Test + public void cancelAfterMiscarriedStart_isNoOp() throws Exception { + // Unlike cancel() *before* start(), cancel() after a start() that miscarried is perfectly + // legal: ClientCallImpl arms its deadline/Context cancellation only after start() returns. + OneWayBinderProxy outgoingBinder = mock(OneWayBinderProxy.class); + BinderClientTransport transport = + new DummyClientTransportBuilder().setOutgoingBinder(outgoingBinder).build(); + ClientStream unstartedStream = newUnstartedStream(transport); + int callId = BinderTransport.FIRST_CALL_ID; + + transport.shutdownNow(Status.UNAVAILABLE.withDescription("test shutdownNow")); + drainExecutors(); + FakeClientStreamListener unstartedListener = new FakeClientStreamListener(); + unstartedStream.start(unstartedListener); + drainExecutors(); + assertThat(unstartedListener.isClosed()).isTrue(); + assertThat(unstartedListener.getClosedProgress()).isEqualTo(RpcProgress.MISCARRIED); + + unstartedStream.cancel(Status.CANCELLED.withDescription("cancelled after miscarry")); + drainExecutors(); + + // FakeClientStreamListener throws if closed() is called twice. + assertThat(unstartedListener.getClosedStatus()).hasCode(Status.Code.UNAVAILABLE); + assertThat(transport.getOngoingCalls()).doesNotContainKey(callId); + // Nothing was ever sent for this call: no prefix, and no out-of-band close for the cancel. + verify(outgoingBinder, never()).transact(eq(callId), any()); + } + + private ClientStream newUnstartedStream(ClientStreamTracer[] tracers) { + return transport.newStream( + newDummyMethodDescriptorBuilder().build(), new Metadata(), CallOptions.DEFAULT, tracers); + } + + private static ClientStream newUnstartedStream(BinderClientTransport transport) { + return transport.newStream( newDummyMethodDescriptorBuilder().build(), new Metadata(), CallOptions.DEFAULT, new ClientStreamTracer[0]); - Inbound.ClientInbound unstartedInbound = - (Inbound.ClientInbound) transport.getOngoingCalls().get(inbound.callId + 1); - assertThat(transport.getOngoingCalls()).containsKey(unstartedInbound.callId); - - newOutOfBandCloseTxnBuilder(Status.CANCELLED.withDescription("remote abort")) - .dispatchTo(unstartedInbound); - - drainExecutors(); - assertThat(transport.getOngoingCalls()).doesNotContainKey(unstartedInbound.callId); } private static final class DummyClientTransportBuilder { private final BinderClientTransportFactory.Builder factoryBuilder; + private OneWayBinderProxy outgoingBinder = mock(OneWayBinderProxy.class); // Blackhole. + private ManagedClientTransport.Listener transportListener = + mock(ManagedClientTransport.Listener.class); DummyClientTransportBuilder() { MainThreadScheduledExecutorService mainThreadExecutor = @@ -519,6 +820,17 @@ private static final class DummyClientTransportBuilder { .setScheduledExecutorPool(new FixedObjectPool<>(mainThreadExecutor)); } + public DummyClientTransportBuilder setOutgoingBinder(OneWayBinderProxy outgoingBinder) { + this.outgoingBinder = outgoingBinder; + return this; + } + + public DummyClientTransportBuilder setTransportListener( + ManagedClientTransport.Listener transportListener) { + this.transportListener = transportListener; + return this; + } + public BinderClientTransport build() { SocketAddress serverAddress = AndroidComponentAddress.forComponent(new ComponentName("fake.pkg", "fake.cls")); @@ -529,10 +841,9 @@ public BinderClientTransport build() { .build(); // This hack lets us create a transport without the need for a real server for handshaking. synchronized (transport) { - // Blackhole Outbound. - transport.setOutgoingBinder(mock(OneWayBinderProxy.class)); + transport.setOutgoingBinder(outgoingBinder); } - Runnable unused = transport.start(mock(ManagedClientTransport.Listener.class)); + Runnable unused = transport.start(transportListener); synchronized (transport) { transport.setState(BinderTransport.TransportState.READY); } diff --git a/binder/src/test/java/io/grpc/binder/internal/RobolectricBinderTransportTest.java b/binder/src/test/java/io/grpc/binder/internal/RobolectricBinderTransportTest.java index 63c47bf4f19..a13e14f7a66 100644 --- a/binder/src/test/java/io/grpc/binder/internal/RobolectricBinderTransportTest.java +++ b/binder/src/test/java/io/grpc/binder/internal/RobolectricBinderTransportTest.java @@ -29,6 +29,7 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS; import static org.junit.Assume.assumeTrue; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -62,6 +63,7 @@ import io.grpc.binder.internal.SettableAsyncSecurityPolicy.AuthRequest; import io.grpc.internal.AbstractTransportTest; import io.grpc.internal.ClientStream; +import io.grpc.internal.ClientStreamListener.RpcProgress; import io.grpc.internal.ClientStreamListenerBase; import io.grpc.internal.ClientTransport; import io.grpc.internal.ClientTransportFactory.ClientTransportOptions; @@ -78,6 +80,7 @@ import java.util.concurrent.BlockingQueue; import java.util.concurrent.Executor; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.atomic.AtomicReference; import org.junit.Before; import org.junit.Ignore; import org.junit.Rule; @@ -420,6 +423,41 @@ public void clientReportsAuthzErrorToServer() throws Exception { assertThat(serverTransportListener.isTerminated()).isTrue(); } + @Test + public void shutdownNowBetweenNewStreamAndStart_closesStreamAsMiscarried() throws Exception { + server.start(serverListener); + client = newClientTransport(server); + startTransport(client, mockClientTransportListener); + + ClientStream stream = + client.newStream(methodDescriptor, new Metadata(), CallOptions.DEFAULT, tracers); + + Status shutdownReason = Status.UNAVAILABLE.withDescription("shutdownNow() before start()"); + client.shutdownNow(shutdownReason); + verify(mockClientTransportListener, timeout(TIMEOUT_MS)).transportTerminated(); + // An unstarted stream doesn't count as "in use". + verify(mockClientTransportListener, never()).transportInUse(anyBoolean()); + + AtomicReference closedProgress = new AtomicReference<>(); + ClientStreamListenerBase clientStreamListener = + new ClientStreamListenerBase() { + @Override + public void closed(Status status, RpcProgress rpcProgress, Metadata trailers) { + closedProgress.set(rpcProgress); + super.closed(status, rpcProgress, trailers); + } + }; + stream.start(clientStreamListener); + + assertThat(clientStreamListener.awaitClose(TIMEOUT_MS, MILLISECONDS)) + .isSameInstanceAs(shutdownReason); + assertThat(clientStreamListener.awaitTrailers(TIMEOUT_MS, MILLISECONDS)).isNotNull(); + assertThat(closedProgress.get()).isEqualTo(RpcProgress.MISCARRIED); + assertThat(clientStreamTracer1.await(TIMEOUT_MS, MILLISECONDS)).isTrue(); + assertThat(clientStreamTracer1.getStatus()).isSameInstanceAs(shutdownReason); + assertThat(clientStreamTracer1.getOutboundHeaders()).isFalse(); + } + @Test @Override // We don't quite pass the official/abstract version of this test yet because diff --git a/binder/src/testFixtures/java/io/grpc/binder/internal/TransactionBuilder.java b/binder/src/testFixtures/java/io/grpc/binder/internal/TransactionBuilder.java index 2dbe3373f63..f6c7f81ab72 100644 --- a/binder/src/testFixtures/java/io/grpc/binder/internal/TransactionBuilder.java +++ b/binder/src/testFixtures/java/io/grpc/binder/internal/TransactionBuilder.java @@ -54,6 +54,11 @@ public static OutOfBandCloseTxnBuilder newOutOfBandCloseTxnBuilder(Status status return new OutOfBandCloseTxnBuilder(status); } + /** Creates a builder for a transport-level SHUTDOWN_TRANSPORT control transaction. */ + public static ShutdownTransportTxnBuilder newShutdownTransportTxnBuilder() { + return new ShutdownTransportTxnBuilder(); + } + /** Functional interface for parcel consumers that may throw checked exceptions. */ @FunctionalInterface public interface ParcelConsumer { @@ -264,6 +269,33 @@ public void dispatchTo(ParcelConsumer consumer) throws Exception { } } + /** + * Builder for the SHUTDOWN_TRANSPORT control transaction a peer sends when it terminates. + * + *

Unlike stream transactions, this one is addressed to the transport as a whole rather than to + * a particular call, so use {@link #dispatchTo(BinderTransport)}. + */ + public static final class ShutdownTransportTxnBuilder extends TransactionBuilder { + private ShutdownTransportTxnBuilder() {} + + /** Dispatches the synthesized parcel to a transport, as if it came from the peer. */ + public void dispatchTo(BinderTransport transport) throws Exception { + dispatchTo(parcel -> transport.handleTransaction(BinderTransport.SHUTDOWN_TRANSPORT, parcel)); + } + + @Override + public void dispatchTo(ParcelConsumer consumer) throws Exception { + Parcel parcel = Parcel.obtain(); + try { + parcel.writeInt(0); // Flags. Always empty but present, mirroring what a real peer sends. + parcel.setDataPosition(0); + consumer.accept(parcel); + } finally { + parcel.recycle(); + } + } + } + /** * Encodes the given string to bytes using UTF-8. * From c6cd8f55a37a998cafc0de3a6730501ea339d5d0 Mon Sep 17 00:00:00 2001 From: jdcormie Date: Sun, 25 Oct 2026 18:18:23 -0700 Subject: [PATCH 2/2] binder: Don't report onReady() for a closed Inbound Inbound.onTransportReady() unconditionally forwarded onReady() to the listener, so flow control unblocking concurrently with a close could invoke onReady() after closed(), which must be the listener's final callback. Only skip the listener, not the Outbound: a server stream's close may still be sitting in its Outbound waiting for the transmit window to open. --- .../main/java/io/grpc/binder/internal/Inbound.java | 7 ++++++- .../io/grpc/binder/internal/ClientInboundTest.java | 14 ++++++++++++++ 2 files changed, 20 insertions(+), 1 deletion(-) diff --git a/binder/src/main/java/io/grpc/binder/internal/Inbound.java b/binder/src/main/java/io/grpc/binder/internal/Inbound.java index b4358d33b2f..cdd61be6118 100644 --- a/binder/src/main/java/io/grpc/binder/internal/Inbound.java +++ b/binder/src/main/java/io/grpc/binder/internal/Inbound.java @@ -313,8 +313,13 @@ final void onTransportReady() { Outbound outbound = null; StreamListener listener = null; synchronized (this) { + // closed() is always the listener's final callback. + if (!isClosed()) { + listener = this.listener; + } + // But poke the Outbound even if we're closed since a server stream's close could be sitting + // in it, blocked on flow control. outbound = this.outbound; - listener = this.listener; } if (listener != null) { listener.onReady(); diff --git a/binder/src/test/java/io/grpc/binder/internal/ClientInboundTest.java b/binder/src/test/java/io/grpc/binder/internal/ClientInboundTest.java index 48b89fde0bd..8c5117f4668 100644 --- a/binder/src/test/java/io/grpc/binder/internal/ClientInboundTest.java +++ b/binder/src/test/java/io/grpc/binder/internal/ClientInboundTest.java @@ -410,6 +410,20 @@ public void transactionsIgnoredAfterClosed() throws Exception { assertThat(listener.getReadMessages()).isEmpty(); } + @Test + public void transportReadyAfterClosed_doesNotReportOnReady() throws Exception { + newStreamTxnToClientBuilder(nextTxIndex++).withPrefix(new Metadata()).dispatchTo(inbound); + newOutOfBandCloseTxnBuilder(Status.CANCELLED).dispatchTo(inbound); + drainExecutors(); + assertThat(listener.isClosed()).isTrue(); + + // Simulates BinderTransport.handleAcknowledgedBytes() picking this Inbound out of ongoingCalls + // just before a concurrent close (e.g. cancel()) unregistered it. FakeClientStreamListener + // throws if onReady() is invoked after closed(). + inbound.onTransportReady(); + drainExecutors(); + } + @Test public void allInOneUnaryTransaction() throws Exception { newStreamTxnToClientBuilder(0)