Treat download error during input prefetching as lost input.

... so that bazel can correctly rewind the build.

Author: coeuvre

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

@@ -49,6 +49,7 @@
 import com.google.devtools.build.lib.profiler.ProfilerTask;
 import com.google.devtools.build.lib.remote.common.CacheNotFoundException;
 import com.google.devtools.build.lib.remote.util.AsyncTaskCache;
+import com.google.devtools.build.lib.remote.util.DigestUtil;
 import com.google.devtools.build.lib.util.TempPathGenerator;
 import com.google.devtools.build.lib.vfs.FileSymlinkLoopException;
 import com.google.devtools.build.lib.vfs.FileSystemUtils;
@@ -569,26 +570,33 @@ private Completable downloadFileNoCheckRx(
     }
 
     Path finalPath = path;
+    PathFragment execPath = finalPath.relativeTo(execRoot);
 
     Completable download =
         usingTempPath(
             (tempPath, alreadyDeleted) ->
                 toCompletable(
                         () ->
                             doDownloadFile(
-                                action,
-                                reporter,
-                                tempPath,
-                                finalPath.relativeTo(execRoot),
-                                metadata,
-                                priority,
-                                reason),
+                                action, reporter, tempPath, execPath, metadata, priority, reason),
                         directExecutor())
                     .doOnComplete(
                         () -> {
                           finalizeDownload(
                               metadata, tempPath, finalPath, dirsWithOutputPermissions);
                           alreadyDeleted.set(true);
+                        })
+                    .onErrorResumeNext(
+                        error -> {
+                          if (error instanceof CacheNotFoundException) {
+                            return Completable.error(error);
+                          }
+
+                          // Treat other download error as CacheNotFoundException so that Bazel can correctly rewind the
+                          // action/build.
+                          var digest =
+                              DigestUtil.buildDigest(metadata.getDigest(), metadata.getSize());
+                          return Completable.error(new CacheNotFoundException(digest, execPath));
                         }));
 
     return downloadCache.executeIfNot(

src/main/java/com/google/devtools/build/lib/remote/BUILD

@@ -246,6 +246,7 @@ java_library(
         "//src/main/java/com/google/devtools/build/lib/remote/common:cache_not_found_exception",
         "//src/main/java/com/google/devtools/build/lib/remote/common:lost_inputs_event",
         "//src/main/java/com/google/devtools/build/lib/remote/util",
+        "//src/main/java/com/google/devtools/build/lib/remote/util:digest_utils",
         "//src/main/java/com/google/devtools/build/lib/util:temp_path_generator",
         "//src/main/java/com/google/devtools/build/lib/vfs",
         "//src/main/java/com/google/devtools/build/lib/vfs:pathfragment",
@@ -255,6 +256,7 @@ java_library(
         "//third_party:guava",
         "//third_party:jsr305",
         "//third_party:rxjava3",
+        "@remoteapis//:build_bazel_remote_execution_v2_remote_execution_java_proto",
     ],
 )
 

src/test/shell/bazel/remote/build_without_the_bytes_test.sh

@@ -2062,6 +2062,65 @@ EOF
   fi
 }
 
+function test_retry_build_if_remote_executor_is_unavailable() {
+  mkdir -p a
+
+  cat > a/BUILD <<'EOF'
+genrule(
+  name = 'foo',
+  srcs = ['foo.in'],
+  outs = ['foo.out'],
+  cmd = 'cat $(SRCS) > $@',
+)
+
+genrule(
+  name = 'bar',
+  srcs = ['foo.out', 'bar.in'],
+  outs = ['bar.out'],
+  cmd = 'cat $(SRCS) > $@',
+  tags = ['no-remote-exec'],
+)
+EOF
+
+  echo foo > a/foo.in
+  echo bar > a/bar.in
+
+  # Populate remote cache
+  bazel build \
+      --remote_executor=grpc://localhost:${worker_port} \
+      --remote_download_minimal \
+      //a:bar >& $TEST_log || fail "Failed to build"
+
+  bazel clean
+
+  # Clean build, foo.out isn't downloaded
+  bazel build \
+      --remote_executor=grpc://localhost:${worker_port} \
+      --remote_download_minimal \
+      //a:bar >& $TEST_log || fail "Failed to build"
+
+  if [[ -f bazel-bin/a/foo.out ]]; then
+    fail "Expected intermediate output bazel-bin/a/foo.out to not be downloaded"
+  fi
+
+  # Make the remote worker unavailable
+  stop_worker
+  start_worker --unavailable
+
+  echo "updated bar" > a/bar.in
+
+  # Incremental build triggers remote cache error but Bazel automatically retries the build and reruns the generating
+  # actions locally for missing blobs
+  bazel build \
+      --remote_executor=grpc://localhost:${worker_port} \
+      --remote_local_fallback \
+      --remote_download_minimal \
+      --experimental_remote_cache_eviction_retries=1 \
+      //a:bar >& $TEST_log || fail "Failed to build"
+
+  expect_log "Found transient remote cache error, retrying the build..."
+}
+
 function test_remote_cache_eviction_retries_toplevel_artifacts() {
   mkdir -p a
 

src/tools/remote/src/main/java/com/google/devtools/build/remote/worker/RemoteWorker.java

@@ -150,6 +150,19 @@ public <ReqT, RespT> Listener<ReqT> interceptCall(
     }
   }
 
+  private static class UnavailableInterceptor implements ServerInterceptor {
+
+    @Override
+    public <ReqT, RespT> Listener<ReqT> interceptCall(
+        ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
+      if (!call.getMethodDescriptor().getServiceName().contains("Capabilities")) {
+        call.close(Status.UNAVAILABLE, new Metadata());
+        return new ServerCall.Listener<ReqT>() {};
+      }
+      return Contexts.interceptCall(Context.current(), call, headers, next);
+    }
+  }
+
   public RemoteWorker(
       FileSystem fs,
       RemoteWorkerOptions workerOptions,
@@ -193,6 +206,9 @@ public RemoteWorker(
 
   public Server startServer() throws IOException {
     List<ServerInterceptor> interceptors = new ArrayList<>();
+    if (workerOptions.unavailable) {
+      interceptors.add(new UnavailableInterceptor());
+    }
     interceptors.add(new TracingMetadataUtils.ServerHeadersInterceptor());
     if (workerOptions.expectedAuthorizationToken != null) {
       interceptors.add(new AuthorizationTokenInterceptor(workerOptions.expectedAuthorizationToken));

src/tools/remote/src/main/java/com/google/devtools/build/remote/worker/RemoteWorkerOptions.java

@@ -188,6 +188,16 @@ public class RemoteWorkerOptions extends OptionsBase {
               + " testing only.")
   public String expectedAuthorizationToken;
 
+  @Option(
+      name = "unavailable",
+      defaultValue = "false",
+      documentationCategory = OptionDocumentationCategory.UNCATEGORIZED,
+      effectTags = {OptionEffectTag.UNKNOWN},
+      help =
+          "If true, all gRPC services, except Capabilities, return UNAVAILABLE. This is useful for"
+              + " testing only.")
+  public boolean unavailable;
+
   private static final int MAX_JOBS = 16384;
 
   /**