External Storage pipeline integration

packages/client/src/workflow-client.ts

@@ -39,6 +39,21 @@ import {
   decodeOptionalSinglePayload,
   encodeMapToPayloads,
   encodeToPayloadsWithContext,
+  extstoreRetrieveOptions,
+  extstoreStoreOptions,
+  visit,
+  walkExecuteMultiOperationRequest,
+  walkExecuteMultiOperationResponse,
+  walkGetWorkflowExecutionHistoryResponse,
+  walkPollWorkflowExecutionUpdateResponse,
+  walkQueryWorkflowRequest,
+  walkQueryWorkflowResponse,
+  walkSignalWithStartWorkflowExecutionRequest,
+  walkSignalWorkflowExecutionRequest,
+  walkStartWorkflowExecutionRequest,
+  walkTerminateWorkflowExecutionRequest,
+  walkUpdateWorkflowExecutionRequest,
+  walkUpdateWorkflowExecutionResponse,
 } from '@temporalio/common/lib/internal-non-workflow';
 import { filterNullAndUndefined } from '@temporalio/common/lib/internal-workflow';
 import { temporal } from '@temporalio/proto';
@@ -818,6 +833,10 @@ export class WorkflowClient extends BaseClient {
       } catch (err) {
         this.rethrowGrpcError(err, 'Failed to get Workflow execution history', { workflowId, runId });
       }
+      const externalStorage = this.dataConverter.externalStorage;
+      if (externalStorage) {
+        await visit(res, walkGetWorkflowExecutionHistoryResponse, extstoreRetrieveOptions(externalStorage));
+      }
       const events = res.history?.events;
 
       if (events == null || events.length === 0) {
@@ -962,6 +981,20 @@ export class WorkflowClient extends BaseClient {
         header: { fields: input.headers },
       },
     };
+    const externalStorage = this.dataConverter.externalStorage;
+    if (externalStorage) {
+      await visit(
+        req,
+        walkQueryWorkflowRequest,
+        extstoreStoreOptions(externalStorage, {
+          initialTarget: {
+            kind: 'workflow',
+            namespace: this.options.namespace,
+            id: input.workflowExecution.workflowId ?? undefined,
+          },
+        })
+      );
+    }
     let response: temporal.api.workflowservice.v1.QueryWorkflowResponse;
     try {
       response = await this.workflowService.queryWorkflow(req);
@@ -974,6 +1007,9 @@ export class WorkflowClient extends BaseClient {
       }
       this.rethrowGrpcError(err, 'Failed to query Workflow', input.workflowExecution);
     }
