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

// Copyright 2021 The Bazel Authors. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//    http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package com.google.devtools.build.lib.remote;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;
import static com.google.common.collect.ImmutableMap.toImmutableMap;
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.transform;
import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
import static com.google.devtools.build.lib.analysis.constraints.ConstraintConstants.getOsFromConstraintsOrHost;
import static com.google.devtools.build.lib.remote.CombinedCache.createFailureDetail;
import static com.google.devtools.build.lib.remote.util.BulkTransfers.waitForBulkTransfer;
import static com.google.devtools.build.lib.remote.util.Futures.getFromFuture;
import static com.google.devtools.build.lib.remote.util.Utils.createExecExceptionForCredentialHelperException;
import static com.google.devtools.build.lib.remote.util.Utils.grpcAwareErrorMessage;
import static com.google.devtools.build.lib.util.StringEncoding.internalToUnicode;
import static com.google.devtools.build.lib.util.StringEncoding.unicodeToInternal;
import static java.util.Collections.min;
import static java.util.Comparator.comparing;

import build.bazel.remote.execution.v2.Action;
import build.bazel.remote.execution.v2.ActionResult;
import build.bazel.remote.execution.v2.Command;
import build.bazel.remote.execution.v2.Digest;
import build.bazel.remote.execution.v2.Directory;
import build.bazel.remote.execution.v2.DirectoryNode;
import build.bazel.remote.execution.v2.ExecuteRequest;
import build.bazel.remote.execution.v2.ExecuteResponse;
import build.bazel.remote.execution.v2.ExecutedActionMetadata;
import build.bazel.remote.execution.v2.FileNode;
import build.bazel.remote.execution.v2.LogFile;
import build.bazel.remote.execution.v2.OutputDirectory;
import build.bazel.remote.execution.v2.OutputFile;
import build.bazel.remote.execution.v2.OutputSymlink;
import build.bazel.remote.execution.v2.Platform;
import build.bazel.remote.execution.v2.RequestMetadata;
import build.bazel.remote.execution.v2.SymlinkNode;
import build.bazel.remote.execution.v2.Tree;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import com.google.common.collect.Collections2;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
import com.google.common.eventbus.Subscribe;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
import com.google.devtools.build.lib.actions.ActionAnalysisMetadata;
import com.google.devtools.build.lib.actions.ActionExecutionMetadata;
import com.google.devtools.build.lib.actions.ActionInput;
import com.google.devtools.build.lib.actions.Artifact;
import com.google.devtools.build.lib.actions.EnvironmentalExecException;
import com.google.devtools.build.lib.actions.ExecException;
import com.google.devtools.build.lib.actions.ExecutionRequirements;
import com.google.devtools.build.lib.actions.FileArtifactValue;
import com.google.devtools.build.lib.actions.ParamFileActionInput;
import com.google.devtools.build.lib.actions.Spawn;
import com.google.devtools.build.lib.actions.SpawnResult;
import com.google.devtools.build.lib.actions.Spawns;
import com.google.devtools.build.lib.actions.VirtualActionInput;
import com.google.devtools.build.lib.analysis.platform.PlatformInfo;
import com.google.devtools.build.lib.analysis.platform.PlatformUtils;
import com.google.devtools.build.lib.authandtls.credentialhelper.CredentialHelperException;
import com.google.devtools.build.lib.buildtool.buildevent.BuildCompleteEvent;
import com.google.devtools.build.lib.buildtool.buildevent.BuildInterruptedEvent;
import com.google.devtools.build.lib.concurrent.CancellableTask;
import com.google.devtools.build.lib.events.Event;
import com.google.devtools.build.lib.events.Reporter;
import com.google.devtools.build.lib.exec.ExecutionOptions;
import com.google.devtools.build.lib.exec.SpawnRunner.SpawnExecutionContext;
import com.google.devtools.build.lib.exec.local.LocalEnvProvider;
import com.google.devtools.build.lib.profiler.Profiler;
import com.google.devtools.build.lib.profiler.ProfilerTask;
import com.google.devtools.build.lib.profiler.SilentCloseable;
import com.google.devtools.build.lib.remote.CombinedCache.CachedActionResult;
import com.google.devtools.build.lib.remote.RemoteExecutionService.ActionResultMetadata.DirectoryMetadata;
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.Scrubber.SpawnScrubber;
import com.google.devtools.build.lib.remote.common.ActionKey;
import com.google.devtools.build.lib.remote.common.BulkTransferException;
import com.google.devtools.build.lib.remote.common.LostInputsEvent;
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;
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.common.RemoteExecutionClient;
import com.google.devtools.build.lib.remote.common.RemotePathResolver;
import com.google.devtools.build.lib.remote.merkletree.MerkleTree;
import com.google.devtools.build.lib.remote.merkletree.MerkleTreeComputer;
import com.google.devtools.build.lib.remote.options.RemoteOptions;
import com.google.devtools.build.lib.remote.options.RemoteOptions.ConcurrentChangesCheckLevel;
import com.google.devtools.build.lib.remote.salt.CacheSalt;
import com.google.devtools.build.lib.remote.util.DigestUtil;
import com.google.devtools.build.lib.remote.util.TracingMetadataUtils;
import com.google.devtools.build.lib.remote.util.Utils;
import com.google.devtools.build.lib.remote.util.Utils.InMemoryOutput;
import com.google.devtools.build.lib.server.FailureDetails.RemoteExecution;
import com.google.devtools.build.lib.util.Fingerprint;
import com.google.devtools.build.lib.util.OS;
import com.google.devtools.build.lib.util.StringEncoding;
import com.google.devtools.build.lib.util.TempPathGenerator;
import com.google.devtools.build.lib.util.io.FileOutErr;
import com.google.devtools.build.lib.vfs.FileSystem;
import com.google.devtools.build.lib.vfs.FileSystemUtils;
import com.google.devtools.build.lib.vfs.OsPathPolicy;
import com.google.devtools.build.lib.vfs.OutputService;
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.devtools.build.lib.worker.WorkerKey;
import com.google.devtools.build.lib.worker.WorkerOptions;
import com.google.devtools.build.lib.worker.WorkerParser;
import com.google.devtools.common.options.Options;
import com.google.protobuf.ByteString;
import com.google.protobuf.ExtensionRegistry;
import com.google.protobuf.Message;
import io.grpc.Status.Code;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Phaser;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Predicate;
import java.util.stream.Stream;
import javax.annotation.Nullable;

/**
 * A layer between spawn execution and remote execution exposing primitive operations for remote
 * cache and execution with spawn specific types.
 */
public class RemoteExecutionService {
  private static final Comparator<String> PROTO_STRING_COMPARATOR =
      comparing(StringEncoding::unicodeToInternal);

  private final Reporter reporter;
  private final boolean verboseFailures;
  private final Path execRoot;

  /**
   * Do not use directly, instead use the per-spawn resolver created in {@link
   * #buildRemoteAction(Spawn, SpawnExecutionContext)}.
   */
  private final RemotePathResolver baseRemotePathResolver;

  private final String buildRequestId;
  private final String commandId;
  private final DigestUtil digestUtil;
  private final MerkleTreeComputer merkleTreeComputer;
  private final RemoteOptions remoteOptions;
  private final ExecutionOptions executionOptions;
  @Nullable private final CombinedCache combinedCache;
  @Nullable private final RemoteExecutionClient remoteExecutor;
  private final TempPathGenerator tempPathGenerator;
  @Nullable private final Path captureCorruptedOutputsDir;
  private final Set<String> reportedErrors = new HashSet<>();

  private final ListeningExecutorService backgroundTaskExecutor =
      MoreExecutors.listeningDecorator(
          Executors.newThreadPerTaskExecutor(
              Thread.ofVirtual().name("remote-execution-bg-", 0).factory()));

  private final AtomicBoolean shutdown = new AtomicBoolean(false);
  private final AtomicBoolean buildInterrupted = new AtomicBoolean(false);

  @Nullable private final RemoteOutputChecker remoteOutputChecker;
  private final OutputService outputService;

  @Nullable private final Scrubber scrubber;
  private final Set<Digest> knownMissingCasDigests;
  private final Predicate<ActionAnalysisMetadata> wasRewound;

  private Boolean useOutputPaths;

  public RemoteExecutionService(
      Reporter reporter,
      boolean verboseFailures,
      Path execRoot,
      RemotePathResolver remotePathResolver,
      String buildRequestId,
      String commandId,
      String workspaceName,
      DigestUtil digestUtil,
      RemoteOptions remoteOptions,
      ExecutionOptions executionOptions,
      @Nullable CombinedCache combinedCache,
      @Nullable RemoteExecutionClient remoteExecutor,
      TempPathGenerator tempPathGenerator,
      @Nullable Path captureCorruptedOutputsDir,
      @Nullable RemoteOutputChecker remoteOutputChecker,
      OutputService outputService,
      Set<Digest> knownMissingCasDigests,
      Predicate<ActionAnalysisMetadata> wasRewound) {
    this.reporter = reporter;
    this.verboseFailures = verboseFailures;
    this.execRoot = execRoot;
    this.baseRemotePathResolver = remotePathResolver;
    this.buildRequestId = buildRequestId;
    this.commandId = commandId;
    this.digestUtil = digestUtil;
    this.remoteOptions = remoteOptions;
    this.executionOptions = executionOptions;
    this.combinedCache = combinedCache;
    this.remoteExecutor = remoteExecutor;
    this.merkleTreeComputer =
        new MerkleTreeComputer(
            digestUtil,
            // Merkle trees only need to be uploaded for actions that are executed remotely.
            combinedCache instanceof RemoteExecutionCache remoteExecutionCache
                ? remoteExecutionCache
                : null,
            buildRequestId,
            commandId,
            workspaceName);

    this.scrubber = remoteOptions.getScrubber();

    this.tempPathGenerator = tempPathGenerator;
    this.captureCorruptedOutputsDir = captureCorruptedOutputsDir;

    this.remoteOutputChecker = remoteOutputChecker;
    this.outputService = outputService;
    this.knownMissingCasDigests = knownMissingCasDigests;
    this.wasRewound = wasRewound;
  }

  /**
   * Returns the spawn's output files excluding the output into which its standard output is
   * redirected, if any.
   */
  private static Collection<? extends ActionInput> outputsExcludingStdout(Spawn spawn) {
    Artifact stdoutOutput = spawn.getStdout();
    if (stdoutOutput == null) {
      return spawn.getOutputFiles();
    }
    return Collections2.filter(spawn.getOutputFiles(), artifact -> !artifact.equals(stdoutOutput));
  }

