RemoteModule.java

// Copyright 2016 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 java.util.concurrent.TimeUnit.SECONDS;

import build.bazel.remote.execution.v2.Digest;
import build.bazel.remote.execution.v2.DigestFunction;
import com.github.benmanes.caffeine.cache.Cache;
import com.google.auth.Credentials;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Ascii;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.ListeningScheduledExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.google.devtools.build.lib.actions.Artifact;
import com.google.devtools.build.lib.actions.ImportantOutputHandler;
import com.google.devtools.build.lib.analysis.AnalysisResult;
import com.google.devtools.build.lib.analysis.BlazeDirectories;
import com.google.devtools.build.lib.analysis.ConfiguredAspect;
import com.google.devtools.build.lib.analysis.ConfiguredTarget;
import com.google.devtools.build.lib.analysis.config.BuildOptions;
import com.google.devtools.build.lib.analysis.config.CoreOptions;
import com.google.devtools.build.lib.authandtls.AuthAndTLSOptions;
import com.google.devtools.build.lib.authandtls.CallCredentialsProvider;
import com.google.devtools.build.lib.authandtls.GoogleAuthUtils;
import com.google.devtools.build.lib.authandtls.credentialhelper.CredentialHelperEnvironment;
import com.google.devtools.build.lib.authandtls.credentialhelper.CredentialModule;
import com.google.devtools.build.lib.authandtls.credentialhelper.GetCredentialsResponse;
import com.google.devtools.build.lib.bazel.repository.downloader.Downloader;
import com.google.devtools.build.lib.buildeventstream.BuildEventArtifactUploader;
import com.google.devtools.build.lib.buildeventstream.LocalFilesArtifactUploader;
import com.google.devtools.build.lib.buildtool.BuildRequest;
import com.google.devtools.build.lib.buildtool.BuildRequestOptions;
import com.google.devtools.build.lib.cmdline.LabelConstants;
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.ExecutorBuilder;
import com.google.devtools.build.lib.exec.ModuleActionContextRegistry;
import com.google.devtools.build.lib.exec.SpawnStrategyRegistry;
import com.google.devtools.build.lib.profiler.Profiler;
import com.google.devtools.build.lib.remote.CombinedCacheClientFactory.CombinedCacheClient;
import com.google.devtools.build.lib.remote.LeaseService.LeaseExtension;
import com.google.devtools.build.lib.remote.RemoteServerCapabilities.ServerCapabilitiesRequirement;
import com.google.devtools.build.lib.remote.Retrier.ResultClassifier;
import com.google.devtools.build.lib.remote.Retrier.ResultClassifier.Result;
import com.google.devtools.build.lib.remote.circuitbreaker.CircuitBreakerFactory;
import com.google.devtools.build.lib.remote.common.RemoteCacheClient;
import com.google.devtools.build.lib.remote.common.RemoteExecutionClient;
import com.google.devtools.build.lib.remote.disk.DiskCacheClient;
import com.google.devtools.build.lib.remote.disk.DiskCacheGarbageCollectorIdleTask;
import com.google.devtools.build.lib.remote.downloader.GrpcRemoteDownloader;
import com.google.devtools.build.lib.remote.http.DownloadTimeoutException;
import com.google.devtools.build.lib.remote.http.HttpException;
import com.google.devtools.build.lib.remote.logging.LoggingInterceptor;
import com.google.devtools.build.lib.remote.logging.RemoteExecutionLog.LogEntry;
import com.google.devtools.build.lib.remote.options.RemoteOptions;
import com.google.devtools.build.lib.remote.options.RemoteOutputsMode;
import com.google.devtools.build.lib.remote.options.RemoteStartupOptions;
import com.google.devtools.build.lib.remote.util.DigestUtil;
import com.google.devtools.build.lib.remote.util.TracingMetadataUtils;
import com.google.devtools.build.lib.runtime.BlazeModule;
import com.google.devtools.build.lib.runtime.BlazeRuntime;
import com.google.devtools.build.lib.runtime.BlazeServerStartupOptions;
import com.google.devtools.build.lib.runtime.BlazeService;
import com.google.devtools.build.lib.runtime.BlockWaitingModule;
import com.google.devtools.build.lib.runtime.BuildEventArtifactUploaderFactory;
import com.google.devtools.build.lib.runtime.CommandEnvironment;
import com.google.devtools.build.lib.runtime.CommandLinePathFactory;
import com.google.devtools.build.lib.runtime.RemoteRepoContentsCache;
import com.google.devtools.build.lib.runtime.RepositoryRemoteExecutor;
import com.google.devtools.build.lib.runtime.RepositoryRemoteHelpersFactory;
import com.google.devtools.build.lib.runtime.ServerBuilder;
import com.google.devtools.build.lib.runtime.WorkspaceBuilder;
import com.google.devtools.build.lib.server.FailureDetails;
import com.google.devtools.build.lib.server.FailureDetails.Execution;
import com.google.devtools.build.lib.server.FailureDetails.FailureDetail;
import com.google.devtools.build.lib.server.FailureDetails.RemoteExecution;
import com.google.devtools.build.lib.server.FailureDetails.RemoteExecution.Code;
import com.google.devtools.build.lib.skyframe.SkyframeExecutorWrappingWalkableGraph;
import com.google.devtools.build.lib.util.AbruptExitException;
import com.google.devtools.build.lib.util.DetailedExitCode;
import com.google.devtools.build.lib.util.ExitCode;
import com.google.devtools.build.lib.util.Fingerprint;
import com.google.devtools.build.lib.util.TempPathGenerator;
import com.google.devtools.build.lib.util.io.AsynchronousMessageOutputStream;
import com.google.devtools.build.lib.vfs.DigestHashFunction;
import com.google.devtools.build.lib.vfs.FileSystem;
import com.google.devtools.build.lib.vfs.OutputPermissions;
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.common.options.Options;
import com.google.devtools.common.options.OptionsBase;
import com.google.devtools.common.options.OptionsParsingResult;
import com.google.devtools.common.options.RegexPatternOption;
import io.grpc.CallCredentials;
import io.grpc.ClientInterceptor;
import io.grpc.ManagedChannel;
import io.netty.handler.codec.DecoderException;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.reactivex.rxjava3.plugins.RxJavaPlugins;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.channels.ClosedChannelException;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.function.Predicate;
import javax.annotation.Nullable;

/** RemoteModule provides distributed cache and remote execution for Bazel. */
public final class RemoteModule extends BlazeModule {
  private final ListeningScheduledExecutorService retryScheduler =
      MoreExecutors.listeningDecorator(Executors.newScheduledThreadPool(1));

  private final Set<Digest> knownMissingCasDigests = Sets.newConcurrentHashSet();
  private final ChunkLocationMap chunkLocationMap = new ChunkLocationMap();
  private boolean useRemoteRepoContentsCache;

  @Nullable private PathFragment outputBase;
  @Nullable private AsynchronousMessageOutputStream<LogEntry> rpcLogFile;
  @Nullable private ExecutorService executorService;
  @Nullable private RemoteActionContextProvider actionContextProvider;
  @Nullable private RemoteActionInputFetcher actionInputFetcher;
  @Nullable private RemoteOptions remoteOptions;
  @Nullable private CommandEnvironment env;
  @Nullable private OutputService outputService;
  @Nullable private TempPathGenerator tempPathGenerator;
  @Nullable private BlockWaitingModule blockWaitingModule;
  @Nullable private RemoteOutputChecker remoteOutputChecker;
  @Nullable private RemoteOutputChecker lastRemoteOutputChecker;
  @Nullable private String lastBuildId;

