Fix ProtobufJSONPayloadConverter to not require Buffer in the workflo… (#1170)

## What was changed
<!-- Describe what has changed in this PR -->
This is a fix to #1169 

The ProtobufJsonPayloadConverter ensures that Buffer is globally
available, and if it is not, it just adds it when needed, but leaving
the global space untouched after that.

It also needs to ensure that no Buffer objects are leaked, replacing
them with Uint8Array objects.

## Why?

Protobufs containing binary fields cannot be properly encoded/decoded
with this converter without the Buffer class, and this class is not
available by default in the workflow context.

## Checklist

1. Closes <!-- add issue number here -->
#1169 

2. How was this tested:
<!--- Please describe how you tested your changes/how we can test them
-->
Added a system test

Author: antlai-temporal

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

@@ -14,6 +14,8 @@ import {
 
 import { encodingTypes, METADATA_ENCODING_KEY, METADATA_MESSAGE_TYPE_KEY } from './types';
 
+const GLOBAL_BUFFER = globalThis.constructor.constructor('return globalThis.Buffer')();
+
 abstract class ProtobufPayloadConverter implements PayloadConverterWithEncoding {
   protected readonly root: Root | undefined;
   public abstract encodingType: string;
@@ -121,17 +123,69 @@ export class ProtobufJsonPayloadConverter extends ProtobufPayloadConverter {
       return undefined;
     }
 
-    const jsonValue = protoJsonSerializer.toProto3JSON(value);
-
-    return this.constructPayload({
-      messageTypeName: getNamespacedTypeName(value.$type),
-      message: encode(JSON.stringify(jsonValue)),
-    });
+    const hasBufferChanged = setBufferInGlobal();
+    try {
+      const jsonValue = protoJsonSerializer.toProto3JSON(value);
+
+      return this.constructPayload({
+        messageTypeName: getNamespacedTypeName(value.$type),
+        message: encode(JSON.stringify(jsonValue)),
+      });
+    } finally {
+      resetBufferInGlobal(hasBufferChanged);
+    }
   }
 
   public fromPayload<T>(content: Payload): T {
-    const { messageType, data } = this.validatePayload(content);
-    return protoJsonSerializer.fromProto3JSON(messageType, JSON.parse(decode(data))) as unknown as T;
+    const hasBufferChanged = setBufferInGlobal();
+    try {
+      const { messageType, data } = this.validatePayload(content);
+      const res = protoJsonSerializer.fromProto3JSON(messageType, JSON.parse(decode(data))) as unknown as T;
+      if (Buffer.isBuffer(res)) {
+        return new Uint8Array(res) as any;
+      }
+      replaceBuffers(res);
+      return res;
+    } finally {
+      resetBufferInGlobal(hasBufferChanged);
+    }
+  }
+}
+
+function replaceBuffers<X>(obj: X) {
+  const replaceBuffersImpl = <Y>(value: any, key: string | number, target: Y) => {
+    if (Buffer.isBuffer(value)) {
+      // Need to copy. `Buffer` manages a pool slab, internally reused when Buffer objects are GC.
+      type T = keyof typeof target;
+      target[key as T] = new Uint8Array(value) as any;
+    } else {
+      replaceBuffers(value);
+    }
+  };
+
+  if (obj != null && typeof obj === 'object') {
+    // Performance optimization for large arrays
+    if (Array.isArray(obj)) {
+      obj.forEach(replaceBuffersImpl);
+    } else {
+      for (const [key, value] of Object.entries(obj)) {
+        replaceBuffersImpl(value, key, obj);
+      }
+    }
+  }
+}
+
+function setBufferInGlobal(): boolean {
+  if (typeof globalThis.Buffer === 'undefined') {
+    globalThis.Buffer = GLOBAL_BUFFER;
+    return true;
+  }
+  return false;
+}
+
+function resetBufferInGlobal(hasChanged: boolean): void {
+  if (hasChanged) {
+    delete (globalThis as any).Buffer;
   }
 }
 

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

@@ -25,6 +25,7 @@ import { RUN_INTEGRATION_TESTS, Worker } from './helpers';
 import { defaultOptions } from './mock-native-worker';
 import { messageInstance } from './payload-converters/proto-payload-converter';
 import { protobufWorkflow } from './workflows/protobufs';
+import { echoBinaryProtobuf } from './workflows/echo-binary-protobuf';
 
 test('UndefinedPayloadConverter converts from undefined only', (t) => {
   const converter = new UndefinedPayloadConverter();
@@ -167,7 +168,7 @@ test('ProtobufJSONPayloadConverter converts binary', (t) => {
   });
 
   const testInstance = converter.fromPayload<root.BinaryMessage>(encoded!);
-  t.deepEqual(testInstance.data, Buffer.from(instance.data));
+  t.deepEqual(testInstance.data, instance.data);
 });
 
 if (RUN_INTEGRATION_TESTS) {
@@ -208,6 +209,30 @@ if (RUN_INTEGRATION_TESTS) {
     });
     t.pass();
   });
+
+  test('Worker encodes/decodes a protobuf containing a binary array', async (t) => {
+    const binaryInstance = root.BinaryMessage.create({ data: encode('abc') });
+    const dataConverter = { payloadConverterPath: require.resolve('./payload-converters/proto-payload-converter') };
+    const taskQueue = `${__filename}/${t.title}`;
+
+    const worker = await Worker.create({
+      ...defaultOptions,
+      workflowsPath: require.resolve('./workflows/echo-binary-protobuf'),
+      taskQueue,
+      dataConverter,
+    });
+
+    const client = new WorkflowClient({ dataConverter });
+
+    await worker.runUntil(async () => {
+      const result = await client.execute(echoBinaryProtobuf, {
+        args: [binaryInstance],
+        workflowId: uuid4(),
+        taskQueue,
+      });
+      t.deepEqual(result, binaryInstance);
+    });
+  });
 }
 
 test('DefaultPayloadConverterWithProtobufs converts protobufs', (t) => {

packages/test/src/workflows/echo-binary-protobuf.ts

@@ -0,0 +1,9 @@
+import { ApplicationFailure } from '@temporalio/common';
+import type { BinaryMessage } from '../../protos/root';
+
+export async function echoBinaryProtobuf(input: BinaryMessage): Promise<BinaryMessage> {
+  if (input.data instanceof Uint8Array) {
+    return input;
+  }
+  throw ApplicationFailure.nonRetryable('input.data is not a Uint8Array');
+}