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

// Copyright 2017 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.util.concurrent.Futures.immediateFailedFuture;
import static com.google.common.util.concurrent.Futures.immediateFuture;
import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
import static com.google.devtools.build.lib.remote.common.ProgressStatusListener.NO_ACTION;
import static com.google.devtools.build.lib.remote.util.Utils.bytesCountToDisplayString;
import static com.google.devtools.build.lib.remote.util.Utils.getFromFuture;

import build.bazel.remote.execution.v2.ActionResult;
import build.bazel.remote.execution.v2.CacheCapabilities;
import build.bazel.remote.execution.v2.Digest;
import build.bazel.remote.execution.v2.ServerCapabilities;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.flogger.GoogleLogger;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.devtools.build.lib.concurrent.ThreadSafety;
import com.google.devtools.build.lib.exec.SpawnCheckingCacheEvent;
import com.google.devtools.build.lib.exec.SpawnProgressEvent;
import com.google.devtools.build.lib.remote.common.CacheNotFoundException;
import com.google.devtools.build.lib.remote.common.LazyFileOutputStream;
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.RemoteCacheClient;
import com.google.devtools.build.lib.remote.common.RemoteCacheClient.ActionKey;
import com.google.devtools.build.lib.remote.disk.DiskCacheClient;
import com.google.devtools.build.lib.remote.options.RemoteOptions;
import com.google.devtools.build.lib.remote.util.AsyncTaskCache;
import com.google.devtools.build.lib.remote.util.DigestUtil;
import com.google.devtools.build.lib.remote.util.RxFutures;
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.util.io.OutErr;
import com.google.devtools.build.lib.vfs.FileSystemUtils;
import com.google.devtools.build.lib.vfs.Path;
import com.google.errorprone.annotations.CanIgnoreReturnValue;
import com.google.protobuf.ByteString;
import io.netty.util.AbstractReferenceCounted;
import io.reactivex.rxjava3.core.Completable;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicLong;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import javax.annotation.Nullable;

/**
 * Provides unified access to a disk cache, remote cache, or both.
 *
 * <p>The cache is reference counted. Initially, the reference count is 1. Use {@link #retain()} to
 * increase and {@link #release()} to decrease the reference count respectively. Once the reference
 * count is reached to 0, the underlying resources will be released (after pending I/O is finished).
 *
 * <p>Use {@link #awaitTermination()} to wait for pending I/O to finish. Use {@link #shutdownNow()}
 * to cancel all pending I/O and reject new requests.
 */
@ThreadSafety.ThreadSafe
public class CombinedCache extends AbstractReferenceCounted {
  private static final GoogleLogger logger = GoogleLogger.forEnclosingClass();

  private static final ListenableFuture<Void> COMPLETED_SUCCESS = immediateFuture(null);
  private static final ListenableFuture<byte[]> EMPTY_BYTES = immediateFuture(new byte[0]);
  private static final SpawnCheckingCacheEvent SPAWN_CHECKING_DISK_CACHE_EVENT =
      SpawnCheckingCacheEvent.create("disk-cache");
  private static final SpawnCheckingCacheEvent SPAWN_CHECKING_REMOTE_CACHE_EVENT =
      SpawnCheckingCacheEvent.create("remote-cache");

  private final CountDownLatch closeCountDownLatch = new CountDownLatch(1);
  protected final AsyncTaskCache.NoResult<Digest> casUploadCache = AsyncTaskCache.NoResult.create();

  @Nullable protected final RemoteCacheClient remoteCacheClient;
  @Nullable protected final DiskCacheClient diskCacheClient;
  protected final RemoteOptions options;
  protected final DigestUtil digestUtil;

  public CombinedCache(
      @Nullable RemoteCacheClient remoteCacheClient,
      @Nullable DiskCacheClient diskCacheClient,
      RemoteOptions options,
      DigestUtil digestUtil) {
    checkArgument(
        remoteCacheClient != null || diskCacheClient != null,
        "remoteCacheClient and diskCacheClient cannot be null at the same time");
    this.remoteCacheClient = remoteCacheClient;
    this.diskCacheClient = diskCacheClient;
    this.options = options;
    this.digestUtil = digestUtil;
  }

  public CacheCapabilities getRemoteCacheCapabilities() throws IOException {
    return getRemoteServerCapabilities().getCacheCapabilities();
  }

