Implementation (but not plumbing) of the gRPC remote downloader

Extracted from #10622

Per discussion on that PR, there's still some unanswered questions about how exactly we plumb the new `Downloader` type into `RemoteModule`. And per #10742 (comment), it is unlikely that even heroic effort from me will get the full end-to-end functionality into v3.0.

Given this, to simplify the review, I'm taking some of the bits the reviewer is happy with and moving them to a separate PR. After merger, `GrpcRemoteDownloader` and its tests will exist in the source tree, but will not yet be available as CLI options.

R: @michajlo
CC: @adunham-stripe @dslomov @EricBurnett  @philwo @sstriker

Closes #10914.

PiperOrigin-RevId: 299908615

src/main/java/com/google/devtools/build/lib/bazel/repository/downloader/DownloadManager.java

@@ -187,7 +187,7 @@ public Path download(
 
     try {
       downloader.download(
-          urls, authHeaders, checksum, destination, eventHandler, clientEnv);
+          urls, authHeaders, checksum, canonicalId, destination, eventHandler, clientEnv);
     } catch (InterruptedIOException e) {
       throw new InterruptedException(e.getMessage());
     }

src/main/java/com/google/devtools/build/lib/bazel/repository/downloader/Downloader.java

@@ -43,6 +43,7 @@ void download(
       List<URL> urls,
       Map<URI, Map<String, String>> authHeaders,
       Optional<Checksum> checksum,
+      String canonicalId,
       Path output,
       ExtendedEventHandler eventHandler,
       Map<String, String> clientEnv)

src/main/java/com/google/devtools/build/lib/bazel/repository/downloader/HashOutputStream.java

@@ -0,0 +1,89 @@
+// Copyright 2020 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.bazel.repository.downloader;
+
+import com.google.common.hash.HashCode;
+import com.google.common.hash.Hasher;
+import com.google.devtools.build.lib.concurrent.ThreadSafety.ThreadCompatible;
+import java.io.IOException;
+import java.io.OutputStream;
+import javax.annotation.Nullable;
+import javax.annotation.WillCloseWhenClosed;
+
+/**
+ * Output stream that guarantees its contents matches a hash code.
+ *
+ * <p>The actual checksum is computed gradually as the output is written. If it doesn't match, then
+ * an {@link IOException} will be thrown when {@link #close()} is called. This error will be thrown
+ * multiple times if these methods are called again for some reason.
+ *
+ * <p>Note that as the checksum can only be computed once the stream is closed, data will be written
+ * to the underlying stream regardless of whether it matches the expected checksum.
+ *
+ * <p>This class is not thread safe, but it is safe to message pass this object between threads.
+ */
+@ThreadCompatible
+public final class HashOutputStream extends OutputStream {
+
+  private final OutputStream delegate;
+  private final Hasher hasher;
+  private final HashCode code;
+  @Nullable private volatile HashCode actual;
+
+  public HashOutputStream(@WillCloseWhenClosed OutputStream delegate, Checksum checksum) {
+    this.delegate = delegate;
+    this.hasher = checksum.getKeyType().newHasher();
+    this.code = checksum.getHashCode();
+  }
+
+  @Override
+  public void write(int buffer) throws IOException {
+    hasher.putByte((byte) buffer);
+    delegate.write(buffer);
+  }
+
+  @Override
+  public void write(byte[] buffer) throws IOException {
+    hasher.putBytes(buffer);
+    delegate.write(buffer);
+  }
+
+  @Override
+  public void write(byte[] buffer, int offset, int length) throws IOException {
+    hasher.putBytes(buffer, offset, length);
+    delegate.write(buffer, offset, length);
+  }
+
+  @Override
+  public void flush() throws IOException {
+    delegate.flush();
+  }
+
+  @Override
+  public void close() throws IOException {
+    delegate.close();
+    check();
+  }
+
+  private void check() throws IOException {
+    if (actual == null) {
+      actual = hasher.hash();
+    }
+    if (!code.equals(actual)) {
+      throw new UnrecoverableHttpException(
+          String.format("Checksum was %s but wanted %s", actual, code));
+    }
+  }
+}

src/main/java/com/google/devtools/build/lib/bazel/repository/downloader/HttpDownloader.java

@@ -62,6 +62,7 @@ public void download(
       List<URL> urls,
       Map<URI, Map<String, String>> authHeaders,
       Optional<Checksum> checksum,
+      String canonicalId,
       Path destination,
       ExtendedEventHandler eventHandler,
       Map<String, String> clientEnv)

src/main/java/com/google/devtools/build/lib/bazel/repository/downloader/UnrecoverableHttpException.java

@@ -16,7 +16,8 @@
 
 import java.io.IOException;
 
