Retry on HTTP remote cache fetch failure

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

@@ -59,15 +59,16 @@ 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 +91,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 +111,7 @@ private static RemoteCacheClient createHttp(
               options.remoteVerifyDownloads,
               ImmutableList.copyOf(options.remoteHeaders),
               digestUtil,
+              retrier,
               creds,
               authAndTlsOptions);
         } else {
@@ -122,6 +125,7 @@ private static RemoteCacheClient createHttp(
             options.remoteVerifyDownloads,
             ImmutableList.copyOf(options.remoteHeaders),
             digestUtil,
+            retrier,
             creds,
             authAndTlsOptions);
       }
@@ -151,15 +155,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

@@ -59,9 +59,11 @@
 import com.google.devtools.build.lib.exec.ModuleActionContextRegistry;
 import com.google.devtools.build.lib.exec.SpawnStrategyRegistry;
 import com.google.devtools.build.lib.remote.RemoteServerCapabilities.ServerCapabilitiesRequirement;
+import com.google.devtools.build.lib.remote.common.CacheNotFoundException;
 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 +103,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 +221,28 @@ 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) {
+              HttpResponseStatus status = ((HttpException) e).response().status();
+              retry = status == HttpResponseStatus.INTERNAL_SERVER_ERROR
+                  || status == HttpResponseStatus.BAD_GATEWAY
+                  || status == HttpResponseStatus.SERVICE_UNAVAILABLE
+                  || status == HttpResponseStatus.GATEWAY_TIMEOUT;
+            } 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,13 @@ 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

@@ -21,6 +21,7 @@ java_library(
         "//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/common",
+        "//src/main/java/com/google/devtools/build/lib/remote:Retrier",
         "//src/main/java/com/google/devtools/build/lib/remote/util",
         "//src/main/java/com/google/devtools/build/lib/vfs",
         "//third_party:auth",

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,6 @@ 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,7 @@ 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 +483,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 +503,11 @@ 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 +599,10 @@ 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 +696,21 @@ 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