  public ListenableFuture<String> getRemoteAuthority() {
    if (remoteCacheClient == null) {
      return immediateFuture("");
    }
    return remoteCacheClient.getAuthority();
  }

  public ServerCapabilities getRemoteServerCapabilities() throws IOException {
    if (remoteCacheClient == null) {
      return ServerCapabilities.getDefaultInstance();
    }
    return remoteCacheClient.getServerCapabilities();
  }

  /**
   * Class to keep track of which cache (disk or remote) a given [cached] ActionResult comes from.
   */
  public record CachedActionResult(ActionResult actionResult, String cacheName) {
    @Nullable
    public static CachedActionResult remote(ActionResult actionResult) {
      if (actionResult == null) {
        return null;
      }
      return new CachedActionResult(actionResult, "remote");
    }

    @Nullable
    public static CachedActionResult disk(ActionResult actionResult) {
      if (actionResult == null) {
        return null;
      }
      return new CachedActionResult(actionResult, "disk");
    }
  }

  public CachedActionResult downloadActionResult(
      RemoteActionExecutionContext context,
      ActionKey actionKey,
      boolean inlineOutErr,
      Set<String> inlineOutputFiles)
      throws IOException, InterruptedException {
    var spawnExecutionContext = context.getSpawnExecutionContext();

    ListenableFuture<CachedActionResult> future = immediateFuture(null);

    if (diskCacheClient != null && context.getReadCachePolicy().allowDiskCache()) {
      // If Build without the Bytes is enabled, the future will likely return null
      // and fallback to remote cache because AC integrity check is enabled and referenced blobs are
      // probably missing from disk cache due to BwoB.
      //
      // TODO(chiwang): With lease service, instead of doing the integrity check against local
      // filesystem, we can check whether referenced blobs are alive in the lease service to
      // increase the cache-hit rate for disk cache.
      if (spawnExecutionContext != null) {
        spawnExecutionContext.report(SPAWN_CHECKING_DISK_CACHE_EVENT);
      }
      future =
          Futures.transform(
              diskCacheClient.downloadActionResult(actionKey),
              CachedActionResult::disk,
              directExecutor());
    }

    if (remoteCacheClient != null && context.getReadCachePolicy().allowRemoteCache()) {
      future =
          Futures.transformAsync(
              future,
              (result) -> {
                if (result == null) {
                  if (spawnExecutionContext != null) {
                    spawnExecutionContext.report(SPAWN_CHECKING_REMOTE_CACHE_EVENT);
                  }
                  return Futures.transform(
                      downloadActionResultFromRemote(
                          context, actionKey, inlineOutErr, inlineOutputFiles),
                      CachedActionResult::remote,
                      directExecutor());
                } else {
                  return immediateFuture(result);
                }
              },
              directExecutor());
    }

    return getFromFuture(future);
  }

  private ListenableFuture<ActionResult> downloadActionResultFromRemote(
      RemoteActionExecutionContext context,
      ActionKey actionKey,
      boolean inlineOutErr,
      Set<String> inlineOutputFiles) {
    checkState(remoteCacheClient != null && context.getReadCachePolicy().allowRemoteCache());
    return Futures.transformAsync(
        remoteCacheClient.downloadActionResult(context, actionKey, inlineOutErr, inlineOutputFiles),
        (actionResult) -> {
          if (actionResult == null) {
            return immediateFuture(null);
          }

          if (diskCacheClient != null && context.getWriteCachePolicy().allowDiskCache()) {
            return Futures.transform(
                diskCacheClient.uploadActionResult(actionKey, actionResult),
                v -> actionResult,
                directExecutor());
          }

          return immediateFuture(actionResult);
        },
        directExecutor());
  }

