core: Fix shutdown failing accepted RPCs

core/src/main/java/io/grpc/internal/DelayedClientCall.java

@@ -64,6 +64,8 @@ public class DelayedClientCall<ReqT, RespT> extends ClientCall<ReqT, RespT> {
    * order, but also used if an error occurs before {@code realCall} is set.
    */
   private Listener<RespT> listener;
+  // No need to synchronize; start() synchronization provides a happens-before
+  private Metadata startHeaders;
   // Must hold {@code this} lock when setting.
   private ClientCall<ReqT, RespT> realCall;
   @GuardedBy("this")
@@ -161,13 +163,23 @@ public void run() {
    */
   // When this method returns, passThrough is guaranteed to be true
   public final Runnable setCall(ClientCall<ReqT, RespT> call) {
+    Listener<RespT> savedDelayedListener;
     synchronized (this) {
       // If realCall != null, then either setCall() or cancel() has been called.
       if (realCall != null) {
         return null;
       }
       setRealCall(checkNotNull(call, "call"));
+      // start() not yet called
+      if (delayedListener == null) {
+        assert pendingRunnables.isEmpty();
+        pendingRunnables = null;
+        passThrough = true;
+        return null;
+      }
+      savedDelayedListener = this.delayedListener;
     }
+    internalStart(savedDelayedListener);
     return new ContextRunnable(context) {
       @Override
       public void runInContext() {
@@ -176,8 +188,15 @@ public void runInContext() {
     };
   }
 
+  private void internalStart(Listener<RespT> listener) {
+    Metadata savedStartHeaders = this.startHeaders;
+    this.startHeaders = null;
+    context.run(() -> realCall.start(listener, savedStartHeaders));
+  }
+
   @Override
   public final void start(Listener<RespT> listener, final Metadata headers) {
+    checkNotNull(headers, "headers");
     checkState(this.listener == null, "already started");
     Status savedError;
     boolean savedPassThrough;
@@ -188,6 +207,7 @@ public final void start(Listener<RespT> listener, final Metadata headers) {
       savedPassThrough = passThrough;
       if (!savedPassThrough) {
         listener = delayedListener = new DelayedListener<>(listener);
+        startHeaders = headers;
       }
     }
     if (savedError != null) {
@@ -196,15 +216,7 @@ public final void start(Listener<RespT> listener, final Metadata headers) {
     }
     if (savedPassThrough) {
       realCall.start(listener, headers);
-    } else {
-      final Listener<RespT> finalListener = listener;
-      delayOrExecute(new Runnable() {
-        @Override
-        public void run() {
-          realCall.start(finalListener, headers);
-        }
-      });
-    }
+    } // else realCall.start() will be called by setCall
   }
 
   // When this method returns, passThrough is guaranteed to be true
@@ -253,6 +265,7 @@ public void run() {
       if (listenerToClose != null) {
         callExecutor.execute(new CloseListenerRunnable(listenerToClose, status));
       }
+      internalStart(listenerToClose); // listener instance doesn't matter
       drainPendingCalls();
     }
     callCancelled();

core/src/test/java/io/grpc/internal/DelayedClientCallTest.java

@@ -151,10 +151,12 @@ public void startThenSetCall() {
     delayedClientCall.request(1);
     Runnable r = delayedClientCall.setCall(mockRealCall);
     assertThat(r).isNotNull();
-    r.run();
     @SuppressWarnings("unchecked")
     ArgumentCaptor<Listener<Integer>> listenerCaptor = ArgumentCaptor.forClass(Listener.class);
+    // start() must be called before setCall() returns (not in runnable), to ensure the in-use
+    // counts keeping the channel alive after shutdown() don't momentarily decrease to zero.
     verify(mockRealCall).start(listenerCaptor.capture(), any(Metadata.class));
+    r.run();
     Listener<Integer> realCallListener = listenerCaptor.getValue();
     verify(mockRealCall).request(1);
     realCallListener.onMessage(1);

core/src/test/java/io/grpc/internal/ManagedChannelImplTest.java

@@ -1343,7 +1343,7 @@ public ClientStreamTracer newClientStreamTracer(StreamInfo info, Metadata header
         PickResult.withSubchannel(subchannel));
 
     updateBalancingStateSafely(helper, READY, mockPicker);
-    assertEquals(2, executor.runDueTasks());
+    assertEquals(3, executor.runDueTasks());
 
     verify(mockPicker).pickSubchannel(any(PickSubchannelArgs.class));
     verify(mockTransport).newStream(