Revert remote execution from tower to grpcio (#7256)

We're seeing weird broken connection errors with tower.

We'll probably just chuck some retries in and be happy, but for now, let's get back to a more stable time.

* Revert "Remove unused operation wrapper (#7194)"

This reverts commit 9400024.

* Revert "Switch operation getting to tower (#7108)"

This reverts commit 0375b30.

* Revert "Remote execution uses tower-grpc to start executions (#7049)"

This reverts commit 28683c7.

src/rust/engine/process_execution/Cargo.toml

@@ -13,28 +13,18 @@ bytes = "0.4.5"
 digest = "0.8"
 fs = { path = "../fs" }
 futures = "^0.1.16"
-# TODO: Switch to a release once https://github.com/alexcrichton/futures-timer/pull/11 and https://github.com/alexcrichton/futures-timer/pull/12 merge
-futures-timer = { git = "https://github.com/pantsbuild/futures-timer", rev = "0b747e565309a58537807ab43c674d8951f9e5a0" }
-h2 = "0.1.13"
+grpcio = { git = "https://github.com/pantsbuild/grpc-rs.git", rev = "4dfafe9355dc996d7d0702e7386a6fedcd9734c0", default_features = false, features = ["protobuf-codec"] }
 hashing = { path = "../hashing" }
-http = "0.1"
 log = "0.4"
-parking_lot = "0.6"
-prost = "0.4"
-prost-types = "0.4"
 protobuf = { version = "2.0.4", features = ["with-bytes"] }
 resettable = { path = "../resettable" }
 sha2 = "0.8"
 tempfile = "3"
+# TODO: Switch to a release once https://github.com/alexcrichton/futures-timer/pull/11 and https://github.com/alexcrichton/futures-timer/pull/12 merge
+futures-timer = { git = "https://github.com/pantsbuild/futures-timer", rev = "0b747e565309a58537807ab43c674d8951f9e5a0" }
 time = "0.1.40"
-tokio = "0.1.14"
 tokio-codec = "0.1"
-tokio-connect = { git = "https://github.com/pantsbuild/tokio-connect.git", rev = "f7ad1ca437973d6e24037ac6f7d5ef1013833c0b" }
 tokio-process = "0.2.1"
-tower-grpc = { git = "https://github.com/pantsbuild/tower-grpc.git", rev = "ef19f2e1715f415ecb699e8f17f5845ad2b45daf" }
-tower-h2 = { git = "https://github.com/pantsbuild/tower-h2.git", rev = "44b0efb4983b769283efd5b2a3bc3decbf7c33de" }
-tower-http = { git = "https://github.com/pantsbuild/tower-http.git", rev = "56049ee7f31d4f6c549f5d1d5fbbfd7937df3d00" }
-tower-util = { git = "https://github.com/pantsbuild/tower.git", rev = "7b61c1fc1992c1df684fd3f179644ef0ca9bfa4c" }
 
 [dev-dependencies]
 mock = { path = "../testutil/mock" }

src/rust/engine/process_execution/bazel_protos/Cargo.toml

@@ -10,7 +10,6 @@ bytes = "0.4.5"
 futures = "^0.1.16"
 grpcio = { git = "https://github.com/pantsbuild/grpc-rs.git", rev = "4dfafe9355dc996d7d0702e7386a6fedcd9734c0", default_features = false, features = ["protobuf-codec"] }
 hashing = { path = "../../hashing" }
-log = "0.4"
 prost = "0.4"
 prost-derive = "0.4"
 prost-types = "0.4"

src/rust/engine/process_execution/bazel_protos/build.rs

@@ -176,11 +176,9 @@ fn generate_for_tower(thirdpartyprotobuf: &Path, out_dir: PathBuf) {
     .enable_server(true)
     .enable_client(true)
     .build(
-      &[
-        PathBuf::from("build/bazel/remote/execution/v2/remote_execution.proto"),
-        PathBuf::from("google/rpc/code.proto"),
-        PathBuf::from("google/rpc/error_details.proto"),
-      ],
+      &[PathBuf::from(
+        "build/bazel/remote/execution/v2/remote_execution.proto",
+      )],
       &std::fs::read_dir(&thirdpartyprotobuf)
         .unwrap()
         .into_iter()

src/rust/engine/process_execution/bazel_protos/src/conversions.rs

@@ -1,7 +1,4 @@
-use bytes::BytesMut;
 use hashing;
-use log::error;
-use prost::Message;
 
 impl<'a> From<&'a hashing::Digest> for crate::remote_execution::Digest {
   fn from(d: &hashing::Digest) -> Self {
@@ -21,31 +18,19 @@ impl<'a> From<&'a hashing::Digest> for crate::build::bazel::remote::execution::v
   }
 }
 
-impl<'a> From<&'a crate::remote_execution::Digest> for Result<hashing::Digest, String> {
-  fn from(d: &crate::remote_execution::Digest) -> Self {
+impl<'a> From<&'a super::remote_execution::Digest> for Result<hashing::Digest, String> {
+  fn from(d: &super::remote_execution::Digest) -> Self {
     hashing::Fingerprint::from_hex_string(d.get_hash())
       .map_err(|err| format!("Bad fingerprint in Digest {:?}: {:?}", d.get_hash(), err))
       .map(|fingerprint| hashing::Digest(fingerprint, d.get_size_bytes() as usize))
   }
 }
 
-impl<'a> From<&'a crate::build::bazel::remote::execution::v2::Digest>
-  for Result<hashing::Digest, String>
-{
-  fn from(d: &crate::build::bazel::remote::execution::v2::Digest) -> Self {
-    hashing::Fingerprint::from_hex_string(&d.hash)
-      .map_err(|err| format!("Bad fingerprint in Digest {:?}: {:?}", d.hash, err))
-      .map(|fingerprint| hashing::Digest(fingerprint, d.size_bytes as usize))
-  }
-}
-
 impl From<crate::google::longrunning::Operation> for crate::operations::Operation {
   fn from(op: crate::google::longrunning::Operation) -> Self {
     let mut dst = Self::new();
     dst.set_name(op.name);
-    if let Some(metadata) = op.metadata {
-      dst.set_metadata(prost_any_to_gcprio_any(metadata));
-    }
+    dst.set_metadata(prost_any_to_gcprio_any(op.metadata.unwrap()));
     dst.set_done(op.done);
     match op.result {
       Some(crate::google::longrunning::operation::Result::Response(response)) => {
@@ -60,87 +45,6 @@ impl From<crate::google::longrunning::Operation> for crate::operations::Operatio
   }
 }
 
-// This should only be used in test contexts. It should be deleted when the mock systems use tower.
-impl From<crate::remote_execution::ExecuteRequest>
-  for crate::build::bazel::remote::execution::v2::ExecuteRequest
-{
-  fn from(req: crate::remote_execution::ExecuteRequest) -> Self {
-    if req.has_execution_policy() || req.has_results_cache_policy() {
-      panic!("Can't convert ExecuteRequest protos with execution policy or results cache policy");
-    }
-    let digest: Result<hashing::Digest, String> = req.get_action_digest().into();
-    Self {
-      action_digest: Some((&digest.expect("Bad digest converting ExecuteRequest proto")).into()),
-      instance_name: req.instance_name,
-      execution_policy: None,
-      results_cache_policy: None,
-      skip_cache_lookup: req.skip_cache_lookup,
-    }
-  }
-}
-
-// This should only be used in test contexts. It should be deleted when the mock systems use tower.
-impl From<crate::build::bazel::remote::execution::v2::ExecuteRequest>
-  for crate::remote_execution::ExecuteRequest
-{
-  fn from(req: crate::build::bazel::remote::execution::v2::ExecuteRequest) -> Self {
-    if req.execution_policy.is_some() || req.results_cache_policy.is_some() {
-      panic!("Can't convert ExecuteRequest protos with execution policy or results cache policy");
-    }
-    let digest: Result<hashing::Digest, String> = (&req
-      .action_digest
-      .expect("Missing digest converting ExecuteRequest proto"))
-      .into();
-
-    let mut ret = Self::new();
-    ret.set_action_digest((&digest.expect("Bad digest converting ExecuteRequest proto")).into());
-    ret.set_instance_name(req.instance_name);
-    ret.set_skip_cache_lookup(req.skip_cache_lookup);
-    ret
-  }
-}
-
-// This should only be used in test contexts. It should be deleted when the mock systems use tower.
-impl Into<grpcio::RpcStatus> for crate::google::rpc::Status {
-  fn into(self) -> grpcio::RpcStatus {
-    let mut buf = BytesMut::with_capacity(self.encoded_len());
-    self.encode(&mut buf).unwrap();
-    grpcio::RpcStatus {
-      status: self.code.into(),
-      details: None,
-      status_proto_bytes: Some(buf.to_vec()),
-    }
-  }
-}
-
-// TODO: Use num_enum or similar here when TryInto is stable.
-pub fn code_from_i32(i: i32) -> crate::google::rpc::Code {
-  use crate::google::rpc::Code::*;
-  match i {
-    0 => Ok,
-    1 => Cancelled,
-    2 => Unknown,
-    3 => InvalidArgument,
-    4 => DeadlineExceeded,
-    5 => NotFound,
-    6 => AlreadyExists,
-    7 => PermissionDenied,
-    8 => ResourceExhausted,
-    9 => FailedPrecondition,
-    10 => Aborted,
-    11 => OutOfRange,
-    12 => Unimplemented,
-    13 => Internal,
-    14 => Unavailable,
-    15 => DataLoss,
-    16 => Unauthenticated,
-    _ => {
-      error!("Unknown grpc error code: {}, default to Unknown", i);
-      Unknown
-    }
-  }
-}
-
 pub fn prost_any_to_gcprio_any(any: prost_types::Any) -> protobuf::well_known_types::Any {
   let prost_types::Any { type_url, value } = any;
   let mut dst = protobuf::well_known_types::Any::new();

src/rust/engine/process_execution/bazel_protos/src/lib.rs

@@ -13,6 +13,5 @@ mod gen_for_tower;
 pub use crate::gen_for_tower::*;
 
 mod conversions;
-pub use crate::conversions::code_from_i32;
 mod verification;
 pub use crate::verification::verify_directory_canonical;