  private ChannelFactory channelFactory =
      new ChannelFactory() {
        @Override
        public ManagedChannel newChannel(
            String target,
            String proxy,
            AuthAndTLSOptions options,
            List<ClientInterceptor> interceptors,
            Map<String, ?> serviceConfig)
            throws IOException {
          return GoogleAuthUtils.newChannel(
              executorService,
              target,
              proxy,
              options,
              interceptors.isEmpty() ? null : interceptors,
              serviceConfig);
        }
      };

  private final BuildEventArtifactUploaderFactoryDelegate
      buildEventArtifactUploaderFactoryDelegate = new BuildEventArtifactUploaderFactoryDelegate();

  private final RepositoryRemoteHelpersFactoryDelegate repositoryRemoteHelpersFactoryDelegate =
      new RepositoryRemoteHelpersFactoryDelegate();

  @Nullable private GrpcRemoteDownloader remoteDownloader;

  private CredentialModule credentialModule;

  @Override
  public ImmutableList<Class<? extends OptionsBase>> getStartupOptions() {
    return ImmutableList.of(RemoteStartupOptions.class);
  }

  @Override
  public void globalInit(
      OptionsParsingResult startupOptions, Iterable<BlazeService> blazeServices) {
    outputBase = startupOptions.getOptions(BlazeServerStartupOptions.class).getOutputBase();
    useRemoteRepoContentsCache =
        startupOptions.getOptions(RemoteStartupOptions.class).getUseRemoteRepoContentsCache();
  }

  @Nullable
  @Override
  public FileSystem getFileSystemForBuildArtifacts(FileSystem nativeFs) {
    if (!useRemoteRepoContentsCache) {
      return null;
    }
    return new RemoteExternalOverlayFileSystem(
        outputBase.getRelative(LabelConstants.EXTERNAL_REPOSITORY_LOCATION), nativeFs);
  }

  @Override
  public void serverInit(OptionsParsingResult startupOptions, ServerBuilder builder) {
    builder.addBuildEventArtifactUploaderFactory(
        buildEventArtifactUploaderFactoryDelegate, "remote");
    builder.setRepositoryHelpersFactory(repositoryRemoteHelpersFactoryDelegate);
  }

  /** Returns whether remote execution should be enabled. */
  public static boolean shouldEnableRemoteExecution(RemoteOptions options) {
    return !Strings.isNullOrEmpty(options.getRemoteExecutor());
  }

  /** Returns whether the remote downloader should be enabled. */
  private static boolean shouldEnableRemoteDownloader(RemoteOptions options) {
    return !Strings.isNullOrEmpty(options.getRemoteDownloader());
  }

  /** Returns whether the remote output service should be enabled. */
  private static boolean shouldEnableRemoteOutputService(RemoteOptions options) {
    return !Strings.isNullOrEmpty(options.getRemoteOutputService());
  }

  public static final ResultClassifier HTTP_RESULT_CLASSIFIER =
      e -> {
        boolean retry = false;
        if (e instanceof ClosedChannelException) {
          retry = true;
        } else if (e instanceof DownloadTimeoutException) {
          retry = true;
        } else if (e instanceof HttpException httpException) {
          int status = httpException.response().status().code();
          if (status == HttpResponseStatus.NOT_FOUND.code()) {
            return Result.SUCCESS;
          }
          retry =
              status == HttpResponseStatus.REQUEST_TIMEOUT.code()
                  || status == HttpResponseStatus.TOO_MANY_REQUESTS.code()
                  || status == HttpResponseStatus.INTERNAL_SERVER_ERROR.code()
                  || status == HttpResponseStatus.BAD_GATEWAY.code()
                  || status == HttpResponseStatus.SERVICE_UNAVAILABLE.code()
                  || status == HttpResponseStatus.GATEWAY_TIMEOUT.code();
        } else if (e instanceof IOException) {
          String msg = Ascii.toLowerCase(e.getMessage());
          if (msg.contains("connection reset")) {
            retry = true;
          } else if (msg.contains("operation timed out")) {
            retry = true;
          }
        } else {
          // Workaround for a netty bug: https://github.com/netty/netty/issues/11815. Remove this
          // once it is fixed in the upstream.
          if (e instanceof DecoderException
              && e.getMessage().endsWith("functions:OPENSSL_internal:BAD_DECRYPT")) {
            retry = true;
          }
        }
        return retry ? Result.TRANSIENT_FAILURE : Result.PERMANENT_FAILURE;
      };

  private void initHttpAndDiskCache(
      CommandEnvironment env,
      Credentials credentials,
      AuthAndTLSOptions authAndTlsOptions,
      RemoteOptions remoteOptions,
      @Nullable PathFragment diskCachePath,
      DigestUtil digestUtil,
      boolean checkDiskCacheActionResultIntegrity) {
    CombinedCacheClient combinedCacheClient;
    Retrier.CircuitBreaker circuitBreaker =
        CircuitBreakerFactory.createCircuitBreaker(remoteOptions);
    try {
      combinedCacheClient =
          CombinedCacheClientFactory.create(
              remoteOptions,
              diskCachePath,
              credentials,
              authAndTlsOptions,
              Preconditions.checkNotNull(env.getWorkingDirectory(), "workingDirectory"),
              digestUtil,
              new RemoteRetrier(
                  remoteOptions, HTTP_RESULT_CLASSIFIER, retryScheduler, circuitBreaker),
              checkDiskCacheActionResultIntegrity);
    } catch (IOException e) {
      handleInitFailure(env, e, Code.CACHE_INIT_FAILURE);
      return;
    }
    CombinedCache combinedCache =
        new CombinedCache(
            combinedCacheClient.remoteCacheClient(),
            combinedCacheClient.diskCacheClient(),
            Strings.emptyToNull(remoteOptions.getRemoteDownloadSymlinkTemplate()),
            digestUtil,
            remoteOptions.getEffectiveChunkingFunction(),
            chunkLocationMap);
    actionContextProvider =
        RemoteActionContextProvider.createForRemoteCaching(
            env,
            combinedCache,
            /* retryScheduler= */ null,
            digestUtil,
            remoteOutputChecker,
            outputService,
            knownMissingCasDigests);
    actionInputFetcher = createActionInputFetcher(combinedCache);
  }

  @Nullable
  private RemoteActionInputFetcher createActionInputFetcher(@Nullable CombinedCache combinedCache) {
    if (combinedCache == null) {
      return null;
    }
    var coreOptions = env.getOptions().getOptions(CoreOptions.class);
    var outputPermissions =
        coreOptions != null && coreOptions.getExperimentalWritableOutputs()
            ? OutputPermissions.WRITABLE
            : OutputPermissions.READONLY;
    return new RemoteActionInputFetcher(
        env.getReporter(),
        env.getBuildRequestId(),
        env.getCommandId().toString(),
        combinedCache,
        env.getExecRoot(),
        tempPathGenerator,
        remoteOutputChecker,
        env.getOptions().getOptions(BuildRequestOptions.class) != null
            ? env.getOutputDirectoryHelper()
            : null,
        outputPermissions);
  }

