binder: Cancel pending authorizations upon transport termination

Fixes #12929

TAG=agy

Author: jdcormie

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

@@ -127,6 +127,11 @@ void notifyShutdown(Status status) {
     // Nothing to do.
   }
 
+  @Override
+  void notifyTerminatedUnlocked() {
+    BinderTransportSecurity.notifyTerminatedUnlocked(getAttributes());
+  }
+
   @Override
   @GuardedBy("this")
   void notifyTerminated() {

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

@@ -251,6 +251,8 @@ final boolean isReady() {
   @GuardedBy("this")
   abstract void notifyTerminated();
 
+  void notifyTerminatedUnlocked() {}
+
   void releaseExecutors() {
     executorServicePool.returnObject(scheduledExecutorService);
   }
@@ -335,6 +337,8 @@ final void shutdownInternal(Status shutdownStatus, boolean forceTerminate) {
               future.cancel(false); // No effect if already isDone().
             }
 
+            notifyTerminatedUnlocked();
+
             synchronized (this) {
               notifyTerminated();
             }

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

@@ -87,6 +87,24 @@ public static void attachAuthAttrs(
         .set(GrpcAttributes.ATTR_SECURITY_LEVEL, SecurityLevel.PRIVACY_AND_INTEGRITY);
   }
 
+  /**
+   * Informs this module that the transport with the specified 'attributes' is terminating.
+   *
+   * <p>Any resources allocated by this module will be released and no further resources will be
+   * allocated. Any ongoing background work will be canceled and no further background work will be
+   * initiated.
+   *
+   * <p>This method completes futures and therefore may execute arbitrary listener code on a
+   * potentially direct Executor. To avoid deadlock, callers must not hold any locks.
+   */
+  @Internal
+  public static void notifyTerminatedUnlocked(Attributes attributes) {
+    TransportAuthorizationState state = attributes.get(TRANSPORT_AUTHORIZATION_STATE);
+    if (state != null) {
+      state.notifyTerminatedUnlocked();
+    }
+  }
+
   /**
    * Intercepts server calls and ensures they're authorized before allowing them to proceed.
    * Authentication state is fetched from the call attributes, inherited from the transport.
@@ -173,6 +191,8 @@ static final class TransportAuthorizationState {
     private final ConcurrentHashMap<String, ListenableFuture<Status>> serviceAuthorization;
     private final Executor executor;
 
+    private volatile boolean isTerminated;
+
     /**
      * @param executor used for calling into the application. Must outlive the transport.
      */
@@ -203,11 +223,17 @@ ListenableFuture<Status> checkAuthorization(MethodDescriptor<?, ?> method) {
         return nonCancellationPropagating(checkThenActRaceWinner);
       }
 
-      // newPendingAuthResult is visible to other threads at this point but not as a SettableFuture
-      // and always wrapped with nonCancellationPropagating(). Since it must not be complete, we can
-      // ignore the result of setFuture() -- it must always succeed.
-      newPendingAuthResult.setFuture(
-          serverPolicyChecker.checkAuthorizationForServiceAsync(uid, serviceName));
+      // We only check isTerminated *after* the new future is visible to other threads in
+      // serviceAuthorization. In case of a race with a simultaneous call to notifyTerminated(),
+      // better to harmlessly cancel the new future twice rather than not cancel it at all.
+      if (!isTerminated) {
+        // If notifyTerminated() already cancelled newPendingAuthResult, setFuture() will forward
+        // that cancellation to its argument so it's safe to ignore the return value here.
+        newPendingAuthResult.setFuture(
+            serverPolicyChecker.checkAuthorizationForServiceAsync(uid, serviceName));
+      } else {
+        newPendingAuthResult.cancel(false);
+      }
 
       Futures.addCallback(
           newPendingAuthResult,
@@ -233,6 +259,22 @@ public void onFailure(Throwable t) {
 
       return nonCancellationPropagating(newPendingAuthResult);
     }
+
+    /**
+     * After this method returns, every future ever returned by a prior or concurrent call to
+     * checkAuthorization() is guaranteed to be complete. Every future returned by subsequent call
+     * to checkAuthorization() is also guaranteed to be complete.
+     */
+    void notifyTerminatedUnlocked() {
+      // Any entries added to serviceAuthorization *after* this assignment will be immediately
+      // canceled by the adding thread in checkAuthorization().
+      isTerminated = true;
+
+      // Cancel any entries added to serviceAuthorization *before* the volatile assignment above.
+      for (ListenableFuture<Status> authResult : serviceAuthorization.values()) {
+        authResult.cancel(false); // No-op for cached results (already done).
+      }
+    }
   }
 
   /**

binder/src/test/java/io/grpc/binder/RobolectricBinderSecurityTest.java

@@ -19,6 +19,8 @@
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.truth.Truth.assertThat;
 import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.junit.Assert.fail;
 import static org.robolectric.Shadows.shadowOf;
 
 import android.app.Application;
@@ -42,12 +44,16 @@
 import io.grpc.ServerMethodDefinition;
 import io.grpc.ServerServiceDefinition;
 import io.grpc.Status;
+import io.grpc.Status.Code;
 import io.grpc.StatusRuntimeException;
 import io.grpc.protobuf.lite.ProtoLiteUtils;
 import io.grpc.stub.ClientCalls;
 import io.grpc.stub.ServerCalls;
 import java.io.IOException;
 import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -175,6 +181,24 @@ public void testAsyncServerSecurityPolicy_allowed_returnsOkStatus() throws Excep
     assertThat(status.get().getCode()).isEqualTo(Status.Code.OK);
   }
 
+  @Test
+  public void testAsyncServerSecurityPolicy_shutdownNow_cancelsAuthFutures() throws Exception {
+    ListenableFuture<Status> callResult = makeCall();
+    SettableFuture<Status> authResultFuture = statusesToSet.take();
+
+    channel.shutdownNow();
+    boolean terminationResult = channel.awaitTermination(10, SECONDS);
+    assertThat(terminationResult).isTrue();
+
+    try {
+      Status authResult = authResultFuture.get(10, SECONDS);
+      fail("Expected authResultFuture cancellation but got " + authResult);
+    } catch (CancellationException expected) {
+    }
+    assertThat(authResultFuture.isCancelled()).isTrue();
+    assertThat(callResult.get().getCode()).isEqualTo(Status.Code.UNAVAILABLE);
+  }
+
   private ListenableFuture<Status> makeCall() {
     ClientCall<Empty, Empty> call = channel.newCall(getMethodDescriptor(), CallOptions.DEFAULT);
     ListenableFuture<Empty> responseFuture =

binder/src/test/java/io/grpc/binder/internal/TransportAuthorizationStateTest.java

@@ -169,6 +169,26 @@ public void checkAuthorization_failedFuture_notCached() throws Exception {
     assertThat(authResult2.get()).isEqualTo(Status.OK);
   }
 
+  @Test
+  public void notifyTerminatedUnlocked_cancelsPendingAuthFutures() {
+    ListenableFuture<Status> authResult = authState.checkAuthorization(CODEGEN_METHOD);
+    assertThat(authResult.isDone()).isFalse();
+
+    authState.notifyTerminatedUnlocked();
+
+    assertThat(authResult.isCancelled()).isTrue();
+  }
+
+  @Test
+  public void checkAuthorization_afterTermination_returnsCancelledFuture() {
+    authState.notifyTerminatedUnlocked();
+
+    ListenableFuture<Status> authResult = authState.checkAuthorization(CODEGEN_METHOD);
+
+    assertThat(authResult.isCancelled()).isTrue();
+    assertThat(fakePolicyChecker.statusesToSet).isEmpty();  // fakePolicyChecker never called.
+  }
+
   private static final class FakeServerPolicyChecker implements ServerPolicyChecker {
     final LinkedBlockingQueue<SettableFuture<Status>> statusesToSet = new LinkedBlockingQueue<>();