feat!: Custom data converter

docs/data-converter.md

@@ -23,7 +23,7 @@ function workflowInclusiveInstanceOf(instance: unknown, type: Function): boolean
 
 ## Decision
 
-Given the possibility of switching or adding other isolation methods in future, we opted to convert to/from Payloads inside the vm (`PayloadConverter`). We also added another transformer layer called `PayloadCodec` that runs outside the vm, can use node modules and Promises, and operates on Payloads. A `DataConverter` is a `PayloadConverter` and a `PayloadCodec`:
+Given the possibility of switching or adding other isolation methods in future, we opted to convert to/from Payloads inside the vm (`PayloadConverter`). We also added another transformer layer called `PayloadCodec` that runs outside the vm, can use node APIs and Promises, and operates on Payloads. A `DataConverter` is a `PayloadConverter` and a `PayloadCodec`:
 
 ```ts
 export interface DataConverter {
@@ -32,7 +32,7 @@ export interface DataConverter {
 }
 
 export interface PayloadConverter {
-  toPayload<T>(value: T): Payload;
+  toPayload<T>(value: T): Payload | undefined;
   fromPayload<T>(payload: Payload): T;
 }
 
@@ -46,29 +46,14 @@ export interface PayloadCodec {
 
 `PayloadCodec` only runs in the main thread.
 
-When `WorkerOptions.dataConverter.payloadConverterPath` is provided, the code at that location is loaded into the main thread, the worker threads, and the webpack Workflow bundle.
+When `WorkerOptions.dataConverter.payloadConverterPath` is provided, the code at that location is loaded into the main thread and the webpack Workflow bundle.
 
 `Worker.create`:
 _main thread_
 
 - imports and validates `options.dataConverter.payloadConverterPath`
 - passes `payloadConverterPath` to `WorkflowCodeBundler`
 
-Execution goes to either:
-
-- `ThreadedVMWorkflowCreator.create`
-  _main thread_
-
-- `VMWorkflowCreator.create`
-  _worker thread (unless in debug mode)_
-
-And then to:
-
-- `VMWorkflowCreator.createWorkflow`
-  _worker thread (unless in debug mode)_
-
-And then to:
-
 `worker-interface.ts#initRuntime`:
 _workflow vm_
 

packages/client/src/workflow-client.ts

@@ -166,6 +166,9 @@ export interface WorkflowClientOptions {
 }
 
 export type WorkflowClientOptionsWithDefaults = Required<WorkflowClientOptions>;
