Add tests verifying grpc-timeout works properly

client/src/lib.rs

@@ -1622,6 +1622,7 @@ impl<T> WfClientExt for T where T: WfHandleClient + Clone + Sized {}
 #[cfg(test)]
 mod tests {
     use super::*;
+    use tonic::metadata::Ascii;
 
     #[test]
     fn applies_headers() {
@@ -1682,6 +1683,16 @@ mod tests {
         let req = interceptor.call(tonic::Request::new(())).unwrap();
         assert!(!req.metadata().contains_key("my-meta-key"));
         assert!(!req.metadata().contains_key("authorization"));
+
+        // Timeout header not overriden
+        let mut req = tonic::Request::new(());
+        req.metadata_mut()
+            .insert("grpc-timeout", "1S".parse().unwrap());
+        let req = interceptor.call(req).unwrap();
+        assert_eq!(
+            req.metadata().get("grpc-timeout").unwrap(),
+            "1S".parse::<MetadataValue<Ascii>>().unwrap()
+        );
     }
 
     #[test]

tests/integ_tests/client_tests.rs

@@ -1,7 +1,15 @@
-use std::time::Duration;
-use temporal_client::{RetryClient, WorkflowClientTrait, WorkflowService};
-use temporal_sdk_core_protos::temporal::api::workflowservice::v1::DescribeNamespaceRequest;
+use assert_matches::assert_matches;
+use std::{collections::HashMap, time::Duration};
+use temporal_client::{Client, WorkflowClientTrait, WorkflowOptions, WorkflowService};
+use temporal_sdk_core_protos::temporal::api::{
+    common::v1::WorkflowExecution,
+    enums::v1::{UpdateWorkflowExecutionLifecycleStage, WorkflowIdReusePolicy},
+    update,
+    update::v1::WaitPolicy,
+    workflowservice::v1::{DescribeNamespaceRequest, UpdateWorkflowExecutionRequest},
+};
 use temporal_sdk_core_test_utils::{get_integ_server_options, CoreWfStarter, NAMESPACE};
+use tonic::{Code, Request};
 
 #[tokio::test]
 async fn can_use_retry_client() {
@@ -17,9 +25,8 @@ async fn can_use_retry_client() {
 #[tokio::test]
 async fn can_use_retry_raw_client() {
     let opts = get_integ_server_options();
-    let raw_client = opts.connect_no_namespace(None).await.unwrap();
-    let mut retry_client = RetryClient::new(raw_client, opts.retry_config);
-    retry_client
+    let mut client = opts.connect_no_namespace(None).await.unwrap();
+    client
         .describe_namespace(DescribeNamespaceRequest {
             namespace: NAMESPACE.to_string(),
             ..Default::default()
@@ -34,3 +41,74 @@ async fn calls_get_system_info() {
     let raw_client = opts.connect_no_namespace(None).await.unwrap();
     assert!(raw_client.get_client().capabilities().is_some());
 }
+
+#[tokio::test]
+async fn per_call_timeout_respected_whole_client() {
+    let opts = get_integ_server_options();
+    let mut raw_client = opts.connect_no_namespace(None).await.unwrap();
+    let mut hm = HashMap::new();
+    hm.insert("grpc-timeout".to_string(), "0S".to_string());
+    raw_client.get_client().set_headers(hm);
+    let err = raw_client
+        .describe_namespace(DescribeNamespaceRequest {
+            namespace: NAMESPACE.to_string(),
+            ..Default::default()
+        })
+        .await
+        .unwrap_err();
+    assert_eq!(err.code(), Code::DeadlineExceeded);
+}
+
+#[tokio::test]
+async fn per_call_timeout_respected_one_call() {
+    let opts = get_integ_server_options();
+    let mut client = opts.connect_no_namespace(None).await.unwrap();
+    // Start a workflow (we don't need to actually make any progress on it)
+    let wfc = Client::new(client.clone().into_inner(), NAMESPACE.to_string());
+    wfc.start_workflow(
+        vec![],
+        "whatever".to_string(),
+        "test-rpc-timeout".to_string(),
+        "whatever".to_string(),
+        None,
+        WorkflowOptions {
+            run_timeout: Some(Duration::from_secs(10)),
+            id_reuse_policy: WorkflowIdReusePolicy::TerminateIfRunning,
+            ..Default::default()
+        },
+    )
+    .await
+    .unwrap();
+
+    let mut req = Request::new(UpdateWorkflowExecutionRequest {
+        namespace: NAMESPACE.to_string(),
+        wait_policy: Some(WaitPolicy {
+            lifecycle_stage: UpdateWorkflowExecutionLifecycleStage::Completed.into(),
+        }),
+        workflow_execution: Some(WorkflowExecution {
+            workflow_id: "test-rpc-timeout".to_string(),
+            run_id: "".to_string(),
+        }),
+        request: Some(update::v1::Request {
+            meta: Some(update::v1::Meta {
+                update_id: "".into(),
+                identity: "aaaa".to_string(),
+            }),
+            input: Some(update::v1::Input {
+                header: None,
+                name: "update".to_string(),
+                args: None,
+            }),
+        }),
+        ..Default::default()
+    });
+    req.set_timeout(Duration::from_millis(500));
+    let start = std::time::Instant::now();
+    let res = client.update_workflow_execution(req).await;
+    dbg!(&res);
+    assert_matches!(
+        res.unwrap_err().code(),
+        Code::DeadlineExceeded | Code::Cancelled
+    );
+    assert!(start.elapsed() < Duration::from_secs(1));
+}