  /**
   * Initializes the repository remote helpers factory and primes the {@link
   * RemoteExternalOverlayFileSystem} (when one is in use) with the per-build state it needs.
   */
  private void initRepoHelpersAndOverlayFs(
      CommandEnvironment env, String buildRequestId, String invocationId, boolean verboseFailures) {
    if (actionContextProvider == null) {
      return;
    }
    CombinedCache combinedCache = actionContextProvider.getCombinedCache();
    if (combinedCache == null) {
      return;
    }
    repositoryRemoteHelpersFactoryDelegate.init(
        new RepositoryRemoteHelpersFactoryImpl(
            env.getDirectories(),
            combinedCache,
            actionContextProvider.getRemoteExecutionClient(),
            buildRequestId,
            invocationId,
            env.getWorkspaceName(),
            remoteOptions.getRemoteInstanceName(),
            remoteOptions.getRemoteAcceptCached(),
            remoteOptions.getRemoteUploadLocalResults(),
            verboseFailures));
    if (env.getDirectories().getOutputBase().getFileSystem()
        instanceof RemoteExternalOverlayFileSystem remoteFs) {
      remoteFs.beforeCommand(
          combinedCache,
          actionInputFetcher,
          env.getReporter(),
          buildRequestId,
          invocationId,
          env.getSkyframeExecutor().getEvaluator(),
          remoteOptions.getRemoteCacheTtl());
    }
  }

  @Override
  public void workspaceInit(
      BlazeRuntime runtime, BlazeDirectories directories, WorkspaceBuilder builder) {
    Preconditions.checkState(blockWaitingModule == null, "blockWaitingModule must be null");
    Preconditions.checkState(credentialModule == null, "credentialModule must be null");
    blockWaitingModule =
        Preconditions.checkNotNull(runtime.getBlazeModule(BlockWaitingModule.class));
    credentialModule = Preconditions.checkNotNull(runtime.getBlazeModule(CredentialModule.class));
  }

  /**
   * Opens the gRPC log at {@code path} for writing.
   *
   * <p>When the command is retried in-process after a transient remote cache error (see {@code
   * --experimental_remote_cache_eviction_retries}), {@code attemptNumber} is greater than 1 and the
   * log written by the previous attempt is still at {@code path}. Truncating it would discard the
   * log of the very attempt that hit the cache eviction, which is exactly the one worth debugging.
   * Instead, rename the existing file to {@code <path>.<previous attempt number>} so every
   * attempt's log is preserved. See https://github.com/bazelbuild/bazel/issues/18695.
   */
  @VisibleForTesting
  static AsynchronousMessageOutputStream<LogEntry> openRpcLogFile(Path path, int attemptNumber)
      throws IOException {
    if (attemptNumber > 1 && path.exists()) {
      path.renameTo(
          path.getParentDirectory().getChild(path.getBaseName() + "." + (attemptNumber - 1)));
    }
    return new AsynchronousMessageOutputStream<>(path);
  }

  @Override
  public void beforeCommand(CommandEnvironment env) throws AbruptExitException {
    Preconditions.checkState(actionContextProvider == null, "actionContextProvider must be null");
    Preconditions.checkState(actionInputFetcher == null, "actionInputFetcher must be null");
    Preconditions.checkState(remoteOptions == null, "remoteOptions must be null");
    Preconditions.checkState(this.env == null, "env must be null");
    Preconditions.checkState(tempPathGenerator == null, "tempPathGenerator must be null");
    Preconditions.checkState(remoteOutputChecker == null, "remoteOutputChecker must be null");
    Preconditions.checkState(outputService == null, "remoteOutputService must be null");

    if ("clean".equals(env.getCommandName())) {
      chunkLocationMap.clear();
      knownMissingCasDigests.clear();
    }

    var cacheAvailable = setup(env);
    if (!cacheAvailable) {
      if (env.getDirectories().getOutputBase().getFileSystem()
          instanceof RemoteExternalOverlayFileSystem remoteFs) {
        remoteFs.notifyNoCacheAvailable(env.getSkyframeExecutor().getEvaluator());
      }
    }
  }

