feat(extstore): Add extstore store/retrieve to pipeline via LoadedDataConverter. Later, we can add ExternalStorage config to DataConverter to 'enable' it.

cconstable · cconstable · commit 041d6f119dc3 · 2026-07-13T14:54:45.000-04:00
diff --git a/packages/common/src/converter/data-converter.ts b/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;
 }
 
 /**
diff --git a/packages/common/src/internal-non-workflow/external-storage-visitor.ts b/packages/common/src/internal-non-workflow/external-storage-visitor.ts
@@ -0,0 +1,67 @@
+/**
+ * Visits payloads for External Storage and applies payload transformations via
+ * {@link ExternalStorageRunner}.
+ *
+ * @module
+ */
+import type { coresdk } from '@temporalio/proto';
+
+import type { ConcurrencyLimit } from '../concurrency/limit';
+import type { ExternalStorage, StorageDriverTargetInfo } from '../converter/extstore';
+import { ExternalStorageRunner } from './external-storage-runner';
+import { visitWorkflowActivation, visitWorkflowActivationCompletion } from './payload-visitor';
+
+/**
+ * Offload every eligible payload in a `WorkflowActivationCompletion` to external storage,
+ * replacing it in place with a reference payload.
+ *
+ * @internal
+ * @experimental
+ */
+export function storeWorkflowActivationCompletion(
+  externalStorage: ExternalStorage,
+  completion: coresdk.workflow_completion.IWorkflowActivationCompletion,
+  {
+    target,
+    limit,
+    abortSignal,
+  }: {
+    /** Identity of the workflow / activity that produced the payloads, used for object naming. */
+    target?: StorageDriverTargetInfo;
+    /** Bounds concurrent transform calls across payload sites. Omit for sequential. */
+    limit?: ConcurrencyLimit;
+    /** Aborts the walk and every in-flight driver call. */
+    abortSignal?: AbortSignal;
+  } = {}
+): Promise<void> {
+  const runner = new ExternalStorageRunner(externalStorage);
+  return visitWorkflowActivationCompletion(completion, {
+    transformPayloads: (payloads, _context, signal) => runner.store(payloads, { target, abortSignal: signal }),
+    transformPayload: (payload, _context, signal) =>
+      runner.store([payload], { target, abortSignal: signal }).then((stored) => stored[0]!),
+    limit,
+    abortSignal,
+  });
+}
+
+/**
+ * Replace every reference payload in a `WorkflowActivation` with the payload bytes fetched
+ * from external storage, in place.
+ *
+ * @internal
+ * @experimental
+ */
+export function retrieveWorkflowActivation(
+  externalStorage: ExternalStorage,
+  activation: coresdk.workflow_activation.IWorkflowActivation,
+  { limit, abortSignal }: { limit?: ConcurrencyLimit; abortSignal?: AbortSignal } = {}
+): Promise<void> {
+  const runner = new ExternalStorageRunner(externalStorage);
+  return visitWorkflowActivation(activation, {
+    transformPayloads: (payloads, _context, signal) => runner.retrieve(payloads, { abortSignal: signal }),
+    transformPayload: (payload, _context, signal) =>
+      runner.retrieve([payload], { abortSignal: signal }).then((retrieved) => retrieved[0]!),
+    limit,
+    abortSignal,
+  });
+}
diff --git a/packages/common/src/internal-non-workflow/index.ts b/packages/common/src/internal-non-workflow/index.ts
@@ -8,6 +8,7 @@ export * from './codec-types';
 export * from './data-converter-helpers';
 export * from './extstore-helpers';
 export * from './external-storage-runner';
+export * from './external-storage-visitor';
 export * from './parse-host-uri';
 export * from './proxy-config';
 export * from './tls-config';
