Lowercaseify Nexus request headers (#1006)

Author: Sushisource

core/src/worker/nexus.rs

@@ -32,7 +32,11 @@ use temporal_sdk_core_protos::{
             CancelNexusTask, NexusTask, NexusTaskCancelReason, nexus_task, nexus_task_completion,
         },
     },
-    temporal::api::nexus::v1::{request::Variant, response, start_operation_response},
+    temporal::api::nexus::{
+        self,
+        v1::{request::Variant, response, start_operation_response},
+    },
+    utilities::normalize_http_headers,
 };
 use tokio::{
     join,
@@ -42,7 +46,7 @@ use tokio::{
 use tokio_stream::wrappers::UnboundedReceiverStream;
 use tokio_util::sync::CancellationToken;
 
-static REQUEST_TIMEOUT_HEADER: &str = "Request-Timeout";
+static REQUEST_TIMEOUT_HEADER: &str = "request-timeout";
 
 /// Centralizes all state related to received nexus tasks
 pub(super) struct NexusManager {
@@ -245,11 +249,18 @@ where
             .filter_map(move |t| {
                 let res = match t {
                     TaskStreamInput::Poll(t) => match *t {
-                        Ok(t) => {
+                        Ok(mut t) => {
                             if let Some(dur) = t.resp.sched_to_start() {
                                 self.metrics.nexus_task_sched_to_start_latency(dur);
                             };
 
+                            if let Some(ref mut req) = t.resp.request {
+                                req.header = normalize_http_headers(std::mem::take(&mut req.header));
+                                if let Some(nexus::v1::request::Variant::StartOperation(ref mut sor)) = req.variant {
+                                    sor.callback_header = normalize_http_headers(std::mem::take(&mut sor.callback_header));
+                                }
+                            }
+
                             let tt = TaskToken(t.resp.task_token.clone());
                             let mut timeout_task = None;
                             if let Some(timeout_str) = t

sdk-core-protos/src/utilities.rs

@@ -1,3 +1,5 @@
+use std::collections::HashMap;
+
 use prost::{EncodeError, Message};
 
 pub trait TryIntoOrNone<F, T> {
@@ -26,3 +28,13 @@ pub fn pack_any<T: Message>(
     Message::encode(msg, &mut value)?;
     Ok(prost_wkt_types::Any { type_url, value })
 }
+
+/// Given a header map, lowercase all the keys and return it as a new map.
+/// Any keys that are duplicated after lowercasing will clobber each other in undefined ordering.
+pub fn normalize_http_headers(headers: HashMap<String, String>) -> HashMap<String, String> {
+    let mut new_headers = HashMap::new();
+    for (header_key, val) in headers.into_iter() {
+        new_headers.insert(header_key.to_lowercase(), val);
+    }
+    new_headers
+}

tests/integ_tests/workflow_tests/nexus.rs

@@ -258,6 +258,16 @@ async fn nexus_async(
     let client = starter.get_client().await.get_client().clone();
     let nexus_task_handle = async {
         let mut nt = core_worker.poll_nexus_task().await.unwrap().unwrap_task();
+        // Verify request header key for timeout exists and is lowercase
+        if outcome == Outcome::Timeout {
+            assert!(
+                nt.request
+                    .as_ref()
+                    .unwrap()
+                    .header
+                    .contains_key("request-timeout")
+            );
+        }
         let start_req = assert_matches!(
             nt.request.unwrap().variant.unwrap(),
             request::Variant::StartOperation(sr) => sr