  /**
   * Sets up all requested remote functionality (caching, execution, downloader, ...) and returns
   * whether any cache (disk or remote) is enabled.
   */
  private boolean setup(CommandEnvironment env) throws AbruptExitException {
    RemoteOptions remoteOptions = env.getOptions().getOptions(RemoteOptions.class);
    if (remoteOptions == null) {
      // Quit if no supported command is being used. See getCommandOptions for details.
      return false;
    }

    this.remoteOptions = remoteOptions;
    this.env = env;

    // Resolve default disk cache location from --disk_cache / --disk_cache=true, etc.
    PathFragment diskCachePath =
        remoteOptions.getDiskCachePath(
            env.getDirectories().getServerDirectories().getOutputUserRoot());

    AuthAndTLSOptions authAndTlsOptions = env.getOptions().getOptions(AuthAndTLSOptions.class);
    DigestHashFunction hashFn = env.getRuntime().getFileSystem().getDigestFunction();
    DigestUtil digestUtil = new DigestUtil(env.getXattrProvider(), hashFn);

    boolean verboseFailures = false;
    ExecutionOptions executionOptions = env.getOptions().getOptions(ExecutionOptions.class);
    if (executionOptions != null) {
      verboseFailures = executionOptions.getVerboseFailures();
    }

    // If --remote_cache is empty but --remote_executor is not, reuse the latter for the former.
    if (!Strings.isNullOrEmpty(remoteOptions.getRemoteExecutor())
        && Strings.isNullOrEmpty(remoteOptions.getRemoteCache())) {
      remoteOptions.setRemoteCache(remoteOptions.getRemoteExecutor());
    }

    if (shouldEnableRemoteOutputService(remoteOptions)) {
      if (diskCachePath != null) {
        diskCachePath = null;
        env.getReporter()
            .handle(
                Event.warn(
                    "--disk_cache is ignored when --experimental_remote_output_service is set."));
      }

      if (Strings.isNullOrEmpty(remoteOptions.getRemoteCache())) {
        throw createOptionsExitException(
            "--experimental_remote_output_service must be used in combination with one of"
                + " --remote_cache or --remote_executor.",
            FailureDetails.RemoteOptions.Code.EXECUTION_WITH_INVALID_CACHE);
      }
    }

    boolean enableDiskCache = diskCachePath != null;
    boolean enableHttpCache = CombinedCacheClientFactory.isHttpCache(remoteOptions);
    boolean enableRemoteExecution = shouldEnableRemoteExecution(remoteOptions);
    boolean enableGrpcCache = GrpcCacheClient.isRemoteCacheOptions(remoteOptions);
    boolean enableRemoteDownloader = shouldEnableRemoteDownloader(remoteOptions);

    if (enableDiskCache) {
      // Check that the disk cache directory, which is managed by a garbage collecting idle task,
      // does not contain the output base. Since the specified output base path may be a symlink,
      // we resolve it fully. Intermediate symlinks do not have to be checked as the garbage
      // collector ignores symlinks. We also resolve the disk cache directory, where intermediate
      // symlinks also don't matter since deletion only occurs under the fully resolved path.
      Path resolvedOutputBase = env.getOutputBase();
      try {
        resolvedOutputBase = resolvedOutputBase.resolveSymbolicLinks();
      } catch (FileNotFoundException ignored) {
        // Will be created later.
      } catch (IOException e) {
        throw createOptionsExitException(
            "Failed to resolve output base: %s".formatted(e.getMessage()),
            FailureDetails.RemoteOptions.Code.EXECUTION_WITH_INVALID_CACHE);
      }
      Path resolvedDiskCache = env.getWorkingDirectory().getRelative(diskCachePath);
      try {
        resolvedDiskCache = resolvedDiskCache.resolveSymbolicLinks();
      } catch (FileNotFoundException ignored) {
        // Will be created later.
      } catch (IOException e) {
        throw createOptionsExitException(
            "Failed to resolve disk cache directory: %s".formatted(e.getMessage()),
            FailureDetails.RemoteOptions.Code.EXECUTION_WITH_INVALID_CACHE);
      }
      if (resolvedOutputBase.startsWith(resolvedDiskCache)) {
        // This is dangerous as the disk cache GC may delete files in the output base.
        throw createOptionsExitException(
            "The output base [%s] cannot be a subdirectory of the --disk_cache directory [%s]"
                .formatted(resolvedOutputBase, resolvedDiskCache),
            FailureDetails.RemoteOptions.Code.EXECUTION_WITH_INVALID_CACHE);
      }
      var gcIdleTask =
          DiskCacheGarbageCollectorIdleTask.create(
              remoteOptions, diskCachePath, env.getWorkingDirectory());
      if (gcIdleTask != null) {
        env.addIdleTask(gcIdleTask);
      }
    }

    if (enableRemoteDownloader && !enableGrpcCache) {
      throw createOptionsExitException(
          "The remote downloader can only be used in combination with gRPC caching",
          FailureDetails.RemoteOptions.Code.DOWNLOADER_WITHOUT_GRPC_CACHE);
    }

    tempPathGenerator = getTempPathGenerator(env);

    if (!enableDiskCache && !enableHttpCache && !enableGrpcCache && !enableRemoteExecution) {
      // Quit if no remote caching or execution was enabled.
      actionContextProvider =
          RemoteActionContextProvider.createForPlaceholder(
              env, retryScheduler, digestUtil, knownMissingCasDigests);
      return false;
    }

    if (enableHttpCache && enableRemoteExecution) {
      throw createOptionsExitException(
          "Cannot combine gRPC based remote execution with HTTP-based caching",
          FailureDetails.RemoteOptions.Code.EXECUTION_WITH_INVALID_CACHE);
    }

    boolean enableScrubbing = remoteOptions.getScrubber() != null;
    if (enableScrubbing && enableRemoteExecution) {
      env.getReporter()
          .handle(
              Event.warn(
                  "Cache key scrubbing is incompatible with remote execution. Actions that are"
                      + " scrubbed per the --experimental_remote_scrubbing_config configuration"
                      + " file will be executed locally instead."));
    }

    if (digestUtil.getDigestFunction() == DigestFunction.Value.UNKNOWN) {
      throw new AbruptExitException(
          DetailedExitCode.of(
              FailureDetail.newBuilder()
                  .setMessage(String.format("Unsupported digest function: %s", hashFn))
                  .setExecution(Execution.newBuilder().setCode(Execution.Code.EXECUTION_UNKNOWN))
                  .build()));
    }

    // TODO(bazel-team): Consider adding a warning or more validation if the remoteDownloadRegex is
    // used without Build without the Bytes.
    ImmutableList.Builder<Predicate<String>> patternsToDownloadBuilder = ImmutableList.builder();
    if (remoteOptions.getRemoteOutputsMode() != RemoteOutputsMode.ALL) {
      for (RegexPatternOption patternOption : remoteOptions.getRemoteDownloadRegex()) {
        patternsToDownloadBuilder.add(patternOption.matcher());
      }
    }

    remoteOutputChecker =
        new RemoteOutputChecker(
            env.getCommandName(),
            remoteOptions.getRemoteOutputsMode(),
            patternsToDownloadBuilder.build(),
            lastRemoteOutputChecker);
    remoteOutputChecker.maybeInvalidateSkyframeValues(env.getSkyframeExecutor().getEvaluator());

    env.getEventBus().register(this);
    String invocationId = env.getCommandId().toString();
    String buildRequestId = env.getBuildRequestId();
    env.getReporter().handle(Event.info(String.format("Invocation ID: %s", invocationId)));

    RxJavaPlugins.setErrorHandler(
        error -> env.getReporter().handle(Event.error(Throwables.getStackTraceAsString(error))));

    Path logDir =
        env.getOutputBase().getRelative(env.getRuntime().getProductName() + "-remote-logs");
    cleanAndCreateRemoteLogsDir(logDir);

    BuildRequestOptions buildRequestOptions =
        env.getOptions().getOptions(BuildRequestOptions.class);

    int jobs = 0;
    if (buildRequestOptions != null) {
      jobs = buildRequestOptions.getJobs();
    }

    ThreadFactory threadFactory =
        new ThreadFactoryBuilder().setNameFormat("remote-executor-%d").build();
    if (jobs != 0) {
      ThreadPoolExecutor tpe =
          new ThreadPoolExecutor(
              jobs, jobs, 60L, SECONDS, new LinkedBlockingQueue<>(), threadFactory);
      tpe.allowCoreThreadTimeOut(true);
      executorService = tpe;
    } else {
      executorService = Executors.newCachedThreadPool(threadFactory);
    }

    Credentials credentials;
    try {
      credentials =
          createCredentials(
              CredentialHelperEnvironment.newBuilder()
                  .setEventReporter(env.getReporter())
                  .setWorkspacePath(env.getWorkspace())
                  .setClientEnvironment(env::getClientEnv)
                  .setHelperExecutionTimeout(authAndTlsOptions.getCredentialHelperTimeout())
                  .build(),
              credentialModule.getCredentialCache(),
              env.getCommandLinePathFactory(),
              env.getRuntime().getFileSystem(),
              authAndTlsOptions,
              remoteOptions);
    } catch (IOException e) {
      handleInitFailure(env, e, Code.CREDENTIALS_INIT_FAILURE);
      return false;
    }

    int maxConcurrencyPerConnection = 0;
    if (remoteOptions.getRemoteMaxConcurrencyPerConnection() > 0) {
      maxConcurrencyPerConnection = remoteOptions.getRemoteMaxConcurrencyPerConnection();
    }
    int maxConnections = 0;
    if (remoteOptions.getRemoteMaxConnections() > 0) {
      maxConnections = remoteOptions.getRemoteMaxConnections();
    }

    Retrier.CircuitBreaker circuitBreaker =
        CircuitBreakerFactory.createCircuitBreaker(remoteOptions);
    RemoteRetrier retrier =
        new RemoteRetrier(
            remoteOptions,
            RemoteRetrier.EXPERIMENTAL_GRPC_RESULT_CLASSIFIER,
            retryScheduler,
            circuitBreaker,
            // Resolved lazily: rpcLogFile is created further below, after this retrier.
            () -> rpcLogFile);

    ImmutableMap<String, ?> remoteGrpcServiceConfig;
    try {
      remoteGrpcServiceConfig =
          RemoteGrpcServiceConfig.create(remoteOptions, env.getWorkingDirectory());
    } catch (IOException e) {
      throw createOptionsExitException(
          "Invalid --remote_grpc_service_config: " + e.getMessage(),
          FailureDetails.RemoteOptions.Code.REMOTE_GRPC_SERVICE_CONFIG_INVALID);
    }
    ClientInterceptor downloadIdleTimeoutInterceptor = null;
    if (!remoteOptions.getRemoteGrpcDownloadIdleTimeout().isZero()) {
      downloadIdleTimeoutInterceptor =
          new RemoteDownloadIdleTimeoutInterceptor(
              remoteOptions.getRemoteGrpcDownloadIdleTimeout(), retryScheduler);
    }

    if (!Strings.isNullOrEmpty(remoteOptions.getRemoteOutputService())) {
      var bazelOutputServiceChannel =
          createChannel(
              executorService,
              remoteOptions,
              // Don't use auth flags for remote output service
              Options.getDefaults(AuthAndTLSOptions.class),
              null,
              null,
              downloadIdleTimeoutInterceptor,
              remoteGrpcServiceConfig,
              channelFactory,
              remoteOptions.getRemoteOutputService(),
              null,
              maxConcurrencyPerConnection,
              maxConnections,
              env.getReporter(),
              null,
              digestUtil.getDigestFunction(),
              ServerCapabilitiesRequirement.NONE);

      outputService =
          new BazelOutputService(
              env.getOutputBase(),
              env::getExecRoot,
              () -> env.getDirectories().getOutputPath(env.getWorkspaceName()),
              digestUtil,
              remoteOptions.getRemoteCache(),
              remoteOptions.getRemoteInstanceName(),
              remoteOptions.getRemoteOutputServiceOutputPathPrefix(),
              remoteOptions.getMaxOutboundMessageSize(),
              verboseFailures,
              retrier,
              bazelOutputServiceChannel,
              lastBuildId);
    } else {
      outputService =
          new RemoteOutputService(
              env.getDirectories(),
              buildRequestOptions != null && buildRequestOptions.getRewindLostInputs());
    }

    // Verifying that the blobs referenced by a disk cache action result are present locally turns
    // most disk cache hits into misses when Build without the Bytes is enabled, as outputs aren't
    // downloaded and thus never added to the disk cache's CAS. With action rewinding, a blob that
    // is missing after all is cheap to recover from, so the check can be skipped. This allows disk
    // cache AC checks to be entirely local, reducing server load and avoiding a network round trip.
    boolean checkDiskCacheActionResultIntegrity =
        buildRequestOptions == null || !buildRequestOptions.getRewindLostInputs();

    if ((enableHttpCache || enableDiskCache) && !enableGrpcCache) {
      initHttpAndDiskCache(
          env,
          credentials,
          authAndTlsOptions,
          remoteOptions,
          diskCachePath,
          digestUtil,
          checkDiskCacheActionResultIntegrity);
      initRepoHelpersAndOverlayFs(env, buildRequestId, invocationId, verboseFailures);
      return true;
    }

    ClientInterceptor loggingInterceptor = null;
    if (remoteOptions.getRemoteGrpcLog() != null) {
      try {
        rpcLogFile =
            openRpcLogFile(
                env.getWorkingDirectory().getRelative(remoteOptions.getRemoteGrpcLog()),
                env.getAttemptNumber());
      } catch (IOException e) {
        handleInitFailure(env, e, Code.RPC_LOG_FAILURE);
        return false;
      }
      loggingInterceptor = new LoggingInterceptor(rpcLogFile, env.getRuntime().getClock());
    }
    CallCredentialsProvider callCredentialsProvider =
        GoogleAuthUtils.newCallCredentialsProvider(credentials);
    CallCredentials callCredentials = callCredentialsProvider.getCallCredentials();

    RemoteServerCapabilities rsc =
        new RemoteServerCapabilities(
            buildRequestId,
            invocationId,
            remoteOptions.getRemoteInstanceName(),
            callCredentials,
            retrier);

    ReferenceCountedChannel execChannel = null;
    ReferenceCountedChannel cacheChannel = null;
    // We only check required capabilities for a given endpoint.
    //
    // If --remote_executor and --remote_cache point to the same endpoint, we require that
    // endpoint has both execution and cache capabilities.
    //
    // If they point to different endpoints, we check the endpoint with execution or cache
    // capabilities respectively.
    try (var s = Profiler.instance().profile("init channel and check server capabilities")) {
      if (enableRemoteExecution) {
        // Create a separate channel if --remote_executor and --remote_cache point to different
        // endpoints.
        if (remoteOptions.getRemoteCache().equals(remoteOptions.getRemoteExecutor())) {
          execChannel =
              createChannel(
                  executorService,
                  remoteOptions,
                  authAndTlsOptions,
                  TracingMetadataUtils.newExecHeadersInterceptor(
                      remoteOptions.getRemoteHeaders(), remoteOptions.getRemoteExecHeaders()),
                  loggingInterceptor,
                  downloadIdleTimeoutInterceptor,
                  remoteGrpcServiceConfig,
                  channelFactory,
                  remoteOptions.getRemoteExecutor(),
                  remoteOptions.getRemoteProxy(),
                  maxConcurrencyPerConnection,
                  maxConnections,
                  env.getReporter(),
                  rsc,
                  digestUtil.getDigestFunction(),
                  ServerCapabilitiesRequirement.EXECUTION_AND_CACHE);
          cacheChannel = execChannel.retain();
        } else {
          execChannel =
              createChannel(
                  executorService,
                  remoteOptions,
                  authAndTlsOptions,
                  TracingMetadataUtils.newExecHeadersInterceptor(
                      remoteOptions.getRemoteHeaders(), remoteOptions.getRemoteExecHeaders()),
                  loggingInterceptor,
                  downloadIdleTimeoutInterceptor,
                  remoteGrpcServiceConfig,
                  channelFactory,
                  remoteOptions.getRemoteExecutor(),
                  remoteOptions.getRemoteProxy(),
                  maxConcurrencyPerConnection,
                  maxConnections,
                  env.getReporter(),
                  rsc,
                  digestUtil.getDigestFunction(),
                  ServerCapabilitiesRequirement.EXECUTION);
        }
      }

      if (cacheChannel == null) {
        cacheChannel =
            createChannel(
                executorService,
                remoteOptions,
                authAndTlsOptions,
                TracingMetadataUtils.newCacheHeadersInterceptor(
                    remoteOptions.getRemoteHeaders(), remoteOptions.getRemoteCacheHeaders()),
                loggingInterceptor,
                downloadIdleTimeoutInterceptor,
                remoteGrpcServiceConfig,
                channelFactory,
                remoteOptions.getRemoteCache(),
                remoteOptions.getRemoteProxy(),
                maxConcurrencyPerConnection,
                maxConnections,
                env.getReporter(),
                rsc,
                digestUtil.getDigestFunction(),
                ServerCapabilitiesRequirement.CACHE);
      }
    }

    RemoteCacheClient remoteCacheClient =
        new GrpcCacheClient(
            cacheChannel.retain(), callCredentialsProvider, remoteOptions, retrier, digestUtil);
    cacheChannel.release();
    DiskCacheClient diskCacheClient = null;

    if (enableRemoteExecution) {
      if (enableDiskCache) {
        try {
          diskCacheClient =
              CombinedCacheClientFactory.createDiskCache(
                  env.getWorkingDirectory(),
                  diskCachePath,
                  digestUtil,
                  checkDiskCacheActionResultIntegrity);
        } catch (Exception e) {
          handleInitFailure(env, e, Code.CACHE_INIT_FAILURE);
          return false;
        }
      }

      RemoteRetrier execRetrier =
          new RemoteRetrier(
              remoteOptions,
              RemoteRetrier.GRPC_RESULT_CLASSIFIER,
              retryScheduler,
              circuitBreaker,
              () -> rpcLogFile);
      RemoteExecutionClient remoteExecutor =
          new GrpcRemoteExecutor(execChannel.retain(), callCredentialsProvider, execRetrier);
      execChannel.release();
      RemoteExecutionCache remoteCache =
          new RemoteExecutionCache(
              remoteCacheClient,
              diskCacheClient,
              Strings.emptyToNull(remoteOptions.getRemoteDownloadSymlinkTemplate()),
              digestUtil,
              remoteOptions.getEffectiveChunkingFunction(),
              chunkLocationMap);
      actionContextProvider =
          RemoteActionContextProvider.createForRemoteExecution(
              env,
              remoteCache,
              remoteExecutor,
              retryScheduler,
              digestUtil,
              logDir,
              remoteOutputChecker,
              outputService,
              knownMissingCasDigests);
    } else {
      if (enableDiskCache) {
        try {
          diskCacheClient =
              CombinedCacheClientFactory.createDiskCache(
                  env.getWorkingDirectory(),
                  diskCachePath,
                  digestUtil,
                  checkDiskCacheActionResultIntegrity);
        } catch (Exception e) {
          handleInitFailure(env, e, Code.CACHE_INIT_FAILURE);
          return false;
        }
      }

      CombinedCache combinedCache =
          new CombinedCache(
              remoteCacheClient,
              diskCacheClient,
              Strings.emptyToNull(remoteOptions.getRemoteDownloadSymlinkTemplate()),
              digestUtil,
              remoteOptions.getEffectiveChunkingFunction(),
              chunkLocationMap);
      actionContextProvider =
          RemoteActionContextProvider.createForRemoteCaching(
              env,
              combinedCache,
              retryScheduler,
              digestUtil,
              remoteOutputChecker,
              outputService,
              knownMissingCasDigests);
    }

    actionInputFetcher = createActionInputFetcher(actionContextProvider.getCombinedCache());

    initRepoHelpersAndOverlayFs(env, buildRequestId, invocationId, verboseFailures);

    buildEventArtifactUploaderFactoryDelegate.init(
        new ByteStreamBuildEventArtifactUploaderFactory(
            executorService,
            env.getReporter(),
            verboseFailures,
            actionContextProvider.getCombinedCache(),
            remoteOptions.getRemoteInstanceName(),
            remoteOptions.getRemoteBytestreamUriPrefix(),
            buildRequestId,
            invocationId,
            remoteOptions.getRemoteBuildEventUploadMode(),
            remoteOptions.getMaximumOpenFiles()));

    if (enableRemoteDownloader) {
      ReferenceCountedChannel downloaderChannel;
      // Create a separate channel if --remote_downloader and --remote_cache point to different
      // endpoints.
      if (remoteOptions.getRemoteDownloader().equals(remoteOptions.getRemoteCache())) {
        downloaderChannel = cacheChannel.retain();
      } else {
        downloaderChannel =
            createChannel(
                executorService,
                remoteOptions,
                authAndTlsOptions,
                /* headersInterceptor= */ null,
                loggingInterceptor,
                downloadIdleTimeoutInterceptor,
                remoteGrpcServiceConfig,
                channelFactory,
                remoteOptions.getRemoteDownloader(),
                remoteOptions.getRemoteProxy(),
                maxConcurrencyPerConnection,
                maxConnections,
                env.getReporter(),
                rsc,
                digestUtil.getDigestFunction(),
                ServerCapabilitiesRequirement.NONE);
      }

      remoteDownloader =
          new GrpcRemoteDownloader(
              buildRequestId,
              invocationId,
              downloaderChannel.retain(),
              Optional.ofNullable(callCredentials),
              retrier,
              remoteCacheClient,
              digestUtil.getDigestFunction(),
              remoteOptions,
              verboseFailures,
              env.getHttpDownloader(),
              remoteOptions.getRemoteDownloaderLocalFallback());
      downloaderChannel.release();
      env.getDownloaderDelegate().setDelegate(remoteDownloader);
    }

    return true;
  }

