Retry on HTTP remote cache fetch failure

Bazel's previous behavior was to rebuild an artifact locally if fetching
it from an HTTP remote cache failed. This behavior is different from
GRPC remote cache case where Bazel will retry the fetch.

The lack of retry is an issue for multiple reasons: On one hand
rebuilding locally can be slower than fetching from the remote cache, on
the other hand if a build action is not bit reproducible, as is the case
with some compilers, then the local rebuild will trigger cache misses on
further build actions that depend on the current artifact.

This change aims to avoid theses issues by retrying the fetch in the
HTTP cache case similarly to how the GRPC cache client does it.

Some care needs to be taken due to the design of Bazel's internal remote
cache client API. For a fetch the client is given an `OutputStream`
object that it is expected to write the fetched data to. This may be a
temporary file on disk that will be moved to the final location after
the fetch completed. On retry, we need to be careful to not duplicate
previously written data when writing into this `OutputStream`. Due to
the generality of the `OutputStream` interface we cannot reset the file
handle or write pointer to start fresh. Instead, this change follows the
same pattern used in the GRPC cache client. Namely, keep track of the
data previously written and continue from that offset on retry.

With this change the HTTP cache client will attempt to fetch the data
from the remote cache via an HTTP range request. So that the server only
needs to send the data that is still missing. If the server replies with
a 206 Partial Content response, then we write the received data directly
into the output stream, if the server does not support range requests
and instead replies with the full data, then we drop the duplicate
prefix and only write into the output stream from the required offset.