  /**
   * Returns a set of digests that the remote cache does not know about. The returned set is
   * guaranteed to be a subset of {@code digests}.
   */
  public ListenableFuture<ImmutableSet<Digest>> findMissingDigests(
      RemoteActionExecutionContext context, Iterable<Digest> digests) {
    if (Iterables.isEmpty(digests)) {
      return immediateFuture(ImmutableSet.of());
    }

    ListenableFuture<ImmutableSet<Digest>> diskQuery = immediateFuture(ImmutableSet.of());
    if (diskCacheClient != null && context.getWriteCachePolicy().allowDiskCache()) {
      diskQuery = diskCacheClient.findMissingDigests(digests);
    }

    ListenableFuture<ImmutableSet<Digest>> remoteQuery = immediateFuture(ImmutableSet.of());
    if (remoteCacheClient != null && context.getWriteCachePolicy().allowRemoteCache()) {
      remoteQuery = remoteCacheClient.findMissingDigests(context, digests);
    }

    ListenableFuture<ImmutableSet<Digest>> diskQueryFinal = diskQuery;
    ListenableFuture<ImmutableSet<Digest>> remoteQueryFinal = remoteQuery;

    return Futures.whenAllSucceed(remoteQueryFinal, diskQueryFinal)
        .call(
            () ->
                ImmutableSet.<Digest>builder()
                    .addAll(remoteQueryFinal.get())
                    .addAll(diskQueryFinal.get())
                    .build(),
            directExecutor());
  }

  /** Returns whether the remote action cache supports updating action results. */
  public boolean remoteActionCacheSupportsUpdate() {
    try {
      return getRemoteCacheCapabilities().getActionCacheUpdateCapabilities().getUpdateEnabled();
    } catch (IOException ignored) {
      return false;
    }
  }

  /** Upload the action result to the remote cache. */
  public ListenableFuture<Void> uploadActionResult(
      RemoteActionExecutionContext context, ActionKey actionKey, ActionResult actionResult) {
    ListenableFuture<Void> diskCacheFuture = Futures.immediateVoidFuture();
    if (diskCacheClient != null && context.getWriteCachePolicy().allowDiskCache()) {
      diskCacheFuture = diskCacheClient.uploadActionResult(actionKey, actionResult);
    }

    ListenableFuture<Void> remoteCacheFuture = Futures.immediateVoidFuture();
    if (remoteCacheClient != null && context.getWriteCachePolicy().allowRemoteCache()) {
      remoteCacheFuture = remoteCacheClient.uploadActionResult(context, actionKey, actionResult);
    }

    return Futures.whenAllSucceed(diskCacheFuture, remoteCacheFuture)
        .call(() -> null, directExecutor());
  }

  /**
   * Upload a local file to the remote cache.
   *
   * <p>Trying to upload the same file multiple times concurrently, results in only one upload being
   * performed.
   *
   * @param context the context for the action.
   * @param digest the digest of the file.
   * @param file the file to upload.
   */
  public ListenableFuture<Void> uploadFile(
      RemoteActionExecutionContext context, Digest digest, Path file) {
    return uploadFile(context, digest, file, /* force= */ false);
  }

  protected ListenableFuture<Void> uploadFile(
      RemoteActionExecutionContext context, Digest digest, Path file, boolean force) {
    if (digest.getSizeBytes() == 0) {
      return COMPLETED_SUCCESS;
    }

    ListenableFuture<Void> diskCacheFuture = Futures.immediateVoidFuture();
    if (diskCacheClient != null && context.getWriteCachePolicy().allowDiskCache()) {
      diskCacheFuture = diskCacheClient.uploadFile(digest, file);
    }

    ListenableFuture<Void> remoteCacheFuture = Futures.immediateVoidFuture();
    if (remoteCacheClient != null && context.getWriteCachePolicy().allowRemoteCache()) {
      Completable upload =
          casUploadCache.execute(
              digest,
              RxFutures.toCompletable(
                  () -> remoteCacheClient.uploadFile(context, digest, file), directExecutor()),
              force);
      remoteCacheFuture = RxFutures.toListenableFuture(upload);
    }

    return Futures.whenAllSucceed(diskCacheFuture, remoteCacheFuture)
        .call(() -> null, directExecutor());
  }

  /**
   * Upload sequence of bytes to the remote cache.
   *
   * <p>Trying to upload the same BLOB multiple times concurrently, results in only one upload being
   * performed.
   *
   * @param context the context for the action.
   * @param digest the digest of the file.
   * @param data the BLOB to upload.
   */
  public ListenableFuture<Void> uploadBlob(
      RemoteActionExecutionContext context, Digest digest, ByteString data) {
    return uploadBlob(context, digest, data, /* force= */ false);
  }