  private static ReferenceCountedChannel createChannel(
      ExecutorService executorService,
      RemoteOptions remoteOptions,
      AuthAndTLSOptions authAndTlsOptions,
      @Nullable ClientInterceptor headersInterceptor,
      @Nullable ClientInterceptor loggingInterceptor,
      @Nullable ClientInterceptor downloadIdleTimeoutInterceptor,
      Map<String, ?> serviceConfig,
      ChannelFactory channelFactory,
      String target,
      String proxy,
      int maxConcurrencyPerConnection,
      int maxConnections,
      Reporter reporter,
      @Nullable RemoteServerCapabilities remoteServerCapabilities,
      DigestFunction.Value digestFunction,
      ServerCapabilitiesRequirement requirement) {
    ImmutableList.Builder<ClientInterceptor> interceptors = ImmutableList.builder();
    if (headersInterceptor != null) {
      interceptors.add(headersInterceptor);
    }
    if (loggingInterceptor != null) {
      interceptors.add(loggingInterceptor);
    }
    if (downloadIdleTimeoutInterceptor != null) {
      interceptors.add(downloadIdleTimeoutInterceptor);
    }
    var channel =
        new ReferenceCountedChannel(
            new GoogleChannelConnectionFactory(
                channelFactory,
                target,
                proxy,
                remoteOptions,
                serviceConfig,
                authAndTlsOptions,
                interceptors.build(),
                maxConcurrencyPerConnection,
                reporter,
                remoteServerCapabilities,
                digestFunction,
                requirement),
            maxConnections);
    // Eagerly start creating the channel and verifying the capabilities in the background.
    // TODO(tjgq): Make sure this task doesn't linger beyond afterCommand().
    var unused =
        executorService.submit(
            () -> {
              var unused2 = channel.withChannelFuture(c -> null);
            });
    return channel;
  }