This patch has been running successfully in production [here](digital-asset/daml#11238).

cc @cocreature

Closes #14258.

PiperOrigin-RevId: 508604846
Change-Id: I10a5d2a658e9c32a9d9fcd6bd29f6a0b95e84566

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

@@ -59,15 +59,22 @@ public static RemoteCacheClient create(
       @Nullable Credentials creds,
       AuthAndTLSOptions authAndTlsOptions,
       Path workingDirectory,
-      DigestUtil digestUtil)
+      DigestUtil digestUtil,
+      RemoteRetrier retrier)
       throws IOException {
     Preconditions.checkNotNull(workingDirectory, "workingDirectory");
     if (isHttpCache(options) && isDiskCache(options)) {
       return createDiskAndHttpCache(
-          workingDirectory, options.diskCache, options, creds, authAndTlsOptions, digestUtil);
+          workingDirectory,
+          options.diskCache,
+          options,
+          creds,
+          authAndTlsOptions,
+          digestUtil,
+          retrier);
     }
     if (isHttpCache(options)) {
-      return createHttp(options, creds, authAndTlsOptions, digestUtil);
+      return createHttp(options, creds, authAndTlsOptions, digestUtil, retrier);
     }
     if (isDiskCache(options)) {
       return createDiskCache(
@@ -90,7 +97,8 @@ private static RemoteCacheClient createHttp(
       RemoteOptions options,
       Credentials creds,
       AuthAndTLSOptions authAndTlsOptions,
-      DigestUtil digestUtil) {
+      DigestUtil digestUtil,
+      RemoteRetrier retrier) {
     Preconditions.checkNotNull(options.remoteCache, "remoteCache");
 
     try {
@@ -109,6 +117,7 @@ private static RemoteCacheClient createHttp(
               options.remoteVerifyDownloads,
               ImmutableList.copyOf(options.remoteHeaders),
               digestUtil,
+              retrier,
               creds,
               authAndTlsOptions);
         } else {
@@ -122,6 +131,7 @@ private static RemoteCacheClient createHttp(
             options.remoteVerifyDownloads,
             ImmutableList.copyOf(options.remoteHeaders),
             digestUtil,
+            retrier,
             creds,
             authAndTlsOptions);
       }
@@ -151,15 +161,16 @@ private static RemoteCacheClient createDiskAndHttpCache(
       RemoteOptions options,
       Credentials cred,
       AuthAndTLSOptions authAndTlsOptions,
-      DigestUtil digestUtil)
+      DigestUtil digestUtil,
+      RemoteRetrier retrier)
       throws IOException {
     Path cacheDir =
         workingDirectory.getRelative(Preconditions.checkNotNull(diskCachePath, "diskCachePath"));
     if (!cacheDir.exists()) {
       cacheDir.createDirectoryAndParents();
     }
 
-    RemoteCacheClient httpCache = createHttp(options, cred, authAndTlsOptions, digestUtil);
+    RemoteCacheClient httpCache = createHttp(options, cred, authAndTlsOptions, digestUtil, retrier);
     return createDiskAndRemoteClient(
         workingDirectory,
         diskCachePath,

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

@@ -62,6 +62,7 @@
 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.downloader.GrpcRemoteDownloader;
+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.options.RemoteBuildEventUploadMode;
 import com.google.devtools.build.lib.remote.options.RemoteOptions;
@@ -101,17 +102,20 @@
 import io.grpc.Channel;
 import io.grpc.ClientInterceptor;
 import io.grpc.ManagedChannel;
+import io.netty.handler.codec.http.HttpResponseStatus;
 import io.reactivex.rxjava3.plugins.RxJavaPlugins;
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.nio.channels.ClosedChannelException;
 import java.util.List;
 import java.util.Optional;
 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 java.util.regex.Pattern;
 import javax.annotation.Nullable;
 
@@ -216,6 +220,29 @@ private static ServerCapabilities getAndVerifyServerCapabilities(
     return capabilities;
   }
 
+  public static final Predicate<? super Exception> RETRIABLE_HTTP_ERRORS =
+      e -> {
+        boolean retry = false;
+        if (e instanceof ClosedChannelException) {
+          retry = true;
+        } else if (e instanceof HttpException) {
+          int status = ((HttpException) e).response().status().code();
+          retry =
+              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 = e.getMessage().toLowerCase();
+          if (msg.contains("connection reset by peer")) {
+            retry = true;
+          } else if (msg.contains("operation timed out")) {
+            retry = true;
+          }
+        }
+        return retry;
+      };
+
   private void initHttpAndDiskCache(
       CommandEnvironment env,
       Credentials credentials,
@@ -230,7 +257,9 @@ private void initHttpAndDiskCache(
               credentials,
               authAndTlsOptions,
               Preconditions.checkNotNull(env.getWorkingDirectory(), "workingDirectory"),
-              digestUtil);
+              digestUtil,
+              new RemoteRetrier(
+                  remoteOptions, RETRIABLE_HTTP_ERRORS, retryScheduler, Retrier.ALLOW_ALL_CALLS));
     } catch (IOException e) {
       handleInitFailure(env, e, Code.CACHE_INIT_FAILURE);
       return;

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

@@ -20,6 +20,7 @@ java_library(
     deps = [
         "//src/main/java/com/google/devtools/build/lib/analysis:blaze_version_info",
         "//src/main/java/com/google/devtools/build/lib/authandtls",
+        "//src/main/java/com/google/devtools/build/lib/remote:Retrier",
         "//src/main/java/com/google/devtools/build/lib/remote/common",
         "//src/main/java/com/google/devtools/build/lib/remote/util",
         "//src/main/java/com/google/devtools/build/lib/vfs",

src/main/java/com/google/devtools/build/lib/remote/http/DownloadCommand.java

@@ -25,12 +25,18 @@ final class DownloadCommand {
   private final boolean casDownload;
   private final Digest digest;
   private final OutputStream out;
+  private final long offset;
 
-  DownloadCommand(URI uri, boolean casDownload, Digest digest, OutputStream out) {
+  DownloadCommand(URI uri, boolean casDownload, Digest digest, OutputStream out, long offset) {
     this.uri = Preconditions.checkNotNull(uri);
     this.casDownload = casDownload;
     this.digest = Preconditions.checkNotNull(digest);
     this.out = Preconditions.checkNotNull(out);
+    this.offset = offset;
+  }
+
+  DownloadCommand(URI uri, boolean casDownload, Digest digest, OutputStream out) {
+    this(uri, casDownload, digest, out, 0);
   }
 
   public URI uri() {
@@ -48,4 +54,8 @@ public Digest digest() {
   public OutputStream out() {
     return out;
   }
+
+  public long offset() {
+    return offset;
+  }
 }

src/main/java/com/google/devtools/build/lib/remote/http/HttpCacheClient.java

@@ -25,6 +25,7 @@
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.SettableFuture;
 import com.google.devtools.build.lib.authandtls.AuthAndTLSOptions;
+import com.google.devtools.build.lib.remote.RemoteRetrier;
 import com.google.devtools.build.lib.remote.common.CacheNotFoundException;
 import com.google.devtools.build.lib.remote.common.RemoteActionExecutionContext;
 import com.google.devtools.build.lib.remote.common.RemoteCacheClient;
@@ -81,8 +82,10 @@
 import java.util.List;
 import java.util.Map.Entry;
 import java.util.NoSuchElementException;
+import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Function;
 import java.util.regex.Pattern;
 import javax.annotation.Nullable;
@@ -129,6 +132,7 @@ public final class HttpCacheClient implements RemoteCacheClient {
   private final boolean useTls;
   private final boolean verifyDownloads;
   private final DigestUtil digestUtil;
+  private final RemoteRetrier retrier;
 
   private final Object closeLock = new Object();
 
@@ -150,6 +154,7 @@ public static HttpCacheClient create(
       boolean verifyDownloads,
       ImmutableList<Entry<String, String>> extraHttpHeaders,
       DigestUtil digestUtil,
+      RemoteRetrier retrier,
       @Nullable final Credentials creds,
       AuthAndTLSOptions authAndTlsOptions)
       throws Exception {
@@ -162,6 +167,7 @@ public static HttpCacheClient create(
         verifyDownloads,
         extraHttpHeaders,
         digestUtil,
+        retrier,
         creds,
         authAndTlsOptions,
         null);
@@ -175,6 +181,7 @@ public static HttpCacheClient create(
       boolean verifyDownloads,
       ImmutableList<Entry<String, String>> extraHttpHeaders,
       DigestUtil digestUtil,
+      RemoteRetrier retrier,
       @Nullable final Credentials creds,
       AuthAndTLSOptions authAndTlsOptions)
       throws Exception {
@@ -189,6 +196,7 @@ public static HttpCacheClient create(
           verifyDownloads,
           extraHttpHeaders,
           digestUtil,
+          retrier,
           creds,
           authAndTlsOptions,
           domainSocketAddress);
@@ -202,6 +210,7 @@ public static HttpCacheClient create(
           verifyDownloads,
           extraHttpHeaders,
           digestUtil,
+          retrier,
           creds,
           authAndTlsOptions,
           domainSocketAddress);
@@ -219,6 +228,7 @@ private HttpCacheClient(
       boolean verifyDownloads,
       ImmutableList<Entry<String, String>> extraHttpHeaders,
       DigestUtil digestUtil,
+      RemoteRetrier retrier,
       @Nullable final Credentials creds,
       AuthAndTLSOptions authAndTlsOptions,
       @Nullable SocketAddress socketAddress)
@@ -284,6 +294,7 @@ public void channelCreated(Channel ch) {
     this.extraHttpHeaders = extraHttpHeaders;
     this.verifyDownloads = verifyDownloads;
     this.digestUtil = digestUtil;
+    this.retrier = retrier;
   }
 
   @SuppressWarnings("FutureReturnValueIgnored")
@@ -441,8 +452,11 @@ public ListenableFuture<Void> downloadBlob(
       RemoteActionExecutionContext context, Digest digest, OutputStream out) {
     final DigestOutputStream digestOut =
         verifyDownloads ? digestUtil.newDigestOutputStream(out) : null;
+    final AtomicLong casBytesDownloaded = new AtomicLong();
     return Futures.transformAsync(
-        get(digest, digestOut != null ? digestOut : out, /* casDownload= */ true),
+        retrier.executeAsync(
+            () ->
+                get(digest, digestOut != null ? digestOut : out, Optional.of(casBytesDownloaded))),
         (v) -> {
           try {
             if (digestOut != null) {
@@ -458,7 +472,8 @@ public ListenableFuture<Void> downloadBlob(
   }
 
   @SuppressWarnings("FutureReturnValueIgnored")
-  private ListenableFuture<Void> get(Digest digest, final OutputStream out, boolean casDownload) {
+  private ListenableFuture<Void> get(
+      Digest digest, final OutputStream out, Optional<AtomicLong> casBytesDownloaded) {
     final AtomicBoolean dataWritten = new AtomicBoolean();
     OutputStream wrappedOut =
         new OutputStream() {
@@ -469,12 +484,18 @@ private ListenableFuture<Void> get(Digest digest, final OutputStream out, boolea
           @Override
           public void write(byte[] b, int offset, int length) throws IOException {
             dataWritten.set(true);
+            if (casBytesDownloaded.isPresent()) {
+              casBytesDownloaded.get().addAndGet(length);
+            }
             out.write(b, offset, length);
           }
 
           @Override
           public void write(int b) throws IOException {
             dataWritten.set(true);
+            if (casBytesDownloaded.isPresent()) {
+              casBytesDownloaded.get().incrementAndGet();
+            }
             out.write(b);
           }
 
@@ -483,7 +504,12 @@ public void flush() throws IOException {
             out.flush();
           }
         };
-    DownloadCommand downloadCmd = new DownloadCommand(uri, casDownload, digest, wrappedOut);
+    long offset = 0;
+    if (casBytesDownloaded.isPresent()) {
+      offset = casBytesDownloaded.get().get();
+    }
+    DownloadCommand downloadCmd =
+        new DownloadCommand(uri, casBytesDownloaded.isPresent(), digest, wrappedOut, offset);
     SettableFuture<Void> outerF = SettableFuture.create();
     acquireDownloadChannel()
         .addListener(
@@ -575,8 +601,11 @@ private void getAfterCredentialRefresh(DownloadCommand cmd, SettableFuture<Void>
   public ListenableFuture<CachedActionResult> downloadActionResult(
       RemoteActionExecutionContext context, ActionKey actionKey, boolean inlineOutErr) {
     return Futures.transform(
-        Utils.downloadAsActionResult(
-            actionKey, (digest, out) -> get(digest, out, /* casDownload= */ false)),
+        retrier.executeAsync(
+            () ->
+                Utils.downloadAsActionResult(
+                    actionKey,
+                    (digest, out) -> get(digest, out, /* casBytesDownloaded= */ Optional.empty()))),
         CachedActionResult::remote,
         MoreExecutors.directExecutor());
   }
@@ -670,20 +699,28 @@ private void uploadAfterCredentialRefresh(UploadCommand upload, SettableFuture<V
   @Override
   public ListenableFuture<Void> uploadFile(
       RemoteActionExecutionContext context, Digest digest, Path file) {
-    try {
-      return uploadAsync(
-          digest.getHash(), digest.getSizeBytes(), file.getInputStream(), /* casUpload= */ true);
-    } catch (IOException e) {
-      // Can be thrown from file.getInputStream.
-      return Futures.immediateFailedFuture(e);
-    }
+    return retrier.executeAsync(
+        () -> {
+          try {
+            return uploadAsync(
+                digest.getHash(),
+                digest.getSizeBytes(),
+                file.getInputStream(),
+                /* casUpload= */ true);
+          } catch (IOException e) {
+            // Can be thrown from file.getInputStream.
+            return Futures.immediateFailedFuture(e);
+          }
+        });
   }
 
   @Override
   public ListenableFuture<Void> uploadBlob(
       RemoteActionExecutionContext context, Digest digest, ByteString data) {
-    return uploadAsync(
-        digest.getHash(), digest.getSizeBytes(), data.newInput(), /* casUpload= */ true);
+    return retrier.executeAsync(
+        () ->
+            uploadAsync(
+                digest.getHash(), digest.getSizeBytes(), data.newInput(), /* casUpload= */ true));
   }
 
   @Override