+    if (externalStorage) {
+      await visit(response, walkQueryWorkflowResponse, extstoreRetrieveOptions(externalStorage));
+    }
     if (response.queryRejected) {
       if (response.queryRejected.status === undefined || response.queryRejected.status === null) {
         throw new TypeError('Received queryRejected from server with no status');
@@ -1034,6 +1070,20 @@ export class WorkflowClient extends BaseClient {
         : UpdateWorkflowExecutionLifecycleStage.UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_ACCEPTED;
 
     const request = await this._createUpdateWorkflowRequest(waitForStageProto, input);
+    const externalStorage = this.dataConverter.externalStorage;
+    if (externalStorage) {
+      await visit(
+        request,
+        walkUpdateWorkflowExecutionRequest,
+        extstoreStoreOptions(externalStorage, {
+          initialTarget: {
+            kind: 'workflow',
+            namespace: this.options.namespace,
+            id: input.workflowExecution.workflowId ?? undefined,
+          },
+        })
+      );
+    }
 
     // Repeatedly send UpdateWorkflowExecution until update is durable (if the server receives a request with
     // an update ID that already exists, it responds with information for the existing update). If the
@@ -1048,6 +1098,9 @@ export class WorkflowClient extends BaseClient {
     } catch (err) {
       this.rethrowUpdateGrpcError(err, 'Workflow Update failed', input.workflowExecution);
     }
+    if (externalStorage) {
+      await visit(response, walkUpdateWorkflowExecutionResponse, extstoreRetrieveOptions(externalStorage));
+    }
     return {
       updateId: request.request!.meta!.updateId!,
 
@@ -1106,8 +1159,25 @@ export class WorkflowClient extends BaseClient {
       // Repeatedly send ExecuteMultiOperation until update is durable (if the server receives a request with
       // an update ID that already exists, it responds with information for the existing update). If the
       // requested wait stage is COMPLETED, further polling is done before returning the UpdateHandle.
+      const externalStorage = this.dataConverter.externalStorage;
+      if (externalStorage) {
+        await visit(
+          multiOpReq,
+          walkExecuteMultiOperationRequest,
+          extstoreStoreOptions(externalStorage, {
+            initialTarget: {
+              kind: 'workflow',
+              namespace: this.options.namespace,
+              id: input.workflowStartOptions.workflowId,
+            },
+          })
+        );
+      }
       do {
         multiOpResp = await this.workflowService.executeMultiOperation(multiOpReq);
+        if (externalStorage) {
+          await visit(multiOpResp, walkExecuteMultiOperationResponse, extstoreRetrieveOptions(externalStorage));
+        }
         startResp = multiOpResp.responses?.[0]
           ?.startWorkflow as temporal.api.workflowservice.v1.IStartWorkflowExecutionResponse;
         if (!seenStart) {
@@ -1193,6 +1263,10 @@ export class WorkflowClient extends BaseClient {
     for (;;) {
       try {
         const response = await this.workflowService.pollWorkflowExecutionUpdate(req);
+        const externalStorage = this.dataConverter.externalStorage;
+        if (externalStorage) {
+          await visit(response, walkPollWorkflowExecutionUpdateResponse, extstoreRetrieveOptions(externalStorage));
+        }
         if (response.outcome) {
           return response.outcome;
         }
@@ -1224,6 +1298,20 @@ export class WorkflowClient extends BaseClient {
       links: internalOptions?.links,
     };
     try {
+      const externalStorage = this.dataConverter.externalStorage;
+      if (externalStorage) {
+        await visit(
+          req,
+          walkSignalWorkflowExecutionRequest,
+          extstoreStoreOptions(externalStorage, {
+            initialTarget: {
+              kind: 'workflow',
+              namespace: this.options.namespace,
+              id: input.workflowExecution.workflowId,
+            },
+          })
+        );
+      }
       const response = await this.workflowService.signalWorkflowExecution(req);
       if (internalOptions != null) {
         // Servers that support CHASM signal response links (1.31 and up) return a response link
@@ -1281,6 +1369,16 @@ export class WorkflowClient extends BaseClient {
       links: internalOptions?.links,
     };
     try {
+      const externalStorage = this.dataConverter.externalStorage;
+      if (externalStorage) {
+        await visit(
+          req,
+          walkSignalWithStartWorkflowExecutionRequest,
+          extstoreStoreOptions(externalStorage, {
+            initialTarget: { kind: 'workflow', namespace: this.options.namespace, id: req.workflowId ?? undefined },
+          })
+        );
+      }
       const response = await this.workflowService.signalWithStartWorkflowExecution(req);
       if (internalOptions != null) {
         // Servers that support CHASM signal response links (1.31 and up) return a response link
@@ -1310,6 +1408,20 @@ export class WorkflowClient extends BaseClient {
     const { options: opts, workflowType } = input;
     const internalOptions = (opts as InternalWorkflowStartOptions)[InternalWorkflowStartOptionsSymbol];
     try {
+      const externalStorage = this.dataConverter.externalStorage;
+      if (externalStorage) {
+        await visit(
+          req,
+          walkStartWorkflowExecutionRequest,
+          extstoreStoreOptions(externalStorage, {
+            initialTarget: {
+              kind: 'workflow',
+              namespace: req.namespace ?? this.options.namespace,
+              id: req.workflowId ?? undefined,
+            },
+          })
+        );
+      }
       const response = await this.workflowService.startWorkflowExecution(req);
       if (internalOptions != null) {
         internalOptions.responseLink = response.link ?? undefined;
@@ -1401,6 +1513,20 @@ export class WorkflowClient extends BaseClient {
       firstExecutionRunId: input.firstExecutionRunId,
     };
     try {
+      const externalStorage = this.dataConverter.externalStorage;
+      if (externalStorage) {
+        await visit(
+          req,
+          walkTerminateWorkflowExecutionRequest,
+          extstoreStoreOptions(externalStorage, {
+            initialTarget: {
+              kind: 'workflow',
+              namespace: this.options.namespace,
+              id: input.workflowExecution.workflowId ?? undefined,
+            },
+          })
+        );
+      }
       return await this.workflowService.terminateWorkflowExecution(req);
     } catch (err) {
       this.rethrowGrpcError(err, 'Failed to terminate Workflow', input.workflowExecution);

packages/common/src/__tests__/test-payload-visitor.ts

@@ -2,7 +2,11 @@ import test from 'ava';
 import { coresdk } from '@temporalio/proto';
 import { limit } from '../concurrency/limit';
 import type { Payload } from '../interfaces';
-import { visitWorkflowActivation, visitWorkflowActivationCompletion } from '../internal-non-workflow/payload-visitor';
+import {
+  visit,
+  walkWorkflowActivation,
+  walkWorkflowActivationCompletion,
+} from '../internal-non-workflow/payload-visitor';
 
 // Lets the event loop run every async step that is currently ready
 const tick = (): Promise<void> => new Promise((resolve) => setImmediate(resolve));
@@ -61,7 +65,7 @@ test('a failing transform surfaces its error and aborts in-flight siblings', asy
   };
 
   const error = await t.throwsAsync(
-    visitWorkflowActivation(activation, {
+    visit(activation, walkWorkflowActivation, {
       transformPayload: async (p, _context, signal) => {
         const tag = read(p);
         if (tag === 'boom') {
@@ -94,7 +98,7 @@ test('visitWorkflowActivationCompletion with an already-aborted signal skips the
   const counter = countingTransforms();
 
   const error = await t.throwsAsync(
-    visitWorkflowActivationCompletion(completionWith('a', 'b'), {
+    visit(completionWith('a', 'b'), walkWorkflowActivationCompletion, {
       ...counter.transforms,
       abortSignal: AbortSignal.abort(new Error('stop')),
     })
@@ -161,7 +165,7 @@ test('visits every payload site exactly once and writes back in place', async (t
   const activation = activationWithEveryPayloadSite();
 
   const seen: string[] = [];
-  await visitWorkflowActivation(activation, {
+  await visit(activation, walkWorkflowActivation, {
     ...perPayload<void>((p) => {
       seen.push(read(p));
       return payload(`${read(p)}#`);
@@ -172,8 +176,9 @@ test('visits every payload site exactly once and writes back in place', async (t
 
   // A second walk should observe the first walk's rewrites, proving writeback landed in place.
   const seenAgain: string[] = [];
-  await visitWorkflowActivation(
+  await visit(
     activation,
+    walkWorkflowActivation,
     perPayload<void>((p) => {
       seenAgain.push(read(p));
       return p;
@@ -189,10 +194,11 @@ test('visits every payload site exactly once and writes back in place', async (t
 test('an activation with no payloads makes no transform calls and resolves', async (t) => {
   const counter = countingTransforms();
 
-  await visitWorkflowActivation({ jobs: [] }, counter.transforms);
-  await visitWorkflowActivation({}, counter.transforms);
-  await visitWorkflowActivation(
+  await visit({ jobs: [] }, walkWorkflowActivation, counter.transforms);
+  await visit({}, walkWorkflowActivation, counter.transforms);
+  await visit(
     { jobs: [{ initializeWorkflow: { arguments: [], headers: {} } }, { fireTimer: {} }] },
+    walkWorkflowActivation,
     counter.transforms
   );
 
@@ -205,11 +211,11 @@ test('a repeated field may return any count, including zero', async (t) => {
   });
 
   const emptied = withInput();
-  await visitWorkflowActivation(emptied, { ...identity, transformPayloads: async () => [] });
+  await visit(emptied, walkWorkflowActivation, { ...identity, transformPayloads: async () => [] });
   t.deepEqual(emptied.jobs![0]!.signalWorkflow!.input, [], 'a repeated list can be emptied');
 
   const reframed = withInput();
-  await visitWorkflowActivation(reframed, { ...identity, transformPayloads: async () => [payload('merged')] });
+  await visit(reframed, walkWorkflowActivation, { ...identity, transformPayloads: async () => [payload('merged')] });
   t.deepEqual(reframed.jobs![0]!.signalWorkflow!.input!.map(read), ['merged'], 'count may shrink');
 });
 
@@ -219,7 +225,7 @@ test('context is derived per message and isolated across sibling jobs', async (t
   };
 
   const seen: Record<string, string> = {};
-  await visitWorkflowActivation<string>(activation, {
+  await visit(activation, walkWorkflowActivation, {
     initialContext: 'root',
     deriveContext: (_message, typeName, context) =>
       typeName === 'coresdk.workflow_activation.SignalWorkflow'
@@ -250,7 +256,7 @@ test('skipHeaders and skipSearchAttributes omit those sites', async (t) => {
   };
 
   const seen: string[] = [];
-  await visitWorkflowActivation(activation, {
+  await visit(activation, walkWorkflowActivation, {
     ...perPayload<void>((p) => {
       seen.push(read(p));
       return p;
@@ -273,8 +279,9 @@ test('visits completion command payloads: user metadata and a rejected update',
   };
 
   const seen: string[] = [];
-  await visitWorkflowActivationCompletion(
+  await visit(
     completion,
+    walkWorkflowActivationCompletion,
     perPayload<void>((p) => {
       seen.push(read(p));
       return payload(`${read(p)}!`);
@@ -292,8 +299,9 @@ test('walks a decoded protobuf message instance, not just object literals', asyn
   const Type = coresdk.workflow_completion.WorkflowActivationCompletion;
   const decoded = Type.decode(Type.encode(completionWith('a', 'b')).finish());
 
-  await visitWorkflowActivationCompletion(
+  await visit(
     decoded,
+    walkWorkflowActivationCompletion,
     perPayload<void>((p) => payload(`${read(p)}!`))
   );
 
@@ -306,7 +314,7 @@ test('an identity transform leaves a real message byte-for-byte unchanged', asyn
   const original = Type.encode(activationWithEveryPayloadSite()).finish();
   const decoded = Type.decode(original);
 
-  await visitWorkflowActivation(decoded, identity);
+  await visit(decoded, walkWorkflowActivation, identity);
 
   t.deepEqual(Buffer.from(Type.encode(decoded).finish()), Buffer.from(original));
 });
@@ -344,7 +352,7 @@ test('a concurrency limit caps in-flight transforms across the whole activation'
   };
 
   const capped = tracked();
-  const cappedVisit = visitWorkflowActivation(activationWithEveryPayloadSite(), {
+  const cappedVisit = visit(activationWithEveryPayloadSite(), walkWorkflowActivation, {
     ...capped.transforms,
     limit: limit(4),
   });
@@ -354,7 +362,7 @@ test('a concurrency limit caps in-flight transforms across the whole activation'
   await cappedVisit;
 
   const sequential = tracked();
-  const sequentialVisit = visitWorkflowActivation(activationWithEveryPayloadSite(), {
+  const sequentialVisit = visit(activationWithEveryPayloadSite(), walkWorkflowActivation, {
     ...sequential.transforms,
     limit: limit(1),
   });
@@ -377,7 +385,7 @@ test('aborting mid-walk rejects and cancels in-flight transforms', async (t) =>
     }
   };
 
-  const visiting = visitWorkflowActivation(activationWithEveryPayloadSite(), {
+  const visiting = visit(activationWithEveryPayloadSite(), walkWorkflowActivation, {
     transformPayload: (p, _context, signal) => waitThenAbort(p, signal),
     transformPayloads: (ps, _context, signal) => waitThenAbort(ps, signal),
     limit: limit(16),

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

@@ -1,3 +1,4 @@
+import type { ExternalStorage } from './extstore';
 import type { FailureConverter } from './failure-converter';
 import { DefaultFailureConverter } from './failure-converter';
 import type { PayloadCodec } from './payload-codec';
@@ -60,6 +61,12 @@ export interface LoadedDataConverter {
   payloadConverter: PayloadConverter;
   failureConverter: FailureConverter;
   payloadCodecs: PayloadCodec[];
+
+  /**
+   * @internal
+   * @experimental
+   */
+  externalStorage?: ExternalStorage;
 }
 
 /**