binder: Cancel pending authorizations upon transport termination

Fixes grpc#12929

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.
@@ -174,6 +192,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.
      */
@@ -204,11 +224,20 @@ ListenableFuture<Status> checkAuthorization(MethodDescriptor<?, ?> method) {
         return nonCancellationPropagating(checkThenActRaceWinner);
       }
 
-      try {
-        newPendingAuthResult.setFuture(
-            serverPolicyChecker.checkAuthorizationForServiceAsync(uid, serviceName));
-      } catch (Exception e) {  // Not just RuntimeException! Handle the "sneaky" checked case too.
-        newPendingAuthResult.setException(e);
+      // 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.
+        try {
+          newPendingAuthResult.setFuture(
+              serverPolicyChecker.checkAuthorizationForServiceAsync(uid, serviceName));
+        } catch (Exception e) {  // Not just RuntimeException! Handle the "sneaky" checked case too.
+          newPendingAuthResult.setException(e);
+        }
+      } else {
+        newPendingAuthResult.cancel(false);
       }
 
       Futures.addCallback(
@@ -236,6 +265,21 @@ 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

@@ -20,6 +20,7 @@
 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.assertThrows;
 import static org.robolectric.Shadows.shadowOf;
 
 import android.app.Application;
@@ -49,6 +50,10 @@
 import io.grpc.stub.ServerCalls;
 import java.io.IOException;
 import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeoutException;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -209,6 +214,19 @@ public void testAsyncServerSecurityPolicy_allowed_returnsOkStatus() throws Excep
     assertThat(awaitResult(status).getCode()).isEqualTo(Status.Code.OK);
   }
 
+  @Test
+  public void testAsyncServerSecurityPolicy_shutdownNow_cancelsAuthFutures() throws Exception {
+    ListenableFuture<Status> callResult = makeCall();
+    SettableFuture<Status> authResultFuture = awaitNext(statusesToSet);
+
+    channel.shutdownNow();
+    boolean terminationResult = channel.awaitTermination(10, SECONDS);
+    assertThat(terminationResult).isTrue();
+
+    assertThrows(CancellationException.class, () -> awaitResult(authResultFuture));
+    assertThat(awaitResult(callResult).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

@@ -183,7 +183,25 @@ 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.
+  }
 
   @Test
   public void checkAuthorization_synchronousException_doesNotLeaveStrandedFuture()