  private Command buildCommand(
      boolean useOutputPaths,
      Collection<? extends ActionInput> outputs,
      List<String> arguments,
      ImmutableMap<String, String> env,
      @Nullable Platform platform,
      RemotePathResolver remotePathResolver,
      @Nullable SpawnScrubber spawnScrubber,
      @Nullable PlatformInfo executionPlatform) {
    Command.Builder command = Command.newBuilder();
    if (useOutputPaths) {
      var outputPaths = new ArrayList<String>();
      for (ActionInput output : outputs) {
        String pathString = internalToUnicode(remotePathResolver.localPathToOutputPath(output));
        outputPaths.add(pathString);
      }
      outputPaths.sort(PROTO_STRING_COMPARATOR);
      command.addAllOutputPaths(outputPaths);
    } else {
      var outputFiles = new ArrayList<String>();
      var outputDirectories = new ArrayList<String>();
      for (ActionInput output : outputs) {
        String pathString = internalToUnicode(remotePathResolver.localPathToOutputPath(output));
        if (output.isDirectory()) {
          outputDirectories.add(pathString);
        } else {
          outputFiles.add(pathString);
        }
      }
      outputFiles.sort(PROTO_STRING_COMPARATOR);
      outputDirectories.sort(PROTO_STRING_COMPARATOR);
      command.addAllOutputFiles(outputFiles).addAllOutputDirectories(outputDirectories);
    }

    if (platform != null) {
      command.setPlatform(platform);
    }
    boolean first = true;
    for (String arg : arguments) {
      if (spawnScrubber != null) {
        arg = spawnScrubber.transformArgument(arg);
      }
      if (first && executionPlatform != null) {
        first = false;
        OS executionOs = getOsFromConstraintsOrHost(executionPlatform);
        arg = OsPathPolicy.of(executionOs).postProcessPathStringForExecution(arg);
      }
      command.addArguments(internalToUnicode(arg));
    }
    // Sorting the environment pairs by variable name.
    TreeSet<String> variables = new TreeSet<>(env.keySet());
    for (String var : variables) {
      command
          .addEnvironmentVariablesBuilder()
          .setName(internalToUnicode(var))
          .setValue(internalToUnicode(env.get(var)));
    }

    return command.build();
  }

  private boolean useRemoteCache() {
    return combinedCache != null && combinedCache.hasRemoteCache();
  }

  private boolean useDiskCache() {
    return combinedCache != null && combinedCache.hasDiskCache();
  }

  public CachePolicy getReadCachePolicy(Spawn spawn) {
    if (combinedCache == null) {
      return CachePolicy.NO_CACHE;
    }

    boolean allowRemoteCache =
        useRemoteCache()
            && remoteOptions.getRemoteAcceptCached()
            && Spawns.mayBeCachedRemotely(spawn);
    boolean allowDiskCache = useDiskCache() && Spawns.mayBeCached(spawn);

    return CachePolicy.create(allowRemoteCache, allowDiskCache);
  }

  public CachePolicy getWriteCachePolicy(Spawn spawn) {
    if (combinedCache == null) {
      return CachePolicy.NO_CACHE;
    }

    boolean allowRemoteCache =
        useRemoteCache()
            && shouldUploadLocalResultsToRemoteCache(remoteOptions, spawn.getExecutionInfo())
            && combinedCache.remoteActionCacheSupportsUpdate();
    boolean allowDiskCache = useDiskCache() && Spawns.mayBeCached(spawn);

    return CachePolicy.create(allowRemoteCache, allowDiskCache);
  }

  private static boolean shouldUploadLocalResultsToRemoteCache(
      RemoteOptions remoteOptions, Map<String, String> executionInfo) {
    return remoteOptions.getRemoteUploadLocalResults()
        && Spawns.mayBeCachedRemotely(executionInfo)
        && !executionInfo.containsKey(ExecutionRequirements.NO_REMOTE_CACHE_UPLOAD);
  }

  /** Returns {@code true} if the spawn may be executed remotely. */
  public boolean mayBeExecutedRemotely(Spawn spawn) {
    return combinedCache instanceof RemoteExecutionCache
        && remoteExecutor != null
        && Spawns.mayBeExecutedRemotely(spawn)
        && !isScrubbedSpawn(spawn, scrubber);
  }

  @Nullable
  private static ByteString buildSalt(Spawn spawn, @Nullable SpawnScrubber spawnScrubber) {
    CacheSalt.Builder saltBuilder =
        CacheSalt.newBuilder().setMayBeExecutedRemotely(Spawns.mayBeExecutedRemotely(spawn));

    String workspace =
        spawn.getExecutionInfo().get(ExecutionRequirements.DIFFERENTIATE_WORKSPACE_CACHE);
    if (workspace != null) {
      saltBuilder.setWorkspace(workspace);
    }

    if (spawnScrubber != null) {
      saltBuilder.setScrubSalt(
          CacheSalt.ScrubSalt.newBuilder().setSalt(spawnScrubber.getSalt()).build());
    }

    return saltBuilder.build().toByteString();
  }

  /**
   * Semaphore for limiting the concurrent number of Merkle tree input roots we compute and keep in
   * memory.
   *
   * <p>When --jobs is set to a high value to let the remote execution service runs many actions in
   * parallel, there is no point in letting the local system compute Merkle trees of input roots
   * with the same amount of parallelism. Not only does this make Bazel feel sluggish and slow to
   * respond to being interrupted, it causes it to exhaust memory.
   *
   * <p>As there is no point in letting Merkle tree input root computation use a higher concurrency
   * than the number of CPUs in the system, use a semaphore to limit the concurrency of
   * buildRemoteAction().
   */
  private final Semaphore remoteActionBuildingSemaphore =
      new Semaphore(Runtime.getRuntime().availableProcessors(), true);

  @Nullable
  private ToolSignature getToolSignature(Spawn spawn, SpawnExecutionContext context)
      throws IOException, ExecException, InterruptedException {
    return remoteOptions.getMarkToolInputs()
            && Spawns.supportsWorkers(spawn)
            && !spawn.getToolFiles().isEmpty()
        ? computePersistentWorkerSignature(spawn, context)
        : null;
  }

  private void maybeAcquireRemoteActionBuildingSemaphore(ProfilerTask task)
      throws InterruptedException {
    if (!remoteOptions.getThrottleRemoteActionBuilding()) {
      return;
    }

    try (var c = Profiler.instance().profile(task, "acquiring semaphore")) {
      remoteActionBuildingSemaphore.acquire();
    }
  }

  private void maybeReleaseRemoteActionBuildingSemaphore() {
    if (!remoteOptions.getThrottleRemoteActionBuilding()) {
      return;
    }

    remoteActionBuildingSemaphore.release();
  }

  private boolean useOutputPaths() {
    if (this.useOutputPaths == null) {
      initUseOutputPaths();
    }
    return this.useOutputPaths;
  }

  private synchronized void initUseOutputPaths() {
    // If this has already been initialized, return
    if (this.useOutputPaths != null) {
      return;
    }
    ApiVersion serverHighestVersion = null;
    try {
      // If both Remote Executor and Remote Cache are configured,
      // use the highest version supported by both.

      ClientApiVersion.ServerSupportedStatus executorSupportStatus = null;
      if (remoteExecutor != null) {
        var serverCapabilities = remoteExecutor.getServerCapabilities();
        if (serverCapabilities != null) {
          executorSupportStatus =
              ClientApiVersion.current.checkServerSupportedVersions(serverCapabilities);
        }
      }

      ClientApiVersion.ServerSupportedStatus cacheSupportStatus = null;
      if (combinedCache != null) {
        var serverCapabilities = combinedCache.getRemoteServerCapabilities();
        if (serverCapabilities != null) {
          cacheSupportStatus =
              ClientApiVersion.current.checkServerSupportedVersions(serverCapabilities);
        }
      }

      ApiVersion executorHighestVersion = null;
      if (executorSupportStatus != null && executorSupportStatus.isSupported()) {
        executorHighestVersion = executorSupportStatus.getHighestSupportedVersion();
      }

      ApiVersion cacheHighestVersion = null;
      if (cacheSupportStatus != null && cacheSupportStatus.isSupported()) {
        cacheHighestVersion = cacheSupportStatus.getHighestSupportedVersion();
      }

      if (executorHighestVersion != null && cacheHighestVersion != null) {
        serverHighestVersion = min(ImmutableList.of(executorHighestVersion, cacheHighestVersion));
      } else if (executorHighestVersion != null) {
        serverHighestVersion = executorHighestVersion;
      } else if (cacheHighestVersion != null) {
        serverHighestVersion = cacheHighestVersion;
      }
    } catch (IOException e) {
      // Intentionally ignored.
    }
    this.useOutputPaths =
        serverHighestVersion == null || serverHighestVersion.compareTo(ApiVersion.twoPointOne) >= 0;
  }

  @VisibleForTesting
  RemoteAction buildRemoteAction(Spawn spawn, SpawnExecutionContext context)
      throws IOException, ExecException, InterruptedException {
    return buildRemoteAction(
        spawn,
        context,
        remoteOptions.getRemoteDiscardMerkleTrees()
            ? MerkleTreeComputer.BlobPolicy.DISCARD
            : MerkleTreeComputer.BlobPolicy.KEEP_AND_REUPLOAD);
  }

