[9.0.1] Fix a deadlock in RemotePathChecker when using a disk cache (https://github.com/bazelbuild/bazel/pull/28416)

src/main/java/com/google/devtools/build/lib/remote/RemoteActionFileSystem.java

@@ -17,12 +17,15 @@
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.util.concurrent.Futures.immediateFailedFuture;
+import static com.google.common.util.concurrent.Futures.immediateVoidFuture;
 import static com.google.devtools.build.lib.remote.util.Utils.getFromFuture;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSortedSet;
 import com.google.common.collect.Iterables;
+import com.google.common.util.concurrent.ListenableFuture;
 import com.google.devtools.build.lib.actions.Action;
 import com.google.devtools.build.lib.actions.ActionExecutionException;
 import com.google.devtools.build.lib.actions.ActionExecutionMetadata;
@@ -373,7 +376,10 @@ public boolean delete(PathFragment path) throws IOException {
   @Override
   public InputStream getInputStream(PathFragment path) throws IOException {
     try {
-      downloadIfRemote(path);
+      getFromFuture(downloadIfRemote(path));
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      throw new IOException(String.format("Received interrupt while fetching file '%s'", path), e);
     } catch (BulkTransferException e) {
       var newlyLostInputs = e.getLostArtifacts(inputArtifactData::getInput);
       if (!newlyLostInputs.isEmpty()) {
@@ -384,29 +390,23 @@ public InputStream getInputStream(PathFragment path) throws IOException {
     return localFs.getPath(path).getInputStream();
   }
 
-  private void downloadIfRemote(PathFragment path) throws IOException {
-    if (!isRemote(path)) {
-      return;
+  /** Downloads the file at {@code path} if it is remote. */
+  public ListenableFuture<Void> downloadIfRemote(PathFragment path) {
+    try {
+      if (!isRemote(path)) {
+        return immediateVoidFuture();
+      }
+    } catch (IOException e) {
+      return immediateFailedFuture(e);
     }
     PathFragment execPath = path.relativeTo(execRoot);
     ActionInput input = inputArtifactData.getInput(execPath);
     if (input == null) {
       // TODO(tjgq): Also look up the remote output tree.
-      return;
-    }
-
-    try {
-      getFromFuture(
-          inputFetcher.prefetchFiles(
-              action,
-              ImmutableList.of(input),
-              inputArtifactData,
-              Priority.CRITICAL,
-              Reason.INPUTS));
-    } catch (InterruptedException e) {
-      Thread.currentThread().interrupt();
-      throw new IOException(String.format("Received interrupt while fetching file '%s'", path), e);
+      return immediateVoidFuture();
     }
+    return inputFetcher.prefetchFiles(
+        action, ImmutableList.of(input), inputArtifactData, Priority.CRITICAL, Reason.INPUTS);
   }
 
   @Override

src/main/java/com/google/devtools/build/lib/remote/RemoteExecutionCache.java

@@ -17,6 +17,7 @@
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static com.google.common.util.concurrent.Futures.immediateFailedFuture;
 import static com.google.common.util.concurrent.Futures.immediateFuture;
+import static com.google.common.util.concurrent.Futures.immediateVoidFuture;
 import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
 import static com.google.devtools.build.lib.remote.util.RxFutures.toCompletable;
 import static com.google.devtools.build.lib.remote.util.RxFutures.toSingle;
@@ -33,6 +34,7 @@
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Iterables;
 import com.google.common.flogger.GoogleLogger;
+import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.devtools.build.lib.actions.cache.VirtualActionInput;
 import com.google.devtools.build.lib.profiler.Profiler;
@@ -74,33 +76,41 @@ public class RemoteExecutionCache extends CombinedCache implements MerkleTreeUpl
   private static final GoogleLogger logger = GoogleLogger.forEnclosingClass();
 
   /**
-   * An interface used to check whether a given {@link Path} is stored in a remote or a disk cache.
+   * An interface used to check whether a given {@link Path} is available without contacting the
+   * remote cache, i.e., it is present on the local disk, perhaps after being downloaded from the
+   * disk cache.
    */
   public interface RemotePathChecker {
-    boolean isRemote(RemoteActionExecutionContext context, Path path) throws IOException;
+    ListenableFuture<Boolean> isAvailableLocally(RemoteActionExecutionContext context, Path path);
   }
 
   private RemotePathChecker remotePathChecker =
       new RemotePathChecker() {
         @Override
-        public boolean isRemote(RemoteActionExecutionContext context, Path path)
-            throws IOException {
+        public ListenableFuture<Boolean> isAvailableLocally(
+            RemoteActionExecutionContext context, Path path) {
           var fs = path.getFileSystem();
-          if (fs instanceof RemoteActionFileSystem remoteActionFileSystem) {
-            if (remoteActionFileSystem.isRemote(path)) {
-              if (context.getReadCachePolicy().allowDiskCache()) {
-                try (var inputStream = path.getInputStream()) {
-                  // If the file exists in the disk cache, download it and continue the upload.
-                  return false;
-                } catch (IOException e) {
-                  logger.atWarning().withCause(e).log(
-                      "Failed to get input stream for %s", path.getPathString());
-                }
-              }
-              return true;
-            }
+          if (!(fs instanceof RemoteActionFileSystem remoteActionFileSystem)) {
+            return immediateFuture(true);
+          }
+          // If the file is available in the disk cache, we can attempt to download it from there.
+          ListenableFuture<Void> downloadFromDiskCache = immediateVoidFuture();
+          if (context.getReadCachePolicy().allowDiskCache()) {
+            downloadFromDiskCache =
+                Futures.catchingAsync(
+                    remoteActionFileSystem.downloadIfRemote(path.asFragment()),
+                    IOException.class,
+                    e -> {
+                      logger.atWarning().withCause(e).log(
+                          "Failed to download %s", path.getPathString());
+                      return immediateVoidFuture();
+                    },
+                    directExecutor());
           }
-          return false;
+          return Futures.transform(
+              downloadFromDiskCache,
+              unused -> remoteActionFileSystem.getHostFileSystem().exists(path.asFragment()),
+              directExecutor());
         }
       };
 
@@ -182,25 +192,26 @@ public ListenableFuture<Void> uploadFile(
       RemotePathResolver remotePathResolver,
       Digest digest,
       Path path) {
-    try {
-      if (remotePathChecker.isRemote(context, path)) {
-        // If we get here, the remote input was determined to exist in the remote or disk
-        // cache at some point before action execution, but reported to be missing when
-        // querying the remote for missing action inputs; possibly because it was evicted in
-        // the interim.
-        if (remotePathResolver != null) {
-          throw new CacheNotFoundException(
-              digest, remotePathResolver.localPathToExecPath(path.asFragment()));
-        } else {
-          // This path should only be taken for RemoteRepositoryRemoteExecutor, which has no
-          // way to handle lost inputs.
-          throw new CacheNotFoundException(digest, path.getPathString());
-        }
-      }
-    } catch (IOException e) {
-      return immediateFailedFuture(e);
-    }
-    return remoteCacheClient.uploadFile(context, digest, path);
+    return Futures.transformAsync(
+        remotePathChecker.isAvailableLocally(context, path),
+        isAvailableLocally -> {
+          if (!isAvailableLocally) {
+            // If we get here, the remote input was determined to exist in the remote or disk
+            // cache at some point before action execution, but reported to be missing when
+            // querying the remote for missing action inputs; possibly because it was evicted in
+            // the interim.
+            if (remotePathResolver != null) {
+              throw new CacheNotFoundException(
+                  digest, remotePathResolver.localPathToExecPath(path.asFragment()));
+            } else {
+              // This path should only be taken for RemoteRepositoryRemoteExecutor, which has no
+              // way to handle lost inputs.
+              throw new CacheNotFoundException(digest, path.getPathString());
+            }
+          }
+          return remoteCacheClient.uploadFile(context, digest, path);
+        },
+        directExecutor());
   }
 
   @Override

src/test/java/com/google/devtools/build/lib/remote/BuildWithoutTheBytesIntegrationTest.java

@@ -39,8 +39,10 @@
 import com.google.devtools.build.lib.vfs.Path;
 import com.google.devtools.build.lib.vfs.PathFragment;
 import com.google.devtools.build.lib.vfs.Symlinks;
+import com.google.testing.junit.testparameterinjector.TestParameter;
 import com.google.testing.junit.testparameterinjector.TestParameterInjector;
 import java.io.IOException;
+import java.util.UUID;
 import org.junit.ClassRule;
 import org.junit.Rule;
 import org.junit.Test;
@@ -51,6 +53,8 @@
 public class BuildWithoutTheBytesIntegrationTest extends BuildWithoutTheBytesIntegrationTestBase {
   @ClassRule @Rule public static final WorkerInstance worker = IntegrationTestUtils.createWorker();
 
+  @TestParameter public boolean useDiskCache;
+
   @Override
   protected ImmutableList<String> getStartupOptions() {
     // Some tests require the ability to create symlinks on Windows.
@@ -74,6 +78,10 @@ protected void setupOptions() throws Exception {
       // The default behavior is to require the target path to exist and make a deep copy.
       addOptions("--action_env=MSYS=winsymlinks:native");
     }
+
+    if (useDiskCache) {
+      addOptions("--disk_cache=" + UUID.randomUUID());
+    }
   }
 
   @Override
@@ -127,6 +135,9 @@ protected void assertOutputContains(String content, String contains) throws Exce
   @Override
   protected void evictAllBlobs() throws Exception {
     worker.reset();
+    if (useDiskCache) {
+      addOptions("--disk_cache=" + UUID.randomUUID());
+    }
   }
 
   @Override

src/test/java/com/google/devtools/build/lib/remote/CombinedCacheTest.java

@@ -14,6 +14,7 @@
 package com.google.devtools.build.lib.remote;
 
 import static com.google.common.truth.Truth.assertThat;
+import static com.google.common.util.concurrent.Futures.immediateFuture;
 import static com.google.devtools.build.lib.remote.util.Utils.getFromFuture;
 import static com.google.devtools.build.lib.remote.util.Utils.waitForBulkTransfer;
 import static java.nio.charset.StandardCharsets.UTF_8;
@@ -350,7 +351,8 @@ public void ensureInputsPresent_missingInputs_exceptionHasLostInputs() throws Ex
     RemoteCacheClient cacheProtocol = spy(new InMemoryCacheClient());
     RemoteExecutionCache remoteCache = spy(newRemoteExecutionCache(cacheProtocol));
     remoteCache.setRemotePathChecker(
-        (context, path) -> path.relativeTo(execRoot).equals(PathFragment.create("foo")));
+        (context, path) ->
+            immediateFuture(!path.relativeTo(execRoot).equals(PathFragment.create("foo"))));
 
     Path path = execRoot.getRelative("foo");
     FileSystemUtils.writeContentAsLatin1(path, "bar");