-final class UnrecoverableHttpException extends IOException {
+/** Indicates an HTTP error that cannot be recovered from. */
+public final class UnrecoverableHttpException extends IOException {
   UnrecoverableHttpException(String message) {
     super(message);
   }

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

@@ -6,6 +6,7 @@ filegroup(
     name = "srcs",
     srcs = glob(["**"]) + [
         "//src/main/java/com/google/devtools/build/lib/remote/common:srcs",
+        "//src/main/java/com/google/devtools/build/lib/remote/downloader:srcs",
         "//src/main/java/com/google/devtools/build/lib/remote/disk:srcs",
         "//src/main/java/com/google/devtools/build/lib/remote/http:srcs",
         "//src/main/java/com/google/devtools/build/lib/remote/logging:srcs",
@@ -18,13 +19,30 @@ filegroup(
 
 java_library(
     name = "remote",
-    srcs = glob(["*.java"]),
+    srcs = glob(
+        ["*.java"],
+        exclude = [
+            "ExecutionStatusException.java",
+            "ReferenceCountedChannel.java",
+            "RemoteRetrier.java",
+            "RemoteRetrierUtils.java",
+            "Retrier.java",
+        ],
+    ),
     tags = ["bazel"],
+    exports = [
+        ":ExecutionStatusException",
+        ":ReferenceCountedChannel",
+        ":Retrier",
+    ],
     runtime_deps = [
         # This is required for client TLS.
         "//third_party:netty_tcnative",
     ],
     deps = [
+        ":ExecutionStatusException",
+        ":ReferenceCountedChannel",
+        ":Retrier",
         "//src/main/java/com/google/devtools/build/lib:build-base",
         "//src/main/java/com/google/devtools/build/lib:events",
         "//src/main/java/com/google/devtools/build/lib:packages-internal",
@@ -65,3 +83,39 @@ java_library(
         "@remoteapis//:build_bazel_semver_semver_java_proto",
     ],
 )
+
+java_library(
+    name = "ExecutionStatusException",
+    srcs = ["ExecutionStatusException.java"],
+    deps = [
+        "//third_party:jsr305",
+        "//third_party/grpc:grpc-jar",
+        "@googleapis//:google_rpc_status_java_proto",
+        "@remoteapis//:build_bazel_remote_execution_v2_remote_execution_java_proto",
+    ],
+)
+
+java_library(
+    name = "ReferenceCountedChannel",
+    srcs = ["ReferenceCountedChannel.java"],
+    deps = [
+        "//third_party:netty",
+        "//third_party/grpc:grpc-jar",
+    ],
+)
+
+java_library(
+    name = "Retrier",
+    srcs = [
+        "RemoteRetrier.java",
+        "RemoteRetrierUtils.java",
+        "Retrier.java",
+    ],
+    deps = [
+        ":ExecutionStatusException",
+        "//src/main/java/com/google/devtools/build/lib/remote/options",
+        "//third_party:guava",
+        "//third_party:jsr305",
+        "//third_party/grpc:grpc-jar",
+    ],
+)

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

@@ -21,13 +21,14 @@
 import io.netty.util.ReferenceCounted;
 import java.util.concurrent.TimeUnit;
 
-/** A wrapper around a {@link io.grpc.ManagedChannel} exposing a reference count.
- * When instantiated the reference count is 1. {@link ManagedChannel#shutdown()} will be called
- * on the wrapped channel when the reference count reaches 0.
+/**
+ * A wrapper around a {@link io.grpc.ManagedChannel} exposing a reference count. When instantiated
+ * the reference count is 1. {@link ManagedChannel#shutdown()} will be called on the wrapped channel
+ * when the reference count reaches 0.
  *
- * See {@link ReferenceCounted} for more information about reference counting.
+ * <p>See {@link ReferenceCounted} for more information about reference counting.
  */
-class ReferenceCountedChannel extends ManagedChannel implements ReferenceCounted {
+public class ReferenceCountedChannel extends ManagedChannel implements ReferenceCounted {
 
   private final ManagedChannel channel;
   private final AbstractReferenceCounted referenceCounted = new AbstractReferenceCounted() {

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

@@ -120,7 +120,8 @@ public <T> T execute(Callable<T> call) throws IOException, InterruptedException
     }
   }
 
-  static class ExponentialBackoff implements Backoff {
+  /** Backoff strategy that backs off exponentially. */
+  public static class ExponentialBackoff implements Backoff {
 
     private final long maxMillis;
     private long nextDelayMillis;
@@ -152,7 +153,7 @@ static class ExponentialBackoff implements Backoff {
       this.maxAttempts = maxAttempts;
     }
 
-    ExponentialBackoff(RemoteOptions options) {
+    public ExponentialBackoff(RemoteOptions options) {
       this(
           /* initial = */ Duration.ofMillis(100),
           /* max = */ Duration.ofSeconds(5),

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

@@ -0,0 +1,32 @@
+load("@rules_java//java:defs.bzl", "java_library")
+
+package(
+    default_visibility = ["//src:__subpackages__"],
+)
+
+filegroup(
+    name = "srcs",
+    srcs = glob(["*"]),
+)
+
+java_library(
+    name = "downloader",
+    srcs = glob(["*.java"]),
+    deps = [
+        "//src/main/java/com/google/devtools/build/lib:events",
+        "//src/main/java/com/google/devtools/build/lib/bazel/repository/downloader",
+        "//src/main/java/com/google/devtools/build/lib/remote",
+        "//src/main/java/com/google/devtools/build/lib/remote:ReferenceCountedChannel",
+        "//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/options",
+        "//src/main/java/com/google/devtools/build/lib/remote/util",
+        "//src/main/java/com/google/devtools/build/lib/vfs",
+        "//third_party:gson",
+        "//third_party:guava",
+        "//third_party/grpc:grpc-jar",
+        "@remoteapis//:build_bazel_remote_asset_v1_remote_asset_java_grpc",
+        "@remoteapis//:build_bazel_remote_asset_v1_remote_asset_java_proto",
+        "@remoteapis//:build_bazel_remote_execution_v2_remote_execution_java_proto",
+    ],
+)