  /** Creates a new {@link RemoteAction} instance from spawn. */
  public RemoteAction buildRemoteAction(
      Spawn spawn, SpawnExecutionContext context, MerkleTreeComputer.BlobPolicy blobPolicy)
      throws IOException, ExecException, InterruptedException {
    maybeAcquireRemoteActionBuildingSemaphore(ProfilerTask.REMOTE_SETUP);
    try {
      // Create a remote path resolver that is aware of the spawn's path mapper, which rewrites
      // the paths of the inputs and outputs as well as paths appearing in the command line for
      // execution. This is necessary to ensure that artifacts are correctly emitted into and staged
      // from the unmapped location locally.
      RemotePathResolver remotePathResolver =
          RemotePathResolver.createMapped(baseRemotePathResolver, execRoot, spawn.getPathMapper());
      ToolSignature toolSignature = getToolSignature(spawn, context);
      MerkleTree merkleTree;
      try {
        merkleTree =
            merkleTreeComputer.buildForSpawn(
                spawn,
                toolSignature != null ? toolSignature.toolInputs : ImmutableSet.of(),
                scrubber,
                context,
                blobPolicy);
      } catch (CredentialHelperException e) {
        throw createExecExceptionForCredentialHelperException(e);
      }

      // Get the remote platform properties.
      Platform platform;
      ImmutableMap.Builder<String, String> additionalPropertiesBuilder = ImmutableMap.builder();
      if (toolSignature != null) {
        additionalPropertiesBuilder.put(
            PlatformProperties.PERSISTENT_WORKER_KEY, toolSignature.key);
        if (spawn.getExecutionInfo().containsKey(ExecutionRequirements.REQUIRES_WORKER_PROTOCOL)) {
          additionalPropertiesBuilder.put(
              PlatformProperties.PERSISTENT_WORKER_PROTOCOL,
              spawn.getExecutionInfo().get(ExecutionRequirements.REQUIRES_WORKER_PROTOCOL));
        }
      }
      platform =
          PlatformUtils.getPlatformProto(spawn, remoteOptions, additionalPropertiesBuilder.build());

      SpawnScrubber spawnScrubber = scrubber != null ? scrubber.forSpawn(spawn) : null;
      Command command =
          buildCommand(
              useOutputPaths(),
              outputsExcludingStdout(spawn),
              spawn.getArguments(),
              spawn.getEnvironment(),
              platform,
              remotePathResolver,
              spawnScrubber,
              spawn.getExecutionPlatform());
      Digest commandHash = digestUtil.compute(command);
      Action action =
          Utils.buildAction(
              commandHash,
              merkleTree.digest(),
              platform,
              context.getTimeout(),
              Spawns.mayBeCachedRemotely(spawn),
              buildSalt(spawn, spawnScrubber));

      ActionKey actionKey = digestUtil.computeActionKey(action);

      ActionExecutionMetadata actionMetadata = spawn.getResourceOwner();
      RequestMetadata metadata =
          TracingMetadataUtils.buildMetadata(
              buildRequestId,
              commandId,
              actionKey.digest().getHash(),
              actionMetadata != null ? actionMetadata.getMnemonic() : null,
              actionMetadata != null && actionMetadata.getOwner().getLabel() != null
                  ? actionMetadata.getOwner().getLabel().getCanonicalForm()
                  : null,
              actionMetadata != null ? actionMetadata.getOwner().getConfigurationChecksum() : null);
      RemoteActionExecutionContext remoteActionExecutionContext =
          RemoteActionExecutionContext.create(
              spawn, context, metadata, getWriteCachePolicy(spawn), getReadCachePolicy(spawn));
      return new RemoteAction(
          spawn,
          context,
          remoteActionExecutionContext,
          remotePathResolver,
          merkleTree,
          commandHash,
          command,
          action,
          actionKey);
    } finally {
      maybeReleaseRemoteActionBuildingSemaphore();
    }
  }

  private ToolSignature computePersistentWorkerSignature(Spawn spawn, SpawnExecutionContext context)
      throws IOException, ExecException, InterruptedException {
    WorkerParser workerParser =
        new WorkerParser(
            execRoot, Options.getDefaults(WorkerOptions.class), LocalEnvProvider.NOOP, null);
    WorkerKey workerKey = workerParser.compute(spawn, context).getWorkerKey();
    Fingerprint fingerprint = new Fingerprint();
    // getWorkerFilesCombinedHash always uses SHA-256, so the hash is always 32 bytes.
    fingerprint.addBytes(workerKey.getWorkerFilesCombinedHash().asBytes());
    fingerprint.addStrings(workerKey.getArgs());
    fingerprint.addStringMap(workerKey.getEnv());
    return new ToolSignature(
        fingerprint.hexDigestAndReset(), workerKey.getWorkerFilesWithDigests().keySet());
  }

  /** A value class representing the result of remotely executed {@link RemoteAction}. */
  public static class RemoteActionResult {
    private final ActionResult actionResult;
    @Nullable private final ExecuteResponse executeResponse;
    @Nullable private final String cacheName;
    @Nullable private ActionResultMetadata metadata;

    /** Creates a new {@link RemoteActionResult} instance from a cached result. */
    public static RemoteActionResult createFromCache(CachedActionResult cachedActionResult) {
      checkArgument(cachedActionResult != null, "cachedActionResult is null");
      return new RemoteActionResult(
          cachedActionResult.actionResult(), null, cachedActionResult.cacheName());
    }

    /** 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);
    }

    public RemoteActionResult(
        ActionResult actionResult,
        @Nullable ExecuteResponse executeResponse,
        @Nullable String cacheName) {
      this.actionResult = actionResult;
      this.executeResponse = executeResponse;
      this.cacheName = cacheName;
    }

    /** Returns the exit code of remote executed action. */
    public int getExitCode() {
      return actionResult.getExitCode();
    }

    public List<OutputFile> getOutputFiles() {
      return actionResult.getOutputFilesList();
    }

    public List<OutputSymlink> getOutputFileSymlinks() {
      return actionResult.getOutputFileSymlinksList();
    }

    public List<OutputDirectory> getOutputDirectories() {
      return actionResult.getOutputDirectoriesList();
    }

    public ActionResultMetadata getOrParseActionResultMetadata(
        CombinedCache combinedCache,
        DigestUtil digestUtil,
        RemoteActionExecutionContext context,
        RemotePathResolver remotePathResolver)
        throws IOException, InterruptedException {
      if (metadata == null) {
        try (SilentCloseable c = Profiler.instance().profile("Remote.parseActionResultMetadata")) {
          metadata =
              parseActionResultMetadata(
                  combinedCache, digestUtil, context, actionResult, remotePathResolver);
        }
      }
      return metadata;
    }

    public List<OutputSymlink> getOutputDirectorySymlinks() {
      return actionResult.getOutputDirectorySymlinksList();
    }

    public List<OutputSymlink> getOutputSymlinks() {
      return actionResult.getOutputSymlinksList();
    }

    /**
     * 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.
     */
    public String getMessage() {
      return executeResponse != null ? executeResponse.getMessage() : "";
    }

    /** Returns the details of the execution that originally produced this result. */
    public ExecutedActionMetadata getExecutionMetadata() {
      return actionResult.getExecutionMetadata();
    }

    /** Returns whether the action is executed successfully. */
    public boolean success() {
      if (executeResponse != null) {
        if (executeResponse.getStatus().getCode() != Code.OK.value()) {
          return false;
        }
      }

      return actionResult.getExitCode() == 0;
    }

    /**
     * Returns an output that is mandatory for the given spawn but missing from this result, if any.
     */
    public Optional<? extends ActionInput> maybeGetMissingMandatoryOutput(RemoteAction action) {
      var outputFiles = Iterables.transform(getOutputFiles(), OutputFile::getPath);
      var outputDirPaths = Iterables.transform(getOutputDirectories(), OutputDirectory::getPath);
      var outputSymlinkPaths =
          Iterables.transform(
              Iterables.concat(
                  getOutputSymlinks(), getOutputFileSymlinks(), getOutputDirectorySymlinks()),
              OutputSymlink::getPath);
      ImmutableSet<String> allOutputPaths =
          ImmutableSet.copyOf(
              Iterables.transform(
                  Iterables.concat(outputFiles, outputDirPaths, outputSymlinkPaths),
                  StringEncoding::unicodeToInternal));
      // Check that all mandatory outputs are created. The stdout output is captured as the action
      // result's stdout rather than as a regular output file of the RemoteAction, so it has to be
      // excluded here.
      var spawn = action.getSpawn();
      var remotePathResolver = action.getRemotePathResolver();
      return outputsExcludingStdout(spawn).stream()
          .filter(spawn::isMandatoryOutput)
          .filter(
              output -> !allOutputPaths.contains(remotePathResolver.localPathToOutputPath(output)))
          .findFirst();
    }

    /** Returns {@code true} if this result is from a cache. */
    public boolean cacheHit() {
      if (executeResponse == null) {
        return true;
      }

      return executeResponse.getCachedResult();
    }

    /** Returns cache name (disk/remote) when {@code cacheHit()} or {@code null} when not */
    @Nullable
    public String cacheName() {
      return cacheName;
    }

    /**
     * Returns the underlying {@link ExecuteResponse} or {@code null} if this result is from a
     * cache.
     */
    @Nullable
    public ExecuteResponse getResponse() {
      return executeResponse;
    }

    @Override
    public boolean equals(Object object) {
      if (!(object instanceof RemoteActionResult that)) {
        return false;
      }

      return Objects.equals(actionResult, that.actionResult)
          && Objects.equals(executeResponse, that.executeResponse);
    }

    @Override
    public int hashCode() {
      return Objects.hash(actionResult, executeResponse);
    }
  }

  /** Lookup the remote cache for the given {@link RemoteAction}. {@code null} if not found. */
  @Nullable
  public RemoteActionResult lookupCache(RemoteAction action)
      throws IOException, InterruptedException {
    checkState(
        action.getRemoteActionExecutionContext().getReadCachePolicy().allowAnyCache(),
        "spawn doesn't accept cached result");

    // A rewound action must regenerate its lost outputs even if the cache still serves its stale
    // action result. Treat this as a cache miss.
    // TODO(https://github.com/bazelbuild/remote-apis/pull/386): Allow cache lookup for rewound
    // actions when CacheCapabilities.verifies_action_results is advertised, while still bypassing
    // unverified cache layers.
    if (wasRewound.test(action.getSpawn().getResourceOwner())) {
      return null;
    }

    ImmutableSet<String> inlineOutputFiles = ImmutableSet.of();
    PathFragment inMemoryOutputPath = getInMemoryOutputPath(action.getSpawn());
    if (inMemoryOutputPath != null) {
      inlineOutputFiles =
          ImmutableSet.of(action.getRemotePathResolver().localPathToOutputPath(inMemoryOutputPath));
    }

    CachedActionResult cachedActionResult =
        combinedCache.downloadActionResult(
            action.getRemoteActionExecutionContext(),
            action.getActionKey(),
            /* inlineOutErr= */ false,
            inlineOutputFiles);

    if (cachedActionResult == null) {
      return null;
    }

    var result = RemoteActionResult.createFromCache(cachedActionResult);

    // The legacy whole-invocation retry path uses knownMissingCasDigests to prevent a retried
    // invocation from accepting the same stale action result. Action rewinding bypasses cache
    // lookup above without populating this set.
    if (!knownMissingCasDigests.isEmpty()) {
      ActionResultMetadata metadata;
      try {
        metadata =
            result.getOrParseActionResultMetadata(
                combinedCache,
                digestUtil,
                action.getRemoteActionExecutionContext(),
                action.getRemotePathResolver());
      } catch (BulkTransferException e) {
        if (!e.allCausedByCacheNotFoundException()) {
          throw e;
        }
        return null; // Handle dangling reference to lost Tree message as AC miss.
      }

      // If we already know digests referenced by this AC is missing from remote cache, ignore it so
      // that we can fall back to execution. This could happen when the remote cache is an HTTP
      // cache, or doesn't implement AC integrity check.
      //
      // See https://github.com/bazelbuild/bazel/issues/18696.
      if (updateKnownMissingCasDigests(knownMissingCasDigests, metadata)) {
        return null;
      }
    }

    return result;
  }