  protected ListenableFuture<Void> uploadBlob(
      RemoteActionExecutionContext context, Digest digest, ByteString data, boolean force) {
    if (digest.getSizeBytes() == 0) {
      return COMPLETED_SUCCESS;
    }

    ListenableFuture<Void> diskCacheFuture = Futures.immediateVoidFuture();
    if (diskCacheClient != null && context.getWriteCachePolicy().allowDiskCache()) {
      diskCacheFuture = diskCacheClient.uploadBlob(digest, data);
    }

    ListenableFuture<Void> remoteCacheFuture = Futures.immediateVoidFuture();
    if (remoteCacheClient != null && context.getWriteCachePolicy().allowRemoteCache()) {
      Completable upload =
          casUploadCache.execute(
              digest,
              RxFutures.toCompletable(
                  () -> remoteCacheClient.uploadBlob(context, digest, data), directExecutor()),
              force);

      remoteCacheFuture = RxFutures.toListenableFuture(upload);
    }

    return Futures.whenAllSucceed(diskCacheFuture, remoteCacheFuture)
        .call(() -> null, directExecutor());
  }

  public ListenableFuture<byte[]> downloadBlob(
      RemoteActionExecutionContext context, Digest digest) {
    return downloadBlob(context, /* blobName= */ "", digest);
  }

  /**
   * Downloads a blob with content hash {@code digest} and stores its content in memory.
   *
   * @return a future that completes after the download completes (succeeds / fails). If successful,
   *     the content is stored in the future's {@code byte[]}.
   */
  public ListenableFuture<byte[]> downloadBlob(
      RemoteActionExecutionContext context, String blobName, Digest digest) {
    if (digest.getSizeBytes() == 0) {
      return EMPTY_BYTES;
    }
    ByteArrayOutputStream bOut = new ByteArrayOutputStream((int) digest.getSizeBytes());
    var download = downloadBlob(context, blobName, digest, bOut);
    return Futures.transform(download, (v) -> bOut.toByteArray(), directExecutor());
  }

  private ListenableFuture<Void> downloadBlob(
      RemoteActionExecutionContext context, String blobName, Digest digest, OutputStream out) {
    if (digest.getSizeBytes() == 0) {
      return COMPLETED_SUCCESS;
    }
    var future = downloadBlob(context, digest, out);
    return Futures.catchingAsync(
        future,
        CacheNotFoundException.class,
        (cacheNotFoundException) -> {
          cacheNotFoundException.setFilename(blobName);
          return immediateFailedFuture(cacheNotFoundException);
        },
        directExecutor());
  }

  private ListenableFuture<Void> downloadBlob(
      RemoteActionExecutionContext context, Digest digest, OutputStream out) {
    ListenableFuture<Void> future = immediateFailedFuture(new CacheNotFoundException(digest));

    if (diskCacheClient != null && context.getReadCachePolicy().allowDiskCache()) {
      future = diskCacheClient.downloadBlob(digest, out);
    }

    if (remoteCacheClient != null && context.getReadCachePolicy().allowRemoteCache()) {
      future =
          Futures.catchingAsync(
              future,
              CacheNotFoundException.class,
              (unused) -> downloadBlobFromRemote(context, digest, out),
              directExecutor());
    }

    return future;
  }

  private ListenableFuture<Void> downloadBlobFromRemote(
      RemoteActionExecutionContext context, Digest digest, OutputStream out) {
    checkState(remoteCacheClient != null && context.getReadCachePolicy().allowRemoteCache());

    if (diskCacheClient != null && context.getWriteCachePolicy().allowDiskCache()) {
      Path tempPath = diskCacheClient.getTempPath();
      LazyFileOutputStream tempOut = new LazyFileOutputStream(tempPath);
      ListenableFuture<Void> download = remoteCacheClient.downloadBlob(context, digest, tempOut);
      return cleanupTempFileOnError(
          Futures.transformAsync(
              download,
              (unused) -> {
                try {
                  // Fsync temp before we rename it to avoid data loss in the case of machine
                  // crashes (the OS may reorder the writes and the rename).
                  tempOut.syncIfPossible();
                  tempOut.close();
                  diskCacheClient.captureFile(tempPath, digest, Store.CAS);
                } catch (IOException e) {
                  return immediateFailedFuture(e);
                }
                return diskCacheClient.downloadBlob(digest, out);
              },
              directExecutor()),
          tempPath,
          tempOut);
    }

    return remoteCacheClient.downloadBlob(context, digest, out);
  }