  private static void handleInitFailure(
      CommandEnvironment env, Exception e, Code remoteExecutionCode) {
    env.getReporter().handle(Event.error(e.getMessage()));
    env.getBlazeModuleEnvironment()
        .exit(
            createExitException(
                "Error initializing RemoteModule",
                ExitCode.COMMAND_LINE_ERROR,
                remoteExecutionCode));
  }

  // This is a Skymeld-only code path. At the same time, afterAnalysis is exclusive to the
  // non-Skymeld code path.
  @Override
  public void afterTopLevelTargetAnalysis(
      CommandEnvironment env,
      BuildRequest request,
      BuildOptions buildOptions,
      ConfiguredTarget configuredTarget) {
    if (remoteOutputChecker != null) {
      remoteOutputChecker.afterTopLevelTargetAnalysis(
          configuredTarget, request::getTopLevelArtifactContext);
    }
  }

  @Override
  public void afterSingleAspectAnalysis(BuildRequest request, ConfiguredAspect configuredTarget) {
    if (remoteOutputChecker != null) {
      remoteOutputChecker.afterAspectAnalysis(
          configuredTarget, request::getTopLevelArtifactContext);
    }
  }

  @Override
  public void afterSingleTestAnalysis(BuildRequest request, ConfiguredTarget configuredTarget) {
    if (remoteOutputChecker != null) {
      remoteOutputChecker.afterTestAnalyzedEvent(configuredTarget);
    }
  }

  @Override
  public void coverageArtifactsKnown(ImmutableSet<Artifact> coverageArtifacts) {
    if (remoteOutputChecker != null) {
      remoteOutputChecker.coverageArtifactsKnown(coverageArtifacts);
    }
  }

  @Override
  public void afterAnalysis(
      CommandEnvironment env,
      BuildRequest request,
      BuildOptions buildOptions,
      AnalysisResult analysisResult) {
    BuildRequestOptions buildRequestOptions =
        env.getOptions().getOptions(BuildRequestOptions.class);
    if (remoteOutputChecker != null && buildRequestOptions.getPerformExecutionPhase()) {
      remoteOutputChecker.afterAnalysis(analysisResult);
    }
  }