  /**
   * Returns the (exec root relative) path of a spawn output that should be made available via
   * {@link SpawnResult#getInMemoryOutput(ActionInput)}.
   */
  @Nullable
  private static PathFragment getInMemoryOutputPath(Spawn spawn) {
    String outputPath =
        spawn.getExecutionInfo().get(ExecutionRequirements.REMOTE_EXECUTION_INLINE_OUTPUTS);
    if (outputPath != null) {
      return PathFragment.create(outputPath);
    }
    return null;
  }

  /**
   * Removes digests referenced by {@code metadata} from {@code knownMissingCasDigests} and returns
   * whether any were removed
   */
  private static boolean updateKnownMissingCasDigests(
      Set<Digest> knownMissingCasDigests, ActionResultMetadata metadata) {
    // Using `remove` below because we assume the missing blob will be uploaded afterwards.
    var result = false;
    for (var file : metadata.files()) {
      if (knownMissingCasDigests.remove(file.digest())) {
        result = true;
      }
    }
    for (var entry : metadata.directories()) {
      for (var file : entry.getValue().files()) {
        if (knownMissingCasDigests.remove(file.digest())) {
          result = true;
        }
      }
    }
    return result;
  }

  private ListenableFuture<FileMetadata> downloadFile(
      RemoteActionExecutionContext context,
      ProgressStatusListener progressStatusListener,
      FileMetadata file,
      Path tmpPath,
      RemotePathResolver remotePathResolver) {
    checkNotNull(combinedCache, "combinedCache can't be null");

    try {
      ListenableFuture<Void> future =
          combinedCache.downloadFile(
              context,
              internalToUnicode(remotePathResolver.localPathToOutputPath(file.path())),
              remotePathResolver.localPathToExecPath(file.path().asFragment()),
              tmpPath,
              /* finalPath= */ file.path(),
              file.digest(),
              new CombinedCache.DownloadProgressReporter(
                  progressStatusListener,
                  internalToUnicode(remotePathResolver.localPathToOutputPath(file.path())),
                  file.digest().getSizeBytes()));
      return transform(future, (d) -> file, directExecutor());
    } catch (IOException e) {
      return immediateFailedFuture(e);
    }
  }

  private void captureCorruptedOutputs(Exception e) {
    if (captureCorruptedOutputsDir != null) {
      if (e instanceof BulkTransferException) {
        for (Throwable suppressed : e.getSuppressed()) {
          if (suppressed instanceof OutputDigestMismatchException) {
            // Capture corrupted outputs
            try {
              String outputPath = ((OutputDigestMismatchException) suppressed).getOutputPath();
              Path localPath = ((OutputDigestMismatchException) suppressed).getLocalPath();
              Path dst = captureCorruptedOutputsDir.getRelative(outputPath);
              dst.createDirectoryAndParents();

              // Make sure dst is still under captureCorruptedOutputsDir, otherwise
              // IllegalArgumentException will be thrown.
              dst.relativeTo(captureCorruptedOutputsDir);

              FileSystemUtils.copyFile(localPath, dst);
            } catch (Exception ee) {
              e.addSuppressed(ee);
            }
          }
        }
      }
    }
  }

  private void deletePartialDownloadedOutputs(
      Map<Path, Path> realToTmpPath, FileOutErr tmpOutErr, Exception e) throws ExecException {
    try {
      // Delete any (partially) downloaded output files.
      for (Path tmpPath : realToTmpPath.values()) {
        tmpPath.delete();
      }

      tmpOutErr.clearOut();
      tmpOutErr.clearErr();
    } catch (IOException ioEx) {
      ioEx.addSuppressed(e);

      // If deleting of output files failed, we abort the build with a decent error message as
      // any subsequent local execution failure would likely be incomprehensible.
      ExecException execEx =
          new EnvironmentalExecException(
              ioEx,
              createFailureDetail(
                  "Failed to delete output files after incomplete download",
                  RemoteExecution.Code.INCOMPLETE_OUTPUT_DOWNLOAD_CLEANUP_FAILURE));
      execEx.addSuppressed(e);
      throw execEx;
    }
  }

  /** Moves the locally created outputs from their temporary location to their declared location. */
  private void moveOutputsToFinalLocation(
      Iterable<Path> localOutputs, Map<Path, Path> realToTmpPath) throws IOException {
    // Move the output files from their temporary name to the actual output file name. Executable
    // bit is ignored since the file permission will be changed to 0555 after execution.
    for (Path realPath : localOutputs) {
      Path tmpPath = Preconditions.checkNotNull(realToTmpPath.get(realPath));
      realPath.getParentDirectory().createDirectoryAndParents();
      FileSystemUtils.moveFile(tmpPath, realPath);
    }
  }

  private void createSymlinks(Iterable<SymlinkMetadata> symlinks) throws IOException {
    for (SymlinkMetadata symlink : symlinks) {
      // Ensure the symlink is materialized inside the exec root. The local path is derived from an
      // ActionResult output symlink path via execRoot.getRelative(...), which returns an absolute
      // path verbatim, so without this check an absolute (or otherwise escaping) symlink path would
      // be created outside the output tree. Output files already enforce containment via
      // file.path.relativeTo(execRoot) in downloadOutputs(); apply the same guarantee to output
      // symlinks (including tree-nested symlinks, which also flow through here).
      if (!symlink.path().startsWith(execRoot)) {
        throw new IOException(
            String.format(
                "Failed to create symlink %s: the output path escapes the output tree %s",
                symlink.path(), execRoot));
      }
      Preconditions.checkNotNull(
              symlink.path().getParentDirectory(),
              "Failed creating directory and parents for %s",
              symlink.path())
          .createDirectoryAndParents();
      // If a directory output is being materialized as a symlink, creating the symlink fails as we
      // must first delete the preexisting empty directory. Since this is rare (and in the future
      // BwoB may no longer eagerly create these directories), we don't delete the directory
      // beforehand.
      try {
        symlink.path().createSymbolicLink(symlink.target());
      } catch (IOException e) {
        if (!symlink.path().isDirectory(Symlinks.NOFOLLOW)) {
          throw e;
        }
        // Retry after deleting the directory.
        symlink.path().delete();
        symlink.path().createSymbolicLink(symlink.target());
      }
    }
  }

  /** In-memory representation of action result metadata. */
  static class ActionResultMetadata {

    static class SymlinkMetadata {
      private final Path path;
      private final PathFragment target;

      private SymlinkMetadata(Path path, PathFragment target) {
        this.path = path;
        this.target = target;
      }

      public Path path() {
        return path;
      }

      public PathFragment target() {
        return target;
      }
    }

    public static class FileMetadata {
      private final Path path;
      private final Digest digest;
      private final boolean isExecutable;
      private final ByteString contents;

      private FileMetadata(Path path, Digest digest, boolean isExecutable, ByteString contents) {
        this.path = path;
        this.digest = digest;
        this.isExecutable = isExecutable;
        this.contents = contents;
      }

      public Path path() {
        return path;
      }

      public Digest digest() {
        return digest;
      }

      public boolean isExecutable() {
        return isExecutable;
      }

      public ByteString content() {
        return contents;
      }
    }

    static class DirectoryMetadata {
      private final ImmutableList<FileMetadata> files;
      private final ImmutableList<SymlinkMetadata> symlinks;

      private DirectoryMetadata(
          ImmutableList<FileMetadata> files, ImmutableList<SymlinkMetadata> symlinks) {
        this.files = files;
        this.symlinks = symlinks;
      }

      public ImmutableList<FileMetadata> files() {
        return files;
      }

      public ImmutableList<SymlinkMetadata> symlinks() {
        return symlinks;
      }
    }

    private final ImmutableMap<Path, FileMetadata> files;
    private final ImmutableMap<Path, SymlinkMetadata> symlinks;
    private final ImmutableMap<Path, DirectoryMetadata> directories;

    private ActionResultMetadata(
        ImmutableMap<Path, FileMetadata> files,
        ImmutableMap<Path, SymlinkMetadata> symlinks,
        ImmutableMap<Path, DirectoryMetadata> directories) {
      this.files = files;
      this.symlinks = symlinks;
      this.directories = directories;
    }

    @Nullable
    public FileMetadata file(Path path) {
      return files.get(path);
    }

    @Nullable
    public DirectoryMetadata directory(Path path) {
      return directories.get(path);
    }

    public Collection<FileMetadata> files() {
      return files.values();
    }

    public ImmutableSet<Entry<Path, DirectoryMetadata>> directories() {
      return directories.entrySet();
    }

    public Collection<SymlinkMetadata> symlinks() {
      return symlinks.values();
    }
  }

  private static void validatePathComponent(String name) throws IOException {
    if (name.isEmpty() || name.contains("/") || name.equals(".") || name.equals("..")) {
      throw new IOException("Malformed path component: " + name);
    }
  }

  private static DirectoryMetadata parseDirectory(
      Path parent, Directory dir, Map<Digest, Directory> childDirectoriesMap) throws IOException {
    ImmutableList.Builder<FileMetadata> filesBuilder = ImmutableList.builder();
    for (FileNode file : dir.getFilesList()) {
      String name = unicodeToInternal(file.getName());
      validatePathComponent(name);
      filesBuilder.add(
          new FileMetadata(
              parent.getRelative(name),
              file.getDigest(),
              file.getIsExecutable(),
              ByteString.EMPTY));
    }

    ImmutableList.Builder<SymlinkMetadata> symlinksBuilder = ImmutableList.builder();
    for (SymlinkNode symlink : dir.getSymlinksList()) {
      String name = unicodeToInternal(symlink.getName());
      validatePathComponent(name);
      symlinksBuilder.add(
          new SymlinkMetadata(
              parent.getRelative(name),
              PathFragment.create(unicodeToInternal(symlink.getTarget()))));
    }

    for (DirectoryNode directoryNode : dir.getDirectoriesList()) {
      String name = unicodeToInternal(directoryNode.getName());
      validatePathComponent(name);
      Path childPath = parent.getRelative(name);
      Directory childDir =
          Preconditions.checkNotNull(childDirectoriesMap.get(directoryNode.getDigest()));
      DirectoryMetadata childMetadata = parseDirectory(childPath, childDir, childDirectoriesMap);
      filesBuilder.addAll(childMetadata.files());
      symlinksBuilder.addAll(childMetadata.symlinks());
    }

    return new DirectoryMetadata(filesBuilder.build(), symlinksBuilder.build());
  }

  // The Tree message representing an empty directory.
  private static final Tree EMPTY_DIRECTORY =
      Tree.newBuilder().setRoot(Directory.getDefaultInstance()).build();

  static {
    // See logic in parseActionResultMetadata below.
    Preconditions.checkState(EMPTY_DIRECTORY.toByteString().size() == 2);
  }

