External Storage: foundational types and config (#2092)

Author: cconstable

CONTRIBUTING.md

@@ -162,6 +162,8 @@ rm -rf ./example "$TMP_DIR"
 
 - Typescript code is linted with [eslint](https://eslint.org/)
 - Files in this repo are formatted with [prettier](https://prettier.io/)
+- Prefer explicit named re-exports and avoid wildcard re-exports where possible (`export * from ...`) in public entrypoint / barrel files.
+- Use `@experimental` and `@internal` to manage API stability and visibility. Mark new or work-in-progress exported APIs `@experimental` to signal their shape may still change. Mark a symbol `@internal` to keep it out of the generated public docs. The two are independent and may be combined. It is fine to ship something `@internal` now and promote it to public later, by removing `@internal` and adding a named re-export, once it is actually usable. The reverse is a breaking change, so prefer starting narrow.
 - Pull request titles SHOULD adhere to the [Conventional Commits specification](https://conventionalcommits.org/), for example:
 
 ```

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

@@ -65,7 +65,7 @@ export interface LoadedDataConverter {
 /**
  * The default {@link FailureConverter} used by the SDK.
  *
- * Error messages and stack traces are serizalized as plain text.
+ * Error messages and stack traces are serialized as plain text.
  */
 export const defaultFailureConverter: FailureConverter = new DefaultFailureConverter();
 

packages/common/src/converter/extstore.ts

@@ -0,0 +1,182 @@
+import { ValueError } from '../errors';
+import type { Payload } from '../interfaces';
+
+/**
+ * Reference returned from {@link StorageDriver.store}. `claimData` is an
+ * opaque key/value map the driver uses to retrieve the payload later.
+ *
+ * @internal
+ * @experimental
+ */
+export class StorageDriverClaim {
+  constructor(readonly claimData: Record<string, string>) {}
+}
+
+/**
+ * Workflow identity information passed to a storage driver.
+ *
+ * @internal
+ * @experimental
+ */
+export interface StorageDriverWorkflowInfo {
+  readonly kind: 'workflow';
+  /** The namespace of the workflow execution. */
+  readonly namespace: string;
+  /** The workflow ID. */
+  readonly id?: string;
+  /** The workflow run ID, if available. */
+  readonly runId?: string;
+  /** The workflow type name, if available. */
+  readonly type?: string;
+}
+
+/**
+ * Activity identity information passed to a storage driver.
+ *
+ * @internal
+ * @experimental
+ */
+export interface StorageDriverActivityInfo {
+  readonly kind: 'activity';
+  /** The namespace of the activity execution. */
+  readonly namespace: string;
+  /** The activity ID. */
+  readonly id?: string;
+  /** The activity run ID (only for standalone activities). */
+  readonly runId?: string;
+  /** The activity type name, if available. */
+  readonly type?: string;
+}
+
+/**
+ * Identity of the workflow or activity that produced the payloads being stored.
+ *
+ * @internal
+ * @experimental
+ */
+export type StorageDriverTargetInfo = StorageDriverWorkflowInfo | StorageDriverActivityInfo;
+
+/**
+ * Context handed to {@link StorageDriver.store} (and to the selector).
+ *
+ * @internal
+ * @experimental
+ */
+export interface StorageDriverStoreContext {
+  /** Aborts the in-flight operation; siblings are cancelled on first error. */
+  abortSignal?: AbortSignal;
+  /** Identity of the workflow / activity that produced the payloads. */
+  target?: StorageDriverTargetInfo;
+}
+
+/**
+ * Context handed to {@link StorageDriver.retrieve}.
+ *
+ * @internal
+ * @experimental
+ */
+export interface StorageDriverRetrieveContext {
+  abortSignal?: AbortSignal;
+}
+
+/**
+ * External storage backend driver.
+ *
+ * `name` is the per-instance routing key written into the wire format.
+ * `type` is a stable cross-language driver-implementation identifier
+ * reported via worker heartbeat (e.g. `"aws.s3driver"`).
+ *
+ * @internal
+ * @experimental
+ */
+export interface StorageDriver {
+  readonly name: string;
+  readonly type: string;
+  store(context: StorageDriverStoreContext, payloads: Payload[]): Promise<StorageDriverClaim[]>;
+  retrieve(context: StorageDriverRetrieveContext, claims: StorageDriverClaim[]): Promise<Payload[]>;
+}
+
+/**
+ * User-supplied function that picks the destination driver for a given
+ * payload, or returns `null` to keep the payload inline.
+ *
+ * @internal
+ * @experimental
+ */
+export type StorageDriverSelector = (context: StorageDriverStoreContext, payload: Payload) => StorageDriver | null;
+
+// ============================================================================
+// Configuration
+// ============================================================================
+
+/** Default {@link ExternalStorage.payloadSizeThreshold}: 256 KiB. */
+const DEFAULT_PAYLOAD_SIZE_THRESHOLD = 256 * 1024;
+
+/**
+ * Configuration for external storage. Holds the registered drivers, an
+ * optional selector, and the size threshold above which payloads are
+ * eligible for offloading to external storage. A selector function is
+ * required when more than one driver is registered.
+ *
+ * @internal
+ * @experimental
+ */
+export class ExternalStorage {
+  readonly drivers: StorageDriver[];
+  /**
+   * Selects the destination driver for each payload, or returns `null` to keep
+   * the payload inline.
+   */
+  readonly driverSelector: StorageDriverSelector;
+  readonly payloadSizeThreshold: number;
+  private readonly driversByName: ReadonlyMap<string, StorageDriver>;
+
+  constructor({
+    drivers,
+    driverSelector,
+    payloadSizeThreshold = DEFAULT_PAYLOAD_SIZE_THRESHOLD,
+  }: {
+    drivers: StorageDriver[];
+    driverSelector?: StorageDriverSelector;
+    /** Omit for default (256 KiB). Set `0` to consider all payloads regardless of size. */
+    payloadSizeThreshold?: number;
+  }) {
+    if (!Array.isArray(drivers) || drivers.length === 0) {
+      throw new ValueError('ExternalStorage requires at least one driver');
+    }
+    if (
+      typeof payloadSizeThreshold !== 'number' ||
+      !Number.isFinite(payloadSizeThreshold) ||
+      payloadSizeThreshold < 0
+    ) {
+      throw new ValueError(
+        `ExternalStorage.payloadSizeThreshold must be a non-negative finite number, got ${String(payloadSizeThreshold)}`
+      );
+    }
+
+    const driversByName = new Map<string, StorageDriver>();
+    for (const driver of drivers) {
+      if (typeof driver?.name !== 'string' || driver.name.length === 0) {
+        throw new ValueError("Storage driver 'name' must be a non-empty string");
+      }
+      if (driversByName.has(driver.name)) {
+        throw new ValueError(`Duplicate storage driver name: '${driver.name}'`);
+      }
+      driversByName.set(driver.name, driver);
+    }
+
+    if (driverSelector === undefined && driversByName.size > 1) {
+      throw new ValueError('ExternalStorage.driverSelector is required when more than one driver is registered');
+    }
+
+    this.drivers = [...drivers];
+    this.driverSelector = driverSelector ?? (() => drivers[0] as StorageDriver);
+    this.payloadSizeThreshold = payloadSizeThreshold;
+    this.driversByName = driversByName;
+  }
+
+  /** Look up a registered driver by name. Returns `null` if no driver with that name is registered. */
+  getDriver(name: string): StorageDriver | null {
+    return this.driversByName.get(name) ?? null;
+  }
+}

packages/common/src/internal-non-workflow/extstore-helpers.ts

@@ -0,0 +1,102 @@
+import Long from 'long';
+import * as proto from '@temporalio/proto';
+import { decode } from '../encoding';
+import { ValueError } from '../errors';
+import type { Payload } from '../interfaces';
+import { ProtobufJsonPayloadConverter } from '../converter/protobuf-payload-converters';
+import { encodingTypes, METADATA_ENCODING_KEY, METADATA_MESSAGE_TYPE_KEY } from '../converter/types';
+import type { StorageDriverClaim } from '../converter/extstore';
+
+const ExternalStorageReferenceProto = proto.temporal.api.sdk.v1.ExternalStorageReference;
+const PayloadProto = proto.temporal.api.common.v1.Payload;
+const storageReferenceConverter = new ProtobufJsonPayloadConverter(proto);
+
+const EXTSTORE_REFERENCE_MESSAGE_TYPE = 'temporal.api.sdk.v1.ExternalStorageReference';
+const EXTSTORE_REFERENCE_ENCODING = encodingTypes.METADATA_ENCODING_PROTOBUF_JSON;
+
+/**
+ * True if the payload is an External Storage reference:
+ * `temporal.api.sdk.v1.ExternalStorageReference`.
+ *
+ * @internal
+ * @experimental
+ */
+export function isReferencePayload(payload: Payload): boolean {
+  const encodingValue = readMetadataString(payload, METADATA_ENCODING_KEY);
+  if (encodingValue === EXTSTORE_REFERENCE_ENCODING) {
+    return readMetadataString(payload, METADATA_MESSAGE_TYPE_KEY) === EXTSTORE_REFERENCE_MESSAGE_TYPE;
+  }
+  return false;
+}
+
+/**
+ * Parsed contents of a reference payload.
+ *
+ * @internal
+ * @experimental
+ */
+export interface DecodedReferencePayload {
+  driverName: string;
+  claimData: Record<string, string>;
+  sizeBytes: number;
+}
+
+/**
+ * Decode a reference payload from the wire format {@link ExternalStorageReference}.
+ * Throws {@link ValueError} if the payload is not a reference or is malformed.
+ * Callers should gate on {@link isReferencePayload} first.
+ *
+ * @internal
+ * @experimental
+ */
+export function decodeReferencePayload(payload: Payload): DecodedReferencePayload {
+  const ref = storageReferenceConverter.fromPayload<proto.temporal.api.sdk.v1.IExternalStorageReference>(payload);
+  if (!ref.driverName) {
+    // TODO: use the new stable error codes here
+    throw new ValueError("Reference payload field 'driverName' must be a non-empty string");
+  }
+  return {
+    driverName: ref.driverName,
+    claimData: { ...ref.claimData },
+    sizeBytes: readSizeBytes(payload),
+  };
+}
+
+/**
+ * Encode a reference payload to wire format.
+ *
+ * @internal
+ * @experimental
+ */
+export function encodeReferencePayload({
+  driverName,
+  claim,
+  sizeBytes,
+}: {
+  driverName: string;
+  claim: StorageDriverClaim;
+  sizeBytes: number;
+}): Payload {
+  const ref = ExternalStorageReferenceProto.create({ driverName, claimData: claim.claimData });
+  const payload = storageReferenceConverter.toPayload(ref);
+  if (payload === undefined) {
+    // TODO: use the new stable error codes here
+    throw new ValueError('Failed to serialize ExternalStorageReference');
+  }
+  return {
+    ...payload,
+    externalPayloads: [PayloadProto.ExternalPayloadDetails.create({ sizeBytes: Long.fromNumber(sizeBytes) })],
+  };
+}
+
+function readMetadataString(payload: Payload, key: string): string | undefined {
+  const raw = payload.metadata?.[key];
+  if (!raw) return undefined;
+  return decode(raw);
+}
+
+function readSizeBytes(payload: Payload): number {
+  const details = payload.externalPayloads?.[0];
+  if (!details?.sizeBytes) return 0;
+  return Long.isLong(details.sizeBytes) ? details.sizeBytes.toNumber() : Number(details.sizeBytes);
+}

packages/common/src/internal-non-workflow/index.ts

@@ -6,6 +6,7 @@
 export * from './codec-helpers';
 export * from './codec-types';
 export * from './data-converter-helpers';
+export * from './extstore-helpers';
 export * from './parse-host-uri';
 export * from './proxy-config';
 export * from './tls-config';

packages/proto/scripts/compile-proto.ts

@@ -87,6 +87,7 @@ async function main() {
     'temporal/api/cloud/cloudservice/v1/service.proto',
     'temporal/api/errordetails/v1/message.proto',
     'temporal/api/sdk/v1/workflow_metadata.proto',
+    'temporal/api/sdk/v1/external_storage.proto',
     'temporal/api/testservice/v1/request_response.proto',
     'temporal/api/testservice/v1/service.proto',
     'grpc/health/v1/health.proto',