diff --git a/packages/test/src/test-extstore-visit.ts b/packages/test/src/test-extstore-visit.ts
@@ -0,0 +1,78 @@
+/* eslint @typescript-eslint/no-non-null-assertion: 0 */
+import test from 'ava';
+import type { Payload } from '@temporalio/common';
+import { ExternalStorage } from '@temporalio/common/lib/converter/extstore';
+import {
+  isReferencePayload,
+  retrieveWorkflowActivation,
+  storeWorkflowActivationCompletion,
+} from '@temporalio/common/lib/internal-non-workflow';
+import { encode } from '@temporalio/common/lib/encoding';
+import { METADATA_ENCODING_KEY } from '@temporalio/common/lib/converter/types';
+import { coresdk } from '@temporalio/proto';
+import { makeFakeDriver } from './extstore-fake-driver';
+
+/** Build a Payload whose proto-encoded size is at least `bodyBytes`, filled with `fill`. */
+function makePayload(bodyBytes: number, fill = 0): Payload {
+  return {
+    metadata: { [METADATA_ENCODING_KEY]: encode('binary/plain') },
+    data: new Uint8Array(bodyBytes).fill(fill),
+  };
+}
+
+function externalStorageWith(driver = makeFakeDriver({ name: 's3' }), payloadSizeThreshold = 96) {
+  return { driver, externalStorage: new ExternalStorage({ drivers: [driver], payloadSizeThreshold }) };
+}
+
+const WORKFLOW_TARGET = { kind: 'workflow', namespace: 'ns', id: 'wf-1', runId: 'run-1' } as const;
+
+test('store offloads a large payload at a singular site and threads the target to the driver', async (t) => {
+  const { driver, externalStorage } = externalStorageWith();
+  const completion: coresdk.workflow_completion.IWorkflowActivationCompletion = {
+    successful: { commands: [{ completeWorkflowExecution: { result: makePayload(256) } }] },
+  };
+
+  await storeWorkflowActivationCompletion(externalStorage, completion, { target: WORKFLOW_TARGET });
+
+  const result = completion.successful!.commands![0]!.completeWorkflowExecution!.result!;
+  t.true(isReferencePayload(result));
+  t.is(driver.storeCalls.length, 1);
+  t.deepEqual(driver.storeCalls[0]!.context.target, WORKFLOW_TARGET);
+});
+
+test('store offloads only above-threshold payloads at a repeated site', async (t) => {
+  const { externalStorage } = externalStorageWith();
+  const small = makePayload(0);
+  const completion: coresdk.workflow_completion.IWorkflowActivationCompletion = {
+    successful: { commands: [{ scheduleActivity: { arguments: [makePayload(256), small] } }] },
+  };
+
+  await storeWorkflowActivationCompletion(externalStorage, completion, { target: WORKFLOW_TARGET });
+
+  const args = completion.successful!.commands![0]!.scheduleActivity!.arguments!;
+  t.true(isReferencePayload(args[0]!));
+  t.deepEqual(args[1], small);
+});
+
+test('store then retrieve round-trips the original payload bytes through a worker message', async (t) => {
+  const { driver, externalStorage } = externalStorageWith();
+  const original = makePayload(256, 7);
+  const completion: coresdk.workflow_completion.IWorkflowActivationCompletion = {
+    successful: { commands: [{ completeWorkflowExecution: { result: original } }] },
+  };
+
+  await storeWorkflowActivationCompletion(externalStorage, completion, { target: WORKFLOW_TARGET });
+  const reference = completion.successful!.commands![0]!.completeWorkflowExecution!.result!;
+  t.true(isReferencePayload(reference));
+
+  // The reference produced on the outbound path arrives back on a later inbound activation.
+  const activation: coresdk.workflow_activation.IWorkflowActivation = {
+    jobs: [{ resolveActivity: { result: { completed: { result: reference } } } }],
+  };
+  await retrieveWorkflowActivation(externalStorage, activation);
+
+  const retrieved = activation.jobs![0]!.resolveActivity!.result!.completed!.result!;
+  t.false(isReferencePayload(retrieved));
+  t.deepEqual(retrieved, original);
+  t.is(driver.retrieveCalls.length, 1);
+});
diff --git a/packages/test/src/test-workflow-codec-runner.ts b/packages/test/src/test-workflow-codec-runner.ts
@@ -20,7 +20,10 @@ function failureWithDetail(label: string) {
   );
 }
 
