feat: allow binary gRPC headers

Author: chris-olszewski

packages/core-bridge/src/client.rs

@@ -3,7 +3,7 @@ use std::time::Duration;
 use std::{collections::HashMap, sync::Arc};
 
 use neon::prelude::*;
-use tonic::metadata::MetadataKey;
+use tonic::metadata::{BinaryMetadataValue, MetadataKey};
 
 use temporal_sdk_core::{ClientOptions as CoreClientOptions, CoreRuntime, RetryClient};
 
@@ -130,10 +130,16 @@ pub struct RpcCall {
     pub rpc: String,
     pub req: Vec<u8>,
     pub retry: bool,
-    pub metadata: HashMap<String, String>,
+    pub metadata: HashMap<String, MetadataValue>,
     pub timeout: Option<Duration>,
 }
 
+#[derive(Debug, Clone, TryFromJs)]
+pub enum MetadataValue {
+    Ascii { value: String },
+    Binary { value: Vec<u8> },
+}
+
 /// Send a request to the Workflow Service using the provided Client
 #[js_function]
 pub fn client_send_workflow_service_request(
@@ -584,16 +590,29 @@ fn rpc_req<P: prost::Message + Default>(call: RpcCall) -> BridgeResult<tonic::Re
 
     let mut req = tonic::Request::new(proto);
     for (k, v) in call.metadata {
-        req.metadata_mut().insert(
-            MetadataKey::from_str(k.as_str()).map_err(|err| BridgeError::TypeError {
-                field: None,
-                message: format!("Invalid metadata key: {err}"),
-            })?,
-            v.parse().map_err(|err| BridgeError::TypeError {
-                field: None,
-                message: format!("Invalid metadata value: {err}"),
-            })?,
-        );
+        match v {
+            MetadataValue::Ascii { value: v } => {
+                req.metadata_mut().insert(
+                    MetadataKey::from_str(k.as_str()).map_err(|err| BridgeError::TypeError {
+                        field: None,
+                        message: format!("Invalid metadata key: {err}"),
+                    })?,
+                    v.parse().map_err(|err| BridgeError::TypeError {
+                        field: None,
+                        message: format!("Invalid metadata value: {err}"),
+                    })?,
+                );
+            }
+            MetadataValue::Binary { value: v } => {
+                req.metadata_mut().insert_bin(
+                    MetadataKey::from_str(k.as_str()).map_err(|err| BridgeError::TypeError {
+                        field: None,
+                        message: format!("Invalid metadata key: {err}"),
+                    })?,
+                    BinaryMetadataValue::from_bytes(&v),
+                );
+            }
+        }
     }
 
     if let Some(timeout) = call.timeout {
@@ -628,7 +647,7 @@ mod config {
 
     use bridge_macros::TryFromJs;
 
-    use crate::helpers::*;
+    use crate::{client::MetadataValue, helpers::*};
 
     #[derive(Debug, Clone, TryFromJs)]
     pub(super) struct ClientOptions {
@@ -637,7 +656,7 @@ mod config {
         client_version: String,
         tls: Option<TlsConfig>,
         http_connect_proxy: Option<HttpConnectProxy>,
-        headers: Option<HashMap<String, String>>,
+        headers: Option<HashMap<String, MetadataValue>>,
         api_key: Option<String>,
         disable_error_code_metric_tags: bool,
     }
@@ -677,13 +696,16 @@ mod config {
                 builder.tls_cfg(tls.into());
             }
 
+            let (ascii_headers, bin_headers) = partition_headers(self.headers);
+
             let client_options = builder
                 .target_url(self.target_url)
                 .client_name(self.client_name)
                 .client_version(self.client_version)
                 // tls_cfg -- above
                 .http_connect_proxy(self.http_connect_proxy.map(Into::into))
-                .headers(self.headers)
+                .headers(ascii_headers)
+                .binary_headers(bin_headers)
                 .api_key(self.api_key)
                 .disable_error_code_metric_tags(self.disable_error_code_metric_tags)
                 // identity -- skipped: will be set on worker
@@ -719,4 +741,32 @@ mod config {
             }
         }
     }
+
+    fn partition_headers(
+        headers: Option<HashMap<String, MetadataValue>>,
+    ) -> (
+        Option<HashMap<String, String>>,
+        Option<HashMap<String, Vec<u8>>>,
+    ) {
+        let Some(headers) = headers else {
+            return (None, None);
+        };
+        // Maybe with_capacity this assuming primarily ascii headers?
+        let mut ascii_headers = HashMap::default();
+        let mut bin_headers = HashMap::default();
+        for (k, v) in headers {
+            match v {
+                MetadataValue::Ascii { value: v } => {
+                    ascii_headers.insert(k, v);
+                }
+                MetadataValue::Binary { value: v } => {
+                    bin_headers.insert(k, v);
+                }
+            }
+        }
+        (
+            (!ascii_headers.is_empty()).then_some(ascii_headers),
+            (!bin_headers.is_empty()).then_some(bin_headers),
+        )
+    }
 }

packages/core-bridge/ts/native.ts

@@ -124,7 +124,7 @@ export interface ClientOptions {
   clientVersion: string;
   tls: Option<TLSConfig>;
   httpConnectProxy: Option<HttpConnectProxy>;
-  headers: Option<Record<string, string>>;
+  headers: Option<Record<string, MetadataValue>>;
   apiKey: Option<string>;
   disableErrorCodeMetricTags: boolean;
 }
@@ -157,7 +157,7 @@ export interface RpcCall {
   rpc: string;
   req: Buffer;
   retry: boolean;
-  metadata: Record<string, string>;
+  metadata: Record<string, MetadataValue>;
   timeout: Option<number>;
 }
 
@@ -191,6 +191,16 @@ export interface Worker {
   type: 'worker';
 }
 
+export type MetadataValue =
+  | {
+      type: 'ascii';
+      value: string;
+    }
+  | {
+      type: 'binary';
+      value: Buffer;
+    };
+
 export interface WorkerOptions {
   identity: string;
   buildId: string;

packages/test/src/test-native-connection.ts

@@ -203,13 +203,18 @@ test('withMetadata and withDeadline propagate metadata and deadline', async (t)
   );
   const connection = await NativeConnection.connect({
     address: `127.0.0.1:${port}`,
+    metadata: { 'default-bin': Buffer.from([0x00]) },
   });
 
   await connection.withDeadline(Date.now() + 10_000, () =>
-    connection.withMetadata({ test: 'true' }, () => connection.workflowService.getSystemInfo({}))
+    connection.withMetadata({ test: 'true', 'other-bin': Buffer.from([0x01]) }, () =>
+      connection.workflowService.getSystemInfo({})
+    )
   );
   t.is(requests.length, 2);
   t.is(requests[1].metadata.get('test').toString(), 'true');
+  t.deepEqual(requests[1].metadata.get('default-bin'), [Buffer.from([0x00])]);
+  t.deepEqual(requests[1].metadata.get('other-bin'), [Buffer.from([0x01])]);
   t.true(typeof requests[1].deadline === 'number' && requests[1].deadline > 5_000);
   await connection.close();
   server.forceShutdown();

packages/worker/src/connection-options.ts

@@ -8,6 +8,7 @@ import {
   TLSConfig,
 } from '@temporalio/common/lib/internal-non-workflow';
 import pkg from './pkg';
+import type { Metadata } from '@temporalio/client';
 
 export { TLSConfig, ProxyConfig };
 
@@ -44,7 +45,7 @@ export interface NativeConnectionOptions {
    *
    * Set statically at connection time, can be replaced later using {@link NativeConnection.setMetadata}.
    */
-  metadata?: Record<string, string>;
+  metadata?: Metadata;
 
   /**
    * API key for Temporal. This becomes the "Authorization" HTTP header with "Bearer " prepended.
@@ -102,13 +103,25 @@ export function toNativeClientOptions(options: NativeConnectionOptions): native.
     );
   }
 
+  let headers: Record<string, native.MetadataValue> | null = null;
+  if (options.metadata) {
+    headers = {};
+    for (const [key, value] of Object.entries(options.metadata)) {
+      if (typeof value === 'string') {
+        headers[key] = { type: 'ascii', value };
+      } else {
+        headers[key] = { type: 'binary', value };
+      }
+    }
+  }
+
   return {
     targetUrl: tls ? `https://${address}` : `http://${address}`,
     clientName: 'temporal-typescript',
     clientVersion: pkg.version,
     tls,
     httpConnectProxy,
-    headers: options.metadata ?? null,
+    headers,
     apiKey: options.apiKey ?? null,
     disableErrorCodeMetricTags: options.disableErrorCodeMetricTags ?? false,
   };

packages/worker/src/connection.ts

@@ -127,7 +127,14 @@ export class NativeConnection implements ConnectionLike {
 
     const ctx = this.callContextStorage.getStore() ?? {};
     const metadata =
-      ctx.metadata != null ? Object.fromEntries(Object.entries(ctx.metadata).map(([k, v]) => [k, v.toString()])) : {};
+      ctx.metadata != null
+        ? Object.fromEntries(
+            Object.entries(ctx.metadata).map(([k, value]) => [
+              k,
+              typeof value === 'string' ? { type: 'ascii' as const, value } : { type: 'binary' as const, value },
+            ])
+          )
+        : {};
 
     const req = {
       rpc: method.name,