Do not cache failed futures for async security policies indefinitely.

Currently, if caching is enabled (as is often the case) and AsyncSecurityPolicy returns a failed future, then this future is cached forever, without giving the SecurityPolicy implementation a chance to be retried. Going forward, new invocations will trigger new security checks if the last one could not be completed successfuly.

Part of #10566.

binder/src/androidTest/java/io/grpc/binder/BinderSecurityTest.java

@@ -45,6 +45,8 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+
 import javax.annotation.Nullable;
 import org.junit.After;
 import org.junit.Before;
@@ -76,6 +78,7 @@ public void setupServiceDefinitionsAndMethods() {
             MethodDescriptor.newBuilder(marshaller, marshaller)
                 .setFullMethodName(name)
                 .setType(MethodDescriptor.MethodType.UNARY)
+                .setSampledToLocalTracing(true)
                 .build();
         ServerCallHandler<Empty, Empty> callHandler =
             ServerCalls.asyncUnaryCall(
@@ -142,7 +145,7 @@ private void assertCallSuccess(MethodDescriptor<Empty, Empty> method) {
   private void assertCallFailure(MethodDescriptor<Empty, Empty> method, Status status) {
     try {
       ClientCalls.blockingUnaryCall(channel, method, CallOptions.DEFAULT, null);
-      fail();
+      fail("Expected call to " + method.getFullMethodName() + " to fail but it succeeded.");
     } catch (StatusRuntimeException sre) {
       assertThat(sre.getStatus().getCode()).isEqualTo(status.getCode());
     }
@@ -172,6 +175,50 @@ public void testServerDisallowsCalls() throws Exception {
     }
   }
 
+  @Test
+  public void testFailedFuturesAreNotCachedPermanently() throws Exception {
+    AtomicReference<Boolean> firstAttempt = new AtomicReference<>(true);
+    createChannel(
+        ServerSecurityPolicy.newBuilder()
+            .servicePolicy("foo", new AsyncSecurityPolicy() {
+              @Override
+              ListenableFuture<Status> checkAuthorizationAsync(int uid) {
+                if (firstAttempt.getAndSet(false)) {
+                  return Futures.immediateFailedFuture(new IllegalStateException());
+                }
+                return Futures.immediateFuture(Status.OK);
+              }
+            })
+            .build(),
+        SecurityPolicies.internalOnly());
+    MethodDescriptor<Empty, Empty> method = methods.get("foo/method0");
+
+    assertCallFailure(method, Status.INTERNAL);
+    assertCallSuccess(method);
+  }
+
+  @Test
+  public void testCancelledFuturesAreNotCachedPermanently() throws Exception {
+    AtomicReference<Boolean> firstAttempt = new AtomicReference<>(true);
+    createChannel(
+        ServerSecurityPolicy.newBuilder()
+            .servicePolicy("foo", new AsyncSecurityPolicy() {
+              @Override
+              ListenableFuture<Status> checkAuthorizationAsync(int uid) {
+                if (firstAttempt.getAndSet(false)) {
+                  return Futures.immediateCancelledFuture();
+                }
+                return Futures.immediateFuture(Status.OK);
+              }
+            })
+            .build(),
+        SecurityPolicies.internalOnly());
+    MethodDescriptor<Empty, Empty> method = methods.get("foo/method0");
+
+    assertCallFailure(method, Status.INTERNAL);
+    assertCallSuccess(method);
+  }
+
   @Test
   public void testClientDoesntTrustServer() throws Exception {
     createChannel(SecurityPolicies.serverInternalOnly(), policy((uid) -> false));

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

@@ -32,9 +32,12 @@
 import io.grpc.Status;
 import io.grpc.internal.GrpcAttributes;
 
+import java.util.concurrent.CancellationException;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
+import java.util.concurrent.Future;
+
 import javax.annotation.CheckReturnValue;
 import javax.annotation.Nullable;
 
@@ -103,7 +106,7 @@ public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
       // Most SecurityPolicy will have synchronous implementations that provide an
       // immediately-resolved Future. In that case, short-circuit to avoid unnecessary allocations
       // and asynchronous code if the authorization result is already present.
-      if (!authStatusFuture.isDone()) {
+      if (!authStatusFuture.isDone() || authStatusFuture.isCancelled()) {
         return newServerCallListenerForPendingAuthResult(authStatusFuture, call, headers, next);
       }
 
@@ -178,9 +181,12 @@ ListenableFuture<Status> checkAuthorization(MethodDescriptor<?, ?> method) {
       boolean useCache = method.isSampledToLocalTracing();
       if (useCache) {
         @Nullable ListenableFuture<Status> authorization = serviceAuthorization.get(serviceName);
-        if (authorization != null) {
+        if (authorization != null && !isFailedOrCancelled(authorization)) {
+          // Authorization check exists and is a pending or successful future (even if for a failed
+          // authorization).
           return authorization;
         }
+        serviceAuthorization.remove(serviceName);
       }
       // Under high load, this may trigger a large number of concurrent authorization checks that
       // perform essentially the same work and have the potential of exhausting the resources they
@@ -198,6 +204,18 @@ ListenableFuture<Status> checkAuthorization(MethodDescriptor<?, ?> method) {
     }
   }
 
+  private static <T> boolean isFailedOrCancelled(Future<T> doneFuture) {
+    if (!doneFuture.isDone()) {
+      return false;
+    }
+    try {
+      T unused = Futures.getDone(doneFuture);
+      return false;
+    } catch (ExecutionException | CancellationException e) {
+      return true;
+    }
+  }
+
   /**
    * Decides whether a given Android UID is authorized to access some resource.
    *