-function decodeCompletion(bytes: Uint8Array): coresdk.workflow_completion.WorkflowActivationCompletion {
+function decodeCompletion(
+  completion: coresdk.workflow_completion.IWorkflowActivationCompletion
+): coresdk.workflow_completion.WorkflowActivationCompletion {
+  const bytes = coresdk.workflow_completion.WorkflowActivationCompletion.encodeDelimited(completion).finish();
   return coresdk.workflow_completion.WorkflowActivationCompletion.decodeDelimited(bytes);
 }
 
diff --git a/packages/worker/src/worker.ts b/packages/worker/src/worker.ts
@@ -31,6 +31,8 @@ import {
   decodeFromPayloadsAtIndex,
   encodeErrorToFailure,
   encodeToPayload,
+  retrieveWorkflowActivation,
+  storeWorkflowActivationCompletion,
 } from '@temporalio/common/lib/internal-non-workflow';
 import { historyFromJSON } from '@temporalio/common/lib/proto-utils';
 import type { Duration } from '@temporalio/common/lib/time';
@@ -1441,6 +1443,10 @@ export class Worker {
           workflowId,
         });
       }
+      const { externalStorage } = this.options.loadedDataConverter;
+      if (externalStorage) {
+        await retrieveWorkflowActivation(externalStorage, activation);
+      }
       const decodedActivation = await workflowCodecRunner.decodeActivation(activation);
 
       if (workflow === undefined) {
@@ -1456,7 +1462,19 @@ export class Worker {
       let isFatalError = false;
       try {
         const unencodedCompletion = await workflow.workflow.activate(decodedActivation);
-        const completion = await workflowCodecRunner.encodeCompletion(unencodedCompletion);
+        const encodedCompletion = await workflowCodecRunner.encodeCompletion(unencodedCompletion);
+        if (externalStorage) {
+          await storeWorkflowActivationCompletion(externalStorage, encodedCompletion, {
+            target: {
+              kind: 'workflow',
+              namespace: workflowCodecRunner.workflowContext.namespace,
+              id: workflowCodecRunner.workflowContext.workflowId,
+              runId: activation.runId,
+            },
+          });
+        }
+        const completion =
+          coresdk.workflow_completion.WorkflowActivationCompletion.encodeDelimited(encodedCompletion).finish();
 
         return { state: workflow, output: { close, completion } };
       } catch (err) {
diff --git a/packages/worker/src/workflow-codec-runner.ts b/packages/worker/src/workflow-codec-runner.ts
@@ -34,7 +34,7 @@ export class WorkflowCodecRunner {
 
   constructor(
     private readonly codecs: PayloadCodec[],
-    private readonly workflowContext: WorkflowSerializationContext
+    public readonly workflowContext: WorkflowSerializationContext
   ) {}
 
   private consumeContext<TContext extends SerializationContext>(
@@ -368,7 +368,7 @@ export class WorkflowCodecRunner {
    */
   public async encodeCompletion(
     completion: coresdk.workflow_completion.IWorkflowActivationCompletion
-  ): Promise<Uint8Array> {
+  ): Promise<Encoded<coresdk.workflow_completion.IWorkflowActivationCompletion>> {
     const encodedCompletion: Encoded<coresdk.workflow_completion.IWorkflowActivationCompletion> = {
       ...completion,
       failed: completion.failed
@@ -608,6 +608,6 @@ export class WorkflowCodecRunner {
         : null,
     };
 
-    return coresdk.workflow_completion.WorkflowActivationCompletion.encodeDelimited(encodedCompletion).finish();
+    return encodedCompletion;
   }
 }

Original file line number	Diff line number	Diff line change
`@@ -20,7 +20,10 @@ function failureWithDetail(label: string) {`
`20`	`20`	`);`
`21`	`21`	`}`
`22`	`22`
`23`		`-function decodeCompletion(bytes: Uint8Array): coresdk.workflow_completion.WorkflowActivationCompletion {`
	`23`	`+function decodeCompletion(`
	`24`	`+ completion: coresdk.workflow_completion.IWorkflowActivationCompletion`
	`25`	`+): coresdk.workflow_completion.WorkflowActivationCompletion {`
	`26`	`+ const bytes = coresdk.workflow_completion.WorkflowActivationCompletion.encodeDelimited(completion).finish();`
`24`	`27`	`return coresdk.workflow_completion.WorkflowActivationCompletion.decodeDelimited(bytes);`
`25`	`28`	`}`
`26`	`29`