binder: Fix listener callbacks in two edge cases

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.
+   *
+   * <p>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.
+   *
+   * <p>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

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 {

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);
@@ -308,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();
@@ -616,6 +626,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()".
+     *
+     * <p>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.
+     *
+     * <p>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());
+    }
   }
 
   // ======================================

binder/src/main/java/io/grpc/binder/internal/MultiMessageClientStream.java

@@ -46,21 +46,26 @@
  */
 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;
   }
 
   @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);
     }
   }
 

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.
+   *
+   * <p>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 {

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,16 +56,20 @@ 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;
   }
 
   @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);
     }
   }