  static ActionResultMetadata parseActionResultMetadata(
      CombinedCache combinedCache,
      DigestUtil digestUtil,
      RemoteActionExecutionContext context,
      ActionResult result,
      RemotePathResolver remotePathResolver)
      throws IOException, InterruptedException {
    checkNotNull(combinedCache, "combinedCache can't be null");

    Map<Path, ListenableFuture<Tree>> dirMetadataDownloads =
        Maps.newHashMapWithExpectedSize(result.getOutputDirectoriesCount());
    for (OutputDirectory dir : result.getOutputDirectoriesList()) {
      var outputPath = dir.getPath();
      var localPath = remotePathResolver.outputPathToLocalPath(unicodeToInternal(outputPath));
      if (dir.getTreeDigest().getSizeBytes() == 2) {
        // A valid Tree message contains at least a non-empty root field. The only way for a Tree
        // message to have a size of 2 bytes is if the root field is the only non-empty field and
        // the Directory message in the root field is empty, which corresponds to one byte for the
        // LEN tag and field number and one byte for the zero-length varint. Since empty tree
        // artifacts are relatively common (e.g., as the undeclared test output directory), we avoid
        // downloading these messages here.
        dirMetadataDownloads.put(localPath, immediateFuture(EMPTY_DIRECTORY));
      } else {
        dirMetadataDownloads.put(
            localPath,
            Futures.transformAsync(
                combinedCache.downloadBlobAsByteString(
                    context,
                    outputPath,
                    remotePathResolver.localPathToExecPath(localPath.asFragment()),
                    dir.getTreeDigest()),
                (treeBytes) ->
                    immediateFuture(
                        Tree.parseFrom(treeBytes, ExtensionRegistry.getEmptyRegistry())),
                directExecutor()));
      }
    }

    waitForBulkTransfer(dirMetadataDownloads.values());

    ImmutableMap.Builder<Path, DirectoryMetadata> directories = ImmutableMap.builder();
    for (Map.Entry<Path, ListenableFuture<Tree>> metadataDownload :
        dirMetadataDownloads.entrySet()) {
      Path path = metadataDownload.getKey();
      Tree directoryTree = getFromFuture(metadataDownload.getValue());
      Map<Digest, Directory> childrenMap = new HashMap<>();
      for (Directory childDir : directoryTree.getChildrenList()) {
        childrenMap.put(digestUtil.compute(childDir), childDir);
      }

      directories.put(path, parseDirectory(path, directoryTree.getRoot(), childrenMap));
    }

    ImmutableMap.Builder<Path, FileMetadata> files = ImmutableMap.builder();
    for (OutputFile outputFile : result.getOutputFilesList()) {
      Path localPath =
          remotePathResolver.outputPathToLocalPath(unicodeToInternal(outputFile.getPath()));
      files.put(
          localPath,
          new FileMetadata(
              localPath,
              outputFile.getDigest(),
              outputFile.getIsExecutable(),
              outputFile.getContents()));
    }

    Artifact stdoutArtifact = context.getSpawn().getStdout();
    if (stdoutArtifact != null) {
      Path localPath = stdoutArtifact.getPath();
      files.put(
          localPath,
          new FileMetadata(
              localPath,
              result.getStdoutDigest(),
              /* isExecutable= */ false,
              result.getStdoutRaw()));
    }

    var symlinkMap = new HashMap<Path, SymlinkMetadata>();
    var outputSymlinks =
        Iterables.concat(
            result.getOutputFileSymlinksList(),
            result.getOutputDirectorySymlinksList(),
            result.getOutputSymlinksList());
    for (var symlink : outputSymlinks) {
      var localPath =
          remotePathResolver.outputPathToLocalPath(unicodeToInternal(symlink.getPath()));
      var target = PathFragment.create(unicodeToInternal(symlink.getTarget()));
      var existingMetadata = symlinkMap.get(localPath);
      if (existingMetadata != null) {
        if (!target.equals(existingMetadata.target())) {
          throw new IOException(
              String.format(
                  "Symlink path collision: '%s' is mapped to both '%s' and '%s'. Action Result"
                      + " should not contain multiple targets for the same symlink.",
                  localPath, existingMetadata.target(), target));
        }
        continue;
      }

      symlinkMap.put(localPath, new SymlinkMetadata(localPath, target));
    }

    return new ActionResultMetadata(
        files.buildOrThrow(), ImmutableMap.copyOf(symlinkMap), directories.buildOrThrow());
  }

  /**
   * Downloads the outputs of a remotely executed action and injects their metadata.
   *
   * <p>For a successful action, the {@link RemoteOutputChecker} is consulted to determine which of
   * the outputs should be downloaded. For a failed action, all outputs are downloaded. The action
   * stdout and stderr, as well as the in-memory output when present, are always downloaded even in
   * the success case. Any outputs that are not downloaded have their metadata injected into the
   * {@link RemoteActionFileSystem}.
   *
   * <p>In case of download failure, all of the already downloaded outputs are deleted.
   *
   * @return The in-memory output if the spawn had one, otherwise null.
   */
  @Nullable
  public InMemoryOutput downloadOutputs(RemoteAction action, RemoteActionResult result)
      throws InterruptedException, IOException, ExecException {
    checkState(!shutdown.get(), "shutdown");
    checkNotNull(combinedCache, "combinedCache can't be null");

    RemoteActionFileSystem remoteActionFileSystem = null;
    boolean hasBazelOutputService = outputService instanceof BazelOutputService;
    if (!hasBazelOutputService) {
      FileSystem actionFileSystem = action.getSpawnExecutionContext().getActionFileSystem();
      checkState(
          actionFileSystem instanceof RemoteActionFileSystem,
          "expected the ActionFileSystem to be a RemoteActionFileSystem");
      remoteActionFileSystem = (RemoteActionFileSystem) actionFileSystem;
    }

    ProgressStatusListener progressStatusListener = action.getSpawnExecutionContext()::report;
    RemoteActionExecutionContext context = action.getRemoteActionExecutionContext();
    if (result.executeResponse != null) {
      // Always read from remote cache for just remotely executed action.
      context = context.withReadCachePolicy(context.getReadCachePolicy().addRemoteCache());
    }

    ActionResultMetadata metadata =
        result.getOrParseActionResultMetadata(
            combinedCache, digestUtil, context, action.getRemotePathResolver());

    // The expiration time for remote cache entries.
    var expirationTime = Instant.now().plus(remoteOptions.getRemoteCacheTtl());

    ActionInput inMemoryOutput = null;
    AtomicReference<ByteString> inMemoryOutputData = new AtomicReference<>(null);
    PathFragment inMemoryOutputPath = getInMemoryOutputPath(action.getSpawn());
    if (inMemoryOutputPath != null) {
      for (ActionInput output : action.getSpawn().getOutputFiles()) {
        if (output.getExecPath().equals(inMemoryOutputPath)) {
          inMemoryOutput = output;
          break;
        }
      }
    }

    // Collect the set of files to download.
    ImmutableList.Builder<ListenableFuture<FileMetadata>> downloadsBuilder =
        ImmutableList.builder();

    // Download into temporary paths, then move everything at the end.
    // This avoids holding the output lock while downloading, which would prevent the local branch
    // from completing sooner under the dynamic execution strategy.
    Map<Path, Path> realToTmpPath = new HashMap<>();

    for (FileMetadata file : metadata.files()) {
      if (realToTmpPath.containsKey(file.path)) {
        continue;
      }

      var execPath = file.path.relativeTo(execRoot);
      var isInMemoryOutputFile = inMemoryOutput != null && execPath.equals(inMemoryOutputPath);
      if (!isInMemoryOutputFile && shouldDownload(result, execPath, /* treeRootExecPath= */ null)) {
        Path tmpPath = tempPathGenerator.generateTempPath();
        realToTmpPath.put(file.path, tmpPath);
        downloadsBuilder.add(
            downloadFile(
                context, progressStatusListener, file, tmpPath, action.getRemotePathResolver()));
      } else {
        if (hasBazelOutputService) {
          downloadsBuilder.add(immediateFuture(file));
        } else {
          checkNotNull(remoteActionFileSystem)
              .injectRemoteFile(
                  file.path().asFragment(),
                  DigestUtil.toBinaryDigest(file.digest()),
                  file.digest().getSizeBytes(),
                  expirationTime,
                  isInMemoryOutputFile);
        }

        if (isInMemoryOutputFile) {
          // Download into memory only; do not write to disk.
          remoteOutputChecker.skipDownload(inMemoryOutputPath);
          if (file.contents.isEmpty()) {
            // As the contents field doesn't have presence information, we use the digest size to
            // distinguish between an empty file and one that wasn't inlined.
            if (file.digest.getSizeBytes() == 0) {
              inMemoryOutputData.set(ByteString.EMPTY);
            } else {
              downloadsBuilder.add(
                  transform(
                      combinedCache.downloadBlobAsByteString(
                          context,
                          inMemoryOutputPath.getPathString(),
                          inMemoryOutputPath,
                          file.digest()),
                      data -> {
                        inMemoryOutputData.set(data);
                        return null;
                      },
                      directExecutor()));
            }
          } else {
            inMemoryOutputData.set(file.contents);
          }
        }
      }
    }

    for (Map.Entry<Path, DirectoryMetadata> entry : metadata.directories()) {
      PathFragment treeRootExecPath = entry.getKey().relativeTo(execRoot);

      for (FileMetadata file : entry.getValue().files()) {
        if (realToTmpPath.containsKey(file.path)) {
          continue;
        }

        if (shouldDownload(result, file.path.relativeTo(execRoot), treeRootExecPath)) {
          Path tmpPath = tempPathGenerator.generateTempPath();
          realToTmpPath.put(file.path, tmpPath);
          downloadsBuilder.add(
              downloadFile(
                  context, progressStatusListener, file, tmpPath, action.getRemotePathResolver()));
        } else if (hasBazelOutputService) {
          downloadsBuilder.add(immediateFuture(file));
        } else {
          checkNotNull(remoteActionFileSystem)
              .injectRemoteFile(
                  file.path().asFragment(),
                  DigestUtil.toBinaryDigest(file.digest()),
                  file.digest().getSizeBytes(),
                  expirationTime,
                  /* inMemoryOutput= */ false);
        }
      }
    }

    FileOutErr outErr = action.getSpawnExecutionContext().getFileOutErr();

    // Always download the action stdout/stderr, except for stdout that was redirected into an
    // output above (in which case it must not be reported as regular action stdout).
    FileOutErr tmpOutErr = outErr.childOutErr();
    List<ListenableFuture<Void>> outErrDownloads =
        combinedCache.downloadOutErr(
            context,
            result.actionResult,
            tmpOutErr,
            /* downloadStdout= */ context.getSpawn().getStdout() == null);
    for (ListenableFuture<Void> future : outErrDownloads) {
      downloadsBuilder.add(transform(future, (v) -> null, directExecutor()));
    }

    ImmutableList<ListenableFuture<FileMetadata>> downloads = downloadsBuilder.build();
    try (SilentCloseable c = Profiler.instance().profile("Remote.download")) {
      waitForBulkTransfer(downloads);
    } catch (Exception e) {
      // TODO(bazel-team): Consider adding better case-by-case exception handling instead of just
      // rethrowing
      captureCorruptedOutputs(e);
      deletePartialDownloadedOutputs(realToTmpPath, tmpOutErr, e);
      throw e;
    }

    FileOutErr.dump(tmpOutErr, outErr);

    // Ensure that we are the only ones writing to the output files when using the dynamic spawn
    // strategy.
    action
        .getSpawnExecutionContext()
        .lockOutputFiles(result.getExitCode(), result.getMessage(), tmpOutErr);
    // Will these be properly garbage-collected if the above throws an exception?
    tmpOutErr.clearOut();
    tmpOutErr.clearErr();

    List<FileMetadata> finishedDownloads = new ArrayList<>(downloads.size());
    for (ListenableFuture<FileMetadata> finishedDownload : downloads) {
      FileMetadata outputFile = getFromFuture(finishedDownload);
      if (outputFile != null) {
        finishedDownloads.add(outputFile);
      }
    }

    if (hasBazelOutputService) {
      // TODO(chiwang): Stage directories directly
      ((BazelOutputService) outputService).stageArtifacts(finishedDownloads);
    } else {
      moveOutputsToFinalLocation(
          Iterables.transform(finishedDownloads, FileMetadata::path), realToTmpPath);
    }

    List<SymlinkMetadata> symlinksInDirectories = new ArrayList<>();
    for (Entry<Path, DirectoryMetadata> entry : metadata.directories()) {
      for (SymlinkMetadata symlink : entry.getValue().symlinks()) {
        symlinksInDirectories.add(symlink);
      }
    }

    Iterable<SymlinkMetadata> symlinks =
        Iterables.concat(metadata.symlinks(), symlinksInDirectories);

    // Create the symbolic links after all downloads are finished, because dangling symlinks
    // might not be supported on all platforms.
    createSymlinks(symlinks);

    if (result.success()) {
      // Check that all mandatory outputs are created.
      var missingMandatoryOutput = result.maybeGetMissingMandatoryOutput(action);
      if (missingMandatoryOutput.isPresent()) {
        throw new IOException(
            "mandatory output %s was not created"
                .formatted(prettyPrint(missingMandatoryOutput.get())));
      }

      if (result.executeResponse != null && !knownMissingCasDigests.isEmpty()) {
        // A succeeded execution uploads outputs to CAS. Refresh our knowledge about missing
        // digests.
        var unused = updateKnownMissingCasDigests(knownMissingCasDigests, metadata);
      }

      // When downloading outputs from just remotely executed action, the action result comes from
      // Execution response which means, if disk cache is enabled, action result hasn't been
      // uploaded to it. Upload action result to disk cache here so next build could hit it.
      if (useDiskCache() && result.executeResponse != null) {
        getFromFuture(
            combinedCache.uploadActionResult(
                context.withWriteCachePolicy(CachePolicy.DISK_CACHE_ONLY),
                action.getActionKey(),
                result.actionResult));
      }
    }

    if (inMemoryOutput != null && inMemoryOutputData.get() != null) {
      return new InMemoryOutput(inMemoryOutput, inMemoryOutputData.get());
    }

    return null;
  }

