feat(transport): Support timeouts with "grpc-timeout" header

tests/integration_tests/Cargo.toml

@@ -16,6 +16,7 @@ bytes = "1.0"
 
 [dev-dependencies]
 tokio = { version = "1.0", features = ["macros", "rt-multi-thread", "net"] }
+tokio-stream = { version = "0.1.5", features = ["net"] }
 
 [build-dependencies]
 tonic-build = { path = "../../tonic-build" }

tests/integration_tests/tests/timeout.rs

@@ -0,0 +1,94 @@
+use integration_tests::pb::{test_client, test_server, Input, Output};
+use std::time::Duration;
+use tokio::net::TcpListener;
+use tonic::{transport::Server, Code, Request, Response, Status};
+
+#[tokio::test]
+#[ignore]
+async fn cancelation_on_timeout() {
+    struct Svc;
+
+    #[tonic::async_trait]
+    impl test_server::Test for Svc {
+        async fn unary_call(&self, _req: Request<Input>) -> Result<Response<Output>, Status> {
+            // Wait for a time longer than the timeout
+            tokio::time::sleep(Duration::from_millis(1_000)).await;
+            Ok(Response::new(Output {}))
+        }
+    }
+
+    let svc = test_server::TestServer::new(Svc);
+
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+
+    tokio::spawn(async move {
+        Server::builder()
+            .add_service(svc)
+            .serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener))
+            .await
+            .unwrap();
+    });
+
+    let mut client = test_client::TestClient::connect(format!("http://{}", addr))
+        .await
+        .unwrap();
+
+    let mut req = Request::new(Input {});
+    req.metadata_mut()
+        .insert("grpc-timeout", "500m".parse().unwrap());
+
+    let res = client.unary_call(req).await;
+    dbg!(&res);
+
+    let err = res.unwrap_err();
+    assert!(err.message().contains("Timeout expired"));
+
+    // TODO(david): make this work. Will require mapping `TimeoutExpired` errors into
+    // `Code::Cancelled` but can't quite figure out how to do that.
+    assert_eq!(err.code(), Code::Cancelled);
+}
+
+#[tokio::test]
+async fn picks_the_shortest_timeout() {
+    struct Svc;
+
+    #[tonic::async_trait]
+    impl test_server::Test for Svc {
+        async fn unary_call(&self, _req: Request<Input>) -> Result<Response<Output>, Status> {
+            // Wait for a time longer than the timeout
+            tokio::time::sleep(Duration::from_secs(1)).await;
+            Ok(Response::new(Output {}))
+        }
+    }
+
+    let svc = test_server::TestServer::new(Svc);
+
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+
+    tokio::spawn(async move {
+        Server::builder()
+            .timeout(Duration::from_millis(100))
+            .add_service(svc)
+            .serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener))
+            .await
+            .unwrap();
+    });
+
+    let mut client = test_client::TestClient::connect(format!("http://{}", addr))
+        .await
+        .unwrap();
+
+    let mut req = Request::new(Input {});
+    req.metadata_mut()
+        // 10 hours
+        .insert("grpc-timeout", "10H".parse().unwrap());
+
+    // TODO(david): for some reason this fails with "h2 protocol error: protocol error: unexpected
+    // internal error encountered". Seems to be happening on `master` as well. Bug?
+    let res = client.unary_call(req).await;
+    dbg!(&res);
+    let err = res.unwrap_err();
+    assert!(err.message().contains("Timeout expired"));
+}

tonic/Cargo.toml

@@ -31,7 +31,8 @@ transport = [
     "tokio",
     "tower",
     "tracing-futures",
-    "tokio/macros"
+    "tokio/macros",
+    "tokio/time",
 ]
 tls = ["transport", "tokio-rustls"]
 tls-roots = ["tls", "rustls-native-certs"]

tonic/src/metadata/map.rs

@@ -198,11 +198,10 @@ pub struct OccupiedEntry<'a, VE: ValueEncoding> {
 
 impl MetadataMap {
     // Headers reserved by the gRPC protocol.
-    pub(crate) const GRPC_RESERVED_HEADERS: [&'static str; 8] = [
+    pub(crate) const GRPC_RESERVED_HEADERS: [&'static str; 7] = [
         "te",
         "user-agent",
         "content-type",
-        "grpc-timeout",
         "grpc-message",
         "grpc-encoding",
         "grpc-message-type",

tonic/src/transport/service/connection.rs

@@ -1,5 +1,7 @@
 use super::super::BoxFuture;
-use super::{layer::ServiceBuilderExt, reconnect::Reconnect, AddOrigin, UserAgent};
+use super::{
+    layer::ServiceBuilderExt, reconnect::Reconnect, timeout::Timeout, AddOrigin, UserAgent,
+};
 use crate::{body::BoxBody, transport::Endpoint};
 use http::Uri;
 use hyper::client::conn::Builder;
@@ -14,7 +16,6 @@ use tower::load::Load;
 use tower::{
     layer::Layer,
     limit::{concurrency::ConcurrencyLimitLayer, rate::RateLimitLayer},
-    timeout::TimeoutLayer,
     util::BoxService,
     ServiceBuilder, ServiceExt,
 };
@@ -53,7 +54,7 @@ impl Connection {
         let stack = ServiceBuilder::new()
             .layer_fn(|s| AddOrigin::new(s, endpoint.uri.clone()))
             .layer_fn(|s| UserAgent::new(s, endpoint.user_agent.clone()))
-            .optional_layer(endpoint.timeout.map(TimeoutLayer::new))
+            .layer_fn(|s| Timeout::new(s, endpoint.timeout))
             .optional_layer(endpoint.concurrency_limit.map(ConcurrencyLimitLayer::new))
             .optional_layer(endpoint.rate_limit.map(|(l, d)| RateLimitLayer::new(l, d)))
             .into_inner();

tonic/src/transport/service/mod.rs

@@ -6,6 +6,7 @@ mod io;
 mod layer;
 mod reconnect;
 mod router;
+mod timeout;
 #[cfg(feature = "tls")]
 mod tls;
 mod user_agent;