bazelbuild · coeuvre · Nov 4, 2022
@@ -31,6 +31,7 @@
 import com.google.devtools.build.lib.buildtool.buildevent.ProfilerStartedEvent;
 import com.google.devtools.build.lib.events.Event;
 import com.google.devtools.build.lib.events.ExtendedEventHandler;
+import com.google.devtools.build.lib.remote.common.MissingDigestsFinder.Intention;
 import com.google.devtools.build.lib.remote.common.RemoteActionExecutionContext;
 import com.google.devtools.build.lib.remote.common.RemoteActionExecutionContext.CachePolicy;
 import com.google.devtools.build.lib.remote.options.RemoteBuildEventUploadMode;
@@ -247,7 +248,7 @@ private Single<List<PathMetadata>> queryRemoteCache(
     if (digestsToQuery.isEmpty()) {
       return Single.just(knownRemotePaths);
     }
-    return toSingle(() -> remoteCache.findMissingDigests(context, digestsToQuery), executor)
+    return toSingle(() -> remoteCache.findMissingDigests(context, Intention.WRITE, digestsToQuery), executor)
         .onErrorResumeNext(
             error -> {
               reporterUploadError(error);

@@ -178,7 +178,7 @@ public static boolean isRemoteCacheOptions(RemoteOptions options) {
 
   @Override
   public ListenableFuture<ImmutableSet<Digest>> findMissingDigests(
-      RemoteActionExecutionContext context, Iterable<Digest> digests) {
+      RemoteActionExecutionContext context, Intention intention, Iterable<Digest> digests) {
     if (Iterables.isEmpty(digests)) {
       return Futures.immediateFuture(ImmutableSet.of());
     }

@@ -34,6 +34,7 @@
 import com.google.devtools.build.lib.concurrent.ThreadSafety;
 import com.google.devtools.build.lib.exec.SpawnProgressEvent;
 import com.google.devtools.build.lib.remote.common.LazyFileOutputStream;
+import com.google.devtools.build.lib.remote.common.MissingDigestsFinder.Intention;
 import com.google.devtools.build.lib.remote.common.OutputDigestMismatchException;
 import com.google.devtools.build.lib.remote.common.ProgressStatusListener;
 import com.google.devtools.build.lib.remote.common.RemoteActionExecutionContext;
@@ -115,11 +116,14 @@ public CachedActionResult downloadActionResult(
    * guaranteed to be a subset of {@code digests}.
    */
   public ListenableFuture<ImmutableSet<Digest>> findMissingDigests(
-      RemoteActionExecutionContext context, Iterable<Digest> digests) {
-    if (Iterables.isEmpty(digests)) {
+      RemoteActionExecutionContext context, Intention intention, Iterable<Digest> digests) {
+    // 0 byte digest is never considered missing because we don't upload nor download it.
+    Iterable<Digest> nonEmptyDigests = Iterables.filter(digests,
+        digest -> digest.getSizeBytes() != 0);
+    if (Iterables.isEmpty(nonEmptyDigests)) {
       return immediateFuture(ImmutableSet.of());
     }
-    return cacheProtocol.findMissingDigests(context, digests);
+    return cacheProtocol.findMissingDigests(context, intention, nonEmptyDigests);
   }
 
   /** Upload the action result to the remote cache. */

@@ -33,6 +33,7 @@
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.devtools.build.lib.profiler.Profiler;
 import com.google.devtools.build.lib.profiler.SilentCloseable;
+import com.google.devtools.build.lib.remote.common.MissingDigestsFinder.Intention;
 import com.google.devtools.build.lib.remote.common.RemoteActionExecutionContext;
 import com.google.devtools.build.lib.remote.common.RemoteCacheClient;
 import com.google.devtools.build.lib.remote.merkletree.MerkleTree;
@@ -252,7 +253,7 @@ private Single<List<UploadTask>> findMissingBlobs(
                                   if (digestsToQuery.isEmpty()) {
                                     return immediateFuture(ImmutableSet.of());
                                   }
-                                  return findMissingDigests(context, digestsToQuery);
+                                  return findMissingDigests(context, Intention.WRITE, digestsToQuery);
                                 },
                                 directExecutor())
                             .map(

@@ -22,6 +22,7 @@
 import static com.google.common.util.concurrent.Futures.transform;
 import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
 import static com.google.devtools.build.lib.remote.RemoteCache.createFailureDetail;
+import static com.google.devtools.build.lib.remote.RemoteCacheClientFactory.isHttpCache;
 import static com.google.devtools.build.lib.remote.util.Utils.getFromFuture;
 import static com.google.devtools.build.lib.remote.util.Utils.getInMemoryOutputPath;
 import static com.google.devtools.build.lib.remote.util.Utils.grpcAwareErrorMessage;
@@ -86,6 +87,8 @@
 import com.google.devtools.build.lib.remote.RemoteExecutionService.ActionResultMetadata.FileMetadata;
 import com.google.devtools.build.lib.remote.RemoteExecutionService.ActionResultMetadata.SymlinkMetadata;
 import com.google.devtools.build.lib.remote.common.BulkTransferException;
+import com.google.devtools.build.lib.remote.common.CacheNotFoundException;
+import com.google.devtools.build.lib.remote.common.MissingDigestsFinder.Intention;
 import com.google.devtools.build.lib.remote.common.OperationObserver;
 import com.google.devtools.build.lib.remote.common.OutputDigestMismatchException;
 import com.google.devtools.build.lib.remote.common.ProgressStatusListener;
@@ -144,6 +147,7 @@
 import java.util.concurrent.Phaser;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Predicate;
+import java.util.stream.Stream;
 import javax.annotation.Nullable;
 
 /**
@@ -514,27 +518,35 @@ public static class RemoteActionResult {
     private final ActionResult actionResult;
     @Nullable private final ExecuteResponse executeResponse;
     @Nullable private final String cacheName;
+    @Nullable private final ActionResultMetadata metadata;
 
-    /** Creates a new {@link RemoteActionResult} instance from a cached result. */
     public static RemoteActionResult createFromCache(CachedActionResult cachedActionResult) {
+      return createFromCache(cachedActionResult, /* metadata= */ null);
+    }
+
+    /** Creates a new {@link RemoteActionResult} instance from a cached result. */
+    public static RemoteActionResult createFromCache(CachedActionResult cachedActionResult,
+        @Nullable ActionResultMetadata metadata) {
       checkArgument(cachedActionResult != null, "cachedActionResult is null");
       return new RemoteActionResult(
-          cachedActionResult.actionResult(), null, cachedActionResult.cacheName());
+          cachedActionResult.actionResult(), null, cachedActionResult.cacheName(), metadata);
     }
 
     /** Creates a new {@link RemoteActionResult} instance from a execute response. */
     public static RemoteActionResult createFromResponse(ExecuteResponse response) {
       checkArgument(response.hasResult(), "response doesn't have result");
-      return new RemoteActionResult(response.getResult(), response, /* cacheName */ null);
+      return new RemoteActionResult(response.getResult(), response, /* cacheName= */ null, /* metadata= */ null);
     }
 
     public RemoteActionResult(
         ActionResult actionResult,
         @Nullable ExecuteResponse executeResponse,
-        @Nullable String cacheName) {
+        @Nullable String cacheName,
+        @Nullable ActionResultMetadata metadata) {
       this.actionResult = actionResult;
       this.executeResponse = executeResponse;
       this.cacheName = cacheName;
+      this.metadata = metadata;
     }
 
     /** Returns the exit code of remote executed action. */
@@ -562,6 +574,15 @@ public List<OutputSymlink> getOutputDirectorySymlinks() {
       return actionResult.getOutputDirectorySymlinksList();
     }
 
+    public ActionResult getActionResult() {
+      return actionResult;
+    }
+
+    @Nullable
+    public ActionResultMetadata getActionResultMetadata() {
+      return metadata;
+    }
+
     /**
      * Returns the freeform informational message with details on the execution of the action that
      * may be displayed to the user upon failure or when requested explicitly.
@@ -631,23 +652,75 @@ public int hashCode() {
   @Nullable
   public RemoteActionResult lookupCache(RemoteAction action)
       throws IOException, InterruptedException {
-    checkState(
-        action.getRemoteActionExecutionContext().getReadCachePolicy().allowAnyCache(),
-        "spawn doesn't accept cached result");
+    RemoteActionExecutionContext context = action.getRemoteActionExecutionContext();
+    checkState(context.getReadCachePolicy().allowAnyCache(), "spawn doesn't accept cached result");
 
-    CachedActionResult cachedActionResult =
-        remoteCache.downloadActionResult(
-            action.getRemoteActionExecutionContext(),
-            action.getActionKey(),
-            /* inlineOutErr= */ false);
+    CachedActionResult cachedActionResult = remoteCache.downloadActionResult(context,
+        action.getActionKey(), /* inlineOutErr= */ false);
 
     if (cachedActionResult == null) {
       return null;
     }
 
+    if (shouldCheckReferencedFiles(cachedActionResult.actionResult())) {
+      // Check whether all the referenced files are available in the remote cache, otherwise, we
+      // consider it as cache miss.
+      try {
+        ActionResultMetadata metadata;
+        try (SilentCloseable c = Profiler.instance().profile("Remote.parseActionResultMetadata")) {
+          metadata = parseActionResultMetadata(context, cachedActionResult.actionResult());
+        }
+        ImmutableSet<Digest> missingDigests = getFromFuture(
+            remoteCache.findMissingDigests(context, Intention.READ,
+                () -> Stream.concat(metadata.files.values().stream(),
+                        metadata.directories.values().stream().flatMap(dir -> dir.files.stream()))
+                    .map(file -> file.digest).iterator()));
+        if (!missingDigests.isEmpty()) {
+          // Some referenced files are missing, ignore the cached action result to rerun the spawn.
+          return null;
+        }
+        return RemoteActionResult.createFromCache(cachedActionResult, metadata);
+      } catch (CacheNotFoundException e) {
+        return null;
+      } catch (BulkTransferException e) {
+        if (e.onlyCausedByCacheNotFoundException()) {
+          return null;
+        }
+        throw e;
+      }
+    }
+
     return RemoteActionResult.createFromCache(cachedActionResult);
   }
 
+  /**
+   * Returns {@code true} if it should also check whether all references files in the action result
+   * are available in the remote cache to call it a cache-hit.
+   */
+  private boolean shouldCheckReferencedFiles(ActionResult actionResult) {
+    // No need to check failed action result since we will rerun the generating spawn anyway.
+    if (actionResult.getExitCode() != 0) {
+      return false;
+    }
+
+    // Only need to check for Build without the Bytes. In the normal mode, we download all the
+    // referenced files immediately, so any missing files wil be detected and the action result is
+    // ignored. The generating spawn will be re-executed. However, for Build without the Bytes, we
+    // don't download the referenced files within spawn execution. If any download attempt later
+    // fails due to file not found, we cannot rerun the generating spawns at that moment. So we need
+    // to check the existence of referenced files now.
+    if (remoteOptions.remoteOutputsMode.downloadAllOutputs()) {
+      return false;
+    }
+
+    // HttpCache doesn't support `findMissingDigests` so we can't check.
+    if (isHttpCache(remoteOptions)) {
+      return false;
+    }
+
+    return true;
+  }
+
   private ListenableFuture<FileMetadata> downloadFile(
       RemoteActionExecutionContext context,
       ProgressStatusListener progressStatusListener,
@@ -921,13 +994,13 @@ private DirectoryMetadata parseDirectory(
   }
 
   ActionResultMetadata parseActionResultMetadata(
-      RemoteActionExecutionContext context, RemoteActionResult result)
+      RemoteActionExecutionContext context, ActionResult result)
       throws IOException, InterruptedException {
     checkNotNull(remoteCache, "remoteCache can't be null");
 
     Map<Path, ListenableFuture<Tree>> dirMetadataDownloads =
         Maps.newHashMapWithExpectedSize(result.getOutputDirectoriesCount());
-    for (OutputDirectory dir : result.getOutputDirectories()) {
+    for (OutputDirectory dir : result.getOutputDirectoriesList()) {
       dirMetadataDownloads.put(
           remotePathResolver.outputPathToLocalPath(encodeBytestringUtf8(dir.getPath())),
           Futures.transformAsync(
@@ -953,7 +1026,7 @@ ActionResultMetadata parseActionResultMetadata(
     }
 
     ImmutableMap.Builder<Path, FileMetadata> files = ImmutableMap.builder();
-    for (OutputFile outputFile : result.getOutputFiles()) {
+    for (OutputFile outputFile : result.getOutputFilesList()) {
       Path localPath =
           remotePathResolver.outputPathToLocalPath(encodeBytestringUtf8(outputFile.getPath()));
       files.put(
@@ -963,7 +1036,7 @@ ActionResultMetadata parseActionResultMetadata(
 
     ImmutableMap.Builder<Path, SymlinkMetadata> symlinks = ImmutableMap.builder();
     Iterable<OutputSymlink> outputSymlinks =
-        Iterables.concat(result.getOutputFileSymlinks(), result.getOutputDirectorySymlinks());
+        Iterables.concat(result.getOutputFileSymlinksList(), result.getOutputDirectorySymlinksList());
     for (OutputSymlink symlink : outputSymlinks) {
       Path localPath =
           remotePathResolver.outputPathToLocalPath(encodeBytestringUtf8(symlink.getPath()));
@@ -999,9 +1072,11 @@ public InMemoryOutput downloadOutputs(RemoteAction action, RemoteActionResult re
       context = context.withReadCachePolicy(context.getReadCachePolicy().addRemoteCache());
     }
 
-    ActionResultMetadata metadata;
-    try (SilentCloseable c = Profiler.instance().profile("Remote.parseActionResultMetadata")) {
-      metadata = parseActionResultMetadata(context, result);
+    ActionResultMetadata metadata = result.getActionResultMetadata();
+    if (metadata == null) {
+      try (SilentCloseable c = Profiler.instance().profile("Remote.parseActionResultMetadata")) {
+        metadata = parseActionResultMetadata(context, result.getActionResult());
+      }
     }
 
     FileOutErr outErr = action.getSpawnExecutionContext().getFileOutErr();

@@ -37,6 +37,7 @@
 import com.google.devtools.build.lib.actions.ExecException;
 import com.google.devtools.build.lib.actions.UserExecException;
 import com.google.devtools.build.lib.events.ExtendedEventHandler;
+import com.google.devtools.build.lib.remote.common.MissingDigestsFinder.Intention;
 import com.google.devtools.build.lib.remote.common.RemoteActionExecutionContext;
 import com.google.devtools.build.lib.remote.common.RemoteCacheClient;
 import com.google.devtools.build.lib.remote.common.RemoteCacheClient.ActionKey;
@@ -465,7 +466,7 @@ public Single<ActionResult> uploadAsync(
 
     String outputPrefix = "cas/";
     Flowable<RxUtils.TransferResult> bulkTransfers =
-        toSingle(() -> remoteCache.findMissingDigests(context, digests), directExecutor())
+        toSingle(() -> remoteCache.findMissingDigests(context, Intention.WRITE, digests), directExecutor())
             .doOnSubscribe(d -> reportUploadStarted(reporter, action, outputPrefix, digests))
             .doOnError(error -> reportUploadFinished(reporter, action, outputPrefix, digests))
             .doOnDispose(() -> reportUploadFinished(reporter, action, outputPrefix, digests))

@@ -20,12 +20,30 @@
 /** Supports querying a remote cache whether it contains a list of blobs. */
 public interface MissingDigestsFinder {
 
+  /**
+   * The intention for the requested digests.
+   */
+  enum Intention {
+    /**
+     * Intents to read the requested digests. In case of a combined cache, the implementation should
+     * return the intersection of missing digests from each cache component because a following read
+     * operation on the digests could return the content.
+     */
+    READ,
+    /**
+     * Intents to upload for the missing digests. In case of a combined cache, the implementation
+     * should return the union of missing digests from each cache component so an upload will occur
+     * later to make sure the digest are stored on each cache component.
+     */
+    WRITE;
+  }
+
   /**
    * Returns a set of digests that the remote cache does not know about. The returned set is
    * guaranteed to be a subset of {@code digests}.
    *
    * @param digests The list of digests to look for.
    */
   ListenableFuture<ImmutableSet<Digest>> findMissingDigests(
-      RemoteActionExecutionContext context, Iterable<Digest> digests);
+      RemoteActionExecutionContext context, Intention intention, Iterable<Digest> digests);
 }
@@ -21,6 +21,7 @@
 import build.bazel.remote.execution.v2.Digest;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Sets;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.devtools.build.lib.remote.common.LazyFileOutputStream;
@@ -108,15 +109,52 @@ public ListenableFuture<Void> uploadBlob(
 
   @Override
   public ListenableFuture<ImmutableSet<Digest>> findMissingDigests(
-      RemoteActionExecutionContext context, Iterable<Digest> digests) {
+      RemoteActionExecutionContext context, Intention intention, Iterable<Digest> digests) {
+    switch (intention) {
+      case READ:
+        return findMissingDigestsForRead(context, digests);
+      case WRITE:
+        return findMissingDigestsForWrite(context, digests);
+    }
+
+    throw new IllegalStateException("unreachable");
+  }
+
+  private ListenableFuture<ImmutableSet<Digest>> findMissingDigestsForRead(
+      RemoteActionExecutionContext context, Iterable<Digest> digests
+  ) {
+    ListenableFuture<ImmutableSet<Digest>> diskQuery = immediateFuture(ImmutableSet.of());
+    if (context.getReadCachePolicy().allowDiskCache()) {
+      diskQuery = diskCache.findMissingDigests(context, Intention.READ, digests);
+    }
+
+    ListenableFuture<ImmutableSet<Digest>> remoteQuery = immediateFuture(ImmutableSet.of());
+    if (context.getReadCachePolicy().allowRemoteCache()) {
+      remoteQuery = remoteCache.findMissingDigests(context, Intention.READ, digests);
+    }
+
+    ListenableFuture<ImmutableSet<Digest>> diskQueryFinal = diskQuery;
+    ListenableFuture<ImmutableSet<Digest>> remoteQueryFinal = remoteQuery;
+
+    return Futures.whenAllSucceed(remoteQueryFinal, diskQueryFinal)
+        .call(
+            () ->
+                ImmutableSet.copyOf(
+                    Sets.intersection(remoteQueryFinal.get(), diskQueryFinal.get())),
+            directExecutor());
+  }
+
+  private ListenableFuture<ImmutableSet<Digest>> findMissingDigestsForWrite(
+      RemoteActionExecutionContext context, Iterable<Digest> digests
+  ) {
     ListenableFuture<ImmutableSet<Digest>> diskQuery = immediateFuture(ImmutableSet.of());
     if (context.getWriteCachePolicy().allowDiskCache()) {
-      diskQuery = diskCache.findMissingDigests(context, digests);
+      diskQuery = diskCache.findMissingDigests(context, Intention.WRITE, digests);
     }
 
     ListenableFuture<ImmutableSet<Digest>> remoteQuery = immediateFuture(ImmutableSet.of());
     if (context.getWriteCachePolicy().allowRemoteCache()) {
-      remoteQuery = remoteCache.findMissingDigests(context, digests);
+      remoteQuery = remoteCache.findMissingDigests(context, Intention.WRITE, digests);
     }
 
     ListenableFuture<ImmutableSet<Digest>> diskQueryFinal = diskQuery;