  private static ListenableFuture<Void> cleanupTempFileOnError(
      ListenableFuture<Void> f, Path tempPath, OutputStream tempOut) {
    return Futures.catchingAsync(
        f,
        Exception.class,
        (rootCause) -> {
          try {
            tempOut.close();
          } catch (IOException e) {
            rootCause.addSuppressed(e);
          }
          try {
            tempPath.delete();
          } catch (IOException e) {
            rootCause.addSuppressed(e);
          }
          return immediateFailedFuture(rootCause);
        },
        directExecutor());
  }

  /** A reporter that reports download progresses. */
  public static class DownloadProgressReporter {
    private static final Pattern PATTERN = Pattern.compile("^bazel-out/[^/]+/[^/]+/");
    private final boolean includeFile;
    private final ProgressStatusListener listener;
    private final String id;
    private final String file;
    private final String totalSize;
    private final AtomicLong downloadedBytes = new AtomicLong(0);

    public DownloadProgressReporter(ProgressStatusListener listener, String file, long totalSize) {
      this(/* includeFile= */ true, listener, file, totalSize);
    }

    public DownloadProgressReporter(
        boolean includeFile, ProgressStatusListener listener, String file, long totalSize) {
      this.includeFile = includeFile;
      this.listener = listener;
      this.id = file;
      this.totalSize = bytesCountToDisplayString(totalSize);

      Matcher matcher = PATTERN.matcher(file);
      this.file = matcher.replaceFirst("");
    }

    void started() {
      reportProgress(false, false);
    }

    void downloadedBytes(int count) {
      downloadedBytes.addAndGet(count);
      reportProgress(true, false);
    }

    void finished() {
      reportProgress(true, true);
    }

    private void reportProgress(boolean includeBytes, boolean finished) {
      String progress;
      if (includeBytes) {
        if (includeFile) {
          progress =
              String.format(
                  "Downloading %s, %s / %s",
                  file, bytesCountToDisplayString(downloadedBytes.get()), totalSize);
        } else {
          progress =
              String.format("%s / %s", bytesCountToDisplayString(downloadedBytes.get()), totalSize);
        }
      } else {
        if (includeFile) {
          progress = String.format("Downloading %s", file);
        } else {
          progress = "";
        }
      }
      listener.onProgressStatus(SpawnProgressEvent.create(id, progress, finished));
    }
  }

  public ListenableFuture<Void> downloadFile(
      RemoteActionExecutionContext context,
      String outputPath,
      Path localPath,
      Digest digest,
      DownloadProgressReporter reporter)
      throws IOException {
    ListenableFuture<Void> f = downloadFile(context, localPath, digest, reporter);
    return Futures.catchingAsync(
        f,
        Throwable.class,
        (throwable) -> {
          if (throwable instanceof CacheNotFoundException cacheNotFoundException) {
            cacheNotFoundException.setFilename(outputPath);
          } else if (throwable instanceof OutputDigestMismatchException e) {
            e.setOutputPath(outputPath);
            e.setLocalPath(localPath);
          }
          return immediateFailedFuture(throwable);
        },
        directExecutor());
  }

  /** Downloads a file (that is not a directory). The content is fetched from the digest. */
  public ListenableFuture<Void> downloadFile(
      RemoteActionExecutionContext context, Path path, Digest digest) throws IOException {
    return downloadFile(
        context,
        path.getPathString(),
        path,
        digest,
        new DownloadProgressReporter(NO_ACTION, "", 0));
  }

  /** Downloads a file (that is not a directory). The content is fetched from the digest. */
  private ListenableFuture<Void> downloadFile(
      RemoteActionExecutionContext context,
      Path path,
      Digest digest,
      DownloadProgressReporter reporter)
      throws IOException {
    checkNotNull(path.getParentDirectory()).createDirectoryAndParents();
    if (digest.getSizeBytes() == 0) {
      // Handle empty file locally.
      FileSystemUtils.writeContent(path, new byte[0]);
      return COMPLETED_SUCCESS;
    }

    if (!options.remoteDownloadSymlinkTemplate.isEmpty()) {
      // Don't actually download files from the CAS. Instead, create a
      // symbolic link that points to a location where CAS objects may
      // be found. This could, for example, be a FUSE file system.
      path.createSymbolicLink(
          path.getRelative(
              options
                  .remoteDownloadSymlinkTemplate
                  .replace("{hash}", digest.getHash())
                  .replace("{size_bytes}", String.valueOf(digest.getSizeBytes()))));
      return COMPLETED_SUCCESS;
    }

    reporter.started();
    OutputStream out = new ReportingOutputStream(new LazyFileOutputStream(path), reporter);

    ListenableFuture<Void> f = downloadBlob(context, digest, out);
    f.addListener(
        () -> {
          try {
            out.close();
          } catch (IOException e) {
            logger.atWarning().withCause(e).log(
                "Unexpected exception closing output stream after downloading %s/%d to %s",
                digest.getHash(), digest.getSizeBytes(), path);
          } finally {
            reporter.finished();
          }
        },
        directExecutor());
    return f;
  }