  /** An ongoing local execution of a spawn. */
  public static final class LocalExecution implements SilentCloseable {
    private final RemoteAction action;
    private final SettableFuture<SpawnResult> spawnResultFuture;
    private final Runnable onClose;
    private final AtomicBoolean closeManually = new AtomicBoolean(false);
    private final Phaser spawnResultConsumers =
        new Phaser(1) {
          @Override
          protected boolean onAdvance(int phase, int registeredParties) {
            // We only use a single phase.
            return true;
          }
        };

    private LocalExecution(RemoteAction action, Runnable onClose) {
      this.action = action;
      this.spawnResultFuture = SettableFuture.create();
      this.onClose = onClose;
    }

    /**
     * Creates a new {@link LocalExecution} instance tracking the potential local execution of the
     * given {@link RemoteAction} if there is a chance that the same action will be executed by a
     * different Spawn.
     *
     * <p>This is only done for local (as in, non-remote) execution as remote executors are expected
     * to already have deduplication mechanisms for actions in place, perhaps even across different
     * builds and clients.
     */
    @Nullable
    public static LocalExecution createIfDeduplicatable(RemoteAction action, Runnable onClose) {
      if (action.getSpawn().getPathMapper().isNoop()) {
        return null;
      }
      return new LocalExecution(action, onClose);
    }

    /**
     * Attempts to register a thread waiting for the {@link #spawnResultFuture} to become available
     * and returns true if successful.
     *
     * <p>Every call to this method must be matched by a call to {@link #unregister()} via
     * try-finally.
     *
     * <p>This always returns true for actions that do not modify their spawns' outputs after
     * execution.
     */
    public boolean registerForOutputReuse() {
      // We only use a single phase.
      return spawnResultConsumers.register() == 0;
    }

    /**
     * Unregisters a thread waiting for the {@link #spawnResultFuture}, either after successful
     * reuse of the outputs or upon failure.
     */
    public void unregister() {
      spawnResultConsumers.arriveAndDeregister();
    }

    /**
     * Waits for all potential consumers of the {@link #spawnResultFuture} to be done with their
     * output reuse.
     */
    public void awaitAllOutputReuse() {
      spawnResultConsumers.arriveAndAwaitAdvance();
    }

    /**
     * Signals to all potential consumers of the {@link #spawnResultFuture} that this execution has
     * finished or been canceled and that the result will no longer be available.
     */
    @Override
    public void close() {
      if (!closeManually.get()) {
        doClose();
      }
    }

    /**
     * Returns a {@link Runnable} that will close this {@link LocalExecution} instance when called.
     * After this method is called, the {@link LocalExecution} instance will not be closed by the
     * {@link #close()} method.
     */
    public Runnable delayClose() {
      if (!closeManually.compareAndSet(false, true)) {
        throw new IllegalStateException("delayClose has already been called");
      }
      return this::doClose;
    }

    private void doClose() {
      spawnResultFuture.cancel(true);
      onClose.run();
    }
  }

  /**
   * Makes the {@link SpawnResult} available to all parallel {@link Spawn}s for the same {@link
   * RemoteAction} waiting for it or notifies them that the spawn failed.
   *
   * @return Whether the spawn result should be uploaded to the cache.
   */
  public boolean commitResultAndDecideWhetherToUpload(
      SpawnResult result, @Nullable LocalExecution execution) {
    if (result.status().equals(SpawnResult.Status.SUCCESS) && result.exitCode() == 0) {
      if (execution != null) {
        execution.spawnResultFuture.set(result);
      }
      return true;
    } else {
      if (execution != null) {
        execution.spawnResultFuture.cancel(true);
      }
      return false;
    }
  }

  /**
   * Reuses the outputs of a concurrent local execution of the same RemoteAction in a different
   * spawn.
   *
   * <p>Since each output file is generated by a unique action and actions generally take care to
   * run a unique spawn for each output file, this method is only useful with path mapping enabled,
   * which allows different spawns in a single build to have the same RemoteAction.ActionKey.
   *
   * @return The {@link SpawnResult} of the previous execution if it was successful, otherwise null.
   */
  @Nullable
  public SpawnResult waitForAndReuseOutputs(RemoteAction action, LocalExecution previousExecution)
      throws InterruptedException, IOException {
    checkState(!shutdown.get(), "shutdown");

    SpawnResult previousSpawnResult;
    try {
      previousSpawnResult = previousExecution.spawnResultFuture.get();
    } catch (CancellationException | ExecutionException e) {
      if (e.getCause() != null) {
        Throwables.throwIfInstanceOf(e.getCause(), InterruptedException.class);
        Throwables.throwIfUnchecked(e.getCause());
      }
      // The spawn this action was deduplicated against failed due to an exception or
      // non-zero exit code. Since it isn't possible to transparently replay its failure for the
      // current spawn, we rerun the action instead.
      return null;
    }

    Preconditions.checkArgument(
        action.getActionKey().equals(previousExecution.action.getActionKey()));

    ImmutableMap<Path, ActionInput> previousOutputs =
        previousExecution.action.getSpawn().getOutputFiles().stream()
            .collect(toImmutableMap(output -> execRoot.getRelative(output.getExecPath()), o -> o));
    Map<Path, Path> realToTmpPath = new HashMap<>();
    ByteString inMemoryOutputContent = null;
    String inMemoryOutputPath = null;
    var outputPathsList =
        useOutputPaths()
            ? action.getCommand().getOutputPathsList()
            : Stream.concat(
                    action.getCommand().getOutputFilesList().stream(),
                    action.getCommand().getOutputDirectoriesList().stream())
                .toList();
    try {
      for (String output : outputPathsList) {
        String reencodedOutput = unicodeToInternal(output);
        Path sourcePath =
            previousExecution.action.getRemotePathResolver().outputPathToLocalPath(reencodedOutput);
        ActionInput outputArtifact = previousOutputs.get(sourcePath);
        Path targetPath = action.getRemotePathResolver().outputPathToLocalPath(reencodedOutput);
        inMemoryOutputContent = previousSpawnResult.getInMemoryOutput(outputArtifact);
        if (inMemoryOutputContent != null) {
          inMemoryOutputPath = targetPath.relativeTo(execRoot).getPathString();
          continue;
        }
        Path tmpPath = tempPathGenerator.generateTempPath();
        tmpPath.getParentDirectory().createDirectoryAndParents();
        try {
          if (outputArtifact.isDirectory()) {
            tmpPath.createDirectory();
            FileSystemUtils.copyTreesBelow(sourcePath, tmpPath);
          } else if (outputArtifact.isSymlink()) {
            FileSystemUtils.ensureSymbolicLink(tmpPath, sourcePath.readSymbolicLink());
          } else {
            FileSystemUtils.copyFile(sourcePath, tmpPath);
          }
          realToTmpPath.put(targetPath, tmpPath);
        } catch (FileNotFoundException e) {
          // The spawn this action was deduplicated against failed to create an output file. If the
          // output is mandatory, we cannot reuse the previous execution.
          if (action.getSpawn().isMandatoryOutput(outputArtifact)) {
            return null;
          }
        }
      }

      // The stdout output is captured as the action result's stdout rather than as a regular
      // output file, so it isn't part of the command's output paths and has to be copied
      // separately.
      Artifact stdoutOutput = action.getSpawn().getStdout();
      if (stdoutOutput != null) {
        Artifact previousStdoutOutput = previousExecution.action.getSpawn().getStdout();
        if (previousStdoutOutput == null) {
          // The previous spawn reported its stdout as regular action output, so it isn't
          // available as a file. Rerun the action instead.
          return null;
        }
        Path tmpPath = tempPathGenerator.generateTempPath();
        tmpPath.getParentDirectory().createDirectoryAndParents();
        try {
          FileSystemUtils.copyFile(previousStdoutOutput.getPath(), tmpPath);
          realToTmpPath.put(stdoutOutput.getPath(), tmpPath);
        } catch (FileNotFoundException e) {
          return null;
        }
      }

      // TODO: FileOutErr is action-scoped, not spawn-scoped, but this is not a problem for the
      //  current use case of supporting deduplication of path mapped spawns:
      //  1. Starlark and C++ compilation actions always create a single spawn.
      //  2. Java compilation actions may run a fallback spawn, but reset the FileOutErr before
      //     running it.
      //  If this changes, we will need to introduce a spawn-scoped OutErr.
      FileOutErr.dump(
          previousExecution.action.getSpawnExecutionContext().getFileOutErr(),
          action.getSpawnExecutionContext().getFileOutErr());

      action
          .getSpawnExecutionContext()
          .lockOutputFiles(
              previousSpawnResult.exitCode(),
              previousSpawnResult.getFailureMessage(),
              action.getSpawnExecutionContext().getFileOutErr());
      // All outputs are created locally.
      moveOutputsToFinalLocation(realToTmpPath.keySet(), realToTmpPath);
    } catch (InterruptedException | IOException e) {
      // Delete any copied output files.
      for (Path tmpPath : realToTmpPath.values()) {
        try {
          tmpPath.delete();
        } catch (IOException ignored) {
          // Best effort, will be cleaned up at server restart.
        }
      }
      throw e;
    }

    if (inMemoryOutputPath != null) {
      String finalInMemoryOutputPath = inMemoryOutputPath;
      ByteString finalInMemoryOutputContent = inMemoryOutputContent;
      return new SpawnResult.DelegateSpawnResult(previousSpawnResult) {
        @Override
        @Nullable
        public ByteString getInMemoryOutput(ActionInput output) {
          if (output.getExecPathString().equals(finalInMemoryOutputPath)) {
            return finalInMemoryOutputContent;
          }
          return null;
        }
      };
    }

    return previousSpawnResult;
  }

