Add support for gRPC binary metadata values

packages/core-bridge/src/client.rs

@@ -3,12 +3,12 @@ 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};
 
 use bridge_macros::{TryFromJs, js_function};
-use temporal_client::{ClientInitError, ConfiguredClient, TemporalServiceClientWithMetrics};
+use temporal_client::{ClientInitError, ConfiguredClient, TemporalServiceClient};
 
 use crate::runtime::Runtime;
 use crate::{helpers::*, runtime::RuntimeExt as _};
@@ -38,7 +38,7 @@ pub fn init(cx: &mut neon::prelude::ModuleContext) -> neon::prelude::NeonResult<
     Ok(())
 }
 
-type CoreClient = RetryClient<ConfiguredClient<TemporalServiceClientWithMetrics>>;
+type CoreClient = RetryClient<ConfiguredClient<TemporalServiceClient>>;
 
 pub struct Client {
     // These fields are pub because they are accessed from Worker::new
@@ -61,6 +61,10 @@ pub fn client_new(
 
         let core_client = match res {
             Ok(core_client) => core_client,
+            Err(ClientInitError::InvalidHeaders(e)) => Err(BridgeError::TypeError {
+                message: format!("Invalid metadata key: {e}"),
+                field: None,
+            })?,
             Err(ClientInitError::SystemInfoCallError(e)) => Err(BridgeError::TransportError(
                 format!("Failed to call GetSystemInfo: {e}"),
             ))?,
@@ -90,7 +94,11 @@ pub fn client_update_headers(
         .borrow()?
         .core_client
         .get_client()
-        .set_headers(headers);
+        .set_headers(headers)
+        .map_err(|err| BridgeError::TypeError {
+            message: format!("Invalid metadata key: {err}"),
+            field: None,
+        })?;
     Ok(())
 }
 
@@ -122,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(
@@ -576,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 {
@@ -620,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 {
@@ -629,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,
     }
@@ -669,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
@@ -711,4 +741,31 @@ 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);
+        };
+        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/src/helpers/try_from_js.rs

@@ -11,7 +11,7 @@ use neon::{
 };
 use temporal_sdk_core::Url;
 
-use super::{BridgeError, BridgeResult};
+use super::{AppendFieldContext, BridgeError, BridgeResult};
 
 /// Trait for Rust types that can be created from JavaScript values, possibly throwing an error.
 pub trait TryFromJs: Sized {
@@ -175,8 +175,9 @@ impl<T: TryFromJs> TryFromJs for HashMap<String, T> {
         let mut map = Self::new();
         for key_handle in props {
             let key = key_handle.to_string(cx)?.value(cx);
-            let value = obj.get_value(cx, key_handle)?;
-            map.insert(key, T::try_from_js(cx, value)?);
+            let js_value = obj.get_value(cx, key_handle)?;
+            let value = T::try_from_js(cx, js_value).field(&key)?;
+            map.insert(key, value);
         }
         Ok(map)
     }

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-client-connection.ts

@@ -60,6 +60,8 @@ async function bindLocalhostTls(server: grpc.Server): Promise<number> {
 
 test('withMetadata / withDeadline / withAbortSignal set the CallContext for RPC call', async (t) => {
   let gotTestHeaders = false;
+  let gotStaticBinValue;
+  let gotOtherBinValue;
   let gotDeadline = false;
   const authTokens: string[] = [];
   const deadline = Date.now() + 10000;
@@ -89,6 +91,8 @@ test('withMetadata / withDeadline / withAbortSignal set the CallContext for RPC
       ) {
         gotTestHeaders = true;
       }
+      gotStaticBinValue = call.metadata.get('staticKey-bin');
+      gotOtherBinValue = call.metadata.get('otherKey-bin');
       const receivedDeadline = call.getDeadline();
       // For some reason the deadline the server gets is slightly different from the one we send in the client
       if (typeof receivedDeadline === 'number' && receivedDeadline >= deadline && receivedDeadline - deadline < 1000) {
@@ -108,16 +112,18 @@ test('withMetadata / withDeadline / withAbortSignal set the CallContext for RPC
   const port = await bindLocalhost(server);
   const conn = await Connection.connect({
     address: `127.0.0.1:${port}`,
-    metadata: { staticKey: 'set' },
+    metadata: { staticKey: 'set', 'staticKey-bin': Buffer.from([0x00]) },
     apiKey: 'test-token',
   });
   await conn.withMetadata({ test: 'true' }, () =>
-    conn.withMetadata({ otherKey: 'set' }, () =>
+    conn.withMetadata({ otherKey: 'set', 'otherKey-bin': Buffer.from([0x01]) }, () =>
       conn.withDeadline(deadline, () => conn.workflowService.registerNamespace({}))
     )
   );
   t.true(gotTestHeaders);
   t.true(gotDeadline);
+  t.deepEqual(gotStaticBinValue, [Buffer.from([0x00])]);
+  t.deepEqual(gotOtherBinValue, [Buffer.from([0x01])]);
   await conn.withApiKey('tt-2', () => conn.workflowService.startWorkflowExecution({}));
   conn.setApiKey('tt-3');
   await conn.workflowService.startWorkflowExecution({});

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

@@ -7,6 +7,7 @@ import {
   ProxyConfig,
   TLSConfig,
 } from '@temporalio/common/lib/internal-non-workflow';
+import type { Metadata } from '@temporalio/client';
 import pkg from './pkg';
 
 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,