Make bazel run works with minimal mode

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

@@ -524,4 +524,8 @@ public void finalizeAction(Action action, MetadataHandler metadataHandler) {
       prefetchFiles(outputsToDownload, metadataHandler, Priority.LOW);
     }
   }
+
+  public void flushOutputTree() throws InterruptedException {
+    downloadCache.awaitInProgressTasks();
+  }
 }

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

@@ -929,7 +929,14 @@ public void executorInit(CommandEnvironment env, BuildRequest request, ExecutorB
       remoteOutputService.setActionInputFetcher(actionInputFetcher);
       actionContextProvider.setActionInputFetcher(actionInputFetcher);
 
-      if (remoteOutputsMode.downloadToplevelOutputsOnly()) {
+      boolean useToplevelDownloader = remoteOutputsMode.downloadToplevelOutputsOnly();
+
+      // Download toplevel artifacts for `run` command.
+      if ("run".equals(env.getCommandName())) {
+        useToplevelDownloader = true;
+      }
+
+      if (useToplevelDownloader) {
         toplevelArtifactsDownloader =
             new ToplevelArtifactsDownloader(
                 env.getCommandName(),

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

@@ -95,6 +95,13 @@ public ModifiedFileSet startBuild(
     return ModifiedFileSet.EVERYTHING_MODIFIED;
   }
 
+  @Override
+  public void flushOutputTree() throws InterruptedException {
+    if (actionInputFetcher != null) {
+      actionInputFetcher.flushOutputTree();
+    }
+  }
+
   @Override
   public void finalizeBuild(boolean buildSuccessful) {
     // Intentionally left empty.

src/main/java/com/google/devtools/build/lib/remote/util/AsyncTaskCache.java

@@ -21,11 +21,14 @@
 import io.reactivex.rxjava3.annotations.NonNull;
 import io.reactivex.rxjava3.core.Completable;
 import io.reactivex.rxjava3.core.CompletableEmitter;
+import io.reactivex.rxjava3.core.Flowable;
 import io.reactivex.rxjava3.core.Single;
 import io.reactivex.rxjava3.core.SingleObserver;
 import io.reactivex.rxjava3.disposables.Disposable;
 import io.reactivex.rxjava3.functions.Action;
+import io.reactivex.rxjava3.subjects.AsyncSubject;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -131,6 +134,8 @@ class Execution extends Single<ValueT> implements SingleObserver<ValueT> {
     @GuardedBy("lock")
     private final List<SingleObserver<? super ValueT>> observers = new ArrayList<>();
 
+    private final AsyncSubject<ValueT> completion = AsyncSubject.create();
+
     Execution(KeyT key, Single<ValueT> upstream) {
       this.key = key;
       this.upstream = upstream;
@@ -182,6 +187,9 @@ public void onSuccess(@NonNull ValueT value) {
             observer.onSuccess(value);
           }
 
+          completion.onNext(value);
+          completion.onComplete();
+
           maybeNotifyTermination();
         }
       }
@@ -198,6 +206,8 @@ public void onError(@NonNull Throwable error) {
             observer.onError(error);
           }
 
+          completion.onError(error);
+
           maybeNotifyTermination();
         }
       }
@@ -348,7 +358,42 @@ public void shutdown() {
     }
   }
 
-  /** Waits for the channel to become terminated. */
+  /**
+   * Waits for the in-progress tasks to finish. Any tasks that are submitted after the call are not
+   * waited.
+   */
+  public void awaitInProgressTasks() throws InterruptedException {
+    Completable completable =
+        Completable.defer(
+            () -> {
+              Collection<Execution> executions;
+              synchronized (lock) {
+                executions = ImmutableList.copyOf(inProgress.values());
+              }
+
+              if (executions.isEmpty()) {
+                return Completable.complete();
+              }
+
+              return Completable.fromPublisher(
+                  Flowable.fromIterable(executions)
+                      .flatMapSingle(e -> Single.fromObservable(e.completion)));
+            });
+
+    try {
+      completable.blockingAwait();
+    } catch (RuntimeException e) {
+      Throwable cause = e.getCause();
+      if (cause != null) {
+        throwIfInstanceOf(cause, InterruptedException.class);
+      }
+      throw e;
+    }
+  }
+
+  /**
+   * Waits for the channel to become terminated.
+   */
   public void awaitTermination() throws InterruptedException {
     Completable completable =
         Completable.create(
@@ -493,7 +538,17 @@ public void shutdown() {
       cache.shutdown();
     }
 
-    /** Waits for the cache to become terminated. */
+    /**
+     * Waits for the in-progress tasks to finish. Any tasks that are submitted after the call are
+     * not waited.
+     */
+    public void awaitInProgressTasks() throws InterruptedException {
+      cache.awaitInProgressTasks();
+    }
+
+    /**
+     * Waits for the cache to become terminated.
+     */
     public void awaitTermination() throws InterruptedException {
       cache.awaitTermination();
     }

src/main/java/com/google/devtools/build/lib/runtime/commands/RunCommand.java

@@ -306,6 +306,14 @@ public BlazeCommandResult exec(CommandEnvironment env, OptionsParsingResult opti
       return BlazeCommandResult.detailedExitCode(result.getDetailedExitCode());
     }
 
+    if (env.getOutputService() != null) {
+      try {
+        env.getOutputService().flushOutputTree();
+      } catch (InterruptedException ignored) {
+        Thread.currentThread().interrupt();
+      }
+    }
+
     // Make sure that we have exactly 1 built target (excluding --run_under),
     // and that it is executable.
     // These checks should only fail if keepGoing is true, because we already did

src/main/java/com/google/devtools/build/lib/skyframe/SkyframeActionExecutor.java

@@ -309,6 +309,10 @@ Path getExecRoot() {
     return executorEngine.getExecRoot();
   }
 
+  ActionInputPrefetcher getActionInputPrefetcher() {
+    return actionInputPrefetcher;
+  }
+
   ActionContextRegistry getActionContextRegistry() {
     return executorEngine;
   }

src/main/java/com/google/devtools/build/lib/skyframe/SkyframeExecutor.java

@@ -2630,6 +2630,10 @@ public MemoizingEvaluator getEvaluator() {
     return memoizingEvaluator;
   }
 
+  public ActionInputPrefetcher getActionInputPrefetcher() {
+    return skyframeActionExecutor.getActionInputPrefetcher();
+  }
+
   @VisibleForTesting
   public ConfiguredRuleClassProvider getRuleClassProviderForTesting() {
     return ruleClassProvider;

src/main/java/com/google/devtools/build/lib/vfs/OutputService.java

@@ -117,6 +117,9 @@ public default boolean shouldTrustRemoteArtifacts() {
   ModifiedFileSet startBuild(EventHandler eventHandler, UUID buildId, boolean finalizeActions)
       throws BuildFailedException, AbruptExitException, InterruptedException;
 
+  /** Flush and wait for in-progress downloads. */
+  void flushOutputTree() throws InterruptedException;
+
   /**
    * Finish the build.
    *

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

@@ -1331,4 +1331,26 @@ EOF
   [[ ! -e "bazel-bin/a/liblib.jdeps" ]] || fail "bazel-bin/a/liblib.jdeps shouldn't exist"
 }
 
+function test_bazel_run_with_minimal() {
+  # Test that `bazel run` works in minimal mode.
+  mkdir -p a
+
+  cat > a/BUILD <<'EOF'
+genrule(
+  name = 'bin',
+  srcs = [],
+  outs = ['bin.out'],
+  cmd = "echo 'echo bin-message' > $@",
+  executable = True,
+)
+EOF
+
+  bazel run \
+    --remote_executor=grpc://localhost:${worker_port} \
+    --remote_download_minimal \
+    //a:bin >& $TEST_log || fail "Failed to run //a:bin"
+
+  expect_log "bin-message"
+}
+
 run_suite "Build without the Bytes tests"