  private boolean shouldDownload(
      RemoteActionResult result, PathFragment execPath, @Nullable PathFragment treeRootExecPath) {
    if (outputService instanceof BazelOutputService) {
      return false;
    }

    // In case the action failed, download all outputs. It might be helpful for debugging and there
    // is no point in injecting output metadata of a failed action.
    if (result.getExitCode() != 0) {
      return true;
    }
    return remoteOutputChecker.shouldDownloadOutput(execPath, treeRootExecPath);
  }

  private static String prettyPrint(ActionInput actionInput) {
    if (actionInput instanceof Artifact artifact) {
      return artifact.prettyPrint();
    } else {
      return actionInput.getExecPathString();
    }
  }

  @VisibleForTesting
  UploadManifest buildUploadManifest(RemoteAction action, SpawnResult spawnResult)
      throws IOException, ExecException, InterruptedException {
    try (SilentCloseable c = Profiler.instance().profile("build upload manifest")) {
      ImmutableList.Builder<Path> outputFiles = ImmutableList.builder();
      // If the spawn redirected its stdout into an output, upload that file as the action's stdout
      // digest rather than as an output file.
      Artifact stdoutOutput = action.getSpawn().getStdout();
      FileOutErr fileOutErr = action.getSpawnExecutionContext().getFileOutErr();
      if (stdoutOutput != null) {
        fileOutErr = new FileOutErr(stdoutOutput.getPath(), fileOutErr.getErrorPath());
      }

      // Check that all mandatory outputs are created.
      for (ActionInput outputFile : outputsExcludingStdout(action.getSpawn())) {
        Symlinks followSymlinks = outputFile.isSymlink() ? Symlinks.NOFOLLOW : Symlinks.FOLLOW;
        Path localPath = execRoot.getRelative(outputFile.getExecPath());
        if (action.getSpawn().isMandatoryOutput(outputFile) && !localPath.exists(followSymlinks)) {
          throw new IOException(
              "Expected output " + prettyPrint(outputFile) + " was not created locally.");
        }
        outputFiles.add(localPath);
      }

      return UploadManifest.create(
          combinedCache.getRemoteCacheCapabilities(),
          digestUtil,
          action.getRemotePathResolver(),
          action.getActionKey(),
          action.getAction(),
          action.getCommand(),
          outputFiles.build(),
          fileOutErr,
          spawnResult.exitCode(),
          spawnResult.getStartTime(),
          spawnResult.getWallTimeInMs(),
          /* preserveExecutableBit= */ false);
    }
  }

  /** Upload outputs of a remote action which was executed locally to remote cache. */
  public void uploadOutputs(
      RemoteAction action,
      SpawnResult spawnResult,
      Runnable onUploadComplete,
      ConcurrentChangesCheckLevel concurrentChangesCheckLevel)
      throws InterruptedException, ExecException {
    checkState(!shutdown.get(), "shutdown");
    checkState(
        action.getRemoteActionExecutionContext().getWriteCachePolicy().allowAnyCache(),
        "spawn shouldn't upload local result");
    checkState(
        SpawnResult.Status.SUCCESS.equals(spawnResult.status()) && spawnResult.exitCode() == 0,
        "shouldn't upload outputs of failed local action");

    try (SilentCloseable c = Profiler.instance().profile("checkForConcurrentModifications")) {
      checkForConcurrentModifications(action, concurrentChangesCheckLevel);
    } catch (IOException e) {
      report(
          Event.warn(
              String.format(
                  "%s: Skipping uploading outputs because of concurrent modifications with"
                      + " --guard_against_concurrent_changes enabled: %s",
                  action.getSpawn().getTargetLabel(), e.getMessage())));
      onUploadComplete.run();
      return;
    }

    // The output blobs of a rewound action may have been evicted after this invocation uploaded
    // them, so bypass the deduplication of completed uploads. Otherwise, the refreshed action
    // result could still reference missing blobs. Evaluate this here rather than in the upload
    // task, which may run after the build's rewinding state has been discarded.
    boolean force = wasRewound.test(action.getSpawn().getResourceOwner());
    if (remoteOptions.getRemoteCacheAsync()
        && !action.getSpawn().getResourceOwner().mayModifySpawnOutputsAfterExecution()) {
      new OutputUploadTask(action, spawnResult, force, onUploadComplete).start();
    } else {
      doUploadOutputs(action, spawnResult, force, onUploadComplete);
    }
  }

  @Nullable
  private RemoteRewoundActionSynchronizer getRewoundActionSynchronizer() {
    if (outputService instanceof RemoteOutputService remoteOutputService
        && remoteOutputService.getRewoundActionSynchronizer()
            instanceof RemoteRewoundActionSynchronizer rewoundActionSynchronizer) {
      return rewoundActionSynchronizer;
    }
    return null;
  }

  /**
   * A cancellable background upload of an action's outputs.
   *
   * <p>Registers itself with the {@link RemoteRewoundActionSynchronizer}, if there is one, before
   * the upload starts and unregisters itself when it is done, so that a rewinding of the action
   * waits for uploads that are still in flight.
   *
   * <p>{@link CancellableTask} ensures that the completion callback runs exactly once and that
   * cancellation only completes once the upload no longer accesses the action's outputs.
   */
  @VisibleForTesting
  final class OutputUploadTask implements RemoteRewoundActionSynchronizer.Cancellable {
    private final CancellableTask<InterruptedException> upload;
    private final ActionExecutionMetadata spawnOwner;
    @Nullable private final RemoteRewoundActionSynchronizer rewoundActionSynchronizer;

    // Written by start() before the upload is handed to the executor, which establishes a
    // happens-before edge to the reads in run() on the executor thread.
    private Runnable unregisterHandle = () -> {};

    OutputUploadTask(
        RemoteAction action, SpawnResult spawnResult, boolean force, Runnable onUploadComplete) {
      this.upload =
          new CancellableTask<>(
              () -> {
                try {
                  doUploadOutputs(action, spawnResult, force, /* onUploadComplete= */ () -> {});
                } catch (ExecException e) {
                  reportUploadError(e);
                }
              },
              onUploadComplete);
      this.spawnOwner = action.getRemoteActionExecutionContext().getSpawnOwner();
      this.rewoundActionSynchronizer = getRewoundActionSynchronizer();
    }

    /** Registers the task for cancellation and starts the upload in the background. */
    void start() {
      // Register before starting the upload so that it can't unregister itself before it has been
      // registered.
      if (rewoundActionSynchronizer != null) {
        unregisterHandle = rewoundActionSynchronizer.registerOutputUploadTask(spawnOwner, this);
      }
      try {
        // Executes rather than submits the upload: the body of a task submitted to an
        // ExecutorService is skipped entirely if its future is cancelled before the body starts,
        // which is why cancellation goes through the task itself instead of its future.
        backgroundTaskExecutor.execute(this::run);
      } catch (RejectedExecutionException e) {
        // The upload will never run. Cancelling it runs the completion callback if no cancellation
        // got there first. A concurrent cancellation may still be running the callback, whose
        // effects are part of the upload's completion contract, so wait for it before propagating
        // the rejection.
        try {
          upload.requestCancellation();
          upload.awaitCompletionUninterruptibly();
        } finally {
          unregisterHandle.run();
        }
        throw e;
      }
    }

    private void run() {
      try {
        var unused = upload.runIfNotCancelled();
      } catch (InterruptedException ignored) {
        // ThreadPerTaskExecutor does not care about interrupt status.
      } finally {
        unregisterHandle.run();
      }
    }

    @Override
    public void requestCancellation() {
      upload.requestCancellation();
    }

    @Override
    public void awaitCompletion() throws InterruptedException {
      upload.awaitCompletion();
    }
  }

  private void doUploadOutputs(
      RemoteAction action, SpawnResult spawnResult, boolean force, Runnable onUploadComplete)
      throws ExecException, InterruptedException {
    try (SilentCloseable c =
        Profiler.instance().profile(ProfilerTask.UPLOAD_TIME, "upload outputs")) {
      UploadManifest manifest = buildUploadManifest(action, spawnResult);
      var unused =
          manifest.upload(action.getRemoteActionExecutionContext(), combinedCache, reporter, force);
    } catch (IOException e) {
      reportUploadError(e);
    } finally {
      onUploadComplete.run();
    }
  }