  /**
   * Download the stdout and stderr of an executed action.
   *
   * @param context the context for the action.
   * @param result the result of the action.
   * @param outErr the {@link OutErr} that the stdout and stderr will be downloaded to.
   */
  public final List<ListenableFuture<Void>> downloadOutErr(
      RemoteActionExecutionContext context, ActionResult result, OutErr outErr) {
    List<ListenableFuture<Void>> downloads = new ArrayList<>();
    if (!result.getStdoutRaw().isEmpty()) {
      try {
        result.getStdoutRaw().writeTo(outErr.getOutputStream());
        outErr.getOutputStream().flush();
      } catch (IOException e) {
        downloads.add(Futures.immediateFailedFuture(e));
      }
    } else if (result.hasStdoutDigest()) {
      downloads.add(
          downloadBlob(
              context,
              /* blobName= */ "<stdout>",
              result.getStdoutDigest(),
              outErr.getOutputStream()));
    }
    if (!result.getStderrRaw().isEmpty()) {
      try {
        result.getStderrRaw().writeTo(outErr.getErrorStream());
        outErr.getErrorStream().flush();
      } catch (IOException e) {
        downloads.add(Futures.immediateFailedFuture(e));
      }
    } else if (result.hasStderrDigest()) {
      downloads.add(
          downloadBlob(
              context,
              /* blobName= */ "<stderr>",
              result.getStderrDigest(),
              outErr.getErrorStream()));
    }
    return downloads;
  }

  public boolean hasRemoteCache() {
    return remoteCacheClient != null;
  }

  public boolean hasDiskCache() {
    return diskCacheClient != null;
  }

  @Override
  protected void deallocate() {
    if (diskCacheClient != null) {
      diskCacheClient.close();
    }
    casUploadCache.shutdown();
    if (remoteCacheClient != null) {
      remoteCacheClient.close();
    }

    closeCountDownLatch.countDown();
  }

  @Override
  public CombinedCache touch(Object o) {
    return this;
  }

  @CanIgnoreReturnValue
  @Override
  public CombinedCache retain() {
    super.retain();
    return this;
  }

  /** Waits for active network I/Os to finish. */
  public void awaitTermination() throws InterruptedException {
    casUploadCache.awaitTermination();
    closeCountDownLatch.await();
  }

  /** Shuts the cache down and cancels active network I/Os. */
  public void shutdownNow() {
    casUploadCache.shutdownNow();
  }

  public static FailureDetail createFailureDetail(String message, Code detailedCode) {
    return FailureDetail.newBuilder()
        .setMessage(message)
        .setRemoteExecution(RemoteExecution.newBuilder().setCode(detailedCode))
        .build();
  }

  /**
   * An {@link OutputStream} that reports all the write operations with {@link
   * DownloadProgressReporter}.
   */
  private static class ReportingOutputStream extends OutputStream {

    private final OutputStream out;
    private final DownloadProgressReporter reporter;

    ReportingOutputStream(OutputStream out, DownloadProgressReporter reporter) {
      this.out = out;
      this.reporter = reporter;
    }

    @Override
    public void write(byte[] b) throws IOException {
      out.write(b);
      reporter.downloadedBytes(b.length);
    }

    @Override
    public void write(byte[] b, int off, int len) throws IOException {
      out.write(b, off, len);
      reporter.downloadedBytes(len);
    }

    @Override
    public void write(int b) throws IOException {
      out.write(b);
      reporter.downloadedBytes(1);
    }

    @Override
    public void flush() throws IOException {
      out.flush();
    }

    @Override
    public void close() throws IOException {
      out.close();
    }
  }
}