+export type LoadedWorkflowClientOptions = WorkflowClientOptionsWithDefaults & {
+  loadedDataConverter: LoadedDataConverter;
+};
 
 export function defaultWorkflowClientOptions(): WorkflowClientOptionsWithDefaults {
   return {
@@ -243,12 +246,14 @@ export type WorkflowStartOptions<T extends Workflow = Workflow> = WithWorkflowAr
  * Client for starting Workflow executions and creating Workflow handles
  */
 export class WorkflowClient {
-  public readonly options: WorkflowClientOptionsWithDefaults;
-  protected readonly dataConverter: LoadedDataConverter;
+  public readonly options: LoadedWorkflowClientOptions;
 
   constructor(public readonly service: WorkflowService = new Connection().service, options?: WorkflowClientOptions) {
-    this.dataConverter = loadDataConverter(options?.dataConverter);
-    this.options = { ...defaultWorkflowClientOptions(), ...options };
+    this.options = {
+      ...defaultWorkflowClientOptions(),
+      ...options,
+      loadedDataConverter: loadDataConverter(options?.dataConverter),
+    };
   }
 
   /**
@@ -429,7 +434,7 @@ export class WorkflowClient {
         // Note that we can only return one value from our workflow function in JS.
         // Ignore any other payloads in result
         const [result] = await decodeArrayFromPayloads(
-          this.dataConverter,
+          this.options.loadedDataConverter,
           ev.workflowExecutionCompletedEventAttributes.result?.payloads
         );
         return result as any;
@@ -442,14 +447,14 @@ export class WorkflowClient {
         const { failure, retryState } = ev.workflowExecutionFailedEventAttributes;
         throw new WorkflowFailedError(
           'Workflow execution failed',
-          await decodeOptionalFailureToOptionalError(this.dataConverter, failure),
+          await decodeOptionalFailureToOptionalError(this.options.loadedDataConverter, failure),
           retryState ?? RetryState.RETRY_STATE_UNSPECIFIED
         );
       } else if (ev.workflowExecutionCanceledEventAttributes) {
         const failure = new CancelledFailure(
           'Workflow canceled',
           await decodeArrayFromPayloads(
-            this.dataConverter,
+            this.options.loadedDataConverter,
             ev.workflowExecutionCanceledEventAttributes.details?.payloads
           )
         );
@@ -529,7 +534,7 @@ export class WorkflowClient {
         execution: input.workflowExecution,
         query: {
           queryType: input.queryType,
-          queryArgs: { payloads: await encodeToPayloads(this.dataConverter, ...input.args) },
+          queryArgs: { payloads: await encodeToPayloads(this.options.loadedDataConverter, ...input.args) },
         },
       });
     } catch (err) {
@@ -545,7 +550,7 @@ export class WorkflowClient {
       throw new TypeError('Invalid response from server');
     }
     // We ignore anything but the first result
-    return await decodeFromPayloadsAtIndex(this.dataConverter, 0, response.queryResult?.payloads);
+    return await decodeFromPayloadsAtIndex(this.options.loadedDataConverter, 0, response.queryResult?.payloads);
   }
 
   /**
@@ -562,7 +567,7 @@ export class WorkflowClient {
         requestId: uuid4(),
         // control is unused,
         signalName: input.signalName,
-        input: { payloads: await encodeToPayloads(this.dataConverter, ...input.args) },
+        input: { payloads: await encodeToPayloads(this.options.loadedDataConverter, ...input.args) },
       });
     } catch (err) {
       this.rethrowGrpcError(err, input.workflowExecution, 'Failed to signal Workflow');
@@ -585,9 +590,9 @@ export class WorkflowClient {
         workflowId: options.workflowId,
         workflowIdReusePolicy: options.workflowIdReusePolicy,
         workflowType: { name: workflowType },
-        input: { payloads: await encodeToPayloads(this.dataConverter, ...options.args) },
+        input: { payloads: await encodeToPayloads(this.options.loadedDataConverter, ...options.args) },
         signalName,
-        signalInput: { payloads: await encodeToPayloads(this.dataConverter, ...signalArgs) },
+        signalInput: { payloads: await encodeToPayloads(this.options.loadedDataConverter, ...signalArgs) },
         taskQueue: {
           kind: temporal.api.enums.v1.TaskQueueKind.TASK_QUEUE_KIND_UNSPECIFIED,
           name: options.taskQueue,
@@ -596,10 +601,12 @@ export class WorkflowClient {
         workflowRunTimeout: options.workflowRunTimeout,
         workflowTaskTimeout: options.workflowTaskTimeout,
         retryPolicy: options.retry ? compileRetryPolicy(options.retry) : undefined,
-        memo: options.memo ? { fields: await encodeMapToPayloads(this.dataConverter, options.memo) } : undefined,
+        memo: options.memo
+          ? { fields: await encodeMapToPayloads(this.options.loadedDataConverter, options.memo) }
+          : undefined,
         searchAttributes: options.searchAttributes
           ? {
-              indexedFields: await encodeMapToPayloads(this.dataConverter, options.searchAttributes),
+              indexedFields: await encodeMapToPayloads(this.options.loadedDataConverter, options.searchAttributes),
             }
           : undefined,
         cronSchedule: options.cronSchedule,
@@ -626,7 +633,7 @@ export class WorkflowClient {
       workflowId: opts.workflowId,
       workflowIdReusePolicy: opts.workflowIdReusePolicy,
       workflowType: { name: workflowType },
-      input: { payloads: await encodeToPayloads(this.dataConverter, ...opts.args) },
+      input: { payloads: await encodeToPayloads(this.options.loadedDataConverter, ...opts.args) },
       taskQueue: {
         kind: temporal.api.enums.v1.TaskQueueKind.TASK_QUEUE_KIND_UNSPECIFIED,
         name: opts.taskQueue,
@@ -635,10 +642,10 @@ export class WorkflowClient {
       workflowRunTimeout: opts.workflowRunTimeout,
       workflowTaskTimeout: opts.workflowTaskTimeout,
       retryPolicy: opts.retry ? compileRetryPolicy(opts.retry) : undefined,
-      memo: opts.memo ? { fields: await encodeMapToPayloads(this.dataConverter, opts.memo) } : undefined,
+      memo: opts.memo ? { fields: await encodeMapToPayloads(this.options.loadedDataConverter, opts.memo) } : undefined,
       searchAttributes: opts.searchAttributes
         ? {
-            indexedFields: await encodeMapToPayloads(this.dataConverter, opts.searchAttributes),
+            indexedFields: await encodeMapToPayloads(this.options.loadedDataConverter, opts.searchAttributes),
           }
         : undefined,
       cronSchedule: opts.cronSchedule,
@@ -672,7 +679,11 @@ export class WorkflowClient {
         namespace: this.options.namespace,
         identity: this.options.identity,
         ...input,
-        details: { payloads: input.details ? await encodeToPayloads(this.dataConverter, ...input.details) : undefined },
+        details: {
+          payloads: input.details
+            ? await encodeToPayloads(this.options.loadedDataConverter, ...input.details)
+            : undefined,
+        },
         firstExecutionRunId: input.firstExecutionRunId,
       });
     } catch (err) {

packages/common/src/converter/data-converter.ts

@@ -2,18 +2,25 @@ import { PayloadCodec } from './payload-codec';
 import { PayloadConverter } from './payload-converter';
 
 /**
- * When your data (arguments and return values) is sent over the wire and stored by Temporal Server,
- * it is encoded in binary in a {@link Payload} Protobuf message.
+ * When your data (arguments and return values) is sent over the wire and stored by Temporal Server, it is encoded in
+ * binary in a {@link Payload} Protobuf message.
  *
- * The default `DataConverter` supports `Uint8Array`, and JSON serializables (so if [`JSON.stringify(yourArgOrRetval)`](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/JSON/stringify#description) works, the default data converter will work).
+ * The default `DataConverter` supports `Uint8Array`, and JSON serializables (so if
+ * [`JSON.stringify(yourArgOrRetval)`](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/JSON/stringify#description)
+ * works, the default data converter will work). Protobufs are supported via [this
+ * API](https://docs.temporal.io/docs/typescript/data-converters#protobufs).
  *
- * Use a custom `DataConverter` to control the contents of your {@link Payload}s.
- * Common reasons for using a custom `DataConverter` are:
- * - Converting values that are not supported by the default `DataConverter` (for example, `JSON.stringify()` doesn't handle `BigInt`s, so if you want to return `{ total: 1000n }` from a Workflow, Signal, or Activity, you need your own `DataConverter`).
- * - Encrypting values that may contain private information that you don't want stored in plaintext in Temporal Server's database.
+ * Use a custom `DataConverter` to control the contents of your {@link Payload}s. Common reasons for using a custom
+ * `DataConverter` are:
+ * - Converting values that are not supported by the default `DataConverter` (for example, `JSON.stringify()` doesn't
+ *   handle `BigInt`s, so if you want to return `{ total: 1000n }` from a Workflow, Signal, or Activity, you need your
+ *   own `DataConverter`).
+ * - Encrypting values that may contain private information that you don't want stored in plaintext in Temporal Server's
+ *   database.
  * - Compressing values to reduce disk or network usage.
  *
- * To use your custom `DataConverter`, provide it to the {@link WorkflowClient}, {@link Worker}, and {@link bundleWorkflowCode} (if you use it):
+ * To use your custom `DataConverter`, provide it to the {@link WorkflowClient}, {@link Worker}, and
+ * {@link bundleWorkflowCode} (if you use it):
  * - `new WorkflowClient({ ..., dataConverter })`
  * - `Worker.create({ ..., dataConverter })`
  * - `bundleWorkflowCode({ ..., payloadConverterPath })`

packages/common/src/converter/payload-codec.ts

@@ -3,7 +3,7 @@ import { Payload } from './types';
 /**
  * `PayloadCodec` is an optional step that happens between the wire and the {@link PayloadConverter}:
  *
- * Temporal Server ↔️ Wire ↔️ `PayloadCodec` ↔️ `PayloadConverter` ↔️ User code
+ * Temporal Server <--> Wire <--> `PayloadCodec` <--> `PayloadConverter` <--> User code
  *
  * Implement this to transform an array of {@link Payload}s to/from the format sent over the wire and stored by Temporal Server.
  * Common transformations are encryption and compression.

packages/common/src/converter/payload-converter.ts

@@ -1,4 +1,4 @@
-import { UnsupportedTypeError, ValueError } from '@temporalio/internal-workflow-common';
+import { PayloadConverterError, ValueError } from '@temporalio/internal-workflow-common';
 import {
   BinaryPayloadConverter,
   JsonPayloadConverter,
@@ -8,8 +8,7 @@ import {
 import { METADATA_ENCODING_KEY, Payload, str } from './types';
 
 /**
- * Used by the framework to serialize/deserialize parameters and return values that need to be
- * sent over the wire.
+ * Used by the framework to serialize/deserialize parameters and return values.
  *
  * This is called inside the [Workflow isolate](https://docs.temporal.io/docs/typescript/determinism).
  * To write async code or use Node APIs (or use packages that use Node APIs), use a {@link PayloadCodec}.
@@ -19,7 +18,7 @@ export interface PayloadConverter {
    * Converts a value to a {@link Payload}.
    * @param value The value to convert. Example values include the Workflow args sent by the client and the values returned by a Workflow or Activity.
    */
-  toPayload<T>(value: T): Payload;
+  toPayload<T>(value: T): Payload | undefined;
 
   /**
    * Converts a {@link Payload} back to a value.
@@ -40,23 +39,24 @@ export class CompositePayloadConverter implements PayloadConverter {
 
   /**
    * Tries to run `.toPayload(value)` on each converter in the order provided at construction.
-   * Returns the first successful result.
-   * @throws {@link ValueError} if no converter can convert the value
+   * Returns the first successful result, or `undefined` if there is no converter that can handle the value.
    */
-  public toPayload<T>(value: T): Payload {
+  public toPayload<T>(value: T): Payload | undefined {
     for (const converter of this.converters) {
       try {
         const result = converter.toPayload(value);
-        return result;
+        if (result !== undefined) {
+          return result;
+        }
       } catch (e: unknown) {
-        if (e instanceof UnsupportedTypeError) {
+        if (e instanceof PayloadConverterError) {
           continue;
         } else {
           throw e;
         }
       }
     }
-    throw new ValueError(`Cannot serialize ${value} of type ${typeof value}`);
+    return undefined;
   }
 
   /**
@@ -75,20 +75,45 @@ export class CompositePayloadConverter implements PayloadConverter {
   }
 }
 
+/**
+ * Tries to convert `value` to a {@link Payload}. Throws if conversion fails.
+ *
+ * @throws {@link PayloadConverterError}
+ */
+export function toPayload(converter: PayloadConverter, value: unknown): Payload {
+  const payload = converter.toPayload(value);
+  if (payload === undefined) {
+    throw new PayloadConverterError(`Failed to convert value: ${value}`);
+  }
+  return payload;
+}
+
 /**
  * Implements conversion of a list of values.
  *
  * @param converter
- * @param values JS values to convert to Payloads.
- * @return converted value
+ * @param values JS values to convert to Payloads
+ * @return converted values
  * @throws PayloadConverterError if conversion of the value passed as parameter failed for any
  *     reason.
  */
 export function toPayloads(converter: PayloadConverter, ...values: unknown[]): Payload[] | undefined {
   if (values.length === 0) {
     return undefined;
   }
-  return values.map((value) => converter.toPayload(value));
+
+  return values.map((value) => toPayload(converter, value));
+}
+
+/**
+ * Run {@link PayloadConverter.toPayload} on each value in the map.
+ *
+ * @throws {@link PayloadConverterError} if conversion of any value in the map fails
+ */
+export function mapToPayloads<K extends string>(converter: PayloadConverter, map: Record<K, any>): Record<K, Payload> {
+  return Object.fromEntries(
+    Object.entries(map).map(([k, v]): [K, Payload] => [k as K, toPayload(converter, v)])
+  ) as Record<K, Payload>;
 }
 
 /**
@@ -99,7 +124,7 @@ export function toPayloads(converter: PayloadConverter, ...values: unknown[]): P
  * @param index index of the value in the payloads
  * @param payloads serialized value to convert to JS values.
  * @return converted JS value
- * @throws PayloadConverterError if conversion of the data passed as parameter failed for any
+ * @throws {@link PayloadConverterError} if conversion of the data passed as parameter failed for any
  *     reason.
  */
 export function fromPayloadsAtIndex<T>(converter: PayloadConverter, index: number, payloads?: Payload[] | null): T {
@@ -120,15 +145,6 @@ export function arrayFromPayloads(converter: PayloadConverter, payloads?: Payloa
   return payloads.map((payload: Payload) => converter.fromPayload(payload));
 }
 
-/**
- * Run {@link PayloadConverter.toPayload} on each value in the map.
- */
-export function mapToPayloads<K extends string>(converter: PayloadConverter, map: Record<K, any>): Record<K, Payload> {
-  return Object.fromEntries(
-    Object.entries(map).map(([k, v]): [K, Payload] => [k as K, converter.toPayload(v)])
-  ) as Record<K, Payload>;
-}
-
 export class DefaultPayloadConverter extends CompositePayloadConverter {
   // Match the order used in other SDKs, but exclude Protobuf converters so that the code, including
   // `proto3-json-serializer`, doesn't take space in Workflow bundles that don't use Protobufs. To use Protobufs, use