  private void checkForConcurrentModifications(
      RemoteAction action, ConcurrentChangesCheckLevel level) throws IOException {
    if (level == ConcurrentChangesCheckLevel.OFF) {
      return;
    }

    // As this check runs after the action has been executed, we can reuse the input map if it
    // has already been created with willAccessRepeatedly = true, but do not need to force its
    // retention.
    for (ActionInput input : action.getInputMap(/* willAccessRepeatedly= */ false).values()) {
      // In lite mode, only check source artifacts in the main repository for modifications.
      // Non-source artifacts are made read-only after execution, and external repositories are
      // rarely modified, with local_repository being the notable exception.
      // TODO: Find a way to include repositories that are symlinks to source directories.
      // On Bazel itself, this reduces the number of wasModifiedSinceDigest calls by 99% compared to
      // the full check. By not checking output files, this mode also avoids spurious false
      // positives (see https://github.com/bazelbuild/bazel/issues/3360).
      if (level == ConcurrentChangesCheckLevel.LITE
          && !(input instanceof Artifact artifact
              && artifact.isSourceArtifact()
              && !artifact.getRoot().isExternal())) {
        continue;
      } else if (input instanceof VirtualActionInput) {
        continue;
      }
      FileArtifactValue metadata =
          action.getSpawnExecutionContext().getInputMetadataProvider().getInputMetadata(input);
      Path path = execRoot.getRelative(input.getExecPath());
      if (metadata.wasModifiedSinceDigest(path)) {
        throw new IOException(path + " was modified during execution");
      }
    }
  }

  private void reportUploadError(Throwable error) {
    if (buildInterrupted.get()) {
      // If build interrupted, ignores all the errors
      return;
    }

    String errorMessage = "Remote Cache: " + grpcAwareErrorMessage(error, verboseFailures);

    report(Event.warn(errorMessage));
  }

  /**
   * Upload inputs of a remote action to remote cache if they are not presented already.
   *
   * <p>Must be called before calling {@link #executeRemotely}.
   */
  public void uploadInputsIfNotPresent(RemoteAction action, boolean force)
      throws IOException, ExecException, InterruptedException {
    checkState(!shutdown.get(), "shutdown");
    checkState(mayBeExecutedRemotely(action.getSpawn()), "spawn can't be executed remotely");

    RemoteExecutionCache remoteExecutionCache = (RemoteExecutionCache) combinedCache;
    // Upload the command and all the inputs into the remote cache.
    Map<Digest, Message> additionalInputs = Maps.newHashMapWithExpectedSize(2);
    additionalInputs.put(action.getActionKey().digest(), action.getAction());
    additionalInputs.put(action.getCommandHash(), action.getCommand());

    // As uploading depends on having the full input root in memory, limit
    // concurrency. This prevents memory exhaustion. We assume that
    // ensureInputsPresent() provides enough parallelism to saturate the
    // network connection.
    maybeAcquireRemoteActionBuildingSemaphore(ProfilerTask.UPLOAD_TIME);
    try {
      MerkleTree.Uploadable merkleTree;
      if (action.getMerkleTree() instanceof MerkleTree.Uploadable uploadable && !force) {
        merkleTree = uploadable;
      } else {
        // --experimental_remote_discard_merkle_trees was provided or the remote lost a shared
        // subtree uploaded previously. Recompute the tree - including all subtrees in the latter
        // case.
        Spawn spawn = action.getSpawn();
        SpawnExecutionContext context = action.getSpawnExecutionContext();
        ToolSignature toolSignature = getToolSignature(spawn, context);
        merkleTree =
            (MerkleTree.Uploadable)
                merkleTreeComputer.buildForSpawn(
                    spawn,
                    toolSignature != null ? toolSignature.toolInputs : ImmutableSet.of(),
                    scrubber,
                    context,
                    force
                        ? MerkleTreeComputer.BlobPolicy.KEEP_AND_REUPLOAD
                        : MerkleTreeComputer.BlobPolicy.KEEP);
      }

      remoteExecutionCache.ensureInputsPresent(
          action
              .getRemoteActionExecutionContext()
              .withWriteCachePolicy(CachePolicy.REMOTE_CACHE_ONLY), // Only upload to remote cache
          merkleTree,
          additionalInputs,
          force);
    } finally {
      maybeReleaseRemoteActionBuildingSemaphore();
    }
  }

  /**
   * Executes the remote action remotely and returns the result.
   *
   * @param acceptCachedResult tells remote execution server whether it should used cached result.
   * @param observer receives status updates during the execution.
   */
  public RemoteActionResult executeRemotely(
      RemoteAction action, boolean acceptCachedResult, OperationObserver observer)
      throws IOException, InterruptedException {
    checkState(!shutdown.get(), "shutdown");
    checkState(mayBeExecutedRemotely(action.getSpawn()), "spawn can't be executed remotely");

    ExecuteRequest.Builder requestBuilder =
        ExecuteRequest.newBuilder()
            .setInstanceName(remoteOptions.getRemoteInstanceName())
            .setDigestFunction(digestUtil.getDigestFunction())
            .setActionDigest(action.getActionKey().digest())
            // TODO(https://github.com/bazelbuild/remote-apis/pull/386): Allow cached execution
            // results for rewound actions when ExecutionCapabilities.verifies_action_results is
            // advertised.
            .setSkipCacheLookup(
                !acceptCachedResult || wasRewound.test(action.getSpawn().getResourceOwner()));
    if (remoteOptions.getRemoteResultCachePriority() != 0) {
      requestBuilder
          .getResultsCachePolicyBuilder()
          .setPriority(remoteOptions.getRemoteResultCachePriority());
    }
    if (remoteOptions.getRemoteExecutionPriority() != 0) {
      requestBuilder
          .getExecutionPolicyBuilder()
          .setPriority(remoteOptions.getRemoteExecutionPriority());
    }
    PathFragment inMemoryOutputPath = getInMemoryOutputPath(action.getSpawn());
    if (inMemoryOutputPath != null) {
      requestBuilder.addInlineOutputFiles(
          internalToUnicode(
              action.getRemotePathResolver().localPathToOutputPath(inMemoryOutputPath)));
    }

    ExecuteRequest request = requestBuilder.build();

    ExecuteResponse reply =
        remoteExecutor.executeRemotely(action.getRemoteActionExecutionContext(), request, observer);

    return RemoteActionResult.createFromResponse(reply);
  }

  /** A value classes representing downloaded server logs. */
  public static class ServerLogs {
    public int logCount;
    public Path directory;
    @Nullable public Path lastLogPath;
  }

  /** Downloads server logs from a remotely executed action if any. */
  public ServerLogs maybeDownloadServerLogs(RemoteAction action, ExecuteResponse resp, Path logDir)
      throws InterruptedException, IOException {
    checkState(!shutdown.get(), "shutdown");
    checkNotNull(combinedCache, "combinedCache can't be null");
    ServerLogs serverLogs = new ServerLogs();
    serverLogs.directory = logDir.getRelative(action.getActionId());

    ActionResult actionResult = resp.getResult();
    if (resp.getServerLogsCount() > 0
        && (actionResult.getExitCode() != 0 || resp.getStatus().getCode() != Code.OK.value())) {
      // Always read from remote cache for just remotely executed action, even if the spawn doesn't
      // accept cached results (e.g. because it is tagged with no-cache).
      var context = action.getRemoteActionExecutionContext();
      context = context.withReadCachePolicy(context.getReadCachePolicy().addRemoteCache());
      for (Map.Entry<String, LogFile> e : resp.getServerLogsMap().entrySet()) {
        if (e.getValue().getHumanReadable()) {
          Path lastLogPath = serverLogs.directory.getRelative(e.getKey());
          if (!lastLogPath.startsWith(serverLogs.directory)) {
            throw new IOException(
                String.format(
                    "Path traversal detected in server log key: %s (resolved: %s, expected"
                        + " descendant of %s)",
                    e.getKey(), lastLogPath, serverLogs.directory));
          }
          serverLogs.lastLogPath = lastLogPath;
          serverLogs.logCount++;
          getFromFuture(
              combinedCache.downloadFile(
                  context, serverLogs.lastLogPath, e.getValue().getDigest()));
        }
      }
    }

    return serverLogs;
  }

  @Subscribe
  public void onBuildInterrupted(BuildInterruptedEvent event) {
    buildInterrupted.set(true);
  }

  @Subscribe
  public void onBuildComplete(BuildCompleteEvent event) {
    if (event.getResult().getSuccess()) {
      // If build succeeded, clear knownMissingCasDigests in case there are missing digests from
      // other targets from previous builds which are not relevant anymore.
      knownMissingCasDigests.clear();
    }
  }

  @Subscribe
  public void onLostInputs(LostInputsEvent event) {
    for (String digest : event.missingDigests()) {
      knownMissingCasDigests.add(DigestUtil.fromString(digest));
    }
  }

  /** Shuts the service down. */
  public void shutdown() {
    if (!shutdown.compareAndSet(false, true)) {
      return;
    }

    if (buildInterrupted.get()) {
      backgroundTaskExecutor.shutdownNow();
      if (combinedCache != null) {
        combinedCache.shutdownNow();
      }
      Thread.currentThread().interrupt();
    }

    // Waits for all background tasks to finish and interrupts them if there is another interrupt.
    backgroundTaskExecutor.close();

    // Release the cache only after background tasks are done as they might be using it.
    if (combinedCache != null) {
      combinedCache.release();
    }

    if (remoteExecutor != null) {
      remoteExecutor.close();
    }
  }

  /** Shuts down the background task executor so that new task submissions are rejected. */
  @VisibleForTesting
  void shutdownBackgroundTaskExecutorForTesting() {
    backgroundTaskExecutor.shutdown();
  }

  /**
   * Whether parameter files should be written locally, even when using remote execution or caching.
   */
  public void maybeWriteParamFilesLocally(Spawn spawn) throws IOException {
    if (!executionOptions.shouldMaterializeParamFiles()) {
      return;
    }
    for (ActionInput actionInput : spawn.getInputFiles().flatten()) {
      if (actionInput instanceof ParamFileActionInput paramFileActionInput) {
        paramFileActionInput.atomicallyWriteRelativeTo(execRoot);
      }
    }
  }

  void report(Event evt) {
    synchronized (this) {
      if (reportedErrors.contains(evt.getMessage())) {
        return;
      }
      reportedErrors.add(evt.getMessage());
      reporter.handle(evt);
    }
  }

  private static boolean isScrubbedSpawn(Spawn spawn, @Nullable Scrubber scrubber) {
    return scrubber != null && scrubber.forSpawn(spawn) != null;
  }

  /**
   * A simple value class combining a hash of the tool inputs (and their digests) as well as a set
   * of the relative paths of all tool inputs.
   */
  private static final class ToolSignature {
    private final String key;
    private final Set<PathFragment> toolInputs;

    private ToolSignature(String key, Set<PathFragment> toolInputs) {
      this.key = key;
      this.toolInputs = toolInputs;
    }
  }
}