  private static void cleanAndCreateRemoteLogsDir(Path logDir) throws AbruptExitException {
    try {
      // Clean out old logs files.
      if (logDir.exists()) {
        logDir.deleteTree();
      }
      logDir.createDirectory();
    } catch (IOException e) {
      String message = String.format("Could not create base directory for remote logs: %s", logDir);
      throw createExitException(
          message, ExitCode.LOCAL_ENVIRONMENTAL_ERROR, Code.LOG_DIR_CLEANUP_FAILURE);
    }
  }

  @Override
  public void afterCommand() {
    Preconditions.checkNotNull(blockWaitingModule, "blockWaitingModule must not be null");

    // Some cleanup tasks must wait until every other BlazeModule's afterCommand() has run, as
    // otherwise we might interfere with asynchronous remote downloads that are in progress.
    RemoteActionContextProvider actionContextProviderRef = actionContextProvider;
    GrpcRemoteDownloader remoteDownloaderRef = remoteDownloader;
    TempPathGenerator tempPathGeneratorRef = tempPathGenerator;
    AsynchronousMessageOutputStream<LogEntry> rpcLogFileRef = rpcLogFile;
    if (actionContextProviderRef != null
        || remoteDownloaderRef != null
        || tempPathGeneratorRef != null
        || rpcLogFileRef != null) {
      blockWaitingModule.submit(
          () ->
              afterCommandTask(
                  actionContextProviderRef,
                  remoteDownloaderRef,
                  tempPathGeneratorRef,
                  rpcLogFileRef));
    }

    lastRemoteOutputChecker = remoteOutputChecker;
    lastBuildId = Preconditions.checkNotNull(env).getCommandId().toString();

    buildEventArtifactUploaderFactoryDelegate.reset();
    repositoryRemoteHelpersFactoryDelegate.reset();
    if (env.getDirectories().getOutputBase().getFileSystem()
        instanceof RemoteExternalOverlayFileSystem remoteFs) {
      remoteFs.afterCommand();
    }
    remoteDownloader = null;
    actionContextProvider = null;
    actionInputFetcher = null;
    remoteOptions = null;
    env = null;
    outputService = null;
    tempPathGenerator = null;
    rpcLogFile = null;
    remoteOutputChecker = null;
  }

  private static void afterCommandTask(
      @Nullable RemoteActionContextProvider actionContextProvider,
      @Nullable GrpcRemoteDownloader remoteDownloader,
      @Nullable TempPathGenerator tempPathGenerator,
      @Nullable AsynchronousMessageOutputStream<LogEntry> rpcLogFile)
      throws AbruptExitException {
    if (actionContextProvider != null) {
      actionContextProvider.afterCommand();
    }

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

    if (tempPathGenerator != null) {
      Path tempDir = tempPathGenerator.getTempDir();
      try {
        tempDir.deleteTree();
      } catch (IOException ignored) {
        // Intentionally ignored.
      }
    }

    if (rpcLogFile != null) {
      try {
        rpcLogFile.close();
      } catch (IOException e) {
        throw createExitException(
            "Partially wrote RPC log file",
            ExitCode.LOCAL_ENVIRONMENTAL_ERROR,
            Code.RPC_LOG_FAILURE);
      }
    }
  }

  @Override
  public void registerSpawnStrategies(
      SpawnStrategyRegistry.Builder registryBuilder, CommandEnvironment env) {
    if (actionContextProvider == null) {
      return;
    }
    RemoteOptions remoteOptions =
        Preconditions.checkNotNull(
            env.getOptions().getOptions(RemoteOptions.class), "RemoteOptions");
    registryBuilder.setRemoteLocalFallbackStrategyIdentifier(
        remoteOptions.getRemoteLocalFallbackStrategy());
    actionContextProvider.registerRemoteSpawnStrategy(registryBuilder);
  }

  @Override
  public void registerActionContexts(
      ModuleActionContextRegistry.Builder registryBuilder,
      CommandEnvironment env,
      BuildRequest buildRequest) {
    if (actionContextProvider == null) {
      return;
    }
    actionContextProvider.registerSpawnCache(registryBuilder);

    // For skymeld, a non-toplevel target might become a toplevel after it has been executed. This
    // is the last chance to download the missing toplevel outputs in this case before sending out
    // TargetCompleteEvent. See https://github.com/bazelbuild/bazel/issues/20737.
    if (env.withMergedAnalysisAndExecutionSourceOfTruth()
        && actionInputFetcher != null
        && remoteOutputChecker != null) {
      registryBuilder.register(
          ImportantOutputHandler.class,
          new RemoteImportantOutputHandler(
              SkyframeExecutorWrappingWalkableGraph.of(env.getSkyframeExecutor()),
              remoteOutputChecker,
              actionInputFetcher,
              Preconditions.checkNotNull(outputService).getRewoundActionSynchronizer()));
    }
  }

  private TempPathGenerator getTempPathGenerator(CommandEnvironment env)
      throws AbruptExitException {
    Path tempDir = env.getActionTempsDirectory().getChild("remote");
    try {
      if (tempDir.exists()) {
        env.getReporter()
            .handle(Event.warn("Found stale downloads from previous build, deleting..."));
        tempDir.deleteTree();
      }
    } catch (IOException e) {
      throw new AbruptExitException(
          DetailedExitCode.of(
              ExitCode.LOCAL_ENVIRONMENTAL_ERROR,
              FailureDetail.newBuilder()
                  .setMessage(String.format("Failed to delete stale downloads: %s", e.getMessage()))
                  .setRemoteExecution(
                      RemoteExecution.newBuilder().setCode(Code.DOWNLOADED_INPUTS_DELETION_FAILURE))
                  .build()));
    }

    return new TempPathGenerator(tempDir);
  }

  @Override
  public void executorInit(CommandEnvironment env, BuildRequest request, ExecutorBuilder builder) {
    Preconditions.checkNotNull(remoteOptions, "remoteOptions must not be null");

    if (actionContextProvider == null) {
      return;
    }

    actionContextProvider.setTempPathGenerator(tempPathGenerator);

    if (actionContextProvider.getCombinedCache() != null) {
      Preconditions.checkNotNull(remoteOutputChecker, "remoteOutputChecker must not be null");
      Preconditions.checkNotNull(outputService, "remoteOutputService must not be null");
      Preconditions.checkNotNull(actionInputFetcher, "actionInputFetcher must not be null");

      env.getEventBus().register(actionInputFetcher);
      builder.setActionInputPrefetcher(actionInputFetcher);
      actionContextProvider.setActionInputFetcher(actionInputFetcher);

      BuildRequestOptions buildRequestOptions =
          env.getOptions().getOptions(BuildRequestOptions.class);
      boolean rewindLostInputs =
          buildRequestOptions != null && buildRequestOptions.getRewindLostInputs();
      LeaseService leaseService = null;
      if (rewindLostInputs) {
        // Action rewinding regenerates lost inputs within the build, so there is no need for the
        // lease service to extend leases or to discard all remote metadata after a build that
        // encountered lost inputs.
        if (remoteOptions.getRemoteCacheLeaseExtension()) {
          env.getReporter()
              .handle(
                  Event.warn(
                      "--experimental_remote_cache_lease_extension has no effect since"
                          + " --rewind_lost_inputs is enabled, which recovers from lost remote"
                          + " cache entries as they are encountered."));
        }
      } else {
        LeaseExtension leaseExtension = null;
        if (remoteOptions.getRemoteCacheLeaseExtension()) {
          leaseExtension =
              new RemoteLeaseExtension(
                  env.getSkyframeExecutor().getEvaluator(),
                  env.getBlazeWorkspace().getPersistentActionCache(),
                  env.getBuildRequestId(),
                  env.getCommandId().toString(),
                  actionContextProvider.getCombinedCache(),
                  remoteOptions.getRemoteCacheTtl());
        }
        leaseService =
            new LeaseService(
                env.getSkyframeExecutor().getEvaluator(),
                () -> env.getBlazeWorkspace().getPersistentActionCache(),
                leaseExtension);
        env.getEventBus().register(leaseService);
      }

      if (outputService instanceof RemoteOutputService remoteOutputService) {
        remoteOutputService.setRemoteOutputChecker(remoteOutputChecker);
        remoteOutputService.setActionInputFetcher(
            actionInputFetcher,
            SkyframeExecutorWrappingWalkableGraph.of(env.getSkyframeExecutor()));
        if (leaseService != null) {
          remoteOutputService.setLeaseService(leaseService);
        }
        env.getEventBus().register(outputService);
      }
    }

    builder.setActionExecutionSalt(computeActionExecutionSalt(remoteOptions));
  }

  @VisibleForTesting
  static String computeActionExecutionSalt(RemoteOptions remoteOptions) {
    Fingerprint fp = new Fingerprint();

    // When building without a remote cache following a build with one, cached actions may reference
    // remotely stored files which cannot be downloaded. For simplicity we also invalidate in the
    // reverse situation (building with a remote cache after building without one) even though
    // cached local actions don't need it.
    // TODO(chiwang): Solve this with build/action rewinding instead. The main difficulty is that if
    // no remote options are set, we lack a prefetcher and cannot trigger rewinding.
    fp.addBoolean(remoteOptions != null && remoteOptions.isRemoteCacheEnabled());

    // The default exec properties may affect how a spawn is remotely executed without affecting the
    // action key. In practice, only spawns with no execution platform or whose execution platform
    // has no exec properties are affected, but we don't have access to this information at the
    // time we're computing the action key, so we unconditionally include the defaults. This
    // shouldn't be too bad, as we don't expect the defaults to change very often.
    fp.addStringMap(
        remoteOptions != null ? remoteOptions.getRemoteDefaultExecProperties() : ImmutableMap.of());

    // When switching cache endpoints/instances across invocations, cached remote actions reference
    // blobs stored in the previous cache instance which cannot be downloaded from the new cache
    // instance. See https://github.com/bazelbuild/bazel/issues/23780.
    if (remoteOptions != null) {
      fp.addNullableString(remoteOptions.getRemoteCache());
      fp.addNullableString(remoteOptions.getRemoteExecutor());
      fp.addNullableString(remoteOptions.getRemoteDownloader());
      fp.addNullableString(remoteOptions.getRemoteInstanceName());
      fp.addNullableString(remoteOptions.getRemoteBytestreamUriPrefix());
      fp.addNullableString(remoteOptions.getRemoteProxy());
    }

    return fp.hexDigestAndReset();
  }

  @Override
  @Nullable
  public OutputService getOutputService() {
    return outputService;
  }

  @Override
  public Iterable<Class<? extends OptionsBase>> getCommonCommandOptions() {
    return ImmutableList.of(RemoteOptions.class, AuthAndTLSOptions.class);
  }

  private static class BuildEventArtifactUploaderFactoryDelegate
      implements BuildEventArtifactUploaderFactory {

    @Nullable private ByteStreamBuildEventArtifactUploaderFactory uploaderFactory;

    public void init(ByteStreamBuildEventArtifactUploaderFactory uploaderFactory) {
      Preconditions.checkState(this.uploaderFactory == null);
      this.uploaderFactory = uploaderFactory;
    }

    public void reset() {
      this.uploaderFactory = null;
    }

    @Override
    public BuildEventArtifactUploader create(CommandEnvironment env)
        throws InvalidPackagePathSymlinkException {
      BuildEventArtifactUploaderFactory uploaderFactory0 = this.uploaderFactory;
      if (uploaderFactory0 == null) {
        return new LocalFilesArtifactUploader();
      }
      return uploaderFactory0.create(env);
    }
  }

  private static AbruptExitException createOptionsExitException(
      String message, FailureDetails.RemoteOptions.Code remoteExecutionCode) {
    return new AbruptExitException(
        DetailedExitCode.of(
            FailureDetail.newBuilder()
                .setMessage(message)
                .setRemoteOptions(
                    FailureDetails.RemoteOptions.newBuilder().setCode(remoteExecutionCode))
                .build()));
  }

  private static AbruptExitException createExitException(
      String message, ExitCode exitCode, Code remoteExecutionCode) {
    return new AbruptExitException(
        DetailedExitCode.of(
            exitCode,
            FailureDetail.newBuilder()
                .setMessage(message)
                .setRemoteExecution(RemoteExecution.newBuilder().setCode(remoteExecutionCode))
                .build()));
  }

  private static class RepositoryRemoteHelpersFactoryDelegate
      implements RepositoryRemoteHelpersFactory {

    private volatile RepositoryRemoteHelpersFactory delegate;

    void init(RepositoryRemoteHelpersFactory delegate) {
      Preconditions.checkState(this.delegate == null);
      this.delegate = delegate;
    }

    public void reset() {
      this.delegate = null;
    }

    @Nullable
    @Override
    public RepositoryRemoteExecutor createExecutor() {
      RepositoryRemoteHelpersFactory delegate = this.delegate;
      if (delegate == null) {
        return null;
      }
      return delegate.createExecutor();
    }

    @Nullable
    @Override
    public RemoteRepoContentsCache createRepoContentsCache() {
      RepositoryRemoteHelpersFactory delegate = this.delegate;
      if (delegate == null) {
        return null;
      }
      return delegate.createRepoContentsCache();
    }
  }

  @VisibleForTesting
  void setChannelFactory(ChannelFactory channelFactory) {
    this.channelFactory = channelFactory;
  }

  @VisibleForTesting
  RemoteActionContextProvider getActionContextProvider() {
    return actionContextProvider;
  }

  @VisibleForTesting
  RepositoryRemoteHelpersFactory getRepositoryRemoteHelpersFactoryDelegate() {
    return repositoryRemoteHelpersFactoryDelegate;
  }

  @VisibleForTesting
  static Credentials createCredentials(
      CredentialHelperEnvironment credentialHelperEnvironment,
      Cache<URI, GetCredentialsResponse> credentialCache,
      CommandLinePathFactory commandLinePathFactory,
      FileSystem fileSystem,
      AuthAndTLSOptions authAndTlsOptions,
      RemoteOptions remoteOptions)
      throws IOException {
    Credentials credentials =
        GoogleAuthUtils.newCredentials(
            credentialHelperEnvironment,
            credentialCache,
            commandLinePathFactory,
            fileSystem,
            authAndTlsOptions);

    try {
      if (remoteOptions.getRemoteCache() != null
          && Ascii.toLowerCase(remoteOptions.getRemoteCache()).startsWith("http://")
          && !credentials.getRequestMetadata(new URI(remoteOptions.getRemoteCache())).isEmpty()) {
        // TODO(yannic): Make this a error aborting the build.
        credentialHelperEnvironment
            .eventReporter()
            .handle(
                Event.warn(
                    "Credentials are transmitted in plaintext to "
                        + remoteOptions.getRemoteCache()
                        + ". Please consider using an HTTPS endpoint."));
      }
    } catch (URISyntaxException e) {
      throw new IOException(e.getMessage(), e);
    }

    return credentials;
  }

  @VisibleForTesting
  Downloader getRemoteDownloader() {
    return remoteDownloader;
  }
}